--- /dev/null
+//===----------------------- 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
//
//===----------------------------------------------------------------------===//
-#include "llvm/ExecutionEngine/Orc/RawByteChannel.h"
#include "llvm/ExecutionEngine/Orc/RPCUtils.h"
+#include "QueueChannel.h"
#include "gtest/gtest.h"
#include <queue>
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 {
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) {}
};
}
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>(
}
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>(
}
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>(
}
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:
}
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>(
}
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>(
}
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>(
}
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>(
}
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>(
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(
[&]() {
}
TEST(DummyRPC, TestRemoveHandler) {
- Queue Q1, Q2;
- DummyRPCEndpoint Server(Q1, Q2);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Server(*Channels.second);
Server.addHandler<DummyRPCAPI::VoidBool>(
[](bool B) {
}
TEST(DummyRPC, TestClearHandlers) {
- Queue Q1, Q2;
- DummyRPCEndpoint Server(Q1, Q2);
+ auto Channels = createPairedQueueChannels();
+ DummyRPCEndpoint Server(*Channels.second);
Server.addHandler<DummyRPCAPI::VoidBool>(
[](bool B) {
projects / llvm / commitdiff