1 //
2 //
3 // Copyright 2016 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 // is % allowed in string
17 //
18
19 #include "src/cpp/thread_manager/thread_manager.h"
20
21 #include <atomic>
22 #include <chrono>
23 #include <climits>
24 #include <memory>
25 #include <thread>
26
27 #include <gtest/gtest.h>
28
29 #include <grpc/support/log.h>
30 #include <grpc/support/port_platform.h>
31 #include <grpcpp/grpcpp.h>
32
33 #include "src/core/lib/gprpp/crash.h"
34 #include "test/core/util/test_config.h"
35
36 namespace grpc {
37 namespace {
38
39 struct TestThreadManagerSettings {
40 // The min number of pollers that SHOULD be active in ThreadManager
41 int min_pollers;
42
43 // The max number of pollers that could be active in ThreadManager
44 int max_pollers;
45
46 // The sleep duration in PollForWork() function to simulate "polling"
47 int poll_duration_ms;
48
49 // The sleep duration in DoWork() function to simulate "work"
50 int work_duration_ms;
51
52 // Max number of times PollForWork() is called before shutting down
53 int max_poll_calls;
54
55 // The thread limit (for use in resource quote)
56 int thread_limit;
57
58 // How many should be instantiated
59 int thread_manager_count;
60 };
61
62 class TestThreadManager final : public grpc::ThreadManager {
63 public:
TestThreadManager(const char * name,grpc_resource_quota * rq,const TestThreadManagerSettings & settings)64 TestThreadManager(const char* name, grpc_resource_quota* rq,
65 const TestThreadManagerSettings& settings)
66 : ThreadManager(name, rq, settings.min_pollers, settings.max_pollers),
67 settings_(settings),
68 num_do_work_(0),
69 num_poll_for_work_(0),
70 num_work_found_(0) {}
71
72 grpc::ThreadManager::WorkStatus PollForWork(void** tag, bool* ok) override;
DoWork(void *,bool,bool)73 void DoWork(void* /* tag */, bool /*ok*/, bool /*resources*/) override {
74 num_do_work_.fetch_add(1, std::memory_order_relaxed);
75
76 // Simulate work by sleeping
77 std::this_thread::sleep_for(
78 std::chrono::milliseconds(settings_.work_duration_ms));
79 }
80
81 // Get number of times PollForWork() was called
num_poll_for_work() const82 int num_poll_for_work() const {
83 return num_poll_for_work_.load(std::memory_order_relaxed);
84 }
85 // Get number of times PollForWork() returned WORK_FOUND
num_work_found() const86 int num_work_found() const {
87 return num_work_found_.load(std::memory_order_relaxed);
88 }
89 // Get number of times DoWork() was called
num_do_work() const90 int num_do_work() const {
91 return num_do_work_.load(std::memory_order_relaxed);
92 }
93
94 private:
95 TestThreadManagerSettings settings_;
96
97 // Counters
98 std::atomic_int num_do_work_; // Number of calls to DoWork
99 std::atomic_int num_poll_for_work_; // Number of calls to PollForWork
100 std::atomic_int num_work_found_; // Number of times WORK_FOUND was returned
101 };
102
PollForWork(void ** tag,bool * ok)103 grpc::ThreadManager::WorkStatus TestThreadManager::PollForWork(void** tag,
104 bool* ok) {
105 int call_num = num_poll_for_work_.fetch_add(1, std::memory_order_relaxed);
106 if (call_num >= settings_.max_poll_calls) {
107 Shutdown();
108 return SHUTDOWN;
109 }
110
111 // Simulate "polling" duration
112 std::this_thread::sleep_for(
113 std::chrono::milliseconds(settings_.poll_duration_ms));
114 *tag = nullptr;
115 *ok = true;
116
117 // Return timeout roughly 1 out of every 3 calls just to make the test a bit
118 // more interesting
119 if (call_num % 3 == 0) {
120 return TIMEOUT;
121 }
122
123 num_work_found_.fetch_add(1, std::memory_order_relaxed);
124 return WORK_FOUND;
125 }
126
127 class ThreadManagerTest
128 : public ::testing::TestWithParam<TestThreadManagerSettings> {
129 protected:
SetUp()130 void SetUp() override {
131 grpc_resource_quota* rq = grpc_resource_quota_create("Thread manager test");
132 if (GetParam().thread_limit > 0) {
133 grpc_resource_quota_set_max_threads(rq, GetParam().thread_limit);
134 }
135 for (int i = 0; i < GetParam().thread_manager_count; i++) {
136 thread_manager_.emplace_back(
137 new TestThreadManager("TestThreadManager", rq, GetParam()));
138 }
139 grpc_resource_quota_unref(rq);
140 for (auto& tm : thread_manager_) {
141 tm->Initialize();
142 }
143 for (auto& tm : thread_manager_) {
144 tm->Wait();
145 }
146 }
147
148 std::vector<std::unique_ptr<TestThreadManager>> thread_manager_;
149 };
150
151 TestThreadManagerSettings scenarios[] = {
152 {2 /* min_pollers */, 10 /* max_pollers */, 10 /* poll_duration_ms */,
153 1 /* work_duration_ms */, 50 /* max_poll_calls */,
154 INT_MAX /* thread_limit */, 1 /* thread_manager_count */},
155 {1 /* min_pollers */, 1 /* max_pollers */, 1 /* poll_duration_ms */,
156 10 /* work_duration_ms */, 50 /* max_poll_calls */, 3 /* thread_limit */,
157 2 /* thread_manager_count */}};
158
159 INSTANTIATE_TEST_SUITE_P(ThreadManagerTest, ThreadManagerTest,
160 ::testing::ValuesIn(scenarios));
161
TEST_P(ThreadManagerTest,TestPollAndWork)162 TEST_P(ThreadManagerTest, TestPollAndWork) {
163 for (auto& tm : thread_manager_) {
164 // Verify that The number of times DoWork() was called is equal to the
165 // number of times WORK_FOUND was returned
166 gpr_log(GPR_DEBUG, "DoWork() called %d times", tm->num_do_work());
167 EXPECT_GE(tm->num_poll_for_work(), GetParam().max_poll_calls);
168 EXPECT_EQ(tm->num_do_work(), tm->num_work_found());
169 }
170 }
171
TEST_P(ThreadManagerTest,TestThreadQuota)172 TEST_P(ThreadManagerTest, TestThreadQuota) {
173 if (GetParam().thread_limit > 0) {
174 for (auto& tm : thread_manager_) {
175 EXPECT_GE(tm->num_poll_for_work(), GetParam().max_poll_calls);
176 EXPECT_LE(tm->GetMaxActiveThreadsSoFar(), GetParam().thread_limit);
177 }
178 }
179 }
180
181 } // namespace
182 } // namespace grpc
183
main(int argc,char ** argv)184 int main(int argc, char** argv) {
185 std::srand(std::time(nullptr));
186 grpc::testing::TestEnvironment env(&argc, argv);
187 ::testing::InitGoogleTest(&argc, argv);
188
189 grpc_init();
190 auto ret = RUN_ALL_TESTS();
191 grpc_shutdown();
192
193 return ret;
194 }
195