Repository: cherrypy/cheroot
Issue category: Performance Issue

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


Hints:
Hey @mar10, thanks for the heads up! v8.1.0 introduced a number of functional regressions that are got some fixes in v8.3.0 and v8.4.0 so please also test with these. It was a result of refactoring the connection management to decouple keep-alive TCP connections from HTTP requests. So I'd say that you should also check whether your tool reuses the connections or creates new ones of each HTTP request.
Hi Sviatoslav,

I already [tested with 8.3.1 and 8.4.0](https://github.com/mar10/wsgidav/blob/master/tests/stressor/results.md) with same results.

[stressor](https://github.com/mar10/stressor/) uses the requests library with sessions, which *should* [support keep-alive](https://requests.readthedocs.io/en/master/user/advanced/#keep-alive), but I can double check.

However, even if it did *not* reuse connections, would the refactoring explain that cheroot now slows down to 25% (for my specific benchmark)?  
I did not change anything in the scenario (test-tool, WSGI app), except for swapping out the WSGI servers.
Hm... I think you use the same session for all the requests meaning that they all are processed sequentially because they come one after another over the wire in the same TCP connection. HTTP/1.1 doesn't have built-in parallelism, unlike HTTP/2 that uses streams within the same connection.

Try checking that it does use just one connection (using tcpdump/wireshark, I guess). Also try make stressor creat a number of sessions and load-balance the queries. Plus try stress-testing with some other tool like ab/wrk2/locust.
Stressor spawns one thread for every virtual user. 
Each 'user' is holding its own `requests/urllib3` session, so every user should re-use its connection.
,
I modified the test. This time a sequence is only one GET request to a single static file:

### Cheroot 8.0

Executed 10,163 activities in 10,163 sequences, using 1 parallel sessions.  => 338.8 requests per sec/user
Executed 11,584 activities in 11,584 sequences, using 2 parallel sessions.  => 193.1 requests per sec/user
Executed 14,545 activities in 14,545 sequences, using 10 parallel sessions. =>  48.5 requests per sec/user
=> 0.003 ..  0.02 seconds per request 

### Cheroot 8.4

Executed 292 activities in 292 sequences, using 1 parallel sessions.      => 9.7 requests per sec/user
Executed 582 activities in 582 sequences, using 2 parallel sessions.      => 9.7 requests per sec/user
Executed 2,943 activities in 2,933 sequences, using 10 parallel sessions. => 9.8 requests per sec/user
=> 0.1 seconds per request

It may sound silly, but it looks like someone introduced a `sleep(0.1)` in v8.1+ ;-)
Weird. This clearly needs profiling. What are the thread num settings in your setup?
@mar10 mind record a flame graph using https://github.com/benfred/py-spy?
numthreads is set to 50.

py-spy looks interesting, I will try it out if I find time.

However, I searched for "0.1" and found that `connections.py` contains this code:
```py
        try:
            for fno in socket_dict:
                self._selector.register(fno, selectors.EVENT_READ)
            rlist = [
                key.fd for key, _event
                in self._selector.select(timeout=0.1)
            ]
        except OSError:
          ...
```

I patched the timeout value and ran the benchmarks again

timeout=0.1 (original)
Executed 2,907 activities in 2,907 sequences, using 10 parallel sessions.

timeout=0.05
Executed 5,766 activities in 5,766 sequences, using 10 parallel sessions.

timeout=0.02
Executed 12,199 activities in 12,199 sequences, using 10 parallel sessions.

timeout=0.01
Executed 13,933 activities in 13,933 sequences, using 10 parallel sessions.

Smaller values don't seem to improve the performance.
Extracting what @liamstask posted in the PR:
>it looks to me that this could be related to the fact that, currently in `server.tick()`, only one request can be added per tick even if several are available to be processed in `ConnectionManager.connections`.
>
>in this case, since `ConnectionManager` does not register connections previously marked as read-ready with the selector, it's possible that _none_ of the remaining connections in the selector are active, meaning the server is forced to wait for the full timeout duration each time `tick()` is called.
>
>fundamentally, the value of this timeout should not really be particularly sensitive, since the selector should wake up once connections are ready to be processed, regardless of the timeout.

_(Originally posted by @liamstask in https://github.com/cherrypy/cheroot/pull/308#issuecomment-664040481)_
8.4.2 fixed it for me, thanks

8.4.1:
Run time 0:00:30.109496, net: 4:59.91 min.
Executed 2,920 activities in 2,920 sequences, using 10 parallel sessions.

8.4.2:
Run time 0:00:30.099942, net: 4:57.40 min.
Executed 14,131 activities in 14,131 sequences, using 10 parallel sessions.



Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 64bd28811f..134e3ff27f 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 8d3bc1f1ab..c38cfd1f19 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 @@ def __init__(self, server):
             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 @@ def put(self, conn):
         # 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 @@ def expire(self):
             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 @@ def _from_server_socket(self, server_socket):  # noqa: C901  # FIXME
 
     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 @@ def close(self):
     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 a5c31b2e65..fb3bc421a2 100644
--- a/cheroot/server.py
+++ b/cheroot/server.py
@@ -7,12 +7,13 @@
 
     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 @@ def __init__(
         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 @@ def prepare(self):  # noqa: C901  # FIXME
 
     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 @@ def resolve_real_bind_addr(socket_):
 
         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 @@ def interrupt(self, interrupt):
 
     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 175b9cfe66..f1da660b65 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -108,6 +108,7 @@
     ('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 b8660cfb15..c33fb2b44b 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 f71ce61dd5..52b8b1b978 100644
--- a/cheroot/test/test_server.py
+++ b/cheroot/test/test_server.py
@@ -10,6 +10,7 @@
 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: