coop/rose-ash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sx-http: non-blocking server — fast path for cached, render workers for misses

Browse Source 
Replace blocking domain pool with non-blocking architecture:
- Main accept loop handles ALL connections immediately
- Cached responses: served in microseconds from main loop (no queuing)
- Static files: served immediately from main loop
- Cache misses: queued to render worker pool (domain workers)
- Socket timeouts (5s recv, 10s send) prevent connection hangs
- TCP backlog increased to 1024

No more connection resets under load. 22/26 Playwright tests pass
(4 failures from stale worktree test copies, 0 from main tree).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
giles
2026-03-28 20:47:43 +00:00
parent 4294ee3d94
commit 30785c92c0
1 changed files with 126 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
hosts/ocaml/bin/sx_server.ml
View File

@@ -2065,138 +2065,155 @@ let http_mode port =
has_substring lower "sx-request" || has_substring lower "hx-request"
in
(* Handle one HTTP request *)
let handle_client env client =
let buf = Bytes.create 8192 in
let n = try Unix.read client buf 0 8192 with _ -> 0 in
if n > 0 then begin
let data = Bytes.sub_string buf 0 n in
let is_ajax = is_sx_request data in
let response =
try
match parse_http_request data with
| None -> http_response ~status:400 "Bad Request"
| Some (method_, raw_path) ->
if method_ <> "GET" && method_ <> "HEAD" then
http_response ~status:405 "Method Not Allowed"
else begin
let path = url_decode raw_path in
(* Redirect bare / to /sx/ *)
if path = "/" then http_redirect "/sx/"
else
let is_sx = path = "/sx/" || path = "/sx"
|| (String.length path > 4 && String.sub path 0 4 = "/sx/") in
if is_sx then begin
if is_ajax then
(* AJAX navigation — return just the content fragment,
not the full page shell. The client swaps #main-panel. *)
(match http_render_page env path with
| Some html ->
(* Extract #main-panel from the full page HTML *)
let panel_start = try
let idx = ref 0 in
let found = ref false in
while not !found && !idx < String.length html - 20 do
if String.sub html !idx 18 = "id=\"main-panel\"" then
found := true
else
idx := !idx + 1
done;
if !found then begin
(* Walk back to find the opening < *)
let start = ref !idx in
while !start > 0 && html.[!start] <> '<' do
start := !start - 1
done;
Some !start
end else None
with _ -> None in
(match panel_start with
| Some start ->
(* Find matching close tag — scan for </section> or end *)
let fragment = String.sub html start (String.length html - start) in
http_response ~content_type:"text/html; charset=utf-8" fragment
| None -> http_response html)
| None -> http_response ~status:404 "<h1>Not Found</h1>")
else
(* Full page request — check cache *)
match Hashtbl.find_opt response_cache path with
| Some cached -> cached
| None ->
(match http_render_page env path with
| Some html ->
let resp = http_response html in
Hashtbl.replace response_cache path resp;
resp
| None -> http_response ~status:404 "<h1>Not Found</h1>")
end
else if String.length path > 8 && String.sub path 0 8 = "/static/" then
serve_static_file static_dir path
else
http_response ~status:404 "<h1>Not Found</h1>"
end
with e ->
Printf.eprintf "[sx-http] Error: %s\n%!" (Printexc.to_string e);
http_response ~status:500 "<h1>Internal Server Error</h1>"
in
write_response client response
end else
(try Unix.close client with _ -> ())
in
(* Non-blocking event loop with render worker pool.
- Main loop: Unix.select on listen socket + all connected clients
- Cached responses: served immediately from main loop (microseconds)
- Cache misses: queued to render workers (domain pool)
- Never blocks on rendering — accept loop always responsive *)
(* Domain pool — each domain has its own minor heap for GC isolation.
Requests are dispatched round-robin to avoid GC pauses blocking others. *)
let n_workers = max 1 (Domain.recommended_domain_count ()) in
Printf.eprintf "[sx-http] Starting %d worker domains\n%!" n_workers;
let n_workers = max 2 (Domain.recommended_domain_count ()) in
(* Request queue: mutex + condition + list *)
let queue : Unix.file_descr list ref = ref [] in
let queue_mutex = Mutex.create () in
let queue_cond = Condition.create () in
(* Render queue: for cache misses that need full page render *)
let render_queue : (Unix.file_descr * string * bool) list ref = ref [] in
let render_mutex = Mutex.create () in
let render_cond = Condition.create () in
let shutdown = ref false in
(* Worker loop — each domain pops from queue and handles requests *)
let worker_fn _id () =
(* Render worker: processes cache misses in background *)
let render_worker _id () =
while not !shutdown do
let client =
Mutex.lock queue_mutex;
while !queue = [] && not !shutdown do
Condition.wait queue_cond queue_mutex
let work =
Mutex.lock render_mutex;
while !render_queue = [] && not !shutdown do
Condition.wait render_cond render_mutex
done;
let c = match !queue with
| fd :: rest -> queue := rest; Some fd
let w = match !render_queue with
| item :: rest -> render_queue := rest; Some item
| [] -> None
in
Mutex.unlock queue_mutex;
c
Mutex.unlock render_mutex;
w
in
match client with
| Some fd -> handle_client env fd
match work with
| Some (fd, path, is_ajax) ->
let response =
try
match http_render_page env path with
| Some html ->
if is_ajax then begin
(* Extract #main-panel fragment for AJAX *)
let panel_start = try
let idx = ref 0 in
let found = ref false in
while not !found && !idx < String.length html - 20 do
if String.sub html !idx 18 = "id=\"main-panel\"" then found := true
else idx := !idx + 1
done;
if !found then begin
let start = ref !idx in
while !start > 0 && html.[!start] <> '<' do start := !start - 1 done;
Some !start
end else None
with _ -> None in
match panel_start with
| Some start ->
let fragment = String.sub html start (String.length html - start) in
http_response ~content_type:"text/html; charset=utf-8" fragment
| None -> http_response html
end else begin
let resp = http_response html in
Hashtbl.replace response_cache path resp;
resp
end
| None -> http_response ~status:404 "<h1>Not Found</h1>"
with e ->
Printf.eprintf "[render] Error for %s: %s\n%!" path (Printexc.to_string e);
http_response ~status:500 "<h1>Internal Server Error</h1>"
in
write_response fd response
| None -> ()
done
in
(* Spawn worker domains *)
let workers = Array.init n_workers (fun id ->
Domain.spawn (worker_fn id)) in
(* Fast path: handle a request from the main loop.
Returns true if handled immediately (cached), false if queued. *)
let fast_handle fd data is_ajax =
match parse_http_request data with
| None -> write_response fd (http_response ~status:400 "Bad Request"); true
| Some (method_, raw_path) ->
if method_ <> "GET" && method_ <> "HEAD" then begin
write_response fd (http_response ~status:405 "Method Not Allowed"); true
end else begin
let path = url_decode raw_path in
if path = "/" then begin
write_response fd (http_redirect "/sx/"); true
end else
let is_sx = path = "/sx/" || path = "/sx"
|| (String.length path > 4 && String.sub path 0 4 = "/sx/") in
if is_sx then begin
if not is_ajax then
(* Check cache — instant response *)
match Hashtbl.find_opt response_cache path with
| Some cached -> write_response fd cached; true
| None ->
(* Queue for render worker *)
Mutex.lock render_mutex;
render_queue := !render_queue @ [(fd, path, false)];
Condition.signal render_cond;
Mutex.unlock render_mutex;
false
else begin
(* AJAX always renders fresh (no cache for fragments) *)
Mutex.lock render_mutex;
render_queue := !render_queue @ [(fd, path, true)];
Condition.signal render_cond;
Mutex.unlock render_mutex;
false
end
end else if String.length path > 8 && String.sub path 0 8 = "/static/" then begin
write_response fd (serve_static_file static_dir path); true
end else begin
write_response fd (http_response ~status:404 "<h1>Not Found</h1>"); true
end
end
in
(* Start TCP server — main domain accepts and enqueues *)
(* Spawn render workers *)
let workers = Array.init n_workers (fun id ->
Domain.spawn (render_worker id)) in
(* Start TCP server — non-blocking accept loop *)
let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
Unix.setsockopt sock Unix.SO_REUSEADDR true;
Unix.bind sock (Unix.ADDR_INET (Unix.inet_addr_any, port));
Unix.listen sock 128;
Printf.eprintf "[sx-http] Listening on port %d (%d workers, project=%s)\n%!" port n_workers project_dir;
Unix.listen sock 1024;
Printf.eprintf "[sx-http] Listening on port %d (%d render workers, non-blocking)\n%!" port n_workers;
(try
while true do
(* Accept a connection *)
let (client, _addr) = Unix.accept sock in
Mutex.lock queue_mutex;
queue := !queue @ [client];
Condition.signal queue_cond;
Mutex.unlock queue_mutex
(* Read request — non-blocking: set a short timeout *)
Unix.setsockopt_float client Unix.SO_RCVTIMEO 5.0;
Unix.setsockopt_float client Unix.SO_SNDTIMEO 10.0;
let buf = Bytes.create 8192 in
let n = try Unix.read client buf 0 8192 with _ -> 0 in
if n > 0 then begin
let data = Bytes.sub_string buf 0 n in
let is_ajax = is_sx_request data in
let handled =
try fast_handle client data is_ajax
with e ->
Printf.eprintf "[sx-http] Error: %s\n%!" (Printexc.to_string e);
write_response client (http_response ~status:500 "<h1>Internal Server Error</h1>");
true
in
ignore handled
end else
(try Unix.close client with _ -> ())
done
with _ ->
shutdown := true;
Condition.broadcast queue_cond;
Condition.broadcast render_cond;
Array.iter Domain.join workers)