Future submit()/result() takes very long time to complete with 8MB Python object

[adbreind]

What happened: Smallish "Hello world" style submit/result takes a long time and sometimes produces memory warnings.

What you expected to happen: Fast end-to-end result (milliseconds)

Minimal Complete Verifiable Example:

Python 3.9.12 | packaged by conda-forge | (main, Mar 24 2022, 23:23:20) 
Type 'copyright', 'credits' or 'license' for more information
IPython 8.3.0 -- An enhanced Interactive Python. Type '?' for help.

In [1]: from distributed import Client, LocalCluster

In [2]: cluster = LocalCluster(n_workers=2, threads_per_worker=1, memory_limit='
   ...: 16GiB')

In [3]: client = Client(cluster)

In [4]: def foo():
   ...:     return [1.5] * 1_000_000
   ...: 

In [5]: %time res = client.submit(foo).result()
CPU times: user 7.44 s, sys: 567 ms, total: 8.01 s
Wall time: 20.9 s

Anything else we need to know?:

This issue has been happening in various flavors for 2+ years (I think 2019-vintage Dask does not have this issue)

Environment:

• Dask version: 2022.1.0
• Python version: 3.9.12
• Operating System: MacOS, Linux
• Install method (conda, pip, source): conda install coiled-runtime (but have observed the issue for years with pip or conda installs of dask and distributed

[mrocklin]

This probably has far less to do with the size of the data, and much more to do with the fact that it's a list. My guess is that for lists we iterate through the list and see if there are any large objects we should be pulling out to send separately (like a list of numpy arrays, or cudf dataframes).

For example, I suspect that if the result here was a numpy array of the same size that it would be trivially fast.

@madsbk or other folks who think about the protocol might be able to provide more insight.

[fjetter]

I think this is connected to us wrapping Serialize ...

from distributed.protocol import Serialize, dumps
long_list = [1.5] * 1_000_000

In [9]: %time _ = dumps(long_list)
CPU times: user 78.9 ms, sys: 90.9 ms, total: 170 ms
Wall time: 172 ms

In [10]: %time _ = dumps(Serialize(long_list))
{}
CPU times: user 5.53 s, sys: 131 ms, total: 5.66 s
Wall time: 5.66 s

[fjetter]

Either way, py-spy flags dumps

[ian-r-rose]

I believe that this issue is the cause of dask/dask#9411. Now, a valid solution is to ensure that we are passing around numpy/pandas objects, but I think that underestimates just how badly the serialization protocol here does with medium-to-large python collections.