BGI/SOAPsnp/CThreadPool.cpp

   1 #include <stdlib.h>
   2 #include "CThreadPool.h"
   3 #include <iostream>
   4
   5 void CTask::SetData(void * data)
   6 {
   7     m_ptrData = data;
   8 }
   9
  10 void CTask::releaseData()
  11 {
  12         if (m_ptrData != NULL)
  13                 delete m_ptrData;
  14         m_ptrData = NULL;
  15 }
  16 TaskList CThreadPool::m_TaskList;         //任务列表
  17 bool CThreadPool::shutdown = false;
  18
  19 pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
  20 pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;
  21 int CThreadPool::m_task_count = 0;
  22
  23 /**
  24  * 线程池管理类构造函数
  25  */
  26 CThreadPool::CThreadPool(int threadNum)
  27 {
  28     this->m_iThreadNum = threadNum;
  29     Create();
  30 }
  31
  32 /**
  33  * 线程回调函数
  34  */
  35 void* CThreadPool::ThreadFunc(void* threadData)
  36 {
  37     pthread_t tid = pthread_self();
  38     while (1)
  39     {
  40         pthread_mutex_lock(&m_pthreadMutex);
  41         while (m_TaskList.size() == 0 && !shutdown)
  42         {
  43             pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);
  44         }
  45
  46         if (shutdown)
  47         {
  48             pthread_mutex_unlock(&m_pthreadMutex);
  49             pthread_exit(NULL);
  50         }
  51
  52         TaskList::iterator iter = m_TaskList.begin();
  53
  54         /**
  55         * 取出一个任务并处理之
  56         */
  57         CTask* task = *iter;
  58         if (iter != m_TaskList.end())
  59         {
  60             task = *iter;
  61                         m_TaskList.pop_front();
  62         }
  63
  64         pthread_mutex_unlock(&m_pthreadMutex);
  65
  66         task->Run(); /** 执行任务 */
  67         pthread_mutex_lock(&m_pthreadMutex);
  68                 --m_task_count;
  69         pthread_mutex_unlock(&m_pthreadMutex);
  70     }
  71     return (void*)0;
  72 }
  73
  74 /**
  75  * 往任务队列里边添加任务并发出线程同步信号
  76  */
  77 int CThreadPool::AddTask(CTask *task)
  78 {
  79     pthread_mutex_lock(&m_pthreadMutex);
  80         this->m_TaskList.push_back(task);
  81         ++m_task_count;
  82     pthread_mutex_unlock(&m_pthreadMutex);
  83     pthread_cond_signal(&m_pthreadCond);
  84
  85     return 0;
  86 }
  87
  88 /**
  89  * 创建线程
  90  */
  91 int CThreadPool::Create()
  92 {
  93     pthread_id = (pthread_t*)malloc(sizeof(pthread_t) * m_iThreadNum);
  94     for(int i = 0; i < m_iThreadNum; i++)
  95     {
  96         pthread_create(&pthread_id[i], NULL, ThreadFunc, NULL);
  97     }
  98     return 0;
  99 }
 100
 101 /**
 102  * 停止所有线程
 103  */
 104 int CThreadPool::StopAll()
 105 {
 106     /** 避免重复调用 */
 107     if (shutdown)
 108     {
 109         return -1;
 110     }
 111     /** 唤醒所有等待线程，线程池要销毁了 */
 112     shutdown = true;
 113     pthread_cond_broadcast(&m_pthreadCond);
 114
 115     /** 阻塞等待线程退出，否则就成僵尸了 */
 116     for (int i = 0; i < m_iThreadNum; i++)
 117     {
 118         pthread_join(pthread_id[i], NULL);
 119     }
 120
 121     free(pthread_id);
 122     pthread_id = NULL;
 123
 124     /** 销毁条件变量和互斥体 */
 125     pthread_mutex_destroy(&m_pthreadMutex);
 126     pthread_cond_destroy(&m_pthreadCond);
 127
 128     return 0;
 129 }
 130
 131 /**
 132  * 获取当前队列中任务数
 133  */
 134 int CThreadPool::getTaskSize()
 135 {
 136     return m_task_count;
 137 }