diff options
author | Craig Tiller <ctiller@google.com> | 2017-09-11 12:11:40 -0700 |
---|---|---|
committer | Craig Tiller <ctiller@google.com> | 2017-09-11 12:11:40 -0700 |
commit | bbe282f600ae030659c97bc5fea0d8ff5107ac17 (patch) | |
tree | d9eca449f8ee5883b7a8f610362addd8d855f5c0 /src/core/ext | |
parent | b61a26cff93b0ef97c98dad0c338d539ff44e7e4 (diff) | |
parent | 55c4b31389d5557b88d39bde6d783d68aa747de7 (diff) |
Merge github.com:grpc/grpc into grpc_millis
Diffstat (limited to 'src/core/ext')
43 files changed, 530 insertions, 394 deletions
diff --git a/src/core/ext/filters/client_channel/channel_connectivity.c b/src/core/ext/filters/client_channel/channel_connectivity.c index 78d275e6b8..416bdefa08 100644 --- a/src/core/ext/filters/client_channel/channel_connectivity.c +++ b/src/core/ext/filters/client_channel/channel_connectivity.c @@ -87,7 +87,7 @@ static void delete_state_watcher(grpc_exec_ctx *exec_ctx, state_watcher *w) { static void finished_completion(grpc_exec_ctx *exec_ctx, void *pw, grpc_cq_completion *ignored) { int delete = 0; - state_watcher *w = pw; + state_watcher *w = (state_watcher *)pw; gpr_mu_lock(&w->mu); switch (w->phase) { case WAITING: @@ -203,7 +203,7 @@ void grpc_channel_watch_connectivity_state( grpc_channel_element *client_channel_elem = grpc_channel_stack_last_element(grpc_channel_get_channel_stack(channel)); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - state_watcher *w = gpr_malloc(sizeof(*w)); + state_watcher *w = (state_watcher *)gpr_malloc(sizeof(*w)); GRPC_API_TRACE( "grpc_channel_watch_connectivity_state(" @@ -228,7 +228,8 @@ void grpc_channel_watch_connectivity_state( w->channel = channel; w->error = NULL; - watcher_timer_init_arg *wa = gpr_malloc(sizeof(watcher_timer_init_arg)); + watcher_timer_init_arg *wa = + (watcher_timer_init_arg *)gpr_malloc(sizeof(watcher_timer_init_arg)); wa->w = w; wa->deadline = deadline; GRPC_CLOSURE_INIT(&w->watcher_timer_init, watcher_timer_init, wa, diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c index 36630680ec..7cdf7d27a4 100644 --- a/src/core/ext/filters/client_channel/client_channel.c +++ b/src/core/ext/filters/client_channel/client_channel.c @@ -150,7 +150,8 @@ static void *method_parameters_create_from_json(const grpc_json *json) { if (!parse_timeout(field, &timeout)) return NULL; } } - method_parameters *value = gpr_malloc(sizeof(method_parameters)); + method_parameters *value = + (method_parameters *)gpr_malloc(sizeof(method_parameters)); gpr_ref_init(&value->refs, 1); value->timeout = timeout; value->wait_for_ready = wait_for_ready; @@ -256,7 +257,7 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx, static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - lb_policy_connectivity_watcher *w = arg; + lb_policy_connectivity_watcher *w = (lb_policy_connectivity_watcher *)arg; grpc_connectivity_state publish_state = w->state; /* check if the notification is for the latest policy */ if (w->lb_policy == w->chand->lb_policy) { @@ -283,7 +284,8 @@ static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx, static void watch_lb_policy_locked(grpc_exec_ctx *exec_ctx, channel_data *chand, grpc_lb_policy *lb_policy, grpc_connectivity_state current_state) { - lb_policy_connectivity_watcher *w = gpr_malloc(sizeof(*w)); + lb_policy_connectivity_watcher *w = + (lb_policy_connectivity_watcher *)gpr_malloc(sizeof(*w)); GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy"); w->chand = chand; GRPC_CLOSURE_INIT(&w->on_changed, on_lb_policy_state_changed_locked, w, @@ -312,7 +314,8 @@ typedef struct { } service_config_parsing_state; static void parse_retry_throttle_params(const grpc_json *field, void *arg) { - service_config_parsing_state *parsing_state = arg; + service_config_parsing_state *parsing_state = + (service_config_parsing_state *)arg; if (strcmp(field->key, "retryThrottling") == 0) { if (parsing_state->retry_throttle_data != NULL) return; // Duplicate. if (field->type != GRPC_JSON_OBJECT) return; @@ -367,7 +370,7 @@ static void parse_retry_throttle_params(const grpc_json *field, void *arg) { static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - channel_data *chand = arg; + channel_data *chand = (channel_data *)arg; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p: got resolver result: error=%s", chand, grpc_error_string(error)); @@ -393,7 +396,8 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, channel_arg = grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_ADDRESSES); if (channel_arg != NULL && channel_arg->type == GRPC_ARG_POINTER) { - grpc_lb_addresses *addresses = channel_arg->value.pointer.p; + grpc_lb_addresses *addresses = + (grpc_lb_addresses *)channel_arg->value.pointer.p; bool found_balancer_address = false; for (size_t i = 0; i < addresses->num_addresses; ++i) { if (addresses->addresses[i].is_balancer) { @@ -588,9 +592,10 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, static void start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error_ignored) { - grpc_transport_op *op = arg; - grpc_channel_element *elem = op->handler_private.extra_arg; - channel_data *chand = elem->channel_data; + grpc_transport_op *op = (grpc_transport_op *)arg; + grpc_channel_element *elem = + (grpc_channel_element *)op->handler_private.extra_arg; + channel_data *chand = (channel_data *)elem->channel_data; if (op->on_connectivity_state_change != NULL) { grpc_connectivity_state_notify_on_state_change( @@ -644,7 +649,7 @@ static void start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg, static void cc_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_transport_op *op) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; GPR_ASSERT(op->set_accept_stream == false); if (op->bind_pollset != NULL) { @@ -664,7 +669,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx, static void cc_get_channel_info(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, const grpc_channel_info *info) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; gpr_mu_lock(&chand->info_mu); if (info->lb_policy_name != NULL) { *info->lb_policy_name = chand->info_lb_policy_name == NULL @@ -684,7 +689,7 @@ static void cc_get_channel_info(grpc_exec_ctx *exec_ctx, static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_channel_element_args *args) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; GPR_ASSERT(args->is_last); GPR_ASSERT(elem->filter == &grpc_client_channel_filter); // Initialize data members. @@ -715,7 +720,8 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx, "client channel factory arg must be a pointer"); } grpc_client_channel_factory_ref(arg->value.pointer.p); - chand->client_channel_factory = arg->value.pointer.p; + chand->client_channel_factory = + (grpc_client_channel_factory *)arg->value.pointer.p; // Get server name to resolve, using proxy mapper if needed. arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVER_URI); if (arg == NULL) { @@ -747,7 +753,7 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx, static void shutdown_resolver_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_resolver *resolver = arg; + grpc_resolver *resolver = (grpc_resolver *)arg; grpc_resolver_shutdown_locked(exec_ctx, resolver); GRPC_RESOLVER_UNREF(exec_ctx, resolver, "channel"); } @@ -755,7 +761,7 @@ static void shutdown_resolver_locked(grpc_exec_ctx *exec_ctx, void *arg, /* Destructor for channel_data */ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; if (chand->resolver != NULL) { GRPC_CLOSURE_SCHED( exec_ctx, GRPC_CLOSURE_CREATE(shutdown_resolver_locked, chand->resolver, @@ -850,7 +856,7 @@ typedef struct client_channel_call_data { grpc_subchannel_call *grpc_client_channel_get_subchannel_call( grpc_call_element *elem) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; return calld->subchannel_call; } @@ -870,7 +876,7 @@ static void waiting_for_pick_batches_add( // This is called via the call combiner, so access to calld is synchronized. static void fail_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - call_data *calld = arg; + call_data *calld = (call_data *)arg; if (calld->waiting_for_pick_batches_count > 0) { --calld->waiting_for_pick_batches_count; grpc_transport_stream_op_batch_finish_with_failure( @@ -884,7 +890,7 @@ static void fail_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, static void waiting_for_pick_batches_fail(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_error *error) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: failing %" PRIdPTR " pending batches: %s", @@ -914,7 +920,7 @@ static void waiting_for_pick_batches_fail(grpc_exec_ctx *exec_ctx, // This is called via the call combiner, so access to calld is synchronized. static void run_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *ignored) { - call_data *calld = arg; + call_data *calld = (call_data *)arg; if (calld->waiting_for_pick_batches_count > 0) { --calld->waiting_for_pick_batches_count; grpc_subchannel_call_process_op( @@ -926,8 +932,8 @@ static void run_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, // This is called via the call combiner, so access to calld is synchronized. static void waiting_for_pick_batches_resume(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: sending %" PRIdPTR " pending batches to subchannel_call=%p", @@ -952,8 +958,8 @@ static void waiting_for_pick_batches_resume(grpc_exec_ctx *exec_ctx, // that the resolver has returned results to the channel. static void apply_service_config_to_call_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: applying service config to call", chand, calld); @@ -963,7 +969,7 @@ static void apply_service_config_to_call_locked(grpc_exec_ctx *exec_ctx, grpc_server_retry_throttle_data_ref(chand->retry_throttle_data); } if (chand->method_params_table != NULL) { - calld->method_params = grpc_method_config_table_get( + calld->method_params = (method_parameters *)grpc_method_config_table_get( exec_ctx, chand->method_params_table, calld->path); if (calld->method_params != NULL) { method_parameters_ref(calld->method_params); @@ -986,8 +992,8 @@ static void apply_service_config_to_call_locked(grpc_exec_ctx *exec_ctx, static void create_subchannel_call_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_error *error) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; const grpc_connected_subchannel_call_args call_args = { .pollent = calld->pollent, .path = calld->path, @@ -1015,8 +1021,8 @@ static void create_subchannel_call_locked(grpc_exec_ctx *exec_ctx, static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_error *error) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent, chand->interested_parties); if (calld->connected_subchannel == NULL) { @@ -1059,15 +1065,16 @@ typedef struct { static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - pick_after_resolver_result_args *args = arg; + pick_after_resolver_result_args *args = + (pick_after_resolver_result_args *)arg; if (args->finished) { gpr_free(args); return; } args->finished = true; grpc_call_element *elem = args->elem; - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; // If we don't yet have a resolver result, then a closure for // pick_after_resolver_result_done_locked() will have been added to // chand->waiting_for_resolver_result_closures, and it may not be invoked @@ -1093,7 +1100,8 @@ static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - pick_after_resolver_result_args *args = arg; + pick_after_resolver_result_args *args = + (pick_after_resolver_result_args *)arg; if (args->finished) { /* cancelled, do nothing */ if (GRPC_TRACER_ON(grpc_client_channel_trace)) { @@ -1104,8 +1112,8 @@ static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, } args->finished = true; grpc_call_element *elem = args->elem; - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (error != GRPC_ERROR_NONE) { if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver failed to return data", @@ -1125,8 +1133,8 @@ static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, static void pick_after_resolver_result_start_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: deferring pick pending resolver result", chand, @@ -1150,9 +1158,9 @@ static void pick_after_resolver_result_start_locked(grpc_exec_ctx *exec_ctx, // holding the call combiner. static void pick_callback_cancel_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call_element *elem = arg; - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)arg; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (error != GRPC_ERROR_NONE && calld->lb_policy != NULL) { if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p", @@ -1169,9 +1177,9 @@ static void pick_callback_cancel_locked(grpc_exec_ctx *exec_ctx, void *arg, // Unrefs the LB policy and invokes subchannel_ready_locked(). static void pick_callback_done_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call_element *elem = arg; - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)arg; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", chand, calld); @@ -1188,8 +1196,8 @@ static void pick_callback_done_locked(grpc_exec_ctx *exec_ctx, void *arg, static bool pick_callback_start_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_lb_policy_pick_args *inputs) { - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: starting pick on lb_policy=%p", chand, calld, chand->lb_policy); @@ -1224,8 +1232,8 @@ static bool pick_callback_start_locked(grpc_exec_ctx *exec_ctx, static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { GPR_TIMER_BEGIN("pick_subchannel", 0); - channel_data *chand = elem->channel_data; - call_data *calld = elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; bool pick_done = false; if (chand->lb_policy != NULL) { apply_service_config_to_call_locked(exec_ctx, elem); @@ -1297,8 +1305,8 @@ static void start_pick_locked(grpc_exec_ctx *exec_ctx, void *arg, } static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call_element *elem = arg; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)arg; + call_data *calld = (call_data *)elem->call_data; if (calld->retry_throttle_data != NULL) { if (error == GRPC_ERROR_NONE) { grpc_server_retry_throttle_data_record_success( @@ -1319,8 +1327,8 @@ static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { static void cc_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *batch) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; if (chand->deadline_checking_enabled) { grpc_deadline_state_client_start_transport_stream_op_batch(exec_ctx, elem, batch); @@ -1413,8 +1421,8 @@ done: static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; // Initialize data members. calld->path = grpc_slice_ref_internal(args->path); calld->call_start_time = args->start_time; @@ -1434,8 +1442,8 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, grpc_closure *then_schedule_closure) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *chand = (channel_data *)elem->channel_data; if (chand->deadline_checking_enabled) { grpc_deadline_state_destroy(exec_ctx, elem); } @@ -1469,7 +1477,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, static void cc_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_polling_entity *pollent) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; calld->pollent = pollent; } @@ -1493,7 +1501,7 @@ const grpc_channel_filter grpc_client_channel_filter = { static void try_to_connect_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error_ignored) { - channel_data *chand = arg; + channel_data *chand = (channel_data *)arg; if (chand->lb_policy != NULL) { grpc_lb_policy_exit_idle_locked(exec_ctx, chand->lb_policy); } else { @@ -1507,7 +1515,7 @@ static void try_to_connect_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_connectivity_state grpc_client_channel_check_connectivity_state( grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; grpc_connectivity_state out = grpc_connectivity_state_check(&chand->state_tracker); if (out == GRPC_CHANNEL_IDLE && try_to_connect) { @@ -1578,7 +1586,7 @@ static void external_connectivity_watcher_list_remove( int grpc_client_channel_num_external_connectivity_watchers( grpc_channel_element *elem) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; int count = 0; gpr_mu_lock(&chand->external_connectivity_watcher_list_mu); @@ -1595,7 +1603,7 @@ int grpc_client_channel_num_external_connectivity_watchers( static void on_external_watch_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - external_connectivity_watcher *w = arg; + external_connectivity_watcher *w = (external_connectivity_watcher *)arg; grpc_closure *follow_up = w->on_complete; grpc_polling_entity_del_from_pollset_set(exec_ctx, &w->pollent, w->chand->interested_parties); @@ -1608,7 +1616,7 @@ static void on_external_watch_complete(grpc_exec_ctx *exec_ctx, void *arg, static void watch_connectivity_state_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error_ignored) { - external_connectivity_watcher *w = arg; + external_connectivity_watcher *w = (external_connectivity_watcher *)arg; external_connectivity_watcher *found = NULL; if (w->state != NULL) { external_connectivity_watcher_list_append(w->chand, w); @@ -1637,8 +1645,9 @@ void grpc_client_channel_watch_connectivity_state( grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_polling_entity pollent, grpc_connectivity_state *state, grpc_closure *closure, grpc_closure *watcher_timer_init) { - channel_data *chand = elem->channel_data; - external_connectivity_watcher *w = gpr_zalloc(sizeof(*w)); + channel_data *chand = (channel_data *)elem->channel_data; + external_connectivity_watcher *w = + (external_connectivity_watcher *)gpr_zalloc(sizeof(*w)); w->chand = chand; w->pollent = pollent; w->on_complete = closure; diff --git a/src/core/ext/filters/client_channel/http_connect_handshaker.c b/src/core/ext/filters/client_channel/http_connect_handshaker.c index 0952dc6d4e..418bb41ef6 100644 --- a/src/core/ext/filters/client_channel/http_connect_handshaker.c +++ b/src/core/ext/filters/client_channel/http_connect_handshaker.c @@ -124,7 +124,7 @@ static void handshake_failed_locked(grpc_exec_ctx* exec_ctx, // Callback invoked when finished writing HTTP CONNECT request. static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - http_connect_handshaker* handshaker = arg; + http_connect_handshaker* handshaker = (http_connect_handshaker*)arg; gpr_mu_lock(&handshaker->mu); if (error != GRPC_ERROR_NONE || handshaker->shutdown) { // If the write failed or we're shutting down, clean up and invoke the @@ -145,7 +145,7 @@ static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg, // Callback invoked for reading HTTP CONNECT response. static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - http_connect_handshaker* handshaker = arg; + http_connect_handshaker* handshaker = (http_connect_handshaker*)arg; gpr_mu_lock(&handshaker->mu); if (error != GRPC_ERROR_NONE || handshaker->shutdown) { // If the read failed or we're shutting down, clean up and invoke the @@ -281,7 +281,8 @@ static void http_connect_handshaker_do_handshake( GPR_ASSERT(arg->type == GRPC_ARG_STRING); gpr_string_split(arg->value.string, "\n", &header_strings, &num_header_strings); - headers = gpr_malloc(sizeof(grpc_http_header) * num_header_strings); + headers = (grpc_http_header*)gpr_malloc(sizeof(grpc_http_header) * + num_header_strings); for (size_t i = 0; i < num_header_strings; ++i) { char* sep = strchr(header_strings[i], ':'); if (sep == NULL) { @@ -308,7 +309,7 @@ static void http_connect_handshaker_do_handshake( grpc_httpcli_request request; memset(&request, 0, sizeof(request)); request.host = server_name; - request.http.method = "CONNECT"; + request.http.method = (char*)"CONNECT"; request.http.path = server_name; request.http.hdrs = headers; request.http.hdr_count = num_headers; @@ -333,7 +334,8 @@ static const grpc_handshaker_vtable http_connect_handshaker_vtable = { http_connect_handshaker_do_handshake}; static grpc_handshaker* grpc_http_connect_handshaker_create() { - http_connect_handshaker* handshaker = gpr_malloc(sizeof(*handshaker)); + http_connect_handshaker* handshaker = + (http_connect_handshaker*)gpr_malloc(sizeof(*handshaker)); memset(handshaker, 0, sizeof(*handshaker)); grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base); gpr_mu_init(&handshaker->mu); diff --git a/src/core/ext/filters/client_channel/lb_policy.c b/src/core/ext/filters/client_channel/lb_policy.c index dd95a135cf..8e6673d737 100644 --- a/src/core/ext/filters/client_channel/lb_policy.c +++ b/src/core/ext/filters/client_channel/lb_policy.c @@ -67,7 +67,7 @@ void grpc_lb_policy_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) { static void shutdown_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_lb_policy *policy = arg; + grpc_lb_policy *policy = (grpc_lb_policy *)arg; policy->vtable->shutdown_locked(exec_ctx, policy); GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, policy, "strong-unref"); } diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c index 299f26b4de..bd290464c8 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c @@ -49,7 +49,7 @@ typedef struct { static void on_complete_for_send(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - call_data *calld = arg; + call_data *calld = (call_data *)arg; if (error == GRPC_ERROR_NONE) { calld->send_initial_metadata_succeeded = true; } @@ -59,7 +59,7 @@ static void on_complete_for_send(grpc_exec_ctx *exec_ctx, void *arg, static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - call_data *calld = arg; + call_data *calld = (call_data *)arg; if (error == GRPC_ERROR_NONE) { calld->recv_initial_metadata_succeeded = true; } @@ -70,7 +70,7 @@ static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, void *arg, static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; // Get stats object from context and take a ref. GPR_ASSERT(args->context != NULL); GPR_ASSERT(args->context[GRPC_GRPCLB_CLIENT_STATS].value != NULL); @@ -84,7 +84,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, grpc_closure *ignored) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; // Record call finished, optionally setting client_failed_to_send and // received. grpc_grpclb_client_stats_add_call_finished( @@ -98,7 +98,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, static void start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *batch) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; GPR_TIMER_BEGIN("clr_start_transport_stream_op_batch", 0); // Intercept send_initial_metadata. if (batch->send_initial_metadata) { diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c index 3c7f0e72a9..137642613c 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c @@ -181,7 +181,7 @@ typedef struct wrapped_rr_closure_arg { * order to unref the round robin instance upon its invocation */ static void wrapped_rr_closure(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - wrapped_rr_closure_arg *wc_arg = arg; + wrapped_rr_closure_arg *wc_arg = (wrapped_rr_closure_arg *)arg; GPR_ASSERT(wc_arg->wrapped_closure != NULL); GRPC_CLOSURE_SCHED(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_REF(error)); @@ -245,7 +245,7 @@ static void add_pending_pick(pending_pick **root, grpc_connected_subchannel **target, grpc_call_context_element *context, grpc_closure *on_complete) { - pending_pick *pp = gpr_zalloc(sizeof(*pp)); + pending_pick *pp = (pending_pick *)gpr_zalloc(sizeof(*pp)); pp->next = *root; pp->pick_args = *pick_args; pp->target = target; @@ -271,7 +271,7 @@ typedef struct pending_ping { } pending_ping; static void add_pending_ping(pending_ping **root, grpc_closure *notify) { - pending_ping *pping = gpr_zalloc(sizeof(*pping)); + pending_ping *pping = (pending_ping *)gpr_zalloc(sizeof(*pping)); pping->wrapped_notify_arg.wrapped_closure = notify; pping->wrapped_notify_arg.free_when_done = pping; pping->next = *root; @@ -671,7 +671,7 @@ static grpc_lb_policy_args *lb_policy_args_create(grpc_exec_ctx *exec_ctx, grpc_lb_addresses *addresses = process_serverlist_locked(exec_ctx, glb_policy->serverlist); GPR_ASSERT(addresses != NULL); - grpc_lb_policy_args *args = gpr_zalloc(sizeof(*args)); + grpc_lb_policy_args *args = (grpc_lb_policy_args *)gpr_zalloc(sizeof(*args)); args->client_channel_factory = glb_policy->cc_factory; args->combiner = glb_policy->base.combiner; // Replace the LB addresses in the channel args that we pass down to @@ -798,7 +798,7 @@ static void rr_handover_locked(grpc_exec_ctx *exec_ctx, static void glb_rr_connectivity_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - rr_connectivity_data *rr_connectivity = arg; + rr_connectivity_data *rr_connectivity = (rr_connectivity_data *)arg; glb_lb_policy *glb_policy = rr_connectivity->glb_policy; if (glb_policy->shutting_down) { GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base, @@ -841,8 +841,8 @@ static grpc_slice_hash_table_entry targets_info_entry_create( } static int balancer_name_cmp_fn(void *a, void *b) { - const char *a_str = a; - const char *b_str = b; + const char *a_str = (const char *)a; + const char *b_str = (const char *)b; return strcmp(a_str, b_str); } @@ -929,14 +929,14 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, if (arg == NULL || arg->type != GRPC_ARG_POINTER) { return NULL; } - grpc_lb_addresses *addresses = arg->value.pointer.p; + grpc_lb_addresses *addresses = (grpc_lb_addresses *)arg->value.pointer.p; size_t num_grpclb_addrs = 0; for (size_t i = 0; i < addresses->num_addresses; ++i) { if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; } if (num_grpclb_addrs == 0) return NULL; - glb_lb_policy *glb_policy = gpr_zalloc(sizeof(*glb_policy)); + glb_lb_policy *glb_policy = (glb_lb_policy *)gpr_zalloc(sizeof(*glb_policy)); /* Get server name. */ arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); @@ -1190,7 +1190,8 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, } GRPC_LB_POLICY_REF(glb_policy->rr_policy, "glb_pick"); - wrapped_rr_closure_arg *wc_arg = gpr_zalloc(sizeof(wrapped_rr_closure_arg)); + wrapped_rr_closure_arg *wc_arg = + (wrapped_rr_closure_arg *)gpr_zalloc(sizeof(wrapped_rr_closure_arg)); GRPC_CLOSURE_INIT(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg, grpc_schedule_on_exec_ctx); @@ -1272,7 +1273,7 @@ static void schedule_next_client_load_report(grpc_exec_ctx *exec_ctx, static void client_load_report_done_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; grpc_byte_buffer_destroy(glb_policy->client_load_report_payload); glb_policy->client_load_report_payload = NULL; if (error != GRPC_ERROR_NONE || glb_policy->lb_call == NULL) { @@ -1312,7 +1313,7 @@ static bool load_report_counters_are_zero(grpc_grpclb_request *request) { static void send_client_load_report_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; if (error == GRPC_ERROR_CANCELLED || glb_policy->lb_call == NULL) { glb_policy->client_load_report_timer_pending = false; GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base, @@ -1517,7 +1518,7 @@ static void query_for_backends_locked(grpc_exec_ctx *exec_ctx, static void lb_on_sent_initial_request_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; glb_policy->initial_request_sent = true; // If we attempted to send a client load report before the initial // request was sent, send the load report now. @@ -1530,7 +1531,7 @@ static void lb_on_sent_initial_request_locked(grpc_exec_ctx *exec_ctx, static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; grpc_op ops[2]; memset(ops, 0, sizeof(ops)); grpc_op *op = ops; @@ -1647,7 +1648,7 @@ static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg, static void lb_call_on_retry_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; glb_policy->retry_timer_active = false; if (!glb_policy->shutting_down && error == GRPC_ERROR_NONE) { if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { @@ -1662,7 +1663,7 @@ static void lb_call_on_retry_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, static void lb_on_server_status_received_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; GPR_ASSERT(glb_policy->lb_call != NULL); if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { char *status_details = @@ -1724,8 +1725,8 @@ static void glb_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, glb_policy->pending_update_args->args); gpr_free(glb_policy->pending_update_args); } - glb_policy->pending_update_args = - gpr_zalloc(sizeof(*glb_policy->pending_update_args)); + glb_policy->pending_update_args = (grpc_lb_policy_args *)gpr_zalloc( + sizeof(*glb_policy->pending_update_args)); glb_policy->pending_update_args->client_channel_factory = args->client_channel_factory; glb_policy->pending_update_args->args = grpc_channel_args_copy(args->args); @@ -1753,7 +1754,8 @@ static void glb_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, (void *)glb_policy); } } - const grpc_lb_addresses *addresses = arg->value.pointer.p; + const grpc_lb_addresses *addresses = + (const grpc_lb_addresses *)arg->value.pointer.p; GPR_ASSERT(glb_policy->lb_channel != NULL); grpc_channel_args *lb_channel_args = build_lb_channel_args( exec_ctx, addresses, glb_policy->response_generator, args->args); @@ -1786,7 +1788,7 @@ static void glb_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, static void glb_lb_channel_on_connectivity_changed_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - glb_lb_policy *glb_policy = arg; + glb_lb_policy *glb_policy = (glb_lb_policy *)arg; if (glb_policy->shutting_down) goto done; // Re-initialize the lb_call. This should also take care of updating the // embedded RR policy. Note that the current RR policy, if any, will stay in diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.c index 5b62623145..903120ca7d 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.c @@ -42,7 +42,8 @@ struct grpc_grpclb_client_stats { }; grpc_grpclb_client_stats* grpc_grpclb_client_stats_create() { - grpc_grpclb_client_stats* client_stats = gpr_zalloc(sizeof(*client_stats)); + grpc_grpclb_client_stats* client_stats = + (grpc_grpclb_client_stats*)gpr_zalloc(sizeof(*client_stats)); gpr_ref_init(&client_stats->refs, 1); return client_stats; } @@ -88,7 +89,8 @@ void grpc_grpclb_client_stats_add_call_dropped_locked( // Record the drop. if (client_stats->drop_token_counts == NULL) { client_stats->drop_token_counts = - gpr_zalloc(sizeof(grpc_grpclb_dropped_call_counts)); + (grpc_grpclb_dropped_call_counts*)gpr_zalloc( + sizeof(grpc_grpclb_dropped_call_counts)); } grpc_grpclb_dropped_call_counts* drop_token_counts = client_stats->drop_token_counts; @@ -103,9 +105,9 @@ void grpc_grpclb_client_stats_add_call_dropped_locked( while (new_num_entries < drop_token_counts->num_entries + 1) { new_num_entries *= 2; } - drop_token_counts->token_counts = - gpr_realloc(drop_token_counts->token_counts, - new_num_entries * sizeof(grpc_grpclb_drop_token_count)); + drop_token_counts->token_counts = (grpc_grpclb_drop_token_count*)gpr_realloc( + drop_token_counts->token_counts, + new_num_entries * sizeof(grpc_grpclb_drop_token_count)); grpc_grpclb_drop_token_count* new_entry = &drop_token_counts->token_counts[drop_token_counts->num_entries++]; new_entry->token = gpr_strdup(token); diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c index 050ebc12bf..393f6256bf 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c @@ -25,7 +25,7 @@ /* invoked once for every Server in ServerList */ static bool count_serverlist(pb_istream_t *stream, const pb_field_t *field, void **arg) { - grpc_grpclb_serverlist *sl = *arg; + grpc_grpclb_serverlist *sl = (grpc_grpclb_serverlist *)*arg; grpc_grpclb_server server; if (!pb_decode(stream, grpc_lb_v1_Server_fields, &server)) { gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream)); @@ -46,9 +46,10 @@ typedef struct decode_serverlist_arg { /* invoked once for every Server in ServerList */ static bool decode_serverlist(pb_istream_t *stream, const pb_field_t *field, void **arg) { - decode_serverlist_arg *dec_arg = *arg; + decode_serverlist_arg *dec_arg = (decode_serverlist_arg *)*arg; GPR_ASSERT(dec_arg->serverlist->num_servers >= dec_arg->decoding_idx); - grpc_grpclb_server *server = gpr_zalloc(sizeof(grpc_grpclb_server)); + grpc_grpclb_server *server = + (grpc_grpclb_server *)gpr_zalloc(sizeof(grpc_grpclb_server)); if (!pb_decode(stream, grpc_lb_v1_Server_fields, server)) { gpr_free(server); gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream)); @@ -59,7 +60,8 @@ static bool decode_serverlist(pb_istream_t *stream, const pb_field_t *field, } grpc_grpclb_request *grpc_grpclb_request_create(const char *lb_service_name) { - grpc_grpclb_request *req = gpr_malloc(sizeof(grpc_grpclb_request)); + grpc_grpclb_request *req = + (grpc_grpclb_request *)gpr_malloc(sizeof(grpc_grpclb_request)); req->has_client_stats = false; req->has_initial_request = true; req->initial_request.has_name = true; @@ -78,14 +80,15 @@ static void populate_timestamp(gpr_timespec timestamp, static bool encode_string(pb_ostream_t *stream, const pb_field_t *field, void *const *arg) { - char *str = *arg; + char *str = (char *)*arg; if (!pb_encode_tag_for_field(stream, field)) return false; return pb_encode_string(stream, (uint8_t *)str, strlen(str)); } static bool encode_drops(pb_ostream_t *stream, const pb_field_t *field, void *const *arg) { - grpc_grpclb_dropped_call_counts *drop_entries = *arg; + grpc_grpclb_dropped_call_counts *drop_entries = + (grpc_grpclb_dropped_call_counts *)*arg; if (drop_entries == NULL) return true; for (size_t i = 0; i < drop_entries->num_entries; ++i) { if (!pb_encode_tag_for_field(stream, field)) return false; @@ -104,7 +107,8 @@ static bool encode_drops(pb_ostream_t *stream, const pb_field_t *field, grpc_grpclb_request *grpc_grpclb_load_report_request_create_locked( grpc_grpclb_client_stats *client_stats) { - grpc_grpclb_request *req = gpr_zalloc(sizeof(grpc_grpclb_request)); + grpc_grpclb_request *req = + (grpc_grpclb_request *)gpr_zalloc(sizeof(grpc_grpclb_request)); req->has_client_stats = true; req->client_stats.has_timestamp = true; populate_timestamp(gpr_now(GPR_CLOCK_REALTIME), &req->client_stats.timestamp); @@ -179,7 +183,8 @@ grpc_grpclb_serverlist *grpc_grpclb_response_parse_serverlist( pb_istream_from_buffer(GRPC_SLICE_START_PTR(encoded_grpc_grpclb_response), GRPC_SLICE_LENGTH(encoded_grpc_grpclb_response)); pb_istream_t stream_at_start = stream; - grpc_grpclb_serverlist *sl = gpr_zalloc(sizeof(grpc_grpclb_serverlist)); + grpc_grpclb_serverlist *sl = + (grpc_grpclb_serverlist *)gpr_zalloc(sizeof(grpc_grpclb_serverlist)); grpc_grpclb_response res; memset(&res, 0, sizeof(grpc_grpclb_response)); // First pass: count number of servers. @@ -193,7 +198,8 @@ grpc_grpclb_serverlist *grpc_grpclb_response_parse_serverlist( } // Second pass: populate servers. if (sl->num_servers > 0) { - sl->servers = gpr_zalloc(sizeof(grpc_grpclb_server *) * sl->num_servers); + sl->servers = (grpc_grpclb_server **)gpr_zalloc( + sizeof(grpc_grpclb_server *) * sl->num_servers); decode_serverlist_arg decode_arg; memset(&decode_arg, 0, sizeof(decode_arg)); decode_arg.serverlist = sl; @@ -226,13 +232,16 @@ void grpc_grpclb_destroy_serverlist(grpc_grpclb_serverlist *serverlist) { grpc_grpclb_serverlist *grpc_grpclb_serverlist_copy( const grpc_grpclb_serverlist *sl) { - grpc_grpclb_serverlist *copy = gpr_zalloc(sizeof(grpc_grpclb_serverlist)); + grpc_grpclb_serverlist *copy = + (grpc_grpclb_serverlist *)gpr_zalloc(sizeof(grpc_grpclb_serverlist)); copy->num_servers = sl->num_servers; memcpy(©->expiration_interval, &sl->expiration_interval, sizeof(grpc_grpclb_duration)); - copy->servers = gpr_malloc(sizeof(grpc_grpclb_server *) * sl->num_servers); + copy->servers = (grpc_grpclb_server **)gpr_malloc( + sizeof(grpc_grpclb_server *) * sl->num_servers); for (size_t i = 0; i < sl->num_servers; i++) { - copy->servers[i] = gpr_malloc(sizeof(grpc_grpclb_server)); + copy->servers[i] = + (grpc_grpclb_server *)gpr_malloc(sizeof(grpc_grpclb_server)); memcpy(copy->servers[i], sl->servers[i], sizeof(grpc_grpclb_server)); } return copy; diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c index a50ba09bf5..fab3073eb9 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c @@ -217,7 +217,7 @@ static int pf_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, if (!p->started_picking) { start_picking_locked(exec_ctx, p); } - pp = gpr_malloc(sizeof(*pp)); + pp = (pending_pick *)gpr_malloc(sizeof(*pp)); pp->next = p->pending_picks; pp->target = target; pp->initial_metadata_flags = pick_args->initial_metadata_flags; @@ -314,7 +314,8 @@ static void pf_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, } return; } - const grpc_lb_addresses *addresses = arg->value.pointer.p; + const grpc_lb_addresses *addresses = + (const grpc_lb_addresses *)arg->value.pointer.p; if (addresses->num_addresses == 0) { // Empty update. Unsubscribe from all current subchannels and put the // channel in TRANSIENT_FAILURE. @@ -392,7 +393,8 @@ static void pf_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_channel_args_destroy(exec_ctx, p->pending_update_args->args); gpr_free(p->pending_update_args); } - p->pending_update_args = gpr_zalloc(sizeof(*p->pending_update_args)); + p->pending_update_args = + (grpc_lb_policy_args *)gpr_zalloc(sizeof(*p->pending_update_args)); p->pending_update_args->client_channel_factory = args->client_channel_factory; p->pending_update_args->args = grpc_channel_args_copy(args->args); @@ -456,7 +458,7 @@ static void pf_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, static void pf_connectivity_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - pick_first_lb_policy *p = arg; + pick_first_lb_policy *p = (pick_first_lb_policy *)arg; grpc_subchannel *selected_subchannel; pending_pick *pp; @@ -678,7 +680,7 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx, grpc_lb_policy_factory *factory, grpc_lb_policy_args *args) { GPR_ASSERT(args->client_channel_factory != NULL); - pick_first_lb_policy *p = gpr_zalloc(sizeof(*p)); + pick_first_lb_policy *p = (pick_first_lb_policy *)gpr_zalloc(sizeof(*p)); if (GRPC_TRACER_ON(grpc_lb_pick_first_trace)) { gpr_log(GPR_DEBUG, "Pick First %p created.", (void *)p); } diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c index 866fb9a1eb..be91d3d651 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c @@ -144,10 +144,11 @@ struct rr_subchannel_list { static rr_subchannel_list *rr_subchannel_list_create(round_robin_lb_policy *p, size_t num_subchannels) { - rr_subchannel_list *subchannel_list = gpr_zalloc(sizeof(*subchannel_list)); + rr_subchannel_list *subchannel_list = + (rr_subchannel_list *)gpr_zalloc(sizeof(*subchannel_list)); subchannel_list->policy = p; subchannel_list->subchannels = - gpr_zalloc(sizeof(subchannel_data) * num_subchannels); + (subchannel_data *)gpr_zalloc(sizeof(subchannel_data) * num_subchannels); subchannel_list->num_subchannels = num_subchannels; gpr_ref_init(&subchannel_list->refcount, 1); if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) { @@ -452,7 +453,7 @@ static int rr_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, if (!p->started_picking) { start_picking_locked(exec_ctx, p); } - pending_pick *pp = gpr_malloc(sizeof(*pp)); + pending_pick *pp = (pending_pick *)gpr_malloc(sizeof(*pp)); pp->next = p->pending_picks; pp->target = target; pp->on_complete = on_complete; @@ -553,7 +554,7 @@ static grpc_connectivity_state update_lb_connectivity_status_locked( static void rr_connectivity_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - subchannel_data *sd = arg; + subchannel_data *sd = (subchannel_data *)arg; round_robin_lb_policy *p = sd->subchannel_list->policy; if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) { gpr_log( @@ -754,7 +755,7 @@ static void rr_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, } return; } - grpc_lb_addresses *addresses = arg->value.pointer.p; + grpc_lb_addresses *addresses = (grpc_lb_addresses *)arg->value.pointer.p; rr_subchannel_list *subchannel_list = rr_subchannel_list_create(p, addresses->num_addresses); if (addresses->num_addresses == 0) { @@ -887,7 +888,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx, grpc_lb_policy_factory *factory, grpc_lb_policy_args *args) { GPR_ASSERT(args->client_channel_factory != NULL); - round_robin_lb_policy *p = gpr_zalloc(sizeof(*p)); + round_robin_lb_policy *p = (round_robin_lb_policy *)gpr_zalloc(sizeof(*p)); grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable, args->combiner); grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, "round_robin"); diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.c b/src/core/ext/filters/client_channel/lb_policy_factory.c index 538d8d65ed..cdcaf17544 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.c +++ b/src/core/ext/filters/client_channel/lb_policy_factory.c @@ -28,11 +28,12 @@ grpc_lb_addresses* grpc_lb_addresses_create( size_t num_addresses, const grpc_lb_user_data_vtable* user_data_vtable) { - grpc_lb_addresses* addresses = gpr_zalloc(sizeof(grpc_lb_addresses)); + grpc_lb_addresses* addresses = + (grpc_lb_addresses*)gpr_zalloc(sizeof(grpc_lb_addresses)); addresses->num_addresses = num_addresses; addresses->user_data_vtable = user_data_vtable; const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses; - addresses->addresses = gpr_zalloc(addresses_size); + addresses->addresses = (grpc_lb_address*)gpr_zalloc(addresses_size); return addresses; } diff --git a/src/core/ext/filters/client_channel/proxy_mapper_registry.c b/src/core/ext/filters/client_channel/proxy_mapper_registry.c index 5f43a0596a..09967eea3c 100644 --- a/src/core/ext/filters/client_channel/proxy_mapper_registry.c +++ b/src/core/ext/filters/client_channel/proxy_mapper_registry.c @@ -34,7 +34,7 @@ typedef struct { static void grpc_proxy_mapper_list_register(grpc_proxy_mapper_list* list, bool at_start, grpc_proxy_mapper* mapper) { - list->list = gpr_realloc( + list->list = (grpc_proxy_mapper**)gpr_realloc( list->list, (list->num_mappers + 1) * sizeof(grpc_proxy_mapper*)); if (at_start) { memmove(list->list + 1, list->list, diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c index c75cba0613..c32b5c5a75 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c @@ -144,7 +144,7 @@ static void dns_ares_channel_saw_error_locked(grpc_exec_ctx *exec_ctx, static void dns_ares_on_retry_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - ares_dns_resolver *r = arg; + ares_dns_resolver *r = (ares_dns_resolver *)arg; r->have_retry_timer = false; if (error == GRPC_ERROR_NONE) { if (!r->resolving) { @@ -227,7 +227,7 @@ static char *choose_service_config(char *service_config_choice_json) { static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - ares_dns_resolver *r = arg; + ares_dns_resolver *r = (ares_dns_resolver *)arg; grpc_channel_args *result = NULL; GPR_ASSERT(r->resolving); r->resolving = false; @@ -361,7 +361,8 @@ static grpc_resolver *dns_ares_create(grpc_exec_ctx *exec_ctx, const char *path = args->uri->path; if (path[0] == '/') ++path; /* Create resolver. */ - ares_dns_resolver *r = gpr_zalloc(sizeof(ares_dns_resolver)); + ares_dns_resolver *r = + (ares_dns_resolver *)gpr_zalloc(sizeof(ares_dns_resolver)); grpc_resolver_init(&r->base, &dns_ares_resolver_vtable, args->combiner); if (0 != strcmp(args->uri->authority, "")) { r->dns_server = gpr_strdup(args->uri->authority); diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c index b696344eab..9747d39a16 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c @@ -111,7 +111,7 @@ static void fd_node_destroy(grpc_exec_ctx *exec_ctx, fd_node *fdn) { grpc_error *grpc_ares_ev_driver_create(grpc_ares_ev_driver **ev_driver, grpc_pollset_set *pollset_set) { - *ev_driver = gpr_malloc(sizeof(grpc_ares_ev_driver)); + *ev_driver = (grpc_ares_ev_driver *)gpr_malloc(sizeof(grpc_ares_ev_driver)); int status = ares_init(&(*ev_driver)->channel); gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create"); if (status != ARES_SUCCESS) { @@ -178,7 +178,7 @@ static fd_node *pop_fd_node(fd_node **head, int fd) { static void on_readable_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - fd_node *fdn = arg; + fd_node *fdn = (fd_node *)arg; grpc_ares_ev_driver *ev_driver = fdn->ev_driver; gpr_mu_lock(&fdn->mu); fdn->readable_registered = false; @@ -205,7 +205,7 @@ static void on_readable_cb(grpc_exec_ctx *exec_ctx, void *arg, static void on_writable_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - fd_node *fdn = arg; + fd_node *fdn = (fd_node *)arg; grpc_ares_ev_driver *ev_driver = fdn->ev_driver; gpr_mu_lock(&fdn->mu); fdn->writable_registered = false; @@ -251,7 +251,7 @@ static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, if (fdn == NULL) { char *fd_name; gpr_asprintf(&fd_name, "ares_ev_driver-%" PRIuPTR, i); - fdn = gpr_malloc(sizeof(fd_node)); + fdn = (fd_node *)gpr_malloc(sizeof(fd_node)); gpr_log(GPR_DEBUG, "new fd: %d", socks[i]); fdn->grpc_fd = grpc_fd_create(socks[i], fd_name); fdn->ev_driver = ev_driver; diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c index e65723a63b..0d71f3560e 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c @@ -158,9 +158,9 @@ static void on_hostbyname_done_cb(void *arg, int status, int timeouts, for (i = 0; hostent->h_addr_list[i] != NULL; i++) { } (*lb_addresses)->num_addresses += i; - (*lb_addresses)->addresses = - gpr_realloc((*lb_addresses)->addresses, - sizeof(grpc_lb_address) * (*lb_addresses)->num_addresses); + (*lb_addresses)->addresses = (grpc_lb_address *)gpr_realloc( + (*lb_addresses)->addresses, + sizeof(grpc_lb_address) * (*lb_addresses)->num_addresses); for (i = prev_naddr; i < (*lb_addresses)->num_addresses; i++) { switch (hostent->h_addrtype) { case AF_INET6: { @@ -293,12 +293,12 @@ static void on_txt_done_cb(void *arg, int status, int timeouts, // Found a service config record. if (result != NULL) { size_t service_config_len = result->length - prefix_len; - *r->service_config_json_out = gpr_malloc(service_config_len + 1); + *r->service_config_json_out = (char *)gpr_malloc(service_config_len + 1); memcpy(*r->service_config_json_out, result->txt + prefix_len, service_config_len); for (result = result->next; result != NULL && !result->record_start; result = result->next) { - *r->service_config_json_out = gpr_realloc( + *r->service_config_json_out = (char *)gpr_realloc( *r->service_config_json_out, service_config_len + result->length + 1); memcpy(*r->service_config_json_out + service_config_len, result->txt, result->length); @@ -360,7 +360,8 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( error = grpc_ares_ev_driver_create(&ev_driver, interested_parties); if (error != GRPC_ERROR_NONE) goto error_cleanup; - grpc_ares_request *r = gpr_zalloc(sizeof(grpc_ares_request)); + grpc_ares_request *r = + (grpc_ares_request *)gpr_zalloc(sizeof(grpc_ares_request)); gpr_mu_init(&r->mu); r->ev_driver = ev_driver; r->on_done = on_done; @@ -502,10 +503,11 @@ static void on_dns_lookup_done_cb(grpc_exec_ctx *exec_ctx, void *arg, if (r->lb_addrs == NULL || r->lb_addrs->num_addresses == 0) { *resolved_addresses = NULL; } else { - *resolved_addresses = gpr_zalloc(sizeof(grpc_resolved_addresses)); + *resolved_addresses = + (grpc_resolved_addresses *)gpr_zalloc(sizeof(grpc_resolved_addresses)); (*resolved_addresses)->naddrs = r->lb_addrs->num_addresses; - (*resolved_addresses)->addrs = gpr_zalloc(sizeof(grpc_resolved_address) * - (*resolved_addresses)->naddrs); + (*resolved_addresses)->addrs = (grpc_resolved_address *)gpr_zalloc( + sizeof(grpc_resolved_address) * (*resolved_addresses)->naddrs); for (size_t i = 0; i < (*resolved_addresses)->naddrs; i++) { GPR_ASSERT(!r->lb_addrs->addresses[i].is_balancer); memcpy(&(*resolved_addresses)->addrs[i], diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c index 3ff081a514..c4676e0299 100644 --- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c +++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c @@ -159,7 +159,7 @@ typedef struct set_response_closure_arg { static void set_response_closure_fn(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - set_response_closure_arg* closure_arg = arg; + set_response_closure_arg* closure_arg = (set_response_closure_arg*)arg; grpc_fake_resolver_response_generator* generator = closure_arg->generator; fake_resolver* r = generator->resolver; if (r->next_results != NULL) { @@ -178,7 +178,8 @@ void grpc_fake_resolver_response_generator_set_response( grpc_exec_ctx* exec_ctx, grpc_fake_resolver_response_generator* generator, grpc_channel_args* next_response) { GPR_ASSERT(generator->resolver != NULL); - set_response_closure_arg* closure_arg = gpr_zalloc(sizeof(*closure_arg)); + set_response_closure_arg* closure_arg = + (set_response_closure_arg*)gpr_zalloc(sizeof(*closure_arg)); closure_arg->generator = generator; closure_arg->next_response = grpc_channel_args_copy(next_response); GRPC_CLOSURE_SCHED(exec_ctx, diff --git a/src/core/ext/filters/client_channel/retry_throttle.c b/src/core/ext/filters/client_channel/retry_throttle.c index 0c7a3ae651..6cd6654b6f 100644 --- a/src/core/ext/filters/client_channel/retry_throttle.c +++ b/src/core/ext/filters/client_channel/retry_throttle.c @@ -139,12 +139,14 @@ static long compare_server_name(void* key1, void* key2, void* unused) { } static void destroy_server_retry_throttle_data(void* value, void* unused) { - grpc_server_retry_throttle_data* throttle_data = value; + grpc_server_retry_throttle_data* throttle_data = + (grpc_server_retry_throttle_data*)value; grpc_server_retry_throttle_data_unref(throttle_data); } static void* copy_server_retry_throttle_data(void* value, void* unused) { - grpc_server_retry_throttle_data* throttle_data = value; + grpc_server_retry_throttle_data* throttle_data = + (grpc_server_retry_throttle_data*)value; return grpc_server_retry_throttle_data_ref(throttle_data); } diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c index 61c059e4a3..2c466f595a 100644 --- a/src/core/ext/filters/client_channel/subchannel.c +++ b/src/core/ext/filters/client_channel/subchannel.c @@ -157,7 +157,7 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel, static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_connected_subchannel *c = arg; + grpc_connected_subchannel *c = (grpc_connected_subchannel *)arg; grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c)); gpr_free(c); } @@ -181,7 +181,7 @@ void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx, static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_subchannel *c = arg; + grpc_subchannel *c = (grpc_subchannel *)arg; gpr_free((void *)c->filters); grpc_channel_args_destroy(exec_ctx, c->args); grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker); @@ -290,21 +290,23 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, return c; } - c = gpr_zalloc(sizeof(*c)); + c = (grpc_subchannel *)gpr_zalloc(sizeof(*c)); c->key = key; gpr_atm_no_barrier_store(&c->ref_pair, 1 << INTERNAL_REF_BITS); c->connector = connector; grpc_connector_ref(c->connector); c->num_filters = args->filter_count; if (c->num_filters > 0) { - c->filters = gpr_malloc(sizeof(grpc_channel_filter *) * c->num_filters); + c->filters = (const grpc_channel_filter **)gpr_malloc( + sizeof(grpc_channel_filter *) * c->num_filters); memcpy((void *)c->filters, args->filters, sizeof(grpc_channel_filter *) * c->num_filters); } else { c->filters = NULL; } c->pollset_set = grpc_pollset_set_create(); - grpc_resolved_address *addr = gpr_malloc(sizeof(*addr)); + grpc_resolved_address *addr = + (grpc_resolved_address *)gpr_malloc(sizeof(*addr)); grpc_get_subchannel_address_arg(exec_ctx, args->args, addr); grpc_resolved_address *new_address = NULL; grpc_channel_args *new_args = NULL; @@ -400,7 +402,7 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c, static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - external_state_watcher *w = arg; + external_state_watcher *w = (external_state_watcher *)arg; grpc_closure *follow_up = w->notify; if (w->pollset_set != NULL) { grpc_pollset_set_del_pollset_set(exec_ctx, w->subchannel->pollset_set, @@ -416,7 +418,7 @@ static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg, } static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_subchannel *c = arg; + grpc_subchannel *c = (grpc_subchannel *)arg; gpr_mu_lock(&c->mu); c->have_alarm = false; if (c->disconnected) { @@ -498,7 +500,7 @@ void grpc_subchannel_notify_on_state_change( } gpr_mu_unlock(&c->mu); } else { - w = gpr_malloc(sizeof(*w)); + w = (external_state_watcher *)gpr_malloc(sizeof(*w)); w->subchannel = c; w->pollset_set = interested_parties; w->notify = notify; @@ -530,7 +532,7 @@ void grpc_connected_subchannel_process_transport_op( static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - state_watcher *sw = p; + state_watcher *sw = (state_watcher *)p; grpc_subchannel *c = sw->subchannel; gpr_mu *mu = &c->mu; @@ -620,7 +622,7 @@ static bool publish_transport_locked(grpc_exec_ctx *exec_ctx, memset(&c->connecting_result, 0, sizeof(c->connecting_result)); /* initialize state watcher */ - sw_subchannel = gpr_malloc(sizeof(*sw_subchannel)); + sw_subchannel = (state_watcher *)gpr_malloc(sizeof(*sw_subchannel)); sw_subchannel->subchannel = c; sw_subchannel->connectivity_state = GRPC_CHANNEL_READY; GRPC_CLOSURE_INIT(&sw_subchannel->closure, subchannel_on_child_state_changed, @@ -657,7 +659,7 @@ static bool publish_transport_locked(grpc_exec_ctx *exec_ctx, static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_subchannel *c = arg; + grpc_subchannel *c = (grpc_subchannel *)arg; grpc_channel_args *delete_channel_args = c->connecting_result.channel_args; GRPC_SUBCHANNEL_WEAK_REF(c, "connected"); @@ -693,7 +695,7 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg, static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call, grpc_error *error) { - grpc_subchannel_call *c = call; + grpc_subchannel_call *c = (grpc_subchannel_call *)call; GPR_ASSERT(c->schedule_closure_after_destroy != NULL); GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0); grpc_connected_subchannel *connection = c->connection; @@ -747,7 +749,7 @@ grpc_error *grpc_connected_subchannel_create_call( const grpc_connected_subchannel_call_args *args, grpc_subchannel_call **call) { grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con); - *call = gpr_arena_alloc( + *call = (grpc_subchannel_call *)gpr_arena_alloc( args->arena, sizeof(grpc_subchannel_call) + chanstk->call_stack_size); grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call); (*call)->connection = GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call"); diff --git a/src/core/ext/filters/client_channel/subchannel_index.c b/src/core/ext/filters/client_channel/subchannel_index.c index ababd05d84..f57b631c41 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.c +++ b/src/core/ext/filters/client_channel/subchannel_index.c @@ -43,11 +43,11 @@ static bool g_force_creation = false; static grpc_subchannel_key *create_key( const grpc_subchannel_args *args, grpc_channel_args *(*copy_channel_args)(const grpc_channel_args *args)) { - grpc_subchannel_key *k = gpr_malloc(sizeof(*k)); + grpc_subchannel_key *k = (grpc_subchannel_key *)gpr_malloc(sizeof(*k)); k->args.filter_count = args->filter_count; if (k->args.filter_count > 0) { - k->args.filters = - gpr_malloc(sizeof(*k->args.filters) * k->args.filter_count); + k->args.filters = (const grpc_channel_filter **)gpr_malloc( + sizeof(*k->args.filters) * k->args.filter_count); memcpy((grpc_channel_filter *)k->args.filters, args->filters, sizeof(*k->args.filters) * k->args.filter_count); } else { @@ -137,7 +137,7 @@ grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx, gpr_mu_unlock(&g_mu); grpc_subchannel *c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF( - gpr_avl_get(index, key, exec_ctx), "index_find"); + (grpc_subchannel *)gpr_avl_get(index, key, exec_ctx), "index_find"); gpr_avl_unref(index, exec_ctx); return c; @@ -159,7 +159,7 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx, gpr_mu_unlock(&g_mu); // - Check to see if a subchannel already exists - c = gpr_avl_get(index, key, exec_ctx); + c = (grpc_subchannel *)gpr_avl_get(index, key, exec_ctx); if (c != NULL) { c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "index_register"); } @@ -207,7 +207,7 @@ void grpc_subchannel_index_unregister(grpc_exec_ctx *exec_ctx, // Check to see if this key still refers to the previously // registered subchannel - grpc_subchannel *c = gpr_avl_get(index, key, exec_ctx); + grpc_subchannel *c = (grpc_subchannel *)gpr_avl_get(index, key, exec_ctx); if (c != constructed) { gpr_avl_unref(index, exec_ctx); break; diff --git a/src/core/ext/filters/client_channel/uri_parser.c b/src/core/ext/filters/client_channel/uri_parser.c index e841928760..fb4fb8e694 100644 --- a/src/core/ext/filters/client_channel/uri_parser.c +++ b/src/core/ext/filters/client_channel/uri_parser.c @@ -45,7 +45,7 @@ static grpc_uri *bad_uri(const char *uri_text, size_t pos, const char *section, gpr_log(GPR_ERROR, "%s%s'", line_prefix, uri_text); gpr_free(line_prefix); - line_prefix = gpr_malloc(pfx_len + 1); + line_prefix = (char *)gpr_malloc(pfx_len + 1); memset(line_prefix, ' ', pfx_len); line_prefix[pfx_len] = 0; gpr_log(GPR_ERROR, "%s^ here", line_prefix); @@ -156,7 +156,8 @@ static void parse_query_parts(grpc_uri *uri) { gpr_string_split(uri->query, QUERY_PARTS_SEPARATOR, &uri->query_parts, &uri->num_query_parts); - uri->query_parts_values = gpr_malloc(uri->num_query_parts * sizeof(char **)); + uri->query_parts_values = + (char **)gpr_malloc(uri->num_query_parts * sizeof(char **)); for (size_t i = 0; i < uri->num_query_parts; i++) { char **query_param_parts; size_t num_query_param_parts; @@ -269,7 +270,7 @@ grpc_uri *grpc_uri_parse(grpc_exec_ctx *exec_ctx, const char *uri_text, fragment_end = i; } - uri = gpr_zalloc(sizeof(*uri)); + uri = (grpc_uri *)gpr_zalloc(sizeof(*uri)); uri->scheme = decode_and_copy_component(exec_ctx, uri_text, scheme_begin, scheme_end); uri->authority = decode_and_copy_component(exec_ctx, uri_text, diff --git a/src/core/ext/filters/deadline/deadline_filter.c b/src/core/ext/filters/deadline/deadline_filter.c index 023378af94..2b2eac2ac5 100644 --- a/src/core/ext/filters/deadline/deadline_filter.c +++ b/src/core/ext/filters/deadline/deadline_filter.c @@ -38,7 +38,7 @@ // filter stack. Yields the call combiner when the batch returns. static void yield_call_combiner(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* ignored) { - grpc_deadline_state* deadline_state = arg; + grpc_deadline_state* deadline_state = (grpc_deadline_state*)arg; GRPC_CALL_COMBINER_STOP(exec_ctx, deadline_state->call_combiner, "got on_complete from cancel_stream batch"); GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer"); @@ -48,8 +48,8 @@ static void yield_call_combiner(grpc_exec_ctx* exec_ctx, void* arg, // synchronized. static void send_cancel_op_in_call_combiner(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - grpc_call_element* elem = arg; - grpc_deadline_state* deadline_state = elem->call_data; + grpc_call_element* elem = (grpc_call_element*)arg; + grpc_deadline_state* deadline_state = (grpc_deadline_state*)elem->call_data; grpc_transport_stream_op_batch* batch = grpc_make_transport_stream_op( GRPC_CLOSURE_INIT(&deadline_state->timer_callback, yield_call_combiner, deadline_state, grpc_schedule_on_exec_ctx)); @@ -158,8 +158,10 @@ struct start_timer_after_init_state { }; static void start_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - struct start_timer_after_init_state* state = arg; - grpc_deadline_state* deadline_state = state->elem->call_data; + struct start_timer_after_init_state* state = + (struct start_timer_after_init_state*)arg; + grpc_deadline_state* deadline_state = + (grpc_deadline_state*)state->elem->call_data; if (!state->in_call_combiner) { // We are initially called without holding the call combiner, so we // need to bounce ourselves into it. @@ -192,7 +194,8 @@ void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, // call stack initialization is finished. To avoid that problem, we // create a closure to start the timer, and we schedule that closure // to be run after call stack initialization is done. - struct start_timer_after_init_state* state = gpr_zalloc(sizeof(*state)); + struct start_timer_after_init_state* state = + (struct start_timer_after_init_state*)gpr_zalloc(sizeof(*state)); state->elem = elem; state->deadline = deadline; GRPC_CLOSURE_INIT(&state->closure, start_timer_after_init, state, diff --git a/src/core/ext/filters/http/client/http_client_filter.c b/src/core/ext/filters/http/client/http_client_filter.c index 2d7429c41e..6208089f2e 100644 --- a/src/core/ext/filters/http/client/http_client_filter.c +++ b/src/core/ext/filters/http/client/http_client_filter.c @@ -139,8 +139,8 @@ static grpc_error *client_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *error) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (error == GRPC_ERROR_NONE) { error = client_filter_incoming_metadata(exec_ctx, elem, calld->recv_initial_metadata); @@ -154,8 +154,8 @@ static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, static void recv_trailing_metadata_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *error) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (error == GRPC_ERROR_NONE) { error = client_filter_incoming_metadata(exec_ctx, elem, calld->recv_trailing_metadata); @@ -234,7 +234,7 @@ static void on_send_message_next_done(grpc_exec_ctx *exec_ctx, void *arg, } static char *slice_buffer_to_string(grpc_slice_buffer *slice_buffer) { - char *payload_bytes = gpr_malloc(slice_buffer->length + 1); + char *payload_bytes = (char *)gpr_malloc(slice_buffer->length + 1); size_t offset = 0; for (size_t i = 0; i < slice_buffer->count; ++i) { memcpy(payload_bytes + offset, @@ -300,8 +300,8 @@ static void remove_if_present(grpc_exec_ctx *exec_ctx, static void hc_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *batch) { - call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *channeld = (channel_data *)elem->channel_data; GPR_TIMER_BEGIN("hc_start_transport_stream_op_batch", 0); if (batch->recv_initial_metadata) { @@ -536,7 +536,7 @@ static grpc_slice user_agent_from_args(const grpc_channel_args *args, static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_channel_element_args *args) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; GPR_ASSERT(!args->is_last); GPR_ASSERT(args->optional_transport != NULL); chand->static_scheme = scheme_from_args(args->channel_args); @@ -552,7 +552,7 @@ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, /* Destructor for channel data */ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem) { - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; GRPC_MDELEM_UNREF(exec_ctx, chand->user_agent); } diff --git a/src/core/ext/filters/http/http_filters_plugin.c b/src/core/ext/filters/http/http_filters_plugin.c index a5c1b92054..88bd2250f9 100644 --- a/src/core/ext/filters/http/http_filters_plugin.c +++ b/src/core/ext/filters/http/http_filters_plugin.c @@ -44,7 +44,7 @@ static bool maybe_add_optional_filter(grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, void *arg) { if (!is_building_http_like_transport(builder)) return true; - optional_filter *filtarg = arg; + optional_filter *filtarg = (optional_filter *)arg; const grpc_channel_args *channel_args = grpc_channel_stack_builder_get_channel_arguments(builder); bool enable = grpc_channel_arg_get_bool( diff --git a/src/core/ext/filters/http/message_compress/message_compress_filter.c b/src/core/ext/filters/http/message_compress/message_compress_filter.c index 98a503cafc..0145dffab0 100644 --- a/src/core/ext/filters/http/message_compress/message_compress_filter.c +++ b/src/core/ext/filters/http/message_compress/message_compress_filter.c @@ -82,8 +82,8 @@ typedef struct channel_data { static bool skip_compression(grpc_call_element *elem, uint32_t flags, bool has_compression_algorithm) { - call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *channeld = (channel_data *)elem->channel_data; if (flags & (GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_INTERNAL_COMPRESS)) { return true; @@ -106,8 +106,8 @@ static grpc_error *process_send_initial_metadata( static grpc_error *process_send_initial_metadata( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_metadata_batch *initial_metadata, bool *has_compression_algorithm) { - call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + channel_data *channeld = (channel_data *)elem->channel_data; *has_compression_algorithm = false; grpc_stream_compression_algorithm stream_compression_algorithm = GRPC_STREAM_COMPRESS_NONE; @@ -285,7 +285,7 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx, static void fail_send_message_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - call_data *calld = arg; + call_data *calld = (call_data *)arg; if (calld->send_message_batch != NULL) { grpc_transport_stream_op_batch_finish_with_failure( exec_ctx, calld->send_message_batch, GRPC_ERROR_REF(error), @@ -374,7 +374,7 @@ static void start_send_message_batch(grpc_exec_ctx *exec_ctx, void *arg, static void compress_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *batch) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; GPR_TIMER_BEGIN("compress_start_transport_stream_op_batch", 0); // Handle cancel_stream. if (batch->cancel_stream) { @@ -473,7 +473,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, grpc_closure *ignored) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices); GRPC_ERROR_UNREF(calld->cancel_error); } @@ -482,7 +482,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_channel_element_args *args) { - channel_data *channeld = elem->channel_data; + channel_data *channeld = (channel_data *)elem->channel_data; /* Configuration for message compression */ channeld->enabled_algorithms_bitset = diff --git a/src/core/ext/filters/http/server/http_server_filter.c b/src/core/ext/filters/http/server/http_server_filter.c index a10e69ba59..554a7f530d 100644 --- a/src/core/ext/filters/http/server/http_server_filter.c +++ b/src/core/ext/filters/http/server/http_server_filter.c @@ -94,7 +94,7 @@ static void add_error(const char *error_name, grpc_error **cumulative, static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_metadata_batch *b) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; grpc_error *error = GRPC_ERROR_NONE; static const char *error_name = "Failed processing incoming headers"; @@ -263,8 +263,8 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx, static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (err == GRPC_ERROR_NONE) { err = server_filter_incoming_metadata(exec_ctx, elem, calld->recv_initial_metadata); @@ -276,8 +276,8 @@ static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, static void hs_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; /* Call recv_message_ready if we got the payload via the path field */ if (calld->seen_path_with_query && calld->recv_message_ready != NULL) { *calld->pp_recv_message = calld->payload_bin_delivered @@ -296,8 +296,8 @@ static void hs_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, static void hs_recv_message_ready(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (calld->seen_path_with_query) { // Do nothing. This is probably a GET request, and payload will be // returned in hs_on_complete callback. @@ -314,7 +314,7 @@ static grpc_error *hs_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *op) { /* grab pointers to our data from the call element */ - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; if (op->send_initial_metadata) { grpc_error *error = GRPC_ERROR_NONE; @@ -376,7 +376,7 @@ static grpc_error *hs_mutate_op(grpc_exec_ctx *exec_ctx, static void hs_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *op) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; GPR_TIMER_BEGIN("hs_start_transport_stream_op_batch", 0); grpc_error *error = hs_mutate_op(exec_ctx, elem, op); if (error != GRPC_ERROR_NONE) { @@ -393,7 +393,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { /* grab pointers to our data from the call element */ - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; /* initialize members */ calld->call_combiner = args->call_combiner; GRPC_CLOSURE_INIT(&calld->hs_on_recv, hs_on_recv, elem, @@ -410,7 +410,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, grpc_closure *ignored) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; grpc_slice_buffer_destroy_internal(exec_ctx, &calld->read_slice_buffer); } diff --git a/src/core/ext/filters/load_reporting/server_load_reporting_filter.c b/src/core/ext/filters/load_reporting/server_load_reporting_filter.c index 7b8cf986f7..ca8a3b2a13 100644 --- a/src/core/ext/filters/load_reporting/server_load_reporting_filter.c +++ b/src/core/ext/filters/load_reporting/server_load_reporting_filter.c @@ -56,8 +56,8 @@ typedef struct channel_data { static void on_initial_md_ready(grpc_exec_ctx *exec_ctx, void *user_data, grpc_error *err) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (err == GRPC_ERROR_NONE) { if (calld->recv_initial_metadata->idx.named.path != NULL) { @@ -88,7 +88,7 @@ static void on_initial_md_ready(grpc_exec_ctx *exec_ctx, void *user_data, static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; calld->id = (intptr_t)args->call_stack; GRPC_CLOSURE_INIT(&calld->on_initial_md_ready, on_initial_md_ready, elem, grpc_schedule_on_exec_ctx); @@ -111,7 +111,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, grpc_closure *ignored) { - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; /* TODO(dgq): do something with the data channel_data *chand = elem->channel_data; @@ -141,7 +141,7 @@ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, grpc_channel_element_args *args) { GPR_ASSERT(!args->is_last); - channel_data *chand = elem->channel_data; + channel_data *chand = (channel_data *)elem->channel_data; chand->id = (intptr_t)args->channel_stack; /* TODO(dgq): do something with the data @@ -176,8 +176,8 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, static grpc_filtered_mdelem lr_trailing_md_filter(grpc_exec_ctx *exec_ctx, void *user_data, grpc_mdelem md) { - grpc_call_element *elem = user_data; - call_data *calld = elem->call_data; + grpc_call_element *elem = (grpc_call_element *)user_data; + call_data *calld = (call_data *)elem->call_data; if (grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_LB_COST_BIN)) { calld->trailing_md_string = GRPC_MDVALUE(md); return GRPC_FILTERED_REMOVE(); @@ -189,7 +189,7 @@ static void lr_start_transport_stream_op_batch( grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op_batch *op) { GPR_TIMER_BEGIN("lr_start_transport_stream_op_batch", 0); - call_data *calld = elem->call_data; + call_data *calld = (call_data *)elem->call_data; if (op->recv_initial_metadata) { /* substitute our callback for the higher callback */ diff --git a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c index 199cb883b3..f56afd27d2 100644 --- a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c +++ b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.c @@ -41,7 +41,7 @@ static bool maybe_add_server_load_reporting_filter( grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, void *arg) { const grpc_channel_args *args = grpc_channel_stack_builder_get_channel_arguments(builder); - const grpc_channel_filter *filter = arg; + const grpc_channel_filter *filter = (const grpc_channel_filter *)arg; grpc_channel_stack_builder_iterator *it = grpc_channel_stack_builder_iterator_find(builder, filter->name); const bool already_has_load_reporting_filter = diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.c b/src/core/ext/transport/chttp2/client/chttp2_connector.c index 983691bbad..0ec9353c04 100644 --- a/src/core/ext/transport/chttp2/client/chttp2_connector.c +++ b/src/core/ext/transport/chttp2/client/chttp2_connector.c @@ -93,8 +93,8 @@ static void chttp2_connector_shutdown(grpc_exec_ctx *exec_ctx, static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_handshaker_args *args = arg; - chttp2_connector *c = args->user_data; + grpc_handshaker_args *args = (grpc_handshaker_args *)arg; + chttp2_connector *c = (chttp2_connector *)args->user_data; gpr_mu_lock(&c->mu); if (error != GRPC_ERROR_NONE || c->shutdown) { if (error == GRPC_ERROR_NONE) { @@ -143,7 +143,7 @@ static void start_handshake_locked(grpc_exec_ctx *exec_ctx, } static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - chttp2_connector *c = arg; + chttp2_connector *c = (chttp2_connector *)arg; gpr_mu_lock(&c->mu); GPR_ASSERT(c->connecting); c->connecting = false; @@ -198,7 +198,7 @@ static const grpc_connector_vtable chttp2_connector_vtable = { chttp2_connector_connect}; grpc_connector *grpc_chttp2_connector_create() { - chttp2_connector *c = gpr_zalloc(sizeof(*c)); + chttp2_connector *c = (chttp2_connector *)gpr_zalloc(sizeof(*c)); c->base.vtable = &chttp2_connector_vtable; gpr_mu_init(&c->mu); gpr_ref_init(&c->refs, 1); diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.c b/src/core/ext/transport/chttp2/server/chttp2_server.c index feb95796b3..c2b2fc03aa 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.c +++ b/src/core/ext/transport/chttp2/server/chttp2_server.c @@ -60,8 +60,9 @@ typedef struct { static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_handshaker_args *args = arg; - server_connection_state *connection_state = args->user_data; + grpc_handshaker_args *args = (grpc_handshaker_args *)arg; + server_connection_state *connection_state = + (server_connection_state *)args->user_data; gpr_mu_lock(&connection_state->server_state->mu); if (error != GRPC_ERROR_NONE || connection_state->server_state->shutdown) { const char *error_str = grpc_error_string(error); @@ -108,7 +109,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, grpc_pollset *accepting_pollset, grpc_tcp_server_acceptor *acceptor) { - server_state *state = arg; + server_state *state = (server_state *)arg; gpr_mu_lock(&state->mu); if (state->shutdown) { gpr_mu_unlock(&state->mu); @@ -143,7 +144,7 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, static void server_start_listener(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg, grpc_pollset **pollsets, size_t pollset_count) { - server_state *state = arg; + server_state *state = (server_state *)arg; gpr_mu_lock(&state->mu); state->shutdown = false; gpr_mu_unlock(&state->mu); @@ -153,7 +154,7 @@ static void server_start_listener(grpc_exec_ctx *exec_ctx, grpc_server *server, static void tcp_server_shutdown_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - server_state *state = arg; + server_state *state = (server_state *)arg; /* ensure all threads have unlocked */ gpr_mu_lock(&state->mu); grpc_closure *destroy_done = state->server_destroy_listener_done; @@ -178,7 +179,7 @@ static void tcp_server_shutdown_complete(grpc_exec_ctx *exec_ctx, void *arg, static void server_destroy_listener(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg, grpc_closure *destroy_done) { - server_state *state = arg; + server_state *state = (server_state *)arg; gpr_mu_lock(&state->mu); state->shutdown = true; state->server_destroy_listener_done = destroy_done; @@ -208,7 +209,7 @@ grpc_error *grpc_chttp2_server_add_port(grpc_exec_ctx *exec_ctx, if (err != GRPC_ERROR_NONE) { goto error; } - state = gpr_zalloc(sizeof(*state)); + state = (server_state *)gpr_zalloc(sizeof(*state)); GRPC_CLOSURE_INIT(&state->tcp_server_shutdown_complete, tcp_server_shutdown_complete, state, grpc_schedule_on_exec_ctx); @@ -225,7 +226,7 @@ grpc_error *grpc_chttp2_server_add_port(grpc_exec_ctx *exec_ctx, gpr_mu_init(&state->mu); const size_t naddrs = resolved->naddrs; - errors = gpr_malloc(sizeof(*errors) * naddrs); + errors = (grpc_error **)gpr_malloc(sizeof(*errors) * naddrs); for (i = 0; i < naddrs; i++) { errors[i] = grpc_tcp_server_add_port(tcp_server, &resolved->addrs[i], &port_temp); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index a0bfa1676c..d77047cb9e 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -84,8 +84,6 @@ grpc_tracer_flag grpc_trace_chttp2_refcount = GRPC_TRACER_INITIALIZER(false, "chttp2_refcount"); #endif -static const grpc_transport_vtable vtable; - /* forward declarations of various callbacks that we'll build closures around */ static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); @@ -248,6 +246,8 @@ void grpc_chttp2_unref_transport(grpc_exec_ctx *exec_ctx, void grpc_chttp2_ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); } #endif +static const grpc_transport_vtable *get_vtable(void); + static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, const grpc_channel_args *channel_args, grpc_endpoint *ep, bool is_client) { @@ -257,7 +257,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) == GRPC_CHTTP2_CLIENT_CONNECT_STRLEN); - t->base.vtable = &vtable; + t->base.vtable = get_vtable(); t->ep = ep; /* one ref is for destroy */ gpr_ref_init(&t->refs, 1); @@ -542,11 +542,6 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } } - GRPC_CLOSURE_INIT(&t->write_action, write_action, t, - t->opt_target == GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT - ? grpc_executor_scheduler - : grpc_schedule_on_exec_ctx); - t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; t->ping_state.is_delayed_ping_timer_set = false; @@ -573,7 +568,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, static void destroy_transport_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; t->destroying = 1; close_transport_locked( exec_ctx, t, @@ -699,7 +694,7 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, grpc_error *error) { - grpc_chttp2_stream *s = sp; + grpc_chttp2_stream *s = (grpc_chttp2_stream *)sp; grpc_chttp2_transport *t = s->t; GPR_TIMER_BEGIN("destroy_stream", 0); @@ -783,7 +778,7 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, grpc_chttp2_stream *grpc_chttp2_parsing_lookup_stream(grpc_chttp2_transport *t, uint32_t id) { - return grpc_chttp2_stream_map_find(&t->stream_map, id); + return (grpc_chttp2_stream *)grpc_chttp2_stream_map_find(&t->stream_map, id); } grpc_chttp2_stream *grpc_chttp2_parsing_accept_stream(grpc_exec_ctx *exec_ctx, @@ -842,6 +837,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, switch (t->write_state) { case GRPC_CHTTP2_WRITE_STATE_IDLE: set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING, reason); + t->is_first_write_in_batch = true; GRPC_CHTTP2_REF_TRANSPORT(t, "writing"); GRPC_CLOSURE_SCHED( exec_ctx, @@ -860,52 +856,100 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, GPR_TIMER_END("grpc_chttp2_initiate_write", 0); } -void grpc_chttp2_become_writable( - grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, - grpc_chttp2_stream_write_type stream_write_type, const char *reason) { +void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + bool also_initiate_write, const char *reason) { if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s)) { GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:become"); } - switch (stream_write_type) { - case GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK: - break; - case GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED: - grpc_chttp2_initiate_write(exec_ctx, t, reason); - break; - case GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED: - grpc_chttp2_initiate_write(exec_ctx, t, reason); - break; + if (also_initiate_write) { + grpc_chttp2_initiate_write(exec_ctx, t, reason); + } +} + +static grpc_closure_scheduler *write_scheduler(grpc_chttp2_transport *t, + bool early_results_scheduled, + bool partial_write) { + /* if it's not the first write in a batch, always offload to the executor: + we'll probably end up queuing against the kernel anyway, so we'll likely + get better latency overall if we switch writing work elsewhere and continue + with application work above */ + if (!t->is_first_write_in_batch) { + return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT); + } + /* equivalently, if it's a partial write, we *know* we're going to be taking a + thread jump to write it because of the above, may as well do so + immediately */ + if (partial_write) { + return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT); } + switch (t->opt_target) { + case GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT: + /* executor gives us the largest probability of being able to batch a + * write with others on this transport */ + return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT); + case GRPC_CHTTP2_OPTIMIZE_FOR_LATENCY: + return grpc_schedule_on_exec_ctx; + } + GPR_UNREACHABLE_CODE(return NULL); +} + +#define WRITE_STATE_TUPLE_TO_INT(p, i) (2 * (int)(p) + (int)(i)) +static const char *begin_writing_desc(bool partial, bool inlined) { + switch (WRITE_STATE_TUPLE_TO_INT(partial, inlined)) { + case WRITE_STATE_TUPLE_TO_INT(false, false): + return "begin write in background"; + case WRITE_STATE_TUPLE_TO_INT(false, true): + return "begin write in current thread"; + case WRITE_STATE_TUPLE_TO_INT(true, false): + return "begin partial write in background"; + case WRITE_STATE_TUPLE_TO_INT(true, true): + return "begin partial write in current thread"; + } + GPR_UNREACHABLE_CODE(return "bad state tuple"); } static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *gt, grpc_error *error_ignored) { GPR_TIMER_BEGIN("write_action_begin_locked", 0); - grpc_chttp2_transport *t = gt; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; GPR_ASSERT(t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE); - switch (t->closed ? GRPC_CHTTP2_NOTHING_TO_WRITE - : grpc_chttp2_begin_write(exec_ctx, t)) { - case GRPC_CHTTP2_NOTHING_TO_WRITE: - set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_IDLE, - "begin writing nothing"); - GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing"); - break; - case GRPC_CHTTP2_PARTIAL_WRITE: - set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE, - "begin writing partial"); - GRPC_CLOSURE_SCHED(exec_ctx, &t->write_action, GRPC_ERROR_NONE); - break; - case GRPC_CHTTP2_FULL_WRITE: - set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING, - "begin writing"); - GRPC_CLOSURE_SCHED(exec_ctx, &t->write_action, GRPC_ERROR_NONE); - break; + grpc_chttp2_begin_write_result r; + if (t->closed) { + r.writing = false; + } else { + r = grpc_chttp2_begin_write(exec_ctx, t); + } + if (r.writing) { + if (r.partial) { + GRPC_STATS_INC_HTTP2_PARTIAL_WRITES(exec_ctx); + } + if (!t->is_first_write_in_batch) { + GRPC_STATS_INC_HTTP2_WRITES_CONTINUED(exec_ctx); + } + grpc_closure_scheduler *scheduler = + write_scheduler(t, r.early_results_scheduled, r.partial); + if (scheduler != grpc_schedule_on_exec_ctx) { + GRPC_STATS_INC_HTTP2_WRITES_OFFLOADED(exec_ctx); + } + set_write_state( + exec_ctx, t, r.partial ? GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE + : GRPC_CHTTP2_WRITE_STATE_WRITING, + begin_writing_desc(r.partial, scheduler == grpc_schedule_on_exec_ctx)); + GRPC_CLOSURE_SCHED(exec_ctx, GRPC_CLOSURE_INIT(&t->write_action, + write_action, t, scheduler), + GRPC_ERROR_NONE); + } else { + set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_IDLE, + "begin writing nothing"); + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing"); } GPR_TIMER_END("write_action_begin_locked", 0); } static void write_action(grpc_exec_ctx *exec_ctx, void *gt, grpc_error *error) { - grpc_chttp2_transport *t = gt; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; GPR_TIMER_BEGIN("write_action", 0); grpc_endpoint_write( exec_ctx, t->ep, &t->outbuf, @@ -917,7 +961,7 @@ static void write_action(grpc_exec_ctx *exec_ctx, void *gt, grpc_error *error) { static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { GPR_TIMER_BEGIN("terminate_writing_with_lock", 0); - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; if (error != GRPC_ERROR_NONE) { close_transport_locked(exec_ctx, t, GRPC_ERROR_REF(error)); @@ -942,7 +986,8 @@ static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp, case GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE: GPR_TIMER_MARK("state=writing_stale_no_poller", 0); set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING, - "continue writing [!covered]"); + "continue writing"); + t->is_first_write_in_batch = false; GRPC_CHTTP2_REF_TRANSPORT(t, "writing"); GRPC_CLOSURE_RUN( exec_ctx, @@ -1042,9 +1087,7 @@ static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream_map_add(&t->stream_map, s->id, s); post_destructive_reclaimer(exec_ctx, t); - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, - "new_stream"); + grpc_chttp2_become_writable(exec_ctx, t, s, true, "new_stream"); } /* cancel out streams that will never be started */ while (t->next_stream_id >= MAX_CLIENT_STREAM_ID && @@ -1093,12 +1136,14 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx, closure->next_data.scratch -= CLOSURE_BARRIER_FIRST_REF_BIT; if (GRPC_TRACER_ON(grpc_http_trace)) { const char *errstr = grpc_error_string(error); - gpr_log(GPR_DEBUG, - "complete_closure_step: %p refs=%d flags=0x%04x desc=%s err=%s", - closure, - (int)(closure->next_data.scratch / CLOSURE_BARRIER_FIRST_REF_BIT), - (int)(closure->next_data.scratch % CLOSURE_BARRIER_FIRST_REF_BIT), - desc, errstr); + gpr_log( + GPR_DEBUG, + "complete_closure_step: t=%p %p refs=%d flags=0x%04x desc=%s err=%s " + "write_state=%s", + t, closure, + (int)(closure->next_data.scratch / CLOSURE_BARRIER_FIRST_REF_BIT), + (int)(closure->next_data.scratch % CLOSURE_BARRIER_FIRST_REF_BIT), desc, + errstr, write_state_name(t->write_state)); } if (error != GRPC_ERROR_NONE) { if (closure->error_data.error == GRPC_ERROR_NONE) { @@ -1139,9 +1184,7 @@ static void maybe_become_writable_due_to_send_msg(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s) { if (s->id != 0 && (!s->write_buffering || s->flow_controlled_buffer.length > t->write_buffer_size)) { - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, - "op.send_message"); + grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message"); } } @@ -1173,15 +1216,19 @@ static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx, } else { grpc_chttp2_write_cb *cb = t->write_cb_pool; if (cb == NULL) { - cb = gpr_malloc(sizeof(*cb)); + cb = (grpc_chttp2_write_cb *)gpr_malloc(sizeof(*cb)); } else { t->write_cb_pool = cb->next; } cb->call_at_byte = notify_offset; cb->closure = s->fetching_send_message_finished; s->fetching_send_message_finished = NULL; - cb->next = s->on_write_finished_cbs; - s->on_write_finished_cbs = cb; + grpc_chttp2_write_cb **list = + s->fetching_send_message->flags & GRPC_WRITE_THROUGH + ? &s->on_write_finished_cbs + : &s->on_flow_controlled_cbs; + cb->next = *list; + *list = cb; } s->fetching_send_message = NULL; return; /* early out */ @@ -1201,7 +1248,7 @@ static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx, static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs, grpc_error *error) { - grpc_chttp2_stream *s = gs; + grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs; grpc_chttp2_transport *t = s->t; if (error == GRPC_ERROR_NONE) { error = grpc_byte_stream_pull(exec_ctx, s->fetching_send_message, @@ -1236,8 +1283,9 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, grpc_error *error_ignored) { GPR_TIMER_BEGIN("perform_stream_op_locked", 0); - grpc_transport_stream_op_batch *op = stream_op; - grpc_chttp2_stream *s = op->handler_private.extra_arg; + grpc_transport_stream_op_batch *op = + (grpc_transport_stream_op_batch *)stream_op; + grpc_chttp2_stream *s = (grpc_chttp2_stream *)op->handler_private.extra_arg; grpc_transport_stream_op_batch_payload *op_payload = op->payload; grpc_chttp2_transport *t = s->t; @@ -1290,7 +1338,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, if ((s->stream_compression_send_enabled = (op_payload->send_initial_metadata.send_initial_metadata->idx.named .content_encoding != NULL)) == true) { - s->compressed_data_buffer = gpr_malloc(sizeof(grpc_slice_buffer)); + s->compressed_data_buffer = + (grpc_slice_buffer *)gpr_malloc(sizeof(grpc_slice_buffer)); grpc_slice_buffer_init(s->compressed_data_buffer); } @@ -1336,14 +1385,13 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, } } else { GPR_ASSERT(s->id != 0); - grpc_chttp2_stream_write_type write_type = - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED; + bool initiate_write = true; if (op->send_message && (op->payload->send_message.send_message->flags & GRPC_WRITE_BUFFER_HINT)) { - write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK; + initiate_write = false; } - grpc_chttp2_become_writable(exec_ctx, t, s, write_type, + grpc_chttp2_become_writable(exec_ctx, t, s, initiate_write, "op.send_initial_metadata"); } } else { @@ -1452,8 +1500,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, } else if (s->id != 0) { /* TODO(ctiller): check if there's flow control for any outstanding bytes before going writable */ - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, + grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_trailing_metadata"); } } @@ -1578,7 +1625,7 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, static void retry_initiate_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; t->ping_state.is_delayed_ping_timer_set = false; grpc_chttp2_initiate_write(exec_ctx, t, "retry_send_ping"); } @@ -1634,8 +1681,9 @@ void grpc_chttp2_add_ping_strike(grpc_exec_ctx *exec_ctx, static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, grpc_error *error_ignored) { - grpc_transport_op *op = stream_op; - grpc_chttp2_transport *t = op->handler_private.extra_arg; + grpc_transport_op *op = (grpc_transport_op *)stream_op; + grpc_chttp2_transport *t = + (grpc_chttp2_transport *)op->handler_private.extra_arg; grpc_error *close_transport = op->disconnect_with_error; if (op->goaway_error) { @@ -1847,7 +1895,8 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_exec_ctx *exec_ctx, static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, uint32_t id, grpc_error *error) { - grpc_chttp2_stream *s = grpc_chttp2_stream_map_delete(&t->stream_map, id); + grpc_chttp2_stream *s = + (grpc_chttp2_stream *)grpc_chttp2_stream_map_delete(&t->stream_map, id); GPR_ASSERT(s); if (t->incoming_stream == s) { t->incoming_stream = NULL; @@ -1979,6 +2028,21 @@ static grpc_error *removal_error(grpc_error *extra_error, grpc_chttp2_stream *s, return error; } +static void flush_write_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, + grpc_chttp2_stream *s, grpc_chttp2_write_cb **list, + grpc_error *error) { + while (*list) { + grpc_chttp2_write_cb *cb = *list; + *list = cb->next; + grpc_chttp2_complete_closure_step(exec_ctx, t, s, &cb->closure, + GRPC_ERROR_REF(error), + "on_write_finished_cb"); + cb->next = t->write_cb_pool; + t->write_cb_pool = cb; + } + GRPC_ERROR_UNREF(error); +} + void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_error *error) { @@ -1998,16 +2062,9 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx, grpc_chttp2_complete_closure_step( exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_REF(error), "fetching_send_message_finished"); - while (s->on_write_finished_cbs) { - grpc_chttp2_write_cb *cb = s->on_write_finished_cbs; - s->on_write_finished_cbs = cb->next; - grpc_chttp2_complete_closure_step(exec_ctx, t, s, &cb->closure, - GRPC_ERROR_REF(error), - "on_write_finished_cb"); - cb->next = t->write_cb_pool; - t->write_cb_pool = cb; - } - GRPC_ERROR_UNREF(error); + flush_write_list(exec_ctx, t, s, &s->on_write_finished_cbs, + GRPC_ERROR_REF(error)); + flush_write_list(exec_ctx, t, s, &s->on_flow_controlled_cbs, error); } void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx, @@ -2227,8 +2284,8 @@ typedef struct { } cancel_stream_cb_args; static void cancel_stream_cb(void *user_data, uint32_t key, void *stream) { - cancel_stream_cb_args *args = user_data; - grpc_chttp2_stream *s = stream; + cancel_stream_cb_args *args = (cancel_stream_cb_args *)user_data; + grpc_chttp2_stream *s = (grpc_chttp2_stream *)stream; grpc_chttp2_cancel_stream(args->exec_ctx, args->t, s, GRPC_ERROR_REF(args->error)); } @@ -2252,13 +2309,11 @@ void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx, case GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED: break; case GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY: - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, + grpc_chttp2_become_writable(exec_ctx, t, s, true, "immediate stream flowctl"); break; case GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE: - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK, + grpc_chttp2_become_writable(exec_ctx, t, s, false, "queue stream flowctl"); break; } @@ -2330,7 +2385,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { GPR_TIMER_BEGIN("reading_action_locked", 0); - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; GRPC_ERROR_REF(error); @@ -2371,9 +2426,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, if (t->flow_control.initial_window_update > 0) { grpc_chttp2_stream *s; while (grpc_chttp2_list_pop_stalled_by_stream(t, &s)) { - grpc_chttp2_become_writable( - exec_ctx, t, s, GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED, - "unstalled"); + grpc_chttp2_become_writable(exec_ctx, t, s, true, "unstalled"); } } t->flow_control.initial_window_update = 0; @@ -2415,7 +2468,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; if (GRPC_TRACER_ON(grpc_http_trace)) { gpr_log(GPR_DEBUG, "%s: Start BDP ping", t->peer_string); } @@ -2428,7 +2481,7 @@ static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; if (GRPC_TRACER_ON(grpc_http_trace)) { gpr_log(GPR_DEBUG, "%s: Complete BDP ping", t->peer_string); } @@ -2477,7 +2530,7 @@ void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args, static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; GPR_ASSERT(t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING); if (t->destroying || t->closed) { t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING; @@ -2507,7 +2560,7 @@ static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, static void start_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; GRPC_CHTTP2_REF_TRANSPORT(t, "keepalive watchdog"); grpc_timer_init(exec_ctx, &t->keepalive_watchdog_timer, grpc_exec_ctx_now(exec_ctx) + t->keepalive_time, @@ -2516,7 +2569,7 @@ static void start_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) { if (error == GRPC_ERROR_NONE) { t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_WAITING; @@ -2532,7 +2585,7 @@ static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) { if (error == GRPC_ERROR_NONE) { t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING; @@ -2613,7 +2666,8 @@ static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx, void *argp, grpc_error *error_ignored) { - grpc_chttp2_incoming_byte_stream *bs = argp; + grpc_chttp2_incoming_byte_stream *bs = + (grpc_chttp2_incoming_byte_stream *)argp; grpc_chttp2_transport *t = bs->transport; grpc_chttp2_stream *s = bs->stream; @@ -2823,7 +2877,8 @@ static const grpc_byte_stream_vtable grpc_chttp2_incoming_byte_stream_vtable = { static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx, void *byte_stream, grpc_error *error_ignored) { - grpc_chttp2_incoming_byte_stream *bs = byte_stream; + grpc_chttp2_incoming_byte_stream *bs = + (grpc_chttp2_incoming_byte_stream *)byte_stream; grpc_chttp2_stream *s = bs->stream; grpc_chttp2_transport *t = s->t; @@ -2879,7 +2934,7 @@ static void post_destructive_reclaimer(grpc_exec_ctx *exec_ctx, static void benign_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; if (error == GRPC_ERROR_NONE && grpc_chttp2_stream_map_size(&t->stream_map) == 0) { /* Channel with no active streams: send a goaway to try and make it @@ -2909,11 +2964,12 @@ static void benign_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, static void destructive_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_chttp2_transport *t = arg; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; size_t n = grpc_chttp2_stream_map_size(&t->stream_map); t->destructive_reclaimer_registered = false; if (error == GRPC_ERROR_NONE && n > 0) { - grpc_chttp2_stream *s = grpc_chttp2_stream_map_rand(&t->stream_map); + grpc_chttp2_stream *s = + (grpc_chttp2_stream *)grpc_chttp2_stream_map_rand(&t->stream_map); if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "HTTP2: %s - abandon stream id %d", t->peer_string, s->id); @@ -2957,10 +3013,13 @@ static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream), destroy_transport, chttp2_get_endpoint}; +static const grpc_transport_vtable *get_vtable(void) { return &vtable; } + grpc_transport *grpc_create_chttp2_transport( grpc_exec_ctx *exec_ctx, const grpc_channel_args *channel_args, grpc_endpoint *ep, int is_client) { - grpc_chttp2_transport *t = gpr_zalloc(sizeof(grpc_chttp2_transport)); + grpc_chttp2_transport *t = + (grpc_chttp2_transport *)gpr_zalloc(sizeof(grpc_chttp2_transport)); init_transport(exec_ctx, t, channel_args, ep, is_client != 0); return &t->base; } diff --git a/src/core/ext/transport/chttp2/transport/frame_goaway.c b/src/core/ext/transport/chttp2/transport/frame_goaway.c index 4bce84f21c..78ec08e177 100644 --- a/src/core/ext/transport/chttp2/transport/frame_goaway.c +++ b/src/core/ext/transport/chttp2/transport/frame_goaway.c @@ -46,7 +46,7 @@ grpc_error *grpc_chttp2_goaway_parser_begin_frame(grpc_chttp2_goaway_parser *p, gpr_free(p->debug_data); p->debug_length = length - 8; - p->debug_data = gpr_malloc(p->debug_length); + p->debug_data = (char *)gpr_malloc(p->debug_length); p->debug_pos = 0; p->state = GRPC_CHTTP2_GOAWAY_LSI0; return GRPC_ERROR_NONE; @@ -60,7 +60,7 @@ grpc_error *grpc_chttp2_goaway_parser_parse(grpc_exec_ctx *exec_ctx, uint8_t *const beg = GRPC_SLICE_START_PTR(slice); uint8_t *const end = GRPC_SLICE_END_PTR(slice); uint8_t *cur = beg; - grpc_chttp2_goaway_parser *p = parser; + grpc_chttp2_goaway_parser *p = (grpc_chttp2_goaway_parser *)parser; switch (p->state) { case GRPC_CHTTP2_GOAWAY_LSI0: diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.c b/src/core/ext/transport/chttp2/transport/frame_ping.c index 31673e622d..e99bc2eb4e 100644 --- a/src/core/ext/transport/chttp2/transport/frame_ping.c +++ b/src/core/ext/transport/chttp2/transport/frame_ping.c @@ -75,7 +75,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, uint8_t *const beg = GRPC_SLICE_START_PTR(slice); uint8_t *const end = GRPC_SLICE_END_PTR(slice); uint8_t *cur = beg; - grpc_chttp2_ping_parser *p = parser; + grpc_chttp2_ping_parser *p = (grpc_chttp2_ping_parser *)parser; while (p->byte != 8 && cur != end) { p->opaque_8bytes |= (((uint64_t)*cur) << (56 - 8 * p->byte)); @@ -111,7 +111,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, if (!g_disable_ping_ack) { if (t->ping_ack_count == t->ping_ack_capacity) { t->ping_ack_capacity = GPR_MAX(t->ping_ack_capacity * 3 / 2, 3); - t->ping_acks = gpr_realloc( + t->ping_acks = (uint64_t *)gpr_realloc( t->ping_acks, t->ping_ack_capacity * sizeof(*t->ping_acks)); } t->ping_acks[t->ping_ack_count++] = p->opaque_8bytes; diff --git a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c index 689dc8935c..0133b6efa2 100644 --- a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c +++ b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c @@ -77,7 +77,7 @@ grpc_error *grpc_chttp2_rst_stream_parser_parse(grpc_exec_ctx *exec_ctx, uint8_t *const beg = GRPC_SLICE_START_PTR(slice); uint8_t *const end = GRPC_SLICE_END_PTR(slice); uint8_t *cur = beg; - grpc_chttp2_rst_stream_parser *p = parser; + grpc_chttp2_rst_stream_parser *p = (grpc_chttp2_rst_stream_parser *)parser; while (p->byte != 4 && cur != end) { p->reason_bytes[p->byte] = *cur; diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.c b/src/core/ext/transport/chttp2/transport/frame_settings.c index 057d3d9ed3..806100adaa 100644 --- a/src/core/ext/transport/chttp2/transport/frame_settings.c +++ b/src/core/ext/transport/chttp2/transport/frame_settings.c @@ -111,7 +111,7 @@ grpc_error *grpc_chttp2_settings_parser_parse(grpc_exec_ctx *exec_ctx, void *p, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_slice slice, int is_last) { - grpc_chttp2_settings_parser *parser = p; + grpc_chttp2_settings_parser *parser = (grpc_chttp2_settings_parser *)p; const uint8_t *cur = GRPC_SLICE_START_PTR(slice); const uint8_t *end = GRPC_SLICE_END_PTR(slice); char *msg; diff --git a/src/core/ext/transport/chttp2/transport/frame_window_update.c b/src/core/ext/transport/chttp2/transport/frame_window_update.c index 65f3b01d77..c94f7725bf 100644 --- a/src/core/ext/transport/chttp2/transport/frame_window_update.c +++ b/src/core/ext/transport/chttp2/transport/frame_window_update.c @@ -70,7 +70,8 @@ grpc_error *grpc_chttp2_window_update_parser_parse( uint8_t *const beg = GRPC_SLICE_START_PTR(slice); uint8_t *const end = GRPC_SLICE_END_PTR(slice); uint8_t *cur = beg; - grpc_chttp2_window_update_parser *p = parser; + grpc_chttp2_window_update_parser *p = + (grpc_chttp2_window_update_parser *)parser; while (p->byte != 4 && cur != end) { p->amount |= ((uint32_t)*cur) << (8 * (3 - p->byte)); @@ -98,9 +99,8 @@ grpc_error *grpc_chttp2_window_update_parser_parse( grpc_chttp2_flowctl_recv_stream_update( &t->flow_control, &s->flow_control, received_update); if (grpc_chttp2_list_remove_stalled_by_stream(t, s)) { - grpc_chttp2_become_writable( - exec_ctx, t, s, GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED, - "stream.read_flow_control"); + grpc_chttp2_become_writable(exec_ctx, t, s, true, + "stream.read_flow_control"); } } } else { diff --git a/src/core/ext/transport/chttp2/transport/hpack_encoder.c b/src/core/ext/transport/chttp2/transport/hpack_encoder.c index 207ef6c451..96ac4b094b 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_encoder.c +++ b/src/core/ext/transport/chttp2/transport/hpack_encoder.c @@ -536,7 +536,7 @@ void grpc_chttp2_hpack_compressor_init(grpc_chttp2_hpack_compressor *c) { c->max_table_elems = c->cap_table_elems; c->max_usable_size = GRPC_CHTTP2_HPACKC_INITIAL_TABLE_SIZE; c->table_elem_size = - gpr_malloc(sizeof(*c->table_elem_size) * c->cap_table_elems); + (uint16_t *)gpr_malloc(sizeof(*c->table_elem_size) * c->cap_table_elems); memset(c->table_elem_size, 0, sizeof(*c->table_elem_size) * c->cap_table_elems); for (size_t i = 0; i < GPR_ARRAY_SIZE(c->entries_keys); i++) { @@ -564,7 +564,8 @@ void grpc_chttp2_hpack_compressor_set_max_usable_size( } static void rebuild_elems(grpc_chttp2_hpack_compressor *c, uint32_t new_cap) { - uint16_t *table_elem_size = gpr_malloc(sizeof(*table_elem_size) * new_cap); + uint16_t *table_elem_size = + (uint16_t *)gpr_malloc(sizeof(*table_elem_size) * new_cap); uint32_t i; memset(table_elem_size, 0, sizeof(*table_elem_size) * new_cap); diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.c b/src/core/ext/transport/chttp2/transport/hpack_parser.c index c21d76ba71..82ff2c8e2c 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_parser.c +++ b/src/core/ext/transport/chttp2/transport/hpack_parser.c @@ -1284,7 +1284,7 @@ static void append_bytes(grpc_chttp2_hpack_parser_string *str, GPR_ASSERT(str->data.copied.length + length <= UINT32_MAX); str->data.copied.capacity = (uint32_t)(str->data.copied.length + length); str->data.copied.str = - gpr_realloc(str->data.copied.str, str->data.copied.capacity); + (char *)gpr_realloc(str->data.copied.str, str->data.copied.capacity); } memcpy(str->data.copied.str + str->data.copied.length, data, length); GPR_ASSERT(length <= UINT32_MAX - str->data.copied.length); @@ -1643,7 +1643,7 @@ static const maybe_complete_func_type maybe_complete_funcs[] = { static void force_client_rst_stream(grpc_exec_ctx *exec_ctx, void *sp, grpc_error *error) { - grpc_chttp2_stream *s = sp; + grpc_chttp2_stream *s = (grpc_chttp2_stream *)sp; grpc_chttp2_transport *t = s->t; if (!s->write_closed) { grpc_slice_buffer_add( @@ -1665,7 +1665,8 @@ static void parse_stream_compression_md(grpc_exec_ctx *exec_ctx, if (!grpc_slice_eq(content_encoding, GRPC_MDSTR_IDENTITY)) { if (grpc_slice_eq(content_encoding, GRPC_MDSTR_GZIP)) { s->stream_compression_recv_enabled = true; - s->decompressed_data_buffer = gpr_malloc(sizeof(grpc_slice_buffer)); + s->decompressed_data_buffer = + (grpc_slice_buffer *)gpr_malloc(sizeof(grpc_slice_buffer)); grpc_slice_buffer_init(s->decompressed_data_buffer); } } @@ -1677,7 +1678,7 @@ grpc_error *grpc_chttp2_header_parser_parse(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_slice slice, int is_last) { - grpc_chttp2_hpack_parser *parser = hpack_parser; + grpc_chttp2_hpack_parser *parser = (grpc_chttp2_hpack_parser *)hpack_parser; GPR_TIMER_BEGIN("grpc_chttp2_hpack_parser_parse", 0); if (s != NULL) { s->stats.incoming.header_bytes += GRPC_SLICE_LENGTH(slice); diff --git a/src/core/ext/transport/chttp2/transport/hpack_table.c b/src/core/ext/transport/chttp2/transport/hpack_table.c index 944d778011..bbd135a318 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_table.c +++ b/src/core/ext/transport/chttp2/transport/hpack_table.c @@ -173,7 +173,7 @@ void grpc_chttp2_hptbl_init(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl) { GRPC_CHTTP2_INITIAL_HPACK_TABLE_SIZE; tbl->max_entries = tbl->cap_entries = entries_for_bytes(tbl->current_table_bytes); - tbl->ents = gpr_malloc(sizeof(*tbl->ents) * tbl->cap_entries); + tbl->ents = (grpc_mdelem *)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_slices( @@ -228,7 +228,7 @@ static void evict1(grpc_exec_ctx *exec_ctx, grpc_chttp2_hptbl *tbl) { } static void rebuild_ents(grpc_chttp2_hptbl *tbl, uint32_t new_cap) { - grpc_mdelem *ents = gpr_malloc(sizeof(*ents) * new_cap); + grpc_mdelem *ents = (grpc_mdelem *)gpr_malloc(sizeof(*ents) * new_cap); uint32_t i; for (i = 0; i < tbl->num_ents; i++) { diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 901be651d0..46930dd22c 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -262,6 +262,10 @@ struct grpc_chttp2_transport { /** write execution state of the transport */ grpc_chttp2_write_state write_state; + /** is this the first write in a series of writes? + set when we initiate writing from idle, cleared when we + initiate writing from writing+more */ + bool is_first_write_in_batch; /** is the transport destroying itself? */ uint8_t destroying; @@ -483,6 +487,7 @@ struct grpc_chttp2_stream { grpc_slice fetching_slice; int64_t next_message_end_offset; int64_t flow_controlled_bytes_written; + int64_t flow_controlled_bytes_flowed; grpc_closure complete_fetch_locked; grpc_closure *fetching_send_message_finished; @@ -555,6 +560,7 @@ struct grpc_chttp2_stream { grpc_slice_buffer flow_controlled_buffer; + grpc_chttp2_write_cb *on_flow_controlled_cbs; grpc_chttp2_write_cb *on_write_finished_cbs; grpc_chttp2_write_cb *finish_after_write; size_t sending_bytes; @@ -595,10 +601,13 @@ struct grpc_chttp2_stream { void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, const char *reason); -typedef enum { - GRPC_CHTTP2_NOTHING_TO_WRITE, - GRPC_CHTTP2_PARTIAL_WRITE, - GRPC_CHTTP2_FULL_WRITE, +typedef struct { + /** are we writing? */ + bool writing; + /** if writing: was it a complete flush (false) or a partial flush (true) */ + bool partial; + /** did we queue any completions as part of beginning the write */ + bool early_results_scheduled; } grpc_chttp2_begin_write_result; grpc_chttp2_begin_write_result grpc_chttp2_begin_write( @@ -840,22 +849,12 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, void grpc_chttp2_add_ping_strike(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t); -typedef enum { - /* don't initiate a transport write, but piggyback on the next one */ - GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK, - /* initiate a covered write */ - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, - /* initiate an uncovered write */ - GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED -} grpc_chttp2_stream_write_type; - /** add a ref to the stream and add it to the writable list; ref will be dropped in writing.c */ void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, - grpc_chttp2_stream_write_type type, - const char *reason); + bool also_initiate_write, const char *reason); void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index bc1e773ae4..1d165c6cda 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -106,7 +106,8 @@ grpc_error *grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx, return err; } ++cur; - ++t->deframe_state; + t->deframe_state = + (grpc_chttp2_deframe_transport_state)(1 + (int)t->deframe_state); } if (cur == end) { return GRPC_ERROR_NONE; @@ -402,7 +403,7 @@ static void free_timeout(void *p) { gpr_free(p); } static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp, grpc_mdelem md) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; grpc_chttp2_stream *s = t->incoming_stream; GPR_TIMER_BEGIN("on_initial_header", 0); @@ -430,7 +431,7 @@ static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp, grpc_millis timeout; if (cached_timeout == NULL) { /* not already parsed: parse it now, and store the result away */ - cached_timeout = gpr_malloc(sizeof(grpc_millis)); + cached_timeout = (grpc_millis *)gpr_malloc(sizeof(grpc_millis)); if (!grpc_http2_decode_timeout(GRPC_MDVALUE(md), cached_timeout)) { char *val = grpc_slice_to_c_string(GRPC_MDVALUE(md)); gpr_log(GPR_ERROR, "Ignoring bad timeout value '%s'", val); @@ -481,7 +482,7 @@ static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp, static void on_trailing_header(grpc_exec_ctx *exec_ctx, void *tp, grpc_mdelem md) { - grpc_chttp2_transport *t = tp; + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; grpc_chttp2_stream *s = t->incoming_stream; GPR_TIMER_BEGIN("on_trailing_header", 0); diff --git a/src/core/ext/transport/chttp2/transport/stream_map.c b/src/core/ext/transport/chttp2/transport/stream_map.c index e2f10bc208..650090d8f0 100644 --- a/src/core/ext/transport/chttp2/transport/stream_map.c +++ b/src/core/ext/transport/chttp2/transport/stream_map.c @@ -27,8 +27,8 @@ void grpc_chttp2_stream_map_init(grpc_chttp2_stream_map *map, size_t initial_capacity) { GPR_ASSERT(initial_capacity > 1); - map->keys = gpr_malloc(sizeof(uint32_t) * initial_capacity); - map->values = gpr_malloc(sizeof(void *) * initial_capacity); + map->keys = (uint32_t *)gpr_malloc(sizeof(uint32_t) * initial_capacity); + map->values = (void **)gpr_malloc(sizeof(void *) * initial_capacity); map->count = 0; map->free = 0; map->capacity = initial_capacity; diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index dea7a9ad5b..998f152398 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -122,15 +122,18 @@ static void maybe_initiate_ping(grpc_exec_ctx *exec_ctx, (t->ping_state.pings_before_data_required != 0); } -static void update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, +static bool update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, int64_t send_bytes, - grpc_chttp2_write_cb **list, grpc_error *error) { + grpc_chttp2_write_cb **list, int64_t *ctr, + grpc_error *error) { + bool sched_any = false; grpc_chttp2_write_cb *cb = *list; *list = NULL; - s->flow_controlled_bytes_written += send_bytes; + *ctr += send_bytes; while (cb) { grpc_chttp2_write_cb *next = cb->next; - if (cb->call_at_byte <= s->flow_controlled_bytes_written) { + if (cb->call_at_byte <= *ctr) { + sched_any = true; finish_write_cb(exec_ctx, t, s, cb, GRPC_ERROR_REF(error)); } else { add_to_write_list(list, cb); @@ -138,6 +141,7 @@ static void update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, cb = next; } GRPC_ERROR_UNREF(error); + return sched_any; } static bool stream_ref_if_not_destroyed(gpr_refcount *r) { @@ -163,6 +167,13 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { grpc_chttp2_stream *s; + /* stats histogram counters: we increment these throughout this function, + and at the end publish to the central stats histograms */ + int flow_control_writes = 0; + int initial_metadata_writes = 0; + int trailing_metadata_writes = 0; + int message_writes = 0; + GRPC_STATS_INC_HTTP2_WRITES_BEGUN(exec_ctx); GPR_TIMER_BEGIN("grpc_chttp2_begin_write", 0); @@ -176,6 +187,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( t->force_send_settings = 0; t->dirtied_local_settings = 0; t->sent_local_settings = 1; + GRPC_STATS_INC_HTTP2_SETTINGS_WRITES(exec_ctx); } /* simple writes are queued to qbuf, and flushed here */ @@ -195,13 +207,13 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( } } - bool partial_write = false; + grpc_chttp2_begin_write_result result = {false, false, false}; /* for each grpc_chttp2_stream that's become writable, frame it's data (according to available window sizes) and add to the output buffer */ while (true) { if (t->outbuf.length > target_write_size(t)) { - partial_write = true; + result.partial = true; break; } @@ -245,13 +257,13 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( .stats = &s->stats.outgoing}; grpc_chttp2_encode_header(exec_ctx, &t->hpack_compressor, NULL, 0, s->send_initial_metadata, &hopt, &t->outbuf); - now_writing = true; t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; if (!t->is_client) { t->ping_recv_state.last_ping_recv_time = 0; t->ping_recv_state.ping_strikes = 0; } + initial_metadata_writes++; } else { GRPC_CHTTP2_IF_TRACING( gpr_log(GPR_INFO, "not sending initial_metadata (Trailers-Only)")); @@ -267,10 +279,15 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( [num_extra_headers_for_trailing_metadata++] = &s->send_initial_metadata->idx.named.content_type->md; } + trailing_metadata_writes++; } s->send_initial_metadata = NULL; s->sent_initial_metadata = true; sent_initial_metadata = true; + result.early_results_scheduled = true; + grpc_chttp2_complete_closure_step( + exec_ctx, t, s, &s->send_initial_metadata_finished, GRPC_ERROR_NONE, + "send_initial_metadata_finished"); } /* send any window updates */ @@ -286,6 +303,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( t->ping_recv_state.last_ping_recv_time = 0; t->ping_recv_state.ping_strikes = 0; } + flow_control_writes++; } if (sent_initial_metadata) { /* send any body bytes, if allowed by flow control */ @@ -304,6 +322,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( if (max_outgoing > 0) { bool is_last_data_frame = false; bool is_last_frame = false; + size_t sending_bytes_before = s->sending_bytes; if (s->stream_compression_send_enabled) { while ((s->flow_controlled_buffer.length > 0 || s->compressed_data_buffer->length > 0) && @@ -370,6 +389,11 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( &s->stats.outgoing)); } } + result.early_results_scheduled |= + update_list(exec_ctx, t, s, + (int64_t)(s->sending_bytes - sending_bytes_before), + &s->on_flow_controlled_cbs, + &s->flow_controlled_bytes_flowed, GRPC_ERROR_NONE); now_writing = true; if (s->flow_controlled_buffer.length > 0 || (s->stream_compression_send_enabled && @@ -377,6 +401,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:fork"); grpc_chttp2_list_add_writable_stream(t, s); } + message_writes++; } else if (t->flow_control.remote_window == 0) { grpc_chttp2_list_add_stalled_by_transport(t, s); now_writing = true; @@ -412,6 +437,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( num_extra_headers_for_trailing_metadata, s->send_trailing_metadata, &hopt, &t->outbuf); + trailing_metadata_writes++; } s->send_trailing_metadata = NULL; s->sent_trailing_metadata = true; @@ -421,10 +447,22 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( s->id, GRPC_HTTP2_NO_ERROR, &s->stats.outgoing)); } now_writing = true; + result.early_results_scheduled = true; + grpc_chttp2_complete_closure_step( + exec_ctx, t, s, &s->send_trailing_metadata_finished, + GRPC_ERROR_NONE, "send_trailing_metadata_finished"); } } if (now_writing) { + GRPC_STATS_INC_HTTP2_SEND_INITIAL_METADATA_PER_WRITE( + exec_ctx, initial_metadata_writes); + GRPC_STATS_INC_HTTP2_SEND_MESSAGE_PER_WRITE(exec_ctx, message_writes); + GRPC_STATS_INC_HTTP2_SEND_TRAILING_METADATA_PER_WRITE( + exec_ctx, trailing_metadata_writes); + GRPC_STATS_INC_HTTP2_SEND_FLOWCTL_PER_WRITE(exec_ctx, + flow_control_writes); + if (!grpc_chttp2_list_add_writing_stream(t, s)) { /* already in writing list: drop ref */ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing:already_writing"); @@ -461,9 +499,8 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( GPR_TIMER_END("grpc_chttp2_begin_write", 0); - return t->outbuf.count > 0 ? (partial_write ? GRPC_CHTTP2_PARTIAL_WRITE - : GRPC_CHTTP2_FULL_WRITE) - : GRPC_CHTTP2_NOTHING_TO_WRITE; + result.writing = t->outbuf.count > 0; + return result; } void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -472,20 +509,13 @@ void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s; while (grpc_chttp2_list_pop_writing_stream(t, &s)) { - if (s->sent_initial_metadata) { - grpc_chttp2_complete_closure_step( - exec_ctx, t, s, &s->send_initial_metadata_finished, - GRPC_ERROR_REF(error), "send_initial_metadata_finished"); - } if (s->sending_bytes != 0) { update_list(exec_ctx, t, s, (int64_t)s->sending_bytes, - &s->on_write_finished_cbs, GRPC_ERROR_REF(error)); + &s->on_write_finished_cbs, &s->flow_controlled_bytes_written, + GRPC_ERROR_REF(error)); s->sending_bytes = 0; } if (s->sent_trailing_metadata) { - grpc_chttp2_complete_closure_step( - exec_ctx, t, s, &s->send_trailing_metadata_finished, - GRPC_ERROR_REF(error), "send_trailing_metadata_finished"); grpc_chttp2_mark_stream_closed(exec_ctx, t, s, !t->is_client, 1, GRPC_ERROR_REF(error)); } diff --git a/src/core/ext/transport/inproc/inproc_transport.c b/src/core/ext/transport/inproc/inproc_transport.c index be50db84ca..49e507cc78 100644 --- a/src/core/ext/transport/inproc/inproc_transport.c +++ b/src/core/ext/transport/inproc/inproc_transport.c @@ -120,7 +120,7 @@ static void slice_buffer_list_append_entry(slice_buffer_list *l, } static grpc_slice_buffer *slice_buffer_list_append(slice_buffer_list *l) { - sb_list_entry *next = gpr_malloc(sizeof(*next)); + sb_list_entry *next = (sb_list_entry *)gpr_malloc(sizeof(*next)); grpc_slice_buffer_init(&next->sb); slice_buffer_list_append_entry(l, next); return &next->sb; @@ -327,7 +327,8 @@ static grpc_error *fill_in_metadata(grpc_exec_ctx *exec_ctx, inproc_stream *s, grpc_error *error = GRPC_ERROR_NONE; for (grpc_linked_mdelem *elem = metadata->list.head; (elem != NULL) && (error == GRPC_ERROR_NONE); elem = elem->next) { - grpc_linked_mdelem *nelem = gpr_arena_alloc(s->arena, sizeof(*nelem)); + grpc_linked_mdelem *nelem = + (grpc_linked_mdelem *)gpr_arena_alloc(s->arena, sizeof(*nelem)); nelem->md = grpc_mdelem_from_slices( exec_ctx, grpc_slice_intern(GRPC_MDKEY(elem->md)), grpc_slice_intern(GRPC_MDVALUE(elem->md))); @@ -531,12 +532,14 @@ static void fail_helper_locked(grpc_exec_ctx *exec_ctx, inproc_stream *s, // since it expects that as well as no error yet grpc_metadata_batch fake_md; grpc_metadata_batch_init(&fake_md); - grpc_linked_mdelem *path_md = gpr_arena_alloc(s->arena, sizeof(*path_md)); + grpc_linked_mdelem *path_md = + (grpc_linked_mdelem *)gpr_arena_alloc(s->arena, sizeof(*path_md)); path_md->md = grpc_mdelem_from_slices(exec_ctx, g_fake_path_key, g_fake_path_value); GPR_ASSERT(grpc_metadata_batch_link_tail(exec_ctx, &fake_md, path_md) == GRPC_ERROR_NONE); - grpc_linked_mdelem *auth_md = gpr_arena_alloc(s->arena, sizeof(*auth_md)); + grpc_linked_mdelem *auth_md = + (grpc_linked_mdelem *)gpr_arena_alloc(s->arena, sizeof(*auth_md)); auth_md->md = grpc_mdelem_from_slices(exec_ctx, g_fake_auth_key, g_fake_auth_value); GPR_ASSERT(grpc_metadata_batch_link_tail(exec_ctx, &fake_md, auth_md) == @@ -1172,8 +1175,8 @@ static void inproc_transports_create(grpc_exec_ctx *exec_ctx, grpc_transport **client_transport, const grpc_channel_args *client_args) { INPROC_LOG(GPR_DEBUG, "inproc_transports_create"); - inproc_transport *st = gpr_zalloc(sizeof(*st)); - inproc_transport *ct = gpr_zalloc(sizeof(*ct)); + inproc_transport *st = (inproc_transport *)gpr_zalloc(sizeof(*st)); + inproc_transport *ct = (inproc_transport *)gpr_zalloc(sizeof(*ct)); // Share one lock between both sides since both sides get affected st->mu = ct->mu = gpr_malloc(sizeof(*st->mu)); gpr_mu_init(&st->mu->mu); |