worker_client KeyError

[gsakkis]

Describe the issue:

I'm getting a similar error to this old issue even though rejoin is called. Based on some print statements in secede and rejoin and assuming that the order of output is the same as the order of execution, it seems it might be due to non deterministic ordering.

Relevant output:

seceded <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>
rejoined <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>
seceded <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>
seceded <Thread(Dask-Default-Threads-4156015-0, started daemon 140222050313984)>
rejoined <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>
2025-03-31 21:12:54,970 - distributed.protocol.pickle - ERROR - Failed to serialize <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>.
Traceback (most recent call last):
  File "/mnt/disks/data/miniforge3/envs/etl/lib/python3.11/site-packages/distributed/worker_client.py", line 71, in worker_client
    secede()  # have this thread secede from the thread pool
    ^^^^^^^^
  File "/mnt/disks/data/miniforge3/envs/etl/lib/python3.11/site-packages/distributed/threadpoolexecutor.py", line 121, in secede
    thread_state.executor._threads.remove(threading.current_thread())
KeyError: <Thread(Dask-Default-Threads-4156020-0, started daemon 140297992402688)>

...
rejoined <Thread(Dask-Default-Threads-4156015-0, started daemon 140222050313984)>

Minimal Complete Verifiable Example:

Although it's reproducible in my environment, coming up with a MCVE may be tricky.

Environment:

• Dask version: 2025.3.0
• Python version: 3.11.10
• Operating System: Linux Debian 5.10.197-1
• Install method (conda, pip, source): pip

[gsakkis]

FWIW worker_client(separate_thread=False) seems to work fine but I don't know its ramifications. The closed issue mentions that the documentation warns about setting it to False can result in a deadlock but there's no such warning in the current docs (or for that matter any hints about when should one pass True vs False).

[gsakkis]

It turns out the issue is that there's a second nested worker_client call. MCVE:

from dask import delayed
from distributed import Client, worker_client, as_completed


def transform(n):
    return n + 1


def load(n):
    print("Loading", n)


def transform_range(n):
    with worker_client() as client:
        futures = client.map(transform, range(n))
        results = [f.result() for f in as_completed(futures)]
    return results


def run(n):
    with worker_client() as client:
        futures = []
        for t in transform_range(n):
            futures.append(client.submit(load, t))
        for f in as_completed(futures):
            print("Loaded", f.result())


if __name__ == "__main__":
    with Client():
        delayed(run)(4).compute()