From b15632730f263e4586e84ba9c2f19d2f7d45a82d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=84=9A=E8=80=85?= Date: Mon, 29 Jun 2026 08:38:54 +0800 Subject: [PATCH] fix(fasthttp): rearm keep-alive poll events (#27585) --- vlib/fasthttp/fasthttp_bsd.c.v | 4 +- .../fasthttp/fasthttp_bsd_regression_test.c.v | 61 +++++++++++++++++++ vlib/fasthttp/fasthttp_linux.v | 27 ++++---- .../fasthttp/fasthttp_linux_regression_test.v | 58 ++++++++++++++++++ 4 files changed, 136 insertions(+), 14 deletions(-) create mode 100644 vlib/fasthttp/fasthttp_bsd_regression_test.c.v create mode 100644 vlib/fasthttp/fasthttp_linux_regression_test.v diff --git a/vlib/fasthttp/fasthttp_bsd.c.v b/vlib/fasthttp/fasthttp_bsd.c.v index 2c1fd9d8c..19ad15868 100644 --- a/vlib/fasthttp/fasthttp_bsd.c.v +++ b/vlib/fasthttp/fasthttp_bsd.c.v @@ -307,6 +307,7 @@ fn complete_response(server &Server, kq int, c_ptr voidptr, mut clients map[int] } c.read_start = 0 c.should_close = false + add_event(kq, u64(c.fd), i16(C.EVFILT_READ), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), c) } // process_request handles a complete HTTP request: decodes, calls the handler, @@ -352,7 +353,8 @@ fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]vo match resp.takeover_mode { .manual { // The handler has taken ownership of the connection. - // Remove from kqueue and tracking, but do NOT close the fd. + // Remove from kqueue and tracking before ending the request, but do + // NOT close the fd. clients.delete(c.fd) delete_event(kq, u64(c.fd), i16(C.EVFILT_READ), c) delete_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), c) diff --git a/vlib/fasthttp/fasthttp_bsd_regression_test.c.v b/vlib/fasthttp/fasthttp_bsd_regression_test.c.v new file mode 100644 index 000000000..3d6ca41e6 --- /dev/null +++ b/vlib/fasthttp/fasthttp_bsd_regression_test.c.v @@ -0,0 +1,61 @@ +// vtest build: macos || freebsd || openbsd || netbsd || dragonfly +module fasthttp + +fn C.socketpair(domain i32, typ i32, protocol i32, sockets &i32) i32 + +fn bsd_reregistration_test_handler(_ HttpRequest) !HttpResponse { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n'.bytes() + } +} + +fn test_keep_alive_completion_rearms_kqueue_read_after_consumed_edge() ! { + server := new_server(ServerConfig{ + family: .ip + port: 0 + max_request_buffer_size: 8192 + handler: bsd_reregistration_test_handler + })! + kq := C.kqueue() + assert kq >= 0 + defer { + C.close(kq) + } + mut sockets := [2]int{} + assert C.socketpair(C.AF_UNIX, C.SOCK_STREAM, 0, &sockets[0]) == 0 + defer { + C.close(sockets[0]) + C.close(sockets[1]) + } + set_nonblocking(sockets[0]) + mut conn := &Conn{ + fd: sockets[0] + request_active: true + file_fd: -1 + } + assert add_event(kq, u64(sockets[0]), i16(C.EVFILT_READ), + u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), conn) == 0 + assert C.write(sockets[1], c'GET ', 4) == 4 + + mut event := C.kevent{} + mut timeout := C.timespec{ + tv_sec: 1 + } + assert C.kevent(kq, unsafe { nil }, 0, &event, 1, &timeout) == 1 + assert event.ident == u64(sockets[0]) + + server.begin_request() + mut clients := { + sockets[0]: voidptr(conn) + } + complete_response(server, kq, conn, mut clients, false) + + assert server.active_request_count() == 0 + assert clients[sockets[0]] or { unsafe { nil } } == voidptr(conn) + timeout = C.timespec{ + tv_sec: 1 + } + assert C.kevent(kq, unsafe { nil }, 0, &event, 1, &timeout) == 1 + assert event.ident == u64(sockets[0]) + close_conn(server, kq, conn, mut clients) +} diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index 2a63c2db3..753edd59d 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -424,7 +424,6 @@ fn arm_epollout(epoll_fd int, client_fd int) int { fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) { state := client_write_states[client_fd] or { return } should_close := state.should_close - epollout_was_armed := state.epollout_armed free_write_state(server, client_fd, mut client_write_states) client_buffers.delete(client_fd) if server.is_shutting_down() || should_close { @@ -432,17 +431,17 @@ fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds ma client_read_starts, mut closing_client_fds, mut client_write_states) return } - // Keep-alive: if EPOLLOUT was registered for the write, drop the fd from - // epoll and re-add it with the original EPOLLIN|EPOLLET mask. The re-add - // causes the kernel to fire a fresh edge for any pipelined request bytes - // that piled up in the recv buffer while we were write-blocked; a plain - // EPOLL_CTL_MOD would not generate that edge. - if epollout_was_armed { - remove_fd_from_epoll(epoll_fd, client_fd) - if add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 { - handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut - client_read_starts, mut closing_client_fds, mut client_write_states) - } + // Keep-alive: drop the fd from epoll and re-add it with the original + // EPOLLIN|EPOLLET mask. The re-add causes the kernel to fire a fresh edge + // for any pipelined request bytes that piled up in the recv buffer while the + // response was being written; a plain EPOLL_CTL_MOD would not generate that + // edge. Do this even when EPOLLOUT was never armed, because the inline write + // path can also consume the current EPOLLIN edge before pipelined bytes are + // observed by the loop again. + remove_fd_from_epoll(epoll_fd, client_fd) + if add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 { + handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) } } @@ -511,7 +510,8 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ match response.takeover_mode { .manual { // The handler has taken ownership of the connection. - // Remove from epoll and tracking, but do NOT close the fd. + // Remove from epoll and tracking before ending the request, but do + // NOT close the fd. client_fds.delete(client_fd) client_buffers.delete(client_fd) client_read_starts.delete(client_fd) @@ -521,6 +521,7 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ response.free_owned_content() if request_active { server.end_request() + request_active = false } return } diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v new file mode 100644 index 000000000..fb7837ca4 --- /dev/null +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -0,0 +1,58 @@ +// vtest build: linux +module fasthttp + +fn C.socketpair(domain i32, typ i32, protocol i32, sockets &i32) i32 + +fn linux_reregistration_test_handler(_ HttpRequest) !HttpResponse { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n'.bytes() + } +} + +fn test_keep_alive_completion_rearms_epollin_after_consumed_edge() ! { + server := new_server(ServerConfig{ + family: .ip + port: 0 + max_request_buffer_size: 8192 + handler: linux_reregistration_test_handler + })! + epoll_fd := C.epoll_create1(0) + assert epoll_fd >= 0 + defer { + C.close(epoll_fd) + } + mut sockets := [2]int{} + assert C.socketpair(C.AF_UNIX, C.SOCK_STREAM, 0, &sockets[0]) == 0 + defer { + C.close(sockets[0]) + C.close(sockets[1]) + } + set_blocking(sockets[0], false) + assert add_fd_to_epoll(epoll_fd, sockets[0], u32(C.EPOLLIN | C.EPOLLET)) == 0 + assert C.write(sockets[1], c'GET ', 4) == 4 + + mut event := C.epoll_event{} + assert C.epoll_wait(epoll_fd, &event, 1, 1000) == 1 + assert unsafe { event.data.fd } == sockets[0] + + server.begin_request() + mut state := &ClientWriteState{ + request_active: true + } + mut client_fds := { + sockets[0]: true + } + mut client_buffers := map[int][]u8{} + mut client_read_starts := map[int]i64{} + mut closing_client_fds := map[int]bool{} + mut client_write_states := { + sockets[0]: state + } + complete_write(server, epoll_fd, sockets[0], mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) + + assert server.active_request_count() == 0 + assert sockets[0] in client_fds + assert C.epoll_wait(epoll_fd, &event, 1, 1000) == 1 + assert unsafe { event.data.fd } == sockets[0] +} -- 2.39.5