Sha256: fec35797428fd59d6f670eb2c265ddbe8f27d93adbfe688b3257a8018f87bf0d
Contents?: true
Size: 1.73 KB
Versions: 9
Compression:
Stored size: 1.73 KB
Contents
#ifndef LIMONP_THREAD_POOL_HPP
#define LIMONP_THREAD_POOL_HPP
#include "Thread.hpp"
#include "BlockingQueue.hpp"
#include "BoundedBlockingQueue.hpp"
#include "Closure.hpp"
namespace limonp {
using namespace std;
//class ThreadPool;
class ThreadPool: NonCopyable {
public:
class Worker: public IThread {
public:
Worker(ThreadPool* pool): ptThreadPool_(pool) {
assert(ptThreadPool_);
}
virtual ~Worker() {
}
virtual void Run() {
while (true) {
ClosureInterface* closure = ptThreadPool_->queue_.Pop();
if (closure == NULL) {
break;
}
try {
closure->Run();
} catch(std::exception& e) {
XLOG(ERROR) << e.what();
} catch(...) {
XLOG(ERROR) << " unknown exception.";
}
delete closure;
}
}
private:
ThreadPool * ptThreadPool_;
}; // class Worker
ThreadPool(size_t thread_num)
: threads_(thread_num),
queue_(thread_num) {
assert(thread_num);
for(size_t i = 0; i < threads_.size(); i ++) {
threads_[i] = new Worker(this);
}
}
~ThreadPool() {
Stop();
}
void Start() {
for(size_t i = 0; i < threads_.size(); i++) {
threads_[i]->Start();
}
}
void Stop() {
for(size_t i = 0; i < threads_.size(); i ++) {
queue_.Push(NULL);
}
for(size_t i = 0; i < threads_.size(); i ++) {
threads_[i]->Join();
delete threads_[i];
}
threads_.clear();
}
void Add(ClosureInterface* task) {
assert(task);
queue_.Push(task);
}
private:
friend class Worker;
vector<IThread*> threads_;
BoundedBlockingQueue<ClosureInterface*> queue_;
}; // class ThreadPool
} // namespace limonp
#endif // LIMONP_THREAD_POOL_HPP
Version data entries
9 entries across 9 versions & 2 rubygems