diff options
98 files changed, 843 insertions, 590 deletions
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index 18678622c5..9d794a4267 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -93,7 +93,7 @@ typedef enum { typedef struct grpc_arg_pointer_vtable { void *(*copy)(void *p); - void (*destroy)(void *p); + void (*destroy)(grpc_exec_ctx *exec_ctx, void *p); int (*cmp)(void *p, void *q); } grpc_arg_pointer_vtable; diff --git a/include/grpc/impl/codegen/slice.h b/include/grpc/impl/codegen/slice.h index 774ba0e95d..ef60ce1220 100644 --- a/include/grpc/impl/codegen/slice.h +++ b/include/grpc/impl/codegen/slice.h @@ -37,6 +37,8 @@ #include <stddef.h> #include <stdint.h> +typedef struct grpc_exec_ctx grpc_exec_ctx; + /* Slice API A slice represents a contiguous reference counted array of bytes. @@ -57,7 +59,7 @@ grpc_slice_new, or grpc_slice_new_with_len instead. */ typedef struct grpc_slice_refcount { void (*ref)(void *); - void (*unref)(void *); + void (*unref)(grpc_exec_ctx *exec_ctx, void *); } grpc_slice_refcount; #define GRPC_SLICE_INLINED_SIZE (sizeof(size_t) + sizeof(uint8_t *) - 1) diff --git a/src/core/ext/client_channel/http_connect_handshaker.c b/src/core/ext/client_channel/http_connect_handshaker.c index 82042897b2..6ce1953209 100644 --- a/src/core/ext/client_channel/http_connect_handshaker.c +++ b/src/core/ext/client_channel/http_connect_handshaker.c @@ -76,7 +76,7 @@ static void http_connect_handshaker_unref(http_connect_handshaker* handshaker) { if (gpr_unref(&handshaker->refcount)) { gpr_free(handshaker->proxy_server); gpr_free(handshaker->server_name); - grpc_slice_buffer_destroy(&handshaker->write_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &handshaker->write_buffer); grpc_http_parser_destroy(&handshaker->http_parser); grpc_http_response_destroy(&handshaker->http_response); gpr_free(handshaker); @@ -142,7 +142,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg, &handshaker->read_buffer->slices[i + 1], handshaker->read_buffer->count - i - 1); grpc_slice_buffer_swap(handshaker->read_buffer, &tmp_buffer); - grpc_slice_buffer_destroy(&tmp_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &tmp_buffer); break; } } @@ -159,7 +159,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg, // complete (e.g., handling chunked transfer encoding or looking // at the Content-Length: header). if (handshaker->http_parser.state != GRPC_HTTP_BODY) { - grpc_slice_buffer_reset_and_unref(handshaker->read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, handshaker->read_buffer); grpc_endpoint_read(exec_ctx, handshaker->endpoint, handshaker->read_buffer, &handshaker->response_read_closure); return; diff --git a/src/core/ext/client_channel/resolver_factory.h b/src/core/ext/client_channel/resolver_factory.h index 4da42e84d2..76b1f45d80 100644 --- a/src/core/ext/client_channel/resolver_factory.h +++ b/src/core/ext/client_channel/resolver_factory.h @@ -55,7 +55,8 @@ struct grpc_resolver_factory_vtable { void (*unref)(grpc_resolver_factory *factory); /** Implementation of grpc_resolver_factory_create_resolver */ - grpc_resolver *(*create_resolver)(grpc_resolver_factory *factory, + grpc_resolver *(*create_resolver)(grpc_exec_ctx *exec_ctx, + grpc_resolver_factory *factory, grpc_resolver_args *args); /** Implementation of grpc_resolver_factory_get_default_authority */ @@ -70,7 +71,8 @@ void grpc_resolver_factory_unref(grpc_resolver_factory *resolver); /** Create a resolver instance for a name */ grpc_resolver *grpc_resolver_factory_create_resolver( - grpc_resolver_factory *factory, grpc_resolver_args *args); + grpc_exec_ctx *exec_ctx, grpc_resolver_factory *factory, + grpc_resolver_args *args); /** Return a (freshly allocated with gpr_malloc) string representing the default authority to use for this scheme. */ diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c index a5d7c0df45..2175d2094e 100644 --- a/src/core/ext/client_channel/subchannel.c +++ b/src/core/ext/client_channel/subchannel.c @@ -46,6 +46,7 @@ #include "src/core/lib/channel/connected_channel.h" #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/backoff.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/channel_init.h" @@ -206,7 +207,7 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg, gpr_free((void *)c->filters); grpc_channel_args_destroy(c->args); gpr_free(c->addr); - grpc_slice_unref(c->initial_connect_string); + grpc_slice_unref_internal(exec_ctx, c->initial_connect_string); grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker); grpc_connector_unref(exec_ctx, c->connector); grpc_pollset_set_destroy(c->pollset_set); diff --git a/src/core/ext/client_channel/uri_parser.c b/src/core/ext/client_channel/uri_parser.c index 0fbc542ef8..bcb6a1dee4 100644 --- a/src/core/ext/client_channel/uri_parser.c +++ b/src/core/ext/client_channel/uri_parser.c @@ -35,14 +35,13 @@ #include <string.h> -#include <grpc/slice.h> -#include <grpc/slice_buffer.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/port_platform.h> +#include <grpc/support/slice.h> +#include <grpc/support/slice_buffer.h> #include <grpc/support/string_util.h> -#include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" /** a size_t default value... maps to all 1's */ @@ -149,38 +148,38 @@ static void parse_query_parts(grpc_uri *uri) { uri->num_query_parts = 0; return; } - grpc_slice query_slice = - grpc_slice_new(uri->query, strlen(uri->query), do_nothing); - grpc_slice_buffer query_parts; /* the &-separated elements of the query */ - grpc_slice_buffer query_param_parts; /* the =-separated subelements */ + gpr_slice query_slice = + gpr_slice_new(uri->query, strlen(uri->query), do_nothing); + gpr_slice_buffer query_parts; /* the &-separated elements of the query */ + gpr_slice_buffer query_param_parts; /* the =-separated subelements */ - grpc_slice_buffer_init(&query_parts); - grpc_slice_buffer_init(&query_param_parts); + gpr_slice_buffer_init(&query_parts); + gpr_slice_buffer_init(&query_param_parts); - grpc_slice_split(query_slice, QUERY_PARTS_SEPARATOR, &query_parts); + gpr_slice_split(query_slice, QUERY_PARTS_SEPARATOR, &query_parts); uri->query_parts = gpr_malloc(query_parts.count * sizeof(char *)); uri->query_parts_values = gpr_malloc(query_parts.count * sizeof(char *)); uri->num_query_parts = query_parts.count; for (size_t i = 0; i < query_parts.count; i++) { - grpc_slice_split(query_parts.slices[i], QUERY_PARTS_VALUE_SEPARATOR, - &query_param_parts); + gpr_slice_split(query_parts.slices[i], QUERY_PARTS_VALUE_SEPARATOR, + &query_param_parts); GPR_ASSERT(query_param_parts.count > 0); uri->query_parts[i] = - grpc_dump_slice(query_param_parts.slices[0], GPR_DUMP_ASCII); + gpr_dump_slice(query_param_parts.slices[0], GPR_DUMP_ASCII); if (query_param_parts.count > 1) { /* TODO(dgq): only the first value after the separator is considered. * Perhaps all chars after the first separator for the query part should * be included, even if they include the separator. */ uri->query_parts_values[i] = - grpc_dump_slice(query_param_parts.slices[1], GPR_DUMP_ASCII); + gpr_dump_slice(query_param_parts.slices[1], GPR_DUMP_ASCII); } else { uri->query_parts_values[i] = NULL; } - grpc_slice_buffer_reset_and_unref(&query_param_parts); + gpr_slice_buffer_reset_and_unref(&query_param_parts); } - grpc_slice_buffer_destroy(&query_parts); - grpc_slice_buffer_destroy(&query_param_parts); - grpc_slice_unref(query_slice); + gpr_slice_buffer_destroy(&query_parts); + gpr_slice_buffer_destroy(&query_param_parts); + gpr_slice_unref(query_slice); } grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors) { diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c index 2ae96c5e8d..00c7468326 100644 --- a/src/core/ext/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/lb_policy/grpclb/grpclb.c @@ -121,6 +121,7 @@ #include "src/core/ext/lb_policy/grpclb/load_balancer_api.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/iomgr/sockaddr_utils.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/call.h" @@ -970,7 +971,8 @@ static void close_sent_cb(grpc_exec_ctx *exec_ctx, void *arg, static void srv_status_rcvd_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); -static lb_client_data *lb_client_data_create(glb_lb_policy *glb_policy) { +static lb_client_data *lb_client_data_create(grpc_exec_ctx *exec_ctx, + glb_lb_policy *glb_policy) { GPR_ASSERT(glb_policy->server_name != NULL); GPR_ASSERT(glb_policy->server_name[0] != '\0'); @@ -1004,7 +1006,7 @@ static lb_client_data *lb_client_data_create(glb_lb_policy *glb_policy) { grpc_slice request_payload_slice = grpc_grpclb_request_encode(request); lb_client->request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); - grpc_slice_unref(request_payload_slice); + grpc_slice_unref_internal(exec_ctx, request_payload_slice); grpc_grpclb_request_destroy(request); lb_client->status_details = NULL; @@ -1035,7 +1037,7 @@ static void query_for_backends(grpc_exec_ctx *exec_ctx, glb_lb_policy *glb_policy) { GPR_ASSERT(glb_policy->lb_channel != NULL); - glb_policy->lb_client = lb_client_data_create(glb_policy); + glb_policy->lb_client = lb_client_data_create(exec_ctx, glb_policy); grpc_call_error call_error; grpc_op ops[1]; memset(ops, 0, sizeof(ops)); @@ -1126,7 +1128,7 @@ static void res_recv_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_grpclb_serverlist *serverlist = grpc_grpclb_response_parse_serverlist(response_slice); if (serverlist != NULL) { - grpc_slice_unref(response_slice); + grpc_slice_unref_internal(exec_ctx, response_slice); if (grpc_lb_glb_trace) { gpr_log(GPR_INFO, "Serverlist with %lu servers received", (unsigned long)serverlist->num_servers); @@ -1185,7 +1187,7 @@ static void res_recv_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { GPR_ASSERT(serverlist == NULL); gpr_log(GPR_ERROR, "Invalid LB response received: '%s'", grpc_dump_slice(response_slice, GPR_DUMP_ASCII)); - grpc_slice_unref(response_slice); + grpc_slice_unref_internal(exec_ctx, response_slice); /* Disconnect from server returning invalid response. */ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; diff --git a/src/core/ext/resolver/sockaddr/sockaddr_resolver.c b/src/core/ext/resolver/sockaddr/sockaddr_resolver.c index 26a650aadd..a3f13cd61f 100644 --- a/src/core/ext/resolver/sockaddr/sockaddr_resolver.c +++ b/src/core/ext/resolver/sockaddr/sockaddr_resolver.c @@ -47,6 +47,7 @@ #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/lib/iomgr/unix_sockets_posix.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" @@ -161,7 +162,8 @@ char *unix_get_default_authority(grpc_resolver_factory *factory, static void do_nothing(void *ignored) {} -static grpc_resolver *sockaddr_create(grpc_resolver_args *args, +static grpc_resolver *sockaddr_create(grpc_exec_ctx *exec_ctx, + grpc_resolver_args *args, int parse(grpc_uri *uri, grpc_resolved_address *dst)) { if (0 != strcmp(args->uri->authority, "")) { @@ -188,8 +190,8 @@ static grpc_resolver *sockaddr_create(grpc_resolver_args *args, gpr_free(part_str); if (errors_found) break; } - grpc_slice_buffer_destroy(&path_parts); - grpc_slice_unref(path_slice); + grpc_slice_buffer_destroy_internal(exec_ctx, &path_parts); + grpc_slice_unref_internal(exec_ctx, path_slice); if (errors_found) { grpc_lb_addresses_destroy(addresses); return NULL; @@ -219,8 +221,9 @@ static void sockaddr_factory_unref(grpc_resolver_factory *factory) {} #define DECL_FACTORY(name) \ static grpc_resolver *name##_factory_create_resolver( \ - grpc_resolver_factory *factory, grpc_resolver_args *args) { \ - return sockaddr_create(args, parse_##name); \ + grpc_exec_ctx *exec_ctx, grpc_resolver_factory *factory, \ + grpc_resolver_args *args) { \ + return sockaddr_create(exec_ctx, args, parse_##name); \ } \ static const grpc_resolver_factory_vtable name##_factory_vtable = { \ sockaddr_factory_ref, sockaddr_factory_unref, \ diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c index aa2ecf5743..f46e849932 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c @@ -62,7 +62,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server *server, grpc_endpoint *server_endpoint = grpc_tcp_create(grpc_fd_create(fd, name), resource_quota, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, name); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_free(name); diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c index 942638ad7f..15ef778ebc 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c @@ -98,7 +98,7 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep, grpc_server_setup_transport( exec_ctx, connection_state->server_state->server, transport, connection_state->accepting_pollset, args_copy); - grpc_channel_args_destroy(args_copy); + grpc_channel_args_destroy(exec_ctx, args_copy); grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL); } else { /* We need to consume this here, because the server may already have @@ -110,7 +110,7 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep, } else { gpr_log(GPR_ERROR, "Secure transport failed with error %d", status); } - grpc_channel_args_destroy(connection_state->args); + grpc_channel_args_destroy(exec_ctx, connection_state->args); grpc_tcp_server_unref(exec_ctx, connection_state->server_state->tcp); gpr_free(connection_state); } @@ -125,7 +125,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint, gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str); grpc_error_free_string(error_str); GRPC_ERROR_UNREF(error); - grpc_channel_args_destroy(args); + grpc_channel_args_destroy(exec_ctx, args); gpr_free(read_buffer); grpc_handshake_manager_shutdown(exec_ctx, connection_state->handshake_mgr); grpc_handshake_manager_destroy(exec_ctx, connection_state->handshake_mgr); diff --git a/src/core/ext/transport/chttp2/transport/bin_decoder.c b/src/core/ext/transport/chttp2/transport/bin_decoder.c index 3eef80b557..8db36e4a7f 100644 --- a/src/core/ext/transport/chttp2/transport/bin_decoder.c +++ b/src/core/ext/transport/chttp2/transport/bin_decoder.c @@ -34,6 +34,7 @@ #include "src/core/ext/transport/chttp2/transport/bin_decoder.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" @@ -143,7 +144,8 @@ bool grpc_base64_decode_partial(struct grpc_base64_decode_context *ctx) { return true; } -grpc_slice grpc_chttp2_base64_decode(grpc_slice input) { +grpc_slice grpc_chttp2_base64_decode(grpc_exec_ctx *exec_ctx, + grpc_slice input) { size_t input_length = GRPC_SLICE_LENGTH(input); size_t output_length = input_length / 4 * 3; struct grpc_base64_decode_context ctx; @@ -179,7 +181,7 @@ grpc_slice grpc_chttp2_base64_decode(grpc_slice input) { char *s = grpc_dump_slice(input, GPR_DUMP_ASCII); gpr_log(GPR_ERROR, "Base64 decoding failed, input string:\n%s\n", s); gpr_free(s); - grpc_slice_unref(output); + grpc_slice_unref_internal(exec_ctx, output); return gpr_empty_slice(); } GPR_ASSERT(ctx.output_cur == GRPC_SLICE_END_PTR(output)); @@ -187,7 +189,8 @@ grpc_slice grpc_chttp2_base64_decode(grpc_slice input) { return output; } -grpc_slice grpc_chttp2_base64_decode_with_length(grpc_slice input, +grpc_slice grpc_chttp2_base64_decode_with_length(grpc_exec_ctx *exec_ctx, + grpc_slice input, size_t output_length) { size_t input_length = GRPC_SLICE_LENGTH(input); grpc_slice output = grpc_slice_malloc(output_length); @@ -200,7 +203,7 @@ grpc_slice grpc_chttp2_base64_decode_with_length(grpc_slice input, "grpc_chttp2_base64_decode_with_length has a length of %d, which " "has a tail of 1 byte.\n", (int)input_length); - grpc_slice_unref(output); + grpc_slice_unref_internal(exec_ctx, output); return gpr_empty_slice(); } @@ -210,7 +213,7 @@ grpc_slice grpc_chttp2_base64_decode_with_length(grpc_slice input, "than the max possible output length %d.\n", (int)output_length, (int)(input_length / 4 * 3 + tail_xtra[input_length % 4])); - grpc_slice_unref(output); + grpc_slice_unref_internal(exec_ctx, output); return gpr_empty_slice(); } @@ -224,7 +227,7 @@ grpc_slice grpc_chttp2_base64_decode_with_length(grpc_slice input, char *s = grpc_dump_slice(input, GPR_DUMP_ASCII); gpr_log(GPR_ERROR, "Base64 decoding failed, input string:\n%s\n", s); gpr_free(s); - grpc_slice_unref(output); + grpc_slice_unref_internal(exec_ctx, output); return gpr_empty_slice(); } GPR_ASSERT(ctx.output_cur == GRPC_SLICE_END_PTR(output)); diff --git a/src/core/ext/transport/chttp2/transport/bin_decoder.h b/src/core/ext/transport/chttp2/transport/bin_decoder.h index 83a90be519..48ffb2ae3b 100644 --- a/src/core/ext/transport/chttp2/transport/bin_decoder.h +++ b/src/core/ext/transport/chttp2/transport/bin_decoder.h @@ -55,12 +55,13 @@ bool grpc_base64_decode_partial(struct grpc_base64_decode_context *ctx); /* base64 decode a slice with pad chars. Returns a new slice, does not take ownership of the input. Returns an empty slice if decoding is failed. */ -grpc_slice grpc_chttp2_base64_decode(grpc_slice input); +grpc_slice grpc_chttp2_base64_decode(grpc_exec_ctx *exec_ctx, grpc_slice input); /* base64 decode a slice without pad chars, data length is needed. Returns a new slice, does not take ownership of the input. Returns an empty slice if decoding is failed. */ -grpc_slice grpc_chttp2_base64_decode_with_length(grpc_slice input, +grpc_slice grpc_chttp2_base64_decode_with_length(grpc_exec_ctx *exec_ctx, + grpc_slice input, size_t output_length); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_BIN_DECODER_H */ diff --git a/src/core/ext/transport/chttp2/transport/bin_encoder.h b/src/core/ext/transport/chttp2/transport/bin_encoder.h index 9e143b46e2..477559d0e2 100644 --- a/src/core/ext/transport/chttp2/transport/bin_encoder.h +++ b/src/core/ext/transport/chttp2/transport/bin_encoder.h @@ -47,7 +47,7 @@ grpc_slice grpc_chttp2_huffman_compress(grpc_slice input); /* equivalent to: grpc_slice x = grpc_chttp2_base64_encode(input); grpc_slice y = grpc_chttp2_huffman_compress(x); - grpc_slice_unref(x); + grpc_slice_unref_internal(exec_ctx, x); return y; */ grpc_slice grpc_chttp2_base64_encode_and_huffman_compress_impl( grpc_slice input); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 1ffa9165b2..0c61159495 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -51,6 +51,7 @@ #include "src/core/lib/http/parser.h" #include "src/core/lib/iomgr/workqueue.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/static_metadata.h" @@ -144,12 +145,12 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, grpc_endpoint_destroy(exec_ctx, t->ep); - grpc_slice_buffer_destroy(&t->qbuf); + grpc_slice_buffer_destroy_internal(exec_ctx, &t->qbuf); - grpc_slice_buffer_destroy(&t->outbuf); - grpc_chttp2_hpack_compressor_destroy(&t->hpack_compressor); + grpc_slice_buffer_destroy_internal(exec_ctx, &t->outbuf); + grpc_chttp2_hpack_compressor_destroy(exec_ctx, &t->hpack_compressor); - grpc_slice_buffer_destroy(&t->read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &t->read_buffer); grpc_chttp2_hpack_parser_destroy(&t->hpack_parser); grpc_chttp2_goaway_parser_destroy(&t->goaway_parser); @@ -532,7 +533,7 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, grpc_chttp2_data_parser_destroy(exec_ctx, &s->data_parser); grpc_chttp2_incoming_metadata_buffer_destroy(&s->metadata_buffer[0]); grpc_chttp2_incoming_metadata_buffer_destroy(&s->metadata_buffer[1]); - grpc_slice_buffer_destroy(&s->flow_controlled_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &s->flow_controlled_buffer); GRPC_ERROR_UNREF(s->read_closed_error); GRPC_ERROR_UNREF(s->write_closed_error); @@ -761,7 +762,7 @@ void grpc_chttp2_add_incoming_goaway(grpc_exec_ctx *exec_ctx, char *msg = grpc_dump_slice(goaway_text, GPR_DUMP_HEX | GPR_DUMP_ASCII); GRPC_CHTTP2_IF_TRACING( gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg)); - grpc_slice_unref(goaway_text); + grpc_slice_unref_internal(exec_ctx, goaway_text); t->seen_goaway = 1; /* lie: use transient failure from the transport to indicate goaway has been * received */ @@ -1244,7 +1245,7 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, if (op->send_goaway) { send_goaway(exec_ctx, t, grpc_chttp2_grpc_status_to_http2_error(op->goaway_status), - grpc_slice_ref(*op->goaway_message)); + grpc_slice_ref_internal(*op->goaway_message)); } if (op->set_accept_stream) { @@ -1474,21 +1475,22 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, char status_string[GPR_LTOA_MIN_BUFSIZE]; gpr_ltoa(status, status_string); grpc_chttp2_incoming_metadata_buffer_add( - &s->metadata_buffer[1], - grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_GRPC_STATUS, grpc_mdstr_from_string(status_string))); + &s->metadata_buffer[1], grpc_mdelem_from_metadata_strings( + exec_ctx, GRPC_MDSTR_GRPC_STATUS, + grpc_mdstr_from_string(status_string))); if (slice) { grpc_chttp2_incoming_metadata_buffer_add( &s->metadata_buffer[1], grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_GRPC_MESSAGE, - grpc_mdstr_from_slice(grpc_slice_ref(*slice)))); + exec_ctx, GRPC_MDSTR_GRPC_MESSAGE, + grpc_mdstr_from_slice(exec_ctx, + grpc_slice_ref_internal(*slice)))); } s->published_metadata[1] = GRPC_METADATA_SYNTHESIZED_FROM_FAKE; grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s); } if (slice) { - grpc_slice_unref(*slice); + grpc_slice_unref_internal(exec_ctx, *slice); } } @@ -1862,7 +1864,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, keep_reading = true; GRPC_CHTTP2_REF_TRANSPORT(t, "keep_reading"); } - grpc_slice_buffer_reset_and_unref(&t->read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &t->read_buffer); if (keep_reading) { grpc_endpoint_read(exec_ctx, t->ep, &t->read_buffer, &t->read_action_begin); @@ -1916,7 +1918,7 @@ static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs) { if (gpr_unref(&bs->refs)) { GRPC_ERROR_UNREF(bs->error); - grpc_slice_buffer_destroy(&bs->slices); + grpc_slice_buffer_destroy_internal(exec_ctx, &bs->slices); gpr_mu_destroy(&bs->slice_mu); gpr_free(bs); } diff --git a/src/core/ext/transport/chttp2/transport/hpack_encoder.c b/src/core/ext/transport/chttp2/transport/hpack_encoder.c index eb68fe3138..49a8326f62 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_encoder.c +++ b/src/core/ext/transport/chttp2/transport/hpack_encoder.c @@ -48,6 +48,7 @@ #include "src/core/ext/transport/chttp2/transport/bin_encoder.h" #include "src/core/ext/transport/chttp2/transport/hpack_table.h" #include "src/core/ext/transport/chttp2/transport/varint.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/transport/metadata.h" #include "src/core/lib/transport/static_metadata.h" #include "src/core/lib/transport/timeout_encoding.h" @@ -183,7 +184,8 @@ static void evict_entry(grpc_chttp2_hpack_compressor *c) { } /* add an element to the decoder table */ -static void add_elem(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem) { +static void add_elem(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_compressor *c, + grpc_mdelem *elem) { uint32_t key_hash = elem->key->hash; uint32_t elem_hash = GRPC_MDSTR_KV_HASH(key_hash, elem->value->hash); uint32_t new_index = c->tail_remote_index + c->table_elems + 1; @@ -227,12 +229,12 @@ static void add_elem(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem) { } else if (c->indices_elems[HASH_FRAGMENT_2(elem_hash)] < c->indices_elems[HASH_FRAGMENT_3(elem_hash)]) { /* not there: replace oldest */ - GRPC_MDELEM_UNREF(c->entries_elems[HASH_FRAGMENT_2(elem_hash)]); + GRPC_MDELEM_UNREF(exec_ctx, c->entries_elems[HASH_FRAGMENT_2(elem_hash)]); c->entries_elems[HASH_FRAGMENT_2(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_2(elem_hash)] = new_index; } else { /* not there: replace oldest */ - GRPC_MDELEM_UNREF(c->entries_elems[HASH_FRAGMENT_3(elem_hash)]); + GRPC_MDELEM_UNREF(exec_ctx, c->entries_elems[HASH_FRAGMENT_3(elem_hash)]); c->entries_elems[HASH_FRAGMENT_3(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_3(elem_hash)] = new_index; } @@ -251,11 +253,11 @@ static void add_elem(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem) { c->indices_keys[HASH_FRAGMENT_3(key_hash)] = new_index; } else if (c->indices_keys[HASH_FRAGMENT_2(key_hash)] < c->indices_keys[HASH_FRAGMENT_3(key_hash)]) { - GRPC_MDSTR_UNREF(c->entries_keys[HASH_FRAGMENT_2(key_hash)]); + GRPC_MDSTR_UNREF(exec_ctx, c->entries_keys[HASH_FRAGMENT_2(key_hash)]); c->entries_keys[HASH_FRAGMENT_2(key_hash)] = GRPC_MDSTR_REF(elem->key); c->indices_keys[HASH_FRAGMENT_2(key_hash)] = new_index; } else { - GRPC_MDSTR_UNREF(c->entries_keys[HASH_FRAGMENT_3(key_hash)]); + GRPC_MDSTR_UNREF(exec_ctx, c->entries_keys[HASH_FRAGMENT_3(key_hash)]); c->entries_keys[HASH_FRAGMENT_3(key_hash)] = GRPC_MDSTR_REF(elem->key); c->indices_keys[HASH_FRAGMENT_3(key_hash)] = new_index; } @@ -294,7 +296,7 @@ static void emit_lithdr_incidx(grpc_chttp2_hpack_compressor *c, add_tiny_header_data(st, len_pfx), len_pfx); GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, huffman_prefix, add_tiny_header_data(st, len_val_len), len_val_len); - add_header_data(st, grpc_slice_ref(value_slice)); + add_header_data(st, grpc_slice_ref_internal(value_slice)); } static void emit_lithdr_noidx(grpc_chttp2_hpack_compressor *c, @@ -311,7 +313,7 @@ static void emit_lithdr_noidx(grpc_chttp2_hpack_compressor *c, add_tiny_header_data(st, len_pfx), len_pfx); GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, huffman_prefix, add_tiny_header_data(st, len_val_len), len_val_len); - add_header_data(st, grpc_slice_ref(value_slice)); + add_header_data(st, grpc_slice_ref_internal(value_slice)); } static void emit_lithdr_incidx_v(grpc_chttp2_hpack_compressor *c, @@ -327,10 +329,10 @@ static void emit_lithdr_incidx_v(grpc_chttp2_hpack_compressor *c, *add_tiny_header_data(st, 1) = 0x40; GRPC_CHTTP2_WRITE_VARINT(len_key, 1, 0x00, add_tiny_header_data(st, len_key_len), len_key_len); - add_header_data(st, grpc_slice_ref(elem->key->slice)); + add_header_data(st, grpc_slice_ref_internal(elem->key->slice)); GRPC_CHTTP2_WRITE_VARINT(len_val, 1, huffman_prefix, add_tiny_header_data(st, len_val_len), len_val_len); - add_header_data(st, grpc_slice_ref(value_slice)); + add_header_data(st, grpc_slice_ref_internal(value_slice)); } static void emit_lithdr_noidx_v(grpc_chttp2_hpack_compressor *c, @@ -346,10 +348,10 @@ static void emit_lithdr_noidx_v(grpc_chttp2_hpack_compressor *c, *add_tiny_header_data(st, 1) = 0x00; GRPC_CHTTP2_WRITE_VARINT(len_key, 1, 0x00, add_tiny_header_data(st, len_key_len), len_key_len); - add_header_data(st, grpc_slice_ref(elem->key->slice)); + add_header_data(st, grpc_slice_ref_internal(elem->key->slice)); GRPC_CHTTP2_WRITE_VARINT(len_val, 1, huffman_prefix, add_tiny_header_data(st, len_val_len), len_val_len); - add_header_data(st, grpc_slice_ref(value_slice)); + add_header_data(st, grpc_slice_ref_internal(value_slice)); } static void emit_advertise_table_size_change(grpc_chttp2_hpack_compressor *c, @@ -366,8 +368,8 @@ static uint32_t dynidx(grpc_chttp2_hpack_compressor *c, uint32_t elem_index) { } /* encode an mdelem */ -static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, - framer_state *st) { +static void hpack_enc(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_compressor *c, + grpc_mdelem *elem, framer_state *st) { uint32_t key_hash = elem->key->hash; uint32_t elem_hash = GRPC_MDSTR_KV_HASH(key_hash, elem->value->hash); size_t decoder_space_usage; @@ -417,7 +419,7 @@ static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, /* HIT: key (first cuckoo hash) */ if (should_add_elem) { emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st); - add_elem(c, elem); + add_elem(exec_ctx, c, elem); return; } else { emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st); @@ -432,7 +434,7 @@ static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, /* HIT: key (first cuckoo hash) */ if (should_add_elem) { emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st); - add_elem(c, elem); + add_elem(exec_ctx, c, elem); return; } else { emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st); @@ -445,7 +447,7 @@ static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, if (should_add_elem) { emit_lithdr_incidx_v(c, elem, st); - add_elem(c, elem); + add_elem(exec_ctx, c, elem); return; } else { emit_lithdr_noidx_v(c, elem, st); @@ -457,16 +459,17 @@ static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, #define STRLEN_LIT(x) (sizeof(x) - 1) #define TIMEOUT_KEY "grpc-timeout" -static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, +static void deadline_enc(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, framer_state *st) { char timeout_str[GRPC_HTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE]; grpc_mdelem *mdelem; grpc_http2_encode_timeout( gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str); mdelem = grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_GRPC_TIMEOUT, grpc_mdstr_from_string(timeout_str)); - hpack_enc(c, mdelem, st); - GRPC_MDELEM_UNREF(mdelem); + exec_ctx, GRPC_MDSTR_GRPC_TIMEOUT, grpc_mdstr_from_string(timeout_str)); + hpack_enc(exec_ctx, c, mdelem, st); + GRPC_MDELEM_UNREF(exec_ctx, mdelem); } static uint32_t elems_for_bytes(uint32_t bytes) { return (bytes + 31) / 32; } @@ -483,11 +486,12 @@ void grpc_chttp2_hpack_compressor_init(grpc_chttp2_hpack_compressor *c) { sizeof(*c->table_elem_size) * c->cap_table_elems); } -void grpc_chttp2_hpack_compressor_destroy(grpc_chttp2_hpack_compressor *c) { +void grpc_chttp2_hpack_compressor_destroy(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hpack_compressor *c) { int i; for (i = 0; i < GRPC_CHTTP2_HPACKC_NUM_VALUES; i++) { - if (c->entries_keys[i]) GRPC_MDSTR_UNREF(c->entries_keys[i]); - if (c->entries_elems[i]) GRPC_MDELEM_UNREF(c->entries_elems[i]); + if (c->entries_keys[i]) GRPC_MDSTR_UNREF(exec_ctx, c->entries_keys[i]); + if (c->entries_elems[i]) GRPC_MDELEM_UNREF(exec_ctx, c->entries_elems[i]); } gpr_free(c->table_elem_size); } @@ -542,7 +546,8 @@ void grpc_chttp2_hpack_compressor_set_max_table_size( } } -void grpc_chttp2_encode_header(grpc_chttp2_hpack_compressor *c, +void grpc_chttp2_encode_header(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hpack_compressor *c, uint32_t stream_id, grpc_metadata_batch *metadata, int is_eof, size_t max_frame_size, @@ -571,11 +576,11 @@ void grpc_chttp2_encode_header(grpc_chttp2_hpack_compressor *c, } grpc_metadata_batch_assert_ok(metadata); for (l = metadata->list.head; l; l = l->next) { - hpack_enc(c, l->md, &st); + hpack_enc(exec_ctx, c, l->md, &st); } deadline = metadata->deadline; if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) { - deadline_enc(c, deadline, &st); + deadline_enc(exec_ctx, c, deadline, &st); } finish_frame(&st, 1, is_eof); diff --git a/src/core/ext/transport/chttp2/transport/hpack_encoder.h b/src/core/ext/transport/chttp2/transport/hpack_encoder.h index bcbd675ca2..3a35496ec8 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_encoder.h +++ b/src/core/ext/transport/chttp2/transport/hpack_encoder.h @@ -83,13 +83,15 @@ typedef struct { } grpc_chttp2_hpack_compressor; void grpc_chttp2_hpack_compressor_init(grpc_chttp2_hpack_compressor *c); -void grpc_chttp2_hpack_compressor_destroy(grpc_chttp2_hpack_compressor *c); +void grpc_chttp2_hpack_compressor_destroy(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hpack_compressor *c); void grpc_chttp2_hpack_compressor_set_max_table_size( grpc_chttp2_hpack_compressor *c, uint32_t max_table_size); void grpc_chttp2_hpack_compressor_set_max_usable_size( grpc_chttp2_hpack_compressor *c, uint32_t max_table_size); -void grpc_chttp2_encode_header(grpc_chttp2_hpack_compressor *c, uint32_t id, +void grpc_chttp2_encode_header(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hpack_compressor *c, uint32_t id, grpc_metadata_batch *metadata, int is_eof, size_t max_frame_size, grpc_transport_one_way_stats *stats, diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.c b/src/core/ext/transport/chttp2/transport/hpack_parser.c index 1046c31dda..e805aac8c4 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_parser.c +++ b/src/core/ext/transport/chttp2/transport/hpack_parser.c @@ -669,11 +669,11 @@ static const uint8_t inverse_base64[256] = { static grpc_error *on_hdr(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, grpc_mdelem *md, int add_to_table) { if (add_to_table) { - grpc_error *err = grpc_chttp2_hptbl_add(&p->table, md); + grpc_error *err = grpc_chttp2_hptbl_add(exec_ctx, &p->table, md); if (err != GRPC_ERROR_NONE) return err; } if (p->on_header == NULL) { - GRPC_MDELEM_UNREF(md); + GRPC_MDELEM_UNREF(exec_ctx, md); return GRPC_ERROR_CREATE("on_header callback not set"); } p->on_header(exec_ctx, p->on_header_user_data, md); @@ -814,10 +814,10 @@ static grpc_error *finish_lithdr_incidx(grpc_exec_ctx *exec_ctx, const uint8_t *end) { grpc_mdelem *md = grpc_chttp2_hptbl_lookup(&p->table, p->index); GPR_ASSERT(md != NULL); /* handled in string parsing */ - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(GRPC_MDSTR_REF(md->key), - take_string(p, &p->value)), - 1); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, GRPC_MDSTR_REF(md->key), + take_string(p, &p->value)), + 1); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -827,10 +827,10 @@ static grpc_error *finish_lithdr_incidx_v(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, const uint8_t *cur, const uint8_t *end) { - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(take_string(p, &p->key), - take_string(p, &p->value)), - 1); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, take_string(p, &p->key), + take_string(p, &p->value)), + 1); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -882,10 +882,10 @@ static grpc_error *finish_lithdr_notidx(grpc_exec_ctx *exec_ctx, const uint8_t *end) { grpc_mdelem *md = grpc_chttp2_hptbl_lookup(&p->table, p->index); GPR_ASSERT(md != NULL); /* handled in string parsing */ - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(GRPC_MDSTR_REF(md->key), - take_string(p, &p->value)), - 0); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, GRPC_MDSTR_REF(md->key), + take_string(p, &p->value)), + 0); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -895,10 +895,10 @@ static grpc_error *finish_lithdr_notidx_v(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, const uint8_t *cur, const uint8_t *end) { - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(take_string(p, &p->key), - take_string(p, &p->value)), - 0); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, take_string(p, &p->key), + take_string(p, &p->value)), + 0); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -950,10 +950,10 @@ static grpc_error *finish_lithdr_nvridx(grpc_exec_ctx *exec_ctx, const uint8_t *end) { grpc_mdelem *md = grpc_chttp2_hptbl_lookup(&p->table, p->index); GPR_ASSERT(md != NULL); /* handled in string parsing */ - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(GRPC_MDSTR_REF(md->key), - take_string(p, &p->value)), - 0); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, GRPC_MDSTR_REF(md->key), + take_string(p, &p->value)), + 0); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -963,10 +963,10 @@ static grpc_error *finish_lithdr_nvridx_v(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, const uint8_t *cur, const uint8_t *end) { - grpc_error *err = on_hdr( - exec_ctx, p, grpc_mdelem_from_metadata_strings(take_string(p, &p->key), - take_string(p, &p->value)), - 0); + grpc_error *err = on_hdr(exec_ctx, p, grpc_mdelem_from_metadata_strings( + exec_ctx, take_string(p, &p->key), + take_string(p, &p->value)), + 0); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -1019,7 +1019,7 @@ static grpc_error *finish_max_tbl_size(grpc_exec_ctx *exec_ctx, gpr_log(GPR_INFO, "MAX TABLE SIZE: %d", p->index); } grpc_error *err = - grpc_chttp2_hptbl_set_current_table_size(&p->table, p->index); + grpc_chttp2_hptbl_set_current_table_size(exec_ctx, &p->table, p->index); if (err != GRPC_ERROR_NONE) return parse_error(exec_ctx, p, cur, end, err); return parse_begin(exec_ctx, p, cur, end); } @@ -1545,7 +1545,7 @@ void grpc_chttp2_hpack_parser_init(grpc_chttp2_hpack_parser *p) { p->value.length = 0; p->dynamic_table_update_allowed = 2; p->last_error = GRPC_ERROR_NONE; - grpc_chttp2_hptbl_init(&p->table); + grpc_chttp2_hptbl_init(exec_ctx, &p->table); } void grpc_chttp2_hpack_parser_set_has_priority(grpc_chttp2_hpack_parser *p) { @@ -1554,7 +1554,7 @@ void grpc_chttp2_hpack_parser_set_has_priority(grpc_chttp2_hpack_parser *p) { } void grpc_chttp2_hpack_parser_destroy(grpc_chttp2_hpack_parser *p) { - grpc_chttp2_hptbl_destroy(&p->table); + grpc_chttp2_hptbl_destroy(exec_ctx, &p->table); GRPC_ERROR_UNREF(p->last_error); gpr_free(p->key.str); gpr_free(p->value.str); diff --git a/src/core/ext/transport/chttp2/transport/hpack_table.c b/src/core/ext/transport/chttp2/transport/hpack_table.c index 2dc793d304..26d4036d49 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_table.c +++ b/src/core/ext/transport/chttp2/transport/hpack_table.c @@ -179,7 +179,7 @@ static uint32_t entries_for_bytes(uint32_t bytes) { GRPC_CHTTP2_HPACK_ENTRY_OVERHEAD; } -void grpc_chttp2_hptbl_init(grpc_chttp2_hptbl *tbl) { +void grpc_chttp2_hptbl_init(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl) { size_t i; memset(tbl, 0, sizeof(*tbl)); @@ -190,18 +190,20 @@ void grpc_chttp2_hptbl_init(grpc_chttp2_hptbl *tbl) { tbl->ents = gpr_malloc(sizeof(*tbl->ents) * tbl->cap_entries); memset(tbl->ents, 0, sizeof(*tbl->ents) * tbl->cap_entries); for (i = 1; i <= GRPC_CHTTP2_LAST_STATIC_ENTRY; i++) { - tbl->static_ents[i - 1] = - grpc_mdelem_from_strings(static_table[i].key, static_table[i].value); + tbl->static_ents[i - 1] = grpc_mdelem_from_strings( + exec_ctx, static_table[i].key, static_table[i].value); } } -void grpc_chttp2_hptbl_destroy(grpc_chttp2_hptbl *tbl) { +void grpc_chttp2_hptbl_destroy(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl) { size_t i; for (i = 0; i < GRPC_CHTTP2_LAST_STATIC_ENTRY; i++) { - GRPC_MDELEM_UNREF(tbl->static_ents[i]); + GRPC_MDELEM_UNREF(exec_ctx, tbl->static_ents[i]); } for (i = 0; i < tbl->num_ents; i++) { - GRPC_MDELEM_UNREF(tbl->ents[(tbl->first_ent + i) % tbl->cap_entries]); + GRPC_MDELEM_UNREF(exec_ctx, + tbl->ents[(tbl->first_ent + i) % tbl->cap_entries]); } gpr_free(tbl->ents); } @@ -224,7 +226,7 @@ grpc_mdelem *grpc_chttp2_hptbl_lookup(const grpc_chttp2_hptbl *tbl, } /* Evict one element from the table */ -static void evict1(grpc_chttp2_hptbl *tbl) { +static void evict1(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl) { grpc_mdelem *first_ent = tbl->ents[tbl->first_ent]; size_t elem_bytes = GRPC_SLICE_LENGTH(first_ent->key->slice) + GRPC_SLICE_LENGTH(first_ent->value->slice) + @@ -233,7 +235,7 @@ static void evict1(grpc_chttp2_hptbl *tbl) { tbl->mem_used -= (uint32_t)elem_bytes; tbl->first_ent = ((tbl->first_ent + 1) % tbl->cap_entries); tbl->num_ents--; - GRPC_MDELEM_UNREF(first_ent); + GRPC_MDELEM_UNREF(exec_ctx, first_ent); } static void rebuild_ents(grpc_chttp2_hptbl *tbl, uint32_t new_cap) { @@ -249,7 +251,8 @@ static void rebuild_ents(grpc_chttp2_hptbl *tbl, uint32_t new_cap) { tbl->first_ent = 0; } -void grpc_chttp2_hptbl_set_max_bytes(grpc_chttp2_hptbl *tbl, +void grpc_chttp2_hptbl_set_max_bytes(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, uint32_t max_bytes) { if (tbl->max_bytes == max_bytes) { return; @@ -258,12 +261,13 @@ void grpc_chttp2_hptbl_set_max_bytes(grpc_chttp2_hptbl *tbl, gpr_log(GPR_DEBUG, "Update hpack parser max size to %d", max_bytes); } while (tbl->mem_used > max_bytes) { - evict1(tbl); + evict1(exec_ctx, tbl); } tbl->max_bytes = max_bytes; } -grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_chttp2_hptbl *tbl, +grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, uint32_t bytes) { if (tbl->current_table_bytes == bytes) { return GRPC_ERROR_NONE; @@ -281,7 +285,7 @@ grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_chttp2_hptbl *tbl, gpr_log(GPR_DEBUG, "Update hpack parser table size to %d", bytes); } while (tbl->mem_used > bytes) { - evict1(tbl); + evict1(exec_ctx, tbl); } tbl->current_table_bytes = bytes; tbl->max_entries = entries_for_bytes(bytes); @@ -296,7 +300,8 @@ grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_chttp2_hptbl *tbl, return GRPC_ERROR_NONE; } -grpc_error *grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { +grpc_error *grpc_chttp2_hptbl_add(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { /* determine how many bytes of buffer this entry represents */ size_t elem_bytes = GRPC_SLICE_LENGTH(md->key->slice) + GRPC_SLICE_LENGTH(md->value->slice) + @@ -326,14 +331,14 @@ grpc_error *grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { * empty table. */ while (tbl->num_ents) { - evict1(tbl); + evict1(exec_ctx, tbl); } return GRPC_ERROR_NONE; } /* evict entries to ensure no overflow */ while (elem_bytes > (size_t)tbl->current_table_bytes - tbl->mem_used) { - evict1(tbl); + evict1(exec_ctx, tbl); } /* copy the finalized entry in */ diff --git a/src/core/ext/transport/chttp2/transport/hpack_table.h b/src/core/ext/transport/chttp2/transport/hpack_table.h index 2ca130e64b..144574ef06 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_table.h +++ b/src/core/ext/transport/chttp2/transport/hpack_table.h @@ -84,18 +84,21 @@ typedef struct { } grpc_chttp2_hptbl; /* initialize a hpack table */ -void grpc_chttp2_hptbl_init(grpc_chttp2_hptbl *tbl); -void grpc_chttp2_hptbl_destroy(grpc_chttp2_hptbl *tbl); -void grpc_chttp2_hptbl_set_max_bytes(grpc_chttp2_hptbl *tbl, +void grpc_chttp2_hptbl_init(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl); +void grpc_chttp2_hptbl_destroy(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl); +void grpc_chttp2_hptbl_set_max_bytes(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, uint32_t max_bytes); -grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_chttp2_hptbl *tbl, +grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, uint32_t bytes); /* lookup a table entry based on its hpack index */ grpc_mdelem *grpc_chttp2_hptbl_lookup(const grpc_chttp2_hptbl *tbl, uint32_t index); /* add a table entry to the index */ -grpc_error *grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, +grpc_error *grpc_chttp2_hptbl_add(grpc_exec_ctx *exec_ctx, + grpc_chttp2_hptbl *tbl, grpc_mdelem *md) GRPC_MUST_USE_RESULT; /* Find a key/value pair in the table... returns the index in the table of the most similar entry, or 0 if the value was not found */ diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index 139e7387c4..d7f45b16ad 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -39,6 +39,7 @@ #include "src/core/ext/transport/chttp2/transport/http2_errors.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" static void add_to_write_list(grpc_chttp2_write_cb **list, grpc_chttp2_write_cb *cb) { @@ -254,7 +255,7 @@ void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing:end"); } - grpc_slice_buffer_reset_and_unref(&t->outbuf); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &t->outbuf); GRPC_ERROR_UNREF(error); GPR_TIMER_END("grpc_chttp2_end_write", 0); } diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c index cfc072c0b5..c956b0febc 100644 --- a/src/core/lib/channel/channel_args.c +++ b/src/core/lib/channel/channel_args.c @@ -184,7 +184,7 @@ grpc_channel_args *grpc_channel_args_normalize(const grpc_channel_args *a) { return b; } -void grpc_channel_args_destroy(grpc_channel_args *a) { +void grpc_channel_args_destroy(grpc_exec_ctx *exec_ctx, grpc_channel_args *a) { size_t i; if (!a) return; for (i = 0; i < a->num_args; i++) { @@ -195,7 +195,8 @@ void grpc_channel_args_destroy(grpc_channel_args *a) { case GRPC_ARG_INTEGER: break; case GRPC_ARG_POINTER: - a->args[i].value.pointer.vtable->destroy(a->args[i].value.pointer.p); + a->args[i].value.pointer.vtable->destroy(exec_ctx, + a->args[i].value.pointer.p); break; } gpr_free(a->args[i].key); @@ -249,7 +250,8 @@ static int find_compression_algorithm_states_bitset(const grpc_channel_args *a, } grpc_channel_args *grpc_channel_args_compression_algorithm_set_state( - grpc_channel_args **a, grpc_compression_algorithm algorithm, int state) { + grpc_exec_ctx *exec_ctx, grpc_channel_args **a, + grpc_compression_algorithm algorithm, int state) { int *states_arg = NULL; grpc_channel_args *result = *a; const int states_arg_found = @@ -282,7 +284,7 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state( GPR_BITCLEAR((unsigned *)&tmp.value.integer, algorithm); } result = grpc_channel_args_copy_and_add(*a, &tmp, 1); - grpc_channel_args_destroy(*a); + grpc_channel_args_destroy(exec_ctx, *a); *a = result; } return result; diff --git a/src/core/lib/channel/channel_args.h b/src/core/lib/channel/channel_args.h index 1e05303471..5933133413 100644 --- a/src/core/lib/channel/channel_args.h +++ b/src/core/lib/channel/channel_args.h @@ -67,7 +67,7 @@ grpc_channel_args *grpc_channel_args_merge(const grpc_channel_args *a, const grpc_channel_args *b); /** Destroy arguments created by \a grpc_channel_args_copy */ -void grpc_channel_args_destroy(grpc_channel_args *a); +void grpc_channel_args_destroy(grpc_exec_ctx *exec_ctx, grpc_channel_args *a); /** Returns the compression algorithm set in \a a. */ grpc_compression_algorithm grpc_channel_args_get_compression_algorithm( @@ -87,7 +87,8 @@ grpc_channel_args *grpc_channel_args_set_compression_algorithm( * modified to point to the returned instance (which may be different from the * input value of \a a). */ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state( - grpc_channel_args **a, grpc_compression_algorithm algorithm, int enabled); + grpc_exec_ctx *exec_ctx, grpc_channel_args **a, + grpc_compression_algorithm algorithm, int enabled); /** Returns the bitset representing the support state (true for enabled, false * for disabled) for compression algorithms. diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c index 9da81959e7..c32e7e6277 100644 --- a/src/core/lib/channel/channel_stack.c +++ b/src/core/lib/channel/channel_stack.c @@ -292,7 +292,7 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); memset(op, 0, sizeof(*op)); op->on_complete = grpc_closure_create(destroy_op, op); - grpc_transport_stream_op_add_cancellation_with_message(op, status, + grpc_transport_stream_op_add_cancellation_with_message(exec_ctx, op, status, optional_message); elem->filter->start_transport_stream_op(exec_ctx, elem, op); } @@ -304,6 +304,6 @@ void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); memset(op, 0, sizeof(*op)); op->on_complete = grpc_closure_create(destroy_op, op); - grpc_transport_stream_op_add_close(op, status, optional_message); + grpc_transport_stream_op_add_close(exec_ctx, op, status, optional_message); elem->filter->start_transport_stream_op(exec_ctx, elem, op); } diff --git a/src/core/lib/channel/channel_stack_builder.c b/src/core/lib/channel/channel_stack_builder.c index eda4968f48..366bd0de29 100644 --- a/src/core/lib/channel/channel_stack_builder.c +++ b/src/core/lib/channel/channel_stack_builder.c @@ -138,9 +138,10 @@ void grpc_channel_stack_builder_set_name(grpc_channel_stack_builder *builder, } void grpc_channel_stack_builder_set_channel_arguments( - grpc_channel_stack_builder *builder, const grpc_channel_args *args) { + grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, + const grpc_channel_args *args) { if (builder->args != NULL) { - grpc_channel_args_destroy(builder->args); + grpc_channel_args_destroy(exec_ctx, builder->args); } builder->args = grpc_channel_args_copy(args); } @@ -213,7 +214,8 @@ bool grpc_channel_stack_builder_add_filter_after( return true; } -void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder) { +void grpc_channel_stack_builder_destroy(grpc_exec_ctx *exec_ctx, + grpc_channel_stack_builder *builder) { filter_node *p = builder->begin.next; while (p != &builder->end) { filter_node *next = p->next; @@ -221,7 +223,7 @@ void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder) { p = next; } if (builder->args != NULL) { - grpc_channel_args_destroy(builder->args); + grpc_channel_args_destroy(exec_ctx, builder->args); } gpr_free(builder->target); gpr_free(builder); @@ -270,7 +272,7 @@ void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx, i++; } - grpc_channel_stack_builder_destroy(builder); + grpc_channel_stack_builder_destroy(exec_ctx, builder); gpr_free((grpc_channel_filter **)filters); return result; diff --git a/src/core/lib/channel/channel_stack_builder.h b/src/core/lib/channel/channel_stack_builder.h index 4a00f7bfdb..3532819a0c 100644 --- a/src/core/lib/channel/channel_stack_builder.h +++ b/src/core/lib/channel/channel_stack_builder.h @@ -73,7 +73,8 @@ grpc_transport *grpc_channel_stack_builder_get_transport( /// Set channel arguments: copies args void grpc_channel_stack_builder_set_channel_arguments( - grpc_channel_stack_builder *builder, const grpc_channel_args *args); + grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, + const grpc_channel_args *args); /// Return a borrowed pointer to the channel arguments const grpc_channel_args *grpc_channel_stack_builder_get_channel_arguments( @@ -158,7 +159,8 @@ void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx, void *destroy_arg); /// Destroy the builder without creating a channel stack -void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder); +void grpc_channel_stack_builder_destroy(grpc_exec_ctx *exec_ctx, + grpc_channel_stack_builder *builder); extern int grpc_trace_channel_stack_builder; diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c index de71bcc22b..9cb52627ce 100644 --- a/src/core/lib/channel/compress_filter.c +++ b/src/core/lib/channel/compress_filter.c @@ -44,6 +44,7 @@ #include "src/core/lib/compression/algorithm_metadata.h" #include "src/core/lib/compression/message_compress.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/static_metadata.h" @@ -126,12 +127,14 @@ static int skip_compression(grpc_call_element *elem) { /** Filter initial metadata */ static void process_send_initial_metadata( - grpc_call_element *elem, grpc_metadata_batch *initial_metadata) { + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_metadata_batch *initial_metadata) { call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; /* Parse incoming request for compression. If any, it'll be available * at calld->compression_algorithm */ - grpc_metadata_batch_filter(initial_metadata, compression_md_filter, elem); + grpc_metadata_batch_filter(exec_ctx, initial_metadata, compression_md_filter, + elem); if (!calld->has_compression_algorithm) { /* If no algorithm was found in the metadata and we aren't * exceptionally skipping compression, fall back to the channel @@ -157,7 +160,7 @@ static void continue_send_message(grpc_exec_ctx *exec_ctx, static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { grpc_call_element *elem = elemp; call_data *calld = elem->call_data; - grpc_slice_buffer_reset_and_unref(&calld->slices); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &calld->slices); calld->post_send->cb(exec_ctx, calld->post_send->cb_arg, error); } @@ -167,8 +170,8 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx, int did_compress; grpc_slice_buffer tmp; grpc_slice_buffer_init(&tmp); - did_compress = - grpc_msg_compress(calld->compression_algorithm, &calld->slices, &tmp); + did_compress = grpc_msg_compress(exec_ctx, calld->compression_algorithm, + &calld->slices, &tmp); if (did_compress) { if (grpc_compression_trace) { char *algo_name; @@ -195,7 +198,7 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx, } } - grpc_slice_buffer_destroy(&tmp); + grpc_slice_buffer_destroy_internal(exec_ctx, &tmp); grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices, calld->send_flags); @@ -239,7 +242,7 @@ static void compress_start_transport_stream_op(grpc_exec_ctx *exec_ctx, GPR_TIMER_BEGIN("compress_start_transport_stream_op", 0); if (op->send_initial_metadata) { - process_send_initial_metadata(elem, op->send_initial_metadata); + process_send_initial_metadata(exec_ctx, elem, op->send_initial_metadata); } if (op->send_message != NULL && !skip_compression(elem) && 0 == (op->send_message->flags & GRPC_WRITE_NO_COMPRESS)) { @@ -277,7 +280,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, void *ignored) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; - grpc_slice_buffer_destroy(&calld->slices); + grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices); } /* Constructor for channel_data */ diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c index 449eb7b8d6..3b24e52ff4 100644 --- a/src/core/lib/channel/deadline_filter.c +++ b/src/core/lib/channel/deadline_filter.c @@ -41,6 +41,7 @@ #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/slice/slice_internal.h" // // grpc_deadline_state @@ -58,7 +59,7 @@ static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice msg = grpc_slice_from_static_string("Deadline Exceeded"); grpc_call_element_send_cancel_with_message( exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &msg); - grpc_slice_unref(msg); + grpc_slice_unref_internal(exec_ctx, msg); } GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer"); } diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c index 0714f31bdd..026e4d486e 100644 --- a/src/core/lib/channel/http_client_filter.c +++ b/src/core/lib/channel/http_client_filter.c @@ -36,6 +36,7 @@ #include <grpc/support/string_util.h> #include <string.h> #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/static_metadata.h" #include "src/core/lib/transport/transport_impl.h" @@ -136,8 +137,8 @@ static void hc_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, client_recv_filter_args a; a.elem = elem; a.exec_ctx = exec_ctx; - grpc_metadata_batch_filter(calld->recv_initial_metadata, client_recv_filter, - &a); + grpc_metadata_batch_filter(exec_ctx, calld->recv_initial_metadata, + client_recv_filter, &a); calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, error); } @@ -155,7 +156,7 @@ static void hc_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { grpc_call_element *elem = elemp; call_data *calld = elem->call_data; - grpc_slice_buffer_reset_and_unref(&calld->slices); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &calld->slices); calld->post_send->cb(exec_ctx, calld->post_send->cb_arg, error); } @@ -244,7 +245,7 @@ static void hc_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, /* when all the send_message data is available, then create a MDELEM and append to headers */ grpc_mdelem *payload_bin = grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_GRPC_PAYLOAD_BIN, + exec_ctx, GRPC_MDSTR_GRPC_PAYLOAD_BIN, grpc_mdstr_from_buffer(calld->payload_bytes, op->send_message->length)); grpc_metadata_batch_add_tail(op->send_initial_metadata, @@ -261,8 +262,8 @@ static void hc_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, } } - grpc_metadata_batch_filter(op->send_initial_metadata, client_strip_filter, - elem); + grpc_metadata_batch_filter(exec_ctx, op->send_initial_metadata, + client_strip_filter, elem); /* Send : prefixed headers, which have to be before any application layer headers. */ grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->method, @@ -324,7 +325,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, void *ignored) { call_data *calld = elem->call_data; - grpc_slice_buffer_destroy(&calld->slices); + grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices); } static grpc_mdelem *scheme_from_args(const grpc_channel_args *args) { @@ -425,7 +426,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx, chand->max_payload_size_for_get = max_payload_size_from_args(args->channel_args); chand->user_agent = grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_USER_AGENT, + exec_ctx, GRPC_MDSTR_USER_AGENT, user_agent_from_args(args->channel_args, args->optional_transport->vtable->name)); } @@ -434,7 +435,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx, static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem) { channel_data *chand = elem->channel_data; - GRPC_MDELEM_UNREF(chand->user_agent); + GRPC_MDELEM_UNREF(exec_ctx, chand->user_agent); } const grpc_channel_filter grpc_http_client_filter = { diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c index 10631850cd..d09a2b13ee 100644 --- a/src/core/lib/channel/http_server_filter.c +++ b/src/core/lib/channel/http_server_filter.c @@ -37,6 +37,7 @@ #include <grpc/support/log.h> #include <string.h> #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/transport/static_metadata.h" #define EXPECTED_CONTENT_TYPE "application/grpc" @@ -155,7 +156,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { /* translate host to :authority since :authority may be omitted */ grpc_mdelem *authority = grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_AUTHORITY, GRPC_MDSTR_REF(md->value)); + a->exec_ctx, GRPC_MDSTR_AUTHORITY, GRPC_MDSTR_REF(md->value)); calld->seen_authority = 1; return authority; } else if (md->key == GRPC_MDSTR_GRPC_PAYLOAD_BIN) { @@ -164,7 +165,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { calld->seen_payload_bin = 1; grpc_slice_buffer_init(&calld->read_slice_buffer); grpc_slice_buffer_add(&calld->read_slice_buffer, - grpc_slice_ref(md->value->slice)); + grpc_slice_ref_internal(md->value->slice)); grpc_slice_buffer_stream_init(&calld->read_stream, &calld->read_slice_buffer, 0); return NULL; @@ -181,7 +182,8 @@ static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, server_filter_args a; a.elem = elem; a.exec_ctx = exec_ctx; - grpc_metadata_batch_filter(calld->recv_initial_metadata, server_filter, &a); + grpc_metadata_batch_filter(exec_ctx, calld->recv_initial_metadata, + server_filter, &a); /* Have we seen the required http2 transport headers? (:method, :scheme, content-type, with :path and :authority covered at the channel level right now) */ diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c index 85c307702f..50118b52fd 100644 --- a/src/core/lib/channel/message_size_filter.c +++ b/src/core/lib/channel/message_size_filter.c @@ -66,8 +66,10 @@ static int message_size_limits_cmp(void* value1, void* value2) { return 0; } +static void free_mem(grpc_exec_ctx* exec_ctx, void* p) { gpr_free(p); } + static const grpc_mdstr_hash_table_vtable message_size_limits_vtable = { - gpr_free, message_size_limits_copy, message_size_limits_cmp}; + free_mem, message_size_limits_copy, message_size_limits_cmp}; static void* method_config_convert_value( const grpc_method_config* method_config) { @@ -171,8 +173,8 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, calld->max_send_size = chand->max_send_size; calld->max_recv_size = chand->max_recv_size; if (chand->method_limit_table != NULL) { - message_size_limits* limits = - grpc_method_config_table_get(chand->method_limit_table, args->path); + message_size_limits* limits = grpc_method_config_table_get( + exec_ctx, chand->method_limit_table, args->path); if (limits != NULL) { if (limits->max_send_size >= 0 && (limits->max_send_size < calld->max_send_size || @@ -225,7 +227,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx, if (channel_arg != NULL) { GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER); chand->method_limit_table = grpc_method_config_table_convert( - (grpc_method_config_table*)channel_arg->value.pointer.p, + exec_ctx, (grpc_method_config_table*)channel_arg->value.pointer.p, method_config_convert_value, &message_size_limits_vtable); } } @@ -234,7 +236,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx, static void destroy_channel_elem(grpc_exec_ctx* exec_ctx, grpc_channel_element* elem) { channel_data* chand = elem->channel_data; - grpc_mdstr_hash_table_unref(chand->method_limit_table); + grpc_mdstr_hash_table_unref(exec_ctx, chand->method_limit_table); } const grpc_channel_filter grpc_message_size_filter = { diff --git a/src/core/lib/compression/message_compress.c b/src/core/lib/compression/message_compress.c index 6c245acf61..49beb953b0 100644 --- a/src/core/lib/compression/message_compress.c +++ b/src/core/lib/compression/message_compress.c @@ -40,10 +40,12 @@ #include <zlib.h> +#include "src/core/lib/slice/slice_internal.h" + #define OUTPUT_BLOCK_SIZE 1024 -static int zlib_body(z_stream* zs, grpc_slice_buffer* input, - grpc_slice_buffer* output, +static int zlib_body(grpc_exec_ctx* exec_ctx, z_stream* zs, + grpc_slice_buffer* input, grpc_slice_buffer* output, int (*flate)(z_stream* zs, int flush)) { int r; int flush; @@ -87,7 +89,7 @@ static int zlib_body(z_stream* zs, grpc_slice_buffer* input, return 1; error: - grpc_slice_unref(outbuf); + grpc_slice_unref_internal(exec_ctx, outbuf); return 0; } @@ -97,8 +99,8 @@ static void* zalloc_gpr(void* opaque, unsigned int items, unsigned int size) { static void zfree_gpr(void* opaque, void* address) { gpr_free(address); } -static int zlib_compress(grpc_slice_buffer* input, grpc_slice_buffer* output, - int gzip) { +static int zlib_compress(grpc_exec_ctx* exec_ctx, grpc_slice_buffer* input, + grpc_slice_buffer* output, int gzip) { z_stream zs; int r; size_t i; @@ -110,10 +112,11 @@ static int zlib_compress(grpc_slice_buffer* input, grpc_slice_buffer* output, r = deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 15 | (gzip ? 16 : 0), 8, Z_DEFAULT_STRATEGY); GPR_ASSERT(r == Z_OK); - r = zlib_body(&zs, input, output, deflate) && output->length < input->length; + r = zlib_body(exec_ctx, &zs, input, output, deflate) && + output->length < input->length; if (!r) { for (i = count_before; i < output->count; i++) { - grpc_slice_unref(output->slices[i]); + grpc_slice_unref_internal(exec_ctx, output->slices[i]); } output->count = count_before; output->length = length_before; @@ -122,8 +125,8 @@ static int zlib_compress(grpc_slice_buffer* input, grpc_slice_buffer* output, return r; } -static int zlib_decompress(grpc_slice_buffer* input, grpc_slice_buffer* output, - int gzip) { +static int zlib_decompress(grpc_exec_ctx* exec_ctx, grpc_slice_buffer* input, + grpc_slice_buffer* output, int gzip) { z_stream zs; int r; size_t i; @@ -134,10 +137,10 @@ static int zlib_decompress(grpc_slice_buffer* input, grpc_slice_buffer* output, zs.zfree = zfree_gpr; r = inflateInit2(&zs, 15 | (gzip ? 16 : 0)); GPR_ASSERT(r == Z_OK); - r = zlib_body(&zs, input, output, inflate); + r = zlib_body(exec_ctx, &zs, input, output, inflate); if (!r) { for (i = count_before; i < output->count; i++) { - grpc_slice_unref(output->slices[i]); + grpc_slice_unref_internal(exec_ctx, output->slices[i]); } output->count = count_before; output->length = length_before; @@ -149,12 +152,13 @@ static int zlib_decompress(grpc_slice_buffer* input, grpc_slice_buffer* output, static int copy(grpc_slice_buffer* input, grpc_slice_buffer* output) { size_t i; for (i = 0; i < input->count; i++) { - grpc_slice_buffer_add(output, grpc_slice_ref(input->slices[i])); + grpc_slice_buffer_add(output, grpc_slice_ref_internal(input->slices[i])); } return 1; } -static int compress_inner(grpc_compression_algorithm algorithm, +static int compress_inner(grpc_exec_ctx* exec_ctx, + grpc_compression_algorithm algorithm, grpc_slice_buffer* input, grpc_slice_buffer* output) { switch (algorithm) { case GRPC_COMPRESS_NONE: @@ -162,9 +166,9 @@ static int compress_inner(grpc_compression_algorithm algorithm, rely on that here */ return 0; case GRPC_COMPRESS_DEFLATE: - return zlib_compress(input, output, 0); + return zlib_compress(exec_ctx, input, output, 0); case GRPC_COMPRESS_GZIP: - return zlib_compress(input, output, 1); + return zlib_compress(exec_ctx, input, output, 1); case GRPC_COMPRESS_ALGORITHMS_COUNT: break; } @@ -172,24 +176,26 @@ static int compress_inner(grpc_compression_algorithm algorithm, return 0; } -int grpc_msg_compress(grpc_compression_algorithm algorithm, +int grpc_msg_compress(grpc_exec_ctx* exec_ctx, + grpc_compression_algorithm algorithm, grpc_slice_buffer* input, grpc_slice_buffer* output) { - if (!compress_inner(algorithm, input, output)) { + if (!compress_inner(exec_ctx, algorithm, input, output)) { copy(input, output); return 0; } return 1; } -int grpc_msg_decompress(grpc_compression_algorithm algorithm, +int grpc_msg_decompress(grpc_exec_ctx* exec_ctx, + grpc_compression_algorithm algorithm, grpc_slice_buffer* input, grpc_slice_buffer* output) { switch (algorithm) { case GRPC_COMPRESS_NONE: return copy(input, output); case GRPC_COMPRESS_DEFLATE: - return zlib_decompress(input, output, 0); + return zlib_decompress(exec_ctx, input, output, 0); case GRPC_COMPRESS_GZIP: - return zlib_decompress(input, output, 1); + return zlib_decompress(exec_ctx, input, output, 1); case GRPC_COMPRESS_ALGORITHMS_COUNT: break; } diff --git a/src/core/lib/compression/message_compress.h b/src/core/lib/compression/message_compress.h index 448d36a863..7bd3d98607 100644 --- a/src/core/lib/compression/message_compress.h +++ b/src/core/lib/compression/message_compress.h @@ -40,13 +40,15 @@ /* compress 'input' to 'output' using 'algorithm'. On success, appends compressed slices to output and returns 1. On failure, appends uncompressed slices to output and returns 0. */ -int grpc_msg_compress(grpc_compression_algorithm algorithm, +int grpc_msg_compress(grpc_exec_ctx* exec_ctx, + grpc_compression_algorithm algorithm, grpc_slice_buffer* input, grpc_slice_buffer* output); /* decompress 'input' to 'output' using 'algorithm'. On success, appends slices to output and returns 1. On failure, output is unchanged, and returns 0. */ -int grpc_msg_decompress(grpc_compression_algorithm algorithm, +int grpc_msg_decompress(grpc_exec_ctx* exec_ctx, + grpc_compression_algorithm algorithm, grpc_slice_buffer* input, grpc_slice_buffer* output); #endif /* GRPC_CORE_LIB_COMPRESSION_MESSAGE_COMPRESS_H */ diff --git a/src/core/lib/http/httpcli.c b/src/core/lib/http/httpcli.c index fdb8abaa2d..6bab7ef275 100644 --- a/src/core/lib/http/httpcli.c +++ b/src/core/lib/http/httpcli.c @@ -47,6 +47,7 @@ #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/iomgr/tcp_client.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/string.h" typedef struct { @@ -111,14 +112,14 @@ static void finish(grpc_exec_ctx *exec_ctx, internal_request *req, if (req->ep != NULL) { grpc_endpoint_destroy(exec_ctx, req->ep); } - grpc_slice_unref(req->request_text); + grpc_slice_unref_internal(exec_ctx, req->request_text); gpr_free(req->host); gpr_free(req->ssl_host_override); grpc_iomgr_unregister_object(&req->iomgr_obj); - grpc_slice_buffer_destroy(&req->incoming); - grpc_slice_buffer_destroy(&req->outgoing); + grpc_slice_buffer_destroy_internal(exec_ctx, &req->incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &req->outgoing); GRPC_ERROR_UNREF(req->overall_error); - grpc_resource_quota_internal_unref(exec_ctx, req->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, req->resource_quota); gpr_free(req); } @@ -178,7 +179,7 @@ static void done_write(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { } static void start_write(grpc_exec_ctx *exec_ctx, internal_request *req) { - grpc_slice_ref(req->request_text); + grpc_slice_ref_internal(req->request_text); grpc_slice_buffer_add(&req->outgoing, req->request_text); grpc_endpoint_write(exec_ctx, req->ep, &req->outgoing, &req->done_write); } @@ -265,7 +266,7 @@ static void internal_request_begin(grpc_exec_ctx *exec_ctx, req->context = context; req->pollent = pollent; req->overall_error = GRPC_ERROR_NONE; - req->resource_quota = grpc_resource_quota_internal_ref(resource_quota); + req->resource_quota = grpc_resource_quota_ref_internal(resource_quota); grpc_closure_init(&req->on_read, on_read, req); grpc_closure_init(&req->done_write, done_write, req); grpc_slice_buffer_init(&req->incoming); diff --git a/src/core/lib/iomgr/resource_quota.c b/src/core/lib/iomgr/resource_quota.c index ddc7a88c5b..eac7c52c51 100644 --- a/src/core/lib/iomgr/resource_quota.c +++ b/src/core/lib/iomgr/resource_quota.c @@ -169,14 +169,14 @@ static void rq_step(grpc_exec_ctx *exec_ctx, void *rq, grpc_error *error) { rq_reclaim(exec_ctx, resource_quota, false) || rq_reclaim(exec_ctx, resource_quota, true); done: - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); } static void rq_step_sched(grpc_exec_ctx *exec_ctx, grpc_resource_quota *resource_quota) { if (resource_quota->step_scheduled) return; resource_quota->step_scheduled = true; - grpc_resource_quota_internal_ref(resource_quota); + grpc_resource_quota_ref_internal(resource_quota); grpc_combiner_execute_finally(exec_ctx, resource_quota->combiner, &resource_quota->rq_step_closure, GRPC_ERROR_NONE, false); @@ -257,7 +257,7 @@ static bool rq_reclaim(grpc_exec_ctx *exec_ctx, destructive ? "destructive" : "benign"); } resource_quota->reclaiming = true; - grpc_resource_quota_internal_ref(resource_quota); + grpc_resource_quota_ref_internal(resource_quota); grpc_closure *c = resource_user->reclaimers[destructive]; resource_user->reclaimers[destructive] = NULL; grpc_closure_run(exec_ctx, c, GRPC_ERROR_NONE); @@ -280,21 +280,10 @@ static void ru_slice_ref(void *p) { gpr_ref(&rc->refs); } -static void ru_slice_unref(void *p) { +static void ru_slice_unref(grpc_exec_ctx *exec_ctx, void *p) { ru_slice_refcount *rc = p; if (gpr_unref(&rc->refs)) { - /* TODO(ctiller): this is dangerous, but I think safe for now: - we have no guarantee here that we're at a safe point for creating an - execution context, but we have no way of writing this code otherwise. - In the future: consider lifting grpc_slice to grpc, and offering an - internal_{ref,unref} pair that is execution context aware. - Alternatively, - make exec_ctx be thread local and 'do the right thing' (whatever that - is) - if NULL */ - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resource_user_free(&exec_ctx, rc->resource_user, rc->size); - grpc_exec_ctx_finish(&exec_ctx); + grpc_resource_user_free(exec_ctx, rc->resource_user, rc->size); gpr_free(rc); } } @@ -419,7 +408,7 @@ static void rq_resize(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { a->resource_quota->size += delta; a->resource_quota->free_pool += delta; rq_step_sched(exec_ctx, a->resource_quota); - grpc_resource_quota_internal_unref(exec_ctx, a->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, a->resource_quota); gpr_free(a); } @@ -428,7 +417,7 @@ static void rq_reclamation_done(grpc_exec_ctx *exec_ctx, void *rq, grpc_resource_quota *resource_quota = rq; resource_quota->reclaiming = false; rq_step_sched(exec_ctx, resource_quota); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); } /******************************************************************************* @@ -459,7 +448,7 @@ grpc_resource_quota *grpc_resource_quota_create(const char *name) { return resource_quota; } -void grpc_resource_quota_internal_unref(grpc_exec_ctx *exec_ctx, +void grpc_resource_quota_unref_internal(grpc_exec_ctx *exec_ctx, grpc_resource_quota *resource_quota) { if (gpr_unref(&resource_quota->refs)) { grpc_combiner_destroy(exec_ctx, resource_quota->combiner); @@ -471,11 +460,11 @@ void grpc_resource_quota_internal_unref(grpc_exec_ctx *exec_ctx, /* Public API */ void grpc_resource_quota_unref(grpc_resource_quota *resource_quota) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_exec_ctx_finish(&exec_ctx); } -grpc_resource_quota *grpc_resource_quota_internal_ref( +grpc_resource_quota *grpc_resource_quota_ref_internal( grpc_resource_quota *resource_quota) { gpr_ref(&resource_quota->refs); return resource_quota; @@ -483,7 +472,7 @@ grpc_resource_quota *grpc_resource_quota_internal_ref( /* Public API */ void grpc_resource_quota_ref(grpc_resource_quota *resource_quota) { - grpc_resource_quota_internal_ref(resource_quota); + grpc_resource_quota_ref_internal(resource_quota); } /* Public API */ @@ -491,7 +480,7 @@ void grpc_resource_quota_resize(grpc_resource_quota *resource_quota, size_t size) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; rq_resize_args *a = gpr_malloc(sizeof(*a)); - a->resource_quota = grpc_resource_quota_internal_ref(resource_quota); + a->resource_quota = grpc_resource_quota_ref_internal(resource_quota); a->size = (int64_t)size; grpc_closure_init(&a->closure, rq_resize, a); grpc_combiner_execute(&exec_ctx, resource_quota->combiner, &a->closure, @@ -508,7 +497,7 @@ grpc_resource_quota *grpc_resource_quota_from_channel_args( for (size_t i = 0; i < channel_args->num_args; i++) { if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) { if (channel_args->args[i].type == GRPC_ARG_POINTER) { - return grpc_resource_quota_internal_ref( + return grpc_resource_quota_ref_internal( channel_args->args[i].value.pointer.p); } else { gpr_log(GPR_DEBUG, GRPC_ARG_RESOURCE_QUOTA " should be a pointer"); @@ -523,7 +512,9 @@ static void *rq_copy(void *rq) { return rq; } -static void rq_destroy(void *rq) { grpc_resource_quota_unref(rq); } +static void rq_destroy(grpc_exec_ctx *exec_ctx, void *rq) { + grpc_resource_quota_unref_internal(exec_ctx, rq); +} static int rq_cmp(void *a, void *b) { return GPR_ICMP(a, b); } @@ -540,7 +531,7 @@ void grpc_resource_user_init(grpc_resource_user *resource_user, grpc_resource_quota *resource_quota, const char *name) { resource_user->resource_quota = - grpc_resource_quota_internal_ref(resource_quota); + grpc_resource_quota_ref_internal(resource_quota); grpc_closure_init(&resource_user->allocate_closure, &ru_allocate, resource_user); grpc_closure_init(&resource_user->add_to_free_pool_closure, @@ -589,7 +580,7 @@ void grpc_resource_user_shutdown(grpc_exec_ctx *exec_ctx, void grpc_resource_user_destroy(grpc_exec_ctx *exec_ctx, grpc_resource_user *resource_user) { - grpc_resource_quota_internal_unref(exec_ctx, resource_user->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_user->resource_quota); gpr_mu_destroy(&resource_user->mu); gpr_free(resource_user->name); } diff --git a/src/core/lib/iomgr/resource_quota.h b/src/core/lib/iomgr/resource_quota.h index f7e5ca6494..f1da73933e 100644 --- a/src/core/lib/iomgr/resource_quota.h +++ b/src/core/lib/iomgr/resource_quota.h @@ -77,9 +77,9 @@ extern int grpc_resource_quota_trace; -grpc_resource_quota *grpc_resource_quota_internal_ref( +grpc_resource_quota *grpc_resource_quota_ref_internal( grpc_resource_quota *resource_quota); -void grpc_resource_quota_internal_unref(grpc_exec_ctx *exec_ctx, +void grpc_resource_quota_unref_internal(grpc_exec_ctx *exec_ctx, grpc_resource_quota *resource_quota); grpc_resource_quota *grpc_resource_quota_from_channel_args( const grpc_channel_args *channel_args); diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c index bc08c94ee0..3b0fe3bc15 100644 --- a/src/core/lib/iomgr/tcp_client_posix.c +++ b/src/core/lib/iomgr/tcp_client_posix.c @@ -116,7 +116,7 @@ static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { if (done) { gpr_mu_destroy(&ac->mu); gpr_free(ac->addr_str); - grpc_channel_args_destroy(ac->channel_args); + grpc_channel_args_destroy(exec_ctx, ac->channel_args); gpr_free(ac); } } @@ -136,8 +136,8 @@ grpc_endpoint *grpc_tcp_client_create_from_fd( &channel_args->args[i], options); } else if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) { - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); - resource_quota = grpc_resource_quota_internal_ref( + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); + resource_quota = grpc_resource_quota_ref_internal( channel_args->args[i].value.pointer.p); } } @@ -145,7 +145,7 @@ grpc_endpoint *grpc_tcp_client_create_from_fd( grpc_endpoint *ep = grpc_tcp_create(fd, resource_quota, tcp_read_chunk_size, addr_str); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); return ep; } @@ -247,7 +247,7 @@ finish: if (done) { gpr_mu_destroy(&ac->mu); gpr_free(ac->addr_str); - grpc_channel_args_destroy(ac->channel_args); + grpc_channel_args_destroy(exec_ctx, ac->channel_args); gpr_free(ac); } grpc_exec_ctx_sched(exec_ctx, closure, error, NULL); diff --git a/src/core/lib/iomgr/tcp_client_windows.c b/src/core/lib/iomgr/tcp_client_windows.c index 4d1e809872..4ad417f77d 100644 --- a/src/core/lib/iomgr/tcp_client_windows.c +++ b/src/core/lib/iomgr/tcp_client_windows.c @@ -71,7 +71,7 @@ static void async_connect_unlock_and_cleanup(grpc_exec_ctx *exec_ctx, int done = (--ac->refs == 0); gpr_mu_unlock(&ac->mu); if (done) { - grpc_resource_quota_internal_unref(exec_ctx, ac->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, ac->resource_quota); gpr_mu_destroy(&ac->mu); gpr_free(ac->addr_name); gpr_free(ac); @@ -153,8 +153,8 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, if (channel_args != NULL) { for (size_t i = 0; i < channel_args->num_args; i++) { if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) { - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); - resource_quota = grpc_resource_quota_internal_ref( + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); + resource_quota = grpc_resource_quota_ref_internal( channel_args->args[i].value.pointer.p); } } @@ -242,7 +242,7 @@ failure: } else if (sock != INVALID_SOCKET) { closesocket(sock); } - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); grpc_exec_ctx_sched(exec_ctx, on_done, final_error, NULL); } diff --git a/src/core/lib/iomgr/tcp_posix.c b/src/core/lib/iomgr/tcp_posix.c index 584fc2fe2e..4bf13bee27 100644 --- a/src/core/lib/iomgr/tcp_posix.c +++ b/src/core/lib/iomgr/tcp_posix.c @@ -56,6 +56,7 @@ #include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/ev_posix.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" @@ -131,7 +132,7 @@ static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { static void tcp_free(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) { grpc_fd_orphan(exec_ctx, tcp->em_fd, tcp->release_fd_cb, tcp->release_fd, "tcp_unref_orphan"); - grpc_slice_buffer_destroy(&tcp->last_read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &tcp->last_read_buffer); grpc_resource_user_destroy(exec_ctx, &tcp->resource_user); gpr_free(tcp->peer_string); gpr_free(tcp); @@ -178,7 +179,7 @@ static void tcp_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) { grpc_network_status_unregister_endpoint(ep); grpc_tcp *tcp = (grpc_tcp *)ep; tcp_maybe_shutdown_resource_user(exec_ctx, tcp); - grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &tcp->last_read_buffer); TCP_UNREF(exec_ctx, tcp, "destroy"); } @@ -245,13 +246,14 @@ static void tcp_do_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) { /* We've consumed the edge, request a new one */ grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure); } else { - grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, + tcp->incoming_buffer); call_read_cb(exec_ctx, tcp, GRPC_OS_ERROR(errno, "recvmsg")); TCP_UNREF(exec_ctx, tcp, "read"); } } else if (read_bytes == 0) { /* 0 read size ==> end of stream */ - grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, tcp->incoming_buffer); call_read_cb(exec_ctx, tcp, GRPC_ERROR_CREATE("Socket closed")); TCP_UNREF(exec_ctx, tcp, "read"); } else { @@ -276,8 +278,9 @@ static void tcp_read_allocation_done(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) { grpc_tcp *tcp = tcpp; if (error != GRPC_ERROR_NONE) { - grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer); - grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, tcp->incoming_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, + &tcp->last_read_buffer); call_read_cb(exec_ctx, tcp, GRPC_ERROR_REF(error)); TCP_UNREF(exec_ctx, tcp, "read"); } else { @@ -302,8 +305,9 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */, GPR_ASSERT(!tcp->finished_edge); if (error != GRPC_ERROR_NONE) { - grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer); - grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, tcp->incoming_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, + &tcp->last_read_buffer); call_read_cb(exec_ctx, tcp, GRPC_ERROR_REF(error)); TCP_UNREF(exec_ctx, tcp, "read"); } else { @@ -317,7 +321,7 @@ static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, GPR_ASSERT(tcp->read_cb == NULL); tcp->read_cb = cb; tcp->incoming_buffer = incoming_buffer; - grpc_slice_buffer_reset_and_unref(incoming_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, incoming_buffer); grpc_slice_buffer_swap(incoming_buffer, &tcp->last_read_buffer); TCP_REF(tcp, "read"); if (tcp->finished_edge) { @@ -578,7 +582,7 @@ void grpc_tcp_destroy_and_release_fd(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, tcp->release_fd = fd; tcp->release_fd_cb = done; tcp_maybe_shutdown_resource_user(exec_ctx, tcp); - grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &tcp->last_read_buffer); TCP_UNREF(exec_ctx, tcp, "destroy"); } diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index b6fc1e4ca2..1a753d1231 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -167,18 +167,18 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, s->so_reuseport = has_so_reuseport && (args->args[i].value.integer != 0); } else { - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); gpr_free(s); return GRPC_ERROR_CREATE(GRPC_ARG_ALLOW_REUSEPORT " must be an integer"); } } else if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) { if (args->args[i].type == GRPC_ARG_POINTER) { - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); s->resource_quota = - grpc_resource_quota_internal_ref(args->args[i].value.pointer.p); + grpc_resource_quota_ref_internal(args->args[i].value.pointer.p); } else { - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); gpr_free(s); return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA " must be a pointer to a buffer pool"); @@ -219,7 +219,7 @@ static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { gpr_free(sp); } - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); gpr_free(s); } diff --git a/src/core/lib/iomgr/tcp_server_windows.c b/src/core/lib/iomgr/tcp_server_windows.c index ae54c70d2d..c2a6d1736e 100644 --- a/src/core/lib/iomgr/tcp_server_windows.c +++ b/src/core/lib/iomgr/tcp_server_windows.c @@ -115,11 +115,11 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, for (size_t i = 0; i < (args == NULL ? 0 : args->num_args); i++) { if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) { if (args->args[i].type == GRPC_ARG_POINTER) { - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); s->resource_quota = - grpc_resource_quota_internal_ref(args->args[i].value.pointer.p); + grpc_resource_quota_ref_internal(args->args[i].value.pointer.p); } else { - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); gpr_free(s); return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA " must be a pointer to a buffer pool"); @@ -155,7 +155,7 @@ static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { grpc_winsocket_destroy(sp->socket); gpr_free(sp); } - grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); gpr_free(s); } diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c index f825057c0e..a97b1b21fe 100644 --- a/src/core/lib/iomgr/tcp_windows.c +++ b/src/core/lib/iomgr/tcp_windows.c @@ -190,13 +190,13 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) { char *utf8_message = gpr_format_message(info->wsa_error); error = GRPC_ERROR_CREATE(utf8_message); gpr_free(utf8_message); - grpc_slice_unref(tcp->read_slice); + grpc_slice_unref_internal(exec_ctx, tcp->read_slice); } else { if (info->bytes_transfered != 0 && !tcp->shutting_down) { sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); grpc_slice_buffer_add(tcp->read_slices, sub); } else { - grpc_slice_unref(tcp->read_slice); + grpc_slice_unref_internal(exec_ctx, tcp->read_slice); error = GRPC_ERROR_CREATE("End of TCP stream"); } } @@ -225,7 +225,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, tcp->read_cb = cb; tcp->read_slices = read_slices; - grpc_slice_buffer_reset_and_unref(read_slices); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, read_slices); tcp->read_slice = grpc_slice_malloc(8192); diff --git a/src/core/lib/security/credentials/credentials_metadata.c b/src/core/lib/security/credentials/credentials_metadata.c index e6cb567734..84e2b8991a 100644 --- a/src/core/lib/security/credentials/credentials_metadata.c +++ b/src/core/lib/security/credentials/credentials_metadata.c @@ -62,8 +62,8 @@ void grpc_credentials_md_store_add(grpc_credentials_md_store *store, grpc_slice key, grpc_slice value) { if (store == NULL) return; store_ensure_capacity(store); - store->entries[store->num_entries].key = grpc_slice_ref(key); - store->entries[store->num_entries].value = grpc_slice_ref(value); + store->entries[store->num_entries].key = grpc_slice_ref_internal(key); + store->entries[store->num_entries].value = grpc_slice_ref_internal(value); store->num_entries++; } @@ -91,8 +91,8 @@ void grpc_credentials_md_store_unref(grpc_credentials_md_store *store) { if (store->entries != NULL) { size_t i; for (i = 0; i < store->num_entries; i++) { - grpc_slice_unref(store->entries[i].key); - grpc_slice_unref(store->entries[i].value); + grpc_slice_unref_internal(exec_ctx, store->entries[i].key); + grpc_slice_unref_internal(exec_ctx, store->entries[i].value); } gpr_free(store->entries); } diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.c b/src/core/lib/security/credentials/google_default/google_default_credentials.c index afe0e3d357..5df97e1671 100644 --- a/src/core/lib/security/credentials/google_default/google_default_credentials.c +++ b/src/core/lib/security/credentials/google_default/google_default_credentials.c @@ -132,7 +132,7 @@ static int is_stack_running_on_compute_engine(void) { gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), max_detection_delay), grpc_closure_create(on_compute_engine_detection_http_response, &detector), &detector.response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_exec_ctx_flush(&exec_ctx); @@ -225,7 +225,7 @@ static grpc_error *create_default_creds_from_path( end: GPR_ASSERT((result == NULL) + (error == GRPC_ERROR_NONE) == 1); if (creds_path != NULL) gpr_free(creds_path); - grpc_slice_unref(creds_data); + grpc_slice_unref_internal(exec_ctx, creds_data); if (json != NULL) grpc_json_destroy(json); *creds = result; return error; diff --git a/src/core/lib/security/credentials/jwt/jwt_verifier.c b/src/core/lib/security/credentials/jwt/jwt_verifier.c index 42bd89dd0a..d551a7c51a 100644 --- a/src/core/lib/security/credentials/jwt/jwt_verifier.c +++ b/src/core/lib/security/credentials/jwt/jwt_verifier.c @@ -96,7 +96,7 @@ static grpc_json *parse_json_part_from_jwt(const char *str, size_t len, json = grpc_json_parse_string_with_len((char *)GRPC_SLICE_START_PTR(*buffer), GRPC_SLICE_LENGTH(*buffer)); if (json == NULL) { - grpc_slice_unref(*buffer); + grpc_slice_unref_internal(exec_ctx, *buffer); gpr_log(GPR_ERROR, "JSON parsing error."); } return json; @@ -133,7 +133,7 @@ typedef struct { } jose_header; static void jose_header_destroy(jose_header *h) { - grpc_slice_unref(h->buffer); + grpc_slice_unref_internal(exec_ctx, h->buffer); gpr_free(h); } @@ -195,7 +195,7 @@ struct grpc_jwt_claims { void grpc_jwt_claims_destroy(grpc_jwt_claims *claims) { grpc_json_destroy(claims->json); - grpc_slice_unref(claims->buffer); + grpc_slice_unref_internal(exec_ctx, claims->buffer); gpr_free(claims); } @@ -365,8 +365,8 @@ static verifier_cb_ctx *verifier_cb_ctx_create( void verifier_cb_ctx_destroy(verifier_cb_ctx *ctx) { if (ctx->audience != NULL) gpr_free(ctx->audience); if (ctx->claims != NULL) grpc_jwt_claims_destroy(ctx->claims); - grpc_slice_unref(ctx->signature); - grpc_slice_unref(ctx->signed_data); + grpc_slice_unref_internal(exec_ctx, ctx->signature); + grpc_slice_unref_internal(exec_ctx, ctx->signed_data); jose_header_destroy(ctx->header); for (size_t i = 0; i < HTTP_RESPONSE_COUNT; i++) { grpc_http_response_destroy(&ctx->responses[i]); @@ -459,7 +459,7 @@ static BIGNUM *bignum_from_base64(const char *b64) { } result = BN_bin2bn(GRPC_SLICE_START_PTR(bin), TSI_SIZE_AS_SIZE(GRPC_SLICE_LENGTH(bin)), NULL); - grpc_slice_unref(bin); + grpc_slice_unref_internal(exec_ctx, bin); return result; } @@ -667,7 +667,7 @@ static void on_openid_config_retrieved(grpc_exec_ctx *exec_ctx, void *user_data, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), grpc_closure_create(on_keys_retrieved, ctx), &ctx->responses[HTTP_RESPONSE_KEYS]); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); grpc_json_destroy(json); gpr_free(req.host); return; @@ -779,7 +779,7 @@ static void retrieve_key_and_verify(grpc_exec_ctx *exec_ctx, exec_ctx, &ctx->verifier->http_ctx, &ctx->pollent, resource_quota, &req, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), http_cb, &ctx->responses[rsp_idx]); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); gpr_free(req.host); gpr_free(req.http.path); return; diff --git a/src/core/lib/security/credentials/oauth2/oauth2_credentials.c b/src/core/lib/security/credentials/oauth2/oauth2_credentials.c index d980577c46..09140bef57 100644 --- a/src/core/lib/security/credentials/oauth2/oauth2_credentials.c +++ b/src/core/lib/security/credentials/oauth2/oauth2_credentials.c @@ -315,7 +315,7 @@ static void compute_engine_fetch_oauth2( grpc_httpcli_get(exec_ctx, httpcli_context, pollent, resource_quota, &request, deadline, grpc_closure_create(response_cb, metadata_req), &metadata_req->response); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); } grpc_call_credentials *grpc_google_compute_engine_credentials_create( @@ -372,7 +372,7 @@ static void refresh_token_fetch_oauth2( &request, body, strlen(body), deadline, grpc_closure_create(response_cb, metadata_req), &metadata_req->response); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); gpr_free(body); } diff --git a/src/core/lib/security/credentials/plugin/plugin_credentials.c b/src/core/lib/security/credentials/plugin/plugin_credentials.c index 61c10862da..16cbb17f46 100644 --- a/src/core/lib/security/credentials/plugin/plugin_credentials.c +++ b/src/core/lib/security/credentials/plugin/plugin_credentials.c @@ -100,8 +100,8 @@ static void plugin_md_request_metadata_ready(void *request, r->cb(&exec_ctx, r->user_data, md_array, num_md, GRPC_CREDENTIALS_OK, NULL); for (i = 0; i < num_md; i++) { - grpc_slice_unref(md_array[i].key); - grpc_slice_unref(md_array[i].value); + grpc_slice_unref_internal(exec_ctx, md_array[i].key); + grpc_slice_unref_internal(exec_ctx, md_array[i].value); } gpr_free(md_array); } diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c index cd4769ea10..22ca99eff8 100644 --- a/src/core/lib/security/transport/client_auth_filter.c +++ b/src/core/lib/security/transport/client_auth_filter.c @@ -121,8 +121,8 @@ static void on_credentials_metadata(grpc_exec_ctx *exec_ctx, void *user_data, for (i = 0; i < num_md; i++) { grpc_metadata_batch_add_tail( mdb, &calld->md_links[i], - grpc_mdelem_from_slices(grpc_slice_ref(md_elems[i].key), - grpc_slice_ref(md_elems[i].value))); + grpc_mdelem_from_slices(grpc_slice_ref_internal(md_elems[i].key), + grpc_slice_ref_internal(md_elems[i].value))); } grpc_call_next_op(exec_ctx, elem, op); } diff --git a/src/core/lib/security/transport/handshake.c b/src/core/lib/security/transport/handshake.c index 01e7fab773..077c1f0aa7 100644 --- a/src/core/lib/security/transport/handshake.c +++ b/src/core/lib/security/transport/handshake.c @@ -104,9 +104,9 @@ static void unref_handshake(grpc_security_handshake *h) { if (gpr_unref(&h->refs)) { if (h->handshaker != NULL) tsi_handshaker_destroy(h->handshaker); if (h->handshake_buffer != NULL) gpr_free(h->handshake_buffer); - grpc_slice_buffer_destroy(&h->left_overs); - grpc_slice_buffer_destroy(&h->outgoing); - grpc_slice_buffer_destroy(&h->incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &h->left_overs); + grpc_slice_buffer_destroy_internal(exec_ctx, &h->outgoing); + grpc_slice_buffer_destroy_internal(exec_ctx, &h->incoming); GRPC_AUTH_CONTEXT_UNREF(h->auth_context, "handshake"); GRPC_SECURITY_CONNECTOR_UNREF(h->connector, "handshake"); gpr_free(h); @@ -213,7 +213,7 @@ static void send_handshake_bytes_to_peer(grpc_exec_ctx *exec_ctx, to_send = grpc_slice_from_copied_buffer((const char *)h->handshake_buffer, offset); - grpc_slice_buffer_reset_and_unref(&h->outgoing); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &h->outgoing); grpc_slice_buffer_add(&h->outgoing, to_send); /* TODO(klempner,jboeuf): This should probably use the client setup deadline */ @@ -280,7 +280,7 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx, grpc_slice_buffer_add( &h->left_overs, grpc_slice_split_tail(&h->incoming.slices[i], consumed_slice_size)); - grpc_slice_unref( + grpc_slice_unref_internal(exec_ctx, h->incoming.slices[i]); /* split_tail above increments refcount. */ } grpc_slice_buffer_addn( diff --git a/src/core/lib/security/transport/secure_endpoint.c b/src/core/lib/security/transport/secure_endpoint.c index fba3314812..78037f8089 100644 --- a/src/core/lib/security/transport/secure_endpoint.c +++ b/src/core/lib/security/transport/secure_endpoint.c @@ -74,11 +74,11 @@ static void destroy(grpc_exec_ctx *exec_ctx, secure_endpoint *secure_ep) { secure_endpoint *ep = secure_ep; grpc_endpoint_destroy(exec_ctx, ep->wrapped_ep); tsi_frame_protector_destroy(ep->protector); - grpc_slice_buffer_destroy(&ep->leftover_bytes); - grpc_slice_unref(ep->read_staging_buffer); - grpc_slice_unref(ep->write_staging_buffer); - grpc_slice_buffer_destroy(&ep->output_buffer); - grpc_slice_buffer_destroy(&ep->source_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &ep->leftover_bytes); + grpc_slice_unref_internal(exec_ctx, ep->read_staging_buffer); + grpc_slice_unref_internal(exec_ctx, ep->write_staging_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &ep->output_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &ep->source_buffer); gpr_mu_destroy(&ep->protector_mu); gpr_free(ep); } @@ -154,7 +154,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, uint8_t *end = GRPC_SLICE_END_PTR(ep->read_staging_buffer); if (error != GRPC_ERROR_NONE) { - grpc_slice_buffer_reset_and_unref(ep->read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, ep->read_buffer); call_read_cb(exec_ctx, ep, GRPC_ERROR_CREATE_REFERENCING( "Secure read failed", &error, 1)); return; @@ -209,10 +209,10 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, /* TODO(yangg) experiment with moving this block after read_cb to see if it helps latency */ - grpc_slice_buffer_reset_and_unref(&ep->source_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &ep->source_buffer); if (result != TSI_OK) { - grpc_slice_buffer_reset_and_unref(ep->read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, ep->read_buffer); call_read_cb(exec_ctx, ep, grpc_set_tsi_error_result( GRPC_ERROR_CREATE("Unwrap failed"), result)); return; @@ -226,7 +226,7 @@ static void endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, secure_endpoint *ep = (secure_endpoint *)secure_ep; ep->read_cb = cb; ep->read_buffer = slices; - grpc_slice_buffer_reset_and_unref(ep->read_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, ep->read_buffer); SECURE_ENDPOINT_REF(ep, "read"); if (ep->leftover_bytes.count) { @@ -258,7 +258,7 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, uint8_t *cur = GRPC_SLICE_START_PTR(ep->write_staging_buffer); uint8_t *end = GRPC_SLICE_END_PTR(ep->write_staging_buffer); - grpc_slice_buffer_reset_and_unref(&ep->output_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &ep->output_buffer); if (grpc_trace_secure_endpoint) { for (i = 0; i < slices->count; i++) { @@ -322,7 +322,7 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, if (result != TSI_OK) { /* TODO(yangg) do different things according to the error type? */ - grpc_slice_buffer_reset_and_unref(&ep->output_buffer); + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &ep->output_buffer); grpc_exec_ctx_sched( exec_ctx, cb, grpc_set_tsi_error_result(GRPC_ERROR_CREATE("Wrap failed"), result), @@ -398,7 +398,7 @@ grpc_endpoint *grpc_secure_endpoint_create( grpc_slice_buffer_init(&ep->leftover_bytes); for (i = 0; i < leftover_nslices; i++) { grpc_slice_buffer_add(&ep->leftover_bytes, - grpc_slice_ref(leftover_slices[i])); + grpc_slice_ref_internal(leftover_slices[i])); } ep->write_staging_buffer = grpc_slice_malloc(STAGING_BUFFER_SIZE); ep->read_staging_buffer = grpc_slice_malloc(STAGING_BUFFER_SIZE); diff --git a/src/core/lib/security/util/b64.c b/src/core/lib/security/util/b64.c index 4892e8e621..c227889726 100644 --- a/src/core/lib/security/util/b64.c +++ b/src/core/lib/security/util/b64.c @@ -228,6 +228,6 @@ grpc_slice grpc_base64_decode_with_len(const char *b64, size_t b64_len, return result; fail: - grpc_slice_unref(result); + grpc_slice_unref_internal(exec_ctx, result); return gpr_empty_slice(); } diff --git a/src/core/lib/slice/percent_encoding.c b/src/core/lib/slice/percent_encoding.c index b9e35f1c71..c76c58d371 100644 --- a/src/core/lib/slice/percent_encoding.c +++ b/src/core/lib/slice/percent_encoding.c @@ -35,6 +35,8 @@ #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" + const uint8_t grpc_url_percent_encoding_unreserved_bytes[256 / 8] = { 0x00, 0x00, 0x00, 0x00, 0x00, 0x60, 0xff, 0x03, 0xfe, 0xff, 0xff, 0x87, 0xfe, 0xff, 0xff, 0x47, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, @@ -66,7 +68,7 @@ grpc_slice grpc_percent_encode_slice(grpc_slice slice, } // no unreserved bytes: return the string unmodified if (!any_reserved_bytes) { - return grpc_slice_ref(slice); + return grpc_slice_ref_internal(slice); } // second pass: actually encode grpc_slice out = grpc_slice_malloc(output_length); @@ -119,7 +121,7 @@ bool grpc_strict_percent_decode_slice(grpc_slice slice_in, } } if (!any_percent_encoded_stuff) { - *slice_out = grpc_slice_ref(slice_in); + *slice_out = grpc_slice_ref_internal(slice_in); return true; } p = GRPC_SLICE_START_PTR(slice_in); @@ -158,7 +160,7 @@ grpc_slice grpc_permissive_percent_decode_slice(grpc_slice slice_in) { } } if (!any_percent_encoded_stuff) { - return grpc_slice_ref(slice_in); + return grpc_slice_ref_internal(slice_in); } p = GRPC_SLICE_START_PTR(slice_in); grpc_slice out = grpc_slice_malloc(out_length); diff --git a/src/core/lib/slice/slice.c b/src/core/lib/slice/slice.c index 3dac18df61..5b8f71a778 100644 --- a/src/core/lib/slice/slice.c +++ b/src/core/lib/slice/slice.c @@ -31,12 +31,16 @@ * */ +#include "src/core/lib/slice/slice_internal.h" + #include <grpc/slice.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <string.h> +#include "src/core/lib/iomgr/exec_ctx.h" + grpc_slice gpr_empty_slice(void) { grpc_slice out; out.refcount = 0; @@ -44,25 +48,37 @@ grpc_slice gpr_empty_slice(void) { return out; } -grpc_slice grpc_slice_ref(grpc_slice slice) { +grpc_slice grpc_slice_ref_internal(grpc_slice slice) { if (slice.refcount) { slice.refcount->ref(slice.refcount); } return slice; } -void grpc_slice_unref(grpc_slice slice) { +void grpc_slice_unref_internal(grpc_exec_ctx *exec_ctx, grpc_slice slice) { if (slice.refcount) { - slice.refcount->unref(slice.refcount); + slice.refcount->unref(exec_ctx, slice.refcount); } } +/* Public API */ +grpc_slice grpc_slice_ref(grpc_slice slice) { + return grpc_slice_ref_internal(slice); +} + +/* Public API */ +void grpc_slice_unref(grpc_slice slice) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_slice_unref_internal(&exec_ctx, slice); + grpc_exec_ctx_finish(&exec_ctx); +} + /* grpc_slice_from_static_string support structure - a refcount that does nothing */ -static void noop_ref_or_unref(void *unused) {} +static void noop_ref(void *unused) {} +static void noop_unref(grpc_exec_ctx *exec_ctx, void *unused) {} -static grpc_slice_refcount noop_refcount = {noop_ref_or_unref, - noop_ref_or_unref}; +static grpc_slice_refcount noop_refcount = {noop_ref, noop_unref}; grpc_slice grpc_slice_from_static_string(const char *s) { grpc_slice slice; @@ -86,7 +102,7 @@ static void new_slice_ref(void *p) { gpr_ref(&r->refs); } -static void new_slice_unref(void *p) { +static void new_slice_unref(grpc_exec_ctx *exec_ctx, void *p) { new_slice_refcount *r = p; if (gpr_unref(&r->refs)) { r->user_destroy(r->user_data); @@ -131,7 +147,7 @@ static void new_with_len_ref(void *p) { gpr_ref(&r->refs); } -static void new_with_len_unref(void *p) { +static void new_with_len_unref(grpc_exec_ctx *exec_ctx, void *p) { new_with_len_slice_refcount *r = p; if (gpr_unref(&r->refs)) { r->user_destroy(r->user_data, r->user_length); @@ -177,7 +193,7 @@ static void malloc_ref(void *p) { gpr_ref(&r->refs); } -static void malloc_unref(void *p) { +static void malloc_unref(grpc_exec_ctx *exec_ctx, void *p) { malloc_refcount *r = p; if (gpr_unref(&r->refs)) { gpr_free(r); diff --git a/src/core/lib/slice/slice_buffer.c b/src/core/lib/slice/slice_buffer.c index 990ef128bd..872bd10a09 100644 --- a/src/core/lib/slice/slice_buffer.c +++ b/src/core/lib/slice/slice_buffer.c @@ -40,6 +40,8 @@ #include <grpc/support/log.h> #include <grpc/support/useful.h> +#include "src/core/lib/slice/slice_internal.h" + /* grow a buffer; requires GRPC_SLICE_BUFFER_INLINE_ELEMENTS > 1 */ #define GROW(x) (3 * (x) / 2) @@ -63,11 +65,21 @@ void grpc_slice_buffer_init(grpc_slice_buffer *sb) { sb->slices = sb->inlined; } +void grpc_slice_buffer_destroy_internal(grpc_exec_ctx *exec_ctx, + grpc_slice_buffer *sb) { + grpc_slice_buffer_reset_and_unref_internal(exec_ctx, sb); + if (sb->slices != sb->inlined) { + gpr_free(sb->slices); + } +} + void grpc_slice_buffer_destroy(grpc_slice_buffer *sb) { - grpc_slice_buffer_reset_and_unref(sb); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_slice_buffer_reset_and_unref_internal(&exec_ctx, sb); if (sb->slices != sb->inlined) { gpr_free(sb->slices); } + grpc_exec_ctx_finish(&exec_ctx); } uint8_t *grpc_slice_buffer_tiny_add(grpc_slice_buffer *sb, size_t n) { @@ -154,17 +166,24 @@ void grpc_slice_buffer_pop(grpc_slice_buffer *sb) { } } -void grpc_slice_buffer_reset_and_unref(grpc_slice_buffer *sb) { +void grpc_slice_buffer_reset_and_unref_internal(grpc_exec_ctx *exec_ctx, + grpc_slice_buffer *sb) { size_t i; for (i = 0; i < sb->count; i++) { - grpc_slice_unref(sb->slices[i]); + grpc_slice_unref_internal(exec_ctx, sb->slices[i]); } sb->count = 0; sb->length = 0; } +void grpc_slice_buffer_reset_and_unref(grpc_slice_buffer *sb) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_slice_buffer_reset_and_unref_internal(&exec_ctx, sb); + grpc_exec_ctx_finish(&exec_ctx); +} + void grpc_slice_buffer_swap(grpc_slice_buffer *a, grpc_slice_buffer *b) { GPR_SWAP(size_t, a->count, b->count); GPR_SWAP(size_t, a->capacity, b->capacity); diff --git a/src/core/lib/slice/slice_internal.h b/src/core/lib/slice/slice_internal.h new file mode 100644 index 0000000000..72b0a590bb --- /dev/null +++ b/src/core/lib/slice/slice_internal.h @@ -0,0 +1,49 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_SUPPORT_SLICE_INTERNAL_H +#define GRPC_CORE_LIB_SUPPORT_SLICE_INTERNAL_H + +#include <grpc/slice.h> +#include <grpc/slice_buffer.h> + +#include "src/core/lib/iomgr/exec_ctx.h" + +grpc_slice grpc_slice_ref_internal(grpc_slice slice); +void grpc_slice_unref_internal(grpc_exec_ctx *exec_ctx, grpc_slice slice); +void grpc_slice_buffer_reset_and_unref_internal(grpc_exec_ctx *exec_ctx, + grpc_slice_buffer *sb); +void grpc_slice_buffer_destroy_internal(grpc_exec_ctx *exec_ctx, + grpc_slice_buffer *sb); + +#endif diff --git a/src/core/lib/slice/slice_string_helpers.c b/src/core/lib/slice/slice_string_helpers.c index 4731762ece..839c366b32 100644 --- a/src/core/lib/slice/slice_string_helpers.c +++ b/src/core/lib/slice/slice_string_helpers.c @@ -37,6 +37,7 @@ #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/string.h" char *grpc_dump_slice(grpc_slice s, uint32_t flags) { @@ -84,6 +85,6 @@ void grpc_slice_split(grpc_slice str, const char *sep, grpc_slice_buffer *dst) { grpc_slice_buffer_add_indexed( dst, grpc_slice_sub(str, end + sep_len, GRPC_SLICE_LENGTH(str))); } else { /* no sep found, add whole input */ - grpc_slice_buffer_add_indexed(dst, grpc_slice_ref(str)); + grpc_slice_buffer_add_indexed(dst, grpc_slice_ref_internal(str)); } } diff --git a/src/core/lib/surface/byte_buffer.c b/src/core/lib/surface/byte_buffer.c index d646591a65..c8e2fdfdad 100644 --- a/src/core/lib/surface/byte_buffer.c +++ b/src/core/lib/surface/byte_buffer.c @@ -35,6 +35,8 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" + grpc_byte_buffer *grpc_raw_byte_buffer_create(grpc_slice *slices, size_t nslices) { return grpc_raw_compressed_byte_buffer_create(slices, nslices, @@ -50,7 +52,7 @@ grpc_byte_buffer *grpc_raw_compressed_byte_buffer_create( bb->data.raw.compression = compression; grpc_slice_buffer_init(&bb->data.raw.slice_buffer); for (i = 0; i < nslices; i++) { - grpc_slice_ref(slices[i]); + grpc_slice_ref_internal(slices[i]); grpc_slice_buffer_add(&bb->data.raw.slice_buffer, slices[i]); } return bb; @@ -82,12 +84,14 @@ grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) { void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) { if (!bb) return; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; switch (bb->type) { case GRPC_BB_RAW: - grpc_slice_buffer_destroy(&bb->data.raw.slice_buffer); + grpc_slice_buffer_destroy_internal(&exec_ctx, &bb->data.raw.slice_buffer); break; } gpr_free(bb); + grpc_exec_ctx_finish(&exec_ctx); } size_t grpc_byte_buffer_length(grpc_byte_buffer *bb) { diff --git a/src/core/lib/surface/byte_buffer_reader.c b/src/core/lib/surface/byte_buffer_reader.c index 0089959fbb..1a6ccdaddb 100644 --- a/src/core/lib/surface/byte_buffer_reader.c +++ b/src/core/lib/surface/byte_buffer_reader.c @@ -42,6 +42,7 @@ #include <grpc/support/log.h> #include "src/core/lib/compression/message_compress.h" +#include "src/core/lib/slice/slice_internal.h" static int is_compressed(grpc_byte_buffer *buffer) { switch (buffer->type) { @@ -56,13 +57,15 @@ static int is_compressed(grpc_byte_buffer *buffer) { int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_slice_buffer decompressed_slices_buffer; reader->buffer_in = buffer; switch (reader->buffer_in->type) { case GRPC_BB_RAW: grpc_slice_buffer_init(&decompressed_slices_buffer); if (is_compressed(reader->buffer_in)) { - if (grpc_msg_decompress(reader->buffer_in->data.raw.compression, + if (grpc_msg_decompress(&exec_ctx, + reader->buffer_in->data.raw.compression, &reader->buffer_in->data.raw.slice_buffer, &decompressed_slices_buffer) == 0) { gpr_log(GPR_ERROR, @@ -76,13 +79,15 @@ int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, grpc_raw_byte_buffer_create(decompressed_slices_buffer.slices, decompressed_slices_buffer.count); } - grpc_slice_buffer_destroy(&decompressed_slices_buffer); + grpc_slice_buffer_destroy_internal(&exec_ctx, + &decompressed_slices_buffer); } else { /* not compressed, use the input buffer as output */ reader->buffer_out = reader->buffer_in; } reader->current.index = 0; break; } + grpc_exec_ctx_finish(&exec_ctx); return 1; } @@ -104,7 +109,8 @@ int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, grpc_slice_buffer *slice_buffer; slice_buffer = &reader->buffer_out->data.raw.slice_buffer; if (reader->current.index < slice_buffer->count) { - *slice = grpc_slice_ref(slice_buffer->slices[reader->current.index]); + *slice = grpc_slice_ref_internal( + slice_buffer->slices[reader->current.index]); reader->current.index += 1; return 1; } @@ -121,12 +127,14 @@ grpc_slice grpc_byte_buffer_reader_readall(grpc_byte_buffer_reader *reader) { grpc_slice out_slice = grpc_slice_malloc(input_size); uint8_t *const outbuf = GRPC_SLICE_START_PTR(out_slice); /* just an alias */ + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; while (grpc_byte_buffer_reader_next(reader, &in_slice) != 0) { const size_t slice_length = GRPC_SLICE_LENGTH(in_slice); memcpy(&(outbuf[bytes_read]), GRPC_SLICE_START_PTR(in_slice), slice_length); bytes_read += slice_length; - grpc_slice_unref(in_slice); + grpc_slice_unref_internal(&exec_ctx, in_slice); GPR_ASSERT(bytes_read <= input_size); } + grpc_exec_ctx_finish(&exec_ctx); return out_slice; } diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 62c0ec83a1..be568feba1 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -49,6 +49,7 @@ #include "src/core/lib/compression/algorithm_metadata.h" #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/api_trace.h" @@ -225,12 +226,12 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack, static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_error *error); -grpc_error *grpc_call_create(const grpc_call_create_args *args, +grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, + const grpc_call_create_args *args, grpc_call **out_call) { size_t i, j; grpc_channel_stack *channel_stack = grpc_channel_get_channel_stack(args->channel); - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_call *call; GPR_TIMER_BEGIN("grpc_call_create", 0); call = gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size); @@ -313,14 +314,14 @@ grpc_error *grpc_call_create(const grpc_call_create_args *args, GRPC_CHANNEL_INTERNAL_REF(args->channel, "call"); /* initial refcount dropped by grpc_call_destroy */ grpc_error *error = - grpc_call_stack_init(&exec_ctx, channel_stack, 1, destroy_call, call, + grpc_call_stack_init(exec_ctx, channel_stack, 1, destroy_call, call, call->context, args->server_transport_data, path, send_deadline, CALL_STACK_FROM_CALL(call)); if (error != GRPC_ERROR_NONE) { grpc_status_code status; const char *error_str; grpc_error_get_status(error, &status, &error_str); - close_with_status(&exec_ctx, call, status, error_str); + close_with_status(exec_ctx, call, status, error_str); } if (args->cq != NULL) { GPR_ASSERT( @@ -336,12 +337,11 @@ grpc_error *grpc_call_create(const grpc_call_create_args *args, } if (!grpc_polling_entity_is_empty(&call->pollent)) { grpc_call_stack_set_pollset_or_pollset_set( - &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); + exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); } - if (path != NULL) GRPC_MDSTR_UNREF(path); + if (path != NULL) GRPC_MDSTR_UNREF(exec_ctx, path); - grpc_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_call_create", 0); return error; } @@ -402,7 +402,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, GPR_TIMER_BEGIN("destroy_call", 0); for (i = 0; i < 2; i++) { grpc_metadata_batch_destroy( - &c->metadata_batch[1 /* is_receiving */][i /* is_initial */]); + exec_ctx, &c->metadata_batch[1 /* is_receiving */][i /* is_initial */]); } if (c->receiving_stream != NULL) { grpc_byte_stream_destroy(exec_ctx, c->receiving_stream); @@ -410,11 +410,11 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, gpr_mu_destroy(&c->mu); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (c->status[i].details) { - GRPC_MDSTR_UNREF(c->status[i].details); + GRPC_MDSTR_UNREF(exec_ctx, c->status[i].details); } } for (ii = 0; ii < c->send_extra_metadata_count; ii++) { - GRPC_MDELEM_UNREF(c->send_extra_metadata[ii].md); + GRPC_MDELEM_UNREF(exec_ctx, c->send_extra_metadata[ii].md); } for (i = 0; i < GRPC_CONTEXT_COUNT; i++) { if (c->context[i].destroy) { @@ -442,22 +442,22 @@ static void set_status_code(grpc_call *call, status_source source, call->status[source].code = (grpc_status_code)status; } -static void set_status_details(grpc_call *call, status_source source, - grpc_mdstr *status) { +static void set_status_details(grpc_exec_ctx *exec_ctx, grpc_call *call, + status_source source, grpc_mdstr *status) { if (call->status[source].details != NULL) { - GRPC_MDSTR_UNREF(status); + GRPC_MDSTR_UNREF(exec_ctx, status); } else { call->status[source].details = status; } } -static void set_status_from_error(grpc_call *call, status_source source, - grpc_error *error) { +static void set_status_from_error(grpc_exec_ctx *exec_ctx, grpc_call *call, + status_source source, grpc_error *error) { grpc_status_code status; const char *msg; grpc_error_get_status(error, &status, &msg); set_status_code(call, source, (uint32_t)status); - set_status_details(call, source, grpc_mdstr_from_string(msg)); + set_status_details(exec_ctx, call, source, grpc_mdstr_from_string(msg)); } static void set_incoming_compression_algorithm( @@ -491,7 +491,8 @@ uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) { static void destroy_encodings_accepted_by_peer(void *p) { return; } -static void set_encodings_accepted_by_peer(grpc_call *call, grpc_mdelem *mdel) { +static void set_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx, + grpc_call *call, grpc_mdelem *mdel) { size_t i; grpc_compression_algorithm algorithm; grpc_slice_buffer accept_encoding_parts; @@ -531,7 +532,7 @@ static void set_encodings_accepted_by_peer(grpc_call *call, grpc_mdelem *mdel) { } } - grpc_slice_buffer_destroy(&accept_encoding_parts); + grpc_slice_buffer_destroy_internal(exec_ctx, &accept_encoding_parts); grpc_mdelem_set_user_data( mdel, destroy_encodings_accepted_by_peer, @@ -589,12 +590,10 @@ static grpc_metadata *get_md_elem(grpc_metadata *metadata, return res; } -static int prepare_application_metadata(grpc_call *call, int count, - grpc_metadata *metadata, - int is_trailing, - int prepend_extra_metadata, - grpc_metadata *additional_metadata, - int additional_metadata_count) { +static int prepare_application_metadata( + grpc_exec_ctx *exec_ctx, grpc_call *call, int count, + grpc_metadata *metadata, int is_trailing, int prepend_extra_metadata, + grpc_metadata *additional_metadata, int additional_metadata_count) { int total_count = count + additional_metadata_count; int i; grpc_metadata_batch *batch = @@ -605,7 +604,7 @@ static int prepare_application_metadata(grpc_call *call, int count, grpc_linked_mdelem *l = (grpc_linked_mdelem *)&md->internal_data; GPR_ASSERT(sizeof(grpc_linked_mdelem) == sizeof(md->internal_data)); l->md = grpc_mdelem_from_string_and_buffer( - md->key, (const uint8_t *)md->value, md->value_length); + exec_ctx, md->key, (const uint8_t *)md->value, md->value_length); if (!grpc_header_key_is_legal(grpc_mdstr_as_c_string(l->md->key), GRPC_MDSTR_LENGTH(l->md->key))) { gpr_log(GPR_ERROR, "attempt to send invalid metadata key: %s", @@ -625,7 +624,7 @@ static int prepare_application_metadata(grpc_call *call, int count, const grpc_metadata *md = get_md_elem(metadata, additional_metadata, j, count); grpc_linked_mdelem *l = (grpc_linked_mdelem *)&md->internal_data; - GRPC_MDELEM_UNREF(l->md); + GRPC_MDELEM_UNREF(exec_ctx, l->md); } return 0; } @@ -808,7 +807,8 @@ static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, grpc_error *error) { static grpc_call_error terminate_with_status(grpc_exec_ctx *exec_ctx, termination_closure *tc) { - set_status_from_error(tc->call, STATUS_FROM_API_OVERRIDE, tc->error); + set_status_from_error(exec_ctx, tc->call, STATUS_FROM_API_OVERRIDE, + tc->error); if (tc->type == TC_CANCEL) { grpc_closure_init(&tc->closure, send_cancel, tc); @@ -925,7 +925,8 @@ static grpc_compression_algorithm decode_compression(grpc_mdelem *md) { return algorithm; } -static grpc_mdelem *recv_common_filter(grpc_call *call, grpc_mdelem *elem) { +static grpc_mdelem *recv_common_filter(grpc_exec_ctx *exec_ctx, grpc_call *call, + grpc_mdelem *elem) { if (elem->key == GRPC_MDSTR_GRPC_STATUS) { GPR_TIMER_BEGIN("status", 0); set_status_code(call, STATUS_FROM_WIRE, decode_status(elem)); @@ -933,7 +934,8 @@ static grpc_mdelem *recv_common_filter(grpc_call *call, grpc_mdelem *elem) { return NULL; } else if (elem->key == GRPC_MDSTR_GRPC_MESSAGE) { GPR_TIMER_BEGIN("status-details", 0); - set_status_details(call, STATUS_FROM_WIRE, GRPC_MDSTR_REF(elem->value)); + set_status_details(exec_ctx, call, STATUS_FROM_WIRE, + GRPC_MDSTR_REF(elem->value)); GPR_TIMER_END("status-details", 0); return NULL; } @@ -959,33 +961,38 @@ static grpc_mdelem *publish_app_metadata(grpc_call *call, grpc_mdelem *elem, return elem; } -static grpc_mdelem *recv_initial_filter(void *callp, grpc_mdelem *elem) { - grpc_call *call = callp; - elem = recv_common_filter(call, elem); +typedef struct { + grpc_exec_ctx *exec_ctx; + grpc_call *call; +} recv_filter_args; + +static grpc_mdelem *recv_initial_filter(void *args, grpc_mdelem *elem) { + recv_filter_args *a = args; + elem = recv_common_filter(a->exec_ctx, a->call, elem); if (elem == NULL) { return NULL; } else if (elem->key == GRPC_MDSTR_GRPC_ENCODING) { GPR_TIMER_BEGIN("incoming_compression_algorithm", 0); - set_incoming_compression_algorithm(call, decode_compression(elem)); + set_incoming_compression_algorithm(a->call, decode_compression(elem)); GPR_TIMER_END("incoming_compression_algorithm", 0); return NULL; } else if (elem->key == GRPC_MDSTR_GRPC_ACCEPT_ENCODING) { GPR_TIMER_BEGIN("encodings_accepted_by_peer", 0); - set_encodings_accepted_by_peer(call, elem); + set_encodings_accepted_by_peer(a->exec_ctx, a->call, elem); GPR_TIMER_END("encodings_accepted_by_peer", 0); return NULL; } else { - return publish_app_metadata(call, elem, 0); + return publish_app_metadata(a->call, elem, 0); } } -static grpc_mdelem *recv_trailing_filter(void *callp, grpc_mdelem *elem) { - grpc_call *call = callp; - elem = recv_common_filter(call, elem); +static grpc_mdelem *recv_trailing_filter(void *args, grpc_mdelem *elem) { + recv_filter_args *a = args; + elem = recv_common_filter(a->exec_ctx, a->call, elem); if (elem == NULL) { return NULL; } else { - return publish_app_metadata(call, elem, 1); + return publish_app_metadata(a->call, elem, 1); } } @@ -1231,7 +1238,8 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, if (error == GRPC_ERROR_NONE) { grpc_metadata_batch *md = &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; - grpc_metadata_batch_filter(md, recv_initial_filter, call); + recv_filter_args args = {exec_ctx, call}; + grpc_metadata_batch_filter(exec_ctx, md, recv_initial_filter, &args); GPR_TIMER_BEGIN("validate_filtered_metadata", 0); validate_filtered_metadata(exec_ctx, bctl); @@ -1275,14 +1283,15 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, intptr_t status; if (error != GRPC_ERROR_NONE && grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) { - set_status_from_error(call, STATUS_FROM_CORE, error); + set_status_from_error(exec_ctx, call, STATUS_FROM_CORE, error); } if (bctl->send_initial_metadata) { if (error != GRPC_ERROR_NONE) { - set_status_from_error(call, STATUS_FROM_CORE, error); + set_status_from_error(exec_ctx, call, STATUS_FROM_CORE, error); } grpc_metadata_batch_destroy( + exec_ctx, &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]); } if (bctl->send_message) { @@ -1290,12 +1299,14 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, } if (bctl->send_final_op) { grpc_metadata_batch_destroy( + exec_ctx, &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]); } if (bctl->recv_final_op) { grpc_metadata_batch *md = &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - grpc_metadata_batch_filter(md, recv_trailing_filter, call); + recv_filter_args args = {exec_ctx, call}; + grpc_metadata_batch_filter(exec_ctx, md, recv_trailing_filter, &args); call->received_final_op = true; /* propagate cancellation to any interested children */ @@ -1432,7 +1443,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, bctl->send_initial_metadata = 1; call->sent_initial_metadata = 1; if (!prepare_application_metadata( - call, (int)op->data.send_initial_metadata.count, + exec_ctx, call, (int)op->data.send_initial_metadata.count, op->data.send_initial_metadata.metadata, 0, call->is_client, &compression_md, (int)additional_metadata_count)) { error = GRPC_CALL_ERROR_INVALID_METADATA; @@ -1506,15 +1517,15 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, call->sent_final_op = 1; call->send_extra_metadata_count = 1; call->send_extra_metadata[0].md = grpc_channel_get_reffed_status_elem( - call->channel, op->data.send_status_from_server.status); + exec_ctx, call->channel, op->data.send_status_from_server.status); if (op->data.send_status_from_server.status_details != NULL) { call->send_extra_metadata[1].md = grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_GRPC_MESSAGE, + exec_ctx, GRPC_MDSTR_GRPC_MESSAGE, grpc_mdstr_from_string( op->data.send_status_from_server.status_details)); call->send_extra_metadata_count++; set_status_details( - call, STATUS_FROM_API_OVERRIDE, + exec_ctx, call, STATUS_FROM_API_OVERRIDE, GRPC_MDSTR_REF(call->send_extra_metadata[1].md->value)); } if (op->data.send_status_from_server.status != GRPC_STATUS_OK) { @@ -1522,7 +1533,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, (uint32_t)op->data.send_status_from_server.status); } if (!prepare_application_metadata( - call, + exec_ctx, call, (int)op->data.send_status_from_server.trailing_metadata_count, op->data.send_status_from_server.trailing_metadata, 1, 1, NULL, 0)) { @@ -1647,7 +1658,7 @@ done_with_error: /* reverse any mutations that occured */ if (bctl->send_initial_metadata) { call->sent_initial_metadata = 0; - grpc_metadata_batch_clear(&call->metadata_batch[0][0]); + grpc_metadata_batch_clear(exec_ctx, &call->metadata_batch[0][0]); } if (bctl->send_message) { call->sending_message = 0; @@ -1655,7 +1666,7 @@ done_with_error: } if (bctl->send_final_op) { call->sent_final_op = 0; - grpc_metadata_batch_clear(&call->metadata_batch[0][1]); + grpc_metadata_batch_clear(exec_ctx, &call->metadata_batch[0][1]); } if (bctl->recv_initial_metadata) { call->received_initial_metadata = 0; diff --git a/src/core/lib/surface/call.h b/src/core/lib/surface/call.h index 18af41b7fb..233340c329 100644 --- a/src/core/lib/surface/call.h +++ b/src/core/lib/surface/call.h @@ -70,7 +70,8 @@ typedef struct grpc_call_create_args { /* Create a new call based on \a args. Regardless of success or failure, always returns a valid new call into *call */ -grpc_error *grpc_call_create(const grpc_call_create_args *args, +grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx, + const grpc_call_create_args *args, grpc_call **call); void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call, diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index 92d783b78d..82617390bb 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -89,13 +89,14 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, bool is_client = grpc_channel_stack_type_is_client(channel_stack_type); grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create(); - grpc_channel_stack_builder_set_channel_arguments(builder, input_args); + grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder, + input_args); grpc_channel_stack_builder_set_target(builder, target); grpc_channel_stack_builder_set_transport(builder, optional_transport); grpc_channel *channel; grpc_channel_args *args; if (!grpc_channel_init_create_stack(exec_ctx, builder, channel_stack_type)) { - grpc_channel_stack_builder_destroy(builder); + grpc_channel_stack_builder_destroy(exec_ctx, builder); return NULL; } else { args = grpc_channel_args_copy( @@ -120,10 +121,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, } else { if (channel->default_authority) { /* setting this takes precedence over anything else */ - GRPC_MDELEM_UNREF(channel->default_authority); + GRPC_MDELEM_UNREF(exec_ctx, channel->default_authority); } channel->default_authority = grpc_mdelem_from_strings( - ":authority", args->args[i].value.string); + exec_ctx, ":authority", args->args[i].value.string); } } else if (0 == strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) { @@ -138,7 +139,7 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG); } else { channel->default_authority = grpc_mdelem_from_strings( - ":authority", args->args[i].value.string); + exec_ctx, ":authority", args->args[i].value.string); } } } else if (0 == strcmp(args->args[i].key, @@ -164,7 +165,7 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, 0x1; /* always support no compression */ } } - grpc_channel_args_destroy(args); + grpc_channel_args_destroy(exec_ctx, args); } return channel; @@ -176,10 +177,10 @@ char *grpc_channel_get_target(grpc_channel *channel) { } static grpc_call *grpc_channel_create_call_internal( - grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, - grpc_completion_queue *cq, grpc_pollset_set *pollset_set_alternative, - grpc_mdelem *path_mdelem, grpc_mdelem *authority_mdelem, - gpr_timespec deadline) { + grpc_exec_ctx *exec_ctx, grpc_channel *channel, grpc_call *parent_call, + uint32_t propagation_mask, grpc_completion_queue *cq, + grpc_pollset_set *pollset_set_alternative, grpc_mdelem *path_mdelem, + grpc_mdelem *authority_mdelem, gpr_timespec deadline) { grpc_mdelem *send_metadata[2]; size_t num_metadata = 0; @@ -206,7 +207,7 @@ static grpc_call *grpc_channel_create_call_internal( args.send_deadline = deadline; grpc_call *call; - GRPC_LOG_IF_ERROR("call_create", grpc_call_create(&args, &call)); + GRPC_LOG_IF_ERROR("call_create", grpc_call_create(exec_ctx, &args, &call)); return call; } @@ -227,26 +228,30 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel, (channel, parent_call, (unsigned)propagation_mask, cq, method, host, deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); - return grpc_channel_create_call_internal( - channel, parent_call, propagation_mask, cq, NULL, - grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_call *call = grpc_channel_create_call_internal( + &exec_ctx, channel, parent_call, propagation_mask, cq, NULL, + grpc_mdelem_from_metadata_strings(&exec_ctx, GRPC_MDSTR_PATH, grpc_mdstr_from_string(method)), - host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY, + host ? grpc_mdelem_from_metadata_strings(&exec_ctx, GRPC_MDSTR_AUTHORITY, grpc_mdstr_from_string(host)) : NULL, deadline); + grpc_exec_ctx_finish(&exec_ctx); + return call; } grpc_call *grpc_channel_create_pollset_set_call( - grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, - grpc_pollset_set *pollset_set, const char *method, const char *host, - gpr_timespec deadline, void *reserved) { + grpc_exec_ctx *exec_ctx, grpc_channel *channel, grpc_call *parent_call, + uint32_t propagation_mask, grpc_pollset_set *pollset_set, + const char *method, const char *host, gpr_timespec deadline, + void *reserved) { GPR_ASSERT(!reserved); return grpc_channel_create_call_internal( - channel, parent_call, propagation_mask, NULL, pollset_set, - grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + exec_ctx, channel, parent_call, propagation_mask, NULL, pollset_set, + grpc_mdelem_from_metadata_strings(exec_ctx, GRPC_MDSTR_PATH, grpc_mdstr_from_string(method)), - host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY, + host ? grpc_mdelem_from_metadata_strings(exec_ctx, GRPC_MDSTR_AUTHORITY, grpc_mdstr_from_string(host)) : NULL, deadline); @@ -259,15 +264,18 @@ void *grpc_channel_register_call(grpc_channel *channel, const char *method, "grpc_channel_register_call(channel=%p, method=%s, host=%s, reserved=%p)", 4, (channel, method, host, reserved)); GPR_ASSERT(!reserved); - rc->path = grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + rc->path = grpc_mdelem_from_metadata_strings(&exec_ctx, GRPC_MDSTR_PATH, grpc_mdstr_from_string(method)); - rc->authority = host ? grpc_mdelem_from_metadata_strings( - GRPC_MDSTR_AUTHORITY, grpc_mdstr_from_string(host)) - : NULL; + rc->authority = + host ? grpc_mdelem_from_metadata_strings(&exec_ctx, GRPC_MDSTR_AUTHORITY, + grpc_mdstr_from_string(host)) + : NULL; gpr_mu_lock(&channel->registered_call_mu); rc->next = channel->registered_calls; channel->registered_calls = rc; gpr_mu_unlock(&channel->registered_call_mu); + grpc_exec_ctx_finish(&exec_ctx); return rc; } @@ -287,10 +295,13 @@ grpc_call *grpc_channel_create_registered_call( registered_call_handle, deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); - return grpc_channel_create_call_internal( - channel, parent_call, propagation_mask, completion_queue, NULL, + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_call *call = grpc_channel_create_call_internal( + &exec_ctx, channel, parent_call, propagation_mask, completion_queue, NULL, GRPC_MDELEM_REF(rc->path), rc->authority ? GRPC_MDELEM_REF(rc->authority) : NULL, deadline); + grpc_exec_ctx_finish(&exec_ctx); + return call; } #ifdef GRPC_STREAM_REFCOUNT_DEBUG @@ -316,14 +327,14 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, while (channel->registered_calls) { registered_call *rc = channel->registered_calls; channel->registered_calls = rc->next; - GRPC_MDELEM_UNREF(rc->path); + GRPC_MDELEM_UNREF(exec_ctx, rc->path); if (rc->authority) { - GRPC_MDELEM_UNREF(rc->authority); + GRPC_MDELEM_UNREF(exec_ctx, rc->authority); } gpr_free(rc); } if (channel->default_authority != NULL) { - GRPC_MDELEM_UNREF(channel->default_authority); + GRPC_MDELEM_UNREF(exec_ctx, channel->default_authority); } gpr_mu_destroy(&channel->registered_call_mu); gpr_free(channel->target); @@ -353,7 +364,8 @@ grpc_compression_options grpc_channel_compression_options( return channel->compression_options; } -grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) { +grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_exec_ctx *exec_ctx, + grpc_channel *channel, int i) { char tmp[GPR_LTOA_MIN_BUFSIZE]; switch (i) { case 0: @@ -364,6 +376,6 @@ grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) { return GRPC_MDELEM_GRPC_STATUS_2; } gpr_ltoa(i, tmp); - return grpc_mdelem_from_metadata_strings(GRPC_MDSTR_GRPC_STATUS, + return grpc_mdelem_from_metadata_strings(exec_ctx, GRPC_MDSTR_GRPC_STATUS, grpc_mdstr_from_string(tmp)); } diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h index 23cc8656ca..2ebadb7a15 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -51,9 +51,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, properties from the server call to this new client call, depending on the value of \a propagation_mask (see propagation_bits.h for possible values) */ grpc_call *grpc_channel_create_pollset_set_call( - grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, - grpc_pollset_set *pollset_set, const char *method, const char *host, - gpr_timespec deadline, void *reserved); + grpc_exec_ctx *exec_ctx, grpc_channel *channel, grpc_call *parent_call, + uint32_t propagation_mask, grpc_pollset_set *pollset_set, + const char *method, const char *host, gpr_timespec deadline, + void *reserved); /** Get a (borrowed) pointer to this channels underlying channel stack */ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel); @@ -62,7 +63,8 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel); status_code. The returned elem is owned by the caller. */ -grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, +grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_exec_ctx *exec_ctx, + grpc_channel *channel, int status_code); #ifdef GRPC_STREAM_REFCOUNT_DEBUG diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c index 7903f57a68..8c82f38c77 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -221,6 +221,7 @@ void grpc_init(void) { void grpc_shutdown(void) { int i; GRPC_API_TRACE("grpc_shutdown(void)", 0, ()); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; gpr_mu_lock(&g_init_mu); if (--g_initializations == 0) { grpc_executor_shutdown(); @@ -233,9 +234,10 @@ void grpc_shutdown(void) { g_all_of_the_plugins[i].destroy(); } } - grpc_mdctx_global_shutdown(); + grpc_mdctx_global_shutdown(&exec_ctx); } gpr_mu_unlock(&g_init_mu); + grpc_exec_ctx_finish(&exec_ctx); } int grpc_is_initialized(void) { diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c index d32c884e8e..1b57c5cd01 100644 --- a/src/core/lib/surface/lame_client.c +++ b/src/core/lib/surface/lame_client.c @@ -55,14 +55,15 @@ typedef struct { const char *error_message; } channel_data; -static void fill_metadata(grpc_call_element *elem, grpc_metadata_batch *mdb) { +static void fill_metadata(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_metadata_batch *mdb) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; char tmp[GPR_LTOA_MIN_BUFSIZE]; gpr_ltoa(chand->error_code, tmp); - calld->status.md = grpc_mdelem_from_strings("grpc-status", tmp); + calld->status.md = grpc_mdelem_from_strings(exec_ctx, "grpc-status", tmp); calld->details.md = - grpc_mdelem_from_strings("grpc-message", chand->error_message); + grpc_mdelem_from_strings(exec_ctx, "grpc-message", chand->error_message); calld->status.prev = calld->details.next = NULL; calld->status.next = &calld->details; calld->details.prev = &calld->status; @@ -76,9 +77,9 @@ static void lame_start_transport_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); if (op->recv_initial_metadata != NULL) { - fill_metadata(elem, op->recv_initial_metadata); + fill_metadata(exec_ctx, elem, op->recv_initial_metadata); } else if (op->recv_trailing_metadata != NULL) { - fill_metadata(elem, op->recv_trailing_metadata); + fill_metadata(exec_ctx, elem, op->recv_trailing_metadata); } grpc_transport_stream_op_finish_with_failure( exec_ctx, op, GRPC_ERROR_CREATE("lame client channel")); diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 798f582cad..6d9d3a92ab 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -45,6 +45,7 @@ #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" #include "src/core/lib/iomgr/iomgr.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/stack_lockfree.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/api_trace.h" @@ -270,7 +271,7 @@ struct shutdown_cleanup_args { static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { struct shutdown_cleanup_args *a = arg; - grpc_slice_unref(a->slice); + grpc_slice_unref_internal(exec_ctx, a->slice); gpr_free(a); } @@ -378,7 +379,7 @@ static void server_ref(grpc_server *server) { static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) { registered_method *rm; size_t i; - grpc_channel_args_destroy(server->channel_args); + grpc_channel_args_destroy(exec_ctx, server->channel_args); gpr_mu_destroy(&server->mu_global); gpr_mu_destroy(&server->mu_call); while ((rm = server->registered_methods) != NULL) { @@ -763,7 +764,8 @@ static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr, gpr_timespec op_deadline; GRPC_ERROR_REF(error); - grpc_metadata_batch_filter(calld->recv_initial_metadata, server_filter, elem); + grpc_metadata_batch_filter(exec_ctx, calld->recv_initial_metadata, + server_filter, elem); op_deadline = calld->recv_initial_metadata->deadline; if (0 != gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) { calld->deadline = op_deadline; @@ -837,7 +839,7 @@ static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd, args.server_transport_data = transport_server_data; args.send_deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); grpc_call *call; - grpc_error *error = grpc_call_create(&args, &call); + grpc_error *error = grpc_call_create(exec_ctx, &args, &call); grpc_call_element *elem = grpc_call_stack_element(grpc_call_get_call_stack(call), 0); if (error != GRPC_ERROR_NONE) { @@ -901,10 +903,10 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, GPR_ASSERT(calld->state != PENDING); if (calld->host) { - GRPC_MDSTR_UNREF(calld->host); + GRPC_MDSTR_UNREF(exec_ctx, calld->host); } if (calld->path) { - GRPC_MDSTR_UNREF(calld->path); + GRPC_MDSTR_UNREF(exec_ctx, calld->path); } grpc_metadata_array_destroy(&calld->initial_metadata); @@ -935,10 +937,10 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, if (chand->registered_methods) { for (i = 0; i < chand->registered_method_slots; i++) { if (chand->registered_methods[i].method) { - GRPC_MDSTR_UNREF(chand->registered_methods[i].method); + GRPC_MDSTR_UNREF(exec_ctx, chand->registered_methods[i].method); } if (chand->registered_methods[i].host) { - GRPC_MDSTR_UNREF(chand->registered_methods[i].host); + GRPC_MDSTR_UNREF(exec_ctx, chand->registered_methods[i].host); } } gpr_free(chand->registered_methods); diff --git a/src/core/lib/transport/byte_stream.c b/src/core/lib/transport/byte_stream.c index 2f1d7b7c60..4d4206189e 100644 --- a/src/core/lib/transport/byte_stream.c +++ b/src/core/lib/transport/byte_stream.c @@ -37,6 +37,8 @@ #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" + int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream, grpc_slice *slice, size_t max_size_hint, grpc_closure *on_complete) { @@ -57,7 +59,8 @@ static int slice_buffer_stream_next(grpc_exec_ctx *exec_ctx, grpc_closure *on_complete) { grpc_slice_buffer_stream *stream = (grpc_slice_buffer_stream *)byte_stream; GPR_ASSERT(stream->cursor < stream->backing_buffer->count); - *slice = grpc_slice_ref(stream->backing_buffer->slices[stream->cursor]); + *slice = + grpc_slice_ref_internal(stream->backing_buffer->slices[stream->cursor]); stream->cursor++; return 1; } diff --git a/src/core/lib/transport/mdstr_hash_table.c b/src/core/lib/transport/mdstr_hash_table.c index 8e914c420b..a3f6bde516 100644 --- a/src/core/lib/transport/mdstr_hash_table.c +++ b/src/core/lib/transport/mdstr_hash_table.c @@ -96,13 +96,14 @@ grpc_mdstr_hash_table* grpc_mdstr_hash_table_ref(grpc_mdstr_hash_table* table) { return table; } -int grpc_mdstr_hash_table_unref(grpc_mdstr_hash_table* table) { +int grpc_mdstr_hash_table_unref(grpc_exec_ctx* exec_ctx, + grpc_mdstr_hash_table* table) { if (table != NULL && gpr_unref(&table->refs)) { for (size_t i = 0; i < table->size; ++i) { grpc_mdstr_hash_table_entry* entry = &table->entries[i]; if (entry->key != NULL) { - GRPC_MDSTR_UNREF(entry->key); - entry->vtable->destroy_value(entry->value); + GRPC_MDSTR_UNREF(exec_ctx, entry->key); + entry->vtable->destroy_value(exec_ctx, entry->value); } } gpr_free(table->entries); diff --git a/src/core/lib/transport/mdstr_hash_table.h b/src/core/lib/transport/mdstr_hash_table.h index bceb4df93d..45e5720063 100644 --- a/src/core/lib/transport/mdstr_hash_table.h +++ b/src/core/lib/transport/mdstr_hash_table.h @@ -49,7 +49,7 @@ typedef struct grpc_mdstr_hash_table grpc_mdstr_hash_table; typedef struct grpc_mdstr_hash_table_vtable { - void (*destroy_value)(void* value); + void (*destroy_value)(grpc_exec_ctx* exec_ctx, void* value); void* (*copy_value)(void* value); int (*compare_value)(void* value1, void* value2); } grpc_mdstr_hash_table_vtable; @@ -68,7 +68,8 @@ grpc_mdstr_hash_table* grpc_mdstr_hash_table_create( grpc_mdstr_hash_table* grpc_mdstr_hash_table_ref(grpc_mdstr_hash_table* table); /** Returns 1 when \a table is destroyed. */ -int grpc_mdstr_hash_table_unref(grpc_mdstr_hash_table* table); +int grpc_mdstr_hash_table_unref(grpc_exec_ctx* exec_ctx, + grpc_mdstr_hash_table* table); /** Returns the number of entries in \a table. */ size_t grpc_mdstr_hash_table_num_entries(const grpc_mdstr_hash_table* table); diff --git a/src/core/lib/transport/metadata.c b/src/core/lib/transport/metadata.c index a1748c033b..ef5fd32b52 100644 --- a/src/core/lib/transport/metadata.c +++ b/src/core/lib/transport/metadata.c @@ -47,6 +47,7 @@ #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/support/murmur_hash.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/static_metadata.h" @@ -153,7 +154,7 @@ static size_t g_static_mdtab_maxprobe; static strtab_shard g_strtab_shard[STRTAB_SHARD_COUNT]; static mdtab_shard g_mdtab_shard[MDTAB_SHARD_COUNT]; -static void gc_mdtab(mdtab_shard *shard); +static void gc_mdtab(grpc_exec_ctx *exec_ctx, mdtab_shard *shard); void grpc_test_only_set_metadata_hash_seed(uint32_t seed) { g_hash_seed = seed; @@ -227,12 +228,12 @@ void grpc_mdctx_global_init(void) { } } -void grpc_mdctx_global_shutdown(void) { +void grpc_mdctx_global_shutdown(grpc_exec_ctx *exec_ctx) { size_t i; for (i = 0; i < MDTAB_SHARD_COUNT; i++) { mdtab_shard *shard = &g_mdtab_shard[i]; gpr_mu_destroy(&shard->mu); - gc_mdtab(shard); + gc_mdtab(exec_ctx, shard); /* TODO(ctiller): GPR_ASSERT(shard->count == 0); */ if (shard->count != 0) { gpr_log(GPR_DEBUG, "WARNING: %" PRIuPTR " metadata elements were leaked", @@ -316,12 +317,13 @@ static void grow_strtab(strtab_shard *shard) { GPR_TIMER_END("grow_strtab", 0); } -static void internal_destroy_string(strtab_shard *shard, internal_string *is) { +static void internal_destroy_string(grpc_exec_ctx *exec_ctx, + strtab_shard *shard, internal_string *is) { internal_string **prev_next; internal_string *cur; GPR_TIMER_BEGIN("internal_destroy_string", 0); if (is->has_base64_and_huffman_encoded) { - grpc_slice_unref(is->base64_and_huffman); + grpc_slice_unref_internal(exec_ctx, is->base64_and_huffman); } for (prev_next = &shard->strs[TABLE_IDX(is->hash, LOG2_STRTAB_SHARD_COUNT, shard->capacity)], @@ -340,20 +342,20 @@ static void slice_ref(void *p) { GRPC_MDSTR_REF((grpc_mdstr *)(is)); } -static void slice_unref(void *p) { +static void slice_unref(grpc_exec_ctx *exec_ctx, void *p) { internal_string *is = (internal_string *)((char *)p - offsetof(internal_string, refcount)); - GRPC_MDSTR_UNREF((grpc_mdstr *)(is)); + GRPC_MDSTR_UNREF(exec_ctx, (grpc_mdstr *)(is)); } grpc_mdstr *grpc_mdstr_from_string(const char *str) { return grpc_mdstr_from_buffer((const uint8_t *)str, strlen(str)); } -grpc_mdstr *grpc_mdstr_from_slice(grpc_slice slice) { +grpc_mdstr *grpc_mdstr_from_slice(grpc_exec_ctx *exec_ctx, grpc_slice slice) { grpc_mdstr *result = grpc_mdstr_from_buffer(GRPC_SLICE_START_PTR(slice), GRPC_SLICE_LENGTH(slice)); - grpc_slice_unref(slice); + grpc_slice_unref_internal(exec_ctx, slice); return result; } @@ -444,7 +446,7 @@ grpc_mdstr *grpc_mdstr_from_buffer(const uint8_t *buf, size_t length) { return (grpc_mdstr *)s; } -static void gc_mdtab(mdtab_shard *shard) { +static void gc_mdtab(grpc_exec_ctx *exec_ctx, mdtab_shard *shard) { size_t i; internal_metadata **prev_next; internal_metadata *md, *next; @@ -457,8 +459,8 @@ static void gc_mdtab(mdtab_shard *shard) { void *user_data = (void *)gpr_atm_no_barrier_load(&md->user_data); next = md->bucket_next; if (gpr_atm_acq_load(&md->refcnt) == 0) { - GRPC_MDSTR_UNREF((grpc_mdstr *)md->key); - GRPC_MDSTR_UNREF((grpc_mdstr *)md->value); + GRPC_MDSTR_UNREF(exec_ctx, (grpc_mdstr *)md->key); + GRPC_MDSTR_UNREF(exec_ctx, (grpc_mdstr *)md->value); if (md->user_data) { ((destroy_user_data_func)gpr_atm_no_barrier_load( &md->destroy_user_data))(user_data); @@ -506,16 +508,17 @@ static void grow_mdtab(mdtab_shard *shard) { GPR_TIMER_END("grow_mdtab", 0); } -static void rehash_mdtab(mdtab_shard *shard) { +static void rehash_mdtab(grpc_exec_ctx *exec_ctx, mdtab_shard *shard) { if (gpr_atm_no_barrier_load(&shard->free_estimate) > (gpr_atm)(shard->capacity / 4)) { - gc_mdtab(shard); + gc_mdtab(exec_ctx, shard); } else { grow_mdtab(shard); } } -grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *mkey, +grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_exec_ctx *exec_ctx, + grpc_mdstr *mkey, grpc_mdstr *mvalue) { internal_string *key = (internal_string *)mkey; internal_string *value = (internal_string *)mvalue; @@ -547,8 +550,8 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *mkey, for (md = shard->elems[idx]; md; md = md->bucket_next) { if (md->key == key && md->value == value) { REF_MD_LOCKED(shard, md); - GRPC_MDSTR_UNREF((grpc_mdstr *)key); - GRPC_MDSTR_UNREF((grpc_mdstr *)value); + GRPC_MDSTR_UNREF(exec_ctx, (grpc_mdstr *)key); + GRPC_MDSTR_UNREF(exec_ctx, (grpc_mdstr *)value); gpr_mu_unlock(&shard->mu); GPR_TIMER_END("grpc_mdelem_from_metadata_strings", 0); return (grpc_mdelem *)md; @@ -574,7 +577,7 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *mkey, shard->count++; if (shard->count > shard->capacity * 2) { - rehash_mdtab(shard); + rehash_mdtab(exec_ctx, shard); } gpr_mu_unlock(&shard->mu); @@ -584,21 +587,26 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *mkey, return (grpc_mdelem *)md; } -grpc_mdelem *grpc_mdelem_from_strings(const char *key, const char *value) { - return grpc_mdelem_from_metadata_strings(grpc_mdstr_from_string(key), - grpc_mdstr_from_string(value)); +grpc_mdelem *grpc_mdelem_from_strings(grpc_exec_ctx *exec_ctx, const char *key, + const char *value) { + return grpc_mdelem_from_metadata_strings( + exec_ctx, grpc_mdstr_from_string(key), grpc_mdstr_from_string(value)); } -grpc_mdelem *grpc_mdelem_from_slices(grpc_slice key, grpc_slice value) { - return grpc_mdelem_from_metadata_strings(grpc_mdstr_from_slice(key), - grpc_mdstr_from_slice(value)); +grpc_mdelem *grpc_mdelem_from_slices(grpc_exec_ctx *exec_ctx, grpc_slice key, + grpc_slice value) { + return grpc_mdelem_from_metadata_strings( + exec_ctx, grpc_mdstr_from_slice(exec_ctx, key), + grpc_mdstr_from_slice(exec_ctx, value)); } -grpc_mdelem *grpc_mdelem_from_string_and_buffer(const char *key, +grpc_mdelem *grpc_mdelem_from_string_and_buffer(grpc_exec_ctx *exec_ctx, + const char *key, const uint8_t *value, size_t value_length) { return grpc_mdelem_from_metadata_strings( - grpc_mdstr_from_string(key), grpc_mdstr_from_buffer(value, value_length)); + exec_ctx, grpc_mdstr_from_string(key), + grpc_mdstr_from_buffer(value, value_length)); } static size_t get_base64_encoded_size(size_t raw_length) { @@ -654,7 +662,7 @@ grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *gmd DEBUG_ARGS) { return gmd; } -void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) { +void grpc_mdelem_unref(grpc_exec_ctx *exec_ctx, grpc_mdelem *gmd DEBUG_ARGS) { internal_metadata *md = (internal_metadata *)gmd; if (!md) return; if (is_mdelem_static(gmd)) return; @@ -691,7 +699,7 @@ grpc_mdstr *grpc_mdstr_ref(grpc_mdstr *gs DEBUG_ARGS) { return gs; } -void grpc_mdstr_unref(grpc_mdstr *gs DEBUG_ARGS) { +void grpc_mdstr_unref(grpc_exec_ctx *exec_ctx, grpc_mdstr *gs DEBUG_ARGS) { internal_string *s = (internal_string *)gs; if (is_mdstr_static(gs)) return; if (1 == gpr_atm_full_fetch_add(&s->refcnt, -1)) { @@ -699,7 +707,7 @@ void grpc_mdstr_unref(grpc_mdstr *gs DEBUG_ARGS) { &g_strtab_shard[SHARD_IDX(s->hash, LOG2_STRTAB_SHARD_COUNT)]; gpr_mu_lock(&shard->mu); GPR_ASSERT(0 == gpr_atm_no_barrier_load(&s->refcnt)); - internal_destroy_string(shard, s); + internal_destroy_string(exec_ctx, shard, s); gpr_mu_unlock(&shard->mu); } } diff --git a/src/core/lib/transport/metadata.h b/src/core/lib/transport/metadata.h index 8dcfbb98bb..cf77753692 100644 --- a/src/core/lib/transport/metadata.h +++ b/src/core/lib/transport/metadata.h @@ -96,7 +96,7 @@ void grpc_test_only_set_metadata_hash_seed(uint32_t seed); clients may have handy */ grpc_mdstr *grpc_mdstr_from_string(const char *str); /* Unrefs the slice. */ -grpc_mdstr *grpc_mdstr_from_slice(grpc_slice slice); +grpc_mdstr *grpc_mdstr_from_slice(grpc_exec_ctx *exec_ctx, grpc_slice slice); grpc_mdstr *grpc_mdstr_from_buffer(const uint8_t *str, size_t length); /* Returns a borrowed slice from the mdstr with its contents base64 encoded @@ -105,12 +105,16 @@ grpc_slice grpc_mdstr_as_base64_encoded_and_huffman_compressed(grpc_mdstr *str); /* Constructors for grpc_mdelem instances; take a variety of data types that clients may have handy */ -grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *key, +grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_exec_ctx *exec_ctx, + grpc_mdstr *key, grpc_mdstr *value); -grpc_mdelem *grpc_mdelem_from_strings(const char *key, const char *value); +grpc_mdelem *grpc_mdelem_from_strings(grpc_exec_ctx *exec_ctx, const char *key, + const char *value); /* Unrefs the slices. */ -grpc_mdelem *grpc_mdelem_from_slices(grpc_slice key, grpc_slice value); -grpc_mdelem *grpc_mdelem_from_string_and_buffer(const char *key, +grpc_mdelem *grpc_mdelem_from_slices(grpc_exec_ctx *exec_ctx, grpc_slice key, + grpc_slice value); +grpc_mdelem *grpc_mdelem_from_string_and_buffer(grpc_exec_ctx *exec_ctx, + const char *key, const uint8_t *value, size_t value_length); @@ -127,22 +131,26 @@ void grpc_mdelem_set_user_data(grpc_mdelem *md, void (*destroy_func)(void *), //#define GRPC_METADATA_REFCOUNT_DEBUG #ifdef GRPC_METADATA_REFCOUNT_DEBUG #define GRPC_MDSTR_REF(s) grpc_mdstr_ref((s), __FILE__, __LINE__) -#define GRPC_MDSTR_UNREF(s) grpc_mdstr_unref((s), __FILE__, __LINE__) +#define GRPC_MDSTR_UNREF(exec_ctx, s) \ + grpc_mdstr_unref((exec_ctx), (s), __FILE__, __LINE__) #define GRPC_MDELEM_REF(s) grpc_mdelem_ref((s), __FILE__, __LINE__) -#define GRPC_MDELEM_UNREF(s) grpc_mdelem_unref((s), __FILE__, __LINE__) +#define GRPC_MDELEM_UNREF(exec_ctx, s) \ + grpc_mdelem_unref((exec_ctx), (s), __FILE__, __LINE__) grpc_mdstr *grpc_mdstr_ref(grpc_mdstr *s, const char *file, int line); -void grpc_mdstr_unref(grpc_mdstr *s, const char *file, int line); +void grpc_mdstr_unref(grpc_exec_ctx *exec_ctx, grpc_mdstr *s, const char *file, + int line); grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *md, const char *file, int line); -void grpc_mdelem_unref(grpc_mdelem *md, const char *file, int line); +void grpc_mdelem_unref(grpc_exec_ctx *exec_ctx, grpc_mdelem *md, + const char *file, int line); #else #define GRPC_MDSTR_REF(s) grpc_mdstr_ref((s)) -#define GRPC_MDSTR_UNREF(s) grpc_mdstr_unref((s)) +#define GRPC_MDSTR_UNREF(exec_ctx, s) grpc_mdstr_unref((exec_ctx), (s)) #define GRPC_MDELEM_REF(s) grpc_mdelem_ref((s)) -#define GRPC_MDELEM_UNREF(s) grpc_mdelem_unref((s)) +#define GRPC_MDELEM_UNREF(exec_ctx, s) grpc_mdelem_unref((exec_ctx), (s)) grpc_mdstr *grpc_mdstr_ref(grpc_mdstr *s); -void grpc_mdstr_unref(grpc_mdstr *s); +void grpc_mdstr_unref(grpc_exec_ctx *exec_ctx, grpc_mdstr *s); grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *md); -void grpc_mdelem_unref(grpc_mdelem *md); +void grpc_mdelem_unref(grpc_exec_ctx *exec_ctx, grpc_mdelem *md); #endif /* Recover a char* from a grpc_mdstr. The returned string is null terminated. @@ -162,7 +170,7 @@ int grpc_mdstr_is_bin_suffixed(grpc_mdstr *s); #define GRPC_MDSTR_KV_HASH(k_hash, v_hash) (GPR_ROTL((k_hash), 2) ^ (v_hash)) void grpc_mdctx_global_init(void); -void grpc_mdctx_global_shutdown(void); +void grpc_mdctx_global_shutdown(grpc_exec_ctx *exec_ctx); /* Implementation provided by chttp2_transport */ extern grpc_slice (*grpc_chttp2_base64_encode_and_huffman_compress)( diff --git a/src/core/lib/transport/metadata_batch.c b/src/core/lib/transport/metadata_batch.c index 84b5a74d51..4430224e70 100644 --- a/src/core/lib/transport/metadata_batch.c +++ b/src/core/lib/transport/metadata_batch.c @@ -72,10 +72,11 @@ void grpc_metadata_batch_init(grpc_metadata_batch *batch) { batch->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); } -void grpc_metadata_batch_destroy(grpc_metadata_batch *batch) { +void grpc_metadata_batch_destroy(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch) { grpc_linked_mdelem *l; for (l = batch->list.head; l; l = l->next) { - GRPC_MDELEM_UNREF(l->md); + GRPC_MDELEM_UNREF(exec_ctx, l->md); } } @@ -140,7 +141,8 @@ void grpc_metadata_batch_move(grpc_metadata_batch *dst, memset(src, 0, sizeof(grpc_metadata_batch)); } -void grpc_metadata_batch_filter(grpc_metadata_batch *batch, +void grpc_metadata_batch_filter(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch, grpc_mdelem *(*filter)(void *user_data, grpc_mdelem *elem), void *user_data) { @@ -168,9 +170,9 @@ void grpc_metadata_batch_filter(grpc_metadata_batch *batch, batch->list.tail = l->prev; } assert_valid_list(&batch->list); - GRPC_MDELEM_UNREF(l->md); + GRPC_MDELEM_UNREF(exec_ctx, l->md); } else if (filt != orig) { - GRPC_MDELEM_UNREF(orig); + GRPC_MDELEM_UNREF(exec_ctx, orig); l->md = filt; } } @@ -183,9 +185,10 @@ static grpc_mdelem *no_metadata_for_you(void *user_data, grpc_mdelem *elem) { return NULL; } -void grpc_metadata_batch_clear(grpc_metadata_batch *batch) { +void grpc_metadata_batch_clear(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch) { batch->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); - grpc_metadata_batch_filter(batch, no_metadata_for_you, NULL); + grpc_metadata_batch_filter(exec_ctx, batch, no_metadata_for_you, NULL); } bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) { diff --git a/src/core/lib/transport/metadata_batch.h b/src/core/lib/transport/metadata_batch.h index 7a9ccb4bc8..862c21b45b 100644 --- a/src/core/lib/transport/metadata_batch.h +++ b/src/core/lib/transport/metadata_batch.h @@ -68,8 +68,10 @@ typedef struct grpc_metadata_batch { } grpc_metadata_batch; void grpc_metadata_batch_init(grpc_metadata_batch *batch); -void grpc_metadata_batch_destroy(grpc_metadata_batch *batch); -void grpc_metadata_batch_clear(grpc_metadata_batch *batch); +void grpc_metadata_batch_destroy(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch); +void grpc_metadata_batch_clear(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch); bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch); /* Returns the transport size of the batch. */ @@ -118,7 +120,8 @@ void grpc_metadata_batch_add_tail(grpc_metadata_batch *batch, The return value from \a filter will be substituted for the grpc_mdelem passed to \a filter. If \a filter returns NULL, the element will be moved to the garbage list. */ -void grpc_metadata_batch_filter(grpc_metadata_batch *batch, +void grpc_metadata_batch_filter(grpc_exec_ctx *exec_ctx, + grpc_metadata_batch *batch, grpc_mdelem *(*filter)(void *user_data, grpc_mdelem *elem), void *user_data); diff --git a/src/core/lib/transport/method_config.c b/src/core/lib/transport/method_config.c index 57d97700bf..25fb54b37d 100644 --- a/src/core/lib/transport/method_config.c +++ b/src/core/lib/transport/method_config.c @@ -63,7 +63,9 @@ static int bool_cmp(void* v1, void* v2) { return 0; } -static grpc_mdstr_hash_table_vtable bool_vtable = {gpr_free, bool_copy, +static void free_mem(grpc_exec_ctx* exec_ctx, void* p) { gpr_free(p); } + +static grpc_mdstr_hash_table_vtable bool_vtable = {free_mem, bool_copy, bool_cmp}; // timespec vtable @@ -79,7 +81,7 @@ static int timespec_cmp(void* v1, void* v2) { return gpr_time_cmp(*(gpr_timespec*)v1, *(gpr_timespec*)v2); } -static grpc_mdstr_hash_table_vtable timespec_vtable = {gpr_free, timespec_copy, +static grpc_mdstr_hash_table_vtable timespec_vtable = {free_mem, timespec_copy, timespec_cmp}; // int32 vtable @@ -99,7 +101,7 @@ static int int32_cmp(void* v1, void* v2) { return 0; } -static grpc_mdstr_hash_table_vtable int32_vtable = {gpr_free, int32_copy, +static grpc_mdstr_hash_table_vtable int32_vtable = {free_mem, int32_copy, int32_cmp}; // Hash table keys. @@ -166,12 +168,13 @@ grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config) { return method_config; } -void grpc_method_config_unref(grpc_method_config* method_config) { - if (grpc_mdstr_hash_table_unref(method_config->table)) { - GRPC_MDSTR_UNREF(method_config->wait_for_ready_key); - GRPC_MDSTR_UNREF(method_config->timeout_key); - GRPC_MDSTR_UNREF(method_config->max_request_message_bytes_key); - GRPC_MDSTR_UNREF(method_config->max_response_message_bytes_key); +void grpc_method_config_unref(grpc_exec_ctx* exec_ctx, + grpc_method_config* method_config) { + if (grpc_mdstr_hash_table_unref(exec_ctx, method_config->table)) { + GRPC_MDSTR_UNREF(exec_ctx, method_config->wait_for_ready_key); + GRPC_MDSTR_UNREF(exec_ctx, method_config->timeout_key); + GRPC_MDSTR_UNREF(exec_ctx, method_config->max_request_message_bytes_key); + GRPC_MDSTR_UNREF(exec_ctx, method_config->max_response_message_bytes_key); gpr_free(method_config); } } @@ -210,8 +213,8 @@ const int32_t* grpc_method_config_get_max_response_message_bytes( // grpc_method_config_table // -static void method_config_unref(void* valuep) { - grpc_method_config_unref(valuep); +static void method_config_unref(grpc_exec_ctx* exec_ctx, void* valuep) { + grpc_method_config_unref(exec_ctx, valuep); } static void* method_config_ref(void* valuep) { @@ -245,8 +248,9 @@ grpc_method_config_table* grpc_method_config_table_ref( return grpc_mdstr_hash_table_ref(table); } -void grpc_method_config_table_unref(grpc_method_config_table* table) { - grpc_mdstr_hash_table_unref(table); +void grpc_method_config_table_unref(grpc_exec_ctx* exec_ctx, + grpc_method_config_table* table) { + grpc_mdstr_hash_table_unref(exec_ctx, table); } int grpc_method_config_table_cmp(const grpc_method_config_table* table1, @@ -254,7 +258,8 @@ int grpc_method_config_table_cmp(const grpc_method_config_table* table1, return grpc_mdstr_hash_table_cmp(table1, table2); } -void* grpc_method_config_table_get(const grpc_mdstr_hash_table* table, +void* grpc_method_config_table_get(grpc_exec_ctx* exec_ctx, + const grpc_mdstr_hash_table* table, const grpc_mdstr* path) { void* value = grpc_mdstr_hash_table_get(table, path); // If we didn't find a match for the path, try looking for a wildcard @@ -270,14 +275,16 @@ void* grpc_method_config_table_get(const grpc_mdstr_hash_table* table, grpc_mdstr* wildcard_path = grpc_mdstr_from_string(buf); gpr_free(buf); value = grpc_mdstr_hash_table_get(table, wildcard_path); - GRPC_MDSTR_UNREF(wildcard_path); + GRPC_MDSTR_UNREF(exec_ctx, wildcard_path); } return value; } static void* copy_arg(void* p) { return grpc_method_config_table_ref(p); } -static void destroy_arg(void* p) { grpc_method_config_table_unref(p); } +static void destroy_arg(grpc_exec_ctx* exec_ctx, void* p) { + grpc_method_config_table_unref(exec_ctx, p); +} static int cmp_arg(void* p1, void* p2) { return grpc_method_config_table_cmp(p1, p2); @@ -315,7 +322,7 @@ static void convert_entry(const grpc_mdstr_hash_table_entry* entry, } grpc_mdstr_hash_table* grpc_method_config_table_convert( - const grpc_method_config_table* table, + grpc_exec_ctx* exec_ctx, const grpc_method_config_table* table, void* (*convert_value)(const grpc_method_config* method_config), const grpc_mdstr_hash_table_vtable* vtable) { // Create an array of the entries in the table with converted values. @@ -331,8 +338,8 @@ grpc_mdstr_hash_table* grpc_method_config_table_convert( grpc_mdstr_hash_table_create(state.num_entries, state.entries); // Clean up the array. for (size_t i = 0; i < state.num_entries; ++i) { - GRPC_MDSTR_UNREF(state.entries[i].key); - vtable->destroy_value(state.entries[i].value); + GRPC_MDSTR_UNREF(exec_ctx, state.entries[i].key); + vtable->destroy_value(exec_ctx, state.entries[i].value); } gpr_free(state.entries); // Return the new table. diff --git a/src/core/lib/transport/method_config.h b/src/core/lib/transport/method_config.h index 58fedd9436..d17a493fd4 100644 --- a/src/core/lib/transport/method_config.h +++ b/src/core/lib/transport/method_config.h @@ -60,7 +60,8 @@ grpc_method_config* grpc_method_config_create( int32_t* max_request_message_bytes, int32_t* max_response_message_bytes); grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config); -void grpc_method_config_unref(grpc_method_config* method_config); +void grpc_method_config_unref(grpc_exec_ctx* exec_ctx, + grpc_method_config* method_config); /// Compares two grpc_method_configs. /// The sort order is stable but undefined. @@ -95,7 +96,8 @@ grpc_method_config_table* grpc_method_config_table_create( grpc_method_config_table* grpc_method_config_table_ref( grpc_method_config_table* table); -void grpc_method_config_table_unref(grpc_method_config_table* table); +void grpc_method_config_table_unref(grpc_exec_ctx* exec_ctx, + grpc_method_config_table* table); /// Compares two grpc_method_config_tables. /// The sort order is stable but undefined. @@ -110,7 +112,8 @@ int grpc_method_config_table_cmp(const grpc_method_config_table* table1, /// Note: This returns a void* instead of a grpc_method_config* so that /// it can also be used for tables constructed via /// grpc_method_config_table_convert(). -void* grpc_method_config_table_get(const grpc_mdstr_hash_table* table, +void* grpc_method_config_table_get(grpc_exec_ctx* exec_ctx, + const grpc_mdstr_hash_table* table, const grpc_mdstr* path); /// Returns a channel arg containing \a table. @@ -129,7 +132,7 @@ grpc_arg grpc_method_config_table_create_channel_arg( /// the grpc_method_config, and \a vtable provides the methods for /// operating on the struct type. grpc_mdstr_hash_table* grpc_method_config_table_convert( - const grpc_method_config_table* table, + grpc_exec_ctx* exec_ctx, const grpc_method_config_table* table, void* (*convert_value)(const grpc_method_config* method_config), const grpc_mdstr_hash_table_vtable* vtable); diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 866cd9ea87..1b79520e68 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -40,6 +40,7 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/transport_impl.h" @@ -207,12 +208,12 @@ void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op, } void grpc_transport_stream_op_add_cancellation_with_message( - grpc_transport_stream_op *op, grpc_status_code status, - grpc_slice *optional_message) { + grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op, + grpc_status_code status, grpc_slice *optional_message) { GPR_ASSERT(status != GRPC_STATUS_OK); if (op->cancel_error != GRPC_ERROR_NONE) { if (optional_message) { - grpc_slice_unref(*optional_message); + grpc_slice_unref_internal(exec_ctx, *optional_message); } return; } @@ -222,7 +223,7 @@ void grpc_transport_stream_op_add_cancellation_with_message( error = grpc_error_set_str(GRPC_ERROR_CREATE(msg), GRPC_ERROR_STR_GRPC_MESSAGE, msg); gpr_free(msg); - grpc_slice_unref(*optional_message); + grpc_slice_unref_internal(exec_ctx, *optional_message); } else { error = GRPC_ERROR_CREATE("Call cancelled"); } @@ -230,14 +231,15 @@ void grpc_transport_stream_op_add_cancellation_with_message( add_error(op, &op->cancel_error, error); } -void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, +void grpc_transport_stream_op_add_close(grpc_exec_ctx *exec_ctx, + grpc_transport_stream_op *op, grpc_status_code status, grpc_slice *optional_message) { GPR_ASSERT(status != GRPC_STATUS_OK); if (op->cancel_error != GRPC_ERROR_NONE || op->close_error != GRPC_ERROR_NONE) { if (optional_message) { - grpc_slice_unref(*optional_message); + grpc_slice_unref_internal(exec_ctx, *optional_message); } return; } @@ -247,7 +249,7 @@ void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, error = grpc_error_set_str(GRPC_ERROR_CREATE(msg), GRPC_ERROR_STR_GRPC_MESSAGE, msg); gpr_free(msg); - grpc_slice_unref(*optional_message); + grpc_slice_unref_internal(exec_ctx, *optional_message); } else { error = GRPC_ERROR_CREATE("Call force closed"); } diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index 8916b28b72..3e38d98f28 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -248,10 +248,11 @@ void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op, grpc_status_code status); void grpc_transport_stream_op_add_cancellation_with_message( - grpc_transport_stream_op *op, grpc_status_code status, - grpc_slice *optional_message); + grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op, + grpc_status_code status, grpc_slice *optional_message); -void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, +void grpc_transport_stream_op_add_close(grpc_exec_ctx *exec_ctx, + grpc_transport_stream_op *op, grpc_status_code status, grpc_slice *optional_message); diff --git a/test/core/bad_client/bad_client.c b/test/core/bad_client/bad_client.c index 07fcd995d7..126ea54b69 100644 --- a/test/core/bad_client/bad_client.c +++ b/test/core/bad_client/bad_client.c @@ -117,7 +117,7 @@ void grpc_run_bad_client_test( grpc_resource_quota *resource_quota = grpc_resource_quota_create("bad_client_test"); sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 65536); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); /* Create server, completion events */ a.server = grpc_server_create(NULL, NULL); @@ -181,7 +181,7 @@ void grpc_run_bad_client_test( grpc_exec_ctx_finish(&exec_ctx); GPR_ASSERT( gpr_event_wait(&args.read_done, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5))); - grpc_slice_buffer_destroy(&args.incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &args.incoming); } // Shutdown. grpc_endpoint_shutdown(&exec_ctx, sfd.client); @@ -194,7 +194,7 @@ void grpc_run_bad_client_test( .type == GRPC_OP_COMPLETE); grpc_server_destroy(a.server); grpc_completion_queue_destroy(a.cq); - grpc_slice_buffer_destroy(&outgoing); + grpc_slice_buffer_destroy_internal(exec_ctx, &outgoing); grpc_exec_ctx_finish(&exec_ctx); grpc_shutdown(); diff --git a/test/core/bad_client/tests/large_metadata.c b/test/core/bad_client/tests/large_metadata.c index 9c804e78c1..809bbe4094 100644 --- a/test/core/bad_client/tests/large_metadata.c +++ b/test/core/bad_client/tests/large_metadata.c @@ -213,7 +213,7 @@ static void client_validator(grpc_slice_buffer *incoming) { *p++ = 11; // Compare actual and expected. GPR_ASSERT(grpc_slice_cmp(last_frame, expected) == 0); - grpc_slice_buffer_destroy(&last_frame_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &last_frame_buffer); } int main(int argc, char **argv) { diff --git a/test/core/client_channel/set_initial_connect_string_test.c b/test/core/client_channel/set_initial_connect_string_test.c index a10d28b30f..d8eca036ed 100644 --- a/test/core/client_channel/set_initial_connect_string_test.c +++ b/test/core/client_channel/set_initial_connect_string_test.c @@ -155,8 +155,8 @@ static void start_rpc(int use_creds, int target_port) { static void cleanup_rpc(void) { grpc_event ev; - grpc_slice_buffer_destroy(&state.incoming_buffer); - grpc_slice_buffer_destroy(&state.temp_incoming_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.incoming_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.temp_incoming_buffer); grpc_channel_credentials_unref(state.creds); grpc_call_destroy(state.call); grpc_completion_queue_shutdown(state.cq); diff --git a/test/core/compression/message_compress_test.c b/test/core/compression/message_compress_test.c index fc53cd9d36..ee4f0dbe40 100644 --- a/test/core/compression/message_compress_test.c +++ b/test/core/compression/message_compress_test.c @@ -105,10 +105,10 @@ static void assert_passthrough(grpc_slice value, final = grpc_slice_merge(output.slices, output.count); GPR_ASSERT(0 == grpc_slice_cmp(value, final)); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&compressed); - grpc_slice_buffer_destroy(&compressed_raw); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &compressed); + grpc_slice_buffer_destroy_internal(exec_ctx, &compressed_raw); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); grpc_slice_unref(final); } @@ -164,8 +164,8 @@ static void test_tiny_data_compress(void) { GPR_ASSERT(1 == output.count); } - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } static void test_bad_decompression_data_crc(void) { @@ -191,9 +191,9 @@ static void test_bad_decompression_data_crc(void) { /* try (and fail) to decompress the corrupted compresed buffer */ GPR_ASSERT(0 == grpc_msg_decompress(GRPC_COMPRESS_GZIP, &corrupted, &output)); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&corrupted); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &corrupted); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } static void test_bad_decompression_data_trailing_garbage(void) { @@ -210,8 +210,8 @@ static void test_bad_decompression_data_trailing_garbage(void) { /* try (and fail) to decompress the invalid compresed buffer */ GPR_ASSERT(0 == grpc_msg_decompress(GRPC_COMPRESS_DEFLATE, &input, &output)); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } static void test_bad_decompression_data_stream(void) { @@ -226,8 +226,8 @@ static void test_bad_decompression_data_stream(void) { /* try (and fail) to decompress the invalid compresed buffer */ GPR_ASSERT(0 == grpc_msg_decompress(GRPC_COMPRESS_DEFLATE, &input, &output)); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } static void test_bad_compression_algorithm(void) { @@ -247,8 +247,8 @@ static void test_bad_compression_algorithm(void) { grpc_msg_compress(GRPC_COMPRESS_ALGORITHMS_COUNT + 123, &input, &output); GPR_ASSERT(0 == was_compressed); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } static void test_bad_decompression_algorithm(void) { @@ -269,8 +269,8 @@ static void test_bad_decompression_algorithm(void) { &input, &output); GPR_ASSERT(0 == was_decompressed); - grpc_slice_buffer_destroy(&input); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &input); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); } int main(int argc, char **argv) { diff --git a/test/core/end2end/bad_server_response_test.c b/test/core/end2end/bad_server_response_test.c index 1c4a17fda8..9dc70d79ec 100644 --- a/test/core/end2end/bad_server_response_test.c +++ b/test/core/end2end/bad_server_response_test.c @@ -228,8 +228,8 @@ static void start_rpc(int target_port, grpc_status_code expected_status, static void cleanup_rpc(void) { grpc_event ev; - grpc_slice_buffer_destroy(&state.temp_incoming_buffer); - grpc_slice_buffer_destroy(&state.outgoing_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.temp_incoming_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.outgoing_buffer); grpc_call_destroy(state.call); grpc_completion_queue_shutdown(state.cq); do { diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index 11e8604f56..dc03861f86 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -145,7 +145,7 @@ void test_connect(const char *server_host, const char *client_host, int port, gpr_free(hosts_with_port[i]); } gpr_free(hosts_with_port); - grpc_slice_buffer_destroy(&uri_parts); + grpc_slice_buffer_destroy_internal(exec_ctx, &uri_parts); grpc_slice_unref(uri_slice); } else { gpr_join_host_port(&client_hostport, client_host, port); diff --git a/test/core/end2end/fake_resolver.c b/test/core/end2end/fake_resolver.c index 32856a5db9..e6cdaf7bfe 100644 --- a/test/core/end2end/fake_resolver.c +++ b/test/core/end2end/fake_resolver.c @@ -189,7 +189,7 @@ static grpc_resolver* fake_resolver_create(grpc_resolver_factory* factory, addresses->addresses[i].is_balancer = lb_enabled; if (errors_found) break; } - grpc_slice_buffer_destroy(&path_parts); + grpc_slice_buffer_destroy_internal(exec_ctx, &path_parts); grpc_slice_unref(path_slice); if (errors_found) { grpc_lb_addresses_destroy(addresses); diff --git a/test/core/end2end/fixtures/http_proxy.c b/test/core/end2end/fixtures/http_proxy.c index 57fc4a38f8..9c808209e3 100644 --- a/test/core/end2end/fixtures/http_proxy.c +++ b/test/core/end2end/fixtures/http_proxy.c @@ -110,12 +110,12 @@ static void proxy_connection_unref(grpc_exec_ctx* exec_ctx, if (conn->server_endpoint != NULL) grpc_endpoint_destroy(exec_ctx, conn->server_endpoint); grpc_pollset_set_destroy(conn->pollset_set); - grpc_slice_buffer_destroy(&conn->client_read_buffer); - grpc_slice_buffer_destroy(&conn->client_deferred_write_buffer); - grpc_slice_buffer_destroy(&conn->client_write_buffer); - grpc_slice_buffer_destroy(&conn->server_read_buffer); - grpc_slice_buffer_destroy(&conn->server_deferred_write_buffer); - grpc_slice_buffer_destroy(&conn->server_write_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_deferred_write_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_write_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_deferred_write_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_write_buffer); grpc_http_parser_destroy(&conn->http_parser); grpc_http_request_destroy(&conn->http_request); gpr_free(conn); diff --git a/test/core/end2end/fuzzers/client_fuzzer.c b/test/core/end2end/fuzzers/client_fuzzer.c index c5260cd287..26b520885b 100644 --- a/test/core/end2end/fuzzers/client_fuzzer.c +++ b/test/core/end2end/fuzzers/client_fuzzer.c @@ -62,7 +62,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_resource_quota_create("client_fuzzer"); grpc_endpoint *mock_endpoint = grpc_mock_endpoint_create(discard_write, resource_quota); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_transport *transport = diff --git a/test/core/end2end/fuzzers/server_fuzzer.c b/test/core/end2end/fuzzers/server_fuzzer.c index 164022ec79..115fb06925 100644 --- a/test/core/end2end/fuzzers/server_fuzzer.c +++ b/test/core/end2end/fuzzers/server_fuzzer.c @@ -60,7 +60,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_resource_quota_create("server_fuzzer"); grpc_endpoint *mock_endpoint = grpc_mock_endpoint_create(discard_write, resource_quota); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_mock_endpoint_put_read( &exec_ctx, mock_endpoint, grpc_slice_from_copied_buffer((const char *)data, size)); diff --git a/test/core/http/httpcli_test.c b/test/core/http/httpcli_test.c index 3e312c1dde..57b779cccf 100644 --- a/test/core/http/httpcli_test.c +++ b/test/core/http/httpcli_test.c @@ -93,7 +93,7 @@ static void test_get(int port) { grpc_httpcli_get(&exec_ctx, &g_context, &g_pops, resource_quota, &req, n_seconds_time(15), grpc_closure_create(on_finish, &response), &response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_mu_lock(g_mu); while (!g_done) { grpc_pollset_worker *worker = NULL; @@ -133,7 +133,7 @@ static void test_post(int port) { grpc_httpcli_post(&exec_ctx, &g_context, &g_pops, resource_quota, &req, "hello", 5, n_seconds_time(15), grpc_closure_create(on_finish, &response), &response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_mu_lock(g_mu); while (!g_done) { grpc_pollset_worker *worker = NULL; diff --git a/test/core/http/httpscli_test.c b/test/core/http/httpscli_test.c index d06035149e..765ae101c5 100644 --- a/test/core/http/httpscli_test.c +++ b/test/core/http/httpscli_test.c @@ -94,7 +94,7 @@ static void test_get(int port) { grpc_httpcli_get(&exec_ctx, &g_context, &g_pops, resource_quota, &req, n_seconds_time(15), grpc_closure_create(on_finish, &response), &response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_mu_lock(g_mu); while (!g_done) { grpc_pollset_worker *worker = NULL; @@ -135,7 +135,7 @@ static void test_post(int port) { grpc_httpcli_post(&exec_ctx, &g_context, &g_pops, resource_quota, &req, "hello", 5, n_seconds_time(15), grpc_closure_create(on_finish, &response), &response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_mu_lock(g_mu); while (!g_done) { grpc_pollset_worker *worker = NULL; diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c index 8186ea7e85..09a1e611f4 100644 --- a/test/core/iomgr/endpoint_tests.c +++ b/test/core/iomgr/endpoint_tests.c @@ -249,8 +249,8 @@ static void read_and_write_test(grpc_endpoint_test_config config, grpc_exec_ctx_flush(&exec_ctx); end_test(config); - grpc_slice_buffer_destroy(&state.outgoing); - grpc_slice_buffer_destroy(&state.incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.outgoing); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.incoming); grpc_endpoint_destroy(&exec_ctx, state.read_ep); grpc_endpoint_destroy(&exec_ctx, state.write_ep); grpc_exec_ctx_finish(&exec_ctx); @@ -304,7 +304,7 @@ static void multiple_shutdown_test(grpc_endpoint_test_config config) { grpc_endpoint_shutdown(&exec_ctx, f.client_ep); wait_for_fail_count(&exec_ctx, &fail_count, 3); - grpc_slice_buffer_destroy(&slice_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &slice_buffer); grpc_endpoint_destroy(&exec_ctx, f.client_ep); grpc_endpoint_destroy(&exec_ctx, f.server_ep); diff --git a/test/core/iomgr/resource_quota_test.c b/test/core/iomgr/resource_quota_test.c index 6dc9f8b76a..8ba8af26a8 100644 --- a/test/core/iomgr/resource_quota_test.c +++ b/test/core/iomgr/resource_quota_test.c @@ -672,7 +672,7 @@ static void test_one_slice(void) { GPR_ASSERT(num_allocs == start_allocs + 1); } - grpc_slice_buffer_destroy(&buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &buffer); destroy_user(&usr); grpc_resource_quota_unref(q); } @@ -712,7 +712,7 @@ static void test_one_slice_deleted_late(void) { } grpc_resource_quota_unref(q); - grpc_slice_buffer_destroy(&buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &buffer); GPR_ASSERT(done); { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c index 5eafa570bb..81b9ef5dff 100644 --- a/test/core/iomgr/tcp_posix_test.c +++ b/test/core/iomgr/tcp_posix_test.c @@ -184,7 +184,7 @@ static void read_test(size_t num_bytes, size_t slice_size) { grpc_resource_quota *resource_quota = grpc_resource_quota_create("read_test"); ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), resource_quota, slice_size, "test"); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket_partial(sv[0], num_bytes); @@ -212,7 +212,7 @@ static void read_test(size_t num_bytes, size_t slice_size) { GPR_ASSERT(state.read_bytes == state.target_read_bytes); gpr_mu_unlock(g_mu); - grpc_slice_buffer_destroy(&state.incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.incoming); grpc_endpoint_destroy(&exec_ctx, ep); grpc_exec_ctx_finish(&exec_ctx); } @@ -235,7 +235,7 @@ static void large_read_test(size_t slice_size) { grpc_resource_quota_create("large_read_test"); ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), resource_quota, slice_size, "test"); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket(sv[0]); @@ -263,7 +263,7 @@ static void large_read_test(size_t slice_size) { GPR_ASSERT(state.read_bytes == state.target_read_bytes); gpr_mu_unlock(g_mu); - grpc_slice_buffer_destroy(&state.incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.incoming); grpc_endpoint_destroy(&exec_ctx, ep); grpc_exec_ctx_finish(&exec_ctx); } @@ -374,7 +374,7 @@ static void write_test(size_t num_bytes, size_t slice_size) { grpc_resource_quota_create("write_test"); ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"), resource_quota, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, "test"); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); state.ep = ep; @@ -404,7 +404,7 @@ static void write_test(size_t num_bytes, size_t slice_size) { } gpr_mu_unlock(g_mu); - grpc_slice_buffer_destroy(&outgoing); + grpc_slice_buffer_destroy_internal(exec_ctx, &outgoing); grpc_endpoint_destroy(&exec_ctx, ep); gpr_free(slices); grpc_exec_ctx_finish(&exec_ctx); @@ -442,7 +442,7 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) { ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), resource_quota, slice_size, "test"); GPR_ASSERT(grpc_tcp_fd(ep) == sv[1] && sv[1] >= 0); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket_partial(sv[0], num_bytes); @@ -472,7 +472,7 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) { GPR_ASSERT(state.read_bytes == state.target_read_bytes); gpr_mu_unlock(g_mu); - grpc_slice_buffer_destroy(&state.incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &state.incoming); grpc_tcp_destroy_and_release_fd(&exec_ctx, ep, &fd, &fd_released_cb); grpc_exec_ctx_flush(&exec_ctx); gpr_mu_lock(g_mu); @@ -534,7 +534,7 @@ static grpc_endpoint_test_fixture create_fixture_tcp_socketpair( resource_quota, slice_size, "test"); f.server_ep = grpc_tcp_create(grpc_fd_create(sv[1], "fixture:server"), resource_quota, slice_size, "test"); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, f.client_ep, g_pollset); grpc_endpoint_add_to_pollset(&exec_ctx, f.server_ep, g_pollset); diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c index b5d95004fe..e49662d428 100644 --- a/test/core/security/secure_endpoint_test.c +++ b/test/core/security/secure_endpoint_test.c @@ -59,7 +59,7 @@ static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair( grpc_resource_quota *resource_quota = grpc_resource_quota_create("secure_endpoint_test"); tcp = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, slice_size); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_endpoint_add_to_pollset(&exec_ctx, tcp.client, g_pollset); grpc_endpoint_add_to_pollset(&exec_ctx, tcp.server, g_pollset); @@ -171,7 +171,7 @@ static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) { grpc_endpoint_destroy(&exec_ctx, f.server_ep); grpc_exec_ctx_finish(&exec_ctx); grpc_slice_unref(s); - grpc_slice_buffer_destroy(&incoming); + grpc_slice_buffer_destroy_internal(exec_ctx, &incoming); clean_up(); } diff --git a/test/core/slice/slice_buffer_test.c b/test/core/slice/slice_buffer_test.c index bf9ae197d2..e5ef3047e5 100644 --- a/test/core/slice/slice_buffer_test.c +++ b/test/core/slice/slice_buffer_test.c @@ -68,7 +68,7 @@ void test_slice_buffer_add() { } GPR_ASSERT(buf.count == 0); GPR_ASSERT(buf.length == 0); - grpc_slice_buffer_destroy(&buf); + grpc_slice_buffer_destroy_internal(exec_ctx, &buf); } void test_slice_buffer_move_first() { diff --git a/test/core/surface/byte_buffer_reader_test.c b/test/core/surface/byte_buffer_reader_test.c index d8d7a52d15..0d1628821b 100644 --- a/test/core/surface/byte_buffer_reader_test.c +++ b/test/core/surface/byte_buffer_reader_test.c @@ -162,8 +162,8 @@ static void read_compressed_slice(grpc_compression_algorithm algorithm, GPR_ASSERT(read_count == input_size); grpc_byte_buffer_reader_destroy(&reader); grpc_byte_buffer_destroy(buffer); - grpc_slice_buffer_destroy(&sliceb_out); - grpc_slice_buffer_destroy(&sliceb_in); + grpc_slice_buffer_destroy_internal(exec_ctx, &sliceb_out); + grpc_slice_buffer_destroy_internal(exec_ctx, &sliceb_in); } static void test_read_gzip_compressed_slice(void) { diff --git a/test/core/transport/chttp2/hpack_encoder_test.c b/test/core/transport/chttp2/hpack_encoder_test.c index 91421e18f4..6f1a1f7223 100644 --- a/test/core/transport/chttp2/hpack_encoder_test.c +++ b/test/core/transport/chttp2/hpack_encoder_test.c @@ -101,7 +101,7 @@ static void verify(size_t window_available, int eof, size_t expect_window_used, grpc_chttp2_encode_header(&g_compressor, 0xdeadbeef, &b, eof, 16384, &stats, &output); merged = grpc_slice_merge(output.slices, output.count); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); grpc_metadata_batch_destroy(&b); if (0 != grpc_slice_cmp(merged, expect)) { @@ -205,7 +205,7 @@ static void verify_table_size_change_match_elem_size(const char *key, memset(&stats, 0, sizeof(stats)); grpc_chttp2_encode_header(&g_compressor, 0xdeadbeef, &b, 0, 16384, &stats, &output); - grpc_slice_buffer_destroy(&output); + grpc_slice_buffer_destroy_internal(exec_ctx, &output); grpc_metadata_batch_destroy(&b); GPR_ASSERT(g_compressor.table_size == elem_size + initial_table_size); diff --git a/test/core/util/mock_endpoint.c b/test/core/util/mock_endpoint.c index 6e09427c2b..1d33566221 100644 --- a/test/core/util/mock_endpoint.c +++ b/test/core/util/mock_endpoint.c @@ -82,7 +82,7 @@ static void unref(grpc_exec_ctx *exec_ctx, grpc_mock_endpoint *m) { gpr_mu_lock(&m->mu); if (0 == --m->refs) { gpr_mu_unlock(&m->mu); - grpc_slice_buffer_destroy(&m->read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &m->read_buffer); grpc_resource_user_destroy(exec_ctx, &m->resource_user); gpr_free(m); } else { diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c index a5bac8aaa8..db10de87d9 100644 --- a/test/core/util/passthru_endpoint.c +++ b/test/core/util/passthru_endpoint.c @@ -132,8 +132,8 @@ static void me_really_destroy(grpc_exec_ctx *exec_ctx, void *ep, if (0 == --p->halves) { gpr_mu_unlock(&p->mu); gpr_mu_destroy(&p->mu); - grpc_slice_buffer_destroy(&p->client.read_buffer); - grpc_slice_buffer_destroy(&p->server.read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &p->client.read_buffer); + grpc_slice_buffer_destroy_internal(exec_ctx, &p->server.read_buffer); gpr_free(p); } else { gpr_mu_unlock(&p->mu); diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c index b2342feeb4..384a158c47 100644 --- a/test/core/util/port_server_client.c +++ b/test/core/util/port_server_client.c @@ -104,7 +104,7 @@ void grpc_free_port_using_server(char *server, int port) { grpc_httpcli_get(&exec_ctx, &context, &pr.pops, resource_quota, &req, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), grpc_closure_create(freed_port_from_server, &pr), &rsp); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); gpr_mu_lock(pr.mu); while (!pr.done) { grpc_pollset_worker *worker = NULL; @@ -176,7 +176,7 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), grpc_closure_create(got_port_from_server, pr), &pr->response); - grpc_resource_quota_internal_unref(exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(exec_ctx, resource_quota); return; } GPR_ASSERT(response); @@ -223,7 +223,7 @@ int grpc_pick_port_using_server(char *server) { GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), grpc_closure_create(got_port_from_server, &pr), &pr.response); - grpc_resource_quota_internal_unref(&exec_ctx, resource_quota); + grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(pr.mu); while (pr.port == -1) { diff --git a/tools/run_tests/sanity/core_banned_functions.py b/tools/run_tests/sanity/core_banned_functions.py new file mode 100755 index 0000000000..a3d4d6337e --- /dev/null +++ b/tools/run_tests/sanity/core_banned_functions.py @@ -0,0 +1,32 @@ +#!/usr/bin/python + +import os +import sys + +os.chdir(os.path.join(os.path.dirname(sys.argv[0]), '../../..')) + +# map of banned function signature to whitelist +BANNED_EXCEPT = { + 'grpc_resource_quota_ref(': ('src/core/lib/iomgr/resource_quota.c'), + 'grpc_resource_quota_unref(': ('src/core/lib/iomgr/resource_quota.c'), + 'grpc_slice_buffer_destroy(': ('src/core/lib/slice/slice_buffer.c'), + 'grpc_slice_buffer_reset_and_unref(': ('src/core/lib/slice/slice_buffer.c'), + 'grpc_slice_ref(': ('src/core/lib/slice/slice.c'), + 'grpc_slice_unref(': ('src/core/lib/slice/slice.c'), +} + +errors = 0 +for root, dirs, files in os.walk('src/core'): + for filename in files: + path = os.path.join(root, filename) + if os.path.splitext(path)[1] != '.c': continue + with open(path) as f: + text = f.read() + for banned, exceptions in BANNED_EXCEPT.items(): + if path in exceptions: continue + if banned in text: + print 'Illegal use of "%s" in %s' % (banned, path) + errors += 1 + +assert errors == 0 + |