SAM now runs the show.
[sam.git] / 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 machine or other
5 ;; machines 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         matchable
15         srfi-18 ; threads
16         srfi-69 ; hash-table
17         uuid ; ids for actors
18         uri-generic
19         udp
20         fifo)
21
22 ;; Global variables
23
24 (define trace #f)
25
26 (define sam-host "localhost")
27 (define sam-port 8000)
28
29 (define sam-version "0.1")
30
31 ;; Actors
32
33
34 (define (make-address host port id)
35   (list id host port))
36
37 (define (make-local-address . args)
38   (make-address sam-host
39                 sam-port
40                 (if (null? args)
41                     (uuid)
42                     (car args))))
43
44 (define (address-id address)
45   (car address))
46 (define (address-host address)
47   (cadr address))
48 (define (address-port address)
49   (caddr address))
50 (define (address->string address)
51   (uri->string
52    (make-uri #:scheme "actor"
53              #:host (address-host address)
54              #:port (address-port address)
55              #:path (list '/ (address-id address)))))
56 (define (string->address str)
57   (let ((uri (uri-reference str)))
58     (make-address (uri-host uri)
59                   (uri-port uri)
60                   (cadr (uri-path uri)))))
61
62 (define (address-local? address)
63   (and (equal? (address-host address) sam-host)
64        (equal? (address-port address) sam-port)))
65
66 (define actor-table (make-hash-table))
67
68 (define (make-actor beh)
69   (let* ((address (make-local-address))
70          (id (address-id address)))
71     (hash-table-set! actor-table id beh)
72     address))
73
74 (define (deliver-message address . message)
75   (if trace (print "Delivering to " address ": " message))
76   (let ((id (address-id address)))
77     (let ((behaviour (hash-table-ref/default actor-table id '())))
78       (if (null? behaviour)
79           (print "Warning: discarded message " message
80                  " to unknown actor id " id)
81           (match (apply (hash-table-ref actor-table id) (cons address message))
82             ('done (hash-table-delete! actor-table id))
83             ('sleep 'do-nothing)
84             (new-beh (hash-table-set! actor-table id new-beh)))))))
85
86 ;; Scheduler
87
88 (define local-queue-mutex (make-mutex "message queue"))
89 (define message-available-mutex (make-mutex "message available"))
90 (mutex-lock! message-available-mutex #f #f)
91 (define local-queue (make-fifo))
92
93 (define (send-message address . message)
94   (apply (if (address-local? address)
95              send-local-message
96              send-network-message)
97          (cons address message)))
98
99 (define (send-local-message address . message)
100   (mutex-lock! local-queue-mutex)
101   (fifo-push local-queue (cons address message))
102   (mutex-unlock! message-available-mutex)
103   (mutex-unlock! local-queue-mutex))
104
105 (define (send-network-message address . message)
106   (let ((s (udp-open-socket))
107         (packet (with-output-to-string
108                   (lambda ()
109                     (write (cons address message))))))
110     (udp-bind! s #f 0)
111     (udp-connect! s
112                   (address-host address)
113                   (address-port address))
114     (udp-send s packet)
115     (udp-close-socket s)))
116
117 (define (send-message-later address time . message)
118   (thread-start!
119    (lambda ()
120      (thread-sleep! time)
121      (apply send-message (cons address message)))))
122
123 (define (next-local-message)
124   (let ((res #f))
125     (mutex-lock! message-available-mutex #f #f)
126     (mutex-lock! local-queue-mutex)
127     (set! res (fifo-pop local-queue))
128     (if (not (fifo-empty? local-queue))
129         (mutex-unlock! message-available-mutex))
130     (mutex-unlock! local-queue-mutex)
131     res))
132
133 (define (start-scheduler)
134   (let loop ()
135     (apply deliver-message (next-local-message))
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 sam-port)
146        (let loop ()
147          (let-values (((n str) (udp-recv s 65536)))
148            (match (with-input-from-string str read)
149              ((address message ...)
150               (apply send-message (cons address message)))
151              (else
152               (print "Warning: received badly formatted message string '" str "'"))))
153          (loop))))))
154
155 ;; System interface
156
157 (define reader-queue-mutex (make-mutex "reader queue"))
158 (define reader-available-mutex (make-mutex "reader available"))
159 (mutex-lock! reader-available-mutex #f #f)
160 (define reader-queue (make-fifo))
161
162 (define (next-reader)
163   (let ((res #f))
164     (mutex-lock! reader-available-mutex #f #f)
165     (mutex-lock! reader-queue-mutex)
166     (set! res (fifo-pop reader-queue))
167     (if (not (fifo-empty? reader-queue))
168         (mutex-unlock! reader-available-mutex))
169     (mutex-unlock! reader-queue-mutex)
170     res))
171
172 (define (start-console)
173   (thread-start!
174    (lambda ()
175      (let loop ()
176        (let ((reader (next-reader)))
177          (##sys#thread-block-for-i/o! (current-thread) 0 #t)
178          (thread-yield!)
179          (send-message reader (read-line)))
180        (loop)))))
181
182 ;; System initialization
183
184 (define (system-beh self . message)
185   (match message
186
187     (('shutdown)
188      (print "## System actor received shutdown message.")
189      (exit 0)
190      'done)
191
192     (('print strings ...)
193      (apply print strings)
194      'sleep)
195
196     (('read reader)
197      (mutex-lock! reader-queue-mutex)
198      (fifo-push reader-queue reader)
199      (mutex-unlock! reader-available-mutex)
200      (mutex-unlock! reader-queue-mutex)
201      'sleep)))
202
203 (define (boot-sam)
204   (start-console)
205   (start-network-listener)
206   (send-message (make-actor main-beh) (make-actor system-beh))
207   (start-scheduler))
208
209 (define (print-usage)
210   (print "Simple Actor Machine v" sam-version "\n"
211          "\n"
212          "Usage: sam -h|--help\n"
213          "       sam [-n hostname] [-p port] source-file-1 [source-file-2 [...]] "))
214
215
216 (let loop ((args (cdr (argv))))
217   (match args
218     (((or "-h" "--help"))
219      (print-usage))
220     (((or "-p" "--port") pstr rest ...)
221      (set! sam-port (string->number pstr))
222      (loop rest))
223     (((or "-n" "--hostname") hstr rest ...)
224      (set! sam-host hstr)
225      (loop rest)
226     (((? file-exists? filename) rest ...))
227      (print* "Loading " filename "...")
228      (load filename)
229      (print " done.")
230      (loop rest))
231     (()
232      (boot-sam host port main-beh))
233     (else
234      (print "Unrecognised argument '" (car args) "'.\n")
235      (print-usage))))
236