diff --git a/README.md b/README.md index 48833ea..53d221c 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ is removed, every other message stays where it was. [[:add n] :when (pos? n) (add n)] ; a guard; one that throws is false [[x x] :same] ; a repeated name must match equal values [msg (log msg)] ; binds the whole message - [:after 1000 :timeout]) ; ms, 0 polls, nil waits forever + [:after 1000 :timeout]) ; ms, 0 polls, nil or :infinity waits forever ``` As in Erlang, a symbol bound where the `receive` is written is not a binder: @@ -102,8 +102,12 @@ Inside any actor, `(act/state)`, `(act/set-state! v)` and ordinary, trappable reason. `:normal` sent to yourself exits you. - Linking to a dead process gives `:noproc`. - `monitor!` returns a ref; the watcher gets `[:DOWN ref :process actor - reason]`, with `:noproc` if the actor was already dead. `demonitor!` with - `{:flush true}` also removes a delivered DOWN. + reason]`, with `:noproc` if the actor was already dead. A monitor made by + name names it in the DOWN as `[:At name node]`, Erlang's `{Name, Node}`, + and is down at once with `:noproc` if no process holds the name. + `demonitor!` with `{:flush true}` also removes a delivered DOWN. +- A ref (`make-ref`) carries the node that made it. Wherever a timeout is + taken, `nil` and `:infinity` both wait forever. - `register!` fails if the name is taken or the actor already has one, and a name is released when its actor exits. @@ -203,11 +207,17 @@ gen_fsm is not provided; OTP deprecated it in favour of gen_statem. ## gen_event `ensemble.gen-event`: a manager hands each event to its handlers in order. -`h-handle-event` returns `[:ok state]` or `:remove-handler`. A handler that -throws is removed, with `h-terminate` called on `[:error reason]`, and the -manager and the other handlers carry on. `add-sup-handler!` ties a handler to -the calling actor both ways. `call!` talks to one handler (`:bad-module` if -there is none), and `delete-handler!` returns what `h-terminate` returned. +`h-handle-event` returns `[:ok state]`, `[:ok state :hibernate]` (the +manager hibernates), `:remove-handler`, or `[:swap-handler reason state id +f]`. A handler that throws is removed, with `h-terminate` called on +`[:error reason]`, and the manager and the other handlers carry on. A +handler that also implements `InfoHandler` gets the manager's other +messages through `h-handle-info`. `add-sup-handler!` ties a handler to the +calling actor both ways, and the link goes with the actor's last handler. +`swap-handler!` and `swap-sup-handler!` replace a handler as OTP's do: the +old one's `h-terminate` runs, and `(f what-it-returned)` is the new one. +`call!` talks to one handler (`:bad-module` if there is none), and +`delete-handler!` returns what `h-terminate` returned. ## Supervisors @@ -333,7 +343,10 @@ process or on another machine. each `monitor-node!` gets `[:nodedown node]`. A link to a remote pid that is gone gives `:noproc`. - `spawn-on` starts a fn named by a symbol on the other node, as - `spawn(Node, M, F, A)` does. + `spawn(Node, M, F, A)` does, if that node allows it: `start!`'s `:spawn` + option is a set of namespaces or a predicate on the symbol, and without + it a node starts nothing for a peer, since peers aren't authenticated + yet. - Connections come from a `Transport`. The loopback transport joins nodes in one VM, but each frame is still printed and read back as EDN, so what crosses is exactly what a wire would carry. A socket transport plugs in diff --git a/src/ensemble/actor.clj b/src/ensemble/actor.clj index 9b42a2c..e6b6537 100644 --- a/src/ensemble/actor.clj +++ b/src/ensemble/actor.clj @@ -112,9 +112,20 @@ (defn make-ref "A unique reference (Erlang's make_ref), for tagging a request so its reply - can be told apart from every other message." + can be told apart from every other message. It carries the node that made + it, so refs from two runtimes never collide." [] - {::ref (swap! counter inc)}) + {::ref (swap! counter inc) ::node (node)}) + +(defn ref-node + "The node a ref was made on." + [r] + (::node r)) + +(defn timeout-ms + "A timeout as ms, or nil for none: nil and :infinity both wait forever." + [t] + (when-not (= :infinity t) t)) ;; process table and registry ------------------------------------------- @@ -401,27 +412,44 @@ (defn- watched? [x] (and (some? x) (not= ::firing x))) +(defn- name-object + "What a DOWN names for a monitor made by name, Erlang's {RegName, Node}: + [:At name node]." + [dest] + (if (vector? dest) + [:At (norm-name (second dest)) (nth dest 2)] + [:At (norm-name dest) (node)])) + (defn monitor! - "Monitor actor from the current actor. Returns a ref. When actor exits, - the current actor receives [:DOWN ref :process actor reason]; if it is - already dead the reason is :noproc. A monitor is one-way: the monitoring - actor is never affected by the exit. Each call makes a new monitor." - [actor] + "Monitor dest -- a process, a name, or [:At name node] -- from the current + actor. Returns a ref. When the process exits, the current actor + receives [:DOWN ref :process object reason], object the process, or + [:At name node] for a monitor made by name; reason is :noproc if it was + already dead, or no process held the name. A monitor is one-way: the + monitoring actor is never affected by the exit. Each call makes a new + monitor." + [dest] (let [me (self) - actor (resolve-dest actor) + obj (when-not (pid? dest) (name-object dest)) + actor (if obj (try (resolve-dest dest) (catch Throwable _ nil)) dest) ref (make-ref) - watching (::monitoring me)] - (swap! watching assoc ref actor) - ;; the entry in watching is a one-shot claim: the DOWN is sent only by - ;; whoever turns it to ::firing, and demonitor! only by removing it, so - ;; a demonitor that wins knows no DOWN will ever come - (proc/-add-monitor actor ref - (fn [reason] - (let [[before _] (swap-vals! watching - (fn [w] (if (watched? (get w ref)) (assoc w ref ::firing) w)))] - (when (watched? (get before ref)) - (enqueue! me [:DOWN ref :process actor reason]) - (swap! watching dissoc ref))))))) + watching (::monitoring me) + who (or obj actor)] + (if (nil? actor) + ;; a name nobody holds: down at once + (do (enqueue! me [:DOWN ref :process who :noproc]) ref) + (do + (swap! watching assoc ref actor) + ;; the entry in watching is a one-shot claim: the DOWN is sent only by + ;; whoever turns it to ::firing, and demonitor! only by removing it, so + ;; a demonitor that wins knows no DOWN will ever come + (proc/-add-monitor actor ref + (fn [reason] + (let [[before _] (swap-vals! watching + (fn [w] (if (watched? (get w ref)) (assoc w ref ::firing) w)))] + (when (watched? (get before ref)) + (enqueue! me [:DOWN ref :process who reason]) + (swap! watching dissoc ref))))))))) (defn- flush! "Remove every mailbox message pred accepts." @@ -815,12 +843,13 @@ (defn receive-match "Block until a message matches one of the compiled patterns pats whose guard (ok? clause-index bindings) accepts it. Returns [clause-index msg env], or - [:timeout] when timeout-ms (nil: wait forever) elapses first. Exit signals + [:timeout] when timeout-ms (nil or :infinity: wait forever) elapses first. Exit signals are acted on before each look at the mailbox." ([pats timeout-ms] (receive-match pats always timeout-ms)) ([pats ok? timeout-ms] (let [me (or (self) (throw (ex-info "receive outside an actor" {}))) - deadline (when timeout-ms (+ (System/currentTimeMillis) timeout-ms))] + timeout-ms (when-not (= :infinity timeout-ms) timeout-ms) + deadline (when timeout-ms (+ (now-ms) timeout-ms))] (loop [start 0] (drain-signals! me) (let [saved @(::saved me) @@ -838,7 +867,7 @@ :else (let [scanned (count @(::saved me))] (if deadline - (let [left (- deadline (System/currentTimeMillis))] + (let [left (- deadline (now-ms))] (if (pos? left) (do (a/alts!! [(::bell me) (a/timeout left)]) (recur scanned)) [:timeout])) @@ -883,7 +912,8 @@ clauses: [:else & body] matches the next message whatever it is - [:after ms & body] runs body if nothing matches within ms (nil: never) + [:after ms & body] runs body if nothing matches within ms (nil or + :infinity: never) Returns the value of the chosen clause's body." [& clauses] diff --git a/src/ensemble/gen_event.clj b/src/ensemble/gen_event.clj index d2d1c5e..dbb1781 100644 --- a/src/ensemble/gen_event.clj +++ b/src/ensemble/gen_event.clj @@ -10,9 +10,12 @@ h-handle-call [:ok reply state] or [:remove-handler reply] h-terminate anything; add-handler!/delete-handler! return it - As in OTP, a handler that throws or returns something else is removed, with - h-terminate called on [:error reason]; the manager and the other handlers - carry on. A handler added with add-sup-handler! is tied to the actor that + h-handle-event may also return [:ok state :hibernate], which hibernates the + manager, or [:swap-handler reason state id f], which swaps the handler as + swap-handler! does. A handler that also implements InfoHandler receives + the manager's other messages through h-handle-info. As in OTP, a handler + that throws or returns something else is removed, with h-terminate called + on [:error reason]; the manager and the other handlers carry on. A handler added with add-sup-handler! is tied to the actor that added it: if that actor exits, the handler is removed with [:stop reason]; if the handler is removed for any reason but delete-handler!, that actor receives [:gen-event-EXIT id reason]." @@ -21,10 +24,16 @@ (defprotocol Handler (h-init [this] "Return the handler's initial state.") - (h-handle-event [this event state] "Return [:ok state] or :remove-handler.") + (h-handle-event [this event state] "Return [:ok state], [:ok state :hibernate], :remove-handler or [:swap-handler reason state id f].") (h-handle-call [this request state] "Return [:ok reply state] or [:remove-handler reply].") (h-terminate [this reason state] "Run once as the handler is removed.")) +(defprotocol InfoHandler + (h-handle-info [this msg state] + "Optional, as in OTP: take a message the manager received that is not an + event, a call or a supervising process's exit. Returns as + h-handle-event.")) + (defn- hfind [hs id] (first (filter #(= id (:id %)) hs))) (defn- hdel [hs id] (filterv #(not= id (:id %)) hs)) @@ -32,51 +41,100 @@ (defn- remove-handler "Remove h for reason: run its terminate (a throw there is ignored, as in OTP) and tell a supervising actor, unless it asked for the removal itself. - Returns [handlers terminate-result]." + An owner left with no handler of its own is unlinked. Returns [handlers + terminate-result]." [hs h reason tell?] - (let [r (try (h-terminate (:handler h) reason (:st h)) (catch Throwable e [:error e]))] - (when (and tell? (:owner h)) - (act/! (:owner h) [:gen-event-EXIT (:id h) reason])) - [(hdel hs (:id h)) r])) - -(defn- event-to - "Hand ev to one handler; return the handlers after it has run." - [hs h ev] - (let [r (try (h-handle-event (:handler h) ev (:st h)) (catch Throwable e [::crash e]))] - (cond - (= :remove-handler r) (first (remove-handler hs h :remove-handler true)) - (and (vector? r) (= :ok (first r)) (= 2 (count r))) - (mapv #(if (= (:id h) (:id %)) (assoc % :st (second r)) %) hs) - (and (vector? r) (= ::crash (first r))) - (first (remove-handler hs h [:error (act/reason-of (second r))] true)) - :else (first (remove-handler hs h [:error [:bad-return-value r]] true))))) - -(defn- notify-all [hs ev] - (reduce (fn [acc h] (if (hfind acc (:id h)) (event-to acc h ev) acc)) hs hs)) - -(defn- add [hs id handler owner] + (let [r (try (h-terminate (:handler h) reason (:st h)) (catch Throwable e [:error e])) + hs* (hdel hs (:id h))] + (when-let [owner (:owner h)] + (when tell? (act/! owner [:gen-event-EXIT (:id h) reason])) + (when-not (some #(= owner (:owner %)) hs*) + (act/unlink! owner))) + [hs* r])) + +(defn- add + "Add handler under id, at position at (the end when nil), supervised by + owner when there is one: [[:ok nil] hs] or [[:error reason] hs]." + [hs id handler owner at] (if (hfind hs id) [[:error :already-added] hs] (let [r (try [:ok (h-init handler)] (catch Throwable e [:error (act/reason-of e)]))] (if (= :ok (first r)) - (do (when owner (act/link! owner)) - [[:ok nil] (conj hs {:id id :handler handler :st (second r) :owner owner})]) + (let [entry {:id id :handler handler :st (second r) :owner owner} + at (or at (count hs))] + (when owner (act/link! owner)) + [[:ok nil] (vec (concat (take at hs) [entry] (drop at hs)))]) [r hs])))) +(defn- swap + "Replace handler h with one made by (f terminate-result) under new-id, as + OTP's swap_handler: h's terminate runs with reason, and what it returns + is the new handler's start. The new one is supervised by owner." + [hs h reason new-id f owner] + (let [at (count (take-while #(not= (:id h) (:id %)) hs)) + [hs* res] (remove-handler hs h reason false) + handler (try (f res) (catch Throwable e e))] + (if (instance? Throwable handler) + [[:error (act/reason-of handler)] hs*] + (add hs* new-id handler owner at)))) + +(defn- run-handler + "Run (f) for handler h, a callback that answers as h-handle-event does, + and return [handlers hibernate?] after it. A handler that throws or + answers anything else is removed with [:error reason]." + [hs h f] + (let [r (try (f) (catch Throwable e [::crash e]))] + (cond + (= :remove-handler r) [(first (remove-handler hs h :remove-handler true)) false] + (and (vector? r) (= :ok (first r)) (<= 2 (count r) 3) + (or (= 2 (count r)) (= :hibernate (nth r 2)))) + [(mapv #(if (= (:id h) (:id %)) (assoc % :st (second r)) %) hs) (= 3 (count r))] + (and (vector? r) (= :swap-handler (first r)) (= 5 (count r))) + (let [[_ reason st new-id f] r] + [(second (swap hs (assoc h :st st) reason new-id f (:owner h))) false]) + (and (vector? r) (= ::crash (first r))) + [(first (remove-handler hs h [:error (act/reason-of (second r))] true)) false] + :else [(first (remove-handler hs h [:error [:bad-return-value r]] true)) false]))) + +(defn- each-handler + "Run (f h) for each handler still present that takes? accepts, in order: + [handlers hibernate?], hibernating when any asked to." + [hs takes? f] + (reduce (fn [[acc hib] h] + (if (and (hfind acc (:id h)) (takes? h)) + (let [[acc h?] (run-handler acc h #(f h))] [acc (or hib h?)]) + [acc hib])) + [hs false] hs)) + +(defn- notify-all [hs ev] + (each-handler hs any? (fn [h] (h-handle-event (:handler h) ev (:st h))))) + +(defn- info-all [hs msg] + (each-handler hs #(satisfies? InfoHandler (:handler %)) + (fn [h] (h-handle-info (:handler h) msg (:st h))))) + +(defn- noreply [[hs hibernate?]] + (if hibernate? [:noreply hs :hibernate] [:noreply hs])) + (defrecord Manager [initial] gs/Server (init [_] - [:ok (reduce (fn [hs [id h]] (second (add hs id h nil))) [] initial)]) + [:ok (reduce (fn [hs [id h]] (second (add hs id h nil nil))) [] initial)]) (handle-call [_ req _ hs] (case (first req) :add (let [[_ id h owner] req - [r hs*] (add hs id h owner)] + [r hs*] (add hs id h owner nil)] [:reply r hs*]) :delete (let [[_ id reason] req] (if-let [h (hfind hs id)] (let [[hs* r] (remove-handler hs h reason false)] [:reply [:ok r] hs*]) [:reply [:error :not-found] hs])) - :sync-notify [:reply [:ok nil] (notify-all hs (second req))] + :swap (let [[_ old-id reason new-id f owner] req] + (if-let [h (hfind hs old-id)] + (let [[r hs*] (swap hs h reason new-id f owner)] [:reply r hs*]) + [:reply [:error :not-found] hs])) + :sync-notify (let [[hs* hib] (notify-all hs (second req))] + (if hib [:reply [:ok nil] hs* :hibernate] [:reply [:ok nil] hs*])) :call (let [[_ id hreq] req h (hfind hs id)] (if (nil? h) @@ -94,17 +152,19 @@ [:reply [:error reason] (first (remove-handler hs h [:error reason] true))]))))) :which [:reply [:ok (mapv :id hs)] hs])) (handle-cast [_ req hs] - [:noreply (notify-all hs (second req))]) + (noreply (notify-all hs (second req)))) (handle-info [_ msg hs] - (if (and (vector? msg) (= :EXIT (first msg))) - (let [[_ from reason] msg] + (let [owner-exit (when (and (vector? msg) (= :EXIT (first msg))) + (let [[_ from] msg] (when (some #(= from (:owner %)) hs) from)))] + (if owner-exit [:noreply (reduce (fn [acc h] - (if (= from (:owner h)) - (first (remove-handler acc h [:stop reason] false)) + (if (= owner-exit (:owner h)) + (first (remove-handler acc h [:stop (nth msg 2)] false)) acc)) - hs hs)]) - [:noreply hs])) + hs hs)] + (noreply (info-all hs msg))))) (handle-timeout [_ hs] [:noreply hs]) + (handle-continue [_ _ hs] [:noreply hs]) (terminate [_ reason hs] (doseq [h hs] (remove-handler hs h (if (= :normal reason) :stop [:stop reason]) true)))) @@ -136,6 +196,22 @@ (result (gs/call! mgr [:add id handler (act/self)])) :ok) +(defn swap-handler! + "Replace handler old-id, as OTP's swap_handler: its terminate runs with + reason, and (f what-terminate-returned) makes the handler added under + new-id in its place. Returns :ok; throws if old-id is not a handler or + the new one's h-init throws." + [mgr old-id reason new-id f] + (result (gs/call! mgr [:swap old-id reason new-id f nil])) + :ok) + +(defn swap-sup-handler! + "swap-handler!, the new handler tied to the calling actor as + add-sup-handler! ties one." + [mgr old-id reason new-id f] + (result (gs/call! mgr [:swap old-id reason new-id f (act/self)])) + :ok) + (defn delete-handler! "Remove handler id, calling its terminate with reason (default :delete). Returns what terminate returned; throws if there is no such handler." diff --git a/src/ensemble/gen_server.clj b/src/ensemble/gen_server.clj index a52407c..c7ee04e 100644 --- a/src/ensemble/gen_server.clj +++ b/src/ensemble/gen_server.clj @@ -88,12 +88,14 @@ (defn call! "Send request to the server srv (an actor or a registered name) and wait for - its reply, up to timeout-ms (default *call-timeout*; nil waits forever). + its reply, up to timeout-ms (default *call-timeout*; nil or :infinity + waits forever). Throws if the server is not running, exits before replying, or the timeout elapses; the ex-data :reason says which." ([srv request] (call! srv request *call-timeout*)) ([srv request timeout-ms] - (let [target (try (act/resolve-dest srv) + (let [timeout-ms (act/timeout-ms timeout-ms) + target (try (act/resolve-dest srv) (catch Throwable _ (throw (failure :noproc request))))] (cond (= target (act/self)) (throw (failure :calling-self request)) @@ -365,7 +367,8 @@ ([srv] (stop! srv :normal nil)) ([srv reason] (stop! srv reason nil)) ([srv reason timeout-ms] - (let [target (try (act/resolve-dest srv) + (let [timeout-ms (act/timeout-ms timeout-ms) + target (try (act/resolve-dest srv) (catch Throwable _ (throw (failure :noproc [:stop reason])))) p (promise) hook (act/on-exit! target (fn [r] (deliver p r)))] diff --git a/src/ensemble/gen_statem.clj b/src/ensemble/gen_statem.clj index ac22417..0d1976e 100644 --- a/src/ensemble/gen_statem.clj +++ b/src/ensemble/gen_statem.clj @@ -36,7 +36,7 @@ (def reply! gs/reply!) -(defn- now [] (System/currentTimeMillis)) +(defn- now [] (act/now-ms)) (defn- arm "Timers as {key [deadline content]} from the pure timers {key [ms content]} diff --git a/src/ensemble/node.clj b/src/ensemble/node.clj index 2c4f7e6..303c2a7 100644 --- a/src/ensemble/node.clj +++ b/src/ensemble/node.clj @@ -242,6 +242,19 @@ ;; --- frames from a peer ------------------------------------------------- +(defn- spawn-allowed? + "May node n start the fn named sym for a peer? Only what its :spawn + option allows: a set of namespace symbols, or a predicate on sym. With + no :spawn nothing is: a peer is not authenticated yet, and a spawn runs + any fn it names." + [n sym] + (let [allow (:spawn (state n))] + (boolean + (cond + (set? allow) (contains? allow (symbol (namespace sym))) + (ifn? allow) (allow sym) + :else false)))) + (defn handle-frame "Perform, as node n, the operation frame from peer asks for." [n peer frame] @@ -272,14 +285,16 @@ (when-let [a (and id (here id))] (proc/-drop-monitor a ref))) :fired (let [[ref reason] args] (fire! n ref reason)) :alias (let [[alias msg] args] (act/send-alias! alias msg)) - :spawn (let [[ref sym fargs link from] args - f (requiring-resolve sym) - ;; the link is made by the new process before it runs a - ;; step, so a crash at once still reaches the spawner - go (promise) - a (act/spawn (fn [] @go (when link (act/link! from)) (apply f fargs)))] - (send-frame! n peer [:spawned ref a]) - (deliver go true)) + :spawn (let [[ref sym fargs link from] args] + (if-not (spawn-allowed? n sym) + (send-frame! n peer [:spawned ref [:refused sym]]) + (let [f (requiring-resolve sym) + ;; the link is made by the new process before it runs a + ;; step, so a crash at once still reaches the spawner + go (promise) + a (act/spawn (fn [] @go (when link (act/link! from)) (apply f fargs)))] + (send-frame! n peer [:spawned ref a]) + (deliver go true)))) :spawned (let [[ref pid] args] (when-let [p (get-in @nodes [n :pending ref])] (deliver p pid))) nil)))) @@ -315,17 +330,24 @@ (defn start! "Start node name in this VM, reachable over transport. Code that runs outside any actor now runs as this node; with-node runs code as another. + Options: + + :spawn what a peer may start here with spawn-on: a set of namespace + symbols whose fns it may name, or a predicate on the fn's + symbol. Without it a peer may start nothing. + Returns name." - [nm transport] + ([nm transport] (start! nm transport {})) + ([nm transport opts] (when-not (and (keyword? nm) (= nm (edn/read-string (pr-str nm)))) (throw (ex-info (str "a node name is a keyword that reads back as itself: " (pr-str nm)) {:reason :badarg :node nm}))) (swap! nodes assoc nm {:transport transport :conns {} :watching {} :exported {} - :node-mons [] :pending {}}) + :node-mons [] :pending {} :spawn (:spawn opts)}) (act/set-remote-resolver! (fn [[_ remote-nm peer]] (->RemoteName peer remote-nm))) (act/set-remote-alias-sender! (fn [alias msg] (send-frame! (act/node) (:ensemble.actor/owner-node alias) [:alias alias msg]))) - nm) + nm)) (defmacro with-node "Run body as node n: what it spawns runs there, and names resolve there." @@ -359,8 +381,10 @@ (defn spawn-on "Start a process on node peer running (apply f args), f named by a - symbol peer can resolve, as spawn(Node, M, F, A). With {:link true} it - is linked to the current actor. Returns its pid, a RemotePid." + symbol peer can resolve and allows (see start!'s :spawn), as + spawn(Node, M, F, A). With {:link true} it is linked to the current + actor. Returns its pid, a RemotePid; throws {:reason [:not-allowed + sym]} when peer refuses." ([peer f-sym args] (spawn-on peer f-sym args {})) ([peer f-sym args {:keys [link timeout] :or {timeout 5000}}] (let [n (act/node) @@ -369,11 +393,15 @@ (swap! nodes assoc-in [n :pending ref] p) (when-not (send-frame! n peer [:spawn ref f-sym (vec args) (boolean link) (act/self)]) (throw (ex-info "cannot reach node" {:reason :noconnection :node peer}))) - (let [r (deref p timeout ::timeout)] + (let [t (act/timeout-ms timeout) + r (if t (deref p t ::timeout) @p)] (swap! nodes update-in [n :pending] dissoc ref) - (if (= ::timeout r) - (throw (ex-info "spawn-on timed out" {:reason :timeout :node peer})) - r))))) + (cond + (= ::timeout r) (throw (ex-info "spawn-on timed out" {:reason :timeout :node peer})) + (and (vector? r) (= :refused (first r))) + (throw (ex-info "the node does not allow that spawn" + {:reason [:not-allowed (second r)] :node peer})) + :else r))))) ;; --- the loopback transport --------------------------------------------- diff --git a/test/ensemble/actor_test.clj b/test/ensemble/actor_test.clj index 0eb6776..efa9c1b 100644 --- a/test/ensemble/actor_test.clj +++ b/test/ensemble/actor_test.clj @@ -309,3 +309,35 @@ (act/exit! a :shutdown) (act/exit-reason a 1000))))))) (is (every? #{:shutdown} (deref out 60000 [:timeout]))))) + +;; --- :infinity is a timeout everywhere one is taken ------------------------ + +(deftest receive-takes-infinity + (let [a (act/spawn (fn [] (receive [:x :got] [:after :infinity :never])))] + (act/! a :x) + (is (= :got (act/join a 1000))))) + +;; --- a monitor by name ------------------------------------------------------ + +(deftest monitoring-a-name-nobody-holds-is-down-at-once + (let [a (act/spawn (fn [] (let [r (act/monitor! ::nobody)] + (receive [[:DOWN r :process obj reason] [obj reason]] + [:after 1000 :no-down]))))] + (is (= [[:At :nobody (act/node)] :noproc] (act/join a 2000))))) + +(deftest a-monitor-by-name-reports-the-name + (let [t (act/spawn (fn [] (receive [:die (act/exit! :bye)])) {:name ::watched}) + a (act/spawn (fn [] (let [r (act/monitor! ::watched)] + (act/! t :die) + (receive [[:DOWN r :process obj reason] [obj reason]] + [:after 1000 :no-down]))))] + (is (= [[:At :watched (act/node)] :bye] (act/join a 2000))))) + +;; --- refs carry their node -------------------------------------------------- + +(deftest refs-made-on-different-nodes-differ + (let [r1 (binding [act/*node* :one.vm] (act/make-ref)) + r2 (binding [act/*node* :two.vm] (act/make-ref))] + (is (= :one.vm (act/ref-node r1))) + (is (= :two.vm (act/ref-node r2))) + (is (not= (assoc r1 :ensemble.actor/ref 0) (assoc r2 :ensemble.actor/ref 0))))) diff --git a/test/ensemble/gen_event_test.clj b/test/ensemble/gen_event_test.clj index e5a844b..0d09f1b 100644 --- a/test/ensemble/gen_event_test.clj +++ b/test/ensemble/gen_event_test.clj @@ -1,5 +1,6 @@ (ns ensemble.gen-event-test - (:require [clojure.test :refer [deftest is]] + (:require [clojure.core.async :as a] + [clojure.test :refer [deftest is]] [ensemble.actor :as act :refer [receive]] [ensemble.gen-event :as ev])) @@ -101,3 +102,65 @@ m (ev/start {:handlers [[:a (->Sink l)]]})] (ev/stop! m) (is (= [[:terminated :stop]] @l)))) + +(defn- eventually [pred] + (loop [i 0] (cond (pred) true (> i 200) false :else (do (a/Infoed log [:from res]))] + :sleep [:ok st :hibernate] + (do (swap! log conj [tag e]) [:ok st]))) + (h-handle-call [_ _ st] [:ok st st]) + (h-terminate [_ reason st] (swap! log conj [:terminated tag reason]) [:handed st]) + ev/InfoHandler + (h-handle-info [_ msg st] (swap! log conj [:info tag msg]) [:ok st])) + +(deftest other-messages-go-to-handle-info + (let [l (atom []) + m (ev/start)] + (ev/add-handler! m :a (->Infoed l :a)) + (ev/add-handler! m :b (->Sink (atom []))) + (act/! m :hello) + (is (eventually #(= [[:info :a :hello]] @l))) + (is (= [:a :b] (ev/which-handlers m))))) + +(deftest swap-handler-hands-terminate-result-to-the-new-one + (let [l (atom []) + m (ev/start)] + (ev/add-handler! m :old (->Infoed l :old)) + (is (= :ok (ev/swap-handler! m :old :bye :new (fn [res] (->Infoed l [:got res]))))) + (is (= [:new] (ev/which-handlers m))) + (is (= [:got [:handed :old]] (ev/call! m :new :state))) + (is (= [[:terminated :old :bye]] @l)))) + +(deftest a-handler-may-swap-itself + (let [l (atom []) + m (ev/start)] + (ev/add-handler! m :old (->Infoed l :old)) + (ev/sync-notify! m :swap) + (is (= [:next] (ev/which-handlers m))) + (is (= [:from [:handed :old]] (ev/call! m :next :state))))) + +(deftest a-handler-may-hibernate-the-manager + (let [m (ev/start)] + (ev/add-handler! m :a (->Infoed (atom []) :a)) + (ev/notify m :sleep) + (is (eventually #(act/hibernating? m))) + (is (= :a (ev/call! m :a :state))))) + +(deftest a-supervised-handler-going-unlinks-its-owner + (let [m (ev/start) + owner (act/spawn (fn [] + (ev/add-sup-handler! m :a (->Sink (atom []))) + (ev/delete-handler! m :a) + ;; linked still, the manager's death would kill it + (receive [:check (do (act/exit! m :kill) (receive [_ nil] [:after 100 nil]) :alive)])))] + (a/SlowInit 5000) {:timeout 50}) (catch Throwable e (:reason (ex-data e))))))] (is (= :timeout (act/join c 2000))))) + +(deftest call-and-stop-take-infinity + (let [s (gs/start (->SlowInit 0) {:timeout :infinity})] + (is (= :up (gs/call! s :get :infinity))) + (is (= :up (act/join (act/spawn (fn [] (gs/call! s :get :infinity))) 1000))) + (is (= :ok (gs/stop! s :normal :infinity))))) + +(deftest wait-response-takes-infinity + (let [s (gs/start (->SlowInit 0)) + a (act/spawn (fn [] (gs/wait-response (gs/send-request s :get) :infinity)))] + (is (= [:reply :up] (act/join a 1000))))) diff --git a/test/ensemble/node_test.clj b/test/ensemble/node_test.clj index 4e1ce8d..ef942d8 100644 --- a/test/ensemble/node_test.clj +++ b/test/ensemble/node_test.clj @@ -20,8 +20,8 @@ [] (let [a (keyword (str (gensym "a") ".vm")) b (keyword (str (gensym "b") ".vm"))] - (node/start! a (node/loopback)) - (node/start! b (node/loopback)) + (node/start! a (node/loopback) {:spawn #{'ensemble.node-test}}) + (node/start! b (node/loopback) {:spawn #{'ensemble.node-test}}) (node/with-node a (node/connect! b)) [a b])) @@ -150,3 +150,14 @@ (is (= 2 (deref p 3000 nil)))) (is (= 5 (node/with-node a (gs/call! (node/with-node b (act/whereis :counter)) [:add 3]))) "a local handle works too") (is (= 5 (node/with-node a (gs/call! [:At :counter b] [:get]))))))) + +(deftest a-node-spawns-only-what-it-allows + (let [a (keyword (str (gensym "a") ".vm")) + b (keyword (str (gensym "b") ".vm"))] + (node/start! a (node/loopback)) + (node/start! b (node/loopback) {:spawn (fn [sym] (= sym `idle))}) + (is (= [:not-allowed `echo] + (try (node/with-node a (node/spawn-on b `echo [])) (catch Throwable e (:reason (ex-data e)))))) + (is (= [:not-allowed `echo] + (try (node/with-node b (node/spawn-on a `echo [])) (catch Throwable e (:reason (ex-data e)))))) + (is (act/pid? (node/with-node a (node/spawn-on b `idle []))))))