diff options
| -rw-r--r-- | include/grpc++/client_context.h | 11 | ||||
| -rw-r--r-- | src/cpp/client/client_context.cc | 23 | ||||
| -rw-r--r-- | src/python/grpcio/setup.py | 2 | ||||
| -rw-r--r-- | test/core/client_config/lb_policies_test.c | 180 | ||||
| -rw-r--r-- | test/core/httpcli/httpcli_test.c | 21 | ||||
| -rw-r--r-- | test/core/httpcli/httpscli_test.c | 21 | ||||
| -rwxr-xr-x | tools/run_tests/run_lcov.sh | 2 |
7 files changed, 223 insertions, 37 deletions
diff --git a/include/grpc++/client_context.h b/include/grpc++/client_context.h index 1b6769dee1..8e7c3579e3 100644 --- a/include/grpc++/client_context.h +++ b/include/grpc++/client_context.h @@ -280,6 +280,17 @@ class ClientContext { /// There is no guarantee the call will be cancelled. void TryCancel(); + /// Global Callbacks + /// + /// Can be set exactly once per application to install hooks whenever + /// a client context is constructed and destructed. + class GlobalCallbacks { + public: + virtual void DefaultConstructor(ClientContext* context) = 0; + virtual void Destructor(ClientContext* context) = 0; + }; + static void SetGlobalCallbacks(GlobalCallbacks* callbacks); + private: // Disallow copy and assign. ClientContext(const ClientContext&); diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc index 9bb358b233..2aa532808c 100644 --- a/src/cpp/client/client_context.cc +++ b/src/cpp/client/client_context.cc @@ -45,17 +45,31 @@ namespace grpc { +class DefaultGlobalClientCallbacks GRPC_FINAL + : public ClientContext::GlobalCallbacks { + public: + void DefaultConstructor(ClientContext* context) GRPC_OVERRIDE {} + void Destructor(ClientContext* context) GRPC_OVERRIDE {} +}; + +static DefaultGlobalClientCallbacks g_default_client_callbacks; +static ClientContext::GlobalCallbacks* g_client_callbacks = + &g_default_client_callbacks; + ClientContext::ClientContext() : initial_metadata_received_(false), call_(nullptr), call_canceled_(false), deadline_(gpr_inf_future(GPR_CLOCK_REALTIME)), - propagate_from_call_(nullptr) {} + propagate_from_call_(nullptr) { + g_client_callbacks->DefaultConstructor(this); +} ClientContext::~ClientContext() { if (call_) { grpc_call_destroy(call_); } + g_client_callbacks->Destructor(this); } std::unique_ptr<ClientContext> ClientContext::FromServerContext( @@ -124,4 +138,11 @@ grpc::string ClientContext::peer() const { return peer; } +void ClientContext::SetGlobalCallbacks(GlobalCallbacks* client_callbacks) { + GPR_ASSERT(g_client_callbacks == &g_default_client_callbacks); + GPR_ASSERT(client_callbacks != NULL); + GPR_ASSERT(client_callbacks != &g_default_client_callbacks); + g_client_callbacks = client_callbacks; +} + } // namespace grpc diff --git a/src/python/grpcio/setup.py b/src/python/grpcio/setup.py index a948ca1fac..b8a98c3d85 100644 --- a/src/python/grpcio/setup.py +++ b/src/python/grpcio/setup.py @@ -163,7 +163,7 @@ else: setuptools.setup( name='grpcio', - version='0.11.0b2', + version='0.12.0b0', ext_modules=CYTHON_EXTENSION_MODULES, packages=list(PACKAGES), package_dir=PACKAGE_DIRECTORIES, diff --git a/test/core/client_config/lb_policies_test.c b/test/core/client_config/lb_policies_test.c index 190bed0b41..9da4a4eec7 100644 --- a/test/core/client_config/lb_policies_test.c +++ b/test/core/client_config/lb_policies_test.c @@ -38,18 +38,19 @@ #include <grpc/support/alloc.h> #include <grpc/support/host_port.h> #include <grpc/support/log.h> -#include <grpc/support/time.h> #include <grpc/support/string_util.h> +#include <grpc/support/time.h> #include "src/core/channel/channel_stack.h" #include "src/core/channel/client_channel.h" +#include "src/core/client_config/lb_policies/round_robin.h" #include "src/core/client_config/lb_policy_registry.h" -#include "src/core/surface/channel.h" #include "src/core/support/string.h" +#include "src/core/surface/channel.h" #include "src/core/surface/server.h" -#include "test/core/util/test_config.h" -#include "test/core/util/port.h" #include "test/core/end2end/cq_verifier.h" +#include "test/core/util/port.h" +#include "test/core/util/test_config.h" typedef struct servers_fixture { size_t num_servers; @@ -136,8 +137,9 @@ static void kill_server(const servers_fixture *f, size_t i) { gpr_log(GPR_INFO, "KILLING SERVER %d", i); GPR_ASSERT(f->servers[i] != NULL); grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), - NULL).type == GRPC_OP_COMPLETE); + GPR_ASSERT( + grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), NULL) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->servers[i]); f->servers[i] = NULL; } @@ -203,8 +205,8 @@ static void teardown_servers(servers_fixture *f) { if (f->servers[i] == NULL) continue; grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), - n_millis_time(5000), - NULL).type == GRPC_OP_COMPLETE); + n_millis_time(5000), NULL) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->servers[i]); } grpc_completion_queue_shutdown(f->cq); @@ -225,8 +227,8 @@ static void teardown_servers(servers_fixture *f) { } /** Returns connection sequence (server indices), which must be freed */ -int *perform_request(servers_fixture *f, grpc_channel *client, - request_data *rdata, const test_spec *spec) { +static int *perform_request(servers_fixture *f, grpc_channel *client, + request_data *rdata, const test_spec *spec) { grpc_call *c; int s_idx; int *s_valid; @@ -242,8 +244,6 @@ int *perform_request(servers_fixture *f, grpc_channel *client, s_valid = gpr_malloc(sizeof(int) * f->num_servers); connection_sequence = gpr_malloc(sizeof(int) * spec->num_iters); - /* Send a trivial request. */ - for (iter_num = 0; iter_num < spec->num_iters; iter_num++) { cq_verifier *cqv = cq_verifier_create(f->cq); rdata->details = NULL; @@ -304,8 +304,8 @@ int *perform_request(servers_fixture *f, grpc_channel *client, s_idx = -1; while ((ev = grpc_completion_queue_next( - f->cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1), NULL)).type != - GRPC_QUEUE_TIMEOUT) { + f->cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1), NULL)) + .type != GRPC_QUEUE_TIMEOUT) { GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); read_tag = ((int)(gpr_intptr)ev.tag); gpr_log(GPR_DEBUG, "EVENT: success:%d, type:%d, tag:%d iter:%d", @@ -324,8 +324,6 @@ int *perform_request(servers_fixture *f, grpc_channel *client, } } - gpr_log(GPR_DEBUG, "s_idx=%d", s_idx); - if (s_idx >= 0) { op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -371,7 +369,7 @@ int *perform_request(servers_fixture *f, grpc_channel *client, &rdata->call_details[s_idx], &f->request_metadata_recv[s_idx], f->cq, f->cq, tag(1000 + (int)s_idx))); - } else { + } else { /* no response from server */ grpc_call_cancel(c, NULL); if (!completed_client) { cq_expect_completion(cqv, tag(1), 1); @@ -397,6 +395,42 @@ int *perform_request(servers_fixture *f, grpc_channel *client, return connection_sequence; } +static grpc_call **perform_multirequest(servers_fixture *f, + grpc_channel *client, + size_t concurrent_calls) { + grpc_call **calls; + grpc_op ops[6]; + grpc_op *op; + size_t i; + + calls = gpr_malloc(sizeof(grpc_call *) * concurrent_calls); + for (i = 0; i < f->num_servers; i++) { + kill_server(f, i); + } + + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; + op->reserved = NULL; + + for (i = 0; i < concurrent_calls; i++) { + calls[i] = grpc_channel_create_call( + client, NULL, GRPC_PROPAGATE_DEFAULTS, f->cq, "/foo", + "foo.test.google.fr", gpr_inf_future(GPR_CLOCK_REALTIME), NULL); + GPR_ASSERT(calls[i]); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(calls[i], ops, + (size_t)(op - ops), tag(1), + NULL)); + } + + return calls; +} + static void assert_channel_connectivity( grpc_channel *ch, size_t num_accepted_conn_states, grpc_connectivity_state accepted_conn_state, ...) { @@ -487,8 +521,110 @@ void run_spec(const test_spec *spec) { gpr_free(actual_connection_sequence); gpr_free(rdata.call_details); + grpc_channel_destroy(client); /* calls the LB's shutdown func */ + teardown_servers(f); +} + +static grpc_channel *create_client(const servers_fixture *f) { + grpc_channel *client; + char *client_hostport; + char *servers_hostports_str; + grpc_arg arg; + grpc_channel_args args; + + servers_hostports_str = gpr_strjoin_sep((const char **)f->servers_hostports, + f->num_servers, ",", NULL); + gpr_asprintf(&client_hostport, "ipv4:%s?lb_policy=round_robin", + servers_hostports_str); + + arg.type = GRPC_ARG_INTEGER; + arg.key = "grpc.testing.fixed_reconnect_backoff"; + arg.value.integer = 100; + args.num_args = 1; + args.args = &arg; + + client = grpc_insecure_channel_create(client_hostport, &args, NULL); + gpr_free(client_hostport); + gpr_free(servers_hostports_str); + + return client; +} + +static void test_ping() { + grpc_channel *client; + request_data rdata; + servers_fixture *f; + cq_verifier *cqv; + grpc_connectivity_state state = GRPC_CHANNEL_IDLE; + const size_t num_servers = 1; + int i; + + rdata.call_details = gpr_malloc(sizeof(grpc_call_details) * num_servers); + f = setup_servers("127.0.0.1", &rdata, num_servers); + cqv = cq_verifier_create(f->cq); + + client = create_client(f); + + grpc_channel_ping(client, f->cq, tag(0), NULL); + cq_expect_completion(cqv, tag(0), 0); + + /* check that we're still in idle, and start connecting */ + GPR_ASSERT(grpc_channel_check_connectivity_state(client, 1) == + GRPC_CHANNEL_IDLE); + /* we'll go through some set of transitions (some might be missed), until + READY is reached */ + while (state != GRPC_CHANNEL_READY) { + grpc_channel_watch_connectivity_state( + client, state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f->cq, tag(99)); + cq_expect_completion(cqv, tag(99), 1); + cq_verify(cqv); + state = grpc_channel_check_connectivity_state(client, 0); + GPR_ASSERT(state == GRPC_CHANNEL_READY || + state == GRPC_CHANNEL_CONNECTING || + state == GRPC_CHANNEL_TRANSIENT_FAILURE); + } + + for (i = 1; i <= 5; i++) { + grpc_channel_ping(client, f->cq, tag(i), NULL); + cq_expect_completion(cqv, tag(i), 1); + cq_verify(cqv); + } + gpr_free(rdata.call_details); + grpc_channel_destroy(client); teardown_servers(f); + + cq_verifier_destroy(cqv); +} + +static void test_pending_calls(size_t concurrent_calls) { + size_t i; + grpc_call **calls; + grpc_channel *client; + request_data rdata; + servers_fixture *f; + test_spec *spec = test_spec_create(0, 4); + rdata.call_details = + gpr_malloc(sizeof(grpc_call_details) * spec->num_servers); + f = setup_servers("127.0.0.1", &rdata, spec->num_servers); + + client = create_client(f); + calls = perform_multirequest(f, client, concurrent_calls); + grpc_call_cancel( + calls[0], + NULL); /* exercise the cancel pick path whilst there are pending picks */ + + gpr_free(rdata.call_details); + + grpc_channel_destroy(client); /* calls the LB's shutdown func */ + /* destroy the calls after the channel so that they are still around for the + * LB's shutdown func to process */ + for (i = 0; i < concurrent_calls; i++) { + grpc_call_destroy(calls[i]); + } + gpr_free(calls); + teardown_servers(f); + test_spec_destroy(spec); } static void print_failed_expectations(const int *expected_connection_sequence, @@ -715,13 +851,14 @@ int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_init(); + grpc_lb_round_robin_trace = 1; GPR_ASSERT(grpc_lb_policy_create("this-lb-policy-does-not-exist", NULL) == NULL); GPR_ASSERT(grpc_lb_policy_create(NULL, NULL) == NULL); - /* everything is fine, all servers stay up the whole time and life's peachy */ spec = test_spec_create(NUM_ITERS, NUM_SERVERS); + /* everything is fine, all servers stay up the whole time and life's peachy */ spec->verifier = verify_vanilla_round_robin; spec->description = "test_all_server_up"; run_spec(spec); @@ -735,7 +872,8 @@ int main(int argc, char **argv) { } run_spec(spec); - /* at the start of the 2nd iteration, kill all but the first and last servers. + /* at the start of the 2nd iteration, kill all but the first and last + * servers. * This should knock down the server bound to be selected next */ test_spec_reset(spec); spec->verifier = verify_vanishing_floor_round_robin; @@ -764,9 +902,11 @@ int main(int argc, char **argv) { spec->revive_at[3][i] = 1; } run_spec(spec); - test_spec_destroy(spec); + test_pending_calls(4); + test_ping(); + grpc_shutdown(); return 0; } diff --git a/test/core/httpcli/httpcli_test.c b/test/core/httpcli/httpcli_test.c index d47774251a..fc51cb0101 100644 --- a/test/core/httpcli/httpcli_test.c +++ b/test/core/httpcli/httpcli_test.c @@ -142,19 +142,26 @@ int main(int argc, char **argv) { char *me = argv[0]; char *lslash = strrchr(me, '/'); char *args[4]; - char root[1024]; int port = grpc_pick_unused_port_or_die(); - /* figure out where we are */ - if (lslash) { - memcpy(root, me, (size_t)(lslash - me)); - root[lslash - me] = 0; + GPR_ASSERT(argc <= 2); + if (argc == 2) { + args[0] = gpr_strdup(argv[1]); } else { - strcpy(root, "."); + /* figure out where we are */ + char *root; + if (lslash) { + root = gpr_malloc(lslash - me + 1); + memcpy(root, me, (size_t)(lslash - me)); + root[lslash - me] = 0; + } else { + root = strdup("."); + } + gpr_asprintf(&args[0], "%s/../../test/core/httpcli/test_server.py", root); + gpr_free(root); } /* start the server */ - gpr_asprintf(&args[0], "%s/../../test/core/httpcli/test_server.py", root); args[1] = "--port"; gpr_asprintf(&args[2], "%d", port); server = gpr_subprocess_create(3, (const char **)args); diff --git a/test/core/httpcli/httpscli_test.c b/test/core/httpcli/httpscli_test.c index b1c1913cae..4cfa9e59b0 100644 --- a/test/core/httpcli/httpscli_test.c +++ b/test/core/httpcli/httpscli_test.c @@ -144,19 +144,26 @@ int main(int argc, char **argv) { char *me = argv[0]; char *lslash = strrchr(me, '/'); char *args[5]; - char root[1024]; int port = grpc_pick_unused_port_or_die(); - /* figure out where we are */ - if (lslash) { - memcpy(root, me, (size_t)(lslash - me)); - root[lslash - me] = 0; + GPR_ASSERT(argc <= 2); + if (argc == 2) { + args[0] = gpr_strdup(argv[1]); } else { - strcpy(root, "."); + /* figure out where we are */ + char *root; + if (lslash) { + root = gpr_malloc(lslash - me + 1); + memcpy(root, me, (size_t)(lslash - me)); + root[lslash - me] = 0; + } else { + strcpy(root, "."); + } + gpr_asprintf(&args[0], "%s/../../test/core/httpcli/test_server.py", root); + gpr_free(root); } /* start the server */ - gpr_asprintf(&args[0], "%s/../../test/core/httpcli/test_server.py", root); args[1] = "--port"; gpr_asprintf(&args[2], "%d", port); args[3] = "--ssl"; diff --git a/tools/run_tests/run_lcov.sh b/tools/run_tests/run_lcov.sh index ec97ebf0a5..796a0b5ceb 100755 --- a/tools/run_tests/run_lcov.sh +++ b/tools/run_tests/run_lcov.sh @@ -33,7 +33,7 @@ set -ex out=$(readlink -f ${1:-coverage}) root=$(readlink -f $(dirname $0)/../..) -shift +shift || true tmp=$(mktemp) cd $root tools/run_tests/run_tests.py -c gcov -l c c++ $@ || true |