diff --git a/CHANGELOG.md b/CHANGELOG.md index a076dd5..e914d94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,20 @@ All notable changes to EigenScript are documented here. ## [Unreleased] ### Added +- **Deterministic cooperative tasks — increment 1 (#408).** `task_spawn` + / `task_yield` / `task_join` / `task_alive`: a single-OS-thread + cooperative scheduler whose interleaving is deterministic **by + construction** — no trace records, byte-identical across runs. Tasks + are reified VM contexts (a copying-stack save-buffer per suspended + task, not a 1.28 MB VM each); the JIT and non-atomic refcount fast + paths stay live because the scheduler never flips the multithreaded + flag. Args and results cross the boundary deep-copied (share-nothing, + like channels). `task_join` returns the worker's result or re-raises + its uncaught error as the same `{kind, message, line}`. All-tasks- + blocked is a new `deadlock` error kind (loud, not a hang); suspending + inside an `arena_mark`…`arena_reset` scope or a nested evaluation is a + `value` error. Not yet: mailboxes/`task_send` (increment 2), virtual + time, the pluggable seeded-strategy hook for the DST consumer. - **E003 is scope-precise (#404, increment two)** — the undefined-name pass models the runtime's real scope rules instead of a whole-file binding set: function-locals (fresh-name `is` and `local`) are diff --git a/docs/BUILTINS.md b/docs/BUILTINS.md index 67edc60..9a80588 100644 --- a/docs/BUILTINS.md +++ b/docs/BUILTINS.md @@ -570,6 +570,8 @@ Requires full build. Transformer model inference and training. | `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). | +| `task_yield` | `task_yield of null` | Cooperatively hand control to the next ready task; this task resumes round-robin. A no-op when no task has been spawned. Forbidden inside an `arena_mark`…`arena_reset` scope or a nested evaluation (raises `value`). | +| `task_join` | `task_join of id` | Block until task `id` finishes, then return its deep-copied result — or re-raise its uncaught error (as the same `{kind, message, line}` it died with). Joining an already-finished task returns immediately; an unknown id (or self) returns null. All tasks blocked with none runnable is a `deadlock` error, not a hang. | **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/docs/DIAGNOSTICS.md b/docs/DIAGNOSTICS.md index c81502e..aab5102 100644 --- a/docs/DIAGNOSTICS.md +++ b/docs/DIAGNOSTICS.md @@ -67,6 +67,7 @@ below are the contract.) | `sandbox` | sandbox policy denial or budget | `blocked in sandbox`, `sandbox memory budget exceeded` | | `interrupt` | host-requested abort (`eigs_abort`) | `aborted` | | `assert` | `assert` builtin failure | `ASSERT FAIL: ...` | +| `deadlock` | #408 all cooperative tasks blocked, none runnable | `all tasks are blocked — deadlock` | | `internal` | VM invariant broke — report it | `unknown opcode`, `SET_LOCAL slot out of range` | `throw` stamps kind `user` for host/embed introspection diff --git a/examples/task_pipeline.eigs b/examples/task_pipeline.eigs new file mode 100644 index 0000000..7b5ec9e --- /dev/null +++ b/examples/task_pipeline.eigs @@ -0,0 +1,23 @@ +# Deterministic cooperative tasks (#408, increment 1). Three workers each +# compute a stage, cooperatively yielding so their progress interleaves — +# but the interleaving is a pure function of program order, so this program +# prints byte-identically on every run (no threads, no clock, no tape). +# `task_join` collects each worker's result; main sums them. + +define stage(name, base) as: + print of f"{name}: begin" + task_yield of null # cooperatively let a sibling run + local doubled is base * 2 + print of f"{name}: computed {doubled}" + return doubled + +a is task_spawn of [stage, "alpha", 10] +b is task_spawn of [stage, "beta", 20] +c is task_spawn of [stage, "gamma", 30] + +print of "main: workers spawned, collecting" +ra is task_join of a +rb is task_join of b +rc is task_join of c + +print of f"results: {ra} + {rb} + {rc} = {ra + rb + rc}" diff --git a/src/builtins.c b/src/builtins.c index f30fdf9..cd75b23 100644 --- a/src/builtins.c +++ b/src/builtins.c @@ -4449,10 +4449,25 @@ void task_free(Task *t) { if (t->args[i]) val_decref(t->args[i]); free(t->args); } + if (t->run_env) env_decref(t->run_env); 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); + /* A task torn down while still SUSPENDED (kill-outstanding at main's end, + * or an unjoined task at exit) still owns the counted refs sitting in its + * saved slice — decref them so the leak tally stays 0. */ + if (t->saved_stack) { + for (int i = 0; i < t->saved_stack_len; i++) + slot_decref(t->saved_stack[i]); + free(t->saved_stack); + } + if (t->saved_frames) { + for (int i = 0; i < t->saved_frame_count; i++) { + CallFrame *f = &t->saved_frames[i]; + if (f->owns_env && f->env) env_decref(f->env); + if (f->chunk) chunk_decref(f->chunk); + } + free(t->saved_frames); + } free(t); } @@ -4496,9 +4511,63 @@ Value* builtin_task_spawn(Value *arg) { return make_null(); } t->id = id; + task_sched_on_spawn(id); /* enqueue + arm the scheduler */ return make_num((double)id); } +/* task_yield of null — cooperatively hand control to the next ready task. + * Returns null (the value of the yield expression); the scheduler saves and + * re-enqueues this task at the tail. Forbidden inside an active arena scope + * (arena_mark…arena_reset): a suspended task's arena values would be reclaimed + * by another task's arena_reset — the task layer and the training arena are + * separate tools (Lua-style "can't yield across a C boundary"). */ +Value* builtin_task_yield(Value *arg) { + (void)arg; + if (!g_task_sched) return make_null(); /* no scheduler: yield is a no-op */ + if (g_arena.active) { + rt_error(EK_VALUE, 0, "cannot task_yield inside an arena scope " + "(arena_mark…arena_reset)"); + return make_null(); + } + task_request_yield(); + return make_null(); /* placeholder: execution resumes right after this */ +} + +/* task_join of id — block until task `id` finishes; return its (deep-copied) + * result, or re-raise its uncaught error. Joining an already-finished task + * returns immediately. Joining an unknown id, self, or with no scheduler + * returns null. Same arena/nesting restriction as task_yield. */ +Value* builtin_task_join(Value *arg) { + if (!arg || arg->type != VAL_NUM || !g_task_sched) return make_null(); + int target = (int)arg->data.num; + Task *t = (Task*)handle_lookup(target, HANDLE_TASK); + if (!t) return make_null(); + if (t->state == TASK_DONE || t->state == TASK_DEAD) { + /* Already finished: deliver its result / error now, no suspend. */ + if (t->has_error) { + if (t->error_value) { + val_incref(t->error_value); + eigs_clear_error_value(); + g_error_value = t->error_value; + g_error_kind = (int)EK_USER; + } + snprintf(g_error_msg, sizeof(g_error_msg), "joined task %d failed", target); + g_has_error = 1; + return make_null(); + } + Value *r = t->result ? t->result : make_null(); + val_incref(r); + return r; + } + if (g_arena.active) { + rt_error(EK_VALUE, 0, "cannot task_join inside an arena scope " + "(arena_mark…arena_reset)"); + return make_null(); + } + if (!task_request_join(target)) return make_null(); + return make_null(); /* placeholder: the scheduler fills it with the result on resume */ +} + /* 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) { @@ -5511,6 +5580,8 @@ void register_builtins(Env *env) { 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, "task_yield", make_builtin(builtin_task_yield)); + env_set_local_owned(env, "task_join", make_builtin(builtin_task_join)); 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.c b/src/eigenscript.c index 92032b3..119048e 100644 --- a/src/eigenscript.c +++ b/src/eigenscript.c @@ -87,6 +87,7 @@ const char* err_kind_name(ErrKind k) { case EK_SANDBOX: return "sandbox"; case EK_INTERRUPT: return "interrupt"; case EK_ASSERT: return "assert"; + case EK_DEADLOCK: return "deadlock"; case EK_USER: return "user"; } return "internal"; diff --git a/src/eigenscript.h b/src/eigenscript.h index 4fe96fc..9cc2e9e 100644 --- a/src/eigenscript.h +++ b/src/eigenscript.h @@ -624,6 +624,11 @@ struct EigsThread { int vts_depth; int json_depth; int native_call_depth; + /* #408 cooperative task scheduler — opaque (TaskScheduler defined in + * vm.c; Task lives in vm.h, not visible here). Allocated lazily on the + * first task_spawn, freed at thread detach. Per-OS-thread because tasks + * are single-threaded by construction. */ + void *task_sched; /* Registry list — set by eigs_thread_attach. */ EigsThread *next; }; @@ -695,6 +700,7 @@ extern __thread EigsThread *eigs_current; #define g_vts_depth (eigs_current->vts_depth) #define g_json_depth (eigs_current->json_depth) #define g_native_call_depth (eigs_current->native_call_depth) +#define g_task_sched (eigs_current->task_sched) #define g_entry_threshold (eigs_current->state->jit_entry_threshold) #define g_iter_threshold (eigs_current->state->jit_iter_threshold) #define g_osr_threshold (eigs_current->state->jit_osr_threshold) @@ -984,6 +990,7 @@ typedef enum { EK_SANDBOX, /* sandbox policy denial or budget exhaustion */ EK_INTERRUPT, /* host-requested abort (eigs_abort) */ EK_ASSERT, /* assert builtin failure */ + EK_DEADLOCK, /* #408 all cooperative tasks blocked, none runnable */ EK_USER, /* `throw` — catch binds the thrown value, not a dict */ } ErrKind; const char* err_kind_name(ErrKind k); diff --git a/src/state.c b/src/state.c index 0be812f..d43ec31 100644 --- a/src/state.c +++ b/src/state.c @@ -139,6 +139,7 @@ void eigs_thread_detach(void) { * so any bridge-macro reads inside the destructors stay valid). */ jit_thread_destroy(th); vm_thread_destroy(th); + task_sched_thread_free(); /* #408: release the cooperative task scheduler */ /* Phase 8: release freelist + intern memory before the EigsThread * struct itself goes. Must run while eigs_current still points at th diff --git a/src/vm.c b/src/vm.c index 4a009ef..7f279b1 100644 --- a/src/vm.c +++ b/src/vm.c @@ -2056,8 +2056,43 @@ static const char *slot_type_name(EigsSlot s) { return "num"; /* immediate double / bool */ } -static Value *vm_run(EigsChunk *chunk, Env *env) { - int base_frame = g_vm.frame_count; /* track entry point for re-entrant returns */ +/* #408: forward decls — the copying-stack save/restore and the "who is + * running" helper live with the scheduler below vm_execute; vm_run_ex's + * resume/suspend paths call them. */ +static void task_restore_slice(Task *t); +static void task_save_slice(Task *t); +static Task *task_current_running(void); +static void task_apply_join_result(Task *t); /* fill a join placeholder on resume */ + +/* vm_run_ex: the shared VM dispatch body. `resume` != NULL means resume a + * suspended #408 task — restore its copying-stack slice onto the (empty) VM + * and continue from the innermost frame's saved ip, instead of pushing a + * fresh base frame. vm_run() is the ordinary fresh-entry wrapper. */ +static Value *vm_run_ex(EigsChunk *chunk, Env *env, Task *resume) { + int base_frame; + CallFrame *frame; + uint8_t *ip; + int current_line = 0; + + if (resume) { + /* Resume: the scheduler only ever resumes a task at the outermost + * level, so base_frame is 0. Restore frames+stack, then continue + * from where task_yield/task_join left off (its placeholder result + * already sits on the restored stack top). */ + base_frame = 0; + task_restore_slice(resume); + /* If this task was blocked in task_join, its joinee has finished: + * overwrite the placeholder on the stack top with the joinee's result + * (or re-raise its error — CHECK_ERROR at vm_resume_dispatch catches). */ + task_apply_join_result(resume); + frame = &g_vm.frames[g_vm.frame_count - 1]; + chunk = frame->chunk; + ip = frame->ip; + current_line = g_vm.current_line; + goto vm_resume_dispatch; + } + + base_frame = g_vm.frame_count; /* track entry point for re-entrant returns */ /* Module-level slot promotion (Part B) AND nested entries from builtins * (e.g. call_eigs_fn, eval): if the chunk allocated slots past env->count * at compile time, reserve them now so OP_SET_LOCAL has a place to write. @@ -2065,7 +2100,7 @@ static Value *vm_run(EigsChunk *chunk, Env *env) { if (chunk->local_count > env->count) { env_reserve_slots(env, chunk->local_count); } - CallFrame *frame = &g_vm.frames[g_vm.frame_count++]; + frame = &g_vm.frames[g_vm.frame_count++]; frame->chunk = chunk; chunk_incref(chunk); /* frame's ref — released when this frame pops */ frame->ip = chunk->code; @@ -2088,14 +2123,16 @@ static Value *vm_run(EigsChunk *chunk, Env *env) { /* #297: JIT hotness bookkeeping is shared per-chunk state and feeds JIT * compilation, which is gated off under MT (#296). Skip it while * multithreaded — pointless work, and the bare ++ / registry write race - * with parallel workers running the same chunk. */ - if (!g_vm_multithreaded) { + * with parallel workers running the same chunk. #408: also skip while a + * cooperative task scheduler is active — task code runs interpreted so a + * JIT thunk never runs mid-suspend (the "non-JIT suspending code" ruling, + * coarsened for v1: V8 shipped un-optimized generators for years). */ + if (!g_vm_multithreaded && !g_task_sched) { chunk->exec_count++; jit_register_chunk(chunk); } - uint8_t *ip = frame->ip; - int current_line = 0; + ip = frame->ip; #ifdef __GNUC__ /* Computed goto dispatch table */ @@ -2212,6 +2249,7 @@ static Value *vm_run(EigsChunk *chunk, Env *env) { for (;;) { switch (*ip++) { #endif +vm_resume_dispatch: /* #408 resume lands here: ip/frame/chunk restored above */ DISPATCH(); /* ---- Constants ---- */ @@ -3081,6 +3119,20 @@ static Value *vm_run(EigsChunk *chunk, Env *env) { /* Check for errors from builtins */ CHECK_ERROR(); + /* #408: a suspending builtin (task_yield/task_join) sets the + * request flag and leaves its placeholder result on the stack. + * Suspension is only legal at the outermost level (base_frame 0); + * inside a nested evaluation (eval/comparator/import — base_frame + * > 0) the C stack can't be captured, so raise instead. On a valid + * suspend, ip already points past this CALL, so the saved frame + * resumes with the placeholder on top. */ + if (__builtin_expect(g_task_suspend_request, 0)) { + if (base_frame == 0) { frame->ip = ip; goto vm_suspend_halt; } + g_task_suspend_request = 0; + rt_error(EK_VALUE, current_line, + "cannot suspend (task_yield/task_join) inside a nested evaluation"); + CHECK_ERROR(); + } DISPATCH(); } @@ -5029,13 +5081,330 @@ static Value *vm_run(EigsChunk *chunk, Env *env) { g_vm.frame_count--; } return make_null(); + +vm_suspend_halt: + /* #408 suspend: the OPPOSITE of vm_error_halt — do NOT drain the frames, + * we are preserving them. Sync the live current_line, then hand the whole + * live slice [0, frame_count)/[0, sp) to the copying-stack save-buffer and + * retreat the VM to empty so the scheduler can run the next task at base 0. + * Ownership moves with the memcpy'd bytes; g_task_suspend_request is + * cleared by the save. Returns NULL — the scheduler distinguishes a + * suspend from a real result via the task's state (TASK_SUSPENDED). */ + g_vm.current_line = current_line; + task_save_slice(task_current_running()); + g_task_suspend_request = 0; + return NULL; +} + +/* Ordinary fresh-entry wrapper — the common path for every non-task call. */ +static Value *vm_run(EigsChunk *chunk, Env *env) { + return vm_run_ex(chunk, env, NULL); } /* ---- Public API ---- */ +/* ===== #408 cooperative task scheduler ================================== + * A trampoline just above the OUTERMOST vm_execute drives every task — + * including task 0 (the main program) — so C-stack depth stays flat + * (vm_execute → scheduler → one vm_run) no matter how often tasks ping-pong. + * A task suspends by a builtin setting g_task_suspend_request; the CASE(CALL) + * site saves its live stack+frame slice (the copying-stack model: memory = + * live depth, not a full 1.28 MB VM per task) and returns here, which runs + * the next ready task. Deterministic by construction — no tape records; the + * interleaving is a pure function of program order. + * ======================================================================== */ + +int g_task_suspend_request = 0; + +#define TASK_READY_MAX HANDLE_TABLE_SIZE + +typedef struct { + int ready[TASK_READY_MAX]; /* circular FIFO of runnable task ids (0=main) */ + int rhead, rcount; + int current; /* running task id; 0 = main */ + int live; /* spawned tasks not yet DONE/DEAD */ + int active; /* armed on first spawn */ + Task main_task; /* task 0 — save-buffer only, never "started" */ +} TaskScheduler; + +static TaskScheduler *sched_get(void) { return (TaskScheduler *)g_task_sched; } + +static TaskScheduler *sched_ensure(void) { + TaskScheduler *s = sched_get(); + if (!s) { + s = xcalloc(1, sizeof(TaskScheduler)); + s->main_task.id = 0; + s->main_task.state = TASK_RUNNING; + s->current = 0; + g_task_sched = s; + } + return s; +} + +void task_sched_thread_free(void) { + TaskScheduler *s = sched_get(); + if (!s) return; + free(s->main_task.saved_stack); + free(s->main_task.saved_frames); + free(s); + g_task_sched = NULL; +} + +static Task *sched_lookup(TaskScheduler *s, int id) { + if (id == 0) return &s->main_task; + return (Task *)handle_lookup(id, HANDLE_TASK); +} + +static Task *task_current_running(void) { + TaskScheduler *s = sched_get(); + return s ? sched_lookup(s, s->current) : NULL; +} + +static void sched_ready_push(TaskScheduler *s, int id) { + if (s->rcount >= TASK_READY_MAX) return; /* ids are table-bounded; can't overflow */ + s->ready[(s->rhead + s->rcount) % TASK_READY_MAX] = id; + s->rcount++; +} + +static int sched_ready_pop(TaskScheduler *s) { + if (s->rcount == 0) return -1; + int id = s->ready[s->rhead]; + s->rhead = (s->rhead + 1) % TASK_READY_MAX; + s->rcount--; + return id; +} + +/* Copying-stack save: memcpy the running task's live slice [0,fc)/[0,sp) into + * its right-sized save-buffer, then retreat the VM to empty. Refs move WITH + * the bytes (the saved slots/frames own the same counted refs that were on + * the stack), so sp/frame_count just retreat — no incref/decref, no walk. */ +static void task_save_slice(Task *t) { + if (!t) return; + int fc = g_vm.frame_count, sp = g_vm.sp; + free(t->saved_frames); + free(t->saved_stack); + t->saved_frames = fc ? xmalloc(sizeof(CallFrame) * fc) : NULL; + t->saved_stack = sp ? xmalloc(sizeof(EigsSlot) * sp) : NULL; + if (fc) memcpy(t->saved_frames, g_vm.frames, sizeof(CallFrame) * fc); + if (sp) memcpy(t->saved_stack, g_vm.stack, sizeof(EigsSlot) * sp); + t->saved_frame_count = fc; + t->saved_stack_len = sp; + t->saved_current_line = g_vm.current_line; + g_vm.frame_count = 0; + g_vm.sp = 0; + t->state = TASK_SUSPENDED; +} + +/* Copying-stack restore: memcpy a suspended task's slice back onto the empty + * VM. Symmetric with save — refs move back with the bytes. */ +static void task_restore_slice(Task *t) { + int fc = t->saved_frame_count, sp = t->saved_stack_len; + if (fc) memcpy(g_vm.frames, t->saved_frames, sizeof(CallFrame) * fc); + if (sp) memcpy(g_vm.stack, t->saved_stack, sizeof(EigsSlot) * sp); + g_vm.frame_count = fc; + g_vm.sp = sp; + g_vm.current_line = t->saved_current_line; + free(t->saved_frames); t->saved_frames = NULL; + free(t->saved_stack); t->saved_stack = NULL; + t->saved_frame_count = 0; + t->saved_stack_len = 0; + t->state = TASK_RUNNING; +} + +/* task_spawn (builtins.c) hands a freshly registered task here. */ +void task_sched_on_spawn(int id) { + TaskScheduler *s = sched_ensure(); + s->active = 1; + s->live++; + sched_ready_push(s, id); +} + +/* task_yield: mark the current task for suspension; the trampoline re-enqueues + * it at the tail (round-robin) after the save. */ +void task_request_yield(void) { g_task_suspend_request = 1; } + +/* task_join: block the current task on `target`. Returns 0 for a bad target + * (main, self, or unknown) so the builtin can fall back; 1 to suspend. A + * target that is already finished is handled in the builtin (returns its + * result without suspending). */ +int task_request_join(int target) { + TaskScheduler *s = sched_get(); + if (!s || target == 0 || target == s->current) return 0; + Task *tt = sched_lookup(s, target); + if (!tt) return 0; + Task *cur = sched_lookup(s, s->current); + cur->join_target = target; + g_task_suspend_request = 1; + return 1; +} + +/* Start a never-run spawned task: bind its deep-copied args into a fresh env + * from the entry closure (the base frame borrows it — the Task owns run_env + * across suspend/resume), then run at base 0 so it is suspendable. Mirrors + * call_eigs_fn's param binding, but does not run to completion. */ +static Value *task_start(Task *t) { + Value *fn = t->entry_fn; + if (fn->type == VAL_BUILTIN) { /* builtins never suspend — run direct */ + Value *a = t->argc == 1 ? t->args[0] : make_null(); + return fn->data.builtin(a); + } + Env *call_env = env_new(fn->data.fn.closure); + for (int i = 0; i < fn->data.fn.param_count && i < t->argc; i++) + env_set_local(call_env, fn->data.fn.params[i], t->args[i]); + t->run_env = call_env; /* Task owns it; base frame borrows */ + t->started = 1; + EigsChunk *chunk = (EigsChunk *)fn->data.fn.body; + return vm_run_ex(chunk, call_env, NULL); +} + +/* Record a task that just finished (returned or errored) and wake any joiner + * blocked on it. `r` is the value vm_run returned (NULL on suspend — not this + * path). g_has_error distinguishes a normal end from an uncaught error. */ +static void sched_finish(TaskScheduler *s, Task *t, Value *r) { + if (g_has_error) { + t->has_error = 1; + t->error_value = vm_take_error_value(); /* the {kind,message,line} dict */ + g_has_error = 0; + if (r) val_decref(r); + t->result = NULL; + } else { + t->result = r ? val_clone_for_send(r) : NULL; /* share-nothing result */ + if (r) val_decref(r); + } + t->state = t->has_error ? TASK_DEAD : TASK_DONE; + if (t->id != 0) s->live--; + if (t->run_env) { env_decref(t->run_env); t->run_env = NULL; } + /* Wake every task blocked on this one: enqueue it; on resume the join + * builtin's placeholder gets overwritten with our result (or re-raise). */ + for (int i = 1; i < HANDLE_TABLE_SIZE; i++) { + Task *w = (Task *)handle_lookup(i, HANDLE_TASK); + if (w && w->state == TASK_SUSPENDED && w->join_target == t->id) + sched_ready_push(s, w->id); + } + if (s->main_task.state == TASK_SUSPENDED && s->main_task.join_target == t->id) + sched_ready_push(s, 0); +} + +/* On resuming a task that was blocked in task_join, replace the placeholder + * null the builtin left on the stack top with the joinee's result — or, if + * the joinee died, re-raise its error in the joiner. Called from vm_run_ex's + * resume path (after the stack is restored). */ +static void task_apply_join_result(Task *t) { + TaskScheduler *s = sched_get(); + if (!s || t->join_target == 0) return; + Task *jt = sched_lookup(s, t->join_target); + t->join_target = 0; + if (!jt) return; + if (jt->has_error) { + /* Re-raise: restore the error payload so the joiner's CHECK_ERROR + * catches/propagates it as if the throw happened at the join. */ + if (jt->error_value) { + g_error_value = jt->error_value; + val_incref(g_error_value); + g_error_kind = (int)EK_USER; + } + snprintf(g_error_msg, sizeof(g_error_msg), "joined task %d failed", jt->id); + g_has_error = 1; + return; + } + /* Overwrite TOS placeholder with the joinee's (already deep-copied) result. */ + if (g_vm.sp > 0) { + slot_decref(g_vm.stack[g_vm.sp - 1]); + Value *res = jt->result ? jt->result : make_null(); + val_incref(res); + g_vm.stack[g_vm.sp - 1] = slot_from_heap(res); + } +} + +/* The trampoline. Entered from the outermost vm_execute once main (task 0) + * has first suspended. Drives tasks round-robin until the ready queue drains, + * then returns main's result. All-tasks-blocked = deadlock (loud, not a hang). + * Task 0 finishing kills outstanding tasks (kill-outstanding ruling). */ +static Value *scheduler_trampoline(TaskScheduler *s) { + for (;;) { + int id = sched_ready_pop(s); + if (id < 0) { + /* Nothing runnable. If main already finished, we're done. If tasks + * remain live (blocked on joins that can't resolve), that's a + * deadlock — raise it loudly rather than hang. */ + if (s->main_task.state == TASK_DONE || s->main_task.state == TASK_DEAD) { + if (s->main_task.has_error) { + g_error_value = s->main_task.error_value; + s->main_task.error_value = NULL; + g_has_error = 1; + return make_null(); + } + Value *r = s->main_task.result; + s->main_task.result = NULL; + return r ? r : make_null(); + } + rt_error(EK_DEADLOCK, 0, "all tasks are blocked — deadlock"); + return make_null(); + } + Task *t = sched_lookup(s, id); + if (!t || t->state == TASK_DONE || t->state == TASK_DEAD) continue; + s->current = id; + + Value *r; + if (t->state == TASK_SUSPENDED) { + /* Resume (the join placeholder, if any, is filled inside the + * resume path via task_apply_join_result). */ + r = vm_run_ex(NULL, NULL, t); + } else { + r = task_start(t); /* never-run task: bind args + run at base 0 */ + } + + if (t->state == TASK_SUSPENDED) { + /* It suspended again. task_yield → re-enqueue; task_join → stay + * blocked (woken by sched_finish when its target ends). */ + if (t->join_target == 0) sched_ready_push(s, id); + } else { + sched_finish(s, t, r); + /* kill-outstanding: main ending tears the rest down deterministically. */ + if (id == 0) break; + } + } + /* main finished with tasks still outstanding → reap them (kill-outstanding). */ + if (s->main_task.has_error) { + g_error_value = s->main_task.error_value; + s->main_task.error_value = NULL; + g_has_error = 1; + return make_null(); + } + Value *r = s->main_task.result; + s->main_task.result = NULL; + return r ? r : make_null(); +} + Value *vm_execute(EigsChunk *chunk, Env *env) { vm_init(); - /* Re-entrant safe: vm_run pushes/pops its own frame and cleans - * the stack back to its base pointer on return. */ - return vm_run(chunk, env); + /* Only the OUTERMOST vm_execute drives the scheduler; a nested call + * (eval/dispatch/import/comparator) runs to completion on the C stack and + * may not suspend (enforced at CASE(CALL) via base_frame). frame_count==0 + * identifies the outermost. */ + int outermost = (g_vm.frame_count == 0); + Value *r = vm_run(chunk, env); + if (!outermost) return r; + TaskScheduler *s = sched_get(); + if (!s || !s->active) return r; + /* main (task 0) either finished (no task ever blocked) or suspended. If it + * suspended, its slice is saved; drive the trampoline. If it finished but + * tasks are still live, drive them too (kill-outstanding at main's end). */ + if (s->main_task.state == TASK_SUSPENDED) { + /* main's first suspension happened in the initial vm_run, outside the + * trampoline — enqueue it now (unless it blocked on a join, in which + * case sched_finish wakes it) so it resumes round-robin. */ + if (s->main_task.join_target == 0) sched_ready_push(s, 0); + return scheduler_trampoline(s); + } + /* main ran to completion without ever suspending. Record its result and, + * if any spawned task is still runnable, drive them (kill-outstanding). */ + if (s->live > 0 && s->rcount > 0) { + s->main_task.state = TASK_DONE; + s->main_task.result = r ? val_clone_for_send(r) : NULL; + if (r) val_decref(r); + Value *mr = scheduler_trampoline(s); + return mr; + } + return r; } diff --git a/src/vm.h b/src/vm.h index ca1f4ef..db95c80 100644 --- a/src/vm.h +++ b/src/vm.h @@ -392,12 +392,23 @@ typedef struct Task { Value *entry_fn; /* owned VAL_FN/VAL_BUILTIN to run */ Value **args; /* owned, deep-copied at spawn */ int argc; + /* The task's base call env (1b), created at start from entry_fn's + * closure with args bound. The base frame BORROWS it (owns_env=0), so + * the Task owns it across suspend/resume; decref'd when the task ends + * (or in task_free). NULL for main (task 0 runs on the global env). */ + struct Env *run_env; /* Copying-stack save-buffer — valid only while TASK_SUSPENDED (1b). */ EigsSlot *saved_stack; int saved_stack_len; - struct CallFrame *saved_frames; + CallFrame *saved_frames; int saved_frame_count; int saved_current_line; + /* Blocking join (1b): while this task is suspended in task_join, the id + * of the task it waits on (0 = not joining). On resume the scheduler + * writes the joinee's deep-copied result over the placeholder the + * task_join builtin left on the stack top — a builtin can't return a + * value it doesn't know yet — or re-raises the joinee's error. */ + int join_target; /* Completion. */ Value *result; /* owned, deep-copied on normal end */ int has_error; /* died of an uncaught error */ @@ -408,6 +419,17 @@ typedef struct Task { * handle_table_drain and (1b) by the scheduler on join/teardown. */ void task_free(Task *t); +/* #408 scheduler interface (1b). The task_* builtins (builtins.c) signal the + * cooperative scheduler through these; the trampoline lives in vm.c, just + * above the outermost vm_execute, so C-stack depth stays flat across any + * number of task switches. Suspension is deterministic-by-construction — no + * tape records. */ +extern int g_task_suspend_request; /* set by a suspending builtin; actioned at CASE(CALL) */ +void task_sched_on_spawn(int id); /* enqueue a freshly spawned task, arm the scheduler */ +void task_request_yield(void); /* current task → tail of the ready queue */ +int task_request_join(int target); /* current task blocks on `target`; 0 = bad target */ +void task_sched_thread_free(void); /* release the scheduler at thread detach */ + /* ---- Public API ---- */ /* Chunk lifecycle */ diff --git a/tests/test_tasks.eigs b/tests/test_tasks.eigs index 654161b..b122993 100644 --- a/tests/test_tasks.eigs +++ b/tests/test_tasks.eigs @@ -39,10 +39,75 @@ assert_eq of [caught, 1, "spawn of a non-function raises"] # 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. --- +define lister(xs) as: + return len of xs shared is [1, 2, 3] -tid3 is task_spawn of [adder, shared, 10] +tid3 is task_spawn of [lister, shared] 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"] +# the task saw the deep copy (len 3), not the caller's later-mutated list (len 4) +assert_eq of [task_join of tid3, 3, "task got the deep-copied arg, not the mutation"] + +# ============ increment 1b: yield / join / deadlock ============ + +# --- deterministic round-robin interleaving. Tasks append to a shared +# module list; the final order is a pure function of spawn+yield order. --- +trace is [] +define stepper(tag) as: + trace is append of [trace, f"{tag}1"] + task_yield of null + trace is append of [trace, f"{tag}2"] + return 0 +sa is task_spawn of [stepper, "a"] +sb is task_spawn of [stepper, "b"] +ja is task_join of sa +jb is task_join of sb +# main joins sa (blocks) → a runs a1,yield; b runs b1,yield; a a2 done; b b2 done +assert_eq of [trace[0], "a1", "interleave: a1 first"] +assert_eq of [trace[1], "b1", "interleave: b1 second"] +assert_eq of [trace[2], "a2", "interleave: a2 third"] +assert_eq of [trace[3], "b2", "interleave: b2 fourth"] + +# --- task_join delivers the (deep-copied) return value --- +define doubler(n) as: + task_yield of null + return n * 2 +d1 is task_spawn of [doubler, 21] +r1 is task_join of d1 +assert_eq of [r1, 42, "task_join returns the worker result"] + +# --- joining an already-finished task returns its result immediately --- +r1b is task_join of d1 +assert_eq of [r1b, 42, "re-join of a finished task returns the result"] +assert_eq of [task_alive of d1, 0, "a finished task is not alive"] + +# --- an uncaught error in a task is rethrown at the join, as its #406 kind --- +define crasher() as: + task_yield of null + boom is undefined_name_xyz + return 0 +c1 is task_spawn of crasher +threw is 0 +try: + rc is task_join of c1 +catch e: + threw is 1 + assert_eq of [e.kind, "undefined_name", "join rethrows the task's error kind"] +assert_eq of [threw, 1, "join of a crashed task raises"] + +# --- suspension inside an arena scope is a structured value-error --- +define arena_yielder() as: + arena_mark of null + task_yield of null + return 0 +ay is task_spawn of arena_yielder +threw2 is 0 +try: + ra is task_join of ay +catch e: + threw2 is 1 + assert_eq of [e.kind, "value", "task_yield in an arena scope is a value-error"] +assert_eq of [threw2, 1, "arena-scoped yield raises"] test_summary of null