Integrate dataflow
This commit is contained in:
parent
e5b87f5bb0
commit
a06b9d188a
|
@ -36,6 +36,9 @@
|
||||||
turn-spawn!
|
turn-spawn!
|
||||||
turn-stop-actor!
|
turn-stop-actor!
|
||||||
turn-crash!
|
turn-crash!
|
||||||
|
turn-field!
|
||||||
|
turn-dataflow!
|
||||||
|
turn-assert/dataflow!
|
||||||
turn-assert!
|
turn-assert!
|
||||||
turn-assert!*
|
turn-assert!*
|
||||||
turn-retract!
|
turn-retract!
|
||||||
|
@ -45,12 +48,15 @@
|
||||||
turn-sync!*
|
turn-sync!*
|
||||||
turn-message!)
|
turn-message!)
|
||||||
|
|
||||||
|
(require (only-in preserves preserve=?))
|
||||||
(require racket/match)
|
(require racket/match)
|
||||||
(require (only-in racket/exn exn->string))
|
(require (only-in racket/exn exn->string))
|
||||||
(require struct-defaults)
|
(require struct-defaults)
|
||||||
|
|
||||||
(require "rewrite.rkt")
|
(require "rewrite.rkt")
|
||||||
(require "engine.rkt")
|
(require "engine.rkt")
|
||||||
|
(require "dataflow.rkt")
|
||||||
|
(require "field.rkt")
|
||||||
(require "support/counter.rkt")
|
(require "support/counter.rkt")
|
||||||
|
|
||||||
(struct entity (assert retract message sync))
|
(struct entity (assert retract message sync))
|
||||||
|
@ -69,6 +75,7 @@
|
||||||
(struct actor (id
|
(struct actor (id
|
||||||
engine
|
engine
|
||||||
[daemon? #:mutable]
|
[daemon? #:mutable]
|
||||||
|
dataflow
|
||||||
[root #:mutable]
|
[root #:mutable]
|
||||||
[exit-reason #:mutable] ;; #f -> running, #t -> terminated OK, exn -> error
|
[exit-reason #:mutable] ;; #f -> running, #t -> terminated OK, exn -> error
|
||||||
[exit-hooks #:mutable])
|
[exit-hooks #:mutable])
|
||||||
|
@ -119,6 +126,7 @@
|
||||||
(define ac (actor (generate-actor-id)
|
(define ac (actor (generate-actor-id)
|
||||||
engine
|
engine
|
||||||
daemon?
|
daemon?
|
||||||
|
(make-dataflow-graph)
|
||||||
'uninitialized
|
'uninitialized
|
||||||
#f
|
#f
|
||||||
'()))
|
'()))
|
||||||
|
@ -213,13 +221,15 @@
|
||||||
;;---------------------------------------------------------------------------
|
;;---------------------------------------------------------------------------
|
||||||
|
|
||||||
(define (turn! f action [zombie-turn? #f])
|
(define (turn! f action [zombie-turn? #f])
|
||||||
(when (or zombie-turn? (and (not (actor-exit-reason (facet-actor f))) (facet-live? f)))
|
(define ac (facet-actor f))
|
||||||
|
(when (or zombie-turn? (and (not (actor-exit-reason ac)) (facet-live? f)))
|
||||||
(let ((turn (turn (generate-turn-id) f (make-hasheq))))
|
(let ((turn (turn (generate-turn-id) f (make-hasheq))))
|
||||||
(with-handlers ([exn? (lambda (e)
|
(with-handlers ([exn? (lambda (e)
|
||||||
(turn! (actor-root (facet-actor f))
|
(turn! (actor-root ac)
|
||||||
(lambda (turn)
|
(lambda (turn)
|
||||||
(actor-terminate! turn (facet-actor f) e))))])
|
(actor-terminate! turn ac e))))])
|
||||||
(action turn)
|
(action turn)
|
||||||
|
(dataflow-repair-damage! (actor-dataflow ac) (lambda (action) (action turn)))
|
||||||
(for [((ff qq) (in-hash (turn-queues turn)))]
|
(for [((ff qq) (in-hash (turn-queues turn)))]
|
||||||
(queue-task! (actor-engine (facet-actor ff))
|
(queue-task! (actor-engine (facet-actor ff))
|
||||||
(lambda ()
|
(lambda ()
|
||||||
|
@ -273,6 +283,22 @@
|
||||||
(define ac (facet-actor (turn-active-facet turn)))
|
(define ac (facet-actor (turn-active-facet turn)))
|
||||||
(turn-enqueue! turn (actor-root ac) (lambda (turn) (actor-terminate! turn ac exn))))
|
(turn-enqueue! turn (actor-root ac) (lambda (turn) (actor-terminate! turn ac exn))))
|
||||||
|
|
||||||
|
(define (turn-field! turn name initial-value)
|
||||||
|
(field (actor-dataflow (facet-actor (turn-active-facet turn))) name initial-value))
|
||||||
|
|
||||||
|
(define (turn-dataflow! turn action)
|
||||||
|
(parameterize ((current-dataflow-subject-id action))
|
||||||
|
(action turn)))
|
||||||
|
|
||||||
|
(define (turn-assert/dataflow! turn peer assertion-action)
|
||||||
|
(define handle #f)
|
||||||
|
(define assertion (void))
|
||||||
|
(turn-dataflow! turn
|
||||||
|
(lambda (turn)
|
||||||
|
(define new-assertion (assertion-action turn))
|
||||||
|
(when (not (preserve=? assertion new-assertion))
|
||||||
|
(set! handle (turn-replace! turn peer handle new-assertion))))))
|
||||||
|
|
||||||
(define (turn-assert! turn peer assertion)
|
(define (turn-assert! turn peer assertion)
|
||||||
(define handle (generate-handle))
|
(define handle (generate-handle))
|
||||||
(turn-assert!* turn peer assertion handle)
|
(turn-assert!* turn peer assertion handle)
|
||||||
|
|
|
@ -0,0 +1,35 @@
|
||||||
|
#lang racket/base
|
||||||
|
|
||||||
|
(provide field?
|
||||||
|
(rename-out [make-field field])
|
||||||
|
field-name
|
||||||
|
field-id)
|
||||||
|
|
||||||
|
(require "dataflow.rkt")
|
||||||
|
(require "support/counter.rkt")
|
||||||
|
|
||||||
|
(struct field (name ;; Symbol
|
||||||
|
id ;; Nat
|
||||||
|
dataflow ;; Dataflow
|
||||||
|
[value #:mutable] ;; Any
|
||||||
|
)
|
||||||
|
#:methods gen:custom-write
|
||||||
|
[(define (write-proc f port mode)
|
||||||
|
(fprintf f "#<field:~a:~a>" (field-id f) (field-name f)))]
|
||||||
|
#:property prop:procedure
|
||||||
|
(case-lambda
|
||||||
|
[(f)
|
||||||
|
(dataflow-record-observation! (field-dataflow f) f)
|
||||||
|
(field-value f)]
|
||||||
|
[(f new-value)
|
||||||
|
(when (not (equal? (field-value f) new-value))
|
||||||
|
(dataflow-record-damage! (field-dataflow f) f)
|
||||||
|
(set-field-value! f new-value))]))
|
||||||
|
|
||||||
|
(define generate-field-id (make-counter))
|
||||||
|
|
||||||
|
(define (make-field dataflow name initial-value)
|
||||||
|
(field name
|
||||||
|
(generate-field-id)
|
||||||
|
dataflow
|
||||||
|
initial-value))
|
|
@ -9,10 +9,8 @@
|
||||||
(require "schemas/gen/dataspace.rkt")
|
(require "schemas/gen/dataspace.rkt")
|
||||||
|
|
||||||
(define ((box ds LIMIT REPORT_EVERY) turn)
|
(define ((box ds LIMIT REPORT_EVERY) turn)
|
||||||
(define value-handle #f)
|
(define value (turn-field! turn 'box-value 0))
|
||||||
(define (set-value turn value)
|
(turn-assert/dataflow! turn ds (lambda (turn) (BoxState->preserves (BoxState (value)))))
|
||||||
(set! value-handle (turn-replace! turn ds value-handle (BoxState->preserves (BoxState value)))))
|
|
||||||
(set-value turn 0)
|
|
||||||
(define start-time (current-inexact-milliseconds))
|
(define start-time (current-inexact-milliseconds))
|
||||||
(define prev-value 0)
|
(define prev-value 0)
|
||||||
(turn-assert! turn ds
|
(turn-assert! turn ds
|
||||||
|
@ -33,7 +31,7 @@
|
||||||
(/ count delta)))
|
(/ count delta)))
|
||||||
(when (= new-value LIMIT)
|
(when (= new-value LIMIT)
|
||||||
(turn-stop-actor! turn))
|
(turn-stop-actor! turn))
|
||||||
(set-value turn new-value))))))))
|
(value new-value))))))))
|
||||||
|
|
||||||
(define ((client ds) turn)
|
(define ((client ds) turn)
|
||||||
(turn-assert! turn ds
|
(turn-assert! turn ds
|
||||||
|
|
Loading…
Reference in New Issue