Merge "Fix process based executor task proxying-back events"

taskflow/engines/action_engine/process_executor.py

@@ -34,6 +34,7 @@ import six
 from taskflow.engines.action_engine import executor as base
 from taskflow import logging
 from taskflow import task as ta
+from taskflow.types import notifier as nt
 from taskflow.utils import iter_utils
 from taskflow.utils import misc
 from taskflow.utils import schema_utils as su
@@ -675,16 +676,38 @@ class ParallelProcessTaskExecutor(base.ParallelTaskExecutor):
             # so that when the clone runs in another process that this task
             # can receive the same notifications (thus making it look like the
             # the notifications are transparently happening in this process).
-            needed = set()
+            proxy_event_types = set()
             for (event_type, listeners) in task.notifier.listeners_iter():
                 if listeners:
-                    needed.add(event_type)
+                    proxy_event_types.add(event_type)
             if progress_callback is not None:
-                needed.add(ta.EVENT_UPDATE_PROGRESS)
+                proxy_event_types.add(ta.EVENT_UPDATE_PROGRESS)
-            if needed:
+            if nt.Notifier.ANY in proxy_event_types:
+                # NOTE(harlowja): If ANY is present, just have it be
+                # the **only** event registered, as all other events will be
+                # sent if ANY is registered (due to the nature of ANY sending
+                # all the things); if we also include the other event types
+                # in this set if ANY is present we will receive duplicate
+                # messages in this process (the one where the local
+                # task callbacks are being triggered). For example the
+                # emissions of the tasks notifier (that is running out
+                # of process) will for specific events send messages for
+                # its ANY event type **and** the specific event
+                # type (2 messages, when we just want one) which will
+                # cause > 1 notify() call on the local tasks notifier, which
+                # causes more local callback triggering than we want
+                # to actually happen.
+                proxy_event_types = set([nt.Notifier.ANY])
+            if proxy_event_types:
+                # This sender acts as our forwarding proxy target, it
+                # will be sent pickled to the process that will execute
+                # the needed task and it will do the work of using the
+                # channel object to send back messages to this process for
+                # dispatch into the local task.
                 sender = EventSender(channel)
-                for event_type in needed:
+                for event_type in proxy_event_types:
                     clone.notifier.register(event_type, sender)
+            return bool(proxy_event_types)
 
         def register():
             if progress_callback is not None:
@@ -698,14 +721,17 @@ class ParallelProcessTaskExecutor(base.ParallelTaskExecutor):
                                          progress_callback)
             self._dispatcher.targets.pop(identity, None)
 
-        rebind_task()
+        should_register = rebind_task()
-        register()
+        if should_register:
+            register()
         try:
             fut = self._executor.submit(func, clone, *args, **kwargs)
         except RuntimeError:
             with excutils.save_and_reraise_exception():
-                deregister()
+                if should_register:
+                    deregister()
 
         fut.atom = task
-        fut.add_done_callback(deregister)
+        if should_register:
+            fut.add_done_callback(deregister)
         return fut

taskflow/tests/unit/test_engines.py

@@ -1527,6 +1527,48 @@ class ParallelEngineWithProcessTest(EngineTaskTest,
                                      max_workers=self._EXECUTOR_WORKERS,
                                      **kwargs)
 
+    def test_update_progress_notifications_proxied(self):
+        captured = collections.defaultdict(list)
+
+        def notify_me(event_type, details):
+            captured[event_type].append(details)
+
+        a = utils.MultiProgressingTask('a')
+        a.notifier.register(a.notifier.ANY, notify_me)
+        progress_chunks = list(x / 10.0 for x in range(1, 10))
+        e = self._make_engine(a, store={'progress_chunks': progress_chunks})
+        e.run()
+
+        self.assertEqual(11, len(captured[task.EVENT_UPDATE_PROGRESS]))
+
+    def test_custom_notifications_proxied(self):
+        captured = collections.defaultdict(list)
+
+        def notify_me(event_type, details):
+            captured[event_type].append(details)
+
+        a = utils.EmittingTask('a')
+        a.notifier.register(a.notifier.ANY, notify_me)
+        e = self._make_engine(a)
+        e.run()
+
+        self.assertEqual(1, len(captured['hi']))
+        self.assertEqual(2, len(captured[task.EVENT_UPDATE_PROGRESS]))
+
+    def test_just_custom_notifications_proxied(self):
+        captured = collections.defaultdict(list)
+
+        def notify_me(event_type, details):
+            captured[event_type].append(details)
+
+        a = utils.EmittingTask('a')
+        a.notifier.register('hi', notify_me)
+        e = self._make_engine(a)
+        e.run()
+
+        self.assertEqual(1, len(captured['hi']))
+        self.assertEqual(0, len(captured[task.EVENT_UPDATE_PROGRESS]))
+
 
 class WorkerBasedEngineTest(EngineTaskTest,
                             EngineMultipleResultsTest,

taskflow/tests/utils.py

@@ -19,6 +19,7 @@ import string
 import threading
 import time
 
+from oslo_utils import timeutils
 import redis
 import six
 
@@ -104,6 +105,15 @@ class DummyTask(task.Task):
         pass
 
 
+class EmittingTask(task.Task):
+    TASK_EVENTS = (task.EVENT_UPDATE_PROGRESS, 'hi')
+
+    def execute(self, *args, **kwargs):
+        self.notifier.notify('hi',
+                             details={'sent_on': timeutils.utcnow(),
+                                      'args': args, 'kwargs': kwargs})
+
+
 class AddOneSameProvidesRequires(task.Task):
     default_provides = 'value'