debugging new syntax in unittests

Author: tclose

new-docs/source/tutorial/3-troubleshooting.ipynb

@@ -6,8 +6,8 @@
    "source": [
     "# Troubleshooting\n",
     "\n",
-    "This tutorial steps through tecnhiques to identify errors and pipeline failures, and\n",
-    "avoid common pitfalls."
+    "This tutorial steps through tecnhiques to identify errors and pipeline failures, as well\n",
+    "as avoid common pitfalls setting up executing over multiple processes."
    ]
   },
   {
@@ -45,9 +45,9 @@
    "source": [
     "### Enclosing multi-process code within `if __name__ == \"__main__\"`\n",
     "\n",
-    "If running a script that executes a workflow with the concurrent futures worker\n",
-    "(i.e. `worker=\"cf\"`) on macOS or Windows, then the submissing/execution call needs to\n",
-    "be enclosed within a `if __name__ == \"__main__\"` blocks, e.g."
+    "When running multi-process Python code on macOS or Windows, as is the case when the \n",
+    "concurrent futures worker is selected (i.e. `worker=\"cf\"`), then scripts that execute\n",
+    "the forking code need to be enclosed within an `if __name__ == \"__main__\"` block, e.g."
    ]
   },
   {
@@ -71,12 +71,24 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "### Remove stray lockfiles\n",
+    "This allows the secondary processes to import the script without executing it. Without\n",
+    "such a block Pydra will lock up and not process the workflow. On Linux this is not an\n",
+    "issue due to the way that processes are forked, but is good practice in any case for\n",
+    "code portability."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Removing stray lockfiles\n",
     "\n",
-    "During the execution of a task, a lockfile is generated to signify that a task is running.\n",
-    "These lockfiles are released after a task completes, either successfully or with an error,\n",
-    "within a *try/finally* block. However, if a task/workflow is terminated by an interactive\n",
-    "debugger the finally block may not be executed causing stray lockfiles to hang around. This\n",
+    "When a Pydra task is executed, a lockfile is generated to signify that the task is running.\n",
+    "Other processes will wait for this lock to be released before attempting to access the\n",
+    "tasks results. The lockfiles are automatically deleted after a task completes, either\n",
+    "successfully or with an error, within a *try/finally* block so should run most of the time.\n",
+    "However, if a task/workflow is terminated by an interactive\n",
+    "debugger, the finally block may not be executed, leaving stray lockfiles. This\n",
     "can cause the Pydra to hang waiting for the lock to be released. If you suspect this to be\n",
     "an issue, and there are no other jobs running, then simply remove all lock files from your\n",
     "cache directory (e.g. `rm <your-run-cache-dir>/*.lock`) and re-submit your job.\n",
@@ -91,7 +103,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## Finding errors\n",
+    "## Inspecting errors\n",
     "\n",
     "### Running in *debug* mode\n",
     "\n",
@@ -116,8 +128,9 @@
     "# This workflow will fail because we are trying to divide by 0\n",
     "wf = UnsafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2 ,0])\n",
     "\n",
-    "with Submitter(worker=\"cf\") as sub:\n",
-    "    result = sub(wf)\n",
+    "if __name__ == \"__main__\":\n",
+    "    with Submitter(worker=\"cf\") as sub:\n",
+    "        result = sub(wf)\n",
     "    \n",
     "if result.errored:\n",
     "    print(\"Workflow failed with errors:\\n\" + str(result.errors))\n",
@@ -129,7 +142,23 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "Work in progress..."
+    "The error pickle files can be loaded using the `cloudpickle` library, noting that it is\n",
+    "important to use the same Python version to load the files that was used to run the Pydra\n",
+    "workflow"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import cloudpickle as cp\n",
+    "\n",
+    "with open(\"<your-cache-root>/<task-cache-dir/_error.pklz\", \"rb\") as f:\n",
+    "    error = cp.load(f)\n",
+    "\n",
+    "print(error)"
    ]
   },
   {
@@ -147,31 +176,69 @@
     "Currently in Pydra you need to step backwards through the tasks of the workflow, load\n",
     "the saved task object and inspect its inputs to find the preceding nodes. If any of the\n",
     "inputs that have been generated by previous nodes are not ok, then you should check the\n",
-    "tasks that generated them in turn.\n",
+    "tasks that generated them in turn. For file-based inputs, you should be able to find\n",
+    "the path of the preceding task's cache directory from the provided file path. However,\n",
+    "for non-file inputs you may need to exhaustively iterate through all the task dirs\n",
+    "in your cache root to find the issue.\n",
     "\n",
-    "For example, in the following example if we are not happy with the mask brain that has\n",
-    "been generated, we can check the mask to see whether it looks sensible by first loading\n",
-    "the apply mask task and then inspecting its inputs."
+    "For example, in the following example workflow, if a divide by 0 occurs within the division\n",
+    "node of the workflow, then an `float('inf')` will be returned, which will then propagate\n",
+    "through the workflow."
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 2,
    "metadata": {},
-   "outputs": [],
-   "source": []
+   "outputs": [
+    {
+     "ename": "NameError",
+     "evalue": "name 'Submitter' is not defined",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
+      "Cell \u001b[0;32mIn[2], line 5\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[38;5;28;01mfrom\u001b[39;00m \u001b[38;5;21;01mpydra\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mtasks\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mtesting\u001b[39;00m \u001b[38;5;28;01mimport\u001b[39;00m SafeDivisionWorkflow\n\u001b[1;32m      3\u001b[0m wf \u001b[38;5;241m=\u001b[39m SafeDivisionWorkflow(a\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m10\u001b[39m, b\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m5\u001b[39m)\u001b[38;5;241m.\u001b[39msplit(denominator\u001b[38;5;241m=\u001b[39m[\u001b[38;5;241m3\u001b[39m, \u001b[38;5;241m2\u001b[39m ,\u001b[38;5;241m0\u001b[39m])\n\u001b[0;32m----> 5\u001b[0m \u001b[38;5;28;01mwith\u001b[39;00m \u001b[43mSubmitter\u001b[49m(worker\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mcf\u001b[39m\u001b[38;5;124m\"\u001b[39m) \u001b[38;5;28;01mas\u001b[39;00m sub:\n\u001b[1;32m      6\u001b[0m     result \u001b[38;5;241m=\u001b[39m sub(wf)\n\u001b[1;32m      8\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mWorkflow completed successfully, results saved in: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mresult\u001b[38;5;241m.\u001b[39moutput_dir\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n",
+      "\u001b[0;31mNameError\u001b[0m: name 'Submitter' is not defined"
+     ]
+    }
+   ],
+   "source": [
+    "from pydra.tasks.testing import SafeDivisionWorkflow\n",
+    "\n",
+    "wf = SafeDivisionWorkflow(a=10, b=5).split(denominator=[3, 2 ,0])\n",
+    "\n",
+    "with Submitter(worker=\"cf\") as sub:\n",
+    "    result = sub(wf)\n",
+    "    \n",
+    "print(f\"Workflow completed successfully, results saved in: {result.output_dir}\")"
+   ]
   },
   {
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "Work in progress..."
+    "To find the task directory where the issue first surfaced, iterate through every task\n",
+    "cache dir and check the results for `float(\"inf\")`s"
    ]
   },
   {
-   "cell_type": "markdown",
+   "cell_type": "code",
+   "execution_count": null,
    "metadata": {},
-   "source": []
+   "outputs": [],
+   "source": [
+    "import cloudpickle as cp\n",
+    "from pydra.utils import user_cache_dir\n",
+    "\n",
+    "run_cache = user_cache_dir / \"run-cache\"\n",
+    "\n",
+    "for task_cache_dir in run_cache.iterdir():\n",
+    "    with open(task_cache_dir / \"_result.pklz\", \"rb\") as f:\n",
+    "        error = cp.load(f)\n",
+    "    for \n",
+    "    "
+   ]
   }
  ],
  "metadata": {

new-docs/source/tutorial/tst.py

@@ -5,7 +5,7 @@
 wf = UnsafeDivisionWorkflow(a=10, b=5, denominator=2)
 
 if __name__ == "__main__":
-    with Submitter(worker="cf") as sub:
+    with Submitter(worker="cf", rerun=True) as sub:
         result = sub(wf)
 
 

pydra/design/python.py

@@ -166,7 +166,7 @@ def make(wrapped: ty.Callable | type) -> PythonDef:
         for i, output in enumerate(parsed_outputs.values()):
             output.order = i
 
-        interface = make_task_def(
+        defn = make_task_def(
             PythonDef,
             PythonOutputs,
             parsed_inputs,
@@ -177,7 +177,7 @@ def make(wrapped: ty.Callable | type) -> PythonDef:
             outputs_bases=outputs_bases,
         )
 
-        return interface
+        return defn
 
     if wrapped is not None:
         if not isinstance(wrapped, (ty.Callable, type)):

pydra/engine/specs.py

@@ -380,6 +380,8 @@ def _compute_hashes(self) -> ty.Tuple[bytes, ty.Dict[str, bytes]]:
             if getattr(field, "container_path", False):
                 continue
             inp_dict[field.name] = getattr(self, field.name)
+        # Include the outputs class, just in case any names or types have changed
+        inp_dict["Outputs"] = self.Outputs
         hash_cache = Cache()
         field_hashes = {
             k: hash_function(v, cache=hash_cache) for k, v in inp_dict.items()

pydra/engine/tests/test_functions.py

@@ -182,18 +182,17 @@ def Indirect(a):
 
     # Run functions to ensure behavior is unaffected
     a = random.randint(0, (1 << 32) - 3)
-    assert Direct(a=a) == Partial(a=a)
-    assert Direct(a=a) == Indirect(a=a)
+    assert hashes(Direct(a=a)) == hashes(Partial(a=a)) == hashes(Indirect(a=a))
 
     # checking if the annotation is properly converted to output_spec if used in task
     assert list_fields(Direct.Outputs) == [
-        python.arg(name="sum", type=int),
-        python.arg(name="sub", type=int),
+        python.out(name="sum", type=int, order=0),
+        python.out(name="sub", type=int, order=1),
     ]
 
 
 def test_invalid_annotation():
-    with pytest.raises(TypeError):
+    with pytest.raises(ValueError, match="Unrecognised input names"):
 
         @python.define(inputs={"b": int})
         def addtwo(a):
@@ -202,50 +201,51 @@ def addtwo(a):
 
 def test_annotated_task():
 
-    def square(in_val: float):
+    @python.define
+    def Square(in_val: float):
         return in_val**2
 
-    res = square(in_val=2.0)()
-    assert res.output.out == 4.0
+    outputs = Square(in_val=2.0)()
+    assert outputs.out == 4.0
 
 
 def test_return_annotated_task():
 
     @python.define(inputs={"in_val": float}, outputs={"squared": float})
-    def square(in_val):
+    def Square(in_val):
         return in_val**2
 
-    res = square(in_val=2.0)()
-    assert res.output.squared == 4.0
+    outputs = Square(in_val=2.0)()
+    assert outputs.squared == 4.0
 
 
 def test_return_halfannotated_annotated_task():
 
     @python.define(inputs={"in_val": float}, outputs={"out": float})
-    def square(in_val):
+    def Square(in_val):
         return in_val**2
 
-    res = square(in_val=2.0)()
-    assert res.output.out == 4.0
+    outputs = Square(in_val=2.0)()
+    assert outputs.out == 4.0
 
 
 def test_return_annotated_task_multiple_output():
 
     @python.define(inputs={"in_val": float}, outputs={"squared": float, "cubed": float})
-    def square(in_val):
+    def Square(in_val):
         return in_val**2, in_val**3
 
-    res = square(in_val=2.0)()
-    assert res.output.squared == 4.0
-    assert res.output.cubed == 8.0
+    outputs = Square(in_val=2.0)()
+    assert outputs.squared == 4.0
+    assert outputs.cubed == 8.0
 
 
 def test_return_halfannotated_task_multiple_output():
 
     @python.define(inputs={"in_val": float}, outputs=(float, float))
-    def square(in_val):
+    def Square(in_val):
         return in_val**2, in_val**3
 
-    res = square(in_val=2.0)()
-    assert res.output.out1 == 4.0
-    assert res.output.out2 == 8.0
+    outputs = Square(in_val=2.0)()
+    assert outputs.out1 == 4.0
+    assert outputs.out2 == 8.0

pydra/tasks/testing/__init__.py

@@ -11,6 +11,13 @@ def Divide(x: float, y: float) -> float:
     return x / y
 
 
+@python.define
+def SafeDivide(x: float, y: float) -> float:
+    if y == 0:
+        return float("inf")
+    return x / y
+
+
 @python.define
 def Subtract(x: float, y: float) -> float:
     return x - y
@@ -41,3 +48,30 @@ def UnsafeDivisionWorkflow(a: float, b: float, denominator: float) -> float:
     divide = workflow.add(Divide(x=add.out, y=denominator))
     subtract = workflow.add(Subtract(x=divide.out, y=b))
     return subtract.out
+
+
+@workflow.define
+def SafeDivisionWorkflow(a: float, b: float, denominator: float) -> float:
+    """Adds 'a' and 'b' together, divides by 'denominator', and then subtracts 'b' from
+    the output. Division by 0 is not guarded against so the workflow will fail if
+    the value passed to the 'denominator' parameter is 0.
+
+    Parameters
+    ----------
+    a : float
+        The first number to add.
+    b : float
+        The second number to add.
+    denominator : float
+        The number to divide the sum of 'a' and 'b' by.
+
+    Returns
+    -------
+    out : float
+        The result of subtracting 'b' from the result of dividing the sum of 'a' and
+        'b' by 'denominator'.
+    """
+    add = workflow.add(Add(x=a, y=b))
+    divide = workflow.add(SafeDivide(x=add.out, y=denominator))
+    subtract = workflow.add(Subtract(x=divide.out, y=b))
+    return subtract.out