Adds cwd propagation to DaskExecutor (#994)

* Adds cwd propagation to DaskExecutor

* changelog

* add test_dask to CI

* Update cluster_tools/cluster_tools/executors/dask.py

Co-authored-by: Daniel <[email protected]>

---------

Co-authored-by: Daniel <[email protected]>

Author: normanrz

.github/workflows/ci.yml

@@ -134,7 +134,7 @@ jobs:
         if: ${{ matrix.executors == 'dask' && matrix.python-version != '3.8' }}
         run: |
           cd tests
-          PYTEST_EXECUTORS=dask poetry run python -m pytest -sv test_all.py
+          PYTEST_EXECUTORS=dask poetry run python -m pytest -sv test_all.py test_dask.py
 
   webknossos_linux:
     needs: changes

cluster_tools/Changelog.md

@@ -16,6 +16,7 @@ For upgrade instructions, please check the respective *Breaking Changes* section
 ### Changed
 
 ### Fixed
+- Fixed working directory propagation in DaskExecutor. [#994](https://github.com/scalableminds/webknossos-libs/pull/994)
 
 
 ## [0.14.14](https://github.com/scalableminds/webknossos-libs/releases/tag/v0.14.14) - 2024-01-12

cluster_tools/cluster_tools/executors/dask.py

@@ -41,6 +41,8 @@ def _run_in_nanny(
         for key, value in __env.items():
             os.environ[key] = value
 
+        if "PWD" in os.environ:
+            os.chdir(os.environ["PWD"])
         ret = __fn(*args, **kwargs)
         queue.put({"value": ret})
     except Exception as exc:
@@ -174,7 +176,9 @@ def submit(  # type: ignore[override]
                 ),
             )
 
-        kwargs["__env"] = os.environ.copy()
+        __env = os.environ.copy()
+        __env["PWD"] = os.getcwd()
+        kwargs["__env"] = __env
 
         # We run the functions in dask as a separate process to not hold the
         # GIL for too long, because dask workers need to be able to communicate

cluster_tools/tests/test_dask.py

@@ -0,0 +1,29 @@
+import os
+from typing import TYPE_CHECKING, List, Optional
+
+if TYPE_CHECKING:
+    from distributed import LocalCluster
+
+import cluster_tools
+
+_dask_cluster: Optional["LocalCluster"] = None
+
+
+def job(_arg: None) -> str:
+    return os.getcwd()
+
+
+def test_pass_cwd() -> None:
+    global _dask_cluster
+    if not _dask_cluster:
+        from distributed import LocalCluster, Worker
+
+        _dask_cluster = LocalCluster(
+            worker_class=Worker, resources={"mem": 20e9, "cpus": 4}, nthreads=6
+        )
+    with cluster_tools.get_executor(
+        "dask", job_resources={"address": _dask_cluster}
+    ) as exec:
+        tmp_path = os.path.realpath("/tmp")  # macOS redirects `/tmp` to `/private/tmp`
+        os.chdir(tmp_path)
+        assert list(exec.map(job, [None])) == [tmp_path]