Update DelegateMQ library source

Author: endurodave

DelegateMQ/delegate/DelegateAsync.h

@@ -146,7 +146,7 @@ class DelegateFreeAsync<RetType(Args...)> : public DelegateFree<RetType(Args...)
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateFreeAsync(const ClassType& rhs) :
-        BaseType(rhs), m_thread(rhs.m_thread), m_priority(rhs.m_priority) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 
@@ -420,14 +420,14 @@ class DelegateMemberAsync<TClass, RetType(Args...)> : public DelegateMember<TCla
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateMemberAsync(const ClassType& rhs) :
-        BaseType(rhs), m_thread(rhs.m_thread) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 
     /// @brief Move constructor that transfers ownership of resources.
     /// @param[in] rhs The object to move from.
     DelegateMemberAsync(ClassType&& rhs) noexcept :
-        BaseType(rhs), m_thread(rhs.m_thread) {
+        BaseType(rhs), m_thread(rhs.m_thread), m_priority(rhs.m_priority) {
         rhs.Clear();
     }
 
@@ -709,14 +709,14 @@ class DelegateFunctionAsync<RetType(Args...)> : public DelegateFunction<RetType(
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateFunctionAsync(const ClassType& rhs) :
-        BaseType(rhs), m_thread(rhs.m_thread) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 
     /// @brief Move constructor that transfers ownership of resources.
     /// @param[in] rhs The object to move from.
     DelegateFunctionAsync(ClassType&& rhs) noexcept :
-        BaseType(rhs), m_thread(rhs.m_thread) {
+        BaseType(rhs), m_thread(rhs.m_thread), m_priority(rhs.m_priority) {
         rhs.Clear();
     }
 

DelegateMQ/delegate/DelegateAsyncWait.h

@@ -542,7 +542,7 @@ class DelegateMemberAsyncWait<TClass, RetType(Args...)> : public DelegateMember<
     /// @brief Move constructor that transfers ownership of resources.
     /// @param[in] rhs The object to move from.
     DelegateMemberAsyncWait(ClassType&& rhs) noexcept :
-        BaseType(rhs), m_thread(rhs.m_thread), m_timeout(rhs.m_timeout), m_success(rhs.m_success), m_retVal(rhs.m_retVal) {
+        BaseType(rhs), m_thread(rhs.m_thread), m_priority(rhs.m_priority), m_timeout(rhs.m_timeout), m_success(rhs.m_success), m_retVal(rhs.m_retVal) {
         rhs.Clear();
     }
 
@@ -923,7 +923,7 @@ class DelegateFunctionAsyncWait<RetType(Args...)> : public DelegateFunction<RetT
     /// @brief Move constructor that transfers ownership of resources.
     /// @param[in] rhs The object to move from.
     DelegateFunctionAsyncWait(ClassType&& rhs) noexcept :
-        BaseType(rhs), m_thread(rhs.m_thread), m_timeout(rhs.m_timeout), m_success(rhs.m_success), m_retVal(rhs.m_retVal) {
+        BaseType(rhs), m_thread(rhs.m_thread), m_priority(rhs.m_priority), m_timeout(rhs.m_timeout), m_success(rhs.m_success), m_retVal(rhs.m_retVal) {
         rhs.Clear();
     }
 

DelegateMQ/delegate/DelegateRemote.h

@@ -147,7 +147,7 @@ class DelegateFreeRemote<RetType(Args...)> : public DelegateFree<RetType(Args...
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateFreeRemote(const ClassType& rhs) :
-        BaseType(rhs), m_id(rhs.m_id) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 
@@ -621,7 +621,7 @@ class DelegateMemberRemote<TClass, RetType(Args...)> : public DelegateMember<TCl
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateMemberRemote(const ClassType& rhs) :
-        BaseType(rhs), m_id(rhs.m_id) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 
@@ -1109,7 +1109,7 @@ class DelegateFunctionRemote<RetType(Args...)> : public DelegateFunction<RetType
     /// set the state of the new instance.
     /// @param[in] rhs The object to copy from.
     DelegateFunctionRemote(const ClassType& rhs) :
-        BaseType(rhs), m_id(rhs.m_id) {
+        BaseType(rhs) {
         Assign(rhs);
     }
 

DelegateMQ/delegate/IThread.h

@@ -21,7 +21,7 @@ class IThread
 
 	/// Dispatch a `DelegateMsg` onto this thread. The implementer is responsible for
 	/// getting the `DelegateMsg` into an OS message queue. Once `DelegateMsg` is
-	/// on the destination thread of control, the `IDelegateInvoker::Invoke()` function
+	/// on the destination thread of control, the `IInvoker::Invoke()` function
 	/// must be called to execute the target function.
 	/// @param[in] msg A shared pointer to the message.
 	/// @post The destination thread calls `IThreadInvoker::Invoke()` when `DelegateMsg`

DelegateMQ/predef/os/freertos/Thread.h

@@ -16,6 +16,7 @@
 /// @TODO: Implement a priority queue for the thread messages like 
 /// std::priority_queue<ThreadMsg*, std::vector<ThreadMsg*>, ThreadMsgComparator>
 /// as implemented in stdlib\Thread.h if necessary.
+/// Optionally implement a watchdog mechanism as shown in stdlib\Thread.h.
 
 class ThreadMsg;
 

DelegateMQ/predef/os/stdlib/Thread.cpp

@@ -1,3 +1,4 @@
+#include "DelegateMQ.h"
 #include "Thread.h"
 #include "predef/util/Fault.h"
 
@@ -29,7 +30,7 @@ Thread::~Thread()
 //----------------------------------------------------------------------------
 // CreateThread
 //----------------------------------------------------------------------------
-bool Thread::CreateThread()
+bool Thread::CreateThread(std::optional<dmq::Duration> watchdogTimeout)
 {
 	if (!m_thread)
 	{
@@ -43,6 +44,27 @@ bool Thread::CreateThread()
 		// Wait for the thread to enter the Process method
 		m_threadStartFuture.get();
 
+		m_lastAliveTime.store(Timer::GetTime());
+
+		// Caller wants a watchdog timer?
+		if (watchdogTimeout.has_value())
+		{
+			// Create watchdog timer
+			m_watchdogTimeout = watchdogTimeout.value();
+
+			// Timer to ensure the Thread instance runs periodically. ThreadCheck invoked
+			// on this thread instance.
+			m_threadTimer = std::make_unique<Timer>();
+			m_threadTimer->Expired = MakeDelegate(this, &Thread::ThreadCheck, *this);
+			m_threadTimer->Start(m_watchdogTimeout.load() / 4);
+
+			// Timer to check that this Thread instance runs. WatchdogCheck invoked 
+			// on Timer::ProcessTimers() thread.
+			m_watchdogTimer = std::make_unique<Timer>();
+			m_watchdogTimer->Expired = MakeDelegate(this, &Thread::WatchdogCheck);
+			m_watchdogTimer->Start(m_watchdogTimeout.load() / 2);
+		}
+
 		LOG_INFO("Thread::CreateThread {}", THREAD_NAME);
 	}
 	return true;
@@ -100,6 +122,11 @@ void Thread::ExitThread()
 	if (!m_thread)
 		return;
 
+	if (m_watchdogTimer)
+		m_watchdogTimer->Stop();
+	if (m_threadTimer)
+		m_threadTimer->Stop();
+
 	// Create a new ThreadMsg
 	std::shared_ptr<ThreadMsg> threadMsg(new ThreadMsg(MSG_EXIT_THREAD, 0));
 
@@ -147,6 +174,34 @@ void Thread::DispatchDelegate(std::shared_ptr<dmq::DelegateMsg> msg)
 		typeid(*threadMsg->GetData()->GetInvoker()).name());
 }
 
+//----------------------------------------------------------------------------
+// WatchdogCheck
+//----------------------------------------------------------------------------
+void Thread::WatchdogCheck()
+{
+	auto now = Timer::GetTime();
+	auto lastAlive = m_lastAliveTime.load();
+
+	auto delta = Timer::Difference(lastAlive, now);
+
+	// Watchdog expired?
+	if (delta > m_watchdogTimeout.load())
+	{
+		LOG_ERROR("Watchdog detected unresponsive thread: {}", THREAD_NAME);
+
+		// @TODO You can optionally trigger recovery, restart, or further actions here
+		// For example, throw or notify external system
+	}
+}
+
+//----------------------------------------------------------------------------
+// ThreadCheck
+//----------------------------------------------------------------------------
+void Thread::ThreadCheck()
+{
+	// Do nothing
+}
+
 //----------------------------------------------------------------------------
 // Process
 //----------------------------------------------------------------------------
@@ -155,10 +210,12 @@ void Thread::Process()
 	// Signal that the thread has started processing to notify CreateThread
 	m_threadStartPromise.set_value();
 
-	LOG_INFO("Thread::Process {}", THREAD_NAME);
+	LOG_INFO("Thread::Process Start {}", THREAD_NAME);
 
 	while (1)
 	{
+		m_lastAliveTime.store(Timer::GetTime());
+
 		std::shared_ptr<ThreadMsg> msg;
 		{
 			// Wait for a message to be added to the queue
@@ -193,11 +250,15 @@ void Thread::Process()
 
 			case MSG_EXIT_THREAD:
 			{
+				LOG_INFO("Thread::Process Exit Thread {}", THREAD_NAME);
                 return;
 			}
 
 			default:
+			{
+				LOG_INFO("Thread::Process Invalid Message {}", THREAD_NAME);
 				throw std::invalid_argument("Invalid message ID");
+			}
 		}
 	}
 }

DelegateMQ/predef/os/stdlib/Thread.h

@@ -5,13 +5,15 @@
 // David Lafreniere, Feb 2017.
 
 #include "delegate/IThread.h"
+#include "./predef/util/Timer.h"
 #include "ThreadMsg.h"
 #include <thread>
 #include <queue>
 #include <mutex>
 #include <atomic>
 #include <condition_variable>
 #include <future>
+#include <optional>
 
 // Comparator for priority queue
 struct ThreadMsgComparator {
@@ -20,6 +22,21 @@ struct ThreadMsgComparator {
 	}
 };
 
+/// @brief Cross-platform thread for any system supporting C++11 std::thread (e.g. Windows, Linux).
+/// @details The Thread class creates a worker thread capable of dispatching and 
+/// invoking asynchronous delegates.
+///
+/// // Create thread with a watchdog timeout
+/// thread->CreateThread(std::chrono::milliseconds(2000));
+///
+/// // Or create without a watchdog
+/// thread->CreateThread();
+///
+/// WatchdogCheck() is invoked by a Timer instance. In a real-time operating system (RTOS),
+/// Timer::ProcessTimers() is typically called from the highest-priority task in the system.
+/// This ensures that any user thread becoming unresponsive can still be detected,
+/// since WatchdogCheck() runs at a higher priority. For mission-critical systems,
+/// a hardware watchdog should also be used as a fail-safe.
 class Thread : public dmq::IThread
 {
 public:
@@ -29,11 +46,13 @@ class Thread : public dmq::IThread
 	/// Destructor
 	~Thread();
 
-	/// Called once to create the worker thread
+	/// Called once to create the worker thread. If watchdogTimeout value 
+	/// provided, the maximum watchdog interval is used. Otherwise no watchdog.
+	/// @param[in] watchdogTimeout - optional watchdog timeout.
 	/// @return TRUE if thread is created. FALSE otherise. 
-	bool CreateThread();
+	bool CreateThread(std::optional<dmq::Duration> watchdogTimeout = std::nullopt);
 
-	/// Called once a program exit to exit the worker thread
+	/// Called once at program exit to shut down the worker thread
 	void ExitThread();
 
 	/// Get the ID of this thread instance
@@ -48,6 +67,9 @@ class Thread : public dmq::IThread
 	/// Get size of thread message queue.
 	size_t GetQueueSize();
 
+	/// Dispatch and invoke a delegate target on the destination thread.
+	/// @param[in] msg - Delegate message containing target function 
+	/// arguments.
 	virtual void DispatchDelegate(std::shared_ptr<dmq::DelegateMsg> msg);
 
 private:
@@ -59,6 +81,18 @@ class Thread : public dmq::IThread
 
 	void SetThreadName(std::thread::native_handle_type handle, const std::string& name);
 
+	/// Check watchdog is expired. This function is called by the thread 
+	/// the calls Timer::ProcessTimers(). This function is thread-safe.
+	/// In a real-time OS, Timer::ProcessTimers() typically is called by the highest
+	/// priority task in the system.
+	void WatchdogCheck();
+
+	/// Timer expiration function used to check that thread loop is running.
+	/// This function is called by this thread context (m_thread). The 
+	/// Thread::Process() function must be called periodically even if no
+	/// other user delegate events to be handled.
+	void ThreadCheck();
+
 	std::unique_ptr<std::thread> m_thread;
 	std::priority_queue<std::shared_ptr<ThreadMsg>,
 		std::vector<std::shared_ptr<ThreadMsg>>,
@@ -72,6 +106,12 @@ class Thread : public dmq::IThread
 	std::future<void> m_threadStartFuture;
 
 	std::atomic<bool> m_exit;
+
+	// Watchdog related members
+	std::atomic<dmq::Duration> m_lastAliveTime;
+	std::unique_ptr<Timer> m_watchdogTimer;
+	std::unique_ptr<Timer> m_threadTimer;
+	std::atomic<dmq::Duration> m_watchdogTimeout;
 };
 
 #endif 

DelegateMQ/predef/transport/mqtt/MqttTransport.h

@@ -27,7 +27,7 @@ class IMqttReceiveHandler
     virtual void Receive() = 0;
 };
 
-void debugTrace(enum MQTTCLIENT_TRACE_LEVELS level, char* message) {
+static void debugTrace(enum MQTTCLIENT_TRACE_LEVELS level, char* message) {
     //printf("TRACE [%d]: %s\n", level, message);
 }
 

DelegateMQ/predef/transport/win32-udp/Win32UdpTransport.h

@@ -35,7 +35,8 @@ class UdpTransport : public ITransport
 
     UdpTransport() : m_thread("UdpTransport"), m_sendTransport(this), m_recvTransport(this)
     {
-        m_thread.CreateThread();
+        // Create thread with a 5s watchdog timeout
+        m_thread.CreateThread(std::chrono::milliseconds(5000));
     }
 
     ~UdpTransport()

DelegateMQ/predef/transport/zeromq/ZeroMqTransport.h

@@ -33,7 +33,8 @@ class ZeroMqTransport : public ITransport
 
     ZeroMqTransport() : m_thread("ZeroMQTransport"), m_sendTransport(this), m_recvTransport(this)
     {
-        m_thread.CreateThread();
+        // Create thread with a 5s watchdog timeout
+        m_thread.CreateThread(std::chrono::milliseconds(5000));
     }
 
     ~ZeroMqTransport()