aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/transport')
-rw-r--r--src/core/transport/chttp2/hpack_encoder.c4
-rw-r--r--src/core/transport/chttp2/internal.h15
-rw-r--r--src/core/transport/chttp2/stream_lists.c21
-rw-r--r--src/core/transport/chttp2/writing.c20
-rw-r--r--src/core/transport/chttp2_transport.c34
-rw-r--r--src/core/transport/connectivity_state.c12
-rw-r--r--src/core/transport/transport.c10
-rw-r--r--src/core/transport/transport.h15
8 files changed, 86 insertions, 45 deletions
diff --git a/src/core/transport/chttp2/hpack_encoder.c b/src/core/transport/chttp2/hpack_encoder.c
index 89a80d896c..f30f574d06 100644
--- a/src/core/transport/chttp2/hpack_encoder.c
+++ b/src/core/transport/chttp2/hpack_encoder.c
@@ -283,7 +283,7 @@ static void emit_lithdr_incidx(grpc_chttp2_hpack_compressor *c,
len_val_len = GRPC_CHTTP2_VARINT_LENGTH((uint32_t)len_val, 1);
GRPC_CHTTP2_WRITE_VARINT(key_index, 2, 0x40,
add_tiny_header_data(st, len_pfx), len_pfx);
- GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, 0x00,
+ GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, huffman_prefix,
add_tiny_header_data(st, len_val_len), len_val_len);
add_header_data(st, gpr_slice_ref(value_slice));
}
@@ -300,7 +300,7 @@ static void emit_lithdr_noidx(grpc_chttp2_hpack_compressor *c,
len_val_len = GRPC_CHTTP2_VARINT_LENGTH((uint32_t)len_val, 1);
GRPC_CHTTP2_WRITE_VARINT(key_index, 4, 0x00,
add_tiny_header_data(st, len_pfx), len_pfx);
- GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, 0x00,
+ GRPC_CHTTP2_WRITE_VARINT((uint32_t)len_val, 1, huffman_prefix,
add_tiny_header_data(st, len_val_len), len_val_len);
add_header_data(st, gpr_slice_ref(value_slice));
}
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index a8262b7af2..c611496e7e 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -35,6 +35,7 @@
#define GRPC_INTERNAL_CORE_CHTTP2_INTERNAL_H
#include <assert.h>
+#include <stdbool.h>
#include "src/core/iomgr/endpoint.h"
#include "src/core/transport/chttp2/frame.h"
@@ -67,6 +68,9 @@ typedef enum {
GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING,
GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_WRITING,
GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT,
+ /* streams waiting for the outgoing window in the writing path, they will be
+ * merged to the stalled list or writable list under transport lock. */
+ GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT,
/** streams that are waiting to start because there are too many concurrent
streams on the connection */
GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY,
@@ -488,7 +492,7 @@ void grpc_chttp2_perform_writes(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing,
grpc_endpoint *endpoint);
void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing, int success);
+ void *transport_writing, bool success);
void grpc_chttp2_cleanup_writing(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *global,
grpc_chttp2_transport_writing *writing);
@@ -504,11 +508,11 @@ void grpc_chttp2_publish_reads(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *global,
grpc_chttp2_transport_parsing *parsing);
-/** Get a writable stream
- returns non-zero if there was a stream available */
void grpc_chttp2_list_add_writable_stream(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global);
+/** Get a writable stream
+ returns non-zero if there was a stream available */
int grpc_chttp2_list_pop_writable_stream(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_transport_writing *transport_writing,
@@ -560,9 +564,12 @@ int grpc_chttp2_list_pop_check_read_ops(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global **stream_global);
-void grpc_chttp2_list_add_stalled_by_transport(
+void grpc_chttp2_list_add_writing_stalled_by_transport(
grpc_chttp2_transport_writing *transport_writing,
grpc_chttp2_stream_writing *stream_writing);
+void grpc_chttp2_list_flush_writing_stalled_by_transport(
+ grpc_chttp2_transport_writing *transport_writing, bool is_window_available);
+
int grpc_chttp2_list_pop_stalled_by_transport(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global **stream_global);
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index 273a513e2f..2f31a47cb3 100644
--- a/src/core/transport/chttp2/stream_lists.c
+++ b/src/core/transport/chttp2/stream_lists.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -313,12 +313,27 @@ int grpc_chttp2_list_pop_check_read_ops(
return r;
}
-void grpc_chttp2_list_add_stalled_by_transport(
+void grpc_chttp2_list_add_writing_stalled_by_transport(
grpc_chttp2_transport_writing *transport_writing,
grpc_chttp2_stream_writing *stream_writing) {
stream_list_add(TRANSPORT_FROM_WRITING(transport_writing),
STREAM_FROM_WRITING(stream_writing),
- GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT);
+ GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT);
+}
+
+void grpc_chttp2_list_flush_writing_stalled_by_transport(
+ grpc_chttp2_transport_writing *transport_writing,
+ bool is_window_available) {
+ grpc_chttp2_stream *stream;
+ grpc_chttp2_transport *transport = TRANSPORT_FROM_WRITING(transport_writing);
+ while (stream_list_pop(transport, &stream,
+ GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) {
+ if (is_window_available) {
+ grpc_chttp2_list_add_writable_stream(&transport->global, &stream->global);
+ } else {
+ stream_list_add(transport, stream, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT);
+ }
+ }
}
int grpc_chttp2_list_pop_stalled_by_transport(
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index fdad05b5fb..095883c66d 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -130,8 +130,8 @@ int grpc_chttp2_unlocking_check_writes(
GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
}
} else {
- grpc_chttp2_list_add_stalled_by_transport(transport_writing,
- stream_writing);
+ grpc_chttp2_list_add_writing_stalled_by_transport(transport_writing,
+ stream_writing);
}
}
if (stream_global->send_trailing_metadata) {
@@ -188,7 +188,7 @@ void grpc_chttp2_perform_writes(
grpc_endpoint_write(exec_ctx, endpoint, &transport_writing->outbuf,
&transport_writing->done_cb);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, &transport_writing->done_cb, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &transport_writing->done_cb, true, NULL);
}
}
@@ -273,8 +273,8 @@ static void finalize_outbuf(grpc_exec_ctx *exec_ctx,
stream_writing->sent_message = 1;
}
} else if (transport_writing->outgoing_window == 0) {
- grpc_chttp2_list_add_stalled_by_transport(transport_writing,
- stream_writing);
+ grpc_chttp2_list_add_writing_stalled_by_transport(transport_writing,
+ stream_writing);
grpc_chttp2_list_add_written_stream(transport_writing, stream_writing);
}
}
@@ -312,8 +312,8 @@ static void finalize_outbuf(grpc_exec_ctx *exec_ctx,
/* do nothing - already reffed */
}
} else {
- grpc_chttp2_list_add_stalled_by_transport(transport_writing,
- stream_writing);
+ grpc_chttp2_list_add_writing_stalled_by_transport(transport_writing,
+ stream_writing);
grpc_chttp2_list_add_written_stream(transport_writing, stream_writing);
}
} else {
@@ -329,6 +329,10 @@ void grpc_chttp2_cleanup_writing(
grpc_chttp2_transport_writing *transport_writing) {
grpc_chttp2_stream_writing *stream_writing;
grpc_chttp2_stream_global *stream_global;
+ bool is_window_available = transport_writing->outgoing_window > 0;
+
+ grpc_chttp2_list_flush_writing_stalled_by_transport(transport_writing,
+ is_window_available);
while (grpc_chttp2_list_pop_written_stream(
transport_global, transport_writing, &stream_global, &stream_writing)) {
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 05b25fd8b0..9298573c7f 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -86,14 +86,14 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
/* forward declarations of various callbacks that we'll build closures around */
static void writing_action(grpc_exec_ctx *exec_ctx, void *t,
- int iomgr_success_ignored);
+ bool iomgr_success_ignored);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
uint32_t value);
/** Endpoint callback to process incoming data */
-static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success);
+static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success);
/** Start disconnection chain */
static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
@@ -183,7 +183,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
and maybe they hold resources that need to be freed */
while (t->global.pings.next != &t->global.pings) {
grpc_chttp2_outstanding_ping *ping = t->global.pings.next;
- grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, false, NULL);
ping->next->prev = ping->prev;
ping->prev->next = ping->next;
gpr_free(ping);
@@ -602,7 +602,7 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
t->parsing_active)) {
t->writing_active = 1;
REF_TRANSPORT(t, "writing");
- grpc_exec_ctx_enqueue(exec_ctx, &t->writing_action, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &t->writing_action, true, NULL);
prevent_endpoint_shutdown(t);
}
check_read_ops(exec_ctx, &t->global);
@@ -631,7 +631,7 @@ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
}
void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing_ptr, int success) {
+ void *transport_writing_ptr, bool success) {
grpc_chttp2_transport_writing *transport_writing = transport_writing_ptr;
grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing);
grpc_chttp2_stream_global *stream_global;
@@ -669,7 +669,7 @@ void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
}
static void writing_action(grpc_exec_ctx *exec_ctx, void *gt,
- int iomgr_success_ignored) {
+ bool iomgr_success_ignored) {
grpc_chttp2_transport *t = gt;
GPR_TIMER_BEGIN("writing_action", 0);
grpc_chttp2_perform_writes(exec_ctx, &t->writing, t->ep);
@@ -759,7 +759,7 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
closure->final_data |= 1;
}
if (closure->final_data < 2) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, closure->final_data == 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, closure->final_data == 0, NULL);
}
*pclosure = NULL;
}
@@ -777,7 +777,7 @@ static int contains_non_ok_status(
return 0;
}
-static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, int success) {}
+static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, bool success) {}
static void perform_stream_op_locked(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
@@ -934,7 +934,7 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
for (ping = transport_global->pings.next; ping != &transport_global->pings;
ping = ping->next) {
if (0 == memcmp(opaque_8bytes, ping->id, 8)) {
- grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, true, NULL);
ping->next->prev = ping->prev;
ping->prev->next = ping->next;
gpr_free(ping);
@@ -951,7 +951,7 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
lock(t);
- grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
if (op->on_connectivity_state_change != NULL) {
grpc_connectivity_state_notify_on_state_change(
@@ -1022,11 +1022,13 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
*stream_global->recv_message = grpc_chttp2_incoming_frame_queue_pop(
&stream_global->incoming_frames);
GPR_ASSERT(*stream_global->recv_message != NULL);
- grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, true,
+ NULL);
stream_global->recv_message_ready = NULL;
} else if (stream_global->published_trailing_metadata) {
*stream_global->recv_message = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, true,
+ NULL);
stream_global->recv_message_ready = NULL;
}
}
@@ -1336,7 +1338,7 @@ static void read_error_locked(grpc_exec_ctx *exec_ctx,
}
/* tcp read callback */
-static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) {
+static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success) {
size_t i;
int keep_reading = 0;
grpc_chttp2_transport *t = tp;
@@ -1523,7 +1525,7 @@ static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
unlock(exec_ctx, bs->transport);
return 1;
} else if (bs->failed) {
- grpc_exec_ctx_enqueue(exec_ctx, on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, on_complete, false, NULL);
unlock(exec_ctx, bs->transport);
return 0;
} else {
@@ -1552,7 +1554,7 @@ void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&bs->transport->mu);
if (bs->on_next != NULL) {
*bs->next = slice;
- grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, true, NULL);
bs->on_next = NULL;
} else {
gpr_slice_buffer_add(&bs->slices, slice);
@@ -1567,7 +1569,7 @@ void grpc_chttp2_incoming_byte_stream_finished(
if (from_parsing_thread) {
gpr_mu_lock(&bs->transport->mu);
}
- grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, false, NULL);
bs->on_next = NULL;
bs->failed = 1;
if (from_parsing_thread) {
diff --git a/src/core/transport/connectivity_state.c b/src/core/transport/connectivity_state.c
index 3c3fd4671d..87765b9799 100644
--- a/src/core/transport/connectivity_state.c
+++ b/src/core/transport/connectivity_state.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -78,7 +78,7 @@ void grpc_connectivity_state_destroy(grpc_exec_ctx *exec_ctx,
} else {
success = 0;
}
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, success);
+ grpc_exec_ctx_enqueue(exec_ctx, w->notify, success, NULL);
gpr_free(w);
}
gpr_free(tracker->name);
@@ -109,7 +109,7 @@ int grpc_connectivity_state_notify_on_state_change(
if (current == NULL) {
grpc_connectivity_state_watcher *w = tracker->watchers;
if (w != NULL && w->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
tracker->watchers = w->next;
gpr_free(w);
return 0;
@@ -117,7 +117,7 @@ int grpc_connectivity_state_notify_on_state_change(
while (w != NULL) {
grpc_connectivity_state_watcher *rm_candidate = w->next;
if (rm_candidate != NULL && rm_candidate->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
w->next = w->next->next;
gpr_free(rm_candidate);
return 0;
@@ -128,7 +128,7 @@ int grpc_connectivity_state_notify_on_state_change(
} else {
if (tracker->current_state != *current) {
*current = tracker->current_state;
- grpc_exec_ctx_enqueue(exec_ctx, notify, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, true, NULL);
} else {
grpc_connectivity_state_watcher *w = gpr_malloc(sizeof(*w));
w->current = current;
@@ -158,7 +158,7 @@ void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
while ((w = tracker->watchers) != NULL) {
*w->current = tracker->current_state;
tracker->watchers = w->next;
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, w->notify, true, NULL);
gpr_free(w);
}
}
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 2ab978be46..08d685668c 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -59,7 +59,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx,
grpc_stream_refcount *refcount) {
#endif
if (gpr_unref(&refcount->refs)) {
- grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, true, NULL);
}
}
@@ -125,8 +125,8 @@ char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx,
void grpc_transport_stream_op_finish_with_failure(
grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) {
- grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0);
- grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, false, NULL);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, false, NULL);
}
void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
@@ -150,7 +150,7 @@ typedef struct {
grpc_closure closure;
} close_message_data;
-static void free_message(grpc_exec_ctx *exec_ctx, void *p, int iomgr_success) {
+static void free_message(grpc_exec_ctx *exec_ctx, void *p, bool iomgr_success) {
close_message_data *cmd = p;
gpr_slice_unref(cmd->message);
if (cmd->then_call != NULL) {
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index f94f0ae76e..f5cac77adc 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -81,16 +81,29 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount);
/* Transport stream op: a set of operations to perform on a transport
against a single stream */
typedef struct grpc_transport_stream_op {
+ /** Send initial metadata to the peer, from the provided metadata batch. */
grpc_metadata_batch *send_initial_metadata;
+
+ /** Send trailing metadata to the peer, from the provided metadata batch. */
grpc_metadata_batch *send_trailing_metadata;
+ /** Send message data to the peer, from the provided byte stream. */
grpc_byte_stream *send_message;
+ /** Receive initial metadata from the stream, into provided metadata batch. */
grpc_metadata_batch *recv_initial_metadata;
+
+ /** Receive message data from the stream, into provided byte stream. */
grpc_byte_stream **recv_message;
+ /** Should be enqueued when one message is ready to be processed. */
grpc_closure *recv_message_ready;
+
+ /** Receive trailing metadata from the stream, into provided metadata batch.
+ */
grpc_metadata_batch *recv_trailing_metadata;
+ /** Should be enqueued when all requested operations (excluding recv_message
+ which has its own closure) in a given batch have been completed. */
grpc_closure *on_complete;
/** If != GRPC_STATUS_OK, cancel this stream */
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-05 04:31:55 +0000