C++ std::thread Event Loop with Message Queue and Timer
Create a worker thread with an event loop, message queue and a timer using the C++11 thread support library.
Table of Contents
Preface
Originally published on CodeProject at: <a href="http://www.codeproject.com/Articles/1169105/Cplusplus-std-thread-Event-Loop-with-Message-Queue"><strong>C++ std::thread Event Loop with Message Queue and Timer</strong></a>
Getting Started
CMake is used to create the project build files on any Windows or Linux machine. The source code works on any C++ compiler with std::thread support.
- Clone the repository.
- From the repository root, run the following CMake command:
cmake -B Build .
- Build and run the project within the
Build directory.
Introduction
<p>An event loop, or sometimes called a message loop, is a thread that waits for and dispatches incoming events. The thread blocks waiting for requests to arrive and then dispatches the event to an event handler function. A message queue is typically used by the loop to hold incoming messages. Each message is sequentially dequeued, decoded, and then an action is performed. Event loops are one way to implement inter-process communication.</p>
<p>All operating systems provide support for multi-threaded applications. Each OS has unique function calls for creating threads, message queues and timers. With the advent of the C++11 thread support library, it’s now possible to create portable code and avoid the OS-specific function calls. This article provides a simple example of how to create a thread event loop, message queue and timer services while only relying upon the C++ Standard Library. Any C++11 compiler supporting the thread library should be able to compile the attached source.</p>
Background
<p>Typically, I need a thread to operate as an event loop. Incoming messages are dequeued by the thread and data is dispatched to an appropriate function handler based on a unique message identifier. Timer support capable of invoking a function is handy for low speed polling or to generate a timeout if something doesn’t happen in the expected amount of time. Many times, the worker thread is created at startup and isn’t destroyed until the application terminates.</p>
<p>A key requirement for the implementation is that the incoming messages must execute on the same thread instance. Whereas say <code>std::async </code>may use a temporary thread from a pool, this class ensures that all incoming messages use the same thread. For instance, a subsystem could be implemented with code that is not thread-safe. A single <code>WorkerThread </code>instance is used to safely dispatch function calls into the subsystem.</p>
<p>At first glance, the C++ thread support seems to be missing some key features. Yes, <code>std::thread </code>is available to spin off a thread but there is no thread-safe queue and no timers – services that most OS’s provide. I’ll show how to use the C++ Standard Library to create these “missing” features and provide an event processing loop familiar to many programmers.</p>
WorkerThread
<p>The <code>WorkerThread </code>class encapsulates all the necessary event loop mechanisms. A simple class interface allows thread creation, posting messages to the event loop, and eventual thread termination. The interface is shown below:</p>
<pre lang="C++">
class WorkerThread
{
public:
/// Constructor
WorkerThread(const char* threadName);
/// Destructor
~WorkerThread();
/// Called once to create the worker thread
/// @return True if thread is created. False otherwise.
bool CreateThread();
/// Called once a program exit to exit the worker thread
void ExitThread();
/// Get the ID of this thread instance
/// @return The worker thread ID
std::thread::id GetThreadId();
/// Get the ID of the currently executing thread
/// @return The current thread ID
static std::thread::id GetCurrentThreadId();
/// Add a message to the thread queue
/// @param[in] data - thread specific message information
void PostMsg(std::shared_ptr<UserData> msg);
private:
WorkerThread(const WorkerThread&) = delete;
WorkerThread& operator=(const WorkerThread&) = delete;
/// Entry point for the worker thread
void Process();
/// Entry point for timer thread
void TimerThread();
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 char* THREAD_NAME;
};</pre>
<p>The first thing to notice is that <code>std::thread </code>is used to create a main worker thread. The main worker thread function is <code>Process()</code>.</p>
<pre lang="C++">
bool WorkerThread::CreateThread()
{
if (!m_thread)
m_thread = new thread(&WorkerThread::Process, this);
return true;
}</pre>
Event Loop
<p>The <code>Process() </code>event loop is shown below. The thread relies upon a <code>std::queue<ThreadMsg*> </code>for the message queue. <code>std::queue </code>is not thread-safe so all access to the queue must be protected by mutex. A <code>std::condition_variable </code>is used to suspend the thread until notified that a new message has been added to the queue.</p>
<pre lang="C++">
void WorkerThread::Process()
{
m_timerExit = false;
std::thread timerThread(&WorkerThread::TimerThread, this);
while (1)
{
std::shared_ptr<ThreadMsg> msg;
{
// Wait for a message to be added to the queue
std::unique_lock<std::mutex> lk(m_mutex);
while (m_queue.empty())
m_cv.wait(lk);
if (m_queue.empty())
continue;
msg = m_queue.front();
m_queue.pop();
}
switch (msg->id)
{
case MSG_POST_USER_DATA:
{
ASSERT_TRUE(msg->msg != NULL);
auto userData = std::static_pointer_cast<UserData>(msg->msg);
cout << userData->msg.c_str() << " " << userData->year << " on " << THREAD_NAME << endl;
break;
}
case MSG_TIMER:
cout << "Timer expired on " << THREAD_NAME << endl;
break;
case MSG_EXIT_THREAD:
{
m_timerExit = true;
timerThread.join();
return;
}
default:
ASSERT();
}
}
}</pre>
<p><code>PostMsg() </code>creates a new <code>ThreadMsg </code>on the heap, adds the message to the queue, and then notifies the worker thread using a condition variable.</p>
<pre lang="C++">
void WorkerThread::PostMsg(std::shared_ptr<UserData> data)
{
ASSERT
