From 58fab6c4158ec568b6ef94789cd8e622183f3cab Mon Sep 17 00:00:00 2001
From: Kevin Conway <kconway@atlassian.com>
Date: Fri, 1 Jul 2016 08:57:30 -0500
Subject: [PATCH] Accept multiple connections on listener EVENT_READ

When a large number of connections are made at once the event loop
struggles to accept them because it only accepts and begins
handling one new connection on each pass of the loop in which the
EVENT_READ is emitted for the listening socket. This causes
incoming connections to time out and sever.

This patch uses the backlog setting of the listening socket to
allow for accepting a bounded number of new incoming connections
rather than a single one.
---
 Lib/asyncio/base_events.py     |  2 +-
 Lib/asyncio/selector_events.py | 71 ++++++++++++++++++++++--------------------
 2 files changed, 39 insertions(+), 34 deletions(-)

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index f55a36e..6a1fdf6 100755
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -975,7 +975,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
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index 34cce6b..55bfd15 100755
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -151,43 +151,48 @@ 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):
-        try:
-            conn, addr = sock.accept()
-            if self._debug:
-                logger.debug("%r got a new connection from %r: %r",
-                             server, addr, conn)
-            conn.setblocking(False)
-        except (BlockingIOError, InterruptedError, ConnectionAbortedError):
-            pass  # False alarm.
-        except OSError as exc:
-            # There's nowhere to send the error, so just log it.
-            if exc.errno in (errno.EMFILE, errno.ENFILE,
-                             errno.ENOBUFS, errno.ENOMEM):
-                # Some platforms (e.g. Linux keep reporting the FD as
-                # ready, so we remove the read handler temporarily.
-                # We'll try again in a while.
-                self.call_exception_handler({
-                    'message': 'socket.accept() out of system resource',
-                    'exception': exc,
-                    'socket': sock,
-                })
-                self.remove_reader(sock.fileno())
-                self.call_later(constants.ACCEPT_RETRY_DELAY,
-                                self._start_serving,
-                                protocol_factory, sock, sslcontext, server)
+                           sslcontext=None, server=None, backlog=100):
+        counter = 0
+        for _ in range(backlog):
+            try:
+                conn, addr = sock.accept()
+                if self._debug:
+                    logger.debug("%r got a new connection from %r: %r",
+                                 server, addr, conn)
+                conn.setblocking(False)
+                counter += 1
+            except (BlockingIOError, InterruptedError, ConnectionAbortedError):
+                # Early exit because the socket accept buffer is empty.
+                return None
+            except OSError as exc:
+                # There's nowhere to send the error, so just log it.
+                if exc.errno in (errno.EMFILE, errno.ENFILE,
+                                 errno.ENOBUFS, errno.ENOMEM):
+                    # Some platforms (e.g. Linux keep reporting the FD as
+                    # ready, so we remove the read handler temporarily.
+                    # We'll try again in a while.
+                    self.call_exception_handler({
+                        'message': 'socket.accept() out of system resource',
+                        '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.
-        else:
-            extra = {'peername': addr}
-            accept = self._accept_connection2(protocol_factory, conn, extra,
-                                              sslcontext, server)
-            self.create_task(accept)
+                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,
--
2.5.0