Chunking receiver

For an overview, refer to Chunking receiver. This page is a reference for the C++ API.

struct chunk_place_data

This structure is designed to be a plain C structure that can easily be handled by language bindings. As far as possible, new fields will be added to the end but existing fields will be retained, to preserve ABI compatibility.

Do not modify any of the pointers in the structure.

Public Members

const std::uint8_t *packet: Pointer to the original packet data.

std::size_t packet_size: Number of bytes referenced by packet.

s_item_pointer_t *items: Values of requested item pointers.

std::int64_t chunk_id: Chunk ID (output). Set to -1 (or leave unmodified) to discard the heap.

std::size_t heap_index: Number of this heap within the chunk (output)

std::size_t heap_offset: Byte offset of this heap within the chunk payload (output)

std::uint64_t *batch_stats: Pointer to staging area for statistics.

std::uint8_t *extra: Pointer to data to be copied to chunk::extra.

std::size_t extra_offset: Offset within chunk::extra to write.

std::size_t extra_size: Number of bytes to copy to chunk::extra.

typedef std::function<void(chunk_place_data *data, std::size_t data_size)> spead2::recv::chunk_place_function

Callback to determine where a heap is placed in the chunk stream.

Ringbuffer convenience API

template<typename DataRingbuffer = ringbuffer<std::unique_ptr<chunk>>, typename FreeRingbuffer = ringbuffer<std::unique_ptr<chunk>>> class chunk_ring_stream : public detail::chunk_ring_pair<ringbuffer<std::unique_ptr<chunk>>, ringbuffer<std::unique_ptr<chunk>>>, public spead2::recv::chunk_stream 

Wrapper around chunk_stream that uses ringbuffers to manage chunks.

When a fresh chunk is needed, it is retrieved from a ringbuffer of free chunks (the “free ring”). When a chunk is flushed, it is pushed to a “data ring”. These may be shared between streams, but both will be stopped as soon as any of the streams using them are stopped. The intended use case is parallel streams that are started and stopped together.

When stop is called, any in-flight chunks (that are not in either of the ringbuffers) will be freed from the thread that called stop.

Public Functions

chunk_ring_stream(io_service_ref io_service, const stream_config &config, const chunk_stream_config &chunk_config, std::shared_ptr<DataRingbuffer> data_ring, std::shared_ptr<FreeRingbuffer> free_ring)

Constructor.

Refer to chunk_stream::chunk_stream for details.

The allocate callback is ignored and should be unset. If a ready callback is defined, it will be called before the chunk is pushed onto the ringbuffer. It must not move from the provided unique_ptr, but it can be used to perform further processing on the chunk before it is pushed.

Calling get_chunk_config will reflect the internally-defined callbacks.

template<typename T, typename DataSemaphore = semaphore, typename SpaceSemaphore = semaphore> class ringbuffer : public spead2::ringbuffer_base<T>

Ring buffer with blocking and non-blocking push and pop.

It supports non-copyable objects using move semantics. The producer may signal that it has finished producing data by calling stop, which will gracefully shut down consumers as well as other producers. This class is fully thread-safe for multiple producers and consumers.

With multiple producers it is sometimes desirable to only stop the ringbuffer once all the producers are finished. To support this, a producer may register itself with add_producer, and indicate completion with remove_producer. If this causes the number of producers to fall to zero, the stream is stopped.

Public Functions

void try_push(T &&value)

Append an item to the queue, if there is space.

It uses move semantics, so on success, the original value is undefined.

Parameters:

value – Value to move

Throws:

ringbuffer_full – if there is no space
ringbuffer_stopped – if stop has already been called

template<typename ...Args> void try_emplace(Args&&... args)

Construct a new item in the queue, if there is space.

Parameters:

args – Arguments to the constructor

Throws:

ringbuffer_full – if there is no space
ringbuffer_stopped – if stop has already been called

template<typename ...SemArgs> void push(T &&value, SemArgs&&... sem_args)

Append an item to the queue, blocking if necessary.

It uses move semantics, so on success, the original value is undefined.

Parameters:

value – Value to move
sem_args – Arbitrary arguments to pass to the space semaphore

Throws:

ringbuffer_stopped – if stop is called first

template<typename ...Args> void emplace(Args&&... args)

Construct a new item in the queue, blocking if necessary.

Parameters:: args – Arguments to the constructor
Throws:: ringbuffer_stopped – if stop is called first

T try_pop()

Retrieve an item from the queue, if there is one.

Throws:

ringbuffer_stopped – if the queue is empty and stop was called
ringbuffer_empty – if the queue is empty but still active

template<typename ...SemArgs> T pop(SemArgs&&... sem_args)

Retrieve an item from the queue, blocking until there is one or until the queue is stopped.

Parameters:: sem_args – Arbitrary arguments to pass to the data semaphore
Throws:: ringbuffer_stopped – if the queue is empty and stop was called

bool stop()

Indicate that no more items will be produced.

This does not immediately stop consumers if there are still items in the queue; instead, consumers will continue to retrieve remaining items, and will only be signalled once the queue has drained.

Returns:: whether the ringbuffer was stopped

bool remove_producer()

Indicate that a producer registered with add_producer is finished with the ringbuffer.

If this was the last producer, the ringbuffer is stopped.

Returns:: whether the ringbuffer was stopped

inline const DataSemaphore &get_data_sem() const: Get access to the data semaphore.

inline const SpaceSemaphore &get_space_sem() const: Get access to the free-space semaphore.

detail::ringbuffer_iterator<ringbuffer> begin()

Begin iteration over the items in the ringbuffer.

This does not return a full-blown iterator; it is only intended to be used for a range-based for loop. For example: for (auto &&item : ringbuffer) { ... }

detail::ringbuffer_sentinel end(): End iterator (see begin).

std::size_t capacity() const: Maximum number of items that can be held at once.

std::size_t size() const

Return the number of items currently in the ringbuffer.

This should only be used for metrics, not for control flow, as the result could be out of date by the time it is returned.

void add_producer()

Register a new producer.

Producers only need to call this if they want to call ringbuffer::remove_producer.

class ringbuffer_empty : public std::runtime_error: Thrown when attempting to do a non-blocking pop from an empty ringbuffer.

class ringbuffer_full : public std::runtime_error: Thrown when attempting to do a non-blocking push to a full ringbuffer.

class ringbuffer_stopped : public std::runtime_error: Thrown when attempting to do a pop from an empty ringbuffer that has been stopped, or a push to a ringbuffer that has been stopped.