[surface] Ensure SEND_STATUS & RECV_MESSAGE do not inhabit the same batch (#31554)
Add a check that SEND_STATUS_FROM_SERVER and RECV_MESSAGE are not in the same batch. This is necessary pre-work for #31204 and implements part of grpc/proposal#336. Also eliminates fling instead of updating it: My expectation is nobody has looked at this corner in many years It's not a benchmark we want: concentrating on a microbenchmark that doesn't include a binding layer caused us to favor designs that emphasized a lightweight core at the expense of a expensive bindings. We should consider the whole. Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/31630/head
Craig Tiller
2 years ago
committed by
GitHub
parent
88a6b304fb
commit
d410f1d0aa
21 changed files with 136 additions and 1102 deletions
@ -1,89 +0,0 @@ |
||||
# Copyright 2017 gRPC authors. |
||||
# |
||||
# Licensed under the Apache License, Version 2.0 (the "License"); |
||||
# you may not use this file except in compliance with the License. |
||||
# You may obtain a copy of the License at |
||||
# |
||||
# http://www.apache.org/licenses/LICENSE-2.0 |
||||
# |
||||
# Unless required by applicable law or agreed to in writing, software |
||||
# distributed under the License is distributed on an "AS IS" BASIS, |
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
# See the License for the specific language governing permissions and |
||||
# limitations under the License. |
||||
|
||||
load("//bazel:grpc_build_system.bzl", "grpc_cc_binary", "grpc_cc_test", "grpc_package") |
||||
|
||||
grpc_package(name = "test/core/fling") |
||||
|
||||
licenses(["notice"]) |
||||
|
||||
grpc_cc_binary( |
||||
name = "fling_client", |
||||
testonly = 1, |
||||
srcs = ["client.cc"], |
||||
language = "C++", |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//test/core/end2end:ssl_test_data", |
||||
"//test/core/util:grpc_test_util", |
||||
"//test/core/util:grpc_test_util_base", |
||||
], |
||||
) |
||||
|
||||
grpc_cc_binary( |
||||
name = "fling_server", |
||||
testonly = 1, |
||||
srcs = ["server.cc"], |
||||
language = "C++", |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//test/core/end2end:ssl_test_data", |
||||
"//test/core/util:grpc_test_util", |
||||
"//test/core/util:grpc_test_util_base", |
||||
], |
||||
) |
||||
|
||||
grpc_cc_test( |
||||
name = "fling_test", |
||||
srcs = ["fling_test.cc"], |
||||
data = [ |
||||
":fling_client", |
||||
":fling_server", |
||||
], |
||||
tags = [ |
||||
"no_windows", |
||||
"requires-net:ipv4", |
||||
"requires-net:loopback", |
||||
], |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//test/core/end2end:ssl_test_data", |
||||
"//test/core/util:grpc_test_util", |
||||
"//test/core/util:grpc_test_util_base", |
||||
], |
||||
) |
||||
|
||||
grpc_cc_test( |
||||
name = "fling_stream_test", |
||||
srcs = ["fling_stream_test.cc"], |
||||
data = [ |
||||
":fling_client", |
||||
":fling_server", |
||||
], |
||||
tags = [ |
||||
"no_windows", |
||||
"requires-net:ipv4", |
||||
"requires-net:loopback", |
||||
], |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//test/core/end2end:ssl_test_data", |
||||
"//test/core/util:grpc_test_util", |
||||
"//test/core/util:grpc_test_util_base", |
||||
], |
||||
) |
||||
@ -1,250 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/byte_buffer.h> |
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/impl/codegen/propagation_bits.h> |
||||
#include <grpc/slice.h> |
||||
#include <grpc/status.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/time.h> |
||||
|
||||
#include "src/core/lib/gpr/useful.h" |
||||
#include "test/core/util/cmdline.h" |
||||
#include "test/core/util/grpc_profiler.h" |
||||
#include "test/core/util/histogram.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
static grpc_histogram* histogram; |
||||
static grpc_byte_buffer* the_buffer; |
||||
static grpc_channel* channel; |
||||
static grpc_completion_queue* cq; |
||||
static grpc_call* call; |
||||
static grpc_op ops[6]; |
||||
static grpc_op stream_init_ops[2]; |
||||
static grpc_op stream_step_ops[2]; |
||||
static grpc_metadata_array initial_metadata_recv; |
||||
static grpc_metadata_array trailing_metadata_recv; |
||||
static grpc_byte_buffer* response_payload_recv = nullptr; |
||||
static grpc_status_code status; |
||||
static grpc_slice details; |
||||
static grpc_op* op; |
||||
|
||||
static void init_ping_pong_request(void) { |
||||
grpc_metadata_array_init(&initial_metadata_recv); |
||||
grpc_metadata_array_init(&trailing_metadata_recv); |
||||
|
||||
memset(ops, 0, sizeof(ops)); |
||||
op = ops; |
||||
|
||||
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
op->data.send_initial_metadata.count = 0; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_MESSAGE; |
||||
op->data.send_message.send_message = the_buffer; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_MESSAGE; |
||||
op->data.recv_message.recv_message = &response_payload_recv; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; |
||||
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; |
||||
op->data.recv_status_on_client.status = &status; |
||||
op->data.recv_status_on_client.status_details = &details; |
||||
op++; |
||||
} |
||||
|
||||
static void step_ping_pong_request(void) { |
||||
grpc_slice host = grpc_slice_from_static_string("localhost"); |
||||
call = grpc_channel_create_call( |
||||
channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq, |
||||
grpc_slice_from_static_string("/Reflector/reflectUnary"), &host, |
||||
gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, ops, |
||||
(size_t)(op - ops), (void*)1, |
||||
nullptr)); |
||||
grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); |
||||
grpc_call_unref(call); |
||||
grpc_byte_buffer_destroy(response_payload_recv); |
||||
call = nullptr; |
||||
} |
||||
|
||||
static void init_ping_pong_stream(void) { |
||||
grpc_metadata_array_init(&initial_metadata_recv); |
||||
|
||||
grpc_call_error error; |
||||
grpc_slice host = grpc_slice_from_static_string("localhost"); |
||||
call = grpc_channel_create_call( |
||||
channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq, |
||||
grpc_slice_from_static_string("/Reflector/reflectStream"), &host, |
||||
gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); |
||||
stream_init_ops[0].op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
stream_init_ops[0].data.send_initial_metadata.count = 0; |
||||
stream_init_ops[1].op = GRPC_OP_RECV_INITIAL_METADATA; |
||||
stream_init_ops[1].data.recv_initial_metadata.recv_initial_metadata = |
||||
&initial_metadata_recv; |
||||
error = grpc_call_start_batch(call, stream_init_ops, 2, |
||||
reinterpret_cast<void*>(1), nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); |
||||
|
||||
grpc_metadata_array_init(&initial_metadata_recv); |
||||
|
||||
stream_step_ops[0].op = GRPC_OP_SEND_MESSAGE; |
||||
stream_step_ops[0].data.send_message.send_message = the_buffer; |
||||
stream_step_ops[1].op = GRPC_OP_RECV_MESSAGE; |
||||
stream_step_ops[1].data.recv_message.recv_message = &response_payload_recv; |
||||
} |
||||
|
||||
static void step_ping_pong_stream(void) { |
||||
grpc_call_error error; |
||||
error = grpc_call_start_batch(call, stream_step_ops, 2, |
||||
reinterpret_cast<void*>(1), nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); |
||||
grpc_byte_buffer_destroy(response_payload_recv); |
||||
} |
||||
|
||||
static double now(void) { |
||||
gpr_timespec tv = gpr_now(GPR_CLOCK_REALTIME); |
||||
return 1e9 * static_cast<double>(tv.tv_sec) + tv.tv_nsec; |
||||
} |
||||
|
||||
typedef struct { |
||||
const char* name; |
||||
void (*init)(); |
||||
void (*do_one_step)(); |
||||
} scenario; |
||||
|
||||
static const scenario scenarios[] = { |
||||
{"ping-pong-request", init_ping_pong_request, step_ping_pong_request}, |
||||
{"ping-pong-stream", init_ping_pong_stream, step_ping_pong_stream}, |
||||
}; |
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc_slice slice = grpc_slice_from_copied_string("x"); |
||||
double start, stop; |
||||
unsigned i; |
||||
|
||||
int fake_argc = 1; |
||||
char* fake_argv[1]; |
||||
|
||||
int payload_size = 1; |
||||
int secure = 0; |
||||
const char* target = "localhost:443"; |
||||
gpr_cmdline* cl; |
||||
grpc_event event; |
||||
const char* scenario_name = "ping-pong-request"; |
||||
scenario sc = {nullptr, nullptr, nullptr}; |
||||
|
||||
GPR_ASSERT(argc >= 1); |
||||
fake_argv[0] = argv[0]; |
||||
grpc::testing::TestEnvironment env(&fake_argc, fake_argv); |
||||
|
||||
grpc_init(); |
||||
|
||||
int warmup_seconds = 1; |
||||
int benchmark_seconds = 5; |
||||
|
||||
cl = gpr_cmdline_create("fling client"); |
||||
gpr_cmdline_add_int(cl, "payload_size", "Size of the payload to send", |
||||
&payload_size); |
||||
gpr_cmdline_add_string(cl, "target", "Target host:port", &target); |
||||
gpr_cmdline_add_flag(cl, "secure", "Run with security?", &secure); |
||||
gpr_cmdline_add_string(cl, "scenario", "Scenario", &scenario_name); |
||||
gpr_cmdline_add_int(cl, "warmup", "Warmup seconds", &warmup_seconds); |
||||
gpr_cmdline_add_int(cl, "benchmark", "Benchmark seconds", &benchmark_seconds); |
||||
gpr_cmdline_parse(cl, argc, argv); |
||||
gpr_cmdline_destroy(cl); |
||||
|
||||
for (i = 0; i < GPR_ARRAY_SIZE(scenarios); i++) { |
||||
if (0 == strcmp(scenarios[i].name, scenario_name)) { |
||||
sc = scenarios[i]; |
||||
} |
||||
} |
||||
if (!sc.name) { |
||||
fprintf(stderr, "unsupported scenario '%s'. Valid are:", scenario_name); |
||||
fflush(stderr); |
||||
for (i = 0; i < GPR_ARRAY_SIZE(scenarios); i++) { |
||||
fprintf(stderr, " %s", scenarios[i].name); |
||||
fflush(stderr); |
||||
} |
||||
return 1; |
||||
} |
||||
|
||||
grpc_channel_credentials* creds = grpc_insecure_credentials_create(); |
||||
channel = grpc_channel_create(target, creds, nullptr); |
||||
grpc_channel_credentials_release(creds); |
||||
cq = grpc_completion_queue_create_for_next(nullptr); |
||||
the_buffer = |
||||
grpc_raw_byte_buffer_create(&slice, static_cast<size_t>(payload_size)); |
||||
histogram = grpc_histogram_create(0.01, 60e9); |
||||
|
||||
sc.init(); |
||||
|
||||
gpr_timespec end_warmup = grpc_timeout_seconds_to_deadline(warmup_seconds); |
||||
gpr_timespec end_profiling = |
||||
grpc_timeout_seconds_to_deadline(warmup_seconds + benchmark_seconds); |
||||
|
||||
while (gpr_time_cmp(gpr_now(end_warmup.clock_type), end_warmup) < 0) { |
||||
sc.do_one_step(); |
||||
} |
||||
|
||||
gpr_log(GPR_INFO, "start profiling"); |
||||
grpc_profiler_start("client.prof"); |
||||
while (gpr_time_cmp(gpr_now(end_profiling.clock_type), end_profiling) < 0) { |
||||
start = now(); |
||||
sc.do_one_step(); |
||||
stop = now(); |
||||
grpc_histogram_add(histogram, stop - start); |
||||
} |
||||
grpc_profiler_stop(); |
||||
|
||||
if (call) { |
||||
grpc_call_unref(call); |
||||
} |
||||
|
||||
grpc_channel_destroy(channel); |
||||
grpc_completion_queue_shutdown(cq); |
||||
do { |
||||
event = grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), |
||||
nullptr); |
||||
} while (event.type != GRPC_QUEUE_SHUTDOWN); |
||||
grpc_completion_queue_destroy(cq); |
||||
grpc_byte_buffer_destroy(the_buffer); |
||||
grpc_slice_unref(slice); |
||||
|
||||
gpr_log(GPR_INFO, "latency (50/95/99/99.9): %f/%f/%f/%f", |
||||
grpc_histogram_percentile(histogram, 50), |
||||
grpc_histogram_percentile(histogram, 95), |
||||
grpc_histogram_percentile(histogram, 99), |
||||
grpc_histogram_percentile(histogram, 99.9)); |
||||
grpc_histogram_destroy(histogram); |
||||
|
||||
grpc_shutdown(); |
||||
|
||||
return 0; |
||||
} |
||||
@ -1,80 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <string> |
||||
|
||||
#include "absl/strings/str_cat.h" |
||||
|
||||
#include "src/core/lib/gprpp/host_port.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/subprocess.h" |
||||
|
||||
int main(int /*argc*/, char** argv) { |
||||
char* me = argv[0]; |
||||
char* lslash = strrchr(me, '/'); |
||||
char root[1024]; |
||||
int port = grpc_pick_unused_port_or_die(); |
||||
char* args[10]; |
||||
int status; |
||||
gpr_subprocess *svr, *cli; |
||||
/* figure out where we are */ |
||||
if (lslash) { |
||||
memcpy(root, me, static_cast<size_t>(lslash - me)); |
||||
root[lslash - me] = 0; |
||||
} else { |
||||
strcpy(root, "."); |
||||
} |
||||
/* start the server */ |
||||
std::string command = |
||||
absl::StrCat(root, "/fling_server", gpr_subprocess_binary_extension()); |
||||
args[0] = const_cast<char*>(command.c_str()); |
||||
args[1] = const_cast<char*>("--bind"); |
||||
std::string joined = grpc_core::JoinHostPort("::", port); |
||||
args[2] = const_cast<char*>(joined.c_str()); |
||||
args[3] = const_cast<char*>("--no-secure"); |
||||
svr = gpr_subprocess_create(4, const_cast<const char**>(args)); |
||||
|
||||
/* start the client */ |
||||
command = |
||||
absl::StrCat(root, "/fling_client", gpr_subprocess_binary_extension()); |
||||
args[0] = const_cast<char*>(command.c_str()); |
||||
args[1] = const_cast<char*>("--target"); |
||||
joined = grpc_core::JoinHostPort("127.0.0.1", port); |
||||
args[2] = const_cast<char*>(joined.c_str()); |
||||
args[3] = const_cast<char*>("--scenario=ping-pong-stream"); |
||||
args[4] = const_cast<char*>("--no-secure"); |
||||
args[5] = nullptr; |
||||
cli = gpr_subprocess_create(6, const_cast<const char**>(args)); |
||||
|
||||
/* wait for completion */ |
||||
printf("waiting for client\n"); |
||||
if ((status = gpr_subprocess_join(cli))) { |
||||
gpr_subprocess_destroy(cli); |
||||
gpr_subprocess_destroy(svr); |
||||
return status; |
||||
} |
||||
gpr_subprocess_destroy(cli); |
||||
|
||||
gpr_subprocess_interrupt(svr); |
||||
status = gpr_subprocess_join(svr); |
||||
gpr_subprocess_destroy(svr); |
||||
return status; |
||||
} |
||||
@ -1,83 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <string> |
||||
|
||||
#include "absl/strings/str_cat.h" |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/gprpp/host_port.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/subprocess.h" |
||||
|
||||
int main(int /*argc*/, const char** argv) { |
||||
gpr_log_verbosity_init(); |
||||
const char* me = argv[0]; |
||||
const char* lslash = strrchr(me, '/'); |
||||
char root[1024]; |
||||
int port = grpc_pick_unused_port_or_die(); |
||||
char* args[10]; |
||||
int status; |
||||
gpr_subprocess *svr, *cli; |
||||
/* figure out where we are */ |
||||
if (lslash) { |
||||
memcpy(root, me, static_cast<size_t>(lslash - me)); |
||||
root[lslash - me] = 0; |
||||
} else { |
||||
strcpy(root, "."); |
||||
} |
||||
/* start the server */ |
||||
std::string command = |
||||
absl::StrCat(root, "/fling_server", gpr_subprocess_binary_extension()); |
||||
args[0] = const_cast<char*>(command.c_str()); |
||||
args[1] = const_cast<char*>("--bind"); |
||||
std::string joined = grpc_core::JoinHostPort("::", port); |
||||
args[2] = const_cast<char*>(joined.c_str()); |
||||
args[3] = const_cast<char*>("--no-secure"); |
||||
svr = gpr_subprocess_create(4, const_cast<const char**>(args)); |
||||
|
||||
/* start the client */ |
||||
command = |
||||
absl::StrCat(root, "/fling_client", gpr_subprocess_binary_extension()); |
||||
args[0] = const_cast<char*>(command.c_str()); |
||||
args[1] = const_cast<char*>("--target"); |
||||
joined = grpc_core::JoinHostPort("127.0.0.1", port); |
||||
args[2] = const_cast<char*>(joined.c_str()); |
||||
args[3] = const_cast<char*>("--scenario=ping-pong-request"); |
||||
args[4] = const_cast<char*>("--no-secure"); |
||||
args[5] = nullptr; |
||||
cli = gpr_subprocess_create(6, const_cast<const char**>(args)); |
||||
|
||||
/* wait for completion */ |
||||
printf("waiting for client\n"); |
||||
if ((status = gpr_subprocess_join(cli))) { |
||||
gpr_subprocess_destroy(cli); |
||||
gpr_subprocess_destroy(svr); |
||||
return status; |
||||
} |
||||
gpr_subprocess_destroy(cli); |
||||
|
||||
gpr_subprocess_interrupt(svr); |
||||
status = gpr_subprocess_join(svr); |
||||
gpr_subprocess_destroy(svr); |
||||
return status; |
||||
} |
||||
@ -1,329 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <signal.h> |
||||
#include <stdint.h> |
||||
#include <stdlib.h> |
||||
#include <time.h> |
||||
|
||||
#include <grpc/byte_buffer.h> |
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/slice.h> |
||||
#include <grpc/status.h> |
||||
#include <grpc/support/sync.h> |
||||
#ifndef _WIN32 |
||||
/* This is for _exit() below, which is temporary. */ |
||||
#include <unistd.h> |
||||
#endif |
||||
|
||||
#include <string> |
||||
|
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/time.h> |
||||
|
||||
#include "src/core/lib/gprpp/host_port.h" |
||||
#include "test/core/end2end/data/ssl_test_data.h" |
||||
#include "test/core/util/cmdline.h" |
||||
#include "test/core/util/grpc_profiler.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
static grpc_completion_queue* cq; |
||||
static grpc_server* server; |
||||
static grpc_call* call; |
||||
static grpc_call_details call_details; |
||||
static grpc_metadata_array request_metadata_recv; |
||||
static grpc_metadata_array initial_metadata_send; |
||||
static grpc_byte_buffer* payload_buffer = nullptr; |
||||
/* Used to drain the terminal read in unary calls. */ |
||||
static grpc_byte_buffer* terminal_buffer = nullptr; |
||||
|
||||
static grpc_op read_op; |
||||
static grpc_op metadata_send_op; |
||||
static grpc_op write_op; |
||||
static grpc_op status_op[2]; |
||||
static int was_cancelled = 2; |
||||
static grpc_op unary_ops[6]; |
||||
static int got_sigint = 0; |
||||
|
||||
static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); } |
||||
|
||||
typedef enum { |
||||
FLING_SERVER_NEW_REQUEST = 1, |
||||
FLING_SERVER_READ_FOR_UNARY, |
||||
FLING_SERVER_BATCH_OPS_FOR_UNARY, |
||||
FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING, |
||||
FLING_SERVER_READ_FOR_STREAMING, |
||||
FLING_SERVER_WRITE_FOR_STREAMING, |
||||
FLING_SERVER_SEND_STATUS_FOR_STREAMING |
||||
} fling_server_tags; |
||||
|
||||
typedef struct { |
||||
gpr_refcount pending_ops; |
||||
uint32_t flags; |
||||
} call_state; |
||||
|
||||
static void request_call(void) { |
||||
grpc_metadata_array_init(&request_metadata_recv); |
||||
GPR_ASSERT(GRPC_CALL_OK == |
||||
grpc_server_request_call(server, &call, &call_details, |
||||
&request_metadata_recv, cq, cq, |
||||
tag(FLING_SERVER_NEW_REQUEST))); |
||||
} |
||||
|
||||
static void handle_unary_method(void) { |
||||
grpc_op* op; |
||||
grpc_call_error error; |
||||
|
||||
grpc_metadata_array_init(&initial_metadata_send); |
||||
|
||||
op = unary_ops; |
||||
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
op->data.send_initial_metadata.count = 0; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_MESSAGE; |
||||
op->data.recv_message.recv_message = &terminal_buffer; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_MESSAGE; |
||||
if (payload_buffer == nullptr) { |
||||
gpr_log(GPR_INFO, "NULL payload buffer !!!"); |
||||
} |
||||
op->data.send_message.send_message = payload_buffer; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; |
||||
op->data.send_status_from_server.status = GRPC_STATUS_OK; |
||||
op->data.send_status_from_server.trailing_metadata_count = 0; |
||||
op->data.send_status_from_server.status_details = nullptr; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||
op++; |
||||
|
||||
error = grpc_call_start_batch(call, unary_ops, |
||||
static_cast<size_t>(op - unary_ops), |
||||
tag(FLING_SERVER_BATCH_OPS_FOR_UNARY), nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
} |
||||
|
||||
static void send_initial_metadata(void) { |
||||
grpc_call_error error; |
||||
void* tagarg = tag(FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING); |
||||
grpc_metadata_array_init(&initial_metadata_send); |
||||
metadata_send_op.op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
metadata_send_op.data.send_initial_metadata.count = 0; |
||||
error = grpc_call_start_batch(call, &metadata_send_op, 1, tagarg, nullptr); |
||||
|
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
} |
||||
|
||||
static void start_read_op(int t) { |
||||
grpc_call_error error; |
||||
/* Starting read at server */ |
||||
read_op.op = GRPC_OP_RECV_MESSAGE; |
||||
read_op.data.recv_message.recv_message = &payload_buffer; |
||||
error = grpc_call_start_batch(call, &read_op, 1, tag(t), nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
} |
||||
|
||||
static void start_write_op(void) { |
||||
grpc_call_error error; |
||||
void* tagarg = tag(FLING_SERVER_WRITE_FOR_STREAMING); |
||||
/* Starting write at server */ |
||||
write_op.op = GRPC_OP_SEND_MESSAGE; |
||||
if (payload_buffer == nullptr) { |
||||
gpr_log(GPR_INFO, "NULL payload buffer !!!"); |
||||
} |
||||
write_op.data.send_message.send_message = payload_buffer; |
||||
error = grpc_call_start_batch(call, &write_op, 1, tagarg, nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
} |
||||
|
||||
static void start_send_status(void) { |
||||
grpc_call_error error; |
||||
void* tagarg = tag(FLING_SERVER_SEND_STATUS_FOR_STREAMING); |
||||
status_op[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER; |
||||
status_op[0].data.send_status_from_server.status = GRPC_STATUS_OK; |
||||
status_op[0].data.send_status_from_server.trailing_metadata_count = 0; |
||||
status_op[0].data.send_status_from_server.status_details = nullptr; |
||||
status_op[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||
status_op[1].data.recv_close_on_server.cancelled = &was_cancelled; |
||||
|
||||
error = grpc_call_start_batch(call, status_op, 2, tagarg, nullptr); |
||||
GPR_ASSERT(GRPC_CALL_OK == error); |
||||
} |
||||
|
||||
/* We have some sort of deadlock, so let's not exit gracefully for now.
|
||||
When that is resolved, please remove the #include <unistd.h> above. */ |
||||
static void sigint_handler(int /*x*/) { _exit(0); } |
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc_event ev; |
||||
call_state* s; |
||||
std::string addr_buf; |
||||
gpr_cmdline* cl; |
||||
grpc_completion_queue* shutdown_cq; |
||||
int shutdown_started = 0; |
||||
int shutdown_finished = 0; |
||||
|
||||
int secure = 0; |
||||
const char* addr = nullptr; |
||||
|
||||
char* fake_argv[1]; |
||||
|
||||
GPR_ASSERT(argc >= 1); |
||||
argc = 1; |
||||
fake_argv[0] = argv[0]; |
||||
grpc_test_init(&argc, fake_argv); |
||||
|
||||
grpc_init(); |
||||
srand(static_cast<unsigned>(clock())); |
||||
|
||||
cl = gpr_cmdline_create("fling server"); |
||||
gpr_cmdline_add_string(cl, "bind", "Bind host:port", &addr); |
||||
gpr_cmdline_add_flag(cl, "secure", "Run with security?", &secure); |
||||
gpr_cmdline_parse(cl, argc, argv); |
||||
gpr_cmdline_destroy(cl); |
||||
|
||||
if (addr == nullptr) { |
||||
addr_buf = grpc_core::JoinHostPort("::", grpc_pick_unused_port_or_die()); |
||||
addr = addr_buf.c_str(); |
||||
} |
||||
gpr_log(GPR_INFO, "creating server on: %s", addr); |
||||
|
||||
cq = grpc_completion_queue_create_for_next(nullptr); |
||||
grpc_server_credentials* creds; |
||||
if (secure) { |
||||
grpc_ssl_pem_key_cert_pair pem_key_cert_pair = {test_server1_key, |
||||
test_server1_cert}; |
||||
creds = grpc_ssl_server_credentials_create(nullptr, &pem_key_cert_pair, 1, |
||||
0, nullptr); |
||||
} else { |
||||
creds = grpc_insecure_server_credentials_create(); |
||||
} |
||||
server = grpc_server_create(nullptr, nullptr); |
||||
GPR_ASSERT(grpc_server_add_http2_port(server, addr, creds)); |
||||
grpc_server_credentials_release(creds); |
||||
grpc_server_register_completion_queue(server, cq, nullptr); |
||||
grpc_server_start(server); |
||||
|
||||
addr = nullptr; |
||||
addr_buf.clear(); |
||||
|
||||
grpc_call_details_init(&call_details); |
||||
|
||||
request_call(); |
||||
|
||||
grpc_profiler_start("server.prof"); |
||||
signal(SIGINT, sigint_handler); |
||||
while (!shutdown_finished) { |
||||
if (got_sigint && !shutdown_started) { |
||||
gpr_log(GPR_INFO, "Shutting down due to SIGINT"); |
||||
|
||||
shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr); |
||||
grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000)); |
||||
|
||||
GPR_ASSERT(grpc_completion_queue_pluck( |
||||
shutdown_cq, tag(1000), |
||||
grpc_timeout_seconds_to_deadline(5), nullptr) |
||||
.type == GRPC_OP_COMPLETE); |
||||
grpc_completion_queue_destroy(shutdown_cq); |
||||
|
||||
grpc_completion_queue_shutdown(cq); |
||||
shutdown_started = 1; |
||||
} |
||||
ev = grpc_completion_queue_next( |
||||
cq, |
||||
gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), |
||||
gpr_time_from_micros(1000000, GPR_TIMESPAN)), |
||||
nullptr); |
||||
s = static_cast<call_state*>(ev.tag); |
||||
switch (ev.type) { |
||||
case GRPC_OP_COMPLETE: |
||||
switch (reinterpret_cast<intptr_t>(s)) { |
||||
case FLING_SERVER_NEW_REQUEST: |
||||
if (call != nullptr) { |
||||
if (0 == grpc_slice_str_cmp(call_details.method, |
||||
"/Reflector/reflectStream")) { |
||||
/* Received streaming call. Send metadata here. */ |
||||
start_read_op(FLING_SERVER_READ_FOR_STREAMING); |
||||
send_initial_metadata(); |
||||
} else { |
||||
/* Received unary call. Can do all ops in one batch. */ |
||||
start_read_op(FLING_SERVER_READ_FOR_UNARY); |
||||
} |
||||
} else { |
||||
GPR_ASSERT(shutdown_started); |
||||
} |
||||
/* request_call();
|
||||
*/ |
||||
break; |
||||
case FLING_SERVER_READ_FOR_STREAMING: |
||||
if (payload_buffer != nullptr) { |
||||
/* Received payload from client. */ |
||||
start_write_op(); |
||||
} else { |
||||
/* Received end of stream from client. */ |
||||
start_send_status(); |
||||
} |
||||
break; |
||||
case FLING_SERVER_WRITE_FOR_STREAMING: |
||||
/* Write completed at server */ |
||||
grpc_byte_buffer_destroy(payload_buffer); |
||||
payload_buffer = nullptr; |
||||
start_read_op(FLING_SERVER_READ_FOR_STREAMING); |
||||
break; |
||||
case FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING: |
||||
/* Metadata send completed at server */ |
||||
break; |
||||
case FLING_SERVER_SEND_STATUS_FOR_STREAMING: |
||||
/* Send status and close completed at server */ |
||||
grpc_call_unref(call); |
||||
if (!shutdown_started) request_call(); |
||||
break; |
||||
case FLING_SERVER_READ_FOR_UNARY: |
||||
/* Finished payload read for unary. Start all reamaining
|
||||
* unary ops in a batch. |
||||
*/ |
||||
handle_unary_method(); |
||||
break; |
||||
case FLING_SERVER_BATCH_OPS_FOR_UNARY: |
||||
/* Finished unary call. */ |
||||
grpc_byte_buffer_destroy(payload_buffer); |
||||
payload_buffer = nullptr; |
||||
grpc_call_unref(call); |
||||
if (!shutdown_started) request_call(); |
||||
break; |
||||
} |
||||
break; |
||||
case GRPC_QUEUE_SHUTDOWN: |
||||
GPR_ASSERT(shutdown_started); |
||||
shutdown_finished = 1; |
||||
break; |
||||
case GRPC_QUEUE_TIMEOUT: |
||||
break; |
||||
} |
||||
} |
||||
grpc_profiler_stop(); |
||||
grpc_call_details_destroy(&call_details); |
||||
|
||||
grpc_server_destroy(server); |
||||
grpc_completion_queue_destroy(cq); |
||||
grpc_shutdown(); |
||||
return 0; |
||||
} |
||||
Loading…
Reference in new issue