Update DelegateMQ library

Author: endurodave

.github/workflows/cmake_clang.yml

@@ -24,3 +24,5 @@ jobs:
 
     - name: Run IntegrationTestFrameworkApp
       run: ./build/IntegrationTestFrameworkApp  # Run the built executable
+
+

.github/workflows/cmake_ubuntu.yml

@@ -24,3 +24,5 @@ jobs:
 
     - name: Run IntegrationTestFrameworkApp
       run: ./build/IntegrationTestFrameworkApp  # Run the built executable
+
+

.github/workflows/cmake_windows.yml

@@ -8,22 +8,22 @@ on:
     branches:
       - main  # Trigger on pull 
 
+
 jobs:
   build:
-    runs-on: windows-latest  # Use Windows environment for the build
+    runs-on: windows-2022  # Use Windows environment for the build
 
     steps:
     - name: Checkout code
       uses: actions/checkout@v4  # Checkout the repository code
 
-    - name: Set up Visual Studio
-      uses: microsoft/setup-msbuild@v2  # Set up Visual Studio environment (MSBuild)
-
     - name: Configure CMake
-      run: cmake -S . -B build -G "Visual Studio 17 2022" -DENABLE_IT=ON # Configure CMake for Visual Studio
+      run: cmake -S . -B build -DENABLE_IT=ON # Configure CMake for Visual Studio
 
     - name: Build
       run: cmake --build build --config Release  # Build the project using CMake with Release configuration
 
     - name: Run IntegrationTestFrameworkApp
       run: .\build\Release\IntegrationTestFrameworkApp.exe  # Run the built executable (adjust path for MSBuild)
+
+

DelegateMQ/DelegateMQ.h

@@ -233,12 +233,15 @@
 #endif
 
 #include "extras/util/Fault.h"
+#include "extras/util/ClockHelper.h"
 
 // Only include Timer and AsyncInvoke if threads exist
 #if !defined(DMQ_THREAD_NONE)
     #include "extras/util/Timer.h"
+    #include "extras/util/TimerDelegate.h"
     #include "extras/util/AsyncInvoke.h"
     #include "extras/util/TransportMonitor.h"
+    #include "extras/util/ThreadMonitor.h"
 #endif
 
 // Only include NetworkEngine if a transport that uses it is active

DelegateMQ/delegate/DelegateAsync.h

@@ -280,7 +280,8 @@ class DelegateFreeAsync<RetType(Args...)> : public DelegateFree<RetType(Args...)
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destintation thread. 
-                thread->DispatchDelegate(msg);
+                bool success = thread->DispatchDelegate(msg);
+                (void)success;
             }
 
             // Do not wait for destination thread return value from async function call
@@ -593,7 +594,8 @@ class DelegateMemberAsync<TClass, RetType(Args...)> : public DelegateMember<TCla
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destintation thread. 
-                thread->DispatchDelegate(msg);
+                bool success = thread->DispatchDelegate(msg);
+                (void)success;
             }
 
             // Do not wait for destination thread return value from async function call
@@ -835,7 +837,8 @@ class DelegateMemberAsyncSp<TClass, RetType(Args...)> : public DelegateMemberSp<
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destintation thread. 
-                thread->DispatchDelegate(msg);
+                bool success = thread->DispatchDelegate(msg);
+                (void)success;
             }
 
             // Do not wait for destination thread return value from async function call
@@ -1089,7 +1092,8 @@ class DelegateFunctionAsync<RetType(Args...)> : public DelegateFunction<RetType(
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destintation thread. 
-                thread->DispatchDelegate(msg);
+                bool success = thread->DispatchDelegate(msg);
+                (void)success;
             }
 
             // Do not wait for destination thread return value from async function call

DelegateMQ/delegate/DelegateAsyncWait.h

@@ -342,14 +342,14 @@ class DelegateFreeAsyncWait<RetType(Args...)> : public DelegateFree<RetType(Args
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destination thread. 
-                thread->DispatchDelegate(msg);
-
-                // Wait for destination thread to execute the delegate function and get return value
-                if (msg->GetSema().Wait(m_timeout)) {
-                    // Wait succeeded. Now acquire lock to safely read the value.
-                    const dmq::LockGuard<Mutex> lock(msg->GetLock());
-                    m_success = true;
-                    m_retVal = delegate->m_retVal;
+                if (thread->DispatchDelegate(msg)) {
+                    // Wait for destination thread to execute the delegate function and get return value
+                    if (msg->GetSema().Wait(m_timeout)) {
+                        // Wait succeeded. Now acquire lock to safely read the value.
+                        const dmq::LockGuard<Mutex> lock(msg->GetLock());
+                        m_success = true;
+                        m_retVal = delegate->m_retVal;
+                    }
                 }
             }
 
@@ -778,14 +778,14 @@ class DelegateMemberAsyncWait<TClass, RetType(Args...)> : public DelegateMember<
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destination thread. 
-                thread->DispatchDelegate(msg);
-
-                // Wait for destination thread to execute the delegate function and get return value
-                if (msg->GetSema().Wait(m_timeout)) {
-                    // Wait succeeded. Now acquire lock to safely read the value.
-                    const dmq::LockGuard<Mutex> lock(msg->GetLock());
-                    m_success = true;
-                    m_retVal = delegate->m_retVal;
+                if (thread->DispatchDelegate(msg)) {
+                    // Wait for destination thread to execute the delegate function and get return value
+                    if (msg->GetSema().Wait(m_timeout)) {
+                        // Wait succeeded. Now acquire lock to safely read the value.
+                        const dmq::LockGuard<Mutex> lock(msg->GetLock());
+                        m_success = true;
+                        m_retVal = delegate->m_retVal;
+                    }
                 }
             }
 
@@ -1131,14 +1131,14 @@ class DelegateMemberAsyncWaitSp<TClass, RetType(Args...)> : public DelegateMembe
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destination thread. 
-                thread->DispatchDelegate(msg);
-
-                // Wait for destination thread to execute the delegate function and get return value
-                if (msg->GetSema().Wait(m_timeout)) {
-                    // Wait succeeded. Now acquire lock to safely read the value.
-                    const dmq::LockGuard<Mutex> lock(msg->GetLock());
-                    m_success = true;
-                    m_retVal = delegate->m_retVal;
+                if (thread->DispatchDelegate(msg)) {
+                    // Wait for destination thread to execute the delegate function and get return value
+                    if (msg->GetSema().Wait(m_timeout)) {
+                        // Wait succeeded. Now acquire lock to safely read the value.
+                        const dmq::LockGuard<Mutex> lock(msg->GetLock());
+                        m_success = true;
+                        m_retVal = delegate->m_retVal;
+                    }
                 }
             }
 
@@ -1486,14 +1486,14 @@ class DelegateFunctionAsyncWait<RetType(Args...)> : public DelegateFunction<RetT
             if (thread) {
                 // Dispatch message onto the callback destination thread. Invoke()
                 // will be called by the destination thread. 
-                thread->DispatchDelegate(msg);
-
-                // Wait for destination thread to execute the delegate function and get return value
-                if (msg->GetSema().Wait(m_timeout)) {
-                    // Wait succeeded. Now acquire lock to safely read the value.
-                    const dmq::LockGuard<Mutex> lock(msg->GetLock());
-                    m_success = true;
-                    m_retVal = delegate->m_retVal;
+                if (thread->DispatchDelegate(msg)) {
+                    // Wait for destination thread to execute the delegate function and get return value
+                    if (msg->GetSema().Wait(m_timeout)) {
+                        // Wait succeeded. Now acquire lock to safely read the value.
+                        const dmq::LockGuard<Mutex> lock(msg->GetLock());
+                        m_success = true;
+                        m_retVal = delegate->m_retVal;
+                    }
                 }
             }
 

DelegateMQ/delegate/DelegateOpt.h

@@ -139,6 +139,25 @@ namespace dmq
     using Duration = typename Clock::duration;
     using TimePoint = typename Clock::time_point;
 
+    /// @brief Default timeout for the TIMEOUT queue-full policy across all Thread ports.
+    /// Override per-thread at construction or project-wide before including this header.
+    inline constexpr std::chrono::seconds DEFAULT_DISPATCH_TIMEOUT{2};
+
+    // --- RESOURCE LIMITS & SBO CONFIGURATION ---
+    
+    /// @brief Max timers processed in one tick without heap allocation.
+    inline constexpr size_t MAX_TIMER_EXPIRED = 16;
+
+    /// @brief Signal Small-Buffer Optimization (SBO) count. 
+    /// Signals with <= this many subscribers are invoked heap-free.
+    inline constexpr size_t SIGNAL_SBO_COUNT = 8;
+
+    /// @brief Default internal queue size for all dmq::os::Thread ports.
+    inline constexpr size_t DEFAULT_QUEUE_SIZE = 20;
+
+    /// @brief Max number of threads that can be monitored by the watchdog.
+    inline constexpr size_t MAX_WATCHDOG_THREADS = 16;
+
     // --- MUTEX / LOCK SELECTION ---
 #if defined(DMQ_THREAD_STDLIB) || defined(DMQ_THREAD_WIN32) || defined(DMQ_THREAD_QT)
     // Windows / Linux / macOS / Qt

DelegateMQ/delegate/IThread.h

@@ -35,7 +35,8 @@ class IThread
 	///
 	/// @param[in] msg A shared pointer to the delegate message. This pointer must remain valid
 	/// until the target thread finishes execution.
-	virtual void DispatchDelegate(std::shared_ptr<DelegateMsg> msg) = 0;
+	/// @return true if the message was successfully enqueued, false otherwise.
+	virtual bool DispatchDelegate(std::shared_ptr<DelegateMsg> msg) = 0;
 
 	/// @brief Returns true if the calling thread is this thread.
 	/// Used to decide whether to marshal an event or execute inline.

DelegateMQ/delegate/MulticastDelegate.h

@@ -233,13 +233,13 @@ class MulticastDelegate<RetType(Args...)>
         int& m_cnt;
         MulticastDelegate* m_container;
     };
-
+protected:
     /// List of registered delegates
     xlist<std::shared_ptr<DelegateType>> m_delegates;
 
     /// Count of active nested broadcasts
     int m_broadcastCount = 0;
-    
+
     /// Flag for handling lazy delete
     bool m_cleanup = false;
 };

DelegateMQ/delegate/MulticastDelegateSafe.h

@@ -47,16 +47,47 @@ class MulticastDelegateSafe<RetType(Args...)> : public MulticastDelegate<RetType
     /// A void return value is used since multiple targets invoked.
     /// @param[in] args The arguments used when invoking the target functions
     void operator()(Args... args) {
-        const dmq::LockGuard<RecursiveMutex> lock(m_lock);
-        BaseType::operator ()(args...);
+        // To prevent deadlocks, the mutex must be released before invoking the 
+        // delegate. Circular lock dependencies can occur if the delegate target 
+        // function itself attempts to acquire a lock that is held by the 
+        // thread invoking the delegate. 
+        // Use a small-buffer optimization to avoid heap allocation in the common case.
+        std::shared_ptr<DelegateType> small_buf[SIGNAL_SBO_COUNT];
+        xlist<std::shared_ptr<DelegateType>> large_buf;
+        size_t count = 0;
+
+        {
+            const dmq::LockGuard<RecursiveMutex> lock(m_lock);
+            count = this->m_delegates.size();
+            if (count <= SIGNAL_SBO_COUNT) {
+                size_t i = 0;
+                for (auto& d : this->m_delegates) {
+                    small_buf[i++] = d;
+                }
+            } else {
+                large_buf = this->m_delegates;
+            }
+        }
+
+        if (count <= SIGNAL_SBO_COUNT) {
+            for (size_t i = 0; i < count; ++i) {
+                if (small_buf[i])
+                    (*small_buf[i])(args...);
+                small_buf[i].reset(); // Clear to release shared_ptr immediately
+            }
+        } else {
+            for (auto& d : large_buf) {
+                if (d)
+                    (*d)(args...);
+            }
+        }
     }
 
     /// Invoke all bound target functions. A void return value is used 
     /// since multiple targets invoked.
     /// @param[in] args The arguments used when invoking the target functions
     void Broadcast(Args... args) {
-        const dmq::LockGuard<RecursiveMutex> lock(m_lock);
-        BaseType::Broadcast(args...);
+        operator()(args...);
     }
 
     /// Insert a delegate into the container.