[Triage] gthread evolution: lifecycle of an accepted connection

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

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()