AI_Slop/calog
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
calog/src/value.c

713 lines
23 KiB
C
Raw Blame History

// value.c -- CalogValueT / CalogAggT / CalogFnT lifecycle for the broker core.
//
// All cross-boundary data is by-value: calogValueCopy deep-copies aggregates and
// strings so no two owners (and, later, no two threads) ever share a heap
// pointer. The single exception is calogFnE: a function value is a refcounted
// CalogFnT handle that is shared by reference and only ever invoked, never
// inspected. Aggregates are required to be acyclic and bounded by
// CALOG_MAX_DEPTH; the copy path enforces the bound, and the free path relies
// on it.
#include "calogInternal.h"
#include <math.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#define BROKER_MIN_CAPACITY 8
#define CALLABLE_INITIAL_REFCOUNT 1
struct CalogFnT {
CalogNativeFnT fn;
void *userData;
CalogReleaseFnT release;
CalogT *runtime; // owning runtime; its hooks route invoke/release
uint64_t ownerCtxId; // the owning context's 64-bit id (0 = host)
_Atomic int32_t refCount;
_Atomic bool alive;
// Set by calogFnReclaim once the owner has run the engine release, so a later finalize does not
// run it again. Atomic, and the ONLY thing reclaim changes: fn/userData/release are written once
// at create and read by other threads (actorInvokeCallable marshals them), so mutating those
// would be a data race against an invoke already in flight.
_Atomic bool reclaimed;
};
static int32_t aggregateCopyDepth(CalogAggT **out, const CalogAggT *src, int32_t depth);
static int32_t valueCopyDepth(CalogValueT *dst, const CalogValueT *src, int32_t depth);
static int32_t aggregateCopyDepth(CalogAggT **out, const CalogAggT *src, int32_t depth) {
CalogAggT *copy;
void *buffer;
int32_t status;
int64_t index;
*out = NULL;
if (depth >= CALOG_MAX_DEPTH) {
return calogErrDepthE;
}
status = calogAggCreate(&copy, src->kind);
if (status != calogOkE) {
return status;
}
for (index = 0; index < src->arrayCount; index++) {
CalogValueT element;
status = valueCopyDepth(&element, &src->array[index], depth + 1);
if (status != calogOkE) {
calogAggFree(copy);
return status;
}
status = calogAggPush(copy, &element);
if (status != calogOkE) {
calogValueFree(&element);
calogAggFree(copy);
return status;
}
}
for (index = 0; index < src->pairCount; index++) {
CalogValueT key;
CalogValueT value;
status = valueCopyDepth(&key, &src->pairs[index].key, depth + 1);
if (status != calogOkE) {
calogAggFree(copy);
return status;
}
status = valueCopyDepth(&value, &src->pairs[index].value, depth + 1);
if (status != calogOkE) {
calogValueFree(&key);
calogAggFree(copy);
return status;
}
// Source pairs are unique by construction, so append directly instead of via
// calogAggSet, whose per-insert duplicate scan made this copy O(n^2).
buffer = copy->pairs;
status = calogGrow(&buffer, &copy->pairCap, copy->pairCount + 1, sizeof(CalogPairT));
if (status != calogOkE) {
calogValueFree(&key);
calogValueFree(&value);
calogAggFree(copy);
return status;
}
copy->pairs = (CalogPairT *)buffer;
calogValueMove(&copy->pairs[copy->pairCount].key, &key);
calogValueMove(&copy->pairs[copy->pairCount].value, &value);
copy->pairCount++;
}
*out = copy;
return calogOkE;
}
int32_t calogAggCreate(CalogAggT **out, CalogKindE kind) {
CalogAggT *aggregate;
*out = NULL;
aggregate = (CalogAggT *)calloc(1, sizeof(*aggregate));
if (aggregate == NULL) {
return calogErrOomE;
}
aggregate->kind = kind;
*out = aggregate;
return calogOkE;
}
void calogAggFree(CalogAggT *aggregate) {
int64_t index;
if (aggregate == NULL) {
return;
}
for (index = 0; index < aggregate->arrayCount; index++) {
calogValueFree(&aggregate->array[index]);
}
for (index = 0; index < aggregate->pairCount; index++) {
calogValueFree(&aggregate->pairs[index].key);
calogValueFree(&aggregate->pairs[index].value);
}
free(aggregate->array);
free(aggregate->pairs);
free(aggregate);
}
CalogValueT *calogAggGet(CalogAggT *aggregate, const CalogValueT *key) {
int64_t index;
for (index = 0; index < aggregate->pairCount; index++) {
if (calogValueEquals(&aggregate->pairs[index].key, key)) {
return &aggregate->pairs[index].value;
}
}
return NULL;
}
bool calogAggIsKeyed(const CalogAggT *aggregate) {
return aggregate->pairCount > 0 || aggregate->kind == calogMapE;
}
int32_t calogAggPush(CalogAggT *aggregate, CalogValueT *value) {
void *buffer;
int32_t status;
// Pun through a real void* lvalue, never (void **)&typedPointer, to avoid a
// strict-aliasing violation when calogGrow stores the resized pointer.
buffer = aggregate->array;
status = calogGrow(&buffer, &aggregate->arrayCap, aggregate->arrayCount + 1, sizeof(CalogValueT));
if (status != calogOkE) {
return status;
}
aggregate->array = (CalogValueT *)buffer;
calogValueMove(&aggregate->array[aggregate->arrayCount], value);
aggregate->arrayCount++;
return calogOkE;
}
int32_t calogAggSet(CalogAggT *aggregate, CalogValueT *key, CalogValueT *value) {
void *buffer;
int32_t status;
int64_t index;
for (index = 0; index < aggregate->pairCount; index++) {
if (calogValueEquals(&aggregate->pairs[index].key, key)) {
calogValueFree(&aggregate->pairs[index].value);
calogValueMove(&aggregate->pairs[index].value, value);
calogValueFree(key);
return calogOkE;
}
}
buffer = aggregate->pairs;
status = calogGrow(&buffer, &aggregate->pairCap, aggregate->pairCount + 1, sizeof(CalogPairT));
if (status != calogOkE) {
return status;
}
aggregate->pairs = (CalogPairT *)buffer;
calogValueMove(&aggregate->pairs[aggregate->pairCount].key, key);
calogValueMove(&aggregate->pairs[aggregate->pairCount].value, value);
aggregate->pairCount++;
return calogOkE;
}
int32_t calogFnCreate(CalogFnT **out, CalogT *runtime, CalogNativeFnT fn, void *userData, CalogReleaseFnT release, uint64_t ownerCtxId) {
CalogFnT *callable;
int32_t status;
*out = NULL;
callable = (CalogFnT *)calloc(1, sizeof(*callable));
if (callable == NULL) {
return calogErrOomE;
}
callable->fn = fn;
callable->userData = userData;
callable->release = release;
callable->runtime = runtime;
callable->ownerCtxId = ownerCtxId;
atomic_init(&callable->refCount, CALLABLE_INITIAL_REFCOUNT);
atomic_init(&callable->alive, true);
atomic_init(&callable->reclaimed, false);
// List it against the owning context, which reclaims the engine handle if it is torn down while
// something else still holds this callable (design.md sec 26). No list, no reclaim -- so a
// failure here fails the create rather than handing back a callable that could strand its handle.
status = calogContextTrackFn(runtime, ownerCtxId, callable);
if (status != calogOkE) {
free(callable);
return status;
}
*out = callable;
return calogOkE;
}
int32_t calogFnFromNative(CalogFnT **out, CalogT *calog, CalogNativeFnT fn, void *userData) {
// A host-owned callable (owner id CALOG_HOST_ID = host): no engine handle to release.
return calogFnCreate(out, calog, fn, userData, NULL, CALOG_HOST_ID);
}
void calogFnFinalize(CalogFnT *callable) {
if (callable == NULL) {
return;
}
// Last reference is gone. If the owner context is still alive, run the engine's closure
// release (luaL_unref / sq_release / ...), which also frees the engine's per-callable
// struct; the actor layer routes this to the owner's thread (design.md sec 10) so the
// interpreter op is safe. If the owner is GONE, its interpreter -- and this callable's
// handle inside it -- were already destroyed, so we must NOT touch it: free the engine's
// per-callable struct directly. This is what lets an exported function outlive its
// context (e.g. a script unloaded and reloaded, re-exporting new versions).
//
// (Freeing userData directly requires every engine's per-callable struct to be a single
// heap allocation; all adapters honor that -- see CalogReleaseFnT in calogInternal.h.)
//
// A callable the owner already RECLAIMED (calogFnReclaim, as its interpreter was torn down)
// has no release hook left, so this frees only the shell -- the engine handle went with the
// interpreter that owned it.
calogContextUntrackFn(callable->runtime, callable->ownerCtxId, callable);
if (callable->release != NULL && !atomic_exchange_explicit(&callable->reclaimed, true, memory_order_acq_rel)) {
if (callable->ownerCtxId == CALOG_HOST_ID || calogContextRegistered(callable->runtime, callable->ownerCtxId)) {
callable->release(callable);
} else {
free(callable->userData);
}
}
free(callable);
}
// The owning context is tearing its interpreter down; this runs on that context's own thread, with
// the interpreter still alive, for every callable the context still owns. Running the engine release
// HERE is what keeps the handle from outliving its VM: whoever else still holds a reference -- a
// pubsub subscription, an export, a value sitting in another engine, an invoke in flight -- finds a
// dead callable and finalizes an empty shell. Marking it dead first makes any such invoke fail
// cleanly instead of reaching into a VM that is going away.
void calogFnReclaim(CalogFnT *callable) {
if (callable == NULL) {
return;
}
calogFnMarkDead(callable);
// Claim the release with an atomic exchange so it runs exactly once, here or in a finalize --
// never both. Nothing else about the callable changes: an invoke already past its alive check
// may still be reading fn/userData on another thread to marshal this call, and those reads must
// not race a write. (That marshal cannot land: reclaim runs after serveLoop has closed this
// context's queue, so the dispatch is refused -- see sec 28.)
if (callable->release != NULL && !atomic_exchange_explicit(&callable->reclaimed, true, memory_order_acq_rel)) {
callable->release(callable); // frees the engine handle AND the adapter's userData block
}
}
CalogNativeFnT calogFnNative(const CalogFnT *callable) {
return callable->fn;
}
int32_t calogFnInvoke(CalogFnT *callable, CalogValueT *args, int32_t argCount, CalogValueT *result) {
calogValueNil(result);
// The owner reclaimed this callable as its interpreter went away (calogFnReclaim), so there is
// nothing left to call. calogErrDeadE is the same answer the routing path gives for a context
// that is gone -- the timer library cancels a timer on exactly that, and pubsub stops counting
// the subscriber as delivered.
if (!atomic_load_explicit(&callable->alive, memory_order_acquire)) {
return calogFail(result, calogErrDeadE, "callable owner no longer exists");
}
// The caller has been stopped (calogAbortAll, or this script's own calogAbortCurrent): a script
// function is script code, so refuse it for the same reason calogCall refuses a native -- a timer
// or subscriber callback already queued must not start running a script body afterwards.
if (callable->runtime != NULL && calogAborting(callable->runtime)) {
return calogFail(result, calogErrAbortE, CALOG_ABORT_MESSAGE);
}
// The owning runtime's actor layer, if present, marshals a foreign-thread invoke
// to the owner's thread; inline otherwise (a bare broker has no hook).
if (callable->runtime != NULL && callable->runtime->invokeHook != NULL) {
return callable->runtime->invokeHook(callable, args, argCount, result);
}
return callable->fn(args, argCount, result, callable->userData);
}
// Finalize a callable WITHOUT running the engine release, for a caller that is not on the owner's
// thread and could not marshal the release there (actorReleaseCallable's fallback). The handle
// INSIDE the interpreter is deliberately left behind -- it is pinned until that interpreter is
// destroyed, which for a long-lived context can be a long time, but leaking beats reaching into a
// live VM from the wrong thread.
//
// userData is freed on exactly the same condition as calogFnFinalize: only when a release hook
// exists. That hook is what makes userData calog's to free (an adapter's single heap block, per the
// CalogReleaseFnT contract). A host-owned callable from calogFnFromNative has NO hook and a userData
// the EMBEDDER owns -- freeing that would corrupt their heap, which is worse than anything this
// function exists to prevent. A callable already reclaimed has both cleared, so it frees the shell.
void calogFnFinalizeForeign(CalogFnT *callable) {
if (callable == NULL) {
return;
}
calogContextUntrackFn(callable->runtime, callable->ownerCtxId, callable);
if (callable->release != NULL && !atomic_exchange_explicit(&callable->reclaimed, true, memory_order_acq_rel)) {
free(callable->userData);
}
free(callable);
}
// Weak-to-strong upgrade: take a reference ONLY if this callable has not already dropped to zero.
// A plain calogFnRetain cannot be used to adopt a callable found in a list, because the count
// reaching zero is what COMMITS calogFnRelease to finalizing it -- retaining after that resurrects a
// corpse the in-flight finalize is about to free, and the second drop would free it twice. The CAS
// loop refuses exactly that case. The caller must hold the owner context's queueMutex, which is what
// keeps the shell itself alive to be inspected: every path that frees one (calogFnFinalize,
// calogFnFinalizeForeign) untracks first, and untrack needs that lock.
bool calogFnRetainIfLive(CalogFnT *callable) {
int32_t current;
current = atomic_load_explicit(&callable->refCount, memory_order_acquire);
while (current > 0) {
if (atomic_compare_exchange_weak_explicit(&callable->refCount, &current, current + 1,
memory_order_acq_rel, memory_order_acquire)) {
return true;
}
}
return false;
}
void calogFnMarkDead(CalogFnT *callable) {
if (callable == NULL) {
return;
}
atomic_store_explicit(&callable->alive, false, memory_order_release);
}
uint64_t calogFnOwner(const CalogFnT *callable) {
return callable->ownerCtxId;
}
void calogFnRelease(CalogFnT *callable) {
int32_t previous;
if (callable == NULL) {
return;
}
previous = atomic_fetch_sub_explicit(&callable->refCount, 1, memory_order_acq_rel);
if (previous == 1) {
// This drop took the count to zero. The owning runtime's actor layer (if
// installed) routes the finalize to the owner's thread; otherwise inline.
if (callable->runtime != NULL && callable->runtime->releaseHook != NULL) {
callable->runtime->releaseHook(callable);
} else {
calogFnFinalize(callable);
}
}
}
void calogFnRetain(CalogFnT *callable) {
if (callable == NULL) {
return;
}
atomic_fetch_add_explicit(&callable->refCount, 1, memory_order_relaxed);
}
CalogT *calogFnRuntime(const CalogFnT *callable) {
return callable->runtime;
}
void *calogFnUserData(const CalogFnT *callable) {
return callable->userData;
}
int32_t calogGrow(void **buffer, int64_t *cap, int64_t needed, size_t elemSize) {
int64_t newCap;
void *resized;
if (*cap >= needed) {
return calogOkE;
}
newCap = (*cap == 0) ? BROKER_MIN_CAPACITY : *cap;
while (newCap < needed) {
if (newCap > INT64_MAX / CALOG_GROWTH_FACTOR) {
return calogErrOomE;
}
newCap *= CALOG_GROWTH_FACTOR;
}
if ((uint64_t)newCap > (uint64_t)(SIZE_MAX / elemSize)) {
return calogErrOomE;
}
resized = realloc(*buffer, (size_t)newCap * elemSize);
if (resized == NULL) {
return calogErrOomE;
}
*buffer = resized;
*cap = newCap;
return calogOkE;
}
int32_t calogValueCopy(CalogValueT *dst, const CalogValueT *src) {
return valueCopyDepth(dst, src, 0);
}
static int32_t valueCopyDepth(CalogValueT *dst, const CalogValueT *src, int32_t depth) {
CalogAggT *aggregate;
int32_t status;
calogValueNil(dst);
switch (src->type) {
case calogNilE:
break;
case calogBoolE:
dst->as.b = src->as.b;
break;
case calogIntE:
dst->as.i = src->as.i;
break;
case calogRealE:
dst->as.r = src->as.r;
break;
case calogStringE:
return calogValueString(dst, src->as.s.bytes, src->as.s.length);
case calogAggE:
status = aggregateCopyDepth(&aggregate, src->as.agg, depth);
if (status != calogOkE) {
return status;
}
calogValueAgg(dst, aggregate);
return calogOkE;
case calogFnE:
calogFnRetain(src->as.fn);
dst->as.fn = src->as.fn;
break;
}
dst->type = src->type;
return calogOkE;
}
// Equality used for map-key matching. Scalars and strings compare by value;
// aggregate and function values compare by pointer identity, so they are only
// useful as keys when the exact same handle is reused (a deep copy yields a new
// pointer). Real keys use IEEE semantics: NaN never matches and +0.0 == -0.0.
bool calogValueEquals(const CalogValueT *a, const CalogValueT *b) {
if (a->type != b->type) {
return false;
}
switch (a->type) {
case calogNilE:
return true;
case calogBoolE:
return a->as.b == b->as.b;
case calogIntE:
return a->as.i == b->as.i;
case calogRealE:
return a->as.r == b->as.r;
case calogStringE:
if (a->as.s.length != b->as.s.length) {
return false;
}
return memcmp(a->as.s.bytes, b->as.s.bytes, (size_t)a->as.s.length) == 0;
case calogAggE:
return a->as.agg == b->as.agg;
case calogFnE:
return a->as.fn == b->as.fn;
}
return false;
}
void calogValueFree(CalogValueT *value) {
if (value == NULL) {
return;
}
switch (value->type) {
case calogNilE:
case calogBoolE:
case calogIntE:
case calogRealE:
break;
case calogStringE:
free(value->as.s.bytes);
break;
case calogAggE:
calogAggFree(value->as.agg);
break;
case calogFnE:
calogFnRelease(value->as.fn);
break;
}
calogValueNil(value);
}
void calogValueAgg(CalogValueT *value, CalogAggT *aggregate) {
value->type = calogAggE;
value->as.agg = aggregate;
}
void calogValueBool(CalogValueT *value, bool b) {
value->type = calogBoolE;
value->as.b = b;
}
void calogValueFn(CalogValueT *value, CalogFnT *callable) {
value->type = calogFnE;
value->as.fn = callable;
}
void calogValueFromDouble(CalogValueT *out, double number) {
if (isfinite(number) && number == floor(number) && number >= CALOG_INT64_MIN_DOUBLE && number < CALOG_INT64_MAX_DOUBLE) {
calogValueInt(out, (int64_t)number);
} else {
calogValueReal(out, number);
}
}
void calogValueInt(CalogValueT *value, int64_t i) {
value->type = calogIntE;
value->as.i = i;
}
void calogValueNil(CalogValueT *value) {
memset(value, 0, sizeof(*value));
value->type = calogNilE;
}
void calogValueReal(CalogValueT *value, double r) {
value->type = calogRealE;
value->as.r = r;
}
int32_t calogValueString(CalogValueT *value, const char *bytes, int64_t length) {
char *buffer;
if (length < 0) {
calogValueNil(value);
return calogErrRangeE;
}
if (bytes == NULL && length > 0) {
calogValueNil(value);
return calogErrArgE;
}
buffer = (char *)malloc((size_t)length + 1);
if (buffer == NULL) {
calogValueNil(value);
return calogErrOomE;
}
if (length > 0) {
memcpy(buffer, bytes, (size_t)length);
}
buffer[length] = '\0';
value->type = calogStringE;
value->as.s.bytes = buffer;
value->as.s.length = length;
return calogOkE;
}
void calogValueMove(CalogValueT *dst, CalogValueT *src) {
if (dst == src) {
return;
}
*dst = *src;
calogValueNil(src);
}
const char *calogTypeName(CalogTypeE type) {
switch (type) {
case calogNilE:
return "nil";
case calogBoolE:
return "bool";
case calogIntE:
return "int";
case calogRealE:
return "real";
case calogStringE:
return "string";
case calogAggE:
return "aggregate";
case calogFnE:
return "function";
}
return "unknown";
}
// ---- shared library helpers (see calogInternal.h) ----
int32_t calogMapSetBool(CalogAggT *map, const char *key, bool flag) {
CalogValueT keyValue;
CalogValueT boolValue;
int32_t status;
status = calogValueString(&keyValue, key, (int64_t)strlen(key));
if (status != calogOkE) {
return status;
}
calogValueBool(&boolValue, flag);
status = calogAggSet(map, &keyValue, &boolValue);
if (status != calogOkE) {
calogValueFree(&keyValue);
}
return status;
}
int32_t calogMapSetInt(CalogAggT *map, const char *key, int64_t value) {
CalogValueT keyValue;
CalogValueT intValue;
int32_t status;
status = calogValueString(&keyValue, key, (int64_t)strlen(key));
if (status != calogOkE) {
return status;
}
calogValueInt(&intValue, value);
status = calogAggSet(map, &keyValue, &intValue);
if (status != calogOkE) {
calogValueFree(&keyValue);
}
return status;
}
int32_t calogMapSetStr(CalogAggT *map, const char *key, const char *bytes, int64_t length) {
CalogValueT keyValue;
CalogValueT stringValue;
int32_t status;
status = calogValueString(&keyValue, key, (int64_t)strlen(key));
if (status != calogOkE) {
return status;
}
status = calogValueString(&stringValue, bytes, length);
if (status != calogOkE) {
calogValueFree(&keyValue);
return status;
}
status = calogAggSet(map, &keyValue, &stringValue);
if (status != calogOkE) {
calogValueFree(&keyValue);
calogValueFree(&stringValue);
}
return status;
}
void calogRegistryRelease(pthread_mutex_t *initMutex, int32_t *refCount, void (*freeAll)(void)) {
pthread_mutex_lock(initMutex);
// freeAll runs exactly once, on the 1 -> 0 transition. A release at refcount 0 (e.g. an
// auto-shutdown after the caller already shut the library down by hand) is a safe no-op.
if (*refCount > 0) {
(*refCount)--;
if (*refCount == 0) {
freeAll();
}
}
pthread_mutex_unlock(initMutex);
}
void calogRegistryRetain(pthread_mutex_t *initMutex, int32_t *refCount) {
pthread_mutex_lock(initMutex);
(*refCount)++;
pthread_mutex_unlock(initMutex);
}
Reference in a new issue View git blame Copy permalink