move scale in at exit code into close method of job status poller

Author: benclifford

parsl/dataflow/dflow.py

@@ -1213,21 +1213,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 pi in self.job_status_poller._poll_items:
-            if not pi.executor.bad_state_is_set:
-                logger.info(f"Scaling in executor {pi.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(pi.status)
-                pi.scale_in(block_count)
-
-            else:  # and bad_state_is_set
-                logger.warning(f"Not scaling in executor {pi.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._poll_items.append(PolledExecutorFacade(executor, self.dfk))
         self._strategy.add_executors(executors)
+
+    def close(self):
+        super().close()
+        for pi in self._poll_items:
+            if not pi.executor.bad_state_is_set:
+                logger.info(f"Scaling in executor {pi.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(pi.status)
+                pi.scale_in(block_count)
+
+            else:  # and bad_state_is_set
+                logger.warning(f"Not scaling in executor {pi.executor.label} because it is in bad state")