When the connection queue is full, respond with a 503 error

The 503 responses are run in a thread

Author: pythonspeed

cheroot/server.py

@@ -1520,6 +1520,11 @@ def _close_kernel_socket(self):
                 raise
 
 
+# Unique identifier used to indicate the thread that handles unservicable
+# connections should shut down.
+_SHUT_DOWN_UNSERVICABLES_THREAD = object()
+
+
 class HTTPServer:
     """An HTTP server."""
 
@@ -1658,6 +1663,8 @@ def __init__(
         self.reuse_port = reuse_port
         self.clear_stats()
 
+        self._unservicable_conns = queue.Queue()
+
     def clear_stats(self):
         """Reset server stat counters.."""
         self._start_time = None
@@ -1866,8 +1873,34 @@ def prepare(self):  # noqa: C901  # FIXME
         self.ready = True
         self._start_time = time.time()
 
+    def _serve_unservicable(self):
+        """Serve connections we can't handle a 503."""
+        while self.ready:
+            conn = self._unservicable_conns.get()
+            if conn is _SHUT_DOWN_UNSERVICABLES_THREAD:
+                return
+            request = HTTPRequest(self, conn)
+            try:
+                request.simple_response('503 Service Unavailable')
+            except (socket.error, errors.FatalSSLAlert):
+                # We're sending the 503 error to be polite, it it fails that's
+                # fine.
+                continue
+            except Exception as ex:
+                self.server.error_log(
+                    repr(ex),
+                    level=logging.ERROR,
+                    traceback=True,
+                )
+            conn.close()
+
     def serve(self):
         """Serve requests, after invoking :func:`prepare()`."""
+        # This thread will handle unservicable connections, as added to
+        # self._unservicable_conns queue. It will run forever, until
+        # self.stop() tells it to shut down.
+        threading.Thread(target=self._serve_unservicable).start()
+
         while self.ready and not self.interrupt:
             try:
                 self._connections.run(self.expiration_interval)
@@ -2162,8 +2195,7 @@ def process_conn(self, conn):
         try:
             self.requests.put(conn)
         except queue.Full:
-            # Just drop the conn. TODO: write 503 back?
-            conn.close()
+            self._unservicable_conns.put(conn)
 
     @property
     def interrupt(self):
@@ -2201,6 +2233,11 @@ def stop(self):  # noqa: C901  # FIXME
             return  # already stopped
 
         self.ready = False
+
+        # This tells the thread that handles unservicable connections to shut
+        # down:
+        self._unservicable_conns.put(_SHUT_DOWN_UNSERVICABLES_THREAD)
+
         if self._start_time is not None:
             self._run_time += time.time() - self._start_time
         self._start_time = None

cheroot/test/test_server.py

@@ -8,6 +8,7 @@
 import types
 import urllib.parse  # noqa: WPS301
 import uuid
+from http import HTTPStatus
 
 import pytest
 
@@ -570,3 +571,36 @@ def test_threadpool_multistart_validation(monkeypatch):
         match='Threadpools can only be started once.',
     ):
         tp.start()
+
+
+def test_overload_results_in_suitable_http_error(request):
+    """A server that can't keep up with requests returns a 503 HTTP error."""
+    localhost = '127.0.0.1'
+    httpserver = HTTPServer(
+        bind_addr=(localhost, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+    # Can only handle on request in parallel:
+    httpserver.requests = ThreadPool(
+        min=1,
+        max=1,
+        accepted_queue_size=1,
+        accepted_queue_timeout=0,
+        server=httpserver,
+    )
+
+    httpserver.prepare()
+    serve_thread = threading.Thread(target=httpserver.serve)
+    serve_thread.start()
+    request.addfinalizer(httpserver.stop)
+    # Stop the thread pool to ensure the queue fills up:
+    httpserver.requests.stop()
+
+    _host, port = httpserver.bind_addr
+
+    # Use up the very limited thread pool queue we've set up, so future
+    # requests fail:
+    httpserver.requests._queue.put(None)
+
+    response = requests.get(f'http://{localhost}:{port}', timeout=20)
+    assert response.status_code == HTTPStatus.SERVICE_UNAVAILABLE