1 /*
2 *
3 * Copyright 2019 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include "src/core/lib/iomgr/executor/threadpool.h"
20
21 #include "test/core/util/test_config.h"
22
23 static const int kSmallThreadPoolSize = 20;
24 static const int kLargeThreadPoolSize = 100;
25 static const int kThreadSmallIter = 100;
26 static const int kThreadLargeIter = 10000;
27
test_size_zero(void)28 static void test_size_zero(void) {
29 gpr_log(GPR_INFO, "test_size_zero");
30 grpc_core::ThreadPool* pool_size_zero = new grpc_core::ThreadPool(0);
31 GPR_ASSERT(pool_size_zero->pool_capacity() == 1);
32 delete pool_size_zero;
33 }
34
test_constructor_option(void)35 static void test_constructor_option(void) {
36 gpr_log(GPR_INFO, "test_constructor_option");
37 // Tests options
38 grpc_core::Thread::Options options;
39 options.set_stack_size(192 * 1024); // Random non-default value
40 grpc_core::ThreadPool* pool =
41 new grpc_core::ThreadPool(0, "test_constructor_option", options);
42 GPR_ASSERT(pool->thread_options().stack_size() == options.stack_size());
43 delete pool;
44 }
45
46 // Simple functor for testing. It will count how many times being called.
47 class SimpleFunctorForAdd : public grpc_experimental_completion_queue_functor {
48 public:
49 friend class SimpleFunctorCheckForAdd;
SimpleFunctorForAdd()50 SimpleFunctorForAdd() {
51 functor_run = &SimpleFunctorForAdd::Run;
52 inlineable = true;
53 internal_next = this;
54 internal_success = 0;
55 }
~SimpleFunctorForAdd()56 ~SimpleFunctorForAdd() {}
Run(struct grpc_experimental_completion_queue_functor * cb,int)57 static void Run(struct grpc_experimental_completion_queue_functor* cb,
58 int /*ok*/) {
59 auto* callback = static_cast<SimpleFunctorForAdd*>(cb);
60 callback->count_.FetchAdd(1, grpc_core::MemoryOrder::RELAXED);
61 }
62
count()63 int count() { return count_.Load(grpc_core::MemoryOrder::RELAXED); }
64
65 private:
66 grpc_core::Atomic<int> count_{0};
67 };
68
test_add(void)69 static void test_add(void) {
70 gpr_log(GPR_INFO, "test_add");
71 grpc_core::ThreadPool* pool =
72 new grpc_core::ThreadPool(kSmallThreadPoolSize, "test_add");
73
74 SimpleFunctorForAdd* functor = new SimpleFunctorForAdd();
75 for (int i = 0; i < kThreadSmallIter; ++i) {
76 pool->Add(functor);
77 }
78 delete pool;
79 GPR_ASSERT(functor->count() == kThreadSmallIter);
80 delete functor;
81 gpr_log(GPR_DEBUG, "Done.");
82 }
83
84 // Thread that adds closures to pool
85 class WorkThread {
86 public:
WorkThread(grpc_core::ThreadPool * pool,SimpleFunctorForAdd * cb,int num_add)87 WorkThread(grpc_core::ThreadPool* pool, SimpleFunctorForAdd* cb, int num_add)
88 : num_add_(num_add), cb_(cb), pool_(pool) {
89 thd_ = grpc_core::Thread(
90 "thread_pool_test_add_thd",
91 [](void* th) { static_cast<WorkThread*>(th)->Run(); }, this);
92 }
~WorkThread()93 ~WorkThread() {}
94
Start()95 void Start() { thd_.Start(); }
Join()96 void Join() { thd_.Join(); }
97
98 private:
Run()99 void Run() {
100 for (int i = 0; i < num_add_; ++i) {
101 pool_->Add(cb_);
102 }
103 }
104
105 int num_add_;
106 SimpleFunctorForAdd* cb_;
107 grpc_core::ThreadPool* pool_;
108 grpc_core::Thread thd_;
109 };
110
test_multi_add(void)111 static void test_multi_add(void) {
112 gpr_log(GPR_INFO, "test_multi_add");
113 const int num_work_thds = 10;
114 grpc_core::ThreadPool* pool =
115 new grpc_core::ThreadPool(kLargeThreadPoolSize, "test_multi_add");
116 SimpleFunctorForAdd* functor = new SimpleFunctorForAdd();
117 WorkThread** work_thds = static_cast<WorkThread**>(
118 gpr_zalloc(sizeof(WorkThread*) * num_work_thds));
119 gpr_log(GPR_DEBUG, "Fork threads for adding...");
120 for (int i = 0; i < num_work_thds; ++i) {
121 work_thds[i] = new WorkThread(pool, functor, kThreadLargeIter);
122 work_thds[i]->Start();
123 }
124 // Wait for all threads finish
125 gpr_log(GPR_DEBUG, "Waiting for all work threads finish...");
126 for (int i = 0; i < num_work_thds; ++i) {
127 work_thds[i]->Join();
128 delete work_thds[i];
129 }
130 gpr_free(work_thds);
131 gpr_log(GPR_DEBUG, "Done.");
132 gpr_log(GPR_DEBUG, "Waiting for all closures finish...");
133 // Destructor of thread pool will wait for all closures to finish
134 delete pool;
135 GPR_ASSERT(functor->count() == kThreadLargeIter * num_work_thds);
136 delete functor;
137 gpr_log(GPR_DEBUG, "Done.");
138 }
139
140 // Checks the current count with a given number.
141 class SimpleFunctorCheckForAdd
142 : public grpc_experimental_completion_queue_functor {
143 public:
SimpleFunctorCheckForAdd(int ok,int * count)144 SimpleFunctorCheckForAdd(int ok, int* count) : count_(count) {
145 functor_run = &SimpleFunctorCheckForAdd::Run;
146 inlineable = true;
147 internal_success = ok;
148 }
~SimpleFunctorCheckForAdd()149 ~SimpleFunctorCheckForAdd() {}
Run(struct grpc_experimental_completion_queue_functor * cb,int)150 static void Run(struct grpc_experimental_completion_queue_functor* cb,
151 int /*ok*/) {
152 auto* callback = static_cast<SimpleFunctorCheckForAdd*>(cb);
153 (*callback->count_)++;
154 GPR_ASSERT(*callback->count_ == callback->internal_success);
155 }
156
157 private:
158 int* count_;
159 };
160
test_one_thread_FIFO(void)161 static void test_one_thread_FIFO(void) {
162 gpr_log(GPR_INFO, "test_one_thread_FIFO");
163 int counter = 0;
164 grpc_core::ThreadPool* pool =
165 new grpc_core::ThreadPool(1, "test_one_thread_FIFO");
166 SimpleFunctorCheckForAdd** check_functors =
167 static_cast<SimpleFunctorCheckForAdd**>(
168 gpr_zalloc(sizeof(SimpleFunctorCheckForAdd*) * kThreadSmallIter));
169 for (int i = 0; i < kThreadSmallIter; ++i) {
170 check_functors[i] = new SimpleFunctorCheckForAdd(i + 1, &counter);
171 pool->Add(check_functors[i]);
172 }
173 // Destructor of pool will wait until all closures finished.
174 delete pool;
175 for (int i = 0; i < kThreadSmallIter; ++i) {
176 delete check_functors[i];
177 }
178 gpr_free(check_functors);
179 gpr_log(GPR_DEBUG, "Done.");
180 }
181
main(int argc,char ** argv)182 int main(int argc, char** argv) {
183 grpc::testing::TestEnvironment env(argc, argv);
184 grpc_init();
185 test_size_zero();
186 test_constructor_option();
187 test_add();
188 test_multi_add();
189 test_one_thread_FIFO();
190 grpc_shutdown();
191 return 0;
192 }
193