Make htex managers track start_time (#3546)

Managers now record their start time and forward this information to the interchange during registration. The ManagerRecord was updated to support this functionality. Adding this will allow for better manager selection by the interchange in the future.
Parsl · Jul 26, 2024 · b96a2dd · b96a2dd
1 parent 0c24d7b
commit b96a2dd
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 0 deletions.
diff --git a/parsl/executors/high_throughput/interchange.py b/parsl/executors/high_throughput/interchange.py
@@ -410,6 +410,7 @@ def process_task_outgoing_incoming(
                 self._ready_managers[manager_id] = {'last_heartbeat': time.time(),
                                                     'idle_since': time.time(),
                                                     'block_id': None,
+                                                    'start_time': msg['start_time'],
                                                     'max_capacity': 0,
                                                     'worker_count': 0,
                                                     'active': True,

diff --git a/parsl/executors/high_throughput/manager_record.py b/parsl/executors/high_throughput/manager_record.py
@@ -6,6 +6,7 @@
 
 class ManagerRecord(TypedDict, total=False):
     block_id: Optional[str]
+    start_time: float
     tasks: List[Any]
     worker_count: int
     max_capacity: int

diff --git a/parsl/executors/high_throughput/process_worker_pool.py b/parsl/executors/high_throughput/process_worker_pool.py
@@ -184,6 +184,7 @@ def __init__(self, *,
 
         self.uid = uid
         self.block_id = block_id
+        self.start_time = time.time()
 
         self.enable_mpi_mode = enable_mpi_mode
         self.mpi_launcher = mpi_launcher
@@ -263,6 +264,7 @@ def create_reg_message(self):
                'worker_count': self.worker_count,
                'uid': self.uid,
                'block_id': self.block_id,
+               'start_time': self.start_time,
                'prefetch_capacity': self.prefetch_capacity,
                'max_capacity': self.worker_count + self.prefetch_capacity,
                'os': platform.system(),