Merge pull request #794 from itamarst/769-thread-leak

Fix thread leak by switching to different `Queue` shutdown mechanism

Author: webknjaz

.pre-commit-config.yaml

@@ -148,7 +148,7 @@ repos:
   - id: name-tests-test
     args:
     - --django
-    exclude: cheroot/test/(helper|webtest|_pytest_plugin).py
+    exclude: cheroot/test/(helper|webtest|_pytest_plugin|threadleakcheck).py
     files: cheroot/test/.+\.py$
   - id: check-added-large-files
   - id: check-case-conflict

cheroot/server.py

@@ -100,37 +100,6 @@
     'get_ssl_adapter_class',
 )
 
-
-if sys.version_info[:2] >= (3, 13):
-    from queue import (
-        Queue as QueueWithShutdown,
-        ShutDown as QueueShutDown,
-    )
-else:
-
-    class QueueShutDown(Exception):
-        """Queue has been shut down."""
-
-    class QueueWithShutdown(queue.Queue):
-        """Add shutdown() similar to Python 3.13+ Queue."""
-
-        _queue_shut_down: bool = False
-
-        def shutdown(self, immediate=False):
-            if immediate:
-                while True:
-                    try:
-                        self.get_nowait()
-                    except queue.Empty:
-                        break
-            self._queue_shut_down = True
-
-        def get(self, *args, **kwargs):
-            if self._queue_shut_down:
-                raise QueueShutDown
-            return super().get(*args, **kwargs)
-
-
 IS_WINDOWS = platform.system() == 'Windows'
 """Flag indicating whether the app is running under Windows."""
 
@@ -1691,7 +1660,7 @@ def __init__(  # pylint: disable=too-many-positional-arguments
         self.reuse_port = reuse_port
         self.clear_stats()
 
-        self._unservicable_conns = QueueWithShutdown()
+        self._unservicable_conns = queue.Queue()
 
     def clear_stats(self):
         """Reset server stat counters.."""
@@ -1904,9 +1873,8 @@ def prepare(self):  # noqa: C901  # FIXME
     def _serve_unservicable(self):
         """Serve connections we can't handle a 503."""
         while self.ready:
-            try:
-                conn = self._unservicable_conns.get()
-            except QueueShutDown:
+            conn = self._unservicable_conns.get()
+            if conn is _STOPPING_FOR_INTERRUPT:
                 return
             request = HTTPRequest(self, conn)
             try:
@@ -2269,7 +2237,7 @@ def stop(self):  # noqa: C901  # FIXME
 
         # This tells the thread that handles unservicable connections to shut
         # down:
-        self._unservicable_conns.shutdown(immediate=True)
+        self._unservicable_conns.put(_STOPPING_FOR_INTERRUPT)
 
         if self._start_time is not None:
             self._run_time += time.time() - self._start_time

cheroot/test/test_server.py

@@ -1,8 +1,10 @@
 """Tests for the HTTP server."""
 
 import os
+import pathlib
 import queue
 import socket
+import subprocess
 import sys
 import tempfile
 import threading
@@ -23,6 +25,7 @@
     ANY_INTERFACE_IPV4,
     ANY_INTERFACE_IPV6,
     EPHEMERAL_PORT,
+    SUCCESSFUL_SUBPROCESS_EXIT,
 )
 from ..workers.threadpool import ThreadPool
 
@@ -611,3 +614,15 @@ def test_overload_results_in_suitable_http_error(request):
 
     response = requests.get(f'http://{localhost}:{port}', timeout=20)
     assert response.status_code == HTTPStatus.SERVICE_UNAVAILABLE
+
+
+def test_overload_thread_does_not_leak():
+    """On shutdown the overload thread exits.
+
+    This is a test for issue #769.
+    """
+    path = pathlib.Path(__file__).parent / 'threadleakcheck.py'
+    process = subprocess.run([sys.executable, path], check=False)
+    # We use special exit code to indicate success, rather than normal zero, so
+    # the test doesn't acidentally pass:
+    assert process.returncode == SUCCESSFUL_SUBPROCESS_EXIT

cheroot/test/threadleakcheck.py

@@ -0,0 +1,43 @@
+"""
+Make sure threads don't leak.
+
+Run in an isolated subprocess by ``test_server.py``, to ensure parallelism of
+any sort don't cause problems.
+"""
+
+import sys
+import threading
+import time
+
+from cheroot.server import Gateway, HTTPServer
+from cheroot.testing import (
+    ANY_INTERFACE_IPV4,
+    EPHEMERAL_PORT,
+    SUCCESSFUL_SUBPROCESS_EXIT,
+)
+
+
+SLEEP_INTERVAL = 0.2
+
+
+def check_for_leaks():
+    """Exit with special success code if no threads were leaked."""
+    before_serv = threading.active_count()
+    for _ in range(5):
+        httpserver = HTTPServer(
+            bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT),
+            gateway=Gateway,
+        )
+        with httpserver._run_in_thread():
+            time.sleep(SLEEP_INTERVAL)
+
+    leaked_threads = threading.active_count() - before_serv
+    if leaked_threads == 0:
+        sys.exit(SUCCESSFUL_SUBPROCESS_EXIT)
+    else:
+        # We leaked a thread:
+        sys.exit(f'Number of leaked threads: {leaked_threads}')
+
+
+if __name__ == '__main__':
+    check_for_leaks()

cheroot/testing.py

@@ -19,6 +19,10 @@
 ANY_INTERFACE_IPV4 = '0.0.0.0'
 ANY_INTERFACE_IPV6 = '::'
 
+# We use special exit code to indicate success, rather than normal zero, so
+# the test doesn't acidentally pass:
+SUCCESSFUL_SUBPROCESS_EXIT = 23
+
 config = {
     cheroot.wsgi.Server: {
         'bind_addr': (NO_INTERFACE, EPHEMERAL_PORT),

docs/changelog-fragments.d/769.bugfix.rst

@@ -0,0 +1 @@
+794.bugfix.rst

docs/changelog-fragments.d/794.bugfix.rst

@@ -0,0 +1,7 @@
+The "service unavailable" thread is now turn down properly when
+the server is shut down -- by :user:`itamarst`.
+
+This fixes a regression in Cheroot originally introduced in v11.0.0
+that would manifest itself under Python 3.12 and older. In certain
+conditions like under CherryPy, it would also lead to hangs on
+tear-down.