hopefully fixed issue with raising errors from cf worker

Author: tclose

pydra/engine/tests/test_task.py

@@ -1344,7 +1344,7 @@ def FunError(x):
     assert "in FunError" in error_tb[-2]
 
 
-def test_traceback_wf(tmpdir):
+def test_traceback_wf(plugin_parallel: str, tmp_path: Path):
     """checking if the error raised in a function is properly returned by a workflow;
     checking if there is an error filename in the error message that contains
     full traceback including the line in the python function
@@ -1361,7 +1361,7 @@ def Workflow(x_list):
 
     wf = Workflow(x_list=[3, 4])
     with pytest.raises(RuntimeError, match="Task 'fun_error' failed.*") as exinfo:
-        with Submitter(worker="cf", cache_dir=tmpdir) as sub:
+        with Submitter(worker=plugin_parallel, cache_dir=tmp_path) as sub:
             sub(wf, raise_errors=True)
 
     # getting error file from the error message

pydra/engine/workers.py

@@ -36,6 +36,7 @@ class Worker(metaclass=abc.ABCMeta):
     """A base class for execution of tasks."""
 
     plugin_name: str
+    loop: asyncio.AbstractEventLoop
 
     def __init__(self, loop=None):
         """Initialize the worker."""
@@ -48,10 +49,11 @@ def run(self, task: "Task[DefType]", rerun: bool = False) -> "Result":
         pass
 
     async def run_async(self, task: "Task[DefType]", rerun: bool = False) -> "Result":
-        if task.is_async:
+        if task.is_async:  # only for workflows at this stage and the foreseeable
+            # These jobs are run in the primary process but farm out the workflows jobs
             return await task.run_async(rerun=rerun)
         else:
-            return task.run(rerun=rerun)
+            return self.run(task=task, rerun=rerun)
 
     def close(self):
         """Close this worker."""
@@ -175,7 +177,8 @@ class ConcurrentFuturesWorker(Worker):
     plugin_name = "cf"
 
     n_procs: int
-    loop: cf.ProcessPoolExecutor
+    loop: asyncio.AbstractEventLoop
+    pool: cf.ProcessPoolExecutor
 
     def __init__(self, n_procs: int | None = None):
         """Initialize Worker."""
@@ -193,13 +196,12 @@ async def run(
     ) -> "Result":
         """Run a task."""
         assert self.loop, "No event loop available to submit tasks"
-        task_pkl = cp.dumps(task)
         return await self.loop.run_in_executor(
-            self.pool, self.unpickle_and_run, task_pkl, rerun
+            self.pool, self.uncloudpickle_and_run, cp.dumps(task), rerun
         )
 
     @classmethod
-    def unpickle_and_run(cls, task_pkl: Path, rerun: bool) -> "Result":
+    def uncloudpickle_and_run(cls, task_pkl: bytes, rerun: bool) -> "Result":
         """Unpickle and run a task."""
         task: Task[DefType] = cp.loads(task_pkl)
         return task.run(rerun=rerun)