limit refers to our target maximal amount of memory. total refers to the amount of memory used currently. I think that the logic was correct before. Are you running into something that caused you to investigate this?

limit refers to our target maximal amount of memory. total refers to the amount of memory used currently.

Doesn't this mean that we want limit > total to create new workers - currently it is the opposite.

Are you running into something that caused you to investigate this?

Yes, the linked dask_jobqueue issue but the same also happens for the LocalCluster (see testcase test_adaptive_memory_limit). The important part is that a non-zero memory limit is set for the workers (it works if it is zero):

import time

from dask.distributed import Client, LocalCluster
from distributed.worker_client import get_worker


def test_fun():
    time.sleep(5)
    return get_worker().name


if __name__ == "__main__":
    # create simple task graph
    n = 4
    graph = {f"test_{i}": (test_fun,) for i in range(n)}
    targets = [f"test_{i}" for i in range(n)]

    workers = 10  # we shouldn't need more than n=4 workers

    # LocalCluster
    kwargs = {
        "n_workers": 1,
        "diagnostics_port": None,
        "threads_per_worker": 1,
        "memory_limit": "2G",
        "processes": True,
    }
    with LocalCluster(**kwargs) as cluster:
        cluster.adapt(minimum=1, maximum=workers)
        with Client(cluster) as client:
            client.register_worker_callbacks(
                lambda dask_worker: print("setup", dask_worker.name)
            )
            print(client.get(graph, targets))  # prints only 0s. If memory_limit=0, more workers are spawned

I think the remaining tests would pass if #4108 was fixed.

@mrocklin if you do not agree with the memory-related part, I can separate the "not spawning more jobs than runnable tasks" into its own PR

bump @mrocklin @guillaumeeb

Other people seem to encounter the same issue (SLURMCluster not adapting) esi-neuroscience/acme#14

I have been using this patch for more than a month on a SLURM cluster and it addressed my main concern (wasting resources) using dask on a cluster:

• do not spawn more workers than necessary and
• down-scale number of workers: I have typically one long'ish task left at the end, no need to block resources with idling workers.

limit refers to our target maximal amount of memory. total refers to the amount of memory used currently.

Doesn't this mean that we want limit > total to create new workers - currently it is the opposite.

I don't think so. For example, if I have a cluster with 100TB of memory (limit=1e14) and it stores no data (total=0) then I think that we don't want to scale up.

thank you @mrocklin for your comments! I think I understand my misconception :)

I will create a separate PR for the number of runnable task (after I switched to TaskGroups) and then spend more time to analyze why Local/SLURMCluster do not spawn more workers in my setup. Based on my current understanding it shouldn't be related to memory at all (must be the CPU part).

I think that all changes in this PR are not addressing the root cause: self.total_occupancy doesn't seem to get updated well enough (neither for SLURMCluster nor LocalCluster). I would expect that if a task is sleeping for 30s, that total_occupancy will increase and at some point reach 30. By default, the targeted duration is 5s. As soon as the total_occupancy is above 5s, dask should spawn more workers. Unfortunately, total_occupancy doesn't ever go above 2.0 in my test examples even though wall time is much larger.

If all tasks take much longer than spawning a worker, something like the following is an ugly workaround:

with dask.config.set({"distributed.adaptive.target-duration": 0.1}):
    ...

The only reason we don't see this behavior when no memory limit is set (limit = 0), is that the if condition for memory is always True for limit = 0.

Iterating over TaskGroup.status does not seem to contain all tasks. Might just be missing the 'no-worker' cases?

For the record, the initial problem was that the graph used for the computation had not enough information to enable adaptive scaling. See dask/dask-jobqueue#463.