com.climate.claypoole — org.clj-commons/claypoole 1.2.2

Threadpool tools for Clojure. Claypoole provides parallel functions and macros that use threads from a pool and otherwise act like key builtins like future, pmap, for, and so on. See the file README.md for an introduction.

A threadpool is just an ExecutorService with a fixed number of threads. In general, you can use your own ExecutorService in place of any threadpool, and you can treat a threadpool as you would any other ExecutorService.

*default-pmap-buffer* clj

This is an advanced configuration option. You probably don't need to set this!

When doing a pmap, Claypoole pushes input tasks into the threadpool. It normally tries to keep the threadpool full, plus it adds a buffer of size nthreads. If it can't find out the number of threads in the threadpool, it just tries to keep default-pmap-buffer tasks in the pool.

*parallel* clj

A dynamic binding to disable parallelism. If you do (binding [parallel false] body) then the body will have no parallelism. Disabling parallelism this way is handy for testing.

completable-future clj macro

(completable-future pool & body)

Like clojure.core/future, but using a threadpool and returns a CompletableFuture.

The threadpool may be one of 3 things:

1. An ExecutorService, e.g. one created by threadpool.
2. The keyword :builtin. In this case, the future will use the built-in agent threadpool, the same threadpool used by an ordinary clojure.core/future.
3. The keyword :serial. In this case, the computation will be performed in serial. This may be helpful during profiling, for example.

completable-future-call clj

(completable-future-call pool f)

Like clojure.core/future-call, but using a threadpool, and returns a CompletableFuture.

The threadpool may be one of 3 things:

1. An ExecutorService, e.g. one created by threadpool.
2. The keyword :builtin. In this case, the future will use the built-in agent threadpool, the same threadpool used by an ordinary clojure.core/future.
3. The keyword :serial. In this case, the computation will be performed in serial. This may be helpful during profiling, for example.

future clj macro

(future pool & body)

Like clojure.core/future, but using a threadpool.

The threadpool may be one of 3 things:

1. An ExecutorService, e.g. one created by threadpool.
2. The keyword :builtin. In this case, the future will use the built-in agent threadpool, the same threadpool used by an ordinary clojure.core/future.
3. The keyword :serial. In this case, the computation will be performed in serial. This may be helpful during profiling, for example.

future-call clj

(future-call pool f)

Like clojure.core/future-call, but using a threadpool.

The threadpool may be one of 3 things:

1. An ExecutorService, e.g. one created by threadpool.
2. The keyword :builtin. In this case, the future will use the built-in agent threadpool, the same threadpool used by an ordinary clojure.core/future.
3. The keyword :serial. In this case, the computation will be performed in serial. This may be helpful during profiling, for example.

ncpus clj

(ncpus)

Get the number of available CPUs.

pcalls clj

(pcalls pool & fs)

Like clojure.core.pcalls, except it takes a threadpool. For more detail on its parallelism and on its threadpool argument, see pmap.

pdoseq clj macro

(pdoseq pool bindings & body)

Like doseq, but in parallel. Unlike the streaming sequence functions (e.g. pmap), pdoseq blocks until all the work is done.

Similar to pfor, only the body is done in parallel. For more details, see pfor.

pfor clj macro

(pfor pool bindings & body)

A parallel version of for. It is like for, except it takes a threadpool and is parallel. For more detail on its parallelism and on its threadpool argument, see pmap.

Note that while the body is executed in parallel, the bindings are executed in serial, so while this will call complex-computation in parallel: (pfor pool [i (range 1000)] (complex-computation i)) this will not have useful parallelism: (pfor pool [i (range 1000) :let [result (complex-computation i)]] result)

You can use the special binding :priority (which must be the last binding) to set the priorities of the tasks. (upfor (priority-threadpool 10) [i (range 1000) :priority (inc i)] (complex-computation i))

pmap clj

(pmap pool f & arg-seqs)

Like clojure.core.pmap, except:

1. It is eager, returning an ordered sequence of completed results as they are available.
2. It uses a threadpool to control the desired level of parallelism.

A word of caution: pmap will consume the entire input sequence and produce as much output as possible--if you pmap over (range) you'll get an Out of Memory Exception! Use this when you definitely want the work to be done.

The threadpool may be one of 4 things:

1. An ExecutorService, e.g. one created by threadpool.
2. An integer. In this case, a threadpool will be created, and it will be destroyed when all the pmap tasks are complete.
3. The keyword :builtin. In this case, pmap will use the Clojure built-in agent threadpool. For pmap, that's probably not what you want, as you'll likely create a thread per task.
4. The keyword :serial. In this case, the computations will be performed in serial via (doall map). This may be helpful during profiling, for example.

priority-threadpool clj

(priority-threadpool
  n
  &
  {:keys [default-priority] :as args :or {default-priority 0}})

Make a threadpool that chooses tasks based on their priorities.

Assign priorities to tasks by wrapping the pool with with-priority or with-priority-fn. You can also set a default priority with keyword argument :default-priority.

Otherwise, this uses the same keyword arguments as threadpool, and functions just like any other ExecutorService.

priority-threadpool? clj

(priority-threadpool? pool)

Returns true iff the argument is a priority-threadpool.

prun! clj

(prun! pool proc coll)

Like run!, but in parallel. Unlike the streaming sequence functions (e.g. pmap), prun! blocks until all the work is done.

pvalues clj macro

(pvalues pool & exprs)

Like clojure.core.pvalues, except it takes a threadpool. For more detail on its parallelism and on its threadpool argument, see pmap.

serial? clj

(serial? pool)

Check if we should run computations on this threadpool in serial.

shutdown clj

(shutdown pool)

Syntactic sugar to stop a pool cleanly. This will stop the pool from accepting any new requests.

shutdown! clj

(shutdown! pool)

Syntactic sugar to forcibly shutdown a pool. This will kill any running threads in the pool!

shutdown? clj

(shutdown? pool)

Syntactic sugar to test if a pool is shutdown.

thread-factory clj

(thread-factory & {:keys [daemon thread-priority] pool-name :name :as args})

Create a ThreadFactory with keyword options including thread daemon status :daemon, the thread name format :name (a string for format with one integer), and a thread priority :thread-priority.

This is exposed as a public function because it's handy if you're instantiating your own ExecutorServices.

threadpool clj

(threadpool n
            &
            {:keys [daemon thread-priority] pool-name :name :or {daemon true}})

Make a threadpool. It should be shutdown when no longer needed.

A threadpool is just an ExecutorService with a fixed number of threads. In general, you can use your own ExecutorService in place of any threadpool, and you can treat a threadpool as you would any other ExecutorService.

This takes optional keyword arguments: :daemon, a boolean indicating whether the threads are daemon threads, which will automatically die when the JVM exits, defaults to true) :name, a string giving the pool name, which will be the prefix of each thread name, resulting in threads named "name-0", "name-1", etc. Defaults to "claypoole-[pool-number]". :thread-priority, an integer in [Thread/MIN_PRIORITY, Thread/MAX_PRIORITY]. The effects of thread priority are system-dependent and should not be confused with Claypoole's priority threadpools that choose tasks based on a priority. For more info about Java thread priority see http://www.javamex.com/tutorials/threads/priority_what.shtml

Note: Returns a ScheduledExecutorService rather than just an ExecutorService because it's the same thing with a few bonus features.

threadpool? clj

(threadpool? pool)

Returns true iff the argument is a threadpool.

upcalls clj

(upcalls pool & fs)

Like clojure.core.pcalls, except it takes a threadpool and returns results ordered by completion time. For more detail on its parallelism and on its threadpool argument, see upmap.

upfor clj macro

(upfor pool bindings & body)

Like pfor, except the return value is a sequence of results ordered by completion time, not by input order.

upmap clj

(upmap pool f & arg-seqs)

Like pmap, except that the return value is a sequence of results ordered by completion time, not by input order.

upvalues clj macro

(upvalues pool & exprs)

Like clojure.core.pvalues, except it takes a threadpool and returns results ordered by completion time. For more detail on its parallelism and on its threadpool argument, see upmap.

with-priority clj

(with-priority pool priority)

Make a priority-threadpool wrapper with a given fixed priority.

All tasks run with this pool wrapper will have the given priority. e.g.

(def t1 (future (with-priority p 1) 1)) (def t2 (future (with-priority p 2) 2)) (def t3 (future (with-priority p 3) 3))

will use pool p to run these tasks with priorities 1, 2, and 3 respectively.

If you nest priorities, the outermost one "wins", so this task will be run at priority 3:

(def wp (with-priority p 1)) (def t1 (future (with-priority (with-priority wp 2) 3) :result))

with-priority-fn clj

(with-priority-fn pool priority-fn)

Make a priority-threadpool wrapper that uses a given priority function.

The priority function is applied to a pmap'd function's arguments. e.g.

(upmap (with-priority-fn pool (fn [x _] x)) + [6 5 4] [1 2 3])

will use pool to run tasks [(+ 6 1) (+ 5 2) (+ 4 3)] with priorities [6 5 4].

with-shutdown! clj macro

(with-shutdown! pool-syms-and-inits & body)

Lets a threadpool from an initializer, then evaluates body in a try expression, calling shutdown! on the threadpool to forcibly shut it down at the end.

The threadpool initializer may be a threadpool. Alternately, it can be any threadpool argument accepted by pmap, e.g. a number, :builtin, or :serial, in which case it will create a threadpool just as pmap would.

Be aware that any unfinished jobs at the end of the body will be killed!

Examples:

(with-shutdown! [pool (threadpool 6)]
  (doall (pmap pool identity (range 1000))))
(with-shutdown! [pool1 6
                 pool2 :serial]
  (doall (pmap pool1 identity (range 1000))))

Bad example:

(with-shutdown! [pool 6]
  ;; Some of these tasks may be killed!
  (pmap pool identity (range 1000)))