C++ std::thread Event Loop

A worker thread with an event loop, priority message queue, back pressure, and watchdog using the C++17 standard library.

Table of Contents

• C++ std::thread Event Loop
• Table of Contents
• Getting Started
• Introduction
• Background
• Thread Class
• Event Loop
• Priority Queue
• Back Pressure
• Watchdog
• Usage
• Unit Tests
• Portable C/C++ Libraries
• References
• Conclusion

Getting Started

CMake is used to create the project build files on any Windows or Linux machine. The source code works on any C++17 compiler with std::thread support.

1. Clone the repository.
2. From the repository root, run the following CMake command: cmake -B Build .
3. Build and run the project within the Build directory. StdWorkerThreadApp runs the feature demos in main.cpp, then the unit tests in UnitTests.cpp, and exits non-zero if a test fails.

Introduction

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.

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 is now possible to create portable code and avoid the OS-specific function calls. This article provides an example of how to create a thread event loop and message queue while relying only upon the C++ standard library. Any C++17 compiler supporting the thread library should be able to compile the attached source.

Background

Typically, a thread needs 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. Many times, the worker thread is created at startup and is not destroyed until the application terminates.

A key requirement for the implementation is that incoming messages must execute on the same thread instance. Whereas std::async 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 Thread instance is used to safely dispatch function calls into the subsystem.

At first glance, the C++ thread support seems to be missing some key features. Yes, std::thread is available to spin off a thread but there is no thread-safe queue and no built-in back pressure or watchdog — services that many OS primitives provide. The sections below show how to use the C++ standard library to create these features and provide an event processing loop familiar to many programmers.

Thread Class

The Thread class encapsulates all the necessary event loop mechanisms. A simple interface allows thread creation, posting messages to the event loop, and eventual thread termination.

class Thread
{
public:
    Thread(const std::string& threadName, size_t maxQueueSize = 0, FullPolicy fullPolicy = FullPolicy::BLOCK);
    ~Thread();

    bool CreateThread(std::optional<std::chrono::milliseconds> watchdogTimeout = std::nullopt);
    void ExitThread();

    std::thread::id GetThreadId();
    static std::thread::id GetCurrentThreadId();
    bool IsCurrentThread();

    std::string GetThreadName();
    size_t GetQueueSize();
    size_t GetWatchdogAlarmCount() const;
    size_t GetProcessedCount() const;

    using MsgHandler = std::function<void(std::shared_ptr<UserData>)>;
    void SetMsgHandler(MsgHandler handler);

    void PostMsg(std::shared_ptr<UserData> msg, Priority priority = Priority::NORMAL);
};

Thread uses std::optional<std::thread> internally, which allows the thread handle to be cleanly reset between CreateThread() and ExitThread() calls without requiring heap allocation.

SetMsgHandler(), called before CreateThread(), sets the function the worker thread calls for each message. Without one, each message is printed to std::cout. The unit tests use it to run test code on the worker thread.

CreateThread() uses std::promise and std::future to guarantee the worker thread has entered its event loop before returning, so callers can safely post messages immediately after creation.

Event Loop

Process() is the worker thread's event loop. It runs for the lifetime of the thread, blocking when there is nothing to do and waking to process one message at a time in priority order.

The loop has three distinct responsibilities that interact with each other, explained below.

1. Watchdog heartbeat

m_lastAliveTime is written at the top of every iteration before the thread blocks. This is the timestamp the watchdog checks. As long as the loop keeps cycling, this stays current and the watchdog stays silent. If the thread gets stuck inside a message handler and stops reaching the top of the loop, the timestamp goes stale and the watchdog fires.

2. Blocking wait — with or without a heartbeat timeout

When no watchdog is configured the thread blocks indefinitely with m_cv.wait() until a message arrives or ExitThread() wakes it. When a watchdog is active the wait is replaced with m_cv.wait_for() using a heartbeat interval of timeout / 4. This causes the loop to wake periodically even when the queue is empty, so m_lastAliveTime stays fresh. Without this, an idle thread would look identical to a stalled one from the watchdog's perspective.

When the timed wait expires with an empty queue, the loop simply continues back to the top to refresh the timestamp and wait again.

3. Dequeue and back pressure release

Messages are dequeued from the first non-empty priority queue, checked in order HIGH, NORMAL, LOW. Each queue is a std::deque drained front-to-back, so messages posted at the same priority are always processed in the order they were posted. After removing a message, m_cvNotFull is signalled to wake any producer that blocked in PostMsg() because the queue was at its MAX_QUEUE_SIZE limit.

void Thread::Process()
{
    // Unblock CreateThread(), which waits on this promise before returning.
    m_threadStartPromise->set_value();

    while (1)
    {
        // Refresh watchdog timestamp before blocking.
        m_lastAliveTime.store(GetNow());

        shared_ptr<ThreadMsg> msg;
        {
            unique_lock<mutex> lk(m_mutex);

            auto empty = [this]() {
                return m_highQueue.empty() && m_normalQueue.empty() && m_lowQueue.empty();
            };

            if (m_watchdogTimeout.load() > steady_clock::duration::zero())
            {
                // Timed wait: wake periodically to keep m_lastAliveTime
                // current even when the queue is empty.
                auto heartbeat = m_watchdogTimeout.load() / 4;
                m_cv.wait_for(lk, heartbeat, [this, &empty]() {
                    return !empty() || m_exit.load();
                });
            }
            else
            {
                // No watchdog: block until a message arrives or exit is set.
                m_cv.wait(lk, [this, &empty]() {
                    return !empty() || m_exit.load();
                });
            }

            if (empty())
            {
                // Either the heartbeat fired (loop back, refresh timestamp)
                // or ExitThread() was called with nothing left (exit).
                if (m_exit.load()) return;
                continue;
            }

            // Dequeue from the highest-priority non-empty queue
            // (HIGH > NORMAL > LOW). Each queue is FIFO.
            if (!m_highQueue.empty())
            {
                msg = m_highQueue.front();
                m_highQueue.pop_front();
            }
            else if (!m_normalQueue.empty())
            {
                msg = m_normalQueue.front();
                m_normalQueue.pop_front();
            }
            else
            {
                msg = m_lowQueue.front();
                m_lowQueue.pop_front();
            }

            // Wake a producer blocked on a full queue.
            if (MAX_QUEUE_SIZE > 0)
                m_cvNotFull.notify_one();
        }

        switch (msg->GetId())
        {
            case MSG_POST_USER_DATA:
            {
                auto userData = msg->GetData();
                if (m_msgHandler)
                    m_msgHandler(userData);
                else
                    cout << userData->msg << " " << userData->year
                         << " on " << THREAD_NAME << endl;
                m_processedCount++;
                break;
            }
            case MSG_EXIT_THREAD:
                return;
            default:
                ASSERT();
        }
    }
}

PostMsg() adds a message to the queue and wakes the consumer:

void Thread::PostMsg(std::shared_ptr<UserData> data, Priority priority)
{
    if (m_exit.load())
        return;

    unique_lock<mutex> lk(m_mutex);

    auto totalSize = [this]() {
        return m_highQueue.size() + m_normalQueue.size() + m_lowQueue.size();
    };

    if (MAX_QUEUE_SIZE > 0 && totalSize() >= MAX_QUEUE_SIZE)
    {
        if (FULL_POLICY == FullPolicy::DROP)
            return;  // silently discard — caller is not stalled

        // BLOCK from the worker thread itself would deadlock
        ASSERT_TRUE(this_thread::get_id() != m_thread->get_id());

        m_cvNotFull.wait(lk, [this, &totalSize]() {
            return totalSize() < MAX_QUEUE_SIZE || m_exit.load();
        });
    }

    if (m_exit.load())
        return;

    auto threadMsg = make_shared<ThreadMsg>(MSG_POST_USER_DATA, data, priority);
    switch (priority)
    {
        case Priority::HIGH:   m_highQueue.push_back(threadMsg);   break;
        case Priority::NORMAL: m_normalQueue.push_back(threadMsg); break;
        case Priority::LOW:    m_lowQueue.push_back(threadMsg);    break;
    }
    m_cv.notify_one();
}

ExitThread() pushes a MSG_EXIT_THREAD message to the back of the Priority::LOW queue, sets the exit flag inside the lock, and notifies both the consumer and any blocked producers before joining the thread. Because the exit message is the last one in priority order, every message posted before ExitThread(), at any priority, is processed before the thread exits; PostMsg() calls made after it are discarded. The watchdog is stopped only after the join, so a thread that hangs while draining on shutdown is still reported.

Priority Queue

Messages are stored in one std::deque per priority level rather than a single combined queue:

enum class Priority { LOW = 0, NORMAL = 1, HIGH = 2 };

std::deque<std::shared_ptr<ThreadMsg>> m_highQueue;
std::deque<std::shared_ptr<ThreadMsg>> m_normalQueue;
std::deque<std::shared_ptr<ThreadMsg>> m_lowQueue;

PostMsg() appends to the deque matching the message's priority; Process() dequeues from the first non-empty deque, checked in order HIGH, NORMAL, LOW. When several messages are queued simultaneously, the worker thread always processes HIGH priority messages first, then NORMAL, then LOW, regardless of post order across levels — useful for giving urgent work, such as a shutdown or error signal, preferential access to the thread without a separate fast-path queue. Because each level is its own FIFO deque, messages posted at the same priority are always processed in the order they were posted.

An earlier version of this class used a single std::priority_queue with a comparator over Priority. That correctly ordered different priorities, but std::priority_queue's underlying binary heap does not preserve insertion order among equal elements — messages posted at the same priority could be dequeued out of order. The per-priority deque design fixes this while keeping the same HIGH > NORMAL > LOW ordering guarantee across levels.

Back Pressure

An optional maxQueueSize constructor argument caps how many messages may sit in the queue at once. When the queue is full, the FullPolicy determines the behavior:

• BLOCK: (default) PostMsg() blocks the calling thread on a second condition variable (m_cvNotFull) until the worker drains at least one message and signals that space is available. This cooperative throttling prevents a fast producer from outrunning a slow consumer and exhausting memory.
• DROP: PostMsg() silently discards the message and returns immediately. This is useful for high-rate, best-effort data like sensor telemetry or UI updates where a stale sample is preferable to stalling the producer.

Thread workerThread1("MyThread", 5); // BLOCK policy by default
Thread workerThread2("MyThread", 5, FullPolicy::DROP); // DROP policy enabled

When no limit is needed, pass 0 (the default) for maxQueueSize and PostMsg() never blocks or drops.

With BLOCK, a worker thread must not post to its own full queue: it would wait for a slot that only it can free, so PostMsg() faults (ASSERT_TRUE) instead of deadlocking. Size the queue for self-posts, or use DROP on a thread that posts to itself.

During shutdown ExitThread() sets the exit flag inside the lock and calls m_cvNotFull.notify_all(), ensuring any thread blocked in PostMsg() unblocks immediately rather than waiting for queue space that will never arrive.

Watchdog

Passing a timeout to CreateThread() enables a lightweight watchdog that runs on a dedicated thread alongside the worker.

workerThread.CreateThread(std::chrono::milliseconds(2000)); // 2-second watchdog

The worker thread stores a steady_clock::time_point (m_lastAliveTime) at the top of every loop iteration. The watchdog wakes every timeout / 2 and compares the current time against that timestamp:

void Thread::WatchdogProcess()
{
    auto timeout      = m_watchdogTimeout.load();
    auto sleepInterval = timeout / 2;

    while (true)
    {
        unique_lock<mutex> lk(m_watchdogMutex);
        if (m_watchdogCv.wait_for(lk, sleepInterval, [this]() { return m_watchdogExit.load(); }))
            break;

        auto delta = GetNow() - m_lastAliveTime.load();
        if (delta > timeout)
        {
            m_watchdogAlarms++;
            cerr << "Watchdog: thread '" << THREAD_NAME
                 << "' is unresponsive!" << endl;
        }
    }
}

If the timestamp goes stale — because the thread is stuck inside a message handler (deadlock, infinite loop, etc.) — the watchdog logs an error. In a production system the watchdog callback can be extended to trigger a controlled shutdown or system reset. The watchdog exits promptly when ExitThread() is called by signalling m_watchdogCv directly rather than waiting for the next sleep to expire; the wait's predicate also catches an exit request made before the wait began. GetWatchdogAlarmCount() returns how many alarms have been raised.

CreateThread() stores the watchdog timeout before starting the worker. The worker reads it to choose a timed or an indefinite queue wait; stored after the worker has started, an idle worker could already be in the indefinite wait, never refresh its timestamp, and be reported as unresponsive although it is only idle. UnitTests.cpp covers this case, an exited or re-created thread, and a stalled thread that must be reported.

Usage

The main() function below, from main.cpp, demonstrates each feature with its own thread, then runs the unit tests.

int main(void)
{
    Thread priorityThread("PriorityThread");                    // unlimited queue
    Thread blockThread("BlockThread", 5);                       // max 5 queued, BLOCK
    Thread dropThread("DropThread", 5, FullPolicy::DROP);       // max 5 queued, DROP
    Thread watchdogThread("WatchdogThread");                    // watchdog enabled
    Thread timerThread("TimerThread");                          // fed by a timer

    // CreateThread() returns once the thread is running, so posting right away is safe
    priorityThread.CreateThread();
    blockThread.CreateThread();
    dropThread.CreateThread();
    watchdogThread.CreateThread(milliseconds(2000));
    timerThread.CreateThread();

    // Posted LOW, NORMAL, HIGH; usually processed HIGH, NORMAL, LOW (the worker
    // may take the first message before the others are queued)
    cout << "\n-- Priority --" << endl;
    priorityThread.PostMsg(Msg("Low priority msg"), Priority::LOW);
    priorityThread.PostMsg(Msg("Normal priority msg"), Priority::NORMAL);
    priorityThread.PostMsg(Msg("High priority msg"), Priority::HIGH);

    // 10 messages into a queue of 5: PostMsg() blocks until a slot frees up
    cout << "\n-- Back pressure (BLOCK, max 5) --" << endl;
    for (int i = 1; i <= 10; i++)
        blockThread.PostMsg(Msg("Back pressure message #" + to_string(i)));

    // 10 messages into a queue of 5: once full, PostMsg() discards and returns
    cout << "\n-- Back pressure (DROP, max 5) --" << endl;
    for (int i = 1; i <= 10; i++)
        dropThread.PostMsg(Msg("Drop message #" + to_string(i)));

    // A healthy thread: the 2s watchdog stays silent. A handler that runs past
    // the timeout is reported (see Watchdog_StalledThreadReported in UnitTests.cpp).
    cout << "\n-- Watchdog (2000ms) --" << endl;
    watchdogThread.PostMsg(Msg("Watchdog-monitored message"));

    // A std::thread "timer" posts a LOW priority message every 250ms
    cout << "\n-- Timer (250ms) --" << endl;
    atomic<bool> timerExit(false);
    thread timer([&]() {
        while (!timerExit)
        {
            this_thread::sleep_for(milliseconds(250));
            if (!timerExit)
                timerThread.PostMsg(Msg("Timer expired"), Priority::LOW);
        }
    });
    this_thread::sleep_for(milliseconds(1000));
    timerExit = true;
    timer.join();   // stop posting before ExitThread()

    // ExitThread() processes everything still queued, then joins
    priorityThread.ExitThread();
    blockThread.ExitThread();
    dropThread.ExitThread();
    watchdogThread.ExitThread();
    timerThread.ExitThread();

    return RunUnitTests() == 0 ? 0 : 1;
}

Unit Tests

UnitTests.cpp tests priority order (across and within levels), FullPolicy (DROP drops and never stalls the producer; BLOCK delivers everything by stalling it; ExitThread() releases a blocked producer), ExitThread() draining the queue and re-creating the thread, and the watchdog (an idle, exited or re-created thread isn't reported; a stalled one is). They are adapted from the DelegateMQ std::thread port's tests. Each test uses a local Thread with SetMsgHandler(), and failures are counted rather than aborting, so every test runs.

Portable C/C++ Libraries

Explore portable-c-cpp-libs for reusable C/C++ components, including state machines, callbacks, threading, memory management, and more.

References

• Active Object C++ State Machine — Active-object C++ finite state machine with async dispatch and pub/sub signals.
• DelegateMQ - The DelegateMQ C++ library can invoke any callable function synchronously, asynchronously, or on a remote endpoint.
• Integration Test Framework - Integration test framework using Google Test and delegates.

Conclusion

The C++ standard library offers a platform-independent way to write multi-threaded application code without reliance on OS-specific APIs. The Thread class presented here builds on the basic event loop pattern to add a priority queue for ordered message processing, back pressure to protect against memory exhaustion, and an optional watchdog to detect stalled threads — all using only the C++17 standard library.