From 58a3870de0f631eccf8ce6822a3d05661971df52 Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 18:26:37 +0200 Subject: [PATCH 1/5] [Refactor] receive_dict() returns the status and the error as dict --- .../standalone/interactive/communication.py | 8 ++---- .../interactive/blockallocation.py | 11 ++++---- .../task_scheduler/interactive/shared.py | 25 ++++++++++--------- .../interactive/test_communication.py | 8 +++--- 4 files changed, 24 insertions(+), 28 deletions(-) diff --git a/src/executorlib/standalone/interactive/communication.py b/src/executorlib/standalone/interactive/communication.py index 665d71f91..c10d38c59 100644 --- a/src/executorlib/standalone/interactive/communication.py +++ b/src/executorlib/standalone/interactive/communication.py @@ -86,11 +86,7 @@ def receive_dict(self) -> dict: self._logger.warning( "Received dictionary of size: " + str(sys.getsizeof(data)) ) - output = cloudpickle.loads(data) - if "result" in output: - return output["result"] - else: - raise output["error"] + return cloudpickle.loads(data) def send_and_receive_dict(self, input_dict: dict) -> dict: """ @@ -154,7 +150,7 @@ def shutdown(self, wait: bool = True): if self._spawner.poll(): result = self.send_and_receive_dict( input_dict={"shutdown": True, "wait": wait} - ) + )["result"] self._spawner.shutdown(wait=wait) self._reset_socket() return result diff --git a/src/executorlib/task_scheduler/interactive/blockallocation.py b/src/executorlib/task_scheduler/interactive/blockallocation.py index 1f7474dfe..02501e786 100644 --- a/src/executorlib/task_scheduler/interactive/blockallocation.py +++ b/src/executorlib/task_scheduler/interactive/blockallocation.py @@ -379,10 +379,9 @@ def _set_init_function( ) -> Optional[Exception]: interface_initialization_exception = None if init_function is not None and interface.status: - try: - _ = interface.send_and_receive_dict( - input_dict={"init": True, "fn": init_function, "args": (), "kwargs": {}} - ) - except Exception as init_exception: - interface_initialization_exception = init_exception + output = interface.send_and_receive_dict( + input_dict={"init": True, "fn": init_function, "args": (), "kwargs": {}} + ) + if "error" in output: + interface_initialization_exception = output["error"] return interface_initialization_exception diff --git a/src/executorlib/task_scheduler/interactive/shared.py b/src/executorlib/task_scheduler/interactive/shared.py index e40842224..1ada2508a 100644 --- a/src/executorlib/task_scheduler/interactive/shared.py +++ b/src/executorlib/task_scheduler/interactive/shared.py @@ -101,13 +101,14 @@ def _execute_task_without_cache( Returns: bool: True if the task was submitted successfully, False otherwise. """ - try: - future_obj.set_result(interface.send_and_receive_dict(input_dict=task_dict)) - except Exception as thread_exception: - if isinstance(thread_exception, ExecutorlibSocketError): + output = interface.send_and_receive_dict(input_dict=task_dict) + if "result" in output: + future_obj.set_result(output["result"]) + else: + if isinstance(output["error"], ExecutorlibSocketError): return False else: - future_obj.set_exception(exception=thread_exception) + future_obj.set_exception(exception=output["error"]) return True @@ -141,18 +142,18 @@ def _execute_task_with_cache( ) file_name = os.path.abspath(os.path.join(cache_directory, task_key + "_o.h5")) if file_name not in get_cache_files(cache_directory=cache_directory): - try: - time_start = time.time() - result = interface.send_and_receive_dict(input_dict=task_dict) - data_dict["output"] = result + time_start = time.time() + output = interface.send_and_receive_dict(input_dict=task_dict) + if "result" in output: + data_dict["output"] = output["result"] data_dict["runtime"] = time.time() - time_start dump(file_name=file_name, data_dict=data_dict) future_obj.set_result(result) - except Exception as thread_exception: - if isinstance(thread_exception, ExecutorlibSocketError): + else: + if isinstance(output["error"], ExecutorlibSocketError): return False else: - future_obj.set_exception(exception=thread_exception) + future_obj.set_exception(exception=output["error"]) else: _, _, result = get_output(file_name=file_name) future_obj.set_result(result) diff --git a/tests/unit/standalone/interactive/test_communication.py b/tests/unit/standalone/interactive/test_communication.py index 4ed5f35ad..e93e4eea9 100644 --- a/tests/unit/standalone/interactive/test_communication.py +++ b/tests/unit/standalone/interactive/test_communication.py @@ -63,7 +63,7 @@ def test_interface_mpi(self): ) self.assertTrue(interface.status) self.assertEqual( - interface.send_and_receive_dict(input_dict=task_dict), np.array(4) + interface.send_and_receive_dict(input_dict=task_dict)["result"], np.array(4) ) interface.shutdown(wait=True) @@ -96,7 +96,7 @@ def test_interface_serial_without_debug(self): ) self.assertTrue(interface.status) self.assertEqual( - interface.send_and_receive_dict(input_dict=task_dict), np.array(4) + interface.send_and_receive_dict(input_dict=task_dict)["result"], np.array(4) ) interface.shutdown(wait=True) @@ -129,7 +129,7 @@ def test_interface_serial_with_debug(self): ) self.assertTrue(interface.status) self.assertEqual( - interface.send_and_receive_dict(input_dict=task_dict), np.array(4) + interface.send_and_receive_dict(input_dict=task_dict)["result"], np.array(4) ) interface.shutdown(wait=True) @@ -195,7 +195,7 @@ def test_interface_serial_with_stopped_process(self): interface.send_dict(input_dict=task_dict) interface._spawner._process.terminate() with self.assertRaises(ExecutorlibSocketError): - interface.receive_dict() + raise interface.receive_dict()["error"] class TestZMQ(unittest.TestCase): From ca6422e95742319e7520ca19b6df80823e90ffce Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 18:28:03 +0200 Subject: [PATCH 2/5] fix test --- tests/unit/standalone/interactive/test_communication.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/standalone/interactive/test_communication.py b/tests/unit/standalone/interactive/test_communication.py index e93e4eea9..6c0832d47 100644 --- a/tests/unit/standalone/interactive/test_communication.py +++ b/tests/unit/standalone/interactive/test_communication.py @@ -194,8 +194,8 @@ def test_interface_serial_with_stopped_process(self): self.assertTrue(interface.status) interface.send_dict(input_dict=task_dict) interface._spawner._process.terminate() - with self.assertRaises(ExecutorlibSocketError): - raise interface.receive_dict()["error"] + output = interface.receive_dict() + self.assertIsInstance(output["error"], ExecutorlibSocketError) class TestZMQ(unittest.TestCase): From b1a8e85f6de6cb76d058a32e609134d47b67f713 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 3 Jul 2026 17:07:52 +0000 Subject: [PATCH 3/5] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- .../task_scheduler/interactive/shared.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/src/executorlib/task_scheduler/interactive/shared.py b/src/executorlib/task_scheduler/interactive/shared.py index 1ada2508a..0450af29a 100644 --- a/src/executorlib/task_scheduler/interactive/shared.py +++ b/src/executorlib/task_scheduler/interactive/shared.py @@ -104,11 +104,10 @@ def _execute_task_without_cache( output = interface.send_and_receive_dict(input_dict=task_dict) if "result" in output: future_obj.set_result(output["result"]) + elif isinstance(output["error"], ExecutorlibSocketError): + return False else: - if isinstance(output["error"], ExecutorlibSocketError): - return False - else: - future_obj.set_exception(exception=output["error"]) + future_obj.set_exception(exception=output["error"]) return True @@ -149,11 +148,10 @@ def _execute_task_with_cache( data_dict["runtime"] = time.time() - time_start dump(file_name=file_name, data_dict=data_dict) future_obj.set_result(result) + elif isinstance(output["error"], ExecutorlibSocketError): + return False else: - if isinstance(output["error"], ExecutorlibSocketError): - return False - else: - future_obj.set_exception(exception=output["error"]) + future_obj.set_exception(exception=output["error"]) else: _, _, result = get_output(file_name=file_name) future_obj.set_result(result) From 7e97e10432314dbfc5870750908f94b7b2d4b951 Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 19:10:40 +0200 Subject: [PATCH 4/5] Fix ruff --- src/executorlib/task_scheduler/interactive/shared.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/executorlib/task_scheduler/interactive/shared.py b/src/executorlib/task_scheduler/interactive/shared.py index 0450af29a..badca5eed 100644 --- a/src/executorlib/task_scheduler/interactive/shared.py +++ b/src/executorlib/task_scheduler/interactive/shared.py @@ -147,7 +147,7 @@ def _execute_task_with_cache( data_dict["output"] = output["result"] data_dict["runtime"] = time.time() - time_start dump(file_name=file_name, data_dict=data_dict) - future_obj.set_result(result) + future_obj.set_result(output["result"]) elif isinstance(output["error"], ExecutorlibSocketError): return False else: From d1d970529a199d1040105ad9b888ff039362b927 Mon Sep 17 00:00:00 2001 From: Jan Janssen Date: Fri, 3 Jul 2026 20:02:32 +0200 Subject: [PATCH 5/5] return error rather than crashing --- src/executorlib/standalone/interactive/communication.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/executorlib/standalone/interactive/communication.py b/src/executorlib/standalone/interactive/communication.py index c10d38c59..a5623dd2c 100644 --- a/src/executorlib/standalone/interactive/communication.py +++ b/src/executorlib/standalone/interactive/communication.py @@ -78,9 +78,11 @@ def receive_dict(self) -> dict: while len(response_lst) == 0: response_lst = self._poller.poll(self._time_out_ms) if not self._spawner.poll(): - raise ExecutorlibSocketError( - "SocketInterface crashed during execution." - ) + return { + "error": ExecutorlibSocketError( + "SocketInterface crashed during execution." + ), + } data = self._socket.recv(zmq.NOBLOCK) if self._logger is not None: self._logger.warning(