diff --git a/server.py b/server.py index a6e330e2..9518f86d 100644 --- a/server.py +++ b/server.py @@ -119,6 +119,7 @@ class QuietHTTPServer(ThreadingHTTPServer): daemon_threads = True request_queue_size = 64 max_request_workers = 128 + max_overflow_reject_workers = 16 _OVERFLOW_RESPONSE = ( b"HTTP/1.1 503 Service Unavailable\r\n" b"Connection: close\r\n" @@ -133,6 +134,7 @@ class QuietHTTPServer(ThreadingHTTPServer): self.ssl_context: object | None = None super().__init__(*args, **kwargs) self._request_worker_slots = threading.BoundedSemaphore(self.max_request_workers) + self._overflow_reject_slots = threading.BoundedSemaphore(self.max_overflow_reject_workers) self.accept_loop_requests_total = 0 self.accept_loop_last_request_at = 0.0 @@ -226,7 +228,24 @@ class QuietHTTPServer(ThreadingHTTPServer): pass def _reject_overflow_request(self, request) -> None: - if getattr(self, "ssl_context", None) is None: + if getattr(self, "ssl_context", None) is not None: + self._close_request_quietly(request) + return + if not self._overflow_reject_slots.acquire(blocking=False): + self._close_request_quietly(request) + return + try: + threading.Thread( + target=self._reject_overflow_request_worker, + args=(request,), + daemon=True, + ).start() + except Exception: + self._overflow_reject_slots.release() + self._close_request_quietly(request) + + def _reject_overflow_request_worker(self, request) -> None: + try: self._drain_request_input_nonblocking(request) try: request.sendall(self._OVERFLOW_RESPONSE) @@ -236,7 +255,9 @@ class QuietHTTPServer(ThreadingHTTPServer): pass except Exception: pass - self._close_request_quietly(request) + finally: + self._close_request_quietly(request) + self._overflow_reject_slots.release() def process_request(self, request, client_address): if not self._request_worker_slots.acquire(blocking=False): diff --git a/tests/test_issue5210_http_worker_bound.py b/tests/test_issue5210_http_worker_bound.py index 3a271722..3acfbeea 100644 --- a/tests/test_issue5210_http_worker_bound.py +++ b/tests/test_issue5210_http_worker_bound.py @@ -22,12 +22,12 @@ def _free_port() -> int: return s.getsockname()[1] -def _worker_count(server: QuietHTTPServer) -> int: +def _worker_count(server: _ObservedQuietHTTPServer) -> int: with server.worker_lock: return server.worker_starts -def _wait_for_worker_count(server: QuietHTTPServer, expected: int, *, timeout: float = 2.0) -> None: +def _wait_for_worker_count(server: _ObservedQuietHTTPServer, expected: int, *, timeout: float = 2.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: current = _worker_count(server) @@ -39,7 +39,7 @@ def _wait_for_worker_count(server: QuietHTTPServer, expected: int, *, timeout: f raise AssertionError(f"timed out waiting for worker count {expected}, got {_worker_count(server)}") -def _assert_worker_count_stays(server: QuietHTTPServer, expected: int, *, duration: float = 0.35) -> None: +def _assert_worker_count_stays(server: _ObservedQuietHTTPServer, expected: int, *, duration: float = 0.35) -> None: deadline = time.monotonic() + duration while time.monotonic() < deadline: current = _worker_count(server) @@ -47,7 +47,7 @@ def _assert_worker_count_stays(server: QuietHTTPServer, expected: int, *, durati time.sleep(0.01) -def _wait_for_worker_slot_release(server: QuietHTTPServer, *, timeout: float = 2.0) -> None: +def _wait_for_worker_slot_release(server: _ObservedQuietHTTPServer, *, timeout: float = 2.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if server._request_worker_slots.acquire(blocking=False): @@ -210,6 +210,47 @@ def test_over_capacity_plain_http_gets_fast_503_without_starting_handler(monkeyp assert hold_result["value"][0] == 200 +def test_slow_overflow_cleanup_does_not_block_later_rejects(monkeypatch): + monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False) + drain_entered = threading.Event() + release_drain = threading.Event() + original_drain = QuietHTTPServer._drain_request_input_nonblocking + calls = 0 + + def blocking_first_drain(self, request) -> None: + nonlocal calls + calls += 1 + if calls == 1: + drain_entered.set() + assert release_drain.wait(timeout=5) + return + original_drain(self, request) + + monkeypatch.setattr(QuietHTTPServer, "_drain_request_input_nonblocking", blocking_first_drain) + + with _ServerRunner(_GateHandler) as srv: + hold_thread, hold_result = _start_request_thread(srv.port, "/hold") + assert srv.httpd.hold_entered.wait(timeout=5) + _wait_for_worker_count(srv.httpd, 1) + + slow_sock = socket.create_connection(("127.0.0.1", srv.port), timeout=2) + try: + assert drain_entered.wait(timeout=5) + status, headers, body = _request(srv.port, "/fast", timeout=1) + assert status == 503 + assert headers["Connection"].lower() == "close" + assert body == b"" + _assert_worker_count_stays(srv.httpd, 1) + finally: + release_drain.set() + slow_sock.close() + + srv.httpd.release_event.set() + hold_thread.join(timeout=5) + assert "error" not in hold_result, hold_result.get("error") + assert hold_result["value"][0] == 200 + + def test_worker_slot_releases_after_request_finishes(monkeypatch): monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False)