diff --git a/lib/tcp_io.h b/lib/tcp_io.h index f26cb9d3..4142d082 100644 --- a/lib/tcp_io.h +++ b/lib/tcp_io.h @@ -19,6 +19,14 @@ // // Ошибка: любая ошибка сокета → tcp_conn_handle_error → close + null коллбэков → on_error. // Реализация on_error обязана вызвать tcp_conn_destroy. После возврата tc недействителен. +// +// Все операции и callbacks — в потоке ua. После create установить consumer на read_queue: +// queue_set_callback(tc->read_queue, on_read, arg). +// Consumer извлекает entry, обрабатывает e->len байт, возвращает dgram в tc->data_pool, +// освобождает entry и вызывает queue_resume_callback(q) для продолжения чтения. +// Для записи выделить entry из tc->entry_pool и dgram из tc->data_pool, заполнить len и dgram_pool; +// затем queue_data_put(tc->write_queue, e). Длина должна помещаться в буфер и uint16_t. +// После принятого FIN новые данные не добавлять; закрывать сокет через push_close/destroy. #ifndef TCP_IO_H #define TCP_IO_H @@ -41,10 +49,10 @@ struct tcp_conn { struct ll_queue* read_queue; // сокет → данные (блоки до entry_data_size) struct ll_queue* write_queue; // данные → сокет (блоки до write_chunk_size) - int read_high_water; - int read_low_water; - int rcvbuf_size; - uint8_t read_paused; + int read_high_water; // Число входных entry для приостановки чтения + int read_low_water; // Порог очереди для возобновления чтения + int rcvbuf_size; // Запрошенный SO_RCVBUF; <=0 — не настраивать буферы + uint8_t read_paused; // Чтение отключено backpressure или pause_read uint8_t write_monitor; // 1 = EPOLLOUT активен uint8_t connected; @@ -73,14 +81,21 @@ struct tcp_conn { // Коллбэки void (*on_connected)(struct tcp_conn* tc, void* arg); // исходящий connect завершён - void (*on_fin)(struct tcp_conn* tc, void* arg); // FIN получен от удалённой стороны (fin_remote=1) + void (*on_fin)(struct tcp_conn* tc, void* arg); // FIN получен; вызов сразу при пустой read_queue, иначе после её дренажа void (*on_fin_sent)(struct tcp_conn* tc, void* arg); // FIN отправлен удалённой стороне (fin_local=1) void (*on_error)(struct tcp_conn* tc, int err, void* arg); // фатальная ошибка (вызывается после close сокета; обязан вызвать tcp_conn_destroy) void (*on_flushed)(struct tcp_conn* tc, void* arg); // все данные записи отправлены (write_queue + write_buf пусты) void (*on_closed)(struct tcp_conn* tc, void* arg); // сокет закрыт через очередь (после close) - void* arg; + void* arg; // Заимствованный контекст всех callbacks, модуль его не освобождает }; +// Регистрирует подключённый/подключающийся неблокирующий sock в ua; сам connect не вызывает. +// entry_data_size — ёмкость recv-блока, write_chunk_size — размер буфера записи; оба 1..UINT16_MAX, +// max этих размеров должен вмещать void* (требование пула). Пороги: 0 <= low < high, проверяет вызывающий. +// rcvbuf_size > 0 пробует SO_SNDBUF=0/SO_RCVBUF=rcvbuf_size; ошибка опции логируется, create продолжается. +// Создаёт две очереди и два пула; on_fin/on_error и arg сохраняются без передачи владения arg. +// Результат — tc или NULL при неверных ua/sock, OOM/ошибке регистрации; при NULL sock остаётся вызывающему. +// При успехе сокетом владеет tc. Пулы и выданные из них блоки живут до отложенного освобождения tc. struct tcp_conn* tcp_conn_create( struct UASYNC* ua, socket_t sock, @@ -97,11 +112,13 @@ struct tcp_conn* tcp_conn_create( // Пулы и tc освобождаются через uasync_call_soon в том же uasync-потоке. // NULL безопасен; повторный destroy допустим только до отложенного free. // Владелец после destroy больше не использует tc; event loop должен выполнить отложенную очистку. +// Ранее извлечённые entry/dgram вернуть до освобождения пулов; перед destroy ua дренировать call_soon вне callbacks. void tcp_conn_destroy(struct tcp_conn* tc); // Запись: queue_data_put(tc->write_queue, entry) напрямую. // Первый автозабор может быть синхронным; queue_resume_callback откладывает продолжение. -// Порог: 32 записи, при заполнении queue_waiter_wait. +// Порог backpressure: count <= 32, не жёсткий лимит ёмкости; производитель ждёт через queue_waiter_wait. +// Обычные буферы записи должны принадлежать data_pool: отправка возвращает их непосредственно в этот пул. // Поставить сигналы в очередь отправки. Все данные в очереди перед сигналом // будут переданы ОС до его обработки. Сигналы имеют len=0: dgram=NULL → CLOSE, @@ -119,6 +136,7 @@ int tcp_conn_push_close(struct tcp_conn* tc); // Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели. // Сбрасывается перед вызовом; callback может установить новое уведомление. +// Setter только заменяет функцию (NULL снимает); для уже пустой очереди сам вызов не планирует. void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg)); // Принудительная остановка чтения: убирает EPOLLIN и отменяет read_waiter. diff --git a/lib/tcp_io_doc.md b/lib/tcp_io_doc.md deleted file mode 100644 index 1526ddd9..00000000 --- a/lib/tcp_io_doc.md +++ /dev/null @@ -1,177 +0,0 @@ -# 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. Как пользоваться - -### Создание соединения - -```c -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 настраиваются отдельно: - -```c -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) - -```c -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`: - -```c -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 - -```c -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. - -### Уничтожение - -```c -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` недействителен. - -```c -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 (одноразовый) - -```c -tcp_conn_set_flushed(tc, my_flushed_cb); -// коллбэк вызовется один раз когда write_queue + write_buf полностью опустеют; сбрасывается перед вызовом -``` - -### Приостановка чтения - -```c -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 или отрицательный код ошибки |