paysages3d/src/system/ParallelWork.cpp

163 lines
3.8 KiB
C++
Raw Normal View History

#include "ParallelWork.h"
#include "Thread.h"
#include "System.h"
#include "Semaphore.h"
#include "Mutex.h"
2014-08-18 10:17:16 +00:00
#include "ParallelWorker.h"
#include <cassert>
2014-08-18 10:17:16 +00:00
/**
* Compatibility class for code that uses ParallelUnitFunction.
*/
class ParallelWorkerCompat : public ParallelWorker {
public:
ParallelWorkerCompat(ParallelWork *work, ParallelWork::ParallelUnitFunction func, void *data)
: ParallelWorker(), work(work), func(func), data(data) {
2014-08-18 10:17:16 +00:00
}
virtual void processParallelUnit(int unit) {
func(work, unit, data);
2014-08-18 10:17:16 +00:00
}
private:
ParallelWork *work;
2014-08-18 10:17:16 +00:00
ParallelWork::ParallelUnitFunction func;
void *data;
2014-08-18 10:17:16 +00:00
};
/**
* Thread sub-class to perform units of work
*/
class ParallelWork::ParallelThread : public Thread {
public:
ParallelThread(ParallelWork *work) : Thread(), work(work), sem(0) {
interrupted = false;
unit = -1;
}
void feedUnit(int unit) {
this->unit = unit;
sem.release();
}
void interrupt() {
interrupted = true;
sem.release();
}
protected:
virtual void run() override {
while (unit >= 0 or not interrupted) {
// Wait for a unit (or interrupt)
sem.acquire();
// Process the unit
if (unit >= 0) {
work->worker->processParallelUnit(unit);
unit = -1;
2014-08-18 14:25:44 +00:00
work->returnThread(this);
}
}
}
private:
ParallelWork *work;
Semaphore sem;
int unit;
bool interrupted;
};
ParallelWork::ParallelWork(ParallelWorker *worker, int units) {
2014-08-18 10:17:16 +00:00
this->units = units;
this->running = 0;
this->worker = worker;
this->worker_compat = false;
}
ParallelWork::ParallelWork(ParallelUnitFunction func, int units, void *data) {
this->units = units;
this->running = 0;
2014-08-18 10:17:16 +00:00
this->worker = new ParallelWorkerCompat(this, func, data);
this->worker_compat = true;
}
ParallelWork::~ParallelWork() {
assert(not running);
if (worker_compat) {
2014-08-18 10:17:16 +00:00
delete worker;
}
}
int ParallelWork::perform(int thread_count) {
int i, done;
assert(not running);
// Get thread count
if (thread_count <= 0) {
2014-08-18 10:17:16 +00:00
thread_count = System::getCoreCount();
}
if (thread_count > PARALLEL_MAX_THREADS) {
2014-08-18 10:17:16 +00:00
thread_count = PARALLEL_MAX_THREADS;
}
this->thread_count = thread_count;
running = 1;
// Init threads
semaphore = new Semaphore(thread_count);
mutex = new Mutex();
threads = new ParallelThread *[thread_count];
available = new ParallelThread *[thread_count];
available_offset = 0;
available_length = thread_count;
for (i = 0; i < thread_count; i++) {
threads[i] = available[i] = new ParallelThread(this);
threads[i]->start();
}
// Perform all units
for (done = 0; done < units; done++) {
// Take first available thread
semaphore->acquire();
mutex->acquire();
ParallelThread *thread = available[available_offset];
available_offset++;
if (available_offset >= thread_count) {
available_offset = 0;
}
available_length--;
mutex->release();
// Feed the unit to it
thread->feedUnit(done);
}
// Wait for all threads to end, then cleanup
for (i = 0; i < thread_count; i++) {
threads[i]->interrupt();
}
for (i = 0; i < thread_count; i++) {
threads[i]->join();
delete threads[i];
}
delete[] threads;
delete[] available;
delete semaphore;
delete mutex;
running = 0;
return done;
}
void ParallelWork::interrupt() {
2014-08-19 07:18:55 +00:00
worker->interrupt();
}
void ParallelWork::returnThread(ParallelWork::ParallelThread *thread) {
mutex->acquire();
available[(available_offset + available_length) % thread_count] = thread;
available_length++;
semaphore->release();
mutex->release();
}