Merge master into execctx

2 files changed, 74 insertions, 15 deletions

diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.cc b/src/core/ext/transport/chttp2/server/chttp2_server.cc
index 08d919144e..5669fa4090 100644
--- a/src/core/ext/transport/chttp2/server/chttp2_server.cc
+++ b/src/core/ext/transport/chttp2/server/chttp2_server.cc
@@ -21,6 +21,7 @@
 #include <grpc/grpc.h>
 
 #include <inttypes.h>
+#include <limits.h>
 #include <string.h>
 
 #include <grpc/support/alloc.h>
@@ -31,6 +32,7 @@
 
 #include "src/core/ext/filters/http/server/http_server_filter.h"
 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/ext/transport/chttp2/transport/internal.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/handshaker.h"
 #include "src/core/lib/channel/handshaker_registry.h"
@@ -53,12 +55,51 @@ typedef struct {
 } server_state;
 
 typedef struct {
+  gpr_refcount refs;
   server_state* svr_state;
   grpc_pollset* accepting_pollset;
   grpc_tcp_server_acceptor* acceptor;
   grpc_handshake_manager* handshake_mgr;
+  // State for enforcing handshake timeout on receiving HTTP/2 settings.
+  grpc_chttp2_transport* transport;
+  grpc_millis deadline;
+  grpc_timer timer;
+  grpc_closure on_timeout;
+  grpc_closure on_receive_settings;
 } server_connection_state;
 
+static void server_connection_state_unref(
+    server_connection_state* connection_state) {
+  if (gpr_unref(&connection_state->refs)) {
+    if (connection_state->transport != nullptr) {
+      GRPC_CHTTP2_UNREF_TRANSPORT(connection_state->transport,
+                                  "receive settings timeout");
+    }
+    gpr_free(connection_state);
+  }
+}
+
+static void on_timeout(void* arg, grpc_error* error) {
+  server_connection_state* connection_state = (server_connection_state*)arg;
+  // Note that we may be called with GRPC_ERROR_NONE when the timer fires
+  // or with an error indicating that the timer system is being shut down.
+  if (error != GRPC_ERROR_CANCELLED) {
+    grpc_transport_op* op = grpc_make_transport_op(nullptr);
+    op->disconnect_with_error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+        "Did not receive HTTP/2 settings before handshake timeout");
+    grpc_transport_perform_op(&connection_state->transport->base, op);
+  }
+  server_connection_state_unref(connection_state);
+}
+
+static void on_receive_settings(void* arg, grpc_error* error) {
+  server_connection_state* connection_state = (server_connection_state*)arg;
+  if (error == GRPC_ERROR_NONE) {
+    grpc_timer_cancel(&connection_state->timer);
+  }
+  server_connection_state_unref(connection_state);
+}
+
 static void on_handshake_done(void* arg, grpc_error* error) {
   grpc_handshaker_args* args = (grpc_handshaker_args*)arg;
   server_connection_state* connection_state =
@@ -67,7 +108,6 @@ static void on_handshake_done(void* arg, grpc_error* error) {
   if (error != GRPC_ERROR_NONE || connection_state->svr_state->shutdown) {
     const char* error_str = grpc_error_string(error);
     gpr_log(GPR_DEBUG, "Handshaking failed: %s", error_str);
-
     if (error == GRPC_ERROR_NONE && args->endpoint != nullptr) {
       // We were shut down after handshaking completed successfully, so
       // destroy the endpoint here.
@@ -87,12 +127,27 @@ static void on_handshake_done(void* arg, grpc_error* error) {
     // code, so we can just clean up here without creating a transport.
     if (args->endpoint != nullptr) {
       grpc_transport* transport =
-          grpc_create_chttp2_transport(args->args, args->endpoint, 0);
+          grpc_create_chttp2_transport(args->args, args->endpoint, false);
       grpc_server_setup_transport(
           connection_state->svr_state->server, transport,
           connection_state->accepting_pollset, args->args);
-      grpc_chttp2_transport_start_reading(transport, args->read_buffer);
+      // Use notify_on_receive_settings callback to enforce the
+      // handshake deadline.
+      connection_state->transport = (grpc_chttp2_transport*)transport;
+      gpr_ref(&connection_state->refs);
+      GRPC_CLOSURE_INIT(&connection_state->on_receive_settings,
+                        on_receive_settings, connection_state,
+                        grpc_schedule_on_exec_ctx);
+      grpc_chttp2_transport_start_reading(
+          transport, args->read_buffer, &connection_state->on_receive_settings);
       grpc_channel_args_destroy(args->args);
+      gpr_ref(&connection_state->refs);
+      GRPC_CHTTP2_REF_TRANSPORT((grpc_chttp2_transport*)transport,
+                                "receive settings timeout");
+      GRPC_CLOSURE_INIT(&connection_state->on_timeout, on_timeout,
+                        connection_state, grpc_schedule_on_exec_ctx);
+      grpc_timer_init(&connection_state->timer, connection_state->deadline,
+                      &connection_state->on_timeout);
     }
   }
   grpc_handshake_manager_pending_list_remove(
@@ -100,9 +155,9 @@ static void on_handshake_done(void* arg, grpc_error* error) {
       connection_state->handshake_mgr);
   gpr_mu_unlock(&connection_state->svr_state->mu);
   grpc_handshake_manager_destroy(connection_state->handshake_mgr);
-  grpc_tcp_server_unref(connection_state->svr_state->tcp_server);
   gpr_free(connection_state->acceptor);
-  gpr_free(connection_state);
+  grpc_tcp_server_unref(connection_state->svr_state->tcp_server);
+  server_connection_state_unref(connection_state);
 }
 
 static void on_accept(void* arg, grpc_endpoint* tcp,
@@ -123,20 +178,24 @@ static void on_accept(void* arg, grpc_endpoint* tcp,
   gpr_mu_unlock(&state->mu);
   grpc_tcp_server_ref(state->tcp_server);
   server_connection_state* connection_state =
-      (server_connection_state*)gpr_malloc(sizeof(*connection_state));
+      (server_connection_state*)gpr_zalloc(sizeof(*connection_state));
+  gpr_ref_init(&connection_state->refs, 1);
   connection_state->svr_state = state;
   connection_state->accepting_pollset = accepting_pollset;
   connection_state->acceptor = acceptor;
   connection_state->handshake_mgr = handshake_mgr;
   grpc_handshakers_add(HANDSHAKER_SERVER, state->args,
                        connection_state->handshake_mgr);
-  // TODO(roth): We should really get this timeout value from channel
-  // args instead of hard-coding it.
-  const grpc_millis deadline =
-      grpc_core::ExecCtx::Get()->Now() + 120 * GPR_MS_PER_SEC;
-  grpc_handshake_manager_do_handshake(connection_state->handshake_mgr, tcp,
-                                      state->args, deadline, acceptor,
-                                      on_handshake_done, connection_state);
+  const grpc_arg* timeout_arg =
+      grpc_channel_args_find(state->args, GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS);
+  connection_state->deadline =
+      grpc_core::ExecCtx::Get()->Now() +
+      grpc_channel_arg_get_integer(timeout_arg,
+                                   {120 * GPR_MS_PER_SEC, 1, INT_MAX});
+  grpc_handshake_manager_do_handshake(
+      connection_state->handshake_mgr, nullptr /* interested_parties */, tcp,
+      state->args, connection_state->deadline, acceptor, on_handshake_done,
+      connection_state);
 }
 
 /* Server callback: start listening on our ports */
diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc
index 210939e6af..dafd4af6ce 100644
--- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc
+++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc
@@ -49,7 +49,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server,
 
   const grpc_channel_args* server_args = grpc_server_get_channel_args(server);
   grpc_transport* transport = grpc_create_chttp2_transport(
-      server_args, server_endpoint, 0 /* is_client */);
+      server_args, server_endpoint, false /* is_client */);
 
   grpc_pollset** pollsets;
   size_t num_pollsets = 0;
@@ -60,7 +60,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server,
   }
 
   grpc_server_setup_transport(server, transport, nullptr, server_args);
-  grpc_chttp2_transport_start_reading(transport, nullptr);
+  grpc_chttp2_transport_start_reading(transport, nullptr, nullptr);
 }
 
 #else  // !GPR_SUPPORT_CHANNELS_FROM_FD