From 4da9d855451884c81db1c6fcf0e2307f3fa639d5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 13:28:44 +0200 Subject: [PATCH] Honor TCP read backpressure across HUP and restore polling --- lib/tcp_io.c | 109 +++++++++++++++++++++++++------------------- lib/tcp_io.h | 2 + tests/test_tcp_io.c | 59 ++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 47 deletions(-) diff --git a/lib/tcp_io.c b/lib/tcp_io.c index 2ea83f23..a0a204a9 100644 --- a/lib/tcp_io.c +++ b/lib/tcp_io.c @@ -24,6 +24,38 @@ static void tcp_conn_handle_error(struct tcp_conn* tc, int err); static void connect_timeout_cb(void* arg); static const uint8_t tcp_fin_sentinel; +// Сохраняем fd при HUP, но не оставляем level-triggered событие в poll без работы. +// Очередь восстанавливает регистрацию с новым generation; write_buf обслуживается независимо от чтения. +static int sync_socket_monitoring(struct tcp_conn* tc) { + if (tc->destroyed || tc->error || tc->closed || tc->sock == SOCKET_INVALID) return -1; + int read = !tc->read_paused && !tc->read_stopped && !tc->fin_remote; + int write = !tc->connected || tc->write_buf != NULL; + if (tc->hup_received && !read && !write) { + if (tc->socket_id) { + uasync_remove_socket_t(tc->ua, tc->sock); tc->socket_id = NULL; + DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: park HUP fd=%d rq=%d paused=%u stopped=%u fin=%u", + (int)tc->sock, tc->read_queue->count, tc->read_paused, tc->read_stopped, tc->fin_remote); + } + tc->write_monitor = 0; + return 0; + } + if (!tc->socket_id) { + tc->socket_id = uasync_add_socket_t(tc->ua, tc->sock, read_cb, write_cb, error_cb, "tcp_io", tc); + if (!tc->socket_id) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot restore socket registration fd=%d", (int)tc->sock); + tcp_conn_handle_error(tc, ENOMEM); return -1; + } + DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: restore HUP fd=%d read=%d write=%d", (int)tc->sock, read, write); + } + if (uasync_set_socket_read(tc->ua, tc->socket_id, read) != ERR_OK || + uasync_set_socket_write(tc->ua, tc->socket_id, write) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: monitoring update failed fd=%d read=%d write=%d", (int)tc->sock, read, write); + tcp_conn_handle_error(tc, EIO); return -1; + } + tc->write_monitor = write; + return 0; +} + struct tcp_conn* tcp_conn_create( struct UASYNC* ua, socket_t sock, size_t entry_data_size, size_t write_chunk_size, @@ -201,7 +233,7 @@ static void tcp_conn_handle_error(struct tcp_conn* tc, int err) static void read_cb(socket_t sock, void* arg) { (void)sock; struct tcp_conn* tc = (struct tcp_conn*)arg; - if (!tc || tc->sock == SOCKET_INVALID || !tc->read_queue) return; + if (!tc || tc->sock == SOCKET_INVALID || !tc->read_queue || tc->read_paused || tc->read_stopped || tc->fin_remote) return; struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); uint8_t* buf = memory_pool_alloc(tc->data_pool); @@ -209,8 +241,7 @@ static void read_cb(socket_t sock, void* arg) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: read_cb alloc failed fd=%d", (int)tc->sock); if (e) queue_entry_free(e); if (buf) memory_pool_free(tc->data_pool, buf); - uasync_set_socket_read(tc->ua, tc->socket_id, 0); - tc->read_paused = 1; + tcp_conn_handle_error(tc, ENOMEM); return; } @@ -222,9 +253,12 @@ static void read_cb(socket_t sock, void* arg) { queue_data_put(rq, e); // tc мог быть уничтожен через callback chain — read_queue обнулён if (rq && tc->read_queue && tc->read_queue->count >= tc->read_high_water && !tc->read_paused) { - uasync_set_socket_read(tc->ua, tc->socket_id, 0); tc->read_paused = 1; - queue_waiter_wait(tc->read_queue, &tc->read_waiter, resume_read_cb, tc); + if (sync_socket_monitoring(tc) < 0) return; + if (queue_waiter_wait(tc->read_queue, &tc->read_waiter, resume_read_cb, tc) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot wait for read queue fd=%d", (int)tc->sock); + tcp_conn_handle_error(tc, ENOMEM); + } } } else if (n == 0) { memory_pool_free(tc->data_pool, buf); @@ -232,8 +266,8 @@ static void read_cb(socket_t sock, void* arg) { if (tc->fin_remote) return; tc->fin_remote = 1; DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: FIN fd=%d", (int)tc->sock); - uasync_set_socket_read(tc->ua, tc->socket_id, 0); queue_waiter_cancel(tc->read_queue, &tc->read_waiter); + if (sync_socket_monitoring(tc) < 0) return; if (tc->read_queue->count == 0) { if (tc->on_fin) tc->on_fin(tc, tc->arg); } else { @@ -251,9 +285,16 @@ static void read_cb(socket_t sock, void* arg) { static void resume_read_cb(struct ll_queue* q, void* arg) { (void)q; struct tcp_conn* tc = (struct tcp_conn*)arg; - if (!tc || tc->sock == SOCKET_INVALID) return; - uasync_set_socket_read(tc->ua, tc->socket_id, 1); + if (!tc || tc->sock == SOCKET_INVALID || tc->destroyed || tc->read_stopped || tc->fin_remote) return; + if (tc->read_queue->count > tc->read_low_water) { + if (queue_waiter_wait(tc->read_queue, &tc->read_waiter, resume_read_cb, tc) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot renew read waiter fd=%d", (int)tc->sock); + tcp_conn_handle_error(tc, ENOMEM); + } + return; + } tc->read_paused = 0; + sync_socket_monitoring(tc); } static void fin_deferred_cb(struct ll_queue* q, void* arg) { @@ -313,8 +354,7 @@ static void write_queue_fetch_cb(struct ll_queue* q, void* arg) { struct ll_entry* e = queue_data_get(q); if (!e) { - uasync_set_socket_write(tc->ua, tc->socket_id, 0); - tc->write_monitor = 0; + if (sync_socket_monitoring(tc) < 0) return; DEBUG_TRACE(DEBUG_CATEGORY_SOCKET, "tcp_io: wq_fetch POLLOUT OFF fd=%d (queue empty)", (int)tc->sock); queue_resume_callback(q); notify_flushed(tc); @@ -370,8 +410,7 @@ static void write_queue_fetch_cb(struct ll_queue* q, void* arg) { tc->write_offset = 0; memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); - uasync_set_socket_write(tc->ua, tc->socket_id, 1); - tc->write_monitor = 1; + if (sync_socket_monitoring(tc) < 0) return; DEBUG_TRACE(DEBUG_CATEGORY_SOCKET, "tcp_io: wq_fetch POLLOUT ON fd=%d (partial send, wbuf=%zu bytes)", (int)tc->sock, tc->write_len); } else if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) { tc->write_buf = memory_pool_alloc(tc->data_pool); @@ -387,8 +426,7 @@ static void write_queue_fetch_cb(struct ll_queue* q, void* arg) { tc->write_offset = 0; memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); - uasync_set_socket_write(tc->ua, tc->socket_id, 1); - tc->write_monitor = 1; + if (sync_socket_monitoring(tc) < 0) return; DEBUG_TRACE(DEBUG_CATEGORY_SOCKET, "tcp_io: wq_fetch POLLOUT ON fd=%d (EAGAIN, wbuf=%zu bytes)", (int)tc->sock, tc->write_len); } else { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: send error fd=%d errno=%d", (int)tc->sock, errno); @@ -427,10 +465,7 @@ static void write_cb(socket_t sock, void* arg) { if (flush_write_buf(tc)) return; if (!tc->write_buf) { DEBUG_TRACE(DEBUG_CATEGORY_SOCKET, "tcp_io: write_cb flushed fd=%d wmon=%d (resume WQ cb)", (int)tc->sock, tc->write_monitor); - if (tc->write_queue->waiter_defer) { - uasync_set_socket_write(tc->ua, tc->socket_id, 0); - tc->write_monitor = 0; - } + if (sync_socket_monitoring(tc) < 0) return; queue_resume_callback(tc->write_queue); notify_flushed(tc); } @@ -446,33 +481,13 @@ static void error_cb(socket_t sock, void* arg) { if (!tc || tc->sock == SOCKET_INVALID) return; int so_err = 0; socklen_t so_len = sizeof(so_err); if (getsockopt(tc->sock, SOL_SOCKET, SO_ERROR, (char*)&so_err, &so_len) == 0 && so_err == 0) { - // Peer полностью закрыл свой сокет (EPOLLHUP) после того, как FIN уже прочитан (fin_remote=1). - // EPOLLHUP — level-triggered и «always reported»: если оставить сокет в epoll, каждая итерация - // epoll_wait мгновенно вернёт HUP и error_cb зациклится на 100% CPU. Закрываем сокет. - if (tc->fin_remote && tc->fin_local) { - // Обе TCP-половины закрыты, но владелец ещё может передавать ранее прочитанные данные. - // Убираем level-triggered HUP из poll; CLOSE остаётся решением владельца. - DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: both FINs fd=%d — detach poll, await owner drain", (int)tc->sock); - if (tc->socket_id) uasync_remove_socket_t(tc->ua, tc->sock); - tc->socket_id = NULL; tc->write_monitor = 0; - return; - } - if (tc->fin_remote) { - DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: async HUP (graceful) fd=%d — peer fully closed after FIN, closing", - (int)tc->sock); - tcp_conn_push_close(tc); - return; - } - // Грациозное закрытие peer (FIN, EPOLLHUP) при приостановленном чтении: FIN пришёл не как - // EPOLLIN (чтение отключено backpressure-ом), а как HUP. Возобновляем чтение, чтобы - // read_cb дослил остаток и прочитал FIN (recv()==0 → fin_remote → отложенный on_fin). - DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: async HUP (graceful) fd=%d fin_r=%d fin_l=%d read_paused=%d rq=%d — resume read", - (int)tc->sock, tc->fin_remote, tc->fin_local, tc->read_paused, - tc->read_queue ? tc->read_queue->count : 0); - queue_waiter_cancel(tc->read_queue, &tc->read_waiter); - uasync_set_socket_read(tc->ua, tc->socket_id, 1); - tc->read_paused = 0; - read_cb(tc->sock, tc); + if (!tc->hup_received) + DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: HUP fd=%d fin_r=%u fin_l=%u paused=%u stopped=%u rq=%d", + (int)tc->sock, tc->fin_remote, tc->fin_local, tc->read_paused, tc->read_stopped, tc->read_queue->count); + tc->hup_received = 1; + if (tc->write_buf && flush_write_buf(tc)) return; + if (!tc->fin_remote && !tc->read_paused && !tc->read_stopped) read_cb(tc->sock, tc); + if (!tc->destroyed && !tc->error && !tc->closed) sync_socket_monitoring(tc); return; } DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: async error fd=%d connected=%d err=%d so_err=%d fin_r=%d fin_l=%d closed=%d wmon=%d", @@ -516,9 +531,9 @@ void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_con void tcp_conn_pause_read(struct tcp_conn* tc) { if (!tc || tc->sock == SOCKET_INVALID) return; - uasync_set_socket_read(tc->ua, tc->socket_id, 0); queue_waiter_cancel(tc->read_queue, &tc->read_waiter); - tc->read_paused = 1; + tc->read_stopped = 1; tc->read_paused = 1; + sync_socket_monitoring(tc); } // ==================================================================== diff --git a/lib/tcp_io.h b/lib/tcp_io.h index 4142d082..658e8638 100644 --- a/lib/tcp_io.h +++ b/lib/tcp_io.h @@ -53,6 +53,8 @@ struct tcp_conn { int read_low_water; // Порог очереди для возобновления чтения int rcvbuf_size; // Запрошенный SO_RCVBUF; <=0 — не настраивать буферы uint8_t read_paused; // Чтение отключено backpressure или pause_read + uint8_t read_stopped; // Владелец запретил чтение; low_water/HUP не снимает запрет + uint8_t hup_received; // Получен HUP с SO_ERROR=0; idle registration снимается до готовности очереди uint8_t write_monitor; // 1 = EPOLLOUT активен uint8_t connected; diff --git a/tests/test_tcp_io.c b/tests/test_tcp_io.c index e6ea34d1..90547220 100644 --- a/tests/test_tcp_io.c +++ b/tests/test_tcp_io.c @@ -832,6 +832,64 @@ static void test_fin_before_close(void) { TEST_PASS(); } +// Закрытый peer держит HUP даже при выключенном IN: consumer не должен терять ограничение очереди. +static void test_hup_backpressure(void) { + for (int backend = 0; backend < 2; backend++) { + TEST_START("HUP respects high_water, restores registration and delivers DATA before FIN"); + size_t allocated = u_get_allocated_count(); + struct UASYNC* ua = uasync_create(); ASSERT_TRUE(ua != NULL, "create ua"); + ua->use_epoll = backend; + reset_counters(); + int sv[2]; ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair"); + for (int i = 0; i < 2; i++) ASSERT_EQ(socket_set_nonblocking(sv[i]), 0, "nonblocking"); + struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 64, 64, 4, 1, 0, on_fin, on_error, NULL); + ASSERT_TRUE(tc != NULL, "create tcp"); + uint8_t bytes[1024]; for (size_t i = 0; i < sizeof(bytes); i++) bytes[i] = (uint8_t)i; + ASSERT_EQ(send(sv[1], bytes, sizeof(bytes), MSG_NOSIGNAL), sizeof(bytes), "send peer data"); + close(sv[1]); + for (int i = 0; i < 30; i++) uasync_poll(ua, 0); + ASSERT_EQ(tc->read_queue->count, 4, "HUP bypassed high water"); + ASSERT_TRUE(tc->read_paused && tc->hup_received && !tc->socket_id, "HUP not parked"); + ASSERT_EQ(g_fin_count, 0, "FIN before buffered DATA consumed"); + size_t received = 0; + for (int i = 0; i < 200 && !g_fin_count; i++) { + struct ll_entry* e = queue_data_get(tc->read_queue); + if (e) { + ASSERT_TRUE(received + e->len <= sizeof(bytes), "extra data"); + ASSERT_EQ(memcmp(e->dgram, bytes + received, e->len), 0, "corrupted data after HUP"); + received += e->len; memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); + queue_resume_callback(tc->read_queue); + } + uasync_poll(ua, 0); + ASSERT_TRUE(tc->read_queue->count <= 4, "restored HUP bypassed high water"); + } + ASSERT_EQ(received, sizeof(bytes), "missing HUP data"); + ASSERT_EQ(g_fin_count, 1, "FIN missing/duplicated"); + ASSERT_EQ(g_error_count, 0, "unexpected socket error"); + tcp_conn_destroy(tc); uasync_drain_immediate(ua); + ASSERT_EQ(ua->socket_alloc_count, ua->socket_free_count, "socket registration leaked"); + uasync_destroy(ua, 0); ASSERT_EQ(u_get_allocated_count(), allocated, "HUP memory leak"); + TEST_PASS(); + + TEST_START("HUP keeps owner pause and pending write independent"); + ua = uasync_create(); ASSERT_TRUE(ua != NULL, "create ua"); ua->use_epoll = backend; + reset_counters(); + ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair"); + for (int i = 0; i < 2; i++) ASSERT_EQ(socket_set_nonblocking(sv[i]), 0, "nonblocking"); + tc = tcp_conn_create(ua, sv[0], 64, 64, 4, 1, 0, on_fin, on_error, NULL); + ASSERT_TRUE(tc != NULL, "create tcp"); tcp_conn_pause_read(tc); + ASSERT_EQ(push_write(tc, bytes, 64), 0, "write while owner paused"); + uint8_t reply[64]; ASSERT_EQ(recv(sv[1], reply, sizeof(reply), 0), sizeof(reply), "paused read blocked write"); + ASSERT_EQ(memcmp(bytes, reply, sizeof(reply)), 0, "write corrupted"); + ASSERT_EQ(send(sv[1], bytes, 64, MSG_NOSIGNAL), 64, "peer send"); close(sv[1]); + for (int i = 0; i < 30; i++) uasync_poll(ua, 0); + ASSERT_TRUE(tc->read_stopped && tc->hup_received && !tc->socket_id, "manual pause not parked"); + ASSERT_EQ(tc->read_queue->count, 0, "HUP undid manual pause"); + tcp_conn_destroy(tc); uasync_drain_immediate(ua); uasync_destroy(ua, 0); + ASSERT_EQ(u_get_allocated_count(), allocated, "manual HUP memory leak"); TEST_PASS(); + } +} + #endif int main(void) { @@ -866,6 +924,7 @@ int main(void) { #endif test_double_push_close(); test_fin_before_close(); + test_hup_backpressure(); DEBUG_INFO(DEBUG_CATEGORY_SYS, "=== Test Statistics ==="); DEBUG_INFO(DEBUG_CATEGORY_SYS, "Tests run: %d", tests_run);