From a74d0c659d729d8bda8c1e0293ba38792026e0cf Mon Sep 17 00:00:00 2001 From: szeka9 Date: Tue, 7 Jul 2026 21:11:03 +0200 Subject: [PATCH] Refactor connection timeout and connection admission; reduce idle timeout Remove HttpServer.can_handle_new_client() responsible for evicting stale/inactive connections. Idle connection timeout is already part of the HTTP connection handling logic (see also HttpConnection.RECV_TIMEOUT_SECONDS). Instead, use the availibility of the buffer pool as the admission control mechanism, and wait for each connection to reserve a sender and receiver buffer pair. Make BaseConnection.close() idempontent, and allow the close method to be called multiple times during error handling for example. In such cases, BaseConnection.close() returns immediately instead of rerunning cleanup steps. Reduce idle timeout from 10 seconds to 3 seconds. This allows other clients to wait less for available stream buffers. Additional changes: - move client removal to the end of the finally branch - remove HttpServer._drop_client() and make it inline, as it is only referenced in one place - update description of class variables in HttpConnection --- docs/application_development/concepts.md | 8 +-- src/pyrobusta/bindings/http_connection.py | 6 +-- src/pyrobusta/server/http_server.py | 65 +++++++---------------- src/pyrobusta/transport/connection.py | 3 +- 4 files changed, 30 insertions(+), 52 deletions(-) diff --git a/docs/application_development/concepts.md b/docs/application_development/concepts.md index 50ce84b..1ca9810 100644 --- a/docs/application_development/concepts.md +++ b/docs/application_development/concepts.md @@ -57,8 +57,9 @@ and associates each connection with a `StreamReader` and `StreamWriter` pair for Each client connection executes as an independent asynchronous task, allowing multiple connections to make progress concurrently. `HttpServer` maintains the set of active connections and accepts -new connections until the configured maximum number of concurrent connections is reached. `HttpServer` -closes idle connections after the configured timeout to reclaim resources. Because each connection is +new connections until the configured maximum number of concurrent connections is reached. When the +server is at capacity, new connections wait for resources to become available until a configurable +timeout expires. If the timeout is reached, the connection is rejected. Because each connection is processed independently, slow clients do not block unrelated connections. However, requests within the same connection are always processed sequentially. @@ -101,7 +102,8 @@ for normal request processing is reserved when a connection is accepted, rather For each new connection, the server reserves a request and response stream buffer from the shared buffer pool. When all stream buffers are reserved, new connections are blocked until a buffer pair becomes available -(either through connection closure or timeout). `HttpServer` automatically closes connections that remain inactive for too long. +(either through connection closure or timeout). Connections that remain inactive for too long will +automatically time out, enforced by `HttpConnection`. Stream buffers are preallocated during server initialization based on the configured memory limits. Their size and number remain fixed throughout the lifetime of the server. **Preallocating stream buffers reduces heap diff --git a/src/pyrobusta/bindings/http_connection.py b/src/pyrobusta/bindings/http_connection.py index 0b8236c..ac04783 100644 --- a/src/pyrobusta/bindings/http_connection.py +++ b/src/pyrobusta/bindings/http_connection.py @@ -17,9 +17,9 @@ class HttpConnection(BaseConnection): buffer management and state machine parser. """ - MTU_SIZE = 1460 - STATE_MACHINE_SLEEP_MS = 2 - RECV_TIMEOUT_SECONDS = 10 + MTU_SIZE = 1460 # Maximum Transmission Unit size for TCP packets, in bytes + STATE_MACHINE_SLEEP_MS = 2 # Duration of sleep between state machine iterations + RECV_TIMEOUT_SECONDS = 3 # Timeout for reading request data from socket, in seconds __slots__ = ("_engine", "_prev_state", "_recv_buf", "_send_buf") diff --git a/src/pyrobusta/server/http_server.py b/src/pyrobusta/server/http_server.py index a9e0219..c8e838e 100644 --- a/src/pyrobusta/server/http_server.py +++ b/src/pyrobusta/server/http_server.py @@ -94,17 +94,6 @@ def _init_pools(cls, max_clients): cls.RECV_POOL = MemoryPool(recv_size, con_limit, wrapper=SlidingBuffer) cls.SEND_POOL = MemoryPool(send_size, con_limit, wrapper=SlidingBuffer) - @classmethod - async def _drop_client(cls, client): - """ - Remove client from the list of active clients. - """ - if client not in cls.ACTIVE_CLIENTS: - return - logging.debug(__name__ + f": {client.id} dropped") - await client.close() - cls.ACTIVE_CLIENTS.remove(client) - # ---------------- # Instance methods # ---------------- @@ -119,30 +108,6 @@ def __init__(self): self._server = None self._max_clients = 0 - async def can_handle_new_client(self): - """ - Decide if the new socket can be handled. - Evict closed/inactive sockets if needed. - :return is_acceptable: true/false - """ - con_timestamp = ticks_ms() - while ticks_diff(ticks_ms(), con_timestamp) < self.CON_ACCEPT_TIMEOUT_MS: - if len(self.ACTIVE_CLIENTS) < self._max_clients: - return True - # Attempt to evict inactive clients - for client in self.ACTIVE_CLIENTS: - client_inactive = int(ticks_diff(ticks_ms(), client.last_event) * 0.001) - if not client.connected or client_inactive > self.CON_TIMEOUT_S: - logging.debug( - ( - __name__ + f": evicted {client.id} " - f"timeout: {self.CON_TIMEOUT_S - client_inactive}s" - ) - ) - await self._drop_client(client) - await sleep_ms(self.CON_ACCEPT_SLEEP_MS) - return False - async def _reserve_buffers(self): """ Reserve and return request and response buffers. @@ -152,8 +117,9 @@ async def _reserve_buffers(self): recv_buf = None send_buf = None + deadline = ticks_ms() + self.CON_ACCEPT_TIMEOUT_MS - while not recv_buf or not send_buf: + while (not recv_buf or not send_buf) and ticks_diff(ticks_ms(), deadline) < 0: if not recv_buf: recv_buf = self.RECV_POOL.reserve() if not send_buf: @@ -169,15 +135,21 @@ async def _accept_socket(self, reader, writer): :param reader: asyncio StreamReader :param reader: asyncio StreamWriter """ - if not await self.can_handle_new_client(): - logging.debug(__name__ + ": cannot accept new client") - writer.close() - await writer.wait_closed() - return - client = None try: recv_buf, send_buf = await self._reserve_buffers() + + if recv_buf is None or send_buf is None: + logging.debug( + __name__ + + ": connection from " + + writer.get_extra_info("peername")[0] + + " rejected (server at capacity)" + ) + writer.close() + await writer.wait_closed() + return + client = HttpConnection(reader, writer, recv_buf, send_buf) logging.debug(__name__ + f": accept {client.id}") self.ACTIVE_CLIENTS.append(client) @@ -186,14 +158,14 @@ async def _accept_socket(self, reader, writer): except Exception as e: # pylint: disable=W0718 logging.warning(__name__ + f": error in run(): {e}") finally: - if client: - self.ACTIVE_CLIENTS.remove(client) if send_buf: send_buf.consume() self.SEND_POOL.release(send_buf) if recv_buf: recv_buf.consume() self.RECV_POOL.release(recv_buf) + if client and client in self.ACTIVE_CLIENTS: + self.ACTIVE_CLIENTS.remove(client) collect() async def start_socket_server(self): @@ -234,7 +206,10 @@ async def terminate(self): """ logging.info(__name__ + ": terminated") while self.ACTIVE_CLIENTS: - await self._drop_client(self.ACTIVE_CLIENTS[0]) + client = self.ACTIVE_CLIENTS[0] + logging.debug(__name__ + f": {client.id} dropped") + self.ACTIVE_CLIENTS.remove(client) + await client.close() if self._server: self._server.close() await self._server.wait_closed() diff --git a/src/pyrobusta/transport/connection.py b/src/pyrobusta/transport/connection.py index aa5d7bd..c0bf8c6 100644 --- a/src/pyrobusta/transport/connection.py +++ b/src/pyrobusta/transport/connection.py @@ -78,7 +78,8 @@ async def close(self): Close the connection, update the internal state accordingly. """ if not self.connected: - return OSError(f"{self.id} already closed") + logging.warning(f"{self.id} already closed") + return self.connected = False logging.debug(__name__ + f": close connection: {self.id}")