101 lines
4.2 KiB
Racket
101 lines
4.2 KiB
Racket
#lang racket/base
|
|
;;; SPDX-License-Identifier: LGPL-3.0-or-later
|
|
;;; SPDX-FileCopyrightText: Copyright © 2021-2024 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
|
|
|
|
(provide engine?
|
|
engine-id
|
|
engine-name
|
|
engine-running?
|
|
engine-custodian
|
|
engine-thread
|
|
engine-inhabitant-count
|
|
make-engine
|
|
adjust-inhabitant-count!
|
|
queue-task!
|
|
engine-register!
|
|
engine-deregister!
|
|
engine-shutdown!
|
|
*dead-engine*)
|
|
|
|
(require racket/match)
|
|
(require (only-in racket/exn exn->string))
|
|
(require "support/counter.rkt")
|
|
|
|
(define-logger syndicate/engine)
|
|
|
|
(struct engine (id name custodian thread [running? #:mutable] actors [inhabitant-count #:mutable])
|
|
#:methods gen:custom-write
|
|
[(define (write-proc e port mode)
|
|
(fprintf port "#<engine:~a:~a>" (engine-id e) (engine-name e)))])
|
|
|
|
(define generate-engine-id (make-counter))
|
|
|
|
(define (make-engine initial-inhabitant-count name termination-handler)
|
|
(define custodian (make-custodian))
|
|
(define e (engine (generate-engine-id)
|
|
name
|
|
custodian
|
|
(parameterize ((current-custodian custodian))
|
|
(thread (lambda ()
|
|
(thread-receive) ;; delay boot until we're ready
|
|
(log-syndicate/engine-debug "~a starting" e)
|
|
(with-handlers ([exn? (handle-unexpected-task-runner-termination e)])
|
|
(let loop ((termination-handler termination-handler))
|
|
(log-syndicate/engine-debug
|
|
"~a task count: ~a" e (engine-inhabitant-count e))
|
|
(if (positive? (engine-inhabitant-count e))
|
|
;; We have some non-daemon users so just block
|
|
(begin ((thread-receive))
|
|
(loop termination-handler))
|
|
;; No non-daemon users, so keep running until there's no more work
|
|
(match (thread-try-receive)
|
|
[#f ;; No work, no non-daemons, we're done.
|
|
(log-syndicate/engine-debug
|
|
"Invoking termination handler ~v" termination-handler)
|
|
(termination-handler loop)]
|
|
[thunk
|
|
(log-syndicate/engine-debug
|
|
"Zero inhabitant count but pending work")
|
|
(thunk)
|
|
(loop termination-handler)])))
|
|
(log-syndicate/engine-debug "~a stopping" e)
|
|
(custodian-shutdown-all custodian)))))
|
|
#t
|
|
(make-hash)
|
|
initial-inhabitant-count))
|
|
(thread-send (engine-thread e) 'boot)
|
|
e)
|
|
|
|
(define (adjust-inhabitant-count! e delta)
|
|
(queue-task! e (lambda ()
|
|
(set-engine-inhabitant-count! e (+ (engine-inhabitant-count e) delta)))))
|
|
|
|
(define (engine-register! e ac)
|
|
(when (not (eq? (current-thread) (engine-thread e)))
|
|
(error 'engine-register! "Called from wrong thread"))
|
|
(hash-set! (engine-actors e) ac #t))
|
|
|
|
(define (engine-deregister! e ac)
|
|
(when (not (eq? (current-thread) (engine-thread e)))
|
|
(error 'engine-deregister! "Called from wrong thread"))
|
|
(hash-remove! (engine-actors e) ac))
|
|
|
|
(define (engine-shutdown! e)
|
|
(log-syndicate/engine-debug "~a shutdown" e)
|
|
(set-engine-running?! e #f)
|
|
(define actors (hash-keys (engine-actors e)))
|
|
(hash-clear! (engine-actors e))
|
|
actors)
|
|
|
|
(define ((handle-unexpected-task-runner-termination e) exn)
|
|
(log-syndicate/engine-error "~a terminated unexpectedly!\n~a" e (exn->string exn))
|
|
(exit 1))
|
|
|
|
(define (queue-task! e thunk)
|
|
(thread-send (engine-thread e)
|
|
thunk
|
|
(lambda ()
|
|
(log-syndicate/engine-debug "Attempt to enqueue task for dead engine ~v" e))))
|
|
|
|
(define *dead-engine* (make-engine 0 'dead-engine void))
|