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.
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.
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.
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.
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.
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.

// 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.

Definition at line 71 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 330 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 487 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 ReadBulk(), and ReadBulkTriggered().

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 277 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 252 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 265 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 258 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 271 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 416 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 284 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 291 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 301 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 310 of file stk_sync_pipe.h.

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

References m_capacity.

Referenced by Read(), and Write().

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 if timeout expired before any data became available.

Definition at line 454 of file stk_sync_pipe.h.

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

References m_count, m_cv_not_empty, m_cv_not_full, m_element_size, m_tail, Next(), Slot(), STK_ASSERT, and STK_MEMCPY().

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

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

References DrainLocked(), stk::GetTicks(), m_count, m_cv_not_empty, m_element_size, stk::NO_WAIT, 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:

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).

Definition at line 567 of file stk_sync_pipe.h.

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

References DrainLocked(), stk::GetTicks(), m_count, m_cv_not_empty, stk::NO_WAIT, 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:

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 614 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 404 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(), and stk::memory::BlockMemoryPool::BlockMemoryPool().

Here is the caller graph for this function:

Slot()

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

inlineprivate

Definition at line 307 of file stk_sync_pipe.h.

{ return m_buffer + (idx * m_element_size); }

References m_buffer, and m_element_size.

Referenced by DrainLocked(), Read(), Write(), and WriteBulk().

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 166 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 199 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 234 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 112 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 145 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 if timeout expired before space became available.

Definition at line 348 of file stk_sync_pipe.h.

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

References CAPACITY_MAX, m_capacity, m_count, m_cv_not_empty, m_cv_not_full, m_element_size, m_head, Next(), Slot(), STK_ASSERT, and STK_MEMCPY().

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

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

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(), and stk::WAIT_INFINITE.

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:

Member Data Documentation

CAPACITY_MAX

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

static

Max capacity supported (number of elements).

Definition at line 76 of file stk_sync_pipe.h.

Referenced by Pipe(), and Write().

m_buffer

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

private

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

Definition at line 316 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 317 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetCapacity(), GetSpace(), IsFull(), Next(), Pipe(), Write(), and WriteBulk().

m_count

size_t stk::sync::Pipe::m_count

private

current number of elements stored in the pipe

Definition at line 319 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetCount(), GetSpace(), IsEmpty(), IsFull(), Pipe(), Read(), ReadBulk(), ReadBulkTriggered(), Reset(), Write(), and WriteBulk().

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

Referenced by Read(), ReadBulk(), ReadBulkTriggered(), Write(), and WriteBulk().

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

Referenced by DrainLocked(), Read(), Reset(), Write(), and WriteBulk().

m_element_size

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

private

size of each element in bytes

Definition at line 318 of file stk_sync_pipe.h.

Referenced by DrainLocked(), GetElementSize(), Pipe(), Read(), ReadBulk(), Slot(), Write(), and WriteBulk().

m_head

size_t stk::sync::Pipe::m_head

private

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

Definition at line 320 of file stk_sync_pipe.h.

Referenced by Pipe(), Reset(), Write(), and WriteBulk().

m_tail

size_t stk::sync::Pipe::m_tail

private

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

Definition at line 321 of file stk_sync_pipe.h.

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

---

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

• stk_sync_pipe.h