Adaptive scaling went above the number of available tasks

[mrocklin]

Currently Dask supports adaptive scaling. This allows it to scale the number of workers based on load. The logic for how many workers to ask for is based on many things, like how much work we have, how much memory we need, and so on.

I recently observed a cluster scale up beyond the number of available tasks. It shouldn't do this. I had three tasks, each of which took several minutes, and I saw my cluster scale up to ten workers. Probably there should be a maximum applied in there somewhere that wasn't being set. There appears to already be some code to do this (the intent is there) but perhaps there is a but. The relevant code is here

distributed/distributed/scheduler.py

Lines 5337 to 5390 in 4115f55

	def adaptive_target(self, comm=None, target_duration=None):
	"""Desired number of workers based on the current workload
	This looks at the current running tasks and memory use, and returns a
	number of desired workers. This is often used by adaptive scheduling.
	Parameters
	----------
	target_duration: str
	A desired duration of time for computations to take. This affects
	how rapidly the scheduler will ask to scale.
	See Also
	--------
	distributed.deploy.Adaptive
	"""
	if target_duration is None:
	target_duration = dask.config.get("distributed.adaptive.target-duration")
	target_duration = parse_timedelta(target_duration)
	# CPU
	cpu = math.ceil(
	self.total_occupancy / target_duration
	) # TODO: threads per worker
	# Avoid a few long tasks from asking for many cores
	tasks_processing = 0
	for ws in self.workers.values():
	tasks_processing += len(ws.processing)
	if tasks_processing > cpu:
	break
	else:
	cpu = min(tasks_processing, cpu)
	if self.unrunnable and not self.workers:
	cpu = max(1, cpu)
	# Memory
	limit_bytes = {addr: ws.memory_limit for addr, ws in self.workers.items()}
	worker_bytes = [ws.nbytes for ws in self.workers.values()]
	limit = sum(limit_bytes.values())
	total = sum(worker_bytes)
	if total > 0.6 * limit:
	memory = 2 * len(self.workers)
	else:
	memory = 0
	target = max(memory, cpu)
	if target >= len(self.workers):
	return target
	else: # Scale down?
	to_close = self.workers_to_close()
	return len(self.workers) - len(to_close)

[quasiben]

Do you have reproducer for the event ?

[twoertwein]

Do you have reproducer for the event ?

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
    with LocalCluster(
        n_workers=1, processes=True, threads_per_worker=1, memory_limit=0
    ) 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:

setup 0
setup 1
setup 2
setup 3
setup 6
setup 7
setup 5
setup 4
setup 8
setup 9
[0, 0, 1, 2]