push cache refresh state into executor

this should not change behaviour, but is prep for next patch that does change behaviour

it does not change behaviour because it is only moving attributes into different objects

Author: benclifford

parsl/executors/status_handling.py

@@ -73,6 +73,8 @@ def __init__(self, *,
         self.blocks_to_job_id = {}  # type: Dict[str, str]
         self.job_ids_to_block = {}  # type: Dict[str, str]
 
+        self._last_poll_time = 0.0
+
     def _make_status_dict(self, block_ids: List[str], status_list: List[JobStatus]) -> Dict[str, JobStatus]:
         """Given a list of block ids and a list of corresponding status strings,
         returns a dictionary mapping each block id to the corresponding status
@@ -236,3 +238,6 @@ def _get_block_and_job_ids(self) -> Tuple[List[str], List[Any]]:
     @abstractproperty
     def workers_per_node(self) -> Union[int, float]:
         pass
+
+    def _should_poll(self, now: float) -> bool:
+        return now >= self._last_poll_time + self.status_polling_interval

parsl/jobs/job_status_poller.py

@@ -20,8 +20,6 @@ class PolledExecutorFacade:
     def __init__(self, executor: BlockProviderExecutor, dfk: Optional["parsl.dataflow.dflow.DataFlowKernel"] = None):
         self._executor = executor
         self._dfk = dfk
-        self._interval = executor.status_polling_interval
-        self._last_poll_time = 0.0
         self.first = True
 
         # Create a ZMQ channel to send poll status to monitoring
@@ -36,15 +34,12 @@ def __init__(self, executor: BlockProviderExecutor, dfk: Optional["parsl.dataflo
             self.hub_channel.connect("tcp://{}:{}".format(hub_address, hub_port))
             logger.info("Monitoring enabled on job status poller")
 
-    def _should_poll(self, now: float) -> bool:
-        return now >= self._last_poll_time + self._interval
-
     def poll(self, now: float) -> None:
         previous_status = self.executor._poller_mutable_status
 
-        if self._should_poll(now):
+        if self.executor._should_poll(now):
             self.executor._poller_mutable_status = self._executor.status()
-            self._last_poll_time = now
+            self.executor._last_poll_time = now
 
         if previous_status != self.executor._poller_mutable_status:
             # short circuit the case where the two objects are identical so