Convert DFK/monitoring message.send() to multiprocessing Queue (#3835)

This consolidates the API that the MonitoringHub offers for sending a
monitoring message to be via a multiprocessing Queue.

Prior to this PR, the MonitoringHub exposed both a multiprocessing Queue
(since #3818) and a .send() method for use in the original process.

This is part of ongoing work to move external communication for
monitoring out of the MonitoringHub, except via this
multiprocessing.Queue.

This removes another constraint on what a replacement MonitoringHub
looks like - something that is being partially driven by Globus Compute
desire for some monitoring information via API, not via the existing
database implementation.

This PR splits the DFK state for monitoring into two: a configured
MonitoringHub (self.monitoring) which comes from the Config object as
before, and a new internal piece of state, self.monitoring_radio which
is now managed separately and configured from self.monitoring. This is
extra complexity for the DFK, as a tradeoff for a cleaner interface for
MonitoringHub, and reflects the DFK's dual role as "thing that deals
with tasks" and "God-object that binds everything together".

# Changed Behaviour

none

## Type of change

- Code maintenance/cleanup

Author: benclifford

parsl/dataflow/dflow.py

@@ -45,6 +45,7 @@
 from parsl.jobs.job_status_poller import JobStatusPoller
 from parsl.monitoring import MonitoringHub
 from parsl.monitoring.message_type import MessageType
+from parsl.monitoring.radios.multiprocessing import MultiprocessingQueueRadioSender
 from parsl.monitoring.remote import monitor_wrapper
 from parsl.process_loggers import wrap_with_logs
 from parsl.usage_tracking.usage import UsageTracker
@@ -110,8 +111,11 @@ def __init__(self, config: Config) -> None:
         self.monitoring: Optional[MonitoringHub]
         self.monitoring = config.monitoring
 
+        self.monitoring_radio = None
+
         if self.monitoring:
             self.monitoring.start(self.run_dir, self.config.run_dir)
+            self.monitoring_radio = MultiprocessingQueueRadioSender(self.monitoring.resource_msgs)
 
         self.time_began = datetime.datetime.now()
         self.time_completed: Optional[datetime.datetime] = None
@@ -156,9 +160,9 @@ def __init__(self, config: Config) -> None:
                 'host': gethostname(),
         }
 
-        if self.monitoring:
-            self.monitoring.send((MessageType.WORKFLOW_INFO,
-                                 workflow_info))
+        if self.monitoring_radio:
+            self.monitoring_radio.send((MessageType.WORKFLOW_INFO,
+                                       workflow_info))
 
         if config.checkpoint_files is not None:
             checkpoint_files = config.checkpoint_files
@@ -231,9 +235,9 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
             raise InternalConsistencyError(f"Exit case for {mode} should be unreachable, validated by typeguard on Config()")
 
     def _send_task_log_info(self, task_record: TaskRecord) -> None:
-        if self.monitoring:
+        if self.monitoring_radio:
             task_log_info = self._create_task_log_info(task_record)
-            self.monitoring.send((MessageType.TASK_INFO, task_log_info))
+            self.monitoring_radio.send((MessageType.TASK_INFO, task_log_info))
 
     def _create_task_log_info(self, task_record: TaskRecord) -> Dict[str, Any]:
         """
@@ -1215,15 +1219,16 @@ def cleanup(self) -> None:
         logger.info("Terminated executors")
         self.time_completed = datetime.datetime.now()
 
-        if self.monitoring:
+        if self.monitoring_radio:
             logger.info("Sending final monitoring message")
-            self.monitoring.send((MessageType.WORKFLOW_INFO,
-                                 {'tasks_failed_count': self.task_state_counts[States.failed],
-                                  'tasks_completed_count': self.task_state_counts[States.exec_done],
-                                  "time_began": self.time_began,
-                                  'time_completed': self.time_completed,
-                                  'run_id': self.run_id, 'rundir': self.run_dir}))
+            self.monitoring_radio.send((MessageType.WORKFLOW_INFO,
+                                       {'tasks_failed_count': self.task_state_counts[States.failed],
+                                        'tasks_completed_count': self.task_state_counts[States.exec_done],
+                                        "time_began": self.time_began,
+                                        'time_completed': self.time_completed,
+                                        'run_id': self.run_id, 'rundir': self.run_dir}))
 
+        if self.monitoring:
             logger.info("Terminating monitoring")
             self.monitoring.close()
             logger.info("Terminated monitoring")

parsl/monitoring/monitoring.py

@@ -12,7 +12,6 @@
 
 from parsl.monitoring.errors import MonitoringHubStartError
 from parsl.monitoring.radios.filesystem_router import filesystem_router_starter
-from parsl.monitoring.radios.multiprocessing import MultiprocessingQueueRadioSender
 from parsl.monitoring.radios.udp_router import udp_router_starter
 from parsl.monitoring.types import TaggedMonitoringMessage
 from parsl.multiprocessing import (
@@ -180,8 +179,6 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
         self.filesystem_proc.start()
         logger.info("Started filesystem radio receiver process %s", self.filesystem_proc.pid)
 
-        self.radio = MultiprocessingQueueRadioSender(self.resource_msgs)
-
         try:
             udp_comm_q_result = udp_comm_q.get(block=True, timeout=120)
             udp_comm_q.close()
@@ -199,10 +196,6 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
 
         logger.info("Monitoring Hub initialized")
 
-    def send(self, message: TaggedMonitoringMessage) -> None:
-        logger.debug("Sending message type %s", message[0])
-        self.radio.send(message)
-
     def close(self) -> None:
         logger.info("Terminating Monitoring Hub")
         if self.monitoring_hub_active: