Adding Fanout node

cpp/CMakeLists.txt

@@ -192,6 +192,7 @@ if(RAPIDSMPF_HAVE_STREAMING)
             src/streaming/coll/shuffler.cpp
             src/streaming/core/channel.cpp
             src/streaming/core/context.cpp
+            src/streaming/core/fanout.cpp
             src/streaming/core/leaf_node.cpp
             src/streaming/core/node.cpp
             src/streaming/core/spillable_messages.cpp

cpp/include/rapidsmpf/streaming/core/channel.hpp

@@ -46,6 +46,8 @@ class Channel {
      * @param msg The msg to send.
      * @return A coroutine that evaluates to true if the msg was successfully sent or
      * false if the channel was shut down.
+     *
+     * @throws std::logic_error If the message is empty.
      */
     coro::task<bool> send(Message msg);
 
@@ -56,6 +58,8 @@ class Channel {
      *
      * @return A coroutine that evaluates to the message, which will be empty if the
      * channel is shut down.
+     *
+     * @throws std::logic_error If the received message is empty.
      */
     coro::task<Message> receive();
 
@@ -85,6 +89,13 @@ class Channel {
      */
     [[nodiscard]] bool empty() const noexcept;
 
+    /**
+     * @brief Check whether the channel is shut down.
+     *
+     * @return True if the channel is shut down.
+     */
+    [[nodiscard]] bool is_shutdown() const noexcept;
+
   private:
     Channel(std::shared_ptr<SpillableMessages> spillable_messages)
         : sm_{std::move(spillable_messages)} {}

cpp/include/rapidsmpf/streaming/core/context.hpp

@@ -78,6 +78,13 @@ class Context {
      */
     [[nodiscard]] std::shared_ptr<Communicator> comm() const noexcept;
 
+    /**
+     * @brief Returns the logger.
+     *
+     * @return Reference to the logger.
+     */
+    [[nodiscard]] Communicator::Logger& logger() const noexcept;
+
     /**
      * @brief Returns the progress thread.
      *

cpp/include/rapidsmpf/streaming/core/fanout.hpp

@@ -0,0 +1,67 @@
+/**
+ * SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <rapidsmpf/streaming/core/channel.hpp>
+#include <rapidsmpf/streaming/core/context.hpp>
+#include <rapidsmpf/streaming/core/node.hpp>
+
+namespace rapidsmpf::streaming::node {
+
+/**
+ * @brief Fanout policy controlling how messages are propagated.
+ */
+enum class FanoutPolicy : uint8_t {
+    /**
+     * @brief 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.
+     */
+    BOUNDED,
+
+    /**
+     * @brief 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.
+     */
+    UNBOUNDED,
+};
+
+/**
+ * @brief Broadcast messages from one input channel to multiple output channels.
+ *
+ * The node 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.
+ *
+ * @param ctx The node context to use.
+ * @param ch_in Input channel from which messages are received.
+ * @param chs_out Output channels to which messages are broadcast. Must be at least 2.
+ * @param policy The fanout strategy to use (see ::FanoutPolicy).
+ *
+ * @return Streaming node representing the fanout operation.
+ *
+ * @throws std::invalid_argument If an unknown fanout policy is specified or if the number
+ * of output channels is less than 2.
+ */
+Node fanout(
+    std::shared_ptr<Context> ctx,
+    std::shared_ptr<Channel> ch_in,
+    std::vector<std::shared_ptr<Channel>> chs_out,
+    FanoutPolicy policy
+);
+
+}  // namespace rapidsmpf::streaming::node

cpp/src/streaming/core/channel.cpp

@@ -9,14 +9,15 @@
 namespace rapidsmpf::streaming {
 
 coro::task<bool> Channel::send(Message msg) {
+    RAPIDSMPF_EXPECTS(!msg.empty(), "message cannot be empty");
     auto result = co_await rb_.produce(sm_->insert(std::move(msg)));
     co_return result == coro::ring_buffer_result::produce::produced;
 }
 
 coro::task<Message> Channel::receive() {
-    auto msg = co_await rb_.consume();
-    if (msg.has_value()) {
-        co_return sm_->extract(*msg);
+    auto msg_id = co_await rb_.consume();
+    if (msg_id.has_value()) {
+        co_return sm_->extract(*msg_id);
     } else {
         co_return Message{};
     }
@@ -34,4 +35,8 @@ bool Channel::empty() const noexcept {
     return rb_.empty();
 }
 
+bool Channel::is_shutdown() const noexcept {
+    return rb_.is_shutdown();
+}
+
 }  // namespace rapidsmpf::streaming

cpp/src/streaming/core/context.cpp

@@ -126,6 +126,10 @@ std::shared_ptr<Communicator> Context::comm() const noexcept {
     return comm_;
 }
 
+Communicator::Logger& Context::logger() const noexcept {
+    return comm_->logger();
+}
+
 std::shared_ptr<ProgressThread> Context::progress_thread() const noexcept {
     return progress_thread_;
 }