add timeout option with default to 10s

[prasunanand]

Fixes https://github.com/dask/distributed/issues/3489

[prasunanand]

Thanks @TomAugspurger, I have modified the Pull Request. Now I raise KeyError if key is not found in the graph.

Thanks @jrbourbeau, I got confused because of line . I thought a callback timeout was excepted.

[gsailer]

Considering the original issue maybe you could go for a test similar to this one:

def test_get_unknown_key(c):
    result = delayed(add)(1, 2)
    graph = result.__dask_graph__()
    with pytest.raises(KeyError):
        c.get(graph, ["does_not_exist"])

[prasunanand]

Thanks @sublinus and @jrbourbeau , I have made the changes as you suggested.

[TomAugspurger]

Just to verify the desired behavior here... On master, if I do

# process 1
>>> client1 = Client("path/to/scheduler")
>>> client1.get({}, "a")

# process 2
>>> client2 = Client("path/to/scheduler")
>>> client2.get({"a": 1}, "a")

Then the behavior I see is that client1 gets it's result now that the scheduler knows about that key.

IIUC, this PR will change that behavior. Is that something we would want to support, or should we require that the requested key be in the actual graph submitted?

[jrbourbeau]

That's a good point, thanks for bringing it up @TomAugspurger. I personally find the current behavior in master non-intuitive - though this is subjective. Instead I'd prefer to guide users towards creating a Future for a key and then calling result() on that Future. I'm certainly open to hearing about what others think though

[TomAugspurger]

Agreed that the current behavior is unintuitive. I'm comfortable claiming that it "accidentally worked" unless we document that behavior somewhere.