Factored out mux.rkt
This commit is contained in:
parent
bf316f792b
commit
ec2eea9e25
|
@ -6,7 +6,6 @@
|
||||||
(rename-out [quit <quit>])
|
(rename-out [quit <quit>])
|
||||||
(except-out (struct-out spawn) spawn)
|
(except-out (struct-out spawn) spawn)
|
||||||
(rename-out [spawn <spawn>])
|
(rename-out [spawn <spawn>])
|
||||||
(struct-out process)
|
|
||||||
(struct-out transition)
|
(struct-out transition)
|
||||||
(struct-out world)
|
(struct-out world)
|
||||||
|
|
||||||
|
@ -52,7 +51,9 @@
|
||||||
sequence-transitions
|
sequence-transitions
|
||||||
|
|
||||||
world-handle-event
|
world-handle-event
|
||||||
clean-transition)
|
clean-transition
|
||||||
|
|
||||||
|
pretty-print-world)
|
||||||
|
|
||||||
(require racket/set)
|
(require racket/set)
|
||||||
(require racket/match)
|
(require racket/match)
|
||||||
|
@ -61,6 +62,7 @@
|
||||||
(require "route.rkt")
|
(require "route.rkt")
|
||||||
(require "patch.rkt")
|
(require "patch.rkt")
|
||||||
(require "trace.rkt")
|
(require "trace.rkt")
|
||||||
|
(require "mux.rkt")
|
||||||
(module+ test (require rackunit))
|
(module+ test (require rackunit))
|
||||||
|
|
||||||
;; Events = Patches ∪ Messages
|
;; Events = Patches ∪ Messages
|
||||||
|
@ -69,14 +71,6 @@
|
||||||
;; Actions ⊃ Events
|
;; Actions ⊃ Events
|
||||||
(struct spawn (boot) #:prefab)
|
(struct spawn (boot) #:prefab)
|
||||||
|
|
||||||
;; Processes (machine states): (process Matcher (U Behavior (disabled Behavior)) Any)
|
|
||||||
(struct process (interests behavior state) #:transparent)
|
|
||||||
|
|
||||||
;; Disabled Behaviors, when found in a Process, indicate that the
|
|
||||||
;; process has been disabled and is waiting out the performance of its
|
|
||||||
;; final actions before finally being removed.
|
|
||||||
(struct disabled (behavior) #:transparent)
|
|
||||||
|
|
||||||
;; A Behavior is a ((Option Event) Any -> Transition): a function
|
;; A Behavior is a ((Option Event) Any -> Transition): a function
|
||||||
;; mapping an Event (or, in the #f case, a poll signal) and a
|
;; mapping an Event (or, in the #f case, a poll signal) and a
|
||||||
;; Process's current state to a Transition.
|
;; Process's current state to a Transition.
|
||||||
|
@ -98,12 +92,11 @@
|
||||||
;; A Label is a PID or 'meta.
|
;; A Label is a PID or 'meta.
|
||||||
|
|
||||||
;; VM private states
|
;; VM private states
|
||||||
(struct world (next-pid ;; PID
|
(struct world (mux ;; Multiplexer
|
||||||
pending-action-queue ;; (Queueof (Cons Label (U Action 'quit)))
|
pending-action-queue ;; (Queueof (Cons Label (U Action 'quit)))
|
||||||
runnable-pids ;; (Setof PID)
|
runnable-pids ;; (Setof PID)
|
||||||
routing-table ;; (Matcherof (Setof Label))
|
behaviors ;; (HashTable PID Behavior)
|
||||||
process-table ;; (HashTable PID Process)
|
states ;; (HashTable PID Any)
|
||||||
environment-interests ;; (Matcherof (set 'meta))
|
|
||||||
) #:transparent)
|
) #:transparent)
|
||||||
|
|
||||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
|
@ -111,8 +104,6 @@
|
||||||
(define (event? x) (or (patch? x) (message? x)))
|
(define (event? x) (or (patch? x) (message? x)))
|
||||||
(define (action? x) (or (event? x) (spawn? x)))
|
(define (action? x) (or (event? x) (spawn? x)))
|
||||||
|
|
||||||
(define (meta-label? x) (eq? x 'meta))
|
|
||||||
|
|
||||||
(define (prepend-at-meta pattern level)
|
(define (prepend-at-meta pattern level)
|
||||||
(if (zero? level)
|
(if (zero? level)
|
||||||
pattern
|
pattern
|
||||||
|
@ -156,27 +147,29 @@
|
||||||
(filter (lambda (x) (and (action? x) (not (patch-empty? x)))) (flatten actions)))
|
(filter (lambda (x) (and (action? x) (not (patch-empty? x)))) (flatten actions)))
|
||||||
|
|
||||||
(define (send-event e pid w)
|
(define (send-event e pid w)
|
||||||
(match (hash-ref (world-process-table w) pid #f)
|
(define behavior (hash-ref (world-behaviors w) pid #f))
|
||||||
[#f w]
|
(define old-state (hash-ref (world-states w) pid #f))
|
||||||
[(process _ (? disabled?) _) w] ;; disabled due to earlier error or quit
|
(if (not behavior)
|
||||||
[(and p (process _ behavior old-state))
|
w
|
||||||
(invoke-process pid
|
(invoke-process pid
|
||||||
(lambda () (clean-transition (ensure-transition (behavior e old-state))))
|
(lambda () (clean-transition (ensure-transition (behavior e old-state))))
|
||||||
(match-lambda
|
(match-lambda
|
||||||
[#f w]
|
[#f w]
|
||||||
[(and q (quit final-actions))
|
[(and q (quit final-actions))
|
||||||
(trace-process-step e pid p #f q)
|
(trace-process-step e pid behavior old-state #f q)
|
||||||
(enqueue-actions (disable-process pid #f w) pid (append final-actions
|
(enqueue-actions (disable-process pid #f w) pid (append final-actions
|
||||||
(list 'quit)))]
|
(list 'quit)))]
|
||||||
[(and t (transition new-state new-actions))
|
[(and t (transition new-state new-actions))
|
||||||
(trace-process-step e pid p #f t)
|
(trace-process-step e pid behavior old-state #f t)
|
||||||
(update-process pid
|
(enqueue-actions (mark-pid-runnable (update-state w pid new-state) pid)
|
||||||
(struct-copy process p [state new-state])
|
pid
|
||||||
new-actions
|
new-actions)])
|
||||||
w)])
|
|
||||||
(lambda (exn)
|
(lambda (exn)
|
||||||
(trace-process-step e pid p exn #f)
|
(trace-process-step e pid behavior old-state exn #f)
|
||||||
(enqueue-actions (disable-process pid exn w) pid (list 'quit))))]))
|
(enqueue-actions (disable-process pid exn w) pid (list 'quit))))))
|
||||||
|
|
||||||
|
(define (update-state w pid s)
|
||||||
|
(struct-copy world w [states (hash-set (world-states w) pid s)]))
|
||||||
|
|
||||||
(define (send-event/guard delta pid w)
|
(define (send-event/guard delta pid w)
|
||||||
(if (patch-empty? delta)
|
(if (patch-empty? delta)
|
||||||
|
@ -188,16 +181,9 @@
|
||||||
(log-error "Process ~a died with exception:\n~a"
|
(log-error "Process ~a died with exception:\n~a"
|
||||||
(cons pid (trace-pid-stack))
|
(cons pid (trace-pid-stack))
|
||||||
(exn->string exn)))
|
(exn->string exn)))
|
||||||
(match (hash-ref (world-process-table w) pid #f)
|
(struct-copy world w
|
||||||
[#f w]
|
[behaviors (hash-remove (world-behaviors w) pid)]
|
||||||
[old-p
|
[states (hash-remove (world-states w) pid)]))
|
||||||
(define new-p (struct-copy process old-p [behavior (disabled (process-behavior old-p))]))
|
|
||||||
(struct-copy world w [process-table (hash-set (world-process-table w) pid new-p)])]))
|
|
||||||
|
|
||||||
(define (update-process pid p actions w)
|
|
||||||
(let* ((w (struct-copy world w [process-table (hash-set (world-process-table w) pid p)]))
|
|
||||||
(w (mark-pid-runnable w pid)))
|
|
||||||
(enqueue-actions w pid actions)))
|
|
||||||
|
|
||||||
(define (invoke-process pid thunk k-ok k-exn)
|
(define (invoke-process pid thunk k-ok k-exn)
|
||||||
(define-values (ok? result)
|
(define-values (ok? result)
|
||||||
|
@ -223,9 +209,8 @@
|
||||||
(quit actions))
|
(quit actions))
|
||||||
|
|
||||||
(define-syntax-rule (spawn-process behavior-exp initial-state-exp initial-patch-exp ...)
|
(define-syntax-rule (spawn-process behavior-exp initial-state-exp initial-patch-exp ...)
|
||||||
(spawn (lambda (pid)
|
(spawn (lambda ()
|
||||||
(process (apply-patch (matcher-empty)
|
(list (patch-seq initial-patch-exp ...)
|
||||||
(label-patch (patch-seq initial-patch-exp ...) (set pid)))
|
|
||||||
behavior-exp
|
behavior-exp
|
||||||
initial-state-exp))))
|
initial-state-exp))))
|
||||||
|
|
||||||
|
@ -244,16 +229,15 @@
|
||||||
(make-spawn-world (lambda () (list boot-action ...))))
|
(make-spawn-world (lambda () (list boot-action ...))))
|
||||||
|
|
||||||
(define (make-world boot-actions)
|
(define (make-world boot-actions)
|
||||||
(world 0
|
(world (mux)
|
||||||
(list->queue (for/list ((a (in-list (clean-actions boot-actions)))) (cons 'meta a)))
|
(list->queue (for/list ((a (in-list (clean-actions boot-actions)))) (cons 'meta a)))
|
||||||
(set)
|
(set)
|
||||||
(matcher-empty)
|
|
||||||
(hash)
|
(hash)
|
||||||
(matcher-empty)))
|
(hash)))
|
||||||
|
|
||||||
(define (make-spawn-world boot-actions-thunk)
|
(define (make-spawn-world boot-actions-thunk)
|
||||||
(spawn (lambda (pid)
|
(spawn (lambda ()
|
||||||
(process (matcher-empty)
|
(list empty-patch
|
||||||
world-handle-event
|
world-handle-event
|
||||||
(make-world (boot-actions-thunk))))))
|
(make-world (boot-actions-thunk))))))
|
||||||
|
|
||||||
|
@ -300,112 +284,56 @@
|
||||||
(define ((perform-action label a) w)
|
(define ((perform-action label a) w)
|
||||||
(match a
|
(match a
|
||||||
[(spawn boot)
|
[(spawn boot)
|
||||||
(define new-pid (world-next-pid w))
|
|
||||||
(invoke-process 'booting
|
(invoke-process 'booting
|
||||||
(lambda ()
|
(lambda ()
|
||||||
(match (boot new-pid)
|
(match (boot)
|
||||||
[(? process? p) p]
|
[(and results (list (? patch?) (? procedure?) _))
|
||||||
[other (error 'spawn
|
results]
|
||||||
"Spawn boot procedure must yield process; received ~v"
|
[other
|
||||||
|
(error 'spawn
|
||||||
|
"Spawn boot procedure must yield boot spec; received ~v"
|
||||||
other)]))
|
other)]))
|
||||||
(lambda (new-p)
|
(lambda (results)
|
||||||
(define new-interests (process-interests new-p))
|
(match-define (list initial-patch behavior initial-state) results)
|
||||||
(define new-w
|
(define-values (new-mux new-pid patches meta-action)
|
||||||
(update-process new-pid
|
(mux-add-stream (world-mux w) initial-patch))
|
||||||
new-p
|
(let* ((w (update-state w new-pid initial-state))
|
||||||
'()
|
(w (mark-pid-runnable w new-pid))
|
||||||
(struct-copy world w [next-pid (+ new-pid 1)])))
|
(w (struct-copy world w
|
||||||
(apply-patch-in-world new-pid (patch new-interests (matcher-empty)) new-w))
|
[mux new-mux]
|
||||||
|
[behaviors (hash-set (world-behaviors w)
|
||||||
|
new-pid
|
||||||
|
behavior)]))
|
||||||
|
(w (deliver-patches w patches meta-action)))
|
||||||
|
w))
|
||||||
(lambda (exn)
|
(lambda (exn)
|
||||||
(log-error "Spawned process in world ~a died with exception:\n~a"
|
(log-error "Spawned process in world ~a died with exception:\n~a"
|
||||||
(trace-pid-stack)
|
(trace-pid-stack)
|
||||||
(exn->string exn))
|
(exn->string exn))
|
||||||
(transition w '())))]
|
(transition w '())))]
|
||||||
['quit
|
['quit
|
||||||
(define pt (world-process-table w))
|
(define-values (new-mux _label patches meta-action) (mux-remove-stream (world-mux w) label))
|
||||||
(match (hash-ref pt label #f)
|
(deliver-patches (struct-copy world w [mux new-mux])
|
||||||
[#f (transition w '())]
|
;; ^ behavior & state already removed by disable-process
|
||||||
[(process interests _ _)
|
patches
|
||||||
(define delta (patch (matcher-empty) interests))
|
meta-action)]
|
||||||
(define-values (discarded-actions retained-actions)
|
|
||||||
(queue-partition (lambda (e) (equal? (car e) label)) (world-pending-action-queue w)))
|
|
||||||
(when (not (queue-empty? discarded-actions))
|
|
||||||
(log-error "Process ~a had ~a queued actions at exit"
|
|
||||||
label
|
|
||||||
(queue-length discarded-actions)))
|
|
||||||
(define new-w (struct-copy world w
|
|
||||||
[process-table (hash-remove pt label)]
|
|
||||||
[pending-action-queue retained-actions]))
|
|
||||||
(apply-patch-in-world label delta new-w)])]
|
|
||||||
[(? patch? delta-orig)
|
[(? patch? delta-orig)
|
||||||
(define p (hash-ref (world-process-table w) label #f))
|
(define-values (new-mux _label patches meta-action)
|
||||||
(cond
|
(mux-update-stream (world-mux w) label delta-orig))
|
||||||
[(or p (meta-label? label))
|
(deliver-patches (struct-copy world w [mux new-mux])
|
||||||
(define old-interests (if (meta-label? label)
|
patches
|
||||||
(world-environment-interests w)
|
meta-action)]
|
||||||
(process-interests p)))
|
|
||||||
(define delta (limit-patch (label-patch delta-orig (set label)) old-interests))
|
|
||||||
(define new-interests (apply-patch old-interests delta))
|
|
||||||
(define new-w
|
|
||||||
(if (meta-label? label)
|
|
||||||
(struct-copy world w [environment-interests new-interests])
|
|
||||||
(let* ((p (struct-copy process p [interests new-interests])))
|
|
||||||
(struct-copy world w [process-table (hash-set (world-process-table w) label p)]))))
|
|
||||||
(apply-patch-in-world label delta new-w)]
|
|
||||||
[else ;; we can still apply actions for nonexistent processes,
|
|
||||||
;; but we have to limit the patches by consulting the
|
|
||||||
;; whole routing table, making their zombie patch actions
|
|
||||||
;; potentially less efficient.
|
|
||||||
(define delta (limit-patch/routing-table (label-patch delta-orig (set label))
|
|
||||||
(world-routing-table w)))
|
|
||||||
(apply-patch-in-world label delta w)])]
|
|
||||||
[(and m (message body))
|
[(and m (message body))
|
||||||
(when (observe? body)
|
(define-values (send-to-meta? affected-pids) (mux-route-message (world-mux w) label body))
|
||||||
(log-warning "Process ~a sent message containing query ~v"
|
(transition (for/fold [(w w)] [(pid (in-list affected-pids))]
|
||||||
(cons label (trace-pid-stack))
|
|
||||||
body))
|
|
||||||
(cond
|
|
||||||
[(matcher-match-value (world-routing-table w) body #f) ;; some other process has declared m
|
|
||||||
(transition w '())]
|
|
||||||
[else
|
|
||||||
(define local-to-meta? (and (not (meta-label? label)) ;; it's from a local process, not envt
|
|
||||||
(at-meta? body))) ;; it relates to envt, not local
|
|
||||||
(define affected-pids (if local-to-meta?
|
|
||||||
(set)
|
|
||||||
(matcher-match-value (world-routing-table w) (observe body))))
|
|
||||||
(transition (for/fold [(w w)] [(pid (in-set affected-pids))]
|
|
||||||
(send-event m pid w))
|
(send-event m pid w))
|
||||||
(and local-to-meta?
|
(and send-to-meta? (message (at-meta-claim body))))]))
|
||||||
(message (at-meta-claim body))))])]))
|
|
||||||
|
|
||||||
;; PRECONDITION: delta has been limited to be minimal with respect to
|
(define (deliver-patches w patches meta-action)
|
||||||
;; existing interests of its label in w's routing table.
|
(transition (for/fold [(w w)] [(entry (in-list patches))]
|
||||||
(define (apply-patch-in-world label delta w)
|
(match-define (cons label event) entry)
|
||||||
(define old-routing-table (world-routing-table w))
|
(send-event/guard event label w))
|
||||||
(define new-routing-table (apply-patch old-routing-table delta))
|
meta-action))
|
||||||
(define delta-aggregate (compute-aggregate-patch delta label old-routing-table))
|
|
||||||
(define affected-pids (let ((pids (compute-affected-pids old-routing-table delta)))
|
|
||||||
(if (meta-label? label) pids (set-add pids label))))
|
|
||||||
(transition (for/fold [(w (struct-copy world w [routing-table new-routing-table]))]
|
|
||||||
[(pid affected-pids)]
|
|
||||||
(cond [(equal? pid label)
|
|
||||||
(define feedback
|
|
||||||
(patch (biased-intersection new-routing-table (patch-added delta))
|
|
||||||
(biased-intersection old-routing-table (patch-removed delta))))
|
|
||||||
(send-event/guard feedback label w)]
|
|
||||||
[else
|
|
||||||
(define p (hash-ref (world-process-table w) pid))
|
|
||||||
(define event (view-patch delta-aggregate (process-interests p)))
|
|
||||||
(send-event/guard event pid w)]))
|
|
||||||
(and (not (meta-label? label))
|
|
||||||
(drop-patch delta-aggregate))))
|
|
||||||
|
|
||||||
(define (compute-affected-pids routing-table delta)
|
|
||||||
(define cover (matcher-union (patch-added delta) (patch-removed delta)))
|
|
||||||
(matcher-match-matcher (pattern->matcher #t (observe (embedded-matcher cover)))
|
|
||||||
routing-table
|
|
||||||
#:seed (set)
|
|
||||||
#:combiner (lambda (v1 v2 acc) (set-union v2 acc))))
|
|
||||||
|
|
||||||
(define (step-children w)
|
(define (step-children w)
|
||||||
(define runnable-pids (world-runnable-pids w))
|
(define runnable-pids (world-runnable-pids w))
|
||||||
|
@ -416,6 +344,33 @@
|
||||||
(send-event #f pid w))
|
(send-event #f pid w))
|
||||||
'())))
|
'())))
|
||||||
|
|
||||||
|
(define (pretty-print-world w [p (current-output-port)])
|
||||||
|
(local-require racket/pretty)
|
||||||
|
(match-define (world mux qs runnable behaviors states) w)
|
||||||
|
(fprintf p "WORLD:\n")
|
||||||
|
(fprintf p " - ~a queued actions\n" (queue-length qs))
|
||||||
|
(fprintf p " - ~a runnable pids ~a\n" (set-count runnable) (set->list runnable))
|
||||||
|
(fprintf p " - ~a live processes (~a with claims)\n"
|
||||||
|
(hash-count states)
|
||||||
|
(hash-count (mux-interest-table mux)))
|
||||||
|
(fprintf p " - next pid: ~a\n" (mux-next-pid mux))
|
||||||
|
(fprintf p " - routing table:\n")
|
||||||
|
(pretty-print-matcher (mux-routing-table mux) p)
|
||||||
|
(for ([pid (set-union (hash-keys (mux-interest-table mux)) (hash-keys states))])
|
||||||
|
(fprintf p " ---- process ~a, behavior ~v, STATE:\n" pid (hash-ref behaviors pid #f))
|
||||||
|
(define state (hash-ref states pid #f))
|
||||||
|
(display (indented-port-output 6 (lambda (p)
|
||||||
|
(if (world? state)
|
||||||
|
(pretty-print-world state p)
|
||||||
|
(pretty-write state p))))
|
||||||
|
p)
|
||||||
|
(newline p)
|
||||||
|
(fprintf p " process ~a, behavior ~v, CLAIMS:\n" pid (hash-ref behaviors pid #f))
|
||||||
|
(display (indented-port-output 6 (lambda (p)
|
||||||
|
(pretty-print-matcher (mux-interests-of mux pid) p)))
|
||||||
|
p)
|
||||||
|
(newline p)))
|
||||||
|
|
||||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
|
|
||||||
(module+ test
|
(module+ test
|
||||||
|
|
|
@ -0,0 +1,94 @@
|
||||||
|
#lang racket/base
|
||||||
|
;; General multiplexer.
|
||||||
|
|
||||||
|
(provide meta-label?
|
||||||
|
(except-out (struct-out mux) mux)
|
||||||
|
(rename-out [mux <mux>] [make-mux mux])
|
||||||
|
mux-add-stream
|
||||||
|
mux-remove-stream
|
||||||
|
mux-update-stream
|
||||||
|
mux-route-message
|
||||||
|
mux-interests-of)
|
||||||
|
|
||||||
|
(require racket/set)
|
||||||
|
(require racket/match)
|
||||||
|
(require "route.rkt")
|
||||||
|
(require "patch.rkt")
|
||||||
|
(require "trace.rkt")
|
||||||
|
|
||||||
|
;; A PID is a Nat.
|
||||||
|
;; A Label is a PID or 'meta.
|
||||||
|
;; Multiplexer private states
|
||||||
|
(struct mux (next-pid ;; PID
|
||||||
|
routing-table ;; (Matcherof (Setof Label))
|
||||||
|
interest-table ;; (HashTable Label Matcher)
|
||||||
|
) #:transparent)
|
||||||
|
|
||||||
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
|
|
||||||
|
(define (meta-label? x) (eq? x 'meta))
|
||||||
|
|
||||||
|
(define (make-mux)
|
||||||
|
(mux 0 (matcher-empty) (hash)))
|
||||||
|
|
||||||
|
(define (mux-add-stream m initial-patch)
|
||||||
|
(define new-pid (mux-next-pid m))
|
||||||
|
(mux-update-stream (struct-copy mux m [next-pid (+ new-pid 1)])
|
||||||
|
new-pid
|
||||||
|
initial-patch))
|
||||||
|
|
||||||
|
(define (mux-remove-stream m label)
|
||||||
|
(mux-update-stream m label (patch (matcher-empty) (pattern->matcher #t ?))))
|
||||||
|
|
||||||
|
(define (mux-update-stream m label delta-orig)
|
||||||
|
(define old-interests (mux-interests-of m label))
|
||||||
|
(define delta (limit-patch (label-patch delta-orig (set label)) old-interests))
|
||||||
|
(define new-interests (apply-patch old-interests delta))
|
||||||
|
(let* ((m (struct-copy mux m
|
||||||
|
[interest-table
|
||||||
|
(if (matcher-empty? new-interests)
|
||||||
|
(hash-remove (mux-interest-table m) label)
|
||||||
|
(hash-set (mux-interest-table m) label new-interests))])))
|
||||||
|
;; CONDITION at this point: delta has been labelled and limited to
|
||||||
|
;; be minimal with respect to existing interests of its label.
|
||||||
|
(define old-routing-table (mux-routing-table m))
|
||||||
|
(define new-routing-table (apply-patch old-routing-table delta))
|
||||||
|
(define delta-aggregate (compute-aggregate-patch delta label old-routing-table))
|
||||||
|
(define affected-pids (let ((pids (compute-affected-pids old-routing-table delta)))
|
||||||
|
(set-remove (set-add pids label) 'meta))) ;; TODO: removing meta is weird
|
||||||
|
(values (struct-copy mux m [routing-table new-routing-table])
|
||||||
|
label
|
||||||
|
(for/list [(pid affected-pids)]
|
||||||
|
(cond [(equal? pid label)
|
||||||
|
(define feedback
|
||||||
|
(patch (biased-intersection new-routing-table (patch-added delta))
|
||||||
|
(biased-intersection old-routing-table (patch-removed delta))))
|
||||||
|
(cons label feedback)]
|
||||||
|
[else
|
||||||
|
(cons pid (view-patch delta-aggregate (mux-interests-of m pid)))]))
|
||||||
|
(and (not (meta-label? label))
|
||||||
|
(drop-patch delta-aggregate)))))
|
||||||
|
|
||||||
|
(define (compute-affected-pids routing-table delta)
|
||||||
|
(define cover (matcher-union (patch-added delta) (patch-removed delta)))
|
||||||
|
(matcher-match-matcher (pattern->matcher #t (observe (embedded-matcher cover)))
|
||||||
|
routing-table
|
||||||
|
#:seed (set)
|
||||||
|
#:combiner (lambda (v1 v2 acc) (set-union v2 acc))))
|
||||||
|
|
||||||
|
(define (mux-route-message m label body)
|
||||||
|
(when (observe? body)
|
||||||
|
(log-warning "Stream ~a sent message containing query ~v"
|
||||||
|
(cons label (trace-pid-stack))
|
||||||
|
body))
|
||||||
|
(cond
|
||||||
|
[(matcher-match-value (mux-routing-table m) body #f) ;; some other stream has declared body
|
||||||
|
(values #f '())]
|
||||||
|
[(and (not (meta-label? label)) ;; it's from a local process, not envt
|
||||||
|
(at-meta? body)) ;; it relates to envt, not local
|
||||||
|
(values #t '())]
|
||||||
|
[else
|
||||||
|
(values #f (set->list (matcher-match-value (mux-routing-table m) (observe body))))]))
|
||||||
|
|
||||||
|
(define (mux-interests-of m label)
|
||||||
|
(hash-ref (mux-interest-table m) label (matcher-empty)))
|
|
@ -8,9 +8,12 @@
|
||||||
trace-process-step
|
trace-process-step
|
||||||
trace-internal-step
|
trace-internal-step
|
||||||
|
|
||||||
exn->string) ;; required from web-server/private/util
|
exn->string ;; required from web-server/private/util
|
||||||
|
string-indent
|
||||||
|
indented-port-output)
|
||||||
|
|
||||||
(require (only-in web-server/private/util exn->string))
|
(require (only-in web-server/private/util exn->string))
|
||||||
|
(require (only-in racket/string string-join string-split))
|
||||||
|
|
||||||
(define trace-logger (make-logger 'minimart-trace))
|
(define trace-logger (make-logger 'minimart-trace))
|
||||||
|
|
||||||
|
@ -30,13 +33,22 @@
|
||||||
(log-message trace-logger 'info name "" r #f)))
|
(log-message trace-logger 'info name "" r #f)))
|
||||||
|
|
||||||
;; Event PID Process (Option Exception) (Option Transition) -> Void
|
;; Event PID Process (Option Exception) (Option Transition) -> Void
|
||||||
(define (trace-process-step e pid p exn t)
|
(define (trace-process-step e pid beh st exn t)
|
||||||
(when exn
|
(when exn
|
||||||
(log-error "Process ~a died with exception:\n~a"
|
(log-error "Process ~a died with exception:\n~a"
|
||||||
(cons pid (trace-pid-stack))
|
(cons pid (trace-pid-stack))
|
||||||
(exn->string exn)))
|
(exn->string exn)))
|
||||||
(record-trace-event 'process-step (list (cons pid (trace-pid-stack)) e p exn t)))
|
(record-trace-event 'process-step (list (cons pid (trace-pid-stack)) e beh st exn t)))
|
||||||
|
|
||||||
;; PID Action World Transition -> Void
|
;; PID Action World Transition -> Void
|
||||||
(define (trace-internal-step pid a w t)
|
(define (trace-internal-step pid a w t)
|
||||||
(record-trace-event 'internal-step (list (cons pid (trace-pid-stack)) a w t)))
|
(record-trace-event 'internal-step (list (cons pid (trace-pid-stack)) a w t)))
|
||||||
|
|
||||||
|
(define (string-indent amount s)
|
||||||
|
(define pad (make-string amount #\space))
|
||||||
|
(string-join (for/list [(line (string-split s "\n"))] (string-append pad line)) "\n"))
|
||||||
|
|
||||||
|
(define (indented-port-output amount f)
|
||||||
|
(define p (open-output-string))
|
||||||
|
(f p)
|
||||||
|
(string-indent amount (get-output-string p)))
|
||||||
|
|
|
@ -9,6 +9,7 @@
|
||||||
(require (only-in web-server/private/util exn->string))
|
(require (only-in web-server/private/util exn->string))
|
||||||
(require "../core.rkt")
|
(require "../core.rkt")
|
||||||
(require "../trace.rkt")
|
(require "../trace.rkt")
|
||||||
|
(require "../mux.rkt")
|
||||||
|
|
||||||
(define (env-aref varname default alist)
|
(define (env-aref varname default alist)
|
||||||
(define key (or (getenv varname) default))
|
(define key (or (getenv varname) default))
|
||||||
|
@ -95,7 +96,7 @@
|
||||||
(let loop ()
|
(let loop ()
|
||||||
(match-define (vector level message-string data event-name) (sync receiver))
|
(match-define (vector level message-string data event-name) (sync receiver))
|
||||||
(match* (event-name data)
|
(match* (event-name data)
|
||||||
[('process-step (list pids e p exn t))
|
[('process-step (list pids e beh st exn t))
|
||||||
(define pidstr (format-pids pids))
|
(define pidstr (format-pids pids))
|
||||||
(define relevant-exn? (and show-exceptions? exn))
|
(define relevant-exn? (and show-exceptions? exn))
|
||||||
(match e
|
(match e
|
||||||
|
@ -113,22 +114,23 @@
|
||||||
(output "~a received a message:\n" pidstr)
|
(output "~a received a message:\n" pidstr)
|
||||||
(pretty-write body (current-error-port))))])
|
(pretty-write body (current-error-port))))])
|
||||||
(when (or relevant-exn? show-process-states-pre?)
|
(when (or relevant-exn? show-process-states-pre?)
|
||||||
(when (or relevant-exn? (not (boring-state? (process-state p))))
|
(when (or relevant-exn? (not (boring-state? st)))
|
||||||
(with-color YELLOW
|
(with-color YELLOW
|
||||||
(output "~a's state just before the event:\n" pidstr)
|
(output "~a's state just before the event:\n" pidstr)
|
||||||
(output-state (process-state p)))))
|
(output-state st))))
|
||||||
(when relevant-exn?
|
(when relevant-exn?
|
||||||
(with-color WHITE-ON-RED
|
(with-color WHITE-ON-RED
|
||||||
(output "Process ~a died with exception:\n~a\n"
|
(output "Process ~a ~v died with exception:\n~a\n"
|
||||||
pidstr
|
pidstr
|
||||||
|
beh
|
||||||
(exn->string exn))))
|
(exn->string exn))))
|
||||||
(when (quit? t)
|
(when (quit? t)
|
||||||
(with-color BRIGHT-RED
|
(with-color BRIGHT-RED
|
||||||
(output "Process ~a exited normally.\n" pidstr)))
|
(output "Process ~a ~v exited normally.\n" pidstr beh)))
|
||||||
(when (or relevant-exn? show-process-states-post?)
|
(when (or relevant-exn? show-process-states-post?)
|
||||||
(when (transition? t)
|
(when (transition? t)
|
||||||
(unless (boring-state? (transition-state t))
|
(unless (boring-state? (transition-state t))
|
||||||
(when (not (equal? (process-state p) (transition-state t)))
|
(when (not (equal? st (transition-state t)))
|
||||||
(with-color YELLOW
|
(with-color YELLOW
|
||||||
(output "~a's state just after the event:\n" pidstr)
|
(output "~a's state just after the event:\n" pidstr)
|
||||||
(output-state (transition-state t)))))))]
|
(output-state (transition-state t)))))))]
|
||||||
|
@ -136,16 +138,16 @@
|
||||||
(when t ;; inert worlds don't change interestingly
|
(when t ;; inert worlds don't change interestingly
|
||||||
(define pidstr (format-pids pids))
|
(define pidstr (format-pids pids))
|
||||||
(define new-w (if (transition? t) (transition-state t) old-w))
|
(define new-w (if (transition? t) (transition-state t) old-w))
|
||||||
(define old-processes (world-process-table old-w))
|
(define newcount (hash-count (world-behaviors new-w)))
|
||||||
(define new-processes (world-process-table new-w))
|
|
||||||
(define newcount (hash-count new-processes))
|
|
||||||
(match a
|
(match a
|
||||||
[(? spawn?)
|
[(? spawn?)
|
||||||
(when (or show-process-lifecycle? show-actions?)
|
(when (or show-process-lifecycle? show-actions?)
|
||||||
(define newpid (set-first (set-subtract (hash-keys new-processes)
|
(define newpid (set-first (set-subtract (hash-keys (world-behaviors new-w))
|
||||||
(hash-keys old-processes))))
|
(hash-keys (world-behaviors old-w)))))
|
||||||
(define newpidstr (format-pids (cons newpid (cdr pids)))) ;; replace parent pid
|
(define newpidstr (format-pids (cons newpid (cdr pids)))) ;; replace parent pid
|
||||||
(match-define (process interests behavior state) (hash-ref new-processes newpid))
|
(define interests (mux-interests-of (world-mux new-w) newpid))
|
||||||
|
(define behavior (hash-ref (world-behaviors new-w) newpid))
|
||||||
|
(define state (hash-ref (world-states new-w) newpid))
|
||||||
(with-color BRIGHT-GREEN
|
(with-color BRIGHT-GREEN
|
||||||
(output "~a ~v spawned from ~a (~a total processes now)\n"
|
(output "~a ~v spawned from ~a (~a total processes now)\n"
|
||||||
newpidstr
|
newpidstr
|
||||||
|
@ -160,20 +162,14 @@
|
||||||
(pretty-print-matcher interests (current-error-port))))]
|
(pretty-print-matcher interests (current-error-port))))]
|
||||||
['quit
|
['quit
|
||||||
(when (or show-process-lifecycle? show-actions?)
|
(when (or show-process-lifecycle? show-actions?)
|
||||||
(match (hash-ref old-processes (car pids) (lambda () #f))
|
(define interests (mux-interests-of (world-mux old-w) (car pids)))
|
||||||
[#f (void)]
|
|
||||||
[(process interests behavior state)
|
|
||||||
(with-color BRIGHT-RED
|
(with-color BRIGHT-RED
|
||||||
(output "~a ~v exited (~a total processes now)\n"
|
(output "~a exited (~a total processes now)\n"
|
||||||
pidstr
|
pidstr
|
||||||
behavior
|
|
||||||
newcount))
|
newcount))
|
||||||
(unless (boring-state? state)
|
|
||||||
(output "~a's final state:\n" pidstr)
|
|
||||||
(output-state state))
|
|
||||||
(unless (matcher-empty? interests)
|
(unless (matcher-empty? interests)
|
||||||
(output "~a's final interests:\n" pidstr)
|
(output "~a's final interests:\n" pidstr)
|
||||||
(pretty-print-matcher interests (current-error-port)))]))]
|
(pretty-print-matcher interests (current-error-port))))]
|
||||||
[(? patch? p)
|
[(? patch? p)
|
||||||
(when (or show-actions? show-patch-actions?)
|
(when (or show-actions? show-patch-actions?)
|
||||||
(output "~a performed a patch:\n" pidstr)
|
(output "~a performed a patch:\n" pidstr)
|
||||||
|
@ -183,11 +179,12 @@
|
||||||
(output "~a sent a message:\n" pidstr)
|
(output "~a sent a message:\n" pidstr)
|
||||||
(pretty-write body (current-error-port)))])
|
(pretty-write body (current-error-port)))])
|
||||||
(when show-routing-table?
|
(when show-routing-table?
|
||||||
(when (not (equal? (world-routing-table old-w) (world-routing-table new-w)))
|
(define old-table (mux-routing-table (world-mux old-w)))
|
||||||
|
(define new-table (mux-routing-table (world-mux new-w)))
|
||||||
|
(when (not (equal? old-table new-table))
|
||||||
(with-color BRIGHT-BLUE
|
(with-color BRIGHT-BLUE
|
||||||
(output "~a's routing table:\n" (format-pids (cdr pids)))
|
(output "~a's routing table:\n" (format-pids (cdr pids)))
|
||||||
(pretty-print-matcher (world-routing-table new-w)
|
(pretty-print-matcher new-table (current-error-port))))))])
|
||||||
(current-error-port))))))])
|
|
||||||
(loop))))
|
(loop))))
|
||||||
|
|
||||||
(void (when (not (set-empty? flags))
|
(void (when (not (set-empty? flags))
|
||||||
|
|
Loading…
Reference in New Issue