From b48a62cf56d5831dfa4144f80654d5d597db116b Mon Sep 17 00:00:00 2001 From: weigang Date: Wed, 17 Jun 2026 13:41:42 +0800 Subject: [PATCH 01/12] fix: improve fasthttp bsd processing using loop command channel --- vlib/fasthttp/fasthttp_bsd.c.v | 199 ++++++++++++++++++++++++--------- vlib/fasthttp/fasthttp_test.v | 68 +++++++++++ 2 files changed, 214 insertions(+), 53 deletions(-) diff --git a/vlib/fasthttp/fasthttp_bsd.c.v b/vlib/fasthttp/fasthttp_bsd.c.v index 2c1fd9d8c54eec..98e3bd70a7c0f7 100644 --- a/vlib/fasthttp/fasthttp_bsd.c.v +++ b/vlib/fasthttp/fasthttp_bsd.c.v @@ -127,6 +127,7 @@ mut: file_pos i64 should_close bool request_arena voidptr + processing bool } fn (mut c Conn) free_write_buf() { @@ -277,41 +278,21 @@ fn send_request_timeout(fd int) { C.send(fd, status_408_response.data, status_408_response.len, send_flags) } -fn handle_write(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { - if send_pending(c_ptr) { - return - } - complete_response(server, kq, c_ptr, mut clients, true) +enum LoopCommandKind { + close_conn + complete_response + arm_write + manual_takeover + enable_read } -fn complete_response(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr, remove_write_event bool) { - mut c := unsafe { &Conn(c_ptr) } - if remove_write_event { - delete_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), c) - } - if server.is_shutting_down() || c.should_close { - close_conn(server, kq, c_ptr, mut clients) - return - } - if c.request_active { - server.end_request() - c.request_active = false - } - c.free_write_buf() - c.free_request_arena() - c.write_pos = 0 - c.read_len = 0 - if c.read_extra.cap > 0 { - unsafe { c.read_extra.free() } - c.read_extra = []u8{} - } - c.read_start = 0 - c.should_close = false +struct LoopCommand { + kind LoopCommandKind + c_ptr voidptr + remove_write_event bool } -// process_request handles a complete HTTP request: decodes, calls the handler, -// sends the response (or handles takeover/sendfile). -fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { +fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { mut c := unsafe { &Conn(c_ptr) } mut request_arena := voidptr(unsafe { nil }) $if prealloc { @@ -327,21 +308,25 @@ fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]vo mut decoded := decode_http_request(req_buf) or { send_bad_request(c.fd) end_request_arena_current_thread(request_arena) - close_conn(server, kq, c_ptr, mut clients) + command_ch <- LoopCommand{ + kind: .close_conn + c_ptr: c_ptr + } return } $if trace_prealloc ? { unsafe { prealloc_scope_checkpoint(c'fasthttp decoded request') } } - server.begin_request() - c.request_active = true decoded.client_conn_fd = c.fd decoded.user_data = server.user_data mut resp := server.request_handler(decoded) or { send_bad_request(c.fd) end_request_arena_current_thread(request_arena) - close_conn(server, kq, c_ptr, mut clients) + command_ch <- LoopCommand{ + kind: .close_conn + c_ptr: c_ptr + } return } $if trace_prealloc ? { @@ -351,18 +336,16 @@ 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. - 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) if c.request_active { server.end_request() c.request_active = false } resp.free_owned_content() resp.abandon_request_arena_current_thread() - unsafe { free(c_ptr) } + command_ch <- LoopCommand{ + kind: .manual_takeover + c_ptr: c_ptr + } return } .reusable { @@ -377,7 +360,15 @@ fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]vo resp.free_owned_content() resp.end_request_arena_current_thread() if server.is_shutting_down() || resp.should_close { - close_conn(server, kq, c_ptr, mut clients) + command_ch <- LoopCommand{ + kind: .close_conn + c_ptr: c_ptr + } + } else { + command_ch <- LoopCommand{ + kind: .enable_read + c_ptr: c_ptr + } } return } @@ -394,15 +385,15 @@ fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]vo } leave_request_arena_current_thread(c.request_arena) if resp.file_path != '' { - fd := C.open(resp.file_path.str, C.O_RDONLY, 0) - if fd != -1 { + file_fd := C.open(resp.file_path.str, C.O_RDONLY, 0) + if file_fd != -1 { mut st := C.stat{} - if C.fstat(fd, &st) == 0 { - c.file_fd = fd + if C.fstat(file_fd, &st) == 0 { + c.file_fd = file_fd c.file_len = st.st_size c.file_pos = 0 } else { - C.close(fd) + C.close(file_fd) } } } @@ -413,11 +404,105 @@ fn process_request(server &Server, kq int, c_ptr voidptr, mut clients map[int]vo c.read_start = 0 if send_pending(c_ptr) { - add_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), c) + command_ch <- LoopCommand{ + kind: .arm_write + c_ptr: c_ptr + } return } - complete_response(server, kq, c_ptr, mut clients, false) + command_ch <- LoopCommand{ + kind: .complete_response + c_ptr: c_ptr + remove_write_event: false + } +} + +fn dispatch_request_async(server &Server, kq int, c_ptr voidptr, command_ch chan LoopCommand) { + mut c := unsafe { &Conn(c_ptr) } + c.processing = true + server.begin_request() + c.request_active = true + delete_event(kq, u64(c.fd), i16(C.EVFILT_READ), c) + delete_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), c) + spawn process_request(server, c_ptr, command_ch) +} + +fn handle_write(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { + if send_pending(c_ptr) { + return + } + complete_response(server, kq, c_ptr, mut clients, true) +} + +fn complete_response(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr, remove_write_event bool) { + mut c := unsafe { &Conn(c_ptr) } + if remove_write_event { + delete_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), c) + } + if server.is_shutting_down() || c.should_close { + close_conn(server, kq, c_ptr, mut clients) + return + } + if c.request_active { + server.end_request() + c.request_active = false + } + c.free_write_buf() + c.free_request_arena() + c.write_pos = 0 + c.read_len = 0 + if c.read_extra.cap > 0 { + unsafe { c.read_extra.free() } + c.read_extra = []u8{} + } + c.read_start = 0 + c.should_close = false + c.processing = false + add_event(kq, u64(c.fd), i16(C.EVFILT_READ), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), c) +} + +fn process_loop_command(server &Server, kq int, cmd LoopCommand, mut clients map[int]voidptr) { + match cmd.kind { + .close_conn { + close_conn(server, kq, cmd.c_ptr, mut clients) + } + .complete_response { + complete_response(server, kq, cmd.c_ptr, mut clients, cmd.remove_write_event) + } + .arm_write { + mut c := unsafe { &Conn(cmd.c_ptr) } + c.processing = false + add_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), + c) + } + .manual_takeover { + mut c := unsafe { &Conn(cmd.c_ptr) } + 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) + unsafe { free(cmd.c_ptr) } + } + .enable_read { + mut c := unsafe { &Conn(cmd.c_ptr) } + c.processing = false + add_event(kq, u64(c.fd), i16(C.EVFILT_READ), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), + c) + } + } +} + +fn drain_loop_commands(server &Server, kq int, command_ch chan LoopCommand, mut clients map[int]voidptr) { + for { + select { + cmd := <-command_ch { + process_loop_command(server, kq, cmd, mut clients) + } + else { + break + } + } + } } // total_read_len returns the total number of request bytes received so far, @@ -439,7 +524,7 @@ fn (c &Conn) get_full_request_data() []u8 { return req_buf } -fn handle_read(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { +fn handle_read(server &Server, kq int, c_ptr voidptr, command_ch chan LoopCommand, mut clients map[int]voidptr) { mut c := unsafe { &Conn(c_ptr) } // Drain the socket for this kqueue notification. EV_CLEAR only rearms once @@ -535,7 +620,7 @@ fn handle_read(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidpt return } - process_request(server, kq, c_ptr, mut clients) + dispatch_request_async(server, kq, c_ptr, command_ch) } fn accept_clients(kq int, listen_fd int, mut clients map[int]voidptr) { @@ -637,7 +722,9 @@ pub fn (mut s Server) run() ! { mut events := [kqueue_max_events]C.kevent{} mut clients := map[int]voidptr{} + command_ch := chan LoopCommand{cap: kqueue_max_events * 4} for { + drain_loop_commands(s, s.poll_fd, command_ch, mut clients) if s.is_shutting_down() && s.active_request_count() == 0 { close_all_conns(s, s.poll_fd, mut clients) break @@ -685,6 +772,11 @@ pub fn (mut s Server) run() ! { continue } + c := unsafe { &Conn(event.udata) } + if c.processing { + continue + } + if event.flags & u16(C.EV_EOF) != 0 { close_conn(s, s.poll_fd, event.udata, mut clients) continue @@ -695,11 +787,12 @@ pub fn (mut s Server) run() ! { close_conn(s, s.poll_fd, event.udata, mut clients) continue } - handle_read(s, s.poll_fd, event.udata, mut clients) + handle_read(s, s.poll_fd, event.udata, command_ch, mut clients) } else if event.filter == i16(C.EVFILT_WRITE) { handle_write(s, s.poll_fd, event.udata, mut clients) } } + drain_loop_commands(s, s.poll_fd, command_ch, mut clients) // Sweep for connections waiting for body data that have timed out if s.timeout_in_seconds > 0 { now := time.sys_mono_now() @@ -707,7 +800,7 @@ pub fn (mut s Server) run() ! { for client_fd in clients.keys() { c_ptr := clients[client_fd] or { continue } c := unsafe { &Conn(c_ptr) } - if c.read_start > 0 && c.read_len > 0 && !c.request_active { + if c.read_start > 0 && c.read_len > 0 && !c.request_active && !c.processing { elapsed := now - c.read_start if elapsed >= timeout_ns { send_request_timeout(c.fd) diff --git a/vlib/fasthttp/fasthttp_test.v b/vlib/fasthttp/fasthttp_test.v index b2fc94901b299f..dff465463ac068 100644 --- a/vlib/fasthttp/fasthttp_test.v +++ b/vlib/fasthttp/fasthttp_test.v @@ -1,12 +1,15 @@ module fasthttp import net +import net.http import os import time const fasthttp_example_exe = os.join_path(os.cache_dir(), 'fasthttp_example_test.exe') const reusable_takeover_port = 13019 const reusable_takeover_addr = '127.0.0.1:${reusable_takeover_port}' +const loopback_request_port = 13020 +const loopback_request_addr = '127.0.0.1:${loopback_request_port}' fn testsuite_begin() { // Clean up old example binary if it exists @@ -195,6 +198,71 @@ fn test_response_takeover_mode_reusable_keeps_connection() { } } +fn test_handler_can_make_loopback_request_to_same_server() { + $if linux || bsd { + mut server := new_server(ServerConfig{ + family: .ip + port: loopback_request_port + timeout_in_seconds: 2 + max_request_buffer_size: 8192 + handler: loopback_request_handler + }) or { + assert false, 'Failed to create server: ${err}' + return + } + handle := server.handle() + spawn server.run() + handle.wait_till_running(max_retries: 1000, retry_period_ms: 10) or { + assert false, 'server did not start: ${err}' + return + } + defer { + handle.shutdown(timeout: 5 * time.second) or {} + } + + resp := http.fetch( + method: .get + url: 'http://${loopback_request_addr}/outer' + read_timeout: 2 * time.second + write_timeout: 2 * time.second + disable_connection_reuse: true + ) or { + assert false, 'loopback request failed: ${err}' + return + } + assert resp.status_code == 200 + assert resp.body == 'outer:inner' + } $else { + return + } +} + +fn loopback_request_handler(req HttpRequest) !HttpResponse { + path := req.buffer[req.path.start..req.path.start + req.path.len].bytestr() + if path == '/inner' { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\ninner'.bytes() + } + } + if path == '/outer' { + inner := http.fetch( + method: .get + url: 'http://${loopback_request_addr}/inner' + read_timeout: time.second + write_timeout: time.second + disable_connection_reuse: true + )! + body := 'outer:${inner.body}' + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: ${body.len}\r\n\r\n${body}'.bytes() + } + } + return HttpResponse{ + content: 'HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n'.bytes() + should_close: true + } +} + fn reusable_takeover_handler(req HttpRequest) !HttpResponse { path := req.buffer[req.path.start..req.path.start + req.path.len].bytestr() if path == '/reusable' { From c7222f8a61cee609ecece7b751083ab86dbd8c0a Mon Sep 17 00:00:00 2001 From: weigang Date: Fri, 19 Jun 2026 09:10:00 +0800 Subject: [PATCH 02/12] fix(fasthttp): fix manual takeover shutdown race and guard test to BSD --- vlib/fasthttp/fasthttp_bsd.c.v | 8 ++++---- vlib/fasthttp/fasthttp_test.v | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/vlib/fasthttp/fasthttp_bsd.c.v b/vlib/fasthttp/fasthttp_bsd.c.v index 98e3bd70a7c0f7..094cc7fcd2f9ae 100644 --- a/vlib/fasthttp/fasthttp_bsd.c.v +++ b/vlib/fasthttp/fasthttp_bsd.c.v @@ -336,10 +336,6 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { match resp.takeover_mode { .manual { - if c.request_active { - server.end_request() - c.request_active = false - } resp.free_owned_content() resp.abandon_request_arena_current_thread() command_ch <- LoopCommand{ @@ -478,6 +474,10 @@ fn process_loop_command(server &Server, kq int, cmd LoopCommand, mut clients map } .manual_takeover { mut c := unsafe { &Conn(cmd.c_ptr) } + if c.request_active { + server.end_request() + c.request_active = false + } 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_test.v b/vlib/fasthttp/fasthttp_test.v index dff465463ac068..8837bac976c636 100644 --- a/vlib/fasthttp/fasthttp_test.v +++ b/vlib/fasthttp/fasthttp_test.v @@ -199,7 +199,7 @@ fn test_response_takeover_mode_reusable_keeps_connection() { } fn test_handler_can_make_loopback_request_to_same_server() { - $if linux || bsd { + $if bsd { mut server := new_server(ServerConfig{ family: .ip port: loopback_request_port From 68402cd3b2ffe959bf24c217649a0ddb9c951a19 Mon Sep 17 00:00:00 2001 From: weigang Date: Fri, 19 Jun 2026 10:29:59 +0800 Subject: [PATCH 03/12] refactor(fasthttp): make linux backend asynchronous and fix loopback race --- vlib/fasthttp/fasthttp_linux.v | 184 +++++++++++++++++++++++++-------- vlib/fasthttp/fasthttp_test.v | 2 +- 2 files changed, 141 insertions(+), 45 deletions(-) diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index 2a63c2db340932..64e925e0bc53c5 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -39,6 +39,20 @@ mut: data C.epoll_data_t } +enum LoopCommandKind { + close_conn + complete_response + arm_write + manual_takeover + enable_read +} + +struct LoopCommand { + kind LoopCommandKind + client_fd int + state &ClientWriteState = unsafe { nil } +} + pub struct Server { pub: family net.AddrFamily = .ip6 @@ -439,10 +453,10 @@ fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds ma // 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) - } + } + 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) } } @@ -465,12 +479,11 @@ fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[ } } -fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer []u8, 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) { +fn process_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand) { mut request_arena := voidptr(unsafe { nil }) $if prealloc { request_arena = unsafe { prealloc_scope_begin() } } - client_read_starts.delete(client_fd) server.begin_request() mut request_active := true @@ -482,8 +495,10 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ if request_active { server.end_request() } - 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) + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + } return } $if trace_prealloc ? { @@ -499,8 +514,10 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ if request_active { server.end_request() } - 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) + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + } return } $if trace_prealloc ? { @@ -510,32 +527,32 @@ 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. - client_fds.delete(client_fd) - client_buffers.delete(client_fd) - client_read_starts.delete(client_fd) - closing_client_fds.delete(client_fd) - remove_fd_from_epoll(epoll_fd, client_fd) response.abandon_request_arena_current_thread() response.free_owned_content() - if request_active { - server.end_request() + command_ch <- LoopCommand{ + kind: .manual_takeover + client_fd: client_fd } return } .reusable { set_blocking(client_fd, false) - client_buffers.delete(client_fd) response.free_owned_content() response.end_request_arena_current_thread() if request_active { server.end_request() + request_active = false } if server.is_shutting_down() || response.should_close { - 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) + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + } + } else { + command_ch <- LoopCommand{ + kind: .enable_read + client_fd: client_fd + } } return } @@ -555,8 +572,10 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ if request_active { server.end_request() } - 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) + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + } return } mut st := C.stat{} @@ -568,8 +587,10 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ if request_active { server.end_request() } - 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) + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + } return } file_fd = fd @@ -597,30 +618,98 @@ fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer [ } // From here on, the state owns end_request bookkeeping. request_active = false - client_write_states[client_fd] = state status := try_drain_write(client_fd, mut state) match status { .done { - complete_write(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut - client_read_starts, mut closing_client_fds, mut client_write_states) + command_ch <- LoopCommand{ + kind: .complete_response + client_fd: client_fd + state: state + } } .blocked { - // Kernel send buffer is full. Park the state and resume on EPOLLOUT. - client_buffers.delete(client_fd) - client_read_starts.delete(client_fd) - if arm_epollout(epoll_fd, client_fd) == -1 { - eprintln('ERROR: epoll_ctl(MOD, EPOLLOUT) failed errno=${C.errno}') - handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut + command_ch <- LoopCommand{ + kind: .arm_write + client_fd: client_fd + state: state + } + } + .failed { + command_ch <- LoopCommand{ + kind: .close_conn + client_fd: client_fd + state: state + } + } + } +} + +fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, 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) { + match cmd.kind { + .close_conn { + if cmd.state != unsafe { nil } { + free_write_state(server, cmd.client_fd, mut client_write_states) + } + handle_client_closure(server, epoll_fd, cmd.client_fd, mut client_fds, mut + client_buffers, mut client_read_starts, mut closing_client_fds, mut + client_write_states) + } + .complete_response { + if cmd.state != unsafe { nil } { + client_write_states[cmd.client_fd] = cmd.state + complete_write(server, epoll_fd, cmd.client_fd, mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) + } + } + .arm_write { + if cmd.state != unsafe { nil } { + mut state := cmd.state + client_write_states[cmd.client_fd] = state + client_buffers.delete(cmd.client_fd) + client_read_starts.delete(cmd.client_fd) + if add_fd_to_epoll(epoll_fd, cmd.client_fd, u32(C.EPOLLIN | C.EPOLLOUT | C.EPOLLET)) == -1 { + eprintln('ERROR: epoll_ctl(ADD, EPOLLOUT) failed errno=${C.errno}') + handle_client_closure(server, epoll_fd, cmd.client_fd, mut client_fds, mut + client_buffers, mut client_read_starts, mut closing_client_fds, mut + client_write_states) + } else { + state.epollout_armed = true + } + } + } + .manual_takeover { + server.end_request() + client_fds.delete(cmd.client_fd) + client_buffers.delete(cmd.client_fd) + client_read_starts.delete(cmd.client_fd) + closing_client_fds.delete(cmd.client_fd) + remove_fd_from_epoll(epoll_fd, cmd.client_fd) + if cmd.state != unsafe { nil } { + free_write_state(server, cmd.client_fd, mut client_write_states) + } + } + .enable_read { + client_buffers.delete(cmd.client_fd) + if add_fd_to_epoll(epoll_fd, cmd.client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 { + handle_client_closure(server, epoll_fd, cmd.client_fd, mut client_fds, mut client_buffers, mut client_read_starts, mut closing_client_fds, mut client_write_states) - } else { - state.epollout_armed = true } } - .failed { - 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) + } +} + +fn drain_loop_commands(server &Server, epoll_fd int, command_ch chan LoopCommand, 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) { + for { + select { + cmd := <-command_ch { + process_loop_command(server, epoll_fd, cmd, mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) + } + else { + break + } } } } @@ -633,10 +722,13 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { mut client_read_starts := map[int]i64{} mut closing_client_fds := map[int]bool{} mut client_write_states := map[int]&ClientWriteState{} + command_ch := chan LoopCommand{cap: max_connection_size * 4} unsafe { request_buffer.flags.set(.noslices | .nogrow | .noshrink) } for { + drain_loop_commands(server, epoll_fd, command_ch, mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) if server.is_shutting_down() && server.active_request_count() == 0 { close_worker_clients(server, epoll_fd, mut client_fds, mut client_buffers, mut client_read_starts, mut closing_client_fds, mut client_write_states) @@ -784,9 +876,11 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { continue } if request_complete { - process_request(server, epoll_fd, client_fd, readed_request_buffer, mut - client_fds, mut client_buffers, mut client_read_starts, mut - closing_client_fds, mut client_write_states) + remove_fd_from_epoll(epoll_fd, client_fd) + client_read_starts.delete(client_fd) + req_buf := readed_request_buffer.clone() + client_buffers.delete(client_fd) + spawn process_request_async(server, client_fd, req_buf, command_ch) } else if recv_error { // Unexpected recv error - send 444 No Response C.send(client_fd, status_444_response.data, status_444_response.len, @@ -804,6 +898,8 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { } } } + drain_loop_commands(server, epoll_fd, command_ch, mut client_fds, mut client_buffers, mut + client_read_starts, mut closing_client_fds, mut client_write_states) if server.timeout_in_seconds > 0 { now := time.sys_mono_now() timeout_ns := i64(server.timeout_in_seconds) * 1_000_000_000 diff --git a/vlib/fasthttp/fasthttp_test.v b/vlib/fasthttp/fasthttp_test.v index 8837bac976c636..dff465463ac068 100644 --- a/vlib/fasthttp/fasthttp_test.v +++ b/vlib/fasthttp/fasthttp_test.v @@ -199,7 +199,7 @@ fn test_response_takeover_mode_reusable_keeps_connection() { } fn test_handler_can_make_loopback_request_to_same_server() { - $if bsd { + $if linux || bsd { mut server := new_server(ServerConfig{ family: .ip port: loopback_request_port From 7845d39e1855ca5c97248e382b0d5336b0f55ef2 Mon Sep 17 00:00:00 2001 From: weigang Date: Sat, 20 Jun 2026 10:58:53 +0800 Subject: [PATCH 04/12] fix(fasthttp): close Linux async request races --- vlib/fasthttp/fasthttp_linux.v | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index 64e925e0bc53c5..bbfeb8584f41dd 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -484,7 +484,6 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com $if prealloc { request_arena = unsafe { prealloc_scope_begin() } } - server.begin_request() mut request_active := true mut decoded_http_request := decode_http_request(request_buffer) or { @@ -649,6 +648,8 @@ fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, mut clien match cmd.kind { .close_conn { if cmd.state != unsafe { nil } { + // Failed first writes have not registered their state yet. + client_write_states[cmd.client_fd] = cmd.state free_write_state(server, cmd.client_fd, mut client_write_states) } handle_client_closure(server, epoll_fd, cmd.client_fd, mut client_fds, mut @@ -880,6 +881,8 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { client_read_starts.delete(client_fd) req_buf := readed_request_buffer.clone() client_buffers.delete(client_fd) + // Account for the request before shutdown can observe the spawned work. + server.begin_request() spawn process_request_async(server, client_fd, req_buf, command_ch) } else if recv_error { // Unexpected recv error - send 444 No Response From 5de24dda047c55c8112abb7cf0b9ae2a19f60454 Mon Sep 17 00:00:00 2001 From: weigang Date: Sat, 20 Jun 2026 11:15:29 +0800 Subject: [PATCH 05/12] test(fasthttp): cover Linux async request lifecycle --- vlib/fasthttp/fasthttp_linux.v | 13 +++- .../fasthttp/fasthttp_linux_regression_test.v | 73 +++++++++++++++++++ 2 files changed, 83 insertions(+), 3 deletions(-) create mode 100644 vlib/fasthttp/fasthttp_linux_regression_test.v diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index bbfeb8584f41dd..ce5731536bfe53 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -480,6 +480,9 @@ fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[ } fn process_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand) { + $if fasthttp_test_delay_async_start ? { + time.sleep(200 * time.millisecond) + } mut request_arena := voidptr(unsafe { nil }) $if prealloc { request_arena = unsafe { prealloc_scope_begin() } @@ -644,6 +647,12 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com } } +fn dispatch_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand) { + // Account for the request before shutdown can observe the spawned work. + server.begin_request() + spawn process_request_async(server, client_fd, request_buffer, command_ch) +} + fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, 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) { match cmd.kind { .close_conn { @@ -881,9 +890,7 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { client_read_starts.delete(client_fd) req_buf := readed_request_buffer.clone() client_buffers.delete(client_fd) - // Account for the request before shutdown can observe the spawned work. - server.begin_request() - spawn process_request_async(server, client_fd, req_buf, command_ch) + dispatch_request_async(server, client_fd, req_buf, command_ch) } else if recv_error { // Unexpected recv error - send 444 No Response C.send(client_fd, status_444_response.data, status_444_response.len, diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v new file mode 100644 index 00000000000000..c823a6e0279deb --- /dev/null +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -0,0 +1,73 @@ +// vtest build: linux +// vtest vflags: -d fasthttp_test_delay_async_start +module fasthttp + +import net +import time + +fn lifecycle_test_handler(_ HttpRequest) !HttpResponse { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n'.bytes() + } +} + +fn new_lifecycle_test_server() !&Server { + return new_server(ServerConfig{ + family: .ip + port: 0 + max_request_buffer_size: 8192 + handler: lifecycle_test_handler + }) +} + +fn test_close_command_releases_unregistered_write_state() ! { + server := new_lifecycle_test_server()! + epoll_fd := C.epoll_create1(0) + assert epoll_fd >= 0 + defer { + C.close(epoll_fd) + } + client_fd := int(C.socket(i32(net.AddrFamily.ip), i32(net.SocketType.tcp), 0)) + assert client_fd >= 0 + assert add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN)) == 0 + + server.begin_request() + state := &ClientWriteState{ + content: 'pending'.bytes() + content_owned: true + request_active: true + } + mut client_fds := { + client_fd: 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 := map[int]&ClientWriteState{} + process_loop_command(server, epoll_fd, LoopCommand{ + kind: .close_conn + client_fd: client_fd + state: state + }, 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 client_fd !in client_fds + assert client_fd !in client_write_states +} + +fn test_dispatch_marks_request_active_before_spawn() ! { + server := new_lifecycle_test_server()! + command_ch := chan LoopCommand{cap: 1} + dispatch_request_async(server, -1, 'invalid request'.bytes(), command_ch) + assert server.active_request_count() == 1 + + select { + _ := <-command_ch { + assert server.active_request_count() == 0 + } + 1 * time.second { + assert false, 'timed out waiting for the async request to finish' + } + } +} From ef66e1b8b7a1e5c15324b4aa42f4406262f564a1 Mon Sep 17 00:00:00 2001 From: weigang Date: Sun, 21 Jun 2026 14:50:24 +0800 Subject: [PATCH 06/12] fix(fasthttp): close async poll lifecycle races --- vlib/fasthttp/fasthttp_bsd.c.v | 63 ++++++---- .../fasthttp/fasthttp_bsd_regression_test.c.v | 75 ++++++++++++ vlib/fasthttp/fasthttp_linux.v | 110 ++++++++++++------ .../fasthttp/fasthttp_linux_regression_test.v | 36 +++++- 4 files changed, 225 insertions(+), 59 deletions(-) create mode 100644 vlib/fasthttp/fasthttp_bsd_regression_test.c.v diff --git a/vlib/fasthttp/fasthttp_bsd.c.v b/vlib/fasthttp/fasthttp_bsd.c.v index 094cc7fcd2f9ae..5706165e02e5c1 100644 --- a/vlib/fasthttp/fasthttp_bsd.c.v +++ b/vlib/fasthttp/fasthttp_bsd.c.v @@ -17,6 +17,7 @@ const buf_size = max_connection_size const kqueue_max_events = 128 const backlog = max_connection_size const kqueue_wait_timeout_ms = 100 +const kqueue_command_ident = u64(0) // send_flags is OR'd into every C.send() call in this file. On OpenBSD, // which lacks the per-socket SO_NOSIGPIPE option, we pass MSG_NOSIGNAL @@ -292,7 +293,21 @@ struct LoopCommand { remove_write_event bool } -fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { +fn wake_event_loop(kq int) { + mut ev := C.kevent{} + ev_set(mut &ev, kqueue_command_ident, i16(C.EVFILT_USER), u16(0), u32(C.NOTE_TRIGGER), + isize(0), unsafe { nil }) + if C.kevent(kq, &ev, 1, unsafe { nil }, 0, unsafe { nil }) < 0 { + eprintln('ERROR: kqueue NOTE_TRIGGER failed with errno=${C.errno}') + } +} + +fn send_loop_command(command_ch chan LoopCommand, kq int, command LoopCommand) { + command_ch <- command + wake_event_loop(kq) +} + +fn process_request(server &Server, kq int, c_ptr voidptr, command_ch chan LoopCommand) { mut c := unsafe { &Conn(c_ptr) } mut request_arena := voidptr(unsafe { nil }) $if prealloc { @@ -308,10 +323,10 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { mut decoded := decode_http_request(req_buf) or { send_bad_request(c.fd) end_request_arena_current_thread(request_arena) - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .close_conn c_ptr: c_ptr - } + }) return } $if trace_prealloc ? { @@ -323,10 +338,10 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { mut resp := server.request_handler(decoded) or { send_bad_request(c.fd) end_request_arena_current_thread(request_arena) - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .close_conn c_ptr: c_ptr - } + }) return } $if trace_prealloc ? { @@ -338,10 +353,10 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { .manual { resp.free_owned_content() resp.abandon_request_arena_current_thread() - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .manual_takeover c_ptr: c_ptr - } + }) return } .reusable { @@ -349,22 +364,18 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { c.read_len = 0 c.read_extra.clear() c.read_start = 0 - if c.request_active { - server.end_request() - c.request_active = false - } resp.free_owned_content() resp.end_request_arena_current_thread() if server.is_shutting_down() || resp.should_close { - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .close_conn c_ptr: c_ptr - } + }) } else { - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .enable_read c_ptr: c_ptr - } + }) } return } @@ -400,18 +411,18 @@ fn process_request(server &Server, c_ptr voidptr, command_ch chan LoopCommand) { c.read_start = 0 if send_pending(c_ptr) { - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .arm_write c_ptr: c_ptr - } + }) return } - command_ch <- LoopCommand{ + send_loop_command(command_ch, kq, LoopCommand{ kind: .complete_response c_ptr: c_ptr remove_write_event: false - } + }) } fn dispatch_request_async(server &Server, kq int, c_ptr voidptr, command_ch chan LoopCommand) { @@ -421,7 +432,7 @@ fn dispatch_request_async(server &Server, kq int, c_ptr voidptr, command_ch chan c.request_active = true delete_event(kq, u64(c.fd), i16(C.EVFILT_READ), c) delete_event(kq, u64(c.fd), i16(C.EVFILT_WRITE), c) - spawn process_request(server, c_ptr, command_ch) + spawn process_request(server, kq, c_ptr, command_ch) } fn handle_write(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { @@ -485,6 +496,10 @@ fn process_loop_command(server &Server, kq int, cmd LoopCommand, mut clients map } .enable_read { mut c := unsafe { &Conn(cmd.c_ptr) } + if c.request_active { + server.end_request() + c.request_active = false + } c.processing = false add_event(kq, u64(c.fd), i16(C.EVFILT_READ), u16(C.EV_ADD | C.EV_ENABLE | C.EV_CLEAR), c) @@ -723,6 +738,10 @@ pub fn (mut s Server) run() ! { mut events := [kqueue_max_events]C.kevent{} mut clients := map[int]voidptr{} command_ch := chan LoopCommand{cap: kqueue_max_events * 4} + if add_event(s.poll_fd, kqueue_command_ident, i16(C.EVFILT_USER), u16(C.EV_ADD | C.EV_CLEAR), + unsafe { nil }) < 0 { + return error('failed to register kqueue command wakeup') + } for { drain_loop_commands(s, s.poll_fd, command_ch, mut clients) if s.is_shutting_down() && s.active_request_count() == 0 { @@ -749,6 +768,10 @@ pub fn (mut s Server) run() ! { for i := 0; i < nev; i++ { event := events[i] + if event.filter == i16(C.EVFILT_USER) && event.ident == kqueue_command_ident { + drain_loop_commands(s, s.poll_fd, command_ch, mut clients) + continue + } if event.flags & u16(C.EV_ERROR) != 0 { if event.ident == u64(listen_fd) { C.perror(c'listener error') 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 00000000000000..b46e0452cedfd4 --- /dev/null +++ b/vlib/fasthttp/fasthttp_bsd_regression_test.c.v @@ -0,0 +1,75 @@ +// vtest build: macos || freebsd || openbsd || netbsd || dragonfly +module fasthttp + +import net +import time + +fn bsd_lifecycle_test_handler(_ HttpRequest) !HttpResponse { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n'.bytes() + } +} + +fn new_bsd_lifecycle_test_server() !&Server { + return new_server(ServerConfig{ + family: .ip + port: 0 + max_request_buffer_size: 8192 + handler: bsd_lifecycle_test_handler + }) +} + +fn test_reusable_completion_ends_request_in_event_loop() ! { + server := new_bsd_lifecycle_test_server()! + kq := C.kqueue() + assert kq >= 0 + defer { + C.close(kq) + } + client_fd := int(C.socket(i32(net.AddrFamily.ip), i32(net.SocketType.tcp), 0)) + assert client_fd >= 0 + mut conn := &Conn{ + fd: client_fd + request_active: true + processing: true + file_fd: -1 + } + mut clients := { + client_fd: voidptr(conn) + } + server.begin_request() + process_loop_command(server, kq, LoopCommand{ + kind: .enable_read + c_ptr: conn + }, mut clients) + + assert server.active_request_count() == 0 + assert !conn.request_active + assert !conn.processing + close_conn(server, kq, conn, mut clients) +} + +fn test_loop_command_wakes_kqueue() ! { + kq := C.kqueue() + assert kq >= 0 + defer { + C.close(kq) + } + assert add_event(kq, kqueue_command_ident, i16(C.EVFILT_USER), u16(C.EV_ADD | C.EV_CLEAR), + unsafe { nil }) == 0 + command_ch := chan LoopCommand{cap: 1} + send_loop_command(command_ch, kq, LoopCommand{ + kind: .close_conn + }) + + mut event := C.kevent{} + timeout := C.timespec{ + tv_sec: 1 + } + watch := time.new_stopwatch() + assert C.kevent(kq, unsafe { nil }, 0, &event, 1, &timeout) == 1 + assert watch.elapsed() < 250 * time.millisecond + assert event.filter == i16(C.EVFILT_USER) + assert event.ident == kqueue_command_ident + _ := <-command_ch +} diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index ce5731536bfe53..fcd33588942f6c 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -5,6 +5,7 @@ import sync.stdatomic import time #include +#include #include #include #include @@ -20,6 +21,8 @@ fn C.epoll_ctl(__epfd i32, __op i32, __fd i32, __event &C.epoll_event) i32 fn C.epoll_wait(__epfd i32, __events &C.epoll_event, __maxevents i32, __timeout i32) i32 +fn C.eventfd(initval u32, flags i32) i32 + fn C.sendfile(out_fd i32, in_fd i32, offset &i64, count usize) i32 fn C.fstat(fd i32, buf &C.stat) i32 @@ -53,6 +56,23 @@ struct LoopCommand { state &ClientWriteState = unsafe { nil } } +fn wake_event_loop(wakeup_fd int) { + one := u64(1) + if C.write(wakeup_fd, &one, sizeof(one)) < 0 && C.errno != C.EAGAIN { + eprintln('ERROR: eventfd write failed with errno=${C.errno}') + } +} + +fn send_loop_command(command_ch chan LoopCommand, wakeup_fd int, command LoopCommand) { + command_ch <- command + wake_event_loop(wakeup_fd) +} + +fn drain_event_loop_wakeup(wakeup_fd int) { + mut value := u64(0) + for C.read(wakeup_fd, &value, sizeof(value)) > 0 {} +} + pub struct Server { pub: family net.AddrFamily = .ip6 @@ -479,7 +499,7 @@ fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[ } } -fn process_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand) { +fn process_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { $if fasthttp_test_delay_async_start ? { time.sleep(200 * time.millisecond) } @@ -497,10 +517,10 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com if request_active { server.end_request() } - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd - } + }) return } $if trace_prealloc ? { @@ -516,10 +536,10 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com if request_active { server.end_request() } - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd - } + }) return } $if trace_prealloc ? { @@ -531,10 +551,10 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com .manual { response.abandon_request_arena_current_thread() response.free_owned_content() - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .manual_takeover client_fd: client_fd - } + }) return } .reusable { @@ -546,15 +566,15 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com request_active = false } if server.is_shutting_down() || response.should_close { - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd - } + }) } else { - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .enable_read client_fd: client_fd - } + }) } return } @@ -574,10 +594,10 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com if request_active { server.end_request() } - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd - } + }) return } mut st := C.stat{} @@ -589,10 +609,10 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com if request_active { server.end_request() } - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd - } + }) return } file_fd = fd @@ -605,52 +625,50 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com content_bytes := response.take_or_clone_content() arena_ptr := response.take_request_arena() leave_request_arena_current_thread(arena_ptr) - - mut state := &ClientWriteState{ - content: content_bytes - content_pos: 0 - content_owned: true - file_fd: file_fd - file_len: file_len - file_pos: 0 - should_close: response.should_close - request_arena: arena_ptr - request_active: request_active - start_ns: time.sys_mono_now() - } + // Bypass the spawned thread's outer prealloc scope. The event loop owns this + // object after the worker exits, so it must be independently allocated. + mut state := unsafe { &ClientWriteState(C.calloc(1, sizeof(ClientWriteState))) } + state.content = content_bytes + state.content_owned = true + state.file_fd = file_fd + state.file_len = file_len + state.should_close = response.should_close + state.request_arena = arena_ptr + state.request_active = request_active + state.start_ns = time.sys_mono_now() // From here on, the state owns end_request bookkeeping. request_active = false status := try_drain_write(client_fd, mut state) match status { .done { - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .complete_response client_fd: client_fd state: state - } + }) } .blocked { - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .arm_write client_fd: client_fd state: state - } + }) } .failed { - command_ch <- LoopCommand{ + send_loop_command(command_ch, wakeup_fd, LoopCommand{ kind: .close_conn client_fd: client_fd state: state - } + }) } } } -fn dispatch_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand) { +fn dispatch_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { // Account for the request before shutdown can observe the spawned work. server.begin_request() - spawn process_request_async(server, client_fd, request_buffer, command_ch) + spawn process_request_async(server, client_fd, request_buffer, command_ch, wakeup_fd) } fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, 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) { @@ -733,6 +751,17 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { mut closing_client_fds := map[int]bool{} mut client_write_states := map[int]&ClientWriteState{} command_ch := chan LoopCommand{cap: max_connection_size * 4} + wakeup_fd := C.eventfd(0, C.EFD_NONBLOCK | C.EFD_CLOEXEC) + if wakeup_fd < 0 { + eprintln('ERROR: eventfd() failed with errno=${C.errno}') + return + } + defer { + close_socket(wakeup_fd) + } + if add_fd_to_epoll(epoll_fd, wakeup_fd, u32(C.EPOLLIN)) == -1 { + return + } unsafe { request_buffer.flags.set(.noslices | .nogrow | .noshrink) } @@ -757,6 +786,13 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { } for i := 0; i < num_events; i++ { client_fd := unsafe { events[i].data.fd } + if client_fd == wakeup_fd { + drain_event_loop_wakeup(wakeup_fd) + drain_loop_commands(server, epoll_fd, command_ch, mut client_fds, mut + client_buffers, mut client_read_starts, mut closing_client_fds, mut + client_write_states) + continue + } // Accept new connections when the listening socket is readable if client_fd == listen_fd { if server.is_shutting_down() { @@ -890,7 +926,7 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { client_read_starts.delete(client_fd) req_buf := readed_request_buffer.clone() client_buffers.delete(client_fd) - dispatch_request_async(server, client_fd, req_buf, command_ch) + dispatch_request_async(server, client_fd, req_buf, command_ch, wakeup_fd) } else if recv_error { // Unexpected recv error - send 444 No Response C.send(client_fd, status_444_response.data, status_444_response.len, diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v index c823a6e0279deb..d7d2e27a2740bf 100644 --- a/vlib/fasthttp/fasthttp_linux_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -1,5 +1,5 @@ // vtest build: linux -// vtest vflags: -d fasthttp_test_delay_async_start +// vtest vflags: -prealloc -d fasthttp_test_delay_async_start module fasthttp import net @@ -59,7 +59,12 @@ fn test_close_command_releases_unregistered_write_state() ! { fn test_dispatch_marks_request_active_before_spawn() ! { server := new_lifecycle_test_server()! command_ch := chan LoopCommand{cap: 1} - dispatch_request_async(server, -1, 'invalid request'.bytes(), command_ch) + wakeup_fd := C.eventfd(0, C.EFD_NONBLOCK | C.EFD_CLOEXEC) + assert wakeup_fd >= 0 + defer { + C.close(wakeup_fd) + } + dispatch_request_async(server, -1, 'invalid request'.bytes(), command_ch, wakeup_fd) assert server.active_request_count() == 1 select { @@ -71,3 +76,30 @@ fn test_dispatch_marks_request_active_before_spawn() ! { } } } + +fn test_loop_command_wakes_epoll() ! { + epoll_fd := C.epoll_create1(0) + assert epoll_fd >= 0 + defer { + C.close(epoll_fd) + } + wakeup_fd := C.eventfd(0, C.EFD_NONBLOCK | C.EFD_CLOEXEC) + assert wakeup_fd >= 0 + defer { + C.close(wakeup_fd) + } + assert add_fd_to_epoll(epoll_fd, wakeup_fd, u32(C.EPOLLIN)) == 0 + command_ch := chan LoopCommand{cap: 1} + send_loop_command(command_ch, wakeup_fd, LoopCommand{ + kind: .close_conn + client_fd: -1 + }) + + mut event := C.epoll_event{} + watch := time.new_stopwatch() + assert C.epoll_wait(epoll_fd, &event, 1, 1000) == 1 + assert watch.elapsed() < 250 * time.millisecond + assert event.data.fd == wakeup_fd + drain_event_loop_wakeup(wakeup_fd) + _ := <-command_ch +} From b7a287ce4ee843cdd0f39a325b479f84a85a2b71 Mon Sep 17 00:00:00 2001 From: weigang Date: Sun, 21 Jun 2026 14:53:02 +0800 Subject: [PATCH 07/12] test(fasthttp): silence epoll union warning --- vlib/fasthttp/fasthttp_linux_regression_test.v | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v index d7d2e27a2740bf..0bf1287b09beba 100644 --- a/vlib/fasthttp/fasthttp_linux_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -99,7 +99,8 @@ fn test_loop_command_wakes_epoll() ! { watch := time.new_stopwatch() assert C.epoll_wait(epoll_fd, &event, 1, 1000) == 1 assert watch.elapsed() < 250 * time.millisecond - assert event.data.fd == wakeup_fd + event_fd := unsafe { event.data.fd } + assert event_fd == wakeup_fd drain_event_loop_wakeup(wakeup_fd) _ := <-command_ch } From 5ec448364198fba65b6590f4b62b9dfb0356c3b9 Mon Sep 17 00:00:00 2001 From: weigang Date: Mon, 22 Jun 2026 14:47:29 +0800 Subject: [PATCH 08/12] fix(fasthttp): support takeover with musl and tcc CI --- vlib/fasthttp/fasthttp_linux.v | 1 - .../fasthttp_linux_prealloc_regression_test.v | 58 +++++++++++++++++++ .../fasthttp/fasthttp_linux_regression_test.v | 2 +- vlib/veb/veb_fasthttp.v | 9 ++- 4 files changed, 65 insertions(+), 5 deletions(-) create mode 100644 vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index fcd33588942f6c..683a78bc834aa6 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -712,7 +712,6 @@ fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, mut clien client_buffers.delete(cmd.client_fd) client_read_starts.delete(cmd.client_fd) closing_client_fds.delete(cmd.client_fd) - remove_fd_from_epoll(epoll_fd, cmd.client_fd) if cmd.state != unsafe { nil } { free_write_state(server, cmd.client_fd, mut client_write_states) } diff --git a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v new file mode 100644 index 00000000000000..d9113bf46c488e --- /dev/null +++ b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v @@ -0,0 +1,58 @@ +// vtest build: linux && !tinyc +// vtest vflags: -prealloc +module fasthttp + +import net +import time + +fn C.socketpair(domain i32, typ i32, protocol i32, sockets &i32) i32 + +fn prealloc_async_state_test_handler(_ HttpRequest) !HttpResponse { + return HttpResponse{ + content: 'HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK'.bytes() + } +} + +fn test_prealloc_async_write_state_outlives_worker() ! { + server := new_server(ServerConfig{ + family: .ip + port: 0 + max_request_buffer_size: 8192 + handler: prealloc_async_state_test_handler + })! + 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]) + } + wakeup_fd := C.eventfd(0, C.EFD_NONBLOCK | C.EFD_CLOEXEC) + assert wakeup_fd >= 0 + defer { + C.close(wakeup_fd) + } + command_ch := chan LoopCommand{cap: 1} + dispatch_request_async(server, sockets[0], 'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n'.bytes(), + command_ch, wakeup_fd) + + mut cmd := LoopCommand{} + select { + command := <-command_ch { + cmd = command + } + 1 * time.second { + assert false, 'timed out waiting for the async response' + return + } + } + // The worker must have exited before the event loop consumes its state. + time.sleep(50 * time.millisecond) + assert cmd.kind == .complete_response + assert cmd.state != unsafe { nil } + assert cmd.state.content.len > 0 + + mut states := map[int]&ClientWriteState{} + states[sockets[0]] = cmd.state + free_write_state(server, sockets[0], mut states) + assert server.active_request_count() == 0 +} diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v index 0bf1287b09beba..853e09222cbc18 100644 --- a/vlib/fasthttp/fasthttp_linux_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -1,5 +1,5 @@ // vtest build: linux -// vtest vflags: -prealloc -d fasthttp_test_delay_async_start +// vtest vflags: -d fasthttp_test_delay_async_start module fasthttp import net diff --git a/vlib/veb/veb_fasthttp.v b/vlib/veb/veb_fasthttp.v index 3a3cbcd4f94718..9d1cbaef565a7b 100644 --- a/vlib/veb/veb_fasthttp.v +++ b/vlib/veb/veb_fasthttp.v @@ -233,7 +233,10 @@ fn handle_request_and_route[A, X](mut app A, req http.Request, _client_fd int, p form: form files: files } - mut user_context := X{ + // A manual takeover handler may spawn work that outlives the async fasthttp + // request worker. Keep its context in the request arena/heap, not on that + // worker's stack. + mut user_context := &X{ Context: ctx } $if trace_prealloc ? { @@ -247,7 +250,7 @@ fn handle_request_and_route[A, X](mut app A, req http.Request, _client_fd int, p app.static_compression_mime_types, app.enable_markdown_negotiation), mut user_context, url, host) { - // Preserve the handled context on the heap before the stack-local user context goes away. + // Preserve the handled context in the base context returned to fasthttp. unsafe { *ctx = user_context.Context } @@ -268,7 +271,7 @@ fn handle_request_and_route[A, X](mut app A, req http.Request, _client_fd int, p $if trace_prealloc ? { unsafe { prealloc_scope_checkpoint(c'veb route returned') } } - // Preserve the handled context on the heap before the stack-local user context goes away. + // Preserve the handled context in the base context returned to fasthttp. unsafe { *ctx = user_context.Context } From 1d6739de089e90832ca1f04c2c6737299002c5ea Mon Sep 17 00:00:00 2001 From: weigang Date: Mon, 22 Jun 2026 14:51:48 +0800 Subject: [PATCH 09/12] test(fasthttp): clean prealloc regression imports --- vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v | 1 - 1 file changed, 1 deletion(-) diff --git a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v index d9113bf46c488e..db233e62cea6db 100644 --- a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v @@ -2,7 +2,6 @@ // vtest vflags: -prealloc module fasthttp -import net import time fn C.socketpair(domain i32, typ i32, protocol i32, sockets &i32) i32 From a3f5a1e5e5ab7044a8744a683ec2441637bcf7da Mon Sep 17 00:00:00 2001 From: weigang Date: Mon, 22 Jun 2026 14:58:13 +0800 Subject: [PATCH 10/12] test(fasthttp): wait for prealloc worker exit --- vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v index db233e62cea6db..9bbac76081b14d 100644 --- a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v @@ -31,8 +31,9 @@ fn test_prealloc_async_write_state_outlives_worker() ! { C.close(wakeup_fd) } command_ch := chan LoopCommand{cap: 1} - dispatch_request_async(server, sockets[0], 'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n'.bytes(), - command_ch, wakeup_fd) + server.begin_request() + worker := spawn process_request_async(server, sockets[0], + 'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n'.bytes(), command_ch, wakeup_fd) mut cmd := LoopCommand{} select { @@ -44,8 +45,7 @@ fn test_prealloc_async_write_state_outlives_worker() ! { return } } - // The worker must have exited before the event loop consumes its state. - time.sleep(50 * time.millisecond) + worker.wait() assert cmd.kind == .complete_response assert cmd.state != unsafe { nil } assert cmd.state.content.len > 0 From 5b68deab9c1a6c7f065a61665d325d7dd8cf278f Mon Sep 17 00:00:00 2001 From: weigang Date: Tue, 23 Jun 2026 14:45:07 +0800 Subject: [PATCH 11/12] fix(fasthttp): guard stale takeover fd cleanup --- vlib/fasthttp/fasthttp_bsd.c.v | 13 +++++-- .../fasthttp/fasthttp_bsd_regression_test.c.v | 35 +++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/vlib/fasthttp/fasthttp_bsd.c.v b/vlib/fasthttp/fasthttp_bsd.c.v index 5706165e02e5c1..5a40c9e2871c6c 100644 --- a/vlib/fasthttp/fasthttp_bsd.c.v +++ b/vlib/fasthttp/fasthttp_bsd.c.v @@ -200,6 +200,11 @@ fn delete_event(kq int, ident u64, filter i16, udata voidptr) { C.kevent(kq, &ev, 1, unsafe { nil }, 0, unsafe { nil }) } +fn client_is_tracked(c &Conn, c_ptr voidptr, clients map[int]voidptr) bool { + tracked := clients[c.fd] or { return false } + return tracked == c_ptr +} + fn close_conn(server &Server, kq int, c_ptr voidptr, mut clients map[int]voidptr) { mut c := unsafe { &Conn(c_ptr) } clients.delete(c.fd) @@ -489,9 +494,11 @@ fn process_loop_command(server &Server, kq int, cmd LoopCommand, mut clients map server.end_request() c.request_active = false } - 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) + if client_is_tracked(c, cmd.c_ptr, clients) { + 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) + } unsafe { free(cmd.c_ptr) } } .enable_read { diff --git a/vlib/fasthttp/fasthttp_bsd_regression_test.c.v b/vlib/fasthttp/fasthttp_bsd_regression_test.c.v index b46e0452cedfd4..988b98aee8b261 100644 --- a/vlib/fasthttp/fasthttp_bsd_regression_test.c.v +++ b/vlib/fasthttp/fasthttp_bsd_regression_test.c.v @@ -49,6 +49,41 @@ fn test_reusable_completion_ends_request_in_event_loop() ! { close_conn(server, kq, conn, mut clients) } +fn test_manual_takeover_keeps_reused_fd_client_tracked() ! { + server := new_bsd_lifecycle_test_server()! + kq := C.kqueue() + assert kq >= 0 + defer { + C.close(kq) + } + client_fd := int(C.socket(i32(net.AddrFamily.ip), i32(net.SocketType.tcp), 0)) + assert client_fd >= 0 + mut old_conn := &Conn{ + fd: client_fd + request_active: true + processing: true + file_fd: -1 + } + mut new_conn := &Conn{ + fd: client_fd + file_fd: -1 + } + mut clients := { + client_fd: voidptr(new_conn) + } + server.begin_request() + process_loop_command(server, kq, LoopCommand{ + kind: .manual_takeover + c_ptr: old_conn + }, mut clients) + + assert server.active_request_count() == 0 + assert clients[client_fd] or { unsafe { nil } } == voidptr(new_conn) + clients.delete(client_fd) + C.close(client_fd) + unsafe { free(new_conn) } +} + fn test_loop_command_wakes_kqueue() ! { kq := C.kqueue() assert kq >= 0 From f609523bd42c8f5fc035ff84fd6009a5c931f8bf Mon Sep 17 00:00:00 2001 From: weigang Date: Wed, 24 Jun 2026 20:31:41 +0800 Subject: [PATCH 12/12] fix(fasthttp): guard stale Linux takeover cleanup --- vlib/fasthttp/fasthttp_linux.v | 119 +++++++++++------- .../fasthttp_linux_prealloc_regression_test.v | 2 +- .../fasthttp/fasthttp_linux_regression_test.v | 50 +++++++- 3 files changed, 120 insertions(+), 51 deletions(-) diff --git a/vlib/fasthttp/fasthttp_linux.v b/vlib/fasthttp/fasthttp_linux.v index 683a78bc834aa6..d4d0a3d2bbd3ed 100644 --- a/vlib/fasthttp/fasthttp_linux.v +++ b/vlib/fasthttp/fasthttp_linux.v @@ -51,9 +51,10 @@ enum LoopCommandKind { } struct LoopCommand { - kind LoopCommandKind - client_fd int - state &ClientWriteState = unsafe { nil } + kind LoopCommandKind + client_fd int + generation u64 + state &ClientWriteState = unsafe { nil } } fn wake_event_loop(wakeup_fd int) { @@ -248,7 +249,8 @@ fn remove_fd_from_epoll(epoll_fd int, fd int) bool { return true } -fn handle_accept_loop(epoll_fd int, listen_fd int, mut client_fds map[int]bool) { +fn handle_accept_loop(epoll_fd int, listen_fd int, mut client_fds map[int]u64, next_generation u64) u64 { + mut current_generation := next_generation for { client_fd := C.accept4(listen_fd, C.NULL, C.NULL, C.SOCK_NONBLOCK) if client_fd < 0 { @@ -267,8 +269,13 @@ fn handle_accept_loop(epoll_fd int, listen_fd int, mut client_fds map[int]bool) close_socket(client_fd) continue } - client_fds[client_fd] = true + current_generation++ + if current_generation == 0 { + current_generation++ + } + client_fds[client_fd] = current_generation } + return current_generation } fn free_write_state(server &Server, client_fd int, mut client_write_states map[int]&ClientWriteState) { @@ -295,7 +302,7 @@ fn free_write_state(server &Server, client_fd int, mut client_write_states map[i unsafe { free(state) } } -fn handle_client_closure(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) { +fn handle_client_closure(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]u64, 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) { // Never close the listening socket here if client_fd == 0 { return @@ -313,14 +320,14 @@ fn handle_client_closure(server &Server, epoll_fd int, client_fd int, mut client close_socket(client_fd) } -fn close_worker_clients(server &Server, epoll_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) { +fn close_worker_clients(server &Server, epoll_fd int, mut client_fds map[int]u64, 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) { for client_fd in client_fds.keys() { 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) } } -fn drain_closing_client(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) { +fn drain_closing_client(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]u64, 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) { mut drain_buf := []u8{len: 4096} for { bytes_read := C.recv(client_fd, unsafe { &drain_buf[0] }, drain_buf.len, 0) @@ -455,7 +462,7 @@ fn arm_epollout(epoll_fd int, client_fd int) int { // complete_write tears down the per-fd write state after a successful drain, // then either closes the connection or leaves it in keep-alive mode. -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) { +fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]u64, 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 @@ -481,7 +488,7 @@ fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds ma } // handle_write resumes a blocked write when EPOLLOUT fires for `client_fd`. -fn handle_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) { +fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]u64, 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) { mut state := client_write_states[client_fd] or { return } status := try_drain_write(client_fd, mut state) match status { @@ -499,7 +506,7 @@ fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[ } } -fn process_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { +fn process_request_async(server &Server, client_fd int, generation u64, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { $if fasthttp_test_delay_async_start ? { time.sleep(200 * time.millisecond) } @@ -518,8 +525,9 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com server.end_request() } send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd + kind: .close_conn + client_fd: client_fd + generation: generation }) return } @@ -537,8 +545,9 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com server.end_request() } send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd + kind: .close_conn + client_fd: client_fd + generation: generation }) return } @@ -552,8 +561,9 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com response.abandon_request_arena_current_thread() response.free_owned_content() send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .manual_takeover - client_fd: client_fd + kind: .manual_takeover + client_fd: client_fd + generation: generation }) return } @@ -567,13 +577,15 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com } if server.is_shutting_down() || response.should_close { send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd + kind: .close_conn + client_fd: client_fd + generation: generation }) } else { send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .enable_read - client_fd: client_fd + kind: .enable_read + client_fd: client_fd + generation: generation }) } return @@ -595,8 +607,9 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com server.end_request() } send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd + kind: .close_conn + client_fd: client_fd + generation: generation }) return } @@ -610,8 +623,9 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com server.end_request() } send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd + kind: .close_conn + client_fd: client_fd + generation: generation }) return } @@ -643,35 +657,44 @@ fn process_request_async(server &Server, client_fd int, request_buffer []u8, com match status { .done { send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .complete_response - client_fd: client_fd - state: state + kind: .complete_response + client_fd: client_fd + generation: generation + state: state }) } .blocked { send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .arm_write - client_fd: client_fd - state: state + kind: .arm_write + client_fd: client_fd + generation: generation + state: state }) } .failed { send_loop_command(command_ch, wakeup_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd - state: state + kind: .close_conn + client_fd: client_fd + generation: generation + state: state }) } } } -fn dispatch_request_async(server &Server, client_fd int, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { +fn dispatch_request_async(server &Server, client_fd int, generation u64, request_buffer []u8, command_ch chan LoopCommand, wakeup_fd int) { // Account for the request before shutdown can observe the spawned work. server.begin_request() - spawn process_request_async(server, client_fd, request_buffer, command_ch, wakeup_fd) + spawn process_request_async(server, client_fd, generation, request_buffer, command_ch, + wakeup_fd) +} + +fn command_matches_client(cmd LoopCommand, client_fds map[int]u64) bool { + generation := client_fds[cmd.client_fd] or { return false } + return generation == cmd.generation } -fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, 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) { +fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, mut client_fds map[int]u64, 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) { match cmd.kind { .close_conn { if cmd.state != unsafe { nil } { @@ -708,10 +731,12 @@ fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, mut clien } .manual_takeover { server.end_request() - client_fds.delete(cmd.client_fd) - client_buffers.delete(cmd.client_fd) - client_read_starts.delete(cmd.client_fd) - closing_client_fds.delete(cmd.client_fd) + if command_matches_client(cmd, client_fds) { + client_fds.delete(cmd.client_fd) + client_buffers.delete(cmd.client_fd) + client_read_starts.delete(cmd.client_fd) + closing_client_fds.delete(cmd.client_fd) + } if cmd.state != unsafe { nil } { free_write_state(server, cmd.client_fd, mut client_write_states) } @@ -727,7 +752,7 @@ fn process_loop_command(server &Server, epoll_fd int, cmd LoopCommand, mut clien } } -fn drain_loop_commands(server &Server, epoll_fd int, command_ch chan LoopCommand, 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) { +fn drain_loop_commands(server &Server, epoll_fd int, command_ch chan LoopCommand, mut client_fds map[int]u64, 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) { for { select { cmd := <-command_ch { @@ -744,11 +769,12 @@ fn drain_loop_commands(server &Server, epoll_fd int, command_ch chan LoopCommand fn process_events(server &Server, epoll_fd int, listen_fd int) { mut events := [max_connection_size]C.epoll_event{} mut request_buffer := []u8{len: server.max_request_buffer_size, cap: server.max_request_buffer_size} - mut client_fds := map[int]bool{} + mut client_fds := map[int]u64{} 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{} + mut next_generation := u64(0) command_ch := chan LoopCommand{cap: max_connection_size * 4} wakeup_fd := C.eventfd(0, C.EFD_NONBLOCK | C.EFD_CLOEXEC) if wakeup_fd < 0 { @@ -797,7 +823,8 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { if server.is_shutting_down() { continue } - handle_accept_loop(epoll_fd, listen_fd, mut client_fds) + next_generation = handle_accept_loop(epoll_fd, listen_fd, mut client_fds, + next_generation) continue } @@ -925,7 +952,9 @@ fn process_events(server &Server, epoll_fd int, listen_fd int) { client_read_starts.delete(client_fd) req_buf := readed_request_buffer.clone() client_buffers.delete(client_fd) - dispatch_request_async(server, client_fd, req_buf, command_ch, wakeup_fd) + generation := client_fds[client_fd] or { 0 } + dispatch_request_async(server, client_fd, generation, req_buf, command_ch, + wakeup_fd) } else if recv_error { // Unexpected recv error - send 444 No Response C.send(client_fd, status_444_response.data, status_444_response.len, diff --git a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v index 9bbac76081b14d..c149bde0b36d34 100644 --- a/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_prealloc_regression_test.v @@ -32,7 +32,7 @@ fn test_prealloc_async_write_state_outlives_worker() ! { } command_ch := chan LoopCommand{cap: 1} server.begin_request() - worker := spawn process_request_async(server, sockets[0], + worker := spawn process_request_async(server, sockets[0], 1, 'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n'.bytes(), command_ch, wakeup_fd) mut cmd := LoopCommand{} diff --git a/vlib/fasthttp/fasthttp_linux_regression_test.v b/vlib/fasthttp/fasthttp_linux_regression_test.v index 853e09222cbc18..e930c8636ac241 100644 --- a/vlib/fasthttp/fasthttp_linux_regression_test.v +++ b/vlib/fasthttp/fasthttp_linux_regression_test.v @@ -38,16 +38,17 @@ fn test_close_command_releases_unregistered_write_state() ! { request_active: true } mut client_fds := { - client_fd: true + client_fd: u64(1) } 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{} process_loop_command(server, epoll_fd, LoopCommand{ - kind: .close_conn - client_fd: client_fd - state: state + kind: .close_conn + client_fd: client_fd + generation: 1 + state: state }, mut client_fds, mut client_buffers, mut client_read_starts, mut closing_client_fds, mut client_write_states) @@ -64,7 +65,7 @@ fn test_dispatch_marks_request_active_before_spawn() ! { defer { C.close(wakeup_fd) } - dispatch_request_async(server, -1, 'invalid request'.bytes(), command_ch, wakeup_fd) + dispatch_request_async(server, -1, 1, 'invalid request'.bytes(), command_ch, wakeup_fd) assert server.active_request_count() == 1 select { @@ -77,6 +78,45 @@ fn test_dispatch_marks_request_active_before_spawn() ! { } } +fn test_manual_takeover_keeps_reused_fd_state() ! { + server := new_lifecycle_test_server()! + epoll_fd := C.epoll_create1(0) + assert epoll_fd >= 0 + defer { + C.close(epoll_fd) + } + client_fd := int(C.socket(i32(net.AddrFamily.ip), i32(net.SocketType.tcp), 0)) + assert client_fd >= 0 + defer { + C.close(client_fd) + } + mut client_fds := { + client_fd: u64(2) + } + mut client_buffers := { + client_fd: 'GET /partial'.bytes() + } + mut client_read_starts := map[int]i64{} + client_read_starts[client_fd] = time.sys_mono_now() + mut closing_client_fds := { + client_fd: true + } + mut client_write_states := map[int]&ClientWriteState{} + server.begin_request() + process_loop_command(server, epoll_fd, LoopCommand{ + kind: .manual_takeover + client_fd: client_fd + generation: 1 + }, 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 client_fds[client_fd] or { 0 } == 2 + assert client_buffers[client_fd] or { []u8{} } == 'GET /partial'.bytes() + assert client_fd in client_read_starts + assert closing_client_fds[client_fd] or { false } +} + fn test_loop_command_wakes_epoll() ! { epoll_fd := C.epoll_create1(0) assert epoll_fd >= 0