The socket loop should have one job: wait for readiness, accept connections, and transfer each accepted socket without doing connection work itself. If it reads requests or waits on a worker inside the loop, one slow client can delay every new connection.
Use nonblocking sockets with a selector. A listening socket can become readable while several connections are queued, so accept repeatedly until the queue is empty. Put a limit on each pass: under a sustained connection flood, the loop still needs to return to the selector regularly. With Python’s level-triggered selectors, any remaining queued connections will make the listener readable again.
import selectors
import socket
def serve(host, port, dispatch, stop_event):
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind((host, port))
listener.listen()
listener.setblocking(False)
try:
with selectors.DefaultSelector() as selector:
selector.register(listener, selectors.EVENT_READ)
while not stop_event.is_set():
for key, _ in selector.select(timeout=0.25):
if key.fileobj is not listener:
continue
for _ in range(64):
try:
client, address = listener.accept()
except BlockingIOError:
break
except InterruptedError:
continue
client.setblocking(False)
try:
# dispatch must return promptly and take ownership
# of client only when it succeeds.
dispatch(client, address)
except Exception:
client.close()
raise
finally:
listener.close()
The accept budget prevents a busy listener from monopolizing the loop. The selector timeout lets the loop notice a shutdown request even when no connection arrives; a wakeup socket or pipe can replace polling when shutdown latency matters. Keep the listening socket’s backlog separate from application capacity: the backlog controls pending connections in the kernel, not how many clients the server can safely process.
The ownership rule matters. Once dispatch succeeds, the loop must not close or otherwise manage that client socket; before success, it remains responsible for closing it if dispatch fails. Keep dispatch limited to a quick handoff, since
OURA Ring 5 Sizing Kit - Size Before You Buy | Unique Sizing, Not Standard Ring Sizing - Receive Amazon Credit for Oura Ring 5 Purchase
$10.00 (as of November 4, 2026 10:12 GMT +00:00 - More infoProduct prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on [relevant Amazon Site(s), as applicable] at the time of purchase will apply to the purchase of this product.)Handing each connection to a worker thread
…the queue operation itself must not wait for a worker. Give the handoff a bounded queue and make saturation an explicit result. An unbounded queue merely turns overload into rising memory use and growing latency; a blocking enqueue moves the same delay back into the accept loop.
import queue
import threading
jobs = queue.Queue(maxsize=256)
def dispatch(client, address):
try:
jobs.put_nowait((client, address))
except queue.Full:
return False
return True
def worker():
while True:
item = jobs.get()
try:
if item is None:
return
client, address = item
try:
client.settimeout(10.0)
handle_connection(client, address)
except (OSError, TimeoutError):
pass
finally:
client.close()
finally:
jobs.task_done()
workers = [
threading.Thread(target=worker, name=f"client-worker-{i}")
for i in range(16)
]
for thread in workers:
thread.start()
The loop must close a socket when the queue refuses it; after a successful enqueue, the worker owns it. That makes the ownership boundary concrete and prevents both leaks and double closes. Replace the handoff in the accept loop with:
client.setblocking(False)
try:
accepted = dispatch(client, address)
except Exception:
client.close()
raise
if not accepted:
client.close()
Each worker changes its socket to blocking mode with a finite timeout before calling the request handler. That keeps ordinary request code simple while ensuring a client that sends only part of a request cannot occupy a thread forever. A timeout is not a substitute for limits on request size, response size, or total processing time; those limits belong in the protocol handler.
Keep the worker count and queue capacity deliberate. More threads can increase throughput when handlers wait on disk or other I/O, but they also consume stack space and add scheduling overhead. A queue of 256 does not mean 256 connections are being processed: it means at most 256 accepted sockets can wait in addition to the sockets already held by workers. When the queue is full, closing the newly accepted connection is a simple overload policy; a server with a different protocol may instead send a small busy response before closing.
Do not let exceptions escape the worker’s outer loop. A malformed request or a disconnected peer is a property of one connection, not a reason to lose a worker thread. Catch expected per-connection failures around the handler, close the socket in finally, and keep task_done paired with every successful get. Shutdown then has to account for both active handlers and queued sockets, since
Keeping shared state safe
…the state shared by those handlers needs its own ownership rules. Threads can read and modify the same objects at any time; the fact that a particular operation appears to be one line of Python does not make a larger sequence of operations indivisible. Protect invariants, not individual assignments. If a request must check a quota and then increment usage, both steps belong inside the same critical section.
import threading
class Usage:
def __init__(self, limit):
self._lock = threading.Lock()
self._limit = limit
self._by_client = {}
def reserve(self, client_id, amount):
with self._lock:
used = self._by_client.get(client_id, 0)
if used + amount > self._limit:
return False
self._by_client[client_id] = used + amount
return True
def release(self, client_id, amount):
with self._lock:
used = self._by_client.get(client_id, 0) - amount
if used > 0:
self._by_client[client_id] = used
else:
self._by_client.pop(client_id, None)
The lock makes the check-and-update one operation from the point of view of other workers. Without it, two requests can both observe the same available capacity and both reserve it. Use a context manager so exceptions cannot leave the lock held. Keep the protected region short: perform parsing, disk access, logging, and network writes outside the lock whenever possible. Holding a global lock while waiting on a client turns a thread pool into a single-threaded server.
For counters, guard the read-modify-write sequence as well. Do not rely on the interpreter’s implementation details or on the global interpreter lock to preserve application-level invariants.
class Metrics:
def __init__(self):
self._lock = threading.Lock()
self._completed = 0
self._failed = 0
def record(self, failed):
with self._lock:
if failed:
self._failed += 1
else:
self._completed += 1
def snapshot(self):
with self._lock:
return self._completed, self._failed
Return a snapshot rather than exposing the mutable dictionary or counters. A caller that receives a reference to internal state can bypass the lock entirely, even if every method in the class uses it correctly. For read-heavy state, an immutable snapshot can also be built under the lock and then used after releasing it; the lock protects the copy, not the work performed with the copy.
When an operation needs two locks, define one acquisition order and use it everywhere. Otherwise, one worker can hold lock A while waiting for lock B as another holds B while waiting for A. Better still, avoid nested locks by combining related state under one lock or by transferring updates through a queue to a single owner. That design costs some coordination but makes the state transition explicit.
Do not invoke callbacks while holding a state lock. A callback may acquire another lock, re-enter the same component, or block on I/O; any of those can create deadlock or unexpectedly long pauses. Copy the data needed for the callback while locked, release the lock, and then call it. If callbacks can mutate the source state, define whether they receive a snapshot or whether their changes must be submitted as a separate operation.
Testing concurrency under load
Concurrency bugs rarely appear in a single request. Test with many clients at once, and check both the answers and the server’s resource use. A test that reports only requests per second can miss a growing queue, stalled workers, or a small number of requests taking seconds while the average remains low.
Measure each request from the client’s point of view. The following harness assumes a newline-delimited request and response; replace the exchange with the protocol the server actually implements. Keep connection setup inside the timed region if production clients open a connection for each request.
import concurrent.futures
import socket
import time
def one_request(host, port, payload, timeout=5.0):
started = time.perf_counter()
try:
with socket.create_connection((host, port), timeout=timeout) as client:
client.settimeout(timeout)
client.sendall(payload + b"n")
with client.makefile("rb") as stream:
response = stream.readline()
if not response:
raise RuntimeError("server closed without a response")
return time.perf_counter() - started, response, None
except Exception as error:
return time.perf_counter() - started, None, error
def run_batch(host, port, payload, requests, concurrency):
results = []
with concurrent.futures.ThreadPoolExecutor(
max_workers=concurrency
) as pool:
futures = [
pool.submit(one_request, host, port, payload)
for _ in range(requests)
]
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
return results
Run several concurrency levels, not just one large batch. A server may handle a few dozen clients efficiently and then hit a sharp limit when its worker pool and queue fill. At each level, record completed requests, failures, response correctness, and latency percentiles. Report the offered load as well: a test that completes 500 requests in a minute says little unless it also says how many clients were trying to make progress.
def percentile(values, fraction):
ordered = sorted(values)
if not ordered:
return None
index = round((len(ordered) - 1) * fraction)
return ordered[index]
for concurrency in (1, 4, 8, 16, 32, 64):
results = run_batch(
"127.0.0.1",
9000,
b"ping",
requests=concurrency * 20,
concurrency=concurrency,
)
latencies = [elapsed for elapsed, _, _ in results]
failures = [error for _, _, error in results if error is not None]
wrong = [
response
for _, response, error in results
if error is None and response != b"pongn"
]
print(
"clients=", concurrency,
"completed=", len(results) - len(failures),
"errors=", len(failures),
"wrong=", len(wrong),
"p50=", percentile(latencies, 0.50),
"p95=", percentile(latencies, 0.95),
"p99=", percentile(latencies, 0.99),
)
Use a fixed request count for one experiment, then use a fixed duration for another. A fixed batch exposes the time needed to drain a burst; a duration test exposes behavior under sustained arrivals. Increase the load gradually and leave each stage running long enough to reveal trends. If latency rises from one sample window to the next while throughput stays flat, work is accumulating somewhere, often in the handoff queue.
Exercise the overload path deliberately. Hold workers busy, fill the bounded queue, and confirm that new connections are rejected or receive the documented busy response. The test should also verify that rejected sockets are closed, worker threads remain alive, and later requests succeed after the load subsides. A server that survives the first burst but never recovers has not passed the test.
Test shared-state invariants with concurrent operations whose expected result is known. For example, let many clients reserve from one quota and assert that the total successful reservation never exceeds the limit. Check the result through the server’s supported interface rather than inspecting its internal dictionary from the test; the public behavior is the invariant that matters.
Watch process-level resources during the run: open file descriptors, thread count, memory, queue depth, and active connections. Repeat the same workload after the server has been idle. If these values keep climbing across otherwise identical runs, investigate sockets that are not closed, queue items that are not completed, or per-connection state that is retained.
Include slow and broken clients, not only well-behaved request generators. Some clients should connect and send nothing; others should send half a request and pause, disconnect during a response, or send an oversized message. Confirm that timeouts and protocol limits release their worker slots, then continue measuring normal clients while those cases are present.
