Bound request-time Wasm compilation
This commit is contained in:
+50
-96
@@ -542,90 +542,9 @@ static void uce_job_reap()
|
||||
}
|
||||
}
|
||||
|
||||
static DValue uce_process_exec(String cmd, String input, StringMap env, u64 timeout_ms)
|
||||
{
|
||||
DValue r;
|
||||
r["exit_code"] = (f64)-1;
|
||||
r["stdout"] = "";
|
||||
r["stderr"] = "";
|
||||
r["timed_out"].set_bool(false);
|
||||
if(timeout_ms == 0) timeout_ms = 5000;
|
||||
int inpipe[2], outpipe[2], errpipe[2];
|
||||
if(pipe(inpipe) || pipe(outpipe) || pipe(errpipe)) { r["stderr"]="pipe failed"; return(r); }
|
||||
// The process-wide handler reaps background children; this caller owns this child's status.
|
||||
WasmSigchldBlock sigchld;
|
||||
unsigned int child_status_snapshot = child_exit_status_snapshot();
|
||||
pid_t pid = fork();
|
||||
if(pid < 0)
|
||||
{
|
||||
close(inpipe[0]); close(inpipe[1]); close(outpipe[0]); close(outpipe[1]); close(errpipe[0]); close(errpipe[1]);
|
||||
r["stderr"] = "fork failed";
|
||||
return(r);
|
||||
}
|
||||
if(pid == 0)
|
||||
{
|
||||
sigchld.restore();
|
||||
setpgid(0, 0);
|
||||
dup2(inpipe[0], 0); dup2(outpipe[1], 1); dup2(errpipe[1], 2);
|
||||
close(inpipe[0]); close(inpipe[1]); close(outpipe[0]); close(outpipe[1]); close(errpipe[0]); close(errpipe[1]);
|
||||
for(auto& kv : env) setenv(kv.first.c_str(), kv.second.c_str(), 1);
|
||||
execl("/bin/sh", "sh", "-c", cmd.c_str(), (char*)0);
|
||||
_exit(127);
|
||||
}
|
||||
setpgid(pid, pid);
|
||||
close(inpipe[0]); close(outpipe[1]); close(errpipe[1]);
|
||||
fcntl(inpipe[1], F_SETFL, fcntl(inpipe[1], F_GETFL, 0) | O_NONBLOCK);
|
||||
fcntl(outpipe[0], F_SETFL, fcntl(outpipe[0], F_GETFL, 0) | O_NONBLOCK);
|
||||
fcntl(errpipe[0], F_SETFL, fcntl(errpipe[0], F_GETFL, 0) | O_NONBLOCK);
|
||||
size_t input_off = 0; bool in_open = true, out_open = true, err_open = true; int status = 0; bool exited = false, status_valid = false;
|
||||
u64 deadline = wasm_monotonic_ms() + timeout_ms;
|
||||
while(out_open || err_open || !exited)
|
||||
{
|
||||
if(!exited)
|
||||
{
|
||||
pid_t w = waitpid(pid, &status, WNOHANG);
|
||||
if(w == pid) { exited = true; status_valid = true; }
|
||||
else if(w < 0 && errno == ECHILD)
|
||||
{
|
||||
u64 transfer_deadline = wasm_deadline_after_ms(50);
|
||||
do { status_valid = child_exit_status_take(pid, status, child_status_snapshot); if(!status_valid) sched_yield(); } while(!status_valid && wasm_monotonic_ms() < transfer_deadline);
|
||||
exited = true;
|
||||
if(!status_valid) r["stderr"] = r["stderr"].to_string() + "lost child exit status";
|
||||
}
|
||||
}
|
||||
if(in_open)
|
||||
{
|
||||
if(input_off < input.size()) { ssize_t n=write(inpipe[1], input.data()+input_off, input.size()-input_off); if(n>0) input_off += (size_t)n; else if(n<0 && errno!=EINTR && errno!=EAGAIN && errno!=EWOULDBLOCK) { close(inpipe[1]); in_open=false; } }
|
||||
else { close(inpipe[1]); in_open=false; }
|
||||
}
|
||||
char buf[4096];
|
||||
ssize_t n;
|
||||
while((n=read(outpipe[0], buf, sizeof(buf))) > 0) r["stdout"] = r["stdout"].to_string() + String(buf, n);
|
||||
if(n == 0 && out_open) { close(outpipe[0]); out_open=false; }
|
||||
while((n=read(errpipe[0], buf, sizeof(buf))) > 0) r["stderr"] = r["stderr"].to_string() + String(buf, n);
|
||||
if(n == 0 && err_open) { close(errpipe[0]); err_open=false; }
|
||||
if((out_open || err_open || !exited) && wasm_monotonic_ms() >= deadline)
|
||||
{
|
||||
r["timed_out"].set_bool(true);
|
||||
kill(-pid, SIGKILL);
|
||||
kill(pid, SIGKILL);
|
||||
if(!exited)
|
||||
status_valid = waitpid(pid, &status, 0) == pid;
|
||||
exited = true;
|
||||
if(in_open) { close(inpipe[1]); in_open=false; }
|
||||
if(out_open) { close(outpipe[0]); out_open=false; }
|
||||
if(err_open) { close(errpipe[0]); err_open=false; }
|
||||
}
|
||||
if((out_open || err_open || !exited)) usleep(10000);
|
||||
}
|
||||
if(status_valid && WIFEXITED(status)) r["exit_code"] = (f64)WEXITSTATUS(status);
|
||||
else if(status_valid && WIFSIGNALED(status)) r["exit_code"] = (f64)(128 + WTERMSIG(status));
|
||||
return(r);
|
||||
}
|
||||
|
||||
static DValue uce_shell_exec_spec(const DValue& spec)
|
||||
{
|
||||
return(uce_process_exec(spec.key("cmd") ? spec.key("cmd")->to_string() : String(""), spec.key("stdin") ? spec.key("stdin")->to_string() : String(""), spec.key("env") ? spec.key("env")->to_stringmap() : StringMap(), spec.key("timeout_ms") ? spec.key("timeout_ms")->to_u64(5000) : 5000));
|
||||
return(process_exec(spec.key("cmd") ? spec.key("cmd")->to_string() : String(""), spec.key("stdin") ? spec.key("stdin")->to_string() : String(""), spec.key("env") ? spec.key("env")->to_stringmap() : StringMap(), spec.key("timeout_ms") ? spec.key("timeout_ms")->to_u64(5000) : 5000));
|
||||
}
|
||||
|
||||
static void uce_job_finish(u64 id, DValue result, String final_state="done")
|
||||
@@ -1606,8 +1525,7 @@ public:
|
||||
return(UINT64_MAX);
|
||||
if(now >= invocation_deadline)
|
||||
return(0);
|
||||
u64 remaining_us = (u64)std::chrono::duration_cast<std::chrono::microseconds>(invocation_deadline - now).count();
|
||||
return(remaining_us / 1000 + (remaining_us % 1000 != 0));
|
||||
return((u64)std::chrono::duration_cast<std::chrono::milliseconds>(invocation_deadline - now).count());
|
||||
}
|
||||
|
||||
bool invocation_expired(InvocationClock::time_point now = InvocationClock::now()) const
|
||||
@@ -1659,7 +1577,7 @@ public:
|
||||
InvocationClock::time_point previous_deadline;
|
||||
u64 previous_budget_ms = 0;
|
||||
public:
|
||||
InvocationScope(WasmWorkspace& workspace, u64 timeout_cap_ms = 0, bool force_new = false) : workspace(workspace)
|
||||
InvocationScope(WasmWorkspace& workspace, u64 timeout_cap_ms = 0, bool force_new = false, u64 reported_budget_ms = 0) : workspace(workspace)
|
||||
{
|
||||
if(!workspace.invocation_active || force_new)
|
||||
{
|
||||
@@ -1668,10 +1586,10 @@ public:
|
||||
previous_deadline = workspace.invocation_deadline;
|
||||
previous_budget_ms = workspace.invocation_budget_ms;
|
||||
u64 budget_ms = workspace.worker.cfg.invocation_timeout_ms;
|
||||
if(timeout_cap_ms > 0)
|
||||
if(timeout_cap_ms != UINT64_MAX)
|
||||
budget_ms = std::min(budget_ms, timeout_cap_ms);
|
||||
workspace.invocation_active = true;
|
||||
workspace.invocation_budget_ms = budget_ms;
|
||||
workspace.invocation_budget_ms = reported_budget_ms > 0 ? reported_budget_ms : budget_ms;
|
||||
workspace.invocation_deadline = InvocationClock::now() + std::chrono::milliseconds(budget_ms);
|
||||
}
|
||||
workspace.arm_guest_deadline(workspace.ctx());
|
||||
@@ -1786,6 +1704,8 @@ public:
|
||||
phase_start = now;
|
||||
return(elapsed);
|
||||
};
|
||||
if(invocation_expired())
|
||||
return(invocation_timeout_error());
|
||||
auto cx = ctx();
|
||||
store.limiter(worker.cfg.memory_limit, -1, -1, -1, -1);
|
||||
arm_guest_deadline(cx);
|
||||
@@ -2096,7 +2016,9 @@ private:
|
||||
String trap_text(const wasmtime::TrapError& error)
|
||||
{
|
||||
String result = wasm_trace_collapse(String(error.message()));
|
||||
if(invocation_expired() && result.find("interrupt") != String::npos && result.find("UCE_INVOCATION_TIMEOUT:") == String::npos)
|
||||
bool invocation_deadline_interrupt = invocation_active &&
|
||||
(invocation_expired() || invocation_remaining_ms() <= worker.cfg.epoch_period_ms);
|
||||
if(invocation_deadline_interrupt && result.find("interrupt") != String::npos && result.find("UCE_INVOCATION_TIMEOUT:") == String::npos)
|
||||
result = invocation_timeout_error() + "\n" + result;
|
||||
struct Frame
|
||||
{
|
||||
@@ -2795,10 +2717,12 @@ private:
|
||||
// `handler` names the export ("render", "component:CARD", "cli",
|
||||
// "serve_http:named", "once") or is "exists" (probe only, loads nothing).
|
||||
int32_t component_resolve(const String& target, const String& handler, const String& current_unit,
|
||||
String& resolved_out, int32_t* once_slot_out = 0)
|
||||
String& resolved_out, int32_t* once_slot_out = 0, bool* compile_timed_out = 0)
|
||||
{
|
||||
if(once_slot_out)
|
||||
*once_slot_out = 0;
|
||||
if(compile_timed_out)
|
||||
*compile_timed_out = false;
|
||||
auto probe_start = std::chrono::steady_clock::now();
|
||||
auto record_probe = [&]() {
|
||||
component_resolve_count += 1;
|
||||
@@ -2925,7 +2849,21 @@ private:
|
||||
}
|
||||
}
|
||||
if(!artifact_exists || (stale && !can_serve_stale))
|
||||
get_shared_unit(context, resolved);
|
||||
{
|
||||
u64 remaining_ms = invocation_remaining_ms();
|
||||
bool timed_out = false;
|
||||
if(remaining_ms == 0)
|
||||
timed_out = true;
|
||||
else
|
||||
get_shared_unit_bounded(context, resolved, remaining_ms, &timed_out);
|
||||
if(timed_out)
|
||||
{
|
||||
if(compile_timed_out)
|
||||
*compile_timed_out = true;
|
||||
record_probe();
|
||||
return(0);
|
||||
}
|
||||
}
|
||||
component_artifact_total_us += (u64)std::chrono::duration_cast<std::chrono::microseconds>(
|
||||
std::chrono::steady_clock::now() - artifact_start).count();
|
||||
|
||||
@@ -3342,7 +3280,7 @@ private:
|
||||
{
|
||||
u64 remaining_ms = self->invocation_remaining_ms();
|
||||
u64 timeout_ms = std::max<u64>(1, self->bounded_hostcall_timeout_ms(5000));
|
||||
DValue execution = uce_process_exec(cmd + " 2>&1", "", StringMap(), timeout_ms);
|
||||
DValue execution = process_exec(cmd + " 2>&1", "", StringMap(), timeout_ms);
|
||||
if(execution["timed_out"].to_bool())
|
||||
{
|
||||
String kind = self->invocation_expired() || remaining_ms <= 5000 ?
|
||||
@@ -3850,7 +3788,14 @@ private:
|
||||
}
|
||||
}
|
||||
else if(op == "compile")
|
||||
response["ok"].set_bool(unit_compile(request["path"].to_string()));
|
||||
{
|
||||
u64 remaining_ms = self->invocation_remaining_ms();
|
||||
bool timed_out = false;
|
||||
bool ok = remaining_ms > 0 && unit_compile_bounded(context, request["path"].to_string(), remaining_ms, &timed_out);
|
||||
if(timed_out || remaining_ms == 0)
|
||||
return(Trap(self->invocation_timeout_error()));
|
||||
response["ok"].set_bool(ok);
|
||||
}
|
||||
else if(op == "call")
|
||||
{
|
||||
DValue* param = request.key("param");
|
||||
@@ -4292,7 +4237,10 @@ private:
|
||||
self->hostcall_read(args[2].i32(), args[3].i32(), handler);
|
||||
self->hostcall_read(args[4].i32(), args[5].i32(), current);
|
||||
int32_t once_slot = 0;
|
||||
int32_t slot = self->component_resolve(target, handler, current, resolved, &once_slot);
|
||||
bool compile_timed_out = false;
|
||||
int32_t slot = self->component_resolve(target, handler, current, resolved, &once_slot, &compile_timed_out);
|
||||
if(compile_timed_out)
|
||||
return(Trap(self->invocation_timeout_error()));
|
||||
u32 cap = (u32)args[7].i32();
|
||||
if(cap > 0)
|
||||
{
|
||||
@@ -4335,7 +4283,7 @@ inline String wasm_worker_prepare(WasmWorker& worker)
|
||||
// ---- public entry: one request through one workspace -----------------------
|
||||
|
||||
inline WasmResponse wasm_worker_serve(WasmWorker& worker, const Request& request, const String& entry_source_path,
|
||||
const String& handler = "render")
|
||||
const String& handler = "render", u64 timeout_cap_ms = UINT64_MAX)
|
||||
{
|
||||
WasmResponse response;
|
||||
f64 serve_started = time_precise();
|
||||
@@ -4345,9 +4293,15 @@ inline WasmResponse wasm_worker_serve(WasmWorker& worker, const Request& request
|
||||
bool thread_runtime_profiled = worker.cfg.profile_thread_runtime && getrusage(RUSAGE_THREAD, &thread_runtime_start) == 0;
|
||||
int thread_cpu_start = thread_runtime_profiled ? sched_getcpu() : -1;
|
||||
WasmWorkspace workspace(worker);
|
||||
f64 setup_cpu_finished = wasm_thread_cpu_time();
|
||||
workspace.workspace_setup_us = (u64)std::chrono::duration_cast<std::chrono::microseconds>(
|
||||
u64 workspace_setup_us = (u64)std::chrono::duration_cast<std::chrono::microseconds>(
|
||||
std::chrono::steady_clock::now() - workspace_start).count();
|
||||
u64 workspace_setup_ms = workspace_setup_us / 1000 + (workspace_setup_us % 1000 != 0);
|
||||
u64 workspace_budget_ms = timeout_cap_ms;
|
||||
if(workspace_budget_ms != UINT64_MAX)
|
||||
workspace_budget_ms = workspace_budget_ms > workspace_setup_ms ? workspace_budget_ms - workspace_setup_ms : 0;
|
||||
WasmWorkspace::InvocationScope invocation(workspace, workspace_budget_ms, false, worker.cfg.invocation_timeout_ms);
|
||||
f64 setup_cpu_finished = wasm_thread_cpu_time();
|
||||
workspace.workspace_setup_us = workspace_setup_us;
|
||||
workspace.workspace_setup_cpu_us = cpu_started > 0 && setup_cpu_finished > cpu_started ?
|
||||
(u64)((setup_cpu_finished - cpu_started) * 1000000.0) : 0;
|
||||
workspace.set_perf_snapshot(my_pid, (u64)parent_pid, request.server ? request.server->request_count : 0,
|
||||
|
||||
Reference in New Issue
Block a user