AI_Slop/calog
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
calog/src/context.c
Scott Duensing 080bc95d8e Some security decisions revised. Bugs fixed.
2026-10-15 17:45:12 -05:00

1997 lines
82 KiB
C
Raw Blame History

// context.c -- actor threading: one thread + one MPSC queue per context, plus an
// implicit "host context" (id 0) driven by the host thread via calogPump.
//
// Model (design.md sec 6):
// - Scripts run fire-and-forget on their own context threads (calogContextEval
// enqueues and returns).
// - A registered native runs on the HOST thread: a script calling one parks and
// posts a CALL onto the host queue, which calogPump services on the host thread.
// So native C code is serialized on one thread and needs no locking. An inline
// native (calogRegisterInline) instead runs on the calling thread.
// - A script function value (CalogFnT) runs on its owning engine's thread (sec 10).
// - A script error is posted to the host queue and delivered to the error handler
// during calogPump.
//
// All per-runtime actor state (the context registry, the host context, the routing
// hooks, the error sink) lives in CalogT, so independent runtimes coexist in one
// process. The dispatch reaches its runtime through the object it already holds -- the
// serving context's ->broker, a callable's ->runtime, or an explicit calog argument --
// never a process global. currentContext (a thread-local) names the calling thread's
// context; calogPump sets it to the pumped runtime's host for the drain, so ONE thread
// can host several runtimes by pumping each in turn. Because context ids number from 1
// per runtime, inline/reply decisions match the runtime too, not the id alone. A value
// or callable is still not shared across runtimes (a cross-runtime reply cannot route).
//
// Everything a thread waits on flows through its own queue; a caller that is itself
// a context pumps its own queue (servicing re-entrant calls) rather than sleeping.
// Ownership across the queue: a CALL deep-copies its args; the target frees them and
// MOVES the result back. No heap pointer is shared between threads at rest.
#define _POSIX_C_SOURCE 200809L
#include "calogInternal.h"
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#define CONTEXT_INITIAL_CONTEXTS 8
#define CONTEXT_INITIAL_ENGINES 4
// The host context id (script contexts get 1..N). currentContext == calog->hostContext
// means "on this runtime's host thread."
// CALOG_HOST_ID is defined in calogInternal.h (shared with value.c's finalize path).
// A script context id packs a 1-based slot index in the low CONTEXT_INDEX_BITS and a
// generation counter above it. When a slot is recycled its generation is bumped, so a
// handle minted against the old occupant no longer resolves (design.md sec 9). The id
// is 64-bit -- a 32-bit slot index + 32-bit generation -- so neither the live-context
// count nor open/close churn hits a preset ceiling in practice.
#define CONTEXT_INDEX_BITS 32
#define CONTEXT_INDEX_MASK 0xFFFFFFFFu
typedef enum MessageKindE {
messageCallE = 0,
messageReplyE = 1,
messageEvalE = 2,
messageReleaseE = 3,
messageErrorE = 4
} MessageKindE;
typedef struct ReplyBoxT {
pthread_mutex_t mutex;
pthread_cond_t cond;
bool done;
int32_t status;
CalogValueT result;
} ReplyBoxT;
typedef struct MessageT {
MessageKindE kind;
CalogNativeFnT fn; // messageCallE: the native to run
void *userData; // messageCallE: its userData
CalogValueT *args; // messageCallE: deep-copied arguments
int32_t argCount;
char *source; // messageEvalE: script source; messageErrorE: error text
CalogFnT *callable; // messageReleaseE: the callable to finalize
uint64_t replyToId; // messageErrorE: the failing context id
ReplyBoxT *replyBox;
uint64_t token;
int32_t status;
CalogValueT result;
struct MessageT *next;
} MessageT;
// One sandbox policy, shared by a context and every task it spawns. Sharing is the whole design: a
// spawned child does not get a fresh budget, it draws on the SAME memory pool, dies at the SAME
// absolute deadline, and is bound by the SAME allow-list, so a limited script cannot escape its
// policy by doing its work in a child. `live` is both the reference count and the number of contexts
// in the tree, which is what maxContexts bounds -- without it, inheriting the other three limits
// would still leave `while true do taskSpawn(...) end` to exhaust the host's threads.
typedef struct SandboxT {
CalogLimitStateT state; // what the engine adapters install their allocator/hook against
char **allow; // sorted allowed-native names, or NULL = every native permitted
int32_t allowCount;
int32_t maxContexts; // 0 = unbounded
bool metered; // a memory cap or a deadline is set, so adapters need `state`
_Atomic int32_t live; // contexts sharing this policy; the last one out frees it
} SandboxT;
struct CalogContextT {
uint64_t id;
CalogT *broker;
const CalogEngineT *engine;
void *interp;
pthread_t thread;
pthread_mutex_t queueMutex;
pthread_cond_t queueCond;
MessageT *head;
MessageT *tail;
MessageT *stash;
uint64_t nextToken;
_Atomic bool shuttingDown; // set by contextRequestShutdown; polled lock-free by calogCurrentShuttingDown
_Atomic bool retireRequested; // set by calogCurrentRetire (taskExit); serveLoop retires AFTER the current eval
_Atomic bool finished; // set by threadMain just before the thread returns; lets any thread safely reap it
bool started;
bool interpDead; // set (under ctxMutex) once the interpreter is torn down
SandboxT *sandbox; // the policy this context runs under, SHARED with any task it
// spawns, or NULL when the context is unrestricted
_Atomic bool abortRaised; // the abort error was raised INTO this VM (see calogAbortRaised)
bool closing; // guarded by broker->ctxMutex: a close already owns this context
bool queueClosed; // guarded by queueMutex: serveLoop has stopped serving, refuse new messages
bool reclaimDone; // guarded by queueMutex: the callable sweep has run, stop adopting
bool callsClosed; // guarded by queueMutex: refuse CALLs only (errors still land)
// Every callable this context created, so it can reclaim their engine handles before its
// interpreter is destroyed (contextReclaimCallables). Guarded by queueMutex -- a finalize can
// reach it from a foreign thread on the best-effort path in actorReleaseCallable.
CalogFnT **ownedFns;
int64_t ownedCount;
int64_t ownedCap;
};
// currentContext is the ONLY process-global actor state -- and it is thread-local: it
// names the context (host or script) the calling thread belongs to. A foreign thread
// (never a calog thread) leaves it NULL and reaches its runtime through a passed
// handle instead.
static _Thread_local CalogContextT *currentContext = NULL;
// Cross-boundary trace: a thread-local stack of the cross-context callable hops the CALLING thread
// is currently inside (each entry = the context a callable is owned by). tracePush/Pop bracket every
// actorInvokeCallable; calogLastTrace formats it for live introspection. Post-mortem, the error
// STRING itself carries the chain -- traceEnrich prepends the owner's "[engine ctx N]" tag as an
// error unwinds out of each boundary, so the tag accumulates across the reply and reaches onError.
#define CALOG_TRACE_MAX 32
typedef struct TraceFrameT {
CalogT *runtime;
uint64_t ctxId;
} TraceFrameT;
static _Thread_local TraceFrameT traceFrames[CALOG_TRACE_MAX];
static _Thread_local int32_t traceDepth = 0;
static int32_t actorInvokeCallable(CalogFnT *callable, CalogValueT *args, int32_t argCount, CalogValueT *result);
static void actorReleaseCallable(CalogFnT *callable);
static int32_t actorRoute(CalogT *calog, CalogEntryT *entry, CalogValueT *args, int32_t argCount, CalogValueT *result);
static int32_t contextDispatch(CalogT *calog, uint64_t targetId, CalogNativeFnT fn, void *userData, CalogValueT *args, int32_t argCount, CalogValueT *result);
static void contextDispatchCall(CalogContextT *context, MessageT *message);
static void contextDispatchError(CalogT *calog, MessageT *message);
static void contextDispatchEval(CalogContextT *context, MessageT *message);
static void contextDispatchRelease(MessageT *message);
static void contextDrainQueue(CalogContextT *context);
static CalogContextT *contextOpenWithSandbox(CalogT *broker, const CalogEngineT *engine, SandboxT *sandbox);
static SandboxT *sandboxCreate(const CalogLimitsT *limits);
static void sandboxRelease(SandboxT *sandbox);
static SandboxT *sandboxRetainForChild(SandboxT *sandbox);
static int32_t contextEnqueue(CalogT *calog, uint64_t targetId, MessageT *message);
static CalogContextT *contextAtIndex(CalogT *calog, int64_t index);
static int32_t contextPostRelease(CalogT *calog, uint64_t targetId, CalogFnT *callable);
static void contextReclaimCallables(CalogContextT *context);
static void contextReply(CalogT *calog, MessageT *message, int32_t status, CalogValueT *result);
static void contextRequestShutdown(CalogContextT *context);
static int32_t contextSendBlocking(CalogT *calog, uint64_t targetId, MessageT *message, CalogValueT *result);
static bool dispatchCommon(CalogContextT *context, MessageT *message);
static bool enqueueRaw(CalogContextT *context, MessageT *message);
static int32_t hookGrow(void **array, int64_t count, int64_t *cap, size_t elemSize);
static void hostCallsClose(CalogT *calog);
static void hostDispatch(CalogT *calog, MessageT *message);
static uint64_t idCompose(int64_t index, uint32_t generation);
static uint32_t idGeneration(uint64_t id);
static int64_t idIndex(uint64_t id);
static int limitNameCompare(const void *a, const void *b);
static MessageT *messageDequeue(CalogContextT *context);
static void messageFree(MessageT *message);
static bool onOwnerThread(CalogT *runtime, uint64_t owner);
static int32_t pumpUntil(CalogContextT *context, uint64_t token, int32_t *outStatus, CalogValueT *result);
static void registryFreePush(CalogT *calog, int64_t index);
static CalogContextT *registryResolveLocked(CalogT *calog, uint64_t id);
static void serveLoop(CalogContextT *context);
static void traceEnrich(CalogT *runtime, uint64_t ctxId, CalogValueT *result);
static const char *traceEngineName(CalogT *runtime, uint64_t ctxId);
static void tracePop(void);
static void tracePush(CalogT *runtime, uint64_t ctxId);
static MessageT *tryDequeue(CalogContextT *context);
static void *threadMain(void *arg);
int32_t calogActorInit(CalogT *calog) {
// Build this runtime's host context (id 0, no thread) and claim the calling thread
// as its host. All actor state hangs off CalogT, so multiple runtimes coexist.
calog->hostContext = (CalogContextT *)calloc(1, sizeof(*calog->hostContext));
if (calog->hostContext == NULL) {
return calogErrOomE;
}
calog->hostContext->id = CALOG_HOST_ID;
calog->hostContext->broker = calog;
pthread_mutex_init(&calog->hostContext->queueMutex, NULL);
pthread_cond_init(&calog->hostContext->queueCond, NULL);
calog->hostThread = pthread_self();
currentContext = calog->hostContext;
pthread_mutex_init(&calog->ctxMutex, NULL);
calog->routeHook = actorRoute;
calog->invokeHook = actorInvokeCallable;
calog->releaseHook = actorReleaseCallable;
return calogOkE;
}
// True when the calling thread is the owner context's own thread. Ids alone are
// ambiguous across runtimes (each numbers contexts from 1), so the runtime must match
// too; a foreign thread (currentContext NULL) is never the owner.
static bool onOwnerThread(CalogT *runtime, uint64_t owner) {
return currentContext != NULL && currentContext->broker == runtime && currentContext->id == owner;
}
static void tracePush(CalogT *runtime, uint64_t ctxId) {
if (traceDepth >= 0 && traceDepth < CALOG_TRACE_MAX) {
traceFrames[traceDepth].runtime = runtime;
traceFrames[traceDepth].ctxId = ctxId;
}
traceDepth++; // still counts past the array so the matching pop stays balanced
}
static void tracePop(void) {
if (traceDepth > 0) {
traceDepth--;
}
}
static const char *traceEngineName(CalogT *runtime, uint64_t ctxId) {
CalogContextT *context;
const char *name;
if (ctxId == CALOG_HOST_ID) {
return "host";
}
name = "?";
pthread_mutex_lock(&runtime->ctxMutex);
context = registryResolveLocked(runtime, ctxId);
if (context != NULL && context->engine != NULL) {
name = context->engine->name;
}
pthread_mutex_unlock(&runtime->ctxMutex);
return name;
}
// Prepend "[engine ctx N] " to an error result as it unwinds out of a cross-context callable, so
// the chain accumulates in the string that crosses the reply back to the caller (and to onError).
static void traceEnrich(CalogT *runtime, uint64_t ctxId, CalogValueT *result) {
char prefix[96];
char *combined;
const char *engine;
size_t prefixLen;
size_t oldLen;
if (result->type != calogStringE) {
return;
}
engine = traceEngineName(runtime, ctxId);
prefixLen = (size_t)snprintf(prefix, sizeof(prefix), "[%s ctx %lld] ", engine, (long long)idIndex(ctxId));
if (prefixLen >= sizeof(prefix)) {
prefixLen = sizeof(prefix) - 1;
}
oldLen = (size_t)result->as.s.length;
combined = (char *)malloc(prefixLen + oldLen + 1);
if (combined == NULL) {
return;
}
memcpy(combined, prefix, prefixLen);
memcpy(combined + prefixLen, result->as.s.bytes, oldLen);
combined[prefixLen + oldLen] = '\0';
calogValueFree(result);
calogValueString(result, combined, (int64_t)(prefixLen + oldLen));
free(combined);
}
static int32_t actorInvokeCallable(CalogFnT *callable, CalogValueT *args, int32_t argCount, CalogValueT *result) {
CalogNativeFnT fn;
CalogT *runtime;
uint64_t owner;
int32_t status;
// Inline when already on the owner's thread; otherwise marshal to the owner (in the callable's
// own runtime). A callable's fn + userData are exactly a native call, so the CALL machinery
// carries it unchanged. The boundary hop is traced so a cross-context failure reads clearly.
runtime = calogFnRuntime(callable);
owner = calogFnOwner(callable);
tracePush(runtime, owner);
if (onOwnerThread(runtime, owner)) {
fn = calogFnNative(callable);
status = fn(args, argCount, result, calogFnUserData(callable));
} else {
status = contextDispatch(runtime, owner, calogFnNative(callable), calogFnUserData(callable), args, argCount, result);
}
tracePop();
if (status != calogOkE) {
traceEnrich(runtime, owner, result);
}
return status;
}
static void actorReleaseCallable(CalogFnT *callable) {
CalogT *runtime;
uint64_t owner;
// Finalize inline when already on the owner's thread; otherwise marshal the
// finalize to the owner so the engine release (luaL_unref / sq_release) runs there.
runtime = calogFnRuntime(callable);
owner = calogFnOwner(callable);
if (onOwnerThread(runtime, owner)) {
calogFnFinalize(callable);
return;
}
if (contextPostRelease(runtime, owner, callable) != calogOkE) {
// The release could not be marshalled: the owner is unreachable (a quiescence violation --
// design.md sec 9), or the message itself would not allocate while the owner is very much
// alive. We are NOT on the owner's thread either way, so finalizing normally could run the
// adapter's engine op (luaL_unref, JS_FreeValue, ...) against a live interpreter from the
// wrong thread. Free the memory and deliberately leak the engine handle instead: it dies
// with its interpreter, and a leak beats corrupting a VM that is still running.
calogFnFinalizeForeign(callable);
}
}
static int32_t actorRoute(CalogT *calog, CalogEntryT *entry, CalogValueT *args, int32_t argCount, CalogValueT *result) {
// Sandboxing: a limited context (allow != NULL -- only a limited SCRIPT context ever sets it;
// the host context never does) may call only the natives on its allow-list. currentContext is
// the CALLING context, so this gates every engine uniformly through the one dispatch choke point.
if (currentContext != NULL && currentContext->sandbox != NULL && currentContext->sandbox->allow != NULL &&
bsearch(&entry->name, currentContext->sandbox->allow, (size_t)currentContext->sandbox->allowCount, sizeof(char *), limitNameCompare) == NULL) {
calogValueNil(result);
return calogFail(result, calogErrUnsupportedE, "native not permitted by this context's allow-list");
}
// An inline native, or any call already on this runtime's host thread, runs inline;
// otherwise the native is marshalled to the host thread (serviced by calogPump).
// currentContext alone is not enough: it is a single per-thread slot, so a thread
// hosting several runtimes in turn (calogCreate, calogCreate) only matches the most
// recently created one. pthread_equal against the physical host thread recorded at
// calogActorInit catches the older runtime too -- nobody else will ever pump its
// host queue, so marshalling there would deadlock forever.
if (entry->runInline || currentContext == calog->hostContext || pthread_equal(pthread_self(), calog->hostThread)) {
return entry->fn(args, argCount, result, entry->userData);
}
return contextDispatch(calog, CALOG_HOST_ID, entry->fn, entry->userData, args, argCount, result);
}
void calogActorShutdown(CalogT *calog) {
// How long to park between checks while waiting out a close that is already in flight (0.5 ms,
// the same tick the host pump uses). It is a rendezvous that almost never happens.
struct timespec closeWait = { 0, 500000 };
int64_t count;
int64_t index;
// Latch the registry closed and read its extent in the same breath. Until now a script was free
// to keep opening contexts (taskSpawn, taskLoad); from here calogContextOpen refuses, so the
// slots below are the complete and final set, ctxSlots cannot move under a realloc, and no
// thread can be started that these loops would miss and then free out from under itself.
pthread_mutex_lock(&calog->ctxMutex);
calog->tearingDown = true;
count = calog->ctxCount;
pthread_mutex_unlock(&calog->ctxMutex);
// A close that was already running when the latch went up (a script's taskClose) owns that
// context's stop-join-free and unlinks its slot when it is done. Wait those out rather than race
// them -- joining one thread from two is undefined, and freeing a context another thread is
// still inside is worse. The latch refuses any new one, so this settles and cannot grow.
for (;;) {
bool pending;
pending = false;
pthread_mutex_lock(&calog->ctxMutex);
for (index = 0; index < count; index++) {
if (calog->ctxSlots[index].context != NULL && calog->ctxSlots[index].context->closing) {
pending = true;
break;
}
}
pthread_mutex_unlock(&calog->ctxMutex);
if (!pending) {
break;
}
nanosleep(&closeWait, NULL);
}
for (index = 0; index < count; index++) {
CalogContextT *context;
context = contextAtIndex(calog, index);
if (context == NULL) {
continue;
}
contextRequestShutdown(context);
}
// Join EVERY thread before freeing ANY context: a context tearing down may release a
// callable owned by a sibling (marshalled across contexts), which resolves/posts to that
// sibling -- so no sibling may be freed while another's thread still runs.
for (index = 0; index < count; index++) {
CalogContextT *context;
context = contextAtIndex(calog, index);
if (context != NULL) {
pthread_join(context->thread, NULL);
}
}
// All context threads are stopped. Unregister each under ctxMutex (so any in-flight
// registryResolveLocked from the host sees NULL rather than a freed slot), drain any
// orphaned release, then free.
for (index = 0; index < count; index++) {
CalogContextT *context;
pthread_mutex_lock(&calog->ctxMutex);
context = calog->ctxSlots[index].context;
calog->ctxSlots[index].context = NULL;
pthread_mutex_unlock(&calog->ctxMutex);
if (context == NULL) {
continue;
}
contextDrainQueue(context);
sandboxRelease(context->sandbox);
pthread_mutex_destroy(&context->queueMutex);
pthread_cond_destroy(&context->queueCond);
free(context);
}
free(calog->ctxSlots);
free(calog->ctxFree);
calog->ctxSlots = NULL;
calog->ctxFree = NULL;
calog->ctxCount = 0;
calog->ctxCap = 0;
calog->ctxFreeCount = 0;
calog->ctxFreeCap = 0;
// Drain and free the host context. Only clear the calling thread's currentContext
// if it named THIS runtime's host -- the thread may still host other runtimes.
if (calog->hostContext != NULL) {
bool wasHost;
wasHost = (currentContext == calog->hostContext);
// contextDrainQueue (not a bare messageFree loop) finalizes any orphaned
// messageReleaseE so a pending callable is not leaked.
contextDrainQueue(calog->hostContext);
pthread_mutex_destroy(&calog->hostContext->queueMutex);
pthread_cond_destroy(&calog->hostContext->queueCond);
free(calog->hostContext);
calog->hostContext = NULL;
if (wasHost) {
currentContext = NULL;
}
}
calog->routeHook = NULL;
calog->invokeHook = NULL;
calog->releaseHook = NULL;
calog->errorHandler = NULL;
calog->errorUserData = NULL;
// ctxMutex is deliberately NOT destroyed here. Finalizing a callable locks it -- to untrack the
// callable and to resolve its owner -- and callables are still being finalized after this
// returns: the after-context destroy hooks (pubsub, export, kv, task) release everything their
// registries hold. calogDestroy destroys it once those have run. Every slot is gone by now, so
// those late locks find nothing, which is the answer they want.
}
CalogT *calogContextBroker(const CalogContextT *context) {
return context->broker;
}
// The public runtime lifecycle: calogCreate composes the registry with the actor
// layer (which builds the host context and installs the routing hooks); calogDestroy
// tears the actor layer down (joining context threads) before freeing the registry.
CalogT *calogCreate(void) {
CalogT *calog;
calog = calogBrokerCreate();
if (calog == NULL) {
return NULL;
}
if (calogActorInit(calog) != calogOkE) {
calogBrokerDestroy(calog);
return NULL;
}
return calog;
}
#define CALOG_HOOK_INITIAL_CAP 4
static int32_t hookGrow(void **array, int64_t count, int64_t *cap, size_t elemSize) {
void *grown;
int64_t newCap;
if (count < *cap) {
return calogOkE;
}
newCap = (*cap == 0) ? CALOG_HOOK_INITIAL_CAP : *cap * CALOG_GROWTH_FACTOR;
grown = realloc(*array, (size_t)newCap * elemSize);
if (grown == NULL) {
return calogErrOomE;
}
*array = grown;
*cap = newCap;
return calogOkE;
}
int32_t calogAtContext(CalogT *calog, CalogContextHookFnT init, CalogContextHookFnT shutdown, void *userData) {
int32_t status;
if (init == NULL && shutdown == NULL) {
return calogOkE;
}
status = hookGrow((void **)&calog->contextHooks, calog->contextHookCount, &calog->contextHookCap, sizeof(*calog->contextHooks));
if (status != calogOkE) {
return status;
}
calog->contextHooks[calog->contextHookCount].init = init;
calog->contextHooks[calog->contextHookCount].shutdown = shutdown;
calog->contextHooks[calog->contextHookCount].userData = userData;
calog->contextHookCount++;
return calogOkE;
}
int32_t calogAtDestroy(CalogT *calog, CalogDestroyHookFnT fn, CalogDestroyPhaseE phase) {
int32_t status;
if (fn == NULL) {
return calogOkE;
}
status = hookGrow((void **)&calog->destroyHooks, calog->destroyHookCount, &calog->destroyHookCap, sizeof(*calog->destroyHooks));
if (status != calogOkE) {
return status;
}
calog->destroyHooks[calog->destroyHookCount].fn = fn;
calog->destroyHooks[calog->destroyHookCount].phase = phase;
calog->destroyHookCount++;
return calogOkE;
}
void calogDestroy(CalogT *calog) {
int64_t index;
if (calog == NULL) {
return;
}
// FIRST, before any hook runs: see hostCallsClose. The hooks below join background threads, and
// a thread waiting on a host reply that can no longer come would never be joinable.
hostCallsClose(calog);
// Before-context destroy hooks: a library with a background thread that invokes context
// callbacks stops it here, while those contexts are still alive. Reverse registration order.
for (index = calog->destroyHookCount - 1; index >= 0; index--) {
if (calog->destroyHooks[index].phase == calogDestroyBeforeContextsE) {
calog->destroyHooks[index].fn();
}
}
calogActorShutdown(calog);
// After-context destroy hooks: free process-global state now that no context thread survives.
for (index = calog->destroyHookCount - 1; index >= 0; index--) {
if (calog->destroyHooks[index].phase == calogDestroyAfterContextsE) {
calog->destroyHooks[index].fn();
}
}
free(calog->destroyHooks);
free(calog->contextHooks);
// Only now is ctxMutex dead. The after-context hooks above release the callables their registries
// hold, and finalizing one locks this mutex (calogContextUntrackFn, and calogContextRegistered to
// decide whether the engine release may run) -- so destroying it inside calogActorShutdown, as
// this used to, left those hooks locking a destroyed mutex. ThreadSanitizer calls that "use of an
// invalid mutex"; the C standard calls it undefined behaviour.
pthread_mutex_destroy(&calog->ctxMutex);
calogBrokerDestroy(calog);
}
// Create + start a context in one step (design: no owned-native registration happens
// between the two, so there is nothing to do in between). Returns NULL on failure.
uint64_t calogMonotonicMillis(void) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000u + (uint64_t)(ts.tv_nsec / 1000000);
}
CalogLimitStateT *calogContextLimitState(CalogContextT *context) {
if (context->sandbox == NULL || !context->sandbox->metered) {
return NULL;
}
return &context->sandbox->state;
}
bool calogContextSandboxed(CalogContextT *context) {
return context->sandbox != NULL;
}
static int limitNameCompare(const void *a, const void *b) {
return strcmp(*(const char *const *)a, *(const char *const *)b);
}
// Build the policy for a freshly limited context. Returns NULL for "no limits at all" (which is not
// a failure) and leaves *failed true only if an allocation failed. The deadline is resolved to an
// ABSOLUTE instant here, once: every context in the tree then shares that instant, so a chain of
// spawns cannot walk the budget forward by restarting it.
static SandboxT *sandboxCreate(const CalogLimitsT *limits) {
SandboxT *sandbox;
if (limits == NULL) {
return NULL;
}
if (limits->memoryBytes <= 0 && limits->wallClockMillis <= 0 && limits->allowList == NULL && limits->maxContexts <= 0) {
return NULL;
}
sandbox = (SandboxT *)calloc(1, sizeof(*sandbox));
if (sandbox == NULL) {
return NULL;
}
atomic_store(&sandbox->live, 1);
sandbox->maxContexts = limits->maxContexts > 0 ? limits->maxContexts : 0;
if (limits->memoryBytes > 0) {
sandbox->state.memCap = limits->memoryBytes;
sandbox->metered = true;
}
if (limits->wallClockMillis > 0) {
sandbox->state.deadlineMs = calogMonotonicMillis() + (uint64_t)limits->wallClockMillis;
sandbox->metered = true;
}
if (limits->allowList != NULL) {
int32_t count;
int32_t i;
count = 0;
while (limits->allowList[count] != NULL) {
count++;
}
sandbox->allow = (char **)calloc((size_t)(count > 0 ? count : 1), sizeof(char *));
if (sandbox->allow == NULL) {
free(sandbox);
return NULL;
}
for (i = 0; i < count; i++) {
sandbox->allow[i] = strdup(limits->allowList[i]);
if (sandbox->allow[i] == NULL) {
sandbox->allowCount = i;
sandboxRelease(sandbox);
return NULL;
}
}
sandbox->allowCount = count;
qsort(sandbox->allow, (size_t)count, sizeof(char *), limitNameCompare);
}
return sandbox;
}
// Drop one context's reference. The policy outlives any single context in its tree -- a parent that
// finishes first must not pull the budget out from under the children it spawned -- so only the last
// one out frees it. NULL (an unrestricted context) is a no-op.
static void sandboxRelease(SandboxT *sandbox) {
int32_t i;
if (sandbox == NULL) {
return;
}
if (atomic_fetch_sub(&sandbox->live, 1) != 1) {
return;
}
for (i = 0; i < sandbox->allowCount; i++) {
free(sandbox->allow[i]);
}
free(sandbox->allow);
free(sandbox);
}
// Take a reference for a context about to be spawned INTO this policy, refusing if that would push
// the tree past maxContexts. The CAS loop is what makes the bound real: two scripts spawning at once
// must not both read live == max-1 and both succeed.
static SandboxT *sandboxRetainForChild(SandboxT *sandbox) {
int32_t current;
if (sandbox == NULL) {
return NULL;
}
current = atomic_load(&sandbox->live);
for (;;) {
if (sandbox->maxContexts > 0 && current >= sandbox->maxContexts) {
return NULL;
}
if (atomic_compare_exchange_weak(&sandbox->live, &current, current + 1)) {
return sandbox;
}
}
}
CalogContextT *calogContextOpen(CalogT *broker, const CalogEngineT *engine) {
return calogContextOpenLimited(broker, engine, NULL);
}
// Open a context already bound to `sandbox`, taking ownership of the reference the caller holds (so
// a failure here releases it). Both public entry points funnel through this: the difference between
// them is only WHERE the policy came from -- freshly built by the host, or inherited from the script
// that asked for the spawn.
static CalogContextT *contextOpenWithSandbox(CalogT *broker, const CalogEngineT *engine, SandboxT *sandbox) {
CalogContextT *context;
int64_t index;
uint32_t generation;
// A limit the engine would silently ignore is refused instead: a limited open, or a sandboxed
// script's taskSpawn, would otherwise hand back a context free of the caps it was asked for.
if (engine != NULL && engine->ignoresLimits && sandbox != NULL && sandbox->metered) {
sandboxRelease(sandbox);
return NULL;
}
context = (CalogContextT *)calloc(1, sizeof(*context));
if (context == NULL) {
sandboxRelease(sandbox);
return NULL;
}
context->broker = broker;
context->engine = engine;
// Bound BEFORE the thread starts: createInterpreter reads the limit state to install its
// allocator/hook and runs on the new thread, which pthread_create orders after this write.
context->sandbox = sandbox;
pthread_mutex_init(&context->queueMutex, NULL);
pthread_cond_init(&context->queueCond, NULL);
pthread_mutex_lock(&broker->ctxMutex);
// The runtime is being destroyed: refuse rather than join a registry that is already being
// walked and freed (a taskSpawn racing calogDestroy would otherwise leave a thread running
// past the teardown that never saw it).
if (broker->tearingDown) {
pthread_mutex_unlock(&broker->ctxMutex);
goto fail;
}
if (broker->ctxFreeCount > 0) {
broker->ctxFreeCount--;
index = broker->ctxFree[broker->ctxFreeCount];
generation = broker->ctxSlots[index].generation + 1u;
} else {
if (broker->ctxCount + 1 > (int64_t)CONTEXT_INDEX_MASK) {
pthread_mutex_unlock(&broker->ctxMutex);
goto fail;
}
if (broker->ctxCount == broker->ctxCap) {
int64_t newCap;
CalogRegistrySlotT *grown;
newCap = (broker->ctxCap == 0) ? CONTEXT_INITIAL_CONTEXTS : broker->ctxCap * CALOG_GROWTH_FACTOR;
grown = (CalogRegistrySlotT *)realloc(broker->ctxSlots, (size_t)newCap * sizeof(CalogRegistrySlotT));
if (grown == NULL) {
pthread_mutex_unlock(&broker->ctxMutex);
goto fail;
}
broker->ctxSlots = grown;
broker->ctxCap = newCap;
}
index = broker->ctxCount;
generation = 0;
broker->ctxCount++;
}
context->id = idCompose(index, generation);
broker->ctxSlots[index].context = context;
broker->ctxSlots[index].generation = generation;
// Start the thread and publish `started` INSIDE the registry lock. A teardown reads both under
// that same lock, so it can never catch this context half-registered -- either the slot is still
// empty, or it is filled and the thread it names is running. The new thread may need the lock
// itself (its first callable, a context init hook); it simply waits the moment or two until the
// unlock below, and nothing here waits on the new thread, so the two cannot deadlock.
if (pthread_create(&context->thread, NULL, threadMain, context) != 0) {
broker->ctxSlots[index].context = NULL;
registryFreePush(broker, index);
pthread_mutex_unlock(&broker->ctxMutex);
goto fail;
}
context->started = true;
pthread_mutex_unlock(&broker->ctxMutex);
return context;
fail:
// Shared teardown for every failure above: nothing beyond the queue's own mutex/cond, this
// context's reference on the policy, and the context shell has been allocated at any of these
// points.
sandboxRelease(context->sandbox);
pthread_mutex_destroy(&context->queueMutex);
pthread_cond_destroy(&context->queueCond);
free(context);
return NULL;
}
CalogContextT *calogContextOpenLimited(CalogT *broker, const CalogEngineT *engine, const CalogLimitsT *limits) {
SandboxT *sandbox;
// A policy that asks for nothing is no policy: sandboxCreate returns NULL for it, which is
// exactly what an unrestricted context wants. Telling that apart from an allocation failure is
// why the emptiness test lives there rather than here.
sandbox = sandboxCreate(limits);
if (sandbox == NULL && limits != NULL && (limits->memoryBytes > 0 || limits->wallClockMillis > 0 ||
limits->allowList != NULL || limits->maxContexts > 0)) {
return NULL;
}
return contextOpenWithSandbox(broker, engine, sandbox);
}
// Open a context under the CALLING script's policy -- what taskSpawn/taskLoad use. An unrestricted
// caller yields an unrestricted child (the ordinary case, and free). A sandboxed caller yields a
// child sharing its memory pool, its deadline and its allow-list, and the spawn is REFUSED once the
// tree is at maxContexts. Returns NULL on refusal, which the task natives report as an error rather
// than silently minting the unlimited context this used to hand out.
CalogContextT *calogContextOpenInheriting(CalogT *broker, const CalogEngineT *engine) {
SandboxT *sandbox;
if (currentContext == NULL || currentContext->sandbox == NULL) {
return contextOpenWithSandbox(broker, engine, NULL);
}
sandbox = sandboxRetainForChild(currentContext->sandbox);
if (sandbox == NULL) {
return NULL;
}
return contextOpenWithSandbox(broker, engine, sandbox);
}
// Search the registered engines for a file "<baseFileName>.<ext>". The first engine
// (in registration order), then its first extension, that names a readable file wins:
// its contents are loaded fire-and-forget into a fresh context on that engine. Reads
// the file on the calling thread. Returns NULL if nothing matched or the load failed.
// Body of both load entry points. `inherit` picks which open to use, so the file search itself has
// one implementation rather than two that can drift.
static CalogContextT *contextLoadCommon(CalogT *calog, const char *baseFileName, bool inherit) {
int64_t engineIndex;
for (engineIndex = 0; engineIndex < calog->engineCount; engineIndex++) {
const CalogEngineT *engine;
int32_t extIndex;
engine = calog->engines[engineIndex];
if (engine->extensions == NULL) {
continue;
}
for (extIndex = 0; engine->extensions[extIndex] != NULL; extIndex++) {
CalogContextT *context;
const char *ext;
char *path;
char *source;
FILE *file;
size_t pathSize;
long fileSize;
size_t readCount;
ext = engine->extensions[extIndex];
pathSize = strlen(baseFileName) + strlen(ext) + 2; // '.' + '\0'
path = (char *)malloc(pathSize);
if (path == NULL) {
return NULL;
}
snprintf(path, pathSize, "%s.%s", baseFileName, ext);
file = fopen(path, "rb");
free(path);
if (file == NULL) {
continue; // not this extension -- try the next
}
// First existing file wins. Read it whole, open a context, load it.
if (fseek(file, 0, SEEK_END) != 0 || (fileSize = ftell(file)) < 0) {
fclose(file);
return NULL;
}
rewind(file);
source = (char *)malloc((size_t)fileSize + 1);
if (source == NULL) {
fclose(file);
return NULL;
}
readCount = fread(source, 1, (size_t)fileSize, file);
fclose(file);
source[readCount] = '\0';
context = inherit ? calogContextOpenInheriting(calog, engine) : calogContextOpen(calog, engine);
if (context == NULL) {
free(source);
return NULL;
}
if (calogContextEval(context, source) != calogOkE) {
free(source);
calogContextClose(context);
return NULL;
}
free(source);
return context;
}
}
return NULL;
}
CalogContextT *calogContextLoad(CalogT *calog, const char *baseFileName) {
return contextLoadCommon(calog, baseFileName, false);
}
// calogContextLoad for a script's own taskLoad: the loaded context inherits the calling script's
// sandbox rather than starting unrestricted (see calogContextOpenInheriting).
CalogContextT *calogContextLoadInheriting(CalogT *calog, const char *baseFileName) {
return contextLoadCommon(calog, baseFileName, true);
}
CalogT *calogCurrent(void) {
return currentContext != NULL ? currentContext->broker : NULL;
}
uint64_t calogCurrentId(void) {
return currentContext != NULL ? currentContext->id : CALOG_HOST_ID;
}
// Ask THIS context to retire itself, requested from inside the task (which has no handle to
// itself). Unlike an owner's taskClose, this does NOT shut the context down mid-eval: it sets
// a deferred flag so the CURRENT native/eval finishes normally (host calls and all), and
// serveLoop stops the context only once that eval returns. The thread then exits and a reaper
// (calogTaskReap) joins and frees it. No-op on the host context, which has no thread to end.
void calogCurrentRetire(void) {
if (currentContext != NULL && currentContext->id != CALOG_HOST_ID) {
atomic_store(&currentContext->retireRequested, true);
}
}
// True once this context's owner has requested it close (calogContextClose set the flag and
// is blocked joining this thread). A long-running script polls it to cooperate; the host
// thread, which has no context to close, always reads false. Lock-free (the flag is atomic).
bool calogCurrentShuttingDown(void) {
if (currentContext == NULL) {
return false;
}
return atomic_load(&currentContext->shuttingDown);
}
// Latch the runtime aborting so no context runs another line of script (see calog.h). Called from a
// native: the engine raises the error returned here, which unwinds the calling script out of its
// chunk at the call site, and every later call -- from this context or any other -- is refused by
// calogCall/calogFnInvoke, so a script that catches the unwind still cannot do anything.
// Idempotent: latching a latched runtime is a no-op.
//
// It deliberately does NOT retire the calling context. Teardown order is the host's (calogDestroy
// releases the cross-context reference holders -- export/pubsub/timer -- while every context is
// still alive, and only then joins them). A context that closed itself here would destroy its own
// interpreter first, stranding the callables those registries still hold: harmless for a VM that
// frees everything on close, fatal for one that asserts it owns no live objects (QuickJS). The
// scripts are stopped either way -- that is what the latch is for.
int32_t calogAbortAll(CalogT *calog, CalogValueT *result) {
atomic_store(&calog->aborting, true);
calogAbortRaise();
return calogFail(result, calogErrAbortE, CALOG_ABORT_MESSAGE);
}
// Record that the abort error is being raised INTO the calling context's VM. Called from the three
// -- and only three -- places the abort can enter running script code: the native that latches the
// runtime (calogAbortAll), the dispatch choke point that refuses every later native (calogCall), and
// an engine's interrupt hook unwinding a script that calls no natives at all. Marking at the moment
// of injection is what lets calogAbortRaised stay exact. No-op off a context thread.
void calogAbortRaise(void) {
if (currentContext == NULL || currentContext->id == CALOG_HOST_ID) {
return;
}
atomic_store(&currentContext->abortRaised, true);
}
// True when the error an engine is holding is one WE put there to unwind this script, rather than
// one the script earned. Engine adapters ask this instead of calogAborting when deciding whether an
// eval failure deserves a diagnostic: "the runtime is aborting" is not the same question, because a
// script can be handed to an engine with a syntax error in it while some other script is tearing the
// runtime down, and that script really did fail -- silently swallowing it hides the one piece of
// information its author needs. Exact for every engine, including the four whose eval API compiles
// and runs in one call and so cannot tell the two apart by return code alone.
bool calogAbortRaised(CalogT *calog) {
if (currentContext == NULL || currentContext->broker != calog) {
return false;
}
return atomic_load(&currentContext->abortRaised);
}
// True when the caller must stop running script code: this runtime was latched by calogAbortAll.
// One query, so the dispatch choke point, every engine adapter and every interrupt hook ask the same
// question. To ask instead whether a FAILURE was caused by that abort, use calogAbortRaised.
bool calogAborting(CalogT *calog) {
return atomic_load(&calog->aborting);
}
// Tear down a single context (quiescence assumed). The thread is stopped and joined,
// then the slot is unlinked under the registry lock so no foreign enqueue can reach
// the freed queue mutex; the freed index returns to the freelist.
void calogContextClose(CalogContextT *context) {
CalogT *broker;
int64_t index;
if (context == NULL) {
return;
}
broker = context->broker;
// Exactly one closer may stop, join and free a given context. A script thread reaches this
// through taskClose, which can land while calogDestroy is tearing the runtime down -- and two
// threads joining one pthread is undefined. So claim it here, under the registry lock, or leave
// it to whoever holds the claim: the teardown (tearingDown is its claim on every context, and it
// waits out any close already in flight) or another close already running.
pthread_mutex_lock(&broker->ctxMutex);
if (broker->tearingDown || context->closing) {
pthread_mutex_unlock(&broker->ctxMutex);
return;
}
context->closing = true;
pthread_mutex_unlock(&broker->ctxMutex);
if (context->started) {
contextRequestShutdown(context);
pthread_join(context->thread, NULL);
}
pthread_mutex_lock(&broker->ctxMutex);
index = idIndex(context->id);
if (index >= 0 && index < broker->ctxCount && broker->ctxSlots[index].context == context) {
broker->ctxSlots[index].context = NULL;
registryFreePush(broker, index);
}
pthread_mutex_unlock(&broker->ctxMutex);
contextDrainQueue(context);
sandboxRelease(context->sandbox);
pthread_mutex_destroy(&context->queueMutex);
pthread_cond_destroy(&context->queueCond);
free(context);
}
// Read one registry slot with ctxMutex held. Teardown walks the registry from the host thread while
// script threads can still be unlinking contexts of their own, so every slot read has to take the
// lock -- an unlocked walk races an open's realloc of the whole array. calogContextOpen fills a slot
// and starts that context's thread inside this same lock, so a slot found here always names a
// context whose thread is running and is therefore joinable.
static CalogContextT *contextAtIndex(CalogT *calog, int64_t index) {
CalogContextT *context;
pthread_mutex_lock(&calog->ctxMutex);
context = calog->ctxSlots[index].context;
pthread_mutex_unlock(&calog->ctxMutex);
return context;
}
static int32_t contextDispatch(CalogT *calog, uint64_t targetId, CalogNativeFnT fn, void *userData, CalogValueT *args, int32_t argCount, CalogValueT *result) {
MessageT *call;
int32_t status;
int32_t index;
calogValueNil(result);
call = (MessageT *)calloc(1, sizeof(*call));
if (call == NULL) {
return calogFail(result, calogErrOomE, "out of memory creating call message");
}
call->kind = messageCallE;
call->fn = fn;
call->userData = userData;
call->argCount = argCount;
if (argCount > 0) {
call->args = (CalogValueT *)calloc((size_t)argCount, sizeof(CalogValueT));
if (call->args == NULL) {
free(call);
return calogFail(result, calogErrOomE, "out of memory copying call arguments");
}
for (index = 0; index < argCount; index++) {
status = calogValueCopy(&call->args[index], &args[index]);
if (status != calogOkE) {
int32_t cleanup;
for (cleanup = 0; cleanup < index; cleanup++) {
calogValueFree(&call->args[cleanup]);
}
free(call->args);
free(call);
return calogFail(result, status, "failed to copy call arguments");
}
}
}
return contextSendBlocking(calog, targetId, call, result);
}
static void contextDispatchCall(CalogContextT *context, MessageT *message) {
CalogValueT result;
int32_t status;
int32_t index;
calogValueNil(&result);
status = message->fn(message->args, message->argCount, &result, message->userData);
if (message->args != NULL) {
for (index = 0; index < message->argCount; index++) {
calogValueFree(&message->args[index]);
}
free(message->args);
message->args = NULL;
}
contextReply(context->broker, message, status, &result);
}
static void contextDispatchError(CalogT *calog, MessageT *message) {
// Runs on the host thread (calogPump / a nested host pump): deliver a fire-and-
// forget script error to the handler, or log it if none is set.
if (calog->errorHandler != NULL) {
calog->errorHandler(message->replyToId, message->source != NULL ? message->source : "", calog->errorUserData);
} else {
fprintf(stderr, "calog: context %llu script error: %s\n", (unsigned long long)message->replyToId, message->source != NULL ? message->source : "");
}
free(message->source);
free(message);
}
static void contextDispatchEval(CalogContextT *context, MessageT *message) {
CalogValueT result;
int32_t status;
// Fire-and-forget: run the script, free the message, and on failure post the
// error to the host thread. No reply.
calogValueNil(&result);
if (context->engine == NULL || context->engine->runSource == NULL) {
status = calogFail(&result, calogErrUnsupportedE, "context has no runnable engine");
} else if (context->interp == NULL) {
// createInterpreter failed (threadMain ignores its status); reject cleanly.
status = calogFail(&result, calogErrUnsupportedE, "engine interpreter unavailable");
} else {
status = context->engine->runSource(context->interp, message->source, &result);
}
free(message->source);
free(message);
if (status != calogOkE) {
calogPostError(context->broker, context->id, &result);
}
calogValueFree(&result);
}
static void contextDispatchRelease(MessageT *message) {
// Runs on the callable's owner thread: perform the engine's closure release via
// calogFnFinalize, then free the message shell.
calogFnFinalize(message->callable);
free(message);
}
static void contextDrainQueue(CalogContextT *context) {
MessageT *message;
// Called at teardown once the owning thread has stopped and the context is unregistered,
// so no new messages can arrive. A release can be orphaned here if a last drop of one of
// this context's callables raced its shutdown (enqueued after serveLoop stopped serving);
// finalize those so the callable is not leaked, and free every other pending message.
while ((message = tryDequeue(context)) != NULL) {
if (message->kind == messageReleaseE) {
calogFnFinalize(message->callable);
free(message);
} else if (message->kind == messageCallE) {
CalogValueT dead;
int32_t argIndex;
// A BLOCKING caller is parked on this message's reply box. Freeing the message would
// leave that thread waiting on a condition nothing can ever signal -- and teardown
// joins that thread, so the wait becomes a process-wide hang rather than one stuck
// call. Fail the call instead: the caller gets calogErrDeadE, which every caller of a
// cross-context native already handles.
if (message->args != NULL) {
for (argIndex = 0; argIndex < message->argCount; argIndex++) {
calogValueFree(&message->args[argIndex]);
}
free(message->args);
message->args = NULL;
}
calogValueNil(&dead);
calogFail(&dead, calogErrDeadE, "calog: the target context is gone");
contextReply(context->broker, message, calogErrDeadE, &dead);
calogValueFree(&dead);
} else if (message->kind == messageErrorE) {
// An error a dying context posted just before its thread ended. Freeing it would throw
// away the one diagnostic the run has for that failure -- and, under bin/calog, the
// failing exit code with it, so a task that died during teardown made the run report
// success. Delivering is safe here: every context thread has been joined by the time
// this runs, so the handler is called on the host thread with nothing left to race.
contextDispatchError(context->broker, message);
} else {
messageFree(message);
}
}
// A nested pumpUntil frame can abandon its token during shutdown (queue-empty exit)
// while a still-active outer frame stashes its late reply (token mismatch) and then
// never claims it once the thread ends. tryDequeue never visits context->stash, so
// free those replies here too.
while (context->stash != NULL) {
message = context->stash;
context->stash = message->next;
messageFree(message);
}
}
static int32_t contextEnqueue(CalogT *calog, uint64_t targetId, MessageT *message) {
CalogContextT *target;
int64_t index;
bool inRange;
pthread_mutex_lock(&calog->ctxMutex);
target = registryResolveLocked(calog, targetId);
if (target != NULL) {
bool queued;
queued = enqueueRaw(target, message);
pthread_mutex_unlock(&calog->ctxMutex);
// A resolvable context whose queue has closed is on its way out: dead for our purposes, and
// reported as such rather than accepting a message no one will serve.
return queued ? calogOkE : calogErrDeadE;
}
// Resolve failed: an in-range index means the slot existed and has since been
// freed or recycled (a dead context); out of range means the id named nothing.
index = idIndex(targetId);
inRange = index >= 0 && index < calog->ctxCount;
pthread_mutex_unlock(&calog->ctxMutex);
if (inRange) {
return calogErrDeadE;
}
return calogErrNotFoundE;
}
// Fire-and-forget: post a callable finalize to its owner thread. On enqueue failure
// the message is freed and actorReleaseCallable finalizes the dead callable itself.
static int32_t contextPostRelease(CalogT *calog, uint64_t targetId, CalogFnT *callable) {
MessageT *message;
int32_t status;
message = (MessageT *)calloc(1, sizeof(*message));
if (message == NULL) {
return calogErrOomE;
}
message->kind = messageReleaseE;
message->callable = callable;
status = contextEnqueue(calog, targetId, message);
if (status != calogOkE) {
free(message);
}
return status;
}
int32_t calogContextEval(CalogContextT *context, const char *source) {
MessageT *eval;
char *sourceCopy;
int32_t status;
sourceCopy = strdup(source);
if (sourceCopy == NULL) {
return calogErrOomE;
}
eval = (MessageT *)calloc(1, sizeof(*eval));
if (eval == NULL) {
free(sourceCopy);
return calogErrOomE;
}
eval->kind = messageEvalE;
eval->source = sourceCopy;
// Fire-and-forget: enqueue onto the context's thread and return. The script runs
// asynchronously; errors surface via the error handler.
status = contextEnqueue(context->broker, context->id, eval);
if (status != calogOkE) {
messageFree(eval);
}
return status;
}
// True once this context's thread has fully exited (serveLoop returned, interpreter destroyed).
// A reaper polls it so it only pthread_joins a context that will not block, then frees it.
bool calogContextFinished(const CalogContextT *context) {
return atomic_load(&context->finished);
}
uint64_t calogContextId(const CalogContextT *context) {
return context->id;
}
void *calogContextInterp(CalogContextT *context) {
return context->interp;
}
bool calogContextRegistered(CalogT *runtime, uint64_t ctxId) {
bool registered;
if (runtime == NULL) {
return false;
}
pthread_mutex_lock(&runtime->ctxMutex);
registered = (registryResolveLocked(runtime, ctxId) != NULL);
pthread_mutex_unlock(&runtime->ctxMutex);
return registered;
}
// List a new callable against the context that owns it, so contextReclaimCallables can release its
// engine handle if that context is torn down while something else still holds it. Locks in the
// runtime's established order (ctxMutex, then the context's queueMutex). An owner that no longer
// resolves needs no entry: its interpreter is already gone, which calogFnFinalize handles.
int32_t calogContextTrackFn(CalogT *runtime, uint64_t ownerCtxId, CalogFnT *fn) {
CalogContextT *owner;
void *buffer;
int64_t capacity;
int32_t status;
if (runtime == NULL || ownerCtxId == CALOG_HOST_ID) {
return calogOkE; // host-owned: no interpreter for the handle to outlive
}
status = calogOkE;
pthread_mutex_lock(&runtime->ctxMutex);
owner = registryResolveLocked(runtime, ownerCtxId);
if (owner != NULL) {
pthread_mutex_lock(&owner->queueMutex);
buffer = owner->ownedFns;
capacity = owner->ownedCap;
status = calogGrow(&buffer, &capacity, owner->ownedCount + 1, sizeof(CalogFnT *));
if (status == calogOkE) {
owner->ownedFns = (CalogFnT **)buffer;
owner->ownedCap = capacity;
owner->ownedFns[owner->ownedCount] = fn;
owner->ownedCount++;
}
pthread_mutex_unlock(&owner->queueMutex);
}
pthread_mutex_unlock(&runtime->ctxMutex);
return status;
}
// Drop a callable from its owner's list as it is finalized. Order within the list carries no
// meaning, so the hole is filled from the end. A callable the owner already reclaimed is not in any
// list (the sweep took the whole array), which is why this can run for one and simply find nothing.
bool calogContextUntrackFn(CalogT *runtime, uint64_t ownerCtxId, CalogFnT *fn) {
CalogContextT *owner;
int64_t index;
int64_t found;
bool mayFinalize;
if (runtime == NULL || ownerCtxId == CALOG_HOST_ID) {
return true;
}
mayFinalize = true;
pthread_mutex_lock(&runtime->ctxMutex);
owner = registryResolveLocked(runtime, ownerCtxId);
if (owner != NULL) {
pthread_mutex_lock(&owner->queueMutex);
found = -1;
for (index = 0; index < owner->ownedCount; index++) {
if (owner->ownedFns[index] == fn) {
found = index;
break;
}
}
// The owner has stopped serving but has not yet reclaimed (threadMain runs the per-context
// shutdown hooks in between, so this window is as long as those take). Finalizing now would
// happen on the wrong thread, which means skipping the engine release -- and the handle
// would then still be live inside an interpreter that is about to be destroyed. Leave the
// callable TRACKED instead: the sweep is the one thread that may run that release, and it
// is about to walk this very list. The finalize's shell holder travels with it.
if (found >= 0 && owner->queueClosed && !owner->reclaimDone) {
calogFnMarkAdopted(fn);
mayFinalize = false;
} else if (found >= 0) {
owner->ownedCount--;
owner->ownedFns[found] = owner->ownedFns[owner->ownedCount];
}
pthread_mutex_unlock(&owner->queueMutex);
}
pthread_mutex_unlock(&runtime->ctxMutex);
return mayFinalize;
}
void calogPump(CalogT *calog) {
CalogContextT *previous;
MessageT *message;
if (calog->hostContext == NULL) {
return;
}
// Present the calling thread as THIS runtime's host for the drain, so a native --
// and anything it calls -- resolves to this runtime. Restoring afterward lets one
// thread pump several runtimes in turn (each drain acts as the right host).
previous = currentContext;
currentContext = calog->hostContext;
// Non-blocking: run every pending host-thread message and return.
while ((message = tryDequeue(calog->hostContext)) != NULL) {
hostDispatch(calog, message);
}
currentContext = previous;
}
// Append an engine to the runtime's calogContextLoad search list (setup-time, host
// thread). On OOM the engine is simply not registered -- load just won't find it.
void calogRegisterEngine(CalogT *calog, const CalogEngineT *engine) {
if (calog->engineCount == calog->engineCap) {
int64_t newCap;
const CalogEngineT **grown;
newCap = (calog->engineCap == 0) ? CONTEXT_INITIAL_ENGINES : calog->engineCap * CALOG_GROWTH_FACTOR;
grown = (const CalogEngineT **)realloc(calog->engines, (size_t)newCap * sizeof(*grown));
if (grown == NULL) {
return;
}
calog->engines = grown;
calog->engineCap = newCap;
}
calog->engines[calog->engineCount] = engine;
calog->engineCount++;
}
void calogSetErrorHandler(CalogT *calog, CalogErrorFnT fn, void *userData) {
calog->errorHandler = fn;
calog->errorUserData = userData;
}
// The reply tail shared by CALL dispatch: hand (status, result) back to whoever is
// blocked on this message -- an external caller's reply box, or a context caller via
// a REPLY enqueued onto its queue (in the serving context's runtime). Consumes message.
// Release the engine handle of every callable this context still owns, on the context's OWN thread,
// while its interpreter is still alive. Anything else holding one of these callables -- a pubsub
// subscription, an export, a value handed to another engine, an invoke in flight -- keeps a valid
// (dead) shell to release later; without this the engine handle would outlive the VM that owns it,
// which most VMs quietly tolerate and QuickJS aborts on. See design.md sec 26.
static void contextReclaimCallables(CalogContextT *context) {
CalogFnT **owned;
CalogFnT **orphans;
int64_t total;
int64_t count;
int64_t orphanCount;
int64_t index;
// Take the whole list first: an engine release can cascade (a VM finalizer dropping another of
// this context's callables), which re-enters untrack, and must not run against a list being
// walked. Holding a reference across the sweep is what keeps such a cascade from freeing an
// entry we have not reached yet.
//
// Both the steal and the retains happen in ONE critical section, and the retain is conditional.
// A foreign thread can drop the last reference to any of these at any moment; retaining after
// that drop would resurrect a callable whose finalize is already committed, and the sweep's own
// release would then free it a second time. calogFnRetainIfLive refuses that case, so an entry
// already being finalized is simply left out -- its finalize frees it, as it was going to. And
// no shell can be freed while this lock is held, because every free path untracks first and
// untrack needs this same mutex, so inspecting the refcount here is safe.
pthread_mutex_lock(&context->queueMutex);
owned = context->ownedFns;
total = context->ownedCount;
count = 0;
orphanCount = 0;
// An entry whose count already hit zero cannot be retained -- a finalize is committed for it on
// another thread. It must not be skipped either: that thread is NOT this one, so it is barred
// from touching the interpreter and finalizes without running the engine release, stranding the
// handle inside a VM that is about to assert it owns nothing. Those go on a second list and get
// their release run here, on the thread that may. A shell holder is taken for each so the
// committed finalize cannot free the memory before this sweep is done with it.
orphans = (total > 0) ? (CalogFnT **)malloc((size_t)total * sizeof(*orphans)) : NULL;
for (index = 0; index < total; index++) {
if (calogFnRetainIfLive(owned[index])) {
owned[count] = owned[index]; // compact: keep only what this sweep now owns
count++;
} else if (orphans != NULL) {
// An adopted entry's finalize already handed its shell holder over (calogContextUntrackFn);
// one that has not been adopted still has a finalize in flight that will drop its own, so
// this sweep takes a holder of its own to keep the shell alive until it is done.
if (!calogFnAdopted(owned[index])) {
calogFnShellHold(owned[index]);
}
orphans[orphanCount] = owned[index];
orphanCount++;
}
}
context->ownedFns = NULL;
context->ownedCount = 0;
context->ownedCap = 0;
context->reclaimDone = true; // from here a finalize completes normally: no more adopting
pthread_mutex_unlock(&context->queueMutex);
for (index = 0; index < orphanCount; index++) {
calogFnReclaimOrphan(orphans[index]);
}
free(orphans);
if (owned == NULL) {
return;
}
for (index = 0; index < count; index++) {
calogFnReclaim(owned[index]);
}
for (index = 0; index < count; index++) {
calogFnRelease(owned[index]); // reclaimed above, so a final drop frees the shell only
}
free(owned);
}
static void contextReply(CalogT *calog, MessageT *message, int32_t status, CalogValueT *result) {
if (message->replyBox != NULL) {
ReplyBoxT *box;
box = message->replyBox;
pthread_mutex_lock(&box->mutex);
box->status = status;
calogValueMove(&box->result, result);
box->done = true;
pthread_cond_signal(&box->cond);
pthread_mutex_unlock(&box->mutex);
free(message);
} else {
// Reuse the request message as the REPLY: it already carries the caller's
// token and id, so the wakeup cannot be lost to an allocation failure here.
uint64_t replyToId;
replyToId = message->replyToId;
message->kind = messageReplyE;
message->status = status;
calogValueMove(&message->result, result);
if (contextEnqueue(calog, replyToId, message) != calogOkE) {
messageFree(message);
}
}
}
// Ask a context's own thread to stop once its queue drains. Allocation-free (unlike an
// enqueued SHUTDOWN message would be) so it cannot fail under OOM and leave the caller's
// pthread_join waiting forever: it just flips the flag messageDequeue already waits on,
// under the same mutex, and signals the cond so a thread parked in cond_wait wakes.
static void contextRequestShutdown(CalogContextT *context) {
pthread_mutex_lock(&context->queueMutex);
context->shuttingDown = true;
pthread_cond_signal(&context->queueCond);
pthread_mutex_unlock(&context->queueMutex);
}
// Enqueue an already-built CALL to targetId and wait for its reply. A context caller
// pumps its own queue (servicing re-entrant work) instead of sleeping; a foreign
// caller blocks on a private reply box. The message is consumed either way.
static int32_t contextSendBlocking(CalogT *calog, uint64_t targetId, MessageT *message, CalogValueT *result) {
CalogContextT *caller;
int32_t status;
// A call that lands on calog's own host queue, made by the exact thread that
// hosts calog, can never be serviced by anyone else -- calogPump(calog) only ever
// runs on that thread, and currentContext may already name a different runtime's
// context (this thread hosts several). Run the call inline instead of enqueuing
// and waiting, or it hangs forever.
if (targetId == CALOG_HOST_ID && pthread_equal(pthread_self(), calog->hostThread)) {
int32_t index;
status = message->fn(message->args, message->argCount, result, message->userData);
if (message->args != NULL) {
for (index = 0; index < message->argCount; index++) {
calogValueFree(&message->args[index]);
}
free(message->args);
}
free(message);
return status;
}
// The token/pump path needs the caller's queue in THIS runtime (its reply routes
// by the caller's id, resolved in calog). A thread that is foreign to calog -- a
// non-calog thread, or one currently hosting a different runtime -- takes the reply
// box instead, so a cross-runtime call cannot misroute its reply.
caller = (currentContext != NULL && currentContext->broker == calog) ? currentContext : NULL;
if (caller != NULL) {
uint64_t token;
int32_t replyStatus;
token = ++caller->nextToken;
message->replyToId = caller->id;
message->token = token;
status = contextEnqueue(calog, targetId, message);
if (status != calogOkE) {
messageFree(message);
return calogFail(result, status, "target context unavailable");
}
status = pumpUntil(caller, token, &replyStatus, result);
if (status != calogOkE) {
return calogFail(result, status, "context is shutting down");
}
return replyStatus;
}
ReplyBoxT box;
pthread_mutex_init(&box.mutex, NULL);
pthread_cond_init(&box.cond, NULL);
box.done = false;
box.status = calogOkE;
calogValueNil(&box.result);
message->replyBox = &box;
status = contextEnqueue(calog, targetId, message);
if (status != calogOkE) {
messageFree(message);
pthread_mutex_destroy(&box.mutex);
pthread_cond_destroy(&box.cond);
return calogFail(result, status, "target context unavailable");
}
pthread_mutex_lock(&box.mutex);
while (!box.done) {
pthread_cond_wait(&box.cond, &box.mutex);
}
pthread_mutex_unlock(&box.mutex);
calogValueMove(result, &box.result);
status = box.status;
pthread_mutex_destroy(&box.mutex);
pthread_cond_destroy(&box.cond);
return status;
}
// The dispatch shared by every loop that drains a queue (serveLoop, pumpUntil,
// hostDispatch): the three message kinds whose handling never varies by loop. Returns
// whether message->kind was one of them (and so was consumed); the caller still owns
// its own loop-specific kinds (REPLY, ERROR) and the default for anything left over.
static bool dispatchCommon(CalogContextT *context, MessageT *message) {
switch (message->kind) {
case messageCallE:
contextDispatchCall(context, message);
return true;
case messageEvalE:
contextDispatchEval(context, message);
return true;
case messageReleaseE:
contextDispatchRelease(message);
return true;
default:
return false;
}
}
// Append to a context's queue, or refuse if that context has stopped serving. Returns false only for
// the refusal; the caller owns the message either way.
static bool enqueueRaw(CalogContextT *context, MessageT *message) {
pthread_mutex_lock(&context->queueMutex);
// callsClosed refuses CALLs but still accepts everything else -- notably a script error, which
// is fire-and-forget and gets delivered by the drain. A CALL is the one kind whose sender WAITS,
// so it is the only kind that can turn "nobody will serve this" into a hang.
if (context->queueClosed || (context->callsClosed && message->kind == messageCallE)) {
pthread_mutex_unlock(&context->queueMutex);
return false;
}
message->next = NULL;
if (context->tail != NULL) {
context->tail->next = message;
} else {
context->head = message;
}
context->tail = message;
pthread_cond_signal(&context->queueCond);
pthread_mutex_unlock(&context->queueMutex);
return true;
}
// Stop the host serving CALLs, and fail everything already queued for it.
//
// Once the host thread is inside calogDestroy it never calls calogPump again, so every host-directed
// CALL from that moment on is guaranteed never to be served -- its sender would block forever. That
// is not hypothetical: the before-contexts destroy hooks stop background threads by JOINING them,
// and a timer thread parked mid-callback on a context that is itself waiting for a host reply
// produces a three-way cycle (host -> timer thread -> context -> host) that hangs teardown outright.
// Refusing new calls and failing the parked ones breaks it before the first join.
static void hostCallsClose(CalogT *calog) {
if (calog->hostContext == NULL) {
return;
}
pthread_mutex_lock(&calog->hostContext->queueMutex);
calog->hostContext->callsClosed = true;
pthread_mutex_unlock(&calog->hostContext->queueMutex);
contextDrainQueue(calog->hostContext);
}
static void hostDispatch(CalogT *calog, MessageT *message) {
if (message->kind == messageErrorE) {
contextDispatchError(calog, message);
return;
}
if (!dispatchCommon(calog->hostContext, message)) {
messageFree(message);
}
}
static uint64_t idCompose(int64_t index, uint32_t generation) {
return ((uint64_t)generation << CONTEXT_INDEX_BITS) | (uint64_t)(index + 1);
}
static uint32_t idGeneration(uint64_t id) {
return (uint32_t)(id >> CONTEXT_INDEX_BITS);
}
static int64_t idIndex(uint64_t id) {
uint64_t low;
low = id & CONTEXT_INDEX_MASK;
if (low == 0) {
return -1;
}
return (int64_t)low - 1;
}
static MessageT *messageDequeue(CalogContextT *context) {
MessageT *message;
pthread_mutex_lock(&context->queueMutex);
while (context->head == NULL && !context->shuttingDown) {
pthread_cond_wait(&context->queueCond, &context->queueMutex);
}
if (context->head == NULL) {
// Queue drained and shutting down: this thread is about to leave serveLoop. Close the queue
// in the SAME critical section that decides to stop, so a sender can never slip a message in
// between -- one that nothing would ever serve, leaving a blocking caller waiting on a reply
// that cannot come. From here contextEnqueue reports the context dead, which every caller
// already handles.
context->queueClosed = true;
pthread_mutex_unlock(&context->queueMutex);
return NULL;
}
message = context->head;
context->head = message->next;
if (context->head == NULL) {
context->tail = NULL;
}
pthread_mutex_unlock(&context->queueMutex);
message->next = NULL;
return message;
}
static void messageFree(MessageT *message) {
int32_t index;
if (message == NULL) {
return;
}
if (message->kind == messageCallE && message->args != NULL) {
for (index = 0; index < message->argCount; index++) {
calogValueFree(&message->args[index]);
}
free(message->args);
}
if (message->kind == messageEvalE || message->kind == messageErrorE) {
free(message->source);
}
if (message->kind == messageReplyE) {
calogValueFree(&message->result);
}
free(message);
}
// Post a fire-and-forget script error to the host thread's error handler.
// Deliver a script failure to the runtime's error handler (on the host thread). Also the way a
// library's own background thread reports a callback that failed -- see calogInternal.h.
void calogPostError(CalogT *calog, uint64_t contextId, const CalogValueT *result) {
MessageT *message;
const char *text;
text = (result->type == calogStringE) ? result->as.s.bytes : "script error";
message = (MessageT *)calloc(1, sizeof(*message));
if (message == NULL) {
return;
}
message->kind = messageErrorE;
message->replyToId = contextId;
message->source = strdup(text);
if (message->source == NULL) {
free(message);
return;
}
if (contextEnqueue(calog, CALOG_HOST_ID, message) != calogOkE) {
free(message->source);
free(message);
}
}
static int32_t pumpUntil(CalogContextT *context, uint64_t token, int32_t *outStatus, CalogValueT *result) {
for (;;) {
MessageT *message;
MessageT *prev;
MessageT *node;
// Re-check the stash: a nested pump may have stashed this token's reply.
prev = NULL;
node = context->stash;
while (node != NULL) {
if (node->token == token) {
if (prev != NULL) {
prev->next = node->next;
} else {
context->stash = node->next;
}
*outStatus = node->status;
calogValueMove(result, &node->result);
free(node);
return calogOkE;
}
prev = node;
node = node->next;
}
message = messageDequeue(context);
if (message == NULL) {
calogValueNil(result);
return calogErrNotFoundE;
}
if (message->kind == messageReplyE) {
if (message->token == token) {
*outStatus = message->status;
calogValueMove(result, &message->result);
free(message);
return calogOkE;
}
message->next = context->stash;
context->stash = message;
continue;
}
if (message->kind == messageErrorE) {
contextDispatchError(context->broker, message);
continue;
}
if (!dispatchCommon(context, message)) {
messageFree(message);
}
}
}
// Push a freed slot index onto the recycle freelist. Called under calog->ctxMutex. On
// OOM the index is simply not recycled (a leaked slot, never a correctness fault) --
// void because neither caller has anything to act on.
static void registryFreePush(CalogT *calog, int64_t index) {
if (calog->ctxFreeCount == calog->ctxFreeCap) {
int64_t newCap;
int64_t *grown;
newCap = (calog->ctxFreeCap == 0) ? CONTEXT_INITIAL_CONTEXTS : calog->ctxFreeCap * CALOG_GROWTH_FACTOR;
grown = (int64_t *)realloc(calog->ctxFree, (size_t)newCap * sizeof(int64_t));
if (grown == NULL) {
return;
}
calog->ctxFree = grown;
calog->ctxFreeCap = newCap;
}
calog->ctxFree[calog->ctxFreeCount] = index;
calog->ctxFreeCount++;
}
static CalogContextT *registryResolveLocked(CalogT *calog, uint64_t id) {
int64_t index;
if (id == CALOG_HOST_ID) {
return calog->hostContext;
}
index = idIndex(id);
if (index < 0 || index >= calog->ctxCount) {
return NULL;
}
if (calog->ctxSlots[index].context == NULL) {
return NULL;
}
if (calog->ctxSlots[index].generation != idGeneration(id)) {
return NULL;
}
// Interpreter already torn down: treat as gone so late releases finalize without touching
// it (the slot itself stays populated until calogContextClose/calogActorShutdown frees it).
if (calog->ctxSlots[index].context->interpDead) {
return NULL;
}
return calog->ctxSlots[index].context;
}
static void serveLoop(CalogContextT *context) {
for (;;) {
MessageT *message;
// messageDequeue returns NULL once the queue is empty and contextRequestShutdown
// has flipped shuttingDown -- there is no SHUTDOWN message kind to look for.
message = messageDequeue(context);
if (message == NULL) {
return;
}
if (!dispatchCommon(context, message)) {
messageFree(message);
}
if (atomic_load(&context->retireRequested)) {
// A native in the eval that just finished called calogCurrentRetire (taskExit): the
// eval ran to completion normally; now stop so this context's thread can exit.
contextRequestShutdown(context);
}
}
}
static MessageT *tryDequeue(CalogContextT *context) {
MessageT *message;
pthread_mutex_lock(&context->queueMutex);
message = context->head;
if (message != NULL) {
context->head = message->next;
if (context->head == NULL) {
context->tail = NULL;
}
}
pthread_mutex_unlock(&context->queueMutex);
if (message != NULL) {
message->next = NULL;
}
return message;
}
static void *threadMain(void *arg) {
CalogContextT *context;
int64_t index;
context = (CalogContextT *)arg;
currentContext = context;
if (context->engine != NULL && context->engine->createInterpreter != NULL) {
context->engine->createInterpreter(context, &context->interp);
}
// Per-context library init, on this context's own thread now that the interpreter exists.
for (index = 0; index < context->broker->contextHookCount; index++) {
if (context->broker->contextHooks[index].init != NULL) {
context->broker->contextHooks[index].init(context, context->broker->contextHooks[index].userData);
}
}
serveLoop(context);
// Per-context library shutdown (reverse order), while the context + interpreter are still valid.
for (index = context->broker->contextHookCount - 1; index >= 0; index--) {
if (context->broker->contextHooks[index].shutdown != NULL) {
context->broker->contextHooks[index].shutdown(context, context->broker->contextHooks[index].userData);
}
}
// Reclaim the engine handle of every callable this context still owns, while the interpreter is
// alive and this is its own thread. Anything still holding one -- a subscription, an export, a
// value another engine received -- is left a dead shell it can safely release later.
contextReclaimCallables(context);
// Mark the interpreter dead BEFORE destroying it, so registryResolveLocked stops handing
// this context out: a callable this context owns that is released from now on (e.g. one
// published via the export library and dropped after the context unloads) finalizes by
// freeing its struct directly instead of running the engine release on a dead interpreter.
pthread_mutex_lock(&context->broker->ctxMutex);
context->interpDead = true;
pthread_mutex_unlock(&context->broker->ctxMutex);
if (context->engine != NULL && context->engine->destroyInterpreter != NULL) {
context->engine->destroyInterpreter(context->interp);
}
currentContext = NULL;
// Last thing the thread does: publish that it has exited, so a reaper (calogTaskReap on
// the host) can pthread_join it without blocking and then free the context.
atomic_store(&context->finished, true);
return NULL;
}
Reference in a new issue View git blame Copy permalink