Add notify_on_receive_settings closure to chttp2 transport.

8 files changed, 54 insertions, 14 deletions

diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.cc b/src/core/ext/transport/chttp2/client/chttp2_connector.cc
index 4efd129384..5870a3e6d5 100644
--- a/src/core/ext/transport/chttp2/client/chttp2_connector.cc
+++ b/src/core/ext/transport/chttp2/client/chttp2_connector.cc
@@ -120,8 +120,32 @@ static void on_handshake_done(grpc_exec_ctx* exec_ctx, void* arg,
     c->result->transport = grpc_create_chttp2_transport(
         exec_ctx, args->args, args->endpoint, true);
     GPR_ASSERT(c->result->transport);
+    // TODO(roth): We ideally want to wait until we receive HTTP/2
+    // settings from the server before we consider the connection
+    // established.  If that doesn't happen before the connection
+    // timeout expires, then we should consider the connection attempt a
+    // failure and feed that information back into the backoff code.
+    // We could pass a notify_on_receive_settings callback to
+    // grpc_chttp2_transport_start_reading() to let us know when
+    // settings are received, but we would need to figure out how to use
+    // that information here.
+    //
+    // Unfortunately, we don't currently have a way to split apart the two
+    // effects of scheduling c->notify: we start sending RPCs immediately
+    // (which we want to do) and we consider the connection attempt successful
+    // (which we don't want to do until we get the notify_on_receive_settings
+    // callback from the transport).  If we could split those things
+    // apart, then we could start sending RPCs but then wait for our
+    // timeout before deciding if the connection attempt is successful.
+    // If the attempt is not successful, then we would tear down the
+    // transport and feed the failure back into the backoff code.
+    //
+    // In addition, even if we did that, we would probably not want to do
+    // so until after transparent retries is implemented.  Otherwise, any
+    // RPC that we attempt to send on the connection before the timeout
+    // would fail instead of being retried on a subsequent attempt.
     grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport,
-                                        args->read_buffer);
+                                        args->read_buffer, nullptr);
     c->result->channel_args = args->args;
   }
   grpc_closure* notify = c->notify;
diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
index fcc2f4249a..ad64f740b8 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
@@ -58,7 +58,7 @@ grpc_channel* grpc_insecure_channel_create_from_fd(
   grpc_channel* channel = grpc_channel_create(
       &exec_ctx, target, final_args, GRPC_CLIENT_DIRECT_CHANNEL, transport);
   grpc_channel_args_destroy(&exec_ctx, final_args);
-  grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+  grpc_chttp2_transport_start_reading(&exec_ctx, transport, nullptr, nullptr);
 
   grpc_exec_ctx_finish(&exec_ctx);
 
diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.cc b/src/core/ext/transport/chttp2/server/chttp2_server.cc
index 1b4d89b5ee..39e8dfd684 100644
--- a/src/core/ext/transport/chttp2/server/chttp2_server.cc
+++ b/src/core/ext/transport/chttp2/server/chttp2_server.cc
@@ -93,8 +93,10 @@ static void on_handshake_done(grpc_exec_ctx* exec_ctx, void* arg,
       grpc_server_setup_transport(
           exec_ctx, connection_state->svr_state->server, transport,
           connection_state->accepting_pollset, args->args);
+// FIXME: set notify_on_receive_settings callback and use it to enforce
+// handshaking deadline
       grpc_chttp2_transport_start_reading(exec_ctx, transport,
-                                          args->read_buffer);
+                                          args->read_buffer, nullptr);
       grpc_channel_args_destroy(exec_ctx, args->args);
     }
   }
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 70d4864710..09ee14c022 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
@@ -61,7 +61,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server,
   }
 
   grpc_server_setup_transport(&exec_ctx, server, transport, NULL, server_args);
-  grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+  grpc_chttp2_transport_start_reading(&exec_ctx, transport, nullptr, nullptr);
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
index b4edb17cde..21808115d7 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
@@ -1788,7 +1788,6 @@ static void perform_transport_op_locked(grpc_exec_ctx* exec_ctx,
   grpc_transport_op* op = (grpc_transport_op*)stream_op;
   grpc_chttp2_transport* t =
       (grpc_chttp2_transport*)op->handler_private.extra_arg;
-  grpc_error* close_transport = op->disconnect_with_error;
 
   if (op->goaway_error) {
     send_goaway(exec_ctx, t, op->goaway_error);
@@ -1820,8 +1819,13 @@ static void perform_transport_op_locked(grpc_exec_ctx* exec_ctx,
         op->on_connectivity_state_change);
   }
 
-  if (close_transport != GRPC_ERROR_NONE) {
-    close_transport_locked(exec_ctx, t, close_transport);
+  if (op->disconnect_with_error != GRPC_ERROR_NONE) {
+    close_transport_locked(exec_ctx, t, op->disconnect_with_error);
+    if (t->notify_on_receive_settings != nullptr) {
+      GRPC_CLOSURE_SCHED(exec_ctx, t->notify_on_receive_settings,
+                         GRPC_ERROR_CANCELLED);
+      t->notify_on_receive_settings = nullptr;
+    }
   }
 
   GRPC_CLOSURE_RUN(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
@@ -3231,15 +3235,16 @@ grpc_transport* grpc_create_chttp2_transport(
   return &t->base;
 }
 
-void grpc_chttp2_transport_start_reading(grpc_exec_ctx* exec_ctx,
-                                         grpc_transport* transport,
-                                         grpc_slice_buffer* read_buffer) {
+void grpc_chttp2_transport_start_reading(
+    grpc_exec_ctx* exec_ctx, grpc_transport* transport,
+    grpc_slice_buffer* read_buffer, grpc_closure* notify_on_receive_settings) {
   grpc_chttp2_transport* t = (grpc_chttp2_transport*)transport;
   GRPC_CHTTP2_REF_TRANSPORT(
       t, "reading_action"); /* matches unref inside reading_action */
-  if (read_buffer != NULL) {
+  if (read_buffer != nullptr) {
     grpc_slice_buffer_move_into(read_buffer, &t->read_buffer);
     gpr_free(read_buffer);
   }
+  t->notify_on_receive_settings = notify_on_receive_settings;
   GRPC_CLOSURE_SCHED(exec_ctx, &t->read_action_locked, GRPC_ERROR_NONE);
 }
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.h b/src/core/ext/transport/chttp2/transport/chttp2_transport.h
index 4fe12d42e9..a349e00498 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.h
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.h
@@ -41,9 +41,11 @@ grpc_transport* grpc_create_chttp2_transport(
 
 /// Takes ownership of \a read_buffer, which (if non-NULL) contains
 /// leftover bytes previously read from the endpoint (e.g., by handshakers).
-void grpc_chttp2_transport_start_reading(grpc_exec_ctx* exec_ctx,
-                                         grpc_transport* transport,
-                                         grpc_slice_buffer* read_buffer);
+/// If non-null, \a notify_on_receive_settings will be scheduled when
+/// HTTP/2 settings are received from the peer.
+void grpc_chttp2_transport_start_reading(
+    grpc_exec_ctx* exec_ctx, grpc_transport* transport,
+    grpc_slice_buffer* read_buffer, grpc_closure* notify_on_receive_settings);
 
 #ifdef __cplusplus
 }
diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.cc b/src/core/ext/transport/chttp2/transport/frame_settings.cc
index d33da721a5..a1da607516 100644
--- a/src/core/ext/transport/chttp2/transport/frame_settings.cc
+++ b/src/core/ext/transport/chttp2/transport/frame_settings.cc
@@ -131,6 +131,11 @@ grpc_error* grpc_chttp2_settings_parser_parse(grpc_exec_ctx* exec_ctx, void* p,
             memcpy(parser->target_settings, parser->incoming_settings,
                    GRPC_CHTTP2_NUM_SETTINGS * sizeof(uint32_t));
             grpc_slice_buffer_add(&t->qbuf, grpc_chttp2_settings_ack_create());
+            if (t->notify_on_receive_settings != nullptr) {
+              GRPC_CLOSURE_SCHED(exec_ctx, t->notify_on_receive_settings,
+                                 GRPC_ERROR_NONE);
+              t->notify_on_receive_settings = nullptr;
+            }
           }
           return GRPC_ERROR_NONE;
         }
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index a5a0a804a2..b4fe6fdcbe 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -245,6 +245,8 @@ struct grpc_chttp2_transport {
 
   grpc_combiner* combiner;
 
+  grpc_closure* notify_on_receive_settings;
+
   /** write execution state of the transport */
   grpc_chttp2_write_state write_state;
   /** is this the first write in a series of writes?