Is this page helpful?

Metadata and Payload Exchange#

std::unique_ptr<Buffer> rapidsmpf::communicator::default_allocate_buffer( BufferResource *br, std::size_t size, MetadataPayloadExchange::AllocationMode mode )#

The default receive buffer allocator of a MetadataPayloadExchange.

See MetadataPayloadExchange::AllocateBufferFn. Reserves memory in ADDRESSABLE_MEMORY_TYPES, in order of preference, without overbooking. If nothing fits:

With AllocationMode::MAY_FAIL, it returns nullptr.
With AllocationMode::MUST_SUCCEED, it overbooks host memory if allow_overbooking_by_default() of br allows it, and otherwise throws std::runtime_error, which is a fatal, unrecoverable error for the exchange.

Parameters:

br – The buffer resource to allocate from.
size – The size of the buffer in bytes.
mode – Whether the allocation may fail.

Throws:

std::runtime_error – If the allocation must succeed, nothing fits, and overbooking is not allowed.

Returns:

The buffer, or nullptr if mode allows failing and nothing fits.

class MetadataPayloadExchange#

#include <core.hpp>

Interface for exchanging serialized metadata and payload between ranks.

The MetadataPayloadExchange class defines an abstract interface for transmitting messages that contain both serialized metadata and a data payload. This abstraction simplifies scenarios where metadata and payload must be exchanged together as a single logical unit.

Concrete implementations, such as TagMetadataPayloadExchange, use the Communicator to implement this interface. In the future, other implementations may leverage specialized features beyond the basic Communicator API to further optimize this communication pattern.

The exchange only communicates with its peers: messages can only be sent to a peer, are only received from peers, and finish() and is_idle() only involve peers. The peers must not include the calling rank or duplicates, and the peer relation must be symmetric: if rank A has rank B as a peer, rank B has rank A as a peer. This is not checked: if rank B does not have rank A as a peer, rank A waits forever for the termination marker of rank B and never becomes idle, and messages from rank A stay unreceived on rank B, where a later exchange reusing the operation ID may receive them. Use all_remote_ranks() to communicate with every other rank.

An exchange is used in three steps:

Send messages with send(), and call progress(), recv() and release_sent() regularly.
Call finish() once no more messages will be sent.
Keep calling progress(), recv() and release_sent() until is_idle() returns true.

Only then can the exchange be destroyed and its operation ID be reused.

Receive buffers are allocated with allocate_buffer_fn, which is called on the thread calling progress() and must not block or spill. Implementations follow these rules:

When called with AllocationMode::MAY_FAIL, the allocator may return nullptr to indicate that memory is not available right now. The message is kept and the allocation is retried on later calls to progress(). A message is never dropped or delivered without its payload.
If no allocation succeeds within allocation_timeout, one pending message is forced: the allocator is called with AllocationMode::MUST_SUCCEED and must return a buffer or throw. An exception thrown by the allocator propagates out of progress() and is a fatal, unrecoverable error: the exchange can’t be used afterwards. Only one message is forced per timeout. The timeout measures time without any successful allocation, not how long a single message has waited, so a message can wait longer while other allocations keep succeeding.
The order in which pending allocations are retried is implementation defined.

Payloads at rest may live on disk:

When br has a disk resource, implementations may move the payload of a message whose receive buffer was forced by the timeout to disk before the message is delivered, so received messages may be disk-backed. The move blocks progress() while the payload is written. Without a disk resource, the payload is delivered in the forced buffer.
Only payloads in sendable_memory_types() of their destination can be sent.

Note

This class is not thread-safe. All methods except sendable_memory_types() must be called from the same thread.

Note

All concrete implementations are expected to provide a constructor with the following signature:

DerivedMetadataPayloadExchange(
    std::shared_ptr<Communicator> comm,
OpID op_id,
    std::vector<Rank> peers,
MetadataPayloadExchange::AllocateBufferFn allocate_buffer_fn,
Duration allocation_timeout,
    BufferResource* br
);

Subclassed by rapidsmpf::communicator::TagMetadataPayloadExchange

Public Types

enum class AllocationMode : bool#

Whether a receive buffer allocation may fail.

See AllocateBufferFn.

Values:

enumerator MAY_FAIL#: The allocator may return nullptr if memory is not available now.

enumerator MUST_SUCCEED#: The allocator must return a buffer or throw.

enum class AfterSend : bool#

What happens to an outgoing message once it has been sent.

Values:

enumerator DISCARD#: The message is discarded.

enumerator RETAIN#: The message is retained and returned by release_sent().

using AllocateBufferFn = std::function<std::unique_ptr<Buffer>(BufferResource*, std::size_t, AllocationMode)>#

Function allocating a receive buffer of the given size.

The exchange calls it with its own buffer resource.

With AllocationMode::MAY_FAIL, it may return nullptr if memory is not available right now. With AllocationMode::MUST_SUCCEED, it must return a buffer, for example by overbooking memory, or throw if its policy does not allow forcing an allocation. Throwing is a fatal, unrecoverable error, so only throw as the very last resort.

Public Functions

virtual std::span<MemoryType const> sendable_memory_types( Rank dst ) const = 0#

Get the memory types of payloads that can be sent to a remote rank.

Callers must only send payloads in one of these memory types. The result includes at least ADDRESSABLE_MEMORY_TYPES, is fixed at construction, and this method may be called from any thread.

Parameters:: dst – The destination rank.
Returns:: The sendable memory types in decreasing order of preference.

virtual void send(std::unique_ptr<Message> message) = 0#

Send a single message to a remote rank.

Takes ownership of a ready message and manages its transmission, including metadata sending and coordination of data transfer.

The messages sent from the calling process to a destination remote rank are guaranteed to be received in the same order as they were sent. No ordering is guaranteed between messages sent to different remote ranks.

Parameters:: message – Message ready to be sent to a remote rank.

virtual void send( std::vector<std::unique_ptr<Message>> &&messages ) = 0#

Send messages to remote ranks.

Takes ownership of ready messages and manages their transmission, including metadata sending and coordination of data transfer.

Parameters:: messages – Vector of messages ready to be sent to remote ranks.

virtual void progress() = 0#

Progress the communication state machine.

Advances the internal state of the communication layer by processing pending operations such as receiving metadata, setting up data transfers, completing data transfers, and cleaning up completed operations. Completed messages are stored internally and can be retrieved via recv().

This method should be called periodically to make progress on communication.

virtual std::vector<std::unique_ptr<Message>> recv() = 0#

Receive messages from remote ranks.

The messages received by the calling process are guaranteed to be received in the same order as they were sent by the source remote rank. No ordering is guaranteed between messages received from different remote ranks.

Returns:: Vector of completed messages ready for local processing.

virtual std::vector<std::unique_ptr<Message>> release_sent() = 0#

Release retained messages whose send has completed.

Returns the messages sent with AfterSend::RETAIN whose send has completed, as detected by progress(). A released message is unchanged: it has its peer rank, metadata and payload, which can be reused. Messages to the same destination rank are returned in the order they were sent. No ordering is guaranteed between messages sent to different ranks.

Returns:: Vector of sent messages.

virtual void finish() = 0#

Signal that no more messages will be sent.

After calling this method, no further calls to send() are permitted. The implementation sends protocol-level termination markers to all peers so that each receiver knows the exact number of application messages to expect. This enables safe reuse of operation IDs: once all termination markers have been received and all expected messages processed, the communication layer considers itself idle and the tag/op_id can be reused.

Calling it again has no effect.

virtual bool is_finished() const = 0#

Check if finish() has been called.

Returns:: true if finish() has been called, otherwise false.

virtual bool is_idle() const = 0#

Check if the communication layer is currently idle.

Indicates whether there are any active or pending communication operations. Before finish() is called, a return value of true means no I/O operations are in progress, and no messages wait to be returned by recv() or release_sent(). After finish() is called, true additionally requires that all peers have sent their termination markers and all expected messages have been received, meaning the op_id can safely be reused.

Returns:: true if the communication layer is idle; false if activity is ongoing.

class Message#

#include <core.hpp>

Message class for communication.

This class contains the essential information needed for communication: data payload, metadata, and peer rank (source/destination).

Public Functions

Message( Rank peer_rank, std::vector<std::uint8_t> &&metadata, std::unique_ptr<Buffer> data = nullptr, AfterSend after_send = AfterSend::DISCARD )#

Construct a new Message.

Parameters:

peer_rank – Destination (outgoing) or source (incoming) rank.
metadata – Serialized metadata.
data – Data buffer (can be nullptr for metadata-only messages).
after_send – What happens to an outgoing message once it has been sent.

inline constexpr Rank peer_rank() const noexcept#

Get the destination rank for outgoing or source rank for incoming messages.

Returns:: The rank of the destination or source.

inline constexpr AfterSend after_send() const noexcept#

Get what happens to an outgoing message once it has been sent.

Returns:: Whether the message is discarded or retained for release_sent().

inline constexpr std::vector<std::uint8_t> const &metadata( ) const noexcept#

Get the serialized metadata for this message.

This metadata is sent first to inform the receiver about the incoming message.

Returns:: The serialized metadata.

std::vector<std::uint8_t> release_metadata() noexcept#

Release ownership of the metadata.

This is typically called when transferring metadata to the communication layer.

Returns:: Metadata with ownership transferred.

Buffer const *data() const#

Get the data buffer for this message.

Returns:: The data buffer, or nullptr if no data.

std::unique_ptr<Buffer> release_data() noexcept#

Release ownership of the data buffer.

This is typically called when transferring a buffer to the communication layer.

Returns:: Data buffer with ownership transferred, or nullptr if no data.

void set_data(std::unique_ptr<Buffer> buffer)#

Set the data buffer for this message.

This method can be used by implementations to update the data buffer.

Parameters:: buffer – Data buffer to be set.

class TagMetadataPayloadExchange : public rapidsmpf::communicator::MetadataPayloadExchange #

#include <tag.hpp>

Tag-based implementation of MetadataPayloadExchange.

This implementation provides the same communication protocol as TagMetadataPayloadExchange but works with the abstract Message.

Public Types

enum class ForcedReceive : bool#

What happens to the payload of a message whose receive buffer was forced.

Values:

enumerator MOVE_TO_DISK#: The payload is moved to disk, if br has a disk resource.

enumerator KEEP_IN_MEMORY#: The payload stays in the forced receive buffer.

Public Functions

TagMetadataPayloadExchange( std::shared_ptr<Communicator> comm, OpID op_id, std::vector<Rank> peers, AllocateBufferFn allocate_buffer_fn, Duration allocation_timeout, BufferResource *br, ForcedReceive forced_receive = ForcedReceive::MOVE_TO_DISK )#

Constructor for TagMetadataPayloadExchange.

See MetadataPayloadExchange for how receive buffers are allocated, how the allocation timeout forces progress, and when payloads are moved to disk.

Parameters:

comm – The communicator to use for operations.
op_id – The operation ID for tagging messages.
peers – The ranks to communicate with. See MetadataPayloadExchange.
allocate_buffer_fn – Function to allocate buffers for incoming data.
allocation_timeout – Progress timeout for receive buffer allocations.
br – Buffer resource used to move payloads to disk and whose statistics track communication operations. Must not be nullptr.
forced_receive – What happens to the payload of a message whose receive buffer was forced.

Throws:

std::out_of_range – If a peer is not a rank of comm.
std::invalid_argument – If peers contains the calling rank or duplicates.

~TagMetadataPayloadExchange() noexcept override#

Destructor.

Logs a warning if the exchange is destroyed before finish() has been called and is_idle() returns true.

virtual void send(std::unique_ptr<Message> message) override#

Send a single message to a remote rank.

Takes ownership of a ready message and manages its transmission, including metadata sending and coordination of data transfer.

Parameters:: message – Message ready to be sent to a remote rank.
Throws:: std::invalid_argument – if the destination is not a peer or the message’s payload is not in sendable_memory_types() of its destination.

inline virtual std::span<MemoryType const> sendable_memory_types( Rank dst ) const override#

Get the memory types of payloads that can be sent to a remote rank.

Callers must only send payloads in one of these memory types. The result includes at least ADDRESSABLE_MEMORY_TYPES, is fixed at construction, and this method may be called from any thread.

Returns ADDRESSABLE_MEMORY_TYPES for every remote rank.

Parameters:: dst – The destination rank.
Returns:: The sendable memory types in decreasing order of preference.

virtual void send( std::vector<std::unique_ptr<Message>> &&messages ) override#

Send messages to remote ranks.

Takes ownership of ready messages and manages their transmission, including metadata sending and coordination of data transfer.

Parameters:: messages – Vector of messages ready to be sent to remote ranks.
Throws:: std::invalid_argument – if a destination is not a peer or a message’s payload is not in sendable_memory_types() of its destination.

virtual void progress() override#

Progress the communication state machine.

This method should be called periodically to make progress on communication.

Advances the communication state machine by:

Receiving incoming message metadata
Setting up data transfers
Handling completed data transfers
Cleaning up completed operations

Receive buffers are allocated in message order per peer. While a message from a peer waits for a buffer, later messages from that peer wait behind it.

Any exception thrown by the receive buffer allocator, for example when it is not allowed to force an allocation, propagates out of this function and is a fatal, unrecoverable error.

When a forced receive buffer is moved to disk, this call blocks until the payload is written.

virtual std::vector<std::unique_ptr<Message>> recv() override#

Receive messages from remote ranks.

Returns:: Vector of completed messages ready for local processing.

virtual std::vector<std::unique_ptr<Message>> release_sent( ) override#

Release retained messages whose send has completed.

Returns:: Vector of sent messages.

virtual void finish() override#

Signal that no more messages will be sent.

Calling it again has no effect.

virtual bool is_finished() const override#

Check if finish() has been called.

Returns:: true if finish() has been called, otherwise false.

virtual bool is_idle() const override#

Check if the communication layer is currently idle.

Returns:: true if the communication layer is idle; false if activity is ongoing.