utun2/lib/tcp_io_doc.md

tcp_io — Управление TCP-соединением на uasync + ll_queue

1. Назначение

Асинхронный TCP-коннектор, построенный на event loop uasync и очередях ll_queue одного потока. Одна структура struct tcp_conn = одно TCP-соединение. Используется в SOCKS/HTTP-прокси, STCP, TCP-прокси и тестах.

Ключевые свойства:

• Чтение, запись, ошибки и connect — через uasync (epoll/kqueue/poll).
• Два независимых ll_queue: read_queue (входящие данные) и write_queue (исходящие).
• Два memory pool: entry_pool (struct ll_entry) и data_pool (буферы), всё выделение на горячем пути через пулы.
• Backpressure на чтение: high_water/low_water + queue_waiter_wait, EPOLLIN вкл/выкл по уровню заполнения очереди.
• Backpressure на запись: порог 32 записи в write_queue + queue_waiter_wait, EPOLLOUT вкл/выкл при частичной отправке.
• Graceful shutdown: FIN (shutdown SHUT_WR) и CLOSE (close сокета) — сентинелы в write_queue; сначала предшествующие данные передаются ОС.

2. Как пользоваться

Создание соединения

struct tcp_conn* tc = tcp_conn_create(
    ua, sock,
    1500,    // entry_data_size — размер буфера для одного recv
    8192,    // write_chunk_size — размер буфера для одной send-операции
    32,      // read_high_water — порог приостановки EPOLLIN
    8,       // read_low_water  — порог возобновления EPOLLIN
    0,       // rcvbuf_size — размер буфера сокета (0 = не менять)
    on_fin_cb, on_error_cb, arg);

После создания нужно установить callback на read_queue. Отложенный первый вызов и уведомления backpressure настраиваются отдельно:

queue_set_callback(tc->read_queue, on_read_cb, my_conn);
queue_set_callback_defer(tc->read_queue, 1);  // необязательно: первый consumer callback через call_soon
queue_set_waiter_defer(tc->read_queue, 1);

Чтение данных (в on_read_cb)

static void on_read_cb(struct ll_queue* q, void* arg) {
    struct my_conn* c = (struct my_conn*)arg;
    struct ll_entry* e = queue_data_get(q);
    if (!e) { queue_resume_callback(q); return; }
    // обработать e->dgram (e->len байт)
    memory_pool_free(c->tc->data_pool, e->dgram);
    queue_entry_free(e);
    queue_resume_callback(q);  // обязательно!
}

Запись данных

Пользователь напрямую кладёт entry в write_queue. Первый callback может выполниться синхронно из queue_data_put; продолжение после queue_resume_callback откладывается через uasync_call_soon:

struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool);
uint8_t* buf = memory_pool_alloc(tc->data_pool);
memcpy(buf, data, len);
e->dgram = buf; e->len = (uint16_t)len;
queue_data_put(tc->write_queue, e);

При заполнении (порог 32) нужно ждать через queue_waiter_wait.

Graceful shutdown

tcp_conn_push_fin(tc);    // shutdown(SHUT_WR) после отправки всех данных → on_fin_sent
tcp_conn_push_close(tc);  // close сокета после отправки всех данных → on_closed

Можно вызывать из callback-ов (on_fin, on_read_cb и т.д.). push_fin возвращает -1 как при уже поставленном FIN (fin_pending=1), так и после успешного shutdown (fin_local=1). push_close возвращает -1 после фактического закрытия (closed=1); до обработки CLOSE повторная постановка не блокируется. При ошибке shutdown вызывается on_error, а on_fin_sent не вызывается. Успешный push означает принятие сентинела в очередь; связанный callback может выполниться ещё до возврата из push.

Уничтожение

void tcp_conn_destroy(struct tcp_conn* tc);

Сразу отменяет connect-таймер и callbacks очередей/сокета, дренирует очереди и закрывает сокет. Пулы и сам tc освобождаются через uasync_call_soon в том же потоке. После destroy владелец не использует указатель. Повторный вызов безопасен только до выполнения отложенного free; NULL безопасен всегда. Перед остановкой/уничтожением uasync нужно выполнить отложенную очистку (uasync_drain_immediate вне текущих callbacks).

Обработка ошибок

on_error(tc, err, arg) вызывается после закрытия сокета. Реализация on_error обязана вызвать tcp_conn_destroy. После возврата из on_error tc недействителен.

static void on_error_cb(struct tcp_conn* tc, int err, void* arg) {
    struct my_conn* c = (struct my_conn*)arg;
    // почистить свои ресурсы
    tcp_conn_destroy(tc);   // tc после этого использовать нельзя
    socket_close_wrapper(other_sock);
    u_free(c);
}

on_flushed (одноразовый)

tcp_conn_set_flushed(tc, my_flushed_cb);
// коллбэк вызовется один раз когда write_queue + write_buf полностью опустеют; сбрасывается перед вызовом

Приостановка чтения

tcp_conn_pause_read(tc);  // убирает EPOLLIN и отменяет read_waiter

Connect-таймаут

tcp_conn_set_connect_timeout(tc, timeout_ms) заменяет интервал ожидания, начиная с текущего вызова. Значение <=0 отменяет прежний таймер; положительное значение ограничено INT_MAX/10 мс из-за единиц uasync (0.1 мс). Для уже установленного соединения настройка сохраняется без создания таймера. Возврат: 0 — применено, -EINVAL — аргументы/состояние, -ENOMEM — создание таймера, -EIO — нарушение состояния отмены. При ошибке прежние настройка и handle сохраняются. Новый таймер резервируется перед отменой старого; операция выполняется целиком в uasync-потоке без вызова пользовательских callbacks. Вызывающий код проверяет результат. Истечение вызывает on_error(ETIMEDOUT); завершение connect, фатальная ошибка и destroy отменяют таймер.

Ограничения и многопоточность

• Однопоточный. Всё работает в одном uasync event loop. Нельзя вызывать из другого потока.
• FIN-сентинел в очереди. После успешного push_fin новые данные в write_queue не помещают; предшествующие данные передаются ОС до shutdown.
• Нельзя вызывать close сокета напрямую. Только через tcp_conn_push_close — иначе нарушится порядок обработки очереди.
• Можно вызывать push_fin/push_close из callback-ов. FIN защищён от повторной постановки; CLOSE проверяет только уже выполненное закрытие. Владелец учитывает возможный синхронный callback.

3. API

Структуры

Поле	Назначение
struct tcp_conn	Всё состояние TCP-соединения: сокет, очереди, пулы, коллбэки, флаги

Жизненный цикл

Функция	Описание
tcp_conn_create(ua, sock, entry_data_size, write_chunk_size, read_high_water, read_low_water, rcvbuf_size, on_fin, on_error, arg)	Создать соединение: аллоцирует tc, два memory pool, два ll_queue, регистрирует сокет в uasync с EPOLLIN+EPOLLOUT. Вовращает NULL при ошибке
tcp_conn_destroy(tc)	Сразу отменяет таймер/callbacks, дренирует очереди и закрывает сокет; пулы и tc освобождает через uasync_call_soon

Коллбэки (поля struct tcp_conn)

Поле	Когда вызывается
on_fin(tc, arg)	FIN получен от удалённой стороны (recv==0): сразу если read_queue пуста, иначе через queue_set_empty_callback после дренажа
on_fin_sent(tc, arg)	FIN отправлен (shutdown(SHUT_WR)) — после отправки всех данных перед FIN-сентинелом
on_error(tc, err, arg)	Фатальная ошибка сокета. Вызывается после close сокета. Обязан вызвать tcp_conn_destroy. После возврата tc недействителен
on_flushed(tc, arg)	write_queue + write_buf полностью опустели. Одноразовый — перед вызовом сбрасывается в NULL. Установка: tcp_conn_set_flushed()
on_closed(tc, arg)	Сокет закрыт через CLOSE-сентинел (tcp_conn_push_close) после отправки всех предшествующих данных

Graceful shutdown

Функция	Описание
tcp_conn_push_fin(tc)	Помещает FIN-сентинел (dgram=&tcp_fin_sentinel, len=0) в write_queue. После отправки предшествующих данных: shutdown(SHUT_WR), fin_local=1, вызывает on_fin_sent. Повторный вызов → -1
tcp_conn_push_close(tc)	Помещает CLOSE-сентинел (dgram=NULL, len=0) в write_queue. После отправки предшествующих данных: close сокета, удаление из uasync, closed=1, вызывает on_closed. После закрытия → -1

Прочее

Функция	Описание
tcp_conn_set_flushed(tc, cb)	Установить одноразовый коллбэк на полное опустошение write_queue
tcp_conn_pause_read(tc)	Принудительно выключить EPOLLIN и отменить ожидание низкого порога
tcp_conn_set_connect_timeout(tc, ms)	Заменить/отключить connect-таймер; возвращает 0 или отрицательный код ошибки