Всем привет! Наверное, многие уже слышали о новом интерфейсе ядра Linux — io_uring. Это новый способ работы с асинхронным I/O (и не только) в Linux. Кстати, новый он не только из-за даты выхода в свет, но и в плане подходов, которые предлагает разработчику.

Заинтересовало? Более подробно разберемся под катом.

Дисклеймер

Это первая статья из серии посвященной io_uring. Данный материал — вводный, поэтому основной упор будет сделан на основы работы с io_uring и примеры программ с комментариями. 

В этой статье я буду только вскользь касаться темы специфических настроек и опций io_uring. Также сегодня не будет практических примеров применения этой технологии. Но не беспокойтесь, эти темы будут освещены будущих публикациях.

Кстати, если вас смутило нахождение статьи в хабе GO — причина будет в конце публикации. 

Долгожданные гости

IO_URING это новый интерфейс ядра Linux для асинхронного ввода/вывода, разработанный Jens Axboe. Доступен для использования с версии ядра 5.1 (но замечу, что примеры статьи проверялись в версии 5.11 и точно не будут работать в версиях до 5.5).

Тень прошлого

И прежде чем мы действительно разберемся, что это за интерфейс, предлагаю немного освежить память и вспомнить инструменты Linux для асинхронного программирования:

• select, poll, epoll — вообще говоря, эти семейства системных вызовов не дают асинхронность как таковую, но позволяют следить за набором файловых дескрипторов и реагировать на готовность определенных дескрипторов к чтению/записи:

  • select — обладает крайне неудобным API, не работает с файлами и проигрывает коллегам по перформансу

  • poll — так же как и select позволяет разработчику следить за готовностью файловых дескрипторов. В отличие от select имеет более приятный API (хотя и не без огрехов, которые были устранены в epoll), не умеет в файлы

  • epoll — усовершенствованный poll, доступен только в linux, существенно улучшает перформанс предшественника, все так же не умеет работать с файлами

• AIO — семейство системных вызовов. Стоит несколько особняком, поскольку предоставляет интерфейс, который действительно похож на нечто асинхронное (ну колбеки там, javascript, вы понимаете). Правда данный инструмент имеет столько вопросов по производительности, API и внутренней реализации, что в реальности сложно найти человека, который им пользовался.

В общем, как видите, даже epoll, хоть и используется повсеместно, имеет свои ограничения.

Самая короткая дорога к асинхронности

И как уже несложно догадаться, задача io_uring — снять эти ограничения, а также дать новый интерфейс для работы с асинхронным I/O в linux. 

По своей сути io_uring - это два кольцевых буфера (отсюда и ring в названии):

1. Submission queue (далее SQ) — сюда пишем операции, которые должно выполнить ядро ОС (например: прочитать файл, принять соединение, закрыть сокет). Операция — это syscall который система выполнит в фоне, не блокируя нашу программу. Элемент SQ — submission queue entry (SQE). Ниже приведена структура, которая описывает SQE. Выглядит довольно страшно, поэтому наиболее часто используемые поля будут описаны отдельно:

   io_uring_sqe

   /*
    * IO submission data structure (Submission Queue Entry)
    */
   struct io_uring_sqe {
       __u8    opcode;         /* type of operation for this sqe */
       __u8    flags;          /* IOSQE_ flags */
       __u16   ioprio;         /* ioprio for the request */
       __s32   fd;             /* file descriptor to do IO on */
       union {
           __u64   off;            /* offset into file */
           __u64   addr2;
       };
       union {
           __u64   addr;       /* pointer to buffer or iovecs */
           __u64   splice_off_in;
       }
       __u32   len;            /* buffer size or number of iovecs */
       union {
           __kernel_rwf_t  rw_flags;
           __u32    fsync_flags;
           __u16    poll_events;   /* compatibility */
           __u32    poll32_events; /* word-reversed for BE */
           __u32    sync_range_flags;
           __u32    msg_flags;
           __u32    timeout_flags;
           __u32    accept_flags;
           __u32    cancel_flags;
           __u32    open_flags;
           __u32    statx_flags;
           __u32    fadvise_advice;
           __u32    splice_flags;
           __u32    rename_flags;
           __u32    unlink_flags;
           __u32    hardlink_flags;
       }; /* op_code flags */
       __u64    user_data;     /* data to be passed back at completion time */
       union {
       struct {
               union {
                   __u16    buf_index;
                   __u16    buf_group;
               }
           __u16    personality;
           union {
               __s32    splice_fd_in;
               __u32    file_index;
       };
       };
       __u64    __pad2[3];
       };
   };

   •  opcode — код операции, можно сказать, набор поддерживаемых io_uring системных вызовов. Но так же есть такие операции, как отмена операции или Nop операция (полезно в тестах)

   • flags — набор флагов, но не для выбранного syscall'а (операции), а для самого SQE. Например, с помощью флага IOSQE_IO_LINK гарантируется последовательное исполнение двух или более SQE

   • fd — файловый дескриптор к которому применяется операция

   • addr, len — сюда обычно помещается буфер для чтения/записи

   • op_code flags — union в котором хранятся флаги специфичные для выбранного syscall'а

   • user_data — это поле разберем чуть позже, при разборе CQE

2. Completion queue (далее CQ) - это очередь из которой вычитываются результаты. Элемент CQ - completion queue event (CQE). Структура описывающая CQE:

   io_uring_cqe

   struct io_uring_cqe {
     __u64     user_data;  /* sqe->data submission passed back */
     __s32     res;        /* result code for this event */
     __u32     flags;
   };

   • res — результат работы системного вызова. Например, количество прочитанных байт в случае ReadV или дескриптор сокета для Accept. В случае ошибки — содержит значение -errno

   • flags — пока не используется

   • user_data — концептуально важное поле. Как вы понимаете, порядок получения CQE никак не зависит от порядка, в котором добавлялись SQE (асинхронность же). Возникает вопрос, как совместить некий результат (CQE) и соответствующий ему запрос (SQE)? Ответ: используем поле user_data которое есть как у SQE, так и у CQE. Значение из поля SQE.user_data будет скопировано в результат работы этой операции — CQE.user_data

Оба буфера шарятся между ядром и userspace для избежания затрат на копирование данных. Пользователь заносит операции в tail SQ буфера, а ядро читает из head. После выполнения операции ядро положит результат в tail CQ буфера, а пользователь должен читать результаты из head:

В ходе дальнейшего изложения будем говорить о SQ и CQ просто как о двух очередях. Чтобы избежать путаницы, мы абстрагируемся от реализации этих очередей через кольцевые буфера.

Начинаем работу с io_uring

Простейший алгоритм работы с io_uring выглядит примерно так: 

1. Инициализировать инстанс io_uring.

2. Добавить в SQ операцию на выполнение (queue SQE).

3. Сообщить ядру что в SQ появились новые элементы.

4. Подождать, пока ядро выполнит операцию.

5. Извлечь из CQ результат выполнения (dequeue CQE).

Для реализации подобного алгоритма понадобится ряд системных вызовов: io_uring_setup, io_uring_enter и io_uring_register.

io_uring_setup

io_uring_setup — создает и конфигурирует экземпляр io_uring. Конфигурация io_uring это отдельная тема для разговора (которую обязательно коснемся в будущих статьях) — есть куча опций, которые могут повлиять как на поведение, так и на производительность системы (в худшую и в лучшую сторону само собой). 

Помимо самого вызова io_uring_setup, для работы необходимо замапить к себе память, которую уже выделило ядро под SQ и CQ, делается это вызовом mmap с флагом MAP_SHARED.

Пример:

// создаем инстанс io_uring, размер CQ и SQ устанавливается параметром entries,
// конфигурация в структуре io_uring_params
int io_uring_setup(unsigned entries, struct io_uring_params *p)
{
    return (int) syscall(__NR_io_uring_setup, entries, p);
}

io_uring_enter

У этого системного вызова есть две основных функции:

1. Сообщить ядру о том в SQ появились новые SQE.

2. Подождать, пока в CQ не появится n результатов выполнения операций.

Можно или ждать CQE или сабмитить SQE, а можно делать обе эти вещи в рамках одного syscall'a.

Пример:

// отправляем 3 операции на выполнение в кольцо ring_fd, возврат блокируется пока io_uring не выполнит 2 операции
syscall(__NR_io_uring_enter, ring_fd, 3, 2,  IORING_ENTER_GETEVENTS, NULL, 0);

io_uring_register

Используется для управления ресурсами связанными с io_uring. Например:

• для регистрации (обновления и дерегистрации) буферов которые будут использоваться нашим приложение и ядром совместно. Теоретически это позволит устранить некоторые копирования данных из userspace в kernel и обратно

• для регистрации (обновления и дерегистрации) набора файловых дескрипторов. Не знаю зачем это нужно, но в старых версиях ядра это требуется делать, чтобы файловый дескриптор был "рабочим" в некоторых режимах работы io_uring

• для получения probe - информации по фичам, которые поддерживает текущая версия io_uring

Пример:

// регистрируем буфера в ядре, передаем набор vectors - указателей на структуры iovec
syscall(__NR_io_uring_register, fd, IORING_REGISTER_BUFFERS, vectors, vectors_len)

Возвращаясь к нашему простейшему^tm алгоритму: естественно он может быть сильно модифицирован. Например, чтение из CQ и запись в SQ могут производиться параллельно, в разных потоках. Или можно писать в SQ не по одной операции, а сразу пачку, для уменьшения количества системных вызовов io_uring_enter. Тут уже все зависит от разработчика, как использовать эти строительные кирпичики для реализации таких концепций как, например, event loop.

В гостях у liburing

Конечно, работать напрямую с системными вызовами не только неудобно, но и не рекомендуется. Поэтому стоит использовать библиотеку liburing. Причина — устранение бойлерплейта и более приятный API. Кроме того, так как обе очереди используются и приложением, и ядром — реализации queue в SQ и dequeue из CQ должны синхронизироваться с ядром. Эти обязанности берет на себя liburing.

Рассмотрим основные функции, которые предлагает эта библиотека:

• io_uring_queue_init — создает io_uring + отображает CQ и SQ в userspace

• io_uring_get_sqe — возвращает указатель на следующее, готовое к использованию, SQE в SQ

• io_uring_prep_* (пример: io_uring_prep_writev, io_uring_prep_accept) — семейство функций, принимают на вход SQE которую конфигурируют в соответствии с выбранной операцией

• io_uring_submit — сообщает ядру о том, что в SQ появились новые SQE

• io_uring_wait_cqes — ждет, пока в CQ не появится заданное число не просмотренных CQE

• io_uring_cqe_seen — помечаем CQE как просмотренное

• io_uring_register_*— обертки над системным вызовом io_uring_register. Позволяют зарегистрировать буфера, файлы, файловые дескрипторы для поллинга, "взять пробу" и так далее

Вот с таким нехитрым набором функций нам и предлагается писать асинхронные приложения. Что же, давайте напишем что-то простое, для разминки.

Hello world

Выведем заветные 13 символов в STDOUT:

#include <liburing.h>
#include <assert.h>
#include <unistd.h>
#include <string.h>

int main() {
    struct io_uring_params params;
    struct io_uring ring;
    memset(&params, 0, sizeof(params));

    /**
     * Создаем инстанс io_uring, не используем никаких кастомных опций.
     * Емкость SQ и CQ буфера указываем как 4096 вхождений.
     */
    int ret = io_uring_queue_init_params(4, &ring, &params);
    assert(ret == 0);

    char hello[] = "hello world!\n";
    
    // Добавляем операцию write в очередь SQ.
    struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
    io_uring_prep_write(sqe, STDOUT_FILENO, hello, 13, 0);

    // Сообщаем io_uring о новых SQE в SQ.
    io_uring_submit(&ring);
    
    // Ждем пока в CQ появится новое CQE.
    struct io_uring_cqe *cqe;
    ret = io_uring_wait_cqe(&ring, &cqe);
    assert(ret == 0);

    // Проверяем отсутствие ошибок.
    assert(cqe->res > 0);

    // Dequeue из очереди CQ.
    io_uring_cqe_seen(&ring, cqe);

    io_uring_queue_exit(&ring);

    return 0;
}

Да уж, кода получилось немало. Да и где же тут асинхронность? Асинхронность заключается в том, что вывод в терминал происходит в фоне от потока приложения, в  момент после подтверждения SQE (io_uring_submit) и перед получением результата операции (io_uring_wait_cqe). Итак, сам по себе системный вызов write (pwrite если быть точным) происходит в одном из тредов ядра. Как? Я об этом не рассказывал? Исправляемся!

Туман над kernel workers

Это, наверное, наиболее "туманная" сторона io_uring. Операции, помещенные в очередь, будут выполнены в "фоне" от нашего приложения. Но кто их выполнит?

Выполнять будут потоки ядра. Для каждого экземпляра io_uring создается пул воркеров io_wqe_worker-*. Управление этим пулом скрыто от прикладного программиста (к сожалению, и в документации нет явного описания алгоритма работы, так что только сурцы и практика). 

Но, все-таки, есть рычаги для косвенного управления. Например, в недавней версии ядра появилась возможность указать максимальное количество воркеров в пуле. Кроме того, ряд опций влияет на то, как io_uring управляет пулом воркеров. 

Ну и наконец, можно использовать несколько экземпляров io_uring — таким образом, поднимая несколько пулов (хотя это поведение можно изменить, попросив несколько экземпляров io_uring работать на одном пуле). 

Зеркало трафика, пишем tcp-echo сервер

Предлагаю финализировать сегодняшнюю информацию и разобрать реализацию tcp-echo сервера написанного с использованием io_uring. Задача tcp-echo сервера — ретрансляция всех входящих данных обратно клиенту. За основу был взят код из этого проекта, слегка модифицирован и снабжен необходимыми комментариями.

#include <liburing.h>
#include <stdio.h>
#include <string.h>
#include <strings.h>
#include <assert.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <sys/socket.h>

#define MAX_CONNECTIONS 4096
#define BACKLOG 512
#define MAX_MESSAGE_LEN 2048
#define IORING_FEAT_FAST_POLL (1U << 5)

void add_accept(struct io_uring *ring, int fd, struct sockaddr *client_addr, socklen_t *client_len);

void add_socket_read(struct io_uring *ring, int fd, size_t size);

void add_socket_write(struct io_uring *ring, int fd, size_t size);

/**
 * Каждое активное соединение в нашем приложение описывается структурой conn_info.
 * fd - файловый дескриптор сокета.
 * type - описывает состояние в котором находится сокет - ждет accept, read или write.
 */
typedef struct conn_info {
    int fd;
    unsigned type;
} conn_info;

enum {
    ACCEPT,
    READ,
    WRITE,
};

// Буфер для соединений.
conn_info conns[MAX_CONNECTIONS];

// Для каждого возможного соединения инициализируем буфер для чтения/записи.
char bufs[MAX_CONNECTIONS][MAX_MESSAGE_LEN];

int main(int argc, char *argv[]) {
    /**
     * Создаем серверный сокет и начинаем прослушивать порт.
     * Обратите внимание что при создании сокета мы НЕ УСТАНАВЛИВАЕМ флаг O_NON_BLOCK,
     * но при этом все чтения и записи не будут блокировать приложение. 
     * Происходит это потому, что io_uring спокойно превращает операции над блокирующими сокетами в non-block системные вызовы.
     */
    int portno = strtol(argv[1], NULL, 10);
    struct sockaddr_in serv_addr, client_addr;
    socklen_t client_len = sizeof(client_addr);

    int sock_listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    const int val = 1;
    setsockopt(sock_listen_fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));

    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(portno);
    serv_addr.sin_addr.s_addr = INADDR_ANY;

    assert(bind(sock_listen_fd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) >= 0);
    assert(listen(sock_listen_fd, BACKLOG) >= 0);

    /**
     * Создаем инстанс io_uring, не используем никаких кастомных опций.
     * Емкость очередей SQ и CQ указываем как 4096 вхождений.
     */
    struct io_uring_params params;
    struct io_uring ring;
    memset(&params, 0, sizeof(params));

    assert(io_uring_queue_init_params(4096, &ring, &params) >= 0);

    /**
     * Проверяем наличие фичи IORING_FEAT_FAST_POLL.
     * Для нас это наиболее "перформящая" фича в данном приложении,
     * фактически это встроенный в io_uring движок для поллинга I/O.
     */
    if (!(params.features & IORING_FEAT_FAST_POLL)) {
        printf("IORING_FEAT_FAST_POLL not available in the kernel, quiting...\n");
        exit(0);
    }

    /**
     * Добавляем в SQ первую операцию - слушаем сокет сервера для приема входящих соединений.
     */
    add_accept(&ring, sock_listen_fd, (struct sockaddr *) &client_addr, &client_len);


    /*
     * event loop
     */
    while (1) {
        struct io_uring_cqe *cqe;
        int ret;

        /**
         * Сабмитим все SQE которые были добавлены на предыдущей итерации.
         */
        io_uring_submit(&ring);

        /**
         * Ждем когда в CQ буфере появится хотя бы одно CQE.
         */
        ret = io_uring_wait_cqe(&ring, &cqe);
        assert(ret == 0);

        /**
         * Положим все "готовые" CQE в буфер cqes.
         */
        struct io_uring_cqe *cqes[BACKLOG];
        int cqe_count = io_uring_peek_batch_cqe(&ring, cqes, sizeof(cqes) / sizeof(cqes[0]));

        for (int i = 0; i < cqe_count; ++i) {
            cqe = cqes[i];

            /**
             * В поле user_data мы заранее положили указатель структуру
             * в которой находится служебная информация по сокету.
             */
            struct conn_info *user_data = (struct conn_info *) io_uring_cqe_get_data(cqe);

            /**
             * Используя тип идентифицируем операцию к которой относится CQE (accept/recv/send).
             */
            unsigned type = user_data->type;
            if (type == ACCEPT) {
                int sock_conn_fd = cqe->res;

                /**
                * Если появилось новое соединение: добавляем в SQ операцию recv - читаем из клиентского сокета,
                * продолжаем слушать серверный сокет.
                */
                add_socket_read(&ring, sock_conn_fd, MAX_MESSAGE_LEN);
                add_accept(&ring, sock_listen_fd, (struct sockaddr *) &client_addr, &client_len);
            } else if (type == READ) {
                int bytes_read = cqe->res;

                /**
                 * В случае чтения из клиентского сокета:
                 * если прочитали 0 байт - закрываем сокет
                 * если чтение успешно: добавляем в SQ операцию send - пересылаем прочитанные данные обратно, на клиент.
                 */
                if (bytes_read <= 0) {
                    shutdown(user_data->fd, SHUT_RDWR);
                } else {
                    add_socket_write(&ring, user_data->fd, bytes_read);
                }
            } else if (type == WRITE) {
                /**
                * Запись в клиентский сокет окончена: добавляем в SQ операцию recv - читаем из клиентского сокета.
                */
                add_socket_read(&ring, user_data->fd, MAX_MESSAGE_LEN);
            }

            io_uring_cqe_seen(&ring, cqe);
        }
    }
}

/**
 *  Помещаем операцию accept в SQ, fd - дескриптор сокета на котором принимаем соединения.
 */
void add_accept(struct io_uring *ring, int fd, struct sockaddr *client_addr, socklen_t *client_len) {
    // Получаем указатель на первый доступный SQE.
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
    // Хелпер io_uring_prep_accept помещает в SQE операцию ACCEPT.
    io_uring_prep_accept(sqe, fd, client_addr, client_len, 0);

    // Устанавливаем состояние серверного сокета в ACCEPT.
    conn_info *conn_i = &conns[fd];
    conn_i->fd = fd;
    conn_i->type = ACCEPT;

    // Устанавливаем в поле user_data указатель на socketInfo соответствующий серверному сокету.
    io_uring_sqe_set_data(sqe, conn_i);
}

/**
 *  Помещаем операцию recv в SQ.
 */
void add_socket_read(struct io_uring *ring, int fd, size_t size) {
    // Получаем указатель на первый доступный SQE.
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
    // Хелпер io_uring_prep_recv помещает в SQE операцию RECV, чтение производится в буфер соответствующий клиентскому сокету.
    io_uring_prep_recv(sqe, fd, &bufs[fd], size, 0);

    // Устанавливаем состояние клиентского сокета в READ.
    conn_info *conn_i = &conns[fd];
    conn_i->fd = fd;
    conn_i->type = READ;

    // Устанавливаем в поле user_data указатель на socketInfo соответствующий клиентскому сокету.
    io_uring_sqe_set_data(sqe, conn_i);
}

/**
 *  Помещаем операцию send в SQ буфер.
 */
void add_socket_write(struct io_uring *ring, int fd, size_t size) {
    // Получаем указатель на первый доступный SQE.
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
    // Хелпер io_uring_prep_send помещает в SQE операцию SEND, запись производится из буфера соответствующего клиентскому сокету.
    io_uring_prep_send(sqe, fd, &bufs[fd], size, 0);

    // Устанавливаем состояние клиентского сокета в WRITE.
    conn_info *conn_i = &conns[fd];
    conn_i->fd = fd;
    conn_i->type = WRITE;

    // Устанавливаем в поле user_data указатель на socketInfo соответсвующий клиентскому сокету.
    io_uring_sqe_set_data(sqe, conn_i);
}

Производительность

Для оценки производительности будем использовать сравнение с таким же tcp-echo сервером, написанным с использованием epoll. Считать RPS будем вот этим инструментом, варьируем количество клиентских соединений (c) и объем передаваемых данных (bytes). 

Ну и характеристики стенда:

• Linux 5.11

• Intel(R) Core(TM) i5-7500 CPU @ 3.40GHz (4 ядра)

• 16gb RAM

Компилируем и запускаем приложение:

gcc tcp-echo.c -o ./tcp-echo -Wall -O2 -D_GNU_SOURCE -luring
./tcp-echo 8080

Затем бенчмарк:

cargo run --release -- --address "127.0.0.1:8080" --number {c} --duration 60 --length {bytes}

В таблице выше представлены request per second полученные в ходе тестов. Нагрузка на процессор в обоих случаях была примерно одинаковая. Можно сделать вывод — io_uring как минимум является достойным конкурентом epoll в плане производительности. 

Промежуточные итоги, а также содержание следующих статей

Данная статья является введением в io_uring. За рамками этого материала осталась гора нюансов связанных, в первую очередь, с настройками io_uring. Но, надеюсь, некоторые из них получится осветить в последующих статьях.

Важно заметить, что механизм сам по себе довольно новый, поэтому:

1. Все еще можно наткнуться на неприятные баги (особенно в "старых" версиях ядра).

2. Фичи активно добавляются.

3. Есть небезосновательные надежды на то, что в последующих версиях производительность будет еще лучше.

Ну и напоследок, наверное, стоит осветить вопрос, при чем тут вообще GO и почему будущие статьи будут касаться в том числе и этого языка?

Ну, во-первых, потому что автор GO разработчик. А во-вторых, и это наиболее важно, мы говорим об асинхронном I/O, работать с которым так удобно в GO. В основе GO-шного I/O лежит такая штука как netpoller который является частью рантайма. А что если попробовать написать свой netpoller или альтернативу ему с использованием io_uring и повоевать с рантаймом? И сделать это, например, в рамках http сервера? 

Думаю может получиться интересно, а по дороге еще раз посмотрим на внутреннее устройство некоторых механизмов GO рантайма. Stay tuned!

Немного полезных ссылок

• https://kernel.dk/io_uring.pdf — whitepaper

• https://unixism.net/loti/index.html — блог с примерами реализаций простых приложений

• https://github.com/axboe/liburing — liburing

Дата-центр ITSOFT — размещение и аренда серверов и стоек в двух дата-центрах в Москве. За последние годы UPTIME 100%. Размещение GPU-ферм и ASIC-майнеров, аренда GPU-серверов, лицензии связи, SSL-сертификаты, администрирование серверов и поддержка сайтов.