stk::sync::PipeT< T, N > Class Template Reference

Thread-safe, type-safe FIFO communication pipe with internal storage. More...

#include <stk_sync_pipe.h>

Inheritance diagram for stk::sync::PipeT< T, N >:

Collaboration diagram for stk::sync::PipeT< T, N >:

Public Member Functions
	PipeT ()
	Constructor.
bool	Write (const T &data, Timeout timeout_ticks=WAIT_INFINITE)
	Write data to the pipe.
bool	TryWrite (const T &data)
	Attempt to write data to the pipe without blocking.
size_t	WriteBulk (const T *src, size_t count, Timeout timeout_ticks=WAIT_INFINITE)
	Write multiple elements to the pipe.
size_t	TryWriteBulk (const T *src, size_t count)
	Attempt to write multiple elements to the pipe without blocking.
bool	Read (T &data, Timeout timeout_ticks=WAIT_INFINITE)
	Read data from the pipe.
bool	TryRead (T &data)
	Attempt to read data from the pipe without blocking.
size_t	ReadBulk (T *dst, size_t count, Timeout timeout_ticks=WAIT_INFINITE)
	Read multiple elements from the pipe.
size_t	TryReadBulk (T *dst, size_t count)
	Attempt to read multiple elements from the pipe without blocking.
void	Reset ()
	Discard all elements and reset the pipe to the empty state.
size_t	GetCapacity () const
	Get the maximum number of elements the pipe can hold.
size_t	GetCount () const
	Get the current number of elements in the pipe.
size_t	GetSpace () const
	Get the number of free slots currently available.
bool	IsEmpty () const
	Check if the pipe is currently empty.
bool	IsFull () const
	Check if the pipe is currently full.

Private Member Functions
	PipeT (const PipeT &)=delete
PipeT &	operator= (const PipeT &)=delete

Private Attributes
T	m_buffer [N]
	static storage for FIFO elements
size_t	m_head
	index of the next slot to be written (producer)
size_t	m_tail
	index of the next slot to be read (consumer)
size_t	m_count
	current number of elements stored in the pipe
ConditionVariable	m_cv_not_empty
	condition variable signaled when the pipe is no longer empty
ConditionVariable	m_cv_not_full
	condition variable signaled when the pipe is no longer full

Detailed Description

template<typename T, size_t N>
class stk::sync::PipeT< T, N >

Thread-safe, type-safe FIFO communication pipe with internal storage.

Template Parameters

T	Data type of elements.
N	Capacity of the pipe (number of elements).

PipeT provides a synchronized ring-buffer mechanism that allows tasks to exchange typed data safely. It implements blocking semantics:

• Write() blocks if the pipe is full until space becomes available.
• Read() blocks if the pipe is empty until data is produced.

Unlike stk::sync::Pipe, which operates on raw byte buffers and uses memcpy for all transfers, PipeT is parameterised on a concrete element type T. This enables:

• Direct typed assignment (no memcpy overhead for scalar types).
• Compile-time constant propagation for capacity N and element size, allowing the compiler to strength-reduce modulo operations and eliminate dead branches in bulk transfer paths.
• A fully type-safe call interface using T references and pointers.

// Example: Producer-Consumer pattern
stk::sync::PipeT<uint32_t, 8> g_DataPipe;
 
void Task_Producer() {
    uint32_t value = 42;
    // blocks if pipe is full
    g_DataPipe.Write(value);
}
 
void Task_Consumer() {
    uint32_t received;
    // blocks until data is available, with a 1 s timeout
    if (g_DataPipe.Read(received, 1000)) {
        // ... process received value ...
    }
}

See also

Pipe, MessageQueueT, ConditionVariable

Note

Only available when kernel is compiled with KERNEL_SYNC mode enabled.

Definition at line 671 of file stk_sync_pipe.h.

Constructor & Destructor Documentation

PipeT() [1/2]

template<typename T, size_t N>

stk::sync::PipeT< T, N >::PipeT ( )

inlineexplicit

Constructor.

Definition at line 676 of file stk_sync_pipe.h.

                     : m_buffer(), m_head(0U), m_tail(0U), m_count(0U), m_cv_not_empty(), m_cv_not_full()
    {}

PipeT() [2/2]

template<typename T, size_t N>

stk::sync::PipeT< T, N >::PipeT ( const PipeT< T, N > & )

privatedelete

Member Function Documentation

GetCapacity()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::GetCapacity ( ) const

inline

Get the maximum number of elements the pipe can hold.

Returns

Compile-time capacity N.

Note

ISR-safe.

Definition at line 1023 of file stk_sync_pipe.h.

{ return N; }

GetCount()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::GetCount ( ) const

inline

Get the current number of elements in the pipe.

Returns

Point-in-time snapshot of the element count. May be stale by the time the caller acts on it in a multi-task environment.

Note

ISR-safe on targets where a size_t-aligned read is atomic.

Definition at line 1030 of file stk_sync_pipe.h.

{ return m_count; }

GetSpace()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::GetSpace ( ) const

inline

Get the number of free slots currently available.

Returns

Point-in-time snapshot of the free-slot count.

Note

ISR-safe.

Definition at line 1036 of file stk_sync_pipe.h.

{ return (N - m_count); }

IsEmpty()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::IsEmpty ( ) const

inline

Check if the pipe is currently empty.

Returns

true if empty, otherwise false.

Note

The returned value is a point-in-time snapshot.

ISR-safe.

Definition at line 1043 of file stk_sync_pipe.h.

{ return (m_count == 0U); }

IsFull()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::IsFull ( ) const

inline

Check if the pipe is currently full.

Returns

true if full, otherwise false.

Note

The returned value is a point-in-time snapshot.

ISR-safe.

Definition at line 1050 of file stk_sync_pipe.h.

{ return (m_count == N); }

operator=()

template<typename T, size_t N>

PipeT & stk::sync::PipeT< T, N >::operator= ( const PipeT< T, N > & )

privatedelete

Read()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::Read ( T & data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read data from the pipe.

Attempts to retrieve an element from the FIFO queue. If pipe is empty, the calling task will be suspended until data is provided by a producer or the timeout expires.

Parameters

[out]	data	Reference to the variable where the retrieved data will be stored.
[in]	timeout_ticks	Maximum time to wait for data (ticks). Default: WAIT_INFINITE.

Warning

ISR-safe only with timeout_ticks=NO_WAIT, ISR-unsafe otherwise.

Returns

true if data was successfully read, false if timeout expired before any data became available.

Definition at line 849 of file stk_sync_pipe.h.

    {
        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;
    }

Referenced by stk::sync::PipeT< Timer *, 32U >::TryRead().

Here is the caller graph for this function:

ReadBulk()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::ReadBulk ( T * dst, size_t count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read multiple elements from the pipe.

Attempts to retrieve a block of data from the FIFO. If pipe does not contain enough elements to satisfy the requested count, it will block until the full amount is read or the timeout expires.

Parameters

[out]	dst	Pointer to the destination array.
[in]	count	Number of elements to read.
[in]	timeout_ticks	Maximum time to wait for data (ticks). Default: WAIT_INFINITE.

Warning

ISR-safe only with timeout_ticks=NO_WAIT, ISR-unsafe otherwise.

Returns

Number of elements actually read. Equal to count unless a timeout occurred.

// Example:
Sample frame[64];
 
size_t result = g_Pipe.ReadBulk(frame, 64, 500);
if (result == 64) {
    // process full frame
} else {
    // handle partial read / timeout
}

Definition at line 906 of file stk_sync_pipe.h.

    {
        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;
    }

Referenced by stk::sync::PipeT< Timer *, 32U >::TryReadBulk().

Here is the caller graph for this function:

Reset()

template<typename T, size_t N>

void stk::sync::PipeT< T, N >::Reset ( )

inline

Discard all elements and reset the pipe to the empty state.

Resets head, tail and count to zero. Any tasks blocked in Write() are woken so they can re-evaluate and enqueue into the now-empty pipe.

Warning

Elements that were in the pipe are silently discarded. Ensure no consumers depend on them before calling Reset().

ISR-safe.

Definition at line 1006 of file stk_sync_pipe.h.

    {
        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();
    }

TryRead()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::TryRead ( T & data )

inline

Attempt to read data from the pipe without blocking.

Dequeues an element only if one is immediately available. Returns false instantly if the pipe is empty.

Parameters

[out]

data

Reference to the variable where the retrieved data will be stored.

Warning

ISR-safe.

Returns

true if data was successfully read, false if the pipe was empty.

Definition at line 883 of file stk_sync_pipe.h.

{ return Read(data, NO_WAIT); }

TryReadBulk()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::TryReadBulk ( T * dst, size_t count )

inline

Attempt to read multiple elements from the pipe without blocking.

Reads as many elements as are currently available without blocking.

Parameters

[out]	dst	Pointer to the destination array.
[in]	count	Number of elements to read.

Warning

ISR-safe.

Returns

Number of elements actually read.

Definition at line 997 of file stk_sync_pipe.h.

{ return ReadBulk(dst, count, NO_WAIT); }

TryWrite()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::TryWrite ( const T & data )

inline

Attempt to write data to the pipe without blocking.

Enqueues the element only if a free slot is immediately available. Returns false instantly if the pipe is full.

Parameters

[in]

data

Reference to the data element to be copied into the pipe.

Warning

ISR-safe.

Returns

true if data was successfully written, false if no space is available.

Definition at line 723 of file stk_sync_pipe.h.

{ return Write(data, NO_WAIT); }

TryWriteBulk()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::TryWriteBulk ( const T * src, size_t count )

inline

Attempt to write multiple elements to the pipe without blocking.

Copies as many elements as possible without blocking. Elements that do not fit are discarded.

Parameters

[in]	src	Pointer to the source array.
[in]	count	Number of elements to write.

Warning

ISR-safe.

Returns

Number of elements actually written.

Definition at line 837 of file stk_sync_pipe.h.

{ return WriteBulk(src, count, NO_WAIT); }

Write()

template<typename T, size_t N>

bool stk::sync::PipeT< T, N >::Write ( const T & data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write data to the pipe.

Attempts to push an element into the FIFO queue. If pipe is full, the calling task will be suspended until space is made available by a consumer or the timeout expires.

Parameters

[in]	data	Reference to the data element to be copied into the pipe.
[in]	timeout_ticks	Maximum time to wait for space (ticks). Default: WAIT_INFINITE.

Warning

ISR-safe only with timeout_ticks=NO_WAIT, ISR-unsafe otherwise.

Returns

true if data was successfully written, false if timeout expired before space became available.

Definition at line 689 of file stk_sync_pipe.h.

    {
        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;
    }

Referenced by stk::sync::PipeT< Timer *, 32U >::TryWrite().

Here is the caller graph for this function:

WriteBulk()

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::WriteBulk ( const T * src, size_t count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write multiple elements to the pipe.

Copies a block of data into the FIFO. If the pipe does not have enough space for the entire block, it will block until the full amount can be written or the timeout expires.

Parameters

[in]	src	Pointer to the source array.
[in]	count	Number of elements to write.
[in]	timeout_ticks	Maximum time to wait for sufficient space (ticks). Default: WAIT_INFINITE.

Warning

ISR-safe only with timeout_ticks=NO_WAIT, ISR-unsafe otherwise.

Returns

Number of elements actually written. Equal to count unless a timeout occurred.

// Example:
Sample frame[64];
FillFrame(frame);
 
size_t result = g_Pipe.WriteBulk(frame, 64, 500);
if (result < 64) {
    // handle partial write / timeout
}

Definition at line 745 of file stk_sync_pipe.h.

    {
        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;
    }

Referenced by stk::sync::PipeT< Timer *, 32U >::TryWriteBulk().

Here is the caller graph for this function:

Member Data Documentation

m_buffer

template<typename T, size_t N>

T stk::sync::PipeT< T, N >::m_buffer[N]

private

static storage for FIFO elements

Definition at line 1055 of file stk_sync_pipe.h.

m_count

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::m_count

private

current number of elements stored in the pipe

Definition at line 1058 of file stk_sync_pipe.h.

m_cv_not_empty

template<typename T, size_t N>

ConditionVariable stk::sync::PipeT< T, N >::m_cv_not_empty

private

condition variable signaled when the pipe is no longer empty

Definition at line 1059 of file stk_sync_pipe.h.

m_cv_not_full

template<typename T, size_t N>

ConditionVariable stk::sync::PipeT< T, N >::m_cv_not_full

private

condition variable signaled when the pipe is no longer full

Definition at line 1060 of file stk_sync_pipe.h.

m_head

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::m_head

private

index of the next slot to be written (producer)

Definition at line 1056 of file stk_sync_pipe.h.

m_tail

template<typename T, size_t N>

size_t stk::sync::PipeT< T, N >::m_tail

private

index of the next slot to be read (consumer)

Definition at line 1057 of file stk_sync_pipe.h.

---

The documentation for this class was generated from the following file:

• stk_sync_pipe.h