线程池工作程序实现

#include <functional>
#include <future>
#include <deque>
#include <thread>

/// <summary> A typical thread worker queue that can execute arbitrary jobs. 
/// </summary>
///
/// <remarks>
///  * Thread Safety   : Full.
///  * Exception Safety: Strong. </remarks>
class WorkQueue{
public:
    /// <summary> Constructors a new work queue object. </summary>
    /// <param name="numWorkers"> (Optional) number of workers, less than 0 to 
    ///     auto-detect (may fail on esoteric platforms). </param>
    explicit WorkQueue(int numWorkers = -1){
        if (numWorkers < 1){
            numWorkers = std::thread::hardware_concurrency() + 1;
        }
        while (numWorkers--){
            m_workers.emplace_back(std::thread(&WorkQueue::doWork, this));
        }
    }

    /// <summary> Will abort all pending jobs and run any in-progress jobs to 
    ///     completion upon destruction. </summary>
    ~WorkQueue(){
        abort();
    }

    /// <summary> Stops work queue and finishes jobs currently being executed.
    ///     Queued jobs that have not begun execution will have their promises 
    ///     broken. </summary>
    void abort(){
        m_exit = true;
        m_finish_work = false;
        m_signal.notify_all();
        joinAll();

        {
            std::lock_guard<std::mutex> lg(m_mutex);
            m_work.clear();
        }
    }

    /// <summary> Stops new work from being submitted to this work queue.</summary>
    void stop(){
        m_exit = true;
        m_finish_work = true;
        m_signal.notify_all();
    }

    /// <summary> Wait for completion of all submitted work. No more work will 
    ///     be allowed to be submitted. </summary>
    void waitForCompletion(){
        stop();
        joinAll();
        assert(m_work.empty());
    }

    /// <summary> Executes the given function asynchronously. </summary>
    /// <exception cref="std::runtime_error"> Thrown if attempting to submit a job
    ///     to a work queue that is terminating. </exception>
    /// <param name="function"> [in] The function to execute. </param>
    /// <returns> A std::future<RETVAL> for the result that will be generated by 
    ///     the function argument. Exceptions from the function will be 
    ///     thrown by get() on the future.</returns>
    template<typename RETVAL>
    std::future<RETVAL> submit(std::function<RETVAL()>&& function){
        if (m_exit){
            throw std::runtime_error("Caught work submission to work queue that is desisting.");
        }

        // Workaround for lack of lambda move capture
        typedef std::pair<std::promise<RETVAL>, std::function<RETVAL()>> pair_t;
        std::shared_ptr<pair_t> data = std::make_shared<pair_t>(std::promise<RETVAL>(), std::move(function));

        std::future<RETVAL> future = data->first.get_future();

        {
            std::lock_guard<std::mutex> lg(m_mutex);
            m_work.emplace_back([data](){
                try{
                    data->first.set_value(data->second());
                }
                catch (...){
                    data->first.set_exception(std::current_exception());
                }
            });
        }
        m_signal.notify_one();
        return std::move(future);
    }

    template<>
    std::future<void> submit(std::function<void()>&& function){
        if (m_exit){
            throw std::runtime_error("Caught work submission to work queue that is desisting.");
        }
        // Workaround for lack of lambda move capture
        typedef std::pair<std::promise<void>, std::function<void()>> pair_t;
        std::shared_ptr<pair_t> data = std::make_shared<pair_t>(std::promise<void>(), std::move(function));

        std::future<void> future = data->first.get_future();

        {
            std::lock_guard<std::mutex> lg(m_mutex);
            m_work.emplace_back([data](){
                try{
                    data->second();
                    data->first.set_value();
                }
                catch (...){
                    data->first.set_exception(std::current_exception());
                }
            });
        }
        m_signal.notify_one();

        return std::move(future);
    }

private:
    std::deque<std::function<void()>> m_work;
    std::mutex m_mutex;
    std::condition_variable m_signal;
    std::atomic<bool> m_exit{ false };
    std::atomic<bool> m_finish_work{ true };
    std::vector<std::thread> m_workers;

    void doWork(){
        std::unique_lock<std::mutex> ul(m_mutex);
        while (!m_exit || (m_finish_work && !m_work.empty())){
            if (!m_work.empty()){
                std::function<void()> work(std::move(m_work.front()));
                m_work.pop_front();
                ul.unlock();
                work();
                ul.lock();
            }
            else{
                m_signal.wait(ul);
            }
        }
    }

    void joinAll(){
        for (auto& thread : m_workers){
            thread.join();
        }
        m_workers.clear();
    }

    void operator=(const WorkQueue&) = delete;
    WorkQueue(const WorkQueue&) = delete;
};

int main(){
    WorkQueue wq;
    wq.submit<void>([](){std::cout << "foo" << std::endl; });
    wq.submit<void>([](){std::cout << "bar" << std::endl; });
    std::future<int> f0 = wq.submit<int>([](){return 4; });
    std::future<int> f1 = wq.submit<int>([](){return 40; });

    std::cout << f1.get() << std::endl;
    wq.waitForCompletion();
    std::cout << f.get() << std::endl;
    return 0;
}

template <typename FunctionObject>
auto submit(FunctionObject &&function) -> std::future<decltype(function())>;

template <>
inline void WorkQueue::execute_and_set_data<void>(const DataPointer<void> &data) {
    data->second();
    data->first.set_value();
}

template <typename ReturnType>
void WorkQueue::execute_and_set_data(const DataPointer<ReturnType> &data) {
    data->first.set_value(data->second());
}

void joinAll(){
    std::vector<std::thread> workers;
    {
        std::lock_guard<std::mutex> lg(m_mutex);
        workers = std::move(m_workers);
    }
    for (auto& thread : workers) {
        thread.join();
    }
}