141. Неблокирующий ввод-вывод: select, poll, epoll, дескрипторы событий

Сервер из раздела 78 обслуживал одно соединение за раз: read ждёт данных, и пока один клиент молчит, остальные стоят. Поток на соединение (раздел 139) решает это до нескольких сотен клиентов, а дальше упирается в память под стеки и в переключения. Третий путь — один поток и ядро, которое говорит, какие дескрипторы готовы: неблокирующие сокеты и мультиплексирование. На нём стоят nginx, Redis и Node.js.

Здесь один эхо-сервер написан трижды — на select, poll и epoll — и прогнан под нагрузкой из 500 и 2000 соединений (_verify/sys141/, stm-sys:1); дальше — таймер, событие и сигнал как дескрипторы, и что стоит за словами «реактор», io_uring и «корутины».

Как это называется в документации

Оригинал	Перевод
blocking, non-blocking I/O	блокирующий, неблокирующий ввод-вывод
EAGAIN / EWOULDBLOCK	«сейчас нечего» — повторить позже
I/O multiplexing	мультиплексирование: ждать многих дескрипторов сразу
readiness notification	уведомление о готовности
level-triggered, edge-triggered	по уровню, по фронту
FD_SETSIZE	предел select — 1024 дескриптора
interest list, ready list	список наблюдаемых и список готовых (epoll)
timerfd, eventfd, signalfd	таймер, событие, сигнал в виде дескриптора
C10K problem	задача десяти тысяч соединений
reactor pattern	реактор: цикл событий с обработчиками
completion-based I/O	ввод-вывод по завершению (io_uring, IOCP)
coroutine	сопрограмма

141.1 Неблокирующий режим

По умолчанию read, write, accept и connect ждут. Флаг O_NONBLOCK меняет договор: если сделать нечего, вызов возвращает −1 с errno == EAGAIN сразу. Исключение — connect: он возвращает EINPROGRESS, а итог соединения узнают, когда сокет станет готов к записи, через getsockopt(fd, SOL_SOCKET, SO_ERROR, …):

static int set_nonblocking(int fd)
{
    int fl = fcntl(fd, F_GETFL, 0);
    return fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

неблокирующий read без данных: -1, errno Resource temporarily unavailable

Сам по себе неблокирующий режим бесполезен — крутиться в цикле, опрашивая сокеты, значит жечь процессор. Нужен вызов, который уснёт и проснётся, когда хоть один дескриптор станет готов.

141.2 Три вызова, один сервер

Сервер делает одно и то же независимо от способа ожидания: слушающий сокет в наборе; готов он — принять всех, кого можно, и добавить в набор; готово соединение — прочитать всё до EAGAIN и вернуть:

static int serve(int fd)
{
    char buf[4096];
    for (;;) {
        ssize_t n = read(fd, buf, sizeof buf);
        if (n < 0) return errno == EAGAIN ? 1 : 0;      /* EAGAIN: данных больше нет — не ошибка */
        if (n == 0) return 0;
        if (write(fd, buf, n) < 0) return 0;
    }
}

Это упрощение: на неблокирующем сокете write может записать часть данных или вернуть EAGAIN, когда буфер отправки полон. Настоящий сервер хранит неотправленный хвост в буфере соединения, подписывается на EPOLLOUT и дописывает, когда сокет снова готов к записи.

select — старейший. Набор дескрипторов — битовая маска fd_set на 1024 бита (128 байт), и это предел: дескриптор с номером 1024 и выше в неё не помещается. Вызов портит набор, поэтому его копируют перед каждым вызовом, а после — перебирают все номера до maxfd:

        fd_set rd = all;                                 /* select портит набор — копия на каждый вызов */
        select(maxfd + 1, &rd, NULL, NULL, NULL);
        for (int fd = 0; fd <= maxfd; fd++) {            /* перебор всех номеров до maxfd */

poll — массив struct pollfd любой длины, без предела 1024, но каждый вызов передаёт в ядро весь массив и получает обратно весь массив с флагами revents; работа растёт с числом соединений, а не с числом готовых.

epoll — ядро хранит список наблюдаемых у себя; epoll_ctl добавляет и удаляет по одному, epoll_wait возвращает только готовые:

        int n = epoll_wait(ep, evs, 64, -1);             /* ядро отдаёт только готовые дескрипторы */

Нагрузка — клиент открывает соединения и по каждому гоняет сообщения туда-обратно:

На 500 соединениях все трое справляются; разница в скорости — от перебора: select и poll на каждое сообщение проходят весь набор. На 2000 соединений select ломается — сервер обязан закрыть всё, что не влезло в fd_set, и клиент получает эхо только по половине. poll справляется, но вчетверо медленнее epoll: 2000 структур в ядро и обратно на каждый из 4000 обменов.

141.3 epoll: по уровню и по фронту

epoll_wait по умолчанию срабатывает по уровню: пока в сокете есть непрочитанные данные, он будет возвращать этот дескриптор снова и снова. Флаг EPOLLET — по фронту: одно уведомление на появление данных, и если прочитать не всё, следующее придёт только с новыми данными. Клиент, который ждёт ответа и молчит, их не пришлёт — хвост так и останется в сокете. Отсюда правило для режима по фронту: сокет неблокирующий, читать до EAGAIN — как в serve выше. Сервер в опыте принимает соединения accept4(…, SOCK_NONBLOCK) и ставит EPOLLIN | EPOLLET:

                    struct epoll_event ce = { .events = EPOLLIN | EPOLLET, .data.fd = c };   /* по фронту */
                    epoll_ctl(ep, EPOLL_CTL_ADD, c, &ce);

Поле data — что угодно: номер дескриптора или указатель на структуру соединения с буферами и состоянием; ядро вернёт его как есть. EPOLLRDHUP сообщает, что клиент закрыл свою сторону; EPOLLONESHOT — одно срабатывание, потом дескриптор надо перевзвести (для пула потоков).

epoll есть только в Linux; в BSD и macOS его роль играет kqueue, в Windows — порты завершения (IOCP). Библиотеки libevent, libuv (ядро Node.js) и libev прячут эту разницу за одним циклом событий и добавляют таймеры, DNS и HTTP.

141.4 Всё — дескриптор

В одном poll рядом с сокетами удобно ждать и другое, если оно тоже дескриптор. Linux даёт три таких:

• timerfd — таймер: read возвращает число сработавших периодов;
• eventfd — счётчик для сигнала между потоками или процессами: write прибавляет, read забирает и обнуляет;
• signalfd — сигналы приходят как записи signalfd_siginfo, а не в обработчик; сигнал при этом блокируют маской, иначе он уйдёт по старому пути.

    struct pollfd p[3] = { { tfd, POLLIN, 0 }, { efd, POLLIN, 0 }, { sfd, POLLIN, 0 } };

таймер: тик 1
eventfd: 1 от другого потока
таймер: тик 2
signalfd: сигнал 10
таймер: тик 3

Так главный цикл сервера получает таймеры, сигналы SIGTERM и сообщения от рабочих потоков тем же путём, что и данные из сокетов, — без гонок между обработчиком сигнала и циклом (раздел 136).

141.5 Реактор, C10K и io_uring

Цикл «жди готовности → вызови обработчик» называется реактором; так устроены nginx (по одному процессу-реактору на ядро), Redis (один поток) и Node.js. Десять тысяч одновременных соединений — задача, на которой в конце девяностых ломались серверы с потоком или процессом на клиента; epoll и kqueue появились как ответ на неё. Сегодня та же схема держит миллион соединений: расходы — только память на буферы.

epoll сообщает о готовности, а читать и писать программа должна сама. io_uring меняет модель: программа кладёт запросы (прочитать столько-то байт в этот буфер) в кольцо в общей с ядром памяти, ядро выполняет и кладёт результаты в другое кольцо — без системного вызова на каждую операцию. Это ввод-вывод по завершению, как IOCP в Windows, и он покрывает и файлы, для которых epoll бесполезен (обычный файл всегда «готов»). Старый POSIX AIO в glibc — потоки с pread внутри, и выигрыша не даёт.

141.6 Сопрограммы: последовательный код на цикле событий

Обработчики реактора рвут логику соединения на куски между событиями. Сопрограмма прячет это: функция «ждёт данных» и приостанавливается, а цикл событий возобновляет её, когда данные пришли; код читается как последовательный. В C сопрограмм в языке нет, их делают библиотеками: переключение контекста через makecontext/swapcontext (libtask) или короткой вставкой на ассемблере (libco), макросы на switch по номеру строки (приём Даффа — без своего стека), или готовые libmill/libdill в стиле Go. Каждая сопрограмма — свой стек в куче (обычно 8–64 КБ) вместо мегабайт у потока. Переключение на ассемблере обходится без ядра — единицы наносекунд; swapcontext из glibc при каждом переключении ещё и сохраняет маску сигналов системным вызовом и выходит в разы медленнее — в опыте больше сотни наносекунд. Обратная сторона: одна сопрограмма, ушедшая в долгий расчёт, останавливает все.

141.7 Грабли

141.8 Итоги

• O_NONBLOCK: вместо ожидания — EAGAIN; читать до него.
• select — 1024 дескриптора и порча набора; poll — без предела, но весь массив в ядро; epoll — только готовые: 154 мс против 698 у poll на 2000 соединений.
• EPOLLET требует дочитывать до EAGAIN; data — свой указатель.
• timerfd, eventfd, signalfd — таймер, событие и сигнал в одном poll.
• Реактор — один поток на ядро; io_uring — по завершению; сопрограммы — последовательный код поверх цикла событий.

Полные исходники

echo_epoll.c — эхо-сервер на epoll, select или poll.

/* Раздел 141: эхо-сервер на epoll — один поток обслуживает любое число соединений.
   Ключ -s — select, -p — poll, по умолчанию epoll. Порт — первый аргумент. */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <unistd.h>
#include <poll.h>
#include <sys/select.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>

static int set_nonblocking(int fd)
{
    int fl = fcntl(fd, F_GETFL, 0);
    return fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

static int listen_on(int port)
{
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    int one = 1; setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
    struct sockaddr_in a = { .sin_family = AF_INET, .sin_port = htons(port), .sin_addr.s_addr = htonl(INADDR_ANY) };
    if (bind(fd, (struct sockaddr *)&a, sizeof a) < 0 || listen(fd, 128) < 0) { perror("bind/listen"); exit(1); }
    set_nonblocking(fd);
    return fd;
}

/* обслужить готовый дескриптор: читать всё, что есть, и вернуть; 0 — соединение закрыто */
static int serve(int fd)
{
    char buf[4096];
    for (;;) {
        ssize_t n = read(fd, buf, sizeof buf);
        if (n < 0) return errno == EAGAIN ? 1 : 0;      /* EAGAIN: данных больше нет — не ошибка */
        if (n == 0) return 0;
        if (write(fd, buf, n) < 0) return 0;
    }
}

static void run_epoll(int lfd)
{
    int ep = epoll_create1(0);
    struct epoll_event ev = { .events = EPOLLIN, .data.fd = lfd }, evs[64];
    epoll_ctl(ep, EPOLL_CTL_ADD, lfd, &ev);
    for (;;) {
        int n = epoll_wait(ep, evs, 64, -1);             /* ядро отдаёт только готовые дескрипторы */
        for (int i = 0; i < n; i++) {
            int fd = evs[i].data.fd;
            if (fd == lfd) {
                int c;
                while ((c = accept4(lfd, NULL, NULL, SOCK_NONBLOCK)) >= 0) {
                    struct epoll_event ce = { .events = EPOLLIN | EPOLLET, .data.fd = c };   /* по фронту */
                    epoll_ctl(ep, EPOLL_CTL_ADD, c, &ce);
                }
            } else if (!serve(fd)) {
                epoll_ctl(ep, EPOLL_CTL_DEL, fd, NULL); close(fd);
            }
        }
    }
}

static void run_poll(int lfd)
{
    static struct pollfd fds[8192]; int n = 1;      /* poll: массив любого размера, но весь идёт в ядро */
    fds[0] = (struct pollfd){ lfd, POLLIN, 0 };
    for (;;) {
        poll(fds, n, -1);                                /* каждый вызов — весь массив в ядро и обратно */
        for (int i = 0; i < n; i++) {
            if (!(fds[i].revents & (POLLIN | POLLHUP))) continue;
            if (fds[i].fd == lfd) {
                int c;
                while ((c = accept4(lfd, NULL, NULL, SOCK_NONBLOCK)) >= 0 && n < 8192)
                    fds[n++] = (struct pollfd){ c, POLLIN, 0 };
            } else if (!serve(fds[i].fd)) {
                close(fds[i].fd); fds[i] = fds[--n]; i--;
            }
        }
    }
}

static void run_select(int lfd)
{
    fd_set all; FD_ZERO(&all); FD_SET(lfd, &all); int maxfd = lfd;
    for (;;) {
        fd_set rd = all;                                 /* select портит набор — копия на каждый вызов */
        select(maxfd + 1, &rd, NULL, NULL, NULL);
        for (int fd = 0; fd <= maxfd; fd++) {            /* перебор всех номеров до maxfd */
            if (!FD_ISSET(fd, &rd)) continue;
            if (fd == lfd) {
                int c;
                while ((c = accept4(lfd, NULL, NULL, SOCK_NONBLOCK)) >= 0) {
                    if (c >= FD_SETSIZE) { close(c); continue; }     /* select не умеет больше 1024 */
                    FD_SET(c, &all); if (c > maxfd) maxfd = c;
                }
            } else if (!serve(fd)) { close(fd); FD_CLR(fd, &all); }
        }
    }
}

int main(int argc, char **argv)
{
    signal(SIGPIPE, SIG_IGN);
    int lfd = listen_on(argc > 1 ? atoi(argv[1]) : 7000);
    const char *mode = argc > 2 ? argv[2] : "-e";
    if (!strcmp(mode, "-s")) run_select(lfd);
    else if (!strcmp(mode, "-p")) run_poll(lfd);
    else run_epoll(lfd);
    return 0;
}

Дальше: раздел 142 — веб-сервисы на C: HTTP, CGI, FastCGI, sendfile.