Merge cd0239acb1 into 6aa8192131

lib/ansible/executor/task_queue_manager.py

@@ -22,6 +22,7 @@ import sys
 import tempfile
 import threading
 import time
+import traceback
 import typing as t
 import multiprocessing.queues
 
@@ -110,6 +111,15 @@ class AnsibleEndPlay(Exception):
         self.result = result
 
 
+def _callback_error(callback_plugin, method_name):
+    def inner(exception):
+        display.warning(
+            "Failure using method (%s) in callback plugin (%s): %s" % (to_text(method_name), to_text(callback_plugin), to_text(exception))
+        )
+        display.vvv('Callback Exception: \n' + ' '.join(traceback.format_exception(exception)))
+    return inner
+
+
 class TaskQueueManager:
 
     '''
@@ -164,6 +174,11 @@ class TaskQueueManager:
             raise AnsibleError("Unable to use multiprocessing, this is normally caused by lack of access to /dev/shm: %s" % to_native(e))
 
         self._callback_lock = threading.Lock()
+        self._callback_pool = multiprocessing_context.Pool(
+            processes=1,
+            initializer=lambda final_q: display.set_queue(final_q),
+            initargs=(self._final_q,),
+        )
 
         # A temporary file (opened pre-fork) used by connection
         # plugins for inter-process locking.
@@ -459,11 +474,12 @@ class TaskQueueManager:
                 continue
 
             for method in methods:
-                try:
-                    method(*new_args, **kwargs)
-                except Exception as e:
-                    # TODO: add config toggle to make this fatal or not?
-                    display.warning(u"Failure using method (%s) in callback plugin (%s): %s" % (to_text(method_name), to_text(callback_plugin), to_text(e)))
-                    from traceback import format_tb
-                    from sys import exc_info
-                    display.vvv('Callback Exception: \n' + ' '.join(format_tb(exc_info()[2])))
+                self._callback_pool.apply_async(method, new_args, kwargs, error_callback=_callback_error(callback_plugin, method))
+                # try:
+                #     self._callback_pool.apply_async(method, new_args, kwargs).get()
+                # except Exception as e:
+                #     # TODO: add config toggle to make this fatal or not?
+                #     display.warning(u"Failure using method (%s) in callback plugin (%s): %s" % (to_text(method_name), to_text(callback_plugin), to_text(e)))
+                #     from traceback import format_tb
+                #     from sys import exc_info
+                #     display.vvv('Callback Exception: \n' + ' '.join(format_tb(exc_info()[2])))

lib/ansible/plugins/callback/__init__.py

@@ -141,11 +141,6 @@ class CallbackBase(AnsiblePlugin):
     '''
 
     def __init__(self, display=None, options=None):
-        if display:
-            self._display = display
-        else:
-            self._display = global_display
-
         if self._display.verbosity >= 4:
             name = getattr(self, 'CALLBACK_NAME', 'unnamed')
             ctype = getattr(self, 'CALLBACK_TYPE', 'old')
@@ -164,6 +159,10 @@ class CallbackBase(AnsiblePlugin):
     # helper for callbacks, so they don't all have to include deepcopy
     _copy_result = deepcopy
 
+    @property
+    def _display(self):
+        return global_display
+
     def set_option(self, k, v):
         self._plugin_options[k] = C.config.get_config_value(k, plugin_type=self.plugin_type, plugin_name=self._load_name, direct={k: v})