UCXX communicator class that implements the Communicator interface. More...

#include <ucxx.hpp>

Inheritance diagram for rapidsmpf::ucxx::UCXX:

Classes
class	Future
	Represents the future result of an UCXX operation. More...

Public Member Functions
	UCXX (std::unique_ptr< InitializedRank > ucxx_initialized_rank, config::Options options)
	Construct the UCXX rank. More...

Rank	rank () const override
	Retrieves the rank of the current node. More...

Rank	nranks () const override
	Retrieves the total number of ranks. More...

std::unique_ptr< Communicator::Future >	send (std::unique_ptr< std::vector< uint8_t >> msg, Rank rank, Tag tag) override
	Sends a host message to a specific rank. More...

std::unique_ptr< Communicator::Future >	send (std::unique_ptr< Buffer > msg, Rank rank, Tag tag) override
	Sends a message (device or host) to a specific rank. More...

std::unique_ptr< Communicator::Future >	recv (Rank rank, Tag tag, std::unique_ptr< Buffer > recv_buffer) override
	Receives a message from a specific rank to a buffer. Use `release_data` to extract the data out of the buffer once the future is completed. More...

std::unique_ptr< Communicator::Future >	recv_sync_host_data (Rank rank, Tag tag, std::unique_ptr< std::vector< uint8_t >> synced_buffer) override
	Receives a message from a specific rank to an allocated (synchronized) host buffer. Use `release_sync_host_data` to extract the data out of the buffer once the future is completed. More...

std::pair< std::unique_ptr< std::vector< uint8_t > >, Rank >	recv_any (Tag tag) override
	Receives a message from any rank (blocking). More...

std::unique_ptr< std::vector< uint8_t > >	recv_from (Rank src, Tag tag) override
	Receives a message from a specific rank (blocking). More...

std::pair< std::vector< std::unique_ptr< Communicator::Future > >, std::vector< std::size_t > >	test_some (std::vector< std::unique_ptr< Communicator::Future >> &future_vector) override
	Tests for completion of multiple futures. More...

std::vector< std::size_t >	test_some (std::unordered_map< std::size_t, std::unique_ptr< Communicator::Future >> const &future_map) override
	Tests for completion of multiple futures in a map. More...

std::unique_ptr< Buffer >	wait (std::unique_ptr< Communicator::Future > future) override
	Wait for a future to complete and return the data buffer. More...

std::unique_ptr< Buffer >	release_data (std::unique_ptr< Communicator::Future > future) override
	Retrieves data associated with a completed future. More...

std::unique_ptr< std::vector< uint8_t > >	release_sync_host_data (std::unique_ptr< Communicator::Future > future) override
	Retrieves synchronized host data associated with a completed future. When the future is completed, the the host data is valid, and ready, but not stream-ordered. More...

Logger &	logger () override
	Retrieves the logger associated with this communicator. More...

std::string	str () const override
	Provides a string representation of the communicator. More...

void	barrier ()
	Barrier to synchronize all ranks. More...

ListenerAddress	listener_address ()
	Get address to which listener is bound. More...

std::shared_ptr< UCXX >	split ()
	Creates a new communicator with a single rank. More...

Public Member Functions inherited from rapidsmpf::Communicator
virtual std::pair< std::vector< std::unique_ptr< Future > >, std::vector< std::size_t > >	test_some (std::vector< std::unique_ptr< Future >> &future_vector)=0
	Tests for completion of multiple futures. More...

virtual std::vector< std::size_t >	test_some (std::unordered_map< std::size_t, std::unique_ptr< Communicator::Future >> const &future_map)=0
	Tests for completion of multiple futures in a map. More...

virtual std::unique_ptr< Buffer >	wait (std::unique_ptr< Future > future)=0
	Wait for a future to complete and return the data buffer. More...

virtual std::unique_ptr< Buffer >	release_data (std::unique_ptr< Communicator::Future > future)=0
	Retrieves data associated with a completed future. More...

virtual std::unique_ptr< std::vector< uint8_t > >	release_sync_host_data (std::unique_ptr< Communicator::Future > future)=0
	Retrieves synchronized host data associated with a completed future. When the future is completed, the the host data is valid, and ready, but not stream-ordered. More...

Detailed Description

UCXX communicator class that implements the Communicator interface.

This class implements communication functions using UCXX, allowing for data exchange between processes in a distributed system. It supports sending and receiving data, both on the CPU and GPU, and provides asynchronous operations with support for future results.

Definition at line 111 of file ucxx.hpp.

Constructor & Destructor Documentation

◆ UCXX()

rapidsmpf::ucxx::UCXX::UCXX	(	std::unique_ptr< InitializedRank >	ucxx_initialized_rank,
		config::Options	options
	)

Construct the UCXX rank.

Construct the UCXX rank using the context previously returned from the call to init().

Parameters

ucxx_initialized_rank	The previously initialized UCXX rank.
options	Configuration options.

Member Function Documentation

◆ barrier()

void rapidsmpf::ucxx::UCXX::barrier ( )

Barrier to synchronize all ranks.

The barrier is not intended to be performant and therefore should not be used as part of regular rapidsmpf logic, it is designed as a mechanism to wait for the cluster to bootstrap and to wait for completion of all tasks.

◆ listener_address()

ListenerAddress rapidsmpf::ucxx::UCXX::listener_address ( )

Get address to which listener is bound.

Returns: The address to which listener is bound.

◆ logger()

Logger& rapidsmpf::ucxx::UCXX::logger ( )

inlineoverridevirtual

Retrieves the logger associated with this communicator.

Returns: Reference to the logger.

Implements rapidsmpf::Communicator.

Definition at line 279 of file ucxx.hpp.

◆ nranks()

Rank rapidsmpf::ucxx::UCXX::nranks ( ) const

overridevirtual

Retrieves the total number of ranks.

Returns: The total number of ranks.

Implements rapidsmpf::Communicator.

◆ rank()

Rank rapidsmpf::ucxx::UCXX::rank ( ) const

overridevirtual

Retrieves the rank of the current node.

Returns: The rank of the node.

Implements rapidsmpf::Communicator.

◆ recv()

std::unique_ptr<Communicator::Future> rapidsmpf::ucxx::UCXX::recv	(	Rank	rank,
		Tag	tag,
		std::unique_ptr< Buffer >	recv_buffer
	)

overridevirtual

Receives a message from a specific rank to a buffer. Use release_data to extract the data out of the buffer once the future is completed.

Parameters

rank	The source rank.
tag	Message tag for identification.
recv_buffer	The receive buffer.

Returns: A unique pointer to a Future representing the asynchronous operation.

Warning: The caller is responsible to ensure the underlying Buffer allocation is already valid before calling, for example, when a CUDA allocation and/or copy are done asynchronously. Specifically, the caller should ensure Buffer::is_ready() returns true before calling this function, if not, a warning is printed and the application will terminate.

Implements rapidsmpf::Communicator.

◆ recv_any()

std::pair<std::unique_ptr<std::vector<uint8_t> >, Rank> rapidsmpf::ucxx::UCXX::recv_any ( Tag tag )

overridevirtual

Receives a message from any rank (blocking).

Parameters

tag	Message tag for identification.

Returns: A pair containing the message data (host memory) and the rank of the sender.

Note: If no message is available this is indicated by returning a nullptr in the first slot of the pair.

Exceptions

ucxx::Error if there is a message but the receive does not complete successfully.

Implements rapidsmpf::Communicator.

◆ recv_from()

std::unique_ptr<std::vector<uint8_t> > rapidsmpf::ucxx::UCXX::recv_from	(	Rank	src,
		Tag	tag
	)

overridevirtual

Receives a message from a specific rank (blocking).

Parameters

src	The source rank from which to receive the message.
tag	Message tag for identification.

Returns: A unique pointer to a vector containing the received message data (host memory).

Note: If no message is available, this function returns a nullptr.

Exceptions

ucxx::Error if there is a message but the receive does not complete successfully.

Implements rapidsmpf::Communicator.

◆ recv_sync_host_data()

std::unique_ptr<Communicator::Future> rapidsmpf::ucxx::UCXX::recv_sync_host_data	(	Rank	rank,
		Tag	tag,
		std::unique_ptr< std::vector< uint8_t >>	synced_buffer
	)

overridevirtual

Receives a message from a specific rank to an allocated (synchronized) host buffer. Use release_sync_host_data to extract the data out of the buffer once the future is completed.

Parameters

rank	The source rank.
tag	Message tag for identification.
synced_buffer	The receive buffer.

Returns: A unique pointer to a Future representing the asynchronous operation.

Implements rapidsmpf::Communicator.

◆ release_data()

std::unique_ptr<Buffer> rapidsmpf::ucxx::UCXX::release_data ( std::unique_ptr< Communicator::Future > future )

override

Retrieves data associated with a completed future.

Parameters

future The completed future.

Returns: A unique pointer to the data buffer.

Exceptions

std::runtime_error if the future has no data.

◆ release_sync_host_data()

std::unique_ptr<std::vector<uint8_t> > rapidsmpf::ucxx::UCXX::release_sync_host_data ( std::unique_ptr< Communicator::Future > future )

override

Retrieves synchronized host data associated with a completed future. When the future is completed, the the host data is valid, and ready, but not stream-ordered.

Parameters

future The completed future.

Returns: A unique pointer to the synchronized host data.

Exceptions

std::runtime_error if the future has no data.

◆ send() [1/2]

std::unique_ptr<Communicator::Future> rapidsmpf::ucxx::UCXX::send	(	std::unique_ptr< Buffer >	msg,
		Rank	rank,
		Tag	tag
	)

overridevirtual

Sends a message (device or host) to a specific rank.

Parameters

msg	Unique pointer to the message data (Buffer).
rank	The destination rank.
tag	Message tag for identification.

Returns: A unique pointer to a Future representing the asynchronous operation.

Warning: The caller is responsible to ensure the underlying Buffer allocation and data are already valid before calling, for example, when a CUDA allocation and/or copy are done asynchronously. Specifically, the caller should ensure Buffer::is_ready() returns true before calling this function, if not, a warning is printed and the application will terminate.

Implements rapidsmpf::Communicator.

◆ send() [2/2]

std::unique_ptr<Communicator::Future> rapidsmpf::ucxx::UCXX::send	(	std::unique_ptr< std::vector< uint8_t >>	msg,
		Rank	rank,
		Tag	tag
	)

overridevirtual

Sends a host message to a specific rank.

This is used to send data that resides in host memory and is guaranteed to be valid at the time of the call.

Parameters

msg	Unique pointer to the message data (host memory).
rank	The destination rank.
tag	Message tag for identification.

Returns: A unique pointer to a Future representing the asynchronous operation.

Implements rapidsmpf::Communicator.

◆ split()

std::shared_ptr<UCXX> rapidsmpf::ucxx::UCXX::split ( )

Creates a new communicator with a single rank.

This method creates a new communicator that acts as if it was a single rank, similar to MPI_Comm_split when color is the rank of the current process and key is 0.

Note: This method is generally used for testing.

Returns: A new UCXX communicator with a single rank.

◆ str()

std::string rapidsmpf::ucxx::UCXX::str ( ) const

overridevirtual

Provides a string representation of the communicator.

Returns: A string describing the communicator.

Implements rapidsmpf::Communicator.

◆ test_some() [1/2]

std::vector<std::size_t> rapidsmpf::ucxx::UCXX::test_some ( std::unordered_map< std::size_t, std::unique_ptr< Communicator::Future >> const & future_map )

override

Tests for completion of multiple futures in a map.

Parameters

future_map Map of futures identified by keys.

Returns: Keys of completed futures.

Exceptions

ucxx::Error if any completed futures did not complete successfully.

◆ test_some() [2/2]

std::pair< std::vector<std::unique_ptr<Communicator::Future> >, std::vector<std::size_t> > rapidsmpf::ucxx::UCXX::test_some ( std::vector< std::unique_ptr< Communicator::Future >> & future_vector )

override

Tests for completion of multiple futures.

Parameters

[in,out] future_vector Vector of Future objects. Completed futures are erased from the vector.

Returns: Pair of completed futures and indices of input vector that were completed.

Exceptions

ucxx::Error if any completed futures did not complete successfully.

◆ wait()

std::unique_ptr<Buffer> rapidsmpf::ucxx::UCXX::wait ( std::unique_ptr< Communicator::Future > future )

override

Wait for a future to complete and return the data buffer.

Parameters

future The future to wait for completion of.

Returns: A unique pointer to the GPU data buffer (or nullptr if the future had no data).

Exceptions

ucxx::Error if the future did not complete successfully.

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

/__w/rapidsmpf/rapidsmpf/cpp/include/rapidsmpf/communicator/ucxx.hpp

Classes

Public Member Functions

Detailed Description

Constructor & Destructor Documentation

◆ UCXX()

Member Function Documentation

◆ barrier()

◆ listener_address()

◆ logger()

◆ nranks()

◆ rank()

◆ recv()

◆ recv_any()

◆ recv_from()

◆ recv_sync_host_data()

◆ release_data()

◆ release_sync_host_data()

◆ send() [1/2]

◆ send() [2/2]

◆ split()

◆ str()

◆ test_some() [1/2]

◆ test_some() [2/2]

◆ wait()