Merge pull request #12600 from sreecha/closure-debug
More detailed error when double-scheduling a closure
diff --git a/doc/environment_variables.md b/doc/environment_variables.md
index b79cd97..f90f1d5 100644
--- a/doc/environment_variables.md
+++ b/doc/environment_variables.md
@@ -50,6 +50,7 @@
- channel_stack_builder - traces information about channel stacks being built
- executor - traces grpc's internal thread pool ('the executor')
- http - traces state in the http2 transport engine
+ - http2_stream_state - traces all http2 stream state mutations.
- http1 - traces HTTP/1.x operations performed by gRPC
- inproc - traces the in-process transport
- flowctl - traces http2 flow control
diff --git a/include/grpc/compression.h b/include/grpc/compression.h
index 3a8de4b..13a8dd6 100644
--- a/include/grpc/compression.h
+++ b/include/grpc/compression.h
@@ -44,13 +44,13 @@
* algorithm. Note that \a name is statically allocated and must *not* be freed.
* Returns 1 upon success, 0 otherwise. */
GRPCAPI int grpc_compression_algorithm_name(
- grpc_compression_algorithm algorithm, char **name);
+ grpc_compression_algorithm algorithm, const char **name);
/** Updates \a name with the encoding name corresponding to a valid \a
* algorithm. Note that \a name is statically allocated and must *not* be freed.
* Returns 1 upon success, 0 otherwise. */
GRPCAPI int grpc_stream_compression_algorithm_name(
- grpc_stream_compression_algorithm algorithm, char **name);
+ grpc_stream_compression_algorithm algorithm, const char **name);
/** Returns the compression algorithm corresponding to \a level for the
* compression algorithms encoded in the \a accepted_encodings bitset.
diff --git a/src/core/ext/census/base_resources.c b/src/core/ext/census/base_resources.c
index 2114bf0..1f2bb39 100644
--- a/src/core/ext/census/base_resources.c
+++ b/src/core/ext/census/base_resources.c
@@ -37,20 +37,20 @@
void define_base_resources() {
google_census_Resource_BasicUnit numerator =
google_census_Resource_BasicUnit_SECS;
- resource r = {"client_rpc_latency", // name
- "Client RPC latency in seconds", // description
- 0, // prefix
- 1, // n_numerators
- &numerator, // numerators
- 0, // n_denominators
- NULL}; // denominators
+ resource r = {(char *)"client_rpc_latency", // name
+ (char *)"Client RPC latency in seconds", // description
+ 0, // prefix
+ 1, // n_numerators
+ &numerator, // numerators
+ 0, // n_denominators
+ NULL}; // denominators
define_resource(&r);
- r = (resource){"server_rpc_latency", // name
- "Server RPC latency in seconds", // description
- 0, // prefix
- 1, // n_numerators
- &numerator, // numerators
- 0, // n_denominators
- NULL}; // denominators
+ r = (resource){(char *)"server_rpc_latency", // name
+ (char *)"Server RPC latency in seconds", // description
+ 0, // prefix
+ 1, // n_numerators
+ &numerator, // numerators
+ 0, // n_denominators
+ NULL}; // denominators
define_resource(&r);
}
diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 129d0f3..016199b 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -375,7 +375,7 @@
}
// Extract the following fields from the resolver result, if non-NULL.
bool lb_policy_updated = false;
- char *lb_policy_name = NULL;
+ char *lb_policy_name_dup = NULL;
bool lb_policy_name_changed = false;
grpc_lb_policy *new_lb_policy = NULL;
char *service_config_json = NULL;
@@ -383,6 +383,7 @@
grpc_slice_hash_table *method_params_table = NULL;
if (chand->resolver_result != NULL) {
// Find LB policy name.
+ const char *lb_policy_name = NULL;
const grpc_arg *channel_arg =
grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_POLICY_NAME);
if (channel_arg != NULL) {
@@ -473,7 +474,7 @@
// Before we clean up, save a copy of lb_policy_name, since it might
// be pointing to data inside chand->resolver_result.
// The copy will be saved in chand->lb_policy_name below.
- lb_policy_name = gpr_strdup(lb_policy_name);
+ lb_policy_name_dup = gpr_strdup(lb_policy_name);
grpc_channel_args_destroy(exec_ctx, chand->resolver_result);
chand->resolver_result = NULL;
}
@@ -481,8 +482,8 @@
gpr_log(GPR_DEBUG,
"chand=%p: resolver result: lb_policy_name=\"%s\"%s, "
"service_config=\"%s\"",
- chand, lb_policy_name, lb_policy_name_changed ? " (changed)" : "",
- service_config_json);
+ chand, lb_policy_name_dup,
+ lb_policy_name_changed ? " (changed)" : "", service_config_json);
}
// Now swap out fields in chand. Note that the new values may still
// be NULL if (e.g.) the resolver failed to return results or the
@@ -490,9 +491,9 @@
//
// First, swap out the data used by cc_get_channel_info().
gpr_mu_lock(&chand->info_mu);
- if (lb_policy_name != NULL) {
+ if (lb_policy_name_dup != NULL) {
gpr_free(chand->info_lb_policy_name);
- chand->info_lb_policy_name = lb_policy_name;
+ chand->info_lb_policy_name = lb_policy_name_dup;
}
if (service_config_json != NULL) {
gpr_free(chand->info_service_config_json);
diff --git a/src/core/ext/filters/client_channel/client_channel_factory.c b/src/core/ext/filters/client_channel/client_channel_factory.c
index e8aa4cd..57eac8f 100644
--- a/src/core/ext/filters/client_channel/client_channel_factory.c
+++ b/src/core/ext/filters/client_channel/client_channel_factory.c
@@ -63,6 +63,6 @@
grpc_arg grpc_client_channel_factory_create_channel_arg(
grpc_client_channel_factory* factory) {
- return grpc_channel_arg_pointer_create(GRPC_ARG_CLIENT_CHANNEL_FACTORY,
+ return grpc_channel_arg_pointer_create((char*)GRPC_ARG_CLIENT_CHANNEL_FACTORY,
factory, &factory_arg_vtable);
}
diff --git a/src/core/ext/filters/client_channel/client_channel_plugin.c b/src/core/ext/filters/client_channel/client_channel_plugin.c
index c32e83d..1f71c5a 100644
--- a/src/core/ext/filters/client_channel/client_channel_plugin.c
+++ b/src/core/ext/filters/client_channel/client_channel_plugin.c
@@ -54,8 +54,8 @@
char *default_authority = grpc_get_default_authority(
exec_ctx, grpc_channel_stack_builder_get_target(builder));
if (default_authority != NULL) {
- grpc_arg arg = grpc_channel_arg_string_create(GRPC_ARG_DEFAULT_AUTHORITY,
- default_authority);
+ grpc_arg arg = grpc_channel_arg_string_create(
+ (char *)GRPC_ARG_DEFAULT_AUTHORITY, default_authority);
grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder,
new_args);
diff --git a/src/core/ext/filters/client_channel/http_proxy.c b/src/core/ext/filters/client_channel/http_proxy.c
index ef3512e..0c3b009 100644
--- a/src/core/ext/filters/client_channel/http_proxy.c
+++ b/src/core/ext/filters/client_channel/http_proxy.c
@@ -157,7 +157,7 @@
}
grpc_arg args_to_add[2];
args_to_add[0] = grpc_channel_arg_string_create(
- GRPC_ARG_HTTP_CONNECT_SERVER,
+ (char*)GRPC_ARG_HTTP_CONNECT_SERVER,
uri->path[0] == '/' ? uri->path + 1 : uri->path);
if (user_cred != NULL) {
/* Use base64 encoding for user credentials as stated in RFC 7617 */
@@ -166,8 +166,8 @@
char* header;
gpr_asprintf(&header, "Proxy-Authorization:Basic %s", encoded_user_cred);
gpr_free(encoded_user_cred);
- args_to_add[1] =
- grpc_channel_arg_string_create(GRPC_ARG_HTTP_CONNECT_HEADERS, header);
+ args_to_add[1] = grpc_channel_arg_string_create(
+ (char*)GRPC_ARG_HTTP_CONNECT_HEADERS, header);
*new_args = grpc_channel_args_copy_and_add(args, args_to_add, 2);
gpr_free(header);
} else {
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
index a776a07..c4c4973 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
@@ -1830,8 +1830,8 @@
// Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args,
// since we use this to trigger the client_load_reporting filter.
- grpc_arg new_arg =
- grpc_channel_arg_string_create(GRPC_ARG_LB_POLICY_NAME, "grpclb");
+ grpc_arg new_arg = grpc_channel_arg_string_create(
+ (char *)GRPC_ARG_LB_POLICY_NAME, (char *)"grpclb");
static const char *args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME};
glb_policy->args = grpc_channel_args_copy_and_add_and_remove(
args->args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1);
diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c
index 8ac1a46..a3a62e9 100644
--- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c
+++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c
@@ -589,7 +589,7 @@
// Dispose of outdated subchannel lists.
if (sd->subchannel_list != p->subchannel_list &&
sd->subchannel_list != p->latest_pending_subchannel_list) {
- char *reason = NULL;
+ const char *reason = NULL;
if (sd->subchannel_list->shutting_down) {
reason = "sl_outdated_straggler";
rr_subchannel_list_unref(exec_ctx, sd->subchannel_list, reason);
diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.c b/src/core/ext/filters/client_channel/lb_policy_factory.c
index acf5929..4d14054 100644
--- a/src/core/ext/filters/client_channel/lb_policy_factory.c
+++ b/src/core/ext/filters/client_channel/lb_policy_factory.c
@@ -141,7 +141,7 @@
grpc_arg grpc_lb_addresses_create_channel_arg(
const grpc_lb_addresses* addresses) {
return grpc_channel_arg_pointer_create(
- GRPC_ARG_LB_ADDRESSES, (void*)addresses, &lb_addresses_arg_vtable);
+ (char*)GRPC_ARG_LB_ADDRESSES, (void*)addresses, &lb_addresses_arg_vtable);
}
grpc_lb_addresses* grpc_lb_addresses_find_channel_arg(
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
index 371c59b..9bb229a 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
@@ -249,7 +249,7 @@
service_config_string);
args_to_remove[num_args_to_remove++] = GRPC_ARG_SERVICE_CONFIG;
new_args[num_args_to_add++] = grpc_channel_arg_string_create(
- GRPC_ARG_SERVICE_CONFIG, service_config_string);
+ (char *)GRPC_ARG_SERVICE_CONFIG, service_config_string);
service_config = grpc_service_config_create(service_config_string);
if (service_config != NULL) {
const char *lb_policy_name =
@@ -257,7 +257,7 @@
if (lb_policy_name != NULL) {
args_to_remove[num_args_to_remove++] = GRPC_ARG_LB_POLICY_NAME;
new_args[num_args_to_add++] = grpc_channel_arg_string_create(
- GRPC_ARG_LB_POLICY_NAME, (char *)lb_policy_name);
+ (char *)GRPC_ARG_LB_POLICY_NAME, (char *)lb_policy_name);
}
}
}
diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c
index c4676e0..69ea440 100644
--- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c
+++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c
@@ -210,7 +210,7 @@
grpc_fake_resolver_response_generator* generator) {
grpc_arg arg;
arg.type = GRPC_ARG_POINTER;
- arg.key = GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR;
+ arg.key = (char*)GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR;
arg.value.pointer.p = generator;
arg.value.pointer.vtable = &response_generator_arg_vtable;
return arg;
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index bc9c3cc..40a51c7 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -811,6 +811,6 @@
grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr) {
return grpc_channel_arg_string_create(
- GRPC_ARG_SUBCHANNEL_ADDRESS,
+ (char *)GRPC_ARG_SUBCHANNEL_ADDRESS,
addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup(""));
}
diff --git a/src/core/ext/filters/http/message_compress/message_compress_filter.c b/src/core/ext/filters/http/message_compress/message_compress_filter.c
index 0145dff..f785e13 100644
--- a/src/core/ext/filters/http/message_compress/message_compress_filter.c
+++ b/src/core/ext/filters/http/message_compress/message_compress_filter.c
@@ -244,7 +244,7 @@
&calld->slices, &tmp);
if (did_compress) {
if (GRPC_TRACER_ON(grpc_compression_trace)) {
- char *algo_name;
+ const char *algo_name;
const size_t before_size = calld->slices.length;
const size_t after_size = tmp.length;
const float savings_ratio = 1.0f - (float)after_size / (float)before_size;
@@ -258,7 +258,7 @@
send_flags |= GRPC_WRITE_INTERNAL_COMPRESS;
} else {
if (GRPC_TRACER_ON(grpc_compression_trace)) {
- char *algo_name;
+ const char *algo_name;
GPR_ASSERT(grpc_compression_algorithm_name(calld->compression_algorithm,
&algo_name));
gpr_log(GPR_DEBUG,
diff --git a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c
index f56afd2..2486ead 100644
--- a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c
+++ b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c
@@ -55,7 +55,8 @@
}
grpc_arg grpc_load_reporting_enable_arg() {
- return grpc_channel_arg_integer_create(GRPC_ARG_ENABLE_LOAD_REPORTING, 1);
+ return grpc_channel_arg_integer_create((char *)GRPC_ARG_ENABLE_LOAD_REPORTING,
+ 1);
}
/* Plugin registration */
diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create.c b/src/core/ext/transport/chttp2/client/insecure/channel_create.c
index cccb347..6410a60 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create.c
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create.c
@@ -55,7 +55,7 @@
}
// Add channel arg containing the server URI.
grpc_arg arg = grpc_channel_arg_string_create(
- GRPC_ARG_SERVER_URI,
+ (char *)GRPC_ARG_SERVER_URI,
grpc_resolver_factory_add_default_prefix_if_needed(exec_ctx, target));
const char *to_remove[] = {GRPC_ARG_SERVER_URI};
grpc_channel_args *new_args =
diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.c b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.c
index 0346d50..dd88136 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.c
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.c
@@ -42,7 +42,7 @@
(target, fd, args));
grpc_arg default_authority_arg = grpc_channel_arg_string_create(
- GRPC_ARG_DEFAULT_AUTHORITY, "test.authority");
+ (char *)GRPC_ARG_DEFAULT_AUTHORITY, (char *)"test.authority");
grpc_channel_args *final_args =
grpc_channel_args_copy_and_add(args, &default_authority_arg, 1);
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_plugin.c b/src/core/ext/transport/chttp2/transport/chttp2_plugin.c
index 78551df..6d09953 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_plugin.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_plugin.c
@@ -23,6 +23,7 @@
void grpc_chttp2_plugin_init(void) {
grpc_register_tracer(&grpc_http_trace);
grpc_register_tracer(&grpc_flowctl_trace);
+ grpc_register_tracer(&grpc_trace_http2_stream_state);
#ifndef NDEBUG
grpc_register_tracer(&grpc_trace_chttp2_refcount);
#endif
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.h b/src/core/ext/transport/chttp2/transport/chttp2_transport.h
index 0c4e2a9..55fb1a8 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.h
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.h
@@ -25,6 +25,7 @@
extern grpc_tracer_flag grpc_http_trace;
extern grpc_tracer_flag grpc_flowctl_trace;
+extern grpc_tracer_flag grpc_trace_http2_stream_state;
#ifndef NDEBUG
extern grpc_tracer_flag grpc_trace_chttp2_refcount;
diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c
index 7cc85de..47cd22d 100644
--- a/src/core/ext/transport/chttp2/transport/stream_lists.c
+++ b/src/core/ext/transport/chttp2/transport/stream_lists.c
@@ -20,6 +20,27 @@
#include <grpc/support/log.h>
+static char *stream_list_id_string(grpc_chttp2_stream_list_id id) {
+ switch (id) {
+ case GRPC_CHTTP2_LIST_WRITABLE:
+ return "writable";
+ case GRPC_CHTTP2_LIST_WRITING:
+ return "writing";
+ case GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT:
+ return "stalled_by_transport";
+ case GRPC_CHTTP2_LIST_STALLED_BY_STREAM:
+ return "stalled_by_stream";
+ case GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY:
+ return "waiting_for_concurrency";
+ case STREAM_LIST_COUNT:
+ GPR_UNREACHABLE_CODE(return "unknown");
+ }
+ GPR_UNREACHABLE_CODE(return "unknown");
+}
+
+grpc_tracer_flag grpc_trace_http2_stream_state =
+ GRPC_TRACER_INITIALIZER(false, "http2_stream_state");
+
/* core list management */
static bool stream_list_empty(grpc_chttp2_transport *t,
@@ -44,6 +65,10 @@
s->included[id] = 0;
}
*stream = s;
+ if (s && GRPC_TRACER_ON(grpc_trace_http2_stream_state)) {
+ gpr_log(GPR_DEBUG, "%p[%d][%s]: pop from %s", t, s->id,
+ t->is_client ? "cli" : "svr", stream_list_id_string(id));
+ }
return s != 0;
}
@@ -62,6 +87,10 @@
} else {
t->lists[id].tail = s->links[id].prev;
}
+ if (GRPC_TRACER_ON(grpc_trace_http2_stream_state)) {
+ gpr_log(GPR_DEBUG, "%p[%d][%s]: remove from %s", t, s->id,
+ t->is_client ? "cli" : "svr", stream_list_id_string(id));
+ }
}
static bool stream_list_maybe_remove(grpc_chttp2_transport *t,
@@ -90,6 +119,10 @@
}
t->lists[id].tail = s;
s->included[id] = 1;
+ if (GRPC_TRACER_ON(grpc_trace_http2_stream_state)) {
+ gpr_log(GPR_DEBUG, "%p[%d][%s]: add to %s", t, s->id,
+ t->is_client ? "cli" : "svr", stream_list_id_string(id));
+ }
}
static bool stream_list_add(grpc_chttp2_transport *t, grpc_chttp2_stream *s,
@@ -150,17 +183,12 @@
void grpc_chttp2_list_add_stalled_by_transport(grpc_chttp2_transport *t,
grpc_chttp2_stream *s) {
- GRPC_FLOW_CONTROL_IF_TRACING(
- gpr_log(GPR_DEBUG, "stream %u stalled by transport", s->id));
stream_list_add(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT);
}
bool grpc_chttp2_list_pop_stalled_by_transport(grpc_chttp2_transport *t,
grpc_chttp2_stream **s) {
- bool ret = stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT);
- GRPC_FLOW_CONTROL_IF_TRACING(if (ret) gpr_log(
- GPR_DEBUG, "stream %u un-stalled by transport", (*s)->id));
- return ret;
+ return stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT);
}
void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport *t,
@@ -170,23 +198,15 @@
void grpc_chttp2_list_add_stalled_by_stream(grpc_chttp2_transport *t,
grpc_chttp2_stream *s) {
- GRPC_FLOW_CONTROL_IF_TRACING(
- gpr_log(GPR_DEBUG, "stream %u stalled by stream", s->id));
stream_list_add(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
}
bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport *t,
grpc_chttp2_stream **s) {
- bool ret = stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
- GRPC_FLOW_CONTROL_IF_TRACING(
- if (ret) gpr_log(GPR_DEBUG, "stream %u un-stalled by stream", (*s)->id));
- return ret;
+ return stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
}
bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport *t,
grpc_chttp2_stream *s) {
- bool ret = stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
- GRPC_FLOW_CONTROL_IF_TRACING(
- if (ret) gpr_log(GPR_DEBUG, "stream %u un-stalled by stream", s->id));
- return ret;
+ return stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
}
diff --git a/src/core/ext/transport/inproc/inproc_transport.c b/src/core/ext/transport/inproc/inproc_transport.c
index cd3e76a..31739d0 100644
--- a/src/core/ext/transport/inproc/inproc_transport.c
+++ b/src/core/ext/transport/inproc/inproc_transport.c
@@ -1263,8 +1263,8 @@
grpc_arg default_authority_arg;
default_authority_arg.type = GRPC_ARG_STRING;
- default_authority_arg.key = GRPC_ARG_DEFAULT_AUTHORITY;
- default_authority_arg.value.string = "inproc.authority";
+ default_authority_arg.key = (char *)GRPC_ARG_DEFAULT_AUTHORITY;
+ default_authority_arg.value.string = (char *)"inproc.authority";
grpc_channel_args *client_args =
grpc_channel_args_copy_and_add(args, &default_authority_arg, 1);
diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c
index 9215707..30248b3 100644
--- a/src/core/lib/channel/channel_args.c
+++ b/src/core/lib/channel/channel_args.c
@@ -243,7 +243,7 @@
GPR_ASSERT(algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT);
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
- tmp.key = GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
+ tmp.key = (char *)GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
tmp.value.integer = algorithm;
return grpc_channel_args_copy_and_add(a, &tmp, 1);
}
@@ -253,7 +253,7 @@
GPR_ASSERT(algorithm < GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT);
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
- tmp.key = GRPC_STREAM_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
+ tmp.key = (char *)GRPC_STREAM_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
tmp.value.integer = algorithm;
return grpc_channel_args_copy_and_add(a, &tmp, 1);
}
@@ -308,7 +308,7 @@
if (grpc_channel_args_get_compression_algorithm(*a) == algorithm &&
state == 0) {
- char *algo_name = NULL;
+ const char *algo_name = NULL;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name) != 0);
gpr_log(GPR_ERROR,
"Tried to disable default compression algorithm '%s'. The "
@@ -324,7 +324,7 @@
/* create a new arg */
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
- tmp.key = GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET;
+ tmp.key = (char *)GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET;
/* all enabled by default */
tmp.value.integer = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1;
if (state != 0) {
@@ -349,7 +349,7 @@
if (grpc_channel_args_get_stream_compression_algorithm(*a) == algorithm &&
state == 0) {
- char *algo_name = NULL;
+ const char *algo_name = NULL;
GPR_ASSERT(grpc_stream_compression_algorithm_name(algorithm, &algo_name) !=
0);
gpr_log(GPR_ERROR,
@@ -366,7 +366,7 @@
/* create a new arg */
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
- tmp.key = GRPC_STREAM_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET;
+ tmp.key = (char *)GRPC_STREAM_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET;
/* all enabled by default */
tmp.value.integer = (1u << GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) - 1;
if (state != 0) {
diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h
index ae1cac3..f0de80f 100644
--- a/src/core/lib/channel/channel_stack.h
+++ b/src/core/lib/channel/channel_stack.h
@@ -281,7 +281,7 @@
/* Given the top element of a call stack, get the call stack itself */
grpc_call_stack *grpc_call_stack_from_top_element(grpc_call_element *elem);
-void grpc_call_log_op(char *file, int line, gpr_log_severity severity,
+void grpc_call_log_op(const char *file, int line, gpr_log_severity severity,
grpc_call_element *elem,
grpc_transport_stream_op_batch *op);
diff --git a/src/core/lib/compression/compression.c b/src/core/lib/compression/compression.c
index ec84c01..1cfac23 100644
--- a/src/core/lib/compression/compression.c
+++ b/src/core/lib/compression/compression.c
@@ -60,7 +60,7 @@
}
int grpc_compression_algorithm_name(grpc_compression_algorithm algorithm,
- char **name) {
+ const char **name) {
GRPC_API_TRACE("grpc_compression_algorithm_parse(algorithm=%d, name=%p)", 2,
((int)algorithm, name));
switch (algorithm) {
@@ -80,7 +80,7 @@
}
int grpc_stream_compression_algorithm_name(
- grpc_stream_compression_algorithm algorithm, char **name) {
+ grpc_stream_compression_algorithm algorithm, const char **name) {
GRPC_API_TRACE(
"grpc_stream_compression_algorithm_parse(algorithm=%d, name=%p)", 2,
((int)algorithm, name));
diff --git a/src/core/lib/debug/trace.h b/src/core/lib/debug/trace.h
index dd9e6a3..64f2e3f 100644
--- a/src/core/lib/debug/trace.h
+++ b/src/core/lib/debug/trace.h
@@ -35,7 +35,7 @@
#else
bool value;
#endif
- char *name;
+ const char *name;
} grpc_tracer_flag;
#ifdef GRPC_THREADSAFE_TRACER
diff --git a/src/core/lib/http/httpcli_security_connector.c b/src/core/lib/http/httpcli_security_connector.c
index 97c2886..c553fa3 100644
--- a/src/core/lib/http/httpcli_security_connector.c
+++ b/src/core/lib/http/httpcli_security_connector.c
@@ -43,7 +43,8 @@
grpc_httpcli_ssl_channel_security_connector *c =
(grpc_httpcli_ssl_channel_security_connector *)sc;
if (c->handshaker_factory != NULL) {
- tsi_ssl_client_handshaker_factory_destroy(c->handshaker_factory);
+ tsi_ssl_client_handshaker_factory_unref(c->handshaker_factory);
+ c->handshaker_factory = NULL;
}
if (c->secure_peer_name != NULL) gpr_free(c->secure_peer_name);
gpr_free(sc);
diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c
index d5e00a8..aa05501 100644
--- a/src/core/lib/iomgr/error.c
+++ b/src/core/lib/iomgr/error.c
@@ -641,7 +641,7 @@
static char *fmt_time(gpr_timespec tm) {
char *out;
- char *pfx = "!!";
+ const char *pfx = "!!";
switch (tm.clock_type) {
case GPR_CLOCK_MONOTONIC:
pfx = "@monotonic:";
diff --git a/src/core/lib/iomgr/resolve_address_posix.c b/src/core/lib/iomgr/resolve_address_posix.c
index 082e3b7..60cfeeb 100644
--- a/src/core/lib/iomgr/resolve_address_posix.c
+++ b/src/core/lib/iomgr/resolve_address_posix.c
@@ -85,7 +85,7 @@
if (s != 0) {
/* Retry if well-known service name is recognized */
- char *svc[][2] = {{"http", "80"}, {"https", "443"}};
+ const char *svc[][2] = {{"http", "80"}, {"https", "443"}};
for (i = 0; i < GPR_ARRAY_SIZE(svc); i++) {
if (strcmp(port, svc[i][0]) == 0) {
GRPC_SCHEDULING_START_BLOCKING_REGION;
diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c
index a7568b9..2a9e939 100644
--- a/src/core/lib/security/transport/security_connector.c
+++ b/src/core/lib/security/transport/security_connector.c
@@ -455,14 +455,14 @@
typedef struct {
grpc_channel_security_connector base;
- tsi_ssl_client_handshaker_factory *handshaker_factory;
+ tsi_ssl_client_handshaker_factory *client_handshaker_factory;
char *target_name;
char *overridden_target_name;
} grpc_ssl_channel_security_connector;
typedef struct {
grpc_server_security_connector base;
- tsi_ssl_server_handshaker_factory *handshaker_factory;
+ tsi_ssl_server_handshaker_factory *server_handshaker_factory;
} grpc_ssl_server_security_connector;
static void ssl_channel_destroy(grpc_exec_ctx *exec_ctx,
@@ -470,9 +470,8 @@
grpc_ssl_channel_security_connector *c =
(grpc_ssl_channel_security_connector *)sc;
grpc_call_credentials_unref(exec_ctx, c->base.request_metadata_creds);
- if (c->handshaker_factory != NULL) {
- tsi_ssl_client_handshaker_factory_destroy(c->handshaker_factory);
- }
+ tsi_ssl_client_handshaker_factory_unref(c->client_handshaker_factory);
+ c->client_handshaker_factory = NULL;
if (c->target_name != NULL) gpr_free(c->target_name);
if (c->overridden_target_name != NULL) gpr_free(c->overridden_target_name);
gpr_free(sc);
@@ -482,9 +481,8 @@
grpc_security_connector *sc) {
grpc_ssl_server_security_connector *c =
(grpc_ssl_server_security_connector *)sc;
- if (c->handshaker_factory != NULL) {
- tsi_ssl_server_handshaker_factory_destroy(c->handshaker_factory);
- }
+ tsi_ssl_server_handshaker_factory_unref(c->server_handshaker_factory);
+ c->server_handshaker_factory = NULL;
gpr_free(sc);
}
@@ -496,7 +494,7 @@
// Instantiate TSI handshaker.
tsi_handshaker *tsi_hs = NULL;
tsi_result result = tsi_ssl_client_handshaker_factory_create_handshaker(
- c->handshaker_factory,
+ c->client_handshaker_factory,
c->overridden_target_name != NULL ? c->overridden_target_name
: c->target_name,
&tsi_hs);
@@ -521,7 +519,7 @@
// Instantiate TSI handshaker.
tsi_handshaker *tsi_hs = NULL;
tsi_result result = tsi_ssl_server_handshaker_factory_create_handshaker(
- c->handshaker_factory, &tsi_hs);
+ c->server_handshaker_factory, &tsi_hs);
if (result != TSI_OK) {
gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.",
tsi_result_to_string(result));
@@ -852,7 +850,7 @@
result = tsi_create_ssl_client_handshaker_factory(
has_key_cert_pair ? &config->pem_key_cert_pair : NULL, pem_root_certs,
ssl_cipher_suites(), alpn_protocol_strings, (uint16_t)num_alpn_protocols,
- &c->handshaker_factory);
+ &c->client_handshaker_factory);
if (result != TSI_OK) {
gpr_log(GPR_ERROR, "Handshaker factory creation failed with %s.",
tsi_result_to_string(result));
@@ -897,7 +895,7 @@
config->pem_root_certs, get_tsi_client_certificate_request_type(
config->client_certificate_request),
ssl_cipher_suites(), alpn_protocol_strings, (uint16_t)num_alpn_protocols,
- &c->handshaker_factory);
+ &c->server_handshaker_factory);
if (result != TSI_OK) {
gpr_log(GPR_ERROR, "Handshaker factory creation failed with %s.",
tsi_result_to_string(result));
diff --git a/src/core/lib/support/log_linux.c b/src/core/lib/support/log_linux.c
index 61d2346..7755018 100644
--- a/src/core/lib/support/log_linux.c
+++ b/src/core/lib/support/log_linux.c
@@ -57,7 +57,7 @@
}
void gpr_default_log(gpr_log_func_args *args) {
- char *final_slash;
+ const char *final_slash;
char *prefix;
const char *display_file;
char time_buffer[64];
diff --git a/src/core/lib/support/string.c b/src/core/lib/support/string.c
index 523e434..6b172df 100644
--- a/src/core/lib/support/string.c
+++ b/src/core/lib/support/string.c
@@ -276,7 +276,7 @@
void gpr_string_split(const char *input, const char *sep, char ***strs,
size_t *nstrs) {
- char *next;
+ const char *next;
*strs = NULL;
*nstrs = 0;
size_t capstrs = 0;
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index e258719..a41eb33 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -1511,7 +1511,7 @@
} else if (grpc_compression_options_is_stream_compression_algorithm_enabled(
&compression_options, algo) == 0) {
/* check if algorithm is supported by current channel config */
- char *algo_name = NULL;
+ const char *algo_name = NULL;
grpc_stream_compression_algorithm_name(algo, &algo_name);
gpr_asprintf(&error_msg, "Stream compression algorithm '%s' is disabled.",
algo_name);
@@ -1525,7 +1525,7 @@
if (!GPR_BITGET(call->stream_encodings_accepted_by_peer,
call->incoming_stream_compression_algorithm)) {
if (GRPC_TRACER_ON(grpc_compression_trace)) {
- char *algo_name = NULL;
+ const char *algo_name = NULL;
grpc_stream_compression_algorithm_name(
call->incoming_stream_compression_algorithm, &algo_name);
gpr_log(
@@ -1552,7 +1552,7 @@
} else if (grpc_compression_options_is_algorithm_enabled(
&compression_options, algo) == 0) {
/* check if algorithm is supported by current channel config */
- char *algo_name = NULL;
+ const char *algo_name = NULL;
grpc_compression_algorithm_name(algo, &algo_name);
gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.",
algo_name);
@@ -1568,7 +1568,7 @@
if (!GPR_BITGET(call->encodings_accepted_by_peer,
call->incoming_compression_algorithm)) {
if (GRPC_TRACER_ON(grpc_compression_trace)) {
- char *algo_name = NULL;
+ const char *algo_name = NULL;
grpc_compression_algorithm_name(call->incoming_compression_algorithm,
&algo_name);
gpr_log(GPR_ERROR,
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index 6452f08..468360f 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -565,13 +565,13 @@
* true if the increment was successful; false if the counter is zero */
static bool atm_inc_if_nonzero(gpr_atm *counter) {
while (true) {
- gpr_atm count = gpr_atm_no_barrier_load(counter);
+ gpr_atm count = gpr_atm_acq_load(counter);
/* If zero, we are done. If not, we must to a CAS (instead of an atomic
* increment) to maintain the contract: do not increment the counter if it
* is zero. */
if (count == 0) {
return false;
- } else if (gpr_atm_no_barrier_cas(counter, count, count + 1)) {
+ } else if (gpr_atm_full_cas(counter, count, count + 1)) {
break;
}
}
@@ -643,8 +643,12 @@
/* Add the completion to the queue */
bool is_first = cq_event_queue_push(&cqd->queue, storage);
gpr_atm_no_barrier_fetch_add(&cqd->things_queued_ever, 1);
- bool will_definitely_shutdown =
- gpr_atm_no_barrier_load(&cqd->pending_events) == 1;
+
+ /* Since we do not hold the cq lock here, it is important to do an 'acquire'
+ load here (instead of a 'no_barrier' load) to match with the release store
+ (done via gpr_atm_full_fetch_add(pending_events, -1)) in cq_shutdown_next
+ */
+ bool will_definitely_shutdown = gpr_atm_acq_load(&cqd->pending_events) == 1;
if (!will_definitely_shutdown) {
/* Only kick if this is the first item queued */
@@ -888,7 +892,7 @@
}
}
- if (gpr_atm_no_barrier_load(&cqd->pending_events) == 0) {
+ if (gpr_atm_acq_load(&cqd->pending_events) == 0) {
/* Before returning, check if the queue has any items left over (since
gpr_mpscq_pop() can sometimes return NULL even if the queue is not
empty. If so, keep retrying but do not return GRPC_QUEUE_SHUTDOWN */
@@ -934,7 +938,7 @@
}
if (cq_event_queue_num_items(&cqd->queue) > 0 &&
- gpr_atm_no_barrier_load(&cqd->pending_events) > 0) {
+ gpr_atm_acq_load(&cqd->pending_events) > 0) {
gpr_mu_lock(cq->mu);
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), NULL);
gpr_mu_unlock(cq->mu);
@@ -985,6 +989,9 @@
return;
}
cqd->shutdown_called = true;
+ /* Doing a full_fetch_add (i.e acq/release) here to match with
+ * cq_begin_op_for_next and and cq_end_op_for_next functions which read/write
+ * on this counter without necessarily holding a lock on cq */
if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) {
cq_finish_shutdown_next(exec_ctx, cq);
}
diff --git a/src/core/lib/transport/transport_op_string.c b/src/core/lib/transport/transport_op_string.c
index 409a6c4..8586647 100644
--- a/src/core/lib/transport/transport_op_string.c
+++ b/src/core/lib/transport/transport_op_string.c
@@ -197,7 +197,7 @@
return out;
}
-void grpc_call_log_op(char *file, int line, gpr_log_severity severity,
+void grpc_call_log_op(const char *file, int line, gpr_log_severity severity,
grpc_call_element *elem,
grpc_transport_stream_op_batch *op) {
char *str = grpc_transport_stream_op_batch_string(op);
diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c
index 1fd6592..7ebf9dd 100644
--- a/src/core/tsi/ssl_transport_security.c
+++ b/src/core/tsi/ssl_transport_security.c
@@ -67,7 +67,13 @@
/* --- Structure definitions. ---*/
+struct tsi_ssl_handshaker_factory {
+ const tsi_ssl_handshaker_factory_vtable *vtable;
+ gpr_refcount refcount;
+};
+
struct tsi_ssl_client_handshaker_factory {
+ tsi_ssl_handshaker_factory base;
SSL_CTX *ssl_context;
unsigned char *alpn_protocol_list;
size_t alpn_protocol_list_length;
@@ -77,6 +83,7 @@
/* Several contexts to support SNI.
The tsi_peer array contains the subject names of the server certificates
associated with the contexts at the same index. */
+ tsi_ssl_handshaker_factory base;
SSL_CTX **ssl_contexts;
tsi_peer *ssl_context_x509_subject_names;
size_t ssl_context_count;
@@ -90,6 +97,7 @@
BIO *into_ssl;
BIO *from_ssl;
tsi_result result;
+ tsi_ssl_handshaker_factory *factory_ref;
} tsi_ssl_handshaker;
typedef struct {
@@ -846,6 +854,47 @@
ssl_protector_destroy,
};
+/* --- tsi_server_handshaker_factory methods implementation. --- */
+
+static void tsi_ssl_handshaker_factory_destroy(
+ tsi_ssl_handshaker_factory *self) {
+ if (self == NULL) return;
+
+ if (self->vtable != NULL && self->vtable->destroy != NULL) {
+ self->vtable->destroy(self);
+ }
+ /* Note, we don't free(self) here because this object is always directly
+ * embedded in another object. If tsi_ssl_handshaker_factory_init allocates
+ * any memory, it should be free'd here. */
+}
+
+static tsi_ssl_handshaker_factory *tsi_ssl_handshaker_factory_ref(
+ tsi_ssl_handshaker_factory *self) {
+ if (self == NULL) return NULL;
+ gpr_refn(&self->refcount, 1);
+ return self;
+}
+
+static void tsi_ssl_handshaker_factory_unref(tsi_ssl_handshaker_factory *self) {
+ if (self == NULL) return;
+
+ if (gpr_unref(&self->refcount)) {
+ tsi_ssl_handshaker_factory_destroy(self);
+ }
+}
+
+static tsi_ssl_handshaker_factory_vtable handshaker_factory_vtable = {NULL};
+
+/* Initializes a tsi_ssl_handshaker_factory object. Caller is responsible for
+ * allocating memory for the factory. */
+static void tsi_ssl_handshaker_factory_init(
+ tsi_ssl_handshaker_factory *factory) {
+ GPR_ASSERT(factory != NULL);
+
+ factory->vtable = &handshaker_factory_vtable;
+ gpr_ref_init(&factory->refcount, 1);
+}
+
/* --- tsi_handshaker methods implementation. ---*/
static tsi_result ssl_handshaker_get_bytes_to_send_to_peer(tsi_handshaker *self,
@@ -1013,6 +1062,7 @@
static void ssl_handshaker_destroy(tsi_handshaker *self) {
tsi_ssl_handshaker *impl = (tsi_ssl_handshaker *)self;
SSL_free(impl->ssl); /* The BIO objects are owned by ssl */
+ tsi_ssl_handshaker_factory_unref(impl->factory_ref);
gpr_free(impl);
}
@@ -1030,6 +1080,7 @@
static tsi_result create_tsi_ssl_handshaker(SSL_CTX *ctx, int is_client,
const char *server_name_indication,
+ tsi_ssl_handshaker_factory *factory,
tsi_handshaker **handshaker) {
SSL *ssl = SSL_new(ctx);
BIO *into_ssl = NULL;
@@ -1085,6 +1136,8 @@
impl->from_ssl = from_ssl;
impl->result = TSI_HANDSHAKE_IN_PROGRESS;
impl->base.vtable = &handshaker_vtable;
+ impl->factory_ref = tsi_ssl_handshaker_factory_ref(factory);
+
*handshaker = &impl->base;
return TSI_OK;
}
@@ -1121,11 +1174,20 @@
tsi_ssl_client_handshaker_factory *self, const char *server_name_indication,
tsi_handshaker **handshaker) {
return create_tsi_ssl_handshaker(self->ssl_context, 1, server_name_indication,
- handshaker);
+ &self->base, handshaker);
}
-void tsi_ssl_client_handshaker_factory_destroy(
+void tsi_ssl_client_handshaker_factory_unref(
tsi_ssl_client_handshaker_factory *self) {
+ if (self == NULL) return;
+ tsi_ssl_handshaker_factory_unref(&self->base);
+}
+
+static void tsi_ssl_client_handshaker_factory_destroy(
+ tsi_ssl_handshaker_factory *factory) {
+ if (factory == NULL) return;
+ tsi_ssl_client_handshaker_factory *self =
+ (tsi_ssl_client_handshaker_factory *)factory;
if (self->ssl_context != NULL) SSL_CTX_free(self->ssl_context);
if (self->alpn_protocol_list != NULL) gpr_free(self->alpn_protocol_list);
gpr_free(self);
@@ -1150,11 +1212,21 @@
if (self->ssl_context_count == 0) return TSI_INVALID_ARGUMENT;
/* Create the handshaker with the first context. We will switch if needed
because of SNI in ssl_server_handshaker_factory_servername_callback. */
- return create_tsi_ssl_handshaker(self->ssl_contexts[0], 0, NULL, handshaker);
+ return create_tsi_ssl_handshaker(self->ssl_contexts[0], 0, NULL, &self->base,
+ handshaker);
}
-void tsi_ssl_server_handshaker_factory_destroy(
+void tsi_ssl_server_handshaker_factory_unref(
tsi_ssl_server_handshaker_factory *self) {
+ if (self == NULL) return;
+ tsi_ssl_handshaker_factory_unref(&self->base);
+}
+
+static void tsi_ssl_server_handshaker_factory_destroy(
+ tsi_ssl_handshaker_factory *factory) {
+ if (factory == NULL) return;
+ tsi_ssl_server_handshaker_factory *self =
+ (tsi_ssl_server_handshaker_factory *)factory;
size_t i;
for (i = 0; i < self->ssl_context_count; i++) {
if (self->ssl_contexts[i] != NULL) {
@@ -1263,6 +1335,9 @@
/* --- tsi_ssl_handshaker_factory constructors. --- */
+static tsi_ssl_handshaker_factory_vtable client_handshaker_factory_vtable = {
+ tsi_ssl_client_handshaker_factory_destroy};
+
tsi_result tsi_create_ssl_client_handshaker_factory(
const tsi_ssl_pem_key_cert_pair *pem_key_cert_pair,
const char *pem_root_certs, const char *cipher_suites,
@@ -1285,6 +1360,9 @@
}
impl = gpr_zalloc(sizeof(*impl));
+ tsi_ssl_handshaker_factory_init(&impl->base);
+ impl->base.vtable = &client_handshaker_factory_vtable;
+
impl->ssl_context = ssl_context;
do {
@@ -1322,7 +1400,7 @@
}
} while (0);
if (result != TSI_OK) {
- tsi_ssl_client_handshaker_factory_destroy(impl);
+ tsi_ssl_handshaker_factory_unref(&impl->base);
return result;
}
SSL_CTX_set_verify(ssl_context, SSL_VERIFY_PEER, NULL);
@@ -1332,6 +1410,9 @@
return TSI_OK;
}
+static tsi_ssl_handshaker_factory_vtable server_handshaker_factory_vtable = {
+ tsi_ssl_server_handshaker_factory_destroy};
+
tsi_result tsi_create_ssl_server_handshaker_factory(
const tsi_ssl_pem_key_cert_pair *pem_key_cert_pairs,
size_t num_key_cert_pairs, const char *pem_client_root_certs,
@@ -1364,12 +1445,15 @@
}
impl = gpr_zalloc(sizeof(*impl));
+ tsi_ssl_handshaker_factory_init(&impl->base);
+ impl->base.vtable = &server_handshaker_factory_vtable;
+
impl->ssl_contexts = gpr_zalloc(num_key_cert_pairs * sizeof(SSL_CTX *));
impl->ssl_context_x509_subject_names =
gpr_zalloc(num_key_cert_pairs * sizeof(tsi_peer));
if (impl->ssl_contexts == NULL ||
impl->ssl_context_x509_subject_names == NULL) {
- tsi_ssl_server_handshaker_factory_destroy(impl);
+ tsi_ssl_handshaker_factory_unref(&impl->base);
return TSI_OUT_OF_RESOURCES;
}
impl->ssl_context_count = num_key_cert_pairs;
@@ -1379,7 +1463,7 @@
&impl->alpn_protocol_list,
&impl->alpn_protocol_list_length);
if (result != TSI_OK) {
- tsi_ssl_server_handshaker_factory_destroy(impl);
+ tsi_ssl_handshaker_factory_unref(&impl->base);
return result;
}
}
@@ -1451,10 +1535,11 @@
} while (0);
if (result != TSI_OK) {
- tsi_ssl_server_handshaker_factory_destroy(impl);
+ tsi_ssl_handshaker_factory_unref(&impl->base);
return result;
}
}
+
*factory = impl;
return TSI_OK;
}
@@ -1501,3 +1586,15 @@
return 0; /* Not found. */
}
+
+/* --- Testing support. --- */
+const tsi_ssl_handshaker_factory_vtable *tsi_ssl_handshaker_factory_swap_vtable(
+ tsi_ssl_handshaker_factory *factory,
+ tsi_ssl_handshaker_factory_vtable *new_vtable) {
+ GPR_ASSERT(factory != NULL);
+ GPR_ASSERT(factory->vtable != NULL);
+
+ const tsi_ssl_handshaker_factory_vtable *orig_vtable = factory->vtable;
+ factory->vtable = new_vtable;
+ return orig_vtable;
+}
diff --git a/src/core/tsi/ssl_transport_security.h b/src/core/tsi/ssl_transport_security.h
index 1775999..3abfdf5 100644
--- a/src/core/tsi/ssl_transport_security.h
+++ b/src/core/tsi/ssl_transport_security.h
@@ -96,10 +96,10 @@
tsi_ssl_client_handshaker_factory *self, const char *server_name_indication,
tsi_handshaker **handshaker);
-/* Destroys the handshaker factory. WARNING: it is unsafe to destroy a factory
- while handshakers created with this factory are still in use. */
-void tsi_ssl_client_handshaker_factory_destroy(
- tsi_ssl_client_handshaker_factory *self);
+/* Decrements reference count of the handshaker factory. Handshaker factory will
+ * be destroyed once no references exist. */
+void tsi_ssl_client_handshaker_factory_unref(
+ tsi_ssl_client_handshaker_factory *factory);
/* --- tsi_ssl_server_handshaker_factory object ---
@@ -158,9 +158,9 @@
tsi_result tsi_ssl_server_handshaker_factory_create_handshaker(
tsi_ssl_server_handshaker_factory *self, tsi_handshaker **handshaker);
-/* Destroys the handshaker factory. WARNING: it is unsafe to destroy a factory
- while handshakers created with this factory are still in use. */
-void tsi_ssl_server_handshaker_factory_destroy(
+/* Decrements reference count of the handshaker factory. Handshaker factory will
+ * be destroyed once no references exist. */
+void tsi_ssl_server_handshaker_factory_unref(
tsi_ssl_server_handshaker_factory *self);
/* Util that checks that an ssl peer matches a specific name.
@@ -170,6 +170,29 @@
- handle public suffix wildchar more strictly (e.g. *.co.uk) */
int tsi_ssl_peer_matches_name(const tsi_peer *peer, const char *name);
+/* --- Testing support. ---
+
+ These functions and typedefs are not intended to be used outside of testing.
+ */
+
+/* Base type of client and server handshaker factories. */
+typedef struct tsi_ssl_handshaker_factory tsi_ssl_handshaker_factory;
+
+/* Function pointer to handshaker_factory destructor. */
+typedef void (*tsi_ssl_handshaker_factory_destructor)(
+ tsi_ssl_handshaker_factory *factory);
+
+/* Virtual table for tsi_ssl_handshaker_factory. */
+typedef struct {
+ tsi_ssl_handshaker_factory_destructor destroy;
+} tsi_ssl_handshaker_factory_vtable;
+
+/* Set destructor of handshaker_factory to new_destructor, returns previous
+ destructor. */
+const tsi_ssl_handshaker_factory_vtable *tsi_ssl_handshaker_factory_swap_vtable(
+ tsi_ssl_handshaker_factory *factory,
+ tsi_ssl_handshaker_factory_vtable *new_vtable);
+
#ifdef __cplusplus
}
#endif
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 3af8bdc..40e95f3 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -96,7 +96,7 @@
void ClientContext::set_compression_algorithm(
grpc_compression_algorithm algorithm) {
- char* algorithm_name = nullptr;
+ const char* algorithm_name = nullptr;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
gpr_log(GPR_ERROR, "Name for compression algorithm '%d' unknown.",
algorithm);
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index 4913682..d7876a0 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -190,7 +190,7 @@
void ServerContext::set_compression_algorithm(
grpc_compression_algorithm algorithm) {
- char* algorithm_name = NULL;
+ const char* algorithm_name = NULL;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
gpr_log(GPR_ERROR, "Name for compression algorithm '%d' unknown.",
algorithm);
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py b/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
index 83f21ec..424b153 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
@@ -12,19 +12,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import argparse
import contextlib
-import distutils.spawn
-import errno
-import itertools
+import importlib
import os
-import pkg_resources
+from os import path
+import pkgutil
import shutil
-import subprocess
import sys
import tempfile
import threading
-import time
import unittest
from six import moves
@@ -33,12 +29,22 @@
from grpc.beta import interfaces
from grpc.framework.foundation import future
from grpc.framework.interfaces.face import face
+from grpc_tools import protoc
from tests.unit.framework.common import test_constants
-import tests.protoc_plugin.protos.payload.test_payload_pb2 as payload_pb2
-import tests.protoc_plugin.protos.requests.r.test_requests_pb2 as request_pb2
-import tests.protoc_plugin.protos.responses.test_responses_pb2 as response_pb2
-import tests.protoc_plugin.protos.service.test_service_pb2 as service_pb2
+_RELATIVE_PROTO_PATH = 'relative_proto_path'
+_RELATIVE_PYTHON_OUT = 'relative_python_out'
+
+_PROTO_FILES_PATH_COMPONENTS = (
+ ('beta_grpc_plugin_test', 'payload', 'test_payload.proto',),
+ ('beta_grpc_plugin_test', 'requests', 'r', 'test_requests.proto',),
+ ('beta_grpc_plugin_test', 'responses', 'test_responses.proto',),
+ ('beta_grpc_plugin_test', 'service', 'test_service.proto',),)
+
+_PAYLOAD_PB2 = 'beta_grpc_plugin_test.payload.test_payload_pb2'
+_REQUESTS_PB2 = 'beta_grpc_plugin_test.requests.r.test_requests_pb2'
+_RESPONSES_PB2 = 'beta_grpc_plugin_test.responses.test_responses_pb2'
+_SERVICE_PB2 = 'beta_grpc_plugin_test.service.test_service_pb2'
# Identifiers of entities we expect to find in the generated module.
SERVICER_IDENTIFIER = 'BetaTestServiceServicer'
@@ -47,12 +53,50 @@
STUB_FACTORY_IDENTIFIER = 'beta_create_TestService_stub'
+@contextlib.contextmanager
+def _system_path(path_insertion):
+ old_system_path = sys.path[:]
+ sys.path = sys.path[0:1] + path_insertion + sys.path[1:]
+ yield
+ sys.path = old_system_path
+
+
+def _create_directory_tree(root, path_components_sequence):
+ created = set()
+ for path_components in path_components_sequence:
+ thus_far = ''
+ for path_component in path_components:
+ relative_path = path.join(thus_far, path_component)
+ if relative_path not in created:
+ os.makedirs(path.join(root, relative_path))
+ created.add(relative_path)
+ thus_far = path.join(thus_far, path_component)
+
+
+def _massage_proto_content(raw_proto_content):
+ imports_substituted = raw_proto_content.replace(
+ b'import "tests/protoc_plugin/protos/',
+ b'import "beta_grpc_plugin_test/')
+ package_statement_substituted = imports_substituted.replace(
+ b'package grpc_protoc_plugin;', b'package beta_grpc_protoc_plugin;')
+ return package_statement_substituted
+
+
+def _packagify(directory):
+ for subdirectory, _, _ in os.walk(directory):
+ init_file_name = path.join(subdirectory, '__init__.py')
+ with open(init_file_name, 'wb') as init_file:
+ init_file.write(b'')
+
+
class _ServicerMethods(object):
- def __init__(self):
+ def __init__(self, payload_pb2, responses_pb2):
self._condition = threading.Condition()
self._paused = False
self._fail = False
+ self._payload_pb2 = payload_pb2
+ self._responses_pb2 = responses_pb2
@contextlib.contextmanager
def pause(self): # pylint: disable=invalid-name
@@ -79,22 +123,22 @@
self._condition.wait()
def UnaryCall(self, request, unused_rpc_context):
- response = response_pb2.SimpleResponse()
- response.payload.payload_type = payload_pb2.COMPRESSABLE
+ response = self._responses_pb2.SimpleResponse()
+ response.payload.payload_type = self._payload_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * request.response_size
self._control()
return response
def StreamingOutputCall(self, request, unused_rpc_context):
for parameter in request.response_parameters:
- response = response_pb2.StreamingOutputCallResponse()
- response.payload.payload_type = payload_pb2.COMPRESSABLE
+ response = self._responses_pb2.StreamingOutputCallResponse()
+ response.payload.payload_type = self._payload_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size
self._control()
yield response
def StreamingInputCall(self, request_iter, unused_rpc_context):
- response = response_pb2.StreamingInputCallResponse()
+ response = self._responses_pb2.StreamingInputCallResponse()
aggregated_payload_size = 0
for request in request_iter:
aggregated_payload_size += len(request.payload.payload_compressable)
@@ -105,8 +149,8 @@
def FullDuplexCall(self, request_iter, unused_rpc_context):
for request in request_iter:
for parameter in request.response_parameters:
- response = response_pb2.StreamingOutputCallResponse()
- response.payload.payload_type = payload_pb2.COMPRESSABLE
+ response = self._responses_pb2.StreamingOutputCallResponse()
+ response.payload.payload_type = self._payload_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size
self._control()
yield response
@@ -115,8 +159,8 @@
responses = []
for request in request_iter:
for parameter in request.response_parameters:
- response = response_pb2.StreamingOutputCallResponse()
- response.payload.payload_type = payload_pb2.COMPRESSABLE
+ response = self._responses_pb2.StreamingOutputCallResponse()
+ response.payload.payload_type = self._payload_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size
self._control()
responses.append(response)
@@ -125,7 +169,7 @@
@contextlib.contextmanager
-def _CreateService():
+def _CreateService(payload_pb2, responses_pb2, service_pb2):
"""Provides a servicer backend and a stub.
The servicer is just the implementation of the actual servicer passed to the
@@ -136,7 +180,7 @@
the service bound to the stub and and stub is the stub on which to invoke
RPCs.
"""
- servicer_methods = _ServicerMethods()
+ servicer_methods = _ServicerMethods(payload_pb2, responses_pb2)
class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
@@ -161,12 +205,12 @@
server.start()
channel = implementations.insecure_channel('localhost', port)
stub = getattr(service_pb2, STUB_FACTORY_IDENTIFIER)(channel)
- yield (servicer_methods, stub)
+ yield servicer_methods, stub,
server.stop(0)
@contextlib.contextmanager
-def _CreateIncompleteService():
+def _CreateIncompleteService(service_pb2):
"""Provides a servicer backend that fails to implement methods and its stub.
The servicer is just the implementation of the actual servicer passed to the
@@ -192,16 +236,16 @@
server.stop(0)
-def _streaming_input_request_iterator():
+def _streaming_input_request_iterator(payload_pb2, requests_pb2):
for _ in range(3):
- request = request_pb2.StreamingInputCallRequest()
+ request = requests_pb2.StreamingInputCallRequest()
request.payload.payload_type = payload_pb2.COMPRESSABLE
request.payload.payload_compressable = 'a'
yield request
-def _streaming_output_request():
- request = request_pb2.StreamingOutputCallRequest()
+def _streaming_output_request(requests_pb2):
+ request = requests_pb2.StreamingOutputCallRequest()
sizes = [1, 2, 3]
request.response_parameters.add(size=sizes[0], interval_us=0)
request.response_parameters.add(size=sizes[1], interval_us=0)
@@ -209,11 +253,11 @@
return request
-def _full_duplex_request_iterator():
- request = request_pb2.StreamingOutputCallRequest()
+def _full_duplex_request_iterator(requests_pb2):
+ request = requests_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0)
yield request
- request = request_pb2.StreamingOutputCallRequest()
+ request = requests_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=2, interval_us=0)
request.response_parameters.add(size=3, interval_us=0)
yield request
@@ -227,22 +271,78 @@
methods and does not exist for response-streaming methods.
"""
+ def setUp(self):
+ self._directory = tempfile.mkdtemp(dir='.')
+ self._proto_path = path.join(self._directory, _RELATIVE_PROTO_PATH)
+ self._python_out = path.join(self._directory, _RELATIVE_PYTHON_OUT)
+
+ os.makedirs(self._proto_path)
+ os.makedirs(self._python_out)
+
+ directories_path_components = {
+ proto_file_path_components[:-1]
+ for proto_file_path_components in _PROTO_FILES_PATH_COMPONENTS
+ }
+ _create_directory_tree(self._proto_path, directories_path_components)
+ self._proto_file_names = set()
+ for proto_file_path_components in _PROTO_FILES_PATH_COMPONENTS:
+ raw_proto_content = pkgutil.get_data(
+ 'tests.protoc_plugin.protos',
+ path.join(*proto_file_path_components[1:]))
+ massaged_proto_content = _massage_proto_content(raw_proto_content)
+ proto_file_name = path.join(self._proto_path,
+ *proto_file_path_components)
+ with open(proto_file_name, 'wb') as proto_file:
+ proto_file.write(massaged_proto_content)
+ self._proto_file_names.add(proto_file_name)
+
+ def tearDown(self):
+ shutil.rmtree(self._directory)
+
+ def _protoc(self):
+ args = [
+ '',
+ '--proto_path={}'.format(self._proto_path),
+ '--python_out={}'.format(self._python_out),
+ '--grpc_python_out=grpc_1_0:{}'.format(self._python_out),
+ ] + list(self._proto_file_names)
+ protoc_exit_code = protoc.main(args)
+ self.assertEqual(0, protoc_exit_code)
+
+ _packagify(self._python_out)
+
+ with _system_path([
+ self._python_out,
+ ]):
+ self._payload_pb2 = importlib.import_module(_PAYLOAD_PB2)
+ self._requests_pb2 = importlib.import_module(_REQUESTS_PB2)
+ self._responses_pb2 = importlib.import_module(_RESPONSES_PB2)
+ self._service_pb2 = importlib.import_module(_SERVICE_PB2)
+
def testImportAttributes(self):
+ self._protoc()
+
# check that we can access the generated module and its members.
- self.assertIsNotNone(getattr(service_pb2, SERVICER_IDENTIFIER, None))
- self.assertIsNotNone(getattr(service_pb2, STUB_IDENTIFIER, None))
self.assertIsNotNone(
- getattr(service_pb2, SERVER_FACTORY_IDENTIFIER, None))
+ getattr(self._service_pb2, SERVICER_IDENTIFIER, None))
+ self.assertIsNotNone(getattr(self._service_pb2, STUB_IDENTIFIER, None))
self.assertIsNotNone(
- getattr(service_pb2, STUB_FACTORY_IDENTIFIER, None))
+ getattr(self._service_pb2, SERVER_FACTORY_IDENTIFIER, None))
+ self.assertIsNotNone(
+ getattr(self._service_pb2, STUB_FACTORY_IDENTIFIER, None))
def testUpDown(self):
- with _CreateService():
- request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2):
+ self._requests_pb2.SimpleRequest(response_size=13)
def testIncompleteServicer(self):
- with _CreateIncompleteService() as (_, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateIncompleteService(self._service_pb2) as (_, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
try:
stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
except face.AbortionError as error:
@@ -250,15 +350,21 @@
error.code)
def testUnaryCall(self):
- with _CreateService() as (methods, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
response = stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
expected_response = methods.UnaryCall(request, 'not a real context!')
self.assertEqual(expected_response, response)
def testUnaryCallFuture(self):
- with _CreateService() as (methods, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
# Check that the call does not block waiting for the server to respond.
with methods.pause():
response_future = stub.UnaryCall.future(
@@ -268,8 +374,11 @@
self.assertEqual(expected_response, response)
def testUnaryCallFutureExpired(self):
- with _CreateService() as (methods, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
with methods.pause():
response_future = stub.UnaryCall.future(
request, test_constants.SHORT_TIMEOUT)
@@ -277,24 +386,33 @@
response_future.result()
def testUnaryCallFutureCancelled(self):
- with _CreateService() as (methods, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
with methods.pause():
response_future = stub.UnaryCall.future(request, 1)
response_future.cancel()
self.assertTrue(response_future.cancelled())
def testUnaryCallFutureFailed(self):
- with _CreateService() as (methods, stub):
- request = request_pb2.SimpleRequest(response_size=13)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = self._requests_pb2.SimpleRequest(response_size=13)
with methods.fail():
response_future = stub.UnaryCall.future(
request, test_constants.LONG_TIMEOUT)
self.assertIsNotNone(response_future.exception())
def testStreamingOutputCall(self):
- with _CreateService() as (methods, stub):
- request = _streaming_output_request()
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = _streaming_output_request(self._requests_pb2)
responses = stub.StreamingOutputCall(request,
test_constants.LONG_TIMEOUT)
expected_responses = methods.StreamingOutputCall(
@@ -304,8 +422,11 @@
self.assertEqual(expected_response, response)
def testStreamingOutputCallExpired(self):
- with _CreateService() as (methods, stub):
- request = _streaming_output_request()
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = _streaming_output_request(self._requests_pb2)
with methods.pause():
responses = stub.StreamingOutputCall(
request, test_constants.SHORT_TIMEOUT)
@@ -313,8 +434,11 @@
list(responses)
def testStreamingOutputCallCancelled(self):
- with _CreateService() as (methods, stub):
- request = _streaming_output_request()
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = _streaming_output_request(self._requests_pb2)
responses = stub.StreamingOutputCall(request,
test_constants.LONG_TIMEOUT)
next(responses)
@@ -323,8 +447,11 @@
next(responses)
def testStreamingOutputCallFailed(self):
- with _CreateService() as (methods, stub):
- request = _streaming_output_request()
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request = _streaming_output_request(self._requests_pb2)
with methods.fail():
responses = stub.StreamingOutputCall(request, 1)
self.assertIsNotNone(responses)
@@ -332,30 +459,46 @@
next(responses)
def testStreamingInputCall(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
response = stub.StreamingInputCall(
- _streaming_input_request_iterator(),
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
test_constants.LONG_TIMEOUT)
expected_response = methods.StreamingInputCall(
- _streaming_input_request_iterator(), 'not a real RpcContext!')
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
+ 'not a real RpcContext!')
self.assertEqual(expected_response, response)
def testStreamingInputCallFuture(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.pause():
response_future = stub.StreamingInputCall.future(
- _streaming_input_request_iterator(),
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
test_constants.LONG_TIMEOUT)
response = response_future.result()
expected_response = methods.StreamingInputCall(
- _streaming_input_request_iterator(), 'not a real RpcContext!')
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
+ 'not a real RpcContext!')
self.assertEqual(expected_response, response)
def testStreamingInputCallFutureExpired(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.pause():
response_future = stub.StreamingInputCall.future(
- _streaming_input_request_iterator(),
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
test_constants.SHORT_TIMEOUT)
with self.assertRaises(face.ExpirationError):
response_future.result()
@@ -363,10 +506,14 @@
face.ExpirationError)
def testStreamingInputCallFutureCancelled(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.pause():
response_future = stub.StreamingInputCall.future(
- _streaming_input_request_iterator(),
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
test_constants.LONG_TIMEOUT)
response_future.cancel()
self.assertTrue(response_future.cancelled())
@@ -374,26 +521,38 @@
response_future.result()
def testStreamingInputCallFutureFailed(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.fail():
response_future = stub.StreamingInputCall.future(
- _streaming_input_request_iterator(),
+ _streaming_input_request_iterator(self._payload_pb2,
+ self._requests_pb2),
test_constants.LONG_TIMEOUT)
self.assertIsNotNone(response_future.exception())
def testFullDuplexCall(self):
- with _CreateService() as (methods, stub):
- responses = stub.FullDuplexCall(_full_duplex_request_iterator(),
- test_constants.LONG_TIMEOUT)
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ responses = stub.FullDuplexCall(
+ _full_duplex_request_iterator(self._requests_pb2),
+ test_constants.LONG_TIMEOUT)
expected_responses = methods.FullDuplexCall(
- _full_duplex_request_iterator(), 'not a real RpcContext!')
+ _full_duplex_request_iterator(self._requests_pb2),
+ 'not a real RpcContext!')
for expected_response, response in moves.zip_longest(
expected_responses, responses):
self.assertEqual(expected_response, response)
def testFullDuplexCallExpired(self):
- request_iterator = _full_duplex_request_iterator()
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ request_iterator = _full_duplex_request_iterator(self._requests_pb2)
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.pause():
responses = stub.FullDuplexCall(request_iterator,
test_constants.SHORT_TIMEOUT)
@@ -401,8 +560,11 @@
list(responses)
def testFullDuplexCallCancelled(self):
- with _CreateService() as (methods, stub):
- request_iterator = _full_duplex_request_iterator()
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
+ request_iterator = _full_duplex_request_iterator(self._requests_pb2)
responses = stub.FullDuplexCall(request_iterator,
test_constants.LONG_TIMEOUT)
next(responses)
@@ -411,8 +573,11 @@
next(responses)
def testFullDuplexCallFailed(self):
- request_iterator = _full_duplex_request_iterator()
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ request_iterator = _full_duplex_request_iterator(self._requests_pb2)
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with methods.fail():
responses = stub.FullDuplexCall(request_iterator,
test_constants.LONG_TIMEOUT)
@@ -421,13 +586,16 @@
next(responses)
def testHalfDuplexCall(self):
- with _CreateService() as (methods, stub):
+ self._protoc()
+
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
def half_duplex_request_iterator():
- request = request_pb2.StreamingOutputCallRequest()
+ request = self._requests_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0)
yield request
- request = request_pb2.StreamingOutputCallRequest()
+ request = self._requests_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=2, interval_us=0)
request.response_parameters.add(size=3, interval_us=0)
yield request
@@ -441,6 +609,8 @@
self.assertEqual(expected_response, response)
def testHalfDuplexCallWedged(self):
+ self._protoc()
+
condition = threading.Condition()
wait_cell = [False]
@@ -455,14 +625,15 @@
condition.notify_all()
def half_duplex_request_iterator():
- request = request_pb2.StreamingOutputCallRequest()
+ request = self._requests_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0)
yield request
with condition:
while wait_cell[0]:
condition.wait()
- with _CreateService() as (methods, stub):
+ with _CreateService(self._payload_pb2, self._responses_pb2,
+ self._service_pb2) as (methods, stub):
with wait():
responses = stub.HalfDuplexCall(half_duplex_request_iterator(),
test_constants.SHORT_TIMEOUT)
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
index c5c848a..55a7eed 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
@@ -164,10 +164,10 @@
typedef int(*grpc_compression_algorithm_parse_type)(grpc_slice value, grpc_compression_algorithm *algorithm);
extern grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_import;
#define grpc_compression_algorithm_parse grpc_compression_algorithm_parse_import
-typedef int(*grpc_compression_algorithm_name_type)(grpc_compression_algorithm algorithm, char **name);
+typedef int(*grpc_compression_algorithm_name_type)(grpc_compression_algorithm algorithm, const char **name);
extern grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
#define grpc_compression_algorithm_name grpc_compression_algorithm_name_import
-typedef int(*grpc_stream_compression_algorithm_name_type)(grpc_stream_compression_algorithm algorithm, char **name);
+typedef int(*grpc_stream_compression_algorithm_name_type)(grpc_stream_compression_algorithm algorithm, const char **name);
extern grpc_stream_compression_algorithm_name_type grpc_stream_compression_algorithm_name_import;
#define grpc_stream_compression_algorithm_name grpc_stream_compression_algorithm_name_import
typedef grpc_compression_algorithm(*grpc_compression_algorithm_for_level_type)(grpc_compression_level level, uint32_t accepted_encodings);
diff --git a/test/core/compression/algorithm_test.c b/test/core/compression/algorithm_test.c
index e37c8c3..a11e6e9 100644
--- a/test/core/compression/algorithm_test.c
+++ b/test/core/compression/algorithm_test.c
@@ -35,7 +35,7 @@
gpr_log(GPR_DEBUG, "test_algorithm_mesh");
for (i = 0; i < GRPC_COMPRESS_ALGORITHMS_COUNT; i++) {
- char *name;
+ const char *name;
grpc_compression_algorithm parsed;
grpc_slice mdstr;
grpc_mdelem mdelem;
diff --git a/test/core/compression/compression_test.c b/test/core/compression/compression_test.c
index 4b6026e..326a800 100644
--- a/test/core/compression/compression_test.c
+++ b/test/core/compression/compression_test.c
@@ -57,7 +57,7 @@
static void test_compression_algorithm_name(void) {
int success;
- char *name;
+ const char *name;
size_t i;
const char *valid_names[] = {"identity", "gzip", "deflate"};
const grpc_compression_algorithm valid_algorithms[] = {
diff --git a/test/core/compression/message_compress_test.c b/test/core/compression/message_compress_test.c
index fd692fe..f7f4893 100644
--- a/test/core/compression/message_compress_test.c
+++ b/test/core/compression/message_compress_test.c
@@ -49,7 +49,7 @@
grpc_slice_buffer output;
grpc_slice final;
int was_compressed;
- char *algorithm_name;
+ const char *algorithm_name;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algorithm_name) != 0);
gpr_log(
diff --git a/test/core/end2end/tests/compressed_payload.c b/test/core/end2end/tests/compressed_payload.c
index 429717a..639af15 100644
--- a/test/core/end2end/tests/compressed_payload.c
+++ b/test/core/end2end/tests/compressed_payload.c
@@ -228,7 +228,7 @@
/* with a certain error */
GPR_ASSERT(status == expected_error);
- char *algo_name = NULL;
+ const char *algo_name = NULL;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm_to_disable, &algo_name));
char *expected_details = NULL;
gpr_asprintf(&expected_details, "Compression algorithm '%s' is disabled.",
diff --git a/test/core/end2end/tests/stream_compression_compressed_payload.c b/test/core/end2end/tests/stream_compression_compressed_payload.c
index 11e7999..5a69091 100644
--- a/test/core/end2end/tests/stream_compression_compressed_payload.c
+++ b/test/core/end2end/tests/stream_compression_compressed_payload.c
@@ -228,7 +228,7 @@
/* with a certain error */
GPR_ASSERT(status == expected_error);
- char *algo_name = NULL;
+ const char *algo_name = NULL;
GPR_ASSERT(
grpc_stream_compression_algorithm_name(algorithm_to_disable, &algo_name));
char *expected_details = NULL;
diff --git a/test/core/tsi/ssl_transport_security_test.c b/test/core/tsi/ssl_transport_security_test.c
index 364dfa1..2399b05 100644
--- a/test/core/tsi/ssl_transport_security_test.c
+++ b/test/core/tsi/ssl_transport_security_test.c
@@ -23,7 +23,9 @@
#include "src/core/lib/iomgr/load_file.h"
#include "src/core/lib/security/transport/security_connector.h"
#include "src/core/tsi/ssl_transport_security.h"
+#include "src/core/tsi/transport_security.h"
#include "src/core/tsi/transport_security_adapter.h"
+#include "src/core/tsi/transport_security_interface.h"
#include "test/core/tsi/transport_security_test_lib.h"
#include "test/core/util/test_config.h"
@@ -312,10 +314,10 @@
key_cert_lib->bad_client_pem_key_cert_pair);
gpr_free(key_cert_lib->root_cert);
gpr_free(key_cert_lib);
- /* Destroy others. */
- tsi_ssl_server_handshaker_factory_destroy(
+ /* Unreference others. */
+ tsi_ssl_server_handshaker_factory_unref(
ssl_fixture->server_handshaker_factory);
- tsi_ssl_client_handshaker_factory_destroy(
+ tsi_ssl_client_handshaker_factory_unref(
ssl_fixture->client_handshaker_factory);
}
@@ -536,6 +538,118 @@
}
}
+static const tsi_ssl_handshaker_factory_vtable *original_vtable;
+static bool handshaker_factory_destructor_called;
+
+static void ssl_tsi_test_handshaker_factory_destructor(
+ tsi_ssl_handshaker_factory *factory) {
+ GPR_ASSERT(factory != NULL);
+ handshaker_factory_destructor_called = true;
+ if (original_vtable != NULL && original_vtable->destroy != NULL) {
+ original_vtable->destroy(factory);
+ }
+}
+
+static tsi_ssl_handshaker_factory_vtable test_handshaker_factory_vtable = {
+ ssl_tsi_test_handshaker_factory_destructor};
+
+void test_tsi_ssl_client_handshaker_factory_refcounting() {
+ int i;
+ const char *cert_chain =
+ load_file(SSL_TSI_TEST_CREDENTIALS_DIR, "client.pem");
+
+ tsi_ssl_client_handshaker_factory *client_handshaker_factory;
+ GPR_ASSERT(tsi_create_ssl_client_handshaker_factory(
+ NULL, cert_chain, NULL, NULL, 0, &client_handshaker_factory) ==
+ TSI_OK);
+
+ handshaker_factory_destructor_called = false;
+ original_vtable = tsi_ssl_handshaker_factory_swap_vtable(
+ (tsi_ssl_handshaker_factory *)client_handshaker_factory,
+ &test_handshaker_factory_vtable);
+
+ tsi_handshaker *handshaker[3];
+
+ for (i = 0; i < 3; ++i) {
+ GPR_ASSERT(tsi_ssl_client_handshaker_factory_create_handshaker(
+ client_handshaker_factory, "google.com", &handshaker[i]) ==
+ TSI_OK);
+ }
+
+ tsi_handshaker_destroy(handshaker[1]);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_handshaker_destroy(handshaker[0]);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_ssl_client_handshaker_factory_unref(client_handshaker_factory);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_handshaker_destroy(handshaker[2]);
+ GPR_ASSERT(handshaker_factory_destructor_called);
+
+ gpr_free((void *)cert_chain);
+}
+
+void test_tsi_ssl_server_handshaker_factory_refcounting() {
+ int i;
+ tsi_ssl_server_handshaker_factory *server_handshaker_factory;
+ tsi_handshaker *handshaker[3];
+ const char *cert_chain =
+ load_file(SSL_TSI_TEST_CREDENTIALS_DIR, "server0.pem");
+ tsi_ssl_pem_key_cert_pair cert_pair;
+
+ cert_pair.cert_chain = cert_chain;
+ cert_pair.private_key =
+ load_file(SSL_TSI_TEST_CREDENTIALS_DIR, "server0.key");
+
+ GPR_ASSERT(tsi_create_ssl_server_handshaker_factory(
+ &cert_pair, 1, cert_chain, 0, NULL, NULL, 0,
+ &server_handshaker_factory) == TSI_OK);
+
+ handshaker_factory_destructor_called = false;
+ original_vtable = tsi_ssl_handshaker_factory_swap_vtable(
+ (tsi_ssl_handshaker_factory *)server_handshaker_factory,
+ &test_handshaker_factory_vtable);
+
+ for (i = 0; i < 3; ++i) {
+ GPR_ASSERT(tsi_ssl_server_handshaker_factory_create_handshaker(
+ server_handshaker_factory, &handshaker[i]) == TSI_OK);
+ }
+
+ tsi_handshaker_destroy(handshaker[1]);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_handshaker_destroy(handshaker[0]);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_ssl_server_handshaker_factory_unref(server_handshaker_factory);
+ GPR_ASSERT(!handshaker_factory_destructor_called);
+
+ tsi_handshaker_destroy(handshaker[2]);
+ GPR_ASSERT(handshaker_factory_destructor_called);
+
+ ssl_test_pem_key_cert_pair_destroy(cert_pair);
+}
+
+/* Attempting to create a handshaker factory with invalid parameters should fail
+ * but not crash. */
+void test_tsi_ssl_client_handshaker_factory_bad_params() {
+ const char *cert_chain = "This is not a valid PEM file.";
+
+ tsi_ssl_client_handshaker_factory *client_handshaker_factory;
+ GPR_ASSERT(tsi_create_ssl_client_handshaker_factory(
+ NULL, cert_chain, NULL, NULL, 0, &client_handshaker_factory) ==
+ TSI_INVALID_ARGUMENT);
+ tsi_ssl_client_handshaker_factory_unref(client_handshaker_factory);
+}
+
+void ssl_tsi_test_handshaker_factory_internals() {
+ test_tsi_ssl_client_handshaker_factory_refcounting();
+ test_tsi_ssl_server_handshaker_factory_refcounting();
+ test_tsi_ssl_client_handshaker_factory_bad_params();
+}
+
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_init();
@@ -553,6 +667,7 @@
ssl_tsi_test_do_handshake_alpn_client_server_ok();
ssl_tsi_test_do_round_trip_for_all_configs();
ssl_tsi_test_do_round_trip_odd_buffer_size();
+ ssl_tsi_test_handshaker_factory_internals();
grpc_shutdown();
return 0;
}
diff --git a/tools/run_tests/python_utils/filter_pull_request_tests.py b/tools/run_tests/python_utils/filter_pull_request_tests.py
index 4ad9812..f99143f 100644
--- a/tools/run_tests/python_utils/filter_pull_request_tests.py
+++ b/tools/run_tests/python_utils/filter_pull_request_tests.py
@@ -86,7 +86,6 @@
'^test/distrib/php/': [_PHP_TEST_SUITE],
'^test/distrib/python/': [_PYTHON_TEST_SUITE],
'^test/distrib/ruby/': [_RUBY_TEST_SUITE],
- '^tools/internal_ci/': [],
'^vsprojects/': [_WINDOWS_TEST_SUITE],
'binding\.gyp$': [_NODE_TEST_SUITE],
'composer\.json$': [_PHP_TEST_SUITE],
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index 591a0be..176c109 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -533,6 +533,7 @@
self.config = config
self.args = args
_check_compiler(self.args.compiler, ['default'])
+ self._make_options = ['EMBED_OPENSSL=true', 'EMBED_ZLIB=true']
def test_specs(self):
return [self.config.job_spec(['src/php/bin/run_tests.sh'],
@@ -545,7 +546,7 @@
return ['static_c', 'shared_c']
def make_options(self):
- return []
+ return self._make_options;
def build_steps(self):
return [['tools/run_tests/helper_scripts/build_php.sh']]
@@ -569,6 +570,7 @@
self.config = config
self.args = args
_check_compiler(self.args.compiler, ['default'])
+ self._make_options = ['EMBED_OPENSSL=true', 'EMBED_ZLIB=true']
def test_specs(self):
return [self.config.job_spec(['src/php/bin/run_tests.sh'],
@@ -581,7 +583,7 @@
return ['static_c', 'shared_c']
def make_options(self):
- return []
+ return self._make_options;
def build_steps(self):
return [['tools/run_tests/helper_scripts/build_php.sh']]