8sync/actors.scm

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