[C++] Consumer-Producerパターン

このページにあるConsumer-ProducerパターンのサンプルプログラムをC++で書いてみた。

#include <stdio.h>
#include <iostream>
#include <string>

#include <boost/lexical_cast.hpp>
#include <boost/thread.hpp>

class Table {
    public:
        void put(const std::string& food) {
            boost::mutex::scoped_lock l(m_);
            while (!food_on_table_.empty()) {
                std::cout << boost::this_thread::get_id() << " waits (put)" << std::endl;
                c_.wait(l);
            }
            std::cout << boost::this_thread::get_id() << " puts" << std::endl;
            food_on_table_ = food;
            heavyJob();
            std::cout << food_on_table_ << " is on the table. >>>>>>>>>>" << std::endl;
            c_.notify_all();
        }
        std::string eat() {
            boost::mutex::scoped_lock l(m_);
            while (food_on_table_.empty()) {
                std::cout << boost::this_thread::get_id() << " waits (eat)" << std::endl;
                c_.wait(l);
            }
            std::string food = food_on_table_;
            std::cout << boost::this_thread::get_id() << " eats " << food << " <<<<<<<<<<" << std::endl;
            food_on_table_ = "";
            heavyJob();
            std::cout << "Nothing is on the table." << std::endl;
            c_.notify_all();
            return food;
        }

    private:
        void heavyJob() {
            int msec = 3000*(rand()/(double)RAND_MAX);
            std::cout << msec << "msec sleep" << std::endl;
            boost::this_thread::sleep(boost::posix_time::milliseconds(3000*(rand()/(double)RAND_MAX)));
            std::cout << "wake up" << std::endl;
        }

        boost::mutex m_;
        boost::condition_variable c_;
        std::string food_on_table_;
};

class Producer {
    public:
        Producer(const std::string& food, Table* table):counter_(0) {
            table_ = table;
            food_ = food;
        }
        void run() {
            while (true) {
                table_->put(food_ + " No." + boost::lexical_cast<std::string>(counter_));
                counter_++;
            }
        }

    private :
        int counter_;
        Table* table_;
        std::string food_;
};

class Consumer {
    public:
        Consumer(Table* table) {
            table_ = table;
        }
        void run() {
            while (true) {
                table_->eat();
            }
        }
    private:
        Table* table_;
};

int main(){
    Table table;
    Producer p1("Apple", &table);
    Producer p2("Banana", &table);
    Producer p3("Banana", &table);
    Consumer c1(&table);
    Consumer c2(&table);

    boost::thread_group g;
    g.create_thread(boost::bind(&Producer::run, &p1));
    g.create_thread(boost::bind(&Producer::run, &p2));
    g.create_thread(boost::bind(&Producer::run, &p3));
    g.create_thread(boost::bind(&Consumer::run, &c1));
    g.create_thread(boost::bind(&Consumer::run, &c2));

    g.join_all();
}

注意点

notify_all()によってwaitしているスレッドが動き出すのは、notify_all()を実行した直後ではなく、mutexがリリースされた直後である。参照