Repository navigation
Task dependency bug in worker #4231
Description
What happened:
The following code fails sometimes with an KeyError error (3 out of 4 times). git bisect says the bug was introduced by #4107.
distributed.worker - ERROR - "('split-simple-shuffle-b50994a1a48d47067bc463a19b005e75', 14, 55)"
Traceback (most recent call last):
File "/home/nfs/mkristensen/repos/distributed/distributed/worker.py", line 1984, in gather_dep
deps_ts = [self.tasks[key] for key in deps]
File "/home/nfs/mkristensen/repos/distributed/distributed/worker.py", line 1984, in <listcomp>
deps_ts = [self.tasks[key] for key in deps]
KeyError: "('split-simple-shuffle-b50994a1a48d47067bc463a19b005e75', 14, 55)"
What you expected to happen:
For some reason the worker calls gather_dep() with a set of deps that the task does not depend on. As far as I can see, the client and scheduler maintains task dependencies correctly.
Minimal Complete Verifiable Example:
import pandas as pd
import numpy as np
import dask.dataframe as dd
from dask.dataframe.shuffle import shuffle
from distributed import wait
import dask
from distributed import Client, LocalCluster
nparts = 100
max_branch = 100
data_size = nparts * max_branch
def main(client):
df = pd.DataFrame({"x": np.arange(data_size)})
ddf = dd.from_pandas(df, npartitions=nparts)
ddf = ddf.persist(optimize_graph=False)
wait(ddf)
with dask.config.set({"optimization.fuse.active": False}):
s = shuffle(
ddf, ddf.x, shuffle="tasks", npartitions=nparts, max_branch=max_branch
)
s = s.persist()
wait(s)
if __name__ == "__main__":
with LocalCluster(scheduler_port=0, asynchronous=False, n_workers=5) as cluster:
with Client(cluster, asynchronous=False) as client:
main(client)cc. @gforsyth
Activity
gforsyth commentedon Nov 10, 2020
Thanks for the reproducer, @madsbk ! I'll take a look
quasiben commentedon Nov 10, 2020
I'm seeing the same thing in the nightly benchmark we are putting together:
quasiben/dask-scheduler-performance#2
gforsyth commentedon Nov 10, 2020
I can confirm the issue. Here's an inelegant fix while I figure out the correct one:
mrocklin commentedon Nov 12, 2020
Checking in. Is this the right approach? @gforsyth what is inelegant about this in particular? It appears to be blocking some benchmarking work as well.
Also, quick nit, I wonder how expensive it is to construct a TaskState object, and if instead we should do something like using
orhere:gforsyth commentedon Nov 12, 2020
I wanted to confirm that this was due to (I think) work stealing and not a request for the "wrong" dependencies. Tried to get a test that would reliably fail on this but didn't quite get there.
yes, I think this is the right way to do this, I'll push up the PR now
Create dependency TaskState as needed in gather_dep
Create dependency TaskState as needed in gather_dep (#4241)