Is this page helpful?

Streaming Engine#

using rapidsmpf::streaming::Actor = coro::task<void>#

Alias for an actor in a streaming graph.

Actors represent coroutine-based asynchronous operations used throughout the streaming graph.

using rapidsmpf::streaming::Semaphore = coro::semaphore<std::numeric_limits<std::ptrdiff_t>::max()>#: An awaitable semaphore to manage acquisition and release of finite resources.

inline ContentDescription rapidsmpf::streaming::get_content_description( PackedData const &obj )#

Generate a content description for PackedData.

Parameters:: obj – The object’s content to describe.
Returns:: A new content description.

Message rapidsmpf::streaming::to_message( std::uint64_t sequence_number, std::unique_ptr<PackedData> chunk )#

Wrap PackedData into a Message.

Parameters:

sequence_number – Ordering identifier for the message.
chunk – The chunk to wrap into a message.

Returns:

A Message encapsulating the provided chunk as its payload.

ContentDescription rapidsmpf::streaming::get_content_description( PartitionMapChunk const &obj )#

Generate a content description for a PartitionMapChunk.

Parameters:: obj – The object’s content to describe.
Returns:: A new content description.

ContentDescription rapidsmpf::streaming::get_content_description( PartitionVectorChunk const &obj )#

Generate a content description for a PartitionVectorChunk.

Parameters:: obj – The object’s content to describe.
Returns:: A new content description.

Message rapidsmpf::streaming::to_message( std::uint64_t sequence_number, std::unique_ptr<PartitionMapChunk> chunk )#

Wrap a PartitionMapChunk into a Message.

Parameters:

sequence_number – Ordering identifier for the message.
chunk – The chunk to wrap into a message.

Returns:

A Message encapsulating the provided chunk as its payload.

Message rapidsmpf::streaming::to_message( std::uint64_t sequence_number, std::unique_ptr<PartitionVectorChunk> chunk )#

Wrap a PartitionVectorChunk into a Message.

Parameters:

sequence_number – Ordering identifier for the message.
chunk – The chunk to wrap into a message.

Returns:

A Message encapsulating the provided chunk as its payload.

void rapidsmpf::streaming::run_actor_network( std::vector<Actor> actors )#

Runs a list of actors concurrently and waits for all to complete.

This function schedules each actor and blocks until all of them have finished execution. Typically used to launch multiple producer/consumer coroutines in parallel.

Parameters:: actors – A vector of actors to run.

template<std::ranges::range Range> auto rapidsmpf::streaming::coro_results( Range &&task_results )#

Collect the results of multiple finished coroutines.

This helper consumes a range of coroutine result objects (e.g., from coro::when_all or coro::when_any) and extracts their return values by invoking .return_value() on each element.

If the tasks produce a non-void type T, all values are collected into a std::vector<T> and returned.
If the tasks return void, the function simply invokes .return_value() on each element to surface any unhandled exceptions and then returns void.

Note

All result types must be the same. If your coroutines produce heterogeneous result types, this helper cannot be used; you must instead extract each result manually by calling .return_value() on each element, or use the tuple form of coro_results.

Note

The return values of libcoro’s gather functions such as coro::when_all and coro::wait_all must always be retrieved by calling .return_value() (either directly or via this helper). Failing to do so leaves exceptions unobserved, which can cause the streaming pipeline to deadlock or hang indefinitely while waiting for error propagation.

Template Parameters:: Range – A range type whose elements support a .return_value() member function. Typically the result of functions and coroutines like coro::when_all and coro::wait_all
Parameters:: task_results – A range of completed coroutine results.
Returns:: std::vector<T> if the underlying tasks return a value of type T or void if the underlying tasks return void.

template<typename ...Args> auto rapidsmpf::streaming::coro_results( std::tuple<Args...> &&results )#

Collect the results of multiple finished coroutines from a tuple.

This overload works with a tuple of coroutine result objects, typically from co_await coro::when_all(...).

If the tasks produce non-void types, all values are collected into a std::tuple<T1, T2, ...> and returned.
If the tasks return void, the function simply invokes .return_value() on each element to surface any unhandled exceptions and then returns void.

Template Parameters:: Args – Types of coroutine result objects in the tuple
Parameters:: results – Tuple of coroutine result objects to extract values from
Returns:: std::tuple<T1, T2, ...> if the underlying tasks return values, or void if all underlying tasks return void.

coro::task<MemoryReservation> rapidsmpf::streaming::reserve_memory( std::shared_ptr<Context> ctx, std::size_t size, std::int64_t net_memory_delta, MemoryType mem_type = MemoryType::DEVICE, std::optional<AllowOverbooking> allow_overbooking = std::nullopt )#

Reserve memory using the context memory reservation mechanism.

Submits a memory reservation request for the configured memory type and suspends until the request is satisfied. If no pending reservation request can be satisfied within the configured "memory_reserve_timeout", the behavior depends on allow_overbooking.

This is a convenience helper that returns only the MemoryReservation. If more control is required, for example inspecting the amount of overbooking, callers should use MemoryReserveOrWait directly, such as ctx.memory(MemoryType::DEVICE).reserve_or_wait_or_overbook(size, net_memory_delta).

Priority and progress semantics are identical to MemoryReserveOrWait::reserve_or_wait(). In particular, net_memory_delta is used as a heuristic to prefer eligible requests that are expected to reduce memory pressure sooner. Smaller values have higher priority.

// Reserve memory inside an actor:
auto res = co_await reserve_memory(
    ctx,
    1024,
    0,  // net_memory_delta
    MemoryType::DEVICE,
    AllowOverbooking::YES
);
EXPECT_EQ(res.size(), 1024);

// Disable overbooking and fail if no progress is possible:
auto res2 = co_await reserve_memory(
    ctx,
    2048,
    0,  // net_memory_delta
    MemoryType::DEVICE,
    AllowOverbooking::NO
);

See also

MemoryReserveOrWait::reserve_or_wait()

Parameters:

ctx – Actor context used to obtain the memory reservation handle.
size – Number of bytes to reserve.
net_memory_delta – Estimated net change in memory usage after the reservation is allocated and the dependent operation completes. Smaller values have higher priority.
mem_type – Memory type for which to reserve memory.
allow_overbooking – Controls the behavior when no progress is possible within the configured timeout:
- If set to AllowOverbooking::YES, the call may overbook memory when forcing progress.
- If set to AllowOverbooking::NO, the call fails if no progress is possible.
- If not provided, the default behavior is determined by the configuration option "allow_overbooking_by_default".

Throws:

std::runtime_error – If shutdown occurs before the request can be processed.
rapidsmpf::reservation_error – If no progress is possible within the timeout and allow_overbooking resolves to AllowOverbooking::NO.

Returns:

The allocated memory reservation.

struct PartitionMapChunk#

#include <partition.hpp>

Chunk of packed partitions identified by partition ID.

Represents a single unit of work in a streaming pipeline where each partition is associated with a PartID and contains packed (serialized) data.

Public Members

std::unordered_map<shuffler::PartID, PackedData> data#: Packed data for each partition, keyed by partition ID.

struct PartitionVectorChunk#

#include <partition.hpp>

Chunk of packed partitions stored as a vector.

Represents a single unit of work in a streaming pipeline where the partitions are stored in a vector.

Public Members

std::vector<PackedData> data#: Packed data for each partition stored in a vector.

class AllGather#

#include <allgather.hpp>

Asynchronous (coroutine) interface to coll::AllGather.

Once the AllGather is created, many tasks may insert data into it. If multiple tasks insert data, the user is responsible for arranging that insert_finished is only called after all insertions have completed. A single consumer task should extract data.

Public Types

using Ordered = rapidsmpf::coll::AllGather::Ordered#: Tag requesting ordering for extraction.

Public Functions

AllGather( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, OpID op_id )#

Construct an asynchronous allgather.

Parameters:

ctx – Streaming context
comm – Communicator for the collective operation.
op_id – Unique identifier for the allgather.

std::shared_ptr<Context> const &ctx() const noexcept#

Gets the streaming context associated with this AllGather object.

Returns:: Shared pointer to context.

std::shared_ptr<Communicator> const &comm() const noexcept#

Gets the communicator associated with this AllGather.

Returns:: Shared pointer to communicator.

void insert(std::uint64_t sequence_number, PackedData &&chunk)#

Insert a chunk into the allgather.

Parameters:

sequence_number – The sequence number for this chunk.
chunk – The chunk to insert.

void insert_finished()#: Mark that this rank has finished contributing data.

coro::task<std::vector<PackedData>> extract_all( Ordered ordered = Ordered::YES )#

Extract all gathered data.

Parameters:: ordered – If the extracted data should be ordered. If ordered, return data will be ordered first by rank and then by sequence number of the inserted chunks on that rank.
Returns:: Coroutine that completes when all data is available for extraction and returns the data.

class AllReduce#

#include <allreduce.hpp>

Asynchronous (coroutine) interface to coll::AllReduce.

A single extraction task must await extract() to obtain the result and ensure that the reduction completes.

Public Functions

AllReduce( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, std::unique_ptr<Buffer> input, std::unique_ptr<Buffer> output, OpID op_id, coll::ReduceOperator reduce_operator )#

Construct an asynchronous allreduce.

Parameters:

ctx – Streaming context
comm – The communicator for communication.
input – Local data to contribute to the reduction.
output – Allocated buffer in which to place reduction result. Must be the same size and memory type as input. Overwritten with the reduction result (values already in the buffer are ignored).
op_id – Unique operation identifier for this allreduce.
reduce_operator – Type-erased reduction operator to use. See ReduceOperator.

std::shared_ptr<Context> const &ctx() const noexcept#

Gets the streaming context associated with this AllReduce object.

Returns:: Shared pointer to context.

std::shared_ptr<Communicator> const &comm() const noexcept#

Gets the communicator associated with this AllReduce.

Returns:: Shared pointer to communicator.

coro::task<std::pair<std::unique_ptr<Buffer>, std::unique_ptr<Buffer>>> extract( )#

Wait for completion and extract the reduced data.

Returns:: Coroutine that completes when the result is available and returns a pair of the two Buffers passed to the constructor. The first Buffer contains an implementation-defined value, the second Buffer contains the final reduced result.

class ShufflerAsync#

#include <shuffler.hpp>

An asynchronous shuffler that wraps the synchronous shuffler with a coroutine interface.

ShufflerAsync provides an asynchronous interface to the shuffler, allowing data to be inserted and then extracted after the shuffle completes. All local partitions complete simultaneously, so extraction is non-blocking after awaiting insert_finished().

Example usage:

auto shuffle = ShufflerAsync(...);
while (...) {
  shuffle.insert(...);
}
co_await shuffle.insert_finished();
for (auto pid : shuffle.local_partitions()) {
  auto chunks = shuffle.extract(pid);
  // process chunks...
}

{}

Warning

The coroutine returned by insert_finished() must be awaited before the object is destroyed, otherwise the shuffle with terminate in destruction and/or deadlocks will occur.

Public Functions

ShufflerAsync( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, OpID op_id, shuffler::PartID total_num_partitions, shuffler::Shuffler::PartitionOwner partition_owner = shuffler::Shuffler::round_robin )#

Constructs a new ShufflerAsync instance.

Note

The caller promises that inserted buffers are stream-ordered with respect to their own stream, and extracted buffers are likewise guaranteed to be stream- ordered with respect to their own stream.

Parameters:

ctx – The streaming context to use.
comm – Communicator for the collective operation.
op_id – Unique operation ID for this shuffle. Must not be reused until all participants have completed the shuffle operation.
total_num_partitions – Total number of partitions to shuffle data into.
partition_owner – Function that maps a partition ID to its owning rank/node. Defaults to round-robin distribution.

inline constexpr std::shared_ptr<Context> const &ctx() const#

Gets the streaming context associated with this shuffler.

Returns:: A reference to the shared context object.

inline std::shared_ptr<Communicator> const &comm() const noexcept#

Gets the communicator associated with this shuffler.

Returns:: Shared pointer to communicator.

inline constexpr shuffler::PartID total_num_partitions() const#

Gets the total number of partitions for this shuffle operation.

Returns:: The total number of partitions that data will be shuffled into.

inline constexpr shuffler::Shuffler::PartitionOwner const &partition_owner( ) const#

Gets the partition owner function used by this shuffler.

Returns:: A const reference to the function that maps partition IDs to owning ranks.

std::span<shuffler::PartID const> local_partitions() const#

Returns the local partition IDs owned by the current node.

Parameters:

comm – The communicator to use.
total_num_partitions – Total number of partitions in the shuffle.
partition_owner – Function that determines partition ownership.

Returns:

A vector of partition IDs owned by the current node.

void insert( std::unordered_map<shuffler::PartID, PackedData> &&chunks )#

Insert a bunch of packed (serialized) chunks into the shuffle.

Note

Concurrent insertion by multiple threads is supported, the caller must ensure that insert_finished() is called after all insert() calls have completed.

Parameters:: chunks – A map of partition IDs and their packed chunks.

Actor insert_finished()#

Signal that no more data will be inserted into the shuffle.

This informs the shuffler that this rank has finished inserting data. Must be called exactly once.

Note

If multiple threads are insert()ing, you must establish a happens-before relationship between the completion of all insert()s and the final call to insert_finished().

Note

This coroutine function must be awaited to ensure the shuffler has fully completed its asynchronous operations.

Returns:: A coroutine that inserts the finish marker and suspends until the shuffle has completed. Once complete,

std::vector<PackedData> extract(shuffler::PartID pid)#

Extract all chunks belonging to the specified partition.

Parameters:: pid – The ID of the partition to extract.
Throws:: std::logic_error – If the partition has already been extracted or is otherwise not available.
Returns:: A vector of PackedData chunks associated with the partition.

class SparseAlltoall#

#include <sparse_alltoall.hpp>

Asynchronous (coroutine) interface to coll::SparseAlltoall.

Many tasks may insert data concurrently. If multiple tasks insert data, the caller is responsible for arranging that insert_finished() is only called after all insert() operations have completed. Once insert_finished() is awaited, extraction is non-blocking.

Public Functions

SparseAlltoall( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, OpID op_id, std::vector<Rank> srcs, std::vector<Rank> dsts )#

Construct an asynchronous sparse all-to-all.

Parameters:

ctx – Streaming context.
comm – Communicator for the collective operation.
op_id – Unique identifier for the collective.
srcs – Ranks this rank expects to receive from.
dsts – Ranks this rank may send to.

std::shared_ptr<Context> const &ctx() const noexcept#

Gets the streaming context associated with this object.

Returns:: Shared pointer to context.

std::shared_ptr<Communicator> const &comm() const noexcept#

Gets the communicator associated with this SparseAlltoall.

Returns:: Shared pointer to communicator.

void insert(Rank dst, PackedData &&packed_data)#

Insert data to send to a destination rank.

The order the destination rank obtains the sent data is given by the insertion order on the send side. If inserting concurrently to the same destination, the caller must establish a total order of the insertions, otherwise the reconstruction order on the receive side is unspecified.

Note

Concurrent insertion by multiple threads is supported.

Note

the caller must ensure that insert_finished() is called after all insert() calls have completed.

Parameters:

dst – Destination rank. Must be present in the constructor’s dsts.
packed_data – Packed payload and metadata to send.

coro::task<void> insert_finished()#

Indicate that no more data will be inserted for any destination.

Must be called exactly once.

Note

If multiple threads are insert()ing, you must establish a happens-before relationship between the completion of all insert()s and the final call to insert_finished().

Returns:: Coroutine that completes once all data is ready for extraction.

std::vector<PackedData> extract(Rank src)#

Extract all received messages from a source rank.

The returned vector is ordered by the sender’s local insertion order.

Note

Concurrent extraction is supported, behaviour is undefined if two threads attempt to extract data from the same source.

Parameters:: src – Source rank. Must be present in the constructor’s srcs.
Throws:: std::logic_error – If extracting before the collective is complete.
Returns:: All messages received from src.

class Channel#

#include <channel.hpp>

A coroutine-based channel for sending and receiving messages asynchronously.

The constructor is private, use the factory method Context::create_channel() to create a new channel.

In addition to sending messages through a channel, channel producers can communicate metadata to consumers through a metadata side-channel.

Note

The Channel is bounded for the purposes of sending messages (via send), but the side-channel is unbounded for the purposes of sending metadata (via send_metadata). There is no implied ordering between metadata messages and ordinary messages (they travel over separate paths and do not interfere).

Note

The metadata side-channel should be drained (or shutdown) like the Channel itself. For convenience, drain and shutdown ensure that both channel and metadata channel are shut down appropriately. One can also drain or shutdown the side-channel independently (drain_metadata and shutdown_metadata).

Public Functions

coro::task<bool> send(Message msg)#

Asynchronously send a message into the channel.

Suspends if the channel is full.

Parameters:: msg – The msg to send.
Throws:: std::logic_error – If the message is empty.
Returns:: A coroutine that evaluates to true if the msg was successfully sent or false if the channel was shut down.

coro::task<Message> receive()#

Asynchronously receive a message from the channel.

Suspends if the channel is empty.

Throws:: std::logic_error – If the received message is empty.
Returns:: A coroutine that evaluates to the message, which will be empty if the channel is shut down.

coro::task<bool> send_metadata(Message msg)#

Asynchronously send a metadata message into the channel.

Note

Sending metadata is always possible, even if the other end of the channel never consumes the metadata.

Note

A typical usage of metadata will have the consumer reading metadata before reading messages from the channel. Hence, the producer should send metadata (or shutdown the side-channel via shutdown_metadata) before proceeding to send messages.

Parameters:: msg – The metadata message to send.
Throws:: std::logic_error – If the message is empty.
Returns:: A coroutine that evaluates to true if the msg was successfully sent or false if the channel was shut down.

coro::task<Message> receive_metadata()#

Asynchronously receive a metadata message from the channel.

Suspends if no metadata is available.

Returns:: A coroutine that evaluates to the message, which will be empty if the metadata queue is shut down.

Actor drain_metadata( std::shared_ptr<CoroThreadPoolExecutor> executor )#

Drains all pending metadata messages from the channel and shuts down the metadata channel.

This is intended to ensure all remaining metadata messages are processed.

Warning

If the consumer has no intention of reading metadata messages it must call shutdown_metadata (directly, or indirectly via shutdown) otherwise when the producer drains the output metadata channel it will block forever.

Parameters:: executor – The thread pool used to process remaining messages.
Returns:: A coroutine representing the completion of the metadata shutdown drain.

Actor drain(std::shared_ptr<CoroThreadPoolExecutor> executor)#

Drains all pending messages from the channel and shuts it down.

This is intended to ensure all remaining messages are processed.

Warning

Parameters:: executor – The thread pool used to process remaining messages.
Returns:: A coroutine representing the completion of the shutdown drain.

Actor shutdown()#

Immediately shuts down the channel.

Any pending or future send/receive operations (including metadata messages) will complete with failure.

Returns:: A coroutine representing the completion of the shutdown.

Actor shutdown_metadata()#

Immediately shuts down the metadata channel.

Any pending or future metadata send/receive operations will complete with failure.

Note

If the producer has no metadata to provide, it should shutdown_metadata before anything else.

Returns:: A coroutine representing the completion of the shutdown.

bool empty() const noexcept#

Check whether the channel is empty.

Returns:: True if there are no messages in the buffer.

bool is_shutdown() const noexcept#

Check whether the channel is shut down.

Returns:: True if the channel is shut down.

class ThrottlingAdaptor#

#include <channel.hpp>

An adaptor to throttle access to a channel.

This adds a semaphore-based throttle to a channel to cap the number of suspended coroutines that can be waiting to send into it. It is useful when writing producer actors that otherwise do not depend on an input channel.

Public Functions

inline explicit ThrottlingAdaptor( std::shared_ptr<Channel> channel, std::ptrdiff_t max_tickets )#

Create an adaptor that throttles sends into a channel.

This adaptor is typically used for producer tasks that have no dependencies but where we nonetheless want to introduce a suspension point before sending into an output channel. Such a task can accept the output channel and wrap it in a ThrottlingAdaptor. Consumers of the adapted channel must first acquire a ticket to send before they can send. At most max_tickets consumers can pass the acquire suspension point at once.

Example usage:

auto ch = ctx->create_channel();
auto throttled = ThrottlingAdaptor(ch, 4);
auto make_task = [&]() {
    auto ticket = co_await throttled.acquire();
    auto data = do_expensive_work();
    auto [_, receipt] = co_await ticket.send(data);
    // Not for correctness, but to allow other threads to pick up awaiters at
    // acquire
    co_await executor->yield();
    co_await receipt;
};
std::vector<coro::task<void>> tasks;
for ( ... ) {
    tasks.push_back(make_task());
}
co_await coro::when_all(std::move(tasks));

Parameters:

channel – Channel to throttle.
max_tickets – Maximum number of simultaneous tickets for sending into the channel.

inline coro::task<Ticket> acquire()#

Obtain a ticket to send a message.

Suspends if all tickets are currently handed out.

Throws:: std::runtime_error – If the semaphore is shut down.
Returns:: A coroutine producing a new Ticket that grants permission to send a message.

class ShutdownAtExit#

#include <channel.hpp>

Helper RAII class to shut down channels when they go out of scope.

When this object is destroyed, it invokes shutdown() on all provided channels in the order they were provided. After shutdown, any pending or future send/receive operations on those channels will fail or yield nullopt.

This is useful inside coroutine bodies to guarantee channels are shut down if an unhandled exception escapes the coroutine. Relying on a channel’s own destructor is insufficient when the channel is shared (e.g., via std::shared_ptr), because other owners keep it alive.

Public Functions

inline explicit ShutdownAtExit( std::vector<std::shared_ptr<Channel>> channels )#

Construct from a vector of channel handles.

The order of elements determines the shutdown order invoked by the destructor.

Parameters:: channels – Vector of shared channel handles to be shut down on destruction.
Throws:: std::invalid_argument – If any channel in the vector is nullptr.

template<class ...T> inline explicit ShutdownAtExit( T&&... channels )#

Variadic convenience constructor.

Enables ShutdownAtExit{ch1, ch2, ...} without explicitly creating a vector. Each argument must be convertible to std::shared_ptr<Channel>. The order of the arguments determines the shutdown order in the destructor.

Template Parameters:: T – Parameter pack of types convertible to std::shared_ptr<Channel>.
Parameters:: channels – One or more channel handles.
Throws:: std::invalid_argument – If any of the provided channel pointers is nullptr.

inline ~ShutdownAtExit() noexcept#

Destructor that synchronously shuts down all channels.

Calls shutdown() on each channel in the same order they were passed.

class Context#

#include <context.hpp>

Context for actors (coroutines) in rapidsmpf.

The context owns shared resources used during execution, including the coroutine executor and memory reservation infrastructure.

A recommended usage pattern is to create a single Context instance up front on the main thread and reuse it throughout the lifetime of the program. This reduces overhead and avoids issues related to destruction on a different thread.

Warning

Shutdown of the context must be initiated from the same thread that created it. Calling shutdown() from a different thread results in program termination. Since the destructor implicitly calls shutdown(), destroying the context from a different thread also results in termination unless the executor has already been shut down explicitly.

Public Functions

Context( config::Options options, std::shared_ptr<Logger> logger, std::shared_ptr<CoroThreadPoolExecutor> executor, std::shared_ptr<BufferResource> br )#

Full constructor for the Context.

All provided pointers must be non-null.

Parameters:

options – Configuration options.
logger – Shared pointer to a logger.
executor – Shared pointer to a coroutine executor.
br – Shared pointer to a buffer resource.

Context( config::Options options, std::shared_ptr<Logger> logger, std::shared_ptr<BufferResource> br )#

Convenience constructor using the provided configuration options.

Parameters:

options – Configuration options.
logger – Shared pointer to a logger.
br – Buffer resource used to reserve host memory and perform data movement.

void shutdown() noexcept#

Shut down the context.

This method is idempotent and only performs shutdown once. Subsequent calls have no effect.

Warning

Shutdown must be initiated from the same thread that constructed the executor. Calling this method from a different thread results in program termination.

config::Options options() const noexcept#

Returns the configuration options.

Returns:: The Options instance.

std::shared_ptr<Logger> const &logger() const noexcept#

Returns:: Shared pointer to the logger.

std::shared_ptr<CoroThreadPoolExecutor> const &executor( ) const noexcept#

Returns the coroutine executor.

Returns:: Shared pointer to the executor.

std::shared_ptr<BufferResource> const &br() const noexcept#

Returns the buffer resource.

Returns:: Shared pointer to the buffer resource.

std::shared_ptr<MemoryReserveOrWait> const &memory( MemoryType mem_type ) const noexcept#

Get the handle for memory reservations for a given memory type.

Returns an object that coordinates asynchronous memory reservation requests for the specified memory type. The returned instance provides backpressure and global progress guarantees, and should be used to reserve memory before performing operations that require memory.

A recommended usage pattern is to reserve all required memory up front as a single atomic reservation. This allows callers to co_await the reservation request and only start executing the operation once all required memory is available.

Parameters:: mem_type – Memory type for which reservations are requested.
Returns:: Shared pointer to the corresponding memory reservation coordinator.

std::shared_ptr<Statistics> statistics() const noexcept#

Returns the statistics collector.

Returns:: Shared pointer to the statistics instance.

std::shared_ptr<Channel> create_channel() const noexcept#

Create a new channel associated with this context.

Returns:: A shared pointer to the newly created channel.

std::shared_ptr<SpillableMessages> const &spillable_messages( ) const noexcept#

Returns the spillable messages collection.

Returns:: Shared pointer to the collection.

std::shared_ptr<BoundedQueue> create_bounded_queue( std::size_t buffer_size ) const noexcept#

Create a new bounded queue associated with this context.

Parameters:: buffer_size – Maximum size of the queue.
Returns:: A shared pointer to the newly created bounded queue.

std::size_t uid() const noexcept#

Return a unique identifier for this context.

The returned value uniquely identifies this Context instance. No two Context objects, past or present, will ever have the same identifier within the same process.

Returns:: A process-unique identifier for this Context.

Public Static Functions

static std::shared_ptr<Context> from_options( any_device_resource mr, std::shared_ptr<Logger> logger, config::Options options, std::shared_ptr<Statistics> statistics = Statistics::disabled() )#

Create a Context based on configuration options.

This is a convenience factory that wires up a fully initialized and usable Context.

Note

The current CUDA device must be set prior to calling this function. Options that depend on device memory availability query the current device.

Warning

Parameters:

mr – Device memory resource used by RapidsMPF. It will be wrapped in an internal RmmResourceAdaptor for allocation tracking.
logger – The logger to use.
options – Configuration options used to initialize the Context and its components.
statistics – The statistics instance to use (disabled by default).

Throws:

std::invalid_argument – If an option value is invalid.
std::out_of_range – If an option value exceeds the representable range.

Returns:

A fully initialized Context.

class CoroThreadPoolExecutor#

#include <coro_executor.hpp>

Executor wrapper around a coro::thread_pool used for coroutine execution.

The executor lifetime defines the lifetime of the underlying thread pool. The number of threads can be provided explicitly or derived from configuration options.

This can be subtle in coroutine-based code, where a scheduled coroutine may unwind its stack on a different thread and trigger destructors. Explicitly calling shutdown() on the creator thread allows the destructor to run safely on any thread afterward.

Warning

Shutdown of the executor must be initiated from the same thread that created it. Calling shutdown() from a different thread results in program termination. Since the destructor implicitly calls shutdown(), destroying the executor from a different thread also results in termination unless the executor has already been shut down explicitly.

Public Functions

CoroThreadPoolExecutor( std::uint32_t num_streaming_threads, std::shared_ptr<Statistics> statistics = Statistics::disabled() )#

Construct an executor with an explicit number of streaming threads.

Parameters:

num_streaming_threads – Number of threads used to execute coroutines. Must be greater than zero.
statistics – Statistics collector associated with the executor. If not provided, statistics collection is disabled. TODO: statistics are not currently collected. In the future, libcoro’s thread start and stop callbacks should be used to track coroutine execution statistics.

CoroThreadPoolExecutor( config::Options options, std::shared_ptr<Statistics> statistics = Statistics::disabled() )#

Construct an executor from configuration options.

Reads the num_streaming_threads option. If the option is not set, a single streaming thread is used by default.

Parameters:

options – Configuration options used to initialize the executor.
statistics – Statistics collector associated with the executor. If not provided, statistics collection is disabled. TODO: statistics are not currently collected. In the future, libcoro’s thread start and stop callbacks should be used to track coroutine execution statistics.

Throws:

std::invalid_argument – If num_streaming_threads is present but not a positive integer.

CoroThreadPoolExecutor(CoroThreadPoolExecutor&&) = delete#: No move and copy constructors and assignment operators.

void shutdown() noexcept#

Shut down the underlying thread pool.

This method is idempotent and only performs shutdown once. Subsequent calls have no effect.

Warning

Shutdown must be initiated from the same thread that constructed the executor. Calling this method from a different thread results in program termination.

inline std::uint32_t num_streaming_threads() const noexcept#

Get the configured number of streaming threads.

Returns:: Number of threads in the underlying libcoro thread pool.

std::unique_ptr<coro::thread_pool> &get() noexcept#

Get access to the underlying thread pool to be used with libcoro.

Note

Ownership of the thread pool remains with the executor.

Returns:: Reference to the owning std::unique_ptr holding the coro::thread_pool.

inline auto schedule()#

Schedule work on the underlying libcoro thread pool.

Returns:: A libcoro awaitable as returned by coro::thread_pool::schedule().

inline auto schedule(auto task)#

Schedule a task on the underlying libcoro thread pool.

Parameters:: task – Task to schedule.
Returns:: A libcoro awaitable as returned by coro::thread_pool::schedule(task).

inline auto yield()#

Yield execution back to the underlying libcoro thread pool.

Returns:: A libcoro awaitable as returned by coro::thread_pool::yield().

inline auto spawn_detached(auto task) noexcept#

Spawn a detached task on the underlying libcoro thread pool.

Parameters:: task – Task to spawn.
Returns:: Result as returned by coro::thread_pool::spawn_detached(task).

inline auto spawn_joinable(auto task) noexcept#

Spawn a joinable task on the underlying libcoro thread pool.

Parameters:: task – Task to spawn.
Returns:: Result as returned by coro::thread_pool::spawn_joinable(task).

class Lineariser#

#include <lineariser.hpp>

Linearise insertion into an output channel from a fixed number of producers by sequence number.

Producers are polled in round-robin fashion, and therefore must deliver messages in round-robin increasing sequence number order. If this guarantee is upheld, then the output of the Lineariser is guaranteed to be in total order of the sequence numbers.

Example usage:

auto ctx = std::make_shared<Context>(...);
auto ch_out = ctx->create_channel();
auto linearise = std::make_shared<Lineariser>(ch_out, 8);
std::vector<Actor> tasks;
// Draining the lineariser will pull from all the input channels until they are
// shutdown and send to the output channel until it is consumed.
tasks.push_back(linearise->drain());
for (auto& ch_in: lineariser->get_inputs()) {
  // Each producer promises to send an increasing stream of sequence ids in
  // round-robin fashion. That is, if there are P producers, producer 0 sends
  // [0, P, 2P, ...], producer 1 sends [1, P+1, 2P + 1, ...] and producer i
  // sends [i, P + i, 2P + i, ...].
  tasks.push_back(producer(ctx, ch_in, ...));
}
coro_results(co_await coro::when_all(std::move(tasks)));
// ch_out will see inputs in global total order of sequence id.

Public Functions

inline Lineariser( std::shared_ptr<Context> ctx, std::shared_ptr<Channel> ch_out, std::size_t num_producers, std::size_t buffer_size = 1 )#

Create a new Lineariser into an output channel.

Parameters:

ctx – Streaming context.
ch_out – The output channel.
num_producers – The number of producers.
buffer_size – The number of messages that are buffered in the lineariser from each producer.

inline std::vector<std::shared_ptr<BoundedQueue>> &get_queues()#

Get a reference to the input queues.

Note

Behaviour is undefined if more than one producer coroutine sends into the same queue.

Returns:: Reference to the BoundedQueues to send into.

inline Actor drain()#

Process inputs and send to the output channel.

Note

This coroutine should be awaited in a coro::when_all with all of the producer tasks.

Returns:: Coroutine representing the linearised sends of all producers.

inline coro::task<void> shutdown()#

Shut down the lineariser, informing both producers and consumer/.

Returns:: Coroutine representing the shutdown of all input queues and the output channel.

class MemoryReserveOrWait#

#include <memory_reserve_or_wait.hpp>

Asynchronous coordinator for memory reservation requests.

MemoryReserveOrWait provides a coroutine-based mechanism for reserving memory with backpressure. Callers submit reservation requests via reserve_or_wait(), which suspends until enough memory is available or progress must be forced.

While requests are pending and none of them fit into the available memory, spilling is triggered on the caller’s behalf. This applies to MemoryType::DEVICE only, since SpillManager measures headroom against device memory. Requests of other memory types wait without spilling.

Public Functions

MemoryReserveOrWait( config::Options options, std::shared_ptr<Logger> logger, MemoryType mem_type, std::shared_ptr<CoroThreadPoolExecutor> executor, std::shared_ptr<BufferResource> br )#

Constructs a MemoryReserveOrWait instance.

If no reservation request can be satisfied within the timeout specified by the "memory_reserve_timeout" key in options, the coroutine forces progress by selecting the smallest pending request and attempting to reserve memory for it. This attempt may result in an empty reservation if the request still cannot be satisfied.

Spilling is attempted before that timeout is reached, see the class docs.

Parameters:

options – Configuration options.
logger – Shared pointer to a logger.
mem_type – The memory type for which reservations are requested.
executor – Shared pointer to a coroutine executor.
br – Buffer resource for memory allocation.*

Actor shutdown()#

Shuts down all pending memory reservation requests.

Returns:: A coroutine that completes only after all pending requests have been cancelled and the periodic memory check task has exited.

coro::task<MemoryReservation> reserve_or_wait( std::size_t size, std::int64_t net_memory_delta )#

Attempts to reserve memory or waits until progress can be made.

This coroutine submits a memory reservation request and then suspends until either sufficient memory becomes available or no reservation request (including other pending requests) makes progress within the configured timeout.

While no pending request fits, spilling is attempted to free the memory the highest-priority one needs, and again just before progress is forced. Both are limited to MemoryType::DEVICE.

The timeout does not apply specifically to this request. Instead, it bounds when a final recovery attempt begins, not when it completes: if no pending reservation request can be satisfied within the timeout, MemoryReserveOrWait forces progress by selecting the smallest pending request, spilling to make room for it, and attempting to reserve memory. That spill waits for any in-flight spill to finish, so the call can return later than the timeout. The forced reservation attempt may result in an empty MemoryReservation if the selected request still cannot be satisfied, for example when nothing is spillable.

When multiple reservation requests are eligible, MemoryReserveOrWait uses net_memory_delta as a heuristic to prefer requests that are expected to reduce memory pressure sooner. The value represents the estimated net change in memory usage after the reservation has been allocated and the dependent operation completes (that is, the memory impact after both allocating size and finishing the work that consumes the reservation):

> 0: expected net increase in memory usage
= 0: memory-neutral
< 0: expected net decrease in memory usage

Smaller values have higher priority. Examples:

Reading data from disk into memory typically has a positive net_memory_delta (memory usage increases).
A row-wise transformation that retains input and output typically has a net delta near zero (memory-neutral).
Writing data to disk or a reduction that frees inputs typically has a negative net_memory_delta (memory usage decreases).

Parameters:

size – Number of bytes to reserve.
net_memory_delta – Estimated net change in memory usage after the reservation is allocated and the dependent operation completes. Smaller values have higher priority.

Throws:

std::runtime_error – If shutdown occurs before the request can be processed.

Returns:

A MemoryReservation representing the allocated memory, or an empty reservation if progress could not be made.

coro::task<std::pair<MemoryReservation, std::size_t>> reserve_or_wait_or_overbook( std::size_t size, std::int64_t net_memory_delta )#

Variant of reserve_or_wait() that allows overbooking on timeout.

This coroutine behaves identically to reserve_or_wait() with respect to request submission, waiting, and progress guarantees. The only difference is the behavior when the progress timeout expires.

If no reservation request can be satisfied before the timeout, this method attempts to reserve the requested memory by allowing overbooking. This guarantees forward progress, but may exceed the configured memory limits.

See also

reserve_or_wait()

Parameters:

size – Number of bytes to reserve.
net_memory_delta – Heuristic used to prioritize eligible requests. See reserve_or_wait() for details and semantics.

Throws:

std::runtime_error – If shutdown occurs before the request can be processed.

Returns:

A pair consisting of:

A MemoryReservation representing the allocated memory.
The number of bytes by which the reservation overbooked the available memory. This value is zero if no overbooking occurred.

coro::task<MemoryReservation> reserve_or_wait_or_fail( std::size_t size, std::int64_t net_memory_delta )#

Variant of reserve_or_wait() that fails if no progress is possible.

This coroutine behaves identically to reserve_or_wait() with respect to request submission, waiting, and progress guarantees until the progress timeout expires.

If no reservation request can be satisfied before the timeout, this method fails instead of forcing progress. Overbooking is not allowed, and no memory reservation is made.

See also

reserve_or_wait()

Parameters:

size – Number of bytes to reserve.
net_memory_delta – Heuristic used to prioritize eligible requests. See reserve_or_wait() for details and semantics.

Throws:

rapidsmpf::reservation_error – If no progress is possible within the timeout.
std::runtime_error – If shutdown occurs before the request can be processed.

Returns:

A MemoryReservation representing the allocated memory.

std::size_t size() const noexcept#

Returns the number of pending memory reservation requests.

It may change concurrently as requests are added or fulfilled.

Returns:: The number of outstanding reservation requests.

std::size_t periodic_memory_check_counter() const noexcept#

Returns the number of iterations performed by periodic_memory_check().

This counter is incremented once per loop iteration inside periodic_memory_check(), and can be useful for diagnostics or testing.

Returns:: The total number of memory-check iterations executed so far.

std::shared_ptr<CoroThreadPoolExecutor> const &executor( ) const noexcept#

Get the coroutine executor used by this instance.

Returns:: Shared pointer to the coroutine executor.

std::shared_ptr<BufferResource> const &br() const noexcept#

Get the buffer resource used for memory reservations.

Returns:: Shared pointer to the buffer resource.

Duration timeout() const noexcept#

Get the configured progress timeout.

Returns:: The progress timeout duration.

Public Static Attributes

static constexpr std::int64_t missing_net_memory_delta = 0#

Sentinel indicating that net_memory_delta estimation has not yet been implemented.

This value is used when a reasonable estimate of the net memory delta is not yet available. Any use of this sentinel should be treated as a TODO, since providing a concrete estimate enables better spilling and scheduling decisions.

See also

reserve_or_wait()

class Message#

#include <message.hpp>

Type-erased message wrapper around a payload.

Public Types

using CopyCallback = std::function<Message(Message const&, MemoryReservation &reservation)>#

Callback for performing a deep copy of a message.

The copy operation allocates new memory for the message’s payload using the provided memory reservation. The memory type specified in the reservation determines where the new copy will primarily reside (e.g., device or host memory).

Param msg:: Source message to copy.
Param reservation:: Memory reservation to consume during allocation.
Return:: A new Message instance containing a deep copy of the payload.

Public Functions

Message() = default#: Create an empty message.

template<typename T> inline Message( std::uint64_t sequence_number, std::unique_ptr<T> payload, ContentDescription content_description, CopyCallback copy_cb = nullptr )#

Construct a new message from a unique pointer to its payload.

The message may optionally support deep-copy and spilling operations through a user-provided CopyCallback. If no callback is provided, copy and spill operations are disabled.

Note

Sequence numbers are used to ensure that when multiple producers send into the same output channel, channel ordering is preserved. Specifically, the guarantee is that Channels always produce elements in increasing sequence number order. To ensure this, single producers must promise to send into the channels in strictly increasing sequence number order. Behaviour is undefined if not. To ensure insertion into an output channel from multiple producers obeys this invariant, use a Lineariser. This promise allows consumers to ensure ordering by buffering at most num_consumers messages, rather than needing to buffer the entire channel input.

Template Parameters:

T – Type of the payload to store inside the message.

Parameters:

sequence_number – Ordering identifier for the message.
payload – Non-null unique pointer to the payload.
content_description – Description of the payload’s content. When a copy callback is provided, this description must accurately reflect the content of the payload (e.g., per-memory-type sizes and spillable status).
copy_cb – Optional callback used to perform deep copies of the message. If nullptr, copying and spilling are disabled.

Throws:

std::invalid_argument – if payload is null.

Message(Message &&other) noexcept = default#

Move construct.

Parameters:: other – Source message.

Message &operator=(Message &&other) noexcept = default#

Move assign.

Parameters:: other – Source message.
Returns:: *this.

inline void reset() noexcept#: Reset the message to empty.

inline bool empty() const noexcept#

Returns true when no payload is stored.

Returns:: true if empty, false otherwise.

inline constexpr std::uint64_t sequence_number() const noexcept#

Returns the sequence number of this message.

Returns:: The sequence number.

template<typename T> inline bool holds() const noexcept#

Compare the payload type.

Template Parameters:: T – Expected payload type.
Returns:: true if the payload is typeid(T), false otherwise.

template<typename T> inline T const &get() const#

Reference to the payload.

The returned reference remains valid until the message is released or reset.

Template Parameters:: T – Payload type.
Throws:: std::invalid_argument – if empty or type mismatch.
Returns:: Reference to the payload.

template<typename T> inline T release()#

Extracts the payload and resets the message.

Template Parameters:: T – Payload type.
Throws:: std::invalid_argument – if empty or type mismatch.
Returns:: The payload.

inline constexpr ContentDescription const &content_description( ) const noexcept#

Returns the content description associated with the message.

Returns:: The message’s content description.

inline constexpr CopyCallback const &copy_cb() const noexcept#

Returns the copy callback associated with the message.

Returns:: The message’s copy callback function.

inline constexpr std::size_t copy_cost() const noexcept#

Returns the total memory size required for a deep copy of the payload.

The computed size represents the total amount of memory that must be reserved to duplicate all content buffers of the message, regardless of their current memory locations. For example, if the payload’s content resides in both host and device memory, the returned size is the sum of both.

See also

copy()

Returns:: Total number of bytes that must be reserved to perform a deep copy of the message’s payload and content buffers.

inline Message copy(MemoryReservation &reservation) const#

Perform a deep copy of this message and its payload.

Invokes the registered copy callback to create a new Message with freshly allocated buffers. The allocation is performed using the provided memory reservation, which also define the target memory type (e.g., host or device).

The resulting message contains a deep copy of the original payload, while preserving the same metadata and callbacks.

Parameters:: reservation – Memory reservation to consume for the copy.
Throws:: std::invalid_argument – if the message does not support copying.
Returns:: A new Message instance containing a deep copy of the payload.

class BoundedQueue#

#include <queue.hpp>

A bounded queue for type-erased Messages.

This adds a semaphore-based ticketing system to coro::queue. The producer must acquire a ticket which is sent with the message to the consumer who can decide when to release the ticket having received the message.

Public Functions

inline coro::task<std::optional<Ticket>> acquire()#

Acquire a ticket to send into the queue.

Returns:: A coroutine to be awaited that provides a ticket (or std::nullopt if the queue is shutdown).

inline coro::task<std::pair<coro::task<void>, Message>> receive( )#

Receive a message from the queue.

Returns:: A coroutine containing the release task and the received message (or a null task and an empty message if the queue is shut down).

inline coro::task<void> drain( std::shared_ptr<CoroThreadPoolExecutor> executor )#

Drain all messages in the queue and shut down.

Parameters:: executor – The thread pool used to process the remaining messages.
Returns:: A coroutine representing completion of the shutdown drain.

inline coro::task<void> shutdown()#

Immediately shut down the queue.

Any pending or future operations will complete with failure.

Returns:: A coroutine representing the shutdown.

inline BoundedQueue::Shutdown raii_shutdown() noexcept#

Obtain an object that will synchronously shutdown the queue when it goes out of scope.

Returns:: A shutdown object.

class SpillableMessages#

#include <spillable_messages.hpp>

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.

Public Types

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

Public Functions

MessageId insert(Message &&message)#

Insert a new message and return its assigned ID.

Parameters:: message – Message to insert.
Returns:: Assigned MessageId of the inserted message.

Message 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.
Throws:: std::out_of_range – If the message ID is invalid or was already extracted.
Returns:: Extracted Message instance.

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

Throws:

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.

Returns:

A deep copy of the referenced Message.

std::size_t 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.

Throws:

std::runtime_error – If there is insufficient host memory to reserve.

Returns:

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

std::map<MessageId, ContentDescription> 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.

ContentDescription get_content_description(MessageId mid) const#

Get the content description of a message by ID.

Parameters:: mid – Message identifier.
Throws:: std::out_of_range – If the message does not exist.
Returns:: Content description of the message.

void clear()#

Clear all outstanding messages.

This is useful for avoiding Items from outliving the BufferResource on which they were allocated. It is the caller’s responsibility to clear the messages before the BufferResource is destroyed.

Actors#

enum class rapidsmpf::streaming::actor::FanoutPolicy : std::uint8_t#

Fanout policy controlling how messages are propagated.

Values:

enumerator BOUNDED#

Process messages as they arrive and immediately forward them.

Messages are forwarded as soon as they are received from the input channel. The next message is not processed until all output channels have completed sending the current one, ensuring backpressure and synchronized flow.

enumerator UNBOUNDED#

Forward messages without enforcing backpressure.

In this mode, messages may be accumulated internally before being broadcast, or they may be forwarded immediately depending on the implementation and downstream consumption rate.

This mode disables coordinated backpressure between outputs, allowing consumers to process at independent rates, but can lead to unbounded buffering and increased memory usage.

Actor rapidsmpf::streaming::actor::allgather( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, std::shared_ptr<Channel> ch_in, std::shared_ptr<Channel> ch_out, OpID op_id, AllGather::Ordered ordered = AllGather::Ordered::YES )#

Create an allgather actor for a single allgather operation.

This is a streaming version of rapidsmpf::coll::AllGather that operates on packed data received through Channels.

Parameters:

ctx – The streaming context to use.
comm – Communicator for the collective operation.
ch_in – Input channel providing PackedDatas to be gathered.
ch_out – Output channel where the gathered PackedDatas are sent.
op_id – Unique identifier for the operation.
ordered – If the extracted data should be sent to the output channel with sequence numbers corresponding to the global total order of input chunks. If yes, then the sequence numbers of the extracted data will be ordered first by rank and then by input sequence number. If no, the sequence number of the extracted chunks will have no relation to any input sequence order.

Returns:

A streaming actor that completes when the allgather is finished and the output channel is drained.

Actor rapidsmpf::streaming::actor::shuffler( std::shared_ptr<Context> ctx, std::shared_ptr<Communicator> comm, std::shared_ptr<Channel> ch_in, std::shared_ptr<Channel> ch_out, OpID op_id, shuffler::PartID total_num_partitions, shuffler::Shuffler::PartitionOwner partition_owner = shuffler::Shuffler::round_robin )#

Launches a shuffler actor for a single shuffle operation.

This is a streaming version of rapidsmpf::shuffler::Shuffler that operates on packed partition chunks using channels.

It consumes partitioned input data from the input channel and produces output chunks grouped by partition_owner.

Parameters:

ctx – The context to use.
comm – Communicator for the collective operation.
ch_in – Input channel providing PartitionMapChunk to be shuffled.
ch_out – Output channel where the resulting PartitionVectorChunks are sent.
op_id – Unique operation ID for this shuffle. Must not be reused until all actors have called Shuffler::shutdown().
total_num_partitions – Total number of partitions to shuffle the data into.
partition_owner – Function that maps a partition ID to its owning rank/node.

Returns:

A streaming actor that completes when the shuffling has finished and the output channel is drained.

Actor rapidsmpf::streaming::actor::fanout( std::shared_ptr<Context> ctx, std::shared_ptr<Channel> ch_in, std::vector<std::shared_ptr<Channel>> chs_out, FanoutPolicy policy )#

Broadcast messages from one input channel to multiple output channels.

The actor continuously receives messages from the input channel and forwards them to all output channels according to the selected fanout policy, see FanoutPolicy.

Each output channel receives a deep copy of the same message.

Parameters:

ctx – The actor context to use.
ch_in – Input channel from which messages are received.
chs_out – Output channels to which messages are broadcast. Must be at least 2.
policy – The fanout strategy to use (see FanoutPolicy).

Throws:

std::invalid_argument – If an unknown fanout policy is specified or if the number of output channels is less than 2.

Returns:

Streaming actor representing the fanout operation.

Actor rapidsmpf::streaming::actor::push_to_channel( std::shared_ptr<Context> ctx, std::shared_ptr<Channel> ch_out, std::vector<Message> messages )#

Asynchronously pushes all messages from a vector into an output channel.

Sends each message of the input vector into the channel in order, marking the end of the stream once done.

Parameters:

ctx – The actor context to use.
ch_out – Output channel to which messages will be sent.
messages – Input vector containing the messages to send.

Throws:

std::invalid_argument – if any of the elements in messages is empty.

Returns:

Streaming actor representing the asynchronous operation.

Actor rapidsmpf::streaming::actor::pull_from_channel( std::shared_ptr<Context> ctx, std::shared_ptr<Channel> ch_in, std::vector<Message> &out_messages )#

Asynchronously pulls all messages from an input channel into a vector.

Receives messages from the channel until it is closed and appends them to the provided output vector.

Parameters:

ctx – The actor context to use.
ch_in – Input channel providing messages.
out_messages – Output vector to store the received messages.

Returns:

Streaming actor representing the asynchronous operation.