got concurrent futures worker to work

Author: tclose

new-docs/source/tutorial/tst.py

@@ -1,16 +1,16 @@
 from pydra.design import python
 
+if __name__ == "__main__":
 
-@python.define
-def TenToThePower(p: int) -> int:
-    return 10**p
+    @python.define
+    def TenToThePower(p: int) -> int:
+        return 10**p
 
+    ten_to_the_power = TenToThePower().split(p=[1, 2, 3, 4, 5])
 
-ten_to_the_power = TenToThePower().split(p=[1, 2, 3, 4, 5])
+    # Run the 5 tasks in parallel split across 3 processes
+    outputs = ten_to_the_power(worker="cf", n_procs=3)
 
-# Run the 5 tasks in parallel split across 3 processes
-outputs = ten_to_the_power(worker="debug")
+    p1, p2, p3, p4, p5 = outputs.out
 
-p1, p2, p3, p4, p5 = outputs.out
-
-print(f"10^5 = {p5}")
+    print(f"10^5 = {p5}")

pydra/engine/core.py

@@ -325,6 +325,8 @@ def _populate_filesystem(self):
         if not self.can_resume and self.output_dir.exists():
             shutil.rmtree(self.output_dir)
         self.output_dir.mkdir(parents=False, exist_ok=self.can_resume)
+        # Save task pkl into the output directory for future reference
+        save(self.output_dir, task=self)
 
     def run(self, rerun: bool = False):
         """Prepare the task working directory, execute the task definition, and save the
@@ -382,7 +384,7 @@ def run(self, rerun: bool = False):
         self._check_for_hash_changes()
         return result
 
-    async def run_async(self, rerun: bool = False):
+    async def run_async(self, rerun: bool = False) -> Result:
         """Prepare the task working directory, execute the task definition asynchronously,
         and save the results. NB: only workflows are run asynchronously at the moment.
 

pydra/engine/helpers.py

@@ -463,7 +463,7 @@ def load_and_run(task_pkl: Path, rerun: bool = False) -> Path:
 
     resultfile = task.output_dir / "_result.pklz"
     try:
-        task(rerun=rerun)
+        task.run(rerun=rerun)
     except Exception as e:
         # creating result and error files if missing
         errorfile = task.output_dir / "_error.pklz"
@@ -479,16 +479,16 @@ def load_and_run(task_pkl: Path, rerun: bool = False) -> Path:
     return resultfile
 
 
-async def load_and_run_async(task_pkl):
-    """
-    loading a task from a pickle file, settings proper input
-    and running the workflow
-    """
-    task = load_task(task_pkl=task_pkl)
-    await task()
+# async def load_and_run_async(task_pkl):
+#     """
+#     loading a task from a pickle file, settings proper input
+#     and running the workflow
+#     """
+#     task = load_task(task_pkl=task_pkl)
+#     await task()
 
 
-def load_task(task_pkl):
+def load_task(task_pkl: Path | str) -> "Task[DefType]":
     """loading a task from a pickle file, settings proper input for the specific ind"""
     if isinstance(task_pkl, str):
         task_pkl = Path(task_pkl)

pydra/engine/specs.py

@@ -876,6 +876,11 @@ def _resolve_value(
 
 class ShellDef(TaskDef[ShellOutputsType]):
 
+    arguments: ty.List[str] = shell.arg(
+        default=attrs.Factory(list),
+        help="Additional arguments to pass to the command.",
+    )
+
     RESERVED_FIELD_NAMES = TaskDef.RESERVED_FIELD_NAMES + ("cmdline",)
 
     def _run(self, task: "Task[ShellDef]") -> None:
@@ -952,7 +957,9 @@ def _command_args(
         # Sort command and arguments by position
         cmd_args = position_sort(pos_args)
         # pos_args values are each a list of arguments, so concatenate lists after sorting
-        return sum(cmd_args, [])
+        command_args = sum(cmd_args, [])
+        command_args += self.arguments
+        return command_args
 
     def _command_shelltask_executable(
         self, field: shell.arg, value: ty.Any

pydra/engine/submitter.py

@@ -9,7 +9,6 @@
 from copy import copy
 from collections import defaultdict
 from .workers import Worker, WORKERS
-from .core import is_workflow
 from .graph import DiGraph
 from .helpers import (
     get_open_loop,
@@ -153,10 +152,10 @@ def Split(defn: TaskDef) -> tuple:
                 "Use the `split` method to split the task before combining."
             )
         task = Task(task_def, submitter=self, name="task", environment=self.environment)
-        if task.is_async:
-            self.loop.run_until_complete(task.run_async(rerun=self.rerun))
+        if task.is_async:  # Only workflow tasks can be async
+            self.loop.run_until_complete(self.worker.run_async(task, rerun=self.rerun))
         else:
-            task.run(rerun=self.rerun)
+            self.worker.run(rerun=self.rerun)
         PersistentCache().clean_up()
         result = task.result()
         if result is None:
@@ -285,8 +284,8 @@ async def expand_workflow_async(self, workflow_task: "Task[WorkflowDef]") -> Non
                             )
                         raise RuntimeError(msg)
             for task in tasks:
-                if is_workflow(task):
-                    await task.run_async(rerun=self.rerun)
+                if task.is_async:
+                    await self.worker.run_async(task, rerun=self.rerun)
                 else:
                     task_futures.add(self.worker.run(task, rerun=self.rerun))
             task_futures = await self.worker.fetch_finished(task_futures)