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 (spawn-inbox
  29             delivery-agent))
  30
  31 (define* (spawn-inbox)
  32   "Spawn an inbox fiber which manages a a buffered queue.
  33
  34 Returns three values to its continuation: a INBOX-ENQ channel to send
  35 messages to, an INBOX-DEQ channel which is what the actor doing the
  36 reading should read from, and a STOP? atomic box which can be set to #t
  37 to stop delivery."
  38   (let ((inbox-enq (make-channel))
  39         (inbox-deq (make-channel))
  40         (stop? (make-atomic-box #f)))
  41     (spawn-fiber (lambda ()
  42                    ;; From the perspective of the delivery-agent,
  43                    ;; deliver-to
  44                    (delivery-agent inbox-enq inbox-deq stop?)))
  45     (values inbox-enq inbox-deq stop?)))
  46
  47 ;; @@: Do we want to add a stop condition?
  48 (define (delivery-agent inbox-enq inbox-deq stop?)
  49   "This starts up a loop doing delivery receiving from INBOX-ENQ and
  50 delivering to INBOX-DEQ, actually managing an (ice-9 q) object QUEUE.
  51 Atomic box STOP? can be set to indicate that this "
  52   (define queue
  53     (make-q))
  54   (define get-or-stop
  55     (choice-operation
  56      (wrap-operation (get-operation inbox-enq)
  57                      (lambda (message)
  58                        (enq! queue message)
  59                        'got-one))
  60      (wrap-operation (wait-operation stop?)
  61                      (const 'stop))))
  62   (let main-lp ()
  63     (cond
  64      ;; No items to deliver?  We need to get one first...
  65      ((q-empty? queue)
  66       (match (perform-operation get-or-stop)
  67         ;; keep looping
  68         ('got-one (main-lp))
  69         ;; halt!
  70         ('stop 'done)))
  71      (else
  72       ;; Pull an item off the queue for delivery...
  73       (let ((this-one (deq! queue)))
  74         ;; But we need to start looping!
  75         (let deliver-this-one ()
  76           (match (perform-operation
  77                   (choice-operation
  78                    ;; get a new message and keep trying to deliver
  79                    ;; this one, or stop
  80                    get-or-stop
  81                    ;; deliver this one and get the next one to deliver
  82                    (wrap-operation (put-operation inbox-deq this-one)
  83                                    (const 'delivered))))
  84             ;; We're dispatching based on which one succeeds.
  85             ;; Maybe this isn't necessary, but I'm not convinced
  86             ;; that looping within the choice-operation would be
  87             ;; properly tail recursive.
  88             ('got-one (deliver-this-one))
  89             ('delivered (main-lp))
  90             ('stop 'done))))))))