hive-weave.pool — io.github.hive-agi/hive-weave 0.2.5

Bounded thread-pool primitives — factory + safe submit/await.

Extends hive-weave with a pool abstraction so downstream code does not reach into java.util.concurrent directly (DIP).

Responsibilities:

• Construct a bounded ThreadPoolExecutor with CallerRunsPolicy backpressure and a named thread factory (for JVM diagnostics).
• Expose submit! returning an opaque Future-like handle.
• Expose await! — submit + block up to a timeout, returning a fallback on timeout/error. Never hangs.
• Re-export pool-stats and shutdown! for lifecycle.

Callers keep pool instances in their own registry (e.g. named io/compute/event/memory pools) and hand them to await! when they need bounded, isolated execution for a piece of work.

Quick start: (require '[hive-weave.pool :as wp])

(def db-pool (wp/make-pool {:name "db" :size 8}))

(wp/await! db-pool (fn [] (query-database ...)) {:timeout-ms 5000 :fallback ::db-timeout}) ;; => result or ::db-timeout

await! clj

(await! pool f {:keys [timeout-ms fallback name] :or {name "pool-task"}})

Submit f to pool and block on its result up to :timeout-ms.

On timeout, cancels the task (with interrupt) and returns :fallback. On exception during execution, logs and returns :fallback.

Never hangs indefinitely.

Options: :timeout-ms — max wait in ms (required) :fallback — value returned on timeout or exception (default nil) :name — diagnostic label used in logs (default "pool-task")

bound-future clj macro

(bound-future & body)

Drop-in replacement for clojure.core/future that conveys the caller's dynamic-var bindings through the active IBindingConveyor.

capture-frame clj

(capture-frame)

Snapshot the current thread's binding frame. Pair with FixedFrameConveyor to inject this frame into work-fns running on other threads.

convey-fn clj

(convey-fn f)

Wrap thunk f via the active conveyor — the canonical entry point for any async submission that must preserve dynamic-var bindings.

get-conveyor clj

(get-conveyor)

Return active conveyor. Falls back to BoundFnConveyor if atom is nil.

IBindingConveyor clj protocol

convey clj

(convey this f)

Wrap thunk f so dynamic-var bindings transfer to the executing thread.

make-pool clj

(make-pool {:keys [name size queue-capacity keep-alive-s]
            :or {queue-capacity default-queue-capacity keep-alive-s 60}})

Create a bounded fixed-size ThreadPoolExecutor.

Options: :name — thread-name prefix and diagnostic label (required) :size — fixed pool size (required) :queue-capacity — bounded LinkedBlockingQueue capacity (default 256) :keep-alive-s — idle keep-alive in seconds (default 60)

CallerRunsPolicy is always used: when both workers and queue are saturated, the submitting thread runs the task itself. This provides upstream backpressure instead of unbounded thread creation or silent task drops.

pool-stats clj

(pool-stats pool)

Snapshot of a pool's runtime counters.

set-conveyor! clj

(set-conveyor! c)

Install conveyor c as the active binding conveyor. Returns prior conveyor.

shutdown! clj

(shutdown! pool & [{:keys [await-ms] :or {await-ms 5000}}])

Orderly shutdown: stop accepting new tasks, wait up to :await-ms for in-flight tasks, then force-shutdown. Default :await-ms is 5000.

submit! clj

(submit! pool f)

Submit f to pool, returning a java.util.concurrent.Future.

f is wrapped via the active IBindingConveyor (default BoundFnConveyor, equivalent to clojure.core/bound-fn*) so the caller's dynamic var frame is conveyed to the pool thread. This matches the behaviour of clojure.core/future and avoids a silent trap where code relying on binding loses its frame at the pool boundary.

On RejectedExecutionException (pool shut down), runs f on the caller thread and returns a synthetic already-completed Future.

with-pool-await clj macro

(with-pool-await pool opts & body)

Submit body to pool, block up to (:timeout-ms opts), return (:fallback opts) on timeout/exception.

(with-pool-await memory-pool {:timeout-ms 30000 :fallback ::failed} (chroma/add-entry! ...))