sam.scm

   1 ;; Simple Actor Machine
   2 ;;
   3 ;; A virtual machine which houses a population of actors which can
   4 ;; communicate using messages with actors on the same host or other
   5 ;; hosts via the network.
   6
   7 (import scheme
   8         (chicken base)
   9         (chicken io)
  10         (chicken string)
  11         (chicken port)
  12         (chicken process-context)
  13         (chicken file)
  14         (chicken condition)
  15         matchable
  16         srfi-18 ; threads
  17         srfi-69 ; hash-table
  18         uuid ; ids for actors
  19         uri-generic
  20         udp
  21         fifo
  22         sam-macros)
  23
  24 ;; Global variables
  25
  26 (define trace #f)
  27
  28 (define sam-host "localhost")
  29 (define sam-port 8000)
  30
  31 (define sam-version "0.1")
  32
  33 ;; Logging
  34
  35 (define (log-msg . args)
  36   (with-output-to-port (current-error-port)
  37     (lambda ()
  38       (apply print (cons "## " args)))))
  39
  40 (define (log-trace . args)
  41   (with-output-to-port (current-error-port)
  42     (lambda ()
  43       (if trace (apply log-msg args)))))
  44
  45 (define (->stringrep arg)
  46   (with-output-to-string
  47     (lambda ()
  48       (write arg))))
  49
  50 ;; Behaviours
  51 ;; (See also macros defined in sam-macros.scm.)
  52
  53 (define (beh-proc beh)
  54   (car beh))
  55 (define (beh-parent beh)
  56   (cdr beh))
  57
  58 (define root-beh
  59   (make-beh : #f (self)
  60             (('ping recipient) =>
  61              (send-message recipient 'pong))))
  62
  63 ;; Actors
  64
  65 (define (make-address host port id)
  66   (list id host port))
  67
  68 (define (make-local-address . args)
  69   (make-address sam-host
  70                 sam-port
  71                 (if (null? args)
  72                     (uuid)
  73                     (car args))))
  74
  75 (define (address-id address)
  76   (car address))
  77 (define (address-host address)
  78   (cadr address))
  79 (define (address-port address)
  80   (caddr address))
  81 (define (address->string address)
  82   (uri->string
  83    (make-uri #:scheme "actor"
  84              #:host (address-host address)
  85              #:port (address-port address)
  86              #:path (list '/ (address-id address)))))
  87 (define (string->address str)
  88   (let ((uri (uri-reference str)))
  89     (make-address (uri-host uri)
  90                   (uri-port uri)
  91                   (cadr (uri-path uri)))))
  92
  93 (define (address-local? address)
  94   (and (equal? (address-host address) sam-host)
  95        (equal? (address-port address) sam-port)))
  96
  97 (define actor-table (make-hash-table))
  98
  99 (define (make-actor beh)
 100   (let* ((address (make-local-address))
 101          (id (address-id address)))
 102     (hash-table-set! actor-table id beh)
 103     address))
 104
 105 (define (deliver-message address . message)
 106   (let ((id (address-id address)))
 107     (log-trace "DELIVERING to " id ": " (->stringrep message))
 108     (let loop ((beh (hash-table-ref/default actor-table id #f)))
 109       (if beh
 110           (condition-case
 111               (match (apply (beh-proc beh) (cons address message))
 112                 ('done (hash-table-delete! actor-table id))
 113                 ('pass
 114                  (log-trace "Passing to parent behaviour...")
 115                  (loop (beh-parent beh)))
 116                 ((? procedure? new-beh) (hash-table-set! actor-table id new-beh))
 117                 (else
 118                  'do-nothing)) ; sleep is now the default
 119             (o (exn)
 120              (log-msg "Warning: actor " id " crashed evaluating message " (->stringrep message))
 121              (print-error-message o)))
 122           (log-msg "Warning: DISCARDING message to unknown actor " id ": " (->stringrep message))))))
 123
 124 ;; Scheduler
 125
 126 (define local-queue-mutex (make-mutex "message queue"))
 127 (define message-available-mutex (make-mutex "message available"))
 128 (mutex-lock! message-available-mutex #f #f)
 129 (define local-queue (make-fifo))
 130
 131 (define (send-message address . message)
 132   (log-trace "SENDING to " address ": " (->stringrep message))
 133   (apply (if (address-local? address)
 134              send-local-message
 135              send-network-message)
 136          (cons address message)))
 137
 138 (define (send-local-message address . message)
 139   (mutex-lock! local-queue-mutex)
 140   (fifo-push local-queue (cons address message))
 141   (mutex-unlock! message-available-mutex)
 142   (mutex-unlock! local-queue-mutex))
 143
 144 (define (send-network-message address . message)
 145   (let ((s (udp-open-socket))
 146         (packet (with-output-to-string
 147                   (lambda ()
 148                     (write (cons address message))))))
 149     (udp-bind! s #f 0)
 150     (udp-connect! s
 151                   (address-host address)
 152                   (address-port address))
 153     (udp-send s packet)
 154     (udp-close-socket s)))
 155
 156 (define (send-message-later address time . message)
 157   (thread-start!
 158    (lambda ()
 159      (thread-sleep! time)
 160      (apply send-message (cons address message)))))
 161
 162 (define (next-local-message)
 163   (let ((res #f))
 164     (mutex-lock! message-available-mutex #f #f)
 165     (mutex-lock! local-queue-mutex)
 166     (set! res (fifo-pop local-queue))
 167     (if (not (fifo-empty? local-queue))
 168         (mutex-unlock! message-available-mutex))
 169     (mutex-unlock! local-queue-mutex)
 170     res))
 171
 172 (define (start-scheduler)
 173   (let loop ()
 174     (apply deliver-message (next-local-message))
 175     (loop)))
 176
 177
 178 ;; Network
 179
 180 (define (start-network-listener)
 181   (thread-start!
 182    (lambda ()
 183      (let ((s (udp-open-socket*)))
 184        (udp-bind! s #f sam-port)
 185        (let loop ()
 186          (let-values (((n str) (udp-recv s 65536)))
 187            (match (with-input-from-string str read)
 188              ((address message ...)
 189               (apply send-message (cons address message)))
 190              (else
 191               (log-msg "Warning: received badly formatted message string '" str "'"))))
 192          (loop))))))
 193
 194 ;; System interface
 195
 196 (define reader-queue-mutex (make-mutex "reader queue"))
 197 (define reader-available-mutex (make-mutex "reader available"))
 198 (mutex-lock! reader-available-mutex #f #f)
 199 (define reader-queue (make-fifo))
 200
 201 (define (next-reader)
 202   (let ((res #f))
 203     (mutex-lock! reader-available-mutex #f #f)
 204     (mutex-lock! reader-queue-mutex)
 205     (set! res (fifo-pop reader-queue))
 206     (if (not (fifo-empty? reader-queue))
 207         (mutex-unlock! reader-available-mutex))
 208     (mutex-unlock! reader-queue-mutex)
 209     res))
 210
 211 (define (start-console)
 212   (thread-start!
 213    (lambda ()
 214      (let loop ()
 215        (let ((reader (next-reader)))
 216          (##sys#thread-block-for-i/o! (current-thread) 0 #t)
 217          (thread-yield!)
 218          (send-message reader (read-line)))
 219        (loop)))))
 220
 221 ;; System initialization
 222
 223 (define-beh system-beh
 224   (self)
 225   (('shutdown) =>
 226    (log-msg "System actor received shutdown message.")
 227    (exit 0)
 228    'done)
 229
 230   (('print strings ...) =>
 231    (apply print strings))
 232
 233   (('read reader) =>
 234    (mutex-lock! reader-queue-mutex)
 235    (fifo-push reader-queue reader)
 236    (mutex-unlock! reader-available-mutex)
 237    (mutex-unlock! reader-queue-mutex)))
 238
 239 (define (boot-sam)
 240   (start-console)
 241   (start-network-listener)
 242   (let ((system (make-actor system-beh))
 243         (main #f))
 244     (condition-case
 245         (begin
 246           (set! main (make-actor main-beh)))
 247       ((exn)
 248        (log-msg "Error starting main actor. Is main-beh defined?")
 249        (exit 1)))
 250     (send-message main system))
 251   (start-scheduler))
 252
 253 (define (print-usage)
 254   (print "Simple Actor Machine v" sam-version "\n"
 255          "\n"
 256          "Usage: sam -h|--help\n"
 257          "       sam [-n hostname] [-p port] source-file-1 [source-file-2 [...]] "))
 258
 259 (let loop ((args (cdr (argv))))
 260   (match args
 261     (((or "-h" "--help"))
 262      (print-usage))
 263     (((or "-p" "--port") pstr rest ...)
 264      (set! sam-port (string->number pstr))
 265      (loop rest))
 266     (((or "-n" "--hostname") hstr rest ...)
 267      (set! sam-host hstr)
 268      (loop rest))
 269     (((or "-t" "--trace") rest ...)
 270      (log-msg "Enabling trace debugging")
 271      (set! trace #t)
 272      (loop rest))
 273     (((? file-exists? filename) rest ...)
 274      (log-msg "Loading " filename)
 275      (load filename)
 276      (loop rest))
 277     (()
 278      (log-msg "Booting SAM\n")
 279      (boot-sam))
 280     (else
 281      (print "Unrecognised argument '" (car args) "'.\n")
 282      (print-usage))))
 283