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.
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 @@
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 @@
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 @@
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)
