Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189// Sync-call mailbox over one SharedArrayBuffer per runtime thread.// Layout (DESIGN.md §Sync mailbox):// Int32 header: [0] STATE [1] JSON_LEN [2] BYTES_LEN [3] HEARTBEAT [4] SIGNALS// arena (1 MiB at byte 32): [json][bytes]// SIGNALS is the pending-signal bitmask (bit n = signal number n). The kernel sets a bit// before it replies to the syscall that raised it; the runtime's loop takes the whole// mask at poll-phase entry, so a signal sent to oneself runs in the next poll phase.// Request: runtime writes json (+ optional bytes), STATE=PENDING, doorbell postMessage.// Reply: kernel writes json (+ optional bytes) into the arena, STATE=DONE, Atomics.notify.// The requester sequence-checks STATE before waiting (lost-wakeup fix from C2).//// Bytes-marker protocol: JSON cannot carry typed arrays, so any Uint8Array in a request// or reply is replaced with BYTES_MARKER and the real bytes ride the arena (one slot per// message — sufficient for step 1, where at most one Uint8Array crosses per syscall).import type { FsError } from "./protocol";
export const M = { STATE: 0, JSON_LEN: 1, BYTES_LEN: 2, HEARTBEAT: 3, SIGNALS: 4 } as const;export const ST = { IDLE: 0, PENDING: 1, DONE: 2 } as const;export const ARENA_BYTES = 1 << 20;const HEADER_WORDS = 5;const ARENA_OFFSET = 32;export const SAB_BYTES = ARENA_OFFSET + ARENA_BYTES;export const BYTES_MARKER = "__nib_bytes_marker_7f3a9c__";
const header = (sab: SharedArrayBuffer): Int32Array => new Int32Array(sab, 0, HEADER_WORDS);const arena = (sab: SharedArrayBuffer): Uint8Array => new Uint8Array(sab, ARENA_OFFSET, ARENA_BYTES);
const now = (): number => Date.now() & 0x7fffffff;const enc = new TextEncoder();const dec = new TextDecoder();
export function bumpHeartbeat(sab: SharedArrayBuffer): void { Atomics.store(header(sab), M.HEARTBEAT, now());}
/** Kernel side: mark signal number `n` pending. */export function raiseSignal(sab: SharedArrayBuffer, n: number): void { Atomics.or(header(sab), M.SIGNALS, 1 << n);}
/** Runtime side: take every pending signal number, clearing the mask. */export function takeSignals(sab: SharedArrayBuffer): number[] { const mask = Atomics.exchange(header(sab), M.SIGNALS, 0); const out: number[] = []; for (let n = 1; n < 32; n++) if (mask & (1 << n)) out.push(n); return out;}
export function startHeartbeat(sab: SharedArrayBuffer): () => void { bumpHeartbeat(sab); const iv = setInterval(() => bumpHeartbeat(sab), 500); return () => clearInterval(iv);}
function put(sab: SharedArrayBuffer, json: string, bytes: Uint8Array | null): void { const a = arena(sab); const jsonBytes = enc.encode(json); if (jsonBytes.length + (bytes?.length ?? 0) > ARENA_BYTES) { throw new Error("EOVERFLOW: reply exceeds mailbox arena"); } a.set(jsonBytes, 0); if (bytes) a.set(bytes, jsonBytes.length); const h = header(sab); h[M.JSON_LEN] = jsonBytes.length; // -1 means "no Uint8Array at all", distinct from 0 ("an empty one") — a 0-byte // read or write is real (an empty file, or writing zero bytes) and must round-trip // as an empty Uint8Array, not as the marker string nothing replaced it with. h[M.BYTES_LEN] = bytes ? bytes.length : -1;}
function getJson(sab: SharedArrayBuffer, lenOffset: number): unknown { const a = arena(sab); const len = header(sab)[lenOffset]!; // slice() copies out of the SharedArrayBuffer — TextDecoder.decode() rejects shared views. return JSON.parse(dec.decode(a.slice(0, len)));}
function getBytes(sab: SharedArrayBuffer): Uint8Array | null { const h = header(sab); const jsonLen = h[M.JSON_LEN]!; const bytesLen = h[M.BYTES_LEN]!; if (bytesLen < 0) return null; return arena(sab).slice(jsonLen, jsonLen + bytesLen);}
// ---- marker packing (shared by both directions) ----
/** Replace every Uint8Array in a value with BYTES_MARKER; return the extracted bytes. */export function extractTypedArrays(value: unknown): { value: unknown; bytes: Uint8Array | null } { let found: Uint8Array | null = null; const walk = (v: unknown): unknown => { if (v instanceof Uint8Array) { if (found) throw new Error("EOVERFLOW: multiple Uint8Arrays in one mailbox message"); found = v; return BYTES_MARKER; } if (Array.isArray(v)) return v.map(walk); if (v && typeof v === "object") { const out: Record<string, unknown> = {}; for (const [k, val] of Object.entries(v)) out[k] = walk(val); return out; } return v; }; return { value: walk(value), bytes: found };}
/** Replace BYTES_MARKER strings with the arena bytes. Inverse of extractTypedArrays. */export function injectTypedArrays(value: unknown, bytes: Uint8Array | null): unknown { if (bytes === null) return value; const walk = (v: unknown): unknown => { if (v === BYTES_MARKER) return bytes; if (Array.isArray(v)) return v.map(walk); if (v && typeof v === "object") { const out: Record<string, unknown> = {}; for (const [k, val] of Object.entries(v)) out[k] = walk(val); return out; } return v; }; return walk(value);}
// ---- runtime side ----
export type SyncReply<T> = { ok: true; val: T } | { ok: false; err: FsError };
/** Write a sync request into the mailbox. Caller must not have one in flight (single thread). */export function writeSyncRequest(sab: SharedArrayBuffer, op: string, args: unknown): void { const h = header(sab); if (Atomics.load(h, M.STATE) === ST.PENDING) throw new Error("EAGAIN: sync call already in flight"); const { value, bytes } = extractTypedArrays(args); put(sab, JSON.stringify({ op, args: value }), bytes); Atomics.store(h, M.STATE, ST.PENDING);}
/** Block until the kernel writes the reply. Sequence-checks state before each wait. */export function awaitSyncReply<T>(sab: SharedArrayBuffer): SyncReply<T> { const h = header(sab); let lastBeat = Atomics.load(h, M.HEARTBEAT); // Stalled means no beat for 10 s while we wait. Timing from our own clock, not the // beat's value, keeps a call made before the kernel's first beat (the mailbox is // still 0 until it handles `attach`) from reading as a stall. let lastBeatSeen = now(); for (;;) { const state = Atomics.load(h, M.STATE); if (state === ST.DONE) break; if (state === ST.PENDING) { const beat = Atomics.load(h, M.HEARTBEAT); if (beat !== lastBeat) { lastBeat = beat; lastBeatSeen = now(); } else if (now() - lastBeatSeen > 10000) throw new Error("ERR_NIB_KERNEL_LOST: kernel heartbeat stalled"); Atomics.wait(h, M.STATE, ST.PENDING, 2000); continue; } // state === IDLE: reply cannot be ready yet, nothing was written. Atomics.wait(h, M.STATE, ST.IDLE, 2000); } const bytes = getBytes(sab); const raw = getJson(sab, M.JSON_LEN) as SyncReply<T>; Atomics.store(h, M.STATE, ST.IDLE); if (!raw.ok) return raw; return { ok: true, val: injectTypedArrays(raw.val, bytes) as T };}
// ---- kernel side ----
export function readSyncRequest(sab: SharedArrayBuffer): { op: string; args: unknown } { const bytes = getBytes(sab); const req = getJson(sab, M.JSON_LEN) as { op: string; args: unknown }; return { op: req.op, args: injectTypedArrays(req.args, bytes) };}
/** Kernel side: write the reply and wake the waiter. */export function writeSyncReply(sab: SharedArrayBuffer, ok: boolean, val: unknown, err: FsError | undefined): void { let bytes: Uint8Array | null = null; let payload = val; if (ok) { const packed = extractTypedArrays(val); payload = packed.value; bytes = packed.bytes; } put(sab, JSON.stringify({ ok, val: payload, err }), bytes); const h = header(sab); Atomics.store(h, M.STATE, ST.DONE); Atomics.notify(h, M.STATE);}