Handle aborted transport connections gracefully

Author: 1st1

uvloop/handles/basetransport.pyx

@@ -117,7 +117,12 @@ cdef class UVBaseTransport(UVSocketHandle):
     cdef _wakeup_waiter(self):
         if self._waiter is not None:
             if not self._waiter.cancelled():
-                self._waiter.set_result(True)
+                if not self._is_alive():
+                    self._waiter.set_exception(
+                        RuntimeError(
+                            'closed Transport handle and unset waiter'))
+                else:
+                    self._waiter.set_result(True)
             self._waiter = None
 
     cdef _call_connection_made(self):
@@ -126,7 +131,10 @@ cdef class UVBaseTransport(UVSocketHandle):
             raise RuntimeError(
                 'protocol is not set, cannot call connection_made()')
 
-        if self._closing:
+        # We use `_is_alive()` and not `_closing`, because we call
+        # `transport._close()` in `loop.create_connection()` if an
+        # exception happens during `await waiter`.
+        if not self._is_alive():
             # A connection waiter can be cancelled between
             # 'await loop.create_connection()' and
             # `_schedule_call_connection_made` and
@@ -147,7 +155,7 @@ cdef class UVBaseTransport(UVSocketHandle):
             self._wakeup_waiter()
             raise
 
-        if self._closing:
+        if not self._is_alive():
             # This might happen when "transport.abort()" is called
             # from "Protocol.connection_made".
             self._wakeup_waiter()

uvloop/handles/stream.pyx

@@ -254,6 +254,10 @@ cdef class UVStream(UVBaseTransport):
 
     cdef _start_reading(self):
         cdef int err
+
+        if self._closing:
+            return
+
         self._ensure_alive()
 
         if self.__reading:

uvloop/loop.pyx

@@ -1755,6 +1755,11 @@ cdef class Loop:
                 tr._init_protocol()
                 await waiter
             except:
+                # It's OK to call `_close()` here, as opposed to
+                # `_force_close()` or `close()` as we want to terminate the
+                # transport immediately.  The `waiter` can only be waken
+                # up in `Transport._call_connection_made()`, and calling
+                # `_close()` before it is fine.
                 tr._close()
                 raise