Skip to content

Add a thread-pool engine for compute behind a flag - #88

Merged
clojure-finance merged 1 commit into
2.x.xfrom
engine-spike
Oct 13, 2026
Merged

clojure-finance merged 1 commit into
2.x.xfrom
engine-spike

Conversation

@buehlmaier

Copy link
Copy Markdown
Collaborator

First stage of replacing Onyx: the plain compute path (no group-by, aggregate or join) can run on a small in-process engine. Onyx stays the default.

What changes

  • New clojask.engine:
    • run: a reader thread splits the input into batches, a fixed pool of num-work daemon threads works them, and the calling thread writes the results in input order.
    • At most 2 * num-work + 2 batches are in flight.
    • Workers return realized vectors, so the work stays on the pool threads.
    • The input is closed through the read function's arity-1 :close, also when the job fails.
    • No global state, so concurrent computes on this engine don't interfere.
    • start-compute: the compute driver, producing the same rows as worker-func-gen-format.
  • DataFrame.compute picks the engine from the system property clojask.engine, else the env var CLOJASK_ENGINE: onyx (default) or local. An unknown value throws.
  • shutdown-test pins itself to Onyx with a fixture, since it tests the Onyx environment.

Behaviour on local

  • Output is always in input order, so :order has no effect.
  • Errors are still ExecutionException, now [run stage (compute)] (original error: ...). The cause is the real exception instead of Onyx's ExceptionInfo, and the message no longer points to .clojask/clojask.log.

Benchmark

Run on a synthetic 1.8M-row, 160 MB Compustat-like CSV, using the cases from benchmark/clojure-benchmark.clj (JDK 25, 8 cores). Times are the fastest run, in seconds:

workers element row full
1 (onyx / local) 8.3 / 3.0 8.3 / 2.9 12.8 / 5.7
4 8.4 / 3.7 9.3 / 3.7 14.2 / 4.4
8 10.5 / 4.3 10.6 / 4.5 16.2 / 4.3
  • Outputs are identical between engines once sorted; local output is byte-identical to the input order on the full case.
  • More workers only help the full case. The per-batch work is cheap, so reading and writing the file are the bottleneck on both engines.
  • With -Xmx96m, local computed the whole file correctly in 9.8 s. Onyx hit OutOfMemoryError in the embedded ZooKeeper and stalled.

Testing

  • lein test: 52 tests, 198 assertions, 0 failures, on both the default engine and CLOJASK_ENGINE=local.
  • A side-by-side smoke test (plain, :select, filter + operate, in-memory, melt, a worker exception with :exception true) gave identical results apart from the error message text above.
  • CI runs only the default engine. Adding a CLOJASK_ENGINE=local leg to the workflow is left for a follow-up.

The plain compute path can now run without Onyx. clojask.engine/run
reads batches on its own thread, works them on a fixed pool of
num-work threads and hands the results to the caller thread in input
order, with at most 2 * num-work + 2 batches in flight. The input is
closed through the read function's :close, also when the job fails.
Workers realize their batch, so the work stays on the pool threads.

The system property clojask.engine, else CLOJASK_ENGINE, selects
"onyx" (the default) or "local". Aggregate, group-by and joins still
run on Onyx. shutdown-test pins itself to Onyx, since it tests the
Onyx environment.
@clojure-finance
clojure-finance merged commit 391c480 into 2.x.x Oct 13, 2026
3 checks passed
@buehlmaier
buehlmaier deleted the engine-spike branch October 13, 2026 08:15
This was referenced Oct 20, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants