payload: {"action":"created","issue":{"url":"https://api.github.com/repos/dask/distributed/issues/2628","repository_url":"https://api.github.com/repos/dask/distributed","labels_url":"https://api.github.com/repos/dask/distributed/issues/2628/labels{/name}","comments_url":"https://api.github.com/repos/dask/distributed/issues/2628/comments","events_url":"https://api.github.com/repos/dask/distributed/issues/2628/events","html_url":"https://github.com/dask/distributed/issues/2628","id":435294422,"node_id":"MDU6SXNzdWU0MzUyOTQ0MjI=","number":2628,"title":"Unexpected `CancelledFuture` error","user":{"login":"cicdw","id":13255838,"node_id":"MDQ6VXNlcjEzMjU1ODM4","avatar_url":"https://avatars0.githubusercontent.com/u/13255838?v=4","gravatar_id":"","url":"https://api.github.com/users/cicdw","html_url":"https://github.com/cicdw","followers_url":"https://api.github.com/users/cicdw/followers","following_url":"https://api.github.com/users/cicdw/following{/other_user}","gists_url":"https://api.github.com/users/cicdw/gists{/gist_id}","starred_url":"https://api.github.com/users/cicdw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cicdw/subscriptions","organizations_url":"https://api.github.com/users/cicdw/orgs","repos_url":"https://api.github.com/users/cicdw/repos","events_url":"https://api.github.com/users/cicdw/events{/privacy}","received_events_url":"https://api.github.com/users/cicdw/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2019-04-19T20:24:41Z","updated_at":"2019-04-23T01:52:55Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"Over at https://github.com/PrefectHQ/prefect/issues/943 we're running into a very interesting situation; we believe we know how to fix it in our own codebase, but I'd like to understand whether this is a fundamentally bad Dask pattern or whether this is a bug.  \r\n\r\nIn short, we are using multiple `worker_client`s and `fire_and_forget`, and are seeing `CancelledFuture` errors in very unexpected circumstances.  These errors seem to only occur in _very_ specific circumstances, but not otherwise.\r\n\r\nTo reproduce, start a local cluster:\r\n```\r\ndask-scheduler\r\ndask-worker --nanny --nthreads 3 127.0.0.1:8786\r\n```\r\n\r\nand run this script (basically a distillation of how we're using Dask in our pipeline):\r\n```python\r\nfrom distributed import Client, fire_and_forget, worker_client\r\n\r\ndef inc(x):\r\n    return x + 1\r\n\r\ndef dec(x):\r\n    return x - 1\r\n\r\ndef mapper(fn, arr):\r\n    with worker_client(separate_thread=True) as client:\r\n        futures = client.map(fn, arr, pure=False)\r\n    fire_and_forget(futures)\r\n    return futures\r\n\r\ndef waiter(futs):\r\n    with worker_client(separate_thread=True) as client:\r\n        return client.gather(futs)\r\n\r\ndef pipeline(fn, arr):\r\n    futs = mapper(fn, arr)\r\n    return waiter(futs)\r\n\r\nwith Client(address='localhost:8786') as client:\r\n    f = client.submit(pipeline, inc, list(range(100)))\r\n    fire_and_forget(f) # this appears to be irrelevant\r\n    g = client.submit(pipeline, dec, list(range(100)))\r\n    fire_and_forget(g) # this appears to be irrelevant\r\n    val = client.gather([f,g])\r\n```\r\nand you _should_ get a `CancelledError`.\r\n\r\nNotes:\r\n- if you write the logs from the scheduler and worker to a file, the error doesn't seem to occur\r\n- if you run this twice, the error won't occur; it seems to only occur whenever the scheduler, worker and python process are all \"fresh\"\r\n- this is on `distributed.__version__` = 1.26.1\r\n- it appears to be threading related (running with 1 thread doesn't reproduce)\r\n\r\nIt seems that removing the `fire_and_forget` call inside `mapper` fixes the situation.\r\n\r\n**cc:** @jlowin "},"comment":{"url":"https://api.github.com/repos/dask/distributed/issues/comments/485612244","html_url":"https://github.com/dask/distributed/issues/2628#issuecomment-485612244","issue_url":"https://api.github.com/repos/dask/distributed/issues/2628","id":485612244,"node_id":"MDEyOklzc3VlQ29tbWVudDQ4NTYxMjI0NA==","user":{"login":"cicdw","id":13255838,"node_id":"MDQ6VXNlcjEzMjU1ODM4","avatar_url":"https://avatars0.githubusercontent.com/u/13255838?v=4","gravatar_id":"","url":"https://api.github.com/users/cicdw","html_url":"https://github.com/cicdw","followers_url":"https://api.github.com/users/cicdw/followers","following_url":"https://api.github.com/users/cicdw/following{/other_user}","gists_url":"https://api.github.com/users/cicdw/gists{/gist_id}","starred_url":"https://api.github.com/users/cicdw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cicdw/subscriptions","organizations_url":"https://api.github.com/users/cicdw/orgs","repos_url":"https://api.github.com/users/cicdw/repos","events_url":"https://api.github.com/users/cicdw/events{/privacy}","received_events_url":"https://api.github.com/users/cicdw/received_events","type":"User","site_admin":false},"created_at":"2019-04-23T01:52:55Z","updated_at":"2019-04-23T01:52:55Z","author_association":"CONTRIBUTOR","body":"Ok so, it seems that `fire_and_forget` was sort of a red herring; ultimately the problem is what @jlowin mentioned above: we want to resolve futures created by one client with another client.  We used `fire_and_forget` to ensure the scheduler kept those futures alive, but it doesn't help us resolve them by other clients (hence the `CancelledError` which is really just a symptom of the client not having that future).\r\n\r\nIt would be amazing if we could \"tag\" a future somehow to let the scheduler know we fully intend to resolve it in the future with another client."}}
other: {"actor":{"display_login":"cicdw"}}