Files
rose-ash/lib/erlang/runtime.sx
giles d191f7cd9e
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Has been cancelled
erlang: send + selective receive via shift/reset (+13 tests)
2026-04-24 20:27:59 +00:00

328 lines
8.8 KiB
Plaintext

;; Erlang runtime — scheduler, process records, mailbox queue.
;; Phase 3 foundation. spawn/send/receive build on these primitives.
;;
;; Scheduler is a single global dict in `er-scheduler` holding:
;; :next-pid INT — counter for fresh pid allocation
;; :processes DICT — pid-key (string) -> process record
;; :runnable QUEUE — FIFO of pids ready to run
;; :current PID — pid currently executing, or nil
;;
;; A pid value is tagged: {:tag "pid" :id INT}. Pids compare by id.
;;
;; Process record fields:
;; :pid — this process's pid
;; :mailbox — queue of received messages (arrival order)
;; :state — "runnable" | "running" | "waiting" | "exiting" | "dead"
;; :continuation — saved k (for receive suspension); nil otherwise
;; :receive-pats — patterns the process is blocked on; nil otherwise
;; :trap-exit — bool
;; :links — list of pids
;; :monitors — list of {:ref :pid}
;; :env — Erlang env at the last yield
;; :exit-reason — nil until the process exits
;;
;; Queue — amortised-O(1) FIFO with head-pointer + slab-compact:
;; {:items (list...) :head-idx INT}
;; ── queue ────────────────────────────────────────────────────────
(define er-q-new (fn () {:head-idx 0 :items (list)}))
(define er-q-push! (fn (q x) (append! (get q :items) x)))
(define
er-q-pop!
(fn
(q)
(let
((h (get q :head-idx)) (items (get q :items)))
(if
(>= h (len items))
nil
(let
((x (nth items h)))
(dict-set! q :head-idx (+ h 1))
(er-q-compact! q)
x)))))
(define
er-q-peek
(fn
(q)
(let
((h (get q :head-idx)) (items (get q :items)))
(if (>= h (len items)) nil (nth items h)))))
(define
er-q-len
(fn (q) (- (len (get q :items)) (get q :head-idx))))
(define er-q-empty? (fn (q) (= (er-q-len q) 0)))
;; Compact the backing list when the head pointer gets large so the
;; queue doesn't grow without bound. Threshold chosen to amortise the
;; O(n) copy — pops are still amortised O(1).
(define
er-q-compact!
(fn
(q)
(let
((h (get q :head-idx)) (items (get q :items)))
(when
(> h 128)
(let
((new (list)))
(for-each
(fn (i) (append! new (nth items i)))
(range h (len items)))
(dict-set! q :items new)
(dict-set! q :head-idx 0))))))
(define
er-q-to-list
(fn
(q)
(let
((h (get q :head-idx)) (items (get q :items)) (out (list)))
(for-each
(fn (i) (append! out (nth items i)))
(range h (len items)))
out)))
;; Read the i'th entry (relative to head) without popping.
(define
er-q-nth
(fn (q i) (nth (get q :items) (+ (get q :head-idx) i))))
;; Remove entry at logical index i, shift tail in.
(define
er-q-delete-at!
(fn
(q i)
(let
((h (get q :head-idx)) (items (get q :items)) (new (list)))
(for-each
(fn
(j)
(when (not (= j (+ h i))) (append! new (nth items j))))
(range h (len items)))
(dict-set! q :items new)
(dict-set! q :head-idx 0))))
;; ── pids ─────────────────────────────────────────────────────────
(define er-mk-pid (fn (id) {:id id :tag "pid"}))
(define er-pid? (fn (v) (er-is-tagged? v "pid")))
(define er-pid-id (fn (pid) (get pid :id)))
(define er-pid-key (fn (pid) (str "p" (er-pid-id pid))))
(define
er-pid-equal?
(fn (a b) (and (er-pid? a) (er-pid? b) (= (er-pid-id a) (er-pid-id b)))))
;; ── scheduler state ──────────────────────────────────────────────
(define er-scheduler (list nil))
(define
er-sched-init!
(fn
()
(set-nth!
er-scheduler
0
{:next-pid 0
:current nil
:processes {}
:runnable (er-q-new)})))
(define er-sched (fn () (nth er-scheduler 0)))
(define
er-pid-new!
(fn
()
(let
((s (er-sched)))
(let
((n (get s :next-pid)))
(dict-set! s :next-pid (+ n 1))
(er-mk-pid n)))))
(define
er-sched-runnable
(fn () (get (er-sched) :runnable)))
(define
er-sched-processes
(fn () (get (er-sched) :processes)))
(define
er-sched-enqueue!
(fn (pid) (er-q-push! (er-sched-runnable) pid)))
(define
er-sched-next-runnable!
(fn () (er-q-pop! (er-sched-runnable))))
(define
er-sched-runnable-count
(fn () (er-q-len (er-sched-runnable))))
(define
er-sched-set-current!
(fn (pid) (dict-set! (er-sched) :current pid)))
(define er-sched-current-pid (fn () (get (er-sched) :current)))
(define
er-sched-process-count
(fn () (len (keys (er-sched-processes)))))
;; ── process records ──────────────────────────────────────────────
(define
er-proc-new!
(fn
(env)
(let
((pid (er-pid-new!)))
(let
((proc
{:pid pid
:env env
:links (list)
:mailbox (er-q-new)
:state "runnable"
:monitors (list)
:continuation nil
:receive-pats nil
:trap-exit false
:exit-reason nil}))
(dict-set! (er-sched-processes) (er-pid-key pid) proc)
(er-sched-enqueue! pid)
proc))))
(define
er-proc-get
(fn (pid) (get (er-sched-processes) (er-pid-key pid))))
(define
er-proc-exists?
(fn (pid) (dict-has? (er-sched-processes) (er-pid-key pid))))
(define
er-proc-field
(fn (pid field) (get (er-proc-get pid) field)))
(define
er-proc-set!
(fn
(pid field val)
(let
((p (er-proc-get pid)))
(if
(= p nil)
(error (str "Erlang: no such process " (er-pid-key pid)))
(dict-set! p field val)))))
(define
er-proc-mailbox-push!
(fn (pid msg) (er-q-push! (er-proc-field pid :mailbox) msg)))
(define
er-proc-mailbox-size
(fn (pid) (er-q-len (er-proc-field pid :mailbox))))
;; ── process BIFs ────────────────────────────────────────────────
(define
er-bif-is-pid
(fn (vs) (er-bool (er-pid? (er-bif-arg1 vs "is_pid")))))
(define
er-bif-self
(fn
(vs)
(if
(not (= (len vs) 0))
(error "Erlang: self/0: arity")
(let
((pid (er-sched-current-pid)))
(if
(= pid nil)
(error "Erlang: self/0: no current process")
pid)))))
(define
er-bif-spawn
(fn
(vs)
(cond
(= (len vs) 1) (er-spawn-fun (nth vs 0))
(= (len vs) 3) (error
"Erlang: spawn/3: module-based spawn deferred to Phase 5 (modules)")
:else (error "Erlang: spawn: wrong arity"))))
(define
er-spawn-fun
(fn
(fv)
(if
(not (er-fun? fv))
(error "Erlang: spawn/1: not a fun")
(let
((proc (er-proc-new! (er-env-new))))
(dict-set! proc :initial-fun fv)
(get proc :pid)))))
;; ── scheduler loop ──────────────────────────────────────────────
;; Each process's entry runs inside a `reset`; `receive` uses `shift`
;; to suspend (saving a continuation on the proc record). When a `!`
;; delivers a message to a waiting process we re-enqueue it — the
;; scheduler step invokes the saved continuation, which retries the
;; receive against the updated mailbox.
(define er-suspend-marker {:tag "er-suspend-marker"})
(define
er-suspended?
(fn
(v)
(and
(= (type-of v) "dict")
(= (get v :tag) "er-suspend-marker"))))
(define
er-sched-run-all!
(fn
()
(let
((pid (er-sched-next-runnable!)))
(when
(not (= pid nil))
(er-sched-step! pid)
(er-sched-run-all!)))))
(define
er-sched-step!
(fn
(pid)
(er-sched-set-current! pid)
(er-proc-set! pid :state "running")
(let
((prev-k (er-proc-field pid :continuation))
(result-ref (list nil)))
(if
(= prev-k nil)
(set-nth!
result-ref
0
(reset (er-apply-fun (er-proc-field pid :initial-fun) (list))))
(do
(er-proc-set! pid :continuation nil)
(set-nth! result-ref 0 (prev-k nil))))
(let
((r (nth result-ref 0)))
(cond
(er-suspended? r) nil
:else (do
(er-proc-set! pid :state "dead")
(er-proc-set! pid :exit-reason (er-mk-atom "normal"))
(er-proc-set! pid :exit-result r)
(er-proc-set! pid :continuation nil)))))
(er-sched-set-current! nil)))