From 3531d9044dbfd15b3bf5ec1abe5b9744fce37464 Mon Sep 17 00:00:00 2001 From: Victor Stinner Date: Fri, 9 Jan 2015 01:42:52 +0100 Subject: [PATCH] asyncio: sync with Tulip * Document why set_result() calls are safe * Cleanup gather(). Use public methods instead of hacks to consume the exception of a future. * sock_connect(): pass directly the fd to _sock_connect_done instead of the socket. --- Lib/asyncio/queues.py | 6 ++++++ Lib/asyncio/selector_events.py | 6 +++--- Lib/asyncio/tasks.py | 11 +++++++---- 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 8f6c257701f..dce0d53c703 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -126,6 +126,8 @@ class Queue: # Use _put and _get instead of passing item straight to getter, in # case a subclass has logic that must run (e.g. JoinableQueue). self._put(item) + + # getter cannot be cancelled, we just removed done getters getter.set_result(self._get()) elif self._maxsize > 0 and self._maxsize <= self.qsize(): @@ -152,6 +154,8 @@ class Queue: # Use _put and _get instead of passing item straight to getter, in # case a subclass has logic that must run (e.g. JoinableQueue). self._put(item) + + # getter cannot be cancelled, we just removed done getters getter.set_result(self._get()) elif self._maxsize > 0 and self._maxsize <= self.qsize(): @@ -200,6 +204,8 @@ class Queue: item, putter = self._putters.popleft() self._put(item) # Wake putter on next tick. + + # getter cannot be cancelled, we just removed done putters putter.set_result(None) return self._get() diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py index 69b649ceeae..58b61f1c89e 100644 --- a/Lib/asyncio/selector_events.py +++ b/Lib/asyncio/selector_events.py @@ -363,15 +363,15 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop): break except BlockingIOError: fut.add_done_callback(functools.partial(self._sock_connect_done, - sock)) + fd)) self.add_writer(fd, self._sock_connect_cb, fut, sock, address) except Exception as exc: fut.set_exception(exc) else: fut.set_result(None) - def _sock_connect_done(self, sock, fut): - self.remove_writer(sock.fileno()) + def _sock_connect_done(self, fd, fut): + self.remove_writer(fd) def _sock_connect_cb(self, fut, sock, address): if fut.cancelled(): diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py index 8fc5beacf97..7959a55afea 100644 --- a/Lib/asyncio/tasks.py +++ b/Lib/asyncio/tasks.py @@ -582,11 +582,12 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False): def _done_callback(i, fut): nonlocal nfinished - if outer._state != futures._PENDING: - if fut._exception is not None: + if outer.done(): + if not fut.cancelled(): # Mark exception retrieved. fut.exception() return + if fut._state == futures._CANCELLED: res = futures.CancelledError() if not return_exceptions: @@ -644,9 +645,11 @@ def shield(arg, *, loop=None): def _done_callback(inner): if outer.cancelled(): - # Mark inner's result as retrieved. - inner.cancelled() or inner.exception() + if not inner.cancelled(): + # Mark inner's result as retrieved. + inner.exception() return + if inner.cancelled(): outer.cancel() else: