Metadata and Payload Exchange#

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.

Note

This class is not thread-safe. All methods 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::function<std::unique_ptr<Buffer>(std::size_t)> allocate_buffer_fn,
    std::shared_ptr<Statistics> statistics
);

Subclassed by rapidsmpf::communicator::TagMetadataPayloadExchange

Public Functions

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

Throws:: std::logic_error – If called more than once.

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

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

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

TagMetadataPayloadExchange( std::shared_ptr<Communicator> comm, OpID op_id, std::function<std::unique_ptr<Buffer>(std::size_t)> allocate_buffer_fn, std::shared_ptr<Statistics> statistics )#

Constructor for TagMetadataPayloadExchange.

Parameters:

comm – The communicator to use for operations.
op_id – The operation ID for tagging messages.
allocate_buffer_fn – Function to allocate buffers for incoming data.
statistics – The statistics to use for tracking communication operations.

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::runtime_error – if a message is sent to itself or if an outgoing message already exists.

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::runtime_error – if a message is sent to itself or if an outgoing message already exists.

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

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

Receive messages from remote ranks.

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

virtual void finish() override#

Signal that no more messages will be sent.

Throws:: std::logic_error – If called more than once.

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.