Fix a thread leak.

Implementing correct shutdown() would involve copying too much code, so switch
to the more traditional way of indicating shutdown to a thread reading from a
Queue.

Signed-off-by: Itamar Turner-Trauring <[email protected]>

Author: pythonspeed

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 None:
                 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(None)
 
         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
@@ -611,3 +613,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 exit code 23 to indicate success, so the test doesn't acidentally
+    # pass:
+    assert process.returncode == 23

cheroot/test/threadleakcheck.py

@@ -0,0 +1,41 @@
+"""
+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 os
+import sys
+import threading
+import time
+
+from cheroot.server import Gateway, HTTPServer
+from cheroot.testing import (
+    ANY_INTERFACE_IPV4,
+    EPHEMERAL_PORT,
+)
+
+
+def check_for_leaks():
+    """Exit with code 23 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(0.2)
+
+    leaked_threads = threading.active_count() - before_serv
+    if leaked_threads == 0:
+        os._exit(23)
+    else:
+        # We leaked a thread:
+        print('Number of leaked threads:', leaked_threads, file=sys.stderr)
+        os._exit(7)
+
+
+if __name__ == '__main__':
+    check_for_leaks()