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

@@ -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()

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():

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: