Merge github.com:grpc/grpc into i-have-no-master

10 files changed, 220 insertions, 7 deletions

diff --git a/src/core/client_config/connector.h b/src/core/client_config/connector.h
index e9b8be4b53..a649f143ae 100644
--- a/src/core/client_config/connector.h
+++ b/src/core/client_config/connector.h
@@ -51,6 +51,8 @@ typedef struct {
   /** address to connect to */
   const struct sockaddr *addr;
   size_t addr_len;
+  /** initial connect string to send */
+  gpr_slice initial_connect_string;
   /** deadline for connection */
   gpr_timespec deadline;
   /** channel arguments (to be passed to transport) */
diff --git a/src/core/client_config/default_initial_connect_string.c b/src/core/client_config/default_initial_connect_string.c
new file mode 100644
index 0000000000..6a4e23e6f2
--- /dev/null
+++ b/src/core/client_config/default_initial_connect_string.c
@@ -0,0 +1,39 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc/support/slice.h>
+#include "src/core/iomgr/sockaddr.h"
+
+void grpc_set_default_initial_connect_string(struct sockaddr **addr,
+                                             size_t *addr_len,
+                                             gpr_slice *initial_str) {}
diff --git a/src/core/client_config/initial_connect_string.c b/src/core/client_config/initial_connect_string.c
new file mode 100644
index 0000000000..19afa1675a
--- /dev/null
+++ b/src/core/client_config/initial_connect_string.c
@@ -0,0 +1,53 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "src/core/client_config/initial_connect_string.h"
+
+#include <stddef.h>
+
+extern void grpc_set_default_initial_connect_string(struct sockaddr **addr,
+                                                    size_t *addr_len,
+                                                    gpr_slice *initial_str);
+
+static grpc_set_initial_connect_string_func g_set_initial_connect_string_func =
+    grpc_set_default_initial_connect_string;
+
+void grpc_test_set_initial_connect_string_function(
+    grpc_set_initial_connect_string_func func) {
+  g_set_initial_connect_string_func = func;
+}
+
+void grpc_set_initial_connect_string(struct sockaddr **addr, size_t *addr_len,
+                                     gpr_slice *initial_str) {
+  g_set_initial_connect_string_func(addr, addr_len, initial_str);
+}
diff --git a/src/core/client_config/initial_connect_string.h b/src/core/client_config/initial_connect_string.h
new file mode 100644
index 0000000000..b6dca7134a
--- /dev/null
+++ b/src/core/client_config/initial_connect_string.h
@@ -0,0 +1,50 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H
+#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H
+
+#include <grpc/support/slice.h>
+#include "src/core/iomgr/sockaddr.h"
+
+typedef void (*grpc_set_initial_connect_string_func)(struct sockaddr **addr,
+                                                     size_t *addr_len,
+                                                     gpr_slice *initial_str);
+void grpc_test_set_initial_connect_string_function(
+    grpc_set_initial_connect_string_func func);
+
+/** Set a string to be sent once connected. Optionally reset addr. */
+void grpc_set_initial_connect_string(struct sockaddr **addr, size_t *addr_len,
+                                     gpr_slice *connect_string);
+
+#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H */
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 958d1a05cf..b210e7f679 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -40,6 +40,7 @@
 #include "src/core/channel/channel_args.h"
 #include "src/core/channel/client_channel.h"
 #include "src/core/channel/connected_channel.h"
+#include "src/core/client_config/initial_connect_string.h"
 #include "src/core/iomgr/timer.h"
 #include "src/core/profiling/timers.h"
 #include "src/core/surface/channel.h"
@@ -80,6 +81,9 @@ struct grpc_subchannel {
   struct sockaddr *addr;
   size_t addr_len;
 
+  /** initial string to send to peer */
+  gpr_slice initial_connect_string;
+
   /** set during connection */
   grpc_connect_out_args connecting_result;
 
@@ -199,6 +203,7 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
   gpr_free((void *)c->filters);
   grpc_channel_args_destroy(c->args);
   gpr_free(c->addr);
+  gpr_slice_unref(c->initial_connect_string);
   grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
   grpc_connector_unref(exec_ctx, c->connector);
   grpc_pollset_set_destroy(&c->pollset_set);
@@ -250,6 +255,8 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
   memcpy(c->addr, args->addr, args->addr_len);
   grpc_pollset_set_init(&c->pollset_set);
   c->addr_len = args->addr_len;
+  grpc_set_initial_connect_string(&c->addr, &c->addr_len,
+                                  &c->initial_connect_string);
   c->args = grpc_channel_args_copy(args->args);
   c->random = random_seed();
   grpc_closure_init(&c->connected, subchannel_connected, c);
@@ -267,6 +274,7 @@ static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   args.addr_len = c->addr_len;
   args.deadline = compute_connect_deadline(c);
   args.channel_args = c->args;
+  args.initial_connect_string = c->initial_connect_string;
 
   grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
                               GRPC_CHANNEL_CONNECTING, "state_change");
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 5d701fa87b..c179248214 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -194,6 +194,7 @@ void grpc_pollset_init(grpc_pollset *pollset) {
   pollset->in_flight_cbs = 0;
   pollset->shutting_down = 0;
   pollset->called_shutdown = 0;
+  pollset->kicked_without_pollers = 0;
   pollset->idle_jobs.head = pollset->idle_jobs.tail = NULL;
   pollset->local_wakeup_cache = NULL;
   pollset->kicked_without_pollers = 0;
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index d7722f26a7..8970b56eea 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -37,6 +37,8 @@
 #include <string.h>
 
 #include <grpc/support/alloc.h>
+#include <grpc/support/slice.h>
+#include <grpc/support/slice_buffer.h>
 
 #include "src/core/census/grpc_filter.h"
 #include "src/core/channel/channel_args.h"
@@ -56,6 +58,8 @@ typedef struct {
   grpc_closure *notify;
   grpc_connect_in_args args;
   grpc_connect_out_args *result;
+  grpc_closure initial_string_sent;
+  gpr_slice_buffer initial_string_buffer;
 
   grpc_endpoint *tcp;
 
@@ -70,15 +74,31 @@ static void connector_ref(grpc_connector *con) {
 static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
   connector *c = (connector *)con;
   if (gpr_unref(&c->refs)) {
+    /* c->initial_string_buffer does not need to be destroyed */
     gpr_free(c);
   }
 }
 
+static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
+                                           int success) {
+  connector_unref(exec_ctx, arg);
+}
+
 static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
   connector *c = arg;
   grpc_closure *notify;
   grpc_endpoint *tcp = c->tcp;
   if (tcp != NULL) {
+    if (!GPR_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
+      grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
+                        c);
+      gpr_slice_buffer_init(&c->initial_string_buffer);
+      gpr_slice_buffer_add(&c->initial_string_buffer,
+                           c->args.initial_connect_string);
+      connector_ref(arg);
+      grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer,
+                          &c->initial_string_sent);
+    }
     c->result->transport =
         grpc_create_chttp2_transport(exec_ctx, c->args.channel_args, tcp, 1);
     grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL,
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 344b4d79f7..826ab8e86c 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -37,6 +37,8 @@
 #include <string.h>
 
 #include <grpc/support/alloc.h>
+#include <grpc/support/slice.h>
+#include <grpc/support/slice_buffer.h>
 
 #include "src/core/census/grpc_filter.h"
 #include "src/core/channel/channel_args.h"
@@ -61,6 +63,8 @@ typedef struct {
   grpc_closure *notify;
   grpc_connect_in_args args;
   grpc_connect_out_args *result;
+  grpc_closure initial_string_sent;
+  gpr_slice_buffer initial_string_buffer;
 
   gpr_mu mu;
   grpc_endpoint *connecting_endpoint;
@@ -77,6 +81,7 @@ static void connector_ref(grpc_connector *con) {
 static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
   connector *c = (connector *)con;
   if (gpr_unref(&c->refs)) {
+    /* c->initial_string_buffer does not need to be destroyed */
     gpr_free(c);
   }
 }
@@ -115,6 +120,14 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
   notify->cb(exec_ctx, notify->cb_arg, 1);
 }
 
+static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
+                                           int success) {
+  connector *c = arg;
+  grpc_security_connector_do_handshake(exec_ctx, &c->security_connector->base,
+                                       c->connecting_endpoint,
+                                       on_secure_handshake_done, c);
+}
+
 static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
   connector *c = arg;
   grpc_closure *notify;
@@ -124,8 +137,19 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
     GPR_ASSERT(c->connecting_endpoint == NULL);
     c->connecting_endpoint = tcp;
     gpr_mu_unlock(&c->mu);
-    grpc_security_connector_do_handshake(exec_ctx, &c->security_connector->base,
-                                         tcp, on_secure_handshake_done, c);
+    if (!GPR_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
+      grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
+                        c);
+      gpr_slice_buffer_init(&c->initial_string_buffer);
+      gpr_slice_buffer_add(&c->initial_string_buffer,
+                           c->args.initial_connect_string);
+      grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer,
+                          &c->initial_string_sent);
+    } else {
+      grpc_security_connector_do_handshake(exec_ctx,
+                                           &c->security_connector->base, tcp,
+                                           on_secure_handshake_done, c);
+    }
   } else {
     memset(c->result, 0, sizeof(*c->result));
     notify = c->notify;
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index ebfa347398..3952f8a0e8 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -191,6 +191,9 @@ typedef struct {
       copied to next_stream_id in parsing when parsing commences */
   gpr_uint32 next_stream_id;
 
+  /** how far to lookahead in a stream? */
+  gpr_uint32 stream_lookahead;
+
   /** last received stream id */
   gpr_uint32 last_incoming_stream_id;
 
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 79ebaa70fe..ff0c616386 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -53,7 +53,6 @@
 #include "src/core/transport/transport_impl.h"
 
 #define DEFAULT_WINDOW 65535
-#define GRPC_CHTTP2_STREAM_LOOKAHEAD DEFAULT_WINDOW
 #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024)
 #define MAX_WINDOW 0x7fffffffu
 
@@ -238,6 +237,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   t->global.is_client = is_client;
   t->writing.outgoing_window = DEFAULT_WINDOW;
   t->parsing.incoming_window = DEFAULT_WINDOW;
+  t->global.stream_lookahead = DEFAULT_WINDOW;
   t->global.connection_window_target = DEFAULT_CONNECTION_WINDOW_TARGET;
   t->global.ping_counter = 1;
   t->global.pings.next = t->global.pings.prev = &t->global.pings;
@@ -328,6 +328,18 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
               (gpr_uint32)channel_args->args[i].value.integer;
         }
       } else if (0 == strcmp(channel_args->args[i].key,
+                             GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES)) {
+        if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
+          gpr_log(GPR_ERROR, "%s: must be an integer",
+                  GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES);
+        } else if (channel_args->args[i].value.integer <= 5) {
+          gpr_log(GPR_ERROR, "%s: must be at least 5",
+                  GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES);
+        } else {
+          t->global.stream_lookahead =
+              (gpr_uint32)channel_args->args[i].value.integer;
+        }
+      } else if (0 == strcmp(channel_args->args[i].key,
                              GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_DECODER)) {
         if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
           gpr_log(GPR_ERROR, "%s: must be an integer",
@@ -1397,8 +1409,8 @@ static void incoming_byte_stream_update_flow_control(
   gpr_uint32 max_recv_bytes;
 
   /* clamp max recv hint to an allowable size */
-  if (max_size_hint >= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD) {
-    max_recv_bytes = GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD;
+  if (max_size_hint >= GPR_UINT32_MAX - transport_global->stream_lookahead) {
+    max_recv_bytes = GPR_UINT32_MAX - transport_global->stream_lookahead;
   } else {
     max_recv_bytes = (gpr_uint32)max_size_hint;
   }
@@ -1411,8 +1423,9 @@ static void incoming_byte_stream_update_flow_control(
   }
 
   /* add some small lookahead to keep pipelines flowing */
-  GPR_ASSERT(max_recv_bytes <= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD);
-  max_recv_bytes += GRPC_CHTTP2_STREAM_LOOKAHEAD;
+  GPR_ASSERT(max_recv_bytes <=
+             GPR_UINT32_MAX - transport_global->stream_lookahead);
+  max_recv_bytes += transport_global->stream_lookahead;
   if (stream_global->max_recv_bytes < max_recv_bytes) {
     gpr_uint32 add_max_recv_bytes =
         max_recv_bytes - stream_global->max_recv_bytes;