Consolidate TCP queue and resource lifecycle documentation in header
This commit is contained in:
evgeny committed
1 parent
10b039686c
commit
16f81575ce
2 files changed
+25
-184
No files matched your search
+25
-7
@@ -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.
|
||||
|
||||
@@ -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 или отрицательный код ошибки |
|
||||
Reference in new issue
Block a user