diff options
Diffstat (limited to 'src/core/ext')
24 files changed, 465 insertions, 343 deletions
diff --git a/src/core/ext/filters/client_channel/OWNERS b/src/core/ext/filters/client_channel/OWNERS index 8f5e92808e..c8760d947b 100644 --- a/src/core/ext/filters/client_channel/OWNERS +++ b/src/core/ext/filters/client_channel/OWNERS @@ -1,4 +1,4 @@ set noparent @markdroth @dgquintas -@a11r +@AspirinSJL diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index df57fe5bfb..80a647fa94 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -174,7 +174,7 @@ static void set_channel_connectivity_state_locked(channel_data* chand, } } if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: setting connectivity state to %s", chand, + gpr_log(GPR_INFO, "chand=%p: setting connectivity state to %s", chand, grpc_connectivity_state_name(state)); } grpc_connectivity_state_set(&chand->state_tracker, state, error, reason); @@ -186,7 +186,7 @@ static void on_lb_policy_state_changed_locked(void* arg, grpc_error* error) { /* check if the notification is for the latest policy */ if (w->lb_policy == w->chand->lb_policy.get()) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: lb_policy=%p state changed to %s", w->chand, + gpr_log(GPR_INFO, "chand=%p: lb_policy=%p state changed to %s", w->chand, w->lb_policy, grpc_connectivity_state_name(w->state)); } set_channel_connectivity_state_locked(w->chand, w->state, @@ -215,7 +215,7 @@ static void watch_lb_policy_locked(channel_data* chand, static void start_resolving_locked(channel_data* chand) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: starting name resolution", chand); + gpr_log(GPR_INFO, "chand=%p: starting name resolution", chand); } GPR_ASSERT(!chand->started_resolving); chand->started_resolving = true; @@ -297,7 +297,7 @@ static void request_reresolution_locked(void* arg, grpc_error* error) { return; } if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: started name re-resolving", chand); + gpr_log(GPR_INFO, "chand=%p: started name re-resolving", chand); } chand->resolver->RequestReresolutionLocked(); // Give back the closure to the LB policy. @@ -311,7 +311,7 @@ static void request_reresolution_locked(void* arg, grpc_error* error) { static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { channel_data* chand = static_cast<channel_data*>(arg); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p: got resolver result: resolver_result=%p error=%s", chand, chand->resolver_result, grpc_error_string(error)); } @@ -431,7 +431,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { } } if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p: resolver result: lb_policy_name=\"%s\"%s, " "service_config=\"%s\"", chand, lb_policy_name_dup, @@ -466,7 +466,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { chand->resolver == nullptr) { if (chand->lb_policy != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: unreffing lb_policy=%p", chand, + gpr_log(GPR_INFO, "chand=%p: unreffing lb_policy=%p", chand, chand->lb_policy.get()); } grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties(), @@ -480,11 +480,11 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { // error or shutdown. if (error != GRPC_ERROR_NONE || chand->resolver == nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: shutting down", chand); + gpr_log(GPR_INFO, "chand=%p: shutting down", chand); } if (chand->resolver != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: shutting down resolver", chand); + gpr_log(GPR_INFO, "chand=%p: shutting down resolver", chand); } chand->resolver.reset(); } @@ -506,7 +506,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { GRPC_ERROR_CREATE_FROM_STATIC_STRING("No load balancing policy"); if (lb_policy_created) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p: initializing new LB policy", chand); + gpr_log(GPR_INFO, "chand=%p: initializing new LB policy", chand); } GRPC_ERROR_UNREF(state_error); state = chand->lb_policy->CheckConnectivityLocked(&state_error); @@ -842,10 +842,11 @@ typedef struct { bool completed_recv_trailing_metadata : 1; // State for callback processing. bool retry_dispatched : 1; - bool recv_initial_metadata_ready_deferred : 1; - bool recv_message_ready_deferred : 1; + subchannel_batch_data* recv_initial_metadata_ready_deferred_batch; grpc_error* recv_initial_metadata_error; + subchannel_batch_data* recv_message_ready_deferred_batch; grpc_error* recv_message_error; + subchannel_batch_data* recv_trailing_metadata_internal_batch; } subchannel_call_retry_state; // Pending batches stored in call data. @@ -994,6 +995,39 @@ static void maybe_cache_send_ops_for_batch(call_data* calld, } } +// Frees cached send_initial_metadata. +static void free_cached_send_initial_metadata(channel_data* chand, + call_data* calld) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: destroying calld->send_initial_metadata", chand, + calld); + } + grpc_metadata_batch_destroy(&calld->send_initial_metadata); +} + +// Frees cached send_message at index idx. +static void free_cached_send_message(channel_data* chand, call_data* calld, + size_t idx) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: destroying calld->send_messages[%" PRIuPTR "]", + chand, calld, idx); + } + (*calld->send_messages)[idx]->Destroy(); +} + +// Frees cached send_trailing_metadata. +static void free_cached_send_trailing_metadata(channel_data* chand, + call_data* calld) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: destroying calld->send_trailing_metadata", + chand, calld); + } + grpc_metadata_batch_destroy(&calld->send_trailing_metadata); +} + // Frees cached send ops that have already been completed after // committing the call. static void free_cached_send_op_data_after_commit( @@ -1001,19 +1035,13 @@ static void free_cached_send_op_data_after_commit( channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (retry_state->completed_send_initial_metadata) { - grpc_metadata_batch_destroy(&calld->send_initial_metadata); + free_cached_send_initial_metadata(chand, calld); } for (size_t i = 0; i < retry_state->completed_send_message_count; ++i) { - if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, - "chand=%p calld=%p: destroying calld->send_messages[%" PRIuPTR - "]", - chand, calld, i); - } - (*calld->send_messages)[i]->Destroy(); + free_cached_send_message(chand, calld, i); } if (retry_state->completed_send_trailing_metadata) { - grpc_metadata_batch_destroy(&calld->send_trailing_metadata); + free_cached_send_trailing_metadata(chand, calld); } } @@ -1025,20 +1053,14 @@ static void free_cached_send_op_data_for_completed_batch( channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (batch_data->batch.send_initial_metadata) { - grpc_metadata_batch_destroy(&calld->send_initial_metadata); + free_cached_send_initial_metadata(chand, calld); } if (batch_data->batch.send_message) { - if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, - "chand=%p calld=%p: destroying calld->send_messages[%" PRIuPTR - "]", - chand, calld, retry_state->completed_send_message_count - 1); - } - (*calld->send_messages)[retry_state->completed_send_message_count - 1] - ->Destroy(); + free_cached_send_message(chand, calld, + retry_state->completed_send_message_count - 1); } if (batch_data->batch.send_trailing_metadata) { - grpc_metadata_batch_destroy(&calld->send_trailing_metadata); + free_cached_send_trailing_metadata(chand, calld); } } @@ -1066,7 +1088,7 @@ static void pending_batches_add(grpc_call_element* elem, call_data* calld = static_cast<call_data*>(elem->call_data); const size_t idx = get_batch_index(batch); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: adding pending batch at index %" PRIuPTR, chand, calld, idx); } @@ -1094,7 +1116,7 @@ static void pending_batches_add(grpc_call_element* elem, } if (calld->bytes_buffered_for_retry > chand->per_rpc_retry_buffer_size) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: exceeded retry buffer size, committing", chand, calld); } @@ -1109,7 +1131,7 @@ static void pending_batches_add(grpc_call_element* elem, // retries are disabled so that we don't bother with retry overhead. if (calld->num_attempts_completed == 0) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: disabling retries before first attempt", chand, calld); } @@ -1156,7 +1178,7 @@ static void pending_batches_fail(grpc_call_element* elem, grpc_error* error, for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { if (calld->pending_batches[i].batch != nullptr) ++num_batches; } - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: failing %" PRIuPTR " pending batches: %s", elem->channel_data, calld, num_batches, grpc_error_string(error)); } @@ -1218,7 +1240,7 @@ static void pending_batches_resume(grpc_call_element* elem) { for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { if (calld->pending_batches[i].batch != nullptr) ++num_batches; } - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: starting %" PRIuPTR " pending batches on subchannel_call=%p", chand, calld, num_batches, calld->subchannel_call); @@ -1263,7 +1285,7 @@ static void maybe_clear_pending_batch(grpc_call_element* elem, (!batch->recv_message || batch->payload->recv_message.recv_message_ready == nullptr)) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: clearing pending batch", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: clearing pending batch", chand, calld); } pending_batch_clear(calld, pending); @@ -1353,7 +1375,7 @@ static void retry_commit(grpc_call_element* elem, if (calld->retry_committed) return; calld->retry_committed = true; if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: committing retries", chand, calld); + gpr_log(GPR_INFO, "chand=%p calld=%p: committing retries", chand, calld); } if (retry_state != nullptr) { free_cached_send_op_data_after_commit(elem, retry_state); @@ -1398,7 +1420,7 @@ static void do_retry(grpc_call_element* elem, next_attempt_time = calld->retry_backoff->NextAttemptTime(); } if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: retrying failed call in %" PRIuPTR " ms", chand, calld, next_attempt_time - grpc_core::ExecCtx::Get()->Now()); } @@ -1432,7 +1454,7 @@ static bool maybe_retry(grpc_call_element* elem, batch_data->subchannel_call)); if (retry_state->retry_dispatched) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: retry already dispatched", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: retry already dispatched", chand, calld); } return true; @@ -1444,14 +1466,14 @@ static bool maybe_retry(grpc_call_element* elem, calld->retry_throttle_data->RecordSuccess(); } if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: call succeeded", chand, calld); + gpr_log(GPR_INFO, "chand=%p calld=%p: call succeeded", chand, calld); } return false; } // Status is not OK. Check whether the status is retryable. if (!retry_policy->retryable_status_codes.Contains(status)) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: status %s not configured as retryable", chand, calld, grpc_status_code_to_string(status)); } @@ -1467,14 +1489,14 @@ static bool maybe_retry(grpc_call_element* elem, if (calld->retry_throttle_data != nullptr && !calld->retry_throttle_data->RecordFailure()) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: retries throttled", chand, calld); + gpr_log(GPR_INFO, "chand=%p calld=%p: retries throttled", chand, calld); } return false; } // Check whether the call is committed. if (calld->retry_committed) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: retries already committed", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: retries already committed", chand, calld); } return false; @@ -1483,7 +1505,7 @@ static bool maybe_retry(grpc_call_element* elem, ++calld->num_attempts_completed; if (calld->num_attempts_completed >= retry_policy->max_attempts) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: exceeded %d retry attempts", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: exceeded %d retry attempts", chand, calld, retry_policy->max_attempts); } return false; @@ -1491,7 +1513,7 @@ static bool maybe_retry(grpc_call_element* elem, // If the call was cancelled from the surface, don't retry. if (calld->cancel_error != GRPC_ERROR_NONE) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: call cancelled from surface, not retrying", chand, calld); } @@ -1504,16 +1526,15 @@ static bool maybe_retry(grpc_call_element* elem, uint32_t ms; if (!grpc_parse_slice_to_uint32(GRPC_MDVALUE(*server_pushback_md), &ms)) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: not retrying due to server push-back", chand, calld); } return false; } else { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, - "chand=%p calld=%p: server push-back: retry in %u ms", chand, - calld, ms); + gpr_log(GPR_INFO, "chand=%p calld=%p: server push-back: retry in %u ms", + chand, calld, ms); } server_pushback_ms = (grpc_millis)ms; } @@ -1586,7 +1607,7 @@ static void invoke_recv_initial_metadata_callback(void* arg, batch->payload->recv_initial_metadata.recv_initial_metadata_ready != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: invoking recv_initial_metadata_ready for " "pending batch at index %" PRIuPTR, chand, calld, i); @@ -1622,7 +1643,7 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: got recv_initial_metadata_ready, error=%s", chand, calld, grpc_error_string(error)); } @@ -1637,12 +1658,12 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) { if ((batch_data->trailing_metadata_available || error != GRPC_ERROR_NONE) && !retry_state->completed_recv_trailing_metadata) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: deferring recv_initial_metadata_ready " "(Trailers-Only)", chand, calld); } - retry_state->recv_initial_metadata_ready_deferred = true; + retry_state->recv_initial_metadata_ready_deferred_batch = batch_data; retry_state->recv_initial_metadata_error = GRPC_ERROR_REF(error); if (!retry_state->started_recv_trailing_metadata) { // recv_trailing_metadata not yet started by application; start it @@ -1679,7 +1700,7 @@ static void invoke_recv_message_callback(void* arg, grpc_error* error) { if (batch != nullptr && batch->recv_message && batch->payload->recv_message.recv_message_ready != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: invoking recv_message_ready for " "pending batch at index %" PRIuPTR, chand, calld, i); @@ -1712,7 +1733,7 @@ static void recv_message_ready(void* arg, grpc_error* error) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: got recv_message_ready, error=%s", + gpr_log(GPR_INFO, "chand=%p calld=%p: got recv_message_ready, error=%s", chand, calld, grpc_error_string(error)); } subchannel_call_retry_state* retry_state = @@ -1726,12 +1747,12 @@ static void recv_message_ready(void* arg, grpc_error* error) { if ((batch_data->recv_message == nullptr || error != GRPC_ERROR_NONE) && !retry_state->completed_recv_trailing_metadata) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: deferring recv_message_ready (nullptr " "message and recv_trailing_metadata pending)", chand, calld); } - retry_state->recv_message_ready_deferred = true; + retry_state->recv_message_ready_deferred_batch = batch_data; retry_state->recv_message_error = GRPC_ERROR_REF(error); if (!retry_state->started_recv_trailing_metadata) { // recv_trailing_metadata not yet started by application; start it @@ -1750,6 +1771,59 @@ static void recv_message_ready(void* arg, grpc_error* error) { } // +// list of closures to execute in call combiner +// + +// Represents a closure that needs to run in the call combiner as part of +// starting or completing a batch. +typedef struct { + grpc_closure* closure; + grpc_error* error; + const char* reason; + bool free_reason = false; +} closure_to_execute; + +static void execute_closures_in_call_combiner(grpc_call_element* elem, + const char* caller, + closure_to_execute* closures, + size_t num_closures) { + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + call_data* calld = static_cast<call_data*>(elem->call_data); + // Note that the call combiner will be yielded for each closure that + // we schedule. We're already running in the call combiner, so one of + // the closures can be scheduled directly, but the others will + // have to re-enter the call combiner. + if (num_closures > 0) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, "chand=%p calld=%p: %s starting closure: %s", chand, + calld, caller, closures[0].reason); + } + GRPC_CLOSURE_SCHED(closures[0].closure, closures[0].error); + if (closures[0].free_reason) { + gpr_free(const_cast<char*>(closures[0].reason)); + } + for (size_t i = 1; i < num_closures; ++i) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: %s starting closure in call combiner: %s", + chand, calld, caller, closures[i].reason); + } + GRPC_CALL_COMBINER_START(calld->call_combiner, closures[i].closure, + closures[i].error, closures[i].reason); + if (closures[i].free_reason) { + gpr_free(const_cast<char*>(closures[i].reason)); + } + } + } else { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, "chand=%p calld=%p: no closures to run for %s", chand, + calld, caller); + } + GRPC_CALL_COMBINER_STOP(calld->call_combiner, "no closures to run"); + } +} + +// // on_complete callback handling // @@ -1777,36 +1851,35 @@ static void update_retry_state_for_completed_batch( } } -// Represents a closure that needs to run as a result of a completed batch. -typedef struct { - grpc_closure* closure; - grpc_error* error; - const char* reason; -} closure_to_execute; - // Adds any necessary closures for deferred recv_initial_metadata and // recv_message callbacks to closures, updating *num_closures as needed. static void add_closures_for_deferred_recv_callbacks( subchannel_batch_data* batch_data, subchannel_call_retry_state* retry_state, closure_to_execute* closures, size_t* num_closures) { - if (batch_data->batch.recv_trailing_metadata && - retry_state->recv_initial_metadata_ready_deferred) { - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = - GRPC_CLOSURE_INIT(&batch_data->recv_initial_metadata_ready, - invoke_recv_initial_metadata_callback, batch_data, - grpc_schedule_on_exec_ctx); - closure->error = retry_state->recv_initial_metadata_error; - closure->reason = "resuming recv_initial_metadata_ready"; - } - if (batch_data->batch.recv_trailing_metadata && - retry_state->recv_message_ready_deferred) { - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = GRPC_CLOSURE_INIT(&batch_data->recv_message_ready, - invoke_recv_message_callback, - batch_data, grpc_schedule_on_exec_ctx); - closure->error = retry_state->recv_message_error; - closure->reason = "resuming recv_message_ready"; + if (batch_data->batch.recv_trailing_metadata) { + // Add closure for deferred recv_initial_metadata_ready. + if (retry_state->recv_initial_metadata_ready_deferred_batch != nullptr) { + closure_to_execute* closure = &closures[(*num_closures)++]; + closure->closure = GRPC_CLOSURE_INIT( + &batch_data->recv_initial_metadata_ready, + invoke_recv_initial_metadata_callback, + retry_state->recv_initial_metadata_ready_deferred_batch, + grpc_schedule_on_exec_ctx); + closure->error = retry_state->recv_initial_metadata_error; + closure->reason = "resuming recv_initial_metadata_ready"; + retry_state->recv_initial_metadata_ready_deferred_batch = nullptr; + } + // Add closure for deferred recv_message_ready. + if (retry_state->recv_message_ready_deferred_batch != nullptr) { + closure_to_execute* closure = &closures[(*num_closures)++]; + closure->closure = GRPC_CLOSURE_INIT( + &batch_data->recv_message_ready, invoke_recv_message_callback, + retry_state->recv_message_ready_deferred_batch, + grpc_schedule_on_exec_ctx); + closure->error = retry_state->recv_message_error; + closure->reason = "resuming recv_message_ready"; + retry_state->recv_message_ready_deferred_batch = nullptr; + } } } @@ -1838,7 +1911,7 @@ static void add_closures_for_replay_or_pending_send_ops( } if (have_pending_send_message_ops || have_pending_send_trailing_metadata_op) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: starting next batch for pending send op(s)", chand, calld); } @@ -1863,7 +1936,7 @@ static void add_closures_for_completed_pending_batches( pending_batch* pending = &calld->pending_batches[i]; if (pending_batch_is_completed(pending, calld, retry_state)) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: pending batch completed at index %" PRIuPTR, chand, calld, i); } @@ -1896,7 +1969,7 @@ static void add_closures_to_fail_unstarted_pending_batches( pending_batch* pending = &calld->pending_batches[i]; if (pending_batch_is_unstarted(pending, calld, retry_state)) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: failing unstarted pending batch at index " "%" PRIuPTR, chand, calld, i); @@ -1940,7 +2013,7 @@ static void on_complete(void* arg, grpc_error* error) { call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { char* batch_str = grpc_transport_stream_op_batch_string(&batch_data->batch); - gpr_log(GPR_DEBUG, "chand=%p calld=%p: got on_complete, error=%s, batch=%s", + gpr_log(GPR_INFO, "chand=%p calld=%p: got on_complete, error=%s, batch=%s", chand, calld, grpc_error_string(error), batch_str); gpr_free(batch_str); } @@ -1951,11 +2024,13 @@ static void on_complete(void* arg, grpc_error* error) { // If we have previously completed recv_trailing_metadata, then the // call is finished. bool call_finished = retry_state->completed_recv_trailing_metadata; + // Record whether we were already committed before receiving this callback. + const bool previously_committed = calld->retry_committed; // Update bookkeeping in retry_state. update_retry_state_for_completed_batch(batch_data, retry_state); if (call_finished) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: call already finished", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: call already finished", chand, calld); } } else { @@ -1979,35 +2054,39 @@ static void on_complete(void* arg, grpc_error* error) { if (md_batch->idx.named.grpc_retry_pushback_ms != nullptr) { server_pushback_md = &md_batch->idx.named.grpc_retry_pushback_ms->md; } - } else if (retry_state->completed_recv_trailing_metadata) { - call_finished = true; } - if (call_finished && grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: call finished, status=%s", chand, - calld, grpc_status_code_to_string(status)); - } - // If the call is finished, check if we should retry. - if (call_finished && - maybe_retry(elem, batch_data, status, server_pushback_md)) { - // Unref batch_data for deferred recv_initial_metadata_ready or - // recv_message_ready callbacks, if any. - if (batch_data->batch.recv_trailing_metadata && - retry_state->recv_initial_metadata_ready_deferred) { - batch_data_unref(batch_data); - GRPC_ERROR_UNREF(retry_state->recv_initial_metadata_error); + // If the call just finished, check if we should retry. + if (call_finished) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, "chand=%p calld=%p: call finished, status=%s", chand, + calld, grpc_status_code_to_string(status)); } - if (batch_data->batch.recv_trailing_metadata && - retry_state->recv_message_ready_deferred) { + if (maybe_retry(elem, batch_data, status, server_pushback_md)) { + // Unref batch_data for deferred recv_initial_metadata_ready or + // recv_message_ready callbacks, if any. + if (batch_data->batch.recv_trailing_metadata && + retry_state->recv_initial_metadata_ready_deferred_batch != + nullptr) { + batch_data_unref(batch_data); + GRPC_ERROR_UNREF(retry_state->recv_initial_metadata_error); + } + if (batch_data->batch.recv_trailing_metadata && + retry_state->recv_message_ready_deferred_batch != nullptr) { + batch_data_unref(batch_data); + GRPC_ERROR_UNREF(retry_state->recv_message_error); + } batch_data_unref(batch_data); - GRPC_ERROR_UNREF(retry_state->recv_message_error); + return; } - batch_data_unref(batch_data); - return; + // Not retrying, so commit the call. + retry_commit(elem, retry_state); } } - // If the call is finished or retries are committed, free cached data for - // send ops that we've just completed. - if (call_finished || calld->retry_committed) { + // If we were already committed before receiving this callback, free + // cached data for send ops that we've just completed. (If the call has + // just now finished, the call to retry_commit() above will have freed all + // cached send ops, so we don't need to do it here.) + if (previously_committed) { free_cached_send_op_data_for_completed_batch(elem, batch_data, retry_state); } // Call not being retried. @@ -2042,20 +2121,8 @@ static void on_complete(void* arg, grpc_error* error) { // Don't need batch_data anymore. batch_data_unref(batch_data); // Schedule all of the closures identified above. - // Note that the call combiner will be yielded for each closure that - // we schedule. We're already running in the call combiner, so one of - // the closures can be scheduled directly, but the others will - // have to re-enter the call combiner. - if (num_closures > 0) { - GRPC_CLOSURE_SCHED(closures[0].closure, closures[0].error); - for (size_t i = 1; i < num_closures; ++i) { - GRPC_CALL_COMBINER_START(calld->call_combiner, closures[i].closure, - closures[i].error, closures[i].reason); - } - } else { - GRPC_CALL_COMBINER_STOP(calld->call_combiner, - "no closures to run for on_complete"); - } + execute_closures_in_call_combiner(elem, "on_complete", closures, + num_closures); } // @@ -2072,6 +2139,31 @@ static void start_batch_in_call_combiner(void* arg, grpc_error* ignored) { grpc_subchannel_call_process_op(subchannel_call, batch); } +// Adds a closure to closures that will execute batch in the call combiner. +static void add_closure_for_subchannel_batch( + call_data* calld, grpc_transport_stream_op_batch* batch, + closure_to_execute* closures, size_t* num_closures) { + batch->handler_private.extra_arg = calld->subchannel_call; + GRPC_CLOSURE_INIT(&batch->handler_private.closure, + start_batch_in_call_combiner, batch, + grpc_schedule_on_exec_ctx); + closure_to_execute* closure = &closures[(*num_closures)++]; + closure->closure = &batch->handler_private.closure; + closure->error = GRPC_ERROR_NONE; + // If the tracer is enabled, we log a more detailed message, which + // requires dynamic allocation. This will be freed in + // start_retriable_subchannel_batches(). + if (grpc_client_channel_trace.enabled()) { + char* batch_str = grpc_transport_stream_op_batch_string(batch); + gpr_asprintf(const_cast<char**>(&closure->reason), + "starting batch in call combiner: %s", batch_str); + gpr_free(batch_str); + closure->free_reason = true; + } else { + closure->reason = "start_subchannel_batch"; + } +} + // Adds retriable send_initial_metadata op to batch_data. static void add_retriable_send_initial_metadata_op( call_data* calld, subchannel_call_retry_state* retry_state, @@ -2131,7 +2223,7 @@ static void add_retriable_send_message_op( channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: starting calld->send_messages[%" PRIuPTR "]", chand, calld, retry_state->started_send_message_count); } @@ -2218,7 +2310,7 @@ static void start_internal_recv_trailing_metadata(grpc_call_element* elem) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: call failed but recv_trailing_metadata not " "started; starting it internally", chand, calld); @@ -2227,8 +2319,12 @@ static void start_internal_recv_trailing_metadata(grpc_call_element* elem) { static_cast<subchannel_call_retry_state*>( grpc_connected_subchannel_call_get_parent_data( calld->subchannel_call)); - subchannel_batch_data* batch_data = batch_data_create(elem, 1); + // Create batch_data with 2 refs, since this batch will be unreffed twice: + // once when the subchannel batch returns, and again when we actually get + // a recv_trailing_metadata op from the surface. + subchannel_batch_data* batch_data = batch_data_create(elem, 2); add_retriable_recv_trailing_metadata_op(calld, retry_state, batch_data); + retry_state->recv_trailing_metadata_internal_batch = batch_data; // Note: This will release the call combiner. grpc_subchannel_call_process_op(calld->subchannel_call, &batch_data->batch); } @@ -2246,7 +2342,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( !retry_state->started_send_initial_metadata && !calld->pending_send_initial_metadata) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: replaying previously completed " "send_initial_metadata op", chand, calld); @@ -2262,7 +2358,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( retry_state->completed_send_message_count && !calld->pending_send_message) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: replaying previously completed " "send_message op", chand, calld); @@ -2281,7 +2377,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( !retry_state->started_send_trailing_metadata && !calld->pending_send_trailing_metadata) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: replaying previously completed " "send_trailing_metadata op", chand, calld); @@ -2299,7 +2395,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( // *num_batches as needed. static void add_subchannel_batches_for_pending_batches( grpc_call_element* elem, subchannel_call_retry_state* retry_state, - grpc_transport_stream_op_batch** batches, size_t* num_batches) { + closure_to_execute* closures, size_t* num_closures) { call_data* calld = static_cast<call_data*>(elem->call_data); for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { pending_batch* pending = &calld->pending_batches[i]; @@ -2342,13 +2438,37 @@ static void add_subchannel_batches_for_pending_batches( } if (batch->recv_trailing_metadata && retry_state->started_recv_trailing_metadata) { + // If we previously completed a recv_trailing_metadata op + // initiated by start_internal_recv_trailing_metadata(), use the + // result of that instead of trying to re-start this op. + if (retry_state->recv_trailing_metadata_internal_batch != nullptr) { + // If the batch completed, then trigger the completion callback + // directly, so that we return the previously returned results to + // the application. Otherwise, just unref the internally + // started subchannel batch, since we'll propagate the + // completion when it completes. + if (retry_state->completed_recv_trailing_metadata) { + subchannel_batch_data* batch_data = + retry_state->recv_trailing_metadata_internal_batch; + closure_to_execute* closure = &closures[(*num_closures)++]; + closure->closure = &batch_data->on_complete; + // Batches containing recv_trailing_metadata always succeed. + closure->error = GRPC_ERROR_NONE; + closure->reason = + "re-executing on_complete for recv_trailing_metadata " + "to propagate internally triggered result"; + } else { + batch_data_unref(retry_state->recv_trailing_metadata_internal_batch); + } + retry_state->recv_trailing_metadata_internal_batch = nullptr; + } continue; } // If we're not retrying, just send the batch as-is. if (calld->method_params == nullptr || calld->method_params->retry_policy() == nullptr || calld->retry_committed) { - batches[(*num_batches)++] = batch; + add_closure_for_subchannel_batch(calld, batch, closures, num_closures); pending_batch_clear(calld, pending); continue; } @@ -2385,7 +2505,8 @@ static void add_subchannel_batches_for_pending_batches( GPR_ASSERT(batch->collect_stats); add_retriable_recv_trailing_metadata_op(calld, retry_state, batch_data); } - batches[(*num_batches)++] = &batch_data->batch; + add_closure_for_subchannel_batch(calld, &batch_data->batch, closures, + num_closures); } } @@ -2396,69 +2517,36 @@ static void start_retriable_subchannel_batches(void* arg, grpc_error* ignored) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: constructing retriable batches", + gpr_log(GPR_INFO, "chand=%p calld=%p: constructing retriable batches", chand, calld); } subchannel_call_retry_state* retry_state = static_cast<subchannel_call_retry_state*>( grpc_connected_subchannel_call_get_parent_data( calld->subchannel_call)); + // Construct list of closures to execute, one for each pending batch. // We can start up to 6 batches. - grpc_transport_stream_op_batch* - batches[GPR_ARRAY_SIZE(calld->pending_batches)]; - size_t num_batches = 0; + closure_to_execute closures[GPR_ARRAY_SIZE(calld->pending_batches)]; + size_t num_closures = 0; // Replay previously-returned send_* ops if needed. subchannel_batch_data* replay_batch_data = maybe_create_subchannel_batch_for_replay(elem, retry_state); if (replay_batch_data != nullptr) { - batches[num_batches++] = &replay_batch_data->batch; + add_closure_for_subchannel_batch(calld, &replay_batch_data->batch, closures, + &num_closures); } // Now add pending batches. - add_subchannel_batches_for_pending_batches(elem, retry_state, batches, - &num_batches); + add_subchannel_batches_for_pending_batches(elem, retry_state, closures, + &num_closures); // Start batches on subchannel call. - // Note that the call combiner will be yielded for each batch that we - // send down. We're already running in the call combiner, so one of - // the batches can be started directly, but the others will have to - // re-enter the call combiner. if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: starting %" PRIuPTR " retriable batches on subchannel_call=%p", - chand, calld, num_batches, calld->subchannel_call); - } - if (num_batches == 0) { - // This should be fairly rare, but it can happen when (e.g.) an - // attempt completes before it has finished replaying all - // previously sent messages. - GRPC_CALL_COMBINER_STOP(calld->call_combiner, - "no retriable subchannel batches to start"); - } else { - for (size_t i = 1; i < num_batches; ++i) { - if (grpc_client_channel_trace.enabled()) { - char* batch_str = grpc_transport_stream_op_batch_string(batches[i]); - gpr_log(GPR_DEBUG, - "chand=%p calld=%p: starting batch in call combiner: %s", chand, - calld, batch_str); - gpr_free(batch_str); - } - batches[i]->handler_private.extra_arg = calld->subchannel_call; - GRPC_CLOSURE_INIT(&batches[i]->handler_private.closure, - start_batch_in_call_combiner, batches[i], - grpc_schedule_on_exec_ctx); - GRPC_CALL_COMBINER_START(calld->call_combiner, - &batches[i]->handler_private.closure, - GRPC_ERROR_NONE, "start_subchannel_batch"); - } - if (grpc_client_channel_trace.enabled()) { - char* batch_str = grpc_transport_stream_op_batch_string(batches[0]); - gpr_log(GPR_DEBUG, "chand=%p calld=%p: starting batch: %s", chand, calld, - batch_str); - gpr_free(batch_str); - } - // Note: This will release the call combiner. - grpc_subchannel_call_process_op(calld->subchannel_call, batches[0]); + chand, calld, num_closures, calld->subchannel_call); } + execute_closures_in_call_combiner(elem, "start_retriable_subchannel_batches", + closures, num_closures); } // @@ -2483,7 +2571,7 @@ static void create_subchannel_call(grpc_call_element* elem, grpc_error* error) { grpc_error* new_error = calld->pick.connected_subchannel->CreateCall( call_args, &calld->subchannel_call); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: create subchannel_call=%p: error=%s", + gpr_log(GPR_INFO, "chand=%p calld=%p: create subchannel_call=%p: error=%s", chand, calld, calld->subchannel_call, grpc_error_string(new_error)); } if (new_error != GRPC_ERROR_NONE) { @@ -2524,7 +2612,7 @@ static void pick_done(void* arg, grpc_error* error) { : GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Failed to create subchannel", &error, 1); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: failed to create subchannel: error=%s", chand, calld, grpc_error_string(new_error)); } @@ -2568,7 +2656,7 @@ static void pick_callback_cancel_locked(void* arg, grpc_error* error) { // the one we started it on. However, this will just be a no-op. if (error != GRPC_ERROR_NONE && chand->lb_policy != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p", + gpr_log(GPR_INFO, "chand=%p calld=%p: cancelling pick from LB policy %p", chand, calld, chand->lb_policy.get()); } chand->lb_policy->CancelPickLocked(&calld->pick, GRPC_ERROR_REF(error)); @@ -2583,8 +2671,8 @@ static void pick_callback_done_locked(void* arg, grpc_error* error) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", - chand, calld); + gpr_log(GPR_INFO, "chand=%p calld=%p: pick completed asynchronously", chand, + calld); } async_pick_done_locked(elem, GRPC_ERROR_REF(error)); GRPC_CALL_STACK_UNREF(calld->owning_call, "pick_callback"); @@ -2596,7 +2684,7 @@ static void apply_service_config_to_call_locked(grpc_call_element* elem) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: applying service config to call", + gpr_log(GPR_INFO, "chand=%p calld=%p: applying service config to call", chand, calld); } if (chand->retry_throttle_data != nullptr) { @@ -2634,8 +2722,8 @@ static bool pick_callback_start_locked(grpc_call_element* elem) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: starting pick on lb_policy=%p", - chand, calld, chand->lb_policy.get()); + gpr_log(GPR_INFO, "chand=%p calld=%p: starting pick on lb_policy=%p", chand, + calld, chand->lb_policy.get()); } // Only get service config data on the first attempt. if (calld->num_attempts_completed == 0) { @@ -2682,7 +2770,7 @@ static bool pick_callback_start_locked(grpc_call_element* elem) { if (pick_done) { // Pick completed synchronously. if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed synchronously", + gpr_log(GPR_INFO, "chand=%p calld=%p: pick completed synchronously", chand, calld); } GRPC_CALL_STACK_UNREF(calld->owning_call, "pick_callback"); @@ -2726,7 +2814,7 @@ static void pick_after_resolver_result_cancel_locked(void* arg, channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: cancelling pick waiting for resolver result", chand, calld); } @@ -2746,7 +2834,7 @@ static void pick_after_resolver_result_done_locked(void* arg, if (args->finished) { /* cancelled, do nothing */ if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "call cancelled before resolver result"); + gpr_log(GPR_INFO, "call cancelled before resolver result"); } gpr_free(args); return; @@ -2757,14 +2845,14 @@ static void pick_after_resolver_result_done_locked(void* arg, call_data* calld = static_cast<call_data*>(elem->call_data); if (error != GRPC_ERROR_NONE) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver failed to return data", + gpr_log(GPR_INFO, "chand=%p calld=%p: resolver failed to return data", chand, calld); } async_pick_done_locked(elem, GRPC_ERROR_REF(error)); } else if (chand->resolver == nullptr) { // Shutting down. if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver disconnected", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: resolver disconnected", chand, calld); } async_pick_done_locked( @@ -2780,7 +2868,7 @@ static void pick_after_resolver_result_done_locked(void* arg, .send_initial_metadata_flags; if (send_initial_metadata_flags & GRPC_INITIAL_METADATA_WAIT_FOR_READY) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: resolver returned but no LB policy; " "wait_for_ready=true; trying again", chand, calld); @@ -2788,7 +2876,7 @@ static void pick_after_resolver_result_done_locked(void* arg, pick_after_resolver_result_start_locked(elem); } else { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: resolver returned but no LB policy; " "wait_for_ready=false; failing", chand, calld); @@ -2801,7 +2889,7 @@ static void pick_after_resolver_result_done_locked(void* arg, } } else { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver returned, doing pick", + gpr_log(GPR_INFO, "chand=%p calld=%p: resolver returned, doing pick", chand, calld); } if (pick_callback_start_locked(elem)) { @@ -2819,7 +2907,7 @@ static void pick_after_resolver_result_start_locked(grpc_call_element* elem) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: deferring pick pending resolver result", chand, calld); } @@ -2886,7 +2974,7 @@ static void cc_start_transport_stream_op_batch( // If we've previously been cancelled, immediately fail any new batches. if (calld->cancel_error != GRPC_ERROR_NONE) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: failing batch with error: %s", + gpr_log(GPR_INFO, "chand=%p calld=%p: failing batch with error: %s", chand, calld, grpc_error_string(calld->cancel_error)); } // Note: This will release the call combiner. @@ -2905,7 +2993,7 @@ static void cc_start_transport_stream_op_batch( calld->cancel_error = GRPC_ERROR_REF(batch->payload->cancel_stream.cancel_error); if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: recording cancel_error=%s", chand, + gpr_log(GPR_INFO, "chand=%p calld=%p: recording cancel_error=%s", chand, calld, grpc_error_string(calld->cancel_error)); } // If we do not have a subchannel call (i.e., a pick has not yet @@ -2931,7 +3019,7 @@ static void cc_start_transport_stream_op_batch( // streaming calls). if (calld->subchannel_call != nullptr) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: starting batch on subchannel_call=%p", chand, calld, calld->subchannel_call); } @@ -2943,7 +3031,7 @@ static void cc_start_transport_stream_op_batch( // combiner to start a pick. if (batch->send_initial_metadata) { if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: entering client_channel combiner", + gpr_log(GPR_INFO, "chand=%p calld=%p: entering client_channel combiner", chand, calld); } GRPC_CLOSURE_SCHED( @@ -2953,7 +3041,7 @@ static void cc_start_transport_stream_op_batch( } else { // For all other batches, release the call combiner. if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "chand=%p calld=%p: saved batch, yeilding call combiner", chand, calld); } diff --git a/src/core/ext/filters/client_channel/http_connect_handshaker.cc b/src/core/ext/filters/client_channel/http_connect_handshaker.cc index fb29fa788d..4e8b8b71db 100644 --- a/src/core/ext/filters/client_channel/http_connect_handshaker.cc +++ b/src/core/ext/filters/client_channel/http_connect_handshaker.cc @@ -326,7 +326,7 @@ static void http_connect_handshaker_do_handshake( static const grpc_handshaker_vtable http_connect_handshaker_vtable = { http_connect_handshaker_destroy, http_connect_handshaker_shutdown, - http_connect_handshaker_do_handshake}; + http_connect_handshaker_do_handshake, "http_connect"}; static grpc_handshaker* grpc_http_connect_handshaker_create() { http_connect_handshaker* handshaker = diff --git a/src/core/ext/filters/client_channel/lb_policy.cc b/src/core/ext/filters/client_channel/lb_policy.cc index fa63dd75b5..e065f45639 100644 --- a/src/core/ext/filters/client_channel/lb_policy.cc +++ b/src/core/ext/filters/client_channel/lb_policy.cc @@ -44,13 +44,13 @@ void LoadBalancingPolicy::TryReresolutionLocked( GRPC_CLOSURE_SCHED(request_reresolution_, error); request_reresolution_ = nullptr; if (grpc_lb_trace->enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "%s %p: scheduling re-resolution closure with error=%s.", grpc_lb_trace->name(), this, grpc_error_string(error)); } } else { if (grpc_lb_trace->enabled()) { - gpr_log(GPR_DEBUG, "%s %p: no available re-resolution closure.", + gpr_log(GPR_INFO, "%s %p: no available re-resolution closure.", grpc_lb_trace->name(), this); } } diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h index c3e43e5ef6..454e00a690 100644 --- a/src/core/ext/filters/client_channel/lb_policy.h +++ b/src/core/ext/filters/client_channel/lb_policy.h @@ -162,6 +162,10 @@ class LoadBalancingPolicy GRPC_ABSTRACT_BASE_CLASS protected: + // So Delete() can access our protected dtor. + template <typename T> + friend void Delete(T*); + explicit LoadBalancingPolicy(const Args& args); virtual ~LoadBalancingPolicy(); diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index 0b2a30818e..1a675476f0 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -189,6 +189,10 @@ class GrpcLb : public LoadBalancingPolicy { bool seen_initial_response() const { return seen_initial_response_; } private: + // So Delete() can access our private dtor. + template <typename T> + friend void grpc_core::Delete(T*); + ~BalancerCallState(); GrpcLb* grpclb_policy() const { @@ -1243,7 +1247,7 @@ bool GrpcLb::PickLocked(PickState* pick) { } } else { // rr_policy_ == NULL if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[grpclb %p] No RR policy. Adding to grpclb's pending picks", this); } @@ -1409,14 +1413,13 @@ void GrpcLb::OnFallbackTimerLocked(void* arg, grpc_error* error) { void GrpcLb::StartBalancerCallRetryTimerLocked() { grpc_millis next_try = lb_call_backoff_.NextAttemptTime(); if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_DEBUG, "[grpclb %p] Connection to LB server lost...", this); + gpr_log(GPR_INFO, "[grpclb %p] Connection to LB server lost...", this); grpc_millis timeout = next_try - ExecCtx::Get()->Now(); if (timeout > 0) { - gpr_log(GPR_DEBUG, - "[grpclb %p] ... retry_timer_active in %" PRIuPTR "ms.", this, - timeout); + gpr_log(GPR_INFO, "[grpclb %p] ... retry_timer_active in %" PRIuPTR "ms.", + this, timeout); } else { - gpr_log(GPR_DEBUG, "[grpclb %p] ... retry_timer_active immediately.", + gpr_log(GPR_INFO, "[grpclb %p] ... retry_timer_active immediately.", this); } } @@ -1724,7 +1727,7 @@ void GrpcLb::CreateOrUpdateRoundRobinPolicyLocked() { GPR_ASSERT(args != nullptr); if (rr_policy_ != nullptr) { if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_DEBUG, "[grpclb %p] Updating RR policy %p", this, + gpr_log(GPR_INFO, "[grpclb %p] Updating RR policy %p", this, rr_policy_.get()); } rr_policy_->UpdateLocked(*args); @@ -1735,7 +1738,7 @@ void GrpcLb::CreateOrUpdateRoundRobinPolicyLocked() { lb_policy_args.args = args; CreateRoundRobinPolicyLocked(lb_policy_args); if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_DEBUG, "[grpclb %p] Created new RR policy %p", this, + gpr_log(GPR_INFO, "[grpclb %p] Created new RR policy %p", this, rr_policy_.get()); } } @@ -1751,7 +1754,7 @@ void GrpcLb::OnRoundRobinRequestReresolutionLocked(void* arg, } if (grpc_lb_glb_trace.enabled()) { gpr_log( - GPR_DEBUG, + GPR_INFO, "[grpclb %p] Re-resolution requested from the internal RR policy (%p).", grpclb_policy, grpclb_policy->rr_policy_.get()); } diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index 9090c34412..a07e90322c 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -95,7 +95,7 @@ PickFirst::PickFirst(const Args& args) : LoadBalancingPolicy(args) { grpc_connectivity_state_init(&state_tracker_, GRPC_CHANNEL_IDLE, "pick_first"); if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_DEBUG, "Pick First %p created.", this); + gpr_log(GPR_INFO, "Pick First %p created.", this); } UpdateLocked(*args.args); grpc_subchannel_index_ref(); @@ -103,7 +103,7 @@ PickFirst::PickFirst(const Args& args) : LoadBalancingPolicy(args) { PickFirst::~PickFirst() { if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_DEBUG, "Destroying Pick First %p", this); + gpr_log(GPR_INFO, "Destroying Pick First %p", this); } GPR_ASSERT(subchannel_list_ == nullptr); GPR_ASSERT(latest_pending_subchannel_list_ == nullptr); @@ -126,7 +126,7 @@ void PickFirst::HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) { void PickFirst::ShutdownLocked() { grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_DEBUG, "Pick First %p Shutting down", this); + gpr_log(GPR_INFO, "Pick First %p Shutting down", this); } shutdown_ = true; PickState* pick; @@ -372,7 +372,7 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { // subchannel list. if (latest_pending_subchannel_list_ != nullptr) { if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "Pick First %p Shutting down latest pending subchannel list " "%p, about to be replaced by newer latest %p", this, latest_pending_subchannel_list_, subchannel_list); @@ -396,7 +396,7 @@ void PickFirst::OnConnectivityChangedLocked(void* arg, grpc_error* error) { grpc_lb_subchannel_data* sd = static_cast<grpc_lb_subchannel_data*>(arg); PickFirst* p = static_cast<PickFirst*>(sd->subchannel_list->policy); if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "Pick First %p connectivity changed for subchannel %p (%" PRIuPTR " of %" PRIuPTR "), subchannel_list %p: state=%s p->shutdown_=%d " diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc index e534131c02..b470016bd9 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc @@ -114,7 +114,7 @@ RoundRobin::RoundRobin(const Args& args) : LoadBalancingPolicy(args) { "round_robin"); UpdateLocked(*args.args); if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] Created with %" PRIuPTR " subchannels", this, + gpr_log(GPR_INFO, "[RR %p] Created with %" PRIuPTR " subchannels", this, subchannel_list_->num_subchannels); } grpc_subchannel_index_ref(); @@ -122,7 +122,7 @@ RoundRobin::RoundRobin(const Args& args) : LoadBalancingPolicy(args) { RoundRobin::~RoundRobin() { if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] Destroying Round Robin policy", this); + gpr_log(GPR_INFO, "[RR %p] Destroying Round Robin policy", this); } GPR_ASSERT(subchannel_list_ == nullptr); GPR_ASSERT(latest_pending_subchannel_list_ == nullptr); @@ -152,7 +152,7 @@ size_t RoundRobin::GetNextReadySubchannelIndexLocked() { subchannel_list_->num_subchannels; if (grpc_lb_round_robin_trace.enabled()) { gpr_log( - GPR_DEBUG, + GPR_INFO, "[RR %p] checking subchannel %p, subchannel_list %p, index %" PRIuPTR ": state=%s", this, subchannel_list_->subchannels[index].subchannel, @@ -163,7 +163,7 @@ size_t RoundRobin::GetNextReadySubchannelIndexLocked() { if (subchannel_list_->subchannels[index].curr_connectivity_state == GRPC_CHANNEL_READY) { if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] found next ready subchannel (%p) at index %" PRIuPTR " of subchannel_list %p", this, subchannel_list_->subchannels[index].subchannel, index, @@ -173,7 +173,7 @@ size_t RoundRobin::GetNextReadySubchannelIndexLocked() { } } if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] no subchannels in ready state", this); + gpr_log(GPR_INFO, "[RR %p] no subchannels in ready state", this); } return subchannel_list_->num_subchannels; } @@ -183,7 +183,7 @@ void RoundRobin::UpdateLastReadySubchannelIndexLocked(size_t last_ready_index) { GPR_ASSERT(last_ready_index < subchannel_list_->num_subchannels); last_ready_subchannel_index_ = last_ready_index; if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] setting last_ready_subchannel_index=%" PRIuPTR " (SC %p, CSC %p)", this, last_ready_index, @@ -207,7 +207,7 @@ void RoundRobin::HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) { void RoundRobin::ShutdownLocked() { grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] Shutting down", this); + gpr_log(GPR_INFO, "[RR %p] Shutting down", this); } shutdown_ = true; PickState* pick; @@ -311,8 +311,7 @@ void RoundRobin::ExitIdleLocked() { bool RoundRobin::PickLocked(PickState* pick) { if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] Trying to pick (shutdown: %d)", this, - shutdown_); + gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", this, shutdown_); } GPR_ASSERT(!shutdown_); if (subchannel_list_ != nullptr) { @@ -327,7 +326,7 @@ bool RoundRobin::PickLocked(PickState* pick) { } if (grpc_lb_round_robin_trace.enabled()) { gpr_log( - GPR_DEBUG, + GPR_INFO, "[RR %p] Picked target <-- Subchannel %p (connected %p) (sl %p, " "index %" PRIuPTR ")", this, sd->subchannel, pick->connected_subchannel.get(), @@ -416,7 +415,7 @@ void RoundRobin::OnConnectivityChangedLocked(void* arg, grpc_error* error) { RoundRobin* p = static_cast<RoundRobin*>(sd->subchannel_list->policy); if (grpc_lb_round_robin_trace.enabled()) { gpr_log( - GPR_DEBUG, + GPR_INFO, "[RR %p] connectivity changed for subchannel %p, subchannel_list %p: " "prev_state=%s new_state=%s p->shutdown=%d " "sd->subchannel_list->shutting_down=%d error=%s", @@ -458,7 +457,7 @@ void RoundRobin::OnConnectivityChangedLocked(void* arg, grpc_error* error) { case GRPC_CHANNEL_TRANSIENT_FAILURE: { sd->connected_subchannel.reset(); if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] Subchannel %p has gone into TRANSIENT_FAILURE. " "Requesting re-resolution", p, sd->subchannel); @@ -483,7 +482,7 @@ void RoundRobin::OnConnectivityChangedLocked(void* arg, grpc_error* error) { p->subchannel_list_ != nullptr ? p->subchannel_list_->num_subchannels : 0; - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] phasing out subchannel list %p (size %" PRIuPTR ") in favor of %p (size %" PRIuPTR ")", p, p->subchannel_list_, num_subchannels, sd->subchannel_list, @@ -517,7 +516,7 @@ void RoundRobin::OnConnectivityChangedLocked(void* arg, grpc_error* error) { *pick->user_data = selected->user_data; } if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] Fulfilling pending pick. Target <-- subchannel %p " "(subchannel_list %p, index %" PRIuPTR ")", p, selected->subchannel, p->subchannel_list_, @@ -584,7 +583,7 @@ void RoundRobin::UpdateLocked(const grpc_channel_args& args) { } grpc_lb_addresses* addresses = (grpc_lb_addresses*)arg->value.pointer.p; if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, "[RR %p] received update with %" PRIuPTR " addresses", + gpr_log(GPR_INFO, "[RR %p] received update with %" PRIuPTR " addresses", this, addresses->num_addresses); } grpc_lb_subchannel_list* subchannel_list = grpc_lb_subchannel_list_create( @@ -629,7 +628,7 @@ void RoundRobin::UpdateLocked(const grpc_channel_args& args) { } if (latest_pending_subchannel_list_ != nullptr) { if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[RR %p] Shutting down latest pending subchannel list %p, " "about to be replaced by newer latest %p", this, latest_pending_subchannel_list_, subchannel_list); diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc index 79cb64c6c6..257db57575 100644 --- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc +++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc @@ -34,7 +34,7 @@ void grpc_lb_subchannel_data_unref_subchannel(grpc_lb_subchannel_data* sd, const char* reason) { if (sd->subchannel != nullptr) { if (sd->subchannel_list->tracer->enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR " (subchannel %p): unreffing subchannel", sd->subchannel_list->tracer->name(), sd->subchannel_list->policy, @@ -57,7 +57,7 @@ void grpc_lb_subchannel_data_start_connectivity_watch( grpc_lb_subchannel_data* sd) { if (sd->subchannel_list->tracer->enabled()) { gpr_log( - GPR_DEBUG, + GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR " (subchannel %p): requesting connectivity change " "notification (from %s)", @@ -77,7 +77,7 @@ void grpc_lb_subchannel_data_start_connectivity_watch( void grpc_lb_subchannel_data_stop_connectivity_watch( grpc_lb_subchannel_data* sd) { if (sd->subchannel_list->tracer->enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR " (subchannel %p): stopping connectivity watch", sd->subchannel_list->tracer->name(), sd->subchannel_list->policy, @@ -98,7 +98,7 @@ grpc_lb_subchannel_list* grpc_lb_subchannel_list_create( static_cast<grpc_lb_subchannel_list*>( gpr_zalloc(sizeof(*subchannel_list))); if (tracer->enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] Creating subchannel list %p for %" PRIuPTR " subchannels", tracer->name(), p, subchannel_list, addresses->num_addresses); } @@ -132,7 +132,7 @@ grpc_lb_subchannel_list* grpc_lb_subchannel_list_create( if (tracer->enabled()) { char* address_uri = grpc_sockaddr_to_uri(&addresses->addresses[i].address); - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] could not create subchannel for address uri %s, " "ignoring", tracer->name(), subchannel_list->policy, address_uri); @@ -143,7 +143,7 @@ grpc_lb_subchannel_list* grpc_lb_subchannel_list_create( if (tracer->enabled()) { char* address_uri = grpc_sockaddr_to_uri(&addresses->addresses[i].address); - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR ": Created subchannel %p for address uri %s", tracer->name(), p, subchannel_list, subchannel_index, subchannel, @@ -175,7 +175,7 @@ grpc_lb_subchannel_list* grpc_lb_subchannel_list_create( static void subchannel_list_destroy(grpc_lb_subchannel_list* subchannel_list) { if (subchannel_list->tracer->enabled()) { - gpr_log(GPR_DEBUG, "[%s %p] Destroying subchannel_list %p", + gpr_log(GPR_INFO, "[%s %p] Destroying subchannel_list %p", subchannel_list->tracer->name(), subchannel_list->policy, subchannel_list); } @@ -192,7 +192,7 @@ void grpc_lb_subchannel_list_ref(grpc_lb_subchannel_list* subchannel_list, gpr_ref_non_zero(&subchannel_list->refcount); if (subchannel_list->tracer->enabled()) { const gpr_atm count = gpr_atm_acq_load(&subchannel_list->refcount.count); - gpr_log(GPR_DEBUG, "[%s %p] subchannel_list %p REF %lu->%lu (%s)", + gpr_log(GPR_INFO, "[%s %p] subchannel_list %p REF %lu->%lu (%s)", subchannel_list->tracer->name(), subchannel_list->policy, subchannel_list, static_cast<unsigned long>(count - 1), static_cast<unsigned long>(count), reason); @@ -204,7 +204,7 @@ void grpc_lb_subchannel_list_unref(grpc_lb_subchannel_list* subchannel_list, const bool done = gpr_unref(&subchannel_list->refcount); if (subchannel_list->tracer->enabled()) { const gpr_atm count = gpr_atm_acq_load(&subchannel_list->refcount.count); - gpr_log(GPR_DEBUG, "[%s %p] subchannel_list %p UNREF %lu->%lu (%s)", + gpr_log(GPR_INFO, "[%s %p] subchannel_list %p UNREF %lu->%lu (%s)", subchannel_list->tracer->name(), subchannel_list->policy, subchannel_list, static_cast<unsigned long>(count + 1), static_cast<unsigned long>(count), reason); @@ -217,7 +217,7 @@ void grpc_lb_subchannel_list_unref(grpc_lb_subchannel_list* subchannel_list, static void subchannel_data_cancel_connectivity_watch( grpc_lb_subchannel_data* sd, const char* reason) { if (sd->subchannel_list->tracer->enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR " (subchannel %p): canceling connectivity watch (%s)", sd->subchannel_list->tracer->name(), sd->subchannel_list->policy, @@ -232,7 +232,7 @@ static void subchannel_data_cancel_connectivity_watch( void grpc_lb_subchannel_list_shutdown_and_unref( grpc_lb_subchannel_list* subchannel_list, const char* reason) { if (subchannel_list->tracer->enabled()) { - gpr_log(GPR_DEBUG, "[%s %p] Shutting down subchannel_list %p (%s)", + gpr_log(GPR_INFO, "[%s %p] Shutting down subchannel_list %p (%s)", subchannel_list->tracer->name(), subchannel_list->policy, subchannel_list, reason); } diff --git a/src/core/ext/filters/client_channel/method_params.h b/src/core/ext/filters/client_channel/method_params.h index 099924edf3..a31d360f17 100644 --- a/src/core/ext/filters/client_channel/method_params.h +++ b/src/core/ext/filters/client_channel/method_params.h @@ -60,6 +60,10 @@ class ClientChannelMethodParams : public RefCounted<ClientChannelMethodParams> { template <typename T, typename... Args> friend T* grpc_core::New(Args&&... args); + // So Delete() can call our private dtor. + template <typename T> + friend void grpc_core::Delete(T*); + ClientChannelMethodParams() {} virtual ~ClientChannelMethodParams() {} diff --git a/src/core/ext/filters/client_channel/resolver.h b/src/core/ext/filters/client_channel/resolver.h index cdb5a20ea3..02380314dd 100644 --- a/src/core/ext/filters/client_channel/resolver.h +++ b/src/core/ext/filters/client_channel/resolver.h @@ -105,6 +105,10 @@ class Resolver : public InternallyRefCountedWithTracing<Resolver> { GRPC_ABSTRACT_BASE_CLASS protected: + // So Delete() can access our protected dtor. + template <typename T> + friend void Delete(T*); + /// Does NOT take ownership of the reference to \a combiner. // TODO(roth): Once we have a C++-like interface for combiners, this // API should change to take a RefCountedPtr<>, so that we always take diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc index 6ac9a77635..c3c62b60bf 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc @@ -363,6 +363,15 @@ void AresDnsResolver::OnResolvedLocked(void* arg, grpc_error* error) { } void AresDnsResolver::MaybeStartResolvingLocked() { + // If there is an existing timer, the time it fires is the earliest time we + // can start the next resolution. + if (have_next_resolution_timer_) { + // TODO(dgq): remove the following two lines once Pick First stops + // discarding subchannels after selecting. + ++resolved_version_; + MaybeFinishNextLocked(); + return; + } if (last_resolution_timestamp_ >= 0) { const grpc_millis earliest_next_resolution = last_resolution_timestamp_ + min_time_between_resolutions_; @@ -375,17 +384,15 @@ void AresDnsResolver::MaybeStartResolvingLocked() { "In cooldown from last resolution (from %" PRIdPTR " ms ago). Will resolve again in %" PRIdPTR " ms", last_resolution_ago, ms_until_next_resolution); - if (!have_next_resolution_timer_) { - have_next_resolution_timer_ = true; - // TODO(roth): We currently deal with this ref manually. Once the - // new closure API is done, find a way to track this ref with the timer - // callback as part of the type system. - RefCountedPtr<Resolver> self = - Ref(DEBUG_LOCATION, "next_resolution_timer_cooldown"); - self.release(); - grpc_timer_init(&next_resolution_timer_, ms_until_next_resolution, - &on_next_resolution_); - } + have_next_resolution_timer_ = true; + // TODO(roth): We currently deal with this ref manually. Once the + // new closure API is done, find a way to track this ref with the timer + // callback as part of the type system. + RefCountedPtr<Resolver> self = + Ref(DEBUG_LOCATION, "next_resolution_timer_cooldown"); + self.release(); + grpc_timer_init(&next_resolution_timer_, ms_until_next_resolution, + &on_next_resolution_); // TODO(dgq): remove the following two lines once Pick First stops // discarding subchannels after selecting. ++resolved_version_; @@ -397,6 +404,7 @@ void AresDnsResolver::MaybeStartResolvingLocked() { } void AresDnsResolver::StartResolvingLocked() { + gpr_log(GPR_DEBUG, "Start resolving."); // TODO(roth): We currently deal with this ref manually. Once the // new closure API is done, find a way to track this ref with the timer // callback as part of the type system. diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc index fb2435749d..e86ab5a37e 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc @@ -153,7 +153,10 @@ static void grpc_ares_request_unref(grpc_ares_request* r) { /* If there are no pending queries, invoke on_done callback and destroy the request */ if (gpr_unref(&r->pending_queries)) { - grpc_cares_wrapper_address_sorting_sort(*(r->lb_addrs_out)); + grpc_lb_addresses* lb_addrs = *(r->lb_addrs_out); + if (lb_addrs != nullptr) { + grpc_cares_wrapper_address_sorting_sort(lb_addrs); + } GRPC_CLOSURE_SCHED(r->on_done, r->error); gpr_mu_destroy(&r->mu); grpc_ares_ev_driver_destroy(r->ev_driver); diff --git a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc index fbab136421..e7842a7951 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc @@ -236,6 +236,15 @@ void NativeDnsResolver::OnResolvedLocked(void* arg, grpc_error* error) { } void NativeDnsResolver::MaybeStartResolvingLocked() { + // If there is an existing timer, the time it fires is the earliest time we + // can start the next resolution. + if (have_next_resolution_timer_) { + // TODO(dgq): remove the following two lines once Pick First stops + // discarding subchannels after selecting. + ++resolved_version_; + MaybeFinishNextLocked(); + return; + } if (last_resolution_timestamp_ >= 0) { const grpc_millis earliest_next_resolution = last_resolution_timestamp_ + min_time_between_resolutions_; @@ -248,17 +257,15 @@ void NativeDnsResolver::MaybeStartResolvingLocked() { "In cooldown from last resolution (from %" PRIdPTR " ms ago). Will resolve again in %" PRIdPTR " ms", last_resolution_ago, ms_until_next_resolution); - if (!have_next_resolution_timer_) { - have_next_resolution_timer_ = true; - // TODO(roth): We currently deal with this ref manually. Once the - // new closure API is done, find a way to track this ref with the timer - // callback as part of the type system. - RefCountedPtr<Resolver> self = - Ref(DEBUG_LOCATION, "next_resolution_timer_cooldown"); - self.release(); - grpc_timer_init(&next_resolution_timer_, ms_until_next_resolution, - &on_next_resolution_); - } + have_next_resolution_timer_ = true; + // TODO(roth): We currently deal with this ref manually. Once the + // new closure API is done, find a way to track this ref with the timer + // callback as part of the type system. + RefCountedPtr<Resolver> self = + Ref(DEBUG_LOCATION, "next_resolution_timer_cooldown"); + self.release(); + grpc_timer_init(&next_resolution_timer_, ms_until_next_resolution, + &on_next_resolution_); // TODO(dgq): remove the following two lines once Pick First stops // discarding subchannels after selecting. ++resolved_version_; @@ -270,6 +277,7 @@ void NativeDnsResolver::MaybeStartResolvingLocked() { } void NativeDnsResolver::StartResolvingLocked() { + gpr_log(GPR_DEBUG, "Start resolving."); // TODO(roth): We currently deal with this ref manually. Once the // new closure API is done, find a way to track this ref with the timer // callback as part of the type system. diff --git a/src/core/ext/filters/client_channel/retry_throttle.h b/src/core/ext/filters/client_channel/retry_throttle.h index 2b6fa0a70b..fddafcd903 100644 --- a/src/core/ext/filters/client_channel/retry_throttle.h +++ b/src/core/ext/filters/client_channel/retry_throttle.h @@ -42,6 +42,10 @@ class ServerRetryThrottleData : public RefCounted<ServerRetryThrottleData> { intptr_t milli_token_ratio() const { return milli_token_ratio_; } private: + // So Delete() can call our private dtor. + template <typename T> + friend void grpc_core::Delete(T*); + ~ServerRetryThrottleData(); void GetReplacementThrottleDataIfNeeded( diff --git a/src/core/ext/filters/http/message_compress/message_compress_filter.cc b/src/core/ext/filters/http/message_compress/message_compress_filter.cc index e7d9949386..f8f478b6c0 100644 --- a/src/core/ext/filters/http/message_compress/message_compress_filter.cc +++ b/src/core/ext/filters/http/message_compress/message_compress_filter.cc @@ -234,7 +234,7 @@ static void finish_send_message(grpc_call_element* elem) { static_cast<float>(before_size); GPR_ASSERT(grpc_message_compression_algorithm_name( calld->message_compression_algorithm, &algo_name)); - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "Compressed[%s] %" PRIuPTR " bytes vs. %" PRIuPTR " bytes (%.2f%% savings)", algo_name, before_size, after_size, 100 * savings_ratio); @@ -246,7 +246,7 @@ static void finish_send_message(grpc_call_element* elem) { const char* algo_name; GPR_ASSERT(grpc_message_compression_algorithm_name( calld->message_compression_algorithm, &algo_name)); - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "Algorithm '%s' enabled but decided not to compress. Input size: " "%" PRIuPTR, algo_name, calld->slices.length); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index 6980b8671e..0ef73961a5 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -807,7 +807,7 @@ static const char* write_state_name(grpc_chttp2_write_state st) { static void set_write_state(grpc_chttp2_transport* t, grpc_chttp2_write_state st, const char* reason) { - GRPC_CHTTP2_IF_TRACING(gpr_log(GPR_DEBUG, "W:%p %s state %s -> %s [%s]", t, + GRPC_CHTTP2_IF_TRACING(gpr_log(GPR_INFO, "W:%p %s state %s -> %s [%s]", t, t->is_client ? "CLIENT" : "SERVER", write_state_name(t->write_state), write_state_name(st), reason)); @@ -1072,7 +1072,7 @@ void grpc_chttp2_add_incoming_goaway(grpc_chttp2_transport* t, uint32_t goaway_error, grpc_slice goaway_text) { // GRPC_CHTTP2_IF_TRACING( - // gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg)); + // gpr_log(GPR_INFO, "got goaway [%d]: %s", goaway_error, msg)); // Discard the error from a previous goaway frame (if any) if (t->goaway_error != GRPC_ERROR_NONE) { @@ -1118,7 +1118,7 @@ static void maybe_start_some_streams(grpc_chttp2_transport* t) { grpc_chttp2_list_pop_waiting_for_concurrency(t, &s)) { /* safe since we can't (legally) be parsing this stream yet */ GRPC_CHTTP2_IF_TRACING(gpr_log( - GPR_DEBUG, "HTTP:%s: Allocating new grpc_chttp2_stream %p to id %d", + GPR_INFO, "HTTP:%s: Allocating new grpc_chttp2_stream %p to id %d", t->is_client ? "CLI" : "SVR", s, t->next_stream_id)); GPR_ASSERT(s->id == 0); @@ -1183,7 +1183,7 @@ void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t, if (grpc_http_trace.enabled()) { const char* errstr = grpc_error_string(error); gpr_log( - GPR_DEBUG, + GPR_INFO, "complete_closure_step: t=%p %p refs=%d flags=0x%04x desc=%s err=%s " "write_state=%s", t, closure, @@ -1336,7 +1336,7 @@ static void perform_stream_op_locked(void* stream_op, if (grpc_http_trace.enabled()) { char* str = grpc_transport_stream_op_batch_string(op); - gpr_log(GPR_DEBUG, "perform_stream_op_locked: %s; on_complete = %p", str, + gpr_log(GPR_INFO, "perform_stream_op_locked: %s; on_complete = %p", str, op->on_complete); gpr_free(str); if (op->send_initial_metadata) { @@ -1638,7 +1638,7 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, if (grpc_http_trace.enabled()) { char* str = grpc_transport_stream_op_batch_string(op); - gpr_log(GPR_DEBUG, "perform_stream_op[s=%p]: %s", s, str); + gpr_log(GPR_INFO, "perform_stream_op[s=%p]: %s", s, str); gpr_free(str); } @@ -2529,7 +2529,7 @@ static void schedule_bdp_ping_locked(grpc_chttp2_transport* t) { static void start_bdp_ping_locked(void* tp, grpc_error* error) { grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(tp); if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "%s: Start BDP ping err=%s", t->peer_string, + gpr_log(GPR_INFO, "%s: Start BDP ping err=%s", t->peer_string, grpc_error_string(error)); } /* Reset the keepalive ping timer */ @@ -2542,7 +2542,7 @@ static void start_bdp_ping_locked(void* tp, grpc_error* error) { static void finish_bdp_ping_locked(void* tp, grpc_error* error) { grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(tp); if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "%s: Complete BDP ping err=%s", t->peer_string, + gpr_log(GPR_INFO, "%s: Complete BDP ping err=%s", t->peer_string, grpc_error_string(error)); } if (error != GRPC_ERROR_NONE) { @@ -2716,8 +2716,7 @@ static void keepalive_watchdog_fired_locked(void* arg, grpc_error* error) { static void connectivity_state_set(grpc_chttp2_transport* t, grpc_connectivity_state state, grpc_error* error, const char* reason) { - GRPC_CHTTP2_IF_TRACING( - gpr_log(GPR_DEBUG, "set connectivity_state=%d", state)); + GRPC_CHTTP2_IF_TRACING(gpr_log(GPR_INFO, "set connectivity_state=%d", state)); grpc_connectivity_state_set(&t->channel_callback.state_tracker, state, error, reason); } @@ -2984,7 +2983,7 @@ static void benign_reclaimer_locked(void* arg, grpc_error* error) { /* Channel with no active streams: send a goaway to try and make it * disconnect cleanly */ if (grpc_resource_quota_trace.enabled()) { - gpr_log(GPR_DEBUG, "HTTP2: %s - send goaway to free memory", + gpr_log(GPR_INFO, "HTTP2: %s - send goaway to free memory", t->peer_string); } send_goaway(t, @@ -2992,7 +2991,7 @@ static void benign_reclaimer_locked(void* arg, grpc_error* error) { GRPC_ERROR_CREATE_FROM_STATIC_STRING("Buffers full"), GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_ENHANCE_YOUR_CALM)); } else if (error == GRPC_ERROR_NONE && grpc_resource_quota_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "HTTP2: %s - skip benign reclamation, there are still %" PRIdPTR " streams", t->peer_string, grpc_chttp2_stream_map_size(&t->stream_map)); @@ -3013,7 +3012,7 @@ static void destructive_reclaimer_locked(void* arg, grpc_error* error) { grpc_chttp2_stream* s = static_cast<grpc_chttp2_stream*>( grpc_chttp2_stream_map_rand(&t->stream_map)); if (grpc_resource_quota_trace.enabled()) { - gpr_log(GPR_DEBUG, "HTTP2: %s - abandon stream id %d", t->peer_string, + gpr_log(GPR_INFO, "HTTP2: %s - abandon stream id %d", t->peer_string, s->id); } grpc_chttp2_cancel_stream( diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.cc b/src/core/ext/transport/chttp2/transport/frame_settings.cc index 9ea27dcd47..987ac0e79d 100644 --- a/src/core/ext/transport/chttp2/transport/frame_settings.cc +++ b/src/core/ext/transport/chttp2/transport/frame_settings.cc @@ -217,14 +217,14 @@ grpc_error* grpc_chttp2_settings_parser_parse(void* p, grpc_chttp2_transport* t, t->initial_window_update += static_cast<int64_t>(parser->value) - parser->incoming_settings[id]; if (grpc_http_trace.enabled() || grpc_flowctl_trace.enabled()) { - gpr_log(GPR_DEBUG, "%p[%s] adding %d for initial_window change", - t, t->is_client ? "cli" : "svr", + gpr_log(GPR_INFO, "%p[%s] adding %d for initial_window change", t, + t->is_client ? "cli" : "svr", static_cast<int>(t->initial_window_update)); } } parser->incoming_settings[id] = parser->value; if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "CHTTP2:%s:%s: got setting %s = %d", + gpr_log(GPR_INFO, "CHTTP2:%s:%s: got setting %s = %d", t->is_client ? "CLI" : "SVR", t->peer_string, sp->name, parser->value); } diff --git a/src/core/ext/transport/chttp2/transport/hpack_encoder.cc b/src/core/ext/transport/chttp2/transport/hpack_encoder.cc index e4f3c1b81e..d5ef063883 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_encoder.cc +++ b/src/core/ext/transport/chttp2/transport/hpack_encoder.cc @@ -470,7 +470,7 @@ static void hpack_enc(grpc_chttp2_hpack_compressor* c, grpc_mdelem elem, v = grpc_slice_to_c_string(GRPC_MDVALUE(elem)); } gpr_log( - GPR_DEBUG, + GPR_INFO, "Encode: '%s: %s', elem_interned=%d [%d], k_interned=%d, v_interned=%d", k, v, GRPC_MDELEM_IS_INTERNED(elem), GRPC_MDELEM_STORAGE(elem), grpc_slice_is_interned(GRPC_MDKEY(elem)), @@ -654,7 +654,7 @@ void grpc_chttp2_hpack_compressor_set_max_table_size( } c->advertise_table_size_change = 1; if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "set max table size from encoder to %d", max_table_size); + gpr_log(GPR_INFO, "set max table size from encoder to %d", max_table_size); } } diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.cc b/src/core/ext/transport/chttp2/transport/hpack_parser.cc index fc96a8b3e4..907ba71178 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_parser.cc +++ b/src/core/ext/transport/chttp2/transport/hpack_parser.cc @@ -633,7 +633,7 @@ static grpc_error* on_hdr(grpc_chttp2_hpack_parser* p, grpc_mdelem md, v = grpc_slice_to_c_string(GRPC_MDVALUE(md)); } gpr_log( - GPR_DEBUG, + GPR_INFO, "Decode: '%s: %s', elem_interned=%d [%d], k_interned=%d, v_interned=%d", k, v, GRPC_MDELEM_IS_INTERNED(md), GRPC_MDELEM_STORAGE(md), grpc_slice_is_interned(GRPC_MDKEY(md)), diff --git a/src/core/ext/transport/chttp2/transport/hpack_table.cc b/src/core/ext/transport/chttp2/transport/hpack_table.cc index f050f502f5..7929258356 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_table.cc +++ b/src/core/ext/transport/chttp2/transport/hpack_table.cc @@ -247,7 +247,7 @@ void grpc_chttp2_hptbl_set_max_bytes(grpc_chttp2_hptbl* tbl, return; } if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "Update hpack parser max size to %d", max_bytes); + gpr_log(GPR_INFO, "Update hpack parser max size to %d", max_bytes); } while (tbl->mem_used > max_bytes) { evict1(tbl); @@ -270,7 +270,7 @@ grpc_error* grpc_chttp2_hptbl_set_current_table_size(grpc_chttp2_hptbl* tbl, return err; } if (grpc_http_trace.enabled()) { - gpr_log(GPR_DEBUG, "Update hpack parser table size to %d", bytes); + gpr_log(GPR_INFO, "Update hpack parser table size to %d", bytes); } while (tbl->mem_used > bytes) { evict1(tbl); diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.cc b/src/core/ext/transport/chttp2/transport/stream_lists.cc index 5d3ec4b53b..6626170a7e 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.cc +++ b/src/core/ext/transport/chttp2/transport/stream_lists.cc @@ -68,7 +68,7 @@ static bool stream_list_pop(grpc_chttp2_transport* t, } *stream = s; if (s && grpc_trace_http2_stream_state.enabled()) { - gpr_log(GPR_DEBUG, "%p[%d][%s]: pop from %s", t, s->id, + gpr_log(GPR_INFO, "%p[%d][%s]: pop from %s", t, s->id, t->is_client ? "cli" : "svr", stream_list_id_string(id)); } return s != nullptr; @@ -90,7 +90,7 @@ static void stream_list_remove(grpc_chttp2_transport* t, grpc_chttp2_stream* s, t->lists[id].tail = s->links[id].prev; } if (grpc_trace_http2_stream_state.enabled()) { - gpr_log(GPR_DEBUG, "%p[%d][%s]: remove from %s", t, s->id, + gpr_log(GPR_INFO, "%p[%d][%s]: remove from %s", t, s->id, t->is_client ? "cli" : "svr", stream_list_id_string(id)); } } @@ -122,7 +122,7 @@ static void stream_list_add_tail(grpc_chttp2_transport* t, t->lists[id].tail = s; s->included[id] = 1; if (grpc_trace_http2_stream_state.enabled()) { - gpr_log(GPR_DEBUG, "%p[%d][%s]: add to %s", t, s->id, + gpr_log(GPR_INFO, "%p[%d][%s]: add to %s", t, s->id, t->is_client ? "cli" : "svr", stream_list_id_string(id)); } } diff --git a/src/core/ext/transport/chttp2/transport/writing.cc b/src/core/ext/transport/chttp2/transport/writing.cc index 6f32397a3a..85efe27080 100644 --- a/src/core/ext/transport/chttp2/transport/writing.cc +++ b/src/core/ext/transport/chttp2/transport/writing.cc @@ -52,7 +52,7 @@ static void maybe_initiate_ping(grpc_chttp2_transport* t) { if (!grpc_closure_list_empty(pq->lists[GRPC_CHTTP2_PCL_INFLIGHT])) { /* ping already in-flight: wait */ if (grpc_http_trace.enabled() || grpc_bdp_estimator_trace.enabled()) { - gpr_log(GPR_DEBUG, "%s: Ping delayed [%p]: already pinging", + gpr_log(GPR_INFO, "%s: Ping delayed [%p]: already pinging", t->is_client ? "CLIENT" : "SERVER", t->peer_string); } return; @@ -61,7 +61,7 @@ static void maybe_initiate_ping(grpc_chttp2_transport* t) { t->ping_policy.max_pings_without_data != 0) { /* need to receive something of substance before sending a ping again */ if (grpc_http_trace.enabled() || grpc_bdp_estimator_trace.enabled()) { - gpr_log(GPR_DEBUG, "%s: Ping delayed [%p]: too many recent pings: %d/%d", + gpr_log(GPR_INFO, "%s: Ping delayed [%p]: too many recent pings: %d/%d", t->is_client ? "CLIENT" : "SERVER", t->peer_string, t->ping_state.pings_before_data_required, t->ping_policy.max_pings_without_data); @@ -81,7 +81,7 @@ static void maybe_initiate_ping(grpc_chttp2_transport* t) { if (next_allowed_ping > now) { /* not enough elapsed time between successive pings */ if (grpc_http_trace.enabled() || grpc_bdp_estimator_trace.enabled()) { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "%s: Ping delayed [%p]: not enough time elapsed since last ping. " " Last ping %f: Next ping %f: Now %f", t->is_client ? "CLIENT" : "SERVER", t->peer_string, @@ -107,7 +107,7 @@ static void maybe_initiate_ping(grpc_chttp2_transport* t) { GRPC_STATS_INC_HTTP2_PINGS_SENT(); t->ping_state.last_ping_sent_time = now; if (grpc_http_trace.enabled() || grpc_bdp_estimator_trace.enabled()) { - gpr_log(GPR_DEBUG, "%s: Ping sent [%p]: %d/%d", + gpr_log(GPR_INFO, "%s: Ping sent [%p]: %d/%d", t->is_client ? "CLIENT" : "SERVER", t->peer_string, t->ping_state.pings_before_data_required, t->ping_policy.max_pings_without_data); @@ -401,7 +401,7 @@ class StreamWriteContext { StreamWriteContext(WriteContext* write_context, grpc_chttp2_stream* s) : write_context_(write_context), t_(write_context->transport()), s_(s) { GRPC_CHTTP2_IF_TRACING( - gpr_log(GPR_DEBUG, "W:%p %s[%d] im-(sent,send)=(%d,%d) announce=%d", t_, + gpr_log(GPR_INFO, "W:%p %s[%d] im-(sent,send)=(%d,%d) announce=%d", t_, t_->is_client ? "CLIENT" : "SERVER", s->id, s->sent_initial_metadata, s->send_initial_metadata != nullptr, (int)(s->flow_control->local_window_delta() - diff --git a/src/core/ext/transport/inproc/inproc_transport.cc b/src/core/ext/transport/inproc/inproc_transport.cc index 67a380077b..2c3bff5c1e 100644 --- a/src/core/ext/transport/inproc/inproc_transport.cc +++ b/src/core/ext/transport/inproc/inproc_transport.cc @@ -125,12 +125,12 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error); static void op_state_machine(void* arg, grpc_error* error); static void ref_transport(inproc_transport* t) { - INPROC_LOG(GPR_DEBUG, "ref_transport %p", t); + INPROC_LOG(GPR_INFO, "ref_transport %p", t); gpr_ref(&t->refs); } static void really_destroy_transport(inproc_transport* t) { - INPROC_LOG(GPR_DEBUG, "really_destroy_transport %p", t); + INPROC_LOG(GPR_INFO, "really_destroy_transport %p", t); grpc_connectivity_state_destroy(&t->connectivity); if (gpr_unref(&t->mu->refs)) { gpr_free(t->mu); @@ -139,7 +139,7 @@ static void really_destroy_transport(inproc_transport* t) { } static void unref_transport(inproc_transport* t) { - INPROC_LOG(GPR_DEBUG, "unref_transport %p", t); + INPROC_LOG(GPR_INFO, "unref_transport %p", t); if (gpr_unref(&t->refs)) { really_destroy_transport(t); } @@ -154,17 +154,17 @@ static void unref_transport(inproc_transport* t) { #endif static void ref_stream(inproc_stream* s, const char* reason) { - INPROC_LOG(GPR_DEBUG, "ref_stream %p %s", s, reason); + INPROC_LOG(GPR_INFO, "ref_stream %p %s", s, reason); STREAM_REF(s->refs, reason); } static void unref_stream(inproc_stream* s, const char* reason) { - INPROC_LOG(GPR_DEBUG, "unref_stream %p %s", s, reason); + INPROC_LOG(GPR_INFO, "unref_stream %p %s", s, reason); STREAM_UNREF(s->refs, reason); } static void really_destroy_stream(inproc_stream* s) { - INPROC_LOG(GPR_DEBUG, "really_destroy_stream %p", s); + INPROC_LOG(GPR_INFO, "really_destroy_stream %p", s); GRPC_ERROR_UNREF(s->write_buffer_cancel_error); GRPC_ERROR_UNREF(s->cancel_self_error); @@ -225,7 +225,7 @@ static grpc_error* fill_in_metadata(inproc_stream* s, static int init_stream(grpc_transport* gt, grpc_stream* gs, grpc_stream_refcount* refcount, const void* server_data, gpr_arena* arena) { - INPROC_LOG(GPR_DEBUG, "init_stream %p %p %p", gt, gs, server_data); + INPROC_LOG(GPR_INFO, "init_stream %p %p %p", gt, gs, server_data); inproc_transport* t = reinterpret_cast<inproc_transport*>(gt); inproc_stream* s = reinterpret_cast<inproc_stream*>(gs); s->arena = arena; @@ -282,8 +282,8 @@ static int init_stream(grpc_transport* gt, grpc_stream* gs, // Pass the client-side stream address to the server-side for a ref ref_stream(s, "inproc_init_stream:clt"); // ref it now on behalf of server // side to avoid destruction - INPROC_LOG(GPR_DEBUG, "calling accept stream cb %p %p", - st->accept_stream_cb, st->accept_stream_data); + INPROC_LOG(GPR_INFO, "calling accept stream cb %p %p", st->accept_stream_cb, + st->accept_stream_data); (*st->accept_stream_cb)(st->accept_stream_data, &st->base, (void*)s); } else { // This is the server-side and is being called through accept_stream_cb @@ -378,7 +378,7 @@ static void complete_if_batch_end_locked(inproc_stream* s, grpc_error* error, int is_rtm = static_cast<int>(op == s->recv_trailing_md_op); if ((is_sm + is_stm + is_rim + is_rm + is_rtm) == 1) { - INPROC_LOG(GPR_DEBUG, "%s %p %p %p", msg, s, op, error); + INPROC_LOG(GPR_INFO, "%s %p %p %p", msg, s, op, error); GRPC_CLOSURE_SCHED(op->on_complete, GRPC_ERROR_REF(error)); } } @@ -393,7 +393,7 @@ static void maybe_schedule_op_closure_locked(inproc_stream* s, } static void fail_helper_locked(inproc_stream* s, grpc_error* error) { - INPROC_LOG(GPR_DEBUG, "op_state_machine %p fail_helper", s); + INPROC_LOG(GPR_INFO, "op_state_machine %p fail_helper", s); // If we're failing this side, we need to make sure that // we also send or have already sent trailing metadata if (!s->trailing_md_sent) { @@ -458,7 +458,7 @@ static void fail_helper_locked(inproc_stream* s, grpc_error* error) { *s->recv_initial_md_op->payload->recv_initial_metadata .trailing_metadata_available = true; } - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "fail_helper %p scheduling initial-metadata-ready %p %p", s, error, err); GRPC_CLOSURE_SCHED(s->recv_initial_md_op->payload->recv_initial_metadata @@ -472,7 +472,7 @@ static void fail_helper_locked(inproc_stream* s, grpc_error* error) { s->recv_initial_md_op = nullptr; } if (s->recv_message_op) { - INPROC_LOG(GPR_DEBUG, "fail_helper %p scheduling message-ready %p", s, + INPROC_LOG(GPR_INFO, "fail_helper %p scheduling message-ready %p", s, error); GRPC_CLOSURE_SCHED( s->recv_message_op->payload->recv_message.recv_message_ready, @@ -496,9 +496,8 @@ static void fail_helper_locked(inproc_stream* s, grpc_error* error) { s->send_trailing_md_op = nullptr; } if (s->recv_trailing_md_op) { - INPROC_LOG(GPR_DEBUG, - "fail_helper %p scheduling trailing-md-on-complete %p", s, - error); + INPROC_LOG(GPR_INFO, "fail_helper %p scheduling trailing-md-on-complete %p", + s, error); complete_if_batch_end_locked( s, error, s->recv_trailing_md_op, "fail_helper scheduling recv-trailing-metadata-on-complete"); @@ -549,7 +548,7 @@ static void message_transfer_locked(inproc_stream* sender, receiver->recv_stream.Init(&receiver->recv_message, 0); receiver->recv_message_op->payload->recv_message.recv_message->reset( receiver->recv_stream.get()); - INPROC_LOG(GPR_DEBUG, "message_transfer_locked %p scheduling message-ready", + INPROC_LOG(GPR_INFO, "message_transfer_locked %p scheduling message-ready", receiver); GRPC_CLOSURE_SCHED( receiver->recv_message_op->payload->recv_message.recv_message_ready, @@ -577,7 +576,7 @@ static void op_state_machine(void* arg, grpc_error* error) { bool needs_close = false; - INPROC_LOG(GPR_DEBUG, "op_state_machine %p", arg); + INPROC_LOG(GPR_INFO, "op_state_machine %p", arg); inproc_stream* s = static_cast<inproc_stream*>(arg); gpr_mu* mu = &s->t->mu->mu; // keep aside in case s gets closed gpr_mu_lock(mu); @@ -626,7 +625,7 @@ static void op_state_machine(void* arg, grpc_error* error) { : &other->to_read_trailing_md_filled; if (*destfilled || s->trailing_md_sent) { // The buffer is already in use; that's an error! - INPROC_LOG(GPR_DEBUG, "Extra trailing metadata %p", s); + INPROC_LOG(GPR_INFO, "Extra trailing metadata %p", s); new_err = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Extra trailing metadata"); fail_helper_locked(s, GRPC_ERROR_REF(new_err)); goto done; @@ -639,7 +638,7 @@ static void op_state_machine(void* arg, grpc_error* error) { } s->trailing_md_sent = true; if (!s->t->is_client && s->trailing_md_recvd && s->recv_trailing_md_op) { - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling trailing-md-on-complete", s); GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->on_complete, GRPC_ERROR_NONE); @@ -658,7 +657,7 @@ static void op_state_machine(void* arg, grpc_error* error) { new_err = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Already recvd initial md"); INPROC_LOG( - GPR_DEBUG, + GPR_INFO, "op_state_machine %p scheduling on_complete errors for already " "recvd initial md %p", s, new_err); @@ -684,7 +683,7 @@ static void op_state_machine(void* arg, grpc_error* error) { } grpc_metadata_batch_clear(&s->to_read_initial_md); s->to_read_initial_md_filled = false; - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling initial-metadata-ready %p", s, new_err); GRPC_CLOSURE_SCHED(s->recv_initial_md_op->payload->recv_initial_metadata @@ -696,7 +695,7 @@ static void op_state_machine(void* arg, grpc_error* error) { s->recv_initial_md_op = nullptr; if (new_err != GRPC_ERROR_NONE) { - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling on_complete errors2 %p", s, new_err); fail_helper_locked(s, GRPC_ERROR_REF(new_err)); @@ -719,7 +718,7 @@ static void op_state_machine(void* arg, grpc_error* error) { new_err = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Already recvd trailing md"); INPROC_LOG( - GPR_DEBUG, + GPR_INFO, "op_state_machine %p scheduling on_complete errors for already " "recvd trailing md %p", s, new_err); @@ -729,7 +728,7 @@ static void op_state_machine(void* arg, grpc_error* error) { if (s->recv_message_op != nullptr) { // This message needs to be wrapped up because it will never be // satisfied - INPROC_LOG(GPR_DEBUG, "op_state_machine %p scheduling message-ready", s); + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling message-ready", s); GRPC_CLOSURE_SCHED( s->recv_message_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); @@ -764,7 +763,7 @@ static void op_state_machine(void* arg, grpc_error* error) { // (If the server hasn't already sent its trailing md, it doesn't have // a final status, so don't mark this op complete) if (s->t->is_client || s->trailing_md_sent) { - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling trailing-md-on-complete %p", s, new_err); GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->on_complete, @@ -772,21 +771,21 @@ static void op_state_machine(void* arg, grpc_error* error) { s->recv_trailing_md_op = nullptr; needs_close = true; } else { - INPROC_LOG(GPR_DEBUG, + INPROC_LOG(GPR_INFO, "op_state_machine %p server needs to delay handling " "trailing-md-on-complete %p", s, new_err); } } else { INPROC_LOG( - GPR_DEBUG, + GPR_INFO, "op_state_machine %p has trailing md but not yet waiting for it", s); } } if (s->trailing_md_recvd && s->recv_message_op) { // No further message will come on this stream, so finish off the // recv_message_op - INPROC_LOG(GPR_DEBUG, "op_state_machine %p scheduling message-ready", s); + INPROC_LOG(GPR_INFO, "op_state_machine %p scheduling message-ready", s); GRPC_CLOSURE_SCHED( s->recv_message_op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE); @@ -810,7 +809,7 @@ static void op_state_machine(void* arg, grpc_error* error) { // Didn't get the item we wanted so we still need to get // rescheduled INPROC_LOG( - GPR_DEBUG, "op_state_machine %p still needs closure %p %p %p %p %p", s, + GPR_INFO, "op_state_machine %p still needs closure %p %p %p %p %p", s, s->send_message_op, s->send_trailing_md_op, s->recv_initial_md_op, s->recv_message_op, s->recv_trailing_md_op); s->ops_needed = true; @@ -826,8 +825,7 @@ done: static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) { bool ret = false; // was the cancel accepted - INPROC_LOG(GPR_DEBUG, "cancel_stream %p with %s", s, - grpc_error_string(error)); + INPROC_LOG(GPR_INFO, "cancel_stream %p with %s", s, grpc_error_string(error)); if (s->cancel_self_error == GRPC_ERROR_NONE) { ret = true; s->cancel_self_error = GRPC_ERROR_REF(error); @@ -877,7 +875,7 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) { static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, grpc_transport_stream_op_batch* op) { - INPROC_LOG(GPR_DEBUG, "perform_stream_op %p %p %p", gt, gs, op); + INPROC_LOG(GPR_INFO, "perform_stream_op %p %p %p", gt, gs, op); inproc_stream* s = reinterpret_cast<inproc_stream*>(gs); gpr_mu* mu = &s->t->mu->mu; // save aside in case s gets closed gpr_mu_lock(mu); @@ -907,7 +905,7 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, // already self-canceled so still give it an error error = GRPC_ERROR_REF(s->cancel_self_error); } else { - INPROC_LOG(GPR_DEBUG, "perform_stream_op %p %s%s%s%s%s%s%s", s, + INPROC_LOG(GPR_INFO, "perform_stream_op %p %s%s%s%s%s%s%s", s, s->t->is_client ? "client" : "server", op->send_initial_metadata ? " send_initial_metadata" : "", op->send_message ? " send_message" : "", @@ -936,7 +934,7 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, : &other->to_read_initial_md_filled; if (*destfilled || s->initial_md_sent) { // The buffer is already in use; that's an error! - INPROC_LOG(GPR_DEBUG, "Extra initial metadata %p", s); + INPROC_LOG(GPR_INFO, "Extra initial metadata %p", s); error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Extra initial metadata"); } else { if (!other || !other->closed) { @@ -1013,7 +1011,7 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, true; } INPROC_LOG( - GPR_DEBUG, + GPR_INFO, "perform_stream_op error %p scheduling initial-metadata-ready %p", s, error); GRPC_CLOSURE_SCHED( @@ -1022,14 +1020,14 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, } if (op->recv_message) { INPROC_LOG( - GPR_DEBUG, + GPR_INFO, "perform_stream_op error %p scheduling recv message-ready %p", s, error); GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready, GRPC_ERROR_REF(error)); } } - INPROC_LOG(GPR_DEBUG, "perform_stream_op %p scheduling on_complete %p", s, + INPROC_LOG(GPR_INFO, "perform_stream_op %p scheduling on_complete %p", s, error); GRPC_CLOSURE_SCHED(on_complete, GRPC_ERROR_REF(error)); } @@ -1042,7 +1040,7 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, } static void close_transport_locked(inproc_transport* t) { - INPROC_LOG(GPR_DEBUG, "close_transport %p %d", t, t->is_closed); + INPROC_LOG(GPR_INFO, "close_transport %p %d", t, t->is_closed); grpc_connectivity_state_set( &t->connectivity, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Closing transport."), @@ -1063,7 +1061,7 @@ static void close_transport_locked(inproc_transport* t) { static void perform_transport_op(grpc_transport* gt, grpc_transport_op* op) { inproc_transport* t = reinterpret_cast<inproc_transport*>(gt); - INPROC_LOG(GPR_DEBUG, "perform_transport_op %p %p", t, op); + INPROC_LOG(GPR_INFO, "perform_transport_op %p %p", t, op); gpr_mu_lock(&t->mu->mu); if (op->on_connectivity_state_change) { grpc_connectivity_state_notify_on_state_change( @@ -1096,7 +1094,7 @@ static void perform_transport_op(grpc_transport* gt, grpc_transport_op* op) { static void destroy_stream(grpc_transport* gt, grpc_stream* gs, grpc_closure* then_schedule_closure) { - INPROC_LOG(GPR_DEBUG, "destroy_stream %p %p", gs, then_schedule_closure); + INPROC_LOG(GPR_INFO, "destroy_stream %p %p", gs, then_schedule_closure); inproc_stream* s = reinterpret_cast<inproc_stream*>(gs); s->closure_at_destroy = then_schedule_closure; really_destroy_stream(s); @@ -1104,7 +1102,7 @@ static void destroy_stream(grpc_transport* gt, grpc_stream* gs, static void destroy_transport(grpc_transport* gt) { inproc_transport* t = reinterpret_cast<inproc_transport*>(gt); - INPROC_LOG(GPR_DEBUG, "destroy_transport %p", t); + INPROC_LOG(GPR_INFO, "destroy_transport %p", t); gpr_mu_lock(&t->mu->mu); close_transport_locked(t); gpr_mu_unlock(&t->mu->mu); @@ -1165,7 +1163,7 @@ static void inproc_transports_create(grpc_transport** server_transport, const grpc_channel_args* server_args, grpc_transport** client_transport, const grpc_channel_args* client_args) { - INPROC_LOG(GPR_DEBUG, "inproc_transports_create"); + INPROC_LOG(GPR_INFO, "inproc_transports_create"); inproc_transport* st = static_cast<inproc_transport*>(gpr_zalloc(sizeof(*st))); inproc_transport* ct = |