Clojurians Log v2
missionary 2024-10-07
I am using (m/ap (m/?> (m/?> (count flows) (m/seed flows)))) to mix multiple flows into one flow which I then use to update state. This works well when the flows send events with huge time delays (100ms - 2000ms). Now I am trying to use this mixing function in a simulation that is effecrively synchronous, I have the problem that the different threads in the mixing flow work lead to my playback events cpming in in an unordered way. I Tried to use m/zip which basically would allows me to always read the next value from each flow. But my issue is that the different flows have different "length". What I would like to make is to transform all input flows so that when I pass them all to m/zip I get nil for flows that don't have value at the moment and the value of the flow if it is available. But I have no idea how. Thanks for listening!
Been studying Leo's answer. So the solution is the composition of two helper functions: Our well-known mix function and a new time-buffered function.
(defn mix [& flows] (m/ap (m/?> (m/?> (count flows) (m/seed flows)))))
(defn time-buffered [duration-ms flow]
(m/ap
(let [restartable (second (m/?> (m/group-by {} flow)))]
(m/? (->> (m/ap (m/amb= (m/?> restartable)
(m/? (m/sleep duration-ms ::end))))
(m/eduction (take-while #(not= % ::end)))
(m/reduce conj))))))
;; consume batch of values every 50 ms
((m/reduce conj (time-buffered 50 (mix producer-slow producer-fast)))
prn prn)@kawas great rewrite! Consuming the batch with reduce alone will block forever;| This works: ((->> (time-buffered 50 (mix producer-slow producer-fast)) (m/eduction (take 20)) (m/reduce conj)) prn prn)
So I think I have somehow of an idea how it could work. So I need to process flow a and flow b to update internal state. I want to read the flows in such a way that for each value of flow a I will also get all new values of flow b. So my idea is this (m/zip vector a (m/relieve conj b)). So in a way this should work to generate partitions in time. Not sure if it will work when I test it.
I guess for it to work flow b must at least have one value each time there is a value of a.
I did it!!
I updateted the code. There are still some issues.
Have you considered this pattern ? Not sure how it fits to your use case
(defn data-producer [max-delay-ms]
(m/ap
(loop [i 0]
(m/amb
(m/? (m/sleep (rand-int max-delay-ms) i))
(recur (inc i))))))
(def producer-slow (data-producer 200))
(def producer-fast (data-producer 30))
((->> (m/ap
(let [[_ batch] (m/?> (m/group-by {} (m/ap (m/?> (m/amb= producer-slow producer-fast)))))]
(m/? (->> (m/ap (m/amb= (m/?> batch)
(m/? (m/sleep 50))))
(m/eduction (take-while some?))
(m/reduce conj)))))
(m/eduction (take 20))
(m/reduce conj))
prn prn)Thanks @leonoel No I have not! Very interesting! I need to study it in detail. Need to understand m/amb= Has this pattern been mentioned somewhere? I really tried my best to find all available docs, but I didnt see it anywhere. Many many thanks!
@leonoel It solves 100% my issue. Absolutely Brilliant! Thank you. Now I "just" need to understand what is going on really. ap structured inside ap. How crazy is this? Amazing how short this code is. Love it!!!
m/amb= is just sugar over m/?>, you can look the source
the essence of the pattern is m/group-by, which splits a flow over space (i.e. partition the value set according to a key function) and time (i.e. cancelling a group consumer starts a new batch). In this case we're not interested in the space partition so we use {} as the key function, i.e. all values have the same key. The time partition is defined by the m/eduction stage which early terminates when the m/sleep result is observed