Client.get_dataset to always create Futures attached to itself #3729

crusaderky · 2020-04-20T15:16:12Z

Fix error "Inputs contain futures that were created by another client"
Closes Published Future retrieved from async client is synchronous #3227
Closes Unable to use published datasets in a different client #2336

This reverts commit 5d2566c.

jcrist

Thanks @crusaderky, I've left a few comments.

jcrist · 2020-04-20T22:01:41Z

distributed/client.py

@@ -90,6 +91,7 @@
 _global_clients = weakref.WeakValueDictionary()
 _global_client_index = [0]

+_deserialize_client = ContextVar("_deserialize_client", default=None)


Do we need to use contextvars here? We currently use a thread local for storing these kind of things, but that doesn't always play nicely with async. Is there a way we could refactor part of distributed to avoid passing around this implicit state?

In general I thinkcontextvars can be quite useful where required, but want to ensure it's needed here before bringing it in. We also already have a fair bit of implicit state, I'm a bit hesitant to add more if we can avoid it.

To avoid using contextvars you need to explicitly propagate the client until you stop calling async functions and just call sync functions. In this case, this means that

client.Client._get_dataset needs to pass self to

core.send_recv, which in turn needs to pass it to

comm.tcp.TCP.read (plus its variants comm.inproc.InProc.read and the 100% untested comm.ucx.UCX.read), which in turn needs to pass it to

comm.utils.from_frames, which can finally put a context manager (setting a thread-local variable) around

distributed/distributed/comm/utils.py

Lines 60 to 62 in 8376f22

return protocol.loads(

frames, deserialize=deserialize, deserializers=deserializers

)

which will alter the behaviour of client.Future.__getstate__

and it should be really self-evident why I don't recommend doing it.

We currently use a thread local for storing these kind of things

For this one you were using a global, client._global_clients.
contextvars exactly fits the use case of thread locals when you use coroutines.

Makes sense to me. Should this logic replace how we access implicit clients everywhere? If so, perhaps we should rename the context var key?

Are there other cases where a method of Client loses self along the way and re-acquires it later from a global variable?

Also note the very big fat caveat of the race conditions on Python 3.6. Not sure how much of the dask.distributed user base still uses Python 3.6 AND has more than one client running.

Sorry, let me clarify.

It feels a bit weird to handle setting the contextvar in the _get_dataset method. It might be cleaner to have a parent context over all coroutines started by a client that includes a reference back to the client. As it stands this is baking in functionality to Future that is only used by one method, when we likely want this everywhere.

It might be cleaner to have a parent context over all coroutines started by a client that includes a reference back to the client.

Are you talking to some sort of blanket decorator applied around all methods of Client? If so, it feels a bit overkill to me, as I do not know of another method that needs it.

As it stands this is baking in functionality to Future that is only used by one method, when we likely want this everywhere.

You definitely want to reuse this same variable every time you have a self reference to a Client instance, invoke a method, lose the reference (because it would be too cumbersome to propagated by hand), and then you need to reacquire it deeper down the line before the end of the method. If you can point me to any other function that does this, I'll be happy to look into it.

@jcrist ping

I'm redesigning the change to make it more generic, hold on

jcrist · 2020-04-20T22:12:08Z

distributed/tests/test_publish.py

+
+
+@gen_cluster(client=True)
+async def test_deserialize_client(c, s, a, b):


Nice test case.

TomAugspurger · 2020-04-22T14:29:32Z

@jcrist are you able to shepherd this PR through?

crusaderky · 2020-04-27T12:25:31Z

@jcrist I redesigned the PR to address your concerns - see if you like it now

crusaderky · 2020-04-29T13:28:01Z

@jcrist ping

crusaderky · 2020-05-04T08:23:02Z

@jcrist any feedback please?

jcrist

Overall I like the direction this PR is going. Just a few cleanups and I think I'm happy to merge.

jcrist · 2020-05-04T22:15:38Z

distributed/client.py

+        # We can still detect a race condition.
+        if sys.version_info < (3, 7) and _current_client.get() not in (self, None):
+            raise RuntimeError(
+                "Detected race condition where get_dataset() is invoked in "


This error message still mentions get_dataset, should probably be made a bit more generic.

distributed/client.py

jcrist

Overall this looks good to me. Thanks for pushing this work through, this definitely has cleaned up some of how clients are implicitly passed around.

Just to make sure I didn't miss anything glaring, I think it would be good to get one other maintainer's eyes on this before we merge. cc @jrbourbeau perhaps?

jcrist · 2020-05-05T13:26:11Z

distributed/tests/test_client.py


-    await c.close()
-    await f.close()
+def test_as_current_is_thread_local(s):


Clever test

crusaderky · 2020-05-05T13:37:17Z

Getting a completely unrelated failure on Windows since the latest merge from master

jcrist · 2020-05-05T13:39:06Z

Yeah, the test failures are unrelated, see #3774

jcrist · 2020-05-07T19:13:06Z

Test failure is unrelated. Thanks @crusaderky, merging.

pborgen · 2020-05-08T17:02:10Z

Thanks for this fix. .Cannot wait to try it out...

crusaderky added 17 commits April 11, 2020 18:25

Python 3.6+ syntax

207c27b

Code polish

02b1079

Merge remote-tracking branch 'upstream/master' into get_dataset_async

026a602

Revert

e07b9f8

Polish

5d2566c

Revert "Polish"

a5dc1be

This reverts commit 5d2566c.

Merge remote-tracking branch 'upstream/master' into get_dataset_async

f2b6e1f

tests

f86126a

Merge branch 'master' into get_dataset_async

7ef5832

revert

77a0d8a

Merge remote-tracking branch 'upstream/master' into get_dataset_async

de6b457

revert

9d6c8c0

Merge remote-tracking branch 'upstream/master' into get_dataset_async

6515c1b

Merge branch 'master' into get_dataset_async

d8fcdc9

xfail

21522df

Better async functions

4b55b31

Use contextvars to deserialize Future

fb8f777

jcrist reviewed Apr 20, 2020

View reviewed changes

crusaderky mentioned this pull request Apr 21, 2020

test_dont_steal_unknown_functions failure #3574

Closed

Merge branch 'master' into get_dataset_async

804b537

crusaderky added 3 commits April 27, 2020 10:44

Merge remote-tracking branch 'upstream/master' into get_dataset_async

2d594f8

Redesign

042eddf

Tweaks

621573d

jcrist requested changes May 4, 2020

View reviewed changes

crusaderky added 2 commits May 5, 2020 11:39

Merge remote-tracking branch 'upstream/master' into get_dataset_async

77fa36b

docstrings

e70303a

crusaderky closed this May 5, 2020

crusaderky reopened this May 5, 2020

jcrist reviewed May 5, 2020

View reviewed changes

distributed/tests/test_client.py

await c.close()

await f.close()

def test_as_current_is_thread_local(s):

Copy link

Member

jcrist May 5, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Clever test

Merge branch 'master' into get_dataset_async

f0f8c6e

jcrist merged commit 77f6c55 into dask:master May 7, 2020

crusaderky deleted the get_dataset_async branch May 7, 2020 23:07

crusaderky mentioned this pull request May 11, 2020

Futures unpickled on Client/Worker are attached to the wrong Client #3790

Open

gjoseph92 mentioned this pull request Nov 16, 2021

Consistent worker Client instance in get_client #5467

Open

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Client.get_dataset to always create Futures attached to itself #3729

Client.get_dataset to always create Futures attached to itself #3729

crusaderky commented Apr 20, 2020

jcrist left a comment

jcrist Apr 20, 2020

crusaderky Apr 21, 2020

jcrist Apr 21, 2020

crusaderky Apr 21, 2020

jcrist Apr 22, 2020

crusaderky Apr 23, 2020

crusaderky Apr 27, 2020

crusaderky Apr 27, 2020

jcrist Apr 20, 2020

TomAugspurger commented Apr 22, 2020

crusaderky commented Apr 27, 2020

crusaderky commented Apr 29, 2020

crusaderky commented May 4, 2020

jcrist left a comment

jcrist May 4, 2020

crusaderky May 5, 2020

jcrist left a comment

jcrist May 5, 2020

crusaderky commented May 5, 2020

jcrist commented May 5, 2020 •

edited

Loading

jcrist commented May 7, 2020

pborgen commented May 8, 2020

	return protocol.loads(
	frames, deserialize=deserialize, deserializers=deserializers
	)



		@gen_cluster(client=True)
		async def test_deserialize_client(c, s, a, b):

Client.get_dataset to always create Futures attached to itself #3729

Client.get_dataset to always create Futures attached to itself #3729

Conversation

crusaderky commented Apr 20, 2020

jcrist left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TomAugspurger commented Apr 22, 2020

crusaderky commented Apr 27, 2020

crusaderky commented Apr 29, 2020

crusaderky commented May 4, 2020

jcrist left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jcrist left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crusaderky commented May 5, 2020

jcrist commented May 5, 2020 • edited Loading

jcrist commented May 7, 2020

pborgen commented May 8, 2020

jcrist commented May 5, 2020 •

edited

Loading