Timed out trying to connect ... : connect() didn't finish in time

[mnarodovitch]

Whereas the following is not a bug we can pinpoint, it really bothers us. After weeks of experimentation and investigation we didn't find a real solution.

We run something like

(a_bag
  .repartition(100_000) # outputs small parttitions
  .map(very_cpu_heavy_computation) # outputs ~4MB partitions
  .repartition(1000).to_dataframe().to_parquet() # writes ~30MB parquet files from ~400MB in-memory-partitions
)

on 100 workers with 1500 cpus on Kubernetes pods.

When it comes to the last re-partition step, dask starts blocking the event loop and the workers start spamming the following logs

distributed.utils_comm - INFO - Retrying get_data_from_worker after exception in attempt 0/2: Timed out trying to connect to 'tcp://10.2.3.20:38449' after 20 s: Timed out trying to connect to 'tcp://10.2.3.20:38449' after 20 s: connect() didn't finish in time

=> 5000 events

distributed.core - INFO - Event loop was unresponsive in Worker for 4.95s. This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.

=> 26,094 events

We are working around that, by setting

# less connections to avoid potential network saturation and connection timeouts
export DASK_DISTRIBUTED__WORKER__CONNECTIONS__OUTGOING=25
export DASK_DISTRIBUTED__WORKER__CONNECTIONS__INCOMING=5
export DASK_DISTRIBUTED__COMM__SOCKET_BACKLOG=16384

# graceful timeout and retry policy
export DASK_DISTRIBUTED__SCHEDULER__ALLOWED_FAILURES=20
export DASK_DISTRIBUTED__COMM__TIMEOUTS__CONNECT=20
export DASK_DISTRIBUTED__COMM__RETRY__COUNT=2

sothat, the job finally finishes, despite the issues in the communication.

Anyways, sometimes workers even start hanging and have to be restarted manually. Also the progress significantly slows down to the end of the job.

I'm sharing this with the community and hoping, that somebody may give pointers what to try, or even ideas for a resolution.

Environment:

• Dask version: 2.23.0
• Python version: 3.6
• Operating System: ubuntu on kubernetes
• Install method (conda, pip, source): pip

[mnarodovitch]

I have the suspicion, that this is related to the handshake, which is failing, when the event loop is blocked
https://github.com/dask/distributed/blob/master/distributed/comm/core.py#L297

[mrocklin]

I have the suspicion, that this is related to the handshake, which is failing, when the event loop is blocked

Interesting. This was added recently. We could test this hypothesis relatively easily by walking back to Dask version 2.22 and seeing if the problem persists. Is this easy for you to do?

[mnarodovitch]

Interesting. This was added recently. We could test this hypothesis relatively easily by walking back to Dask version 2.22 and seeing if the problem persists. Is this easy for you to do?

Had a look into our history and we tried 2.20, 2.21 and 2.23. All 'failing' with the timeouts. The runs are somewhat costly and take some hours. We didn't setup cheap reproducers, sofar. This rather happens at scale.

[mnarodovitch]

Digging out more logs

distributed.worker - ERROR - Worker stream died during communication: tcp://10.2.37.70:44241 |  
-- | --
  | Traceback (most recent call last): |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/comm/core.py", line 237, in connect |  
  | _raise(error) |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/comm/core.py", line 216, in _raise |  
  | raise IOError(msg) |  
  | OSError: Timed out trying to connect to 'tcp://10.2.37.70:44241' after 20 s: connect() didn't finish in time |  
  |   |  
  | During handling of the above exception, another exception occurred: |  
  |   |  
  | Traceback (most recent call last): |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/worker.py", line 1983, in gather_dep |  
  | self.rpc, deps, worker, who=self.address |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/worker.py", line 3258, in get_data_from_worker |  
  | return await retry_operation(_get_data, operation="get_data_from_worker") |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/utils_comm.py", line 390, in retry_operation |  
  | operation=operation, |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/utils_comm.py", line 370, in retry |  
  | return await coro() |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/worker.py", line 3235, in _get_data |  
  | comm = await rpc.connect(worker) |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/core.py", line 1013, in connect |  
  | **self.connection_args, |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/comm/core.py", line 249, in connect |  
  | _raise(error) |  
  | File "/usr/local/lib/python3.6/site-packages/distributed/comm/core.py", line 216, in _raise |  
  | raise IOError(msg) |  
  | OSError: Timed out trying to connect to 'tcp://10.2.37.70:44241' after 20 s: Timed out trying to connect to 'tcp://10.2.37.70:44241' after 20 s: connect() didn't finish in time

along with

 log='distributed.utils_comm - INFO - Retrying get_data_from_worker after exception in attempt 0/2: Timed out trying to connect to 'tcp://10.2.3.20:38449' after 20 s: Timed out trying to connect to 'tcp://10.2.3.20:38449' after 20 s: connect() didn't finish in time'

=> ~ 1,364 events

[mrocklin]

OK, if this was happening prior to 2.23 then it probably isn't the handshake.

I'm surprised to see the event loop being inactive for 5s . The next thing I would probably ask is what is slowing down the workers. One way to get this information is to create a performance_report or watch the dashboard and look at the /profile-server tab. That might give us a sense about what is getting in the way.

Separately, if I come up with some time I'll create a small snippet to test how long it takes for every worker to talk to every other worker. My guess is that this will return quickly, and that there is some issue with the combination of comms and whatever work is going on. I'm curious, is there anything suspect in the functions that you're running, or would this happen just as often with normal Pandas functions?

[mrocklin]

I would be curious what this returns on your system. On my laptop it's 15ms.

import asyncio
import time
from typing import List

from dask.distributed import Client, get_worker
from dask.utils import format_time


async def ping(addresses: List[str]):
    await asyncio.gather(*[
        get_worker().rpc(addr).identity()
        for addr in addresses
        if addr != get_worker().address
    ])


def all_ping(client: Client):
    workers = list(client.scheduler_info()["workers"])
    start = time.time()
    client.run(ping, workers)
    stop = time.time()
    print(format_time(stop - start))


if __name__ == "__main__":
    with Client() as client:
        all_ping(client)

Also, due to connection pooling, running this before your comptuation might make this smoother short term.

[mnarodovitch]

Thank you so much for looking into this! I'm happy to run all_ping and try the performance report to create better evidence on what's going on.
Will update the issue with more data in the upcoming week.

In respect to the functions I'm running: very_cpu_heavy_computation runs ML code in a child process spawned from a dask-worker-thread and outputs a list(dict). From my point of view, this is not very suspect, as the child process should not affect the event-loop or GIL anyhow. Dask performs very well on that tasks - without any hiccups.

The problems arise, when it starts re-partitioning the 100,000 output partitions to 1000.

I suspect, that the same would happen, if a repartition a synthetic bag of a comparable scale from 100,000 to 1000 partitions. This is also something to try out on our end.

[mnarodovitch]

Thanks again for looking into this. I got hooked up on that topic and happy to share a sort-of-reproducer running on the full-scale-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()

The reproducer is producing connection errors and timeouts for the comm between workers and scheduler. This is not exactly the failure mode, we have faced. In our case, the workers are loosing connection between themselves. I'm sharing the results anyways.

I produced a performance report and uploaded that to https://bl.ocks.org/michaelnarodovitch/19923648bc21002b0a4c39556aea00f9 .

I also produced verbose logs and share some highlights here. If somebody is interested, I also may share the logs themselves.

On the scheduler
distributed.core - DEBUG - Lost connection to <WORKER> while reading message: in <closed TCP>: Stream is closed. Last operation: None
=> 141,539 events

On the workers
distributed.utils_comm - INFO - Retrying functools.partial(<function PooledRPCCall.__getattr__.<locals>.send_recv_from_rpc at 0x7f71c27aa268>, keys=("('lambda-fe761bebe828abd999db601e6011be16', 55634)",)) after exception in attempt 0/2: in <closed TCP>: ConnectionResetError: [Errno 104] Connection reset by peer
=> 2,194 events

distributed.utils_comm - INFO - Retrying functools.partial(<function PooledRPCCall.__getattr__.<locals>.send_recv_from_rpc at 0x7ff3d7b1a840>, keys=("('lambda-fe761bebe828abd999db601e6011be16', 71105)",)) after exception in attempt 1/2: in <closed TCP>: TimeoutError: [Errno 110] Connection timed out
=> 11,370 events

OSError: Timed out trying to connect to '<SCHEDULER>' after 20 s: Timed out trying to connect to 'tcp://10.3.91.248:8786' after 20 s: connect() didn't finish in time
=> 39,922 events

Will share more information on our failure mode, when we run very_cpu_heavy_computation next time.

@mrocklin reporting back on the ping_all. Executed in 15ms. The network is very fast. AWS + some k8s overlay.

[mrocklin]

Well, looking at the performance report nothing is coming up unfortunately (we're entertainingly spending most of our time in the workers logging). We're not entirely sensitive here though. If we're spending a lot of time in GIL-holding functions we wouldn't see that.

This example is great though in that it's really well isolated, and probably stresses only our networking and serialization stack. I'll take a longer look later today on a cluster.

[mrocklin]

@mrocklin reporting back on the ping_all. Executed in 15ms. The network is very fast. AWS + some k8s overlay.

That's good to hear. I was curious about how things would scale there. Thanks!

[mrocklin]

OK, so I just ran this myself on a larger cluster. Here are the results

import coiled
cluster = coiled.Cluster(n_workers=40)

from dask.distributed import Client
client = Client(cluster)

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

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

# Get logs
logs = cluster.get_logs()
print(list(logs.values())[-1])

distributed.nanny - INFO -         Start Nanny at: 'tls://10.1.11.97:36845'
distributed.worker - INFO -       Start worker at:     tls://10.1.11.97:32879
distributed.worker - INFO -          Listening to:     tls://10.1.11.97:32879
distributed.worker - INFO -          dashboard at:           10.1.11.97:33013
distributed.worker - INFO - Waiting to connect to: tls://ip-10-1-13-173.us-east-2.compute.internal:8786
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                          4
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO -       Local Directory: /dask-worker-space/dask-worker-space/worker-z4lq642y
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO - Starting Worker plugin <https://beta.coiled.io/preloads/aws-credentials.p-7c0bb291-0d76-4b28-8fe0-e9ff648b9c53
distributed.worker - INFO -         Registered to: tls://ip-10-1-13-173.us-east-2.compute.internal:8786
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.batched - ERROR - Error in batched write
Traceback (most recent call last):
  File "/opt/conda/lib/python3.7/site-packages/distributed/batched.py", line 93, in _background_send
    payload, serializers=self.serializers, on_error="raise"
  File "/opt/conda/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
  File "/opt/conda/lib/python3.7/site-packages/distributed/comm/tcp.py", line 241, in write
    stream.write(b)
  File "/opt/conda/lib/python3.7/site-packages/tornado/iostream.py", line 553, in write
    self._write_buffer.append(data)
  File "/opt/conda/lib/python3.7/site-packages/tornado/iostream.py", line 177, in append
    b += data  # type: ignore
BufferError: Existing exports of data: object cannot be re-sized

cc @jakirkham as the new bytes-handling expert does the error message above mean anything to you?

[mnarodovitch]

That's something new to me :) Thanks for trying it out.

I digged a little bit into the comm code and didn't find anything suspicious there. I can't exclude that the connection timeouts on our end are related to our networking setup. I will update here, if I find anything interesting to share.

I suppose, that the reproducer captures the problem (on our end) quite well.

Maybe some other folks are also facing timeouts and can share their experiences here.

[mrocklin]

@quasiben is anyone on your team able to help deal with the comms issue here? It seems like it might be related to recent changes.