Skip to content

refactor(fasthttp): make both Linux (epoll) and BSD (kqueue) backends asynchronous #27477

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
guweigang wants to merge 14 commits into from
Closed

refactor(fasthttp): make both Linux (epoll) and BSD (kqueue) backends asynchronous #27477

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
b48a62c
fix: improve fasthttp bsd processing using loop command channel
guweigang Jun 17, 2026
c7222f8
fix(fasthttp): fix manual takeover shutdown race and guard test to BSD
guweigang Jun 19, 2026
68402cd
refactor(fasthttp): make linux backend asynchronous and fix loopback …
guweigang Jun 19, 2026
7845d39
fix(fasthttp): close Linux async request races
guweigang Jun 20, 2026
5de24dd
test(fasthttp): cover Linux async request lifecycle
guweigang Jun 20, 2026
ef66e1b
fix(fasthttp): close async poll lifecycle races
guweigang Jun 21, 2026
b7a287c
test(fasthttp): silence epoll union warning
guweigang Jun 21, 2026
5ec4483
fix(fasthttp): support takeover with musl and tcc CI
guweigang Jun 22, 2026
1d6739d
test(fasthttp): clean prealloc regression imports
guweigang Jun 22, 2026
a3f5a1e
test(fasthttp): wait for prealloc worker exit
guweigang Jun 22, 2026
5b68dea
fix(fasthttp): guard stale takeover fd cleanup
guweigang Jun 23, 2026
dd86d21
Merge remote-tracking branch 'upstream/master' into fix-fasthttp-bsd
guweigang Jun 23, 2026
f609523
fix(fasthttp): guard stale Linux takeover cleanup
guweigang Jun 24, 2026
02a18d4
Merge remote-tracking branch 'upstream/master' into fix-fasthttp-bsd
guweigang Jun 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
199 changes: 146 additions & 53 deletions vlib/fasthttp/fasthttp_bsd.c.v
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ mut:
file_pos i64
should_close bool
request_arena voidptr
processing bool
}

fn (mut c Conn) free_write_buf() {
Expand Down Expand Up @@ -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 {
Expand All @@ -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 ? {
Expand All @@ -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{
Comment thread
guweigang marked this conversation as resolved.
Outdated
kind: .manual_takeover
c_ptr: c_ptr
}
return
}
.reusable {
Expand All @@ -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
}
Expand All @@ -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)
}
}
}
Expand All @@ -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)
Comment thread
guweigang marked this conversation as resolved.
Outdated
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,
Expand All @@ -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
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -695,19 +787,20 @@ 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()
timeout_ns := i64(s.timeout_in_seconds) * 1_000_000_000
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)
Expand Down
Loading
Loading