aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/surface/call.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/surface/call.cc')
-rw-r--r--src/core/lib/surface/call.cc176
1 files changed, 96 insertions, 80 deletions
diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc
index a9349afa68..735a78ad08 100644
--- a/src/core/lib/surface/call.cc
+++ b/src/core/lib/surface/call.cc
@@ -72,8 +72,11 @@
// Used to create arena for the first call.
#define ESTIMATED_MDELEM_COUNT 16
-typedef struct batch_control {
- grpc_call* call;
+struct batch_control {
+ batch_control() { gpr_ref_init(&steps_to_complete, 0); }
+
+ grpc_call* call = nullptr;
+ grpc_transport_stream_op_batch op;
/* Share memory for cq_completion and notify_tag as they are never needed
simultaneously. Each byte used in this data structure count as six bytes
per call, so any savings we can make are worthwhile,
@@ -96,84 +99,110 @@ typedef struct batch_control {
grpc_closure start_batch;
grpc_closure finish_batch;
gpr_refcount steps_to_complete;
- gpr_atm batch_error;
- grpc_transport_stream_op_batch op;
-} batch_control;
+ gpr_atm batch_error = reinterpret_cast<gpr_atm>(GRPC_ERROR_NONE);
+};
+
+struct parent_call {
+ parent_call() { gpr_mu_init(&child_list_mu); }
+ ~parent_call() { gpr_mu_destroy(&child_list_mu); }
-typedef struct {
gpr_mu child_list_mu;
- grpc_call* first_child;
-} parent_call;
+ grpc_call* first_child = nullptr;
+};
-typedef struct {
+struct child_call {
+ child_call(grpc_call* parent) : parent(parent) {}
grpc_call* parent;
/** siblings: children of the same parent form a list, and this list is
protected under
parent->mu */
- grpc_call* sibling_next;
- grpc_call* sibling_prev;
-} child_call;
+ grpc_call* sibling_next = nullptr;
+ grpc_call* sibling_prev = nullptr;
+};
#define RECV_NONE ((gpr_atm)0)
#define RECV_INITIAL_METADATA_FIRST ((gpr_atm)1)
struct grpc_call {
+ grpc_call(gpr_arena* arena, const grpc_call_create_args& args)
+ : arena(arena),
+ cq(args.cq),
+ channel(args.channel),
+ is_client(args.server_transport_data == nullptr),
+ stream_op_payload(context) {
+ gpr_ref_init(&ext_ref, 1);
+ grpc_call_combiner_init(&call_combiner);
+ for (int i = 0; i < 2; i++) {
+ for (int j = 0; j < 2; j++) {
+ metadata_batch[i][j].deadline = GRPC_MILLIS_INF_FUTURE;
+ }
+ }
+ }
+
+ ~grpc_call() {
+ gpr_free(static_cast<void*>(const_cast<char*>(final_info.error_string)));
+ grpc_call_combiner_destroy(&call_combiner);
+ }
+
gpr_refcount ext_ref;
gpr_arena* arena;
grpc_call_combiner call_combiner;
grpc_completion_queue* cq;
grpc_polling_entity pollent;
grpc_channel* channel;
- gpr_timespec start_time;
- /* parent_call* */ gpr_atm parent_call_atm;
- child_call* child;
+ gpr_timespec start_time = gpr_now(GPR_CLOCK_MONOTONIC);
+ /* parent_call* */ gpr_atm parent_call_atm = 0;
+ child_call* child = nullptr;
/* client or server call */
bool is_client;
/** has grpc_call_unref been called */
- bool destroy_called;
+ bool destroy_called = false;
/** flag indicating that cancellation is inherited */
- bool cancellation_is_inherited;
+ bool cancellation_is_inherited = false;
/** which ops are in-flight */
- bool sent_initial_metadata;
- bool sending_message;
- bool sent_final_op;
- bool received_initial_metadata;
- bool receiving_message;
- bool requested_final_op;
- gpr_atm any_ops_sent_atm;
- gpr_atm received_final_op_atm;
-
- batch_control* active_batches[MAX_CONCURRENT_BATCHES];
+ bool sent_initial_metadata = false;
+ bool sending_message = false;
+ bool sent_final_op = false;
+ bool received_initial_metadata = false;
+ bool receiving_message = false;
+ bool requested_final_op = false;
+ gpr_atm any_ops_sent_atm = 0;
+ gpr_atm received_final_op_atm = 0;
+
+ batch_control* active_batches[MAX_CONCURRENT_BATCHES] = {};
grpc_transport_stream_op_batch_payload stream_op_payload;
/* first idx: is_receiving, second idx: is_trailing */
- grpc_metadata_batch metadata_batch[2][2];
+ grpc_metadata_batch metadata_batch[2][2] = {};
/* Buffered read metadata waiting to be returned to the application.
Element 0 is initial metadata, element 1 is trailing metadata. */
- grpc_metadata_array* buffered_metadata[2];
+ grpc_metadata_array* buffered_metadata[2] = {};
grpc_metadata compression_md;
// A char* indicating the peer name.
- gpr_atm peer_string;
+ gpr_atm peer_string = 0;
/* Call data useful used for reporting. Only valid after the call has
* completed */
grpc_call_final_info final_info;
/* Compression algorithm for *incoming* data */
- grpc_message_compression_algorithm incoming_message_compression_algorithm;
+ grpc_message_compression_algorithm incoming_message_compression_algorithm =
+ GRPC_MESSAGE_COMPRESS_NONE;
/* Stream compression algorithm for *incoming* data */
- grpc_stream_compression_algorithm incoming_stream_compression_algorithm;
- /* Supported encodings (compression algorithms), a bitset */
- uint32_t encodings_accepted_by_peer;
+ grpc_stream_compression_algorithm incoming_stream_compression_algorithm =
+ GRPC_STREAM_COMPRESS_NONE;
+ /* Supported encodings (compression algorithms), a bitset.
+ * Always support no compression. */
+ uint32_t encodings_accepted_by_peer = 1 << GRPC_MESSAGE_COMPRESS_NONE;
/* Supported stream encodings (stream compression algorithms), a bitset */
- uint32_t stream_encodings_accepted_by_peer;
+ uint32_t stream_encodings_accepted_by_peer = 0;
/* Contexts for various subsystems (security, tracing, ...). */
- grpc_call_context_element context[GRPC_CONTEXT_COUNT];
+ grpc_call_context_element context[GRPC_CONTEXT_COUNT] = {};
/* for the client, extra metadata is initial metadata; for the
server, it's trailing metadata */
@@ -184,14 +213,14 @@ struct grpc_call {
grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream> sending_stream;
grpc_core::OrphanablePtr<grpc_core::ByteStream> receiving_stream;
- grpc_byte_buffer** receiving_buffer;
- grpc_slice receiving_slice;
+ grpc_byte_buffer** receiving_buffer = nullptr;
+ grpc_slice receiving_slice = grpc_empty_slice();
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;
- gpr_atm cancelled;
+ uint32_t test_only_last_message_flags = 0;
+ gpr_atm cancelled = 0;
grpc_closure release_call;
@@ -207,7 +236,7 @@ struct grpc_call {
grpc_server* server;
} server;
} final_op;
- gpr_atm status_error;
+ gpr_atm status_error = 0;
/* recv_state can contain one of the following values:
RECV_NONE : : no initial metadata and messages received
@@ -225,7 +254,7 @@ struct grpc_call {
For 1, 4: See receiving_initial_metadata_ready() function
For 2, 3: See receiving_stream_ready() function */
- gpr_atm recv_state;
+ gpr_atm recv_state = 0;
};
grpc_core::TraceFlag grpc_call_error_trace(false, "call_error");
@@ -269,11 +298,10 @@ void* grpc_call_arena_alloc(grpc_call* call, size_t size) {
static parent_call* get_or_create_parent_call(grpc_call* call) {
parent_call* p = (parent_call*)gpr_atm_acq_load(&call->parent_call_atm);
if (p == nullptr) {
- p = static_cast<parent_call*>(gpr_arena_alloc(call->arena, sizeof(*p)));
- gpr_mu_init(&p->child_list_mu);
+ p = new (gpr_arena_alloc(call->arena, sizeof(*p))) parent_call();
if (!gpr_atm_rel_cas(&call->parent_call_atm, (gpr_atm) nullptr,
(gpr_atm)p)) {
- gpr_mu_destroy(&p->child_list_mu);
+ p->~parent_call();
p = (parent_call*)gpr_atm_acq_load(&call->parent_call_atm);
}
}
@@ -292,7 +320,9 @@ size_t grpc_call_get_initial_size_estimate() {
grpc_error* grpc_call_create(const grpc_call_create_args* args,
grpc_call** out_call) {
GPR_TIMER_SCOPE("grpc_call_create", 0);
- size_t i, j;
+
+ GRPC_CHANNEL_INTERNAL_REF(args->channel, "call");
+
grpc_error* error = GRPC_ERROR_NONE;
grpc_channel_stack* channel_stack =
grpc_channel_get_channel_stack(args->channel);
@@ -300,27 +330,19 @@ 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, GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)) +
- channel_stack->call_stack_size));
- gpr_ref_init(&call->ext_ref, 1);
- gpr_atm_no_barrier_store(&call->cancelled, 0);
- call->arena = arena;
- grpc_call_combiner_init(&call->call_combiner);
+ call = new (gpr_arena_alloc(
+ arena, GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)) +
+ channel_stack->call_stack_size)) grpc_call(arena, *args);
*out_call = call;
- call->channel = args->channel;
- call->cq = args->cq;
- call->start_time = gpr_now(GPR_CLOCK_MONOTONIC);
- /* Always support no compression */
- GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_MESSAGE_COMPRESS_NONE);
- call->is_client = args->server_transport_data == nullptr;
- call->stream_op_payload.context = call->context;
grpc_slice path = grpc_empty_slice();
if (call->is_client) {
+ call->final_op.client.status_details = nullptr;
+ call->final_op.client.status = nullptr;
+ call->final_op.client.error_string = nullptr;
GRPC_STATS_INC_CLIENT_CALLS_CREATED();
GPR_ASSERT(args->add_initial_metadata_count <
MAX_SEND_EXTRA_METADATA_COUNT);
- for (i = 0; i < args->add_initial_metadata_count; i++) {
+ for (size_t i = 0; i < args->add_initial_metadata_count; i++) {
call->send_extra_metadata[i].md = args->add_initial_metadata[i];
if (grpc_slice_eq(GRPC_MDKEY(args->add_initial_metadata[i]),
GRPC_MDSTR_PATH)) {
@@ -332,23 +354,18 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
static_cast<int>(args->add_initial_metadata_count);
} else {
GRPC_STATS_INC_SERVER_CALLS_CREATED();
+ call->final_op.server.cancelled = nullptr;
call->final_op.server.server = args->server;
GPR_ASSERT(args->add_initial_metadata_count == 0);
call->send_extra_metadata_count = 0;
}
- for (i = 0; i < 2; i++) {
- for (j = 0; j < 2; j++) {
- call->metadata_batch[i][j].deadline = GRPC_MILLIS_INF_FUTURE;
- }
- }
- grpc_millis send_deadline = args->send_deadline;
+ grpc_millis send_deadline = args->send_deadline;
bool immediately_cancel = false;
if (args->parent != nullptr) {
- call->child =
- static_cast<child_call*>(gpr_arena_alloc(arena, sizeof(child_call)));
- call->child->parent = args->parent;
+ call->child = new (gpr_arena_alloc(arena, sizeof(child_call)))
+ child_call(args->parent);
GRPC_CALL_INTERNAL_REF(args->parent, "child");
GPR_ASSERT(call->is_client);
@@ -382,10 +399,7 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
}
}
}
-
call->send_deadline = send_deadline;
-
- GRPC_CHANNEL_INTERNAL_REF(args->channel, "call");
/* initial refcount dropped by grpc_call_unref */
grpc_call_element_args call_args = {CALL_STACK_FROM_CALL(call),
args->server_transport_data,
@@ -413,6 +427,7 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
}
gpr_mu_unlock(&pc->child_list_mu);
}
+
if (error != GRPC_ERROR_NONE) {
cancel_with_error(call, GRPC_ERROR_REF(error));
}
@@ -487,9 +502,9 @@ void grpc_call_internal_unref(grpc_call* c REF_ARG) {
static void release_call(void* call, grpc_error* error) {
grpc_call* c = static_cast<grpc_call*>(call);
grpc_channel* channel = c->channel;
- gpr_free(static_cast<void*>(const_cast<char*>(c->final_info.error_string)));
- grpc_call_combiner_destroy(&c->call_combiner);
- grpc_channel_update_call_size_estimate(channel, gpr_arena_destroy(c->arena));
+ gpr_arena* arena = c->arena;
+ c->~grpc_call();
+ grpc_channel_update_call_size_estimate(channel, gpr_arena_destroy(arena));
GRPC_CHANNEL_INTERNAL_UNREF(channel, "call");
}
@@ -505,7 +520,7 @@ static void destroy_call(void* call, grpc_error* error) {
c->receiving_stream.reset();
parent_call* pc = get_parent_call(c);
if (pc != nullptr) {
- gpr_mu_destroy(&pc->child_list_mu);
+ pc->~parent_call();
}
for (ii = 0; ii < c->send_extra_metadata_count; ii++) {
GRPC_MDELEM_UNREF(c->send_extra_metadata[ii].md);
@@ -1100,10 +1115,11 @@ static batch_control* reuse_or_allocate_batch_control(grpc_call* call,
if (bctl->call != nullptr) {
return nullptr;
}
- memset(bctl, 0, sizeof(*bctl));
+ bctl->~batch_control();
+ bctl->op = {};
} else {
- bctl = static_cast<batch_control*>(
- gpr_arena_alloc(call->arena, sizeof(batch_control)));
+ bctl = new (gpr_arena_alloc(call->arena, sizeof(batch_control)))
+ batch_control();
*pslot = bctl;
}
bctl->call = call;
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-09 19:19:03 +0000