Make UDP radio receiver shutdown period configurable

.. and use that new parameter in stderr/out tests which don't make use of
the UDP radio.

Author: benclifford

parsl/monitoring/monitoring.py

@@ -51,7 +51,8 @@ def __init__(self,
                  logdir: Optional[str] = None,
                  monitoring_debug: bool = False,
                  resource_monitoring_enabled: bool = True,
-                 resource_monitoring_interval: float = 30):  # in seconds
+                 resource_monitoring_interval: float = 30,  # in seconds
+                 udp_atexit_timeout: float = 3):
         """
         Parameters
         ----------
@@ -90,6 +91,10 @@ def __init__(self,
              If set to 0, only start and end information will be logged, and no periodic monitoring will
              be made.
              Default: 30 seconds
+        udp_atexit_timeout : float
+             The amount of time in seconds to wait for more UDP messages at shutdown, after the last DFK
+             workflow message is received.
+
         """
 
         if _db_manager_excepts:
@@ -109,6 +114,8 @@ def __init__(self,
         self.resource_monitoring_enabled = resource_monitoring_enabled
         self.resource_monitoring_interval = resource_monitoring_interval
 
+        self.udp_atexit_timeout = udp_atexit_timeout
+
     def start(self, run_id: str, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> int:
 
         logger.debug("Starting MonitoringHub")
@@ -164,7 +171,8 @@ def start(self, run_id: str, dfk_run_dir: str, config_run_dir: Union[str, os.Pat
                                                "zmq_port_range": self.hub_port_range,
                                                "logdir": self.logdir,
                                                "logging_level": logging.DEBUG if self.monitoring_debug else logging.INFO,
-                                               "run_id": run_id
+                                               "run_id": run_id,
+                                               "udp_atexit_timeout": self.udp_atexit_timeout
                                                },
                                        name="Monitoring-Router-Process",
                                        daemon=True,

parsl/monitoring/router.py

@@ -35,7 +35,7 @@ def __init__(self,
                  logdir: str = ".",
                  run_id: str,
                  logging_level: int = logging.INFO,
-                 atexit_timeout: int = 3    # in seconds
+                 atexit_timeout: float
                  ):
         """ Initializes a monitoring configuration class.
 
@@ -53,7 +53,8 @@ def __init__(self,
         logging_level : int
              Logging level as defined in the logging module. Default: logging.INFO
         atexit_timeout : float, optional
-            The amount of time in seconds to terminate the hub without receiving any messages, after the last dfk workflow message is received.
+            The amount of time in seconds to wait for more UDP messages at shutdown, after the last DFK
+            workflow message is received.
 
         """
         os.makedirs(logdir, exist_ok=True)
@@ -184,6 +185,8 @@ def router_starter(comm_q: "queue.Queue[Union[Tuple[int, int], str]]",
                    udp_port: Optional[int],
                    zmq_port_range: Tuple[int, int],
 
+                   udp_atexit_timeout: float,
+
                    logdir: str,
                    logging_level: int,
                    run_id: str) -> None:
@@ -194,7 +197,8 @@ def router_starter(comm_q: "queue.Queue[Union[Tuple[int, int], str]]",
                                   zmq_port_range=zmq_port_range,
                                   logdir=logdir,
                                   logging_level=logging_level,
-                                  run_id=run_id)
+                                  run_id=run_id,
+                                  atexit_timeout=udp_atexit_timeout)
     except Exception as e:
         logger.error("MonitoringRouter construction failed.", exc_info=True)
         comm_q.put(f"Monitoring router construction failed: {e}")

parsl/tests/test_monitoring/test_stdouterr.py

@@ -38,6 +38,7 @@ def fresh_config(run_dir):
         monitoring=MonitoringHub(
                         hub_address="localhost",
                         hub_port=55055,
+                        udp_atexit_timeout=0
         )
     )