stk::sync::Pipe Class Reference

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

#include <stk_sync_pipe.h>

Inheritance diagram for stk::sync::Pipe:

Collaboration diagram for stk::sync::Pipe:

Public Member Functions
	Pipe (uint8_t *buf, size_t capacity, size_t element_size)
	Constructor.
	~Pipe ()=default
	Destructor.
EWaitResult	WriteEx (const void *data, Timeout timeout_ticks=WAIT_INFINITE)
	Write a single element to the pipe, preserving the full wait outcome.
bool	Write (const void *data, Timeout timeout_ticks=WAIT_INFINITE)
	Write a single element to the pipe.
bool	TryWrite (const void *data)
	Attempt to write a single element to the pipe without blocking.
EWaitResult	WriteBulkEx (const void *src, size_t count, size_t &written, Timeout timeout_ticks=WAIT_INFINITE)
	Write multiple elements to the pipe, preserving the full wait outcome.
size_t	WriteBulk (const void *src, size_t count, Timeout timeout_ticks=WAIT_INFINITE)
	Write multiple elements to the pipe.
size_t	TryWriteBulk (const void *src, size_t count)
	Attempt to write multiple elements to the pipe without blocking.
EWaitResult	ReadEx (void *data, Timeout timeout_ticks=WAIT_INFINITE)
	Read a single element from the pipe, preserving the full wait outcome.
bool	Read (void *data, Timeout timeout_ticks=WAIT_INFINITE)
	Read a single element from the pipe.
bool	TryRead (void *data)
	Attempt to read a single element from the pipe without blocking.
EWaitResult	ReadBulkEx (void *dst, size_t count, size_t &read_count, Timeout timeout_ticks=WAIT_INFINITE)
	Read multiple elements from the pipe, preserving the full wait outcome.
size_t	ReadBulk (void *dst, size_t count, Timeout timeout_ticks=WAIT_INFINITE)
	Read multiple elements from the pipe.
size_t	TryReadBulk (void *dst, size_t count)
	Attempt to read multiple elements from the pipe without blocking.
EWaitResult	ReadBulkTriggeredEx (void *dst, size_t trigger, size_t max_count, size_t &read_count, Timeout timeout_ticks=WAIT_INFINITE)
	Read at least trigger elements, then drain up to max_count without blocking, preserving the full wait outcome.
size_t	ReadBulkTriggered (void *dst, size_t trigger, size_t max_count, Timeout timeout_ticks=WAIT_INFINITE)
	Read at least trigger elements, then drain up to max_count without blocking.
size_t	TryReadBulkTriggered (void *dst, size_t max_count)
	Non-blocking variant of ReadBulkTriggered.
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	GetElementSize () const
	Get the size of each element in bytes.
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.
uint8_t *	GetBuffer ()
	Get a pointer to the raw backing buffer.
bool	IsEmpty () const
	Check if the pipe is currently empty.
bool	IsFull () const
	Check if the pipe is currently full.
bool	IsStorageValid () const
	Verify that the backing storage is valid and the pipe is ready for use.
void	SetTraceName (const char *name)
	Set name.
const char *	GetTraceName () const
	Get name.

Static Public Attributes
static const size_t	CAPACITY_MAX = 0xFFFEU
	Max capacity supported (number of elements).

Private Member Functions
	Pipe (const Pipe &)=delete
Pipe &	operator= (const Pipe &)=delete
uint8_t *	Slot (size_t idx) const
size_t	Next (size_t idx) const
size_t	DrainLocked (uint8_t *dst_bytes, size_t count)

Private Attributes
uint8_t *	m_buffer
	flat byte ring-buffer: capacity slots of element_size bytes each
const size_t	m_capacity
	maximum number of elements stored in the pipe
const size_t	m_element_size
	size of each element in bytes
size_t	m_count
	current number of elements stored in the pipe
size_t	m_head
	write index (next slot to be written by Write())
size_t	m_tail
	read index (next slot to be read by Read())
ConditionVariable	m_cv_not_empty
	signaled by Write() when the pipe transitions from empty
ConditionVariable	m_cv_not_full
	signaled by Read()/Reset() when the pipe is no longer full

Detailed Description

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

Pipe provides a synchronized ring-buffer mechanism that allows tasks to exchange data safely. It is parameterised on an element byte count (element_size) rather than a C++ type, making it suitable for passing heterogeneous or C-ABI structs without requiring the element type to be copyable via the C++ assignment operator. All element payloads are copied with memcpy.

It follows the following blocking semantics:

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

The caller is responsible for providing an appropriately sized external buffer and ensuring it remains valid for the entire lifetime of the Pipe object. For a self-contained, type-safe alternative with compile-time capacity and direct typed assignment, see stk::sync::PipeT.

Note

The boolean / count-returning calls (Write(), Read(), WriteBulk(), ReadBulk(), ReadBulkTriggered()) collapse a timeout and a cancellation (IKernel::CancelTaskWait()) into the same failure result. Use the *Ex() variants (WriteEx(), ReadEx(), WriteBulkEx(), ReadBulkEx(), ReadBulkTriggeredEx()) if the caller needs to tell them apart.

// Example: Producer-Consumer pattern with external buffer
struct Sample { uint32_t timestamp_ms; int16_t value; };
static uint8_t s_buf[8 * sizeof(Sample)];
stk::sync::Pipe g_DataPipe(s_buf, 8, sizeof(Sample));
 
void Task_Producer() {
    Sample s = { GetTicks(), ReadSensor() };
    // blocks if pipe is full
    g_DataPipe.Write(&s);
}
 
void Task_Consumer() {
    Sample received;
    // blocks until data is available, with a 1 s timeout
    if (g_DataPipe.Read(&received, 1000)) {
        // ... process received value ...
    }
}

See also

PipeT, MessageQueue, ConditionVariable

Note

Only available when kernel is compiled with KERNEL_SYNC mode enabled.

Examples

D:/SVN/private/stk/interop/c/include/stk_c.h.

Definition at line 77 of file stk_sync_pipe.h.

Constructor & Destructor Documentation

Pipe() [1/2]

stk::sync::Pipe::Pipe ( uint8_t * buf, size_t capacity, size_t element_size )

inlineexplicit

Constructor.

Parameters

[in]	buf	Pointer to externally-allocated storage. Must be at least capacity * element_size bytes.
[in]	capacity	Maximum number of elements [1, CAPACITY_MAX].
[in]	element_size	Size of each element in bytes (>= 1).

Definition at line 456 of file stk_sync_pipe.h.

: 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);
}

References CAPACITY_MAX, m_buffer, m_capacity, m_count, m_element_size, m_head, m_tail, and STK_ASSERT.

Referenced by operator=().

Here is the caller graph for this function:

~Pipe()

stk::sync::Pipe::~Pipe ( )

default

Destructor.

Note

If tasks are still waiting at destruction time it is considered a logical error (dangling waiters). An assertion is triggered in debug builds via the ConditionVariable destructors.

MISRA deviation: [STK-DEV-005] Rule 10-3-2.

References STK_VIRT_DTOR, and stk::WAIT_INFINITE.

Pipe() [2/2]

stk::sync::Pipe::Pipe ( const Pipe & )

privatedelete

Member Function Documentation

DrainLocked()

size_t stk::sync::Pipe::DrainLocked ( uint8_t * dst_bytes, size_t count )

inlineprivate

Definition at line 650 of file stk_sync_pipe.h.

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

References m_capacity, m_count, m_cv_not_full, m_element_size, m_tail, stk::Min(), Slot(), and STK_MEMCPY().

Referenced by ReadBulkEx(), and ReadBulkTriggeredEx().

Here is the call graph for this function:

Here is the caller graph for this function:

GetBuffer()

uint8_t * stk::sync::Pipe::GetBuffer ( )

inline

Get a pointer to the raw backing buffer.

Returns

Pointer to the beginning of the element buffer.

Note

ISR-safe.

Definition at line 403 of file stk_sync_pipe.h.

{ return m_buffer; }

References m_buffer.

GetCapacity()

size_t stk::sync::Pipe::GetCapacity ( ) const

inline

Get the maximum number of elements the pipe can hold.

Returns

Construction-time capacity.

Note

ISR-safe.

Definition at line 378 of file stk_sync_pipe.h.

{ return m_capacity; }

References m_capacity.

Referenced by stk_pipe_get_capacity(), and xStreamBufferSetTriggerLevel().

Here is the caller graph for this function:

GetCount()

size_t stk::sync::Pipe::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 391 of file stk_sync_pipe.h.

{ return m_count; }

References m_count.

Referenced by stk_pipe_get_count().

Here is the caller graph for this function:

GetElementSize()

size_t stk::sync::Pipe::GetElementSize ( ) const

inline

Get the size of each element in bytes.

Returns

Construction-time element size.

Note

ISR-safe.

Definition at line 384 of file stk_sync_pipe.h.

{ return m_element_size; }

References m_element_size.

Referenced by stk_pipe_get_element_size().

Here is the caller graph for this function:

GetSpace()

size_t stk::sync::Pipe::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 397 of file stk_sync_pipe.h.

{ return (m_capacity - m_count); }

References m_capacity, and m_count.

Referenced by stk_pipe_get_space().

Here is the caller graph for this function:

GetTraceName()

const char * stk::ITraceable::GetTraceName ( ) const

inlineinherited

Get name.

Returns

Name string, or NULL if not set or if STK_SYNC_DEBUG_NAMES is 0.

Definition at line 582 of file stk_common.h.

    {
    #if STK_SYNC_DEBUG_NAMES
        return m_trace_name;
    #else
        return nullptr;
    #endif
    }

IsEmpty()

bool stk::sync::Pipe::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 410 of file stk_sync_pipe.h.

{ return (m_count == 0U); }

References m_count.

Referenced by stk_pipe_is_empty().

Here is the caller graph for this function:

IsFull()

bool stk::sync::Pipe::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 417 of file stk_sync_pipe.h.

{ return (m_count == m_capacity); }

References m_capacity, and m_count.

Referenced by stk_pipe_is_full().

Here is the caller graph for this function:

IsStorageValid()

bool stk::sync::Pipe::IsStorageValid ( ) const

inline

Verify that the backing storage is valid and the pipe is ready for use.

Always true for pipes constructed with external storage. For heap-constructed pipes, false if operator new failed. Must be checked after the heap constructor when operating without exceptions (the typical embedded configuration).

Returns

true if the pipe is ready for use.

Note

ISR-safe.

Definition at line 427 of file stk_sync_pipe.h.

{ return (m_buffer != nullptr); }

References m_buffer.

Referenced by stk_pipe_is_storage_valid().

Here is the caller graph for this function:

Next()

size_t stk::sync::Pipe::Next ( size_t idx ) const

inlineprivate

Definition at line 436 of file stk_sync_pipe.h.

{ return (idx + 1U) % m_capacity; }

References m_capacity.

Referenced by ReadEx(), and WriteEx().

Here is the caller graph for this function:

operator=()

Pipe & stk::sync::Pipe::operator= ( const Pipe & )

privatedelete

References Pipe().

Here is the call graph for this function:

Read()

bool stk::sync::Pipe::Read ( void * data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read a single element from the pipe.

Copies element_size bytes from the oldest slot in the ring buffer into the buffer pointed to by data. If the pipe is empty, the calling task is suspended until data is produced or the timeout expires.

Parameters

[out]	data	Destination buffer for the retrieved element (must be at least element_size bytes).
[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 otherwise (timeout, cancellation or NO_WAIT on an empty pipe). Use ReadEx() to tell these apart.

Definition at line 221 of file stk_sync_pipe.h.

    {
        return (ReadEx(data, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

References ReadEx(), stk::WAIT_INFINITE, and stk::WAIT_RESULT_SIGNAL.

Referenced by stk_pipe_read(), and TryRead().

Here is the call graph for this function:

Here is the caller graph for this function:

ReadBulk()

size_t stk::sync::Pipe::ReadBulk ( void * dst, size_t count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read multiple elements from the pipe.

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

Parameters

[out]	dst	Pointer to the destination array (must hold at least count elements of element_size bytes each).
[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 or a cancellation occurred. Use ReadBulkEx() to tell these apart.

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

    {
        size_t read_count = 0U;
        const EWaitResult result = ReadBulkEx(dst, count, read_count, timeout_ticks);
        STK_UNUSED(result);
        return read_count;
    }

References ReadBulkEx(), STK_UNUSED, and stk::WAIT_INFINITE.

Referenced by stk_pipe_read_bulk(), and TryReadBulk().

Here is the call graph for this function:

Here is the caller graph for this function:

ReadBulkEx()

EWaitResult stk::sync::Pipe::ReadBulkEx ( void * dst, size_t count, size_t & read_count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read multiple elements from the pipe, preserving the full wait outcome.

Identical to ReadBulk(), except the caller gets the raw EWaitResult and the number of elements read through read_count, so a cancellation (via IKernel::CancelTaskWait()) can be distinguished from a timeout. Elements already copied into dst before a timeout or cancellation stay valid and are counted.

Parameters

[out]	dst	Pointer to the destination array (must hold at least count elements of element_size bytes each).
[in]	count	Number of elements to read.
[out]	read_count	Number of elements actually read (always set, 0 on WAIT_RESULT_FAIL).
[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

WAIT_RESULT_SIGNAL if all count elements were read (also for count = 0), WAIT_RESULT_TIMEOUT if the timeout expired (or NO_WAIT was passed) before all elements were available, WAIT_RESULT_CANCELED if the wait was cancelled via IKernel::CancelTaskWait(), WAIT_RESULT_FAIL if dst is nullptr or the kernel rejected the wait. On any non-signal result read_count holds the partial count.

Definition at line 676 of file stk_sync_pipe.h.

{
    read_count = 0U;
 
    // null destination is rejected, an empty request is trivially satisfied
    EWaitResult result = ((dst != nullptr) ? WAIT_RESULT_SIGNAL : WAIT_RESULT_FAIL);
 
    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)
        {
            // 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));
                }
 
                const EWaitResult wr = m_cv_not_empty.WaitEx(cs_, remaining);
                if (wr != WAIT_RESULT_SIGNAL)
                {
                    result = wr;
                    break; // break inner condition variable loop
                }
            }
 
            // if the wait failed (timeout, cancellation), drop out of the chunk processing loop
            if (result != WAIT_RESULT_SIGNAL)
            {
                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 result;
}

References DrainLocked(), stk::GetTicks(), m_count, m_cv_not_empty, m_element_size, stk::NO_WAIT, stk::WAIT_INFINITE, stk::WAIT_RESULT_FAIL, and stk::WAIT_RESULT_SIGNAL.

Referenced by ReadBulk(), and stk_pipe_read_bulk_ex().

Here is the call graph for this function:

Here is the caller graph for this function:

ReadBulkTriggered()

size_t stk::sync::Pipe::ReadBulkTriggered ( void * dst, size_t trigger, size_t max_count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read at least trigger elements, then drain up to max_count without blocking.

Blocks until at least trigger elements are simultaneously available in the pipe, or the timeout expires. Once the threshold is reached the call dequeues min(max_count, available) elements in a single critical-section pass — no busy-spin, no second call.

Parameters

[out]	dst	Destination buffer; must hold at least max_count elements of element_size bytes.
[in]	trigger	Minimum number of elements that must be available before any data is dequeued. Clamped to [1, max_count] internally.
[in]	max_count	Maximum number of elements to return in total.
[in]	timeout_ticks	Maximum time to wait for trigger elements. Default: WAIT_INFINITE.

Warning

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

Returns

Number of elements actually read.

• Equal to zero if timeout fired before a single element arrived.
• Greater than zero but less than trigger if a partial trigger's worth of data was written before timeout (this can only happen when trigger > 1 and the timeout expires after some — but not enough — elements arrive).
• Greater than or equal to trigger if the threshold was satisfied (the caller may then assume the trigger callback condition is met).

Note

A cancellation (IKernel::CancelTaskWait()) is reported the same way as a timeout. Use ReadBulkTriggeredEx() to tell these apart.

Definition at line 342 of file stk_sync_pipe.h.

    {
        size_t read_count = 0U;
        const EWaitResult result = ReadBulkTriggeredEx(dst, trigger, max_count, read_count, timeout_ticks);
        STK_UNUSED(result);
        return read_count;
    }

References ReadBulkTriggeredEx(), STK_UNUSED, and stk::WAIT_INFINITE.

Referenced by stk_pipe_read_bulk_triggered(), TryReadBulkTriggered(), and xStreamBufferReceive().

Here is the call graph for this function:

Here is the caller graph for this function:

ReadBulkTriggeredEx()

EWaitResult stk::sync::Pipe::ReadBulkTriggeredEx ( void * dst, size_t trigger, size_t max_count, size_t & read_count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read at least trigger elements, then drain up to max_count without blocking, preserving the full wait outcome.

Identical to ReadBulkTriggered(), except the caller gets the raw EWaitResult and the number of elements read through read_count, so a cancellation (via IKernel::CancelTaskWait()) can be distinguished from a timeout. Like the plain variant, whatever is available is drained even when the wait did not succeed.

Parameters

[out]	dst	Destination buffer; must hold at least max_count elements of element_size bytes.
[in]	trigger	Minimum number of elements that must be available before the wait is considered successful. Clamped to [1, max_count] internally.
[in]	max_count	Maximum number of elements to return in total.
[out]	read_count	Number of elements actually read (always set, 0 on WAIT_RESULT_FAIL).
[in]	timeout_ticks	Maximum time to wait for trigger elements. Default: WAIT_INFINITE.

Warning

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

Returns

WAIT_RESULT_SIGNAL if the trigger threshold was reached (also for max_count = 0), WAIT_RESULT_TIMEOUT if the timeout expired before the threshold was reached, WAIT_RESULT_CANCELED if the wait was cancelled via IKernel::CancelTaskWait(), WAIT_RESULT_FAIL if dst is nullptr or the kernel rejected the wait. On a timeout or cancellation read_count holds the (possibly zero) number of elements that were available and drained.

Definition at line 733 of file stk_sync_pipe.h.

{
    read_count = 0U;
 
    // null destination is rejected, an empty request is trivially satisfied
    EWaitResult result = ((dst != nullptr) ? WAIT_RESULT_SIGNAL : WAIT_RESULT_FAIL);
 
    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));
            }
 
            const EWaitResult wr = m_cv_not_empty.WaitEx(cs_, remaining);
            if (wr != WAIT_RESULT_SIGNAL)
            {
                result = wr;
                break; // break the waiting loop on timeout, cancellation or failure
            }
        }
 
        // whether we broke out via satisfying the trigger or a failed wait (timeout,
        // cancellation), we drain whatever is currently available up to max_count
        read_count = DrainLocked(dst_bytes, max_count);
    }
 
    return result;
}

References DrainLocked(), stk::GetTicks(), m_count, m_cv_not_empty, stk::NO_WAIT, stk::WAIT_INFINITE, stk::WAIT_RESULT_FAIL, and stk::WAIT_RESULT_SIGNAL.

Referenced by ReadBulkTriggered(), and stk_pipe_read_bulk_triggered_ex().

Here is the call graph for this function:

Here is the caller graph for this function:

ReadEx()

EWaitResult stk::sync::Pipe::ReadEx ( void * data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Read a single element from the pipe, preserving the full wait outcome.

Identical to Read(), except the caller gets the raw EWaitResult, so a cancellation (via IKernel::CancelTaskWait()) can be distinguished from a timeout.

Parameters

[out]	data	Destination buffer for the retrieved element (must be at least element_size bytes).
[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

WAIT_RESULT_SIGNAL if an element was read, WAIT_RESULT_TIMEOUT if no data became available before the timeout expired (or NO_WAIT was passed on an empty pipe), WAIT_RESULT_CANCELED if the wait was cancelled via IKernel::CancelTaskWait(), WAIT_RESULT_FAIL if the kernel rejected the wait. Nothing is read on a non-signal result.

Definition at line 600 of file stk_sync_pipe.h.

{
    STK_ASSERT(data != nullptr);
 
    ScopedCriticalSection cs_;
    EWaitResult result = WAIT_RESULT_SIGNAL;
 
    const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
 
    // capture an absolute deadline once, before entering the wait loop; a different
    // consumer can steal the just-produced element between NotifyOne_CS() and this
    // task re-acquiring the critical section (Mesa-style wakeup, not a direct handoff),
    // so re-passing the original timeout_ticks on each retry would silently restart it
    const Timeout deadline = (timed_wait ?
        static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
    while (m_count == 0U)
    {
        Timeout remaining = deadline;
        if (timed_wait)
        {
            const Timeout now = static_cast<Timeout>(GetTicks());
            remaining = (now >= deadline ? NO_WAIT : (deadline - now));
        }
 
        result = m_cv_not_empty.WaitEx(cs_, remaining);
        if (result != WAIT_RESULT_SIGNAL)
        {
            // timeout, cancellation or kernel failure: nothing is read
            break;
        }
    }
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        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 result;
}

References stk::GetTicks(), m_count, m_cv_not_empty, m_cv_not_full, m_element_size, m_tail, Next(), stk::NO_WAIT, Slot(), STK_ASSERT, STK_MEMCPY(), stk::WAIT_INFINITE, and stk::WAIT_RESULT_SIGNAL.

Referenced by Read(), and stk_pipe_read_ex().

Here is the call graph for this function:

Here is the caller graph for this function:

Reset()

void stk::sync::Pipe::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 786 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();
}

References m_count, m_cv_not_full, m_head, and m_tail.

Referenced by stk_pipe_reset().

Here is the caller graph for this function:

SetTraceName()

void stk::ITraceable::SetTraceName ( const char * name )

inlineinherited

Set name.

Parameters

[in]

name

Null-terminated string or NULL.

Note

If STK_SYNC_DEBUG_NAMES is 0 then calling this function has no effect.

Definition at line 570 of file stk_common.h.

    {
    #if STK_SYNC_DEBUG_NAMES
        m_trace_name = name;
    #else
        STK_UNUSED(name);
    #endif
    }

References STK_UNUSED.

Referenced by stk::memory::BlockMemoryPool::BlockMemoryPool(), stk::memory::BlockMemoryPool::BlockMemoryPool(), stk::memory::BytePool::BytePool(), stk::memory::BytePool::BytePool(), tx_block_pool_create(), tx_event_flags_create(), tx_mutex_create(), tx_queue_create(), and tx_semaphore_create().

Here is the caller graph for this function:

Slot()

uint8_t * stk::sync::Pipe::Slot ( size_t idx ) const

inlineprivate

Definition at line 433 of file stk_sync_pipe.h.

{ return m_buffer + (idx * m_element_size); }

References m_buffer, and m_element_size.

Referenced by DrainLocked(), ReadEx(), WriteBulkEx(), and WriteEx().

Here is the caller graph for this function:

TryRead()

bool stk::sync::Pipe::TryRead ( void * data )

inline

Attempt to read a single element 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

Destination buffer for the retrieved element.

Warning

ISR-safe.

Returns

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

Definition at line 233 of file stk_sync_pipe.h.

{ return Read(data, NO_WAIT); }

References stk::NO_WAIT, and Read().

Referenced by stk_pipe_tryread().

Here is the call graph for this function:

Here is the caller graph for this function:

TryReadBulk()

size_t stk::sync::Pipe::TryReadBulk ( void * 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 293 of file stk_sync_pipe.h.

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

References stk::NO_WAIT, and ReadBulk().

Referenced by stk_pipe_tryread_bulk(), and xStreamBufferReceiveFromISR().

Here is the call graph for this function:

Here is the caller graph for this function:

TryReadBulkTriggered()

size_t stk::sync::Pipe::TryReadBulkTriggered ( void * dst, size_t max_count )

inline

Non-blocking variant of ReadBulkTriggered.

Returns immediately with however many elements are available, up to max_count. The trigger threshold is not enforced (equivalent to trigger = 1 with NO_WAIT).

Parameters

[out]	dst	Destination buffer.
[in]	max_count	Maximum number of elements to read.

Warning

ISR-safe.

Returns

Number of elements actually read.

Definition at line 360 of file stk_sync_pipe.h.

    {
        return ReadBulkTriggered(dst, 1U, max_count, NO_WAIT);
    }

References stk::NO_WAIT, and ReadBulkTriggered().

Referenced by stk_pipe_tryread_bulk_triggered().

Here is the call graph for this function:

Here is the caller graph for this function:

TryWrite()

bool stk::sync::Pipe::TryWrite ( const void * data )

inline

Attempt to write a single element 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

Pointer to the element payload.

Warning

ISR-safe.

Returns

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

Definition at line 135 of file stk_sync_pipe.h.

{ return Write(data, NO_WAIT); }

References stk::NO_WAIT, and Write().

Referenced by stk_pipe_trywrite().

Here is the call graph for this function:

Here is the caller graph for this function:

TryWriteBulk()

size_t stk::sync::Pipe::TryWriteBulk ( const void * 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 195 of file stk_sync_pipe.h.

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

References stk::NO_WAIT, and WriteBulk().

Referenced by stk_pipe_trywrite_bulk(), and xStreamBufferSendFromISR().

Here is the call graph for this function:

Here is the caller graph for this function:

Write()

bool stk::sync::Pipe::Write ( const void * data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write a single element to the pipe.

Copies element_size bytes from data into the next available slot in the ring buffer. If the pipe is full, the calling task is suspended until space becomes available or the timeout expires.

Parameters

[in]	data	Pointer to the element payload (must be at least element_size bytes).
[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 otherwise (timeout, cancellation or NO_WAIT on a full pipe). Use WriteEx() to tell these apart.

Definition at line 123 of file stk_sync_pipe.h.

    {
        return (WriteEx(data, timeout_ticks) == WAIT_RESULT_SIGNAL);
    }

References stk::WAIT_INFINITE, stk::WAIT_RESULT_SIGNAL, and WriteEx().

Referenced by stk_pipe_write(), and TryWrite().

Here is the call graph for this function:

Here is the caller graph for this function:

WriteBulk()

size_t stk::sync::Pipe::WriteBulk ( const void * src, size_t count, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write multiple elements to the pipe.

Copies a block of count elements 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 (must hold at least count elements of element_size bytes each).
[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 or a cancellation occurred. Use WriteBulkEx() to tell these apart.

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

    {
        size_t written = 0U;
        const EWaitResult result = WriteBulkEx(src, count, written, timeout_ticks);
        STK_UNUSED(result);
        return written;
    }

References STK_UNUSED, stk::WAIT_INFINITE, and WriteBulkEx().

Referenced by stk_pipe_write_bulk(), TryWriteBulk(), and xStreamBufferSend().

Here is the call graph for this function:

Here is the caller graph for this function:

WriteBulkEx()

EWaitResult stk::sync::Pipe::WriteBulkEx ( const void * src, size_t count, size_t & written, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write multiple elements to the pipe, preserving the full wait outcome.

Identical to WriteBulk(), except the caller gets the raw EWaitResult and the number of elements written through written, so a cancellation (via IKernel::CancelTaskWait()) can be distinguished from a timeout. Elements already copied into the pipe before a timeout or cancellation stay in the pipe.

Parameters

[in]	src	Pointer to the source array (must hold at least count elements of element_size bytes each).
[in]	count	Number of elements to write.
[out]	written	Number of elements actually written (always set, 0 on WAIT_RESULT_FAIL).
[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

WAIT_RESULT_SIGNAL if all count elements were written (also for count = 0), WAIT_RESULT_TIMEOUT if the timeout expired (or NO_WAIT was passed) before all elements fitted, WAIT_RESULT_CANCELED if the wait was cancelled via IKernel::CancelTaskWait(), WAIT_RESULT_FAIL if src is nullptr or the kernel rejected the wait. On any non-signal result written holds the partial count.

Definition at line 525 of file stk_sync_pipe.h.

{
    written = 0U;
 
    // null source is rejected, an empty request is trivially satisfied
    EWaitResult result = ((src != nullptr) ? WAIT_RESULT_SIGNAL : WAIT_RESULT_FAIL);
 
    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)
        {
            // wait until there is at least 1 free slot
            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));
                }
 
                const EWaitResult wr = m_cv_not_full.WaitEx(cs_, remaining);
                if (wr != WAIT_RESULT_SIGNAL)
                {
                    result = wr;
                    break; // break inner condition variable loop
                }
            }
 
            // if the wait failed (timeout, cancellation), drop out of the chunk processing loop
            if (result != WAIT_RESULT_SIGNAL)
            {
                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 result;
}

References stk::GetTicks(), m_capacity, m_count, m_cv_not_empty, m_cv_not_full, m_element_size, m_head, stk::NO_WAIT, Slot(), STK_MEMCPY(), stk::WAIT_INFINITE, stk::WAIT_RESULT_FAIL, and stk::WAIT_RESULT_SIGNAL.

Referenced by stk_pipe_write_bulk_ex(), and WriteBulk().

Here is the call graph for this function:

Here is the caller graph for this function:

WriteEx()

EWaitResult stk::sync::Pipe::WriteEx ( const void * data, Timeout timeout_ticks = WAIT_INFINITE )

inline

Write a single element to the pipe, preserving the full wait outcome.

Identical to Write(), except the caller gets the raw EWaitResult, so a cancellation (via IKernel::CancelTaskWait()) can be distinguished from a timeout.

Parameters

[in]	data	Pointer to the element payload (must be at least element_size bytes).
[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

WAIT_RESULT_SIGNAL if data was written, WAIT_RESULT_TIMEOUT if no space became available before the timeout expired (or NO_WAIT was passed on a full pipe), WAIT_RESULT_CANCELED if the wait was cancelled via IKernel::CancelTaskWait(), WAIT_RESULT_FAIL if the kernel rejected the wait. Nothing is written on a non-signal result.

Definition at line 474 of file stk_sync_pipe.h.

{
    STK_ASSERT(data    != nullptr);
    STK_ASSERT(m_count <= (CAPACITY_MAX - 1U));
 
    ScopedCriticalSection cs_;
    EWaitResult result = WAIT_RESULT_SIGNAL;
 
    const bool timed_wait = (timeout_ticks != WAIT_INFINITE) && (timeout_ticks != NO_WAIT);
 
    // capture an absolute deadline once, before entering the wait loop; a different
    // producer can steal the just-freed slot between NotifyOne_CS() and this task
    // re-acquiring the critical section (Mesa-style wakeup, not a direct handoff),
    // so re-passing the original timeout_ticks on each retry would silently restart it
    const Timeout deadline = (timed_wait ?
        static_cast<Timeout>(GetTicks() + timeout_ticks) : timeout_ticks);
 
    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));
        }
 
        result = m_cv_not_full.WaitEx(cs_, remaining);
        if (result != WAIT_RESULT_SIGNAL)
        {
            // timeout, cancellation or kernel failure: nothing is written
            break;
        }
    }
 
    if (result == WAIT_RESULT_SIGNAL)
    {
        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 result;
}

References CAPACITY_MAX, stk::GetTicks(), m_capacity, m_count, m_cv_not_empty, m_cv_not_full, m_element_size, m_head, Next(), stk::NO_WAIT, Slot(), STK_ASSERT, STK_MEMCPY(), stk::WAIT_INFINITE, and stk::WAIT_RESULT_SIGNAL.

Referenced by stk_pipe_write_ex(), and Write().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

CAPACITY_MAX

const size_t stk::sync::Pipe::CAPACITY_MAX = 0xFFFEU

static

Max capacity supported (number of elements).

Definition at line 82 of file stk_sync_pipe.h.

Referenced by Pipe(), and WriteEx().

m_buffer

uint8_t* stk::sync::Pipe::m_buffer

private

flat byte ring-buffer: capacity slots of element_size bytes each

Definition at line 442 of file stk_sync_pipe.h.

Referenced by GetBuffer(), IsStorageValid(), Pipe(), and Slot().

m_capacity

const size_t stk::sync::Pipe::m_capacity

private

maximum number of elements stored in the pipe

Definition at line 443 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetCapacity(), GetSpace(), IsFull(), Next(), Pipe(), WriteBulkEx(), and WriteEx().

m_count

size_t stk::sync::Pipe::m_count

private

current number of elements stored in the pipe

Definition at line 445 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetCount(), GetSpace(), IsEmpty(), IsFull(), Pipe(), ReadBulkEx(), ReadBulkTriggeredEx(), ReadEx(), Reset(), WriteBulkEx(), and WriteEx().

m_cv_not_empty

ConditionVariable stk::sync::Pipe::m_cv_not_empty

private

signaled by Write() when the pipe transitions from empty

Definition at line 448 of file stk_sync_pipe.h.

Referenced by ReadBulkEx(), ReadBulkTriggeredEx(), ReadEx(), WriteBulkEx(), and WriteEx().

m_cv_not_full

ConditionVariable stk::sync::Pipe::m_cv_not_full

private

signaled by Read()/Reset() when the pipe is no longer full

Definition at line 449 of file stk_sync_pipe.h.

Referenced by DrainLocked(), ReadEx(), Reset(), WriteBulkEx(), and WriteEx().

m_element_size

const size_t stk::sync::Pipe::m_element_size

private

size of each element in bytes

Definition at line 444 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetElementSize(), Pipe(), ReadBulkEx(), ReadEx(), Slot(), WriteBulkEx(), and WriteEx().

m_head

size_t stk::sync::Pipe::m_head

private

write index (next slot to be written by Write())

Definition at line 446 of file stk_sync_pipe.h.

Referenced by Pipe(), Reset(), WriteBulkEx(), and WriteEx().

m_tail

size_t stk::sync::Pipe::m_tail

private

read index (next slot to be read by Read())

Definition at line 447 of file stk_sync_pipe.h.

Referenced by DrainLocked(), Pipe(), ReadEx(), and Reset().

---

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

• stk_sync_pipe.h