aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/transport')
-rw-r--r--src/core/transport/chttp2/internal.h21
-rw-r--r--src/core/transport/chttp2/parsing.c3
-rw-r--r--src/core/transport/chttp2/stream_lists.c68
-rw-r--r--src/core/transport/chttp2/writing.c94
-rw-r--r--src/core/transport/chttp2_transport.c17
-rw-r--r--src/core/transport/transport_op_string.c10
6 files changed, 115 insertions, 98 deletions
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index e7901da510..f0eeb6de50 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -60,7 +60,6 @@ typedef enum {
GRPC_CHTTP2_LIST_WRITABLE,
GRPC_CHTTP2_LIST_WRITING,
GRPC_CHTTP2_LIST_WRITTEN,
- GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE,
GRPC_CHTTP2_LIST_PARSING_SEEN,
GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING,
GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING,
@@ -383,6 +382,8 @@ typedef struct {
gpr_uint8 published_cancelled;
/** is this stream in the stream map? (boolean) */
gpr_uint8 in_stream_map;
+ /** is this stream actively being written? */
+ gpr_uint8 writing_now;
/** stream state already published to the upper layer */
grpc_stream_state published_state;
@@ -475,11 +476,17 @@ void grpc_chttp2_publish_reads(grpc_chttp2_transport_global *global,
void grpc_chttp2_list_add_writable_stream(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global);
+void grpc_chttp2_list_add_first_writable_stream(
+ grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_stream_global *stream_global);
int grpc_chttp2_list_pop_writable_stream(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_transport_writing *transport_writing,
grpc_chttp2_stream_global **stream_global,
grpc_chttp2_stream_writing **stream_writing);
+void grpc_chttp2_list_remove_writable_stream(
+ grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_stream_global *stream_global);
void grpc_chttp2_list_add_incoming_window_updated(
grpc_chttp2_transport_global *transport_global,
@@ -511,18 +518,6 @@ int grpc_chttp2_list_pop_written_stream(
grpc_chttp2_stream_global **stream_global,
grpc_chttp2_stream_writing **stream_writing);
-void grpc_chttp2_list_add_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_global **stream_global,
- grpc_chttp2_stream_writing **stream_writing);
-void grpc_chttp2_list_remove_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-
void grpc_chttp2_list_add_parsing_seen_stream(
grpc_chttp2_transport_parsing *transport_parsing,
grpc_chttp2_stream_parsing *stream_parsing);
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index 50a2f752f6..d84960009b 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -182,8 +182,7 @@ void grpc_chttp2_publish_reads(
stream_global->max_recv_bytes -=
stream_parsing->incoming_window_delta;
stream_parsing->incoming_window_delta = 0;
- grpc_chttp2_list_add_writable_window_update_stream(transport_global,
- stream_global);
+ grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
}
/* update outgoing flow control window */
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index 590f6abfbc..9e68c1e146 100644
--- a/src/core/transport/chttp2/stream_lists.c
+++ b/src/core/transport/chttp2/stream_lists.c
@@ -108,6 +108,23 @@ static void stream_list_maybe_remove(grpc_chttp2_transport *t,
}
}
+static void stream_list_add_head(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ grpc_chttp2_stream_list_id id) {
+ grpc_chttp2_stream *old_head;
+ GPR_ASSERT(!s->included[id]);
+ old_head = t->lists[id].head;
+ s->links[id].next = old_head;
+ s->links[id].prev = NULL;
+ if (old_head) {
+ old_head->links[id].prev = s;
+ } else {
+ t->lists[id].tail = s;
+ }
+ t->lists[id].head = s;
+ s->included[id] = 1;
+}
+
static void stream_list_add_tail(grpc_chttp2_transport *t,
grpc_chttp2_stream *s,
grpc_chttp2_stream_list_id id) {
@@ -119,7 +136,6 @@ static void stream_list_add_tail(grpc_chttp2_transport *t,
if (old_tail) {
old_tail->links[id].next = s;
} else {
- s->links[id].prev = NULL;
t->lists[id].head = s;
}
t->lists[id].tail = s;
@@ -144,6 +160,18 @@ void grpc_chttp2_list_add_writable_stream(
STREAM_FROM_GLOBAL(stream_global), GRPC_CHTTP2_LIST_WRITABLE);
}
+void grpc_chttp2_list_add_first_writable_stream(
+ grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_stream_global *stream_global) {
+ GPR_ASSERT(stream_global->id != 0);
+ gpr_log(GPR_DEBUG, "add:%d:%d:%d:%d", stream_global->id,
+ stream_global->write_state, stream_global->in_stream_map,
+ stream_global->read_closed);
+ stream_list_add_head(TRANSPORT_FROM_GLOBAL(transport_global),
+ STREAM_FROM_GLOBAL(stream_global),
+ GRPC_CHTTP2_LIST_WRITABLE);
+}
+
int grpc_chttp2_list_pop_writable_stream(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_transport_writing *transport_writing,
@@ -157,6 +185,14 @@ int grpc_chttp2_list_pop_writable_stream(
return r;
}
+void grpc_chttp2_list_remove_writable_stream(
+ grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_stream_global *stream_global) {
+ stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
+ STREAM_FROM_GLOBAL(stream_global),
+ GRPC_CHTTP2_LIST_WRITABLE);
+}
+
void grpc_chttp2_list_add_writing_stream(
grpc_chttp2_transport_writing *transport_writing,
grpc_chttp2_stream_writing *stream_writing) {
@@ -202,36 +238,6 @@ int grpc_chttp2_list_pop_written_stream(
return r;
}
-void grpc_chttp2_list_add_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global) {
- GPR_ASSERT(stream_global->id != 0);
- stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
- STREAM_FROM_GLOBAL(stream_global),
- GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
-int grpc_chttp2_list_pop_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_global **stream_global,
- grpc_chttp2_stream_writing **stream_writing) {
- grpc_chttp2_stream *stream;
- int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream,
- GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
- *stream_global = &stream->global;
- *stream_writing = &stream->writing;
- return r;
-}
-
-void grpc_chttp2_list_remove_writable_window_update_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global) {
- stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
- STREAM_FROM_GLOBAL(stream_global),
- GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
void grpc_chttp2_list_add_parsing_seen_stream(
grpc_chttp2_transport_parsing *transport_parsing,
grpc_chttp2_stream_parsing *stream_parsing) {
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index d8ec117aa5..d39b0c42f7 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -44,6 +44,7 @@ int grpc_chttp2_unlocking_check_writes(
grpc_chttp2_transport_writing *transport_writing) {
grpc_chttp2_stream_global *stream_global;
grpc_chttp2_stream_writing *stream_writing;
+ grpc_chttp2_stream_global *first_reinserted_stream = NULL;
gpr_uint32 window_delta;
/* simple writes are queued to qbuf, and flushed here */
@@ -64,50 +65,54 @@ int grpc_chttp2_unlocking_check_writes(
}
/* for each grpc_chttp2_stream that's become writable, frame it's data
- (according to
- available window sizes) and add to the output buffer */
- while (grpc_chttp2_list_pop_writable_stream(transport_global,
- transport_writing, &stream_global,
- &stream_writing)) {
+ (according to available window sizes) and add to the output buffer */
+ while (grpc_chttp2_list_pop_writable_stream(
+ transport_global, transport_writing, &stream_global, &stream_writing)) {
+ if (stream_global == first_reinserted_stream) {
+ /* prevent infinite loop */
+ grpc_chttp2_list_add_first_writable_stream(transport_global,
+ stream_global);
+ break;
+ }
+
stream_writing->id = stream_global->id;
- window_delta = grpc_chttp2_preencode(
- stream_global->outgoing_sopb->ops, &stream_global->outgoing_sopb->nops,
- GPR_MIN(transport_global->outgoing_window,
- stream_global->outgoing_window),
- &stream_writing->sopb);
- GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
- "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
- GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
- outgoing_window, -(gpr_int64)window_delta);
- transport_global->outgoing_window -= window_delta;
- stream_global->outgoing_window -= window_delta;
-
- if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
- stream_global->outgoing_sopb->nops == 0) {
- if (!transport_global->is_client && !stream_global->read_closed) {
- stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
- } else {
- stream_writing->send_closed = GRPC_SEND_CLOSED;
+ stream_writing->send_closed = GRPC_DONT_SEND_CLOSED;
+ GPR_ASSERT(!stream_global->writing_now);
+
+ if (stream_global->outgoing_sopb) {
+ window_delta =
+ grpc_chttp2_preencode(stream_global->outgoing_sopb->ops,
+ &stream_global->outgoing_sopb->nops,
+ GPR_MIN(transport_global->outgoing_window,
+ stream_global->outgoing_window),
+ &stream_writing->sopb);
+ GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
+ "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
+ GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
+ outgoing_window,
+ -(gpr_int64)window_delta);
+ transport_global->outgoing_window -= window_delta;
+ stream_global->outgoing_window -= window_delta;
+
+ if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
+ stream_global->outgoing_sopb->nops == 0) {
+ if (!transport_global->is_client && !stream_global->read_closed) {
+ stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
+ } else {
+ stream_writing->send_closed = GRPC_SEND_CLOSED;
+ }
}
- }
- if (stream_writing->sopb.nops > 0 ||
- stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
- grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
- }
- if (stream_global->outgoing_window > 0 &&
- stream_global->outgoing_sopb->nops != 0) {
- grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+ if (stream_global->outgoing_window > 0 &&
+ stream_global->outgoing_sopb->nops != 0) {
+ grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+ if (first_reinserted_stream == NULL &&
+ transport_global->outgoing_window == 0) {
+ first_reinserted_stream = stream_global;
+ }
+ }
}
- }
- /* for each grpc_chttp2_stream that wants to update its window, add that
- * window here */
- while (grpc_chttp2_list_pop_writable_window_update_stream(transport_global,
- transport_writing,
- &stream_global,
- &stream_writing)) {
- stream_writing->id = stream_global->id;
if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) {
stream_writing->announce_window = stream_global->unannounced_incoming_window;
GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
@@ -118,6 +123,11 @@ int grpc_chttp2_unlocking_check_writes(
stream_global->unannounced_incoming_window = 0;
grpc_chttp2_list_add_incoming_window_updated(transport_global,
stream_global);
+ stream_global->writing_now = 1;
+ grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
+ } else if (stream_writing->sopb.nops > 0 ||
+ stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
+ stream_global->writing_now = 1;
grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
}
}
@@ -205,6 +215,8 @@ void grpc_chttp2_cleanup_writing(
while (grpc_chttp2_list_pop_written_stream(
transport_global, transport_writing, &stream_global, &stream_writing)) {
+ GPR_ASSERT(stream_global->writing_now);
+ stream_global->writing_now = 0;
if (stream_global->outgoing_sopb != NULL &&
stream_global->outgoing_sopb->nops == 0) {
stream_global->outgoing_sopb = NULL;
@@ -216,9 +228,9 @@ void grpc_chttp2_cleanup_writing(
if (!transport_global->is_client) {
stream_global->read_closed = 1;
}
- grpc_chttp2_list_add_read_write_state_changed(transport_global,
- stream_global);
}
+ grpc_chttp2_list_add_read_write_state_changed(transport_global,
+ stream_global);
}
transport_writing->outbuf.count = 0;
transport_writing->outbuf.length = 0;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index eb435a2ee8..5f49b2ddd6 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -395,12 +395,16 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) {
}
grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global);
- grpc_chttp2_list_remove_writable_window_update_stream(&t->global, &s->global);
+ grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
gpr_mu_unlock(&t->mu);
for (i = 0; i < STREAM_LIST_COUNT; i++) {
- GPR_ASSERT(!s->included[i]);
+ if (s->included[i]) {
+ gpr_log(GPR_ERROR, "%s stream %d still included in list %d",
+ t->global.is_client ? "client" : "server", s->global.id, i);
+ abort();
+ }
}
GPR_ASSERT(s->global.outgoing_sopb == NULL);
@@ -576,8 +580,6 @@ static void maybe_start_some_streams(
grpc_chttp2_list_add_incoming_window_updated(transport_global,
stream_global);
grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
- grpc_chttp2_list_add_writable_window_update_stream(transport_global,
- stream_global);
}
/* cancel out streams that will never be started */
@@ -643,8 +645,7 @@ static void perform_stream_op_locked(
if (stream_global->id != 0) {
grpc_chttp2_list_add_read_write_state_changed(transport_global,
stream_global);
- grpc_chttp2_list_add_writable_window_update_stream(transport_global,
- stream_global);
+ grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
}
}
@@ -752,6 +753,7 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) {
if (!s) {
s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id);
}
+ grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
GPR_ASSERT(s);
s->global.in_stream_map = 0;
if (t->parsing.incoming_stream == &s->parsing) {
@@ -833,6 +835,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
if (!stream_global->publish_sopb) {
continue;
}
+ if (stream_global->writing_now) {
+ continue;
+ }
/* FIXME(ctiller): we include in_stream_map in our computation of
whether the stream is write-closed. This is completely bogus,
but has the effect of delaying stream-closed until the stream
diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c
index 10d796fc15..f62c340e97 100644
--- a/src/core/transport/transport_op_string.c
+++ b/src/core/transport/transport_op_string.c
@@ -116,10 +116,9 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) {
if (op->send_ops) {
if (!first) gpr_strvec_add(&b, gpr_strdup(" "));
first = 0;
- gpr_strvec_add(&b, gpr_strdup("SEND"));
- if (op->is_last_send) {
- gpr_strvec_add(&b, gpr_strdup("_LAST"));
- }
+ gpr_asprintf(&tmp, "SEND%s:%p", op->is_last_send ? "_LAST" : "",
+ op->on_done_send);
+ gpr_strvec_add(&b, tmp);
gpr_strvec_add(&b, gpr_strdup("["));
gpr_strvec_add(&b, grpc_sopb_string(op->send_ops));
gpr_strvec_add(&b, gpr_strdup("]"));
@@ -128,7 +127,8 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) {
if (op->recv_ops) {
if (!first) gpr_strvec_add(&b, gpr_strdup(" "));
first = 0;
- gpr_asprintf(&tmp, "RECV:max_recv_bytes=%d", op->max_recv_bytes);
+ gpr_asprintf(&tmp, "RECV:%p:max_recv_bytes=%d", op->on_done_recv,
+ op->max_recv_bytes);
gpr_strvec_add(&b, tmp);
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-25 00:02:07 +0000