Refactor naive scale in behaviour for Work Queue and Task Vine (#3526)

The intended behaviour of this scale in code, which is only for scaling in
all blocks (for example at the end of a workflow) makes sense as a default
for all BlockProviderExecutors. This PR makes that refactor.

This code is buggy (before and after) - see issue #3471. This PR does not
attempt to fix that, but moves code into a better place for bugfixing, and
a subsequent PR will fix it.

Author: benclifford

parsl/executors/status_handling.py

@@ -193,15 +193,32 @@ def _scale_out(self, blocks: int = 1) -> List[str]:
                 self._simulated_status[block_id] = JobStatus(JobState.FAILED, "Failed to start block {}: {}".format(block_id, ex))
         return block_ids
 
-    @abstractmethod
     def scale_in(self, blocks: int) -> List[str]:
         """Scale in method.
 
         Cause the executor to reduce the number of blocks by count.
 
+        The default implementation will kill blocks without regard to their
+        status or whether they are executing tasks. Executors with more
+        nuanced scaling strategies might overload this method to work with
+        that strategy - see the HighThroughputExecutor for an example of that.
+
         :return: A list of block ids corresponding to the blocks that were removed.
         """
-        pass
+        # Obtain list of blocks to kill
+        to_kill = list(self.blocks_to_job_id.keys())[:blocks]
+        kill_ids = [self.blocks_to_job_id[block] for block in to_kill]
+
+        # Cancel the blocks provisioned
+        if self.provider:
+            logger.info(f"Scaling in jobs: {kill_ids}")
+            r = self.provider.cancel(kill_ids)
+            job_ids = self._filter_scale_in_ids(kill_ids, r)
+            block_ids_killed = [self.job_ids_to_block[jid] for jid in job_ids]
+            return block_ids_killed
+        else:
+            logger.error("No execution provider available to scale in")
+            return []
 
     def _launch_block(self, block_id: str) -> Any:
         launch_cmd = self._get_launch_command(block_id)

parsl/executors/taskvine/executor.py

@@ -573,24 +573,6 @@ def outstanding(self) -> int:
     def workers_per_node(self) -> Union[int, float]:
         return 1
 
-    def scale_in(self, count: int) -> List[str]:
-        """Scale in method. Cancel a given number of blocks
-        """
-        # Obtain list of blocks to kill
-        to_kill = list(self.blocks_to_job_id.keys())[:count]
-        kill_ids = [self.blocks_to_job_id[block] for block in to_kill]
-
-        # Cancel the blocks provisioned
-        if self.provider:
-            logger.info(f"Scaling in jobs: {kill_ids}")
-            r = self.provider.cancel(kill_ids)
-            job_ids = self._filter_scale_in_ids(kill_ids, r)
-            block_ids_killed = [self.job_ids_to_block[jid] for jid in job_ids]
-            return block_ids_killed
-        else:
-            logger.error("No execution provider available to scale")
-            return []
-
     def shutdown(self, *args, **kwargs):
         """Shutdown the executor. Sets flag to cancel the submit process and
         collector thread, which shuts down the TaskVine system submission.

parsl/executors/workqueue/executor.py

@@ -689,24 +689,6 @@ def outstanding(self) -> int:
     def workers_per_node(self) -> Union[int, float]:
         return self.scaling_cores_per_worker
 
-    def scale_in(self, count: int) -> List[str]:
-        """Scale in method.
-        """
-        # Obtain list of blocks to kill
-        to_kill = list(self.blocks_to_job_id.keys())[:count]
-        kill_ids = [self.blocks_to_job_id[block] for block in to_kill]
-
-        # Cancel the blocks provisioned
-        if self.provider:
-            logger.info(f"Scaling in jobs: {kill_ids}")
-            r = self.provider.cancel(kill_ids)
-            job_ids = self._filter_scale_in_ids(kill_ids, r)
-            block_ids_killed = [self.job_ids_to_block[jid] for jid in job_ids]
-            return block_ids_killed
-        else:
-            logger.error("No execution provider available to scale in")
-            return []
-
     def shutdown(self, *args, **kwargs):
         """Shutdown the executor. Sets flag to cancel the submit process and
         collector thread, which shuts down the Work Queue system submission.