utility/ThreadPool.h at master · arayq2/utility · 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
#pragma once

#include <vector>
#include <algorithm>
#include <thread>
#include <utility>
#include <pthread.h>

namespace Utility
{
    template<typename Functor>
    class ThreadPool
    {
    public:
        using Workers = std::vector<std::thread>;

        explicit
        ThreadPool(Functor fn, size_t count = 0)
        : fn_(fn)
        , workers_()
        , stopped_(true)
        {
            if ( count > 0 ) { start( count ); }
        }

        ~ThreadPool() { stop(); }

        void start( size_t count )
        {
            if ( stopped_ )
            {
                fill_( count );
                stopped_ = false;
            }
        }

        void stop()
        {
            if ( !stopped_ )
            {
                stopped_ = true;
                drain_();
            }
        }

        void cancel( std::thread::id const& id = std::thread::id(), bool detach = false )
        {
            if ( stopped_ ) { return; } // nothing to do
            if ( id == std::thread::id() )
            {
                stopped_ = true;
                cancel_all_( detach );
            }
            else { cancel_( id, detach ); }
        }

    private:
        Functor     fn_;
        Workers     workers_;
        bool        stopped_;

        void fill_( size_t count )
        {
            std::generate_n( std::back_inserter(workers_), count, [this]()
            {
                return std::thread(fn_);
            } );
        }

        void drain_()
        {
            for ( auto& _worker : workers_ ) { _worker.join(); }
            workers_.clear();
        }

        void cancel_thread_( std::thread& worker, bool detach )
        {
            ::pthread_cancel( worker.native_handle() );
            if ( detach ) { worker.detach(); }
            else { worker.join(); }
        }

        void cancel_( std::thread::id id, bool detach )
        {
            auto    _lambda([id, detach]( std::thread& worker ) -> bool
            {
                if ( worker.get_id() != id ) { return false; }
                cancel_thread_( worker, detach );
                return true;
            });
            workers_.erase( std::remove_if( workers_.begin(), workers_.end(), _lambda ), workers_.end() );
        }

        void cancel_all_( bool detach )
        {
            for ( auto& _worker : workers_ ) { cancel_thread_( _worker, detach ); }
            workers_.clear();
        }
    };
} // namespace Utility