X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/a4175b2e216a20b86cc872dea8a08005c60617a5..be7cf844f59fa483423724e8e4b5e663e5b88ddd:/rpc/thr_pool.cc

diff --git a/rpc/thr_pool.cc b/rpc/thr_pool.cc
index 26226cd..7d3cf7d 100644
--- a/rpc/thr_pool.cc
+++ b/rpc/thr_pool.cc
@@ -1,60 +1,33 @@
 #include "thr_pool.h"
-#include <stdlib.h>
-#include <errno.h>
-#include "lang/verify.h"
 
-static void
-do_worker(void *arg)
-{
-	ThrPool *tp = (ThrPool *)arg;
-	while (1) {
-		ThrPool::job_t j;
-		if (!tp->takeJob(&j))
-			break; //die
-
-		(void)(j.f)(j.a);
-	}
-}
-
-//if blocking, then addJob() blocks when queue is full
-//otherwise, addJob() simply returns false when queue is full
-ThrPool::ThrPool(int sz, bool blocking)
-: nthreads_(sz),blockadd_(blocking),jobq_(100*sz) 
-{
-	for (int i = 0; i < sz; i++) {
-        th_.push_back(std::thread(do_worker, this));
-	}
+// if blocking, then addJob() blocks when queue is full
+// otherwise, addJob() simply returns false when queue is full
+ThrPool::ThrPool(size_t sz, bool blocking)
+: nthreads_(sz),blockadd_(blocking),jobq_(100*sz) {
+	for (size_t i=0; i<nthreads_; i++)
+        th_.emplace_back(&ThrPool::do_worker, this);
 }
 
-//IMPORTANT: this function can be called only when no external thread 
-//will ever use this thread pool again or is currently blocking on it
-ThrPool::~ThrPool()
-{
-	for (int i = 0; i < nthreads_; i++) {
-		job_t j;
-		j.f = (void (*)(void *))NULL; //poison pill to tell worker threads to exit
-		jobq_.enq(j);
-	}
+// IMPORTANT: this function can be called only when no external thread 
+// will ever use this thread pool again or is currently blocking on it
+ThrPool::~ThrPool() {
+	for (size_t i=0; i<nthreads_; i++)
+		jobq_.enq(job_t());
 
-	for (int i = 0; i < nthreads_; i++) {
+	for (size_t i=0; i<nthreads_; i++)
         th_[i].join();
-	}
 }
 
-bool 
-ThrPool::addJob(void (*f)(void *), void *a)
-{
-	job_t j;
-	j.f = f;
-	j.a = a;
-
+bool ThrPool::addJob(const job_t &j) {
 	return jobq_.enq(j,blockadd_);
 }
 
-bool 
-ThrPool::takeJob(job_t *j)
-{
-	jobq_.deq(j);
-	return (j->f!=NULL);
+void ThrPool::do_worker() {
+    job_t j;
+	while (1) {
+        jobq_.deq(&j);
+		if (!j)
+			break;
+		j();
+	}
 }
-