EventThread.hpp

Go to the documentation of this file.

#pragma once
 
#include <functional>
#include <variant>
 
#include <zephyr/kernel.h>
#include <zephyr/logging/log.h>
 
#include "Timer.hpp"
#include "TimerManager.hpp"
 
namespace zct {
 
// Don't use constexpr here, it seg faults!
// static constexpr int LOG_LEVEL = LOG_LEVEL_DBG;
#define ZCT_EVENT_THREAD_LOG_LEVEL LOG_LEVEL_WRN
 
template <typename EventType>
class EventThread {
 
public:
 
    using MsgQueueItem = std::variant<EventType, std::function<void()>>;
 
    EventThread(
        const char* name,
        k_thread_stack_t* threadStack,
        size_t threadStackSize,
        int threadPriority,
        size_t eventQueueBufferNumItems
    ) :
        m_threadStack(threadStack),
        m_threadStackSize(threadStackSize),
        m_threadPriority(threadPriority),
        m_timerManager(10)
    {
        LOG_MODULE_DECLARE(EventThread, ZCT_EVENT_THREAD_LOG_LEVEL);
        LOG_DBG("EventThread constructor called.");
        m_name = name;
 
        // Create event queue buffer and then init queue with it
        m_msgQueueBuffer = new MsgQueueItem[eventQueueBufferNumItems];
        __ASSERT_NO_MSG(m_msgQueueBuffer != nullptr);
        k_msgq_init(&m_threadMsgQueue, (char*)m_msgQueueBuffer, sizeof(MsgQueueItem), eventQueueBufferNumItems);
    };
 
    ~EventThread() {
        LOG_MODULE_DECLARE(EventThread, ZCT_EVENT_THREAD_LOG_LEVEL);
        LOG_DBG("%s() called.", __FUNCTION__);
        k_thread_join(&m_thread, K_FOREVER);
    }
 
    void start() {
        LOG_MODULE_DECLARE(EventThread, ZCT_EVENT_THREAD_LOG_LEVEL);
        LOG_DBG("EventThread start() called.");
        
        // Start the thread
        k_thread_create(
            &m_thread,
            m_threadStack,
            m_threadStackSize,
            staticThreadFunction,
            this, // Pass in the instance of the class
            NULL,
            NULL,
            m_threadPriority,
            0,
            K_NO_WAIT);
        // Name the thread for easier debugging/logging
        k_thread_name_set(&m_thread, m_name);
    }
 
    void onExternalEvent(std::function<void(const EventType&)> callback) {
        m_externalEventCallback = callback;
    }
 
    void sendEvent(const EventType& event) {
        MsgQueueItem item = event;
        k_msgq_put(&m_threadMsgQueue, &item, K_NO_WAIT);
    }
 
    TimerManager& timerManager() {
        return m_timerManager;
    }
 
    void exitEventLoop() {
        m_exitEventLoop = true;
    }
 
    void runInLoop(std::function<void()> func) {
        MsgQueueItem item = func;
        k_msgq_put(&m_threadMsgQueue, &item, K_NO_WAIT);
    }
 
protected:
 
    static void staticThreadFunction(void* arg1, void* arg2, void* arg3)
    {
        LOG_MODULE_DECLARE(EventThread, ZCT_EVENT_THREAD_LOG_LEVEL);
        ARG_UNUSED(arg2);
        ARG_UNUSED(arg3);
 
        // First passed in argument is the instance of the class
        EventThread* obj = static_cast<EventThread*>(arg1);
        // Run the event loop. This should not return unless the user calls exitEventLoop().
        obj->runEventLoop();
        // If we get here, the user decided to exit the thread
    }
 
    void runEventLoop() {
        LOG_MODULE_DECLARE(EventThread, ZCT_EVENT_THREAD_LOG_LEVEL);
        LOG_DBG("%s() called.", __FUNCTION__);
 
        while (!m_exitEventLoop) {
            // Get the next timer to expire
            auto nextTimerInfo = m_timerManager.getNextExpiringTimer();
 
        // Check for expired timers and call their callbacks
        while (nextTimerInfo.m_timer != nullptr && nextTimerInfo.m_durationToWaitUs == 0) {
            LOG_DBG("Timer expired. Timer: %p.", nextTimerInfo.m_timer);
            // Update the timer after expiry
            nextTimerInfo.m_timer->updateAfterExpiry();
            
            // Call the callback
            const auto& callback = nextTimerInfo.m_timer->getExpiryCallback();
            if (callback) {
                callback();
                // If the callback calls exitEventLoop(), we will exit the event loop
                // and return from runEventLoop().
                if (m_exitEventLoop) {
                    return;
                }
            }
            
            // Check for the next expired timer
            nextTimerInfo = m_timerManager.getNextExpiringTimer();
        }
 
        // If we get here, we have handled all expired timers.
        k_timeout_t timeout;
        if (nextTimerInfo.m_timer != nullptr) {
            timeout = Z_TIMEOUT_US(nextTimerInfo.m_durationToWaitUs);
        } else {
            timeout = K_FOREVER;
        }
 
        // Block on message queue until next timer expiry
        // Create storage in the case we receive an external event
        MsgQueueItem msgQueueItem;
        int queueRc = k_msgq_get(&m_threadMsgQueue, &msgQueueItem, timeout);
        if (queueRc == 0) {
            // We got a message from the queue, it will either be an event or a function to run in the context of the event thread.
            if (std::holds_alternative<EventType>(msgQueueItem)) {
                // It's an event, call the external event callback
                const auto& event = std::get<EventType>(msgQueueItem);
                if (m_externalEventCallback) {
                    m_externalEventCallback(event);
                    // If the callback calls exitEventLoop(), we will exit the event loop
                    // and return from runEventLoop().
                    if (m_exitEventLoop) {
                        return;
                    }
                } else {
                    LOG_WRN("Received external event in event thread \"%s\" but no external event callback is registered.", m_name);
                }
            } else {
                // It's a function to run in the context of the event thread, run it
                std::get<std::function<void()>>(msgQueueItem)();
            }
            continue;
        } else if (queueRc == -EAGAIN) {
            // Queue timed out, which means we need to handle the timer expiry,
            // jump back to start of while loop to handle the timer expiry
            LOG_DBG("Queue timed out, which means we need to handle timer expiry.");
            continue;
        } else if (queueRc == -ENOMSG) {
            // This means the queue must have been purged
            __ASSERT(false, "Got -ENOMSG from queue, was not expecting this.");
        } else {
            __ASSERT(false, "Got unexpected return code from queue: %d.", queueRc);
        }
        } // End of while (true) loop
    }
 
    const char* m_name = nullptr;
    void* m_msgQueueBuffer = nullptr;
 
    struct k_thread m_thread;
    struct k_msgq m_threadMsgQueue;
 
    k_thread_stack_t* m_threadStack;
    size_t m_threadStackSize;
    int m_threadPriority;
 
    TimerManager m_timerManager;
    std::function<void(const EventType&)> m_externalEventCallback = nullptr;
 
    bool m_exitEventLoop = false;
};
 
} // namespace zct