Set of Hopper buckets keyed by an integer. More...

#include <StormByte/buffer/sink.hxx>

Classes
class	Lane
	One-key redirect: `from.To(key)` >> dest. More...

Public Types
using	Select = std::function< std::size_t(std::size_t)>
	Index chooser for Pop selection across multiple buckets.

Public Member Functions
Lifecycle
	Sink () noexcept
	Constructs an empty Sink with zero buckets.

	Sink (const Sink &)=delete
	Copy constructor is deleted (Sink is non-copyable).

	Sink (Sink &&) noexcept=delete
	Move constructor is deleted (Sink is non-movable).

	~Sink () noexcept
	Destructor.

Sink &	operator= (const Sink &)=delete
	Copy assignment operator is deleted (Sink is non-copyable).

Sink &	operator= (Sink &&) noexcept=delete
	Move assignment operator is deleted (Sink is non-movable).

Producer
void	Push (int key, T item) noexcept
	Pushes an item into the bucket specified by key.

void	Eof () noexcept
	Closes the Sink and releases writer holds on its hoppers.

Wiring & Connection
Lane	To (int key) noexcept
	Redirect of one hopper key.

Sink &	operator>> (Sink &dest) noexcept
	Share every existing hopper with `dest`.

Sink &	operator<< (Sink &src) noexcept
	Same as `src` >> *this (all hoppers).

Sink &	operator<< (Lane lane) noexcept
	Same as `lane` >> *this.

void	Bind (Sink &consumer)
	Shares all existing hoppers on this Sink with the consumer Sink.

void	Bind (int key, Sink &consumer)
	Creates or retrieves the hopper for key and shares it with consumer.

void	Drain () noexcept
	Marks Sink as a terminal producer (un-wired Push calls drop instead of waiting).

bool	Draining () const noexcept
	Checks whether Drain was called.

void	Notify (std::condition_variable &consumer) noexcept
	Registers consumer condition variable to notify on all current and future hoppers.

void	Unnotify () noexcept
	Clears Notify on this Sink and on every hopper.

Bucket Capacity & Query
std::size_t	Capacity (int key) const noexcept
	Gets capacity ceiling of bucket key.

void	Capacity (int key, std::size_t capacity) noexcept
	Sets capacity ceiling of bucket key.

std::size_t	Size (int key) const noexcept
	Gets pending item count in bucket key.

bool	Full (int key) const noexcept
	Checks if bucket key is full.

Consumer
T	Pop () noexcept
	Pops one item from the Sink using default round-robin order.

T	Pop (const Select &select) noexcept
	Pops one item from the Sink using a custom selection function.

bool	EoF () const noexcept
	Checks if the Sink is finished.

bool	Ready () const noexcept
	Checks whether a Pop call can return immediately (item available or EoF).

Detailed Description

template<Type::MoveConstructible T>
class StormByte::Buffer::Sink< T >

Set of Hopper buckets keyed by an integer.

Sink manages a collection of Hopper queues keyed by an integer (representing channels, tracks, sessions, etc.). Sink does not interpret the integer key.

Key characteristics:

Wiring: producer >> consumer shares every existing hopper. producer.To(key) >> consumer creates or shares one key. When this Sink already holds that hopper, dest is a co-writer: Eof on this Sink then only closes the hopper when the last writer closes. consumer << producer and consumer << producer.To(key) are the same edges. Bind remains for one or two releases and is deprecated.
Drain: Terminal producer flag. Push to an un-wired key discards the item without waiting for a consumer.
Pop: Retrieves items across buckets using Round-Robin or custom Select.
EoF: Closes the Sink and CloseWriter on hoppers this Sink writes.

Template Parameters

T	Item type stored in the hoppers (must be MoveConstructible).

Member Typedef Documentation

◆ Select

template<Type::MoveConstructible T>

using StormByte::Buffer::Sink< T >::Select = std::function<std::size_t(std::size_t)>

Index chooser for Pop selection across multiple buckets.

Given the count of active buckets, returns an index in range [0, count).

Constructor & Destructor Documentation

◆ Sink() [1/3]

template<Type::MoveConstructible T>

StormByte::Buffer::Sink< T >::Sink ( )

noexcept

Constructs an empty Sink with zero buckets.

◆ Sink() [2/3]

template<Type::MoveConstructible T>

StormByte::Buffer::Sink< T >::Sink ( const Sink< T > & )

delete

Copy constructor is deleted (Sink is non-copyable).

◆ Sink() [3/3]

template<Type::MoveConstructible T>

StormByte::Buffer::Sink< T >::Sink ( Sink< T > && )

deletenoexcept

Move constructor is deleted (Sink is non-movable).

◆ ~Sink()

template<Type::MoveConstructible T>

StormByte::Buffer::Sink< T >::~Sink ( )

noexcept

Destructor.

Wakes any threads blocked on wiring.

Member Function Documentation

◆ Bind() [1/2]

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Bind	(	int	key,
		Sink< T > &	consumer
	)

Creates or retrieves the hopper for key and shares it with consumer.

Parameters

key	Bucket key identifier.
consumer	Target consumer Sink.

Deprecated:: Use To(key) >> consumer. Kept for one or two releases.

This Sink has no hopper yet: creates it (this is the writer, consumer is the reader). This Sink already has the hopper: consumer is attached as a co-writer (extra producer).

◆ Bind() [2/2]

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Bind ( Sink< T > & consumer )

Shares all existing hoppers on this Sink with the consumer Sink.

Parameters

consumer Target consumer Sink.

Deprecated:: Use *this >> consumer. Kept for one or two releases.

◆ Capacity() [1/2]

template<Type::MoveConstructible T>

std::size_t StormByte::Buffer::Sink< T >::Capacity ( int key ) const

noexcept

Gets capacity ceiling of bucket key.

Parameters

key	Bucket key identifier.

Returns: Hopper capacity, or 0 if key bucket does not exist.

◆ Capacity() [2/2]

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Capacity	(	int	key,
		std::size_t	capacity
	)

noexcept

Sets capacity ceiling of bucket key.

Parameters

key	Bucket key identifier.
capacity	Maximum items allowed (0 = unbounded).

◆ Drain()

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Drain ( )

noexcept

Marks Sink as a terminal producer (un-wired Push calls drop instead of waiting).

◆ Draining()

template<Type::MoveConstructible T>

bool StormByte::Buffer::Sink< T >::Draining ( ) const

noexcept

Checks whether Drain was called.

Returns: true if Drain was set.

◆ EoF()

template<Type::MoveConstructible T>

bool StormByte::Buffer::Sink< T >::EoF ( ) const

noexcept

Checks if the Sink is finished.

Returns: true if no items remain and no new items can arrive in the current hoppers.

Contract details for EoF():

With zero hoppers: EoF() returns true only if this Sink is closed (via Eof() or destruction).
With hoppers: EoF() returns true when all hoppers are empty and Hopper::EoF() is true, even if this Sink itself did not call Eof() (because wiring shares the Hopper and the producer closed it from the other Sink).
Wiring a new key after EoF() returned true may cause EoF() to return false again if new work becomes available.
EoF() does not mean "this object called Eof()", but "no items remain and none can enter current buckets".

◆ Eof()

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Eof ( )

noexcept

Closes the Sink and releases writer holds on its hoppers.

Hoppers this Sink created (or attached to as a co-writer) decrement their writer count. A hopper shared by several producers Eofs when the last writer closes. Idempotent.

Wakes Push and Pop waiters.

◆ Full()

template<Type::MoveConstructible T>

bool StormByte::Buffer::Sink< T >::Full ( int key ) const

noexcept

Checks if bucket key is full.

Parameters

key	Bucket key identifier.

Returns: true if full, or false if key does not exist.

◆ Notify()

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Notify ( std::condition_variable & consumer )

noexcept

Registers consumer condition variable to notify on all current and future hoppers.

Parameters

consumer Consumer condition variable reference. Not owned.

◆ operator<<() [1/2]

template<Type::MoveConstructible T>

Sink & StormByte::Buffer::Sink< T >::operator<< ( Lane lane )

noexcept

Same as lane >> *this.

Parameters

lane	To result from the producer.

Returns: *this.

◆ operator<<() [2/2]

template<Type::MoveConstructible T>

Sink & StormByte::Buffer::Sink< T >::operator<< ( Sink< T > & src )

noexcept

Same as src >> *this (all hoppers).

Parameters

src	Producer Sink.

Returns: *this.

◆ operator=() [1/2]

template<Type::MoveConstructible T>

Sink & StormByte::Buffer::Sink< T >::operator= ( const Sink< T > & )

delete

Copy assignment operator is deleted (Sink is non-copyable).

◆ operator=() [2/2]

template<Type::MoveConstructible T>

Sink & StormByte::Buffer::Sink< T >::operator= ( Sink< T > && )

deletenoexcept

Move assignment operator is deleted (Sink is non-movable).

◆ operator>>()

template<Type::MoveConstructible T>

Sink & StormByte::Buffer::Sink< T >::operator>> ( Sink< T > & dest )

noexcept

Share every existing hopper with dest.

Parameters

dest	Consumer Sink.

Returns: dest.

Zero buckets: no-op. Does not create keys.

◆ Pop() [1/2]

template<Type::MoveConstructible T>

T StormByte::Buffer::Sink< T >::Pop ( )

noexcept

Pops one item from the Sink using default round-robin order.

Returns: Next item, or default T if empty/closed.

Blocks while zero buckets exist and Sink is not closed.

◆ Pop() [2/2]

template<Type::MoveConstructible T>

T StormByte::Buffer::Sink< T >::Pop ( const Select & select )

noexcept

Pops one item from the Sink using a custom selection function.

Parameters

select Bucket index selection callback.

Returns: Next item, or default T if empty/closed.

Blocks while zero buckets exist and Sink is not closed.

◆ Push()

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Push	(	int	key,
		T	item
	)

noexcept

Pushes an item into the bucket specified by key.

Parameters

key	Integer bucket identifier.
item	Item to enqueue. Empty smart pointers are discarded.

Waits until the bucket exists, the Sink is closed, or Drain is set. If closed or draining and no bucket exists for key, item is discarded.

◆ Ready()

template<Type::MoveConstructible T>

bool StormByte::Buffer::Sink< T >::Ready ( ) const

noexcept

Checks whether a Pop call can return immediately (item available or EoF).

Returns: true if any hopper has an item or if Sink is EoF.

◆ Size()

template<Type::MoveConstructible T>

std::size_t StormByte::Buffer::Sink< T >::Size ( int key ) const

noexcept

Gets pending item count in bucket key.

Parameters

key	Bucket key identifier.

Returns: Item count, or 0 if bucket does not exist.

◆ To()

template<Type::MoveConstructible T>

Lane StormByte::Buffer::Sink< T >::To ( int key )

noexcept

Redirect of one hopper key.

Parameters

key	Bucket key.

Returns: Lane for >> dest.

◆ Unnotify()

template<Type::MoveConstructible T>

void StormByte::Buffer::Sink< T >::Unnotify ( )

noexcept

Clears Notify on this Sink and on every hopper.

Call after Eof when the consumer condition variable is about to die. Later Eof from a co-writer will not signal it.

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

lib/public/StormByte/buffer/hopper.hxx
lib/public/StormByte/buffer/sink.hxx

Classes

Public Types

Public Member Functions

Detailed Description

Member Typedef Documentation

◆ Select

Constructor & Destructor Documentation

◆ Sink() [1/3]

◆ Sink() [2/3]

◆ Sink() [3/3]

◆ ~Sink()

Member Function Documentation

◆ Bind() [1/2]

◆ Bind() [2/2]

◆ Capacity() [1/2]

◆ Capacity() [2/2]

◆ Drain()

◆ Draining()

◆ EoF()

◆ Eof()

◆ Full()

◆ Notify()

◆ operator<<() [1/2]

◆ operator<<() [2/2]

◆ operator=() [1/2]

◆ operator=() [2/2]

◆ operator>>()

◆ Pop() [1/2]

◆ Pop() [2/2]

◆ Push()

◆ Ready()

◆ Size()

◆ To()

◆ Unnotify()