stk_sync_msgqueue.h

Go to the documentation of this file.

/*
 * SuperTinyKernel(TM) RTOS: Lightweight High-Performance Deterministic C++ RTOS for Embedded Systems.
 *
 * Source: https://github.com/SuperTinyKernel-RTOS
 *
 * Copyright (c) 2022-2026 Neutron Code Limited <stk@neutroncode.com>. All Rights Reserved.
 * License: MIT License, see LICENSE for a full text.
 */
 
#ifndef STK_SYNC_MSGQUEUE_H_
#define STK_SYNC_MSGQUEUE_H_
 
#include <cstddef> // for std::max_align_t
 
#include "stk_sync_cv.h"

 
namespace stk {
namespace sync {

class MessageQueue : public ITraceable
{
public:
    static const size_t CAPACITY_MAX = 0xFFFEU;

    explicit MessageQueue(uint8_t *buf, size_t capacity, size_t msg_size);

    STK_VIRT_DTOR ~MessageQueue() = default;

    EWaitResult PutEx(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    bool Put(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE)
    {
        return (PutEx(msg_ptr, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

    bool TryPut(const void *msg_ptr) { return Put(msg_ptr, NO_WAIT); }

    EWaitResult PutFrontEx(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    bool PutFront(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE)
    {
        return (PutFrontEx(msg_ptr, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

    bool TryPutFront(const void *msg_ptr) { return PutFront(msg_ptr, NO_WAIT); }

    EWaitResult GetEx(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    bool Get(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE)
    {
        return (GetEx(msg_ptr, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

    bool TryGet(void *msg_ptr) { return Get(msg_ptr, NO_WAIT); }

    EWaitResult PeekEx(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    bool Peek(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE)
    {
        return (PeekEx(msg_ptr, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

    bool TryPeek(void *msg_ptr) { return Peek(msg_ptr, NO_WAIT); }

    EWaitResult PeekFrontEx(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    bool PeekFront(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE)
    {
        return (PeekFrontEx(msg_ptr, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

    bool TryPeekFront(void *msg_ptr) { return PeekFront(msg_ptr, NO_WAIT); }

    void Reset();

    size_t GetCapacity() const { return m_capacity; }

    size_t GetMsgSize() const { return m_msg_size; }

    size_t GetCount() const { return m_count; }

    size_t GetSpace() const { return (m_capacity - m_count); }

    uint8_t *GetBuffer() { return m_buffer; }

    bool IsEmpty() const { return (m_count == 0U); }

    bool IsFull() const { return (m_count == m_capacity); }

    bool IsStorageValid() const { return (m_buffer != nullptr); }
 
private:
    STK_NONCOPYABLE_CLASS(MessageQueue);
 
    // Get pointer to the raw storage for slot index \a idx.
    uint8_t *Slot(size_t idx) const { return m_buffer + (idx * m_msg_size); }
 
    // Advance a ring-buffer index forward by one with wrap-around.
    size_t Next(size_t idx) const { return (idx + 1U) % m_capacity; }
 
    // Retreat a ring-buffer index backward by one with wrap-around.
    // Uses a branch instead of modular arithmetic to avoid unsigned underflow.
    size_t Prev(size_t idx) const { return (idx == 0U) ? (m_capacity - 1U) : (idx - 1U); }
 
    // Wait until the queue has a free slot or the wait fails. Caller MUST hold the critical
    // section \a cs. Returns WAIT_RESULT_SIGNAL if a free slot is available, otherwise the
    // failed wait outcome (timeout, cancellation, kernel failure).
    EWaitResult WaitNotFull(IMutex &cs, Timeout timeout_ticks);
 
    // Wait until the queue has a message or the wait fails. Caller MUST hold the critical
    // section \a cs. Returns WAIT_RESULT_SIGNAL if a message is available, otherwise the
    // failed wait outcome (timeout, cancellation, kernel failure).
    EWaitResult WaitNotEmpty(IMutex &cs, Timeout timeout_ticks);
 
    uint8_t          *m_buffer;       
    const size_t      m_capacity;     
    const size_t      m_msg_size;     
    size_t            m_count;        
    size_t            m_head;         
    size_t            m_tail;         
    ConditionVariable m_cv_not_empty; 
    ConditionVariable m_cv_not_full;  
};
 
// ---------------------------------------------------------------------------
// Constructor
// ---------------------------------------------------------------------------
 
inline MessageQueue::MessageQueue(uint8_t *buf, size_t capacity, size_t msg_size)
: m_buffer(buf),
  m_capacity(capacity),
  m_msg_size(msg_size),
  m_count(0U),
  m_head(0U),
  m_tail(0U)
{
    STK_ASSERT(buf      != nullptr);
    STK_ASSERT(capacity >= 1U);
    STK_ASSERT(capacity <= CAPACITY_MAX);
    STK_ASSERT(msg_size >= 1U);
}
 
// ---------------------------------------------------------------------------
// WaitNotFull  (private helper, caller must hold the critical section)
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::WaitNotFull(IMutex &cs, Timeout timeout_ticks)
{
    EWaitResult result = WAIT_RESULT_SIGNAL;
 
    const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
 
    // capture an absolute deadline once, before entering the wait loop; a different
    // producer can steal the just-freed slot between NotifyOne_CS() and this task
    // re-acquiring the critical section (Mesa-style wakeup, not a direct handoff),
    // so re-passing the original timeout_ticks on each retry would silently restart it
    const Timeout deadline = (timed_wait ?
        static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
    while (m_count == m_capacity)
    {
        Timeout remaining = deadline;
        if (timed_wait)
        {
            const Timeout now = static_cast<Timeout>(GetTicks());
            remaining = (now >= deadline ? NO_WAIT : (deadline - now));
        }
 
        result = m_cv_not_full.WaitEx(cs, remaining);
        if (result != WAIT_RESULT_SIGNAL)
        {
            // timeout, cancellation or kernel failure
            break;
        }
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// WaitNotEmpty  (private helper, caller must hold the critical section)
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::WaitNotEmpty(IMutex &cs, Timeout timeout_ticks)
{
    EWaitResult result = WAIT_RESULT_SIGNAL;
 
    const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
 
    // capture an absolute deadline once, before entering the wait loop; a different
    // consumer can steal the just-produced message (or drain the queue via Get()/TryGet())
    // between NotifyOne_CS() and this task re-acquiring the critical section (Mesa-style
    // wakeup, not a direct handoff), so re-passing the original timeout_ticks on each
    // retry would silently restart it
    const Timeout deadline = (timed_wait ?
        static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
    while (m_count == 0U)
    {
        Timeout remaining = deadline;
        if (timed_wait)
        {
            const Timeout now = static_cast<Timeout>(GetTicks());
            remaining = (now >= deadline ? NO_WAIT : (deadline - now));
        }
 
        result = m_cv_not_empty.WaitEx(cs, remaining);
        if (result != WAIT_RESULT_SIGNAL)
        {
            // timeout, cancellation or kernel failure
            break;
        }
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// PutEx
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::PutEx(const void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr);             // API contract: msg_ptr must not be null
    STK_ASSERT(m_count <= (CAPACITY_MAX - 1U)); // API contract: must not exceed capacity
 
    ScopedCriticalSection cs_;
 
    const EWaitResult result = WaitNotFull(cs_, timeout_ticks);
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        STK_MEMCPY(Slot(m_head), msg_ptr, m_msg_size);
        m_head = Next(m_head);
        m_count++;
 
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// PutFrontEx
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::PutFrontEx(const void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr);             // API contract: msg_ptr must not be null
    STK_ASSERT(m_count <= (CAPACITY_MAX - 1U)); // API contract: must not exceed capacity
 
    ScopedCriticalSection cs_;
 
    const EWaitResult result = WaitNotFull(cs_, timeout_ticks);
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        // retreat the tail pointer to claim the slot that Get() would read next,
        // then write the message there; this makes the new message the head of
        // the logical sequence without touching m_head at all
        m_tail = Prev(m_tail);
        STK_MEMCPY(Slot(m_tail), msg_ptr, m_msg_size);
        m_count++;
 
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// GetEx
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::GetEx(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
 
    const EWaitResult result = WaitNotEmpty(cs_, timeout_ticks);
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        STK_MEMCPY(msg_ptr, Slot(m_tail), m_msg_size);
        m_tail = Next(m_tail);
        m_count--;
 
        m_cv_not_full.NotifyOne_CS();
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// PeekEx
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::PeekEx(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
 
    const EWaitResult result = WaitNotEmpty(cs_, timeout_ticks);
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        // copy from the tail slot without advancing the index or decrementing
        // the count, so the message remains available for the next Get()
        STK_MEMCPY(msg_ptr, Slot(m_tail), m_msg_size);
 
        // the message was not consumed: pass the wake-up on to the next waiter (if any),
        // otherwise a Get() waiter queued behind this Peek() would stay blocked although
        // a message is available (Put() notifies only one waiter)
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// PeekFrontEx
// ---------------------------------------------------------------------------
 
inline EWaitResult MessageQueue::PeekFrontEx(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
 
    const EWaitResult result = WaitNotEmpty(cs_, timeout_ticks);
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        // the front-inserted message is at m_tail (PutFront retreats m_tail then
        // writes, so the newly placed message is always at the current m_tail);
        // for a pure-Put queue this is equally correct: m_tail is the oldest slot
        STK_MEMCPY(msg_ptr, Slot(m_tail), m_msg_size);
 
        // the message was not consumed: pass the wake-up on to the next waiter (if any),
        // see PeekEx()
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return result;
}
 
// ---------------------------------------------------------------------------
// Reset
// ---------------------------------------------------------------------------
 
inline void MessageQueue::Reset()
{
    const ScopedCriticalSection cs_;
 
    m_count = 0U;
    m_head  = 0U;
    m_tail  = 0U;
 
    // wake all blocked producers: the queue is now entirely empty
    m_cv_not_full.NotifyAll_CS();
}
 
// ---------------------------------------------------------------------------

template <size_t N, size_t MSG>
class MessageQueueT : public MessageQueue
{
public:
    STK_STATIC_ASSERT_DESC(N   >= 1U,           "MessageQueueT: capacity N must be at least 1");
    STK_STATIC_ASSERT_DESC(N   <= CAPACITY_MAX, "MessageQueueT: capacity N must not exceed CAPACITY_MAX");
    STK_STATIC_ASSERT_DESC(MSG >= 1U,           "MessageQueueT: message size MSG must be at least 1");

    MessageQueueT() : MessageQueue(m_storage, N, MSG)
    {}

    STK_VIRT_DTOR ~MessageQueueT() = default;
 
private:
    STK_NONCOPYABLE_CLASS(MessageQueueT);
 
    alignas(alignof(std::max_align_t)) uint8_t m_storage[N * MSG]; 
};
 
} // namespace sync
} // namespace stk
 
#endif /* STK_SYNC_MSGQUEUE_H_ */