forked from taskflow/taskflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy paththreadpool.cpp
More file actions
149 lines (116 loc) · 3.44 KB
/
threadpool.cpp
File metadata and controls
149 lines (116 loc) · 3.44 KB
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
// 2018/09/03 - added by Tsung-Wei Huang
// - refactored ProactiveThreadpool unittest
// - added tests for SimpleThreadpool
//
// 2018/09/02 - created by Guannan
// - test_silent_async
// - test_async
// - test_wait_for_all
#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include <doctest.h>
#include <taskflow/threadpool/threadpool.hpp>
// Procedure: test_silent_async
template <typename ThreadpoolType>
void test_silent_async(ThreadpoolType& tp, const size_t task_num) {
std::atomic<size_t> counter{0};
size_t sum = 0;
for(size_t i=0; i<task_num; i++){
sum += i;
tp.silent_async([i=i, &counter](){ counter += i; });
}
tp.shutdown(); //make sure all silent threads end
REQUIRE(counter == sum);
}
// Procedure: test_async
template <typename ThreadpoolType>
void test_async(ThreadpoolType& tp, const size_t task_num){
std::vector<std::future<int>> int_future;
std::vector<int> int_result;
for(size_t i=0; i<task_num; i++){
int_future.emplace_back(tp.async(
[size = i](){
int sum = 0;
for(int i=0; i<=static_cast<int>(size); i++){
sum += i;
}
return sum;
}
)
);
int sum_result = 0;
for(int j=0; j<=static_cast<int>(i); j++) sum_result += j;
int_result.push_back(sum_result);
}
REQUIRE(int_future.size() == int_result.size());
for(size_t i=0; i<int_future.size(); i++){
REQUIRE(int_future[i].get() == int_result[i]);
}
}
// Procedure: test_wait_for_all
template <typename ThreadpoolType>
void test_wait_for_all(ThreadpoolType& tp){
using namespace std::literals::chrono_literals;
const size_t worker_num = tp.num_workers();
const size_t task_num = 20;
std::atomic<size_t> counter{0};
for(size_t i=0; i<task_num; i++){
tp.silent_async([&counter](){
std::this_thread::sleep_for(200us);
counter++;
});
}
REQUIRE(counter <= task_num); // pay attention to the case of 0 worker
tp.shutdown();
REQUIRE(counter == task_num);
REQUIRE(tp.num_workers() == 0);
tp.spawn(static_cast<unsigned>(worker_num));
REQUIRE(tp.num_workers() == worker_num);
counter = 0;
for(size_t i=0; i<task_num; i++){
tp.silent_async([&counter](){
std::this_thread::sleep_for(200us);
counter++;
});
}
tp.wait_for_all();
REQUIRE(counter == task_num);
REQUIRE(tp.num_workers() == worker_num);
}
// --------------------------------------------------------
// Testcase: SimpleThreadpool
// --------------------------------------------------------
TEST_CASE("SimpleThreadpool" * doctest::timeout(300)) {
const size_t task_num = 100;
SUBCASE("PlaceTask"){
for(unsigned i=0; i<=4; ++i) {
tf::SimpleThreadpool tp(i);
test_async(tp, task_num);
test_silent_async(tp, task_num);
}
}
SUBCASE("WaitForAll"){
for(unsigned i=0; i<=4; ++i) {
tf::SimpleThreadpool tp(i);
test_wait_for_all(tp);
}
}
}
// --------------------------------------------------------
// Testcase: ProactiveThreadpool
// --------------------------------------------------------
TEST_CASE("ProactiveThreadpool" * doctest::timeout(300)) {
const size_t task_num = 100;
SUBCASE("PlaceTask"){
for(unsigned i=0; i<=4; ++i) {
tf::ProactiveThreadpool tp(i);
test_async(tp, task_num);
test_silent_async(tp, task_num);
}
}
SUBCASE("WaitForAll"){
for(unsigned i=0; i<=4; ++i) {
tf::ProactiveThreadpool tp(i);
test_wait_for_all(tp);
}
}
}