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 "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;

    bool Put(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

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

    bool PutFront(const void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

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

    bool Get(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

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

    bool Peek(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

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

    bool PeekFront(void *msg_ptr, Timeout timeout_ticks = WAIT_INFINITE);

    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); }
 
    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);
}
 
// ---------------------------------------------------------------------------
// Put
// ---------------------------------------------------------------------------
 
inline bool MessageQueue::Put(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_;
    bool success = true;
 
    while (m_count == m_capacity)
    {
        if (!m_cv_not_full.Wait(cs_, timeout_ticks))
        {
            success = false;
            break;
        }
    }
 
    if (success)
    {
        STK_MEMCPY(Slot(m_head), msg_ptr, m_msg_size);
        m_head = Next(m_head);
        m_count++;
 
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// PutFront
// ---------------------------------------------------------------------------
 
inline bool MessageQueue::PutFront(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_;
    bool success = true;
 
    while (m_count == m_capacity)
    {
        if (!m_cv_not_full.Wait(cs_, timeout_ticks))
        {
            success = false;
            break;
        }
    }
 
    if (success)
    {
        // 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 success;
}
 
// ---------------------------------------------------------------------------
// Get
// ---------------------------------------------------------------------------
 
inline bool MessageQueue::Get(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
    bool success = true;
 
    while (m_count == 0U)
    {
        if (!m_cv_not_empty.Wait(cs_, timeout_ticks))
        {
            success = false;
            break;
        }
    }
 
    if (success)
    {
        STK_MEMCPY(msg_ptr, Slot(m_tail), m_msg_size);
        m_tail = Next(m_tail);
        m_count--;
 
        m_cv_not_full.NotifyOne_CS();
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// Peek
// ---------------------------------------------------------------------------
 
inline bool MessageQueue::Peek(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
    bool success = true;
 
    while (m_count == 0U)
    {
        if (!m_cv_not_empty.Wait(cs_, timeout_ticks))
        {
            success = false;
            break;
        }
    }
 
    if (success)
    {
        // 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);
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// PeekFront
// ---------------------------------------------------------------------------
 
inline bool MessageQueue::PeekFront(void *msg_ptr, Timeout timeout_ticks)
{
    STK_ASSERT(msg_ptr != nullptr); // API contract: msg_ptr must not be null
 
    ScopedCriticalSection cs_;
    bool success = true;
 
    while (m_count == 0U)
    {
        if (!m_cv_not_empty.Wait(cs_, timeout_ticks))
        {
            success = false;
            break;
        }
    }
 
    if (success)
    {
        // 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);
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// 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(max_align_t)) uint8_t m_storage[N * MSG]; 
};
 
} // namespace sync
} // namespace stk
 
#endif /* STK_SYNC_MSGQUEUE_H_ */