Fix executor requests publishing bug

Make sure request is set to the PENDING state before putting
it into the requests cache to prevent the notify processing
thread get list of waiting requests and publish it before it
is published in the executor thread. So request wouldn't be
published twice.

Change-Id: If6f39a82c5b8554cd429f9ad342021a0baad2a00
Closes-Bug: #1294046

taskflow/engines/worker_based/executor.py

@@ -85,6 +85,7 @@ class WorkerTaskExecutor(executor.TaskExecutorBase):
 
         # publish waiting requests
         for request in self._requests_cache.get_waiting_requests(tasks):
+            request.set_pending()
             self._publish_request(request, topic)
 
     def _process_response(self, response, message):
@@ -133,12 +134,19 @@ class WorkerTaskExecutor(executor.TaskExecutorBase):
         """Submit task request to workers."""
         request = pr.Request(task, task_uuid, action, arguments,
                              progress_callback, timeout, **kwargs)
-        self._requests_cache.set(request.uuid, request)
 
         # Get task's topic and publish request if topic was found.
         topic = self._workers_cache.get_topic_by_task(request.task_cls)
         if topic is not None:
+            # NOTE(skudriashev): Make sure request is set to the PENDING state
+            # before putting it into the requests cache to prevent the notify
+            # processing thread get list of waiting requests and publish it
+            # before it is published here, so it wouldn't be published twice.
+            request.set_pending()
+            self._requests_cache.set(request.uuid, request)
             self._publish_request(request, topic)
+        else:
+            self._requests_cache.set(request.uuid, request)
 
         return request.result
 
@@ -156,8 +164,6 @@ class WorkerTaskExecutor(executor.TaskExecutorBase):
                               request)
                 self._requests_cache.delete(request.uuid)
                 request.set_result(failure)
-        else:
-            request.set_pending()
 
     def _notify_topics(self):
         """Cyclically publish notify message to each topic."""

taskflow/tests/unit/worker_based/test_executor.py

@@ -226,11 +226,11 @@ class TestWorkerTaskExecutor(test.MockTestCase):
         expected_calls = [
             mock.call.Request(self.task, self.task_uuid, 'execute',
                               self.task_args, None, self.timeout),
+            mock.call.request.set_pending(),
             mock.call.proxy.publish(msg=self.request_inst_mock,
                                     routing_key=self.executor_topic,
                                     reply_to=self.executor_uuid,
-                                    correlation_id=self.task_uuid),
-            mock.call.request.set_pending()
+                                    correlation_id=self.task_uuid)
         ]
         self.assertEqual(self.master_mock.mock_calls, expected_calls)
 
@@ -247,11 +247,11 @@ class TestWorkerTaskExecutor(test.MockTestCase):
                               self.task_args, None, self.timeout,
                               failures=self.task_failures,
                               result=self.task_result),
+            mock.call.request.set_pending(),
             mock.call.proxy.publish(msg=self.request_inst_mock,
                                     routing_key=self.executor_topic,
                                     reply_to=self.executor_uuid,
-                                    correlation_id=self.task_uuid),
-            mock.call.request.set_pending()
+                                    correlation_id=self.task_uuid)
         ]
         self.assertEqual(self.master_mock.mock_calls, expected_calls)
 
@@ -277,6 +277,7 @@ class TestWorkerTaskExecutor(test.MockTestCase):
         expected_calls = [
             mock.call.Request(self.task, self.task_uuid, 'execute',
                               self.task_args, None, self.timeout),
+            mock.call.request.set_pending(),
             mock.call.proxy.publish(msg=self.request_inst_mock,
                                     routing_key=self.executor_topic,
                                     reply_to=self.executor_uuid,