diff options
Diffstat (limited to 'src/core')
31 files changed, 1196 insertions, 774 deletions
diff --git a/src/core/ext/census/grpc_context.cc b/src/core/ext/census/grpc_context.cc deleted file mode 100644 index 599a798dda..0000000000 --- a/src/core/ext/census/grpc_context.cc +++ /dev/null @@ -1,38 +0,0 @@ -/* - * - * Copyright 2015 gRPC authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -#include <grpc/support/port_platform.h> - -#include <grpc/census.h> -#include <grpc/grpc.h> -#include "src/core/lib/surface/api_trace.h" -#include "src/core/lib/surface/call.h" - -void grpc_census_call_set_context(grpc_call* call, census_context* context) { - GRPC_API_TRACE("grpc_census_call_set_context(call=%p, census_context=%p)", 2, - (call, context)); - if (context != nullptr) { - grpc_call_context_set(call, GRPC_CONTEXT_TRACING, context, nullptr); - } -} - -census_context* grpc_census_call_get_context(grpc_call* call) { - GRPC_API_TRACE("grpc_census_call_get_context(call=%p)", 1, (call)); - return static_cast<census_context*>( - grpc_call_context_get(call, GRPC_CONTEXT_TRACING)); -} diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index ea6775a8d8..520431e63b 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -794,6 +794,15 @@ typedef struct { // The batch to use in the subchannel call. // Its payload field points to subchannel_call_retry_state.batch_payload. grpc_transport_stream_op_batch batch; + // For intercepting on_complete. + grpc_closure on_complete; +} subchannel_batch_data; + +// Retry state associated with a subchannel call. +// Stored in the parent_data of the subchannel call object. +typedef struct { + // subchannel_batch_data.batch.payload points to this. + grpc_transport_stream_op_batch_payload batch_payload; // For send_initial_metadata. // Note that we need to make a copy of the initial metadata for each // subchannel call instead of just referring to the copy in call_data, @@ -817,15 +826,7 @@ typedef struct { // For intercepting recv_trailing_metadata. grpc_metadata_batch recv_trailing_metadata; grpc_transport_stream_stats collect_stats; - // For intercepting on_complete. - grpc_closure on_complete; -} subchannel_batch_data; - -// Retry state associated with a subchannel call. -// Stored in the parent_data of the subchannel call object. -typedef struct { - // subchannel_batch_data.batch.payload points to this. - grpc_transport_stream_op_batch_payload batch_payload; + grpc_closure recv_trailing_metadata_ready; // These fields indicate which ops have been started and completed on // this subchannel call. size_t started_send_message_count; @@ -1192,35 +1193,24 @@ static void pending_batches_fail(grpc_call_element* elem, grpc_error* error, "chand=%p calld=%p: failing %" PRIuPTR " pending batches: %s", elem->channel_data, calld, num_batches, grpc_error_string(error)); } - grpc_transport_stream_op_batch* - batches[GPR_ARRAY_SIZE(calld->pending_batches)]; - size_t num_batches = 0; + grpc_core::CallCombinerClosureList closures; for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { pending_batch* pending = &calld->pending_batches[i]; grpc_transport_stream_op_batch* batch = pending->batch; if (batch != nullptr) { - batches[num_batches++] = batch; + batch->handler_private.extra_arg = calld; + GRPC_CLOSURE_INIT(&batch->handler_private.closure, + fail_pending_batch_in_call_combiner, batch, + grpc_schedule_on_exec_ctx); + closures.Add(&batch->handler_private.closure, GRPC_ERROR_REF(error), + "pending_batches_fail"); pending_batch_clear(calld, pending); } } - for (size_t i = yield_call_combiner ? 1 : 0; i < num_batches; ++i) { - grpc_transport_stream_op_batch* batch = batches[i]; - batch->handler_private.extra_arg = calld; - GRPC_CLOSURE_INIT(&batch->handler_private.closure, - fail_pending_batch_in_call_combiner, batch, - grpc_schedule_on_exec_ctx); - GRPC_CALL_COMBINER_START(calld->call_combiner, - &batch->handler_private.closure, - GRPC_ERROR_REF(error), "pending_batches_fail"); - } if (yield_call_combiner) { - if (num_batches > 0) { - // Note: This will release the call combiner. - grpc_transport_stream_op_batch_finish_with_failure( - batches[0], GRPC_ERROR_REF(error), calld->call_combiner); - } else { - GRPC_CALL_COMBINER_STOP(calld->call_combiner, "pending_batches_fail"); - } + closures.RunClosures(calld->call_combiner); + } else { + closures.RunClosuresWithoutYielding(calld->call_combiner); } GRPC_ERROR_UNREF(error); } @@ -1255,30 +1245,22 @@ static void pending_batches_resume(grpc_call_element* elem) { " pending batches on subchannel_call=%p", chand, calld, num_batches, calld->subchannel_call); } - grpc_transport_stream_op_batch* - batches[GPR_ARRAY_SIZE(calld->pending_batches)]; - size_t num_batches = 0; + grpc_core::CallCombinerClosureList closures; for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { pending_batch* pending = &calld->pending_batches[i]; grpc_transport_stream_op_batch* batch = pending->batch; if (batch != nullptr) { - batches[num_batches++] = batch; + batch->handler_private.extra_arg = calld->subchannel_call; + GRPC_CLOSURE_INIT(&batch->handler_private.closure, + resume_pending_batch_in_call_combiner, batch, + grpc_schedule_on_exec_ctx); + closures.Add(&batch->handler_private.closure, GRPC_ERROR_NONE, + "pending_batches_resume"); pending_batch_clear(calld, pending); } } - for (size_t i = 1; i < num_batches; ++i) { - grpc_transport_stream_op_batch* batch = batches[i]; - batch->handler_private.extra_arg = calld->subchannel_call; - GRPC_CLOSURE_INIT(&batch->handler_private.closure, - resume_pending_batch_in_call_combiner, batch, - grpc_schedule_on_exec_ctx); - GRPC_CALL_COMBINER_START(calld->call_combiner, - &batch->handler_private.closure, GRPC_ERROR_NONE, - "pending_batches_resume"); - } - GPR_ASSERT(num_batches > 0); // Note: This will release the call combiner. - grpc_subchannel_call_process_op(calld->subchannel_call, batches[0]); + closures.RunClosures(calld->call_combiner); } static void maybe_clear_pending_batch(grpc_call_element* elem, @@ -1293,7 +1275,10 @@ static void maybe_clear_pending_batch(grpc_call_element* elem, batch->payload->recv_initial_metadata.recv_initial_metadata_ready == nullptr) && (!batch->recv_message || - batch->payload->recv_message.recv_message_ready == nullptr)) { + batch->payload->recv_message.recv_message_ready == nullptr) && + (!batch->recv_trailing_metadata || + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready == + nullptr)) { if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, "chand=%p calld=%p: clearing pending batch", chand, calld); @@ -1302,75 +1287,27 @@ static void maybe_clear_pending_batch(grpc_call_element* elem, } } -// Returns true if all ops in the pending batch have been completed. -static bool pending_batch_is_completed( - pending_batch* pending, call_data* calld, - subchannel_call_retry_state* retry_state) { - if (pending->batch == nullptr || pending->batch->on_complete == nullptr) { - return false; - } - if (pending->batch->send_initial_metadata && - !retry_state->completed_send_initial_metadata) { - return false; - } - if (pending->batch->send_message && - retry_state->completed_send_message_count < - calld->send_messages->size()) { - return false; - } - if (pending->batch->send_trailing_metadata && - !retry_state->completed_send_trailing_metadata) { - return false; - } - if (pending->batch->recv_initial_metadata && - !retry_state->completed_recv_initial_metadata) { - return false; - } - if (pending->batch->recv_message && - retry_state->completed_recv_message_count < - retry_state->started_recv_message_count) { - return false; - } - if (pending->batch->recv_trailing_metadata && - !retry_state->completed_recv_trailing_metadata) { - return false; - } - return true; -} - -// Returns true if any op in the batch was not yet started. -static bool pending_batch_is_unstarted( - pending_batch* pending, call_data* calld, - subchannel_call_retry_state* retry_state) { - if (pending->batch == nullptr || pending->batch->on_complete == nullptr) { - return false; - } - if (pending->batch->send_initial_metadata && - !retry_state->started_send_initial_metadata) { - return true; - } - if (pending->batch->send_message && - retry_state->started_send_message_count < calld->send_messages->size()) { - return true; - } - if (pending->batch->send_trailing_metadata && - !retry_state->started_send_trailing_metadata) { - return true; - } - if (pending->batch->recv_initial_metadata && - !retry_state->started_recv_initial_metadata) { - return true; - } - if (pending->batch->recv_message && - retry_state->completed_recv_message_count == - retry_state->started_recv_message_count) { - return true; - } - if (pending->batch->recv_trailing_metadata && - !retry_state->started_recv_trailing_metadata) { - return true; +// Returns a pointer to the first pending batch for which predicate(batch) +// returns true, or null if not found. +template <typename Predicate> +static pending_batch* pending_batch_find(grpc_call_element* elem, + const char* log_message, + Predicate predicate) { + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + 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]; + grpc_transport_stream_op_batch* batch = pending->batch; + if (batch != nullptr && predicate(batch)) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: %s pending batch at index %" PRIuPTR, chand, + calld, log_message, i); + } + return pending; + } } - return false; + return nullptr; } // @@ -1557,8 +1494,13 @@ static bool maybe_retry(grpc_call_element* elem, // subchannel_batch_data // +// Creates a subchannel_batch_data object on the call's arena with the +// specified refcount. If set_on_complete is true, the batch's +// on_complete callback will be set to point to on_complete(); +// otherwise, the batch's on_complete callback will be null. static subchannel_batch_data* batch_data_create(grpc_call_element* elem, - int refcount) { + int refcount, + bool set_on_complete) { call_data* calld = static_cast<call_data*>(elem->call_data); subchannel_call_retry_state* retry_state = static_cast<subchannel_call_retry_state*>( @@ -1571,26 +1513,32 @@ static subchannel_batch_data* batch_data_create(grpc_call_element* elem, GRPC_SUBCHANNEL_CALL_REF(calld->subchannel_call, "batch_data_create"); batch_data->batch.payload = &retry_state->batch_payload; gpr_ref_init(&batch_data->refs, refcount); - GRPC_CLOSURE_INIT(&batch_data->on_complete, on_complete, batch_data, - grpc_schedule_on_exec_ctx); - batch_data->batch.on_complete = &batch_data->on_complete; + if (set_on_complete) { + GRPC_CLOSURE_INIT(&batch_data->on_complete, on_complete, batch_data, + grpc_schedule_on_exec_ctx); + batch_data->batch.on_complete = &batch_data->on_complete; + } GRPC_CALL_STACK_REF(calld->owning_call, "batch_data"); return batch_data; } static void batch_data_unref(subchannel_batch_data* batch_data) { if (gpr_unref(&batch_data->refs)) { - if (batch_data->send_initial_metadata_storage != nullptr) { - grpc_metadata_batch_destroy(&batch_data->send_initial_metadata); + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); + if (batch_data->batch.send_initial_metadata) { + grpc_metadata_batch_destroy(&retry_state->send_initial_metadata); } - if (batch_data->send_trailing_metadata_storage != nullptr) { - grpc_metadata_batch_destroy(&batch_data->send_trailing_metadata); + if (batch_data->batch.send_trailing_metadata) { + grpc_metadata_batch_destroy(&retry_state->send_trailing_metadata); } if (batch_data->batch.recv_initial_metadata) { - grpc_metadata_batch_destroy(&batch_data->recv_initial_metadata); + grpc_metadata_batch_destroy(&retry_state->recv_initial_metadata); } if (batch_data->batch.recv_trailing_metadata) { - grpc_metadata_batch_destroy(&batch_data->recv_trailing_metadata); + grpc_metadata_batch_destroy(&retry_state->recv_trailing_metadata); } GRPC_SUBCHANNEL_CALL_UNREF(batch_data->subchannel_call, "batch_data_unref"); call_data* calld = static_cast<call_data*>(batch_data->elem->call_data); @@ -1606,30 +1554,22 @@ static void batch_data_unref(subchannel_batch_data* batch_data) { static void invoke_recv_initial_metadata_callback(void* arg, grpc_error* error) { subchannel_batch_data* batch_data = static_cast<subchannel_batch_data*>(arg); - channel_data* chand = - static_cast<channel_data*>(batch_data->elem->channel_data); - call_data* calld = static_cast<call_data*>(batch_data->elem->call_data); // Find pending batch. - pending_batch* pending = nullptr; - for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { - grpc_transport_stream_op_batch* batch = calld->pending_batches[i].batch; - if (batch != nullptr && batch->recv_initial_metadata && - batch->payload->recv_initial_metadata.recv_initial_metadata_ready != - nullptr) { - if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_INFO, - "chand=%p calld=%p: invoking recv_initial_metadata_ready for " - "pending batch at index %" PRIuPTR, - chand, calld, i); - } - pending = &calld->pending_batches[i]; - break; - } - } + pending_batch* pending = pending_batch_find( + batch_data->elem, "invoking recv_initial_metadata_ready for", + [](grpc_transport_stream_op_batch* batch) { + return batch->recv_initial_metadata && + batch->payload->recv_initial_metadata + .recv_initial_metadata_ready != nullptr; + }); GPR_ASSERT(pending != nullptr); // Return metadata. + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); grpc_metadata_batch_move( - &batch_data->recv_initial_metadata, + &retry_state->recv_initial_metadata, pending->batch->payload->recv_initial_metadata.recv_initial_metadata); // Update bookkeeping. // Note: Need to do this before invoking the callback, since invoking @@ -1661,11 +1601,20 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) { static_cast<subchannel_call_retry_state*>( grpc_connected_subchannel_call_get_parent_data( batch_data->subchannel_call)); + retry_state->completed_recv_initial_metadata = true; + // If a retry was already dispatched, then we're not going to use the + // result of this recv_initial_metadata op, so do nothing. + if (retry_state->retry_dispatched) { + GRPC_CALL_COMBINER_STOP( + calld->call_combiner, + "recv_initial_metadata_ready after retry dispatched"); + return; + } // If we got an error or a Trailers-Only response and have not yet gotten - // the recv_trailing_metadata on_complete callback, then defer - // propagating this callback back to the surface. We can evaluate whether - // to retry when recv_trailing_metadata comes back. - if (GPR_UNLIKELY((batch_data->trailing_metadata_available || + // the recv_trailing_metadata_ready callback, then defer propagating this + // callback back to the surface. We can evaluate whether to retry when + // recv_trailing_metadata comes back. + if (GPR_UNLIKELY((retry_state->trailing_metadata_available || error != GRPC_ERROR_NONE) && !retry_state->completed_recv_trailing_metadata)) { if (grpc_client_channel_trace.enabled()) { @@ -1689,9 +1638,9 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) { } // Received valid initial metadata, so commit the call. retry_commit(elem, retry_state); + // Invoke the callback to return the result to the surface. // Manually invoking a callback function; it does not take ownership of error. invoke_recv_initial_metadata_callback(batch_data, error); - GRPC_ERROR_UNREF(error); } // @@ -1701,29 +1650,21 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) { // Invokes recv_message_ready for a subchannel batch. static void invoke_recv_message_callback(void* arg, grpc_error* error) { subchannel_batch_data* batch_data = static_cast<subchannel_batch_data*>(arg); - channel_data* chand = - static_cast<channel_data*>(batch_data->elem->channel_data); - call_data* calld = static_cast<call_data*>(batch_data->elem->call_data); // Find pending op. - pending_batch* pending = nullptr; - for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { - grpc_transport_stream_op_batch* batch = calld->pending_batches[i].batch; - if (batch != nullptr && batch->recv_message && - batch->payload->recv_message.recv_message_ready != nullptr) { - if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_INFO, - "chand=%p calld=%p: invoking recv_message_ready for " - "pending batch at index %" PRIuPTR, - chand, calld, i); - } - pending = &calld->pending_batches[i]; - break; - } - } + pending_batch* pending = pending_batch_find( + batch_data->elem, "invoking recv_message_ready for", + [](grpc_transport_stream_op_batch* batch) { + return batch->recv_message && + batch->payload->recv_message.recv_message_ready != nullptr; + }); GPR_ASSERT(pending != nullptr); // Return payload. + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); *pending->batch->payload->recv_message.recv_message = - std::move(batch_data->recv_message); + std::move(retry_state->recv_message); // Update bookkeeping. // Note: Need to do this before invoking the callback, since invoking // the callback will result in yielding the call combiner. @@ -1751,12 +1692,20 @@ static void recv_message_ready(void* arg, grpc_error* error) { static_cast<subchannel_call_retry_state*>( grpc_connected_subchannel_call_get_parent_data( batch_data->subchannel_call)); + ++retry_state->completed_recv_message_count; + // If a retry was already dispatched, then we're not going to use the + // result of this recv_message op, so do nothing. + if (retry_state->retry_dispatched) { + GRPC_CALL_COMBINER_STOP(calld->call_combiner, + "recv_message_ready after retry dispatched"); + return; + } // If we got an error or the payload was nullptr and we have not yet gotten - // the recv_trailing_metadata on_complete callback, then defer - // propagating this callback back to the surface. We can evaluate whether - // to retry when recv_trailing_metadata comes back. + // the recv_trailing_metadata_ready callback, then defer propagating this + // callback back to the surface. We can evaluate whether to retry when + // recv_trailing_metadata comes back. if (GPR_UNLIKELY( - (batch_data->recv_message == nullptr || error != GRPC_ERROR_NONE) && + (retry_state->recv_message == nullptr || error != GRPC_ERROR_NONE) && !retry_state->completed_recv_trailing_metadata)) { if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, @@ -1777,133 +1726,272 @@ static void recv_message_ready(void* arg, grpc_error* error) { } // Received a valid message, so commit the call. retry_commit(elem, retry_state); + // Invoke the callback to return the result to the surface. // Manually invoking a callback function; it does not take ownership of error. invoke_recv_message_callback(batch_data, error); - GRPC_ERROR_UNREF(error); } // -// list of closures to execute in call combiner +// recv_trailing_metadata handling // -// 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); +// Sets *status and *server_pushback_md based on batch_data and error. +static void get_call_status(subchannel_batch_data* batch_data, + grpc_error* error, grpc_status_code* status, + grpc_mdelem** server_pushback_md) { + grpc_call_element* elem = batch_data->elem; 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)); - } - } + if (error != GRPC_ERROR_NONE) { + grpc_error_get_status(error, calld->deadline, status, nullptr, nullptr, + nullptr); } 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_metadata_batch* md_batch = + batch_data->batch.payload->recv_trailing_metadata + .recv_trailing_metadata; + GPR_ASSERT(md_batch->idx.named.grpc_status != nullptr); + *status = + grpc_get_status_code_from_metadata(md_batch->idx.named.grpc_status->md); + if (md_batch->idx.named.grpc_retry_pushback_ms != nullptr) { + *server_pushback_md = &md_batch->idx.named.grpc_retry_pushback_ms->md; } - GRPC_CALL_COMBINER_STOP(calld->call_combiner, "no closures to run"); } + GRPC_ERROR_UNREF(error); } -// -// on_complete callback handling -// - -// Updates retry_state to reflect the ops completed in batch_data. -static void update_retry_state_for_completed_batch( - subchannel_batch_data* batch_data, - subchannel_call_retry_state* retry_state) { - if (batch_data->batch.send_initial_metadata) { - retry_state->completed_send_initial_metadata = true; - } - if (batch_data->batch.send_message) { - ++retry_state->completed_send_message_count; - } - if (batch_data->batch.send_trailing_metadata) { - retry_state->completed_send_trailing_metadata = true; - } - if (batch_data->batch.recv_initial_metadata) { - retry_state->completed_recv_initial_metadata = true; - } - if (batch_data->batch.recv_message) { - ++retry_state->completed_recv_message_count; - } - if (batch_data->batch.recv_trailing_metadata) { - retry_state->completed_recv_trailing_metadata = true; +// Adds recv_trailing_metadata_ready closure to closures. +static void add_closure_for_recv_trailing_metadata_ready( + grpc_call_element* elem, subchannel_batch_data* batch_data, + grpc_error* error, grpc_core::CallCombinerClosureList* closures) { + // Find pending batch. + pending_batch* pending = pending_batch_find( + elem, "invoking recv_trailing_metadata for", + [](grpc_transport_stream_op_batch* batch) { + return batch->recv_trailing_metadata && + batch->payload->recv_trailing_metadata + .recv_trailing_metadata_ready != nullptr; + }); + // If we generated the recv_trailing_metadata op internally via + // start_internal_recv_trailing_metadata(), then there will be no + // pending batch. + if (pending == nullptr) { + GRPC_ERROR_UNREF(error); + return; } + // Return metadata. + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); + grpc_metadata_batch_move( + &retry_state->recv_trailing_metadata, + pending->batch->payload->recv_trailing_metadata.recv_trailing_metadata); + // Add closure. + closures->Add(pending->batch->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + error, "recv_trailing_metadata_ready for pending batch"); + // Update bookkeeping. + pending->batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready = + nullptr; + maybe_clear_pending_batch(elem, pending); } // Adds any necessary closures for deferred recv_initial_metadata and -// recv_message callbacks to closures, updating *num_closures as needed. +// recv_message callbacks to closures. 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) { + grpc_core::CallCombinerClosureList* closures) { if (batch_data->batch.recv_trailing_metadata) { // Add closure for deferred recv_initial_metadata_ready. if (GPR_UNLIKELY(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"; + GRPC_CLOSURE_INIT(&retry_state->recv_initial_metadata_ready, + invoke_recv_initial_metadata_callback, + retry_state->recv_initial_metadata_ready_deferred_batch, + grpc_schedule_on_exec_ctx); + closures->Add(&retry_state->recv_initial_metadata_ready, + retry_state->recv_initial_metadata_error, + "resuming recv_initial_metadata_ready"); retry_state->recv_initial_metadata_ready_deferred_batch = nullptr; } // Add closure for deferred recv_message_ready. if (GPR_UNLIKELY(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"; + GRPC_CLOSURE_INIT(&retry_state->recv_message_ready, + invoke_recv_message_callback, + retry_state->recv_message_ready_deferred_batch, + grpc_schedule_on_exec_ctx); + closures->Add(&retry_state->recv_message_ready, + retry_state->recv_message_error, + "resuming recv_message_ready"); retry_state->recv_message_ready_deferred_batch = nullptr; } } } +// Returns true if any op in the batch was not yet started. +// Only looks at send ops, since recv ops are always started immediately. +static bool pending_batch_is_unstarted( + pending_batch* pending, call_data* calld, + subchannel_call_retry_state* retry_state) { + if (pending->batch == nullptr || pending->batch->on_complete == nullptr) { + return false; + } + if (pending->batch->send_initial_metadata && + !retry_state->started_send_initial_metadata) { + return true; + } + if (pending->batch->send_message && + retry_state->started_send_message_count < calld->send_messages->size()) { + return true; + } + if (pending->batch->send_trailing_metadata && + !retry_state->started_send_trailing_metadata) { + return true; + } + return false; +} + +// For any pending batch containing an op that has not yet been started, +// adds the pending batch's completion closures to closures. +static void add_closures_to_fail_unstarted_pending_batches( + grpc_call_element* elem, subchannel_call_retry_state* retry_state, + grpc_error* error, grpc_core::CallCombinerClosureList* closures) { + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + 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]; + if (pending_batch_is_unstarted(pending, calld, retry_state)) { + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: failing unstarted pending batch at index " + "%" PRIuPTR, + chand, calld, i); + } + closures->Add(pending->batch->on_complete, GRPC_ERROR_REF(error), + "failing on_complete for pending batch"); + pending->batch->on_complete = nullptr; + maybe_clear_pending_batch(elem, pending); + } + } + GRPC_ERROR_UNREF(error); +} + +// Runs necessary closures upon completion of a call attempt. +static void run_closures_for_completed_call(subchannel_batch_data* batch_data, + grpc_error* error) { + grpc_call_element* elem = batch_data->elem; + call_data* calld = static_cast<call_data*>(elem->call_data); + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); + // Construct list of closures to execute. + grpc_core::CallCombinerClosureList closures; + // First, add closure for recv_trailing_metadata_ready. + add_closure_for_recv_trailing_metadata_ready( + elem, batch_data, GRPC_ERROR_REF(error), &closures); + // If there are deferred recv_initial_metadata_ready or recv_message_ready + // callbacks, add them to closures. + add_closures_for_deferred_recv_callbacks(batch_data, retry_state, &closures); + // Add closures to fail any pending batches that have not yet been started. + add_closures_to_fail_unstarted_pending_batches( + elem, retry_state, GRPC_ERROR_REF(error), &closures); + // Don't need batch_data anymore. + batch_data_unref(batch_data); + // Schedule all of the closures identified above. + // Note: This will release the call combiner. + closures.RunClosures(calld->call_combiner); + GRPC_ERROR_UNREF(error); +} + +// Intercepts recv_trailing_metadata_ready callback for retries. +// Commits the call and returns the trailing metadata up the stack. +static void recv_trailing_metadata_ready(void* arg, grpc_error* error) { + subchannel_batch_data* batch_data = static_cast<subchannel_batch_data*>(arg); + grpc_call_element* elem = batch_data->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_INFO, + "chand=%p calld=%p: got recv_trailing_metadata_ready, error=%s", + chand, calld, grpc_error_string(error)); + } + subchannel_call_retry_state* retry_state = + static_cast<subchannel_call_retry_state*>( + grpc_connected_subchannel_call_get_parent_data( + batch_data->subchannel_call)); + retry_state->completed_recv_trailing_metadata = true; + // Get the call's status and check for server pushback metadata. + grpc_status_code status = GRPC_STATUS_OK; + grpc_mdelem* server_pushback_md = nullptr; + get_call_status(batch_data, GRPC_ERROR_REF(error), &status, + &server_pushback_md); + 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)); + } + // Check if we should retry. + 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 (retry_state->recv_initial_metadata_ready_deferred_batch != nullptr) { + batch_data_unref(batch_data); + GRPC_ERROR_UNREF(retry_state->recv_initial_metadata_error); + } + if (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); + return; + } + // Not retrying, so commit the call. + retry_commit(elem, retry_state); + // Run any necessary closures. + run_closures_for_completed_call(batch_data, GRPC_ERROR_REF(error)); +} + +// +// on_complete callback handling +// + +// Adds the on_complete closure for the pending batch completed in +// batch_data to closures. +static void add_closure_for_completed_pending_batch( + grpc_call_element* elem, subchannel_batch_data* batch_data, + subchannel_call_retry_state* retry_state, grpc_error* error, + grpc_core::CallCombinerClosureList* closures) { + pending_batch* pending = pending_batch_find( + elem, "completed", [batch_data](grpc_transport_stream_op_batch* batch) { + // Match the pending batch with the same set of send ops as the + // subchannel batch we've just completed. + return batch->on_complete != nullptr && + batch_data->batch.send_initial_metadata == + batch->send_initial_metadata && + batch_data->batch.send_message == batch->send_message && + batch_data->batch.send_trailing_metadata == + batch->send_trailing_metadata; + }); + // If batch_data is a replay batch, then there will be no pending + // batch to complete. + if (pending == nullptr) { + GRPC_ERROR_UNREF(error); + return; + } + // Add closure. + closures->Add(pending->batch->on_complete, error, + "on_complete for pending batch"); + pending->batch->on_complete = nullptr; + maybe_clear_pending_batch(elem, pending); +} + // If there are any cached ops to replay or pending ops to start on the // subchannel call, adds a closure to closures to invoke -// start_retriable_subchannel_batches(), updating *num_closures as needed. +// start_retriable_subchannel_batches(). static void add_closures_for_replay_or_pending_send_ops( grpc_call_element* elem, subchannel_batch_data* batch_data, - subchannel_call_retry_state* retry_state, closure_to_execute* closures, - size_t* num_closures) { + subchannel_call_retry_state* retry_state, + grpc_core::CallCombinerClosureList* closures) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); call_data* calld = static_cast<call_data*>(elem->call_data); bool have_pending_send_message_ops = @@ -1929,93 +2017,12 @@ static void add_closures_for_replay_or_pending_send_ops( "chand=%p calld=%p: starting next batch for pending send op(s)", chand, calld); } - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = GRPC_CLOSURE_INIT( - &batch_data->batch.handler_private.closure, - start_retriable_subchannel_batches, elem, grpc_schedule_on_exec_ctx); - closure->error = GRPC_ERROR_NONE; - closure->reason = "starting next batch for send_* op(s)"; - } -} - -// For any pending batch completed in batch_data, adds the necessary -// completion closures to closures, updating *num_closures as needed. -static void add_closures_for_completed_pending_batches( - grpc_call_element* elem, subchannel_batch_data* batch_data, - subchannel_call_retry_state* retry_state, grpc_error* error, - 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); - for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { - 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_INFO, - "chand=%p calld=%p: pending batch completed at index %" PRIuPTR, - chand, calld, i); - } - // Copy the trailing metadata to return it to the surface. - if (batch_data->batch.recv_trailing_metadata) { - grpc_metadata_batch_move(&batch_data->recv_trailing_metadata, - pending->batch->payload->recv_trailing_metadata - .recv_trailing_metadata); - } - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = pending->batch->on_complete; - closure->error = GRPC_ERROR_REF(error); - closure->reason = "on_complete for pending batch"; - pending->batch->on_complete = nullptr; - maybe_clear_pending_batch(elem, pending); - } - } - GRPC_ERROR_UNREF(error); -} - -// For any pending batch containing an op that has not yet been started, -// adds the pending batch's completion closures to closures, updating -// *num_closures as needed. -static void add_closures_to_fail_unstarted_pending_batches( - grpc_call_element* elem, subchannel_call_retry_state* retry_state, - grpc_error* error, 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); - for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) { - 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_INFO, - "chand=%p calld=%p: failing unstarted pending batch at index " - "%" PRIuPTR, - chand, calld, i); - } - if (pending->batch->recv_initial_metadata) { - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = pending->batch->payload->recv_initial_metadata - .recv_initial_metadata_ready; - closure->error = GRPC_ERROR_REF(error); - closure->reason = - "failing recv_initial_metadata_ready for pending batch"; - pending->batch->payload->recv_initial_metadata - .recv_initial_metadata_ready = nullptr; - } - if (pending->batch->recv_message) { - *pending->batch->payload->recv_message.recv_message = nullptr; - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = - pending->batch->payload->recv_message.recv_message_ready; - closure->error = GRPC_ERROR_REF(error); - closure->reason = "failing recv_message_ready for pending batch"; - pending->batch->payload->recv_message.recv_message_ready = nullptr; - } - closure_to_execute* closure = &closures[(*num_closures)++]; - closure->closure = pending->batch->on_complete; - closure->error = GRPC_ERROR_REF(error); - closure->reason = "failing on_complete for pending batch"; - pending->batch->on_complete = nullptr; - maybe_clear_pending_batch(elem, pending); - } + GRPC_CLOSURE_INIT(&batch_data->batch.handler_private.closure, + start_retriable_subchannel_batches, elem, + grpc_schedule_on_exec_ctx); + closures->Add(&batch_data->batch.handler_private.closure, GRPC_ERROR_NONE, + "starting next batch for send_* op(s)"); } - GRPC_ERROR_UNREF(error); } // Callback used to intercept on_complete from subchannel calls. @@ -2035,136 +2042,49 @@ static void on_complete(void* arg, grpc_error* error) { static_cast<subchannel_call_retry_state*>( grpc_connected_subchannel_call_get_parent_data( batch_data->subchannel_call)); - // 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_INFO, "chand=%p calld=%p: call already finished", chand, - calld); - } - } else { - // Check if this batch finished the call, and if so, get its status. - // The call is finished if either (a) this callback was invoked with - // an error or (b) we receive status. - grpc_status_code status = GRPC_STATUS_OK; - grpc_mdelem* server_pushback_md = nullptr; - if (GPR_UNLIKELY(error != GRPC_ERROR_NONE)) { // Case (a). - call_finished = true; - grpc_error_get_status(error, calld->deadline, &status, nullptr, nullptr, - nullptr); - } else if (batch_data->batch.recv_trailing_metadata) { // Case (b). - call_finished = true; - grpc_metadata_batch* md_batch = - batch_data->batch.payload->recv_trailing_metadata - .recv_trailing_metadata; - GPR_ASSERT(md_batch->idx.named.grpc_status != nullptr); - status = grpc_get_status_code_from_metadata( - md_batch->idx.named.grpc_status->md); - if (md_batch->idx.named.grpc_retry_pushback_ms != nullptr) { - server_pushback_md = &md_batch->idx.named.grpc_retry_pushback_ms->md; - } - } - // 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 (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); - } - // Track number of pending subchannel send batches and determine if - // this was the last one. - bool last_callback_complete = false; - if (batch_data->batch.send_initial_metadata || - batch_data->batch.send_message || - batch_data->batch.send_trailing_metadata) { - --calld->num_pending_retriable_subchannel_send_batches; - last_callback_complete = - calld->num_pending_retriable_subchannel_send_batches == 0; - } - batch_data_unref(batch_data); - // If we just completed the last subchannel send batch, unref the - // call stack. - if (last_callback_complete) { - GRPC_CALL_STACK_UNREF(calld->owning_call, "subchannel_send_batches"); - } - return; - } - // Not retrying, so commit the call. - retry_commit(elem, retry_state); - } + if (batch_data->batch.send_initial_metadata) { + retry_state->completed_send_initial_metadata = true; + } + if (batch_data->batch.send_message) { + ++retry_state->completed_send_message_count; + } + if (batch_data->batch.send_trailing_metadata) { + retry_state->completed_send_trailing_metadata = true; } - // 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) { + // If the call is committed, free cached data for send ops that we've just + // completed. + if (calld->retry_committed) { free_cached_send_op_data_for_completed_batch(elem, batch_data, retry_state); } - // Call not being retried. // Construct list of closures to execute. - // Max number of closures is number of pending batches plus one for - // each of: - // - recv_initial_metadata_ready (either deferred or unstarted) - // - recv_message_ready (either deferred or unstarted) - // - starting a new batch for pending send ops - closure_to_execute closures[GPR_ARRAY_SIZE(calld->pending_batches) + 3]; - size_t num_closures = 0; - // If there are deferred recv_initial_metadata_ready or recv_message_ready - // callbacks, add them to closures. - add_closures_for_deferred_recv_callbacks(batch_data, retry_state, closures, - &num_closures); - // Find pending batches whose ops are now complete and add their - // on_complete callbacks to closures. - add_closures_for_completed_pending_batches(elem, batch_data, retry_state, - GRPC_ERROR_REF(error), closures, - &num_closures); - // Add closures to handle any pending batches that have not yet been started. - // If the call is finished, we fail these batches; otherwise, we add a - // callback to start_retriable_subchannel_batches() to start them on - // the subchannel call. - if (call_finished) { - add_closures_to_fail_unstarted_pending_batches( - elem, retry_state, GRPC_ERROR_REF(error), closures, &num_closures); - } else { - add_closures_for_replay_or_pending_send_ops(elem, batch_data, retry_state, - closures, &num_closures); + grpc_core::CallCombinerClosureList closures; + // If a retry was already dispatched, that means we saw + // recv_trailing_metadata before this, so we do nothing here. + // Otherwise, invoke the callback to return the result to the surface. + if (!retry_state->retry_dispatched) { + // Add closure for the completed pending batch, if any. + add_closure_for_completed_pending_batch(elem, batch_data, retry_state, + GRPC_ERROR_REF(error), &closures); + // If needed, add a callback to start any replay or pending send ops on + // the subchannel call. + if (!retry_state->completed_recv_trailing_metadata) { + add_closures_for_replay_or_pending_send_ops(elem, batch_data, retry_state, + &closures); + } } // Track number of pending subchannel send batches and determine if this // was the last one. - bool last_callback_complete = false; - if (batch_data->batch.send_initial_metadata || - batch_data->batch.send_message || - batch_data->batch.send_trailing_metadata) { - --calld->num_pending_retriable_subchannel_send_batches; - last_callback_complete = - calld->num_pending_retriable_subchannel_send_batches == 0; - } + --calld->num_pending_retriable_subchannel_send_batches; + const bool last_send_batch_complete = + calld->num_pending_retriable_subchannel_send_batches == 0; // Don't need batch_data anymore. batch_data_unref(batch_data); // Schedule all of the closures identified above. // Note: This yeilds the call combiner. - execute_closures_in_call_combiner(elem, "on_complete", closures, - num_closures); - // If we just completed the last subchannel send batch, unref the call stack. - if (last_callback_complete) { + closures.RunClosures(calld->call_combiner); + // If this was the last subchannel send batch, unref the call stack. + if (last_send_batch_complete) { GRPC_CALL_STACK_UNREF(calld->owning_call, "subchannel_send_batches"); } } @@ -2185,27 +2105,22 @@ static void start_batch_in_call_combiner(void* arg, grpc_error* ignored) { // 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) { + grpc_call_element* elem, grpc_transport_stream_op_batch* batch, + grpc_core::CallCombinerClosureList* closures) { + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + call_data* calld = static_cast<call_data*>(elem->call_data); 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_log(GPR_INFO, "chand=%p calld=%p: starting subchannel batch: %s", chand, + calld, batch_str); gpr_free(batch_str); - closure->free_reason = true; - } else { - closure->reason = "start_subchannel_batch"; } + closures->Add(&batch->handler_private.closure, GRPC_ERROR_NONE, + "start_subchannel_batch"); } // Adds retriable send_initial_metadata op to batch_data. @@ -2221,28 +2136,28 @@ static void add_retriable_send_initial_metadata_op( // // If we've already completed one or more attempts, add the // grpc-retry-attempts header. - batch_data->send_initial_metadata_storage = + retry_state->send_initial_metadata_storage = static_cast<grpc_linked_mdelem*>(gpr_arena_alloc( calld->arena, sizeof(grpc_linked_mdelem) * (calld->send_initial_metadata.list.count + (calld->num_attempts_completed > 0)))); grpc_metadata_batch_copy(&calld->send_initial_metadata, - &batch_data->send_initial_metadata, - batch_data->send_initial_metadata_storage); - if (GPR_UNLIKELY(batch_data->send_initial_metadata.idx.named + &retry_state->send_initial_metadata, + retry_state->send_initial_metadata_storage); + if (GPR_UNLIKELY(retry_state->send_initial_metadata.idx.named .grpc_previous_rpc_attempts != nullptr)) { - grpc_metadata_batch_remove( - &batch_data->send_initial_metadata, - batch_data->send_initial_metadata.idx.named.grpc_previous_rpc_attempts); + grpc_metadata_batch_remove(&retry_state->send_initial_metadata, + retry_state->send_initial_metadata.idx.named + .grpc_previous_rpc_attempts); } if (GPR_UNLIKELY(calld->num_attempts_completed > 0)) { grpc_mdelem retry_md = grpc_mdelem_from_slices( GRPC_MDSTR_GRPC_PREVIOUS_RPC_ATTEMPTS, *retry_count_strings[calld->num_attempts_completed - 1]); grpc_error* error = grpc_metadata_batch_add_tail( - &batch_data->send_initial_metadata, - &batch_data->send_initial_metadata_storage[calld->send_initial_metadata - .list.count], + &retry_state->send_initial_metadata, + &retry_state->send_initial_metadata_storage[calld->send_initial_metadata + .list.count], retry_md); if (GPR_UNLIKELY(error != GRPC_ERROR_NONE)) { gpr_log(GPR_ERROR, "error adding retry metadata: %s", @@ -2253,7 +2168,7 @@ static void add_retriable_send_initial_metadata_op( retry_state->started_send_initial_metadata = true; batch_data->batch.send_initial_metadata = true; batch_data->batch.payload->send_initial_metadata.send_initial_metadata = - &batch_data->send_initial_metadata; + &retry_state->send_initial_metadata; batch_data->batch.payload->send_initial_metadata.send_initial_metadata_flags = calld->send_initial_metadata_flags; batch_data->batch.payload->send_initial_metadata.peer_string = @@ -2274,10 +2189,10 @@ static void add_retriable_send_message_op( grpc_core::ByteStreamCache* cache = (*calld->send_messages)[retry_state->started_send_message_count]; ++retry_state->started_send_message_count; - batch_data->send_message.Init(cache); + retry_state->send_message.Init(cache); batch_data->batch.send_message = true; batch_data->batch.payload->send_message.send_message.reset( - batch_data->send_message.get()); + retry_state->send_message.get()); } // Adds retriable send_trailing_metadata op to batch_data. @@ -2287,17 +2202,17 @@ static void add_retriable_send_trailing_metadata_op( // We need to make a copy of the metadata batch for each attempt, since // the filters in the subchannel stack may modify this batch, and we don't // want those modifications to be passed forward to subsequent attempts. - batch_data->send_trailing_metadata_storage = + retry_state->send_trailing_metadata_storage = static_cast<grpc_linked_mdelem*>(gpr_arena_alloc( calld->arena, sizeof(grpc_linked_mdelem) * calld->send_trailing_metadata.list.count)); grpc_metadata_batch_copy(&calld->send_trailing_metadata, - &batch_data->send_trailing_metadata, - batch_data->send_trailing_metadata_storage); + &retry_state->send_trailing_metadata, + retry_state->send_trailing_metadata_storage); retry_state->started_send_trailing_metadata = true; batch_data->batch.send_trailing_metadata = true; batch_data->batch.payload->send_trailing_metadata.send_trailing_metadata = - &batch_data->send_trailing_metadata; + &retry_state->send_trailing_metadata; } // Adds retriable recv_initial_metadata op to batch_data. @@ -2306,16 +2221,16 @@ static void add_retriable_recv_initial_metadata_op( subchannel_batch_data* batch_data) { retry_state->started_recv_initial_metadata = true; batch_data->batch.recv_initial_metadata = true; - grpc_metadata_batch_init(&batch_data->recv_initial_metadata); + grpc_metadata_batch_init(&retry_state->recv_initial_metadata); batch_data->batch.payload->recv_initial_metadata.recv_initial_metadata = - &batch_data->recv_initial_metadata; + &retry_state->recv_initial_metadata; batch_data->batch.payload->recv_initial_metadata.trailing_metadata_available = - &batch_data->trailing_metadata_available; - GRPC_CLOSURE_INIT(&batch_data->recv_initial_metadata_ready, + &retry_state->trailing_metadata_available; + GRPC_CLOSURE_INIT(&retry_state->recv_initial_metadata_ready, recv_initial_metadata_ready, batch_data, grpc_schedule_on_exec_ctx); batch_data->batch.payload->recv_initial_metadata.recv_initial_metadata_ready = - &batch_data->recv_initial_metadata_ready; + &retry_state->recv_initial_metadata_ready; } // Adds retriable recv_message op to batch_data. @@ -2325,11 +2240,11 @@ static void add_retriable_recv_message_op( ++retry_state->started_recv_message_count; batch_data->batch.recv_message = true; batch_data->batch.payload->recv_message.recv_message = - &batch_data->recv_message; - GRPC_CLOSURE_INIT(&batch_data->recv_message_ready, recv_message_ready, + &retry_state->recv_message; + GRPC_CLOSURE_INIT(&retry_state->recv_message_ready, recv_message_ready, batch_data, grpc_schedule_on_exec_ctx); batch_data->batch.payload->recv_message.recv_message_ready = - &batch_data->recv_message_ready; + &retry_state->recv_message_ready; } // Adds retriable recv_trailing_metadata op to batch_data. @@ -2338,12 +2253,17 @@ static void add_retriable_recv_trailing_metadata_op( subchannel_batch_data* batch_data) { retry_state->started_recv_trailing_metadata = true; batch_data->batch.recv_trailing_metadata = true; - grpc_metadata_batch_init(&batch_data->recv_trailing_metadata); + grpc_metadata_batch_init(&retry_state->recv_trailing_metadata); batch_data->batch.payload->recv_trailing_metadata.recv_trailing_metadata = - &batch_data->recv_trailing_metadata; - batch_data->batch.collect_stats = true; - batch_data->batch.payload->collect_stats.collect_stats = - &batch_data->collect_stats; + &retry_state->recv_trailing_metadata; + batch_data->batch.payload->recv_trailing_metadata.collect_stats = + &retry_state->collect_stats; + GRPC_CLOSURE_INIT(&retry_state->recv_trailing_metadata_ready, + recv_trailing_metadata_ready, batch_data, + grpc_schedule_on_exec_ctx); + batch_data->batch.payload->recv_trailing_metadata + .recv_trailing_metadata_ready = + &retry_state->recv_trailing_metadata_ready; } // Helper function used to start a recv_trailing_metadata batch. This @@ -2364,9 +2284,11 @@ static void start_internal_recv_trailing_metadata(grpc_call_element* elem) { grpc_connected_subchannel_call_get_parent_data( calld->subchannel_call)); // 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); + // once for the recv_trailing_metadata_ready callback 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, false /* set_on_complete */); 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. @@ -2391,7 +2313,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( "send_initial_metadata op", chand, calld); } - replay_batch_data = batch_data_create(elem, 1); + replay_batch_data = batch_data_create(elem, 1, true /* set_on_complete */); add_retriable_send_initial_metadata_op(calld, retry_state, replay_batch_data); } @@ -2408,7 +2330,8 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( chand, calld); } if (replay_batch_data == nullptr) { - replay_batch_data = batch_data_create(elem, 1); + replay_batch_data = + batch_data_create(elem, 1, true /* set_on_complete */); } add_retriable_send_message_op(elem, retry_state, replay_batch_data); } @@ -2427,7 +2350,8 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay( chand, calld); } if (replay_batch_data == nullptr) { - replay_batch_data = batch_data_create(elem, 1); + replay_batch_data = + batch_data_create(elem, 1, true /* set_on_complete */); } add_retriable_send_trailing_metadata_op(calld, retry_state, replay_batch_data); @@ -2439,7 +2363,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, - closure_to_execute* closures, size_t* num_closures) { + grpc_core::CallCombinerClosureList* 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]; @@ -2493,15 +2417,11 @@ static void add_subchannel_batches_for_pending_batches( // 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"; + closures->Add( + &retry_state->recv_trailing_metadata_ready, GRPC_ERROR_NONE, + "re-executing recv_trailing_metadata_ready to propagate " + "internally triggered result"); } else { batch_data_unref(retry_state->recv_trailing_metadata_internal_batch); } @@ -2513,14 +2433,19 @@ static void add_subchannel_batches_for_pending_batches( if (calld->method_params == nullptr || calld->method_params->retry_policy() == nullptr || calld->retry_committed) { - add_closure_for_subchannel_batch(calld, batch, closures, num_closures); + add_closure_for_subchannel_batch(elem, batch, closures); pending_batch_clear(calld, pending); continue; } // Create batch with the right number of callbacks. - const int num_callbacks = - 1 + batch->recv_initial_metadata + batch->recv_message; - subchannel_batch_data* batch_data = batch_data_create(elem, num_callbacks); + const bool has_send_ops = batch->send_initial_metadata || + batch->send_message || + batch->send_trailing_metadata; + const int num_callbacks = has_send_ops + batch->recv_initial_metadata + + batch->recv_message + + batch->recv_trailing_metadata; + subchannel_batch_data* batch_data = batch_data_create( + elem, num_callbacks, has_send_ops /* set_on_complete */); // Cache send ops if needed. maybe_cache_send_ops_for_batch(calld, pending); // send_initial_metadata. @@ -2547,11 +2472,9 @@ static void add_subchannel_batches_for_pending_batches( } // recv_trailing_metadata. if (batch->recv_trailing_metadata) { - GPR_ASSERT(batch->collect_stats); add_retriable_recv_trailing_metadata_op(calld, retry_state, batch_data); } - add_closure_for_subchannel_batch(calld, &batch_data->batch, closures, - num_closures); + add_closure_for_subchannel_batch(elem, &batch_data->batch, closures); // Track number of pending subchannel send batches. // If this is the first one, take a ref to the call stack. if (batch->send_initial_metadata || batch->send_message || @@ -2579,15 +2502,13 @@ static void start_retriable_subchannel_batches(void* arg, grpc_error* ignored) { 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. - closure_to_execute closures[GPR_ARRAY_SIZE(calld->pending_batches)]; - size_t num_closures = 0; + grpc_core::CallCombinerClosureList closures; // 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) { - add_closure_for_subchannel_batch(calld, &replay_batch_data->batch, closures, - &num_closures); + add_closure_for_subchannel_batch(elem, &replay_batch_data->batch, + &closures); // Track number of pending subchannel send batches. // If this is the first one, take a ref to the call stack. if (calld->num_pending_retriable_subchannel_send_batches == 0) { @@ -2596,17 +2517,16 @@ static void start_retriable_subchannel_batches(void* arg, grpc_error* ignored) { ++calld->num_pending_retriable_subchannel_send_batches; } // Now add pending batches. - add_subchannel_batches_for_pending_batches(elem, retry_state, closures, - &num_closures); + add_subchannel_batches_for_pending_batches(elem, retry_state, &closures); // Start batches on subchannel call. if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, "chand=%p calld=%p: starting %" PRIuPTR " retriable batches on subchannel_call=%p", - chand, calld, num_closures, calld->subchannel_call); + chand, calld, closures.size(), calld->subchannel_call); } - execute_closures_in_call_combiner(elem, "start_retriable_subchannel_batches", - closures, num_closures); + // Note: This will yield the call combiner. + closures.RunClosures(calld->call_combiner); } // diff --git a/src/core/ext/filters/deadline/deadline_filter.cc b/src/core/ext/filters/deadline/deadline_filter.cc index e0a41a3637..d23ad67ad5 100644 --- a/src/core/ext/filters/deadline/deadline_filter.cc +++ b/src/core/ext/filters/deadline/deadline_filter.cc @@ -128,21 +128,25 @@ static void cancel_timer_if_needed(grpc_deadline_state* deadline_state) { } } -// Callback run when the call is complete. -static void on_complete(void* arg, grpc_error* error) { +// Callback run when we receive trailing metadata. +static void recv_trailing_metadata_ready(void* arg, grpc_error* error) { grpc_deadline_state* deadline_state = static_cast<grpc_deadline_state*>(arg); cancel_timer_if_needed(deadline_state); - // Invoke the next callback. - GRPC_CLOSURE_RUN(deadline_state->next_on_complete, GRPC_ERROR_REF(error)); + // Invoke the original callback. + GRPC_CLOSURE_RUN(deadline_state->original_recv_trailing_metadata_ready, + GRPC_ERROR_REF(error)); } -// Inject our own on_complete callback into op. -static void inject_on_complete_cb(grpc_deadline_state* deadline_state, - grpc_transport_stream_op_batch* op) { - deadline_state->next_on_complete = op->on_complete; - GRPC_CLOSURE_INIT(&deadline_state->on_complete, on_complete, deadline_state, +// Inject our own recv_trailing_metadata_ready callback into op. +static void inject_recv_trailing_metadata_ready( + grpc_deadline_state* deadline_state, grpc_transport_stream_op_batch* op) { + deadline_state->original_recv_trailing_metadata_ready = + op->payload->recv_trailing_metadata.recv_trailing_metadata_ready; + GRPC_CLOSURE_INIT(&deadline_state->recv_trailing_metadata_ready, + recv_trailing_metadata_ready, deadline_state, grpc_schedule_on_exec_ctx); - op->on_complete = &deadline_state->on_complete; + op->payload->recv_trailing_metadata.recv_trailing_metadata_ready = + &deadline_state->recv_trailing_metadata_ready; } // Callback and associated state for starting the timer after call stack @@ -226,7 +230,7 @@ void grpc_deadline_state_client_start_transport_stream_op_batch( // Make sure we know when the call is complete, so that we can cancel // the timer. if (op->recv_trailing_metadata) { - inject_on_complete_cb(deadline_state, op); + inject_recv_trailing_metadata_ready(deadline_state, op); } } } @@ -322,7 +326,7 @@ static void server_start_transport_stream_op_batch( // the client never sends trailing metadata, because this is the // hook that tells us when the call is complete on the server side. if (op->recv_trailing_metadata) { - inject_on_complete_cb(&calld->base.deadline_state, op); + inject_recv_trailing_metadata_ready(&calld->base.deadline_state, op); } } // Chain to next filter. diff --git a/src/core/ext/filters/deadline/deadline_filter.h b/src/core/ext/filters/deadline/deadline_filter.h index 13207cbd6f..1d797f445a 100644 --- a/src/core/ext/filters/deadline/deadline_filter.h +++ b/src/core/ext/filters/deadline/deadline_filter.h @@ -37,12 +37,12 @@ typedef struct grpc_deadline_state { grpc_deadline_timer_state timer_state; grpc_timer timer; grpc_closure timer_callback; - // Closure to invoke when the call is complete. + // Closure to invoke when we receive trailing metadata. // We use this to cancel the timer. - grpc_closure on_complete; - // The original on_complete closure, which we chain to after our own - // closure is invoked. - grpc_closure* next_on_complete; + grpc_closure recv_trailing_metadata_ready; + // The original recv_trailing_metadata_ready closure, which we chain to + // after our own closure is invoked. + grpc_closure* original_recv_trailing_metadata_ready; } grpc_deadline_state; // diff --git a/src/core/ext/filters/http/client/http_client_filter.cc b/src/core/ext/filters/http/client/http_client_filter.cc index ae94ce47b9..1678051beb 100644 --- a/src/core/ext/filters/http/client/http_client_filter.cc +++ b/src/core/ext/filters/http/client/http_client_filter.cc @@ -55,8 +55,8 @@ struct call_data { grpc_closure recv_initial_metadata_ready; // State for handling recv_trailing_metadata ops. grpc_metadata_batch* recv_trailing_metadata; - grpc_closure* original_recv_trailing_metadata_on_complete; - grpc_closure recv_trailing_metadata_on_complete; + grpc_closure* original_recv_trailing_metadata_ready; + grpc_closure recv_trailing_metadata_ready; // State for handling send_message ops. grpc_transport_stream_op_batch* send_message_batch; size_t send_message_bytes_read; @@ -153,8 +153,7 @@ static void recv_initial_metadata_ready(void* user_data, grpc_error* error) { GRPC_CLOSURE_RUN(calld->original_recv_initial_metadata_ready, error); } -static void recv_trailing_metadata_on_complete(void* user_data, - grpc_error* error) { +static void recv_trailing_metadata_ready(void* user_data, grpc_error* error) { grpc_call_element* elem = static_cast<grpc_call_element*>(user_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (error == GRPC_ERROR_NONE) { @@ -163,7 +162,7 @@ static void recv_trailing_metadata_on_complete(void* user_data, } else { GRPC_ERROR_REF(error); } - GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_on_complete, error); + GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_ready, error); } static void send_message_on_complete(void* arg, grpc_error* error) { @@ -312,8 +311,10 @@ static void hc_start_transport_stream_op_batch( /* substitute our callback for the higher callback */ calld->recv_trailing_metadata = batch->payload->recv_trailing_metadata.recv_trailing_metadata; - calld->original_recv_trailing_metadata_on_complete = batch->on_complete; - batch->on_complete = &calld->recv_trailing_metadata_on_complete; + calld->original_recv_trailing_metadata_ready = + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready; + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready = + &calld->recv_trailing_metadata_ready; } grpc_error* error = GRPC_ERROR_NONE; @@ -420,8 +421,8 @@ static grpc_error* init_call_elem(grpc_call_element* elem, GRPC_CLOSURE_INIT(&calld->recv_initial_metadata_ready, recv_initial_metadata_ready, elem, grpc_schedule_on_exec_ctx); - GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_on_complete, - recv_trailing_metadata_on_complete, elem, + GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready, + recv_trailing_metadata_ready, elem, grpc_schedule_on_exec_ctx); GRPC_CLOSURE_INIT(&calld->send_message_on_complete, send_message_on_complete, elem, grpc_schedule_on_exec_ctx); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index a8090d18a6..0d6b72c66e 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -1149,12 +1149,10 @@ static void maybe_start_some_streams(grpc_chttp2_transport* t) { } } -/* Flag that this closure barrier wants stats to be updated before finishing */ -#define CLOSURE_BARRIER_STATS_BIT (1 << 0) /* Flag that this closure barrier may be covering a write in a pollset, and so we should not complete this closure until we can prove that the write got scheduled */ -#define CLOSURE_BARRIER_MAY_COVER_WRITE (1 << 1) +#define CLOSURE_BARRIER_MAY_COVER_WRITE (1 << 0) /* First bit of the reference count, stored in the high order bits (with the low bits being used for flags defined above) */ #define CLOSURE_BARRIER_FIRST_REF_BIT (1 << 16) @@ -1206,10 +1204,6 @@ void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t, grpc_error_add_child(closure->error_data.error, error); } if (closure->next_data.scratch < CLOSURE_BARRIER_FIRST_REF_BIT) { - if (closure->next_data.scratch & CLOSURE_BARRIER_STATS_BIT) { - grpc_transport_move_stats(&s->stats, s->collecting_stats); - s->collecting_stats = nullptr; - } if ((t->write_state == GRPC_CHTTP2_WRITE_STATE_IDLE) || !(closure->next_data.scratch & CLOSURE_BARRIER_MAY_COVER_WRITE)) { GRPC_CLOSURE_RUN(closure, closure->error_data.error); @@ -1351,9 +1345,14 @@ static void perform_stream_op_locked(void* stream_op, } grpc_closure* on_complete = op->on_complete; + // TODO(roth): This is a hack needed because we use data inside of the + // closure itself to do the barrier calculation (i.e., to ensure that + // we don't schedule the closure until all ops in the batch have been + // completed). This can go away once we move to a new C++ closure API + // that provides the ability to create a barrier closure. if (on_complete == nullptr) { - on_complete = - GRPC_CLOSURE_CREATE(do_nothing, nullptr, grpc_schedule_on_exec_ctx); + on_complete = GRPC_CLOSURE_INIT(&op->handler_private.closure, do_nothing, + nullptr, grpc_schedule_on_exec_ctx); } /* use final_data as a barrier until enqueue time; the inital counter is @@ -1361,12 +1360,6 @@ static void perform_stream_op_locked(void* stream_op, on_complete->next_data.scratch = CLOSURE_BARRIER_FIRST_REF_BIT; on_complete->error_data.error = GRPC_ERROR_NONE; - if (op->collect_stats) { - GPR_ASSERT(s->collecting_stats == nullptr); - s->collecting_stats = op_payload->collect_stats.collect_stats; - on_complete->next_data.scratch |= CLOSURE_BARRIER_STATS_BIT; - } - if (op->cancel_stream) { GRPC_STATS_INC_HTTP2_OP_CANCEL(); grpc_chttp2_cancel_stream(t, s, op_payload->cancel_stream.cancel_error); @@ -1600,8 +1593,11 @@ static void perform_stream_op_locked(void* stream_op, if (op->recv_trailing_metadata) { GRPC_STATS_INC_HTTP2_OP_RECV_TRAILING_METADATA(); + GPR_ASSERT(s->collecting_stats == nullptr); + s->collecting_stats = op_payload->recv_trailing_metadata.collect_stats; GPR_ASSERT(s->recv_trailing_metadata_finished == nullptr); - s->recv_trailing_metadata_finished = add_closure_barrier(on_complete); + s->recv_trailing_metadata_finished = + op_payload->recv_trailing_metadata.recv_trailing_metadata_ready; s->recv_trailing_metadata = op_payload->recv_trailing_metadata.recv_trailing_metadata; s->final_metadata_requested = true; @@ -1960,11 +1956,12 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_chttp2_transport* t, } if (s->read_closed && s->frame_storage.length == 0 && !pending_data && s->recv_trailing_metadata_finished != nullptr) { + grpc_transport_move_stats(&s->stats, s->collecting_stats); + s->collecting_stats = nullptr; grpc_chttp2_incoming_metadata_buffer_publish(&s->metadata_buffer[1], s->recv_trailing_metadata); - grpc_chttp2_complete_closure_step( - t, s, &s->recv_trailing_metadata_finished, GRPC_ERROR_NONE, - "recv_trailing_metadata_finished"); + null_then_run_closure(&s->recv_trailing_metadata_finished, + GRPC_ERROR_NONE); } } } diff --git a/src/core/ext/transport/chttp2/transport/writing.cc b/src/core/ext/transport/chttp2/transport/writing.cc index 34d5e6e218..8b73b01dea 100644 --- a/src/core/ext/transport/chttp2/transport/writing.cc +++ b/src/core/ext/transport/chttp2/transport/writing.cc @@ -139,22 +139,27 @@ static bool update_list(grpc_chttp2_transport* t, grpc_chttp2_stream* s, static void report_stall(grpc_chttp2_transport* t, grpc_chttp2_stream* s, const char* staller) { - gpr_log( - GPR_DEBUG, - "%s:%p stream %d stalled by %s [fc:pending=%" PRIdPTR - ":pending-compressed=%" PRIdPTR ":flowed=%" PRId64 - ":peer_initwin=%d:t_win=%" PRId64 ":s_win=%d:s_delta=%" PRId64 "]", - t->peer_string, t, s->id, staller, s->flow_controlled_buffer.length, - s->compressed_data_buffer.length, s->flow_controlled_bytes_flowed, - t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], - t->flow_control->remote_window(), - static_cast<uint32_t> GPR_MAX( - 0, - s->flow_control->remote_window_delta() + - (int64_t)t->settings[GRPC_PEER_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]), - s->flow_control->remote_window_delta()); + if (grpc_flowctl_trace.enabled()) { + gpr_log( + GPR_DEBUG, + "%s:%p stream %d moved to stalled list by %s. This is FULLY expected " + "to happen in a healthy program that is not seeing flow control stalls." + " However, if you know that there are unwanted stalls, here is some " + "helpful data: [fc:pending=%" PRIdPTR ":pending-compressed=%" PRIdPTR + ":flowed=%" PRId64 ":peer_initwin=%d:t_win=%" PRId64 + ":s_win=%d:s_delta=%" PRId64 "]", + t->peer_string, t, s->id, staller, s->flow_controlled_buffer.length, + s->compressed_data_buffer.length, s->flow_controlled_bytes_flowed, + t->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], + t->flow_control->remote_window(), + static_cast<uint32_t> GPR_MAX( + 0, + s->flow_control->remote_window_delta() + + (int64_t)t->settings[GRPC_PEER_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]), + s->flow_control->remote_window_delta()); + } } static bool stream_ref_if_not_destroyed(gpr_refcount* r) { diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.cc b/src/core/ext/transport/cronet/transport/cronet_transport.cc index 420c2d13e1..4a252d972d 100644 --- a/src/core/ext/transport/cronet/transport/cronet_transport.cc +++ b/src/core/ext/transport/cronet/transport/cronet_transport.cc @@ -925,6 +925,10 @@ static bool op_can_be_run(grpc_transport_stream_op_batch* curr_op, result = false; } /* Check if every op that was asked for is done. */ + /* TODO(muxi): We should not consider the recv ops here, since they + * have their own callbacks. We should invoke a batch's on_complete + * as soon as all of the batch's send ops are complete, even if + * there are still recv ops pending. */ else if (curr_op->send_initial_metadata && !stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) { CRONET_LOG(GPR_DEBUG, "Because"); @@ -1280,12 +1284,20 @@ static enum e_op_result execute_stream_op(struct op_and_state* oas) { op_can_be_run(stream_op, s, &oas->state, OP_RECV_TRAILING_METADATA)) { CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_TRAILING_METADATA", oas); - if (oas->s->state.rs.trailing_metadata_valid) { + grpc_error* error = GRPC_ERROR_NONE; + if (stream_state->state_op_done[OP_CANCEL_ERROR]) { + error = GRPC_ERROR_REF(stream_state->cancel_error); + } else if (stream_state->state_op_done[OP_FAILED]) { + error = make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable."); + } else if (oas->s->state.rs.trailing_metadata_valid) { grpc_chttp2_incoming_metadata_buffer_publish( &oas->s->state.rs.trailing_metadata, stream_op->payload->recv_trailing_metadata.recv_trailing_metadata); stream_state->rs.trailing_metadata_valid = false; } + GRPC_CLOSURE_SCHED( + stream_op->payload->recv_trailing_metadata.recv_trailing_metadata_ready, + error); stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true; result = ACTION_TAKEN_NO_CALLBACK; } else if (stream_op->cancel_stream && @@ -1398,6 +1410,11 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready, GRPC_ERROR_CANCELLED); } + if (op->recv_trailing_metadata) { + GRPC_CLOSURE_SCHED( + op->payload->recv_trailing_metadata.recv_trailing_metadata_ready, + GRPC_ERROR_CANCELLED); + } GRPC_CLOSURE_SCHED(op->on_complete, GRPC_ERROR_CANCELLED); return; } diff --git a/src/core/ext/transport/inproc/inproc_transport.cc b/src/core/ext/transport/inproc/inproc_transport.cc index 2c3bff5c1e..b0ca7f8207 100644 --- a/src/core/ext/transport/inproc/inproc_transport.cc +++ b/src/core/ext/transport/inproc/inproc_transport.cc @@ -120,7 +120,6 @@ typedef struct inproc_stream { struct inproc_stream* stream_list_next; } inproc_stream; -static grpc_closure do_nothing_closure; static bool cancel_stream_locked(inproc_stream* s, grpc_error* error); static void op_state_machine(void* arg, grpc_error* error); @@ -373,6 +372,10 @@ static void complete_if_batch_end_locked(inproc_stream* s, grpc_error* error, const char* msg) { int is_sm = static_cast<int>(op == s->send_message_op); int is_stm = static_cast<int>(op == s->send_trailing_md_op); + // TODO(vjpai): We should not consider the recv ops here, since they + // have their own callbacks. We should invoke a batch's on_complete + // as soon as all of the batch's send ops are complete, even if there + // are still recv ops pending. int is_rim = static_cast<int>(op == s->recv_initial_md_op); int is_rm = static_cast<int>(op == s->recv_message_op); int is_rtm = static_cast<int>(op == s->recv_trailing_md_op); @@ -496,6 +499,11 @@ 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_INFO, "fail_helper %p scheduling trailing-metadata-ready %p", + s, error); + GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + GRPC_ERROR_REF(error)); INPROC_LOG(GPR_INFO, "fail_helper %p scheduling trailing-md-on-complete %p", s, error); complete_if_batch_end_locked( @@ -639,6 +647,12 @@ 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_INFO, + "op_state_machine %p scheduling trailing-metadata-ready", s); + GRPC_CLOSURE_SCHED( + s->recv_trailing_md_op->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + GRPC_ERROR_NONE); + 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); @@ -711,6 +725,12 @@ static void op_state_machine(void* arg, grpc_error* error) { } if (s->recv_trailing_md_op && s->t->is_client && other && other->send_message_op) { + INPROC_LOG(GPR_INFO, + "op_state_machine %p scheduling trailing-metadata-ready %p", s, + GRPC_ERROR_NONE); + GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + GRPC_ERROR_NONE); maybe_schedule_op_closure_locked(other, GRPC_ERROR_NONE); } if (s->to_read_trailing_md_filled) { @@ -766,6 +786,10 @@ static void op_state_machine(void* arg, grpc_error* error) { 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->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + GRPC_ERROR_REF(new_err)); GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->on_complete, GRPC_ERROR_REF(new_err)); s->recv_trailing_md_op = nullptr; @@ -859,6 +883,9 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) { // couldn't complete that because we hadn't yet sent out trailing // md, now's the chance if (!s->t->is_client && s->trailing_md_recvd && s->recv_trailing_md_op) { + GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata + .recv_trailing_metadata_ready, + GRPC_ERROR_REF(s->cancel_self_error)); complete_if_batch_end_locked( s, s->cancel_self_error, s->recv_trailing_md_op, "cancel_stream scheduling trailing-md-on-complete"); @@ -873,6 +900,8 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) { return ret; } +static void do_nothing(void* arg, grpc_error* error) {} + static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, grpc_transport_stream_op_batch* op) { INPROC_LOG(GPR_INFO, "perform_stream_op %p %p %p", gt, gs, op); @@ -892,8 +921,14 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, } grpc_error* error = GRPC_ERROR_NONE; grpc_closure* on_complete = op->on_complete; + // TODO(roth): This is a hack needed because we use data inside of the + // closure itself to do the barrier calculation (i.e., to ensure that + // we don't schedule the closure until all ops in the batch have been + // completed). This can go away once we move to a new C++ closure API + // that provides the ability to create a barrier closure. if (on_complete == nullptr) { - on_complete = &do_nothing_closure; + on_complete = GRPC_CLOSURE_INIT(&op->handler_private.closure, do_nothing, + nullptr, grpc_schedule_on_exec_ctx); } if (op->cancel_stream) { @@ -1026,6 +1061,15 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready, GRPC_ERROR_REF(error)); } + if (op->recv_trailing_metadata) { + INPROC_LOG( + GPR_INFO, + "perform_stream_op error %p scheduling trailing-metadata-ready %p", + s, error); + GRPC_CLOSURE_SCHED( + op->payload->recv_trailing_metadata.recv_trailing_metadata_ready, + GRPC_ERROR_REF(error)); + } } INPROC_LOG(GPR_INFO, "perform_stream_op %p scheduling on_complete %p", s, error); @@ -1129,12 +1173,8 @@ static grpc_endpoint* get_endpoint(grpc_transport* t) { return nullptr; } /******************************************************************************* * GLOBAL INIT AND DESTROY */ -static void do_nothing(void* arg, grpc_error* error) {} - void grpc_inproc_transport_init(void) { grpc_core::ExecCtx exec_ctx; - GRPC_CLOSURE_INIT(&do_nothing_closure, do_nothing, nullptr, - grpc_schedule_on_exec_ctx); g_empty_slice = grpc_slice_from_static_buffer(nullptr, 0); grpc_slice key_tmp = grpc_slice_from_static_string(":path"); diff --git a/src/core/lib/channel/channel_trace.cc b/src/core/lib/channel/channel_trace.cc index eb7214b355..0f655d8716 100644 --- a/src/core/lib/channel/channel_trace.cc +++ b/src/core/lib/channel/channel_trace.cc @@ -28,7 +28,6 @@ #include <stdlib.h> #include <string.h> -#include "src/core/lib/channel/channelz_registry.h" #include "src/core/lib/channel/status_util.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/gpr/useful.h" @@ -40,16 +39,17 @@ #include "src/core/lib/transport/error_utils.h" namespace grpc_core { +namespace channelz { ChannelTrace::TraceEvent::TraceEvent( Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer, ReferencedType type) + RefCountedPtr<ChannelNode> referenced_channel, ReferencedType type) : severity_(severity), data_(data), timestamp_(grpc_millis_to_timespec(grpc_core::ExecCtx::Get()->Now(), GPR_CLOCK_REALTIME)), next_(nullptr), - referenced_tracer_(std::move(referenced_tracer)), + referenced_channel_(std::move(referenced_channel)), referenced_type_(type) {} ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data) @@ -62,15 +62,13 @@ ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data) ChannelTrace::TraceEvent::~TraceEvent() { grpc_slice_unref_internal(data_); } ChannelTrace::ChannelTrace(size_t max_events) - : channel_uuid_(-1), - num_events_logged_(0), + : num_events_logged_(0), list_size_(0), max_list_size_(max_events), head_trace_(nullptr), tail_trace_(nullptr) { if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0 gpr_mu_init(&tracer_mu_); - channel_uuid_ = ChannelzRegistry::Register(this); time_created_ = grpc_millis_to_timespec(grpc_core::ExecCtx::Get()->Now(), GPR_CLOCK_REALTIME); } @@ -83,12 +81,9 @@ ChannelTrace::~ChannelTrace() { it = it->next(); Delete<TraceEvent>(to_free); } - ChannelzRegistry::Unregister(channel_uuid_); gpr_mu_destroy(&tracer_mu_); } -intptr_t ChannelTrace::GetUuid() const { return channel_uuid_; } - void ChannelTrace::AddTraceEventHelper(TraceEvent* new_trace_event) { ++num_events_logged_; // first event case @@ -117,20 +112,21 @@ void ChannelTrace::AddTraceEvent(Severity severity, grpc_slice data) { void ChannelTrace::AddTraceEventReferencingChannel( Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer) { + RefCountedPtr<ChannelNode> referenced_channel) { if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0 // create and fill up the new event - AddTraceEventHelper( - New<TraceEvent>(severity, data, std::move(referenced_tracer), Channel)); + AddTraceEventHelper(New<TraceEvent>( + severity, data, std::move(referenced_channel), ReferencedType::Channel)); } void ChannelTrace::AddTraceEventReferencingSubchannel( Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer) { + RefCountedPtr<ChannelNode> referenced_subchannel) { if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0 // create and fill up the new event - AddTraceEventHelper(New<TraceEvent>( - severity, data, std::move(referenced_tracer), Subchannel)); + AddTraceEventHelper(New<TraceEvent>(severity, data, + std::move(referenced_subchannel), + ReferencedType::Subchannel)); } namespace { @@ -193,22 +189,24 @@ void ChannelTrace::TraceEvent::RenderTraceEvent(grpc_json* json) const { json_iterator = grpc_json_create_child(json_iterator, json, "timestamp", fmt_time(timestamp_), GRPC_JSON_STRING, true); - if (referenced_tracer_ != nullptr) { + if (referenced_channel_ != nullptr) { char* uuid_str; - gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_tracer_->channel_uuid_); + gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_channel_->channel_uuid()); grpc_json* child_ref = grpc_json_create_child( json_iterator, json, - (referenced_type_ == Channel) ? "channelRef" : "subchannelRef", nullptr, - GRPC_JSON_OBJECT, false); + (referenced_type_ == ReferencedType::Channel) ? "channelRef" + : "subchannelRef", + nullptr, GRPC_JSON_OBJECT, false); json_iterator = grpc_json_create_child( nullptr, child_ref, - (referenced_type_ == Channel) ? "channelId" : "subchannelId", uuid_str, - GRPC_JSON_STRING, true); + (referenced_type_ == ReferencedType::Channel) ? "channelId" + : "subchannelId", + uuid_str, GRPC_JSON_STRING, true); json_iterator = child_ref; } } -char* ChannelTrace::RenderTrace() const { +grpc_json* ChannelTrace::RenderJSON() const { if (!max_list_size_) return nullptr; // tracing is disabled if max_events == 0 grpc_json* json = grpc_json_create(GRPC_JSON_OBJECT); @@ -219,7 +217,7 @@ char* ChannelTrace::RenderTrace() const { grpc_json_create_child(json_iterator, json, "numEventsLogged", num_events_logged_str, GRPC_JSON_STRING, true); json_iterator = - grpc_json_create_child(json_iterator, json, "creationTime", + grpc_json_create_child(json_iterator, json, "creationTimestamp", fmt_time(time_created_), GRPC_JSON_STRING, true); grpc_json* events = grpc_json_create_child(json_iterator, json, "events", nullptr, GRPC_JSON_ARRAY, false); @@ -231,9 +229,8 @@ char* ChannelTrace::RenderTrace() const { it->RenderTraceEvent(json_iterator); it = it->next(); } - char* json_str = grpc_json_dump_to_string(json, 0); - grpc_json_destroy(json); - return json_str; + return json; } +} // namespace channelz } // namespace grpc_core diff --git a/src/core/lib/channel/channel_trace.h b/src/core/lib/channel/channel_trace.h index 1df1e585f2..0dd162a777 100644 --- a/src/core/lib/channel/channel_trace.h +++ b/src/core/lib/channel/channel_trace.h @@ -28,18 +28,18 @@ #include "src/core/lib/json/json.h" namespace grpc_core { +namespace channelz { + +class ChannelNode; // Object used to hold live data for a channel. This data is exposed via the // channelz service: // https://github.com/grpc/proposal/blob/master/A14-channelz.md -class ChannelTrace : public RefCounted<ChannelTrace> { +class ChannelTrace { public: ChannelTrace(size_t max_events); ~ChannelTrace(); - // returns the tracer's uuid - intptr_t GetUuid() const; - enum Severity { Unset = 0, // never to be used Info, // we start at 1 to avoid using proto default values @@ -59,34 +59,30 @@ class ChannelTrace : public RefCounted<ChannelTrace> { // created a new subchannel, then it would record that with a TraceEvent // referencing the new subchannel. // - // TODO(ncteisen): Once channelz is implemented, the events should reference - // the overall channelz object, not just the ChannelTrace object. // TODO(ncteisen): as this call is used more and more throughout the gRPC // stack, determine if it makes more sense to accept a char* instead of a // slice. void AddTraceEventReferencingChannel( Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer); + RefCountedPtr<ChannelNode> referenced_channel); void AddTraceEventReferencingSubchannel( Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer); + RefCountedPtr<ChannelNode> referenced_subchannel); - // Returns the tracing data rendered as a grpc json string. - // The string is owned by the caller and must be freed. - char* RenderTrace() const; + // Creates and returns the raw grpc_json object, so a parent channelz + // object may incorporate the json before rendering. + grpc_json* RenderJSON() const; private: // Types of objects that can be references by trace events. - enum ReferencedType { Channel, Subchannel }; + enum class ReferencedType { Channel, Subchannel }; // Private class to encapsulate all the data and bookkeeping needed for a // a trace event. class TraceEvent { public: // Constructor for a TraceEvent that references a different channel. - // TODO(ncteisen): once channelz is implemented, this should reference the - // overall channelz object, not just the ChannelTrace object TraceEvent(Severity severity, grpc_slice data, - RefCountedPtr<ChannelTrace> referenced_tracer, + RefCountedPtr<ChannelNode> referenced_channel, ReferencedType type); // Constructor for a TraceEvent that does not reverence a different @@ -109,7 +105,7 @@ class ChannelTrace : public RefCounted<ChannelTrace> { gpr_timespec timestamp_; TraceEvent* next_; // the tracer object for the (sub)channel that this trace event refers to. - RefCountedPtr<ChannelTrace> referenced_tracer_; + RefCountedPtr<ChannelNode> referenced_channel_; // the type that the referenced tracer points to. Unused if this trace // does not point to any channel or subchannel ReferencedType referenced_type_; @@ -119,7 +115,6 @@ class ChannelTrace : public RefCounted<ChannelTrace> { void AddTraceEventHelper(TraceEvent* new_trace_event); gpr_mu tracer_mu_; - intptr_t channel_uuid_; uint64_t num_events_logged_; size_t list_size_; size_t max_list_size_; @@ -128,6 +123,7 @@ class ChannelTrace : public RefCounted<ChannelTrace> { gpr_timespec time_created_; }; +} // namespace channelz } // namespace grpc_core #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_TRACE_H */ diff --git a/src/core/lib/channel/channelz.cc b/src/core/lib/channel/channelz.cc new file mode 100644 index 0000000000..3550fc0551 --- /dev/null +++ b/src/core/lib/channel/channelz.cc @@ -0,0 +1,185 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include <grpc/impl/codegen/port_platform.h> + +#include "src/core/lib/channel/channelz.h" + +#include <grpc/grpc.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> + +#include "src/core/lib/channel/channelz_registry.h" +#include "src/core/lib/channel/status_util.h" +#include "src/core/lib/gpr/string.h" +#include "src/core/lib/gpr/useful.h" +#include "src/core/lib/gprpp/memory.h" +#include "src/core/lib/iomgr/error.h" +#include "src/core/lib/slice/slice_internal.h" +#include "src/core/lib/surface/channel.h" +#include "src/core/lib/transport/connectivity_state.h" +#include "src/core/lib/transport/error_utils.h" + +namespace grpc_core { +namespace channelz { + +namespace { + +// TODO(ncteisen): move this function to a common helper location. +// +// returns an allocated string that represents tm according to RFC-3339, and, +// more specifically, follows: +// https://developers.google.com/protocol-buffers/docs/proto3#json +// +// "Uses RFC 3339, where generated output will always be Z-normalized and uses +// 0, 3, 6 or 9 fractional digits." +char* fmt_time(gpr_timespec tm) { + char time_buffer[35]; + char ns_buffer[11]; // '.' + 9 digits of precision + struct tm* tm_info = localtime((const time_t*)&tm.tv_sec); + strftime(time_buffer, sizeof(time_buffer), "%Y-%m-%dT%H:%M:%S", tm_info); + snprintf(ns_buffer, 11, ".%09d", tm.tv_nsec); + // This loop trims off trailing zeros by inserting a null character that the + // right point. We iterate in chunks of three because we want 0, 3, 6, or 9 + // fractional digits. + for (int i = 7; i >= 1; i -= 3) { + if (ns_buffer[i] == '0' && ns_buffer[i + 1] == '0' && + ns_buffer[i + 2] == '0') { + ns_buffer[i] = '\0'; + // Edge case in which all fractional digits were 0. + if (i == 1) { + ns_buffer[0] = '\0'; + } + } else { + break; + } + } + char* full_time_str; + gpr_asprintf(&full_time_str, "%s%sZ", time_buffer, ns_buffer); + return full_time_str; +} + +// TODO(ncteisen); move this to json library +grpc_json* add_num_str(grpc_json* parent, grpc_json* it, const char* name, + int64_t num) { + char* num_str; + gpr_asprintf(&num_str, "%" PRId64, num); + return grpc_json_create_child(it, parent, name, num_str, GRPC_JSON_STRING, + true); +} + +} // namespace + +ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes) + : channel_(channel), target_(nullptr), channel_uuid_(-1) { + trace_.Init(channel_tracer_max_nodes); + target_ = UniquePtr<char>(grpc_channel_get_target(channel_)); + channel_uuid_ = ChannelzRegistry::Register(this); + gpr_atm_no_barrier_store(&last_call_started_millis_, + (gpr_atm)ExecCtx::Get()->Now()); +} + +ChannelNode::~ChannelNode() { + trace_.Destroy(); + ChannelzRegistry::Unregister(channel_uuid_); +} + +void ChannelNode::RecordCallStarted() { + gpr_atm_no_barrier_fetch_add(&calls_started_, (gpr_atm)1); + gpr_atm_no_barrier_store(&last_call_started_millis_, + (gpr_atm)ExecCtx::Get()->Now()); +} + +grpc_connectivity_state ChannelNode::GetConnectivityState() { + if (channel_ == nullptr) { + return GRPC_CHANNEL_SHUTDOWN; + } else { + return grpc_channel_check_connectivity_state(channel_, false); + } +} + +char* ChannelNode::RenderJSON() { + // We need to track these three json objects to build our object + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + // create and fill the ref child + json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr, + GRPC_JSON_OBJECT, false); + json = json_iterator; + json_iterator = nullptr; + json_iterator = add_num_str(json, json_iterator, "channelId", channel_uuid_); + // reset json iterators to top level object + json = top_level_json; + json_iterator = nullptr; + // create and fill the data child. + grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr, + GRPC_JSON_OBJECT, false); + json = data; + json_iterator = nullptr; + // create and fill the connectivity state child. + grpc_connectivity_state connectivity_state = GetConnectivityState(); + json_iterator = grpc_json_create_child(json_iterator, json, "state", nullptr, + GRPC_JSON_OBJECT, false); + json = json_iterator; + grpc_json_create_child(nullptr, json, "state", + grpc_connectivity_state_name(connectivity_state), + GRPC_JSON_STRING, false); + // reset the parent to be the data object. + json = data; + json_iterator = grpc_json_create_child( + json_iterator, json, "target", target_.get(), GRPC_JSON_STRING, false); + // fill in the channel trace if applicable + grpc_json* trace = trace_->RenderJSON(); + if (trace != nullptr) { + // we manuall link up and fill the child since it was created for us in + // ChannelTrace::RenderJSON + json_iterator = grpc_json_link_child(json, trace, json_iterator); + trace->parent = json; + trace->value = nullptr; + trace->key = "trace"; + trace->owns_value = false; + } + // reset the parent to be the data object. + json = data; + json_iterator = nullptr; + // We use -1 as sentinel values since proto default value for integers is + // zero, and the confuses the parser into thinking the value weren't present + json_iterator = + add_num_str(json, json_iterator, "callsStarted", calls_started_); + json_iterator = + add_num_str(json, json_iterator, "callsSucceeded", calls_succeeded_); + json_iterator = + add_num_str(json, json_iterator, "callsFailed", calls_failed_); + gpr_timespec ts = + grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME); + json_iterator = + grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp", + fmt_time(ts), GRPC_JSON_STRING, true); + // render and return the over json object + char* json_str = grpc_json_dump_to_string(top_level_json, 0); + grpc_json_destroy(top_level_json); + return json_str; +} + +} // namespace channelz +} // namespace grpc_core diff --git a/src/core/lib/channel/channelz.h b/src/core/lib/channel/channelz.h new file mode 100644 index 0000000000..2aad1e82f4 --- /dev/null +++ b/src/core/lib/channel/channelz.h @@ -0,0 +1,85 @@ +/* + * + * Copyright 2018 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPC_CORE_LIB_CHANNEL_CHANNELZ_H +#define GRPC_CORE_LIB_CHANNEL_CHANNELZ_H + +#include <grpc/impl/codegen/port_platform.h> + +#include <grpc/grpc.h> + +#include "src/core/lib/channel/channel_trace.h" +#include "src/core/lib/gprpp/manual_constructor.h" +#include "src/core/lib/gprpp/ref_counted.h" +#include "src/core/lib/gprpp/ref_counted_ptr.h" +#include "src/core/lib/iomgr/error.h" +#include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/json/json.h" + +namespace grpc_core { +namespace channelz { + +namespace testing { +class ChannelNodePeer; +} + +class ChannelNode : public RefCounted<ChannelNode> { + public: + ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes); + ~ChannelNode(); + + void RecordCallStarted(); + void RecordCallFailed() { + gpr_atm_no_barrier_fetch_add(&calls_failed_, (gpr_atm(1))); + } + void RecordCallSucceeded() { + gpr_atm_no_barrier_fetch_add(&calls_succeeded_, (gpr_atm(1))); + } + + char* RenderJSON(); + + ChannelTrace* trace() { return trace_.get(); } + + void set_channel_destroyed() { + GPR_ASSERT(channel_ != nullptr); + channel_ = nullptr; + } + + intptr_t channel_uuid() { return channel_uuid_; } + + private: + // testing peer friend. + friend class testing::ChannelNodePeer; + + // helper for getting connectivity state. + grpc_connectivity_state GetConnectivityState(); + + grpc_channel* channel_ = nullptr; + UniquePtr<char> target_; + gpr_atm calls_started_ = 0; + gpr_atm calls_succeeded_ = 0; + gpr_atm calls_failed_ = 0; + gpr_atm last_call_started_millis_ = 0; + intptr_t channel_uuid_; + ManualConstructor<ChannelTrace> trace_; +}; + +} // namespace channelz +} // namespace grpc_core + +#endif /* GRPC_CORE_LIB_CHANNEL_CHANNELZ_H */ diff --git a/src/core/lib/channel/connected_channel.cc b/src/core/lib/channel/connected_channel.cc index ddd3029402..e2ea334ded 100644 --- a/src/core/lib/channel/connected_channel.cc +++ b/src/core/lib/channel/connected_channel.cc @@ -51,6 +51,7 @@ typedef struct connected_channel_call_data { callback_state on_complete[6]; // Max number of pending batches. callback_state recv_initial_metadata_ready; callback_state recv_message_ready; + callback_state recv_trailing_metadata_ready; } call_data; static void run_in_call_combiner(void* arg, grpc_error* error) { @@ -111,6 +112,12 @@ static void con_start_transport_stream_op_batch( intercept_callback(calld, state, false, "recv_message_ready", &batch->payload->recv_message.recv_message_ready); } + if (batch->recv_trailing_metadata) { + callback_state* state = &calld->recv_trailing_metadata_ready; + intercept_callback( + calld, state, false, "recv_trailing_metadata_ready", + &batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready); + } if (batch->cancel_stream) { // There can be more than one cancellation batch in flight at any // given time, so we can't just pick out a fixed index into @@ -121,7 +128,7 @@ static void con_start_transport_stream_op_batch( static_cast<callback_state*>(gpr_malloc(sizeof(*state))); intercept_callback(calld, state, true, "on_complete (cancel_stream)", &batch->on_complete); - } else { + } else if (batch->on_complete != nullptr) { callback_state* state = get_state_for_batch(calld, batch); intercept_callback(calld, state, false, "on_complete", &batch->on_complete); } diff --git a/src/core/lib/gprpp/inlined_vector.h b/src/core/lib/gprpp/inlined_vector.h index f36f6cb706..0d2586e507 100644 --- a/src/core/lib/gprpp/inlined_vector.h +++ b/src/core/lib/gprpp/inlined_vector.h @@ -99,6 +99,8 @@ class InlinedVector { void push_back(T&& value) { emplace_back(std::move(value)); } size_t size() const { return size_; } + bool empty() const { return size_ == 0; } + size_t capacity() const { return capacity_; } void clear() { diff --git a/src/core/lib/iomgr/call_combiner.h b/src/core/lib/iomgr/call_combiner.h index 0ccd08ea57..641fa18082 100644 --- a/src/core/lib/iomgr/call_combiner.h +++ b/src/core/lib/iomgr/call_combiner.h @@ -26,6 +26,7 @@ #include <grpc/support/atm.h> #include "src/core/lib/gpr/mpscq.h" +#include "src/core/lib/gprpp/inlined_vector.h" #include "src/core/lib/iomgr/closure.h" // A simple, lock-free mechanism for serializing activity related to a @@ -109,4 +110,83 @@ void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner, void grpc_call_combiner_cancel(grpc_call_combiner* call_combiner, grpc_error* error); +namespace grpc_core { + +// Helper for running a list of closures in a call combiner. +// +// Each callback running in the call combiner will eventually be +// returned to the surface, at which point the surface will yield the +// call combiner. So when we are running in the call combiner and have +// more than one callback to return to the surface, we need to re-enter +// the call combiner for all but one of those callbacks. +class CallCombinerClosureList { + public: + CallCombinerClosureList() {} + + // Adds a closure to the list. The closure must eventually result in + // the call combiner being yielded. + void Add(grpc_closure* closure, grpc_error* error, const char* reason) { + closures_.emplace_back(closure, error, reason); + } + + // Runs all closures in the call combiner and yields the call combiner. + // + // All but one of the closures in the list will be scheduled via + // GRPC_CALL_COMBINER_START(), and the remaining closure will be + // scheduled via GRPC_CLOSURE_SCHED(), which will eventually result in + // yielding the call combiner. If the list is empty, then the call + // combiner will be yielded immediately. + void RunClosures(grpc_call_combiner* call_combiner) { + if (closures_.empty()) { + GRPC_CALL_COMBINER_STOP(call_combiner, "no closures to schedule"); + return; + } + for (size_t i = 1; i < closures_.size(); ++i) { + auto& closure = closures_[i]; + GRPC_CALL_COMBINER_START(call_combiner, closure.closure, closure.error, + closure.reason); + } + if (grpc_call_combiner_trace.enabled()) { + gpr_log(GPR_INFO, + "CallCombinerClosureList executing closure while already " + "holding call_combiner %p: closure=%p error=%s reason=%s", + call_combiner, closures_[0].closure, + grpc_error_string(closures_[0].error), closures_[0].reason); + } + // This will release the call combiner. + GRPC_CLOSURE_SCHED(closures_[0].closure, closures_[0].error); + closures_.clear(); + } + + // Runs all closures in the call combiner, but does NOT yield the call + // combiner. All closures will be scheduled via GRPC_CALL_COMBINER_START(). + void RunClosuresWithoutYielding(grpc_call_combiner* call_combiner) { + for (size_t i = 0; i < closures_.size(); ++i) { + auto& closure = closures_[i]; + GRPC_CALL_COMBINER_START(call_combiner, closure.closure, closure.error, + closure.reason); + } + closures_.clear(); + } + + size_t size() const { return closures_.size(); } + + private: + struct CallCombinerClosure { + grpc_closure* closure; + grpc_error* error; + const char* reason; + + CallCombinerClosure(grpc_closure* closure, grpc_error* error, + const char* reason) + : closure(closure), error(error), reason(reason) {} + }; + + // There are generally a maximum of 6 closures to run in the call + // combiner, one for each pending op. + InlinedVector<CallCombinerClosure, 6> closures_; +}; + +} // namespace grpc_core + #endif /* GRPC_CORE_LIB_IOMGR_CALL_COMBINER_H */ diff --git a/src/core/lib/iomgr/closure.h b/src/core/lib/iomgr/closure.h index 34a494485d..f14c723844 100644 --- a/src/core/lib/iomgr/closure.h +++ b/src/core/lib/iomgr/closure.h @@ -283,9 +283,10 @@ inline void grpc_closure_sched(grpc_closure* c, grpc_error* error) { if (c->scheduled) { gpr_log(GPR_ERROR, "Closure already scheduled. (closure: %p, created: [%s:%d], " - "previously scheduled at: [%s: %d] run?: %s", + "previously scheduled at: [%s: %d], newly scheduled at [%s: %d], " + "run?: %s", c, c->file_created, c->line_created, c->file_initiated, - c->line_initiated, c->run ? "true" : "false"); + c->line_initiated, file, line, c->run ? "true" : "false"); abort(); } c->scheduled = true; diff --git a/src/core/lib/iomgr/socket_utils.h b/src/core/lib/iomgr/socket_utils.h index cf1a7be648..14bb081e93 100644 --- a/src/core/lib/iomgr/socket_utils.h +++ b/src/core/lib/iomgr/socket_utils.h @@ -23,12 +23,21 @@ #include <stddef.h> +// TODO(juanlishen): The following functions might be simple enough to implement +// ourselves, so that they don't cause any portability hassle. + /* A wrapper for htons on POSIX and Windows */ uint16_t grpc_htons(uint16_t hostshort); /* A wrapper for ntohs on POSIX and WINDOWS */ uint16_t grpc_ntohs(uint16_t netshort); +/* A wrapper for htonl on POSIX and Windows */ +uint32_t grpc_htonl(uint32_t hostlong); + +/* A wrapper for ntohl on POSIX and WINDOWS */ +uint32_t grpc_ntohl(uint32_t netlong); + /* A wrapper for inet_pton on POSIX and WINDOWS */ int grpc_inet_pton(int af, const char* src, void* dst); diff --git a/src/core/lib/iomgr/socket_utils_common_posix.cc b/src/core/lib/iomgr/socket_utils_common_posix.cc index caee652307..c4b991c94d 100644 --- a/src/core/lib/iomgr/socket_utils_common_posix.cc +++ b/src/core/lib/iomgr/socket_utils_common_posix.cc @@ -339,6 +339,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); } uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); } +uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); } + +uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); } + int grpc_inet_pton(int af, const char* src, void* dst) { return inet_pton(af, src, dst); } diff --git a/src/core/lib/iomgr/socket_utils_uv.cc b/src/core/lib/iomgr/socket_utils_uv.cc index 7eba40c46b..b5f96b52df 100644 --- a/src/core/lib/iomgr/socket_utils_uv.cc +++ b/src/core/lib/iomgr/socket_utils_uv.cc @@ -33,6 +33,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); } uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); } +uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); } + +uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); } + int grpc_inet_pton(int af, const char* src, void* dst) { return inet_pton(af, src, dst); } diff --git a/src/core/lib/iomgr/socket_utils_windows.cc b/src/core/lib/iomgr/socket_utils_windows.cc index 3e7b5b812d..9137ab98e6 100644 --- a/src/core/lib/iomgr/socket_utils_windows.cc +++ b/src/core/lib/iomgr/socket_utils_windows.cc @@ -31,6 +31,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); } uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); } +uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); } + +uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); } + int grpc_inet_pton(int af, const char* src, void* dst) { return inet_pton(af, src, dst); } diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.cc b/src/core/lib/security/credentials/google_default/google_default_credentials.cc index 38c9175717..c456ffaf5d 100644 --- a/src/core/lib/security/credentials/google_default/google_default_credentials.cc +++ b/src/core/lib/security/credentials/google_default/google_default_credentials.cc @@ -231,7 +231,8 @@ end: creds->base.vtable = &google_default_credentials_vtable; creds->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_GOOGLE_DEFAULT; gpr_ref_init(&creds->base.refcount, 1); - creds->ssl_creds = grpc_ssl_credentials_create(nullptr, nullptr, nullptr); + creds->ssl_creds = + grpc_ssl_credentials_create(nullptr, nullptr, nullptr, nullptr); GPR_ASSERT(creds->ssl_creds != nullptr); grpc_alts_credentials_options* options = grpc_alts_credentials_client_options_create(); diff --git a/src/core/lib/security/credentials/ssl/ssl_credentials.cc b/src/core/lib/security/credentials/ssl/ssl_credentials.cc index 2b6377d3ec..3d6f2f200a 100644 --- a/src/core/lib/security/credentials/ssl/ssl_credentials.cc +++ b/src/core/lib/security/credentials/ssl/ssl_credentials.cc @@ -48,6 +48,10 @@ static void ssl_destruct(grpc_channel_credentials* creds) { grpc_ssl_credentials* c = reinterpret_cast<grpc_ssl_credentials*>(creds); gpr_free(c->config.pem_root_certs); grpc_tsi_ssl_pem_key_cert_pairs_destroy(c->config.pem_key_cert_pair, 1); + if (c->config.verify_options.verify_peer_destruct != nullptr) { + c->config.verify_options.verify_peer_destruct( + c->config.verify_options.verify_peer_callback_userdata); + } } static grpc_security_status ssl_create_security_connector( @@ -87,6 +91,7 @@ static grpc_channel_credentials_vtable ssl_vtable = { static void ssl_build_config(const char* pem_root_certs, grpc_ssl_pem_key_cert_pair* pem_key_cert_pair, + const verify_peer_options* verify_options, grpc_ssl_config* config) { if (pem_root_certs != nullptr) { config->pem_root_certs = gpr_strdup(pem_root_certs); @@ -101,23 +106,32 @@ static void ssl_build_config(const char* pem_root_certs, config->pem_key_cert_pair->private_key = gpr_strdup(pem_key_cert_pair->private_key); } + if (verify_options != nullptr) { + memcpy(&config->verify_options, verify_options, + sizeof(verify_peer_options)); + } else { + // Otherwise set all options to default values + memset(&config->verify_options, 0, sizeof(verify_peer_options)); + } } grpc_channel_credentials* grpc_ssl_credentials_create( const char* pem_root_certs, grpc_ssl_pem_key_cert_pair* pem_key_cert_pair, - void* reserved) { + const verify_peer_options* verify_options, void* reserved) { grpc_ssl_credentials* c = static_cast<grpc_ssl_credentials*>( gpr_zalloc(sizeof(grpc_ssl_credentials))); GRPC_API_TRACE( "grpc_ssl_credentials_create(pem_root_certs=%s, " "pem_key_cert_pair=%p, " + "verify_options=%p, " "reserved=%p)", - 3, (pem_root_certs, pem_key_cert_pair, reserved)); + 4, (pem_root_certs, pem_key_cert_pair, verify_options, reserved)); GPR_ASSERT(reserved == nullptr); c->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_SSL; c->base.vtable = &ssl_vtable; gpr_ref_init(&c->base.refcount, 1); - ssl_build_config(pem_root_certs, pem_key_cert_pair, &c->config); + ssl_build_config(pem_root_certs, pem_key_cert_pair, verify_options, + &c->config); return &c->base; } diff --git a/src/core/lib/security/security_connector/security_connector.cc b/src/core/lib/security/security_connector/security_connector.cc index b54a7643e4..cc72bb6164 100644 --- a/src/core/lib/security/security_connector/security_connector.cc +++ b/src/core/lib/security/security_connector/security_connector.cc @@ -620,6 +620,7 @@ typedef struct { tsi_ssl_client_handshaker_factory* client_handshaker_factory; char* target_name; char* overridden_target_name; + const verify_peer_options* verify_options; } grpc_ssl_channel_security_connector; typedef struct { @@ -878,11 +879,34 @@ static void ssl_channel_check_peer(grpc_security_connector* sc, tsi_peer peer, grpc_closure* on_peer_checked) { grpc_ssl_channel_security_connector* c = reinterpret_cast<grpc_ssl_channel_security_connector*>(sc); - grpc_error* error = ssl_check_peer(sc, - c->overridden_target_name != nullptr - ? c->overridden_target_name - : c->target_name, - &peer, auth_context); + const char* target_name = c->overridden_target_name != nullptr + ? c->overridden_target_name + : c->target_name; + grpc_error* error = ssl_check_peer(sc, target_name, &peer, auth_context); + if (error == GRPC_ERROR_NONE && + c->verify_options->verify_peer_callback != nullptr) { + const tsi_peer_property* p = + tsi_peer_get_property_by_name(&peer, TSI_X509_PEM_CERT_PROPERTY); + if (p == nullptr) { + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Cannot check peer: missing pem cert property."); + } else { + char* peer_pem = static_cast<char*>(gpr_malloc(p->value.length + 1)); + memcpy(peer_pem, p->value.data, p->value.length); + peer_pem[p->value.length] = '\0'; + int callback_status = c->verify_options->verify_peer_callback( + target_name, peer_pem, + c->verify_options->verify_peer_callback_userdata); + gpr_free(peer_pem); + if (callback_status) { + char* msg; + gpr_asprintf(&msg, "Verify peer callback returned a failure (%d)", + callback_status); + error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + gpr_free(msg); + } + } + } GRPC_CLOSURE_SCHED(on_peer_checked, error); tsi_peer_destruct(&peer); } @@ -1047,6 +1071,7 @@ grpc_security_status grpc_ssl_channel_security_connector_create( if (overridden_target_name != nullptr) { c->overridden_target_name = gpr_strdup(overridden_target_name); } + c->verify_options = &config->verify_options; has_key_cert_pair = config->pem_key_cert_pair != nullptr && config->pem_key_cert_pair->private_key != nullptr && diff --git a/src/core/lib/security/security_connector/security_connector.h b/src/core/lib/security/security_connector/security_connector.h index f9723166d0..67a506b576 100644 --- a/src/core/lib/security/security_connector/security_connector.h +++ b/src/core/lib/security/security_connector/security_connector.h @@ -193,6 +193,7 @@ grpc_server_security_connector* grpc_fake_server_security_connector_create( typedef struct { tsi_ssl_pem_key_cert_pair* pem_key_cert_pair; char* pem_root_certs; + verify_peer_options verify_options; } grpc_ssl_config; /* Creates an SSL channel_security_connector. diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc index 1cf8ea94e7..556eb234b4 100644 --- a/src/core/lib/surface/call.cc +++ b/src/core/lib/surface/call.cc @@ -233,6 +233,7 @@ struct grpc_call { grpc_closure receiving_slice_ready; grpc_closure receiving_stream_ready; grpc_closure receiving_initial_metadata_ready; + grpc_closure receiving_trailing_metadata_ready; uint32_t test_only_last_message_flags; grpc_closure release_call; @@ -270,8 +271,17 @@ struct grpc_call { grpc_core::TraceFlag grpc_call_error_trace(false, "call_error"); grpc_core::TraceFlag grpc_compression_trace(false, "compression"); -#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack*)((call) + 1)) -#define CALL_FROM_CALL_STACK(call_stack) (((grpc_call*)(call_stack)) - 1) +/* Given a size, round up to the next multiple of sizeof(void*) */ +#define ROUND_UP_TO_ALIGNMENT_SIZE(x) \ + (((x) + GPR_MAX_ALIGNMENT - 1u) & ~(GPR_MAX_ALIGNMENT - 1u)) + +#define CALL_STACK_FROM_CALL(call) \ + (grpc_call_stack*)((char*)(call) + \ + ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call))) +#define CALL_FROM_CALL_STACK(call_stack) \ + (grpc_call*)(((char*)(call_stack)) - \ + ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call))) + #define CALL_ELEM_FROM_CALL(call, idx) \ grpc_call_stack_element(CALL_STACK_FROM_CALL(call), idx) #define CALL_FROM_TOP_ELEM(top_elem) \ @@ -342,8 +352,9 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, size_t initial_size = grpc_channel_get_call_size_estimate(args->channel); GRPC_STATS_INC_CALL_INITIAL_SIZE(initial_size); gpr_arena* arena = gpr_arena_create(initial_size); - call = static_cast<grpc_call*>(gpr_arena_alloc( - arena, sizeof(grpc_call) + channel_stack->call_stack_size)); + call = static_cast<grpc_call*>( + gpr_arena_alloc(arena, ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)) + + channel_stack->call_stack_size)); gpr_ref_init(&call->ext_ref, 1); call->arena = arena; grpc_call_combiner_init(&call->call_combiner); @@ -478,6 +489,12 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, &call->pollent); } + grpc_core::channelz::ChannelNode* channelz_channel = + grpc_channel_get_channelz_node(call->channel); + if (channelz_channel != nullptr) { + channelz_channel->RecordCallStarted(); + } + grpc_slice_unref_internal(path); return error; @@ -520,7 +537,6 @@ static void release_call(void* call, grpc_error* error) { GRPC_CHANNEL_INTERNAL_UNREF(channel, "call"); } -static void set_status_value_directly(grpc_status_code status, void* dest); static void destroy_call(void* call, grpc_error* error) { GPR_TIMER_SCOPE("destroy_call", 0); size_t i; @@ -1076,13 +1092,12 @@ static void recv_trailing_filter(void* args, grpc_metadata_batch* b) { if (b->idx.named.grpc_status != nullptr) { grpc_status_code status_code = grpc_get_status_code_from_metadata(b->idx.named.grpc_status->md); - grpc_error* error = - status_code == GRPC_STATUS_OK - ? GRPC_ERROR_NONE - : grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Error received from peer"), - GRPC_ERROR_INT_GRPC_STATUS, - static_cast<intptr_t>(status_code)); + grpc_error* error = GRPC_ERROR_NONE; + if (status_code != GRPC_STATUS_OK) { + error = grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Error received from peer"), + GRPC_ERROR_INT_GRPC_STATUS, static_cast<intptr_t>(status_code)); + } if (b->idx.named.grpc_message != nullptr) { error = grpc_error_set_str( error, GRPC_ERROR_STR_GRPC_MESSAGE, @@ -1209,7 +1224,6 @@ static void post_batch_completion(batch_control* bctl) { if (bctl->op.send_initial_metadata) { grpc_metadata_batch_destroy( - &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]); } if (bctl->op.send_message) { @@ -1217,14 +1231,9 @@ static void post_batch_completion(batch_control* bctl) { } if (bctl->op.send_trailing_metadata) { grpc_metadata_batch_destroy( - &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]); } if (bctl->op.recv_trailing_metadata) { - grpc_metadata_batch* md = - &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - recv_trailing_filter(call, md); - /* propagate cancellation to any interested children */ gpr_atm_rel_store(&call->received_final_op_atm, 1); parent_call* pc = get_parent_call(call); @@ -1246,7 +1255,6 @@ static void post_batch_completion(batch_control* bctl) { } gpr_mu_unlock(&pc->child_list_mu); } - if (call->is_client) { get_final_status(call, set_status_value_directly, call->final_op.client.status, @@ -1256,7 +1264,15 @@ static void post_batch_completion(batch_control* bctl) { get_final_status(call, set_cancelled_value, call->final_op.server.cancelled, nullptr, nullptr); } - + grpc_core::channelz::ChannelNode* channelz_channel = + grpc_channel_get_channelz_node(call->channel); + if (channelz_channel != nullptr) { + if (*call->final_op.client.status != GRPC_STATUS_OK) { + channelz_channel->RecordCallFailed(); + } else { + channelz_channel->RecordCallSucceeded(); + } + } GRPC_ERROR_UNREF(error); error = GRPC_ERROR_NONE; } @@ -1538,6 +1554,17 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) { finish_batch_step(bctl); } +static void receiving_trailing_metadata_ready(void* bctlp, grpc_error* error) { + batch_control* bctl = static_cast<batch_control*>(bctlp); + grpc_call* call = bctl->call; + GRPC_CALL_COMBINER_STOP(&call->call_combiner, "recv_trailing_metadata_ready"); + add_batch_error(bctl, GRPC_ERROR_REF(error), false); + grpc_metadata_batch* md = + &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; + recv_trailing_filter(call, md); + finish_batch_step(bctl); +} + static void finish_batch(void* bctlp, grpc_error* error) { batch_control* bctl = static_cast<batch_control*>(bctlp); grpc_call* call = bctl->call; @@ -1558,7 +1585,8 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, size_t i; const grpc_op* op; batch_control* bctl; - int num_completion_callbacks_needed = 1; + bool has_send_ops = false; + int num_recv_ops = 0; grpc_call_error error = GRPC_CALL_OK; grpc_transport_stream_op_batch* stream_op; grpc_transport_stream_op_batch_payload* stream_op_payload; @@ -1664,6 +1692,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, stream_op_payload->send_initial_metadata.peer_string = &call->peer_string; } + has_send_ops = true; break; } case GRPC_OP_SEND_MESSAGE: { @@ -1693,6 +1722,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, &op->data.send_message.send_message->data.raw.slice_buffer, flags); stream_op_payload->send_message.send_message.reset( call->sending_stream.get()); + has_send_ops = true; break; } case GRPC_OP_SEND_CLOSE_FROM_CLIENT: { @@ -1713,6 +1743,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, call->sent_final_op = true; stream_op_payload->send_trailing_metadata.send_trailing_metadata = &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]; + has_send_ops = true; break; } case GRPC_OP_SEND_STATUS_FROM_SERVER: { @@ -1777,6 +1808,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, } stream_op_payload->send_trailing_metadata.send_trailing_metadata = &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]; + has_send_ops = true; break; } case GRPC_OP_RECV_INITIAL_METADATA: { @@ -1804,7 +1836,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, stream_op_payload->recv_initial_metadata.peer_string = &call->peer_string; } - num_completion_callbacks_needed++; + ++num_recv_ops; break; } case GRPC_OP_RECV_MESSAGE: { @@ -1826,7 +1858,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, grpc_schedule_on_exec_ctx); stream_op_payload->recv_message.recv_message_ready = &call->receiving_stream_ready; - num_completion_callbacks_needed++; + ++num_recv_ops; break; } case GRPC_OP_RECV_STATUS_ON_CLIENT: { @@ -1852,11 +1884,16 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, call->final_op.client.error_string = op->data.recv_status_on_client.error_string; stream_op->recv_trailing_metadata = true; - stream_op->collect_stats = true; stream_op_payload->recv_trailing_metadata.recv_trailing_metadata = &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - stream_op_payload->collect_stats.collect_stats = + stream_op_payload->recv_trailing_metadata.collect_stats = &call->final_info.stats.transport_stream_stats; + GRPC_CLOSURE_INIT(&call->receiving_trailing_metadata_ready, + receiving_trailing_metadata_ready, bctl, + grpc_schedule_on_exec_ctx); + stream_op_payload->recv_trailing_metadata.recv_trailing_metadata_ready = + &call->receiving_trailing_metadata_ready; + ++num_recv_ops; break; } case GRPC_OP_RECV_CLOSE_ON_SERVER: { @@ -1877,11 +1914,16 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, call->final_op.server.cancelled = op->data.recv_close_on_server.cancelled; stream_op->recv_trailing_metadata = true; - stream_op->collect_stats = true; stream_op_payload->recv_trailing_metadata.recv_trailing_metadata = &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - stream_op_payload->collect_stats.collect_stats = + stream_op_payload->recv_trailing_metadata.collect_stats = &call->final_info.stats.transport_stream_stats; + GRPC_CLOSURE_INIT(&call->receiving_trailing_metadata_ready, + receiving_trailing_metadata_ready, bctl, + grpc_schedule_on_exec_ctx); + stream_op_payload->recv_trailing_metadata.recv_trailing_metadata_ready = + &call->receiving_trailing_metadata_ready; + ++num_recv_ops; break; } } @@ -1891,13 +1933,15 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, if (!is_notify_tag_closure) { GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag)); } - gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed); + gpr_ref_init(&bctl->steps_to_complete, (has_send_ops ? 1 : 0) + num_recv_ops); - GRPC_CLOSURE_INIT(&bctl->finish_batch, finish_batch, bctl, - grpc_schedule_on_exec_ctx); - stream_op->on_complete = &bctl->finish_batch; - gpr_atm_rel_store(&call->any_ops_sent_atm, 1); + if (has_send_ops) { + GRPC_CLOSURE_INIT(&bctl->finish_batch, finish_batch, bctl, + grpc_schedule_on_exec_ctx); + stream_op->on_complete = &bctl->finish_batch; + } + gpr_atm_rel_store(&call->any_ops_sent_atm, 1); execute_batch(call, stream_op, &bctl->start_batch); done: diff --git a/src/core/lib/surface/channel.cc b/src/core/lib/surface/channel.cc index a466b325be..d5d75fcb2a 100644 --- a/src/core/lib/surface/channel.cc +++ b/src/core/lib/surface/channel.cc @@ -32,6 +32,7 @@ #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channel_trace.h" +#include "src/core/lib/channel/channelz.h" #include "src/core/lib/debug/stats.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/gprpp/manual_constructor.h" @@ -66,7 +67,7 @@ struct grpc_channel { gpr_mu registered_call_mu; registered_call* registered_calls; - grpc_core::RefCountedPtr<grpc_core::ChannelTrace> tracer; + grpc_core::RefCountedPtr<grpc_core::channelz::ChannelNode> channelz_channel; char* target; }; @@ -103,6 +104,7 @@ grpc_channel* grpc_channel_create_with_builder( channel->target = target; channel->is_client = grpc_channel_stack_type_is_client(channel_stack_type); size_t channel_tracer_max_nodes = 0; // default to off + bool channelz_enabled = false; gpr_mu_init(&channel->registered_call_mu); channel->registered_calls = nullptr; @@ -141,15 +143,20 @@ grpc_channel* grpc_channel_create_with_builder( const grpc_integer_options options = {0, 0, INT_MAX}; channel_tracer_max_nodes = (size_t)grpc_channel_arg_get_integer(&args->args[i], options); + } else if (0 == strcmp(args->args[i].key, GRPC_ARG_ENABLE_CHANNELZ)) { + channelz_enabled = grpc_channel_arg_get_bool(&args->args[i], false); } } grpc_channel_args_destroy(args); - channel->tracer = grpc_core::MakeRefCounted<grpc_core::ChannelTrace>( - channel_tracer_max_nodes); - channel->tracer->AddTraceEvent( - grpc_core::ChannelTrace::Severity::Info, - grpc_slice_from_static_string("Channel created")); + if (channelz_enabled) { + channel->channelz_channel = + grpc_core::MakeRefCounted<grpc_core::channelz::ChannelNode>( + channel, channel_tracer_max_nodes); + channel->channelz_channel->trace()->AddTraceEvent( + grpc_core::channelz::ChannelTrace::Severity::Info, + grpc_slice_from_static_string("Channel created")); + } return channel; } @@ -184,12 +191,9 @@ static grpc_channel_args* build_channel_args( return grpc_channel_args_copy_and_add(input_args, new_args, num_new_args); } -char* grpc_channel_get_trace(grpc_channel* channel) { - return channel->tracer->RenderTrace(); -} - -intptr_t grpc_channel_get_uuid(grpc_channel* channel) { - return channel->tracer->GetUuid(); +grpc_core::channelz::ChannelNode* grpc_channel_get_channelz_node( + grpc_channel* channel) { + return channel->channelz_channel.get(); } grpc_channel* grpc_channel_create(const char* target, @@ -395,6 +399,10 @@ void grpc_channel_internal_unref(grpc_channel* c REF_ARG) { static void destroy_channel(void* arg, grpc_error* error) { grpc_channel* channel = static_cast<grpc_channel*>(arg); + if (channel->channelz_channel != nullptr) { + channel->channelz_channel->set_channel_destroyed(); + channel->channelz_channel.reset(); + } grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CHANNEL(channel)); while (channel->registered_calls) { registered_call* rc = channel->registered_calls; @@ -403,7 +411,6 @@ static void destroy_channel(void* arg, grpc_error* error) { GRPC_MDELEM_UNREF(rc->authority); gpr_free(rc); } - channel->tracer.reset(); gpr_mu_destroy(&channel->registered_call_mu); gpr_free(channel->target); gpr_free(channel); diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h index 288313951e..e5ff2c3596 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -23,6 +23,7 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/channel/channel_stack_builder.h" +#include "src/core/lib/channel/channelz.h" #include "src/core/lib/surface/channel_stack_type.h" grpc_channel* grpc_channel_create(const char* target, @@ -50,6 +51,9 @@ grpc_call* grpc_channel_create_pollset_set_call( /** Get a (borrowed) pointer to this channels underlying channel stack */ grpc_channel_stack* grpc_channel_get_channel_stack(grpc_channel* channel); +grpc_core::channelz::ChannelNode* grpc_channel_get_channelz_node( + grpc_channel* channel); + /** Get a grpc_mdelem of grpc-status: X where X is the numeric value of status_code. diff --git a/src/core/lib/transport/transport.cc b/src/core/lib/transport/transport.cc index 039d603394..cbdb77c844 100644 --- a/src/core/lib/transport/transport.cc +++ b/src/core/lib/transport/transport.cc @@ -212,21 +212,32 @@ void grpc_transport_stream_op_batch_finish_with_failure( if (batch->send_message) { batch->payload->send_message.send_message.reset(); } - if (batch->recv_message) { - GRPC_CALL_COMBINER_START( - call_combiner, batch->payload->recv_message.recv_message_ready, - GRPC_ERROR_REF(error), "failing recv_message_ready"); + if (batch->cancel_stream) { + GRPC_ERROR_UNREF(batch->payload->cancel_stream.cancel_error); } + // Construct a list of closures to execute. + grpc_core::CallCombinerClosureList closures; if (batch->recv_initial_metadata) { - GRPC_CALL_COMBINER_START( - call_combiner, + closures.Add( batch->payload->recv_initial_metadata.recv_initial_metadata_ready, GRPC_ERROR_REF(error), "failing recv_initial_metadata_ready"); } - GRPC_CLOSURE_SCHED(batch->on_complete, error); - if (batch->cancel_stream) { - GRPC_ERROR_UNREF(batch->payload->cancel_stream.cancel_error); + if (batch->recv_message) { + closures.Add(batch->payload->recv_message.recv_message_ready, + GRPC_ERROR_REF(error), "failing recv_message_ready"); + } + if (batch->recv_trailing_metadata) { + closures.Add( + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready, + GRPC_ERROR_REF(error), "failing recv_trailing_metadata_ready"); + } + if (batch->on_complete != nullptr) { + closures.Add(batch->on_complete, GRPC_ERROR_REF(error), + "failing on_complete"); } + // Execute closures. + closures.RunClosures(call_combiner); + GRPC_ERROR_UNREF(error); } typedef struct { diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index b2e252d939..585b9dfae9 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -122,9 +122,15 @@ typedef struct grpc_transport_stream_op_batch_payload /* Transport stream op: a set of operations to perform on a transport against a single stream */ typedef struct grpc_transport_stream_op_batch { - /** Should be enqueued when all requested operations (excluding recv_message - and recv_initial_metadata which have their own closures) in a given batch - have been completed. */ + /** Should be scheduled when all of the non-recv operations in the batch + are complete. + + The recv ops (recv_initial_metadata, recv_message, and + recv_trailing_metadata) each have their own callbacks. If a batch + contains both recv ops and non-recv ops, on_complete should be + scheduled as soon as the non-recv ops are complete, regardless of + whether or not the recv ops are complete. If a batch contains + only recv ops, on_complete can be null. */ grpc_closure* on_complete; /** Values for the stream op (fields set are determined by flags above) */ @@ -149,9 +155,6 @@ typedef struct grpc_transport_stream_op_batch { */ bool recv_trailing_metadata : 1; - /** Collect any stats into provided buffer, zero internal stat counters */ - bool collect_stats : 1; - /** Cancel this stream with the provided error */ bool cancel_stream : 1; @@ -219,11 +222,10 @@ struct grpc_transport_stream_op_batch_payload { struct { grpc_metadata_batch* recv_trailing_metadata; - } recv_trailing_metadata; - - struct { grpc_transport_stream_stats* collect_stats; - } collect_stats; + /** Should be enqueued when initial metadata is ready to be processed. */ + grpc_closure* recv_trailing_metadata_ready; + } recv_trailing_metadata; /** Forcefully close this stream. The HTTP2 semantics should be: diff --git a/src/core/lib/transport/transport_op_string.cc b/src/core/lib/transport/transport_op_string.cc index 25ab492f3a..8c7db642a5 100644 --- a/src/core/lib/transport/transport_op_string.cc +++ b/src/core/lib/transport/transport_op_string.cc @@ -120,13 +120,6 @@ char* grpc_transport_stream_op_batch_string( gpr_strvec_add(&b, tmp); } - if (op->collect_stats) { - gpr_strvec_add(&b, gpr_strdup(" ")); - gpr_asprintf(&tmp, "COLLECT_STATS:%p", - op->payload->collect_stats.collect_stats); - gpr_strvec_add(&b, tmp); - } - out = gpr_strvec_flatten(&b, nullptr); gpr_strvec_destroy(&b); |