#ifndef __thread_worker_h__
#define __thread_worker_h__

//std
#include <assert.h>
//boost
#include <boost/thread.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>

class thread_worker : public boost::enable_shared_from_this<thread_worker>
{
public:
    enum eState {
        s_NotReady, s_WaitForTask, s_DataReady, s_DataProcessed, s_Finalize
    };

    struct notification_callbacks
    {
        boost::function<void(boost::shared_ptr<thread_worker>)> fn_notify_ready;
    };


public:
    virtual ~thread_worker() {
        assert(s_NotReady == m_state || s_Finalize == m_state);
    }

//protected:
public:
    thread_worker()
        : m_state(s_NotReady)
    {}

public:
    inline eState state() const { 
        return m_state; 
    }
    
    inline void join() { 
        m_thread.join(); 
    }
    
    inline void processed() { 
        m_state = s_WaitForTask; 
    }

public:
    void run(notification_callbacks &notify_cb)
    {
        m_thread = boost::move(boost::thread([this, notify_cb]
            {
                assert(notify_cb.fn_notify_ready);

                m_state = s_WaitForTask;

                boost::unique_lock<boost::mutex> lock(m_mtx_data_ready);

                do 
                {
                    notify_cb.fn_notify_ready(shared_from_this());
                    m_cv_data_ready.wait(lock, [this]() -> bool { return s_DataReady == m_state || s_Finalize == m_state; });

                    if (s_DataReady == m_state)
                    {
                        do_job();
                        m_state = s_DataProcessed;
                    }

                } while (s_Finalize != m_state);
            }        
        ));
    }

    void finalize()
    {
        m_state = s_Finalize;
        m_cv_data_ready.notify_all();
    }

//protected:
public:
    void push()
    {
        m_state = s_DataReady;
        m_cv_data_ready.notify_all();
    }

public:
        //virtual void do_job() = 0;
    virtual void do_job() {}

private:
    boost::thread m_thread;
    volatile eState m_state;
    boost::mutex m_mtx_data_ready;
    boost::condition_variable m_cv_data_ready;
};

#endif

typedef boost::shared_ptr<thread_worker > ptr_thread_worker;

std::set<ptr_thread_worker> v_threads;
std::queue<ptr_thread_worker>  q_ready_workers;

//data redy sync mutex
boost::mutex mtx_data_ready;
boost::condition_variable cv_data_ready;


//notification from worker
paralel_for_each_worker::notification_callbacks callbacks;
callbacks.fn_notify_ready = [&mtx_data_ready, &q_ready_workers, &cv_data_ready, ](boost::shared_ptr<thread_worker> ptr_worker)
{
    assert(ptr_worker);
    assert(thread_worker::s_WaitForTask == ptr_worker->state() || thread_worker::s_DataProcessed == ptr_worker->state());

    //lock queue
    boost::lock_guard<boost::mutex> lock(mtx_data_ready);
    q_ready_workers.push(ptr_worker);
    
    cv_data_ready.notify_one();
};

unsigned int active_tasks = 0;

//run threads
while (threads--)
{
    ptr_thread_worker worker = boost::make_shared<ptr_thread_worker::element_type>();
    v_threads.insert(worker);
    ++active_tasks;
    worker->run(callbacks);
}

unsigned int y = 0;

boost::unique_lock<boost::mutex> lock(mtx_data_ready);

do
{
    //wait for !empty() queue
    cv_data_ready.wait(lock, [&q_ready_workers]() -> bool { return !q_ready_workers.empty(); });

    while (!q_ready_workers.empty())
    {
        //pop work thread
        ptr_thread_worker ptr_thread_chunk = q_ready_workers.front();
        q_ready_workers.pop();

        ptr_thread_chunk->processed();

        assert(thread_worker::s_WaitForTask == ptr_thread_chunk->state());

        if (y < 10) {
            ptr_thread_chunk->push();
            ++y;
        }
        else {
            --active_tasks;
            ptr_thread_chunk->finalize();
        }
    }

} while (y < 10 || active_tasks);

//wait for end all threads
std::for_each(v_threads.begin(), v_threads.end(), [](ptr_thread_worker ptr_worker) { ptr_worker->finalize(); ptr_worker->join(); });

void finalize()
{
    boost::unique_lock<boost::mutex> lock(m_mtx_data_ready);
    m_state = s_Finalize;
    m_cv_data_ready.notify_all();
}

void push()
{
    boost::unique_lock<boost::mutex> lock(m_mtx_data_ready);
    m_state = s_DataReady;
    m_cv_data_ready.notify_all();
}