Define error log file in resource_dict

Author: jan-janssen

.github/workflows/pipeline.yml

@@ -402,7 +402,7 @@ jobs:
       run: echo -e "channels:\n  - conda-forge\n" > .condarc
     - uses: conda-incubator/setup-miniconda@v3
       with:
-        python-version: '3.9'
+        python-version: '3.10'
         miniforge-version: latest
         condarc-file: .condarc
         environment-file: .ci_support/environment-old.yml

executorlib/backend/cache_parallel.py

@@ -4,6 +4,7 @@
 
 import cloudpickle
 
+from executorlib.standalone.error import backend_write_error_file
 from executorlib.task_scheduler.file.backend import (
     backend_load_file,
     backend_write_file,
@@ -53,6 +54,10 @@ def main() -> None:
                 output={"error": error},
                 runtime=time.time() - time_start,
             )
+            backend_write_error_file(
+                error=error,
+                apply_dict=apply_dict,
+            )
     else:
         if mpi_rank_zero:
             backend_write_file(

executorlib/backend/interactive_parallel.py

@@ -6,6 +6,7 @@
 import cloudpickle
 import zmq
 
+from executorlib.standalone.error import backend_write_error_file
 from executorlib.standalone.interactive.backend import call_funct, parse_arguments
 from executorlib.standalone.interactive.communication import (
     interface_connect,
@@ -82,6 +83,10 @@ def main() -> None:
                         socket=socket,
                         result_dict={"error": error},
                     )
+                    backend_write_error_file(
+                        error=error,
+                        apply_dict=input_dict,
+                    )
             else:
                 # Send output
                 if mpi_rank_zero:

executorlib/backend/interactive_serial.py

@@ -2,6 +2,7 @@
 from os.path import abspath
 from typing import Optional
 
+from executorlib.standalone.error import backend_write_error_file
 from executorlib.standalone.interactive.backend import call_funct, parse_arguments
 from executorlib.standalone.interactive.communication import (
     interface_connect,
@@ -58,6 +59,10 @@ def main(argument_lst: Optional[list[str]] = None):
                     socket=socket,
                     result_dict={"error": error},
                 )
+                backend_write_error_file(
+                    error=error,
+                    apply_dict=input_dict,
+                )
             else:
                 # Send output
                 interface_send(socket=socket, result_dict={"result": output})

executorlib/standalone/error.py

@@ -0,0 +1,22 @@
+
+import traceback
+
+
+def backend_write_error_file(error: Exception, apply_dict: dict) -> None:
+    """
+    Write an error to a file if specified in the apply_dict.
+
+    Args:
+        error (Exception): The error to be written.
+        apply_dict (dict): Dictionary containing additional parameters.
+
+    Returns:
+        None
+    """
+    error_log_file = apply_dict.get("error_log_file", None)
+    if error_log_file is not None:
+        with open(error_log_file, "a") as f:
+            f.write("function: " + str(apply_dict["fn"]) + "\n")
+            f.write("args: " + str(apply_dict["args"]) + "\n")
+            f.write("kwargs: " + str(apply_dict["kwargs"]) + "\n")
+            traceback.print_exception(error, file=f)

executorlib/task_scheduler/file/backend.py

@@ -2,6 +2,7 @@
 import time
 from typing import Any
 
+from executorlib.standalone.error import backend_write_error_file
 from executorlib.task_scheduler.file.hdf import dump, load
 from executorlib.task_scheduler.file.shared import FutureItem
 
@@ -77,6 +78,10 @@ def backend_execute_task_in_file(file_name: str) -> None:
         }
     except Exception as error:
         result = {"error": error}
+        backend_write_error_file(
+            error=error,
+            apply_dict=apply_dict,
+        )
 
     backend_write_file(
         file_name=file_name,

executorlib/task_scheduler/file/hdf.py

@@ -52,6 +52,10 @@ def load(file_name: str) -> dict:
             data_dict["kwargs"] = cloudpickle.loads(np.void(hdf["/input_kwargs"]))
         else:
             data_dict["kwargs"] = {}
+        if "error_log_file" in hdf:
+            data_dict["error_log_file"] = cloudpickle.loads(
+                np.void(hdf["/error_log_file"])
+            )
         return data_dict
 
 

executorlib/task_scheduler/file/shared.py

@@ -126,13 +126,15 @@ def execute_tasks_h5(
             )
             cache_key = task_resource_dict.pop("cache_key", None)
             cache_directory = os.path.abspath(task_resource_dict.pop("cache_directory"))
+            error_log_file = task_resource_dict.pop("error_log_file", None)
             task_key, data_dict = serialize_funct_h5(
                 fn=task_dict["fn"],
                 fn_args=task_args,
                 fn_kwargs=task_kwargs,
                 resource_dict=task_resource_dict,
                 cache_key=cache_key,
             )
+            data_dict["error_log_file"] = error_log_file
             if task_key not in memory_dict:
                 if os.path.join(
                     cache_directory, task_key + "_o.h5"

executorlib/task_scheduler/interactive/shared.py

@@ -26,6 +26,7 @@ def execute_tasks(
     cache_key: Optional[str] = None,
     queue_join_on_shutdown: bool = True,
     log_obj_size: bool = False,
+    error_log_file: Optional[str] = None,
     **kwargs,
 ) -> None:
     """
@@ -70,6 +71,8 @@ def execute_tasks(
                 future_queue.join()
             break
         elif "fn" in task_dict and "future" in task_dict:
+            if error_log_file is not None:
+                task_dict["error_log_file"] = error_log_file
             if cache_directory is None:
                 _execute_task_without_cache(
                     interface=interface, task_dict=task_dict, future_queue=future_queue

pyproject.toml

@@ -23,7 +23,6 @@ classifiers = [
     "License :: OSI Approved :: BSD License",
     "Intended Audience :: Science/Research",
     "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",