dask / dask/distributed

gather(direct=True) may fall back to direct=False

Open
#7,993 0 comments 0 reactions 0 assignees View on GitHub
bug help wanted
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

The current implementation for

- `Client(...).gather(..., direct=True)`
- `Client(direct_to_workers=True).gather(...)`
- `get_client().gather(...)` (worker clients)

is that, if *anything* bad happens on the direct connection to the workers, the keys that failed to be fetched directly are fetched through the scheduler `(direct=False)` instead.

I believe that the intent of this design is to build resilience when the user specifies direct=True while in fact they have a firewall preventing comms.

First, I think even the original intent is not a good idea to begin with - most firewalls will just silently drop the packets, thus causing the user to wait ~30 seconds for timeout. If the workflow takes e.g. ~2 minutes to compute, the user may not realise there's a substantial slowdown.

Second, there are many cases where a key will fail to be gathered on the first try - notably, the who_has dict can fall out of sync with the scheduler, or a worker's GIL can be temporarily clogged. This may cause a workflow that gathers huge amounts of data from workers to client to run fine *most times* and then, once in a while, kill off the scheduler due to memory pressure. This is particularly true when the client mounts vastly more RAM than the scheduler - which is not an uncommon configuration, particularly when worker_clients are involved.

I suggest we drop this feature completely. If workers are permanently unreachable with direct=True due to network configuration, gather(direct=True) should just not work. Note that the default is direct=False.

CC @fjetter @hendrikmakait

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.