Simple thread supervisor

supervisor.rkt

#lang racket/base

(require racket/exn
         racket/set
         "util.rkt")

(provide supervise)

(define-logger supervise)

(define restart-window 60)
(define max-restarts 4)

(struct supervised (name thunk* restarts thread))

(define ((supervised-uncaught-exception s) e)
  (log-supervise-error "uncaught exception in thread ~a" (supervised-name s))
  (log-supervise-error (exn->string e)))

(define (supervised-thunk s)
  (lambda ()
    (with-handlers ([exn? (supervised-uncaught-exception s)])
      ((supervised-thunk* s)))))

(define (supervise-shutdown tds)
  (for ([s (in-set tds)])
    (log-supervise-info "sending shutdown to ~a" (supervised-name s))
    (break-thread (supervised-thread s)))
  (for ([s (in-set tds)])
    (thread-wait (supervised-thread s))))

(define (supervised-restart s tds)
  (log-supervise-info "restarting thread ~a" (supervised-name s))
  (define restart-threshold (- (current-seconds) restart-window))
  (define pruned-restarts 
    (for/list ([r (in-list (supervised-restarts s))] #:when (> r restart-threshold)) r))
  (define num-restarts (length pruned-restarts))
  (log-supervise-info "~a restarts in the last ~a seconds" num-restarts restart-window)
  (when (> num-restarts max-restarts)
    (error 'supervised-restart "too many restarts for ~a: ~a" (supervised-name s) num-restarts))
  (define new-s
    (struct-copy supervised s 
      [restarts (cons (current-seconds) pruned-restarts)]
      [thread   (thread (supervised-thunk s))]))
  (set-replace tds s new-s))

(define (supervised-restart1-evt s tds)
  (handle-evt 
    (thread-dead-evt (supervised-thread s))
    (λ (ignored)
      (log-supervise-info "thread ~a is dead" (supervised-name s))
      (supervised-restart s tds))))

(define (supervised-restart-evt tds kont)
  (handle-evt
    (apply 
      choice-evt
      (for/list ([s (in-set tds)])
        (cond 
          [(supervised-thread s) (supervised-restart1-evt s tds)]
          [else
            (handle-evt always-evt (λ (ignore) (supervised-restart s tds)))])))
    kont))

(define (supervise* tds)
  (with-handlers ([exn? (λ (e) (log-supervise-error "supervisor got exception")
                               (log-supervise-error (exn->string e))
                               (supervise-shutdown tds))])
    (sync (supervised-restart-evt tds supervise*))))

(define supervise
  (make-keyword-procedure
    (lambda (kws kas)
      (define tds
        (for/set ([kw (in-list kws)]
                  [ka (in-list kas)])
          (supervised (keyword->string kw) ka null #f)))
      (supervise* tds))))