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 //
17 //
18
19 #include <signal.h>
20 #include <stdint.h>
21 #include <stdlib.h>
22 #include <time.h>
23
24 #ifndef _WIN32
25 // This is for _exit() below, which is temporary.
26 #include <unistd.h>
27 #endif
28
29 #include <algorithm>
30 #include <string>
31 #include <vector>
32
33 #include "absl/base/attributes.h"
34 #include "absl/flags/flag.h"
35 #include "absl/flags/parse.h"
36 #include "absl/status/status.h"
37
38 #include <grpc/byte_buffer.h>
39 #include <grpc/grpc.h>
40 #include <grpc/grpc_security.h>
41 #include <grpc/impl/channel_arg_names.h>
42 #include <grpc/slice.h>
43 #include <grpc/status.h>
44 #include <grpc/support/alloc.h>
45 #include <grpc/support/log.h>
46 #include <grpc/support/time.h>
47
48 #include "src/core/ext/xds/xds_enabled_server.h"
49 #include "src/core/lib/channel/channel_args.h"
50 #include "src/core/lib/gprpp/host_port.h"
51 #include "test/core/end2end/data/ssl_test_data.h"
52 #include "test/core/memory_usage/memstats.h"
53 #include "test/core/util/port.h"
54 #include "test/core/util/test_config.h"
55
56 ABSL_FLAG(std::string, bind, "", "Bind host:port");
57 ABSL_FLAG(bool, secure, false, "Use security");
58 ABSL_FLAG(bool, minstack, false, "Use minimal stack");
59 ABSL_FLAG(bool, use_xds, false, "Use xDS");
60
61 static grpc_completion_queue* cq;
62 static grpc_server* server;
63 static grpc_op metadata_ops[2];
64 static grpc_op snapshot_ops[5];
65 static grpc_op status_op;
66 static int got_sigint = 0;
67 static grpc_byte_buffer* payload_buffer = nullptr;
68 static int was_cancelled = 2;
69
tag(intptr_t t)70 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
71
72 typedef enum {
73 FLING_SERVER_NEW_REQUEST = 1,
74 FLING_SERVER_SEND_INIT_METADATA,
75 FLING_SERVER_WAIT_FOR_DESTROY,
76 FLING_SERVER_SEND_STATUS_FLING_CALL,
77 FLING_SERVER_SEND_STATUS_SNAPSHOT,
78 FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL
79 } fling_server_tags;
80
81 typedef struct {
82 fling_server_tags state;
83 grpc_call* call;
84 grpc_call_details call_details;
85 grpc_metadata_array request_metadata_recv;
86 grpc_metadata_array initial_metadata_send;
87 } fling_call;
88
89 // hold up to 100000 calls and 6 snaphost calls
90 static fling_call calls[1000006];
91
request_call_unary(int call_idx)92 static void request_call_unary(int call_idx) {
93 if (call_idx == static_cast<int>(sizeof(calls) / sizeof(fling_call))) {
94 gpr_log(GPR_INFO, "Used all call slots (10000) on server. Server exit.");
95 _exit(0);
96 }
97 grpc_metadata_array_init(&calls[call_idx].request_metadata_recv);
98 grpc_server_request_call(
99 server, &calls[call_idx].call, &calls[call_idx].call_details,
100 &calls[call_idx].request_metadata_recv, cq, cq, &calls[call_idx]);
101 }
102
send_initial_metadata_unary(void * tag)103 static void send_initial_metadata_unary(void* tag) {
104 grpc_metadata_array_init(
105 &(*static_cast<fling_call*>(tag)).initial_metadata_send);
106 metadata_ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
107 metadata_ops[0].data.send_initial_metadata.count = 0;
108
109 GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch((*(fling_call*)tag).call,
110 metadata_ops, 1, tag,
111 nullptr));
112 }
113
send_status(void * tag)114 static void send_status(void* tag) {
115 status_op.op = GRPC_OP_SEND_STATUS_FROM_SERVER;
116 status_op.data.send_status_from_server.status = GRPC_STATUS_OK;
117 status_op.data.send_status_from_server.trailing_metadata_count = 0;
118 grpc_slice details = grpc_slice_from_static_string("");
119 status_op.data.send_status_from_server.status_details = &details;
120
121 GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch((*(fling_call*)tag).call,
122 &status_op, 1, tag,
123 nullptr));
124 }
125
send_snapshot(void * tag,MemStats * snapshot)126 static void send_snapshot(void* tag, MemStats* snapshot) {
127 grpc_op* op;
128
129 grpc_slice snapshot_slice =
130 grpc_slice_new(snapshot, sizeof(*snapshot), gpr_free);
131 payload_buffer = grpc_raw_byte_buffer_create(&snapshot_slice, 1);
132 grpc_metadata_array_init(
133 &(*static_cast<fling_call*>(tag)).initial_metadata_send);
134
135 op = snapshot_ops;
136 op->op = GRPC_OP_SEND_INITIAL_METADATA;
137 op->data.send_initial_metadata.count = 0;
138 op++;
139 op->op = GRPC_OP_SEND_MESSAGE;
140 if (payload_buffer == nullptr) {
141 gpr_log(GPR_INFO, "NULL payload buffer !!!");
142 }
143 op->data.send_message.send_message = payload_buffer;
144 op++;
145 op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
146 op->data.send_status_from_server.status = GRPC_STATUS_OK;
147 op->data.send_status_from_server.trailing_metadata_count = 0;
148 grpc_slice details = grpc_slice_from_static_string("");
149 op->data.send_status_from_server.status_details = &details;
150 op++;
151 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
152 op->data.recv_close_on_server.cancelled = &was_cancelled;
153 op++;
154
155 GPR_ASSERT(GRPC_CALL_OK ==
156 grpc_call_start_batch((*(fling_call*)tag).call, snapshot_ops,
157 (size_t)(op - snapshot_ops), tag, nullptr));
158 }
159 // We have some sort of deadlock, so let's not exit gracefully for now.
160 // When that is resolved, please remove the #include <unistd.h> above.
sigint_handler(int)161 static void sigint_handler(int /*x*/) { _exit(0); }
162
OnServingStatusUpdate(void *,const char * uri,grpc_serving_status_update update)163 static void OnServingStatusUpdate(void* /*user_data*/, const char* uri,
164 grpc_serving_status_update update) {
165 absl::Status status(static_cast<absl::StatusCode>(update.code),
166 update.error_message);
167 gpr_log(GPR_INFO, "xDS serving status notification: uri=\"%s\", status=%s",
168 uri, status.ToString().c_str());
169 }
170
main(int argc,char ** argv)171 int main(int argc, char** argv) {
172 absl::ParseCommandLine(argc, argv);
173
174 grpc_event ev;
175 grpc_completion_queue* shutdown_cq;
176 int shutdown_started = 0;
177 int shutdown_finished = 0;
178
179 char* fake_argv[1];
180
181 GPR_ASSERT(argc >= 1);
182 fake_argv[0] = argv[0];
183 grpc::testing::TestEnvironment env(&argc, argv);
184
185 grpc_init();
186 srand(static_cast<unsigned>(clock()));
187
188 std::string addr = absl::GetFlag(FLAGS_bind);
189 if (addr.empty()) {
190 addr = grpc_core::JoinHostPort("::", grpc_pick_unused_port_or_die());
191 }
192 gpr_log(GPR_INFO, "creating server on: %s", addr.c_str());
193
194 cq = grpc_completion_queue_create_for_next(nullptr);
195
196 std::vector<grpc_arg> args_vec;
197 if (absl::GetFlag(FLAGS_minstack)) {
198 args_vec.push_back(grpc_channel_arg_integer_create(
199 const_cast<char*>(GRPC_ARG_MINIMAL_STACK), 1));
200 }
201 // TODO(roth): The xDS code here duplicates the functionality in
202 // XdsServerBuilder, which is undesirable. We should ideally convert
203 // this to use the C++ API instead of the C-core API, so that we can
204 // avoid this duplication.
205 if (absl::GetFlag(FLAGS_use_xds)) {
206 args_vec.push_back(grpc_channel_arg_integer_create(
207 const_cast<char*>(GRPC_ARG_XDS_ENABLED_SERVER), 1));
208 }
209
210 grpc_channel_args args = {args_vec.size(), args_vec.data()};
211 server = grpc_server_create(&args, nullptr);
212
213 if (absl::GetFlag(FLAGS_use_xds)) {
214 grpc_server_config_fetcher* config_fetcher =
215 grpc_server_config_fetcher_xds_create({OnServingStatusUpdate, nullptr},
216 &args);
217 if (config_fetcher != nullptr) {
218 grpc_server_set_config_fetcher(server, config_fetcher);
219 }
220 }
221
222 MemStats before_server_create = MemStats::Snapshot();
223 if (absl::GetFlag(FLAGS_secure)) {
224 grpc_ssl_pem_key_cert_pair pem_key_cert_pair = {test_server1_key,
225 test_server1_cert};
226 grpc_server_credentials* ssl_creds = grpc_ssl_server_credentials_create(
227 nullptr, &pem_key_cert_pair, 1, 0, nullptr);
228 GPR_ASSERT(grpc_server_add_http2_port(server, addr.c_str(), ssl_creds));
229 grpc_server_credentials_release(ssl_creds);
230 } else {
231 GPR_ASSERT(grpc_server_add_http2_port(
232 server, addr.c_str(), grpc_insecure_server_credentials_create()));
233 }
234
235 grpc_server_register_completion_queue(server, cq, nullptr);
236 grpc_server_start(server);
237
238 MemStats after_server_create = MemStats::Snapshot();
239
240 // initialize call instances
241 for (int i = 0; i < static_cast<int>(sizeof(calls) / sizeof(fling_call));
242 i++) {
243 grpc_call_details_init(&calls[i].call_details);
244 calls[i].state = FLING_SERVER_NEW_REQUEST;
245 }
246
247 int next_call_idx = 0;
248 MemStats current_snapshot;
249
250 request_call_unary(next_call_idx);
251
252 signal(SIGINT, sigint_handler);
253
254 while (!shutdown_finished) {
255 if (got_sigint && !shutdown_started) {
256 gpr_log(GPR_INFO, "Shutting down due to SIGINT");
257
258 shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
259 grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000));
260 GPR_ASSERT(grpc_completion_queue_pluck(
261 shutdown_cq, tag(1000),
262 grpc_timeout_seconds_to_deadline(5), nullptr)
263 .type == GRPC_OP_COMPLETE);
264 grpc_completion_queue_destroy(shutdown_cq);
265 grpc_completion_queue_shutdown(cq);
266 shutdown_started = 1;
267 }
268 ev = grpc_completion_queue_next(
269 cq,
270 gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
271 gpr_time_from_micros(1000000, GPR_TIMESPAN)),
272 nullptr);
273 fling_call* s = static_cast<fling_call*>(ev.tag);
274 switch (ev.type) {
275 case GRPC_OP_COMPLETE:
276 switch (s->state) {
277 case FLING_SERVER_NEW_REQUEST:
278 request_call_unary(++next_call_idx);
279 if (0 == grpc_slice_str_cmp(s->call_details.method,
280 "/Reflector/reflectUnary")) {
281 s->state = FLING_SERVER_SEND_INIT_METADATA;
282 send_initial_metadata_unary(s);
283 } else if (0 ==
284 grpc_slice_str_cmp(s->call_details.method,
285 "Reflector/GetBeforeSvrCreation")) {
286 s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
287 send_snapshot(s, &before_server_create);
288 } else if (0 ==
289 grpc_slice_str_cmp(s->call_details.method,
290 "Reflector/GetAfterSvrCreation")) {
291 s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
292 send_snapshot(s, &after_server_create);
293 } else if (0 == grpc_slice_str_cmp(s->call_details.method,
294 "Reflector/SimpleSnapshot")) {
295 s->state = FLING_SERVER_SEND_STATUS_SNAPSHOT;
296 current_snapshot = MemStats::Snapshot();
297 send_snapshot(s, ¤t_snapshot);
298 } else if (0 == grpc_slice_str_cmp(s->call_details.method,
299 "Reflector/DestroyCalls")) {
300 s->state = FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL;
301 current_snapshot = MemStats::Snapshot();
302 send_snapshot(s, ¤t_snapshot);
303 } else {
304 gpr_log(GPR_ERROR, "Wrong call method");
305 }
306 break;
307 case FLING_SERVER_SEND_INIT_METADATA:
308 s->state = FLING_SERVER_WAIT_FOR_DESTROY;
309 break;
310 case FLING_SERVER_WAIT_FOR_DESTROY:
311 break;
312 case FLING_SERVER_SEND_STATUS_FLING_CALL:
313 grpc_call_unref(s->call);
314 grpc_call_details_destroy(&s->call_details);
315 grpc_metadata_array_destroy(&s->initial_metadata_send);
316 grpc_metadata_array_destroy(&s->request_metadata_recv);
317 break;
318 case FLING_SERVER_BATCH_SEND_STATUS_FLING_CALL:
319 for (int k = 0;
320 k < static_cast<int>(sizeof(calls) / sizeof(fling_call));
321 ++k) {
322 if (calls[k].state == FLING_SERVER_WAIT_FOR_DESTROY) {
323 calls[k].state = FLING_SERVER_SEND_STATUS_FLING_CALL;
324 send_status(&calls[k]);
325 }
326 }
327 ABSL_FALLTHROUGH_INTENDED;
328 // no break here since we want to continue to case
329 // FLING_SERVER_SEND_STATUS_SNAPSHOT to destroy the snapshot call
330 case FLING_SERVER_SEND_STATUS_SNAPSHOT:
331 grpc_byte_buffer_destroy(payload_buffer);
332 grpc_call_unref(s->call);
333 grpc_call_details_destroy(&s->call_details);
334 grpc_metadata_array_destroy(&s->initial_metadata_send);
335 grpc_metadata_array_destroy(&s->request_metadata_recv);
336 payload_buffer = nullptr;
337 break;
338 }
339 break;
340 case GRPC_QUEUE_SHUTDOWN:
341 GPR_ASSERT(shutdown_started);
342 shutdown_finished = 1;
343 break;
344 case GRPC_QUEUE_TIMEOUT:
345 break;
346 }
347 }
348
349 grpc_server_destroy(server);
350 grpc_completion_queue_destroy(cq);
351 grpc_shutdown_blocking();
352 return 0;
353 }
354