Replace tornado.locks with asyncio for Events/Locks/Conditions/Semaphore - #3397
Conversation
Apparently asyncio.Conditions don't support async context managers yet
This shouldn't change anything, but it should be informative
|
OK, this is running now. cc @jrbourbeau for review (also maybe @jcrist) A couple of points:
|
jrbourbeau
left a comment
jrbourbeau
left a comment
There was a problem hiding this comment.
Thanks for taking the time to go through and make these changes @mrocklin, I imagine it was tedious to do. Generally things look really good, I left a few small comments. Looks like there are a few tornado Events remaining at
Can these also be updated?
| self._loop_runner = LoopRunner(loop=loop, asynchronous=asynchronous) | ||
| self.loop = self._loop_runner.loop | ||
|
|
||
| self._gather_semaphore = asyncio.Semaphore(5, loop=self.loop.asyncio_loop) |
There was a problem hiding this comment.
Is specifying s loop= parameter required for using asyncio.Semaphore here? I ask because we weren't passing a loop in before and doing is has been depreciated starting in Python 3.8
There was a problem hiding this comment.
This object is created by the user outside of the event loop. The semaphore will grab the event loop of the current thread, which when we're operating in normal synchronous mode, will not be the same event loop as the one that the Client is running on. So here we have to be explicit and specify the event loop that we want to use.
This usually doesn't come up because we create these objects within async functions or functions that are only being called from within the event loop. The client code can get strange in this way due to the two threads that are active (user/jupyter/ipython thread and the event loop thread)
There was a problem hiding this comment.
I see, thanks for the explaination @mrocklin.
Since Semaphore, Condition, etc. will attach to the current event loop, could temporarily switch the current loop to be the Client's self.loop.asyncio_loop, create the Semaphore, and then switch the current loop back to what it was before? (Not 100% sure this wouldn't have other consequences, just proposing as a possible option).
Alternatively, I'm happy to keep this as is and open a separate issue for working around the loop= keyword depreciation
There was a problem hiding this comment.
Ah, I missed the statement in your original message that this was being deprecated. The clean way to do this is to to move this to some __await__ definition within this class and make sure that we only create the Semaphore in an async function. I can do this, but I wouldn't mind it being in another PR if that's easy.
There was a problem hiding this comment.
Gotcha, that does sound cleaner. A separate PR for that would be appreciated
| self.lock = threading.Lock() | ||
| self.loop = loop or default_client().loop | ||
| self.condition = Condition() | ||
| self.condition = asyncio.Condition(loop=self.loop.asyncio_loop) |
There was a problem hiding this comment.
Same question about the loop= parameter
| self.name = name | ||
| self.buffer = deque() | ||
| self.condition = tornado.locks.Condition() | ||
| self.condition = asyncio.Condition(loop=self.loop.asyncio_loop) |
There was a problem hiding this comment.
Same question about the loop= parameter
jrbourbeau
left a comment
jrbourbeau
left a comment
There was a problem hiding this comment.
The only remaining tornado lock objects being used are a couple of tornado.locks.Events in the BatchedSend class in distributed/batched.py. Everything else looks good to merge
|
Batched.py is a bit strange. We're using gen.coroutine there in order to avoid leaving lingering asyncio coroutines around that cause warnings on shutdown. I'm sort of ignoring that file for now. |
jrbourbeau
left a comment
jrbourbeau
left a comment
There was a problem hiding this comment.
Okay, good to know. Then in that case, this is good to go. Merging, thanks @mrocklin!
2 participants
Fixes #3395
This is a work in progress. Things are broken. I thought I'd put it up here for now though.