std.concurrent.executor

thread pools, scheduling, submission, and executor lifecycle

Construct JVM executors with consistent queue handling, tracking, health information, and component-compatible lifecycle operations.

Concurrency overview

1    Overview

The namespace wraps ExecutorService and ScheduledThreadPoolExecutor with constructors for single, pooled, cached, scheduled, and shared executors. Submission functions return foundation futures and support delay, minimum runtime, timeout, and default values.

2    Walkthrough

2.1    Submit work to a pool

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

(def pool
  (executor/executor
   {:type :pool
    :size 2
    :max 4
    :keep-alive 1000
    :queue {:size 32}}))

@(executor/submit pool
                  (fn [] {:status :complete})
                  {:max 5000})

(executor/executor:info pool)
(executor/exec:shutdown pool)

2.2    Schedule repeated work

Create a scheduled executor for delayed or repeated tasks. Fixed-rate scheduling targets a wall-clock cadence, while fixed-delay scheduling waits after each completion.

(def scheduler (executor/executor:scheduled 1))

(executor/schedule scheduler #(println :once) 250)
(executor/schedule:fixed-rate scheduler #(println :tick) 1000)
(executor/schedule:fixed-delay scheduler #(println :poll) 1000)

(executor/exec:shutdown-now scheduler)

2.3    Share an executor

Register a long-lived executor under an application-specific ID and retrieve it through the generic executor constructor. Always unshare it before final shutdown.

(executor/executor:share :application/background pool)
(executor/executor {:type :shared :id :application/background})
(executor/executor:unshare :application/background)

3    API

*shared* ^

NONE
(defonce ^:dynamic ^:private *shared* (atom {}))
link

exec:at-capacity? ^

[executor]
Added 3.0

checks if executor is at capacity

v 3.0
(defn exec:at-capacity?
  ([executor]
   (= (exec:current-active executor)
      (exec:pool-max executor))))
link
(let [exe (executor:single 1)]
  (try
    (try (submit exe (fn [] (Thread/sleep 1000)))
         (catch Throwable t))
    (exec:at-capacity? exe)
    (finally (exec:shutdown-now exe))))
=> boolean?

exec:await-termination ^

[service] [service ms]
Added 3.0

await termination for executor service

v 3.0
(defn exec:await-termination
  ([^ExecutorService service]
   (exec:await-termination service Long/MAX_VALUE))
  ([^ExecutorService service ^long ms]
   (.awaitTermination service ms TimeUnit/MILLISECONDS)))
link
;; Shutdown
(def -counter- (atom 0))

(-> (executor:single)
    (doto (submit (fn [] (Thread/sleep 100) (swap! -counter- inc))))
    (doto (submit (fn [] (Thread/sleep 100) (swap! -counter- inc))))
    (exec:shutdown)
    (doto (exec:await-termination)))

@-counter-
=> 2

;; Shutdown now
(def -counter- (atom 0))

(-> (executor:single)
    (doto (submit (fn [] (Thread/sleep 100) (swap! -counter- inc))))
    (doto (submit (fn [] (Thread/sleep 100) (swap! -counter- inc))))
    (exec:shutdown-now)
    (doto (exec:await-termination)))

@-counter-
=> 0

exec:current-active ^

[service]
Added 3.0

returns number of active threads in pool

v 3.0
(defn exec:current-active
  ([^ThreadPoolExecutor service]
   (.getActiveCount service)))
link
(let [exe (executor:single)]
  (try
    (exec:current-active exe)
    (finally (exec:shutdown exe))))
=> number?

exec:current-completed ^

[service]
Added 3.0

returns number of completed tasks

v 3.0
(defn exec:current-completed
  ([^ThreadPoolExecutor service]
   (.getCompletedTaskCount service)))
link
(let [exe (executor:single)]
  (try
    (exec:current-completed exe)
    (finally (exec:shutdown exe))))
=> number?

exec:current-size ^

[service]
Added 3.0

returns number of threads in pool

v 3.0
(defn exec:current-size
  ([^ThreadPoolExecutor service]
   (.getPoolSize service)))
link
(let [exe (executor:single)]
  (try
    (exec:current-size exe)
    (finally (exec:shutdown exe))))
=> number?

exec:current-submitted ^

[service]
Added 3.0

returns number of submitted tasks

v 3.0
(defn exec:current-submitted
  ([^ThreadPoolExecutor service]
   (.getTaskCount service)))
link
(let [exe (executor:single)]
  (try
    (exec:current-submitted exe)
    (finally (exec:shutdown exe))))
=> number?

exec:get-queue ^

[service]
Added 3.0

gets the queue from the executor

v 3.0
(defn ^BlockingQueue exec:get-queue
  ([service]
   (if (instance? ThreadPoolExecutor service)
     (.getQueue ^ThreadPoolExecutor service)
     (throw (ex-info "Cannot access queue" {:service service})))))
link
(let [queue (q/queue)
      exe   (executor:pool 10 10 1000 queue)]
  (try
    (= queue (exec:get-queue exe))
    (finally
      (exec:shutdown exe))))
=> true

exec:increase-capacity ^

[executor] [executor n]
Added 3.0

increases the capacity of the executor

v 3.0
(defn exec:increase-capacity
  ([executor]
   (exec:increase-capacity executor (* 2 (exec:pool-max executor))))
  ([executor n]
   (let [max (exec:pool-max executor)]
     (if (< max n) (exec:pool-max executor n))
     (exec:pool-size executor n))))
link
(let [exe (executor:single)]
  (try
    (doto exe (exec:increase-capacity))
    (exec:pool-size exe)
    (finally (exec:shutdown exe))))
=> 2

exec:keep-alive ^

[service] [service length]
Added 3.0

gets and sets the keep alive time

v 3.0
(defn exec:keep-alive
  ([^ThreadPoolExecutor service]
   (.getKeepAliveTime service TimeUnit/MILLISECONDS))
  ([^ThreadPoolExecutor service length]
   (.setKeepAliveTime service length TimeUnit/MILLISECONDS)))
link
(let [exe (executor:single)]
  (try
    (doto exe (exec:keep-alive 1000))
    (exec:keep-alive exe)
    (finally (exec:shutdown exe))))
=> 1000

exec:pool-max ^

[service] [service size]
Added 3.0

gets and sets the core pool max

v 3.0
(defn exec:pool-max
  ([^ThreadPoolExecutor service]
   (.getMaximumPoolSize service))
  ([^ThreadPoolExecutor service size]
   (.setMaximumPoolSize service size)))
link
(let [exe (executor:single)]
  (try
    (doto exe (exec:pool-max 10))
    (exec:pool-max exe)
    (finally (exec:shutdown exe))))
=> 10

exec:pool-size ^

[service] [service size]
Added 3.0

gets and sets the core pool size

v 3.0
(defn exec:pool-size
  ([^ThreadPoolExecutor service]
   (.getCorePoolSize service))
  ([^ThreadPoolExecutor service size]
   (.setCorePoolSize service (int size))))
link
(let [exe (executor:single)]
  (try
    (doto exe
      (exec:pool-max 20)
      (exec:pool-size 10))
    (exec:pool-size exe)
    (finally
      (exec:shutdown exe))))
=> 10

exec:queue ^

[] [arg]
Added 3.0

contructs a raw queue in different ways

v 3.0
(defn ^BlockingQueue exec:queue
  ([]
   (exec:queue nil))
  ([arg]
   (cond (q/queue? arg) arg

         (nil? arg) (q/queue)

         (integer? arg) (q/queue:fixed arg)

         (map? arg)
         (cond (integer? (:size arg))
               (q/queue:fixed (:size arg))

               :else
               (q/queue))

         :else (throw (ex-info "Invalid input" {:input arg})))))
link
(exec:queue)

(exec:queue 1)

(exec:queue {:size 1})

(exec:queue {})
=> java.util.concurrent.LinkedBlockingQueue

(exec:queue (q/queue))
=> java.util.concurrent.LinkedBlockingQueue

exec:rejected-handler ^

[service] [service f]
Added 3.0

sets the rejected task handler

v 3.0
(defn exec:rejected-handler
  ([^ThreadPoolExecutor service]
   (.getRejectedExecutionHandler service))
  ([^ThreadPoolExecutor service f]
   (.setRejectedExecutionHandler service
                                 (reify RejectedExecutionHandler
                                   (rejectedExecution [_ task executor]
                                     (f task executor))))))
link
(def -counter- (atom 0))
(doto (executor:single 1)
  (exec:rejected-handler (fn [_ _] (swap! -counter- inc)))
  (f/-> (doseq [i (range 10)]
          (submit % (fn [] (Thread/sleep 1000)))))
  (exec:shutdown-now))

(<= 8 @-counter-)
=> true

(let [exe (executor:single)]
  (try
    (exec:rejected-handler exe)
    (finally (exec:shutdown exe))))
=> anything

exec:shutdown ^

[service]
Added 3.0

shuts down executor

v 3.0
(defn exec:shutdown
  ([^ExecutorService service]
   (doto service
     (track/untrack)
     (.shutdown))))
link
(-> (executor:single)
    (doto (submit (fn [] (Thread/sleep 1000))))
    (doto (submit (fn [] (Thread/sleep 1000))))
    (doto (exec:shutdown)))
=> (all exec:shutdown?
        exec:terminating?
        (comp not exec:terminated?))

exec:shutdown-now ^

[service]
Added 3.0

shuts down executor immediately

v 3.0
(defn exec:shutdown-now
  ([^ExecutorService service]
   (doto service
     (track/untrack)
     (.shutdownNow))))
link
(-> (executor:single)
    (doto (submit (fn [] (Thread/sleep 1000))))
    (doto (submit (fn [] (Thread/sleep 1000))))
    (exec:shutdown-now)
    (exec:shutdown?))
=> true

exec:shutdown? ^

[service]
Added 3.0

checks if executor is shutdown

v 3.0
(defn exec:shutdown?
  ([^ExecutorService service]
   (.isShutdown service)))
link
(-> (executor:single)
    (doto (exec:shutdown))
    (exec:shutdown?))
=> true

exec:terminated? ^

[service]
Added 3.0

checks if executor is shutdown and all threads have finished

v 3.0
(defn exec:terminated?
  ([^ExecutorService service]
   (.isTerminated service)))
link
(-> (executor:single)
    (doto (exec:shutdown))
    (doto (exec:await-termination))
    (exec:terminated?))
=> true

exec:terminating? ^

[service]
Added 3.0

check that executor is terminating

v 3.0
(defn exec:terminating?
  ([^ThreadPoolExecutor service]
   (.isTerminating service)))
link
(-> (executor:single)
    (doto (submit (fn [] (Thread/sleep 1000))))
    (doto (exec:shutdown))
    (exec:terminating?))
=> true

executor ^

Added 3.0

creates an executor

v 3.0
(defmulti executor
  :type)
link
(doto (executor {:type :pool
                 :size 3
                 :max 3
                 :keep-alive 1000})
  (exec:shutdown))

(doto (executor {:type :single
                 :size 1})
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

executor-string ^

[executor]

NONE
(defn- executor-string
  [executor]
  (let [tag (if (instance? ScheduledThreadPoolExecutor executor)
              "scheduled"
              "raw")]
    (str "#" tag ".executor" (executor:info executor [:type :counter :current]))))
link

executor:cached ^

[]
Added 3.0

creates a cached executor

v 3.0
(defn ^ExecutorService executor:cached
  ([]
   (doto (Executors/newCachedThreadPool)
     (track/track))))
link
(doto (executor:cached)
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

executor:health ^

[executor]
Added 3.0

returns health of the executor

v 3.0
(defn executor:health
  ([executor]
   (if (exec:shutdown? executor)
     {:status :not-healthy}
     {:status :ok})))
link
(let [exe (executor:single)]
  (try
    (executor:health exe)
    (finally (exec:shutdown exe))))
=> {:status :ok}

(-> (executor:single)
    (exec:shutdown)
    (executor:health))
=> {:status :not-healthy}

executor:info ^

[executor] [executor k]
Added 3.0

returns executor service info

v 3.0
(defn executor:info
  ([^ThreadPoolExecutor executor]
   (executor:info executor #{:type :running :current :counter :options}))
  ([^ThreadPoolExecutor executor k]
   (let [[return items]  (cond (keyword? k)
                               [k #{k}]

                               :else
                               [identity (set k)])
         queue  (.getQueue executor)]
     (cond-> {}
       (:type items)    (assoc :type (executor:type executor))
       (:running items) (assoc :running (not (exec:shutdown? executor)))
       (:current items) (assoc :current {:threads (exec:current-size executor)
                                         :active  (exec:current-active executor)
                                         :queued  (count queue)
                                         :terminated (exec:terminated? executor)})
       (:counter items) (assoc :counter {:submit   (exec:current-submitted executor)
                                         :complete (exec:current-completed executor)})
       (:options items) (assoc :options {:pool {:size (exec:pool-size executor)
                                                :max (exec:pool-max executor)
                                                :keep-alive (exec:keep-alive executor)}
                                         :queue {:remaining (q/remaining-capacity queue)
                                                 :total (count queue)}})
       :then return))))
link
(let [exe (executor:single)]
  (try
    (executor:info exe)
    (finally (exec:shutdown exe))))
=> {:type :single,
    :running true,
    :current {:threads 0, :active 0, :queued 0, :terminated false},
    :counter {:submit 0, :complete 0},
    :options {:pool {:size 0, :max 1, :keep-alive 0},
              :queue {:remaining 2147483647, :total 0}}}

executor:kill ^

NONE
(def executor:kill  exec:shutdown-now)
link

executor:pool ^

[size max keep-alive] [size max keep-alive size-or-queue]
Added 3.0

constructs a pool executor

v 3.0
(defn ^ExecutorService executor:pool
  ([size max keep-alive]
   (executor:pool size max keep-alive (q/queue)))
  ([^long size ^long max ^long keep-alive size-or-queue]
   (doto (ThreadPoolExecutor. size max keep-alive
                              TimeUnit/MILLISECONDS (exec:queue size-or-queue))
     (track/track))))
link
(doto (executor:pool 10 10 1000 {:size 10})
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

executor:props ^

[executor]
Added 3.0

returns props for getters and setters

v 3.0
(defn executor:props
  ([^ThreadPoolExecutor executor]
   {:pool {:size       {:get exec:pool-size
                        :set exec:pool-size}
           :max        {:get exec:pool-max
                        :set exec:pool-max}
           :keep-alive {:get exec:keep-alive
                        :set exec:keep-alive}}
    :rejected-handler  {:get exec:rejected-handler
                        :set exec:rejected-handler}}))
link
(let [exe (executor:single)]
  (try
    (executor:props exe)
    (finally (exec:shutdown exe))))
=> {:pool {:size {:get exec:pool-size
                  :set exec:pool-size},
           :max {:get exec:pool-max
                 :set exec:pool-max},
           :keep-alive {:get exec:keep-alive
                        :set exec:keep-alive}}
    :rejected-handler {:get exec:rejected-handler
                       :set exec:rejected-handler}}

executor:scheduled ^

[size]
Added 3.0

constructs a scheduled executor

v 3.0
(defn ^ScheduledThreadPoolExecutor executor:scheduled
  ([^long size]
   (doto (ScheduledThreadPoolExecutor. size)
     (track/track))))
link
(doto (executor:scheduled 10)
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

executor:share ^

[id executor]
Added 3.0

registers a shared executor

v 3.0
(defn executor:share
  ([id ^ExecutorService executor]
   (swap! *shared* (fn [m]
                     (if (or (get @f/*pools* id)
                             (get m id))
                       (throw (ex-info "Already exists" {:id id}))
                       (assoc m id executor))))))
link
(let [exe (executor:single)]
  (try
    (executor:share :test exe)
    (get (executor:shared) :test)
    (finally
      (executor:unshare :test)
      (exec:shutdown exe))))
=> anything

(executor:unshare :test)

executor:shared ^

[]
Added 3.0

lists all shared executors

v 3.0
(defn executor:shared
  ([]
   (merge @*shared* @f/*pools*)))
link
(-> (executor:shared) keys sort)
=> [:async :default :pooled]

executor:single ^

[] [size-or-queue]
Added 3.0

constructs a single executor

v 3.0
(defn ^ExecutorService executor:single
  ([]
   (executor:single (q/queue)))
  ([size-or-queue]
   (doto (ThreadPoolExecutor. 0 1 0 TimeUnit/MILLISECONDS (exec:queue size-or-queue))
     (track/track))))
link
;; any sized pool
(doto (executor:single)
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

;; fixed pool
(doto (executor:single {:size 10})
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

executor:start ^

NONE
(def executor:start  identity)
link

executor:started? ^

NONE
(def executor:started? (comp not exec:shutdown?))
link

executor:stop ^

NONE
(def executor:stop  exec:shutdown)
link

executor:stopped? ^

NONE
(def executor:stopped? exec:shutdown?)
link

executor:submit ^

NONE
(def executor:submit submit)
link

executor:type ^

[executor]
Added 3.0

returns executor service type

v 3.0
(defn executor:type
  ([^ThreadPoolExecutor executor]
   (cond (instance? ScheduledThreadPoolExecutor executor)
         :scheduled

         (= 1 (exec:pool-max executor))
         :single

         (and (= Integer/MAX_VALUE (exec:pool-max executor))
              (let [queue (exec:get-queue executor)]
                (and (zero? (q/remaining-capacity queue))
                     (zero? (count queue)))))
         :cached

         :else :pool)))
link
(let [exe (executor:single)]
  (try
    (executor:type exe)
    (finally (exec:shutdown exe))))
=> :single

executor:unshare ^

[id]
Added 3.0

deregisters a shared executor

v 3.0
(defn executor:unshare
  ([id]
   (atom/swap-return! *shared* (fn [m] [(get m id) (dissoc m id)]))))
link
(let [exe (executor:single)]
  (try
    (executor:share :test exe)
    (executor:unshare :test)
    (get (executor:shared) :test)
    (finally
      (executor:unshare :test)
      (exec:shutdown exe))))
=> nil

schedule ^

[service f interval] [service f interval {:keys [min]}]
Added 3.0

schedules task for execution

v 3.0
(defn schedule
  ([^ScheduledThreadPoolExecutor service ^Callable f ^long interval]
   (schedule service f interval nil))
  ([^ScheduledThreadPoolExecutor service ^Callable f ^long interval {:keys [min]}]
   (let [^Callable f (cond-> f
                       min (wrap-min-time min))]
     (.schedule service f interval TimeUnit/MILLISECONDS))))
link
(f/-> (doto (executor:scheduled 1)
        (schedule (fn [] (Thread/sleep 10)) 10)
        (schedule (fn [] (Thread/sleep 10)) 20)
        (schedule (fn [] (Thread/sleep 10)) 30)
        (schedule (fn [] (Thread/sleep 10)) 40))
      (do (Thread/sleep 100)
          (exec:current-completed %)))
=> 4

schedule:fixed-delay ^

[service f delay] [service f delay {:keys [initial min]}]
Added 3.0

schedules task at fixed delay

v 3.0
(defn schedule:fixed-delay
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long delay]
   (schedule:fixed-delay service f delay nil))
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long delay {:keys [initial min]}]
   (let [^Runnable f (cond-> f
                       min (wrap-min-time min))]
     (.scheduleWithFixedDelay service f (or initial 0) delay TimeUnit/MILLISECONDS))))
link
(def -counter- (atom 0))

(f/-> (doto (executor:scheduled 1)
        (schedule:fixed-delay (fn [] (Thread/sleep 20)) 30))
      (do (Thread/sleep 100)
          (exec:shutdown-now %)
          (exec:current-completed %)))
=> #(<= 1 %)

schedule:fixed-rate ^

[service f interval] [service f interval {:keys [initial min]}]
Added 3.0

schedules task at a fixed rate

v 3.0
(defn schedule:fixed-rate
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long interval]
   (schedule:fixed-rate service f interval nil))
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long interval {:keys [initial min]}]
   (let [^Runnable f (cond-> f
                       min (wrap-min-time min))]
     (.scheduleAtFixedRate service f (or initial 0) interval TimeUnit/MILLISECONDS))))
link
(def -counter- (atom 0))

(f/-> (doto (executor:scheduled 1)
        (schedule:fixed-rate (fn [] (Thread/sleep 20)) 30))
      (do (Thread/sleep 100)
          (exec:shutdown-now %)
          (exec:current-completed %)))
=> #(<= 3 %)

submit ^

[service f] [service f {:keys [min max delay default], :as m}]
Added 3.0

submits a task to an executor

v 3.0
(defn submit
  ([^ExecutorService service ^Callable f]
   (submit service f nil))
  ([^ExecutorService service f {:keys [min max delay default] :as m}]
   (let [^Callable f (cond-> f
                       min (wrap-min-time min (or delay 0)))
         opts (cond-> {:pool service}
                max     (assoc :timeout max)
                default (assoc :default default)
                delay   (assoc :delay delay))]
     (f/future:run f opts))))
link
(let [exe (executor:single)]
  (try
    @(submit exe
             (fn [])
             {:min 100})
    (finally
      (exec:shutdown exe))))

(let [exe (executor:single)]
  (try
    (try
      @(submit exe
               (fn []
                 (Thread/sleep 1000))
               {:max 100})
      (catch java.util.concurrent.ExecutionException e
        (if (instance? java.util.concurrent.TimeoutException (.getCause e))
          :timeout
          (throw e))))
    (finally
      (exec:shutdown exe))))
=> :timeout

submit-notify ^

[service f] [service f {:keys [min max delay default], :as m}]
Added 3.0

submits a task (generally to a fixed size queue)

v 3.0
(defn submit-notify
  ([^ExecutorService service ^Callable f]
   (submit-notify service f nil))
  ([^ExecutorService service f {:keys [min max delay default] :as m}]
   (if (pos? (q/remaining-capacity (exec:get-queue service)))
     (h/suppress (submit service f m)))))
link
(doto (executor:single 1)
  (submit-notify (fn [])
                 1000)
  (submit-notify (fn [])
                 1000)
  (submit-notify (fn [])
                 1000)
  (exec:shutdown))
=> java.util.concurrent.ThreadPoolExecutor

wrap-min-time ^

[f total] [f total delay]
Added 3.0

wraps a function with min-time and delay

v 3.0
(defn ^Callable wrap-min-time
  ([f total]
   (wrap-min-time f total 0))
  ([f total delay]
   (fn []
     (let [start (System/currentTimeMillis)
           _ (if (and delay (pos? delay)) (Thread/sleep (long delay)))
           _ (f)
           end (System/currentTimeMillis)
           duration (- end start)]
       (if (> total duration)
         (Thread/sleep (long (- total duration))))))))
link
((wrap-min-time (fn []) 20 0))
=> nil

((wrap-min-time (fn []) 100 10))
=> nil

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