pyfracval.dask_runner#

Dask client helpers for distributed aggregate generation.

Module Contents#

pyfracval.dask_runner.install_wheel_on_workers(client, wheel_path, package_name, expected_version)[source]#

Install an arbitrary wheel on a Dask scheduler and all its workers, verified via a runtime version fingerprint.

Generic version of the mechanism this module has always used for installing pyfracval itself onto Dask Docker workers that don’t have it preinstalled (e.g. a generic ghcr.io/dask/dask image) – reusable for any package, not just this one. In particular: a compiled extension (e.g. pyfastmm’s f2py extension) needs a wheel actually built for the worker’s platform/Python ABI – this function only ships and installs whatever wheel you hand it, it does not build one (see _register_package below for pyfracval’s own “build one first” case).

Installs via a WorkerPlugin (client.register_plugin()), not a one-shot client.run() sweep: a plugin also runs its setup() on any worker that joins after this call (Docker restart, autoscaling, …), where a client.run() snapshot of currently-connected workers would silently miss it and leave that worker without the package. The installer function and plugin class are both defined inline so cloudpickle serialises them by value (bytecode), not by reference to a module – which would fail on the scheduler/workers before package_name is installed there.

Parameters:

  • client – A connected dask.distributed.Client.

  • wheel_path – Path to a .whl file, already built for the worker’s platform and Python version (this function does no cross-compilation or compatibility checking – get that part right before calling this).

  • package_name – The distribution/import name (e.g. "pyfracval", "pyfastmm", "spcwth") – used for the post-install version check, the stale-module cache eviction, and the <PACKAGE>_INSTALLED_WHEEL/ <PACKAGE>_EXPECTED_VERSION env vars set on each worker.

  • expected_version – Version string the installed wheel must report after installing, or this raises RuntimeError.

pyfracval.dask_runner.get_client(scheduler_address=None, n_workers=None, install_package=False)[source]#

Return a Dask distributed Client.

If scheduler_address is given, connect to a running scheduler at that address (e.g. "tcp://host:8786"). Otherwise start a local LocalCluster with n_workers workers (defaults to the number of CPU cores when n_workers is None).

When install_package is True and a remote scheduler is used, the local pyfracval package is built into a wheel and installed on all workers via a WorkerPlugin before the client is returned. This is required whenever the workers do not have pyfracval pre-installed (e.g. a generic Dask Docker image).

Parameters:

  • scheduler_address – Address of a remote Dask scheduler. None → use a local cluster.

  • n_workers – Number of workers for a local cluster. Ignored when connecting to a remote scheduler.

  • install_package – When True and using a remote scheduler, build + install pyfracval on all workers before returning.

Return type:

dask.distributed.Client