Adaptive target duration is scaled by `threads_per` worker

[willirath]

Adaptive Clusters with threads_per_worker > 1 will scale to reach a duration target_duration / threads_per_worker:

from dask.distributed import LocalCluster, Client, Adaptive
from dask import bag as db
from time import time, sleep

cluster = LocalCluster(n_workers=1, threads_per_worker=2, memory_limit=1e9)
ca = cluster.adapt(minimum=1, maximum=10, target_duration="20s", scale_factor=1)

client = Client(cluster)

start = time()
db.from_sequence((0.5 for n in range(100)), npartitions=100).map(sleep).compute()
print(time() - start, "seconds with", len(cluster.workers), "workers")

9.84409236907959 seconds with 3 workers

(Note that I set scale_factor=1 to prevent anything but the target duration from scaling up the cluster.)

With single-threaded workers, we get what we desired (approx. 20s):

[...]
cluster = LocalCluster(n_workers=1, threads_per_worker=1, memory_limit=1e9)
ca = cluster.adapt(minimum=1, maximum=10, target_duration="20s", scale_factor=1)
[...]

18.264188766479492 seconds with 3 workers

• Should this be documented?
• Or should behaviour of Adaptive be corrected to include the number of threads per worker in the estimate of the expected target duration?

[guillaumeeb]

Yep, cores per worker is not taken into account here :

distributed/distributed/deploy/adaptive.py

Line 295 in 7ebe659

target = math.ceil(self.scheduler.total_occupancy / self.target_duration)

.

Not sure if we can get and be sure of the number of cores per new workers, but de can probably make a good assumption using the number of workers and total cores.

I think the current behavior is not the one intended.

[AndreaGiardini]

distributed/distributed/scheduler.py

Lines 5357 to 5360 in 4115f55

	# CPU
	cpu = math.ceil(
	self.total_occupancy / target_duration
	) # TODO: threads per worker

This started to hit us heavily. Can you point me in the right direction to implement this? Any suggestion?