aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext/filters/client_channel/client_channel.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/ext/filters/client_channel/client_channel.cc')
-rw-r--r--src/core/ext/filters/client_channel/client_channel.cc270
1 files changed, 153 insertions, 117 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc
index 58db7a05b4..d42961fc60 100644
--- a/src/core/ext/filters/client_channel/client_channel.cc
+++ b/src/core/ext/filters/client_channel/client_channel.cc
@@ -938,12 +938,26 @@ static void cc_destroy_channel_elem(grpc_channel_element* elem) {
// (census filter is on top of this one)
// - add census stats for retries
+namespace {
+struct call_data;
+
// State used for starting a retryable batch on a subchannel call.
// This provides its own grpc_transport_stream_op_batch and other data
// structures needed to populate the ops in the batch.
// We allocate one struct on the arena for each attempt at starting a
// batch on a given subchannel call.
-typedef struct {
+struct subchannel_batch_data {
+ subchannel_batch_data(grpc_call_element* elem, call_data* calld, int refcount,
+ bool set_on_complete);
+ // All dtor code must be added in `destroy`. This is because we may
+ // call closures in `subchannel_batch_data` after they are unrefed by
+ // `batch_data_unref`, and msan would complain about accessing this class
+ // after calling dtor. As a result we cannot call the `dtor` in
+ // `batch_data_unref`.
+ // TODO(soheil): We should try to call the dtor in `batch_data_unref`.
+ ~subchannel_batch_data() { destroy(); }
+ void destroy();
+
gpr_refcount refs;
grpc_call_element* elem;
grpc_subchannel_call* subchannel_call; // Holds a ref.
@@ -952,11 +966,23 @@ typedef struct {
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 {
+struct subchannel_call_retry_state {
+ explicit subchannel_call_retry_state(grpc_call_context_element* context)
+ : batch_payload(context),
+ started_send_initial_metadata(false),
+ completed_send_initial_metadata(false),
+ started_send_trailing_metadata(false),
+ completed_send_trailing_metadata(false),
+ started_recv_initial_metadata(false),
+ completed_recv_initial_metadata(false),
+ started_recv_trailing_metadata(false),
+ completed_recv_trailing_metadata(false),
+ retry_dispatched(false) {}
+
// subchannel_batch_data.batch.payload points to this.
grpc_transport_stream_op_batch_payload batch_payload;
// For send_initial_metadata.
@@ -975,7 +1001,7 @@ typedef struct {
// For intercepting recv_initial_metadata.
grpc_metadata_batch recv_initial_metadata;
grpc_closure recv_initial_metadata_ready;
- bool trailing_metadata_available;
+ bool trailing_metadata_available = false;
// For intercepting recv_message.
grpc_closure recv_message_ready;
grpc_core::OrphanablePtr<grpc_core::ByteStream> recv_message;
@@ -985,10 +1011,10 @@ typedef struct {
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;
- size_t completed_send_message_count;
- size_t started_recv_message_count;
- size_t completed_recv_message_count;
+ size_t started_send_message_count = 0;
+ size_t completed_send_message_count = 0;
+ size_t started_recv_message_count = 0;
+ size_t completed_recv_message_count = 0;
bool started_send_initial_metadata : 1;
bool completed_send_initial_metadata : 1;
bool started_send_trailing_metadata : 1;
@@ -999,12 +1025,12 @@ typedef struct {
bool completed_recv_trailing_metadata : 1;
// State for callback processing.
bool retry_dispatched : 1;
- subchannel_batch_data* recv_initial_metadata_ready_deferred_batch;
- grpc_error* recv_initial_metadata_error;
- subchannel_batch_data* recv_message_ready_deferred_batch;
- grpc_error* recv_message_error;
- subchannel_batch_data* recv_trailing_metadata_internal_batch;
-} subchannel_call_retry_state;
+ subchannel_batch_data* recv_initial_metadata_ready_deferred_batch = nullptr;
+ grpc_error* recv_initial_metadata_error = GRPC_ERROR_NONE;
+ subchannel_batch_data* recv_message_ready_deferred_batch = nullptr;
+ grpc_error* recv_message_error = GRPC_ERROR_NONE;
+ subchannel_batch_data* recv_trailing_metadata_internal_batch = nullptr;
+};
// Pending batches stored in call data.
typedef struct {
@@ -1019,7 +1045,44 @@ typedef struct {
Handles queueing of stream ops until a call object is ready, waiting
for initial metadata before trying to create a call object,
and handling cancellation gracefully. */
-typedef struct client_channel_call_data {
+struct call_data {
+ call_data(grpc_call_element* elem, const channel_data& chand,
+ const grpc_call_element_args& args)
+ : deadline_state(elem, args.call_stack, args.call_combiner,
+ GPR_LIKELY(chand.deadline_checking_enabled)
+ ? args.deadline
+ : GRPC_MILLIS_INF_FUTURE),
+ path(grpc_slice_ref_internal(args.path)),
+ call_start_time(args.start_time),
+ deadline(args.deadline),
+ arena(args.arena),
+ owning_call(args.call_stack),
+ call_combiner(args.call_combiner),
+ pending_send_initial_metadata(false),
+ pending_send_message(false),
+ pending_send_trailing_metadata(false),
+ enable_retries(chand.enable_retries),
+ retry_committed(false),
+ last_attempt_got_server_pushback(false) {}
+
+ ~call_data() {
+ if (GPR_LIKELY(subchannel_call != nullptr)) {
+ GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call,
+ "client_channel_destroy_call");
+ }
+ grpc_slice_unref_internal(path);
+ GRPC_ERROR_UNREF(cancel_error);
+ for (size_t i = 0; i < GPR_ARRAY_SIZE(pending_batches); ++i) {
+ GPR_ASSERT(pending_batches[i].batch == nullptr);
+ }
+ for (size_t i = 0; i < GRPC_CONTEXT_COUNT; ++i) {
+ if (pick.subchannel_call_context[i].value != nullptr) {
+ pick.subchannel_call_context[i].destroy(
+ pick.subchannel_call_context[i].value);
+ }
+ }
+ }
+
// State for handling deadlines.
// The code in deadline_filter.c requires this to be the first field.
// TODO(roth): This is slightly sub-optimal in that grpc_deadline_state
@@ -1038,24 +1101,24 @@ typedef struct client_channel_call_data {
grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data;
grpc_core::RefCountedPtr<ClientChannelMethodParams> method_params;
- grpc_subchannel_call* subchannel_call;
+ grpc_subchannel_call* subchannel_call = nullptr;
// Set when we get a cancel_stream op.
- grpc_error* cancel_error;
+ grpc_error* cancel_error = GRPC_ERROR_NONE;
grpc_core::LoadBalancingPolicy::PickState pick;
grpc_closure pick_closure;
grpc_closure pick_cancel_closure;
- grpc_polling_entity* pollent;
- bool pollent_added_to_interested_parties;
+ grpc_polling_entity* pollent = nullptr;
+ bool pollent_added_to_interested_parties = false;
// Batches are added to this list when received from above.
// They are removed when we are done handling the batch (i.e., when
// either we have invoked all of the batch's callbacks or we have
// passed the batch down to the subchannel call and are not
// intercepting any of its callbacks).
- pending_batch pending_batches[MAX_PENDING_BATCHES];
+ pending_batch pending_batches[MAX_PENDING_BATCHES] = {};
bool pending_send_initial_metadata : 1;
bool pending_send_message : 1;
bool pending_send_trailing_metadata : 1;
@@ -1064,8 +1127,8 @@ typedef struct client_channel_call_data {
bool enable_retries : 1;
bool retry_committed : 1;
bool last_attempt_got_server_pushback : 1;
- int num_attempts_completed;
- size_t bytes_buffered_for_retry;
+ int num_attempts_completed = 0;
+ size_t bytes_buffered_for_retry = 0;
grpc_core::ManualConstructor<grpc_core::BackOff> retry_backoff;
grpc_timer retry_timer;
@@ -1076,12 +1139,12 @@ typedef struct client_channel_call_data {
// until all of these batches have completed.
// Note that we actually only need to track replay batches, but it's
// easier to track all batches with send ops.
- int num_pending_retriable_subchannel_send_batches;
+ int num_pending_retriable_subchannel_send_batches = 0;
// Cached data for retrying send ops.
// send_initial_metadata
- bool seen_send_initial_metadata;
- grpc_linked_mdelem* send_initial_metadata_storage;
+ bool seen_send_initial_metadata = false;
+ grpc_linked_mdelem* send_initial_metadata_storage = nullptr;
grpc_metadata_batch send_initial_metadata;
uint32_t send_initial_metadata_flags;
gpr_atm* peer_string;
@@ -1092,14 +1155,13 @@ typedef struct client_channel_call_data {
// Note: We inline the cache for the first 3 send_message ops and use
// dynamic allocation after that. This number was essentially picked
// at random; it could be changed in the future to tune performance.
- grpc_core::ManualConstructor<
- grpc_core::InlinedVector<grpc_core::ByteStreamCache*, 3>>
- send_messages;
+ grpc_core::InlinedVector<grpc_core::ByteStreamCache*, 3> send_messages;
// send_trailing_metadata
- bool seen_send_trailing_metadata;
- grpc_linked_mdelem* send_trailing_metadata_storage;
+ bool seen_send_trailing_metadata = false;
+ grpc_linked_mdelem* send_trailing_metadata_storage = nullptr;
grpc_metadata_batch send_trailing_metadata;
-} call_data;
+};
+} // namespace
// Forward declarations.
static void retry_commit(grpc_call_element* elem,
@@ -1143,7 +1205,7 @@ static void maybe_cache_send_ops_for_batch(call_data* calld,
gpr_arena_alloc(calld->arena, sizeof(grpc_core::ByteStreamCache)));
new (cache) grpc_core::ByteStreamCache(
std::move(batch->payload->send_message.send_message));
- calld->send_messages->push_back(cache);
+ calld->send_messages.push_back(cache);
}
// Save metadata batch for send_trailing_metadata ops.
if (batch->send_trailing_metadata) {
@@ -1180,7 +1242,7 @@ static void free_cached_send_message(channel_data* chand, call_data* calld,
"chand=%p calld=%p: destroying calld->send_messages[%" PRIuPTR "]",
chand, calld, idx);
}
- (*calld->send_messages)[idx]->Destroy();
+ calld->send_messages[idx]->Destroy();
}
// Frees cached send_trailing_metadata.
@@ -1650,55 +1712,66 @@ 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,
- bool set_on_complete) {
- call_data* calld = static_cast<call_data*>(elem->call_data);
+namespace {
+subchannel_batch_data::subchannel_batch_data(grpc_call_element* elem,
+ call_data* calld, int refcount,
+ bool set_on_complete)
+ : elem(elem),
+ subchannel_call(GRPC_SUBCHANNEL_CALL_REF(calld->subchannel_call,
+ "batch_data_create")) {
subchannel_call_retry_state* retry_state =
static_cast<subchannel_call_retry_state*>(
grpc_connected_subchannel_call_get_parent_data(
calld->subchannel_call));
- subchannel_batch_data* batch_data = static_cast<subchannel_batch_data*>(
- gpr_arena_alloc(calld->arena, sizeof(*batch_data)));
- batch_data->elem = elem;
- batch_data->subchannel_call =
- 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);
+ batch.payload = &retry_state->batch_payload;
+ gpr_ref_init(&refs, refcount);
if (set_on_complete) {
- GRPC_CLOSURE_INIT(&batch_data->on_complete, on_complete, batch_data,
+ GRPC_CLOSURE_INIT(&on_complete, ::on_complete, this,
grpc_schedule_on_exec_ctx);
- batch_data->batch.on_complete = &batch_data->on_complete;
+ batch.on_complete = &on_complete;
}
GRPC_CALL_STACK_REF(calld->owning_call, "batch_data");
+}
+
+void subchannel_batch_data::destroy() {
+ subchannel_call_retry_state* retry_state =
+ static_cast<subchannel_call_retry_state*>(
+ grpc_connected_subchannel_call_get_parent_data(subchannel_call));
+ if (batch.send_initial_metadata) {
+ grpc_metadata_batch_destroy(&retry_state->send_initial_metadata);
+ }
+ if (batch.send_trailing_metadata) {
+ grpc_metadata_batch_destroy(&retry_state->send_trailing_metadata);
+ }
+ if (batch.recv_initial_metadata) {
+ grpc_metadata_batch_destroy(&retry_state->recv_initial_metadata);
+ }
+ if (batch.recv_trailing_metadata) {
+ grpc_metadata_batch_destroy(&retry_state->recv_trailing_metadata);
+ }
+ GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "batch_data_unref");
+ call_data* calld = static_cast<call_data*>(elem->call_data);
+ GRPC_CALL_STACK_UNREF(calld->owning_call, "batch_data");
+}
+} // namespace
+
+// 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,
+ bool set_on_complete) {
+ call_data* calld = static_cast<call_data*>(elem->call_data);
+ subchannel_batch_data* batch_data =
+ new (gpr_arena_alloc(calld->arena, sizeof(*batch_data)))
+ subchannel_batch_data(elem, calld, refcount, set_on_complete);
return batch_data;
}
static void batch_data_unref(subchannel_batch_data* batch_data) {
if (gpr_unref(&batch_data->refs)) {
- 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->batch.send_trailing_metadata) {
- grpc_metadata_batch_destroy(&retry_state->send_trailing_metadata);
- }
- if (batch_data->batch.recv_initial_metadata) {
- grpc_metadata_batch_destroy(&retry_state->recv_initial_metadata);
- }
- if (batch_data->batch.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);
- GRPC_CALL_STACK_UNREF(calld->owning_call, "batch_data");
+ batch_data->destroy();
}
}
@@ -1996,7 +2069,7 @@ static bool pending_batch_is_unstarted(
return true;
}
if (pending->batch->send_message &&
- retry_state->started_send_message_count < calld->send_messages->size()) {
+ retry_state->started_send_message_count < calld->send_messages.size()) {
return true;
}
if (pending->batch->send_trailing_metadata &&
@@ -2152,7 +2225,7 @@ static void add_closures_for_replay_or_pending_send_ops(
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 =
- retry_state->started_send_message_count < calld->send_messages->size();
+ retry_state->started_send_message_count < calld->send_messages.size();
bool have_pending_send_trailing_metadata_op =
calld->seen_send_trailing_metadata &&
!retry_state->started_send_trailing_metadata;
@@ -2344,7 +2417,7 @@ static void add_retriable_send_message_op(
chand, calld, retry_state->started_send_message_count);
}
grpc_core::ByteStreamCache* cache =
- (*calld->send_messages)[retry_state->started_send_message_count];
+ calld->send_messages[retry_state->started_send_message_count];
++retry_state->started_send_message_count;
retry_state->send_message.Init(cache);
batch_data->batch.send_message = true;
@@ -2476,7 +2549,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay(
}
// send_message.
// Note that we can only have one send_message op in flight at a time.
- if (retry_state->started_send_message_count < calld->send_messages->size() &&
+ if (retry_state->started_send_message_count < calld->send_messages.size() &&
retry_state->started_send_message_count ==
retry_state->completed_send_message_count &&
!calld->pending_send_message) {
@@ -2497,7 +2570,7 @@ static subchannel_batch_data* maybe_create_subchannel_batch_for_replay(
// to start, since we can't send down any more send_message ops after
// send_trailing_metadata.
if (calld->seen_send_trailing_metadata &&
- retry_state->started_send_message_count == calld->send_messages->size() &&
+ retry_state->started_send_message_count == calld->send_messages.size() &&
!retry_state->started_send_trailing_metadata &&
!calld->pending_send_trailing_metadata) {
if (grpc_client_channel_trace.enabled()) {
@@ -2549,7 +2622,7 @@ static void add_subchannel_batches_for_pending_batches(
// send_message ops after send_trailing_metadata.
if (batch->send_trailing_metadata &&
(retry_state->started_send_message_count + batch->send_message <
- calld->send_messages->size() ||
+ calld->send_messages.size() ||
retry_state->started_send_trailing_metadata)) {
continue;
}
@@ -2716,11 +2789,9 @@ static void create_subchannel_call(grpc_call_element* elem, grpc_error* error) {
pending_batches_fail(elem, new_error, true /* yield_call_combiner */);
} else {
if (parent_data_size > 0) {
- subchannel_call_retry_state* retry_state =
- static_cast<subchannel_call_retry_state*>(
- grpc_connected_subchannel_call_get_parent_data(
- calld->subchannel_call));
- retry_state->batch_payload.context = calld->pick.subchannel_call_context;
+ new (grpc_connected_subchannel_call_get_parent_data(
+ calld->subchannel_call))
+ subchannel_call_retry_state(calld->pick.subchannel_call_context);
}
pending_batches_resume(elem);
}
@@ -3260,21 +3331,8 @@ static void cc_start_transport_stream_op_batch(
/* Constructor for call_data */
static grpc_error* cc_init_call_elem(grpc_call_element* elem,
const grpc_call_element_args* args) {
- call_data* calld = static_cast<call_data*>(elem->call_data);
channel_data* chand = static_cast<channel_data*>(elem->channel_data);
- // Initialize data members.
- calld->path = grpc_slice_ref_internal(args->path);
- calld->call_start_time = args->start_time;
- calld->deadline = args->deadline;
- calld->arena = args->arena;
- calld->owning_call = args->call_stack;
- calld->call_combiner = args->call_combiner;
- if (GPR_LIKELY(chand->deadline_checking_enabled)) {
- grpc_deadline_state_init(elem, args->call_stack, args->call_combiner,
- calld->deadline);
- }
- calld->enable_retries = chand->enable_retries;
- calld->send_messages.Init();
+ new (elem->call_data) call_data(elem, *chand, *args);
return GRPC_ERROR_NONE;
}
@@ -3283,34 +3341,12 @@ static void cc_destroy_call_elem(grpc_call_element* elem,
const grpc_call_final_info* final_info,
grpc_closure* then_schedule_closure) {
call_data* calld = static_cast<call_data*>(elem->call_data);
- channel_data* chand = static_cast<channel_data*>(elem->channel_data);
- if (GPR_LIKELY(chand->deadline_checking_enabled)) {
- grpc_deadline_state_destroy(elem);
- }
- grpc_slice_unref_internal(calld->path);
- calld->retry_throttle_data.reset();
- calld->method_params.reset();
- GRPC_ERROR_UNREF(calld->cancel_error);
if (GPR_LIKELY(calld->subchannel_call != nullptr)) {
grpc_subchannel_call_set_cleanup_closure(calld->subchannel_call,
then_schedule_closure);
then_schedule_closure = nullptr;
- GRPC_SUBCHANNEL_CALL_UNREF(calld->subchannel_call,
- "client_channel_destroy_call");
- }
- for (size_t i = 0; i < GPR_ARRAY_SIZE(calld->pending_batches); ++i) {
- GPR_ASSERT(calld->pending_batches[i].batch == nullptr);
- }
- if (GPR_LIKELY(calld->pick.connected_subchannel != nullptr)) {
- calld->pick.connected_subchannel.reset();
- }
- for (size_t i = 0; i < GRPC_CONTEXT_COUNT; ++i) {
- if (calld->pick.subchannel_call_context[i].value != nullptr) {
- calld->pick.subchannel_call_context[i].destroy(
- calld->pick.subchannel_call_context[i].value);
- }
}
- calld->send_messages.Destroy();
+ calld->~call_data();
GRPC_CLOSURE_SCHED(then_schedule_closure, GRPC_ERROR_NONE);
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-09 11:53:35 +0000