Merge branch 'buffer_pools_for_realsies' into grpc_slice

6 files changed, 98 insertions, 134 deletions

diff --git a/src/core/ext/transport/chttp2/alpn/alpn.c b/src/core/ext/transport/chttp2/alpn/alpn.c
index 48b0217265..55710dc5ae 100644
--- a/src/core/ext/transport/chttp2/alpn/alpn.c
+++ b/src/core/ext/transport/chttp2/alpn/alpn.c
@@ -36,7 +36,7 @@
 #include <grpc/support/useful.h>
 
 /* in order of preference */
-static const char *const supported_versions[] = {"h2"};
+static const char *const supported_versions[] = {"grpc-exp", "h2"};
 
 int grpc_chttp2_is_alpn_version_supported(const char *version, size_t size) {
   size_t i;
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 0c4dbee8d9..03dc955b27 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create.c
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create.c
@@ -52,6 +52,10 @@
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/channel.h"
 
+//
+// connector
+//
+
 typedef struct {
   grpc_connector base;
   gpr_refcount refs;
@@ -151,41 +155,26 @@ static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con,
   grpc_closure_init(&c->connected, connected, c);
   grpc_tcp_client_connect(exec_ctx, &c->connected, &c->tcp,
                           args->interested_parties, args->channel_args,
-                          args->addr, args->addr_len, args->deadline);
+                          args->addr, args->deadline);
 }
 
 static const grpc_connector_vtable connector_vtable = {
     connector_ref, connector_unref, connector_shutdown, connector_connect};
 
-typedef struct {
-  grpc_client_channel_factory base;
-  gpr_refcount refs;
-  grpc_channel_args *merge_args;
-} client_channel_factory;
+//
+// client_channel_factory
+//
 
 static void client_channel_factory_ref(
-    grpc_client_channel_factory *cc_factory) {
-  client_channel_factory *f = (client_channel_factory *)cc_factory;
-  gpr_ref(&f->refs);
-}
+    grpc_client_channel_factory *cc_factory) {}
 
 static void client_channel_factory_unref(
-    grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory) {
-  client_channel_factory *f = (client_channel_factory *)cc_factory;
-  if (gpr_unref(&f->refs)) {
-    grpc_channel_args_destroy(f->merge_args);
-    gpr_free(f);
-  }
-}
+    grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory) {}
 
 static grpc_subchannel *client_channel_factory_create_subchannel(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
-    grpc_subchannel_args *args) {
-  client_channel_factory *f = (client_channel_factory *)cc_factory;
+    const grpc_subchannel_args *args) {
   connector *c = gpr_malloc(sizeof(*c));
-  grpc_channel_args *final_args =
-      grpc_channel_args_merge(args->args, f->merge_args);
-  grpc_subchannel *s;
   memset(c, 0, sizeof(*c));
   c->base.vtable = &connector_vtable;
   gpr_ref_init(&c->refs, 1);
@@ -197,23 +186,18 @@ static grpc_subchannel *client_channel_factory_create_subchannel(
         grpc_http_connect_handshaker_create(proxy_name, args->server_name));
     gpr_free(proxy_name);
   }
-  args->args = final_args;
-  s = grpc_subchannel_create(exec_ctx, &c->base, args);
+  grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args);
   grpc_connector_unref(exec_ctx, &c->base);
-  grpc_channel_args_destroy(final_args);
   return s;
 }
 
 static grpc_channel *client_channel_factory_create_channel(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
     const char *target, grpc_client_channel_type type,
-    grpc_channel_args *args) {
-  client_channel_factory *f = (client_channel_factory *)cc_factory;
-  grpc_channel_args *final_args = grpc_channel_args_merge(args, f->merge_args);
-  grpc_channel *channel = grpc_channel_create(exec_ctx, target, final_args,
-                                              GRPC_CLIENT_CHANNEL, NULL);
-  grpc_channel_args_destroy(final_args);
-  grpc_resolver *resolver = grpc_resolver_create(target);
+    const grpc_channel_args *args) {
+  grpc_channel *channel =
+      grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL);
+  grpc_resolver *resolver = grpc_resolver_create(target, args);
   if (!resolver) {
     GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel,
                                 "client_channel_factory_create_channel");
@@ -221,7 +205,7 @@ static grpc_channel *client_channel_factory_create_channel(
   }
 
   grpc_client_channel_finish_initialization(
-      exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base);
+      exec_ctx, grpc_channel_get_channel_stack(channel), resolver, cc_factory);
   GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel");
 
   return channel;
@@ -232,6 +216,9 @@ static const grpc_client_channel_factory_vtable client_channel_factory_vtable =
      client_channel_factory_create_subchannel,
      client_channel_factory_create_channel};
 
+static grpc_client_channel_factory client_channel_factory = {
+    &client_channel_factory_vtable};
+
 /* Create a client channel:
    Asynchronously: - resolve target
                    - connect to it (trying alternatives as presented)
@@ -245,16 +232,12 @@ grpc_channel *grpc_insecure_channel_create(const char *target,
       (target, args, reserved));
   GPR_ASSERT(!reserved);
 
-  client_channel_factory *f = gpr_malloc(sizeof(*f));
-  memset(f, 0, sizeof(*f));
-  f->base.vtable = &client_channel_factory_vtable;
-  gpr_ref_init(&f->refs, 1);
-  f->merge_args = grpc_channel_args_copy(args);
-
+  grpc_client_channel_factory *factory =
+      (grpc_client_channel_factory *)&client_channel_factory;
   grpc_channel *channel = client_channel_factory_create_channel(
-      &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, NULL);
+      &exec_ctx, factory, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, args);
 
-  grpc_client_channel_factory_unref(&exec_ctx, &f->base);
+  grpc_client_channel_factory_unref(&exec_ctx, factory);
   grpc_exec_ctx_finish(&exec_ctx);
 
   return channel != NULL ? channel : grpc_lame_client_channel_create(
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 c135f69820..7f9c365799 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
@@ -54,6 +54,10 @@
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/tsi/transport_security_interface.h"
 
+//
+// connector
+//
+
 typedef struct {
   grpc_connector base;
   gpr_refcount refs;
@@ -208,19 +212,21 @@ static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con,
   GPR_ASSERT(c->connecting_endpoint == NULL);
   gpr_mu_unlock(&c->mu);
   grpc_closure_init(&c->connected_closure, connected, c);
-  grpc_tcp_client_connect(exec_ctx, &c->connected_closure,
-                          &c->newly_connecting_endpoint,
-                          args->interested_parties, args->channel_args,
-                          args->addr, args->addr_len, args->deadline);
+  grpc_tcp_client_connect(
+      exec_ctx, &c->connected_closure, &c->newly_connecting_endpoint,
+      args->interested_parties, args->channel_args, args->addr, args->deadline);
 }
 
 static const grpc_connector_vtable connector_vtable = {
     connector_ref, connector_unref, connector_shutdown, connector_connect};
 
+//
+// client_channel_factory
+//
+
 typedef struct {
   grpc_client_channel_factory base;
   gpr_refcount refs;
-  grpc_channel_args *merge_args;
   grpc_channel_security_connector *security_connector;
 } client_channel_factory;
 
@@ -236,19 +242,15 @@ static void client_channel_factory_unref(
   if (gpr_unref(&f->refs)) {
     GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base,
                                   "client_channel_factory");
-    grpc_channel_args_destroy(f->merge_args);
     gpr_free(f);
   }
 }
 
 static grpc_subchannel *client_channel_factory_create_subchannel(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
-    grpc_subchannel_args *args) {
+    const grpc_subchannel_args *args) {
   client_channel_factory *f = (client_channel_factory *)cc_factory;
   connector *c = gpr_malloc(sizeof(*c));
-  grpc_channel_args *final_args =
-      grpc_channel_args_merge(args->args, f->merge_args);
-  grpc_subchannel *s;
   memset(c, 0, sizeof(*c));
   c->base.vtable = &connector_vtable;
   c->security_connector = f->security_connector;
@@ -262,25 +264,19 @@ static grpc_subchannel *client_channel_factory_create_subchannel(
   }
   gpr_mu_init(&c->mu);
   gpr_ref_init(&c->refs, 1);
-  args->args = final_args;
-  s = grpc_subchannel_create(exec_ctx, &c->base, args);
+  grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args);
   grpc_connector_unref(exec_ctx, &c->base);
-  grpc_channel_args_destroy(final_args);
   return s;
 }
 
 static grpc_channel *client_channel_factory_create_channel(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
     const char *target, grpc_client_channel_type type,
-    grpc_channel_args *args) {
+    const grpc_channel_args *args) {
   client_channel_factory *f = (client_channel_factory *)cc_factory;
-
-  grpc_channel_args *final_args = grpc_channel_args_merge(args, f->merge_args);
-  grpc_channel *channel = grpc_channel_create(exec_ctx, target, final_args,
-                                              GRPC_CLIENT_CHANNEL, NULL);
-  grpc_channel_args_destroy(final_args);
-
-  grpc_resolver *resolver = grpc_resolver_create(target);
+  grpc_channel *channel =
+      grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL);
+  grpc_resolver *resolver = grpc_resolver_create(target, args);
   if (resolver != NULL) {
     grpc_client_channel_finish_initialization(
         exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base);
@@ -290,7 +286,6 @@ static grpc_channel *client_channel_factory_create_channel(
                                 "client_channel_factory_create_channel");
     channel = NULL;
   }
-
   return channel;
 }
 
@@ -307,19 +302,13 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
                                          const char *target,
                                          const grpc_channel_args *args,
                                          void *reserved) {
-  grpc_arg connector_arg;
-  grpc_channel_args *args_copy;
-  grpc_channel_args *new_args_from_connector;
-  grpc_channel_security_connector *security_connector;
-  client_channel_factory *f;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-
   GRPC_API_TRACE(
       "grpc_secure_channel_create(creds=%p, target=%s, args=%p, "
       "reserved=%p)",
       4, (creds, target, args, reserved));
   GPR_ASSERT(reserved == NULL);
-
+  // Make sure security connector does not already exist in args.
   if (grpc_find_security_connector_in_args(args) != NULL) {
     gpr_log(GPR_ERROR, "Cannot set security context in channel args.");
     grpc_exec_ctx_finish(&exec_ctx);
@@ -327,7 +316,9 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
         target, GRPC_STATUS_INTERNAL,
         "Security connector exists in channel args.");
   }
-
+  // Create security connector and construct new channel args.
+  grpc_channel_security_connector *security_connector;
+  grpc_channel_args *new_args_from_connector;
   if (grpc_channel_credentials_create_security_connector(
           creds, target, args, &security_connector, &new_args_from_connector) !=
       GRPC_SECURITY_OK) {
@@ -335,35 +326,30 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
     return grpc_lame_client_channel_create(
         target, GRPC_STATUS_INTERNAL, "Failed to create security connector.");
   }
-
-  connector_arg = grpc_security_connector_to_arg(&security_connector->base);
-  args_copy = grpc_channel_args_copy_and_add(
+  grpc_arg connector_arg =
+      grpc_security_connector_to_arg(&security_connector->base);
+  grpc_channel_args *new_args = grpc_channel_args_copy_and_add(
       new_args_from_connector != NULL ? new_args_from_connector : args,
       &connector_arg, 1);
-
-  f = gpr_malloc(sizeof(*f));
-  memset(f, 0, sizeof(*f));
-  f->base.vtable = &client_channel_factory_vtable;
-  gpr_ref_init(&f->refs, 1);
-
-  f->merge_args = grpc_channel_args_copy(args_copy);
-  grpc_channel_args_destroy(args_copy);
   if (new_args_from_connector != NULL) {
     grpc_channel_args_destroy(new_args_from_connector);
   }
-
+  // Create client channel factory.
+  client_channel_factory *f = gpr_malloc(sizeof(*f));
+  memset(f, 0, sizeof(*f));
+  f->base.vtable = &client_channel_factory_vtable;
+  gpr_ref_init(&f->refs, 1);
   GRPC_SECURITY_CONNECTOR_REF(&security_connector->base,
                               "grpc_secure_channel_create");
   f->security_connector = security_connector;
-
+  // Create channel.
   grpc_channel *channel = client_channel_factory_create_channel(
-      &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, NULL);
-
+      &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args);
+  // Clean up.
   GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base,
                                 "client_channel_factory_create_channel");
-
+  grpc_channel_args_destroy(new_args);
   grpc_client_channel_factory_unref(&exec_ctx, &f->base);
   grpc_exec_ctx_finish(&exec_ctx);
-
   return channel; /* may be NULL */
 }
diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
index e5b4025c5f..c18d618f96 100644
--- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
+++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
@@ -148,9 +148,7 @@ int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) {
   const size_t naddrs = resolved->naddrs;
   errors = gpr_malloc(sizeof(*errors) * naddrs);
   for (i = 0; i < naddrs; i++) {
-    errors[i] = grpc_tcp_server_add_port(
-        tcp, (struct sockaddr *)&resolved->addrs[i].addr,
-        resolved->addrs[i].len, &port_temp);
+    errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp);
     if (errors[i] == GRPC_ERROR_NONE) {
       if (port_num == -1) {
         port_num = port_temp;
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 a998dddd37..942638ad7f 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
@@ -287,9 +287,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
 
   errors = gpr_malloc(sizeof(*errors) * resolved->naddrs);
   for (i = 0; i < resolved->naddrs; i++) {
-    errors[i] = grpc_tcp_server_add_port(
-        tcp, (struct sockaddr *)&resolved->addrs[i].addr,
-        resolved->addrs[i].len, &port_temp);
+    errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp);
     if (errors[i] == GRPC_ERROR_NONE) {
       if (port_num == -1) {
         port_num = port_temp;
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 98190f7fb9..07639c1223 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -883,53 +883,51 @@ static bool contains_non_ok_status(grpc_metadata_batch *batch) {
 
 static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx,
                                      grpc_chttp2_transport *t,
-                                     grpc_chttp2_stream *s);
+                                     grpc_chttp2_stream *s) {
+  s->fetched_send_message_length +=
+      (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice);
+  grpc_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice);
+  if (s->id != 0) {
+    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
+  }
+}
 
 static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx,
                                           grpc_chttp2_transport *t,
                                           grpc_chttp2_stream *s) {
-  if (s->fetching_send_message == NULL) {
-    /* Stream was cancelled before message fetch completed */
-    abort(); /* TODO(ctiller): what cleanup here? */
-    return;
-  }
-  if (s->fetched_send_message_length == s->fetching_send_message->length) {
-    int64_t notify_offset = s->next_message_end_offset;
-    if (notify_offset <= s->flow_controlled_bytes_written) {
-      grpc_chttp2_complete_closure_step(
-          exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE,
-          "fetching_send_message_finished");
-    } else {
-      grpc_chttp2_write_cb *cb = t->write_cb_pool;
-      if (cb == NULL) {
-        cb = gpr_malloc(sizeof(*cb));
+  for (;;) {
+    if (s->fetching_send_message == NULL) {
+      /* Stream was cancelled before message fetch completed */
+      abort(); /* TODO(ctiller): what cleanup here? */
+      return;  /* early out */
+    }
+    if (s->fetched_send_message_length == s->fetching_send_message->length) {
+      int64_t notify_offset = s->next_message_end_offset;
+      if (notify_offset <= s->flow_controlled_bytes_written) {
+        grpc_chttp2_complete_closure_step(
+            exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE,
+            "fetching_send_message_finished");
       } else {
-        t->write_cb_pool = cb->next;
+        grpc_chttp2_write_cb *cb = t->write_cb_pool;
+        if (cb == NULL) {
+          cb = gpr_malloc(sizeof(*cb));
+        } else {
+          t->write_cb_pool = cb->next;
+        }
+        cb->call_at_byte = notify_offset;
+        cb->closure = s->fetching_send_message_finished;
+        s->fetching_send_message_finished = NULL;
+        cb->next = s->on_write_finished_cbs;
+        s->on_write_finished_cbs = cb;
       }
-      cb->call_at_byte = notify_offset;
-      cb->closure = s->fetching_send_message_finished;
-      s->fetching_send_message_finished = NULL;
-      cb->next = s->on_write_finished_cbs;
-      s->on_write_finished_cbs = cb;
+      s->fetching_send_message = NULL;
+      return; /* early out */
+    } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message,
+                                     &s->fetching_slice, UINT32_MAX,
+                                     &s->complete_fetch)) {
+      add_fetched_slice_locked(exec_ctx, t, s);
     }
-    s->fetching_send_message = NULL;
-  } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message,
-                                   &s->fetching_slice, UINT32_MAX,
-                                   &s->complete_fetch)) {
-    add_fetched_slice_locked(exec_ctx, t, s);
-  }
-}
-
-static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx,
-                                     grpc_chttp2_transport *t,
-                                     grpc_chttp2_stream *s) {
-  s->fetched_send_message_length +=
-      (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice);
-  grpc_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice);
-  if (s->id != 0) {
-    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
   }
-  continue_fetching_send_locked(exec_ctx, t, s);
 }
 
 static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs,
@@ -938,6 +936,7 @@ static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs,
   grpc_chttp2_transport *t = s->t;
   if (error == GRPC_ERROR_NONE) {
     add_fetched_slice_locked(exec_ctx, t, s);
+    continue_fetching_send_locked(exec_ctx, t, s);
   } else {
     /* TODO(ctiller): what to do here */
     abort();