posix和c11都实现的跨平台线程库

#include "ThreadPool.h" #include <cstdio> #include <Windows.h> namespace ThreadAPP { static volatile int pool_seq = 0; static MutexLock pool_seq_locker; ThreadPool::ThreadPool(int num) : closed(false) , poolSeq(0) , threadNum(num) , taskLocker() , taskCond() , waitCompleteCond() #ifdef USING_C11THREAD #endif { pool_seq_locker.lock(); this->poolSeq = ++pool_seq; pool_seq_locker.unlock(); printf("create thread pool %d\n", poolSeq); init(); } ThreadPool::~ThreadPool() { for (list<ThreadWrapper*>::iterator iter = threads.begin(); iter != threads.end(); iter++) { ThreadWrapper* tw = *iter; delete tw; } threads.clear(); while (!taskQueue.empty()) { printf("thread pool release task is not empty"); TaskWrapper* taskw = taskQueue.front(); taskQueue.pop(); delete taskw; } printf("delete thread pool %d\n", poolSeq); } void* ThreadPool::__run__thread(void *arg) { ThreadPool* pool = (ThreadPool*)arg; pool->process(); printf("thread run end..."); return NULL; } void ThreadPool::init() { taskLocker.lock(); for (int i = 0; i < threadNum; i++) { #ifdef USING_C11THREAD char name[16]; sprintf(name, "p-%d-t-%d", poolSeq, i + 1); threads.push_back(new ThreadWrapper(__run__thread, this, name)); #else pthread_t pid; pthread_create(&pid, NULL, __run__thread, this); pthread_detach(pid); char name[16]; sprintf(name, "p-%d-t-%d", poolSeq, i + 1); threads.push_back(new ThreadWrapper(pid, name)); #endif } taskLocker.unlock(); } ThreadWrapper* ThreadPool::findThreadWrapper() const { for (list<ThreadWrapper*>::const_iterator iter = threads.begin(); iter != threads.end(); iter++) { ThreadWrapper* tw = *iter; #ifdef USING_C11THREAD if (this_thread::get_id() == tw->getPid()) #else if (pthread_equal(pthread_self(), tw->getPid())) #endif { return tw; } } return NULL; } #ifndef USING_C11THREAD void ThreadPool::process() { TaskWrapper* taskw; while (true) { taskLocker.lock(); while (taskQueue.empty()) { ThreadWrapper* tw = findThreadWrapper(); if (closed) { if (tw != NULL) { threads.remove(tw); delete tw; } bool allClosed = threads.empty(); taskLocker.unlock(); if (allClosed) { delete this; } return; } tw->setIdle(true); waitCompleteCond.notify(); taskCond.wait(taskLocker); tw->setIdle(false); } taskw = taskQueue.front(); printf("process : %p\n", taskw); taskQueue.pop(); #ifndef USING_C11THREAD taskLocker.unlock(); #endif taskw->getTask()->run(); delete taskw; } } #else void ThreadPool::process() { TaskWrapper* taskw; while (true) { std::unique_lock<std::mutex> locker(taskLocker.getPthreadMutexByRef()); while (taskQueue.empty()) { ThreadWrapper* tw = findThreadWrapper(); if (closed) { if (tw != NULL) { threads.remove(tw); delete tw; } bool allClosed = threads.empty(); if (allClosed) { delete this; } return; } tw->setIdle(true); waitCompleteCond.notify(); #ifdef USING_C11THREAD taskCond.wait(locker); #else taskCond.wait(taskLocker); #endif tw->setIdle(false); } taskw = taskQueue.front(); taskQueue.pop(); #ifndef USING_C11THREAD taskLocker.unlock(); #endif taskw->getTask()->run(); delete taskw; } } #endif void ThreadPool::addTask(Task* task, bool needDeleteAfterRun) { TaskWrapper* taskw = new TaskWrapper(task, needDeleteAfterRun); taskLocker.lock(); taskQueue.push(taskw); #ifndef USING_C11THREAD taskCond.notify(); #else taskCond.notify(); #endif taskLocker.unlock(); } void ThreadPool::addTasks(vector<Task*>* tasks, bool needDeleteAfterRun) { taskLocker.lock(); for (unsigned int i = 0; i < tasks->size(); i++) { taskQueue.push(new TaskWrapper(tasks->at(i), needDeleteAfterRun)); } #ifndef USING_C11THREAD taskCond.notifyAll(); #else condTask.notify_all(); #endif taskLocker.unlock(); } void ThreadPool::shutdownAfterProcess() { if (!closed) { closed = true; taskCond.notifyAll(); } } #ifndef USING_C11THREAD void ThreadPool::waitComplete() { while (true) { taskLocker.lock(); if (taskQueue.empty()) { bool allIdle = true; for (list<ThreadWrapper*>::iterator iter = threads.begin(); iter != threads.end(); iter++) { ThreadWrapper* tw = *iter; if (!tw->isIdle()) { allIdle = false; break; } } if (allIdle) { taskLocker.unlock(); return; } } waitCompleteCond.wait(taskLocker); taskLocker.unlock(); } } #else void ThreadPool::waitComplete() { while (true) { std::unique_lock<std::mutex> locker(taskLocker.getPthreadMutexByRef()); if (taskQueue.empty()) { bool allIdle = true; for (list<ThreadWrapper*>::iterator iter = threads.begin(); iter != threads.end(); iter++) { ThreadWrapper* tw = *iter; if (!tw->isIdle()) { allIdle = false; break; } } if (allIdle) { return; } } waitCompleteCond.wait(locker); } } #endif }