update daskworker

Author: adi611

.github/workflows/testdask.yml

@@ -0,0 +1,42 @@
+name: Dask Tests
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.9, 3.10, 3.11]
+      fail-fast: false
+
+    steps:
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+
+      - name: Checkout Pydra repo
+        uses: actions/checkout@v3
+        with:
+          repository: ${{ github.repository }}
+
+      - name: Install pydra with Dask and test dependencies
+        run: |
+          pip install -e ".[test,dask]"
+
+      - name: Run tests
+        run: |
+          pytest -v --dask pydra/engine --cov pydra --cov-config .coveragerc --cov-report xml:cov.xml
+
+      - name: Upload to codecov
+        run: codecov -f cov.xml -F unittests -e GITHUB_WORKFLOW

pydra/engine/workers.py

@@ -879,8 +879,14 @@ async def exec_dask(self, runnable, rerun=False):
             from dask.distributed import Client
 
             self.client = await Client(**self.client_args, asynchronous=True)
-        future = self.client.submit(runnable._run, rerun)
-        result = await future
+
+        if isinstance(runnable, TaskBase):
+            future = self.client.submit(runnable._run, rerun)
+            result = await future
+        else:  # it could be tuple that includes pickle files with tasks and inputs
+            ind, task_main_pkl, task_orig = runnable
+            future = self.client.submit(load_and_run, task_main_pkl, ind, rerun)
+            result = await future
         return result
 
     def close(self):
@@ -894,4 +900,4 @@ def close(self):
     "slurm": SlurmWorker,
     "dask": DaskWorker,
     "sge": SGEWorker,
-}
+}