Fix UDPRadio atexit_timeout and use it in tests (#3899)

This parameter wasn't wired through, but because there was a default
deeper in the code, nothing detected that.

See #2973 for some context around this deep-default coding style: this
PR removes the duplicated defaults and leaves the default only on the
user facing UDPRadio class.

This PR uses the now-working parameter to make UDPRadioReceiver shutdown
faster in tests: because in these test cases, we know that we have
received all messages that we will process, there is no need to wait for
any further outstanding messages. This saves around 9 seconds of test
runtime - both theoretically (3 tests, each 3 second saving) and in
practice on my laptop.

Now that this is wired through properly, a type mismatch between the
original atexit_timeout (int) and the unwired UDPRadio parameter (float
| int) is revealed and fixed: everything uses the original int type now.

## Type of change

- Bug fix

Author: benclifford

parsl/monitoring/radios/udp.py

@@ -2,7 +2,7 @@
 import pickle
 import socket
 from multiprocessing.queues import Queue
-from typing import Optional, Union
+from typing import Optional
 
 from parsl.monitoring.radios.base import (
     MonitoringRadioReceiver,
@@ -15,7 +15,7 @@
 
 
 class UDPRadio(RadioConfig):
-    def __init__(self, *, port: Optional[int] = None, atexit_timeout: Union[int, float] = 3, address: str, debug: bool = False):
+    def __init__(self, *, port: Optional[int] = None, atexit_timeout: int = 3, address: str, debug: bool = False):
         self.port = port
         self.atexit_timeout = atexit_timeout
         self.address = address
@@ -29,7 +29,8 @@ def create_receiver(self, run_dir: str, resource_msgs: Queue) -> MonitoringRadio
         udp_receiver = start_udp_receiver(logdir=run_dir,
                                           monitoring_messages=resource_msgs,
                                           port=self.port,
-                                          debug=self.debug
+                                          debug=self.debug,
+                                          atexit_timeout=self.atexit_timeout
                                           )
         self.port = udp_receiver.port
         return udp_receiver

parsl/monitoring/radios/udp_router.py

@@ -38,7 +38,7 @@ def __init__(self,
                  udp_port: Optional[int] = None,
                  run_dir: str = ".",
                  logging_level: int = logging.INFO,
-                 atexit_timeout: int = 3,   # in seconds
+                 atexit_timeout: int,   # in seconds
                  resource_msgs: mpq.Queue,
                  exit_event: Event,
                  ):
@@ -129,14 +129,16 @@ def udp_router_starter(*,
                        udp_port: Optional[int],
 
                        run_dir: str,
-                       logging_level: int) -> None:
+                       logging_level: int,
+                       atexit_timeout: int) -> None:
     setproctitle("parsl: monitoring UDP router")
     try:
         router = MonitoringRouter(udp_port=udp_port,
                                   run_dir=run_dir,
                                   logging_level=logging_level,
                                   resource_msgs=resource_msgs,
-                                  exit_event=exit_event)
+                                  exit_event=exit_event,
+                                  atexit_timeout=atexit_timeout)
     except Exception as e:
         logger.error("MonitoringRouter construction failed.", exc_info=True)
         comm_q.put(f"Monitoring router construction failed: {e}")
@@ -165,7 +167,8 @@ def start_udp_receiver(*,
                        monitoring_messages: Queue,
                        port: Optional[int],
                        logdir: str,
-                       debug: bool) -> UDPRadioReceiver:
+                       debug: bool,
+                       atexit_timeout: int) -> UDPRadioReceiver:
 
     udp_comm_q: Queue[Union[int, str]]
     udp_comm_q = SizedQueue(maxsize=10)
@@ -179,6 +182,7 @@ def start_udp_receiver(*,
                                        "udp_port": port,
                                        "run_dir": logdir,
                                        "logging_level": logging.DEBUG if debug else logging.INFO,
+                                       "atexit_timeout": atexit_timeout,
                                        },
                                name="Monitoring-UDP-Router-Process",
                                daemon=True,

parsl/tests/test_monitoring/test_basic.py

@@ -28,7 +28,7 @@ def this_app():
 # a configuration that is suitably configured for monitoring.
 
 def thread_config():
-    c = Config(executors=[ThreadPoolExecutor(remote_monitoring_radio=UDPRadio(address="localhost"))],
+    c = Config(executors=[ThreadPoolExecutor(remote_monitoring_radio=UDPRadio(address="localhost", atexit_timeout=0))],
                monitoring=MonitoringHub(resource_monitoring_interval=0))
     return c
 
@@ -47,7 +47,7 @@ def htex_udp_config():
     ex = c.executors[0]
 
     assert isinstance(ex.remote_monitoring_radio, HTEXRadio), "precondition: htex is configured for the HTEXRadio"
-    ex.remote_monitoring_radio = UDPRadio(address="localhost")
+    ex.remote_monitoring_radio = UDPRadio(address="localhost", atexit_timeout=0)
 
     return c
 

parsl/tests/test_monitoring/test_radio_udp.py

@@ -15,7 +15,7 @@ def test_udp(tmpd_cwd):
 
     resource_msgs = SpawnQueue()
 
-    radio_config = UDPRadio(address="localhost")
+    radio_config = UDPRadio(address="localhost", atexit_timeout=0)
 
     # start receiver
     udp_receiver = radio_config.create_receiver(run_dir=str(tmpd_cwd),