std.concurrent.pool

bounded reusable resource pools with lifecycle and health checks

Manage expensive reusable objects with acquisition tracking, idle cleanup, disposal, and component lifecycle support.

Concurrency overview

1    Overview

A pool separates resource construction from resource use. It tracks idle and busy objects, enforces a maximum, records utilization, and can dispose unhealthy or long-idle resources in the background.

2    Walkthrough

2.1    Create a pool

(require '[std.concurrent.pool :as pool])

(def connections
  (pool/pool
   {:size 2
    :max 8
    :keep-alive 30000
    :poll 5000
    :resource {:create (fn [] {:opened-at (System/currentTimeMillis)})
               :start identity
               :stop  (fn [_] nil)
               :health (fn [_] {:status :ok})
               :initial 0.5}}))

2.2    Acquire and release safely

Use pool:with-resource for normal work. It releases the object even when the body throws. Lower-level code can call pool:acquire and pool:release directly.

(pool/pool:with-resource [connection connections]
  (:opened-at connection))

(let [[id connection] (pool/pool:acquire connections)]
  (try
    connection
    (finally
      (pool/pool:release connections id))))

2.3    Inspect and dispose

(pool/pool:info connections)
(pool/pool:resources:busy connections)
(pool/pool:resources:idle connections)
(pool/pool:cleanup connections)
(pool/pool:stop connections)

Call pool:dispose:mark inside a pooled operation when the current resource should be discarded rather than returned to idle storage.

3    API

*current* ^

NONE
(def ^:dynamic *current* nil)
link

*defaults* ^

NONE
(def ^:dynamic *defaults*)
link

*dispose* ^

NONE
(def ^:dynamic *dispose* nil)
link

*stop-fn* ^

NONE
(def ^:dynamic *stop-fn* e/exec:shutdown-now)
link

+opts+ ^

NONE
(def +opts+
  [:size :max :keep-alive :poll])
link

->Pool ^

[size max state]

NONE
(impl/defimpl Pool [size max state]
  :prefix "pool:"
  :string pool-string
  :protocols [std.protocol.track/ITrack
              protocol.component/IComponent
              :body {-remote?  false}])
link

->PoolResource ^

[id thread-id object status create-time update-time busy used]

NONE
(impl/defimpl PoolResource [id thread-id object status create-time update-time busy used]
  :stream resource-string)
link

Pool ^

NONE
(impl/defimpl Pool [size max state]
  :prefix "pool:"
  :string pool-string
  :protocols [std.protocol.track/ITrack
              protocol.component/IComponent
              :body {-remote?  false}])
link

PoolResource ^

NONE
(impl/defimpl PoolResource [id thread-id object status create-time update-time busy used]
  :stream resource-string)
link

dispose-fn ^

[m id] [m id stop-fn]
Added 3.0

helper function for `dispose` and `cleanup`

v 3.0
(defn dispose-fn
  ([m id]
   (dispose-fn m id nil))
  ([m id stop-fn]
   (let [update-fn   (fn [m id resource]
                       (let [current (time/time-ns)
                             {:keys [create-time busy]} resource]
                         (-> m
                             (update :idle dissoc id)
                             (update-in [:stats :total-time] + (- current create-time))
                             (update-in [:stats :busy-time] + busy)
                             (update-in [:stats :dispose] inc))))
         {:keys [idle options]} m
         resource (get idle id)
         stop-fn (or stop-fn identity)]
     (if resource
       (let [_ (stop-fn (:object resource))]
         (update-fn m id resource))
       m))))
link
(dispose-fn {:idle {"hello" {:create-time 0
                             :busy 0.0
                             :object :resource}}
             :stats {:total-time 0.0
                     :busy-time 0.0
                     :dispose 0}}
            "hello"
            (fn [_] :stopped))
=> (contains-in {:idle {}
                 :stats {:dispose 1}})

map->Pool ^

[m__7997__auto__]

NONE
(impl/defimpl Pool [size max state]
  :prefix "pool:"
  :string pool-string
  :protocols [std.protocol.track/ITrack
              protocol.component/IComponent
              :body {-remote?  false}])
link

map->PoolResource ^

[m__7997__auto__]

NONE
(impl/defimpl PoolResource [id thread-id object status create-time update-time busy used]
  :stream resource-string)
link

pool ^

[{:keys [size max keep-alive poll resource], :as m}]
Added 3.0

creates and starts the pool

v 3.0
(defn pool
  ([{:keys [size max keep-alive poll resource] :as m}]
   (-> (pool:create m)
       (pool:start))))
link
(component/with [p (pool {:size 2
                            :max 10
                            :keep-alive 10000
                            :poll 20000
                            :resource {:create (fn [] (rand))
                                       :initial 0.3
                                       :thread-local true}})]
  p => pool?)

pool-handler ^

[{:keys [state], :as pool}]
Added 3.0

creates a handler loop for cleanup

v 3.0
(defn pool-handler
  ([{:keys [state] :as pool}]
   (fn pool-handler-fn []
     (let [{:keys [executor options]} @state]
       (pool:cleanup pool)
       (Thread/sleep (long (:poll options)))
       (if executor
         (e/submit executor pool-handler-fn))))))
link
(pool-handler {:state (atom {:running true})
               :cleanup (fn [])
               :poll 100})
=> fn?

pool-props-fn ^

[key]

NONE
(defn- pool-props-fn
  ([key]
   (fn
     ([{:keys [state]}]
      (get-in @state [:options key]))
     ([{:keys [state]} input]
      (swap! state assoc-in [:options key] input)))))
link

pool-resource ^

[id {:keys [resource], :as pool}]
Added 3.0

creates a pool resource

v 3.0
(defn pool-resource
  ([id {:keys [resource] :as pool}]
   (let [{:keys [create start]
          :or {start identity}} resource
         current (time/time-ns)]
     (PoolResource. id
                    nil
                    (-> (create)
                        (start))
                    :idle
                    current
                    current
                    0.0
                    0))))
link
(pool-resource "hello"
               {:resource {:create (fn [] ')}})
=> (contains {:id "hello",
              :object '
              :status :idle,
              :create-time number?
              :update-time number?
              :busy 0.0, :used 0})

pool-string ^

[{:keys [tag], :as pool}]

NONE
(defn- pool-string
  ([{:keys [tag] :as pool}]
   (let [str-fn (comp time/format-ns long)]
     (str "#" (or tag "pool") " "
          (-> (pool:info pool)
              (update :resource dissoc :total :busy)
              (update-in [:resource :duration] str-fn))))))
link

pool:acquire ^

[{:keys [state resource], :as pool}]
Added 3.0

acquires a resource from the pool

v 3.0
(defn pool:acquire
  ([{:keys [state resource] :as pool}]
   (let [{:keys [thread-local]} resource
         resource-fn (fn [resource thread-id]
                       (-> resource
                           (update :used inc)
                           (assoc  :update-time (time/time-ns)
                                   :status :busy
                                   :thread-id thread-id)))
         update-fn   (fn [m thread-id id resource]
                       (-> m
                           (update :busy assoc id
                                   (resource-fn resource thread-id))
                           (update-in [:stats :acquire] inc)
                           (update-in [:lookup thread-id]
                                      (fnil #(conj % id) #{}))))
         acquire-fn  (fn [{:keys [busy idle options lookup] :as m}]
                       (let [{:keys [max]} options
                             [id resource] (first idle)
                             thread-id (t/thread:id)
                             local-id (first (get lookup thread-id))]
                         (cond (and thread-local local-id)
                               [[:success local-id
                                 (f/completed (:object (get busy local-id)))]
                                m]

                               id
                               [[:success id
                                 (f/completed (:object resource))]
                                (-> (update-fn m thread-id id resource)
                                    (update :idle dissoc id))]

                               (and max
                                    (<= max
                                        (+ (count busy)
                                           (count idle))))
                               [[:reject nil (ex-info "Maximum limit reached" {:max max})]
                                (update-in m [:stats :reject] inc)]

                               :else
                               [[:success (std.lib.foundation/sid) (f/incomplete)]
                                (update-in m [:stats :create] inc)])))
         [status id boxed] (atom/swap-return! state acquire-fn)
         allocate-fn (fn [state id boxed]
                       (f/fulfil boxed
                                 (fn []
                                   (let [resource (pool-resource id pool)]
                                     (swap! state (fn [m] (update-fn m (t/thread:id) id resource)))
                                     (:object resource)))))
         object  (if (f/future? boxed)
                   (do (if (not (f/future:complete? boxed))
                         (allocate-fn state id boxed))
                       @boxed)
                   boxed)]
     (case status
       :success [id object]
       :reject  (throw object)))))
link
(component/with [|pool| (create-pool)]
  (pool:acquire |pool|))
=> (contains [string? '])

pool:cleanup ^

[{:keys [state], :as pool}]
Added 3.0

runs cleanup on the pool

v 3.0
(defn pool:cleanup
  ([{:keys [state] :as pool}]
   (let [current (time/time-ns)
         {:keys [stop]} (:resource pool)
         {:keys [health] :or {health (constantly {:status :ok})}} (:resource pool)
         {:keys [idle]} @state
         dead-ids (keep (fn [[id resource]]
                          (if-not (-> (health resource)
                                      (:status)
                                      (= :ok))
                            id))
                        idle)
         dispose (fn [m id] (dispose-fn m id stop))
         _    (swap! state (fn [m] (reduce dispose  m dead-ids)))
         cleanup-fn  (fn [{:keys [idle options] :as m}]
                       (let [{:keys [size keep-alive]} options
                             over (- (count idle) size)
                             [ids m] (cond (pos? over)
                                           (let [ids (->> idle
                                                          (keep (fn [[id {:keys [update-time]}]]
                                                                  (if (< (+ update-time keep-alive)
                                                                         current)
                                                                    id)))
                                                          (take over))]
                                             [ids (reduce dispose m ids)])

                                           :else
                                           [[] m])]
                         [ids (update-in m [:stats :cleanup] inc)]))
         ids  (atom/swap-return! state cleanup-fn)]
     ids)))
link
(component/with [|pool| (create-pool)]
  (def -ids- (->> (for [i (range 8)]
                    (pool:acquire |pool|))
                  (mapv first)))

  (count (pool:resources:busy |pool|))
  => 8

  (doseq [id -ids-]
    (pool:release |pool| id))

  (count (pool:resources:busy |pool|))
  => 0

  (pool:cleanup |pool|)

  (count (pool:resources:idle |pool|))
  => 3)

pool:create ^

[m]
Added 3.0

creates an initial pool

v 3.0
(defn pool:create
  ([m]
   (-> (assoc (apply dissoc m +opts+)
              :state (atom {:stats {:create  0
                                    :reject  0
                                    :dispose 0
                                    :acquire 0
                                    :release 0
                                    :history []
                                    :cleanup 0
                                    :total-time 0.0
                                    :busy-time  0.0}
                            :busy    {}
                            :idle    {}
                            :lookup  {}
                            :options (merge *defaults* (select-keys m +opts+))}))
       (map->Pool))))
link
(let [pool (pool:create {:size 5
                         :max 8
                         :keep-alive 10000
                         :poll 20000
                         :resource {:create (fn [] ')
                                    :initial 0.3
                                    :thread-local true}})]
  [(pool? pool)
   (f/atom? (:state pool))
   (-> pool :resource :thread-local)])
=> [true true true]

pool:dispose ^

[{:keys [state resource], :as pool} id]
Added 3.0

disposes an idle object

v 3.0
(defn pool:dispose
  ([{:keys [state resource] :as pool} id]
   (swap! state dispose-fn id (:stop resource)) id))
link
(component/with [|pool| (create-pool)]
  (let [id (first (keys (pool:resources:idle |pool|)))]
    (pool:dispose |pool| id)
    (contains? (pool:resources:idle |pool|) id)))
=> false

pool:dispose-over ^

[{:keys [state resource], :as pool} id]
Added 3.0

disposes if idle and busy are over size limit

v 3.0
(defn pool:dispose-over
  ([{:keys [state resource] :as pool} id]
   (atom/swap-return! state (fn [{:keys [idle options] :as m}]
                           (if (get idle id)
                             [id  (dispose-fn m id (:stop resource))]
                             [nil m])))))
link
(component/with [|pool| (create-pool)]
  (let [[id _] (pool:acquire |pool|)]
    (pool:release |pool| id)
    (pool:dispose-over |pool| id)))
=> string?

pool:dispose:mark ^

[]
Added 3.0

marks the current resource for dispose

v 3.0
(defn pool:dispose:mark
  ([]
   (vreset! *dispose* true)))
link
(component/with [|pool| (create-pool)]
  (dotimes [i 2]
    ((wrap-pool-resource
      (fn [pool]
        (pool:dispose:mark))
      |pool|)))

  (count (pool:resources:idle |pool|))
  => 0)

pool:dispose:unmark ^

[]
Added 3.0

unmarks the current resource for dispose

v 3.0
(defn pool:dispose:unmark
  ([]
   (vreset! *dispose* nil)))
link
(component/with [|pool| (create-pool)]
  (dotimes [i 2]
    ((wrap-pool-resource
      (fn [pool]
        (pool:dispose:mark)
        (pool:dispose:unmark))
      |pool|)))

  (count (pool:resources:idle |pool|))
  => 2)

pool:health ^

[pool]
Added 3.0

returns health of the pool

v 3.0
(defn pool:health
  ([pool]
   (if (pool:started? pool)
     {:status :ok}
     {:status :not-healthy})))
link
(component/with [|pool| (create-pool)]
  (pool:health |pool|))
=> {:status :ok}

pool:info ^

[pool] [{:keys [state], :as pool} _]
Added 3.0

returns information about the pool

v 3.0
(defn pool:info
  ([pool]
   (pool:info pool :default))
  ([{:keys [state] :as pool} _]
   (let [{:keys [stats idle busy executor options]} @state
         {:keys [total-time busy-time acquire]} stats
         util (binding [*current* (time/time-ns)]
                (->> (concat (vals busy) (vals idle))
                     (mapv resource-info)
                     (apply merge-with + {:total (or total-time 0)
                                          :busy (or busy-time 0)})
                     (merge {:count (or acquire 0)})))
         util (assoc util
                     :utilization (float (/ (:busy util) (+ (:total util) 1)))
                     :duration    (long  (/ (:busy util) (+ (:count util) 1))))]
     {:running (boolean (and executor (not (e/exec:shutdown? executor))))
      :idle (count idle)
      :busy (count busy)
      :resource util})))
link
(component/with [|pool| (create-pool)]
  (pool:info |pool|))
=> (contains-in {:running true,
                 :idle 2, :busy 0,
                 :resource {:count 0, :total number?
                            :busy 0.0, :utilization 0.0,
                            :duration 0}})

pool:kill ^

[pool]
Added 3.0

kills the pool

v 3.0
(defn pool:kill
  ([pool]
   (binding [*stop-fn* e/exec:shutdown-now]
     (pool:stop pool))))
link
(component/with [|pool| (create-pool)]
  (pool:kill |pool|)
  => pool:stopped?)

pool:props ^

[pool]
Added 3.0

gets props for the pool

v 3.0
(defn pool:props
  ([pool]
   {:size {:get pool:props:size
           :set pool:props:size}
    :max  {:get pool:props:max
           :set pool:props:max}
    :keep-alive {:get pool:props:keep-alive
                 :set pool:props:keep-alive}
    :poll  {:get pool:props:poll
            :set pool:props:poll}}))
link
(component/with [|pool| (create-pool)]
  (keys (pool:props |pool|)))
=> (contains [:size :max :keep-alive :poll])

pool:props:keep-alive ^

NONE
(def pool:props:keep-alive (pool-props-fn :keep-alive))
link

pool:props:max ^

NONE
(def pool:props:max (pool-props-fn :max))
link

pool:props:poll ^

NONE
(def pool:props:poll (pool-props-fn :poll))
link

pool:props:size ^

NONE
(def pool:props:size (pool-props-fn :size))
link

pool:release ^

[pool id] [{:keys [state], :as pool} id dispose?]
Added 3.0

releases a resource back to the pool

v 3.0
(defn pool:release
  ([pool id]
   (pool:release pool id false))
  ([{:keys [state] :as pool} id dispose?]
   (let [resource-fn (fn [{:keys [update-time busy] :as resource}]
                       (let [current (time/time-ns)
                             diff (- current update-time)]
                         (assoc resource
                                :update-time current
                                :busy (+ busy diff)
                                :status :idle
                                :thread-id nil)))
         update-fn   (fn [m thread-id id resource]
                       (-> m
                           (update :idle assoc id (resource-fn resource))
                           (update :busy dissoc id)
                           (update-in [:stats :release] inc)
                           (update :lookup
                                   (fn [lu]
                                     (let [nentries (disj (get lu thread-id) id)]
                                       (if (empty? nentries)
                                         (dissoc lu thread-id)
                                         (assoc  lu thread-id nentries)))))))
         release-fn  (fn [{:keys [busy options] :as m}]
                       (let [{:keys [thread-id] :as resource} (get busy id)]
                         (cond (nil? resource)
                               [[:error (ex-info "Cannot find id" {:id id})]
                                m]

                               :else
                               [[:success id]
                                (update-fn m thread-id id resource)])))
         [[status id] {:keys [idle busy options executor]}] (atom/swap-return! state release-fn true)
         _  (if dispose? (pool:dispose pool id))
         _  (if (and executor
                     (< (:size options)
                        (+ (count idle)
                           (count busy))))
              (f/future:run (fn []
                              (pool:dispose-over pool id))
                            {:pool executor
                             :delay (:keep-alive options)}))]
     (case status
       :success id
       :error   nil))))
link
(component/with [|pool| (create-pool)]
  (let [[id _] (pool:acquire |pool|)]
    (pool:release |pool| id)))
=> string?

pool:resources:busy ^

[{:keys [state], :as pool}]
Added 3.0

returns all the busy resources

v 3.0
(defn pool:resources:busy
  ([{:keys [state] :as pool}]
   (collection/map-vals :object (:busy @state))))
link
(component/with [|pool| (create-pool)]
  (pool:resources:busy |pool|))
=> {}

(component/with [|pool| (create-pool)]
  (-> (doto |pool|
        (pool:acquire)
        (pool:acquire))
      (pool:resources:busy)
      count))
=> 2

pool:resources:idle ^

[{:keys [state], :as pool}]
Added 3.0

returns all the idle resources

v 3.0
(defn pool:resources:idle
  ([{:keys [state] :as pool}]
   (collection/map-vals :object (:idle @state))))
link
(component/with [|pool| (create-pool)]
  (count (pool:resources:idle |pool|)))
=> 2

(component/with [|pool| (create-pool)]
  (-> (doto |pool|
        (pool:acquire)
        (pool:acquire))
      (pool:resources:idle)))
=> {}

pool:resources:thread ^

[pool] [{:keys [state], :as pool} thread]
Added 3.0

returns acquired resources for a given thread

v 3.0
(defn pool:resources:thread
  ([pool]
   (pool:resources:thread pool (t/thread:current)))
  ([{:keys [state] :as pool} thread]
   (let [thread-id (t/thread:id thread)
         {:keys [busy lookup]} @state
         ids (get lookup thread-id)]
     (collection/map-juxt [identity (comp :object #(get busy %))] ids))))
link
(component/with [|pool| (create-pool)]
  (-> (doto |pool|
        (pool:acquire)
        (pool:acquire))
      (pool:resources:thread)
      count))
=> 2

pool:start ^

[{:keys [state resource poll], :as pool}]
Added 3.0

starts the pool

v 3.0
(defn pool:start
  ([{:keys [state resource poll] :as pool}]
   (if-not (pool:started? pool)
     (swap! state (fn [{:keys [options] :as m}]
                    (let [{:keys [size]} options
                          idle    (->> (range (long (* size (or (:initial resource)
                                                                0))))
                                       (map (fn [_]
                                              (let [id (std.lib.foundation/sid)]
                                                [id (pool-resource id pool)])))
                                       (into {}))
                          executor (e/executor:pool 2 2 1000)
                          _  (e/submit executor (pool-handler pool))]
                      (assoc m :executor executor :idle idle)))))
   pool))
link
(component/with [|pool| (create-pool)]
  (pool:start |pool|)
  => pool:started?)

pool:started? ^

[{:keys [state], :as pool}]
Added 3.0

checks if pool has started

v 3.0
(defn pool:started?
  ([{:keys [state] :as pool}]
   (let [{:keys [executor]} @state]
     (boolean (and executor (not (e/exec:shutdown? executor)))))))
link
(component/with [|pool| (create-pool)]
  (pool:started? |pool|))
=> true

pool:stop ^

[{:keys [state], :as pool}]
Added 3.0

stops the pool

v 3.0
(defn pool:stop
  ([{:keys [state] :as pool}]
   (if (pool:started? pool)
     (let [_ (swap! state (fn [{:keys [executor] :as m}]
                            (*stop-fn* executor)
                            (dissoc m :executor)))
           busy-ids (keys (:busy @state))
           _ (doseq [id busy-ids]
               (pool:release pool id))
           idle-ids (keys (:idle @state))
           _  (doseq [id idle-ids]
                (pool:dispose pool id))]))
   pool))
link
(component/with [|pool| (create-pool)]
  (pool:stop |pool|)
  => pool:stopped?)

pool:stopped? ^

[pool]
Added 3.0

checks if pool has stopped

v 3.0
(defn pool:stopped?
  ([pool]
   (not (pool:started? pool))))
link
(component/with [|pool| (create-pool)]
  (pool:stopped? |pool|))
=> false

pool:track-path ^

[pool]
Added 3.0

gets props for the pool

v 3.0
(defn pool:track-path
  ([pool]
   (or (get-in pool [:track :path])
       [:raw :pool])))
link
(component/with [|pool| (create-pool)]
  (pool:track-path |pool|))
=> [:raw :pool]

pool:with-resource ^

[[obj pool] & body]
Added 3.0

takes an object from the pool, performs operation then returns it

v 3.0
(defmacro pool:with-resource
  ([[obj pool] & body]
   `((wrap-pool-resource
      (fn [obj#]
        (let [~obj obj#]
          ~@body))
      ~pool))))
link
(component/with [|pool| (create-pool)]
  (pool:with-resource [obj |pool|]
    (str obj))
  => "")

pool? ^

[obj]
Added 3.0

checks that object is a pool

v 3.0
(defn pool?
  ([obj]
   (instance? Pool obj)))
link
(component/with [|pool| (create-pool)]
  (pool? |pool|))
=> true

resource-info ^

[res] [{:keys [create-time update-time busy used status]} mode]
Added 3.0

returns info about the pool resource

v 3.0
(defn resource-info
  ([res]
   (resource-info res :basic))
  ([{:keys [create-time update-time busy used status]} mode]
   (let [current (or *current* (time/time-ns))
         total-time   (- current create-time)
         busy-time  (case status
                      :busy (+ busy (- current update-time))
                      :idle busy)]
     (cond-> {:total       total-time
              :busy        busy-time}
       (= mode :full) (assoc :count       used
                             :utilization (float (/ busy-time (+ total-time 1)))
                             :duration    (long (/ busy-time (+ used 1))))))))
link
(component/with [|pool| (create-pool)]
  (->  (pool-resource "hello" |pool|)
       (resource-info :full)))
=> (contains {:total number?,
              :busy 0.0,
              :count 0,
              :utilization 0.0,
              :duration 0})

resource-string ^

[res]
Added 3.0

returns a string describing the resource

v 3.0
(defn resource-string
  ([res]
   (str "#res" (-> (resource-info res :full)
                   (update :total time/format-ns)
                   (update :duration time/format-ns)))))
link
(boolean
 (re-find #"^#res"
          (binding [*current* 10]
            (resource-string {:create-time 0
                              :update-time 0
                              :busy 0.0
                              :used 0
                              :status :idle}))))
=> true

wrap-pool-resource ^

[f pool]
Added 3.0

wraps a function to operate on a pool resource

v 3.0
(defn wrap-pool-resource
  ([f pool]
   (fn [& args]
     (let [[id obj] (pool:acquire pool)
           [out exception dispose?]
           (binding [*dispose* (volatile! nil)]
             (try
               [(apply f obj args) nil @*dispose*])
             (catch Throwable t
               [nil t @*dispose*]))]
       (pool:release pool id dispose?)
       (if exception
         (throw exception)
         out)))))
link
(component/with [|pool| (create-pool)]
  ((wrap-pool-resource (fn [obj]
                         (str obj))
                       |pool|)))
=> ""

Related

jvm.artifactdocs

artifact coordinates and paths

jvm.classloaderdocs

classpath and classloaders

jvm.depsdocs

JVM dependency resolution

jvm.monitordocs

JVM metrics

jvm.namespacedocs

namespace management

jvm.protocoldocs

JVM extension protocols

jvm.reflectdocs

Java reflection

jvm.requiredocs

dependency-aware requiring