move first field into strategy executor state, out of job status poller

this should not change behaviour

Author: benclifford

parsl/jobs/job_status_poller.py

@@ -19,7 +19,6 @@ class PolledExecutorFacade:
     def __init__(self, executor: BlockProviderExecutor, dfk: Optional["parsl.dataflow.dflow.DataFlowKernel"] = None):
         self._executor = executor
         self._dfk = dfk
-        self.first = True
 
         # Create a ZMQ channel to send poll status to monitoring
         self.monitoring_enabled = False

parsl/jobs/strategy.py

@@ -26,6 +26,10 @@ class ExecutorState(TypedDict):
     If the executor is not idle, then None.
     """
 
+    first: bool
+    """Is this the first poll for this executor?
+    """
+
 
 class Strategy:
     """Scaling strategy.
@@ -144,17 +148,17 @@ def __init__(self, *, strategy: Optional[str], max_idletime: float) -> None:
 
     def add_executors(self, executors: Sequence[ParslExecutor]) -> None:
         for executor in executors:
-            self.executors[executor.label] = {'idle_since': None}
+            self.executors[executor.label] = {'idle_since': None, 'first': True}
 
     def _strategy_init_only(self, status_list: List[jsp.PolledExecutorFacade]) -> None:
         """Scale up to init_blocks at the start, then nothing more.
         """
         for exec_status in status_list:
-            if exec_status.first:
-                executor = exec_status.executor
+            executor = exec_status.executor
+            if self.executors[executor.label]['first']:
                 logger.debug(f"strategy_init_only: scaling out {executor.provider.init_blocks} initial blocks for {executor.label}")
                 exec_status.scale_out(executor.provider.init_blocks)
-                exec_status.first = False
+                self.executors[executor.label]['first'] = False
             else:
                 logger.debug("strategy_init_only: doing nothing")
 
@@ -190,11 +194,11 @@ def _general_strategy(self, status_list, *, strategy_type):
                 continue
             logger.debug(f"Strategizing for executor {label}")
 
-            if exec_status.first:
+            if self.executors[label]['first']:
                 executor = exec_status.executor
                 logger.debug(f"Scaling out {executor.provider.init_blocks} initial blocks for {label}")
                 exec_status.scale_out(executor.provider.init_blocks)
-                exec_status.first = False
+                self.executors[label]['first'] = False
 
             # Tasks that are either pending completion
             active_tasks = executor.outstanding