Class ProcessingService

java.lang.Object

org.ojalgo.concurrent.ProcessingService

public final class ProcessingService
extends Object

A simple wrapper around an ExecutorService that makes it easier to process collections of items in parallel. The work items are processed by a Consumer, Function or TwoStepMapper. In particular the TwoStepMapper can be used to aggregate/reduce data from the work items and then combine the collected data into a final result.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
(package private) static final class	ProcessingService.CallableConsumer<W>
(package private) static final class	ProcessingService.CallableMapper<W,R>

Field Summary

Fields

Modifier and Type	Field	Description
static final ProcessingService	INSTANCE
private final ExecutorService	myExecutor

Constructor Summary

Constructors

Constructor	Description
ProcessingService(ExecutorService executor)

Method Summary

Modifier and Type	Method	Description
<W,R> Map<W,R>	compute(Collection<W> work, int parallelism, Function<W,R> computer)	Compute an output item for each (unique) input item, and return the results as a Map.
<W,R> Map<W,R>	compute(Collection<W> work, Function<W,R> computer)	Using parallelism Parallelism.CORES.
<W,R> Map<W,R>	compute(Collection<W> work, IntSupplier parallelism, Function<W,R> computer)
DivideAndConquer.Divider	divider()	Deprecated. v56 Use newDivider() or ParallelismSupplier.newDivider(int) instead
ExecutorService	getExecutor()
<W,R> Collection<R>	map(Collection<W> work, int parallelism, Function<W,R> mapper)	Simply map each (unique) input item to an output item - a Collection of input results in a Collection of output.
<W,R> Collection<R>	map(Collection<W> work, Function<W,R> mapper)	Using parallelism Parallelism.CORES.
<W,R> Collection<R>	map(Collection<W> work, IntSupplier parallelism, Function<W,R> mapper)
DivideAndConquer.Divider	newDivider()
static ProcessingService	newInstance(String name)
<W> void	process(Collection<? extends W> work, int parallelism, Consumer<W> processor)	Will create at most parallelism tasks to work through the work items, processing them with processor.
<W> void	process(Collection<? extends W> work, Consumer<W> processor)	Using parallelism Parallelism.CORES.
<W> void	process(Collection<? extends W> work, IntSupplier parallelism, Consumer<W> processor)
<W> void	processPair(W work1, W work2, Consumer<W> processor)	Just 2 work items.
<W> void	processTriplet(W work1, W work2, W work3, Consumer<W> processor)	Just 3 work items.
<W,R> R	reduce(Collection<W> work, int parallelism, Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)	Deprecated. v54 Use invalid reference #reduceMergeable(Collection<W>,int,Supplier<? extends TwoStepMapper.Mergeable<W, R>>) instead
<W,R> R	reduce(Collection<W> work, IntSupplier parallelism, Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)	Deprecated. v54 Use invalid reference #reduceMergeable(Collection<W>,IntSupplier,Supplier<? extends TwoStepMapper.Mergeable<W, R>>) instead
<W,R> R	reduce(Collection<W> work, Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)	Deprecated. v54 Use invalid reference #reduceMergeable(Collection<W>,Supplier<? extends TwoStepMapper.Mergeable<W, R>>) instead
<W, R, A extends TwoStepMapper.Combineable<W,R,A>> R	reduceCombineable(Collection<W> work, int parallelism, Supplier<A> reducer)	Will create at most parallelism tasks to work through the work items, processing them with reducer.
<W, R, A extends TwoStepMapper.Combineable<W,R,A>> R	reduceCombineable(Collection<W> work, IntSupplier parallelism, Supplier<A> reducer)
<W, R, A extends TwoStepMapper.Combineable<W,R,A>> R	reduceCombineable(Collection<W> work, Supplier<A> reducer)	Using parallelism Parallelism.CORES.
<W, R, A extends TwoStepMapper.Mergeable<W,R>> R	reduceMergeable(Collection<W> work, int parallelism, Supplier<A> reducer)	Will create at most parallelism tasks to work through the work items, processing them with reducer.
<W, R, A extends TwoStepMapper.Mergeable<W,R>> R	reduceMergeable(Collection<W> work, IntSupplier parallelism, Supplier<A> reducer)
<W, R, A extends TwoStepMapper.Mergeable<W,R>> R	reduceMergeable(Collection<W> work, Supplier<A> reducer)	Using parallelism Parallelism.CORES.
void	run(int parallelism, Runnable processor)	Will create precisely parallelism tasks that each execute the processor.
void	run(Runnable task1, Runnable task2)
void	run(IntSupplier parallelism, Runnable processor)
<T> AtomicBoolean	take(BlockingQueue<T> queue, int parallelism, Consumer<T> processor)	Will submit precisely parallelism tasks that each take from the queue feeding the items to the processor.
<T> AtomicBoolean	take(BlockingQueue<T> queue, IntSupplier parallelism, Consumer<T> processor)

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

INSTANCE

public static final ProcessingService INSTANCE

myExecutor

private final ExecutorService myExecutor

Constructor Details

ProcessingService

public ProcessingService(ExecutorService executor)

Method Details

newInstance

public static ProcessingService newInstance(String name)

compute

public <W,R> Map<W,R> compute(Collection<W> work,
 Function<W,R> computer)

Using parallelism Parallelism.CORES.

See Also:

• compute(Collection, IntSupplier, Function)

compute

public <W,R> Map<W,R> compute(Collection<W> work,
 int parallelism,
 Function<W,R> computer)

Compute an output item for each (unique) input item, and return the results as a Map. If the input contains duplicates, the output will have fewer items. It is therefore vital that the input type implements Object.hashCode() and Object.equals(Object) properly.

Will create at most parallelism tasks to work through the work items, processing them with computer and collectiing the results in a Map.

Type Parameters:

W - The work item type

R - The function return type

Parameters:

work - The collection of work items

parallelism - The maximum number of concurrent workers that will process the work items

computer - The processing code

Returns:

A map of function input to output

compute

public <W,R> Map<W,R> compute(Collection<W> work,
 IntSupplier parallelism,
 Function<W,R> computer)

See Also:

• compute(Collection, int, Function)

divider

@Deprecated
public DivideAndConquer.Divider divider()

Deprecated.

v56 Use newDivider() or ParallelismSupplier.newDivider(int) instead

getExecutor

public ExecutorService getExecutor()

Returns:

The underlying ExecutorService

map

public <W,R> Collection<R> map(Collection<W> work,
 Function<W,R> mapper)

Using parallelism Parallelism.CORES.

See Also:

• map(Collection, IntSupplier, Function)

map

public <W,R> Collection<R> map(Collection<W> work,
 int parallelism,
 Function<W,R> mapper)

Simply map each (unique) input item to an output item - a Collection of input results in a Collection of output. If the input contains duplicates, the output will have fewer items. It is therefore vital that the input type implements Object.hashCode() and Object.equals(Object) properly.

Type Parameters:

W - The input item type

R - The output item type

Parameters:

work - The collection of work items

parallelism - The maximum number of concurrent workers that will process the work items

mapper - The mapper functiom

Returns:

The mapped results

map

public <W,R> Collection<R> map(Collection<W> work,
 IntSupplier parallelism,
 Function<W,R> mapper)

See Also:

• map(Collection, int, Function)

newDivider

public DivideAndConquer.Divider newDivider()

process

public <W> void process(Collection<? extends W> work,
 Consumer<W> processor)

Using parallelism Parallelism.CORES.

See Also:

• process(Collection, IntSupplier, Consumer)

process

public <W> void process(Collection<? extends W> work,
 int parallelism,
 Consumer<W> processor)

Will create at most parallelism tasks to work through the work items, processing them with processor.

Type Parameters:

W - The work item type

Parameters:

work - The collection of work items

parallelism - The maximum number of concurrent workers that will process the work items

processor - The processing code

process

public <W> void process(Collection<? extends W> work,
 IntSupplier parallelism,
 Consumer<W> processor)

See Also:

• process(Collection, int, Consumer)

processPair

public <W> void processPair(W work1,
 W work2,
 Consumer<W> processor)

Just 2 work items.

See Also:

• process(Collection, Consumer)

processTriplet

public <W> void processTriplet(W work1,
 W work2,
 W work3,
 Consumer<W> processor)

Just 3 work items.

See Also:

• process(Collection, Consumer)

reduce

@Deprecated
public <W,R> R reduce(Collection<W> work,
 int parallelism,
 Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)

Deprecated.

v54 Use

invalid reference

#reduceMergeable(Collection<W>,int,Supplier<? extends TwoStepMapper.Mergeable<W, R>>)

instead

reduce

@Deprecated
public <W,R> R reduce(Collection<W> work,
 IntSupplier parallelism,
 Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)

Deprecated.

v54 Use

invalid reference

#reduceMergeable(Collection<W>,IntSupplier,Supplier<? extends TwoStepMapper.Mergeable<W, R>>)

instead

reduce

@Deprecated
public <W,R> R reduce(Collection<W> work,
 Supplier<? extends TwoStepMapper.Mergeable<W,R>> reducer)

Deprecated.

v54 Use

invalid reference

#reduceMergeable(Collection<W>,Supplier<? extends TwoStepMapper.Mergeable<W, R>>)

instead

reduceCombineable

public <W, R, A extends TwoStepMapper.Combineable<W,R,A>>
R reduceCombineable(Collection<W> work,
 int parallelism,
 Supplier<A> reducer)

Will create at most parallelism tasks to work through the work items, processing them with reducer. The state of each task's reducer will be combined into a single instance, and the results of that instance will be returned.

Each TwoStepMapper.Combineable is only worked on by a single thread, and the results are combined into a single instance. The instances are not reused.

Parameters:

work - The collection of work items

parallelism - The maximum number of concurrent workers that will process the work items

reducer - A TwoStepMapper.Combineable implementation that does what you want.

Returns:

The results...

reduceCombineable

public <W, R, A extends TwoStepMapper.Combineable<W,R,A>>
R reduceCombineable(Collection<W> work,
 IntSupplier parallelism,
 Supplier<A> reducer)

See Also:

• reduceCombineable(Collection, int, Supplier)

reduceCombineable

public <W, R, A extends TwoStepMapper.Combineable<W,R,A>>
R reduceCombineable(Collection<W> work,
 Supplier<A> reducer)

Using parallelism Parallelism.CORES.

See Also:

• reduceCombineable(Collection, int, Supplier)

reduceMergeable

public <W, R, A extends TwoStepMapper.Mergeable<W,R>>
R reduceMergeable(Collection<W> work,
 int parallelism,
 Supplier<A> reducer)

Will create at most parallelism tasks to work through the work items, processing them with reducer. The results of each task's reducer will be merged into a single instance, and the results of that instance will be returned.

Each TwoStepMapper.Mergeable is only worked on by a single thread, and the results are combined into a single instance. The instances are not reused.

Parameters:

work - The collection of work items

parallelism - The maximum number of concurrent workers that will process the work items

reducer - A TwoStepMapper.Mergeable implementation that does what you want.

Returns:

The results...

reduceMergeable

public <W, R, A extends TwoStepMapper.Mergeable<W,R>>
R reduceMergeable(Collection<W> work,
 IntSupplier parallelism,
 Supplier<A> reducer)

See Also:

• reduceMergeable(Collection, int, Supplier)

reduceMergeable

public <W, R, A extends TwoStepMapper.Mergeable<W,R>>
R reduceMergeable(Collection<W> work,
 Supplier<A> reducer)

Using parallelism Parallelism.CORES.

See Also:

• reduceMergeable(Collection, int, Supplier)

run

public void run(int parallelism,
 Runnable processor)

Will create precisely parallelism tasks that each execute the processor.

Parameters:

parallelism - The number of concurrent workers/threads that will run

processor - The processing code

run

public void run(IntSupplier parallelism,
 Runnable processor)

See Also:

• run(int, Runnable)

run

public void run(Runnable task1,
 Runnable task2)

take

public <T> AtomicBoolean take(BlockingQueue<T> queue,
 int parallelism,
 Consumer<T> processor)

Will submit precisely parallelism tasks that each take from the queue feeding the items to the processor. The tasks will continue to run until the returned AtomicBoolean is set to false (or the thread is interrupted).

If the threads of the underlying ExecutorService are daemon threads, the JVM will not wait for them to finish before it exits. The default behaviour, using INSTANCE or newInstance(String), is to make use of ojAlgo's DaemonPoolExecutor.

Type Parameters:

T - The work item type

Parameters:

queue - The queue to take from

parallelism - How many parallel workers to create

processor - What to do with each of the work items

Returns:

A flag that can be used to signal the tasks to stop

take

public <T> AtomicBoolean take(BlockingQueue<T> queue,
 IntSupplier parallelism,
 Consumer<T> processor)

See Also:

• take(BlockingQueue, int, Consumer)