Move scale in at exit code into close method of job status poller (#3311

)

Changed behaviour: this will now happen slightly before it used to, but the
differences are only changed place in the call stack, and log messages will
appear in a different order now - so the actual scaling-in behaviour should
be unchanged.

parsl/dataflow/dflow.py

@@ -1218,21 +1218,7 @@ def cleanup(self) -> None:
         self.job_status_poller.close()
         logger.info("Terminated job status poller")
 
-        logger.info("Scaling in and shutting down executors")
-
-        for ef in self.job_status_poller._executor_facades:
-            if not ef.executor.bad_state_is_set:
-                logger.info(f"Scaling in executor {ef.executor.label}")
-
-                # this code needs to be at least as many blocks as need
-                # cancelling, but it is safe to be more, as the scaling
-                # code will cope with being asked to cancel more blocks
-                # than exist.
-                block_count = len(ef.status)
-                ef.scale_in(block_count)
-
-            else:  # and bad_state_is_set
-                logger.warning(f"Not scaling in executor {ef.executor.label} because it is in bad state")
+        logger.info("Shutting down executors")
 
         for executor in self.executors.values():
             logger.info(f"Shutting down executor {executor.label}")

parsl/jobs/job_status_poller.py

@@ -136,3 +136,19 @@ def add_executors(self, executors: Sequence[BlockProviderExecutor]) -> None:
                 logger.debug("Adding executor {}".format(executor.label))
                 self._executor_facades.append(PolledExecutorFacade(executor, self.dfk))
         self._strategy.add_executors(executors)
+
+    def close(self):
+        super().close()
+        for ef in self._executor_facades:
+            if not ef.executor.bad_state_is_set:
+                logger.info(f"Scaling in executor {ef.executor.label}")
+
+                # this code needs to be at least as many blocks as need
+                # cancelling, but it is safe to be more, as the scaling
+                # code will cope with being asked to cancel more blocks
+                # than exist.
+                block_count = len(ef.status)
+                ef.scale_in(block_count)
+
+            else:  # and bad_state_is_set
+                logger.warning(f"Not scaling in executor {ef.executor.label} because it is in bad state")