class CoWork
This class is indented as general parallelization tool. Whenever jobs (e.g. loop iterations) are independent (they do not share any data between iterations), CoWork can be used to relatively easily spawn loop iterations over threads and thus over CPU cores. Note that previous statement does not preclude CoWork iterations to share data at all - sharing data using Mutex or similar serialization mechanisms still works. CoWork works with fixed-size thread pool, which is created during initialization phase (which first CoWork constructor is called). No more thread are created or destroyed during normal work. Nesting of CoWork instances is also possible. Thread pool is normally terminated when the main thread finishes.
No synchronization is required to access CoWork instances from various threads (CoWork is internally synchronized).
If an exception is thrown in worker thread, which is not handled by worker thread, it is caught and rethrown in CoWork thread in Finish routine. Any such exception also causes the Cancel of the CoWork. For this reason, CoWork destructor CAN throw exceptions and CoWork should be usually used as automatick (stack) variable. If you need CoWork that does not throw exceptions in destructor, use CoWorkNX.
Implementation notes: Current implementation has single global FIFO stack for 2048 scheduled jobs. When there is no slot available when scheduling the job, it is performed immediately by Do. Finish method has to wait until all jobs scheduled by CoWork instance are finished, while waiting it attempts to perform scheduled jobs from the same instance. That way work always progresses even if there is shortage of worker threads.
static bool TrySchedule(Function<void ()>&& fn)
static bool TrySchedule(const Function<void ()>& fn)
This is a low-level function that attempts to schedule fn to be executed by worker thread. Returns true if fn was scheduled, false if not (in case there is no slot left in scheduling stacks). Note that this function only schedules the function, the exact time of execution is unknown.
static void Schedule(Function<void ()>&& fn)
static void Schedule(const Function<void ()>& fn)
Similar to TrySchedule, but always schedules fn - even if it has to wait for slot to become available.
void Do(Function<void ()>&& fn)
void Do(const Function<void ()>& fn)
CoWork& operator&(const Function<void ()>& fn)
CoWork& operator&(Function<void ()>&& fn)
Schedules fn to be executed. All changes to data done before Do are visible in the scheduled code. The order of execution or whether the code is execute in another or calling thread is not specified. In certain situations (no scheduling slot available), Do can perform scheduled job immediately in calling thread.
int GetScheduledCount() const
Returns a number of remaining unfinished jobs scheduled by Do in this CoWork.
static void FinLock()
This functions is to be called in scheduled routine. Its purpose is to serialize access to shared data at the end of the routine. The rationale is that CoWork has to lock some mutex anyway after scheduled code finishes, so FinLock can lock this mutex a bit earlier, joining two mutex locks into single one. Of course, as with all locks, execution of locked code should be short as not to cause congestion of CoWork scheduling.
void Cancel()
Removes all jobs scheduled by this thread that has not started yet from the queue and then waits for any jobs already started to finish. Rethrows the exception thrown in worker threads. If more than single worker thread throws the exception, the first exception thrown in absolute time is rethrown.
static bool IsCanceled()
This methods returns true in worker thread when the worker thread is a part of some CoWork instance and that instance was canceled.
void Finish()
Waits until all jobs scheduled using Do (or operator&) are finished. All changes to data performed by scheduled threads are visible after Finish. While waiting, Finish can perform scheduled jobs. Can
bool IsFinished()
Checkes whether all jobs scheduled using Do (or operator&) are finished. All changes to data performed by scheduled threads are visible after IsFinished returns true (so this is basically non-blocking variant of Finish).
void Reset()
Calls Cancel, catches and ignores all exceptions eventually thrown by worker threads. Then resets CoWork to the initial state as if it was just constructed. Useful when using CoWork as nonlocal variable.
~CoWork()
Calls Finish(). Can eventually rethrow worker thread exception. If there is a chance of destructor being involved in stack unwinding, Finish should be called separately before destructor.
static bool IsWorker()
Returns true if current thread is CoWork worker thread.
static int GetWorkerIndex()
Returns the index of current worker thread - index is >= 0 and < GetPoolSize(). This is useful if there is a need for per-thread resources. -1 means that thread is not worker.
static int GetPoolSize()
Returns the current count of worker threads.
static void SetPoolSize(int n)
Adjusts the thread pool size (default pool size is CPU_Cores() + 2).
void Loop(Function<void ()>&& fn)
void Loop(const Function<void ()>& fn)
CoWork& operator*(const Function<void ()>& fn)
CoWork& operator*(Function<void ()>&& fn)
Schedules fn to be run on all worker threads and on calling thread. After the first thread returns from fn, all other scheduled fn jobs that has not started yet are unscheduled. Waits for all started jobs to finish. The function also sets internal index counter to zero in CoWork before starting any worker thread. Worker thread should acquire job quantum in internal loop - internal CoWork index can be used for this purpose. Deprecated - use CoDo instead.
int Next()
Atomically increments internal index counter and returns its previous value (thus the first value returned is 0). Supposed to be used with Loop. Deprecated - use std::atomic<int> variable with CoDo instead.
struct CoWorkNX : public CoWork
This simple helper class just removes "noexcept(false)" from CoWork destructor, removing exception throws from the destructor and allowing it to be used as member variable. Obviously, if exception handling is required, you need to use Finish (not to depend on destructor to call it).
Loop parallelisation functions
|
|
void CoDo(Function<void ()>&& fn)
Schedules fn to be run on all worker threads and on calling thread. After the first thread returns from fn, all other scheduled fn jobs that has not started yet are unscheduled. Waits for all started jobs to finish. Worker thread should acquire job quantum in internal loop - usually std::atomic is used for this purpose. This is the most effective and flexible way to parallelise iteration.
void CoDo_ST(Function<void ()>&& fn)
This function simply calls fn. It is diagnostics tool - it allows to change CoDo parallel iteration into serial one by adding "_ST" text.
void CoDo(bool co, Function<void ()>&& fn)
If co is true, calls CoDo(fn), otherwise CoDo_ST(fn). This allows to parametrize algorithms with respect to parallelization.
template <typename Fn> void CoFor(int n, Fn iterator)
Based on CoDo, runs in parallel iterator for values 0..n passing the value as argument.
template <typename Fn> void CoFor_ST(int n, Fn iterator)
Single threaded variant of CoFor, for diagnostics purposes.
template <typename Fn> void CoFor(bool co, int n, Fn iterator)
Calls CoFor(n, iterator) if co is true, CoFor_ST(n, iterator) otherwise. This allows to parametrize algorithms with respect to parallelization.
template <class T>
class CoWorkerResources
This is a simple helper class that provides per-worker resources. For example, certain calculation requires the instance of computation model that is not immutable, but can be reused over iterations. In single-threaded code, a single instance of such model would be used over the whole loop, however in multi-threaded code, each worker thread, plus thread that created CoWork need its own instance. CoWorkerResources helps to manage such situation.
NOTE: The problem that this helper class solves is in majority cases better solved by using CoDo function.
CoWorkerResources()
Creates a required number of instances so that each sub-job of CoWork has its unique instance.
CoWorkerResources(Event<T&> initializer)
Creates a required number of instances so that each sub-job of CoWork has its unique instance.and initializes them using initializer.
int GetCount() const
Returns the number of instances. Note that this is equal to CoWork::GetPoolSize() if thread that created CoWorkerResources is itself CoWork worker, or CoWork::GetPoolSize() + 1 if it is any other thread. The reason for this is that CoWork owner can execute CoWork jobs too (while waiting in Finish).
T& operator[](int i)
Returns instance i. Together with GetCount can be used to initialize resources (alternative method to using constructor with initializer.
T& Get()
T& operator~()
Supposed to be called in CoWork job, returns a reference to resources unique for the thread.
T *begin()
T *end()
Standard iterator access.
|
