Merge pull request #178 from unknownbrackets/threadpool

Cut down on overhead from std::bind copies in the threadpool.

file/path.cpp

@@ -52,8 +52,7 @@ void PathBrowser::Navigate(const std::string &path) {
 			if (slash != std::string::npos)
 				path_ = path_.substr(0, slash + 1);
 		}
-	}
-	else {
+	} else {
 		if (path[1] == ':' && path_ == "/")
 			path_ = path;
 		else

file/path.h

@@ -19,7 +19,7 @@ public:
 	void GetListing(std::vector<FileInfo> &fileInfo, const char *filter = 0);
 	void Navigate(const std::string &path);
 
-	std::string GetPath() {
+	std::string GetPath() const {
 		if (path_ != "/")
 			return path_;
 		else
@@ -28,16 +28,16 @@ public:
 	std::string GetFriendlyPath() {
 		std::string str = GetPath();
 #if defined(BLACKBERRY)
-		char* home = getenv("PERIMETER_HOME");
-		if (!strncmp(str.c_str(), home, strlen(home))) {
+		char *home = getenv("PERIMETER_HOME");
+		if (home != NULL && !strncmp(str.c_str(), home, strlen(home))) {
 			str = str.substr(strlen(home));
 			str.insert(0, 1, '~');
 		}
 #elif defined(ANDROID)
 		// Do nothing
 #elif defined(__linux)
-		char* home = getenv("HOME");
-		if (!strncmp(str.c_str(), home, strlen(home))) {
+		char *home = getenv("HOME");
+		if (home != NULL && !strncmp(str.c_str(), home, strlen(home))) {
 			str = str.substr(strlen(home));
 			str.insert(0, 1, '~');
 		}

thread/threadpool.cpp

@@ -31,9 +31,9 @@ void WorkerThread::WaitForCompletion() {
 void WorkerThread::WorkFunc() {
 	mutex.lock();
 	started = true;
-	while(active) {
+	while (active) {
 		signal.wait(mutex);
-		if(active) { 
+		if (active) {
 			work_();
 			doneMutex.lock();
 			done.notify_one();
@@ -42,6 +42,35 @@ void WorkerThread::WorkFunc() {
 	}
 }
 
+LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) {
+	thread = new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this));
+	doneMutex.lock();
+	while(!started) { };
+}
+
+void LoopWorkerThread::Process(const std::function<void(int, int)> &work, int start, int end) {
+	mutex.lock();
+	work_ = work;
+	start_ = start;
+	end_ = end;
+	signal.notify_one();
+	mutex.unlock();
+}
+
+void LoopWorkerThread::WorkFunc() {
+	mutex.lock();
+	started = true;
+	while (active) {
+		signal.wait(mutex);
+		if (active) {
+			work_(start_, end_);
+			doneMutex.lock();
+			done.notify_one();
+			doneMutex.unlock();
+		}
+	}
+}
+
 ///////////////////////////// ThreadPool
 
 ThreadPool::ThreadPool(int numThreads) : numThreads(numThreads), workersStarted(false) {
@@ -50,13 +79,13 @@ ThreadPool::ThreadPool(int numThreads) : numThreads(numThreads), workersStarted(
 void ThreadPool::StartWorkers() {
 	if(!workersStarted) {
 		for(int i=0; i<numThreads; ++i) {
-			workers.push_back(std::make_shared<WorkerThread>());
+			workers.push_back(std::make_shared<LoopWorkerThread>());
 		}
 		workersStarted = true;
 	}
 }
 
-void ThreadPool::ParallelLoop(std::function<void(int,int)> loop, int lower, int upper) {
+void ThreadPool::ParallelLoop(const std::function<void(int,int)> &loop, int lower, int upper) {
 	int range = upper - lower;
 	if (range >= numThreads * 2) { // don't parallelize tiny loops (this could be better, maybe add optional parameter that estimates work per iteration)
 		lock_guard guard(mutex);
@@ -67,7 +96,7 @@ void ThreadPool::ParallelLoop(std::function<void(int,int)> loop, int lower, int
 		int chunk = range / numThreads;
 		int s = lower;
 		for (int i = 0; i < numThreads - 1; ++i) {
-			workers[i]->Process(std::bind(loop, s, s+chunk));
+			workers[i]->Process(loop, s, s+chunk);
 			s+=chunk;
 		}
 		// This is the final chunk.

thread/threadpool.h

@@ -17,20 +17,37 @@ public:
 	// wait for a submitted work item to be completed
 	void WaitForCompletion();
 
-private:
+protected:
+	WorkerThread(bool ignored) : active(true), started(false) {}
+	virtual void WorkFunc();
+
 	std::thread *thread; // the worker thread
 	::condition_variable signal; // used to signal new work
 	::condition_variable done; // used to signal work completion
 	::recursive_mutex mutex, doneMutex; // associated with each respective condition variable
 	volatile bool active, started;
-	std::function<void()> work_; // the work to be done by this thread
 
-	void WorkFunc();
+private:
+	std::function<void()> work_; // the work to be done by this thread
 
 	WorkerThread(const WorkerThread& other); // prevent copies
 	void operator =(const WorkerThread &other);
 };
 
+class LoopWorkerThread : public WorkerThread {
+public:
+	LoopWorkerThread();
+	void Process(const std::function<void(int, int)> &work, int start, int end);
+
+protected:
+	virtual void WorkFunc();
+
+private:
+	int start_;
+	int end_;
+	std::function<void(int, int)> work_; // the work to be done by this thread
+};
+
 // A thread pool manages a set of worker threads, and allows the execution of parallel loops on them
 // individual parallel loops are fully sequentialized to simplify synchronization, which should not 
 // be a problem as they should each use the entire system
@@ -40,17 +57,17 @@ public:
 	// don't need a destructor, "workers" is cleared on delete, 
 	// leading to the stopping and joining of all worker threads (RAII and all that)
 
-	void ParallelLoop(std::function<void(int,int)> loop, int lower, int upper);
+	void ParallelLoop(const std::function<void(int,int)> &loop, int lower, int upper);
 
 private:
 	const int numThreads;
-	std::vector<std::shared_ptr<WorkerThread>> workers;
+	std::vector<std::shared_ptr<LoopWorkerThread>> workers;
 	::recursive_mutex mutex; // used to sequentialize loop execution
 
 	bool workersStarted;
 	void StartWorkers();
 	
 	ThreadPool(const ThreadPool& other); // prevent copies
-	void operator =(const WorkerThread &other);
+	void operator =(const ThreadPool &other);
 };