diff options
Diffstat (limited to 'src/core/ext/transport/chttp2')
21 files changed, 613 insertions, 312 deletions
diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.c b/src/core/ext/transport/chttp2/client/chttp2_connector.c index d0a762a280..2b226c1bf7 100644 --- a/src/core/ext/transport/chttp2/client/chttp2_connector.c +++ b/src/core/ext/transport/chttp2/client/chttp2_connector.c @@ -63,8 +63,6 @@ typedef struct { grpc_closure *notify; grpc_connect_in_args args; grpc_connect_out_args *result; - grpc_closure initial_string_sent; - grpc_slice_buffer initial_string_buffer; grpc_endpoint *endpoint; // Non-NULL until handshaking starts. @@ -82,7 +80,6 @@ static void chttp2_connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) { chttp2_connector *c = (chttp2_connector *)con; if (gpr_unref(&c->refs)) { - /* c->initial_string_buffer does not need to be destroyed */ gpr_mu_destroy(&c->mu); // If handshaking is not yet in progress, destroy the endpoint. // Otherwise, the handshaker will do this for us. @@ -116,7 +113,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, gpr_mu_lock(&c->mu); if (error != GRPC_ERROR_NONE || c->shutdown) { if (error == GRPC_ERROR_NONE) { - error = GRPC_ERROR_CREATE("connector shutdown"); + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); // We were shut down after handshaking completed successfully, so // destroy the endpoint here. // TODO(ctiller): It is currently necessary to shutdown endpoints @@ -160,28 +157,6 @@ static void start_handshake_locked(grpc_exec_ctx *exec_ctx, c->endpoint = NULL; // Endpoint handed off to handshake manager. } -static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - chttp2_connector *c = arg; - gpr_mu_lock(&c->mu); - if (error != GRPC_ERROR_NONE || c->shutdown) { - if (error == GRPC_ERROR_NONE) { - error = GRPC_ERROR_CREATE("connector shutdown"); - } else { - error = GRPC_ERROR_REF(error); - } - memset(c->result, 0, sizeof(*c->result)); - grpc_closure *notify = c->notify; - c->notify = NULL; - grpc_closure_sched(exec_ctx, notify, error); - gpr_mu_unlock(&c->mu); - chttp2_connector_unref(exec_ctx, arg); - } else { - start_handshake_locked(exec_ctx, c); - gpr_mu_unlock(&c->mu); - } -} - static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { chttp2_connector *c = arg; gpr_mu_lock(&c->mu); @@ -189,7 +164,7 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { c->connecting = false; if (error != GRPC_ERROR_NONE || c->shutdown) { if (error == GRPC_ERROR_NONE) { - error = GRPC_ERROR_CREATE("connector shutdown"); + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); } else { error = GRPC_ERROR_REF(error); } @@ -204,17 +179,7 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { chttp2_connector_unref(exec_ctx, arg); } else { GPR_ASSERT(c->endpoint != NULL); - if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) { - grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent, - c, grpc_schedule_on_exec_ctx); - grpc_slice_buffer_init(&c->initial_string_buffer); - grpc_slice_buffer_add(&c->initial_string_buffer, - c->args.initial_connect_string); - grpc_endpoint_write(exec_ctx, c->endpoint, &c->initial_string_buffer, - &c->initial_string_sent); - } else { - start_handshake_locked(exec_ctx, c); - } + start_handshake_locked(exec_ctx, c); gpr_mu_unlock(&c->mu); } } @@ -226,7 +191,7 @@ static void chttp2_connector_connect(grpc_exec_ctx *exec_ctx, grpc_closure *notify) { chttp2_connector *c = (chttp2_connector *)con; grpc_resolved_address addr; - grpc_get_subchannel_address_arg(args->channel_args, &addr); + grpc_get_subchannel_address_arg(exec_ctx, args->channel_args, &addr); gpr_mu_lock(&c->mu); GPR_ASSERT(c->notify == NULL); c->notify = notify; @@ -248,8 +213,7 @@ static const grpc_connector_vtable chttp2_connector_vtable = { chttp2_connector_connect}; grpc_connector *grpc_chttp2_connector_create() { - chttp2_connector *c = gpr_malloc(sizeof(*c)); - memset(c, 0, sizeof(*c)); + chttp2_connector *c = gpr_zalloc(sizeof(*c)); c->base.vtable = &chttp2_connector_vtable; gpr_mu_init(&c->mu); gpr_ref_init(&c->refs, 1); diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create.c b/src/core/ext/transport/chttp2/client/insecure/channel_create.c index 490a0c560e..067ac35a5a 100644 --- a/src/core/ext/transport/chttp2/client/insecure/channel_create.c +++ b/src/core/ext/transport/chttp2/client/insecure/channel_create.c @@ -72,8 +72,11 @@ static grpc_channel *client_channel_factory_create_channel( grpc_arg arg; arg.type = GRPC_ARG_STRING; arg.key = GRPC_ARG_SERVER_URI; - arg.value.string = grpc_resolver_factory_add_default_prefix_if_needed(target); - grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1); + arg.value.string = + grpc_resolver_factory_add_default_prefix_if_needed(exec_ctx, target); + const char *to_remove[] = {GRPC_ARG_SERVER_URI}; + grpc_channel_args *new_args = + grpc_channel_args_copy_and_add_and_remove(args, to_remove, 1, &arg, 1); gpr_free(arg.value.string); grpc_channel *channel = grpc_channel_create(exec_ctx, target, new_args, GRPC_CLIENT_CHANNEL, NULL); diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c index d8c18eb122..f0c241d68e 100644 --- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c +++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c @@ -83,7 +83,7 @@ static grpc_subchannel_args *get_secure_naming_subchannel_args( const char *server_uri_str = server_uri_arg->value.string; GPR_ASSERT(server_uri_str != NULL); grpc_uri *server_uri = - grpc_uri_parse(server_uri_str, true /* supress errors */); + grpc_uri_parse(exec_ctx, server_uri_str, true /* supress errors */); GPR_ASSERT(server_uri != NULL); const char *server_uri_path; server_uri_path = @@ -96,7 +96,7 @@ static grpc_subchannel_args *get_secure_naming_subchannel_args( const char *target_uri_str = grpc_get_subchannel_address_uri_arg(args->args); grpc_uri *target_uri = - grpc_uri_parse(target_uri_str, false /* suppress errors */); + grpc_uri_parse(exec_ctx, target_uri_str, false /* suppress errors */); GPR_ASSERT(target_uri != NULL); if (target_uri->path[0] != '\0') { // "path" may be empty const grpc_slice key = grpc_slice_from_static_string( @@ -181,8 +181,11 @@ static grpc_channel *client_channel_factory_create_channel( grpc_arg arg; arg.type = GRPC_ARG_STRING; arg.key = GRPC_ARG_SERVER_URI; - arg.value.string = grpc_resolver_factory_add_default_prefix_if_needed(target); - grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1); + arg.value.string = + grpc_resolver_factory_add_default_prefix_if_needed(exec_ctx, target); + const char *to_remove[] = {GRPC_ARG_SERVER_URI}; + grpc_channel_args *new_args = + grpc_channel_args_copy_and_add_and_remove(args, to_remove, 1, &arg, 1); gpr_free(arg.value.string); grpc_channel *channel = grpc_channel_create(exec_ctx, target, new_args, GRPC_CLIENT_CHANNEL, NULL); diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.c b/src/core/ext/transport/chttp2/server/chttp2_server.c index 0fc180d52f..6433968ca5 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.c +++ b/src/core/ext/transport/chttp2/server/chttp2_server.c @@ -222,8 +222,7 @@ grpc_error *grpc_chttp2_server_add_port(grpc_exec_ctx *exec_ctx, if (err != GRPC_ERROR_NONE) { goto error; } - state = gpr_malloc(sizeof(*state)); - memset(state, 0, sizeof(*state)); + state = gpr_zalloc(sizeof(*state)); grpc_closure_init(&state->tcp_server_shutdown_complete, tcp_server_shutdown_complete, state, grpc_schedule_on_exec_ctx); @@ -257,7 +256,7 @@ grpc_error *grpc_chttp2_server_add_port(grpc_exec_ctx *exec_ctx, char *msg; gpr_asprintf(&msg, "No address added out of total %" PRIuPTR " resolved", naddrs); - err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs); + err = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(msg, errors, naddrs); gpr_free(msg); goto error; } else if (count != naddrs) { @@ -265,7 +264,7 @@ grpc_error *grpc_chttp2_server_add_port(grpc_exec_ctx *exec_ctx, gpr_asprintf(&msg, "Only %" PRIuPTR " addresses added out of total %" PRIuPTR " resolved", count, naddrs); - err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs); + err = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(msg, errors, naddrs); gpr_free(msg); const char *warning_message = grpc_error_string(err); diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c index cb2b3f5502..88c813f3be 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c @@ -61,7 +61,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, 3, (server, addr, creds)); // Create security context. if (creds == NULL) { - err = GRPC_ERROR_CREATE( + err = GRPC_ERROR_CREATE_FROM_STATIC_STRING( "No credentials specified for secure server port (creds==NULL)"); goto done; } @@ -72,7 +72,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, gpr_asprintf(&msg, "Unable to create secure server with credentials of type %s.", creds->type); - err = grpc_error_set_int(GRPC_ERROR_CREATE(msg), + err = grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg), GRPC_ERROR_INT_SECURITY_STATUS, status); gpr_free(msg); goto done; diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 1f1627caf7..26f9449f4b 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -48,16 +48,19 @@ #include "src/core/ext/transport/chttp2/transport/varint.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/http/parser.h" +#include "src/core/lib/iomgr/timer.h" #include "src/core/lib/iomgr/workqueue.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" +#include "src/core/lib/support/env.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/error_utils.h" #include "src/core/lib/transport/http2_errors.h" #include "src/core/lib/transport/static_metadata.h" #include "src/core/lib/transport/status_conversion.h" #include "src/core/lib/transport/timeout_encoding.h" +#include "src/core/lib/transport/transport.h" #include "src/core/lib/transport/transport_impl.h" #define DEFAULT_WINDOW 65535 @@ -66,6 +69,16 @@ #define MAX_WRITE_BUFFER_SIZE (64 * 1024 * 1024) #define DEFAULT_MAX_HEADER_LIST_SIZE (16 * 1024) +#define DEFAULT_CLIENT_KEEPALIVE_TIME_S INT_MAX +#define DEFAULT_CLIENT_KEEPALIVE_TIMEOUT_S 20 +#define DEFAULT_KEEPALIVE_PERMIT_WITHOUT_CALLS false + +static int g_default_client_keepalive_time_s = DEFAULT_CLIENT_KEEPALIVE_TIME_S; +static int g_default_client_keepalive_timeout_s = + DEFAULT_CLIENT_KEEPALIVE_TIMEOUT_S; +static bool g_default_keepalive_permit_without_calls = + DEFAULT_KEEPALIVE_PERMIT_WITHOUT_CALLS; + #define MAX_CLIENT_STREAM_ID 0x7fffffffu int grpc_http_trace = 0; int grpc_flowctl_trace = 0; @@ -135,10 +148,22 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_ping_type ping_type, grpc_closure *on_initiate, grpc_closure *on_complete); +static void retry_initiate_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, + grpc_error *error); #define DEFAULT_MIN_TIME_BETWEEN_PINGS_MS 0 #define DEFAULT_MAX_PINGS_BETWEEN_DATA 3 +/** keepalive-relevant functions */ +static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); +static void start_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); +static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); +static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); + /******************************************************************************* * CONSTRUCTION/DESTRUCTION/REFCOUNTING */ @@ -170,7 +195,8 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, GRPC_COMBINER_UNREF(exec_ctx, t->combiner, "chttp2_transport"); - cancel_pings(exec_ctx, t, GRPC_ERROR_CREATE("Transport destroyed")); + cancel_pings(exec_ctx, t, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport destroyed")); while (t->write_cb_pool) { grpc_chttp2_write_cb *next = t->write_cb_pool->next; @@ -218,8 +244,6 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) == GRPC_CHTTP2_CLIENT_CONNECT_STRLEN); - memset(t, 0, sizeof(*t)); - t->base.vtable = &vtable; t->ep = ep; /* one ref is for destroy */ @@ -251,10 +275,23 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_closure_init(&t->destructive_reclaimer_locked, destructive_reclaimer_locked, t, grpc_combiner_scheduler(t->combiner, false)); + grpc_closure_init(&t->retry_initiate_ping_locked, retry_initiate_ping_locked, + t, grpc_combiner_scheduler(t->combiner, false)); grpc_closure_init(&t->start_bdp_ping_locked, start_bdp_ping_locked, t, grpc_combiner_scheduler(t->combiner, false)); grpc_closure_init(&t->finish_bdp_ping_locked, finish_bdp_ping_locked, t, grpc_combiner_scheduler(t->combiner, false)); + grpc_closure_init(&t->init_keepalive_ping_locked, init_keepalive_ping_locked, + t, grpc_combiner_scheduler(t->combiner, false)); + grpc_closure_init(&t->start_keepalive_ping_locked, + start_keepalive_ping_locked, t, + grpc_combiner_scheduler(t->combiner, false)); + grpc_closure_init(&t->finish_keepalive_ping_locked, + finish_keepalive_ping_locked, t, + grpc_combiner_scheduler(t->combiner, false)); + grpc_closure_init(&t->keepalive_watchdog_fired_locked, + keepalive_watchdog_fired_locked, t, + grpc_combiner_scheduler(t->combiner, false)); grpc_bdp_estimator_init(&t->bdp_estimator, t->peer_string); t->last_pid_update = gpr_now(GPR_CLOCK_MONOTONIC); @@ -316,6 +353,19 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_time_from_millis(DEFAULT_MIN_TIME_BETWEEN_PINGS_MS, GPR_TIMESPAN), }; + /* client-side keepalive setting */ + t->keepalive_time = + g_default_client_keepalive_time_s == INT_MAX + ? gpr_inf_future(GPR_TIMESPAN) + : gpr_time_from_seconds(g_default_client_keepalive_time_s, + GPR_TIMESPAN); + t->keepalive_timeout = + g_default_client_keepalive_timeout_s == INT_MAX + ? gpr_inf_future(GPR_TIMESPAN) + : gpr_time_from_seconds(g_default_client_keepalive_timeout_s, + GPR_TIMESPAN); + t->keepalive_permit_without_calls = g_default_keepalive_permit_without_calls; + if (channel_args) { for (i = 0; i < channel_args->num_args; i++) { if (0 == strcmp(channel_args->args[i].key, @@ -363,6 +413,29 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, strcmp(channel_args->args[i].key, GRPC_ARG_HTTP2_BDP_PROBE)) { t->enable_bdp_probe = grpc_channel_arg_get_integer( &channel_args->args[i], (grpc_integer_options){1, 0, 1}); + } else if (0 == strcmp(channel_args->args[i].key, + GRPC_ARG_CLIENT_KEEPALIVE_TIME_S)) { + const int value = grpc_channel_arg_get_integer( + &channel_args->args[i], + (grpc_integer_options){g_default_client_keepalive_time_s, 1, + INT_MAX}); + t->keepalive_time = value == INT_MAX + ? gpr_inf_future(GPR_TIMESPAN) + : gpr_time_from_seconds(value, GPR_TIMESPAN); + } else if (0 == strcmp(channel_args->args[i].key, + GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S)) { + const int value = grpc_channel_arg_get_integer( + &channel_args->args[i], + (grpc_integer_options){g_default_client_keepalive_timeout_s, 0, + INT_MAX}); + t->keepalive_timeout = value == INT_MAX + ? gpr_inf_future(GPR_TIMESPAN) + : gpr_time_from_seconds(value, GPR_TIMESPAN); + } else if (0 == strcmp(channel_args->args[i].key, + GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS)) { + t->keepalive_permit_without_calls = + (uint32_t)grpc_channel_arg_get_integer( + &channel_args->args[i], (grpc_integer_options){0, 0, 1}); } else { static const struct { const char *channel_arg_name; @@ -413,6 +486,17 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; + t->ping_state.is_delayed_ping_timer_set = false; + + /** Start client-side keepalive pings */ + if (t->is_client) { + t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_WAITING; + GRPC_CHTTP2_REF_TRANSPORT(t, "init keepalive ping"); + grpc_timer_init( + exec_ctx, &t->keepalive_ping_timer, + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), t->keepalive_time), + &t->init_keepalive_ping_locked, gpr_now(GPR_CLOCK_MONOTONIC)); + } grpc_chttp2_initiate_write(exec_ctx, t, false, "init"); post_benign_reclaimer(exec_ctx, t); @@ -424,8 +508,9 @@ static void destroy_transport_locked(grpc_exec_ctx *exec_ctx, void *tp, t->destroying = 1; close_transport_locked( exec_ctx, t, - grpc_error_set_int(GRPC_ERROR_CREATE("Transport destroyed"), - GRPC_ERROR_INT_OCCURRED_DURING_WRITE, t->write_state)); + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport destroyed"), + GRPC_ERROR_INT_OCCURRED_DURING_WRITE, t->write_state)); GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "destroy"); } @@ -441,23 +526,40 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error) { if (!t->closed) { + if (!grpc_error_has_clear_grpc_status(error)) { + error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, + GRPC_STATUS_UNAVAILABLE); + } if (t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE) { if (t->close_transport_on_writes_finished == NULL) { t->close_transport_on_writes_finished = - GRPC_ERROR_CREATE("Delayed close due to in-progress write"); + GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Delayed close due to in-progress write"); } t->close_transport_on_writes_finished = grpc_error_add_child(t->close_transport_on_writes_finished, error); return; } - if (!grpc_error_has_clear_grpc_status(error)) { - error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_UNAVAILABLE); - } t->closed = 1; connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "close_transport"); grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error)); + if (t->is_client) { + switch (t->keepalive_state) { + case GRPC_CHTTP2_KEEPALIVE_STATE_WAITING: { + grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer); + break; + } + case GRPC_CHTTP2_KEEPALIVE_STATE_PINGING: { + grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer); + grpc_timer_cancel(exec_ctx, &t->keepalive_watchdog_timer); + break; + } + case GRPC_CHTTP2_KEEPALIVE_STATE_DYING: { + break; + } + } + } /* flush writable stream list to avoid dangling references */ grpc_chttp2_stream *s; @@ -489,13 +591,11 @@ void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s) { static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, grpc_stream *gs, grpc_stream_refcount *refcount, - const void *server_data) { + const void *server_data, gpr_arena *arena) { 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 @@ -504,8 +604,8 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, gpr_ref_init(&s->active_streams, 1); GRPC_CHTTP2_STREAM_REF(s, "chttp2"); - grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0]); - grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[1]); + grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0], arena); + grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[1], arena); grpc_chttp2_data_parser_init(&s->data_parser); grpc_slice_buffer_init(&s->flow_controlled_buffer); s->deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); @@ -581,16 +681,17 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, GPR_TIMER_END("destroy_stream", 0); - gpr_free(s->destroy_stream_arg); + grpc_closure_sched(exec_ctx, s->destroy_stream_arg, GRPC_ERROR_NONE); } static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, - grpc_stream *gs, void *and_free_memory) { + grpc_stream *gs, + grpc_closure *then_schedule_closure) { GPR_TIMER_BEGIN("destroy_stream", 0); grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs; - s->destroy_stream_arg = and_free_memory; + s->destroy_stream_arg = then_schedule_closure; grpc_closure_sched( exec_ctx, grpc_closure_init(&s->destroy_stream, destroy_stream_locked, s, grpc_combiner_scheduler(t->combiner, false)), @@ -749,7 +850,8 @@ static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp, if (t->sent_goaway_state == GRPC_CHTTP2_GOAWAY_SEND_SCHEDULED) { t->sent_goaway_state = GRPC_CHTTP2_GOAWAY_SENT; if (grpc_chttp2_stream_map_size(&t->stream_map) == 0) { - close_transport_locked(exec_ctx, t, GRPC_ERROR_CREATE("goaway sent")); + close_transport_locked( + exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING("goaway sent")); } } @@ -813,22 +915,19 @@ void grpc_chttp2_add_incoming_goaway(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, uint32_t goaway_error, grpc_slice goaway_text) { - char *msg = grpc_dump_slice(goaway_text, GPR_DUMP_HEX | GPR_DUMP_ASCII); - GRPC_CHTTP2_IF_TRACING( - gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg)); - grpc_slice_unref_internal(exec_ctx, goaway_text); + // GRPC_CHTTP2_IF_TRACING( + // gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg)); t->seen_goaway = 1; /* lie: use transient failure from the transport to indicate goaway has been * received */ connectivity_state_set( exec_ctx, t, GRPC_CHANNEL_TRANSIENT_FAILURE, grpc_error_set_str( - grpc_error_set_int(GRPC_ERROR_CREATE("GOAWAY received"), - GRPC_ERROR_INT_HTTP2_ERROR, - (intptr_t)goaway_error), - GRPC_ERROR_STR_RAW_BYTES, msg), + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("GOAWAY received"), + GRPC_ERROR_INT_HTTP2_ERROR, (intptr_t)goaway_error), + GRPC_ERROR_STR_RAW_BYTES, goaway_text), "got_goaway"); - gpr_free(msg); } static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx, @@ -851,9 +950,10 @@ static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx, t->next_stream_id += 2; if (t->next_stream_id >= MAX_CLIENT_STREAM_ID) { - connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_TRANSIENT_FAILURE, - GRPC_ERROR_CREATE("Stream IDs exhausted"), - "no_more_stream_ids"); + connectivity_state_set( + exec_ctx, t, GRPC_CHANNEL_TRANSIENT_FAILURE, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Stream IDs exhausted"), + "no_more_stream_ids"); } grpc_chttp2_stream_map_add(&t->stream_map, s->id, s); @@ -867,9 +967,9 @@ static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx, grpc_chttp2_list_pop_waiting_for_concurrency(t, &s)) { grpc_chttp2_cancel_stream( exec_ctx, t, s, - grpc_error_set_int(GRPC_ERROR_CREATE("Stream IDs exhausted"), - GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_UNAVAILABLE)); + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Stream IDs exhausted"), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE)); } } @@ -918,11 +1018,11 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx, } if (error != GRPC_ERROR_NONE) { if (closure->error_data.error == GRPC_ERROR_NONE) { - closure->error_data.error = - GRPC_ERROR_CREATE("Error in HTTP transport completing operation"); - closure->error_data.error = - grpc_error_set_str(closure->error_data.error, - GRPC_ERROR_STR_TARGET_ADDRESS, t->peer_string); + closure->error_data.error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Error in HTTP transport completing operation"); + closure->error_data.error = grpc_error_set_str( + closure->error_data.error, GRPC_ERROR_STR_TARGET_ADDRESS, + grpc_slice_from_copied_string(t->peer_string)); } closure->error_data.error = grpc_error_add_child(closure->error_data.error, error); @@ -1097,10 +1197,11 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, exec_ctx, t, s, grpc_error_set_int( grpc_error_set_int( - grpc_error_set_int( - GRPC_ERROR_CREATE("to-be-sent initial metadata size " - "exceeds peer limit"), - GRPC_ERROR_INT_SIZE, (intptr_t)metadata_size), + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "to-be-sent initial metadata size " + "exceeds peer limit"), + GRPC_ERROR_INT_SIZE, + (intptr_t)metadata_size), GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED)); } else { @@ -1116,21 +1217,26 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, } else { grpc_chttp2_cancel_stream( exec_ctx, t, s, - grpc_error_set_int(GRPC_ERROR_CREATE("Transport closed"), - GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_UNAVAILABLE)); + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed"), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE)); } } else { GPR_ASSERT(s->id != 0); - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED, + grpc_chttp2_stream_write_type write_type = + GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED; + if (op->send_message != NULL && + (op->send_message->flags & GRPC_WRITE_BUFFER_HINT)) { + write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK; + } + grpc_chttp2_become_writable(exec_ctx, t, s, write_type, "op.send_initial_metadata"); } } else { s->send_initial_metadata = NULL; grpc_chttp2_complete_closure_step( exec_ctx, t, s, &s->send_initial_metadata_finished, - GRPC_ERROR_CREATE_REFERENCING( + GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Attempt to send initial metadata after stream was closed", &s->write_closed_error, 1), "send_initial_metadata_finished"); @@ -1144,7 +1250,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, if (s->write_closed) { grpc_chttp2_complete_closure_step( exec_ctx, t, s, &s->fetching_send_message_finished, - GRPC_ERROR_CREATE_REFERENCING( + GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Attempt to send message after stream was closed", &s->write_closed_error, 1), "fetching_send_message_finished"); @@ -1192,10 +1298,11 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, exec_ctx, t, s, grpc_error_set_int( grpc_error_set_int( - grpc_error_set_int( - GRPC_ERROR_CREATE("to-be-sent trailing metadata size " - "exceeds peer limit"), - GRPC_ERROR_INT_SIZE, (intptr_t)metadata_size), + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "to-be-sent trailing metadata size " + "exceeds peer limit"), + GRPC_ERROR_INT_SIZE, + (intptr_t)metadata_size), GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED)); } else { @@ -1208,8 +1315,9 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, exec_ctx, t, s, &s->send_trailing_metadata_finished, grpc_metadata_batch_is_empty(op->send_trailing_metadata) ? GRPC_ERROR_NONE - : GRPC_ERROR_CREATE("Attempt to send trailing metadata after " - "stream was closed"), + : GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Attempt to send trailing metadata after " + "stream was closed"), "send_trailing_metadata_finished"); } else if (s->id != 0) { /* TODO(ctiller): check if there's flow control for any outstanding @@ -1304,6 +1412,13 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } } +static void retry_initiate_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, + grpc_error *error) { + grpc_chttp2_transport *t = tp; + t->ping_state.is_delayed_ping_timer_set = false; + grpc_chttp2_initiate_write(exec_ctx, t, false, "retry_send_ping"); +} + void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, uint64_t id) { grpc_chttp2_ping_queue *pq = @@ -1324,11 +1439,11 @@ static void send_goaway(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error) { t->sent_goaway_state = GRPC_CHTTP2_GOAWAY_SEND_SCHEDULED; grpc_http2_error_code http_error; - const char *msg; - grpc_error_get_status(error, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL, &msg, - &http_error); + grpc_slice slice; + grpc_error_get_status(error, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL, + &slice, &http_error); grpc_chttp2_goaway_append(t->last_new_stream_id, (uint32_t)http_error, - grpc_slice_from_copied_string(msg), &t->qbuf); + grpc_slice_ref_internal(slice), &t->qbuf); grpc_chttp2_initiate_write(exec_ctx, t, false, "goaway_sent"); GRPC_ERROR_UNREF(error); } @@ -1488,7 +1603,7 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, if (t->sent_goaway_state == GRPC_CHTTP2_GOAWAY_SENT) { close_transport_locked( exec_ctx, t, - GRPC_ERROR_CREATE_REFERENCING( + GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Last stream closed after sending GOAWAY", &error, 1)); } } @@ -1529,8 +1644,8 @@ void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx, void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_error *error) { grpc_status_code status; - const char *msg; - grpc_error_get_status(error, s->deadline, &status, &msg, NULL); + grpc_slice slice; + grpc_error_get_status(error, s->deadline, &status, &slice, NULL); if (status != GRPC_STATUS_OK) { s->seen_error = true; @@ -1545,15 +1660,19 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, s->recv_trailing_metadata_finished != NULL) { char status_string[GPR_LTOA_MIN_BUFSIZE]; gpr_ltoa(status, status_string); - grpc_chttp2_incoming_metadata_buffer_replace_or_add( - exec_ctx, &s->metadata_buffer[1], - grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_STATUS, - grpc_slice_from_copied_string(status_string))); - if (msg != NULL) { - grpc_chttp2_incoming_metadata_buffer_replace_or_add( - exec_ctx, &s->metadata_buffer[1], - grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_MESSAGE, - grpc_slice_from_copied_string(msg))); + GRPC_LOG_IF_ERROR("add_status", + grpc_chttp2_incoming_metadata_buffer_replace_or_add( + exec_ctx, &s->metadata_buffer[1], + grpc_mdelem_from_slices( + exec_ctx, GRPC_MDSTR_GRPC_STATUS, + grpc_slice_from_copied_string(status_string)))); + if (!GRPC_SLICE_IS_EMPTY(slice)) { + GRPC_LOG_IF_ERROR( + "add_status_message", + grpc_chttp2_incoming_metadata_buffer_replace_or_add( + exec_ctx, &s->metadata_buffer[1], + grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_MESSAGE, + grpc_slice_ref_internal(slice)))); } s->published_metadata[1] = GRPC_METADATA_SYNTHESIZED_FROM_FAKE; grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s); @@ -1582,7 +1701,8 @@ static grpc_error *removal_error(grpc_error *extra_error, grpc_chttp2_stream *s, add_error(extra_error, refs, &nrefs); grpc_error *error = GRPC_ERROR_NONE; if (nrefs > 0) { - error = GRPC_ERROR_CREATE_REFERENCING(master_error_msg, refs, nrefs); + error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(master_error_msg, + refs, nrefs); } GRPC_ERROR_UNREF(extra_error); return error; @@ -1676,12 +1796,13 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_error *error) { grpc_slice hdr; grpc_slice status_hdr; + grpc_slice http_status_hdr; grpc_slice message_pfx; uint8_t *p; uint32_t len = 0; grpc_status_code grpc_status; - const char *msg; - grpc_error_get_status(error, s->deadline, &grpc_status, &msg, NULL); + grpc_slice slice; + grpc_error_get_status(error, s->deadline, &grpc_status, &slice, NULL); GPR_ASSERT(grpc_status >= 0 && (int)grpc_status < 100); @@ -1691,6 +1812,26 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, It's complicated by the fact that our send machinery would be dead by the time we got around to sending this, so instead we ignore HPACK compression and just write the uncompressed bytes onto the wire. */ + if (!s->sent_initial_metadata) { + http_status_hdr = grpc_slice_malloc(13); + p = GRPC_SLICE_START_PTR(http_status_hdr); + *p++ = 0x00; + *p++ = 7; + *p++ = ':'; + *p++ = 's'; + *p++ = 't'; + *p++ = 'a'; + *p++ = 't'; + *p++ = 'u'; + *p++ = 's'; + *p++ = 3; + *p++ = '2'; + *p++ = '0'; + *p++ = '0'; + GPR_ASSERT(p == GRPC_SLICE_END_PTR(http_status_hdr)); + len += (uint32_t)GRPC_SLICE_LENGTH(http_status_hdr); + } + status_hdr = grpc_slice_malloc(15 + (grpc_status >= 10)); p = GRPC_SLICE_START_PTR(status_hdr); *p++ = 0x00; /* literal header, not indexed */ @@ -1717,32 +1858,30 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GPR_ASSERT(p == GRPC_SLICE_END_PTR(status_hdr)); len += (uint32_t)GRPC_SLICE_LENGTH(status_hdr); - if (msg != NULL) { - size_t msg_len = strlen(msg); - GPR_ASSERT(msg_len <= UINT32_MAX); - uint32_t msg_len_len = GRPC_CHTTP2_VARINT_LENGTH((uint32_t)msg_len, 0); - message_pfx = grpc_slice_malloc(14 + msg_len_len); - p = GRPC_SLICE_START_PTR(message_pfx); - *p++ = 0x00; /* literal header, not indexed */ - *p++ = 12; /* len(grpc-message) */ - *p++ = 'g'; - *p++ = 'r'; - *p++ = 'p'; - *p++ = 'c'; - *p++ = '-'; - *p++ = 'm'; - *p++ = 'e'; - *p++ = 's'; - *p++ = 's'; - *p++ = 'a'; - *p++ = 'g'; - *p++ = 'e'; - GRPC_CHTTP2_WRITE_VARINT((uint32_t)msg_len, 0, 0, p, (uint32_t)msg_len_len); - p += msg_len_len; - GPR_ASSERT(p == GRPC_SLICE_END_PTR(message_pfx)); - len += (uint32_t)GRPC_SLICE_LENGTH(message_pfx); - len += (uint32_t)msg_len; - } + size_t msg_len = GRPC_SLICE_LENGTH(slice); + GPR_ASSERT(msg_len <= UINT32_MAX); + uint32_t msg_len_len = GRPC_CHTTP2_VARINT_LENGTH((uint32_t)msg_len, 1); + message_pfx = grpc_slice_malloc(14 + msg_len_len); + p = GRPC_SLICE_START_PTR(message_pfx); + *p++ = 0x00; /* literal header, not indexed */ + *p++ = 12; /* len(grpc-message) */ + *p++ = 'g'; + *p++ = 'r'; + *p++ = 'p'; + *p++ = 'c'; + *p++ = '-'; + *p++ = 'm'; + *p++ = 'e'; + *p++ = 's'; + *p++ = 's'; + *p++ = 'a'; + *p++ = 'g'; + *p++ = 'e'; + GRPC_CHTTP2_WRITE_VARINT((uint32_t)msg_len, 1, 0, p, (uint32_t)msg_len_len); + p += msg_len_len; + GPR_ASSERT(p == GRPC_SLICE_END_PTR(message_pfx)); + len += (uint32_t)GRPC_SLICE_LENGTH(message_pfx); + len += (uint32_t)msg_len; hdr = grpc_slice_malloc(9); p = GRPC_SLICE_START_PTR(hdr); @@ -1758,11 +1897,12 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GPR_ASSERT(p == GRPC_SLICE_END_PTR(hdr)); grpc_slice_buffer_add(&t->qbuf, hdr); - grpc_slice_buffer_add(&t->qbuf, status_hdr); - if (msg != NULL) { - grpc_slice_buffer_add(&t->qbuf, message_pfx); - grpc_slice_buffer_add(&t->qbuf, grpc_slice_from_copied_string(msg)); + if (!s->sent_initial_metadata) { + grpc_slice_buffer_add(&t->qbuf, http_status_hdr); } + grpc_slice_buffer_add(&t->qbuf, status_hdr); + grpc_slice_buffer_add(&t->qbuf, message_pfx); + grpc_slice_buffer_add(&t->qbuf, grpc_slice_ref_internal(slice)); grpc_slice_buffer_add( &t->qbuf, grpc_chttp2_rst_stream_create(s->id, GRPC_HTTP2_NO_ERROR, &s->stats.outgoing)); @@ -1837,9 +1977,9 @@ static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx, if (parse_error == GRPC_ERROR_NONE && (parse_error = grpc_http_parser_eof(&parser)) == GRPC_ERROR_NONE) { error = grpc_error_set_int( - grpc_error_set_int( - GRPC_ERROR_CREATE("Trying to connect an http1.x server"), - GRPC_ERROR_INT_HTTP_STATUS, response.status), + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Trying to connect an http1.x server"), + GRPC_ERROR_INT_HTTP_STATUS, response.status), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE); } GRPC_ERROR_UNREF(parse_error); @@ -1860,9 +2000,10 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *err = error; if (err != GRPC_ERROR_NONE) { - err = grpc_error_set_int( - GRPC_ERROR_CREATE_REFERENCING("Endpoint read failed", &err, 1), - GRPC_ERROR_INT_OCCURRED_DURING_WRITE, t->write_state); + err = grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( + "Endpoint read failed", &err, 1), + GRPC_ERROR_INT_OCCURRED_DURING_WRITE, + t->write_state); } GPR_SWAP(grpc_error *, err, error); GRPC_ERROR_UNREF(err); @@ -1883,8 +2024,8 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, if (errors[1] != GRPC_ERROR_NONE) { errors[2] = try_http_parsing(exec_ctx, t); GRPC_ERROR_UNREF(error); - error = GRPC_ERROR_CREATE_REFERENCING("Failed parsing HTTP/2", errors, - GPR_ARRAY_SIZE(errors)); + error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( + "Failed parsing HTTP/2", errors, GPR_ARRAY_SIZE(errors)); } for (i = 0; i < GPR_ARRAY_SIZE(errors); i++) { GRPC_ERROR_UNREF(errors[i]); @@ -1909,7 +2050,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, GPR_TIMER_BEGIN("post_reading_action_locked", 0); bool keep_reading = false; if (error == GRPC_ERROR_NONE && t->closed) { - error = GRPC_ERROR_CREATE("Transport closed"); + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed"); } if (error != GRPC_ERROR_NONE) { close_transport_locked(exec_ctx, t, GRPC_ERROR_REF(error)); @@ -1987,6 +2128,101 @@ static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping"); } +void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args) { + size_t i; + if (args) { + for (i = 0; i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_CLIENT_KEEPALIVE_TIME_S)) { + g_default_client_keepalive_time_s = grpc_channel_arg_get_integer( + &args->args[i], (grpc_integer_options){ + g_default_client_keepalive_time_s, 1, INT_MAX}); + } else if (0 == strcmp(args->args[i].key, + GRPC_ARG_CLIENT_KEEPALIVE_TIMEOUT_S)) { + g_default_client_keepalive_timeout_s = grpc_channel_arg_get_integer( + &args->args[i], + (grpc_integer_options){g_default_client_keepalive_timeout_s, 0, + INT_MAX}); + } else if (0 == strcmp(args->args[i].key, + GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS)) { + g_default_keepalive_permit_without_calls = + (uint32_t)grpc_channel_arg_get_integer( + &args->args[i], + (grpc_integer_options){g_default_keepalive_permit_without_calls, + 0, 1}); + } + } + } +} + +static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_transport *t = arg; + GPR_ASSERT(t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING); + if (error == GRPC_ERROR_NONE && !(t->destroying || t->closed)) { + if (t->keepalive_permit_without_calls || t->stream_map.count > 0) { + t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_PINGING; + GRPC_CHTTP2_REF_TRANSPORT(t, "keepalive ping end"); + send_ping_locked(exec_ctx, t, GRPC_CHTTP2_PING_ON_NEXT_WRITE, + &t->start_keepalive_ping_locked, + &t->finish_keepalive_ping_locked); + } else { + GRPC_CHTTP2_REF_TRANSPORT(t, "init keepalive ping"); + grpc_timer_init( + exec_ctx, &t->keepalive_ping_timer, + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), t->keepalive_time), + &t->init_keepalive_ping_locked, gpr_now(GPR_CLOCK_MONOTONIC)); + } + } + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "init keepalive ping"); +} + +static void start_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_transport *t = arg; + GRPC_CHTTP2_REF_TRANSPORT(t, "keepalive watchdog"); + grpc_timer_init( + exec_ctx, &t->keepalive_watchdog_timer, + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), t->keepalive_timeout), + &t->keepalive_watchdog_fired_locked, gpr_now(GPR_CLOCK_MONOTONIC)); +} + +static void finish_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_transport *t = arg; + if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) { + if (error == GRPC_ERROR_NONE) { + t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_WAITING; + grpc_timer_cancel(exec_ctx, &t->keepalive_watchdog_timer); + GRPC_CHTTP2_REF_TRANSPORT(t, "init keepalive ping"); + grpc_timer_init( + exec_ctx, &t->keepalive_ping_timer, + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), t->keepalive_time), + &t->init_keepalive_ping_locked, gpr_now(GPR_CLOCK_MONOTONIC)); + } + } + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "keepalive ping end"); +} + +static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_transport *t = arg; + if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) { + if (error == GRPC_ERROR_NONE) { + t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING; + close_transport_locked(exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "keepalive watchdog timeout")); + } + } else { + /** The watchdog timer should have been cancelled by + finish_keepalive_ping_locked. */ + if (error != GRPC_ERROR_CANCELLED) { + gpr_log(GPR_ERROR, "keepalive_ping_end state error: %d (expect: %d)", + t->keepalive_state, GRPC_CHTTP2_KEEPALIVE_STATE_PINGING); + } + } + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "keepalive watchdog"); +} + /******************************************************************************* * CALLBACK LOOP */ @@ -2173,7 +2409,8 @@ void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx, gpr_mu_lock(&bs->slice_mu); if (bs->remaining_bytes < GRPC_SLICE_LENGTH(slice)) { incoming_byte_stream_publish_error( - exec_ctx, bs, GRPC_ERROR_CREATE("Too many bytes in stream")); + exec_ctx, bs, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many bytes in stream")); } else { bs->remaining_bytes -= (uint32_t)GRPC_SLICE_LENGTH(slice); if (bs->on_next != NULL) { @@ -2193,7 +2430,7 @@ void grpc_chttp2_incoming_byte_stream_finished( if (error == GRPC_ERROR_NONE) { gpr_mu_lock(&bs->slice_mu); if (bs->remaining_bytes != 0) { - error = GRPC_ERROR_CREATE("Truncated message"); + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); } gpr_mu_unlock(&bs->slice_mu); } @@ -2273,9 +2510,9 @@ static void benign_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, t->peer_string); } send_goaway(exec_ctx, t, - grpc_error_set_int(GRPC_ERROR_CREATE("Buffers full"), - GRPC_ERROR_INT_HTTP2_ERROR, - GRPC_HTTP2_ENHANCE_YOUR_CALM)); + grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Buffers full"), + GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_ENHANCE_YOUR_CALM)); } else if (error == GRPC_ERROR_NONE && grpc_resource_quota_trace) { gpr_log(GPR_DEBUG, "HTTP2: %s - skip benign reclamation, there are still %" PRIdPTR @@ -2302,9 +2539,10 @@ static void destructive_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, s->id); } grpc_chttp2_cancel_stream( - exec_ctx, t, s, grpc_error_set_int(GRPC_ERROR_CREATE("Buffers full"), - GRPC_ERROR_INT_HTTP2_ERROR, - GRPC_HTTP2_ENHANCE_YOUR_CALM)); + exec_ctx, t, s, + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("Buffers full"), + GRPC_ERROR_INT_HTTP2_ERROR, + GRPC_HTTP2_ENHANCE_YOUR_CALM)); if (n > 1) { /* Since we cancel one stream per destructive reclamation, if there are more streams left, we can immediately post a new @@ -2428,7 +2666,7 @@ static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream), grpc_transport *grpc_create_chttp2_transport( grpc_exec_ctx *exec_ctx, const grpc_channel_args *channel_args, grpc_endpoint *ep, int is_client) { - grpc_chttp2_transport *t = gpr_malloc(sizeof(grpc_chttp2_transport)); + grpc_chttp2_transport *t = gpr_zalloc(sizeof(grpc_chttp2_transport)); init_transport(exec_ctx, t, channel_args, ep, is_client != 0); return &t->base; } diff --git a/src/core/ext/transport/chttp2/transport/frame_data.c b/src/core/ext/transport/chttp2/transport/frame_data.c index f9b9e1b309..6e9258ee7e 100644 --- a/src/core/ext/transport/chttp2/transport/frame_data.c +++ b/src/core/ext/transport/chttp2/transport/frame_data.c @@ -54,7 +54,8 @@ void grpc_chttp2_data_parser_destroy(grpc_exec_ctx *exec_ctx, grpc_chttp2_data_parser *parser) { if (parser->parsing_frame != NULL) { grpc_chttp2_incoming_byte_stream_finished( - exec_ctx, parser->parsing_frame, GRPC_ERROR_CREATE("Parser destroyed")); + exec_ctx, parser->parsing_frame, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Parser destroyed")); } GRPC_ERROR_UNREF(parser->error); } @@ -65,8 +66,9 @@ grpc_error *grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser *parser, if (flags & ~GRPC_CHTTP2_DATA_FLAG_END_STREAM) { char *msg; gpr_asprintf(&msg, "unsupported data flags: 0x%02x", flags); - grpc_error *err = grpc_error_set_int( - GRPC_ERROR_CREATE(msg), GRPC_ERROR_INT_STREAM_ID, (intptr_t)stream_id); + grpc_error *err = + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg), + GRPC_ERROR_INT_STREAM_ID, (intptr_t)stream_id); gpr_free(msg); return err; } @@ -173,13 +175,13 @@ static grpc_error *parse_inner(grpc_exec_ctx *exec_ctx, break; default: gpr_asprintf(&msg, "Bad GRPC frame type 0x%02x", p->frame_type); - p->error = GRPC_ERROR_CREATE(msg); + p->error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); p->error = grpc_error_set_int(p->error, GRPC_ERROR_INT_STREAM_ID, (intptr_t)s->id); gpr_free(msg); msg = grpc_dump_slice(slice, GPR_DUMP_HEX | GPR_DUMP_ASCII); - p->error = - grpc_error_set_str(p->error, GRPC_ERROR_STR_RAW_BYTES, msg); + p->error = grpc_error_set_str(p->error, GRPC_ERROR_STR_RAW_BYTES, + grpc_slice_from_copied_string(msg)); gpr_free(msg); p->error = grpc_error_set_int(p->error, GRPC_ERROR_INT_OFFSET, cur - beg); @@ -265,7 +267,8 @@ static grpc_error *parse_inner(grpc_exec_ctx *exec_ctx, } } - GPR_UNREACHABLE_CODE(return GRPC_ERROR_CREATE("Should never reach here")); + GPR_UNREACHABLE_CODE( + return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Should never reach here")); } grpc_error *grpc_chttp2_data_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, diff --git a/src/core/ext/transport/chttp2/transport/frame_goaway.c b/src/core/ext/transport/chttp2/transport/frame_goaway.c index d99d486c1b..001271dd22 100644 --- a/src/core/ext/transport/chttp2/transport/frame_goaway.c +++ b/src/core/ext/transport/chttp2/transport/frame_goaway.c @@ -54,7 +54,7 @@ grpc_error *grpc_chttp2_goaway_parser_begin_frame(grpc_chttp2_goaway_parser *p, if (length < 8) { char *msg; gpr_asprintf(&msg, "goaway frame too short (%d bytes)", length); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -156,7 +156,8 @@ grpc_error *grpc_chttp2_goaway_parser_parse(grpc_exec_ctx *exec_ctx, } return GRPC_ERROR_NONE; } - GPR_UNREACHABLE_CODE(return GRPC_ERROR_CREATE("Should never reach here")); + GPR_UNREACHABLE_CODE( + return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Should never reach here")); } void grpc_chttp2_goaway_append(uint32_t last_stream_id, uint32_t error_code, diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.c b/src/core/ext/transport/chttp2/transport/frame_ping.c index f487533c41..46dafdb62f 100644 --- a/src/core/ext/transport/chttp2/transport/frame_ping.c +++ b/src/core/ext/transport/chttp2/transport/frame_ping.c @@ -40,6 +40,8 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> +static bool g_disable_ping_ack = false; + grpc_slice grpc_chttp2_ping_create(uint8_t ack, uint64_t opaque_8bytes) { grpc_slice slice = grpc_slice_malloc(9 + 8); uint8_t *p = GRPC_SLICE_START_PTR(slice); @@ -71,7 +73,7 @@ grpc_error *grpc_chttp2_ping_parser_begin_frame(grpc_chttp2_ping_parser *parser, if (flags & 0xfe || length != 8) { char *msg; gpr_asprintf(&msg, "invalid ping: length=%d, flags=%02x", length, flags); - grpc_error *error = GRPC_ERROR_CREATE(msg); + grpc_error *error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return error; } @@ -91,7 +93,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_ping_parser *p = parser; while (p->byte != 8 && cur != end) { - p->opaque_8bytes |= (((uint64_t)*cur) << (8 * p->byte)); + p->opaque_8bytes |= (((uint64_t)*cur) << (56 - 8 * p->byte)); cur++; p->byte++; } @@ -101,15 +103,21 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, if (p->is_ack) { grpc_chttp2_ack_ping(exec_ctx, t, p->opaque_8bytes); } else { - if (t->ping_ack_count == t->ping_ack_capacity) { - t->ping_ack_capacity = GPR_MAX(t->ping_ack_capacity * 3 / 2, 3); - t->ping_acks = gpr_realloc( - t->ping_acks, t->ping_ack_capacity * sizeof(*t->ping_acks)); + if (!g_disable_ping_ack) { + if (t->ping_ack_count == t->ping_ack_capacity) { + t->ping_ack_capacity = GPR_MAX(t->ping_ack_capacity * 3 / 2, 3); + t->ping_acks = gpr_realloc( + t->ping_acks, t->ping_ack_capacity * sizeof(*t->ping_acks)); + } + t->ping_acks[t->ping_ack_count++] = p->opaque_8bytes; + grpc_chttp2_initiate_write(exec_ctx, t, false, "ping response"); } - t->ping_acks[t->ping_ack_count++] = p->opaque_8bytes; - grpc_chttp2_initiate_write(exec_ctx, t, false, "ping response"); } } return GRPC_ERROR_NONE; } + +void grpc_set_disable_ping_ack(bool disable_ping_ack) { + g_disable_ping_ack = disable_ping_ack; +} diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.h b/src/core/ext/transport/chttp2/transport/frame_ping.h index ef642465d7..01983d2b12 100644 --- a/src/core/ext/transport/chttp2/transport/frame_ping.h +++ b/src/core/ext/transport/chttp2/transport/frame_ping.h @@ -53,4 +53,7 @@ grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_stream *s, grpc_slice slice, int is_last); +/* Test-only function for disabling ping ack */ +void grpc_set_disable_ping_ack(bool disable_ping_ack); + #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_PING_H */ diff --git a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c index cb017e75f0..225f15c77c 100644 --- a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c +++ b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c @@ -76,7 +76,7 @@ grpc_error *grpc_chttp2_rst_stream_parser_begin_frame( char *msg; gpr_asprintf(&msg, "invalid rst_stream: length=%d, flags=%02x", length, flags); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -112,8 +112,9 @@ grpc_error *grpc_chttp2_rst_stream_parser_parse(grpc_exec_ctx *exec_ctx, char *message; gpr_asprintf(&message, "Received RST_STREAM with error code %d", reason); error = grpc_error_set_int( - grpc_error_set_str(GRPC_ERROR_CREATE("RST_STREAM"), - GRPC_ERROR_STR_GRPC_MESSAGE, message), + grpc_error_set_str(GRPC_ERROR_CREATE_FROM_STATIC_STRING("RST_STREAM"), + GRPC_ERROR_STR_GRPC_MESSAGE, + grpc_slice_from_copied_string(message)), GRPC_ERROR_INT_HTTP2_ERROR, (intptr_t)reason); gpr_free(message); } diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.c b/src/core/ext/transport/chttp2/transport/frame_settings.c index 82290e34cd..16881c0707 100644 --- a/src/core/ext/transport/chttp2/transport/frame_settings.c +++ b/src/core/ext/transport/chttp2/transport/frame_settings.c @@ -131,13 +131,16 @@ grpc_error *grpc_chttp2_settings_parser_begin_frame( if (flags == GRPC_CHTTP2_FLAG_ACK) { parser->is_ack = 1; if (length != 0) { - return GRPC_ERROR_CREATE("non-empty settings ack frame received"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "non-empty settings ack frame received"); } return GRPC_ERROR_NONE; } else if (flags != 0) { - return GRPC_ERROR_CREATE("invalid flags on settings frame"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "invalid flags on settings frame"); } else if (length % 6 != 0) { - return GRPC_ERROR_CREATE("settings frames must be a multiple of six bytes"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "settings frames must be a multiple of six bytes"); } else { return GRPC_ERROR_NONE; } @@ -229,7 +232,7 @@ grpc_error *grpc_chttp2_settings_parser_parse(grpc_exec_ctx *exec_ctx, void *p, &t->qbuf); gpr_asprintf(&msg, "invalid value %u passed for %s", parser->value, sp->name); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } diff --git a/src/core/ext/transport/chttp2/transport/frame_window_update.c b/src/core/ext/transport/chttp2/transport/frame_window_update.c index 8fa0bb471a..b76b6f6f47 100644 --- a/src/core/ext/transport/chttp2/transport/frame_window_update.c +++ b/src/core/ext/transport/chttp2/transport/frame_window_update.c @@ -70,7 +70,7 @@ grpc_error *grpc_chttp2_window_update_parser_begin_frame( char *msg; gpr_asprintf(&msg, "invalid window update: length=%d, flags=%02x", length, flags); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -102,7 +102,7 @@ grpc_error *grpc_chttp2_window_update_parser_parse( if (received_update == 0 || (received_update & 0x80000000u)) { char *msg; gpr_asprintf(&msg, "invalid window update bytes: %d", p->amount); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } diff --git a/src/core/ext/transport/chttp2/transport/hpack_encoder.c b/src/core/ext/transport/chttp2/transport/hpack_encoder.c index 63df8e135f..84586cd998 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_encoder.c +++ b/src/core/ext/transport/chttp2/transport/hpack_encoder.c @@ -173,6 +173,7 @@ static void add_header_data(framer_state *st, grpc_slice slice) { static uint8_t *add_tiny_header_data(framer_state *st, size_t len) { ensure_space(st, len); + st->stats->header_bytes += len; return grpc_slice_buffer_tiny_add(st->output, len); } diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.c b/src/core/ext/transport/chttp2/transport/hpack_parser.c index 40f5120308..5099d736bf 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_parser.c +++ b/src/core/ext/transport/chttp2/transport/hpack_parser.c @@ -693,7 +693,7 @@ static grpc_error *on_hdr(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, } if (p->on_header == NULL) { GRPC_MDELEM_UNREF(exec_ctx, md); - return GRPC_ERROR_CREATE("on_header callback not set"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING("on_header callback not set"); } p->on_header(exec_ctx, p->on_header_user_data, md); return GRPC_ERROR_NONE; @@ -810,7 +810,8 @@ static grpc_error *finish_indexed_field(grpc_exec_ctx *exec_ctx, grpc_mdelem md = grpc_chttp2_hptbl_lookup(&p->table, p->index); if (GRPC_MDISNULL(md)) { return grpc_error_set_int( - grpc_error_set_int(GRPC_ERROR_CREATE("Invalid HPACK index received"), + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Invalid HPACK index received"), GRPC_ERROR_INT_INDEX, (intptr_t)p->index), GRPC_ERROR_INT_SIZE, (intptr_t)p->table.num_ents); } @@ -1074,7 +1075,7 @@ static grpc_error *parse_max_tbl_size(grpc_exec_ctx *exec_ctx, if (p->dynamic_table_update_allowed == 0) { return parse_error( exec_ctx, p, cur, end, - GRPC_ERROR_CREATE( + GRPC_ERROR_CREATE_FROM_STATIC_STRING( "More than two max table size changes in a single frame")); } p->dynamic_table_update_allowed--; @@ -1092,7 +1093,7 @@ static grpc_error *parse_max_tbl_size_x(grpc_exec_ctx *exec_ctx, if (p->dynamic_table_update_allowed == 0) { return parse_error( exec_ctx, p, cur, end, - GRPC_ERROR_CREATE( + GRPC_ERROR_CREATE_FROM_STATIC_STRING( "More than two max table size changes in a single frame")); } p->dynamic_table_update_allowed--; @@ -1126,7 +1127,7 @@ static grpc_error *parse_illegal_op(grpc_exec_ctx *exec_ctx, GPR_ASSERT(cur != end); char *msg; gpr_asprintf(&msg, "Illegal hpack op code %d", *cur); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return parse_error(exec_ctx, p, cur, end, err); } @@ -1246,7 +1247,7 @@ error: "integer overflow in hpack integer decoding: have 0x%08x, " "got byte 0x%02x on byte 5", *p->parsing.value, *cur); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return parse_error(exec_ctx, p, cur, end, err); } @@ -1275,7 +1276,7 @@ static grpc_error *parse_value5up(grpc_exec_ctx *exec_ctx, "integer overflow in hpack integer decoding: have 0x%08x, " "got byte 0x%02x sometime after byte 5", *p->parsing.value, *cur); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return parse_error(exec_ctx, p, cur, end, err); } @@ -1333,8 +1334,9 @@ static grpc_error *append_string(grpc_exec_ctx *exec_ctx, bits = inverse_base64[*cur]; ++cur; if (bits == 255) - return parse_error(exec_ctx, p, cur, end, - GRPC_ERROR_CREATE("Illegal base64 character")); + return parse_error( + exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal base64 character")); else if (bits == 64) goto b64_byte0; p->base64_buffer = bits << 18; @@ -1348,8 +1350,9 @@ static grpc_error *append_string(grpc_exec_ctx *exec_ctx, bits = inverse_base64[*cur]; ++cur; if (bits == 255) - return parse_error(exec_ctx, p, cur, end, - GRPC_ERROR_CREATE("Illegal base64 character")); + return parse_error( + exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal base64 character")); else if (bits == 64) goto b64_byte1; p->base64_buffer |= bits << 12; @@ -1363,8 +1366,9 @@ static grpc_error *append_string(grpc_exec_ctx *exec_ctx, bits = inverse_base64[*cur]; ++cur; if (bits == 255) - return parse_error(exec_ctx, p, cur, end, - GRPC_ERROR_CREATE("Illegal base64 character")); + return parse_error( + exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal base64 character")); else if (bits == 64) goto b64_byte2; p->base64_buffer |= bits << 6; @@ -1378,8 +1382,9 @@ static grpc_error *append_string(grpc_exec_ctx *exec_ctx, bits = inverse_base64[*cur]; ++cur; if (bits == 255) - return parse_error(exec_ctx, p, cur, end, - GRPC_ERROR_CREATE("Illegal base64 character")); + return parse_error( + exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Illegal base64 character")); else if (bits == 64) goto b64_byte3; p->base64_buffer |= bits; @@ -1391,7 +1396,8 @@ static grpc_error *append_string(grpc_exec_ctx *exec_ctx, goto b64_byte0; } GPR_UNREACHABLE_CODE(return parse_error( - exec_ctx, p, cur, end, GRPC_ERROR_CREATE("Should never reach here"))); + exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Should never reach here"))); } static grpc_error *finish_str(grpc_exec_ctx *exec_ctx, @@ -1406,16 +1412,16 @@ static grpc_error *finish_str(grpc_exec_ctx *exec_ctx, case B64_BYTE0: break; case B64_BYTE1: - return parse_error( - exec_ctx, p, cur, end, - GRPC_ERROR_CREATE("illegal base64 encoding")); /* illegal encoding */ + return parse_error(exec_ctx, p, cur, end, + GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "illegal base64 encoding")); /* illegal encoding */ case B64_BYTE2: bits = p->base64_buffer; if (bits & 0xffff) { char *msg; gpr_asprintf(&msg, "trailing bits in base64 encoding: 0x%04x", bits & 0xffff); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return parse_error(exec_ctx, p, cur, end, err); } @@ -1428,7 +1434,7 @@ static grpc_error *finish_str(grpc_exec_ctx *exec_ctx, char *msg; gpr_asprintf(&msg, "trailing bits in base64 encoding: 0x%02x", bits & 0xff); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return parse_error(exec_ctx, p, cur, end, err); } @@ -1550,7 +1556,8 @@ static grpc_error *is_binary_indexed_header(grpc_chttp2_hpack_parser *p, grpc_mdelem elem = grpc_chttp2_hptbl_lookup(&p->table, p->index); if (GRPC_MDISNULL(elem)) { return grpc_error_set_int( - grpc_error_set_int(GRPC_ERROR_CREATE("Invalid HPACK index received"), + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Invalid HPACK index received"), GRPC_ERROR_INT_INDEX, (intptr_t)p->index), GRPC_ERROR_INT_SIZE, (intptr_t)p->table.num_ents); } @@ -1620,13 +1627,18 @@ void grpc_chttp2_hpack_parser_destroy(grpc_exec_ctx *exec_ctx, grpc_error *grpc_chttp2_hpack_parser_parse(grpc_exec_ctx *exec_ctx, grpc_chttp2_hpack_parser *p, grpc_slice slice) { - /* TODO(ctiller): limit the distance of end from beg, and perform multiple - steps in the event of a large chunk of data to limit - stack space usage when no tail call optimization is - available */ +/* max number of bytes to parse at a time... limits call stack depth on + * compilers without TCO */ +#define MAX_PARSE_LENGTH 1024 p->current_slice_refcount = slice.refcount; - grpc_error *error = p->state(exec_ctx, p, GRPC_SLICE_START_PTR(slice), - GRPC_SLICE_END_PTR(slice)); + uint8_t *start = GRPC_SLICE_START_PTR(slice); + uint8_t *end = GRPC_SLICE_END_PTR(slice); + grpc_error *error = GRPC_ERROR_NONE; + while (start != end && error == GRPC_ERROR_NONE) { + uint8_t *target = start + GPR_MIN(MAX_PARSE_LENGTH, end - start); + error = p->state(exec_ctx, p, start, target); + start = target; + } p->current_slice_refcount = NULL; return error; } @@ -1670,7 +1682,7 @@ grpc_error *grpc_chttp2_header_parser_parse(grpc_exec_ctx *exec_ctx, if (is_last) { if (parser->is_boundary && parser->state != parse_begin) { GPR_TIMER_END("grpc_chttp2_hpack_parser_parse", 0); - return GRPC_ERROR_CREATE( + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( "end of header frame not aligned with a hpack record boundary"); } /* need to check for null stream: this can occur if we receive an invalid @@ -1678,7 +1690,8 @@ grpc_error *grpc_chttp2_header_parser_parse(grpc_exec_ctx *exec_ctx, if (s != NULL) { if (parser->is_boundary) { if (s->header_frames_received == GPR_ARRAY_SIZE(s->metadata_buffer)) { - return GRPC_ERROR_CREATE("Too many trailer frames"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Too many trailer frames"); } s->published_metadata[s->header_frames_received] = GRPC_METADATA_PUBLISHED_FROM_WIRE; diff --git a/src/core/ext/transport/chttp2/transport/hpack_table.c b/src/core/ext/transport/chttp2/transport/hpack_table.c index 62dd1b8cab..9dd41fdbe1 100644 --- a/src/core/ext/transport/chttp2/transport/hpack_table.c +++ b/src/core/ext/transport/chttp2/transport/hpack_table.c @@ -280,7 +280,7 @@ grpc_error *grpc_chttp2_hptbl_set_current_table_size(grpc_exec_ctx *exec_ctx, gpr_asprintf(&msg, "Attempt to make hpack table %d bytes when max is %d bytes", bytes, tbl->max_bytes); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -317,7 +317,7 @@ grpc_error *grpc_chttp2_hptbl_add(grpc_exec_ctx *exec_ctx, "HPACK max table size reduced to %d but not reflected by hpack " "stream (still at %d)", tbl->max_bytes, tbl->current_table_bytes); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } diff --git a/src/core/ext/transport/chttp2/transport/incoming_metadata.c b/src/core/ext/transport/chttp2/transport/incoming_metadata.c index c91b019aa0..da0a34d32a 100644 --- a/src/core/ext/transport/chttp2/transport/incoming_metadata.c +++ b/src/core/ext/transport/chttp2/transport/incoming_metadata.c @@ -41,69 +41,48 @@ #include <grpc/support/log.h> void grpc_chttp2_incoming_metadata_buffer_init( - grpc_chttp2_incoming_metadata_buffer *buffer) { - buffer->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); + grpc_chttp2_incoming_metadata_buffer *buffer, gpr_arena *arena) { + buffer->arena = arena; + grpc_metadata_batch_init(&buffer->batch); + buffer->batch.deadline = gpr_inf_future(GPR_CLOCK_REALTIME); } void grpc_chttp2_incoming_metadata_buffer_destroy( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer) { - size_t i; - if (!buffer->published) { - for (i = 0; i < buffer->count; i++) { - GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md); - } - } - gpr_free(buffer->elems); + grpc_metadata_batch_destroy(exec_ctx, &buffer->batch); } -void grpc_chttp2_incoming_metadata_buffer_add( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem elem) { - GPR_ASSERT(!buffer->published); - if (buffer->capacity == buffer->count) { - buffer->capacity = GPR_MAX(8, 2 * buffer->capacity); - buffer->elems = - gpr_realloc(buffer->elems, sizeof(*buffer->elems) * buffer->capacity); - } - buffer->elems[buffer->count++].md = elem; +grpc_error *grpc_chttp2_incoming_metadata_buffer_add( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, + grpc_mdelem elem) { buffer->size += GRPC_MDELEM_LENGTH(elem); + return grpc_metadata_batch_add_tail( + exec_ctx, &buffer->batch, + gpr_arena_alloc(buffer->arena, sizeof(grpc_linked_mdelem)), elem); } -void grpc_chttp2_incoming_metadata_buffer_replace_or_add( +grpc_error *grpc_chttp2_incoming_metadata_buffer_replace_or_add( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem elem) { - for (size_t i = 0; i < buffer->count; i++) { - if (grpc_slice_eq(GRPC_MDKEY(buffer->elems[i].md), GRPC_MDKEY(elem))) { - GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md); - buffer->elems[i].md = elem; - return; + for (grpc_linked_mdelem *l = buffer->batch.list.head; l != NULL; + l = l->next) { + if (grpc_slice_eq(GRPC_MDKEY(l->md), GRPC_MDKEY(elem))) { + GRPC_MDELEM_UNREF(exec_ctx, l->md); + l->md = elem; + return GRPC_ERROR_NONE; } } - grpc_chttp2_incoming_metadata_buffer_add(buffer, elem); + return grpc_chttp2_incoming_metadata_buffer_add(exec_ctx, buffer, elem); } void grpc_chttp2_incoming_metadata_buffer_set_deadline( grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline) { - GPR_ASSERT(!buffer->published); - buffer->deadline = deadline; + buffer->batch.deadline = deadline; } void grpc_chttp2_incoming_metadata_buffer_publish( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, grpc_metadata_batch *batch) { - GPR_ASSERT(!buffer->published); - buffer->published = 1; - if (buffer->count > 0) { - size_t i; - for (i = 0; i < buffer->count; i++) { - /* TODO(ctiller): do something better here */ - if (!GRPC_LOG_IF_ERROR("grpc_chttp2_incoming_metadata_buffer_publish", - grpc_metadata_batch_link_tail( - exec_ctx, batch, &buffer->elems[i]))) { - GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md); - } - } - } else { - batch->list.head = batch->list.tail = NULL; - } - batch->deadline = buffer->deadline; + *batch = buffer->batch; + grpc_metadata_batch_init(&buffer->batch); } diff --git a/src/core/ext/transport/chttp2/transport/incoming_metadata.h b/src/core/ext/transport/chttp2/transport/incoming_metadata.h index 1eac6fc150..288c917e65 100644 --- a/src/core/ext/transport/chttp2/transport/incoming_metadata.h +++ b/src/core/ext/transport/chttp2/transport/incoming_metadata.h @@ -37,28 +37,26 @@ #include "src/core/lib/transport/transport.h" typedef struct { - grpc_linked_mdelem *elems; - size_t count; - size_t capacity; - gpr_timespec deadline; - int published; + gpr_arena *arena; + grpc_metadata_batch batch; size_t size; // total size of metadata } grpc_chttp2_incoming_metadata_buffer; /** assumes everything initially zeroed */ void grpc_chttp2_incoming_metadata_buffer_init( - grpc_chttp2_incoming_metadata_buffer *buffer); + grpc_chttp2_incoming_metadata_buffer *buffer, gpr_arena *arena); void grpc_chttp2_incoming_metadata_buffer_destroy( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer); void grpc_chttp2_incoming_metadata_buffer_publish( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, grpc_metadata_batch *batch); -void grpc_chttp2_incoming_metadata_buffer_add( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem elem); -void grpc_chttp2_incoming_metadata_buffer_replace_or_add( +grpc_error *grpc_chttp2_incoming_metadata_buffer_add( grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, - grpc_mdelem elem); + grpc_mdelem elem) GRPC_MUST_USE_RESULT; +grpc_error *grpc_chttp2_incoming_metadata_buffer_replace_or_add( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer, + grpc_mdelem elem) GRPC_MUST_USE_RESULT; void grpc_chttp2_incoming_metadata_buffer_set_deadline( grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline); diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index c5b8416547..50993e4ecd 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -50,6 +50,7 @@ #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/iomgr/timer.h" #include "src/core/lib/transport/bdp_estimator.h" #include "src/core/lib/transport/connectivity_state.h" #include "src/core/lib/transport/pid_controller.h" @@ -101,6 +102,8 @@ typedef struct { typedef struct { gpr_timespec last_ping_sent_time; int pings_before_data_required; + grpc_timer delayed_ping_timer; + bool is_delayed_ping_timer_set; } grpc_chttp2_repeated_ping_state; /* deframer state for the overall http2 stream of bytes */ @@ -208,6 +211,12 @@ struct grpc_chttp2_incoming_byte_stream { grpc_closure finished_action; }; +typedef enum { + GRPC_CHTTP2_KEEPALIVE_STATE_WAITING, + GRPC_CHTTP2_KEEPALIVE_STATE_PINGING, + GRPC_CHTTP2_KEEPALIVE_STATE_DYING, +} grpc_chttp2_keepalive_state; + struct grpc_chttp2_transport { grpc_transport base; /* must be first */ gpr_refcount refs; @@ -301,6 +310,7 @@ struct grpc_chttp2_transport { grpc_chttp2_repeated_ping_policy ping_policy; grpc_chttp2_repeated_ping_state ping_state; uint64_t ping_ctr; /* unique id for pings */ + grpc_closure retry_initiate_ping_locked; /** ping acks */ size_t ping_ack_count; @@ -382,6 +392,28 @@ struct grpc_chttp2_transport { grpc_closure benign_reclaimer_locked; /** destructive cleanup closure */ grpc_closure destructive_reclaimer_locked; + + /* keep-alive ping support */ + /** Closure to initialize a keepalive ping */ + grpc_closure init_keepalive_ping_locked; + /** Closure to run when the keepalive ping is sent */ + grpc_closure start_keepalive_ping_locked; + /** Cousure to run when the keepalive ping ack is received */ + grpc_closure finish_keepalive_ping_locked; + /** Closrue to run when the keepalive ping timeouts */ + grpc_closure keepalive_watchdog_fired_locked; + /** timer to initiate ping events */ + grpc_timer keepalive_ping_timer; + /** watchdog to kill the transport when waiting for the keepalive ping */ + grpc_timer keepalive_watchdog_timer; + /** time duration in between pings */ + gpr_timespec keepalive_time; + /** grace period for a ping to complete before watchdog kicks in */ + gpr_timespec keepalive_timeout; + /** if keepalive pings are allowed when there's no outstanding streams */ + bool keepalive_permit_without_calls; + /** keep-alive state machine state */ + grpc_chttp2_keepalive_state keepalive_state; }; typedef enum { @@ -396,7 +428,7 @@ struct grpc_chttp2_stream { grpc_stream_refcount *refcount; grpc_closure destroy_stream; - void *destroy_stream_arg; + grpc_closure *destroy_stream_arg; grpc_chttp2_stream_link links[STREAM_LIST_COUNT]; uint8_t included[STREAM_LIST_COUNT]; @@ -797,4 +829,8 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx, uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t); +/** Set the default keepalive configurations, must only be called at + initialization */ +void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args); + #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */ diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 7ed00522c3..7e457ced27 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -116,7 +116,7 @@ grpc_error *grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx, GRPC_CHTTP2_CLIENT_CONNECT_STRING[t->deframe_state], (int)(uint8_t)GRPC_CHTTP2_CLIENT_CONNECT_STRING[t->deframe_state], *cur, (int)*cur, t->deframe_state); - err = GRPC_ERROR_CREATE(msg); + err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -219,7 +219,7 @@ grpc_error *grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx, t->incoming_frame_size, t->settings[GRPC_ACKED_SETTINGS] [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]); - err = GRPC_ERROR_CREATE(msg); + err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -278,7 +278,7 @@ static grpc_error *init_frame_parser(grpc_exec_ctx *exec_ctx, gpr_asprintf( &msg, "Expected SETTINGS frame as the first frame, got frame type %d", t->incoming_frame_type); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -288,7 +288,7 @@ static grpc_error *init_frame_parser(grpc_exec_ctx *exec_ctx, char *msg; gpr_asprintf(&msg, "Expected CONTINUATION frame, got frame type %02x", t->incoming_frame_type); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -299,7 +299,7 @@ static grpc_error *init_frame_parser(grpc_exec_ctx *exec_ctx, "Expected CONTINUATION frame for grpc_chttp2_stream %08x, got " "grpc_chttp2_stream %08x", t->expect_continuation_stream_id, t->incoming_stream_id); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -311,7 +311,8 @@ static grpc_error *init_frame_parser(grpc_exec_ctx *exec_ctx, case GRPC_CHTTP2_FRAME_HEADER: return init_header_frame_parser(exec_ctx, t, 0); case GRPC_CHTTP2_FRAME_CONTINUATION: - return GRPC_ERROR_CREATE("Unexpected CONTINUATION frame"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Unexpected CONTINUATION frame"); case GRPC_CHTTP2_FRAME_RST_STREAM: return init_rst_stream_parser(exec_ctx, t); case GRPC_CHTTP2_FRAME_SETTINGS: @@ -371,7 +372,7 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx, char *msg; gpr_asprintf(&msg, "frame of size %d overflows incoming window of %" PRId64, t->incoming_frame_size, t->incoming_window); - grpc_error *err = GRPC_ERROR_CREATE(msg); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } @@ -381,16 +382,38 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx, s->incoming_window_delta + t->settings[GRPC_ACKED_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]) { - char *msg; - gpr_asprintf(&msg, - "frame of size %d overflows incoming window of %" PRId64, - t->incoming_frame_size, - s->incoming_window_delta + - t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); - grpc_error *err = GRPC_ERROR_CREATE(msg); - gpr_free(msg); - return err; + if (incoming_frame_size <= + s->incoming_window_delta + + t->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]) { + gpr_log( + GPR_ERROR, + "Incoming frame of size %d exceeds incoming window size of %" PRId64 + ".\n" + "The (un-acked, future) window size would be %" PRId64 + " which is not exceeded.\n" + "This would usually cause a disconnection, but allowing it due to " + "broken HTTP2 implementations in the wild.\n" + "See (for example) https://github.com/netty/netty/issues/6520.", + t->incoming_frame_size, + s->incoming_window_delta + + t->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], + s->incoming_window_delta + + t->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); + } else { + char *msg; + gpr_asprintf(&msg, + "frame of size %d overflows incoming window of %" PRId64, + t->incoming_frame_size, + s->incoming_window_delta + + t->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); + grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + gpr_free(msg); + return err; + } } GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA("parse", t, s, @@ -520,13 +543,21 @@ static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp, grpc_chttp2_cancel_stream( exec_ctx, t, s, grpc_error_set_int( - GRPC_ERROR_CREATE("received initial metadata size exceeds limit"), + GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "received initial metadata size exceeds limit"), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED)); grpc_chttp2_parsing_become_skip_parser(exec_ctx, t); s->seen_error = true; GRPC_MDELEM_UNREF(exec_ctx, md); } else { - grpc_chttp2_incoming_metadata_buffer_add(&s->metadata_buffer[0], md); + grpc_error *error = grpc_chttp2_incoming_metadata_buffer_add( + exec_ctx, &s->metadata_buffer[0], md); + if (error != GRPC_ERROR_NONE) { + grpc_chttp2_cancel_stream(exec_ctx, t, s, error); + grpc_chttp2_parsing_become_skip_parser(exec_ctx, t); + s->seen_error = true; + GRPC_MDELEM_UNREF(exec_ctx, md); + } } } @@ -569,14 +600,22 @@ static void on_trailing_header(grpc_exec_ctx *exec_ctx, void *tp, new_size, metadata_size_limit); grpc_chttp2_cancel_stream( exec_ctx, t, s, - grpc_error_set_int( - GRPC_ERROR_CREATE("received trailing metadata size exceeds limit"), - GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED)); + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "received trailing metadata size exceeds limit"), + GRPC_ERROR_INT_GRPC_STATUS, + GRPC_STATUS_RESOURCE_EXHAUSTED)); grpc_chttp2_parsing_become_skip_parser(exec_ctx, t); s->seen_error = true; GRPC_MDELEM_UNREF(exec_ctx, md); } else { - grpc_chttp2_incoming_metadata_buffer_add(&s->metadata_buffer[1], md); + grpc_error *error = grpc_chttp2_incoming_metadata_buffer_add( + exec_ctx, &s->metadata_buffer[1], md); + if (error != GRPC_ERROR_NONE) { + grpc_chttp2_cancel_stream(exec_ctx, t, s, error); + grpc_chttp2_parsing_become_skip_parser(exec_ctx, t); + s->seen_error = true; + GRPC_MDELEM_UNREF(exec_ctx, md); + } } GPR_TIMER_END("on_trailing_header", 0); @@ -735,7 +774,8 @@ static grpc_error *init_goaway_parser(grpc_exec_ctx *exec_ctx, static grpc_error *init_settings_frame_parser(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { if (t->incoming_stream_id != 0) { - return GRPC_ERROR_CREATE("Settings frame received for grpc_chttp2_stream"); + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Settings frame received for grpc_chttp2_stream"); } grpc_error *err = grpc_chttp2_settings_parser_begin_frame( diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index 2b9d93cae7..0869056f56 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -101,6 +101,14 @@ static void maybe_initiate_ping(grpc_exec_ctx *exec_ctx, "Ping delayed [%p]: not enough time elapsed since last ping", t->peer_string); } + if (!t->ping_state.is_delayed_ping_timer_set) { + t->ping_state.is_delayed_ping_timer_set = true; + grpc_timer_init(exec_ctx, &t->ping_state.delayed_ping_timer, + gpr_time_add(t->ping_state.last_ping_sent_time, + t->ping_policy.min_time_between_pings), + &t->retry_initiate_ping_locked, + gpr_now(GPR_CLOCK_MONOTONIC)); + } return; } /* coalesce equivalent pings into this one */ |