Потоко-безопасная Очередь

// https://stackoverflow.com/questions/19593427/actor-calculation-model-using-boostthread
/*
I have added event queue code into your example.
You have concurrent queue in both EventQueueImpl and Actor, but for different types. 
It is possible to extract common part into separate entity concurrent_queue<T> which works for any type. 
It would be much easier to debug and test queue in one place 
than catching bugs scattered over different classes
*/
// http://stackoverflow.com/questions/19593427/actor-calculation-model-using-boostthread
// g++ -std=c++11 -O3 -mtune=native -Wall -pedantic -pthread main.cpp -lboost_system -lboost_thread -lboost_chrono && ./a.out
 
#include <thread>
#include <future>
#include <functional>
#include <atomic>
#include <iostream>
#include <cassert>
#include <string>
#include <queue>
 
using namespace std;
 
template<typename T>
class concurrent_queue
{
    std::queue<T> q;
    mutable mutex m;
    mutable condition_variable c;
 
    void dequeue(T &result)
    {
        result = q.front();
        q.pop();
    }
public:
    void push(const T &t)
    {
        lock_guard<mutex> l(m);
        q.push(t);
        c.notify_one();
    }
    void pop(T &result)
    {
        unique_lock<mutex> u(m);
        c.wait(u, [this]{ return !q.empty(); } );
        dequeue(result);
    }
    bool try_pop(T &result)
    {
        scoped_lock u(m);
        if(q.empty()) return false;
 
        dequeue(result);
        return true;
    }
    template<typename Duration>
    bool try_pop(T &result, Duration timeout)
    {
        unique_lock<mutex> u(m);
        c.wait_for(u, timeout, [this]{ return !q.empty(); } );
 
        if(q.empty()) return false;
 
        dequeue(result);
        return true;
    }
};
 
struct Actor
{
    typedef function<int()> Job;
 
    concurrent_queue<Job> d_jobQueue;
 
    bool d_keepWorkerRunning;
    thread d_worker;
 
    Actor()
        : d_keepWorkerRunning(true), d_worker(&Actor::workerThread, this)
    {}
 
    ~Actor()
    {
        execJobSync([this]()->int
        {
            d_keepWorkerRunning = false;
            return 0;
        });
        d_worker.join();
    }
 
    void execJobAsync(const Job& job)
    {
        d_jobQueue.push(job);
    }
 
    int execJobSync(const Job& job)
    {
        std::string error;
        promise<int> promise;
        future<int> fut= promise.get_future();  // unique_
 
        d_jobQueue.push([&]() -> int
        {
                int rc = job();    
                promise.set_value(rc);
                return 0;
        });
 
        int rc = fut.get();
        return rc;
    }
 
    void workerThread()
    {
        Job job;
        while (d_keepWorkerRunning)
        {
            d_jobQueue.pop(job);
            job();
        }
    }
};
 
typedef int EventImpl;
typedef concurrent_queue<EventImpl *> EventQueueImpl;
 
int main()
{
    using namespace std;
    cout << "start" << endl;
    {
        Actor x;
        EventQueueImpl eventQueue;
        for(int i=0; i!=16; ++i)
        {
            cout << i << endl;
            x.execJobSync([=]() -> int { 
                        cout << "Sync " << i << endl; 
                        return 0; });
            x.execJobAsync([=]() -> int { 
                        cout << "Async " << i << endl; 
                        return 0; });
        }
        EventImpl e;
        x.execJobSync([&]() -> int {
            cout << "gen event" << endl; 
            eventQueue.push(&e); 
            return 0;
        });
 
        EventImpl *result = nullptr;
        eventQueue.try_pop(result, chrono::milliseconds(10*1000));
        cout << (&e == result) << endl;
    }
    cout << "end" << endl;
}

// Листинг 4.5. Полное определение класса потокобезопасной очереди на базе условных переменных
// Энтони Уильямс Параллельное программирование на С++ в действии - c.119
 
#include <iostream>
#include <queue>
#include <memory>
#include <mutex>
#include <condition_variable>
 
using namespace std;
 
template<typename T>
class threadsafe_queue
{
private:
    mutable std::mutex mut; // Мьютекс должен быть изменяемым
    std::queue<T> data_queue;
    std::condition_variable data_cond;
    
public:     
    threadsafe_queue(){}
    
    threadsafe_queue(threadsafe_queue const& other) {
        std::lock_guard<std::mutex> lk(other.mut);
        data_queue=other.data_queue;
    }
    void push(T new_value) {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(new_value);
        data_cond.notify_one();
    }
    void wait_and_pop(T& value) {
        std::unique_lock<std::mutex> lk(mut);   // другие потоки могут хранить неконстантные ссылки на объект и вызывать изменяющие функции-члены, которые захватывают мьютекс.Поэтому захват мьютекса –это изменяющая операция, следовательно, член mut необходимо пометить как mutable ������, чтобы его можно было захватить в функции empty() и в копирующем конструкторе.
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        value=data_queue.front();
        data_queue.pop();
    }
    std::shared_ptr<T> wait_and_pop() {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return res;
    }
    bool try_pop(T& value) {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
        return false;
        value=data_queue.front();
        data_queue.pop();       
        return true;
    }
    std::shared_ptr<T> try_pop() {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
        return std::shared_ptr<T>();
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return res;
    }
    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};
 
int main()  
{
    threadsafe_queue<int> tsq;
    for (int i=0; i<5;i++) tsq.push(i);
    for (int i=0; i<5;i++) cout << *tsq.wait_and_pop() << endl;
    
  return 0;
}