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 <stdint.h>
20 #include <stdio.h>
21 #include <string.h>
22
23 #include <algorithm>
24 #include <string>
25 #include <utility>
26 #include <vector>
27
28 #include "absl/flags/flag.h"
29 #include "absl/flags/parse.h"
30 #include "absl/strings/match.h"
31
32 #include <grpc/byte_buffer.h>
33 #include <grpc/byte_buffer_reader.h>
34 #include <grpc/grpc.h>
35 #include <grpc/grpc_security.h>
36 #include <grpc/impl/channel_arg_names.h>
37 #include <grpc/impl/propagation_bits.h>
38 #include <grpc/slice.h>
39 #include <grpc/status.h>
40 #include <grpc/support/log.h>
41 #include <grpc/support/time.h>
42
43 #include "src/core/lib/channel/channel_args.h"
44 #include "src/core/lib/gpr/useful.h"
45 #include "src/core/lib/slice/slice_internal.h"
46 #include "test/core/memory_usage/memstats.h"
47 #include "test/core/util/test_config.h"
48
49 static grpc_channel* channel;
50 static grpc_completion_queue* cq;
51 static grpc_op metadata_ops[2];
52 static grpc_op status_ops[2];
53 static grpc_op snapshot_ops[6];
54 static grpc_op* op;
55
56 typedef struct {
57 grpc_call* call;
58 grpc_metadata_array initial_metadata_recv;
59 grpc_status_code status;
60 grpc_slice details;
61 grpc_metadata_array trailing_metadata_recv;
62 } fling_call;
63
64 // Statically allocate call data structs. Enough to accommodate 100000 ping-pong
65 // calls and 1 extra for the snapshot calls.
66 static fling_call calls[100001];
67
tag(intptr_t t)68 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
69
70 // A call is intentionally divided into two steps. First step is to initiate a
71 // call (i.e send and recv metadata). A call is outstanding after we initated,
72 // so we can measure the call memory usage.
init_ping_pong_request(int call_idx)73 static void init_ping_pong_request(int call_idx) {
74 grpc_metadata_array_init(&calls[call_idx].initial_metadata_recv);
75
76 memset(metadata_ops, 0, sizeof(metadata_ops));
77 op = metadata_ops;
78 op->op = GRPC_OP_SEND_INITIAL_METADATA;
79 op->data.send_initial_metadata.count = 0;
80 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
81 op++;
82 op->op = GRPC_OP_RECV_INITIAL_METADATA;
83 op->data.recv_initial_metadata.recv_initial_metadata =
84 &calls[call_idx].initial_metadata_recv;
85 op++;
86
87 grpc_slice hostname = grpc_slice_from_static_string("localhost");
88 calls[call_idx].call = grpc_channel_create_call(
89 channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
90 grpc_slice_from_static_string("/Reflector/reflectUnary"), &hostname,
91 gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
92
93 GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[call_idx].call,
94 metadata_ops,
95 (size_t)(op - metadata_ops),
96 tag(call_idx), nullptr));
97 grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
98 }
99
100 // Second step is to finish the call (i.e recv status) and destroy the call.
finish_ping_pong_request(int call_idx)101 static void finish_ping_pong_request(int call_idx) {
102 grpc_metadata_array_init(&calls[call_idx].trailing_metadata_recv);
103
104 memset(status_ops, 0, sizeof(status_ops));
105 op = status_ops;
106 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
107 op->data.recv_status_on_client.trailing_metadata =
108 &calls[call_idx].trailing_metadata_recv;
109 op->data.recv_status_on_client.status = &calls[call_idx].status;
110 op->data.recv_status_on_client.status_details = &calls[call_idx].details;
111 op++;
112
113 GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[call_idx].call,
114 status_ops,
115 (size_t)(op - status_ops),
116 tag(call_idx), nullptr));
117 grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
118 grpc_metadata_array_destroy(&calls[call_idx].initial_metadata_recv);
119 grpc_metadata_array_destroy(&calls[call_idx].trailing_metadata_recv);
120 grpc_slice_unref(calls[call_idx].details);
121 grpc_call_unref(calls[call_idx].call);
122 calls[call_idx].call = nullptr;
123 }
124
send_snapshot_request(int call_idx,grpc_slice call_type)125 static MemStats send_snapshot_request(int call_idx, grpc_slice call_type) {
126 grpc_metadata_array_init(&calls[call_idx].initial_metadata_recv);
127 grpc_metadata_array_init(&calls[call_idx].trailing_metadata_recv);
128
129 grpc_byte_buffer* response_payload_recv = nullptr;
130 memset(snapshot_ops, 0, sizeof(snapshot_ops));
131 op = snapshot_ops;
132
133 op->op = GRPC_OP_SEND_INITIAL_METADATA;
134 op->data.send_initial_metadata.count = 0;
135 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
136 op++;
137 op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
138 op++;
139 op->op = GRPC_OP_RECV_INITIAL_METADATA;
140 op->data.recv_initial_metadata.recv_initial_metadata =
141 &calls[call_idx].initial_metadata_recv;
142 op++;
143 op->op = GRPC_OP_RECV_MESSAGE;
144 op->data.recv_message.recv_message = &response_payload_recv;
145 op++;
146 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
147 op->data.recv_status_on_client.trailing_metadata =
148 &calls[call_idx].trailing_metadata_recv;
149 op->data.recv_status_on_client.status = &calls[call_idx].status;
150 op->data.recv_status_on_client.status_details = &calls[call_idx].details;
151 op++;
152
153 grpc_slice hostname = grpc_slice_from_static_string("localhost");
154 calls[call_idx].call = grpc_channel_create_call(
155 channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq, call_type, &hostname,
156 gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
157 GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[call_idx].call,
158 snapshot_ops,
159 (size_t)(op - snapshot_ops),
160 (void*)nullptr, nullptr));
161 grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
162
163 gpr_log(GPR_INFO, "Call %d status %d (%s)", call_idx, calls[call_idx].status,
164 std::string(grpc_core::StringViewFromSlice(calls[call_idx].details))
165 .c_str());
166
167 GPR_ASSERT(response_payload_recv != nullptr);
168 grpc_byte_buffer_reader reader;
169 grpc_byte_buffer_reader_init(&reader, response_payload_recv);
170 grpc_slice response = grpc_byte_buffer_reader_readall(&reader);
171 MemStats snapshot =
172 *reinterpret_cast<MemStats*>(GRPC_SLICE_START_PTR(response));
173
174 grpc_metadata_array_destroy(&calls[call_idx].initial_metadata_recv);
175 grpc_metadata_array_destroy(&calls[call_idx].trailing_metadata_recv);
176 grpc_slice_unref(response);
177 grpc_byte_buffer_reader_destroy(&reader);
178 grpc_byte_buffer_destroy(response_payload_recv);
179 grpc_slice_unref(calls[call_idx].details);
180 calls[call_idx].details = grpc_empty_slice();
181 grpc_call_unref(calls[call_idx].call);
182 calls[call_idx].call = nullptr;
183
184 return snapshot;
185 }
186
187 // Create iterations calls, return MemStats when all outstanding
run_test_loop(int iterations,int * call_idx)188 std::pair<MemStats, MemStats> run_test_loop(int iterations, int* call_idx) {
189 grpc_event event;
190
191 // benchmark period
192 for (int i = 0; i < iterations; ++i) {
193 init_ping_pong_request(*call_idx + i + 1);
194 }
195
196 auto peak = std::make_pair(
197 // client
198 MemStats::Snapshot(),
199 // server
200 send_snapshot_request(
201 0, grpc_slice_from_static_string("Reflector/DestroyCalls")));
202
203 do {
204 event = grpc_completion_queue_next(
205 cq,
206 gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
207 gpr_time_from_micros(10000, GPR_TIMESPAN)),
208 nullptr);
209 } while (event.type != GRPC_QUEUE_TIMEOUT);
210
211 // second step - recv status and destroy call
212 for (int i = 0; i < iterations; ++i) {
213 finish_ping_pong_request(*call_idx + i + 1);
214 }
215
216 do {
217 event = grpc_completion_queue_next(
218 cq,
219 gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
220 gpr_time_from_micros(10000, GPR_TIMESPAN)),
221 nullptr);
222 } while (event.type != GRPC_QUEUE_TIMEOUT);
223
224 *call_idx += iterations;
225
226 return peak;
227 }
228
229 ABSL_FLAG(std::string, target, "localhost:443", "Target host:port");
230 ABSL_FLAG(int, warmup, 100, "Warmup iterations");
231 ABSL_FLAG(int, benchmark, 1000, "Benchmark iterations");
232 ABSL_FLAG(bool, minstack, false, "Use minimal stack");
233
main(int argc,char ** argv)234 int main(int argc, char** argv) {
235 absl::ParseCommandLine(argc, argv);
236
237 grpc_slice slice = grpc_slice_from_copied_string("x");
238 char* fake_argv[1];
239
240 GPR_ASSERT(argc >= 1);
241 fake_argv[0] = argv[0];
242 grpc::testing::TestEnvironment env(&argc, argv);
243
244 grpc_init();
245
246 for (size_t k = 0; k < GPR_ARRAY_SIZE(calls); k++) {
247 calls[k].details = grpc_empty_slice();
248 }
249
250 cq = grpc_completion_queue_create_for_next(nullptr);
251
252 std::vector<grpc_arg> args_vec;
253 if (absl::GetFlag(FLAGS_minstack)) {
254 args_vec.push_back(grpc_channel_arg_integer_create(
255 const_cast<char*>(GRPC_ARG_MINIMAL_STACK), 1));
256 }
257 grpc_channel_args args = {args_vec.size(), args_vec.data()};
258
259 channel = grpc_channel_create(absl::GetFlag(FLAGS_target).c_str(),
260 grpc_insecure_credentials_create(), &args);
261
262 int call_idx = 0;
263 const int warmup_iterations = absl::GetFlag(FLAGS_warmup);
264 const int benchmark_iterations = absl::GetFlag(FLAGS_benchmark);
265
266 // warmup period
267 MemStats server_benchmark_calls_start = send_snapshot_request(
268 0, grpc_slice_from_static_string("Reflector/SimpleSnapshot"));
269 MemStats client_benchmark_calls_start = MemStats::Snapshot();
270
271 run_test_loop(warmup_iterations, &call_idx);
272
273 std::pair<MemStats, MemStats> peak =
274 run_test_loop(benchmark_iterations, &call_idx);
275
276 MemStats client_calls_inflight = peak.first;
277 MemStats server_calls_inflight = peak.second;
278
279 grpc_channel_destroy(channel);
280 grpc_completion_queue_shutdown(cq);
281
282 grpc_event event;
283 do {
284 event = grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
285 nullptr);
286 } while (event.type != GRPC_QUEUE_SHUTDOWN);
287 grpc_slice_unref(slice);
288
289 grpc_completion_queue_destroy(cq);
290 grpc_shutdown_blocking();
291
292 const char* prefix = "";
293 if (absl::StartsWith(absl::GetFlag(FLAGS_target), "xds:")) prefix = "xds ";
294 printf("---------client stats--------\n");
295 printf("%sclient call memory usage: %f bytes per call\n", prefix,
296 static_cast<double>(client_calls_inflight.rss -
297 client_benchmark_calls_start.rss) /
298 benchmark_iterations * 1024);
299
300 printf("---------server stats--------\n");
301 printf("%sserver call memory usage: %f bytes per call\n", prefix,
302 static_cast<double>(server_calls_inflight.rss -
303 server_benchmark_calls_start.rss) /
304 benchmark_iterations * 1024);
305
306 return 0;
307 }
308