X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fsystems%2Factors.scm;h=debb606eb09fe743d7454a96b6abf7d1a7a531e2;hp=270b25f452909e5a000780bd7de5e968d47d63d2;hb=22d2cc32b43143a32a2989fdc1917e4b5dc7ee96;hpb=d3202269c645127c176cad41185fa0f95be34d89
diff --git a/8sync/systems/actors.scm b/8sync/systems/actors.scm
index 270b25f..debb606 100644
--- a/8sync/systems/actors.scm
+++ b/8sync/systems/actors.scm
@@ -22,6 +22,7 @@
#:use-module (oop goops)
#:use-module (srfi srfi-9)
#:use-module (srfi srfi-9 gnu)
+ #:use-module (ice-9 control)
#:use-module (ice-9 format)
#:use-module (ice-9 match)
#:use-module (ice-9 pretty-print)
@@ -36,6 +37,8 @@
actor-id
actor-message-handler
+ %current-actor
+
;;; Commenting out the
type for now;
;;; it may be back when we have better serializers
;;
@@ -47,8 +50,7 @@
actor-id-hive
actor-id-string
- mlambda define-mhandler
- make-action-dispatch
+ simple-dispatcher build-actions make-action-dispatch
define-simple-actor
@@ -66,13 +68,13 @@
message-to message-action message-from
message-id message-body message-in-reply-to
message-wants-reply
- message-ref
- send-message send-message-wait
- reply-message reply-message-wait
+ message-auto-reply?
<- <-wait <-reply <-reply-wait
+ call-with-message msg-receive msg-val
+
ez-run-hive
bootstrap-message
@@ -111,13 +113,15 @@
;;; ========
+;; @@: We may want to add a deferred-reply to the below, similar to
+;; what we had in XUDD, for actors which do their own response
+;; queueing.... ie, that might receive messages but need to shelve
+;; them to be acted upon after something else is taken care of.
+
(define-record-type
(make-message-intern id to from action
body in-reply-to wants-reply
- replied
- ;; @@: Not used yet.
- ;; Will we ever find a real use case?
- deferred-reply)
+ replied)
message?
(id message-id)
(to message-to)
@@ -126,55 +130,22 @@
(body message-body)
(in-reply-to message-in-reply-to)
(wants-reply message-wants-reply)
-
- ;; See XUDD source for these. Not use yet, maybe eventually will be?
- ;; XUDD uses them for autoreply.
- ;; Requiring mutation on message objects is clearly not great,
- ;; but it may be worth it...? Investigate!
- (replied message-replied set-message-replied!)
- (deferred-reply message-deferred-reply set-message-deferred-reply!))
+ (replied message-replied set-message-replied!))
(define* (make-message id to from action body
#:key in-reply-to wants-reply
- replied deferred-reply)
+ replied)
(make-message-intern id to from action body
- in-reply-to wants-reply replied
- deferred-reply))
-
-;; Note: the body of messages is currently an alist, but it's created
-;; from a keyword based property list (see the following two functions).
-;; But, that's an extra conversion step, and maybe totally unnecessary:
-;; we already have message-ref, and this could just pull a keyword
-;; from a property list.
-;; The main ways this might be useful are error checking,
-;; serialization across the wire (but even that might require some
-;; change), and using existing tooling (though adding new tooling
-;; would be negligible in implementation effort.)
-
-;; This cons cell is immutable and unique (for eq? tests)
-(define %nothing-provided (cons 'nothing 'provided))
-
-(define* (message-ref message key #:optional (dflt %nothing-provided))
- "Extract KEY from body of MESSAGE.
-
-Optionally set default with [DFLT]
-If key not found and DFLT not provided, throw an error."
- (let ((result (assoc key (message-body message))))
- (if result (cdr result)
- (if (eq? dflt %nothing-provided)
- (throw 'message-missing-key
- "Message body does not contain key and no default provided"
- #:key key
- #:message message)
- dflt))))
-
-
-(define (message-needs-reply message)
+ in-reply-to wants-reply replied))
+
+(define (message-auto-reply? message)
+ (eq? (message-action message) '*auto-reply*))
+
+(define (message-needs-reply? message)
"See if this message needs a reply still"
(and (message-wants-reply message)
- (not (or (message-replied message)
- (message-deferred-reply message)))))
+ (not (message-replied message))))
(define (kwarg-list-to-alist args)
@@ -191,21 +162,24 @@ If key not found and DFLT not provided, throw an error."
args)))))
-(define (send-message from-actor to-id action . message-body-args)
+;;; See: https://web.archive.org/web/20081223021934/http://mumble.net/~jar/articles/oo-moon-weinreb.html
+;;; (also worth seeing: http://mumble.net/~jar/articles/oo.html )
+
+(define (<- from-actor to-id action . message-body-args)
"Send a message from an actor to another actor"
(let* ((hive (actor-hive from-actor))
(message (make-message (hive-gen-message-id hive) to-id
(actor-id from-actor) action
- (kwarg-list-to-alist message-body-args))))
- (8sync-nowait (hive-process-message hive message))))
+ message-body-args)))
+ (8sync (hive-process-message hive message))))
-(define (send-message-wait from-actor to-id action . message-body-args)
+(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"
(let* ((hive (actor-hive from-actor))
(abort-to (hive-prompt (actor-hive from-actor)))
(message (make-message (hive-gen-message-id hive) to-id
(actor-id from-actor) action
- (kwarg-list-to-alist message-body-args)
+ message-body-args
#:wants-reply #t)))
(abort-to-prompt abort-to from-actor message)))
@@ -214,20 +188,29 @@ If key not found and DFLT not provided, throw an error."
;; not have an exception thrown and instead just have a message with
;; the appropriate '*error* message returned.
-(define (reply-message from-actor original-message
- . message-body-args)
+(define (<-reply from-actor original-message . message-body-args)
"Reply to a message"
(set-message-replied! original-message #t)
(let* ((hive (actor-hive from-actor))
(new-message (make-message (hive-gen-message-id hive)
(message-from original-message)
(actor-id from-actor) '*reply*
- (kwarg-list-to-alist message-body-args)
+ message-body-args
#:in-reply-to (message-id original-message))))
- (8sync-nowait (hive-process-message hive new-message))))
+ (8sync (hive-process-message hive new-message))))
-(define (reply-message-wait from-actor original-message
- . message-body-args)
+(define (<-auto-reply from-actor original-message)
+ "Auto-reply to a message. Internal use only!"
+ (set-message-replied! original-message #t)
+ (let* ((hive (actor-hive from-actor))
+ (new-message (make-message (hive-gen-message-id hive)
+ (message-from original-message)
+ (actor-id from-actor) '*auto-reply*
+ '()
+ #:in-reply-to (message-id original-message))))
+ (8sync (hive-process-message hive new-message))))
+
+(define (<-reply-wait from-actor original-message . message-body-args)
"Reply to a messsage, but wait until we get a response"
(set-message-replied! original-message #t)
(let* ((hive (actor-hive from-actor))
@@ -235,22 +218,12 @@ If key not found and DFLT not provided, throw an error."
(new-message (make-message (hive-gen-message-id hive)
(message-from original-message)
(actor-id from-actor) '*reply*
- (kwarg-list-to-alist message-body-args)
+ message-body-args
#:wants-reply #t
#:in-reply-to (message-id original-message))))
(abort-to-prompt abort-to from-actor new-message)))
-;;; Aliases!
-;;; See: http://mumble.net/~jar/articles/oo-moon-weinreb.html
-;;; (also worth seeing: http://mumble.net/~jar/articles/oo.html )
-
-(define <- send-message)
-(define <-wait send-message-wait)
-(define <-reply reply-message)
-(define <-reply-wait reply-message-wait)
-
-
;;; Main actor implementation
;;; =========================
@@ -264,7 +237,38 @@ If key not found and DFLT not provided, throw an error."
(hive #:init-keyword #:hive
#:accessor actor-hive)
;; How we receive and process new messages
- (message-handler #:allocation #:each-subclass))
+ (message-handler #:init-value (wrap-apply actor-inheritable-message-handler)
+ ;; @@: 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)
+
+ ;; This is the default, "simple" way to inherit and process messages.
+ (actions #:init-value '()
+ #:allocation #:each-subclass))
+
+(define (actor-inheritable-message-handler actor message)
+ (define action (message-action message))
+ (define (find-message-handler return)
+ (for-each (lambda (this-class)
+ (define actions
+ (or (and (class-slot-definition this-class 'actions)
+ (class-slot-ref this-class 'actions))
+ '()))
+ (for-each (match-lambda
+ ((action-name . method)
+ (when (eq? action-name action)
+ (return method))))
+ actions))
+ (class-precedence-list (class-of actor)))
+ (throw 'action-not-found
+ "No appropriate action handler found for actor"
+ #:action action
+ #:actor actor
+ #:message message))
+ (define method
+ (call/ec find-message-handler))
+ (apply method actor message (message-body message)))
(define-method (actor-message-handler (actor ))
(slot-ref actor 'message-handler))
@@ -311,64 +315,43 @@ If key not found and DFLT not provided, throw an error."
"Render the full actor id as a human-readable string"
(address->string (actor-id actor)))
+(define %current-actor
+ (make-parameter #f))
+
;;; Actor utilities
;;; ===============
-
-(define-syntax-rule (with-message-args (message message-arg ...)
- body body* ...)
- (let ((message-arg (message-ref message (quote message-arg))) ...)
- body body* ...))
-
-(define-syntax mlambda
- (syntax-rules ()
- "A lambda for building message handlers.
-
-Use it like:
- (mlambda (actor message foo)
- ...)
-
-Which is like doing manually:
- (lambda (actor message)
- (let ((foo (message-ref message foo)))
- ...))"
- ((_ (actor message message-arg ...)
- body body* ...)
- (lambda (actor message)
- (with-message-args (message message-arg ...) body body* ...)))))
-
-;; @@: Sadly, docstrings won't work with this...
-;; I think we need to bust out syntax-case to make that happen...
-(define-syntax-rule (define-mhandler (name actor message message-arg ...)
- body ...)
- (define name
- (mlambda (actor message message-arg ...)
- body ...)))
-
(define (simple-dispatcher action-map)
(lambda (actor message)
(let* ((action (message-action message))
(method (assoc-ref action-map action)))
(if (not method)
+ ;; @@: There's every possibility this should be handled in
+ ;; hive-process-message instead.
(throw 'action-not-found
"No appropriate action handler found for actor"
#:action action
#:actor actor
#:message message
#:available-actions (map car action-map)))
- (method actor message))))
+ (apply method actor message (message-body message)))))
(define-syntax %expand-action-item
(syntax-rules ()
- ((_ ((action-name action-args ...) body ...))
- (cons (quote action-name)
- (mlambda (action-args ...)
- body ...)))
((_ (action-name handler))
(cons (quote action-name) handler))))
+(define-syntax-rule (build-actions action-item ...)
+ "Build a mapping of actions. Same syntax as make-action-dispatch
+but this doesn't build the dispatcher for you (you probably want to
+pass it to simple-dispatcher).
+
+The advantage here is that since this simply builds an alist, you can
+compose it with other action maps."
+ (list (%expand-action-item action-item) ...))
+
(define-syntax make-action-dispatch
(syntax-rules ()
"Expand a list of action names and actions into an alist
@@ -390,8 +373,7 @@ more compact following syntax:
((party actor message)
(display \"Life of the party!\")))"
((make-action-dispatch action-item ...)
- (simple-dispatcher
- (list (%expand-action-item action-item) ...)))))
+ (simple-dispatcher (build-actions action-item ...)))))
(define-syntax-rule (define-simple-actor class actions ...)
(define-class class ()
@@ -487,26 +469,27 @@ more compact following syntax:
(set-message-replied! original-message #t)
(let* ((new-message-body
(if (orig-actor-on-same-hive?)
- `((original-message . ,original-message)
- (error-key . ,error-key)
- (error-args . ,error-args))
- `((original-message . ,original-message)
- (error-key . ,error-key))))
+ `(#:original-message ,original-message
+ #:error-key ,error-key
+ #:error-args ,error-args)
+ `(#:original-message ,original-message
+ #:error-key ,error-key)))
(new-message (make-message (hive-gen-message-id hive)
(message-from original-message)
(actor-id hive) '*error*
new-message-body
#:in-reply-to (message-id original-message))))
- (8sync-nowait (hive-process-message hive new-message))))
+ ;; We only return a thunk, rather than run 8sync here, because if
+ ;; we ran 8sync in the middle of a catch we'd end up with an
+ ;; unresumable continuation.
+ (lambda () (hive-process-message hive new-message))))
(define-method (hive-process-message (hive ) message)
"Handle one message, or forward it via an ambassador"
(define (maybe-autoreply actor)
;; Possibly autoreply
- (if (message-needs-reply message)
- ;; @@: Should we give *autoreply* as the action instead of *reply*?
- (reply-message actor message
- #:*auto-reply* #t)))
+ (if (message-needs-reply? message)
+ (<-auto-reply actor message)))
(define (resolve-actor-to)
"Get the actor the message was aimed at"
@@ -521,14 +504,33 @@ more compact following syntax:
actor))
(define (call-catching-coroutine thunk)
+ (define queued-error-handling-thunk #f)
(define (call-catching-errors)
- (with-throw-handler
- #t thunk
- (lambda (key . args)
- (if (message-needs-reply message)
- ;; If the message is waiting on a reply, let them know
- ;; something went wrong.
- (hive-reply-with-error hive message key args)))))
+ ;; TODO: maybe parameterize (or attach to hive) and use
+ ;; maybe-catch-all from agenda.scm
+ ;; @@: Why not just use with-throw-handler and let the catch
+ ;; happen at the agenda? That's what we used to do, but
+ ;; it ended up with a SIGABRT. See:
+ ;; http://lists.gnu.org/archive/html/bug-guile/2016-05/msg00003.html
+ (catch #t
+ thunk
+ ;; In the actor model, we don't totally crash on errors.
+ (lambda _ #f)
+ ;; If an error happens, we raise it
+ (lambda (key . args)
+ (if (message-needs-reply? message)
+ ;; If the message is waiting on a reply, let them know
+ ;; something went wrong.
+ ;; However, we have to do it outside of this catch
+ ;; routine, or we'll end up in an unrewindable continuation
+ ;; situation.
+ (set! queued-error-handling-thunk
+ (hive-reply-with-error hive message key args)))
+ ;; print error message
+ (apply print-error-and-continue key args)))
+ ;; @@: This is a kludge. See above for why.
+ (if queued-error-handling-thunk
+ (8sync (queued-error-handling-thunk))))
(call-with-prompt (hive-prompt hive)
call-catching-errors
(lambda (kont actor message)
@@ -545,17 +547,19 @@ more compact following syntax:
(lambda ()
(define message-handler (actor-message-handler actor))
;; @@: Should a more general error handling happen here?
- (let ((result
- (message-handler actor message)))
- (maybe-autoreply actor)
- ;; Returning result allows actors to possibly make a run-request
- ;; at the end of handling a message.
- ;; ... We do want that, right?
- result)))))
+ (parameterize ((%current-actor actor))
+ (let ((result
+ (message-handler actor message)))
+ (maybe-autoreply actor)
+ ;; Returning result allows actors to possibly make a run-request
+ ;; at the end of handling a message.
+ ;; ... We do want that, right?
+ result))))))
(define (resume-waiting-coroutine)
(cond
- ((eq? (message-action message) '*reply*)
+ ((or (eq? (message-action message) '*reply*)
+ (eq? (message-action message) '*auto-reply*))
(call-catching-coroutine
(lambda ()
(match (hash-remove! (hive-waiting-coroutines hive)
@@ -615,7 +619,7 @@ more compact following syntax:
(process-remote-message))))
(define-method (hive-actor-local? (hive ) address)
- (hash-ref (hive-actor-registry hive) address))
+ (equal? (hive-id hive) (address-hive-id address)))
(define-method (hive-register-actor! (hive ) (actor ))
(hash-set! (hive-actor-registry hive) (actor-id actor) actor))
@@ -644,6 +648,31 @@ that method for documentation."
(%hive-create-actor hive actor-class
init id-cookie))
+(define (call-with-message message proc)
+ "Applies message body arguments into procedure, with message as first
+argument. Similar to call-with-values in concept."
+ (apply proc message (message-body message)))
+
+;; (msg-receive (<- bar baz)
+;; (baz)
+;; basil)
+
+;; Emacs: (put 'msg-receive 'scheme-indent-function 2)
+
+;; @@: Or receive-msg or receieve-message or??
+(define-syntax-rule (msg-receive arglist message body ...)
+ "Call body with arglist (which can accept arguments like lambda*)
+applied from the message-body of message."
+ (call-with-message message
+ (lambda* arglist
+ body ...)))
+
+(define (msg-val message)
+ "Retrieve the first value from the message-body of message.
+Like single value return from a procedure call. Probably the most
+common case when waiting on a reply from some action invocation."
+ (call-with-message message
+ (lambda (_ val) val)))
;;; Various API methods for actors to interact with the system
@@ -657,16 +686,16 @@ that method for documentation."
This is the method actors should call directly (unless they want
to supply an id-cookie, in which case they should use
create-actor*)."
- (8sync (%hive-create-actor (actor-hive from-actor) actor-class
- init #f)))
+ (%hive-create-actor (actor-hive from-actor) actor-class
+ init #f))
(define* (create-actor* from-actor actor-class id-cookie #:rest init)
"Create an instance of actor-class. Return the new actor's id.
Like create-actor, but permits supplying an id-cookie."
- (8sync (%hive-create-actor (actor-hive from-actor) actor-class
- init id-cookie)))
+ (%hive-create-actor (actor-hive from-actor) actor-class
+ init id-cookie))
(define (self-destruct actor)
@@ -698,7 +727,7 @@ an integer."
(define (bootstrap-message hive to-id action . message-body-args)
(wrap
- (apply send-message hive to-id action message-body-args)))
+ (apply <- hive to-id action message-body-args)))
@@ -715,8 +744,7 @@ an integer."
(message-body message)
(message-in-reply-to message)
(message-wants-reply message)
- (message-replied message)
- (message-deferred-reply message)))
+ (message-replied message)))
(define* (write-message message #:optional (port (current-output-port)))
"Write out a message to a port for easy reading later.
@@ -736,8 +764,7 @@ to improve that. You'll need a better serializer for that.."
(body ,(message-body message))
(in-reply-to ,(message-in-reply-to message))
(wants-reply ,(message-wants-reply message))
- (replied ,(message-replied message))
- (deferred-reply ,(message-deferred-reply message))))
+ (replied ,(message-replied message))))
(define (pprint-message message)
"Pretty print a message."
@@ -746,10 +773,10 @@ to improve that. You'll need a better serializer for that.."
(define* (read-message #:optional (port (current-input-port)))
"Read a message serialized via serialize-message from PORT"
(match (read port)
- ((id to from action body in-reply-to wants-reply replied deferred-reply)
+ ((id to from action body in-reply-to wants-reply replied)
(make-message-intern
id to from action body
- in-reply-to wants-reply replied deferred-reply))
+ in-reply-to wants-reply replied))
(anything-else
(throw 'message-read-bad-structure
"Could not read message from structure"