syndicate-rkt/syndicate/engine.rkt

60 lines
2.5 KiB
Racket

#lang racket/base
;;; SPDX-License-Identifier: LGPL-3.0-or-later
;;; SPDX-FileCopyrightText: Copyright © 2021 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
(provide engine?
engine-thread
engine-inhabitant-count
make-engine
adjust-inhabitant-count!
queue-task!)
(require racket/match)
(require (only-in racket/exn exn->string))
(require "support/counter.rkt")
(define-logger syndicate/task)
(struct engine (id thread [inhabitant-count #:mutable])
#:methods gen:custom-write
[(define (write-proc e port mode)
(fprintf port "#<engine:~a>" (engine-id e)))])
(define generate-engine-id (make-counter))
(define (make-engine initial-inhabitant-count)
(define e (engine (generate-engine-id)
(thread (lambda ()
(thread-receive) ;; delay boot until we're ready
(log-syndicate/task-info "~a starting" e)
(with-handlers ([exn? (handle-unexpected-task-runner-termination e)])
(let loop ()
(log-syndicate/task-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))
;; 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.
(void)]
[thunk
(thunk)
(loop)])))
(log-syndicate/task-info "~a stopping" e))))
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 ((handle-unexpected-task-runner-termination e) exn)
(log-syndicate/task-error "~a terminated unexpectedly!\n~a" e (exn->string exn))
(exit 1))
(define (queue-task! e thunk)
(thread-send (engine-thread e) thunk))