Move ZMQ monitoring radio initialization into HTEX (#3813)

This is part of an ongoing project to move monitoring message forwarding
into relevant executors, with executors only initializing whichever
routers are needed.

In this step of the project:

The ZMQ radio/router is only used by the high throughput executor to
receive messages from the interchange. This PR makes the initialization
of that look more like other interchange<->submit side ZMQ
initialiation, rather than look like monitoring system initialization.

This does not stop other future components which want to use
monitoring-over-ZMQ: those components would initialize ZMQ routing
themselves.

API-wise:

Parsl executors lose the hub_zmq_port attribute which was configured by
the DFK after initialising monitoring.

As a replacement, Parsl executors gain a monitoring_messages attribute,
which is the multiprocessing.Queue where executors should send
monitoring messages when they have received them from their own
components. This queue already existed between monitoring components,
but is now exposed as an interface for other Parsl components.

Configuration-wise:

Several ZMQ-related monitoring configuration options now come from HTEX
configuration, rather than from MonitoringHub configuration. There are
no new configuration options.

* Choice of ZMQ port now comes from the same range as other HTEX ZMQ
ports, rather than from the port range in MonitoringHub. This means the
minimum number of available ports there is now 4, instead of 3.

* ZMQ router log files now go into the High Throughput Executor's log
directory, usually a subdirectory of the main log directory.

* ZMQ log verbosity is configured by High Throughput Executor's
worker_debug configuration parameter, rather than by
MonitoringHub.monitoring_debug.

* The ZMQ router will now listen on the interchange loopback address,
configured by HighThroughputExecutor.loopback_address, rather than on
all addresses.

## Type of change

- New feature
- Code maintenance/cleanup

Author: benclifford

parsl/dataflow/dflow.py

@@ -1129,7 +1129,7 @@ def add_executors(self, executors: Sequence[ParslExecutor]) -> None:
             executor.run_dir = self.run_dir
             if self.monitoring:
                 executor.hub_address = self.monitoring.hub_address
-                executor.hub_zmq_port = self.monitoring.hub_zmq_port
+                executor.monitoring_messages = self.monitoring.resource_msgs
                 executor.submit_monitoring_radio = self.monitoring.radio
             if hasattr(executor, 'provider'):
                 if hasattr(executor.provider, 'script_dir'):

parsl/executors/base.py

@@ -1,11 +1,15 @@
+from __future__ import annotations
+
 import os
 from abc import ABCMeta, abstractmethod
 from concurrent.futures import Future
+from multiprocessing.queues import Queue
 from typing import Any, Callable, Dict, Optional
 
 from typing_extensions import Literal, Self
 
 from parsl.monitoring.radios.base import MonitoringRadioSender
+from parsl.monitoring.types import TaggedMonitoringMessage
 
 
 class ParslExecutor(metaclass=ABCMeta):
@@ -42,6 +46,16 @@ class ParslExecutor(metaclass=ABCMeta):
               invariant, not co-variant, and it looks like @typeguard cannot be
               persuaded otherwise. So if you're implementing an executor and want to
               @typeguard the constructor, you'll have to use List[Any] here.
+
+    The DataFlowKernel will set these two attributes before calling .start(),
+    if monitoring is enabled:
+
+        monitoring_messages: Optional[Queue[TaggedMonitoringMessage]] - an executor
+            can send messages to the monitoring hub by putting them into
+            this queue.
+
+        submit_monitoring_radio: Optional[MonitoringRadioSender] - an executor can
+            send messages to the monitoring hub by sending them using this sender.
     """
 
     label: str = "undefined"
@@ -51,13 +65,13 @@ def __init__(
         self,
         *,
         hub_address: Optional[str] = None,
-        hub_zmq_port: Optional[int] = None,
+        monitoring_messages: Optional[Queue[TaggedMonitoringMessage]] = None,
         submit_monitoring_radio: Optional[MonitoringRadioSender] = None,
         run_dir: str = ".",
         run_id: Optional[str] = None,
     ):
         self.hub_address = hub_address
-        self.hub_zmq_port = hub_zmq_port
+        self.monitoring_messages = monitoring_messages
         self.submit_monitoring_radio = submit_monitoring_radio
         self.run_dir = os.path.abspath(run_dir)
         self.run_id = run_id
@@ -136,16 +150,6 @@ def hub_address(self) -> Optional[str]:
     def hub_address(self, value: Optional[str]) -> None:
         self._hub_address = value
 
-    @property
-    def hub_zmq_port(self) -> Optional[int]:
-        """Port to the Hub for monitoring.
-        """
-        return self._hub_zmq_port
-
-    @hub_zmq_port.setter
-    def hub_zmq_port(self, value: Optional[int]) -> None:
-        self._hub_zmq_port = value
-
     @property
     def submit_monitoring_radio(self) -> Optional[MonitoringRadioSender]:
         """Local radio for sending monitoring messages

parsl/executors/high_throughput/executor.py

@@ -29,6 +29,7 @@
 )
 from parsl.executors.status_handling import BlockProviderExecutor
 from parsl.jobs.states import TERMINAL_STATES, JobState, JobStatus
+from parsl.monitoring.radios.zmq_router import ZMQRadioReceiver, start_zmq_receiver
 from parsl.process_loggers import wrap_with_logs
 from parsl.providers import LocalProvider
 from parsl.providers.base import ExecutionProvider
@@ -334,6 +335,10 @@ def __init__(self,
         self._result_queue_thread_exit = threading.Event()
         self._result_queue_thread: Optional[threading.Thread] = None
 
+        self.zmq_monitoring: Optional[ZMQRadioReceiver]
+        self.zmq_monitoring = None
+        self.hub_zmq_port = None
+
     radio_mode = "htex"
     enable_mpi_mode: bool = False
     mpi_launcher: str = "mpiexec"
@@ -427,6 +432,15 @@ def start(self):
             self.loopback_address, self.interchange_port_range, self.cert_dir
         )
 
+        if self.monitoring_messages is not None:
+            self.zmq_monitoring = start_zmq_receiver(monitoring_messages=self.monitoring_messages,
+                                                     loopback_address=self.loopback_address,
+                                                     port_range=self.interchange_port_range,
+                                                     logdir=self.logdir,
+                                                     worker_debug=self.worker_debug,
+                                                     )
+            self.hub_zmq_port = self.zmq_monitoring.port
+
         self._result_queue_thread = None
         self._start_result_queue_thread()
         self._start_local_interchange_process()
@@ -861,6 +875,9 @@ def shutdown(self, timeout: float = 10.0):
         if self._result_queue_thread:
             self._result_queue_thread.join()
 
+        if self.zmq_monitoring:
+            self.zmq_monitoring.close()
+
         logger.info("Finished HighThroughputExecutor shutdown attempt")
 
     def get_usage_information(self):

parsl/monitoring/errors.py

@@ -4,3 +4,8 @@
 class MonitoringHubStartError(ParslError):
     def __str__(self) -> str:
         return "Hub failed to start"
+
+
+class MonitoringRouterStartError(ParslError):
+    def __str__(self) -> str:
+        return "Monitoring router failed to start"

parsl/monitoring/monitoring.py

@@ -4,16 +4,16 @@
 import multiprocessing.synchronize as ms
 import os
 import queue
+import warnings
 from multiprocessing.queues import Queue
-from typing import TYPE_CHECKING, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Optional, Union
 
 import typeguard
 
 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.radios.zmq_router import zmq_router_starter
 from parsl.monitoring.types import TaggedMonitoringMessage
 from parsl.multiprocessing import (
     SizedQueue,
@@ -41,7 +41,7 @@ class MonitoringHub(RepresentationMixin):
     def __init__(self,
                  hub_address: str,
                  hub_port: Optional[int] = None,
-                 hub_port_range: Tuple[int, int] = (55050, 56000),
+                 hub_port_range: Any = None,
 
                  workflow_name: Optional[str] = None,
                  workflow_version: Optional[str] = None,
@@ -60,12 +60,11 @@ def __init__(self,
              Note that despite the similar name, this is not related to
              hub_port_range.
              Default: None
-        hub_port_range : tuple(int, int)
-             The port range for a ZMQ channel from an executor process
-             (for example, the interchange in the High Throughput Executor)
-             to deliver monitoring messages to the monitoring router.
-             Note that despite the similar name, this is not related to hub_port.
-             Default: (55050, 56000)
+        hub_port_range : unused
+             Unused, but retained until 2025-09-14 to avoid configuration errors.
+             This value previously configured one ZMQ channel inside the
+             HighThroughputExecutor. That ZMQ channel is now configured by the
+             interchange_port_range parameter of HighThroughputExecutor.
         workflow_name : str
              The name for the workflow. Default to the name of the parsl script
         workflow_version : str
@@ -92,6 +91,13 @@ def __init__(self,
 
         self.hub_address = hub_address
         self.hub_port = hub_port
+
+        if hub_port_range is not None:
+            message = "Instead of MonitoringHub.hub_port_range, Use HighThroughputExecutor.interchange_port_range"
+            warnings.warn(message, DeprecationWarning)
+            logger.warning(message)
+        # This is used by RepresentationMixin so needs to exist as an attribute
+        # even though now it is otherwise unused.
         self.hub_port_range = hub_port_range
 
         self.logging_endpoint = logging_endpoint
@@ -123,13 +129,10 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
         # in the future, Queue will allow runtime subscripts.
 
         if TYPE_CHECKING:
-            zmq_comm_q: Queue[Union[int, str]]
             udp_comm_q: Queue[Union[int, str]]
         else:
-            zmq_comm_q: Queue
             udp_comm_q: Queue
 
-        zmq_comm_q = SizedQueue(maxsize=10)
         udp_comm_q = SizedQueue(maxsize=10)
 
         self.resource_msgs: Queue[TaggedMonitoringMessage]
@@ -138,20 +141,6 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
         self.router_exit_event: ms.Event
         self.router_exit_event = SpawnEvent()
 
-        self.zmq_router_proc = SpawnProcess(target=zmq_router_starter,
-                                            kwargs={"comm_q": zmq_comm_q,
-                                                    "resource_msgs": self.resource_msgs,
-                                                    "exit_event": self.router_exit_event,
-                                                    "hub_address": self.hub_address,
-                                                    "zmq_port_range": self.hub_port_range,
-                                                    "run_dir": dfk_run_dir,
-                                                    "logging_level": logging.DEBUG if self.monitoring_debug else logging.INFO,
-                                                    },
-                                            name="Monitoring-ZMQ-Router-Process",
-                                            daemon=True,
-                                            )
-        self.zmq_router_proc.start()
-
         self.udp_router_proc = SpawnProcess(target=udp_router_starter,
                                             kwargs={"comm_q": udp_comm_q,
                                                     "resource_msgs": self.resource_msgs,
@@ -180,8 +169,8 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
                                      daemon=True,
                                      )
         self.dbm_proc.start()
-        logger.info("Started ZMQ router process %s, UDP router process %s and DBM process %s",
-                    self.zmq_router_proc.pid, self.udp_router_proc.pid, self.dbm_proc.pid)
+        logger.info("Started UDP router process %s and DBM process %s",
+                    self.udp_router_proc.pid, self.dbm_proc.pid)
 
         self.filesystem_proc = SpawnProcess(target=filesystem_router_starter,
                                             args=(self.resource_msgs, dfk_run_dir, self.router_exit_event),
@@ -193,20 +182,6 @@ def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> No
 
         self.radio = MultiprocessingQueueRadioSender(self.resource_msgs)
 
-        try:
-            zmq_comm_q_result = zmq_comm_q.get(block=True, timeout=120)
-            zmq_comm_q.close()
-            zmq_comm_q.join_thread()
-        except queue.Empty:
-            logger.error("Monitoring ZMQ Router has not reported port in 120s. Aborting")
-            raise MonitoringHubStartError()
-
-        if isinstance(zmq_comm_q_result, str):
-            logger.error("MonitoringRouter sent an error message: %s", zmq_comm_q_result)
-            raise RuntimeError(f"MonitoringRouter failed to start: {zmq_comm_q_result}")
-
-        self.hub_zmq_port = zmq_comm_q_result
-
         try:
             udp_comm_q_result = udp_comm_q.get(block=True, timeout=120)
             udp_comm_q.close()
@@ -235,9 +210,6 @@ def close(self) -> None:
             logger.info("Setting router termination event")
             self.router_exit_event.set()
 
-            logger.info("Waiting for ZMQ router to terminate")
-            join_terminate_close_proc(self.zmq_router_proc)
-
             logger.info("Waiting for UDP router to terminate")
             join_terminate_close_proc(self.udp_router_proc)