Function get_scheduler_lock

dask/utils.py:1349–1373

Get an instance of the appropriate lock for a certain situation based on scheduler used.

(collection=None, scheduler=None)

def get_scheduler_lock(collection=None, scheduler=None):
    """Get an instance of the appropriate lock for a certain situation based on
    scheduler used."""
    from dask import multiprocessing
    from dask.base import get_scheduler

    actual_get = get_scheduler(collections=[collection], scheduler=scheduler)

    if actual_get == multiprocessing.get:
        return multiprocessing.get_context().Manager().Lock()
    else:
        # if this is a distributed client, we need to lock on
        # the level between processes, SerializableLock won't work
        try:
            import distributed.lock
            from distributed.worker import get_client

            client = get_client()
        except (ImportError, ValueError):
            pass
        else:
            if actual_get == client.get:
                return distributed.lock.Lock()

    return SerializableLock()


def ensure_dict(d: Mapping[K, V], *, copy: bool = False) -> dict[K, V]: