[2/3] incubator-ariatosca git commit: ARIA-156 Better handle exceptions in the process executor

[mx...@apache.org]

ARIA-156 Better handle exceptions in the process executor

Previously, if an exception was raised during the starting of a task,
the task's process was permanently blocked on receiving a message.

The reason was that the exception caused the 'listener thread' to
not send a response to the task's process, as the exception was not
handled inside the 'with' block of the listener thread.

The first change I introduced was to wrap the yielding of the message and
the response inside a try-except-finally block, so the exception will be
handled within the 'with' scope, and to ensure a response is sent to the
task's process.

The second change is to move the sending of the 'task started' message in
the task's process to a place where encountering an exception will be
handled via sending a 'task failed' message back to the listener thread.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/1cb3086f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/1cb3086f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/1cb3086f

Branch: refs/heads/ARIA-160-Operation-toolbelt-unit-tests-fail-spordically
Commit: 1cb3086f38bc35a6a5e588aa5e340cb4fa5cacf5
Parents: 8553977
Author: Avia Efrat <av...@gigaspaces.com>
Authored: Sun May 7 11:42:58 2017 +0300
Committer: Avia Efrat <av...@gigaspaces.com>
Committed: Sun May 7 14:44:01 2017 +0300

----------------------------------------------------------------------
 aria/orchestrator/workflows/executor/process.py | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/1cb3086f/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index 2378e0a..8481406 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -213,8 +213,13 @@ class ProcessExecutor(base.BaseExecutor):
         with contextlib.closing(self._server_socket.accept()[0]) as connection:
             message = _recv_message(connection)
             response = {}
-            yield message, response
-            _send_message(connection, response)
+            try:
+                yield message, response
+            except BaseException as e:
+                response['exception'] = exceptions.wrap_if_needed(e)
+                raise
+            finally:
+                _send_message(connection, response)
 
     def _handle_task_started_request(self, task_id, **kwargs):
         self._task_started(self._tasks[task_id])
@@ -378,7 +383,6 @@ def _main():
     task_id = arguments['task_id']
     port = arguments['port']
     messenger = _Messenger(task_id=task_id, port=port)
-    messenger.started()
 
     implementation = arguments['implementation']
     operation_inputs = arguments['operation_inputs']
@@ -390,6 +394,7 @@ def _main():
 
     with instrumentation.track_changes() as instrument:
         try:
+            messenger.started()
             ctx = context_dict['context_cls'].deserialize_from_dict(**context_dict['context'])
             _patch_session(ctx=ctx, messenger=messenger, instrument=instrument)
             task_func = imports.load_attribute(implementation)