Create and use a serial retry executor

To make it easily possible to change the retry
atom execution from being in the engine thread this
creates a retry executor (which is similar to the task
executor) and provide that a serial executor (which it will
use to execute with). This makes the retry and task actions
closer to being the same and makes the surrounding code that
much similar (which makes understanding it easier).

Change-Id: I993e938280df3bd97f8076293183ef21989e2dba

1 files changed, 15 insertions, 7 deletions

diff --git a/taskflow/engines/action_engine/engine.py b/taskflow/engines/action_engine/engine.py
index 6d9ee26..fa6247d 100644
--- a/taskflow/engines/action_engine/engine.py
+++ b/taskflow/engines/action_engine/engine.py
@@ -41,13 +41,17 @@ LOG = logging.getLogger(__name__)
 
 
 @contextlib.contextmanager
-def _start_stop(executor):
-    # A teenie helper context manager to safely start/stop a executor...
-    executor.start()
+def _start_stop(task_executor, retry_executor):
+    # A teenie helper context manager to safely start/stop engine executors...
+    task_executor.start()
     try:
-        yield executor
+        retry_executor.start()
+        try:
+            yield (task_executor, retry_executor)
+        finally:
+            retry_executor.stop()
     finally:
-        executor.stop()
+        task_executor.stop()
 
 
 class ActionEngine(base.Engine):
@@ -82,6 +86,9 @@ class ActionEngine(base.Engine):
         self._lock = threading.RLock()
         self._state_lock = threading.RLock()
         self._storage_ensured = False
+        # Retries are not *currently* executed out of the engines process
+        # or thread (this could change in the future if we desire it to).
+        self._retry_executor = executor.SerialRetryExecutor()
 
     def _check(self, name, check_compiled, check_storage_ensured):
         """Check (and raise) if the engine has not reached a certain stage."""
@@ -167,7 +174,7 @@ class ActionEngine(base.Engine):
         self.validate()
         runner = self._runtime.runner
         last_state = None
-        with _start_stop(self._task_executor):
+        with _start_stop(self._task_executor, self._retry_executor):
             self._change_state(states.RUNNING)
             try:
                 closed = False
@@ -294,7 +301,8 @@ class ActionEngine(base.Engine):
         self._runtime = runtime.Runtime(self._compilation,
                                         self.storage,
                                         self.atom_notifier,
-                                        self._task_executor)
+                                        self._task_executor,
+                                        self._retry_executor)
         self._runtime.compile()
         self._compiled = True