Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #ifndef TEST_QPS_CLIENT_H
35 : #define TEST_QPS_CLIENT_H
36 :
37 : #include <condition_variable>
38 : #include <mutex>
39 :
40 : #include "test/cpp/qps/histogram.h"
41 : #include "test/cpp/qps/interarrival.h"
42 : #include "test/cpp/qps/timer.h"
43 : #include "test/cpp/qps/qpstest.grpc.pb.h"
44 : #include "test/cpp/util/create_test_channel.h"
45 :
46 : namespace grpc {
47 :
48 : #if defined(__APPLE__)
49 : // Specialize Timepoint for high res clock as we need that
50 : template <>
51 : class TimePoint<std::chrono::high_resolution_clock::time_point> {
52 : public:
53 : TimePoint(const std::chrono::high_resolution_clock::time_point& time) {
54 : TimepointHR2Timespec(time, &time_);
55 : }
56 : gpr_timespec raw_time() const { return time_; }
57 :
58 : private:
59 : gpr_timespec time_;
60 : };
61 : #endif
62 :
63 : namespace testing {
64 :
65 : typedef std::chrono::high_resolution_clock grpc_time_source;
66 : typedef std::chrono::time_point<grpc_time_source> grpc_time;
67 :
68 : class Client {
69 : public:
70 6 : explicit Client(const ClientConfig& config)
71 6 : : channels_(config.client_channels()),
72 6 : timer_(new Timer),
73 18 : interarrival_timer_() {
74 26 : for (int i = 0; i < config.client_channels(); i++) {
75 20 : channels_[i].init(config.server_targets(i % config.server_targets_size()),
76 20 : config);
77 : }
78 6 : request_.set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
79 6 : request_.set_response_size(config.payload_size());
80 6 : }
81 6 : virtual ~Client() {}
82 :
83 12 : ClientStats Mark() {
84 12 : Histogram latencies;
85 : // avoid std::vector for old compilers that expect a copy constructor
86 12 : Histogram* to_merge = new Histogram[threads_.size()];
87 52 : for (size_t i = 0; i < threads_.size(); i++) {
88 40 : threads_[i]->BeginSwap(&to_merge[i]);
89 : }
90 24 : std::unique_ptr<Timer> timer(new Timer);
91 12 : timer_.swap(timer);
92 52 : for (size_t i = 0; i < threads_.size(); i++) {
93 40 : threads_[i]->EndSwap();
94 40 : latencies.Merge(&to_merge[i]);
95 : }
96 12 : delete[] to_merge;
97 :
98 12 : auto timer_result = timer->Mark();
99 :
100 12 : ClientStats stats;
101 12 : latencies.FillProto(stats.mutable_latencies());
102 12 : stats.set_time_elapsed(timer_result.wall);
103 12 : stats.set_time_system(timer_result.system);
104 12 : stats.set_time_user(timer_result.user);
105 24 : return stats;
106 : }
107 :
108 : protected:
109 : SimpleRequest request_;
110 : bool closed_loop_;
111 :
112 20 : class ClientChannelInfo {
113 : public:
114 20 : ClientChannelInfo() {}
115 : ClientChannelInfo(const ClientChannelInfo& i) {
116 : // The copy constructor is to satisfy old compilers
117 : // that need it for using std::vector . It is only ever
118 : // used for empty entries
119 : GPR_ASSERT(!i.channel_ && !i.stub_);
120 : }
121 20 : void init(const grpc::string& target, const ClientConfig& config) {
122 : // We have to use a 2-phase init like this with a default
123 : // constructor followed by an initializer function to make
124 : // old compilers happy with using this in std::vector
125 20 : channel_ = CreateTestChannel(target, config.enable_ssl());
126 20 : stub_ = TestService::NewStub(channel_);
127 20 : }
128 : Channel* get_channel() { return channel_.get(); }
129 74806 : TestService::Stub* get_stub() { return stub_.get(); }
130 :
131 : private:
132 : std::shared_ptr<Channel> channel_;
133 : std::unique_ptr<TestService::Stub> stub_;
134 : };
135 : std::vector<ClientChannelInfo> channels_;
136 :
137 6 : void StartThreads(size_t num_threads) {
138 26 : for (size_t i = 0; i < num_threads; i++) {
139 20 : threads_.emplace_back(new Thread(this, i));
140 : }
141 6 : }
142 :
143 6 : void EndThreads() { threads_.clear(); }
144 :
145 : virtual bool ThreadFunc(Histogram* histogram, size_t thread_idx) = 0;
146 :
147 6 : void SetupLoadTest(const ClientConfig& config, size_t num_threads) {
148 : // Set up the load distribution based on the number of threads
149 6 : if (config.load_type() == CLOSED_LOOP) {
150 5 : closed_loop_ = true;
151 : } else {
152 1 : closed_loop_ = false;
153 :
154 1 : std::unique_ptr<RandomDist> random_dist;
155 1 : const auto& load = config.load_params();
156 1 : switch (config.load_type()) {
157 : case POISSON:
158 : random_dist.reset(
159 1 : new ExpDist(load.poisson().offered_load() / num_threads));
160 1 : break;
161 : case UNIFORM:
162 : random_dist.reset(
163 0 : new UniformDist(load.uniform().interarrival_lo() * num_threads,
164 0 : load.uniform().interarrival_hi() * num_threads));
165 0 : break;
166 : case DETERMINISTIC:
167 : random_dist.reset(
168 0 : new DetDist(num_threads / load.determ().offered_load()));
169 0 : break;
170 : case PARETO:
171 : random_dist.reset(
172 0 : new ParetoDist(load.pareto().interarrival_base() * num_threads,
173 0 : load.pareto().alpha()));
174 0 : break;
175 : default:
176 0 : GPR_ASSERT(false);
177 : break;
178 : }
179 :
180 1 : interarrival_timer_.init(*random_dist, num_threads);
181 9 : for (size_t i = 0; i < num_threads; i++) {
182 : next_time_.push_back(
183 16 : grpc_time_source::now() +
184 : std::chrono::duration_cast<grpc_time_source::duration>(
185 24 : interarrival_timer_(i)));
186 1 : }
187 : }
188 6 : }
189 :
190 366615 : bool NextIssueTime(int thread_idx, grpc_time* time_delay) {
191 366615 : if (closed_loop_) {
192 111951 : return false;
193 : } else {
194 254664 : *time_delay = next_time_[thread_idx];
195 509336 : next_time_[thread_idx] +=
196 : std::chrono::duration_cast<grpc_time_source::duration>(
197 764008 : interarrival_timer_(thread_idx));
198 254673 : return true;
199 : }
200 : }
201 :
202 : private:
203 : class Thread {
204 : public:
205 20 : Thread(Client* client, size_t idx)
206 : : done_(false),
207 : new_(nullptr),
208 : client_(client),
209 : idx_(idx),
210 20 : impl_(&Thread::ThreadFunc, this) {}
211 :
212 40 : ~Thread() {
213 : {
214 20 : std::lock_guard<std::mutex> g(mu_);
215 20 : done_ = true;
216 : }
217 20 : impl_.join();
218 20 : }
219 :
220 40 : void BeginSwap(Histogram* n) {
221 40 : std::lock_guard<std::mutex> g(mu_);
222 40 : new_ = n;
223 40 : }
224 :
225 40 : void EndSwap() {
226 40 : std::unique_lock<std::mutex> g(mu_);
227 96 : while (new_ != nullptr) {
228 16 : cv_.wait(g);
229 40 : };
230 40 : }
231 :
232 : private:
233 : Thread(const Thread&);
234 : Thread& operator=(const Thread&);
235 :
236 3548779 : void ThreadFunc() {
237 : for (;;) {
238 : // run the loop body
239 3548779 : const bool thread_still_ok = client_->ThreadFunc(&histogram_, idx_);
240 : // lock, see if we're done
241 3535715 : std::lock_guard<std::mutex> g(mu_);
242 3553324 : if (!thread_still_ok) {
243 0 : gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
244 0 : done_ = true;
245 : }
246 3553324 : if (done_) {
247 40 : return;
248 : }
249 : // check if we're marking, swap out the histogram if so
250 3553304 : if (new_) {
251 40 : new_->Swap(&histogram_);
252 40 : new_ = nullptr;
253 40 : cv_.notify_one();
254 : }
255 3548759 : }
256 : }
257 :
258 : TestService::Stub* stub_;
259 : ClientConfig config_;
260 : std::mutex mu_;
261 : std::condition_variable cv_;
262 : bool done_;
263 : Histogram* new_;
264 : Histogram histogram_;
265 : Client* client_;
266 : size_t idx_;
267 : std::thread impl_;
268 : };
269 :
270 : std::vector<std::unique_ptr<Thread>> threads_;
271 : std::unique_ptr<Timer> timer_;
272 :
273 : InterarrivalTimer interarrival_timer_;
274 : std::vector<grpc_time> next_time_;
275 : };
276 :
277 : std::unique_ptr<Client> CreateSynchronousUnaryClient(const ClientConfig& args);
278 : std::unique_ptr<Client> CreateSynchronousStreamingClient(
279 : const ClientConfig& args);
280 : std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig& args);
281 : std::unique_ptr<Client> CreateAsyncStreamingClient(const ClientConfig& args);
282 :
283 : } // namespace testing
284 : } // namespace grpc
285 :
286 : #endif
|
