[orc-rt] Replace RunWrapperCall with generalized Dispatch (#210591)

RunWrapperCall dispatched wrapper-function calls, but the Session also
needs to dispatch continuations for results returned by the controller.
Replace it with a generalized Dispatch mechanism (DispatchFn) that is
handed an opaque Task for each unit of work, covering both.

QueueingRunner is correspondingly simplified: it no longer knows
anything about wrapper functions and just enqueues the Task it is given.
Tests and the noDispatch helper are updated to the new interface.

orc-rt/include/orc-rt/QueueingRunner.h

@@ -13,12 +13,8 @@
 #ifndef ORC_RT_QUEUEINGRUNNER_H
 #define ORC_RT_QUEUEINGRUNNER_H
 
-#include "orc-rt/WrapperFunction.h"
+#include "orc-rt/move_only_function.h"
 
-#include "orc-rt-c/CoreTypes.h"
-#include "orc-rt-c/WrapperFunction.h"
-
-#include <cstdint>
 #include <deque>
 #include <mutex>
 #include <optional>
@@ -59,9 +55,9 @@ private:
 
 } // namespace detail
 
-/// A wrapper-call runner that pushes incoming calls onto a caller-owned work
-/// queue, leaving the caller free to drain the queue however and whenever
-/// they choose.
+/// A task runner that pushes dispatched tasks onto a caller-owned work queue,
+/// leaving the caller free to drain the queue however and whenever they
+/// choose.
 ///
 /// QueueingRunner is intended for use on systems where threads are not
 /// available, and for unit tests. For most uses of the ORC runtime,
@@ -82,13 +78,9 @@ public:
 
   QueueingRunner(WorkQueueT &Pending) : Pending(Pending) {}
 
-  /// Enqueue a wrapper-function call to be run later.
-  void operator()(orc_rt_SessionRef S, uint64_t CallId,
-                  orc_rt_WrapperFunctionReturn Return,
-                  orc_rt_WrapperFunction Fn, WrapperFunctionBuffer ArgBytes) {
-    Pending.push_back([=, ArgBytes = std::move(ArgBytes)]() mutable {
-      Fn(S, CallId, Return, ArgBytes.release());
-    });
+  /// Enqueue a task to be run later.
+  void operator()(move_only_function<void()> Task) {
+    Pending.push_back(std::move(Task));
   }
 
   /// Run all currently-queued calls in last-in-first-out order, returning when

orc-rt/include/orc-rt/Session.h

@@ -79,17 +79,17 @@ public:
   using OnControllerCallReturnFn =
       move_only_function<void(WrapperFunctionBuffer)>;
 
-  /// Callback used by the Session to run incoming wrapper-function calls.
+  /// A unit of work handed to the Session's DispatchFn for execution.
+  using Task = move_only_function<void()>;
+
+  /// Callback used by the Session to dispatch tasks for execution.
   ///
-  /// A ManagedCodeTaskGroup token is created for each call to this callback,
-  /// and implementations must eventually call either Fn (typically as
-  /// Fn(S, CallId, Return, ArgBytes.release())), or call Return directly to
-  /// bail out of the call (typically with
-  /// WrapperFunctionBuffer::createOutOfBandError(...)). Failing to do either
-  /// will block Session shutdown indefinitely.
-  using RunWrapperCall = move_only_function<void(
-      orc_rt_SessionRef S, uint64_t CallId, orc_rt_WrapperFunctionReturn Return,
-      orc_rt_WrapperFunction Fn, WrapperFunctionBuffer ArgBytes)>;
+  /// The Session builds a Task for each unit of work it needs run -- an
+  /// incoming wrapper-function call, or a continuation for a result returned
+  /// by the controller -- and hands it to this callback, which is responsible
+  /// for arranging the task to be run inline, queued, or posted to a thread
+  /// pool.
+  using DispatchFn = move_only_function<void(Task)>;
 
   /// Tag used to identify executor-callable functions in the controller.
   /// See callController.
@@ -192,13 +192,14 @@ public:
   /// program are not generally visible to ORC-RT, but can optionally be
   /// reported by calling the orc_rt_Session_reportError function.)
   ///
-  /// The RunCall callback will be invoked for every incoming wrapper-function
-  /// call, and is responsible for arranging the call to be run (inline,
-  /// queued, or posted to a thread pool, at the caller's discretion).
+  /// The Dispatch callback is invoked to run tasks generated by the Session
+  /// (incoming wrapper-function calls, and continuations for results returned
+  /// by the controller), and is responsible for arranging each task to be run
+  /// inline, queued, or posted to a thread pool.
   ///
   /// Note that entry into the reporter is not synchronized: it may be
   /// called from multiple threads concurrently.
-  Session(ExecutorProcessInfo EPI, RunWrapperCall RunCall,
+  Session(ExecutorProcessInfo EPI, DispatchFn Dispatch,
           ErrorReporterFn ReportError);
 
   // Sessions are not copyable or moveable.
@@ -470,7 +471,9 @@ private:
       return;
     }
 
-    RunCall(wrap(this), CallId, &wrapperReturn, Fn, std::move(ArgBytes));
+    Dispatch([this, CallId, Fn, ArgBytes = std::move(ArgBytes)]() mutable {
+      Fn(wrap(this), CallId, &wrapperReturn, ArgBytes.release());
+    });
   }
 
   void sendWrapperResult(uint64_t CallId, WrapperFunctionBuffer ResultBytes);
@@ -478,7 +481,7 @@ private:
                             orc_rt_WrapperFunctionBuffer ResultBytes);
 
   ExecutorProcessInfo EPI;
-  RunWrapperCall RunCall;
+  DispatchFn Dispatch;
   std::shared_ptr<TaskGroup> ManagedCodeTaskGroup = TaskGroup::Create();
   std::shared_ptr<ControllerAccess> CA;
   ErrorReporterFn ReportError;

orc-rt/lib/executor/Session.cpp

@@ -49,9 +49,9 @@ private:
 
 Session::ControllerAccess::~ControllerAccess() = default;
 
-Session::Session(ExecutorProcessInfo EPI, RunWrapperCall RunCall,
+Session::Session(ExecutorProcessInfo EPI, DispatchFn Dispatch,
                  ErrorReporterFn ReportError)
-    : EPI(std::move(EPI)), RunCall(std::move(RunCall)),
+    : EPI(std::move(EPI)), Dispatch(std::move(Dispatch)),
       ReportError(std::move(ReportError)),
       Notifiers(createService<NotificationService>()) {}
 

orc-rt/test/unit/CommonTestUtils.h

@@ -46,19 +46,14 @@ inline orc_rt::ExecutorProcessInfo mockExecutorProcessInfo() noexcept {
   return orc_rt::ExecutorProcessInfo("arm64-apple-darwin", 16384);
 }
 
-/// RunWrapperCall callback for tests that should never dispatch a wrapper
-/// call. Records a test failure on invocation, then completes the call with an
-/// out-of-band error so that any caller awaiting the result unblocks and fails
-/// too (rather than hanging), even in -Asserts builds or when the dispatch
-/// arrives on a non-test thread.
-inline void noDispatch(orc_rt_SessionRef S, uint64_t CallId,
-                       orc_rt_WrapperFunctionReturn Return,
-                       orc_rt_WrapperFunction, orc_rt::WrapperFunctionBuffer) {
-  ADD_FAILURE() << "unexpected wrapper-call dispatch in a no-dispatch session";
-  Return(S, CallId,
-         orc_rt::WrapperFunctionBuffer::createOutOfBandError(
-             "unexpected wrapper-call dispatch in a no-dispatch session")
-             .release());
+/// DispatchFn for tests that should never dispatch a task. Records a test
+/// failure on invocation, then runs the task inline so that any caller
+/// 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) {
+  ADD_FAILURE() << "unexpected dispatch in a no-dispatch session";
+  Task();
 }
 
 template <size_t Idx = 0> class OpCounter {

orc-rt/test/unit/InProcessControllerAccessTest.cpp

@@ -287,7 +287,7 @@ static void echoWrapper(orc_rt_SessionRef S, uint64_t CallId,
 
 TEST(InProcessControllerAccessTest, CallFromControllerSuccess) {
   // The mock IPEPC initiates a wrapper call into IPCA. The Session's
-  // RunWrapperCall hook (a QueueingRunner over `Tasks`) enqueues the
+  // dispatch hook (a QueueingRunner over `Tasks`) enqueues the
   // invocation; draining the queue runs the wrapper, which echoes its
   // arguments back. Verify the mock receives the echoed bytes via
   // ReturnWrapperResult.

orc-rt/test/unit/QueueingRunnerTest.cpp

@@ -11,7 +11,6 @@
 #include "gtest/gtest.h"
 
 #include <cstdint>
-#include <deque>
 #include <thread>
 #include <vector>
 
@@ -19,78 +18,60 @@ using namespace orc_rt;
 
 namespace {
 
-// A dummy SessionRef value used purely to thread an opaque pointer through
-// the runner's enqueue path.
-inline orc_rt_SessionRef dummySession() noexcept {
-  return reinterpret_cast<orc_rt_SessionRef>(uintptr_t{0xABCD});
-}
-
-inline orc_rt_WrapperFunctionReturn dummyReturn() noexcept {
-  return [](orc_rt_SessionRef, uint64_t, orc_rt_WrapperFunctionBuffer) {};
-}
-
-// Test wrapper-function that records each invocation in a globally-accessible
-// log via its CallId.
-struct CallRecord {
-  orc_rt_SessionRef Session;
-  uint64_t CallId;
-};
-
-static std::vector<CallRecord> *RecordingLog = nullptr;
-
-static void recordingFn(orc_rt_SessionRef S, uint64_t CallId,
-                        orc_rt_WrapperFunctionReturn,
-                        orc_rt_WrapperFunctionBuffer ArgBytes) {
-  WrapperFunctionBuffer Owned(ArgBytes);
-  RecordingLog->push_back({S, CallId});
-}
+// Log of task ids, in the order the tasks ran. Tasks record their own id here
+// when run, letting tests observe both whether and in what order tasks ran.
+static std::vector<uint64_t> *RecordingLog = nullptr;
 
 class QueueingRunnerTest : public ::testing::Test {
 protected:
   void SetUp() override { RecordingLog = &Log; }
   void TearDown() override { RecordingLog = nullptr; }
 
-  std::vector<CallRecord> Log;
+  // Build a task that records the given id in the log when run.
+  static move_only_function<void()> recordingTask(uint64_t Id) {
+    return [Id]() { RecordingLog->push_back(Id); };
+  }
+
+  std::vector<uint64_t> Log;
   QueueingRunner<>::WorkQueue Q;
 };
 
 TEST_F(QueueingRunnerTest, EnqueueDoesNotRunImmediately) {
   QueueingRunner<> R(Q);
-  R(dummySession(), /*CallId=*/0, dummyReturn(), recordingFn,
-    WrapperFunctionBuffer());
-  EXPECT_EQ(Log.size(), 0u) << "Enqueue should not run the call";
-  // Pop initial call.
+  R(recordingTask(0));
+  EXPECT_EQ(Log.size(), 0u) << "Enqueue should not run the task";
+  // Pop initial task.
   EXPECT_TRUE(Q.pop_back())
-      << "At least one call should be sitting in the queue";
+      << "At least one task should be sitting in the queue";
   EXPECT_FALSE(Q.pop_back())
-      << "Exactly one call should have been sitting in the queue";
+      << "Exactly one task should have been sitting in the queue";
 }
 
 TEST_F(QueueingRunnerTest, RunFIFOUntilEmpty) {
   QueueingRunner R(Q);
   for (uint64_t I = 0; I < 3; ++I)
-    R(dummySession(), I, dummyReturn(), recordingFn, WrapperFunctionBuffer());
+    R(recordingTask(I));
 
   QueueingRunner<>::runFIFOUntilEmpty(Q);
 
   ASSERT_EQ(Log.size(), 3u);
-  EXPECT_EQ(Log[0].CallId, 0u);
-  EXPECT_EQ(Log[1].CallId, 1u);
-  EXPECT_EQ(Log[2].CallId, 2u);
+  EXPECT_EQ(Log[0], 0u);
+  EXPECT_EQ(Log[1], 1u);
+  EXPECT_EQ(Log[2], 2u);
   EXPECT_FALSE(Q.pop_back()); // Expect queue to be empty.
 }
 
 TEST_F(QueueingRunnerTest, RunLIFOUntilEmpty) {
   QueueingRunner R(Q);
   for (uint64_t I = 0; I < 3; ++I)
-    R(dummySession(), I, dummyReturn(), recordingFn, WrapperFunctionBuffer());
+    R(recordingTask(I));
 
   QueueingRunner<>::runLIFOUntilEmpty(Q);
 
   ASSERT_EQ(Log.size(), 3u);
-  EXPECT_EQ(Log[0].CallId, 2u);
-  EXPECT_EQ(Log[1].CallId, 1u);
-  EXPECT_EQ(Log[2].CallId, 0u);
+  EXPECT_EQ(Log[0], 2u);
+  EXPECT_EQ(Log[1], 1u);
+  EXPECT_EQ(Log[2], 0u);
   EXPECT_FALSE(Q.pop_back()); // Expect queue to be empty.
 }
 
@@ -102,58 +83,45 @@ TEST_F(QueueingRunnerTest, DrainOnEmptyQueueIsNoOp) {
   EXPECT_EQ(Log.size(), 0u);
 }
 
-TEST_F(QueueingRunnerTest, DrainPicksUpCallsEnqueuedDuringDrain) {
-  // A call enqueued by a running call should also be drained in the same
+TEST_F(QueueingRunnerTest, DrainPicksUpTasksEnqueuedDuringDrain) {
+  // A task enqueued by a running task should also be drained in the same
   // runFIFOUntilEmpty call.
   QueueingRunner R(Q);
 
-  // First call enqueues a second call from inside its body. We use a custom
-  // wrapper-function (not recordingFn) to do that, since recordingFn doesn't
-  // know about the queue.
-  static QueueingRunner<> *PendingR = nullptr;
-  PendingR = &R;
-  static auto reentrantFn = [](orc_rt_SessionRef S, uint64_t CallId,
-                               orc_rt_WrapperFunctionReturn,
-                               orc_rt_WrapperFunctionBuffer ArgBytes) {
-    WrapperFunctionBuffer Owned(ArgBytes);
-    RecordingLog->push_back({S, CallId});
-    if (CallId == 0)
-      (*PendingR)(S, /*CallId=*/1, dummyReturn(), recordingFn,
-                  WrapperFunctionBuffer());
-  };
-
-  R(dummySession(), /*CallId=*/0, dummyReturn(), reentrantFn,
-    WrapperFunctionBuffer());
+  // The first task enqueues a second task from inside its body.
+  R([&]() {
+    RecordingLog->push_back(0);
+    R(recordingTask(1));
+  });
 
   QueueingRunner<>::runFIFOUntilEmpty(Q);
 
   ASSERT_EQ(Log.size(), 2u);
-  EXPECT_EQ(Log[0].CallId, 0u);
-  EXPECT_EQ(Log[1].CallId, 1u);
-  PendingR = nullptr;
+  EXPECT_EQ(Log[0], 0u);
+  EXPECT_EQ(Log[1], 1u);
 }
 
 TEST_F(QueueingRunnerTest, ConcurrentProducerAndDrainer) {
   // Verify that QueueingRunner's default WorkQueue (SynchronizedDeque)
   // tolerates concurrent push from one thread and drain from another.
   //
-  // A producer thread enqueues NumCalls wrapper-function invocations while
-  // the main thread spins draining the queue. Once the producer has finished
-  // enqueueing, the main thread joins it and then performs a final drain to
-  // pick up any tail of calls enqueued after its last loop iteration.
-  constexpr uint64_t NumCalls = 1024;
+  // A producer thread enqueues NumTasks tasks while the main thread spins
+  // draining the queue. Once the producer has finished enqueueing, the main
+  // thread joins it and then performs a final drain to pick up any tail of
+  // tasks enqueued after its last loop iteration.
+  constexpr uint64_t NumTasks = 1024;
 
   QueueingRunner<> R(Q);
 
   std::thread Producer([&]() {
-    for (uint64_t I = 0; I < NumCalls; ++I)
-      R(dummySession(), I, dummyReturn(), recordingFn, WrapperFunctionBuffer());
+    for (uint64_t I = 0; I < NumTasks; ++I)
+      R(recordingTask(I));
   });
 
   // Drain concurrently with the producer. The drainer doesn't know when the
   // producer is done, so we just spin until the producer thread has joined
   // (after which a final drain will be definitive).
-  while (Log.size() < NumCalls) {
+  while (Log.size() < NumTasks) {
     QueueingRunner<>::runFIFOUntilEmpty(Q);
     std::this_thread::yield();
   }
@@ -161,12 +129,12 @@ TEST_F(QueueingRunnerTest, ConcurrentProducerAndDrainer) {
   Producer.join();
   QueueingRunner<>::runFIFOUntilEmpty(Q); // pick up any tail.
 
-  ASSERT_EQ(Log.size(), NumCalls);
-  // Producer enqueues in order 0..NumCalls; FIFO drain must observe the same
+  ASSERT_EQ(Log.size(), NumTasks);
+  // Producer enqueues in order 0..NumTasks; FIFO drain must observe the same
   // order. (Concurrent draining doesn't reorder per-producer enqueues for a
   // single producer.)
-  for (uint64_t I = 0; I < NumCalls; ++I)
-    EXPECT_EQ(Log[I].CallId, I);
+  for (uint64_t I = 0; I < NumTasks; ++I)
+    EXPECT_EQ(Log[I], I);
 
   EXPECT_FALSE(Q.pop_back()) << "Queue should be empty after final drain";
 }