fix(streaming): use direct socket fd writes for real-time SSE streaming

Previously, PicoevStreamWriter used picohttpparser.Response.raw() to write
streaming data, which only buffered the data until the entire request handler
completed. This caused SSE responses to be delayed until all sleep() calls
finished (e.g., 2.89s delay for /stream-text endpoint).

Changes:
- Modified PicoevStreamWriter to use socket fd directly from picohttpparser.Response
- Implemented write_to_fd() using C.send() for immediate data delivery
- Updated handle_picoev_streaming_response() to write HTTP headers directly to socket
- Added C.send() function declaration for direct socket writes

This fix enables true real-time streaming where each write() or write_sse()
call immediately sends data to the client, rather than buffering until
the callback completes.

All 74 test files pass with no regressions.
This commit is contained in:
meiseayoung
2026-01-04 15:43:01 +08:00
parent 9ae9170bcb
commit e9d14a35f6
2 changed files with 86 additions and 30 deletions
+41 -17
View File
@@ -583,8 +583,8 @@ fn send_picoev_response(mut res picohttpparser.Response, ctx Context, response h
// handle_picoev_streaming_response - Handle a streaming response using PicoevStreamWriter // handle_picoev_streaming_response - Handle a streaming response using PicoevStreamWriter
// This function: // This function:
// 1. Sends the HTTP headers for streaming // 1. Sends the HTTP headers for streaming directly to socket fd
// 2. Creates a PicoevStreamWriter // 2. Creates a PicoevStreamWriter with the socket fd
// 3. Executes the streaming callback // 3. Executes the streaming callback
// 4. Handles errors and cleanup // 4. Handles errors and cleanup
fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context, config PicoevConfig) { fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context, config PicoevConfig) {
@@ -601,14 +601,10 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
// Determine status code // Determine status code
status_code := if ctx.status_code != 0 { ctx.status_code } else { 200 } status_code := if ctx.status_code != 0 { ctx.status_code } else { 200 }
// Send HTTP response line // Build HTTP headers string for direct socket write
if status_code == 200 { mut headers_str := 'HTTP/1.1 ${status_code} ${get_status_text(status_code)}\r\n'
res.http_ok()
} else {
res.raw('HTTP/1.1 ${status_code} ${get_status_text(status_code)}\r\n')
}
// Send headers from context // Add headers from context
mut has_transfer_encoding := false mut has_transfer_encoding := false
mut has_connection := false mut has_connection := false
@@ -618,7 +614,7 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
if key_len == 14 && eq_ignore_case(key, 'content-length') { if key_len == 14 && eq_ignore_case(key, 'content-length') {
continue continue
} }
res.header(key, value) headers_str += '${key}: ${value}\r\n'
if key_len == 17 && eq_ignore_case(key, 'transfer-encoding') { if key_len == 17 && eq_ignore_case(key, 'transfer-encoding') {
has_transfer_encoding = true has_transfer_encoding = true
} else if key_len == 10 && eq_ignore_case(key, 'connection') { } else if key_len == 10 && eq_ignore_case(key, 'connection') {
@@ -628,20 +624,27 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
// Ensure Transfer-Encoding: chunked is set // Ensure Transfer-Encoding: chunked is set
if !has_transfer_encoding { if !has_transfer_encoding {
res.header('Transfer-Encoding', 'chunked') headers_str += 'Transfer-Encoding: chunked\r\n'
} }
// Ensure Connection header is set for streaming // Ensure Connection header is set for streaming
if !has_connection { if !has_connection {
res.header('Connection', 'keep-alive') headers_str += 'Connection: keep-alive\r\n'
res.header('Keep-Alive', 'timeout=${config.timeout_secs}') headers_str += 'Keep-Alive: timeout=${config.timeout_secs}\r\n'
} }
// End headers section (empty body to signal headers are complete) // End headers section
// Note: We don't call res.body() or res.end() here because we're streaming headers_str += '\r\n'
res.raw('\r\n')
// Create PicoevStreamWriter for streaming // Write headers directly to socket fd for immediate delivery
fd := res.fd
write_to_socket_fd(fd, headers_str.bytes()) or {
eprintln('[Picoev Stream Error] Failed to write headers: ${err.msg()}')
cleanup_stream_config(ctx)
return
}
// Create PicoevStreamWriter for streaming (uses fd directly)
mut writer := PicoevStreamWriter.new(res) mut writer := PicoevStreamWriter.new(res)
// Create StreamContext with the writer // Create StreamContext with the writer
@@ -665,6 +668,27 @@ fn handle_picoev_streaming_response(mut res picohttpparser.Response, ctx Context
cleanup_stream_config(ctx) cleanup_stream_config(ctx)
} }
// write_to_socket_fd - Helper function to write bytes directly to socket fd
// Used for sending HTTP headers before streaming begins
fn write_to_socket_fd(fd int, data []u8) ! {
if data.len == 0 {
return
}
mut total_written := 0
for total_written < data.len {
remaining := data.len - total_written
ptr := unsafe { &u8(data.data) + total_written }
written := C.send(fd, ptr, remaining, 0)
if written < 0 {
return error('Failed to write to socket')
}
if written == 0 {
return error('Connection closed by peer')
}
total_written += written
}
}
// 获取状态码文本 // 获取状态码文本
fn get_status_text(code int) string { fn get_status_text(code int) string {
return match code { return match code {
+45 -13
View File
@@ -8,6 +8,9 @@ import time
import picohttpparser import picohttpparser
import usockets import usockets
// C function declaration for direct socket writes
fn C.send(sockfd int, buf voidptr, len int, flags int) int
// ============================================================================ // ============================================================================
// SSE Event Structure // SSE Event Structure
// ============================================================================ // ============================================================================
@@ -184,23 +187,26 @@ pub fn (mut ctx StreamContext) trigger_abort() {
// PicoevStreamWriter - StreamWriter implementation for picoev server // PicoevStreamWriter - StreamWriter implementation for picoev server
// Supports chunked transfer encoding for streaming responses // Supports chunked transfer encoding for streaming responses
// Uses direct socket fd writes for immediate data delivery (true streaming)
@[heap] @[heap]
pub struct PicoevStreamWriter { pub struct PicoevStreamWriter {
mut: mut:
res &picohttpparser.Response fd int // Socket file descriptor for direct writes
connected bool connected bool
} }
// PicoevStreamWriter.new - Create a new PicoevStreamWriter with the given response // PicoevStreamWriter.new - Create a new PicoevStreamWriter with the given response
// Extracts the fd from picohttpparser.Response for direct socket writes
pub fn PicoevStreamWriter.new(res &picohttpparser.Response) &PicoevStreamWriter { pub fn PicoevStreamWriter.new(res &picohttpparser.Response) &PicoevStreamWriter {
return &PicoevStreamWriter{ return &PicoevStreamWriter{
res: unsafe { res } fd: res.fd
connected: true connected: true
} }
} }
// write - Write raw bytes to the stream using chunked transfer encoding // write - Write raw bytes to the stream using chunked transfer encoding
// Format: {hex_size}\r\n{data}\r\n // Format: {hex_size}\r\n{data}\r\n
// Writes directly to socket fd for immediate delivery
pub fn (mut w PicoevStreamWriter) write(data []u8) ! { pub fn (mut w PicoevStreamWriter) write(data []u8) ! {
if !w.connected { if !w.connected {
return error('Connection closed') return error('Connection closed')
@@ -210,9 +216,10 @@ pub fn (mut w PicoevStreamWriter) write(data []u8) ! {
} }
// Write chunked format: size in hex, CRLF, data, CRLF // Write chunked format: size in hex, CRLF, data, CRLF
chunk_header := format_chunk_size(data.len) chunk_header := format_chunk_size(data.len)
w.res.raw(chunk_header) // Write directly to socket fd
w.res.raw(data.bytestr()) w.write_to_fd(chunk_header.bytes())!
w.res.raw('\r\n') w.write_to_fd(data)!
w.write_to_fd('\r\n'.bytes())!
} }
// write_string - Write a string to the stream using chunked transfer encoding // write_string - Write a string to the stream using chunked transfer encoding
@@ -225,15 +232,40 @@ pub fn (mut w PicoevStreamWriter) write_string(data string) ! {
} }
// Write chunked format: size in hex, CRLF, data, CRLF // Write chunked format: size in hex, CRLF, data, CRLF
chunk_header := format_chunk_size(data.len) chunk_header := format_chunk_size(data.len)
w.res.raw(chunk_header) // Write directly to socket fd
w.res.raw(data) w.write_to_fd(chunk_header.bytes())!
w.res.raw('\r\n') w.write_to_fd(data.bytes())!
w.write_to_fd('\r\n'.bytes())!
} }
// flush - Flush the buffer (picoev handles this automatically) // write_to_fd - Write bytes directly to the socket file descriptor
// This ensures immediate delivery without buffering
@[inline]
fn (mut w PicoevStreamWriter) write_to_fd(data []u8) ! {
if data.len == 0 {
return
}
mut total_written := 0
for total_written < data.len {
remaining := data.len - total_written
ptr := unsafe { &u8(data.data) + total_written }
written := C.send(w.fd, ptr, remaining, 0)
if written < 0 {
w.connected = false
return error('Failed to write to socket')
}
if written == 0 {
w.connected = false
return error('Connection closed by peer')
}
total_written += written
}
}
// flush - Flush the buffer (direct fd writes are already unbuffered)
pub fn (mut w PicoevStreamWriter) flush() ! { pub fn (mut w PicoevStreamWriter) flush() ! {
// picoev automatically flushes data // Direct fd writes are unbuffered, no explicit flush needed
// No explicit flush needed // But we can use TCP_NODELAY to disable Nagle's algorithm if needed
} }
// close - Close the stream by sending the final chunk // close - Close the stream by sending the final chunk
@@ -241,9 +273,9 @@ pub fn (mut w PicoevStreamWriter) flush() ! {
pub fn (mut w PicoevStreamWriter) close() ! { pub fn (mut w PicoevStreamWriter) close() ! {
if w.connected { if w.connected {
// Write the final chunk (zero-length chunk) to signal end of stream // Write the final chunk (zero-length chunk) to signal end of stream
w.res.raw('0\r\n\r\n') w.write_to_fd('0\r\n\r\n'.bytes()) or {}
w.res.end()
w.connected = false w.connected = false
// Note: Don't close the fd here, picoev manages the connection lifecycle
} }
} }