Merge 3.5 (issue #27906)

Lib/asyncio/base_events.py

@@ -1035,7 +1035,7 @@ class BaseEventLoop(events.AbstractEventLoop):
         for sock in sockets:
             sock.listen(backlog)
             sock.setblocking(False)
-            self._start_serving(protocol_factory, sock, ssl, server)
+            self._start_serving(protocol_factory, sock, ssl, server, backlog)
         if self._debug:
             logger.info("%r is serving", server)
         return server

Lib/asyncio/proactor_events.py

@@ -495,7 +495,7 @@ class BaseProactorEventLoop(base_events.BaseEventLoop):
         self._csock.send(b'\0')
 
     def _start_serving(self, protocol_factory, sock,
-                       sslcontext=None, server=None):
+                       sslcontext=None, server=None, backlog=100):
 
         def loop(f=None):
             try:

Lib/asyncio/selector_events.py

@@ -162,43 +162,50 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
                                  exc_info=True)
 
     def _start_serving(self, protocol_factory, sock,
-                       sslcontext=None, server=None):
+                       sslcontext=None, server=None, backlog=100):
         self.add_reader(sock.fileno(), self._accept_connection,
-                        protocol_factory, sock, sslcontext, server)
+                        protocol_factory, sock, sslcontext, server, backlog)
 
     def _accept_connection(self, protocol_factory, sock,
-                           sslcontext=None, server=None):
+                           sslcontext=None, server=None, backlog=100):
-        try:
+        # This method is only called once for each event loop tick where the
-            conn, addr = sock.accept()
+        # listening socket has triggered an EVENT_READ. There may be multiple
-            if self._debug:
+        # connections waiting for an .accept() so it is called in a loop.
-                logger.debug("%r got a new connection from %r: %r",
+        # See https://bugs.python.org/issue27906 for more details.
-                             server, addr, conn)
+        for _ in range(backlog):
-            conn.setblocking(False)
+            try:
-        except (BlockingIOError, InterruptedError, ConnectionAbortedError):
+                conn, addr = sock.accept()
-            pass  # False alarm.
+                if self._debug:
-        except OSError as exc:
+                    logger.debug("%r got a new connection from %r: %r",
-            # There's nowhere to send the error, so just log it.
+                                 server, addr, conn)
-            if exc.errno in (errno.EMFILE, errno.ENFILE,
+                conn.setblocking(False)
-                             errno.ENOBUFS, errno.ENOMEM):
+            except (BlockingIOError, InterruptedError, ConnectionAbortedError):
-                # Some platforms (e.g. Linux keep reporting the FD as
+                # Early exit because the socket accept buffer is empty.
-                # ready, so we remove the read handler temporarily.
+                return None
-                # We'll try again in a while.
+            except OSError as exc:
-                self.call_exception_handler({
+                # There's nowhere to send the error, so just log it.
-                    'message': 'socket.accept() out of system resource',
+                if exc.errno in (errno.EMFILE, errno.ENFILE,
-                    'exception': exc,
+                                 errno.ENOBUFS, errno.ENOMEM):
-                    'socket': sock,
+                    # Some platforms (e.g. Linux keep reporting the FD as
-                })
+                    # ready, so we remove the read handler temporarily.
-                self.remove_reader(sock.fileno())
+                    # We'll try again in a while.
-                self.call_later(constants.ACCEPT_RETRY_DELAY,
+                    self.call_exception_handler({
-                                self._start_serving,
+                        'message': 'socket.accept() out of system resource',
-                                protocol_factory, sock, sslcontext, server)
+                        'exception': exc,
+                        'socket': sock,
+                    })
+                    self.remove_reader(sock.fileno())
+                    self.call_later(constants.ACCEPT_RETRY_DELAY,
+                                    self._start_serving,
+                                    protocol_factory, sock, sslcontext, server,
+                                    backlog)
+                else:
+                    raise  # The event loop will catch, log and ignore it.
             else:
-                raise  # The event loop will catch, log and ignore it.
+                extra = {'peername': addr}
-        else:
+                accept = self._accept_connection2(protocol_factory, conn, extra,
-            extra = {'peername': addr}
+                                                  sslcontext, server)
-            accept = self._accept_connection2(protocol_factory, conn, extra,
+                self.create_task(accept)
-                                              sslcontext, server)
-            self.create_task(accept)
 
     @coroutine
     def _accept_connection2(self, protocol_factory, conn, extra,

Lib/test/test_asyncio/test_base_events.py

@@ -1634,7 +1634,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         self.loop.call_later.assert_called_with(constants.ACCEPT_RETRY_DELAY,
                                                 # self.loop._start_serving
                                                 mock.ANY,
-                                                MyProto, sock, None, None)
+                                                MyProto, sock, None, None, mock.ANY)
 
     def test_call_coroutine(self):
         @asyncio.coroutine

Lib/test/test_asyncio/test_selector_events.py

@@ -687,6 +687,20 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
               selectors.EVENT_WRITE)])
         self.loop.remove_writer.assert_called_with(1)
 
+    def test_accept_connection_multiple(self):
+        sock = mock.Mock()
+        sock.accept.return_value = (mock.Mock(), mock.Mock())
+        backlog = 100
+        # Mock the coroutine generation for a connection to prevent
+        # warnings related to un-awaited coroutines.
+        mock_obj = mock.patch.object
+        with mock_obj(self.loop, '_accept_connection2') as accept2_mock:
+            accept2_mock.return_value = None
+            with mock_obj(self.loop, 'create_task') as task_mock:
+                task_mock.return_value = None
+                self.loop._accept_connection(mock.Mock(), sock, backlog=backlog)
+        self.assertEqual(sock.accept.call_count, backlog)
+
 
 class SelectorTransportTests(test_utils.TestCase):
 

Misc/NEWS

@@ -418,6 +418,9 @@ Library
 
 - Issue #27456: asyncio: Set TCP_NODELAY by default.
 
+- Issue #27906: Fix socket accept exhaustion during high TCP traffic.
+  Patch by Kevin Conway.
+
 IDLE
 ----