replica
/
hanchenye-llvm-project
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[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.

llvm-svn: 280051
This commit is contained in:
Lang Hames 2016-08-30 01:57:06 +00:00
parent ea982ed35a
commit 8427a09d58
2 changed files with 70 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
llvm/include/llvm/ExecutionEngine/Orc/RPCUtils.h
View File

@ -71,7 +71,7 @@ protected:
// 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 @@ protected:
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 @@ protected:
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 @@ public:
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 @@ public:
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 @@ public:
/// 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();
}
/// This can be used in single-threaded mode.
/// 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();
}
/// 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 @@ public:
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...);

6
llvm/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp
View File

@ -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";