Skip to content

Commit f268d71

Browse files
kpayson64kpayson64
authored
Merge pull request grpc#15983 from kpayson64/write_stalling_fix
Write stalling fix
2 parents e36ca9e + 5c1e47a commit f268d71

3 files changed

Lines changed: 40 additions & 5 deletions

File tree

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

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -813,7 +813,11 @@ static void set_write_state(grpc_chttp2_transport* t,
813813
write_state_name(st), reason));
814814
t->write_state = st;
815815
if (st == GRPC_CHTTP2_WRITE_STATE_IDLE) {
816-
GRPC_CLOSURE_LIST_SCHED(&t->run_after_write);
816+
grpc_chttp2_stream* s;
817+
while (grpc_chttp2_list_pop_waiting_for_write_stream(t, &s)) {
818+
GRPC_CLOSURE_LIST_SCHED(&s->run_after_write);
819+
GRPC_CHTTP2_STREAM_UNREF(s, "chttp2:write_closure_sched");
820+
}
817821
if (t->close_transport_on_writes_finished != nullptr) {
818822
grpc_error* err = t->close_transport_on_writes_finished;
819823
t->close_transport_on_writes_finished = nullptr;
@@ -1208,7 +1212,10 @@ void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t,
12081212
!(closure->next_data.scratch & CLOSURE_BARRIER_MAY_COVER_WRITE)) {
12091213
GRPC_CLOSURE_RUN(closure, closure->error_data.error);
12101214
} else {
1211-
grpc_closure_list_append(&t->run_after_write, closure,
1215+
if (grpc_chttp2_list_add_waiting_for_write_stream(t, s)) {
1216+
GRPC_CHTTP2_STREAM_REF(s, "chttp2:pending_write_closure");
1217+
}
1218+
grpc_closure_list_append(&s->run_after_write, closure,
12121219
closure->error_data.error);
12131220
}
12141221
}
@@ -2009,6 +2016,10 @@ static void remove_stream(grpc_chttp2_transport* t, uint32_t id,
20092016

20102017
void grpc_chttp2_cancel_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s,
20112018
grpc_error* due_to_error) {
2019+
GRPC_CLOSURE_LIST_SCHED(&s->run_after_write);
2020+
if (grpc_chttp2_list_remove_waiting_for_write_stream(t, s)) {
2021+
GRPC_CHTTP2_STREAM_UNREF(s, "chttp2:pending_write_closure");
2022+
}
20122023
if (!t->is_client && !s->sent_trailing_metadata &&
20132024
grpc_error_has_clear_grpc_status(due_to_error)) {
20142025
close_from_api(t, s, due_to_error);

‎src/core/ext/transport/chttp2/transport/internal.h‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,8 @@ typedef enum {
5454
/** streams that are waiting to start because there are too many concurrent
5555
streams on the connection */
5656
GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY,
57+
/** streams with closures waiting to be run on a write **/
58+
GRPC_CHTTP2_LIST_WAITING_FOR_WRITE,
5759
STREAM_LIST_COUNT /* must be last */
5860
} grpc_chttp2_stream_list_id;
5961

@@ -431,9 +433,6 @@ struct grpc_chttp2_transport {
431433
*/
432434
grpc_error* close_transport_on_writes_finished;
433435

434-
/* a list of closures to run after writes are finished */
435-
grpc_closure_list run_after_write;
436-
437436
/* buffer pool state */
438437
/** have we scheduled a benign cleanup? */
439438
bool benign_reclaimer_registered;
@@ -584,6 +583,7 @@ struct grpc_chttp2_stream {
584583

585584
grpc_slice_buffer flow_controlled_buffer;
586585

586+
grpc_closure_list run_after_write;
587587
grpc_chttp2_write_cb* on_flow_controlled_cbs;
588588
grpc_chttp2_write_cb* on_write_finished_cbs;
589589
grpc_chttp2_write_cb* finish_after_write;
@@ -686,6 +686,13 @@ bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport* t,
686686
bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport* t,
687687
grpc_chttp2_stream* s);
688688

689+
bool grpc_chttp2_list_add_waiting_for_write_stream(grpc_chttp2_transport* t,
690+
grpc_chttp2_stream* s);
691+
bool grpc_chttp2_list_pop_waiting_for_write_stream(grpc_chttp2_transport* t,
692+
grpc_chttp2_stream** s);
693+
bool grpc_chttp2_list_remove_waiting_for_write_stream(grpc_chttp2_transport* t,
694+
grpc_chttp2_stream* s);
695+
689696
/********* Flow Control ***************/
690697

691698
// Takes in a flow control action and performs all the needed operations.

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

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@ static const char* stream_list_id_string(grpc_chttp2_stream_list_id id) {
3535
return "stalled_by_stream";
3636
case GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY:
3737
return "waiting_for_concurrency";
38+
case GRPC_CHTTP2_LIST_WAITING_FOR_WRITE:
39+
return "waiting_for_write";
3840
case STREAM_LIST_COUNT:
3941
GPR_UNREACHABLE_CODE(return "unknown");
4042
}
@@ -214,3 +216,18 @@ bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport* t,
214216
grpc_chttp2_stream* s) {
215217
return stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM);
216218
}
219+
220+
bool grpc_chttp2_list_add_waiting_for_write_stream(grpc_chttp2_transport* t,
221+
grpc_chttp2_stream* s) {
222+
return stream_list_add(t, s, GRPC_CHTTP2_LIST_WAITING_FOR_WRITE);
223+
}
224+
225+
bool grpc_chttp2_list_pop_waiting_for_write_stream(grpc_chttp2_transport* t,
226+
grpc_chttp2_stream** s) {
227+
return stream_list_pop(t, s, GRPC_CHTTP2_LIST_WAITING_FOR_WRITE);
228+
}
229+
230+
bool grpc_chttp2_list_remove_waiting_for_write_stream(grpc_chttp2_transport* t,
231+
grpc_chttp2_stream* s) {
232+
return stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_WAITING_FOR_WRITE);
233+
}

0 commit comments

Comments
 (0)