基于pthread的线程池实现

Posted 2020-06-25

最近在做项目的过程中，需要使用线程池来实现任务的异步处理。即线程池中包含提前创建好的线程，客户将任务提交到线程池中，线程池中的线程对任务进行获取并执行。针对项目所使用的pthread线程库，我们设计与实现了一个简单的线程池。

在介绍线程池的实现之前，首先整理一下pthread库的一些接口。pthread是由POSIX提出的线程实现，广泛地被各种unix平台所支持。创建线程的接口为

 

pthread_create(pthread_t* thread, const pthread_attr_t* attr, void *(func)(void*), void* arg)

为了进行线程之间的同步，pthread提供了锁和条件变量机制，分别为pthread_mutex_t和pthread_cond_t及其相关函数。两者结合，可用于实现操作系统经典的管程机制。其中，最关键的函数为

pthread_cond_wait(pthread_cond_t* cond, pthread_mutex_t* lock)

 

这个函数必须在线程持有锁lock的时候才可以调用。调用这个函数之后，调用线程原子地放弃锁并阻塞到条件变量cond上，从而防止该操作进行中有其它线程改变条件。待线程被唤醒后，线程会重新参与到锁lock的竞争中，一量竞争到锁，线程从该函数调用的下一条语句开始执行。值得注意的是此时线程已重新获取到锁。
线程退出是通过调用

pthread_cancel(pthread_t pid)

向线程发送信号来实现的。信号是unix提供的一种进程或线程间通信的机制，当一个线程收到信号后，该信号对应找域会被进行标记，线程在从系统态退出到用户态的过程中会处理收到的信号。另外，如果线程阻塞在某些系统调用如read，write上时，该线程会被唤醒并开始进行信号处理，信号处理完成后如果未退出一般会直接退出系统调用并返回EINTR，有的操作系统选择将被中断的系统调用重启。用于终止线程的信号为实时信号，即该信号收到的次数与发送的次数相同，且信号处理具有较高的优先级。线程收到信号后的退出式有两种，一种是异步退出，即线程收到信号后会立即退出，而不论执行到什么位置。另一种为延迟退出，即收到终止信号后会进行标记，待到特定的执行点后再退出。POSIX给出了一组必须包含执行点的函数，这些函数的处理方式一般如下：

//检查终止标记，如果已标记，立即退出, 原子地设置退出模式为异步退出
//正常的函数处理，包括阻塞的系统调用
//检查终止标记，如果已标记，立即退出, 原子地设置退出模式为延迟退出

关于pthread的具体接口请参阅《UNIX环境高级编程》一书。但关于线程退出，有几点需要强调：

1. 一个线程可以使用pthread_cleanup_push来注册一个线程退出时执行的函数，该函数不包含检查点。因此，将该函数调用作为线程执行的第一个语句可以保证该函数被注册。
2. 在C++中，当一个线程退出（包括被cancel和调用pthread_exit）时，已构造的栈上对象的析构函数保证会被调用。但如果该对象未完全构造完成则不会。 
3. pthread_mutex_lock函数中不一定包含检查点（至少linux是这个样子）。这样，如果线程阻塞在锁的获取上，在收到pthread_cancel信号时不一定会被唤醒。因此，任一线程在退出时，必须释放其持有的所有锁已避免其它线程死锁。由于线程退出保证对象析构函数的调用，因此最好将锁的获取封装到对象的构造函数中：

1 class MutexGetLock{
2 public:
3     MutexGetLock(pthread_mutex_t* _lock):lock(_lock)   {pthread_mutex_lock(lock);};
4     ~MutexGetLock(){pthread_mutex_unlock(lock);};
5 private:
6     pthread_mutex_t* lock;
7 };

获取锁可以使用

MutexGetLock get_lock(&lock);

但要注意在该类的构造函数中不要添加输入输出语句如printf，因为读写操作都包含检查点，可能导致在该对象部分构造的时候线程退出，从而析构函数无法执行，造成死锁。

下面开始描述线程池的实现。实现线程池所需要的一个基本组件是阻塞队列，用于客户提交任务的线程池中的线程获取任务。在不设置队列中元素上限的情况下，阻塞队列可以用一个锁与一个条件变量实现:

class BlockingQueue{
    public:
        T* dequeue(){
            MutexGetLock get_lock(lock);
            if(没有元素){
                pthread_cond_wait(empty, lock);
            }    
            //获取元素
            pthread_mutex_unlock(lock);
        }
        void enqueue(){
            MutexGetLock get_lock(lock);
            //插入元素
            pthread_cond_signal(empty);    
        }
    private:
        pthread_mutex_t* lock;    
        pthread_cond_t* empty;
    };

线程池的接口如下：

 1 class ThreadPool{
 2     public:
 3         ThreadPool(int core_size, int max_size, BlockingQueue<Runnable*>* queue = 0);
 4         void submit(Runnable* r);
 5         void shutdown();
 6         void shutdownImmidiately();
 7         void shutdownAbruptly();
 8         ~ThreadPool();
 9         friend class WorkerExiter;
10         friend class WorkerRunnable;
11     private:
12         enum POOL_STATE{
13             INIT, RUNNING, SHUTDOWN, SHUTDOWN_IMMI, ABRUPT, STOPPED
14         };
15 
16         int core_size;
17         int max_size;
18         
19         BlockingQueue<Runnable*>* queue;
20         int current_size;
21         unordered_map<int, DaemonThread*>* workers;
22         
23         int next_wid;
24 
25         //state
26         POOL_STATE state;
27         pthread_mutex_t main_lock; //guard state
28         pthread_cond_t term_cond;
29         
30 
31         //exit
32         void onWorkerExit(int wid);
33         DaemonThread* addThread();
34     };

其中DaemonThread是对pthread的面向对象的简单封装。调用线程池的所有函数都必须先获取main_lock，因此线程池是线程安全的。onWorkerExit是在线程池中的线程退出时调用的，用于保持线程池状态的一致性。term_cond用于其它线程等待线程池终止。线程提供了三种终止方式：

1. shutdown: 执行完当前队列中所有任务后退出
2. shutdownImmidiately: 如果当前线程未执行任务，立即退出，否则执行完当前任务后退出。
3. shutdownAbruptly: 所有线程立即退出，无论执行到什么位置。
另外，线程池的扩张策略为开始有core_size个线程，一旦添加任务时发现队列中的任务数等于当前线程数，就创建一个新线程，但线程最多不会超过max_size。
由于线程池的所有操作均要先获取main_lock，所以均是原子操作，因此，线程池中最关键的是线程池的正确退出，以下将重点说明线程池的退出机制。
线程池的worker线程结构如下：

 1 class WorkerRunnable : public Runnable{
 2     public:
 3         WorkerRunnable(ThreadPool* _pool, int _wid, BlockingQueue<Runnable*>* _queue):
 4             pool(_pool), wid(_wid), queue(_queue), run_lock(PTHREAD_MUTEX_INITIALIZER){};
 5         void run();
 6         void setThread(DaemonThread* _thread){thread = _thread;};
 7         void cancelIfIdel();
 8         void cancelNow();
 9         ~WorkerRunnable(){};
10     private:
11         ThreadPool* pool;    
12         int wid;
13         BlockingQueue<Runnable*>* queue;
14         DaemonThread* thread;
15         pthread_mutex_t run_lock; //If thread is running, lock run_lock, then pool can tell it with try_lock
16         Runnable* getTask();
17     };

其中，最关键的是run_lock。线程在从queue中获取一个新任务后，会首先获取锁，再执行该任务。这样，其它线程便可以使用pthread_try_lock来获取该线程是否在执行任务，并且在完成针对该线程的操作之前，该线程无法正式开始一个新任务。cancelIfIdel即采用这样的策略实现:

void WorkerRunnable::cancelIfIdel{
        if(pthread_try_lock(&run_lock)){
            pthread_cancel(thread->getPid());
        }    
    }

线程池的三种退出逻辑实现如下：

1. shutdown: 设置线程池状态，禁止submit调用提交新任务。然后，向任务队列中插入current_size个特殊的退出任务EXIT_TASK，工作线程一旦读取到该任务后就会退出，因此，可以保证所有线程恰如获取一个退出任务然后退出。然后，调用shutdown的线程阻塞到term_cond条件变量上等待线程池退出。由于每个线程退出时都会调用onWorkerExit，如果某个工作线程退出时发现自已已经是最后一个线程，它就会唤醒阻塞到term_cond上的所有线程。
2. shutdownImmidiately: 设置线程池状态禁止提交新任务。然后调用cancelIfIdel来通知所有未执行任务的线程退出。对于正在执行任务的线程，在它执行完当前任务获取下一个任务之前会检查线程池状态并开始执行退出逻辑。term_cond的逻辑同shutdown。
3. shutdownAbruptly: 对所有线程均调用cancel（cancelNow()）。如果当前执行的任务中有退出点，则它会退出，如果没有的话，它也会等到当前任务执行完成后再退出。

根据以下逻辑，工作线程的基本执行逻辑如下：

 1     void WorkerRunnable::run(){
 2         pthread_cleanup_push(...); //确保线程退出时调用ThreadPool::onWorkerExit        
 3 
 4         while(true){
 5             Runnable* task = getTask();
 6             if(task == EXIT_TASK){
 7                 //退出
 8             }
 9             pthread_lock(run_lock);
10             pthread_testcancel();
11             task->run();
12             pthread_unlock(run_lock);
13         }
14     }
15 
16     Runnable* getTask(){
17         if(线程池已处理SHUTDOWN_IMMI或ABRUPT状态){
18             return EXIT_TASK;
19         }        
20         Runnable* task = queue->dequeue();
21         return task;
22 
23     }