tech.core.parallel — tech/core.parallel 0.1.5

async-channel-to-lazy-seq clj

(async-channel-to-lazy-seq to-chan)

Convert a core-async channel into a lazy sequence where each item is read via async/<!!. Sequence ends when channel returns nil.

buffered-seq clj

(buffered-seq buffer-depth input-seq)

Given an input lazy sequence, realize up to N items ahead but produce the same sequence. Defaults to the original seq for buffer-depth of 0.

channel-seq->item-seq clj

(channel-seq->item-seq channel-seq)

Convert a sequence of channels lazily into a sequence of the first item read from a given channel.

create-next-item-fn clj

(create-next-item-fn item-sequence)

Given a sequence return a function that each time called (with no arguments) returns the next item in the sequence in a mutable fashion.

get-default-parallelism clj

(get-default-parallelism)

launch-parallel-for clj

(launch-parallel-for num-iters parallel-for-fn)

Given a function that takes exactly 2 arguments, a start-index and a length, call this function exactly N times where N is ForkJoinPool/getCommonPoolParallelism. Indexes will be split as evenly as possible among the invocations.

memoize clj

(memoize memo-fn)

order-indexed-sequence clj

(order-indexed-sequence index-fn original-sequence)

Given a possibly unordered original sequence and a function that returns indexes starting from 0 produce an ordered sequence.

parallel-for clj macro

(parallel-for idx-var num-iters & body)

Like clojure.core.matrix.macros c-for except this expects index that run from 0 -> num-iters. Idx is named idx-var and body will be called for each idx in parallel.

queued-pmap clj

(queued-pmap queue-depth map-fn & args)

Given a queue depth and a mapping function, run a pmap like operation.

Not for use with infinite sequences as the threads will hang around forever processing the infinite sequence. Call queued-sequence directly and use the shutdown-fn when the infinite sequence isn't necessary any more.

Note that there will possibly be queue-depth + 1 items in flight as the instant the first output item is dereferenced there is a chance for the processing threads to grab an item and both will be in flight, adding up to queue-depth + 1.

A queue depth of zero indicates to use a normal map operation.

queued-sequence clj

(queued-sequence map-fn
                 map-args
                 &
                 {:keys [queue-depth num-threads thread-init-fn
                         executor-service]
                  :or {queue-depth (get-default-parallelism)
                       num-threads (get-default-parallelism)
                       thread-init-fn nil}})

Returns a map containing a shutdown function and a sequence derived from the queue operation: {:shutdown-fn :sequence} Shutting down the sequence is necessary in the case of an infinite so you can free the resources associated with this queued sequence. When using ordering it does not make any sense to have num-threads > queue-depth because we cannot read more than queue-depth ahead into the src seq.

There is an additional invariant that there are never more that queue-depth items in flight. This invariant means there has to be blocking on the read-head of the input sequence.

When callers dereference the output sequence, however, there may at that instant be queue-depth + 1 items in flight. Callers need to be aware of this.

A thread initialization function is available in case you have an operation that needs to happen exactly once per thread.

Clients can dictate which executor service to use for the thread pool. If a service is not specified this function internally will allocate a forkjoinpool with num-threads parallelism.

If any errors leak into the processing thread the entire system is immediately halted and the error propagated to the calling thread.

require-resolve clj

Clojure's require is not threadsafe. So in order to do dynamic require and resolution of target functions we need to wrap it in a threadsafe memoizeation of the value. Usage: (require-resolve 'clojure.core.async/admix)