Skip to content

Commit 3137d59

Browse files
srini100srini100
authored
Merge pull request grpc#15697 from srini100/backport15690
Backport grpc#15690 from master to v1.13.x
2 parents 37e0ab7 + c6664b0 commit 3137d59

15 files changed

Lines changed: 607 additions & 669 deletions

File tree

‎src/core/ext/filters/client_channel/client_channel.cc‎

Lines changed: 498 additions & 376 deletions
Large diffs are not rendered by default.

‎src/core/ext/filters/deadline/deadline_filter.cc‎

Lines changed: 12 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -128,25 +128,21 @@ static void cancel_timer_if_needed(grpc_deadline_state* deadline_state) {
128128
}
129129
}
130130

131-
// Callback run when we receive trailing metadata.
132-
static void recv_trailing_metadata_ready(void* arg, grpc_error* error) {
131+
// Callback run when the call is complete.
132+
static void on_complete(void* arg, grpc_error* error) {
133133
grpc_deadline_state* deadline_state = static_cast<grpc_deadline_state*>(arg);
134134
cancel_timer_if_needed(deadline_state);
135-
// Invoke the original callback.
136-
GRPC_CLOSURE_RUN(deadline_state->original_recv_trailing_metadata_ready,
137-
GRPC_ERROR_REF(error));
135+
// Invoke the next callback.
136+
GRPC_CLOSURE_RUN(deadline_state->next_on_complete, GRPC_ERROR_REF(error));
138137
}
139138

140-
// Inject our own recv_trailing_metadata_ready callback into op.
141-
static void inject_recv_trailing_metadata_ready(
142-
grpc_deadline_state* deadline_state, grpc_transport_stream_op_batch* op) {
143-
deadline_state->original_recv_trailing_metadata_ready =
144-
op->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
145-
GRPC_CLOSURE_INIT(&deadline_state->recv_trailing_metadata_ready,
146-
recv_trailing_metadata_ready, deadline_state,
139+
// Inject our own on_complete callback into op.
140+
static void inject_on_complete_cb(grpc_deadline_state* deadline_state,
141+
grpc_transport_stream_op_batch* op) {
142+
deadline_state->next_on_complete = op->on_complete;
143+
GRPC_CLOSURE_INIT(&deadline_state->on_complete, on_complete, deadline_state,
147144
grpc_schedule_on_exec_ctx);
148-
op->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
149-
&deadline_state->recv_trailing_metadata_ready;
145+
op->on_complete = &deadline_state->on_complete;
150146
}
151147

152148
// Callback and associated state for starting the timer after call stack
@@ -230,7 +226,7 @@ void grpc_deadline_state_client_start_transport_stream_op_batch(
230226
// Make sure we know when the call is complete, so that we can cancel
231227
// the timer.
232228
if (op->recv_trailing_metadata) {
233-
inject_recv_trailing_metadata_ready(deadline_state, op);
229+
inject_on_complete_cb(deadline_state, op);
234230
}
235231
}
236232
}
@@ -327,7 +323,7 @@ static void server_start_transport_stream_op_batch(
327323
// the client never sends trailing metadata, because this is the
328324
// hook that tells us when the call is complete on the server side.
329325
if (op->recv_trailing_metadata) {
330-
inject_recv_trailing_metadata_ready(&calld->base.deadline_state, op);
326+
inject_on_complete_cb(&calld->base.deadline_state, op);
331327
}
332328
}
333329
// Chain to next filter.

‎src/core/ext/filters/deadline/deadline_filter.h‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,12 @@ typedef struct grpc_deadline_state {
3737
grpc_deadline_timer_state timer_state;
3838
grpc_timer timer;
3939
grpc_closure timer_callback;
40-
// Closure to invoke when we receive trailing metadata.
40+
// Closure to invoke when the call is complete.
4141
// We use this to cancel the timer.
42-
grpc_closure recv_trailing_metadata_ready;
43-
// The original recv_trailing_metadata_ready closure, which we chain to
44-
// after our own closure is invoked.
45-
grpc_closure* original_recv_trailing_metadata_ready;
42+
grpc_closure on_complete;
43+
// The original on_complete closure, which we chain to after our own
44+
// closure is invoked.
45+
grpc_closure* next_on_complete;
4646
} grpc_deadline_state;
4747

4848
//

‎src/core/ext/filters/http/client/http_client_filter.cc‎

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -55,8 +55,8 @@ struct call_data {
5555
grpc_closure recv_initial_metadata_ready;
5656
// State for handling recv_trailing_metadata ops.
5757
grpc_metadata_batch* recv_trailing_metadata;
58-
grpc_closure* original_recv_trailing_metadata_ready;
59-
grpc_closure recv_trailing_metadata_ready;
58+
grpc_closure* original_recv_trailing_metadata_on_complete;
59+
grpc_closure recv_trailing_metadata_on_complete;
6060
// State for handling send_message ops.
6161
grpc_transport_stream_op_batch* send_message_batch;
6262
size_t send_message_bytes_read;
@@ -153,7 +153,8 @@ static void recv_initial_metadata_ready(void* user_data, grpc_error* error) {
153153
GRPC_CLOSURE_RUN(calld->original_recv_initial_metadata_ready, error);
154154
}
155155

156-
static void recv_trailing_metadata_ready(void* user_data, grpc_error* error) {
156+
static void recv_trailing_metadata_on_complete(void* user_data,
157+
grpc_error* error) {
157158
grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
158159
call_data* calld = static_cast<call_data*>(elem->call_data);
159160
if (error == GRPC_ERROR_NONE) {
@@ -162,7 +163,7 @@ static void recv_trailing_metadata_ready(void* user_data, grpc_error* error) {
162163
} else {
163164
GRPC_ERROR_REF(error);
164165
}
165-
GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_ready, error);
166+
GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_on_complete, error);
166167
}
167168

168169
static void send_message_on_complete(void* arg, grpc_error* error) {
@@ -311,10 +312,8 @@ static void hc_start_transport_stream_op_batch(
311312
/* substitute our callback for the higher callback */
312313
calld->recv_trailing_metadata =
313314
batch->payload->recv_trailing_metadata.recv_trailing_metadata;
314-
calld->original_recv_trailing_metadata_ready =
315-
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
316-
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
317-
&calld->recv_trailing_metadata_ready;
315+
calld->original_recv_trailing_metadata_on_complete = batch->on_complete;
316+
batch->on_complete = &calld->recv_trailing_metadata_on_complete;
318317
}
319318

320319
grpc_error* error = GRPC_ERROR_NONE;
@@ -421,8 +420,8 @@ static grpc_error* init_call_elem(grpc_call_element* elem,
421420
GRPC_CLOSURE_INIT(&calld->recv_initial_metadata_ready,
422421
recv_initial_metadata_ready, elem,
423422
grpc_schedule_on_exec_ctx);
424-
GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready,
425-
recv_trailing_metadata_ready, elem,
423+
GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_on_complete,
424+
recv_trailing_metadata_on_complete, elem,
426425
grpc_schedule_on_exec_ctx);
427426
GRPC_CLOSURE_INIT(&calld->send_message_on_complete, send_message_on_complete,
428427
elem, grpc_schedule_on_exec_ctx);

‎src/core/ext/transport/chttp2/transport/chttp2_transport.cc‎

Lines changed: 19 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1149,10 +1149,12 @@ static void maybe_start_some_streams(grpc_chttp2_transport* t) {
11491149
}
11501150
}
11511151

1152+
/* Flag that this closure barrier wants stats to be updated before finishing */
1153+
#define CLOSURE_BARRIER_STATS_BIT (1 << 0)
11521154
/* Flag that this closure barrier may be covering a write in a pollset, and so
11531155
we should not complete this closure until we can prove that the write got
11541156
scheduled */
1155-
#define CLOSURE_BARRIER_MAY_COVER_WRITE (1 << 0)
1157+
#define CLOSURE_BARRIER_MAY_COVER_WRITE (1 << 1)
11561158
/* First bit of the reference count, stored in the high order bits (with the low
11571159
bits being used for flags defined above) */
11581160
#define CLOSURE_BARRIER_FIRST_REF_BIT (1 << 16)
@@ -1204,6 +1206,10 @@ void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t,
12041206
grpc_error_add_child(closure->error_data.error, error);
12051207
}
12061208
if (closure->next_data.scratch < CLOSURE_BARRIER_FIRST_REF_BIT) {
1209+
if (closure->next_data.scratch & CLOSURE_BARRIER_STATS_BIT) {
1210+
grpc_transport_move_stats(&s->stats, s->collecting_stats);
1211+
s->collecting_stats = nullptr;
1212+
}
12071213
if ((t->write_state == GRPC_CHTTP2_WRITE_STATE_IDLE) ||
12081214
!(closure->next_data.scratch & CLOSURE_BARRIER_MAY_COVER_WRITE)) {
12091215
GRPC_CLOSURE_RUN(closure, closure->error_data.error);
@@ -1345,21 +1351,22 @@ static void perform_stream_op_locked(void* stream_op,
13451351
}
13461352

13471353
grpc_closure* on_complete = op->on_complete;
1348-
// TODO(roth): This is a hack needed because we use data inside of the
1349-
// closure itself to do the barrier calculation (i.e., to ensure that
1350-
// we don't schedule the closure until all ops in the batch have been
1351-
// completed). This can go away once we move to a new C++ closure API
1352-
// that provides the ability to create a barrier closure.
13531354
if (on_complete == nullptr) {
1354-
on_complete = GRPC_CLOSURE_INIT(&op->handler_private.closure, do_nothing,
1355-
nullptr, grpc_schedule_on_exec_ctx);
1355+
on_complete =
1356+
GRPC_CLOSURE_CREATE(do_nothing, nullptr, grpc_schedule_on_exec_ctx);
13561357
}
13571358

13581359
/* use final_data as a barrier until enqueue time; the inital counter is
13591360
dropped at the end of this function */
13601361
on_complete->next_data.scratch = CLOSURE_BARRIER_FIRST_REF_BIT;
13611362
on_complete->error_data.error = GRPC_ERROR_NONE;
13621363

1364+
if (op->collect_stats) {
1365+
GPR_ASSERT(s->collecting_stats == nullptr);
1366+
s->collecting_stats = op_payload->collect_stats.collect_stats;
1367+
on_complete->next_data.scratch |= CLOSURE_BARRIER_STATS_BIT;
1368+
}
1369+
13631370
if (op->cancel_stream) {
13641371
GRPC_STATS_INC_HTTP2_OP_CANCEL();
13651372
grpc_chttp2_cancel_stream(t, s, op_payload->cancel_stream.cancel_error);
@@ -1593,11 +1600,8 @@ static void perform_stream_op_locked(void* stream_op,
15931600

15941601
if (op->recv_trailing_metadata) {
15951602
GRPC_STATS_INC_HTTP2_OP_RECV_TRAILING_METADATA();
1596-
GPR_ASSERT(s->collecting_stats == nullptr);
1597-
s->collecting_stats = op_payload->recv_trailing_metadata.collect_stats;
15981603
GPR_ASSERT(s->recv_trailing_metadata_finished == nullptr);
1599-
s->recv_trailing_metadata_finished =
1600-
op_payload->recv_trailing_metadata.recv_trailing_metadata_ready;
1604+
s->recv_trailing_metadata_finished = add_closure_barrier(on_complete);
16011605
s->recv_trailing_metadata =
16021606
op_payload->recv_trailing_metadata.recv_trailing_metadata;
16031607
s->final_metadata_requested = true;
@@ -1956,12 +1960,11 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_chttp2_transport* t,
19561960
}
19571961
if (s->read_closed && s->frame_storage.length == 0 && !pending_data &&
19581962
s->recv_trailing_metadata_finished != nullptr) {
1959-
grpc_transport_move_stats(&s->stats, s->collecting_stats);
1960-
s->collecting_stats = nullptr;
19611963
grpc_chttp2_incoming_metadata_buffer_publish(&s->metadata_buffer[1],
19621964
s->recv_trailing_metadata);
1963-
null_then_run_closure(&s->recv_trailing_metadata_finished,
1964-
GRPC_ERROR_NONE);
1965+
grpc_chttp2_complete_closure_step(
1966+
t, s, &s->recv_trailing_metadata_finished, GRPC_ERROR_NONE,
1967+
"recv_trailing_metadata_finished");
19651968
}
19661969
}
19671970
}

‎src/core/ext/transport/cronet/transport/cronet_transport.cc‎

Lines changed: 1 addition & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -925,10 +925,6 @@ static bool op_can_be_run(grpc_transport_stream_op_batch* curr_op,
925925
result = false;
926926
}
927927
/* Check if every op that was asked for is done. */
928-
/* TODO(muxi): We should not consider the recv ops here, since they
929-
* have their own callbacks. We should invoke a batch's on_complete
930-
* as soon as all of the batch's send ops are complete, even if
931-
* there are still recv ops pending. */
932928
else if (curr_op->send_initial_metadata &&
933929
!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) {
934930
CRONET_LOG(GPR_DEBUG, "Because");
@@ -1284,20 +1280,12 @@ static enum e_op_result execute_stream_op(struct op_and_state* oas) {
12841280
op_can_be_run(stream_op, s, &oas->state,
12851281
OP_RECV_TRAILING_METADATA)) {
12861282
CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_TRAILING_METADATA", oas);
1287-
grpc_error* error = GRPC_ERROR_NONE;
1288-
if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
1289-
error = GRPC_ERROR_REF(stream_state->cancel_error);
1290-
} else if (stream_state->state_op_done[OP_FAILED]) {
1291-
error = make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable.");
1292-
} else if (oas->s->state.rs.trailing_metadata_valid) {
1283+
if (oas->s->state.rs.trailing_metadata_valid) {
12931284
grpc_chttp2_incoming_metadata_buffer_publish(
12941285
&oas->s->state.rs.trailing_metadata,
12951286
stream_op->payload->recv_trailing_metadata.recv_trailing_metadata);
12961287
stream_state->rs.trailing_metadata_valid = false;
12971288
}
1298-
GRPC_CLOSURE_SCHED(
1299-
stream_op->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
1300-
error);
13011289
stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
13021290
result = ACTION_TAKEN_NO_CALLBACK;
13031291
} else if (stream_op->cancel_stream &&
@@ -1410,11 +1398,6 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs,
14101398
GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready,
14111399
GRPC_ERROR_CANCELLED);
14121400
}
1413-
if (op->recv_trailing_metadata) {
1414-
GRPC_CLOSURE_SCHED(
1415-
op->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
1416-
GRPC_ERROR_CANCELLED);
1417-
}
14181401
GRPC_CLOSURE_SCHED(op->on_complete, GRPC_ERROR_CANCELLED);
14191402
return;
14201403
}

‎src/core/ext/transport/inproc/inproc_transport.cc‎

Lines changed: 6 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,7 @@ typedef struct inproc_stream {
120120
struct inproc_stream* stream_list_next;
121121
} inproc_stream;
122122

123+
static grpc_closure do_nothing_closure;
123124
static bool cancel_stream_locked(inproc_stream* s, grpc_error* error);
124125
static void op_state_machine(void* arg, grpc_error* error);
125126

@@ -372,10 +373,6 @@ static void complete_if_batch_end_locked(inproc_stream* s, grpc_error* error,
372373
const char* msg) {
373374
int is_sm = static_cast<int>(op == s->send_message_op);
374375
int is_stm = static_cast<int>(op == s->send_trailing_md_op);
375-
// TODO(vjpai): We should not consider the recv ops here, since they
376-
// have their own callbacks. We should invoke a batch's on_complete
377-
// as soon as all of the batch's send ops are complete, even if there
378-
// are still recv ops pending.
379376
int is_rim = static_cast<int>(op == s->recv_initial_md_op);
380377
int is_rm = static_cast<int>(op == s->recv_message_op);
381378
int is_rtm = static_cast<int>(op == s->recv_trailing_md_op);
@@ -499,11 +496,6 @@ static void fail_helper_locked(inproc_stream* s, grpc_error* error) {
499496
s->send_trailing_md_op = nullptr;
500497
}
501498
if (s->recv_trailing_md_op) {
502-
INPROC_LOG(GPR_INFO, "fail_helper %p scheduling trailing-metadata-ready %p",
503-
s, error);
504-
GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata
505-
.recv_trailing_metadata_ready,
506-
GRPC_ERROR_REF(error));
507499
INPROC_LOG(GPR_INFO, "fail_helper %p scheduling trailing-md-on-complete %p",
508500
s, error);
509501
complete_if_batch_end_locked(
@@ -646,12 +638,6 @@ static void op_state_machine(void* arg, grpc_error* error) {
646638
}
647639
s->trailing_md_sent = true;
648640
if (!s->t->is_client && s->trailing_md_recvd && s->recv_trailing_md_op) {
649-
INPROC_LOG(GPR_INFO,
650-
"op_state_machine %p scheduling trailing-metadata-ready", s);
651-
GRPC_CLOSURE_SCHED(
652-
s->recv_trailing_md_op->payload->recv_trailing_metadata
653-
.recv_trailing_metadata_ready,
654-
GRPC_ERROR_NONE);
655641
INPROC_LOG(GPR_INFO,
656642
"op_state_machine %p scheduling trailing-md-on-complete", s);
657643
GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->on_complete,
@@ -725,12 +711,6 @@ static void op_state_machine(void* arg, grpc_error* error) {
725711
}
726712
if (s->recv_trailing_md_op && s->t->is_client && other &&
727713
other->send_message_op) {
728-
INPROC_LOG(GPR_INFO,
729-
"op_state_machine %p scheduling trailing-metadata-ready %p", s,
730-
GRPC_ERROR_NONE);
731-
GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata
732-
.recv_trailing_metadata_ready,
733-
GRPC_ERROR_NONE);
734714
maybe_schedule_op_closure_locked(other, GRPC_ERROR_NONE);
735715
}
736716
if (s->to_read_trailing_md_filled) {
@@ -786,10 +766,6 @@ static void op_state_machine(void* arg, grpc_error* error) {
786766
INPROC_LOG(GPR_INFO,
787767
"op_state_machine %p scheduling trailing-md-on-complete %p",
788768
s, new_err);
789-
GRPC_CLOSURE_SCHED(
790-
s->recv_trailing_md_op->payload->recv_trailing_metadata
791-
.recv_trailing_metadata_ready,
792-
GRPC_ERROR_REF(new_err));
793769
GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->on_complete,
794770
GRPC_ERROR_REF(new_err));
795771
s->recv_trailing_md_op = nullptr;
@@ -883,9 +859,6 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) {
883859
// couldn't complete that because we hadn't yet sent out trailing
884860
// md, now's the chance
885861
if (!s->t->is_client && s->trailing_md_recvd && s->recv_trailing_md_op) {
886-
GRPC_CLOSURE_SCHED(s->recv_trailing_md_op->payload->recv_trailing_metadata
887-
.recv_trailing_metadata_ready,
888-
GRPC_ERROR_REF(s->cancel_self_error));
889862
complete_if_batch_end_locked(
890863
s, s->cancel_self_error, s->recv_trailing_md_op,
891864
"cancel_stream scheduling trailing-md-on-complete");
@@ -900,8 +873,6 @@ static bool cancel_stream_locked(inproc_stream* s, grpc_error* error) {
900873
return ret;
901874
}
902875

903-
static void do_nothing(void* arg, grpc_error* error) {}
904-
905876
static void perform_stream_op(grpc_transport* gt, grpc_stream* gs,
906877
grpc_transport_stream_op_batch* op) {
907878
INPROC_LOG(GPR_INFO, "perform_stream_op %p %p %p", gt, gs, op);
@@ -921,14 +892,8 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs,
921892
}
922893
grpc_error* error = GRPC_ERROR_NONE;
923894
grpc_closure* on_complete = op->on_complete;
924-
// TODO(roth): This is a hack needed because we use data inside of the
925-
// closure itself to do the barrier calculation (i.e., to ensure that
926-
// we don't schedule the closure until all ops in the batch have been
927-
// completed). This can go away once we move to a new C++ closure API
928-
// that provides the ability to create a barrier closure.
929895
if (on_complete == nullptr) {
930-
on_complete = GRPC_CLOSURE_INIT(&op->handler_private.closure, do_nothing,
931-
nullptr, grpc_schedule_on_exec_ctx);
896+
on_complete = &do_nothing_closure;
932897
}
933898

934899
if (op->cancel_stream) {
@@ -1061,15 +1026,6 @@ static void perform_stream_op(grpc_transport* gt, grpc_stream* gs,
10611026
GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready,
10621027
GRPC_ERROR_REF(error));
10631028
}
1064-
if (op->recv_trailing_metadata) {
1065-
INPROC_LOG(
1066-
GPR_INFO,
1067-
"perform_stream_op error %p scheduling trailing-metadata-ready %p",
1068-
s, error);
1069-
GRPC_CLOSURE_SCHED(
1070-
op->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
1071-
GRPC_ERROR_REF(error));
1072-
}
10731029
}
10741030
INPROC_LOG(GPR_INFO, "perform_stream_op %p scheduling on_complete %p", s,
10751031
error);
@@ -1173,8 +1129,12 @@ static grpc_endpoint* get_endpoint(grpc_transport* t) { return nullptr; }
11731129
/*******************************************************************************
11741130
* GLOBAL INIT AND DESTROY
11751131
*/
1132+
static void do_nothing(void* arg, grpc_error* error) {}
1133+
11761134
void grpc_inproc_transport_init(void) {
11771135
grpc_core::ExecCtx exec_ctx;
1136+
GRPC_CLOSURE_INIT(&do_nothing_closure, do_nothing, nullptr,
1137+
grpc_schedule_on_exec_ctx);
11781138
g_empty_slice = grpc_slice_from_static_buffer(nullptr, 0);
11791139

11801140
grpc_slice key_tmp = grpc_slice_from_static_string(":path");

0 commit comments

Comments
 (0)