diff options
Diffstat (limited to 'test')
34 files changed, 1108 insertions, 411 deletions
diff --git a/test/core/channel/channel_stack_test.c b/test/core/channel/channel_stack_test.c index 76bb57346c..af551c4928 100644 --- a/test/core/channel/channel_stack_test.c +++ b/test/core/channel/channel_stack_test.c @@ -68,7 +68,7 @@ static void channel_destroy_func(grpc_exec_ctx *exec_ctx, static void call_destroy_func(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *ignored) { + grpc_closure *ignored) { ++*(int *)(elem->channel_data); } @@ -139,10 +139,16 @@ static void test_create_channel_stack(void) { GPR_ASSERT(*channel_data == 0); call_stack = gpr_malloc(channel_stack->call_stack_size); - grpc_error *error = - grpc_call_stack_init(&exec_ctx, channel_stack, 1, free_call, call_stack, - NULL, NULL, path, gpr_now(GPR_CLOCK_MONOTONIC), - gpr_inf_future(GPR_CLOCK_MONOTONIC), call_stack); + const grpc_call_element_args args = { + .call_stack = call_stack, + .server_transport_data = NULL, + .context = NULL, + .path = path, + .start_time = gpr_now(GPR_CLOCK_MONOTONIC), + .deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC), + .arena = NULL}; + grpc_error *error = grpc_call_stack_init(&exec_ctx, channel_stack, 1, + free_call, call_stack, &args); GPR_ASSERT(error == GRPC_ERROR_NONE); GPR_ASSERT(call_stack->count == 1); call_elem = grpc_call_stack_element(call_stack, 0); diff --git a/test/core/client_channel/BUILD b/test/core/client_channel/BUILD index a29e9aca4e..55a74c6d01 100644 --- a/test/core/client_channel/BUILD +++ b/test/core/client_channel/BUILD @@ -45,10 +45,3 @@ cc_test( deps = ["//:grpc", "//test/core/util:grpc_test_util", "//:gpr", "//test/core/util:gpr_test_util", "//test/core/end2end:cq_verifier"], copts = ['-std=c99'] ) - -cc_test( - name = "set_initial_connect_string_test", - srcs = ["set_initial_connect_string_test.c"], - deps = ["//:grpc", "//test/core/util:grpc_test_util", "//:gpr", "//test/core/util:gpr_test_util"], - copts = ['-std=c99'] -) diff --git a/test/core/client_channel/set_initial_connect_string_test.c b/test/core/client_channel/set_initial_connect_string_test.c deleted file mode 100644 index 7bed80f726..0000000000 --- a/test/core/client_channel/set_initial_connect_string_test.c +++ /dev/null @@ -1,269 +0,0 @@ -/* - * - * Copyright 2015, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -/* With the addition of a libuv endpoint, sockaddr.h now includes uv.h when - using that endpoint. Because of various transitive includes in uv.h, - including windows.h on Windows, uv.h must be included before other system - headers. Therefore, sockaddr.h must always be included first */ -#include "src/core/lib/iomgr/sockaddr.h" - -#include <string.h> - -#include <grpc/grpc.h> -#include <grpc/slice.h> -#include <grpc/support/alloc.h> -#include <grpc/support/host_port.h> -#include <grpc/support/log.h> -#include <grpc/support/thd.h> - -#include "src/core/ext/client_channel/initial_connect_string.h" -#include "src/core/lib/iomgr/sockaddr.h" -#include "src/core/lib/security/credentials/fake/fake_credentials.h" -#include "src/core/lib/slice/slice_string_helpers.h" -#include "src/core/lib/support/string.h" -#include "test/core/util/port.h" -#include "test/core/util/test_config.h" -#include "test/core/util/test_tcp_server.h" - -struct rpc_state { - char *target; - grpc_channel_credentials *creds; - grpc_completion_queue *cq; - grpc_channel *channel; - grpc_call *call; - grpc_op op; - grpc_slice_buffer incoming_buffer; - grpc_slice_buffer temp_incoming_buffer; - grpc_endpoint *tcp; - gpr_atm done_atm; -}; - -static const char *magic_connect_string = "magic initial string"; -static int server_port; -static struct rpc_state state; -static grpc_closure on_read; - -static void handle_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - GPR_ASSERT(error == GRPC_ERROR_NONE); - grpc_slice_buffer_move_into(&state.temp_incoming_buffer, - &state.incoming_buffer); - gpr_log(GPR_DEBUG, "got %" PRIuPTR " bytes, magic is %" PRIuPTR " bytes", - state.incoming_buffer.length, strlen(magic_connect_string)); - if (state.incoming_buffer.length > strlen(magic_connect_string)) { - gpr_atm_rel_store(&state.done_atm, 1); - grpc_endpoint_shutdown( - exec_ctx, state.tcp, - GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Incoming buffer longer than magic_connect_string")); - grpc_endpoint_destroy(exec_ctx, state.tcp); - } else { - grpc_endpoint_read(exec_ctx, state.tcp, &state.temp_incoming_buffer, - &on_read); - } -} - -static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, - grpc_pollset *accepting_pollset, - grpc_tcp_server_acceptor *acceptor) { - gpr_free(acceptor); - test_tcp_server *server = arg; - grpc_closure_init(&on_read, handle_read, NULL, grpc_schedule_on_exec_ctx); - grpc_slice_buffer_init(&state.incoming_buffer); - grpc_slice_buffer_init(&state.temp_incoming_buffer); - state.tcp = tcp; - grpc_endpoint_add_to_pollset(exec_ctx, tcp, server->pollset); - grpc_endpoint_read(exec_ctx, tcp, &state.temp_incoming_buffer, &on_read); -} - -static void set_magic_initial_string(grpc_resolved_address **addr, - grpc_slice *connect_string) { - GPR_ASSERT(addr); - GPR_ASSERT((*addr)->len); - *connect_string = grpc_slice_from_copied_string(magic_connect_string); -} - -static void reset_addr_and_set_magic_string(grpc_resolved_address **addr, - grpc_slice *connect_string) { - struct sockaddr_in target; - *connect_string = grpc_slice_from_copied_string(magic_connect_string); - gpr_free(*addr); - target.sin_family = AF_INET; - target.sin_addr.s_addr = htonl(INADDR_LOOPBACK); - target.sin_port = htons((uint16_t)server_port); - *addr = (grpc_resolved_address *)gpr_malloc(sizeof(grpc_resolved_address)); - (*addr)->len = sizeof(target); - memcpy((*addr)->addr, &target, sizeof(target)); -} - -static gpr_timespec n_sec_deadline(int seconds) { - return gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), - gpr_time_from_seconds(seconds, GPR_TIMESPAN)); -} - -static void start_rpc(int use_creds, int target_port) { - state.cq = grpc_completion_queue_create(NULL); - if (use_creds) { - state.creds = grpc_fake_transport_security_credentials_create(); - } else { - state.creds = NULL; - } - gpr_join_host_port(&state.target, "127.0.0.1", target_port); - if (use_creds) { - state.channel = - grpc_secure_channel_create(state.creds, state.target, NULL, NULL); - } else { - state.channel = grpc_insecure_channel_create(state.target, NULL, NULL); - } - grpc_slice host = grpc_slice_from_static_string("localhost"); - state.call = grpc_channel_create_call( - state.channel, NULL, GRPC_PROPAGATE_DEFAULTS, state.cq, - grpc_slice_from_static_string("/Service/Method"), &host, - gpr_inf_future(GPR_CLOCK_REALTIME), NULL); - memset(&state.op, 0, sizeof(state.op)); - state.op.op = GRPC_OP_SEND_INITIAL_METADATA; - state.op.data.send_initial_metadata.count = 0; - state.op.flags = 0; - state.op.reserved = NULL; - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(state.call, &state.op, - (size_t)(1), NULL, NULL)); - grpc_completion_queue_next(state.cq, n_sec_deadline(5), NULL); -} - -static void cleanup_rpc(void) { - grpc_event ev; - grpc_slice_buffer_destroy(&state.incoming_buffer); - grpc_slice_buffer_destroy(&state.temp_incoming_buffer); - grpc_channel_credentials_release(state.creds); - grpc_call_destroy(state.call); - grpc_completion_queue_shutdown(state.cq); - do { - ev = grpc_completion_queue_next(state.cq, n_sec_deadline(1), NULL); - } while (ev.type != GRPC_QUEUE_SHUTDOWN); - grpc_completion_queue_destroy(state.cq); - grpc_channel_destroy(state.channel); - gpr_free(state.target); -} - -typedef struct { - test_tcp_server *server; - gpr_event *signal_when_done; -} poll_args; - -static void actually_poll_server(void *arg) { - poll_args *pa = arg; - gpr_timespec deadline = n_sec_deadline(10); - while (true) { - bool done = gpr_atm_acq_load(&state.done_atm) != 0; - gpr_timespec time_left = - gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)); - gpr_log(GPR_DEBUG, "done=%d, time_left=%" PRId64 ".%09" PRId32, done, - time_left.tv_sec, time_left.tv_nsec); - if (done || gpr_time_cmp(time_left, gpr_time_0(GPR_TIMESPAN)) < 0) { - break; - } - test_tcp_server_poll(pa->server, 1); - } - gpr_event_set(pa->signal_when_done, (void *)1); - gpr_free(pa); -} - -static void poll_server_until_read_done(test_tcp_server *server, - gpr_event *signal_when_done) { - gpr_atm_rel_store(&state.done_atm, 0); - gpr_thd_id id; - poll_args *pa = gpr_malloc(sizeof(*pa)); - pa->server = server; - pa->signal_when_done = signal_when_done; - gpr_thd_new(&id, actually_poll_server, pa, NULL); -} - -static void match_initial_magic_string(grpc_slice_buffer *buffer) { - size_t i, j, cmp_length; - size_t magic_length = strlen(magic_connect_string); - GPR_ASSERT(buffer->length >= magic_length); - for (i = 0, j = 0; i < state.incoming_buffer.count && j < magic_length; i++) { - char *dump = grpc_slice_to_c_string(state.incoming_buffer.slices[i]); - cmp_length = GPR_MIN(strlen(dump), magic_length - j); - GPR_ASSERT(strncmp(dump, magic_connect_string + j, cmp_length) == 0); - j += cmp_length; - gpr_free(dump); - } -} - -static void test_initial_string(test_tcp_server *server, int secure) { - gpr_event ev; - gpr_event_init(&ev); - grpc_test_set_initial_connect_string_function(set_magic_initial_string); - poll_server_until_read_done(server, &ev); - start_rpc(secure, server_port); - gpr_event_wait(&ev, gpr_inf_future(GPR_CLOCK_REALTIME)); - match_initial_magic_string(&state.incoming_buffer); - cleanup_rpc(); -} - -static void test_initial_string_with_redirect(test_tcp_server *server, - int secure) { - gpr_event ev; - gpr_event_init(&ev); - int another_port = grpc_pick_unused_port_or_die(); - grpc_test_set_initial_connect_string_function( - reset_addr_and_set_magic_string); - poll_server_until_read_done(server, &ev); - start_rpc(secure, another_port); - gpr_event_wait(&ev, gpr_inf_future(GPR_CLOCK_REALTIME)); - match_initial_magic_string(&state.incoming_buffer); - cleanup_rpc(); -} - -static void run_test(void (*test)(test_tcp_server *server, int secure), - int secure) { - test_tcp_server test_server; - server_port = grpc_pick_unused_port_or_die(); - test_tcp_server_init(&test_server, on_connect, &test_server); - test_tcp_server_start(&test_server, server_port); - test(&test_server, secure); - test_tcp_server_destroy(&test_server); -} - -int main(int argc, char **argv) { - grpc_test_init(argc, argv); - grpc_init(); - - run_test(test_initial_string, 0); - run_test(test_initial_string, 1); - run_test(test_initial_string_with_redirect, 0); - run_test(test_initial_string_with_redirect, 1); - - grpc_shutdown(); - return 0; -} diff --git a/test/core/end2end/end2end_tests.h b/test/core/end2end/end2end_tests.h index cb0afd9cd9..cdb26a67e9 100644 --- a/test/core/end2end/end2end_tests.h +++ b/test/core/end2end/end2end_tests.h @@ -39,12 +39,15 @@ typedef struct grpc_end2end_test_fixture grpc_end2end_test_fixture; typedef struct grpc_end2end_test_config grpc_end2end_test_config; +/* Test feature flags. */ #define FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION 1 #define FEATURE_MASK_SUPPORTS_HOSTNAME_VERIFICATION 2 #define FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS 4 #define FEATURE_MASK_SUPPORTS_REQUEST_PROXYING 8 #define FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL 16 #define FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER 32 +#define FEATURE_MASK_DOES_NOT_SUPPORT_RESOURCE_QUOTA_SERVER 64 +#define FEATURE_MASK_DOES_NOT_SUPPORT_NETWORK_STATUS_CHANGE 128 #define FAIL_AUTH_CHECK_SERVER_ARG_NAME "fail_auth_check" @@ -56,8 +59,12 @@ struct grpc_end2end_test_fixture { }; struct grpc_end2end_test_config { + /* A descriptive name for this test fixture. */ const char *name; + + /* Which features are supported by this fixture. See feature flags above. */ uint32_t feature_mask; + grpc_end2end_test_fixture (*create_fixture)(grpc_channel_args *client_args, grpc_channel_args *server_args); void (*init_client)(grpc_end2end_test_fixture *f, diff --git a/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6723650944237568 b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6723650944237568 Binary files differnew file mode 100644 index 0000000000..e140fdc705 --- /dev/null +++ b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6723650944237568 diff --git a/test/core/end2end/tests/filter_call_init_fails.c b/test/core/end2end/tests/filter_call_init_fails.c index e2df272bab..ebfe3b03dc 100644 --- a/test/core/end2end/tests/filter_call_init_fails.c +++ b/test/core/end2end/tests/filter_call_init_fails.c @@ -213,7 +213,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) {} + grpc_closure *ignored) {} static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, diff --git a/test/core/end2end/tests/filter_causes_close.c b/test/core/end2end/tests/filter_causes_close.c index f4bfb5c804..e6b02eaeee 100644 --- a/test/core/end2end/tests/filter_causes_close.c +++ b/test/core/end2end/tests/filter_causes_close.c @@ -236,7 +236,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) {} + grpc_closure *ignored) {} static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, diff --git a/test/core/end2end/tests/filter_latency.c b/test/core/end2end/tests/filter_latency.c index d05e9e79a1..2428c92a42 100644 --- a/test/core/end2end/tests/filter_latency.c +++ b/test/core/end2end/tests/filter_latency.c @@ -267,7 +267,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) { + grpc_closure *ignored) { gpr_mu_lock(&g_mu); g_client_latency = final_info->stats.latency; gpr_mu_unlock(&g_mu); @@ -276,7 +276,7 @@ static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx, static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) { + grpc_closure *ignored) { gpr_mu_lock(&g_mu); g_server_latency = final_info->stats.latency; gpr_mu_unlock(&g_mu); diff --git a/test/core/end2end/tests/network_status_change.c b/test/core/end2end/tests/network_status_change.c index 7540ce93a1..d7a4106459 100644 --- a/test/core/end2end/tests/network_status_change.c +++ b/test/core/end2end/tests/network_status_change.c @@ -240,6 +240,10 @@ static void test_invoke_network_status_change(grpc_end2end_test_config config) { } void network_status_change(grpc_end2end_test_config config) { + if (config.feature_mask & + FEATURE_MASK_DOES_NOT_SUPPORT_NETWORK_STATUS_CHANGE) { + return; + } test_invoke_network_status_change(config); } diff --git a/test/core/end2end/tests/resource_quota_server.c b/test/core/end2end/tests/resource_quota_server.c index 4f9ed7a3a1..db26b4480e 100644 --- a/test/core/end2end/tests/resource_quota_server.c +++ b/test/core/end2end/tests/resource_quota_server.c @@ -113,6 +113,10 @@ static grpc_slice generate_random_slice() { } void resource_quota_server(grpc_end2end_test_config config) { + if (config.feature_mask & + FEATURE_MASK_DOES_NOT_SUPPORT_RESOURCE_QUOTA_SERVER) { + return; + } grpc_resource_quota *resource_quota = grpc_resource_quota_create("test_server"); grpc_resource_quota_resize(resource_quota, 5 * 1024 * 1024); diff --git a/test/core/security/BUILD b/test/core/security/BUILD index e750c39b7c..1cb03c5cfe 100644 --- a/test/core/security/BUILD +++ b/test/core/security/BUILD @@ -34,7 +34,7 @@ load("//test/core/util:grpc_fuzzer.bzl", "grpc_fuzzer") grpc_fuzzer( name = "ssl_server_fuzzer", srcs = ["ssl_server_fuzzer.c"], - deps = ["//:gpr", "//:grpc", "//test/core/util:grpc_test_util"], + deps = ["//:gpr", "//:grpc", "//test/core/util:grpc_test_util", "//test/core/end2end:ssl_test_data"], corpus = "corpus", copts = ["-std=c99"], ) diff --git a/test/core/security/ssl_server_fuzzer.c b/test/core/security/ssl_server_fuzzer.c index 0219385643..cbbaf9f298 100644 --- a/test/core/security/ssl_server_fuzzer.c +++ b/test/core/security/ssl_server_fuzzer.c @@ -38,6 +38,7 @@ #include "src/core/lib/iomgr/load_file.h" #include "src/core/lib/security/credentials/credentials.h" #include "src/core/lib/security/transport/security_connector.h" +#include "test/core/end2end/data/ssl_test_data.h" #include "test/core/util/memory_counters.h" #include "test/core/util/mock_endpoint.h" @@ -46,10 +47,6 @@ bool squelch = true; // Turning this on will fail the leak check. bool leak_check = false; -#define SSL_CERT_PATH "src/core/lib/tsi/test_creds/server1.pem" -#define SSL_KEY_PATH "src/core/lib/tsi/test_creds/server1.key" -#define SSL_CA_PATH "src/core/lib/tsi/test_creds/ca.pem" - static void discard_write(grpc_slice slice) {} static void dont_log(gpr_log_func_args *args) {} @@ -88,12 +85,9 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { // Load key pair and establish server SSL credentials. grpc_ssl_pem_key_cert_pair pem_key_cert_pair; grpc_slice ca_slice, cert_slice, key_slice; - GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", - grpc_load_file(SSL_CA_PATH, 1, &ca_slice))); - GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", - grpc_load_file(SSL_CERT_PATH, 1, &cert_slice))); - GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", - grpc_load_file(SSL_KEY_PATH, 1, &key_slice))); + ca_slice = grpc_slice_from_static_string(test_root_cert); + cert_slice = grpc_slice_from_static_string(test_server1_cert); + key_slice = grpc_slice_from_static_string(test_server1_key); const char *ca_cert = (const char *)GRPC_SLICE_START_PTR(ca_slice); pem_key_cert_pair.private_key = (const char *)GRPC_SLICE_START_PTR(key_slice); pem_key_cert_pair.cert_chain = (const char *)GRPC_SLICE_START_PTR(cert_slice); diff --git a/test/core/support/arena_test.c b/test/core/support/arena_test.c new file mode 100644 index 0000000000..35b2bbd1b1 --- /dev/null +++ b/test/core/support/arena_test.c @@ -0,0 +1,139 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/support/arena.h" + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/sync.h> +#include <grpc/support/thd.h> +#include <grpc/support/useful.h> +#include <inttypes.h> +#include <string.h> + +#include "src/core/lib/support/string.h" +#include "test/core/util/test_config.h" + +static void test_noop(void) { gpr_arena_destroy(gpr_arena_create(1)); } + +static void test(const char *name, size_t init_size, const size_t *allocs, + size_t nallocs) { + gpr_strvec v; + char *s; + gpr_strvec_init(&v); + gpr_asprintf(&s, "test '%s': %" PRIdPTR " <- {", name, init_size); + gpr_strvec_add(&v, s); + for (size_t i = 0; i < nallocs; i++) { + gpr_asprintf(&s, "%" PRIdPTR ",", allocs[i]); + gpr_strvec_add(&v, s); + } + gpr_strvec_add(&v, gpr_strdup("}")); + s = gpr_strvec_flatten(&v, NULL); + gpr_strvec_destroy(&v); + gpr_log(GPR_INFO, "%s", s); + gpr_free(s); + + gpr_arena *a = gpr_arena_create(init_size); + void **ps = gpr_zalloc(sizeof(*ps) * nallocs); + for (size_t i = 0; i < nallocs; i++) { + ps[i] = gpr_arena_alloc(a, allocs[i]); + // ensure no duplicate results + for (size_t j = 0; j < i; j++) { + GPR_ASSERT(ps[i] != ps[j]); + } + // ensure writable + memset(ps[i], 1, allocs[i]); + } + gpr_arena_destroy(a); + gpr_free(ps); +} + +#define TEST(name, init_size, ...) \ + static const size_t allocs_##name[] = {__VA_ARGS__}; \ + test(#name, init_size, allocs_##name, GPR_ARRAY_SIZE(allocs_##name)) + +#define CONCURRENT_TEST_ITERATIONS 100000 +#define CONCURRENT_TEST_THREADS 100 + +typedef struct { + gpr_event ev_start; + gpr_arena *arena; +} concurrent_test_args; + +static void concurrent_test_body(void *arg) { + concurrent_test_args *a = arg; + gpr_event_wait(&a->ev_start, gpr_inf_future(GPR_CLOCK_REALTIME)); + for (size_t i = 0; i < CONCURRENT_TEST_ITERATIONS; i++) { + *(char *)gpr_arena_alloc(a->arena, 1) = (char)i; + } +} + +static void concurrent_test(void) { + gpr_log(GPR_DEBUG, "concurrent_test"); + + concurrent_test_args args; + gpr_event_init(&args.ev_start); + args.arena = gpr_arena_create(1024); + + gpr_thd_id thds[CONCURRENT_TEST_THREADS]; + + for (int i = 0; i < CONCURRENT_TEST_THREADS; i++) { + gpr_thd_options opt = gpr_thd_options_default(); + gpr_thd_options_set_joinable(&opt); + gpr_thd_new(&thds[i], concurrent_test_body, &args, &opt); + } + + gpr_event_set(&args.ev_start, (void *)1); + + for (int i = 0; i < CONCURRENT_TEST_THREADS; i++) { + gpr_thd_join(thds[i]); + } + + gpr_arena_destroy(args.arena); +} + +int main(int argc, char *argv[]) { + grpc_test_init(argc, argv); + + test_noop(); + TEST(0_1, 0, 1); + TEST(1_1, 1, 1); + TEST(1_2, 1, 2); + TEST(1_3, 1, 3); + TEST(1_inc, 1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11); + TEST(6_123, 6, 1, 2, 3); + concurrent_test(); + + return 0; +} diff --git a/test/core/support/cpu_test.c b/test/core/support/cpu_test.c index ca0fe0ccb5..7b9bf6c5e1 100644 --- a/test/core/support/cpu_test.c +++ b/test/core/support/cpu_test.c @@ -81,9 +81,12 @@ static void worker_thread(void *arg) { uint32_t cpu; unsigned r = 12345678; unsigned i, j; - for (i = 0; i < 1000 / grpc_test_slowdown_factor(); i++) { + /* Avoid repetitive division calculations */ + int64_t max_i = 1000 / grpc_test_slowdown_factor(); + int64_t max_j = 1000000 / grpc_test_slowdown_factor(); + for (i = 0; i < max_i; i++) { /* run for a bit - just calculate something random. */ - for (j = 0; j < 1000000 / grpc_test_slowdown_factor(); j++) { + for (j = 0; j < max_j; j++) { r = (r * 17) & ((r - i) | (r * i)); } cpu = gpr_cpu_current_cpu(); diff --git a/test/core/transport/chttp2/hpack_parser_corpus/clusterfuzz-testcase-5298216461402112 b/test/core/transport/chttp2/hpack_parser_corpus/clusterfuzz-testcase-5298216461402112 Binary files differnew file mode 100644 index 0000000000..04d48d6d76 --- /dev/null +++ b/test/core/transport/chttp2/hpack_parser_corpus/clusterfuzz-testcase-5298216461402112 diff --git a/test/cpp/common/channel_arguments_test.cc b/test/cpp/common/channel_arguments_test.cc index 190d32ce06..9bcc9f99f6 100644 --- a/test/cpp/common/channel_arguments_test.cc +++ b/test/cpp/common/channel_arguments_test.cc @@ -230,13 +230,6 @@ TEST_F(ChannelArgumentsTest, SetSocketMutator) { EXPECT_TRUE(HasArg(arg1)); // arg0 is replaced by arg1 EXPECT_FALSE(HasArg(arg0)); - - // arg0 is destroyed by grpc_socket_mutator_to_arg(mutator1) - { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - arg1.value.pointer.vtable->destroy(&exec_ctx, arg1.value.pointer.p); - grpc_exec_ctx_finish(&exec_ctx); - } } TEST_F(ChannelArgumentsTest, SetUserAgentPrefix) { diff --git a/test/cpp/interop/http2_client.cc b/test/cpp/interop/http2_client.cc index 01c07823cf..38a437f39f 100644 --- a/test/cpp/interop/http2_client.cc +++ b/test/cpp/interop/http2_client.cc @@ -41,7 +41,7 @@ #include <grpc/support/useful.h> #include "src/core/lib/transport/byte_stream.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/cpp/interop/http2_client.h" diff --git a/test/cpp/interop/http2_client.h b/test/cpp/interop/http2_client.h index 12df5d26bc..e57d695205 100644 --- a/test/cpp/interop/http2_client.h +++ b/test/cpp/interop/http2_client.h @@ -38,7 +38,7 @@ #include <grpc++/channel.h> #include <grpc/grpc.h> -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" namespace grpc { diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index b7f2723c39..55ba324cc7 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -46,8 +46,8 @@ #include <grpc/support/useful.h> #include "src/core/lib/transport/byte_stream.h" -#include "src/proto/grpc/testing/empty.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/empty.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/cpp/interop/client_helper.h" #include "test/cpp/interop/interop_client.h" diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h index 74f4db6b78..efcb7d2860 100644 --- a/test/cpp/interop/interop_client.h +++ b/test/cpp/interop/interop_client.h @@ -38,7 +38,7 @@ #include <grpc++/channel.h> #include <grpc/grpc.h> -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" namespace grpc { diff --git a/test/cpp/interop/interop_server.cc b/test/cpp/interop/interop_server.cc index 5a810b45ef..1cbca17928 100644 --- a/test/cpp/interop/interop_server.cc +++ b/test/cpp/interop/interop_server.cc @@ -48,8 +48,8 @@ #include "src/core/lib/support/string.h" #include "src/core/lib/transport/byte_stream.h" -#include "src/proto/grpc/testing/empty.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/empty.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/cpp/interop/server_helper.h" #include "test/cpp/util/test_config.h" diff --git a/test/cpp/interop/reconnect_interop_client.cc b/test/cpp/interop/reconnect_interop_client.cc index 1c2f606637..01d985068d 100644 --- a/test/cpp/interop/reconnect_interop_client.cc +++ b/test/cpp/interop/reconnect_interop_client.cc @@ -40,8 +40,8 @@ #include <grpc++/support/channel_arguments.h> #include <grpc/grpc.h> #include <grpc/support/log.h> -#include "src/proto/grpc/testing/empty.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/empty.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/cpp/util/create_test_channel.h" #include "test/cpp/util/test_config.h" diff --git a/test/cpp/interop/reconnect_interop_server.cc b/test/cpp/interop/reconnect_interop_server.cc index 634d0a90fc..8d1b884af9 100644 --- a/test/cpp/interop/reconnect_interop_server.cc +++ b/test/cpp/interop/reconnect_interop_server.cc @@ -47,8 +47,8 @@ #include <grpc/grpc.h> #include <grpc/support/log.h> -#include "src/proto/grpc/testing/empty.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/empty.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/core/util/reconnect_server.h" #include "test/cpp/util/test_config.h" diff --git a/test/cpp/microbenchmarks/bm_arena.cc b/test/cpp/microbenchmarks/bm_arena.cc new file mode 100644 index 0000000000..770c0b6d47 --- /dev/null +++ b/test/cpp/microbenchmarks/bm_arena.cc @@ -0,0 +1,76 @@ +/* + * + * Copyright 2017, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* Benchmark arenas */ + +extern "C" { +#include "src/core/lib/support/arena.h" +} +#include "test/cpp/microbenchmarks/helpers.h" +#include "third_party/benchmark/include/benchmark/benchmark.h" + +static void BM_Arena_NoOp(benchmark::State& state) { + while (state.KeepRunning()) { + gpr_arena_destroy(gpr_arena_create(state.range(0))); + } +} +BENCHMARK(BM_Arena_NoOp)->Range(1, 1024 * 1024); + +static void BM_Arena_ManyAlloc(benchmark::State& state) { + gpr_arena* a = gpr_arena_create(state.range(0)); + const size_t realloc_after = + 1024 * 1024 * 1024 / ((state.range(1) + 15) & 0xffffff0u); + while (state.KeepRunning()) { + gpr_arena_alloc(a, state.range(1)); + // periodically recreate arena to avoid OOM + if (state.iterations() % realloc_after == 0) { + gpr_arena_destroy(a); + a = gpr_arena_create(state.range(0)); + } + } + gpr_arena_destroy(a); +} +BENCHMARK(BM_Arena_ManyAlloc)->Ranges({{1, 1024 * 1024}, {1, 32 * 1024}}); + +static void BM_Arena_Batch(benchmark::State& state) { + while (state.KeepRunning()) { + gpr_arena* a = gpr_arena_create(state.range(0)); + for (int i = 0; i < state.range(1); i++) { + gpr_arena_alloc(a, state.range(2)); + } + gpr_arena_destroy(a); + } +} +BENCHMARK(BM_Arena_Batch)->Ranges({{1, 64 * 1024}, {1, 64}, {1, 1024}}); + +BENCHMARK_MAIN(); diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc index 014e2b96b5..5ef40abb97 100644 --- a/test/cpp/microbenchmarks/bm_call_create.cc +++ b/test/cpp/microbenchmarks/bm_call_create.cc @@ -232,7 +232,7 @@ static void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx, static void DestroyCallElem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, - void *and_free_memory) {} + grpc_closure *then_sched_closure) {} grpc_error *InitChannelElem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_channel_element_args *args) { @@ -275,7 +275,7 @@ const char *name; /* implementation of grpc_transport_init_stream */ int InitStream(grpc_exec_ctx *exec_ctx, grpc_transport *self, grpc_stream *stream, grpc_stream_refcount *refcount, - const void *server_data) { + const void *server_data, gpr_arena *arena) { return 0; } @@ -299,7 +299,7 @@ void PerformOp(grpc_exec_ctx *exec_ctx, grpc_transport *self, /* implementation of grpc_transport_destroy_stream */ void DestroyStream(grpc_exec_ctx *exec_ctx, grpc_transport *self, - grpc_stream *stream, void *and_free_memory) {} + grpc_stream *stream, grpc_closure *then_sched_closure) {} /* implementation of grpc_transport_destroy */ void Destroy(grpc_exec_ctx *exec_ctx, grpc_transport *self) {} @@ -394,7 +394,7 @@ static void BM_IsolatedFilter(benchmark::State &state) { grpc_channel_stack *channel_stack = static_cast<grpc_channel_stack *>(gpr_zalloc(channel_size)); GPR_ASSERT(GRPC_LOG_IF_ERROR( - "call_stack_init", + "channel_stack_init", grpc_channel_stack_init(&exec_ctx, 1, FilterDestroy, channel_stack, &filters[0], filters.size(), &channel_args, fixture.flags & REQUIRES_TRANSPORT @@ -409,15 +409,29 @@ static void BM_IsolatedFilter(benchmark::State &state) { grpc_slice method = grpc_slice_from_static_string("/foo/bar"); grpc_call_final_info final_info; TestOp test_op_data; + grpc_call_element_args call_args; + call_args.call_stack = call_stack; + call_args.server_transport_data = NULL; + call_args.context = NULL; + call_args.path = method; + call_args.start_time = start_time; + call_args.deadline = deadline; + const int kArenaSize = 4096; + call_args.arena = gpr_arena_create(kArenaSize); while (state.KeepRunning()) { GRPC_ERROR_UNREF(grpc_call_stack_init(&exec_ctx, channel_stack, 1, - DoNothing, NULL, NULL, NULL, method, - start_time, deadline, call_stack)); + DoNothing, NULL, &call_args)); typename TestOp::Op op(&exec_ctx, &test_op_data, call_stack); grpc_call_stack_destroy(&exec_ctx, call_stack, &final_info, NULL); op.Finish(&exec_ctx); grpc_exec_ctx_flush(&exec_ctx); + // recreate arena every 64k iterations to avoid oom + if (0 == (state.iterations() & 0xffff)) { + gpr_arena_destroy(call_args.arena); + call_args.arena = gpr_arena_create(kArenaSize); + } } + gpr_arena_destroy(call_args.arena); grpc_channel_stack_destroy(&exec_ctx, channel_stack); grpc_exec_ctx_finish(&exec_ctx); gpr_free(channel_stack); diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc new file mode 100644 index 0000000000..254d57de20 --- /dev/null +++ b/test/cpp/microbenchmarks/bm_chttp2_transport.cc @@ -0,0 +1,587 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* Microbenchmarks around CHTTP2 transport operations */ + +#include <grpc++/support/channel_arguments.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <string.h> +#include <memory> +#include <queue> +#include <sstream> +extern "C" { +#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" +#include "src/core/ext/transport/chttp2/transport/internal.h" +#include "src/core/lib/iomgr/resource_quota.h" +#include "src/core/lib/slice/slice_internal.h" +#include "src/core/lib/transport/static_metadata.h" +} +#include "test/cpp/microbenchmarks/helpers.h" +#include "third_party/benchmark/include/benchmark/benchmark.h" + +auto &force_library_initialization = Library::get(); + +//////////////////////////////////////////////////////////////////////////////// +// Helper classes +// + +class DummyEndpoint : public grpc_endpoint { + public: + DummyEndpoint() { + static const grpc_endpoint_vtable my_vtable = {read, + write, + get_workqueue, + add_to_pollset, + add_to_pollset_set, + shutdown, + destroy, + get_resource_user, + get_peer, + get_fd}; + grpc_endpoint::vtable = &my_vtable; + ru_ = grpc_resource_user_create(Library::get().rq(), "dummy_endpoint"); + } + + void PushInput(grpc_exec_ctx *exec_ctx, grpc_slice slice) { + if (read_cb_ == nullptr) { + GPR_ASSERT(!have_slice_); + buffered_slice_ = slice; + have_slice_ = true; + return; + } + grpc_slice_buffer_add(slices_, slice); + grpc_closure_sched(exec_ctx, read_cb_, GRPC_ERROR_NONE); + read_cb_ = nullptr; + } + + private: + grpc_resource_user *ru_; + grpc_closure *read_cb_ = nullptr; + grpc_slice_buffer *slices_ = nullptr; + bool have_slice_ = false; + grpc_slice buffered_slice_; + + void QueueRead(grpc_exec_ctx *exec_ctx, grpc_slice_buffer *slices, + grpc_closure *cb) { + GPR_ASSERT(read_cb_ == nullptr); + if (have_slice_) { + have_slice_ = false; + grpc_slice_buffer_add(slices, buffered_slice_); + grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_NONE); + return; + } + read_cb_ = cb; + slices_ = slices; + } + + static void read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_slice_buffer *slices, grpc_closure *cb) { + static_cast<DummyEndpoint *>(ep)->QueueRead(exec_ctx, slices, cb); + } + + static void write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_slice_buffer *slices, grpc_closure *cb) { + grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_NONE); + } + + static grpc_workqueue *get_workqueue(grpc_endpoint *ep) { return NULL; } + + static void add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_pollset *pollset) {} + + static void add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_pollset_set *pollset) {} + + static void shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, + grpc_error *why) { + grpc_resource_user_shutdown(exec_ctx, + static_cast<DummyEndpoint *>(ep)->ru_); + grpc_closure_sched(exec_ctx, static_cast<DummyEndpoint *>(ep)->read_cb_, + why); + } + + static void destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { + grpc_resource_user_unref(exec_ctx, static_cast<DummyEndpoint *>(ep)->ru_); + delete static_cast<DummyEndpoint *>(ep); + } + + static grpc_resource_user *get_resource_user(grpc_endpoint *ep) { + return static_cast<DummyEndpoint *>(ep)->ru_; + } + static char *get_peer(grpc_endpoint *ep) { return gpr_strdup("test"); } + static int get_fd(grpc_endpoint *ep) { return 0; } +}; + +class Fixture { + public: + Fixture(const grpc::ChannelArguments &args, bool client) { + grpc_channel_args c_args = args.c_channel_args(); + ep_ = new DummyEndpoint; + t_ = grpc_create_chttp2_transport(exec_ctx(), &c_args, ep_, client); + grpc_chttp2_transport_start_reading(exec_ctx(), t_, NULL); + FlushExecCtx(); + } + + void FlushExecCtx() { grpc_exec_ctx_flush(&exec_ctx_); } + + ~Fixture() { + grpc_transport_destroy(&exec_ctx_, t_); + grpc_exec_ctx_finish(&exec_ctx_); + } + + grpc_chttp2_transport *chttp2_transport() { + return reinterpret_cast<grpc_chttp2_transport *>(t_); + } + grpc_transport *transport() { return t_; } + grpc_exec_ctx *exec_ctx() { return &exec_ctx_; } + + void PushInput(grpc_slice slice) { ep_->PushInput(exec_ctx(), slice); } + + private: + DummyEndpoint *ep_; + grpc_exec_ctx exec_ctx_ = GRPC_EXEC_CTX_INIT; + grpc_transport *t_; +}; + +static void DoNothing(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {} + +class Stream { + public: + Stream(Fixture *f) : f_(f) { + GRPC_STREAM_REF_INIT(&refcount_, 1, DoNothing, nullptr, "test_stream"); + stream_size_ = grpc_transport_stream_size(f->transport()); + stream_ = gpr_malloc(stream_size_); + arena_ = gpr_arena_create(4096); + } + + ~Stream() { + gpr_free(stream_); + gpr_arena_destroy(arena_); + } + + void Init(benchmark::State &state) { + memset(stream_, 0, stream_size_); + if ((state.iterations() & 0xffff) == 0) { + gpr_arena_destroy(arena_); + arena_ = gpr_arena_create(4096); + } + grpc_transport_init_stream(f_->exec_ctx(), f_->transport(), + static_cast<grpc_stream *>(stream_), &refcount_, + NULL, arena_); + } + + void DestroyThen(grpc_closure *closure) { + grpc_transport_destroy_stream(f_->exec_ctx(), f_->transport(), + static_cast<grpc_stream *>(stream_), closure); + } + + void Op(grpc_transport_stream_op *op) { + grpc_transport_perform_stream_op(f_->exec_ctx(), f_->transport(), + static_cast<grpc_stream *>(stream_), op); + } + + grpc_chttp2_stream *chttp2_stream() { + return static_cast<grpc_chttp2_stream *>(stream_); + } + + private: + Fixture *f_; + grpc_stream_refcount refcount_; + gpr_arena *arena_; + size_t stream_size_; + void *stream_; +}; + +class Closure : public grpc_closure { + public: + virtual ~Closure() {} +}; + +template <class F> +std::unique_ptr<Closure> MakeClosure( + F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) { + struct C : public Closure { + C(const F &f, grpc_closure_scheduler *sched) : f_(f) { + grpc_closure_init(this, Execute, this, sched); + } + F f_; + static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + static_cast<C *>(arg)->f_(exec_ctx, error); + } + }; + return std::unique_ptr<Closure>(new C(f, sched)); +} + +template <class F> +grpc_closure *MakeOnceClosure( + F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) { + struct C : public grpc_closure { + C(const F &f) : f_(f) {} + F f_; + static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + static_cast<C *>(arg)->f_(exec_ctx, error); + delete static_cast<C *>(arg); + } + }; + auto *c = new C{f}; + return grpc_closure_init(c, C::Execute, c, sched); +} + +//////////////////////////////////////////////////////////////////////////////// +// Benchmarks +// + +static void BM_StreamCreateDestroy(benchmark::State &state) { + TrackCounters track_counters; + Fixture f(grpc::ChannelArguments(), true); + Stream s(&f); + std::unique_ptr<Closure> next = + MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (!state.KeepRunning()) return; + s.Init(state); + s.DestroyThen(next.get()); + }); + grpc_closure_run(f.exec_ctx(), next.get(), GRPC_ERROR_NONE); + f.FlushExecCtx(); + track_counters.Finish(state); +} +BENCHMARK(BM_StreamCreateDestroy); + +class RepresentativeClientInitialMetadata { + public: + static std::vector<grpc_mdelem> GetElems(grpc_exec_ctx *exec_ctx) { + return { + GRPC_MDELEM_SCHEME_HTTP, GRPC_MDELEM_METHOD_POST, + grpc_mdelem_from_slices( + exec_ctx, GRPC_MDSTR_PATH, + grpc_slice_intern(grpc_slice_from_static_string("/foo/bar"))), + grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_AUTHORITY, + grpc_slice_intern(grpc_slice_from_static_string( + "foo.test.google.fr:1234"))), + GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_DEFLATE_COMMA_GZIP, + GRPC_MDELEM_TE_TRAILERS, + GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC, + grpc_mdelem_from_slices( + exec_ctx, GRPC_MDSTR_USER_AGENT, + grpc_slice_intern(grpc_slice_from_static_string( + "grpc-c/3.0.0-dev (linux; chttp2; green)")))}; + } +}; + +template <class Metadata> +static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State &state) { + TrackCounters track_counters; + Fixture f(grpc::ChannelArguments(), true); + Stream s(&f); + grpc_transport_stream_op op; + std::unique_ptr<Closure> start; + std::unique_ptr<Closure> done; + + grpc_metadata_batch b; + grpc_metadata_batch_init(&b); + b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); + std::vector<grpc_mdelem> elems = Metadata::GetElems(f.exec_ctx()); + std::vector<grpc_linked_mdelem> storage(elems.size()); + for (size_t i = 0; i < elems.size(); i++) { + GPR_ASSERT(GRPC_LOG_IF_ERROR( + "addmd", + grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i]))); + } + + f.FlushExecCtx(); + start = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (!state.KeepRunning()) return; + s.Init(state); + memset(&op, 0, sizeof(op)); + op.on_complete = done.get(); + op.send_initial_metadata = &b; + s.Op(&op); + }); + done = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + memset(&op, 0, sizeof(op)); + op.cancel_error = GRPC_ERROR_CANCELLED; + s.Op(&op); + s.DestroyThen(start.get()); + }); + grpc_closure_sched(f.exec_ctx(), start.get(), GRPC_ERROR_NONE); + f.FlushExecCtx(); + grpc_metadata_batch_destroy(f.exec_ctx(), &b); + track_counters.Finish(state); +} +BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy, + RepresentativeClientInitialMetadata); + +static void BM_TransportEmptyOp(benchmark::State &state) { + TrackCounters track_counters; + Fixture f(grpc::ChannelArguments(), true); + Stream s(&f); + s.Init(state); + grpc_transport_stream_op op; + std::unique_ptr<Closure> c = + MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (!state.KeepRunning()) return; + memset(&op, 0, sizeof(op)); + op.on_complete = c.get(); + s.Op(&op); + }); + grpc_closure_sched(f.exec_ctx(), c.get(), GRPC_ERROR_NONE); + f.FlushExecCtx(); + s.DestroyThen( + MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {})); + f.FlushExecCtx(); + track_counters.Finish(state); +} +BENCHMARK(BM_TransportEmptyOp); + +static void BM_TransportStreamSend(benchmark::State &state) { + TrackCounters track_counters; + Fixture f(grpc::ChannelArguments(), true); + Stream s(&f); + s.Init(state); + grpc_transport_stream_op op; + grpc_slice_buffer_stream send_stream; + grpc_slice_buffer send_buffer; + grpc_slice_buffer_init(&send_buffer); + grpc_slice_buffer_add(&send_buffer, gpr_slice_malloc(state.range(0))); + memset(GRPC_SLICE_START_PTR(send_buffer.slices[0]), 0, + GRPC_SLICE_LENGTH(send_buffer.slices[0])); + + grpc_metadata_batch b; + grpc_metadata_batch_init(&b); + b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); + std::vector<grpc_mdelem> elems = + RepresentativeClientInitialMetadata::GetElems(f.exec_ctx()); + std::vector<grpc_linked_mdelem> storage(elems.size()); + for (size_t i = 0; i < elems.size(); i++) { + GPR_ASSERT(GRPC_LOG_IF_ERROR( + "addmd", + grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i]))); + } + + std::unique_ptr<Closure> c = + MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (!state.KeepRunning()) return; + // force outgoing window to be yuge + s.chttp2_stream()->outgoing_window_delta = 1024 * 1024 * 1024; + f.chttp2_transport()->outgoing_window = 1024 * 1024 * 1024; + grpc_slice_buffer_stream_init(&send_stream, &send_buffer, 0); + memset(&op, 0, sizeof(op)); + op.on_complete = c.get(); + op.send_message = &send_stream.base; + s.Op(&op); + }); + + memset(&op, 0, sizeof(op)); + op.send_initial_metadata = &b; + op.on_complete = c.get(); + s.Op(&op); + + f.FlushExecCtx(); + memset(&op, 0, sizeof(op)); + op.cancel_error = GRPC_ERROR_CANCELLED; + s.Op(&op); + s.DestroyThen( + MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {})); + f.FlushExecCtx(); + track_counters.Finish(state); + grpc_metadata_batch_destroy(f.exec_ctx(), &b); + grpc_slice_buffer_destroy(&send_buffer); +} +BENCHMARK(BM_TransportStreamSend)->Range(0, 128 * 1024 * 1024); + +#define SLICE_FROM_BUFFER(s) grpc_slice_from_static_buffer(s, sizeof(s) - 1) + +static grpc_slice CreateIncomingDataSlice(size_t length, size_t frame_size) { + std::queue<char> unframed; + + unframed.push(static_cast<uint8_t>(0)); + unframed.push(static_cast<uint8_t>(length >> 24)); + unframed.push(static_cast<uint8_t>(length >> 16)); + unframed.push(static_cast<uint8_t>(length >> 8)); + unframed.push(static_cast<uint8_t>(length)); + for (size_t i = 0; i < length; i++) { + unframed.push('a'); + } + + std::vector<char> framed; + while (unframed.size() > frame_size) { + // frame size + framed.push_back(static_cast<uint8_t>(frame_size >> 16)); + framed.push_back(static_cast<uint8_t>(frame_size >> 8)); + framed.push_back(static_cast<uint8_t>(frame_size)); + // data frame + framed.push_back(0); + // no flags + framed.push_back(0); + // stream id + framed.push_back(0); + framed.push_back(0); + framed.push_back(0); + framed.push_back(1); + // frame data + for (size_t i = 0; i < frame_size; i++) { + framed.push_back(unframed.front()); + unframed.pop(); + } + } + + // frame size + framed.push_back(static_cast<uint8_t>(unframed.size() >> 16)); + framed.push_back(static_cast<uint8_t>(unframed.size() >> 8)); + framed.push_back(static_cast<uint8_t>(unframed.size())); + // data frame + framed.push_back(0); + // no flags + framed.push_back(0); + // stream id + framed.push_back(0); + framed.push_back(0); + framed.push_back(0); + framed.push_back(1); + while (!unframed.empty()) { + framed.push_back(unframed.front()); + unframed.pop(); + } + + return grpc_slice_from_copied_buffer(framed.data(), framed.size()); +} + +static void BM_TransportStreamRecv(benchmark::State &state) { + TrackCounters track_counters; + Fixture f(grpc::ChannelArguments(), true); + Stream s(&f); + s.Init(state); + grpc_transport_stream_op op; + grpc_byte_stream *recv_stream; + grpc_slice incoming_data = CreateIncomingDataSlice(state.range(0), 16384); + + grpc_metadata_batch b; + grpc_metadata_batch_init(&b); + grpc_metadata_batch b_recv; + grpc_metadata_batch_init(&b_recv); + b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); + std::vector<grpc_mdelem> elems = + RepresentativeClientInitialMetadata::GetElems(f.exec_ctx()); + std::vector<grpc_linked_mdelem> storage(elems.size()); + for (size_t i = 0; i < elems.size(); i++) { + GPR_ASSERT(GRPC_LOG_IF_ERROR( + "addmd", + grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i]))); + } + + std::unique_ptr<Closure> do_nothing = + MakeClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {}); + + uint32_t received; + + std::unique_ptr<Closure> drain_start; + std::unique_ptr<Closure> drain; + std::unique_ptr<Closure> drain_continue; + grpc_slice recv_slice; + + std::unique_ptr<Closure> c = + MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (!state.KeepRunning()) return; + // force outgoing window to be yuge + s.chttp2_stream()->incoming_window_delta = 1024 * 1024 * 1024; + f.chttp2_transport()->incoming_window = 1024 * 1024 * 1024; + received = 0; + memset(&op, 0, sizeof(op)); + op.on_complete = do_nothing.get(); + op.recv_message = &recv_stream; + op.recv_message_ready = drain_start.get(); + s.Op(&op); + f.PushInput(grpc_slice_ref(incoming_data)); + }); + + drain_start = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + if (recv_stream == NULL) { + GPR_ASSERT(!state.KeepRunning()); + return; + } + grpc_closure_run(exec_ctx, drain.get(), GRPC_ERROR_NONE); + }); + + drain = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + do { + if (received == recv_stream->length) { + grpc_byte_stream_destroy(exec_ctx, recv_stream); + grpc_closure_sched(exec_ctx, c.get(), GRPC_ERROR_NONE); + return; + } + } while (grpc_byte_stream_next(exec_ctx, recv_stream, &recv_slice, + recv_stream->length - received, + drain_continue.get())); + }); + + drain_continue = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { + received += GRPC_SLICE_LENGTH(recv_slice); + grpc_slice_unref_internal(exec_ctx, recv_slice); + grpc_closure_run(exec_ctx, drain.get(), GRPC_ERROR_NONE); + }); + + memset(&op, 0, sizeof(op)); + op.send_initial_metadata = &b; + op.recv_initial_metadata = &b_recv; + op.on_complete = c.get(); + s.Op(&op); + f.PushInput(SLICE_FROM_BUFFER( + "\x00\x00\x00\x04\x00\x00\x00\x00\x00" + // Generated using: + // tools/codegen/core/gen_header_frame.py < + // test/cpp/microbenchmarks/representative_server_initial_metadata.headers + "\x00\x00X\x01\x04\x00\x00\x00\x01" + "\x10\x07:status\x03" + "200" + "\x10\x0c" + "content-type\x10" + "application/grpc" + "\x10\x14grpc-accept-encoding\x15identity,deflate,gzip")); + + f.FlushExecCtx(); + memset(&op, 0, sizeof(op)); + op.cancel_error = GRPC_ERROR_CANCELLED; + s.Op(&op); + s.DestroyThen( + MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {})); + f.FlushExecCtx(); + track_counters.Finish(state); + grpc_metadata_batch_destroy(f.exec_ctx(), &b); + grpc_metadata_batch_destroy(f.exec_ctx(), &b_recv); + grpc_slice_unref(incoming_data); +} +BENCHMARK(BM_TransportStreamRecv)->Range(0, 128 * 1024 * 1024); + +BENCHMARK_MAIN(); diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc index dc0e7d769a..00e37f7912 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc @@ -54,86 +54,141 @@ auto& force_library_initialization = Library::get(); static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); } -template <class Fixture> -static void BM_PumpStreamClientToServer(benchmark::State& state) { +// Repeatedly makes Streaming Bidi calls (exchanging a configurable number of +// messages in each call) in a loop on a single channel +// +// First parmeter (i.e state.range(0)): Message size (in bytes) to use +// Second parameter (i.e state.range(1)): Number of ping pong messages. +// Note: One ping-pong means two messages (one from client to server and +// the other from server to client): +template <class Fixture, class ClientContextMutator, class ServerContextMutator> +static void BM_StreamingPingPong(benchmark::State& state) { + const int msg_size = state.range(0); + const int max_ping_pongs = state.range(1); + EchoTestService::AsyncService service; std::unique_ptr<Fixture> fixture(new Fixture(&service)); { + EchoResponse send_response; + EchoResponse recv_response; EchoRequest send_request; EchoRequest recv_request; - if (state.range(0) > 0) { - send_request.set_message(std::string(state.range(0), 'a')); + + if (msg_size > 0) { + send_request.set_message(std::string(msg_size, 'a')); + send_response.set_message(std::string(msg_size, 'b')); } - Status recv_status; - ServerContext svr_ctx; - ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx); - service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(), - fixture->cq(), tag(0)); + std::unique_ptr<EchoTestService::Stub> stub( EchoTestService::NewStub(fixture->channel())); - ClientContext cli_ctx; - auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1)); - int need_tags = (1 << 0) | (1 << 1); - void* t; - bool ok; - while (need_tags) { - GPR_ASSERT(fixture->cq()->Next(&t, &ok)); - GPR_ASSERT(ok); - int i = (int)(intptr_t)t; - GPR_ASSERT(need_tags & (1 << i)); - need_tags &= ~(1 << i); - } - response_rw.Read(&recv_request, tag(0)); + while (state.KeepRunning()) { - GPR_TIMER_SCOPE("BenchmarkCycle", 0); - request_rw->Write(send_request, tag(1)); - while (true) { + ServerContext svr_ctx; + ServerContextMutator svr_ctx_mut(&svr_ctx); + ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx); + service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(), + fixture->cq(), tag(0)); + + ClientContext cli_ctx; + ClientContextMutator cli_ctx_mut(&cli_ctx); + auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1)); + + // Establish async stream between client side and server side + void* t; + bool ok; + int need_tags = (1 << 0) | (1 << 1); + while (need_tags) { GPR_ASSERT(fixture->cq()->Next(&t, &ok)); - if (t == tag(0)) { - response_rw.Read(&recv_request, tag(0)); - } else if (t == tag(1)) { - break; - } else { - GPR_ASSERT(false); + GPR_ASSERT(ok); + int i = (int)(intptr_t)t; + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); + } + + // Send 'max_ping_pongs' number of ping pong messages + int ping_pong_cnt = 0; + while (ping_pong_cnt < max_ping_pongs) { + request_rw->Write(send_request, tag(0)); // Start client send + response_rw.Read(&recv_request, tag(1)); // Start server recv + request_rw->Read(&recv_response, tag(2)); // Start client recv + + need_tags = (1 << 0) | (1 << 1) | (1 << 2) | (1 << 3); + while (need_tags) { + GPR_ASSERT(fixture->cq()->Next(&t, &ok)); + GPR_ASSERT(ok); + int i = (int)(intptr_t)t; + + // If server recv is complete, start the server send operation + if (i == 1) { + response_rw.Write(send_response, tag(3)); + } + + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); } + + ping_pong_cnt++; } - } - request_rw->WritesDone(tag(1)); - need_tags = (1 << 0) | (1 << 1); - while (need_tags) { - GPR_ASSERT(fixture->cq()->Next(&t, &ok)); - int i = (int)(intptr_t)t; - GPR_ASSERT(need_tags & (1 << i)); - need_tags &= ~(1 << i); + + request_rw->WritesDone(tag(0)); + response_rw.Finish(Status::OK, tag(1)); + + Status recv_status; + request_rw->Finish(&recv_status, tag(2)); + + need_tags = (1 << 0) | (1 << 1) | (1 << 2); + while (need_tags) { + GPR_ASSERT(fixture->cq()->Next(&t, &ok)); + int i = (int)(intptr_t)t; + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); + } + + GPR_ASSERT(recv_status.ok()); } } + fixture->Finish(state); fixture.reset(); - state.SetBytesProcessed(state.range(0) * state.iterations()); + state.SetBytesProcessed(msg_size * state.iterations() * max_ping_pongs * 2); } -template <class Fixture> -static void BM_PumpStreamServerToClient(benchmark::State& state) { +// Repeatedly sends ping pong messages in a single streaming Bidi call in a loop +// First parmeter (i.e state.range(0)): Message size (in bytes) to use +template <class Fixture, class ClientContextMutator, class ServerContextMutator> +static void BM_StreamingPingPongMsgs(benchmark::State& state) { + const int msg_size = state.range(0); + EchoTestService::AsyncService service; std::unique_ptr<Fixture> fixture(new Fixture(&service)); { EchoResponse send_response; EchoResponse recv_response; - if (state.range(0) > 0) { - send_response.set_message(std::string(state.range(0), 'a')); + EchoRequest send_request; + EchoRequest recv_request; + + if (msg_size > 0) { + send_request.set_message(std::string(msg_size, 'a')); + send_response.set_message(std::string(msg_size, 'b')); } - Status recv_status; + + std::unique_ptr<EchoTestService::Stub> stub( + EchoTestService::NewStub(fixture->channel())); + ServerContext svr_ctx; + ServerContextMutator svr_ctx_mut(&svr_ctx); ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx); service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(), fixture->cq(), tag(0)); - std::unique_ptr<EchoTestService::Stub> stub( - EchoTestService::NewStub(fixture->channel())); + ClientContext cli_ctx; + ClientContextMutator cli_ctx_mut(&cli_ctx); auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1)); - int need_tags = (1 << 0) | (1 << 1); + + // Establish async stream between client side and server side void* t; bool ok; + int need_tags = (1 << 0) | (1 << 1); while (need_tags) { GPR_ASSERT(fixture->cq()->Next(&t, &ok)); GPR_ASSERT(ok); @@ -141,54 +196,78 @@ static void BM_PumpStreamServerToClient(benchmark::State& state) { GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } - request_rw->Read(&recv_response, tag(0)); + while (state.KeepRunning()) { GPR_TIMER_SCOPE("BenchmarkCycle", 0); - response_rw.Write(send_response, tag(1)); - while (true) { + request_rw->Write(send_request, tag(0)); // Start client send + response_rw.Read(&recv_request, tag(1)); // Start server recv + request_rw->Read(&recv_response, tag(2)); // Start client recv + + need_tags = (1 << 0) | (1 << 1) | (1 << 2) | (1 << 3); + while (need_tags) { GPR_ASSERT(fixture->cq()->Next(&t, &ok)); - if (t == tag(0)) { - request_rw->Read(&recv_response, tag(0)); - } else if (t == tag(1)) { - break; - } else { - GPR_ASSERT(false); + GPR_ASSERT(ok); + int i = (int)(intptr_t)t; + + // If server recv is complete, start the server send operation + if (i == 1) { + response_rw.Write(send_response, tag(3)); } + + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); } } + + request_rw->WritesDone(tag(0)); response_rw.Finish(Status::OK, tag(1)); - need_tags = (1 << 0) | (1 << 1); + Status recv_status; + request_rw->Finish(&recv_status, tag(2)); + + need_tags = (1 << 0) | (1 << 1) | (1 << 2); while (need_tags) { GPR_ASSERT(fixture->cq()->Next(&t, &ok)); int i = (int)(intptr_t)t; GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } + + GPR_ASSERT(recv_status.ok()); } + fixture->Finish(state); fixture.reset(); - state.SetBytesProcessed(state.range(0) * state.iterations()); + state.SetBytesProcessed(msg_size * state.iterations() * 2); } /******************************************************************************* * CONFIGURATIONS */ -BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, TCP) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, UDS) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, SockPair) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, InProcessCHTTP2) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, TCP) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, UDS) - ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair) +// Generate Args for StreamingPingPong benchmarks. Currently generates args for +// only "small streams" (i.e streams with 0, 1 or 2 messages) +static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) { + int msg_size = 0; + + b->Args({0, 0}); // spl case: 0 ping-pong msgs (msg_size doesn't matter here) + + for (msg_size = 0; msg_size <= 128 * 1024 * 1024; + msg_size == 0 ? msg_size++ : msg_size *= 8) { + b->Args({msg_size, 1}); + b->Args({msg_size, 2}); + } +} + +BENCHMARK_TEMPLATE(BM_StreamingPingPong, InProcessCHTTP2, NoOpMutator, + NoOpMutator) + ->Apply(StreamingPingPongArgs); +BENCHMARK_TEMPLATE(BM_StreamingPingPong, TCP, NoOpMutator, NoOpMutator) + ->Apply(StreamingPingPongArgs); + +BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator, + NoOpMutator) ->Range(0, 128 * 1024 * 1024); -BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) +BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator) ->Range(0, 128 * 1024 * 1024); } // namespace testing diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index baa9304cc2..25a19a5a74 100644 --- a/test/cpp/qps/client.h +++ b/test/cpp/qps/client.h @@ -46,7 +46,7 @@ #include <grpc/support/log.h> #include <grpc/support/time.h> -#include "src/proto/grpc/testing/payloads.grpc.pb.h" +#include "src/proto/grpc/testing/payloads.pb.h" #include "src/proto/grpc/testing/services.grpc.pb.h" #include "test/cpp/qps/histogram.h" diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h index e72d30a4ef..dd32a16c87 100644 --- a/test/cpp/qps/driver.h +++ b/test/cpp/qps/driver.h @@ -36,7 +36,7 @@ #include <memory> -#include "src/proto/grpc/testing/control.grpc.pb.h" +#include "src/proto/grpc/testing/control.pb.h" #include "test/cpp/qps/histogram.h" namespace grpc { diff --git a/test/cpp/qps/histogram.h b/test/cpp/qps/histogram.h index acb415f0a1..470a394301 100644 --- a/test/cpp/qps/histogram.h +++ b/test/cpp/qps/histogram.h @@ -35,7 +35,7 @@ #define TEST_QPS_HISTOGRAM_H #include <grpc/support/histogram.h> -#include "src/proto/grpc/testing/stats.grpc.pb.h" +#include "src/proto/grpc/testing/stats.pb.h" namespace grpc { namespace testing { diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h index 821d5935be..8fbf37a095 100644 --- a/test/cpp/qps/server.h +++ b/test/cpp/qps/server.h @@ -38,8 +38,8 @@ #include <grpc/support/cpu.h> #include <vector> -#include "src/proto/grpc/testing/control.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/control.pb.h" +#include "src/proto/grpc/testing/messages.pb.h" #include "test/core/end2end/data/ssl_test_data.h" #include "test/core/util/port.h" #include "test/cpp/qps/usage_timer.h" diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc index b3a06aeaf5..b58d91eea6 100644 --- a/test/cpp/qps/server_async.cc +++ b/test/cpp/qps/server_async.cc @@ -103,24 +103,25 @@ class AsyncQpsServerTest final : public grpc::testing::Server { server_ = builder.BuildAndStart(); - using namespace std::placeholders; - auto process_rpc_bound = - std::bind(process_rpc, config.payload_config(), _1, _2); + std::bind(process_rpc, config.payload_config(), std::placeholders::_1, + std::placeholders::_2); for (int i = 0; i < 15000; i++) { for (int j = 0; j < num_threads; j++) { if (request_unary_function) { - auto request_unary = - std::bind(request_unary_function, &async_service_, _1, _2, _3, - srv_cqs_[j].get(), srv_cqs_[j].get(), _4); + auto request_unary = std::bind( + request_unary_function, &async_service_, std::placeholders::_1, + std::placeholders::_2, std::placeholders::_3, srv_cqs_[j].get(), + srv_cqs_[j].get(), std::placeholders::_4); contexts_.emplace_back( new ServerRpcContextUnaryImpl(request_unary, process_rpc_bound)); } if (request_streaming_function) { - auto request_streaming = - std::bind(request_streaming_function, &async_service_, _1, _2, - srv_cqs_[j].get(), srv_cqs_[j].get(), _3); + auto request_streaming = std::bind( + request_streaming_function, &async_service_, + std::placeholders::_1, std::placeholders::_2, srv_cqs_[j].get(), + srv_cqs_[j].get(), std::placeholders::_3); contexts_.emplace_back(new ServerRpcContextStreamingImpl( request_streaming, process_rpc_bound)); } diff --git a/test/http2_test/http2_server_health_check.py b/test/http2_test/http2_server_health_check.py new file mode 100644 index 0000000000..dd9402b855 --- /dev/null +++ b/test/http2_test/http2_server_health_check.py @@ -0,0 +1,49 @@ +# Copyright 2017, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import argparse +import hyper +import sys + +# Utility to healthcheck the http2 server. Used when starting the server to +# verify that the server is live before tests begin. +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('--server_host', type=str, default='localhost') + parser.add_argument('--server_port', type=int, default=8080) + args = parser.parse_args() + server_host = args.server_host + server_port = args.server_port + conn = hyper.HTTP20Connection('%s:%d' % (server_host, server_port)) + conn.request('POST', '/grpc.testing.TestService/UnaryCall') + resp = conn.get_response() + if resp.headers.get('grpc-encoding') is None: + sys.exit(1) + else: + sys.exit(0) diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py index abde3433ad..46c3e00d18 100644 --- a/test/http2_test/http2_test_server.py +++ b/test/http2_test/http2_test_server.py @@ -31,6 +31,7 @@ import argparse import logging +import sys import twisted import twisted.internet import twisted.internet.endpoints @@ -53,9 +54,11 @@ _TEST_CASE_MAPPING = { 'max_streams': test_max_streams.TestcaseSettingsMaxStreams, } +_exit_code = 0 + class H2Factory(twisted.internet.protocol.Factory): def __init__(self, testcase): - logging.info('Creating H2Factory for new connection.') + logging.info('Creating H2Factory for new connection (%s)', testcase) self._num_streams = 0 self._testcase = testcase @@ -83,6 +86,17 @@ def parse_arguments(): ) return parser.parse_args() +def listen(endpoint, test_case): + deferred = endpoint.listen(H2Factory(test_case)) + def listen_error(reason): + # If listening fails, we stop the reactor and exit the program + # with exit code 1. + global _exit_code + _exit_code = 1 + logging.error('Listening failed: %s' % reason.value) + twisted.internet.reactor.stop() + deferred.addErrback(listen_error) + def start_test_servers(base_port): """ Start one server per test case on incrementing port numbers beginning with base_port """ @@ -92,7 +106,9 @@ def start_test_servers(base_port): logging.warning('serving on port %d : %s'%(portnum, test_case)) endpoint = twisted.internet.endpoints.TCP4ServerEndpoint( twisted.internet.reactor, portnum, backlog=128) - endpoint.listen(H2Factory(test_case)) + # Wait until the reactor is running before calling endpoint.listen(). + twisted.internet.reactor.callWhenRunning(listen, endpoint, test_case) + index += 1 if __name__ == '__main__': @@ -102,3 +118,4 @@ if __name__ == '__main__': args = parse_arguments() start_test_servers(args.base_port) twisted.internet.reactor.run() + sys.exit(_exit_code) |