pyfracval.dask_runner
=====================

.. py:module:: pyfracval.dask_runner

.. autoapi-nested-parse::

   Dask client helpers for distributed aggregate generation.





Module Contents
---------------

.. py:function:: install_wheel_on_workers(client, wheel_path, package_name, expected_version)

   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.

   :param client: A connected ``dask.distributed.Client``.
   :param 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).
   :param 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.
   :param expected_version: Version string the installed wheel must report after installing,
                            or this raises ``RuntimeError``.


.. py:function:: get_client(scheduler_address = None, n_workers = None, install_package = False)

   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).

   :param scheduler_address: Address of a remote Dask scheduler.  ``None`` → use a local cluster.
   :param n_workers: Number of workers for a local cluster.  Ignored when connecting to a
                     remote scheduler.
   :param install_package: When ``True`` and using a remote scheduler, build + install
                           ``pyfracval`` on all workers before returning.

   :rtype: dask.distributed.Client