hive-weave.parallel — io.github.hive-agi/hive-weave 0.4.6

Bounded parallel execution — safe alternatives to pmap and raw futures.

• bounded-pmap — pmap with concurrency limit + per-item timeout
• fork-join — concurrent futures with collective timeout budget
• fan-out — fire N tasks, collect results with timeout

Unlike pmap, these primitives:

1. Bound concurrency (no unbounded thread creation)
2. Have timeouts (no indefinite hangs)
3. Return fallback values on timeout (graceful degradation)

bounded-pmap clj

(bounded-pmap {:keys [concurrency timeout-ms fallback stack-bytes virtual?]
               :or {concurrency 4 timeout-ms 10000 fallback nil}}
              f
              coll)

Like pmap but with bounded concurrency and per-item timeout.

Options: :concurrency max items in flight at once (default 4) :timeout-ms per-item timeout in ms (default 10000) :fallback value for timed-out/failed items (default nil) :stack-bytes explicit worker stack size (default: the JVM's) :virtual? run on virtual threads (default: whatever this JVM has)

(bounded-pmap {:concurrency 3 :timeout-ms 5000} fetch-entry-preview entry-ids) ;; => [result1 result2 nil result4 ...] (nil = timed out)

On a JVM with virtual threads the bound is a SEMAPHORE, not a thread count: every item is submitted at once and waits its turn for a permit. The number of items running together is exactly :concurrency either way, so the observable behaviour is unchanged; what goes away is the fixed pool of N platform threads that each call used to create and destroy. A caller that ran four of these per request was starting and stopping sixteen OS threads to do sixteen short reads. The carrier threads underneath the virtual ones are not free either, but they are created once, up to the core count, and shared by every fan-out in the process rather than minted per call.

Below 21 the fixed pool is still the mechanism, and :virtual? false forces it on any JVM, which is how that branch stays testable.

Pass :stack-bytes when f calls into native code that recurses: the default worker stack is sized for Clojure, and a native overflow is a SIGSEGV that takes the process down instead of yielding the fallback (hive-weave.stack). A virtual thread's stack is not configurable, so :stack-bytes always means platform threads and overrides :virtual? rather than being quietly dropped.

bounded-pmap-strict clj

(bounded-pmap-strict opts f coll)

Like bounded-pmap, but an item that times out or throws is an ERROR, not a fallback value. Throws ex-info {:type :hive-weave/pmap-incomplete :missing [index ...] :count n} when any item is missing; otherwise returns the full result vector.

Use it for partition-and-combine work, where a substituted nil is indistinguishable from "this chunk found nothing" and a sum would come back silently low. Accepts the same options as bounded-pmap except :fallback, which it owns.

fan-out clj

(fan-out opts f coll)

Apply f to each item in coll concurrently, collect results with timeout. Like bounded-pmap but returns a Result for the whole batch.

(fan-out {:concurrency 4 :timeout-ms 5000} fetch-preview ids) ;; => (ok [r1 r2 r3 ...]) — all completed ;; => (ok [r1 nil r3 ...]) — some timed out (nils)

Always returns (ok ...) — individual failures become nil/fallback.

fork-join clj

(fork-join {:keys [budget-ms] :or {budget-ms 15000}} & tasks)

Execute named tasks concurrently with a collective timeout budget. Each task is a [key thunk] or [key thunk fallback] triple. Returns a map of {key result} — timed-out tasks get their fallback.

(fork-join {:budget-ms 15000} [:tags #(query-tags candidate-tags) {}] [:kg #(expand-via-kg vanilla-ids) #{}]) ;; => {:tags {...} :kg #{...}}

Options: :budget-ms — total time budget for all tasks (default 15000)

map-within clj

(map-within {:keys [pool deadline-ms name] :or {name "map-within"}} f coll)

Apply f to every item of coll on :pool, all under ONE deadline. Returns a vector of Results, one per item, in input order:

(ok value) the item finished inside the deadline (err :weave/timeout {...}) the deadline passed first; the task is cancelled (err :weave/exception {...}) f threw (err :weave/rejected {...}) a saturated :abort pool refused the item

Every error carries :index (position in coll) and :name.

Options: :pool ExecutorService to run on (required). It is NOT shut down: the caller owns it, which is what makes the concurrency bound a property of the pool rather than of one call. :deadline-ms wall-clock budget for the WHOLE batch (required, pos-int) :name diagnostic label (default map-within)

The difference from bounded-pmap: its timeout is per item and the batch bound is that figure times the number of waves, so a stalled backend costs a caller timeout x waves. Here the budget is the one number the caller states, however many items there are. Items still queued or running when it passes are cancelled, so an expired batch does not keep the pool busy.

Items go through hive-weave.pool/submit!, so the caller's dynamic bindings reach the pool threads. Pair this with a pool built with :rejection :abort: a :caller-runs pool runs the overflow on the submitting thread, where the deadline cannot interrupt it. Nothing is thrown for a failed item; the Result carries it, so a caller can tell no-data from the-read-failed.

missing-indices clj

(missing-indices sentinel results)

Indices of results that hold sentinel: the items a bounded-pmap run could not answer (timed out or threw). Pure.