feat(reactive): isolate dispatcher contexts
This commit is contained in:
Kinneyzhang
parent
01f072e4fd
commit
aeb1b0f7c7
13
Makefile
13
Makefile
@ -1,10 +1,10 @@
|
||||
EMACS ?= emacs
|
||||
LOAD_PATH = -L . -L examples -L scripts -L ../ebox -L ../tp -L ../ecss
|
||||
SOURCES = etaf-view.el etaf-compiler.el etaf-component.el etaf-reactive.el etaf-observer.el etaf-context.el etaf-theme-tp.el etaf-resource.el etaf-data.el etaf-generation.el etaf-host.el etaf-retirement.el etaf-render-port.el etaf-renderer.el etaf-runtime.el etaf-behavior.el etaf-actions.el etaf-events.el etaf-performance.el etaf.el scripts/emacs-gui-verifier.el
|
||||
SOURCES = etaf-view.el etaf-compiler.el etaf-component.el etaf-scheduler.el etaf-reactive.el etaf-observer.el etaf-context.el etaf-theme-tp.el etaf-resource.el etaf-data.el etaf-generation.el etaf-host.el etaf-retirement.el etaf-render-port.el etaf-renderer.el etaf-runtime.el etaf-behavior.el etaf-actions.el etaf-events.el etaf-performance.el etaf.el scripts/emacs-gui-verifier.el scripts/benchmark-scheduler-context.el
|
||||
EXAMPLES = examples/etaf-counter-example.el examples/etaf-data-example.el examples/etaf-resource-example.el
|
||||
TESTS = tests/etaf-tests.el tests/etaf-compiler-tests.el tests/etaf-component-frontends-tests.el tests/etaf-resource-tests.el tests/etaf-data-tests.el tests/etaf-theme-tp-tests.el tests/etaf-examples-tests.el tests/etaf-observer-tests.el tests/etaf-performance-tests.el tests/etaf-gui-verifier-tests.el tests/etaf-m0a-current-characterization-tests.el tests/etaf-interaction-contract-tests.el tests/etaf-m0b-component-manifest-tests.el tests/etaf-render-port-tests.el tests/etaf-generation-tests.el tests/etaf-host-tests.el tests/etaf-retirement-tests.el
|
||||
TESTS = tests/etaf-tests.el tests/etaf-compiler-tests.el tests/etaf-component-frontends-tests.el tests/etaf-resource-tests.el tests/etaf-data-tests.el tests/etaf-theme-tp-tests.el tests/etaf-examples-tests.el tests/etaf-observer-tests.el tests/etaf-performance-tests.el tests/etaf-gui-verifier-tests.el tests/etaf-m0a-current-characterization-tests.el tests/etaf-interaction-contract-tests.el tests/etaf-m0b-component-manifest-tests.el tests/etaf-render-port-tests.el tests/etaf-generation-tests.el tests/etaf-host-tests.el tests/etaf-retirement-tests.el tests/etaf-scheduler-tests.el
|
||||
|
||||
.PHONY: test compile load checkdoc docs-check check clean
|
||||
.PHONY: test compile load checkdoc docs-check scheduler-benchmark check clean
|
||||
|
||||
test: compile
|
||||
$(EMACS) -Q --batch $(LOAD_PATH) --eval "(setq load-prefer-newer t)" \
|
||||
@ -24,10 +24,15 @@ docs-check:
|
||||
$(EMACS) -Q --batch $(LOAD_PATH) --eval "(setq load-prefer-newer t)" -l tests/etaf-docs-tests.el \
|
||||
-f ert-run-tests-batch-and-exit
|
||||
|
||||
scheduler-benchmark:
|
||||
$(EMACS) -Q --batch $(LOAD_PATH) --eval "(setq load-prefer-newer t)" \
|
||||
-l scripts/benchmark-scheduler-context.el \
|
||||
-f etaf-scheduler-benchmark-run
|
||||
|
||||
checkdoc:
|
||||
$(EMACS) -Q --batch --eval '(progn (require (quote checkdoc)) (dolist (directory (list "." "examples" "scripts")) (dolist (file (directory-files directory t)) (when (string-suffix-p ".el" file) (checkdoc-file file)))))'
|
||||
|
||||
check: checkdoc compile test docs-check
|
||||
check: checkdoc compile test docs-check scheduler-benchmark
|
||||
|
||||
clean:
|
||||
rm -f *.elc examples/*.elc scripts/*.elc tests/*.elc
|
||||
|
||||
@ -493,6 +493,34 @@ can distinguish “committed, then callback failed” from a rollback failure.
|
||||
The buffer-kill path drains or contains retirement work but never throws a
|
||||
retirement condition from the kill hook.
|
||||
|
||||
Reactive publication is coordinated by an explicit `etaf-scheduler-context`.
|
||||
The context owns the source and Runtime FIFOs, their dedupe sets, effect claims,
|
||||
turn and projection epochs, nesting/busy state, fault diagnostics, and cost
|
||||
counters; it does not own Component, resource, or generation state. Existing
|
||||
callers use `etaf-scheduler-default-context`, while a mount may supply
|
||||
`:scheduler-context` to isolate its dispatch authority. Scopes, Effects, and
|
||||
opaque Runtime routes inherit and retain that context.
|
||||
|
||||
One logical projection groups every changed source by its live subscriber
|
||||
contexts in one subscriber-table scan before draining. Source propagation
|
||||
settles across all touched contexts before any Runtime callback publishes.
|
||||
Within a context, each source and Effect is delivered once per scheduler turn,
|
||||
and multiple changed sources enqueue a Runtime once. A reentrant write to an
|
||||
already delivered source is deferred to the next turn; a per-context turn
|
||||
budget contains cross-context cycles with reusable fault diagnostics. Dedupe
|
||||
in one context never suppresses another, and registry/token/Host validation
|
||||
filters stale Runtime routes before fan-out.
|
||||
|
||||
Runtime callbacks are detached as a turn, so a lifecycle write enters the
|
||||
following turn. Data success/error multi-ref publication and event/action
|
||||
callbacks use this same projection boundary, while the legacy facade continues
|
||||
through the default context. Data source failures update the Controller error
|
||||
state; a later projection/render failure propagates unchanged and cannot be
|
||||
reclassified as a source failure. Runtime operation reports include both the
|
||||
local context deltas and full cross-context projection summaries for source
|
||||
delivery, subscriber visits, effect work, Runtime work, stale drops, turns,
|
||||
and faults.
|
||||
|
||||
Each Runtime flush records a candidate-aware effect tuple containing the
|
||||
generation id, effect-to-source edges and source versions, plus an immutable
|
||||
semantic-node stamp for candidate input/context/output facts. A repeated tuple
|
||||
|
||||
@ -477,6 +477,31 @@ Ebox revision 与 diagnostic-journal ID。`etaf-condition-postcommit-info` 会
|
||||
failure。buffer-kill 路径会 drain 或 contain retirement 工作，但绝不会从 kill
|
||||
hook 抛 retirement condition。
|
||||
|
||||
Reactive publication 由显式 `etaf-scheduler-context` 协调。context 拥有 source
|
||||
与 Runtime FIFO、对应 dedupe set、effect claim、turn/projection epoch、
|
||||
nesting/busy 状态、fault diagnostics 与成本计数器；它不拥有 Component、
|
||||
resource 或 generation state。既有调用者继续使用
|
||||
`etaf-scheduler-default-context`，mount 也可以传入 `:scheduler-context` 隔离
|
||||
dispatch authority。Scope、Effect 与 opaque Runtime route 会继承并保留该
|
||||
context。
|
||||
|
||||
一次 logical projection 会先按 live subscriber context 对所有 changed source
|
||||
在一次 subscriber-table scan 中分组；所有已触及 context 的 source propagation
|
||||
都稳定后，Runtime callback 才能 publish。在同一 context 中，每个 source 与
|
||||
Effect 每个 scheduler turn 只 delivery 一次，多个 changed source 也只 enqueue
|
||||
同一 Runtime 一次。已经 delivery 的 source 若重入写入，会延后到下一个 turn；
|
||||
per-context turn budget 会包含跨 context cycle，并留下可复用的 fault diagnostics。
|
||||
context A 的 dedupe 不会压制 context B；registry/token/Host 校验会在 fan-out 前
|
||||
过滤 stale Runtime route。
|
||||
|
||||
Runtime callback 先作为一个完整 turn detach，因此 lifecycle write 会进入下一个
|
||||
turn。Data success/error 的 multi-ref publication 与 event/action callback 都复用
|
||||
同一个 projection boundary，legacy façade 则继续走 default context。Data source
|
||||
failure 会更新 Controller error state；之后发生的 projection/render failure 保持
|
||||
原 condition，不能被重新归类为 source failure。Runtime operation report 同时携带
|
||||
本 context delta 与完整 cross-context projection summary，覆盖 source delivery、
|
||||
subscriber visit、effect work、Runtime work、stale drop、turn 与 fault。
|
||||
|
||||
每次 Runtime flush 都记录 candidate-aware effect tuple：其中包含 generation id、
|
||||
effect→source 边和 source version，以及 candidate input/context/output facts 的
|
||||
immutable semantic-node stamp。重复 tuple 会报告有序的 effect/edge path；step
|
||||
|
||||
@ -11,6 +11,8 @@
|
||||
;;; Code:
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'etaf-scheduler)
|
||||
(require 'etaf-reactive)
|
||||
(require 'etaf-runtime)
|
||||
|
||||
(defvar etaf--current-runtime)
|
||||
@ -108,14 +110,20 @@ RUNTIME ACTION ...)'. Action functions receive Runtime first."
|
||||
(unless spec
|
||||
(signal 'etaf-action-error
|
||||
(list (format "Unknown ETAF Action: %S" action))))
|
||||
(if (or etaf--observer-context (etaf-runtime-observer runtime))
|
||||
(etaf-runtime-call-operation
|
||||
runtime 'action (format "%S" action)
|
||||
(lambda ()
|
||||
(let ((etaf--current-runtime runtime))
|
||||
(apply (etaf-action-spec-function spec) runtime arguments))))
|
||||
(let ((etaf--current-runtime runtime))
|
||||
(apply (etaf-action-spec-function spec) runtime arguments)))))
|
||||
(let ((run
|
||||
(lambda ()
|
||||
(etaf-scheduler-call-with-context
|
||||
(etaf-runtime-scheduler-context runtime)
|
||||
(lambda ()
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(let ((etaf--current-runtime runtime))
|
||||
(apply (etaf-action-spec-function spec)
|
||||
runtime arguments)))))))))
|
||||
(if (or etaf--observer-context (etaf-runtime-observer runtime))
|
||||
(etaf-runtime-call-operation
|
||||
runtime 'action (format "%S" action) run)
|
||||
(funcall run)))))
|
||||
|
||||
;;;###autoload
|
||||
(defun etaf-action-undefine (name)
|
||||
|
||||
110
etaf-data.el
110
etaf-data.el
@ -13,6 +13,7 @@
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'etaf-observer)
|
||||
(require 'etaf-scheduler)
|
||||
(require 'etaf-reactive)
|
||||
|
||||
(define-error 'etaf-data-error "Invalid ETAF data operation")
|
||||
@ -285,36 +286,41 @@ use its result as `:initial-result' for `etaf-data-controller'."
|
||||
(when (= request-id (etaf-data--controller-request-id controller))
|
||||
(let ((normalized (etaf-data--normalize-result result))
|
||||
(auto-load-p (etaf-data--controller-auto-load-p controller)))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(setf (etaf-data--controller-auto-load-p controller) nil)
|
||||
;; ITEMS publication may synchronously schedule a DataGrid render.
|
||||
;; Materialize keyed selection dependencies first so render only
|
||||
;; reads retained reactive state.
|
||||
(etaf-data--prepare-selected-refs
|
||||
controller (plist-get normalized :items))
|
||||
(setf (etaf-value (etaf-data--controller-items controller))
|
||||
(plist-get normalized :items))
|
||||
(setf (etaf-value (etaf-data--controller-total controller))
|
||||
(or (plist-get normalized :total)
|
||||
(length (plist-get normalized :items))))
|
||||
(when (plist-member normalized :page)
|
||||
(setf (etaf-value (etaf-data--controller-page controller))
|
||||
(plist-get normalized :page)))
|
||||
(when (plist-member normalized :page-size)
|
||||
(setf (etaf-value (etaf-data--controller-page-size controller))
|
||||
(plist-get normalized :page-size)))
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)
|
||||
(setf (etaf-value (etaf-data--controller-status controller))
|
||||
'success))
|
||||
(setf (etaf-data--controller-auto-load-p controller) auto-load-p))))
|
||||
;; ITEMS publication may synchronously schedule a DataGrid render.
|
||||
;; Materialize keyed selection dependencies first so render only reads
|
||||
;; retained reactive state.
|
||||
(etaf-data--prepare-selected-refs
|
||||
controller (plist-get normalized :items))
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-data--controller-auto-load-p controller) nil)
|
||||
(etaf-scheduler-defer-finalizer
|
||||
(lambda ()
|
||||
(setf (etaf-data--controller-auto-load-p controller)
|
||||
auto-load-p)))
|
||||
(setf (etaf-value (etaf-data--controller-items controller))
|
||||
(plist-get normalized :items))
|
||||
(setf (etaf-value (etaf-data--controller-total controller))
|
||||
(or (plist-get normalized :total)
|
||||
(length (plist-get normalized :items))))
|
||||
(when (plist-member normalized :page)
|
||||
(setf (etaf-value (etaf-data--controller-page controller))
|
||||
(plist-get normalized :page)))
|
||||
(when (plist-member normalized :page-size)
|
||||
(setf (etaf-value (etaf-data--controller-page-size controller))
|
||||
(plist-get normalized :page-size)))
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)
|
||||
(setf (etaf-value (etaf-data--controller-status controller))
|
||||
'success)))))
|
||||
result)
|
||||
|
||||
(defun etaf-data--apply-error (controller request-id error-data)
|
||||
"Publish ERROR-DATA for CONTROLLER when REQUEST-ID is current."
|
||||
(when (= request-id (etaf-data--controller-request-id controller))
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) error-data)
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'error)))
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) error-data)
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'error)))))
|
||||
|
||||
;;;###autoload
|
||||
(cl-defun etaf-data-controller
|
||||
@ -447,20 +453,23 @@ load. INITIAL-RESULT and AUTO-LOAD are mutually exclusive."
|
||||
(etaf-data--require-controller controller)
|
||||
(let ((request-id (1+ (etaf-data--controller-request-id controller))))
|
||||
(setf (etaf-data--controller-request-id controller) request-id)
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'loading)
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)
|
||||
(condition-case err
|
||||
(etaf-data--apply-load-success
|
||||
controller
|
||||
request-id
|
||||
(etaf-data--source-load
|
||||
(etaf-data--controller-source controller)
|
||||
(etaf-value (etaf-data--controller-query controller))
|
||||
(etaf-value (etaf-data--controller-page controller))
|
||||
(etaf-value (etaf-data--controller-page-size controller))))
|
||||
(error
|
||||
(etaf-data--apply-error controller request-id err)
|
||||
(signal (car err) (cdr err))))))
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'loading)
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)))
|
||||
(let ((result
|
||||
(condition-case err
|
||||
(etaf-data--source-load
|
||||
(etaf-data--controller-source controller)
|
||||
(etaf-value (etaf-data--controller-query controller))
|
||||
(etaf-value (etaf-data--controller-page controller))
|
||||
(etaf-value (etaf-data--controller-page-size controller)))
|
||||
(error
|
||||
(etaf-data--apply-error controller request-id err)
|
||||
(signal (car err) (cdr err))))))
|
||||
;; Projection/render failures after a successful source call are not
|
||||
;; source failures and must not rewrite the committed Controller refs.
|
||||
(etaf-data--apply-load-success controller request-id result))))
|
||||
|
||||
;;;###autoload
|
||||
(defun etaf-data-reload (controller)
|
||||
@ -476,18 +485,21 @@ source mutation result after the reload succeeds."
|
||||
(etaf-data--require-controller controller)
|
||||
(let ((request-id (1+ (etaf-data--controller-request-id controller))))
|
||||
(setf (etaf-data--controller-request-id controller) request-id)
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'loading)
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)
|
||||
(condition-case err
|
||||
(let ((result
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value (etaf-data--controller-status controller)) 'loading)
|
||||
(setf (etaf-value (etaf-data--controller-error controller)) nil)))
|
||||
(let ((result
|
||||
(condition-case err
|
||||
(etaf-data--source-mutate
|
||||
(etaf-data--controller-source controller)
|
||||
operation payload)))
|
||||
(etaf-data-load controller)
|
||||
result)
|
||||
(error
|
||||
(etaf-data--apply-error controller request-id err)
|
||||
(signal (car err) (cdr err))))))
|
||||
operation payload)
|
||||
(error
|
||||
(etaf-data--apply-error controller request-id err)
|
||||
(signal (car err) (cdr err))))))
|
||||
;; Reload owns its own source/projection classification.
|
||||
(etaf-data-load controller)
|
||||
result)))
|
||||
|
||||
;;;###autoload
|
||||
(defun etaf-data-stop (controller)
|
||||
|
||||
@ -12,6 +12,8 @@
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'ebox)
|
||||
(require 'etaf-scheduler)
|
||||
(require 'etaf-reactive)
|
||||
(require 'etaf-runtime)
|
||||
|
||||
(defvar etaf--current-runtime)
|
||||
@ -67,7 +69,7 @@
|
||||
otherwise call the local callback with no arguments."
|
||||
(etaf--assert-not-rendering 'dispatch-event)
|
||||
(setq runtime (etaf-runtime-require-mounted runtime))
|
||||
(let ((dispatch
|
||||
(let* ((dispatch
|
||||
(lambda ()
|
||||
(let ((callback (etaf--event-handler runtime host-ref kind)))
|
||||
(unless callback
|
||||
@ -83,13 +85,19 @@ otherwise call the local callback with no arguments."
|
||||
(if payload-p
|
||||
(funcall callback payload)
|
||||
(funcall callback))
|
||||
(etaf-runtime-event-end runtime)))))))
|
||||
(etaf-runtime-event-end runtime))))))
|
||||
(run
|
||||
(lambda ()
|
||||
(etaf-scheduler-call-with-context
|
||||
(etaf-runtime-scheduler-context runtime)
|
||||
(lambda ()
|
||||
(etaf-reactive-call-with-batch dispatch))))))
|
||||
(if (or etaf--observer-context (etaf-runtime-observer runtime))
|
||||
(etaf-runtime-call-operation
|
||||
runtime 'event
|
||||
(format "%s %S" (etaf-event-kind kind) host-ref)
|
||||
(lambda () (ebox-call-with-render-burst dispatch)))
|
||||
(ebox-call-with-render-burst dispatch))))
|
||||
(lambda () (ebox-call-with-render-burst run)))
|
||||
(ebox-call-with-render-burst run))))
|
||||
|
||||
;;;###autoload
|
||||
(defun etaf-host-ref-bounds (runtime host-ref)
|
||||
|
||||
189
etaf-reactive.el
189
etaf-reactive.el
@ -12,6 +12,7 @@
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'gv)
|
||||
(require 'etaf-scheduler)
|
||||
|
||||
(define-error 'etaf-reactive-error "Invalid ETAF reactive operation")
|
||||
(define-error 'etaf-render-side-effect-error
|
||||
@ -53,6 +54,7 @@
|
||||
scheduler
|
||||
deps
|
||||
owner-scope
|
||||
scheduler-context
|
||||
on-stop
|
||||
name
|
||||
(active-p t)
|
||||
@ -65,13 +67,16 @@
|
||||
effects
|
||||
children
|
||||
cleanups
|
||||
scheduler-context
|
||||
name
|
||||
(active-p t))
|
||||
|
||||
(cl-defstruct (etaf-runtime-route
|
||||
(:constructor etaf-runtime-route-create))
|
||||
"Opaque Runtime route stored by reactive sources."
|
||||
runtime-id mount-epoch authority-token scheduler)
|
||||
runtime-id mount-epoch authority-token scheduler scheduler-context
|
||||
accepts-p
|
||||
(active-p t))
|
||||
|
||||
(defvar etaf--active-effect nil
|
||||
"The effect currently collecting dependencies.")
|
||||
@ -107,72 +112,50 @@
|
||||
The function receives a zero-argument job and a phase symbol. Outside a
|
||||
Runtime, watchers run synchronously.")
|
||||
|
||||
(defvar etaf--dispatch-depth 0)
|
||||
(defvar etaf--dispatch-source-queue nil)
|
||||
(defvar etaf--dispatch-source-queue-tail nil)
|
||||
(defvar etaf--dispatch-source-set (make-hash-table :test #'eq))
|
||||
(defvar etaf--dispatch-runtime-queue nil)
|
||||
(defvar etaf--dispatch-runtime-queue-tail nil)
|
||||
(defvar etaf--dispatch-runtime-set (make-hash-table :test #'eq))
|
||||
(defvar etaf--dispatch-effect-set (make-hash-table :test #'eq))
|
||||
(defun etaf--subscriber-scheduler-context (subscriber)
|
||||
"Return SUBSCRIBER's scheduler context or the default context."
|
||||
(or (cond
|
||||
((etaf-runtime-route-p subscriber)
|
||||
(etaf-runtime-route-scheduler-context subscriber))
|
||||
((etaf-effect-p subscriber)
|
||||
(etaf-effect-scheduler-context subscriber)))
|
||||
etaf-scheduler-default-context))
|
||||
|
||||
(defun etaf--dispatch-append-source (source)
|
||||
"Append SOURCE to the reactive FIFO in constant time."
|
||||
(let ((cell (list source)))
|
||||
(if etaf--dispatch-source-queue-tail
|
||||
(setcdr etaf--dispatch-source-queue-tail cell)
|
||||
(setq etaf--dispatch-source-queue cell))
|
||||
(setq etaf--dispatch-source-queue-tail cell)))
|
||||
(defun etaf-runtime-route-live-p (route)
|
||||
"Return non-nil when opaque Runtime ROUTE still owns dispatch authority."
|
||||
(and (etaf-runtime-route-p route)
|
||||
(etaf-runtime-route-active-p route)
|
||||
(let ((predicate (etaf-runtime-route-accepts-p route)))
|
||||
(or (null predicate)
|
||||
(condition-case nil
|
||||
(funcall predicate route)
|
||||
((error quit) nil))))))
|
||||
|
||||
(defun etaf--dispatch-append-runtime (runtime)
|
||||
"Append RUNTIME to the reactive publication FIFO in constant time."
|
||||
(let ((cell (list runtime)))
|
||||
(if etaf--dispatch-runtime-queue-tail
|
||||
(setcdr etaf--dispatch-runtime-queue-tail cell)
|
||||
(setq etaf--dispatch-runtime-queue cell))
|
||||
(setq etaf--dispatch-runtime-queue-tail cell)))
|
||||
(defun etaf-reactive-enqueue-runtime-flush
|
||||
(runtime function &optional scheduler-context)
|
||||
"Queue FUNCTION once for RUNTIME in SCHEDULER-CONTEXT.
|
||||
The default context preserves the legacy two-argument facade."
|
||||
(etaf-scheduler-enqueue-runtime
|
||||
(or scheduler-context (etaf-scheduler-current-context))
|
||||
runtime function))
|
||||
|
||||
(defun etaf-reactive-enqueue-runtime-flush (runtime function)
|
||||
"Queue FUNCTION once for RUNTIME after the outer reactive dispatch settles."
|
||||
(unless (gethash runtime etaf--dispatch-runtime-set)
|
||||
(puthash runtime function etaf--dispatch-runtime-set)
|
||||
(etaf--dispatch-append-runtime runtime)))
|
||||
|
||||
(defun etaf--dispatch-source-now (source)
|
||||
"Notify SOURCE subscribers without opening another dispatch boundary."
|
||||
(let ((subscribers (copy-hash-table (etaf--source-subscribers source))))
|
||||
(maphash
|
||||
(lambda (subscriber _)
|
||||
(cond
|
||||
((etaf-runtime-route-p subscriber)
|
||||
(funcall (etaf-runtime-route-scheduler subscriber)
|
||||
subscriber source))
|
||||
((etaf-effect-active-p subscriber)
|
||||
(unless (gethash subscriber etaf--dispatch-effect-set)
|
||||
(puthash subscriber t etaf--dispatch-effect-set)
|
||||
(if-let* ((scheduler (etaf-effect-scheduler subscriber)))
|
||||
(funcall scheduler subscriber)
|
||||
(etaf-reactive-effect-run subscriber))))))
|
||||
subscribers)))
|
||||
|
||||
(defun etaf--drain-dispatch ()
|
||||
"Drain reactive sources and Runtime work to a stable outer fixed point."
|
||||
(while (or etaf--dispatch-source-queue etaf--dispatch-runtime-queue)
|
||||
(while etaf--dispatch-source-queue
|
||||
(let ((source (pop etaf--dispatch-source-queue)))
|
||||
(unless etaf--dispatch-source-queue
|
||||
(setq etaf--dispatch-source-queue-tail nil))
|
||||
(remhash source etaf--dispatch-source-set)
|
||||
(etaf--dispatch-source-now source)))
|
||||
;; Detach this turn. A lifecycle write may enqueue a source and the same
|
||||
;; Runtime again for the following turn without merging it into this one.
|
||||
(let ((turn etaf--dispatch-runtime-queue))
|
||||
(setq etaf--dispatch-runtime-queue nil
|
||||
etaf--dispatch-runtime-queue-tail nil)
|
||||
(dolist (runtime turn)
|
||||
(let ((function (gethash runtime etaf--dispatch-runtime-set)))
|
||||
(remhash runtime etaf--dispatch-runtime-set)
|
||||
(funcall function))))))
|
||||
(defun etaf--dispatch-subscriber-group
|
||||
(context source _projection-epoch subscribers)
|
||||
"Notify SOURCE SUBSCRIBERS already grouped for scheduler CONTEXT."
|
||||
(dolist (subscriber subscribers)
|
||||
(etaf-scheduler-record-subscriber-visit context)
|
||||
(cond
|
||||
((etaf-runtime-route-p subscriber)
|
||||
(if (etaf-runtime-route-live-p subscriber)
|
||||
(funcall (etaf-runtime-route-scheduler subscriber)
|
||||
subscriber source)
|
||||
(etaf-scheduler-record-stale-route-drop context)))
|
||||
((and (etaf-effect-p subscriber)
|
||||
(etaf-effect-active-p subscriber)
|
||||
(etaf-scheduler-claim-effect context subscriber))
|
||||
(if-let* ((scheduler (etaf-effect-scheduler subscriber)))
|
||||
(funcall scheduler subscriber)
|
||||
(etaf-reactive-effect-run subscriber))))))
|
||||
|
||||
(defun etaf--reactive-same-p (left right)
|
||||
"Return whether LEFT and RIGHT are equal under ETAF's shallow rule."
|
||||
@ -234,10 +217,14 @@ when the effect is disposed."
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(let* ((owner (or scope etaf--active-scope))
|
||||
(scheduler-context
|
||||
(or (and owner (etaf-effect-scope-scheduler-context owner))
|
||||
(etaf-scheduler-current-context)))
|
||||
(effect (etaf--effect-create
|
||||
:function function
|
||||
:scheduler scheduler
|
||||
:owner-scope owner
|
||||
:scheduler-context scheduler-context
|
||||
:on-stop on-stop
|
||||
:name name)))
|
||||
(when owner
|
||||
@ -253,7 +240,12 @@ When RENDERING is non-nil, `etaf-value' writes are rejected for the duration
|
||||
of the run."
|
||||
(unless (etaf-effect-p effect)
|
||||
(signal 'wrong-type-argument (list 'etaf-effect-p effect)))
|
||||
(cond
|
||||
(let* ((scheduler-context
|
||||
(or (etaf-effect-scheduler-context effect)
|
||||
(etaf-scheduler-current-context)))
|
||||
(etaf--scheduler-context scheduler-context))
|
||||
(etaf-scheduler-record-effect-evaluation scheduler-context)
|
||||
(cond
|
||||
((etaf-effect-running-p effect)
|
||||
(error "Recursive ETAF effect execution: %S"
|
||||
(or (etaf-effect-name effect) effect)))
|
||||
@ -280,23 +272,45 @@ of the run."
|
||||
(etaf--clear-effect-deps effect)
|
||||
(dolist (source old-deps)
|
||||
(puthash effect t (etaf--source-subscribers source)))
|
||||
(setf (etaf-effect-deps effect) old-deps))))))))
|
||||
(setf (etaf-effect-deps effect) old-deps)))))))))
|
||||
|
||||
(defun etaf--dispatch-source (source)
|
||||
"Notify every current subscriber of SOURCE once."
|
||||
(unless (gethash source etaf--dispatch-source-set)
|
||||
(puthash source t etaf--dispatch-source-set)
|
||||
(etaf--dispatch-append-source source))
|
||||
(when (zerop etaf--dispatch-depth)
|
||||
(let ((etaf--dispatch-depth 1))
|
||||
(unwind-protect (etaf--drain-dispatch)
|
||||
(setq etaf--dispatch-source-queue nil
|
||||
etaf--dispatch-source-queue-tail nil
|
||||
etaf--dispatch-runtime-queue nil
|
||||
etaf--dispatch-runtime-queue-tail nil)
|
||||
(clrhash etaf--dispatch-source-set)
|
||||
(clrhash etaf--dispatch-runtime-set)
|
||||
(clrhash etaf--dispatch-effect-set)))))
|
||||
(etaf-scheduler-call-with-projection
|
||||
(lambda ()
|
||||
(let ((groups (make-hash-table :test #'eq)))
|
||||
(maphash
|
||||
(lambda (subscriber _)
|
||||
(let ((context (etaf--subscriber-scheduler-context subscriber)))
|
||||
(cond
|
||||
((etaf-runtime-route-p subscriber)
|
||||
(if (etaf-runtime-route-live-p subscriber)
|
||||
(puthash context
|
||||
(cons subscriber (gethash context groups)) groups)
|
||||
(etaf-scheduler-record-stale-route-drop context)))
|
||||
((and (etaf-effect-p subscriber)
|
||||
(etaf-effect-active-p subscriber))
|
||||
(puthash context
|
||||
(cons subscriber (gethash context groups)) groups)))))
|
||||
(etaf--source-subscribers source))
|
||||
(let (contexts)
|
||||
(maphash (lambda (context _) (push context contexts)) groups)
|
||||
(dolist (context
|
||||
(sort contexts
|
||||
(lambda (left right)
|
||||
(< (etaf-scheduler-context-id left)
|
||||
(etaf-scheduler-context-id right)))))
|
||||
(let ((subscribers (nreverse (gethash context groups))))
|
||||
(etaf-scheduler-enqueue-source
|
||||
context source
|
||||
(lambda (delivery-context delivery-source projection-epoch)
|
||||
(etaf--dispatch-subscriber-group
|
||||
delivery-context delivery-source projection-epoch
|
||||
subscribers))))))))))
|
||||
|
||||
(defun etaf-reactive-call-with-batch (function)
|
||||
"Call FUNCTION inside one cross-context reactive projection."
|
||||
(etaf-scheduler-call-with-projection function))
|
||||
|
||||
;;;###autoload
|
||||
(cl-defun etaf-ref (initial-value &key test name)
|
||||
@ -477,11 +491,20 @@ effect. If FUNCTION returns a function, it cleans up the previous run."
|
||||
(lambda () (etaf--stop-effect effect))))
|
||||
|
||||
;;;###autoload
|
||||
(cl-defun etaf-effect-scope (&key detached name)
|
||||
"Create a Scope named NAME, owned by the current Scope unless DETACHED."
|
||||
(cl-defun etaf-effect-scope (&key detached name scheduler-context)
|
||||
"Create a Scope named NAME, owned by the current Scope unless DETACHED.
|
||||
SCHEDULER-CONTEXT defaults to the parent or dynamically active context."
|
||||
(etaf--assert-not-rendering 'create-scope)
|
||||
(let* ((parent (and (not detached) etaf--active-scope))
|
||||
(scope (etaf--effect-scope-create :parent parent :name name)))
|
||||
(scheduler-context
|
||||
(or scheduler-context
|
||||
(and parent (etaf-effect-scope-scheduler-context parent))
|
||||
(etaf-scheduler-current-context)))
|
||||
(scope (etaf--effect-scope-create
|
||||
:parent parent
|
||||
:scheduler-context
|
||||
(etaf-scheduler-context-resolve scheduler-context)
|
||||
:name name)))
|
||||
(when parent
|
||||
(push scope (etaf-effect-scope-children parent)))
|
||||
scope))
|
||||
@ -499,6 +522,8 @@ effect. If FUNCTION returns a function, it cleans up the previous run."
|
||||
(etaf-effect-scope-active-p scope))
|
||||
(error "Cannot enter an inactive ETAF Scope"))
|
||||
(let ((etaf--active-scope scope)
|
||||
(etaf--scheduler-context
|
||||
(etaf-effect-scope-scheduler-context scope))
|
||||
(etaf--watch-scheduler
|
||||
(if watch-scheduler-p watch-scheduler etaf--watch-scheduler)))
|
||||
(funcall function)))
|
||||
|
||||
212
etaf-runtime.el
212
etaf-runtime.el
@ -14,6 +14,7 @@
|
||||
(require 'ebox)
|
||||
(require 'etaf-view)
|
||||
(require 'etaf-component)
|
||||
(require 'etaf-scheduler)
|
||||
(require 'etaf-reactive)
|
||||
(require 'etaf-generation)
|
||||
(require 'etaf-host)
|
||||
@ -276,6 +277,7 @@ the sequential `etaf--pvec-put' contract."
|
||||
root-view
|
||||
root-node
|
||||
scope
|
||||
scheduler-context
|
||||
root-effect-id
|
||||
root-range-id
|
||||
candidate-root-deps
|
||||
@ -1239,15 +1241,25 @@ effect-to-source edges. It is intentionally immutable and suitable as an
|
||||
(puthash tuple t etaf--runtime-fixed-point-stamps)
|
||||
(push entry etaf--runtime-fixed-point-history))))
|
||||
|
||||
(defun etaf--runtime-authorized-route-runtime (route)
|
||||
"Return the live Runtime authorized by opaque ROUTE, or nil."
|
||||
(when (and (etaf-runtime-route-p route)
|
||||
(etaf-runtime-route-active-p route))
|
||||
(when-let* ((runtime
|
||||
(gethash (etaf-runtime-route-mount-epoch route)
|
||||
etaf--runtime-route-registry)))
|
||||
(and (eq route (etaf-runtime-route-token runtime))
|
||||
(eq (etaf-runtime-route-scheduler-context route)
|
||||
(etaf-runtime-scheduler-context runtime))
|
||||
(etaf-runtime-mounted-p runtime)
|
||||
(etaf-host-authority-accepts-token-p
|
||||
(etaf-runtime-host-authority runtime)
|
||||
(etaf-runtime-route-authority-token route))
|
||||
runtime))))
|
||||
|
||||
(defun etaf--runtime-route-scheduler (route source)
|
||||
"Route dirty SOURCE through opaque ROUTE to its current generation."
|
||||
(when-let* ((runtime
|
||||
(gethash (etaf-runtime-route-mount-epoch route)
|
||||
etaf--runtime-route-registry)))
|
||||
(when (and (etaf-runtime-mounted-p runtime)
|
||||
(etaf-host-authority-accepts-token-p
|
||||
(etaf-runtime-host-authority runtime)
|
||||
(etaf-runtime-route-authority-token route)))
|
||||
(when-let* ((runtime (etaf--runtime-authorized-route-runtime route)))
|
||||
(dolist (effect-id
|
||||
(etaf--generation-source-effects
|
||||
(etaf-runtime-current-generation runtime) source))
|
||||
@ -1258,7 +1270,8 @@ effect-to-source edges. It is intentionally immutable and suitable as an
|
||||
(etaf--runtime-mark-root-dirty runtime))))
|
||||
(etaf-reactive-enqueue-runtime-flush
|
||||
(etaf-runtime-mount-epoch runtime)
|
||||
(lambda () (etaf--runtime-request-flush runtime))))))
|
||||
(lambda () (etaf--runtime-request-flush runtime))
|
||||
(etaf-runtime-scheduler-context runtime))))
|
||||
|
||||
(defun etaf--runtime-evaluate-root-candidate (runtime)
|
||||
"Evaluate RUNTIME root while privately collecting dependencies."
|
||||
@ -1362,6 +1375,11 @@ error or quit."
|
||||
(let* ((operation-id
|
||||
(1+ (or (etaf-runtime-next-operation-id runtime) 0)))
|
||||
(generation-before (etaf-runtime-generation runtime))
|
||||
(scheduler-context
|
||||
(or (etaf-runtime-scheduler-context runtime)
|
||||
etaf-scheduler-default-context))
|
||||
(scheduler-before
|
||||
(etaf-scheduler-context-metrics scheduler-context))
|
||||
(started (float-time))
|
||||
(gc-count-before gcs-done)
|
||||
(gc-elapsed-before gc-elapsed)
|
||||
@ -1376,34 +1394,101 @@ error or quit."
|
||||
(lambda (diagnostic)
|
||||
(etaf--runtime-record-observer-diagnostic
|
||||
runtime diagnostic))))
|
||||
projection-summaries
|
||||
result)
|
||||
(setf (etaf-runtime-next-operation-id runtime) operation-id)
|
||||
(etaf--observer-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(unwind-protect
|
||||
(condition-case condition
|
||||
(prog1 (setq result (funcall function))
|
||||
(setq status 'success))
|
||||
(quit
|
||||
(setq status 'quit)
|
||||
(signal (car condition) (cdr condition))))
|
||||
(etaf-observer-emit
|
||||
(list
|
||||
:provider 'etaf
|
||||
:stage 'runtime-operation
|
||||
:kind kind
|
||||
:label (copy-sequence label)
|
||||
:generation-before generation-before
|
||||
:generation-after (etaf-runtime-generation runtime)
|
||||
:status status
|
||||
:duration-ms
|
||||
(max 0.0 (* 1000.0 (- (float-time) started)))
|
||||
:gc-count (- gcs-done gc-count-before)
|
||||
:gc-duration-ms
|
||||
(max 0.0
|
||||
(* 1000.0 (- gc-elapsed gc-elapsed-before))))))))
|
||||
result)))))
|
||||
(cl-labels
|
||||
((run-body
|
||||
()
|
||||
(etaf--observer-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(condition-case condition
|
||||
(prog1 (setq result (funcall function))
|
||||
(setq status 'success))
|
||||
(quit
|
||||
(setq status 'quit)
|
||||
(signal (car condition) (cdr condition)))))))
|
||||
(emit-final
|
||||
(summaries)
|
||||
(let ((scheduler-after
|
||||
(etaf-scheduler-context-metrics scheduler-context)))
|
||||
(etaf--observer-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-observer-emit
|
||||
(list
|
||||
:provider 'etaf
|
||||
:stage 'runtime-operation
|
||||
:kind kind
|
||||
:label (copy-sequence label)
|
||||
:generation-before generation-before
|
||||
:generation-after (etaf-runtime-generation runtime)
|
||||
:status status
|
||||
:scheduler-context-id
|
||||
(plist-get scheduler-after :context-id)
|
||||
:scheduler-projection-epoch-before
|
||||
(plist-get scheduler-before :projection-epoch)
|
||||
:scheduler-projection-epoch-after
|
||||
(plist-get scheduler-after :projection-epoch)
|
||||
:scheduler-turns
|
||||
(- (plist-get scheduler-after :turn-count)
|
||||
(plist-get scheduler-before :turn-count))
|
||||
:scheduler-source-enqueues
|
||||
(- (plist-get scheduler-after :source-enqueues)
|
||||
(plist-get scheduler-before :source-enqueues))
|
||||
:scheduler-source-dedupes
|
||||
(- (plist-get scheduler-after :source-dedupes)
|
||||
(plist-get scheduler-before :source-dedupes))
|
||||
:scheduler-source-deliveries
|
||||
(- (plist-get scheduler-after :source-deliveries)
|
||||
(plist-get scheduler-before :source-deliveries))
|
||||
:scheduler-subscriber-visits
|
||||
(- (plist-get scheduler-after :subscriber-visits)
|
||||
(plist-get scheduler-before :subscriber-visits))
|
||||
:scheduler-effect-claims
|
||||
(- (plist-get scheduler-after :effect-claims)
|
||||
(plist-get scheduler-before :effect-claims))
|
||||
:scheduler-effect-evaluations
|
||||
(- (plist-get scheduler-after :effect-evaluations)
|
||||
(plist-get scheduler-before :effect-evaluations))
|
||||
:scheduler-runtime-enqueues
|
||||
(- (plist-get scheduler-after :runtime-enqueues)
|
||||
(plist-get scheduler-before :runtime-enqueues))
|
||||
:scheduler-runtime-dedupes
|
||||
(- (plist-get scheduler-after :runtime-dedupes)
|
||||
(plist-get scheduler-before :runtime-dedupes))
|
||||
:scheduler-runtime-executions
|
||||
(- (plist-get scheduler-after :runtime-executions)
|
||||
(plist-get scheduler-before :runtime-executions))
|
||||
:scheduler-stale-route-drops
|
||||
(- (plist-get scheduler-after :stale-route-drops)
|
||||
(plist-get scheduler-before :stale-route-drops))
|
||||
:scheduler-faults
|
||||
(- (plist-get scheduler-after :fault-count)
|
||||
(plist-get scheduler-before :fault-count))
|
||||
:scheduler-projections (copy-tree summaries)
|
||||
:scheduler-fault-state
|
||||
(plist-get scheduler-after :fault-state)
|
||||
:duration-ms
|
||||
(max 0.0 (* 1000.0 (- (float-time) started)))
|
||||
:gc-count (- gcs-done gc-count-before)
|
||||
:gc-duration-ms
|
||||
(max 0.0
|
||||
(* 1000.0
|
||||
(- gc-elapsed gc-elapsed-before))))))))))
|
||||
(if (etaf-scheduler-projection-active-p)
|
||||
(unwind-protect
|
||||
(run-body)
|
||||
(etaf-scheduler-on-projection-complete
|
||||
(lambda (summary) (emit-final (list summary)))))
|
||||
(let ((etaf-scheduler-projection-observer
|
||||
(lambda (summary)
|
||||
(push summary projection-summaries))))
|
||||
(unwind-protect
|
||||
(run-body)
|
||||
(emit-final (nreverse projection-summaries)))))
|
||||
result))))))
|
||||
|
||||
(cl-defmacro etaf--runtime-with-operation ((runtime kind label) &rest body)
|
||||
"Evaluate BODY in RUNTIME operation KIND and LABEL when observed."
|
||||
@ -1425,15 +1510,19 @@ error or quit."
|
||||
"Enter one logical event batch for mounted RUNTIME."
|
||||
(setq runtime (etaf-runtime-require-mounted runtime))
|
||||
(cl-incf (etaf-runtime-event-depth runtime))
|
||||
(etaf-scheduler-context-event-begin
|
||||
(etaf-runtime-scheduler-context runtime))
|
||||
runtime)
|
||||
|
||||
(defun etaf-runtime-event-end (runtime)
|
||||
"Leave RUNTIME's logical event batch and publish pending state once."
|
||||
(when (and (etaf-runtime-p runtime)
|
||||
(etaf-runtime-mounted-p runtime))
|
||||
(when (etaf-runtime-p runtime)
|
||||
(setf (etaf-runtime-event-depth runtime)
|
||||
(max 0 (1- (etaf-runtime-event-depth runtime))))
|
||||
(when (and (zerop (etaf-runtime-event-depth runtime))
|
||||
(etaf-scheduler-context-event-end
|
||||
(etaf-runtime-scheduler-context runtime))
|
||||
(when (and (etaf-runtime-mounted-p runtime)
|
||||
(zerop (etaf-runtime-event-depth runtime))
|
||||
(etaf-runtime-pending-p runtime))
|
||||
(setf (etaf-runtime-pending-p runtime) nil)
|
||||
(etaf--runtime-request-flush runtime)))
|
||||
@ -6109,7 +6198,10 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
|
||||
(defun etaf--runtime-request-flush-now (runtime)
|
||||
"Flush RUNTIME now, or mark one follow-up flush while busy."
|
||||
(when (etaf-runtime-mounted-p runtime)
|
||||
(let ((etaf--scheduler-context
|
||||
(or (etaf-runtime-scheduler-context runtime)
|
||||
etaf-scheduler-default-context)))
|
||||
(when (etaf-runtime-mounted-p runtime)
|
||||
(if (or (> (etaf-runtime-event-depth runtime) 0)
|
||||
(etaf-runtime-flushing-p runtime))
|
||||
(setf (etaf-runtime-pending-p runtime) t)
|
||||
@ -6137,7 +6229,7 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
(when (eq (etaf--runtime-component-overlay runtime)
|
||||
:root-fallback)
|
||||
(etaf--runtime-render-root-turn runtime t)))))
|
||||
(setf (etaf-runtime-flushing-p runtime) nil))))))
|
||||
(setf (etaf-runtime-flushing-p runtime) nil)))))))
|
||||
|
||||
(defun etaf--runtime-request-flush (runtime)
|
||||
"Flush RUNTIME within one optional Runtime operation boundary."
|
||||
@ -6151,13 +6243,19 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
(etaf--runtime-request-flush runtime)
|
||||
(etaf-runtime-root-node runtime)))
|
||||
|
||||
(defun etaf--runtime-mount-now (buffer-or-name view observer)
|
||||
"Mount VIEW with optional OBSERVER into BUFFER-OR-NAME."
|
||||
(let* ((buffer (get-buffer-create buffer-or-name))
|
||||
(defun etaf--runtime-mount-now
|
||||
(buffer-or-name view observer &optional scheduler-context)
|
||||
"Mount VIEW into BUFFER-OR-NAME with OBSERVER and SCHEDULER-CONTEXT."
|
||||
(setq scheduler-context
|
||||
(or scheduler-context etaf-scheduler-default-context))
|
||||
(let ((etaf--scheduler-context scheduler-context))
|
||||
(let* ((buffer (get-buffer-create buffer-or-name))
|
||||
(old (gethash buffer etaf--runtime-table)))
|
||||
(when old
|
||||
(etaf-runtime-unmount old))
|
||||
(let* ((scope (etaf-effect-scope :detached t :name buffer))
|
||||
(let* ((scope (etaf-effect-scope
|
||||
:detached t :name buffer
|
||||
:scheduler-context scheduler-context))
|
||||
(mount-epoch (cl-incf etaf--mount-epoch-counter))
|
||||
(host-authority
|
||||
(etaf-host-authority-create
|
||||
@ -6166,6 +6264,7 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
:buffer buffer
|
||||
:root-view view
|
||||
:scope scope
|
||||
:scheduler-context scheduler-context
|
||||
:instances (make-hash-table :test #'equal)
|
||||
:resource-registry (make-hash-table :test #'equal)
|
||||
:mount-epoch mount-epoch
|
||||
@ -6199,7 +6298,9 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
:mount-epoch (etaf-runtime-mount-epoch runtime)
|
||||
:authority-token
|
||||
(etaf-host-authority-token host-authority)
|
||||
:scheduler 'etaf--runtime-route-scheduler)))
|
||||
:scheduler 'etaf--runtime-route-scheduler
|
||||
:scheduler-context scheduler-context
|
||||
:accepts-p 'etaf--runtime-authorized-route-runtime)))
|
||||
(setf (etaf-runtime-route-token runtime) route)
|
||||
(puthash (etaf-runtime-mount-epoch runtime) runtime
|
||||
etaf--runtime-route-registry))
|
||||
@ -6225,6 +6326,8 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
;; Preaccept failures have no Host authority and discard registration.
|
||||
(unless (etaf-host-authority-attached-p host-authority)
|
||||
(setf (etaf-runtime-mounted-p runtime) nil)
|
||||
(when-let* ((route (etaf-runtime-route-token runtime)))
|
||||
(setf (etaf-runtime-route-active-p route) nil))
|
||||
(etaf-host-authority-rollback-attach host-authority)
|
||||
(when (buffer-live-p buffer)
|
||||
(with-current-buffer buffer
|
||||
@ -6235,7 +6338,7 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
etaf--runtime-route-registry)
|
||||
(etaf-scope-stop scope))
|
||||
(signal (car err) (cdr err))))
|
||||
buffer)))
|
||||
buffer))))
|
||||
|
||||
(defun etaf--runtime-validate-mount-options (options)
|
||||
"Validate and return mount OPTIONS."
|
||||
@ -6245,7 +6348,8 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
(unless tail
|
||||
(error "ETAF mount option %S has no value" key))
|
||||
(pop tail)
|
||||
(unless (memq key '(:viewport-width :viewport-height :observer))
|
||||
(unless (memq key '(:viewport-width :viewport-height :observer
|
||||
:scheduler-context))
|
||||
(error "Unknown ETAF mount option: %S" key)))))
|
||||
(when-let* ((width (plist-get options :viewport-width)))
|
||||
(unless (and (numberp width) (> width 0))
|
||||
@ -6258,6 +6362,12 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
(functionp (plist-get options :observer)))))
|
||||
(signal 'wrong-type-argument
|
||||
(list 'functionp (plist-get options :observer))))
|
||||
(when (and (plist-member options :scheduler-context)
|
||||
(not (etaf-scheduler-context-p
|
||||
(plist-get options :scheduler-context))))
|
||||
(signal 'wrong-type-argument
|
||||
(list 'etaf-scheduler-context-p
|
||||
(plist-get options :scheduler-context))))
|
||||
options)
|
||||
|
||||
;;;###autoload
|
||||
@ -6266,15 +6376,17 @@ backend anchor proof failed; ordinary root turns keep their artifact reuse."
|
||||
Setup, reactive publication, Ebox rendering, and lifecycle work share one
|
||||
framework render-burst allocation budget when the installed Ebox supports it.
|
||||
OPTIONS may provide `:viewport-width' in pixels, `:viewport-height' in lines,
|
||||
and `:observer' as a one-argument flat-report sink. The observer is installed
|
||||
before the first Ebox publication."
|
||||
`:observer' as a one-argument flat-report sink, and `:scheduler-context' for
|
||||
explicit dispatch isolation. The observer is installed before publication."
|
||||
(etaf--assert-not-rendering 'mount-runtime)
|
||||
(setq options (etaf--runtime-validate-mount-options options))
|
||||
(let ((ebox-viewport-width (plist-get options :viewport-width))
|
||||
(ebox-viewport-height (plist-get options :viewport-height)))
|
||||
(ebox-call-with-render-burst
|
||||
#'etaf--runtime-mount-now buffer-or-name view
|
||||
(plist-get options :observer))))
|
||||
(plist-get options :observer)
|
||||
(or (plist-get options :scheduler-context)
|
||||
etaf-scheduler-default-context))))
|
||||
|
||||
(defun etaf--runtime-unmount-now (runtime &optional cause)
|
||||
"Unmount RUNTIME idempotently and retire its Component scopes.
|
||||
@ -6295,6 +6407,8 @@ unmount. Host authority is invalidated before any unbounded cleanup."
|
||||
(etaf-host-authority-begin-detach authority))
|
||||
(unless (eq (etaf-host-authority-state authority) 'terminal)
|
||||
(etaf-host-authority-invalidate authority))
|
||||
(when-let* ((route (etaf-runtime-route-token runtime)))
|
||||
(setf (etaf-runtime-route-active-p route) nil))
|
||||
(setf (etaf-runtime-mounted-p runtime) nil)
|
||||
(unwind-protect
|
||||
(progn
|
||||
|
||||
691
etaf-scheduler.el
Normal file
691
etaf-scheduler.el
Normal file
@ -0,0 +1,691 @@
|
||||
;;; etaf-scheduler.el --- Explicit reactive dispatcher contexts -*- lexical-binding: t; -*-
|
||||
|
||||
;; SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
;;; Commentary:
|
||||
|
||||
;; Owns ETAF's cross-source dispatch queues. Contexts isolate FIFO and dedupe
|
||||
;; authority, while a short-lived projection coordinates fan-out across every
|
||||
;; context touched by one logical reactive publication.
|
||||
|
||||
;;; Code:
|
||||
|
||||
(require 'cl-lib)
|
||||
|
||||
(define-error 'etaf-scheduler-error "Invalid ETAF scheduler operation")
|
||||
|
||||
(defgroup etaf-scheduler nil
|
||||
"Explicit ETAF reactive dispatcher contexts."
|
||||
:group 'applications
|
||||
:prefix "etaf-scheduler-")
|
||||
|
||||
(defcustom etaf-scheduler-default-step-budget 128
|
||||
"Maximum dispatcher turns one context may run in one projection."
|
||||
:type 'positive-integer
|
||||
:group 'etaf-scheduler)
|
||||
|
||||
(defvar etaf-scheduler--context-id-counter 0)
|
||||
(defvar etaf-scheduler--projection-id-counter 0)
|
||||
|
||||
(cl-defstruct
|
||||
(etaf-scheduler-context
|
||||
(:constructor etaf-scheduler-context--create))
|
||||
"Mutable queues and diagnostics for one isolated dispatcher."
|
||||
id name
|
||||
source-queue source-queue-tail source-set
|
||||
deferred-source-queue deferred-source-queue-tail deferred-source-set
|
||||
delivered-source-set
|
||||
runtime-queue runtime-queue-tail runtime-set
|
||||
effect-set
|
||||
(active-turn-id 0)
|
||||
(projection-epoch 0)
|
||||
(completed-projection-epoch 0)
|
||||
(depth 0)
|
||||
(event-depth 0)
|
||||
(busy-depth 0)
|
||||
phase
|
||||
fixed-point-step-budget
|
||||
fault-state diagnostics
|
||||
(source-enqueue-count 0)
|
||||
(source-dedupe-count 0)
|
||||
(source-delivery-count 0)
|
||||
(subscriber-visit-count 0)
|
||||
(effect-claim-count 0)
|
||||
(effect-evaluation-count 0)
|
||||
(runtime-enqueue-count 0)
|
||||
(runtime-dedupe-count 0)
|
||||
(runtime-execution-count 0)
|
||||
(stale-route-drop-count 0)
|
||||
(projection-fault-count 0)
|
||||
(turn-count 0))
|
||||
|
||||
(cl-defstruct
|
||||
(etaf-scheduler--projection
|
||||
(:constructor etaf-scheduler--projection-create))
|
||||
"One cross-context reactive projection."
|
||||
id all-contexts all-context-set context-before finalizers
|
||||
completion-observers)
|
||||
|
||||
(cl-defun etaf-scheduler-context-create (&key name fixed-point-step-budget)
|
||||
"Create an isolated scheduler context named NAME.
|
||||
FIXED-POINT-STEP-BUDGET is optional diagnostic metadata for its owner."
|
||||
(when (and fixed-point-step-budget
|
||||
(not (and (integerp fixed-point-step-budget)
|
||||
(> fixed-point-step-budget 0))))
|
||||
(signal 'etaf-scheduler-error
|
||||
(list :invalid-fixed-point-step-budget fixed-point-step-budget)))
|
||||
(etaf-scheduler-context--create
|
||||
:id (cl-incf etaf-scheduler--context-id-counter)
|
||||
:name name
|
||||
:source-set (make-hash-table :test #'eq)
|
||||
:deferred-source-set (make-hash-table :test #'eq)
|
||||
:delivered-source-set (make-hash-table :test #'eq)
|
||||
:runtime-set (make-hash-table :test #'eq)
|
||||
:effect-set (make-hash-table :test #'eq)
|
||||
:fixed-point-step-budget
|
||||
(or fixed-point-step-budget etaf-scheduler-default-step-budget)))
|
||||
|
||||
(defvar etaf-scheduler-default-context
|
||||
(etaf-scheduler-context-create :name 'default)
|
||||
"Default scheduler context used by the compatibility facade.")
|
||||
|
||||
(defvar etaf--scheduler-context nil
|
||||
"Dynamically active scheduler context, or nil for the default context.")
|
||||
|
||||
(defvar etaf-scheduler--active-projection nil
|
||||
"Dynamically active cross-context projection.")
|
||||
|
||||
(defvar etaf-scheduler-projection-observer nil
|
||||
"Optional function receiving each completed projection summary.")
|
||||
|
||||
(defun etaf-scheduler-current-context ()
|
||||
"Return the dynamically active scheduler context or the default context."
|
||||
(or etaf--scheduler-context etaf-scheduler-default-context))
|
||||
|
||||
(defun etaf-scheduler-context-resolve (context)
|
||||
"Return validated CONTEXT, defaulting nil to the current context."
|
||||
(setq context (or context (etaf-scheduler-current-context)))
|
||||
(unless (etaf-scheduler-context-p context)
|
||||
(signal 'wrong-type-argument
|
||||
(list 'etaf-scheduler-context-p context)))
|
||||
context)
|
||||
|
||||
(defun etaf-scheduler-call-with-context (context function)
|
||||
"Call FUNCTION with CONTEXT as the active scheduler context."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(cl-incf (etaf-scheduler-context-depth context))
|
||||
(let ((etaf--scheduler-context context))
|
||||
(unwind-protect
|
||||
(funcall function)
|
||||
(setf (etaf-scheduler-context-depth context)
|
||||
(max 0 (1- (etaf-scheduler-context-depth context)))))))
|
||||
|
||||
(defun etaf-scheduler-context-event-begin (context)
|
||||
"Increment CONTEXT's logical event nesting depth."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(cl-incf (etaf-scheduler-context-event-depth context))
|
||||
context)
|
||||
|
||||
(defun etaf-scheduler-context-event-end (context)
|
||||
"Decrement CONTEXT's logical event nesting depth."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(setf (etaf-scheduler-context-event-depth context)
|
||||
(max 0 (1- (etaf-scheduler-context-event-depth context))))
|
||||
context)
|
||||
|
||||
(defun etaf-scheduler--append-context (projection context)
|
||||
"Register CONTEXT with PROJECTION once."
|
||||
(unless (gethash context
|
||||
(etaf-scheduler--projection-all-context-set projection))
|
||||
(puthash context t
|
||||
(etaf-scheduler--projection-all-context-set projection))
|
||||
(puthash context (etaf-scheduler-context-metrics context)
|
||||
(etaf-scheduler--projection-context-before projection))
|
||||
(push context (etaf-scheduler--projection-all-contexts projection))))
|
||||
|
||||
(defun etaf-scheduler--append-source (context source)
|
||||
"Append SOURCE to CONTEXT's FIFO in constant time."
|
||||
(let ((cell (list source)))
|
||||
(if (etaf-scheduler-context-source-queue-tail context)
|
||||
(setcdr (etaf-scheduler-context-source-queue-tail context) cell)
|
||||
(setf (etaf-scheduler-context-source-queue context) cell))
|
||||
(setf (etaf-scheduler-context-source-queue-tail context) cell)))
|
||||
|
||||
(defun etaf-scheduler--append-runtime (context runtime)
|
||||
"Append RUNTIME to CONTEXT's FIFO in constant time."
|
||||
(let ((cell (list runtime)))
|
||||
(if (etaf-scheduler-context-runtime-queue-tail context)
|
||||
(setcdr (etaf-scheduler-context-runtime-queue-tail context) cell)
|
||||
(setf (etaf-scheduler-context-runtime-queue context) cell))
|
||||
(setf (etaf-scheduler-context-runtime-queue-tail context) cell)))
|
||||
|
||||
(defun etaf-scheduler--append-deferred-source (context source)
|
||||
"Append SOURCE to CONTEXT's following-turn FIFO."
|
||||
(let ((cell (list source)))
|
||||
(if (etaf-scheduler-context-deferred-source-queue-tail context)
|
||||
(setcdr
|
||||
(etaf-scheduler-context-deferred-source-queue-tail context) cell)
|
||||
(setf (etaf-scheduler-context-deferred-source-queue context) cell))
|
||||
(setf (etaf-scheduler-context-deferred-source-queue-tail context) cell)))
|
||||
|
||||
(defun etaf-scheduler--enqueue-source-now
|
||||
(projection context source delivery)
|
||||
"Enqueue SOURCE and DELIVERY in CONTEXT under PROJECTION."
|
||||
(etaf-scheduler--append-context projection context)
|
||||
(let ((set (etaf-scheduler-context-source-set context))
|
||||
(deferred-set
|
||||
(etaf-scheduler-context-deferred-source-set context)))
|
||||
(cond
|
||||
((or (gethash source set) (gethash source deferred-set))
|
||||
(cl-incf (etaf-scheduler-context-source-dedupe-count context)))
|
||||
((and (eq (etaf-scheduler-context-phase context) 'source)
|
||||
(gethash source
|
||||
(etaf-scheduler-context-delivered-source-set context)))
|
||||
(puthash source delivery deferred-set)
|
||||
(etaf-scheduler--append-deferred-source context source)
|
||||
(cl-incf (etaf-scheduler-context-source-enqueue-count context)))
|
||||
(t
|
||||
(puthash source delivery set)
|
||||
(etaf-scheduler--append-source context source)
|
||||
(cl-incf (etaf-scheduler-context-source-enqueue-count context))))))
|
||||
|
||||
(defun etaf-scheduler--enqueue-runtime-now
|
||||
(projection context runtime function)
|
||||
"Enqueue RUNTIME and FUNCTION in CONTEXT under PROJECTION."
|
||||
(etaf-scheduler--append-context projection context)
|
||||
(let ((set (etaf-scheduler-context-runtime-set context)))
|
||||
(if (gethash runtime set)
|
||||
(cl-incf (etaf-scheduler-context-runtime-dedupe-count context))
|
||||
(puthash runtime function set)
|
||||
(etaf-scheduler--append-runtime context runtime)
|
||||
(cl-incf (etaf-scheduler-context-runtime-enqueue-count context)))))
|
||||
|
||||
(defun etaf-scheduler-enqueue-source (context source delivery)
|
||||
"Queue DELIVERY of SOURCE once in CONTEXT for the active projection.
|
||||
DELIVERY receives CONTEXT, SOURCE, and the projection epoch."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(unless (functionp delivery)
|
||||
(signal 'wrong-type-argument (list 'functionp delivery)))
|
||||
(if etaf-scheduler--active-projection
|
||||
(etaf-scheduler--enqueue-source-now
|
||||
etaf-scheduler--active-projection context source delivery)
|
||||
(etaf-scheduler-call-with-projection
|
||||
(lambda ()
|
||||
(etaf-scheduler--enqueue-source-now
|
||||
etaf-scheduler--active-projection context source delivery))))
|
||||
source)
|
||||
|
||||
(defun etaf-scheduler-enqueue-runtime (context runtime function)
|
||||
"Queue FUNCTION once for RUNTIME in CONTEXT after source delivery."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(if etaf-scheduler--active-projection
|
||||
(etaf-scheduler--enqueue-runtime-now
|
||||
etaf-scheduler--active-projection context runtime function)
|
||||
(etaf-scheduler-call-with-projection
|
||||
(lambda ()
|
||||
(etaf-scheduler--enqueue-runtime-now
|
||||
etaf-scheduler--active-projection context runtime function))))
|
||||
runtime)
|
||||
|
||||
(defun etaf-scheduler-claim-effect (context effect)
|
||||
"Return non-nil after claiming EFFECT once in CONTEXT's projection."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(unless (gethash effect (etaf-scheduler-context-effect-set context))
|
||||
(puthash effect t (etaf-scheduler-context-effect-set context))
|
||||
(cl-incf (etaf-scheduler-context-effect-claim-count context))
|
||||
t))
|
||||
|
||||
(defun etaf-scheduler-record-subscriber-visit (context)
|
||||
"Record one subscriber visit in scheduler CONTEXT."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(when etaf-scheduler--active-projection
|
||||
(etaf-scheduler--append-context
|
||||
etaf-scheduler--active-projection context))
|
||||
(cl-incf (etaf-scheduler-context-subscriber-visit-count context)))
|
||||
|
||||
(defun etaf-scheduler-record-effect-evaluation (context)
|
||||
"Record one reactive effect evaluation in scheduler CONTEXT."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(when etaf-scheduler--active-projection
|
||||
(etaf-scheduler--append-context
|
||||
etaf-scheduler--active-projection context))
|
||||
(cl-incf (etaf-scheduler-context-effect-evaluation-count context)))
|
||||
|
||||
(defun etaf-scheduler-record-stale-route-drop (context)
|
||||
"Record one stale route filtered before fan-out in CONTEXT."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(when etaf-scheduler--active-projection
|
||||
(etaf-scheduler--append-context
|
||||
etaf-scheduler--active-projection context))
|
||||
(cl-incf (etaf-scheduler-context-stale-route-drop-count context)))
|
||||
|
||||
(defun etaf-scheduler-defer-finalizer (function)
|
||||
"Run FUNCTION after the active projection drains, or immediately if idle."
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(if etaf-scheduler--active-projection
|
||||
(push function
|
||||
(etaf-scheduler--projection-finalizers
|
||||
etaf-scheduler--active-projection))
|
||||
(funcall function))
|
||||
function)
|
||||
|
||||
(defun etaf-scheduler-projection-active-p ()
|
||||
"Return non-nil while a scheduler projection is active."
|
||||
(not (null etaf-scheduler--active-projection)))
|
||||
|
||||
(defun etaf-scheduler-on-projection-complete (function)
|
||||
"Call FUNCTION with the active projection summary when it completes."
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(if etaf-scheduler--active-projection
|
||||
(push function
|
||||
(etaf-scheduler--projection-completion-observers
|
||||
etaf-scheduler--active-projection))
|
||||
(funcall function nil))
|
||||
function)
|
||||
|
||||
(defun etaf-scheduler--reset-pending (context)
|
||||
"Discard CONTEXT's pending queues and dedupe membership."
|
||||
(setf (etaf-scheduler-context-source-queue context) nil
|
||||
(etaf-scheduler-context-source-queue-tail context) nil
|
||||
(etaf-scheduler-context-deferred-source-queue context) nil
|
||||
(etaf-scheduler-context-deferred-source-queue-tail context) nil
|
||||
(etaf-scheduler-context-runtime-queue context) nil
|
||||
(etaf-scheduler-context-runtime-queue-tail context) nil
|
||||
(etaf-scheduler-context-phase context) nil)
|
||||
(clrhash (etaf-scheduler-context-source-set context))
|
||||
(clrhash (etaf-scheduler-context-deferred-source-set context))
|
||||
(clrhash (etaf-scheduler-context-delivered-source-set context))
|
||||
(clrhash (etaf-scheduler-context-runtime-set context))
|
||||
context)
|
||||
|
||||
(defun etaf-scheduler--promote-deferred-sources (context)
|
||||
"Move CONTEXT's deferred sources into its following-turn FIFO."
|
||||
(let ((deferred (etaf-scheduler-context-deferred-source-queue context))
|
||||
(deferred-set
|
||||
(etaf-scheduler-context-deferred-source-set context))
|
||||
(source-set (etaf-scheduler-context-source-set context)))
|
||||
(setf (etaf-scheduler-context-deferred-source-queue context) nil
|
||||
(etaf-scheduler-context-deferred-source-queue-tail context) nil)
|
||||
(dolist (source deferred)
|
||||
(let ((delivery (gethash source deferred-set)))
|
||||
(remhash source deferred-set)
|
||||
(unless (gethash source source-set)
|
||||
(puthash source delivery source-set)
|
||||
(etaf-scheduler--append-source context source)))))
|
||||
context)
|
||||
|
||||
(defun etaf-scheduler--projection-contexts (projection)
|
||||
"Return PROJECTION contexts in first-touch order."
|
||||
(nreverse
|
||||
(copy-sequence (etaf-scheduler--projection-all-contexts projection))))
|
||||
|
||||
(defun etaf-scheduler--projection-source-work-p (projection)
|
||||
"Return non-nil when PROJECTION has pending source work."
|
||||
(cl-some #'etaf-scheduler-context-source-queue
|
||||
(etaf-scheduler--projection-contexts projection)))
|
||||
|
||||
(defun etaf-scheduler--projection-work-p (projection)
|
||||
"Return non-nil when PROJECTION has pending source or Runtime work."
|
||||
(cl-some
|
||||
(lambda (context)
|
||||
(or (etaf-scheduler-context-source-queue context)
|
||||
(etaf-scheduler-context-runtime-queue context)))
|
||||
(etaf-scheduler--projection-contexts projection)))
|
||||
|
||||
(defun etaf-scheduler--mark-turn
|
||||
(context projection-id marked-contexts context-steps)
|
||||
"Mark CONTEXT for PROJECTION-ID in MARKED-CONTEXTS and CONTEXT-STEPS."
|
||||
(unless (gethash context marked-contexts)
|
||||
(let* ((steps (1+ (gethash context context-steps 0)))
|
||||
(budget (etaf-scheduler-context-fixed-point-step-budget context)))
|
||||
(when (> steps budget)
|
||||
(signal 'etaf-scheduler-error
|
||||
(list :kind 'fixed-point-step-budget
|
||||
:context-id (etaf-scheduler-context-id context)
|
||||
:projection-id projection-id
|
||||
:steps steps :budget budget)))
|
||||
(puthash context steps context-steps))
|
||||
(puthash context t marked-contexts)
|
||||
(setf (etaf-scheduler-context-projection-epoch context) projection-id)
|
||||
(clrhash (etaf-scheduler-context-delivered-source-set context))
|
||||
(clrhash (etaf-scheduler-context-effect-set context))
|
||||
(cl-incf (etaf-scheduler-context-active-turn-id context))
|
||||
(cl-incf (etaf-scheduler-context-turn-count context))))
|
||||
|
||||
(defun etaf-scheduler--call-context-work (context function)
|
||||
"Call FUNCTION as busy work owned by scheduler CONTEXT."
|
||||
(cl-incf (etaf-scheduler-context-busy-depth context))
|
||||
(unwind-protect
|
||||
(etaf-scheduler-call-with-context context function)
|
||||
(setf (etaf-scheduler-context-busy-depth context)
|
||||
(max 0 (1- (etaf-scheduler-context-busy-depth context))))))
|
||||
|
||||
(defun etaf-scheduler--drain-context-sources (context projection-id)
|
||||
"Drain CONTEXT source FIFO for PROJECTION-ID."
|
||||
(setf (etaf-scheduler-context-phase context) 'source)
|
||||
(etaf-scheduler--call-context-work
|
||||
context
|
||||
(lambda ()
|
||||
(while (etaf-scheduler-context-source-queue context)
|
||||
(let* ((source (pop (etaf-scheduler-context-source-queue context)))
|
||||
(delivery
|
||||
(gethash source
|
||||
(etaf-scheduler-context-source-set context))))
|
||||
(unless (etaf-scheduler-context-source-queue context)
|
||||
(setf (etaf-scheduler-context-source-queue-tail context) nil))
|
||||
(remhash source (etaf-scheduler-context-source-set context))
|
||||
(puthash source t
|
||||
(etaf-scheduler-context-delivered-source-set context))
|
||||
(cl-incf (etaf-scheduler-context-source-delivery-count context))
|
||||
(when delivery
|
||||
(funcall delivery context source projection-id)))))))
|
||||
|
||||
(defun etaf-scheduler--detach-runtime-turns (projection)
|
||||
"Detach and return PROJECTION's Runtime queues in context order."
|
||||
(let (turns)
|
||||
(dolist (context (etaf-scheduler--projection-contexts projection))
|
||||
(when-let* ((turn (etaf-scheduler-context-runtime-queue context)))
|
||||
(setf (etaf-scheduler-context-runtime-queue context) nil
|
||||
(etaf-scheduler-context-runtime-queue-tail context) nil)
|
||||
(push (cons context turn) turns)))
|
||||
(nreverse turns)))
|
||||
|
||||
(defun etaf-scheduler--run-runtime-turn (context turn)
|
||||
"Run detached Runtime TURN owned by CONTEXT."
|
||||
(setf (etaf-scheduler-context-phase context) 'runtime)
|
||||
(etaf-scheduler--call-context-work
|
||||
context
|
||||
(lambda ()
|
||||
(dolist (runtime turn)
|
||||
(let ((function
|
||||
(gethash runtime
|
||||
(etaf-scheduler-context-runtime-set context))))
|
||||
(remhash runtime (etaf-scheduler-context-runtime-set context))
|
||||
(when function
|
||||
(cl-incf (etaf-scheduler-context-runtime-execution-count context))
|
||||
(funcall function)))))))
|
||||
|
||||
(defun etaf-scheduler--record-fault (context projection-id condition)
|
||||
"Record CONTEXT failure CONDITION for PROJECTION-ID."
|
||||
(setf (etaf-scheduler-context-fault-state context) (copy-tree condition))
|
||||
(cl-incf (etaf-scheduler-context-projection-fault-count context))
|
||||
(push (list :projection-epoch projection-id
|
||||
:turn-id (etaf-scheduler-context-active-turn-id context)
|
||||
:condition (copy-tree condition))
|
||||
(etaf-scheduler-context-diagnostics context))
|
||||
(when (> (length (etaf-scheduler-context-diagnostics context)) 64)
|
||||
(setcdr (nthcdr 63 (etaf-scheduler-context-diagnostics context)) nil))
|
||||
condition)
|
||||
|
||||
(defun etaf-scheduler--drain-projection (projection)
|
||||
"Drain PROJECTION source-first and re-signal its first context fault."
|
||||
(let ((projection-id (etaf-scheduler--projection-id projection))
|
||||
(failed-contexts (make-hash-table :test #'eq))
|
||||
(context-steps (make-hash-table :test #'eq))
|
||||
first-condition)
|
||||
(while (etaf-scheduler--projection-work-p projection)
|
||||
(let ((marked-contexts (make-hash-table :test #'eq)))
|
||||
;; All source propagation across every context settles before any
|
||||
;; Runtime can observe and publish the resulting reactive state.
|
||||
(while (etaf-scheduler--projection-source-work-p projection)
|
||||
(dolist (context (etaf-scheduler--projection-contexts projection))
|
||||
(cond
|
||||
((gethash context failed-contexts)
|
||||
(etaf-scheduler--reset-pending context))
|
||||
((etaf-scheduler-context-source-queue context)
|
||||
(condition-case condition
|
||||
(progn
|
||||
(etaf-scheduler--mark-turn
|
||||
context projection-id marked-contexts context-steps)
|
||||
(etaf-scheduler--drain-context-sources
|
||||
context projection-id))
|
||||
((error quit)
|
||||
(puthash context t failed-contexts)
|
||||
(unless first-condition (setq first-condition condition))
|
||||
(etaf-scheduler--record-fault
|
||||
context projection-id condition)
|
||||
(etaf-scheduler--reset-pending context)))))))
|
||||
;; Detach every context's Runtime queue before callbacks run. A write
|
||||
;; from one callback therefore belongs to the following scheduler turn.
|
||||
(dolist (entry (etaf-scheduler--detach-runtime-turns projection))
|
||||
(let ((context (car entry)) (turn (cdr entry)))
|
||||
(unless (gethash context failed-contexts)
|
||||
(condition-case condition
|
||||
(progn
|
||||
(etaf-scheduler--mark-turn
|
||||
context projection-id marked-contexts context-steps)
|
||||
(etaf-scheduler--run-runtime-turn context turn))
|
||||
((error quit)
|
||||
(puthash context t failed-contexts)
|
||||
(unless first-condition (setq first-condition condition))
|
||||
(etaf-scheduler--record-fault
|
||||
context projection-id condition)
|
||||
(etaf-scheduler--reset-pending context))))))
|
||||
(dolist (context (etaf-scheduler--projection-contexts projection))
|
||||
(setf (etaf-scheduler-context-phase context) nil)
|
||||
(unless (gethash context failed-contexts)
|
||||
(etaf-scheduler--promote-deferred-sources context)))))
|
||||
(dolist (context (etaf-scheduler--projection-contexts projection))
|
||||
(unless (gethash context failed-contexts)
|
||||
(setf (etaf-scheduler-context-completed-projection-epoch context)
|
||||
projection-id
|
||||
(etaf-scheduler-context-fault-state context) nil)))
|
||||
(when first-condition
|
||||
(signal (car first-condition) (cdr first-condition)))))
|
||||
|
||||
(defun etaf-scheduler--finish-projection (projection)
|
||||
"Restore PROJECTION invariants and return contained finalizer errors."
|
||||
(let (errors)
|
||||
(dolist (context (etaf-scheduler--projection-all-contexts projection))
|
||||
(when (or (etaf-scheduler-context-source-queue context)
|
||||
(etaf-scheduler-context-deferred-source-queue context)
|
||||
(etaf-scheduler-context-runtime-queue context))
|
||||
(etaf-scheduler--reset-pending context))
|
||||
(setf (etaf-scheduler-context-phase context) nil)
|
||||
(clrhash (etaf-scheduler-context-delivered-source-set context))
|
||||
(clrhash (etaf-scheduler-context-effect-set context)))
|
||||
(let ((etaf-scheduler--active-projection nil))
|
||||
(dolist (finalizer
|
||||
(nreverse (etaf-scheduler--projection-finalizers projection)))
|
||||
(condition-case condition
|
||||
(funcall finalizer)
|
||||
((error quit) (push condition errors)))))
|
||||
(setf (etaf-scheduler--projection-finalizers projection) nil)
|
||||
(nreverse errors)))
|
||||
|
||||
(defun etaf-scheduler--notify-projection-complete (projection summary)
|
||||
"Notify PROJECTION observers with detached SUMMARY, containing failures."
|
||||
(let ((etaf-scheduler--active-projection nil)
|
||||
errors)
|
||||
(when (functionp etaf-scheduler-projection-observer)
|
||||
(condition-case condition
|
||||
(funcall etaf-scheduler-projection-observer (copy-tree summary))
|
||||
((error quit) (push condition errors))))
|
||||
(dolist
|
||||
(observer
|
||||
(nreverse
|
||||
(etaf-scheduler--projection-completion-observers projection)))
|
||||
(condition-case condition
|
||||
(funcall observer (copy-tree summary))
|
||||
((error quit) (push condition errors))))
|
||||
(setf (etaf-scheduler--projection-completion-observers projection) nil)
|
||||
(nreverse errors)))
|
||||
|
||||
(defun etaf-scheduler--metric-delta (before after key)
|
||||
"Return non-negative KEY delta between BEFORE and AFTER metric plists."
|
||||
(max 0 (- (or (plist-get after key) 0)
|
||||
(or (plist-get before key) 0))))
|
||||
|
||||
(defun etaf-scheduler--projection-summary (projection condition)
|
||||
"Return detached PROJECTION metrics with optional failure CONDITION."
|
||||
(let ((before-table
|
||||
(etaf-scheduler--projection-context-before projection))
|
||||
context-summaries
|
||||
(source-enqueues 0) (source-dedupes 0) (source-deliveries 0)
|
||||
(subscriber-visits 0) (effect-claims 0) (effect-evaluations 0)
|
||||
(runtime-enqueues 0) (runtime-dedupes 0) (runtime-executions 0)
|
||||
(stale-route-drops 0) (turns 0) (faults 0))
|
||||
(dolist (context (etaf-scheduler--projection-contexts projection))
|
||||
(let* ((before (gethash context before-table))
|
||||
(after (etaf-scheduler-context-metrics context))
|
||||
(entry
|
||||
(list
|
||||
:context-id (etaf-scheduler-context-id context)
|
||||
:source-enqueues
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :source-enqueues)
|
||||
:source-dedupes
|
||||
(etaf-scheduler--metric-delta before after :source-dedupes)
|
||||
:source-deliveries
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :source-deliveries)
|
||||
:subscriber-visits
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :subscriber-visits)
|
||||
:effect-claims
|
||||
(etaf-scheduler--metric-delta before after :effect-claims)
|
||||
:effect-evaluations
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :effect-evaluations)
|
||||
:runtime-enqueues
|
||||
(etaf-scheduler--metric-delta before after :runtime-enqueues)
|
||||
:runtime-dedupes
|
||||
(etaf-scheduler--metric-delta before after :runtime-dedupes)
|
||||
:runtime-executions
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :runtime-executions)
|
||||
:stale-route-drops
|
||||
(etaf-scheduler--metric-delta
|
||||
before after :stale-route-drops)
|
||||
:turns (etaf-scheduler--metric-delta before after :turn-count)
|
||||
:faults
|
||||
(etaf-scheduler--metric-delta before after :fault-count))))
|
||||
(cl-incf source-enqueues (plist-get entry :source-enqueues))
|
||||
(cl-incf source-dedupes (plist-get entry :source-dedupes))
|
||||
(cl-incf source-deliveries (plist-get entry :source-deliveries))
|
||||
(cl-incf subscriber-visits (plist-get entry :subscriber-visits))
|
||||
(cl-incf effect-claims (plist-get entry :effect-claims))
|
||||
(cl-incf effect-evaluations (plist-get entry :effect-evaluations))
|
||||
(cl-incf runtime-enqueues (plist-get entry :runtime-enqueues))
|
||||
(cl-incf runtime-dedupes (plist-get entry :runtime-dedupes))
|
||||
(cl-incf runtime-executions (plist-get entry :runtime-executions))
|
||||
(cl-incf stale-route-drops (plist-get entry :stale-route-drops))
|
||||
(cl-incf turns (plist-get entry :turns))
|
||||
(cl-incf faults (plist-get entry :faults))
|
||||
(push entry context-summaries)))
|
||||
(list :projection-id (etaf-scheduler--projection-id projection)
|
||||
:completed-p (null condition)
|
||||
:context-count (length context-summaries)
|
||||
:source-enqueues source-enqueues
|
||||
:source-dedupes source-dedupes
|
||||
:source-deliveries source-deliveries
|
||||
:subscriber-visits subscriber-visits
|
||||
:effect-claims effect-claims
|
||||
:effect-evaluations effect-evaluations
|
||||
:runtime-enqueues runtime-enqueues
|
||||
:runtime-dedupes runtime-dedupes
|
||||
:runtime-executions runtime-executions
|
||||
:stale-route-drops stale-route-drops
|
||||
:turns turns :faults faults
|
||||
:condition (and condition (copy-tree condition))
|
||||
:contexts (nreverse context-summaries))))
|
||||
|
||||
(defun etaf-scheduler-call-with-projection (function)
|
||||
"Call FUNCTION and drain all contexts touched by its reactive writes.
|
||||
Nested calls join the active projection. If FUNCTION signals after changing
|
||||
state, queued notifications still drain before the original condition is
|
||||
re-signaled."
|
||||
(unless (functionp function)
|
||||
(signal 'wrong-type-argument (list 'functionp function)))
|
||||
(if etaf-scheduler--active-projection
|
||||
(funcall function)
|
||||
(let* ((projection
|
||||
(etaf-scheduler--projection-create
|
||||
:id (cl-incf etaf-scheduler--projection-id-counter)
|
||||
:all-context-set (make-hash-table :test #'eq)
|
||||
:context-before (make-hash-table :test #'eq)))
|
||||
result primary-condition drain-condition summary finalizer-errors)
|
||||
(let ((etaf-scheduler--active-projection projection))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(condition-case condition
|
||||
(setq result (funcall function))
|
||||
((error quit) (setq primary-condition condition)))
|
||||
(condition-case condition
|
||||
(etaf-scheduler--drain-projection projection)
|
||||
((error quit) (setq drain-condition condition)))
|
||||
(setq summary
|
||||
(etaf-scheduler--projection-summary
|
||||
projection drain-condition)))
|
||||
(setq finalizer-errors
|
||||
(etaf-scheduler--finish-projection projection))
|
||||
(when finalizer-errors
|
||||
(setq summary
|
||||
(plist-put summary :completed-p nil)
|
||||
summary
|
||||
(plist-put summary :finalizer-errors
|
||||
(copy-tree finalizer-errors))))
|
||||
(etaf-scheduler--notify-projection-complete
|
||||
projection summary)))
|
||||
(cond
|
||||
(primary-condition
|
||||
(signal (car primary-condition) (cdr primary-condition)))
|
||||
(drain-condition
|
||||
(signal (car drain-condition) (cdr drain-condition)))
|
||||
(t result)))))
|
||||
|
||||
(defun etaf-scheduler-context-idle-p (context)
|
||||
"Return non-nil when CONTEXT has no queued or running dispatcher work."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(and (null (etaf-scheduler-context-source-queue context))
|
||||
(null (etaf-scheduler-context-deferred-source-queue context))
|
||||
(null (etaf-scheduler-context-runtime-queue context))
|
||||
(zerop (hash-table-count
|
||||
(etaf-scheduler-context-source-set context)))
|
||||
(zerop (hash-table-count
|
||||
(etaf-scheduler-context-runtime-set context)))
|
||||
(zerop (etaf-scheduler-context-busy-depth context))))
|
||||
|
||||
(defun etaf-scheduler-context-metrics (context)
|
||||
"Return a detached metrics plist for scheduler CONTEXT."
|
||||
(setq context (etaf-scheduler-context-resolve context))
|
||||
(list :context-id (etaf-scheduler-context-id context)
|
||||
:projection-epoch (etaf-scheduler-context-projection-epoch context)
|
||||
:completed-projection-epoch
|
||||
(etaf-scheduler-context-completed-projection-epoch context)
|
||||
:active-turn-id (etaf-scheduler-context-active-turn-id context)
|
||||
:turn-count (etaf-scheduler-context-turn-count context)
|
||||
:source-enqueues
|
||||
(etaf-scheduler-context-source-enqueue-count context)
|
||||
:source-dedupes
|
||||
(etaf-scheduler-context-source-dedupe-count context)
|
||||
:source-deliveries
|
||||
(etaf-scheduler-context-source-delivery-count context)
|
||||
:subscriber-visits
|
||||
(etaf-scheduler-context-subscriber-visit-count context)
|
||||
:effect-claims
|
||||
(etaf-scheduler-context-effect-claim-count context)
|
||||
:effect-evaluations
|
||||
(etaf-scheduler-context-effect-evaluation-count context)
|
||||
:runtime-enqueues
|
||||
(etaf-scheduler-context-runtime-enqueue-count context)
|
||||
:runtime-dedupes
|
||||
(etaf-scheduler-context-runtime-dedupe-count context)
|
||||
:runtime-executions
|
||||
(etaf-scheduler-context-runtime-execution-count context)
|
||||
:stale-route-drops
|
||||
(etaf-scheduler-context-stale-route-drop-count context)
|
||||
:fault-count
|
||||
(etaf-scheduler-context-projection-fault-count context)
|
||||
:fault-state
|
||||
(copy-tree (etaf-scheduler-context-fault-state context))))
|
||||
|
||||
(provide 'etaf-scheduler)
|
||||
|
||||
;;; etaf-scheduler.el ends here
|
||||
1
etaf.el
1
etaf.el
@ -30,6 +30,7 @@
|
||||
(require 'etaf-view)
|
||||
(require 'etaf-compiler)
|
||||
(require 'etaf-component)
|
||||
(require 'etaf-scheduler)
|
||||
(require 'etaf-reactive)
|
||||
(require 'etaf-observer)
|
||||
(require 'etaf-context)
|
||||
|
||||
142
scripts/benchmark-scheduler-context.el
Normal file
142
scripts/benchmark-scheduler-context.el
Normal file
@ -0,0 +1,142 @@
|
||||
;;; benchmark-scheduler-context.el --- ETAF dispatcher microbenchmark -*- lexical-binding: t; -*-
|
||||
|
||||
;; SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
;;; Commentary:
|
||||
|
||||
;; Runs the M3a T6 fixed scheduler scenario: 40 changed sources fan out to one
|
||||
;; Effect in each of four isolated contexts. Five warmups precede thirty timed
|
||||
;; samples. Exact work counters and a 50 ms p95/max gate make the result useful
|
||||
;; as a small dispatcher regression baseline; product GUI performance is a
|
||||
;; separate end-to-end gate.
|
||||
|
||||
;;; Code:
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'etaf)
|
||||
|
||||
(defconst etaf-scheduler-benchmark-warmups 5)
|
||||
(defconst etaf-scheduler-benchmark-samples 30)
|
||||
(defconst etaf-scheduler-benchmark-source-count 40)
|
||||
(defconst etaf-scheduler-benchmark-context-count 4)
|
||||
(defconst etaf-scheduler-benchmark-limit-ms 50.0)
|
||||
|
||||
(defun etaf-scheduler-benchmark--percentile (values percentile)
|
||||
"Return nearest-rank PERCENTILE from numeric VALUES."
|
||||
(let* ((ordered (sort (copy-sequence values) #'<))
|
||||
(rank (max 1 (ceiling (* percentile (length ordered))))))
|
||||
(nth (1- rank) ordered)))
|
||||
|
||||
(defun etaf-scheduler-benchmark--metric-total (contexts key)
|
||||
"Return the sum of metric KEY across scheduler CONTEXTS."
|
||||
(cl-loop for context in contexts
|
||||
sum (or (plist-get (etaf-scheduler-context-metrics context) key)
|
||||
0)))
|
||||
|
||||
(defun etaf-scheduler-benchmark--iteration (sources iteration)
|
||||
"Publish ITERATION through all reactive SOURCES once."
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(cl-loop for source in sources
|
||||
for index from 0
|
||||
do (setf (etaf-value source)
|
||||
(+ (* iteration 1000) index))))))
|
||||
|
||||
;;;###autoload
|
||||
(defun etaf-scheduler-benchmark-run ()
|
||||
"Run the fixed scheduler benchmark, print its result, and return it."
|
||||
(let* ((contexts
|
||||
(cl-loop for index below etaf-scheduler-benchmark-context-count
|
||||
collect
|
||||
(etaf-scheduler-context-create
|
||||
:name (list 'benchmark index))))
|
||||
(sources
|
||||
(cl-loop repeat etaf-scheduler-benchmark-source-count
|
||||
collect (etaf-ref 0)))
|
||||
effects durations result)
|
||||
(unwind-protect
|
||||
(progn
|
||||
(dolist (context contexts)
|
||||
(let ((effect
|
||||
(etaf-scheduler-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda () (mapcar #'etaf-value sources)))))))
|
||||
(push effect effects)
|
||||
(etaf-reactive-effect-run effect)))
|
||||
(dotimes (index etaf-scheduler-benchmark-warmups)
|
||||
(etaf-scheduler-benchmark--iteration sources (1+ index)))
|
||||
(let ((source-before
|
||||
(etaf-scheduler-benchmark--metric-total
|
||||
contexts :source-enqueues))
|
||||
(visit-before
|
||||
(etaf-scheduler-benchmark--metric-total
|
||||
contexts :subscriber-visits))
|
||||
(effect-before
|
||||
(etaf-scheduler-benchmark--metric-total
|
||||
contexts :effect-evaluations))
|
||||
(turn-before
|
||||
(etaf-scheduler-benchmark--metric-total
|
||||
contexts :turn-count)))
|
||||
(dotimes (index etaf-scheduler-benchmark-samples)
|
||||
(let ((started (float-time)))
|
||||
(etaf-scheduler-benchmark--iteration
|
||||
sources (+ etaf-scheduler-benchmark-warmups index 1))
|
||||
(push (* 1000.0 (- (float-time) started)) durations)))
|
||||
(setq durations (nreverse durations))
|
||||
(let* ((expected-source-work
|
||||
(* etaf-scheduler-benchmark-samples
|
||||
etaf-scheduler-benchmark-source-count
|
||||
etaf-scheduler-benchmark-context-count))
|
||||
(expected-context-work
|
||||
(* etaf-scheduler-benchmark-samples
|
||||
etaf-scheduler-benchmark-context-count))
|
||||
(source-work
|
||||
(- (etaf-scheduler-benchmark--metric-total
|
||||
contexts :source-enqueues)
|
||||
source-before))
|
||||
(subscriber-visits
|
||||
(- (etaf-scheduler-benchmark--metric-total
|
||||
contexts :subscriber-visits)
|
||||
visit-before))
|
||||
(effect-work
|
||||
(- (etaf-scheduler-benchmark--metric-total
|
||||
contexts :effect-evaluations)
|
||||
effect-before))
|
||||
(turn-work
|
||||
(- (etaf-scheduler-benchmark--metric-total
|
||||
contexts :turn-count)
|
||||
turn-before))
|
||||
(p95
|
||||
(etaf-scheduler-benchmark--percentile durations 0.95))
|
||||
(maximum (apply #'max durations)))
|
||||
(unless (and (= source-work expected-source-work)
|
||||
(= subscriber-visits expected-source-work)
|
||||
(= effect-work expected-context-work)
|
||||
(= turn-work expected-context-work))
|
||||
(error "Scheduler work counters diverged: %S"
|
||||
(list source-work subscriber-visits
|
||||
effect-work turn-work)))
|
||||
(setq result
|
||||
(list
|
||||
:scenario 'scheduler-40-sources-4-contexts
|
||||
:warmups etaf-scheduler-benchmark-warmups
|
||||
:samples etaf-scheduler-benchmark-samples
|
||||
:p95-ms p95 :max-ms maximum
|
||||
:source-enqueues source-work
|
||||
:subscriber-visits subscriber-visits
|
||||
:effect-evaluations effect-work
|
||||
:turns turn-work))
|
||||
(when (or (> p95 etaf-scheduler-benchmark-limit-ms)
|
||||
(> maximum etaf-scheduler-benchmark-limit-ms))
|
||||
(error "Scheduler benchmark exceeds %.1f ms: %S"
|
||||
etaf-scheduler-benchmark-limit-ms result)))))
|
||||
(dolist (effect effects) (etaf--stop-effect effect)))
|
||||
(prin1 result)
|
||||
(terpri)
|
||||
result))
|
||||
|
||||
(provide 'benchmark-scheduler-context)
|
||||
|
||||
;;; benchmark-scheduler-context.el ends here
|
||||
733
tests/etaf-scheduler-tests.el
Normal file
733
tests/etaf-scheduler-tests.el
Normal file
@ -0,0 +1,733 @@
|
||||
;;; etaf-scheduler-tests.el --- M3a dispatcher context gates -*- lexical-binding: t; -*-
|
||||
|
||||
;; SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
(require 'cl-lib)
|
||||
(require 'ert)
|
||||
(require 'etaf)
|
||||
|
||||
(define-error 'etaf-scheduler-test-condition
|
||||
"ETAF scheduler test condition")
|
||||
(define-error 'etaf-scheduler-test-source-condition
|
||||
"ETAF scheduler source test condition")
|
||||
(define-error 'etaf-scheduler-test-projection-condition
|
||||
"ETAF scheduler projection test condition")
|
||||
(define-error 'etaf-scheduler-test-body-condition
|
||||
"ETAF scheduler body test condition")
|
||||
|
||||
(etaf-define-component etaf-scheduler-test-pair (&key left right)
|
||||
"Render reactive LEFT and RIGHT values."
|
||||
:view
|
||||
(text (expr (format "%s/%s" (etaf-value left) (etaf-value right)))))
|
||||
|
||||
(etaf-define-component etaf-scheduler-test-single (&key source)
|
||||
"Render one reactive SOURCE."
|
||||
:view (text (expr (etaf-value source))))
|
||||
|
||||
(etaf-define-component etaf-scheduler-test-data (&key controller)
|
||||
"Render CONTROLLER status, item count, and total."
|
||||
:view
|
||||
(text
|
||||
(expr
|
||||
(format "%s:%d:%d"
|
||||
(etaf-value (etaf-data-status controller))
|
||||
(length (etaf-value (etaf-data-items controller)))
|
||||
(etaf-value (etaf-data-total controller))))))
|
||||
|
||||
(etaf-define-component etaf-scheduler-test-data-projection-failure
|
||||
(&key controller)
|
||||
"Fail rendering when CONTROLLER reaches success."
|
||||
:render
|
||||
(let ((status (etaf-value (etaf-data-status controller))))
|
||||
(when (eq status 'success)
|
||||
(signal 'etaf-scheduler-test-projection-condition '("projection")))
|
||||
(etaf-node 'text nil (list (symbol-name status)))))
|
||||
|
||||
(defun etaf-scheduler-test--text (buffer-name)
|
||||
"Return BUFFER-NAME text without properties."
|
||||
(with-current-buffer buffer-name
|
||||
(substring-no-properties (buffer-string))))
|
||||
|
||||
(defun etaf-scheduler-test--metric (context key)
|
||||
"Return scheduler CONTEXT metric KEY."
|
||||
(plist-get (etaf-scheduler-context-metrics context) key))
|
||||
|
||||
(defun etaf-scheduler-test--source (symbol)
|
||||
"Return Lisp source text defining SYMBOL."
|
||||
(let* ((loaded (symbol-file symbol 'defun))
|
||||
(source (if (and loaded (string-suffix-p ".elc" loaded))
|
||||
(substring loaded 0 -1)
|
||||
loaded)))
|
||||
(with-temp-buffer
|
||||
(insert-file-contents source)
|
||||
(buffer-string))))
|
||||
|
||||
(ert-deftest etaf-scheduler-is-the-only-dispatch-state-owner ()
|
||||
"Reactive semantics use the scheduler and own no process-global queues."
|
||||
(let ((reactive (etaf-scheduler-test--source 'etaf--dispatch-source))
|
||||
(scheduler
|
||||
(etaf-scheduler-test--source 'etaf-scheduler-context-create)))
|
||||
(should-not (string-match-p "(defvar etaf--dispatch-" reactive))
|
||||
(dolist (field '("source-queue" "runtime-queue" "effect-set"
|
||||
"active-turn-id" "projection-epoch" "fault-state"))
|
||||
(should (string-match-p field scheduler)))))
|
||||
|
||||
(ert-deftest etaf-scheduler-shared-sources-fan-out-once-per-context ()
|
||||
"Two changed refs schedule each isolated Runtime exactly once."
|
||||
(let* ((left-buffer " *etaf-scheduler-left*")
|
||||
(left-peer-buffer " *etaf-scheduler-left-peer*")
|
||||
(right-buffer " *etaf-scheduler-right*")
|
||||
(left-context (etaf-scheduler-context-create :name 'left))
|
||||
(right-context (etaf-scheduler-context-create :name 'right))
|
||||
(left (etaf-ref "A"))
|
||||
(right (etaf-ref "1"))
|
||||
(view (etaf--view-call
|
||||
'etaf-scheduler-test-pair
|
||||
(list :left left :right right) nil)))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount left-buffer view
|
||||
(list :scheduler-context left-context))
|
||||
(etaf-mount left-peer-buffer view
|
||||
(list :scheduler-context left-context))
|
||||
(etaf-mount right-buffer view
|
||||
(list :scheduler-context right-context))
|
||||
(let* ((left-runtime (etaf-runtime-for-buffer left-buffer))
|
||||
(left-peer-runtime
|
||||
(etaf-runtime-for-buffer left-peer-buffer))
|
||||
(right-runtime (etaf-runtime-for-buffer right-buffer))
|
||||
(left-generation (etaf-runtime-generation left-runtime))
|
||||
(left-peer-generation
|
||||
(etaf-runtime-generation left-peer-runtime))
|
||||
(right-generation (etaf-runtime-generation right-runtime))
|
||||
(left-source-before
|
||||
(etaf-scheduler-test--metric
|
||||
left-context :source-enqueues))
|
||||
(right-source-before
|
||||
(etaf-scheduler-test--metric
|
||||
right-context :source-enqueues))
|
||||
(left-runtime-before
|
||||
(etaf-scheduler-test--metric
|
||||
left-context :runtime-enqueues))
|
||||
(right-runtime-before
|
||||
(etaf-scheduler-test--metric
|
||||
right-context :runtime-enqueues)))
|
||||
(should (eq left-context
|
||||
(etaf-runtime-scheduler-context left-runtime)))
|
||||
(should (eq right-context
|
||||
(etaf-runtime-scheduler-context right-runtime)))
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value left) "B")
|
||||
(setf (etaf-value right) "2")))
|
||||
(should (equal "B/2" (etaf-scheduler-test--text left-buffer)))
|
||||
(should (equal "B/2"
|
||||
(etaf-scheduler-test--text left-peer-buffer)))
|
||||
(should (equal "B/2" (etaf-scheduler-test--text right-buffer)))
|
||||
(should (= 1 (- (etaf-runtime-generation left-runtime)
|
||||
left-generation)))
|
||||
(should (= 1 (- (etaf-runtime-generation left-peer-runtime)
|
||||
left-peer-generation)))
|
||||
(should (= 1 (- (etaf-runtime-generation right-runtime)
|
||||
right-generation)))
|
||||
(should (= 2 (- (etaf-scheduler-test--metric
|
||||
left-context :source-enqueues)
|
||||
left-source-before)))
|
||||
(should (= 2 (- (etaf-scheduler-test--metric
|
||||
right-context :source-enqueues)
|
||||
right-source-before)))
|
||||
(should (= 2 (- (etaf-scheduler-test--metric
|
||||
left-context :runtime-enqueues)
|
||||
left-runtime-before)))
|
||||
(should (= 1 (- (etaf-scheduler-test--metric
|
||||
right-context :runtime-enqueues)
|
||||
right-runtime-before)))
|
||||
(should (etaf-scheduler-context-idle-p left-context))
|
||||
(should (etaf-scheduler-context-idle-p right-context))))
|
||||
(dolist (buffer-name
|
||||
(list left-buffer left-peer-buffer right-buffer))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer buffer-name)))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer))))))
|
||||
|
||||
(ert-deftest etaf-scheduler-cross-context-computed-settles-before-runtime ()
|
||||
"A computed in one context settles before another context publishes."
|
||||
(let* ((buffer-name " *etaf-scheduler-computed*")
|
||||
(computed-context
|
||||
(etaf-scheduler-context-create :name 'computed-owner))
|
||||
(runtime-context
|
||||
(etaf-scheduler-context-create :name 'computed-consumer))
|
||||
(base (etaf-ref 1))
|
||||
(computed
|
||||
(etaf-scheduler-call-with-context
|
||||
computed-context
|
||||
(lambda ()
|
||||
(etaf-computed (lambda () (* 2 (etaf-value base))))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-pair
|
||||
(list :left base :right computed) nil)
|
||||
(list :scheduler-context runtime-context))
|
||||
(let* ((runtime (etaf-runtime-for-buffer buffer-name))
|
||||
(generation (etaf-runtime-generation runtime))
|
||||
(executions
|
||||
(etaf-scheduler-test--metric
|
||||
runtime-context :runtime-executions)))
|
||||
(should (eq computed-context
|
||||
(etaf-effect-scheduler-context
|
||||
(etaf-computed-effect computed))))
|
||||
(setf (etaf-value base) 2)
|
||||
(should (equal "2/4"
|
||||
(etaf-scheduler-test--text buffer-name)))
|
||||
(should (= 1 (- (etaf-runtime-generation runtime) generation)))
|
||||
(should (= 1 (- (etaf-scheduler-test--metric
|
||||
runtime-context :runtime-executions)
|
||||
executions)))))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer buffer-name)))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer))
|
||||
(etaf--stop-effect (etaf-computed-effect computed)))))
|
||||
|
||||
(ert-deftest etaf-scheduler-context-fault-does-not-suppress-peer ()
|
||||
"A failing context is contained until every peer receives the source."
|
||||
(let* ((source (etaf-ref 0))
|
||||
(left-context (etaf-scheduler-context-create :name 'fault-left))
|
||||
(right-context (etaf-scheduler-context-create :name 'fault-right))
|
||||
(fail-p nil)
|
||||
(right-value nil)
|
||||
left-effect right-effect captured)
|
||||
(setq left-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
left-context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(prog1 (etaf-value source)
|
||||
(when fail-p
|
||||
(signal 'etaf-scheduler-test-condition '("left")))))))))
|
||||
(setq right-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
right-context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda () (setq right-value (etaf-value source)))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-reactive-effect-run left-effect)
|
||||
(etaf-reactive-effect-run right-effect)
|
||||
(setq fail-p t)
|
||||
(condition-case condition
|
||||
(setf (etaf-value source) 1)
|
||||
(etaf-scheduler-test-condition (setq captured condition)))
|
||||
(should captured)
|
||||
(should (= right-value 1))
|
||||
(should (eq (car (etaf-scheduler-context-fault-state left-context))
|
||||
'etaf-scheduler-test-condition))
|
||||
(should-not (etaf-scheduler-context-fault-state right-context))
|
||||
(should (etaf-scheduler-context-idle-p left-context))
|
||||
(should (etaf-scheduler-context-idle-p right-context)))
|
||||
(etaf--stop-effect left-effect)
|
||||
(etaf--stop-effect right-effect))))
|
||||
|
||||
(ert-deftest etaf-scheduler-body-error-precedes-drain-error-and-quit-drains ()
|
||||
"Body failure wins over drain failure; quit still drains queued effects."
|
||||
(let* ((context (etaf-scheduler-context-create :name 'precedence))
|
||||
(source (etaf-ref 0))
|
||||
(quit-source (etaf-ref 0))
|
||||
(fail-p nil) (quit-seen nil) finalizer-ran summary
|
||||
effect quit-effect captured quit-captured)
|
||||
(setq effect
|
||||
(etaf-scheduler-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(prog1 (etaf-value source)
|
||||
(when fail-p
|
||||
(signal 'etaf-scheduler-test-condition '("drain")))))))))
|
||||
(setq quit-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(setq quit-seen (etaf-value quit-source)))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-reactive-effect-run effect)
|
||||
(etaf-reactive-effect-run quit-effect)
|
||||
(setq fail-p t)
|
||||
(let ((etaf-scheduler-projection-observer
|
||||
(lambda (value) (setq summary value))))
|
||||
(condition-case condition
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(etaf-scheduler-defer-finalizer
|
||||
(lambda () (error "contained finalizer")))
|
||||
(etaf-scheduler-defer-finalizer
|
||||
(lambda () (setq finalizer-ran t)))
|
||||
(setf (etaf-value source) 1)
|
||||
(signal 'etaf-scheduler-test-body-condition '("body"))))
|
||||
(etaf-scheduler-test-body-condition
|
||||
(setq captured condition))))
|
||||
(should (equal captured
|
||||
'(etaf-scheduler-test-body-condition "body")))
|
||||
(should finalizer-ran)
|
||||
(should (= 1 (length (plist-get summary :finalizer-errors))))
|
||||
(should (eq (car (etaf-scheduler-context-fault-state context))
|
||||
'etaf-scheduler-test-condition))
|
||||
(setq fail-p nil)
|
||||
(condition-case condition
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value quit-source) 1)
|
||||
(signal 'quit nil)))
|
||||
(quit (setq quit-captured condition)))
|
||||
(should (eq (car quit-captured) 'quit))
|
||||
(should (= quit-seen 1))
|
||||
(should (etaf-scheduler-context-idle-p context)))
|
||||
(etaf--stop-effect effect)
|
||||
(etaf--stop-effect quit-effect))))
|
||||
|
||||
(ert-deftest etaf-scheduler-reentrant-source-runs-in-following-turn ()
|
||||
"A same-source reentrant write is deferred to one following turn."
|
||||
(let* ((context (etaf-scheduler-context-create :name 'reentrant))
|
||||
(source (etaf-ref 0))
|
||||
values turns effect)
|
||||
(setq effect
|
||||
(etaf-scheduler-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(let ((value (etaf-value source)))
|
||||
(push value values)
|
||||
(push (etaf-scheduler-context-active-turn-id context)
|
||||
turns)
|
||||
(when (= value 1)
|
||||
(setf (etaf-value source) 2))))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-reactive-effect-run effect)
|
||||
(let ((deliveries
|
||||
(etaf-scheduler-test--metric
|
||||
context :source-deliveries))
|
||||
(turn-count
|
||||
(etaf-scheduler-test--metric context :turn-count)))
|
||||
(setf (etaf-value source) 1)
|
||||
(should (equal '(0 1 2) (nreverse values)))
|
||||
(should (equal '(0 1 2) (nreverse turns)))
|
||||
(should (= 2 (- (etaf-scheduler-test--metric
|
||||
context :source-deliveries)
|
||||
deliveries)))
|
||||
(should (= 2 (- (etaf-scheduler-test--metric
|
||||
context :turn-count)
|
||||
turn-count)))))
|
||||
(etaf--stop-effect effect))))
|
||||
|
||||
(ert-deftest etaf-scheduler-cross-context-cycle-is-bounded-and-reusable ()
|
||||
"Cross-context ping-pong fails at the exact budget and leaves idle contexts."
|
||||
(let* ((left-context
|
||||
(etaf-scheduler-context-create
|
||||
:name 'cycle-left :fixed-point-step-budget 3))
|
||||
(right-context
|
||||
(etaf-scheduler-context-create
|
||||
:name 'cycle-right :fixed-point-step-budget 3))
|
||||
(left-source (etaf-ref 0))
|
||||
(right-source (etaf-ref 0))
|
||||
left-effect right-effect captured recovery-effect)
|
||||
(setq left-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
left-context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(let ((value (etaf-value left-source)))
|
||||
(when (> value 0)
|
||||
(setf (etaf-value right-source) value))))))))
|
||||
(setq right-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
right-context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(let ((value (etaf-value right-source)))
|
||||
(when (> value 0)
|
||||
(setf (etaf-value left-source) (1+ value)))))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-reactive-effect-run left-effect)
|
||||
(etaf-reactive-effect-run right-effect)
|
||||
(condition-case condition
|
||||
(setf (etaf-value left-source) 1)
|
||||
(etaf-scheduler-error (setq captured condition)))
|
||||
(should captured)
|
||||
(should (eq 'fixed-point-step-budget
|
||||
(plist-get (cdr captured) :kind)))
|
||||
(should (= 4 (plist-get (cdr captured) :steps)))
|
||||
(should (= 3 (plist-get (cdr captured) :budget)))
|
||||
(should (etaf-scheduler-context-idle-p left-context))
|
||||
(should (etaf-scheduler-context-idle-p right-context))
|
||||
(etaf--stop-effect left-effect)
|
||||
(etaf--stop-effect right-effect)
|
||||
(let ((recovery-source (etaf-ref 0)) (seen nil))
|
||||
(setq recovery-effect
|
||||
(etaf-scheduler-call-with-context
|
||||
left-context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(setq seen (etaf-value recovery-source)))))))
|
||||
(etaf-reactive-effect-run recovery-effect)
|
||||
(setf (etaf-value recovery-source) 1)
|
||||
(should (= seen 1))
|
||||
(should-not
|
||||
(etaf-scheduler-context-fault-state left-context))))
|
||||
(when (and left-effect (etaf-effect-active-p left-effect))
|
||||
(etaf--stop-effect left-effect))
|
||||
(when (and right-effect (etaf-effect-active-p right-effect))
|
||||
(etaf--stop-effect right-effect))
|
||||
(when recovery-effect (etaf--stop-effect recovery-effect)))))
|
||||
|
||||
(ert-deftest etaf-scheduler-data-success-is-one-turn-per-context ()
|
||||
"One Data success projection coalesces all changed refs in each context."
|
||||
(let* ((left-buffer " *etaf-scheduler-data-left*")
|
||||
(right-buffer " *etaf-scheduler-data-right*")
|
||||
(left-context (etaf-scheduler-context-create :name 'data-left))
|
||||
(right-context (etaf-scheduler-context-create :name 'data-right))
|
||||
left-runtime-before right-runtime-before
|
||||
(source
|
||||
(etaf-data-source
|
||||
:load
|
||||
(lambda (_query _page _page-size)
|
||||
(setq left-runtime-before
|
||||
(etaf-scheduler-test--metric
|
||||
left-context :runtime-enqueues)
|
||||
right-runtime-before
|
||||
(etaf-scheduler-test--metric
|
||||
right-context :runtime-enqueues))
|
||||
(list :items '(one two three) :total 9))))
|
||||
(controller (etaf-data-controller source))
|
||||
(view (etaf--view-call
|
||||
'etaf-scheduler-test-data
|
||||
(list :controller controller) nil)))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount left-buffer view
|
||||
(list :scheduler-context left-context))
|
||||
(etaf-mount right-buffer view
|
||||
(list :scheduler-context right-context))
|
||||
(etaf-data-load controller)
|
||||
(should (equal "success:3:9"
|
||||
(etaf-scheduler-test--text left-buffer)))
|
||||
(should (equal "success:3:9"
|
||||
(etaf-scheduler-test--text right-buffer)))
|
||||
(should (= 1 (- (etaf-scheduler-test--metric
|
||||
left-context :runtime-enqueues)
|
||||
left-runtime-before)))
|
||||
(should (= 1 (- (etaf-scheduler-test--metric
|
||||
right-context :runtime-enqueues)
|
||||
right-runtime-before))))
|
||||
(dolist (buffer-name (list left-buffer right-buffer))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer buffer-name)))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer)))
|
||||
(etaf-data-stop controller))))
|
||||
|
||||
(ert-deftest etaf-scheduler-data-separates-source-and-projection-errors ()
|
||||
"Source failure sets Data error; render failure leaves successful refs."
|
||||
(let* ((buffer-name " *etaf-scheduler-data-projection-error*")
|
||||
(context (etaf-scheduler-context-create :name 'data-projection))
|
||||
(source
|
||||
(etaf-data-source
|
||||
:load (lambda (_query _page _page-size)
|
||||
(list :items '(committed) :total 1))))
|
||||
(controller (etaf-data-controller source))
|
||||
projection-condition)
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-data-projection-failure
|
||||
(list :controller controller) nil)
|
||||
(list :scheduler-context context))
|
||||
(condition-case condition
|
||||
(etaf-data-load controller)
|
||||
(etaf-scheduler-test-projection-condition
|
||||
(setq projection-condition condition)))
|
||||
(should projection-condition)
|
||||
(should (eq 'success
|
||||
(etaf-value (etaf-data-status controller))))
|
||||
(should-not (etaf-value (etaf-data-error controller)))
|
||||
(should (equal '(committed)
|
||||
(etaf-value (etaf-data-items controller))))
|
||||
(should (equal "loading"
|
||||
(etaf-scheduler-test--text buffer-name))))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer buffer-name)))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer))
|
||||
(etaf-data-stop controller)))
|
||||
(let* ((source
|
||||
(etaf-data-source
|
||||
:load (lambda (_query _page _page-size)
|
||||
(signal 'etaf-scheduler-test-source-condition
|
||||
'("source")))))
|
||||
(controller (etaf-data-controller source))
|
||||
captured)
|
||||
(unwind-protect
|
||||
(progn
|
||||
(condition-case condition
|
||||
(etaf-data-load controller)
|
||||
(etaf-scheduler-test-source-condition
|
||||
(setq captured condition)))
|
||||
(should captured)
|
||||
(should (eq 'error (etaf-value (etaf-data-status controller))))
|
||||
(should (equal captured
|
||||
(etaf-value (etaf-data-error controller)))))
|
||||
(etaf-data-stop controller))))
|
||||
|
||||
(ert-deftest etaf-scheduler-detached-route-does-not-enter-fan-out ()
|
||||
"An invalidated Runtime route cannot enqueue work in its old context."
|
||||
(let* ((buffer-name " *etaf-scheduler-stale-route*")
|
||||
(context (etaf-scheduler-context-create :name 'stale))
|
||||
(source (etaf-ref "A")))
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call 'etaf-scheduler-test-single (list :source source) nil)
|
||||
(list :scheduler-context context))
|
||||
(let* ((runtime (etaf-runtime-for-buffer buffer-name))
|
||||
(route (etaf-runtime-route-token runtime)))
|
||||
(etaf-unmount runtime)
|
||||
(let ((before (etaf-scheduler-test--metric context :source-enqueues)))
|
||||
(should-not (etaf-runtime-route-active-p route))
|
||||
(setf (etaf-value source) "B")
|
||||
(should (= before
|
||||
(etaf-scheduler-test--metric
|
||||
context :source-enqueues)))))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer))))
|
||||
|
||||
(ert-deftest etaf-scheduler-stale-authority-is-filtered-before-fan-out ()
|
||||
"Registry, token, attaching, and detached mismatches enqueue no source."
|
||||
(let* ((buffer-name " *etaf-scheduler-stale-authority*")
|
||||
(context (etaf-scheduler-context-create :name 'stale-authority))
|
||||
(source (etaf-ref "A")))
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call 'etaf-scheduler-test-single (list :source source) nil)
|
||||
(list :scheduler-context context))
|
||||
(let* ((runtime (etaf-runtime-for-buffer buffer-name))
|
||||
(route (etaf-runtime-route-token runtime))
|
||||
(epoch (etaf-runtime-mount-epoch runtime))
|
||||
(authority (etaf-runtime-host-authority runtime))
|
||||
(slots (etaf-host-authority-slots authority))
|
||||
(token (etaf-runtime-route-authority-token route))
|
||||
(state (etaf-host-authority-state authority))
|
||||
(drops (etaf-scheduler-test--metric
|
||||
context :stale-route-drops))
|
||||
(assert-drop
|
||||
(lambda (value)
|
||||
(let ((before (etaf-scheduler-test--metric
|
||||
context :source-enqueues)))
|
||||
(setf (etaf-value source) value)
|
||||
(should (= before
|
||||
(etaf-scheduler-test--metric
|
||||
context :source-enqueues)))))))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(remhash epoch etaf--runtime-route-registry)
|
||||
(funcall assert-drop "missing-registry")
|
||||
(puthash epoch runtime etaf--runtime-route-registry)
|
||||
(setf (etaf-runtime-route-authority-token route)
|
||||
(make-symbol "stale-token"))
|
||||
(funcall assert-drop "wrong-token")
|
||||
(setf (etaf-runtime-route-authority-token route) token)
|
||||
(aset slots 0 'attaching)
|
||||
(funcall assert-drop "attaching")
|
||||
(aset slots 0 'detached)
|
||||
(funcall assert-drop "detached")
|
||||
(aset slots 0 state)
|
||||
(should (= 4 (- (etaf-scheduler-test--metric
|
||||
context :stale-route-drops)
|
||||
drops)))
|
||||
(setf (etaf-value source) "live")
|
||||
(should (equal "live"
|
||||
(etaf-scheduler-test--text buffer-name))))
|
||||
(puthash epoch runtime etaf--runtime-route-registry)
|
||||
(setf (etaf-runtime-route-authority-token route) token)
|
||||
(aset slots 0 state)
|
||||
(when (etaf-runtime-mounted-p runtime)
|
||||
(etaf-unmount runtime))))
|
||||
(when-let* ((buffer (get-buffer buffer-name)))
|
||||
(kill-buffer buffer))))
|
||||
|
||||
(ert-deftest etaf-scheduler-operation-report-exposes-cost-counters ()
|
||||
"Observed operations expose scheduler enqueue, dedupe, and turn costs."
|
||||
(let* ((buffer-name " *etaf-scheduler-observer*")
|
||||
(peer-buffer " *etaf-scheduler-observer-peer*")
|
||||
(context (etaf-scheduler-context-create :name 'observer))
|
||||
(peer-context
|
||||
(etaf-scheduler-context-create :name 'observer-peer))
|
||||
(left (etaf-ref "A"))
|
||||
(right (etaf-ref "1"))
|
||||
reports)
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-pair
|
||||
(list :left left :right right) nil)
|
||||
(list :scheduler-context context
|
||||
:observer (lambda (report) (push report reports))))
|
||||
(etaf-mount
|
||||
peer-buffer
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-pair
|
||||
(list :left left :right right) nil)
|
||||
(list :scheduler-context peer-context))
|
||||
(setq reports nil)
|
||||
(let ((runtime (etaf-runtime-for-buffer buffer-name)))
|
||||
(etaf-runtime-call-operation
|
||||
runtime 'scheduler-test "scheduler counters"
|
||||
(lambda ()
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(setf (etaf-value left) "B")
|
||||
(setf (etaf-value right) "2")))))
|
||||
(let ((final (car reports)))
|
||||
(should (eq 'runtime-operation (plist-get final :stage)))
|
||||
(should (= (etaf-scheduler-context-id context)
|
||||
(plist-get final :scheduler-context-id)))
|
||||
(should (= 2 (plist-get final :scheduler-source-enqueues)))
|
||||
(should (= 2 (plist-get final :scheduler-source-deliveries)))
|
||||
(should (= 2 (plist-get final :scheduler-subscriber-visits)))
|
||||
(should (= 1 (plist-get final :scheduler-runtime-enqueues)))
|
||||
(should (= 1 (plist-get final :scheduler-runtime-dedupes)))
|
||||
(should (= 1 (plist-get final :scheduler-runtime-executions)))
|
||||
(should (= 1 (plist-get final :scheduler-turns)))
|
||||
(let ((projection
|
||||
(car (plist-get final :scheduler-projections))))
|
||||
(should (= 2 (plist-get projection :context-count)))
|
||||
(should (= 4 (plist-get projection :source-enqueues)))
|
||||
(should (= 4 (plist-get projection :subscriber-visits)))
|
||||
(should (= 2 (plist-get projection :runtime-executions))))
|
||||
(should (> (plist-get final :scheduler-projection-epoch-after)
|
||||
(plist-get final
|
||||
:scheduler-projection-epoch-before))))))
|
||||
(dolist (name (list buffer-name peer-buffer))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer name)))
|
||||
(when (eq name buffer-name)
|
||||
(etaf-runtime-set-observer runtime nil))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer name)))
|
||||
(kill-buffer buffer))))))
|
||||
|
||||
(ert-deftest etaf-scheduler-real-flush-defers-final-report-to-projection-end ()
|
||||
"A direct ref write reports the complete cross-context projection."
|
||||
(let* ((buffer-name " *etaf-scheduler-real-flush*")
|
||||
(peer-buffer " *etaf-scheduler-real-flush-peer*")
|
||||
(context (etaf-scheduler-context-create :name 'real-flush))
|
||||
(peer-context
|
||||
(etaf-scheduler-context-create :name 'real-flush-peer))
|
||||
(source (etaf-ref "A")) reports)
|
||||
(unwind-protect
|
||||
(progn
|
||||
(etaf-mount
|
||||
buffer-name
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-single (list :source source) nil)
|
||||
(list :scheduler-context context
|
||||
:observer (lambda (report) (push report reports))))
|
||||
(etaf-mount
|
||||
peer-buffer
|
||||
(etaf--view-call
|
||||
'etaf-scheduler-test-single (list :source source) nil)
|
||||
(list :scheduler-context peer-context))
|
||||
(setq reports nil)
|
||||
(setf (etaf-value source) "B")
|
||||
(let* ((final (car reports))
|
||||
(projection
|
||||
(car (plist-get final :scheduler-projections))))
|
||||
(should (eq 'runtime-operation (plist-get final :stage)))
|
||||
(should (= 2 (plist-get projection :context-count)))
|
||||
(should (= 2 (plist-get projection :source-enqueues)))
|
||||
(should (= 2 (plist-get projection :subscriber-visits)))
|
||||
(should (= 2 (plist-get projection :runtime-executions)))
|
||||
(should (equal "B" (etaf-scheduler-test--text buffer-name)))
|
||||
(should (equal "B" (etaf-scheduler-test--text peer-buffer)))))
|
||||
(dolist (name (list buffer-name peer-buffer))
|
||||
(when-let* ((runtime (etaf-runtime-for-buffer name)))
|
||||
(when (equal name buffer-name)
|
||||
(etaf-runtime-set-observer runtime nil))
|
||||
(etaf-unmount runtime))
|
||||
(when-let* ((buffer (get-buffer name)))
|
||||
(kill-buffer buffer))))))
|
||||
|
||||
(ert-deftest etaf-scheduler-scale-matrix-scans-each-subscriber-once ()
|
||||
"Source-by-context fan-out has exact linear subscriber and effect costs."
|
||||
(let* ((contexts
|
||||
(cl-loop for index below 4
|
||||
collect (etaf-scheduler-context-create
|
||||
:name (list 'scale index))))
|
||||
(sources (cl-loop repeat 40 collect (etaf-ref 0)))
|
||||
effects summary
|
||||
(started (float-time)))
|
||||
(unwind-protect
|
||||
(progn
|
||||
(dolist (context contexts)
|
||||
(let ((effect
|
||||
(etaf-scheduler-call-with-context
|
||||
context
|
||||
(lambda ()
|
||||
(etaf-reactive-effect-create
|
||||
(lambda ()
|
||||
(mapcar #'etaf-value sources)))))))
|
||||
(push effect effects)
|
||||
(etaf-reactive-effect-run effect)))
|
||||
(let ((etaf-scheduler-projection-observer
|
||||
(lambda (value) (setq summary value))))
|
||||
(etaf-reactive-call-with-batch
|
||||
(lambda ()
|
||||
(cl-loop for source in sources
|
||||
for value from 1
|
||||
do (setf (etaf-value source) value)))))
|
||||
(should (= 4 (plist-get summary :context-count)))
|
||||
(should (= 160 (plist-get summary :source-enqueues)))
|
||||
(should (= 160 (plist-get summary :source-deliveries)))
|
||||
(should (= 160 (plist-get summary :subscriber-visits)))
|
||||
(should (= 4 (plist-get summary :effect-claims)))
|
||||
(should (= 4 (plist-get summary :effect-evaluations)))
|
||||
(should (= 4 (plist-get summary :turns)))
|
||||
(should (< (* 1000.0 (- (float-time) started)) 1000.0)))
|
||||
(dolist (effect effects) (etaf--stop-effect effect)))))
|
||||
|
||||
(ert-deftest etaf-scheduler-fifo-cost-counters-are-linear ()
|
||||
"A bounded synthetic turn records one enqueue and one dedupe per key."
|
||||
(let* ((context (etaf-scheduler-context-create :name 'cost))
|
||||
(sources (cl-loop repeat 500 collect (list (make-symbol "source"))))
|
||||
(started (float-time)))
|
||||
(etaf-scheduler-call-with-projection
|
||||
(lambda ()
|
||||
(dolist (source sources)
|
||||
(etaf-scheduler-enqueue-source context source #'ignore)
|
||||
(etaf-scheduler-enqueue-source context source #'ignore))))
|
||||
(let ((elapsed-ms (* 1000.0 (- (float-time) started))))
|
||||
(should (= 500 (etaf-scheduler-test--metric
|
||||
context :source-enqueues)))
|
||||
(should (= 500 (etaf-scheduler-test--metric
|
||||
context :source-dedupes)))
|
||||
(should (< elapsed-ms 1000.0))
|
||||
(should (etaf-scheduler-context-idle-p context)))))
|
||||
|
||||
(provide 'etaf-scheduler-tests)
|
||||
|
||||
;;; etaf-scheduler-tests.el ends here
|
||||
@ -2240,24 +2240,26 @@ Event composition is a Runtime contract, not a UI-library helper contract."
|
||||
|
||||
(ert-deftest etaf-reactive-dispatch-fifos-append-in-constant-time-order ()
|
||||
"Keep source and Runtime dispatch order with explicit FIFO tails."
|
||||
(let ((etaf--dispatch-depth 1)
|
||||
(etaf--dispatch-source-queue nil)
|
||||
(etaf--dispatch-source-queue-tail nil)
|
||||
(etaf--dispatch-source-set (make-hash-table :test #'eq))
|
||||
(etaf--dispatch-runtime-queue nil)
|
||||
(etaf--dispatch-runtime-queue-tail nil)
|
||||
(etaf--dispatch-runtime-set (make-hash-table :test #'eq))
|
||||
(let ((context (etaf-scheduler-context-create :name 'fifo-test))
|
||||
(sources (cl-loop repeat 100 collect (etaf-ref nil))))
|
||||
(dolist (source sources)
|
||||
(etaf--dispatch-source source))
|
||||
(etaf--dispatch-source (car sources))
|
||||
(etaf-reactive-enqueue-runtime-flush 'first #'ignore)
|
||||
(etaf-reactive-enqueue-runtime-flush 'second #'ignore)
|
||||
(etaf-reactive-enqueue-runtime-flush 'first #'ignore)
|
||||
(should (equal sources etaf--dispatch-source-queue))
|
||||
(should (eq (car etaf--dispatch-source-queue-tail) (car (last sources))))
|
||||
(should (equal '(first second) etaf--dispatch-runtime-queue))
|
||||
(should (eq 'second (car etaf--dispatch-runtime-queue-tail)))))
|
||||
(etaf-scheduler-call-with-projection
|
||||
(lambda ()
|
||||
(dolist (source sources)
|
||||
(etaf-scheduler-enqueue-source context source #'ignore))
|
||||
(etaf-scheduler-enqueue-source context (car sources) #'ignore)
|
||||
(etaf-scheduler-enqueue-runtime context 'first #'ignore)
|
||||
(etaf-scheduler-enqueue-runtime context 'second #'ignore)
|
||||
(etaf-scheduler-enqueue-runtime context 'first #'ignore)
|
||||
(should (equal sources
|
||||
(etaf-scheduler-context-source-queue context)))
|
||||
(should (eq (car (etaf-scheduler-context-source-queue-tail context))
|
||||
(car (last sources))))
|
||||
(should (equal '(first second)
|
||||
(etaf-scheduler-context-runtime-queue context)))
|
||||
(should
|
||||
(eq 'second
|
||||
(car (etaf-scheduler-context-runtime-queue-tail context))))))
|
||||
(should (etaf-scheduler-context-idle-p context))))
|
||||
|
||||
(ert-deftest etaf-runtime-dirty-effect-fifo-and-priority-stay-stable ()
|
||||
"Append dirty effects in O(1) and sort each immutable effect only once."
|
||||
|
||||
10
tests/fixtures/etaf-m0a-condition-consumers.sexp
vendored
10
tests/fixtures/etaf-m0a-condition-consumers.sexp
vendored
@ -8,6 +8,7 @@
|
||||
(:file "etaf-performance.el" :form condition-case :conditions (error) :owner etaf-performance :policy generic-containment)
|
||||
(:file "etaf-reactive.el" :form condition-case :conditions (error quit) :owner etaf-reactive :policy generic-containment)
|
||||
(:file "etaf-reactive.el" :form condition-case :conditions (error quit) :owner etaf-reactive :policy generic-containment)
|
||||
(:file "etaf-reactive.el" :form condition-case :conditions (error quit) :owner etaf-reactive :policy generic-containment)
|
||||
(:file "etaf-render-port.el" :form condition-case :conditions (error quit) :owner etaf-render-port :policy generic-containment)
|
||||
(:file "etaf-render-port.el" :form condition-case :conditions (error quit) :owner etaf-render-port :policy generic-containment)
|
||||
(:file "etaf-render-port.el" :form condition-case :conditions (error quit) :owner etaf-render-port :policy generic-containment)
|
||||
@ -24,4 +25,11 @@
|
||||
(:file "etaf-runtime.el" :form condition-case :conditions (error quit) :owner etaf-runtime :policy generic-containment)
|
||||
(:file "etaf-runtime.el" :form condition-case :conditions (error quit) :owner etaf-runtime :policy generic-containment)
|
||||
(:file "etaf-runtime.el" :form condition-case :conditions (error quit) :owner etaf-runtime :policy generic-containment)
|
||||
(:file "etaf-runtime.el" :form condition-case :conditions (error quit) :owner etaf-runtime :policy generic-containment))
|
||||
(:file "etaf-runtime.el" :form condition-case :conditions (error quit) :owner etaf-runtime :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment)
|
||||
(:file "etaf-scheduler.el" :form condition-case :conditions (error quit) :owner etaf-scheduler :policy generic-containment))
|
||||
|
||||
Loading…
Reference in New Issue
Block a user