Merge branch 'master' into benc-channels-hard-deprecate

Parsl · Aug 7, 2024 · 42c5a56 · 42c5a56
2 parents a16a846 + 10a6a00
commit 42c5a56
Show file tree

Hide file tree

Showing 6 changed files with 10 additions and 10 deletions.
diff --git a/parsl/dataflow/dflow.py b/parsl/dataflow/dflow.py
@@ -116,7 +116,7 @@ def __init__(self, config: Config) -> None:
         if self.monitoring:
             if self.monitoring.logdir is None:
                 self.monitoring.logdir = self.run_dir
-            self.monitoring.start(self.run_id, self.run_dir, self.config.run_dir)
+            self.monitoring.start(self.run_dir, self.config.run_dir)
 
         self.time_began = datetime.datetime.now()
         self.time_completed: Optional[datetime.datetime] = None

diff --git a/parsl/executors/high_throughput/executor.py b/parsl/executors/high_throughput/executor.py
@@ -551,6 +551,7 @@ def _start_local_interchange_process(self) -> None:
                               "logging_level": logging.DEBUG if self.worker_debug else logging.INFO,
                               "cert_dir": self.cert_dir,
                               "manager_selector": self.manager_selector,
+                              "run_id": self.run_id,
                               }
 
         config_pickle = pickle.dumps(interchange_config)

diff --git a/parsl/executors/high_throughput/interchange.py b/parsl/executors/high_throughput/interchange.py
@@ -55,6 +55,7 @@ def __init__(self,
                  poll_period: int,
                  cert_dir: Optional[str],
                  manager_selector: ManagerSelector,
+                 run_id: str,
                  ) -> None:
         """
         Parameters
@@ -125,6 +126,8 @@ def __init__(self,
         self.command_channel.connect("tcp://{}:{}".format(client_address, client_ports[2]))
         logger.info("Connected to client")
 
+        self.run_id = run_id
+
         self.hub_address = hub_address
         self.hub_zmq_port = hub_zmq_port
 
@@ -227,6 +230,7 @@ def _send_monitoring_info(self, monitoring_radio: Optional[MonitoringRadioSender
             d: Dict = cast(Dict, manager.copy())
             d['timestamp'] = datetime.datetime.now()
             d['last_heartbeat'] = datetime.datetime.fromtimestamp(d['last_heartbeat'])
+            d['run_id'] = self.run_id
 
             monitoring_radio.send((MessageType.NODE_INFO, d))
 

diff --git a/parsl/monitoring/monitoring.py b/parsl/monitoring/monitoring.py
@@ -106,7 +106,7 @@ def __init__(self,
         self.resource_monitoring_enabled = resource_monitoring_enabled
         self.resource_monitoring_interval = resource_monitoring_interval
 
-    def start(self, run_id: str, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> None:
+    def start(self, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> None:
 
         logger.debug("Starting MonitoringHub")
 
@@ -161,7 +161,6 @@ 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
                                                },
                                        name="Monitoring-Router-Process",
                                        daemon=True,

diff --git a/parsl/monitoring/router.py b/parsl/monitoring/router.py
@@ -31,7 +31,6 @@ def __init__(self,
 
                  monitoring_hub_address: str = "127.0.0.1",
                  logdir: str = ".",
-                 run_id: str,
                  logging_level: int = logging.INFO,
                  atexit_timeout: int = 3,   # in seconds
                  priority_msgs: "queue.Queue[AddressedMonitoringMessage]",
@@ -71,7 +70,6 @@ def __init__(self,
 
         self.hub_address = hub_address
         self.atexit_timeout = atexit_timeout
-        self.run_id = run_id
 
         self.loop_freq = 10.0  # milliseconds
 
@@ -172,7 +170,6 @@ def start_zmq_listener(self) -> None:
                         msg_0 = (msg, 0)
 
                         if msg[0] == MessageType.NODE_INFO:
-                            msg[1]['run_id'] = self.run_id
                             self.node_msgs.put(msg_0)
                         elif msg[0] == MessageType.RESOURCE_INFO:
                             self.resource_msgs.put(msg_0)
@@ -218,16 +215,14 @@ def router_starter(comm_q: "queue.Queue[Union[Tuple[int, int], str]]",
                    zmq_port_range: Tuple[int, int],
 
                    logdir: str,
-                   logging_level: int,
-                   run_id: str) -> None:
+                   logging_level: int) -> None:
     setproctitle("parsl: monitoring router")
     try:
         router = MonitoringRouter(hub_address=hub_address,
                                   udp_port=udp_port,
                                   zmq_port_range=zmq_port_range,
                                   logdir=logdir,
                                   logging_level=logging_level,
-                                  run_id=run_id,
                                   priority_msgs=priority_msgs,
                                   node_msgs=node_msgs,
                                   block_msgs=block_msgs,

diff --git a/parsl/tests/test_htex/test_zmq_binding.py b/parsl/tests/test_htex/test_zmq_binding.py
@@ -25,7 +25,8 @@ def make_interchange(*, interchange_address: Optional[str], cert_dir: Optional[s
                        logdir=".",
                        logging_level=logging.INFO,
                        manager_selector=RandomManagerSelector(),
-                       poll_period=10)
+                       poll_period=10,
+                       run_id="test_run_id")
 
 
 @pytest.fixture