aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext
diff options
context:
space:
mode:
authorGravatar Mark D. Roth <roth@google.com>2016-09-16 13:04:49 -0700
committerGravatar Mark D. Roth <roth@google.com>2016-09-16 13:04:49 -0700
commit999b49bf39bbc7764b50c50548d0f61280f873e2 (patch)
tree0a922710fde9dcf2610065fc84e3fffd215cdce4 /src/core/ext
parentfef585f1be2dcc03ee813d1083a50e4ca0ba8486 (diff)
parentc1ac8df181c766cd72866e2bc559aca817120da2 (diff)
Merge branch 'grpclb_resolver_changes' into grpclb_resolver_changes2
Diffstat (limited to 'src/core/ext')
-rw-r--r--src/core/ext/client_config/client_channel.c14
-rw-r--r--src/core/ext/client_config/lb_policy_factory.c3
-rw-r--r--src/core/ext/client_config/subchannel.c18
-rw-r--r--src/core/ext/lb_policy/grpclb/grpclb.c3
-rw-r--r--src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c2
-rw-r--r--src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h2
-rw-r--r--src/core/ext/transport/chttp2/transport/chttp2_transport.c766
-rw-r--r--src/core/ext/transport/chttp2/transport/internal.h54
-rw-r--r--src/core/ext/transport/chttp2/transport/parsing.c12
-rw-r--r--src/core/ext/transport/chttp2/transport/stream_lists.c9
-rw-r--r--src/core/ext/transport/chttp2/transport/writing.c21
11 files changed, 396 insertions, 508 deletions
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index 1696c005c8..e4500cc50d 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -399,7 +399,7 @@ typedef struct client_channel_call_data {
grpc_connected_subchannel *connected_subchannel;
grpc_polling_entity *pollent;
- grpc_transport_stream_op *waiting_ops;
+ grpc_transport_stream_op **waiting_ops;
size_t waiting_ops_count;
size_t waiting_ops_capacity;
@@ -418,7 +418,7 @@ static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
gpr_realloc(calld->waiting_ops,
calld->waiting_ops_capacity * sizeof(*calld->waiting_ops));
}
- calld->waiting_ops[calld->waiting_ops_count++] = *op;
+ calld->waiting_ops[calld->waiting_ops_count++] = op;
GPR_TIMER_END("add_waiting_locked", 0);
}
@@ -427,14 +427,14 @@ static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld,
size_t i;
for (i = 0; i < calld->waiting_ops_count; i++) {
grpc_transport_stream_op_finish_with_failure(
- exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error));
+ exec_ctx, calld->waiting_ops[i], GRPC_ERROR_REF(error));
}
calld->waiting_ops_count = 0;
GRPC_ERROR_UNREF(error);
}
typedef struct {
- grpc_transport_stream_op *ops;
+ grpc_transport_stream_op **ops;
size_t nops;
grpc_subchannel_call *call;
} retry_ops_args;
@@ -443,7 +443,7 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
retry_ops_args *a = args;
size_t i;
for (i = 0; i < a->nops; i++) {
- grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]);
+ grpc_subchannel_call_process_op(exec_ctx, a->call, a->ops[i]);
}
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops");
gpr_free(a->ops);
@@ -451,6 +451,10 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
}
static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
+ if (calld->waiting_ops_count == 0) {
+ return;
+ }
+
retry_ops_args *a = gpr_malloc(sizeof(*a));
a->ops = calld->waiting_ops;
a->nops = calld->waiting_ops_count;
diff --git a/src/core/ext/client_config/lb_policy_factory.c b/src/core/ext/client_config/lb_policy_factory.c
index 2489505791..8ce8745504 100644
--- a/src/core/ext/client_config/lb_policy_factory.c
+++ b/src/core/ext/client_config/lb_policy_factory.c
@@ -81,8 +81,9 @@ void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses,
void (*user_data_destroy)(void*)) {
for (size_t i = 0; i < addresses->num_addresses; ++i) {
gpr_free(addresses->addresses[i].balancer_name);
- if (user_data_destroy != NULL)
+ if (user_data_destroy != NULL) {
user_data_destroy(addresses->addresses[i].user_data);
+ }
}
gpr_free(addresses->addresses);
gpr_free(addresses);
diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c
index df35904b85..2c4364b259 100644
--- a/src/core/ext/client_config/subchannel.c
+++ b/src/core/ext/client_config/subchannel.c
@@ -504,14 +504,13 @@ static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *interested_parties,
grpc_connectivity_state *state,
grpc_closure *closure) {
- grpc_transport_op op;
+ grpc_transport_op *op = grpc_make_transport_op(NULL);
grpc_channel_element *elem;
- memset(&op, 0, sizeof(op));
- op.connectivity_state = state;
- op.on_connectivity_state_change = closure;
- op.bind_pollset_set = interested_parties;
+ op->connectivity_state = state;
+ op->on_connectivity_state_change = closure;
+ op->bind_pollset_set = interested_parties;
elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
- elem->filter->start_transport_op(exec_ctx, elem, &op);
+ elem->filter->start_transport_op(exec_ctx, elem, op);
}
void grpc_connected_subchannel_notify_on_state_change(
@@ -525,12 +524,11 @@ void grpc_connected_subchannel_notify_on_state_change(
void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx,
grpc_connected_subchannel *con,
grpc_closure *closure) {
- grpc_transport_op op;
+ grpc_transport_op *op = grpc_make_transport_op(NULL);
grpc_channel_element *elem;
- memset(&op, 0, sizeof(op));
- op.send_ping = closure;
+ op->send_ping = closure;
elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
- elem->filter->start_transport_op(exec_ctx, elem, &op);
+ elem->filter->start_transport_op(exec_ctx, elem, op);
}
static void publish_transport_locked(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c
index 999122fbe0..36db8ab00d 100644
--- a/src/core/ext/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/lb_policy/grpclb/grpclb.c
@@ -354,8 +354,7 @@ static bool is_server_valid(const grpc_grpclb_server *server, size_t idx,
return true;
}
-/* populate \a addresses according to \a serverlist. Returns the number of
- * addresses successfully parsed and added to \a addresses */
+/* Returns addresses extracted from \a serverlist. */
static grpc_lb_addresses *process_serverlist(
const grpc_grpclb_serverlist *serverlist) {
size_t num_valid = 0;
diff --git a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
index 2676714175..afecb716fb 100644
--- a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
+++ b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
@@ -74,7 +74,7 @@ const pb_field_t grpc_lb_v1_LoadBalanceResponse_fields[3] = {
const pb_field_t grpc_lb_v1_InitialLoadBalanceResponse_fields[3] = {
PB_FIELD( 1, STRING , OPTIONAL, STATIC , FIRST, grpc_lb_v1_InitialLoadBalanceResponse, load_balancer_delegate, load_balancer_delegate, 0),
- PB_FIELD( 3, MESSAGE , OPTIONAL, STATIC , OTHER, grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval, load_balancer_delegate, &grpc_lb_v1_Duration_fields),
+ PB_FIELD( 2, MESSAGE , OPTIONAL, STATIC , OTHER, grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval, load_balancer_delegate, &grpc_lb_v1_Duration_fields),
PB_LAST_FIELD
};
diff --git a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
index 4f1031ec7b..53fed22bae 100644
--- a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
+++ b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
@@ -146,7 +146,7 @@ typedef struct _grpc_lb_v1_LoadBalanceResponse {
#define grpc_lb_v1_Server_load_balance_token_tag 3
#define grpc_lb_v1_Server_drop_request_tag 4
#define grpc_lb_v1_InitialLoadBalanceResponse_load_balancer_delegate_tag 1
-#define grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval_tag 3
+#define grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval_tag 2
#define grpc_lb_v1_LoadBalanceRequest_initial_request_tag 1
#define grpc_lb_v1_LoadBalanceRequest_client_stats_tag 2
#define grpc_lb_v1_ServerList_servers_tag 1
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 00999e3b94..53da0e5d70 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -89,13 +89,20 @@ static const grpc_transport_vtable vtable;
static void writing_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error);
static void reading_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error);
static void parsing_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error);
-static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t,
- grpc_error *error);
+static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error);
+static void post_parse_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error);
+static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, void *t,
+ grpc_error *error);
+static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *t,
+ grpc_error *error);
+static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *t,
+ grpc_error *error);
static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error,
- const char *reason);
+ grpc_chttp2_transport *t, grpc_error *error);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -105,11 +112,6 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_error *error);
-/** Perform a transport_op */
-static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s, void *transport_op);
-
/** Cancel a stream: coming from the transport API */
static void cancel_from_api(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *transport_global,
@@ -121,22 +123,10 @@ static void close_from_api(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_global *stream_global,
grpc_error *error);
-/** Add endpoint from this transport to pollset */
-static void add_to_pollset_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_ignored, void *pollset);
-static void add_to_pollset_set_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_ignored,
- void *pollset_set);
-
/** Start new streams that have been created if we can */
static void maybe_start_some_streams(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global);
-static void finish_global_actions(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t);
-
static void connectivity_state_set(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
grpc_connectivity_state state, grpc_error *error, const char *reason);
@@ -149,14 +139,16 @@ static void incoming_byte_stream_update_flow_control(
grpc_chttp2_stream_global *stream_global, size_t max_size_hint,
size_t have_already);
static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *byte_stream);
+ void *byte_stream,
+ grpc_error *error_ignored);
static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global,
grpc_error *error);
+static void set_write_state(grpc_chttp2_transport *t,
+ grpc_chttp2_write_state state, const char *reason);
+
/*******************************************************************************
* CONSTRUCTION/DESTRUCTION/REFCOUNTING
*/
@@ -165,9 +157,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
size_t i;
- gpr_mu_lock(&t->executor.mu);
-
- GPR_ASSERT(t->ep == NULL);
+ grpc_endpoint_destroy(exec_ctx, t->ep);
gpr_slice_buffer_destroy(&t->global.qbuf);
@@ -191,8 +181,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_map_destroy(&t->new_stream_map);
grpc_connectivity_state_destroy(exec_ctx, &t->channel_callback.state_tracker);
- gpr_mu_unlock(&t->executor.mu);
- gpr_mu_destroy(&t->executor.mu);
+ grpc_combiner_destroy(exec_ctx, t->executor.combiner);
/* callback remaining pings: they're not allowed to call into the transpot,
and maybe they hold resources that need to be freed */
@@ -250,12 +239,13 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
memset(t, 0, sizeof(*t));
t->base.vtable = &vtable;
+ t->executor.write_state = GRPC_CHTTP2_WRITES_CORKED;
t->ep = ep;
- /* one ref is for destroy, the other for when ep becomes NULL */
- gpr_ref_init(&t->refs, 2);
+ /* one ref is for destroy */
+ gpr_ref_init(&t->refs, 1);
/* ref is dropped at transport close() */
gpr_ref_init(&t->shutdown_ep_refs, 1);
- gpr_mu_init(&t->executor.mu);
+ t->executor.combiner = grpc_combiner_create(grpc_endpoint_get_workqueue(ep));
t->peer_string = grpc_endpoint_get_peer(ep);
t->endpoint_reading = 1;
t->global.next_stream_id = is_client ? 1 : 2;
@@ -281,23 +271,22 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_chttp2_hpack_compressor_init(&t->writing.hpack_compressor);
grpc_closure_init(&t->writing_action, writing_action, t);
grpc_closure_init(&t->reading_action, reading_action, t);
+ grpc_closure_init(&t->reading_action_locked, reading_action_locked, t);
grpc_closure_init(&t->parsing_action, parsing_action, t);
- grpc_closure_init(&t->initiate_writing, initiate_writing, t);
+ grpc_closure_init(&t->post_parse_locked, post_parse_locked, t);
+ grpc_closure_init(&t->initiate_writing, initiate_writing_locked, t);
+ grpc_closure_init(&t->terminate_writing, terminate_writing_with_lock, t);
+ grpc_closure_init(&t->initiate_read_flush_locked, initiate_read_flush_locked,
+ t);
+ grpc_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing,
+ &t->writing);
gpr_slice_buffer_init(&t->parsing.qbuf);
grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser);
grpc_chttp2_hpack_parser_init(&t->parsing.hpack_parser);
- grpc_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing,
- &t->writing);
gpr_slice_buffer_init(&t->read_buffer);
- if (is_client) {
- gpr_slice_buffer_add(
- &t->global.qbuf,
- gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING));
- grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "initial_write");
- }
/* 8 is a random stab in the dark as to a good initial size: it's small enough
that it shouldn't waste memory for infrequently used connections, yet
large enough that the exponential growth should happen nicely when it's
@@ -320,6 +309,13 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
t->global.force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
t->global.sent_local_settings = 0;
+ if (is_client) {
+ gpr_slice_buffer_add(
+ &t->writing.outbuf,
+ gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING));
+ grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "initial_write");
+ }
+
/* configure http2 the way we like it */
if (is_client) {
push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_ENABLE_PUSH, 0);
@@ -424,47 +420,39 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
}
}
}
+
+ set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "uncork");
+ grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "init");
}
-static void destroy_transport_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_ignored,
- void *arg_ignored) {
+static void destroy_transport_locked(grpc_exec_ctx *exec_ctx, void *tp,
+ grpc_error *error) {
+ grpc_chttp2_transport *t = tp;
t->destroying = 1;
drop_connection(exec_ctx, t, GRPC_ERROR_CREATE("Transport destroyed"));
+ UNREF_TRANSPORT(exec_ctx, t, "destroy");
}
static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
- grpc_chttp2_run_with_global_lock(exec_ctx, t, NULL, destroy_transport_locked,
- NULL, 0);
- UNREF_TRANSPORT(exec_ctx, t, "destroy");
+ grpc_combiner_execute(exec_ctx, t->executor.combiner,
+ grpc_closure_create(destroy_transport_locked, t),
+ GRPC_ERROR_NONE);
}
/** block grpc_endpoint_shutdown being called until a paired
allow_endpoint_shutdown is made */
static void prevent_endpoint_shutdown(grpc_chttp2_transport *t) {
- GPR_ASSERT(t->ep);
gpr_ref(&t->shutdown_ep_refs);
}
static void allow_endpoint_shutdown_locked(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
if (gpr_unref(&t->shutdown_ep_refs)) {
- if (t->ep) {
- grpc_endpoint_shutdown(exec_ctx, t->ep);
- }
+ grpc_endpoint_shutdown(exec_ctx, t->ep);
}
}
-static void destroy_endpoint(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t) {
- grpc_endpoint_destroy(exec_ctx, t->ep);
- t->ep = NULL;
- /* safe because we'll still have the ref for write */
- UNREF_TRANSPORT(exec_ctx, t, "disconnect");
-}
-
static void close_transport_locked(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t,
grpc_error *error) {
@@ -475,9 +463,7 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
t->closed = 1;
connectivity_state_set(exec_ctx, &t->global, GRPC_CHANNEL_SHUTDOWN,
GRPC_ERROR_REF(error), "close_transport");
- if (t->ep) {
- allow_endpoint_shutdown_locked(exec_ctx, t);
- }
+ allow_endpoint_shutdown_locked(exec_ctx, t);
/* flush writable stream list to avoid dangling references */
grpc_chttp2_stream_global *stream_global;
@@ -511,21 +497,23 @@ void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
}
#endif
-static void finish_init_stream_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *arg_ignored) {
- grpc_chttp2_register_stream(t, s);
+static void finish_init_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
+ grpc_error *error) {
+ grpc_chttp2_stream *s = sp;
+ grpc_chttp2_register_stream(s->t, s);
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "init");
}
static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_stream_refcount *refcount,
const void *server_data) {
+ GPR_TIMER_BEGIN("init_stream", 0);
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
memset(s, 0, sizeof(*s));
+ s->t = t;
s->refcount = refcount;
/* We reserve one 'active stream' that's dropped when the stream is
read-closed. The others are for incoming_byte_streams that are actively
@@ -560,16 +548,21 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
s->global.in_stream_map = true;
}
- grpc_chttp2_run_with_global_lock(exec_ctx, t, s, finish_init_stream_locked,
- NULL, 0);
+ grpc_closure_init(&s->init_stream, finish_init_stream_locked, s);
+ GRPC_CHTTP2_STREAM_REF(&s->global, "init");
+ grpc_combiner_execute(exec_ctx, t->executor.combiner, &s->init_stream,
+ GRPC_ERROR_NONE);
+
+ GPR_TIMER_END("init_stream", 0);
return 0;
}
-static void destroy_stream_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s, void *arg) {
+static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
+ grpc_error *error) {
grpc_byte_stream *bs;
+ grpc_chttp2_stream *s = sp;
+ grpc_chttp2_transport *t = s->t;
GPR_TIMER_BEGIN("destroy_stream", 0);
@@ -588,7 +581,7 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx,
while (
(bs = grpc_chttp2_incoming_frame_queue_pop(&s->global.incoming_frames))) {
- incoming_byte_stream_destroy_locked(exec_ctx, NULL, NULL, bs);
+ incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
grpc_chttp2_list_remove_unannounced_incoming_window_available(&t->global,
@@ -625,16 +618,20 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx,
GPR_TIMER_END("destroy_stream", 0);
- gpr_free(arg);
+ gpr_free(s->destroy_stream_arg);
}
static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, void *and_free_memory) {
+ GPR_TIMER_BEGIN("destroy_stream", 0);
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
- grpc_chttp2_run_with_global_lock(exec_ctx, t, s, destroy_stream_locked,
- and_free_memory, 0);
+ s->destroy_stream_arg = and_free_memory;
+ grpc_closure_init(&s->destroy_stream, destroy_stream_locked, s);
+ grpc_combiner_execute(exec_ctx, t->executor.combiner, &s->destroy_stream,
+ GRPC_ERROR_NONE);
+ GPR_TIMER_END("destroy_stream", 0);
}
grpc_chttp2_stream_parsing *grpc_chttp2_parsing_lookup_stream(
@@ -665,12 +662,10 @@ grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream(
static const char *write_state_name(grpc_chttp2_write_state state) {
switch (state) {
+ case GRPC_CHTTP2_WRITES_CORKED:
+ return "CORKED";
case GRPC_CHTTP2_WRITING_INACTIVE:
return "INACTIVE";
- case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER:
- return "REQUESTED[p=0]";
- case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER:
- return "REQUESTED[p=1]";
case GRPC_CHTTP2_WRITE_SCHEDULED:
return "SCHEDULED";
case GRPC_CHTTP2_WRITING:
@@ -693,120 +688,18 @@ static void set_write_state(grpc_chttp2_transport *t,
t->executor.write_state = state;
}
-static void finish_global_actions(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t) {
- grpc_chttp2_executor_action_header *hdr;
- grpc_chttp2_executor_action_header *next;
-
- GPR_TIMER_BEGIN("finish_global_actions", 0);
-
- for (;;) {
- check_read_ops(exec_ctx, &t->global);
-
- gpr_mu_lock(&t->executor.mu);
- if (t->executor.pending_actions_head != NULL) {
- hdr = t->executor.pending_actions_head;
- t->executor.pending_actions_head = t->executor.pending_actions_tail =
- NULL;
- gpr_mu_unlock(&t->executor.mu);
- while (hdr != NULL) {
- GPR_TIMER_BEGIN("chttp2:locked_action", 0);
- hdr->action(exec_ctx, t, hdr->stream, hdr->arg);
- GPR_TIMER_END("chttp2:locked_action", 0);
- next = hdr->next;
- gpr_free(hdr);
- UNREF_TRANSPORT(exec_ctx, t, "pending_action");
- hdr = next;
- }
- continue;
- } else {
- t->executor.global_active = false;
- switch (t->executor.write_state) {
- case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER:
- set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "unlocking");
- REF_TRANSPORT(t, "initiate_writing");
- gpr_mu_unlock(&t->executor.mu);
- grpc_exec_ctx_sched(
- exec_ctx, &t->initiate_writing, GRPC_ERROR_NONE,
- t->ep != NULL ? grpc_endpoint_get_workqueue(t->ep) : NULL);
- break;
- case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER:
- start_writing(exec_ctx, t);
- gpr_mu_unlock(&t->executor.mu);
- break;
- case GRPC_CHTTP2_WRITING_INACTIVE:
- case GRPC_CHTTP2_WRITING:
- case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER:
- case GRPC_CHTTP2_WRITING_STALE_NO_POLLER:
- case GRPC_CHTTP2_WRITE_SCHEDULED:
- gpr_mu_unlock(&t->executor.mu);
- break;
- }
- }
- break;
- }
-
- GPR_TIMER_END("finish_global_actions", 0);
+static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, void *tp,
+ grpc_error *error) {
+ grpc_chttp2_transport *t = tp;
+ GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED);
+ start_writing(exec_ctx, t);
}
-void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *optional_stream,
- grpc_chttp2_locked_action action,
- void *arg, size_t sizeof_arg) {
- grpc_chttp2_executor_action_header *hdr;
-
- GPR_TIMER_BEGIN("grpc_chttp2_run_with_global_lock", 0);
-
- REF_TRANSPORT(t, "run_global");
- gpr_mu_lock(&t->executor.mu);
-
- for (;;) {
- if (!t->executor.global_active) {
- t->executor.global_active = 1;
- gpr_mu_unlock(&t->executor.mu);
-
- GPR_TIMER_BEGIN("chttp2:locked_action", 0);
- action(exec_ctx, t, optional_stream, arg);
- GPR_TIMER_END("chttp2:locked_action", 0);
-
- finish_global_actions(exec_ctx, t);
- } else {
- gpr_mu_unlock(&t->executor.mu);
-
- hdr = gpr_malloc(sizeof(*hdr) + sizeof_arg);
- hdr->stream = optional_stream;
- hdr->action = action;
- if (sizeof_arg == 0) {
- hdr->arg = arg;
- } else {
- hdr->arg = hdr + 1;
- memcpy(hdr->arg, arg, sizeof_arg);
- }
-
- gpr_mu_lock(&t->executor.mu);
- if (!t->executor.global_active) {
- /* global lock was released while allocating memory: release & retry */
- gpr_free(hdr);
- continue;
- }
- hdr->next = NULL;
- if (t->executor.pending_actions_head != NULL) {
- t->executor.pending_actions_tail =
- t->executor.pending_actions_tail->next = hdr;
- } else {
- t->executor.pending_actions_tail = t->executor.pending_actions_head =
- hdr;
- }
- REF_TRANSPORT(t, "pending_action");
- gpr_mu_unlock(&t->executor.mu);
- }
- break;
- }
-
- UNREF_TRANSPORT(exec_ctx, t, "run_global");
-
- GPR_TIMER_END("grpc_chttp2_run_with_global_lock", 0);
+static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *tp,
+ grpc_error *error) {
+ grpc_chttp2_transport *t = tp;
+ t->executor.check_read_ops_scheduled = false;
+ check_read_ops(exec_ctx, &t->global);
}
/*******************************************************************************
@@ -816,11 +709,12 @@ void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx,
void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *transport_global,
bool covered_by_poller, const char *reason) {
+ GPR_TIMER_BEGIN("grpc_chttp2_initiate_write", 0);
+
/* Perform state checks, and transition to a scheduled state if appropriate.
- Each time we finish the global lock execution, we check if we need to
- write. If we do:
- - (if there is a poller surrounding the write) schedule
- initiate_writing, which locks and calls initiate_writing_locked to...
+ If we are inactive, schedule a write chain to begin once the transport
+ combiner finishes any executions in its current batch (which may be
+ scheduled AFTER this code executes). The write chain will:
- call start_writing, which verifies (under the global lock) that there
are things that need to be written by calling
grpc_chttp2_unlocking_check_writes, and if so schedules writing_action
@@ -830,31 +724,28 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
to do *another* write immediately, and if so loops back to
start_writing.
- Current problems:
+ Current problems:
- too much lock entry/exiting
- the writing thread can become stuck indefinitely (punt through the
workqueue periodically to fix) */
grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
switch (t->executor.write_state) {
- case GRPC_CHTTP2_WRITING_INACTIVE:
- set_write_state(t, covered_by_poller
- ? GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER
- : GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER,
- reason);
+ case GRPC_CHTTP2_WRITES_CORKED:
break;
- case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER:
- /* nothing to do: write already requested */
+ case GRPC_CHTTP2_WRITING_INACTIVE:
+ set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, reason);
+ REF_TRANSPORT(t, "writing");
+ grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
+ &t->initiate_writing, GRPC_ERROR_NONE,
+ covered_by_poller);
break;
- case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER:
+ case GRPC_CHTTP2_WRITE_SCHEDULED:
if (covered_by_poller) {
/* upgrade to note poller is available to cover the write */
- set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, reason);
+ grpc_combiner_force_async_finally(t->executor.combiner);
}
break;
- case GRPC_CHTTP2_WRITE_SCHEDULED:
- /* nothing to do: write already scheduled */
- break;
case GRPC_CHTTP2_WRITING:
set_write_state(t,
covered_by_poller ? GRPC_CHTTP2_WRITING_STALE_WITH_POLLER
@@ -871,15 +762,15 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
}
break;
}
+ GPR_TIMER_END("grpc_chttp2_initiate_write", 0);
}
static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
- GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED ||
- t->executor.write_state == GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER);
+ GPR_TIMER_BEGIN("start_writing", 0);
+ GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED);
if (!t->closed &&
grpc_chttp2_unlocking_check_writes(exec_ctx, &t->global, &t->writing)) {
set_write_state(t, GRPC_CHTTP2_WRITING, "start_writing");
- REF_TRANSPORT(t, "writing");
prevent_endpoint_shutdown(t);
grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL);
} else {
@@ -890,25 +781,10 @@ static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE,
"start_writing:nothing_to_write");
}
- end_waiting_for_write(exec_ctx, t, GRPC_ERROR_NONE, "Nothing to write");
- if (t->ep && !t->endpoint_reading) {
- destroy_endpoint(exec_ctx, t);
- }
+ end_waiting_for_write(exec_ctx, t, GRPC_ERROR_NONE);
+ UNREF_TRANSPORT(exec_ctx, t, "writing");
}
-}
-
-static void initiate_writing_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused,
- void *arg_ignored) {
- start_writing(exec_ctx, t);
- UNREF_TRANSPORT(exec_ctx, t, "initiate_writing");
-}
-
-static void initiate_writing(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- grpc_chttp2_run_with_global_lock(exec_ctx, arg, NULL, initiate_writing_locked,
- NULL, 0);
+ GPR_TIMER_END("start_writing", 0);
}
void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
@@ -942,15 +818,10 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
/* error may be GRPC_ERROR_NONE if there is no error allocated yet.
In that case, use "reason" as the text for a new error. */
static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error,
- const char *reason) {
+ grpc_chttp2_transport *t, grpc_error *error) {
grpc_chttp2_stream_global *stream_global;
while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global,
&stream_global)) {
- if (error == GRPC_ERROR_NONE && reason != NULL) {
- /* create error object. */
- error = GRPC_ERROR_CREATE(reason);
- }
fail_pending_writes(exec_ctx, &t->global, stream_global,
GRPC_ERROR_REF(error));
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes");
@@ -958,12 +829,10 @@ static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_UNREF(error);
}
-static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_ignored,
- void *a) {
- grpc_error *error = a;
-
+static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *tp,
+ grpc_error *error) {
+ GPR_TIMER_BEGIN("terminate_writing_with_lock", 0);
+ grpc_chttp2_transport *t = tp;
allow_endpoint_shutdown_locked(exec_ctx, t);
if (error != GRPC_ERROR_NONE) {
@@ -972,39 +841,46 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx,
grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing);
- end_waiting_for_write(exec_ctx, t, error, NULL);
+ end_waiting_for_write(exec_ctx, t, GRPC_ERROR_REF(error));
switch (t->executor.write_state) {
+ case GRPC_CHTTP2_WRITES_CORKED:
case GRPC_CHTTP2_WRITING_INACTIVE:
- case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER:
- case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER:
case GRPC_CHTTP2_WRITE_SCHEDULED:
GPR_UNREACHABLE_CODE(break);
case GRPC_CHTTP2_WRITING:
+ GPR_TIMER_MARK("state=writing", 0);
set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "terminate_writing");
break;
case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER:
- set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER,
- "terminate_writing");
+ GPR_TIMER_MARK("state=writing_stale_with_poller", 0);
+ set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "terminate_writing");
+ REF_TRANSPORT(t, "writing");
+ grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
+ &t->initiate_writing, GRPC_ERROR_NONE,
+ true);
break;
case GRPC_CHTTP2_WRITING_STALE_NO_POLLER:
- set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER,
- "terminate_writing");
+ GPR_TIMER_MARK("state=writing_stale_no_poller", 0);
+ set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "terminate_writing");
+ REF_TRANSPORT(t, "writing");
+ grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
+ &t->initiate_writing, GRPC_ERROR_NONE,
+ false);
break;
}
- if (t->ep && !t->endpoint_reading) {
- destroy_endpoint(exec_ctx, t);
- }
-
UNREF_TRANSPORT(exec_ctx, t, "writing");
+ GPR_TIMER_END("terminate_writing_with_lock", 0);
}
void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
void *transport_writing, grpc_error *error) {
+ GPR_TIMER_BEGIN("grpc_chttp2_terminate_writing", 0);
grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing);
- grpc_chttp2_run_with_global_lock(
- exec_ctx, t, NULL, terminate_writing_with_lock, GRPC_ERROR_REF(error), 0);
+ grpc_combiner_execute(exec_ctx, t->executor.combiner, &t->terminate_writing,
+ GRPC_ERROR_REF(error));
+ GPR_TIMER_END("grpc_chttp2_terminate_writing", 0);
}
static void writing_action(grpc_exec_ctx *exec_ctx, void *gt,
@@ -1148,15 +1024,22 @@ static int contains_non_ok_status(
static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {}
-static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s, void *stream_op) {
+static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
+ grpc_error *error_ignored) {
GPR_TIMER_BEGIN("perform_stream_op_locked", 0);
grpc_transport_stream_op *op = stream_op;
+ grpc_chttp2_transport *t = op->transport_private.args[0];
+ grpc_chttp2_stream *s = op->transport_private.args[1];
grpc_chttp2_transport_global *transport_global = &t->global;
grpc_chttp2_stream_global *stream_global = &s->global;
+ if (grpc_http_trace) {
+ char *str = grpc_transport_stream_op_string(op);
+ gpr_log(GPR_DEBUG, "perform_stream_op_locked: %s", str);
+ gpr_free(str);
+ }
+
grpc_closure *on_complete = op->on_complete;
if (on_complete == NULL) {
on_complete = grpc_closure_create(do_nothing, NULL);
@@ -1211,7 +1094,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
} else {
if (contains_non_ok_status(transport_global, op->send_initial_metadata)) {
stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (!stream_global->write_closed) {
if (transport_global->is_client) {
@@ -1278,7 +1162,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
if (contains_non_ok_status(transport_global,
op->send_trailing_metadata)) {
stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (stream_global->write_closed) {
stream_global->send_trailing_metadata = NULL;
@@ -1303,7 +1188,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
stream_global->recv_initial_metadata_ready =
op->recv_initial_metadata_ready;
stream_global->recv_initial_metadata = op->recv_initial_metadata;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (op->recv_message != NULL) {
@@ -1317,7 +1203,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
exec_ctx, transport_global, stream_global,
transport_global->stream_lookahead, 0);
}
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (op->recv_trailing_metadata != NULL) {
@@ -1326,21 +1213,30 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
add_closure_barrier(on_complete);
stream_global->recv_trailing_metadata = op->recv_trailing_metadata;
stream_global->final_metadata_requested = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
grpc_chttp2_complete_closure_step(exec_ctx, transport_global, stream_global,
&on_complete, GRPC_ERROR_NONE);
GPR_TIMER_END("perform_stream_op_locked", 0);
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "perform_stream_op");
}
static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_transport_stream_op *op) {
+ GPR_TIMER_BEGIN("perform_stream_op", 0);
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
- grpc_chttp2_run_with_global_lock(exec_ctx, t, s, perform_stream_op_locked, op,
- sizeof(*op));
+ grpc_closure_init(&op->transport_private.closure, perform_stream_op_locked,
+ op);
+ op->transport_private.args[0] = gt;
+ op->transport_private.args[1] = gs;
+ GRPC_CHTTP2_STREAM_REF(&s->global, "perform_stream_op");
+ grpc_combiner_execute(exec_ctx, t->executor.combiner,
+ &op->transport_private.closure, GRPC_ERROR_NONE);
+ GPR_TIMER_END("perform_stream_op", 0);
}
static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -1362,13 +1258,20 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_chttp2_initiate_write(exec_ctx, &t->global, true, "send_ping");
}
-static void ack_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
- grpc_chttp2_stream *s, void *opaque_8bytes) {
+typedef struct ack_ping_args {
+ grpc_closure closure;
+ grpc_chttp2_transport *t;
+ uint8_t opaque_8bytes[8];
+} ack_ping_args;
+
+static void ack_ping_locked(grpc_exec_ctx *exec_ctx, void *a,
+ grpc_error *error_ignored) {
+ ack_ping_args *args = a;
grpc_chttp2_outstanding_ping *ping;
- grpc_chttp2_transport_global *transport_global = &t->global;
+ grpc_chttp2_transport_global *transport_global = &args->t->global;
for (ping = transport_global->pings.next; ping != &transport_global->pings;
ping = ping->next) {
- if (0 == memcmp(opaque_8bytes, ping->id, 8)) {
+ if (0 == memcmp(args->opaque_8bytes, ping->id, 8)) {
grpc_exec_ctx_sched(exec_ctx, ping->on_recv, GRPC_ERROR_NONE, NULL);
ping->next->prev = ping->prev;
ping->prev->next = ping->next;
@@ -1376,21 +1279,27 @@ static void ack_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
break;
}
}
+ UNREF_TRANSPORT(exec_ctx, args->t, "ack_ping");
+ gpr_free(args);
}
void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_parsing *transport_parsing,
const uint8_t *opaque_8bytes) {
- grpc_chttp2_run_with_global_lock(
- exec_ctx, TRANSPORT_FROM_PARSING(transport_parsing), NULL,
- ack_ping_locked, (void *)opaque_8bytes, 8);
+ ack_ping_args *args = gpr_malloc(sizeof(*args));
+ args->t = TRANSPORT_FROM_PARSING(transport_parsing);
+ memcpy(args->opaque_8bytes, opaque_8bytes, sizeof(args->opaque_8bytes));
+ grpc_closure_init(&args->closure, ack_ping_locked, args);
+ REF_TRANSPORT(args->t, "ack_ping");
+ grpc_combiner_execute(exec_ctx, args->t->executor.combiner, &args->closure,
+ GRPC_ERROR_NONE);
}
static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused,
- void *stream_op) {
+ void *stream_op,
+ grpc_error *error_ignored) {
grpc_transport_op *op = stream_op;
+ grpc_chttp2_transport *t = op->transport_private.args[0];
grpc_error *close_transport = op->disconnect_with_error;
/* If there's a set_accept_stream ensure that we're not parsing
@@ -1402,8 +1311,6 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
return;
}
- grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL);
-
if (op->on_connectivity_state_change != NULL) {
grpc_connectivity_state_notify_on_state_change(
exec_ctx, &t->channel_callback.state_tracker, op->connectivity_state,
@@ -1429,11 +1336,11 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
}
if (op->bind_pollset) {
- add_to_pollset_locked(exec_ctx, t, NULL, op->bind_pollset);
+ grpc_endpoint_add_to_pollset(exec_ctx, t->ep, op->bind_pollset);
}
if (op->bind_pollset_set) {
- add_to_pollset_set_locked(exec_ctx, t, NULL, op->bind_pollset_set);
+ grpc_endpoint_add_to_pollset_set(exec_ctx, t->ep, op->bind_pollset_set);
}
if (op->send_ping) {
@@ -1443,13 +1350,21 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
if (close_transport != GRPC_ERROR_NONE) {
close_transport_locked(exec_ctx, t, close_transport);
}
+
+ grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL);
+
+ UNREF_TRANSPORT(exec_ctx, t, "transport_op");
}
static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_transport_op *op) {
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
- grpc_chttp2_run_with_global_lock(
- exec_ctx, t, NULL, perform_transport_op_locked, op, sizeof(*op));
+ op->transport_private.args[0] = gt;
+ grpc_closure_init(&op->transport_private.closure, perform_transport_op_locked,
+ op);
+ REF_TRANSPORT(t, "transport_op");
+ grpc_combiner_execute(exec_ctx, t->executor.combiner,
+ &op->transport_private.closure, GRPC_ERROR_NONE);
}
/*******************************************************************************
@@ -1458,6 +1373,7 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
static void check_read_ops(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *transport_global) {
+ GPR_TIMER_BEGIN("check_read_ops", 0);
grpc_chttp2_stream_global *stream_global;
grpc_byte_stream *bs;
while (
@@ -1467,7 +1383,7 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
if (stream_global->seen_error) {
while ((bs = grpc_chttp2_incoming_frame_queue_pop(
&stream_global->incoming_frames)) != NULL) {
- incoming_byte_stream_destroy_locked(exec_ctx, NULL, NULL, bs);
+ incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
if (stream_global->exceeded_metadata_size) {
cancel_from_api(
@@ -1490,7 +1406,7 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
stream_global->seen_error &&
(bs = grpc_chttp2_incoming_frame_queue_pop(
&stream_global->incoming_frames)) != NULL) {
- incoming_byte_stream_destroy_locked(exec_ctx, NULL, NULL, bs);
+ incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
if (stream_global->incoming_frames.head != NULL) {
*stream_global->recv_message = grpc_chttp2_incoming_frame_queue_pop(
@@ -1511,7 +1427,7 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
if (stream_global->seen_error) {
while ((bs = grpc_chttp2_incoming_frame_queue_pop(
&stream_global->incoming_frames)) != NULL) {
- incoming_byte_stream_destroy_locked(exec_ctx, NULL, NULL, bs);
+ incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
if (stream_global->exceeded_metadata_size) {
cancel_from_api(
@@ -1532,6 +1448,7 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
}
}
}
+ GPR_TIMER_END("check_read_ops", 0);
}
static void decrement_active_streams_locked(
@@ -1539,7 +1456,8 @@ static void decrement_active_streams_locked(
grpc_chttp2_stream_global *stream_global) {
if ((stream_global->all_incoming_byte_streams_finished =
gpr_unref(&stream_global->active_streams))) {
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
}
@@ -1643,7 +1561,8 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx,
}
if (due_to_error != GRPC_ERROR_NONE && !stream_global->seen_error) {
stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1,
1, due_to_error);
@@ -1655,7 +1574,8 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
grpc_status_code status, gpr_slice *slice) {
if (status != GRPC_STATUS_OK) {
stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
/* stream_global->recv_trailing_metadata_finished gives us a
last chance replacement: we've received trailing metadata,
@@ -1679,7 +1599,8 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
grpc_mdstr_from_slice(gpr_slice_ref(*slice))));
}
stream_global->published_trailing_metadata = true;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (slice) {
gpr_slice_unref(*slice);
@@ -1739,7 +1660,8 @@ void grpc_chttp2_mark_stream_closed(
GRPC_ERROR_UNREF(error);
return;
}
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
if (close_reads && !stream_global->read_closed) {
stream_global->read_closed_error = GRPC_ERROR_REF(error);
stream_global->read_closed = true;
@@ -1920,6 +1842,10 @@ static void end_all_the_calls(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_error *error) {
+ if (!grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, NULL)) {
+ error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS,
+ GRPC_STATUS_UNAVAILABLE);
+ }
close_transport_locked(exec_ctx, t, GRPC_ERROR_REF(error));
end_all_the_calls(exec_ctx, t, error);
}
@@ -1955,16 +1881,12 @@ static void update_global_window(void *args, uint32_t id, void *stream) {
* INPUT PROCESSING - PARSING
*/
-static void reading_action_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *arg);
static void parsing_action(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error);
-static void post_reading_action_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *arg);
-static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *arg);
+static void post_reading_action_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error);
+static void post_parse_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error);
static void reading_action(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *error) {
@@ -1972,16 +1894,20 @@ static void reading_action(grpc_exec_ctx *exec_ctx, void *tp,
reading_action_locked ->
(parse_unlocked -> post_parse_locked)? ->
post_reading_action_locked */
- grpc_chttp2_run_with_global_lock(exec_ctx, tp, NULL, reading_action_locked,
- GRPC_ERROR_REF(error), 0);
+ GPR_TIMER_BEGIN("reading_action", 0);
+ grpc_chttp2_transport *t = tp;
+ grpc_combiner_execute(exec_ctx, t->executor.combiner,
+ &t->reading_action_locked, GRPC_ERROR_REF(error));
+ GPR_TIMER_END("reading_action", 0);
}
-static void reading_action_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *arg) {
+static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
+ grpc_error *error) {
+ GPR_TIMER_BEGIN("reading_action_locked", 0);
+
+ grpc_chttp2_transport *t = tp;
grpc_chttp2_transport_global *transport_global = &t->global;
grpc_chttp2_transport_parsing *transport_parsing = &t->parsing;
- grpc_error *error = arg;
GPR_ASSERT(!t->executor.parsing_active);
if (!t->closed) {
@@ -1990,10 +1916,13 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_map_move_into(&t->new_stream_map,
&t->parsing_stream_map);
grpc_chttp2_prepare_to_read(transport_global, transport_parsing);
- grpc_exec_ctx_sched(exec_ctx, &t->parsing_action, error, NULL);
+ grpc_exec_ctx_sched(exec_ctx, &t->parsing_action, GRPC_ERROR_REF(error),
+ NULL);
} else {
- post_reading_action_locked(exec_ctx, t, s_unused, arg);
+ post_reading_action_locked(exec_ctx, t, error);
}
+
+ GPR_TIMER_END("reading_action_locked", 0);
}
static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
@@ -2045,13 +1974,15 @@ static void parsing_action(grpc_exec_ctx *exec_ctx, void *arg,
for (i = 0; i < GPR_ARRAY_SIZE(errors); i++) {
GRPC_ERROR_UNREF(errors[i]);
}
+ grpc_combiner_execute(exec_ctx, t->executor.combiner, &t->post_parse_locked,
+ err);
GPR_TIMER_END("reading_action.parse", 0);
- grpc_chttp2_run_with_global_lock(exec_ctx, t, NULL, post_parse_locked, err,
- 0);
}
-static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *arg) {
+static void post_parse_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ GPR_TIMER_BEGIN("post_parse_locked", 0);
+ grpc_chttp2_transport *t = arg;
grpc_chttp2_transport_global *transport_global = &t->global;
grpc_chttp2_transport_parsing *transport_parsing = &t->parsing;
/* copy parsing qbuf to global qbuf */
@@ -2077,7 +2008,7 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
if (t->post_parsing_op) {
grpc_transport_op *op = t->post_parsing_op;
t->post_parsing_op = NULL;
- perform_transport_op_locked(exec_ctx, t, NULL, op);
+ perform_transport_op_locked(exec_ctx, op, GRPC_ERROR_NONE);
gpr_free(op);
}
/* if a stream is in the stream map, and gets cancelled, we need to
@@ -2094,39 +2025,32 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2");
}
- post_reading_action_locked(exec_ctx, t, s_unused, arg);
+ post_reading_action_locked(exec_ctx, t, error);
+ GPR_TIMER_END("post_parse_locked", 0);
}
-static void post_reading_action_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused,
- void *arg) {
- grpc_error *error = arg;
+static void post_reading_action_locked(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ GPR_TIMER_BEGIN("post_reading_action_locked", 0);
+ grpc_chttp2_transport *t = arg;
bool keep_reading = false;
+ GRPC_ERROR_REF(error);
if (error == GRPC_ERROR_NONE && t->closed) {
error = GRPC_ERROR_CREATE("Transport closed");
}
if (error != GRPC_ERROR_NONE) {
- if (!grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, NULL)) {
- error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS,
- GRPC_STATUS_UNAVAILABLE);
- }
drop_connection(exec_ctx, t, GRPC_ERROR_REF(error));
t->endpoint_reading = 0;
if (grpc_http_write_state_trace) {
gpr_log(GPR_DEBUG, "R:%p -> 0 ws=%s", t,
write_state_name(t->executor.write_state));
}
- if (t->executor.write_state == GRPC_CHTTP2_WRITING_INACTIVE && t->ep) {
- destroy_endpoint(exec_ctx, t);
- }
} else if (!t->closed) {
keep_reading = true;
REF_TRANSPORT(t, "keep_reading");
prevent_endpoint_shutdown(t);
}
gpr_slice_buffer_reset_and_unref(&t->read_buffer);
- GRPC_ERROR_UNREF(error);
if (keep_reading) {
grpc_endpoint_read(exec_ctx, t->ep, &t->read_buffer, &t->reading_action);
@@ -2135,6 +2059,9 @@ static void post_reading_action_locked(grpc_exec_ctx *exec_ctx,
} else {
UNREF_TRANSPORT(exec_ctx, t, "reading_action");
}
+ GRPC_ERROR_UNREF(error);
+
+ GPR_TIMER_END("post_reading_action_locked", 0);
}
/*******************************************************************************
@@ -2156,36 +2083,16 @@ static void connectivity_state_set(
* POLLSET STUFF
*/
-static void add_to_pollset_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused, void *pollset) {
- if (t->ep) {
- grpc_endpoint_add_to_pollset(exec_ctx, t->ep, pollset);
- }
-}
-
-static void add_to_pollset_set_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s_unused,
- void *pollset_set) {
- if (t->ep) {
- grpc_endpoint_add_to_pollset_set(exec_ctx, t->ep, pollset_set);
- }
-}
-
static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_pollset *pollset) {
- /* TODO(ctiller): keep pollset alive */
- grpc_chttp2_run_with_global_lock(exec_ctx, (grpc_chttp2_transport *)gt,
- (grpc_chttp2_stream *)gs,
- add_to_pollset_locked, pollset, 0);
+ grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+ grpc_endpoint_add_to_pollset(exec_ctx, t->ep, pollset);
}
static void set_pollset_set(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_pollset_set *pollset_set) {
- grpc_chttp2_run_with_global_lock(exec_ctx, (grpc_chttp2_transport *)gt,
- (grpc_chttp2_stream *)gs,
- add_to_pollset_set_locked, pollset_set, 0);
+ grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+ grpc_endpoint_add_to_pollset_set(exec_ctx, t->ep, pollset_set);
}
/*******************************************************************************
@@ -2197,6 +2104,7 @@ static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx,
if (gpr_unref(&bs->refs)) {
GRPC_ERROR_UNREF(bs->error);
gpr_slice_buffer_destroy(&bs->slices);
+ gpr_mu_destroy(&bs->slice_mu);
gpr_free(bs);
}
}
@@ -2242,38 +2150,34 @@ static void incoming_byte_stream_update_flow_control(
}
}
-typedef struct {
- grpc_chttp2_incoming_byte_stream *byte_stream;
- gpr_slice *slice;
- size_t max_size_hint;
- grpc_closure *on_complete;
-} incoming_byte_stream_next_arg;
-
static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *argp) {
- incoming_byte_stream_next_arg *arg = argp;
- grpc_chttp2_incoming_byte_stream *bs =
- (grpc_chttp2_incoming_byte_stream *)arg->byte_stream;
+ void *argp,
+ grpc_error *error_ignored) {
+ grpc_chttp2_incoming_byte_stream *bs = argp;
grpc_chttp2_transport_global *transport_global = &bs->transport->global;
grpc_chttp2_stream_global *stream_global = &bs->stream->global;
if (bs->is_tail) {
- incoming_byte_stream_update_flow_control(exec_ctx, transport_global,
- stream_global, arg->max_size_hint,
- bs->slices.length);
- }
+ gpr_mu_lock(&bs->slice_mu);
+ size_t cur_length = bs->slices.length;
+ gpr_mu_unlock(&bs->slice_mu);
+ incoming_byte_stream_update_flow_control(
+ exec_ctx, transport_global, stream_global,
+ bs->next_action.max_size_hint, cur_length);
+ }
+ gpr_mu_lock(&bs->slice_mu);
if (bs->slices.count > 0) {
- *arg->slice = gpr_slice_buffer_take_first(&bs->slices);
- grpc_exec_ctx_sched(exec_ctx, arg->on_complete, GRPC_ERROR_NONE, NULL);
- } else if (bs->error != GRPC_ERROR_NONE) {
- grpc_exec_ctx_sched(exec_ctx, arg->on_complete, GRPC_ERROR_REF(bs->error),
+ *bs->next_action.slice = gpr_slice_buffer_take_first(&bs->slices);
+ grpc_exec_ctx_sched(exec_ctx, bs->next_action.on_complete, GRPC_ERROR_NONE,
NULL);
+ } else if (bs->error != GRPC_ERROR_NONE) {
+ grpc_exec_ctx_sched(exec_ctx, bs->next_action.on_complete,
+ GRPC_ERROR_REF(bs->error), NULL);
} else {
- bs->on_next = arg->on_complete;
- bs->next = arg->slice;
+ bs->on_next = bs->next_action.on_complete;
+ bs->next = bs->next_action.slice;
}
+ gpr_mu_unlock(&bs->slice_mu);
incoming_byte_stream_unref(exec_ctx, bs);
}
@@ -2281,13 +2185,18 @@ static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
grpc_byte_stream *byte_stream,
gpr_slice *slice, size_t max_size_hint,
grpc_closure *on_complete) {
+ GPR_TIMER_BEGIN("incoming_byte_stream_next", 0);
grpc_chttp2_incoming_byte_stream *bs =
(grpc_chttp2_incoming_byte_stream *)byte_stream;
- incoming_byte_stream_next_arg arg = {bs, slice, max_size_hint, on_complete};
gpr_ref(&bs->refs);
- grpc_chttp2_run_with_global_lock(exec_ctx, bs->transport, bs->stream,
- incoming_byte_stream_next_locked, &arg,
- sizeof(arg));
+ bs->next_action.slice = slice;
+ bs->next_action.max_size_hint = max_size_hint;
+ bs->next_action.on_complete = on_complete;
+ grpc_closure_init(&bs->next_action.closure, incoming_byte_stream_next_locked,
+ bs);
+ grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
+ &bs->next_action.closure, GRPC_ERROR_NONE);
+ GPR_TIMER_END("incoming_byte_stream_next", 0);
return 0;
}
@@ -2295,9 +2204,8 @@ static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx,
grpc_byte_stream *byte_stream);
static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *byte_stream) {
+ void *byte_stream,
+ grpc_error *error_ignored) {
grpc_chttp2_incoming_byte_stream *bs = byte_stream;
GPR_ASSERT(bs->base.destroy == incoming_byte_stream_destroy);
decrement_active_streams_locked(exec_ctx, &bs->transport->global,
@@ -2307,10 +2215,14 @@ static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx,
static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx,
grpc_byte_stream *byte_stream) {
+ GPR_TIMER_BEGIN("incoming_byte_stream_destroy", 0);
grpc_chttp2_incoming_byte_stream *bs =
(grpc_chttp2_incoming_byte_stream *)byte_stream;
- grpc_chttp2_run_with_global_lock(exec_ctx, bs->transport, bs->stream,
- incoming_byte_stream_destroy_locked, bs, 0);
+ grpc_closure_init(&bs->destroy_action, incoming_byte_stream_destroy_locked,
+ bs);
+ grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
+ &bs->destroy_action, GRPC_ERROR_NONE);
+ GPR_TIMER_END("incoming_byte_stream_destroy", 0);
}
typedef struct {
@@ -2318,90 +2230,45 @@ typedef struct {
gpr_slice slice;
} incoming_byte_stream_push_arg;
-static void incoming_byte_stream_push_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *argp) {
- incoming_byte_stream_push_arg *arg = argp;
- grpc_chttp2_incoming_byte_stream *bs = arg->byte_stream;
+void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_incoming_byte_stream *bs,
+ gpr_slice slice) {
+ gpr_mu_lock(&bs->slice_mu);
if (bs->on_next != NULL) {
- *bs->next = arg->slice;
+ *bs->next = slice;
grpc_exec_ctx_sched(exec_ctx, bs->on_next, GRPC_ERROR_NONE, NULL);
bs->on_next = NULL;
} else {
- gpr_slice_buffer_add(&bs->slices, arg->slice);
+ gpr_slice_buffer_add(&bs->slices, slice);
}
- incoming_byte_stream_unref(exec_ctx, bs);
+ gpr_mu_unlock(&bs->slice_mu);
}
-void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_incoming_byte_stream *bs,
- gpr_slice slice) {
- incoming_byte_stream_push_arg arg = {bs, slice};
- gpr_ref(&bs->refs);
- grpc_chttp2_run_with_global_lock(exec_ctx, bs->transport, bs->stream,
- incoming_byte_stream_push_locked, &arg,
- sizeof(arg));
-}
-
-typedef struct {
- grpc_chttp2_incoming_byte_stream *bs;
- grpc_error *error;
-} bs_fail_args;
-
-static bs_fail_args *make_bs_fail_args(grpc_chttp2_incoming_byte_stream *bs,
- grpc_error *error) {
- bs_fail_args *a = gpr_malloc(sizeof(*a));
- a->bs = bs;
- a->error = error;
- return a;
-}
-
-static void incoming_byte_stream_finished_failed_locked(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s,
- void *argp) {
- bs_fail_args *a = argp;
- grpc_chttp2_incoming_byte_stream *bs = a->bs;
- grpc_error *error = a->error;
- gpr_free(a);
- grpc_exec_ctx_sched(exec_ctx, bs->on_next, GRPC_ERROR_REF(error), NULL);
- bs->on_next = NULL;
- GRPC_ERROR_UNREF(bs->error);
- bs->error = error;
- incoming_byte_stream_unref(exec_ctx, bs);
-}
-
-static void incoming_byte_stream_finished_ok_locked(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s,
- void *argp) {
- grpc_chttp2_incoming_byte_stream *bs = argp;
+static void incoming_byte_stream_finished_locked(grpc_exec_ctx *exec_ctx,
+ void *bsp, grpc_error *error) {
+ grpc_chttp2_incoming_byte_stream *bs = bsp;
+ if (error != GRPC_ERROR_NONE) {
+ grpc_exec_ctx_sched(exec_ctx, bs->on_next, GRPC_ERROR_REF(error), NULL);
+ bs->on_next = NULL;
+ GRPC_ERROR_UNREF(bs->error);
+ bs->error = error;
+ }
incoming_byte_stream_unref(exec_ctx, bs);
}
void grpc_chttp2_incoming_byte_stream_finished(
grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs,
grpc_error *error, int from_parsing_thread) {
+ GPR_TIMER_BEGIN("grpc_chttp2_incoming_byte_stream_finished", 0);
if (from_parsing_thread) {
- if (error == GRPC_ERROR_NONE) {
- grpc_chttp2_run_with_global_lock(exec_ctx, bs->transport, bs->stream,
- incoming_byte_stream_finished_ok_locked,
- bs, 0);
- } else {
- grpc_chttp2_run_with_global_lock(
- exec_ctx, bs->transport, bs->stream,
- incoming_byte_stream_finished_failed_locked,
- make_bs_fail_args(bs, error), 0);
- }
+ grpc_closure_init(&bs->finished_action,
+ incoming_byte_stream_finished_locked, bs);
+ grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
+ &bs->finished_action, GRPC_ERROR_REF(error));
} else {
- if (error == GRPC_ERROR_NONE) {
- incoming_byte_stream_finished_ok_locked(exec_ctx, bs->transport,
- bs->stream, bs);
- } else {
- incoming_byte_stream_finished_failed_locked(
- exec_ctx, bs->transport, bs->stream, make_bs_fail_args(bs, error));
- }
+ incoming_byte_stream_finished_locked(exec_ctx, bs, error);
}
+ GPR_TIMER_END("grpc_chttp2_incoming_byte_stream_finished", 0);
}
grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
@@ -2414,6 +2281,7 @@ grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
incoming_byte_stream->base.flags = flags;
incoming_byte_stream->base.next = incoming_byte_stream_next;
incoming_byte_stream->base.destroy = incoming_byte_stream_destroy;
+ gpr_mu_init(&incoming_byte_stream->slice_mu);
gpr_ref_init(&incoming_byte_stream->refs, 2);
incoming_byte_stream->next_message = NULL;
incoming_byte_stream->transport = TRANSPORT_FROM_PARSING(transport_parsing);
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index d67c014e54..04b788b702 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -48,6 +48,7 @@
#include "src/core/ext/transport/chttp2/transport/hpack_parser.h"
#include "src/core/ext/transport/chttp2/transport/incoming_metadata.h"
#include "src/core/ext/transport/chttp2/transport/stream_map.h"
+#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/transport/connectivity_state.h"
#include "src/core/lib/transport/transport_impl.h"
@@ -161,9 +162,20 @@ struct grpc_chttp2_incoming_byte_stream {
grpc_chttp2_transport *transport;
grpc_chttp2_stream *stream;
int is_tail;
+
+ gpr_mu slice_mu; // protects slices, on_next
gpr_slice_buffer slices;
grpc_closure *on_next;
gpr_slice *next;
+
+ struct {
+ grpc_closure closure;
+ gpr_slice *slice;
+ size_t max_size_hint;
+ grpc_closure *on_complete;
+ } next_action;
+ grpc_closure destroy_action;
+ grpc_closure finished_action;
};
typedef struct {
@@ -296,23 +308,11 @@ struct grpc_chttp2_transport_parsing {
int64_t outgoing_window;
};
-typedef void (*grpc_chttp2_locked_action)(grpc_exec_ctx *ctx,
- grpc_chttp2_transport *t,
- grpc_chttp2_stream *s, void *arg);
-
-typedef struct grpc_chttp2_executor_action_header {
- grpc_chttp2_stream *stream;
- grpc_chttp2_locked_action action;
- struct grpc_chttp2_executor_action_header *next;
- void *arg;
-} grpc_chttp2_executor_action_header;
-
typedef enum {
+ /** no writing activity allowed */
+ GRPC_CHTTP2_WRITES_CORKED,
/** no writing activity */
GRPC_CHTTP2_WRITING_INACTIVE,
- /** write has been requested, but not scheduled yet */
- GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER,
- GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER,
/** write has been requested and scheduled against the workqueue */
GRPC_CHTTP2_WRITE_SCHEDULED,
/** write has been initiated after being reaped from the workqueue */
@@ -333,7 +333,7 @@ struct grpc_chttp2_transport {
gpr_refcount shutdown_ep_refs;
struct {
- gpr_mu mu;
+ grpc_combiner *combiner;
/** is a thread currently in the global lock */
bool global_active;
@@ -341,9 +341,8 @@ struct grpc_chttp2_transport {
bool parsing_active;
/** write execution state of the transport */
grpc_chttp2_write_state write_state;
-
- grpc_chttp2_executor_action_header *pending_actions_head;
- grpc_chttp2_executor_action_header *pending_actions_tail;
+ /** has a check_read_ops been scheduled */
+ bool check_read_ops_scheduled;
} executor;
/** is the transport destroying itself? */
@@ -380,10 +379,16 @@ struct grpc_chttp2_transport {
grpc_closure writing_action;
/** closure to start reading from the endpoint */
grpc_closure reading_action;
+ grpc_closure reading_action_locked;
+ grpc_closure post_parse_locked;
/** closure to actually do parsing */
grpc_closure parsing_action;
/** closure to initiate writing */
grpc_closure initiate_writing;
+ /** closure to finish writing */
+ grpc_closure terminate_writing;
+ /** closure to flush read state up the stack */
+ grpc_closure initiate_read_flush_locked;
/** incoming read bytes */
gpr_slice_buffer read_buffer;
@@ -527,11 +532,16 @@ struct grpc_chttp2_stream_parsing {
};
struct grpc_chttp2_stream {
+ grpc_chttp2_transport *t;
grpc_stream_refcount *refcount;
grpc_chttp2_stream_global global;
grpc_chttp2_stream_writing writing;
grpc_chttp2_stream_parsing parsing;
+ grpc_closure init_stream;
+ grpc_closure destroy_stream;
+ void *destroy_stream_arg;
+
grpc_chttp2_stream_link links[STREAM_LIST_COUNT];
uint8_t included[STREAM_LIST_COUNT];
};
@@ -626,7 +636,7 @@ int grpc_chttp2_list_pop_waiting_for_concurrency(
grpc_chttp2_stream_global **stream_global);
void grpc_chttp2_list_add_check_read_ops(
- grpc_chttp2_transport_global *transport_global,
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global);
bool grpc_chttp2_list_remove_check_read_ops(
grpc_chttp2_transport_global *transport_global,
@@ -706,12 +716,6 @@ void grpc_chttp2_complete_closure_step(
grpc_chttp2_stream_global *stream_global, grpc_closure **pclosure,
grpc_error *error);
-void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *transport,
- grpc_chttp2_stream *optional_stream,
- grpc_chttp2_locked_action action,
- void *arg, size_t sizeof_arg);
-
#define GRPC_CHTTP2_CLIENT_CONNECT_STRING "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
#define GRPC_CHTTP2_CLIENT_CONNECT_STRLEN \
(sizeof(GRPC_CHTTP2_CLIENT_CONNECT_STRING) - 1)
diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c
index 482cd55c44..0e6d579ba9 100644
--- a/src/core/ext/transport/chttp2/transport/parsing.c
+++ b/src/core/ext/transport/chttp2/transport/parsing.c
@@ -177,7 +177,8 @@ void grpc_chttp2_publish_reads(
stream_global->seen_error = true;
stream_global->exceeded_metadata_size =
stream_parsing->exceeded_metadata_size;
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
/* flush stats to global stream state */
@@ -203,7 +204,8 @@ void grpc_chttp2_publish_reads(
stream_global->incoming_frames.tail->is_tail = 0;
}
if (stream_parsing->data_parser.incoming_frames.head != NULL) {
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
grpc_chttp2_incoming_frame_queue_merge(
&stream_global->incoming_frames,
@@ -219,7 +221,8 @@ void grpc_chttp2_publish_reads(
GPR_SWAP(grpc_chttp2_incoming_metadata_buffer,
stream_parsing->metadata_buffer[0],
stream_global->received_initial_metadata);
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (!stream_global->published_trailing_metadata &&
stream_parsing->got_metadata_on_parse[1]) {
@@ -228,7 +231,8 @@ void grpc_chttp2_publish_reads(
GPR_SWAP(grpc_chttp2_incoming_metadata_buffer,
stream_parsing->metadata_buffer[1],
stream_global->received_trailing_metadata);
- grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+ stream_global);
}
if (stream_parsing->forced_close_error != GRPC_ERROR_NONE) {
diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c
index 2eb5f5f632..4dc4968248 100644
--- a/src/core/ext/transport/chttp2/transport/stream_lists.c
+++ b/src/core/ext/transport/chttp2/transport/stream_lists.c
@@ -298,8 +298,15 @@ int grpc_chttp2_list_pop_waiting_for_concurrency(
}
void grpc_chttp2_list_add_check_read_ops(
- grpc_chttp2_transport_global *transport_global,
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global) {
+ grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
+ if (!t->executor.check_read_ops_scheduled) {
+ grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
+ &t->initiate_read_flush_locked,
+ GRPC_ERROR_NONE, false);
+ t->executor.check_read_ops_scheduled = true;
+ }
stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
STREAM_FROM_GLOBAL(stream_global),
GRPC_CHTTP2_LIST_CHECK_READ_OPS);
diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c
index 311b26e354..979515bd54 100644
--- a/src/core/ext/transport/chttp2/transport/writing.c
+++ b/src/core/ext/transport/chttp2/transport/writing.c
@@ -55,15 +55,6 @@ int grpc_chttp2_unlocking_check_writes(
transport_global->settings[GRPC_ACKED_SETTINGS]
[GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE];
- /* simple writes are queued to qbuf, and flushed here */
- gpr_slice_buffer_swap(&transport_global->qbuf, &transport_writing->outbuf);
- GPR_ASSERT(transport_global->qbuf.count == 0);
-
- grpc_chttp2_hpack_compressor_set_max_table_size(
- &transport_writing->hpack_compressor,
- transport_global->settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]);
-
if (transport_global->dirtied_local_settings &&
!transport_global->sent_local_settings) {
gpr_slice_buffer_add(
@@ -77,6 +68,16 @@ int grpc_chttp2_unlocking_check_writes(
transport_global->sent_local_settings = 1;
}
+ /* simple writes are queued to qbuf, and flushed here */
+ gpr_slice_buffer_move_into(&transport_global->qbuf,
+ &transport_writing->outbuf);
+ GPR_ASSERT(transport_global->qbuf.count == 0);
+
+ grpc_chttp2_hpack_compressor_set_max_table_size(
+ &transport_writing->hpack_compressor,
+ transport_global->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]);
+
GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("write", transport_writing, outgoing_window,
transport_global, outgoing_window);
if (transport_writing->outgoing_window > 0) {
@@ -344,6 +345,7 @@ static void finalize_outbuf(grpc_exec_ctx *exec_ctx,
void grpc_chttp2_cleanup_writing(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
grpc_chttp2_transport_writing *transport_writing) {
+ GPR_TIMER_BEGIN("grpc_chttp2_cleanup_writing", 0);
grpc_chttp2_stream_writing *stream_writing;
grpc_chttp2_stream_global *stream_global;
@@ -382,4 +384,5 @@ void grpc_chttp2_cleanup_writing(
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2_writing");
}
gpr_slice_buffer_reset_and_unref(&transport_writing->outbuf);
+ GPR_TIMER_END("grpc_chttp2_cleanup_writing", 0);
}