Move PolledExecutorFacade status into BlockProviderExecutor

This makes this status available to the scale_in methods of
BlockProviderExecutors, and so can be used as part of a scaling bugfix
in a subsequent PR.

This is part of ongoing work to minimise and possibly eliminate the
PolledExecutorFacade class.

This PR leaves accesses to those structures in the PolledExecutorFacade,
with future PRs intended to evaporate away more of the PolledExecutorFacade.

This should not change behaviour, as it only moves an attribute from one
object to a 1:1 paired object.

Author: benclifford

parsl/executors/status_handling.py

@@ -62,6 +62,8 @@ def __init__(self, *,
         # to keep track of such errors so that they can be handled in one place
         # together with errors reported by status()
         self._simulated_status: Dict[str, JobStatus] = {}
+        self._poller_mutable_status: Dict[str, JobStatus] = {}
+
         self._executor_bad_state = threading.Event()
         self._executor_exception: Optional[Exception] = None
 

parsl/jobs/job_status_poller.py

@@ -19,20 +19,19 @@ class PolledExecutorFacade:
     def __init__(self, executor: BlockProviderExecutor, monitoring: Optional["parsl.monitoring.radios.MonitoringRadio"] = None):
         self._executor = executor
         self._last_poll_time = 0.0
-        self._status = {}  # type: Dict[str, JobStatus]
         self._monitoring = monitoring
 
     def poll(self) -> None:
         now = time.time()
         if now >= self._last_poll_time + self._executor.status_polling_interval:
-            previous_status = self._status
-            self._status = self._executor.status()
+            previous_status = self._executor._poller_mutable_status
+            self._executor._poller_mutable_status = self._executor.status()
             self._last_poll_time = now
             delta_status = {}
-            for block_id in self._status:
+            for block_id in self._executor._poller_mutable_status:
                 if block_id not in previous_status \
-                   or previous_status[block_id].state != self._status[block_id].state:
-                    delta_status[block_id] = self._status[block_id]
+                   or previous_status[block_id].state != self._executor._poller_mutable_status[block_id].state:
+                    delta_status[block_id] = self._executor._poller_mutable_status[block_id]
 
             if delta_status:
                 self.send_monitoring_info(delta_status)
@@ -50,7 +49,7 @@ def status(self) -> Dict[str, JobStatus]:
 
         :return: a dictionary mapping block ids (in string) to job status
         """
-        return self._status
+        return self._executor._poller_mutable_status
 
     @property
     def executor(self) -> BlockProviderExecutor:
@@ -71,7 +70,7 @@ def scale_in(self, n: int, max_idletime: Optional[float] = None) -> List[str]:
             new_status = {}
             for block_id in block_ids:
                 new_status[block_id] = JobStatus(JobState.CANCELLED)
-                del self._status[block_id]
+                del self._executor._poller_mutable_status[block_id]
             self.send_monitoring_info(new_status)
         return block_ids
 
@@ -82,11 +81,11 @@ def scale_out(self, n: int) -> List[str]:
             for block_id in block_ids:
                 new_status[block_id] = JobStatus(JobState.PENDING)
             self.send_monitoring_info(new_status)
-            self._status.update(new_status)
+            self._executor._poller_mutable_status.update(new_status)
         return block_ids
 
     def __repr__(self) -> str:
-        return self._status.__repr__()
+        return self._executor._poller_mutable_status.__repr__()
 
 
 class JobStatusPoller(Timer):