diff --git a/lib/tcp_io.c b/lib/tcp_io.c index bcac4e38..6705ab68 100644 --- a/lib/tcp_io.c +++ b/lib/tcp_io.c @@ -68,6 +68,7 @@ struct tcp_conn* tcp_conn_create( u_free(tc); return NULL; } queue_set_threshold(tc->read_queue, read_low_water, 0); + queue_set_threshold(tc->write_queue, 32, 0); queue_set_callback(tc->write_queue, write_queue_fetch_cb, tc); queue_set_waiter_defer(tc->write_queue, 1); @@ -326,33 +327,3 @@ void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_con if (!tc) return; tc->on_flushed = on_flushed; } - -// ==================================================================== -// Отправка данных в сокет (внешний интерфейс) -// ==================================================================== - -int tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len) { - if (!tc || !data || len == 0) return -1; - if (tc->error || tc->fin) return -1; - - size_t offset = 0; - while (offset < len) { - size_t chunk = len - offset; - if (chunk > tc->write_chunk_size) chunk = tc->write_chunk_size; - - struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); - uint8_t* buf = memory_pool_alloc(tc->data_pool); - if (!e || !buf) { - DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: push_write alloc failed fd=%d chunk=%zu", (int)tc->sock, chunk); - if (e) queue_entry_free(e); - if (buf) memory_pool_free(tc->data_pool, buf); - return -1; - } - memcpy(buf, data + offset, chunk); - e->dgram = buf; - e->len = (uint16_t)chunk; - queue_data_put(tc->write_queue, e); - offset += chunk; - } - return 0; -} diff --git a/lib/tcp_io.h b/lib/tcp_io.h index ec783bff..6df99757 100644 --- a/lib/tcp_io.h +++ b/lib/tcp_io.h @@ -85,9 +85,10 @@ struct tcp_conn* tcp_conn_create( void tcp_conn_destroy(struct tcp_conn* tc); -// Отправка данных в сокет: режет на чанки по write_chunk_size, кладёт в write_queue. -// Автозабор write_queue сам отправит когда сокет будет готов. -int tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len); +// Внешний код пишет данные в tc->write_queue напрямую (queue_data_put). +// Автозабор write_queue (deferred) сам отправляет когда сокет готов. +// Перед push проверять порог: queue_set_threshold в tcp_conn_create (32 entries). +// При заполнении — queue_waiter_wait на освобождение. // Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели. // После вызова сбрасывается. Установить повторно можно в любой момент. diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c index 1b5d0b3d..f560a5bb 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -238,7 +238,19 @@ int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* c queue_dgram_free(entry); queue_entry_free(entry); return -1; } size_t data_len = entry->len - TCP_PROXY_HDR_SIZE; - if (data_len > 0) tcp_conn_push_write(rc->tc, entry->dgram + TCP_PROXY_HDR_SIZE, data_len); + if (data_len > 0) { + struct ll_entry* e = queue_entry_new_from_pool(rc->tc->entry_pool); + uint8_t* buf = memory_pool_alloc(rc->tc->data_pool); + if (e && buf) { + memcpy(buf, entry->dgram + TCP_PROXY_HDR_SIZE, data_len); + e->dgram = buf; e->len = (uint16_t)data_len; + queue_data_put(rc->tc->write_queue, e); + } else { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server: handle_data alloc failed sid=%08x", stream_id); + if (e) queue_entry_free(e); + if (buf) memory_pool_free(rc->tc->data_pool, buf); + } + } queue_dgram_free(entry); queue_entry_free(entry); return 0; } diff --git a/tests/test_tcp_io.c b/tests/test_tcp_io.c index c313c5f7..1de58aaf 100644 --- a/tests/test_tcp_io.c +++ b/tests/test_tcp_io.c @@ -48,6 +48,22 @@ static void reset_counters(void) { g_last_fin_tc = NULL; } +static int push_write(struct tcp_conn* tc, const uint8_t* data, size_t len) { + size_t offset = 0; + while (offset < len) { + size_t chunk = len - offset; + if (chunk > tc->write_chunk_size) chunk = tc->write_chunk_size; + struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); + uint8_t* buf = memory_pool_alloc(tc->data_pool); + if (!e || !buf) { if (e) queue_entry_free(e); if (buf) memory_pool_free(tc->data_pool, buf); return -1; } + memcpy(buf, data + offset, chunk); + e->dgram = buf; e->len = (uint16_t)chunk; + queue_data_put(tc->write_queue, e); + offset += chunk; + } + return 0; +} + static void test_basic_send_recv(void) { TEST_START("Basic send and recv via tcp_conn"); @@ -65,8 +81,8 @@ static void test_basic_send_recv(void) { // Отправляем данные через tcp_conn uint8_t send_buf[3000]; memset(send_buf, 'A', sizeof(send_buf)); - int ret = tcp_conn_push_write(tc, send_buf, sizeof(send_buf)); - ASSERT_EQ(ret, 0, "tcp_conn_push_write first call failed"); + int ret = push_write(tc, send_buf, sizeof(send_buf)); + ASSERT_EQ(ret, 0, "push_write first call failed"); uasync_poll(ua, 1); // write_cb должен отправить // Читаем с другой стороны @@ -126,8 +142,8 @@ static void test_partial_write(void) { ASSERT_TRUE(big_buf != NULL, "malloc failed"); memset(big_buf, 'X', 200000); - int ret = tcp_conn_push_write(tc, big_buf, 200000); - ASSERT_EQ(ret, 0, "tcp_conn_push_write large failed"); + int ret = push_write(tc, big_buf, 200000); + ASSERT_EQ(ret, 0, "push_write large failed"); uasync_poll(ua, 1); // write_cb отправляет что может // Дрейним sv[1] и проверяем что все данные приходят diff --git a/tests/test_tcp_proxy_server.c b/tests/test_tcp_proxy_server.c index bc478770..ca69c2e9 100644 --- a/tests/test_tcp_proxy_server.c +++ b/tests/test_tcp_proxy_server.c @@ -43,6 +43,16 @@ done: close(cli); close(srv); static void on_fin_cb(struct tcp_conn* tc, void* arg) { (void)tc; (void)arg; } static void on_error_cb(struct tcp_conn* tc, int err, void* arg) { (void)tc; (void)arg; (void)err; } +static int push_write(struct tcp_conn* tc, const uint8_t* data, size_t len) { + struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); + uint8_t* buf = memory_pool_alloc(tc->data_pool); + if (!e || !buf) { if (e) queue_entry_free(e); if (buf) memory_pool_free(tc->data_pool, buf); return -1; } + memcpy(buf, data, len); + e->dgram = buf; e->len = (uint16_t)len; + queue_data_put(tc->write_queue, e); + return 0; +} + static void timeout_cb(void* arg) { (void)arg; printf("[FAIL] timeout\n"); @@ -84,7 +94,7 @@ int main(void) { if (phase == 0 && tc->connected) { phase = 1; for (int i = 0; i < PAYLOAD_SIZE; i++) send_buf[i] = (uint8_t)(rand() & 0xFF); - tcp_conn_push_write(tc, send_buf, PAYLOAD_SIZE); + push_write(tc, send_buf, PAYLOAD_SIZE); } if (phase == 1 && !tc->error) {