ceph代码src/common/WorkQueue.h提供了一个功能强大的线程池。简单分析其代码，介绍其使用方法。

主要成员

/// Pool of threads that share work submitted to multiple work queues.
class ThreadPool : public md_config_obs_t {
  CephContext *cct;
  string name;  //线程池名字
  string thread_name; 
  string lockname; //锁的名字
  Mutex _lock;  //线程和工作队列的互斥锁
  Cond _cond;  //锁对应的条件变量
  bool _stop;  //控制线程停止
  int _pause;  //控制线程暂停
  int _draining;  
  Cond _wait_cond;
  int ioprio_class, ioprio_priority;

  // track thread pool size changes
  unsigned _num_threads;  //线程数
  string _thread_num_option; //读取配置中的线程数的key，用于动态增减线程池中的线程数
  const char **_conf_keys;

  vector<WorkQueue_ *> work_queues;  //工作队列数组
  int next_work_queue = 0;  //用于循环从工作队列中取出任务

  set<WorkThread *> _threads; //工作线程集合
  list<WorkThread *> _old_threads;  ///等待被joined的线程
  int processing;
};

与一般线程池实现不同的是，ceph的线程池实现了多种不同的工作队列。一般情况下，一个线程池对应一个类型的工作队列。在要求不高的情况下，也可以一个线程池对应多种类型的工作队列，让线程池处理不同类型的任务。

对外接口

class ThreadPool : public md_config_obs_t {
  //......
public:
  ThreadPool(CephContext *cct_, string nm, string tn, int n,
             const char *option = NULL);
  ~ThreadPool() override;

  /// return number of threads currently running
  int get_num_threads() {
    Mutex::Locker l(_lock);
    return _num_threads;
  }

  /// assign a work queue to this thread pool
  void add_work_queue(WorkQueue_ *wq) {
    Mutex::Locker l(_lock);
    work_queues.push_back(wq);
  }
 
  /// remove a work queue from this thread pool
  void remove_work_queue(WorkQueue_ *wq) {
    Mutex::Locker l(_lock);
    unsigned i = 0;
    while (work_queues[i] != wq) i++;
    for (i++; i < work_queues.size(); i++) work_queues[i - 1] = work_queues[i];
    assert(i == work_queues.size());
    work_queues.resize(i - 1);
  }
  /// start thread pool thread
  void start();
  /// stop thread pool thread
  void stop(bool clear_after = true);
  /// pause thread pool (if it not already paused)
  void pause();
  /// pause initiation of new work
  void pause_new();
  /// resume work in thread pool.  must match each pause() call 1:1 to resume.
  void unpause();
  /** @brief Wait until work completes.
   * If the parameter is NULL, blocks until all threads are idle.
   * If it is not NULL, blocks until the given work queue does not have
   * any items left to process. */
  void drain(WorkQueue_ *wq = 0);

  /// set io priority
  void set_ioprio(int cls, int priority);
  
  
  // 对线程池内部锁和条件变量的简单封装
  // 调用线程池接口时不需要使用，线程池对外函数内部已经有加锁逻辑
  /// take thread pool lock
  void lock() { _lock.Lock(); }
  /// release thread pool lock
  void unlock() { _lock.Unlock(); }
  /// wait for a kick on this thread pool
  void wait(Cond &c) { c.Wait(_lock); }
  /// wake up a waiter (with lock already held)
  void _wake() { _cond.Signal(); }
  /// wake up a waiter (without lock held)
  void wake() {
    Mutex::Locker l(_lock);
    _cond.Signal();
  }
  void _wait() { _cond.Wait(_lock); }
};

工作队列

在上面的数据成员中，vector<WorkQueue_ *> work_queues;项就是工作队列的数组，其中WorkQueue_类是所有不同类型工作队列的基类。所有工作队列都继承自WorkQueue_，并实现对应的接口。

目前主要有4种队列：
队列还有其他的函数，也可以重写，但不是必须重写。详细可以去看代码。

指针类型队列

/** @brief Template by-pointer work queue.
   * Skeleton implementation of a queue that processes items of a given type
   * submitted as pointers. This is useful when the work item are large or
   * include dynamically allocated memory. The queue will automatically add
   * itself to the thread pool on construction and remove itself on destruction.
   */
template <class T>
class WorkQueue : public WorkQueue_ {
    // 需要自己增加queue成员
    // 必须重写的函数
    /// Remove all work items from the queue.
    virtual void _clear() = 0;
    /// Check whether there is anything to do.
    virtual bool _empty() = 0;
    /// Add a work item to the queue.
    virtual bool _enqueue(T *) = 0;
    /// Dequeue a previously submitted work item.
    virtual void _dequeue(T *) = 0;
    /// Dequeue a work item and return the original submitted pointer.
    virtual T *_dequeue() = 0;

    /// Process a work item. Called from the worker threads.
    virtual void _process(T *t, TPHandle &) = 0;
};

简单的指针队列

//同为指针传递，相比于WorkQueue实现的更为完善
template <typename T>
class PointerWQ : public WorkQueue_ {
    // 需要重写的函数
    // T*为enqueue的类型，可以是函数指针，也可以是仿函数或者普通类对象等等
    // process函数的作用就是让你自己执行T所代表的任务
    virtual void process(T *item) = 0;
    /// Remove all work items from the queue.
    virtual void _clear() = 0;
};

批任务队列

/** @brief Work queue that processes several submitted items at once.
   * The queue will automatically add itself to the thread pool on construction
   * and remove itself on destruction. */
template <class T>
class BatchWorkQueue : public WorkQueue_ {
    //必须重写的函数
    virtual void _process(const list<T *> &items, TPHandle &handle) = 0;
    /// Remove all work items from the queue.
    virtual void _clear() = 0;
    /// Check whether there is anything to do.
    virtual bool _empty() = 0;
    virtual bool _enqueue(T *) = 0;
    virtual void _dequeue(T *) = 0;
    virtual void _dequeue(list<T *> *) = 0;
};

值类型队列

/** @brief Templated by-value work queue.
   * Skeleton implementation of a queue that processes items submitted by value.
   * This is useful if the items are single primitive values or very small
   * objects (a few bytes). The queue will automatically add itself to the
   * thread pool on construction and remove itself on destruction. */
  template <typename T, typename U = T>
class WorkQueueVal : public WorkQueue_ {
    // 需要重写的函数
    /// Remove all work items from the queue.
    virtual void _clear() = 0;
    /// Check whether there is anything to do.
    bool _empty() override = 0;
    virtual void _enqueue(T) = 0;
    virtual void _enqueue_front(T) = 0;
    virtual U _dequeue() = 0;
    virtual void _process(U u, TPHandle &) = 0;
};

使用

要使用ThreadPool。

1. 需要选择一个或多个WorkQueue继承，实现对应的方法，要注意文末提到的超时检查。
2. 创建线程池对象，并调用ThreadPool::start()方法，该方法会在加锁的情况下调用ThreadPool::start_threads()函数启动工作线程。
3. 创建工作队列对象，并调用ThreadPool::add_work_queue(WorkQueue_ *wq)将工作队列加入线程池。
4. 向工作队列中添加任务，任务会自动被线程池调度执行。也可以使用工作队列的其他函数控制状态。

工作线程的执行逻辑

执行逻辑在void ThreadPool::worker(WorkThread *wt)函数，概括如下：

1. 判断_stop是否为true，true则退出循环，做收尾工作
2. 调用join_old_threads函数join掉old队列中的线程，并将其从队列中删除。直到old队列为空
3. 判断当前线程队列中的线程数是否大于_num_threads设定的数目，如果是，则将当前线程从工作线程队列删除，放入old队列。
4. 如果_pause为false并且工作队列不为空，从next_work_queue指向的工作队列中取出一个任务，执行任务。执行过程会接连调用队列的_void_process和_void_process_finish函数。而_void_process则最终会调用我们重写的_process函数。默认的_void_process_finish函数一般不做实质性工作，需要的话，我们可以重写它。

超时检查

我们看到在_void_process(void *item, TPHandle &handle)函数中还有第二个参数，TPHandle。

TPHandle的结构如下：

  class TPHandle {
    friend class ThreadPool;
    CephContext *cct;
    heartbeat_handle_d *hb;
    time_t grace; //超时时间，超时后状态为unhealthy
    time_t suicide_grace;//自杀时间，超时后自杀

   public:
    TPHandle(CephContext *cct, heartbeat_handle_d *hb, time_t grace,
             time_t suicide_grace)
        : cct(cct), hb(hb), grace(grace), suicide_grace(suicide_grace) {}
    void reset_tp_timeout();
    void suspend_tp_timeout();
  };

在每次worker执行任务的时候，都会创建一个tphandle，并根据配置设置其超时时间和自杀时间。若当前任务的的执行时间超过grace，会导致cct->get_heartbeat_map()->is_healthy()返回false，当超过suicide_grace时，会导致线程被kill。

void ThreadPool::worker(WorkThread *wt) 函数片段：

  // 线程启动后，将当前线程加入heartbeatmap
  heartbeat_handle_d *hb =
      cct->get_heartbeat_map()->add_worker(ss.str(), pthread_self());
  while(!_stop){
      ......
      // 创建一个handle对象，设置超时时间
      TPHandle tp_handle(cct, hb, wq->timeout_interval,
                         wq->suicide_interval);
      tp_handle.reset_tp_timeout();
      _lock.Unlock();
      // 执行任务i
      wq->_void_process(item, tp_handle);
      _lock.Lock();
      // 收尾工作，如果有的话
      wq->_void_process_finish(item);
    }
    //线程结束前， 从map中移除
    cct->get_heartbeat_map()->remove_worker(hb);

HeartbeatMap中检测超时的逻辑如下。
可以看到，grace超时仅仅设置healthy = false;。而suicide_grace超时则使用pthread_kill(h->thread_id, SIGABRT);kill掉当前工作线程。

bool HeartbeatMap::_check(const heartbeat_handle_d *h, const char *who, time_t now)
{
  bool healthy = true;
  time_t was;

  was = h->timeout;
  if (was && was < now) {
    ldout(m_cct, 1) << who << " '" << h->name << "'"
            << " had timed out after " << h->grace << dendl;
    healthy = false;
  }
  was = h->suicide_timeout;
  if (was && was < now) {
    ldout(m_cct, 1) << who << " '" << h->name << "'"
            << " had suicide timed out after " << h->suicide_grace << dendl;
    pthread_kill(h->thread_id, SIGABRT);
    sleep(1);
    assert(0 == "hit suicide timeout");
  }
  return healthy;
}

如果你不想使用超时检查特性，可以在重写的_process函数中调用suspend_tp_timeout函数关闭当前handle的定时。