stk_sync_pipe.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_PIPE_H_
#define STK_SYNC_PIPE_H_
 
#include <type_traits> // for std::is_scalar
 
#include "stk_sync_cv.h"

 
namespace stk {
namespace sync {

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

    explicit Pipe(uint8_t *buf, size_t capacity, size_t element_size);

    STK_VIRT_DTOR ~Pipe() = default;

    bool Write(const void *data, Timeout timeout_ticks = WAIT_INFINITE);

    bool TryWrite(const void *data) { return Write(data, NO_WAIT); }

    size_t WriteBulk(const void *src, size_t count, Timeout timeout_ticks = WAIT_INFINITE);

    size_t TryWriteBulk(const void *src, size_t count) { return WriteBulk(src, count, NO_WAIT); }

    bool Read(void *data, Timeout timeout_ticks = WAIT_INFINITE);

    bool TryRead(void *data) { return Read(data, NO_WAIT); }

    size_t ReadBulk(void *dst, size_t count, Timeout timeout_ticks = WAIT_INFINITE);

    size_t TryReadBulk(void *dst, size_t count) { return ReadBulk(dst, count, NO_WAIT); }

    size_t ReadBulkTriggered(void *dst, size_t trigger, size_t max_count, 
                             Timeout timeout_ticks = WAIT_INFINITE);

    size_t TryReadBulkTriggered(void *dst, size_t max_count)
    {
        return ReadBulkTriggered(dst, 1U, max_count, NO_WAIT);
    }

    void Reset();

    size_t GetCapacity() const { return m_capacity; }

    size_t GetElementSize() const { return m_element_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(Pipe);
 
    // Get a byte pointer to the raw storage for slot index \a idx.
    uint8_t *Slot(size_t idx) const { return m_buffer + (idx * m_element_size); }
 
    // Advance a ring-buffer index by one with wrap-around.
    size_t Next(size_t idx) const { return (idx + 1U) % m_capacity; }
 
    // Copy Min(count, m_count) elements into dst_bytes and update tail/count.
    // Caller MUST hold the critical section. Returns the number of elements copied.
    size_t DrainLocked(uint8_t *dst_bytes, size_t count);
 
    uint8_t          *m_buffer;       
    const size_t      m_capacity;     
    const size_t      m_element_size; 
    size_t            m_count;        
    size_t            m_head;         
    size_t            m_tail;         
    ConditionVariable m_cv_not_empty; 
    ConditionVariable m_cv_not_full;  
};
 
// ---------------------------------------------------------------------------
// Pipe: Constructor
// ---------------------------------------------------------------------------
 
inline Pipe::Pipe(uint8_t *buf, size_t capacity, size_t element_size)
: m_buffer(buf),
  m_capacity(capacity),
  m_element_size(element_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(element_size >= 1U);
}
 
// ---------------------------------------------------------------------------
// Pipe: Write
// ---------------------------------------------------------------------------
 
inline bool Pipe::Write(const void *data, Timeout timeout_ticks)
{
    STK_ASSERT(data    != nullptr);
    STK_ASSERT(m_count <= (CAPACITY_MAX - 1U));
 
    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), data, m_element_size);
        m_head = Next(m_head);
        m_count++;
 
        // notify consumer that data is ready
        m_cv_not_empty.NotifyOne_CS();
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// Pipe: WriteBulk
// ---------------------------------------------------------------------------
 
inline size_t Pipe::WriteBulk(const void *src, size_t count, Timeout timeout_ticks)
{
    size_t written = 0U;
 
    if ((src != nullptr) && (count != 0U))
    {
        const uint8_t *const src_bytes = static_cast<const uint8_t *>(src);
        const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
        
        // capture an absolute deadline once, before entering the wait loop,
        // preventing the timeout from resetting on intermediate partial writes
        const Timeout deadline = (timed_wait ? 
            static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
        ScopedCriticalSection cs_;
 
        while (written < count)
        {
            bool is_timeout = false;
 
            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));
                }
 
                if (!m_cv_not_full.Wait(cs_, remaining))
                {
                    is_timeout = true;
                    break; // break inner condition variable loop
                }
            }
 
            // if a timeout occurred, drop out of the chunk processing loop
            if (is_timeout)
            {
                break;
            }
 
            const size_t available  = m_capacity - m_count;
            const size_t to_write   = ((count - written) < available) ? (count - written) : available;
            const size_t first_part = m_capacity - m_head;
 
            if (to_write <= first_part)
            {
                STK_MEMCPY(Slot(m_head), src_bytes + (written * m_element_size), to_write * m_element_size);
            }
            else
            {
                STK_MEMCPY(Slot(m_head), src_bytes + (written                * m_element_size), first_part              * m_element_size);
                STK_MEMCPY(Slot(0U),     src_bytes + ((written + first_part) * m_element_size), (to_write - first_part) * m_element_size);
            }
 
            written += to_write;
            m_head   = (m_head + to_write) % m_capacity;
            m_count += to_write;
 
            // notify consumers that data is ready
            m_cv_not_empty.NotifyAll_CS();
        }
    }
 
    return written;
}
 
// ---------------------------------------------------------------------------
// Pipe: Read
// ---------------------------------------------------------------------------
 
inline bool Pipe::Read(void *data, Timeout timeout_ticks)
{
    STK_ASSERT(data != nullptr);
 
    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(data, Slot(m_tail), m_element_size);
        m_tail = Next(m_tail);
        m_count--;
 
        // notify producer that space is now available
        m_cv_not_full.NotifyOne_CS();
    }
 
    return success;
}
 
// ---------------------------------------------------------------------------
// Pipe: DrainLocked  (private helper, caller must hold the critical section)
// ---------------------------------------------------------------------------
 
inline size_t Pipe::DrainLocked(uint8_t *const dst_bytes, const size_t count)
{
    const size_t to_read    = Min(count, m_count);
    const size_t first_part = m_capacity - m_tail;
 
    if (to_read <= first_part)
    {
        STK_MEMCPY(dst_bytes, Slot(m_tail), to_read * m_element_size);
    }
    else
    {
        STK_MEMCPY(dst_bytes,                                 Slot(m_tail), first_part             * m_element_size);
        STK_MEMCPY(dst_bytes + (first_part * m_element_size), Slot(0U),     (to_read - first_part) * m_element_size);
    }
 
    m_tail   = (m_tail + to_read) % m_capacity;
    m_count -= to_read;
 
    m_cv_not_full.NotifyAll_CS();
    return to_read;
}
 
// ---------------------------------------------------------------------------
// Pipe: ReadBulk
// ---------------------------------------------------------------------------
 
inline size_t Pipe::ReadBulk(void *dst, size_t count, Timeout timeout_ticks)
{
    size_t read_count = 0U;
 
    if ((dst != nullptr) && (count != 0U))
    {
        uint8_t *const dst_bytes = static_cast<uint8_t *>(dst);
        const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
        
        // capture an absolute deadline once, before entering the wait loop,
        // preventing the timeout from resetting on intermediate partial reads
        const Timeout deadline = (timed_wait ? 
            static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
        ScopedCriticalSection cs_;
 
        while (read_count < count)
        {
            bool is_timeout = false;
 
            while (m_count == 0U)
            {
                Timeout remaining = deadline;
                if (timed_wait)
                {
                    const Timeout now = static_cast<Timeout>(GetTicks());
                    remaining = (now >= deadline ? NO_WAIT : (deadline - now));
                }
 
                if (!m_cv_not_empty.Wait(cs_, remaining))
                {
                    is_timeout = true;
                    break; // break inner condition variable loop
                }
            }
 
            // if a timeout occurred, drop out of the chunk processing loop
            if (is_timeout)
            {
                break;
            }
 
            // drain data using the state tracker's relative byte offsets
            read_count += DrainLocked(dst_bytes + (read_count * m_element_size), count - read_count);
        }
    }
 
    return read_count;
}
 
// ---------------------------------------------------------------------------
// Pipe: ReadBulkTriggered
// ---------------------------------------------------------------------------
 
inline size_t Pipe::ReadBulkTriggered(void *dst, size_t trigger, size_t max_count, Timeout timeout_ticks)
{
    size_t read_count = 0U;
 
    if ((dst != nullptr) && (max_count != 0U))
    {
        // trigger must be in [1, max_count]
        if (trigger == 0U)       { trigger = 1U; }
        if (trigger > max_count) { trigger = max_count; }
 
        uint8_t *const dst_bytes = static_cast<uint8_t *>(dst);
        const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
        
        // capture an absolute deadline once, before entering the wait loop,
        // preventing the timeout from resetting on intermediate spurious wakeups
        const Timeout deadline = (timed_wait ? 
            static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
        ScopedCriticalSection cs_;
 
        while (m_count < trigger)
        {
            Timeout remaining = deadline;
            if (timed_wait)
            {
                const Timeout now = static_cast<Timeout>(GetTicks());
                remaining = (now >= deadline ? NO_WAIT : (deadline - now));
            }
 
            if (!m_cv_not_empty.Wait(cs_, remaining))
            {
                break; // break the waiting loop on timeout
            }
        }
 
        // whether we broke out via satisfying the trigger or hitting a timeout,
        // we drain whatever is currently available up to max_count
        read_count = DrainLocked(dst_bytes, max_count);
    }
 
    return read_count;
}
 
// ---------------------------------------------------------------------------
// Pipe: Reset
// ---------------------------------------------------------------------------
 
inline void Pipe::Reset()
{
    const ScopedCriticalSection cs_;
 
    m_count = 0U;
    m_head  = 0U;
    m_tail  = 0U;
 
    // wake all blocked producers: the pipe is now entirely empty, every slot is free
    // note: we do not release readers here
    m_cv_not_full.NotifyAll_CS();
}
 
// ---------------------------------------------------------------------------

template <typename T, size_t N>
class PipeT
{
public:
    explicit PipeT() : m_buffer(), m_head(0U), m_tail(0U), m_count(0U), m_cv_not_empty(), m_cv_not_full()
    {}

    bool Write(const T &data, Timeout timeout_ticks = WAIT_INFINITE)
    {
        ScopedCriticalSection cs_;
        bool success = true;
 
        while (m_count == N)
        {
            if (!m_cv_not_full.Wait(cs_, timeout_ticks))
            {
                success = false;
                break;
            }
        }
 
        if (success)
        {
            m_buffer[m_head] = data;
            m_head = (m_head + 1U) % N;
            m_count += 1U;
 
            // notify consumer that data is ready
            m_cv_not_empty.NotifyOne_CS();
        }
 
        return success;
    }

    bool TryWrite(const T &data) { return Write(data, NO_WAIT); }

    size_t WriteBulk(const T *src, size_t count, Timeout timeout_ticks = WAIT_INFINITE)
    {
        size_t written = 0U;
 
        if ((src != nullptr) && (count != 0U))
        {
            const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
            
            // capture an absolute deadline once, before entering the wait loop,
            // preventing the timeout from resetting on intermediate partial writes
            const Timeout deadline = (timed_wait ? 
                static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
            ScopedCriticalSection cs_;
 
            while (written < count)
            {
                bool is_timeout = false;
 
                while (m_count == N)
                {
                    Timeout remaining = deadline;
                    if (timed_wait)
                    {
                        const Timeout now = static_cast<Timeout>(GetTicks());
                        remaining = (now >= deadline ? NO_WAIT : (deadline - now));
                    }
 
                    if (!m_cv_not_full.Wait(cs_, remaining))
                    {
                        is_timeout = true; 
                        break; // break inner condition variable loop
                    }
                }
 
                // if timeout, drop out of the chunk processing loop
                if (is_timeout)
                {
                    break; 
                }
 
                // calculate how many we can copy in this contiguous stretch
                const size_t available = N - m_count;
                const size_t to_write  = ((count - written) < available) ? (count - written) : available;
 
                // copy from source
                // note: if value type is not scalar or queue is small we copy with a for loop,
                //       otherwise using faster memcpy version for large scalar arrays
                if (!std::is_scalar<T>::value || (N < 8U))
                {
                    for (size_t i = 0U; i < to_write; ++i)
                    {
                        m_buffer[m_head] = src[written++];
                        m_head           = (m_head + 1U) % N;
                        m_count         += 1U;
                    }
                }
                else
                {
                    const size_t first_part = N - m_head;
                    
                    if (to_write <= first_part)
                    {
                        STK_MEMCPY(&m_buffer[m_head], &src[written], to_write * sizeof(T));
                    }
                    else
                    {
                        STK_MEMCPY(&m_buffer[m_head], &src[written],              first_part              * sizeof(T));
                        STK_MEMCPY(&m_buffer[0U],     &src[written + first_part], (to_write - first_part) * sizeof(T));
                    }
                    
                    written += to_write;
                    m_head   = (m_head + to_write) % N;
                    m_count += to_write;
                }
 
                // notify consumers that data is ready
                m_cv_not_empty.NotifyAll_CS();
            }
        }
 
        return written;
    }

    size_t TryWriteBulk(const T *src, size_t count) { return WriteBulk(src, count, NO_WAIT); }

    bool Read(T &data, Timeout timeout_ticks = WAIT_INFINITE)
    {
        ScopedCriticalSection cs_;
        bool success = true;
 
        while (m_count == 0U)
        {
            if (!m_cv_not_empty.Wait(cs_, timeout_ticks))
            {
                success = false;
                break;
            }
        }
 
        if (success)
        {
            data     = m_buffer[m_tail];
            m_tail   = (m_tail + 1U) % N;
            m_count -= 1U;
 
            // notify producer that space is available
            m_cv_not_full.NotifyOne_CS();
        }
 
        return success;
    }
    
    bool TryRead(T &data) { return Read(data, NO_WAIT); }

    size_t ReadBulk(T *dst, size_t count, Timeout timeout_ticks = WAIT_INFINITE)
    {
        size_t read_count = 0U;
 
        if ((dst != nullptr) && (count != 0U))
        {
            const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
            
            // capture an absolute deadline once, before entering the wait loop,
            // this prevents the timeout from being silently restarted on each
            // spurious wakeup (e.g. a partial Set() that does not satisfy WAIT_ALL)
            const Timeout deadline = (timed_wait ? 
                static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
            ScopedCriticalSection cs_;
 
            while (read_count < count)
            {
                bool is_timeout = false;
              
                // wait until there is at least 1 element available
                while (m_count == 0U)
                {
                    Timeout remaining = deadline;
                    if (timed_wait)
                    {
                        const Timeout now = static_cast<Timeout>(GetTicks());
                        remaining = (now >= deadline ? NO_WAIT : (deadline - now));
                    }
 
                    if (!m_cv_not_empty.Wait(cs_, remaining))
                    {
                        is_timeout = true;
                        break; // break inner condition variable loop
                    }
                }
 
                // if a timeout, drop out of the chunk processing loop
                if (is_timeout)
                {
                    break;
                }
 
                // determine how many we can pull in this stretch
                const size_t to_read = (count - read_count) < m_count ? (count - read_count) : m_count;
 
                // note: if value type is not scalar or queue is small we copy with a for loop,
                //       otherwise using faster memcpy version for large scalar arrays
                if (!std::is_scalar<T>::value || (N < 8U))
                {
                    for (size_t i = 0U; i < to_read; ++i)
                    {
                        dst[read_count++] = m_buffer[m_tail];
                        m_tail            = (m_tail + 1U) % N;
                        m_count          -= 1U;
                    }
                }
                else
                {
                    const size_t first_part = N - m_tail;
                    
                    if (to_read <= first_part)
                    {
                        STK_MEMCPY(&dst[read_count], &m_buffer[m_tail], to_read * sizeof(T));
                    }
                    else
                    {
                        STK_MEMCPY(&dst[read_count],              &m_buffer[m_tail], first_part              * sizeof(T));
                        STK_MEMCPY(&dst[read_count + first_part], &m_buffer[0U],     (to_read - first_part)  * sizeof(T));
                    }
                    
                    read_count += to_read;
                    m_tail      = (m_tail + to_read) % N;
                    m_count    -= to_read;
                }
 
                // notify producers that space is now available
                m_cv_not_full.NotifyAll_CS();
            }
        }
 
        return read_count;
    }

    size_t TryReadBulk(T *dst, size_t count) { return ReadBulk(dst, count, NO_WAIT); }

    void Reset()
    {
        const ScopedCriticalSection cs_;
 
        m_count = 0U;
        m_head  = 0U;
        m_tail  = 0U;
 
        // wake all blocked producers: the pipe is now entirely empty, every slot is free
        // note: we do not release readers here
        m_cv_not_full.NotifyAll_CS();
    }

    size_t GetCapacity() const { return N; }

    size_t GetCount() const { return m_count; }

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

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

    bool IsFull() const { return (m_count == N); }
 
private:
    STK_NONCOPYABLE_CLASS(PipeT);
 
    T                 m_buffer[N];    
    size_t            m_head;         
    size_t            m_tail;         
    size_t            m_count;        
    ConditionVariable m_cv_not_empty; 
    ConditionVariable m_cv_not_full;  
};
 
} // namespace sync
} // namespace stk
 
#endif /* STK_SYNC_PIPE_H_ */