diff --git a/docs/BUILTINS.md b/docs/BUILTINS.md index 84716b1..67edc60 100644 --- a/docs/BUILTINS.md +++ b/docs/BUILTINS.md @@ -568,6 +568,8 @@ Requires full build. Transformer model inference and training. | `recv_timeout` | `recv_timeout of [channel, ms]` | Bounded-wait receive. Returns the value if one arrives before `ms` milliseconds elapse, else `null`. A close while waiting also returns `null`. Fractional `ms` is honored (ns precision on Linux); negative `ms` degenerates to a `try_recv`. | | `close_channel` | `close_channel of channel` | Close the channel. Wakes all blocked senders/receivers. | | `channel_closed` | `channel_closed of channel` | Returns 1 if closed, 0 otherwise. | +| `task_spawn` | `task_spawn of fn` or `task_spawn of [fn, arg1, ...]` | Create a cooperative task (#408) running `fn` on the single OS thread — deterministic by construction, unlike `spawn`'s OS thread. Args are deep-COPIED (share-nothing, like channel sends), not shared by reference. Returns a numeric task id. (Increment 1a: the task is recorded and reported by `task_alive`; the copying-stack scheduler that runs and interleaves tasks — `task_yield`/`task_join` — lands in a later increment.) | +| `task_alive` | `task_alive of id` | Returns 1 while the task is runnable or suspended, 0 once it has finished (or for an unknown id). | **Thread safety:** Values sent through a channel (or returned through `thread_join`) are deep-COPIED via `val_clone_for_send` — messages are diff --git a/src/builtins.c b/src/builtins.c index deed954..f30fdf9 100644 --- a/src/builtins.c +++ b/src/builtins.c @@ -4428,6 +4428,88 @@ Value* builtin_channel_closed(Value *arg) { return make_num(closed ? 1 : 0); } +/* ==== #408 cooperative task layer (increment 1a: spawn + alive) ==== + * + * A Task rides the single OS thread; it never flips g_vm_multithreaded, so + * the JIT stays live and the refcount fast paths stay non-atomic. Increment + * 1a records tasks and reports liveness; the copying-stack scheduler that + * actually runs and interleaves them lands in 1b (task_yield/task_join and + * the suspend/resume surgery). Held refs (entry_fn/args) are kept live by + * the trial-deletion cycle collector automatically — a counted ref exceeds + * the collector's internal-edge count within its candidate set U, exactly + * as ThreadHandle->fn does — so no root registration is needed. These + * builtins take NO trace records: the scheduling order is a pure function of + * program order, not host nondeterminism (the #408 deterministic-by- + * construction ruling), so wrapping them in TRACE_NONDET_RET would be wrong. */ +void task_free(Task *t) { + if (!t) return; + if (t->entry_fn) val_decref(t->entry_fn); + if (t->args) { + for (int i = 0; i < t->argc; i++) + if (t->args[i]) val_decref(t->args[i]); + free(t->args); + } + if (t->result) val_decref(t->result); + if (t->error_value) val_decref(t->error_value); + free(t->saved_stack); /* NULL in 1a; populated in 1b */ + free(t->saved_frames); + free(t); +} + +/* task_spawn of fn / task_spawn of [fn, arg1, ...] → task id (a number). + * Args are deep-copied to the heap (share-nothing at the boundary, exactly + * like channel sends / thread_join results — #293's val_clone_for_send), so + * a task never shares a mutable arena/heap value with its spawner by + * reference. The task does not run yet (1a); the scheduler starts it in 1b. */ +Value* builtin_task_spawn(Value *arg) { + Value *fn = arg; + Value **args = NULL; + int argc = 0; + if (arg && arg->type == VAL_LIST && arg->data.list.count >= 1) { + fn = arg->data.list.items[0]; + argc = arg->data.list.count - 1; + if (argc > 0) { + args = xmalloc(sizeof(Value*) * argc); + for (int i = 0; i < argc; i++) + args[i] = val_clone_for_send(arg->data.list.items[i + 1]); + } + } + if (!fn || (fn->type != VAL_FN && fn->type != VAL_BUILTIN)) { + if (args) { + for (int i = 0; i < argc; i++) val_decref(args[i]); + free(args); + } + rt_error(EK_TYPE, 0, "task_spawn requires a function or [function, arg1, ...]"); + return make_null(); + } + Task *t = xcalloc(1, sizeof(Task)); + t->state = TASK_READY; + t->entry_fn = fn; + val_incref(fn); + t->args = args; + t->argc = argc; + int id = handle_register(t, HANDLE_TASK); + if (id < 0) { + task_free(t); + rt_error(EK_LIMIT, 0, "task_spawn: too many live tasks (max %d)", + HANDLE_TABLE_SIZE - 1); + return make_null(); + } + t->id = id; + return make_num((double)id); +} + +/* task_alive of id → 1 while the task is READY/RUNNING/SUSPENDED, else 0 + * (DONE, DEAD, or an unknown id). */ +Value* builtin_task_alive(Value *arg) { + if (!arg || arg->type != VAL_NUM) return make_num(0); + Task *t = (Task*)handle_lookup((int)arg->data.num, HANDLE_TASK); + if (!t) return make_num(0); + int alive = (t->state == TASK_READY || t->state == TASK_RUNNING || + t->state == TASK_SUSPENDED); + return make_num(alive ? 1 : 0); +} + /* Deterministic teardown of OS-resource handles, run once the program has * finished executing (the full value world is still alive, so buffered-message * decrefs are safe). Channels and thread handles live in the process handle @@ -4492,6 +4574,18 @@ void handle_table_drain(EigsState *st) { free(ch); st->handle_table[i].ptr = NULL; } + /* #408 tasks: cooperative, single-thread, no OS resource — just held + * refs. An outstanding task at exit (never joined, or the program ended + * with it still live) is reclaimed here. Its held entry_fn/args/result + * decref through task_free; the value world is still alive so this is + * safe, and it composes with the same leak-tally-0 gate as channels. */ + for (int i = 1; i < HANDLE_TABLE_SIZE; i++) { + if (st->handle_table[i].type != HANDLE_TASK) continue; + Task *t = (Task*)st->handle_table[i].ptr; + if (!t) continue; + task_free(t); + st->handle_table[i].ptr = NULL; + } /* Every spawned worker is now joined → the process is single-threaded * again. Clear the multithreaded flag so the cycle collector resumes: * collection was gated off during the MT window (it races mutators), but @@ -5415,6 +5509,8 @@ void register_builtins(Env *env) { /* ---- Concurrency builtins ---- */ env_set_local_owned(env, "spawn", make_builtin(builtin_spawn)); + env_set_local_owned(env, "task_spawn", make_builtin(builtin_task_spawn)); + env_set_local_owned(env, "task_alive", make_builtin(builtin_task_alive)); env_set_local_owned(env, "thread_join", make_builtin(builtin_thread_join)); env_set_local_owned(env, "channel", make_builtin(builtin_channel)); env_set_local_owned(env, "send", make_builtin(builtin_send)); diff --git a/src/eigenscript.h b/src/eigenscript.h index 13d3582..4fe96fc 100644 --- a/src/eigenscript.h +++ b/src/eigenscript.h @@ -429,7 +429,8 @@ struct EigsHttpServer; typedef enum { HANDLE_STORE, HANDLE_THREAD, - HANDLE_CHANNEL + HANDLE_CHANNEL, + HANDLE_TASK /* #408 cooperative task — id-keyed, drained at teardown */ } HandleType; typedef struct { diff --git a/src/vm.h b/src/vm.h index 888f1ce..ca1f4ef 100644 --- a/src/vm.h +++ b/src/vm.h @@ -367,6 +367,47 @@ typedef struct VM { struct EigsThread *owner; } VM; +/* ---- #408 cooperative task layer ------------------------------------ + * A Task is a reified VM context on the single OS thread — cooperatively + * scheduled, deterministic by construction (no tape records: interleaving + * is a pure function of program order). Increment 1a defines the full + * struct; the copying-stack save-buffer fields (saved_*) are populated by + * the suspend/resume surgery in increment 1b. Held refs (entry_fn/args/ + * result/error_value) keep their objects live under the trial-deletion + * cycle collector automatically — a counted ref exceeds the collector's + * internal edge count within U, exactly as ThreadHandle->fn does; no + * special root registration needed (docs/CLOSURE_CYCLE_GC.md). */ +typedef enum { + TASK_READY, /* runnable, not started or between yields */ + TASK_RUNNING, /* currently executing (== scheduler current) */ + TASK_SUSPENDED, /* yielded/blocked with a live save-buffer (1b) */ + TASK_DONE, /* returned normally */ + TASK_DEAD /* died of an uncaught error */ +} TaskState; + +typedef struct Task { + int id; /* == handle-table id; 1-based */ + TaskState state; + int started; /* 0 until first scheduled (1b) */ + Value *entry_fn; /* owned VAL_FN/VAL_BUILTIN to run */ + Value **args; /* owned, deep-copied at spawn */ + int argc; + /* Copying-stack save-buffer — valid only while TASK_SUSPENDED (1b). */ + EigsSlot *saved_stack; + int saved_stack_len; + struct CallFrame *saved_frames; + int saved_frame_count; + int saved_current_line; + /* Completion. */ + Value *result; /* owned, deep-copied on normal end */ + int has_error; /* died of an uncaught error */ + Value *error_value; /* owned {kind,message,line} dict */ +} Task; + +/* Free a Task's held refs and the struct. Safe on any state; called by + * handle_table_drain and (1b) by the scheduler on join/teardown. */ +void task_free(Task *t); + /* ---- Public API ---- */ /* Chunk lifecycle */ diff --git a/tests/run_all_tests.sh b/tests/run_all_tests.sh index c84eb10..4c76549 100755 --- a/tests/run_all_tests.sh +++ b/tests/run_all_tests.sh @@ -2140,6 +2140,10 @@ check_eigs_suite "recv-blocked worker doesn't hang exit" test_spawn_channel_exit echo "[104] Worker Arena Return (no cross-thread UAF, #302)" check_eigs_suite "worker arena return deep-copied before detach" test_spawn_arena_return.eigs "All tests passed" 1 +# #408 cooperative task layer — increment 1a: spawn + alive + leak-clean +# teardown of unrun tasks (the interleaving scheduler lands in 1b). +check_eigs_suite "cooperative tasks: spawn + alive + teardown (#408 inc 1a)" test_tasks.eigs "All tests passed" 1 + # [105] Builtin contract fixes (#312 negative indices, #316 predicate # type-rejection, #317 min/max N-ary reduction) + #314: a directory as the # script path must take the clean cannot-read-file exit, not xmalloc's diff --git a/tests/test_tasks.eigs b/tests/test_tasks.eigs new file mode 100644 index 0000000..654161b --- /dev/null +++ b/tests/test_tasks.eigs @@ -0,0 +1,48 @@ +# #408 cooperative task layer — increment 1a (spawn + alive + teardown). +# The copying-stack scheduler that runs/interleaves tasks lands in 1b; this +# pins the foundational surface: a spawned task exists, is alive, holds its +# closure + deep-copied args without leaking, and an unrun task is reclaimed +# deterministically at teardown (the leak-tally-0 gate covers that half). + +load_file of "lib/test.eigs" + +# --- spawn returns a numeric id; a fresh task is alive --- +define worker() as: + return 42 +tid is task_spawn of worker +assert_eq of [type of tid, "num", "task_spawn returns a numeric id"] +assert_eq of [task_alive of tid, 1, "a freshly spawned task is alive"] + +# --- unknown id is not alive (no crash, no false positive) --- +assert_eq of [task_alive of 99999, 0, "unknown task id is not alive"] +assert_eq of [task_alive of 0, 0, "id 0 (reserved/invalid) is not alive"] + +# --- args form: [fn, arg1, ...] spawns and stays alive --- +define adder(a, b) as: + return a + b +tid2 is task_spawn of [adder, 3, 4] +assert_eq of [task_alive of tid2, 1, "spawn with args is alive"] + +# --- distinct ids per spawn --- +assert_true of [tid2 != tid, "each spawn gets a distinct id"] + +# --- non-callable target is a structured type error, caught cleanly --- +caught is 0 +try: + junk is task_spawn of 5 +catch e: + caught is 1 + assert_eq of [e.kind, "type_mismatch", "spawn of a non-function is type_mismatch"] +assert_eq of [caught, 1, "spawn of a non-function raises"] + +# --- args are deep-copied at the boundary: mutating the caller's value +# after spawn cannot be observed as a shared reference here (1a just +# holds the copy; 1b will prove it on run). We at least prove spawn +# accepts a compound arg and stays alive without aliasing the caller. --- +shared is [1, 2, 3] +tid3 is task_spawn of [adder, shared, 10] +shared is append of [shared, 4] # mutate caller's list post-spawn +assert_eq of [task_alive of tid3, 1, "spawn with a compound arg is alive"] +assert_eq of [len of shared, 4, "caller's list still mutable post-spawn"] + +test_summary of null