Sending

Heaps

class heap

Heap that is constructed for transmission.

Subclassed by spead2::send::heap_wrapper

Public Types

typedef std::vector<item>::size_type item_handle

Opaque handle type for retrieving previously added items.

Public Functions

heap(const flavour &flavour_ = flavour())

Constructor.

Parameters
  • flavour_: SPEAD flavour that will be used to encode the heap

const flavour &get_flavour() const

Return flavour.

template<typename ...Args>
item_handle add_item(s_item_pointer_t id, Args&&... args)

Construct a new item.

Return
A handle that can be passed to get_item to update the item

item &get_item(item_handle handle)

Get a reference to a previously added item.

The retrieved item reference may be modified to update the heap in place. Behaviour is undefined if handle is not a handle previously returned by add_item.

Parameters
  • handle: Item handle previously returned from add_item

const item &get_item(item_handle handle) const

Get a reference to a previously added item.

Behaviour is undefined if handle is not a handle previously returned by add_item.

Parameters
  • handle: Item handle previously returned from add_item

void add_pointer(std::unique_ptr<std::uint8_t[]> &&pointer)

Take over ownership of pointer and arrange for it to be freed when the heap is freed.

void add_descriptor(const descriptor &descriptor)

Encode a descriptor to an item and add it to the heap.

void add_start()

Add a start-of-stream control item.

void add_end()

Add an end-of-stream control item.

void set_repeat_pointers(bool repeat)

Enable/disable repetition of item pointers in all packets.

Usually this is not needed, but it can enable some specialised use cases where immediates can be recovered from incomplete heaps or where the receiver examines the item pointers in each packet to decide how to handle it. The packet size must be large enough to fit all the item pointers for the heap (the implementation also reserves a little space, so do not rely on a tight fit working).

The default is disabled.

bool get_repeat_pointers() const

Return the flag set by set_repeat_pointers.

struct item

An item to be inserted into a heap.

An item does not own its memory.

Public Functions

item()

Default constructor.

This item has undefined values and is not usable.

item(s_item_pointer_t id, const void *ptr, std::size_t length, bool allow_immediate)

Create an item referencing existing memory.

item(s_item_pointer_t id, s_item_pointer_t immediate)

Create an item with a value to be encoded as an immediate.

item(s_item_pointer_t id, const std::string &value, bool allow_immediate)

Construct an item referencing the data in a string.

item(s_item_pointer_t id, const std::vector<std::uint8_t> &value, bool allow_immediate)

Construct an item referencing the data in a vector.

Public Members

s_item_pointer_t id

Item ID.

bool is_inline

If true, the item’s value is stored in-place and must be encoded as an immediate.

Non-inline values can still be encoded as immediates if they have the right length.

bool allow_immediate

If true, the item’s value may be encoded as an immediate.

This must be false if the item is variable-sized, because in that case the actual size can only be determined from address differences.

If is_inline is true, then this must be true as well.

const std::uint8_t *ptr

Pointer to the value.

std::size_t length

Length of the value.

s_item_pointer_t immediate

Integer value to store (host endian).

This is used if and only if is_inline is true.

Streams

All stream types are derived from spead2::send::stream using the curiously recurring template pattern and implementing an async_send_packet function.

typedef std::function<void(const boost::system::error_code &ec, item_pointer_t bytes_transferred)> spead2::send::stream::completion_handler
class stream

Abstract base class for streams.

Subclassed by spead2::send::stream_impl< Derived >, spead2::send::stream_impl< inproc_stream >, spead2::send::stream_impl< streambuf_stream >, spead2::send::stream_impl< tcp_stream >, spead2::send::stream_impl< udp_ibv_stream >, spead2::send::stream_impl< udp_stream >

Public Functions

boost::asio::io_service &get_io_service() const

Retrieve the io_service used for processing the stream.

virtual void set_cnt_sequence(item_pointer_t next, item_pointer_t step) = 0

Modify the linear sequence used to generate heap cnts.

The next heap will have cnt next, and each following cnt will be incremented by step. When using this, it is the user’s responsibility to ensure that the generated values remain unique. The initial state is next = 1, cnt = 1.

This is useful when multiple senders will send heaps to the same receiver, and need to keep their heap cnts separate.

virtual bool async_send_heap(const heap &h, completion_handler handler, s_item_pointer_t cnt = -1) = 0

Send h asynchronously, with handler called on completion.

The caller must ensure that h remains valid (as well as any memory it points to) until handler is called.

If this function returns true, then the heap has been added to the queue. The completion handlers for such heaps are guaranteed to be called in order.

If this function returns false, the heap was rejected due to insufficient space. The handler is called as soon as possible (from a thread running the io_service), with error code boost::asio::error::would_block.

By default the heap cnt is chosen automatically (see set_cnt_sequence). An explicit value can instead be chosen by passing a non-negative value for cnt. When doing this, it is entirely the responsibility of the user to avoid collisions, both with other explicit values and with the automatic counter. This feature is useful when multiple senders contribute to a single stream and must keep their heap cnts disjoint, which the automatic assignment would not do.

Return Value
  • false: If the heap was immediately discarded
  • true: If the heap was enqueued

virtual void flush() = 0

Block until all enqueued heaps have been sent.

This function is thread-safe, but can be live-locked if more heaps are added while it is running.

class udp_stream : public spead2::send::stream_impl<udp_stream>

Public Functions

udp_stream(io_service_ref io_service, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config = stream_config(), std::size_t buffer_size = default_buffer_size, const boost::asio::ip::address &interface_address = boost::asio::ip::address())

Constructor.

This constructor can handle unicast or multicast destinations, but is primarily intended for unicast as it does not provide all the options that the multicast-specific constructors do.

Parameters
  • io_service: I/O service for sending data
  • endpoint: Destination address and port
  • config: Stream configuration
  • buffer_size: Socket buffer size (0 for OS default)
  • interface_address: Source address

    (see tips on Routing)

udp_stream(boost::asio::ip::udp::socket &&socket, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config, std::size_t buffer_size)

Constructor using an existing socket.

The socket must be open but not bound.

udp_stream(boost::asio::ip::udp::socket &&socket, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config = stream_config())

Constructor using an existing socket.

The socket must be open but not connected.

udp_stream(io_service_ref io_service, boost::asio::ip::udp::socket &&socket, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config, std::size_t buffer_size)

Constructor using an existing socket and an explicit io_service or thread pool.

The socket must be open but not bound, and the io_service must match the socket’s.

udp_stream(io_service_ref io_service, boost::asio::ip::udp::socket &&socket, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config = stream_config())

Constructor using an existing socket and an explicit io_service or thread pool.

The socket must be open but not connected, and the io_service must match the socket’s.

udp_stream(io_service_ref io_service, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config, std::size_t buffer_size, int ttl)

Constructor with multicast hop count.

Parameters
  • io_service: I/O service for sending data
  • endpoint: Multicast group and port
  • config: Stream configuration
  • buffer_size: Socket buffer size (0 for OS default)
  • ttl: Maximum number of hops
Exceptions
  • std::invalid_argument: if endpoint is not a multicast address

udp_stream(io_service_ref io_service, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config, std::size_t buffer_size, int ttl, const boost::asio::ip::address &interface_address)

Constructor with multicast hop count and outgoing interface address (IPv4 only).

Parameters
  • io_service: I/O service for sending data
  • endpoint: Multicast group and port
  • config: Stream configuration
  • buffer_size: Socket buffer size (0 for OS default)
  • ttl: Maximum number of hops
  • interface_address: Address of the outgoing interface
Exceptions
  • std::invalid_argument: if endpoint is not an IPv4 multicast address
  • std::invalid_argument: if interface_address is not an IPv4 address

udp_stream(io_service_ref io_service, const boost::asio::ip::udp::endpoint &endpoint, const stream_config &config, std::size_t buffer_size, int ttl, unsigned int interface_index)

Constructor with multicast hop count and outgoing interface address (IPv6 only).

See
if_nametoindex(3)
Parameters
  • io_service: I/O service for sending data
  • endpoint: Multicast group and port
  • config: Stream configuration
  • buffer_size: Socket buffer size (0 for OS default)
  • ttl: Maximum number of hops
  • interface_index: Index of the outgoing interface
Exceptions
  • std::invalid_argument: if endpoint is not an IPv6 multicast address

class tcp_stream : public spead2::send::stream_impl<tcp_stream>

Public Functions

template<typename ConnectHandler>
tcp_stream(io_service_ref io_service, ConnectHandler &&connect_handler, const boost::asio::ip::tcp::endpoint &endpoint, const stream_config &config = stream_config(), std::size_t buffer_size = default_buffer_size, const boost::asio::ip::address &interface_address = boost::asio::ip::address())

Constructor.

A callback is provided to indicate when the connection is established.

Warning
The callback may be called before the constructor returns. The implementation of the callback needs to be prepared to handle this case.
Parameters
  • io_service: I/O service for sending data
  • connect_handler: Callback when connection is established. It is called with a boost::system::error_code to indicate whether connection was successful.
  • endpoint: Destination host and port
  • config: Stream configuration
  • buffer_size: Socket buffer size (0 for OS default)
  • interface_address: Source address

    (see tips on Routing)

tcp_stream(boost::asio::ip::tcp::socket &&socket, const stream_config &config = stream_config())

Constructor using an existing socket.

The socket must be connected.

tcp_stream(io_service_ref io_service, boost::asio::ip::tcp::socket &&socket, const stream_config &config = stream_config())

Constructor using an existing socket.

The socket must be connected.

class inproc_stream : public spead2::send::stream_impl<inproc_stream>

Public Functions

inproc_stream(io_service_ref io_service, std::shared_ptr<inproc_queue> queue, const stream_config &config = stream_config())

Constructor.

class streambuf_stream : public spead2::send::stream_impl<streambuf_stream>

Puts packets into a streambuf (which could come from an ostream).

This should not be used for a blocking stream such as a wrapper around TCP, because doing so will block the asio handler thread.

Subclassed by spead2::send::stream_wrapper< streambuf_stream >

Public Functions

streambuf_stream(io_service_ref io_service, std::streambuf &streambuf, const stream_config &config = stream_config())

Constructor.