gh-76002: asyncio: allow pause_reading() from connection_made()

Lib/asyncio/selector_events.py

@@ -683,14 +683,16 @@ def __init__(self, loop, sock, protocol, waiter=None,
         # decreases the latency (in some cases significantly.)
         _set_nodelay(self._sock)
 
-        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._sock_fd, self._read_ready)
-        if waiter is not None:
+        def _call_connection_made():
+            self._protocol.connection_made(self)
+            # only start reading when connection_made() has been called
+            if not self._paused:
+                self._loop._add_reader(self._sock_fd, self._read_ready)
             # only wake up the waiter when connection_made() has been called
-            self._loop.call_soon(futures._set_result_unless_cancelled,
-                                 waiter, None)
+            if waiter is not None:
+                futures._set_result_unless_cancelled(waiter, None)
+
+        self._loop.call_soon(_call_connection_made)
 
     def pause_reading(self):
         if self._closing:

Lib/test/test_asyncio/test_events.py

@@ -1846,6 +1846,35 @@ def test():
         with self.assertRaises(RuntimeError):
             self.loop.add_signal_handler(signal.SIGTERM, func)
 
+    def test_pause_reading_in_connection_made(self):
+        class PausingProto(MyBaseProto):
+
+            def connection_made(self, transport):
+                super().connection_made(transport)
+                transport.pause_reading()
+
+        proto = PausingProto(self.loop)
+        f = self.loop.create_server(lambda: proto, '127.0.0.1', 0)
+        server = self.loop.run_until_complete(f)
+        self.assertEqual(len(server.sockets), 1)
+        host, port = server.sockets[0].getsockname()
+        client = socket.socket()
+        client.connect(('127.0.0.1', port))
+        client.sendall(b'xxx')
+
+        self.loop.run_until_complete(proto.connected)
+        self.loop.run_until_complete(asyncio.sleep(0.1, loop=self.loop))
+        self.assertEqual(0, proto.nbytes)
+
+        proto.transport.resume_reading()
+
+        test_utils.run_until(self.loop, lambda: proto.nbytes > 0)
+        self.assertEqual(3, proto.nbytes)
+
+        proto.transport.close()
+        client.close()
+        server.close()
+
 
 class SubprocessTestsMixin:
 

Misc/NEWS.d/next/Library/2017-10-19-20-51-57.bpo-31821.DqFAx8.rst

@@ -0,0 +1,2 @@
+Allow ``asyncio.Transport.pause_reading()`` from
+``asyncio.Protocol.connection_made()``.