[ORC][RPC] Reword 'async' to 'non-blocking' to better reflect call primitive

behaviors, and add a callB (blacking call) primitive.

callB is a blocking call primitive for threaded code where the RPC responses are
being processed on a separate thread. (For single threaded code callST should
continue to be used instead).

No unit test yet: Last time I commited a threaded unit test it deadlocked on
one of the s390x builders. I'll try to re-enable that test first, and add a new
test if I can sort out the deadlock issue.



git-svn-id: https://llvm.org/svn/llvm-project/llvm/trunk@280051 91177308-0d34-0410-b5e6-96231b3b80d8

Author: lhames

include/llvm/ExecutionEngine/Orc/RPCUtils.h

@@ -71,7 +71,7 @@ class RPCBase {
   // Id - The function's unique identifier.
   // ErrorReturn - The return type for blocking calls.
   // readResult - Deserialize a result from a channel.
-  // abandon - Abandon a promised (asynchronous) result.
+  // abandon - Abandon a promised result.
   // respond - Retun a result on the channel.
   template <typename FunctionIdT, FunctionIdT FuncId, typename FnT>
   class FunctionHelper {};
@@ -109,6 +109,10 @@ class RPCBase {
                                 inconvertibleErrorCode()));
     }
 
+    static void consumeAbandoned(std::future<ErrorReturn> &P) {
+      consumeError(P.get().takeError());
+    }
+
     template <typename ChannelT, typename SequenceNumberT>
     static Error respond(ChannelT &C, SequenceNumberT SeqNo,
                          ErrorReturn &Result) {
@@ -153,6 +157,10 @@ class RPCBase {
                                 inconvertibleErrorCode()));
     }
 
+    static void consumeAbandoned(std::future<ErrorReturn> &P) {
+      consumeError(P.get());
+    }
+
     template <typename ChannelT, typename SequenceNumberT>
     static Error respond(ChannelT &C, SequenceNumberT SeqNo,
                          ErrorReturn &Result) {
@@ -366,28 +374,25 @@ class RPC : public RPCBase {
   template <FunctionIdT FuncId, typename FnT>
   using Function = FunctionHelper<FunctionIdT, FuncId, FnT>;
 
-  /// Return type for asynchronous call primitives.
+  /// Return type for non-blocking call primitives.
   template <typename Func>
-  using AsyncCallResult = std::future<typename Func::ErrorReturn>;
+  using NonBlockingCallResult = std::future<typename Func::ErrorReturn>;
 
-  /// Return type for asynchronous call-with-seq primitives.
+  /// Return type for non-blocking call-with-seq primitives.
   template <typename Func>
-  using AsyncCallWithSeqResult =
-      std::pair<AsyncCallResult<Func>, SequenceNumberT>;
+  using NonBlockingCallWithSeqResult =
+      std::pair<NonBlockingCallResult<Func>, SequenceNumberT>;
 
-  /// Serialize Args... to channel C, but do not call C.send().
-  ///
-  /// Returns an error (on serialization failure) or a pair of:
-  /// (1) A future Expected<T> (or future<Error> for void functions), and
-  /// (2) A sequence number.
+  /// Call Func on Channel C. Does not block, does not call send. Returns a pair
+  /// of a future result and the sequence number assigned to the result.
   ///
   /// This utility function is primarily used for single-threaded mode support,
   /// where the sequence number can be used to wait for the corresponding
-  /// result. In multi-threaded mode the appendCallAsync method, which does not
+  /// result. In multi-threaded mode the appendCallNB method, which does not
   /// return the sequence numeber, should be preferred.
   template <typename Func, typename... ArgTs>
-  Expected<AsyncCallWithSeqResult<Func>>
-  appendCallAsyncWithSeq(ChannelT &C, const ArgTs &... Args) {
+  Expected<NonBlockingCallWithSeqResult<Func>>
+  appendCallNBWithSeq(ChannelT &C, const ArgTs &... Args) {
     auto SeqNo = SequenceNumberMgr.getSequenceNumber();
     std::promise<typename Func::ErrorReturn> Promise;
     auto Result = Promise.get_future();
@@ -397,21 +402,23 @@ class RPC : public RPCBase {
     if (auto Err = CallHelper<ChannelT, SequenceNumberT, Func>::call(C, SeqNo,
                                                                      Args...)) {
       abandonOutstandingResults();
+      Func::consumeAbandoned(Result);
       return std::move(Err);
     } else
-      return AsyncCallWithSeqResult<Func>(std::move(Result), SeqNo);
+      return NonBlockingCallWithSeqResult<Func>(std::move(Result), SeqNo);
   }
 
-  /// The same as appendCallAsyncWithSeq, except that it calls C.send() to
+  /// The same as appendCallNBWithSeq, except that it calls C.send() to
   /// flush the channel after serializing the call.
   template <typename Func, typename... ArgTs>
-  Expected<AsyncCallWithSeqResult<Func>>
-  callAsyncWithSeq(ChannelT &C, const ArgTs &... Args) {
-    auto Result = appendCallAsyncWithSeq<Func>(C, Args...);
+  Expected<NonBlockingCallWithSeqResult<Func>>
+  callNBWithSeq(ChannelT &C, const ArgTs &... Args) {
+    auto Result = appendCallNBWithSeq<Func>(C, Args...);
     if (!Result)
       return Result;
     if (auto Err = C.send()) {
       abandonOutstandingResults();
+      Func::consumeAbandoned(Result->first);
       return std::move(Err);
     }
     return Result;
@@ -421,30 +428,54 @@ class RPC : public RPCBase {
   /// Returns an error if serialization fails, otherwise returns a
   /// std::future<Expected<T>> (or a future<Error> for void functions).
   template <typename Func, typename... ArgTs>
-  Expected<AsyncCallResult<Func>> appendCallAsync(ChannelT &C,
-                                                  const ArgTs &... Args) {
-    auto ResAndSeqOrErr = appendCallAsyncWithSeq<Func>(C, Args...);
-    if (ResAndSeqOrErr)
-      return std::move(ResAndSeqOrErr->first);
-    return ResAndSeqOrErr.getError();
+  Expected<NonBlockingCallResult<Func>> appendCallNB(ChannelT &C,
+                                                     const ArgTs &... Args) {
+    auto FutureResAndSeqOrErr = appendCallNBWithSeq<Func>(C, Args...);
+    if (FutureResAndSeqOrErr)
+      return std::move(FutureResAndSeqOrErr->first);
+    return FutureResAndSeqOrErr.getError();
   }
 
-  /// The same as appendCallAsync, except that it calls C.send to flush the
+  /// The same as appendCallNB, except that it calls C.send to flush the
   /// channel after serializing the call.
   template <typename Func, typename... ArgTs>
-  Expected<AsyncCallResult<Func>> callAsync(ChannelT &C,
-                                            const ArgTs &... Args) {
-    auto ResAndSeqOrErr = callAsyncWithSeq<Func>(C, Args...);
-    if (ResAndSeqOrErr)
-      return std::move(ResAndSeqOrErr->first);
-    return ResAndSeqOrErr.getError();
+  Expected<NonBlockingCallResult<Func>> callNB(ChannelT &C,
+                                               const ArgTs &... Args) {
+    auto FutureResAndSeqOrErr = callNBWithSeq<Func>(C, Args...);
+    if (FutureResAndSeqOrErr)
+      return std::move(FutureResAndSeqOrErr->first);
+    return FutureResAndSeqOrErr.getError();
+  }
+
+  /// Call Func on Channel C. Blocks waiting for a result. Returns an Error
+  /// for void functions or an Expected<T> for functions returning a T.
+  ///
+  /// This function is for use in threaded code where another thread is
+  /// handling responses and incoming calls.
+  template <typename Func, typename... ArgTs>
+  typename Func::ErrorReturn callB(ChannelT &C, const ArgTs &... Args) {
+    if (auto FutureResOrErr = callNBWithSeq(C, Args...)) {
+      if (auto Err = C.send()) {
+        abandonOutstandingResults();
+        Func::consumeAbandoned(*FutureResOrErr);
+        return std::move(Err);
+      }
+      return FutureResOrErr->get();
+    } else
+      return FutureResOrErr.takeError();
   }
 
-  /// This can be used in single-threaded mode.
+  /// Call Func on Channel C. Block waiting for a result. While blocked, run
+  /// HandleOther to handle incoming calls (Response calls will be handled
+  /// implicitly before calling HandleOther). Returns an Error for void
+  /// functions or an Expected<T> for functions returning a T.
+  ///
+  /// This function is for use in single threaded mode when the calling thread
+  /// must act as both sender and receiver.
   template <typename Func, typename HandleFtor, typename... ArgTs>
   typename Func::ErrorReturn
   callSTHandling(ChannelT &C, HandleFtor &HandleOther, const ArgTs &... Args) {
-    if (auto ResultAndSeqNoOrErr = callAsyncWithSeq<Func>(C, Args...)) {
+    if (auto ResultAndSeqNoOrErr = callNBWithSeq<Func>(C, Args...)) {
       auto &ResultAndSeqNo = *ResultAndSeqNoOrErr;
       if (auto Err = waitForResult(C, ResultAndSeqNo.second, HandleOther))
         return std::move(Err);
@@ -453,7 +484,8 @@ class RPC : public RPCBase {
       return ResultAndSeqNoOrErr.takeError();
   }
 
-  // This can be used in single-threaded mode.
+  /// Call Func on Channel C. Block waiting for a result. Returns an Error for
+  /// void functions or an Expected<T> for functions returning a T.
   template <typename Func, typename... ArgTs>
   typename Func::ErrorReturn callST(ChannelT &C, const ArgTs &... Args) {
     return callSTHandling<Func>(C, handleNone, Args...);

unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp

@@ -83,7 +83,7 @@ TEST_F(DummyRPC, TestAsyncVoidBool) {
   QueueChannel C2(Q2, Q1);
 
   // Make an async call.
-  auto ResOrErr = callAsyncWithSeq<VoidBool>(C1, true);
+  auto ResOrErr = callNBWithSeq<VoidBool>(C1, true);
   EXPECT_TRUE(!!ResOrErr) << "Simple call over queue failed";
 
   {
@@ -112,7 +112,7 @@ TEST_F(DummyRPC, TestAsyncIntInt) {
   QueueChannel C2(Q2, Q1);
 
   // Make an async call.
-  auto ResOrErr = callAsyncWithSeq<IntInt>(C1, 21);
+  auto ResOrErr = callNBWithSeq<IntInt>(C1, 21);
   EXPECT_TRUE(!!ResOrErr) << "Simple call over queue failed";
 
   {
@@ -143,7 +143,7 @@ TEST_F(DummyRPC, TestSerialization) {
 
   // Make a call to Proc1.
   std::vector<int> v({42, 7});
-  auto ResOrErr = callAsyncWithSeq<AllTheTypes>(
+  auto ResOrErr = callNBWithSeq<AllTheTypes>(
       C1, -101, 250, -10000, 10000, -1000000000, 1000000000, -10000000000,
       10000000000, true, "foo", v);
   EXPECT_TRUE(!!ResOrErr) << "Big (serialization test) call over queue failed";