tech.parallel — techascent/tech.parallel 2.8

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.

create-thread-pool clj

(create-thread-pool num-threads)

(create-thread-pool num-threads thread-name-fn)

Create a new thread pool. This is often done as a variable defined in a namespace and then used in several thread-pool-pmap function calls. The advantage is that multiple pmap sequences can use the same underlying thread pool.

num-threads - Number of threads to create in the thread pool. thread-name-fn - Function from pool-index->string thread name

get-default-parallelism clj

(get-default-parallelism)

indexed-map-reduce clj

(indexed-map-reduce num-iters indexed-map-fn)

(indexed-map-reduce num-iters indexed-map-fn reduce-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. Uses ForkJoinPool/commonPool for parallelism. The entire list of results (outputs of indexed-map-fn) are passed to reduce-fn; reduce-fn is called once. When called with 2 arguments the reduction function is dorun

launch-parallel-for clj

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

(launch-parallel-for num-iters indexed-map-fn reduce-fn)

Legacy name. See indexed-map-reduce

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)

Run a side effecting operator over a contiguous set of indexes

psink! clj

(psink! doseq-fn! item-seq)

(psink! doseq-fn!
        item-seq
        {:keys [n-threads thread-name]
         :or {n-threads 16 thread-name "psink-thread"}})

Terminate a sequence with a side effecting function using n-threads named thread-name-{idx}. !!Items are satisifed in a non-order-dependent fashion!! Blocks until execution of the items is complete. :n-threads - num thread to use. :thread-name - thread name prefix to use. Is appended with '-{thread-idx}'

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)

serial-for clj macro

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

Run a side effecting operator over a contiguous set of indexes

thread-pool-num-threads clj

(thread-pool-num-threads thread-pool)

thread-pool-pmap clj

(thread-pool-pmap thread-pool map-fn & args)

Given a thread pool, do the simplest possible pmap operation over the pool. This consumes the source sequence greedily and returns a lazy sequence of the result.

thread-pool-queued-pmap clj

(thread-pool-queued-pmap thread-pool queue-depth map-fn & sequences)

Pmap functionality with preexisting thread pool. Convenience wrapper around using custom thread pools (ExecutorService). thread-pool - thread pool to use. queue-depth - Queue-depth - how far ahead to allow threads to get. map-fn - map fn args - sequences