Merge branch 'master' into fray

20 files changed, 1534 insertions, 105 deletions

diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index c430b56fa2..54ee75af28 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -37,6 +37,7 @@
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
+#include <grpc/support/useful.h>
 
 #include <string.h>
 
@@ -146,3 +147,65 @@ grpc_channel_args *grpc_channel_args_set_compression_algorithm(
   tmp.value.integer = algorithm;
   return grpc_channel_args_copy_and_add(a, &tmp, 1);
 }
+
+/** Returns 1 if the argument for compression algorithm's enabled states bitset
+ * was found in \a a, returning the arg's value in \a states. Otherwise, returns
+ * 0. */
+static int find_compression_algorithm_states_bitset(
+    const grpc_channel_args *a, int **states_arg) {
+  if (a != NULL) {
+    size_t i;
+    for (i = 0; i < a->num_args; ++i) {
+      if (a->args[i].type == GRPC_ARG_INTEGER &&
+          !strcmp(GRPC_COMPRESSION_ALGORITHM_STATE_ARG, a->args[i].key)) {
+        *states_arg = &a->args[i].value.integer;
+        return 1; /* GPR_TRUE */
+      }
+    }
+  }
+  return 0; /* GPR_FALSE */
+}
+
+grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
+    grpc_channel_args **a,
+    grpc_compression_algorithm algorithm,
+    int state) {
+  int *states_arg;
+  grpc_channel_args *result = *a;
+  const int states_arg_found =
+      find_compression_algorithm_states_bitset(*a, &states_arg);
+
+  if (states_arg_found) {
+    if (state != 0) {
+      GPR_BITSET(states_arg, algorithm);
+    } else {
+      GPR_BITCLEAR(states_arg, algorithm);
+    }
+  } else {
+    /* create a new arg */
+    grpc_arg tmp;
+    tmp.type = GRPC_ARG_INTEGER;
+    tmp.key = GRPC_COMPRESSION_ALGORITHM_STATE_ARG;
+    /* all enabled by default */
+    tmp.value.integer = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1;
+    if (state != 0) {
+      GPR_BITSET(&tmp.value.integer, algorithm);
+    } else {
+      GPR_BITCLEAR(&tmp.value.integer, algorithm);
+    }
+    result = grpc_channel_args_copy_and_add(*a, &tmp, 1);
+    grpc_channel_args_destroy(*a);
+    *a = result;
+  }
+  return result;
+}
+
+int grpc_channel_args_compression_algorithm_get_states(
+    const grpc_channel_args *a) {
+  int *states_arg;
+  if (find_compression_algorithm_states_bitset(a, &states_arg)) {
+    return *states_arg;
+  } else {
+    return (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; /* All algs. enabled */
+  }
+}
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index 7e6ddd3997..06a6012dee 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -67,4 +67,24 @@ grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
 grpc_channel_args *grpc_channel_args_set_compression_algorithm(
     grpc_channel_args *a, grpc_compression_algorithm algorithm);
 
+/** Sets the support for the given compression algorithm. By default, all
+ * compression algorithms are enabled. It's an error to disable an algorithm set
+ * by grpc_channel_args_set_compression_algorithm.
+ *
+ * Returns an instance will the updated algorithm states. The \a a pointer is
+ * modified to point to the returned instance (which may be different from the
+ * input value of \a a). */
+grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
+    grpc_channel_args **a,
+    grpc_compression_algorithm algorithm,
+    int enabled);
+
+/** Returns the bitset representing the support state (true for enabled, false
+ * for disabled) for compression algorithms.
+ *
+ * The i-th bit of the returned bitset corresponds to the i-th entry in the
+ * grpc_compression_algorithm enum. */
+int grpc_channel_args_compression_algorithm_get_states(
+    const grpc_channel_args *a);
+
 #endif /* GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H */
diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h
index c474e4dbf1..337596cb74 100644
--- a/src/core/iomgr/pollset.h
+++ b/src/core/iomgr/pollset.h
@@ -74,10 +74,9 @@ void grpc_pollset_destroy(grpc_pollset *pollset);
    grpc_pollset_work, and it is guaranteed that GRPC_POLLSET_MU(pollset) will
    not be released by grpc_pollset_work AFTER worker has been destroyed.
 
-   Returns true if some work has been done, and false if the deadline
-   expired. */
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline);
+   Tries not to block past deadline. */
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
+                       gpr_timespec now, gpr_timespec deadline);
 
 /* Break one polling thread out of polling work for this pollset.
    If specific_worker is GRPC_POLLSET_KICK_BROADCAST, kick ALL the workers.
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index 5ea9dd2101..fe66ebed77 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -181,7 +181,7 @@ static void multipoll_with_epoll_pollset_maybe_work(
   pfds[1].events = POLLIN;
   pfds[1].revents = 0;
 
-  poll_rv = poll(pfds, 2, timeout_ms);
+  poll_rv = grpc_poll_function(pfds, 2, timeout_ms);
 
   if (poll_rv < 0) {
     if (errno != EINTR) {
diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
index 001fcecf76..30ee6e24db 100644
--- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c
+++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
@@ -144,7 +144,7 @@ static void multipoll_with_poll_pollset_maybe_work(
                                         POLLOUT, &watchers[i]);
   }
 
-  r = poll(pfds, pfd_count, timeout);
+  r = grpc_poll_function(pfds, pfd_count, timeout);
 
   for (i = 1; i < pfd_count; i++) {
     grpc_fd_end_poll(&watchers[i], pfds[i].revents & POLLIN,
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index a01f9ff727..6bd1b61f24 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -38,7 +38,6 @@
 #include "src/core/iomgr/pollset_posix.h"
 
 #include <errno.h>
-#include <poll.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
@@ -57,6 +56,8 @@
 GPR_TLS_DECL(g_current_thread_poller);
 GPR_TLS_DECL(g_current_thread_worker);
 
+grpc_poll_function_type grpc_poll_function = poll;
+
 static void remove_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
   worker->prev->next = worker->next;
   worker->next->prev = worker->prev;
@@ -89,6 +90,7 @@ static void push_front_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
 }
 
 void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) {
+  /* pollset->mu already held */
   if (specific_worker != NULL) {
     if (specific_worker == GRPC_POLLSET_KICK_BROADCAST) {
       for (specific_worker = p->root_worker.next;
@@ -168,14 +170,10 @@ static void finish_shutdown(grpc_pollset *pollset) {
   pollset->shutdown_done_cb(pollset->shutdown_done_arg);
 }
 
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline) {
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
+                       gpr_timespec now, gpr_timespec deadline) {
   /* pollset->mu already held */
-  gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
   int added_worker = 0;
-  if (gpr_time_cmp(now, deadline) > 0) {
-    return 0;
-  }
   /* this must happen before we (potentially) drop pollset->mu */
   worker->next = worker->prev = NULL;
   /* TODO(ctiller): pool these */
@@ -217,7 +215,6 @@ done:
       gpr_mu_lock(&pollset->mu);
     }
   }
-  return 1;
 }
 
 void grpc_pollset_shutdown(grpc_pollset *pollset,
@@ -456,7 +453,7 @@ static void basic_pollset_maybe_work(grpc_pollset *pollset,
 
   /* poll fd count (argument 2) is shortened by one if we have no events
      to poll on - such that it only includes the kicker */
-  r = poll(pfd, nfds, timeout);
+  r = grpc_poll_function(pfd, nfds, timeout);
   GRPC_TIMER_MARK(GRPC_PTAG_POLL_FINISHED, r);
 
   if (fd) {
diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h
index a3ea353de6..69bd9cca8c 100644
--- a/src/core/iomgr/pollset_posix.h
+++ b/src/core/iomgr/pollset_posix.h
@@ -34,6 +34,8 @@
 #ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H
 #define GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H
 
+#include <poll.h>
+
 #include <grpc/support/sync.h>
 #include "src/core/iomgr/wakeup_fd_posix.h"
 
@@ -118,4 +120,8 @@ void grpc_poll_become_multipoller(grpc_pollset *pollset, struct grpc_fd **fds,
  * be locked) */
 int grpc_pollset_has_workers(grpc_pollset *pollset);
 
+/* override to allow tests to hook poll() usage */
+typedef int (*grpc_poll_function_type)(struct pollfd *, nfds_t, int);
+extern grpc_poll_function_type grpc_poll_function;
+
 #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H */
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 8710395ab3..07522c8a0c 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -99,14 +99,9 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
   gpr_mu_destroy(&pollset->mu);
 }
 
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline) {
-  gpr_timespec now;
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, 
+                       gpr_timespec now, gpr_timespec deadline) {
   int added_worker = 0;
-  now = gpr_now(GPR_CLOCK_MONOTONIC);
-  if (gpr_time_cmp(now, deadline) > 0) {
-    return 0 /* GPR_FALSE */;
-  }
   worker->next = worker->prev = NULL;
   gpr_cv_init(&worker->cv);
   if (grpc_maybe_call_delayed_callbacks(&pollset->mu, 1 /* GPR_TRUE */)) {
@@ -127,7 +122,6 @@ done:
   if (added_worker) {
     remove_worker(pollset, worker);
   }
-  return 1 /* GPR_TRUE */;
 }
 
 void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) {
diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c
index d6092ece32..3631de867a 100644
--- a/src/core/security/google_default_credentials.c
+++ b/src/core/security/google_default_credentials.c
@@ -115,7 +115,7 @@ static int is_stack_running_on_compute_engine(void) {
   gpr_mu_lock(GRPC_POLLSET_MU(&detector.pollset));
   while (!detector.is_done) {
     grpc_pollset_worker worker;
-    grpc_pollset_work(&detector.pollset, &worker,
+    grpc_pollset_work(&detector.pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC),
                       gpr_inf_future(GPR_CLOCK_MONOTONIC));
   }
   gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset));
diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c
index 2f42f01f53..6e831431fa 100644
--- a/src/core/security/server_auth_filter.c
+++ b/src/core/security/server_auth_filter.c
@@ -104,24 +104,34 @@ static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) {
   return md;
 }
 
-static void on_md_processing_done(void *user_data,
-                                  const grpc_metadata *consumed_md,
-                                  size_t num_consumed_md, int success) {
+static void on_md_processing_done(
+    void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md,
+    const grpc_metadata *response_md, size_t num_response_md,
+    grpc_status_code status, const char *error_details) {
   grpc_call_element *elem = user_data;
   call_data *calld = elem->call_data;
 
-  if (success) {
+  /* TODO(jboeuf): Implement support for response_md. */
+  if (response_md != NULL && num_response_md > 0) {
+    gpr_log(GPR_INFO,
+            "response_md in auth metadata processing not supported for now. "
+            "Ignoring...");
+  }
+
+  if (status == GRPC_STATUS_OK) {
     calld->consumed_md = consumed_md;
     calld->num_consumed_md = num_consumed_md;
     grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md,
                                elem);
-    calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
+    calld->on_done_recv->cb(calld->on_done_recv->cb_arg, 1);
   } else {
-    gpr_slice message = gpr_slice_from_copied_string(
-        "Authentication metadata processing failed.");
+    gpr_slice message;
+    error_details = error_details != NULL
+                    ? error_details
+                    : "Authentication metadata processing failed.";
+    message = gpr_slice_from_copied_string(error_details);
     grpc_sopb_reset(calld->recv_ops);
-    grpc_transport_stream_op_add_close(&calld->transport_op,
-                                       GRPC_STATUS_UNAUTHENTICATED, &message);
+    grpc_transport_stream_op_add_close(&calld->transport_op, status, &message);
     grpc_call_next_op(elem, &calld->transport_op);
   }
 }
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 77443a7ae8..b58115a93f 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -170,6 +170,9 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
                                       gpr_timespec deadline, void *reserved) {
   grpc_event ret;
   grpc_pollset_worker worker;
+  int first_loop = 1;
+  gpr_timespec now;
+
   GPR_ASSERT(!reserved);
 
   deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
@@ -196,12 +199,15 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
       ret.type = GRPC_QUEUE_SHUTDOWN;
       break;
     }
-    if (!grpc_pollset_work(&cc->pollset, &worker, deadline)) {
+    now = gpr_now(GPR_CLOCK_MONOTONIC);
+    if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
+    first_loop = 0;
+    grpc_pollset_work(&cc->pollset, &worker, now, deadline);
   }
   GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
   GRPC_CQ_INTERNAL_UNREF(cc, "next");
@@ -239,6 +245,9 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
   grpc_cq_completion *c;
   grpc_cq_completion *prev;
   grpc_pollset_worker worker;
+  gpr_timespec now;
+  int first_loop = 1;
+
   GPR_ASSERT(!reserved);
 
   deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
@@ -281,13 +290,16 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
-    if (!grpc_pollset_work(&cc->pollset, &worker, deadline)) {
+    now = gpr_now(GPR_CLOCK_MONOTONIC);
+    if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
       del_plucker(cc, tag, &worker);
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
+    first_loop = 0;
+    grpc_pollset_work(&cc->pollset, &worker, now, deadline);
     del_plucker(cc, tag, &worker);
   }
 done:
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 57707121f3..3db23c2abe 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -50,6 +50,52 @@
 
 namespace grpc {
 
+class Server::UnimplementedAsyncRequestContext {
+ protected:
+  UnimplementedAsyncRequestContext() : generic_stream_(&server_context_) {}
+
+  GenericServerContext server_context_;
+  GenericServerAsyncReaderWriter generic_stream_;
+};
+
+class Server::UnimplementedAsyncRequest GRPC_FINAL
+    : public UnimplementedAsyncRequestContext,
+      public GenericAsyncRequest {
+ public:
+  UnimplementedAsyncRequest(Server* server, ServerCompletionQueue* cq)
+      : GenericAsyncRequest(server, &server_context_, &generic_stream_, cq, cq,
+                            NULL, false),
+        server_(server),
+        cq_(cq) {}
+
+  bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
+
+  ServerContext* context() { return &server_context_; }
+  GenericServerAsyncReaderWriter* stream() { return &generic_stream_; }
+
+ private:
+  Server* const server_;
+  ServerCompletionQueue* const cq_;
+};
+
+typedef SneakyCallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus>
+    UnimplementedAsyncResponseOp;
+class Server::UnimplementedAsyncResponse GRPC_FINAL
+    : public UnimplementedAsyncResponseOp {
+ public:
+  UnimplementedAsyncResponse(UnimplementedAsyncRequest* request);
+  ~UnimplementedAsyncResponse() { delete request_; }
+
+  bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE {
+    bool r = UnimplementedAsyncResponseOp::FinalizeResult(tag, status);
+    delete this;
+    return r;
+  }
+
+ private:
+  UnimplementedAsyncRequest* const request_;
+};
+
 class Server::ShutdownRequest GRPC_FINAL : public CompletionQueueTag {
  public:
   bool FinalizeResult(void** tag, bool* status) {
@@ -297,18 +343,23 @@ int Server::AddListeningPort(const grpc::string& addr,
   return creds->AddPortToServer(addr, server_);
 }
 
-bool Server::Start() {
+bool Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) {
   GPR_ASSERT(!started_);
   started_ = true;
   grpc_server_start(server_);
 
   if (!has_generic_service_) {
-    unknown_method_.reset(new RpcServiceMethod(
-        "unknown", RpcMethod::BIDI_STREAMING, new UnknownMethodHandler));
-    // Use of emplace_back with just constructor arguments is not accepted here
-    // by gcc-4.4 because it can't match the anonymous nullptr with a proper
-    // constructor implicitly. Construct the object and use push_back.
-    sync_methods_->push_back(SyncRequest(unknown_method_.get(), nullptr));
+    if (!sync_methods_->empty()) {
+      unknown_method_.reset(new RpcServiceMethod(
+          "unknown", RpcMethod::BIDI_STREAMING, new UnknownMethodHandler));
+      // Use of emplace_back with just constructor arguments is not accepted
+      // here by gcc-4.4 because it can't match the anonymous nullptr with a 
+      // proper constructor implicitly. Construct the object and use push_back.
+      sync_methods_->push_back(SyncRequest(unknown_method_.get(), nullptr));
+    }
+    for (size_t i = 0; i < num_cqs; i++) {
+      new UnimplementedAsyncRequest(this, cqs[i]);
+    }
   }
   // Start processing rpcs.
   if (!sync_methods_->empty()) {
@@ -370,12 +421,14 @@ void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
 
 Server::BaseAsyncRequest::BaseAsyncRequest(
     Server* server, ServerContext* context,
-    ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
+    ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag,
+    bool delete_on_finalize)
     : server_(server),
       context_(context),
       stream_(stream),
       call_cq_(call_cq),
       tag_(tag),
+      delete_on_finalize_(delete_on_finalize),
       call_(nullptr) {
   memset(&initial_metadata_array_, 0, sizeof(initial_metadata_array_));
 }
@@ -402,14 +455,16 @@ bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) {
   // just the pointers inside call are copied here
   stream_->BindCall(&call);
   *tag = tag_;
-  delete this;
+  if (delete_on_finalize_) {
+    delete this;
+  }
   return true;
 }
 
 Server::RegisteredAsyncRequest::RegisteredAsyncRequest(
     Server* server, ServerContext* context,
     ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
-    : BaseAsyncRequest(server, context, stream, call_cq, tag) {}
+    : BaseAsyncRequest(server, context, stream, call_cq, tag, true) {}
 
 void Server::RegisteredAsyncRequest::IssueRequest(
     void* registered_method, grpc_byte_buffer** payload,
@@ -423,8 +478,9 @@ void Server::RegisteredAsyncRequest::IssueRequest(
 Server::GenericAsyncRequest::GenericAsyncRequest(
     Server* server, GenericServerContext* context,
     ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
-    ServerCompletionQueue* notification_cq, void* tag)
-    : BaseAsyncRequest(server, context, stream, call_cq, tag) {
+    ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize)
+    : BaseAsyncRequest(server, context, stream, call_cq, tag,
+                       delete_on_finalize) {
   grpc_call_details_init(&call_details_);
   GPR_ASSERT(notification_cq);
   GPR_ASSERT(call_cq);
@@ -445,6 +501,25 @@ bool Server::GenericAsyncRequest::FinalizeResult(void** tag, bool* status) {
   return BaseAsyncRequest::FinalizeResult(tag, status);
 }
 
+bool Server::UnimplementedAsyncRequest::FinalizeResult(void** tag,
+                                                       bool* status) {
+  if (GenericAsyncRequest::FinalizeResult(tag, status) && *status) {
+    new UnimplementedAsyncRequest(server_, cq_);
+    new UnimplementedAsyncResponse(this);
+  } else {
+    delete this;
+  }
+  return false;
+}
+
+Server::UnimplementedAsyncResponse::UnimplementedAsyncResponse(
+    UnimplementedAsyncRequest* request)
+    : request_(request) {
+  Status status(StatusCode::UNIMPLEMENTED, "");
+  UnknownMethodHandler::FillOps(request_->context(), this);
+  request_->stream()->call_.PerformOps(this);
+}
+
 void Server::ScheduleCallback() {
   {
     grpc::unique_lock<grpc::mutex> lock(mu_);
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index db13c84a2b..b739cbfe62 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -99,12 +99,6 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
     thread_pool_ = CreateDefaultThreadPool();
     thread_pool_owned = true;
   }
-  // Async services only, create a thread pool to handle requests to unknown
-  // services.
-  if (!thread_pool_ && !generic_service_ && !async_services_.empty()) {
-    thread_pool_ = new FixedSizeThreadPool(1);
-    thread_pool_owned = true;
-  }
   std::unique_ptr<Server> server(
       new Server(thread_pool_, thread_pool_owned, max_message_size_));
   for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
@@ -134,7 +128,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
       *port->selected_port = r;
     }
   }
-  if (!server->Start()) {
+  if (!server->Start(&cqs_[0], cqs_.size())) {
     return nullptr;
   }
   return server;
diff --git a/src/python/grpcio/grpc/_adapter/_c/types.h b/src/python/grpcio/grpc/_adapter/_c/types.h
index f646465c63..f6ff957baa 100644
--- a/src/python/grpcio/grpc/_adapter/_c/types.h
+++ b/src/python/grpcio/grpc/_adapter/_c/types.h
@@ -146,6 +146,7 @@ typedef struct Server {
   PyObject_HEAD
   grpc_server *c_serv;
   CompletionQueue *cq;
+  int shutdown_called;
 } Server;
 Server *pygrpc_Server_new(PyTypeObject *type, PyObject *args, PyObject *kwargs);
 void pygrpc_Server_dealloc(Server *self);
@@ -156,6 +157,7 @@ PyObject *pygrpc_Server_add_http2_port(
 PyObject *pygrpc_Server_start(Server *self, PyObject *ignored);
 PyObject *pygrpc_Server_shutdown(
     Server *self, PyObject *args, PyObject *kwargs);
+PyObject *pygrpc_Server_cancel_all_calls(Server *self, PyObject *unused);
 extern PyTypeObject pygrpc_Server_type;
 
 /*=========*/
diff --git a/src/python/grpcio/grpc/_adapter/_c/types/server.c b/src/python/grpcio/grpc/_adapter/_c/types/server.c
index 15c98f28eb..8feab8aab1 100644
--- a/src/python/grpcio/grpc/_adapter/_c/types/server.c
+++ b/src/python/grpcio/grpc/_adapter/_c/types/server.c
@@ -45,6 +45,8 @@ PyMethodDef pygrpc_Server_methods[] = {
      METH_KEYWORDS, ""},
     {"start", (PyCFunction)pygrpc_Server_start, METH_NOARGS, ""},
     {"shutdown", (PyCFunction)pygrpc_Server_shutdown, METH_KEYWORDS, ""},
+    {"cancel_all_calls", (PyCFunction)pygrpc_Server_cancel_all_calls,
+     METH_NOARGS, ""},
     {NULL}
 };
 const char pygrpc_Server_doc[] = "See grpc._adapter._types.Server.";
@@ -109,6 +111,7 @@ Server *pygrpc_Server_new(PyTypeObject *type, PyObject *args, PyObject *kwargs)
   pygrpc_discard_channel_args(c_args);
   self->cq = cq;
   Py_INCREF(self->cq);
+  self->shutdown_called = 0;
   return self;
 }
 
@@ -163,6 +166,7 @@ PyObject *pygrpc_Server_add_http2_port(
 
 PyObject *pygrpc_Server_start(Server *self, PyObject *ignored) {
   grpc_server_start(self->c_serv);
+  self->shutdown_called = 0;
   Py_RETURN_NONE;
 }
 
@@ -176,5 +180,17 @@ PyObject *pygrpc_Server_shutdown(
   }
   tag = pygrpc_produce_server_shutdown_tag(user_tag);
   grpc_server_shutdown_and_notify(self->c_serv, self->cq->c_cq, tag);
+  self->shutdown_called = 1;
+  Py_RETURN_NONE;
+}
+
+PyObject *pygrpc_Server_cancel_all_calls(Server *self, PyObject *unused) {
+  if (!self->shutdown_called) {
+    PyErr_SetString(
+        PyExc_RuntimeError,
+        "shutdown must have been called prior to calling cancel_all_calls!");
+    return NULL;
+  }
+  grpc_server_cancel_all_calls(self->c_serv);
   Py_RETURN_NONE;
 }
diff --git a/src/python/grpcio/grpc/_adapter/_low.py b/src/python/grpcio/grpc/_adapter/_low.py
index 147086e725..3859ebb0e2 100644
--- a/src/python/grpcio/grpc/_adapter/_low.py
+++ b/src/python/grpcio/grpc/_adapter/_low.py
@@ -124,3 +124,6 @@ class Server(_types.Server):
 
   def request_call(self, completion_queue, tag):
     return self.server.request_call(completion_queue.completion_queue, tag)
+
+  def cancel_all_calls(self):
+    return self.server.cancel_all_calls()
diff --git a/src/python/grpcio/grpc/framework/interfaces/face/__init__.py b/src/python/grpcio/grpc/framework/interfaces/face/__init__.py
new file mode 100644
index 0000000000..7086519106
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/interfaces/face/__init__.py
@@ -0,0 +1,30 @@
+# 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.
+
+
diff --git a/src/python/grpcio/grpc/framework/interfaces/face/face.py b/src/python/grpcio/grpc/framework/interfaces/face/face.py
new file mode 100644
index 0000000000..948e7505b6
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/interfaces/face/face.py
@@ -0,0 +1,933 @@
+# 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.
+
+"""Interfaces defining the Face layer of RPC Framework."""
+
+import abc
+import collections
+import enum
+
+# cardinality, style, abandonment, future, and stream are
+# referenced from specification in this module.
+from grpc.framework.common import cardinality  # pylint: disable=unused-import
+from grpc.framework.common import style  # pylint: disable=unused-import
+from grpc.framework.foundation import abandonment  # pylint: disable=unused-import
+from grpc.framework.foundation import future  # pylint: disable=unused-import
+from grpc.framework.foundation import stream  # pylint: disable=unused-import
+
+
+class NoSuchMethodError(Exception):
+  """Raised by customer code to indicate an unrecognized method.
+
+  Attributes:
+    group: The group of the unrecognized method.
+    name: The name of the unrecognized method.
+  """
+
+  def __init__(self, group, method):
+    """Constructor.
+
+    Args:
+      group: The group identifier of the unrecognized RPC name.
+      method: The method identifier of the unrecognized RPC name.
+    """
+    super(NoSuchMethodError, self).__init__()
+    self.group = group
+    self.method = method
+
+  def __repr__(self):
+    return 'face.NoSuchMethodError(%s, %s)' % (self.group, self.method,)
+
+
+class Abortion(
+    collections.namedtuple(
+        'Abortion',
+        ('kind', 'initial_metadata', 'terminal_metadata', 'code', 'details',))):
+  """A value describing RPC abortion.
+
+  Attributes:
+    kind: A Kind value identifying how the RPC failed.
+    initial_metadata: The initial metadata from the other side of the RPC or
+      None if no initial metadata value was received.
+    terminal_metadata: The terminal metadata from the other side of the RPC or
+      None if no terminal metadata value was received.
+    code: The code value from the other side of the RPC or None if no code value
+      was received.
+    details: The details value from the other side of the RPC or None if no
+      details value was received.
+  """
+
+  @enum.unique
+  class Kind(enum.Enum):
+    """Types of RPC abortion."""
+
+    CANCELLED = 'cancelled'
+    EXPIRED = 'expired'
+    LOCAL_SHUTDOWN = 'local shutdown'
+    REMOTE_SHUTDOWN = 'remote shutdown'
+    NETWORK_FAILURE = 'network failure'
+    LOCAL_FAILURE = 'local failure'
+    REMOTE_FAILURE = 'remote failure'
+
+
+class AbortionError(Exception):
+  """Common super type for exceptions indicating RPC abortion.
+
+    initial_metadata: The initial metadata from the other side of the RPC or
+      None if no initial metadata value was received.
+    terminal_metadata: The terminal metadata from the other side of the RPC or
+      None if no terminal metadata value was received.
+    code: The code value from the other side of the RPC or None if no code value
+      was received.
+    details: The details value from the other side of the RPC or None if no
+      details value was received.
+  """
+  __metaclass__ = abc.ABCMeta
+
+  def __init__(self, initial_metadata, terminal_metadata, code, details):
+    super(AbortionError, self).__init__()
+    self.initial_metadata = initial_metadata
+    self.terminal_metadata = terminal_metadata
+    self.code = code
+    self.details = details
+
+
+class CancellationError(AbortionError):
+  """Indicates that an RPC has been cancelled."""
+
+
+class ExpirationError(AbortionError):
+  """Indicates that an RPC has expired ("timed out")."""
+
+
+class LocalShutdownError(AbortionError):
+  """Indicates that an RPC has terminated due to local shutdown of RPCs."""
+
+
+class RemoteShutdownError(AbortionError):
+  """Indicates that an RPC has terminated due to remote shutdown of RPCs."""
+
+
+class NetworkError(AbortionError):
+  """Indicates that some error occurred on the network."""
+
+
+class LocalError(AbortionError):
+  """Indicates that an RPC has terminated due to a local defect."""
+
+
+class RemoteError(AbortionError):
+  """Indicates that an RPC has terminated due to a remote defect."""
+
+
+class RpcContext(object):
+  """Provides RPC-related information and action."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def is_active(self):
+    """Describes whether the RPC is active or has terminated."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def time_remaining(self):
+    """Describes the length of allowed time remaining for the RPC.
+
+    Returns:
+      A nonnegative float indicating the length of allowed time in seconds
+      remaining for the RPC to complete before it is considered to have timed
+      out.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def add_abortion_callback(self, abortion_callback):
+    """Registers a callback to be called if the RPC is aborted.
+
+    Args:
+      abortion_callback: A callable to be called and passed an Abortion value
+        in the event of RPC abortion.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def cancel(self):
+    """Cancels the RPC.
+
+    Idempotent and has no effect if the RPC has already terminated.
+    """
+    raise NotImplementedError()
+
+
+class Call(RpcContext):
+  """Invocation-side utility object for an RPC."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def initial_metadata(self):
+    """Accesses the initial metadata from the service-side of the RPC.
+
+    This method blocks until the value is available or is known not to have been
+    emitted from the service-side of the RPC.
+
+    Returns:
+      The initial metadata object emitted by the service-side of the RPC, or
+        None if there was no such value.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def terminal_metadata(self):
+    """Accesses the terminal metadata from the service-side of the RPC.
+
+    This method blocks until the value is available or is known not to have been
+    emitted from the service-side of the RPC.
+
+    Returns:
+      The terminal metadata object emitted by the service-side of the RPC, or
+        None if there was no such value.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def code(self):
+    """Accesses the code emitted by the service-side of the RPC.
+
+    This method blocks until the value is available or is known not to have been
+    emitted from the service-side of the RPC.
+
+    Returns:
+      The code object emitted by the service-side of the RPC, or None if there
+        was no such value.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def details(self):
+    """Accesses the details value emitted by the service-side of the RPC.
+
+    This method blocks until the value is available or is known not to have been
+    emitted from the service-side of the RPC.
+
+    Returns:
+      The details value emitted by the service-side of the RPC, or None if there
+        was no such value.
+    """
+    raise NotImplementedError()
+
+
+class ServicerContext(RpcContext):
+  """A context object passed to method implementations."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def invocation_metadata(self):
+    """Accesses the metadata from the invocation-side of the RPC.
+
+    This method blocks until the value is available or is known not to have been
+    emitted from the invocation-side of the RPC.
+
+    Returns:
+      The metadata object emitted by the invocation-side of the RPC, or None if
+        there was no such value.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def initial_metadata(self, initial_metadata):
+    """Accepts the service-side initial metadata value of the RPC.
+
+    This method need not be called by method implementations if they have no
+    service-side initial metadata to transmit.
+
+    Args:
+      initial_metadata: The service-side initial metadata value of the RPC to
+        be transmitted to the invocation side of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def terminal_metadata(self, terminal_metadata):
+    """Accepts the service-side terminal metadata value of the RPC.
+
+    This method need not be called by method implementations if they have no
+    service-side terminal metadata to transmit.
+
+    Args:
+      terminal_metadata: The service-side terminal metadata value of the RPC to
+        be transmitted to the invocation side of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def code(self, code):
+    """Accepts the service-side code of the RPC.
+
+    This method need not be called by method implementations if they have no
+    code to transmit.
+
+    Args:
+      code: The code of the RPC to be transmitted to the invocation side of the
+        RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def details(self, details):
+    """Accepts the service-side details of the RPC.
+
+    This method need not be called by method implementations if they have no
+    service-side details to transmit.
+
+    Args:
+      details: The service-side details value of the RPC to be transmitted to
+        the invocation side of the RPC.
+    """
+    raise NotImplementedError()
+
+
+class ResponseReceiver(object):
+  """Invocation-side object used to accept the output of an RPC."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def initial_metadata(self, initial_metadata):
+    """Receives the initial metadata from the service-side of the RPC.
+
+    Args:
+      initial_metadata: The initial metadata object emitted from the
+        service-side of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def response(self, response):
+    """Receives a response from the service-side of the RPC.
+
+    Args:
+      response: A response object emitted from the service-side of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def complete(self, terminal_metadata, code, details):
+    """Receives the completion values emitted from the service-side of the RPC.
+
+    Args:
+      terminal_metadata: The terminal metadata object emitted from the
+        service-side of the RPC.
+      code: The code object emitted from the service-side of the RPC.
+      details: The details object emitted from the service-side of the RPC.
+    """
+    raise NotImplementedError()
+
+
+class UnaryUnaryMultiCallable(object):
+  """Affords invoking a unary-unary RPC in any call style."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def __call__(
+      self, request, timeout, metadata=None, with_call=False):
+    """Synchronously invokes the underlying RPC.
+
+    Args:
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+      with_call: Whether or not to include return a Call for the RPC in addition
+        to the reponse.
+
+    Returns:
+      The response value for the RPC, and a Call for the RPC if with_call was
+        set to True at invocation.
+
+    Raises:
+      AbortionError: Indicating that the RPC was aborted.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def future(self, request, timeout, metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and a future.Future. In the
+        event of RPC completion, the return Future's result value will be the
+        response value of the RPC. In the event of RPC abortion, the returned
+        Future's exception value will be an AbortionError.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event(
+      self, request, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      request: The request value for the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      A Call for the RPC.
+    """
+    raise NotImplementedError()
+
+
+class UnaryStreamMultiCallable(object):
+  """Affords invoking a unary-stream RPC in any call style."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def __call__(self, request, timeout, metadata=None):
+    """Invokes the underlying RPC.
+
+    Args:
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and an iterator of response
+        values. Drawing response values from the returned iterator may raise
+        AbortionError indicating abortion of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event(
+      self, request, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      request: The request value for the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      A Call object for the RPC.
+    """
+    raise NotImplementedError()
+
+
+class StreamUnaryMultiCallable(object):
+  """Affords invoking a stream-unary RPC in any call style."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def __call__(
+      self, request_iterator, timeout, metadata=None,
+      with_call=False):
+    """Synchronously invokes the underlying RPC.
+
+    Args:
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+      with_call: Whether or not to include return a Call for the RPC in addition
+        to the reponse.
+
+    Returns:
+      The response value for the RPC, and a Call for the RPC if with_call was
+        set to True at invocation.
+
+    Raises:
+      AbortionError: Indicating that the RPC was aborted.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def future(self, request_iterator, timeout, metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and a future.Future. In the
+        event of RPC completion, the return Future's result value will be the
+        response value of the RPC. In the event of RPC abortion, the returned
+        Future's exception value will be an AbortionError.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event(
+      self, receiver, abortion_callback, timeout, metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      A single object that is both a Call object for the RPC and a
+        stream.Consumer to which the request values of the RPC should be passed.
+    """
+    raise NotImplementedError()
+
+
+class StreamStreamMultiCallable(object):
+  """Affords invoking a stream-stream RPC in any call style."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def __call__(self, request_iterator, timeout, metadata=None):
+    """Invokes the underlying RPC.
+
+    Args:
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and an iterator of response
+        values. Drawing response values from the returned iterator may raise
+        AbortionError indicating abortion of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event(
+      self, receiver, abortion_callback, timeout, metadata=None):
+    """Asynchronously invokes the underlying RPC.
+
+    Args:
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of
+        the RPC.
+
+    Returns:
+      A single object that is both a Call object for the RPC and a
+        stream.Consumer to which the request values of the RPC should be passed.
+    """
+    raise NotImplementedError()
+
+
+class MethodImplementation(object):
+  """A sum type that describes a method implementation.
+
+  Attributes:
+    cardinality: A cardinality.Cardinality value.
+    style: A style.Service value.
+    unary_unary_inline: The implementation of the method as a callable value
+      that takes a request value and a ServicerContext object and returns a
+      response value. Only non-None if cardinality is
+      cardinality.Cardinality.UNARY_UNARY and style is style.Service.INLINE.
+    unary_stream_inline: The implementation of the method as a callable value
+      that takes a request value and a ServicerContext object and returns an
+      iterator of response values. Only non-None if cardinality is
+      cardinality.Cardinality.UNARY_STREAM and style is style.Service.INLINE.
+    stream_unary_inline: The implementation of the method as a callable value
+      that takes an iterator of request values and a ServicerContext object and
+      returns a response value. Only non-None if cardinality is
+      cardinality.Cardinality.STREAM_UNARY and style is style.Service.INLINE.
+    stream_stream_inline: The implementation of the method as a callable value
+      that takes an iterator of request values and a ServicerContext object and
+      returns an iterator of response values. Only non-None if cardinality is
+      cardinality.Cardinality.STREAM_STREAM and style is style.Service.INLINE.
+    unary_unary_event: The implementation of the method as a callable value that
+      takes a request value, a response callback to which to pass the response
+      value of the RPC, and a ServicerContext. Only non-None if cardinality is
+      cardinality.Cardinality.UNARY_UNARY and style is style.Service.EVENT.
+    unary_stream_event: The implementation of the method as a callable value
+      that takes a request value, a stream.Consumer to which to pass the
+      response values of the RPC, and a ServicerContext. Only non-None if
+      cardinality is cardinality.Cardinality.UNARY_STREAM and style is
+      style.Service.EVENT.
+    stream_unary_event: The implementation of the method as a callable value
+      that takes a response callback to which to pass the response value of the
+      RPC and a ServicerContext and returns a stream.Consumer to which the
+      request values of the RPC should be passed. Only non-None if cardinality
+      is cardinality.Cardinality.STREAM_UNARY and style is style.Service.EVENT.
+    stream_stream_event: The implementation of the method as a callable value
+      that takes a stream.Consumer to which to pass the response values of the
+      RPC and a ServicerContext and returns a stream.Consumer to which the
+      request values of the RPC should be passed. Only non-None if cardinality
+      is cardinality.Cardinality.STREAM_STREAM and style is
+      style.Service.EVENT.
+  """
+  __metaclass__ = abc.ABCMeta
+
+
+class MultiMethodImplementation(object):
+  """A general type able to service many methods."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def service(self, group, method, response_consumer, context):
+    """Services an RPC.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      response_consumer: A stream.Consumer to be called to accept the response
+        values of the RPC.
+      context: a ServicerContext object.
+
+    Returns:
+      A stream.Consumer with which to accept the request values of the RPC. The
+        consumer returned from this method may or may not be invoked to
+        completion: in the case of RPC abortion, RPC Framework will simply stop
+        passing values to this object. Implementations must not assume that this
+        object will be called to completion of the request stream or even called
+        at all.
+
+    Raises:
+      abandonment.Abandoned: May or may not be raised when the RPC has been
+        aborted.
+      NoSuchMethodError: If this MultiMethod does not recognize the given group
+        and name for the RPC and is not able to service the RPC.
+    """
+    raise NotImplementedError()
+
+
+class GenericStub(object):
+  """Affords RPC invocation via generic methods."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def blocking_unary_unary(
+      self, group, method, request, timeout, metadata=None,
+      with_call=False):
+    """Invokes a unary-request-unary-response method.
+
+    This method blocks until either returning the response value of the RPC
+    (in the event of RPC completion) or raising an exception (in the event of
+    RPC abortion).
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+      with_call: Whether or not to include return a Call for the RPC in addition
+        to the reponse.
+
+    Returns:
+      The response value for the RPC, and a Call for the RPC if with_call was
+        set to True at invocation.
+
+    Raises:
+      AbortionError: Indicating that the RPC was aborted.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def future_unary_unary(
+      self, group, method, request, timeout, metadata=None):
+    """Invokes a unary-request-unary-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and a future.Future. In the
+        event of RPC completion, the return Future's result value will be the
+        response value of the RPC. In the event of RPC abortion, the returned
+        Future's exception value will be an AbortionError.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def inline_unary_stream(
+      self, group, method, request, timeout, metadata=None):
+    """Invokes a unary-request-stream-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request: The request value for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and an iterator of response
+        values. Drawing response values from the returned iterator may raise
+        AbortionError indicating abortion of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def blocking_stream_unary(
+      self, group, method, request_iterator, timeout, metadata=None,
+      with_call=False):
+    """Invokes a stream-request-unary-response method.
+
+    This method blocks until either returning the response value of the RPC
+    (in the event of RPC completion) or raising an exception (in the event of
+    RPC abortion).
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+      with_call: Whether or not to include return a Call for the RPC in addition
+        to the reponse.
+
+    Returns:
+      The response value for the RPC, and a Call for the RPC if with_call was
+        set to True at invocation.
+
+    Raises:
+      AbortionError: Indicating that the RPC was aborted.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def future_stream_unary(
+      self, group, method, request_iterator, timeout, metadata=None):
+    """Invokes a stream-request-unary-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and a future.Future. In the
+        event of RPC completion, the return Future's result value will be the
+        response value of the RPC. In the event of RPC abortion, the returned
+        Future's exception value will be an AbortionError.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def inline_stream_stream(
+      self, group, method, request_iterator, timeout, metadata=None):
+    """Invokes a stream-request-stream-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request_iterator: An iterator that yields request values for the RPC.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      An object that is both a Call for the RPC and an iterator of response
+        values. Drawing response values from the returned iterator may raise
+        AbortionError indicating abortion of the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event_unary_unary(
+      self, group, method, request, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Event-driven invocation of a unary-request-unary-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request: The request value for the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      A Call for the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event_unary_stream(
+      self, group, method, request, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Event-driven invocation of a unary-request-stream-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      request: The request value for the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      A Call for the RPC.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event_stream_unary(
+      self, group, method, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Event-driven invocation of a unary-request-unary-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      A pair of a Call object for the RPC and a stream.Consumer to which the
+        request values of the RPC should be passed.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def event_stream_stream(
+      self, group, method, receiver, abortion_callback, timeout,
+      metadata=None):
+    """Event-driven invocation of a unary-request-stream-response method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+      receiver: A ResponseReceiver to be passed the response data of the RPC.
+      abortion_callback: A callback to be called and passed an Abortion value
+        in the event of RPC abortion.
+      timeout: A duration of time in seconds to allow for the RPC.
+      metadata: A metadata value to be passed to the service-side of the RPC.
+
+    Returns:
+      A pair of a Call object for the RPC and a stream.Consumer to which the
+        request values of the RPC should be passed.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def unary_unary(self, group, method):
+    """Creates a UnaryUnaryMultiCallable for a unary-unary method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+
+    Returns:
+      A UnaryUnaryMultiCallable value for the named unary-unary method.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def unary_stream(self, group, method):
+    """Creates a UnaryStreamMultiCallable for a unary-stream method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+
+    Returns:
+      A UnaryStreamMultiCallable value for the name unary-stream method.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def stream_unary(self, group, method):
+    """Creates a StreamUnaryMultiCallable for a stream-unary method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+
+    Returns:
+      A StreamUnaryMultiCallable value for the named stream-unary method.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def stream_stream(self, group, method):
+    """Creates a StreamStreamMultiCallable for a stream-stream method.
+
+    Args:
+      group: The group identifier of the RPC.
+      method: The method identifier of the RPC.
+
+    Returns:
+      A StreamStreamMultiCallable value for the named stream-stream method.
+    """
+    raise NotImplementedError()
+
+
+class DynamicStub(object):
+  """Affords RPC invocation via attributes corresponding to afforded methods.
+
+  Instances of this type may be scoped to a single group so that attribute
+  access is unambiguous.
+
+  Instances of this type respond to attribute access as follows: if the
+  requested attribute is the name of a unary-unary method, the value of the
+  attribute will be a UnaryUnaryMultiCallable with which to invoke an RPC; if
+  the requested attribute is the name of a unary-stream method, the value of the
+  attribute will be a UnaryStreamMultiCallable with which to invoke an RPC; if
+  the requested attribute is the name of a stream-unary method, the value of the
+  attribute will be a StreamUnaryMultiCallable with which to invoke an RPC; and
+  if the requested attribute is the name of a stream-stream method, the value of
+  the attribute will be a StreamStreamMultiCallable with which to invoke an RPC.
+  """
+  __metaclass__ = abc.ABCMeta
diff --git a/src/python/grpcio/grpc/framework/interfaces/face/utilities.py b/src/python/grpcio/grpc/framework/interfaces/face/utilities.py
new file mode 100644
index 0000000000..db2ec6ed87
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/interfaces/face/utilities.py
@@ -0,0 +1,178 @@
+# 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.
+
+"""Utilities for RPC Framework's Face interface."""
+
+import collections
+
+# stream is referenced from specification in this module.
+from grpc.framework.common import cardinality
+from grpc.framework.common import style
+from grpc.framework.foundation import stream  # pylint: disable=unused-import
+from grpc.framework.interfaces.face import face
+
+
+class _MethodImplementation(
+    face.MethodImplementation,
+    collections.namedtuple(
+        '_MethodImplementation',
+        ['cardinality', 'style', 'unary_unary_inline', 'unary_stream_inline',
+         'stream_unary_inline', 'stream_stream_inline', 'unary_unary_event',
+         'unary_stream_event', 'stream_unary_event', 'stream_stream_event',])):
+  pass
+
+
+def unary_unary_inline(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a unary-unary RPC method as a callable value
+      that takes a request value and an face.ServicerContext object and
+      returns a response value.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.UNARY_UNARY, style.Service.INLINE, behavior,
+      None, None, None, None, None, None, None)
+
+
+def unary_stream_inline(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a unary-stream RPC method as a callable
+      value that takes a request value and an face.ServicerContext object and
+      returns an iterator of response values.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.UNARY_STREAM, style.Service.INLINE, None,
+      behavior, None, None, None, None, None, None)
+
+
+def stream_unary_inline(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a stream-unary RPC method as a callable
+      value that takes an iterator of request values and an
+      face.ServicerContext object and returns a response value.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.STREAM_UNARY, style.Service.INLINE, None, None,
+      behavior, None, None, None, None, None)
+
+
+def stream_stream_inline(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a stream-stream RPC method as a callable
+      value that takes an iterator of request values and an
+      face.ServicerContext object and returns an iterator of response values.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.STREAM_STREAM, style.Service.INLINE, None, None,
+      None, behavior, None, None, None, None)
+
+
+def unary_unary_event(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a unary-unary RPC method as a callable
+      value that takes a request value, a response callback to which to pass
+      the response value of the RPC, and an face.ServicerContext.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.UNARY_UNARY, style.Service.EVENT, None, None,
+      None, None, behavior, None, None, None)
+
+
+def unary_stream_event(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a unary-stream RPC method as a callable
+      value that takes a request value, a stream.Consumer to which to pass the
+      the response values of the RPC, and an face.ServicerContext.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.UNARY_STREAM, style.Service.EVENT, None, None,
+      None, None, None, behavior, None, None)
+
+
+def stream_unary_event(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a stream-unary RPC method as a callable
+      value that takes a response callback to which to pass the response value
+      of the RPC and an face.ServicerContext and returns a stream.Consumer to
+      which the request values of the RPC should be passed.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.STREAM_UNARY, style.Service.EVENT, None, None,
+      None, None, None, None, behavior, None)
+
+
+def stream_stream_event(behavior):
+  """Creates an face.MethodImplementation for the given behavior.
+
+  Args:
+    behavior: The implementation of a stream-stream RPC method as a callable
+      value that takes a stream.Consumer to which to pass the response values
+      of the RPC and an face.ServicerContext and returns a stream.Consumer to
+      which the request values of the RPC should be passed.
+
+  Returns:
+    An face.MethodImplementation derived from the given behavior.
+  """
+  return _MethodImplementation(
+      cardinality.Cardinality.STREAM_STREAM, style.Service.EVENT, None, None,
+      None, None, None, None, None, behavior)
diff --git a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py b/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
index 44fe760fbc..70149127da 100644
--- a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
+++ b/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
@@ -52,7 +52,6 @@ def wait_for_events(completion_queues, deadline):
   def set_ith_result(i, completion_queue):
     result = completion_queue.next(deadline)
     with lock:
-      print i, completion_queue, result, time.time() - deadline
       results[i] = result
   for i, completion_queue in enumerate(completion_queues):
     thread = threading.Thread(target=set_ith_result,
@@ -80,10 +79,12 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del self.client_channel
 
     self.client_completion_queue.shutdown()
-    while self.client_completion_queue.next().type != _types.EventType.QUEUE_SHUTDOWN:
+    while (self.client_completion_queue.next().type !=
+           _types.EventType.QUEUE_SHUTDOWN):
       pass
     self.server_completion_queue.shutdown()
-    while self.server_completion_queue.next().type != _types.EventType.QUEUE_SHUTDOWN:
+    while (self.server_completion_queue.next().type !=
+           _types.EventType.QUEUE_SHUTDOWN):
       pass
 
     del self.client_completion_queue
@@ -91,58 +92,68 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del self.server
 
   def testEcho(self):
-    DEADLINE = time.time()+5
-    DEADLINE_TOLERANCE = 0.25
-    CLIENT_METADATA_ASCII_KEY = 'key'
-    CLIENT_METADATA_ASCII_VALUE = 'val'
-    CLIENT_METADATA_BIN_KEY = 'key-bin'
-    CLIENT_METADATA_BIN_VALUE = b'\0'*1000
-    SERVER_INITIAL_METADATA_KEY = 'init_me_me_me'
-    SERVER_INITIAL_METADATA_VALUE = 'whodawha?'
-    SERVER_TRAILING_METADATA_KEY = 'california_is_in_a_drought'
-    SERVER_TRAILING_METADATA_VALUE = 'zomg it is'
-    SERVER_STATUS_CODE = _types.StatusCode.OK
-    SERVER_STATUS_DETAILS = 'our work is never over'
-    REQUEST = 'in death a member of project mayhem has a name'
-    RESPONSE = 'his name is robert paulson'
-    METHOD = 'twinkies'
-    HOST = 'hostess'
+    deadline = time.time() + 5
+    event_time_tolerance = 2
+    deadline_tolerance = 0.25
+    client_metadata_ascii_key = 'key'
+    client_metadata_ascii_value = 'val'
+    client_metadata_bin_key = 'key-bin'
+    client_metadata_bin_value = b'\0'*1000
+    server_initial_metadata_key = 'init_me_me_me'
+    server_initial_metadata_value = 'whodawha?'
+    server_trailing_metadata_key = 'california_is_in_a_drought'
+    server_trailing_metadata_value = 'zomg it is'
+    server_status_code = _types.StatusCode.OK
+    server_status_details = 'our work is never over'
+    request = 'blarghaflargh'
+    response = 'his name is robert paulson'
+    method = 'twinkies'
+    host = 'hostess'
     server_request_tag = object()
-    request_call_result = self.server.request_call(self.server_completion_queue, server_request_tag)
+    request_call_result = self.server.request_call(self.server_completion_queue,
+                                                   server_request_tag)
 
-    self.assertEquals(_types.CallError.OK, request_call_result)
+    self.assertEqual(_types.CallError.OK, request_call_result)
 
     client_call_tag = object()
-    client_call = self.client_channel.create_call(self.client_completion_queue, METHOD, HOST, DEADLINE)
-    client_initial_metadata = [(CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE), (CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)]
+    client_call = self.client_channel.create_call(
+        self.client_completion_queue, method, host, deadline)
+    client_initial_metadata = [
+        (client_metadata_ascii_key, client_metadata_ascii_value),
+        (client_metadata_bin_key, client_metadata_bin_value)
+    ]
     client_start_batch_result = client_call.start_batch([
         _types.OpArgs.send_initial_metadata(client_initial_metadata),
-        _types.OpArgs.send_message(REQUEST, 0),
+        _types.OpArgs.send_message(request, 0),
         _types.OpArgs.send_close_from_client(),
         _types.OpArgs.recv_initial_metadata(),
         _types.OpArgs.recv_message(),
         _types.OpArgs.recv_status_on_client()
     ], client_call_tag)
-    self.assertEquals(_types.CallError.OK, client_start_batch_result)
+    self.assertEqual(_types.CallError.OK, client_start_batch_result)
 
-    client_no_event, request_event, = wait_for_events([self.client_completion_queue, self.server_completion_queue], time.time() + 2)
-    self.assertEquals(client_no_event, None)
-    self.assertEquals(_types.EventType.OP_COMPLETE, request_event.type)
+    client_no_event, request_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+    self.assertEqual(client_no_event, None)
+    self.assertEqual(_types.EventType.OP_COMPLETE, request_event.type)
     self.assertIsInstance(request_event.call, _low.Call)
     self.assertIs(server_request_tag, request_event.tag)
-    self.assertEquals(1, len(request_event.results))
+    self.assertEqual(1, len(request_event.results))
     received_initial_metadata = dict(request_event.results[0].initial_metadata)
     # Check that our metadata were transmitted
-    self.assertEquals(
+    self.assertEqual(
         dict(client_initial_metadata),
-        dict((x, received_initial_metadata[x]) for x in zip(*client_initial_metadata)[0]))
+        dict((x, received_initial_metadata[x])
+             for x in zip(*client_initial_metadata)[0]))
     # Check that Python's user agent string is a part of the full user agent
     # string
     self.assertIn('Python-gRPC-{}'.format(_grpcio_metadata.__version__),
                   received_initial_metadata['user-agent'])
-    self.assertEquals(METHOD, request_event.call_details.method)
-    self.assertEquals(HOST, request_event.call_details.host)
-    self.assertLess(abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE)
+    self.assertEqual(method, request_event.call_details.method)
+    self.assertEqual(host, request_event.call_details.host)
+    self.assertLess(abs(deadline - request_event.call_details.deadline),
+                    deadline_tolerance)
 
     # Check that the channel is connected, and that both it and the call have
     # the proper target and peer; do this after the first flurry of messages to
@@ -155,33 +166,43 @@ class InsecureServerInsecureClient(unittest.TestCase):
 
     server_call_tag = object()
     server_call = request_event.call
-    server_initial_metadata = [(SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE)]
-    server_trailing_metadata = [(SERVER_TRAILING_METADATA_KEY, SERVER_TRAILING_METADATA_VALUE)]
+    server_initial_metadata = [
+        (server_initial_metadata_key, server_initial_metadata_value)
+    ]
+    server_trailing_metadata = [
+        (server_trailing_metadata_key, server_trailing_metadata_value)
+    ]
     server_start_batch_result = server_call.start_batch([
         _types.OpArgs.send_initial_metadata(server_initial_metadata),
         _types.OpArgs.recv_message(),
-        _types.OpArgs.send_message(RESPONSE, 0),
+        _types.OpArgs.send_message(response, 0),
         _types.OpArgs.recv_close_on_server(),
-        _types.OpArgs.send_status_from_server(server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS)
+        _types.OpArgs.send_status_from_server(
+            server_trailing_metadata, server_status_code, server_status_details)
     ], server_call_tag)
-    self.assertEquals(_types.CallError.OK, server_start_batch_result)
+    self.assertEqual(_types.CallError.OK, server_start_batch_result)
 
-    client_event, server_event, = wait_for_events([self.client_completion_queue, self.server_completion_queue], time.time() + 1)
+    client_event, server_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
 
-    self.assertEquals(6, len(client_event.results))
+    self.assertEqual(6, len(client_event.results))
     found_client_op_types = set()
     for client_result in client_event.results:
-      self.assertNotIn(client_result.type, found_client_op_types)  # we expect each op type to be unique
+      # we expect each op type to be unique
+      self.assertNotIn(client_result.type, found_client_op_types)
       found_client_op_types.add(client_result.type)
       if client_result.type == _types.OpType.RECV_INITIAL_METADATA:
-        self.assertEquals(dict(server_initial_metadata), dict(client_result.initial_metadata))
+        self.assertEqual(dict(server_initial_metadata),
+                         dict(client_result.initial_metadata))
       elif client_result.type == _types.OpType.RECV_MESSAGE:
-        self.assertEquals(RESPONSE, client_result.message)
+        self.assertEqual(response, client_result.message)
       elif client_result.type == _types.OpType.RECV_STATUS_ON_CLIENT:
-        self.assertEquals(dict(server_trailing_metadata), dict(client_result.trailing_metadata))
-        self.assertEquals(SERVER_STATUS_DETAILS, client_result.status.details)
-        self.assertEquals(SERVER_STATUS_CODE, client_result.status.code)
-    self.assertEquals(set([
+        self.assertEqual(dict(server_trailing_metadata),
+                         dict(client_result.trailing_metadata))
+        self.assertEqual(server_status_details, client_result.status.details)
+        self.assertEqual(server_status_code, client_result.status.code)
+    self.assertEqual(set([
           _types.OpType.SEND_INITIAL_METADATA,
           _types.OpType.SEND_MESSAGE,
           _types.OpType.SEND_CLOSE_FROM_CLIENT,
@@ -190,16 +211,16 @@ class InsecureServerInsecureClient(unittest.TestCase):
           _types.OpType.RECV_STATUS_ON_CLIENT
       ]), found_client_op_types)
 
-    self.assertEquals(5, len(server_event.results))
+    self.assertEqual(5, len(server_event.results))
     found_server_op_types = set()
     for server_result in server_event.results:
       self.assertNotIn(client_result.type, found_server_op_types)
       found_server_op_types.add(server_result.type)
       if server_result.type == _types.OpType.RECV_MESSAGE:
-        self.assertEquals(REQUEST, server_result.message)
+        self.assertEqual(request, server_result.message)
       elif server_result.type == _types.OpType.RECV_CLOSE_ON_SERVER:
         self.assertFalse(server_result.cancelled)
-    self.assertEquals(set([
+    self.assertEqual(set([
           _types.OpType.SEND_INITIAL_METADATA,
           _types.OpType.RECV_MESSAGE,
           _types.OpType.SEND_MESSAGE,
@@ -211,5 +232,81 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del server_call
 
 
+class HangingServerShutdown(unittest.TestCase):
+
+  def setUp(self):
+    self.server_completion_queue = _low.CompletionQueue()
+    self.server = _low.Server(self.server_completion_queue, [])
+    self.port = self.server.add_http2_port('[::]:0')
+    self.client_completion_queue = _low.CompletionQueue()
+    self.client_channel = _low.Channel('localhost:%d'%self.port, [])
+
+    self.server.start()
+
+  def tearDown(self):
+    self.server.shutdown()
+    del self.client_channel
+
+    self.client_completion_queue.shutdown()
+    self.server_completion_queue.shutdown()
+    while True:
+      client_event, server_event = wait_for_events(
+          [self.client_completion_queue, self.server_completion_queue],
+          float("+inf"))
+      if (client_event.type == _types.EventType.QUEUE_SHUTDOWN and
+          server_event.type == _types.EventType.QUEUE_SHUTDOWN):
+        break
+
+    del self.client_completion_queue
+    del self.server_completion_queue
+    del self.server
+
+  def testHangingServerCall(self):
+    deadline = time.time() + 5
+    deadline_tolerance = 0.25
+    event_time_tolerance = 2
+    cancel_all_calls_time_tolerance = 0.5
+    request = 'blarghaflargh'
+    method = 'twinkies'
+    host = 'hostess'
+    server_request_tag = object()
+    request_call_result = self.server.request_call(self.server_completion_queue,
+                                                   server_request_tag)
+
+    client_call_tag = object()
+    client_call = self.client_channel.create_call(self.client_completion_queue,
+                                                  method, host, deadline)
+    client_start_batch_result = client_call.start_batch([
+        _types.OpArgs.send_initial_metadata([]),
+        _types.OpArgs.send_message(request, 0),
+        _types.OpArgs.send_close_from_client(),
+        _types.OpArgs.recv_initial_metadata(),
+        _types.OpArgs.recv_message(),
+        _types.OpArgs.recv_status_on_client()
+    ], client_call_tag)
+
+    client_no_event, request_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+
+    # Now try to shutdown the server and expect that we see server shutdown
+    # almost immediately after calling cancel_all_calls.
+    with self.assertRaises(RuntimeError):
+      self.server.cancel_all_calls()
+    shutdown_tag = object()
+    self.server.shutdown(shutdown_tag)
+    pre_cancel_timestamp = time.time()
+    self.server.cancel_all_calls()
+    finish_shutdown_timestamp = None
+    client_call_event, server_shutdown_event = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+    self.assertIs(shutdown_tag, server_shutdown_event.tag)
+    self.assertGreater(pre_cancel_timestamp + cancel_all_calls_time_tolerance,
+                       time.time())
+
+    del client_call
+
+
 if __name__ == '__main__':
   unittest.main(verbosity=2)