Message batcher

batcher.rkt

;; A message-batcher batches messages to be sent over communicore, for
;; efficiency.
;;
;; When the first message is queued in an empty batch, a timer is
;; started. When that timer expires, all queued messages will be sent
;; as a single communicore message (containing an array with the
;; messages). If messages are queued before the timer expires, they
;; will be added to the batch.

(struct message-batcher (semaphore ccore tm path delay [messages #:mutable]))


;; Makes a message-batcher that sends messages via _ccore_ on _path_
;; with at most the specified _delay_ in seconds.

(define (make-message-batcher ccore path delay)
  (message-batcher (make-semaphore 1) ccore (start-timer-manager) path delay '()))

(define (queue-message-for-batching mb msg)
  (call-with-semaphore/enable-break
   (message-batcher-semaphore mb)
   (lambda ()
     (set-message-batcher-messages! mb (cons msg (message-batcher-messages mb)))
     (start-timer (message-batcher-tm mb)
                  (message-batcher-delay mb)
                  (lambda ()
                    (send-batched-messages mb))))))

(define (send-batched-messages mb)
  (call-with-semaphore/enable-break
   (message-batcher-semaphore mb)
   (lambda ()
     (let ([messages (reverse (message-batcher-messages mb))])
       (when (not (empty? messages))
         (set-message-batcher-messages! mb '())
         (%ccore-send (message-batcher-ccore mb)
                      (message-batcher-path mb)
                      messages))))))