1 ;; Simple Actor Machine
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.
12 (chicken process-context)
28 (define sam-host "localhost")
29 (define sam-port 8000)
31 (define sam-version "0.1")
35 (define (log-msg . args)
36 (with-output-to-port (current-error-port)
38 (apply print (cons "## " args)))))
40 (define (log-trace . args)
41 (with-output-to-port (current-error-port)
43 (if trace (apply log-msg args)))))
45 (define (->stringrep arg)
46 (with-output-to-string
51 ;; (See also macros defined in sam-macros.scm.)
53 (define (beh-proc beh)
55 (define (beh-parent beh)
61 (send-message recipient 'pong))))
70 (define (make-address host port id)
73 (define (make-local-address . args)
74 (make-address sam-host
80 (define (address-id address)
82 (define (address-host address)
84 (define (address-port address)
86 (define (address->string address)
88 (make-uri #:scheme "actor"
89 #:host (address-host address)
90 #:port (address-port address)
91 #:path (list '/ (address-id address)))))
92 (define (string->address str)
93 (let ((uri (uri-reference str)))
94 (make-address (uri-host uri)
96 (cadr (uri-path uri)))))
98 (define (address-local? address)
99 (and (equal? (address-host address) sam-host)
100 (equal? (address-port address) sam-port)))
102 (define actor-table (make-hash-table))
104 (define (make-actor beh)
105 (let* ((address (make-local-address))
106 (id (address-id address)))
107 (hash-table-set! actor-table id beh)
110 (define (deliver-message address . message)
111 (let ((id (address-id address)))
112 (log-trace "DELIVERING to " id ": " (->stringrep message))
113 (let loop ((beh (hash-table-ref/default actor-table id #f)))
116 (match (apply (beh-proc beh) (cons address message))
117 ('done (hash-table-delete! actor-table id))
119 (log-trace "Passing to parent behaviour...")
120 (loop (beh-parent beh)))
121 ((? beh? new-beh) (hash-table-set! actor-table id new-beh))
123 'do-nothing)) ; sleep is now the default
125 (log-msg "Warning: actor " id " crashed evaluating message " (->stringrep message))
126 (print-error-message o)))
127 (log-msg "Warning: DISCARDING message to unknown actor " id ": " (->stringrep message))))))
131 (define local-queue-mutex (make-mutex "message queue"))
132 (define message-available-mutex (make-mutex "message available"))
133 (mutex-lock! message-available-mutex #f #f)
134 (define local-queue (make-fifo))
136 (define (send-message address . message)
137 (log-trace "SENDING to " address ": " (->stringrep message))
138 (apply (if (address-local? address)
140 send-network-message)
141 (cons address message)))
143 (define (send-local-message address . message)
144 (mutex-lock! local-queue-mutex)
145 (fifo-push local-queue (cons address message))
146 (mutex-unlock! message-available-mutex)
147 (mutex-unlock! local-queue-mutex))
149 (define (send-network-message address . message)
150 (let ((s (udp-open-socket))
151 (packet (with-output-to-string
153 (write (cons address message))))))
156 (address-host address)
157 (address-port address))
159 (udp-close-socket s)))
161 (define (send-message-later address time . message)
165 (apply send-message (cons address message)))))
167 (define (next-local-message)
169 (mutex-lock! message-available-mutex #f #f)
170 (mutex-lock! local-queue-mutex)
171 (set! res (fifo-pop local-queue))
172 (if (not (fifo-empty? local-queue))
173 (mutex-unlock! message-available-mutex))
174 (mutex-unlock! local-queue-mutex)
177 (define (start-scheduler)
179 (apply deliver-message (next-local-message))
185 (define (start-network-listener)
188 (let ((s (udp-open-socket*)))
189 (udp-bind! s #f sam-port)
191 (let-values (((n str) (udp-recv s 65536)))
192 (match (with-input-from-string str read)
193 ((address message ...)
194 (apply send-message (cons address message)))
196 (log-msg "Warning: received badly formatted message string '" str "'"))))
201 (define reader-queue-mutex (make-mutex "reader queue"))
202 (define reader-available-mutex (make-mutex "reader available"))
203 (mutex-lock! reader-available-mutex #f #f)
204 (define reader-queue (make-fifo))
206 (define (next-reader)
208 (mutex-lock! reader-available-mutex #f #f)
209 (mutex-lock! reader-queue-mutex)
210 (set! res (fifo-pop reader-queue))
211 (if (not (fifo-empty? reader-queue))
212 (mutex-unlock! reader-available-mutex))
213 (mutex-unlock! reader-queue-mutex)
216 (define (start-console)
220 (let ((reader (next-reader)))
221 (##sys#thread-block-for-i/o! (current-thread) 0 #t)
223 (send-message reader (read-line)))
226 ;; System initialization
228 (define-beh system-beh
232 (log-msg "System actor received shutdown message.")
236 (('print strings ...) =>
237 (apply print strings))
240 (mutex-lock! reader-queue-mutex)
241 (fifo-push reader-queue reader)
242 (mutex-unlock! reader-available-mutex)
243 (mutex-unlock! reader-queue-mutex)))
247 (start-network-listener)
248 (let ((system (make-actor system-beh))
252 (set! main (make-actor main-beh)))
254 (log-msg "Error starting main actor. Is main-beh defined?")
256 (send-message main system))
259 (define (print-usage)
260 (print "Simple Actor Machine v" sam-version "\n"
262 "Usage: sam -h|--help\n"
263 " sam [-n hostname] [-p port] source-file-1 [source-file-2 [...]] "))
265 (let loop ((args (cdr (argv))))
267 (((or "-h" "--help"))
269 (((or "-p" "--port") pstr rest ...)
270 (set! sam-port (string->number pstr))
272 (((or "-n" "--hostname") hstr rest ...)
275 (((or "-t" "--trace") rest ...)
276 (log-msg "Enabling trace debugging")
279 (((? file-exists? filename) rest ...)
280 (log-msg "Loading " filename)
284 (log-msg "Booting SAM\n")
287 (print "Unrecognised argument '" (car args) "'.\n")