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>
#include <cstring> // for memcpy
 
#include "stk_sync_cv.h"

 
namespace stk {
namespace sync {

template <typename T, size_t N>
class Pipe
{
public:
    explicit Pipe() : m_buffer(), m_head(0U), m_tail(0U), m_count(0U), m_cv_empty(), m_cv_full()
    {}

    bool Write(const T &data, Timeout timeout = WAIT_INFINITE)
    {
        ScopedCriticalSection cs_;
 
        while (m_count == N)
        {
            if (!m_cv_full.Wait(cs_, timeout))
                return false;
        }
 
        m_buffer[m_head] = data;
        m_head = (m_head + 1U) % N;
        m_count += 1U;
 
        // notify consumer that data is available
        m_cv_empty.NotifyOne();
 
        return true;
    }

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

    size_t WriteBulk(const T *src, size_t count, Timeout timeout = WAIT_INFINITE)
    {
        if ((src == nullptr) || (count == 0U))
            return 0U;
 
        size_t written = 0U;
        ScopedCriticalSection cs_;
 
        while (written < count)
        {
            // wait until there is at least 1 slot available
            while (m_count == N)
            {
                if (!m_cv_full.Wait(cs_, timeout))
                    return written; // Return partial count on timeout
            }
 
            // calculate how many we can copy in this contiguous stretch
            size_t available = N - m_count;
            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 < 8))
            {
                for (size_t i = 0; i < to_write; ++i)
                {
                    m_buffer[m_head] = src[written++];
                    m_head = (m_head + 1U) % N;
                    m_count += 1U;
                }
            }
            else
            {
                size_t first_part = N - m_head;
                if (to_write <= first_part)
                {
                    memcpy(&m_buffer[m_head], &src[written], (to_write * sizeof(T)));
                }
                else
                {
                    memcpy(&m_buffer[m_head], &src[written], (first_part * sizeof(T)));
                    memcpy(&m_buffer[0], &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_empty.NotifyAll();
        }
 
        return written;
    }

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

    bool Read(T &data, Timeout timeout = WAIT_INFINITE)
    {
        ScopedCriticalSection cs_;
 
        while (m_count == 0U)
        {
            if (!m_cv_empty.Wait(cs_, timeout))
                return false;
        }
 
        data = m_buffer[m_tail];
        m_tail = (m_tail + 1U) % N;
        m_count -= 1U;
 
        // notify producer that space is available
        m_cv_full.NotifyOne();
 
        return true;
    }

    bool TryRead(T &data) { return Read(data, NO_WAIT); }

    size_t ReadBulk(T *dst, size_t count, Timeout timeout = WAIT_INFINITE)
    {
        if ((dst == nullptr) || (count == 0U))
            return 0U;
 
        size_t read_count = 0U;
 
        ScopedCriticalSection cs_;
 
        while (read_count < count)
        {
            // wait until there is at least 1 element available
            while (m_count == 0U)
            {
                if (!m_cv_empty.Wait(cs_, timeout))
                    return read_count; // return partial count on timeout
            }
 
            // determine how many we can pull in this stretch
            size_t to_read = (count - read_count) < m_count ? (count - read_count) : m_count;
 
            // copy to destination
            // 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 < 8))
            {
                for (size_t i = 0U; i < to_read; ++i)
                {
                    dst[read_count++] = m_buffer[m_tail];
                    m_tail = (m_tail + 1) % N;
                    m_count -= 1U;
                }
            }
            else
            {
                size_t first_part = N - m_tail;
                if (to_read <= first_part)
                {
                    memcpy(&dst[read_count], &m_buffer[m_tail], (to_read * sizeof(T)));
                }
                else
                {
                    memcpy(&dst[read_count], &m_buffer[m_tail], (first_part * sizeof(T)));
                    memcpy(&dst[read_count + first_part], &m_buffer[0], ((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_full.NotifyAll();
        }
 
        return read_count;
    }

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

    size_t GetSize() const { return m_count; }

    bool IsEmpty() const { return (GetSize() == 0); }
 
private:
    STK_NONCOPYABLE_CLASS(Pipe);
 
    T                 m_buffer[N]; 
    size_t            m_head;      
    size_t            m_tail;      
    size_t            m_count;     
    ConditionVariable m_cv_empty;  
    ConditionVariable m_cv_full;   
};
 
} // namespace sync
} // namespace stk
 
#endif /* STK_SYNC_PIPE_H_ */