rapidsmpf::streaming::SpillableMessages Class Reference

Container for individually spillable messages. More...

#include <spillable_messages.hpp>

Public Types
using	MessageId = std::uint64_t
	Unique identifier assigned to each message.

Public Member Functions
	SpillableMessages (SpillableMessages const &)=delete
SpillableMessages &	operator= (SpillableMessages const &)=delete
	SpillableMessages (SpillableMessages &&) noexcept=delete
SpillableMessages &	operator= (SpillableMessages &&) noexcept=delete
MessageId	insert (Message &&message)
	Insert a new message and return its assigned ID. More...
Message	extract (MessageId mid)
	Extract and remove a message by ID. More...
Message	copy (MessageId mid, MemoryReservation &reservation)
	Create a deep copy of a message without removing it. More...
std::size_t	spill (MessageId mid, BufferResource *br) const
	Spill a message's device memory to host memory. More...
std::map< MessageId, ContentDescription >	get_content_descriptions () const
	Get a snapshot of current messages' content descriptions. More...
ContentDescription	get_content_description (MessageId mid) const
	Get the content description of a message by ID. More...

Detailed Description

Container for individually spillable messages.

SpillableMessages manages a collection of Message instances that can be spilled or extracted independently. Each message is assigned a unique MessageId upon insertion, which can later be used to extract or spill that message.

The container is thread-safe for concurrent insertions, extractions, and spills.

Example

SpillableMessages messages;
auto id = messages.insert(to_message(TableChunk{...}));
messages.spill(id, ...);
auto msg = messages.extract(id);

Definition at line 39 of file spillable_messages.hpp.

Member Function Documentation

copy()

Message rapidsmpf::streaming::SpillableMessages::copy	(	MessageId	mid,
		MemoryReservation &	reservation
	)

Create a deep copy of a message without removing it.

This method duplicates the message identified by mid while leaving the original message intact inside the container. The returned message is a full deep copy of the payload. If the message is currently being spilled by another thread, this call waits until spilling completes.

Parameters

mid	Message identifier.
reservation	Memory reservation used for allocating buffers during the deep copy. The reservation also determines the memory type of the returned message.

Returns

A deep copy of the referenced Message.

Exceptions

std::out_of_range	If the message has already been extracted or the message identifier is invalid.
std::runtime_error	If required memory cannot be allocated using the provided reservation.

extract()

Message rapidsmpf::streaming::SpillableMessages::extract

(

MessageId

mid

)

Extract and remove a message by ID.

If the message is currently being spilled, this method blocks until spilling completes.

Parameters

mid	Message identifier.

Returns

Extracted Message instance.

Exceptions

std::out_of_range

If the message ID is invalid or was already extracted.

get_content_description()

ContentDescription rapidsmpf::streaming::SpillableMessages::get_content_description

(

MessageId

mid

)

const

Get the content description of a message by ID.

Parameters

mid	Message identifier.

Returns

Content description of the message.

Exceptions

std::out_of_range

If the message does not exist.

get_content_descriptions()

std::map<MessageId, ContentDescription> rapidsmpf::streaming::SpillableMessages::get_content_descriptions

(

)

const

Get a snapshot of current messages' content descriptions.

The returned map may become outdated immediately if other threads modify the container after this call.

Use this snapshot to decide which messages to spill, but keep in mind that the information may no longer be accurate when the actual spill occurs. When calling spill(), the returned size reflects what was actually spilled.

Returns

Copy of a map from MessageId to ContentDescription.

insert()

MessageId rapidsmpf::streaming::SpillableMessages::insert

(

Message &&

message

)

Insert a new message and return its assigned ID.

Parameters

message

Message to insert.

Returns

Assigned MessageId of the inserted message.

spill()

std::size_t rapidsmpf::streaming::SpillableMessages::spill	(	MessageId	mid,
		BufferResource *	br
	)		const

Spill a message's device memory to host memory.

Performs an in-place deep copy of the message's payload from device to host memory using the specified buffer resource.

If the message is currently being accessed by another thread, is already spilled, not spillable, or does not exist, the operation returns immediately without spilling.

Parameters

mid	Message identifier. If the message does not exist, zero is returned.
br	Buffer resource used for allocations during the spill operation.

Returns

Number of bytes released from device memory (0 if nothing was spilled).

Exceptions

std::runtime_error

If there is insufficient host memory to reserve.

---

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

• /__w/rapidsmpf/rapidsmpf/cpp/include/rapidsmpf/streaming/core/spillable_messages.hpp