Run dask with a matching interpreter

When Debian is in the process of migrating from one Python version to
another, there's a transitional period when we run tests on each of two
Python versions, although `/usr/bin/python3` can only point to one at a
time.  `distributed` runs `dask` from the executable search path, and
when it's installed as a system package rather than in a virtual
environment, it's possible during this transitional period for
`distributed` to be running under (e.g.) Python 3.13 while
`/usr/bin/dask` has a `#!` line that causes it to be run under (e.g.)
Python 3.12.  This results in the following obscure failure mode in some
tests in `distributed/deploy/tests/test_subprocess.py`:

  ______________________________ test_basic ______________________________

      @pytest.mark.skipif(WINDOWS, reason="distributed#7434")
      @gen_test()
      async def test_basic():
          async with SubprocessCluster(
              asynchronous=True,
              dashboard_address=":0",
              scheduler_kwargs={"idle_timeout": "5s"},
              worker_kwargs={"death_timeout": "5s"},
          ) as cluster:
              async with Client(cluster, asynchronous=True) as client:
  >               result = await client.submit(lambda x: x + 1, 10)

  distributed/deploy/tests/test_subprocess.py:27:
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

  self = <Future: cancelled, key: lambda-e426db6b5bc630ab037e145b5e8f631c>, raiseit = True

      async def _result(self, raiseit=True):
          await self._state.wait()
          if self.status == "error":
              exc = clean_exception(self._state.exception, self._state.traceback)
              if raiseit:
                  typ, exc, tb = exc
  >               raise exc.with_traceback(tb)
  E               SystemError: no locals found when setting up annotations

  distributed/client.py:410: SystemError

Explicitly running `dask` using `sys.executable` fixes this.  It's only
_strictly_ necessary to change the invocations of `dask` in
`distributed.deploy.subprocess`, but it seemed better to apply the same
change to the whole codebase for consistency.

Author: cjwatson

distributed/cli/tests/test_dask_scheduler.py

@@ -40,7 +40,7 @@ def _get_dashboard_port(client: Client) -> int:
 
 
 def test_defaults(loop, requires_default_ports):
-    with popen(["dask", "scheduler"]):
+    with popen([sys.executable, "-m", "dask", "scheduler"]):
 
         async def f():
             # Default behaviour is to listen on all addresses
@@ -53,7 +53,17 @@ async def f():
 
 def test_hostport(loop):
     port = open_port()
-    with popen(["dask", "scheduler", "--no-dashboard", "--host", f"127.0.0.1:{port}"]):
+    with popen(
+        [
+            sys.executable,
+            "-m",
+            "dask",
+            "scheduler",
+            "--no-dashboard",
+            "--host",
+            f"127.0.0.1:{port}",
+        ]
+    ):
 
         async def f():
             # The scheduler's main port can't be contacted from the outside
@@ -65,7 +75,7 @@ async def f():
 
 
 def test_no_dashboard(loop, requires_default_ports):
-    with popen(["dask", "scheduler", "--no-dashboard"]):
+    with popen([sys.executable, "-m", "dask", "scheduler", "--no-dashboard"]):
         with Client(f"127.0.0.1:{Scheduler.default_port}", loop=loop):
             response = requests.get("http://127.0.0.1:8787/status/")
             assert response.status_code == 404
@@ -76,7 +86,7 @@ def test_dashboard(loop):
     port = open_port()
 
     with popen(
-        ["dask", "scheduler", "--host", f"127.0.0.1:{port}"],
+        [sys.executable, "-m", "dask", "scheduler", "--host", f"127.0.0.1:{port}"],
     ):
         with Client(f"127.0.0.1:{port}", loop=loop) as c:
             dashboard_port = _get_dashboard_port(c)
@@ -109,6 +119,8 @@ def test_dashboard_non_standard_ports(loop):
     port2 = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "scheduler",
             f"--port={port1}",
@@ -137,6 +149,8 @@ def test_multiple_protocols(loop):
     port2 = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "scheduler",
             "--protocol=tcp,ws",
@@ -158,6 +172,8 @@ def test_dashboard_allowlist(loop):
     port = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "scheduler",
             f"--port={port}",
@@ -198,6 +214,8 @@ def test_interface(loop):
     port = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "scheduler",
             f"--port={port}",
@@ -208,6 +226,8 @@ def test_interface(loop):
     ) as s:
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "worker",
                 f"127.0.0.1:{port}",
@@ -252,12 +272,24 @@ def check_pidfile(proc, pidfile):
             assert proc.pid == pid
 
     with tmpfile() as s:
-        with popen(["dask", "scheduler", "--pid-file", s, "--no-dashboard"]) as sched:
+        with popen(
+            [
+                sys.executable,
+                "-m",
+                "dask",
+                "scheduler",
+                "--pid-file",
+                s,
+                "--no-dashboard",
+            ]
+        ) as sched:
             check_pidfile(sched, s)
 
         with tmpfile() as w:
             with popen(
                 [
+                    sys.executable,
+                    "-m",
                     "dask",
                     "worker",
                     f"127.0.0.1:{port}",
@@ -273,6 +305,8 @@ def test_scheduler_port_zero(loop):
     with tmpfile() as fn:
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "scheduler",
                 "--no-dashboard",
@@ -292,6 +326,8 @@ def test_dashboard_port_zero(loop):
     port = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "scheduler",
             "--host",
@@ -329,6 +365,8 @@ def check_scheduler():
     with tmpfile() as fn:
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "scheduler",
                 "--scheduler-file",
@@ -359,6 +397,8 @@ def check_scheduler():
     with tmpfile() as fn:
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "scheduler",
                 "--scheduler-file",
@@ -382,6 +422,8 @@ def test_preload_remote_module(loop, tmp_path):
     ):
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "scheduler",
                 "--scheduler-file",
@@ -408,7 +450,9 @@ def test_preload_config(loop):
     with tmpfile() as fn:
         env = os.environ.copy()
         env["DASK_DISTRIBUTED__SCHEDULER__PRELOAD"] = PRELOAD_TEXT
-        with popen(["dask", "scheduler", "--scheduler-file", fn], env=env):
+        with popen(
+            [sys.executable, "-m", "dask", "scheduler", "--scheduler-file", fn], env=env
+        ):
             with Client(scheduler_file=fn, loop=loop) as c:
                 assert (
                     c.run_on_scheduler(lambda dask_scheduler: dask_scheduler.foo)
@@ -446,6 +490,8 @@ def check_passthrough():
             print(fn)
             with popen(
                 [
+                    sys.executable,
+                    "-m",
                     "dask",
                     "scheduler",
                     "--scheduler-file",
@@ -477,7 +523,16 @@ def check_passthrough():
         with tmpfile() as fn2:
             print(fn2)
             with popen(
-                ["dask", "scheduler", "--scheduler-file", fn2, "--preload", path],
+                [
+                    sys.executable,
+                    "-m",
+                    "dask",
+                    "scheduler",
+                    "--scheduler-file",
+                    fn2,
+                    "--preload",
+                    path,
+                ],
                 stdout=sys.stdout,
                 stderr=sys.stderr,
             ):
@@ -530,10 +585,20 @@ def dask_setup(worker):
 """
     port = open_port()
     with popen(
-        ["dask", "scheduler", "--no-dashboard", "--host", f"127.0.0.1:{port}"]
+        [
+            sys.executable,
+            "-m",
+            "dask",
+            "scheduler",
+            "--no-dashboard",
+            "--host",
+            f"127.0.0.1:{port}",
+        ]
     ) as s:
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "worker",
                 f"localhost:{port}",
@@ -555,9 +620,37 @@ def dask_setup(worker):
 def test_multiple_workers(loop):
     scheduler_address = f"127.0.0.1:{open_port()}"
     with (
-        popen(["dask", "scheduler", "--no-dashboard", "--host", scheduler_address]),
-        popen(["dask", "worker", scheduler_address, "--no-dashboard"]),
-        popen(["dask", "worker", scheduler_address, "--no-dashboard"]),
+        popen(
+            [
+                sys.executable,
+                "-m",
+                "dask",
+                "scheduler",
+                "--no-dashboard",
+                "--host",
+                scheduler_address,
+            ]
+        ),
+        popen(
+            [
+                sys.executable,
+                "-m",
+                "dask",
+                "worker",
+                scheduler_address,
+                "--no-dashboard",
+            ]
+        ),
+        popen(
+            [
+                sys.executable,
+                "-m",
+                "dask",
+                "worker",
+                scheduler_address,
+                "--no-dashboard",
+            ]
+        ),
         Client(scheduler_address, loop=loop) as c,
     ):
         start = time()

distributed/cli/tests/test_dask_spec.py

@@ -4,6 +4,7 @@
 import json
 import signal
 import subprocess
+import sys
 
 import pytest
 import yaml
@@ -19,6 +20,8 @@ async def test_text():
     port = open_port()
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "spec",
             "--spec",
@@ -27,6 +30,8 @@ async def test_text():
     ):
         with popen(
             [
+                sys.executable,
+                "-m",
                 "dask",
                 "spec",
                 "tcp://localhost:%d" % port,
@@ -55,6 +60,8 @@ async def test_file(c, s, tmp_path):
         )
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "spec",
             s.address,
@@ -72,6 +79,8 @@ async def test_file(c, s, tmp_path):
 def test_errors():
     with popen(
         [
+            sys.executable,
+            "-m",
             "dask",
             "spec",
             "--spec",
@@ -85,7 +94,7 @@ def test_errors():
         assert "exactly one" in line
         assert "--spec" in line and "--spec-file" in line
 
-    with popen(["dask", "spec"], capture_output=True) as proc:
+    with popen([sys.executable, "-m", "dask", "spec"], capture_output=True) as proc:
         line = proc.stdout.readline().decode()
         assert "exactly one" in line
         assert "--spec" in line and "--spec-file" in line
@@ -97,6 +106,8 @@ def test_errors():
 @gen_cluster(client=True, nthreads=[])
 async def test_signal_handling_worker(c, s, worker_type, sig):
     worker = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-m",
         "dask",
         "spec",
         "--spec",
@@ -136,6 +147,8 @@ async def test_signal_handling_worker(c, s, worker_type, sig):
 async def test_signal_handling_scheduler(sig):
     port = open_port()
     scheduler = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-m",
         "dask",
         "spec",
         "--spec",