[llvm] r299611 - [Orc] Break QueueChannel out into its own header and add a utility,
Lang Hames via llvm-commits
llvm-commits at lists.llvm.org
Wed Apr 5 18:49:21 PDT 2017
Author: lhames
Date: Wed Apr 5 20:49:21 2017
New Revision: 299611
URL: http://llvm.org/viewvc/llvm-project?rev=299611&view=rev
Log:
[Orc] Break QueueChannel out into its own header and add a utility,
createPairedQueueChannels, to simplify channel creation in the RPC unit tests.
Added:
llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.cpp
llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.h
Modified:
llvm/trunk/unittests/ExecutionEngine/Orc/CMakeLists.txt
llvm/trunk/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp
Modified: llvm/trunk/unittests/ExecutionEngine/Orc/CMakeLists.txt
URL: http://llvm.org/viewvc/llvm-project/llvm/trunk/unittests/ExecutionEngine/Orc/CMakeLists.txt?rev=299611&r1=299610&r2=299611&view=diff
==============================================================================
--- llvm/trunk/unittests/ExecutionEngine/Orc/CMakeLists.txt (original)
+++ llvm/trunk/unittests/ExecutionEngine/Orc/CMakeLists.txt Wed Apr 5 20:49:21 2017
@@ -14,11 +14,12 @@ add_llvm_unittest(OrcJITTests
IndirectionUtilsTest.cpp
GlobalMappingLayerTest.cpp
LazyEmittingLayerTest.cpp
- RTDyldObjectLinkingLayerTest.cpp
ObjectTransformLayerTest.cpp
OrcCAPITest.cpp
OrcTestCommon.cpp
+ QueueChannel.cpp
RPCUtilsTest.cpp
+ RTDyldObjectLinkingLayerTest.cpp
)
target_link_libraries(OrcJITTests ${LLVM_PTHREAD_LIB})
Added: llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.cpp
URL: http://llvm.org/viewvc/llvm-project/llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.cpp?rev=299611&view=auto
==============================================================================
--- llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.cpp (added)
+++ llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.cpp Wed Apr 5 20:49:21 2017
@@ -0,0 +1,14 @@
+//===-------- QueueChannel.cpp - Unit tests the remote executors ----------===//
+//
+// The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+
+#include "QueueChannel.h"
+
+char llvm::QueueChannelError::ID;
+char llvm::QueueChannelClosedError::ID;
+
Added: llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.h
URL: http://llvm.org/viewvc/llvm-project/llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.h?rev=299611&view=auto
==============================================================================
--- llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.h (added)
+++ llvm/trunk/unittests/ExecutionEngine/Orc/QueueChannel.h Wed Apr 5 20:49:21 2017
@@ -0,0 +1,145 @@
+//===----------------------- Queue.h - RPC Queue ------------------*-c++-*-===//
+//
+// The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+
+#ifndef LLVM_UNITTESTS_EXECUTIONENGINE_ORC_QUEUECHANNEL_H
+#define LLVM_UNITTESTS_EXECUTIONENGINE_ORC_QUEUECHANNEL_H
+
+#include "llvm/ExecutionEngine/Orc/RawByteChannel.h"
+#include "llvm/Support/Error.h"
+
+#include <queue>
+
+namespace llvm {
+
+class QueueChannelError : public ErrorInfo<QueueChannelError> {
+public:
+ static char ID;
+};
+
+class QueueChannelClosedError
+ : public ErrorInfo<QueueChannelClosedError, QueueChannelError> {
+public:
+ static char ID;
+ std::error_code convertToErrorCode() const override {
+ return inconvertibleErrorCode();
+ }
+
+ void log(raw_ostream &OS) const override {
+ OS << "Queue closed";
+ }
+};
+
+class Queue : public std::queue<char> {
+public:
+ using ErrorInjector = std::function<Error()>;
+
+ Queue()
+ : ReadError([]() { return Error::success(); }),
+ WriteError([]() { return Error::success(); }) {}
+
+ Queue(const Queue&) = delete;
+ Queue& operator=(const Queue&) = delete;
+ Queue(Queue&&) = delete;
+ Queue& operator=(Queue&&) = delete;
+
+ std::mutex &getMutex() { return M; }
+ std::condition_variable &getCondVar() { return CV; }
+ Error checkReadError() { return ReadError(); }
+ Error checkWriteError() { return WriteError(); }
+ void setReadError(ErrorInjector NewReadError) {
+ {
+ std::lock_guard<std::mutex> Lock(M);
+ ReadError = std::move(NewReadError);
+ }
+ CV.notify_one();
+ }
+ void setWriteError(ErrorInjector NewWriteError) {
+ std::lock_guard<std::mutex> Lock(M);
+ WriteError = std::move(NewWriteError);
+ }
+private:
+ std::mutex M;
+ std::condition_variable CV;
+ std::function<Error()> ReadError, WriteError;
+};
+
+class QueueChannel : public orc::rpc::RawByteChannel {
+public:
+ QueueChannel(std::shared_ptr<Queue> InQueue,
+ std::shared_ptr<Queue> OutQueue)
+ : InQueue(InQueue), OutQueue(OutQueue) {}
+
+ QueueChannel(const QueueChannel&) = delete;
+ QueueChannel& operator=(const QueueChannel&) = delete;
+ QueueChannel(QueueChannel&&) = delete;
+ QueueChannel& operator=(QueueChannel&&) = delete;
+
+ Error readBytes(char *Dst, unsigned Size) override {
+ std::unique_lock<std::mutex> Lock(InQueue->getMutex());
+ while (Size) {
+ {
+ Error Err = InQueue->checkReadError();
+ while (!Err && InQueue->empty()) {
+ InQueue->getCondVar().wait(Lock);
+ Err = InQueue->checkReadError();
+ }
+ if (Err)
+ return Err;
+ }
+ *Dst++ = InQueue->front();
+ --Size;
+ ++NumRead;
+ InQueue->pop();
+ }
+ return Error::success();
+ }
+
+ Error appendBytes(const char *Src, unsigned Size) override {
+ std::unique_lock<std::mutex> Lock(OutQueue->getMutex());
+ while (Size--) {
+ if (Error Err = OutQueue->checkWriteError())
+ return Err;
+ OutQueue->push(*Src++);
+ ++NumWritten;
+ }
+ OutQueue->getCondVar().notify_one();
+ return Error::success();
+ }
+
+ Error send() override { return Error::success(); }
+
+ void close() {
+ auto ChannelClosed = []() { return make_error<QueueChannelClosedError>(); };
+ InQueue->setReadError(ChannelClosed);
+ InQueue->setWriteError(ChannelClosed);
+ OutQueue->setReadError(ChannelClosed);
+ OutQueue->setWriteError(ChannelClosed);
+ }
+
+ uint64_t NumWritten = 0;
+ uint64_t NumRead = 0;
+
+private:
+
+ std::shared_ptr<Queue> InQueue;
+ std::shared_ptr<Queue> OutQueue;
+};
+
+inline std::pair<std::unique_ptr<QueueChannel>, std::unique_ptr<QueueChannel>>
+createPairedQueueChannels() {
+ auto Q1 = std::make_shared<Queue>();
+ auto Q2 = std::make_shared<Queue>();
+ auto C1 = llvm::make_unique<QueueChannel>(Q1, Q2);
+ auto C2 = llvm::make_unique<QueueChannel>(Q2, Q1);
+ return std::make_pair(std::move(C1), std::move(C2));
+}
+
+}
+
+#endif
Modified: llvm/trunk/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp
URL: http://llvm.org/viewvc/llvm-project/llvm/trunk/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp?rev=299611&r1=299610&r2=299611&view=diff
==============================================================================
--- llvm/trunk/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp (original)
+++ llvm/trunk/unittests/ExecutionEngine/Orc/RPCUtilsTest.cpp Wed Apr 5 20:49:21 2017
@@ -7,8 +7,8 @@
//
//===----------------------------------------------------------------------===//
-#include "llvm/ExecutionEngine/Orc/RawByteChannel.h"
#include "llvm/ExecutionEngine/Orc/RPCUtils.h"
+#include "QueueChannel.h"
#include "gtest/gtest.h"
#include <queue>
@@ -17,47 +17,6 @@ using namespace llvm;
using namespace llvm::orc;
using namespace llvm::orc::rpc;
-class Queue : public std::queue<char> {
-public:
- std::mutex &getMutex() { return M; }
- std::condition_variable &getCondVar() { return CV; }
-private:
- std::mutex M;
- std::condition_variable CV;
-};
-
-class QueueChannel : public RawByteChannel {
-public:
- QueueChannel(Queue &InQueue, Queue &OutQueue)
- : InQueue(InQueue), OutQueue(OutQueue) {}
-
- Error readBytes(char *Dst, unsigned Size) override {
- std::unique_lock<std::mutex> Lock(InQueue.getMutex());
- while (Size) {
- while (InQueue.empty())
- InQueue.getCondVar().wait(Lock);
- *Dst++ = InQueue.front();
- --Size;
- InQueue.pop();
- }
- return Error::success();
- }
-
- Error appendBytes(const char *Src, unsigned Size) override {
- std::unique_lock<std::mutex> Lock(OutQueue.getMutex());
- while (Size--)
- OutQueue.push(*Src++);
- OutQueue.getCondVar().notify_one();
- return Error::success();
- }
-
- Error send() override { return Error::success(); }
-
-private:
- Queue &InQueue;
- Queue &OutQueue;
-};
-
class RPCFoo {};
namespace llvm {
@@ -143,10 +102,8 @@ namespace DummyRPCAPI {
class DummyRPCEndpoint : public SingleThreadedRPCEndpoint<QueueChannel> {
public:
- DummyRPCEndpoint(Queue &Q1, Queue &Q2)
- : SingleThreadedRPCEndpoint(C, true), C(Q1, Q2) {}
-private:
- QueueChannel C;
+ DummyRPCEndpoint(QueueChannel &C)
+ : SingleThreadedRPCEndpoint(C, true) {}
};
@@ -154,15 +111,15 @@ void freeVoidBool(bool B) {
}
TEST(DummyRPC, TestFreeFunctionHandler) {
- Queue Q1, Q2;
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Server(*Channels.first);
Server.addHandler<DummyRPCAPI::VoidBool>(freeVoidBool);
}
TEST(DummyRPC, TestCallAsyncVoidBool) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::VoidBool>(
@@ -204,9 +161,9 @@ TEST(DummyRPC, TestCallAsyncVoidBool) {
}
TEST(DummyRPC, TestCallAsyncIntInt) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::IntInt>(
@@ -249,9 +206,9 @@ TEST(DummyRPC, TestCallAsyncIntInt) {
}
TEST(DummyRPC, TestAsyncIntIntHandler) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addAsyncHandler<DummyRPCAPI::IntInt>(
@@ -295,9 +252,9 @@ TEST(DummyRPC, TestAsyncIntIntHandler) {
}
TEST(DummyRPC, TestAsyncIntIntHandlerMethod) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
class Dummy {
public:
@@ -346,9 +303,9 @@ TEST(DummyRPC, TestAsyncIntIntHandlerMet
}
TEST(DummyRPC, TestCallAsyncVoidString) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::VoidString>(
@@ -386,9 +343,9 @@ TEST(DummyRPC, TestCallAsyncVoidString)
}
TEST(DummyRPC, TestSerialization) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::AllTheTypes>(
@@ -451,9 +408,9 @@ TEST(DummyRPC, TestSerialization) {
}
TEST(DummyRPC, TestCustomType) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::CustomType>(
@@ -494,9 +451,9 @@ TEST(DummyRPC, TestCustomType) {
}
TEST(DummyRPC, TestWithAltCustomType) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::CustomType>(
@@ -537,9 +494,9 @@ TEST(DummyRPC, TestWithAltCustomType) {
}
TEST(DummyRPC, TestParallelCallGroup) {
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread([&]() {
Server.addHandler<DummyRPCAPI::IntInt>(
@@ -619,9 +576,9 @@ TEST(DummyRPC, TestAPICalls) {
static_assert(!DummyCalls1::Contains<DummyRPCAPI::CustomType>::value,
"Contains<Func> template should return false here");
- Queue Q1, Q2;
- DummyRPCEndpoint Client(Q1, Q2);
- DummyRPCEndpoint Server(Q2, Q1);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Client(*Channels.first);
+ DummyRPCEndpoint Server(*Channels.second);
std::thread ServerThread(
[&]() {
@@ -657,8 +614,8 @@ TEST(DummyRPC, TestAPICalls) {
}
TEST(DummyRPC, TestRemoveHandler) {
- Queue Q1, Q2;
- DummyRPCEndpoint Server(Q1, Q2);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Server(*Channels.second);
Server.addHandler<DummyRPCAPI::VoidBool>(
[](bool B) {
@@ -670,8 +627,8 @@ TEST(DummyRPC, TestRemoveHandler) {
}
TEST(DummyRPC, TestClearHandlers) {
- Queue Q1, Q2;
- DummyRPCEndpoint Server(Q1, Q2);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Server(*Channels.second);
Server.addHandler<DummyRPCAPI::VoidBool>(
[](bool B) {
More information about the llvm-commits
mailing list