proxy: rename + simplify + backpressure

- tcp_proxy/remote_proxy → tcp_proxy_client/tcp_proxy_server
- Все proxy файлы перенесены в src/proxy/
- Конфиг: [tcp_proxy]→[tcp_proxy_client], [remote_proxy]→[tcp_proxy_server]
- Убран subcmd CONNECTED (0x02) — клиент шлёт DATA сразу после CONNECT
- Backpressure: etcp_router_waiter_register/cancel через ll_queue threshold waiter
- Клиент: recv_cb send fail → tx_buf+waiter, не tcp_recved() (окно lwIP=0)
- Сервер: read_cb send fail → pause_buf+waiter, пауза read_id
- CLOSE/ERROR отправляется только после drain всех очередей
- Fail: conn не найден/closed → send ERROR вместо CLOSE/silent drop
- PROTOCOL.md: полная архитектура с диаграммами состояний и потоков
This commit is contained in:
Evgeny committed 2026-06-02 18:02:37 +03:00
1 parent e7a9163011
commit 1f4fe37fe5
41 files changed
+1659 -659

No files matched your search

+93 -40
View File
@@ -46,8 +46,6 @@ static inline void queue_check_thread(struct ll_queue* q) {
// Предварительные объявления внутренних функций
static void queue_resume_timeout_cb(void* arg);
static void check_waiters(struct ll_queue* q);
static void add_to_hash(struct ll_queue* q, struct ll_entry* entry);
static void remove_from_hash(struct ll_queue* q, struct ll_entry* entry);
// ==================== Управление очередью ====================
@@ -60,6 +58,8 @@ struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size, uint16_t index_o
q->name = name;
q->ua = ua;
q->size_limit = -1; // Без ограничения по умолчанию
q->threshold_max_packets = 0; // По умолчанию: ждать пустой очереди
q->threshold_max_bytes = 0; // По умолчанию: не проверять байты
q->hash_size = hash_size;
q->index_offset = index_offset;
q->index_size = index_size;
@@ -112,6 +112,17 @@ void queue_free(struct ll_queue* q) {
// ВАЖНО: Не освобождаем элементы в очереди - они должны быть извлечены отдельно
// Это упрощает архитектуру и предотвращает double-u_free
// Освободить список waiters (сбросить handle->internal)
struct queue_waiter* w = q->waiter_head;
while (w) {
struct queue_waiter* next = w->next;
if (w->handle) w->handle->internal = NULL;
u_free(w);
w = next;
}
q->waiter_head = NULL;
q->waiter_tail = NULL;
// Освободить хеш-таблицу
if (q->hash_table) {
u_free(q->hash_table);
@@ -266,24 +277,27 @@ static uint32_t make_hash(const void* data, uint16_t len) {// алгоритм F
return hash;
}
// Проверить и запустить ожидающие коллбэки
// Проверить и запустить первый ожидающий коллбэк (FIFO / round-robin)
static void check_waiters(struct ll_queue* q) {
if (!q) return;
if (!q || !q->waiter_head) return;
// DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: checking waiters, count=%d, bytes=%zu", q->count, q->total_bytes);
const int cond = q->count <= q->threshold_max_packets
&& (q->threshold_max_bytes == 0 || q->total_bytes <= q->threshold_max_bytes);
if (!cond) return;
struct queue_waiter* waiter = &q->waiter;
struct queue_waiter* waiter = q->waiter_head;
q->waiter_head = waiter->next;
if (!q->waiter_head) q->waiter_tail = NULL;
if (waiter->callback) {
// Проверить условие: не больше max_packets и не больше max_bytes
// max_bytes = 0 означает "не проверять байты"
if (q->count <= waiter->max_packets && (waiter->max_bytes == 0 || q->total_bytes <= waiter->max_bytes)) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
waiter->callback(q, waiter->callback_arg);
memset(waiter, 0, sizeof(*waiter));
}
struct queue_waiter_handle* h = waiter->handle;
if (h) h->internal = NULL;
u_free(waiter);
if (h) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: waking head waiter, count=%d<=%d, bytes=%zu<=%zu",
q->count, q->threshold_max_packets, q->total_bytes, q->threshold_max_bytes);
h->callback(q, h->callback_arg);
}
}
@@ -532,34 +546,73 @@ int queue_check_consistency(struct ll_queue* q) {
return 0;
}
// ==================== Асинхронное ожидание ====================
// ==================== Пороговое ожидание (backpressure) ====================
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes,
queue_threshold_callback_fn callback, void* arg) {
if (!q || !callback) return NULL;
struct queue_waiter* waiter = &q->waiter;
// Проверить условие немедленно
if (q->count <= max_packets && (max_bytes == 0 || q->total_bytes <= max_bytes)) {
// Условие уже выполнено - вызвать коллбэк немедленно
callback(q, arg);
return NULL;
}
// Установить waiter для отложенного вызова
waiter->max_packets = max_packets;
waiter->max_bytes = max_bytes;
waiter->callback = callback;
waiter->callback_arg = arg;
return waiter;
void queue_set_threshold(struct ll_queue* q, int max_packets, size_t max_bytes) {
if (!q) return;
q->threshold_max_packets = max_packets;
q->threshold_max_bytes = max_bytes;
}
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter) {
if (!q || !waiter || waiter != &q->waiter) return;
void queue_waiter_handle_init(struct queue_waiter_handle* h,
queue_threshold_callback_fn callback, void* arg) {
if (!h) return;
h->internal = NULL;
h->callback = callback;
h->callback_arg = arg;
}
int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h) {
if (!q || !h) return 1;
memset(waiter, 0, sizeof(*waiter));
if (h->internal) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "[%s] queue_waiter_wait: waiter already registered", q->name);
return 1;
}
// Проверить условие немедленно
if (q->count <= q->threshold_max_packets
&& (q->threshold_max_bytes == 0 || q->total_bytes <= q->threshold_max_bytes)) {
h->callback(q, h->callback_arg);
return 1;
}
struct queue_waiter* waiter = u_malloc(sizeof(struct queue_waiter));
if (!waiter) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "[%s] queue_waiter_wait: u_malloc failed", q->name);
return -1;
}
waiter->next = NULL;
waiter->handle = h;
h->internal = waiter;
if (q->waiter_tail) q->waiter_tail->next = waiter; else q->waiter_head = waiter;
q->waiter_tail = waiter;
return 0;
}
void queue_waiter_cancel(struct ll_queue* q, struct queue_waiter_handle* h) {
if (!q || !h || !h->internal) return;
struct queue_waiter* waiter = h->internal;
struct queue_waiter* prev = NULL;
struct queue_waiter* curr = q->waiter_head;
while (curr && curr != waiter) {
prev = curr;
curr = curr->next;
}
if (!curr) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "[%s] queue_waiter_cancel: waiter not in list", q->name);
h->internal = NULL;
return;
}
if (prev) prev->next = waiter->next; else q->waiter_head = waiter->next;
if (waiter == q->waiter_tail) q->waiter_tail = prev;
h->internal = NULL;
u_free(waiter);
}
// ==================== Поиск и удаление по ID ====================
+57 -19
View File
@@ -26,7 +26,7 @@
* - Обязательный вызов queue_resume_callback после обработки элемента
* - Поддержка пулов памяти для entry и отдельно для dgram
* - Хеш-таблица для быстрого поиска по ID (опционально)
* - Встроенный одиночный waiter для ожидания освобождения очереди до заданного порога
* - Связный список waiters для backpressure c общим порогом (round-robin)
* - Использует uasync для отложенного возобновления callback'ов (без рекурсии в стеке)
*
* @note Важные правила использования автозабора (queue_set_callback):
@@ -37,13 +37,15 @@
* 5. queue_data_get() автоматически suspend'ит коллбэки для предотвращения рекурсии
*
* @note Память:
* - queue_free() освобождает ТОЛЬКО структуру очереди и хеш-таблицу
* - queue_free() освобождает структуру очереди, хеш-таблицу и список waiters
* (handle->internal каждого ожидающего сбрасывается в NULL)
* - Все элементы должны быть извлечены через queue_data_get() и освобождены через queue_entry_free()
* - dgram освобождается отдельно через queue_dgram_free() или кастомную dgram_free_fn
*/
// Forward declaration
// Forward declarations
struct ll_queue;
struct queue_waiter_handle;
/**
* @struct ll_entry
@@ -84,7 +86,7 @@ typedef void (*queue_callback_fn)(struct ll_queue* q, void* arg);
/**
* @typedef queue_threshold_callback_fn
* @brief Одноразовый коллбэк при достижении порога очереди (queue_wait_threshold).
* @brief Одноразовый коллбэк при достижении порога очереди.
* @param q указатель на очередь
* @param arg пользовательский аргумент
*/
@@ -92,13 +94,27 @@ typedef void (*queue_threshold_callback_fn)(struct ll_queue* q, void* arg);
/**
* @struct queue_waiter
* @brief Описание ожидания освобождения места в очереди (встроен в ll_queue — только один).
* @brief Внутренний узел в связном списке ожидающих (очередь владеет, аллоцирует/освобождает сама).
*/
struct queue_waiter {
int max_packets; ///< Максимально допустимое количество элементов
size_t max_bytes; ///< Максимально допустимый объём данных (0 = не проверять)
queue_threshold_callback_fn callback; ///< Коллбэк (вызывается один раз)
void* callback_arg; ///< Аргумент коллбэка
struct queue_waiter* next; ///< Следующий в списке ожидающих
struct queue_waiter_handle* handle; ///< Обратная ссылка на публичную структуру
};
/**
* @struct queue_waiter_handle
* @brief Публичная управляющая структура (встраивается в структуру вызывающей стороны).
*
* Вызывающая сторона:
* 1. Объявляет поле struct queue_waiter_handle в своей структуре
* 2. Вызывает queue_waiter_handle_init() один раз при инициализации
* 3. Вызывает queue_waiter_wait() когда хочет дождаться освобождения очереди
* 4. Вызывает queue_waiter_cancel() при деинициализации для отмены ожидания
*/
struct queue_waiter_handle {
struct queue_waiter* internal; ///< NULL = не ждём; иначе — внутренний узел
queue_threshold_callback_fn callback;
void* callback_arg;
};
/**
@@ -120,7 +136,10 @@ struct ll_queue {
void* resume_timeout_id; // ID таймера uasync для отложенного resume
struct UASYNC* ua; // Экземпляр uasync (обязателен для таймеров)
struct queue_waiter waiter; // Встроенный waiter (только один)
struct queue_waiter* waiter_head; // Голова списка ожидающих
struct queue_waiter* waiter_tail; // Хвост списка ожидающих
int threshold_max_packets; // Общий порог: макс. кол-во элементов (по умолчанию 0)
size_t threshold_max_bytes; // Общий порог: макс. объём данных (0 = не проверять)
struct ll_entry** hash_table; // Хеш-таблица для поиска по id (если hash_size > 0)
size_t hash_size; // Размер хеш-таблицы
@@ -190,26 +209,45 @@ void queue_set_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg)
*/
void queue_resume_callback(struct ll_queue* q);
/* ==================== Пороговое ожидание ==================== */
/* ==================== Пороговое ожидание (backpressure) ==================== */
/**
* @brief Регистрирует одноразовый коллбэк, который сработает, когда очередь освободится до заданного порога.
* @brief Устанавливает общий порог освобождения очереди.
* @param q очередь
* @param max_packets максимальное количество элементов
* @param max_packets максимальное количество элементов (условие: count <= max_packets)
* @param max_bytes максимальный объём данных (0 = не проверять)
*
* Порог — общий на всю очередь. Все waiters в связном списке используют этот порог.
* При достижении порога вызывается следующий waiter из списка (FIFO / round-robin).
* По умолчанию: max_packets=0, max_bytes=0 (пустая очередь).
*/
void queue_set_threshold(struct ll_queue* q, int max_packets, size_t max_bytes);
/**
* @brief Инициализирует публичную управляющую структуру waiter.
* @param h указатель на handle (встраивается в структуру вызывающей стороны)
* @param callback функция, вызываемая при достижении порога
* @param arg аргумент коллбэка
* @return указатель на waiter (для отмены) или NULL, если условие уже выполнено (коллбэк вызван сразу)
*/
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes,
queue_threshold_callback_fn callback, void* arg);
void queue_waiter_handle_init(struct queue_waiter_handle* h,
queue_threshold_callback_fn callback, void* arg);
/**
* @brief Отменяет ранее установленное ожидание порога.
* @brief Регистрирует ожидание освобождения очереди до общего порога.
* @param q очередь
* @param waiter указатель, возвращённый queue_wait_threshold()
* @param h указатель на инициализированный handle
* @return 1 — условие уже выполнено (callback вызван немедленно), 0 — зарегистрирован в списке
*/
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter);
int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h);
/**
* @brief Отменяет ожидание (удаляет из списка, если зарегистрирован).
* @param q очередь
* @param h указатель на handle
*
* Безопасно вызывать в любом состоянии: если waiter не зарегистрирован — ничего не делает.
*/
void queue_waiter_cancel(struct ll_queue* q, struct queue_waiter_handle* h);
/* ==================== Работа с данными ==================== */