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