Clojurians Log v2
missionary
hi all! i've spent quite a bit of time getting into missionary recently and want to incorporate it into a project i'm working on. it's my first time posting so apologies if it's too long; let me know if you want it broken up somehow. i'm interested in your thoughts on how to go about modeling the following situation.
my application is a plugin for a certain game client. interacting with the client is very similar to interacting with java swing and its edt. the client's main loop runs in the so called "client thread" and dispatches events to registered handlers when things happen. the client is very stateful in nature and you have to make sure to access its state only from the client thread. there is also a mechanism to schedule arbitrary runnables for execution on the client thread, just like swing's invokeLater.
concurrent with (and unrelated to) the client is a messaging loop that is used as a two-way communication mechanism between the application and the user. the application uses it to report information to the user, and the user uses it to issue commands to the application. the style of interaction is very similar to the client's: messaging events are dispatched by a loop one after the other on a dedicated thread.
i want to implement a "module" (a process, really) that has to react to events from the client, the messaging loop and possibly other sources. it has to track some state and react to the events in order. essentially what i want is almost like an actor, except that i don't want it to be asynchronous and have its own unbounded mailbox. what is left is a plain old state machine, powered by the events and updated synchronously as they're dispatched (i.e. updated on the dispatching thread). this synchronicity is important due to the synchronization requirements of the client's state, etc.
so far the above doesn't sound too hard. m/observe can be used to construct discrete flows for the events, which can then be combined into a single interleaved flow with a utility such as select below. the state machine itself can be modeled similarly to what is done for actors in the docstring of m/mbx: a series of "behaviors" (plain clojure functions) that take an event to handle and return the behavior to use next. below i model the machine as a flow of states using m/reductions and use m/reduce to run it.
(defn select [& flows]
(m/ap
(loop [[f & fs] flows]
(if f
(m/amb= (m/?> f) (recur fs))
(m/amb)))))(defn machine-flow [init events]
(m/reductions (fn [[_ f] event] [event (f event)]) [nil init] events))
(defn machine-runner [init & flows]
(m/reduce {} nil (machine-flow init (apply select flows))))
(defn foo-machine []
(letfn [(behavior-1 [event] (do (foo event) behavior-2))
(behavior-2 [event] (do (bar event) behavior-1))]
behavior-1))
(def foo-proc
((machine-runner foo-machine flow1 ... flown)
#(prn :ok %)
#(prn :ko %))m/sleep, are we guaranteed to be running on missionary's sleep scheduler thread? or for example, if we fork on a flow and a transfer happens from a specific thread (say the client thread from above), are we guaranteed to stay on it until the next park?
(defn continually [task]
(m/ap (loop [] (m/amb (m/? task) (recur)))))
(defn executor [mbx]
;; NOTE: Error handling omitted.
(m/ap (m/? (m/?> ##Inf (continually mbx)))))
1. I would probably go for an imperative solution e.g. with local mutable state inside an ap block used to store the current behavior. The only functional solution I can imagine would involve a cycle to reinject the asynchronously computed state in the input flow of events. Example :
(defn poll [task]
(m/ap (m/? (m/?> (m/seed (repeat task))))))
(defn feedback [f & args]
(m/ap
(let [rdv (m/rdv)]
(m/? (rdv (m/?> (apply f (poll rdv) args)))))))
(defn async-state-machine [behaviors init events]
(m/ap
(m/amb init
(m/? (m/?> (m/zip #(%1 %2) behaviors events))))))
(def asm
(m/reduce {} nil
(feedback async-state-machine
(letfn [(init [event]
(m/sp
(prn :behavior 'init :event event)
(case event
:foo (m/? (m/sleep 1 step)))))
(step [event]
(m/sp
(prn :behavior 'step :event event)
(case event
:bar (m/? (m/sleep 1 done))
step)))
(done [event]
(m/sp
(prn :behavior 'done :event event)
done))]
init)
(m/seed [:foo :bar :baz]))))
(asm prn prn)
:behavior init :event :foo
:behavior step :event :bar
:behavior done :event :baz2. A top-level mailbox-based executor would work, but it would be cleaner for the application itself to be an input flow of the executor.
3. Yes, guaranteed. Everything is synchronous by default, thread switching must be explicit
meta : Welcome ! Your post is not too long, you provide useful context, you state what you've tried, you ask good questions. I'm happy to help
2.
;; ok - executor and app are siblings of a parent supervisor, they communicate via a side-channel
(m/ap
(let [m (m/mbx)]
(m/?> (m/amb= (executor m) (app m)))))
;; better - app is supervised by executor, they communicate via the flow protocol
(executor (app))then as a consequence i assume there's no 100% sure way to tell whether a process terminated spontaneously or by cancellation? i've had a need for this at the top-level where it would be useful to report spontaneous failures of tasks that shouldn't terminate spontaneously but do due to a bug. something like:
(task
#(log/error "Bug" %)
#(if (cancelled? %) (log/trace "Finished" %) (log/error "Bug" %)))(defn cancelled? [e]
(or (instance? Cancelled e)
(and (instance? ExceptionInfo e)
(let [errors (get (ex-data e) ::m/errors ::none)]
(and (not= errors ::none)
(every? cancelled? errors))))))ExceptionInfo case is to take care of m/join and m/race , but i realize the whole thing is not very robust to begin with. is there a better way to differentiate spontaneous from non-spontaneous termination?also, does cancellation being a best-effort operation make it moot to talk about "termination due to a cancellation request"? it seems like the only notion to work with is whether you terminated before (aka spontaneously) or after (non-spontaneously?) cancellation was requested, because even when cancelled the process might still terminate successfully or with a failure unrelated to cancellation? in other words, there's no guarantee that it was the cancellation that triggered the termination, right? regardless, i'd still be fine with differentiating spontaneous from non-spontaneous termination and just assuming the latter was the result of a cancellation for the purpose of the above question
that's right, none of these questions have a general answer
in my experience, the only place where it matters to discriminate spontaneous termination is to recover from a switch in ap - but even this is a historic design mistake that will eventually be fixed
i see, interesting. what would then be a good way to implement the kind of error reporting described above, where i only log spontaneous termination? what are the pitfalls of an approach like this:
(defn log-spontaneous [task]
(fn [s! f!]
(let [cancelled? (atom false)
cancel (task #(do (when-not @cancelled?
(log/error "Bug!"))
(s! %))
#(do (when-not @cancelled?
(log/error "Bug!"))
(f! %)))]
#(do (reset! cancelled? true)
(cancel)))))which kind of process is in charge of cancelling and why does it care about post-cancellation events ?
this particular task is a top-level one and is stored in a mount state for easier management, like this:
(defstate my-state
:start (task #(...) #(...))
:stop (@my-state))> if i'm not mistaken i believe that makes (app) a child of (executor) , right?
yes, but I'm a bit confused about how you implemented executor. If the goal is to run a succession of tasks sequentially, and the first task is (app) which supposedly doesn't terminate spontaneously, won't the remaining tasks be waiting forever ?
My suggestion was to get rid of the mbx and implement app not as a task pushing events to the mailbox, but as a flow emitting these events. It is generally cleaner, but not necessarily a good fit for your use case - which I don't understand well enough
ah, the executor runs all of the tasks concurrently. the implementation is this one from my initial message:
(defn continually [task]
(m/ap (loop [] (m/amb (m/? task) (recur)))))
(defn executor [mbx]
;; NOTE: Error handling omitted.
(m/ap (m/? (m/?> ##Inf (continually mbx)))))m/join wouldn't do as i don't know in advance how many tasks will be startedside question: what exactly does it mean for a task to "terminate spontaneously"? i've seen this terminology in the docstrings as well but i'm not sure if i fully understand it
it means to terminate before getting cancelled
is that synonymous with calling the success continuation, or the failure continuation but not with with a missionary.Cancelled ?
by convention a cancelled process terminates as failure with missionary.Cancelled but it's not enforced, e.g. a cancelled m/via will just propagate cancellation as an interruption of the running thread, the result ultimately depends on how the current call chain reacts to this signal (`InterruptedException` is another common convention on the JVM, but it's not enforced either)
hi @leonoel, it's been a little while but i've come back to this once again! regarding (2), are you suggesting that (app) return a flow of tasks to execute, one of which would be the main machine task? if so, how would i construct this task flow (which is independent of the machine's state flow) so that i can "push" tasks to it from the machine's handlers? would i use an (rdv) or an (mbx) that's incorporated into the flow (e.g. via select) and pass the port into the machine task, or did you envision a different kind of structure?
the way i launch (app) is by passing it as a task to the executor, something like: (m/sp (mbx (app mbx)) (m/? (executor mbx))) . if i'm not mistaken i believe that makes (app) a child of (executor) , right? is there a meaningful difference between that approach and the one you suggested above? won't they both end up having to use some sort of port to communicate and spawn additional tasks?
i'm also very much still stuck on (1). any comments/thoughts/suggestions are very appreciated, though i realize the description is non-trivial. if you ever find the time i'm more than happy to discuss it and/or try to explain it in a better way. thanks again for creating missionary!
hi leo, thanks for the kind words and the fast reply!
1. nice! that particular approach of supporting missionary tasks as behaviors is what i went for as well, except that i used a volatile for my state. and indeed, i also cannot think of a way that doesn't involve a cycle.
(defn machine [f events]
(m/ap
(m/amb f (let [cell (volatile! f)
event (m/?> events)]
(vreset! cell (m/? (@cell event)))))))
;;; Amusing sidenote: the version below is incorrect because the continuation
;;; of m/?> does *not* include the evalution of the operator position @cell.
(defn machine [f events]
(m/ap
(m/amb f (let [cell (volatile! f)]
(vreset! cell (m/? (@cell (m/?> events))))))))m/sleep parks will cause a temporary block, and the problem only gets worse if the parks take longer. is there something architectural that can be changed to fix this? it might be impossible to say without knowing more about the specific interactions between the jobs and the events, so let me give a few example use cases:
• after receiving an event e1, the machine launches a task t and transitions to a behavior that continues handling events by ignoring them until t terminates. once t terminates, the machine transitions to some new behavior based on its result.
• after receiving an event e2, the machine launches a task t and transitions to behavior b that ignores all but a select few events, say e3 and e4, until t terminates. if t terminates, the machine transitions to some new behavior based on its result. if e3 is received, t is cancelled and the machine transitions to some new behavior. if e4 is received, t is cancelled and launched again, and the machine loops back to behavior b.
in my current implementation where behaviors are still plain clojure functions, i've solved the above cases by running t on the aforementioned mailbox-based executor, but wrapped so that its result is posted as an event to the machine's event flow (via an rdv, i.e. a (poll rdv) flow that is an argument to the top-level select flow). every child task has to be wrapped in a similar way so that its success or failure produces an event that the machine can consume. it quickly turns into a bit of a message-passing mess though and reminds me of erlang-style async request-reply patterns. i wonder if it's possible to restructure things or introduce some combinators to make such patterns more elegant/obvious/"inline"? perhaps some special way of parking within a behavior that would still continue to handle events so as not to block the application? any ideas?2. i don't fully understand what you mean here. can you elaborate on "for the application itself to be an input flow of the executor"?
3. speaking of, i've got a bit of a philosophical dilemma that i've been struggling with for a long while and would love to hear your thoughts about. after lots of confusion and digging i've come to realize that people use "synchronous" in two distinct and seemingly contradictory ways, as can be seen even on https://en.wiktionary.org/wiki/synchronous.
the first definition is the idea of events happening at the same time, i.e. simultaneously. you use this definition at https://github.com/leonoel/missionary-website/blob/master/assets/topics/core-principles.md https://youtu.be/xtTCdT6e9-0?t=1179 and to me this is the more "traditional" and "mathematical" one, which i like.
the second definition is the more "modern" "computing" one, which states that two events are synchronous if one blocks the other, i.e. if one happens after the other. this seems to be the prevalent use of the word these days, especially by lots of libraries in relation to io.
are these two definitions not in complete opposition to one another? or are they describing the same idea but from different angles? have you ever thought about this confusing state of affairs and found a good way to deal with it? do you accept that the word means different things to different people and use whichever definition based on the context, or have you found a way to reconcile the two somehow?
another issue is the frequent use of the word as a unary relation rather than a binary one. people will often write that "an event is synchronous" when they really ought to write that "an event is synchronous with another event". the latter event is often left implicit and in many instances even varies depending on the context, leading to more confusion. i was very pleasantly surprised when i found out that you carefully documented the synchronicity requirements of each operator (e.g. https://github.com/leonoel/missionary-website/blob/master/assets/topics/api/missionary.core/eduction.md), treating it as a proper binary relation and explicitly specifying both events. major props for the rigor, it is deeply satisfying! :)
this got me wondering about the guarantees that missionary provides, especially for its discrete flows. say i've got a (select f1 f2) flow where both f1 and f2 are m/observes that require registration. following the chain of synchronicity requirements of the involved operators, is it correct to conclude that the spawn of the select flow is synchronous with the registration of both m/observes? is it also correct that the select flow will transfer only after both m/observes have been registered, i.e. that it's impossible to receive a value while the process is in a "partially registered" state? essentially that no values can flow until the whole pipeline has been constructed, so to speak?
3. I agree, the second definition of synchronous is confusing and problematic. I can understand where it comes from, i.e. the rest of the computation is "synchronized" with the result of the (asynchronous) computation, but it's still abusive to describe the computation as synchronous when it's just parking/blocking the thread.
Regarding (select f1 f2) yes the composition model guarantees both registrations are synchronous with the spawn event, and no transfer can happen before the spawn event terminates. However, a process is allowed to transfer an input immediately after spawning it, during its own spawning event. It is generally not the case but e.g. m/ap does that, so if an m/observe is immediately ready to transfer after spawn then m/ap will indeed transfer it then resume its evaluation until the next parking point or the end of body.