Merge WorkQueue into AppRuntime

Core/AppRuntime/CMakeLists.txt

@@ -9,9 +9,7 @@ set(SOURCES
     "Include/Babylon/AppRuntime.h"
     "Source/AppRuntime.cpp"
     "Source/AppRuntime_${NAPI_JAVASCRIPT_ENGINE}.cpp"
-    "Source/AppRuntime_${JSRUNTIMEHOST_PLATFORM}.${IMPL_EXT}"
-    "Source/WorkQueue.cpp"
-    "Source/WorkQueue.h")
+    "Source/AppRuntime_${JSRUNTIMEHOST_PLATFORM}.${IMPL_EXT}")
 
 add_library(AppRuntime ${SOURCES})
 warnings_as_errors(AppRuntime)

Core/AppRuntime/Include/Babylon/AppRuntime.h

@@ -6,14 +6,19 @@
 
 #include <napi/utilities.h>
 
+#include <arcana/threading/cancellation.h>
+#include <arcana/threading/dispatcher.h>
+
 #include <memory>
 #include <functional>
 #include <exception>
+#include <optional>
+#include <mutex>
+#include <thread>
+#include <type_traits>
 
 namespace Babylon
 {
-    class WorkQueue;
-
     class AppRuntime final
     {
     public:
@@ -43,6 +48,23 @@ namespace Babylon
         static void BABYLON_API DefaultUnhandledExceptionHandler(const Napi::Error& error);
 
     private:
+        template<typename CallableT>
+        void Append(CallableT callable)
+        {
+            if constexpr (std::is_copy_constructible<CallableT>::value)
+            {
+                m_dispatcher.queue([this, callable = std::move(callable)]() {
+                    callable(m_env.value());
+                });
+            }
+            else
+            {
+                m_dispatcher.queue([this, callablePtr = std::make_shared<CallableT>(std::move(callable))]() {
+                    (*callablePtr)(m_env.value());
+                });
+            }
+        }
+
         // These three methods are the mechanism by which platform- and JavaScript-specific
         // code can be "injected" into the execution of the JavaScript thread. These three
         // functions are implemented in separate files, thus allowing implementations to be
@@ -62,6 +84,10 @@ namespace Babylon
         void Execute(Dispatchable<void()> callback);
 
         Options m_options;
-        std::unique_ptr<WorkQueue> m_workQueue;
+        std::optional<Napi::Env> m_env{};
+        std::optional<std::scoped_lock<std::mutex>> m_suspensionLock{};
+        arcana::cancellation_source m_cancelSource{};
+        arcana::manual_dispatcher<128> m_dispatcher{};
+        std::thread m_thread;
     };
 }

Core/AppRuntime/Source/AppRuntime.cpp

@@ -1,5 +1,4 @@
 #include "AppRuntime.h"
-#include "WorkQueue.h"
 #include <cassert>
 
 namespace Babylon
@@ -11,7 +10,7 @@ namespace Babylon
 
     AppRuntime::AppRuntime(Options options)
         : m_options{std::move(options)}
-        , m_workQueue{std::make_unique<WorkQueue>([this] { RunPlatformTier(); })}
+        , m_thread{[this] { RunPlatformTier(); }}
     {
         Dispatch([this](Napi::Env env) {
             JsRuntime::CreateForJavaScript(env, [this](auto func) { Dispatch(std::move(func)); });
@@ -20,26 +19,58 @@ namespace Babylon
 
     AppRuntime::~AppRuntime()
     {
+        if (m_suspensionLock.has_value())
+        {
+            m_suspensionLock.reset();
+        }
+
+        // Cancel immediately so pending work is dropped promptly, then append
+        // a no-op work item to wake the worker thread from blocking_tick. The
+        // no-op goes through push() which acquires the queue mutex, avoiding
+        // the race where a bare notify_all() can be missed by wait().
+        //
+        // NOTE: This preserves the existing shutdown behavior where pending
+        // callbacks are dropped on cancellation. A more complete solution
+        // would add cooperative shutdown (e.g. NotifyDisposing/Rundown) so
+        // consumers can finish cleanup work before the runtime is destroyed.
+        m_cancelSource.cancel();
+        Append([](Napi::Env) {});
+
+        m_thread.join();
     }
 
     void AppRuntime::Run(Napi::Env env)
     {
-        m_workQueue->Run(env);
+        m_env = std::make_optional(env);
+
+        m_dispatcher.set_affinity(std::this_thread::get_id());
+
+        while (!m_cancelSource.cancelled())
+        {
+            m_dispatcher.blocking_tick(m_cancelSource);
+        }
+
+        // The dispatcher can be non-empty if something is dispatched after cancellation.
+        m_dispatcher.clear();
     }
 
     void AppRuntime::Suspend()
     {
-        m_workQueue->Suspend();
+        auto suspensionMutex = std::make_shared<std::mutex>();
+        m_suspensionLock.emplace(*suspensionMutex);
+        Append([suspensionMutex{std::move(suspensionMutex)}](Napi::Env) {
+            std::scoped_lock lock{*suspensionMutex};
+        });
     }
 
     void AppRuntime::Resume()
     {
-        m_workQueue->Resume();
+        m_suspensionLock.reset();
     }
 
     void AppRuntime::Dispatch(Dispatchable<void(Napi::Env)> func)
     {
-        m_workQueue->Append([this, func{std::move(func)}](Napi::Env env) mutable {
+        Append([this, func{std::move(func)}](Napi::Env env) mutable {
             Execute([this, env, func{std::move(func)}]() mutable {
                 try
                 {

Core/AppRuntime/Source/AppRuntime_JSI.cpp

@@ -1,5 +1,4 @@
 #include "AppRuntime.h"
-#include "WorkQueue.h"
 
 #include <napi/env.h>
 #include <V8JsiRuntime.h>
@@ -10,15 +9,15 @@ namespace
     class TaskRunnerAdapter : public v8runtime::JSITaskRunner
     {
     public:
-        TaskRunnerAdapter(Babylon::WorkQueue& workQueue)
-            : m_workQueue(workQueue)
+        TaskRunnerAdapter(Babylon::AppRuntime& runtime)
+            : m_runtime(runtime)
         {
         }
 
         void postTask(std::unique_ptr<v8runtime::JSITask> task) override
         {
-            std::shared_ptr<v8runtime::JSITask> shared_task(task.release());
-            m_workQueue.Append([shared_task2 = std::move(shared_task)](Napi::Env) {
+            std::shared_ptr<v8runtime::JSITask> shared_task(std::move(task));
+            m_runtime.Dispatch([shared_task2 = std::move(shared_task)](Napi::Env) {
                 shared_task2->run();
             });
         }
@@ -27,7 +26,7 @@ namespace
         TaskRunnerAdapter(const TaskRunnerAdapter&) = delete;
         TaskRunnerAdapter& operator=(const TaskRunnerAdapter&) = delete;
 
-        Babylon::WorkQueue& m_workQueue;
+        Babylon::AppRuntime& m_runtime;
     };
 }
 
@@ -37,7 +36,7 @@ namespace Babylon
     {
         v8runtime::V8RuntimeArgs args{};
         args.inspectorPort = 5643;
-        args.foreground_task_runner = std::make_shared<TaskRunnerAdapter>(*m_workQueue);
+        args.foreground_task_runner = std::make_shared<TaskRunnerAdapter>(*this);
         const auto runtime = v8runtime::makeV8Runtime(std::move(args));
 
         const auto env = Napi::Attach(*runtime);