Add a function to check for killed processes in all strategies (#16684)

* Add a function to check for killed processes so that if any
threads are sigkilled or sigtermed, the entire playbook execution is aborted.

(cherry picked from commit 238c6461f6)

lib/ansible/executor/task_queue_manager.py

@@ -309,6 +309,18 @@ class TaskQueueManager:
     def terminate(self):
         self._terminated = True
 
+    def has_dead_workers(self):
+
+        # [<WorkerProcess(WorkerProcess-2, stopped[SIGKILL])>,
+        # <WorkerProcess(WorkerProcess-2, stopped[SIGTERM])>
+
+        defunct = False
+        for idx,x in enumerate(self._workers):
+            if hasattr(x[0], 'exitcode'):
+                if x[0].exitcode in [-9, -15]:
+                    defunct = True
+        return defunct
+
     def send_callback(self, method_name, *args, **kwargs):
         for callback_plugin in [self._stdout_callback] + self._callback_plugins:
             # a plugin that set self.disabled to True will not be called

lib/ansible/plugins/strategy/__init__.py

@@ -443,9 +443,14 @@ class StrategyBase:
 
         display.debug("waiting for pending results...")
         while self._pending_results > 0 and not self._tqm._terminated:
+
+            if self._tqm.has_dead_workers():
+                raise AnsibleError("A worker was found in a dead state")
+
             results = self._process_pending_results(iterator)
             ret_results.extend(results)
-            time.sleep(0.0001)
+            time.sleep(0.005)
+
         display.debug("no more pending results, returning what we have")
 
         return ret_results

test/units/plugins/strategies/test_strategy_base.py

@@ -162,7 +162,7 @@ class TestStrategyBase(unittest.TestCase):
                 raise Queue.Empty
             else:
                 return queue_items.pop()
-            
+
         mock_queue = MagicMock()
         mock_queue.empty.side_effect = _queue_empty
         mock_queue.get.side_effect = _queue_get
@@ -228,6 +228,10 @@ class TestStrategyBase(unittest.TestCase):
         strategy_base._variable_manager = mock_var_mgr
         strategy_base._blocked_hosts = dict()
 
+        def _has_dead_workers():
+            return False            
+
+        strategy_base._tqm.has_dead_workers = _has_dead_workers
         results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
         self.assertEqual(len(results), 0)