[orc-rt] Run controller-call continuations from Session (#211166)

A controller call's on-complete handler may run managed code, so it must
be dispatched by the Session under a ManagedCodeTaskGroup token, not
invoked directly by the ControllerAccess (which previously held it as a
plain callable and called it itself -- untokened and undispatched).

Wrap it in an opaque ControllerAccess::OnControllerCallReturn that
implementations can hold but not call. They complete it via exactly one
of three Session-provided paths:

- handleControllerCallResult -- the controller returned a result;
dispatches the handler under a fresh token. Asserts the managed-code
group is still open (i.e. before notifyDisconnected).

- failPendingControllerCall -- an already-enqueued call is failed
because the connection dropped before a result arrived (the disconnect
drain); dispatches the handler under a token with a disconnect error.

- failControllerCallInline -- a call made from within callController
while already disconnecting, which can never be enqueued; runs the
handler inline with a disconnect error, where the caller (and its token)
is still on the stack.

Together these guarantee a controller call's handler always runs exactly
once, with a result or a disconnect error, and never runs managed code
without a token. Only the success path takes a result value; the two
failure paths produce the canonical disconnect error internally, so it
can't be gotten wrong. InProcessControllerAccess and the Session unit
tests are updated.
This commit is contained in:
Lang Hames 2026-07-22 14:13:10 +10:00 committed by GitHub
parent d4e853f370
commit 5602b98193
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 311 additions and 93 deletions

View file

@ -107,13 +107,13 @@ public:
void disconnect() override;
void callController(OnControllerCallReturnFn OnComplete, HandlerTag T,
void callController(OnControllerCallReturn OnComplete, HandlerTag T,
WrapperFunctionBuffer ArgBytes) override;
void sendWrapperResult(uint64_t CallId,
WrapperFunctionBuffer ResultBytes) override;
private:
uint64_t registerPendingHandler(OnControllerCallReturnFn OnComplete);
uint64_t registerPendingHandler(OnControllerCallReturn OnComplete);
void doDisconnect();
void callWrapper(uint64_t CallId, void *Fn,
@ -133,8 +133,7 @@ private:
std::mutex M;
uint64_t NextPendingCall = 0;
using PendingCallsMap =
std::unordered_map<uint64_t, OnControllerCallReturnFn>;
using PendingCallsMap = std::unordered_map<uint64_t, OnControllerCallReturn>;
PendingCallsMap PendingCalls;
};

View file

@ -104,7 +104,44 @@ public:
protected:
using HandlerTag = Session::HandlerTag;
using OnControllerCallReturnFn = Session::OnControllerCallReturnFn;
/// Opaque wrapper for a controller-call result handler.
///
/// ControllerAccess implementations hold these for pending calls but cannot
/// invoke them directly. Each must be completed exactly once, via one of
/// three Session-provided paths, so the handler runs in a context where it
/// may safely touch managed code:
///
/// - handleControllerCallResult(OnComplete, ResultBytes): the controller
/// returned a result. Dispatches the handler under a fresh managed-code
/// token.
///
/// - failPendingControllerCall(OnComplete): a call that was enqueued
/// while connected is failed because the connection dropped before a
/// result arrived (the disconnect drain). Dispatches the handler under
/// a fresh token with a disconnect error.
///
/// - failControllerCallInline(OnComplete): a call made from within
/// callController while already disconnecting, which can therefore
/// never be enqueued. Runs the handler inline with a disconnect error,
/// on the caller's stack, covered by the caller's token (see
/// callController).
///
/// Together these guarantee the handler always runs exactly once -- with a
/// result or a disconnect error.
class OnControllerCallReturn {
friend class Session;
public:
OnControllerCallReturn() = default;
explicit operator bool() const noexcept { return !!Wrapped; }
private:
OnControllerCallReturn(OnControllerCallReturnFn Wrapped)
: Wrapped(std::move(Wrapped)) {}
OnControllerCallReturnFn Wrapped;
};
ControllerAccess(Session &S) : S(S) {}
@ -146,14 +183,42 @@ public:
/// to be a cheap operation (e.g. signaling a shutdown flag) with the
/// actual disconnection and notifyDisconnected call happening on another
/// thread.
///
/// When disconnecting, a ControllerAccess must fail every still-pending
/// controller call via failPendingControllerCall(OnComplete). This drain
/// must complete before notifyDisconnected, while the managed-code group is
/// still open, so the completions are dispatched rather than dropped
/// (failPendingControllerCall dispatches under a token and asserts the
/// group is open). The drain must be serialized with the disconnecting
/// check in callController, so a call racing disconnection is either
/// drained here or failed inline there -- never left pending.
virtual void disconnect() = 0;
/// Report an error to the session.
void reportError(Error Err) { S.reportError(std::move(Err)); }
/// Call the handler in the controller associated with the given tag.
virtual void callController(OnControllerCallReturnFn OnComplete,
HandlerTag T,
///
/// OnComplete must be completed exactly once (see OnControllerCallReturn).
/// On entry, check whether this ControllerAccess has begun disconnecting --
/// its connection state is closing or closed, whether triggered locally by
/// disconnect() or by the remote:
///
/// - Still connected: enqueue the continuation. It is later completed via
/// handleControllerCallResult when the controller returns a result, or
/// via failPendingControllerCall if disconnection drains it first (see
/// disconnect).
///
/// - Disconnecting: the call can never receive a result, so fail it
/// immediately via failControllerCallInline, on the current thread.
///
/// Because the disconnecting case completes the handler inline, the handler
/// may run before callController returns, on the calling thread; callers
/// must tolerate this. It is safe: the caller is still on the stack, so a
/// managed-code caller's token still covers the handler, and a non-managed
/// caller needs none (a handler that re-enters managed code must acquire
/// its own token and handle denial).
virtual void callController(OnControllerCallReturn OnComplete, HandlerTag T,
WrapperFunctionBuffer ArgBytes) = 0;
/// Send the result of the given wrapper function call to the controller.
@ -170,8 +235,8 @@ public:
/// exactly-once semantics for this method, even when disconnect is called
/// concurrently with controller-initiated disconnection.
///
/// No calls should be made to reportError or handleWrapperCall after this
/// method is called.
/// No calls should be made to reportError, handleWrapperCall, or
/// handleControllerCallResult after this method is called.
void notifyDisconnected() { S.handleDisconnect(); }
/// Ask the Session to run the given wrapper function.
@ -182,6 +247,43 @@ public:
S.handleWrapperCall(CallId, Fn, std::move(ArgBytes));
}
/// Complete a controller call with a result the controller returned, by
/// dispatching its handler under a fresh managed-code token. Must be called
/// while the managed-code group is still open -- i.e. before
/// notifyDisconnected; the Session asserts this.
///
/// To fail a pending call on disconnect, use failPendingControllerCall.
void handleControllerCallResult(OnControllerCallReturn OnComplete,
WrapperFunctionBuffer ResultBytes) {
S.handleControllerCallResult(std::move(OnComplete),
std::move(ResultBytes));
}
/// Fail a pending (already-enqueued) controller call because the connection
/// dropped before it received a result -- the disconnect drain.
///
/// Equivalent to completing it with a disconnect error via
/// handleControllerCallResult (dispatched under a token, same open-group
/// requirement), but named for the drain and taking no result, so the
/// failure value can't be gotten wrong.
void failPendingControllerCall(OnControllerCallReturn OnComplete) {
S.failPendingControllerCall(std::move(OnComplete));
}
/// Fail a controller call by running its handler inline, on the current
/// thread, with a disconnect error -- without acquiring a managed-code
/// token.
///
/// Use ONLY from within callController, to fail a call that arrives while
/// already disconnecting (see callController). The original caller is then
/// still on the stack, so the handler is covered by the caller's token, or
/// needs none. A call that was successfully enqueued is instead failed via
/// failPendingControllerCall; a real result uses
/// handleControllerCallResult.
void failControllerCallInline(OnControllerCallReturn OnComplete) {
S.failControllerCallInline(std::move(OnComplete));
}
private:
Session &S;
};
@ -478,6 +580,48 @@ private:
});
}
void handleControllerCallResult(
ControllerAccess::OnControllerCallReturn OnComplete,
WrapperFunctionBuffer ResultBytes) {
TaskGroup::Token T(ManagedCodeTaskGroup);
if (!T) {
// Contract violation: a deferred completion must precede
// notifyDisconnected (while the group is still open), and a synchronous
// failure must use failControllerCallInline instead. Reaching here means
// one of those was broken; falling through would run the handler
// unbracketed into a possibly-torn-down Session. Fail loudly.
assert(false && "handleControllerCallResult on a closed "
"ManagedCodeTaskGroup");
abort();
}
Dispatch(
[OnComplete = std::move(OnComplete.Wrapped),
ResultBytes = std::move(ResultBytes),
T = std::move(T)]() mutable { OnComplete(std::move(ResultBytes)); });
}
/// The canonical out-of-band error delivered when a controller call is
/// failed because of disconnection. Used by failPendingControllerCall and
/// failControllerCallInline; not for direct use.
static WrapperFunctionBuffer disconnectError() {
return WrapperFunctionBuffer::createOutOfBandError("disconnected");
}
void failPendingControllerCall(
ControllerAccess::OnControllerCallReturn OnComplete) {
handleControllerCallResult(std::move(OnComplete), disconnectError());
}
void failControllerCallInline(
ControllerAccess::OnControllerCallReturn OnComplete) {
// Runs inline, on the caller's stack, without a token: valid only for a
// synchronous failure from within callController, where the caller (and its
// token, if any) is still on the stack. See callController.
OnComplete.Wrapped(disconnectError());
}
void sendWrapperResult(uint64_t CallId, WrapperFunctionBuffer ResultBytes);
static void wrapperReturn(orc_rt_SessionRef S, uint64_t CallId,
orc_rt_WrapperFunctionBuffer ResultBytes);

View file

@ -201,7 +201,7 @@ void InProcessControllerAccess::disconnect() {
}
void InProcessControllerAccess::callController(
OnControllerCallReturnFn OnComplete, HandlerTag T,
OnControllerCallReturn OnComplete, HandlerTag T,
WrapperFunctionBuffer ArgBytes) {
assert(C && "callController called before connect");
if (C->EnterMessageScope(C)) {
@ -209,8 +209,9 @@ void InProcessControllerAccess::callController(
T, ArgBytes.release());
C->LeaveMessageScope(C);
} else
OnComplete(
WrapperFunctionBuffer::createOutOfBandError("connection closed"));
// Synchronous failure: the connection is already gone, so the call can't be
// sent. The caller is still on the stack, so fail the handler inline.
failControllerCallInline(std::move(OnComplete));
}
void InProcessControllerAccess::sendWrapperResult(
@ -223,7 +224,7 @@ void InProcessControllerAccess::sendWrapperResult(
}
uint64_t InProcessControllerAccess::registerPendingHandler(
OnControllerCallReturnFn OnComplete) {
OnControllerCallReturn OnComplete) {
std::scoped_lock<std::mutex> Lock(M);
PendingCalls[NextPendingCall] = std::move(OnComplete);
return NextPendingCall++;
@ -237,7 +238,7 @@ void InProcessControllerAccess::doDisconnect() {
ToDrain = std::move(PendingCalls);
}
for (auto &[_, H] : ToDrain)
H(WrapperFunctionBuffer::createOutOfBandError("disconnected"));
failPendingControllerCall(std::move(H));
notifyDisconnected();
}
@ -259,7 +260,7 @@ void InProcessControllerAccess::callWrapperEntry(
void InProcessControllerAccess::returnJITDispatchResult(
uint64_t CallId, orc_rt_WrapperFunctionBuffer ResultBytes) {
OnControllerCallReturnFn OnComplete;
OnControllerCallReturn OnComplete;
{
std::scoped_lock<std::mutex> Lock(M);
auto I = PendingCalls.find(CallId);
@ -272,7 +273,8 @@ void InProcessControllerAccess::returnJITDispatchResult(
if (!OnComplete)
return reportError(make_error<StringError>("Invalid call id"));
OnComplete(WrapperFunctionBuffer(ResultBytes));
handleControllerCallResult(std::move(OnComplete),
WrapperFunctionBuffer(ResultBytes));
}
void InProcessControllerAccess::returnJITDispatchResultEntry(

View file

@ -11,6 +11,7 @@
#include "orc-rt/Error.h"
#include "orc-rt/ExecutorProcessInfo.h"
#include "orc-rt/Session.h"
#include "orc-rt/WrapperFunction.h"
#include "orc-rt/move_only_function.h"
@ -51,11 +52,14 @@ inline orc_rt::ExecutorProcessInfo mockExecutorProcessInfo() noexcept {
/// awaiting a result unblocks (rather than hanging) and the managed-code token
/// is released, even in -Asserts builds or when the dispatch arrives on a
/// non-test thread.
inline void noDispatch(orc_rt::move_only_function<void()> Task) {
inline void noDispatch(orc_rt::Session::Task T) {
ADD_FAILURE() << "unexpected dispatch in a no-dispatch session";
Task();
T();
}
/// DispatchFn that runs tasks on the current thread.
inline void inlineDispatch(orc_rt::Session::Task T) { T(); }
template <size_t Idx = 0> class OpCounter {
public:
OpCounter() { ++DefaultConstructions; }

View file

@ -191,7 +191,7 @@ TEST(InProcessControllerAccessTest, OnConnectFailureIsReportedAndDetaches) {
TEST(InProcessControllerAccessTest, CallControllerSuccess) {
// A callController call routed through MockIPEPC, which echoes the args
// back as the result. Verify OnComplete fires with the payload.
Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
std::unique_ptr<MockIPEPC> Mock;
attachWithMock(S, Mock);
@ -219,7 +219,7 @@ TEST(InProcessControllerAccessTest, CallControllerSuccess) {
TEST(InProcessControllerAccessTest, CallControllerOutOfBandError) {
// A callController call where the mock responds with an out-of-band error.
// OnComplete should observe the error message intact.
Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
std::unique_ptr<MockIPEPC> Mock;
attachWithMock(S, Mock);
@ -248,7 +248,7 @@ TEST(InProcessControllerAccessTest, DisconnectDrainsPendingCalls) {
// A callController call is in-flight when the connection drops (the mock
// never responds). Verify that doDisconnect drains the pending handler with
// a "disconnected" out-of-band error rather than leaving it stranded.
Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
Session S(mockExecutorProcessInfo(), inlineDispatch, noErrors);
std::unique_ptr<MockIPEPC> Mock;
attachWithMock(S, Mock);

View file

@ -80,6 +80,42 @@ public:
};
class MockControllerAccess : public Session::ControllerAccess {
private:
class ConnectGuard {
public:
ConnectGuard() = default;
ConnectGuard(MockControllerAccess *MCA) : MCA(MCA) {
// Note: Assumes already locked.
++MCA->Outstanding;
}
ConnectGuard(const ConnectGuard &) = delete;
ConnectGuard &operator=(const ConnectGuard &) = delete;
ConnectGuard(ConnectGuard &&Other) : MCA(Other.MCA) { Other.MCA = nullptr; }
ConnectGuard &operator=(ConnectGuard &&Other) {
reset();
MCA = Other.MCA;
Other.MCA = nullptr;
return *this;
};
~ConnectGuard() { reset(); }
private:
void reset() {
if (MCA) {
bool Notify;
{
std::scoped_lock Lock(MCA->M);
--MCA->Outstanding;
Notify = MCA->Shutdown && MCA->Outstanding == 0;
}
if (Notify)
MCA->ShutdownCV.notify_all();
}
}
MockControllerAccess *MCA = nullptr;
};
public:
using OnConnectFn = move_only_function<Error(BootstrapInfo &BI)>;
@ -124,121 +160,113 @@ public:
}
void disconnect() override {
std::unique_lock<std::mutex> Lock(M);
Shutdown = true;
ShutdownCV.wait(Lock, [this]() { return Shutdown && Outstanding == 0; });
// Drain any still-pending controller calls before notifying, so their
// handlers are failed rather than silently dropped -- see the
// ControllerAccess::disconnect contract. Shutdown gates callController
// (which bails via failControllerCallInline once it is set), so no new call
// can be registered after the snapshot below.
std::unordered_map<size_t, OnControllerCallReturn> ToDrain;
{
std::unique_lock<std::mutex> Lock(M);
Shutdown = true;
ShutdownCV.wait(Lock, [this]() { return Shutdown && Outstanding == 0; });
ToDrain = std::move(PendingOut);
}
for (auto &[_, OnComplete] : ToDrain)
failPendingControllerCall(std::move(OnComplete));
notifyDisconnected();
}
void callController(OnControllerCallReturnFn OnComplete, HandlerTag T,
void callController(OnControllerCallReturn OnComplete, HandlerTag T,
WrapperFunctionBuffer ArgBytes) override {
// Simulate a call to the controller by running the requested function via
// the test-supplied Post hook (or inline, if no hook was provided).
size_t CId;
ConnectGuard CG;
size_t CId = 0;
bool BailOut = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (Shutdown)
return;
CId = CallId++;
Pending[CId] = std::move(OnComplete);
++Outstanding;
if (!Shutdown) {
CG = ConnectGuard(this);
CId = CallId++;
assert(!PendingOut.count(CId));
PendingOut[CId] = std::move(OnComplete);
} else
BailOut = true;
}
if (BailOut)
return failControllerCallInline(std::move(OnComplete));
runOrPost([this, CId, T, ArgBytes = std::move(ArgBytes)]() mutable {
auto Fn = reinterpret_cast<orc_rt_WrapperFunction>(T);
Fn(reinterpret_cast<orc_rt_SessionRef>(this), CId, wfReturn,
Fn(reinterpret_cast<orc_rt_SessionRef>(this), CId, ccReturn,
ArgBytes.release());
});
bool Notify = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (--Outstanding == 0 && Shutdown)
Notify = true;
}
if (Notify)
ShutdownCV.notify_all();
}
void sendWrapperResult(uint64_t CallId,
WrapperFunctionBuffer ResultBytes) override {
// Respond to a simulated call by the controller.
OnControllerCallReturnFn OnComplete;
ConnectGuard CG;
Session::OnControllerCallReturnFn OnComplete;
{
std::scoped_lock<std::mutex> Lock(M);
if (Shutdown) {
assert(Pending.empty() && "Shut down but results still pending?");
assert(PendingIn.empty() && "Shut down but results still pending?");
return;
}
auto I = Pending.find(CallId);
assert(I != Pending.end());
CG = ConnectGuard(this);
auto I = PendingIn.find(CallId);
assert(I != PendingIn.end());
OnComplete = std::move(I->second);
Pending.erase(I);
++Outstanding;
PendingIn.erase(I);
}
runOrPost([OnComplete = std::move(OnComplete),
ResultBytes = std::move(ResultBytes)]() mutable {
OnComplete(std::move(ResultBytes));
});
bool Notify = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (--Outstanding == 0 && Shutdown)
Notify = true;
}
if (Notify)
ShutdownCV.notify_all();
}
void callFromController(OnControllerCallReturnFn OnComplete,
void returnFromController(uint64_t CallId,
WrapperFunctionBuffer ResultBytes) {
ConnectGuard CG;
OnControllerCallReturn OnComplete;
{
std::scoped_lock<std::mutex> Lock(M);
CG = ConnectGuard(this);
auto I = PendingOut.find(CallId);
assert(I != PendingOut.end());
OnComplete = std::move(I->second);
PendingOut.erase(I);
}
handleControllerCallResult(std::move(OnComplete), std::move(ResultBytes));
}
void callFromController(Session::OnControllerCallReturnFn OnComplete,
orc_rt_WrapperFunction Fn,
WrapperFunctionBuffer ArgBytes) {
ConnectGuard CG;
size_t CId = 0;
bool BailOut = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (!Shutdown) {
CG = ConnectGuard(this);
CId = CallId++;
Pending[CId] = std::move(OnComplete);
++Outstanding;
PendingIn[CId] = std::move(OnComplete);
} else
BailOut = true;
}
if (BailOut)
return OnComplete(WrapperFunctionBuffer::createOutOfBandError(
"Controller disconnected"));
return runOrPost([OnComplete = std::move(OnComplete)]() mutable {
OnComplete(WrapperFunctionBuffer::createOutOfBandError(
"Controller disconnected"));
});
handleWrapperCall(CId, Fn, std::move(ArgBytes));
bool Notify = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (--Outstanding == 0 && Shutdown)
Notify = true;
}
if (Notify)
ShutdownCV.notify_all();
}
/// Simulate start of outstanding operation.
void incOutstanding() {
std::scoped_lock<std::mutex> Lock(M);
++Outstanding;
}
/// Simulate end of outstanding operation.
void decOutstanding() {
bool Notify = false;
{
std::scoped_lock<std::mutex> Lock(M);
if (--Outstanding == 0 && Shutdown)
Notify = true;
}
if (Notify)
ShutdownCV.notify_all();
}
private:
@ -249,11 +277,11 @@ private:
Work();
}
static void wfReturn(orc_rt_SessionRef S, uint64_t CallId,
static void ccReturn(orc_rt_SessionRef S, uint64_t CallId,
orc_rt_WrapperFunctionBuffer ResultBytes) {
// Abuse "session" to refer to the ControllerAccess object.
// We can just re-use sendFunctionResult for this.
reinterpret_cast<MockControllerAccess *>(S)->sendWrapperResult(
reinterpret_cast<MockControllerAccess *>(S)->returnFromController(
CallId, WrapperFunctionBuffer(ResultBytes));
}
@ -261,9 +289,10 @@ private:
std::mutex M;
bool Shutdown = false;
size_t Outstanding = 0;
size_t CallId = 0;
std::unordered_map<size_t, OnControllerCallReturnFn> Pending;
size_t Outstanding = 0;
std::unordered_map<size_t, OnControllerCallReturn> PendingOut;
std::unordered_map<size_t, Session::OnControllerCallReturnFn> PendingIn;
std::condition_variable ShutdownCV;
OnConnectFn OnConnect;
};
@ -593,6 +622,46 @@ TEST(ControllerAccessTest, Basics) {
QueueingRunner<>::runFIFOUntilEmpty(Tasks);
}
// A ControllerAccess that is attached but whose controller connection is
// already gone, so every callController fails synchronously. Used to exercise
// the failControllerCallInline path in isolation.
class DeadControllerAccess : public Session::ControllerAccess {
public:
DeadControllerAccess(Session &S) : ControllerAccess(S) {}
void connect(BootstrapInfo) override {}
void disconnect() override { notifyDisconnected(); }
void callController(OnControllerCallReturn OnComplete, HandlerTag,
WrapperFunctionBuffer) override {
failControllerCallInline(std::move(OnComplete));
}
void sendWrapperResult(uint64_t, WrapperFunctionBuffer) override {}
};
TEST(ControllerAccessTest, SynchronousCallControllerFailureRunsInline) {
// A controller call that fails synchronously (connection already gone) must
// complete its handler inline -- on the calling thread, before callController
// returns, and WITHOUT going through Dispatch -- with the canonical
// "disconnected" out-of-band error. noDispatch ADD_FAILUREs if anything is
// dispatched, so it doubles as the "ran inline, not via the task queue"
// check.
Session S(mockExecutorProcessInfo(), noDispatch, noErrors);
S.attach<DeadControllerAccess>(BootstrapInfo(S));
bool HandlerRan = false;
std::string ErrMsg;
S.callController(
[&](WrapperFunctionBuffer Result) {
HandlerRan = true;
if (const char *E = Result.getOutOfBandError())
ErrMsg = E;
},
/*T=*/nullptr, WrapperFunctionBuffer());
// Set synchronously, during the call above -> the handler ran inline.
EXPECT_TRUE(HandlerRan);
EXPECT_EQ(ErrMsg, "disconnected");
}
static void add_sps_wrapper(orc_rt_SessionRef S, uint64_t CallId,
orc_rt_WrapperFunctionReturn Return,
orc_rt_WrapperFunctionBuffer ArgBytes) {