Truncate large payloads when debug-logging received messages via comms

[hendrikmakait]

This issue came out of #8463. Currently, we log all incoming messages as DEBUG logs:

distributed/distributed/core.py

Line 901 in 33b2c72

logger.debug("Message from %r: %s", address, msg)

This can become a severe performance issue with large incoming requests, in particular with push-based systems like P2P. We should probably truncate large incoming payloads.

[nprihodko]

The issue seems to be present not only in debug-level messages, but at least also in info messages.

Managed to get this (cannot come up with a short reproducible example but it happened during merging):

2024-01-17 10:47:07,146 - distributed.utils_comm - INFO - Retrying functools.partial(<bound method ShuffleRun._send of <DataFrameShuffleRun: id='77201d9aa0bf473b1451a90af7e015de', run_id=6, local_address='tls://10.164.0.24:38325', closed=False,
transferred=False>>, 'tls://10.164.0.7:34781', [(33, b'\xff\xff\xff\xff\xa0\x05\x00\x00\x10\x00\x00\x00\x00\x00\n\x00\x0e\x00\x06\x00\x05\x00\x08\x00\n\x00\x00\x00\x00\x01\x04\x00\x10\x00\x00\x00\x00\x00\n\x00\x0c\x00\x00\x00\x04\x00\x08\x00\n\x
00\x00\x00(\x04\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x0c\x00\x00\x00\x08\x00\x0c\x00\x04\x00\x08\x00\x08\x00\x00\x00\x08\x00\x00\x00\x10\x00\x00\x00\x06\x00\x00\x00pandas\x00\x00\xf2\x03\x00\x00{"index_columns": ["__index_level_0__"], "column
_indexes": [{"name": null, "field_name": null, "pandas_type": "unicode", "numpy_type": "object", "metadata": {"encoding": "UTF-8"}}], "columns": [{"name": "id", "field_name": "id", "pandas_type": "int64", "numpy_type": "Int64",
 "metadata": null}, {"name": "date "field_name": "date", "pandas_type": "datetime", "numpy_type": "datetime64[ns]", "metadata": null}, {"name": "value", "field_name": "value", "pandas_type": "int32", "numpy_ty
pe": "Int32", "metadata": null}, {"name": "__hash_partition", "field_name": "__hash_partition", "pandas_type": "uint64", "numpy_type": "uint64", "metadata": null}, {"name": "_worker", "field_name": "_worker", "pandas_type": "int8", "numpy_type":
 "int8", "metadata": null}, {"name": null, "field_name": "__index_level_0__", "pandas_type": "int64", "numpy_type": "int64", "metadata": null}], "creator": {"library": "pyarrow", "version": "14.0.2"}, "pandas_version": "2.1.4"}\x00\x00\x05\x00\x
00\x00\x0c\x01\x00\x00\xc4\x00\x00\x00\x84\x00\x00\x00D\x00\x00\x00\x04\x00\x00\x00\x18\xff\xff\xff\x00\x00\x01\x02\x10\x00\x00\x00$\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00__index_level_0__\x00\x00\x00\x10\xff\xff\xff\x00\x00
\x00\x01@\x00\x00\x00T\xff\xff\xff\x00\x00\x01\x02\x10\x00\x00\x00(\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x00__hash_partition\x00\x00\x06\x00\x08\x00\x04\x00\x06\x00\x00\x00@\x00\x00\x00\x90\xff\xff\xff\x00\x00\x01\x02\x10\x00\
x00\x00$\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00value\x00\x00\x00\x88\xff\xff\xff\x00\x00\x00\x01 \x00\x00\x00\xcc\xff\xff\xff\x00\x00\x01\n\x10\x00\x00\x00 \x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\t\x00\x00\x
00date\x00\x06\x00\x08\x00\x06\x00\x06\x00\x00\x00\x00\x00\x03\x00\x10\x00\x14\x00\x08\x00\x06\x00\x07\x00\x0c\x00\x00\x00\x10\x00\x10\x00\x00\x00\x00\x00\x01\x02\x10\x00\x00\x00$\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x0b\x00\x00\x00id
\x00\x08\x00\x0c\x00\x08\x00\x07\x00\x08\x00\x00\x00\x00\x00\x00\x01@\x00\x00\x00\xff\xff\xff\xffH\x01\x00\x00\x14\x00\x00\x00\x00\x00\x00\x00\x0c\x00\x16\x00\x06\x00\x05\x00\x08\x00\x0c\x00\x0c\x00\x00\x00\x00\x03\x04\x00\x18\x00\x00\
x00\x80\x9e\x99\x00\x00\x00\x00\x00\x00\x00\n\x00\x18\x00\x0c\x00\x04\x00\x08\x00\n\x00\x00\x00\xbc\x00\x00\x00\x10\x00\x00\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\n\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x
00\x00\x00\x00\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x00pFD\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00pFD\x00\x00\x00\x00\x00\xa0\x11\
x11\x00\x00\x00\x00\x00\x10XU\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x10XU\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x00H{w\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00H{w\x00\x00\x00\x00\x008#"\x00\x00\x00\x00\x00\x00\x00\x00\x
00\x05\x00\x00\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x0
0\x00gD\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x
00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x0
0\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xd5\x07\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2
\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x0
0\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00\xf2\t\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x0
0K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00K\n\x00\x00\x00\x00\x00\x00\x12\x13\x00\x00\x00

+ 17 MiB of similar data.

[maawoo]

Also happens with ArrayRechunkRun:

2024-01-24 16:45:06,450 - distributed.utils_comm - INFO - Retrying functools.partial(<bound method ShuffleRun._send of <ArrayRechunkRun: id='d6f8b2d9ed84fca0a899360d7ddf2b2b', run_id=2, local_address='tcp://172.18.1.10:33341', closed=False, transferred=False>>, 'tcp://172.18.1.11:41893', [((1616, 0, 0), [((0, 0, 31), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        ...,
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan]]], dtype=float32))), ((0, 0, 32), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        ...,
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan]]], dtype=float32))), ((0, 1, 31), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        ...,
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan]]], dtype=float32))), ((0, 1, 32), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        ...,
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan]]], dtype=float32))), ((0, 2, 31), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        ...,
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan]]], dtype=float32))), ((0, 2, 32), ((1616, 0, 0), array([[[nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],
        [nan, nan, nan, ..., nan, nan, nan],

...

• 17 MiB of similar data.

I can top that with around 200 MiB