From a7576f3db7e728d16a08f16a3952b2482d4f989c Mon Sep 17 00:00:00 2001 From: udo Date: Sun, 19 Jul 2026 00:44:18 +0000 Subject: [PATCH] Enforce absolute Wasm invocation deadlines --- docs/setup.md | 11 + docs/wasm-runtime-architecture.md | 18 +- etc/uce/settings.cfg | 1 + scripts/run_cli_tests.sh | 3 +- scripts/test_wasm_invocation_timeout.sh | 126 +++++++ site/tests/cli_runner.uce | 27 +- src/lib/sys.cpp | 43 +++ src/lib/sys.h | 2 + src/wasm/backend.cpp | 24 +- src/wasm/worker.cpp | 428 +++++++++++++++++++----- 10 files changed, 591 insertions(+), 92 deletions(-) create mode 100755 scripts/test_wasm_invocation_timeout.sh diff --git a/docs/setup.md b/docs/setup.md index 55c9ee2..c0320ba 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -199,6 +199,7 @@ WASM_CORE_PATH=/bin/wasm/core.wasm WASM_MEMORY_LIMIT_BYTES=536870912 WASM_EPOCH_DEADLINE_TICKS=200 WASM_EPOCH_PERIOD_MS=50 +WASM_INVOCATION_TIMEOUT_MS=30000 MYSQL_PERSISTENT_POOL_SIZE=8 WORKER_COUNT=4 @@ -227,6 +228,16 @@ Important settings: - `WASM_COMPILE_SCRIPT` must point to `scripts/compile_wasm_unit` unless you provide an equivalent compiler. Relative paths are resolved from the runtime root/`COMPILER_SYS_PATH`. That script calls `scripts/check_unit_wasm.py` after linking each unit and uses the pinned WASI SDK on every deployment host. - `PROACTIVE_COMPILE_JOBS` selects 1–16 low-priority full-site scanner processes (default `2`). Each canonical unit path has one scanner owner. The separate higher-priority demand compiler remains reserved for stale units requested over HTTP, so total background compile concurrency can reach this value plus one. - `WASM_CORE_PATH` must point at the built `core.wasm` file. +- `WASM_EPOCH_DEADLINE_TICKS` and `WASM_EPOCH_PERIOD_MS` bound one + uninterrupted guest CPU segment. `WASM_INVOCATION_TIMEOUT_MS` is the + separate absolute wall-clock bound for app-owned unit loading, + initialization, the selected handler, and all nested component/unit calls. + The three values must be positive integers; the ticker period is capped at + `1000` ms and the invocation timeout at `86400000` ms. Invalid values prevent + the Wasm backend from starting. The invocation timeout defaults to `30000` and is + enforced to the epoch ticker's period resolution. Blocking host helpers + retain their own shorter limits and are capped to the remaining invocation + budget where the underlying operation is cancellable. After editing settings, restart UCE: diff --git a/docs/wasm-runtime-architecture.md b/docs/wasm-runtime-architecture.md index a1d269c..5425711 100644 --- a/docs/wasm-runtime-architecture.md +++ b/docs/wasm-runtime-architecture.md @@ -200,11 +200,16 @@ The graceful signal handler belongs to the parent and render workers. Generic `task()` children restore default termination signals after fork so `task_kill()` and `server_stop()` retain their immediate stop contract. -Epoch interruption measures uninterrupted guest CPU segments. The common -hostcall membrane re-arms the store deadline after every native call, excluding -blocking I/O, process waits, hashing, and other host work without weakening a -guest loop that makes no hostcalls. Keeping this at the membrane also covers new -hostcalls without per-import timeout bookkeeping. +Epoch interruption measures uninterrupted guest CPU segments. A separate +absolute workspace invocation deadline starts before app-owned entry-unit +loading and initialization and remains unchanged through the selected handler, +ONCE, and every nested component/unit call. The common hostcall membrane checks +that deadline before and after every native call and re-arms the store with the +smaller of the remaining absolute budget and the CPU-segment budget. A cheap +hostcall loop therefore cannot renew an invocation indefinitely. Blocking host +helpers retain operation-specific limits and cap them to the remaining +invocation budget where the underlying operation is cancellable. Forked task +callbacks receive a fresh invocation deadline capped by the task lifetime. `request_perf()` reports worker module-cache hits and misses and divides a miss into artifact lookup, wasm read, custom-section parse, serialized-module @@ -506,6 +511,9 @@ header free-functions are `inline`. The wasm backend exposes only declarations | Key | Default | Meaning | |---|---|---| | `WASM_BACKEND_VERBOSE` | `0` | Emit `X-UCE-Wasm-*` workspace timing headers (benchmark only). | +| `WASM_EPOCH_DEADLINE_TICKS` | `200` | Maximum uninterrupted guest CPU segment in epoch ticks; must be positive. | +| `WASM_EPOCH_PERIOD_MS` | `50` | Worker epoch-ticker period and timeout resolution; range `1`–`1000` ms. | +| `WASM_INVOCATION_TIMEOUT_MS` | `30000` | Absolute app-owned unit load/init/handler/nested-call deadline; range `1`–`86400000` ms and nested calls cannot renew it. | | `FCGI_SOCKET_PATH` | runtime-configured (`/run/uce/fastcgi.sock` in this doc) | Worker pool FastCGI socket (brokers forward here). | | `CLI_SOCKET_PATH` | `/run/uce/cli.sock` | Worker CLI/admin socket. Keep private; reference `CLI_SOCKET_MODE` is `0600`. | | `FCGI_SOCKET_MODE` | `0666` | Permission mode applied to `FCGI_SOCKET_PATH` after bind; set tighter if nginx/Apache can use a trusted group. | diff --git a/etc/uce/settings.cfg b/etc/uce/settings.cfg index 2f832af..60ab1e0 100644 --- a/etc/uce/settings.cfg +++ b/etc/uce/settings.cfg @@ -48,6 +48,7 @@ WASM_CORE_PATH=bin/wasm/core.wasm WASM_MEMORY_LIMIT_BYTES=536870912 WASM_EPOCH_DEADLINE_TICKS=200 WASM_EPOCH_PERIOD_MS=50 +WASM_INVOCATION_TIMEOUT_MS=30000 MYSQL_PERSISTENT_POOL_SIZE=8 # ENABLE THE BACKGROUND PROACTIVE COMPILER LOOP diff --git a/scripts/run_cli_tests.sh b/scripts/run_cli_tests.sh index 286afad..2308649 100755 --- a/scripts/run_cli_tests.sh +++ b/scripts/run_cli_tests.sh @@ -58,7 +58,7 @@ if [[ "$action" == "list" ]]; then fi if [[ "$action" == "run" ]]; then - groups=(demo http site doc-gate security task-lifetime pool-isolation starter tcp) + groups=(demo http site doc-gate-{1..30} security task-lifetime pool-isolation starter tcp) if [[ "$include_kill" == "1" ]]; then groups+=(wasm-kill) fi @@ -71,6 +71,7 @@ if [[ "$action" == "run" ]]; then scripts/test_parallel_precompile.sh timeout --signal=TERM --kill-after=5s 175s scripts/test_parallel_proactive_compile.sh scripts/test_cold_component_deadline.sh + scripts/test_wasm_invocation_timeout.sh scripts/test_nested_component_props.sh scripts/test_component_once_prefetch.sh scripts/test_relative_component_cache.sh diff --git a/scripts/test_wasm_invocation_timeout.sh b/scripts/test_wasm_invocation_timeout.sh new file mode 100755 index 0000000..06c91dd --- /dev/null +++ b/scripts/test_wasm_invocation_timeout.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "$0")/.." + +raw_calls=$(grep -En -- '(->|\.)call\(' src/wasm/worker.cpp) +if [[ $(wc -l <<<"$raw_calls") -ne 1 || "$raw_calls" != *'return(func.call(context, args));'* ]]; then + echo "raw Wasmtime call bypasses call_guest(): $raw_calls" >&2 + exit 1 +fi +if [[ "${1:-}" == "--static-only" ]]; then + echo "wasm invocation timeout static call gate passed" + exit 0 +fi + +settings_file="${UCE_SETTINGS_FILE:-/etc/uce/settings.cfg}" +site_directory="${UCE_TEST_SITE_DIRECTORY:-site}" +socket_path="${UCE_CLI_SOCKET:-/run/uce/cli.sock}" +bin_directory="${BIN_DIRECTORY:-/tmp/uce/work}" +if [[ -r "$settings_file" ]]; then + [[ -n "${UCE_TEST_SITE_DIRECTORY:-}" ]] || site_directory=$(awk -F= '/^[[:space:]]*SITE_DIRECTORY[[:space:]]*=/ {gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$settings_file") + [[ -n "${UCE_CLI_SOCKET:-}" ]] || socket_path=$(awk -F= '/^[[:space:]]*CLI_SOCKET_PATH[[:space:]]*=/ {gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$settings_file") + [[ -n "${BIN_DIRECTORY:-}" ]] || bin_directory=$(awk -F= '/^[[:space:]]*BIN_DIRECTORY[[:space:]]*=/ {gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$settings_file") + invocation_ms=$(awk -F= '/^[[:space:]]*WASM_INVOCATION_TIMEOUT_MS[[:space:]]*=/ {gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$settings_file") +fi +site_directory="${site_directory:-site}" +socket_path="${socket_path:-/run/uce/cli.sock}" +bin_directory="${bin_directory:-/tmp/uce/work}" +invocation_ms="${invocation_ms:-30000}" +test_name="invocation-timeout-test-$$" +source_dir="$site_directory/$test_name" +pid_file="/tmp/uce-$test_name-worker" +cache_dir="" +body_file="/tmp/uce-$test_name-body" + +cleanup() { + rm -rf "$source_dir" + [[ -z "$cache_dir" ]] || rm -rf "$cache_dir" + rm -f "$pid_file" "$body_file" +} +trap cleanup EXIT +mkdir -p "$source_dir" +cache_dir="$(scripts/unit_cache_directory "$bin_directory")$(realpath "$source_dir")" + +printf '%s\n' \ + 'CLI(Request& context) {' \ + ' if(context.get["warm"] == "1") { print("warm"); return; }' \ + " file_put_contents(\"$pid_file\", std::to_string(request_perf()[\"worker_pid\"].to_u64()));" \ + ' while(true) time_precise();' \ + '}' >"$source_dir/hostcall-loop.uce" +printf '%s\n' \ + 'CLI(Request& context) {' \ + ' if(context.get["warm"] == "1") { print("warm"); return; }' \ + ' if(context.get["quick"] == "1") { print(shell_exec("printf quick")); return; }' \ + ' if(context.get["status"] == "1") { DValue spec; spec["cmd"] = "exit 7"; spec["timeout_ms"] = (f64)500; print(shell_exec(spec)["exit_code"].to_u64()); return; }' \ + ' if(context.get["zero"] == "1") { DValue spec; spec["cmd"] = "printf zero"; spec["timeout_ms"] = (f64)0; print(shell_exec(spec)["stdout"].to_string()); return; }' \ + ' if(context.get["job"] == "1") { DValue spec; spec["cmd"] = "sleep 2"; spec["timeout_ms"] = (f64)5000; u64 job = shell_spawn(spec); f64 started = time_precise(); job_await(job, 300); u64 elapsed = (u64)((time_precise() - started) * 1000); job_cancel(job); print(elapsed); return; }' \ + " file_put_contents(\"$pid_file\", std::to_string(request_perf()[\"worker_pid\"].to_u64()));" \ + ' print(shell_exec("printf shell-start; sleep 60 & printf shell-end"));' \ + '}' >"$source_dir/legacy-shell.uce" +printf '%s\n' \ + 'CLI(Request& context) {' \ + ' if(context.get["warm"] == "1") { print("warm"); return; }' \ + " file_put_contents(\"$pid_file\", std::to_string(request_perf()[\"worker_pid\"].to_u64()));" \ + ' while(true) sleep(60);' \ + ' print("sleep-end");' \ + '}' >"$source_dir/sleep.uce" +printf '%s\n' \ + 'CLI(Request& context) { print(request_perf()["worker_pid"].to_u64(), "|health"); }' >"$source_dir/health.uce" + +max_seconds=$(( (invocation_ms + 15000) / 1000 )) +(( max_seconds >= 10 )) || max_seconds=10 + +request() { + local unit="$1" + curl -sS --max-time "$max_seconds" -o "$body_file" -w '%{http_code}' --unix-socket "$socket_path" "http://localhost/$test_name/$unit" +} + +same_worker_health() { + local expected_pid="$1" + for _ in $(seq 1 32); do + local health + health=$(curl -sS --max-time 5 --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/health.uce") + [[ "$health" == "$expected_pid|health" ]] && return 0 + done + return 1 +} + +curl -sS --max-time "$max_seconds" --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/hostcall-loop.uce?warm=1" >/dev/null +curl -sS --max-time "$max_seconds" --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/legacy-shell.uce?warm=1" >/dev/null +curl -sS --max-time "$max_seconds" --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/sleep.uce?warm=1" >/dev/null +curl -sS --max-time "$max_seconds" --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/health.uce" >/dev/null +quick=$(curl -sS --max-time 5 --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/legacy-shell.uce?quick=1") +[[ "$quick" == quick ]] || { echo "quick legacy shell returned: $quick" >&2; exit 1; } +exit_status=$(curl -sS --max-time 5 --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/legacy-shell.uce?status=1") +[[ "$exit_status" == 7 ]] || { echo "structured shell lost exit status: $exit_status" >&2; exit 1; } +zero_timeout=$(curl -sS --max-time 5 --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/legacy-shell.uce?zero=1") +[[ "$zero_timeout" == zero ]] || { echo "structured shell zero-timeout default changed: $zero_timeout" >&2; exit 1; } +job_elapsed=$(curl -sS --max-time 5 --fail-with-body --unix-socket "$socket_path" "http://localhost/$test_name/legacy-shell.uce?job=1") +[[ "$job_elapsed" =~ ^[0-9]+$ && "$job_elapsed" -ge 200 && "$job_elapsed" -lt 500 ]] || { echo "job_await two-call duration was ${job_elapsed}ms" >&2; exit 1; } + +for unit in hostcall-loop.uce legacy-shell.uce sleep.uce; do + rm -f "$pid_file" + started_ns=$(date +%s%N) + status=$(request "$unit") + elapsed_ms=$(( ($(date +%s%N) - started_ns) / 1000000 )) + [[ "$status" == "500" ]] || { echo "$unit returned HTTP $status" >&2; exit 1; } + if [[ "$unit" == legacy-shell.uce ]]; then + if (( invocation_ms <= 5000 )); then + grep -q 'UCE_INVOCATION_TIMEOUT:' "$body_file" || { echo "$unit lacked invocation-timeout classification" >&2; exit 1; } + expected_ms=$invocation_ms + else + grep -q 'UCE_HOSTCALL_TIMEOUT:' "$body_file" || { echo "$unit lacked hostcall-timeout classification" >&2; exit 1; } + expected_ms=5000 + fi + else + grep -q 'UCE_INVOCATION_TIMEOUT:' "$body_file" || { echo "$unit lacked invocation-timeout classification" >&2; exit 1; } + expected_ms=$invocation_ms + fi + (( elapsed_ms >= expected_ms - 1000 && elapsed_ms <= expected_ms + 3000 )) || { echo "$unit completed in ${elapsed_ms}ms; expected about ${expected_ms}ms" >&2; exit 1; } + grep -q "$unit" "$body_file" || { echo "$unit lacked a source-mapped trace" >&2; exit 1; } + [[ -s "$pid_file" ]] || { echo "$unit did not record its worker" >&2; exit 1; } + worker_pid=$(<"$pid_file") + same_worker_health "$worker_pid" || { echo "worker $worker_pid did not survive $unit" >&2; exit 1; } +done + +echo "wasm invocation timeout passed (absolute ${invocation_ms}ms; sleep and process hostcalls bounded)" diff --git a/site/tests/cli_runner.uce b/site/tests/cli_runner.uce index 23b74eb..3e473dd 100644 --- a/site/tests/cli_runner.uce +++ b/site/tests/cli_runner.uce @@ -197,7 +197,7 @@ bool cli_doc_example_is_placeholder(String source) return(false); } -void cli_run_doc_pages_gate() +void cli_run_doc_pages_gate(u64 shard = 0, u64 shard_count = 1) { StringList error_markers; error_markers.push_back("doc example error"); @@ -207,12 +207,15 @@ void cli_run_doc_pages_gate() error_markers.push_back(String("uncaught exception during ") + "request"); error_markers.push_back("timed out acquiring compile lock"); - for(String file_name : ls("../doc/pages/")) + u64 page_index = 0; + for(String file_name : ls("../doc/pages/").sort()) { String source = file_get_contents("../doc/pages/" + file_name); String page = nibble(file_name, "."); if(page == "") continue; + if(page_index++ % shard_count != shard) + continue; bool has_example = cli_doc_source_has_example(source); String path = "/doc/index.uce?p=" + uri_encode(page); CliHttpResponse res; @@ -465,7 +468,7 @@ void cli_run_wasm_kill(bool include_kill) void cli_print_list(bool include_kill) { print("UCE CLI test groups:\n"); - print(" demo, http, site, doc-gate, security, task-lifetime, pool-isolation, starter, tcp"); + print(" demo, http, site, doc-gate-1..30, security, task-lifetime, pool-isolation, starter, tcp"); if(include_kill) print(", wasm-kill"); print("\n"); @@ -480,8 +483,16 @@ bool cli_run_group(String group, bool skip_local_service_pages, bool include_kil cli_run_http_smoke(); else if(group == "site") cli_run_site_suite(skip_local_service_pages); - else if(group == "doc-gate") - cli_run_doc_pages_gate(); + else if(str_starts_with(group, "doc-gate-")) + { + u64 shard = int_val(group.substr(9)); + if(shard < 1 || shard > 30) + { + cli_test_case("uce_cli_runner:known doc gate shard", false, "unknown group: " + group); + return(false); + } + cli_run_doc_pages_gate(shard - 1, 30); + } else if(group == "security") cli_run_security_smoke(); else if(group == "task-lifetime") @@ -526,7 +537,11 @@ CLI(Request& context) cli_run_group(group, input["skip_local_service_pages"].to_bool(), include_kill); else { - for(String item : { "demo", "http", "site", "doc-gate", "security", "task-lifetime", "pool-isolation", "starter", "tcp" }) + for(String item : { "demo", "http", "site" }) + cli_run_group(item, input["skip_local_service_pages"].to_bool(), include_kill); + for(u64 shard = 1; shard <= 30; shard++) + cli_run_group("doc-gate-" + std::to_string(shard), input["skip_local_service_pages"].to_bool(), include_kill); + for(String item : { "security", "task-lifetime", "pool-isolation", "starter", "tcp" }) cli_run_group(item, input["skip_local_service_pages"].to_bool(), include_kill); cli_run_group("wasm-kill", input["skip_local_service_pages"].to_bool(), include_kill); } diff --git a/src/lib/sys.cpp b/src/lib/sys.cpp index a6b31f9..456484d 100644 --- a/src/lib/sys.cpp +++ b/src/lib/sys.cpp @@ -1396,6 +1396,48 @@ std::map workers; #include #include +// The process-wide reaper may run on a runtime thread while a request thread +// synchronously owns a child. Preserve unknown statuses until that owner takes +// them instead of forcing it to guess success after waitpid reports ECHILD. +static const size_t child_exit_status_capacity = 256; +static volatile sig_atomic_t child_exit_status_cursor = 0; +static volatile sig_atomic_t child_exit_status_pids[child_exit_status_capacity] = {0}; +static volatile sig_atomic_t child_exit_status_values[child_exit_status_capacity] = {0}; +static volatile sig_atomic_t child_exit_status_sequences[child_exit_status_capacity] = {0}; + +static void child_exit_status_publish(pid_t pid, int status) +{ + sig_atomic_t cursor = __atomic_fetch_add(&child_exit_status_cursor, 1, __ATOMIC_RELAXED); + size_t slot = (size_t)(unsigned int)cursor % child_exit_status_capacity; + __atomic_store_n(&child_exit_status_pids[slot], 0, __ATOMIC_RELEASE); + __atomic_store_n(&child_exit_status_values[slot], (sig_atomic_t)status, __ATOMIC_RELAXED); + __atomic_store_n(&child_exit_status_sequences[slot], cursor + 1, __ATOMIC_RELAXED); + __atomic_store_n(&child_exit_status_pids[slot], (sig_atomic_t)pid, __ATOMIC_RELEASE); +} + +unsigned int child_exit_status_snapshot() +{ + return((unsigned int)__atomic_load_n(&child_exit_status_cursor, __ATOMIC_ACQUIRE)); +} + +bool child_exit_status_take(pid_t pid, int& status, unsigned int since) +{ + for(size_t slot = 0; slot < child_exit_status_capacity; slot++) + { + unsigned int sequence = (unsigned int)__atomic_load_n(&child_exit_status_sequences[slot], __ATOMIC_ACQUIRE); + unsigned int distance = sequence - since; + if(sequence == 0 || distance == 0 || distance >= 0x80000000u) + continue; + sig_atomic_t expected = (sig_atomic_t)pid; + if(__atomic_compare_exchange_n(&child_exit_status_pids[slot], &expected, 0, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) + { + status = (int)__atomic_load_n(&child_exit_status_values[slot], __ATOMIC_RELAXED); + return(true); + } + } + return(false); +} + pid_t spawn_subprocess(std::function exec_after_spawn) { parent_pid = getpid(); @@ -1648,6 +1690,7 @@ void on_child_exit(int sig) } else { + child_exit_status_publish(pid, status); printf("(P) task child reaped (PID:%i)\n", pid); } } diff --git a/src/lib/sys.h b/src/lib/sys.h index 702d7a2..090bcb5 100644 --- a/src/lib/sys.h +++ b/src/lib/sys.h @@ -21,6 +21,8 @@ s32 usleep(u32 usec); #else #include #include +unsigned int child_exit_status_snapshot(); +bool child_exit_status_take(pid_t pid, int& status, unsigned int since); #endif #include #include diff --git a/src/wasm/backend.cpp b/src/wasm/backend.cpp index a181652..eff0287 100644 --- a/src/wasm/backend.cpp +++ b/src/wasm/backend.cpp @@ -50,7 +50,24 @@ static String wasm_backend_ensure_started(Request* context) if(cfg[key] != "") wc.write_roots.push_back(cfg[key]); wc.memory_limit = (int64_t)to_u64(cfg["WASM_MEMORY_LIMIT_BYTES"], 512ull * 1024 * 1024); - wc.epoch_deadline_ticks = to_u64(cfg["WASM_EPOCH_DEADLINE_TICKS"], 200); + wc.epoch_deadline_ticks = to_u64(first(cfg["WASM_EPOCH_DEADLINE_TICKS"], "200"), 0); + wc.epoch_period_ms = to_u64(first(cfg["WASM_EPOCH_PERIOD_MS"], "50"), 0); + wc.invocation_timeout_ms = to_u64(first(cfg["WASM_INVOCATION_TIMEOUT_MS"], "30000"), 0); + if(wc.epoch_deadline_ticks == 0 || wc.epoch_deadline_ticks > (u64)INT64_MAX) + { + g_wasm_init_error = "WASM_EPOCH_DEADLINE_TICKS must be a positive integer no greater than INT64_MAX"; + return(g_wasm_init_error); + } + if(wc.epoch_period_ms == 0 || wc.epoch_period_ms > 1000) + { + g_wasm_init_error = "WASM_EPOCH_PERIOD_MS must be a positive integer no greater than 1000"; + return(g_wasm_init_error); + } + if(wc.invocation_timeout_ms == 0 || wc.invocation_timeout_ms > 86400000) + { + g_wasm_init_error = "WASM_INVOCATION_TIMEOUT_MS must be a positive integer no greater than 86400000"; + return(g_wasm_init_error); + } wc.mysql_persistent_pool_size = std::min(to_u64(cfg["MYSQL_PERSISTENT_POOL_SIZE"], 8), 64); wc.profile_hostcall_cpu = to_bool(cfg["WASM_PROFILE_HOSTCALL_CPU"], false); wc.profile_thread_runtime = to_bool(cfg["WASM_PROFILE_THREAD_RUNTIME"], false); @@ -84,7 +101,9 @@ static String wasm_backend_ensure_started(Request* context) g_wasm_epoch_running.store(true); WasmWorker* worker = g_wasm_worker; - u64 period_ms = to_u64(cfg["WASM_EPOCH_PERIOD_MS"], 50); + u64 period_ms = wc.epoch_period_ms; + // The ticker inherits this block so it cannot run the process-wide child reaper. + WasmSigchldBlock sigchld; g_wasm_epoch_ticker = new std::thread([worker, period_ms] { while(g_wasm_epoch_running.load()) { @@ -92,6 +111,7 @@ static String wasm_backend_ensure_started(Request* context) worker->engine.increment_epoch(); } }); + sigchld.restore(); return(""); } diff --git a/src/wasm/worker.cpp b/src/wasm/worker.cpp index 56d059d..a6e0f58 100644 --- a/src/wasm/worker.cpp +++ b/src/wasm/worker.cpp @@ -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(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 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 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 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_off0) 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()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(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> call_guest(wasmtime::Func& func, std::vector args) + { + auto context = ctx(); + arm_guest_deadline(context); + return(func.call(context, args)); + } + String call_core(const String& name, std::vector argv, int32_t* result_out) { auto func = core_func(name); @@ -1960,7 +2226,7 @@ private: std::vector 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 args, Span results) mutable -> Result { 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::steady_clock::now() - started).count(); + u64 elapsed = (u64)std::chrono::duration_cast(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(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 args, Span results) -> Result { 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(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(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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { u64 timeout=std::min((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 args, Span results) -> Result { + 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(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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { 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 args, Span results) -> Result { 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(UINT32_MAX, unslept_seconds)); return(std::monostate()); })); if(mod == "env" && name == "uce_host_regex")