asyncio: sync with Tulip

* Cleanup gather(): use cancelled() method instead of using private Future
  attribute
* Fix _UnixReadPipeTransport and _UnixWritePipeTransport. Only start reading
  when connection_made() has been called.
* Issue #23333: Fix BaseSelectorEventLoop._accept_connection(). Close the
  transport on error. In debug mode, log errors using call_exception_handler()

5 files changed, 85 insertions, 44 deletions

diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index 3195f622da..914783266d 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -22,6 +22,7 @@ from . import futures
 from . import selectors
 from . import transports
 from . import sslproto
+from .coroutines import coroutine
 from .log import logger
 
 
@@ -181,16 +182,47 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
             else:
                 raise  # The event loop will catch, log and ignore it.
         else:
+            extra = {'peername': addr}
+            accept = self._accept_connection2(protocol_factory, conn, extra,
+                                              sslcontext, server)
+            self.create_task(accept)
+
+    @coroutine
+    def _accept_connection2(self, protocol_factory, conn, extra,
+                            sslcontext=None, server=None):
+        protocol = None
+        transport = None
+        try:
             protocol = protocol_factory()
+            waiter = futures.Future(loop=self)
             if sslcontext:
-                self._make_ssl_transport(
-                    conn, protocol, sslcontext,
-                    server_side=True, extra={'peername': addr}, server=server)
+                transport = self._make_ssl_transport(
+                    conn, protocol, sslcontext, waiter=waiter,
+                    server_side=True, extra=extra, server=server)
             else:
-                self._make_socket_transport(
-                    conn, protocol , extra={'peername': addr},
+                transport = self._make_socket_transport(
+                    conn, protocol, waiter=waiter, extra=extra,
                     server=server)
-        # It's now up to the protocol to handle the connection.
+
+            try:
+                yield from waiter
+            except:
+                transport.close()
+                raise
+
+            # It's now up to the protocol to handle the connection.
+        except Exception as exc:
+            if self.get_debug():
+                context = {
+                    'message': ('Error on transport creation '
+                                'for incoming connection'),
+                    'exception': exc,
+                }
+                if protocol is not None:
+                    context['protocol'] = protocol
+                if transport is not None:
+                    context['transport'] = transport
+                self.call_exception_handler(context)
 
     def add_reader(self, fd, callback, *args):
         """Add a reader callback."""
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 63412a97e6..4f19a252ff 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -592,7 +592,7 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
                 fut.exception()
             return
 
-        if fut._state == futures._CANCELLED:
+        if fut.cancelled():
             res = futures.CancelledError()
             if not return_exceptions:
                 outer.set_exception(res)
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index 67973f14f3..7e1265a091 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -298,8 +298,10 @@ class _UnixReadPipeTransport(transports.ReadTransport):
         _set_nonblocking(self._fileno)
         self._protocol = protocol
         self._closing = False
-        self._loop.add_reader(self._fileno, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
+        # only start reading when connection_made() has been called
+        self._loop.call_soon(self._loop.add_reader,
+                             self._fileno, self._read_ready)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
@@ -401,13 +403,16 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
         self._conn_lost = 0
         self._closing = False  # Set when close() or write_eof() called.
 
-        # On AIX, the reader trick only works for sockets.
-        # On other platforms it works for pipes and sockets.
-        # (Exception: OS X 10.4?  Issue #19294.)
+        self._loop.call_soon(self._protocol.connection_made, self)
+
+        # On AIX, the reader trick (to be notified when the read end of the
+        # socket is closed) only works for sockets. On other platforms it
+        # works for pipes and sockets. (Exception: OS X 10.4?  Issue #19294.)
         if is_socket or not sys.platform.startswith("aix"):
-            self._loop.add_reader(self._fileno, self._read_ready)
+            # only start reading when connection_made() has been called
+            self._loop.call_soon(self._loop.add_reader,
+                                 self._fileno, self._read_ready)
 
-        self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index a38c90ebe8..12af62b209 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -886,13 +886,18 @@ class EventLoopTestsMixin:
         if hasattr(sslcontext_client, 'check_hostname'):
             sslcontext_client.check_hostname = True
 
+
         # no CA loaded
         f_c = self.loop.create_connection(MyProto, host, port,
                                           ssl=sslcontext_client)
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(ssl.SSLError,
-                                        'certificate verify failed '):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(ssl.SSLError,
+                                            'certificate verify failed '):
+                    self.loop.run_until_complete(f_c)
+
+            # execute the loop to log the connection error
+            test_utils.run_briefly(self.loop)
 
         # close connection
         self.assertIsNone(proto.transport)
@@ -919,15 +924,20 @@ class EventLoopTestsMixin:
         f_c = self.loop.create_unix_connection(MyProto, path,
                                                ssl=sslcontext_client,
                                                server_hostname='invalid')
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(ssl.SSLError,
-                                        'certificate verify failed '):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(ssl.SSLError,
+                                            'certificate verify failed '):
+                    self.loop.run_until_complete(f_c)
+
+            # execute the loop to log the connection error
+            test_utils.run_briefly(self.loop)
 
         # close connection
         self.assertIsNone(proto.transport)
         server.close()
 
+
     def test_legacy_create_unix_server_ssl_verify_failed(self):
         with test_utils.force_legacy_ssl_support():
             self.test_create_unix_server_ssl_verify_failed()
@@ -949,11 +959,12 @@ class EventLoopTestsMixin:
         # incorrect server_hostname
         f_c = self.loop.create_connection(MyProto, host, port,
                                           ssl=sslcontext_client)
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(
-                    ssl.CertificateError,
-                    "hostname '127.0.0.1' doesn't match 'localhost'"):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(
+                        ssl.CertificateError,
+                        "hostname '127.0.0.1' doesn't match 'localhost'"):
+                    self.loop.run_until_complete(f_c)
 
         # close connection
         proto.transport.close()
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 126196daac..41249ff024 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -350,16 +350,13 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
         return transport
 
     def test_ctor(self):
-        tr = self.read_pipe_transport()
-        self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.protocol.connection_made.assert_called_with(tr)
+        waiter = asyncio.Future(loop=self.loop)
+        tr = self.read_pipe_transport(waiter=waiter)
+        self.loop.run_until_complete(waiter)
 
-    def test_ctor_with_waiter(self):
-        fut = asyncio.Future(loop=self.loop)
-        tr = self.read_pipe_transport(waiter=fut)
-        test_utils.run_briefly(self.loop)
-        self.assertIsNone(fut.result())
+        self.protocol.connection_made.assert_called_with(tr)
+        self.loop.assert_reader(5, tr._read_ready)
+        self.assertIsNone(waiter.result())
 
     @mock.patch('os.read')
     def test__read_ready(self, m_read):
@@ -502,17 +499,13 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
         return transport
 
     def test_ctor(self):
-        tr = self.write_pipe_transport()
-        self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.protocol.connection_made.assert_called_with(tr)
+        waiter = asyncio.Future(loop=self.loop)
+        tr = self.write_pipe_transport(waiter=waiter)
+        self.loop.run_until_complete(waiter)
 
-    def test_ctor_with_waiter(self):
-        fut = asyncio.Future(loop=self.loop)
-        tr = self.write_pipe_transport(waiter=fut)
+        self.protocol.connection_made.assert_called_with(tr)
         self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.assertEqual(None, fut.result())
+        self.assertEqual(None, waiter.result())
 
     def test_can_write_eof(self):
         tr = self.write_pipe_transport()