Конкуррентность в сетевых серверах — threads, select/poll/epoll, async

В уроке 2 мы построили простой TCP-сервер, обрабатывающий клиентов sequentially — одного за раз. Это работает для демо, но в production unacceptable: один медленный клиент блокирует всех. Нужна concurrency — способность обслуживать множество клиентов одновременно.

В этом уроке — обзор моделей concurrency для сетевых серверов: от классических threads до современного async I/O. Каждая модель имеет свою область применимости, и понимание trade-offs позволит выбирать правильно.

Модель 1: Process per client (fork)

Самая старая модель. Главный процесс делает accept, затем fork() — ОС создаёт копию процесса. Child обрабатывает клиента, parent продолжает accept.

# Этот код работает только на Linux/macOS, не на Windows
import os
import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 9999))
server.listen(128)

while True:
    client, addr = server.accept()
    pid = os.fork()
    if pid == 0:
        # Child process
        server.close()  # closing parent's listening socket
        # Handle client
        while True:
            data = client.recv(4096)
            if not data:
                break
            client.sendall(data)
        client.close()
        os._exit(0)
    else:
        # Parent
        client.close()  # parent не работает с этим клиентом
        # Здесь имеет смысл reap zombie processes через signal handler

print("Listening")

Плюсы:

• Полная изоляция — crash в child не падает main.
• Параллельность на multi-core CPU.

Минусы:

• Дорого. Fork создаёт новый процесс — ~10MB+ memory each (хотя copy-on-write помогает).
• Сложно делиться state. Между процессами нужно IPC.
• Дорогой context switch между процессами.

Эта модель использовалась в классических серверах (Apache prefork). Сейчас редко — thread per client дешевле, async ещё дешевле.

Модель 2: Thread per client

Для каждого нового client — свой thread. В Python:

import socket
import threading

def handle_client(client_sock, client_addr):
    try:
        while True:
            data = client_sock.recv(4096)
            if not data:
                break
            client_sock.sendall(data)
    finally:
        client_sock.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 9999))
server.listen(128)

while True:
    client, addr = server.accept()
    threading.Thread(
        target=handle_client,
        args=(client, addr),
        daemon=True
    ).start()

Плюсы:

• Простой код — каждый thread пишется как sequential code.
• Дешевле processes (~1-2MB per thread vs ~10MB per process).
• Shared memory — легко делиться state (с осторожностью к race conditions).

Минусы:

• Не масштабируется до тысяч. Каждый thread = 1-2MB stack. 10000 threads = 10-20GB только на stacks.
• Context switching overhead. Чем больше threads, тем больше CPU тратится на switching.
• В Python: GIL. Глобальный Lock не даёт реальный CPU-parallelism в одном процессе. Для I/O-bound (network) — работает, для CPU-bound — нет.

Thread per client — хороший выбор для средних серверов с десятками-сотнями одновременных клиентов. Большинство Java/Go HTTP-серверов работают так (Go goroutines — легче threads, ~8KB вместо 1MB).

Thread pool

Вариант — иметь готовый pool из N threads, не создавать новые на каждого клиента:

from concurrent.futures import ThreadPoolExecutor
import socket

def handle_client(client_sock, client_addr):
    try:
        while True:
            data = client_sock.recv(4096)
            if not data:
                break
            client_sock.sendall(data)
    finally:
        client_sock.close()

executor = ThreadPoolExecutor(max_workers=100)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 9999))
server.listen(128)

while True:
    client, addr = server.accept()
    executor.submit(handle_client, client, addr)

Pool ограничивает максимальное число одновременно обрабатываемых клиентов — защищает от overload. Если все workers busy, новые клиенты ждут (но connection accepted, в backlog).

I/O multiplexing: select, poll, epoll

Идея: вместо thread per client — один thread следит за множеством sockets через специальный syscall. ОС сообщает: “вот эти sockets имеют данные для чтения”. Thread обрабатывает их, потом снова ждёт.

select() — классический

Самый старый API (1980s). Передаёшь массивы read_fds, write_fds, error_fds — получаешь те же массивы с активными fds.

import socket
import select

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 9999))
server.listen(128)
server.setblocking(False)

clients = []  # список client sockets

while True:
    # Hint select: "следи за этими sockets"
    rlist = [server] + clients
    readable, _, _ = select.select(rlist, [], [], 1.0)

    for sock in readable:
        if sock is server:
            # Это server socket -- accept new client
            client, addr = server.accept()
            client.setblocking(False)
            clients.append(client)
            print(f"New client: {addr}")
        else:
            # Это client socket -- recv
            try:
                data = sock.recv(4096)
                if not data:
                    print("Client disconnected")
                    clients.remove(sock)
                    sock.close()
                else:
                    sock.sendall(data)  # echo
            except ConnectionResetError:
                clients.remove(sock)
                sock.close()

Один thread обрабатывает сотни клиентов.

Минусы select:

1. O(n) на каждый вызов. Ядро итерирует ВСЕ переданные fds, проверяя каждый. На 10000 sockets это медленно.
2. Лимит FD_SETSIZE — обычно 1024 max sockets.
3. Копирование массивов на каждый вызов.

poll() — улучшенный select

poll решает FD_SETSIZE limit — передаёшь массив структур pollfd, без лимита. Но всё ещё O(n) per call.

import select

poller = select.poll()
poller.register(server.fileno(), select.POLLIN)

while True:
    events = poller.poll(1000)  # timeout 1000ms
    for fd, event in events:
        if event & select.POLLIN:
            # Этот fd имеет данные для чтения
            ...

В Python poll используется реже select, потому что эквивалентно по производительности.

epoll/kqueue — современный подход

epoll (Linux), kqueue (macOS/BSD) — самые эффективные I/O multiplexors. Ключевая идея: ядро хранит список fds постоянно (не передаётся на каждый вызов), и возвращает только активные.

import select

epoll = select.epoll()
epoll.register(server.fileno(), select.EPOLLIN)

connections = {}  # fd -> client socket

while True:
    events = epoll.poll(1.0)
    for fd, event in events:
        if fd == server.fileno():
            client, addr = server.accept()
            client.setblocking(False)
            epoll.register(client.fileno(), select.EPOLLIN)
            connections[client.fileno()] = client
        elif event & select.EPOLLIN:
            client = connections[fd]
            data = client.recv(4096)
            if not data:
                epoll.unregister(fd)
                client.close()
                del connections[fd]
            else:
                client.sendall(data)
        elif event & select.EPOLLHUP:
            epoll.unregister(fd)
            connections[fd].close()
            del connections[fd]

epoll O(1) на event — возвращает только готовые sockets. Масштабируется до миллионов одновременных connections (физический лимит — memory).

Nginx, HAProxy, Redis, Node.js (libuv) — все используют epoll/kqueue под капотом. Это основа high-performance servers.

Async I/O в Python: asyncio

Низкоуровневый epoll сложен. asyncio — библиотека Python, которая прячет epoll за elegant high-level API.

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"New client: {addr}")

    try:
        while True:
            data = await reader.read(4096)
            if not data:
                break
            writer.write(data)
            await writer.drain()
    finally:
        writer.close()
        await writer.wait_closed()
        print(f"Closed {addr}")

async def main():
    server = await asyncio.start_server(handle_client, '0.0.0.0', 9999)
    print("Listening on 0.0.0.0:9999")
    async with server:
        await server.serve_forever()

asyncio.run(main())

Гораздо проще, чем epoll руками. Под капотом всё то же — single thread, event loop, epoll/kqueue. Но писать как обычный sequential code с await.

Плюсы asyncio:

1. Масштабируется до тысяч-десятков тысяч одновременных connections. Один thread, низкий memory overhead.
2. Простой код. await вместо callbacks.
3. Стандартная библиотека в Python.

Минусы asyncio:

1. Нужно специальные async-aware libraries. requests не работает — нужен aiohttp. psycopg2 не работает — asyncpg. Каждая lib должна быть async.
2. CPU-bound задачи блокируют event loop. Если в handler делаешь sleep или computation — остальные клиенты ждут. Решение — loop.run_in_executor() для blocking calls.
3. Сложнее debug. Stack traces длинные через async machinery.

Сравнение моделей

Какая модель когда:

Sequential — учебные примеры. Никогда в production.

Thread per client — проще писать, работает для сотен. Если вы строите internal admin tool с десятками клиентов — OK.

Thread pool — бизнес-приложения с известной нагрузкой. Java/Python web-сервисы с N workers.

asyncio / epoll — если ваша нагрузка I/O-bound с тысячами connections. APIs, gateways, proxies, chat-серверы. Web-фреймворки FastAPI, aiohttp.

Hybrid (multi-process + async) — production high-traffic. Nginx запускает N worker processes (один на CPU core), каждый имеет async event loop. Lazy load balancing между ними через accept’у на shared listening socket.

Go’s goroutines — особый случай

Go использует goroutines — lightweight threads с 8KB initial stack (vs 1MB OS thread), которые managed Go runtime, не ядром. Goroutine vs OS thread: ratio M:N — million goroutines может map на десятки OS threads.

С точки зрения programmer goroutine выглядит как thread — блокирующие операции работают как обычно. Но runtime under the hood делает async I/O (через epoll/kqueue). Это лучшее из обоих миров: простой код + высокая performance.

В Python asyncio — ближайший эквивалент, но требует await. В Java — Project Loom (virtual threads, Java 21+) — то же самое.

Реальный пример: production-grade async chat server

import asyncio

clients = set()

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"[+] Client connected: {addr}")
    clients.add(writer)

    # Уведомить всех о новом клиенте
    msg = f"[server] {addr} joined\n".encode()
    await broadcast(msg, exclude=writer)

    try:
        while True:
            data = await reader.readline()
            if not data:
                break
            msg = f"[{addr}] {data.decode().strip()}\n".encode()
            await broadcast(msg, exclude=writer)
    except ConnectionError:
        pass
    finally:
        clients.remove(writer)
        writer.close()
        await writer.wait_closed()
        await broadcast(f"[server] {addr} left\n".encode())
        print(f"[-] Client disconnected: {addr}")

async def broadcast(message, exclude=None):
    """Послать всем клиентам, кроме отправителя."""
    for writer in clients:
        if writer is exclude:
            continue
        try:
            writer.write(message)
        except Exception:
            pass

    # Wait for all writes to complete
    await asyncio.gather(
        *(w.drain() for w in clients if w is not exclude),
        return_exceptions=True
    )

async def main():
    server = await asyncio.start_server(handle_client, '0.0.0.0', 9999)
    print("Chat server on 0.0.0.0:9999")
    async with server:
        await server.serve_forever()

asyncio.run(main())

50 строк — работающий многопользовательский chat-сервер на тысячи одновременных клиентов. Без threads, без epoll-рук. asyncio elegant.

Попробуй сам

# 1. Sequential server: подключиться несколькими nc одновременно
# Терминал 1: запустить sequential server (Урок 2)
# Терминалы 2-4: nc localhost 9999 -- только один из них работает в момент

# 2. Thread-based server: то же, но теперь все работают
# Замените цикл на threading.Thread(target=handle_client) version

# 3. asyncio chat server -- запустить пример выше
python3 chat_server.py
# Терминалы 2-4:
nc localhost 9999
# Введите имя, пишите сообщения -- видны во всех окнах

# 4. Бенчмарк: ab (ApacheBench) или wrk
# Запустить простой HTTP-сервер
python3 -m http.server 8000

# Apache Bench: 1000 запросов, 100 concurrent
ab -n 1000 -c 100 http://localhost:8000/

# wrk -- более современный
# wrk -t 4 -c 100 -d 30s http://localhost:8000/

# 5. Сравнить производительность sync vs async HTTP-сервер
# Установить aiohttp: pip install aiohttp

# Sync (Flask или http.server):
python3 -m http.server 8000  # single threaded

# Async (aiohttp):
python3 -c "
from aiohttp import web

async def hello(request):
    return web.Response(text='Hello')

app = web.Application()
app.router.add_get('/', hello)
web.run_app(app, port=8001)
"

# Сравнить wrk:
# wrk -t 4 -c 100 -d 10s http://localhost:8000/  -- sync
# wrk -t 4 -c 100 -d 10s http://localhost:8001/  -- async
# Видим разницу в throughput

# 6. Посмотреть, сколько threads используют известные серверы
ps -eLf | grep nginx | head
ps -eLf | grep redis | head
# Видим количество threads

Что вы должны вынести

1. Concurrency модели: sequential -> process -> thread -> thread pool -> epoll/async.
2. Каждая модель имеет свой sweet spot по количеству клиентов и типу workload.
3. Thread per client — для сотен клиентов, простой код.
4. epoll/asyncio — для тысяч+, I/O-bound, требует async-aware код.
5. Производство часто: hybrid — multi-process для CPU cores + async внутри.
6. Go goroutines / Java virtual threads — M:N модели, скрывают async от programmer.
7. Python asyncio — современный default для concurrent network servers.