From 1bc34473ce470517d51bf162db1f99cd939be7fe Mon Sep 17 00:00:00 2001
From: Yuchen Zeng <zyc@google.com>
Date: Sun, 26 Mar 2017 23:15:38 -0700
Subject: [PATCH] Add max age enforcement for server channels
---
include/grpc/impl/codegen/grpc_types.h | 4 ++
src/core/lib/surface/server.c | 58 ++++++++++++++++++++++++++
2 files changed, 62 insertions(+)
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index 887c176f1a..6b380bf87e 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -163,6 +163,10 @@ typedef struct {
/** Maximum message length that the channel can send. Int valued, bytes.
-1 means unlimited. */
#define GRPC_ARG_MAX_SEND_MESSAGE_LENGTH "grpc.max_send_message_length"
+
+#define GPRC_ARG_MAX_CONNECION_AGE_S "grpc.max_connection_age"
+
+#define GPRC_ARG_MAX_CONNECION_AGE_GRACE_S "grpc.max_connection_age_grace"
/** Initial sequence number for http2 transports. Int valued. */
#define GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER \
"grpc.http2.initial_sequence_number"
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index a123c9ca43..dee722e67e 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -45,6 +45,7 @@
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/connected_channel.h"
#include "src/core/lib/iomgr/iomgr.h"
+#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/support/stack_lockfree.h"
#include "src/core/lib/support/string.h"
@@ -53,9 +54,12 @@
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/completion_queue.h"
#include "src/core/lib/surface/init.h"
+#include "src/core/lib/transport/http2_errors.h"
#include "src/core/lib/transport/metadata.h"
#include "src/core/lib/transport/static_metadata.h"
+#define DEFAULT_MAX_CONNECTION_AGE_S INT_MAX
+
typedef struct listener {
void *arg;
void (*start)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
@@ -116,6 +120,9 @@ struct channel_data {
uint32_t registered_method_max_probes;
grpc_closure finish_destroy_channel_closure;
grpc_closure channel_connectivity_changed;
+ grpc_timer max_age_timer;
+ gpr_timespec max_connection_age;
+ grpc_closure close_max_age_channel;
};
typedef struct shutdown_tag {
@@ -381,6 +388,7 @@ static void request_matcher_kill_requests(grpc_exec_ctx *exec_ctx,
static void server_ref(grpc_server *server) {
gpr_ref(&server->internal_refcount);
+ gpr_log(GPR_DEBUG, "server_ref");
}
static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) {
@@ -442,9 +450,11 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
grpc_error *error) {
+ gpr_log(GPR_DEBUG, "destroy_channel");
if (is_channel_orphaned(chand)) return;
GPR_ASSERT(chand->server != NULL);
orphan_channel(chand);
+ grpc_timer_cancel(exec_ctx, &chand->max_age_timer);
server_ref(chand->server);
maybe_finish_shutdown(exec_ctx, chand->server);
grpc_closure_init(&chand->finish_destroy_channel_closure,
@@ -831,6 +841,7 @@ static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd,
grpc_transport *transport,
const void *transport_server_data) {
+ gpr_log(GPR_DEBUG, "accept_stream");
channel_data *chand = cd;
/* create a call */
grpc_call_create_args args;
@@ -882,6 +893,7 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
const grpc_call_element_args *args) {
+ gpr_log(GPR_DEBUG, "init_call_elem");
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
memset(calld, 0, sizeof(call_data));
@@ -903,6 +915,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
+ gpr_log(GPR_DEBUG, "destroy_call_elem");
GPR_ASSERT(calld->state != PENDING);
if (calld->host_set) {
@@ -918,6 +931,23 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
server_unref(exec_ctx, chand->server);
}
+static void close_max_age_channel(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ channel_data *chand = arg;
+ if (error == GRPC_ERROR_NONE) {
+ grpc_transport_op *op = grpc_make_transport_op(NULL);
+ op->goaway_error =
+ grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("max_age"),
+ GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_NO_ERROR);
+ grpc_channel_element *elem = grpc_channel_stack_element(
+ grpc_channel_get_channel_stack(chand->channel), 0);
+ elem->filter->start_transport_op(exec_ctx, elem, op);
+ } else if (error != GRPC_ERROR_CANCELLED) {
+ GRPC_LOG_IF_ERROR("close_max_age_channel", error);
+ }
+ GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "max age");
+}
+
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
@@ -929,6 +959,28 @@ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
chand->next = chand->prev = chand;
chand->registered_methods = NULL;
chand->connectivity_state = GRPC_CHANNEL_IDLE;
+ chand->max_connection_age =
+ DEFAULT_MAX_CONNECTION_AGE_S == INT_MAX
+ ? gpr_inf_future(GPR_TIMESPAN)
+ : gpr_time_from_seconds(DEFAULT_MAX_CONNECTION_AGE_S, GPR_TIMESPAN);
+ grpc_closure_init(&chand->close_max_age_channel, close_max_age_channel, chand,
+ grpc_schedule_on_exec_ctx);
+ const grpc_channel_args *channel_args = args->channel_args;
+ if (channel_args) {
+ size_t i;
+ for (i = 0; i < channel_args->num_args; i++) {
+ if (0 ==
+ strcmp(channel_args->args[i].key, GPRC_ARG_MAX_CONNECION_AGE_S)) {
+ const int value = grpc_channel_arg_get_integer(
+ &channel_args->args[i],
+ (grpc_integer_options){DEFAULT_MAX_CONNECTION_AGE_S, 1, INT_MAX});
+ chand->max_connection_age =
+ value == INT_MAX ? gpr_inf_future(GPR_TIMESPAN)
+ : gpr_time_from_seconds(value, GPR_TIMESPAN);
+ }
+ }
+ }
+
grpc_closure_init(&chand->channel_connectivity_changed,
channel_connectivity_changed, chand,
grpc_schedule_on_exec_ctx);
@@ -1132,6 +1184,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
grpc_transport *transport,
grpc_pollset *accepting_pollset,
const grpc_channel_args *args) {
+ gpr_log(GPR_DEBUG, "grpc_server_setup_transport");
size_t num_registered_methods;
size_t alloc;
registered_method *rm;
@@ -1152,6 +1205,11 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
chand->server = s;
server_ref(s);
chand->channel = channel;
+ GRPC_CHANNEL_INTERNAL_REF(channel, "max age");
+ grpc_timer_init(
+ exec_ctx, &chand->max_age_timer,
+ gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), chand->max_connection_age),
+ &chand->close_max_age_channel, gpr_now(GPR_CLOCK_MONOTONIC));
size_t cq_idx;
grpc_completion_queue *accepting_cq = grpc_cq_from_pollset(accepting_pollset);
--
GitLab