work on pluggable API for worker-side monitoring radio

patch so far is overloading self.monitoring_radio for two different uses that should be clarified:
submit side radio and worker side radio

its a bit complicated for having a default radio (UDPRadio or HTEXRadio) even when monitoring is turned off:
I should check that the radio receiver does not get activated in this case: for example:
  * test that broken radio activation causes a startup error
  * test that configuration with broken radio doesn't cause a startup error when monitoringhub is not configured

zmq radio should always listen, and be the place where all radio receivers send their data.

udp radio and filesystem radio should turn into separate... something... processes?

for prototyping i guess it doesn't matter where I make them live, but the most behaviour preserving would keep them somehow separated?

docs/userguide/monitoring.rst

@@ -42,7 +42,6 @@ configuration. Here the `parsl.monitoring.MonitoringHub` is specified to use por
       ],
       monitoring=MonitoringHub(
           hub_address=address_by_hostname(),
-          hub_port=55055,
           monitoring_debug=False,
           resource_monitoring_interval=10,
       ),

parsl/configs/ASPIRE1.py

@@ -34,7 +34,6 @@
         ],
         monitoring=MonitoringHub(
             hub_address=address_by_interface('ib0'),
-            hub_port=55055,
             resource_monitoring_interval=10,
         ),
         strategy='simple',

parsl/dataflow/dflow.py

@@ -744,11 +744,10 @@ def launch_task(self, task_record: TaskRecord) -> Future:
                                                        kwargs=kwargs,
                                                        x_try_id=try_id,
                                                        x_task_id=task_id,
-                                                       monitoring_hub_url=self.monitoring.monitoring_hub_url,
+                                                       radio_config=executor.remote_monitoring_radio_config,
                                                        run_id=self.run_id,
                                                        logging_level=wrapper_logging_level,
                                                        sleep_dur=self.monitoring.resource_monitoring_interval,
-                                                       radio_mode=executor.radio_mode,
                                                        monitor_resources=executor.monitor_resources(),
                                                        run_dir=self.run_dir)
 
@@ -1146,6 +1145,19 @@ def add_executors(self, executors: Sequence[ParslExecutor]) -> None:
                 executor.hub_address = self.monitoring.hub_address
                 executor.hub_zmq_port = self.monitoring.hub_zmq_port
                 executor.submit_monitoring_radio = self.monitoring.radio
+                # this will modify the radio config object: it will add relevant parameters needed
+                # for the particular remote radio sender to communicate back
+                logger.info("starting monitoring receiver "
+                            f"for executor {executor} "
+                            f"with remote monitoring radio config {executor.remote_monitoring_radio_config}")
+                executor.monitoring_receiver = self.monitoring.start_receiver(executor.remote_monitoring_radio_config,
+                                                                              ip=self.monitoring.hub_address,
+                                                                              run_dir=self.run_dir)
+                # TODO: this is a weird way to start the receiver.
+                # Rather than in executor.start, but there's a tangle here
+                # trying to make the executors usable in a non-pure-parsl
+                # context where there is no DFK to grab config out of?
+                # (and no monitoring...)
             if hasattr(executor, 'provider'):
                 if hasattr(executor.provider, 'script_dir'):
                     executor.provider.script_dir = os.path.join(self.run_dir, 'submit_scripts')

parsl/executors/base.py

@@ -1,11 +1,19 @@
+import logging
 import os
 from abc import ABCMeta, abstractmethod
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, Optional
 
 from typing_extensions import Literal, Self
 
-from parsl.monitoring.radios.base import MonitoringRadioSender
+from parsl.monitoring.radios.base import (
+    MonitoringRadioReceiver,
+    MonitoringRadioSender,
+    RadioConfig,
+)
+from parsl.monitoring.radios.udp import UDPRadio
+
+logger = logging.getLogger(__name__)
 
 
 class ParslExecutor(metaclass=ABCMeta):
@@ -19,15 +27,13 @@ class ParslExecutor(metaclass=ABCMeta):
     no arguments and re-raises any thrown exception.
 
     In addition to the listed methods, a ParslExecutor instance must always
-    have a member field:
+    have these member fields:
 
        label: str - a human readable label for the executor, unique
               with respect to other executors.
 
-    Per-executor monitoring behaviour can be influenced by exposing:
-
-       radio_mode: str - a string describing which radio mode should be used to
-              send task resource data back to the submit side.
+       remote_monitoring_radio_config: RadioConfig describing how tasks on this executor
+              should report task resource status
 
     An executor may optionally expose:
 
@@ -45,11 +51,16 @@ class ParslExecutor(metaclass=ABCMeta):
     """
 
     label: str = "undefined"
-    radio_mode: str = "udp"
 
     def __init__(
         self,
         *,
+
+        # TODO: I'd like these two to go away but they're needed right now
+        # to configure the interchange monitoring radio, that is
+        # in addition to the submit and worker monitoring radios (!). They
+        # are effectivley a third monitoring radio config, though, so what
+        # should that look like for the interchange?
         hub_address: Optional[str] = None,
         hub_zmq_port: Optional[int] = None,
         submit_monitoring_radio: Optional[MonitoringRadioSender] = None,
@@ -58,10 +69,19 @@ def __init__(
     ):
         self.hub_address = hub_address
         self.hub_zmq_port = hub_zmq_port
+
+        # these are parameters for the monitoring radio to be used on the remote side
+        # eg. in workers - to send results back, and they should end up encapsulated
+        # inside a RadioConfig.
         self.submit_monitoring_radio = submit_monitoring_radio
+        self.remote_monitoring_radio_config: RadioConfig = UDPRadio()
+
         self.run_dir = os.path.abspath(run_dir)
         self.run_id = run_id
 
+        # will be set externally later, which is pretty ugly
+        self.monitoring_receiver: Optional[MonitoringRadioReceiver] = None
+
     def __enter__(self) -> Self:
         return self
 
@@ -94,7 +114,13 @@ def shutdown(self) -> None:
 
         This includes all attached resources such as workers and controllers.
         """
-        pass
+        logger.debug("Starting base executor shutdown")
+        # logger.error(f"BENC: monitoring receiver on {self} is {self.monitoring_receiver}")
+        if self.monitoring_receiver is not None:
+            logger.debug("Starting monitoring receiver shutdown")
+            self.monitoring_receiver.shutdown()
+            logger.debug("Done with monitoring receiver shutdown")
+        logger.debug("Done with base executor shutdown")
 
     def monitor_resources(self) -> bool:
         """Should resource monitoring happen for tasks on running on this executor?

parsl/executors/high_throughput/executor.py

@@ -29,6 +29,8 @@
 )
 from parsl.executors.status_handling import BlockProviderExecutor
 from parsl.jobs.states import TERMINAL_STATES, JobState, JobStatus
+from parsl.monitoring.radios.base import RadioConfig
+from parsl.monitoring.radios.htex import HTEXRadio
 from parsl.process_loggers import wrap_with_logs
 from parsl.providers import LocalProvider
 from parsl.providers.base import ExecutionProvider
@@ -253,11 +255,13 @@ def __init__(self,
                  worker_logdir_root: Optional[str] = None,
                  manager_selector: ManagerSelector = RandomManagerSelector(),
                  block_error_handler: Union[bool, Callable[[BlockProviderExecutor, Dict[str, JobStatus]], None]] = True,
-                 encrypted: bool = False):
+                 encrypted: bool = False,
+                 remote_monitoring_radio_config: Optional[RadioConfig] = None):
 
         logger.debug("Initializing HighThroughputExecutor")
 
         BlockProviderExecutor.__init__(self, provider=provider, block_error_handler=block_error_handler)
+
         self.label = label
         self.worker_debug = worker_debug
         self.storage_access = storage_access
@@ -300,6 +304,12 @@ def __init__(self,
             self._workers_per_node = 1  # our best guess-- we do not have any provider hints
 
         self._task_counter = 0
+
+        if remote_monitoring_radio_config is not None:
+            self.remote_monitoring_radio_config = remote_monitoring_radio_config
+        else:
+            self.remote_monitoring_radio_config = HTEXRadio()
+
         self.worker_ports = worker_ports
         self.worker_port_range = worker_port_range
         self.interchange_proc: Optional[subprocess.Popen] = None
@@ -322,7 +332,6 @@ def __init__(self,
             interchange_launch_cmd = DEFAULT_INTERCHANGE_LAUNCH_CMD
         self.interchange_launch_cmd = interchange_launch_cmd
 
-    radio_mode = "htex"
     enable_mpi_mode: bool = False
     mpi_launcher: str = "mpiexec"
 
@@ -832,6 +841,9 @@ def shutdown(self, timeout: float = 10.0):
             logger.info("Closing command client")
             self.command_client.close()
 
+        # TODO: implement this across all executors
+        super().shutdown()
+
         logger.info("Finished HighThroughputExecutor shutdown attempt")
 
     def get_usage_information(self):

parsl/executors/high_throughput/mpi_executor.py

@@ -15,6 +15,7 @@
 from parsl.executors.status_handling import BlockProviderExecutor
 from parsl.jobs.states import JobStatus
 from parsl.launchers import SimpleLauncher
+from parsl.monitoring.radios.base import RadioConfig
 from parsl.providers import LocalProvider
 from parsl.providers.base import ExecutionProvider
 
@@ -66,7 +67,8 @@ def __init__(self,
                  worker_logdir_root: Optional[str] = None,
                  mpi_launcher: str = "mpiexec",
                  block_error_handler: Union[bool, Callable[[BlockProviderExecutor, Dict[str, JobStatus]], None]] = True,
-                 encrypted: bool = False):
+                 encrypted: bool = False,
+                 remote_monitoring_radio_config: Optional[RadioConfig] = None):
         super().__init__(
             # Hard-coded settings
             cores_per_worker=1e-9,  # Ensures there will be at least an absurd number of workers
@@ -92,7 +94,15 @@ def __init__(self,
             address_probe_timeout=address_probe_timeout,
             worker_logdir_root=worker_logdir_root,
             block_error_handler=block_error_handler,
-            encrypted=encrypted
+            encrypted=encrypted,
+
+            # TODO:
+            # worker-side monitoring in MPI-style code is probably going to be
+            # broken - resource monitoring won't see any worker processes
+            # most likely, as so perhaps it should have worker resource
+            # monitoring disabled like the thread pool executor has?
+            # (for related but different reasons...)
+            remote_monitoring_radio_config=remote_monitoring_radio_config
         )
         self.enable_mpi_mode = True
         self.mpi_launcher = mpi_launcher

parsl/executors/taskvine/executor.py

@@ -605,6 +605,8 @@ def shutdown(self, *args, **kwargs):
         self._finished_task_queue.close()
         self._finished_task_queue.join_thread()
 
+        super().shutdown()
+
         logger.debug("TaskVine shutdown completed")
 
     @wrap_with_logs

parsl/executors/threads.py

@@ -73,6 +73,7 @@ def shutdown(self, block=True):
         """
         logger.debug("Shutting down executor, which involves waiting for running tasks to complete")
         self.executor.shutdown(wait=block)
+        super().shutdown()
         logger.debug("Done with executor shutdown")
 
     def monitor_resources(self):

parsl/executors/workqueue/executor.py

@@ -716,6 +716,8 @@ def shutdown(self, *args, **kwargs):
         self.collector_queue.close()
         self.collector_queue.join_thread()
 
+        super().shutdown()
+
         logger.debug("Work Queue shutdown completed")
 
     @wrap_with_logs