diff --git a/README.md b/README.md index ed34b3233..300299317 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,7 @@ Beam.demonitor(ref); ## Supervision -Runtimes are OTP children with crash recovery: +Runtimes and context pools are OTP children with crash recovery: ```elixir children = [ @@ -82,6 +82,9 @@ children = [ "db.query" => fn [sql, params] -> Repo.query!(sql, params).rows end, }}, {QuickBEAM, name: :worker, id: :worker}, + + # Context pool for high-concurrency use cases + {QuickBEAM.ContextPool, name: MyApp.JSPool, size: 4}, ] Supervisor.start_link(children, strategy: :one_for_one) @@ -92,6 +95,77 @@ Supervisor.start_link(children, strategy: :one_for_one) The `:script` option loads a JS file at startup. If the runtime crashes, the supervisor restarts it with a fresh context and re-evaluates the script. +Individual `Context` processes are typically started dynamically (e.g. +from a LiveView `mount`) and linked to the connection process. + +## Context Pool + +For high-concurrency scenarios (thousands of connections), use +`ContextPool` instead of individual runtimes. Many lightweight JS +contexts share a small number of runtime threads: + +```elixir +# Start a pool with N runtime threads (defaults to scheduler count) +{:ok, pool} = QuickBEAM.ContextPool.start_link(name: MyApp.JSPool, size: 4) + +# Each context is a GenServer with its own JS global scope +{:ok, ctx} = QuickBEAM.Context.start_link(pool: MyApp.JSPool) +{:ok, 3} = QuickBEAM.Context.eval(ctx, "1 + 2") +{:ok, "HELLO"} = QuickBEAM.Context.eval(ctx, "'hello'.toUpperCase()") +QuickBEAM.Context.stop(ctx) +``` + +Contexts support the full API — `eval`, `call`, `Beam.call`/`callSync`, +DOM, messaging, browser/node APIs, handlers, and supervision: + +```elixir +# In a Phoenix LiveView +def mount(_params, _session, socket) do + {:ok, ctx} = QuickBEAM.Context.start_link( + pool: MyApp.JSPool, + handlers: %{"db.query" => &MyApp.query/1} + ) + {:ok, assign(socket, js: ctx)} +end + +``` + +The context is linked to the LiveView process — it terminates and +cleans up automatically when the connection closes. No explicit +`terminate` callback needed. + +### Granular API groups + +Contexts can load individual API groups instead of the full browser bundle: + +```elixir +QuickBEAM.Context.start_link(pool: pool, apis: [:beam, :fetch]) # 231 KB +QuickBEAM.Context.start_link(pool: pool, apis: [:beam, :url]) # 108 KB +QuickBEAM.Context.start_link(pool: pool, apis: false) # 58 KB +QuickBEAM.Context.start_link(pool: pool) # 429 KB (all browser APIs) +``` + +Available groups: `:fetch`, `:websocket`, `:worker`, `:channel`, +`:eventsource`, `:url`, `:crypto`, `:compression`, `:buffer`, `:dom`, +`:console`, `:storage`, `:locks`. Dependencies auto-resolve. + +### Per-context resource limits + +```elixir +{:ok, ctx} = QuickBEAM.Context.start_link( + pool: pool, + memory_limit: 512_000, # per-context allocation limit (bytes) + max_reductions: 100_000 # opcode budget per eval/call +) + +# Track per-context memory +{:ok, %{context_malloc_size: 92_000}} = QuickBEAM.Context.memory_usage(ctx) +``` + +Exceeding `memory_limit` triggers OOM. Exceeding `max_reductions` +interrupts the current eval but keeps the context usable for +subsequent calls. + ## API surfaces QuickBEAM can load browser APIs, Node.js APIs, or both: @@ -327,8 +401,26 @@ vs QuickJSEx 0.3.1 (Rust/Rustler, JSON serialization): | `Beam.callSync` (JS→BEAM) | 5 μs overhead (unique to QuickBEAM) | | Startup | ~600 μs (parity) | +Context pool vs individual runtimes at scale: + +| | Runtime (1:1 thread) | Context (pooled) | +|---|---|---| +| JS heap per instance | ~530 KB | ~429 KB (full) / ~58 KB (bare) | +| OS thread stack | ~2.5 MB each | shared (4 threads total) | +| OS threads at 10K | 10,000 | 4 (configurable) | +| Total RAM at 10K | ~30 GB | ~4.2 GB (full) / ~570 MB (bare) | + See [`bench/`](https://github.com/elixir-volt/quickbeam/tree/master/bench) for details. +## When to use what + +| Use case | Module | Why | +|---|---|---| +| One-off eval, scripting | `QuickBEAM` (Runtime) | Simple, full isolation | +| SSR request pool | `QuickBEAM.Pool` | Checkout/checkin with reset | +| Per-connection state (LiveView) | `QuickBEAM.Context` | Lightweight, thousands concurrent | +| Sandboxed user code | `QuickBEAM` or `Context` with `apis: false` | Memory limits, reduction limits, timeouts | + ## Examples - [`examples/ssr/`](https://github.com/elixir-volt/quickbeam/tree/master/examples/ssr) — Preact SSR with a pool of runtimes diff --git a/docs/architecture.md b/docs/architecture.md index dbc1354a2..4335a4782 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -199,6 +199,129 @@ Each checkout gets a runtime, each checkin resets it and re-runs the init function. This gives a clean JS context per request while amortizing startup cost. +## Context Pool + +`QuickBEAM.ContextPool` is a different approach to concurrency — +lightweight JS contexts that share runtime threads, rather than +whole runtimes in a checkout pool. + +### The problem + +A full `QuickBEAM.Runtime` dedicates an OS thread and `JSRuntime` per +GenServer (~2MB+ each). At 10K concurrent connections (e.g. Phoenix +LiveView), that's 10K threads and ~25GB of memory. + +### The solution + +QuickJS natively supports multiple `JSContext` instances per +`JSRuntime`. Each context has its own global object, prototypes, and +execution state, but shares the runtime's GC heap and parser. A +`ContextPool` exploits this: + +``` +┌─────────────────────────────────────────────────────┐ +│ ContextPool (GenServer) │ +│ Round-robin assignment: context → thread │ +├──────────┬──────────┬──────────┬───────────────────┐│ +│ Thread 0 │ Thread 1 │ Thread 2 │ Thread N-1 ││ +│ JSRuntime│ JSRuntime│ JSRuntime│ JSRuntime ││ +│ ┌──────┐ │ ┌──────┐ │ ┌──────┐ │ ┌──────┐ ││ +│ │Ctx 1 │ │ │Ctx 2 │ │ │Ctx 3 │ │ │Ctx N │ ││ +│ │Ctx 5 │ │ │Ctx 6 │ │ │Ctx 7 │ │ │Ctx ..│ ││ +│ │Ctx 9 │ │ │... │ │ │... │ │ │ │ ││ +│ └──────┘ │ └──────┘ │ └──────┘ │ └──────┘ ││ +└──────────┴──────────┴──────────┴───────────────────┘│ +└─────────────────────────────────────────────────────┘ +``` + +Marginal memory per context depends on API surface: ~58 KB bare, +~71 KB with Beam API, ~108 KB beam+url, ~231 KB beam+fetch, +~429 KB with full browser APIs. Individual runtimes cost ~530 KB +JS heap plus a ~2.5 MB OS thread stack each. + +### How it works + +Each pool thread has a lock-free message queue and a `HashMap` of +`ContextId → ContextEntry` (QuickJS context + `RuntimeData`). The +worker loop dequeues messages, looks up the target context by ID, +and dispatches operations (eval, call, reset, destroy, DOM queries, +message delivery, resolve/reject for `Beam.call`). + +`Beam.callSync` uses per-context `SyncCallSlot`s stored in a +`RuntimeData` referenced by both the JS thread and NIF layer. The +NIF writes the result and signals the slot directly — no round-trip +through the pool queue — so the blocked JS thread wakes immediately. + +`Beam.call` (async) works through a drain callback: when the JS +thread is in `await_promise` waiting for a Promise to resolve, it +periodically calls `drain_fn` which pulls messages from the pool queue +and routes resolve/reject messages to the correct context. + +### Context lifecycle + +Each `QuickBEAM.Context` is a lightweight GenServer that: +1. On `init`: asks the pool to create a `JSContext` on one of its + threads, installs polyfills (browser/node/beam), snapshots builtins +2. On `eval`/`call`: enqueues work to the pool thread via NIF, + receives the result as a message +3. On `terminate`: sends a destroy command to free the `JSContext` + +Contexts are isolated — separate globals, separate prototypes — but +share the runtime's GC and parser. Prototype pollution in one context +does not affect another. + +### Granular API groups + +Instead of loading all browser APIs, contexts can request individual +groups to minimize memory: + +```elixir +QuickBEAM.Context.start_link(pool: pool, apis: [:beam, :fetch]) # 231 KB +QuickBEAM.Context.start_link(pool: pool, apis: [:beam, :url]) # 108 KB +``` + +Available groups: `:fetch`, `:websocket`, `:worker`, `:channel`, +`:eventsource`, `:url`, `:crypto`, `:compression`, `:buffer`, `:dom`, +`:console`, `:storage`, `:locks`. Dependencies auto-resolve (e.g. +`:fetch` includes EventTarget/AbortController, `:websocket` includes +the message dispatcher). + +The `:browser` atom expands to all groups but uses a monolithic bundle +for better code sharing. + +### Precompiled bytecode + +Polyfill JS is compiled to QuickJS bytecode once (on first use) and +cached in `persistent_term`. New contexts load bytecodes via +`JS_ReadObject` + `JS_EvalFunction` instead of parsing JS text — +~3.2x faster context creation. + +### QuickJS patches + +QuickBEAM patches QuickJS-NG with per-context resource controls: + +**Per-context memory tracking** — All context-level allocators +(`js_malloc`, `js_calloc`, `js_realloc`, `js_free`) track a +`malloc_size` counter on the `JSContext`. When `malloc_limit` is set, +allocations exceeding the limit trigger OOM. The runtime-level memory +tracking remains unchanged (cumulative across all contexts). + +```elixir +{:ok, ctx} = QuickBEAM.Context.start_link(pool: pool, memory_limit: 512_000) +{:ok, %{context_malloc_size: 92_000}} = QuickBEAM.Context.memory_usage(ctx) +``` + +**Per-context reduction limits** — Each interrupt check (~10K opcodes) +increments a `reduction_count` on the `JSContext`. When it exceeds +`reduction_limit`, an uncatchable error terminates the current eval. +The count resets before each eval/call, so the limit is per-operation. +The context remains usable after hitting the limit. + +```elixir +{:ok, ctx} = QuickBEAM.Context.start_link(pool: pool, max_reductions: 100_000) +# A 10M-iteration loop gets interrupted; next eval works fine +``` + ## Supervision Runtimes are GenServers — they fit naturally into OTP supervision diff --git a/lib/quickbeam.ex b/lib/quickbeam.ex index c7b65c3d8..761842e96 100644 --- a/lib/quickbeam.ex +++ b/lib/quickbeam.ex @@ -247,13 +247,6 @@ defmodule QuickBEAM do QuickBEAM.Runtime.send_message(runtime, message) end - @user_globals_js """ - (() => { - const names = Object.getOwnPropertyNames(globalThis).sort(); - return names.filter(k => !k.startsWith("__qb_") && !(k in globalThis.__qb_builtins)); - })() - """ - @doc """ List global names defined in the JS context. @@ -271,10 +264,10 @@ defmodule QuickBEAM do """ @spec globals(runtime(), keyword()) :: {:ok, [String.t()]} | {:error, QuickBEAM.JSError.t()} def globals(runtime, opts \\ []) do - if Keyword.get(opts, :user_only, false) do - eval(runtime, @user_globals_js) - else - eval(runtime, "Object.getOwnPropertyNames(globalThis).sort()") + user_only = Keyword.get(opts, :user_only, false) + + with {:ok, names} <- GenServer.call(runtime, {:list_globals, user_only}, :infinity) do + {:ok, Enum.sort(names)} end end @@ -298,7 +291,7 @@ defmodule QuickBEAM do """ @spec get_global(runtime(), String.t()) :: js_result() def get_global(runtime, name) when is_binary(name) do - eval(runtime, "globalThis[#{inspect(name)}]") + GenServer.call(runtime, {:get_global, name}, :infinity) end @doc """ diff --git a/lib/quickbeam/beam_call.zig b/lib/quickbeam/beam_call.zig index 639596f07..2fc0a574c 100644 --- a/lib/quickbeam/beam_call.zig +++ b/lib/quickbeam/beam_call.zig @@ -110,7 +110,10 @@ fn beam_call_sync_impl( self.rd.sync_slots_mutex.unlock(); return qjs.JS_ThrowInternalError(ctx, "runtime shutting down"); } - slot.done.timedWait(10_000_000) catch {}; + if (self.drain_fn) |drain| { + drain(self); + } + slot.done.timedWait(1_000_000) catch {}; } self.rd.sync_slots_mutex.lock(); diff --git a/lib/quickbeam/context.ex b/lib/quickbeam/context.ex new file mode 100644 index 000000000..d7f976034 --- /dev/null +++ b/lib/quickbeam/context.ex @@ -0,0 +1,614 @@ +defmodule QuickBEAM.Context do + @moduledoc """ + A lightweight JS execution context on a shared runtime thread. + + Unlike `QuickBEAM.Runtime`, a context does not spawn a dedicated OS thread. + Many contexts share a single `JSRuntime` thread managed by a + `QuickBEAM.ContextPool`. This makes each context ~58 KB (bare) to + ~429 KB (full browser APIs) vs ~2 MB+ for a full runtime — ideal for + per-connection state in Phoenix LiveView. + + ## Example + + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, 3} = QuickBEAM.Context.eval(ctx, "1 + 2") + QuickBEAM.Context.stop(ctx) + + ## With LiveView + + def mount(_params, _session, socket) do + {:ok, ctx} = QuickBEAM.Context.start_link( + pool: MyApp.JSPool, + handlers: %{"db.query" => &MyApp.query/1} + ) + {:ok, assign(socket, js: ctx)} + end + + def handle_event("click", params, socket) do + {:ok, html} = QuickBEAM.Context.call(socket.assigns.js, "handleClick", [params]) + {:noreply, push_event(socket, "update", %{html: html})} + end + + `start_link/1` links the context to the calling process, so it + automatically terminates (and cleans up its JS context) when the + LiveView process exits. No explicit `terminate` callback needed. + """ + use GenServer + + @enforce_keys [:pool_resource, :context_id] + defstruct [ + :pool_resource, + :context_id, + :pool, + handlers: %{}, + pending: %{}, + workers: %{}, + next_worker_id: 1 + ] + + @type t :: %__MODULE__{ + pool_resource: reference(), + context_id: pos_integer(), + pool: GenServer.server() | nil, + handlers: map(), + pending: map(), + workers: map(), + next_worker_id: pos_integer() + } + + def child_spec(opts) do + id = Keyword.get(opts, :id, Keyword.get(opts, :name, __MODULE__)) + + %{ + id: id, + start: {__MODULE__, :start_link, [opts]} + } + end + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts) do + {pool, opts} = Keyword.pop!(opts, :pool) + + GenServer.start_link(__MODULE__, [{:pool, pool} | opts], Keyword.take(opts, [:name])) + end + + @spec eval(GenServer.server(), String.t(), keyword()) :: {:ok, term()} | {:error, String.t()} + def eval(server, code, opts \\ []) when is_binary(code) do + timeout_ms = Keyword.get(opts, :timeout, 0) + GenServer.call(server, {:eval, code, timeout_ms}, :infinity) + end + + @spec call(GenServer.server(), String.t(), list(), keyword()) :: + {:ok, term()} | {:error, String.t()} + def call(server, fn_name, args \\ [], opts \\ []) when is_binary(fn_name) and is_list(args) do + timeout_ms = Keyword.get(opts, :timeout, 0) + GenServer.call(server, {:call, fn_name, args, timeout_ms}, :infinity) + end + + @spec reset(GenServer.server()) :: :ok | {:error, String.t()} + def reset(server) do + GenServer.call(server, :reset, :infinity) + end + + @spec stop(GenServer.server()) :: :ok + def stop(server) do + GenServer.stop(server) + end + + @spec get_global(GenServer.server(), String.t()) :: {:ok, term()} + def get_global(server, name) when is_binary(name) do + GenServer.call(server, {:get_global, name}, :infinity) + end + + @spec set_global(GenServer.server(), String.t(), term()) :: :ok + def set_global(server, name, value) when is_binary(name) do + GenServer.call(server, {:set_global, name, value}, :infinity) + end + + @spec send_message(GenServer.server(), term()) :: :ok + def send_message(server, message) do + GenServer.cast(server, {:send_message, message}) + end + + @spec dom_find(GenServer.server(), String.t()) :: {:ok, tuple() | nil} + def dom_find(server, selector) do + GenServer.call(server, {:dom_find, selector}, :infinity) + end + + @spec dom_find_all(GenServer.server(), String.t()) :: {:ok, list()} + def dom_find_all(server, selector) do + GenServer.call(server, {:dom_find_all, selector}, :infinity) + end + + @spec dom_text(GenServer.server(), String.t()) :: {:ok, String.t()} + def dom_text(server, selector) do + GenServer.call(server, {:dom_text, selector}, :infinity) + end + + @spec dom_html(GenServer.server()) :: {:ok, String.t()} + def dom_html(server) do + GenServer.call(server, :dom_html, :infinity) + end + + @spec memory_usage(GenServer.server()) :: {:ok, map()} + def memory_usage(server) do + GenServer.call(server, :memory_usage, :infinity) + end + + @beam_js QuickBEAM.JS.beam_js() + @node_js QuickBEAM.JS.node_js() + + @impl true + def init(opts) do + pool = Keyword.fetch!(opts, :pool) + user_handlers = Keyword.get(opts, :handlers, %{}) + + apis = + case Keyword.get(opts, :apis, [:browser]) do + false -> [] + nil -> [] + api when is_atom(api) -> [api] + list when is_list(list) -> list + end + + has_browser_apis = Enum.any?(apis, &(&1 not in [:beam, :node])) + + builtin_handlers = QuickBEAM.Runtime.beam_handlers() + + builtin_handlers = + if has_browser_apis, + do: Map.merge(builtin_handlers, QuickBEAM.Runtime.browser_handlers()), + else: builtin_handlers + + builtin_handlers = + if :node in apis, + do: Map.merge(builtin_handlers, QuickBEAM.Runtime.node_handlers()), + else: builtin_handlers + + worker_handlers = + if has_browser_apis do + %{ + "__worker_spawn" => {:context_worker, :spawn}, + "__worker_terminate" => {:context_worker, :terminate}, + "__worker_post_to_child" => {:context_worker, :post_to_child} + } + else + %{} + end + + merged_handlers = builtin_handlers |> Map.merge(worker_handlers) |> Map.merge(user_handlers) + + memory_limit = Keyword.get(opts, :memory_limit, 0) + max_reductions = Keyword.get(opts, :max_reductions, 0) + + {pool_resource, context_id} = + QuickBEAM.ContextPool.create_context(pool, self(), + memory_limit: memory_limit, + max_reductions: max_reductions + ) + + state = %__MODULE__{ + pool_resource: pool_resource, + context_id: context_id, + pool: pool, + handlers: merged_handlers + } + + install_builtins(state, apis) + + case Keyword.fetch(opts, :script) do + :error -> + {:ok, state} + + {:ok, path} -> + case load_script(state, path) do + {:ok, state} -> {:ok, state} + {:error, reason} -> {:stop, reason} + end + end + end + + defp install_builtins(state, apis) do + js_sources = QuickBEAM.JS.js_for_apis(apis) + + for bc <- get_bytecode(apis, js_sources), do: sync_load_bytecode(state, bc) + + if :node in apis do + for bc <- get_bytecode_for(:node, @node_js), do: sync_load_bytecode(state, bc) + end + + if apis != [] do + for bc <- get_bytecode_for(:beam, @beam_js), do: sync_load_bytecode(state, bc) + end + end + + defp get_bytecode(apis, js_sources) do + key = {__MODULE__, :bytecode, :crypto.hash(:md5, :erlang.term_to_binary(apis))} + + case :persistent_term.get(key, nil) do + nil -> + bytecodes = compile_to_bytecode(js_sources) + :persistent_term.put(key, bytecodes) + bytecodes + + cached -> + cached + end + end + + defp get_bytecode_for(group, source) do + key = {__MODULE__, :bytecode, group} + + case :persistent_term.get(key, nil) do + nil -> + bytecodes = compile_to_bytecode(source) + :persistent_term.put(key, bytecodes) + bytecodes + + cached -> + cached + end + end + + defp compile_to_bytecode(source_list) do + {:ok, rt} = QuickBEAM.start(apis: false) + + bytecodes = + Enum.map(source_list, fn js -> + {:ok, bc} = QuickBEAM.compile(rt, js) + bc + end) + + QuickBEAM.stop(rt) + bytecodes + end + + defp sync_load_bytecode(state, bytecode) do + ref = + QuickBEAM.Native.pool_load_bytecode( + state.pool_resource, + state.context_id, + bytecode + ) + + receive do + {^ref, result} -> result + after + 30_000 -> {:error, "NIF timeout"} + end + end + + defp load_script(state, path) do + case File.read(path) do + {:ok, code} -> + ref = QuickBEAM.Native.pool_eval(state.pool_resource, state.context_id, code, 0) + await_eval_ref(ref, state) + + {:error, reason} -> + {:error, {:script_not_found, path, reason}} + end + end + + defp await_eval_ref(ref, state) do + receive do + {^ref, {:ok, _}} -> + {:ok, state} + + {^ref, {:error, reason}} -> + {:error, {:script_error, reason}} + + {:beam_call, _call_id, _handler, _args} = msg -> + {:noreply, state} = handle_info(msg, state) + await_eval_ref(ref, state) + after + 30_000 -> {:error, :script_timeout} + end + end + + @impl true + def handle_call({:eval, code, timeout_ms}, from, state) do + ref = QuickBEAM.Native.pool_eval(state.pool_resource, state.context_id, code, timeout_ms) + + transform = fn + {:ok, value} -> {:ok, value} + {:error, value} -> {:error, QuickBEAM.JSError.from_js_value(value)} + end + + {:noreply, put_pending(state, ref, from, transform)} + end + + def handle_call({:call, fn_name, args, timeout_ms}, from, state) do + ref = + QuickBEAM.Native.pool_call_function( + state.pool_resource, + state.context_id, + fn_name, + args, + timeout_ms + ) + + transform = fn + {:ok, value} -> {:ok, value} + {:error, value} -> {:error, QuickBEAM.JSError.from_js_value(value)} + end + + {:noreply, put_pending(state, ref, from, transform)} + end + + def handle_call({:dom_find, selector}, from, state) do + ref = QuickBEAM.Native.pool_dom_find(state.pool_resource, state.context_id, selector) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call({:dom_find_all, selector}, from, state) do + ref = QuickBEAM.Native.pool_dom_find_all(state.pool_resource, state.context_id, selector) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call({:dom_text, selector}, from, state) do + ref = QuickBEAM.Native.pool_dom_text(state.pool_resource, state.context_id, selector) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call(:dom_html, from, state) do + ref = QuickBEAM.Native.pool_dom_html(state.pool_resource, state.context_id) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call(:memory_usage, from, state) do + ref = QuickBEAM.Native.pool_memory_usage(state.pool_resource, state.context_id) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call(:reset, from, state) do + ref = QuickBEAM.Native.pool_reset_context(state.pool_resource, state.context_id) + + transform = fn + {:ok, _} -> :ok + {:error, msg} -> {:error, msg} + end + + {:noreply, put_pending(state, ref, from, transform)} + end + + def handle_call({:get_global, name}, from, state) do + ref = QuickBEAM.Native.pool_get_global(state.pool_resource, state.context_id, name) + {:noreply, put_pending(state, ref, from, nil)} + end + + def handle_call({:set_global, name, value}, _from, state) do + QuickBEAM.Native.pool_define_global(state.pool_resource, state.context_id, name, value) + {:reply, :ok, state} + end + + @impl true + def handle_cast({:send_message, message}, state) do + QuickBEAM.Native.pool_send_message(state.pool_resource, state.context_id, message) + {:noreply, state} + end + + @impl true + def handle_info({:beam_call, call_id, handler_name, args}, state) do + resource = state.pool_resource + context_id = state.context_id + handlers = state.handlers + + case Map.get(handlers, handler_name) do + nil -> + QuickBEAM.Native.pool_reject_call_term( + resource, + context_id, + call_id, + "Unknown handler: #{handler_name}" + ) + + {:noreply, state} + + {:context_worker, action} -> + handle_worker_call(action, args, call_id, state) + + handler -> + Task.start(fn -> + try do + args = if is_list(args), do: args, else: [args] + result = handler.(args) + + QuickBEAM.Native.pool_resolve_call_term(resource, context_id, call_id, result) + rescue + e -> + QuickBEAM.Native.pool_reject_call_term( + resource, + context_id, + call_id, + Exception.message(e) + ) + end + end) + + {:noreply, state} + end + end + + def handle_info({:worker_started, worker_id, child_pid}, state) do + ref = Process.monitor(child_pid) + workers = Map.put(state.workers, ref, {child_pid, worker_id}) + {:noreply, %{state | workers: workers}} + end + + def handle_info({:worker_msg, worker_id, data}, state) do + QuickBEAM.Native.pool_send_message( + state.pool_resource, + state.context_id, + ["__worker_msg", worker_id, data] + ) + + {:noreply, state} + end + + def handle_info({:worker_error, worker_id, error}, state) do + message = + if is_struct(error), do: Map.get(error, :message, "Worker error"), else: "Worker error" + + QuickBEAM.Native.pool_send_message( + state.pool_resource, + state.context_id, + ["__worker_err", worker_id, message] + ) + + {:noreply, state} + end + + def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do + {_worker_id, workers} = Map.pop(state.workers, ref) + {:noreply, %{state | workers: workers}} + end + + def handle_info({ref, result}, state) when is_reference(ref) do + case Map.pop(state.pending, ref) do + {nil, _} -> + {:noreply, state} + + {{from, nil}, pending} -> + GenServer.reply(from, result) + {:noreply, %{state | pending: pending}} + + {{from, transform}, pending} -> + GenServer.reply(from, transform.(result)) + {:noreply, %{state | pending: pending}} + end + end + + def handle_info(_msg, state) do + {:noreply, state} + end + + @impl true + def terminate(_reason, state) do + for {_ref, {pid, _id}} <- state.workers do + Process.exit(pid, :shutdown) + end + + QuickBEAM.Native.pool_destroy_context(state.pool_resource, state.context_id) + :ok + end + + # ── Worker lifecycle ── + + @worker_bootstrap """ + globalThis.self = globalThis; + self.postMessage = function(data) { + Beam.call("__worker_post", data); + }; + Object.defineProperty(self, "onmessage", { + set(handler) { + Beam.onMessage(msg => handler({ data: msg })); + }, + configurable: true, + }); + """ + + defp handle_worker_call(:spawn, args, call_id, state) do + [script] = if is_list(args), do: args, else: [args] + parent_pid = self() + resource = state.pool_resource + pool = state.pool + + worker_id = state.next_worker_id + + Task.start(fn -> + {:ok, child} = + QuickBEAM.Context.start_link( + pool: pool, + apis: false, + handlers: %{ + "__worker_post" => fn [data] -> + send(parent_pid, {:worker_msg, worker_id, data}) + nil + end + } + ) + + send(parent_pid, {:worker_started, worker_id, child}) + + QuickBEAM.Context.eval(child, @worker_bootstrap) + + case QuickBEAM.Context.eval(child, script) do + {:ok, _} -> :ok + {:error, err} -> send(parent_pid, {:worker_error, worker_id, err}) + end + end) + + QuickBEAM.Native.pool_resolve_call_term(resource, state.context_id, call_id, worker_id) + {:noreply, %{state | next_worker_id: worker_id + 1}} + end + + defp handle_worker_call(:terminate, args, call_id, state) do + [worker_id] = if is_list(args), do: args, else: [args] + + case find_worker(state.workers, worker_id) do + {ref, pid} -> + Process.demonitor(ref, [:flush]) + + Task.start(fn -> + try do + QuickBEAM.Context.stop(pid) + catch + :exit, _ -> :ok + end + end) + + workers = Map.delete(state.workers, ref) + + QuickBEAM.Native.pool_resolve_call_term( + state.pool_resource, + state.context_id, + call_id, + nil + ) + + {:noreply, %{state | workers: workers}} + + nil -> + QuickBEAM.Native.pool_resolve_call_term( + state.pool_resource, + state.context_id, + call_id, + nil + ) + + {:noreply, state} + end + end + + defp handle_worker_call(:post_to_child, args, call_id, state) do + [worker_id, data] = if is_list(args), do: args, else: [args] + + case find_worker(state.workers, worker_id) do + {_ref, pid} -> + QuickBEAM.Context.send_message(pid, data) + + nil -> + :ok + end + + QuickBEAM.Native.pool_resolve_call_term( + state.pool_resource, + state.context_id, + call_id, + nil + ) + + {:noreply, state} + end + + defp find_worker(workers, worker_id) do + Enum.find_value(workers, fn {ref, {pid, id}} -> + if id == worker_id, do: {ref, pid} + end) + end + + defp put_pending(state, ref, from, transform) do + %{state | pending: Map.put(state.pending, ref, {from, transform})} + end +end diff --git a/lib/quickbeam/context_pool.ex b/lib/quickbeam/context_pool.ex new file mode 100644 index 000000000..acd0971d7 --- /dev/null +++ b/lib/quickbeam/context_pool.ex @@ -0,0 +1,89 @@ +defmodule QuickBEAM.ContextPool do + @moduledoc """ + A pool of JS runtime threads that host lightweight contexts. + + Each pool thread runs a single `JSRuntime` that can hold many + `JSContext` instances. Contexts are ~58 KB to ~429 KB each depending + on API surface (no dedicated OS thread), making it practical to run + thousands concurrently. + + ## Example + + # Start a pool with 4 runtime threads + {:ok, pool} = QuickBEAM.ContextPool.start_link(name: MyApp.JSPool, size: 4) + + # Create lightweight contexts on it + {:ok, ctx} = QuickBEAM.Context.start_link(pool: MyApp.JSPool) + {:ok, 42} = QuickBEAM.Context.eval(ctx, "40 + 2") + + ## Options + + * `:name` — registered name for the pool + * `:size` — number of runtime threads (default: `System.schedulers_online()`) + * `:memory_limit` — maximum JS heap per thread in bytes (default: 256 MB) + * `:max_stack_size` — maximum JS call stack in bytes (default: 1 MB) + """ + use GenServer + + defstruct [:threads, next_id: 1, next_thread: 0] + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []) do + GenServer.start_link(__MODULE__, opts, Keyword.take(opts, [:name])) + end + + @doc false + @spec create_context(GenServer.server(), pid(), keyword()) :: {reference(), pos_integer()} + def create_context(pool, owner_pid, opts \\ []) do + GenServer.call(pool, {:create_context, owner_pid, opts}, :infinity) + end + + @impl true + def init(opts) do + size = Keyword.get(opts, :size, System.schedulers_online()) + + nif_opts = + opts + |> Keyword.take([:memory_limit, :max_stack_size]) + |> Map.new() + + threads = + for _ <- 1..size do + QuickBEAM.Native.pool_start(nif_opts) + end + |> List.to_tuple() + + {:ok, %__MODULE__{threads: threads}} + end + + @impl true + def handle_call({:create_context, owner_pid, opts}, _from, state) do + context_id = state.next_id + thread_idx = rem(state.next_thread, tuple_size(state.threads)) + resource = elem(state.threads, thread_idx) + memory_limit = Keyword.get(opts, :memory_limit, 0) + max_reductions = Keyword.get(opts, :max_reductions, 0) + + ref = QuickBEAM.Native.pool_create_context(resource, context_id, owner_pid, memory_limit, max_reductions) + + receive do + {^ref, {:ok, ^context_id}} -> + new_state = %{state | next_id: context_id + 1, next_thread: thread_idx + 1} + {:reply, {resource, context_id}, new_state} + + {^ref, {:error, reason}} -> + {:reply, {:error, reason}, state} + after + 30_000 -> {:reply, {:error, :timeout}, state} + end + end + + @impl true + def terminate(_reason, state) do + for i <- 0..(tuple_size(state.threads) - 1) do + QuickBEAM.Native.pool_stop(elem(state.threads, i)) + end + + :ok + end +end diff --git a/lib/quickbeam/context_types.zig b/lib/quickbeam/context_types.zig new file mode 100644 index 000000000..981705383 --- /dev/null +++ b/lib/quickbeam/context_types.zig @@ -0,0 +1,202 @@ +const types = @import("types.zig"); +const worker = @import("worker.zig"); + +pub const std = types.std; +pub const beam = types.beam; +pub const e = types.e; +pub const qjs = types.qjs; +pub const gpa = types.gpa; + +pub const ContextId = u64; + +pub const ContextEntry = struct { + state: worker.WorkerState, + rd: types.RuntimeData, + owner_pid: beam.pid, + id: ContextId, +}; + +pub const PoolMessage = union(enum) { + create_context: CreateContextPayload, + destroy_context: DestroyContextPayload, + ctx_eval: CtxEvalPayload, + ctx_load_bytecode: CtxEvalPayload, + ctx_call_fn: CtxCallPayload, + ctx_reset: CtxResetPayload, + ctx_send_message: CtxMessagePayload, + ctx_define_global: CtxDefineGlobalPayload, + ctx_get_global: CtxGetGlobalPayload, + ctx_memory_usage: CtxMemoryPayload, + ctx_dom_op: CtxDomPayload, + ctx_resolve_call: CtxCallResponse, + ctx_reject_call: CtxCallResponse, + ctx_resolve_call_term: CtxCallResponseTerm, + stop, +}; + +pub const CreateContextPayload = struct { + context_id: ContextId, + owner_pid: beam.pid, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, + memory_limit: usize = 0, + max_reductions: i64 = 0, +}; + +pub const DestroyContextPayload = struct { + context_id: ContextId, +}; + +pub const CtxEvalPayload = struct { + context_id: ContextId, + code: []const u8, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, + timeout_ns: u64 = 0, +}; + +pub const CtxCallPayload = struct { + context_id: ContextId, + name: []const u8, + args_env: ?*e.ErlNifEnv, + args_term: e.ErlNifTerm, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, + timeout_ns: u64 = 0, +}; + +pub const CtxResetPayload = struct { + context_id: ContextId, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + +pub const CtxMessagePayload = struct { + context_id: ContextId, + env: ?*e.ErlNifEnv, + term: e.ErlNifTerm, +}; + +pub const CtxDefineGlobalPayload = struct { + context_id: ContextId, + name: [:0]const u8, + env: ?*e.ErlNifEnv, + term: e.ErlNifTerm, +}; + +pub const CtxGetGlobalPayload = struct { + context_id: ContextId, + name: [:0]const u8, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + +pub const CtxMemoryPayload = struct { + context_id: ContextId, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + +pub const CtxDomPayload = struct { + context_id: ContextId, + op: types.DomOp, + selector: []const u8 = "", + attr_name: []const u8 = "", + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + +pub const CtxCallResponse = struct { + context_id: ContextId, + id: u64, + json: []const u8, +}; + +pub const CtxCallResponseTerm = struct { + context_id: ContextId, + id: u64, + env: ?*e.ErlNifEnv, + term: e.ErlNifTerm, + ok: bool, +}; + +pub const PoolMessageNode = struct { + msg: PoolMessage, + next: ?*PoolMessageNode, +}; + +pub const PoolData = struct { + mutex: std.Thread.Mutex, + cond: std.Thread.Condition, + queue_head: ?*PoolMessageNode, + queue_tail: ?*PoolMessageNode, + stopped: bool, + thread: ?std.Thread, + memory_limit: usize = 256 * 1024 * 1024, + max_stack_size: usize = 1024 * 1024, + shutting_down: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), + deadline: ?i128 = null, + // Maps context_id → pointer to context's RuntimeData (for sync call resolution) + rd_map_mutex: std.Thread.Mutex = .{}, + rd_map: std.AutoHashMapUnmanaged(ContextId, *types.RuntimeData) = .{}, +}; + +pub fn pool_enqueue(pd: *PoolData, msg: PoolMessage) void { + const node = gpa.create(PoolMessageNode) catch return; + node.* = .{ .msg = msg, .next = null }; + + pd.mutex.lock(); + defer pd.mutex.unlock(); + + if (pd.queue_tail) |tail| { + tail.next = node; + } else { + pd.queue_head = node; + } + pd.queue_tail = node; + pd.cond.signal(); +} + +pub fn pool_dequeue(pd: *PoolData) ?PoolMessage { + pd.mutex.lock(); + defer pd.mutex.unlock(); + + const node = pd.queue_head orelse return null; + pd.queue_head = node.next; + if (pd.queue_head == null) pd.queue_tail = null; + const msg = node.msg; + gpa.destroy(node); + return msg; +} + +pub fn pool_dequeue_blocking(pd: *PoolData, timeout_ns: ?u64) ?PoolMessage { + pd.mutex.lock(); + + while (pd.queue_head == null and !pd.stopped) { + if (timeout_ns) |t| { + pd.cond.timedWait(&pd.mutex, t) catch break; + } else { + pd.cond.wait(&pd.mutex); + } + } + + const node = pd.queue_head; + if (node) |n| { + pd.queue_head = n.next; + if (pd.queue_head == null) pd.queue_tail = null; + pd.mutex.unlock(); + const msg = n.msg; + gpa.destroy(n); + return msg; + } + + pd.mutex.unlock(); + return null; +} diff --git a/lib/quickbeam/context_worker.zig b/lib/quickbeam/context_worker.zig new file mode 100644 index 000000000..3c0dc9022 --- /dev/null +++ b/lib/quickbeam/context_worker.zig @@ -0,0 +1,483 @@ +const ct = @import("context_types.zig"); +const types = @import("types.zig"); +const worker = @import("worker.zig"); +const beam_proxy = @import("beam_proxy.zig"); +const dom = @import("dom.zig"); + +const std = ct.std; +const beam = ct.beam; +const e = ct.e; +const qjs = ct.qjs; +const gpa = ct.gpa; + +fn interrupt_handler(_: ?*qjs.JSRuntime, user_data: ?*anyopaque) callconv(.c) c_int { + const pd: *ct.PoolData = @ptrCast(@alignCast(user_data)); + if (pd.deadline) |deadline| { + if (std.time.nanoTimestamp() > deadline) return 1; + } + return 0; +} + +// Thread-local state for the drain callback. +// Set before calling do_eval/do_call, cleared after. +threadlocal var tl_pool_data: ?*ct.PoolData = null; +threadlocal var tl_contexts: ?*std.AutoHashMap(ct.ContextId, *ct.ContextEntry) = null; +threadlocal var tl_context_id: ct.ContextId = 0; + +fn install_pump( + pd: *ct.PoolData, + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + context_id: ct.ContextId, + entry: *ct.ContextEntry, +) void { + tl_pool_data = pd; + tl_contexts = contexts; + tl_context_id = context_id; + entry.state.drain_fn = &pool_drain_callback; +} + +fn uninstall_pump(entry: *ct.ContextEntry) void { + entry.state.drain_fn = null; + tl_pool_data = null; + tl_contexts = null; +} + +fn pool_drain_callback(state: *worker.WorkerState) void { + const pd = tl_pool_data orelse return; + const contexts = tl_contexts orelse return; + const active_id = tl_context_id; + + const msg = ct.pool_dequeue(pd) orelse return; + + switch (msg) { + .ctx_resolve_call => |p| { + if (p.context_id == active_id) { + state.resolve_pending(p.id, p.json); + } else { + handle_ctx_resolve_call(contexts, p); + } + }, + .ctx_reject_call => |p| { + if (p.context_id == active_id) { + state.reject_pending(p.id, p.json); + } else { + handle_ctx_reject_call(contexts, p); + } + }, + .ctx_resolve_call_term => |p| { + if (p.context_id == active_id) { + state.resolve_pending_term(p.env, p.term, p.id); + } else { + handle_ctx_resolve_call_term(contexts, p); + } + }, + .ctx_send_message => |p| { + if (p.context_id == active_id) { + state.deliver_message(.{ .env = p.env, .term = p.term }); + } else { + handle_ctx_message(contexts, p); + } + }, + .ctx_define_global => |p| { + if (p.context_id == active_id) { + state.define_global_property(.{ .name = p.name, .env = p.env, .term = p.term }); + } else { + handle_ctx_define_global(contexts, p); + } + }, + .ctx_get_global => |p| { + if (p.context_id == active_id) { + state.get_global_property(.{ .name = p.name, .caller_pid = p.caller_pid, .ref_env = p.ref_env, .ref_term = p.ref_term }); + } else { + handle_ctx_get_global(contexts, p); + } + }, + // Re-enqueue messages that can't be processed during a promise wait + .ctx_eval, .ctx_load_bytecode, .ctx_call_fn, .ctx_reset, .ctx_memory_usage, .ctx_dom_op => { + ct.pool_enqueue(pd, msg); + }, + .create_context, .destroy_context, .stop => { + ct.pool_enqueue(pd, msg); + }, + } +} + +pub fn pool_worker_main(pd: *ct.PoolData) void { + const rt = qjs.JS_NewRuntime() orelse return; + defer qjs.JS_FreeRuntime(rt); + + qjs.JS_SetMemoryLimit(rt, pd.memory_limit); + qjs.JS_SetMaxStackSize(rt, pd.max_stack_size); + qjs.JS_UpdateStackTop(rt); + qjs.JS_SetInterruptHandler(rt, &interrupt_handler, @ptrCast(pd)); + + types.class_ids_mutex.lock(); + _ = qjs.JS_NewClassID(rt, &beam_proxy.class_id); + _ = qjs.JS_NewClassID(rt, &dom.document_class_id); + _ = qjs.JS_NewClassID(rt, &dom.element_class_id); + types.class_ids_mutex.unlock(); + + beam_proxy.initRuntime(rt); + + var contexts = std.AutoHashMap(ct.ContextId, *ct.ContextEntry).init(gpa); + defer { + var it = contexts.valueIterator(); + while (it.next()) |entry| { + entry.*.state.deinit(); + gpa.destroy(entry.*); + } + contexts.deinit(); + } + + while (true) { + const min_timer_ns = find_min_timer(&contexts); + const msg = if (min_timer_ns != null and min_timer_ns.? == 0) + ct.pool_dequeue(pd) + else + ct.pool_dequeue_blocking(pd, min_timer_ns orelse null); + + if (msg) |m| { + switch (m) { + .create_context => |p| handle_create_context(rt, &contexts, pd, p), + .destroy_context => |p| handle_destroy_context(&contexts, pd, p), + .ctx_eval => |p| handle_ctx_eval(&contexts, pd, p), + .ctx_load_bytecode => |p| handle_ctx_load_bytecode(&contexts, pd, p), + .ctx_call_fn => |p| handle_ctx_call(&contexts, pd, p), + .ctx_reset => |p| handle_ctx_reset(&contexts, p), + .ctx_send_message => |p| handle_ctx_message(&contexts, p), + .ctx_define_global => |p| handle_ctx_define_global(&contexts, p), + .ctx_get_global => |p| handle_ctx_get_global(&contexts, p), + .ctx_memory_usage => |p| handle_ctx_memory_usage(&contexts, p), + .ctx_dom_op => |p| handle_ctx_dom_op(&contexts, p), + .ctx_resolve_call => |p| handle_ctx_resolve_call(&contexts, p), + .ctx_reject_call => |p| handle_ctx_reject_call(&contexts, p), + .ctx_resolve_call_term => |p| handle_ctx_resolve_call_term(&contexts, p), + .stop => break, + } + } + + fire_all_timers(&contexts); + } + + pd.mutex.lock(); + pd.stopped = true; + pd.mutex.unlock(); +} + +fn find_min_timer(contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry)) ?u64 { + var min: ?u64 = null; + var it = contexts.valueIterator(); + while (it.next()) |entry| { + if (entry.*.state.next_timer_timeout_ns()) |ns| { + if (min == null or ns < min.?) min = ns; + } + } + return min; +} + +fn fire_all_timers(contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry)) void { + var it = contexts.valueIterator(); + while (it.next()) |entry| { + entry.*.state.fire_expired_timers(); + entry.*.state.drain_jobs(); + } +} + +fn handle_create_context( + rt: *qjs.JSRuntime, + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + pd: *ct.PoolData, + p: ct.CreateContextPayload, +) void { + const ctx = qjs.JS_NewContext(rt) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Failed to create JS context"); + return; + }; + + const entry = gpa.create(ct.ContextEntry) catch { + qjs.JS_FreeContext(ctx); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Out of memory"); + return; + }; + + entry.* = .{ + .rd = .{ + .mutex = .{}, + .cond = .{}, + .queue_head = null, + .queue_tail = null, + .stopped = false, + .thread = null, + }, + .state = undefined, + .owner_pid = p.owner_pid, + .id = p.context_id, + }; + + entry.state = .{ + .ctx = ctx, + .rt = rt, + .owner_pid = p.owner_pid, + .rd = &entry.rd, + .pending_calls = std.AutoHashMap(u64, worker.PendingCall).init(gpa), + .timers = std.AutoHashMap(u64, worker.TimerEntry).init(gpa), + .start_time = std.time.nanoTimestamp(), + }; + + entry.state.install_globals(); + + if (p.memory_limit > 0) { + qjs.JS_SetContextMemoryLimit(ctx, p.memory_limit); + } + if (p.max_reductions > 0) { + qjs.JS_SetContextReductionLimit(ctx, p.max_reductions); + } + + // Register rd pointer so NIFs can find it for sync call resolution + pd.rd_map_mutex.lock(); + pd.rd_map.put(gpa, p.context_id, &entry.rd) catch {}; + pd.rd_map_mutex.unlock(); + + contexts.put(p.context_id, entry) catch { + entry.state.deinit(); + gpa.destroy(entry); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Out of memory"); + return; + }; + + const renv = beam.alloc_env(); + const term = beam.make(p.context_id, .{ .env = renv }); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, true, renv, term.v, ""); +} + +fn handle_destroy_context( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + pd: *ct.PoolData, + p: ct.DestroyContextPayload, +) void { + pd.rd_map_mutex.lock(); + _ = pd.rd_map.remove(p.context_id); + pd.rd_map_mutex.unlock(); + + if (contexts.fetchRemove(p.context_id)) |kv| { + var entry = kv.value; + entry.state.deinit(); + gpa.destroy(entry); + } +} + +fn handle_ctx_eval( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + pd: *ct.PoolData, + p: ct.CtxEvalPayload, +) void { + defer gpa.free(p.code); + + const entry_ptr = contexts.getPtr(p.context_id) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + if (p.timeout_ns > 0) { + pd.deadline = std.time.nanoTimestamp() + @as(i128, p.timeout_ns); + } + + // Pump resolve/reject messages from the pool queue into the context's rd queue + // so that await_promise (which drains rd) can pick them up. + install_pump(pd, contexts, p.context_id, entry); + qjs.JS_ResetContextReductionCount(entry.state.ctx); + var result = worker.Result{}; + entry.state.do_eval(p.code, &result); + uninstall_pump(entry); + + pd.deadline = null; + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, result.ok, result.env, result.term, result.json); +} + +fn handle_ctx_load_bytecode( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + pd: *ct.PoolData, + p: ct.CtxEvalPayload, +) void { + defer gpa.free(p.code); + + const entry_ptr = contexts.getPtr(p.context_id) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + install_pump(pd, contexts, p.context_id, entry); + var result = worker.Result{}; + entry.state.do_load_bytecode(p.code, &result); + uninstall_pump(entry); + + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, result.ok, result.env, result.term, result.json); +} + +fn handle_ctx_call( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + pd: *ct.PoolData, + p: ct.CtxCallPayload, +) void { + defer gpa.free(p.name); + + const entry_ptr = contexts.getPtr(p.context_id) orelse { + if (p.args_env) |ae| beam.free_env(ae); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + if (p.timeout_ns > 0) { + pd.deadline = std.time.nanoTimestamp() + @as(i128, p.timeout_ns); + } + + install_pump(pd, contexts, p.context_id, entry); + qjs.JS_ResetContextReductionCount(entry.state.ctx); + var result = worker.Result{}; + entry.state.do_call(p.name, p.args_env, p.args_term, &result); + uninstall_pump(entry); + + pd.deadline = null; + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, result.ok, result.env, result.term, result.json); +} + +fn handle_ctx_reset( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxResetPayload, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + var result = worker.Result{}; + entry.state.do_reset(&result); + + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, result.ok, result.env, result.term, result.json); +} + +fn handle_ctx_message( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxMessagePayload, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + if (p.env) |env| beam.free_env(env); + return; + }; + const entry = entry_ptr.*; + entry.state.deliver_message(.{ .env = p.env, .term = p.term }); +} + +fn handle_ctx_define_global( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxDefineGlobalPayload, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + if (p.env) |env| beam.free_env(env); + gpa.free(p.name); + return; + }; + const entry = entry_ptr.*; + entry.state.define_global_property(.{ .name = p.name, .env = p.env, .term = p.term }); +} + +fn handle_ctx_get_global( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxGetGlobalPayload, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + gpa.free(p.name); + if (p.ref_env) |env| beam.free_env(env); + return; + }; + const entry = entry_ptr.*; + entry.state.get_global_property(.{ .name = p.name, .caller_pid = p.caller_pid, .ref_env = p.ref_env, .ref_term = p.ref_term }); +} + +fn handle_ctx_memory_usage( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxMemoryPayload, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + var usage: qjs.JSMemoryUsage = undefined; + qjs.JS_ComputeMemoryUsage(entry.state.rt, &usage); + const renv = beam.alloc_env(); + const result_term = beam.make(.{ + .malloc_size = usage.malloc_size, + .malloc_count = usage.malloc_count, + .memory_used_size = usage.memory_used_size, + .atom_count = usage.atom_count, + .str_count = usage.str_count, + .obj_count = usage.obj_count, + .prop_count = usage.prop_count, + .shape_count = usage.shape_count, + .js_func_count = usage.js_func_count, + .c_func_count = usage.c_func_count, + .array_count = usage.array_count, + .context_malloc_size = qjs.JS_GetContextMallocSize(entry.state.ctx), + }, .{ .env = renv }); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, true, renv, result_term.v, ""); +} + +fn handle_ctx_dom_op( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxDomPayload, +) void { + defer gpa.free(p.selector); + defer gpa.free(p.attr_name); + + const entry_ptr = contexts.getPtr(p.context_id) orelse { + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, false, null, null, "Context not found"); + return; + }; + const entry = entry_ptr.*; + + var result = worker.Result{}; + entry.state.do_dom_op_result(p.op, p.selector, p.attr_name, &result); + types.send_reply(p.caller_pid, p.ref_env, p.ref_term, result.ok, result.env, result.term, result.json); +} + +fn handle_ctx_resolve_call( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxCallResponse, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + gpa.free(p.json); + return; + }; + const entry = entry_ptr.*; + entry.state.resolve_pending(p.id, p.json); +} + +fn handle_ctx_reject_call( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxCallResponse, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + gpa.free(p.json); + return; + }; + const entry = entry_ptr.*; + entry.state.reject_pending(p.id, p.json); +} + +fn handle_ctx_resolve_call_term( + contexts: *std.AutoHashMap(ct.ContextId, *ct.ContextEntry), + p: ct.CtxCallResponseTerm, +) void { + const entry_ptr = contexts.getPtr(p.context_id) orelse { + if (p.env) |env| beam.free_env(env); + return; + }; + const entry = entry_ptr.*; + entry.state.resolve_pending_term(p.env, p.term, p.id); +} diff --git a/lib/quickbeam/js.ex b/lib/quickbeam/js.ex index aa3a2b968..3c68451e7 100644 --- a/lib/quickbeam/js.ex +++ b/lib/quickbeam/js.ex @@ -9,6 +9,228 @@ defmodule QuickBEAM.JS do for full option details. """ + # ── Polyfill compilation (compile-time only) ── + + @ts_dir Path.join([__DIR__, "../../priv/ts"]) |> Path.expand() + + for ts <- Path.wildcard(Path.join(@ts_dir, "*.ts")), + not String.ends_with?(ts, ".d.ts") do + @external_resource ts + end + + defmodule Compiler do + @moduledoc false + + def standalone(ts_dir, names) do + for name <- names do + path = Path.join(ts_dir, "#{name}.ts") + source = File.read!(path) + + OXC.transform!(source, Path.basename(path)) + |> then(&"(() => {\n#{&1}\n})();\n") + end + end + + def bundle(ts_dir, barrel) do + barrel_source = File.read!(Path.join(ts_dir, barrel)) + {:ok, specifiers} = OXC.imports(barrel_source, barrel) + + import_names = + specifiers + |> Enum.filter(&String.starts_with?(&1, "./")) + |> Enum.map(&String.trim_leading(&1, "./")) + + all_names = Enum.uniq([Path.rootname(barrel) | import_names]) + + files = + for name <- all_names do + path = Path.join(ts_dir, "#{name}.ts") + {"#{name}.ts", File.read!(path)} + end + + OXC.bundle!(files) + end + + def bundle_modules(ts_dir, modules, exports_barrel) do + files = + for mod <- modules do + path = Path.join(ts_dir, "#{mod}.ts") + {"#{mod}.ts", File.read!(path)} + end + + barrel = {"_barrel.ts", exports_barrel} + OXC.bundle!([barrel | files]) + end + end + + # ── Granular API groups ── + # + # Core events (Event, EventTarget, AbortController, DOMException) are + # auto-included when any group that needs them is requested. + # Each group's barrel assigns its exports to globalThis. + + @core_js [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target abort], + """ + import { AbortSignal, AbortController } from './abort' + import { DOMException } from './dom-exception' + import { Event, MessageEvent, CloseEvent, ErrorEvent } from './event' + import { EventTarget } from './event-target' + Object.assign(globalThis, { + DOMException, Event, MessageEvent, CloseEvent, ErrorEvent, + EventTarget, AbortSignal, AbortController + }) + """ + )] + + @process_js Compiler.standalone(@ts_dir, ~w[process]) + + # Groups that need core events — loaded automatically + @needs_core ~w[fetch websocket worker channel eventsource console locks dom]a + # Groups that need process.ts (message dispatcher) + @needs_process ~w[worker websocket eventsource]a + + @api_groups %{ + fetch: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target abort headers blob streams form-data fetch text-streams], + """ + import { Blob, File } from './blob' + import { Request, Response, fetch } from './fetch' + import { FormData } from './form-data' + import { Headers } from './headers' + import { ReadableStream, ReadableStreamDefaultReader, WritableStream, WritableStreamDefaultWriter, TransformStream } from './streams' + import { TextDecoderStream, TextEncoderStream } from './text-streams' + Object.assign(globalThis, { + ReadableStream, ReadableStreamDefaultReader, + WritableStream, WritableStreamDefaultWriter, TransformStream, + TextEncoderStream, TextDecoderStream, Blob, File, FormData, Headers, + Request, Response, fetch + }) + """ + )], + websocket: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target abort blob streams websocket], + """ + import { WebSocket } from './websocket' + globalThis.WebSocket = WebSocket + """ + )], + worker: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target worker], + """ + import { Worker } from './worker' + globalThis.Worker = Worker + """ + )], + channel: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target broadcast-channel message-channel], + """ + import { BroadcastChannel } from './broadcast-channel' + import { MessageChannel, MessagePort } from './message-channel' + Object.assign(globalThis, { BroadcastChannel, MessageChannel, MessagePort }) + """ + )], + eventsource: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target event-source], + """ + import { EventSource } from './event-source' + globalThis.EventSource = EventSource + """ + )], + url: Compiler.standalone(@ts_dir, ~w[url]), + crypto: Compiler.standalone(@ts_dir, ~w[crypto-subtle]), + compression: Compiler.standalone(@ts_dir, ~w[compression]), + buffer: Compiler.standalone(@ts_dir, ~w[buffer]), + dom: Compiler.standalone(@ts_dir, ~w[class-list style dom-events performance]), + console: [Compiler.bundle_modules( + @ts_dir, + ~w[console-ext], + "import './console-ext'" + )], + storage: [Compiler.bundle_modules( + @ts_dir, + ~w[storage], + "import './storage'" + )], + locks: [Compiler.bundle_modules( + @ts_dir, + ~w[event dom-exception event-target locks], + "import './locks'" + )], + } + + @browser_groups ~w[fetch websocket worker channel eventsource url crypto compression buffer dom console storage locks]a + + @browser_js Compiler.standalone( + @ts_dir, + ~w[url crypto-subtle compression buffer process class-list style] + ) ++ + [Compiler.bundle(@ts_dir, "web-apis.ts")] ++ + Compiler.standalone(@ts_dir, ~w[dom-events performance]) + + @beam_js Compiler.standalone(@ts_dir, ~w[beam-api]) + + @node_js Compiler.standalone( + @ts_dir, + ~w[node-process node-path node-fs node-os node-child-process] + ) + + @doc false + def browser_js, do: @browser_js + @doc false + def beam_js, do: @beam_js + @doc false + def node_js, do: @node_js + @doc false + def core_js, do: @core_js + @doc false + def process_js, do: @process_js + @doc false + def api_group(name), do: Map.fetch!(@api_groups, name) + @doc false + def browser_groups, do: @browser_groups + @doc false + def needs_core, do: @needs_core + @doc false + def needs_process, do: @needs_process + + @doc false + def js_for_apis(apis) do + if :browser in apis do + @browser_js + else + groups = expand_apis(apis) + needs_core? = Enum.any?(groups, &(&1 in @needs_core)) + needs_process? = Enum.any?(groups, &(&1 in @needs_process)) + + js = if needs_core?, do: @core_js, else: [] + js = if needs_process?, do: js ++ @process_js, else: js + + js ++ + Enum.flat_map(groups, fn group -> + Map.fetch!(@api_groups, group) + end) + end + end + + defp expand_apis(apis) do + Enum.flat_map(apis, fn + :browser -> @browser_groups + :beam -> [] + :node -> [] + group when is_atom(group) -> [group] + end) + |> Enum.uniq() + end + + # ── OXC toolchain delegations ── + @doc """ Parse JS/TS source into an AST. @@ -70,6 +292,23 @@ defmodule QuickBEAM.JS do @spec minify!(String.t(), String.t(), keyword()) :: String.t() defdelegate minify!(source, filename, opts \\ []), to: OXC + @doc """ + Extract import specifiers from JS/TS source. + + Faster than `parse/2` + `collect/2` — skips full AST serialization + and returns only the import source strings. Type-only imports + (`import type { ... }`) are excluded. + + {:ok, imports} = QuickBEAM.JS.imports("import { ref } from 'vue'", "test.ts") + # => {:ok, ["vue"]} + """ + @spec imports(String.t(), String.t()) :: {:ok, [String.t()]} | {:error, [String.t()]} + defdelegate imports(source, filename), to: OXC + + @doc "Like `imports/2` but raises on errors." + @spec imports!(String.t(), String.t()) :: [String.t()] + defdelegate imports!(source, filename), to: OXC + @doc """ Bundle multiple TS/JS modules into a single self-executing script. @@ -130,6 +369,22 @@ defmodule QuickBEAM.JS do @spec walk(map(), (map() -> any())) :: :ok defdelegate walk(node, fun), to: OXC + @doc """ + Depth-first post-order AST traversal. Like `Macro.postwalk/2`. + + See `OXC.postwalk/2` for details. + """ + @spec postwalk(map(), (map() -> map())) :: map() + defdelegate postwalk(node, fun), to: OXC + + @doc """ + Depth-first post-order AST traversal with accumulator. Like `Macro.postwalk/3`. + + See `OXC.postwalk/3` for details. + """ + @spec postwalk(map(), acc, (map(), acc -> {map(), acc})) :: {map(), acc} when acc: term() + defdelegate postwalk(node, acc, fun), to: OXC + @doc """ Collect values from an AST tree by walking and filtering nodes. @@ -137,4 +392,12 @@ defmodule QuickBEAM.JS do """ @spec collect(map(), (map() -> {:keep, any()} | :skip)) :: [any()] defdelegate collect(node, fun), to: OXC + + @doc """ + Apply position-based patches to a source string. + + See `OXC.patch_string/2` for details. + """ + @spec patch_string(String.t(), [map()]) :: String.t() + defdelegate patch_string(source, patches), to: OXC end diff --git a/lib/quickbeam/native.ex b/lib/quickbeam/native.ex index 494ec49c6..8f4c5c3ca 100644 --- a/lib/quickbeam/native.ex +++ b/lib/quickbeam/native.ex @@ -49,7 +49,7 @@ defmodule QuickBEAM.Native do {:priv, "c_src/lexbor_bridge.c", @lexbor_cflags} ] ++ @lexbor_src ], - resources: [:RuntimeResource], + resources: [:RuntimeResource, :PoolResource], nifs: [ eval: 3, compile: 2, @@ -65,11 +65,33 @@ defmodule QuickBEAM.Native do reject_call_term: 3, send_message: 2, define_global: 3, + get_global: 2, + delete_globals: 2, + snapshot_globals: 1, + list_globals: 2, memory_usage: 1, dom_find: 2, dom_find_all: 2, dom_text: 2, dom_attr: 3, - dom_html: 1 + dom_html: 1, + pool_start: 1, + pool_stop: 1, + pool_create_context: 5, + pool_destroy_context: 2, + pool_eval: 4, + pool_call_function: 5, + pool_reset_context: 2, + pool_send_message: 3, + pool_define_global: 4, + pool_load_bytecode: 3, + pool_get_global: 3, + pool_memory_usage: 2, + pool_resolve_call_term: 4, + pool_reject_call_term: 4, + pool_dom_find: 3, + pool_dom_find_all: 3, + pool_dom_text: 3, + pool_dom_html: 2 ] end diff --git a/lib/quickbeam/quickbeam.zig b/lib/quickbeam/quickbeam.zig index bbecc1e90..cc6bd1465 100644 --- a/lib/quickbeam/quickbeam.zig +++ b/lib/quickbeam/quickbeam.zig @@ -1,5 +1,7 @@ const types = @import("types.zig"); const worker = @import("worker.zig"); +const ct = @import("context_types.zig"); +const context_worker = @import("context_worker.zig"); const std = types.std; const beam = @import("beam"); @@ -7,6 +9,7 @@ const e = types.e; const gpa = types.gpa; const RuntimeData = types.RuntimeData; const enqueue = types.enqueue; +const pool_enqueue = ct.pool_enqueue; // ──────────────────── Resource ──────────────────── @@ -393,3 +396,381 @@ pub fn define_global(resource: RuntimeResource, name: []const u8, value: beam.te enqueue(resource.unpack(), .{ .define_global = .{ .name = name_copy, .env = val_env, .term = copied } }); return beam.make(.ok, .{}); } + +pub fn snapshot_globals(resource: RuntimeResource) beam.term { + enqueue(resource.unpack(), .{ .snapshot_globals = .{} }); + return beam.make(.ok, .{}); +} + +pub fn list_globals(resource: RuntimeResource, user_only: u8) beam.term { + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + enqueue(resource.unpack(), .{ .list_globals = .{ .user_only = user_only != 0, .caller_pid = caller_pid, .ref_env = ref_env, .ref_term = ref_term } }); + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn delete_globals(resource: RuntimeResource, names: []const beam.term) beam.term { + var name_slices = types.gpa.alloc([:0]const u8, names.len) catch return beam.make(.{ .@"error", "OOM" }, .{}); + for (names, 0..) |name_term, i| { + const name_str = beam.get([]const u8, name_term, .{}) catch { + for (0..i) |j| types.gpa.free(name_slices[j]); + types.gpa.free(name_slices); + return beam.make(.{ .@"error", "bad_name" }, .{}); + }; + name_slices[i] = types.gpa.dupeZ(u8, name_str) catch { + for (0..i) |j| types.gpa.free(name_slices[j]); + types.gpa.free(name_slices); + return beam.make(.{ .@"error", "OOM" }, .{}); + }; + } + enqueue(resource.unpack(), .{ .delete_globals = .{ .names = name_slices } }); + return beam.make(.ok, .{}); +} + +pub fn get_global(resource: RuntimeResource, name: []const u8) beam.term { + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + const name_copy = types.gpa.dupeZ(u8, name) catch return beam.make(.{ .@"error", "enomem" }, .{}); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + enqueue(resource.unpack(), .{ .get_global = .{ .name = name_copy, .caller_pid = caller_pid, .ref_env = ref_env, .ref_term = ref_term } }); + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +// ──────────────────── Context Pool ──────────────────── + +pub const PoolResource = beam.Resource(*ct.PoolData, @import("root"), .{ + .Callbacks = struct { + pub fn dtor(ptr: **ct.PoolData) void { + const data = ptr.*; + data.shutting_down.store(true, .release); + pool_enqueue(data, .{ .stop = {} }); + if (data.thread) |t_| t_.join(); + gpa.destroy(data); + } + }, +}); + +pub fn pool_start(opts: beam.term) !PoolResource { + const data = try gpa.create(ct.PoolData); + data.* = .{ + .mutex = .{}, + .cond = .{}, + .queue_head = null, + .queue_tail = null, + .stopped = false, + .thread = null, + }; + + const env = beam.context.env orelse return error.NoEnv; + if (get_map_uint(env, opts.v, "memory_limit")) |v| { + data.memory_limit = v; + } + if (get_map_uint(env, opts.v, "max_stack_size")) |v| { + data.max_stack_size = v; + } + + const res = try PoolResource.create(data, .{}); + + const min_thread_stack = 2 * 1024 * 1024; + const thread_stack = @max(data.max_stack_size + min_thread_stack, min_thread_stack); + data.thread = std.Thread.spawn(.{ .stack_size = thread_stack }, context_worker.pool_worker_main, .{data}) catch { + gpa.destroy(data); + return error.ThreadSpawn; + }; + + return res; +} + +pub fn pool_stop(resource: PoolResource) beam.term { + const data = resource.unpack(); + data.shutting_down.store(true, .release); + pool_enqueue(data, .{ .stop = {} }); + if (data.thread) |th| { + th.join(); + data.thread = null; + } + return beam.make(.ok, .{}); +} + +pub fn pool_create_context(resource: PoolResource, context_id: u64, owner_pid: beam.pid, memory_limit: u64, max_reductions: i64) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + pool_enqueue(data, .{ .create_context = .{ + .context_id = context_id, + .owner_pid = owner_pid, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + .memory_limit = memory_limit, + .max_reductions = max_reductions, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_destroy_context(resource: PoolResource, context_id: u64) beam.term { + pool_enqueue(resource.unpack(), .{ .destroy_context = .{ .context_id = context_id } }); + return beam.make(.ok, .{}); +} + +pub fn pool_eval(resource: PoolResource, context_id: u64, code: []const u8, timeout_ms: u64) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + const code_copy = gpa.dupe(u8, code) catch return beam.make(.{ .@"error", "OOM" }, .{}); + + pool_enqueue(data, .{ .ctx_eval = .{ + .context_id = context_id, + .code = code_copy, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + .timeout_ns = if (timeout_ms > 0) timeout_ms * 1_000_000 else 0, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_call_function(resource: PoolResource, context_id: u64, name: []const u8, args: beam.term, timeout_ms: u64) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + const args_env = beam.alloc_env(); + const args_copy = e.enif_make_copy(args_env, args.v); + const name_copy = gpa.dupe(u8, name) catch return beam.make(.{ .@"error", "OOM" }, .{}); + + pool_enqueue(data, .{ .ctx_call_fn = .{ + .context_id = context_id, + .name = name_copy, + .args_env = args_env, + .args_term = args_copy, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + .timeout_ns = if (timeout_ms > 0) timeout_ms * 1_000_000 else 0, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_reset_context(resource: PoolResource, context_id: u64) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + pool_enqueue(data, .{ .ctx_reset = .{ + .context_id = context_id, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_send_message(resource: PoolResource, context_id: u64, message: beam.term) beam.term { + const msg_env = beam.alloc_env(); + const copied = e.enif_make_copy(msg_env, message.v); + pool_enqueue(resource.unpack(), .{ .ctx_send_message = .{ + .context_id = context_id, + .env = msg_env, + .term = copied, + } }); + return beam.make(.ok, .{}); +} + +pub fn pool_define_global(resource: PoolResource, context_id: u64, name: []const u8, value: beam.term) beam.term { + const val_env = beam.alloc_env(); + const copied = e.enif_make_copy(val_env, value.v); + const name_copy = gpa.dupeZ(u8, name) catch return beam.make(.{ .@"error", "OOM" }, .{}); + pool_enqueue(resource.unpack(), .{ .ctx_define_global = .{ + .context_id = context_id, + .name = name_copy, + .env = val_env, + .term = copied, + } }); + return beam.make(.ok, .{}); +} + +pub fn pool_get_global(resource: PoolResource, context_id: u64, name: []const u8) beam.term { + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + const name_copy = gpa.dupeZ(u8, name) catch return beam.make(.{ .@"error", "OOM" }, .{}); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + pool_enqueue(resource.unpack(), .{ .ctx_get_global = .{ + .context_id = context_id, + .name = name_copy, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + } }); + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_load_bytecode(resource: PoolResource, context_id: u64, bytecode: []const u8) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + const code_copy = gpa.dupe(u8, bytecode) catch return beam.make(.{ .@"error", "OOM" }, .{}); + + pool_enqueue(data, .{ .ctx_load_bytecode = .{ + .context_id = context_id, + .code = code_copy, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + .timeout_ns = 0, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_memory_usage(resource: PoolResource, context_id: u64) beam.term { + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + pool_enqueue(resource.unpack(), .{ .ctx_memory_usage = .{ + .context_id = context_id, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + } }); + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +pub fn pool_dom_find(resource: PoolResource, context_id: u64, selector: []const u8) beam.term { + return pool_dom_op(resource, context_id, .find, selector, ""); +} + +pub fn pool_dom_find_all(resource: PoolResource, context_id: u64, selector: []const u8) beam.term { + return pool_dom_op(resource, context_id, .find_all, selector, ""); +} + +pub fn pool_dom_text(resource: PoolResource, context_id: u64, selector: []const u8) beam.term { + return pool_dom_op(resource, context_id, .text, selector, ""); +} + +pub fn pool_dom_html(resource: PoolResource, context_id: u64) beam.term { + return pool_dom_op(resource, context_id, .html, "", ""); +} + +fn pool_dom_op(resource: PoolResource, context_id: u64, op: types.DomOp, selector: []const u8, attr_name: []const u8) beam.term { + const data = resource.unpack(); + const env = beam.context.env orelse return beam.make(.{ .@"error", "no env" }, .{}); + + var caller_pid: beam.pid = undefined; + _ = e.enif_self(env, &caller_pid); + const ref_env = beam.alloc_env(); + const ref_term = e.enif_make_ref(ref_env); + + const sel_copy = gpa.dupe(u8, selector) catch return beam.make(.{ .@"error", "OOM" }, .{}); + const attr_copy = gpa.dupe(u8, attr_name) catch { + gpa.free(sel_copy); + return beam.make(.{ .@"error", "OOM" }, .{}); + }; + + pool_enqueue(data, .{ .ctx_dom_op = .{ + .context_id = context_id, + .op = op, + .selector = sel_copy, + .attr_name = attr_copy, + .caller_pid = caller_pid, + .ref_env = ref_env, + .ref_term = ref_term, + } }); + + return beam.term{ .v = e.enif_make_copy(env, ref_term) }; +} + +fn pool_lookup_sync_slot(data: *ct.PoolData, context_id: u64, call_id: u64) ?*types.SyncCallSlot { + data.rd_map_mutex.lock(); + const rd = data.rd_map.get(context_id); + data.rd_map_mutex.unlock(); + + if (rd) |r| { + r.sync_slots_mutex.lock(); + const slot = r.sync_slots.get(call_id); + r.sync_slots_mutex.unlock(); + return slot; + } + return null; +} + +pub fn pool_resolve_call_term(resource: PoolResource, context_id: u64, call_id: u64, value: beam.term) beam.term { + const data = resource.unpack(); + + if (pool_lookup_sync_slot(data, context_id, call_id)) |s| { + const term_env = beam.alloc_env(); + s.result_env = term_env; + s.result_term = e.enif_make_copy(term_env, value.v); + s.ok = true; + s.done.set(); + return beam.make(.ok, .{}); + } + + const msg_env = beam.alloc_env(); + const copied = e.enif_make_copy(msg_env, value.v); + pool_enqueue(data, .{ .ctx_resolve_call_term = .{ + .context_id = context_id, + .id = call_id, + .env = msg_env, + .term = copied, + .ok = true, + } }); + return beam.make(.ok, .{}); +} + +pub fn pool_reject_call_term(resource: PoolResource, context_id: u64, call_id: u64, reason: []const u8) beam.term { + const data = resource.unpack(); + + if (pool_lookup_sync_slot(data, context_id, call_id)) |s| { + const term_env = beam.alloc_env(); + s.result_env = term_env; + s.result_term = beam.make(reason, .{ .env = term_env }).v; + s.ok = false; + s.done.set(); + return beam.make(.ok, .{}); + } + + const reason_copy = gpa.dupe(u8, reason) catch return beam.make(.ok, .{}); + pool_enqueue(data, .{ .ctx_reject_call = .{ + .context_id = context_id, + .id = call_id, + .json = reason_copy, + } }); + return beam.make(.ok, .{}); +} diff --git a/lib/quickbeam/runtime.ex b/lib/quickbeam/runtime.ex index d6977ecbc..6d8ac682c 100644 --- a/lib/quickbeam/runtime.ex +++ b/lib/quickbeam/runtime.ex @@ -161,7 +161,8 @@ defmodule QuickBEAM.Runtime do "__broadcast_post" => {:with_caller, &QuickBEAM.BroadcastChannel.post/2}, "__broadcast_leave" => {:with_caller, &QuickBEAM.BroadcastChannel.leave/2}, "__worker_spawn" => {:with_caller, &QuickBEAM.WorkerAPI.spawn_worker/2}, - "__worker_terminate" => &QuickBEAM.WorkerAPI.terminate_worker/1, + "__worker_terminate" => {:with_caller, &QuickBEAM.WorkerAPI.terminate_worker/2}, + "__worker_post_to_child" => {:with_caller, &QuickBEAM.WorkerAPI.post_to_child/2}, "__locks_request" => {:with_caller, &QuickBEAM.LocksAPI.request_lock/2}, "__locks_release" => {:with_caller, &QuickBEAM.LocksAPI.release_lock/2}, "__locks_query" => &QuickBEAM.LocksAPI.query_locks/1, @@ -233,61 +234,13 @@ defmodule QuickBEAM.Runtime do "__child_process_exec_sync" => &QuickBEAM.NodeChildProcess.exec_sync/1 } - @ts_dir Path.join([__DIR__, "../../priv/ts"]) |> Path.expand() + @browser_js QuickBEAM.JS.browser_js() + @beam_js QuickBEAM.JS.beam_js() + @node_js QuickBEAM.JS.node_js() - # Register @external_resource for all TS source files - for ts <- Path.wildcard(Path.join(@ts_dir, "*.ts")), - not String.ends_with?(ts, ".d.ts") do - @external_resource ts - end - - defmodule Compiler do - @moduledoc false - - def standalone(ts_dir, names) do - for name <- names do - path = Path.join(ts_dir, "#{name}.ts") - source = File.read!(path) - - OXC.transform!(source, Path.basename(path)) - |> then(&"(() => {\n#{&1}\n})();\n") - end - end - - def bundle(ts_dir, barrel) do - barrel_source = File.read!(Path.join(ts_dir, barrel)) - {:ok, specifiers} = OXC.imports(barrel_source, barrel) - - import_names = - specifiers - |> Enum.filter(&String.starts_with?(&1, "./")) - |> Enum.map(&String.trim_leading(&1, "./")) - - all_names = Enum.uniq([Path.rootname(barrel) | import_names]) - - files = - for name <- all_names do - path = Path.join(ts_dir, "#{name}.ts") - {"#{name}.ts", File.read!(path)} - end - - OXC.bundle!(files) - end - end - - @browser_js Compiler.standalone( - @ts_dir, - ~w[url crypto-subtle compression buffer process class-list style] - ) ++ - [Compiler.bundle(@ts_dir, "web-apis.ts")] ++ - Compiler.standalone(@ts_dir, ~w[dom-events performance]) - - @beam_js Compiler.standalone(@ts_dir, ~w[beam-api]) - - @node_js Compiler.standalone( - @ts_dir, - ~w[node-process node-path node-fs node-os node-child-process] - ) + def browser_handlers, do: @browser_handlers + def beam_handlers, do: @beam_handlers + def node_handlers, do: @node_handlers @impl true def init(opts) do @@ -414,12 +367,6 @@ defmodule QuickBEAM.Runtime do end end - @snapshot_builtins_js """ - globalThis.__qb_builtins = Object.create(null); - for (const k of Object.getOwnPropertyNames(globalThis)) - globalThis.__qb_builtins[k] = true; - """ - defp install_defines(_state, defines) when map_size(defines) == 0, do: :ok defp install_defines(state, defines) do @@ -441,7 +388,7 @@ defmodule QuickBEAM.Runtime do for js <- @beam_js, do: sync_eval(state.resource, js) end - sync_eval(state.resource, @snapshot_builtins_js) + QuickBEAM.Native.snapshot_globals(state.resource) end defp sync_eval(resource, code) do @@ -487,14 +434,15 @@ defmodule QuickBEAM.Runtime do QuickBEAM.Native.define_global(state.resource, name, value) end) - deletes = Enum.map_join(names, "; ", fn n -> "delete globalThis[#{inspect(n)}]" end) - wrapped = "try { #{code}\n } finally { #{deletes} }" + ref = QuickBEAM.Native.eval(state.resource, code, timeout_ms) - ref = QuickBEAM.Native.eval(state.resource, wrapped, timeout_ms) + transform = fn result -> + QuickBEAM.Native.delete_globals(state.resource, names) - transform = fn - {:ok, value} -> {:ok, value} - {:error, value} -> {:error, QuickBEAM.JSError.from_js_value(value)} + case result do + {:ok, value} -> {:ok, value} + {:error, value} -> {:error, QuickBEAM.JSError.from_js_value(value)} + end end {:noreply, put_pending(state, ref, from, transform)} @@ -505,6 +453,28 @@ defmodule QuickBEAM.Runtime do {:reply, :ok, state} end + def handle_call({:get_global, name}, from, state) do + ref = QuickBEAM.Native.get_global(state.resource, name) + + transform = fn + {:ok, value} -> {:ok, value} + {:error, value} -> {:error, QuickBEAM.JSError.from_js_value(value)} + end + + {:noreply, %{state | pending: Map.put(state.pending, ref, {from, transform})}} + end + + def handle_call({:list_globals, user_only}, from, state) do + ref = QuickBEAM.Native.list_globals(state.resource, if(user_only, do: 1, else: 0)) + + transform = fn + {:ok, names} -> {:ok, names} + other -> other + end + + {:noreply, %{state | pending: Map.put(state.pending, ref, {from, transform})}} + end + def handle_call({:compile, code}, from, state) do ref = QuickBEAM.Native.compile(state.resource, code) @@ -654,20 +624,46 @@ defmodule QuickBEAM.Runtime do {:noreply, state} end - def handle_info({:worker_monitor, child_pid}, state) do + def handle_info({:worker_register, worker_id, child_pid}, state) do ref = Process.monitor(child_pid) - workers = Map.put(state.workers, ref, child_pid) + workers = Map.put(state.workers, worker_id, {child_pid, ref}) {:noreply, %{state | workers: workers}} end - def handle_info({:worker_error_from_child, child_pid, error}, state) do + def handle_info({:worker_msg, worker_id, data}, state) do + QuickBEAM.Native.send_message(state.resource, ["__worker_msg", worker_id, data]) + {:noreply, state} + end + + def handle_info({:worker_error, worker_id, error}, state) do message = if is_struct(error), do: Map.get(error, :message, "Worker error"), else: "Worker error" - QuickBEAM.Native.send_message(state.resource, ["__worker_err", child_pid, message]) + QuickBEAM.Native.send_message(state.resource, ["__worker_err", worker_id, message]) + {:noreply, state} + end + + def handle_info({:worker_post_to_child, worker_id, data}, state) do + case Map.get(state.workers, worker_id) do + {child_pid, _ref} -> QuickBEAM.send_message(child_pid, data) + nil -> :ok + end + {:noreply, state} end + def handle_info({:worker_terminate, worker_id}, state) do + case Map.pop(state.workers, worker_id) do + {nil, _} -> + {:noreply, state} + + {{child_pid, ref}, workers} -> + Process.demonitor(ref, [:flush]) + Task.start(fn -> QuickBEAM.stop(child_pid) end) + {:noreply, %{state | workers: workers}} + end + end + def handle_info({:eventsource_open, id}, state) do QuickBEAM.Native.send_message(state.resource, ["__eventsource_open", id]) {:noreply, state} @@ -701,8 +697,18 @@ defmodule QuickBEAM.Runtime do end def handle_info({:DOWN, ref, :process, _pid, reason}, state) do - case Map.pop(state.workers, ref) do - {nil, _} -> + case find_worker_by_ref(state.workers, ref) do + {worker_id, _child_pid} -> + workers = Map.delete(state.workers, worker_id) + + unless reason == :normal do + message = inspect(reason) + QuickBEAM.Native.send_message(state.resource, ["__worker_err", worker_id, message]) + end + + {:noreply, %{state | workers: workers}} + + nil -> case Map.pop(state.monitors, ref) do {nil, _} -> {:noreply, state} @@ -711,14 +717,6 @@ defmodule QuickBEAM.Runtime do QuickBEAM.Native.send_message(state.resource, ["__qb_down", callback_id, reason]) {:noreply, %{state | monitors: monitors}} end - - {child_pid, workers} -> - unless reason == :normal do - message = inspect(reason) - QuickBEAM.Native.send_message(state.resource, ["__worker_err", child_pid, message]) - end - - {:noreply, %{state | workers: workers}} end end @@ -742,6 +740,12 @@ defmodule QuickBEAM.Runtime do {:noreply, state} end + defp find_worker_by_ref(workers, ref) do + Enum.find_value(workers, fn {worker_id, {pid, worker_ref}} -> + if worker_ref == ref, do: {worker_id, pid} + end) + end + @impl true def terminate(_reason, %{resource: resource} = state) do drain_beam_calls(resource, state.handlers) diff --git a/lib/quickbeam/types.zig b/lib/quickbeam/types.zig index 9c830fac9..5a9aae7aa 100644 --- a/lib/quickbeam/types.zig +++ b/lib/quickbeam/types.zig @@ -42,6 +42,10 @@ pub const Message = union(enum) { resolve_call_term: CallResponseTerm, send_message: MessagePayload, define_global: SetGlobalPayload, + get_global: GetGlobalPayload, + delete_globals: DeleteGlobalsPayload, + snapshot_globals: SnapshotGlobalsPayload, + list_globals: ListGlobalsPayload, memory_usage: AsyncMemoryPayload, dom_op: AsyncDomPayload, stop, @@ -119,6 +123,26 @@ pub const SetGlobalPayload = struct { term: e.ErlNifTerm, }; +pub const GetGlobalPayload = struct { + name: [:0]const u8, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + +pub const DeleteGlobalsPayload = struct { + names: []const [:0]const u8, +}; + +pub const SnapshotGlobalsPayload = struct {}; + +pub const ListGlobalsPayload = struct { + user_only: bool, + caller_pid: beam.pid, + ref_env: ?*e.ErlNifEnv, + ref_term: e.ErlNifTerm, +}; + pub const MessageNode = struct { msg: Message, next: ?*MessageNode, diff --git a/lib/quickbeam/worker.zig b/lib/quickbeam/worker.zig index fbdaa8e59..0e833b5f4 100644 --- a/lib/quickbeam/worker.zig +++ b/lib/quickbeam/worker.zig @@ -30,6 +30,8 @@ pub const TimerEntry = struct { interval_ns: ?u64, }; +pub const DrainFn = *const fn (*WorkerState) void; + pub const WorkerState = struct { ctx: *qjs.JSContext, rt: *qjs.JSRuntime, @@ -43,7 +45,9 @@ pub const WorkerState = struct { message_handler: qjs.JSValue = js.JS_UNDEFINED, atoms: atom_cache.AtomCache = .{}, dom_data: ?*dom.DocumentData = null, + builtin_snapshot: ?std.StringHashMap(void) = null, buf: [4096]u8 = @splat(0), + drain_fn: ?DrainFn = null, pub fn deinit(self: *WorkerState) void { var call_it = self.pending_calls.valueIterator(); @@ -63,6 +67,12 @@ pub const WorkerState = struct { qjs.JS_FreeValue(self.ctx, self.message_handler); } + if (self.builtin_snapshot) |*snap| { + var kit = snap.keyIterator(); + while (kit.next()) |k| types.gpa.free(k.*); + snap.deinit(); + } + self.atoms.deinit(self.ctx); qjs.JS_FreeContext(self.ctx); } @@ -146,6 +156,118 @@ pub const WorkerState = struct { _ = qjs.JS_SetPropertyStr(self.ctx, global, sg.name.ptr, val); } + pub fn get_global_property(self: *WorkerState, gg: types.GetGlobalPayload) void { + defer types.gpa.free(gg.name); + + const global = qjs.JS_GetGlobalObject(self.ctx); + defer qjs.JS_FreeValue(self.ctx, global); + const val = qjs.JS_GetPropertyStr(self.ctx, global, gg.name.ptr); + defer qjs.JS_FreeValue(self.ctx, val); + + const result_env = beam.alloc_env(); + const result_term = js_to_beam.convert(self.ctx, val, result_env); + + types.send_reply(gg.caller_pid, gg.ref_env, gg.ref_term, true, result_env, result_term, ""); + } + + pub fn snapshot_globals(self: *WorkerState) void { + if (self.builtin_snapshot) |*old| { + var kit = old.keyIterator(); + while (kit.next()) |k| types.gpa.free(k.*); + old.deinit(); + } + + var snap = std.StringHashMap(void).init(types.gpa); + const global = qjs.JS_GetGlobalObject(self.ctx); + defer qjs.JS_FreeValue(self.ctx, global); + + var ptab: [*c]qjs.JSPropertyEnum = null; + var plen: u32 = 0; + if (qjs.JS_GetOwnPropertyNames(self.ctx, &ptab, &plen, global, qjs.JS_GPN_STRING_MASK) < 0) return; + defer { + for (0..plen) |i| qjs.JS_FreeAtom(self.ctx, ptab[i].atom); + qjs.js_free(self.ctx, ptab); + } + + for (0..plen) |i| { + const cstr = qjs.JS_AtomToCString(self.ctx, ptab[i].atom); + if (cstr == null) continue; + defer qjs.JS_FreeCString(self.ctx, cstr); + const name = std.mem.span(cstr); + const duped = types.gpa.dupe(u8, name) catch continue; + snap.put(duped, {}) catch { + types.gpa.free(duped); + }; + } + + self.builtin_snapshot = snap; + } + + pub fn list_globals(self: *WorkerState, lg: types.ListGlobalsPayload) void { + const global = qjs.JS_GetGlobalObject(self.ctx); + defer qjs.JS_FreeValue(self.ctx, global); + + var ptab: [*c]qjs.JSPropertyEnum = null; + var plen: u32 = 0; + if (qjs.JS_GetOwnPropertyNames(self.ctx, &ptab, &plen, global, qjs.JS_GPN_STRING_MASK) < 0) { + const renv = e.enif_alloc_env(); + const empty = e.enif_make_list(renv, 0); + types.send_reply(lg.caller_pid, lg.ref_env, lg.ref_term, true, renv, empty, ""); + return; + } + + const result_env = e.enif_alloc_env(); + var list = e.enif_make_list(result_env, 0); + + var i: usize = plen; + while (i > 0) { + i -= 1; + const cstr = qjs.JS_AtomToCString(self.ctx, ptab[i].atom); + if (cstr == null) continue; + const name_slice = std.mem.span(cstr); + const name_len = name_slice.len; + + var skip = false; + if (lg.user_only) { + if (name_len >= 5 and std.mem.eql(u8, name_slice[0..5], "__qb_")) skip = true; + if (!skip) { + if (self.builtin_snapshot) |snap| { + if (snap.contains(name_slice)) skip = true; + } + } + } + + if (!skip) { + var bin: e.ErlNifBinary = undefined; + if (e.enif_alloc_binary(name_len, &bin) != 0) { + @memcpy(bin.data[0..name_len], name_slice[0..name_len]); + const name_term = e.enif_make_binary(result_env, &bin); + list = e.enif_make_list_cell(result_env, name_term, list); + } + } + + qjs.JS_FreeCString(self.ctx, cstr); + } + + for (0..plen) |j| qjs.JS_FreeAtom(self.ctx, ptab[j].atom); + qjs.js_free(self.ctx, ptab); + + types.send_reply(lg.caller_pid, lg.ref_env, lg.ref_term, true, result_env, list, ""); + } + + pub fn delete_global_names(self: *WorkerState, dg: types.DeleteGlobalsPayload) void { + const global = qjs.JS_GetGlobalObject(self.ctx); + defer qjs.JS_FreeValue(self.ctx, global); + + for (dg.names) |name| { + const atom = qjs.JS_NewAtomLen(self.ctx, name.ptr, name.len); + defer qjs.JS_FreeAtom(self.ctx, atom); + _ = qjs.JS_DeleteProperty(self.ctx, global, atom, 0); + types.gpa.free(name); + } + types.gpa.free(dg.names); + } + pub fn deliver_message(self: *WorkerState, sm: types.MessagePayload) void { const env = sm.env orelse return; defer beam.free_env(env); @@ -451,13 +573,19 @@ pub const WorkerState = struct { } // Still pending — process messages that might resolve it - if (types.dequeue(self.rd)) |msg| { + if (self.drain_fn) |dfn| { + dfn(self); + } else if (types.dequeue(self.rd)) |msg| { switch (msg) { .resolve_call => |rc| self.resolve_pending(rc.id, rc.json), .reject_call => |rc| self.reject_pending(rc.id, rc.json), .resolve_call_term => |rc| self.resolve_pending_term(rc.env, rc.term, rc.id), .send_message => |sm| self.deliver_message(sm), .define_global => |sg| self.define_global_property(sg), + .get_global => |gg| self.get_global_property(gg), + .delete_globals => |dg| self.delete_global_names(dg), + .snapshot_globals => self.snapshot_globals(), + .list_globals => |lg| self.list_globals(lg), .stop => { result.ok = false; result.json = "Runtime stopped"; @@ -627,6 +755,10 @@ pub fn worker_main(rd: *types.RuntimeData, owner_pid: beam.pid) void { .resolve_call_term => |rc| state.resolve_pending_term(rc.env, rc.term, rc.id), .send_message => |sm| state.deliver_message(sm), .define_global => |sg| state.define_global_property(sg), + .get_global => |gg| state.get_global_property(gg), + .delete_globals => |dg| state.delete_global_names(dg), + .snapshot_globals => state.snapshot_globals(), + .list_globals => |lg| state.list_globals(lg), .dom_op => |p| { var result = Result{}; state.do_dom_op_result(p.op, p.selector, p.attr_name, &result); diff --git a/lib/quickbeam/worker_api.ex b/lib/quickbeam/worker_api.ex index 9099491ea..e867fd956 100644 --- a/lib/quickbeam/worker_api.ex +++ b/lib/quickbeam/worker_api.ex @@ -3,44 +3,51 @@ defmodule QuickBEAM.WorkerAPI do @worker_bootstrap """ globalThis.self = globalThis; - const __parentPid = Beam.callSync("__worker_parent"); self.postMessage = function(data) { - Beam.send(__parentPid, ["__worker_msg", Beam.self(), data]); + Beam.call("__worker_post", data); }; Object.defineProperty(self, "onmessage", { - set(handler) { Beam.onMessage(msg => { - if (Array.isArray(msg) && msg[0] === "__worker_msg") { - handler({ data: msg[1] }); - } - }); }, + set(handler) { + Beam.onMessage(msg => handler({ data: msg })); + }, configurable: true, }); """ def spawn_worker([script], parent_pid) do + worker_id = System.unique_integer([:positive]) + {:ok, child} = QuickBEAM.start( handlers: %{ - "__worker_parent" => fn [] -> parent_pid end + "__worker_post" => fn [data] -> + send(parent_pid, {:worker_msg, worker_id, data}) + nil + end } ) - send(parent_pid, {:worker_monitor, child}) + send(parent_pid, {:worker_register, worker_id, child}) QuickBEAM.eval(child, @worker_bootstrap) Task.start(fn -> case QuickBEAM.eval(child, script) do {:ok, _} -> :ok - {:error, err} -> send(parent_pid, {:worker_error_from_child, child, err}) + {:error, err} -> send(parent_pid, {:worker_error, worker_id, err}) end end) - child + worker_id + end + + def post_to_child([worker_id, data], parent_pid) do + send(parent_pid, {:worker_post_to_child, worker_id, data}) + nil end - def terminate_worker([worker_pid]) do - Task.start(fn -> QuickBEAM.stop(worker_pid) end) + def terminate_worker([worker_id], parent_pid) do + send(parent_pid, {:worker_terminate, worker_id}) nil end end diff --git a/priv/c_src/quickjs.c b/priv/c_src/quickjs.c index 277dd7af5..9afc0d1f8 100644 --- a/priv/c_src/quickjs.c +++ b/priv/c_src/quickjs.c @@ -525,6 +525,14 @@ struct JSContext { /* when the counter reaches zero, JSRutime.interrupt_handler is called */ int interrupt_counter; + /* per-context memory tracking */ + size_t malloc_size; + size_t malloc_limit; /* 0 = unlimited */ + + /* per-context reduction counting */ + int64_t reduction_count; + int64_t reduction_limit; /* 0 = unlimited */ + struct list_head loaded_modules; /* list of JSModuleDef.link */ /* if NULL, RegExp compilation is not supported */ @@ -1693,11 +1701,19 @@ void *js_mallocz_rt(JSRuntime *rt, size_t size) void *js_calloc(JSContext *ctx, size_t count, size_t size) { void *ptr; + size_t total = count * size; + if (ctx->malloc_limit > 0 && + ctx->malloc_size + total > ctx->malloc_limit) + { + JS_ThrowOutOfMemory(ctx); + return NULL; + } ptr = js_calloc_rt(ctx->rt, count, size); if (unlikely(!ptr)) { JS_ThrowOutOfMemory(ctx); return NULL; } + ctx->malloc_size += ctx->rt->mf.js_malloc_usable_size(ptr) + MALLOC_OVERHEAD; return ptr; } @@ -1705,11 +1721,18 @@ void *js_calloc(JSContext *ctx, size_t count, size_t size) void *js_malloc(JSContext *ctx, size_t size) { void *ptr; + if (ctx->malloc_limit > 0 && + ctx->malloc_size + size > ctx->malloc_limit) + { + JS_ThrowOutOfMemory(ctx); + return NULL; + } ptr = js_malloc_rt(ctx->rt, size); if (unlikely(!ptr)) { JS_ThrowOutOfMemory(ctx); return NULL; } + ctx->malloc_size += ctx->rt->mf.js_malloc_usable_size(ptr) + MALLOC_OVERHEAD; return ptr; } @@ -1717,16 +1740,30 @@ void *js_malloc(JSContext *ctx, size_t size) void *js_mallocz(JSContext *ctx, size_t size) { void *ptr; + if (ctx->malloc_limit > 0 && + ctx->malloc_size + size > ctx->malloc_limit) + { + JS_ThrowOutOfMemory(ctx); + return NULL; + } ptr = js_mallocz_rt(ctx->rt, size); if (unlikely(!ptr)) { JS_ThrowOutOfMemory(ctx); return NULL; } + ctx->malloc_size += ctx->rt->mf.js_malloc_usable_size(ptr) + MALLOC_OVERHEAD; return ptr; } void js_free(JSContext *ctx, void *ptr) { + if (ptr) { + size_t free_size = ctx->rt->mf.js_malloc_usable_size(ptr) + MALLOC_OVERHEAD; + if (free_size <= ctx->malloc_size) + ctx->malloc_size -= free_size; + else + ctx->malloc_size = 0; + } js_free_rt(ctx->rt, ptr); } @@ -1734,11 +1771,29 @@ void js_free(JSContext *ctx, void *ptr) void *js_realloc(JSContext *ctx, void *ptr, size_t size) { void *ret; + size_t old_size = ptr ? ctx->rt->mf.js_malloc_usable_size(ptr) : 0; + size_t delta = size > old_size ? size - old_size : 0; + if (ctx->malloc_limit > 0 && delta > 0 && + ctx->malloc_size + delta > ctx->malloc_limit) + { + JS_ThrowOutOfMemory(ctx); + return NULL; + } ret = js_realloc_rt(ctx->rt, ptr, size); if (unlikely(!ret && size != 0)) { JS_ThrowOutOfMemory(ctx); return NULL; } + if (ret) { + size_t new_size = ctx->rt->mf.js_malloc_usable_size(ret) + MALLOC_OVERHEAD; + size_t old_tracked = old_size + MALLOC_OVERHEAD; + if (new_size >= old_tracked) + ctx->malloc_size += new_size - old_tracked; + else if (old_tracked - new_size <= ctx->malloc_size) + ctx->malloc_size -= old_tracked - new_size; + else + ctx->malloc_size = 0; + } return ret; } @@ -1746,11 +1801,29 @@ void *js_realloc(JSContext *ctx, void *ptr, size_t size) void *js_realloc2(JSContext *ctx, void *ptr, size_t size, size_t *pslack) { void *ret; + size_t old_size = ptr ? ctx->rt->mf.js_malloc_usable_size(ptr) : 0; + size_t delta = size > old_size ? size - old_size : 0; + if (ctx->malloc_limit > 0 && delta > 0 && + ctx->malloc_size + delta > ctx->malloc_limit) + { + JS_ThrowOutOfMemory(ctx); + return NULL; + } ret = js_realloc_rt(ctx->rt, ptr, size); if (unlikely(!ret && size != 0)) { JS_ThrowOutOfMemory(ctx); return NULL; } + if (ret) { + size_t new_size = ctx->rt->mf.js_malloc_usable_size(ret) + MALLOC_OVERHEAD; + size_t old_tracked = old_size + MALLOC_OVERHEAD; + if (new_size >= old_tracked) + ctx->malloc_size += new_size - old_tracked; + else if (old_tracked - new_size <= ctx->malloc_size) + ctx->malloc_size -= old_tracked - new_size; + else + ctx->malloc_size = 0; + } if (pslack) { size_t new_size = js_malloc_usable_size_rt(ctx->rt, ret); *pslack = (new_size > size) ? new_size - size : 0; @@ -2057,6 +2130,32 @@ void JS_SetMemoryLimit(JSRuntime *rt, size_t limit) rt->malloc_state.malloc_limit = limit; } +void JS_SetContextMemoryLimit(JSContext *ctx, size_t limit) +{ + ctx->malloc_limit = limit; +} + +size_t JS_GetContextMallocSize(JSContext *ctx) +{ + return ctx->malloc_size; +} + +void JS_SetContextReductionLimit(JSContext *ctx, int64_t limit) +{ + ctx->reduction_limit = limit; + ctx->reduction_count = 0; +} + +int64_t JS_GetContextReductionCount(JSContext *ctx) +{ + return ctx->reduction_count; +} + +void JS_ResetContextReductionCount(JSContext *ctx) +{ + ctx->reduction_count = 0; +} + void JS_SetDumpFlags(JSRuntime *rt, uint64_t flags) { #ifdef ENABLE_DUMPS @@ -2473,6 +2572,10 @@ JSContext *JS_NewContextRaw(JSRuntime *rt) ctx->error_back_trace = JS_UNDEFINED; ctx->error_prepare_stack = JS_UNDEFINED; ctx->error_stack_trace_limit = js_int32(10); + ctx->malloc_size = 0; + ctx->malloc_limit = 0; + ctx->reduction_count = 0; + ctx->reduction_limit = 0; init_list_head(&ctx->loaded_modules); JS_AddIntrinsicBasicObjects(ctx); @@ -8116,6 +8219,17 @@ static no_inline __exception int __js_poll_interrupts(JSContext *ctx) { JSRuntime *rt = ctx->rt; ctx->interrupt_counter = JS_INTERRUPT_COUNTER_INIT; + + /* per-context reduction counting */ + if (ctx->reduction_limit > 0) { + ctx->reduction_count += JS_INTERRUPT_COUNTER_INIT; + if (ctx->reduction_count >= ctx->reduction_limit) { + JS_ThrowInternalError(ctx, "reduction limit exceeded"); + JS_SetUncatchableError(ctx, rt->current_exception); + return -1; + } + } + if (rt->interrupt_handler) { if (rt->interrupt_handler(rt, rt->interrupt_opaque)) { JS_ThrowInterrupted(ctx); diff --git a/priv/c_src/quickjs.h b/priv/c_src/quickjs.h index 0f77edeb9..0ae8ee823 100644 --- a/priv/c_src/quickjs.h +++ b/priv/c_src/quickjs.h @@ -511,6 +511,11 @@ JS_EXTERN JSRuntime *JS_NewRuntime(void); JS_EXTERN void JS_SetRuntimeInfo(JSRuntime *rt, const char *info); /* use 0 to disable memory limit */ JS_EXTERN void JS_SetMemoryLimit(JSRuntime *rt, size_t limit); +JS_EXTERN void JS_SetContextMemoryLimit(JSContext *ctx, size_t limit); +JS_EXTERN size_t JS_GetContextMallocSize(JSContext *ctx); +JS_EXTERN void JS_SetContextReductionLimit(JSContext *ctx, int64_t limit); +JS_EXTERN int64_t JS_GetContextReductionCount(JSContext *ctx); +JS_EXTERN void JS_ResetContextReductionCount(JSContext *ctx); JS_EXTERN void JS_SetDumpFlags(JSRuntime *rt, uint64_t flags); JS_EXTERN uint64_t JS_GetDumpFlags(JSRuntime *rt); JS_EXTERN size_t JS_GetGCThreshold(JSRuntime *rt); diff --git a/priv/ts/worker.ts b/priv/ts/worker.ts index a2cb69da7..e577baef7 100644 --- a/priv/ts/worker.ts +++ b/priv/ts/worker.ts @@ -5,10 +5,10 @@ import { EventTarget } from './event-target' type MessageHandler = ((event: { data: unknown }) => void) | null type ErrorHandler = ((event: { message: string; error: unknown }) => void) | null -const workerRegistry = new Map() +const workerRegistry = new Map() class Worker extends EventTarget { - #pid: unknown + #id: number #terminated = false #earlyMessages: unknown[] = [] #onmessage: MessageHandler = null @@ -16,9 +16,8 @@ class Worker extends EventTarget { constructor(script: string) { super() - this.#pid = Beam.callSync('__worker_spawn', script) - const pidKey = JSON.stringify(this.#pid) - workerRegistry.set(pidKey, this) + this.#id = Beam.callSync('__worker_spawn', script) as number + workerRegistry.set(this.#id, this) } get onmessage(): MessageHandler { @@ -37,15 +36,14 @@ class Worker extends EventTarget { postMessage(data: unknown): void { if (this.#terminated) throw new DOMException('Worker has been terminated', 'InvalidStateError') - Beam.send(this.#pid, ['__worker_msg', data]) + Beam.callSync('__worker_post_to_child', this.#id, data) } terminate(): void { if (this.#terminated) return this.#terminated = true - const pidKey = JSON.stringify(this.#pid) - workerRegistry.delete(pidKey) - void Beam.call('__worker_terminate', this.#pid) + workerRegistry.delete(this.#id) + void Beam.call('__worker_terminate', this.#id) } _dispatch(data: unknown): void { @@ -69,12 +67,12 @@ declare const __qb_register_dispatcher: (fn: (msg: unknown) => boolean) => void __qb_register_dispatcher((msg: unknown): boolean => { if (!Array.isArray(msg) || msg.length < 3) return false - const [type, pid, payload] = msg + const [type, id, payload] = msg if (type !== '__worker_msg' && type !== '__worker_err') return false + if (typeof id !== 'number') return false - const pidKey = JSON.stringify(pid) - const worker = workerRegistry.get(pidKey) + const worker = workerRegistry.get(id) if (!worker) return false if (type === '__worker_msg') { diff --git a/test/core/context_pool_stress_test.exs b/test/core/context_pool_stress_test.exs new file mode 100644 index 000000000..45842fc13 --- /dev/null +++ b/test/core/context_pool_stress_test.exs @@ -0,0 +1,720 @@ +defmodule QuickBEAM.Core.ContextPoolStressTest do + use ExUnit.Case + + @moduletag timeout: 120_000 + + # ──────────────────── 1. Scale ──────────────────── + + describe "mass context creation" do + test "1000 contexts on a 4-thread pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..1000 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + ctx + end + + # Spot-check 50 random contexts + samples = Enum.take_random(Enum.with_index(contexts, 1), 50) + + for {ctx, i} <- samples do + {:ok, val} = QuickBEAM.Context.eval(ctx, "id") + assert val == i, "Context #{i} returned #{val}" + end + + for ctx <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "rapid create/destroy churn — 500 cycles" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + + for i <- 1..500 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, val} = QuickBEAM.Context.eval(ctx, "#{i} * 3") + assert val == i * 3 + QuickBEAM.Context.stop(ctx) + end + + # Pool still healthy after churn + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + assert {:ok, 42} = QuickBEAM.Context.eval(ctx, "42") + QuickBEAM.Context.stop(ctx) + end + + test "destroy context while siblings are active on same thread" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx_a} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_b} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_c} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = QuickBEAM.Context.eval(ctx_a, "globalThis.x = 'a'") + {:ok, _} = QuickBEAM.Context.eval(ctx_b, "globalThis.x = 'b'") + {:ok, _} = QuickBEAM.Context.eval(ctx_c, "globalThis.x = 'c'") + + # Destroy B while A and C are still alive + QuickBEAM.Context.stop(ctx_b) + + assert {:ok, "a"} = QuickBEAM.Context.eval(ctx_a, "x") + assert {:ok, "c"} = QuickBEAM.Context.eval(ctx_c, "x") + + QuickBEAM.Context.stop(ctx_a) + QuickBEAM.Context.stop(ctx_c) + end + end + + # ──────────────────── 2. Concurrency ──────────────────── + + describe "thundering herd" do + test "200 tasks hitting 50 contexts simultaneously" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..50 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + {i, ctx} + end + + tasks = + for _ <- 1..200 do + {expected_id, ctx} = Enum.random(contexts) + + Task.async(fn -> + {:ok, val} = QuickBEAM.Context.eval(ctx, "id") + assert val == expected_id + val + end) + end + + results = Task.await_many(tasks, 30_000) + assert length(results) == 200 + assert Enum.all?(results, &is_integer/1) + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "Beam.call from 100 contexts simultaneously" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..100 do + {:ok, ctx} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "slow_echo" => fn [val] -> + Process.sleep(10) + val + end + } + ) + + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.myId = #{i}") + {i, ctx} + end + + tasks = + for {i, ctx} <- contexts do + Task.async(fn -> + {:ok, result} = QuickBEAM.Context.eval(ctx, "await Beam.call('slow_echo', myId)") + assert result == i + result + end) + end + + results = Task.await_many(tasks, 60_000) + assert Enum.sort(results) == Enum.to_list(1..100) + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "Beam.callSync from many contexts on same thread serializes correctly" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + counter = :counters.new(1, [:atomics]) + + contexts = + for i <- 1..20 do + {:ok, ctx} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "count" => fn [val] -> + :counters.add(counter, 1, 1) + val * 2 + end + } + ) + + {i, ctx} + end + + # Sequential — each callSync blocks the single thread + for {i, ctx} <- contexts do + {:ok, result} = QuickBEAM.Context.eval(ctx, "Beam.callSync('count', #{i})") + assert result == i * 2 + end + + assert :counters.get(counter, 1) == 20 + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + end + + # ──────────────────── 3. Memory ──────────────────── + + describe "memory stability" do + test "context create/destroy cycle doesn't leak" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + + # Warm up + for _ <- 1..5 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + QuickBEAM.Context.eval(ctx, """ + globalThis.data = []; + for (let i = 0; i < 1000; i++) data.push({x: i, y: 'test'.repeat(10)}); + """) + + QuickBEAM.Context.stop(ctx) + end + + :erlang.garbage_collect() + Process.sleep(100) + mem_before = :erlang.memory(:total) + + for _ <- 1..100 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + QuickBEAM.Context.eval(ctx, """ + globalThis.data = []; + for (let i = 0; i < 1000; i++) data.push({x: i, y: 'test'.repeat(10)}); + """) + + QuickBEAM.Context.stop(ctx) + end + + :erlang.garbage_collect() + Process.sleep(100) + mem_after = :erlang.memory(:total) + + growth = mem_after - mem_before + + assert growth < 4 * 1024 * 1024, + "BEAM memory grew by #{div(growth, 1024)}KB over 100 create/destroy cycles" + end + + test "rolling context churn over 3 seconds" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + alive = :ets.new(:alive_contexts, [:set, :public]) + + # Seed 50 contexts + for i <- 1..50 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + :ets.insert(alive, {i, ctx}) + end + + :erlang.garbage_collect() + Process.sleep(50) + mem_before = :erlang.memory(:total) + + # Churn for 3 seconds: destroy oldest, create new + deadline = System.monotonic_time(:millisecond) + 3_000 + next_id = 51 + + {_, next_id} = + Enum.reduce_while(Stream.iterate(1, &(&1 + 1)), {1, next_id}, fn destroy_id, + {_, nid} -> + if System.monotonic_time(:millisecond) >= deadline do + {:halt, {destroy_id, nid}} + else + case :ets.lookup(alive, destroy_id) do + [{^destroy_id, ctx}] -> + QuickBEAM.Context.stop(ctx) + :ets.delete(alive, destroy_id) + + [] -> + :ok + end + + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{nid}") + :ets.insert(alive, {nid, ctx}) + {:cont, {destroy_id + 1, nid + 1}} + end + end) + + :erlang.garbage_collect() + Process.sleep(100) + mem_after = :erlang.memory(:total) + + growth = mem_after - mem_before + churn_count = next_id - 51 + + assert growth < 8 * 1024 * 1024, + "BEAM memory grew by #{div(growth, 1024)}KB over #{churn_count} churn cycles" + + # Clean up remaining + :ets.foldl(fn {_, ctx}, _ -> QuickBEAM.Context.stop(ctx) end, nil, alive) + :ets.delete(alive) + end + + test "eval cycles on long-lived context don't leak" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + for _ <- 1..10, do: QuickBEAM.Context.eval(ctx, "JSON.parse(JSON.stringify({a: 1}))") + + :erlang.garbage_collect() + Process.sleep(50) + mem_before = :erlang.memory(:total) + + for _ <- 1..2000 do + QuickBEAM.Context.eval(ctx, "JSON.parse(JSON.stringify({a: [1,2,3], b: 'hello'}))") + end + + :erlang.garbage_collect() + Process.sleep(50) + mem_after = :erlang.memory(:total) + + growth = mem_after - mem_before + + assert growth < 2 * 1024 * 1024, + "BEAM memory grew by #{div(growth, 1024)}KB over 2000 eval cycles" + + QuickBEAM.Context.stop(ctx) + end + end + + # ──────────────────── 4. State Isolation ──────────────────── + + describe "state isolation" do + test "100 contexts each with unique global, no cross-contamination" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..100 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.x = #{i}") + {i, ctx} + end + + # Shuffle and verify — if isolation is broken, we'd see wrong values + shuffled = Enum.shuffle(contexts) + + for {expected, ctx} <- shuffled do + {:ok, val} = QuickBEAM.Context.eval(ctx, "x") + assert val == expected, "Expected #{expected}, got #{val}" + end + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "prototype modification in one context doesn't leak to another" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx_polluter} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_clean} = QuickBEAM.Context.start_link(pool: pool) + + # Pollute prototypes in one context + {:ok, _} = + QuickBEAM.Context.eval(ctx_polluter, """ + Array.prototype.customMethod = function() { return 'polluted'; }; + Object.prototype.leaked = true; + String.prototype.evil = () => 'hacked'; + """) + + # Verify pollution worked locally + {:ok, "polluted"} = QuickBEAM.Context.eval(ctx_polluter, "[].customMethod()") + + # Verify clean context is unaffected + {:ok, "undefined"} = + QuickBEAM.Context.eval(ctx_clean, "typeof [].customMethod") + + {:ok, "undefined"} = + QuickBEAM.Context.eval(ctx_clean, "typeof ({}).leaked") + + {:ok, "undefined"} = + QuickBEAM.Context.eval(ctx_clean, "typeof ''.evil") + + QuickBEAM.Context.stop(ctx_polluter) + QuickBEAM.Context.stop(ctx_clean) + end + + test "globalThis.constructor tampering is isolated" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx_a} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_b} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = + QuickBEAM.Context.eval(ctx_a, """ + globalThis.Math = { PI: 999 }; + globalThis.parseInt = (x) => 'hijacked'; + """) + + {:ok, 999} = QuickBEAM.Context.eval(ctx_a, "Math.PI") + + # ctx_b should have original Math and parseInt + {:ok, pi} = QuickBEAM.Context.eval(ctx_b, "Math.PI") + assert_in_delta pi, 3.14159, 0.001 + + {:ok, 42} = QuickBEAM.Context.eval(ctx_b, "parseInt('42')") + + QuickBEAM.Context.stop(ctx_a) + QuickBEAM.Context.stop(ctx_b) + end + end + + # ──────────────────── 5. Error Recovery ──────────────────── + + describe "error recovery" do + test "errors in one context don't poison siblings" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx_bad} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_good} = QuickBEAM.Context.start_link(pool: pool) + + # Throw errors, stack overflow, syntax error + {:error, _} = QuickBEAM.Context.eval(ctx_bad, "throw new Error('boom')") + + {:error, _} = + QuickBEAM.Context.eval(ctx_bad, "function f() { f() }; f()") + + {:error, _} = QuickBEAM.Context.eval(ctx_bad, "this is not valid javascript !!!") + + # Good context is completely unaffected + assert {:ok, 42} = QuickBEAM.Context.eval(ctx_good, "42") + assert {:ok, "hello"} = QuickBEAM.Context.eval(ctx_good, "'hello'") + + # Bad context recovers too + assert {:ok, 99} = QuickBEAM.Context.eval(ctx_bad, "99") + + QuickBEAM.Context.stop(ctx_bad) + QuickBEAM.Context.stop(ctx_good) + end + + test "100 sequential errors on one context don't corrupt pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + {:ok, ctx_err} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_ok} = QuickBEAM.Context.start_link(pool: pool) + + for i <- 1..100 do + {:error, _} = QuickBEAM.Context.eval(ctx_err, "throw new Error('err #{i}')") + end + + # Pool and sibling healthy + assert {:ok, "alive"} = QuickBEAM.Context.eval(ctx_ok, "'alive'") + assert {:ok, "recovered"} = QuickBEAM.Context.eval(ctx_err, "'recovered'") + + QuickBEAM.Context.stop(ctx_err) + QuickBEAM.Context.stop(ctx_ok) + end + + test "timeout on one context doesn't block others" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx_slow} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_fast} = QuickBEAM.Context.start_link(pool: pool) + + # Start infinite loop with timeout on ctx_slow + task_slow = + Task.async(fn -> + QuickBEAM.Context.eval(ctx_slow, "while(true) {}", timeout: 200) + end) + + # Wait for the slow one to finish (with timeout error) + {:error, _} = Task.await(task_slow, 5_000) + + # Now ctx_fast should respond promptly + start = System.monotonic_time(:millisecond) + {:ok, 7} = QuickBEAM.Context.eval(ctx_fast, "3 + 4") + elapsed = System.monotonic_time(:millisecond) - start + + assert elapsed < 500, "Fast context took #{elapsed}ms, expected < 500ms" + + QuickBEAM.Context.stop(ctx_slow) + QuickBEAM.Context.stop(ctx_fast) + end + + test "OOM in one context doesn't crash the pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1, memory_limit: 4 * 1024 * 1024) + + {:ok, ctx_oom} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx_ok} = QuickBEAM.Context.start_link(pool: pool) + + {:error, _} = + QuickBEAM.Context.eval(ctx_oom, """ + const arrays = []; + while (true) arrays.push(new Array(10000).fill('x')); + """) + + # Sibling and pool still work + assert {:ok, "fine"} = QuickBEAM.Context.eval(ctx_ok, "'fine'") + + QuickBEAM.Context.stop(ctx_oom) + QuickBEAM.Context.stop(ctx_ok) + end + end + + # ──────────────────── 6. Handler Contention ──────────────────── + + describe "handler contention" do + test "slow handler doesn't starve fast contexts on multi-thread pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + {:ok, ctx_slow} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "block" => fn [] -> + Process.sleep(500) + "done" + end + } + ) + + fast_contexts = + for i <- 1..10 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + {i, ctx} + end + + # Start the slow call + slow_task = + Task.async(fn -> + QuickBEAM.Context.eval(ctx_slow, "await Beam.call('block')") + end) + + Process.sleep(50) + + # Fast contexts should respond quickly (on other threads) + fast_tasks = + for {i, ctx} <- fast_contexts do + Task.async(fn -> + start = System.monotonic_time(:millisecond) + {:ok, val} = QuickBEAM.Context.eval(ctx, "id") + elapsed = System.monotonic_time(:millisecond) - start + assert val == i + elapsed + end) + end + + fast_times = Task.await_many(fast_tasks, 5_000) + avg_fast = Enum.sum(fast_times) / length(fast_times) + + assert avg_fast < 200, + "Average fast eval took #{avg_fast}ms during slow handler" + + {:ok, "done"} = Task.await(slow_task, 5_000) + + QuickBEAM.Context.stop(ctx_slow) + for {_, ctx} <- fast_contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "handler error doesn't leak into other contexts' calls" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + + {:ok, ctx_fail} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{"fail" => fn _ -> raise "handler exploded" end} + ) + + {:ok, ctx_ok} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{"echo" => fn [val] -> val end} + ) + + # Fire both in parallel + task_fail = + Task.async(fn -> + QuickBEAM.Context.eval(ctx_fail, "await Beam.call('fail')") + end) + + task_ok = + Task.async(fn -> + QuickBEAM.Context.eval(ctx_ok, "await Beam.call('echo', 42)") + end) + + {:error, _} = Task.await(task_fail, 5_000) + {:ok, 42} = Task.await(task_ok, 5_000) + + # Both contexts still usable + assert {:ok, 1} = QuickBEAM.Context.eval(ctx_fail, "1") + assert {:ok, 2} = QuickBEAM.Context.eval(ctx_ok, "2") + + QuickBEAM.Context.stop(ctx_fail) + QuickBEAM.Context.stop(ctx_ok) + end + + test "Beam.call with concurrent handlers across 50 contexts" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..50 do + {:ok, ctx} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "compute" => fn [val] -> val * val end + } + ) + + {i, ctx} + end + + tasks = + for {i, ctx} <- contexts do + Task.async(fn -> + {:ok, result} = + QuickBEAM.Context.eval(ctx, "await Beam.call('compute', #{i})") + + assert result == i * i + result + end) + end + + results = Task.await_many(tasks, 30_000) + assert Enum.sort(results) == Enum.map(1..50, &(&1 * &1)) |> Enum.sort() + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + end + + # ──────────────────── 7. Messaging Under Load ──────────────────── + + describe "messaging under load" do + test "1000 messages spread across 50 contexts — no cross-delivery" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + contexts = + for i <- 1..50 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = + QuickBEAM.Context.eval(ctx, """ + globalThis.received = []; + globalThis.myId = #{i}; + Beam.onMessage((msg) => { + globalThis.received.push(msg); + }); + """) + + {i, ctx} + end + + # Send 20 messages to each context, each tagged with a unique value + for {i, ctx} <- contexts do + for j <- 1..20 do + QuickBEAM.Context.send_message(ctx, i * 1000 + j) + end + end + + # Wait for delivery + Process.sleep(500) + + # Verify each context received exactly its messages + for {i, ctx} <- contexts do + {:ok, received} = QuickBEAM.Context.eval(ctx, "globalThis.received") + expected = for j <- 1..20, do: i * 1000 + j + + assert Enum.sort(received) == Enum.sort(expected), + "Context #{i}: expected #{inspect(expected)}, got #{inspect(received)}" + end + + for {_, ctx} <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "messages during Beam.call don't get dropped" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + + {:ok, ctx} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "slow" => fn [] -> + Process.sleep(200) + "done" + end + } + ) + + {:ok, _} = + QuickBEAM.Context.eval(ctx, """ + globalThis.msgs = []; + Beam.onMessage((m) => globalThis.msgs.push(m)); + """) + + # Start a slow Beam.call + task = + Task.async(fn -> + QuickBEAM.Context.eval(ctx, "await Beam.call('slow')") + end) + + # While it's running, send messages + Process.sleep(50) + for i <- 1..20, do: QuickBEAM.Context.send_message(ctx, i) + + {:ok, "done"} = Task.await(task, 10_000) + + # Wait for message delivery + eventually(fn -> + {:ok, msgs} = QuickBEAM.Context.eval(ctx, "globalThis.msgs") + assert length(msgs) == 20 + end) + + QuickBEAM.Context.stop(ctx) + end + + test "burst of 500 messages to a single context" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = + QuickBEAM.Context.eval(ctx, """ + globalThis.received = []; + Beam.onMessage((msg) => globalThis.received.push(msg)); + """) + + for i <- 1..500, do: QuickBEAM.Context.send_message(ctx, i) + + eventually(fn -> + {:ok, count} = QuickBEAM.Context.eval(ctx, "globalThis.received.length") + assert count == 500 + end) + + {:ok, received} = QuickBEAM.Context.eval(ctx, "globalThis.received") + assert Enum.sort(received) == Enum.to_list(1..500) + + QuickBEAM.Context.stop(ctx) + end + end + + # ──────────────────── Helpers ──────────────────── + + defp eventually(fun, attempts \\ 40) do + fun.() + rescue + e in [ExUnit.AssertionError] -> + if attempts > 0 do + Process.sleep(50) + eventually(fun, attempts - 1) + else + reraise e, __STACKTRACE__ + end + catch + :exit, reason -> + if attempts > 0 do + Process.sleep(50) + eventually(fun, attempts - 1) + else + exit(reason) + end + end +end diff --git a/test/core/context_pool_test.exs b/test/core/context_pool_test.exs new file mode 100644 index 000000000..611d2462c --- /dev/null +++ b/test/core/context_pool_test.exs @@ -0,0 +1,371 @@ +defmodule QuickBEAM.Core.ContextPoolTest do + use ExUnit.Case, async: true + + test "create pool and context, eval simple expression" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + assert {:ok, 3} = QuickBEAM.Context.eval(ctx, "1 + 2") + + QuickBEAM.Context.stop(ctx) + end + + test "context state persists across evals" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.x = 42") + assert {:ok, 42} = QuickBEAM.Context.eval(ctx, "x") + + QuickBEAM.Context.stop(ctx) + end + + test "multiple contexts are isolated" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx1} = QuickBEAM.Context.start_link(pool: pool) + {:ok, ctx2} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = QuickBEAM.Context.eval(ctx1, "globalThis.x = 'from_ctx1'") + + assert {:ok, "from_ctx1"} = QuickBEAM.Context.eval(ctx1, "x") + assert {:ok, "undefined"} = QuickBEAM.Context.eval(ctx2, "typeof x") + + QuickBEAM.Context.stop(ctx1) + QuickBEAM.Context.stop(ctx2) + end + + test "call JS function" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = QuickBEAM.Context.eval(ctx, "function add(a, b) { return a + b }") + assert {:ok, 5} = QuickBEAM.Context.call(ctx, "add", [2, 3]) + + QuickBEAM.Context.stop(ctx) + end + + test "reset clears context state" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.x = 42") + :ok = QuickBEAM.Context.reset(ctx) + assert {:ok, "undefined"} = QuickBEAM.Context.eval(ctx, "typeof x") + + QuickBEAM.Context.stop(ctx) + end + + test "Beam.call handler" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + + {:ok, ctx} = + QuickBEAM.Context.start_link( + pool: pool, + handlers: %{ + "greet" => fn [name] -> "Hello, #{name}!" end + } + ) + + assert {:ok, "Hello, world!"} = + QuickBEAM.Context.eval(ctx, ~s[await Beam.call("greet", "world")]) + + QuickBEAM.Context.stop(ctx) + end + + test "many contexts on one pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + + contexts = + for i <- 1..50 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + ctx + end + + results = + for ctx <- contexts do + {:ok, val} = QuickBEAM.Context.eval(ctx, "id") + val + end + + assert results == Enum.to_list(1..50) + + for ctx <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "concurrent eval on different contexts" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + + contexts = + for i <- 1..10 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.val = #{i}") + ctx + end + + tasks = + for {ctx, i} <- Enum.with_index(contexts, 1) do + Task.async(fn -> + {:ok, result} = QuickBEAM.Context.eval(ctx, "val * 2") + assert result == i * 2 + result + end) + end + + results = Task.await_many(tasks) + assert results == Enum.map(1..10, &(&1 * 2)) + + for ctx <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "context cleanup on stop" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, 42} = QuickBEAM.Context.eval(ctx, "42") + QuickBEAM.Context.stop(ctx) + + # Pool still works after context is destroyed + {:ok, ctx2} = QuickBEAM.Context.start_link(pool: pool) + assert {:ok, 7} = QuickBEAM.Context.eval(ctx2, "3 + 4") + QuickBEAM.Context.stop(ctx2) + end + + test "multi-thread pool distributes contexts across threads" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + + # Create contexts that will be distributed across 4 threads + contexts = + for i <- 1..20 do + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + {:ok, _} = QuickBEAM.Context.eval(ctx, "globalThis.id = #{i}") + ctx + end + + # All contexts work independently + tasks = + for {ctx, i} <- Enum.with_index(contexts, 1) do + Task.async(fn -> + {:ok, val} = QuickBEAM.Context.eval(ctx, "id") + assert val == i + val + end) + end + + results = Task.await_many(tasks) + assert results == Enum.to_list(1..20) + + for ctx <- contexts, do: QuickBEAM.Context.stop(ctx) + end + + test "browser APIs available in context" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + # URL parsing (browser API backed by Beam handler) + assert {:ok, "example.com"} = + QuickBEAM.Context.eval(ctx, "new URL('https://example.com/path').hostname") + + # crypto.getRandomValues (native Zig) + assert {:ok, 16} = + QuickBEAM.Context.eval(ctx, "crypto.getRandomValues(new Uint8Array(16)).length") + + # performance.now (native Zig) + {:ok, ms} = QuickBEAM.Context.eval(ctx, "performance.now()") + assert is_float(ms) and ms >= 0 + + # console (logs to Logger) + assert {:ok, nil} = QuickBEAM.Context.eval(ctx, "console.log('from context')") + + # setTimeout + assert {:ok, "done"} = + QuickBEAM.Context.eval(ctx, """ + await new Promise(resolve => setTimeout(() => resolve('done'), 10)) + """) + + QuickBEAM.Context.stop(ctx) + end + + test "DOM operations on context" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = + QuickBEAM.Context.eval(ctx, """ + document.body.innerHTML = '

Title

Content

' + """) + + assert {:ok, {"h1", [], ["Title"]}} = QuickBEAM.Context.dom_find(ctx, "h1") + assert {:ok, "Title"} = QuickBEAM.Context.dom_text(ctx, "h1") + {:ok, html} = QuickBEAM.Context.dom_html(ctx) + assert html =~ "

Title

" + + {:ok, items} = QuickBEAM.Context.dom_find_all(ctx, "div.app > *") + assert length(items) == 2 + + QuickBEAM.Context.stop(ctx) + end + + test "send_message to context" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + {:ok, _} = + QuickBEAM.Context.eval(ctx, """ + globalThis.lastMsg = null; + Beam.onMessage((msg) => { globalThis.lastMsg = msg; }); + """) + + QuickBEAM.Context.send_message(ctx, "hello") + Process.sleep(50) + + assert {:ok, "hello"} = QuickBEAM.Context.eval(ctx, "lastMsg") + + QuickBEAM.Context.stop(ctx) + end + + test "Worker on context pool sends message back to parent" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + assert {:ok, "hello from worker"} = + QuickBEAM.Context.eval(ctx, """ + await new Promise((resolve) => { + const w = new Worker(`self.postMessage("hello from worker")`); + w.onmessage = (e) => resolve(e.data); + }) + """) + + QuickBEAM.Context.stop(ctx) + end + + test "Worker on context pool receives message from parent" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + assert {:ok, "pong: ping"} = + QuickBEAM.Context.eval(ctx, """ + await new Promise((resolve) => { + const w = new Worker(` + self.onmessage = (e) => { + self.postMessage("pong: " + e.data); + }; + `); + setTimeout(() => { + w.onmessage = (e) => resolve(e.data); + w.postMessage("ping"); + }, 50); + }) + """) + + QuickBEAM.Context.stop(ctx) + end + + test "multiple Workers on context pool run concurrently" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 4) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + assert {:ok, [1, 2, 3]} = + QuickBEAM.Context.eval( + ctx, + """ + await new Promise((resolve) => { + const results = []; + let count = 0; + for (let i = 1; i <= 3; i++) { + const w = new Worker(`self.postMessage(${i})`); + w.onmessage = (e) => { + results.push(e.data); + count++; + if (count === 3) resolve(results.sort()); + }; + } + }) + """, + timeout: 10_000 + ) + + QuickBEAM.Context.stop(ctx) + end + + test "Worker can be terminated on context pool" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 2) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + assert {:ok, "terminated"} = + QuickBEAM.Context.eval(ctx, """ + const w = new Worker(` + setTimeout(() => self.postMessage("should not arrive"), 500); + `); + w.terminate(); + "terminated" + """) + + QuickBEAM.Context.stop(ctx) + end + + test "get_global and set_global on context" do + {:ok, pool} = QuickBEAM.ContextPool.start_link(size: 1) + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool) + + :ok = QuickBEAM.Context.set_global(ctx, "myVal", 42) + assert {:ok, 42} = QuickBEAM.Context.get_global(ctx, "myVal") + + :ok = QuickBEAM.Context.set_global(ctx, "myObj", %{"a" => 1}) + assert {:ok, %{"a" => 1}} = QuickBEAM.Context.get_global(ctx, "myObj") + + QuickBEAM.Context.stop(ctx) + end + + test "memory_limit rejects large allocations" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool, apis: false, memory_limit: 200_000) + + assert {:error, _} = + QuickBEAM.Context.eval(ctx, "new Array(100000).fill('hello world')") + + QuickBEAM.Context.stop(ctx) + end + + test "max_reductions interrupts long loops" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + + {:ok, ctx} = + QuickBEAM.Context.start_link(pool: pool, apis: false, max_reductions: 100_000) + + assert {:error, %QuickBEAM.JSError{message: "reduction limit exceeded"}} = + QuickBEAM.Context.eval( + ctx, + "(() => { let s = 0; for(let i = 0; i < 10000000; i++) s += i; return s })()" + ) + + QuickBEAM.Context.stop(ctx) + end + + test "context recovers after hitting reduction limit" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + + {:ok, ctx} = + QuickBEAM.Context.start_link(pool: pool, apis: false, max_reductions: 100_000) + + assert {:error, _} = + QuickBEAM.Context.eval( + ctx, + "(() => { let s = 0; for(let i = 0; i < 10000000; i++) s += i; return s })()" + ) + + assert {:ok, 42} = QuickBEAM.Context.eval(ctx, "42") + + QuickBEAM.Context.stop(ctx) + end + + test "memory_usage includes context_malloc_size" do + {:ok, pool} = QuickBEAM.ContextPool.start_link() + {:ok, ctx} = QuickBEAM.Context.start_link(pool: pool, apis: false) + + {:ok, mem} = QuickBEAM.Context.memory_usage(ctx) + assert is_integer(mem.context_malloc_size) + assert mem.context_malloc_size > 0 + + QuickBEAM.Context.stop(ctx) + end +end diff --git a/test/web_apis/locks_test.exs b/test/web_apis/locks_test.exs index c4ec0f444..21d34332e 100644 --- a/test/web_apis/locks_test.exs +++ b/test/web_apis/locks_test.exs @@ -2,7 +2,6 @@ defmodule QuickBEAM.WebAPIs.LocksTest do use ExUnit.Case, async: false setup do - start_supervised!(QuickBEAM.LockManager) {:ok, rt} = QuickBEAM.start() on_exit(fn ->