Repository: cherrypy/cheroot

Issue:
[cheroot==8.1.0 regression] Significant performance drop
❓ **I'm submitting a ...**
- [X] 🐞 bug report
- [ ] 🐣 feature request
- [ ] ❓ question about the decisions made in the repository

🐞 **Describe the bug. What is the current behavior?**

This more a note than a bug report:
I run some benchmark against my project ([WsgiDAV](https://github.com/mar10/wsgidav)] and found that there is a drastical drop in performance between cheroot versions 8.0.0 and 8.1.0.

The test uses a homegrown tool ([stressor](https://github.com/mar10/stressor)) that runs a test script ([see here](https://github.com/mar10/wsgidav/blob/master/tests/stressor/test_rw.yaml)):

- 10 users in parallel
  - GET a common file 
  - PUT another file (one per user)
  - GET that file
- Running for 30 seconds

Benchmarks show 11k activities per 30 seconds until v8.0.
Starting with cheroot 8.1 this drops to 2.9k activities per 30 sec.

I also tried with gevent as backend, and it delivered 12.4k activities.

Please note that these results may be due to poor implementation on my tools, but all I changed was WSGI server behind WsgiDAV, so I thought I let you know.
 

💡 **To Reproduce**

See https://github.com/mar10/wsgidav/blob/master/tests/stressor/results.md

If you are interested, I can provide more details.

Cheers
Martin

Discussion:
webknjaz: Hm. This introduces a number of windows failures in GHA that haven't been happening in the past: https://github.com/cherrypy/cheroot/pull/311/checks?check_run_id=979099442#step:14:351.

(only py2 win jobs were flaky)
webknjaz: @liamstask I believe this needs rebasing
liamstask: @webknjaz @cyraxjoe i also introduced a commit to address the issues observed in #317 - f3e09256257e335c2fd2a6005355ad2b52273322 (relies on an as yet unimplemented backport on some platforms)

i think it's preferable to only add contention in the case that `put()` is being called (infrequent) rather than all times the selector is accessed. let me know what you think.
liamstask: also split this out as a separate PR in https://github.com/cherrypy/cheroot/pull/319
liamstask: @webknjaz i probably shouldn't have added that commit to this PR, since it's addressing a separate issue. if you agree, i've created a separate PR with just the fix in #319 - i can always come and rebase this afterwards.
webknjaz: @liamstask alright, let's work on #319 first.
liamstask: marking as draft, will revisit after #337 
jaraco: Now that #317 is closed, I believe this one is tee'd up for a refresh. Please resolve the conflicts and take a look.
liamstask: ok, i believe this is ready for another look. this PR is not strictly needed if you're in the process of preparing a more major release, but this should address the issues from #305 in a more direct/robust way, per the notes in the PR description.
webknjaz: @liamstask did you verify how this works with CherryPy? Maybe run its test suite?
liamstask: updated, and tested successfully locally against cherrypy/cherrypy master by:
* editing the [cheroot-master testenv](https://github.com/cherrypy/cherrypy/blob/2a8aaccd649eb1011382c39f5cd93f76f980c0b1/tox.ini#L30) to point to this PR:  `deps = git+git://github.com/liamstask/cheroot.git@0c3fb0d597c492d891361cc6dcc003045ec37da6#egg=cheroot`
* running `python3 -m tox -e cheroot-master`

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 64bd2881..134e3ff2 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -1,3 +1,11 @@
+.. scm-version-title:: v8.5.0
+
+- :issue:`305` via :pr:`311`: In
+  :py:class:`~cheroot.connections.ConnectionManager`,
+  process connections as they become active rather than
+  waiting for a ``tick`` event, addressing performance
+  degradation introduced in v8.1.0.
+
 .. scm-version-title:: v8.4.8
 
 - :issue:`317` via :pr:`337`: Fixed a regression in
diff --git a/cheroot/connections.py b/cheroot/connections.py
index 8d3bc1f1..c38cfd1f 100644
--- a/cheroot/connections.py
+++ b/cheroot/connections.py
@@ -3,7 +3,6 @@
 from __future__ import absolute_import, division, print_function
 __metaclass__ = type
 
-import collections
 import io
 import os
 import socket
@@ -126,8 +125,10 @@ class ConnectionManager:
             server (cheroot.server.HTTPServer): web server object
                 that uses this ConnectionManager instance.
         """
+        self._serving = False
+        self._stop_requested = False
+
         self.server = server
-        self._readable_conns = collections.deque()
         self._selector = _ThreadsafeSelector()
 
         self._selector.register(
@@ -145,13 +146,13 @@ class ConnectionManager:
         # if this conn doesn't have any more data waiting to be read,
         # register it with the selector.
         if conn.rfile.has_data():
-            self._readable_conns.append(conn)
+            self.server.process_conn(conn)
         else:
             self._selector.register(
                 conn.socket.fileno(), selectors.EVENT_READ, data=conn,
             )
 
-    def expire(self):
+    def _expire(self):
         """Expire least recently used connections.
 
         This happens if there are either too many open connections, or if the
@@ -171,48 +172,57 @@ class ConnectionManager:
             self._selector.unregister(sock_fd)
             conn.close()
 
-    def get_conn(self):
-        """Return a HTTPConnection object which is ready to be handled.
+    def stop(self):
+        """Stop the selector loop in run() synchronously.
+
+        May take up to half a second.
+        """
+        self._stop_requested = True
+        while self._serving:
+            time.sleep(0.01)
+
+    def run(self, expiration_interval):
+        """Run the connections selector indefinitely.
 
-        A connection returned by this method should be ready for a worker
-        to handle it. If there are no connections ready, None will be
-        returned.
+        Args:
+            expiration_interval (float): Interval, in seconds, at which
+                connections will be checked for expiration.
+
+        Connections that are ready to process are submitted via
+        self.server.process_conn()
 
-        Any connection returned by this method will need to be `put`
-        back if it should be examined again for another request.
+        Connections submitted for processing must be `put()`
+        back if they should be examined again for another request.
 
-        :returns: HTTPConnection instance, or None
-        :rtype: cheroot.server.HTTPConnection
+        Can be shut down by calling `stop()`.
         """
-        # return a readable connection if any exist
-        with suppress(IndexError):
-            return self._readable_conns.popleft()
+        self._serving = True
+        last_expiration_check = time.time()
 
-        # Will require a select call.
-        try:
-            # The timeout value impacts performance and should be carefully
-            # chosen. Ref:
-            # github.com/cherrypy/cheroot/issues/305#issuecomment-663985165
-            active_list = self._selector.select(timeout=0.01)
-        except OSError:
-            self._remove_invalid_sockets()
-            # Wait for the next tick to occur.
-            return None
-
-        for (sock_fd, conn) in active_list:
-            if conn is self.server:
-                # New connection
-                return self._from_server_socket(self.server.socket)
+        while not self._stop_requested:
+            try:
+                active_list = self._selector.select(timeout=0.01)
+            except OSError:
+                self._remove_invalid_sockets()
+                continue
 
-            # unregister connection from the selector until the server
-            # has read from it and returned it via put()
-            self._selector.unregister(sock_fd)
-            self._readable_conns.append(conn)
+            for (sock_fd, conn) in active_list:
+                if conn is self.server:
+                    # New connection
+                    new_conn = self._from_server_socket(self.server.socket)
+                    self.server.process_conn(new_conn)
+                else:
+                    # unregister connection from the selector until the server
+                    # has read from it and returned it via put()
+                    self._selector.unregister(sock_fd)
+                    self.server.process_conn(conn)
 
-        try:
-            return self._readable_conns.popleft()
-        except IndexError:
-            return None
+            now = time.time()
+            if (now - last_expiration_check) > expiration_interval:
+                self._expire()
+                last_expiration_check = now
+
+        self._serving = False
 
     def _remove_invalid_sockets(self):
         """Clean up the resources of any broken connections.
@@ -332,10 +342,6 @@ class ConnectionManager:
 
     def close(self):
         """Close all monitored connections."""
-        for conn in self._readable_conns:
-            conn.close()
-        self._readable_conns.clear()
-
         for (_, conn) in self._selector.connections:
             if conn is not self.server:  # server closes its own socket
                 conn.close()
@@ -345,11 +351,11 @@ class ConnectionManager:
     def _num_connections(self):
         """Return the current number of connections.
 
-        Includes any in the readable list or registered with the selector,
+        Includes all conns registered with the selector,
         minus one for the server socket, which is always registered
         with the selector.
         """
-        return len(self._readable_conns) + len(self._selector) - 1
+        return len(self._selector) - 1
 
     @property
     def can_add_keepalive_connection(self):
diff --git a/cheroot/server.py b/cheroot/server.py
index a5c31b2e..fb3bc421 100644
--- a/cheroot/server.py
+++ b/cheroot/server.py
@@ -7,12 +7,13 @@ sticking incoming connections onto a Queue::
 
     server = HTTPServer(...)
     server.start()
-    ->  while True:
-            tick()
-            # This blocks until a request comes in:
-            child = socket.accept()
-            conn = HTTPConnection(child, ...)
-            server.requests.put(conn)
+    ->  serve()
+        while ready:
+            _connections.run()
+                while not stop_requested:
+                    child = socket.accept()  # blocks until a request comes in
+                    conn = HTTPConnection(child, ...)
+                    server.process_conn(conn)  # adds conn to threadpool
 
 Worker threads are kept in a pool and poll the Queue, popping off and then
 handling each connection in turn. Each connection can consist of an arbitrary
@@ -1523,6 +1524,11 @@ class HTTPServer:
     timeout = 10
     """The timeout in seconds for accepted connections (default 10)."""
 
+    expiration_interval = 0.5
+    """The interval, in seconds, at which the server checks for
+    expired connections (default 0.5).
+    """
+
     version = 'Cheroot/{version!s}'.format(version=__version__)
     """A version string for the HTTPServer."""
 
@@ -1592,7 +1598,6 @@ class HTTPServer:
         self.requests = threadpool.ThreadPool(
             self, min=minthreads or 1, max=maxthreads,
         )
-        self.serving = False
 
         if not server_name:
             server_name = self.version
@@ -1797,21 +1802,17 @@ class HTTPServer:
 
     def serve(self):
         """Serve requests, after invoking :func:`prepare()`."""
-        self.serving = True
         while self.ready:
             try:
-                self.tick()
+                self._connections.run(self.expiration_interval)
             except (KeyboardInterrupt, SystemExit):
-                self.serving = False
                 raise
             except Exception:
                 self.error_log(
-                    'Error in HTTPServer.tick', level=logging.ERROR,
+                    'Error in HTTPServer.serve', level=logging.ERROR,
                     traceback=True,
                 )
 
-        self.serving = False
-
     def start(self):
         """Run the server forever.
 
@@ -2039,17 +2040,13 @@ class HTTPServer:
 
         return bind_addr
 
-    def tick(self):
-        """Accept a new connection and put it on the Queue."""
-        conn = self._connections.get_conn()
-        if conn:
-            try:
-                self.requests.put(conn)
-            except queue.Full:
-                # Just drop the conn. TODO: write 503 back?
-                conn.close()
-
-        self._connections.expire()
+    def process_conn(self, conn):
+        """Process an incoming HTTPConnection."""
+        try:
+            self.requests.put(conn)
+        except queue.Full:
+            # Just drop the conn. TODO: write 503 back?
+            conn.close()
 
     @property
     def interrupt(self):
@@ -2067,14 +2064,15 @@ class HTTPServer:
 
     def stop(self):  # noqa: C901  # FIXME
         """Gracefully shutdown a server that is serving forever."""
+        if not self.ready:
+            return  # already stopped
+
         self.ready = False
         if self._start_time is not None:
             self._run_time += (time.time() - self._start_time)
         self._start_time = None
 
-        # ensure serve is no longer accessing socket, connections
-        while self.serving:
-            time.sleep(0.1)
+        self._connections.stop()
 
         sock = getattr(self, 'socket', None)
         if sock:
diff --git a/docs/conf.py b/docs/conf.py
index 175b9cfe..f1da660b 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -108,6 +108,7 @@ nitpick_ignore = [
     ('py:class', '_pyio.BufferedWriter'),
     ('py:class', '_pyio.BufferedReader'),
     ('py:class', 'unittest.case.TestCase'),
+    ('py:meth', 'cheroot.connections.ConnectionManager.get_conn'),
 ]
 
 # Ref:


Test patch:
diff --git a/cheroot/test/test_conn.py b/cheroot/test/test_conn.py
index b8660cfb..c33fb2b4 100644
--- a/cheroot/test/test_conn.py
+++ b/cheroot/test/test_conn.py
@@ -1145,6 +1145,6 @@ def test_invalid_selected_connection(test_client, monkeypatch):
     # trigger the internal errors
     faux_get_map.sabotage_conn = faux_select.request_served = True
     # give time to make sure the error gets handled
-    time.sleep(0.2)
+    time.sleep(test_client.server_instance.expiration_interval * 2)
     assert faux_select.os_error_triggered
     assert faux_get_map.conn_closed
diff --git a/cheroot/test/test_server.py b/cheroot/test/test_server.py
index f71ce61d..52b8b1b9 100644
--- a/cheroot/test/test_server.py
+++ b/cheroot/test/test_server.py
@@ -10,6 +10,7 @@ import os
 import socket
 import tempfile
 import threading
+import time
 import uuid
 
 import pytest
@@ -125,20 +126,20 @@ def test_serving_is_false_and_stop_returns_after_ctrlc():
 
     httpserver.prepare()
 
-    # Simulate a Ctrl-C on the first call to `get_conn`.
+    # Simulate a Ctrl-C on the first call to `run`.
     def raise_keyboard_interrupt(*args):
         raise KeyboardInterrupt()
 
-    httpserver._connections.get_conn = raise_keyboard_interrupt
+    httpserver._connections.run = raise_keyboard_interrupt
 
     serve_thread = threading.Thread(target=httpserver.serve)
     serve_thread.start()
 
     # The thread should exit right away due to the interrupt.
-    serve_thread.join(0.5)
+    serve_thread.join(httpserver.expiration_interval * 2)
     assert not serve_thread.is_alive()
 
-    assert not httpserver.serving
+    assert not httpserver._connections._serving
     httpserver.stop()
 
 
@@ -295,11 +296,12 @@ def test_high_number_of_file_descriptors(resource_limit):
     httpserver = HTTPServer(
         bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT), gateway=Gateway,
     )
-    httpserver.prepare()
 
     try:
         # This will trigger a crash if select() is used in the implementation
-        httpserver.tick()
+        with httpserver._run_in_thread():
+            # allow server to run long enough to invoke select()
+            time.sleep(1.0)
     except:  # noqa: E722
         raise  # only needed for `else` to work
     else: