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 : #include <forward_list>
35 : #include <functional>
36 : #include <memory>
37 : #include <mutex>
38 : #include <thread>
39 :
40 : #include <gflags/gflags.h>
41 : #include <grpc/grpc.h>
42 : #include <grpc/support/alloc.h>
43 : #include <grpc/support/host_port.h>
44 : #include <grpc/support/log.h>
45 : #include <grpc++/support/config.h>
46 : #include <grpc++/server.h>
47 : #include <grpc++/server_builder.h>
48 : #include <grpc++/server_context.h>
49 : #include <grpc++/security/server_credentials.h>
50 : #include <gtest/gtest.h>
51 :
52 : #include "test/cpp/qps/qpstest.grpc.pb.h"
53 : #include "test/cpp/qps/server.h"
54 :
55 : namespace grpc {
56 : namespace testing {
57 :
58 : class AsyncQpsServerTest : public Server {
59 : public:
60 4 : AsyncQpsServerTest(const ServerConfig &config, int port) {
61 4 : char *server_address = NULL;
62 4 : gpr_join_host_port(&server_address, "::", port);
63 :
64 4 : ServerBuilder builder;
65 4 : builder.AddListeningPort(server_address, InsecureServerCredentials());
66 4 : gpr_free(server_address);
67 :
68 4 : builder.RegisterAsyncService(&async_service_);
69 18 : for (int i = 0; i < config.threads(); i++) {
70 14 : srv_cqs_.emplace_back(builder.AddCompletionQueue());
71 : }
72 :
73 4 : server_ = builder.BuildAndStart();
74 :
75 : using namespace std::placeholders;
76 23754 : for (int i = 0; i < 10000 / config.threads(); i++) {
77 63750 : for (int j = 0; j < config.threads(); j++) {
78 : auto request_unary = std::bind(
79 : &TestService::AsyncService::RequestUnaryCall, &async_service_, _1,
80 40000 : _2, _3, srv_cqs_[j].get(), srv_cqs_[j].get(), _4);
81 : auto request_streaming = std::bind(
82 : &TestService::AsyncService::RequestStreamingCall, &async_service_,
83 40000 : _1, _2, srv_cqs_[j].get(), srv_cqs_[j].get(), _3);
84 : contexts_.push_front(
85 : new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
86 40000 : request_unary, ProcessRPC));
87 : contexts_.push_front(
88 : new ServerRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
89 40000 : request_streaming, ProcessRPC));
90 : }
91 : }
92 18 : for (int i = 0; i < config.threads(); i++) {
93 14 : shutdown_state_.emplace_back(new PerThreadShutdownState());
94 : }
95 18 : for (int i = 0; i < config.threads(); i++) {
96 14 : threads_.emplace_back(&AsyncQpsServerTest::ThreadFunc, this, i);
97 4 : }
98 4 : }
99 12 : ~AsyncQpsServerTest() {
100 4 : auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(10);
101 4 : server_->Shutdown(deadline);
102 18 : for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
103 14 : (*ss)->set_shutdown();
104 : }
105 18 : for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
106 14 : thr->join();
107 : }
108 18 : for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
109 14 : (*cq)->Shutdown();
110 : bool ok;
111 : void *got_tag;
112 14 : while ((*cq)->Next(&got_tag, &ok))
113 : ;
114 : }
115 80008 : while (!contexts_.empty()) {
116 80000 : delete contexts_.front();
117 80000 : contexts_.pop_front();
118 : }
119 8 : }
120 :
121 : private:
122 14 : void ThreadFunc(int rank) {
123 : // Wait until work is available or we are shutting down
124 : bool ok;
125 : void *got_tag;
126 2358405 : while (srv_cqs_[rank]->Next(&got_tag, &ok)) {
127 2357152 : ServerRpcContext *ctx = detag(got_tag);
128 : // The tag is a pointer to an RPC context to invoke
129 2357811 : const bool still_going = ctx->RunNextState(ok);
130 2357847 : if (!shutdown_state_[rank]->shutdown()) {
131 : // this RPC context is done, so refresh it
132 2358430 : if (!still_going) {
133 1165897 : ctx->Reset();
134 : }
135 : } else {
136 14 : return;
137 : }
138 : }
139 0 : return;
140 : }
141 :
142 : class ServerRpcContext {
143 : public:
144 80000 : ServerRpcContext() {}
145 80000 : virtual ~ServerRpcContext(){};
146 : virtual bool RunNextState(bool) = 0; // next state, return false if done
147 : virtual void Reset() = 0; // start this back at a clean state
148 : };
149 2437250 : static void *tag(ServerRpcContext *func) {
150 2437250 : return reinterpret_cast<void *>(func);
151 : }
152 2357867 : static ServerRpcContext *detag(void *tag) {
153 2357867 : return reinterpret_cast<ServerRpcContext *>(tag);
154 : }
155 :
156 : template <class RequestType, class ResponseType>
157 : class ServerRpcContextUnaryImpl GRPC_FINAL : public ServerRpcContext {
158 : public:
159 40000 : ServerRpcContextUnaryImpl(
160 : std::function<void(ServerContext *, RequestType *,
161 : grpc::ServerAsyncResponseWriter<ResponseType> *,
162 : void *)> request_method,
163 : std::function<grpc::Status(const RequestType *, ResponseType *)>
164 : invoke_method)
165 : : srv_ctx_(new ServerContext),
166 : next_state_(&ServerRpcContextUnaryImpl::invoker),
167 : request_method_(request_method),
168 : invoke_method_(invoke_method),
169 40000 : response_writer_(srv_ctx_.get()) {
170 40000 : request_method_(srv_ctx_.get(), &req_, &response_writer_,
171 : AsyncQpsServerTest::tag(this));
172 40000 : }
173 80000 : ~ServerRpcContextUnaryImpl() GRPC_OVERRIDE {}
174 2279654 : bool RunNextState(bool ok) GRPC_OVERRIDE {
175 2279654 : return (this->*next_state_)(ok);
176 : }
177 1140662 : void Reset() GRPC_OVERRIDE {
178 1140662 : srv_ctx_.reset(new ServerContext);
179 1140624 : req_ = RequestType();
180 1140588 : response_writer_ =
181 : grpc::ServerAsyncResponseWriter<ResponseType>(srv_ctx_.get());
182 :
183 : // Then request the method
184 1140474 : next_state_ = &ServerRpcContextUnaryImpl::invoker;
185 1140474 : request_method_(srv_ctx_.get(), &req_, &response_writer_,
186 : AsyncQpsServerTest::tag(this));
187 1140520 : }
188 :
189 : private:
190 1139536 : bool finisher(bool) { return false; }
191 1140619 : bool invoker(bool ok) {
192 1140619 : if (!ok) {
193 1075 : return false;
194 : }
195 :
196 1139544 : ResponseType response;
197 :
198 : // Call the RPC processing function
199 2278465 : grpc::Status status = invoke_method_(&req_, &response);
200 :
201 : // Have the response writer work and invoke on_finish when done
202 1139534 : next_state_ = &ServerRpcContextUnaryImpl::finisher;
203 1139534 : response_writer_.Finish(response, status, AsyncQpsServerTest::tag(this));
204 2278494 : return true;
205 : }
206 : std::unique_ptr<ServerContext> srv_ctx_;
207 : RequestType req_;
208 : bool (ServerRpcContextUnaryImpl::*next_state_)(bool);
209 : std::function<void(ServerContext *, RequestType *,
210 : grpc::ServerAsyncResponseWriter<ResponseType> *, void *)>
211 : request_method_;
212 : std::function<grpc::Status(const RequestType *, ResponseType *)>
213 : invoke_method_;
214 : grpc::ServerAsyncResponseWriter<ResponseType> response_writer_;
215 : };
216 :
217 : template <class RequestType, class ResponseType>
218 : class ServerRpcContextStreamingImpl GRPC_FINAL : public ServerRpcContext {
219 : public:
220 40000 : ServerRpcContextStreamingImpl(
221 : std::function<void(ServerContext *, grpc::ServerAsyncReaderWriter<
222 : ResponseType, RequestType> *,
223 : void *)> request_method,
224 : std::function<grpc::Status(const RequestType *, ResponseType *)>
225 : invoke_method)
226 : : srv_ctx_(new ServerContext),
227 : next_state_(&ServerRpcContextStreamingImpl::request_done),
228 : request_method_(request_method),
229 : invoke_method_(invoke_method),
230 40000 : stream_(srv_ctx_.get()) {
231 40000 : request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
232 40000 : }
233 80000 : ~ServerRpcContextStreamingImpl() GRPC_OVERRIDE {}
234 78306 : bool RunNextState(bool ok) GRPC_OVERRIDE {
235 78306 : return (this->*next_state_)(ok);
236 : }
237 25238 : void Reset() GRPC_OVERRIDE {
238 25238 : srv_ctx_.reset(new ServerContext);
239 25255 : req_ = RequestType();
240 25205 : stream_ = grpc::ServerAsyncReaderWriter<ResponseType, RequestType>(
241 : srv_ctx_.get());
242 :
243 : // Then request the method
244 24760 : next_state_ = &ServerRpcContextStreamingImpl::request_done;
245 24760 : request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
246 25252 : }
247 :
248 : private:
249 25238 : bool request_done(bool ok) {
250 25238 : if (!ok) {
251 25237 : return false;
252 : }
253 1 : stream_.Read(&req_, AsyncQpsServerTest::tag(this));
254 1 : next_state_ = &ServerRpcContextStreamingImpl::read_done;
255 1 : return true;
256 : }
257 :
258 26525 : bool read_done(bool ok) {
259 26525 : if (ok) {
260 : // invoke the method
261 26525 : ResponseType response;
262 : // Call the RPC processing function
263 53050 : grpc::Status status = invoke_method_(&req_, &response);
264 : // initiate the write
265 26525 : stream_.Write(response, AsyncQpsServerTest::tag(this));
266 53050 : next_state_ = &ServerRpcContextStreamingImpl::write_done;
267 : } else { // client has sent writes done
268 : // finish the stream
269 0 : stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
270 0 : next_state_ = &ServerRpcContextStreamingImpl::finish_done;
271 : }
272 26525 : return true;
273 : }
274 26525 : bool write_done(bool ok) {
275 : // now go back and get another streaming read!
276 26525 : if (ok) {
277 26525 : stream_.Read(&req_, AsyncQpsServerTest::tag(this));
278 26525 : next_state_ = &ServerRpcContextStreamingImpl::read_done;
279 : } else {
280 0 : stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
281 0 : next_state_ = &ServerRpcContextStreamingImpl::finish_done;
282 : }
283 26525 : return true;
284 : }
285 0 : bool finish_done(bool ok) { return false; /* reset the context */ }
286 :
287 : std::unique_ptr<ServerContext> srv_ctx_;
288 : RequestType req_;
289 : bool (ServerRpcContextStreamingImpl::*next_state_)(bool);
290 : std::function<void(
291 : ServerContext *,
292 : grpc::ServerAsyncReaderWriter<ResponseType, RequestType> *, void *)>
293 : request_method_;
294 : std::function<grpc::Status(const RequestType *, ResponseType *)>
295 : invoke_method_;
296 : grpc::ServerAsyncReaderWriter<ResponseType, RequestType> stream_;
297 : };
298 :
299 1165889 : static Status ProcessRPC(const SimpleRequest *request,
300 : SimpleResponse *response) {
301 1165889 : if (request->response_size() > 0) {
302 1166047 : if (!SetPayload(request->response_type(), request->response_size(),
303 1166020 : response->mutable_payload())) {
304 0 : return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
305 : }
306 : }
307 1166047 : return Status::OK;
308 : }
309 : std::vector<std::thread> threads_;
310 : std::unique_ptr<grpc::Server> server_;
311 : std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> srv_cqs_;
312 : TestService::AsyncService async_service_;
313 : std::forward_list<ServerRpcContext *> contexts_;
314 :
315 : class PerThreadShutdownState {
316 : public:
317 14 : PerThreadShutdownState() : shutdown_(false) {}
318 :
319 2357516 : bool shutdown() const {
320 2357516 : std::lock_guard<std::mutex> lock(mutex_);
321 2358594 : return shutdown_;
322 : }
323 :
324 14 : void set_shutdown() {
325 14 : std::lock_guard<std::mutex> lock(mutex_);
326 14 : shutdown_ = true;
327 14 : }
328 :
329 : private:
330 : mutable std::mutex mutex_;
331 : bool shutdown_;
332 : };
333 : std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
334 : };
335 :
336 4 : std::unique_ptr<Server> CreateAsyncServer(const ServerConfig &config,
337 : int port) {
338 4 : return std::unique_ptr<Server>(new AsyncQpsServerTest(config, port));
339 : }
340 :
341 : } // namespace testing
342 18 : } // namespace grpc
|
