Merge github.com:grpc/grpc into grand-unified-closures

36 files changed, 544 insertions, 294 deletions

diff --git a/composer.json b/composer.json
index 6e7f24b451..0ebe0a1108 100644
--- a/composer.json
+++ b/composer.json
@@ -5,15 +5,9 @@
   "keywords": ["rpc"],
   "homepage": "http://grpc.io",
   "license": "BSD-3-Clause",
-  "repositories": [
-    {
-      "type": "vcs",
-      "url": "https://github.com/stanley-cheung/Protobuf-PHP"
-    }
-  ],
   "require": {
     "php": ">=5.5.0",
-    "datto/protobuf-php": "dev-master"
+    "stanley-cheung/protobuf-php": "dev-master"
   },
   "require-dev": {
     "google/auth": "v0.9"
diff --git a/examples/php/README.md b/examples/php/README.md
index 6889a6cb7e..54cc97d8c2 100644
--- a/examples/php/README.md
+++ b/examples/php/README.md
@@ -11,7 +11,7 @@ INSTALL
  - Install the gRPC PHP extension
 
    ```sh
-   $ [sudo] pecl install grpc-beta
+   $ [sudo] pecl install grpc
    ```
 
  - Clone this repository
diff --git a/examples/php/composer.json b/examples/php/composer.json
index 950e11367d..a8b790b1de 100644
--- a/examples/php/composer.json
+++ b/examples/php/composer.json
@@ -2,12 +2,6 @@
   "name": "grpc/grpc-demo",
   "description": "gRPC example for PHP",
   "minimum-stability": "dev",
-  "repositories": [
-    {
-      "type": "vcs",
-      "url": "https://github.com/stanley-cheung/Protobuf-PHP"
-    }
-  ],
   "require": {
     "grpc/grpc": "v0.15.0"
   }
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index 7a8858ef19..6876961e21 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -179,10 +179,13 @@ class Server GRPC_FINAL : public ServerInterface, private GrpcLibraryCodegen {
   grpc::mutex mu_;
   bool started_;
   bool shutdown_;
+  bool shutdown_notified_;
   // The number of threads which are running callbacks.
   int num_running_cb_;
   grpc::condition_variable callback_cv_;
 
+  grpc::condition_variable shutdown_cv_;
+
   std::shared_ptr<GlobalCallbacks> global_callbacks_;
 
   std::list<SyncRequest>* sync_methods_;
diff --git a/package.xml b/package.xml
index e6c7970ed1..22e523bce6 100644
--- a/package.xml
+++ b/package.xml
@@ -10,18 +10,19 @@
   <email>grpc-packages@google.com</email>
   <active>yes</active>
  </lead>
- <date>2016-06-30</date>
+ <date>2016-07-13</date>
  <time>16:06:07</time>
  <version>
   <release>1.1.0</release>
   <api>1.1.0</api>
  </version>
  <stability>
-  <release>beta</release>
-  <api>beta</api>
+  <release>stable</release>
+  <api>stable</api>
  </stability>
  <license>BSD</license>
  <notes>
+- GA release
 - Fix shutdown hang problem #4017
  </notes>
  <contents>
@@ -1090,16 +1091,17 @@ Update to wrap gRPC C Core version 0.10.0
   </release>
   <release>
    <version>
-    <release>0.15.1</release>
-    <api>0.15.1</api>
+    <release>1.0.0</release>
+    <api>1.0.0</api>
    </version>
    <stability>
-    <release>beta</release>
-    <api>beta</api>
+    <release>stable</release>
+    <api>stable</api>
    </stability>
-   <date>2016-06-30</date>
+   <date>2016-07-13</date>
    <license>BSD</license>
    <notes>
+- GA release
 - Fix shutdown hang problem #4017
    </notes>
   </release>
diff --git a/setup.py b/setup.py
index de346a2150..6dbc169053 100644
--- a/setup.py
+++ b/setup.py
@@ -84,9 +84,40 @@ ENABLE_CYTHON_TRACING = os.environ.get(
 # entirely ignored/dropped/forgotten by distutils and its Cygwin/MinGW support.
 # We use these environment variables to thus get around that without locking
 # ourselves in w.r.t. the multitude of operating systems this ought to build on.
-# By default we assume a GCC-like compiler.
-EXTRA_COMPILE_ARGS = shlex.split(os.environ.get('GRPC_PYTHON_CFLAGS', ''))
-EXTRA_LINK_ARGS = shlex.split(os.environ.get('GRPC_PYTHON_LDFLAGS', ''))
+# We can also use these variables as a way to inject environment-specific
+# compiler/linker flags. We assume GCC-like compilers and/or MinGW as a
+# reasonable default.
+EXTRA_ENV_COMPILE_ARGS = os.environ.get('GRPC_PYTHON_CFLAGS', None)
+EXTRA_ENV_LINK_ARGS = os.environ.get('GRPC_PYTHON_LDFLAGS', None)
+if EXTRA_ENV_COMPILE_ARGS is None:
+  EXTRA_ENV_COMPILE_ARGS = '-fno-wrapv'
+  if 'win32' in sys.platform:
+    # We use define flags here and don't directly add to DEFINE_MACROS below to
+    # ensure that the expert user/builder has a way of turning it off (via the
+    # envvars) without adding yet more GRPC-specific envvars.
+    # See https://sourceforge.net/p/mingw-w64/bugs/363/
+    if '32' in platform.architecture()[0]:
+      EXTRA_ENV_COMPILE_ARGS += ' -D_ftime=_ftime32 -D_timeb=__timeb32 -D_ftime_s=_ftime32_s'
+    else:
+      EXTRA_ENV_COMPILE_ARGS += ' -D_ftime=_ftime64 -D_timeb=__timeb64'
+  elif "linux" in sys.platform or "darwin" in sys.platform:
+    EXTRA_ENV_COMPILE_ARGS += ' -fvisibility=hidden'
+if EXTRA_ENV_LINK_ARGS is None:
+  EXTRA_ENV_LINK_ARGS = '-lpthread'
+  if 'win32' in sys.platform:
+    # TODO(atash) check if this is actually safe to just import and call on
+    # non-Windows (to avoid breaking import style)
+    from distutils.cygwinccompiler import get_msvcr
+    msvcr = get_msvcr()[0]
+    # TODO(atash) sift through the GCC specs to see if libstdc++ can have any
+    # influence on the linkage outcome on MinGW for non-C++ programs.
+    EXTRA_ENV_LINK_ARGS += (
+        ' -static-libgcc -static-libstdc++ -mcrtdll={msvcr} '
+        '-static'.format(msvcr=msvcr))
+  elif "linux" in sys.platform:
+    EXTRA_ENV_LINK_ARGS += ' -Wl,-wrap,memcpy'
+EXTRA_COMPILE_ARGS = shlex.split(EXTRA_ENV_COMPILE_ARGS)
+EXTRA_LINK_ARGS = shlex.split(EXTRA_ENV_LINK_ARGS)
 
 CYTHON_EXTENSION_PACKAGE_NAMES = ()
 
@@ -118,13 +149,8 @@ if "win32" in sys.platform:
 
 LDFLAGS = tuple(EXTRA_LINK_ARGS)
 CFLAGS = tuple(EXTRA_COMPILE_ARGS)
-if "linux" in sys.platform:
-  LDFLAGS += ('-Wl,-wrap,memcpy',)
 if "linux" in sys.platform or "darwin" in sys.platform:
-  CFLAGS += ('-fvisibility=hidden',)
-
   pymodinit_type = 'PyObject*' if PY3 else 'void'
-
   pymodinit = '__attribute__((visibility ("default"))) {}'.format(pymodinit_type)
   DEFINE_MACROS += (('PyMODINIT_FUNC', pymodinit),)
 
@@ -139,8 +165,13 @@ if 'darwin' in sys.platform and PY3:
     os.environ['MACOSX_DEPLOYMENT_TARGET'] = '10.7'
 
 
-def cython_extensions(module_names, extra_sources, include_dirs,
-                      libraries, define_macros, build_with_cython=False):
+def cython_extensions():
+  module_names = list(CYTHON_EXTENSION_MODULE_NAMES)
+  extra_sources = list(CYTHON_HELPER_C_FILES) + list(CORE_C_FILES)
+  include_dirs = list(EXTENSION_INCLUDE_DIRECTORIES)
+  libraries = list(EXTENSION_LIBRARIES)
+  define_macros = list(DEFINE_MACROS)
+  build_with_cython = bool(BUILD_WITH_CYTHON)
   # Set compiler directives linetrace argument only if we care about tracing;
   # this is due to Cython having different behavior between linetrace being
   # False and linetrace being unset. See issue #5689.
@@ -181,11 +212,7 @@ def cython_extensions(module_names, extra_sources, include_dirs,
   else:
     return extensions
 
-CYTHON_EXTENSION_MODULES = cython_extensions(
-    list(CYTHON_EXTENSION_MODULE_NAMES),
-    list(CYTHON_HELPER_C_FILES) + list(CORE_C_FILES),
-    list(EXTENSION_INCLUDE_DIRECTORIES), list(EXTENSION_LIBRARIES),
-    list(DEFINE_MACROS), bool(BUILD_WITH_CYTHON))
+CYTHON_EXTENSION_MODULES = cython_extensions()
 
 PACKAGE_DIRECTORIES = {
     '': PYTHON_STEM,
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 23389047e4..98b01426a4 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -697,6 +697,25 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
                                 grpc_chttp2_transport_global *transport_global,
                                 bool covered_by_poller, const char *reason) {
   GPR_TIMER_BEGIN("grpc_chttp2_initiate_write", 0);
+
+  /* Perform state checks, and transition to a scheduled state if appropriate.
+     If we are inactive, schedule a write chain to begin once the transport
+     combiner finishes any executions in its current batch (which may be
+     scheduled AFTER this code executes). The write chain will:
+      - call start_writing, which verifies (under the global lock) that there
+        are things that need to be written by calling
+        grpc_chttp2_unlocking_check_writes, and if so schedules writing_action
+        against the current exec_ctx, to be executed OUTSIDE of the global lock
+      - eventually writing_action results in grpc_chttp2_terminate_writing being
+        called, which re-takes the global lock, updates state, checks if we need
+        to do *another* write immediately, and if so loops back to
+        start_writing.
+
+     Current problems:
+       - too much lock entry/exiting
+       - the writing thread can become stuck indefinitely (punt through the
+         workqueue periodically to fix) */
+
   grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
   switch (t->executor.write_state) {
     case GRPC_CHTTP2_WRITES_CORKED:
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index 066bca354d..b72cd61fcf 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -357,7 +357,8 @@ struct grpc_chttp2_transport {
   /** global state for reading/writing */
   grpc_chttp2_transport_global global;
   /** state only accessible by the chain of execution that
-      set writing_active=1 */
+      set writing_state >= GRPC_WRITING, and only by the writing closure
+      chain. */
   grpc_chttp2_transport_writing writing;
   /** state only accessible by the chain of execution that
       set parsing_active=1 */
@@ -551,7 +552,10 @@ struct grpc_chttp2_stream {
     to write.
     The global lock is dropped and we do the syscall to write.
     After writing, a follow-up check is made to see if another round of writing
-    should be performed. */
+    should be performed.
+
+    The actual call chain is documented in the implementation of this function.
+    */
 void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
                                 grpc_chttp2_transport_global *transport_global,
                                 bool covered_by_poller, const char *reason);
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 4282d01a2b..6a63c4d1d1 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -517,14 +517,10 @@ static polling_island *polling_island_create(grpc_exec_ctx *exec_ctx,
 
 done:
   if (*error != GRPC_ERROR_NONE) {
-    if (pi->epoll_fd < 0) {
-      close(pi->epoll_fd);
-    }
     if (pi->workqueue != NULL) {
       GRPC_WORKQUEUE_UNREF(exec_ctx, pi->workqueue, "polling_island");
     }
-    gpr_mu_destroy(&pi->mu);
-    gpr_free(pi);
+    polling_island_delete(exec_ctx, pi);
     pi = NULL;
   }
   return pi;
@@ -533,9 +529,9 @@ done:
 static void polling_island_delete(grpc_exec_ctx *exec_ctx, polling_island *pi) {
   GPR_ASSERT(pi->fd_cnt == 0);
 
-  gpr_atm_rel_store(&pi->merged_to, (gpr_atm)NULL);
-
-  close(pi->epoll_fd);
+  if (pi->epoll_fd >= 0) {
+    close(pi->epoll_fd);
+  }
   gpr_mu_destroy(&pi->mu);
   gpr_free(pi->fds);
   gpr_free(pi);
@@ -936,6 +932,10 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
   gpr_mu_unlock(&fd->mu);
   UNREF_BY(fd, 2, reason); /* Drop the reference */
   if (unref_pi != NULL) {
+    /* Unref stale polling island here, outside the fd lock above.
+       The polling island owns a workqueue which owns an fd, and unreffing
+       inside the lock can cause an eventual lock loop that makes TSAN very
+       unhappy. */
     PI_UNREF(exec_ctx, unref_pi, "fd_orphan");
   }
   GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error));
@@ -1559,9 +1559,19 @@ retry:
   if (fd->polling_island == pollset->polling_island) {
     pi_new = fd->polling_island;
     if (pi_new == NULL) {
+      /* Unlock before creating a new polling island: the polling island will
+         create a workqueue which creates a file descriptor, and holding an fd
+         lock here can eventually cause a loop to appear to TSAN (making it
+         unhappy). We don't think it's a real loop (there's an epoch point where
+         that loop possibility disappears), but the advantages of keeping TSAN
+         happy outweigh any performance advantage we might have by keeping the
+         lock held. */
       gpr_mu_unlock(&fd->mu);
       pi_new = polling_island_create(exec_ctx, fd, &error);
       gpr_mu_lock(&fd->mu);
+      /* Need to reverify any assumptions made between the initial lock and
+         getting to this branch: if they've changed, we need to throw away our
+         work and figure things out again. */
       if (fd->polling_island != NULL) {
         GRPC_POLLING_TRACE(
             "pollset_add_fd: Raced creating new polling island. pi_new: %p "
diff --git a/src/core/lib/iomgr/iomgr.c b/src/core/lib/iomgr/iomgr.c
index 89292a153e..d67d388b8c 100644
--- a/src/core/lib/iomgr/iomgr.c
+++ b/src/core/lib/iomgr/iomgr.c
@@ -45,6 +45,7 @@
 
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/iomgr/iomgr_internal.h"
+#include "src/core/lib/iomgr/network_status_tracker.h"
 #include "src/core/lib/iomgr/timer.h"
 #include "src/core/lib/support/env.h"
 #include "src/core/lib/support/string.h"
@@ -62,6 +63,7 @@ void grpc_iomgr_init(void) {
   grpc_timer_list_init(gpr_now(GPR_CLOCK_MONOTONIC));
   g_root_object.next = g_root_object.prev = &g_root_object;
   g_root_object.name = "root";
+  grpc_network_status_init();
   grpc_iomgr_platform_init();
 }
 
@@ -140,6 +142,7 @@ void grpc_iomgr_shutdown(void) {
 
   grpc_iomgr_platform_shutdown();
   grpc_exec_ctx_global_shutdown();
+  grpc_network_status_shutdown();
   gpr_mu_destroy(&g_mu);
   gpr_cv_destroy(&g_rcv);
 }
diff --git a/src/core/lib/iomgr/network_status_tracker.c b/src/core/lib/iomgr/network_status_tracker.c
index ccbe136db9..90c074b007 100644
--- a/src/core/lib/iomgr/network_status_tracker.c
+++ b/src/core/lib/iomgr/network_status_tracker.c
@@ -42,9 +42,8 @@ typedef struct endpoint_ll_node {
 
 static endpoint_ll_node *head = NULL;
 static gpr_mu g_endpoint_mutex;
-static gpr_once g_once_init = GPR_ONCE_INIT;
 
-static void destroy_network_status_monitor(void) {
+void grpc_network_status_shutdown(void) {
   if (head != NULL) {
     gpr_log(GPR_ERROR,
             "Memory leaked as all network endpoints were not shut down");
@@ -52,14 +51,12 @@ static void destroy_network_status_monitor(void) {
   gpr_mu_destroy(&g_endpoint_mutex);
 }
 
-static void initialize_network_status_monitor(void) {
+void grpc_network_status_init(void) {
   gpr_mu_init(&g_endpoint_mutex);
-  atexit(destroy_network_status_monitor);
   // TODO(makarandd): Install callback with OS to monitor network status.
 }
 
 void grpc_network_status_register_endpoint(grpc_endpoint *ep) {
-  gpr_once_init(&g_once_init, initialize_network_status_monitor);
   gpr_mu_lock(&g_endpoint_mutex);
   if (head == NULL) {
     head = (endpoint_ll_node *)gpr_malloc(sizeof(endpoint_ll_node));
diff --git a/src/core/lib/iomgr/network_status_tracker.h b/src/core/lib/iomgr/network_status_tracker.h
index 74a1aa8135..67cb645f44 100644
--- a/src/core/lib/iomgr/network_status_tracker.h
+++ b/src/core/lib/iomgr/network_status_tracker.h
@@ -35,7 +35,11 @@
 #define GRPC_CORE_LIB_IOMGR_NETWORK_STATUS_TRACKER_H
 #include "src/core/lib/iomgr/endpoint.h"
 
+void grpc_network_status_init(void);
+void grpc_network_status_shutdown(void);
+
 void grpc_network_status_register_endpoint(grpc_endpoint *ep);
 void grpc_network_status_unregister_endpoint(grpc_endpoint *ep);
 void grpc_network_status_shutdown_all_endpoints();
+
 #endif /* GRPC_CORE_LIB_IOMGR_NETWORK_STATUS_TRACKER_H */
diff --git a/src/core/lib/iomgr/workqueue.h b/src/core/lib/iomgr/workqueue.h
index 416618e258..b2805dc66c 100644
--- a/src/core/lib/iomgr/workqueue.h
+++ b/src/core/lib/iomgr/workqueue.h
@@ -50,12 +50,20 @@
 
 /* grpc_workqueue is forward declared in exec_ctx.h */
 
+/* Reference counting functions. Use the macro's always
+   (GRPC_WORKQUEUE_{REF,UNREF}).
+
+   Pass in a descriptive reason string for reffing/unreffing as the last
+   argument to each macro. When GRPC_WORKQUEUE_REFCOUNT_DEBUG is defined, that
+   string will be printed alongside the refcount. When it is not defined, the
+   string will be discarded at compilation time. */
+
 //#define GRPC_WORKQUEUE_REFCOUNT_DEBUG
 #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
 #define GRPC_WORKQUEUE_REF(p, r) \
   (grpc_workqueue_ref((p), __FILE__, __LINE__, (r)), (p))
-#define GRPC_WORKQUEUE_UNREF(cl, p, r) \
-  grpc_workqueue_unref((cl), (p), __FILE__, __LINE__, (r))
+#define GRPC_WORKQUEUE_UNREF(exec_ctx, p, r) \
+  grpc_workqueue_unref((exec_ctx), (p), __FILE__, __LINE__, (r))
 void grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line,
                         const char *reason);
 void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue,
@@ -67,7 +75,16 @@ void grpc_workqueue_ref(grpc_workqueue *workqueue);
 void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue);
 #endif
 
-/** Add a work item to a workqueue */
+/** Add a work item to a workqueue. Items added to a work queue will be started
+    in approximately the order they were enqueued, on some thread that may or
+    may not be the current thread. Successive closures enqueued onto a workqueue
+    MAY be executed concurrently.
+
+    It is generally more expensive to add a closure to a workqueue than to the
+    execution context, both in terms of CPU work and in execution latency.
+
+    Use work queues when it's important that other threads be given a chance to
+    tackle some workload. */
 void grpc_workqueue_enqueue(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue,
                             grpc_closure *closure, grpc_error *error);
 
diff --git a/src/core/lib/iomgr/workqueue_posix.h b/src/core/lib/iomgr/workqueue_posix.h
index c69ae8a941..03ee21cef7 100644
--- a/src/core/lib/iomgr/workqueue_posix.h
+++ b/src/core/lib/iomgr/workqueue_posix.h
@@ -53,7 +53,8 @@ struct grpc_workqueue {
   grpc_closure read_closure;
 };
 
-/** Create a work queue */
+/** Create a work queue. Returns an error if creation fails. If creation
+    succeeds, sets *workqueue to point to it. */
 grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx,
                                   grpc_workqueue **workqueue);
 
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index 2641604901..e0b5ed6297 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -207,7 +207,7 @@ struct grpc_server {
   registered_method *registered_methods;
   /** one request matcher for unregistered methods */
   request_matcher unregistered_request_matcher;
-  /** free list of available requested_calls indices */
+  /** free list of available requested_calls_per_cq indices */
   gpr_stack_lockfree **request_freelist_per_cq;
   /** requested call backing data */
   requested_call **requested_calls_per_cq;
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index fb4c68ebe4..af04fd4ca6 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -281,6 +281,7 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
     : max_message_size_(max_message_size),
       started_(false),
       shutdown_(false),
+      shutdown_notified_(false),
       num_running_cb_(0),
       sync_methods_(new std::list<SyncRequest>),
       has_generic_service_(false),
@@ -462,13 +463,16 @@ void Server::ShutdownInternal(gpr_timespec deadline) {
     while (num_running_cb_ != 0) {
       callback_cv_.wait(lock);
     }
+
+    shutdown_notified_ = true;
+    shutdown_cv_.notify_all();
   }
 }
 
 void Server::Wait() {
   grpc::unique_lock<grpc::mutex> lock(mu_);
-  while (num_running_cb_ != 0) {
-    callback_cv_.wait(lock);
+  while (started_ && !shutdown_notified_) {
+    shutdown_cv_.wait(lock);
   }
 }
 
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index c670ea65c7..3d0947c03d 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -253,8 +253,9 @@ GPR_EXPORT intptr_t GPR_CALLTYPE grpcsharp_batch_context_recv_message_length(
   if (!ctx->recv_message) {
     return -1;
   }
-  /* TODO(issue:#7206): check return value of grpc_byte_buffer_reader_init. */
-  grpc_byte_buffer_reader_init(&reader, ctx->recv_message);
+
+  GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, ctx->recv_message));
+
   return (intptr_t)grpc_byte_buffer_length(reader.buffer_out);
 }
 
@@ -268,8 +269,7 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_batch_context_recv_message_to_buffer(
   gpr_slice slice;
   size_t offset = 0;
 
-  /* TODO(issue:#7206): check return value of grpc_byte_buffer_reader_init. */
-  grpc_byte_buffer_reader_init(&reader, ctx->recv_message);
+  GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, ctx->recv_message));
 
   while (grpc_byte_buffer_reader_next(&reader, &slice)) {
     size_t len = GPR_SLICE_LENGTH(slice);
diff --git a/src/php/README.md b/src/php/README.md
index 6cc1ba4d46..8abedc40a3 100644
--- a/src/php/README.md
+++ b/src/php/README.md
@@ -5,7 +5,7 @@ This directory contains source code for PHP implementation of gRPC layered on sh
 
 #Status
 
-Beta
+GA
 
 ## Environment
 
@@ -43,7 +43,7 @@ $ sudo mv phpunit-old.phar /usr/bin/phpunit
 Install the gRPC PHP extension
 
 ```sh
-sudo pecl install grpc-beta
+sudo pecl install grpc
 ```
 
 This will compile and install the gRPC PHP extension into the standard PHP extension directory. You should be able to run the [unit tests](#unit-tests), with the PHP extension installed.
@@ -75,7 +75,7 @@ $ sudo make install
 Install the gRPC PHP extension from PECL
 
 ```sh
-$ sudo pecl install grpc-beta
+$ sudo pecl install grpc
 ```
 
 Or, compile from source
@@ -148,7 +148,7 @@ Alternatively, you can download `protoc` binaries from [the protocol buffers Git
 You need to install `protoc-gen-php` to generate stub class `.php` files from service definition `.proto` files.
 
 ```sh
-$ cd grpc/src/php/vendor/datto/protobuf-php # if you had run `composer install` in the previous step
+$ cd grpc/src/php/vendor/stanley-cheung/protobuf-php # if you had run `composer install` in the previous step
 
 OR
 
diff --git a/src/php/composer.json b/src/php/composer.json
index 2ad73223c6..ca8e085fca 100644
--- a/src/php/composer.json
+++ b/src/php/composer.json
@@ -5,15 +5,10 @@
   "keywords": ["rpc"],
   "homepage": "http://grpc.io",
   "license": "BSD-3-Clause",
-  "repositories": [
-    {
-      "type": "vcs",
-      "url": "https://github.com/stanley-cheung/Protobuf-PHP"
-    }
-  ],
+  "version": "1.0.0",
   "require": {
     "php": ">=5.5.0",
-    "datto/protobuf-php": "dev-master",
+    "stanley-cheung/protobuf-php": "dev-master",
     "google/auth": "v0.7"
   },
   "autoload": {
diff --git a/src/proto/grpc/testing/control.proto b/src/proto/grpc/testing/control.proto
index 20496a8116..ece6910815 100644
--- a/src/proto/grpc/testing/control.proto
+++ b/src/proto/grpc/testing/control.proto
@@ -229,4 +229,7 @@ message ScenarioResult {
   repeated int32 server_cores = 5;
   // An after-the-fact computed summary
   ScenarioResultSummary summary = 6;
+  // Information on success or failure of each worker
+  repeated bool client_success = 7;
+  repeated bool server_success = 8;
 }
diff --git a/src/python/grpcio_tests/tests/unit/_exit_test.py b/src/python/grpcio_tests/tests/unit/_exit_test.py
index b0d6af73e5..5a4a32887c 100644
--- a/src/python/grpcio_tests/tests/unit/_exit_test.py
+++ b/src/python/grpcio_tests/tests/unit/_exit_test.py
@@ -84,6 +84,7 @@ def wait(process):
   process.wait()
 
 
+@unittest.skip('https://github.com/grpc/grpc/issues/7311')
 class ExitTest(unittest.TestCase):
 
   def test_unstarted_server(self):
diff --git a/src/python/grpcio_tests/tests/unit/_rpc_test.py b/src/python/grpcio_tests/tests/unit/_rpc_test.py
index 59bf240d28..ab6546bf87 100644
--- a/src/python/grpcio_tests/tests/unit/_rpc_test.py
+++ b/src/python/grpcio_tests/tests/unit/_rpc_test.py
@@ -233,7 +233,11 @@ class RPCTest(unittest.TestCase):
             ('test', 'SuccessfulUnaryRequestFutureUnaryResponse'),))
     response = response_future.result()
 
+    self.assertIsInstance(response_future, grpc.Future)
+    self.assertIsInstance(response_future, grpc.Call)
     self.assertEqual(expected_response, response)
+    self.assertIsNone(response_future.exception())
+    self.assertIsNone(response_future.traceback())
 
   def testSuccessfulUnaryRequestStreamResponse(self):
     request = b'\x37\x58'
@@ -287,6 +291,8 @@ class RPCTest(unittest.TestCase):
     response = response_future.result()
 
     self.assertEqual(expected_response, response)
+    self.assertIsNone(response_future.exception())
+    self.assertIsNone(response_future.traceback())
 
   def testSuccessfulStreamRequestStreamResponse(self):
     requests = tuple(b'\x77\x58' for _ in range(test_constants.STREAM_LENGTH))
@@ -459,6 +465,10 @@ class RPCTest(unittest.TestCase):
     self.assertTrue(response_future.cancelled())
     with self.assertRaises(grpc.FutureCancelledError):
       response_future.result()
+    with self.assertRaises(grpc.FutureCancelledError):
+      response_future.exception()
+    with self.assertRaises(grpc.FutureCancelledError):
+      response_future.traceback()
     self.assertIs(grpc.StatusCode.CANCELLED, response_future.code())
 
   def testCancelledUnaryRequestStreamResponse(self):
@@ -495,6 +505,10 @@ class RPCTest(unittest.TestCase):
     self.assertTrue(response_future.cancelled())
     with self.assertRaises(grpc.FutureCancelledError):
       response_future.result()
+    with self.assertRaises(grpc.FutureCancelledError):
+      response_future.exception()
+    with self.assertRaises(grpc.FutureCancelledError):
+      response_future.traceback()
     self.assertIsNotNone(response_future.initial_metadata())
     self.assertIs(grpc.StatusCode.CANCELLED, response_future.code())
     self.assertIsNotNone(response_future.details())
@@ -528,6 +542,7 @@ class RPCTest(unittest.TestCase):
             request, timeout=test_constants.SHORT_TIMEOUT,
             metadata=(('test', 'ExpiredUnaryRequestBlockingUnaryResponse'),))
 
+    self.assertIsInstance(exception_context.exception, grpc.Call)
     self.assertIsNotNone(exception_context.exception.initial_metadata())
     self.assertIs(
         grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code())
@@ -556,6 +571,7 @@ class RPCTest(unittest.TestCase):
     self.assertIs(
         grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code())
     self.assertIsInstance(response_future.exception(), grpc.RpcError)
+    self.assertIsNotNone(response_future.traceback())
     self.assertIs(
         grpc.StatusCode.DEADLINE_EXCEEDED, response_future.exception().code())
 
@@ -585,6 +601,8 @@ class RPCTest(unittest.TestCase):
             request_iterator, timeout=test_constants.SHORT_TIMEOUT,
             metadata=(('test', 'ExpiredStreamRequestBlockingUnaryResponse'),))
 
+    self.assertIsInstance(exception_context.exception, grpc.RpcError)
+    self.assertIsInstance(exception_context.exception, grpc.Call)
     self.assertIsNotNone(exception_context.exception.initial_metadata())
     self.assertIs(
         grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code())
@@ -601,6 +619,8 @@ class RPCTest(unittest.TestCase):
       response_future = multi_callable.future(
           request_iterator, timeout=test_constants.SHORT_TIMEOUT,
           metadata=(('test', 'ExpiredStreamRequestFutureUnaryResponse'),))
+      with self.assertRaises(grpc.FutureTimeoutError):
+        response_future.result(timeout=test_constants.SHORT_TIMEOUT / 2.0)
       response_future.add_done_callback(callback)
       value_passed_to_callback = callback.value()
 
@@ -610,6 +630,7 @@ class RPCTest(unittest.TestCase):
     self.assertIs(
         grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code())
     self.assertIsInstance(response_future.exception(), grpc.RpcError)
+    self.assertIsNotNone(response_future.traceback())
     self.assertIs(response_future, value_passed_to_callback)
     self.assertIsNotNone(response_future.initial_metadata())
     self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, response_future.code())
@@ -656,11 +677,14 @@ class RPCTest(unittest.TestCase):
       response_future.add_done_callback(callback)
       value_passed_to_callback = callback.value()
 
+    self.assertIsInstance(response_future, grpc.Future)
+    self.assertIsInstance(response_future, grpc.Call)
     with self.assertRaises(grpc.RpcError) as exception_context:
       response_future.result()
     self.assertIs(
         grpc.StatusCode.UNKNOWN, exception_context.exception.code())
     self.assertIsInstance(response_future.exception(), grpc.RpcError)
+    self.assertIsNotNone(response_future.traceback())
     self.assertIs(grpc.StatusCode.UNKNOWN, response_future.exception().code())
     self.assertIs(response_future, value_passed_to_callback)
 
@@ -709,6 +733,7 @@ class RPCTest(unittest.TestCase):
     self.assertIs(
         grpc.StatusCode.UNKNOWN, exception_context.exception.code())
     self.assertIsInstance(response_future.exception(), grpc.RpcError)
+    self.assertIsNotNone(response_future.traceback())
     self.assertIs(response_future, value_passed_to_callback)
 
   def testFailedStreamRequestStreamResponse(self):
diff --git a/templates/composer.json.template b/templates/composer.json.template
index c9ffbbcbd0..07ab1f20eb 100644
--- a/templates/composer.json.template
+++ b/templates/composer.json.template
@@ -7,15 +7,9 @@
     "keywords": ["rpc"],
     "homepage": "http://grpc.io",
     "license": "BSD-3-Clause",
-    "repositories": [
-      {
-        "type": "vcs",
-        "url": "https://github.com/stanley-cheung/Protobuf-PHP"
-      }
-    ],
     "require": {
       "php": ">=5.5.0",
-      "datto/protobuf-php": "dev-master"
+      "stanley-cheung/protobuf-php": "dev-master"
     },
     "require-dev": {
       "google/auth": "v0.9"
diff --git a/templates/package.xml.template b/templates/package.xml.template
index d8155cdd82..153823ece5 100644
--- a/templates/package.xml.template
+++ b/templates/package.xml.template
@@ -12,18 +12,19 @@
     <email>grpc-packages@google.com</email>
     <active>yes</active>
    </lead>
-   <date>2016-06-30</date>
+   <date>2016-07-13</date>
    <time>16:06:07</time>
    <version>
     <release>${settings.php_version.php()}</release>
     <api>${settings.php_version.php()}</api>
    </version>
    <stability>
-    <release>beta</release>
-    <api>beta</api>
+    <release>stable</release>
+    <api>stable</api>
    </stability>
    <license>BSD</license>
    <notes>
+  - GA release
   - Fix shutdown hang problem #4017
    </notes>
    <contents>
@@ -205,16 +206,17 @@
     </release>
     <release>
      <version>
-      <release>0.15.1</release>
-      <api>0.15.1</api>
+      <release>1.0.0</release>
+      <api>1.0.0</api>
      </version>
      <stability>
-      <release>beta</release>
-      <api>beta</api>
+      <release>stable</release>
+      <api>stable</api>
      </stability>
-     <date>2016-06-30</date>
+     <date>2016-07-13</date>
      <license>BSD</license>
      <notes>
+  - GA release
   - Fix shutdown hang problem #4017
      </notes>
     </release>
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index 6c7eae53a4..4a8936d281 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -345,6 +345,31 @@ TEST_P(AsyncEnd2endTest, SequentialRpcs) {
   SendRpc(10);
 }
 
+// We do not need to protect notify because the use is synchronized.
+void ServerWait(Server* server, int* notify) {
+  server->Wait();
+  *notify = 1;
+}
+TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) {
+  int notify = 0;
+  std::thread* wait_thread =
+      new std::thread(&ServerWait, server_.get(), &notify);
+  ResetStub();
+  SendRpc(1);
+  EXPECT_EQ(0, notify);
+  server_->Shutdown();
+  wait_thread->join();
+  EXPECT_EQ(1, notify);
+  delete wait_thread;
+}
+
+TEST_P(AsyncEnd2endTest, ShutdownThenWait) {
+  ResetStub();
+  SendRpc(1);
+  server_->Shutdown();
+  server_->Wait();
+}
+
 // Test a simple RPC using the async version of Next
 TEST_P(AsyncEnd2endTest, AsyncNextRpc) {
   ResetStub();
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 047bd16408..4045e13460 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -112,6 +112,21 @@ class ClientRequestCreator<ByteBuffer> {
   }
 };
 
+class HistogramEntry GRPC_FINAL {
+ public:
+  HistogramEntry() : used_(false) {}
+  bool used() const { return used_; }
+  double value() const { return value_; }
+  void set_value(double v) {
+    used_ = true;
+    value_ = v;
+  }
+
+ private:
+  bool used_;
+  double value_;
+};
+
 class Client {
  public:
   Client() : timer_(new UsageTimer), interarrival_timer_() {}
@@ -151,10 +166,21 @@ class Client {
     return stats;
   }
 
+  // Must call AwaitThreadsCompletion before destructor to avoid a race
+  // between destructor and invocation of virtual ThreadFunc
+  void AwaitThreadsCompletion() {
+    DestroyMultithreading();
+    std::unique_lock<std::mutex> g(thread_completion_mu_);
+    while (threads_remaining_ != 0) {
+      threads_complete_.wait(g);
+    }
+  }
+
  protected:
   bool closed_loop_;
 
   void StartThreads(size_t num_threads) {
+    threads_remaining_ = num_threads;
     for (size_t i = 0; i < num_threads; i++) {
       threads_.emplace_back(new Thread(this, i));
     }
@@ -162,7 +188,8 @@ class Client {
 
   void EndThreads() { threads_.clear(); }
 
-  virtual bool ThreadFunc(Histogram* histogram, size_t thread_idx) = 0;
+  virtual void DestroyMultithreading() = 0;
+  virtual bool ThreadFunc(HistogramEntry* histogram, size_t thread_idx) = 0;
 
   void SetupLoadTest(const ClientConfig& config, size_t num_threads) {
     // Set up the load distribution based on the number of threads
@@ -215,7 +242,6 @@ class Client {
    public:
     Thread(Client* client, size_t idx)
         : done_(false),
-          new_stats_(nullptr),
           client_(client),
           idx_(idx),
           impl_(&Thread::ThreadFunc, this) {}
@@ -230,15 +256,10 @@ class Client {
 
     void BeginSwap(Histogram* n) {
       std::lock_guard<std::mutex> g(mu_);
-      new_stats_ = n;
+      n->Swap(&histogram_);
     }
 
-    void EndSwap() {
-      std::unique_lock<std::mutex> g(mu_);
-      while (new_stats_ != nullptr) {
-        cv_.wait(g);
-      };
-    }
+    void EndSwap() {}
 
     void MergeStatsInto(Histogram* hist) {
       std::unique_lock<std::mutex> g(mu_);
@@ -252,29 +273,26 @@ class Client {
     void ThreadFunc() {
       for (;;) {
         // run the loop body
-        const bool thread_still_ok = client_->ThreadFunc(&histogram_, idx_);
-        // lock, see if we're done
+        HistogramEntry entry;
+        const bool thread_still_ok = client_->ThreadFunc(&entry, idx_);
+        // lock, update histogram if needed and see if we're done
         std::lock_guard<std::mutex> g(mu_);
+        if (entry.used()) {
+          histogram_.Add(entry.value());
+        }
         if (!thread_still_ok) {
           gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
           done_ = true;
         }
         if (done_) {
+          client_->CompleteThread();
           return;
         }
-        // check if we're resetting stats, swap out the histogram if so
-        if (new_stats_) {
-          new_stats_->Swap(&histogram_);
-          new_stats_ = nullptr;
-          cv_.notify_one();
-        }
       }
     }
 
     std::mutex mu_;
-    std::condition_variable cv_;
     bool done_;
-    Histogram* new_stats_;
     Histogram histogram_;
     Client* client_;
     const size_t idx_;
@@ -286,6 +304,18 @@ class Client {
 
   InterarrivalTimer interarrival_timer_;
   std::vector<gpr_timespec> next_time_;
+
+  std::mutex thread_completion_mu_;
+  size_t threads_remaining_;
+  std::condition_variable threads_complete_;
+
+  void CompleteThread() {
+    std::lock_guard<std::mutex> g(thread_completion_mu_);
+    threads_remaining_--;
+    if (threads_remaining_ == 0) {
+      threads_complete_.notify_all();
+    }
+  }
 };
 
 template <class StubType, class RequestType>
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 24bc0eb5f4..5d9cb4bd0c 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -31,7 +31,6 @@
  *
  */
 
-#include <cassert>
 #include <forward_list>
 #include <functional>
 #include <list>
@@ -48,7 +47,6 @@
 #include <grpc++/generic/generic_stub.h>
 #include <grpc/grpc.h>
 #include <grpc/support/cpu.h>
-#include <grpc/support/histogram.h>
 #include <grpc/support/log.h>
 
 #include "src/proto/grpc/testing/services.grpc.pb.h"
@@ -64,7 +62,7 @@ class ClientRpcContext {
   ClientRpcContext() {}
   virtual ~ClientRpcContext() {}
   // next state, return false if done. Collect stats when appropriate
-  virtual bool RunNextState(bool, Histogram* hist) = 0;
+  virtual bool RunNextState(bool, HistogramEntry* entry) = 0;
   virtual ClientRpcContext* StartNewClone() = 0;
   static void* tag(ClientRpcContext* c) { return reinterpret_cast<void*>(c); }
   static ClientRpcContext* detag(void* t) {
@@ -104,7 +102,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
       alarm_.reset(new Alarm(cq_, next_issue_(), ClientRpcContext::tag(this)));
     }
   }
-  bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
+  bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
     switch (next_state_) {
       case State::READY:
         start_ = UsageTimer::Now();
@@ -114,7 +112,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
         next_state_ = State::RESP_DONE;
         return true;
       case State::RESP_DONE:
-        hist->Add((UsageTimer::Now() - start_) * 1e9);
+        entry->set_value((UsageTimer::Now() - start_) * 1e9);
         callback_(status_, &response_);
         next_state_ = State::INVALID;
         return false;
@@ -176,6 +174,7 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
     for (int i = 0; i < num_async_threads_; i++) {
       cli_cqs_.emplace_back(new CompletionQueue);
       next_issuers_.emplace_back(NextIssuer(i));
+      shutdown_state_.emplace_back(new PerThreadShutdownState());
     }
 
     using namespace std::placeholders;
@@ -192,7 +191,6 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
   }
   virtual ~AsyncClient() {
     for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
-      (*cq)->Shutdown();
       void* got_tag;
       bool ok;
       while ((*cq)->Next(&got_tag, &ok)) {
@@ -201,7 +199,36 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
     }
   }
 
-  bool ThreadFunc(Histogram* histogram,
+ protected:
+  const int num_async_threads_;
+
+ private:
+  struct PerThreadShutdownState {
+    mutable std::mutex mutex;
+    bool shutdown;
+    PerThreadShutdownState() : shutdown(false) {}
+  };
+
+  int NumThreads(const ClientConfig& config) {
+    int num_threads = config.async_client_threads();
+    if (num_threads <= 0) {  // Use dynamic sizing
+      num_threads = cores_;
+      gpr_log(GPR_INFO, "Sizing async client to %d threads", num_threads);
+    }
+    return num_threads;
+  }
+  void DestroyMultithreading() GRPC_OVERRIDE GRPC_FINAL {
+    for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
+      std::lock_guard<std::mutex> lock((*ss)->mutex);
+      (*ss)->shutdown = true;
+    }
+    for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
+      (*cq)->Shutdown();
+    }
+    this->EndThreads();  // this needed for resolution
+  }
+
+  bool ThreadFunc(HistogramEntry* entry,
                   size_t thread_idx) GRPC_OVERRIDE GRPC_FINAL {
     void* got_tag;
     bool ok;
@@ -209,12 +236,15 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
     switch (cli_cqs_[thread_idx]->AsyncNext(
         &got_tag, &ok,
         std::chrono::system_clock::now() + std::chrono::milliseconds(10))) {
-      case CompletionQueue::SHUTDOWN:
-        return false;
       case CompletionQueue::GOT_EVENT: {
         // Got a regular event, so process it
         ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
-        if (!ctx->RunNextState(ok, histogram)) {
+        // Proceed while holding a lock to make sure that
+        // this thread isn't supposed to shut down
+        std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
+        if (shutdown_state_[thread_idx]->shutdown) {
+          return true;
+        } else if (!ctx->RunNextState(ok, entry)) {
           // The RPC and callback are done, so clone the ctx
           // and kickstart the new one
           auto clone = ctx->StartNewClone();
@@ -224,27 +254,23 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
         }
         return true;
       }
-      case CompletionQueue::TIMEOUT:
+      case CompletionQueue::TIMEOUT: {
+        std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
+        if (shutdown_state_[thread_idx]->shutdown) {
+          return true;
+        }
+        return true;
+      }
+      case CompletionQueue::SHUTDOWN:  // queue is shutting down, so we must be
+                                       // done
         return true;
     }
-    GPR_UNREACHABLE_CODE(return false);
-  }
-
- protected:
-  const int num_async_threads_;
-
- private:
-  int NumThreads(const ClientConfig& config) {
-    int num_threads = config.async_client_threads();
-    if (num_threads <= 0) {  // Use dynamic sizing
-      num_threads = cores_;
-      gpr_log(GPR_INFO, "Sizing async client to %d threads", num_threads);
-    }
-    return num_threads;
+    GPR_UNREACHABLE_CODE(return true);
   }
 
   std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
   std::vector<std::function<gpr_timespec()>> next_issuers_;
+  std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
 };
 
 static std::unique_ptr<BenchmarkService::Stub> BenchmarkStubCreator(
@@ -260,7 +286,7 @@ class AsyncUnaryClient GRPC_FINAL
             config, SetupCtx, BenchmarkStubCreator) {
     StartThreads(num_async_threads_);
   }
-  ~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
+  ~AsyncUnaryClient() GRPC_OVERRIDE {}
 
  private:
   static void CheckDone(grpc::Status s, SimpleResponse* response) {}
@@ -305,7 +331,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
     stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this));
     next_state_ = State::STREAM_IDLE;
   }
-  bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
+  bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
     while (true) {
       switch (next_state_) {
         case State::STREAM_IDLE:
@@ -337,7 +363,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
           return true;
           break;
         case State::READ_DONE:
-          hist->Add((UsageTimer::Now() - start_) * 1e9);
+          entry->set_value((UsageTimer::Now() - start_) * 1e9);
           callback_(status_, &response_);
           next_state_ = State::STREAM_IDLE;
           break;  // loop around
@@ -389,7 +415,7 @@ class AsyncStreamingClient GRPC_FINAL
     StartThreads(num_async_threads_);
   }
 
-  ~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
+  ~AsyncStreamingClient() GRPC_OVERRIDE {}
 
  private:
   static void CheckDone(grpc::Status s, SimpleResponse* response) {}
@@ -437,7 +463,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
                          ClientRpcContext::tag(this));
     next_state_ = State::STREAM_IDLE;
   }
-  bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
+  bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
     while (true) {
       switch (next_state_) {
         case State::STREAM_IDLE:
@@ -469,7 +495,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
           return true;
           break;
         case State::READ_DONE:
-          hist->Add((UsageTimer::Now() - start_) * 1e9);
+          entry->set_value((UsageTimer::Now() - start_) * 1e9);
           callback_(status_, &response_);
           next_state_ = State::STREAM_IDLE;
           break;  // loop around
@@ -525,7 +551,7 @@ class GenericAsyncStreamingClient GRPC_FINAL
     StartThreads(num_async_threads_);
   }
 
-  ~GenericAsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
+  ~GenericAsyncStreamingClient() GRPC_OVERRIDE {}
 
  private:
   static void CheckDone(grpc::Status s, ByteBuffer* response) {}
diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index c88e95b80e..25c7823553 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -31,7 +31,6 @@
  *
  */
 
-#include <cassert>
 #include <chrono>
 #include <memory>
 #include <mutex>
@@ -46,7 +45,6 @@
 #include <grpc++/server_builder.h>
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
-#include <grpc/support/histogram.h>
 #include <grpc/support/host_port.h>
 #include <grpc/support/log.h>
 #include <grpc/support/time.h>
@@ -55,7 +53,6 @@
 #include "src/core/lib/profiling/timers.h"
 #include "src/proto/grpc/testing/services.grpc.pb.h"
 #include "test/cpp/qps/client.h"
-#include "test/cpp/qps/histogram.h"
 #include "test/cpp/qps/interarrival.h"
 #include "test/cpp/qps/usage_timer.h"
 
@@ -90,6 +87,9 @@ class SynchronousClient
 
   size_t num_threads_;
   std::vector<SimpleResponse> responses_;
+
+ private:
+  void DestroyMultithreading() GRPC_OVERRIDE GRPC_FINAL { EndThreads(); }
 };
 
 class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
@@ -98,9 +98,9 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
       : SynchronousClient(config) {
     StartThreads(num_threads_);
   }
-  ~SynchronousUnaryClient() { EndThreads(); }
+  ~SynchronousUnaryClient() {}
 
-  bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
+  bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) GRPC_OVERRIDE {
     WaitToIssue(thread_idx);
     auto* stub = channels_[thread_idx % channels_.size()].get_stub();
     double start = UsageTimer::Now();
@@ -108,7 +108,7 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
     grpc::ClientContext context;
     grpc::Status s =
         stub->UnaryCall(&context, request_, &responses_[thread_idx]);
-    histogram->Add((UsageTimer::Now() - start) * 1e9);
+    entry->set_value((UsageTimer::Now() - start) * 1e9);
     return s.ok();
   }
 };
@@ -127,25 +127,29 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
     StartThreads(num_threads_);
   }
   ~SynchronousStreamingClient() {
-    EndThreads();
-    for (auto stream = &stream_[0]; stream != &stream_[num_threads_];
-         stream++) {
+    for (size_t i = 0; i < num_threads_; i++) {
+      auto stream = &stream_[i];
       if (*stream) {
         (*stream)->WritesDone();
-        EXPECT_TRUE((*stream)->Finish().ok());
+        Status s = (*stream)->Finish();
+        EXPECT_TRUE(s.ok());
+        if (!s.ok()) {
+          gpr_log(GPR_ERROR, "Stream %zu received an error %s", i,
+                  s.error_message().c_str());
+        }
       }
     }
     delete[] stream_;
     delete[] context_;
   }
 
-  bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
+  bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) GRPC_OVERRIDE {
     WaitToIssue(thread_idx);
     GPR_TIMER_SCOPE("SynchronousStreamingClient::ThreadFunc", 0);
     double start = UsageTimer::Now();
     if (stream_[thread_idx]->Write(request_) &&
         stream_[thread_idx]->Read(&responses_[thread_idx])) {
-      histogram->Add((UsageTimer::Now() - start) * 1e9);
+      entry->set_value((UsageTimer::Now() - start) * 1e9);
       return true;
     }
     return false;
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index 08bf045883..2aeaea51f2 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -87,7 +87,7 @@ static std::unordered_map<string, std::deque<int>> get_hosts_and_cores(
       CoreRequest dummy;
       CoreResponse cores;
       grpc::Status s = stub->CoreCount(&ctx, dummy, &cores);
-      assert(s.ok());
+      GPR_ASSERT(s.ok());
       std::deque<int> dq;
       for (int i = 0; i < cores.cores(); i++) {
         dq.push_back(i);
@@ -289,9 +289,13 @@ std::unique_ptr<ScenarioResult> RunScenario(
     *args.mutable_setup() = server_config;
     servers[i].stream =
         servers[i].stub->RunServer(runsc::AllocContext(&contexts));
-    GPR_ASSERT(servers[i].stream->Write(args));
+    if (!servers[i].stream->Write(args)) {
+      gpr_log(GPR_ERROR, "Could not write args to server %zu", i);
+    }
     ServerStatus init_status;
-    GPR_ASSERT(servers[i].stream->Read(&init_status));
+    if (!servers[i].stream->Read(&init_status)) {
+      gpr_log(GPR_ERROR, "Server %zu did not yield initial status", i);
+    }
     gpr_join_host_port(&cli_target, host, init_status.port());
     client_config.add_server_targets(cli_target);
     gpr_free(host);
@@ -345,9 +349,13 @@ std::unique_ptr<ScenarioResult> RunScenario(
     *args.mutable_setup() = per_client_config;
     clients[i].stream =
         clients[i].stub->RunClient(runsc::AllocContext(&contexts));
-    GPR_ASSERT(clients[i].stream->Write(args));
+    if (!clients[i].stream->Write(args)) {
+      gpr_log(GPR_ERROR, "Could not write args to client %zu", i);
+    }
     ClientStatus init_status;
-    GPR_ASSERT(clients[i].stream->Read(&init_status));
+    if (!clients[i].stream->Read(&init_status)) {
+      gpr_log(GPR_ERROR, "Client %zu did not yield initial status", i);
+    }
   }
 
   // Let everything warmup
@@ -362,19 +370,31 @@ std::unique_ptr<ScenarioResult> RunScenario(
   server_mark.mutable_mark()->set_reset(true);
   ClientArgs client_mark;
   client_mark.mutable_mark()->set_reset(true);
-  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
-    GPR_ASSERT(server->stream->Write(server_mark));
+  for (size_t i = 0; i < num_servers; i++) {
+    auto server = &servers[i];
+    if (!server->stream->Write(server_mark)) {
+      gpr_log(GPR_ERROR, "Couldn't write mark to server %zu", i);
+    }
   }
-  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
-    GPR_ASSERT(client->stream->Write(client_mark));
+  for (size_t i = 0; i < num_clients; i++) {
+    auto client = &clients[i];
+    if (!client->stream->Write(client_mark)) {
+      gpr_log(GPR_ERROR, "Couldn't write mark to client %zu", i);
+    }
   }
   ServerStatus server_status;
   ClientStatus client_status;
-  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
-    GPR_ASSERT(server->stream->Read(&server_status));
+  for (size_t i = 0; i < num_servers; i++) {
+    auto server = &servers[i];
+    if (!server->stream->Read(&server_status)) {
+      gpr_log(GPR_ERROR, "Couldn't get status from server %zu", i);
+    }
   }
-  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
-    GPR_ASSERT(client->stream->Read(&client_status));
+  for (size_t i = 0; i < num_clients; i++) {
+    auto client = &clients[i];
+    if (!client->stream->Read(&client_status)) {
+      gpr_log(GPR_ERROR, "Couldn't get status from client %zu", i);
+    }
   }
 
   // Wait some time
@@ -390,37 +410,73 @@ std::unique_ptr<ScenarioResult> RunScenario(
   Histogram merged_latencies;
 
   gpr_log(GPR_INFO, "Finishing clients");
-  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
-    GPR_ASSERT(client->stream->Write(client_mark));
-    GPR_ASSERT(client->stream->WritesDone());
+  for (size_t i = 0; i < num_clients; i++) {
+    auto client = &clients[i];
+    if (!client->stream->Write(client_mark)) {
+      gpr_log(GPR_ERROR, "Couldn't write mark to client %zu", i);
+    }
+    if (!client->stream->WritesDone()) {
+      gpr_log(GPR_ERROR, "Failed WritesDone for client %zu", i);
+    }
   }
-  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
-    GPR_ASSERT(client->stream->Read(&client_status));
-    const auto& stats = client_status.stats();
-    merged_latencies.MergeProto(stats.latencies());
-    result->add_client_stats()->CopyFrom(stats);
-    GPR_ASSERT(!client->stream->Read(&client_status));
+  for (size_t i = 0; i < num_clients; i++) {
+    auto client = &clients[i];
+    // Read the client final status
+    if (client->stream->Read(&client_status)) {
+      gpr_log(GPR_INFO, "Received final status from client %zu", i);
+      const auto& stats = client_status.stats();
+      merged_latencies.MergeProto(stats.latencies());
+      result->add_client_stats()->CopyFrom(stats);
+      // That final status should be the last message on the client stream
+      GPR_ASSERT(!client->stream->Read(&client_status));
+    } else {
+      gpr_log(GPR_ERROR, "Couldn't get final status from client %zu", i);
+    }
   }
-  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
-    GPR_ASSERT(client->stream->Finish().ok());
+  for (size_t i = 0; i < num_clients; i++) {
+    auto client = &clients[i];
+    Status s = client->stream->Finish();
+    result->add_client_success(s.ok());
+    if (!s.ok()) {
+      gpr_log(GPR_ERROR, "Client %zu had an error %s", i,
+              s.error_message().c_str());
+    }
   }
   delete[] clients;
 
   merged_latencies.FillProto(result->mutable_latencies());
 
   gpr_log(GPR_INFO, "Finishing servers");
-  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
-    GPR_ASSERT(server->stream->Write(server_mark));
-    GPR_ASSERT(server->stream->WritesDone());
+  for (size_t i = 0; i < num_servers; i++) {
+    auto server = &servers[i];
+    if (!server->stream->Write(server_mark)) {
+      gpr_log(GPR_ERROR, "Couldn't write mark to server %zu", i);
+    }
+    if (!server->stream->WritesDone()) {
+      gpr_log(GPR_ERROR, "Failed WritesDone for server %zu", i);
+    }
   }
-  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
-    GPR_ASSERT(server->stream->Read(&server_status));
-    result->add_server_stats()->CopyFrom(server_status.stats());
-    result->add_server_cores(server_status.cores());
-    GPR_ASSERT(!server->stream->Read(&server_status));
+  for (size_t i = 0; i < num_servers; i++) {
+    auto server = &servers[i];
+    // Read the server final status
+    if (server->stream->Read(&server_status)) {
+      gpr_log(GPR_INFO, "Received final status from server %zu", i);
+      result->add_server_stats()->CopyFrom(server_status.stats());
+      result->add_server_cores(server_status.cores());
+      // That final status should be the last message on the server stream
+      GPR_ASSERT(!server->stream->Read(&server_status));
+    } else {
+      gpr_log(GPR_ERROR, "Couldn't get final status from server %zu", i);
+    }
   }
-  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
-    GPR_ASSERT(server->stream->Finish().ok());
+  for (size_t i = 0; i < num_servers; i++) {
+    auto server = &servers[i];
+    Status s = server->stream->Finish();
+    result->add_server_success(s.ok());
+    if (!s.ok()) {
+      gpr_log(GPR_ERROR, "Server %zu had an error %s", i,
+              s.error_message().c_str());
+    }
   }
 
   delete[] servers;
@@ -429,8 +485,9 @@ std::unique_ptr<ScenarioResult> RunScenario(
   return result;
 }
 
-void RunQuit() {
+bool RunQuit() {
   // Get client, server lists
+  bool result = true;
   auto workers = get_workers("QPS_WORKERS");
   for (size_t i = 0; i < workers.size(); i++) {
     auto stub = WorkerService::NewStub(
@@ -438,8 +495,14 @@ void RunQuit() {
     Void dummy;
     grpc::ClientContext ctx;
     ctx.set_fail_fast(false);
-    GPR_ASSERT(stub->QuitWorker(&ctx, dummy, &dummy).ok());
+    Status s = stub->QuitWorker(&ctx, dummy, &dummy);
+    if (!s.ok()) {
+      gpr_log(GPR_ERROR, "Worker %zu could not be properly quit because %s", i,
+              s.error_message().c_str());
+      result = false;
+    }
   }
+  return result;
 }
 
 }  // namespace testing
diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h
index 3a5cf138f1..93f4370caf 100644
--- a/test/cpp/qps/driver.h
+++ b/test/cpp/qps/driver.h
@@ -47,7 +47,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
     const grpc::testing::ServerConfig& server_config, size_t num_servers,
     int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count);
 
-void RunQuit();
+bool RunQuit();
 }  // namespace testing
 }  // namespace grpc
 
diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc
index f5d739f893..1524ebbc38 100644
--- a/test/cpp/qps/qps_json_driver.cc
+++ b/test/cpp/qps/qps_json_driver.cc
@@ -53,7 +53,7 @@ DEFINE_bool(quit, false, "Quit the workers");
 namespace grpc {
 namespace testing {
 
-static void QpsDriver() {
+static bool QpsDriver() {
   grpc::string json;
 
   bool scfile = (FLAGS_scenarios_file != "");
@@ -81,13 +81,13 @@ static void QpsDriver() {
   } else if (scjson) {
     json = FLAGS_scenarios_json.c_str();
   } else if (FLAGS_quit) {
-    RunQuit();
-    return;
+    return RunQuit();
   }
 
   // Parse into an array of scenarios
   Scenarios scenarios;
   ParseJson(json.c_str(), "grpc.testing.Scenarios", &scenarios);
+  bool success = true;
 
   // Make sure that there is at least some valid scenario here
   GPR_ASSERT(scenarios.scenarios_size() > 0);
@@ -109,7 +109,15 @@ static void QpsDriver() {
     GetReporter()->ReportQPSPerCore(*result);
     GetReporter()->ReportLatency(*result);
     GetReporter()->ReportTimes(*result);
+
+    for (int i = 0; success && i < result->client_success_size(); i++) {
+      success = result->client_success(i);
+    }
+    for (int i = 0; success && i < result->server_success_size(); i++) {
+      success = result->server_success(i);
+    }
   }
+  return success;
 }
 
 }  // namespace testing
@@ -118,7 +126,7 @@ static void QpsDriver() {
 int main(int argc, char **argv) {
   grpc::testing::InitBenchmark(&argc, &argv, true);
 
-  grpc::testing::QpsDriver();
+  bool ok = grpc::testing::QpsDriver();
 
-  return 0;
+  return ok ? 0 : 1;
 }
diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc
index f514e23e85..d3e53fe14a 100644
--- a/test/cpp/qps/qps_worker.cc
+++ b/test/cpp/qps/qps_worker.cc
@@ -33,7 +33,6 @@
 
 #include "test/cpp/qps/qps_worker.h"
 
-#include <cassert>
 #include <memory>
 #include <mutex>
 #include <sstream>
@@ -124,11 +123,12 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
       GRPC_OVERRIDE {
     InstanceGuard g(this);
     if (!g.Acquired()) {
-      return Status(StatusCode::RESOURCE_EXHAUSTED, "");
+      return Status(StatusCode::RESOURCE_EXHAUSTED, "Client worker busy");
     }
 
     ScopedProfile profile("qps_client.prof", false);
     Status ret = RunClientBody(ctx, stream);
+    gpr_log(GPR_INFO, "RunClient: Returning");
     return ret;
   }
 
@@ -137,11 +137,12 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
       GRPC_OVERRIDE {
     InstanceGuard g(this);
     if (!g.Acquired()) {
-      return Status(StatusCode::RESOURCE_EXHAUSTED, "");
+      return Status(StatusCode::RESOURCE_EXHAUSTED, "Server worker busy");
     }
 
     ScopedProfile profile("qps_server.prof", false);
     Status ret = RunServerBody(ctx, stream);
+    gpr_log(GPR_INFO, "RunServer: Returning");
     return ret;
   }
 
@@ -154,7 +155,7 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
   Status QuitWorker(ServerContext* ctx, const Void*, Void*) GRPC_OVERRIDE {
     InstanceGuard g(this);
     if (!g.Acquired()) {
-      return Status(StatusCode::RESOURCE_EXHAUSTED, "");
+      return Status(StatusCode::RESOURCE_EXHAUSTED, "Quitting worker busy");
     }
 
     worker_->MarkDone();
@@ -197,33 +198,38 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
                        ServerReaderWriter<ClientStatus, ClientArgs>* stream) {
     ClientArgs args;
     if (!stream->Read(&args)) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Couldn't read args");
     }
     if (!args.has_setup()) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Invalid setup arg");
     }
     gpr_log(GPR_INFO, "RunClientBody: about to create client");
     auto client = CreateClient(args.setup());
     if (!client) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Couldn't create client");
     }
     gpr_log(GPR_INFO, "RunClientBody: client created");
     ClientStatus status;
     if (!stream->Write(status)) {
-      return Status(StatusCode::UNKNOWN, "");
+      return Status(StatusCode::UNKNOWN, "Client couldn't report init status");
     }
     gpr_log(GPR_INFO, "RunClientBody: creation status reported");
     while (stream->Read(&args)) {
       gpr_log(GPR_INFO, "RunClientBody: Message read");
       if (!args.has_mark()) {
         gpr_log(GPR_INFO, "RunClientBody: Message is not a mark!");
-        return Status(StatusCode::INVALID_ARGUMENT, "");
+        return Status(StatusCode::INVALID_ARGUMENT, "Invalid mark");
       }
       *status.mutable_stats() = client->Mark(args.mark().reset());
-      stream->Write(status);
+      if (!stream->Write(status)) {
+        return Status(StatusCode::UNKNOWN, "Client couldn't respond to mark");
+      }
       gpr_log(GPR_INFO, "RunClientBody: Mark response given");
     }
 
+    gpr_log(GPR_INFO, "RunClientBody: Awaiting Threads Completion");
+    client->AwaitThreadsCompletion();
+
     gpr_log(GPR_INFO, "RunClientBody: Returning");
     return Status::OK;
   }
@@ -232,10 +238,10 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
                        ServerReaderWriter<ServerStatus, ServerArgs>* stream) {
     ServerArgs args;
     if (!stream->Read(&args)) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Couldn't read server args");
     }
     if (!args.has_setup()) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Bad server creation args");
     }
     if (server_port_ != 0) {
       args.mutable_setup()->set_port(server_port_);
@@ -243,24 +249,26 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
     gpr_log(GPR_INFO, "RunServerBody: about to create server");
     auto server = CreateServer(args.setup());
     if (!server) {
-      return Status(StatusCode::INVALID_ARGUMENT, "");
+      return Status(StatusCode::INVALID_ARGUMENT, "Couldn't create server");
     }
     gpr_log(GPR_INFO, "RunServerBody: server created");
     ServerStatus status;
     status.set_port(server->port());
     status.set_cores(server->cores());
     if (!stream->Write(status)) {
-      return Status(StatusCode::UNKNOWN, "");
+      return Status(StatusCode::UNKNOWN, "Server couldn't report init status");
     }
     gpr_log(GPR_INFO, "RunServerBody: creation status reported");
     while (stream->Read(&args)) {
       gpr_log(GPR_INFO, "RunServerBody: Message read");
       if (!args.has_mark()) {
         gpr_log(GPR_INFO, "RunServerBody: Message not a mark!");
-        return Status(StatusCode::INVALID_ARGUMENT, "");
+        return Status(StatusCode::INVALID_ARGUMENT, "Invalid mark");
       }
       *status.mutable_stats() = server->Mark(args.mark().reset());
-      stream->Write(status);
+      if (!stream->Write(status)) {
+        return Status(StatusCode::UNKNOWN, "Server couldn't respond to mark");
+      }
       gpr_log(GPR_INFO, "RunServerBody: Mark response given");
     }
 
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index 73ca19148b..dea8746331 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -123,22 +123,24 @@ class AsyncQpsServerTest : public Server {
 
     for (int i = 0; i < num_threads; i++) {
       shutdown_state_.emplace_back(new PerThreadShutdownState());
-    }
-    for (int i = 0; i < num_threads; i++) {
       threads_.emplace_back(&AsyncQpsServerTest::ThreadFunc, this, i);
     }
   }
   ~AsyncQpsServerTest() {
     for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
-      (*ss)->set_shutdown();
+      std::lock_guard<std::mutex> lock((*ss)->mutex);
+      (*ss)->shutdown = true;
+    }
+    // TODO (vpai): Remove this deadline and allow Shutdown to finish properly
+    auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(3);
+    server_->Shutdown(deadline);
+    for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
+      (*cq)->Shutdown();
     }
-    server_->Shutdown(std::chrono::system_clock::now() +
-                      std::chrono::seconds(3));
     for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
       thr->join();
     }
     for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
-      (*cq)->Shutdown();
       bool ok;
       void *got_tag;
       while ((*cq)->Next(&got_tag, &ok))
@@ -151,22 +153,24 @@ class AsyncQpsServerTest : public Server {
   }
 
  private:
-  void ThreadFunc(int rank) {
+  void ThreadFunc(int thread_idx) {
     // Wait until work is available or we are shutting down
     bool ok;
     void *got_tag;
-    while (srv_cqs_[rank]->Next(&got_tag, &ok)) {
+    while (srv_cqs_[thread_idx]->Next(&got_tag, &ok)) {
       ServerRpcContext *ctx = detag(got_tag);
       // The tag is a pointer to an RPC context to invoke
-      const bool still_going = ctx->RunNextState(ok);
-      if (!shutdown_state_[rank]->shutdown()) {
-        // this RPC context is done, so refresh it
-        if (!still_going) {
-          ctx->Reset();
-        }
-      } else {
+      // Proceed while holding a lock to make sure that
+      // this thread isn't supposed to shut down
+      std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
+      if (shutdown_state_[thread_idx]->shutdown) {
         return;
       }
+      const bool still_going = ctx->RunNextState(ok);
+      // if this RPC context is done, refresh it
+      if (!still_going) {
+        ctx->Reset();
+      }
     }
     return;
   }
@@ -334,24 +338,12 @@ class AsyncQpsServerTest : public Server {
   ServiceType async_service_;
   std::forward_list<ServerRpcContext *> contexts_;
 
-  class PerThreadShutdownState {
-   public:
-    PerThreadShutdownState() : shutdown_(false) {}
-
-    bool shutdown() const {
-      std::lock_guard<std::mutex> lock(mutex_);
-      return shutdown_;
-    }
-
-    void set_shutdown() {
-      std::lock_guard<std::mutex> lock(mutex_);
-      shutdown_ = true;
-    }
-
-   private:
-    mutable std::mutex mutex_;
-    bool shutdown_;
+  struct PerThreadShutdownState {
+    mutable std::mutex mutex;
+    bool shutdown;
+    PerThreadShutdownState() : shutdown(false) {}
   };
+
   std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
 };
 
diff --git a/tools/distrib/python/grpcio_tools/setup.py b/tools/distrib/python/grpcio_tools/setup.py
index d804f34fc6..bb2c71d843 100644
--- a/tools/distrib/python/grpcio_tools/setup.py
+++ b/tools/distrib/python/grpcio_tools/setup.py
@@ -51,15 +51,43 @@ import grpc_version
 
 PY3 = sys.version_info.major == 3
 
+# Environment variable to determine whether or not the Cython extension should
+# *use* Cython or use the generated C files. Note that this requires the C files
+# to have been generated by building first *with* Cython support.
+BUILD_WITH_CYTHON = os.environ.get('GRPC_PYTHON_BUILD_WITH_CYTHON', False)
+
 # There are some situations (like on Windows) where CC, CFLAGS, and LDFLAGS are
 # entirely ignored/dropped/forgotten by distutils and its Cygwin/MinGW support.
 # We use these environment variables to thus get around that without locking
 # ourselves in w.r.t. the multitude of operating systems this ought to build on.
-# By default we assume a GCC-like compiler.
-EXTRA_COMPILE_ARGS = shlex.split(os.environ.get('GRPC_PYTHON_CFLAGS',
-                                                '-fno-wrapv -frtti -std=c++11'))
-EXTRA_LINK_ARGS = shlex.split(os.environ.get('GRPC_PYTHON_LDFLAGS',
-                                             '-lpthread'))
+# We can also use these variables as a way to inject environment-specific
+# compiler/linker flags. We assume GCC-like compilers and/or MinGW as a
+# reasonable default.
+EXTRA_ENV_COMPILE_ARGS = os.environ.get('GRPC_PYTHON_CFLAGS', None)
+EXTRA_ENV_LINK_ARGS = os.environ.get('GRPC_PYTHON_LDFLAGS', None)
+if EXTRA_ENV_COMPILE_ARGS is None:
+  EXTRA_ENV_COMPILE_ARGS = '-fno-wrapv -frtti -std=c++11'
+  if 'win32' in sys.platform:
+    # We use define flags here and don't directly add to DEFINE_MACROS below to
+    # ensure that the expert user/builder has a way of turning it off (via the
+    # envvars) without adding yet more GRPC-specific envvars.
+    # See https://sourceforge.net/p/mingw-w64/bugs/363/
+    if '32' in platform.architecture()[0]:
+      EXTRA_ENV_COMPILE_ARGS += ' -D_ftime=_ftime32 -D_timeb=__timeb32 -D_ftime_s=_ftime32_s'
+    else:
+      EXTRA_ENV_COMPILE_ARGS += ' -D_ftime=_ftime64 -D_timeb=__timeb64'
+if EXTRA_ENV_LINK_ARGS is None:
+  EXTRA_ENV_LINK_ARGS = '-lpthread'
+  if 'win32' in sys.platform:
+    # TODO(atash) check if this is actually safe to just import and call on
+    # non-Windows (to avoid breaking import style)
+    from distutils.cygwinccompiler import get_msvcr
+    msvcr = get_msvcr()[0]
+    EXTRA_ENV_LINK_ARGS += (
+        ' -static-libgcc -static-libstdc++ -mcrtdll={msvcr} '
+        '-static'.format(msvcr=msvcr))
+EXTRA_COMPILE_ARGS = shlex.split(EXTRA_ENV_COMPILE_ARGS)
+EXTRA_LINK_ARGS = shlex.split(EXTRA_ENV_LINK_ARGS)
 
 GRPC_PYTHON_TOOLS_PACKAGE = 'grpc.tools'
 GRPC_PYTHON_PROTO_RESOURCES_NAME = '_proto'
@@ -97,15 +125,19 @@ def package_data():
     proto_files.append(relative_target)
   return {GRPC_PYTHON_TOOLS_PACKAGE: proto_files}
 
-def protoc_ext_module():
-  plugin_sources = [
+def extension_modules():
+  if BUILD_WITH_CYTHON:
+    plugin_sources = ['grpc/tools/_protoc_compiler.pyx']
+  else:
+    plugin_sources = ['grpc/tools/_protoc_compiler.cpp']
+  plugin_sources += [
       'grpc/tools/main.cc',
       'grpc_root/src/compiler/python_generator.cc'] + [
       os.path.join(protoc_lib_deps.CC_INCLUDE, cc_file)
       for cc_file in protoc_lib_deps.CC_FILES]
   plugin_ext = extension.Extension(
       name='grpc.tools._protoc_compiler',
-      sources=['grpc/tools/_protoc_compiler.pyx'] + plugin_sources,
+      sources=plugin_sources,
       include_dirs=[
           '.',
           'grpc_root',
@@ -117,19 +149,18 @@ def protoc_ext_module():
       extra_compile_args=list(EXTRA_COMPILE_ARGS),
       extra_link_args=list(EXTRA_LINK_ARGS),
   )
-  return plugin_ext
-
-def maybe_cythonize(exts):
-  from Cython import Build
-  return Build.cythonize(exts)
+  extensions = [plugin_ext]
+  if BUILD_WITH_CYTHON:
+    from Cython import Build
+    return Build.cythonize(extensions)
+  else:
+    return extensions
 
 setuptools.setup(
   name='grpcio_tools',
   version=grpc_version.VERSION,
   license='3-clause BSD',
-  ext_modules=maybe_cythonize([
-      protoc_ext_module(),
-  ]),
+  ext_modules=extension_modules(),
   packages=setuptools.find_packages('.'),
   namespace_packages=['grpc'],
   install_requires=[
diff --git a/tools/run_tests/build_artifact_python.bat b/tools/run_tests/build_artifact_python.bat
index 7c8c2aa12d..a7b1a58284 100644
--- a/tools/run_tests/build_artifact_python.bat
+++ b/tools/run_tests/build_artifact_python.bat
@@ -34,29 +34,6 @@ pip install --upgrade six
 pip install --upgrade setuptools
 pip install -rrequirements.txt
 
-@rem Because this is windows and *everything seems to hate Windows* we have to
-@rem set all of these flags ourselves because Python won't help us (see the
-@rem setup.py of the grpcio_tools project).
-set GRPC_PYTHON_CFLAGS=-fno-wrapv -frtti -std=c++11
-
-@rem See https://sourceforge.net/p/mingw-w64/bugs/363/
-if %2 == 32 (
-  set GRPC_PYTHON_CFLAGS=%GRPC_PYTHON_CFLAGS% -D_ftime=_ftime32 -D_timeb=__timeb32 -D_ftime_s=_ftime32_s
-) else (
-  set GRPC_PYTHON_CFLAGS=%GRPC_PYTHON_CFLAGS% -D_ftime=_ftime64 -D_timeb=__timeb64
-)
-
-@rem Further confusing things, MSYS2's mingw64 tries to dynamically link
-@rem libgcc, libstdc++, and winpthreads. We have to override this or our
-@rem extensions end up linking to MSYS2 DLLs, which the normal Python on
-@rem Windows user won't have... and ON TOP OF THIS, there's MinGW's GCC default
-@rem behavior of linking msvcrt.dll as the C runtime library, which we need to
-@rem override so that Python's distutils doesn't link us against multiple C
-@rem runtimes.
-python -c "from distutils.cygwinccompiler import get_msvcr; print(get_msvcr()[0])" > temp.txt
-set /p PYTHON_MSVCR=<temp.txt
-set GRPC_PYTHON_LDFLAGS=-static-libgcc -static-libstdc++ -mcrtdll=%PYTHON_MSVCR% -static -lpthread
-
 set GRPC_PYTHON_BUILD_WITH_CYTHON=1
 
 
diff --git a/tools/run_tests/build_python.sh b/tools/run_tests/build_python.sh
index a3fa8200d5..9cb3cb12a9 100755
--- a/tools/run_tests/build_python.sh
+++ b/tools/run_tests/build_python.sh
@@ -127,19 +127,6 @@ if [ $(is_linux) ]; then
     fi
   fi
 fi
-# TODO(atash) consider conceptualizing MinGW as a first-class platform and move
-# these flags into our `setup.py`s
-if [ $(is_mingw) ]; then
-  # We're on MinGW, and our CFLAGS and LDFLAGS will be eaten by the void. Use
-  # our work-around environment variables instead.
-  PYTHON_MSVCR=`$PYTHON -c "from distutils.cygwinccompiler import get_msvcr; print(get_msvcr()[0])"`
-  export GRPC_PYTHON_LDFLAGS="-static-libgcc -static-libstdc++ -mcrtdll=$PYTHON_MSVCR -static -lpthread"
-  # See https://sourceforge.net/p/mingw-w64/bugs/363/
-  export GRPC_PYTHON_CFLAGS="-D_ftime=_ftime64 -D_timeb=__timeb64"
-  # TODO(atash) set these flags for only grpcio-tools (they don't do any harm to
-  # grpcio, but they result in noisy warnings).
-  export GRPC_PYTHON_CFLAGS="-frtti -std=c++11 $GRPC_PYTHON_CFLAGS"
-fi
 
 ############################
 # Perform build operations #