Find file
Fetching contributors…
Cannot retrieve contributors at this time
121 lines (105 sloc) 5.65 KB
#lang scheme/base
(require "util.scm")
;; Task Queues
;; This library allows you to create synchronized "task queues". In normal use,
;; you add tasks (which are just thunks) onto the queue. Meanwhile, the queue itself
;; tries to evaluate the thunks, in FIFO order, one at a time. When the thread runs
;; out of tasks, then it suspends itself. But, when a new task is received, the
;; thread is automatically resumed, so, in effect, it's always ready to work.
;; Use (make-threaded-task-queue) to generate a VALUES of a thread that is suspended
;; (which contains an empty task queue), and a procedure of one argument (a task thunk)
;; which is the task adder. You may never need to use the returned thread, but if
;; you want to put a task thread on hiatus, use, e.g.,
;; (sleep-task-thread-for-at-least <task-thread> 30)
;; which suspends processing of tasks for at least 30 seconds. In practice, the time
;; the thread waits should be just every so slightly more than the given seconds value.
;; We allow for "one off" LIFO queue semantics. That is, if you add a task using
;; the task adder function (assume you've named it add!), you can write
;; (add! (lambda () ...some task...) #:add-to-front #t)
;; and this task will be the next one to be chosen for evaluation.
(provide make-threaded-task-queue
;; returns a VALUES of a thread that is suspended and a procedure for adding thunk tasks
;; to the queue; if return-inspector is #t, return a third value which is an inspector
;; object, which can be accessed with the task-inspector-* fns.
(define (make-threaded-task-queue #:return-inspector (return-inspector #f))
(let ((queue (make-task-queue))
(just-created #t))
(letrec ((thread-thunk (lambda ()
(when just-created
(set! just-created #f)
(thread-suspend the-thread))
(let lp ()
(let ((msg (thread-try-receive)))
(if msg
(let ((wait-until (task-queue-msg-wait-until msg))
(now (current-seconds)))
(if (> wait-until now)
;; so we received a message that told us to
;; sleep the thread for that long:
(begin (sleep (- wait-until now))
;; otherwise, the msg is no longer relevant,
;; so try again:
;; no msg? then attack the queue:
(let ((more? (task-pop-and-handle! queue)))
(if more?
(begin (thread-suspend the-thread)
(the-thread (thread thread-thunk)))
(let ((add-task-fn (lambda (task-thunk #:add-to-front (add-to-front #f))
(task-push! queue task-thunk #:add-to-front add-to-front)
(thread-resume the-thread))))
(if return-inspector
(values the-thread add-task-fn (make-a-task-inspector queue))
(values the-thread add-task-fn))))))
(define-struct task-inspector (lock num-tasks-thunk))
(define (make-a-task-inspector queue)
(make-task-inspector (task-queue-dat-lock queue)
(lambda () (length (task-queue-dat-lst queue)))))
(define-struct task-queue-msg (wait-until))
(define (sleep-task-thread-for-at-least a-thread secs-to-wait)
(thread-send a-thread (make-task-queue-msg (+ (current-seconds) secs-to-wait))))
;; IDEA: have tasks that are "infinite" ..i.e., once you execute one, it pops back on
;; to the end of the queue.
(define-struct task-queue-dat (lock lst last-task-ptr) #:mutable)
(define (make-task-queue)
(make-task-queue-dat (make-lock) '() #f))
;; We implement the queue as a mutable list, where the next task to do is the
;; car of the list. We keep a last task pointer so that we can easily add to the end
;; of the queue.
;; executes the next task if one is available; returns #t if there *are* more tasks left
;; (after the one just executed). returns #f if no tasks are available.
(define (task-pop-and-handle! queue)
(task-queue-dat-lock queue)
(let ((lst (task-queue-dat-lst queue)))
(when (tasks-todo? queue)
(let ((next-task (mcar lst)))
(set-task-queue-dat-lst! queue (mcdr lst))
;; next-task is a we invoke it
(tasks-todo? queue))))
(define (task-push! queue task-thunk #:add-to-front (add-to-front #f))
(let ((cell-for-end (mcons task-thunk '())))
(task-queue-dat-lock queue)
(if (not (tasks-todo? queue))
(set-task-queue-dat-lst! queue cell-for-end)
(if add-to-front
(set-task-queue-dat-lst! queue (mcons task-thunk (task-queue-dat-lst queue)))
(set-mcdr! (task-queue-dat-last-task-ptr queue) cell-for-end)))
(unless add-to-front
(set-task-queue-dat-last-task-ptr! queue cell-for-end)))))
;; not synchronized, so make sure you use in a sync block.
(define (tasks-todo? queue)
(not (null? (task-queue-dat-lst queue))))