udo/uce
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wasm runtime: central WS broker, unified handlers, W7d holdouts, membrane completeness

Browse Source 
- WS: a dedicated broker process owns HTTP_PORT + every connection; it forwards
  renders to the worker pool over uce.sock (non-blocking) and applies ws_*
  command batches flushed back at workspace teardown. Removes the now-dead
  per-worker websocket executor (-509 lines).
- Dispatch: unify CLI / WebSocket / serve_http / page render through one
  serve_via_wasm(entry_unit, handler) path; handler string -> __uce_<handler>
  export symbol.
- W7d: rewrite zip.uce to the membrane return-value error contract (no C++
  try/catch), error-reporting.uce to genuine wasm traps instead of throw, and
  sharedunit.uce to unit_info(); empty the native-only token gate.
- Membrane: wire ls / mkdir / file_mtime through new uce_host_file_list /
  uce_host_file_mkdir / uce_host_file_mtime hostcalls (resolve_guest_file gains
  directory support). Fixes /doc/index.uce listing nothing; adds a regression
  assertion that the index enumerates items.
- Docs: add docs/wasm-runtime-architecture.md; record the W7e staged native-
  deletion plan in WASM-PROPOSAL.md.

Verified: scripts/run_cli_tests.sh --include-wasm-kill -> 87 passed, 0 failed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
root
2026-06-14 17:51:51 +00:00
co-authored by Claude Opus 4.8
parent 15e8d092bc
commit 8587fbc5aa
14 changed files with 1219 additions and 769 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/linux_fastcgi.cpp
+283 -583
View File
@@ -3,8 +3,9 @@
// main object only needs its declarations, so editing the wasm runtime no
// longer recompiles the whole native TU.
#include "wasm/backend.h"
// Minimal FastCGI client: connection brokers forward to a clean-engine worker.
#include "lib/fcgi_forward.h"
#include <csetjmp>
#include <deque>
#include <errno.h>
#include <fcntl.h>
#include <sys/socket.h>
@@ -14,10 +15,15 @@ ServerState server_state;
#include "fastcgi/src/fcgicc.cc"
FastCGIServer server;
pid_t http_worker_pid = 0;
pid_t proactive_compiler_pid = 0;
pid_t websocket_exec_pid = 0;
bool worker_accepts_http = false;
// The central WS broker process: owns the WS port + every connection, forwards
// renders to the worker pool over uce.sock, and applies ws_* commands flushed
// back from workers. ws_broker_outbound holds in-flight async render
// connections (fd -> bytes still to write; empty value means draining the reply).
FastCGIServer ws_broker;
pid_t ws_broker_pid = 0;
std::map<int, String> ws_broker_outbound;
static sigjmp_buf request_fault_jmp;
static volatile sig_atomic_t request_fault_active = 0;
static volatile sig_atomic_t request_fault_signal = 0;
@@ -26,11 +32,6 @@ static Request* request_fault_request = 0;
// captured here and symbolized after the siglongjmp.
static void* request_fault_frames[64];
static volatile sig_atomic_t request_fault_frame_count = 0;
static int websocket_exec_fd = -1;
static String websocket_exec_read_buffer = "";
static std::deque<DValue> websocket_exec_pending_jobs;
static DValue websocket_exec_inflight_job;
static String websocket_exec_write_buffer = "";
void close_inherited_server_sockets();
u64 request_seed_from_time(f64 time_value);
@@ -162,471 +163,6 @@ void restore_request_fault_handlers()
signal(SIGALRM, on_segfault);
}
namespace {
bool websocket_ipc_set_nonblocking(int fd)
{
int flags = fcntl(fd, F_GETFL, 0);
if(flags == -1)
return(false);
return(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
}
bool websocket_exec_enabled_for_process()
{
return(worker_accepts_http);
}
u64 websocket_exec_queue_limit_bytes()
{
u64 configured = int_val(server_state.config["WEBSOCKET_EXEC_QUEUE_BYTES"]);
if(configured < 64 * 1024)
configured = 1024 * 1024;
return(configured);
}
bool websocket_exec_has_inflight_job()
{
return(websocket_exec_inflight_job.to_bool());
}
FastCGIServer::Connection* websocket_find_connection(String connection_id)
{
for(auto& item : server.client_sockets)
{
FastCGIServer::Connection* connection = item.second;
if(connection->is_websocket && connection->websocket_connection_id == connection_id)
return(connection);
}
return(0);
}
void websocket_exec_clear_ipc_state()
{
if(websocket_exec_fd != -1)
close(websocket_exec_fd);
websocket_exec_fd = -1;
websocket_exec_read_buffer = "";
websocket_exec_write_buffer = "";
websocket_exec_inflight_job.clear();
}
void websocket_exec_close_connection(String connection_id, u16 status_code = 1011, String reason = "websocket handler unavailable")
{
if(connection_id == "")
return;
server.websocket_close(connection_id, status_code, reason);
}
void websocket_exec_fail_inflight_job(String reason = "websocket handler unavailable")
{
if(!websocket_exec_has_inflight_job())
return;
websocket_exec_close_connection(websocket_exec_inflight_job["connection_id"].to_string(), 1011, reason);
websocket_exec_inflight_job.clear();
websocket_exec_write_buffer = "";
}
void websocket_exec_queue_job(DValue job)
{
if(job["connection_id"].to_string() == "")
return;
u64 queued_bytes = websocket_exec_write_buffer.length();
if(websocket_exec_has_inflight_job())
queued_bytes += websocket_exec_inflight_job["serialized"].to_string().length();
for(auto& pending : websocket_exec_pending_jobs)
queued_bytes += pending["serialized"].to_string().length();
queued_bytes += json_encode(job).length();
if(queued_bytes > websocket_exec_queue_limit_bytes())
{
printf("(!) websocket dispatch queue overflow for %s\n", job["connection_id"].to_string().c_str());
websocket_exec_close_connection(job["connection_id"].to_string(), 1013, "websocket server busy");
return;
}
job["serialized"] = json_encode(job) + "\n";
websocket_exec_pending_jobs.push_back(job);
}
StringList websocket_exec_snapshot_connections(String scope)
{
return(server.websocket_connection_ids(scope));
}
void websocket_exec_append_command(DValue command)
{
if(!context)
return;
context->resources.websocket_dispatch_commands.push(command);
}
DValue websocket_exec_make_message_command(String action, String message, bool binary)
{
DValue command;
command["action"] = action;
command["binary"].set_bool(binary);
command["message_b64"] = base64_encode(message);
return(command);
}
DValue websocket_exec_make_close_command(String connection_id, u16 status_code = 1000, String reason = "")
{
DValue command;
command["action"] = "close";
command["connection_id"] = connection_id;
command["status_code"] = (f64)status_code;
command["reason"] = reason;
return(command);
}
void websocket_exec_apply_command(DValue command)
{
String action = command["action"].to_string();
if(action == "broadcast")
{
bool ok = false;
String payload = base64_decode(command["message_b64"].to_string(), ok);
if(!ok)
return;
server.websocket_broadcast(command["scope"].to_string(), payload, command["binary"].to_bool());
return;
}
if(action == "send_to")
{
bool ok = false;
String payload = base64_decode(command["message_b64"].to_string(), ok);
if(!ok)
return;
server.websocket_send_to(command["connection_id"].to_string(), payload, command["binary"].to_bool());
return;
}
if(action == "close")
{
server.websocket_close(
command["connection_id"].to_string(),
(u16)command["status_code"].to_u64(),
command["reason"].to_string()
);
}
}
void websocket_exec_apply_result(DValue result)
{
if(result["type"].to_string() != "result")
return;
String inflight_connection_id = websocket_exec_inflight_job["connection_id"].to_string();
String result_connection_id = result["connection_id"].to_string();
if(inflight_connection_id != "" && result_connection_id != "" && inflight_connection_id != result_connection_id)
{
printf("(!) websocket dispatch result mismatch: expected %s got %s\n",
inflight_connection_id.c_str(),
result_connection_id.c_str());
}
FastCGIServer::Connection* connection = websocket_find_connection(result_connection_id);
if(connection)
connection->websocket_state = result["connection_state"];
result["commands"].each([] (DValue command, String) {
websocket_exec_apply_command(command);
});
websocket_exec_inflight_job.clear();
}
void websocket_exec_handle_ipc_line(String line)
{
line = trim(line);
if(line == "")
return;
DValue result = json_decode(line);
websocket_exec_apply_result(result);
}
void websocket_exec_read_results()
{
if(websocket_exec_fd == -1)
return;
char buffer[4096];
for(;;)
{
ssize_t read_result = read(websocket_exec_fd, buffer, sizeof(buffer));
if(read_result == 0)
{
printf("(!) websocket executor disconnected\n");
websocket_exec_fail_inflight_job();
websocket_exec_clear_ipc_state();
return;
}
if(read_result < 0)
{
if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
break;
perror("websocket executor read");
websocket_exec_fail_inflight_job();
websocket_exec_clear_ipc_state();
return;
}
websocket_exec_read_buffer.append(buffer, read_result);
for(;;)
{
size_t line_end = websocket_exec_read_buffer.find('\n');
if(line_end == String::npos)
break;
String line = websocket_exec_read_buffer.substr(0, line_end);
websocket_exec_read_buffer.erase(0, line_end + 1);
websocket_exec_handle_ipc_line(line);
}
}
}
void websocket_exec_flush_queue()
{
if(websocket_exec_fd == -1)
return;
if(websocket_exec_write_buffer == "" && !websocket_exec_has_inflight_job() && websocket_exec_pending_jobs.size() > 0)
{
websocket_exec_inflight_job = websocket_exec_pending_jobs.front();
websocket_exec_pending_jobs.pop_front();
websocket_exec_write_buffer = websocket_exec_inflight_job["serialized"].to_string();
}
while(websocket_exec_write_buffer != "")
{
ssize_t write_result = write(websocket_exec_fd, websocket_exec_write_buffer.data(), websocket_exec_write_buffer.length());
if(write_result < 0)
{
if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
return;
perror("websocket executor write");
websocket_exec_fail_inflight_job();
websocket_exec_clear_ipc_state();
return;
}
if(write_result == 0)
return;
websocket_exec_write_buffer.erase(0, write_result);
}
}
Request websocket_exec_build_event_request(DValue job, String message)
{
Request event_request;
event_request.server = &server_state;
event_request.params = job["params"].to_stringmap();
event_request.params["REQUEST_METHOD"] = "WEBSOCKET";
prepare_request_body_maps(event_request);
event_request.resources.is_websocket = true;
event_request.resources.websocket_connection_id = job["connection_id"].to_string();
event_request.resources.websocket_scope = job["scope"].to_string();
event_request.resources.websocket_opcode = (u8)job["opcode"].to_u64();
event_request.resources.websocket_is_binary = job["is_binary"].to_bool();
event_request.resources.websocket_is_text = job["is_text"].to_bool();
event_request.resources.websocket_dispatch_capture = true;
job["scope_connections"].each([&] (DValue item, String) {
event_request.resources.websocket_scope_connection_ids.push_back(item.to_string());
});
event_request.connection = job["connection_state"];
event_request.stats.time_init = time_precise();
event_request.stats.time_start = event_request.stats.time_init;
event_request.random_index = 0;
event_request.random_seed = request_seed_from_time(event_request.stats.time_start);
event_request.response_code = "WEBSOCKET";
event_request.header["Content-Type"] = server_state.config["CONTENT_TYPE"];
event_request.in = message;
event_request.params["WS_MESSAGE"] = message;
event_request.params["WS_CONNECTION_ID"] = event_request.resources.websocket_connection_id;
event_request.params["WS_SCOPE"] = event_request.resources.websocket_scope;
event_request.params["WS_CONNECTION_COUNT"] = (f64)event_request.resources.websocket_scope_connection_ids.size();
event_request.params["WS_OPCODE"] = (f64)event_request.resources.websocket_opcode;
event_request.params["WS_MESSAGE_TYPE"] = (event_request.resources.websocket_is_binary ? "BINARY" : "TEXT");
event_request.params["WS_DOCUMENT_URI"] = first(
event_request.params["DOCUMENT_URI"],
event_request.params["REQUEST_URI"]
);
return(event_request);
}
bool websocket_exec_send_response(int fd, DValue response)
{
String encoded = json_encode(response) + "\n";
size_t offset = 0;
while(offset < encoded.length())
{
ssize_t write_result = write(fd, encoded.data() + offset, encoded.length() - offset);
if(write_result < 0)
{
if(errno == EINTR)
continue;
return(false);
}
offset += (size_t)write_result;
}
return(true);
}
void websocket_exec_process_job_line(int fd, String line)
{
line = trim(line);
if(line == "")
return;
DValue job = json_decode(line);
if(job["type"].to_string() != "dispatch")
return;
bool decoded = false;
String message = base64_decode(job["message_b64"].to_string(), decoded);
if(!decoded)
{
printf("(!) invalid websocket IPC payload for %s\n", job["connection_id"].to_string().c_str());
return;
}
Request event_request = websocket_exec_build_event_request(job, message);
Request* previous_context = set_active_request(event_request);
server_state.request_count += 1;
compiler_invoke_websocket(&event_request, event_request.params["SCRIPT_FILENAME"]);
if(event_request.session_id.length() > 0)
save_session_data(event_request.session_id, event_request.session);
cleanup_mysql_connections();
cleanup_sqlite_connections();
DValue response;
response["type"] = "result";
response["connection_id"] = event_request.resources.websocket_connection_id;
response["connection_state"] = event_request.connection;
response["commands"] = event_request.resources.websocket_dispatch_commands;
restore_active_request(previous_context);
if(!websocket_exec_send_response(fd, response))
exit(1);
}
void websocket_exec_child_loop(int fd)
{
Request background_context;
my_pid = getpid();
context = &background_context;
close_inherited_server_sockets();
signal(SIGSEGV, on_segfault);
signal(SIGABRT, on_segfault);
signal(SIGBUS, on_segfault);
signal(SIGILL, on_segfault);
signal(SIGFPE, on_segfault);
signal(SIGPIPE, SIG_IGN);
setpriority(PRIO_PROCESS, 0, 5);
String read_buffer;
char buffer[4096];
for(;;)
{
ssize_t read_result = read(fd, buffer, sizeof(buffer));
if(read_result == 0)
exit(0);
if(read_result < 0)
{
if(errno == EINTR)
continue;
exit(1);
}
read_buffer.append(buffer, read_result);
for(;;)
{
size_t line_end = read_buffer.find('\n');
if(line_end == String::npos)
break;
String line = read_buffer.substr(0, line_end);
read_buffer.erase(0, line_end + 1);
websocket_exec_process_job_line(fd, line);
}
}
}
bool websocket_exec_alive()
{
return(websocket_exec_pid > 0 && task_kill(websocket_exec_pid, 0) == 0 && websocket_exec_fd != -1);
}
void ensure_websocket_executor()
{
if(!websocket_exec_enabled_for_process())
return;
if(websocket_exec_alive())
return;
if(websocket_exec_pid > 0 || websocket_exec_fd != -1)
{
websocket_exec_fail_inflight_job();
websocket_exec_clear_ipc_state();
websocket_exec_pid = 0;
}
int sockets[2] = {-1, -1};
if(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) != 0)
{
perror("socketpair");
return;
}
pid_t p = fork();
if(p < 0)
{
perror("fork");
close(sockets[0]);
close(sockets[1]);
return;
}
if(p == 0)
{
parent_pid = getppid();
file_release_process_locks("websocket executor fork");
prctl(PR_SET_PDEATHSIG, SIGHUP);
close(sockets[0]);
websocket_exec_child_loop(sockets[1]);
exit(0);
}
close(sockets[1]);
if(!websocket_ipc_set_nonblocking(sockets[0]))
{
printf("(!) failed to set websocket executor socket nonblocking\n");
close(sockets[0]);
return;
}
websocket_exec_fd = sockets[0];
websocket_exec_pid = p;
printf("(P) websocket executor spawned: PID %i\n", p);
}
void websocket_exec_tick()
{
if(!websocket_exec_enabled_for_process())
return;
if(!websocket_exec_alive())
{
websocket_exec_fail_inflight_job();
websocket_exec_clear_ipc_state();
websocket_exec_pid = 0;
ensure_websocket_executor();
}
websocket_exec_read_results();
websocket_exec_flush_queue();
}
}
String current_ws_scope()
{
@@ -682,13 +218,6 @@ bool ws_is_binary()
StringList ws_connections(String scope)
{
if(context && context->resources.websocket_dispatch_capture)
{
String normalized_scope = normalize_ws_scope(scope);
if(normalized_scope == context->resources.websocket_scope)
return(context->resources.websocket_scope_connection_ids);
return(StringList());
}
return(server.websocket_connection_ids(normalize_ws_scope(scope)));
}
@@ -699,26 +228,11 @@ u64 ws_connection_count(String scope)
bool ws_send(String message, bool binary, String scope)
{
String normalized_scope = normalize_ws_scope(scope);
if(context && context->resources.websocket_dispatch_capture)
{
DValue command = websocket_exec_make_message_command("broadcast", message, binary);
command["scope"] = normalized_scope;
websocket_exec_append_command(command);
return(true);
}
return(server.websocket_broadcast(normalized_scope, message, binary) > 0);
return(server.websocket_broadcast(normalize_ws_scope(scope), message, binary) > 0);
}
bool ws_send_to(String connection_id, String message, bool binary)
{
if(context && context->resources.websocket_dispatch_capture)
{
DValue command = websocket_exec_make_message_command("send_to", message, binary);
command["connection_id"] = connection_id;
websocket_exec_append_command(command);
return(true);
}
return(server.websocket_send_to(connection_id, message, binary));
}
@@ -728,11 +242,6 @@ bool ws_close(String connection_id)
connection_id = ws_connection_id();
if(connection_id == "")
return(false);
if(context && context->resources.websocket_dispatch_capture)
{
websocket_exec_append_command(websocket_exec_make_close_command(connection_id));
return(true);
}
return(server.websocket_close(connection_id));
}
@@ -861,7 +370,7 @@ int handle_cli_complete(FastCGIRequest& request)
String cli_unit = compiler_normalize_unit_path(&request, script_filename);
if(wasm_backend_should_handle(request, cli_unit))
{
String wasm_error = wasm_backend_serve(request, cli_unit, wasm_kind::CLI);
String wasm_error = wasm_backend_serve(request, cli_unit, "cli");
if(wasm_error != "")
{
request.set_status(500, "Internal Server Error");
@@ -913,11 +422,6 @@ int handle_data(FastCGIRequest& request) {
}
int handle_complete(FastCGIRequest& request) {
// The event handler can also be a class member function. This
// event occurs when the parameters and standard input streams are
// both closed, and thus the request is complete.
// printf("(i) request handle\n");
Request* previous_context = set_active_request(request);
server_state.request_count += 1;
request.server = &server_state;
@@ -958,10 +462,10 @@ int handle_complete(FastCGIRequest& request) {
// wasm invocation: Wasmtime uses host signals internally to implement
// guest traps, and the native handler would otherwise turn a clean
// guest trap into a native fatal signal. Sets failure_* on error.
auto serve_via_wasm = [&](const String& entry_unit, int32_t kind) {
auto serve_via_wasm = [&](const String& entry_unit, const String& handler) {
request_fault_active = 0;
restore_request_fault_handlers();
String wasm_error = wasm_backend_serve(request, entry_unit, kind);
String wasm_error = wasm_backend_serve(request, entry_unit, handler);
install_request_fault_handlers();
request_fault_active = 1;
if(wasm_error != "")
@@ -973,22 +477,54 @@ int handle_complete(FastCGIRequest& request) {
};
String entry_unit = compiler_normalize_unit_path(&request, request.params["SCRIPT_FILENAME"]);
if(request.resources.is_cli)
if(request.params["UCE_WS"] == "1")
{
// A WS message the broker forwarded here: rebuild the connection
// context the broker passed as params, then run __uce_websocket.
request.resources.websocket_connection_id = request.params["UCE_WS_CONNECTION_ID"];
request.resources.websocket_scope = request.params["UCE_WS_SCOPE"];
request.resources.websocket_opcode = (u8)int_val(request.params["UCE_WS_OPCODE"]);
request.resources.websocket_is_binary = request.params["UCE_WS_BINARY"] == "1";
bool msg_ok = false;
request.in = base64_decode(request.params["UCE_WS_MESSAGE"], msg_ok);
for(auto& id : split(request.params["UCE_WS_CONNECTIONS"], "\n"))
if(id != "")
request.resources.websocket_scope_connection_ids.push_back(id);
bool decoded = false;
String state_raw = base64_decode(request.params["UCE_WS_STATE"], decoded);
if(state_raw != "")
{
DValue state; String e;
if(ucb_decode(state_raw, state, &e))
request.connection = state;
}
if(wasm_backend_should_handle(request, entry_unit))
serve_via_wasm(entry_unit, "websocket");
else
compiler_invoke_websocket(&request, request.params["SCRIPT_FILENAME"]);
}
else if(request.resources.is_cli)
{
if(wasm_backend_should_handle(request, entry_unit))
serve_via_wasm(entry_unit, wasm_kind::CLI);
serve_via_wasm(entry_unit, "cli");
else
compiler_invoke_cli(&request, request.params["SCRIPT_FILENAME"]);
}
else if(request.params["UCE_SERVE_HTTP"] == "1")
// W7c pending: the custom-server dispatcher is forked from a worker
// that already holds a live Wasmtime engine, so re-creating an engine
// in that child currently hangs under the in-process dispatcher path.
// The fix is to have the broker forward to the worker pool rather than
// render in the fork; until then serve_http renders natively.
compiler_invoke_serve_http(&request, request.params["SCRIPT_FILENAME"], request.params["UCE_SERVE_HTTP_FUNCTION"]);
{
// W7c: this runs in a normal worker (clean engine) — the custom-server
// dispatcher forwarded the request here via FastCGI rather than render
// wasm in its own fork (Wasmtime cannot be re-created across fork).
if(wasm_backend_should_handle(request, entry_unit))
{
String fn = request.params["UCE_SERVE_HTTP_FUNCTION"];
serve_via_wasm(entry_unit, fn == "" ? String("serve_http") : "serve_http:" + fn);
}
else
compiler_invoke_serve_http(&request, request.params["SCRIPT_FILENAME"], request.params["UCE_SERVE_HTTP_FUNCTION"]);
}
else if(wasm_backend_should_handle(request, entry_unit))
serve_via_wasm(entry_unit, wasm_kind::RENDER);
serve_via_wasm(entry_unit, "render");
else
compiler_invoke(&request, request.params["SCRIPT_FILENAME"]);
}
@@ -1027,40 +563,6 @@ int handle_complete(FastCGIRequest& request) {
return request.flags.status;
}
int handle_websocket_message(FastCGIRequest& request, const String& message, u8 opcode)
{
ensure_websocket_executor();
if(!websocket_exec_alive())
{
printf("(!) websocket executor unavailable for %s\n", request.resources.websocket_connection_id.c_str());
server.websocket_close(request.resources.websocket_connection_id, 1011, "websocket handler unavailable");
return(0);
}
DValue job;
job["type"] = "dispatch";
job["connection_id"] = request.resources.websocket_connection_id;
job["scope"] = request.resources.websocket_scope;
job["opcode"] = (f64)opcode;
job["is_binary"].set_bool(request.resources.websocket_is_binary);
job["is_text"].set_bool(request.resources.websocket_is_text);
job["message_b64"] = base64_encode(message);
if(request.resources.websocket_connection_state)
job["connection_state"] = *request.resources.websocket_connection_state;
for(auto& item : request.params)
job["params"][item.first] = item.second;
for(auto& connection_id : websocket_exec_snapshot_connections(request.resources.websocket_scope))
{
DValue snapshot_item;
snapshot_item = connection_id;
job["scope_connections"].push(snapshot_item);
}
websocket_exec_queue_job(job);
return 0;
}
volatile bool termination_signal_received = false;
@@ -1218,6 +720,31 @@ StringMap custom_server_read_config(String key)
static String custom_server_dispatcher_key = "";
// Forward a request to a normal worker over the FastCGI socket and populate the
// response from the worker's reply. Shared by the connection brokers (the custom
// HTTP server dispatcher and the WS broker): they own a listening socket but
// render through the clean-engine worker pool rather than host a Wasmtime engine
// (which cannot be re-created across fork). The caller sets the routing params
// (UCE_SERVE_HTTP / UCE_WS, SCRIPT_FILENAME, etc.) before calling.
int forward_request_to_worker(FastCGIRequest& request, u32 timeout_seconds = 30)
{
String fcgi_socket = first(server_state.config["FCGI_SOCKET_PATH"], "/run/uce.sock");
FcgiForwardResult fwd = fcgi_forward_request(fcgi_socket, request.params, request.in, timeout_seconds);
request.ob_start();
if(!fwd.ok)
{
request.set_status(502, "Bad Gateway");
request.header["Content-Type"] = "text/plain; charset=utf-8";
(*request.ob) << "worker forward failed: " << fwd.error << "\n";
return(request.flags.status);
}
request.set_status(fwd.status);
for(auto& h : fwd.headers)
request.header[h.first] = h.second;
request.ob->write(fwd.body.data(), fwd.body.size());
return(request.flags.status);
}
int custom_server_http_complete(FastCGIRequest& request)
{
String key = first(request.params["UCE_SERVER_KEY"], custom_server_dispatcher_key);
@@ -1237,12 +764,193 @@ int custom_server_http_complete(FastCGIRequest& request)
request.params["UCE_SERVE_HTTP_FUNCTION"] = cfg["function"];
request.params["SCRIPT_FILENAME"] = cfg["file"];
u64 timeout = config_map_u64(server_state.config, "CUSTOM_SERVER_HANDLER_TIMEOUT_SECONDS", 30);
if(timeout > 0)
alarm(timeout);
int status = handle_complete(request);
if(timeout > 0)
alarm(0);
return(status);
return(forward_request_to_worker(request, timeout > 0 ? (u32)timeout : 30));
}
// ---- central WS broker -----------------------------------------------------
// Connect to a unix socket (blocking — local, sub-ms) and set it non-blocking.
static int ws_broker_connect_unix(const String& path)
{
int fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
if(fd < 0)
return(-1);
struct sockaddr_un addr;
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
if(::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
{
::close(fd);
return(-1);
}
int fl = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, fl | O_NONBLOCK);
return(fd);
}
// Apply a ws_* dispatch batch a worker flushed at workspace teardown. The broker
// owns every connection, so any target (id / scope / broadcast / close) resolves.
void ws_broker_apply_commands(FastCGIRequest& request)
{
DValue batch;
String err;
bool ok = ucb_decode(request.in, batch, &err);
if(ok)
{
if(DValue* cmds = batch.key("commands"))
cmds->each([&](const DValue& cmd_const, String) {
DValue cmd = cmd_const; // .each yields const; copy for [] access
String action = cmd["action"].to_string();
bool ok = false;
String msg = base64_decode(cmd["message_b64"].to_string(), ok);
bool binary = cmd["binary"].to_bool();
if(action == "broadcast")
ws_broker.websocket_broadcast(cmd["scope"].to_string(), msg, binary);
else if(action == "send_to")
ws_broker.websocket_send_to(cmd["connection_id"].to_string(), msg, binary);
else if(action == "close")
ws_broker.websocket_close(cmd["connection_id"].to_string(),
(u16)cmd["status_code"].to_u64(), cmd["reason"].to_string());
});
// persist any updated per-connection state back onto the live connection
String cid = batch["connection_id"].to_string();
if(cid != "" && batch.key("connection_state"))
for(auto& item : ws_broker.client_sockets)
if(item.second->is_websocket && item.second->websocket_connection_id == cid)
item.second->websocket_state = batch["connection_state"];
}
request.set_status(200);
request.ob_start();
}
// on_complete for the broker: either a worker's command flush, or an un-upgraded
// HTTP request that hit the WS port (forwarded to the pool via the shared facility).
int ws_broker_complete(FastCGIRequest& request)
{
if(request.params["UCE_WS_DISPATCH"] == "1")
{
ws_broker_apply_commands(request);
return(request.flags.status);
}
return(forward_request_to_worker(request));
}
// A complete WS message: fire a render to the worker pool WITHOUT blocking the
// broker loop (the connection identity + state ride as params; the message is
// the body). The worker renders __uce_websocket and flushes ws_* back to us.
int ws_broker_ws_message(FastCGIRequest& request, const String& message, u8 opcode)
{
StringMap params;
params["SCRIPT_FILENAME"] = request.params["SCRIPT_FILENAME"];
params["REQUEST_METHOD"] = "GET";
// handle_request() rejects any request without REQUEST_URI before on_complete.
params["REQUEST_URI"] = first(request.params["REQUEST_URI"], request.params["DOCUMENT_URI"],
request.params["SCRIPT_FILENAME"]);
params["UCE_WS"] = "1";
// The message rides as a param (empty STDIN) so the forwarded request
// completes cleanly — a STDIN body makes the FastCGI transport flush a
// premature response before on_complete (handle_complete) ever runs.
params["UCE_WS_MESSAGE"] = base64_encode(message);
params["UCE_WS_CONNECTION_ID"] = request.resources.websocket_connection_id;
params["UCE_WS_SCOPE"] = request.resources.websocket_scope;
params["UCE_WS_OPCODE"] = std::to_string((int)opcode);
params["UCE_WS_BINARY"] = request.resources.websocket_is_binary ? "1" : "0";
params["UCE_WS_CONNECTIONS"] = join(ws_broker.websocket_connection_ids(request.resources.websocket_scope), "\n");
if(request.resources.websocket_connection_state)
params["UCE_WS_STATE"] = base64_encode(ucb_encode(*request.resources.websocket_connection_state));
int fd = ws_broker_connect_unix(first(server_state.config["FCGI_SOCKET_PATH"], "/run/uce.sock"));
if(fd < 0)
return(0);
ws_broker_outbound[fd] = fcgi_build_request(params, "");
return(0);
}
// Drive in-flight render forwards non-blocking: finish writing the request, then
// drain/discard the reply (the unit's ws_* came back via the command socket).
void ws_broker_drain_outbound()
{
for(auto it = ws_broker_outbound.begin(); it != ws_broker_outbound.end(); )
{
int fd = it->first;
String& pending = it->second;
bool done = false;
if(!pending.empty())
{
ssize_t n = ::send(fd, pending.data(), pending.size(), MSG_NOSIGNAL | MSG_DONTWAIT);
if(n > 0)
pending.erase(0, n);
else if(n < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
done = true;
}
if(!done && pending.empty())
{
char buf[8192];
ssize_t r = ::recv(fd, buf, sizeof(buf), MSG_DONTWAIT);
if(r == 0)
done = true; // worker closed the connection: render complete
else if(r < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
done = true;
}
if(done)
{
::close(fd);
it = ws_broker_outbound.erase(it);
}
else
++it;
}
}
void run_ws_broker()
{
my_pid = getpid();
close_inherited_server_sockets(); // drop the worker listeners we inherited
install_process_fault_handlers();
ws_broker.calls_until_termination = -1;
// The broker renders nothing itself, so accept every request through to
// on_complete (it either forwards to the pool or applies a ws_* batch).
ws_broker.on_request = [](FastCGIRequest&) { return 0; };
ws_broker.on_data = [](FastCGIRequest&) { return 0; };
ws_broker.on_complete = &ws_broker_complete;
ws_broker.on_websocket_message = &ws_broker_ws_message;
if(server_state.config["HTTP_PORT"] != "")
ws_broker.listen_http(int_val(server_state.config["HTTP_PORT"]));
if(server_state.config["WS_BROKER_SOCKET_PATH"] != "")
{
ws_broker.listen(server_state.config["WS_BROKER_SOCKET_PATH"]);
chmod(server_state.config["WS_BROKER_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP);
}
for(;;)
{
ws_broker.process(50);
ws_broker_drain_outbound();
}
}
bool ws_broker_alive()
{
return(ws_broker_pid > 0 && task_kill(ws_broker_pid, 0) == 0);
}
void ensure_ws_broker()
{
if(ws_broker_alive())
return;
pid_t p = fork();
if(p < 0)
{
perror("fork ws_broker");
return;
}
if(p == 0)
{
run_ws_broker();
exit(0);
}
ws_broker_pid = p;
printf("(P) WS broker spawned: PID %i\n", p);
}
void custom_server_http_dispatcher_loop(String key)
@@ -1512,29 +1220,17 @@ void ensure_proactive_compiler()
void listen_for_connections()
{
install_process_fault_handlers();
if(worker_accepts_http)
{
// Keep the dedicated HTTP/WebSocket worker alive. If it ages out like a
// normal FastCGI worker, nginx can connect to the shared listening socket
// while no child is actively accepting, which makes `.ws.uce` page loads
// appear to hang until the parent respawns a replacement worker.
server.calls_until_termination = -1;
}
if(!worker_accepts_http)
server.close_http_listeners();
// Workers are uniform FastCGI/CLI renderers; the WS broker owns the HTTP/WS
// port and every connection, so workers never accept raw HTTP themselves.
server.close_http_listeners();
server.on_request = &handle_request;
server.on_data = &handle_data;
server.on_complete = &handle_complete;
server.on_cli_complete = &handle_cli_complete;
server.on_websocket_message = &handle_websocket_message;
if(worker_accepts_http)
ensure_websocket_executor();
for(;;)
{
file_release_process_locks("worker loop cleanup");
server.process(worker_accepts_http ? 50 : -1);
if(worker_accepts_http)
websocket_exec_tick();
server.process(-1);
}
}
@@ -1566,8 +1262,8 @@ void init_base_process()
chmod(server_state.config["CLI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP);
}
if(server_state.config["HTTP_PORT"] != "")
server.listen_http(int_val(server_state.config["HTTP_PORT"]));
// HTTP_PORT (WebSocket + raw HTTP) is owned by the dedicated WS broker, not
// the worker pool — see ensure_ws_broker(). Workers handle FastCGI + CLI only.
mkdir(server_state.config["BIN_DIRECTORY"]);
mkdir(server_state.config["TMP_UPLOAD_PATH"]);
@@ -1588,6 +1284,7 @@ int main(int argc, char** argv)
init_base_process();
ensure_proactive_compiler();
ensure_ws_broker();
for(;;)
{
@@ -1596,15 +1293,18 @@ int main(int argc, char** argv)
if(!termination_signal_received)
ensure_proactive_compiler();
// One dedicated WS broker owns all connections; respawn it if it dies
// (live connections are lost on a broker restart, but unit-code crashes
// happen in workers, not here, so the broker stays up in practice).
if(!ws_broker_alive())
ws_broker_pid = 0;
if(!termination_signal_received)
ensure_ws_broker();
while(workers.size() < int_val(server_state.config["WORKER_COUNT"]))
{
if(!termination_signal_received)
{
worker_accepts_http = (http_worker_pid == 0 || workers.count(http_worker_pid) == 0);
pid_t child_pid = spawn_subprocess(listen_for_connections);
if(child_pid > 0 && worker_accepts_http)
http_worker_pid = child_pid;
}
spawn_subprocess(listen_for_connections);
}
sleep(1);
}