Repository navigation
Deadlock after few workers take all tasks #5635
Description
What happened:
I have a dask-gateway cluster on k8s, and a 63k task graph. The processing has essentially stalled. Five workers have 375 tasks each, while the remaining 75 have zero. The scheduler logs have a bunch of timeout errors. The workers with all the tasks have the "unresponsive... long-running ... GIL" errors -- though it look like the running tasks are stack.
I attempted to use dump_cluster_state, but hit a timeout error.
Anything else we need to know?:
After deleting the pods with the 5 wedged workers, the graph completed.
Environment:
- Dask version: 2021.12.0 (distributed was from a few commits after 2021.12.0)
- Python version: 3.8.12
- Operating System: Ubuntu 20.04
- Install method (conda, pip, source): pip
Cluster Dump State:
---------------------------------------------------------------------------
TimeoutError Traceback (most recent call last)
/opt/conda/lib/python3.8/site-packages/distributed/comm/core.py in connect()
318 # write, handshake = await asyncio.gather(comm.write(local_info), comm.read())
--> 319 handshake = await asyncio.wait_for(comm.read(), time_left())
320 await asyncio.wait_for(comm.write(local_info), time_left())
/opt/conda/lib/python3.8/asyncio/tasks.py in wait_for()
500 await _cancel_and_wait(fut, loop=loop)
--> 501 raise exceptions.TimeoutError()
502 finally:
TimeoutError:
The above exception was the direct cause of the following exception:
OSError Traceback (most recent call last)
/tmp/ipykernel_894/3630066435.py in <module>
----> 1 client.dump_cluster_state(f"dump_{name}_{date}")
/opt/conda/lib/python3.8/site-packages/distributed/client.py in dump_cluster_state(self, filename, exclude, format)
3592
3593 """
-> 3594 return self.sync(
3595 self._dump_cluster_state,
3596 filename=filename,
/opt/conda/lib/python3.8/site-packages/distributed/utils.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
308 return future
309 else:
--> 310 return sync(
311 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
312 )
/opt/conda/lib/python3.8/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
362 if error[0]:
363 typ, exc, tb = error[0]
--> 364 raise exc.with_traceback(tb)
365 else:
366 return result[0]
/opt/conda/lib/python3.8/site-packages/distributed/utils.py in f()
347 if callback_timeout is not None:
348 future = asyncio.wait_for(future, callback_timeout)
--> 349 result[0] = yield future
350 except Exception:
351 error[0] = sys.exc_info()
/opt/conda/lib/python3.8/site-packages/tornado/gen.py in run(self)
760
761 try:
--> 762 value = future.result()
763 except Exception:
764 exc_info = sys.exc_info()
/opt/conda/lib/python3.8/site-packages/distributed/client.py in _dump_cluster_state(self, filename, exclude, format)
3495 )
3496 versions = self._get_versions()
-> 3497 scheduler_info, worker_info, versions_info = await asyncio.gather(
3498 scheduler_info, worker_info, versions
3499 )
/opt/conda/lib/python3.8/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
884 name, comm.name = comm.name, "ConnectionPool." + key
885 try:
--> 886 result = await send_recv(comm=comm, op=key, **kwargs)
887 finally:
888 self.pool.reuse(self.addr, comm)
/opt/conda/lib/python3.8/site-packages/distributed/core.py in send_recv(comm, reply, serializers, deserializers, **kwargs)
677 if comm.deserialize:
678 typ, exc, tb = clean_exception(**response)
--> 679 raise exc.with_traceback(tb)
680 else:
681 raise Exception(response["exception_text"])
/opt/conda/lib/python3.8/site-packages/distributed/core.py in handle_comm()
519 result = asyncio.ensure_future(result)
520 self._ongoing_coroutines.add(result)
--> 521 result = await result
522 except (CommClosedError, CancelledError):
523 if self.status in (Status.running, Status.paused):
/opt/conda/lib/python3.8/site-packages/distributed/scheduler.py in broadcast()
6018 return resp
6019
-> 6020 results = await All(
6021 [send_message(address) for address in addresses if address is not None]
6022 )
/opt/conda/lib/python3.8/site-packages/distributed/utils.py in All()
206 while not tasks.done():
207 try:
--> 208 result = await tasks.next()
209 except Exception:
210
/opt/conda/lib/python3.8/site-packages/distributed/scheduler.py in send_message()
6010
6011 async def send_message(addr):
-> 6012 comm = await self.rpc.connect(addr)
6013 comm.name = "Scheduler Broadcast"
6014 try:
/opt/conda/lib/python3.8/site-packages/distributed/core.py in connect()
1069 except Exception as exc:
1070 self.semaphore.release()
-> 1071 raise exc
1072 finally:
1073 self._connecting.discard(fut)
/opt/conda/lib/python3.8/site-packages/distributed/core.py in connect()
1053 )
1054 self._connecting.add(fut)
-> 1055 comm = await fut
1056 comm.name = "ConnectionPool"
1057 comm._pool = weakref.ref(self)
/opt/conda/lib/python3.8/site-packages/distributed/comm/core.py in connect()
322 with suppress(Exception):
323 await comm.close()
--> 324 raise OSError(
325 f"Timed out during handshake while connecting to {addr} after {timeout} s"
326 ) from exc
OSError: Timed out during handshake while connecting to tls://10.16.184.3:46165 after 30 s
Reacted by Christos Antoniou and vladvalkov
Activity
fjetter commentedon Jan 14, 2022
I assume these are the five workers which have so much data spilled on the dashboard? (See bytes stored by worker)
Anything interesting happening in this function? E.g. is there native code running which may not release the GIL?
That strongly indicates another state machine problem.
chrisroat commentedon Jan 14, 2022
I think the
stacktask is just ada.stackcall, though it's possible that optimization fused some other things. The CPU has dropped to zero.I'm not sure how to better help diagnose the state machine issues.
I find most problems with autoscaling clusters. I could set the cluster size to very high numbers, but the workloads vary a lot and this would be wasteful. I already waste a lot because my autoscale lower bound needs to be high enough to prevent
KilledWorkerissues while waiting for new workers to come online... so the tail end of the graph where a few tasks are left has very low utilization.fjetter commentedon Jan 18, 2022
These timeouts are not unexpected in such an auto scaling situation. The latest release (2022.01.0) includes a fix to the cluster dump function to not fail in case there are timeouts to individual workers.
#5590
chrisroat commentedon Jan 20, 2022
I just encountered a new mode, where the dashboard says 3 tasks are processing but the graph in the lower left shows 0 (and the Info page shows no workers with anything in "processing"). I typically have to delete a pod to unstick things, but this time I don't know which pod!
I searched through the logs for "overlap-getitem", since that is the most upstream task prefix that is wedged. In the scheduler, I see 12 lines in succession (always the same key, but several different workers) in the scheduler:
I tried deleting the workers mentioned, some of which are still around, but that did not bring back processing of the missing keys.
About 25 seconds before those log entries, I see:
Some 50 minutes later:
That key is seen in the logs as a worker being retired just a few seconds prior to the "Unexpected worker completed task" messages. The same key is re-registered in the middle of those "Unexpected worker completed task" messages.
The cluster dump would not upload to github, so I put it here:
https://drive.google.com/file/d/1okQ6M8M8smP0oBhvlrdbQLJJbWp7_32U/view?usp=sharing
bnaul commentedon Jan 22, 2022
We have recently started seeing this behavior as well, currently on dask+distributed 2022.1.0 but I think it's been happening for a couple of releases. Same pattern as @chrisroat:
client.processing()shows nothing in progress but the Status page still shows the correct progress value@crusaderky @fjetter any chance #5665 could be related given that this worker was left for dead but then kept going...?
crusaderky commentedon Jan 24, 2022
@bnaul #5665 mitigates, but does not fix, the issue.
Everything in the opening post, particularly the memory usage graph of the stuck workers, seems to indicate that this may be a duplicate of #3761:
I can see from your dashboard that 4 workers are paused with extremely high amounts of spilled memory.
They never get out of paused state and they hold a bunch of queued tasks, which are not returned to the scheduler.
The reason why they don't get out of paused state may be (educated guess) that you ran out of disk space and they can't swap out anymore. As of the latest release, the error handling for this use case is fragile - see #5364 for improvement.
I currently have two changes in my pipeline:
While you wait for those, I would suggest you try to figure out why you have 180 GiB worth of managed data per worker on 4 workers while the rest of the cluster is relatively free.
Do those workers contain replicas of data that already exists elsewhere? In that case, the new Active Memory Manager could help you: https://coiled.io/blog/introducing-the-dask-active-memory-manager/
You can also try running rebalance() periodically. Word of warning - the method is not safe to run on a busy scheduler. Making it safe and automatic is also in my pipeline.
chrisroat commentedon Jan 24, 2022
I see this consolidation of tasks a few workers when I use auto-scaling and spot instances. I think the addition/deletion of workers is throwing a curveball to the scheduler. I tried to repro in #4471 (no deadlock -- just bad scheduling), where you see a small number of workers get all the tasks as the cluster scales up.
The graph above is a map_overlap on a 3d-chunked array with 21720=680 chunks, reading from and writing to the zarr format. This leads to 82k tasks(!). I notice in an issue you link that one can adjust fuse parameters. Is that worth trying -- could it reduce the number of tasks and perhaps lower the rate of wonky scheduling & deadlocks?
FWIW, I did attempt the AMM at one point. If I set the env var DASK_DISTRIBUTED__SCHEDULER__ACTIVE_MEMORY-MANAGER__START to "true", it turns on -- do I need additional settings?
crusaderky commentedon Jan 24, 2022
Do you mean that the 4 workers that I see saturated in the dashboard are the initial ones and the rest of the cluster only goes online when they are already very full?
fuse parameters can reduce the number of tasks, which in turn translates in a reduction in CPU load on the scheduler. If your scheduler CPU is frequently near 100%, this should in turn translate in a reduction in random timeouts and disconnects. I don't think this is your case - unless you're frequently seeing them in the logs.
Should be enough. Please run
client.amm.running()to double check.Could you run
client.rebalance()every once in a while? I wouldn't recommend it as a production fix (it's not robust in the main branch while computations are running), but if it does help you it will give me a case for raising the priority of the rework I'm doing on it.chrisroat commentedon Jan 24, 2022
I don't think so. In the issue I references (#4471), I scale from 1 to 4 to 20 workers. It looks like 6 workers get the bulk of the load, and a few others get some load.
I do see a lot of comm errors in the logs.
I could try that on some future run. I do see tons of GIL errors, which I guess must be the read/write to disk that is happening. It does seem like distributing the work better would mitigate things.
fjetter commentedon Feb 15, 2022
FYI I recently merged a PR which might help in this situation. At least the failure scenario I fixed there is likely to happen in scaling situations, particularly if workers are removed #5786
Note that we got already another deadlock reported (off github) after this fix so we're not over the hill, yet.
mrocklin commentedon Apr 12, 2022
The "few workers taking all the tasks" thing is explainable and fixable by #6115 .
This doesn't explain things fully halting though, unless possibly you've run out of disk space (is this possible?)