diff --git a/gunicorn/workers/gthread.py b/gunicorn/workers/gthread.py index bf7711fc4..dd928b397 100644 --- a/gunicorn/workers/gthread.py +++ b/gunicorn/workers/gthread.py @@ -29,13 +29,16 @@ from ..http import wsgi -# Sentinel value to indicate connection should be deferred back to poller -_DEFER = object() +# Default minimum timeout (in seconds) for waiting for request data after accept connection. +# This value serves as a lower bound: the worker uses the larger of this default and the keep-alive timeout. +# If no data arrives within the effective wait period, the connection is closed to avoid holding resources indefinitely. +DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT = 5.0 -# Default timeout (in seconds) for waiting for request data in worker thread. -# If no data arrives within this timeout, the connection is deferred back to -# the main poller to prevent thread pool exhaustion from slow clients. -DEFAULT_WORKER_DATA_TIMEOUT = 5.0 +# Default timeout (in seconds) allowed for fully draining any unread request body +# before reusing the connection in keep-alive mode. +# If the body cannot be drained within this time (for example, due to a stalled client), +# keep-alive is abandoned and the connection is closed, so that the worker thread is not blocked indefinitely. +DEFAULT_KEEPALIVE_BODY_DRAIN_TIMEOUT = 5.0 class TConn: @@ -50,8 +53,6 @@ def __init__(self, cfg, sock, client, server): self.parser = None self.initialized = False self.is_http2 = False - # Track if we've already waited for data (to avoid waiting again after defer) - self.data_ready = False # set the socket to non blocking self.sock.setblocking(False) @@ -89,37 +90,6 @@ def set_timeout(self): # Use monotonic clock for reliability (time.time() can jump due to NTP) self.timeout = time.monotonic() + self.cfg.keepalive - def wait_for_data(self, timeout): - """Wait for data to be available on the socket. - - Uses selectors to wait for the socket to become readable within - the given timeout. This prevents slow clients from blocking - thread pool slots indefinitely. - - Args: - timeout: Maximum time to wait in seconds. - - Returns: - True if data is available, False if timeout expired. - """ - if self.data_ready: - return True - - # Use a temporary selector to wait for data - sel = selectors.DefaultSelector() - try: - sel.register(self.sock, selectors.EVENT_READ) - events = sel.select(timeout=timeout) - if events: - self.data_ready = True - return True - return False - except (OSError, ValueError): - # Socket closed or invalid - return False - finally: - sel.close() - def close(self, graceful=False): if graceful: self.sock.setblocking(True) @@ -283,12 +253,18 @@ def accept(self, listener): try: client_sock, client_addr = listener.accept() self.nr_conns += 1 - client_sock.setblocking(True) conn = TConn(self.cfg, client_sock, client_addr, listener.getsockname()) - # Submit directly to thread pool for processing - self.enqueue_req(conn) + conn.timeout = time.monotonic() + max(DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT, self.cfg.keepalive) + + self.pending_conns.append(conn) + self.poller.register( + conn.sock, + selectors.EVENT_READ, + partial(self.on_pending_socket_readable, conn), + ) + except OSError as e: if e.errno not in (errno.EAGAIN, errno.ECONNABORTED, errno.EWOULDBLOCK): raise @@ -306,9 +282,6 @@ def on_pending_socket_readable(self, conn, client): self.poller.unregister(client) self.pending_conns.remove(conn) - # Mark data as ready so we don't wait again in handle() - conn.data_ready = True - # Submit to thread pool for processing self.enqueue_req(conn) @@ -418,18 +391,9 @@ def run(self): def finish_request(self, conn, fs): """Handle completion of a request (called via method_queue on main thread).""" try: - result = fs.result() if not fs.cancelled() else False + keepalive = not fs.cancelled() and fs.result() - if result is _DEFER and self.alive: - # Connection deferred - no data arrived within timeout. - # Put it on the poller to wait for data without consuming a thread. - conn.sock.setblocking(False) - # Use keepalive timeout for pending connections too - conn.timeout = time.monotonic() + self.cfg.keepalive - self.pending_conns.append(conn) - self.poller.register(conn.sock, selectors.EVENT_READ, - partial(self.on_pending_socket_readable, conn)) - elif result and self.alive: + if keepalive and self.alive: # Keepalive - put connection back in the poller conn.sock.setblocking(False) conn.set_timeout() @@ -447,16 +411,6 @@ def handle(self, conn): """Handle a request on a connection. Runs in a worker thread.""" req = None try: - # For new connections (not yet initialized), wait for data with timeout - # to prevent slow clients from blocking thread pool slots indefinitely. - # Skip this for already-initialized connections (keepalive, deferred) - # since they're coming from the poller and data is already available. - if not conn.initialized and not conn.data_ready: - # Wait for data with timeout before committing this thread - if not conn.wait_for_data(DEFAULT_WORKER_DATA_TIMEOUT): - # No data within timeout - defer to poller - return _DEFER - # Always ensure blocking mode in worker thread. # Critical for keepalive connections: the socket is set to non-blocking # for the selector in finish_request(), but must be blocking for @@ -480,9 +434,9 @@ def handle(self, conn): if keepalive: # Discard any unread request body before keepalive to prevent # the socket from appearing readable due to leftover bytes. - # Bound the drain by the worker data timeout: a stalled client + # Bound the drain by the worker drain timeout: a stalled client # must not keep this thread blocked. - drain_deadline = time.monotonic() + DEFAULT_WORKER_DATA_TIMEOUT + drain_deadline = time.monotonic() + DEFAULT_KEEPALIVE_BODY_DRAIN_TIMEOUT if not conn.parser.finish_body(deadline=drain_deadline): # Abandon keepalive when the body could not be fully drained. return False diff --git a/tests/test_gthread.py b/tests/test_gthread.py index fbaf38109..b26a2518b 100644 --- a/tests/test_gthread.py +++ b/tests/test_gthread.py @@ -499,13 +499,16 @@ def test_finish_request_exception(self): class TestAccept: """Tests for connection acceptance.""" - def create_worker(self): + def create_worker(self, keepalive=None): """Create a worker for testing.""" cfg = Config() cfg.set('workers', 1) cfg.set('threads', 4) cfg.set('worker_connections', 1000) + if keepalive is not None: + cfg.set('keepalive', keepalive) + worker = gthread.ThreadWorker( age=1, ppid=os.getpid(), @@ -520,9 +523,10 @@ def create_worker(self): worker.method_queue = mock.Mock() return worker - def test_accept_success(self): + @pytest.mark.parametrize("keepalive", [None, 1, 6]) + def test_accept_success(self, keepalive): """Test successful connection acceptance.""" - worker = self.create_worker() + worker = self.create_worker(keepalive) worker.nr_conns = 0 client_sock = FakeSocket() @@ -534,7 +538,11 @@ def test_accept_success(self): worker.accept(listener) assert worker.nr_conns == 1 - worker.tpool.submit.assert_called_once() + + assert len(worker.pending_conns) == 1 + assert worker.pending_conns[0].sock == client_sock + + worker.tpool.submit.assert_not_called() def test_accept_eagain(self): """Test handling of EAGAIN during accept.""" @@ -1569,10 +1577,9 @@ def send_trailers_h2(trailers): class TestSlowClientResilience: """Tests for slow client handling to prevent thread pool exhaustion.""" - def create_worker(self, cfg=None): + def create_worker(self): """Helper to create a ThreadWorker for testing.""" - if cfg is None: - cfg = Config() + cfg = Config() cfg.set('threads', 4) cfg.set('worker_connections', 1000) cfg.set('keepalive', 5) @@ -1588,88 +1595,7 @@ def create_worker(self, cfg=None): ) return worker - def test_tconn_wait_for_data_returns_true_when_ready(self): - """Test wait_for_data returns True when data_ready is already set.""" - cfg = Config() - sock = FakeSocket() - conn = gthread.TConn(cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.data_ready = True - - # Should return True immediately without waiting - assert conn.wait_for_data(5.0) is True - - def test_tconn_wait_for_data_sets_data_ready(self): - """Test wait_for_data sets data_ready flag when data arrives.""" - import socket as stdlib_socket - # Create a real socket pair to test selector behavior - server, client = stdlib_socket.socketpair() - try: - cfg = Config() - conn = gthread.TConn(cfg, server, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.data_ready = False - - # Send data from client - client.send(b'GET / HTTP/1.1\r\n') - - # Should detect data is ready - result = conn.wait_for_data(1.0) - - assert result is True - assert conn.data_ready is True - finally: - server.close() - client.close() - - def test_tconn_wait_for_data_timeout(self): - """Test wait_for_data returns False on timeout.""" - import socket as stdlib_socket - # Create a real socket pair but don't send any data - server, client = stdlib_socket.socketpair() - try: - cfg = Config() - conn = gthread.TConn(cfg, server, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.data_ready = False - - # Don't send any data - should timeout - start = time.monotonic() - result = conn.wait_for_data(0.1) # Short timeout - elapsed = time.monotonic() - start - - assert result is False - assert conn.data_ready is False - assert elapsed >= 0.1 - finally: - server.close() - client.close() - - def test_finish_request_handles_defer(self): - """Test finish_request puts deferred connections back on poller.""" - worker = self.create_worker() - worker.poller = mock.Mock() - worker.pending_conns = deque() - worker.nr_conns = 1 - worker.alive = True - - sock = FakeSocket() - conn = gthread.TConn(worker.cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - - # Create a future that returns _DEFER - fs = mock.Mock() - fs.cancelled.return_value = False - fs.result.return_value = gthread._DEFER - - worker.finish_request(conn, fs) - - # Connection should be in pending_conns, not closed - assert len(worker.pending_conns) == 1 - assert worker.pending_conns[0] is conn - assert worker.nr_conns == 1 # Still counted - assert not sock.closed - - # Should be registered with poller - worker.poller.register.assert_called_once() - - def test_on_pending_socket_readable_sets_data_ready(self): + def test_on_pending_socket_readable(self): """Test on_pending_socket_readable marks connection data as ready.""" worker = self.create_worker() worker.poller = mock.Mock() @@ -1679,13 +1605,11 @@ def test_on_pending_socket_readable_sets_data_ready(self): sock = FakeSocket() conn = gthread.TConn(worker.cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.data_ready = False worker.pending_conns.append(conn) # Simulate socket becoming readable worker.on_pending_socket_readable(conn, sock) - assert conn.data_ready is True assert conn not in worker.pending_conns worker.poller.unregister.assert_called_once_with(sock) worker.tpool.submit.assert_called_once() @@ -1717,62 +1641,3 @@ def test_murder_pending_closes_expired_connections(self): assert len(worker.pending_conns) == 1 assert worker.pending_conns[0] is conn2 assert worker.nr_conns == 1 - - def test_handle_defers_slow_connection(self): - """Test that handle() returns _DEFER for connections without data.""" - worker = self.create_worker() - - # Create a connection that will timeout waiting for data - sock = mock.Mock() - conn = gthread.TConn(worker.cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.initialized = False - conn.data_ready = False - - # Mock wait_for_data to simulate timeout - conn.wait_for_data = mock.Mock(return_value=False) - - result = worker.handle(conn) - - assert result is gthread._DEFER - conn.wait_for_data.assert_called_once() - - def test_handle_processes_fast_connection(self): - """Test that handle() processes connections with data immediately.""" - worker = self.create_worker() - worker.wsgi = mock.Mock(return_value=[b'OK']) - - # Create a connection with data ready - sock = mock.Mock() - conn = gthread.TConn(worker.cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.initialized = False - conn.data_ready = True # Data is ready - - # Mock init and parser - conn.init = mock.Mock() - conn.parser = mock.Mock() - conn.parser.__next__ = mock.Mock(return_value=None) # No request parsed - - result = worker.handle(conn) - - # Should not return _DEFER since data was ready - assert result is not gthread._DEFER - conn.init.assert_called_once() - - def test_handle_skips_wait_for_initialized_connections(self): - """Test handle() skips wait_for_data for already initialized (keepalive) connections.""" - worker = self.create_worker() - worker.wsgi = mock.Mock(return_value=[b'OK']) - - sock = mock.Mock() - conn = gthread.TConn(worker.cfg, sock, ('127.0.0.1', 12345), ('127.0.0.1', 8000)) - conn.initialized = True # Already initialized (keepalive) - conn.data_ready = False - conn.wait_for_data = mock.Mock() - - conn.parser = mock.Mock() - conn.parser.__next__ = mock.Mock(return_value=None) - - worker.handle(conn) - - # wait_for_data should not be called for initialized connections - conn.wait_for_data.assert_not_called() diff --git a/tests/test_gthread_integration.py b/tests/test_gthread_integration.py new file mode 100644 index 000000000..e7b2a3829 --- /dev/null +++ b/tests/test_gthread_integration.py @@ -0,0 +1,224 @@ +"""Integration tests for gthread worker.""" +import os +import signal +import socket +import subprocess +import sys +import time + +import pytest + + +# Timeout for CI environments +CI_TIMEOUT = 30 + + +# Simple WSGI app +SIMPLE_APP = ''' +def application(environ, start_response): + """Basic hello world response.""" + status = '200 OK' + body = b'Hello, World!' + headers = [ + ('Content-Type', 'text/plain'), + ('Content-Length', str(len(body))), + ] + start_response(status, headers) + return [body] +''' + + +def find_free_port(): + """Find a free port to bind to.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('127.0.0.1', 0)) + return s.getsockname()[1] + + +def wait_for_server(host, port, timeout=CI_TIMEOUT): + """Wait until server is accepting connections.""" + start = time.monotonic() + while time.monotonic() - start < timeout: + try: + with socket.create_connection((host, port), timeout=1): + return True + except (ConnectionRefusedError, socket.timeout, OSError): + time.sleep(0.1) + return False + + +def make_request(host, port, path='/', sleep_after_create_connection=0): + """Make a simple HTTP request and return the response body. + + Raises ConnectionError if the server closes the connection unexpectedly + (e.g. due to pending request timeout). + """ + with socket.create_connection((host, port), timeout=5) as sock: + # emulate stalled connection + time.sleep(sleep_after_create_connection) + + request = f'GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n' + sock.sendall(request.encode()) + + # Signal that we're done sending; helps detect connection loss on write + try: + sock.shutdown(socket.SHUT_WR) + except OSError: + # Server already closed the connection before or during our send + raise ConnectionError("Server closed the connection before completing the request") + + response = b'' + while True: + chunk = sock.recv(4096) + if not chunk: + break + response += chunk + + if not response: + raise ConnectionError( + "Server closed the connection unexpectedly; no response received" + ) + + return response + + +@pytest.fixture +def app_module(tmp_path): + """Create a temporary app module.""" + app_file = tmp_path / "app.py" + app_file.write_text(SIMPLE_APP) + return str(app_file.parent), "app:application" + + +def start_gunicorn(app_dir, app_name, port, keepalive=None): + """Start a gunicorn server with specified worker class and control socket.""" + cmd = [ + sys.executable, '-m', 'gunicorn', + '--bind', f'127.0.0.1:{port}', + '--workers', '1', + '--worker-class', 'gthread', + '--threads', '2', + '--access-logfile', '-', + '--error-logfile', '-', + '--log-level', 'debug', + '--timeout', '30', + '--no-control-socket', + ] + if keepalive is not None: + cmd += ['--keep-alive', str(keepalive)] + + cmd += [app_name] + + proc = subprocess.Popen( + cmd, + cwd=app_dir, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env={**os.environ, 'PYTHONPATH': app_dir}, + preexec_fn=os.setsid + ) + + return proc + + +def cleanup_gunicorn(proc): + """Clean up a gunicorn process.""" + if proc.poll() is None: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) + except (ProcessLookupError, OSError): + pass + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + proc.wait() + + +class TestPendingRequestLifecycle: + """Test pending requests safety in gthread worker.""" + + @pytest.mark.parametrize( + ('keepalive', 'sleep_after_create_connection'), [ + (None, 3), # used DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT + (2, 3), # used DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT + (8, 6), # used keep-alive timeout + ]) + def test_send_request_inside_pending_request_timeout( + self, + app_module, + tmp_path, + keepalive, + sleep_after_create_connection, + ): + """Request succeeds when client sends data within the pending request window.""" + + app_dir, app_name = app_module + port = find_free_port() + + proc = start_gunicorn(app_dir, app_name, port, keepalive=keepalive) + + try: + # Wait for server to start - should not deadlock + if not wait_for_server('127.0.0.1', port, timeout=15): + stdout, stderr = proc.communicate(timeout=1) + pytest.fail( + f"Gthread worker deadlocked during startup:\n" + f"stdout: {stdout.decode()}\n" + f"stderr: {stderr.decode()}" + ) + + # Verify server responds + response = make_request( + host='127.0.0.1', + port=port, + sleep_after_create_connection=sleep_after_create_connection, + ) + + assert b'Hello, World!' in response + + finally: + cleanup_gunicorn(proc) + + @pytest.mark.parametrize( + ('keepalive', 'sleep_after_create_connection'), [ + (None, 7), # used DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT + (2, 7), # used DEFAULT_PENDING_REQUEST_MIN_WAIT_TIMEOUT + (8, 10), # used keep-alive timeout + ]) + def test_send_request_outside_pending_request_timeout( + self, + app_module, + tmp_path, + keepalive, + sleep_after_create_connection, + ): + """Client gets a connection error when data arrives after the pending request window.""" + + app_dir, app_name = app_module + port = find_free_port() + + proc = start_gunicorn(app_dir, app_name, port, keepalive=keepalive) + + try: + # Wait for server to start - should not deadlock + if not wait_for_server('127.0.0.1', port, timeout=15): + stdout, stderr = proc.communicate(timeout=1) + pytest.fail( + f"Gthread worker deadlocked during startup:\n" + f"stdout: {stdout.decode()}\n" + f"stderr: {stderr.decode()}" + ) + + with pytest.raises(ConnectionError): + make_request( + host='127.0.0.1', + port=port, + sleep_after_create_connection=sleep_after_create_connection, + ) + + finally: + cleanup_gunicorn(proc)