example for "libfiber.hpp"

example.cpp

/*
 * example for "libfiber.hpp" - 2生産者－3消費者デモ
 */
#include <iostream>
#include <vector>
#include <boost/circular_buffer.hpp>
#include "libfiber.hpp"


// 上限付きFIFOキュー
template <class T, std::size_t N>
class bound_queue {
public:
    typedef T value_type;
    bound_queue() : q_(N) {}
    void push(value_type data)
    {
        fiber::unique_lock<fiber::mutex> lk(mtx_);
        cv_pop_.wait(lk, [=]{ return !q_.full(); });
        q_.push_back(data);
        cv_push_.notify_all();
    }
    value_type pop()
    {
        fiber::unique_lock<fiber::mutex> lk(mtx_);
        cv_push_.wait(lk, [=]{ return !q_.empty(); });
        value_type result = q_.front();
        q_.pop_front();
        cv_pop_.notify_all();
        return result;
    }
private:
    boost::circular_buffer<value_type> q_; // FIFOキュー
    fiber::mutex mtx_;                  // キュー操作保護
    fiber::condition_variable cv_push_; // 有効データが1個以上あるとき通知
    fiber::condition_variable cv_pop_;  // 空き容量が1個以上あるとき通知
};

typedef bound_queue<int, 5> QueueType;

// 標準出力用保護 一時ロックオブジェクト生成
fiber::unique_lock<fiber::mutex> autolk()
{
    static fiber::mutex cout_mutex;
    return fiber::unique_lock<fiber::mutex>(cout_mutex);
}


// 素数判定関数
bool is_prime(int n)
{
    if (n % 2 == 0) return (n == 2);
    for (int i = 3; i < n; i += 2)
        if (n % i == 0) return false;
    return true;
}

// 消費者スレッド処理: データ値を出力(0=終端通知)
void dump_data(QueueType& queue, int id)
{
    for (;;) {
        int data = queue.pop();
        if (data == 0) break;
        autolk(), std::cout << "pop " << data << "@" << id << std::endl;
    }
    autolk(), std::cout << "exit@" << id << std::endl;
}

int main()
{
    const int NUM = 10; // データ生成個数

    // 消費者別のFIFOキュー(3個)を作成
    typedef std::vector<QueueType> QueueGroup;
    QueueGroup queue(3);

    // データセレクタ: (入力値%3)番目のキューへ振分
    struct select_queue {
        select_queue(QueueGroup& qg) : qg_(qg) {}
        void push(int n) { qg_[n % 3].push(n); }
        void broadcast(int n) { for (auto& q : qg_) q.push(n); }
        QueueGroup& qg_;
    } selector(queue);

    std::vector<fiber::fiber> producers, consumers;
    // 消費者x3
    consumers.emplace_back(&dump_data, std::ref(queue[0]), 0);
    consumers.emplace_back(&dump_data, std::ref(queue[1]), 1);
    consumers.emplace_back(&dump_data, std::ref(queue[2]), 2);
    // 生産者A: Fibonacci数列を生成
    producers.emplace_back([&]{
        int a, b, t, i;
        selector.push(a = 1);
        selector.push(b = 1);
        for (i = 2; i < NUM; ++i) {
            t = b; b +=a; a = t;
            selector.push(b);
        }
    });
    // 生産者B: 素数列を生成
    producers.emplace_back([&]{
        int n, i;
        selector.push(n = 2);
        for (i = 1, n = 3; i < NUM; ++i, n += 2) {
            while (!is_prime(n)) n += 2;
            selector.push(n);
        }
    });

    // 全生産者の処理完了を待機
    for (auto& r : producers) r.join();

    // 全消費者にデータ列終端通知＆完了待機
    selector.broadcast(0);
    for (auto& r : consumers) r.join();
}

see also https://gist.github.com/2318086