intel tbb::concurrent_queue를 multi level queue로 이용해보기

ubuntu에서는 아래와 같이 설치 필요

* 직접 github에서 받아도 되나 편리한걸 쓰자

sudo apt install libtbb-dev

build는 요런식으로..

g++ multi_concurrent_queue.cpp -pthread -ltbb -std=c++14

 

MAX개(1,000,000)의 Entry class를 push하는 thread를 1, 2, 4개씩, pop 하는 1개의 thread를 생성하여 시간 측정

 

#include <iostream>
#include <memory>
#include <thread>
#include <chrono>
#include <mutex>
#include <queue>
#include <map>
#include <functional>
#include <tbb/concurrent_queue.h>

using namespace std;
using std::chrono::system_clock;
using std::chrono::duration;

class Entry
{
public:
    Entry(uint64_t id) : id_(id)
    {
    }

private:
    uint64_t id_;
};

enum class Priority
{
    TopPriority = 0,
    Level_0 = TopPriority,
    Level_1,
    Level_2,
    MAX
};

template<class Item, class PriorityEnum>
class MultiLevelQueue
{
public:
    MultiLevelQueue(void)
    {
    }

    virtual ~MultiLevelQueue(void)
    {
    }

    void Push(PriorityEnum p, shared_ptr<Item> entry)
    {
        queue_[p].push(entry);
    }

    shared_ptr<Item> Pop(void)
    {
        for (auto& q : queue_)
        {
            shared_ptr<Item> t;
            if (q.second.try_pop(t))
                return t;
        }

        return shared_ptr<Item>(nullptr);
    }

    size_t Size(PriorityEnum p)
    {
        return queue_[p].size();
    }

private:
    map<PriorityEnum, tbb::concurrent_queue<shared_ptr<Item>>> queue_;
};

using TbbMultiLevelQueue = MultiLevelQueue<Entry, Priority>;
static const uint64_t MAX = 1000000;

void p_thread(TbbMultiLevelQueue& queue, uint64_t count)
{
    while (count != 0)
    {
        queue.Push(Priority::Level_0, make_shared<Entry>(count));
        count--;
    }
}

void c_thread(TbbMultiLevelQueue& queue, uint64_t count)
{
    while (count != 0)
    {
        auto i = queue.Pop();
        if (i == nullptr)
            continue;
        count--;
    }
}

void Run(TbbMultiLevelQueue& q, int t_count)
{
    cout << typeid(q).name() << " thread count: " << t_count << endl;
    auto start = system_clock::now();

    vector<thread> v;
    thread c1(c_thread, std::ref(q), MAX);

    for (int i = 0; i < t_count; ++i)
        v.emplace_back(thread(p_thread, std::ref(q), MAX / t_count));

    for (auto& t : v)
        t.join();
    c1.join();

    auto end = system_clock::now();
    cout << "elpased " << chrono::duration_cast<chrono::milliseconds>(end - start).count() << " ms" << endl;
}

int main()
{
    TbbMultiLevelQueue q1;
    Run(q1, 1);
    Run(q1, 2);
    Run(q1, 4);

    return 0;
}