1 /*
2 *
3 * Copyright 2015 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 "test/cpp/qps/qps_worker.h"
20
21 #include <memory>
22 #include <mutex>
23 #include <sstream>
24 #include <string>
25 #include <thread>
26 #include <vector>
27
28 #include <grpc/grpc.h>
29 #include <grpc/support/alloc.h>
30 #include <grpc/support/cpu.h>
31 #include <grpc/support/log.h>
32 #include <grpcpp/client_context.h>
33 #include <grpcpp/security/server_credentials.h>
34 #include <grpcpp/server.h>
35 #include <grpcpp/server_builder.h>
36
37 #include "src/core/lib/gprpp/host_port.h"
38 #include "src/proto/grpc/testing/worker_service.grpc.pb.h"
39 #include "test/core/util/grpc_profiler.h"
40 #include "test/core/util/histogram.h"
41 #include "test/cpp/qps/client.h"
42 #include "test/cpp/qps/qps_server_builder.h"
43 #include "test/cpp/qps/server.h"
44 #include "test/cpp/util/create_test_channel.h"
45 #include "test/cpp/util/test_credentials_provider.h"
46
47 namespace grpc {
48 namespace testing {
49
CreateClient(const ClientConfig & config)50 static std::unique_ptr<Client> CreateClient(const ClientConfig& config) {
51 gpr_log(GPR_INFO, "Starting client of type %s %s %d",
52 ClientType_Name(config.client_type()).c_str(),
53 RpcType_Name(config.rpc_type()).c_str(),
54 config.payload_config().has_bytebuf_params());
55
56 switch (config.client_type()) {
57 case ClientType::SYNC_CLIENT:
58 return CreateSynchronousClient(config);
59 case ClientType::ASYNC_CLIENT:
60 return config.payload_config().has_bytebuf_params()
61 ? CreateGenericAsyncStreamingClient(config)
62 : CreateAsyncClient(config);
63 case ClientType::CALLBACK_CLIENT:
64 return CreateCallbackClient(config);
65 default:
66 abort();
67 }
68 abort();
69 }
70
CreateServer(const ServerConfig & config)71 static std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
72 gpr_log(GPR_INFO, "Starting server of type %s",
73 ServerType_Name(config.server_type()).c_str());
74
75 switch (config.server_type()) {
76 case ServerType::SYNC_SERVER:
77 return CreateSynchronousServer(config);
78 case ServerType::ASYNC_SERVER:
79 return CreateAsyncServer(config);
80 case ServerType::ASYNC_GENERIC_SERVER:
81 return CreateAsyncGenericServer(config);
82 case ServerType::CALLBACK_SERVER:
83 return CreateCallbackServer(config);
84 default:
85 abort();
86 }
87 abort();
88 }
89
90 class ScopedProfile final {
91 public:
ScopedProfile(const char * filename,bool enable)92 ScopedProfile(const char* filename, bool enable) : enable_(enable) {
93 if (enable_) grpc_profiler_start(filename);
94 }
~ScopedProfile()95 ~ScopedProfile() {
96 if (enable_) grpc_profiler_stop();
97 }
98
99 private:
100 const bool enable_;
101 };
102
103 class WorkerServiceImpl final : public WorkerService::Service {
104 public:
WorkerServiceImpl(int server_port,QpsWorker * worker)105 WorkerServiceImpl(int server_port, QpsWorker* worker)
106 : acquired_(false), server_port_(server_port), worker_(worker) {}
107
RunClient(ServerContext * ctx,ServerReaderWriter<ClientStatus,ClientArgs> * stream)108 Status RunClient(
109 ServerContext* ctx,
110 ServerReaderWriter<ClientStatus, ClientArgs>* stream) override {
111 gpr_log(GPR_INFO, "RunClient: Entering");
112 InstanceGuard g(this);
113 if (!g.Acquired()) {
114 return Status(StatusCode::RESOURCE_EXHAUSTED, "Client worker busy");
115 }
116
117 ScopedProfile profile("qps_client.prof", false);
118 Status ret = RunClientBody(ctx, stream);
119 gpr_log(GPR_INFO, "RunClient: Returning");
120 return ret;
121 }
122
RunServer(ServerContext * ctx,ServerReaderWriter<ServerStatus,ServerArgs> * stream)123 Status RunServer(
124 ServerContext* ctx,
125 ServerReaderWriter<ServerStatus, ServerArgs>* stream) override {
126 gpr_log(GPR_INFO, "RunServer: Entering");
127 InstanceGuard g(this);
128 if (!g.Acquired()) {
129 return Status(StatusCode::RESOURCE_EXHAUSTED, "Server worker busy");
130 }
131
132 ScopedProfile profile("qps_server.prof", false);
133 Status ret = RunServerBody(ctx, stream);
134 gpr_log(GPR_INFO, "RunServer: Returning");
135 return ret;
136 }
137
CoreCount(ServerContext *,const CoreRequest *,CoreResponse * resp)138 Status CoreCount(ServerContext* /*ctx*/, const CoreRequest*,
139 CoreResponse* resp) override {
140 resp->set_cores(gpr_cpu_num_cores());
141 return Status::OK;
142 }
143
QuitWorker(ServerContext *,const Void *,Void *)144 Status QuitWorker(ServerContext* /*ctx*/, const Void*, Void*) override {
145 InstanceGuard g(this);
146 if (!g.Acquired()) {
147 return Status(StatusCode::RESOURCE_EXHAUSTED, "Quitting worker busy");
148 }
149
150 worker_->MarkDone();
151 return Status::OK;
152 }
153
154 private:
155 // Protect against multiple clients using this worker at once.
156 class InstanceGuard {
157 public:
InstanceGuard(WorkerServiceImpl * impl)158 InstanceGuard(WorkerServiceImpl* impl)
159 : impl_(impl), acquired_(impl->TryAcquireInstance()) {}
~InstanceGuard()160 ~InstanceGuard() {
161 if (acquired_) {
162 impl_->ReleaseInstance();
163 }
164 }
165
Acquired() const166 bool Acquired() const { return acquired_; }
167
168 private:
169 WorkerServiceImpl* const impl_;
170 const bool acquired_;
171 };
172
TryAcquireInstance()173 bool TryAcquireInstance() {
174 std::lock_guard<std::mutex> g(mu_);
175 if (acquired_) return false;
176 acquired_ = true;
177 return true;
178 }
179
ReleaseInstance()180 void ReleaseInstance() {
181 std::lock_guard<std::mutex> g(mu_);
182 GPR_ASSERT(acquired_);
183 acquired_ = false;
184 }
185
RunClientBody(ServerContext *,ServerReaderWriter<ClientStatus,ClientArgs> * stream)186 Status RunClientBody(ServerContext* /*ctx*/,
187 ServerReaderWriter<ClientStatus, ClientArgs>* stream) {
188 ClientArgs args;
189 if (!stream->Read(&args)) {
190 return Status(StatusCode::INVALID_ARGUMENT, "Couldn't read args");
191 }
192 if (!args.has_setup()) {
193 return Status(StatusCode::INVALID_ARGUMENT, "Invalid setup arg");
194 }
195 gpr_log(GPR_INFO, "RunClientBody: about to create client");
196 std::unique_ptr<Client> client = CreateClient(args.setup());
197 if (!client) {
198 return Status(StatusCode::INVALID_ARGUMENT, "Couldn't create client");
199 }
200 gpr_log(GPR_INFO, "RunClientBody: client created");
201 ClientStatus status;
202 if (!stream->Write(status)) {
203 return Status(StatusCode::UNKNOWN, "Client couldn't report init status");
204 }
205 gpr_log(GPR_INFO, "RunClientBody: creation status reported");
206 while (stream->Read(&args)) {
207 gpr_log(GPR_INFO, "RunClientBody: Message read");
208 if (!args.has_mark()) {
209 gpr_log(GPR_INFO, "RunClientBody: Message is not a mark!");
210 return Status(StatusCode::INVALID_ARGUMENT, "Invalid mark");
211 }
212 *status.mutable_stats() = client->Mark(args.mark().reset());
213 if (!stream->Write(status)) {
214 return Status(StatusCode::UNKNOWN, "Client couldn't respond to mark");
215 }
216 gpr_log(GPR_INFO, "RunClientBody: Mark response given");
217 }
218
219 gpr_log(GPR_INFO, "RunClientBody: Awaiting Threads Completion");
220 client->AwaitThreadsCompletion();
221
222 gpr_log(GPR_INFO, "RunClientBody: Returning");
223 return Status::OK;
224 }
225
RunServerBody(ServerContext *,ServerReaderWriter<ServerStatus,ServerArgs> * stream)226 Status RunServerBody(ServerContext* /*ctx*/,
227 ServerReaderWriter<ServerStatus, ServerArgs>* stream) {
228 ServerArgs args;
229 if (!stream->Read(&args)) {
230 return Status(StatusCode::INVALID_ARGUMENT, "Couldn't read server args");
231 }
232 if (!args.has_setup()) {
233 return Status(StatusCode::INVALID_ARGUMENT, "Bad server creation args");
234 }
235 if (server_port_ > 0) {
236 args.mutable_setup()->set_port(server_port_);
237 }
238 gpr_log(GPR_INFO, "RunServerBody: about to create server");
239 std::unique_ptr<Server> server = CreateServer(args.setup());
240 if (g_inproc_servers != nullptr) {
241 g_inproc_servers->push_back(server.get());
242 }
243 if (!server) {
244 return Status(StatusCode::INVALID_ARGUMENT, "Couldn't create server");
245 }
246 gpr_log(GPR_INFO, "RunServerBody: server created");
247 ServerStatus status;
248 status.set_port(server->port());
249 status.set_cores(server->cores());
250 if (!stream->Write(status)) {
251 return Status(StatusCode::UNKNOWN, "Server couldn't report init status");
252 }
253 gpr_log(GPR_INFO, "RunServerBody: creation status reported");
254 while (stream->Read(&args)) {
255 gpr_log(GPR_INFO, "RunServerBody: Message read");
256 if (!args.has_mark()) {
257 gpr_log(GPR_INFO, "RunServerBody: Message not a mark!");
258 return Status(StatusCode::INVALID_ARGUMENT, "Invalid mark");
259 }
260 *status.mutable_stats() = server->Mark(args.mark().reset());
261 if (!stream->Write(status)) {
262 return Status(StatusCode::UNKNOWN, "Server couldn't respond to mark");
263 }
264 gpr_log(GPR_INFO, "RunServerBody: Mark response given");
265 }
266
267 gpr_log(GPR_INFO, "RunServerBody: Returning");
268 return Status::OK;
269 }
270
271 std::mutex mu_;
272 bool acquired_;
273 int server_port_;
274 QpsWorker* worker_;
275 };
276
QpsWorker(int driver_port,int server_port,const std::string & credential_type)277 QpsWorker::QpsWorker(int driver_port, int server_port,
278 const std::string& credential_type) {
279 impl_.reset(new WorkerServiceImpl(server_port, this));
280 gpr_atm_rel_store(&done_, static_cast<gpr_atm>(0));
281
282 std::unique_ptr<ServerBuilder> builder = CreateQpsServerBuilder();
283 builder->AddChannelArgument(GRPC_ARG_ALLOW_REUSEPORT, 0);
284 if (driver_port >= 0) {
285 std::string server_address = grpc_core::JoinHostPort("::", driver_port);
286 builder->AddListeningPort(
287 server_address.c_str(),
288 GetCredentialsProvider()->GetServerCredentials(credential_type));
289 }
290 builder->RegisterService(impl_.get());
291
292 server_ = builder->BuildAndStart();
293 if (server_ == nullptr) {
294 gpr_log(GPR_ERROR,
295 "QpsWorker: Fail to BuildAndStart(driver_port=%d, server_port=%d)",
296 driver_port, server_port);
297 } else {
298 gpr_log(GPR_INFO,
299 "QpsWorker: BuildAndStart(driver_port=%d, server_port=%d) done",
300 driver_port, server_port);
301 }
302 }
303
~QpsWorker()304 QpsWorker::~QpsWorker() {}
305
Done() const306 bool QpsWorker::Done() const {
307 return (gpr_atm_acq_load(&done_) != static_cast<gpr_atm>(0));
308 }
MarkDone()309 void QpsWorker::MarkDone() {
310 gpr_atm_rel_store(&done_, static_cast<gpr_atm>(1));
311 }
312 } // namespace testing
313 } // namespace grpc
314