diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e70c5c..fe841bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,20 @@ All notable changes to EigenScript are documented here. ## [Unreleased] ### Added +- **Cooperative task mailboxes — increment 2 (#408).** `task_send` / + `task_recv` / `task_try_recv` / `task_kill`: unbounded FIFO mailboxes + (Erlang-style; bounded/backpressure is a later add) with messages + deep-copied across the boundary (share-nothing, like channels). + `task_recv` blocks cooperatively on an empty mailbox — reusing the + increment-1 suspend/resume machinery, delivered on wake — while + `task_try_recv` is the non-blocking form. Sending to a finished or + unknown task is a **silent drop** returning 0 (Akka dead-letters / + Erlang cast — an error path here would be a nondeterminism magnet), + not an error. `task_kill` tears a task down deterministically and + wakes any joiner with an `interrupt` error. Still deterministic by + construction (byte-identical across runs). Not yet: virtual time + (`task_sleep`) and the pluggable seeded-strategy hook for the DST + consumer. - **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 diff --git a/docs/BUILTINS.md b/docs/BUILTINS.md index 9a80588..a1d2efb 100644 --- a/docs/BUILTINS.md +++ b/docs/BUILTINS.md @@ -572,6 +572,10 @@ Requires full build. Transformer model inference and training. | `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. | +| `task_send` | `task_send of [id, value]` | Append a deep-copied message to task `id`'s unbounded FIFO mailbox, waking it if it waits in `task_recv`. Returns 1 if delivered, 0 if `id` is finished/unknown (a silent drop — send-to-dead is never an error). Never blocks. | +| `task_recv` | `task_recv of null` | Return the next message from this task's mailbox, or block cooperatively until one arrives. Forbidden inside an `arena_mark`…`arena_reset` scope or a nested evaluation (raises `value`). | +| `task_try_recv` | `task_try_recv of null` | Non-blocking receive: the next mailbox message, or `null` if empty. Never suspends. | +| `task_kill` | `task_kill of id` | Tear down task `id`: drop its mailbox, mark it dead, wake any joiner with an `interrupt` error. Returns 1 if killed, 0 for a finished/unknown/self target. | **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/SPEC.md b/docs/SPEC.md index cde8de1..ce66ca4 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -1167,6 +1167,28 @@ When the main program returns, any tasks still running are torn down example two tasks each `task_join`-ing the other — that is a `deadlock` error, reported loudly rather than hanging. +Tasks communicate through **mailboxes**. `task_send of [id, value]` +appends a deep-copied message to task `id`'s FIFO mailbox (share- +nothing, like channel sends); `task_recv of null` returns the next +message or blocks cooperatively until one arrives. `task_try_recv` +is the non-blocking form. + +```eigenscript +worker_id is 0 +define worker() as: + a is task_recv of null + b is task_recv of null + return a + b + +worker_id is task_spawn of worker +task_send of [worker_id, 10] +task_send of [worker_id, 32] +print of (task_join of worker_id) +``` +```output +42 +``` + ## Buffers `buffer of count` allocates a flat array of `count` nums (all 0). diff --git a/src/builtins.c b/src/builtins.c index cd75b23..8a1facf 100644 --- a/src/builtins.c +++ b/src/builtins.c @@ -4452,6 +4452,12 @@ void task_free(Task *t) { 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); + /* inc 2: drain any undelivered mailbox messages. */ + if (t->mbox) { + for (int i = 0; i < t->mbox_count; i++) + val_decref(t->mbox[(t->mbox_head + i) % t->mbox_cap]); + free(t->mbox); + } /* 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. */ @@ -4568,6 +4574,54 @@ Value* builtin_task_join(Value *arg) { return make_null(); /* placeholder: the scheduler fills it with the result on resume */ } +/* task_send of [id, value] — append a deep-copied message to task `id`'s + * unbounded FIFO mailbox and wake it if it is waiting in task_recv. Sending to + * a finished or unknown task is a silent drop (counted as a dead letter), not + * an error — Akka dead-letters / Erlang cast. Returns 1 if delivered, 0 if + * dropped. Never blocks (the mailbox is unbounded in v1). */ +Value* builtin_task_send(Value *arg) { + if (!arg || arg->type != VAL_LIST || arg->data.list.count < 2 || !g_task_sched) + return make_num(0); + Value *idv = arg->data.list.items[0]; + if (!idv || idv->type != VAL_NUM) return make_num(0); + Value *copy = val_clone_for_send(arg->data.list.items[1]); /* share-nothing */ + int sent = task_deliver((int)idv->data.num, copy); + if (!sent) val_decref(copy); /* dropped to a dead task — release the copy */ + return make_num(sent ? 1 : 0); +} + +/* task_recv of null — return the next message from this task's mailbox, or + * block cooperatively until one arrives (woken by task_send). Same arena / + * nested-evaluation restriction as the other suspending builtins. */ +Value* builtin_task_recv(Value *arg) { + (void)arg; + if (!g_task_sched) return make_null(); /* no tasks: nothing can arrive */ + if (task_mbox_has()) return task_mbox_pop(); + if (g_arena.active) { + rt_error(EK_VALUE, 0, "cannot task_recv inside an arena scope " + "(arena_mark…arena_reset)"); + return make_null(); + } + task_request_recv(); + return make_null(); /* placeholder: the scheduler delivers the message on resume */ +} + +/* task_try_recv of null — non-blocking receive: the next message, or null if + * the mailbox is empty. Never suspends. */ +Value* builtin_task_try_recv(Value *arg) { + (void)arg; + if (!g_task_sched || !task_mbox_has()) return make_null(); + return task_mbox_pop(); +} + +/* task_kill of id — deterministically tear down task `id`: drop its mailbox + * and suspended slice, wake any joiner with an `interrupt` error, mark it + * dead. Returns 1 if killed, 0 for a bad/self/finished target. */ +Value* builtin_task_kill(Value *arg) { + if (!arg || arg->type != VAL_NUM || !g_task_sched) return make_num(0); + return make_num(task_do_kill((int)arg->data.num) ? 1 : 0); +} + /* 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) { @@ -5582,6 +5636,10 @@ void register_builtins(Env *env) { 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, "task_send", make_builtin(builtin_task_send)); + env_set_local_owned(env, "task_recv", make_builtin(builtin_task_recv)); + env_set_local_owned(env, "task_try_recv", make_builtin(builtin_task_try_recv)); + env_set_local_owned(env, "task_kill", make_builtin(builtin_task_kill)); 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/vm.c b/src/vm.c index 6af3196..c0d2731 100644 --- a/src/vm.c +++ b/src/vm.c @@ -2063,6 +2063,7 @@ 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 */ +static void task_apply_recv_result(Task *t); /* fill a recv 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 @@ -2081,10 +2082,12 @@ static Value *vm_run_ex(EigsChunk *chunk, Env *env, Task *resume) { * 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). */ + /* If this task was blocked in task_join / task_recv, fill the + * placeholder on the stack top with the joinee's result (or re-raise + * its error) / the delivered message. A task is blocked on at most one + * of the two, so both are safe to run. */ task_apply_join_result(resume); + task_apply_recv_result(resume); frame = &g_vm.frames[g_vm.frame_count - 1]; chunk = frame->chunk; ip = frame->ip; @@ -5124,6 +5127,7 @@ typedef struct { int current; /* running task id; 0 = main */ int live; /* spawned tasks not yet DONE/DEAD */ int active; /* armed on first spawn */ + int dead_letters; /* inc 2: sends to finished/unknown tasks */ Task main_task; /* task 0 — save-buffer only, never "started" */ } TaskScheduler; @@ -5144,8 +5148,19 @@ static TaskScheduler *sched_ensure(void) { 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); + Task *m = &s->main_task; + /* main runs to completion before we get here, so its saved slice is empty; + * free defensively. Drain any messages left in main's mailbox + its result + * (nulled by the trampoline on a clean exit, but be defensive). */ + free(m->saved_stack); + free(m->saved_frames); + if (m->mbox) { + for (int i = 0; i < m->mbox_count; i++) + val_decref(m->mbox[(m->mbox_head + i) % m->mbox_cap]); + free(m->mbox); + } + if (m->result) val_decref(m->result); + if (m->error_value) val_decref(m->error_value); free(s); g_task_sched = NULL; } @@ -5238,6 +5253,136 @@ int task_request_join(int target) { return 1; } +/* ---- Inc 2: mailboxes -------------------------------------------------- */ + +/* Append msg (ownership transferred in) to task `tid`'s FIFO mailbox and wake + * it if it is blocked in task_recv. Returns 1 if delivered, 0 if dropped + * because the target is gone (finished/unknown) — send-to-dead is a silent + * drop plus a dead-letter count (Akka dead-letters / Erlang cast), NOT an + * error: an error path here would be a nondeterminism magnet. */ +int task_deliver(int tid, Value *msg_owned) { + TaskScheduler *s = sched_get(); + Task *t = s ? sched_lookup(s, tid) : NULL; + if (!t || t->state == TASK_DONE || t->state == TASK_DEAD) { + if (s) s->dead_letters++; + return 0; + } + if (t->mbox_count >= t->mbox_cap) { + int nc = t->mbox_cap ? t->mbox_cap * 2 : 8; + Value **nb = xmalloc(sizeof(Value *) * nc); + for (int i = 0; i < t->mbox_count; i++) + nb[i] = t->mbox[(t->mbox_head + i) % t->mbox_cap]; + free(t->mbox); + t->mbox = nb; t->mbox_cap = nc; t->mbox_head = 0; + } + t->mbox[(t->mbox_head + t->mbox_count) % t->mbox_cap] = msg_owned; + t->mbox_count++; + /* Wake a recv-blocked receiver on the FIRST message that arrives while it + * waits on an empty mailbox (mbox_count just became 1). recv_blocked stays + * set so the resume path (task_apply_recv_result) delivers this message and + * clears it; the mbox_count==1 guard makes the enqueue idempotent — a + * second send before the receiver resumes finds count>1 and does not + * re-enqueue (which would put the task in the ready queue twice). */ + if (t->recv_blocked && t->state == TASK_SUSPENDED && t->mbox_count == 1) + sched_ready_push(s, tid); + return 1; +} + +int task_mbox_has(void) { + Task *t = task_current_running(); + return (t && t->mbox_count > 0) ? 1 : 0; +} + +Value *task_mbox_pop(void) { + Task *t = task_current_running(); + if (!t || t->mbox_count == 0) return make_null(); + Value *v = t->mbox[t->mbox_head]; + t->mbox_head = (t->mbox_head + 1) % t->mbox_cap; + t->mbox_count--; + return v; /* owned ref transfers to caller */ +} + +void task_request_recv(void) { + Task *t = task_current_running(); + if (t) t->recv_blocked = 1; + g_task_suspend_request = 1; +} + +/* Deterministic teardown of a task mid-run (task_kill): drop its mailbox and + * saved slice, wake any joiner with an `interrupt` error, mark it DEAD. The + * handle entry stays (task_alive → 0, joiners see DEAD); handle_table_drain + * frees the struct at exit. Returns 0 for a bad/self/finished target. */ +int task_do_kill(int tid) { + TaskScheduler *s = sched_get(); + if (!s || tid == 0 || tid == s->current) return 0; + Task *t = sched_lookup(s, tid); + if (!t || t->state == TASK_DONE || t->state == TASK_DEAD) return 0; + /* Drain the mailbox. */ + while (t->mbox_count > 0) { + val_decref(t->mbox[t->mbox_head]); + t->mbox_head = (t->mbox_head + 1) % t->mbox_cap; + t->mbox_count--; + } + free(t->mbox); t->mbox = NULL; t->mbox_cap = 0; t->mbox_head = 0; + /* Release the suspended slice's counted refs (mirror task_free's slice + * teardown) so a killed suspended task doesn't leak. */ + if (t->saved_stack) { + for (int i = 0; i < t->saved_stack_len; i++) slot_decref(t->saved_stack[i]); + free(t->saved_stack); t->saved_stack = NULL; t->saved_stack_len = 0; + } + 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); t->saved_frames = NULL; t->saved_frame_count = 0; + } + if (t->run_env) { env_decref(t->run_env); t->run_env = NULL; } + t->has_error = 1; + t->state = TASK_DEAD; + s->live--; + /* Wake joiners with an interrupt: on resume task_apply_join_result sees + * has_error and re-raises. Give them an error payload. */ + if (!t->error_value) { + Value *ev = make_dict(3); + dict_set_owned(ev, "kind", make_str(err_kind_name(EK_INTERRUPT))); + dict_set_owned(ev, "message", make_str("task was killed")); + dict_set_owned(ev, "line", make_num(0)); + t->error_value = ev; + } + 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 == tid) + sched_ready_push(s, w->id); + } + if (s->main_task.state == TASK_SUSPENDED && s->main_task.join_target == tid) + sched_ready_push(s, 0); + return 1; +} + +/* On resuming a recv-blocked task, fill the placeholder the task_recv builtin + * left on the stack top with the next mailbox message. */ +static void task_apply_recv_result(Task *t) { + /* Only a task that suspended INSIDE task_recv has a placeholder to fill. + * A task resuming from a plain task_yield/task_join must NOT have its + * mailbox drained here, even if a message arrived meanwhile. */ + if (!t->recv_blocked) return; + t->recv_blocked = 0; + if (g_vm.sp > 0) { + slot_decref(g_vm.stack[g_vm.sp - 1]); + Value *msg; + if (t->mbox_count > 0) { + msg = t->mbox[t->mbox_head]; + t->mbox_head = (t->mbox_head + 1) % t->mbox_cap; + t->mbox_count--; + } else { + msg = make_null(); /* woken without a message (killed sender race) */ + } + g_vm.stack[g_vm.sp - 1] = slot_from_heap(msg); + } +} + /* 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 @@ -5365,8 +5510,9 @@ static Value *scheduler_trampoline(TaskScheduler *s) { 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); + * blocked (woken by sched_finish); task_recv on an empty mailbox → + * stay blocked (woken by task_deliver). */ + if (t->join_target == 0 && !t->recv_blocked) sched_ready_push(s, id); } else { sched_finish(s, t, r); /* kill-outstanding: main ending tears the rest down deterministically. */ diff --git a/src/vm.h b/src/vm.h index db95c80..2b593c1 100644 --- a/src/vm.h +++ b/src/vm.h @@ -409,6 +409,14 @@ typedef struct Task { * 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; + /* Inc 2: unbounded FIFO mailbox of deep-copied messages (share-nothing, + * Erlang-style — bounded/backpressure is a cheap later add). Circular + * buffer; grows on demand. recv_blocked is 1 while this task is suspended + * in task_recv on an empty mailbox — woken by task_send, delivered to the + * stack top on resume (same placeholder mechanism as join). */ + Value **mbox; + int mbox_head, mbox_count, mbox_cap; + int recv_blocked; /* Completion. */ Value *result; /* owned, deep-copied on normal end */ int has_error; /* died of an uncaught error */ @@ -429,6 +437,12 @@ void task_sched_on_spawn(int id); /* enqueue a freshly spawned task, arm the 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 */ +/* Inc 2 mailbox interface (builtins.c task_send/task_recv/task_try_recv/task_kill). */ +int task_deliver(int tid, Value *msg_owned); /* append msg (ownership taken); 1 sent / 0 dropped-to-dead */ +int task_mbox_has(void); /* current task has a queued message? */ +Value *task_mbox_pop(void); /* pop current task's front message (owned ref) */ +void task_request_recv(void); /* current task blocks awaiting a message */ +int task_do_kill(int tid); /* deterministic teardown of `tid`; 0 = bad target */ /* ---- Public API ---- */ diff --git a/tests/test_tasks.eigs b/tests/test_tasks.eigs index 333125c..9323a23 100644 --- a/tests/test_tasks.eigs +++ b/tests/test_tasks.eigs @@ -139,4 +139,79 @@ define after_arena() as: aa is task_spawn of after_arena assert_eq of [task_join of aa, 17, "task after an arena-dier gets a clean arena"] +# ============ increment 2: mailboxes ============ + +# --- producer/consumer: consumer blocks on recv, producer sends + yields; +# delivery is FIFO and deterministic. --- +cons_id is 0 +define consumer() as: + total is 0 + n is 0 + loop n < 4: + m is task_recv of null + total is total + m + n is n + 1 + return total +define producer() as: + task_send of [cons_id, 10] + task_send of [cons_id, 20] + task_yield of null + task_send of [cons_id, 30] + task_send of [cons_id, 40] + return 0 +cons_id is task_spawn of consumer +pr is task_spawn of producer +assert_eq of [task_join of cons_id, 100, "mailbox FIFO delivery (10+20+30+40)"] + +# --- task_try_recv is non-blocking: null when empty, the message when present --- +box_id is 0 +define boxed() as: + e1 is task_try_recv of null # empty → null + task_yield of null # (main sends 7 meanwhile) + v is task_try_recv of null # → 7 + e2 is task_try_recv of null # empty again → null + return [e1, v, e2] +box_id is task_spawn of boxed +task_yield of null +task_send of [box_id, 7] +tr is task_join of box_id +assert_eq of [tr[0], null, "try_recv on empty mailbox is null"] +assert_eq of [tr[1], 7, "try_recv returns the queued message"] +assert_eq of [tr[2], null, "try_recv empty after draining is null"] + +# --- send to a finished/unknown task is a silent drop returning 0 --- +define fast() as: + return 1 +f1 is task_spawn of fast +jr is task_join of f1 # f1 finishes +assert_eq of [task_send of [f1, 99], 0, "send to a finished task drops (returns 0)"] +assert_eq of [task_send of [88888, 1], 0, "send to an unknown task drops (returns 0)"] + +# --- messages cross the boundary deep-copied (share-nothing) --- +echo_id is 0 +define echoer() as: + m is task_recv of null + return len of m +echo_id is task_spawn of echoer +payload is [1, 2, 3] +task_send of [echo_id, payload] +payload is append of [payload, 4] # mutate after send +assert_eq of [task_join of echo_id, 3, "received message is a deep copy, not aliased"] + +# --- task_kill: a joiner on a killed task gets an `interrupt` error --- +define victim() as: + task_recv of null # blocks forever + return 1 +vk is task_spawn of victim +task_yield of null +assert_eq of [task_kill of vk, 1, "task_kill of a live task returns 1"] +assert_eq of [task_alive of vk, 0, "a killed task is not alive"] +kthrew is 0 +try: + kr is task_join of vk +catch e: + kthrew is 1 + assert_eq of [e.kind, "interrupt", "join of a killed task raises interrupt"] +assert_eq of [kthrew, 1, "join of a killed task raises"] + test_summary of null