Merge pull request #14793 from xiaket/devel

remove main_q for simplicity.

lib/ansible/executor/process/result.py

@@ -65,7 +65,7 @@ class ResultProcess(multiprocessing.Process):
         result = None
         starting_point = self._cur_worker
         while True:
-            (worker_prc, main_q, rslt_q) = self._workers[self._cur_worker]
+            (worker_prc, rslt_q) = self._workers[self._cur_worker]
             self._cur_worker += 1
             if self._cur_worker >= len(self._workers):
                 self._cur_worker = 0

lib/ansible/executor/task_queue_manager.py

@@ -99,9 +99,8 @@ class TaskQueueManager:
         self._workers = []
 
         for i in range(num):
-            main_q = multiprocessing.Queue()
             rslt_q = multiprocessing.Queue()
-            self._workers.append([None, main_q, rslt_q])
+            self._workers.append([None, rslt_q])
 
         self._result_prc = ResultProcess(self._final_q, self._workers)
         self._result_prc.start()
@@ -249,9 +248,8 @@ class TaskQueueManager:
         if self._result_prc:
             self._result_prc.terminate()
 
-            for (worker_prc, main_q, rslt_q) in self._workers:
+            for (worker_prc, rslt_q) in self._workers:
                 rslt_q.close()
-                main_q.close()
                 if worker_prc and worker_prc.is_alive():
                     try:
                         worker_prc.terminate()

lib/ansible/plugins/strategy/__init__.py

@@ -153,7 +153,7 @@ class StrategyBase:
 
             queued = False
             while True:
-                (worker_prc, main_q, rslt_q) = self._workers[self._cur_worker]
+                (worker_prc, rslt_q) = self._workers[self._cur_worker]
                 if worker_prc is None or not worker_prc.is_alive():
                     worker_prc = WorkerProcess(rslt_q, task_vars, host, task, play_context, self._loader, self._variable_manager, shared_loader_obj)
                     self._workers[self._cur_worker][0] = worker_prc