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

Thread-safe FIFO communication pipe for inter-task data passing. More...

#include <stk_sync_pipe.h>

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

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

Public Member Functions
	Pipe ()
	Constructor.
bool	Write (const T &data, Timeout timeout=WAIT_INFINITE)
	Write data to the pipe.
bool	TryWrite (const T &data)
	Attempt to write data to the pipe.
size_t	WriteBulk (const T *src, size_t count, Timeout timeout=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.
bool	Read (T &data, Timeout timeout=WAIT_INFINITE)
	Read data from the pipe.
bool	TryRead (T &data)
	Attempt to read data from the pipe.
size_t	ReadBulk (T *dst, size_t count, Timeout timeout=WAIT_INFINITE)
	Read multiple elements from the pipe.
size_t	TryReadBulk (T *dst, size_t count)
	Attempt to read multiple elements from the pipe.
size_t	GetSize () const
	Get the current number of elements in the pipe.
bool	IsEmpty () const
	Check if queue is currently empty.

Private Member Functions
	STK_NONCOPYABLE_CLASS (Pipe)

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_empty
	condition variable signaled when the pipe is no longer empty
ConditionVariable	m_cv_full
	condition variable signaled when the pipe is no longer full

Detailed Description

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

Thread-safe FIFO communication pipe for inter-task data passing.

Template Parameters

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

Pipe provides a synchronized ring-buffer mechanism that allows tasks to exchange 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.

// Example: Producer-Consumer pattern
stk::sync::Pipe<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 1s timeout
    if (g_DataPipe.Read(received, 1000)) {
        // ... process received value ...
    }
}

See also

Mutex, ConditionVariable

Note

Only available when kernel is compiled with KERNEL_SYNC mode enabled.

Definition at line 58 of file stk_sync_pipe.h.

Constructor & Destructor Documentation

Pipe()

template<typename T, size_t N>

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

inlineexplicit

Constructor.

Definition at line 63 of file stk_sync_pipe.h.

                    : m_buffer(), m_head(0U), m_tail(0U), m_count(0U), m_cv_empty(), m_cv_full()
    {}

Member Function Documentation

GetSize()

template<typename T, size_t N>

size_t stk::sync::Pipe< T, N >::GetSize ( ) const

inline

Get the current number of elements in the pipe.

Returns

The number of elements currently stored in the FIFO buffer.

Note

The returned value is a point-in-time snapshot. In a multi-tasking environment, queue size may change immediately after function returns if a producer or consumer is active in another task.

Definition at line 337 of file stk_sync_pipe.h.

{ return m_count; }

Referenced by stk::sync::Pipe< Timer *, 32 >::IsEmpty(), and stk_pipe_get_size().

Here is the caller graph for this function:

IsEmpty()

template<typename T, size_t N>

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

inline

Check if queue is currently empty.

Returns

True if empty, otherwise false.

Note

The returned value is a point-in-time snapshot. In a multi-tasking environment, queue size may change immediately after function returns if a producer or consumer is active in another task.

Definition at line 346 of file stk_sync_pipe.h.

{ return (GetSize() == 0); }

Read()

template<typename T, size_t N>

bool stk::sync::Pipe< T, N >::Read ( T & data, Timeout timeout = 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	Maximum time to wait for data (ticks). Default: WAIT_INFINITE.

Warning

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

Returns

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

Definition at line 207 of file stk_sync_pipe.h.

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

Referenced by stk_pipe_read(), and stk::sync::Pipe< Timer *, 32 >::TryRead().

Here is the caller graph for this function:

ReadBulk()

template<typename T, size_t N>

size_t stk::sync::Pipe< T, N >::ReadBulk ( T * dst, size_t count, Timeout timeout = 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	Maximum time to wait for data (ticks). Default: WAIT_INFINITE.

Warning

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

Returns

Number of elements actually read. This will be equal to count unless a timeout occurred.

// Example:
Sample frame[64];
 
// read the whole block
size_t result = g_Pipe.ReadBulk(frame, 64, 500);
if (result == 64) {
    // handle data processing
} else {
    // handle partial data due to underrun/timeout
}

Definition at line 260 of file stk_sync_pipe.h.

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

Referenced by stk_pipe_read_bulk(), and stk::sync::Pipe< Timer *, 32 >::TryReadBulk().

Here is the caller graph for this function:

STK_NONCOPYABLE_CLASS()

template<typename T, size_t N>

stk::sync::Pipe< T, N >::STK_NONCOPYABLE_CLASS ( Pipe< T, N > )

private

TryRead()

template<typename T, size_t N>

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

inline

Attempt to read data from the pipe.

Checks if an element is available in the FIFO and retrieves it immediately without blocking. If the pipe is empty, returns false without yielding the CPU or touching the kernel wait list.

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 235 of file stk_sync_pipe.h.

{ return Read(data, NO_WAIT); }

TryReadBulk()

template<typename T, size_t N>

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

inline

Attempt to read multiple elements from the pipe.

Reads as many elements as are currently available in the FIFO without blocking. If fewer elements than requested are available, only the available elements are read.

Parameters

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

Warning

ISR-safe.

Returns

Number of elements actually read. This will be equal to count if all requested elements were available, or less if the pipe was empty or became empty before all elements could be read.

Definition at line 328 of file stk_sync_pipe.h.

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

TryWrite()

template<typename T, size_t N>

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

inline

Attempt to write data to the pipe.

Attempts to push an element into the FIFO queue. If pipe is full, a false will be returned without waiting for a free space.

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 103 of file stk_sync_pipe.h.

{ return Write(data, NO_WAIT); }

TryWriteBulk()

template<typename T, size_t N>

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

inline

Attempt to write multiple elements to the pipe.

Copies as many elements as possible into the FIFO without blocking. If the pipe does not have enough space for the entire block, only the elements that fit are written and the rest 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. This will be equal to count if all elements were written successfully, or less if the pipe was full or became full before all elements could be written.

Definition at line 195 of file stk_sync_pipe.h.

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

Write()

template<typename T, size_t N>

bool stk::sync::Pipe< T, N >::Write ( const T & data, Timeout timeout = 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	Maximum time to wait for space (ticks). Default: WAIT_INFINITE.

Warning

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

Returns

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

Definition at line 76 of file stk_sync_pipe.h.

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

Referenced by stk_pipe_write(), and stk::sync::Pipe< Timer *, 32 >::TryWrite().

Here is the caller graph for this function:

WriteBulk()

template<typename T, size_t N>

size_t stk::sync::Pipe< T, N >::WriteBulk ( const T * src, size_t count, Timeout timeout = 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	Maximum time to wait for sufficient space (ticks).

Warning

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

Returns

Number of elements actually written. This will be equal to count unless a timeout occurred.

// Example:
Sample frame[64];
FillFrame(frame); // e.g., from decoder
 
// move the whole block
size_t result = g_Pipe.WriteBulk(frame, 64, 500);
if (result < 64) {
    // handle buffer underrun/timeout
}

Definition at line 127 of file stk_sync_pipe.h.

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

Referenced by stk_pipe_write_bulk(), and stk::sync::Pipe< Timer *, 32 >::TryWriteBulk().

Here is the caller graph for this function:

Member Data Documentation

m_buffer

template<typename T, size_t N>

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

private

static storage for FIFO elements

Definition at line 351 of file stk_sync_pipe.h.

m_count

template<typename T, size_t N>

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

private

current number of elements stored in the pipe

Definition at line 354 of file stk_sync_pipe.h.

m_cv_empty

template<typename T, size_t N>

ConditionVariable stk::sync::Pipe< T, N >::m_cv_empty

private

condition variable signaled when the pipe is no longer empty

Definition at line 355 of file stk_sync_pipe.h.

m_cv_full

template<typename T, size_t N>

ConditionVariable stk::sync::Pipe< T, N >::m_cv_full

private

condition variable signaled when the pipe is no longer full

Definition at line 356 of file stk_sync_pipe.h.

m_head

template<typename T, size_t N>

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

private

index of the next slot to be written (producer)

Definition at line 352 of file stk_sync_pipe.h.

m_tail

template<typename T, size_t N>

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

private

index of the next slot to be read (consumer)

Definition at line 353 of file stk_sync_pipe.h.

---

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

• stk_sync_pipe.h