Add RapidsMPF launcher rrun

README.md

@@ -80,6 +80,24 @@ The UCX test suite uses, for convenience, MPI to bootstrap, therefore we need to
 mpirun -np 2 cpp/build/gtests/ucxx_tests
 ```
 
+### rrun - Distributed Launcher
+
+RapidsMPF includes `rrun`, a lightweight launcher that eliminates the MPI dependency for multi-GPU workloads. This is particularly useful for development, testing, and environments where MPI is not available.
+
+#### Single-Node Usage
+
+```bash
+# Build rrun
+cd cpp/build
+cmake --build . --target rrun
+
+# Launch 2 ranks in the local node
+./tools/rrun -n 2 ./benchmarks/bench_comm -C ucxx -O all-to-all
+
+# With verbose output and specific GPUs
+./tools/rrun -v -n 4 -g 0,1,2,3 ./benchmarks/bench_comm -C ucxx
+```
+
 ## Algorithms
 ### Table Shuffle Service
 Example of a MPI program that uses the shuffler:

ci/run_cpp_benchmark_smoketests.sh

@@ -40,3 +40,15 @@ for i in {0..2}; do
     exit 1
   fi
 done
+
+# Test with rrun
+
+# Confirm no dependencies on OpenMPI variables
+unset OMPI_ALLOW_RUN_AS_ROOT
+unset OMPI_ALLOW_RUN_AS_ROOT_CONFIRM
+unset OMPI_MCA_opal_cuda_support
+
+python "${TIMEOUT_TOOL_PATH}" 30 \
+    rrun -n 3 -g 0,0,0 ./bench_comm -m cuda -C ucxx
+python "${TIMEOUT_TOOL_PATH}" 30 \
+    rrun --tag-output -n 3 -g 0,0,0 ./bench_comm -m cuda -C ucxx

conda/recipes/librapidsmpf/recipe.yaml

@@ -74,6 +74,7 @@ outputs:
         content: |
           cmake --install cpp/build
           cmake --install cpp/build --component=benchmarking
+          cmake --install cpp/build --component=tools
       dynamic_linking:
         overlinking_behavior: "error"
       prefix_detection:

cpp/CMakeLists.txt

@@ -155,6 +155,8 @@ target_link_options(maybe_asan INTERFACE "$<$<BOOL:${RAPIDSMPF_ASAN}>:-fsanitize
 add_library(
   rapidsmpf
   src/allgather/allgather.cpp
+  src/bootstrap/bootstrap.cpp
+  src/bootstrap/file_backend.cpp
   src/buffer/buffer.cpp
   src/buffer/pinned_memory_resource.cu
   src/buffer/resource.cpp
@@ -190,7 +192,7 @@ if(RAPIDSMPF_HAVE_STREAMING)
   )
 endif()
 if(RAPIDSMPF_HAVE_UCXX)
-  target_sources(rapidsmpf PRIVATE src/communicator/ucxx.cpp)
+  target_sources(rapidsmpf PRIVATE src/bootstrap/ucxx.cpp src/communicator/ucxx.cpp)
 endif()
 if(RAPIDSMPF_HAVE_MPI)
   target_sources(rapidsmpf PRIVATE src/communicator/mpi.cpp)
@@ -324,6 +326,11 @@ if(RAPIDSMPF_BUILD_EXAMPLES)
   add_subdirectory(examples)
 endif()
 
+# ##################################################################################################
+# * add tools
+# -------------------------------------------------------------------------------------
+add_subdirectory(tools)
+
 # ##################################################################################################
 # * install targets -------------------------------------------------------------------------------
 rapids_cmake_install_lib_dir(lib_dir)

cpp/benchmarks/bench_comm.cpp

@@ -8,6 +8,8 @@
 
 #include <mpi.h>
 
+#include <rapidsmpf/bootstrap/bootstrap.hpp>
+#include <rapidsmpf/bootstrap/ucxx.hpp>
 #include <rapidsmpf/communicator/communicator.hpp>
 #include <rapidsmpf/communicator/mpi.hpp>
 #include <rapidsmpf/communicator/ucxx_utils.hpp>
@@ -27,12 +29,15 @@ using namespace rapidsmpf;
 
 class ArgumentParser {
   public:
-    ArgumentParser(int argc, char* const* argv) {
-        RAPIDSMPF_EXPECTS(mpi::is_initialized() == true, "MPI is not initialized");
-
-        int rank, nranks;
-        RAPIDSMPF_MPI(MPI_Comm_rank(MPI_COMM_WORLD, &rank));
-        RAPIDSMPF_MPI(MPI_Comm_size(MPI_COMM_WORLD, &nranks));
+    ArgumentParser(int argc, char* const* argv, bool use_mpi = true) {
+        int rank = 0;
+        int nranks = 1;
+
+        if (use_mpi) {
+            RAPIDSMPF_EXPECTS(mpi::is_initialized() == true, "MPI is not initialized");
+            RAPIDSMPF_MPI(MPI_Comm_rank(MPI_COMM_WORLD, &rank));
+            RAPIDSMPF_MPI(MPI_Comm_size(MPI_COMM_WORLD, &nranks));
+        }
 
         try {
             int option;
@@ -44,6 +49,8 @@ class ArgumentParser {
                         ss << "Usage: " << argv[0] << " [options]\n"
                            << "Options:\n"
                            << "  -C <comm>  Communicator {mpi, ucxx} (default: mpi)\n"
+                           << "             ucxx automatically detects launcher (mpirun "
+                              "or rrun)\n"
                            << "  -O <op>    Operation {all-to-all} (default: "
                               "all-to-all)\n"
                            << "  -n <num>   Message size in bytes (default: 1M)\n"
@@ -63,7 +70,11 @@ class ArgumentParser {
                         if (rank == 0) {
                             std::cerr << ss.str();
                         }
-                        RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, 0));
+                        if (use_mpi) {
+                            RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, 0));
+                        } else {
+                            std::exit(0);
+                        }
                     }
                     break;
                 case 'C':
@@ -73,7 +84,11 @@ class ArgumentParser {
                             std::cerr << "-C (Communicator) must be one of {mpi, ucxx}"
                                       << std::endl;
                         }
-                        RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+                        if (use_mpi) {
+                            RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+                        } else {
+                            std::exit(-1);
+                        }
                     }
                     break;
                 case 'O':
@@ -114,7 +129,11 @@ class ArgumentParser {
                     break;
 #endif
                 case '?':
-                    RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+                    if (use_mpi) {
+                        RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+                    } else {
+                        std::exit(-1);
+                    }
                     break;
                 default:
                     RAPIDSMPF_FAIL("unknown option", std::invalid_argument);
@@ -127,7 +146,11 @@ class ArgumentParser {
             if (rank == 0) {
                 std::cerr << "Error parsing arguments: " << e.what() << std::endl;
             }
-            RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+            if (use_mpi) {
+                RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1));
+            } else {
+                std::exit(-1);
+            }
         }
 
         if (rmm_mr == "cuda") {
@@ -229,27 +252,49 @@ Duration run(
 }
 
 int main(int argc, char** argv) {
-    // Explicitly initialize MPI with thread support, as this is needed for both mpi and
-    // ucxx communicators.
-    int provided;
-    RAPIDSMPF_MPI(MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided));
-
-    RAPIDSMPF_EXPECTS(
-        provided == MPI_THREAD_MULTIPLE,
-        "didn't get the requested thread level support: MPI_THREAD_MULTIPLE"
-    );
+    // Check if we should use bootstrap mode with rrun
+    // This is determined by checking for RAPIDSMPF_RANK environment variable
+    bool use_bootstrap = std::getenv("RAPIDSMPF_RANK") != nullptr;
+
+    int provided = 0;
+    if (!use_bootstrap) {
+        // Explicitly initialize MPI with thread support, as this is needed for both mpi
+        // and ucxx communicators.
+        RAPIDSMPF_MPI(MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided));
+
+        RAPIDSMPF_EXPECTS(
+            provided == MPI_THREAD_MULTIPLE,
+            "didn't get the requested thread level support: MPI_THREAD_MULTIPLE"
+        );
+    }
 
-    ArgumentParser args{argc, argv};
+    ArgumentParser args{argc, argv, !use_bootstrap};
 
     // Initialize configuration options from environment variables.
     rapidsmpf::config::Options options{rapidsmpf::config::get_environment_variables()};
 
     std::shared_ptr<Communicator> comm;
     if (args.comm_type == "mpi") {
+        if (use_bootstrap) {
+            std::cerr << "Error: MPI communicator requires MPI initialization. "
+                      << "Don't use with rrun or unset RAPIDSMPF_RANK." << std::endl;
+            return 1;
+        }
         mpi::init(&argc, &argv);
         comm = std::make_shared<MPI>(MPI_COMM_WORLD, options);
-    } else {  // ucxx
-        comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options);
+    } else if (args.comm_type == "ucxx") {
+        if (use_bootstrap) {
+            // Launched with rrun - use bootstrap backend
+            comm = rapidsmpf::bootstrap::create_ucxx_comm(
+                rapidsmpf::bootstrap::Backend::AUTO, options
+            );
+        } else {
+            // Launched with mpirun - use MPI bootstrap
+            comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options);
+        }
+    } else {
+        std::cerr << "Error: Unknown communicator type: " << args.comm_type << std::endl;
+        return 1;
     }
 
     auto& log = comm->logger();
@@ -350,6 +395,8 @@ int main(int argc, char** argv) {
     }
 #endif
 
-    RAPIDSMPF_MPI(MPI_Finalize());
+    if (!use_bootstrap) {
+        RAPIDSMPF_MPI(MPI_Finalize());
+    }
     return 0;
 }

cpp/include/rapidsmpf/bootstrap/bootstrap.hpp

@@ -0,0 +1,139 @@
+/**
+ * SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+#include <rapidsmpf/config.hpp>
+#include <rapidsmpf/utils.hpp>
+
+namespace rapidsmpf::bootstrap {
+
+/// @brief Type alias for communicator::Rank
+using Rank = std::int32_t;
+
+/// @brief Type alias for duration type
+using rapidsmpf::Duration;
+
+/**
+ * @brief Backend types for process coordination and bootstrapping.
+ */
+enum class Backend {
+    /**
+     * @brief Automatically detect the best backend based on environment.
+     *
+     * Detection order:
+     * 1. File-based (default fallback)
+     */
+    AUTO,
+
+    /**
+     * @brief File-based coordination using a shared directory.
+     *
+     * Uses filesystem for rank coordination and address exchange.  Works on single-node
+     * and multi-node with shared storage (e.g., NFS) via SSH. Requires RAPIDSMPF_RANK,
+     * RAPIDSMPF_NRANKS, RAPIDSMPF_COORD_DIR environment variables.
+     */
+    FILE,
+};
+
+/**
+ * @brief Context information for the current process/rank.
+ *
+ * This structure contains the rank assignment and total rank count,
+ * along with additional metadata about the execution environment.
+ */
+struct Context {
+    /** @brief This process's rank (0-indexed). */
+    Rank rank;
+
+    /** @brief Total number of ranks in the job. */
+    Rank nranks;
+
+    /** @brief Backend used for coordination. */
+    Backend backend;
+
+    /** @brief Coordination directory (for FILE backend). */
+    std::optional<std::string> coord_dir;
+};
+
+/**
+ * @brief Initialize the bootstrap context from environment variables.
+ *
+ * This function reads environment variables to determine rank, nranks, and
+ * backend configuration. It should be called early in the application lifecycle.
+ *
+ * Environment variables checked (in order of precedence):
+ * - RAPIDSMPF_RANK: Explicitly set rank
+ * - RAPIDSMPF_NRANKS: Explicitly set total rank count
+ * - RAPIDSMPF_COORD_DIR: File-based coordination directory
+ *
+ * @param backend Backend to use (default: AUTO for auto-detection).
+ * @return Context object containing rank and coordination information.
+ * @throws std::runtime_error if environment is not properly configured.
+ *
+ * @code
+ * auto ctx = rapidsmpf::bootstrap::init();
+ * std::cout << "I am rank " << ctx.rank << " of " << ctx.nranks << std::endl;
+ * @endcode
+ */
+Context init(Backend backend = Backend::AUTO);
+
+/**
+ * @brief Broadcast data from root rank to all other ranks.
+ *
+ * This is a helper function for broadcasting small amounts of data during
+ * bootstrapping. It uses the underlying backend's coordination mechanism.
+ *
+ * @param ctx Bootstrap context.
+ * @param data Data buffer to broadcast (both input on root, output on others).
+ * @param size Size of data in bytes.
+ * @param root Root rank performing the broadcast (default: 0).
+ */
+void broadcast(Context const& ctx, void* data, std::size_t size, Rank root = 0);
+
+/**
+ * @brief Perform a barrier synchronization across all ranks.
+ *
+ * This ensures all ranks reach this point before any rank proceeds.
+ *
+ * @param ctx Bootstrap context.
+ */
+void barrier(Context const& ctx);
+
+/**
+ * @brief Store a key-value pair in the coordination backend.
+ *
+ * This is useful for custom coordination beyond UCXX address exchange.
+ *
+ * @param ctx Bootstrap context.
+ * @param key Key name.
+ * @param value Value to store.
+ */
+void put(Context const& ctx, std::string const& key, std::string const& value);
+
+/**
+ * @brief Retrieve a value from the coordination backend.
+ *
+ * This function blocks until the key is available or timeout occurs.
+ *
+ * @param ctx Bootstrap context.
+ * @param key Key name to retrieve.
+ * @param timeout Timeout duration.
+ * @return Value associated with the key.
+ * @throws std::runtime_error if key not found within timeout.
+ */
+std::string get(
+    Context const& ctx,
+    std::string const& key,
+    Duration timeout = std::chrono::milliseconds{30000}
+);
+
+}  // namespace rapidsmpf::bootstrap