actors: Rename msg-receive, msg-val to mbody-receive, mbody-val.
[8sync.git] / 8sync / actors.scm
1 ;;; 8sync --- Asynchronous programming for Guile
2 ;;; Copyright (C) 2016 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 actors)
20   #:use-module (oop goops)
21   #:use-module (srfi srfi-9)
22   #:use-module (srfi srfi-9 gnu)
23   #:use-module (ice-9 control)
24   #:use-module (ice-9 format)
25   #:use-module (ice-9 match)
26   #:use-module (ice-9 pretty-print)
27   #:use-module (8sync agenda)
28   #:export (;; utilities... ought to go in their own module
29             big-random-number
30             big-random-number-string
31             simple-message-id-generator
32
33             <actor>
34             actor-id
35             actor-message-handler
36
37             %current-actor
38
39             ;;; Commenting out the <address> type for now;
40             ;;; it may be back when we have better serializers
41             ;; <address>
42             make-address address?
43             address-actor-id address-hive-id
44
45             address->string
46             actor-id-actor
47             actor-id-hive
48             actor-id-string
49
50             actor-alive?
51
52             build-actions
53
54             define-simple-actor
55
56             <hive>
57             make-hive
58             ;; There are more methods for the hive, but there's
59             ;; no reason for the outside world to look at them maybe?
60             hive-id
61             bootstrap-actor bootstrap-actor*
62
63             create-actor create-actor*
64             self-destruct
65
66             <message>
67             make-message message?
68             message-to message-action message-from
69             message-id message-body message-in-reply-to
70             message-wants-reply
71
72             message-auto-reply?
73
74             <- <-* <-wait <-wait* <-reply <-reply* <-reply-wait <-reply-wait*
75
76             call-with-message mbody-receive mbody-val
77
78             run-hive
79             bootstrap-message
80
81             serialize-message write-message
82             serialize-message-pretty pprint-message
83             read-message read-message-from-string))
84
85 ;; For ids
86 (define %random-state
87   (make-parameter (random-state-from-platform)))
88
89 ;; Same size as a uuid4 I think...
90 (define random-number-size (expt 2 128))
91
92 (define (big-random-number)
93   (random random-number-size (%random-state)))
94
95 ;; Would be great to get this base64 encoded instead.
96 (define (big-random-number-string)
97   ;; @@: This is slow.  Using format here is wasteful.
98   (format #f "~x" (big-random-number)))
99
100 ;; @@: This is slow.  A mere ~275k / second on my (old) machine.
101 ;;   The main cost seems to be in number->string.
102 (define (simple-message-id-generator)
103   ;; Prepending this cookie makes message ids unique per hive
104   (let ((prefix (format #f "~x:" (big-random-number)))
105         (counter 0))
106     (lambda ()
107       (set! counter (1+ counter))
108       (string-append prefix (number->string counter)))))
109
110
111 \f
112 ;;; Messages
113 ;;; ========
114
115
116 ;; @@: We may want to add a deferred-reply to the below, similar to
117 ;;   what we had in XUDD, for actors which do their own response
118 ;;   queueing.... ie, that might receive messages but need to shelve
119 ;;   them to be acted upon after something else is taken care of.
120
121 (define-record-type <message>
122   (make-message-intern id to from action
123                        body in-reply-to wants-reply
124                        replied)
125   message?
126   (id message-id)                    ; id of this message
127   (to message-to)                    ; actor id this is going to
128   (from message-from)                ; actor id of sender
129   (action message-action)            ; action (a symbol) to be handled
130   (body message-body)                ; argument list "body" of message
131   (in-reply-to message-in-reply-to)  ; message id this is in reply to, if any
132   (wants-reply message-wants-reply)  ; whether caller is waiting for reply
133   (replied message-replied           ; was this message replied to?
134            set-message-replied!))
135
136
137 (define* (make-message id to from action body
138                        #:key in-reply-to wants-reply
139                        replied)
140   (make-message-intern id to from action body
141                        in-reply-to wants-reply replied))
142
143 (define (message-auto-reply? message)
144   (eq? (message-action message) '*auto-reply*))
145
146 (define (message-needs-reply? message)
147   "See if this message needs a reply still"
148   (and (message-wants-reply message)
149        (not (message-replied message))))
150
151
152 (define (kwarg-list-to-alist args)
153   (let loop ((remaining args)
154              (result '()))
155     (match remaining
156       (((? keyword? key) val rest ...)
157        (loop rest
158              (cons (cons (keyword->symbol key) val) 
159                    result)))
160       (() result)
161       (_ (throw 'invalid-kwarg-list
162                 "Invalid keyword argument list"
163                 args)))))
164
165
166 ;;; See: https://web.archive.org/web/20081223021934/http://mumble.net/~jar/articles/oo-moon-weinreb.html
167 ;;;   (also worth seeing: http://mumble.net/~jar/articles/oo.html )
168
169 ;; This is the internal, generalized message sending method.
170 ;; Users shouldn't use it!  Use the <-foo forms instead.
171
172 ;; @@: Could we get rid of some of the conditional checks through
173 ;;   some macro-foo?
174
175 (define-inlinable (send-message send-options from-actor to-id action
176                                 replying-to-message wants-reply?
177                                 message-body-args)
178   (if replying-to-message
179       (set-message-replied! replying-to-message #t))
180   (let* ((hive (actor-hive from-actor))
181          (new-message
182           (make-message (hive-gen-message-id hive) to-id
183                         (actor-id from-actor) action
184                         message-body-args
185                         #:wants-reply wants-reply?
186                         #:in-reply-to
187                         (if replying-to-message
188                             (message-id replying-to-message)
189                             #f))))
190     (if wants-reply?
191         (abort-to-prompt (hive-prompt (actor-hive from-actor))
192                          from-actor new-message send-options)
193         ;; @@: It might be that eventually we pass in send-options
194         ;;   here too.  Since <-wait and <-reply-wait are the only ones
195         ;;   that use it yet, for now it kind of just makes things
196         ;;   confusing.
197         (8sync (hive-process-message hive new-message)))))
198
199 (define (<- to-id action . message-body-args)
200   "Send a message from an actor to another actor"
201   (send-message '() (%current-actor) to-id action
202                 #f #f message-body-args))
203
204 (define (<-* send-options to-id action . message-body-args)
205   "Like <-*, but allows extra parameters via send-options"
206   (define* (really-send #:key (actor (%current-actor))
207                         #:allow-other-keys)
208     (send-message send-options actor to-id action
209                   #f #f message-body-args))
210   (apply really-send send-options))
211
212 (define (<-wait to-id action . message-body-args)
213   "Send a message from an actor to another, but wait until we get a response"
214   (wait-maybe-handle-errors
215    (send-message '() (%current-actor) to-id action
216                  #f #t message-body-args)))
217
218 (define (<-wait* send-options to-id action . message-body-args)
219   "Like <-wait, but allows extra parameters, for example whether to
220 #:accept-errors"
221   (define* (really-send #:key (actor (%current-actor))
222                         #:allow-other-keys)
223     (apply wait-maybe-handle-errors
224            (send-message send-options actor to-id action
225                          #f #t message-body-args)
226            send-options))
227   (apply really-send send-options))
228
229 ;; TODO: Intelligently ~propagate(ish) errors on -wait functions.
230 ;;   We might have `send-message-wait-brazen' to allow callers to
231 ;;   not have an exception thrown and instead just have a message with
232 ;;   the appropriate '*error* message returned.
233
234 (define (<-reply original-message . message-body-args)
235   "Reply to a message"
236   (send-message '() (%current-actor) (message-from original-message) '*reply*
237                 original-message #f message-body-args))
238
239 (define (<-reply* send-options original-message . message-body-args)
240   "Like <-reply, but allows extra parameters via send-options"
241   (define* (really-send #:key (actor (%current-actor))
242                         #:allow-other-keys)
243     (send-message send-options actor
244                   (message-from original-message) '*reply*
245                   original-message #f message-body-args))
246   (apply really-send send-options))
247
248 (define (<-auto-reply actor original-message)
249   "Auto-reply to a message.  Internal use only!"
250   (send-message '() actor (message-from original-message) '*auto-reply*
251                 original-message #f '()))
252
253 (define (<-reply-wait original-message . message-body-args)
254   "Reply to a messsage, but wait until we get a response"
255   (wait-maybe-handle-errors
256    (send-message '() (%current-actor)
257                  (message-from original-message) '*reply*
258                  original-message #t message-body-args)))
259
260 (define (<-reply-wait* send-options original-message
261                        . message-body-args)
262   "Like <-reply-wait, but allows extra parameters via send-options"
263   (define* (really-send #:key (actor (%current-actor))
264                         #:allow-other-keys)
265     (apply wait-maybe-handle-errors
266            (send-message send-options actor
267                          (message-from original-message) '*reply*
268                          original-message #t message-body-args)
269            send-options))
270   (apply really-send send-options))
271
272 (define* (wait-maybe-handle-errors message
273                                    #:key accept-errors
274                                    #:allow-other-keys)
275   "Before returning a message to a waiting caller, see if we need to
276 raise an exception if an error."
277   (define action (message-action message))
278   (cond ((and (eq? action '*error*)
279               (not accept-errors))
280          (throw 'hive-unresumable-coroutine
281                 "Won't resume coroutine; got an *error* as a reply"
282                 #:message message))
283         (else message)))
284
285
286 \f
287 ;;; Main actor implementation
288 ;;; =========================
289
290 (define (actor-inheritable-message-handler actor message)
291   (define action (message-action message))
292   (define (find-message-handler return)
293     (for-each (lambda (this-class)
294                 (define actions
295                   (or (and (class-slot-definition this-class 'actions)
296                            (class-slot-ref this-class 'actions))
297                       '()))
298                 (for-each (match-lambda
299                             ((action-name . method)
300                              (when (eq? action-name action)
301                                (return method))))
302                           actions))
303               (class-precedence-list (class-of actor)))
304     (throw 'action-not-found
305            "No appropriate action handler found for actor"
306            #:action action
307            #:actor actor
308            #:message message))
309   (define method
310     (call/ec find-message-handler))
311   (apply method actor message (message-body message)))
312
313 (define-syntax-rule (build-actions (symbol method) ...)
314   "Construct an alist of (symbol . method), where the method is wrapped
315 with wrap-apply to facilitate live hacking and allow the method definition
316 to come after class definition."
317   (list
318    (cons (quote symbol)
319          (wrap-apply method)) ...))
320
321 (define-class <actor> ()
322   ;; An address object
323   (id #:init-keyword #:id
324       #:getter actor-id)
325   ;; The hive we're connected to.
326   ;; We need this to be able to send messages.
327   (hive #:init-keyword #:hive
328         #:accessor actor-hive)
329   ;; How we receive and process new messages
330   (message-handler #:init-value actor-inheritable-message-handler
331                    ;; @@: There's no reason not to use #:class instead of
332                    ;;   #:each-subclass anywhere in this file, except for
333                    ;;   Guile bug #25211 (#:class is broken in Guile 2.2)
334                    #:allocation #:each-subclass
335                    #:getter actor-message-handler)
336
337   ;; This is the default, "simple" way to inherit and process messages.
338   (actions #:init-value (build-actions
339                          ;; Default init method is to do nothing.
340                          (*init* (const #f))
341                          ;; Default cleanup method is to do nothing.
342                          (*cleanup* (const #f)))
343            #:allocation #:each-subclass))
344
345 ;;; So these are the nicer representations of addresses.
346 ;;; However, they don't serialize so easily with scheme read/write, so we're
347 ;;; using the simpler cons cell version below for now.
348
349 ;; (define-record-type <address>
350 ;;   (make-address actor-id hive-id)  ; @@: Do we want the trailing -id?
351 ;;   address?
352 ;;   (actor-id address-actor-id)
353 ;;   (hive-id address-hive-id))
354 ;;
355 ;; (set-record-type-printer!
356 ;;  <address>
357 ;;  (lambda (record port)
358 ;;    (format port "<address: ~s@~s>"
359 ;;            (address-actor-id record) (address-hive-id record))))
360 ;;
361
362 (define (make-address actor-id hive-id)
363   (vector actor-id hive-id))
364
365 (define (address-actor-id address)
366   (vector-ref address 0))
367
368 (define (address-hive-id address)
369   (vector-ref address 1))
370
371 (define (address->string address)
372   (string-append (address-actor-id address) "@"
373                  (address-hive-id address)))
374
375 (define-method (actor-id-actor (actor <actor>))
376   "Get the actor id component of the actor-id"
377   (address-actor-id (actor-id actor)))
378
379 (define-method (actor-id-hive (actor <actor>))
380   "Get the hive id component of the actor-id"
381   (address-hive-id (actor-id actor)))
382
383 (define-method (actor-id-string (actor <actor>))
384   "Render the full actor id as a human-readable string"
385   (address->string (actor-id actor)))
386
387 (define %current-actor
388   (make-parameter #f))
389
390 (define (actor-alive? actor)
391   (hive-resolve-local-actor (actor-hive actor) (actor-id actor)))
392
393
394 \f
395 ;;; Actor utilities
396 ;;; ===============
397
398 (define-syntax-rule (define-simple-actor class action ...)
399   (define-class class (<actor>)
400     (actions #:init-value (build-actions action ...)
401              #:allocation #:each-subclass)))
402
403 \f
404 ;;; The Hive
405 ;;; ========
406 ;;;   Every actor has a hive.  The hive is a kind of "meta-actor"
407 ;;;   which routes all the rest of the actors in a system.
408
409 (define-generic hive-handle-failed-forward)
410
411 (define-class <hive> (<actor>)
412   (actor-registry #:init-thunk make-hash-table
413                   #:getter hive-actor-registry)
414   (msg-id-generator #:init-thunk simple-message-id-generator
415                     #:getter hive-msg-id-generator)
416   ;; Ambassadors are used (or will be) for inter-hive communication.
417   ;; These are special actors that know how to route messages to other hives.
418   (ambassadors #:init-thunk make-weak-key-hash-table
419                #:getter hive-ambassadors)
420   ;; Waiting coroutines
421   ;; This is a map from cons cell of message-id
422   ;;   to a cons cell of (actor-id . coroutine)
423   ;; @@: Should we have a <waiting-coroutine> record type?
424   ;; @@: Should there be any way to clear out "old" coroutines?
425   (waiting-coroutines #:init-thunk make-hash-table
426                       #:getter hive-waiting-coroutines)
427   ;; Message prompt
428   ;; When actors send messages to each other they abort to this prompt
429   ;; to send the message, then carry on their way
430   (prompt #:init-thunk make-prompt-tag
431           #:getter hive-prompt)
432   (actions #:allocation #:each-subclass
433            #:init-value
434            (build-actions
435             ;; This is in the case of an ambassador failing to forward a
436             ;; message... it reports it back to the hive
437             (*failed-forward* hive-handle-failed-forward)
438             ;; These are called at start and end of run-hive
439             (*init-all* hive-handle-init-all)
440             (*cleanup-all* hive-handle-cleanup-all))))
441
442 (define-method (hive-handle-init-all (hive <hive>) message)
443   "Run *init* method on all actors in registry"
444   ;; We have to do this hack and run over the list
445   ;; twice, because hash-for-each would result in an unrewindable
446   ;; continuation, and to avoid the hash-map changing during the
447   ;; middle of this.
448   (define actor-ids
449     (hash-map->list (lambda (actor-id actor) actor-id)
450                     (hive-actor-registry hive)))
451   (for-each (lambda (actor-id)
452               ;; @@: This could maybe just be <-, but we want actors
453               ;;   to be used to the expectation in all circumstances
454               ;;   that their init method is "waited on".
455               (<-wait actor-id '*init*))
456             actor-ids))
457
458 (define-method (hive-handle-failed-forward (hive <hive>) message)
459   "Handle an ambassador failing to forward a message"
460   'TODO)
461
462 (define-method (hive-handle-cleanup-all (hive <hive>) message)
463   "Send a message to all actors in our registry to clean themselves up."
464   ;; We have to do this hack and run over the list
465   ;; twice, because hash-for-each would result in an unrewindable
466   ;; continuation, and to avoid the hash-map changing during the
467   ;; middle of this.
468   (define actor-ids
469     (hash-map->list (lambda (actor-id actor) actor-id)
470                     (hive-actor-registry hive)))
471   (for-each (lambda (actor-id)
472               (<- actor-id '*cleanup*))
473             actor-ids))
474
475 (define* (make-hive #:key hive-id)
476   (let ((hive (make <hive>
477                 #:id (make-address
478                       "hive" (or hive-id
479                                  (big-random-number-string))))))
480     ;; Set the hive's actor reference to itself
481     (set! (actor-hive hive) hive)
482     ;; Register the actor with itself
483     (hive-register-actor! hive hive)
484     hive))
485
486 (define-method (hive-id (hive <hive>))
487   (actor-id-hive hive))
488
489 (define-method (hive-gen-actor-id (hive <hive>) cookie)
490   (make-address (if cookie
491                     (string-append cookie ":" (big-random-number-string))
492                     (big-random-number-string))
493                 (hive-id hive)))
494
495 (define-method (hive-gen-message-id (hive <hive>))
496   "Generate a message id using HIVE's message id generator"
497   ((hive-msg-id-generator hive)))
498
499 (define-method (hive-resolve-local-actor (hive <hive>) actor-address)
500   (hash-ref (hive-actor-registry hive) actor-address))
501
502 (define-method (hive-resolve-ambassador (hive <hive>) ambassador-address)
503   (hash-ref (hive-ambassadors hive) ambassador-address))
504
505 (define-method (make-forward-request (hive <hive>) (ambassador <actor>) message)
506   (make-message (hive-gen-message-id hive) (actor-id ambassador)
507                 ;; If we make the hive not an actor, we could either switch this
508                 ;; to #f or to the original actor...?
509                 ;; Maybe some more thinking should be done on what should
510                 ;; happen in case of failure to forward?  Handling ambassador failures
511                 ;; seems like the primary motivation for the hive remaining an actor.
512                 (actor-id hive)
513                 '*forward*
514                 `((original . ,message))))
515
516 (define-method (hive-reply-with-error (hive <hive>) original-message
517                                       error-key error-args)
518   ;; We only supply the error-args if the original sender is on the same hive
519   (define (orig-actor-on-same-hive?)
520     (equal? (hive-id hive)
521             (address-hive-id (message-from original-message))))
522   (set-message-replied! original-message #t)
523   (let* ((new-message-body
524           (if (orig-actor-on-same-hive?)
525               `(#:original-message ,original-message
526                 #:error-key ,error-key
527                 #:error-args ,error-args)
528               `(#:original-message ,original-message
529                 #:error-key ,error-key)))
530          (new-message (make-message (hive-gen-message-id hive)
531                                     (message-from original-message)
532                                     (actor-id hive) '*error*
533                                     new-message-body
534                                     #:in-reply-to (message-id original-message))))
535     ;; We only return a thunk, rather than run 8sync here, because if
536     ;; we ran 8sync in the middle of a catch we'd end up with an
537     ;; unresumable continuation.
538     (lambda () (hive-process-message hive new-message))))
539
540 (define-record-type <waiting-on-reply>
541   (make-waiting-on-reply actor-id kont send-options)
542   waiting-on-reply?
543   (actor-id waiting-on-reply-actor-id)
544   (kont waiting-on-reply-kont)
545   (send-options waiting-on-reply-send-options))
546
547
548 (define-method (hive-process-message (hive <hive>) message)
549   "Handle one message, or forward it via an ambassador"
550   (define (maybe-autoreply actor)
551     ;; Possibly autoreply
552     (if (message-needs-reply? message)
553         (<-auto-reply actor message)))
554
555   (define (resolve-actor-to)
556     "Get the actor the message was aimed at"
557     (let ((actor (hive-resolve-local-actor hive (message-to message))))
558       (if (not actor)
559           (throw 'actor-not-found
560                  (format #f "Message ~a from ~a directed to nonexistant actor ~a"
561                          (message-id message)
562                          (address->string (message-from message))
563                          (address->string (message-to message)))
564                  message))
565       actor))
566
567   (define (call-catching-coroutine thunk)
568     (define queued-error-handling-thunk #f)
569     (define (call-catching-errors)
570       ;; TODO: maybe parameterize (or attach to hive) and use
571       ;;   maybe-catch-all from agenda.scm
572       ;; @@: Why not just use with-throw-handler and let the catch
573       ;;   happen at the agenda?  That's what we used to do, but
574       ;;   it ended up with a SIGABRT.  See:
575       ;;     http://lists.gnu.org/archive/html/bug-guile/2016-05/msg00003.html
576       (catch #t
577         thunk
578         ;; In the actor model, we don't totally crash on errors.
579         (lambda _ #f)
580         ;; If an error happens, we raise it
581         (lambda (key . args)
582           (if (message-needs-reply? message)
583               ;; If the message is waiting on a reply, let them know
584               ;; something went wrong.
585               ;; However, we have to do it outside of this catch
586               ;; routine, or we'll end up in an unrewindable continuation
587               ;; situation.
588               (set! queued-error-handling-thunk
589                     (hive-reply-with-error hive message key args)))
590           ;; print error message
591           (apply print-error-and-continue key args)))
592       ;; @@: This is a kludge.  See above for why.
593       (if queued-error-handling-thunk
594           (8sync (queued-error-handling-thunk))))
595     (call-with-prompt (hive-prompt hive)
596       call-catching-errors
597       (lambda (kont actor message send-options)
598         ;; Register the coroutine
599         (hash-set! (hive-waiting-coroutines hive)
600                    (message-id message)
601                    (make-waiting-on-reply
602                     (actor-id actor) kont send-options))
603         ;; Send off the message
604         (8sync (hive-process-message hive message)))))
605
606   (define (process-local-message)
607     (let ((actor (resolve-actor-to)))
608       (call-catching-coroutine
609        (lambda ()
610          (define message-handler (actor-message-handler actor))
611          ;; @@: Should a more general error handling happen here?
612          (parameterize ((%current-actor actor))
613            (let ((result
614                   (message-handler actor message)))
615              (maybe-autoreply actor)
616              ;; Returning result allows actors to possibly make a run-request
617              ;; at the end of handling a message.
618              ;; ... We do want that, right?
619              result))))))
620
621   (define (resume-waiting-coroutine)
622     (case (message-action message)
623       ;; standard reply / auto-reply
624       ((*reply* *auto-reply* *error*)
625        (call-catching-coroutine
626         (lambda ()
627           (match (hash-remove! (hive-waiting-coroutines hive)
628                                (message-in-reply-to message))
629             ((_ . waiting)
630              (if (not (equal? (message-to message)
631                               (waiting-on-reply-actor-id waiting)))
632                  (throw 'resuming-to-wrong-actor
633                         "Attempted to resume a coroutine to the wrong actor!"
634                         #:expected-actor-id (message-to message)
635                         #:got-actor-id (waiting-on-reply-actor-id waiting)
636                         #:message message))
637              (let* (;; @@: How should we resolve resuming coroutines to actors who are
638                     ;;   now gone?
639                     (actor (resolve-actor-to))
640                     (kont (waiting-on-reply-kont waiting))
641                     (result (kont message)))
642                (maybe-autoreply actor)
643                result))
644             (#f (throw 'no-waiting-coroutine
645                        "message in-reply-to tries to resume nonexistent coroutine"
646                        message))))))
647       ;; Unhandled action for a reply!
648       (else
649        (throw 'hive-unresumable-coroutine
650               "Won't resume coroutine, nonsense action on reply message"
651               #:action (message-action message)
652               #:message message))))
653
654   (define (process-remote-message)
655     ;; Find the ambassador
656     (let* ((remote-hive-id (hive-id (message-to message)))
657            (ambassador (hive-resolve-ambassador remote-hive-id))
658            (message-handler (actor-message-handler ambassador))
659            (forward-request (make-forward-request hive ambassador message)))
660       (message-handler ambassador forward-request)))
661
662   (let ((to (message-to message)))
663     ;; This seems to be an easy mistake to make, so check that addressing
664     ;; is correct here
665     (if (not to)
666         (throw 'missing-addressee
667                "`to' field is missing on message"
668                #:message message))
669     (if (hive-actor-local? hive to)
670         (if (message-in-reply-to message)
671             (resume-waiting-coroutine)
672             (process-local-message))
673         (process-remote-message))))
674
675 (define-method (hive-actor-local? (hive <hive>) address)
676   (equal? (hive-id hive) (address-hive-id address)))
677
678 (define-method (hive-register-actor! (hive <hive>) (actor <actor>))
679   (hash-set! (hive-actor-registry hive) (actor-id actor) actor))
680
681 (define-method (%hive-create-actor (hive <hive>) actor-class
682                                    init-args id-cookie send-init?)
683   "Actual method called by bootstrap-actor / create-actor.
684
685 Since this is a define-method it can't accept fancy define* arguments,
686 so this gets called from the nicer bootstrap-actor interface.  See
687 that method for documentation."
688   (let* ((actor-id (hive-gen-actor-id hive id-cookie))
689          (actor (apply make actor-class
690                        #:hive hive
691                        #:id actor-id
692                        init-args)))
693     (hive-register-actor! hive actor)
694     ;; Wait on actor to init
695     (when send-init?
696       (<-wait actor-id '*init*))
697     ;; return the actor id
698     actor-id))
699
700 (define* (bootstrap-actor hive actor-class #:rest init-args)
701   "Create an actor on HIVE using ACTOR-CLASS passing in INIT-ARGS args"
702   (%hive-create-actor hive actor-class
703                     init-args (symbol->string (class-name actor-class))
704                     #f))
705
706 (define* (bootstrap-actor* hive actor-class id-cookie #:rest init-args)
707   "Create an actor, but also allow customizing a 'cookie' added to the id
708 for debugging"
709   (%hive-create-actor hive actor-class
710                     init-args id-cookie
711                     #f))
712
713 (define (call-with-message message proc)
714   "Applies message body arguments into procedure, with message as first
715 argument.  Similar to call-with-values in concept."
716   (apply proc message (message-body message)))
717
718 ;; (mbody-receive (<- bar baz)
719 ;;     (baz)
720 ;;   basil)
721
722 ;; Emacs: (put 'mbody-receive 'scheme-indent-function 2)
723
724 ;; @@: Or receive-msg or receieve-message or??
725 (define-syntax-rule (mbody-receive arglist message body ...)
726   "Call body with arglist (which can accept arguments like lambda*)
727 applied from the message-body of message."
728   (call-with-message message
729                      (lambda* arglist
730                        body ...)))
731
732 (define (mbody-val message)
733   "Retrieve the first value from the message-body of message.
734 Like single value return from a procedure call.  Probably the most
735 common case when waiting on a reply from some action invocation."
736   (call-with-message message
737                      (lambda (_ val) val)))
738
739 \f
740 ;;; Various API methods for actors to interact with the system
741 ;;; ==========================================================
742
743 ;; TODO: move send-message and friends here...?
744
745 (define* (create-actor from-actor actor-class #:rest init-args)
746   "Create an instance of actor-class.  Return the new actor's id.
747
748 This is the method actors should call directly (unless they want
749 to supply an id-cookie, in which case they should use
750 create-actor*)."
751   (%hive-create-actor (actor-hive from-actor) actor-class
752                       init-args #f #t))
753
754
755 (define* (create-actor* from-actor actor-class id-cookie #:rest init-args)
756   "Create an instance of actor-class.  Return the new actor's id.
757
758 Like create-actor, but permits supplying an id-cookie."
759   (%hive-create-actor (actor-hive from-actor) actor-class
760                       init-args id-cookie #t))
761
762
763 (define* (self-destruct actor #:key (cleanup #t))
764   "Remove an actor from the hive.
765
766 Unless #:cleanup is set to #f, this will first have the actor handle
767 its '*cleanup* action handler."
768   (when cleanup
769     (<-wait (actor-id actor) '*cleanup*))
770   (hash-remove! (hive-actor-registry (actor-hive actor))
771                 (actor-id actor)))
772
773
774 \f
775 ;;; 8sync bootstrap utilities
776 ;;; =========================
777
778 (define* (run-hive hive initial-tasks
779                    #:key (cleanup #t))
780   "Start up an agenda and run HIVE in it with INITIAL-TASKS."
781   (dynamic-wind
782     (const #f)
783     (lambda ()
784       (let* ((queue (list->q
785                      (cons (bootstrap-message hive (actor-id hive) '*init-all*)
786                            initial-tasks)))
787              (agenda (make-agenda #:pre-unwind-handler print-error-and-continue
788                                   #:queue queue)))
789         (run-agenda agenda)))
790     ;; Run cleanup
791     (lambda ()
792       (when cleanup
793         (run-hive-cleanup hive)))))
794
795 (define (run-hive-cleanup hive)
796   (let ((queue (list->q (list (bootstrap-message hive (actor-id hive)
797                                                  '*cleanup-all*)))))
798     (run-agenda
799      (make-agenda #:queue queue))))
800
801 (define (bootstrap-message hive to-id action . message-body-args)
802   (wrap
803    (apply <-* `(#:actor ,hive) to-id action message-body-args)))
804
805
806 \f
807 ;;; Basic readers / writers
808 ;;; =======================
809
810 (define (serialize-message message)
811   "Serialize a message for read/write"
812   (list
813    (message-id message)
814    (message-to message)
815    (message-from message)
816    (message-action message)
817    (message-body message)
818    (message-in-reply-to message)
819    (message-wants-reply message)
820    (message-replied message)))
821
822 (define* (write-message message #:optional (port (current-output-port)))
823   "Write out a message to a port for easy reading later.
824
825 Note that if a sub-value can't be easily written to something
826 Guile's `read' procedure knows how to read, this doesn't do anything
827 to improve that.  You'll need a better serializer for that.."
828   (write (serialize-message message) port))
829
830 (define (serialize-message-pretty message)
831   "Serialize a message in a way that's easy for humans to read."
832   `(*message*
833     (id ,(message-id message))
834     (to ,(message-to message))
835     (from ,(message-from message))
836     (action ,(message-action message))
837     (body ,(message-body message))
838     (in-reply-to ,(message-in-reply-to message))
839     (wants-reply ,(message-wants-reply message))
840     (replied ,(message-replied message))))
841
842 (define (pprint-message message)
843   "Pretty print a message."
844   (pretty-print (serialize-message-pretty message)))
845
846 (define* (read-message #:optional (port (current-input-port)))
847   "Read a message serialized via serialize-message from PORT"
848   (match (read port)
849     ((id to from action body in-reply-to wants-reply replied)
850      (make-message-intern
851       id to from action body
852       in-reply-to wants-reply replied))
853     (anything-else
854      (throw 'message-read-bad-structure
855             "Could not read message from structure"
856             anything-else))))
857
858 (define (read-message-from-string message-str)
859   "Read message from MESSAGE-STR"
860   (with-input-from-string message-str
861     (lambda ()
862       (read-message (current-input-port)))))