com.biomatters.geneious.publicapi.utilities

Class MultiThreadedProcessingSystem<T>

java.lang.Object

com.biomatters.geneious.publicapi.utilities.MultiThreadedProcessingSystem<T>

Type Parameters:

T -

public abstract class MultiThreadedProcessingSystem<T>
extends Object

A utility class providing a system in which a number of elements are each independently processed in some way (in multiple threads) before calling finishedProcessing on each element in the same order as they originally came in.

Usage:

1) Call add(Object) for each entry. This is not thread safe.

2) The MultiThreadedProcessingSystem will invoke process(Object) for each entry, potentially out of order if numberOfProcessingThreads>1 and potentially in a different thread from add(Object)

3) Then the MultiThreadedProcessingSystem will invoke finishedProcessing(Object) for each entry, guaranteed to be in the same order as they were added but potentially in a different thread from add(Object)

4) Since the calls to finishedProcessing are guaranteed to be called in order (via synchronization), these results can be passed on to another chained MultiThreadedProcessingSystem via add(Object) without synchronization

Example:

    final AtomicLong totalLength = new AtomicLong();
    MultiThreadedProcessingSystem<NucleotideSequenceDocument> processingSystem = new MultiThreadedProcessingSystem<NucleotideSequenceDocument>("testProcess",10,10,Runtime.getRuntime().availableProcessors()) {

Since:

API 4.600 (Geneious 6.0.0)

See Also:

MultiThreadedTask

Constructor Summary

Constructors

Constructor and Description
MultiThreadedProcessingSystem(String threadNamePrefix, int maxBacklogSize, int valuesPerBacklogEntry, int numberOfProcessingThreads)

Method Summary

Methods

Modifier and Type	Method and Description
void	add(T entry) Adds a new entry to be processed.
protected void	batchProcess(List<T> values) Processes multiple values in a single thread.
void	finishedProcessing(T entry) An implementation may optionally override this to perform any additional cleanup.
abstract void	process(T t) Processes a single entry previously added using add(Object)
void	setThresholdToStartThreads(int thresholdToStartThreads) Sets a threshold for the number of items to be added before this processing systems starts creating additional threads.
void	waitForCompletion() Waits until all items added using add(Object) have been processed by process(Object) and have returned from finishedProcessing(Object)

Methods inherited from class java.lang.Object

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

Constructor Detail

MultiThreadedProcessingSystem

public MultiThreadedProcessingSystem(String threadNamePrefix,
                             int maxBacklogSize,
                             int valuesPerBacklogEntry,
                             int numberOfProcessingThreads)

Parameters:

threadNamePrefix - a prefix for the name to use for any threads created

maxBacklogSize - the maximum number of batches of valuesPerBacklogEntry entries to store in memory ready awaiting processing. add(Object) will block as long as this threshold is reached.

valuesPerBacklogEntry - the number of entries added using add(Object) to process in a block. Up to this many entries will passed to batchProcess(java.util.List)

numberOfProcessingThreads - the number of threads to use for processing once a minimum number of items have been added as specified by setThresholdToStartThreads(int). You may want to use Runtime.getRuntime().availableProcessors()

Method Detail

setThresholdToStartThreads

public void setThresholdToStartThreads(int thresholdToStartThreads)

Sets a threshold for the number of items to be added before this processing systems starts creating additional threads. The default value is 1000. This is to prevent thread overhead from impacting performance on small cases where it is unnecessary.

Parameters:

thresholdToStartThreads - the number of items to be be added before this processing systems starts creating additional threads

add

public void add(T entry)
         throws DocumentOperationException

Adds a new entry to be processed. This entry may be processed in this or a background thread. This method may block if there are too many entries awaiting processing as defined in the constructor.

Parameters:

entry - the entry to be processed.

Throws:

DocumentOperationException - if the processing method (process(Object) throws a DocumentOperationException

batchProcess

protected void batchProcess(List<T> values)
                     throws DocumentOperationException

Processes multiple values in a single thread. A subclass my optionally override this to allow improved performance. The default implementation just calls process(Object) for each value.

Parameters:

values - all the values to process

Throws:

DocumentOperationException - if there is a problem processing any of the entries

process

public abstract void process(T t)
                      throws DocumentOperationException

Processes a single entry previously added using add(Object)

Parameters:

t - the entry to process

Throws:

DocumentOperationException - if there is a problem processing this entry. In this case this exception will be thrown by either add(Object) or waitForCompletion()

finishedProcessing

public void finishedProcessing(T entry)
                        throws DocumentOperationException

An implementation may optionally override this to perform any additional cleanup. This is guaranteed to be called for each entry in the same order as they were originally added.

Parameters:

entry - the entry that has been processed.

Throws:

DocumentOperationException - if there is a problem completing processing of the entry

waitForCompletion

public void waitForCompletion()
                       throws DocumentOperationException

Waits until all items added using add(Object) have been processed by process(Object) and have returned from finishedProcessing(Object)

Throws:

DocumentOperationException - if either process(Object) or finishedProcessing(Object) threw a DocumentOperationException