Merge github.com:grpc/grpc into c++api

INSTALL

@@ -90,6 +90,11 @@ these dependencies this way:
 
   # apt-get install autoconf libtool
 
+If you want to run the tests using one of the sanitized configurations, you
+will need clang and its instrumented libc++:
+
+  # apt-get install clang libc++-dev
+
 
 A word on OpenSSL
 -----------------

Makefile

@@ -66,13 +66,13 @@ DEFINES_asan = NDEBUG
 VALID_CONFIG_msan = 1
 REQUIRE_CUSTOM_LIBRARIES_msan = 1
 CC_msan = clang
-CXX_msan = clang++
+CXX_msan = clang++-libc++
 LD_msan = clang
-LDXX_msan = clang++
-CPPFLAGS_msan = -O1 -fsanitize=memory -fno-omit-frame-pointer
+LDXX_msan = clang++-libc++
+CPPFLAGS_msan = -O1 -fsanitize=memory -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1
 OPENSSL_CFLAGS_msan = -DPURIFY
 OPENSSL_CONFIG_msan = no-asm
-LDFLAGS_msan = -fsanitize=memory
+LDFLAGS_msan = -fsanitize=memory -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1
 DEFINES_msan = NDEBUG
 
 VALID_CONFIG_ubsan = 1

include/grpc/grpc.h

@@ -44,7 +44,7 @@
 extern "C" {
 #endif
 
-/* Completion Channels enable notification of the completion of asynchronous
+/* Completion Queues enable notification of the completion of asynchronous
    actions. */
 typedef struct grpc_completion_queue grpc_completion_queue;
 
@@ -156,7 +156,8 @@ typedef enum grpc_op_error {
 struct grpc_byte_buffer;
 typedef struct grpc_byte_buffer grpc_byte_buffer;
 
-/* Sample helpers to obtain byte buffers (these will certainly move place */
+/* Sample helpers to obtain byte buffers (these will certainly move
+   someplace else) */
 grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices);
 grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb);
 size_t grpc_byte_buffer_length(grpc_byte_buffer *bb);

include/grpc/support/port_platform.h

@@ -37,10 +37,6 @@
 /* Override this file with one for your platform if you need to redefine
    things.  */
 
-/* For a common case, assume that the platform has a C99-like stdint.h */
-
-#include <stdint.h>
-
 #if !defined(GPR_NO_AUTODETECT_PLATFORM)
 #if defined(_WIN64) || defined(WIN64)
 #define GPR_WIN32 1
@@ -70,20 +66,40 @@
 #define GPR_POSIX_TIME 1
 #define GPR_GETPID_IN_UNISTD_H 1
 #elif defined(__linux__)
+#ifndef _BSD_SOURCE
+#define _BSD_SOURCE
+#endif
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
 #include <features.h>
 #define GPR_CPU_LINUX 1
 #define GPR_GCC_ATOMIC 1
 #define GPR_LINUX 1
 #define GPR_LINUX_MULTIPOLL_WITH_EPOLL 1
 #define GPR_POSIX_WAKEUP_FD 1
-#define GPR_LINUX_EVENTFD 1
 #define GPR_POSIX_SOCKET 1
 #define GPR_POSIX_SOCKETADDR 1
 #ifdef __GLIBC_PREREQ
+#if __GLIBC_PREREQ(2, 9)
+#define GPR_LINUX_EVENTFD 1
+#endif
+#if __GLIBC_PREREQ(2, 10)
+#define GPR_LINUX_SOCKETUTILS 1
+#endif
 #if __GLIBC_PREREQ(2, 17)
 #define GPR_LINUX_ENV 1
 #endif
 #endif
+#ifndef GPR_LINUX_EVENTFD
+#define GPR_POSIX_NO_SPECIAL_WAKEUP_FD 1
+#endif
+#ifndef GPR_LINUX_SOCKETUTILS
+#define GPR_POSIX_SOCKETUTILS
+#endif
 #ifndef GPR_LINUX_ENV
 #define GPR_POSIX_ENV 1
 #endif
@@ -98,6 +114,9 @@
 #define GPR_ARCH_32 1
 #endif /* _LP64 */
 #elif defined(__APPLE__)
+#ifndef _BSD_SOURCE
+#define _BSD_SOURCE
+#endif
 #define GPR_CPU_POSIX 1
 #define GPR_GCC_ATOMIC 1
 #define GPR_POSIX_LOG 1
@@ -123,6 +142,10 @@
 #endif
 #endif /* GPR_NO_AUTODETECT_PLATFORM */
 
+/* For a common case, assume that the platform has a C99-like stdint.h */
+
+#include <stdint.h>
+
 /* Cache line alignment */
 #ifndef GPR_CACHELINE_SIZE
 #if defined(__i386__) || defined(__x86_64__)

src/core/iomgr/socket_utils_linux.c

@@ -31,12 +31,9 @@
  *
  */
 
-#ifndef _GNU_SOURCE
-#define _GNU_SOURCE
-#endif
 #include <grpc/support/port_platform.h>
 
-#ifdef GPR_LINUX
+#ifdef GPR_LINUX_SOCKETUTILS
 
 #include "src/core/iomgr/socket_utils_posix.h"
 

src/core/support/file_posix.c

@@ -31,19 +31,6 @@
  *
  */
 
-/* Posix code for gpr fdopen and mkstemp support. */
-
-#if !defined _POSIX_C_SOURCE || _POSIX_C_SOURCE < 200112L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 200112L
-#endif
-
-/* Don't know why I have to do this for mkstemp, looks like _POSIX_C_SOURCE
-   should be enough... */
-#ifndef _BSD_SOURCE
-#define _BSD_SOURCE
-#endif
-
 #include <grpc/support/port_platform.h>
 
 #ifdef GPR_POSIX_FILE

src/core/support/log_posix.c

@@ -31,16 +31,6 @@
  *
  */
 
-#if !defined _POSIX_C_SOURCE || _POSIX_C_SOURCE < 200112L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 200112L
-#endif
-
-/* FIXME: "posix" files probably shouldn't depend on _GNU_SOURCE */
-#ifndef _GNU_SOURCE
-#define _GNU_SOURCE
-#endif
-
 #include <grpc/support/port_platform.h>
 
 #if defined(GPR_POSIX_LOG)

src/core/support/string_posix.c

@@ -31,13 +31,6 @@
  *
  */
 
-/* Posix code for gpr snprintf support. */
-
-#if !defined _POSIX_C_SOURCE || _POSIX_C_SOURCE < 200112L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 200112L
-#endif
-
 #include <grpc/support/port_platform.h>
 
 #ifdef GPR_POSIX_STRING

src/core/support/sync_posix.c

@@ -31,13 +31,6 @@
  *
  */
 
-/* Posix gpr synchroization support code. */
-
-#if !defined _POSIX_C_SOURCE || _POSIX_C_SOURCE < 199309L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 199309L
-#endif
-
 #include <grpc/support/port_platform.h>
 
 #ifdef GPR_POSIX_SYNC

src/core/support/time_posix.c

@@ -31,14 +31,6 @@
  *
  */
 
-/* Posix code for gpr time support. */
-
-/* So we get nanosleep and clock_* */
-#if !defined _POSIX_C_SOURCE || _POSIX_C_SOURCE < 199309L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 199309L
-#endif
-
 #include <grpc/support/port_platform.h>
 
 #ifdef GPR_POSIX_TIME
@@ -70,7 +62,9 @@ gpr_timespec gpr_now(void) {
 }
 #else
 /* For some reason Apple's OSes haven't implemented clock_gettime. */
-/* TODO(klempner): Add special handling for Apple. */
+
+#include <sys/time.h>
+
 gpr_timespec gpr_now(void) {
   gpr_timespec now;
   struct timeval now_tv;

src/python/src/_framework/face/_calls.py

@@ -29,6 +29,7 @@
 
 """Utility functions for invoking RPCs."""
 
+import sys
 import threading
 
 from _framework.base import interfaces as base_interfaces
@@ -79,20 +80,46 @@ def _stream_event_subscription(result_consumer, abortion_callback):
       _EventServicedIngestor(result_consumer, abortion_callback))
 
 
+# NOTE(nathaniel): This class has some extremely special semantics around
+# cancellation that allow it to be used by both "blocking" APIs and "futures"
+# APIs.
+#
+# Since futures.Future defines its own exception for cancellation, we want these
+# objects, when returned by methods of a returning-Futures-from-other-methods
+# object, to raise the same exception for cancellation. But that's weird in a
+# blocking API - why should this object, also returned by methods of blocking
+# APIs, raise exceptions from the "future" module? Should we do something like
+# have this class be parameterized by the type of exception that it raises in
+# cancellation circumstances?
+#
+# We don't have to take such a dramatic step: since blocking APIs define no
+# cancellation semantics whatsoever, there is no supported way for
+# blocking-API-users of these objects to cancel RPCs, and thus no supported way
+# for them to see an exception the type of which would be weird to them.
+#
+# Bonus: in both blocking and futures APIs, this object still properly raises
+# exceptions.CancellationError for any *server-side cancellation* of an RPC.
 class _OperationCancellableIterator(interfaces.CancellableIterator):
   """An interfaces.CancellableIterator for response-streaming operations."""
 
   def __init__(self, rendezvous, operation):
+    self._lock = threading.Lock()
     self._rendezvous = rendezvous
     self._operation = operation
+    self._cancelled = False
 
   def __iter__(self):
     return self
 
   def next(self):
+    with self._lock:
+      if self._cancelled:
+        raise future.CancelledError()
     return next(self._rendezvous)
 
   def cancel(self):
+    with self._lock:
+      self._cancelled = True
     self._operation.cancel()
     self._rendezvous.set_outcome(base_interfaces.Outcome.CANCELLED)
 
@@ -105,46 +132,126 @@ class _OperationFuture(future.Future):
     self._rendezvous = rendezvous
     self._operation = operation
 
-    self._outcome = None
+    self._cancelled = False
+    self._computed = False
+    self._payload = None
+    self._exception = None
+    self._traceback = None
     self._callbacks = []
 
   def cancel(self):
     """See future.Future.cancel for specification."""
     with self._condition:
-      if self._outcome is None:
+      if not self._cancelled and not self._computed:
         self._operation.cancel()
-        self._outcome = future.aborted()
+        self._cancelled = True
         self._condition.notify_all()
     return False
 
   def cancelled(self):
     """See future.Future.cancelled for specification."""
-    return False
+    with self._condition:
+      return self._cancelled
+
+  def running(self):
+    """See future.Future.running for specification."""
+    with self._condition:
+      return not self._cancelled and not self._computed
 
   def done(self):
     """See future.Future.done for specification."""
     with self._condition:
-      return (self._outcome is not None and
-              self._outcome.category is not future.ABORTED)
+      return self._cancelled or self._computed
 
-  def outcome(self):
-    """See future.Future.outcome for specification."""
+  def result(self, timeout=None):
+    """See future.Future.result for specification."""
     with self._condition:
-      while self._outcome is None:
-        self._condition.wait()
-      return self._outcome
+      if self._cancelled:
+        raise future.CancelledError()
+      if self._computed:
+        if self._payload is None:
+          raise self._exception  # pylint: disable=raising-bad-type
+        else:
+          return self._payload
+
+      condition = threading.Condition()
+      def notify_condition(unused_future):
+        with condition:
+          condition.notify()
+      self._callbacks.append(notify_condition)
+
+    with condition:
+      condition.wait(timeout=timeout)
+
+    with self._condition:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        if self._payload is None:
+          raise self._exception  # pylint: disable=raising-bad-type
+        else:
+          return self._payload
+      else:
+        raise future.TimeoutError()
+
+  def exception(self, timeout=None):
+    """See future.Future.exception for specification."""
+    with self._condition:
+      if self._cancelled:
+        raise future.CancelledError()
+      if self._computed:
+        return self._exception
+
+      condition = threading.Condition()
+      def notify_condition(unused_future):
+        with condition:
+          condition.notify()
+      self._callbacks.append(notify_condition)
+
+    with condition:
+      condition.wait(timeout=timeout)
+
+    with self._condition:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._exception
+      else:
+        raise future.TimeoutError()
 
-  def add_done_callback(self, callback):
+  def traceback(self, timeout=None):
+    """See future.Future.traceback for specification."""
+    with self._condition:
+      if self._cancelled:
+        raise future.CancelledError()
+      if self._computed:
+        return self._traceback
+
+      condition = threading.Condition()
+      def notify_condition(unused_future):
+        with condition:
+          condition.notify()
+      self._callbacks.append(notify_condition)
+
+    with condition:
+      condition.wait(timeout=timeout)
+
+    with self._condition:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._traceback
+      else:
+        raise future.TimeoutError()
+
+  def add_done_callback(self, fn):
     """See future.Future.add_done_callback for specification."""
     with self._condition:
       if self._callbacks is not None:
-        self._callbacks.add(callback)
+        self._callbacks.add(fn)
         return
 
-      outcome = self._outcome
-
-    callable_util.call_logging_exceptions(
-        callback, _DONE_CALLBACK_LOG_MESSAGE, outcome)
+    callable_util.call_logging_exceptions(fn, _DONE_CALLBACK_LOG_MESSAGE, self)
 
   def on_operation_termination(self, operation_outcome):
     """Indicates to this object that the operation has terminated.
@@ -154,34 +261,42 @@ class _OperationFuture(future.Future):
         outcome of the operation.
     """
     with self._condition:
-      if (self._outcome is None and
-          operation_outcome is not base_interfaces.Outcome.COMPLETED):
-        self._outcome = future.raised(
-            _control.abortion_outcome_to_exception(operation_outcome))
-        self._condition.notify_all()
-
-      outcome = self._outcome
-      rendezvous = self._rendezvous
+      cancelled = self._cancelled
+      if cancelled:
         callbacks = list(self._callbacks)
         self._callbacks = None
+      else:
+        rendezvous = self._rendezvous
 
-    if outcome is None:
+    if not cancelled:
+      payload = None
+      exception = None
+      traceback = None
+      if operation_outcome == base_interfaces.Outcome.COMPLETED:
         try:
-        return_value = next(rendezvous)
+          payload = next(rendezvous)
         except Exception as e:  # pylint: disable=broad-except
-        outcome = future.raised(e)
+          exception = e
+          traceback = sys.exc_info()[2]
       else:
-        outcome = future.returned(return_value)
+        try:
+          # We raise and then immediately catch in order to create a traceback.
+          raise _control.abortion_outcome_to_exception(operation_outcome)
+        except Exception as e:  # pylint: disable=broad-except
+          exception = e
+          traceback = sys.exc_info()[2]
       with self._condition:
-        if self._outcome is None:
-          self._outcome = outcome
-          self._condition.notify_all()
-        else:
-          outcome = self._outcome
+        if not self._cancelled:
+          self._computed = True
+          self._payload = payload
+          self._exception = exception
+          self._traceback = traceback
+        callbacks = list(self._callbacks)
+        self._callbacks = None
 
     for callback in callbacks:
       callable_util.call_logging_exceptions(
-          callback, _DONE_CALLBACK_LOG_MESSAGE, outcome)
+          callback, _DONE_CALLBACK_LOG_MESSAGE, self)
 
 
 class _Call(interfaces.Call):

src/python/src/_framework/face/testing/future_invocation_asynchronous_event_service_test_case.py

@@ -116,7 +116,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         response_future = self.stub.future_value_in_value_out(
             name, request, _TIMEOUT)
-        response = response_future.outcome().return_value
+        response = response_future.result()
 
         test_messages.verify(request, response, self)
 
@@ -144,7 +144,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with request_iterator.pause():
           response_future = self.stub.future_stream_in_value_out(
               name, request_iterator, _TIMEOUT)
-        response = response_future.outcome().return_value
+        response = response_future.result()
 
         test_messages.verify(requests, response, self)
 
@@ -173,13 +173,13 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         first_response_future = self.stub.future_value_in_value_out(
             name, first_request, _TIMEOUT)
-        first_response = first_response_future.outcome().return_value
+        first_response = first_response_future.result()
 
         test_messages.verify(first_request, first_response, self)
 
         second_response_future = self.stub.future_value_in_value_out(
             name, second_request, _TIMEOUT)
-        second_response = second_response_future.outcome().return_value
+        second_response = second_response_future.result()
 
         test_messages.verify(second_request, second_response, self)
 
@@ -192,10 +192,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.pause():
           response_future = self.stub.future_value_in_value_out(
               name, request, _TIMEOUT)
-          outcome = response_future.outcome()
-
           self.assertIsInstance(
-            outcome.exception, exceptions.ExpirationError)
+              response_future.exception(), exceptions.ExpirationError)
+          with self.assertRaises(exceptions.ExpirationError):
+            response_future.result()
 
   def testExpiredUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -203,10 +203,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
       for test_messages in test_messages_sequence:
         request = test_messages.request()
 
-        with self.control.pause(), self.assertRaises(
-            exceptions.ExpirationError):
+        with self.control.pause():
           response_iterator = self.stub.inline_value_in_stream_out(
               name, request, _TIMEOUT)
+          with self.assertRaises(exceptions.ExpirationError):
             list(response_iterator)
 
   def testExpiredStreamRequestUnaryResponse(self):
@@ -218,10 +218,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.pause():
           response_future = self.stub.future_stream_in_value_out(
               name, iter(requests), _TIMEOUT)
-          outcome = response_future.outcome()
-
           self.assertIsInstance(
-            outcome.exception, exceptions.ExpirationError)
+              response_future.exception(), exceptions.ExpirationError)
+          with self.assertRaises(exceptions.ExpirationError):
+            response_future.result()
 
   def testExpiredStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -229,10 +229,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
       for test_messages in test_messages_sequence:
         requests = test_messages.requests()
 
-        with self.control.pause(), self.assertRaises(
-            exceptions.ExpirationError):
+        with self.control.pause():
           response_iterator = self.stub.inline_stream_in_stream_out(
               name, iter(requests), _TIMEOUT)
+          with self.assertRaises(exceptions.ExpirationError):
             list(response_iterator)
 
   def testFailedUnaryRequestUnaryResponse(self):
@@ -244,13 +244,15 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.fail():
           response_future = self.stub.future_value_in_value_out(
               name, request, _TIMEOUT)
-          outcome = response_future.outcome()
 
           # Because the servicer fails outside of the thread from which the
-        # servicer-side runtime called into it its failure is indistinguishable
-        # from simply not having called its response_callback before the
-        # expiration of the RPC.
-        self.assertIsInstance(outcome.exception, exceptions.ExpirationError)
+          # servicer-side runtime called into it its failure is
+          # indistinguishable from simply not having called its
+          # response_callback before the expiration of the RPC.
+          self.assertIsInstance(
+              response_future.exception(), exceptions.ExpirationError)
+          with self.assertRaises(exceptions.ExpirationError):
+            response_future.result()
 
   def testFailedUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -276,13 +278,15 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.fail():
           response_future = self.stub.future_stream_in_value_out(
               name, iter(requests), _TIMEOUT)
-          outcome = response_future.outcome()
 
           # Because the servicer fails outside of the thread from which the
-        # servicer-side runtime called into it its failure is indistinguishable
-        # from simply not having called its response_callback before the
-        # expiration of the RPC.
-        self.assertIsInstance(outcome.exception, exceptions.ExpirationError)
+          # servicer-side runtime called into it its failure is
+          # indistinguishable from simply not having called its
+          # response_callback before the expiration of the RPC.
+          self.assertIsInstance(
+              response_future.exception(), exceptions.ExpirationError)
+          with self.assertRaises(exceptions.ExpirationError):
+            response_future.result()
 
   def testFailedStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -310,8 +314,8 @@ class FutureInvocationAsynchronousEventServiceTestCase(
             name, first_request, _TIMEOUT)
         second_response_future = self.stub.future_value_in_value_out(
             name, second_request, _TIMEOUT)
-        first_response = first_response_future.outcome().return_value
-        second_response = second_response_future.outcome().return_value
+        first_response = first_response_future.result()
+        second_response = second_response_future.result()
 
         test_messages.verify(first_request, first_response, self)
         test_messages.verify(second_request, second_response, self)
@@ -329,10 +333,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.pause():
           response_future = self.stub.future_value_in_value_out(
               name, request, _TIMEOUT)
-          cancelled = response_future.cancel()
+          cancel_method_return_value = response_future.cancel()
 
-        self.assertFalse(cancelled)
-        self.assertEqual(future.ABORTED, response_future.outcome().category)
+        self.assertFalse(cancel_method_return_value)
+        self.assertTrue(response_future.cancelled())
 
   def testCancelledUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -345,7 +349,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
               name, request, _TIMEOUT)
           response_iterator.cancel()
 
-        with self.assertRaises(exceptions.CancellationError):
+        with self.assertRaises(future.CancelledError):
           next(response_iterator)
 
   def testCancelledStreamRequestUnaryResponse(self):
@@ -357,10 +361,10 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         with self.control.pause():
           response_future = self.stub.future_stream_in_value_out(
               name, iter(requests), _TIMEOUT)
-          cancelled = response_future.cancel()
+          cancel_method_return_value = response_future.cancel()
 
-        self.assertFalse(cancelled)
-        self.assertEqual(future.ABORTED, response_future.outcome().category)
+        self.assertFalse(cancel_method_return_value)
+        self.assertTrue(response_future.cancelled())
 
   def testCancelledStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -373,5 +377,5 @@ class FutureInvocationAsynchronousEventServiceTestCase(
               name, iter(requests), _TIMEOUT)
           response_iterator.cancel()
 
-        with self.assertRaises(exceptions.CancellationError):
+        with self.assertRaises(future.CancelledError):
           next(response_iterator)

src/python/src/_framework/foundation/_later_test.py

@@ -33,7 +33,6 @@ import threading
 import time
 import unittest
 
-from _framework.foundation import future
 from _framework.foundation import later
 
 TICK = 0.1
@@ -44,10 +43,14 @@ class LaterTest(unittest.TestCase):
   def test_simple_delay(self):
     lock = threading.Lock()
     cell = [0]
-    def increment_cell():
+    return_value = object()
+
+    def computation():
       with lock:
         cell[0] += 1
-    computation_future = later.later(TICK * 2, increment_cell)
+      return return_value
+    computation_future = later.later(TICK * 2, computation)
+
     self.assertFalse(computation_future.done())
     self.assertFalse(computation_future.cancelled())
     time.sleep(TICK)
@@ -60,22 +63,21 @@ class LaterTest(unittest.TestCase):
     self.assertFalse(computation_future.cancelled())
     with lock:
       self.assertEqual(1, cell[0])
-    outcome = computation_future.outcome()
-    self.assertEqual(future.RETURNED, outcome.category)
+    self.assertEqual(return_value, computation_future.result())
 
   def test_callback(self):
     lock = threading.Lock()
     cell = [0]
     callback_called = [False]
-    outcome_passed_to_callback = [None]
-    def increment_cell():
+    future_passed_to_callback = [None]
+    def computation():
       with lock:
         cell[0] += 1
-    computation_future = later.later(TICK * 2, increment_cell)
+    computation_future = later.later(TICK * 2, computation)
     def callback(outcome):
       with lock:
         callback_called[0] = True
-        outcome_passed_to_callback[0] = outcome
+        future_passed_to_callback[0] = outcome
     computation_future.add_done_callback(callback)
     time.sleep(TICK)
     with lock:
@@ -83,63 +85,67 @@ class LaterTest(unittest.TestCase):
     time.sleep(TICK * 2)
     with lock:
       self.assertTrue(callback_called[0])
-      self.assertEqual(future.RETURNED, outcome_passed_to_callback[0].category)
+      self.assertTrue(future_passed_to_callback[0].done())
 
       callback_called[0] = False
-      outcome_passed_to_callback[0] = None
+      future_passed_to_callback[0] = None
 
     computation_future.add_done_callback(callback)
     with lock:
       self.assertTrue(callback_called[0])
-      self.assertEqual(future.RETURNED, outcome_passed_to_callback[0].category)
+      self.assertTrue(future_passed_to_callback[0].done())
 
   def test_cancel(self):
     lock = threading.Lock()
     cell = [0]
     callback_called = [False]
-    outcome_passed_to_callback = [None]
-    def increment_cell():
+    future_passed_to_callback = [None]
+    def computation():
       with lock:
         cell[0] += 1
-    computation_future = later.later(TICK * 2, increment_cell)
+    computation_future = later.later(TICK * 2, computation)
     def callback(outcome):
       with lock:
         callback_called[0] = True
-        outcome_passed_to_callback[0] = outcome
+        future_passed_to_callback[0] = outcome
     computation_future.add_done_callback(callback)
     time.sleep(TICK)
     with lock:
       self.assertFalse(callback_called[0])
     computation_future.cancel()
     self.assertTrue(computation_future.cancelled())
-    self.assertFalse(computation_future.done())
-    self.assertEqual(future.ABORTED, computation_future.outcome().category)
+    self.assertFalse(computation_future.running())
+    self.assertTrue(computation_future.done())
     with lock:
       self.assertTrue(callback_called[0])
-      self.assertEqual(future.ABORTED, outcome_passed_to_callback[0].category)
+      self.assertTrue(future_passed_to_callback[0].cancelled())
 
-  def test_outcome(self):
+  def test_result(self):
     lock = threading.Lock()
     cell = [0]
     callback_called = [False]
-    outcome_passed_to_callback = [None]
-    def increment_cell():
+    future_passed_to_callback_cell = [None]
+    return_value = object()
+
+    def computation():
       with lock:
         cell[0] += 1
-    computation_future = later.later(TICK * 2, increment_cell)
-    def callback(outcome):
+      return return_value
+    computation_future = later.later(TICK * 2, computation)
+
+    def callback(future_passed_to_callback):
       with lock:
         callback_called[0] = True
-        outcome_passed_to_callback[0] = outcome
+        future_passed_to_callback_cell[0] = future_passed_to_callback
     computation_future.add_done_callback(callback)
-    returned_outcome = computation_future.outcome()
-    self.assertEqual(future.RETURNED, returned_outcome.category)
+    returned_value = computation_future.result()
+    self.assertEqual(return_value, returned_value)
 
     # The callback may not yet have been called! Sleep a tick.
     time.sleep(TICK)
     with lock:
       self.assertTrue(callback_called[0])
-      self.assertEqual(future.RETURNED, outcome_passed_to_callback[0].category)
+      self.assertEqual(return_value, future_passed_to_callback_cell[0].result())
 
 if __name__ == '__main__':
   unittest.main()

src/python/src/_framework/foundation/_timer_future.py

@@ -29,6 +29,7 @@
 
 """Affords a Future implementation based on Python's threading.Timer."""
 
+import sys
 import threading
 import time
 
@@ -52,7 +53,9 @@ class TimerFuture(future.Future):
     self._computing = False
     self._computed = False
     self._cancelled = False
-    self._outcome = None
+    self._result = None
+    self._exception = None
+    self._traceback = None
     self._waiting = []
 
   def _compute(self):
@@ -70,19 +73,24 @@ class TimerFuture(future.Future):
         self._computing = True
 
     try:
-      returned_value = self._computation()
-      outcome = future.returned(returned_value)
+      return_value = self._computation()
+      exception = None
+      traceback = None
     except Exception as e:  # pylint: disable=broad-except
-      outcome = future.raised(e)
+      return_value = None
+      exception = e
+      traceback = sys.exc_info()[2]
 
     with self._lock:
       self._computing = False
       self._computed = True
-      self._outcome = outcome
+      self._return_value = return_value
+      self._exception = exception
+      self._traceback = traceback
       waiting = self._waiting
 
     for callback in waiting:
-      callback(outcome)
+      callback(self)
 
   def start(self):
     """Starts this Future.
@@ -104,13 +112,11 @@ class TimerFuture(future.Future):
       else:
         self._timer.cancel()
         self._cancelled = True
-        self._outcome = future.aborted()
-        outcome = self._outcome
         waiting = self._waiting
 
     for callback in waiting:
       try:
-        callback(outcome)
+        callback(self)
       except Exception:  # pylint: disable=broad-except
         pass
 
@@ -121,36 +127,102 @@ class TimerFuture(future.Future):
     with self._lock:
       return self._cancelled
 
+  def running(self):
+    """See future.Future.running for specification."""
+    with self._lock:
+      return not self._computed and not self._cancelled
+
   def done(self):
     """See future.Future.done for specification."""
     with self._lock:
-      return self._computed
+      return self._computed or self._cancelled
+
+  def result(self, timeout=None):
+    """See future.Future.result for specification."""
+    with self._lock:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        if self._exception is None:
+          return self._return_value
+        else:
+          raise self._exception  # pylint: disable=raising-bad-type
+
+      condition = threading.Condition()
+      def notify_condition(unused_future):
+        with condition:
+          condition.notify()
+      self._waiting.append(notify_condition)
+
+    with condition:
+      condition.wait(timeout=timeout)
+
+    with self._lock:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        if self._exception is None:
+          return self._return_value
+        else:
+          raise self._exception  # pylint: disable=raising-bad-type
+      else:
+        raise future.TimeoutError()
+
+  def exception(self, timeout=None):
+    """See future.Future.exception for specification."""
+    with self._lock:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._exception
+
+      condition = threading.Condition()
+      def notify_condition(unused_future):
+        with condition:
+          condition.notify()
+      self._waiting.append(notify_condition)
+
+    with condition:
+      condition.wait(timeout=timeout)
 
-  def outcome(self):
-    """See future.Future.outcome for specification."""
     with self._lock:
-      if self._computed or self._cancelled:
-        return self._outcome
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._exception
+      else:
+        raise future.TimeoutError()
+
+  def traceback(self, timeout=None):
+    """See future.Future.traceback for specification."""
+    with self._lock:
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._traceback
 
       condition = threading.Condition()
-      def notify_condition(unused_outcome):
+      def notify_condition(unused_future):
         with condition:
           condition.notify()
       self._waiting.append(notify_condition)
 
     with condition:
-      condition.wait()
+      condition.wait(timeout=timeout)
 
     with self._lock:
-      return self._outcome
+      if self._cancelled:
+        raise future.CancelledError()
+      elif self._computed:
+        return self._traceback
+      else:
+        raise future.TimeoutError()
 
-  def add_done_callback(self, callback):
+  def add_done_callback(self, fn):
     """See future.Future.add_done_callback for specification."""
     with self._lock:
       if not self._computed and not self._cancelled:
-        self._waiting.append(callback)
+        self._waiting.append(fn)
         return
-      else:
-        outcome = self._outcome
 
-    callback(outcome)
+    fn(self)

src/python/src/_framework/foundation/callable_util.py

@@ -29,18 +29,47 @@
 
 """Utilities for working with callables."""
 
+import abc
+import collections
+import enum
 import functools
 import logging
 
-from _framework.foundation import future
+
+class Outcome(object):
+  """A sum type describing the outcome of some call.
+
+  Attributes:
+    kind: One of Kind.RETURNED or Kind.RAISED respectively indicating that the
+      call returned a value or raised an exception.
+    return_value: The value returned by the call. Must be present if kind is
+      Kind.RETURNED.
+    exception: The exception raised by the call. Must be present if kind is
+      Kind.RAISED.
+  """
+  __metaclass__ = abc.ABCMeta
+
+  @enum.unique
+  class Kind(enum.Enum):
+    """Identifies the general kind of the outcome of some call."""
+
+    RETURNED = object()
+    RAISED = object()
+
+
+class _EasyOutcome(
+    collections.namedtuple(
+        '_EasyOutcome', ['kind', 'return_value', 'exception']),
+    Outcome):
+  """A trivial implementation of Outcome."""
 
 
 def _call_logging_exceptions(behavior, message, *args, **kwargs):
   try:
-    return future.returned(behavior(*args, **kwargs))
+    return _EasyOutcome(Outcome.Kind.RETURNED, behavior(*args, **kwargs), None)
   except Exception as e:  # pylint: disable=broad-except
     logging.exception(message)
-    return future.raised(e)
+    return _EasyOutcome(Outcome.Kind.RAISED, None, e)
 
 
 def with_exceptions_logged(behavior, message):
@@ -72,7 +101,7 @@ def call_logging_exceptions(behavior, message, *args, **kwargs):
     **kwargs: Keyword arguments to pass to the given behavior.
 
   Returns:
-    A future.Outcome describing whether the given behavior returned a value or
-      raised an exception.
+    An Outcome describing whether the given behavior returned a value or raised
+      an exception.
   """
   return _call_logging_exceptions(behavior, message, *args, **kwargs)

src/python/src/_framework/foundation/future.py

@@ -27,146 +27,210 @@
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-"""The Future interface missing from Python's standard library.
+"""A Future interface.
 
-Python's concurrent.futures library defines a Future class very much like the
-Future defined here, but since that class is concrete and without construction
-semantics it is only available within the concurrent.futures library itself.
-The Future class defined here is an entirely abstract interface that anyone may
+Python doesn't have a Future interface in its standard library. In the absence
+of such a standard, three separate, incompatible implementations
+(concurrent.futures.Future, ndb.Future, and asyncio.Future) have appeared. This
+interface attempts to be as compatible as possible with
+concurrent.futures.Future. From ndb.Future it adopts a traceback-object accessor
+method.
+
+Unlike the concrete and implemented Future classes listed above, the Future
+class defined in this module is an entirely abstract interface that anyone may
 implement and use.
+
+The one known incompatibility between this interface and the interface of
+concurrent.futures.Future is that this interface defines its own CancelledError
+and TimeoutError exceptions rather than raising the implementation-private
+concurrent.futures._base.CancelledError and the
+built-in-but-only-in-3.3-and-later TimeoutError.
 """
 
 import abc
-import collections
-
-RETURNED = object()
-RAISED = object()
-ABORTED = object()
-
 
-class Outcome(object):
-  """A sum type describing the outcome of some computation.
-
-  Attributes:
-    category: One of RETURNED, RAISED, or ABORTED, respectively indicating
-      that the computation returned a value, raised an exception, or was
-      aborted.
-    return_value: The value returned by the computation. Must be present if
-      category is RETURNED.
-    exception: The exception raised by the computation. Must be present if
-      category is RAISED.
-  """
-  __metaclass__ = abc.ABCMeta
 
+class TimeoutError(Exception):
+  """Indicates that a particular call timed out."""
 
-class _EasyOutcome(
-    collections.namedtuple('_EasyOutcome',
-                           ['category', 'return_value', 'exception']),
-    Outcome):
-  """A trivial implementation of Outcome."""
 
-# All Outcomes describing abortion are indistinguishable so there might as well
-# be only one.
-_ABORTED_OUTCOME = _EasyOutcome(ABORTED, None, None)
+class CancelledError(Exception):
+  """Indicates that the computation underlying a Future was cancelled."""
 
 
-def aborted():
-  """Returns an Outcome indicating that a computation was aborted.
+class Future(object):
+  """A representation of a computation in another control flow.
 
-  Returns:
-    An Outcome indicating that a computation was aborted.
+  Computations represented by a Future may be yet to be begun, may be ongoing,
+  or may have already completed.
   """
-  return _ABORTED_OUTCOME
-
+  __metaclass__ = abc.ABCMeta
 
-def raised(exception):
-  """Returns an Outcome indicating that a computation raised an exception.
+  # NOTE(nathaniel): This isn't the return type that I would want to have if it
+  # were up to me. Were this interface being written from scratch, the return
+  # type of this method would probably be a sum type like:
+  #
+  # NOT_COMMENCED
+  # COMMENCED_AND_NOT_COMPLETED
+  # PARTIAL_RESULT<Partial_Result_Type>
+  # COMPLETED<Result_Type>
+  # UNCANCELLABLE
+  # NOT_IMMEDIATELY_DETERMINABLE
+  @abc.abstractmethod
+  def cancel(self):
+    """Attempts to cancel the computation.
 
-  Args:
-    exception: The exception raised by the computation.
+    This method does not block.
 
     Returns:
-    An Outcome indicating that a computation raised the given exception.
+      True if the computation has not yet begun, will not be allowed to take
+        place, and determination of both was possible without blocking. False
+        under all other circumstances including but not limited to the
+        computation's already having begun, the computation's already having
+        finished, and the computation's having been scheduled for execution on a
+        remote system for which a determination of whether or not it commenced
+        before being cancelled cannot be made without blocking.
     """
-  return _EasyOutcome(RAISED, None, exception)
-
+    raise NotImplementedError()
 
-def returned(value):
-  """Returns an Outcome indicating that a computation returned a value.
+  # NOTE(nathaniel): Here too this isn't the return type that I'd want this
+  # method to have if it were up to me. I think I'd go with another sum type
+  # like:
+  #
+  # NOT_CANCELLED (this object's cancel method hasn't been called)
+  # NOT_COMMENCED
+  # COMMENCED_AND_NOT_COMPLETED
+  # PARTIAL_RESULT<Partial_Result_Type>
+  # COMPLETED<Result_Type>
+  # UNCANCELLABLE
+  # NOT_IMMEDIATELY_DETERMINABLE
+  #
+  # Notice how giving the cancel method the right semantics obviates most
+  # reasons for this method to exist.
+  @abc.abstractmethod
+  def cancelled(self):
+    """Describes whether the computation was cancelled.
 
-  Args:
-    value: The value returned by the computation.
+    This method does not block.
 
     Returns:
-    An Outcome indicating that a computation returned the given value.
+      True if the computation was cancelled any time before its result became
+        immediately available. False under all other circumstances including but
+        not limited to this object's cancel method not having been called and
+        the computation's result having become immediately available.
     """
-  return _EasyOutcome(RETURNED, value, None)
-
+    raise NotImplementedError()
 
-class Future(object):
-  """A representation of a computation happening in another control flow.
+  @abc.abstractmethod
+  def running(self):
+    """Describes whether the computation is taking place.
 
-  Computations represented by a Future may have already completed, may be
-  ongoing, or may be yet to be begun.
+    This method does not block.
 
-  Computations represented by a Future are considered uninterruptable; once
-  started they will be allowed to terminate either by returning or raising
-  an exception.
+    Returns:
+      True if the computation is scheduled to take place in the future or is
+        taking place now, or False if the computation took place in the past or
+        was cancelled.
     """
-  __metaclass__ = abc.ABCMeta
+    raise NotImplementedError()
 
+  # NOTE(nathaniel): These aren't quite the semantics I'd like here either. I
+  # would rather this only returned True in cases in which the underlying
+  # computation completed successfully. A computation's having been cancelled
+  # conflicts with considering that computation "done".
   @abc.abstractmethod
-  def cancel(self):
-    """Attempts to cancel the computation.
+  def done(self):
+    """Describes whether the computation has taken place.
+
+    This method does not block.
 
     Returns:
-      True if the computation will not be allowed to take place or False if
-        the computation has already taken place or is currently taking place.
+      True if the computation is known to have either completed or have been
+        unscheduled or interrupted. False if the computation may possibly be
+        executing or scheduled to execute later.
     """
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def cancelled(self):
-    """Describes whether the computation was cancelled.
+  def result(self, timeout=None):
+    """Accesses the outcome of the computation or raises its exception.
+
+    This method may return immediately or may block.
+
+    Args:
+      timeout: The length of time in seconds to wait for the computation to
+        finish or be cancelled, or None if this method should block until the
+        computation has finished or is cancelled no matter how long that takes.
 
     Returns:
-      True if the computation was cancelled and did not take place or False
-        if the computation took place, is taking place, or is scheduled to
-        take place in the future.
+      The return value of the computation.
+
+    Raises:
+      TimeoutError: If a timeout value is passed and the computation does not
+        terminate within the allotted time.
+      CancelledError: If the computation was cancelled.
+      Exception: If the computation raised an exception, this call will raise
+        the same exception.
     """
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def done(self):
-    """Describes whether the computation has taken place.
+  def exception(self, timeout=None):
+    """Return the exception raised by the computation.
+
+    This method may return immediately or may block.
+
+    Args:
+      timeout: The length of time in seconds to wait for the computation to
+        terminate or be cancelled, or None if this method should block until
+        the computation is terminated or is cancelled no matter how long that
+        takes.
 
     Returns:
-      True if the computation took place; False otherwise.
+      The exception raised by the computation, or None if the computation did
+        not raise an exception.
+
+    Raises:
+      TimeoutError: If a timeout value is passed and the computation does not
+        terminate within the allotted time.
+      CancelledError: If the computation was cancelled.
     """
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def outcome(self):
-    """Accesses the outcome of the computation.
+  def traceback(self, timeout=None):
+    """Access the traceback of the exception raised by the computation.
 
-    If the computation has not yet completed, this method blocks until it has.
+    This method may return immediately or may block.
+
+    Args:
+      timeout: The length of time in seconds to wait for the computation to
+        terminate or be cancelled, or None if this method should block until
+        the computation is terminated or is cancelled no matter how long that
+        takes.
 
     Returns:
-      An Outcome describing the outcome of the computation.
+      The traceback of the exception raised by the computation, or None if the
+        computation did not raise an exception.
+
+    Raises:
+      TimeoutError: If a timeout value is passed and the computation does not
+        terminate within the allotted time.
+      CancelledError: If the computation was cancelled.
     """
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def add_done_callback(self, callback):
+  def add_done_callback(self, fn):
     """Adds a function to be called at completion of the computation.
 
-    The callback will be passed an Outcome object describing the outcome of
+    The callback will be passed this Future object describing the outcome of
     the computation.
 
     If the computation has already completed, the callback will be called
     immediately.
 
     Args:
-      callback: A callable taking an Outcome as its single parameter.
+      fn: A callable taking a this Future object as its single parameter.
     """
     raise NotImplementedError()

templates/Makefile.template

@@ -83,13 +83,13 @@ DEFINES_asan = NDEBUG
 VALID_CONFIG_msan = 1
 REQUIRE_CUSTOM_LIBRARIES_msan = 1
 CC_msan = clang
-CXX_msan = clang++
+CXX_msan = clang++-libc++
 LD_msan = clang
-LDXX_msan = clang++
-CPPFLAGS_msan = -O1 -fsanitize=memory -fno-omit-frame-pointer
+LDXX_msan = clang++-libc++
+CPPFLAGS_msan = -O1 -fsanitize=memory -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1
 OPENSSL_CFLAGS_msan = -DPURIFY
 OPENSSL_CONFIG_msan = no-asm
-LDFLAGS_msan = -fsanitize=memory
+LDFLAGS_msan = -fsanitize=memory -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1
 DEFINES_msan = NDEBUG
 
 VALID_CONFIG_ubsan = 1

test/core/fling/client.c

@@ -49,40 +49,77 @@ static grpc_byte_buffer *the_buffer;
 static grpc_channel *channel;
 static grpc_completion_queue *cq;
 static grpc_call *call;
-
-static void init_ping_pong_request(void) {}
+static grpc_op ops[6];
+static grpc_op stream_init_op;
+static grpc_op stream_step_ops[2];
+static grpc_metadata_array initial_metadata_recv;
+static grpc_metadata_array trailing_metadata_recv;
+static grpc_byte_buffer *response_payload_recv = NULL;
+static grpc_call_details call_details;
+static grpc_status_code status;
+static char *details = NULL;
+static size_t details_capacity = 0;
+static grpc_op *op;
+
+static void init_ping_pong_request(void) {
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_call_details_init(&call_details);
+
+  op = ops;
+
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message = the_buffer;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata = &initial_metadata_recv;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &response_payload_recv;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op->data.recv_status_on_client.status_details_capacity = &details_capacity;
+  op++;
+}
 
 static void step_ping_pong_request(void) {
-  call = grpc_channel_create_call_old(channel, "/Reflector/reflectUnary",
+  call = grpc_channel_create_call(channel, cq, "/Reflector/reflectUnary",
                                   "localhost", gpr_inf_future);
-  GPR_ASSERT(grpc_call_invoke_old(call, cq, (void *)1, (void *)1,
-                                  GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
-  GPR_ASSERT(grpc_call_start_write_old(call, the_buffer, (void *)1,
-                                       GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
-  GPR_ASSERT(grpc_call_start_read_old(call, (void *)1) == GRPC_CALL_OK);
-  GPR_ASSERT(grpc_call_writes_done_old(call, (void *)1) == GRPC_CALL_OK);
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(call, ops, op - ops, (void *)1));
   grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
   grpc_call_destroy(call);
   call = NULL;
 }
 
 static void init_ping_pong_stream(void) {
-  call = grpc_channel_create_call_old(channel, "/Reflector/reflectStream",
+  call = grpc_channel_create_call(channel, cq, "/Reflector/reflectStream",
                                   "localhost", gpr_inf_future);
-  GPR_ASSERT(grpc_call_invoke_old(call, cq, (void *)1, (void *)1, 0) ==
-             GRPC_CALL_OK);
+  stream_init_op.op = GRPC_OP_SEND_INITIAL_METADATA;
+  stream_init_op.data.send_initial_metadata.count = 0;
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(call, &stream_init_op, 1, (void *)1));
   grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+
+  stream_step_ops[0].op = GRPC_OP_SEND_MESSAGE;
+  stream_step_ops[0].data.send_message = the_buffer;
+  stream_step_ops[1].op = GRPC_OP_RECV_MESSAGE;
+  stream_step_ops[1].data.recv_message = &response_payload_recv;
 }
 
 static void step_ping_pong_stream(void) {
-  GPR_ASSERT(grpc_call_start_write_old(call, the_buffer, (void *)1, 0) ==
-             GRPC_CALL_OK);
-  GPR_ASSERT(grpc_call_start_read_old(call, (void *)1) == GRPC_CALL_OK);
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(call, stream_step_ops, 2, (void *)1));
   grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
 }
 
@@ -99,7 +136,8 @@ typedef struct {
 
 static const scenario scenarios[] = {
     {"ping-pong-request", init_ping_pong_request, step_ping_pong_request},
-    {"ping-pong-stream", init_ping_pong_stream, step_ping_pong_stream}, };
+    {"ping-pong-stream", init_ping_pong_stream, step_ping_pong_stream},
+};
 
 int main(int argc, char **argv) {
   gpr_slice slice = gpr_slice_from_copied_string("x");
@@ -148,6 +186,7 @@ int main(int argc, char **argv) {
   cq = grpc_completion_queue_create();
   the_buffer = grpc_byte_buffer_create(&slice, payload_size);
   histogram = gpr_histogram_create(0.01, 60e9);
+
   sc.init();
 
   for (i = 0; i < 1000; i++) {

test/core/fling/server.c

@@ -52,17 +52,118 @@
 
 static grpc_completion_queue *cq;
 static grpc_server *server;
+static grpc_call *call;
+static grpc_call_details call_details;
+static grpc_metadata_array request_metadata_recv;
+static grpc_metadata_array initial_metadata_send;
+static grpc_byte_buffer *payload_buffer = NULL;
+/* Used to drain the terminal read in unary calls. */
+static grpc_byte_buffer *terminal_buffer = NULL;
+
+static grpc_op read_op;
+static grpc_op metadata_send_op;
+static grpc_op write_op;
+static grpc_op status_op[2];
+static int was_cancelled = 2;
+static grpc_op unary_ops[6];
 static int got_sigint = 0;
 
+static void *tag(gpr_intptr t) { return (void *)t; }
+
+typedef enum {
+  FLING_SERVER_NEW_REQUEST = 1,
+  FLING_SERVER_READ_FOR_UNARY,
+  FLING_SERVER_BATCH_OPS_FOR_UNARY,
+  FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING,
+  FLING_SERVER_READ_FOR_STREAMING,
+  FLING_SERVER_WRITE_FOR_STREAMING,
+  FLING_SERVER_SEND_STATUS_FOR_STREAMING
+} fling_server_tags;
+
 typedef struct {
   gpr_refcount pending_ops;
   gpr_uint32 flags;
 } call_state;
 
 static void request_call(void) {
-  call_state *s = gpr_malloc(sizeof(call_state));
-  gpr_ref_init(&s->pending_ops, 2);
-  grpc_server_request_call_old(server, s);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+  grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
+                           cq, tag(FLING_SERVER_NEW_REQUEST));
+}
+
+static void handle_unary_method(void) {
+  grpc_op *op;
+
+  grpc_metadata_array_init(&initial_metadata_send);
+
+  op = unary_ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &terminal_buffer;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  if (payload_buffer == NULL) {
+    gpr_log(GPR_INFO, "NULL payload buffer !!!");
+  }
+  op->data.send_message = payload_buffer;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.status = GRPC_STATUS_OK;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status_details = "";
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op++;
+
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(call, unary_ops, op - unary_ops,
+                                   tag(FLING_SERVER_BATCH_OPS_FOR_UNARY)));
+}
+
+static void send_initial_metadata(void) {
+  grpc_metadata_array_init(&initial_metadata_send);
+  metadata_send_op.op = GRPC_OP_SEND_INITIAL_METADATA;
+  metadata_send_op.data.send_initial_metadata.count = 0;
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(
+                 call, &metadata_send_op, 1,
+                 tag(FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING)));
+}
+
+static void start_read_op(int t) {
+  /* Starting read at server */
+  read_op.op = GRPC_OP_RECV_MESSAGE;
+  read_op.data.recv_message = &payload_buffer;
+  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, &read_op, 1, tag(t)));
+}
+
+static void start_write_op(void) {
+  /* Starting write at server */
+  write_op.op = GRPC_OP_SEND_MESSAGE;
+  if (payload_buffer == NULL) {
+    gpr_log(GPR_INFO, "NULL payload buffer !!!");
+  }
+  write_op.data.send_message = payload_buffer;
+  GPR_ASSERT(GRPC_CALL_OK ==
+             grpc_call_start_batch(call, &write_op, 1,
+                                   tag(FLING_SERVER_WRITE_FOR_STREAMING)));
+}
+
+static void start_send_status(void) {
+  status_op[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  status_op[0].data.send_status_from_server.status = GRPC_STATUS_OK;
+  status_op[0].data.send_status_from_server.trailing_metadata_count = 0;
+  status_op[0].data.send_status_from_server.status_details = "";
+  status_op[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  status_op[1].data.recv_close_on_server.cancelled = &was_cancelled;
+
+  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(
+                                 call, status_op, 2,
+                                 tag(FLING_SERVER_SEND_STATUS_FOR_STREAMING)));
 }
 
 static void sigint_handler(int x) { got_sigint = 1; }
@@ -133,43 +234,66 @@ int main(int argc, char **argv) {
     if (!ev) continue;
     s = ev->tag;
     switch (ev->type) {
-      case GRPC_SERVER_RPC_NEW:
-        if (ev->call != NULL) {
-          /* initial ops are already started in request_call */
-          if (0 == strcmp(ev->data.server_rpc_new.method,
-                          "/Reflector/reflectStream")) {
-            s->flags = 0;
+      case GRPC_OP_COMPLETE:
+        switch ((gpr_intptr)s) {
+          case FLING_SERVER_NEW_REQUEST:
+            if (call != NULL) {
+              if (0 ==
+                  strcmp(call_details.method, "/Reflector/reflectStream")) {
+                /* Received streaming call. Send metadata here. */
+                start_read_op(FLING_SERVER_READ_FOR_STREAMING);
+                send_initial_metadata();
               } else {
-            s->flags = GRPC_WRITE_BUFFER_HINT;
+                /* Received unary call. Can do all ops in one batch. */
+                start_read_op(FLING_SERVER_READ_FOR_UNARY);
               }
-          grpc_call_server_accept_old(ev->call, cq, s);
-          grpc_call_server_end_initial_metadata_old(ev->call, s->flags);
-          GPR_ASSERT(grpc_call_start_read_old(ev->call, s) == GRPC_CALL_OK);
-          request_call();
             } else {
               GPR_ASSERT(shutdown_started);
-          gpr_free(s);
             }
+            /*	    request_call();
+             */
             break;
-      case GRPC_WRITE_ACCEPTED:
-        GPR_ASSERT(ev->data.write_accepted == GRPC_OP_OK);
-        GPR_ASSERT(grpc_call_start_read_old(ev->call, s) == GRPC_CALL_OK);
-        break;
-      case GRPC_READ:
-        if (ev->data.read) {
-          GPR_ASSERT(grpc_call_start_write_old(ev->call, ev->data.read, s,
-                                               s->flags) == GRPC_CALL_OK);
+          case FLING_SERVER_READ_FOR_STREAMING:
+            if (payload_buffer != NULL) {
+              /* Received payload from client. */
+              start_write_op();
             } else {
-          GPR_ASSERT(grpc_call_start_write_status_old(ev->call, GRPC_STATUS_OK,
-                                                      NULL, s) == GRPC_CALL_OK);
+              /* Received end of stream from client. */
+              start_send_status();
             }
             break;
+          case FLING_SERVER_WRITE_FOR_STREAMING:
+            /* Write completed at server  */
+            start_read_op(FLING_SERVER_READ_FOR_STREAMING);
+            break;
+          case FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING:
+            /* Metadata send completed at server */
+            break;
+          case FLING_SERVER_SEND_STATUS_FOR_STREAMING:
+            /* Send status and close completed at server */
+            grpc_call_destroy(call);
+            request_call();
+            break;
+          case FLING_SERVER_READ_FOR_UNARY:
+            /* Finished payload read for unary. Start all reamaining
+             *  unary ops in a batch.
+             */
+            handle_unary_method();
+            break;
+          case FLING_SERVER_BATCH_OPS_FOR_UNARY:
+            /* Finished unary call. */
+            grpc_call_destroy(call);
+            request_call();
+            break;
+        }
+        break;
+      case GRPC_SERVER_RPC_NEW:
+      case GRPC_WRITE_ACCEPTED:
+      case GRPC_READ:
       case GRPC_FINISH_ACCEPTED:
       case GRPC_FINISHED:
-        if (gpr_unref(&s->pending_ops)) {
-          grpc_call_destroy(ev->call);
-          gpr_free(s);
-        }
+        gpr_log(GPR_ERROR, "Unexpected event type.");
+        abort();
         break;
       case GRPC_QUEUE_SHUTDOWN:
         GPR_ASSERT(shutdown_started);

tools/dockerfile/grpc_clang/Dockerfile

@@ -23,7 +23,10 @@ RUN mv libcxx llvm/projects
 RUN mv libcxxabi llvm/projects
 
 RUN mkdir llvm-build
-RUN cd llvm-build && cmake ../llvm
+RUN cd llvm-build && cmake \
+  -DCMAKE_BUILD_TYPE:STRING=Release \
+  -DLLVM_TARGETS_TO_BUILD:STRING=X86 \
+  ../llvm
 RUN make -C llvm-build && make -C llvm-build install && rm -rf llvm-build
 
 CMD ["bash"]