Merge remote-tracking branch 'upstream/master' into handshaker_api

20 files changed, 675 insertions, 30 deletions

diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c
index dec25efe61..af913d8a9d 100644
--- a/src/core/ext/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/lb_policy/grpclb/grpclb.c
@@ -767,6 +767,9 @@ static lb_client_data *lb_client_data_create(glb_lb_policy *glb_policy) {
   lb_client->deadline = gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
                                      gpr_time_from_seconds(3, GPR_TIMESPAN));
 
+  /* Note the following LB call progresses every time there's activity in \a
+   * glb_policy->base.interested_parties, which is comprised of the polling
+   * entities passed to glb_pick(). */
   lb_client->lb_call = grpc_channel_create_pollset_set_call(
       glb_policy->lb_channel, NULL, GRPC_PROPAGATE_DEFAULTS,
       glb_policy->base.interested_parties, "/BalanceLoad",
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index f2f5465201..0e8dfb7d96 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -94,7 +94,8 @@ static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t,
 
 static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
 static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
-                                  grpc_chttp2_transport *t, grpc_error *error);
+                                  grpc_chttp2_transport *t, grpc_error *error,
+                                  const char *reason);
 
 /** Set a transport level setting, and push it to our peer */
 static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -876,7 +877,7 @@ static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
       set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE,
                       "start_writing:nothing_to_write");
     }
-    end_waiting_for_write(exec_ctx, t, GRPC_ERROR_CREATE("Nothing to write"));
+    end_waiting_for_write(exec_ctx, t, GRPC_ERROR_NONE, "Nothing to write");
     if (t->ep && !t->endpoint_reading) {
       destroy_endpoint(exec_ctx, t);
     }
@@ -925,11 +926,18 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   }
 }
 
+/* error may be GRPC_ERROR_NONE if there is no error allocated yet.
+   In that case, use "reason" as the text for a new error. */
 static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
-                                  grpc_chttp2_transport *t, grpc_error *error) {
+                                  grpc_chttp2_transport *t, grpc_error *error,
+                                  const char *reason) {
   grpc_chttp2_stream_global *stream_global;
   while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global,
                                                          &stream_global)) {
+    if (error == GRPC_ERROR_NONE && reason != NULL) {
+      /* create error object. */
+      error = GRPC_ERROR_CREATE(reason);
+    }
     fail_pending_writes(exec_ctx, &t->global, stream_global,
                         GRPC_ERROR_REF(error));
     GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes");
@@ -951,7 +959,7 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx,
 
   grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing);
 
-  end_waiting_for_write(exec_ctx, t, error);
+  end_waiting_for_write(exec_ctx, t, error, NULL);
 
   switch (t->executor.write_state) {
     case GRPC_CHTTP2_WRITING_INACTIVE:
diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h
index ee68701456..6b73cce380 100644
--- a/src/core/lib/channel/channel_stack.h
+++ b/src/core/lib/channel/channel_stack.h
@@ -51,6 +51,10 @@
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/transport/transport.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 typedef struct grpc_channel_element grpc_channel_element;
 typedef struct grpc_call_element grpc_call_element;
 
@@ -291,4 +295,8 @@ extern int grpc_trace_channel;
 #define GRPC_CALL_LOG_OP(sev, elem, op) \
   if (grpc_trace_channel) grpc_call_log_op(sev, elem, op)
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_STACK_H */
diff --git a/src/core/lib/channel/channel_stack_builder.h b/src/core/lib/channel/channel_stack_builder.h
index 0e6bfd9aa6..4a00f7bfdb 100644
--- a/src/core/lib/channel/channel_stack_builder.h
+++ b/src/core/lib/channel/channel_stack_builder.h
@@ -39,6 +39,10 @@
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/channel_stack.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 /// grpc_channel_stack_builder offers a programmatic interface to selected
 /// and order channel filters
 typedef struct grpc_channel_stack_builder grpc_channel_stack_builder;
@@ -158,4 +162,8 @@ void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder);
 
 extern int grpc_trace_channel_stack_builder;
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_STACK_BUILDER_H */
diff --git a/src/core/lib/channel/context.h b/src/core/lib/channel/context.h
index c50e84279d..071c5f695c 100644
--- a/src/core/lib/channel/context.h
+++ b/src/core/lib/channel/context.h
@@ -34,10 +34,19 @@
 #ifndef GRPC_CORE_LIB_CHANNEL_CONTEXT_H
 #define GRPC_CORE_LIB_CHANNEL_CONTEXT_H
 
-/* Call object context pointers */
+/// Call object context pointers.
+
+/// Call context is represented as an array of \a grpc_call_context_elements.
+/// This enum represents the indexes into the array, where each index
+/// contains a different type of value.
 typedef enum {
+  /// Value is either a \a grpc_client_security_context or a
+  /// \a grpc_server_security_context.
   GRPC_CONTEXT_SECURITY = 0,
+
+  /// Value is a \a census_context.
   GRPC_CONTEXT_TRACING,
+
   GRPC_CONTEXT_COUNT
 } grpc_context_index;
 
diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c
index 38ebd2dbcb..ea8d241154 100644
--- a/src/core/lib/iomgr/tcp_server_posix.c
+++ b/src/core/lib/iomgr/tcp_server_posix.c
@@ -90,10 +90,12 @@ struct grpc_tcp_listener {
   grpc_closure read_closure;
   grpc_closure destroyed_closure;
   struct grpc_tcp_listener *next;
-  /* When we add a listener, more than one can be created, mainly because of
-     IPv6. A sibling will still be in the normal list, but will be flagged
-     as such. Any action, such as ref or unref, will affect all of the
-     siblings in the list. */
+  /* sibling is a linked list of all listeners for a given port. add_port and
+     clone_port place all new listeners in the same sibling list. A member of
+     the 'sibling' list is also a member of the 'next' list. The head of each
+     sibling list has is_sibling==0, and subsequent members of sibling lists
+     have is_sibling==1. is_sibling allows separate sibling lists to be
+     identified while iterating through 'next'. */
   struct grpc_tcp_listener *sibling;
   int is_sibling;
 };
@@ -306,7 +308,7 @@ static grpc_error *prepare_socket(int fd, const struct sockaddr *addr,
 
   GPR_ASSERT(fd >= 0);
 
-  if (so_reuseport) {
+  if (so_reuseport && !grpc_is_unix_socket(addr)) {
     err = grpc_set_socket_reuse_port(fd, 1);
     if (err != GRPC_ERROR_NONE) goto error;
   }
@@ -480,6 +482,9 @@ static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd,
   return err;
 }
 
+/* Insert count new listeners after listener. Every new listener will have the
+   same listen address as listener (SO_REUSEPORT must be enabled). Every new
+   listener is a sibling of listener. */
 static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
   grpc_tcp_listener *sp = NULL;
   char *addr_str;
@@ -506,6 +511,11 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
     sp = gpr_malloc(sizeof(grpc_tcp_listener));
     sp->next = listener->next;
     listener->next = sp;
+    /* sp (the new listener) is a sibling of 'listener' (the original
+       listener). */
+    sp->is_sibling = 1;
+    sp->sibling = listener->sibling;
+    listener->sibling = sp;
     sp->server = listener->server;
     sp->fd = fd;
     sp->emfd = grpc_fd_create(fd, name);
@@ -514,8 +524,6 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
     sp->port = port;
     sp->port_index = listener->port_index;
     sp->fd_index = listener->fd_index + count - i;
-    sp->is_sibling = 1;
-    sp->sibling = listener->is_sibling ? listener->sibling : listener;
     GPR_ASSERT(sp->emfd);
     while (listener->server->tail->next != NULL) {
       listener->server->tail = listener->server->tail->next;
@@ -685,7 +693,8 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
   s->pollset_count = pollset_count;
   sp = s->head;
   while (sp != NULL) {
-    if (s->so_reuseport && pollset_count > 1) {
+    if (s->so_reuseport && !grpc_is_unix_socket(&sp->addr.sockaddr) &&
+        pollset_count > 1) {
       GPR_ASSERT(GRPC_LOG_IF_ERROR(
           "clone_port", clone_port(sp, (unsigned)(pollset_count - 1))));
       for (i = 0; i < pollset_count; i++) {
diff --git a/src/core/lib/security/context/security_context.h b/src/core/lib/security/context/security_context.h
index ef0c06b1fb..4e7666dfe3 100644
--- a/src/core/lib/security/context/security_context.h
+++ b/src/core/lib/security/context/security_context.h
@@ -37,6 +37,10 @@
 #include "src/core/lib/iomgr/pollset.h"
 #include "src/core/lib/security/credentials/credentials.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 /* --- grpc_auth_context ---
 
    High level authentication context object. Can optionally be chained. */
@@ -111,4 +115,8 @@ grpc_auth_context *grpc_auth_context_from_arg(const grpc_arg *arg);
 grpc_auth_context *grpc_find_auth_context_in_args(
     const grpc_channel_args *args);
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_SECURITY_CONTEXT_SECURITY_CONTEXT_H */
diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h
index 7eff7b8883..4c62974346 100644
--- a/src/core/lib/surface/channel.h
+++ b/src/core/lib/surface/channel.h
@@ -42,6 +42,14 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
                                   grpc_channel_stack_type channel_stack_type,
                                   grpc_transport *optional_transport);
 
+/** Create a call given a grpc_channel, in order to call \a method.
+    Progress is tied to activity on \a pollset_set. The returned call object is
+    meant to be used with \a grpc_call_start_batch_and_execute, which relies on
+    callbacks to signal completions. \a method and \a host need
+    only live through the invocation of this function. If \a parent_call is
+    non-NULL, it must be a server-side call. It will be used to propagate
+    properties from the server call to this new client call, depending on the
+    value of \a propagation_mask (see propagation_bits.h for possible values) */
 grpc_call *grpc_channel_create_pollset_set_call(
     grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask,
     grpc_pollset_set *pollset_set, const char *method, const char *host,
diff --git a/src/core/lib/surface/channel_init.h b/src/core/lib/surface/channel_init.h
index 3a18a61ddb..b53f2aefb9 100644
--- a/src/core/lib/surface/channel_init.h
+++ b/src/core/lib/surface/channel_init.h
@@ -40,6 +40,10 @@
 
 #define GRPC_CHANNEL_INIT_BUILTIN_PRIORITY 10000
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 /// This module provides a way for plugins (and the grpc core library itself)
 /// to register mutators for channel stacks.
 /// It also provides a universal entry path to run those mutators to build
@@ -84,4 +88,8 @@ bool grpc_channel_init_create_stack(grpc_exec_ctx *exec_ctx,
                                     grpc_channel_stack_builder *builder,
                                     grpc_channel_stack_type type);
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_SURFACE_CHANNEL_INIT_H */
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index 1d61ec3bbb..64afcecc07 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -272,7 +272,7 @@ static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
 }
 
 static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
-                          int send_goaway, grpc_error *send_disconnect) {
+                          bool send_goaway, grpc_error *send_disconnect) {
   grpc_transport_op op;
   struct shutdown_cleanup_args *sc;
   grpc_channel_element *elem;
@@ -293,7 +293,7 @@ static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
 
 static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
                                          channel_broadcaster *cb,
-                                         int send_goaway,
+                                         bool send_goaway,
                                          grpc_error *force_disconnect) {
   size_t i;
 
@@ -1252,7 +1252,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
     l->destroy(&exec_ctx, server, l->arg, &l->destroy_done);
   }
 
-  channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 1, 0);
+  channel_broadcaster_shutdown(&exec_ctx, &broadcaster, true /* send_goaway */,
+                               GRPC_ERROR_NONE);
 
 done:
   grpc_exec_ctx_finish(&exec_ctx);
@@ -1268,7 +1269,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
   channel_broadcaster_init(server, &broadcaster);
   gpr_mu_unlock(&server->mu_global);
 
-  channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 0,
+  channel_broadcaster_shutdown(&exec_ctx, &broadcaster, false /* send_goaway */,
                                GRPC_ERROR_CREATE("Cancelling all calls"));
   grpc_exec_ctx_finish(&exec_ctx);
 }
diff --git a/src/core/lib/transport/byte_stream.h b/src/core/lib/transport/byte_stream.h
index 95519a9eaf..e64dce6283 100644
--- a/src/core/lib/transport/byte_stream.h
+++ b/src/core/lib/transport/byte_stream.h
@@ -59,13 +59,9 @@ struct grpc_byte_stream {
  * on_complete will not be called), 0 if the bytes will be available
  * asynchronously.
  *
- * on entry, *remaining can be set as a hint as to the maximum number
+ * max_size_hint can be set as a hint as to the maximum number
  * of bytes that would be acceptable to read.
  *
- * fills *buffer, *length, *remaining with the bytes, length of bytes
- * and length of data remaining to be read before either returning 1
- * or calling on_complete.
- *
  * once a slice is returned into *slice, it is owned by the caller.
  */
 int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/transport/metadata.h b/src/core/lib/transport/metadata.h
index 6d82f4d681..2b0921c8d7 100644
--- a/src/core/lib/transport/metadata.h
+++ b/src/core/lib/transport/metadata.h
@@ -37,6 +37,10 @@
 #include <grpc/support/slice.h>
 #include <grpc/support/useful.h>
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 /* This file provides a mechanism for tracking metadata through the grpc stack.
    It's not intended for consumption outside of the library.
 
@@ -164,4 +168,8 @@ void grpc_mdctx_global_shutdown(void);
 extern gpr_slice (*grpc_chttp2_base64_encode_and_huffman_compress)(
     gpr_slice input);
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_H */
diff --git a/src/core/lib/transport/metadata_batch.c b/src/core/lib/transport/metadata_batch.c
index e4398abeb7..84b5a74d51 100644
--- a/src/core/lib/transport/metadata_batch.c
+++ b/src/core/lib/transport/metadata_batch.c
@@ -33,6 +33,7 @@
 
 #include "src/core/lib/transport/metadata_batch.h"
 
+#include <stdbool.h>
 #include <string.h>
 
 #include <grpc/support/alloc.h>
@@ -187,7 +188,7 @@ void grpc_metadata_batch_clear(grpc_metadata_batch *batch) {
   grpc_metadata_batch_filter(batch, no_metadata_for_you, NULL);
 }
 
-int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) {
+bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) {
   return batch->list.head == NULL &&
          gpr_time_cmp(gpr_inf_future(batch->deadline.clock_type),
                       batch->deadline) == 0;
diff --git a/src/core/lib/transport/metadata_batch.h b/src/core/lib/transport/metadata_batch.h
index 7af823f7ca..0424b4db98 100644
--- a/src/core/lib/transport/metadata_batch.h
+++ b/src/core/lib/transport/metadata_batch.h
@@ -34,12 +34,18 @@
 #ifndef GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H
 #define GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H
 
+#include <stdbool.h>
+
 #include <grpc/grpc.h>
 #include <grpc/support/port_platform.h>
 #include <grpc/support/slice.h>
 #include <grpc/support/time.h>
 #include "src/core/lib/transport/metadata.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 typedef struct grpc_linked_mdelem {
   grpc_mdelem *md;
   struct grpc_linked_mdelem *next;
@@ -64,7 +70,7 @@ typedef struct grpc_metadata_batch {
 void grpc_metadata_batch_init(grpc_metadata_batch *batch);
 void grpc_metadata_batch_destroy(grpc_metadata_batch *batch);
 void grpc_metadata_batch_clear(grpc_metadata_batch *batch);
-int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch);
+bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch);
 
 /* Returns the transport size of the batch. */
 size_t grpc_metadata_batch_size(grpc_metadata_batch *batch);
@@ -125,4 +131,8 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch *comd);
   } while (0)
 #endif
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H */
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index e33fc5c761..1705e6e582 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -43,6 +43,10 @@
 #include "src/core/lib/transport/byte_stream.h"
 #include "src/core/lib/transport/metadata_batch.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 /* forward declarations */
 typedef struct grpc_transport grpc_transport;
 
@@ -158,7 +162,7 @@ typedef struct grpc_transport_op {
   /** should we send a goaway?
       after a goaway is sent, once there are no more active calls on
       the transport, the transport should disconnect */
-  int send_goaway;
+  bool send_goaway;
   /** what should the goaway contain? */
   grpc_status_code goaway_status;
   gpr_slice *goaway_message;
@@ -268,4 +272,8 @@ void grpc_transport_destroy(grpc_exec_ctx *exec_ctx, grpc_transport *transport);
 char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx,
                               grpc_transport *transport);
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif /* GRPC_CORE_LIB_TRANSPORT_TRANSPORT_H */
diff --git a/src/cpp/common/channel_filter.cc b/src/cpp/common/channel_filter.cc
new file mode 100644
index 0000000000..25cd49cb7c
--- /dev/null
+++ b/src/cpp/common/channel_filter.cc
@@ -0,0 +1,112 @@
+/*
+ *
+ * Copyright 2016, 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 <string.h>
+
+#include "src/core/lib/channel/channel_stack.h"
+#include "src/cpp/common/channel_filter.h"
+
+namespace grpc {
+
+// MetadataBatch
+
+grpc_linked_mdelem *MetadataBatch::AddMetadata(const string &key,
+                                               const string &value) {
+  grpc_linked_mdelem *storage = new grpc_linked_mdelem;
+  memset(storage, 0, sizeof(grpc_linked_mdelem));
+  storage->md = grpc_mdelem_from_strings(key.c_str(), value.c_str());
+  grpc_metadata_batch_link_head(batch_, storage);
+  return storage;
+}
+
+// ChannelData
+
+void ChannelData::StartTransportOp(grpc_exec_ctx *exec_ctx,
+                                   grpc_channel_element *elem,
+                                   TransportOp *op) {
+  grpc_channel_next_op(exec_ctx, elem, op->op());
+}
+
+// CallData
+
+void CallData::StartTransportStreamOp(grpc_exec_ctx *exec_ctx,
+                                      grpc_call_element *elem,
+                                      TransportStreamOp *op) {
+  grpc_call_next_op(exec_ctx, elem, op->op());
+}
+
+void CallData::SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,
+                                      grpc_call_element *elem,
+                                      grpc_polling_entity *pollent) {
+  grpc_call_stack_ignore_set_pollset_or_pollset_set(exec_ctx, elem, pollent);
+}
+
+char *CallData::GetPeer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
+  return grpc_call_next_get_peer(exec_ctx, elem);
+}
+
+// internal code used by RegisterChannelFilter()
+
+namespace internal {
+
+// Note: Implicitly initialized to nullptr due to static lifetime.
+std::vector<FilterRecord> *channel_filters;
+
+namespace {
+
+bool MaybeAddFilter(grpc_channel_stack_builder *builder, void *arg) {
+  const FilterRecord &filter = *(FilterRecord *)arg;
+  if (filter.include_filter) {
+    const grpc_channel_args *args =
+        grpc_channel_stack_builder_get_channel_arguments(builder);
+    if (!filter.include_filter(*args)) return true;
+  }
+  return grpc_channel_stack_builder_prepend_filter(builder, &filter.filter,
+                                                   nullptr, nullptr);
+}
+
+}  // namespace
+
+void ChannelFilterPluginInit() {
+  for (size_t i = 0; i < channel_filters->size(); ++i) {
+    FilterRecord &filter = (*channel_filters)[i];
+    grpc_channel_init_register_stage(filter.stack_type, filter.priority,
+                                     MaybeAddFilter, (void *)&filter);
+  }
+}
+
+void ChannelFilterPluginShutdown() {}
+
+}  // namespace internal
+
+}  // namespace grpc
diff --git a/src/cpp/common/channel_filter.h b/src/cpp/common/channel_filter.h
new file mode 100644
index 0000000000..3fbacd824c
--- /dev/null
+++ b/src/cpp/common/channel_filter.h
@@ -0,0 +1,386 @@
+/*
+ *
+ * Copyright 2016, 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 GRPCXX_CHANNEL_FILTER_H
+#define GRPCXX_CHANNEL_FILTER_H
+
+#include <grpc++/impl/codegen/config.h>
+#include <grpc/census.h>
+#include <grpc/grpc.h>
+#include <grpc/impl/codegen/alloc.h>
+
+#include <functional>
+#include <vector>
+
+#include "src/core/lib/channel/channel_stack.h"
+#include "src/core/lib/security/context/security_context.h"
+#include "src/core/lib/surface/channel_init.h"
+#include "src/core/lib/transport/metadata_batch.h"
+
+/// An interface to define filters.
+///
+/// To define a filter, implement a subclass of each of \c CallData and
+/// \c ChannelData. Then register the filter using something like this:
+/// \code{.cpp}
+///   RegisterChannelFilter<MyChannelDataSubclass, MyCallDataSubclass>(
+///       "name-of-filter", GRPC_SERVER_CHANNEL, INT_MAX, nullptr);
+/// \endcode
+
+namespace grpc {
+
+/// A C++ wrapper for the \c grpc_metadata_batch struct.
+class MetadataBatch {
+ public:
+  /// Borrows a pointer to \a batch, but does NOT take ownership.
+  /// The caller must ensure that \a batch continues to exist for as
+  /// long as the MetadataBatch object does.
+  explicit MetadataBatch(grpc_metadata_batch *batch) : batch_(batch) {}
+
+  grpc_metadata_batch *batch() const { return batch_; }
+
+  /// Adds metadata and returns the newly allocated storage.
+  /// The caller takes ownership of the result, which must exist for the
+  /// lifetime of the gRPC call.
+  grpc_linked_mdelem *AddMetadata(const string &key, const string &value);
+
+  class const_iterator : public std::iterator<std::bidirectional_iterator_tag,
+                                              const grpc_mdelem> {
+   public:
+    const grpc_mdelem &operator*() const { return *elem_->md; }
+    const grpc_mdelem *operator->() const { return elem_->md; }
+
+    const_iterator &operator++() {
+      elem_ = elem_->next;
+      return *this;
+    }
+    const_iterator operator++(int) {
+      const_iterator tmp(*this);
+      operator++();
+      return tmp;
+    }
+    const_iterator &operator--() {
+      elem_ = elem_->prev;
+      return *this;
+    }
+    const_iterator operator--(int) {
+      const_iterator tmp(*this);
+      operator--();
+      return tmp;
+    }
+
+    bool operator==(const const_iterator &other) const {
+      return elem_ == other.elem_;
+    }
+    bool operator!=(const const_iterator &other) const {
+      return elem_ != other.elem_;
+    }
+
+   private:
+    friend class MetadataBatch;
+    explicit const_iterator(grpc_linked_mdelem *elem) : elem_(elem) {}
+
+    grpc_linked_mdelem *elem_;
+  };
+
+  const_iterator begin() const { return const_iterator(batch_->list.head); }
+  const_iterator end() const { return const_iterator(nullptr); }
+
+ private:
+  grpc_metadata_batch *batch_;  // Not owned.
+};
+
+/// A C++ wrapper for the \c grpc_transport_op struct.
+class TransportOp {
+ public:
+  /// Borrows a pointer to \a op, but does NOT take ownership.
+  /// The caller must ensure that \a op continues to exist for as
+  /// long as the TransportOp object does.
+  explicit TransportOp(grpc_transport_op *op) : op_(op) {}
+
+  grpc_transport_op *op() const { return op_; }
+
+  // TODO(roth): Add a C++ wrapper for grpc_error?
+  grpc_error *disconnect_with_error() const {
+    return op_->disconnect_with_error;
+  }
+  bool send_goaway() const { return op_->send_goaway; }
+
+  // TODO(roth): Add methods for additional fields as needed.
+
+ private:
+  grpc_transport_op *op_;  // Not owned.
+};
+
+/// A C++ wrapper for the \c grpc_transport_stream_op struct.
+class TransportStreamOp {
+ public:
+  /// Borrows a pointer to \a op, but does NOT take ownership.
+  /// The caller must ensure that \a op continues to exist for as
+  /// long as the TransportStreamOp object does.
+  explicit TransportStreamOp(grpc_transport_stream_op *op)
+      : op_(op),
+        send_initial_metadata_(op->send_initial_metadata),
+        send_trailing_metadata_(op->send_trailing_metadata),
+        recv_initial_metadata_(op->recv_initial_metadata),
+        recv_trailing_metadata_(op->recv_trailing_metadata) {}
+
+  grpc_transport_stream_op *op() const { return op_; }
+
+  grpc_closure *on_complete() const { return op_->on_complete; }
+  void set_on_complete(grpc_closure *closure) { op_->on_complete = closure; }
+
+  MetadataBatch *send_initial_metadata() {
+    return op_->send_initial_metadata == nullptr ? nullptr
+                                                 : &send_initial_metadata_;
+  }
+  MetadataBatch *send_trailing_metadata() {
+    return op_->send_trailing_metadata == nullptr ? nullptr
+                                                  : &send_trailing_metadata_;
+  }
+  MetadataBatch *recv_initial_metadata() {
+    return op_->recv_initial_metadata == nullptr ? nullptr
+                                                 : &recv_initial_metadata_;
+  }
+  MetadataBatch *recv_trailing_metadata() {
+    return op_->recv_trailing_metadata == nullptr ? nullptr
+                                                  : &recv_trailing_metadata_;
+  }
+
+  uint32_t *send_initial_metadata_flags() const {
+    return &op_->send_initial_metadata_flags;
+  }
+
+  grpc_closure *recv_initial_metadata_ready() const {
+    return op_->recv_initial_metadata_ready;
+  }
+  void set_recv_initial_metadata_ready(grpc_closure *closure) {
+    op_->recv_initial_metadata_ready = closure;
+  }
+
+  grpc_byte_stream *send_message() const { return op_->send_message; }
+  void set_send_message(grpc_byte_stream *send_message) {
+    op_->send_message = send_message;
+  }
+
+  /// To be called only on clients and servers, respectively.
+  grpc_client_security_context *client_security_context() const {
+    return (grpc_client_security_context *)op_->context[GRPC_CONTEXT_SECURITY]
+        .value;
+  }
+  grpc_server_security_context *server_security_context() const {
+    return (grpc_server_security_context *)op_->context[GRPC_CONTEXT_SECURITY]
+        .value;
+  }
+
+  census_context *get_census_context() const {
+    return (census_context *)op_->context[GRPC_CONTEXT_TRACING].value;
+  }
+
+ private:
+  grpc_transport_stream_op *op_;  // Not owned.
+  MetadataBatch send_initial_metadata_;
+  MetadataBatch send_trailing_metadata_;
+  MetadataBatch recv_initial_metadata_;
+  MetadataBatch recv_trailing_metadata_;
+};
+
+/// Represents channel data.
+class ChannelData {
+ public:
+  virtual ~ChannelData() {
+    if (peer_) gpr_free((void *)peer_);
+  }
+
+  /// Caller does NOT take ownership of result.
+  const char *peer() const { return peer_; }
+
+  // TODO(roth): Find a way to avoid passing elem into these methods.
+  virtual void StartTransportOp(grpc_exec_ctx *exec_ctx,
+                                grpc_channel_element *elem, TransportOp *op);
+
+ protected:
+  /// Takes ownership of \a peer.
+  ChannelData(const grpc_channel_args &args, const char *peer) : peer_(peer) {}
+
+ private:
+  const char *peer_;
+};
+
+/// Represents call data.
+class CallData {
+ public:
+  virtual ~CallData() {}
+
+  /// Initializes the call data.
+  virtual grpc_error *Init() { return GRPC_ERROR_NONE; }
+
+  // TODO(roth): Find a way to avoid passing elem into these methods.
+
+  /// Starts a new stream operation.
+  virtual void StartTransportStreamOp(grpc_exec_ctx *exec_ctx,
+                                      grpc_call_element *elem,
+                                      TransportStreamOp *op);
+
+  /// Sets a pollset or pollset set.
+  virtual void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,
+                                      grpc_call_element *elem,
+                                      grpc_polling_entity *pollent);
+
+  /// Gets the peer name.
+  virtual char *GetPeer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem);
+
+ protected:
+  explicit CallData(const ChannelData &) {}
+};
+
+namespace internal {
+
+// Defines static members for passing to C core.
+// Members of this class correspond to the members of the C
+// grpc_channel_filter struct.
+template <typename ChannelDataType, typename CallDataType>
+class ChannelFilter GRPC_FINAL {
+ public:
+  static const size_t channel_data_size = sizeof(ChannelDataType);
+
+  static void InitChannelElement(grpc_exec_ctx *exec_ctx,
+                                 grpc_channel_element *elem,
+                                 grpc_channel_element_args *args) {
+    const char *peer =
+        args->optional_transport
+            ? grpc_transport_get_peer(exec_ctx, args->optional_transport)
+            : nullptr;
+    // Construct the object in the already-allocated memory.
+    new (elem->channel_data) ChannelDataType(*args->channel_args, peer);
+  }
+
+  static void DestroyChannelElement(grpc_exec_ctx *exec_ctx,
+                                    grpc_channel_element *elem) {
+    reinterpret_cast<ChannelDataType *>(elem->channel_data)->~ChannelDataType();
+  }
+
+  static void StartTransportOp(grpc_exec_ctx *exec_ctx,
+                               grpc_channel_element *elem,
+                               grpc_transport_op *op) {
+    ChannelDataType *channel_data = (ChannelDataType *)elem->channel_data;
+    TransportOp op_wrapper(op);
+    channel_data->StartTransportOp(exec_ctx, elem, &op_wrapper);
+  }
+
+  static const size_t call_data_size = sizeof(CallDataType);
+
+  static grpc_error *InitCallElement(grpc_exec_ctx *exec_ctx,
+                                     grpc_call_element *elem,
+                                     grpc_call_element_args *args) {
+    const ChannelDataType &channel_data =
+        *(ChannelDataType *)elem->channel_data;
+    // Construct the object in the already-allocated memory.
+    CallDataType *call_data = new (elem->call_data) CallDataType(channel_data);
+    return call_data->Init();
+  }
+
+  static void DestroyCallElement(grpc_exec_ctx *exec_ctx,
+                                 grpc_call_element *elem,
+                                 const grpc_call_final_info *final_info,
+                                 void *and_free_memory) {
+    reinterpret_cast<CallDataType *>(elem->call_data)->~CallDataType();
+  }
+
+  static void StartTransportStreamOp(grpc_exec_ctx *exec_ctx,
+                                     grpc_call_element *elem,
+                                     grpc_transport_stream_op *op) {
+    CallDataType *call_data = (CallDataType *)elem->call_data;
+    TransportStreamOp op_wrapper(op);
+    call_data->StartTransportStreamOp(exec_ctx, elem, &op_wrapper);
+  }
+
+  static void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,
+                                     grpc_call_element *elem,
+                                     grpc_polling_entity *pollent) {
+    CallDataType *call_data = (CallDataType *)elem->call_data;
+    call_data->SetPollsetOrPollsetSet(exec_ctx, elem, pollent);
+  }
+
+  static char *GetPeer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
+    CallDataType *call_data = (CallDataType *)elem->call_data;
+    return call_data->GetPeer(exec_ctx, elem);
+  }
+};
+
+struct FilterRecord {
+  grpc_channel_stack_type stack_type;
+  int priority;
+  std::function<bool(const grpc_channel_args &)> include_filter;
+  grpc_channel_filter filter;
+};
+extern std::vector<FilterRecord> *channel_filters;
+
+void ChannelFilterPluginInit();
+void ChannelFilterPluginShutdown();
+
+}  // namespace internal
+
+/// Registers a new filter.
+/// Must be called by only one thread at a time.
+/// The \a include_filter argument specifies a function that will be called
+/// to determine at run-time whether or not to add the filter. If the
+/// value is nullptr, the filter will be added unconditionally.
+template <typename ChannelDataType, typename CallDataType>
+void RegisterChannelFilter(
+    const char *name, grpc_channel_stack_type stack_type, int priority,
+    std::function<bool(const grpc_channel_args &)> include_filter) {
+  // If we haven't been called before, initialize channel_filters and
+  // call grpc_register_plugin().
+  if (internal::channel_filters == nullptr) {
+    grpc_register_plugin(internal::ChannelFilterPluginInit,
+                         internal::ChannelFilterPluginShutdown);
+    internal::channel_filters = new std::vector<internal::FilterRecord>();
+  }
+  // Add an entry to channel_filters. The filter will be added when the
+  // C-core initialization code calls ChannelFilterPluginInit().
+  typedef internal::ChannelFilter<ChannelDataType, CallDataType> FilterType;
+  internal::FilterRecord filter_record = {
+      stack_type,
+      priority,
+      include_filter,
+      {FilterType::StartTransportStreamOp, FilterType::StartTransportOp,
+       FilterType::call_data_size, FilterType::InitCallElement,
+       FilterType::SetPollsetOrPollsetSet, FilterType::DestroyCallElement,
+       FilterType::channel_data_size, FilterType::InitChannelElement,
+       FilterType::DestroyChannelElement, FilterType::GetPeer, name}};
+  internal::channel_filters->push_back(filter_record);
+}
+
+}  // namespace grpc
+
+#endif  // GRPCXX_CHANNEL_FILTER_H
diff --git a/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py b/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py
index 3c00f686ce..9cae96a00d 100644
--- a/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py
+++ b/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py
@@ -32,12 +32,12 @@
 import threading
 import time
 import unittest
-from concurrent import futures
 
 import grpc
 from grpc import _channel
 from grpc import _server
 from tests.unit.framework.common import test_constants
+from tests.unit import _thread_pool
 
 
 def _ready_in_connectivities(connectivities):
@@ -104,7 +104,8 @@ class ChannelConnectivityTest(unittest.TestCase):
         grpc.ChannelConnectivity.READY, fifth_connectivities)
 
   def test_immediately_connectable_channel_connectivity(self):
-    server = _server.Server(futures.ThreadPoolExecutor(max_workers=0), ())
+    thread_pool = _thread_pool.RecordingThreadPool(max_workers=None)
+    server = _server.Server(thread_pool, ())
     port = server.add_insecure_port('[::]:0')
     server.start()
     first_callback = _Callback()
@@ -141,9 +142,11 @@ class ChannelConnectivityTest(unittest.TestCase):
         fourth_connectivities)
     self.assertNotIn(
         grpc.ChannelConnectivity.SHUTDOWN, fourth_connectivities)
+    self.assertFalse(thread_pool.was_used())
 
   def test_reachable_then_unreachable_channel_connectivity(self):
-    server = _server.Server(futures.ThreadPoolExecutor(max_workers=0), ())
+    thread_pool = _thread_pool.RecordingThreadPool(max_workers=None)
+    server = _server.Server(thread_pool, ())
     port = server.add_insecure_port('[::]:0')
     server.start()
     callback = _Callback()
@@ -155,6 +158,7 @@ class ChannelConnectivityTest(unittest.TestCase):
     server.stop(None)
     callback.block_until_connectivities_satisfy(_last_connectivity_is_not_ready)
     channel.unsubscribe(callback.update)
+    self.assertFalse(thread_pool.was_used())
 
 
 if __name__ == '__main__':
diff --git a/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py b/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py
index e8982ed2de..24f5b45b18 100644
--- a/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py
+++ b/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py
@@ -31,12 +31,12 @@
 
 import threading
 import unittest
-from concurrent import futures
 
 import grpc
 from grpc import _channel
 from grpc import _server
 from tests.unit.framework.common import test_constants
+from tests.unit import _thread_pool
 
 
 class _Callback(object):
@@ -78,7 +78,8 @@ class ChannelReadyFutureTest(unittest.TestCase):
     self.assertFalse(ready_future.running())
 
   def test_immediately_connectable_channel_connectivity(self):
-    server = _server.Server(futures.ThreadPoolExecutor(max_workers=0), ())
+    thread_pool = _thread_pool.RecordingThreadPool(max_workers=None)
+    server = _server.Server(thread_pool, ())
     port = server.add_insecure_port('[::]:0')
     server.start()
     channel = grpc.insecure_channel('localhost:{}'.format(port))
@@ -97,6 +98,7 @@ class ChannelReadyFutureTest(unittest.TestCase):
     self.assertFalse(ready_future.cancelled())
     self.assertTrue(ready_future.done())
     self.assertFalse(ready_future.running())
+    self.assertFalse(thread_pool.was_used())
 
 
 if __name__ == '__main__':
diff --git a/src/python/grpcio_tests/tests/unit/_thread_pool.py b/src/python/grpcio_tests/tests/unit/_thread_pool.py
new file mode 100644
index 0000000000..f13cc2f86f
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_thread_pool.py
@@ -0,0 +1,48 @@
+# Copyright 2016, 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.
+
+import threading
+from concurrent import futures
+
+
+class RecordingThreadPool(futures.Executor):
+  """A thread pool that records if used."""
+  def __init__(self, max_workers):
+    self._tp_executor = futures.ThreadPoolExecutor(max_workers=max_workers)
+    self._lock = threading.Lock()
+    self._was_used = False
+
+  def submit(self, fn, *args, **kwargs):
+    with self._lock:
+      self._was_used = True
+    self._tp_executor.submit(fn, *args, **kwargs)
+
+  def was_used(self):
+    with self._lock:
+      return self._was_used