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

Enforce absolute Wasm invocation deadlines

Browse Source
This commit is contained in:
udo
2026-07-19 00:44:18 +00:00
parent fd6d472187
commit a7576f3db7
10 changed files with 591 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wasm/worker.cpp
+350 -78
View File
@@ -148,6 +148,8 @@ struct WasmWorkerConfig
int64_t memory_limit = 512ll * 1024 * 1024;
u32 table_headroom = 4096;
u64 epoch_deadline_ticks = 200; // ticker period × ticks = CPU budget
u64 epoch_period_ms = 50;
u64 invocation_timeout_ms = 30000;
u64 mysql_persistent_pool_size = 8;
bool profile_hostcall_cpu = false;
bool profile_thread_runtime = false;
@@ -291,6 +293,130 @@ static u64 wasm_monotonic_ms()
return((u64)ts.tv_sec * 1000ull + (u64)ts.tv_nsec / 1000000ull);
}
class WasmSigchldBlock
{
sigset_t previous;
bool blocked = false;
public:
WasmSigchldBlock()
{
sigset_t mask;
sigemptyset(&mask);
sigaddset(&mask, SIGCHLD);
blocked = pthread_sigmask(SIG_BLOCK, &mask, &previous) == 0;
}
void restore()
{
if(blocked)
{
pthread_sigmask(SIG_SETMASK, &previous, 0);
blocked = false;
}
}
~WasmSigchldBlock() { restore(); }
};
static u64 wasm_deadline_after_ms(u64 timeout_ms)
{
u64 now = wasm_monotonic_ms();
return(timeout_ms > UINT64_MAX - now ? UINT64_MAX : now + timeout_ms);
}
static bool wasm_socket_wait(int fd, short events, u64 deadline)
{
while(true)
{
u64 now = wasm_monotonic_ms();
if(now >= deadline)
return(false);
u64 remaining_ms = deadline - now;
struct pollfd item = { fd, events, 0 };
int rc = poll(&item, 1, (int)std::min<u64>(INT_MAX, remaining_ms));
if(rc > 0)
return((item.revents & (events | POLLERR | POLLHUP)) != 0);
if(rc == 0)
return(false);
if(errno != EINTR)
return(false);
}
}
static u64 wasm_socket_connect_bounded(const String& host, u16 port, u64 timeout_ms)
{
int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if(fd < 0)
return(0);
int flags = fcntl(fd, F_GETFL, 0);
if(flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
{
close(fd);
return(0);
}
struct sockaddr_in address = {0};
address.sin_family = AF_INET;
address.sin_port = htons(port);
if(inet_pton(AF_INET, host.c_str(), &address.sin_addr) != 1)
{
close(fd);
return(0);
}
int rc = connect(fd, (struct sockaddr*)&address, sizeof(address));
if(rc != 0 && errno == EINPROGRESS && wasm_socket_wait(fd, POLLOUT, wasm_deadline_after_ms(timeout_ms)))
{
int error = 0;
socklen_t error_size = sizeof(error);
rc = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &error_size) == 0 && error == 0 ? 0 : -1;
}
if(rc != 0)
{
close(fd);
return(0);
}
if(fcntl(fd, F_SETFL, flags) != 0)
{
close(fd);
return(0);
}
if(fd == 0)
{
int moved = dup(fd);
close(fd);
fd = moved;
}
if(fd <= 0)
return(0);
if(context)
context->resources.sockets.push_back(fd);
return((u64)fd);
}
static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 timeout_ms)
{
int fd = (int)socket_fd;
u64 deadline = wasm_deadline_after_ms(timeout_ms);
size_t offset = 0;
while(offset < data.size())
{
if(!wasm_socket_wait(fd, POLLOUT, deadline))
return(false);
ssize_t written = send(fd, data.data() + offset, data.size() - offset, MSG_DONTWAIT | MSG_NOSIGNAL);
if(written > 0)
offset += (size_t)written;
else if(written < 0 && errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)
return(false);
}
return(true);
}
static String wasm_socket_read_bounded(u64 socket_fd, u32 max_length, u64 timeout_ms)
{
if(max_length == 0 || !wasm_socket_wait((int)socket_fd, POLLIN, wasm_deadline_after_ms(timeout_ms)))
return("");
std::vector<char> buffer(max_length);
ssize_t count = recv((int)socket_fd, buffer.data(), buffer.size(), MSG_DONTWAIT);
return(count > 0 ? String(buffer.data(), (size_t)count) : String(""));
}
static f64 wasm_thread_cpu_time()
{
struct timespec ts;
@@ -426,30 +552,50 @@ static DValue uce_process_exec(String cmd, String input, StringMap env, u64 time
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;
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;
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) { close(inpipe[1]); in_open=false; } }
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];
@@ -458,17 +604,22 @@ static DValue uce_process_exec(String cmd, String input, StringMap env, u64 time
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(!exited && wasm_monotonic_ms() >= deadline)
if((out_open || err_open || !exited) && wasm_monotonic_ms() >= deadline)
{
r["timed_out"].set_bool(true);
kill(-pid, SIGKILL);
kill(pid, SIGKILL);
waitpid(pid, &status, 0);
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(WIFEXITED(status)) { r["exit_code"] = (f64)WEXITSTATUS(status); r["timed_out"].set_bool(false); }
else if(WIFSIGNALED(status)) r["exit_code"] = (f64)(128 + WTERMSIG(status));
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);
}
@@ -514,27 +665,38 @@ static DValue uce_exec_argv_capture(std::vector<String> argv, String input, u64
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); }
WasmSigchldBlock sigchld;
unsigned int child_status_snapshot=child_exit_status_snapshot();
pid_t pid=fork();
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]);
std::vector<char*> args; for(auto& a: argv) args.push_back((char*)a.c_str()); args.push_back(0);
execvp(args[0], args.data()); _exit(127);
}
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);
}
setpgid(pid,pid);
close(inpipe[0]); close(outpipe[1]); close(errpipe[1]);
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,exited=false; int status=0; u64 deadline=wasm_monotonic_ms()+timeout_ms;
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,exited=false,status_valid=false; int status=0; 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; }
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) { close(inpipe[1]); in_open=false; } } else { close(inpipe[1]); in_open=false; } }
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(!exited && wasm_monotonic_ms() >= deadline) { r["timed_out"].set_bool(true); kill(pid,SIGKILL); waitpid(pid,&status,0); exited=true; }
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(WIFEXITED(status)) { r["exit_code"]=(f64)WEXITSTATUS(status); r["timed_out"].set_bool(false); } else if(WIFSIGNALED(status)) r["exit_code"]=(f64)(128+WTERMSIG(status));
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);
}
@@ -1433,6 +1595,98 @@ public:
worker.active_workspace = this;
}
using InvocationClock = std::chrono::steady_clock;
bool invocation_active = false;
InvocationClock::time_point invocation_deadline;
u64 invocation_budget_ms = 0;
u64 invocation_remaining_ms(InvocationClock::time_point now = InvocationClock::now()) const
{
if(!invocation_active)
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));
}
bool invocation_expired(InvocationClock::time_point now = InvocationClock::now()) const
{
return(invocation_active && now >= invocation_deadline);
}
String invocation_timeout_error() const
{
return("UCE_INVOCATION_TIMEOUT: wasm invocation exceeded " + std::to_string(invocation_budget_ms) + " ms");
}
void arm_guest_deadline(wasmtime::Store::Context context)
{
u64 ticks = worker.cfg.epoch_deadline_ticks;
if(invocation_active)
{
u64 remaining_ms = invocation_remaining_ms();
if(remaining_ms == 0)
ticks = 0;
else
{
u64 remaining_ticks = remaining_ms / worker.cfg.epoch_period_ms +
(remaining_ms % worker.cfg.epoch_period_ms != 0);
u64 segment_ms = worker.cfg.epoch_deadline_ticks > UINT64_MAX / worker.cfg.epoch_period_ms ?
UINT64_MAX : worker.cfg.epoch_deadline_ticks * worker.cfg.epoch_period_ms;
// One extra tick prevents the engine ticker's current phase from
// interrupting just before the absolute steady-clock deadline.
if(remaining_ms <= segment_ms)
ticks = remaining_ticks == UINT64_MAX ? UINT64_MAX : remaining_ticks + 1;
}
}
context.set_epoch_deadline(ticks);
}
u64 bounded_hostcall_timeout_ms(u64 requested_ms) const
{
if(!invocation_active)
return(requested_ms);
u64 remaining_ms = invocation_remaining_ms();
return(std::min(requested_ms, remaining_ms));
}
class InvocationScope
{
WasmWorkspace& workspace;
bool replaced = false;
bool previous_active = false;
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)
{
if(!workspace.invocation_active || force_new)
{
replaced = true;
previous_active = workspace.invocation_active;
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)
budget_ms = std::min(budget_ms, timeout_cap_ms);
workspace.invocation_active = true;
workspace.invocation_budget_ms = budget_ms;
workspace.invocation_deadline = InvocationClock::now() + std::chrono::milliseconds(budget_ms);
}
workspace.arm_guest_deadline(workspace.ctx());
}
~InvocationScope()
{
if(!replaced)
return;
workspace.invocation_active = previous_active;
workspace.invocation_deadline = previous_deadline;
workspace.invocation_budget_ms = previous_budget_ms;
workspace.arm_guest_deadline(workspace.ctx());
}
};
void set_perf_snapshot(u64 worker_pid, u64 parent_pid, u64 request_count,
f64 time_init, f64 time_params, f64 time_input, f64 time_start,
u64 ready_normalize_us, u64 ready_mutation_check_us, u64 ready_artifact_stat_us,
@@ -1534,7 +1788,7 @@ public:
};
auto cx = ctx();
store.limiter(worker.cfg.memory_limit, -1, -1, -1, -1);
cx.set_epoch_deadline(worker.cfg.epoch_deadline_ticks);
arm_guest_deadline(cx);
birth_policy_us = phase_us();
auto& module = *worker.core_module;
@@ -1708,6 +1962,7 @@ public:
// empty body, a missing cli/serve handler to a 404).
String invoke_entry(const String& entry_source_path, const String& handler, bool* handler_present = 0)
{
InvocationScope invocation(*this);
auto phase_started = std::chrono::steady_clock::now();
auto phase_us = [&]() {
auto now = std::chrono::steady_clock::now();
@@ -1719,6 +1974,8 @@ public:
size_t unit_index = 0;
String error = load_unit(entry_source_path, "entry", unit_index);
entry_load_us = phase_us();
if(invocation_expired())
return(invocation_timeout_error());
if(error != "")
return(error);
String handler_symbol = handler_export_symbol(handler);
@@ -1761,7 +2018,7 @@ public:
if(error != "")
return(error);
entry_link_us = phase_us();
auto result = entry->call(ctx(), { wasmtime::Val((int32_t)handler_slot), wasmtime::Val((int32_t)once_slot) });
auto result = call_guest(*entry, { wasmtime::Val((int32_t)handler_slot), wasmtime::Val((int32_t)once_slot) });
entry_dispatch_us = phase_us();
if(!result)
return(trap_text(result.err()));
@@ -1839,6 +2096,8 @@ 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)
result = invocation_timeout_error() + "\n" + result;
struct Frame
{
String module;
@@ -1952,6 +2211,13 @@ private:
return(std::nullopt);
}
wasmtime::TrapResult<std::vector<wasmtime::Val>> call_guest(wasmtime::Func& func, std::vector<wasmtime::Val> args)
{
auto context = ctx();
arm_guest_deadline(context);
return(func.call(context, args));
}
String call_core(const String& name, std::vector<int32_t> argv, int32_t* result_out)
{
auto func = core_func(name);
@@ -1960,7 +2226,7 @@ private:
std::vector<wasmtime::Val> args;
for(auto value : argv)
args.push_back(wasmtime::Val(value));
auto result = func->call(ctx(), args);
auto result = call_guest(*func, args);
if(!result)
return(trap_text(result.err()));
auto values = result.ok();
@@ -2082,7 +2348,7 @@ private:
// watchdog before the first core call so that wall time cannot make the
// otherwise harmless malloc/reloc sequence trap immediately.
auto allocate_start = std::chrono::steady_clock::now();
ctx().set_epoch_deadline(worker.cfg.epoch_deadline_ticks);
arm_guest_deadline(ctx());
if(mod->abi.version != abi_version)
return(mod->wasm_path + ": uce.abi version " + std::to_string(mod->abi.version)
+ " does not match core ABI " + std::to_string(abi_version));
@@ -2255,19 +2521,19 @@ private:
// init sequence, then bind this unit's context to the request
if(auto relocs = unit_func(unit_index, "__wasm_apply_data_relocs"))
{
auto result = relocs->call(ctx(), {});
auto result = call_guest(*relocs, {});
if(!result)
return(trap_text(result.err()));
}
if(auto ctors = unit_func(unit_index, "__wasm_call_ctors"))
{
auto result = ctors->call(ctx(), {});
auto result = call_guest(*ctors, {});
if(!result)
return(trap_text(result.err()));
}
if(auto set_request = unit_func(unit_index, "__uce_set_current_request"))
{
auto result = set_request->call(ctx(), { wasmtime::Val(request_ptr) });
auto result = call_guest(*set_request, { wasmtime::Val(request_ptr) });
if(!result)
return(trap_text(result.err()));
}
@@ -2295,7 +2561,7 @@ private:
unit_module_operations[module_operation_index].materialize_us = materialize_us;
// Exclude the rest of host-side loading as well. A genuine runaway loop
// makes no loads, so it still trips the deadline.
ctx().set_epoch_deadline(worker.cfg.epoch_deadline_ticks);
arm_guest_deadline(ctx());
return("");
}
@@ -2714,12 +2980,13 @@ private:
return(slot);
}
String run_task_callback(u64 callback_id)
String run_task_callback(u64 callback_id, u64 timeout_cap_ms)
{
InvocationScope invocation(*this, timeout_cap_ms, true);
auto runner = core_func("uce_wasm_task_run");
if(!runner)
return("core does not export uce_wasm_task_run");
auto result = runner->call(ctx(), { wasmtime::Val((int64_t)callback_id) });
auto result = call_guest(*runner, { wasmtime::Val((int64_t)callback_id) });
if(!result)
return(trap_text(result.err()));
return("");
@@ -2749,17 +3016,16 @@ private:
bool profile_memcache = name == "uce_host_memcache_command";
auto profiled = [self, callback, profile_index, profile_enabled, profile_mysql, profile_memcache](Caller caller, Span<const Val> args, Span<Val> results) mutable -> Result<std::monostate, Trap> {
auto started = std::chrono::steady_clock::now();
if(self->invocation_expired(started))
return(Trap(self->invocation_timeout_error()));
f64 cpu_started = profile_enabled && self->worker.cfg.profile_hostcall_cpu ? wasm_thread_cpu_time() : 0;
auto result = callback(caller, args, results);
// Epoch interruption limits guest CPU, not time spent in native I/O,
// process management, hashing, or other host work. Re-arm at the one
// membrane every hostcall crosses so newly added blocking imports
// cannot silently consume the next guest segment's budget.
caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks);
auto finished = std::chrono::steady_clock::now();
self->arm_guest_deadline(caller.context());
f64 cpu_finished = profile_enabled && self->worker.cfg.profile_hostcall_cpu ? wasm_thread_cpu_time() : 0;
if(profile_enabled)
{
u64 elapsed = (u64)std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - started).count();
u64 elapsed = (u64)std::chrono::duration_cast<std::chrono::microseconds>(finished - started).count();
u64 cpu_elapsed = cpu_started > 0 && cpu_finished > cpu_started ?
(u64)((cpu_finished - cpu_started) * 1000000.0) : 0;
self->hostcall_count++;
@@ -2782,6 +3048,8 @@ private:
self->memcache_hostcall_total_us += elapsed;
}
}
if(self->invocation_expired(finished))
return(Trap(self->invocation_timeout_error()));
return(result);
};
auto defined = linker.func_new(mod, name, func_type, profiled);
@@ -3072,7 +3340,16 @@ private:
String stage_key = "shell:" + cmd;
if(!self->hostcall_staged(stage_key, out))
{
out = ::shell_exec(cmd);
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);
if(execution["timed_out"].to_bool())
{
String kind = self->invocation_expired() || remaining_ms <= 5000 ?
"UCE_INVOCATION_TIMEOUT" : "UCE_HOSTCALL_TIMEOUT";
return(Trap(kind + ": shell_exec exceeded " + std::to_string(timeout_ms) + " ms"));
}
out = execution["stdout"].to_string();
if(buf == 0)
self->hostcall_stage(stage_key, out);
}
@@ -3084,7 +3361,7 @@ private:
if(mod == "env" && name == "uce_host_http_request")
return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); String out; String stage_key="http:"+encoded;
if(!self->hostcall_staged(stage_key,out)) { DValue req,response; String err; if(ucb_decode(encoded,req,&err)) response=uce_http_request_value(req); else response["error"]="http_request decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
if(!self->hostcall_staged(stage_key,out)) { DValue req,response; String err; if(ucb_decode(encoded,req,&err)) { u64 requested=req.key("timeout_ms")?req.key("timeout_ms")->to_u64(5000):5000; if(requested==0)requested=5000; req["timeout_ms"]=(f64)std::max<u64>(1,self->bounded_hostcall_timeout_ms(requested)); response=uce_http_request_value(req); } else response["error"]="http_request decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate());
}));
if(mod == "env" && name == "uce_host_http_request_async")
@@ -3094,7 +3371,7 @@ private:
String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded);
u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32();
String out; String stage_key="shell_dv:"+encoded;
if(!self->hostcall_staged(stage_key,out)) { DValue spec, response; String err; if(ucb_decode(encoded,spec,&err)) response=uce_shell_exec_spec(spec); else response["error"]="shell_exec spec decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
if(!self->hostcall_staged(stage_key,out)) { DValue spec, response; String err; if(ucb_decode(encoded,spec,&err)) { u64 requested=spec.key("timeout_ms")?spec.key("timeout_ms")->to_u64(5000):5000; if(requested==0)requested=5000; spec["timeout_ms"]=(f64)std::max<u64>(1,self->bounded_hostcall_timeout_ms(requested)); response=uce_shell_exec_spec(spec); } else response["error"]="shell_exec spec decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
if(buf&&cap>=out.size()) self->hostcall_write(buf,out);
results[0]=Val((int32_t)out.size()); return(std::monostate());
}));
@@ -3103,11 +3380,23 @@ private:
String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded); DValue spec; String err; u64 id=0; if(ucb_decode(encoded,spec,&err)) id=uce_shell_spawn_spec(spec); results[0]=Val((int64_t)id); return(std::monostate());
}));
if(mod == "env" && name == "uce_host_job_status")
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String out=ucb_encode(uce_job_status_value((u64)args[0].i64())); u32 cap=(u32)args[2].i32(); int32_t buf=args[1].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { u64 job_id=(u64)args[0].i64(); String out,stage_key="job_status:"+std::to_string(job_id); u32 cap=(u32)args[2].i32(); int32_t buf=args[1].i32(); if(!self->hostcall_staged(stage_key,out)){out=ucb_encode(uce_job_status_value(job_id));if(buf==0)self->hostcall_stage(stage_key,out);} if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
if(mod == "env" && name == "uce_host_job_result")
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String out=ucb_encode(uce_job_result_value((u64)args[0].i64(), 100)); u32 cap=(u32)args[2].i32(); int32_t buf=args[1].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { u64 job_id=(u64)args[0].i64(); String out,stage_key="job_result:"+std::to_string(job_id); u32 cap=(u32)args[2].i32(); int32_t buf=args[1].i32(); if(!self->hostcall_staged(stage_key,out)){out=ucb_encode(uce_job_result_value(job_id, self->bounded_hostcall_timeout_ms(100)));if(buf==0)self->hostcall_stage(stage_key,out);} if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
if(mod == "env" && name == "uce_host_job_await")
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { u64 timeout=std::min<u64>((u64)args[1].i64(), 30000); String out=ucb_encode(uce_job_result_value((u64)args[0].i64(), timeout)); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
u64 job_id=(u64)args[0].i64(), requested_timeout=(u64)args[1].i64();
u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32();
String out, stage_key="job_await:"+std::to_string(job_id)+":"+std::to_string(requested_timeout);
if(!self->hostcall_staged(stage_key,out))
{
u64 timeout=self->bounded_hostcall_timeout_ms(std::min<u64>(requested_timeout, 30000));
out=ucb_encode(uce_job_result_value(job_id, timeout));
if(buf==0) self->hostcall_stage(stage_key,out);
}
if(buf&&cap>=out.size()) self->hostcall_write(buf,out);
results[0]=Val((int32_t)out.size()); return(std::monostate());
}));
if(mod == "env" && name == "uce_host_job_cancel")
return(add([](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { results[0]=Val((int32_t)(uce_job_cancel_value((u64)args[0].i64())?1:0)); return(std::monostate()); }));
if(mod == "env" && name == "uce_host_path_real")
@@ -3671,8 +3960,9 @@ private:
}
else
{
::socket_write((u64)args[0].i64(), command + "\r\n");
out = ::socket_read((u64)args[0].i64());
u64 socket_fd = (u64)args[0].i64();
wasm_socket_write_bounded(socket_fd, command + "\r\n", self->bounded_hostcall_timeout_ms(1000));
out = wasm_socket_read_bounded(socket_fd, 1024 * 128, self->bounded_hostcall_timeout_ms(1000));
if(buf == 0)
{
self->staged_memcache_key = key;
@@ -3810,28 +4100,8 @@ private:
return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
String host;
self->hostcall_read(args[0].i32(), args[1].i32(), host);
int fd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if(fd >= 0)
{
struct sockaddr_in addr = {0};
addr.sin_family = AF_INET;
addr.sin_port = htons((short)args[2].i32());
addr.sin_addr.s_addr = inet_addr(host.c_str());
if(::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
{
::close(fd);
fd = -1;
}
else if(fd == 0)
{
int moved = ::dup(fd);
::close(fd);
fd = moved;
}
if(fd > 0 && context)
context->resources.sockets.push_back(fd);
}
results[0] = Val((int64_t)(fd > 0 ? fd : 0));
u64 fd = wasm_socket_connect_bounded(host, (u16)args[2].i32(), self->bounded_hostcall_timeout_ms(self->worker.cfg.invocation_timeout_ms));
results[0] = Val((int64_t)fd);
return(std::monostate());
}));
if(mod == "env" && name == "uce_host_socket_close")
@@ -3843,17 +4113,20 @@ private:
return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
String data;
self->hostcall_read(args[1].i32(), args[2].i32(), data);
results[0] = Val(::socket_write((u64)args[0].i64(), data) ? (int32_t)1 : (int32_t)0);
results[0] = Val(wasm_socket_write_bounded((u64)args[0].i64(), data, self->bounded_hostcall_timeout_ms(self->worker.cfg.invocation_timeout_ms)) ? (int32_t)1 : (int32_t)0);
return(std::monostate());
}));
if(mod == "env" && name == "uce_host_socket_read")
return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
u64 sockfd = (u64)args[0].i64();
u32 max_length = (u32)args[1].i32();
u32 timeout = (u32)args[2].i32();
u32 requested_timeout = (u32)args[2].i32();
u64 requested_ms = requested_timeout == 0 ? self->invocation_remaining_ms() : (u64)requested_timeout * 1000;
u64 bounded_ms = self->bounded_hostcall_timeout_ms(requested_ms);
int32_t buf = args[3].i32();
u32 cap = (u32)args[4].i32();
String key = std::to_string(sockfd) + ":" + std::to_string(max_length) + ":" + std::to_string(timeout);
// The size and fetch calls share this key, while the remaining budget may change between them.
String key = std::to_string(sockfd) + ":" + std::to_string(max_length) + ":" + std::to_string(requested_timeout);
String out;
if(buf != 0 && self->staged_socket_read_key == key)
{
@@ -3863,7 +4136,7 @@ private:
}
else
{
out = ::socket_read(sockfd, max_length, timeout);
out = wasm_socket_read_bounded(sockfd, max_length, bounded_ms);
if(buf == 0)
{
self->staged_socket_read_key = key;
@@ -3920,8 +4193,9 @@ private:
// before the hostcall stack unwinds, so `self` points to the child's
// copy of this per-request workspace. The parent request can return and
// destroy its workspace without invalidating the child copy.
auto run_callback = [self, callback_id]() {
String error = self->run_task_callback(callback_id);
u64 task_timeout_ms = timeout > UINT64_MAX / 1000 ? UINT64_MAX : timeout * 1000;
auto run_callback = [self, callback_id, task_timeout_ms]() {
String error = self->run_task_callback(callback_id, task_timeout_ms);
if(error != "")
fprintf(stderr, "[wasm task] callback failed: %s\n", error.c_str());
};
@@ -3959,19 +4233,17 @@ private:
if(mod == "env" && name == "uce_host_sleep_us")
return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
u64 usec = (u64)args[0].i64();
while(usec >= 1000000ull)
{
unsigned int remaining = ::sleep((unsigned int)(usec / 1000000ull));
if(remaining != 0)
{
results[0] = Val((int32_t)remaining);
return(std::monostate());
}
usec %= 1000000ull;
}
if(usec > 0)
::usleep((useconds_t)usec);
results[0] = Val((int32_t)0);
u64 requested_ms = usec / 1000 + (usec % 1000 != 0);
u64 bounded_ms = self->bounded_hostcall_timeout_ms(requested_ms);
u64 bounded_usec = bounded_ms > UINT64_MAX / 1000 ? UINT64_MAX : bounded_ms * 1000;
u64 sleep_usec = std::min(usec, bounded_usec);
struct timespec requested = { (time_t)(sleep_usec / 1000000), (long)((sleep_usec % 1000000) * 1000) };
struct timespec interrupted = { 0, 0 };
u64 unslept_usec = usec - sleep_usec;
if(sleep_usec > 0 && nanosleep(&requested, &interrupted) != 0 && errno == EINTR)
unslept_usec += (u64)interrupted.tv_sec * 1000000 + (u64)interrupted.tv_nsec / 1000;
u64 unslept_seconds = unslept_usec / 1000000 + (unslept_usec % 1000000 != 0);
results[0] = Val((int32_t)std::min<u64>(UINT32_MAX, unslept_seconds));
return(std::monostate());
}));
if(mod == "env" && name == "uce_host_regex")