Merge branch 'master' into failhijackedrecv

17 files changed, 427 insertions, 139 deletions

diff --git a/src/core/lib/iomgr/resource_quota.cc b/src/core/lib/iomgr/resource_quota.cc
index 7e4b3c9b2f..61c366098e 100644
--- a/src/core/lib/iomgr/resource_quota.cc
+++ b/src/core/lib/iomgr/resource_quota.cc
@@ -665,6 +665,7 @@ void grpc_resource_quota_unref_internal(grpc_resource_quota* resource_quota) {
     GPR_ASSERT(resource_quota->num_threads_allocated == 0);
     GRPC_COMBINER_UNREF(resource_quota->combiner, "resource_quota");
     gpr_free(resource_quota->name);
+    gpr_mu_destroy(&resource_quota->thread_count_mu);
     gpr_free(resource_quota);
   }
 }
diff --git a/src/core/lib/iomgr/tcp_windows.cc b/src/core/lib/iomgr/tcp_windows.cc
index 4b5250803d..86ee1010cf 100644
--- a/src/core/lib/iomgr/tcp_windows.cc
+++ b/src/core/lib/iomgr/tcp_windows.cc
@@ -42,6 +42,7 @@
 #include "src/core/lib/iomgr/tcp_windows.h"
 #include "src/core/lib/iomgr/timer.h"
 #include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/slice/slice_string_helpers.h"
 
 #if defined(__MSYS__) && defined(GPR_ARCH_64)
 /* Nasty workaround for nasty bug when using the 64 bits msys compiler
@@ -112,7 +113,10 @@ typedef struct grpc_tcp {
 
   grpc_closure* read_cb;
   grpc_closure* write_cb;
-  grpc_slice read_slice;
+
+  /* garbage after the last read */
+  grpc_slice_buffer last_read_buffer;
+
   grpc_slice_buffer* write_slices;
   grpc_slice_buffer* read_slices;
 
@@ -131,6 +135,7 @@ static void tcp_free(grpc_tcp* tcp) {
   grpc_winsocket_destroy(tcp->socket);
   gpr_mu_destroy(&tcp->mu);
   gpr_free(tcp->peer_string);
+  grpc_slice_buffer_destroy_internal(&tcp->last_read_buffer);
   grpc_resource_user_unref(tcp->resource_user);
   if (tcp->shutting_down) GRPC_ERROR_UNREF(tcp->shutdown_error);
   gpr_free(tcp);
@@ -179,9 +184,12 @@ static void on_read(void* tcpp, grpc_error* error) {
   grpc_tcp* tcp = (grpc_tcp*)tcpp;
   grpc_closure* cb = tcp->read_cb;
   grpc_winsocket* socket = tcp->socket;
-  grpc_slice sub;
   grpc_winsocket_callback_info* info = &socket->read_info;
 
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_INFO, "TCP:%p on_read", tcp);
+  }
+
   GRPC_ERROR_REF(error);
 
   if (error == GRPC_ERROR_NONE) {
@@ -189,13 +197,35 @@ static void on_read(void* tcpp, grpc_error* error) {
       char* utf8_message = gpr_format_message(info->wsa_error);
       error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(utf8_message);
       gpr_free(utf8_message);
-      grpc_slice_unref_internal(tcp->read_slice);
+      grpc_slice_buffer_reset_and_unref_internal(tcp->read_slices);
     } else {
       if (info->bytes_transfered != 0 && !tcp->shutting_down) {
-        sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
-        grpc_slice_buffer_add(tcp->read_slices, sub);
+        GPR_ASSERT((size_t)info->bytes_transfered <= tcp->read_slices->length);
+        if (static_cast<size_t>(info->bytes_transfered) !=
+            tcp->read_slices->length) {
+          grpc_slice_buffer_trim_end(
+              tcp->read_slices,
+              tcp->read_slices->length -
+                  static_cast<size_t>(info->bytes_transfered),
+              &tcp->last_read_buffer);
+        }
+        GPR_ASSERT((size_t)info->bytes_transfered == tcp->read_slices->length);
+
+        if (grpc_tcp_trace.enabled()) {
+          size_t i;
+          for (i = 0; i < tcp->read_slices->count; i++) {
+            char* dump = grpc_dump_slice(tcp->read_slices->slices[i],
+                                         GPR_DUMP_HEX | GPR_DUMP_ASCII);
+            gpr_log(GPR_INFO, "READ %p (peer=%s): %s", tcp, tcp->peer_string,
+                    dump);
+            gpr_free(dump);
+          }
+        }
       } else {
-        grpc_slice_unref_internal(tcp->read_slice);
+        if (grpc_tcp_trace.enabled()) {
+          gpr_log(GPR_INFO, "TCP:%p unref read_slice", tcp);
+        }
+        grpc_slice_buffer_reset_and_unref_internal(tcp->read_slices);
         error = tcp->shutting_down
                     ? GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
                           "TCP stream shutting down", &tcp->shutdown_error, 1)
@@ -209,6 +239,8 @@ static void on_read(void* tcpp, grpc_error* error) {
   GRPC_CLOSURE_SCHED(cb, error);
 }
 
+#define DEFAULT_TARGET_READ_SIZE 8192
+#define MAX_WSABUF_COUNT 16
 static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
                      grpc_closure* cb) {
   grpc_tcp* tcp = (grpc_tcp*)ep;
@@ -217,7 +249,12 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
   int status;
   DWORD bytes_read = 0;
   DWORD flags = 0;
-  WSABUF buffer;
+  WSABUF buffers[MAX_WSABUF_COUNT];
+  size_t i;
+
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_INFO, "TCP:%p win_read", tcp);
+  }
 
   if (tcp->shutting_down) {
     GRPC_CLOSURE_SCHED(
@@ -229,18 +266,27 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
   tcp->read_cb = cb;
   tcp->read_slices = read_slices;
   grpc_slice_buffer_reset_and_unref_internal(read_slices);
+  grpc_slice_buffer_swap(read_slices, &tcp->last_read_buffer);
 
-  tcp->read_slice = GRPC_SLICE_MALLOC(8192);
+  if (tcp->read_slices->length < DEFAULT_TARGET_READ_SIZE / 2 &&
+      tcp->read_slices->count < MAX_WSABUF_COUNT) {
+    // TODO(jtattermusch): slice should be allocated using resource quota
+    grpc_slice_buffer_add(tcp->read_slices,
+                          GRPC_SLICE_MALLOC(DEFAULT_TARGET_READ_SIZE));
+  }
 
-  buffer.len = (ULONG)GRPC_SLICE_LENGTH(
-      tcp->read_slice);  // we know slice size fits in 32bit.
-  buffer.buf = (char*)GRPC_SLICE_START_PTR(tcp->read_slice);
+  GPR_ASSERT(tcp->read_slices->count <= MAX_WSABUF_COUNT);
+  for (i = 0; i < tcp->read_slices->count; i++) {
+    buffers[i].len = (ULONG)GRPC_SLICE_LENGTH(
+        tcp->read_slices->slices[i]);  // we know slice size fits in 32bit.
+    buffers[i].buf = (char*)GRPC_SLICE_START_PTR(tcp->read_slices->slices[i]);
+  }
 
   TCP_REF(tcp, "read");
 
   /* First let's try a synchronous, non-blocking read. */
-  status =
-      WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, NULL, NULL);
+  status = WSARecv(tcp->socket->socket, buffers, (DWORD)tcp->read_slices->count,
+                   &bytes_read, &flags, NULL, NULL);
   info->wsa_error = status == 0 ? 0 : WSAGetLastError();
 
   /* Did we get data immediately ? Yay. */
@@ -252,8 +298,8 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
 
   /* Otherwise, let's retry, by queuing a read. */
   memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED));
-  status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
-                   &info->overlapped, NULL);
+  status = WSARecv(tcp->socket->socket, buffers, (DWORD)tcp->read_slices->count,
+                   &bytes_read, &flags, &info->overlapped, NULL);
 
   if (status != 0) {
     int wsa_error = WSAGetLastError();
@@ -275,6 +321,10 @@ static void on_write(void* tcpp, grpc_error* error) {
   grpc_winsocket_callback_info* info = &handle->write_info;
   grpc_closure* cb;
 
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_INFO, "TCP:%p on_write", tcp);
+  }
+
   GRPC_ERROR_REF(error);
 
   gpr_mu_lock(&tcp->mu);
@@ -303,11 +353,21 @@ static void win_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
   unsigned i;
   DWORD bytes_sent;
   int status;
-  WSABUF local_buffers[16];
+  WSABUF local_buffers[MAX_WSABUF_COUNT];
   WSABUF* allocated = NULL;
   WSABUF* buffers = local_buffers;
   size_t len;
 
+  if (grpc_tcp_trace.enabled()) {
+    size_t i;
+    for (i = 0; i < slices->count; i++) {
+      char* data =
+          grpc_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII);
+      gpr_log(GPR_INFO, "WRITE %p (peer=%s): %s", tcp, tcp->peer_string, data);
+      gpr_free(data);
+    }
+  }
+
   if (tcp->shutting_down) {
     GRPC_CLOSURE_SCHED(
         cb, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
@@ -412,6 +472,7 @@ static void win_shutdown(grpc_endpoint* ep, grpc_error* why) {
 static void win_destroy(grpc_endpoint* ep) {
   grpc_network_status_unregister_endpoint(ep);
   grpc_tcp* tcp = (grpc_tcp*)ep;
+  grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer);
   TCP_UNREF(tcp, "destroy");
 }
 
@@ -463,6 +524,7 @@ grpc_endpoint* grpc_tcp_create(grpc_winsocket* socket,
   GRPC_CLOSURE_INIT(&tcp->on_read, on_read, tcp, grpc_schedule_on_exec_ctx);
   GRPC_CLOSURE_INIT(&tcp->on_write, on_write, tcp, grpc_schedule_on_exec_ctx);
   tcp->peer_string = gpr_strdup(peer_string);
+  grpc_slice_buffer_init(&tcp->last_read_buffer);
   tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
   /* Tell network status tracking code about the new endpoint */
   grpc_network_status_register_endpoint(&tcp->base);
diff --git a/src/core/lib/surface/server.cc b/src/core/lib/surface/server.cc
index 67b38e6f0c..7ae6e51a5f 100644
--- a/src/core/lib/surface/server.cc
+++ b/src/core/lib/surface/server.cc
@@ -194,13 +194,10 @@ struct call_data {
 };
 
 struct request_matcher {
-  request_matcher(grpc_server* server);
-  ~request_matcher();
-
   grpc_server* server;
-  std::atomic<call_data*> pending_head{nullptr};
-  call_data* pending_tail = nullptr;
-  gpr_locked_mpscq* requests_per_cq = nullptr;
+  call_data* pending_head;
+  call_data* pending_tail;
+  gpr_locked_mpscq* requests_per_cq;
 };
 
 struct registered_method {
@@ -349,30 +346,22 @@ static void channel_broadcaster_shutdown(channel_broadcaster* cb,
  * request_matcher
  */
 
-namespace {
-request_matcher::request_matcher(grpc_server* server) : server(server) {
-  requests_per_cq = static_cast<gpr_locked_mpscq*>(
-      gpr_malloc(sizeof(*requests_per_cq) * server->cq_count));
-  for (size_t i = 0; i < server->cq_count; i++) {
-    gpr_locked_mpscq_init(&requests_per_cq[i]);
-  }
-}
-
-request_matcher::~request_matcher() {
+static void request_matcher_init(request_matcher* rm, grpc_server* server) {
+  memset(rm, 0, sizeof(*rm));
+  rm->server = server;
+  rm->requests_per_cq = static_cast<gpr_locked_mpscq*>(
+      gpr_malloc(sizeof(*rm->requests_per_cq) * server->cq_count));
   for (size_t i = 0; i < server->cq_count; i++) {
-    GPR_ASSERT(gpr_locked_mpscq_pop(&requests_per_cq[i]) == nullptr);
-    gpr_locked_mpscq_destroy(&requests_per_cq[i]);
+    gpr_locked_mpscq_init(&rm->requests_per_cq[i]);
   }
-  gpr_free(requests_per_cq);
-}
-}  // namespace
-
-static void request_matcher_init(request_matcher* rm, grpc_server* server) {
-  new (rm) request_matcher(server);
 }
 
 static void request_matcher_destroy(request_matcher* rm) {
-  rm->~request_matcher();
+  for (size_t i = 0; i < rm->server->cq_count; i++) {
+    GPR_ASSERT(gpr_locked_mpscq_pop(&rm->requests_per_cq[i]) == nullptr);
+    gpr_locked_mpscq_destroy(&rm->requests_per_cq[i]);
+  }
+  gpr_free(rm->requests_per_cq);
 }
 
 static void kill_zombie(void* elem, grpc_error* error) {
@@ -381,10 +370,9 @@ static void kill_zombie(void* elem, grpc_error* error) {
 }
 
 static void request_matcher_zombify_all_pending_calls(request_matcher* rm) {
-  call_data* calld;
-  while ((calld = rm->pending_head.load(std::memory_order_relaxed)) !=
-         nullptr) {
-    rm->pending_head.store(calld->pending_next, std::memory_order_relaxed);
+  while (rm->pending_head) {
+    call_data* calld = rm->pending_head;
+    rm->pending_head = calld->pending_next;
     gpr_atm_no_barrier_store(&calld->state, ZOMBIED);
     GRPC_CLOSURE_INIT(
         &calld->kill_zombie_closure, kill_zombie,
@@ -582,9 +570,8 @@ static void publish_new_rpc(void* arg, grpc_error* error) {
   }
 
   gpr_atm_no_barrier_store(&calld->state, PENDING);
-  if (rm->pending_head.load(std::memory_order_relaxed) == nullptr) {
-    rm->pending_head.store(calld, std::memory_order_relaxed);
-    rm->pending_tail = calld;
+  if (rm->pending_head == nullptr) {
+    rm->pending_tail = rm->pending_head = calld;
   } else {
     rm->pending_tail->pending_next = calld;
     rm->pending_tail = calld;
@@ -1448,39 +1435,30 @@ static grpc_call_error queue_call_request(grpc_server* server, size_t cq_idx,
       rm = &rc->data.registered.method->matcher;
       break;
   }
-
-  // Fast path: if there is no pending request to be processed, immediately
-  // return.
-  if (!gpr_locked_mpscq_push(&rm->requests_per_cq[cq_idx], &rc->request_link) ||
-      // Note: We are reading the pending_head without holding the server's call
-      //       mutex. Even if we read a non-null value here due to reordering,
-      //       we will check it below again after grabbing the lock.
-      rm->pending_head.load(std::memory_order_relaxed) == nullptr) {
-    return GRPC_CALL_OK;
-  }
-  // Slow path: This was the first queued request and there are pendings:
-  //            We need to lock and start matching calls.
-  gpr_mu_lock(&server->mu_call);
-  while ((calld = rm->pending_head.load(std::memory_order_relaxed)) !=
-         nullptr) {
-    rc = reinterpret_cast<requested_call*>(
-        gpr_locked_mpscq_pop(&rm->requests_per_cq[cq_idx]));
-    if (rc == nullptr) break;
-    rm->pending_head.store(calld->pending_next, std::memory_order_relaxed);
-    gpr_mu_unlock(&server->mu_call);
-    if (!gpr_atm_full_cas(&calld->state, PENDING, ACTIVATED)) {
-      // Zombied Call
-      GRPC_CLOSURE_INIT(
-          &calld->kill_zombie_closure, kill_zombie,
-          grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0),
-          grpc_schedule_on_exec_ctx);
-      GRPC_CLOSURE_SCHED(&calld->kill_zombie_closure, GRPC_ERROR_NONE);
-    } else {
-      publish_call(server, calld, cq_idx, rc);
-    }
+  if (gpr_locked_mpscq_push(&rm->requests_per_cq[cq_idx], &rc->request_link)) {
+    /* this was the first queued request: we need to lock and start
+       matching calls */
     gpr_mu_lock(&server->mu_call);
+    while ((calld = rm->pending_head) != nullptr) {
+      rc = reinterpret_cast<requested_call*>(
+          gpr_locked_mpscq_pop(&rm->requests_per_cq[cq_idx]));
+      if (rc == nullptr) break;
+      rm->pending_head = calld->pending_next;
+      gpr_mu_unlock(&server->mu_call);
+      if (!gpr_atm_full_cas(&calld->state, PENDING, ACTIVATED)) {
+        // Zombied Call
+        GRPC_CLOSURE_INIT(
+            &calld->kill_zombie_closure, kill_zombie,
+            grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0),
+            grpc_schedule_on_exec_ctx);
+        GRPC_CLOSURE_SCHED(&calld->kill_zombie_closure, GRPC_ERROR_NONE);
+      } else {
+        publish_call(server, calld, cq_idx, rc);
+      }
+      gpr_mu_lock(&server->mu_call);
+    }
+    gpr_mu_unlock(&server->mu_call);
   }
-  gpr_mu_unlock(&server->mu_call);
   return GRPC_CALL_OK;
 }
 
diff --git a/src/core/lib/transport/metadata.cc b/src/core/lib/transport/metadata.cc
index 60af22393e..30482a1b3b 100644
--- a/src/core/lib/transport/metadata.cc
+++ b/src/core/lib/transport/metadata.cc
@@ -187,6 +187,7 @@ static void gc_mdtab(mdtab_shard* shard) {
           ((destroy_user_data_func)gpr_atm_no_barrier_load(
               &md->destroy_user_data))(user_data);
         }
+        gpr_mu_destroy(&md->mu_user_data);
         gpr_free(md);
         *prev_next = next;
         num_freed++;
diff --git a/src/python/grpcio/grpc/__init__.py b/src/python/grpcio/grpc/__init__.py
index daf869b156..70d7618e05 100644
--- a/src/python/grpcio/grpc/__init__.py
+++ b/src/python/grpcio/grpc/__init__.py
@@ -23,6 +23,11 @@ from grpc._cython import cygrpc as _cygrpc
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())
 
+try:
+    from ._grpcio_metadata import __version__
+except ImportError:
+    __version__ = "dev0"
+
 ############################## Future Interface  ###############################
 
 
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
index e0e068e452..01b8237484 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
@@ -28,19 +28,22 @@ cdef tuple _wrap_grpc_arg(grpc_arg arg)
 cdef grpc_arg _unwrap_grpc_arg(tuple wrapped_arg)
 
 
-cdef class _ArgumentProcessor:
+cdef class _ChannelArg:
 
   cdef grpc_arg c_argument
 
   cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except *
 
 
-cdef class _ArgumentsProcessor:
+cdef class _ChannelArgs:
 
   cdef readonly tuple _arguments
-  cdef list _argument_processors
+  cdef list _channel_args
   cdef readonly list _references
   cdef grpc_channel_args _c_arguments
 
-  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except *
-  cdef un_c(self)
+  cdef void _c(self, grpc_arg_pointer_vtable *vtable) except *
+  cdef grpc_channel_args *c_args(self) except *
+
+  @staticmethod
+  cdef _ChannelArgs from_args(object arguments, grpc_arg_pointer_vtable * vtable)
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
index b7a4277ff6..bf12871015 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
@@ -50,7 +50,7 @@ cdef grpc_arg _unwrap_grpc_arg(tuple wrapped_arg):
   return wrapped.arg
 
 
-cdef class _ArgumentProcessor:
+cdef class _ChannelArg:
 
   cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except *:
     key, value = argument
@@ -82,27 +82,34 @@ cdef class _ArgumentProcessor:
           'Expected int, bytes, or behavior, got {}'.format(type(value)))
 
 
-cdef class _ArgumentsProcessor:
+cdef class _ChannelArgs:
 
   def __cinit__(self, arguments):
     self._arguments = () if arguments is None else tuple(arguments)
-    self._argument_processors = []
+    self._channel_args = []
     self._references = []
+    self._c_arguments.arguments = NULL
 
-  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except *:
+  cdef void _c(self, grpc_arg_pointer_vtable *vtable) except *:
     self._c_arguments.arguments_length = len(self._arguments)
-    if self._c_arguments.arguments_length == 0:
-      return NULL
-    else:
+    if self._c_arguments.arguments_length != 0:
       self._c_arguments.arguments = <grpc_arg *>gpr_malloc(
           self._c_arguments.arguments_length * sizeof(grpc_arg))
       for index, argument in enumerate(self._arguments):
-        argument_processor = _ArgumentProcessor()
-        argument_processor.c(argument, vtable, self._references)
-        self._c_arguments.arguments[index] = argument_processor.c_argument
-        self._argument_processors.append(argument_processor)
-      return &self._c_arguments
-
-  cdef un_c(self):
-    if self._arguments:
+        channel_arg = _ChannelArg()
+        channel_arg.c(argument, vtable, self._references)
+        self._c_arguments.arguments[index] = channel_arg.c_argument
+        self._channel_args.append(channel_arg)
+
+  cdef grpc_channel_args *c_args(self) except *:
+    return &self._c_arguments
+
+  def __dealloc__(self):
+    if self._c_arguments.arguments != NULL:
       gpr_free(self._c_arguments.arguments)
+
+  @staticmethod
+  cdef _ChannelArgs from_args(object arguments, grpc_arg_pointer_vtable * vtable):
+    cdef _ChannelArgs channel_args = _ChannelArgs(arguments)
+    channel_args._c(vtable)
+    return channel_args
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi
index 135d224095..70d4abb730 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi
@@ -423,16 +423,15 @@ cdef class Channel:
     self._vtable.copy = &_copy_pointer
     self._vtable.destroy = &_destroy_pointer
     self._vtable.cmp = &_compare_pointer
-    cdef _ArgumentsProcessor arguments_processor = _ArgumentsProcessor(
-        arguments)
-    cdef grpc_channel_args *c_arguments = arguments_processor.c(&self._vtable)
+    cdef _ChannelArgs channel_args = _ChannelArgs.from_args(
+        arguments, &self._vtable)
     if channel_credentials is None:
       self._state.c_channel = grpc_insecure_channel_create(
-          <char *>target, c_arguments, NULL)
+          <char *>target, channel_args.c_args(), NULL)
     else:
       c_channel_credentials = channel_credentials.c()
       self._state.c_channel = grpc_secure_channel_create(
-          c_channel_credentials, <char *>target, c_arguments, NULL)
+          c_channel_credentials, <char *>target, channel_args.c_args(), NULL)
       grpc_channel_credentials_release(c_channel_credentials)
     self._state.c_call_completion_queue = (
         grpc_completion_queue_create_for_next(NULL))
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi
index 52cfccb677..4a6fbe0f96 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi
@@ -16,7 +16,6 @@
 cdef class Server:
 
   cdef grpc_arg_pointer_vtable _vtable
-  cdef readonly _ArgumentsProcessor _arguments_processor
   cdef grpc_server *c_server
   cdef bint is_started  # start has been called
   cdef bint is_shutting_down  # shutdown has been called
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
index e89e02b171..d72648a35d 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
@@ -29,11 +29,9 @@ cdef class Server:
     self._vtable.copy = &_copy_pointer
     self._vtable.destroy = &_destroy_pointer
     self._vtable.cmp = &_compare_pointer
-    cdef _ArgumentsProcessor arguments_processor = _ArgumentsProcessor(
-        arguments)
-    cdef grpc_channel_args *c_arguments = arguments_processor.c(&self._vtable)
-    self.c_server = grpc_server_create(c_arguments, NULL)
-    arguments_processor.un_c()
+    cdef _ChannelArgs channel_args = _ChannelArgs.from_args(
+        arguments, &self._vtable)
+    self.c_server = grpc_server_create(channel_args.c_args(), NULL)
     self.references.append(arguments)
     self.is_started = False
     self.is_shutting_down = False
diff --git a/src/python/grpcio_health_checking/grpc_health/v1/health.py b/src/python/grpcio_health_checking/grpc_health/v1/health.py
index 0583659428..0a5bbb5504 100644
--- a/src/python/grpcio_health_checking/grpc_health/v1/health.py
+++ b/src/python/grpcio_health_checking/grpc_health/v1/health.py
@@ -23,15 +23,61 @@ from grpc_health.v1 import health_pb2_grpc as _health_pb2_grpc
 SERVICE_NAME = _health_pb2.DESCRIPTOR.services_by_name['Health'].full_name
 
 
+class _Watcher():
+
+    def __init__(self):
+        self._condition = threading.Condition()
+        self._responses = list()
+        self._open = True
+
+    def __iter__(self):
+        return self
+
+    def _next(self):
+        with self._condition:
+            while not self._responses and self._open:
+                self._condition.wait()
+            if self._responses:
+                return self._responses.pop(0)
+            else:
+                raise StopIteration()
+
+    def next(self):
+        return self._next()
+
+    def __next__(self):
+        return self._next()
+
+    def add(self, response):
+        with self._condition:
+            self._responses.append(response)
+            self._condition.notify()
+
+    def close(self):
+        with self._condition:
+            self._open = False
+            self._condition.notify()
+
+
 class HealthServicer(_health_pb2_grpc.HealthServicer):
     """Servicer handling RPCs for service statuses."""
 
     def __init__(self):
-        self._server_status_lock = threading.Lock()
+        self._lock = threading.RLock()
         self._server_status = {}
+        self._watchers = {}
+
+    def _on_close_callback(self, watcher, service):
+
+        def callback():
+            with self._lock:
+                self._watchers[service].remove(watcher)
+            watcher.close()
+
+        return callback
 
     def Check(self, request, context):
-        with self._server_status_lock:
+        with self._lock:
             status = self._server_status.get(request.service)
             if status is None:
                 context.set_code(grpc.StatusCode.NOT_FOUND)
@@ -39,14 +85,30 @@ class HealthServicer(_health_pb2_grpc.HealthServicer):
             else:
                 return _health_pb2.HealthCheckResponse(status=status)
 
+    def Watch(self, request, context):
+        service = request.service
+        with self._lock:
+            status = self._server_status.get(service)
+            if status is None:
+                status = _health_pb2.HealthCheckResponse.SERVICE_UNKNOWN  # pylint: disable=no-member
+            watcher = _Watcher()
+            watcher.add(_health_pb2.HealthCheckResponse(status=status))
+            if service not in self._watchers:
+                self._watchers[service] = set()
+            self._watchers[service].add(watcher)
+            context.add_callback(self._on_close_callback(watcher, service))
+        return watcher
+
     def set(self, service, status):
         """Sets the status of a service.
 
-    Args:
-        service: string, the name of the service.
-            NOTE, '' must be set.
-        status: HealthCheckResponse.status enum value indicating
-            the status of the service
-    """
-        with self._server_status_lock:
+        Args:
+          service: string, the name of the service. NOTE, '' must be set.
+          status: HealthCheckResponse.status enum value indicating the status of
+            the service
+        """
+        with self._lock:
             self._server_status[service] = status
+            if service in self._watchers:
+                for watcher in self._watchers[service]:
+                    watcher.add(_health_pb2.HealthCheckResponse(status=status))
diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py
index d5327711d3..582ce898de 100644
--- a/src/python/grpcio_tests/commands.py
+++ b/src/python/grpcio_tests/commands.py
@@ -141,6 +141,7 @@ class TestGevent(setuptools.Command):
         'unit._exit_test.ExitTest.test_in_flight_partial_unary_stream_call',
         'unit._exit_test.ExitTest.test_in_flight_partial_stream_unary_call',
         'unit._exit_test.ExitTest.test_in_flight_partial_stream_stream_call',
+        'health_check._health_servicer_test.HealthServicerTest.test_cancelled_watch_removed_from_watch_list',
         # TODO(https://github.com/grpc/grpc/issues/17330) enable these three tests
         'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels',
         'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels_and_sockets',
diff --git a/src/python/grpcio_tests/tests/health_check/BUILD.bazel b/src/python/grpcio_tests/tests/health_check/BUILD.bazel
index 19e1e1b2e1..77bc61aa30 100644
--- a/src/python/grpcio_tests/tests/health_check/BUILD.bazel
+++ b/src/python/grpcio_tests/tests/health_check/BUILD.bazel
@@ -9,6 +9,7 @@ py_test(
         "//src/python/grpcio/grpc:grpcio",
         "//src/python/grpcio_health_checking/grpc_health/v1:grpc_health",
         "//src/python/grpcio_tests/tests/unit:test_common",
+        "//src/python/grpcio_tests/tests/unit/framework/common:common",
     ],
     imports = ["../../",],
 )
diff --git a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
index c1d9436c2f..35794987bc 100644
--- a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
+++ b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
@@ -13,6 +13,8 @@
 # limitations under the License.
 """Tests of grpc_health.v1.health."""
 
+import threading
+import time
 import unittest
 
 import grpc
@@ -21,22 +23,36 @@ from grpc_health.v1 import health_pb2
 from grpc_health.v1 import health_pb2_grpc
 
 from tests.unit import test_common
+from tests.unit.framework.common import test_constants
+
+from six.moves import queue
+
+_SERVING_SERVICE = 'grpc.test.TestServiceServing'
+_UNKNOWN_SERVICE = 'grpc.test.TestServiceUnknown'
+_NOT_SERVING_SERVICE = 'grpc.test.TestServiceNotServing'
+_WATCH_SERVICE = 'grpc.test.WatchService'
+
+
+def _consume_responses(response_iterator, response_queue):
+    for response in response_iterator:
+        response_queue.put(response)
 
 
 class HealthServicerTest(unittest.TestCase):
 
     def setUp(self):
-        servicer = health.HealthServicer()
-        servicer.set('', health_pb2.HealthCheckResponse.SERVING)
-        servicer.set('grpc.test.TestServiceServing',
-                     health_pb2.HealthCheckResponse.SERVING)
-        servicer.set('grpc.test.TestServiceUnknown',
-                     health_pb2.HealthCheckResponse.UNKNOWN)
-        servicer.set('grpc.test.TestServiceNotServing',
-                     health_pb2.HealthCheckResponse.NOT_SERVING)
+        self._servicer = health.HealthServicer()
+        self._servicer.set('', health_pb2.HealthCheckResponse.SERVING)
+        self._servicer.set(_SERVING_SERVICE,
+                           health_pb2.HealthCheckResponse.SERVING)
+        self._servicer.set(_UNKNOWN_SERVICE,
+                           health_pb2.HealthCheckResponse.UNKNOWN)
+        self._servicer.set(_NOT_SERVING_SERVICE,
+                           health_pb2.HealthCheckResponse.NOT_SERVING)
         self._server = test_common.test_server()
         port = self._server.add_insecure_port('[::]:0')
-        health_pb2_grpc.add_HealthServicer_to_server(servicer, self._server)
+        health_pb2_grpc.add_HealthServicer_to_server(self._servicer,
+                                                     self._server)
         self._server.start()
 
         self._channel = grpc.insecure_channel('localhost:%d' % port)
@@ -46,37 +62,160 @@ class HealthServicerTest(unittest.TestCase):
         self._server.stop(None)
         self._channel.close()
 
-    def test_empty_service(self):
+    def test_check_empty_service(self):
         request = health_pb2.HealthCheckRequest()
         resp = self._stub.Check(request)
         self.assertEqual(health_pb2.HealthCheckResponse.SERVING, resp.status)
 
-    def test_serving_service(self):
-        request = health_pb2.HealthCheckRequest(
-            service='grpc.test.TestServiceServing')
+    def test_check_serving_service(self):
+        request = health_pb2.HealthCheckRequest(service=_SERVING_SERVICE)
         resp = self._stub.Check(request)
         self.assertEqual(health_pb2.HealthCheckResponse.SERVING, resp.status)
 
-    def test_unknown_serivce(self):
-        request = health_pb2.HealthCheckRequest(
-            service='grpc.test.TestServiceUnknown')
+    def test_check_unknown_serivce(self):
+        request = health_pb2.HealthCheckRequest(service=_UNKNOWN_SERVICE)
         resp = self._stub.Check(request)
         self.assertEqual(health_pb2.HealthCheckResponse.UNKNOWN, resp.status)
 
-    def test_not_serving_service(self):
-        request = health_pb2.HealthCheckRequest(
-            service='grpc.test.TestServiceNotServing')
+    def test_check_not_serving_service(self):
+        request = health_pb2.HealthCheckRequest(service=_NOT_SERVING_SERVICE)
         resp = self._stub.Check(request)
         self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING,
                          resp.status)
 
-    def test_not_found_service(self):
+    def test_check_not_found_service(self):
         request = health_pb2.HealthCheckRequest(service='not-found')
         with self.assertRaises(grpc.RpcError) as context:
             resp = self._stub.Check(request)
 
         self.assertEqual(grpc.StatusCode.NOT_FOUND, context.exception.code())
 
+    def test_watch_empty_service(self):
+        request = health_pb2.HealthCheckRequest(service='')
+        response_queue = queue.Queue()
+        rendezvous = self._stub.Watch(request)
+        thread = threading.Thread(
+            target=_consume_responses, args=(rendezvous, response_queue))
+        thread.start()
+
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
+                         response.status)
+
+        rendezvous.cancel()
+        thread.join()
+        self.assertTrue(response_queue.empty())
+
+    def test_watch_new_service(self):
+        request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
+        response_queue = queue.Queue()
+        rendezvous = self._stub.Watch(request)
+        thread = threading.Thread(
+            target=_consume_responses, args=(rendezvous, response_queue))
+        thread.start()
+
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
+                         response.status)
+
+        self._servicer.set(_WATCH_SERVICE,
+                           health_pb2.HealthCheckResponse.SERVING)
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
+                         response.status)
+
+        self._servicer.set(_WATCH_SERVICE,
+                           health_pb2.HealthCheckResponse.NOT_SERVING)
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING,
+                         response.status)
+
+        rendezvous.cancel()
+        thread.join()
+        self.assertTrue(response_queue.empty())
+
+    def test_watch_service_isolation(self):
+        request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
+        response_queue = queue.Queue()
+        rendezvous = self._stub.Watch(request)
+        thread = threading.Thread(
+            target=_consume_responses, args=(rendezvous, response_queue))
+        thread.start()
+
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
+                         response.status)
+
+        self._servicer.set('some-other-service',
+                           health_pb2.HealthCheckResponse.SERVING)
+        with self.assertRaises(queue.Empty):
+            response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+
+        rendezvous.cancel()
+        thread.join()
+        self.assertTrue(response_queue.empty())
+
+    def test_two_watchers(self):
+        request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
+        response_queue1 = queue.Queue()
+        response_queue2 = queue.Queue()
+        rendezvous1 = self._stub.Watch(request)
+        rendezvous2 = self._stub.Watch(request)
+        thread1 = threading.Thread(
+            target=_consume_responses, args=(rendezvous1, response_queue1))
+        thread2 = threading.Thread(
+            target=_consume_responses, args=(rendezvous2, response_queue2))
+        thread1.start()
+        thread2.start()
+
+        response1 = response_queue1.get(timeout=test_constants.SHORT_TIMEOUT)
+        response2 = response_queue2.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
+                         response1.status)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
+                         response2.status)
+
+        self._servicer.set(_WATCH_SERVICE,
+                           health_pb2.HealthCheckResponse.SERVING)
+        response1 = response_queue1.get(timeout=test_constants.SHORT_TIMEOUT)
+        response2 = response_queue2.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
+                         response1.status)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
+                         response2.status)
+
+        rendezvous1.cancel()
+        rendezvous2.cancel()
+        thread1.join()
+        thread2.join()
+        self.assertTrue(response_queue1.empty())
+        self.assertTrue(response_queue2.empty())
+
+    def test_cancelled_watch_removed_from_watch_list(self):
+        request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
+        response_queue = queue.Queue()
+        rendezvous = self._stub.Watch(request)
+        thread = threading.Thread(
+            target=_consume_responses, args=(rendezvous, response_queue))
+        thread.start()
+
+        response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+        self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
+                         response.status)
+
+        rendezvous.cancel()
+        self._servicer.set(_WATCH_SERVICE,
+                           health_pb2.HealthCheckResponse.SERVING)
+        thread.join()
+
+        # Wait, if necessary, for serving thread to process client cancellation
+        timeout = time.time() + test_constants.SHORT_TIMEOUT
+        while time.time() < timeout and self._servicer._watchers[_WATCH_SERVICE]:
+            time.sleep(1)
+        self.assertFalse(self._servicer._watchers[_WATCH_SERVICE],
+                         'watch set should be empty')
+        self.assertTrue(response_queue.empty())
+
     def test_health_service_name(self):
         self.assertEqual(health.SERVICE_NAME, 'grpc.health.v1.Health')
 
diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json
index f202a3f932..de4c2c1fdd 100644
--- a/src/python/grpcio_tests/tests/tests.json
+++ b/src/python/grpcio_tests/tests/tests.json
@@ -64,6 +64,7 @@
   "unit._server_ssl_cert_config_test.ServerSSLCertReloadTestWithoutClientAuth",
   "unit._server_test.ServerTest",
   "unit._session_cache_test.SSLSessionCacheTest",
+  "unit._version_test.VersionTest",
   "unit.beta._beta_features_test.BetaFeaturesTest",
   "unit.beta._beta_features_test.ContextManagementAndLifecycleTest",
   "unit.beta._connectivity_channel_test.ConnectivityStatesTest",
diff --git a/src/python/grpcio_tests/tests/unit/BUILD.bazel b/src/python/grpcio_tests/tests/unit/BUILD.bazel
index 1b462ec67a..a9bcd9f304 100644
--- a/src/python/grpcio_tests/tests/unit/BUILD.bazel
+++ b/src/python/grpcio_tests/tests/unit/BUILD.bazel
@@ -7,6 +7,7 @@ GRPCIO_TESTS_UNIT = [
     "_api_test.py",
     "_auth_context_test.py",
     "_auth_test.py",
+    "_version_test.py",
     "_channel_args_test.py",
     "_channel_close_test.py",
     "_channel_connectivity_test.py",
diff --git a/src/python/grpcio_tests/tests/unit/_version_test.py b/src/python/grpcio_tests/tests/unit/_version_test.py
new file mode 100644
index 0000000000..3d37b319e5
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_version_test.py
@@ -0,0 +1,30 @@
+# Copyright 2018 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Test for grpc.__version__"""
+
+import unittest
+import grpc
+import logging
+from grpc import _grpcio_metadata
+
+
+class VersionTest(unittest.TestCase):
+
+    def test_get_version(self):
+        self.assertEqual(grpc.__version__, _grpcio_metadata.__version__)
+
+
+if __name__ == '__main__':
+    logging.basicConfig()
+    unittest.main(verbosity=2)