Merge pull request #1 from arago/remove_deadlock

[bugfix] Avoid read-in deadlock

Author: tryptichon

CHANGELOG.md

@@ -1,3 +1,11 @@
+# v1.0.1
+
+[bugfix] Avoid read-in deadlock
+
+* When an error is thrown while data is read in, the queues
+  within HiroGraphBatch need to be closed cleanly or a
+  deadlock will occur. Fixed.
+
 # v1.0.0
 
 Created project by splitting `batchclient.py` from [hiro-graph-client](https://github.com/arago/hiro-client-python).

src/hiro_batch_client/VERSION

@@ -1 +1 @@
-1.0.0.dev0
+1.0.1

src/hiro_batch_client/batchclient.py

@@ -4,10 +4,9 @@
 from enum import Enum
 from typing import Optional, Tuple, Any, Iterator, IO
 
-from requests.exceptions import RequestException
-
 from hiro_graph_client.client import HiroGraph
 from hiro_graph_client.clientlib import AbstractTokenApiHandler
+from requests.exceptions import RequestException
 
 
 class Result(Enum):
@@ -491,7 +490,8 @@ def error_message(entity: Entity,
                       action: Action,
                       error: Exception,
                       original: dict,
-                      status_code: int = None) -> dict:
+                      status_code: int = None,
+                      interrupted: bool = None) -> dict:
         """
         Failure message format
 
@@ -513,16 +513,23 @@ def error_message(entity: Entity,
         :param error: The exception raised
         :param original: The data that lead to the exception
         :param status_code: HTTP status code if available
+        :param interrupted: Indicates, that the current batch processing has been interrupted.
         :return: The message
         """
+
+        message = str(error)
+        if interrupted:
+            message = "BATCH PROCESSING ABORTED! " + message + \
+                      " All further data has been ignored after this error occurred."
+
         return {
             "status": Result.FAILURE.value,
             "entity": entity.value,
             "action": action.value,
             "data": {
                 "error": error.__class__.__name__,
                 "code": status_code,
-                "message": str(error),
+                "message": message,
                 "original_data": original
             }
         }
@@ -1037,67 +1044,85 @@ def _request_queue_put(_command: str,
                 self.request_queue.put((_command, _attributes))
                 return _parallel_workers
 
-            session = SessionData(self.use_xid_cache)
+            parallel_workers = 0
 
-            collected_results = [] if self.callback is None else None
+            try:
+                session = SessionData(self.use_xid_cache)
 
-            parallel_workers = 0
+                collected_results = [] if self.callback is None else None
 
-            executor.submit(HiroGraphBatch._reader, self, collected_results)
+                executor.submit(HiroGraphBatch._reader, self, collected_results)
 
-            handle_session_data = False
-            for command_entry in command_iter:
-                for command, attributes in command_entry.items():
+                handle_session_data = False
+                for command_entry in command_iter:
+                    for command, attributes in command_entry.items():
 
-                    if command == "handle_vertices_combined":
-                        command = "handle_vertices"
-                        handle_session_data = True
+                        if command == "handle_vertices_combined":
+                            command = "handle_vertices"
+                            handle_session_data = True
 
-                    try:
-                        if command not in self.command_map:
-                            raise SourceValueError("No such command \"{}\".".format(command))
+                        try:
+                            if command not in self.command_map:
+                                raise SourceValueError("No such command \"{}\".".format(command))
 
-                        if isinstance(attributes, list):
-                            for attribute_entry in attributes:
+                            if isinstance(attributes, list):
+                                for attribute_entry in attributes:
+                                    parallel_workers = _request_queue_put(
+                                        command,
+                                        attribute_entry,
+                                        parallel_workers)
+                            else:
                                 parallel_workers = _request_queue_put(
                                     command,
-                                    attribute_entry,
+                                    attributes,
                                     parallel_workers)
-                        else:
-                            parallel_workers = _request_queue_put(
-                                command,
-                                attributes,
-                                parallel_workers)
-
-                    except SourceValueError as err:
-                        sub_result, sub_code = HiroBatchRunner.error_message(
-                            Entity.UNDEFINED,
-                            Action.UNDEFINED,
-                            err,
-                            attributes,
-                            400), 400
-
-                        self.result_queue.put((sub_result, sub_code))
-
-            if handle_session_data:
-                self.request_queue.join()
-                self._edges_from_session(session)
-                self._timeseries_from_session(session)
-                self._attachments_from_session(session)
-                # Ensure, that all data related to the original vertex has been sent
-                # before creating any issues.
-                self.request_queue.join()
-                self._issues_from_session(session)
 
-            self.request_queue.join()
-            self.result_queue.join()
-
-            for _ in range(parallel_workers):
-                self.request_queue.put(None)
+                            # Empty old attributes, so they do not show up on exceptions
+                            # that might be thrown before the new attributes have been
+                            # read (i.e. read-in-exceptions while iterating over *command_iter*).
+                            attributes = None
+
+                        except SourceValueError as err:
+                            sub_result, sub_code = HiroBatchRunner.error_message(
+                                entity=Entity.UNDEFINED,
+                                action=Action.UNDEFINED,
+                                error=err,
+                                original=attributes,
+                                status_code=400), 400
+
+                            self.result_queue.put((sub_result, sub_code))
+
+                if handle_session_data:
+                    self.request_queue.join()
+                    self._edges_from_session(session)
+                    self._timeseries_from_session(session)
+                    self._attachments_from_session(session)
+                    # Ensure, that all data related to the original vertex has been sent
+                    # before creating any issues.
+                    self.request_queue.join()
+                    self._issues_from_session(session)
+
+                return collected_results
+
+            except Exception as err:
+                sub_result, sub_code = HiroBatchRunner.error_message(
+                    entity=Entity.UNDEFINED,
+                    action=Action.UNDEFINED,
+                    error=err,
+                    original=attributes,
+                    status_code=400,
+                    interrupted=True), 400
+
+                self.result_queue.put((sub_result, sub_code))
+
+            finally:
+                self.request_queue.join()
+                self.result_queue.join()
 
-            self.result_queue.put(None)
+                for _ in range(parallel_workers):
+                    self.request_queue.put(None)
 
-            return collected_results
+                self.result_queue.put(None)
 
 
 class SourceValueError(ValueError):