jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Honor TCP read backpressure across HUP and restore polling

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 13:28:44 +02:00
1 parent d5444fb9bc
commit 4da9d85545
3 files changed
+123 -47
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lib/tcp_io.c
+62 -47
View File
@@ -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);
}
// ====================================================================
lib/tcp_io.h
+2
View File
@@ -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;
tests/test_tcp_io.c
+59
View File
@@ -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);