std.concurrent.atom
batched atom queues and trackable asynchronous hubs
Use atom-backed queues when producers should submit quickly and a single executor should process entries in controlled batches.
1 Overview
std.concurrent.atom provides two related patterns. aq:* functions maintain a simple vector-backed queue, while hub:* functions attach a future ticket to each batch so callers can observe completion.
2 Walkthrough
2.1 Create a batched submitter
Create an executor with a target, a bulk handler, a maximum batch size, and a short collection interval. The returned map contains the queue, executor, and submission function.
(require '[std.concurrent.atom :as atom])
(def worker
(atom/aq:executor
{:target :events
:handler (fn [target entries]
{:target target :entries entries})
:max-batch 100
:interval 25}))
((:submit worker) {:id 1} {:id 2})2.2 Wait for a tracked batch
Use a hub executor when the caller needs a future representing the batch result. hub:wait blocks only while queued work remains.
(def tracked
(atom/hub:executor
nil
{:handler (fn [_ entries] entries)
:max-batch 100
:interval 25}))
(let [[ticket start count submitted?]
((:submit tracked) :a :b :c)]
{:result @ticket
:range [start count]
:submitted? submitted?})
(atom/hub:wait (:queue tracked))3 API
- aq:executor
- aq:new
- aq:process
- aq:submit
- hub-state
- hub:add-entries
- hub:executor
- hub:new
- hub:process
- hub:submit
- hub:wait
v 3.0
(defn aq:executor
([{:keys [target] :as opts}]
(let [queue (aq:new)
executor (exe/executor:single 1)
submit (aq:submit executor queue opts)]
{:queue queue
:executor executor
:submit submit
:options opts})))link
(let [-res- (atom []) -aq- (aq:executor {:handler (fn [_ items] (swap! -res- concat items)) :interval 10 :max-batch 10}) -exe- (:executor -aq-)] (try ((:submit -aq-) 1 2 3) (Thread/sleep 100) @-res- (finally (executor/exec:shutdown -exe-)))) => [1 2 3]
v 3.0
(defn aq:process
([f aq max-batch]
(loop []
(let [queue (at/swap-return! aq
(fn [queue] [queue []]))]
(->> (partition-all max-batch queue)
(mapv f))))))link
(let [+state+ (atom [])] (aq:process (fn [elems] (swap! +state+ conj elems)) (atom [1 2 3 4 5]) 3) @+state+) => [[1 2 3] [4 5]]
aq:submit ^
[executor queue {:keys [target handler max-batch interval], :as opts}]
Added 3.0
submission function for one or multiple entries to aq
v 3.0
(defn aq:submit
([executor queue {:keys [target handler max-batch interval] :as opts}]
(let [bulk-fn (fn []
(aq:process (fn [items]
(-> (handler target items)
(env/explode)))
queue
max-batch))]
(fn [& entries]
(do (apply swap! queue conj entries)
(exe/submit-notify executor
bulk-fn
interval)
nil)))))link
(let [-out- (atom []) -exe- (executor/executor:single) -q- (atom []) -sub- (aq:submit -exe- -q- {:handler (fn [_ items] (swap! -out- concat items)) :interval 10 :max-batch 10})] (try (-sub- 1 2 3) (Thread/sleep 100) @-out- (finally (executor/exec:shutdown -exe-)))) => [1 2 3]
v 4.0
(defn hub-state
[elems]
{:ticket (f/incomplete)
:queue elems})link
(hub-state [1 2 3]) => (contains {:ticket f/future? :queue [1 2 3]})
v 3.0
(defn hub:add-entries
([hub entries]
(at/swap-return! hub
(fn [{:keys [ticket queue] :as m}]
(let [start (count queue)]
[[ticket start (count entries)] {:ticket ticket
:queue (into queue entries)}])))))link
(hub:add-entries (hub:new) [1 2 3 4 5]) => (contains [f/future? 0 5])
hub:executor ^
[object {:keys [handler max-batch interval], :as opts}]
Added 3.0
creates a hub based executor
v 3.0
(defn hub:executor
([object {:keys [handler max-batch interval]
:as opts}]
(let [queue (hub:new)
executor (exe/executor:single 1)
submit (hub:submit object executor queue opts)]
{:queue queue
:executor executor
:submit submit})))link
(let [-exe- (hub:executor nil {:handler (fn [& args] args) :interval 50 :max-batch 1000}) -exec- (:executor -exe-)] (try (do (def -res- ((:submit -exe-) 1 2 3 4 5 6)) @(first -res-)) => '[(nil (1 2 3 4 5 6))] (finally (executor/exec:shutdown -exec-))))
v 3.0
(defn hub:new
([]
(hub:new []))
([elems]
(atom (hub-state elems))))link
@(hub:new [1 2 3]) => (contains {:ticket f/future? :queue [1 2 3]})
v 3.0
(defn hub:process
([f hub max-batch]
(let [[ticket queue] (at/swap-return! hub
(fn [{:keys [ticket queue]}]
[[ticket queue] (hub-state [])]))]
(->> (partition-all max-batch queue)
(mapv f)
(f/future:force ticket)))))link
(let [+state+ (atom [])] (hub:process (fn [elems] (swap! +state+ conj elems)) (hub:new [1 2 3 4 5]) 3) @+state+) => [[1 2 3] [4 5]]
hub:submit ^
[object executor hub {:keys [handler max-batch interval]}]
Added 3.0
submission function for the hub
v 3.0
(defn hub:submit
([object executor hub {:keys [handler max-batch interval]}]
(let [bulk-fn (fn [] (hub:process (fn [items]
(handler object items))
hub
max-batch))]
(fn [& entries]
(let [[ticket start count] (hub:add-entries hub entries)
hit (exe/submit-notify executor
bulk-fn
interval)]
[ticket start count hit])))))link
(let [-out- (atom []) -exe- (executor/executor:single) -hub- (hub:new) -sub- (hub:submit nil -exe- -hub- {:handler (fn [_ items] (swap! -out- concat items) items) :interval 10 :max-batch 10})] (try (let [[ticket] (-sub- 1 2 3)] @ticket) @-out- (finally (executor/exec:shutdown -exe-)))) => [1 2 3]