X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fsystems%2Factors.scm;h=286fef92f24103589e4599d361619a9855b560e9;hp=8bff1d4365795b11c4980178cb342ef5f6a3b1a2;hb=cc0480943c62fbfefd7192e9536f2598faf8e909;hpb=cd2611c04e1bf009d2bf6d57ec9eef5ecf0214cc diff --git a/8sync/systems/actors.scm b/8sync/systems/actors.scm index 8bff1d4..286fef9 100644 --- a/8sync/systems/actors.scm +++ b/8sync/systems/actors.scm @@ -59,6 +59,9 @@ hive-id hive-create-actor hive-create-actor* + create-actor create-actor* + self-destruct + make-message message? message-to message-action message-from @@ -109,6 +112,139 @@ slot-name))) + +;;; Messages +;;; ======== + + +(define-record-type + (make-message-intern id to from action + body in-reply-to wants-reply ; do we need hive-proxy? + ;; Are these still needed? + replied deferred-reply) + message? + (id message-id) + (to message-to) + (from message-from) + (action message-action) + (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!)) + + +(define* (make-message id to from action body + #:key in-reply-to wants-reply + replied deferred-reply) + (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) + "See if this message needs a reply still" + (and (message-wants-reply message) + (not (or (message-replied message) + (message-deferred-reply message))))) + + +(define (kwarg-list-to-alist args) + (let loop ((remaining args) + (result '())) + (match remaining + (((? keyword? key) val rest ...) + (loop rest + (cons (cons (keyword->symbol key) val) + result))) + (() result) + (_ (throw 'invalid-kwarg-list + "Invalid keyword argument list" + args))))) + + +(define (send-message 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 (hive-process-message hive message)))) + +(define (send-message-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) + #:wants-reply #t))) + (abort-to-prompt abort-to from-actor message))) + +;; TODO: Intelligently ~propagate(ish) errors on -wait functions. +;; We might have `send-message-wait-brazen' to allow callers to +;; 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) + "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) + #:in-reply-to (message-id original-message)))) + (8sync (hive-process-message hive new-message)))) + +(define (reply-message-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)) + (abort-to (hive-prompt (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) + #:wants-reply #t + #:in-reply-to (message-id original-message)))) + (abort-to-prompt abort-to from-actor new-message))) + + ;;; Main actor implementation ;;; ========================= @@ -223,7 +359,7 @@ more compact following syntax: (simple-dispatcher (list (%expand-action-item action-item) ...))))) -(define-syntax-rule (define-simple-actor class (actions ...)) +(define-syntax-rule (define-simple-actor class actions ...) (define-class class () (message-handler #:init-value (make-action-dispatch actions ...) @@ -252,6 +388,7 @@ more compact following syntax: ;; This is a map from cons cell of message-id ;; to a cons cell of (actor-id . coroutine) ;; @@: Should we have a record type? + ;; @@: Should there be any way to clear out "old" coroutines? (waiting-coroutines #:init-thunk make-hash-table #:getter hive-waiting-coroutines) ;; Message prompt @@ -311,7 +448,15 @@ more compact following syntax: (define-method (hive-process-message (hive ) message) "Handle one message, or forward it via an ambassador" - (define (process-local-message) + (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))) + + (define (resolve-actor-to) + "Get the actor the message was aimed at" (let ((actor (hive-resolve-local-actor hive (message-to message)))) (if (not actor) (throw 'actor-not-found @@ -320,29 +465,56 @@ more compact following syntax: (address->string (message-from message)) (address->string (message-to message))) message)) - (call-with-prompt (hive-prompt hive) - (lambda () - (define message-handler (actor-message-handler actor)) - ;; @@: Should a more general error handling happen here? - (message-handler actor message)) - - (lambda (kont actor message) - (let ((hive (actor-hive actor))) - ;; Register the coroutine - (hash-set! (hive-waiting-coroutines hive) - (message-id message) - (cons (actor-id actor) kont)) - ;; Send off the message - (8sync (hive-process-message hive message))))))) + actor)) + + (define (call-catching-coroutine thunk) + (call-with-prompt (hive-prompt hive) + thunk + (lambda (kont actor message) + (let ((hive (actor-hive actor))) + ;; Register the coroutine + (hash-set! (hive-waiting-coroutines hive) + (message-id message) + (cons (actor-id actor) kont)) + ;; Send off the message + (8sync (hive-process-message hive message)))))) + + (define (process-local-message) + (let ((actor (resolve-actor-to))) + (call-catching-coroutine + (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))))) (define (resume-waiting-coroutine) - (match (hash-remove! (hive-waiting-coroutines hive) - (message-in-reply-to message)) - ((_ . kont) - (kont message)) - (#f (throw 'no-waiting-coroutine - "message in-reply-to tries to resume nonexistent coroutine" - message)))) + (call-catching-coroutine + (lambda () + (match (hash-remove! (hive-waiting-coroutines hive) + (message-in-reply-to message)) + ((_ . (resume-actor-id . kont)) + (if (not (equal? (message-to message) + resume-actor-id)) + (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 + #:message message)) + (let (;; @@: How should we resolve resuming coroutines to actors who are + ;; now gone? + (actor (resolve-actor-to)) + (result (kont message))) + (maybe-autoreply actor) + result)) + (#f (throw 'no-waiting-coroutine + "message in-reply-to tries to resume nonexistent coroutine" + message)))))) (define (process-remote-message) ;; Find the ambassador @@ -422,129 +594,33 @@ Instead, actors should call create-actor." -;;; Messages -;;; ======== - - -(define-record-type - (make-message-intern id to from action - body in-reply-to wants-reply ; do we need hive-proxy? - ;; Are these still needed? - replied deferred-reply) - message? - (id message-id) - (to message-to) - (from message-from) - (action message-action) - (body message-body) - (in-reply-to message-in-reply-to) - (wants-reply message-wants-reply) +;;; Various API methods for actors to interact with the system +;;; ========================================================== - ;; 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!)) +;; TODO: move send-message and friends here...? +;; TODO: Rewrite this inside of a ? +(define* (create-actor from-actor actor-class #:rest init) + "Create an instance of actor-class. Return the new actor's id. -(define* (make-message id to from action body - #:key in-reply-to wants-reply - replied deferred-reply) - (make-message-intern id to from action body - in-reply-to wants-reply replied - deferred-reply)) +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))) -;; 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* (create-actor* from-actor actor-class id-cookie #:rest init) + "Create an instance of actor-class. Return the new actor's id. -(define* (message-ref message key #:optional (dflt %nothing-provided)) - "Extract KEY from body of MESSAGE. +Like create-actor, but permits supplying an id-cookie." + (8sync (%hive-create-actor (actor-hive from-actor) actor-class + init id-cookie))) -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 (kwarg-list-to-alist args) - (let loop ((remaining args) - (result '())) - (match remaining - (((? keyword? key) val rest ...) - (loop rest - (cons (cons (keyword->symbol key) val) - result))) - (() result) - (_ (throw 'invalid-kwarg-list - "Invalid keyword argument list" - args))))) - - -(define (send-message 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 (hive-process-message hive message)))) - -(define (send-message-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)) - (agenda-prompt (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) - #:wants-reply #t))) - (abort-to-prompt agenda-prompt from-actor message))) - -;; TODO: Intelligently ~propagate(ish) errors on -wait functions. -;; We might have `send-message-wait-brazen' to allow callers to -;; 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) - "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) - #:in-reply-to (message-id original-message)))) - (8sync (hive-process-message hive new-message)))) - -(define (reply-message-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)) - (agenda-prompt (hive-prompt (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) - #:wants-reply #t - #:in-reply-to (message-id original-message)))) - (abort-to-prompt agenda-prompt from-actor new-message))) +(define (self-destruct actor) + "Remove an actor from the hive." + (hash-remove! hive-actor-registry (actor-id actor))) @@ -574,8 +650,8 @@ an integer." -;;; Convenience procedures -;;; ====================== +;;; Basic readers / writers +;;; ======================= (define (serialize-message message) "Serialize a message for read/write"