#4080

did more digging with the reproducer and identified the failure mode, which is causing connection timeouts and failed tasks on our 50 worker cluster.

b = bag.from_sequence([1]*100_000, partition_size=1)
bb = b.map_partitions(lambda *_, **__: [b'1'*2**20]*5).persist()
bc = bb.repartition(1000).persist()

1. For the repartition step, the workers will hit the scheduler with 100_000 who_has requests in a quite short period in time.
2. The hits will go through the ConnectionPool of Worker.scheduler, which has the default connection limit of 512. This will result in ~25,000 concurrent connection attempts and saturate the scheduler listener socket.

Logs would typically output:
Tornado connection attempt - message: Connect start: <SCHEDULER> 170,000 events - Worker side
Tornado connection success - message: Connect done 16,000 events - Worker side
Tornado connection accept - message: On connection 13,000 events - Scheduler side
Dask handshake success - message: handshake: 6,000 events - Worker and Scheduler side

Workers would typically fail tasks or crash due to connection timeouts.

3. Setting a limit of 8 concurrent connections to the scheduler per worker resolved the issue

I doubt that this is a good solution and do not intend to do further work on this PR. Nevertheless, I believe that this 'fix' helps to shed light on the root-cause of the issue.