syndicate-lang
/
syndicate-2017
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reimplement supervise.rkt to use spawn->process+transition at each reboot.

This commit is contained in:
Tony Garnock-Jones 2016-11-29 14:47:22 +13:00
parent 726b936ed3
commit 8cd60417c4
2 changed files with 92 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
racket/syndicate/examples/actor/example-supervise.rkt
View File

@ -5,7 +5,10 @@
(require/activate syndicate/drivers/timestate) (require/activate syndicate/drivers/timestate)
(supervise (supervise
#:name 'ward-supervisor
(actor #:name 'ward (actor #:name 'ward
(on-start (log-info "Starting ward"))
(on-stop (log-info "Stopping ward"))
(on (message 'crash) (on (message 'crash)
(log-info "Crashing") (log-info "Crashing")
(error 'ward "Eep!")) (error 'ward "Eep!"))

228
racket/syndicate/supervise.rkt
View File

@ -11,162 +11,112 @@
(require "store.rkt") (require "store.rkt")
(require (submod "actor.rkt" implementation-details)) (require (submod "actor.rkt" implementation-details))
(require (for-syntax syntax/parse))
(require/activate "drivers/timestate.rkt") (require/activate "drivers/timestate.rkt")
(define-syntax-rule (supervise expr ...) (define-syntax (supervise stx)
(supervise* (lambda () expr ...))) (syntax-parse stx
[(_ name:actor-name expr ...)
(syntax/loc stx
(supervise* #:name name.N (lambda () expr ...)))]))
(define (supervise* actor-producing-thunk) (define (supervise* #:name [supervisor-name #f] actor-producing-thunk)
(with-store [(current-action-transformer (supervise-spawn (current-action-transformer)))] ;; Awkward: the name applies to any and all potential supervisors
;; produced by actor spawns in actor-producing-thunk.
(with-store [(current-action-transformer
(supervise-spawn supervisor-name (current-action-transformer)))]
(actor-producing-thunk))) (actor-producing-thunk)))
(define ((supervise-spawn previous-action-transformer) ac) (define ((supervise-spawn supervisor-name previous-action-transformer) ac)
(match (previous-action-transformer ac) (match (previous-action-transformer ac)
[(? spawn? s) (supervise** s)] [(? spawn? s) (supervise** supervisor-name s)]
[other other])) [other other]))
(define (supervise** s) (define (supervise** supervisor-name supervisee-spawn-action)
(make-spawn (lambda () (actor-action #:name supervisor-name
;; TODO: Consider closing supervisor-main over s (react
;; rather than over st0 and acs. That way, effects
;; from spawn->process+transition will reappear at
;; each reboot.
;;
;; %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
;; %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
;; BUG: Exception thrown inside
;; spawn->process+transition will kill the supervisor!
;; %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
;; %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
;;
(define-values (proc initial-transition) (spawn->process+transition s))
(match initial-transition
[(transition st0 acs)
(list actor-behavior
(boot-actor (lambda ()
(react
(supervisor-main proc st0 acs))))
(list 'supervised (process-name proc)))]
[other
(list (process-behavior proc)
other
(list 'supervised (process-name proc)))]))))
(define (supervisor-main proc boot-state boot-actions) (field [done? #f])
(field [done? #f]) (stop-when (rising-edge (done?)))
(stop-when (rising-edge (done?)))
(field [should-run? #f] (field [supervisee-name 'unknown])
[ready? #f])
(define inner-handle-event (process-behavior proc)) (define intensity 1)
(define period 5000) ;; milliseconds
(define sleep-time 10) ;; seconds
(field [restarts '()])
(define intensity 1) (define (add-restart!)
(define period 5000) ;; milliseconds (define now (current-inexact-milliseconds))
(define sleep-time 10) ;; seconds (define oldest-to-keep (- now period))
(field [restarts '()]) (restarts (filter (lambda (r) (>= r oldest-to-keep))
(cons (current-inexact-milliseconds) (restarts))))
(when (> (length (restarts)) intensity)
(log-error "Supervised process ~s/~s ~a reached max restart intensity. Sleeping for ~a seconds"
supervisor-name
(supervisee-name)
(current-actor-path)
sleep-time)
(sleep sleep-time)))
(define (add-restart!) (field [should-run? #f]
(define now (current-inexact-milliseconds)) [ok? #f])
(define oldest-to-keep (- now period))
(restarts (filter (lambda (r) (>= r oldest-to-keep))
(cons (current-inexact-milliseconds) (restarts))))
(when (> (length (restarts)) intensity)
(log-error "Supervised process ~v ~a reached max restart intensity. Sleeping for ~a seconds"
(process-name proc)
(current-actor-path)
sleep-time)
(sleep sleep-time)))
(on (rising-edge (not (ready?))) (on (rising-edge (not (ok?)))
(should-run? #f) (should-run? #f)
(retract! ?) (ok? #t)
(flush!) (retract! ?)
(should-run? #t)) (flush!)
(should-run? #t))
(on (rising-edge (should-run?)) (define (catch-exns thunk k)
(react (stop-when (rising-edge (not (should-run?)))) (with-handlers ([(lambda (e) #t)
(field [inner-state boot-state]) (lambda (e)
(on-start (perform-actions! boot-actions) (log-error "Supervised process ~s/~s ~a died with exception:\n~a"
(ready? #t)) supervisor-name
(on-event (supervisee-name)
[e (with-handlers ([(lambda (exn) #t) (current-actor-path)
(lambda (e) (if (exn? e)
(log-error "Supervised process ~v ~a died with exception:\n~a" (exn->string e)
(process-name proc) (format "~v" e)))
(current-actor-path) (add-restart!)
(if (exn? e) (ok? #f))])
(exn->string e) (call-with-values thunk k)))
(format "~v" e)))
(add-restart!)
(ready? #f))])
(match (inner-handle-event e (inner-state))
[#f (void)] ;; N.B. TODO: Polling (event of #f)
;; will never reach the inner actor, since
;; actor-behavior doesn't bother executing anything
;; if it is given #f.
[(<quit> _ acs) (perform-actions! acs) (done? #t)]
;; N.B. TODO: what to do with the exception carried
;; in the quit struct?
[(transition st acs) (perform-actions! acs) (inner-state st)]))]))))
;; BROKENish APPROACH: See uni.org (search for 30048bda-b5c2-11e6-afe4-73eb3e5180cc) (on (rising-edge (should-run?))
(react (stop-when (rising-edge (not (should-run?))))
(field [proc #f])
;; (define (supervisor-main proc boot-state boot-actions) (define (handle-transition! txn)
;; (field [done? #f]) (match txn
;; (stop-when (rising-edge (done?))) [#f
;; N.B. TODO: Polling (event of #f) will never
;; reach the inner actor, since actor-behavior
;; doesn't bother executing anything if it is
;; given #f.
(void)]
[(<quit> _ acs)
(perform-actions! acs)
;; N.B. TODO: what to do with the exception
;; carried in the quit struct?
(done? #t)]
[(transition st acs)
(perform-actions! acs)
(proc (update-process-state (proc) st))]))
;; (log-info "ORIGINAL: ~v ~v" (process-name proc) boot-state) (on-start
(catch-exns
(lambda ()
(define-values (initial-proc initial-transition)
(spawn->process+transition supervisee-spawn-action))
(proc initial-proc)
(supervisee-name (process-name initial-proc))
initial-transition)
handle-transition!))
;; (define inner-handle-event (process-behavior proc)) (on-event
;; (field [inner-state boot-state] [running? #t]) [e (when (proc)
;; (on-start (perform-actions! boot-actions)) (catch-exns
;; (define (reboot!) (lambda () ((process-behavior (proc)) e (process-state (proc))))
;; (log-info "--------------------------------------------------") handle-transition!))]))))))
;; (running? #f)
;; (retract! ?)
;; ;; (flush!)
;; (inner-state boot-state)
;; (log-info "--------------------------------------------------")
;; (running? #t)
;; (log-info "REBOOT: ~v ~v" (process-name proc) boot-state)
;; (perform-actions! boot-actions))
;; (define intensity 1)
;; (define period 5000) ;; milliseconds
;; (define sleep-time 10) ;; seconds
;; (field [restarts '()])
;; (define (add-restart!)
;; (define now (current-inexact-milliseconds))
;; (define oldest-to-keep (- now period))
;; (restarts (filter (lambda (r) (>= r oldest-to-keep))
;; (cons (current-inexact-milliseconds) (restarts))))
;; (when (> (length (restarts)) intensity)
;; (log-error "Supervised process ~v ~a reached max restart intensity. Sleeping for ~a seconds"
;; (process-name proc)
;; (current-actor-path)
;; sleep-time)
;; (sleep sleep-time)))
;; (on-event [e (when (running?)
;; (with-handlers ([(lambda (exn) #t)
;; (lambda (e)
;; (log-error "Supervised process ~v ~a died with exception:\n~a"
;; (process-name proc)
;; (current-actor-path)
;; (if (exn? e)
;; (exn->string e)
;; (format "~v" e)))
;; (add-restart!)
;; (reboot!))])
;; (match (inner-handle-event e (inner-state))
;; [#f (void)] ;; N.B. TODO: Polling (event of #f)
;; ;; will never reach the inner actor, since
;; ;; actor-behavior doesn't bother executing anything
;; ;; if it is given #f.
;; [(<quit> _ acs) (perform-actions! acs) (done? #t)]
;; ;; N.B. TODO: what to do with the exception carried
;; ;; in the quit struct?
;; [(transition st acs) (perform-actions! acs) (inner-state st)])))]))