Work queue shutdown fixes

Author: bghgary

Core/AppRuntime/Source/WorkQueue.cpp

@@ -40,9 +40,10 @@ namespace Babylon
 
     void WorkQueue::Run(Napi::Env env)
     {
-        m_env = std::make_optional(env);
         m_dispatcher.set_affinity(std::this_thread::get_id());
 
+        m_env = std::make_optional(env);
+
         while (!m_cancellationSource.cancelled())
         {
             m_dispatcher.blocking_tick(m_cancellationSource);
@@ -53,5 +54,10 @@ namespace Babylon
 
         // There should no longer be any outstanding work once the queue is drained.
         assert(m_dispatcher.empty());
+
+        // Clear the shutdown queue to make sure the callables are destroyed on this thread.
+        m_shutdownQueue.clear();
+
+        m_env.reset();
     }
 }

Core/AppRuntime/Source/WorkQueue.h

@@ -19,21 +19,33 @@ namespace Babylon
         template<typename CallableT>
         void Append(CallableT callable)
         {
-            if (m_cancellationSource.cancelled())
-            {
-                // There is likely a coding error if this exception is thrown.
-                throw std::runtime_error{"Cannot append to the work queue while shutting down"};
-            }
-
             // Manual dispatcher queueing requires a copyable CallableT, we use a shared pointer trick to make a
             // copyable callable if necessary.
             if constexpr (std::is_copy_constructible<CallableT>::value)
             {
-                m_dispatcher.queue([this, callable = std::move(callable)]() { Invoke(callable); });
+                if (m_cancellationSource.cancelled())
+                {
+                    m_shutdownQueue.push([callable = std::move(callable)] {});
+                }
+                else
+                {
+                    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))]() { Invoke(*callablePtr); });
+                if (m_cancellationSource.cancelled())
+                {
+                    m_shutdownQueue.push([callablePtr = std::make_shared<CallableT>(std::move(callable))] {});
+                }
+                else
+                {
+                    m_dispatcher.queue([this, callablePtr = std::make_shared<CallableT>(std::move(callable))]() {
+                        (*callablePtr)(m_env.value());
+                    });
+                }
             }
         }
 
@@ -42,18 +54,17 @@ namespace Babylon
         void Run(Napi::Env);
 
     private:
-        template<typename CallableT>
-        void Invoke(CallableT& callable)
-        {
-            callable(m_env.value());
-        }
-
         std::optional<Napi::Env> m_env{};
 
         std::optional<std::scoped_lock<std::mutex>> m_suspensionLock{};
 
         arcana::cancellation_source m_cancellationSource{};
-        arcana::manual_dispatcher<128> m_dispatcher{};
+
+        using DispatcherT = arcana::manual_dispatcher<128>;
+        DispatcherT m_dispatcher{};
+
+        // Put the callables in a separate queue during shutdown to ensure the callables are destroyed on the right thread.
+        arcana::blocking_concurrent_queue<DispatcherT::callback_t> m_shutdownQueue{};
 
         std::thread m_thread{};
     };