Make it possible to terminate command pools

executor/__init__.py

@@ -61,7 +61,7 @@
     unicode = str
 
 # Semi-standard module versioning.
-__version__ = '5.0.1'
+__version__ = '5.1'
 
 # Initialize a logger.
 logger = logging.getLogger(__name__)
@@ -832,6 +832,10 @@ def terminate(self):
         """
         Terminate a running process.
 
+        :returns: :data:`True` if the process was terminated, :data:`False`
+                  otherwise (e.g. because :func:`start()` was never called or
+                  the process just finished).
+
         Uses the :func:`subprocess.Popen.terminate()` function. Calls
         :func:`wait()` after terminating the process so that the external
         command's output is loaded and temporary resources are cleaned up. The
@@ -844,6 +848,9 @@ def terminate(self):
             self.logger.debug("Terminating external command: %s", quote(self.command_line))
             self.subprocess.terminate()
             self.wait(check=False)
+            return True
+        else:
+            return False
 
     def load_output(self):
         """

executor/concurrent.py

@@ -208,7 +208,9 @@ def collect(self):
                  :data:`True`.
 
         .. warning:: If an exception is raised then commands that are still
-                     running will not be aborted!
+                     running will not be aborted! If this concerns you then
+                     consider calling :func:`terminate()` from a
+                     :keyword:`finally` block.
         """
         num_collected = 0
         for identifier, command in self.commands:
@@ -223,3 +225,22 @@ def collect(self):
         if num_collected > 0:
             logger.debug("Collected %i external commands ..", num_collected)
         return num_collected
+
+    def terminate(self):
+        """
+        Terminate any commands that are currently running.
+
+        :returns: The number of commands that were terminated (an integer).
+
+        If :func:`terminate()` successfully terminates commands, you then call
+        :func:`collect()` and the :attr:`.check` property of a terminated
+        command is :data:`True` you will get an exception because terminated
+        commands (by definition) report a nonzero :attr:`.returncode`.
+        """
+        num_terminated = 0
+        for identifier, command in self.commands:
+            if command.terminate():
+                num_terminated += 1
+        if num_terminated > 0:
+            logger.debug("Terminated %i external commands ..", num_terminated)
+        return num_terminated

executor/tests.py

@@ -348,6 +348,26 @@ def test_command_pool_resumable(self):
         e2 = intercept(pool.collect)
         assert e2.command is c2
 
+    def test_command_pool_termination(self):
+        """Make sure command pools can be terminated on failure."""
+        pool = CommandPool()
+        # Include a command that just sleeps for a minute.
+        sleep_cmd = ExternalCommand('sleep 60')
+        pool.add(sleep_cmd)
+        # Include a command that immediately exits with a nonzero return code.
+        pool.add(ExternalCommand('exit 1', check=True))
+        # Start the command pool and terminate it as soon as the control flow
+        # returns to us (because `exit 1' causes an exception to be raised).
+        try:
+            pool.run()
+            assert False, "Assumed CommandPool.run() to raise ExternalCommandFailed!"
+        except ExternalCommandFailed:
+            pass
+        finally:
+            pool.terminate()
+        # Make sure the sleep command was terminated.
+        assert sleep_cmd.is_terminated
+
     def test_command_pool_logs_directory(self):
         """Make sure command pools can log output of commands in a directory."""
         directory = tempfile.mkdtemp()