Add optional testing hooks for blocking_concurrent_queue

Add a per-instance before-wait callback to blocking_concurrent_queue and
expose it through dispatcher and manual_dispatcher. The callback is invoked
while holding the queue mutex, right before condition_variable::wait().

This enables deterministic testing of race conditions involving the
condition variable (e.g. lost wakeups during shutdown). All changes are
behind #ifdef ARCANA_TESTING_HOOKS and have zero cost when not enabled.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: bghgary

Source/Shared/arcana/threading/blocking_concurrent_queue.h

@@ -7,6 +7,10 @@
 #include <queue>
 #include <condition_variable>
 
+#ifdef ARCANA_TESTING_HOOKS
+#include <functional>
+#endif
+
 namespace arcana
 {
     template<typename T, size_t max_size = std::numeric_limits<size_t>::max()>
@@ -86,6 +90,14 @@ namespace arcana
             m_dataReady.notify_all();
         }
 
+#ifdef ARCANA_TESTING_HOOKS
+        void set_before_wait_callback(std::function<void()> callback)
+        {
+            std::unique_lock<std::mutex> lock{ m_mutex };
+            m_beforeWaitCallback = std::move(callback);
+        }
+#endif
+
     private:
         bool internal_pop(T& dest, const cancellation& cancel, bool block)
         {
@@ -95,6 +107,9 @@ namespace arcana
             {
                 while (!cancel.cancelled() && m_data.empty())
                 {
+#ifdef ARCANA_TESTING_HOOKS
+                    if (m_beforeWaitCallback) m_beforeWaitCallback();
+#endif
                     m_dataReady.wait(lock);
                 }
             }
@@ -116,6 +131,9 @@ namespace arcana
             {
                 while (!cancel.cancelled() && m_data.empty())
                 {
+#ifdef ARCANA_TESTING_HOOKS
+                    if (m_beforeWaitCallback) m_beforeWaitCallback();
+#endif
                     m_dataReady.wait(lock);
                 }
             }
@@ -135,5 +153,9 @@ namespace arcana
         std::queue<T> m_data;
         mutable std::mutex m_mutex;
         std::condition_variable m_dataReady;
+
+#ifdef ARCANA_TESTING_HOOKS
+        std::function<void()> m_beforeWaitCallback;
+#endif
     };
 }

Source/Shared/arcana/threading/dispatcher.h

@@ -45,6 +45,13 @@ namespace arcana
         dispatcher& operator=(dispatcher&&) = default;
         dispatcher(dispatcher&&) = default;
 
+#ifdef ARCANA_TESTING_HOOKS
+        void set_before_wait_callback(std::function<void()> callback)
+        {
+            m_work.set_before_wait_callback(std::move(callback));
+        }
+#endif
+
         bool tick(const cancellation& token)
         {
             return internal_tick(token, false);
@@ -114,6 +121,9 @@ namespace arcana
         using dispatcher<WorkSize>::clear;
         using dispatcher<WorkSize>::set_affinity;
         using dispatcher<WorkSize>::tick;
+#ifdef ARCANA_TESTING_HOOKS
+        using dispatcher<WorkSize>::set_before_wait_callback;
+#endif
     };
 
     template<size_t WorkSize>