sam.scm

   1 ;; Simple Actor Machine
   2 ;;
   3 ;; Houses a population of actors which can communicate using messages
   4 ;; with actors on the same machine or other machines via the network.
   5
   6 (import (chicken io)
   7         (chicken string)
   8         (chicken port)
   9         matchable
  10         srfi-18 ; threads
  11         srfi-69 ; hash-table
  12         uuid ; ids for actors
  13         uri-generic
  14         udp
  15         fifo)
  16
  17 ;; Actors
  18
  19 (define this-host "localhost")
  20 (define this-port 8000)
  21
  22 (define (make-address host port id)
  23   (make-uri #:scheme "actor"
  24             #:host host
  25             #:port port
  26             #:path (list '/ id)))
  27
  28 (define (make-local-address . args)
  29   (make-address this-host
  30                 this-port
  31                 (if (null? args)
  32                     (uuid)
  33                     (car args))))
  34
  35 (define (address-id address) (cadr (uri-path address)))
  36
  37 (define (address-local? address)
  38   (and (equal? (uri-host address) this-host)
  39        (equal? (uri-port address) this-port)))
  40
  41 (define actor-table (make-hash-table))
  42
  43 (define (make-actor beh)
  44   (let* ((address (make-local-address))
  45          (id (address-id address)))
  46     (hash-table-set! actor-table id beh)
  47     address))
  48
  49 (define (deliver-message address . message)
  50   (let ((id (address-id address)))
  51     (let ((behaviour (hash-table-ref/default actor-table id '())))
  52       (if (null? behaviour)
  53           (print "Warning: discarded message" message " to unknown actor " address)
  54           (match (apply (hash-table-ref actor-table id) (cons address message))
  55             ('done (hash-table-delete! actor-table id))
  56             ('sleep 'do-nothing)
  57             (new-beh (hash-table-set! actor-table id new-beh)))))))
  58
  59 ;; Scheduler
  60
  61 (define local-queue-mutex (make-mutex "message queue"))
  62 (define message-available-mutex (make-mutex "message available"))
  63 (mutex-lock! message-available-mutex #f #f)
  64 (define local-queue (make-fifo))
  65
  66 (define (send-message address . message)
  67   (apply (if (address-local? address)
  68              send-local-message
  69              send-network-message)
  70          (cons address message)))
  71
  72 (define (send-local-message address . message)
  73   (mutex-lock! local-queue-mutex)
  74   (fifo-push local-queue (cons address message))
  75   (mutex-unlock! message-available-mutex)
  76   (mutex-unlock! local-queue-mutex))
  77
  78 (define (send-network-message address . message)
  79   (let ((s (udp-open-socket))
  80         (machine (address-machine address)))
  81     (udp-bind! s #f 0)
  82     (udp-connect! s
  83                   (machine-host machine)
  84                   (machine-port machine))
  85     (udp-send s message)
  86     (udp-close-socket s)))
  87
  88 (define (next-local-message)
  89   (let ((res #f))
  90     (mutex-lock! message-available-mutex #f #f)
  91     (mutex-lock! local-queue-mutex)
  92     (set! res (fifo-pop local-queue))
  93     (if (not (fifo-empty? local-queue))
  94       (mutex-unlock! message-available-mutex))
  95     (mutex-unlock! local-queue-mutex)
  96     res))
  97
  98 (define (start-scheduler)
  99   (thread-start!
 100    (lambda ()
 101      (let loop ()
 102        (apply deliver-message (next-local-message))
 103        (loop)))))
 104
 105 ;; Console
 106
 107 (define reader-queue-mutex (make-mutex "reader queue"))
 108 (define reader-available-mutex (make-mutex "reader available"))
 109 (mutex-lock! reader-available-mutex #f #f)
 110 (define reader-queue (make-fifo))
 111
 112 (define console
 113   (make-actor (lambda (self . message)
 114                 (mutex-lock! reader-queue-mutex)
 115                 (fifo-push reader-queue (car message))
 116                 (mutex-unlock! reader-available-mutex)
 117                 (mutex-unlock! reader-queue-mutex)
 118                 'sleep)))
 119
 120 (define (next-reader)
 121   (let ((res #f))
 122     (mutex-lock! reader-available-mutex #f #f)
 123     (mutex-lock! reader-queue-mutex)
 124     (set! res (fifo-pop reader-queue))
 125     (if (not (fifo-empty? reader-queue))
 126         (mutex-unlock! reader-available-mutex))
 127     (mutex-unlock! reader-queue-mutex)
 128     res))
 129
 130 (define (start-console)
 131   (let loop ()
 132     (let ((reader (next-reader)))
 133       (##sys#thread-block-for-i/o! (current-thread) 0 #t)
 134       (thread-yield!)
 135       (send-message reader (read-line)))
 136     (loop)))
 137
 138
 139 ;; Network
 140
 141 (define (start-network-listener)
 142   (thread-start!
 143    (lambda ()
 144      (let ((s (udp-open-socket*)))
 145        (udp-bind! s #f this-port)
 146        (let loop ()
 147          (let-values (((n str) (udp-recv s 1024)))
 148            (print "Received " n " bytes over network: " str)
 149            (match (with-input-from-string str read)
 150              ((uri-str message ...)
 151               (apply send-message (cons (uri-reference uri-str) message)))
 152              (else
 153               (print "Warning: received badly formatted message string '" str "'"))))
 154          (loop))))))
 155
 156 ;; System interface
 157
 158 (define system
 159   (make-actor (lambda (self . message)
 160                 (match message
 161                   (('shutdown)
 162                    (print "## System actor received shutdown message.")
 163                    (exit 0)
 164                    'done)
 165                   (('println strings ...)
 166                    (apply print strings)
 167                    'sleep)))))
 168
 169 ;; Testing
 170
 171
 172 (send-message system 'println "Hello, what is your name?")
 173 (send-message console
 174               (make-actor (lambda (self . message)
 175                             (match message
 176                               ((name)
 177                                (send-message system 'println "Hello, " name "!")
 178                                'done)))))
 179
 180 (thread-start!
 181  (lambda ()
 182    (thread-sleep! 120)
 183    (send-message system 'shutdown)))
 184
 185 (print (uri->string system))
 186
 187 (start-scheduler)
 188 (start-network-listener)
 189 (start-console)
 190
 191 ;; (thread-join! scheduler-thread)