[llvm] [orc-rt] Reduce SimpleRemoteCA to transport-independent utils (PR #223141)
Lang Hames via llvm-commits
llvm-commits at lists.llvm.org
Sat Sep 12 04:49:35 PDT 2026
https://github.com/lhames created https://github.com/llvm/llvm-project/pull/223141
Reduce SimpleRemoteCA to the transport-independent parts: opcodes, setup and hang-up payload encoding, message validation and dispatch, and the table of calls awaiting a result. Subclasses are now responsible for the connection state and synchronization, since they'll usually want to implement these things for their transports anyway (or the transport will already manage connection-state/synchronization).
takePendingCall becomes pure virtual, since subclasses must synchronize access to the pending-call table.
The header moves from orc-rt-internal/ to orc-rt/: with the transport a subclass concern, there's no reason to preclude library clients from writing one.
>From 19e9b06f5d2e38c2e8003262f9b673eec2d7011a Mon Sep 17 00:00:00 2001
From: Lang Hames <lhames at gmail.com>
Date: Tue, 8 Sep 2026 15:53:41 +1000
Subject: [PATCH] [orc-rt] Reduce SimpleRemoteCA to transport-independent utils
Reduce SimpleRemoteCA to the transport-independent parts: opcodes, setup
and hang-up payload encoding, message validation and dispatch, and the
table of calls awaiting a result. Subclasses are now responsible for the
connection state and synchronization, since they'll usually want to
implement these things for their transports anyway (or the transport will
already manage connection-state/synchronization).
takePendingCall becomes pure virtual, since subclasses must synchronize
access to the pending-call table.
The header moves from orc-rt-internal/ to orc-rt/: with the transport a
subclass concern, there's no reason to preclude library clients from
writing one.
---
.../orc-rt/bedrock/sps/SimpleRemoteCA.h | 180 +++----
orc-rt/lib/bedrock/sps/SimpleRemoteCA.cpp | 181 +++----
.../unit/bedrock/sps/SimpleRemoteCATest.cpp | 452 ++++--------------
3 files changed, 230 insertions(+), 583 deletions(-)
diff --git a/orc-rt/include/orc-rt/bedrock/sps/SimpleRemoteCA.h b/orc-rt/include/orc-rt/bedrock/sps/SimpleRemoteCA.h
index 35742b6adcd68..bf34a9adaebd2 100644
--- a/orc-rt/include/orc-rt/bedrock/sps/SimpleRemoteCA.h
+++ b/orc-rt/include/orc-rt/bedrock/sps/SimpleRemoteCA.h
@@ -6,7 +6,7 @@
//
//===----------------------------------------------------------------------===//
//
-// A ControllerAccess base class implementing the SimpleRemote wire protocol.
+// Transport-independent half of the SimpleRemote protocol.
//
//===----------------------------------------------------------------------===//
@@ -19,140 +19,114 @@
#include "orc-rt/support/ExecutorAddress.h"
#include "orc-rt/support/WrapperFunction.h"
-#include <mutex>
-#include <optional>
+#include <cstdint>
#include <unordered_map>
namespace orc_rt {
-/// ControllerAccess base for the SimpleRemote protocol.
+/// Base for ControllerAccess implementations that speak the SimpleRemote
+/// protocol: payload encoding, message semantics, and the table of calls
+/// awaiting a result. A subclass supplies the transport -- framing, I/O, and
+/// the synchronization those need.
///
-/// Implements the protocol semantics -- setup and hang-up message encoding,
-/// message dispatch and validation, pending-call tracking, connection state and
-/// teardown sequencing -- while leaving wire framing and byte transport to
-/// subclasses. Subclasses feed received messages to handleMessage and implement
-/// two hooks, sendMessage and beginTeardown.
+/// This class does not enforce synchronization or track connection state: that
+/// is left to subclasses (who usually have those those things for the sake of
+/// the transport). Subclasses must ensure that notifyDisconnect is called
+/// exactly once, and that every registered handler is notified with either a
+/// result or an error.
class SimpleRemoteCA : public Session::ControllerAccess {
-public:
- void disconnect() final;
-
- void callController(OnControllerCallReturn OnComplete,
- orc_rt_ControllerHandlerTag T,
- WrapperFunctionBuffer ArgBytes) final;
-
- void sendWrapperResult(WrapperFunctionBuffer ResultBytes,
- uint64_t CallId) final;
-
protected:
/// SimpleRemote message opcodes.
///
- /// These values are on-the-wire values, shared with LLVM's
- /// SimpleRemoteEPCOpcode: do not renumber or reorder them.
- enum class Opcode { Setup, Hangup, Result, Call, LastOpcode = Call };
+ /// On-the-wire values, shared with LLVM's SimpleRemoteEPCOpcode: do not
+ /// renumber or reorder.
+ enum class Opcode : uint64_t {
+ Setup,
+ Hangup,
+ Result,
+ Call,
+ LastOpcode = Call
+ };
- /// Result of handleMessage: whether the session continues, or the controller
- /// has hung up and the session should end.
+ /// The name of Op, for logging.
+ static const char *getOpcodeName(Opcode Op) noexcept;
+
+ /// Whether the session continues, or the controller has hung up and it
+ /// should end.
enum class Action { Continue, End };
- static const char *getOpcodeName(Opcode Op) noexcept;
+ using PendingCallsMap = std::unordered_map<uint64_t, OnControllerCallReturn>;
SimpleRemoteCA(Session &S) : ControllerAccess(S) {}
+ /// Serializes BI as the payload of a setup message.
+ static WrapperFunctionBuffer encodeSetup(const BootstrapInfo &BI);
+
/// Serializes Err as the payload of a hang-up message.
///
- /// A hang-up always carries a serialized Error saying why the session is
- /// ending: a success value for an orderly disconnect, otherwise the reason.
- static WrapperFunctionBuffer encodeHangupPayload(Error Err);
+ /// A hang-up always carries a reason: success for an orderly disconnect,
+ /// otherwise what went wrong.
+ static WrapperFunctionBuffer encodeHangup(Error Err);
- /// Starts accepting controller calls and sends the setup message. Subclasses
- /// call this from connect once the transport can carry messages.
- ///
- /// Safe to call whether or not the transport has started receiving. A
- /// transport that has already dropped out and finished teardown gets a no-op:
- /// the Session has been notified, and no setup is sent.
- void beginAccepting(const BootstrapInfo &BI);
-
- /// Completes teardown: drains pending calls, then notifies the Session
- /// exactly once. Subclasses call this when the transport is finished, however
- /// teardown began.
+ /// Decodes a hang-up payload produced by encodeHangup, returning the reason
+ /// it carries.
///
- /// Call it exactly once.
+ /// A payload that will not decode -- including an empty one, which is never
+ /// valid -- comes back as an error describing itself. Both outcomes end the
+ /// session, so both are reported the same way.
+ static Error decodeHangup(WrapperFunctionBuffer Payload);
+
+ /// Registers OnComplete and returns the sequence number to send its call
+ /// under.
///
- /// Err is the disconnection mode: success for an orderly hang-up from either
- /// side, otherwise what went wrong. May synchronously destroy *this, so
- /// nothing may touch it afterwards.
- void finishTeardown(Error Err);
-
- /// Dispatches a received message. Transports call this once per message,
- /// after de-framing. OpC is the raw wire opcode: this validates it and the
- /// per-opcode header semantics. Transports are responsible only for verifying
- /// that the payload deserializes as the message requires.
+ /// Unsynchronized. Register and queue the message in one critical section, so
+ /// that a call is never left pending with nothing to answer it, nor sent with
+ /// no handler to complete.
+ uint64_t registerCall(OnControllerCallReturn OnComplete);
+
+ /// Removes and returns the handler registered under SeqNo, or a null handler
+ /// if there is none.
///
- /// Every error returned here is terminal for the session: shut the transport
- /// down and pass the error to finishTeardown, rather than continuing to read.
- /// Action::End means the controller hung up cleanly -- shut down and pass
- /// Error::success().
+ /// Unsynchronized. Run the handler outside the lock: it runs managed code.
+ OnControllerCallReturn takeCall(uint64_t SeqNo);
+
+ /// Removes and returns every registered handler, for a transport to fail on
+ /// its way out.
///
- /// Calls may come from any thread, but must be serialized with one another
- /// and must all complete before finishTeardown: the Result path completes a
- /// pending call, which is only legal while the managed-code group is still
- /// open.
- Expected<Action> handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
- WrapperFunctionBuffer Payload);
+ /// Unsynchronized. Call before notifying the Session, while the managed-code
+ /// group is still open, or the handlers are dropped rather than dispatched.
+ PendingCallsMap takeAllCalls();
- /// Hands a framed message to the transport. Called with no lock held, so
- /// framing or packaging happens off the critical section.
+ /// Acts on one de-framed message. OpC is the raw wire opcode: this validates
+ /// it along with the header semantics each opcode requires, so a transport
+ /// need only deliver the fields and payload intact.
///
- /// Best-effort: a transport that has gone away should drop the message rather
- /// than report anything. A Call's handler is failed by the drain, and a
- /// Result has no pending call on this side, so nothing is left unsettled.
- virtual void sendMessage(Opcode Op, uint64_t SeqNo,
- orc_rt_ControllerHandlerTag T,
- WrapperFunctionBuffer Payload) = 0;
-
- /// Starts shutting the transport down. Must not block: teardown holds up
- /// Session detach. Call finishTeardown once the transport is finished.
+ /// Every error returned is terminal: stop reading and end the session with
+ /// it. Action::End means the controller hung up cleanly.
///
- /// Called only for a local disconnect, which is orderly by definition, so
- /// make a best effort to send encodeHangupPayload(Error::success()) as the
- /// last message before shutting down. Only the transport can place it last.
+ /// Call with no transport lock held: dispatching a Call, and completing a
+ /// Result, run managed code that may re-enter this object -- through
+ /// callController, and so into that same lock.
///
- /// Teardown the transport initiates instead -- a hang-up from the controller,
- /// or an error out of handleMessage -- goes straight to finishTeardown,
- /// having sent its own reason first if it has one.
- virtual void beginTeardown() = 0;
-
-private:
- /// Whether the connection will accept new controller calls, and how far
- /// teardown has progressed. Guarded by M.
- enum class State {
- NotConnected, ///< Before beginAccepting, or connect failed.
- Accepting, ///< Calls may be registered.
- TearingDown, ///< Latched closed; the transport is shutting down.
- Disconnected, ///< finishTeardown has run; the Session has been notified.
- };
+ /// Calls must be serialized with one another, and must all complete before
+ /// the Session is notified, since a Result completes a pending call.
+ Expected<Action> handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
+ WrapperFunctionBuffer Payload);
- /// Registers OnComplete and returns the sequence number to send the call
- /// under, or nullopt if the connection is no longer accepting calls -- in
- /// which case OnComplete is untouched and the caller must fail it inline.
+ /// Removes the handler for SeqNo, or returns a null handler if there is none.
///
- /// The state check and the registration are one critical section, so a call
- /// racing teardown is either drained by finishTeardown or failed inline:
- /// never left pending with no result to come.
- std::optional<uint64_t> tryRegisterCall(OnControllerCallReturn &OnComplete);
+ /// handleMessage needs this to complete a Result, and a result arriving on
+ /// the transport's reader can race a call being registered on another thread.
+ /// Implement it as takeCall under whatever guards registerCall.
+ virtual OnControllerCallReturn takePendingCall(uint64_t SeqNo) = 0;
- void failAllPendingCalls();
+private:
+ /// Completes the pending call SeqNo with ResultBytes. Fails if no such call
+ /// is outstanding, which means the peer answered a call that was never made.
Error handleResult(uint64_t SeqNo, WrapperFunctionBuffer ResultBytes);
- static WrapperFunctionBuffer encodeSetupMessage(const BootstrapInfo &BI);
-
- std::mutex M;
- State ConnState = State::NotConnected;
-
- // SeqNo zero is reserved for messages with no pending result (Setup, Hangup).
+ // Guarded by the transport's lock. See the class comment.
uint64_t NextSeqNo = 1;
-
- using PendingCallsMap = std::unordered_map<uint64_t, OnControllerCallReturn>;
PendingCallsMap PendingCalls;
};
diff --git a/orc-rt/lib/bedrock/sps/SimpleRemoteCA.cpp b/orc-rt/lib/bedrock/sps/SimpleRemoteCA.cpp
index fde228e102f2d..8082666e22461 100644
--- a/orc-rt/lib/bedrock/sps/SimpleRemoteCA.cpp
+++ b/orc-rt/lib/bedrock/sps/SimpleRemoteCA.cpp
@@ -6,93 +6,21 @@
//
//===----------------------------------------------------------------------===//
//
-// SimpleRemote-protocol ControllerAccess base class.
+// Transport-independent half of the SimpleRemote protocol.
//
//===----------------------------------------------------------------------===//
#include "orc-rt/bedrock/sps/SimpleRemoteCA.h"
#include "orc-rt/support/Compiler.h"
+#include "orc-rt/support/iterator_range.h"
#include "orc-rt/support/sps/SimplePackedSerialization.h"
-#include <cassert>
#include <string>
+#include <utility>
namespace orc_rt {
-void SimpleRemoteCA::disconnect() {
- {
- std::scoped_lock<std::mutex> Lock(M);
- // Not Accepting means teardown is already under way or finished, or the
- // connection never opened. All are no-ops: the Session tolerates a
- // disconnect racing a controller-initiated one.
- if (ConnState != State::Accepting)
- return;
- ConnState = State::TearingDown;
- }
-
- // The transport sends the hang-up itself, so that it lands after the queue it
- // is about to discard rather than behind a result that raced this.
- beginTeardown();
-}
-
-void SimpleRemoteCA::callController(OnControllerCallReturn OnComplete,
- orc_rt_ControllerHandlerTag T,
- WrapperFunctionBuffer ArgBytes) {
- // Sent outside tryRegisterCall's critical section, so framing and packaging
- // stay off it. Safe because the Session holds a shared_ptr<ControllerAccess>
- // across this call, so a teardown racing the send cannot destroy *this.
- if (auto SeqNo = tryRegisterCall(OnComplete))
- return sendMessage(Opcode::Call, *SeqNo, T, std::move(ArgBytes));
-
- // The connection is gone, so no result can arrive. The caller is still on the
- // stack, so fail the handler there.
- failControllerCallInline(std::move(OnComplete));
-}
-
-void SimpleRemoteCA::sendWrapperResult(WrapperFunctionBuffer ResultBytes,
- uint64_t CallId) {
- // No state check: a result has no pending call on this side, so a transport
- // that has gone away can drop it with nothing left unsettled.
- sendMessage(Opcode::Result, CallId, nullptr, std::move(ResultBytes));
-}
-
-void SimpleRemoteCA::beginAccepting(const BootstrapInfo &BI) {
- {
- std::scoped_lock<std::mutex> Lock(M);
- // A transport that enabled delivery before calling this can have dropped
- // out and completed teardown already. The Session has been notified, so
- // there is nothing left to accept on and nothing to send.
- if (ConnState == State::Disconnected)
- return;
- assert(ConnState == State::NotConnected && "beginAccepting called twice");
- ConnState = State::Accepting;
- }
-
- // State first: a teardown landing in the window then closes an accepting
- // connection, where sending first would leave this to reopen one teardown had
- // just closed. A call landing there registers instead of failing spuriously.
- // Setup may reach a transport that has already gone, which drops it.
- sendMessage(Opcode::Setup, 0, nullptr, encodeSetupMessage(BI));
-}
-
-void SimpleRemoteCA::finishTeardown(Error Err) {
- {
- std::scoped_lock<std::mutex> Lock(M);
- // Transports owe exactly one call, and both natural implementations give
- // it: a socket reactor on its way out, and XPC on the single invalidation
- // event it guarantees.
- assert(ConnState != State::Disconnected &&
- "finishTeardown called more than once");
- ConnState = State::Disconnected;
- }
-
- // The drain must precede the notification, while the keepalive group is still
- // open, or the handlers are dropped rather than dispatched.
- failAllPendingCalls();
- notifyDisconnected(std::move(Err));
-}
-
const char *SimpleRemoteCA::getOpcodeName(Opcode Op) noexcept {
switch (Op) {
case Opcode::Setup:
@@ -107,8 +35,7 @@ const char *SimpleRemoteCA::getOpcodeName(Opcode Op) noexcept {
ORC_RT_UNREACHABLE("Unrecognized opcode");
}
-WrapperFunctionBuffer
-SimpleRemoteCA::encodeSetupMessage(const BootstrapInfo &BI) {
+WrapperFunctionBuffer SimpleRemoteCA::encodeSetup(const BootstrapInfo &BI) {
using SPSSetup = SPSTuple<SPSString, uint64_t,
SPSSequence<SPSTuple<SPSString, SPSSequence<char>>>,
SPSSequence<SPSTuple<SPSString, SPSExecutorAddr>>>;
@@ -117,31 +44,59 @@ SimpleRemoteCA::encodeSetupMessage(const BootstrapInfo &BI) {
// FIXME: Remove once we allow size_t serialization.
uint64_t PageSize = BI.processInfo().pageSize();
auto Symbols = iterator_range(BI.symbols());
- auto BootstrapTuple =
+ auto Tuple =
std::tie(BI.processInfo().targetTriple(), PageSize, BI.values(), Symbols);
+ using Serialize = SPSSerializationTraits<SPSSetup, decltype(Tuple)>;
- using SetupSerialize =
- SPSSerializationTraits<SPSSetup, decltype(BootstrapTuple)>;
-
- auto Payload =
- WrapperFunctionBuffer::allocate(SetupSerialize::size(BootstrapTuple));
+ auto Payload = WrapperFunctionBuffer::allocate(Serialize::size(Tuple));
SPSOutputBuffer OB(Payload.data(), Payload.size());
- if (!SetupSerialize::serialize(OB, BootstrapTuple))
+ if (!Serialize::serialize(OB, Tuple))
ORC_RT_UNREACHABLE("serialization should not fail");
-
return Payload;
}
-WrapperFunctionBuffer SimpleRemoteCA::encodeHangupPayload(Error Err) {
+WrapperFunctionBuffer SimpleRemoteCA::encodeHangup(Error Err) {
SPSSerializableError SE(std::move(Err));
- using SPSSerialize = SPSArgList<SPSError>;
- auto Payload = WrapperFunctionBuffer::allocate(SPSSerialize::size(SE));
+ using Serialize = SPSArgList<SPSError>;
+ auto Payload = WrapperFunctionBuffer::allocate(Serialize::size(SE));
SPSOutputBuffer OB(Payload.data(), Payload.size());
- if (!SPSSerialize::serialize(OB, SE))
+ if (!Serialize::serialize(OB, SE))
ORC_RT_UNREACHABLE("serialization should not fail");
return Payload;
}
+Error SimpleRemoteCA::decodeHangup(WrapperFunctionBuffer Payload) {
+ SPSSerializableError SE;
+ SPSInputBuffer IB(Payload.data(), Payload.size());
+ if (!SPSArgList<SPSError>::deserialize(IB, SE))
+ return make_error<StringError>(
+ "Malformed hang-up message: could not deserialize reason");
+ return SE.toError();
+}
+
+uint64_t SimpleRemoteCA::registerCall(OnControllerCallReturn OnComplete) {
+ assert(OnComplete && "Registered handler must contain a value");
+ uint64_t SeqNo = NextSeqNo++;
+ PendingCalls.try_emplace(SeqNo, std::move(OnComplete));
+ return SeqNo;
+}
+
+SimpleRemoteCA::OnControllerCallReturn
+SimpleRemoteCA::takeCall(uint64_t SeqNo) {
+ auto I = PendingCalls.find(SeqNo);
+ if (I == PendingCalls.end())
+ return OnControllerCallReturn();
+ auto OnComplete = std::move(I->second);
+ PendingCalls.erase(I);
+ return OnComplete;
+}
+
+SimpleRemoteCA::PendingCallsMap SimpleRemoteCA::takeAllCalls() {
+ PendingCallsMap Taken;
+ std::swap(Taken, PendingCalls);
+ return Taken;
+}
+
Expected<SimpleRemoteCA::Action>
SimpleRemoteCA::handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
WrapperFunctionBuffer Payload) {
@@ -151,22 +106,19 @@ SimpleRemoteCA::handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
switch (static_cast<Opcode>(OpC)) {
case Opcode::Setup:
return make_error<StringError>("Unexpected Setup message");
+
case Opcode::Hangup: {
// A hang-up carries no sequence number or tag, and a payload holding the
// reason the controller is going away.
if (SeqNo != 0 || Tag)
return make_error<StringError>("Malformed hang-up message");
- SPSSerializableError SE;
- SPSInputBuffer IB(Payload.data(), Payload.size());
- if (!SPSArgList<SPSError>::deserialize(IB, SE))
- return make_error<StringError>(
- "Malformed hang-up message: could not deserialize reason");
- // An orderly hang-up ends the session; one carrying a reason ends it with
- // that reason, which the reactor reports as the disconnection error.
- if (Error Err = SE.toError())
+ // A reason ends the session with that reason; an orderly hang-up, or a
+ // payload that will not decode, just ends it.
+ if (Error Err = decodeHangup(std::move(Payload)))
return std::move(Err);
return Action::End;
}
+
case Opcode::Result:
// A result is not associated with a handler tag.
if (Tag)
@@ -175,6 +127,7 @@ SimpleRemoteCA::handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
if (auto Err = handleResult(SeqNo, std::move(Payload)))
return std::move(Err);
return Action::Continue;
+
case Opcode::Call:
handleWrapperCall(Tag.toPtr<orc_rt_WrapperFunction>(), std::move(Payload),
SeqNo);
@@ -183,38 +136,12 @@ SimpleRemoteCA::handleMessage(uint64_t OpC, uint64_t SeqNo, ExecutorAddr Tag,
ORC_RT_UNREACHABLE("Unrecognized opcode");
}
-std::optional<uint64_t>
-SimpleRemoteCA::tryRegisterCall(OnControllerCallReturn &OnComplete) {
- std::scoped_lock<std::mutex> Lock(M);
- if (ConnState != State::Accepting)
- return std::nullopt;
- PendingCalls.try_emplace(NextSeqNo, std::move(OnComplete));
- return NextSeqNo++;
-}
-
-void SimpleRemoteCA::failAllPendingCalls() {
- PendingCallsMap Failed;
- {
- std::scoped_lock<std::mutex> Lock(M);
- std::swap(Failed, PendingCalls);
- }
-
- for (auto &[SeqNo, OnComplete] : Failed)
- failPendingControllerCall(std::move(OnComplete));
-}
-
Error SimpleRemoteCA::handleResult(uint64_t SeqNo,
WrapperFunctionBuffer ResultBytes) {
- OnControllerCallReturn OnComplete;
- {
- std::scoped_lock<std::mutex> Lock(M);
- auto I = PendingCalls.find(SeqNo);
- if (I == PendingCalls.end())
- return make_error<StringError>("No pending call for sequence number " +
- std::to_string(SeqNo));
- OnComplete = std::move(I->second);
- PendingCalls.erase(I);
- }
+ auto OnComplete = takePendingCall(SeqNo);
+ if (!OnComplete)
+ return make_error<StringError>("No pending call for sequence number " +
+ std::to_string(SeqNo));
handleControllerCallResult(std::move(OnComplete), std::move(ResultBytes));
return Error::success();
diff --git a/orc-rt/test/unit/bedrock/sps/SimpleRemoteCATest.cpp b/orc-rt/test/unit/bedrock/sps/SimpleRemoteCATest.cpp
index 0e580edd1aeb1..cb96c938756f3 100644
--- a/orc-rt/test/unit/bedrock/sps/SimpleRemoteCATest.cpp
+++ b/orc-rt/test/unit/bedrock/sps/SimpleRemoteCATest.cpp
@@ -6,11 +6,11 @@
//
//===----------------------------------------------------------------------===//
//
-// Tests for orc-rt's SimpleRemoteCA protocol base class.
+// Tests for the transport-independent half of the SimpleRemote protocol.
//
-// SimpleRemoteCA is transport-independent, so these tests drive its protocol
-// operations directly through a capture-only test double and observe the
-// results via a Session.
+// These cover the payload encodings and the pending-call table on their own.
+// When a call may be registered, what ends a session, and who notifies the
+// Session are all the transport's to decide, and are tested with it.
//
//===----------------------------------------------------------------------===//
@@ -23,110 +23,65 @@
#include "orc-rt/support/sps/SimplePackedSerialization.h"
-#include <deque>
-#include <optional>
#include <string>
#include <unordered_map>
-#include <utility>
#include <vector>
using namespace orc_rt;
namespace {
-// A SimpleRemoteCA with no real transport. It exposes the protected protocol
-// operations for tests, records every message the base asks it to send, and
-// completes teardown as soon as the base begins it -- the shortest thing a real
-// transport could do.
-class TestSimpleRemoteCA : public SimpleRemoteCA {
+/// Exposes the protected utilities, and captures the handler that
+/// Session::callController hands over so tests have a real one to register.
+class TestCA : public SimpleRemoteCA {
public:
- using SimpleRemoteCA::Action;
- using SimpleRemoteCA::encodeHangupPayload;
- using SimpleRemoteCA::finishTeardown;
- using SimpleRemoteCA::handleMessage;
- using SimpleRemoteCA::Opcode;
-
- // A message the base handed to the transport.
- struct Sent {
- Opcode Op;
- uint64_t SeqNo;
- orc_rt_ControllerHandlerTag Tag;
- WrapperFunctionBuffer Payload;
- };
-
- /// Messages is owned by the caller so that it outlives a CA that teardown
- /// destroys.
- TestSimpleRemoteCA(Session &S, std::deque<Sent> &Messages,
- TestSimpleRemoteCA **Self = nullptr,
- bool DropOutDuringConnect = false)
- : SimpleRemoteCA(S), Messages(Messages),
- DropOutDuringConnect(DropOutDuringConnect) {
+ using ControllerAccess::failPendingControllerCall;
+ using ControllerAccess::OnControllerCallReturn;
+ using SimpleRemoteCA::decodeHangup;
+ using SimpleRemoteCA::encodeHangup;
+ using SimpleRemoteCA::encodeSetup;
+ using SimpleRemoteCA::PendingCallsMap;
+ using SimpleRemoteCA::registerCall;
+ using SimpleRemoteCA::takeAllCalls;
+ using SimpleRemoteCA::takeCall;
+
+ TestCA(Session &S, TestCA **Self = nullptr) : SimpleRemoteCA(S) {
if (Self)
*Self = this;
}
- void connect(BootstrapInfo BI) override {
- // Models a transport that enabled delivery and then dropped out before it
- // got as far as accepting -- the sanctioned failed-connect path.
- if (DropOutDuringConnect)
- finishTeardown(make_error<StringError>("dropped during connect"));
- beginAccepting(BI);
- }
+ void connect(BootstrapInfo BI) override {}
- void sendMessage(Opcode Op, uint64_t SeqNo, orc_rt_ControllerHandlerTag Tag,
- WrapperFunctionBuffer Payload) override {
- Messages.push_back(Sent{Op, SeqNo, Tag, std::move(Payload)});
+ void disconnect() override {
+ // What a transport owes on the way out: fail what is pending, then notify.
+ for (auto &[SeqNo, OnComplete] : takeAllCalls())
+ failPendingControllerCall(std::move(OnComplete));
+ notifyDisconnected(Error::success());
}
- void beginTeardown() override {
- ++TeardownsBegun;
- // What a transport owes for a local disconnect: an orderly reason, sent
- // last.
- sendMessage(Opcode::Hangup, 0, nullptr,
- encodeHangupPayload(Error::success()));
- if (FinishTeardownOnBegin)
- finishTeardown(Error::success());
+ void callController(OnControllerCallReturn OnComplete,
+ orc_rt_ControllerHandlerTag T,
+ WrapperFunctionBuffer ArgBytes) override {
+ Captured = std::move(OnComplete);
}
- // A deque, not a vector: messages() returns pointers into this, and a later
- // send must not invalidate them.
- std::deque<Sent> &Messages;
- /// Clear to model a transport whose shutdown is asynchronous.
- bool FinishTeardownOnBegin = true;
- /// Set to tear down from inside connect, before beginAccepting runs.
- bool DropOutDuringConnect;
-
- unsigned TeardownsBegun = 0;
-};
-
-using SentMessages = std::deque<TestSimpleRemoteCA::Sent>;
-
-// Messages of the given opcode, in send order.
-std::vector<const TestSimpleRemoteCA::Sent *>
-messages(const SentMessages &Msgs, TestSimpleRemoteCA::Opcode Op) {
- std::vector<const TestSimpleRemoteCA::Sent *> Result;
- for (auto &M : Msgs)
- if (M.Op == Op)
- Result.push_back(&M);
- return Result;
-}
+ void sendWrapperResult(WrapperFunctionBuffer ResultBytes,
+ uint64_t CallId) override {}
-constexpr uint64_t opc(TestSimpleRemoteCA::Opcode Op) {
- return static_cast<uint64_t>(Op);
-}
+ /// Single-threaded here, so no lock: a real transport takes its own.
+ OnControllerCallReturn takePendingCall(uint64_t SeqNo) override {
+ return takeCall(SeqNo);
+ }
-// Wrapper that echoes its arguments back as the result.
-void echoWrapper(orc_rt_SessionRef S, orc_rt_WrapperFunctionBuffer ArgBytes,
- orc_rt_WrapperFunctionReturn Return, uint64_t CallId) {
- Return(S, ArgBytes, CallId);
-}
+ /// A handler taken from the Session, for registering.
+ OnControllerCallReturn Captured;
+};
-// Expect an Expected<T> to hold an error whose message equals ExpectedMsg.
-template <typename T> void expectError(Expected<T> R, const char *ExpectedMsg) {
- if (R)
- ADD_FAILURE() << "expected error \"" << ExpectedMsg << "\", got a value";
- else
- EXPECT_EQ(toString(R.takeError()), ExpectedMsg);
+/// Drives Session::callController once and returns the handler it produced.
+TestCA::OnControllerCallReturn borrowHandler(Session &S, TestCA &CA) {
+ S.callController([](WrapperFunctionBuffer) {}, nullptr,
+ WrapperFunctionBuffer());
+ return std::move(CA.Captured);
}
} // namespace
@@ -139,9 +94,6 @@ TEST(SimpleRemoteCATest, SetupMessageRoundTrips) {
SPSSequence<SPSTuple<SPSString, SPSSequence<char>>>,
SPSSequence<SPSTuple<SPSString, SPSExecutorAddr>>>;
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
int SomeSymbol = 0;
@@ -152,14 +104,7 @@ TEST(SimpleRemoteCATest, SetupMessageRoundTrips) {
BootstrapInfo BI(S, std::move(Symbols),
BootstrapInfo::ValueMap{{"key", "value"}});
-
- // connect sends setup; take the payload from the message the base produced.
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(std::move(BI), Msgs, &CA);
- ASSERT_TRUE(CA);
- auto Setups = messages(Msgs, TestSimpleRemoteCA::Opcode::Setup);
- ASSERT_EQ(Setups.size(), 1u);
- auto &Payload = Setups[0]->Payload;
+ auto Payload = TestCA::encodeSetup(BI);
std::string Triple;
uint64_t PageSize = 0;
@@ -179,288 +124,89 @@ TEST(SimpleRemoteCATest, SetupMessageRoundTrips) {
// The payload holds exactly the setup fields: no padding, and nothing the
// controller would be left to interpret.
EXPECT_EQ(static_cast<size_t>(IB.data() - Payload.data()), Payload.size());
-}
-
-TEST(SimpleRemoteCATest, HandleMessageRejectsInvalidOpcode) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- expectError(CA.handleMessage(/*OpC=*/99, /*SeqNo=*/0, ExecutorAddr(),
- WrapperFunctionBuffer()),
- "Invalid opcode 99");
-}
-
-TEST(SimpleRemoteCATest, HandleMessageRejectsUnexpectedSetup) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- expectError(CA.handleMessage(opc(TestSimpleRemoteCA::Opcode::Setup), 0,
- ExecutorAddr(), WrapperFunctionBuffer()),
- "Unexpected Setup message");
-}
-
-TEST(SimpleRemoteCATest, HandleMessageAcceptsWellFormedHangup) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- auto R = CA.handleMessage(
- opc(TestSimpleRemoteCA::Opcode::Hangup), 0, ExecutorAddr(),
- TestSimpleRemoteCA::encodeHangupPayload(Error::success()));
- if (!R)
- ADD_FAILURE() << "unexpected error: " << toString(R.takeError());
- else
- EXPECT_EQ(*R, TestSimpleRemoteCA::Action::End);
-}
-
-TEST(SimpleRemoteCATest, HandleMessageReportsHangupReason) {
- // A hang-up carrying a reason ends the session with that reason, rather than
- // reporting a plain End: the reactor turns it into the disconnection error.
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- expectError(CA.handleMessage(
- opc(TestSimpleRemoteCA::Opcode::Hangup), 0, ExecutorAddr(),
- TestSimpleRemoteCA::encodeHangupPayload(
- make_error<StringError>("controller ran out of x"))),
- "controller ran out of x");
-}
-TEST(SimpleRemoteCATest, HandleMessageRejectsMalformedHangup) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
-
- // A hang-up must carry no sequence number and no tag.
- expectError(CA.handleMessage(
- opc(TestSimpleRemoteCA::Opcode::Hangup), /*SeqNo=*/5,
- ExecutorAddr(),
- TestSimpleRemoteCA::encodeHangupPayload(Error::success())),
- "Malformed hang-up message");
- expectError(CA.handleMessage(
- opc(TestSimpleRemoteCA::Opcode::Hangup), 0,
- ExecutorAddr(0x1000),
- TestSimpleRemoteCA::encodeHangupPayload(Error::success())),
- "Malformed hang-up message");
-
- // It must carry a deserializable reason. An empty payload is never valid --
- // the two ends are rev-locked, so this is a bug in the peer rather than skew.
- expectError(CA.handleMessage(opc(TestSimpleRemoteCA::Opcode::Hangup), 0,
- ExecutorAddr(), WrapperFunctionBuffer()),
- "Malformed hang-up message: could not deserialize reason");
+ S.detach([] {});
}
-TEST(SimpleRemoteCATest, HandleMessageRejectsResultWithTag) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- expectError(CA.handleMessage(opc(TestSimpleRemoteCA::Opcode::Result),
- /*SeqNo=*/1, ExecutorAddr(0x1000),
- WrapperFunctionBuffer()),
- "Result message should not carry a handler tag");
+TEST(SimpleRemoteCATest, OrderlyHangupRoundTrips) {
+ // Both ends encode and decode hang-ups through these, so a success value must
+ // survive the trip as a success.
+ auto Err = TestCA::decodeHangup(TestCA::encodeHangup(Error::success()));
+ EXPECT_FALSE(!!Err) << "an orderly hang-up is not an error";
}
-TEST(SimpleRemoteCATest, HandleMessageRejectsResultForUnknownSequenceNumber) {
- Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
- SentMessages Msgs;
- TestSimpleRemoteCA CA(S, Msgs);
- expectError(CA.handleMessage(opc(TestSimpleRemoteCA::Opcode::Result),
- /*SeqNo=*/7, ExecutorAddr(),
- WrapperFunctionBuffer::copyFrom("r", 1)),
- "No pending call for sequence number 7");
+TEST(SimpleRemoteCATest, HangupReasonRoundTrips) {
+ auto Err = TestCA::decodeHangup(
+ TestCA::encodeHangup(make_error<StringError>("controller ran out of x")));
+ ASSERT_TRUE(!!Err);
+ EXPECT_EQ(toString(std::move(Err)), "controller ran out of x");
}
-TEST(SimpleRemoteCATest, ResultCompletesPendingCall) {
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
- Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
- ASSERT_TRUE(CA);
-
- // Originate a controller call; the test double registers it as pending.
- std::optional<std::string> Res;
- S.callController(
- [&](WrapperFunctionBuffer R) {
- ASSERT_FALSE(R.getOutOfBandError()) << R.getOutOfBandError();
- Res = std::string(R.data(), R.size());
- },
- nullptr, WrapperFunctionBuffer());
-
- auto Calls = messages(Msgs, TestSimpleRemoteCA::Opcode::Call);
- ASSERT_EQ(Calls.size(), 1u);
- uint64_t SeqNo = Calls[0]->SeqNo;
- ASSERT_NE(SeqNo, 0u);
- ASSERT_FALSE(Res) << "handler fired before the result arrived";
-
- // Deliver the matching result; handleMessage completes the call inline.
- auto R = CA->handleMessage(opc(TestSimpleRemoteCA::Opcode::Result), SeqNo,
- ExecutorAddr(),
- WrapperFunctionBuffer::copyFrom("a", 1));
- if (!R)
- ADD_FAILURE() << "unexpected error: " << toString(R.takeError());
- else
- EXPECT_EQ(*R, TestSimpleRemoteCA::Action::Continue);
-
- ASSERT_TRUE(Res);
- EXPECT_EQ(*Res, "a");
+TEST(SimpleRemoteCATest, DecodeHangupRejectsAnEmptyPayload) {
+ // Never valid: the two ends are rev-locked, so this is a bug in the peer
+ // rather than version skew. It comes back as an error like any other reason,
+ // since both outcomes end the session.
+ auto Err = TestCA::decodeHangup(WrapperFunctionBuffer());
+ ASSERT_TRUE(!!Err);
+ EXPECT_EQ(toString(std::move(Err)),
+ "Malformed hang-up message: could not deserialize reason");
}
-TEST(SimpleRemoteCATest, CallDispatchesWrapperAndReturnsResult) {
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
+TEST(SimpleRemoteCATest, RegisterCallReturnsDistinctNonZeroSequenceNumbers) {
+ // Zero is reserved for messages with no result to await, so a registered call
+ // must never get it.
+ TestCA *CA = nullptr;
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
+ S.attach<TestCA>(BootstrapInfo(S), &CA);
ASSERT_TRUE(CA);
- // A Call names echoWrapper via the tag; SeqNo is the call id.
- auto R = CA->handleMessage(
- opc(TestSimpleRemoteCA::Opcode::Call), /*SeqNo=*/42,
- ExecutorAddr::fromPtr(reinterpret_cast<void *>(echoWrapper)),
- WrapperFunctionBuffer::copyFrom("world", 5));
- if (!R)
- ADD_FAILURE() << "unexpected error: " << toString(R.takeError());
- else
- EXPECT_EQ(*R, TestSimpleRemoteCA::Action::Continue);
-
- auto Results = messages(Msgs, TestSimpleRemoteCA::Opcode::Result);
- ASSERT_EQ(Results.size(), 1u);
- EXPECT_EQ(Results[0]->SeqNo, 42u);
- auto &Result = Results[0]->Payload;
- EXPECT_EQ(std::string(Result.data(), Result.size()), "world");
-}
+ uint64_t First = CA->registerCall(borrowHandler(S, *CA));
+ uint64_t Second = CA->registerCall(borrowHandler(S, *CA));
+ EXPECT_NE(First, 0u);
+ EXPECT_NE(Second, 0u);
+ EXPECT_NE(First, Second);
-TEST(SimpleRemoteCATest, DisconnectDrainsPendingCalls) {
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
- Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
- ASSERT_TRUE(CA);
-
- // A controller call is in flight (no result will arrive).
- std::optional<std::string> Err;
- S.callController(
- [&](WrapperFunctionBuffer R) {
- if (const char *M = R.getOutOfBandError())
- Err = M;
- },
- nullptr, WrapperFunctionBuffer());
- ASSERT_EQ(messages(Msgs, TestSimpleRemoteCA::Opcode::Call).size(), 1u);
- ASSERT_FALSE(Err) << "handler fired before disconnect";
-
- // Detach drives disconnect, which drains the pending call with a
- // "disconnected" error.
S.detach([] {});
-
- ASSERT_TRUE(Err);
- EXPECT_EQ(*Err, "disconnected");
}
-TEST(SimpleRemoteCATest, CallRacingTeardownFailsInline) {
- // Teardown has begun but the transport hasn't finished, so the call reaches
- // the CA and must be failed on the spot rather than left pending -- nothing
- // will drain it if the drain has already run.
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
+TEST(SimpleRemoteCATest, TakeCallYieldsTheHandlerExactlyOnce) {
+ TestCA *CA = nullptr;
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
+ S.attach<TestCA>(BootstrapInfo(S), &CA);
ASSERT_TRUE(CA);
- CA->FinishTeardownOnBegin = false;
- CA->disconnect();
- ASSERT_EQ(CA->TeardownsBegun, 1u);
+ uint64_t SeqNo = CA->registerCall(borrowHandler(S, *CA));
+ auto Taken = CA->takeCall(SeqNo);
+ EXPECT_TRUE(!!Taken) << "the registered handler should come back";
- std::optional<std::string> Err;
- S.callController(
- [&](WrapperFunctionBuffer R) {
- if (const char *M = R.getOutOfBandError())
- Err = M;
- },
- nullptr, WrapperFunctionBuffer());
+ // A second take finds nothing: a result for a call that was never made, or
+ // answered twice, which the transport reports as terminal.
+ EXPECT_FALSE(!!CA->takeCall(SeqNo));
+ EXPECT_FALSE(!!CA->takeCall(/*SeqNo=*/9999)) << "never registered";
- // The handler ran before callController returned, and nothing was sent.
- ASSERT_TRUE(Err);
- EXPECT_EQ(*Err, "disconnected");
- EXPECT_TRUE(messages(Msgs, TestSimpleRemoteCA::Opcode::Call).empty());
-
- CA->finishTeardown(Error::success());
+ CA->failPendingControllerCall(std::move(Taken));
+ S.detach([] {});
}
-TEST(SimpleRemoteCATest, DisconnectIsIdempotentAndDefersNotification) {
- unsigned Notifications = 0;
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
+TEST(SimpleRemoteCATest, TakeAllCallsEmptiesTheTable) {
+ TestCA *CA = nullptr;
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- S.setOnDisconnect([&](Error Err) {
- ++Notifications;
- cantFail(std::move(Err));
- });
-
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
+ S.attach<TestCA>(BootstrapInfo(S), &CA);
ASSERT_TRUE(CA);
- CA->FinishTeardownOnBegin = false;
-
- CA->disconnect();
- CA->disconnect();
- EXPECT_EQ(CA->TeardownsBegun, 1u) << "second disconnect was not a no-op";
- EXPECT_EQ(Notifications, 0u) << "notified before the transport finished";
-
- // finishTeardown may destroy *CA, so it is the last thing to touch it.
- CA->finishTeardown(Error::success());
- EXPECT_EQ(Notifications, 1u);
-}
-
-TEST(SimpleRemoteCATest, TransportInitiatedTeardownNotifiesWithoutHangup) {
- // A hang-up from the controller, or a transport failure, goes straight to
- // finishTeardown: the peer already knows, so nothing is sent.
- unsigned Notifications = 0;
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
- Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- S.setOnDisconnect([&](Error Err) {
- ++Notifications;
- EXPECT_EQ(toString(std::move(Err)), "controller vanished");
- });
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA);
- ASSERT_TRUE(CA);
+ uint64_t First = CA->registerCall(borrowHandler(S, *CA));
+ uint64_t Second = CA->registerCall(borrowHandler(S, *CA));
- // Nothing sends a hang-up on this path, because beginTeardown never runs.
- EXPECT_EQ(CA->TeardownsBegun, 0u);
- CA->finishTeardown(make_error<StringError>("controller vanished"));
+ auto All = CA->takeAllCalls();
+ EXPECT_EQ(All.size(), 2u);
+ EXPECT_EQ(All.count(First), 1u);
+ EXPECT_EQ(All.count(Second), 1u);
- EXPECT_EQ(Notifications, 1u);
-}
+ // Emptied, so a second drain finds nothing and a late result finds no call.
+ EXPECT_TRUE(CA->takeAllCalls().empty());
+ EXPECT_FALSE(!!CA->takeCall(First));
-TEST(SimpleRemoteCATest, DropOutBeforeAcceptingIsANoOp) {
- // A transport that enables delivery before calling beginAccepting can lose
- // the connection first, so beginAccepting has to tolerate running after
- // teardown has already completed: it accepts nothing and sends nothing.
- unsigned Notifications = 0;
- // Declared before the Session: its destructor drives teardown, which
- // records a hang-up.
- SentMessages Msgs;
- Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
- S.setOnDisconnect([&](Error Err) {
- ++Notifications;
- EXPECT_EQ(toString(std::move(Err)), "dropped during connect");
- });
-
- // Msgs outlives the CA, which is freed as attach returns.
- TestSimpleRemoteCA *CA = nullptr;
- S.attach<TestSimpleRemoteCA>(BootstrapInfo(S), Msgs, &CA,
- /*DropOutDuringConnect=*/true);
-
- EXPECT_EQ(Notifications, 1u);
- EXPECT_TRUE(Msgs.empty()) << "setup was sent after teardown completed";
+ for (auto &[SeqNo, OnComplete] : All)
+ CA->failPendingControllerCall(std::move(OnComplete));
+ S.detach([] {});
}
More information about the llvm-commits
mailing list