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.
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.
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.
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()
Removed 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 worker - index is >= 0 and < GetPoolSize(). This is useful if there is a need for per-thread resources. -1 means that thread is now worker (this can happen when Finish is using calling thread to perform jobs).
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).
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 depend on destructor to call it).
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.
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.
|
