Processing Queue Sets

ProcessingQueueSet

class ProcessingQueueSet

fluxEngine Processing Queue Set

This class represents a processing queue set. A processing queue set is used for processing queues.

The default constructor does not actually create an object, the user must use one of the other constructors to actually create and initialize a processing queue set.

Each processing queue set is associated with a number of processing threads that the user can manage. A processing queue set may only be used to process a single model at the same time, but multiple processing queue sets may be created.

Public Functions

ProcessingQueueSet() = default

Default constructor.

This does not actually create a processing queue set, but allows the user to create variables that can be used to move a handle around.

inline ProcessingQueueSet(Handle &handle)

Create a new processing queue set for a given handle.

This constructor creates a new processing queue set. The following example shows a typical usage: 

{
    fluxEngine::ProcessingQueueSet pqs(handle);
    // do something with processing queue set
}
// handle has fallen out of scope and is destroyed
The resulting processing queue set has move semantics, see the following example: 
fluxEngine::ProcessingQueueSet pqs(handle);
// do some initial things
someObject.setProcessingQueueSet(std::move(pqs));
If an error occurs, an exception will be thrown. The following exceptions may be thrown by this constructor:

  • std::bad_alloc

  • std::invalid_argument

  • Error

Parameters:

handle – A reference to the fluxEngine handle

inline ProcessingQueueSet(ProcessingQueueSet &&other) noexcept

Move constructor.

Parameters:

other – The processing queue set to move into the newly created object

inline ProcessingQueueSet &operator=(ProcessingQueueSet &&other) noexcept

Move assignment operator.

Parameters:

other – The processing queue set to move into this object

Returns:

A reference to this object

inline ~ProcessingQueueSet()

Destructor.

Destroy a processing queue set, freeing its resources. All background threads will be stopped in the same manner as if stopProcessingThreads() had been called.

Any processing context associated with this processing queue set will be marked as invalid and may hence not be used anymore. However, some memory associated with remaining processing contexts that have not been freed previous to a call to this method may still be in use until each remaining processing context is freed by the user.

inline void createProcessingThreads(int count)

Create processing threads.

fluxEngine may use parallization to speed up processing. In order to achieve this background threads must be created to run on additional CPU cores.

Calling this method is optional: by default processing will be single-threaded.

This method will start count - 1 threads when called, as the thread that asks for processing is always considered to be the first thread (with id 0). For example, if 4 is supplied to count this function will start 3 threads that run in the background. The thread that the user uses to call ProcessingContext::processNext() will be considered the thread with id 0, making processing use a total of 4 threads, which is the value supplied for count.

This method may only be called if there are currently no background threads associated with this processing queue set. Otherwise stopProcessingThreads() must be called first to change the number of threads.

Any processing context associated with this processing queue set that was created before a call to this method was made is marked as invalid and can only be destroyed, but not used anymore.

If an error occurs, an exception will be thrown. The following exceptions may be thrown by this method:

Parameters:

count – The number of threads to use for parallel processing (one less than this number will be created by this method, see the description for details)

template<typename T>
inline void createProcessingThreads(int count, T &&initFunction)

Create processing threads (extended version)

Please read the documentation of the primary overload of createProcessingThreads() for a general overview.

This extended method allows the user to supply a thread initialization function that will be called at the beginning of the newly created background threads. This allows the user to customize the thread properties (such as the thread priority or the CPU affinity) themselves.

This function will only return once all thread initialization functions have run.

The thread initialization functions are only called for the backgronud threads that are started by this method; this means that for a count of 4 the initialization function will be called in 3 background threads, and it is up to the user to alter the thread in which they call ProcessingContext::processNext() to process data with fluxEngine.

The threads will be created sequentially, the next thread being created only after the previous thread’s initialization function has completed. This allows the user to directly modify global data structures in the initialization functions without the need for locking.

An example call to this method could be something like this: 

handle.createProcessingThreads(4, [&] (int threadId, int threadCount) {
    (void) threadCount;
    // Run thread 1 on core 1, thread 2 on core 2, etc.
    int cpuCoreId = threadId;
    pinCurrentThreadCPUCoreWithId(cpuCoreId);
    setCurrentThreadPriority(ThreadPriority::High);
    // The thread 0 will be the thread that calls
    // processNext() on a processing context, and will have
    // to have these settings applied elsewhere
});
(The functions called within the initialization function are just example names. It would be up to the user to implement them.)

If the user detects an error condition during thread initialization and wants to abort, they should throw an exception in the initialization function they supplied, which will be propagated out of the call to this method.

Important: any attempt to call a method that accesses this handle inside the initialization functions will create a deadlock.

This method may only be called if there are currently no background threads associated with this processing queue set. Otherwise stopProcessingThreads() must be called first to change the number of threads.

Any processing context associated with this processing queue set that was created before a call to this method was made is marked as invalid and can only be destroyed, but not used anymore.

If an error occurs, an exception will be thrown. The following exceptions may be thrown by this method:

Parameters:

  • count – The number of threads to use for parallel processing (one less than this number will be created by this method, see the description for details)

  • initFunction – The thread initialization function. This may be a lambda function

inline void stopProcessingThreads() noexcept

Stop all existing processing threads.

This will stop any background threads that are currently associated with a given processing queue set. If processing is currently active on the processing queue set, it will be aborted, as if ProcessingContext::abort() had been called. In that case this method may take a bit of time, as abort operations are not immediate, and this method will wait until the abort has completed.

Any processing context associated with this processing queue set that was created before a call to this method was made is marked as invalid and can only be destroyed, but not used anymore.

This method is always successful: the only errors that could occur when calling this method would be non-recoverable.

This method may safely be called on an invalid processing queue set, that would have no effect.

inline explicit operator bool() const noexcept

Boolean conversion operator.

This allows the user to easily check if a variable of this type currently holds a valid processing queue set. For example: 

if (processingQueueSet) {
    // the processing queue set is valid
}