mooca/mooca_threadpool.hpp at master · labchan-com/mooca · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
/*
 *
 *  mooca_threadpool.hpp
 *  mooca
 *
 *  Created by jelly on 19/02/2018.
 *  Copyright © 2018 mooca.io. All rights reserved.
 *
 */


#ifndef mooca_threadpool_hpp
#define mooca_threadpool_hpp

#include <iostream>
#include <functional>
#include <condition_variable>
#include <queue>
#include <vector>
#include <thread>
#include <utility>

namespace mooca
{
class ThreadPool
{
private:
    std::queue<std::function<void()> >  tasks;
    std::vector<std::thread>        works;
    std::condition_variable         cv;
    std::mutex              queue_mutex;
    bool                    stop;

public:
    ThreadPool( size_t threads );
    template <typename F, typename... Args> void enqueue( F && f, Args &&... args );
    ~ThreadPool();
};


inline ThreadPool::ThreadPool( size_t threads ) : stop( false )
{
    for ( size_t i = 0; i < threads; i++ )
    {
        works.push_back( std::thread(
                             [this]()
        {
            for (;; )
            {
                std::function<void()> task;
                {
                    std::unique_lock<std::mutex> lock( this->queue_mutex );
                    this->cv.wait(lock, [this] {return this->stop || !this->tasks.empty();});

                    if ( this->stop && this->tasks.empty() )
                    {
                        return;
                    }

                    task = std::move(this->tasks.front());
                    this->tasks.pop();
                }
                task();
            }
        }
                         ) );
    }
}

template <typename F, typename... Args>
void ThreadPool::enqueue( F && f, Args &&... args )
{
    auto task = std::bind( std::move( f ), std::move( args ) ... );
    {
        std::unique_lock<std::mutex> lock( this->queue_mutex );
        if (stop)
        {
            throw std::runtime_error( "enqueue on stopped ThreadPool" );
        }
        tasks.push( std::move(task) );
    }
    cv.notify_one();
}


inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock( this->queue_mutex );
        this->stop = true;
    }

    cv.notify_all();
    for ( auto &thread : this->works )
    {
        if (thread.joinable())
        {
            thread.join();
        }
    }
}
}

#endif /* mooca_threadpool_hpp */