Improve thread startup and shutdown handling.

Author: endurodave

README.md

@@ -19,7 +19,7 @@ Create a worker thread with an event loop, message queue and a timer using the C
   - [Event Loop (Win32)](#event-loop-win32)
 - [Timer](#timer)
 - [Usage](#usage)
-- [Star History](#star-history)
+- [Related Repositories](#related-repositories)
 - [Conclusion](#conclusion)
 
 # Preface
@@ -314,11 +314,16 @@ int main(void)
     return 0;
 }</pre>
 
-# Star History
+# Related Repositories
 
-Find this repository useful? Consider giving it a star!
+Real-world projects using `WorkerThread`.
 
-[![Star History Chart](https://api.star-history.com/svg?repos=endurodave/StdWorkerThread&type=Date)](https://star-history.com/#endurodave/StdWorkerThread&Date)
+* <a href="https://github.com/endurodave/DelegateMQ">Asynchronous Delegates in C++</a> - Invoke any C++ callable function synchronously, asynchronously, or on a remote endpoint.
+* <a href="https://github.com/endurodave/StateMachineWithThreads">C++ State Machine with Threads</a> - A framework combining C++ state machines and multicast asynchronous callbacks.
+* <a href="https://github.com/endurodave/StateMachineWithModernDelegates">C++ State Machine with Asynchronous Delegates</a> - A framework combining C++ state machines and asynchronous delegate callbacks.
+* <a href="https://github.com/endurodave/AsyncStateMachine">Asynchronous State Machine Design in C++</a> - An asynchronous C++ state machine implemented using an asynchronous delegate library.
+* <a href="https://github.com/endurodave/IntegrationTestFramework">Integration Test Framework using Google Test and Delegates</a> - A multi-threaded C++ software integration test framework using Google Test and DelegateMQ libraries.
+* <a href="https://github.com/endurodave/Async-SQLite">Asynchronous SQLite API using C++ Delegates</a> - An asynchronous SQLite wrapper implemented using an asynchronous delegate library.
 
 # Conclusion
 

WorkerThread.cpp

@@ -22,7 +22,11 @@ struct ThreadMsg
 //----------------------------------------------------------------------------
 // WorkerThread
 //----------------------------------------------------------------------------
-WorkerThread::WorkerThread(const std::string& threadName) : m_thread(nullptr), m_timerExit(false), THREAD_NAME(threadName)
+WorkerThread::WorkerThread(const std::string& threadName) : 
+	m_thread(nullptr), 
+	m_exit(false), 
+	m_timerExit(false), 
+	THREAD_NAME(threadName)
 {
 }
 
@@ -41,20 +45,15 @@ bool WorkerThread::CreateThread()
 {
 	if (!m_thread)
 	{
+		m_threadStartFuture = m_threadStartPromise.get_future();
+
 		m_thread = std::unique_ptr<std::thread>(new thread(&WorkerThread::Process, this));
 
-#ifdef WIN32
-		// Get the thread's native Windows handle
 		auto handle = m_thread->native_handle();
+		SetThreadName(handle, THREAD_NAME);
 
-		// Set the thread name so it shows in the Visual Studio Debug Location toolbar
-		std::wstring wstr(THREAD_NAME.begin(), THREAD_NAME.end());
-		HRESULT hr = SetThreadDescription(handle, wstr.c_str());
-		if (FAILED(hr))
-		{
-			// Handle error if needed
-		}
-#endif
+		// Wait for the thread to enter the Process method
+		m_threadStartFuture.get();
 	}
 
 	return true;
@@ -77,6 +76,31 @@ std::thread::id WorkerThread::GetCurrentThreadId()
 	return this_thread::get_id();
 }
 
+//----------------------------------------------------------------------------
+// GetQueueSize
+//----------------------------------------------------------------------------
+size_t WorkerThread::GetQueueSize()
+{
+	lock_guard<mutex> lock(m_mutex);
+	return m_queue.size();
+}
+
+//----------------------------------------------------------------------------
+// SetThreadName
+//----------------------------------------------------------------------------
+void WorkerThread::SetThreadName(std::thread::native_handle_type handle, const std::string& name)
+{
+#ifdef WIN32
+	// Set the thread name so it shows in the Visual Studio Debug Location toolbar
+	std::wstring wstr(name.begin(), name.end());
+	HRESULT hr = SetThreadDescription(handle, wstr.c_str());
+	if (FAILED(hr))
+	{
+		// Handle error if needed
+	}
+#endif
+}
+
 //----------------------------------------------------------------------------
 // ExitThread
 //----------------------------------------------------------------------------
@@ -95,15 +119,25 @@ void WorkerThread::ExitThread()
 		m_cv.notify_one();
 	}
 
-    m_thread->join();
-    m_thread = nullptr;
+	m_exit.store(true);
+	m_thread->join();
+
+	// Clear the queue if anything added while waiting for join
+	{
+		lock_guard<mutex> lock(m_mutex);
+		m_thread = nullptr;
+		while (!m_queue.empty())
+			m_queue.pop();
+	}
 }
 
 //----------------------------------------------------------------------------
 // PostMsg
 //----------------------------------------------------------------------------
 void WorkerThread::PostMsg(std::shared_ptr<UserData> data)
 {
+	if (m_exit.load())
+		return;
 	ASSERT_TRUE(m_thread);
 
 	// Create a new ThreadMsg
@@ -123,7 +157,7 @@ void WorkerThread::TimerThread()
     while (!m_timerExit)
     {
         // Sleep for 250mS then put a MSG_TIMER into the message queue
-        std::this_thread::sleep_for(250ms);
+        std::this_thread::sleep_for(std::chrono::milliseconds(250));
 
         std::shared_ptr<ThreadMsg> threadMsg (new ThreadMsg(MSG_TIMER, 0));
 
@@ -139,6 +173,9 @@ void WorkerThread::TimerThread()
 //----------------------------------------------------------------------------
 void WorkerThread::Process()
 {
+	// Signal that the thread has started processing to notify CreateThread
+	m_threadStartPromise.set_value();
+
     m_timerExit = false;
     std::thread timerThread(&WorkerThread::TimerThread, this);
 

WorkerThread.h

@@ -10,6 +10,7 @@
 #include <atomic>
 #include <condition_variable>
 #include <string>
+#include <future>
 
 struct UserData
 {
@@ -47,6 +48,12 @@ class WorkerThread
     /// @param[in] data - thread specific message information
     void PostMsg(std::shared_ptr<UserData> msg);
 
+    /// Get size of thread message queue.
+    size_t GetQueueSize();
+
+    /// Get thread name
+    std::string GetThreadName() { return THREAD_NAME; }
+
 private:
     WorkerThread(const WorkerThread&) = delete;
     WorkerThread& operator=(const WorkerThread&) = delete;
@@ -57,12 +64,20 @@ class WorkerThread
     /// Entry point for timer thread
     void TimerThread();
 
+    void SetThreadName(std::thread::native_handle_type handle, const std::string& name);
+
     std::unique_ptr<std::thread> m_thread;
     std::queue<std::shared_ptr<ThreadMsg>> m_queue;
     std::mutex m_mutex;
     std::condition_variable m_cv;
     std::atomic<bool> m_timerExit;
     const std::string THREAD_NAME;
+
+    // Promise and future to synchronize thread start
+    std::promise<void> m_threadStartPromise;
+    std::future<void> m_threadStartFuture;
+
+    std::atomic<bool> m_exit;
 };
 
 #endif 

main.cpp

@@ -36,8 +36,7 @@ int main(void)
 	// Post the message to worker thread 2
 	workerThread2.PostMsg(userData2);
 
-	// Give time for messages processing on worker threads
-	this_thread::sleep_for(1s);
+	std::this_thread::sleep_for(std::chrono::seconds(1));
 
 	workerThread1.ExitThread();
 	workerThread2.ExitThread();