actors: Add cleanup facility to run-hive.
[8sync.git] / 8sync / actors.scm
index b651d553232742f263b511c7cf76212b0d32b9c6..045a43a1ddc97dc9d9fb4f27767b4ac8edecfbd9 100644 (file)
@@ -25,7 +25,6 @@
   #:use-module (ice-9 match)
   #:use-module (ice-9 pretty-print)
   #:use-module (8sync agenda)
-  #:use-module (8sync repl)
   #:export (;; utilities... ought to go in their own module
             big-random-number
             big-random-number-string
@@ -48,6 +47,8 @@
             actor-id-hive
             actor-id-string
 
+            actor-am-i-alive?
+
             build-actions
 
             define-simple-actor
 
             message-auto-reply?
 
-            <- <-wait <-reply <-reply-wait
+            <- <-wait <-wait* <-reply <-reply-wait <-reply-wait*
 
             call-with-message msg-receive msg-val
 
-            ez-run-hive
+            run-hive
             bootstrap-message
 
             serialize-message write-message
 
 ;; @@: Could we get rid of some of the conditional checks through
 ;;   some macro-foo?
-(define-inlinable (send-message xtra-params from-actor to-id action
+
+(define-inlinable (send-message send-options from-actor to-id action
                                 replying-to-message wants-reply?
                                 message-body-args)
   (if replying-to-message
                         (if replying-to-message
                             (message-id replying-to-message)
                             #f))))
-    ;; TODO: add xtra-params to both of these
     (if wants-reply?
         (abort-to-prompt (hive-prompt (actor-hive from-actor))
-                         from-actor new-message)
+                         from-actor new-message send-options)
+        ;; @@: It might be that eventually we pass in send-options
+        ;;   here too.  Since <-wait and <-reply-wait are the only ones
+        ;;   that use it yet, for now it kind of just makes things
+        ;;   confusing.
         (8sync (hive-process-message hive new-message)))))
 
 
   (send-message '() from-actor to-id action
                 #f #f message-body-args))
 
+(define (<-wait* send-options from-actor to-id action . message-body-args)
+  "Like <-wait, but allows extra parameters, for example whether to
+#:accept-errors"
+  (apply wait-maybe-handle-errors
+         (send-message send-options from-actor to-id action
+                       #f #t message-body-args)
+         send-options))
+
 (define (<-wait from-actor to-id action . message-body-args)
   "Send a message from an actor to another, but wait until we get a response"
-  (send-message '() from-actor to-id action
-                #f #t message-body-args))
+  (apply <-wait* '() from-actor to-id action message-body-args))
 
 ;; TODO: Intelligently ~propagate(ish) errors on -wait functions.
 ;;   We might have `send-message-wait-brazen' to allow callers to
   (send-message '() from-actor (message-from original-message) '*auto-reply*
                 original-message #f '()))
 
+(define (<-reply-wait* send-options from-actor original-message
+                       . message-body-args)
+  "Reply to a messsage, but wait until we get a response"
+  (apply wait-maybe-handle-errors
+         (send-message send-options from-actor
+                       (message-from original-message) '*reply*
+                       original-message #t message-body-args)
+         send-options))
+
 (define (<-reply-wait from-actor original-message . message-body-args)
   "Reply to a messsage, but wait until we get a response"
-  (send-message '() from-actor (message-from original-message) '*reply*
-                original-message #t message-body-args))
+  (apply <-reply-wait* '() from-actor original-message message-body-args))
+
+(define* (wait-maybe-handle-errors message
+                                   #:key accept-errors
+                                   #:allow-other-keys)
+  "Before returning a message to a waiting caller, see if we need to
+raise an exception if an error."
+  (define action (message-action message))
+  (cond ((and (eq? action '*error*)
+              (not accept-errors))
+         (throw 'hive-unresumable-coroutine
+                "Won't resume coroutine; got an *error* as a reply"
+                #:message message))
+        (else message)))
 
 
 \f
     (call/ec find-message-handler))
   (apply method actor message (message-body message)))
 
+(define-syntax-rule (build-actions (symbol method) ...)
+  "Construct an alist of (symbol . method), where the method is wrapped
+with wrap-apply to facilitate live hacking and allow the method definition
+to come after class definition."
+  (list
+   (cons (quote symbol)
+         (wrap-apply method)) ...))
+
 (define-class <actor> ()
   ;; An address object
   (id #:init-keyword #:id
                    ;; @@: There's no reason not to use #:class instead of
                    ;;   #:each-subclass anywhere in this file, except for
                    ;;   Guile bug #25211 (#:class is broken in Guile 2.2)
-                   #:allocation #:each-subclass)
+                   #:allocation #:each-subclass
+                   #:getter actor-message-handler)
 
   ;; This is the default, "simple" way to inherit and process messages.
-  (actions #:init-value '()
+  (actions #:init-value (build-actions
+                         ;; Default cleanup method is to do nothing.
+                         (*cleanup* (const #f)))
            #:allocation #:each-subclass))
 
-(define-method (actor-message-handler (actor <actor>))
-  (slot-ref actor 'message-handler))
-
 ;;; So these are the nicer representations of addresses.
 ;;; However, they don't serialize so easily with scheme read/write, so we're
 ;;; using the simpler cons cell version below for now.
 ;;
 
 (define (make-address actor-id hive-id)
-  (cons actor-id hive-id))
+  (vector actor-id hive-id))
 
 (define (address-actor-id address)
-  (car address))
+  (vector-ref address 0))
 
 (define (address-hive-id address)
-  (cdr address))
+  (vector-ref address 1))
 
 (define (address->string address)
   (string-append (address-actor-id address) "@"
 (define %current-actor
   (make-parameter #f))
 
+(define (actor-am-i-alive? actor)
+  (hive-resolve-local-actor (actor-hive actor) (actor-id actor)))
+
 
 \f
 ;;; Actor utilities
 ;;; ===============
 
-(define-syntax-rule (build-actions (symbol method) ...)
-  "Construct an alist of (symbol . method), where the method is wrapped
-with wrap-apply to facilitate live hacking and allow the method definition
-to come after class definition."
-  (list
-   (cons (quote symbol)
-         (wrap-apply method)) ...))
-
 (define-syntax-rule (define-simple-actor class action ...)
   (define-class class (<actor>)
     (actions #:init-value (build-actions action ...)
@@ -369,12 +405,25 @@ to come after class definition."
            (build-actions
             ;; This is in the case of an ambassador failing to forward a
             ;; message... it reports it back to the hive
-            (*failed-forward* hive-handle-failed-forward))))
+            (*failed-forward* hive-handle-failed-forward)
+            (*cleanup-all* hive-handle-cleanup-all))))
 
 (define-method (hive-handle-failed-forward (hive <hive>) message)
   "Handle an ambassador failing to forward a message"
   'TODO)
 
+(define-method (hive-handle-cleanup-all (hive <hive>) message)
+  "Send a message to all actors in our registry to clean themselves up."
+  ;; Unfortunately we have to do this hack and run over the list
+  ;; twice, because hash-for-each would result in an unrewindable
+  ;; continuation.
+  (define actor-ids
+    (hash-map->list (lambda (actor-id actor) actor-id)
+                    (hive-actor-registry hive)))
+  (for-each (lambda (actor-id)
+              (<- hive actor-id '*cleanup*))
+            actor-ids))
+
 (define* (make-hive #:key hive-id)
   (let ((hive (make <hive>
                 #:id (make-address
@@ -382,6 +431,8 @@ to come after class definition."
                                  (big-random-number-string))))))
     ;; Set the hive's actor reference to itself
     (set! (actor-hive hive) hive)
+    ;; Register the actor with itself
+    (hive-register-actor! hive hive)
     hive))
 
 (define-method (hive-id (hive <hive>))
@@ -438,6 +489,14 @@ to come after class definition."
     ;; unresumable continuation.
     (lambda () (hive-process-message hive new-message))))
 
+(define-record-type <waiting-on-reply>
+  (make-waiting-on-reply actor-id kont send-options)
+  waiting-on-reply?
+  (actor-id waiting-on-reply-actor-id)
+  (kont waiting-on-reply-kont)
+  (send-options waiting-on-reply-send-options))
+
+
 (define-method (hive-process-message (hive <hive>) message)
   "Handle one message, or forward it via an ambassador"
   (define (maybe-autoreply actor)
@@ -487,11 +546,12 @@ to come after class definition."
           (8sync (queued-error-handling-thunk))))
     (call-with-prompt (hive-prompt hive)
       call-catching-errors
-      (lambda (kont actor message)
+      (lambda (kont actor message send-options)
         ;; Register the coroutine
         (hash-set! (hive-waiting-coroutines hive)
                    (message-id message)
-                   (cons (actor-id actor) kont))
+                   (make-waiting-on-reply
+                    (actor-id actor) kont send-options))
         ;; Send off the message
         (8sync (hive-process-message hive message)))))
 
@@ -513,42 +573,29 @@ to come after class definition."
   (define (resume-waiting-coroutine)
     (case (message-action message)
       ;; standard reply / auto-reply
-      ((*reply* *auto-reply*)
+      ((*reply* *auto-reply* *error*)
        (call-catching-coroutine
         (lambda ()
           (match (hash-remove! (hive-waiting-coroutines hive)
                                (message-in-reply-to message))
-            ((_ . (resume-actor-id . kont))
+            ((_ . waiting)
              (if (not (equal? (message-to message)
-                              resume-actor-id))
+                              (waiting-on-reply-actor-id waiting)))
                  (throw 'resuming-to-wrong-actor
                         "Attempted to resume a coroutine to the wrong actor!"
                         #:expected-actor-id (message-to message)
-                        #:got-actor-id resume-actor-id
+                        #:got-actor-id (waiting-on-reply-actor-id waiting)
                         #:message message))
-             (let (;; @@: How should we resolve resuming coroutines to actors who are
-                   ;;   now gone?
-                   (actor (resolve-actor-to))
-                   (result (kont message)))
+             (let* (;; @@: How should we resolve resuming coroutines to actors who are
+                    ;;   now gone?
+                    (actor (resolve-actor-to))
+                    (kont (waiting-on-reply-kont waiting))
+                    (result (kont message)))
                (maybe-autoreply actor)
                result))
             (#f (throw 'no-waiting-coroutine
                        "message in-reply-to tries to resume nonexistent coroutine"
                        message))))))
-      ;; Yikes, an error!
-      ((*error*)
-       ;; @@: Not what we want in the long run?
-       ;; What we'd *prefer* to do is to resume this message
-       ;; and throw an error inside the message handler
-       ;; (say, from send-mesage-wait), but that causes a SIGABRT (??!!)
-       (hash-remove! (hive-waiting-coroutines hive)
-                     (message-in-reply-to message))
-       (let ((explaination
-              (if (eq? (message-action message) '*reply*)
-                  "Won't resume coroutine; got an *error* as a reply"
-                  "Won't resume coroutine because action is not *reply*")))
-         (throw 'hive-unresumable-coroutine
-                explaination #:message message)))
       ;; Unhandled action for a reply!
       (else
        (throw 'hive-unresumable-coroutine
@@ -604,6 +651,7 @@ that method for documentation."
                       init #f))
 
 (define* (hive-create-actor* hive actor-class id-cookie #:rest init)
+  "Create an actor, but also add a 'cookie' to the name for debugging"
   (%hive-create-actor hive actor-class
                       init id-cookie))
 
@@ -667,22 +715,24 @@ Like create-actor, but permits supplying an id-cookie."
 ;;; 8sync bootstrap utilities
 ;;; =========================
 
-(define* (ez-run-hive hive initial-tasks #:key repl-server)
-  "Start up an agenda and run HIVE in it with INITIAL-TASKS.
-
-Should we start up a cooperative REPL for live hacking?  REPL-SERVER
-wants to know!  You can pass it #t or #f, or if you want to specify a port,
-an integer."
-  (let* ((queue (list->q initial-tasks))
-         (agenda (make-agenda #:pre-unwind-handler print-error-and-continue
-                              #:queue queue)))
-    (cond
-     ;; If repl-server is an integer, we'll use that as the port
-     ((integer? repl-server)
-      (spawn-and-queue-repl-server! agenda repl-server))
-     (repl-server
-      (spawn-and-queue-repl-server! agenda)))
-    (start-agenda agenda)))
+(define* (run-hive hive initial-tasks)
+  "Start up an agenda and run HIVE in it with INITIAL-TASKS."
+  (dynamic-wind
+    (const #f)
+    (lambda ()
+      (let* ((queue (list->q initial-tasks))
+             (agenda (make-agenda #:pre-unwind-handler print-error-and-continue
+                                  #:queue queue)))
+        (start-agenda agenda)))
+    ;; Run cleanup
+    (lambda ()
+      (run-hive-cleanup hive))))
+
+(define (run-hive-cleanup hive)
+  (let ((queue (list->q (list (bootstrap-message hive (actor-id hive)
+                                                 '*cleanup-all*)))))
+    (start-agenda
+     (make-agenda #:queue queue))))
 
 (define (bootstrap-message hive to-id action . message-body-args)
   (wrap