[execution][local] Improve error handling on algorithm execution

Currently, when an error occurred in one of the scripts, it
could just hang or the error itself wasn't properly processed.

This patch improves exception handling as well as error
message returned.

beat/core/execution/local.py

@@ -188,8 +188,12 @@ class LocalExecutor(BaseExecutor):
 
         self.zmq_context = None
 
-    def __cleanup(self):
+    def __cleanup(self, early=False):
         if self.loop_executor:
+            if early:
+                self.loop_socket.send_string("don")
+                self.loop_socket.recv()  # ack
+
             self.loop_executor.wait()
             self.loop_executor.close()
 
@@ -311,19 +315,35 @@ class LocalExecutor(BaseExecutor):
                 cache_root=self.cache,
             )
 
-            retval = self.loop_executor.setup()
+            try:
+                retval = self.loop_executor.setup()
+            except Exception as e:
+                message = _process_exception(e, self.prefix, "algorithms")
+                retval = False
+            else:
+                message = None
+
             if not retval:
                 self.__cleanup()
-                raise RuntimeError(
-                    "Loop algorithm {} setup failed".format(self.algorithm.name)
-                )
+                error = "Loop algorithm {} setup failed".format(self.algorithm.name)
+                if message:
+                    error += ": {}".format(message)
+                raise RuntimeError(error)
+
+            try:
+                prepared = self.loop_executor.prepare()
+            except Exception as e:
+                message = _process_exception(e, self.prefix, "algorithms")
+                prepared = False
+            else:
+                message = None
 
-            prepared = self.loop_executor.prepare()
             if not prepared:
                 self.__cleanup()
-                raise RuntimeError(
-                    "Loop algorithm {} prepare failed".format(self.algorithm.name)
-                )
+                error = "Loop algorithm {} prepare failed".format(self.algorithm.name)
+                if message:
+                    error += ": {}".format(message)
+                raise RuntimeError(error)
 
             self.loop_executor.process()
 
@@ -337,26 +357,40 @@ class LocalExecutor(BaseExecutor):
 
         try:
             status = self.executor.setup()
-        except Exception:
+        except Exception as e:
+            message = _process_exception(e, self.prefix, "algorithms")
             status = 0
+        else:
+            message = None
 
         if not status:
-            self.__cleanup()
-            raise RuntimeError("Algorithm {} setup failed".format(self.algorithm.name))
+            self.__cleanup(early=True)
+            error = "Algorithm {} setup failed".format(self.algorithm.name)
+            if message:
+                error += ": {}".format(message)
+            raise RuntimeError(error)
+
+        try:
+            prepared = self.executor.prepare()
+        except Exception as e:
+            message = _process_exception(e, self.prefix, "algorithms")
+            prepared = 0
+        else:
+            message = None
 
-        prepared = self.executor.prepare()
         if not prepared:
-            self.__cleanup()
-            raise RuntimeError(
-                "Algorithm {} prepare failed".format(self.algorithm.name)
-            )
+            self.__cleanup(early=True)
+            error = "Algorithm {} prepare failed".format(self.algorithm.name)
+            if message:
+                error += ": {}".format(message)
+            raise RuntimeError(error)
 
         _start = time.time()
 
         try:
             processed = self.executor.process()
         except Exception as e:
-            message = _process_exception(e, self.prefix, "databases")
+            message = _process_exception(e, self.prefix, "algorithms")
             self.__cleanup()
             return _create_result(1, message)