In practice we often see longish delays for inter-worker communication or client-worker communication.
Round trip delay
On the very short end of this is a baseline 10-20ms roundtrip delay:
In [1]: from distributed import Client
In [2]: c = Client()
In [3]: %time c.submit(lambda x: x + 1, 1).result()
CPU times: user 16 ms, sys: 0 ns, total: 16 ms
Wall time: 17.7 ms
Out[3]: 2
I genuinely don't know how to measure what is responsible for this delay. It isn't incredibly important that we bring this down, but it is a simple case that might create efficiencies that are useful in other applications
Delay when under heavy load
Under heavy load we often find delays for communicating data between workers that are in the 100's of milliseconds up to, in some cases, full seconds. This is typically during large dataframe shuffles.
We have some mechanisms to help measure here. Using the digests in workers we can measure how far off a regular 20ms tornado heartbeat deviates. We also time the duration around yield connect statements. This time can be non-trivial in real networks. It may be worth keeping inter-worker connections alive in a connection pool. If we do this then we should ensure that we don't run up against open file limits.
I don't have a very good approach on how to identify these issues. @pitrou , if you have interest in diving in here I would appreciate it.
Here is a tiny example, I don't think that this is sufficiently representative though:
In [1]: from distributed import LocalCluster, Client, progress
In [2]: cluster = LocalCluster(nanny=False, n_workers=30, threads_per_worker=1)
In [3]: client = Client(cluster)
In [4]: import dask.dataframe as dd
df =
In [5]: df = dd.demo.make_timeseries('2000', '2001',
...: {'value': float, 'name': str, 'id': int},
...: freq='2H', partition_freq='7D', seed=1)
...: df = client.persist(df)
...: df
...:
Out[5]:
Dask DataFrame Structure:
id name value
npartitions=52
2000-01-01 int64 object float64
2000-01-08 ... ... ...
... ... ... ...
2000-12-23 ... ... ...
2000-12-30 ... ... ...
Dask Name: make-timeseries, 52 tasks
In [6]: %time len(df)
CPU times: user 220 ms, sys: 44 ms, total: 264 ms
Wall time: 263 ms
Out[6]: 4368
In practice we often see longish delays for inter-worker communication or client-worker communication.
Round trip delay
On the very short end of this is a baseline 10-20ms roundtrip delay:
I genuinely don't know how to measure what is responsible for this delay. It isn't incredibly important that we bring this down, but it is a simple case that might create efficiencies that are useful in other applications
Delay when under heavy load
Under heavy load we often find delays for communicating data between workers that are in the 100's of milliseconds up to, in some cases, full seconds. This is typically during large dataframe shuffles.
We have some mechanisms to help measure here. Using the digests in workers we can measure how far off a regular 20ms tornado heartbeat deviates. We also time the duration around
yield connectstatements. This time can be non-trivial in real networks. It may be worth keeping inter-worker connections alive in a connection pool. If we do this then we should ensure that we don't run up against open file limits.I don't have a very good approach on how to identify these issues. @pitrou , if you have interest in diving in here I would appreciate it.
Here is a tiny example, I don't think that this is sufficiently representative though: