proxy 'most' plugin instanciations

connections or others that rely on complex objects are ignore to avoid passing those through queue

lib/ansible/executor/task_queue_manager.py

@@ -66,8 +66,10 @@ class DisplaySend:
         self.args = args
         self.kwargs = kwargs
 
+
 class PluginLoaderSend:
-    def __init__(self, method, *args, **kwargs):
+    def __init__(self, loader_name, method, *args, **kwargs):
+        self.loader_name = loader_name
         self.method = method
         self.args = args
         self.kwargs = kwargs
@@ -112,8 +114,9 @@ class FinalQueue(multiprocessing.queues.SimpleQueue):
             PromptSend(**kwargs),
         )
 
-    def send_plugin_loader(self, *args, **kwargs):
-        self.put(PluginLoaderSend(*args, **kwargs))
+    def send_plugin_loader(self, loader_name, method, *args, **kwargs):
+        self.put(PluginLoaderSend(loader_name, method, *args, **kwargs))
+
 
 class AnsibleEndPlay(Exception):
     def __init__(self, result):

lib/ansible/plugins/loader.py

@@ -59,15 +59,20 @@ _final_q = None
 def proxy_load(method):
 
     def proxyit(self, *args, **kwargs):
-        if sys.modules[__name__]._final_q:
+        if sys.modules[__name__]._final_q and 'play_context' not in kwargs and self.type != 'connection':
             # If _final_q is set, that means we are in a WorkerProcess
             # and instead of displaying messages directly from the fork
             # we will proxy them through the queue
-            sys.modules[__name__]._final_q.send_plugin_loader(method.__name__, *args, **kwargs)
+            if self.class_name == 'ModuleDocFragment':
+                loader_name = 'fragment_loader'
+            else:
+                loader_name = '%s_loader' % self.type
+            sys.modules[__name__]._final_q.send_plugin_loader(loader_name, method.__name__, *args, **kwargs)
         return method(self, *args, **kwargs)
 
     return proxyit
 
+
 def set_queue(queue):
     """Set the _final_q on PLuginLoader, so that we know to proxy requests over the queue
     This is only needed in ansible.executor.process.worker:WorkerProcess._run

lib/ansible/plugins/strategy/__init__.py

@@ -41,7 +41,7 @@ from ansible.executor import action_write_locks
 from ansible.executor.play_iterator import IteratingStates, PlayIterator
 from ansible.executor.process.worker import WorkerProcess
 from ansible.executor.task_result import TaskResult
-from ansible.executor.task_queue_manager import CallbackSend, DisplaySend, PromptSend
+from ansible.executor.task_queue_manager import CallbackSend, DisplaySend, PromptSend, PluginLoaderSend
 from ansible.module_utils.six import string_types
 from ansible.module_utils.common.text.converters import to_text
 from ansible.module_utils.connection import Connection, ConnectionError
@@ -146,6 +146,12 @@ def results_thread_main(strategy):
                     except AnsibleError as e:
                         value = e
                 strategy._workers[result.worker_id].worker_queue.put(value)
+            elif isinstance(result, PluginLoaderSend):
+                loader = getattr(plugin_loader, result.loader_name)
+                if loader is not None:
+                    method = getattr(loader, result.method)
+                    if method is not None:
+                        method(*result.args, **result.kwargs)
             else:
                 display.warning('Received an invalid object (%s) in the result queue: %r' % (type(result), result))
         except (IOError, EOFError):