syndicate-2017/racket/syndicate/examples/actor/queue-no-credit2.rkt

104 lines
3.9 KiB
Racket

#lang syndicate
;; A Queue with no flow control.
(require racket/set)
(require syndicate/functional-queue)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Queue Protocol
;; Assertion. Scopes flow from source to target.
(struct subscription (source target) #:prefab)
;; Message. Delivery from source to target.
(struct delivery (source target body) #:prefab)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Metrics Protocol
;; Assertion. Describes some attribute of monitoringish interest.
(struct metric (key value) #:prefab)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Utilities
;; (Fieldof (Queueof X)) -> X
;; EFFECT: Changes f.
;; EFFECT: Error if f contains the empty queue.
(define (deq! f)
(define-values (item remainder) (dequeue (f)))
(f remainder)
item)
;; (Fieldof (Queueof X)) X -> Void
;; EFFECT: Changes f.
(define (enq! f v)
(f (enqueue (f) v)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Queue Implementation
;; EFFECT: Spawn a queue process named `queue-id`.
(define (spawn-queue queue-id)
(spawn #:name (list 'queue queue-id)
(field [waiters (make-queue)])
(field [messages (make-queue)])
(on (asserted (subscription queue-id $who)) (enq! waiters who))
(on (retracted (subscription queue-id $who)) (waiters (queue-remove who (waiters))))
(on (message (delivery $who queue-id $body)) (enq! messages body))
(begin/dataflow
(when (and (not (queue-empty? (waiters)))
(not (queue-empty? (messages))))
(define who (deq! waiters))
(define msg (deq! messages))
(log-info "~a: sending ~a message ~a" queue-id who msg)
(send! (delivery queue-id who msg))
(enq! waiters who)))
(assert (metric (list 'subscriber-count queue-id) (queue-length (waiters))))
(assert (metric (list 'backlog queue-id) (queue-length (messages))))
;;------------------------------------------------------------
(local-require (submod syndicate/actor priorities))
(begin/dataflow #:priority *idle-priority* ;; Check invariants
(define has-waiters? (not (queue-empty? (waiters))))
(define has-messages? (not (queue-empty? (messages))))
(unless (and (or (not has-waiters?) (not has-messages?))
(or (not has-messages?) (not has-waiters?)))
(error 'queue
"~a: invariant violated: ~v"
queue-id
`((has-waiters? ,has-waiters?)
(has-messages? ,has-messages?)
(waiters ,(queue->list (waiters)))
(messages ,(queue->list (messages)))))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Example
(define (spawn-consumer consumer-id #:variant [variant 'normal])
(spawn #:name (list 'consumer consumer-id)
(assert (subscription 'q consumer-id))
(on (message (delivery 'q consumer-id $body))
(log-info "Consumer ~a got: ~a" consumer-id body)
(when (eq? variant 'crashy)
(error consumer-id
"Hark, canst thou hear me? I will play the swan / and die in music.")))))
(spawn (define/query-hash metrics (metric $k $v) k v)
(begin/dataflow (log-info " ~a" (hash->list (metrics)))))
(spawn-queue 'q)
(spawn-consumer 'c1)
(spawn-consumer 'c2 #:variant 'crashy)
(spawn-consumer 'c3)
(spawn* (until (asserted (observe (delivery _ 'q _))))
(for ((n (in-range 10)))
(send! (delivery #f 'q n))
(when (odd? n) (flush!))
))