From 04aecf5ba953516eaaeea423c6428bbc5a4dc8ac Mon Sep 17 00:00:00 2001 From: Kevin Paul Date: Wed, 3 Jul 2024 02:44:08 -0600 Subject: [PATCH] Custom client/scheduler MPI rank placement (#110) * Generalizing starting coroutines in CLI * Ignore scratch space * scheduler_rank and exclusive_workers options with tests * allow 1-rank clusters * Correction to min world size calculation * Rename module for clarity * Create execute function * NOQA on unused imports * Set worker type if deprecated "--no-nanny" option set * Set worker type before raise * Import from dask not distributed * Update versioneer script to new Python * Temporary fix for python 3.12 changes * Custom rank placement logic * Move no_exit test into main initialize test and rename * Renaming for better clarity * Add execute tests * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Set python version in readthedocs env * Try fixing python version for readthedocs build * Revert * Rename to match test name * Rename test / add no_exit test for execute * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Possible to not supply a function * Reorder cli options * Add more execute options * move send_close_signal to execute * send_close_signal has moved * Deprecate initialize, now that execute does everything * Revert initialize deprecation warning for now * Attempt RTF fix --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .gitignore | 1 + dask_mpi/__init__.py | 7 +- dask_mpi/cli.py | 104 +++++---- dask_mpi/execute.py | 210 ++++++++++++++++++ dask_mpi/{core.py => initialize.py} | 21 +- dask_mpi/tests/execute_basic.py | 40 ++++ dask_mpi/tests/execute_no_exit.py | 31 +++ .../{core_basic.py => initialize_basic.py} | 0 ...{core_no_exit.py => initialize_no_exit.py} | 0 dask_mpi/tests/test_cli.py | 91 ++++++-- dask_mpi/tests/test_execute.py | 47 ++++ .../{test_core.py => test_initialize.py} | 15 +- dask_mpi/tests/test_no_exit.py | 20 -- docs/environment.yml | 10 +- setup.py | 2 +- 15 files changed, 495 insertions(+), 104 deletions(-) create mode 100644 dask_mpi/execute.py rename dask_mpi/{core.py => initialize.py} (91%) create mode 100644 dask_mpi/tests/execute_basic.py create mode 100644 dask_mpi/tests/execute_no_exit.py rename dask_mpi/tests/{core_basic.py => initialize_basic.py} (100%) rename dask_mpi/tests/{core_no_exit.py => initialize_no_exit.py} (100%) create mode 100644 dask_mpi/tests/test_execute.py rename dask_mpi/tests/{test_core.py => test_initialize.py} (58%) delete mode 100644 dask_mpi/tests/test_no_exit.py diff --git a/.gitignore b/.gitignore index 5fc18e6..3227925 100644 --- a/.gitignore +++ b/.gitignore @@ -115,6 +115,7 @@ global.lock purge.lock /temp/ /dask-worker-space/ +/dask-scratch-space/ # VSCode files .vscode/ diff --git a/dask_mpi/__init__.py b/dask_mpi/__init__.py index c2ec8cb..43ece32 100644 --- a/dask_mpi/__init__.py +++ b/dask_mpi/__init__.py @@ -1,5 +1,6 @@ -from . import _version -from .core import initialize, send_close_signal -from .exceptions import WorldTooSmallException +from ._version import get_versions +from .exceptions import WorldTooSmallException # noqa +from .execute import execute, send_close_signal # noqa +from .initialize import initialize # noqa __version__ = _version.get_versions()["version"] diff --git a/dask_mpi/cli.py b/dask_mpi/cli.py index 2cccdfd..fc2d848 100644 --- a/dask_mpi/cli.py +++ b/dask_mpi/cli.py @@ -23,6 +23,12 @@ type=int, help="Specify scheduler port number. Defaults to random.", ) +@click.option( + "--scheduler-rank", + default=0, + type=int, + help="The MPI rank on which the scheduler will launch. Defaults to 0.", +) @click.option( "--interface", type=str, default=None, help="Network interface like 'eth0' or 'ib0'" ) @@ -56,6 +62,14 @@ default=True, help="Start workers in nanny process for management (deprecated use --worker-class instead)", ) +@click.option( + "--exclusive-workers/--inclusive-workers", + default=True, + help=( + "Whether to force workers to run on unoccupied MPI ranks. If false, " + "then a worker will be launched on the same rank as the scheduler." + ), +) @click.option( "--worker-class", type=str, @@ -90,27 +104,30 @@ def main( scheduler_address, scheduler_file, + scheduler_port, + scheduler_rank, interface, + protocol, nthreads, - local_directory, memory_limit, + local_directory, scheduler, dashboard, dashboard_address, nanny, + exclusive_workers, worker_class, worker_options, - scheduler_port, - protocol, name, ): comm = MPI.COMM_WORLD world_size = comm.Get_size() - if scheduler and world_size < 2: + min_world_size = 1 + scheduler * max(scheduler_rank, exclusive_workers) + if world_size < min_world_size: raise WorldTooSmallException( - f"Not enough MPI ranks to start cluster, found {world_size}, " - "needs at least 2, one each for the scheduler and a worker." + f"Not enough MPI ranks to start cluster with exclusive_workers={exclusive_workers} and " + f"scheduler_rank={scheduler_rank}, found {world_size} MPI ranks but needs {min_world_size}." ) rank = comm.Get_rank() @@ -120,47 +137,52 @@ def main( except TypeError: worker_options = {} - if rank == 0 and scheduler: + async def run_worker(): + WorkerType = import_term(worker_class) + if not nanny: + WorkerType = Worker + raise DeprecationWarning( + "Option --no-nanny is deprectaed, use --worker-class instead" + ) + opts = { + "interface": interface, + "protocol": protocol, + "nthreads": nthreads, + "memory_limit": memory_limit, + "local_directory": local_directory, + "name": f"{name}-{rank}", + "scheduler_file": scheduler_file, + **worker_options, + } + if scheduler_address: + opts["scheduler_ip"] = scheduler_address - async def run_scheduler(): - async with Scheduler( - interface=interface, - protocol=protocol, - dashboard=dashboard, - dashboard_address=dashboard_address, - scheduler_file=scheduler_file, - port=scheduler_port, - ) as s: - comm.Barrier() - await s.finished() + async with WorkerType(**opts) as worker: + await worker.finished() - asyncio.get_event_loop().run_until_complete(run_scheduler()) + async def run_scheduler(launch_worker=False): + async with Scheduler( + interface=interface, + protocol=protocol, + dashboard=dashboard, + dashboard_address=dashboard_address, + scheduler_file=scheduler_file, + port=scheduler_port, + ) as scheduler: + comm.Barrier() + if launch_worker: + asyncio.get_event_loop().create_task(run_worker()) + + await scheduler.finished() + + if rank == scheduler_rank and scheduler: + asyncio.get_event_loop().run_until_complete( + run_scheduler(launch_worker=not exclusive_workers) + ) else: comm.Barrier() - async def run_worker(): - WorkerType = import_term(worker_class) - if not nanny: - raise DeprecationWarning( - "Option --no-nanny is deprectaed, use --worker-class instead" - ) - WorkerType = Worker - opts = { - "interface": interface, - "protocol": protocol, - "nthreads": nthreads, - "memory_limit": memory_limit, - "local_directory": local_directory, - "name": f"{name}-{rank}", - "scheduler_file": scheduler_file, - **worker_options, - } - if scheduler_address: - opts["scheduler_ip"] = scheduler_address - async with WorkerType(**opts) as worker: - await worker.finished() - asyncio.get_event_loop().run_until_complete(run_worker()) diff --git a/dask_mpi/execute.py b/dask_mpi/execute.py new file mode 100644 index 0000000..fb6df43 --- /dev/null +++ b/dask_mpi/execute.py @@ -0,0 +1,210 @@ +import asyncio +import threading + +import dask +from distributed import Client, Nanny, Scheduler +from distributed.utils import import_term + +from .exceptions import WorldTooSmallException + + +def execute( + client_function=None, + client_args=(), + client_kwargs=None, + client_rank=1, + scheduler=True, + scheduler_rank=0, + scheduler_address=None, + scheduler_port=None, + scheduler_file=None, + interface=None, + nthreads=1, + local_directory="", + memory_limit="auto", + nanny=False, + dashboard=True, + dashboard_address=":8787", + protocol=None, + exclusive_workers=True, + worker_class="distributed.Worker", + worker_options=None, + worker_name=None, + comm=None, +): + """ + Execute a function on a given MPI rank with a Dask cluster launched using mpi4py + + Using mpi4py, MPI rank 0 launches the Scheduler, MPI rank 1 passes through to the + client script, and all other MPI ranks launch workers. All MPI ranks other than + MPI rank 1 block while their event loops run. + + In normal operation these ranks exit once rank 1 ends. If exit=False is set they + instead return an bool indicating whether they are the client and should execute + more client code, or a worker/scheduler who should not. In this case the user is + responsible for the client calling send_close_signal when work is complete, and + checking the returned value to choose further actions. + + Parameters + ---------- + func : callable + A function containing Dask client code to execute with a Dask cluster. If + func it not callable, then no client code will be executed. + args : list + Arguments to the client function + client_rank : int + The MPI rank on which to run func. + scheduler_rank : int + The MPI rank on which to run the Dask scheduler + scheduler_address : str + IP Address of the scheduler, used if scheduler is not launched + scheduler_port : int + Specify scheduler port number. Defaults to random. + scheduler_file : str + Filename to JSON encoded scheduler information. + interface : str + Network interface like 'eth0' or 'ib0' + nthreads : int + Number of threads per worker + local_directory : str + Directory to place worker files + memory_limit : int, float, or 'auto' + Number of bytes before spilling data to disk. This can be an + integer (nbytes), float (fraction of total memory), or 'auto'. + nanny : bool + Start workers in nanny process for management (deprecated, use worker_class instead) + dashboard : bool + Enable Bokeh visual diagnostics + dashboard_address : str + Bokeh port for visual diagnostics + protocol : str + Protocol like 'inproc' or 'tcp' + exclusive_workers : bool + Whether to only run Dask workers on their own MPI ranks + worker_class : str + Class to use when creating workers + worker_options : dict + Options to pass to workers + worker_name : str + Prefix for name given to workers. If defined, each worker will be named + '{worker_name}-{rank}'. Otherwise, the name of each worker is just '{rank}'. + comm : mpi4py.MPI.Intracomm + Optional MPI communicator to use instead of COMM_WORLD + kwargs : dict + Keyword arguments to the client function + """ + if comm is None: + from mpi4py import MPI + + comm = MPI.COMM_WORLD + + world_size = comm.Get_size() + min_world_size = 1 + max(client_rank, scheduler_rank, exclusive_workers) + if world_size < min_world_size: + raise WorldTooSmallException( + f"Not enough MPI ranks to start cluster with exclusive_workers={exclusive_workers} and " + f"scheduler_rank={scheduler_rank}, found {world_size} MPI ranks but needs {min_world_size}." + ) + + rank = comm.Get_rank() + + if not worker_options: + worker_options = {} + + async def run_client(): + def wrapped_function(*args, **kwargs): + client_function(*args, **kwargs) + send_close_signal() + + threading.Thread( + target=wrapped_function, args=client_args, kwargs=client_kwargs + ).start() + + async def run_worker(with_client=False): + WorkerType = import_term(worker_class) + if nanny: + WorkerType = Nanny + raise DeprecationWarning( + "Option nanny=True is deprectaed, use worker_class='distributed.Nanny' instead" + ) + opts = { + "interface": interface, + "protocol": protocol, + "nthreads": nthreads, + "memory_limit": memory_limit, + "local_directory": local_directory, + "name": rank if not worker_name else f"{worker_name}-{rank}", + **worker_options, + } + if not scheduler and scheduler_address: + opts["scheduler_ip"] = scheduler_address + async with WorkerType(**opts) as worker: + if with_client: + asyncio.get_event_loop().create_task(run_client()) + + await worker.finished() + + async def run_scheduler(with_worker=False, with_client=False): + async with Scheduler( + interface=interface, + protocol=protocol, + dashboard=dashboard, + dashboard_address=dashboard_address, + scheduler_file=scheduler_file, + port=scheduler_port, + ) as scheduler: + dask.config.set(scheduler_address=scheduler.address) + comm.bcast(scheduler.address, root=scheduler_rank) + comm.Barrier() + + if with_worker: + asyncio.get_event_loop().create_task( + run_worker(with_client=with_client) + ) + + elif with_client: + asyncio.get_event_loop().create_task(run_client()) + + await scheduler.finished() + + with_scheduler = scheduler and (rank == scheduler_rank) + with_client = callable(client_function) and (rank == client_rank) + + if with_scheduler: + run_coro = run_scheduler( + with_worker=not exclusive_workers, + with_client=with_client, + ) + + else: + if scheduler: + scheduler_address = comm.bcast(None, root=scheduler_rank) + elif scheduler_address is None: + raise ValueError( + "Must provide scheduler_address if executing with scheduler=False" + ) + dask.config.set(scheduler_address=scheduler_address) + comm.Barrier() + + if with_client and exclusive_workers: + run_coro = run_client() + else: + run_coro = run_worker(with_client=with_client) + + asyncio.get_event_loop().run_until_complete(run_coro) + + +def send_close_signal(): + """ + The client can call this function to explicitly stop + the event loop. + + This is not needed in normal usage, where it is run + automatically when the client code exits python. + + You only need to call this manually when using exit=False + in initialize. + """ + + with Client() as c: + c.shutdown() diff --git a/dask_mpi/core.py b/dask_mpi/initialize.py similarity index 91% rename from dask_mpi/core.py rename to dask_mpi/initialize.py index e133a42..63c8219 100644 --- a/dask_mpi/core.py +++ b/dask_mpi/initialize.py @@ -3,10 +3,11 @@ import sys import dask -from distributed import Client, Nanny, Scheduler +from distributed import Nanny, Scheduler from distributed.utils import import_term from .exceptions import WorldTooSmallException +from .execute import send_close_signal def initialize( @@ -121,10 +122,10 @@ async def run_scheduler(): async def run_worker(): WorkerType = import_term(worker_class) if nanny: + WorkerType = Nanny raise DeprecationWarning( "Option nanny=True is deprectaed, use worker_class='distributed.Nanny' instead" ) - WorkerType = Nanny opts = { "interface": interface, "protocol": protocol, @@ -142,19 +143,3 @@ async def run_worker(): sys.exit() else: return False - - -def send_close_signal(): - """ - The client can call this function to explicitly stop - the event loop. - - This is not needed in normal usage, where it is run - automatically when the client code exits python. - - You only need to call this manually when using exit=False - in initialize. - """ - - with Client() as c: - c.shutdown() diff --git a/dask_mpi/tests/execute_basic.py b/dask_mpi/tests/execute_basic.py new file mode 100644 index 0000000..7af4e9f --- /dev/null +++ b/dask_mpi/tests/execute_basic.py @@ -0,0 +1,40 @@ +import sys +from time import sleep + +from distributed import Client +from distributed.metrics import time + +from dask_mpi import execute + + +def client_func(m, c, s, x): + xranks = {c, s} if x else set() + worker_ranks = set(i for i in range(m) if i not in xranks) + + with Client() as c: + start = time() + while len(c.scheduler_info()["workers"]) != len(worker_ranks): + assert time() < start + 10 + sleep(0.2) + + actual_worker_ranks = set( + v["name"] for k, v in c.scheduler_info()["workers"].items() + ) + assert actual_worker_ranks == worker_ranks + + for i in actual_worker_ranks: + assert c.submit(lambda x: x + 1, 10, workers=i).result() == 11 + + +if __name__ == "__main__": + vmap = {"True": True, "False": False, "None": None} + int_or_bool = lambda s: vmap[s] if s in vmap else int(s) + args = [int_or_bool(i) for i in sys.argv[1:]] + + execute( + client_function=client_func, + client_args=args, + client_rank=args[1], + scheduler_rank=args[2], + exclusive_workers=args[3], + ) diff --git a/dask_mpi/tests/execute_no_exit.py b/dask_mpi/tests/execute_no_exit.py new file mode 100644 index 0000000..0735684 --- /dev/null +++ b/dask_mpi/tests/execute_no_exit.py @@ -0,0 +1,31 @@ +from time import sleep, time + +from distributed import Client +from mpi4py.MPI import COMM_WORLD as world + +from dask_mpi import execute + +# Split our MPI world into two pieces, one consisting just of +# the old rank 3 process and the other with everything else +new_comm_assignment = 1 if world.rank == 3 else 0 +comm = world.Split(new_comm_assignment) + +if world.rank != 3: + + def client_code(): + with Client() as c: + start = time() + while len(c.scheduler_info()["workers"]) != 1: + assert time() < start + 10 + sleep(0.2) + + c.submit(lambda x: x + 1, 10).result() == 11 + c.submit(lambda x: x + 1, 20).result() == 21 + + execute(client_code, comm=comm) + +# check that our original comm is intact +world.Barrier() +x = 100 if world.rank == 0 else 200 +x = world.bcast(x) +assert x == 100 diff --git a/dask_mpi/tests/core_basic.py b/dask_mpi/tests/initialize_basic.py similarity index 100% rename from dask_mpi/tests/core_basic.py rename to dask_mpi/tests/initialize_basic.py diff --git a/dask_mpi/tests/core_no_exit.py b/dask_mpi/tests/initialize_no_exit.py similarity index 100% rename from dask_mpi/tests/core_no_exit.py rename to dask_mpi/tests/initialize_no_exit.py diff --git a/dask_mpi/tests/test_cli.py b/dask_mpi/tests/test_cli.py index e10e370..4a2f754 100644 --- a/dask_mpi/tests/test_cli.py +++ b/dask_mpi/tests/test_cli.py @@ -8,10 +8,11 @@ import pytest import requests +from dask.utils import tmpfile from distributed import Client from distributed.comm.addressing import get_address_host_port from distributed.metrics import time -from distributed.utils import import_term, tmpfile +from distributed.utils import import_term from distributed.utils_test import cleanup, loop, loop_in_thread, popen # noqa: F401 pytest.importorskip("mpi4py") @@ -51,6 +52,27 @@ def test_basic(loop, worker_class, mpirun): assert c.submit(lambda x: x + 1, 10).result() == 11 +def test_inclusive_workers(loop, mpirun): + with tmpfile(extension="json") as fn: + cmd = mpirun + [ + "-np", + "4", + "dask-mpi", + "--scheduler-file", + fn, + "--inclusive-workers", + ] + + with popen(cmd): + with Client(scheduler_file=fn) as client: + start = time() + while len(client.scheduler_info()["workers"]) < 4: + assert time() < start + 10 + sleep(0.1) + + assert client.submit(lambda x: x + 1, 10).result() == 11 + + def test_small_world(mpirun): with tmpfile(extension="json") as fn: # Set too few processes to start cluster @@ -69,6 +91,27 @@ def test_small_world(mpirun): assert p.returncode != 0 +def test_inclusive_small_world(mpirun): + with tmpfile(extension="json") as fn: + cmd = mpirun + [ + "-np", + "1", + "dask-mpi", + "--scheduler-file", + fn, + "--inclusive-workers", + ] + + with popen(cmd): + with Client(scheduler_file=fn) as client: + start = time() + while len(client.scheduler_info()["workers"]) < 1: + assert time() < start + 10 + sleep(0.1) + + assert client.submit(lambda x: x + 1, 10).result() == 11 + + def test_no_scheduler(loop, mpirun): with tmpfile(extension="json") as fn: cmd = mpirun + ["-np", "2", "dask-mpi", "--scheduler-file", fn] @@ -98,6 +141,35 @@ def test_no_scheduler(loop, mpirun): sleep(0.2) +def test_scheduler_rank(loop, mpirun): + with tmpfile(extension="json") as fn: + cmd = mpirun + [ + "-np", + "2", + "dask-mpi", + "--scheduler-file", + fn, + "--exclusive-workers", + "--scheduler-rank", + "1", + ] + + with popen(cmd, stdin=FNULL): + with Client(scheduler_file=fn) as client: + start = time() + while len(client.scheduler_info()["workers"]) < 1: + assert time() < start + 10 + sleep(0.2) + + worker_infos = client.scheduler_info()["workers"] + assert len(worker_infos) == 1 + + worker_info = next(iter(worker_infos.values())) + assert worker_info["name"].rsplit("-")[-1] == "0" + + assert client.submit(lambda x: x + 1, 10).result() == 11 + + @pytest.mark.parametrize("nanny", ["--nanny", "--no-nanny"]) def test_non_default_ports(loop, nanny, mpirun): with tmpfile(extension="json") as fn: @@ -150,23 +222,6 @@ def test_dashboard(loop, mpirun): requests.get("http://localhost:59583/status/") -@pytest.mark.skip(reason="Should we expose this option?") -def test_bokeh_worker(loop, mpirun): - with tmpfile(extension="json") as fn: - cmd = mpirun + [ - "-np", - "2", - "dask-mpi", - "--scheduler-file", - fn, - "--bokeh-worker-port", - "59584", - ] - - with popen(cmd, stdin=FNULL): - check_port_okay(59584) - - def tmpfile_static(extension="", dir=None): """ utility function for test_stale_sched test diff --git a/dask_mpi/tests/test_execute.py b/dask_mpi/tests/test_execute.py new file mode 100644 index 0000000..99b74fa --- /dev/null +++ b/dask_mpi/tests/test_execute.py @@ -0,0 +1,47 @@ +from __future__ import absolute_import, division, print_function + +import os +import subprocess +import sys + +import pytest + +pytest.importorskip("mpi4py") + + +@pytest.mark.parametrize( + "mpisize,crank,srank,xworkers,retcode", + [ + (4, 1, 0, True, 0), # DEFAULTS + (1, 1, 0, True, 1), # Set too few processes to start cluster + (4, 2, 3, True, 0), + (5, 1, 3, True, 0), + (3, 2, 2, True, 0), + (2, 0, 0, False, 0), + (1, 0, 0, False, 0), + (1, 0, 0, True, 1), + ], +) +def test_basic(mpisize, crank, srank, xworkers, retcode, mpirun): + script_file = os.path.join( + os.path.dirname(os.path.realpath(__file__)), "execute_basic.py" + ) + + script_args = [str(v) for v in (mpisize, crank, srank, xworkers)] + p = subprocess.Popen( + mpirun + ["-n", script_args[0], sys.executable, script_file] + script_args + ) + + p.communicate() + assert p.returncode == retcode + + +def test_no_exit(mpirun): + script_file = os.path.join( + os.path.dirname(os.path.realpath(__file__)), "execute_no_exit.py" + ) + + p = subprocess.Popen(mpirun + ["-np", "4", sys.executable, script_file]) + + p.communicate() + assert p.returncode == 0 diff --git a/dask_mpi/tests/test_core.py b/dask_mpi/tests/test_initialize.py similarity index 58% rename from dask_mpi/tests/test_core.py rename to dask_mpi/tests/test_initialize.py index abe4a60..56757c4 100644 --- a/dask_mpi/tests/test_core.py +++ b/dask_mpi/tests/test_initialize.py @@ -11,7 +11,7 @@ def test_basic(mpirun): script_file = os.path.join( - os.path.dirname(os.path.realpath(__file__)), "core_basic.py" + os.path.dirname(os.path.realpath(__file__)), "initialize_basic.py" ) p = subprocess.Popen(mpirun + ["-np", "4", sys.executable, script_file]) @@ -22,7 +22,7 @@ def test_basic(mpirun): def test_small_world(mpirun): script_file = os.path.join( - os.path.dirname(os.path.realpath(__file__)), "core_basic.py" + os.path.dirname(os.path.realpath(__file__)), "initialize_basic.py" ) # Set too few processes to start cluster @@ -30,3 +30,14 @@ def test_small_world(mpirun): p.communicate() assert p.returncode != 0 + + +def test_no_exit(mpirun): + script_file = os.path.join( + os.path.dirname(os.path.realpath(__file__)), "initialize_no_exit.py" + ) + + p = subprocess.Popen(mpirun + ["-np", "4", sys.executable, script_file]) + + p.communicate() + assert p.returncode == 0 diff --git a/dask_mpi/tests/test_no_exit.py b/dask_mpi/tests/test_no_exit.py deleted file mode 100644 index 65c27d5..0000000 --- a/dask_mpi/tests/test_no_exit.py +++ /dev/null @@ -1,20 +0,0 @@ -from __future__ import absolute_import, division, print_function - -import os -import subprocess -import sys - -import pytest - -pytest.importorskip("mpi4py") - - -def test_no_exit(mpirun): - script_file = os.path.join( - os.path.dirname(os.path.realpath(__file__)), "core_no_exit.py" - ) - - p = subprocess.Popen(mpirun + ["-np", "4", sys.executable, script_file]) - - p.communicate() - assert p.returncode == 0 diff --git a/docs/environment.yml b/docs/environment.yml index 209c20f..3c70b3d 100644 --- a/docs/environment.yml +++ b/docs/environment.yml @@ -3,15 +3,23 @@ channels: - conda-forge - nodefaults dependencies: + - python<3.12 - dask>=2.19 - distributed>=2.19 - mpich - mpi4py>=3.0.3 - versioneer - - sphinx>=5.0 + - sphinx - pygments - pip - pip: + #>>>> See: https://github.com/dask/dask-sphinx-theme/issues/68 + - sphinxcontrib-applehelp<1.0.5 + - sphinxcontrib-devhelp<1.0.6 + - sphinxcontrib-htmlhelp<2.0.5 + - sphinxcontrib-serializinghtml<1.1.10 + - sphinxcontrib-qthelp<1.0.7 + #<<<< - dask-sphinx-theme>=3.0.5 - numpydoc - sphinx-click diff --git a/setup.py b/setup.py index 8a1aedf..d197b71 100644 --- a/setup.py +++ b/setup.py @@ -47,7 +47,7 @@ def environment_dependencies(obj, dependencies=None): license="BSD 3-Clause", include_package_data=True, install_requires=install_requires, - python_requires=">=3.6", + python_requires=">=3.6,<3.12", packages=["dask_mpi"], long_description=long_description, entry_points="""