From d467d3aa8fa8ef53edbd2dc400d79de90b1c41d4 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 21 Oct 2026 16:44:19 +0200 Subject: [PATCH] Fix TCP connect timer ownership and pending FIN state --- lib/tcp_io.c | 65 ++++++++++++++++++++++++++++-------- lib/tcp_io.h | 33 +++++++++++------- src/proxy/tcp_proxy_server.c | 9 ++++- 3 files changed, 80 insertions(+), 27 deletions(-) diff --git a/lib/tcp_io.c b/lib/tcp_io.c index 8bca3f5d..6aea666d 100644 --- a/lib/tcp_io.c +++ b/lib/tcp_io.c @@ -5,11 +5,7 @@ #include #include #include -#ifndef _WIN32 -#include -#include -#include -#endif +#include #ifndef MSG_NOSIGNAL #define MSG_NOSIGNAL 0 #endif @@ -119,8 +115,8 @@ void tcp_conn_destroy(struct tcp_conn* tc) { if (!tc) return; if (tc->destroyed) return; tc->destroyed = 1; - DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin_remote=%d fin_local=%d closed=%d", - (int)tc->sock, tc->connected, tc->error, tc->fin_remote, tc->fin_local, tc->closed); + DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin_remote=%d fin_pending=%d fin_local=%d closed=%d", + (int)tc->sock, tc->connected, tc->error, tc->fin_remote, tc->fin_pending, tc->fin_local, tc->closed); if (tc->connect_timer) { uasync_cancel_timeout(tc->ua, tc->connect_timer); tc->connect_timer = NULL; } @@ -153,6 +149,7 @@ void tcp_conn_destroy(struct tcp_conn* tc) { static void tcp_conn_handle_error(struct tcp_conn* tc, int err) { tc->error = 1; + if (tc->connect_timer) { uasync_cancel_timeout(tc->ua, tc->connect_timer); tc->connect_timer = NULL; } queue_waiter_cancel(tc->read_queue, &tc->read_waiter); queue_set_empty_callback(tc->read_queue, NULL, NULL); @@ -325,7 +322,14 @@ static void write_queue_fetch_cb(struct ll_queue* q, void* arg) { } else if (e->dgram == &tcp_fin_sentinel) { DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: FIN fd=%d", (int)tc->sock); queue_entry_free(e); queue_resume_callback(q); - shutdown(tc->sock, SHUT_WR); + tc->fin_pending = 0; + if (shutdown(tc->sock, SHUT_WR) != 0) { + int err = socket_get_error(); + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: shutdown write failed fd=%d err=%d (%s)", + (int)tc->sock, err, socket_strerror(err)); + tcp_conn_handle_error(tc, err); + return; + } tc->fin_local = 1; if (tc->on_fin_sent) tc->on_fin_sent(tc, tc->arg); } else { @@ -467,11 +471,13 @@ static void error_cb(socket_t sock, void* arg) { int tcp_conn_push_fin(struct tcp_conn* tc) { if (!tc || tc->sock == SOCKET_INVALID) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin invalid tc"); return -1; } + if (tc->fin_pending) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already pending fd=%d", (int)tc->sock); return -1; } if (tc->fin_local) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already fin_local fd=%d", (int)tc->sock); return -1; } if (tc->closed) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already closed fd=%d", (int)tc->sock); return -1; } struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin entry_pool exhausted fd=%d", (int)tc->sock); return -1; } e->dgram = (uint8_t*)&tcp_fin_sentinel; e->len = 0; + tc->fin_pending = 1; DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin fd=%d wq=%d", (int)tc->sock, tc->write_queue->count + 1); queue_data_put(tc->write_queue, e); return 0; @@ -514,11 +520,42 @@ static void connect_timeout_cb(void* arg) { } } -void tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms) { - if (!tc) return; - if (timeout_ms < 0) timeout_ms = 0; - tc->connect_timeout_ms = timeout_ms; - if (!tc->connected && timeout_ms > 0) { - tc->connect_timer = uasync_set_timeout(tc->ua, timeout_ms * 10, tc, connect_timeout_cb, "tcp_conn_ct"); +/* Замена атомарна относительно event loop: сначала резервируем новый таймер, затем отменяем прежний. */ +int tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms) { + if (!tc) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: set connect timeout: tc=NULL timeout=%d", timeout_ms); + return -EINVAL; } + if (tc->destroyed || tc->error || tc->closed || tc->sock == SOCKET_INVALID) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: set connect timeout on inactive tc fd=%d destroyed=%d error=%d closed=%d", + (int)tc->sock, tc->destroyed, tc->error, tc->closed); + return -EINVAL; + } + if (timeout_ms > INT_MAX / 10) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: connect timeout=%d exceeds limit=%d ms fd=%d", + timeout_ms, INT_MAX / 10, (int)tc->sock); + return -EINVAL; + } + if (timeout_ms < 0) timeout_ms = 0; + void* timer = NULL; + if (!tc->connected && timeout_ms > 0) { + timer = uasync_set_timeout(tc->ua, timeout_ms * 10, tc, connect_timeout_cb, "tcp_conn_ct"); + if (!timer) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot create connect timer fd=%d timeout=%d ms; old timeout=%d retained", + (int)tc->sock, timeout_ms, tc->connect_timeout_ms); + return -ENOMEM; + } + } + void* old_timer = tc->connect_timer; + if (old_timer && uasync_cancel_timeout(tc->ua, old_timer) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot cancel connect timer fd=%d timer=%p", (int)tc->sock, old_timer); + if (timer && uasync_cancel_timeout(tc->ua, timer) != ERR_OK) + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot cancel replacement connect timer fd=%d timer=%p", (int)tc->sock, timer); + return -EIO; + } + tc->connect_timer = timer; + tc->connect_timeout_ms = timeout_ms; + DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: connect timeout updated fd=%d connected=%d timeout=%d ms old_timer=%p timer=%p", + (int)tc->sock, tc->connected, timeout_ms, old_timer, timer); + return 0; } diff --git a/lib/tcp_io.h b/lib/tcp_io.h index e3dab4c9..f26cb9d3 100644 --- a/lib/tcp_io.h +++ b/lib/tcp_io.h @@ -9,7 +9,8 @@ // FIN (recv==0) → on_fin (сразу или через empty_callback после дренажа очереди). // // Запись: внешний код кладёт entry в write_queue через queue_data_put(). -// write_queue_fetch_cb (deferred) → send(), при EAGAIN → write_buf + EPOLLOUT. +// Первый write_queue_fetch_cb может быть синхронным; продолжение — через uasync_call_soon. +// send(), при EAGAIN → write_buf + EPOLLOUT. // Всё отправлено → on_flushed. // // FIN/Close: tcp_conn_push_fin/close ставят сентинел в write_queue. @@ -49,6 +50,7 @@ struct tcp_conn { uint8_t connected; uint8_t error; uint8_t fin_remote; // FIN получен от удалённой стороны (recv == 0) + uint8_t fin_pending; // FIN поставлен в write_queue, shutdown ещё не выполнен uint8_t fin_local; // FIN отправлен удалённой стороне (shutdown SHUT_WR) uint8_t closed; // сокет полностью закрыт (close) uint8_t destroyed; // 1 = tcp_conn_destroy вызван, tc ожидает отложенного free @@ -91,37 +93,44 @@ struct tcp_conn* tcp_conn_create( void (*on_error)(struct tcp_conn* tc, int err, void* arg), void* arg); -// Дренирует очереди, закрывает сокет, уничтожает пулы, освобождает tc. +// Сразу отменяет callbacks/таймер, дренирует очереди и закрывает сокет. +// Пулы и tc освобождаются через uasync_call_soon в том же uasync-потоке. +// NULL безопасен; повторный destroy допустим только до отложенного free. +// Владелец после destroy больше не использует tc; event loop должен выполнить отложенную очистку. void tcp_conn_destroy(struct tcp_conn* tc); // Запись: queue_data_put(tc->write_queue, entry) напрямую. -// Автозабор (deferred) сам шлёт когда сокет готов. +// Первый автозабор может быть синхронным; queue_resume_callback откладывает продолжение. // Порог: 32 записи, при заполнении queue_waiter_wait. // Поставить сигналы в очередь отправки. Все данные в очереди перед сигналом -// будут отправлены до его обработки. Кодирование: dgram=NULL → close, len=0+dgram!=NULL → FIN. +// будут переданы ОС до его обработки. Сигналы имеют len=0: dgram=NULL → CLOSE, +// dgram=&tcp_fin_sentinel → FIN (внутренний указатель, сигналы ставить только через API). // tcp_conn_push_fin(tc) — FIN: после отправки предшествующих данных вызывает // shutdown(SHUT_WR) и on_fin_sent. -// Повторный вызов игнорируется (fin_sent уже установлен). +// Повторный вызов возвращает -1 при fin_pending или fin_local. // tcp_conn_push_close(tc) — CLOSE: после отправки предшествующих данных закрывает // сокет (close), вызывает on_closed. // После close сокет удалён из uasync, tc->closed=1. -// Повторный вызов игнорируется. +// После фактического закрытия повторный вызов возвращает -1. +// Оба API: 0 = сигнал принят; -1 = ошибка. Callback может выполниться до возврата из push. int tcp_conn_push_fin(struct tcp_conn* tc); int tcp_conn_push_close(struct tcp_conn* tc); // Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели. -// После вызова сбрасывается. Установить повторно можно в любой момент. +// Сбрасывается перед вызовом; callback может установить новое уведомление. void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg)); // Принудительная остановка чтения: убирает EPOLLIN и отменяет read_waiter. void tcp_conn_pause_read(struct tcp_conn* tc); -// Таймаут установления соединения (мс, 0 = без таймаута). Если соединение ещё не -// установлено (connected=0), взводит таймер; по истечении вызывает -// tcp_conn_handle_error(ETIMEDOUT) → on_error. Таймер гасится при успешном connect -// и при tcp_conn_destroy. -void tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms); +// Заменить connect-таймаут: мс от текущего вызова, <=0 отменяет таймер. +// Только для живого tc в его uasync-потоке. Для connected=1 таймер не создаётся. +// Допустимый положительный диапазон: <= INT_MAX/10 (uasync принимает 0.1 мс). +// Возврат: 0 = применено; -EINVAL = аргументы/состояние, -ENOMEM = создание таймера, +// -EIO = нарушение состояния отменяемого таймера. При ошибке прежняя настройка сохраняется. +// Функция не вызывает callbacks. Истечение → on_error(ETIMEDOUT); отмена при connect/error/destroy. +int tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms); #ifdef __cplusplus diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c index f1209594..eba13533 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -330,7 +330,14 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* rc->next = ctx->conns; ctx->conns = rc; if (ret == 0) { rc->tc->connected = 1; on_connected_cb(rc->tc, rc); } - tcp_conn_set_connect_timeout(rc->tc, g->tcp_proxy_server_connect_timeout_ms); + int timeout_rc = tcp_conn_set_connect_timeout(rc->tc, g->tcp_proxy_server_connect_timeout_ms); + if (timeout_rc < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: cannot set connect timeout sid=%08x timeout=%d rc=%d", + stream_id, g->tcp_proxy_server_connect_timeout_ms, timeout_rc); + on_error_cb(rc->tc, -timeout_rc, rc); + queue_dgram_free(entry); queue_entry_free(entry); + return -1; + } rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag"); queue_dgram_free(entry); queue_entry_free(entry);