アクティブなスレッドをプールするクラスを作ってみた

スレッドを事前に生成しプールしておくと、実際に使用するときにスレッド生成のオーバーヘッドを避けることができる。このようなパターンをWoker threadパターンと言うらしいので、実際に実装してみた。

仕様はこんな感じ。

• スレッドプールを作成するときにプールサイズを指定できる。
• スレッドプールを作成するときにプールサイズを指定しない場合は、デフォルト値のスレッドが用意される。
• リクエストメソッドでスレッドにタスクを渡す
• クローズメソッドで残っているタスクが完了次第、すべてのスレッドを削除する。
• スレッド追加メソッドで引数個のスレッドをプールに追加する。
• スレッド削除メソッドで引数個のスレッドをプールから削除する。引数が現在のスレッド数を超えている場合はfalseを返す。
• スレッドを削除するときはアイドル状態のスレッドから削除する。

実装はboost C++ Librariesのthreadを使用した。

クラスの定義は次の通り。

1.   class DThreadPool : boost::noncopyable  {
2.   private:
3.     std::queue<boost::function0<void> > queue_; //タスクを入れるキュー
4.     std::map<boost::thread::id, boost::thread *> threadMap_; //スレッドIDとスレッドのマップ
5.     std::map<boost::thread::id, bool> idleMap_; //スレッドIDをスレッド状態のマップ
6.     boost::thread_group thGrp_;
8.     boost::mutex mutex_;
9.     boost::condition_variable_any condition_;
11.     bool isAcceptable_; //タスク受付可能かどうかのフラグ
13.     void run(); //スレッドで実行するタスク待ち関数
14.     void setIdleFlag(boost::thread::id threadID, bool idleFlag);
15.   public:
16.     DThreadPool();
17.     DThreadPool(int size);
18.     ~DThreadPool();
20.     void request(boost::function0<void> func); //タスクをキューに入れる関数
21.     void close(); //クローズ関数
22.     int size();
24.     void addThread(int n); //スレッド追加関数
25.     bool removeThread(int n); //スレッド削除関数
26.   };

スレッドの生成は次のようになっている。

1. DThreadPool::DThreadPool(int size)
2. {
3.   this->isAcceptable_ = true;
4.   for (int i = 0; i < size; i++){
5.     boost::thread* th = thGrp_.create_thread(boost::bind(&DThreadPool::run,   this));
6.     this->threadMap_[th->get_id()] = th;
7.   }
8. }

まず、3行目でタスクを受付可能とする。4〜6行目でsize個のスレッドにrun関数を渡して生成する。それぞれのスレッドをスレッドIDをキー、スレッドへのポインタを値としてマップに入れる。

run関数は次のようになっている。

1. void DThreadPool::run()
2. {
3.   while (isAcceptable_){
4.     boost::mutex::scoped_lock lock(this->mutex_);
6.     setIdleFlag(boost::this_thread::get_id(), true);
7.     while (this->queue_.empty()){
8.       this->condition_.wait(lock);
9.     }
10.     setIdleFlag(boost::this_thread::get_id(), false);
12.     boost::function0<void> func = this->queue_.front();
13.     this->queue_.pop();
14.     func();
15.   }
16.   while (!this->queue_.empty()){
17.     boost::function0<void> func = this->queue_.front();
18.     this->queue_.pop();
19.     func();
20.   }
21. }

3〜15行目は受付可能な状態での処理。まず、ロックをかけキューにタスクが入るの待機する状態になる。このとき自分のスレッドにアイドルフラグをセットする。キューに何かが入ってくると、アイドルフラグを外し、キューから取り出して実行する。16〜20行目は受付不可の状態での処理。キューが空になるまで、キューからタスクを取り出し実行する。キューが空になった時点で、関数が終了する。

タスクをキューに入れる処理は次のようになっている。

1. void DThreadPool::request(boost::function0<void> func)
2. {
3.   boost::mutex::scoped_lock lock(this->mutex_);
4.   this->queue_.push(func);
5.   this->condition_.notify_all();
6. }

ただロックかけて、キューにプッシュして、ロックを外しているだけ。

クローズ処理は次のようになっている。

1. void DThreadPool::close()
2. {
3.   this->isAcceptable_ = false;
4.   for(std::map<boost::thread::id, boost::thread*>::iterator iter =  this->threadMap_.begin();
5.         iter != this->threadMap_.end();
6.         ++iter){
7.     (iter->second)->interrupt();
8.     (iter->second)->join();
9.     this->thGrp_.remove_thread(iter->second);
10.   }
11. }

まず、3行目で受付不可状態にする。4〜10行目ですべてのスレッドに割り込みをかけて、終了を待つ。終了したスレッドはプールから削除する。デストラクタ内でこの関数を呼ぶことで、安全に終了することができる。

スレッドの追加は次の通り。

1. void DThreadPool::addThread(int n)
2. {
3.   if(!this->isAcceptable_){
4.     this->isAcceptable_ = true;
5.   }
6.   for (int i = 0; i < n; i++){
7.     boost::thread* th = thGrp_.create_thread(boost::bind(&DThreadPool::run,   this));
8.     this->threadMap_[th->get_id()] = th;
9.   }
10. }

要はコンストラクタ内でやっていることと同じことをしている。違う点は受付状態を確認してから受付可能にするところ。

スレッドの削除は次の通り。

1. bool DThreadPool::removeThread(int n)
2. {
3.   if(n > this->thGrp_.size()){
4.     return false;
5.   }
6.   int count = 0;
7.   while(count < n){
8.     for(std::map<boost::thread::id, boost::thread*>::iterator iter =  this->threadMap_.begin();
9.           iter != this->threadMap_.end();
10.           ++iter){
11.       if(count < n){
12.         if(this->idleMap_[iter->first]){
13.           boost::thread* th = this->threadMap_[iter->first];
14.           th->interrupt();
15.           th->join();
16.           this->thGrp_.remove_thread(th);
17.           this->idleMap_.erase(iter->first);
18.           count++;
19.         }
20.       }
21.     }
22.   }
23.   if(this->thGrp_.size() == 0){
24.     this->isAcceptable_ = false;
25.   }
26.   return true;
27. }

まず、3〜5行目で削除するスレッド数が現在のスレッド数より多いかをチェックする。多い場合はfalseを返す。次に7〜20行目で削除する個数分だけ、アイドルフラグが立っているスレッドを探して、割り込み処理→終了待ち→削除としていく。このときスレッド状態マップから該当するスレッドIDを削除する。23〜25行目でサイズが0になった場合は受付不可にする。

ソースはこちら→ threadpool.tar.gzをダウンロード

参考サイト

• Let's Boost
  
• metl日記 Boost によるデザインパターン - Worker Thread