executors get current time directly from time, not from job status poller

this is a behavioural change - but what's the behavioural change?
if there are multiple executors, then cache refreshes might now happen
on a slightly different cadence: the 2nd executor now time will be
later (it will be the end time of the 1st executor polling process)
which means it might now fire a bit earlier

Author: benclifford

parsl/executors/status_handling.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 import logging
 import threading
+import time
 from itertools import compress
 from abc import abstractmethod, abstractproperty
 from concurrent.futures import Future
@@ -242,7 +243,8 @@ def workers_per_node(self) -> Union[int, float]:
     def _should_poll(self, now: float) -> bool:
         return now >= self._last_poll_time + self.status_polling_interval
 
-    def _refresh_poll_mutable_status_if_time(self, now):
+    def _refresh_poll_mutable_status_if_time(self):
+        now = time.time()
         if self._should_poll(now):
             self._poller_mutable_status = self.status()
             self._last_poll_time = now

parsl/jobs/job_status_poller.py

@@ -1,6 +1,5 @@
 import logging
 import parsl
-import time
 import zmq
 from typing import Dict, List, Sequence, Optional, Union
 
@@ -34,10 +33,10 @@ 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 poll(self, now: float) -> None:
+    def poll(self) -> None:
         previous_status = self.executor._poller_mutable_status
 
-        self.executor._refresh_poll_mutable_status_if_time(now)
+        self.executor._refresh_poll_mutable_status_if_time()
 
         if previous_status != self.executor._poller_mutable_status:
             # short circuit the case where the two objects are identical so
@@ -124,9 +123,8 @@ def _run_error_handlers(self, status: List[PolledExecutorFacade]) -> None:
             es.executor.handle_errors(es.status)
 
     def _update_state(self) -> None:
-        now = time.time()
         for item in self._executor_facades:
-            item.poll(now)
+            item.poll()
 
     def add_executors(self, executors: Sequence[BlockProviderExecutor]) -> None:
         for executor in executors: