erlang: erlang:send_after/3 + cancel_timer/1 + monotonic_time (T1+T2, 766/766)
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m2s
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 1m2s
Logical-clock timer wheel in the scheduler. send_after schedules a message-delivery event at an absolute deadline (clock + Time ms); cancel_timer marks a live timer cancelled and reports remaining ms, or false. Time advances only when the runnable queue drains, jumping to the earliest pending deadline (deterministic, no wall clock). monotonic_time/0,1 exposes the logical ms clock. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -135,6 +135,56 @@
|
||||
(dict-set! s :next-ref (+ n 1))
|
||||
(er-mk-ref n)))))
|
||||
|
||||
;; ── logical clock + timer wheel ──────────────────────────────────
|
||||
;; The scheduler runs a synchronous model: logical time advances only
|
||||
;; when the runnable queue drains (see `er-sched-advance-time!`). The
|
||||
;; clock is in milliseconds, monotonic, never derived from wall time
|
||||
;; — deterministic and time-travel-safe. `send_after` schedules a
|
||||
;; message-delivery event at an absolute deadline; `receive after Ms`
|
||||
;; schedules a timeout event the same way. When no process is runnable
|
||||
;; the scheduler jumps the clock to the earliest pending deadline and
|
||||
;; fires that single event, then re-runs.
|
||||
(define er-clock (fn () (get (er-sched) :clock)))
|
||||
|
||||
;; Advance the clock to `ms`, but never backwards (monotonicity).
|
||||
(define
|
||||
er-clock-set!
|
||||
(fn (ms) (dict-set! (er-sched) :clock (max (er-clock) ms))))
|
||||
|
||||
(define er-sched-timers (fn () (get (er-sched) :timers)))
|
||||
|
||||
;; Register a timer event. `dest` is a pid or registered-atom value,
|
||||
;; resolved to a live process at fire time. Returns the timer ref.
|
||||
(define
|
||||
er-timer-add!
|
||||
(fn
|
||||
(deadline dest msg ref)
|
||||
(append!
|
||||
(er-sched-timers)
|
||||
{:ref ref :deadline deadline :dest dest :msg msg :alive true})
|
||||
ref))
|
||||
|
||||
;; Find the live timer with the given ref, or nil.
|
||||
(define
|
||||
er-timer-find-alive
|
||||
(fn
|
||||
(ref)
|
||||
(let
|
||||
((ts (er-sched-timers)) (found (list nil)))
|
||||
(for-each
|
||||
(fn
|
||||
(i)
|
||||
(let
|
||||
((t (nth ts i)))
|
||||
(when
|
||||
(and
|
||||
(= (nth found 0) nil)
|
||||
(get t :alive)
|
||||
(er-ref-equal? (get t :ref) ref))
|
||||
(set-nth! found 0 t))))
|
||||
(range 0 (len ts)))
|
||||
(nth found 0))))
|
||||
|
||||
;; ── scheduler state ──────────────────────────────────────────────
|
||||
(define er-scheduler (list nil))
|
||||
|
||||
@@ -151,6 +201,8 @@
|
||||
:processes {}
|
||||
:registered {}
|
||||
:ets {}
|
||||
:clock 0
|
||||
:timers (list)
|
||||
:runnable (er-q-new)})))
|
||||
|
||||
(define er-sched (fn () (nth er-scheduler 0)))
|
||||
@@ -217,6 +269,7 @@
|
||||
:trap-exit false
|
||||
:has-timeout false
|
||||
:timed-out false
|
||||
:timeout-deadline nil
|
||||
:exit-reason nil}))
|
||||
(dict-set! (er-sched-processes) (er-pid-key pid) proc)
|
||||
(er-sched-enqueue! pid)
|
||||
@@ -456,6 +509,69 @@
|
||||
(error "Erlang: make_ref/0: arity")
|
||||
(er-ref-new!))))
|
||||
|
||||
;; ── timer BIFs ───────────────────────────────────────────────────
|
||||
;; erlang:send_after(Time, Dest, Msg) -> Ref
|
||||
;; Schedules Msg to be delivered to Dest after Time ms (logical).
|
||||
;; Time must be a non-negative integer; Dest a pid or registered
|
||||
;; atom name. Returns a fresh timer reference.
|
||||
(define
|
||||
er-bif-send-after
|
||||
(fn
|
||||
(vs)
|
||||
(let
|
||||
((time (nth vs 0)) (dest (nth vs 1)) (msg (nth vs 2)))
|
||||
(cond
|
||||
(not (and (= (type-of time) "number") (>= time 0)))
|
||||
(raise (er-mk-error-marker (er-mk-atom "badarg")))
|
||||
(not (or (er-pid? dest) (er-atom? dest)))
|
||||
(raise (er-mk-error-marker (er-mk-atom "badarg")))
|
||||
:else
|
||||
(er-timer-add!
|
||||
(+ (er-clock) (truncate time))
|
||||
dest
|
||||
msg
|
||||
(er-ref-new!))))))
|
||||
|
||||
;; erlang:cancel_timer(Ref) -> RemainingMs | false
|
||||
;; For a live (not-yet-fired) timer, marks it cancelled and returns
|
||||
;; the milliseconds left until its deadline. For an already-fired,
|
||||
;; already-cancelled, or unknown ref, returns the atom `false`.
|
||||
(define
|
||||
er-bif-cancel-timer
|
||||
(fn
|
||||
(vs)
|
||||
(let
|
||||
((ref (nth vs 0)))
|
||||
(cond
|
||||
(not (er-ref? ref))
|
||||
(raise (er-mk-error-marker (er-mk-atom "badarg")))
|
||||
:else
|
||||
(let
|
||||
((t (er-timer-find-alive ref)))
|
||||
(cond
|
||||
(= t nil) (er-mk-atom "false")
|
||||
:else (do
|
||||
(dict-set! t :alive false)
|
||||
(max 0 (- (get t :deadline) (er-clock))))))))))
|
||||
|
||||
;; erlang:monotonic_time() | erlang:monotonic_time(Unit) -> Integer
|
||||
;; Returns the scheduler's logical monotonic clock in milliseconds.
|
||||
;; Unit (millisecond / second / native) is accepted for API
|
||||
;; compatibility; all units report from the same ms-resolution clock.
|
||||
(define
|
||||
er-bif-monotonic-time
|
||||
(fn
|
||||
(vs)
|
||||
(cond
|
||||
(= (len vs) 0) (er-clock)
|
||||
(and (= (len vs) 1) (er-atom? (nth vs 0)))
|
||||
(let
|
||||
((unit (get (nth vs 0) :name)))
|
||||
(cond
|
||||
(= unit "second") (truncate (/ (er-clock) 1000))
|
||||
:else (er-clock)))
|
||||
:else (raise (er-mk-error-marker (er-mk-atom "badarg"))))))
|
||||
|
||||
;; Add `target` to `pid`'s :links list if not already there.
|
||||
(define
|
||||
er-link-add-one!
|
||||
@@ -664,37 +780,122 @@
|
||||
(cond
|
||||
(not (= pid nil))
|
||||
(do (er-sched-step! pid) (er-sched-run-all!))
|
||||
;; Queue empty — fire one pending receive-with-timeout and go again.
|
||||
(er-sched-fire-one-timeout!) (er-sched-run-all!)
|
||||
;; Queue empty — advance logical time to the next pending
|
||||
;; deadline (timer delivery or receive-timeout) and go again.
|
||||
(er-sched-advance-time!) (er-sched-run-all!)
|
||||
:else nil))))
|
||||
|
||||
;; Wake one waiting process whose receive had an `after Ms` clause.
|
||||
;; Returns true if one fired. In our synchronous model "time passes"
|
||||
;; once the runnable queue drains — timeouts only fire then.
|
||||
;; ── time advance ─────────────────────────────────────────────────
|
||||
;; Called when the runnable queue is empty. Two kinds of pending event
|
||||
;; carry a deadline: live `send_after` timers and waiting processes in
|
||||
;; a `receive ... after Ms` block. Find the single earliest deadline
|
||||
;; across both, jump the clock to it, and fire just that one event
|
||||
;; (timer wins ties — a message delivered exactly at the timeout
|
||||
;; arrives "first"). Returns true if an event fired, false when there
|
||||
;; is nothing left to wake (genuine idle / termination).
|
||||
(define
|
||||
er-sched-fire-one-timeout!
|
||||
er-sched-advance-time!
|
||||
(fn
|
||||
()
|
||||
(let
|
||||
((ks (keys (er-sched-processes))) (fired (list false)))
|
||||
((best (er-sched-next-event)))
|
||||
(cond
|
||||
(= best nil) false
|
||||
:else (do
|
||||
(er-clock-set! (get best :deadline))
|
||||
(cond
|
||||
(= (get best :kind) "timer")
|
||||
(er-timer-fire! (get best :timer))
|
||||
:else (er-recv-timeout-fire! (get best :proc)))
|
||||
true)))))
|
||||
|
||||
;; Scan timers and waiting-with-timeout processes for the earliest
|
||||
;; deadline. Returns {:kind "timer"|"recv" :deadline D ...} or nil.
|
||||
(define
|
||||
er-sched-next-event
|
||||
(fn
|
||||
()
|
||||
(let
|
||||
((best (list nil)))
|
||||
(for-each
|
||||
(fn
|
||||
(i)
|
||||
(let
|
||||
((t (nth (er-sched-timers) i)))
|
||||
(when
|
||||
(get t :alive)
|
||||
(er-event-consider!
|
||||
best
|
||||
{:kind "timer" :deadline (get t :deadline) :timer t}))))
|
||||
(range 0 (len (er-sched-timers))))
|
||||
(for-each
|
||||
(fn
|
||||
(k)
|
||||
(when
|
||||
(not (nth fired 0))
|
||||
(let
|
||||
((p (get (er-sched-processes) k)))
|
||||
(when
|
||||
(and
|
||||
(= (get p :state) "waiting")
|
||||
(get p :has-timeout))
|
||||
(dict-set! p :timed-out true)
|
||||
(dict-set! p :has-timeout false)
|
||||
(dict-set! p :state "runnable")
|
||||
(er-sched-enqueue! (get p :pid))
|
||||
(set-nth! fired 0 true)))))
|
||||
ks)
|
||||
(nth fired 0))))
|
||||
(let
|
||||
((p (get (er-sched-processes) k)))
|
||||
(when
|
||||
(and (= (get p :state) "waiting") (get p :has-timeout))
|
||||
(er-event-consider!
|
||||
best
|
||||
{:kind "recv"
|
||||
:deadline (get p :timeout-deadline)
|
||||
:proc p}))))
|
||||
(keys (er-sched-processes)))
|
||||
(nth best 0))))
|
||||
|
||||
;; Keep the earlier-deadline candidate in the single-cell `best`.
|
||||
;; Strictly-earlier replaces; equal deadlines keep the incumbent so a
|
||||
;; timer registered first (and timers over recv-timeouts) win ties.
|
||||
(define
|
||||
er-event-consider!
|
||||
(fn
|
||||
(best cand)
|
||||
(when
|
||||
(or
|
||||
(= (nth best 0) nil)
|
||||
(< (get cand :deadline) (get (nth best 0) :deadline)))
|
||||
(set-nth! best 0 cand))))
|
||||
|
||||
;; Deliver a fired timer's message to its destination and retire it.
|
||||
;; Destination is resolved at fire time; a dead/missing target (or an
|
||||
;; unregistered name) silently drops the message, as in real Erlang.
|
||||
(define
|
||||
er-timer-fire!
|
||||
(fn
|
||||
(t)
|
||||
(dict-set! t :alive false)
|
||||
(let
|
||||
((pid (er-timer-resolve-dest (get t :dest))))
|
||||
(when
|
||||
(and (not (= pid nil)) (er-proc-exists? pid))
|
||||
(er-proc-mailbox-push! pid (get t :msg))
|
||||
(when
|
||||
(= (er-proc-field pid :state) "waiting")
|
||||
(er-proc-set! pid :state "runnable")
|
||||
(er-sched-enqueue! pid))))))
|
||||
|
||||
;; Non-raising destination resolver for timer delivery.
|
||||
(define
|
||||
er-timer-resolve-dest
|
||||
(fn
|
||||
(v)
|
||||
(cond
|
||||
(er-pid? v) v
|
||||
(er-atom? v)
|
||||
(let
|
||||
((name (get v :name)))
|
||||
(if (dict-has? (er-registered) name) (get (er-registered) name) nil))
|
||||
:else nil)))
|
||||
|
||||
;; Wake a process whose `receive ... after Ms` deadline elapsed.
|
||||
(define
|
||||
er-recv-timeout-fire!
|
||||
(fn
|
||||
(p)
|
||||
(dict-set! p :timed-out true)
|
||||
(dict-set! p :has-timeout false)
|
||||
(dict-set! p :state "runnable")
|
||||
(er-sched-enqueue! (get p :pid))))
|
||||
|
||||
(define
|
||||
er-sched-step!
|
||||
@@ -1506,6 +1707,10 @@
|
||||
(er-register-bif! "erlang" "exit" 1 er-bif-exit)
|
||||
(er-register-bif! "erlang" "exit" 2 er-bif-exit)
|
||||
(er-register-bif! "erlang" "make_ref" 0 er-bif-make-ref)
|
||||
(er-register-bif! "erlang" "send_after" 3 er-bif-send-after)
|
||||
(er-register-bif! "erlang" "cancel_timer" 1 er-bif-cancel-timer)
|
||||
(er-register-bif! "erlang" "monotonic_time" 0 er-bif-monotonic-time)
|
||||
(er-register-bif! "erlang" "monotonic_time" 1 er-bif-monotonic-time)
|
||||
(er-register-bif! "erlang" "link" 1 er-bif-link)
|
||||
(er-register-bif! "erlang" "unlink" 1 er-bif-unlink)
|
||||
(er-register-bif! "erlang" "monitor" 2 er-bif-monitor)
|
||||
|
||||
Reference in New Issue
Block a user