8sync/inbox.scm

   1 ;;; 8sync --- Asynchronous programming for Guile
   2 ;;; Copyright © 2017 Christopher Allan Webber <cwebber@dustycloud.org>
   3 ;;;
   4 ;;; This file is part of 8sync.
   5 ;;;
   6 ;;; 8sync is free software: you can redistribute it and/or modify it
   7 ;;; under the terms of the GNU Lesser General Public License as
   8 ;;; published by the Free Software Foundation, either version 3 of the
   9 ;;; License, or (at your option) any later version.
  10 ;;;
  11 ;;; 8sync is distributed in the hope that it will be useful,
  12 ;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ;;; GNU Lesser General Public License for more details.
  15 ;;;
  16 ;;; You should have received a copy of the GNU Lesser General Public
  17 ;;; License along with 8sync.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 (define-module (8sync inbox)
  20   #:use-module (fibers)
  21   #:use-module (fibers channels)
  22   #:use-module (fibers conditions)
  23   #:use-module (fibers operations)
  24   #:use-module (ice-9 match)
  25   #:use-module (ice-9 q)
  26   #:use-module (srfi srfi-9)
  27   #:use-module (ice-9 atomic)
  28   #:export (delivery-agent))
  29
  30 ;; TODO: Add backpressure limit
  31 (define (delivery-agent inbox-enq inbox-deq stop?)
  32   "This starts up a loop doing delivery receiving from INBOX-ENQ and
  33 delivering to INBOX-DEQ, actually managing an internal object QUEUE.
  34 Condidtion variable STOP? can be set to indicate that this agent
  35 should stop."
  36   (define queue
  37     (make-q))
  38   (define get-or-stop
  39     (choice-operation
  40      (wrap-operation (get-operation inbox-enq)
  41                      (lambda (message)
  42                        (enq! queue message)
  43                        'got-one))
  44      (wrap-operation (wait-operation stop?)
  45                      (const 'stop))))
  46   (let main-lp ()
  47     (cond
  48      ;; No items to deliver?  We need to get one first...
  49      ((q-empty? queue)
  50       (match (perform-operation get-or-stop)
  51         ;; keep looping
  52         ('got-one (main-lp))
  53         ;; halt!
  54         ('stop 'done)))
  55      (else
  56       ;; Pull an item off the queue for delivery...
  57       (let ((this-one (deq! queue)))
  58         ;; But we need to start looping!
  59         (let deliver-this-one ()
  60           (match (perform-operation
  61                   (choice-operation
  62                    ;; get a new message and keep trying to deliver
  63                    ;; this one, or stop
  64                    get-or-stop
  65                    ;; deliver this one and get the next one to deliver
  66                    (wrap-operation (put-operation inbox-deq this-one)
  67                                    (const 'delivered))))
  68             ;; We're dispatching based on which one succeeds.
  69             ;; Maybe this isn't necessary, but I'm not convinced
  70             ;; that looping within the choice-operation would be
  71             ;; properly tail recursive.
  72             ('got-one (deliver-this-one))
  73             ('delivered (main-lp))
  74             ('stop 'done))))))))