diff --git a/cpp/libcudf_streaming/CMakeLists.txt b/cpp/libcudf_streaming/CMakeLists.txt index 904162463906..43a543d0ac56 100644 --- a/cpp/libcudf_streaming/CMakeLists.txt +++ b/cpp/libcudf_streaming/CMakeLists.txt @@ -61,29 +61,12 @@ message(VERBOSE "CUDF_STREAMING: Build examples: ${BUILD_EXAMPLES}") # add third party dependencies using CPM rapids_cpm_init() +include(cmake/ConfigureOptionalCommunication.cmake) include(cmake/thirdparty/get_cudf.cmake) include(cmake/thirdparty/get_rapidsmpf.cmake) +cudf_streaming_configure_optional_communication() include(../cmake/thirdparty/get_cucollections.cmake) -if(BUILD_BENCHMARKS) - if(RAPIDSMPF_HAVE_UCXX) - rapids_find_package(ucxx REQUIRED) - else() - message( - STATUS - "CUDF_STREAMING: Skipping UCXX-dependent benchmarks (RAPIDSMPF_HAVE_UCXX=${RAPIDSMPF_HAVE_UCXX})" - ) - endif() - if(RAPIDSMPF_HAVE_MPI) - rapids_find_package(MPI REQUIRED) - else() - message( - STATUS - "CUDF_STREAMING: Skipping MPI-dependent benchmarks (RAPIDSMPF_HAVE_MPI=${RAPIDSMPF_HAVE_MPI})" - ) - endif() -endif() - # ################################################################################################## # * library target -------------------------------------------------------------------------------- add_library( diff --git a/cpp/libcudf_streaming/benchmarks/CMakeLists.txt b/cpp/libcudf_streaming/benchmarks/CMakeLists.txt index 5ed80023f67e..446f65661d7e 100644 --- a/cpp/libcudf_streaming/benchmarks/CMakeLists.txt +++ b/cpp/libcudf_streaming/benchmarks/CMakeLists.txt @@ -13,7 +13,7 @@ add_library(bench_utils INTERFACE) target_sources(bench_utils INTERFACE utils/random_data.cu) target_compile_options(bench_utils INTERFACE $<$:--expt-extended-lambda>) -if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) +if(CUDF_STREAMING_HAVE_COMM) add_executable(bench_shuffle "bench_shuffle.cpp") set_target_properties( bench_shuffle @@ -25,8 +25,9 @@ if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) CUDA_STANDARD_REQUIRED ON ) target_link_libraries( - bench_shuffle PRIVATE cudf_streaming rapidsmpf::rapidsmpf ucxx::ucxx MPI::MPI_CXX - $ bench_utils + bench_shuffle + PRIVATE cudf_streaming rapidsmpf::rapidsmpf cudf_streaming::optional_communication + $ bench_utils ) install( TARGETS bench_shuffle @@ -34,6 +35,8 @@ if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) DESTINATION bin/benchmarks/libcudf_streaming EXCLUDE_FROM_ALL ) +else() + cudf_streaming_skip_optional_target(bench_shuffle "requires MPI or UCXX") endif() add_executable(bench_partition "bench_partition.cpp") @@ -79,6 +82,4 @@ install( EXCLUDE_FROM_ALL ) -if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) - add_subdirectory(streaming) -endif() +add_subdirectory(streaming) diff --git a/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp b/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp index 808c088fc228..dd99c4696b5a 100644 --- a/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp +++ b/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp @@ -4,14 +4,9 @@ */ #include -#include #include -#include #include #include -#include -#include -#include #include #include #include @@ -30,6 +25,21 @@ #include #endif +#ifdef CUDF_STREAMING_HAVE_MPI +#include +#include +#endif + +#ifdef CUDF_STREAMING_HAVE_UCXX +#include +#include +#endif + +#if defined(CUDF_STREAMING_HAVE_MPI) && defined(CUDF_STREAMING_HAVE_UCXX) +#include +#endif + +#include "utils/comm.hpp" #include "utils/misc.hpp" #include "utils/random_data.hpp" #include "utils/rmm_utils.hpp" @@ -38,17 +48,31 @@ class ArgumentParser { public: ArgumentParser(int argc, char* const* argv, bool use_mpi = true) { - int rank = 0; - int nranks = 1; + int rank = 0; + int nranks = 1; + auto abort_or_exit = [&](int code) { +#ifdef CUDF_STREAMING_HAVE_MPI + if (use_mpi) { RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, code)); } +#endif + std::exit(code); + }; if (use_mpi) { +#ifdef CUDF_STREAMING_HAVE_MPI RAPIDSMPF_EXPECTS(rapidsmpf::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)); +#else + RAPIDSMPF_FAIL("MPI support is not available in this build", std::runtime_error); +#endif } else { // When not using MPI, expect to be using bootstrap mode (rrun) +#ifdef CUDF_STREAMING_HAVE_UCXX nranks = rapidsmpf::bootstrap::get_nranks(); +#else + RAPIDSMPF_FAIL("UCXX bootstrap support is not available in this build", std::runtime_error); +#endif } try { int option; @@ -58,7 +82,9 @@ class ArgumentParser { std::stringstream ss; ss << "Usage: " << argv[0] << " [options]\n" << "Options:\n" - << " -C Communicator {mpi, ucxx} (default: mpi)\n" + << " -C Communicator {" + << cudf_streaming::benchmarks::available_communicators() + << "} (default: " << comm_type << ")\n" << " -r Number of runs (default: 1)\n" << " -w Number of warmup runs (default: 0)\n" << " -c Number of columns in the input tables " @@ -88,23 +114,17 @@ class ArgumentParser { #endif << " -h Display this help message\n"; if (rank == 0) { std::cerr << ss.str(); } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, 0)); - } else { - std::exit(0); - } + abort_or_exit(0); } break; case 'C': comm_type = std::string{optarg}; - if (!(comm_type == "mpi" || comm_type == "ucxx")) { + if (!cudf_streaming::benchmarks::is_communicator_available(comm_type)) { if (rank == 0) { - std::cerr << "-C (Communicator) must be one of {mpi, ucxx}" << std::endl; - } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); + std::cerr << "-C (Communicator) must be one of {" + << cudf_streaming::benchmarks::available_communicators() << "}" + << std::endl; } + abort_or_exit(-1); } break; case 'r': parse_integer(num_runs, optarg); break; @@ -122,11 +142,7 @@ class ArgumentParser { "{cuda, pool, async, managed}" << std::endl; } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } + abort_or_exit(-1); } break; case 'l': parse_integer(device_mem_limit_mb, optarg); break; @@ -141,24 +157,14 @@ class ArgumentParser { enable_cupti_monitoring = true; break; #endif - case '?': - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } - break; + case '?': abort_or_exit(-1); break; default: RAPIDSMPF_FAIL("unknown option", std::invalid_argument); } } if (optind < argc) { RAPIDSMPF_FAIL("unknown option", std::invalid_argument); } } catch (std::exception const& e) { if (rank == 0) { std::cerr << "Error parsing arguments: " << e.what() << std::endl; } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } + abort_or_exit(-1); } local_nbytes = num_columns * num_local_rows * num_local_partitions * sizeof(std::int32_t); @@ -212,7 +218,7 @@ class ArgumentParser { rapidsmpf::shuffler::PartID num_local_partitions{1}; rapidsmpf::shuffler::PartID num_output_partitions{1}; std::string rmm_mr{"pool"}; - std::string comm_type{"mpi"}; + std::string comm_type{cudf_streaming::benchmarks::default_communicator()}; std::uint64_t local_nbytes; std::uint64_t total_nbytes; bool enable_output_discard{false}; @@ -229,9 +235,22 @@ void barrier(std::shared_ptr& comm) { bool use_bootstrap = rapidsmpf::bootstrap::is_running_with_rrun(); if (!use_bootstrap) { +#ifdef CUDF_STREAMING_HAVE_MPI RAPIDSMPF_MPI(MPI_Barrier(MPI_COMM_WORLD)); +#else + RAPIDSMPF_FAIL("MPI barrier requested, but MPI support is not available in this build", + std::runtime_error); +#endif } else { - std::dynamic_pointer_cast(comm)->barrier(); +#ifdef CUDF_STREAMING_HAVE_UCXX + auto ucxx = std::dynamic_pointer_cast(comm); + RAPIDSMPF_EXPECTS( + ucxx != nullptr, "Expected UCXX communicator when using bootstrap mode", std::runtime_error); + ucxx->barrier(); +#else + RAPIDSMPF_FAIL("UCXX bootstrap barrier requested, but UCXX support is not available", + std::runtime_error); +#endif } } @@ -449,10 +468,22 @@ int main(int argc, char** argv) // and ucxx communicators when not using bootstrap mode. int provided = 0; if (!use_bootstrap) { +#ifdef CUDF_STREAMING_HAVE_MPI 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"); +#else + std::cerr << "Error: this build has no MPI support. Use UCXX bootstrap mode or build with MPI." + << std::endl; + return 1; +#endif + } else { +#ifndef CUDF_STREAMING_HAVE_UCXX + std::cerr << "Error: this build has no UCXX support. Bootstrap mode is unavailable." + << std::endl; + return 1; +#endif } ArgumentParser args{argc, argv, !use_bootstrap}; @@ -484,6 +515,7 @@ int main(int argc, char** argv) std::shared_ptr comm; auto progress_thread = std::make_shared(stats); if (args.comm_type == "mpi") { +#ifdef CUDF_STREAMING_HAVE_MPI if (use_bootstrap) { std::cerr << "Error: MPI communicator requires MPI initialization. Don't use with " "rrun or unset RRUN_RANK." @@ -492,15 +524,29 @@ int main(int argc, char** argv) } rapidsmpf::mpi::init(&argc, &argv); comm = std::make_shared(MPI_COMM_WORLD, options, progress_thread); +#else + std::cerr << "Error: MPI communicator is not available in this build." << std::endl; + return 1; +#endif } else if (args.comm_type == "ucxx") { +#ifdef CUDF_STREAMING_HAVE_UCXX if (use_bootstrap) { // Launched with rrun - use bootstrap backend comm = rapidsmpf::bootstrap::create_ucxx_comm( progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options); } else { +#ifdef CUDF_STREAMING_HAVE_MPI // Launched with mpirun - use MPI bootstrap comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); +#else + std::cerr << "Error: UCXX without MPI support requires bootstrap mode." << std::endl; + return 1; +#endif } +#else + std::cerr << "Error: UCXX communicator is not available in this build." << std::endl; + return 1; +#endif } else { std::cerr << "Error: Unknown communicator type: " << args.comm_type << std::endl; return 1; @@ -606,6 +652,8 @@ int main(int argc, char** argv) } #endif +#ifdef CUDF_STREAMING_HAVE_MPI if (!use_bootstrap) { RAPIDSMPF_MPI(MPI_Finalize()); } +#endif return 0; } diff --git a/cpp/libcudf_streaming/benchmarks/streaming/CMakeLists.txt b/cpp/libcudf_streaming/benchmarks/streaming/CMakeLists.txt index 1795b5dadef9..5319bc9ad519 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/CMakeLists.txt +++ b/cpp/libcudf_streaming/benchmarks/streaming/CMakeLists.txt @@ -5,7 +5,7 @@ # cmake-format: on # ============================================================================= -if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) +if(CUDF_STREAMING_HAVE_COMM) add_executable(bench_streaming_shuffle "bench_streaming_shuffle.cpp") set_target_properties( bench_streaming_shuffle @@ -16,8 +16,9 @@ if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) CUDA_STANDARD_REQUIRED ON ) target_link_libraries( - bench_streaming_shuffle PRIVATE cudf_streaming rapidsmpf::rapidsmpf ucxx::ucxx MPI::MPI_CXX - $ bench_utils + bench_streaming_shuffle + PRIVATE cudf_streaming rapidsmpf::rapidsmpf cudf_streaming::optional_communication + $ bench_utils ) install( TARGETS bench_streaming_shuffle @@ -25,6 +26,8 @@ if(RAPIDSMPF_HAVE_MPI AND RAPIDSMPF_HAVE_UCXX) DESTINATION bin/benchmarks/libcudf_streaming EXCLUDE_FROM_ALL ) - - add_subdirectory(ndsh) +else() + cudf_streaming_skip_optional_target(bench_streaming_shuffle "requires MPI or UCXX") endif() + +add_subdirectory(ndsh) diff --git a/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp b/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp index 0ab0bebcff98..c93c348d90f5 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp @@ -3,6 +3,7 @@ * reserved. SPDX-License-Identifier: Apache-2.0 */ +#include "../utils/comm.hpp" #include "../utils/misc.hpp" #include "../utils/rmm_utils.hpp" #include "data_generator.hpp" @@ -10,14 +11,9 @@ #include #include #include -#include #include -#include #include #include -#include -#include -#include #include #include #include @@ -35,21 +31,49 @@ #include #include +#ifdef CUDF_STREAMING_HAVE_MPI +#include +#include +#endif + +#ifdef CUDF_STREAMING_HAVE_UCXX +#include +#include +#endif + +#if defined(CUDF_STREAMING_HAVE_MPI) && defined(CUDF_STREAMING_HAVE_UCXX) +#include +#endif + class ArgumentParser { public: ArgumentParser(int argc, char* const* argv, bool use_mpi = true) { - int rank = 0; - int nranks = 1; + int rank = 0; + int nranks = 1; + auto abort_or_exit = [&](int code) { +#ifdef CUDF_STREAMING_HAVE_MPI + if (use_mpi) { RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, code)); } +#endif + std::exit(code); + }; if (use_mpi) { +#ifdef CUDF_STREAMING_HAVE_MPI RAPIDSMPF_EXPECTS(rapidsmpf::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)); +#else + RAPIDSMPF_FAIL("MPI support is not available in this build", std::runtime_error); +#endif } else { // When not using MPI, expect to be using bootstrap mode (rrun) +#ifdef CUDF_STREAMING_HAVE_UCXX nranks = rapidsmpf::bootstrap::get_nranks(); +#else + RAPIDSMPF_FAIL("UCXX bootstrap support is not available in this build", std::runtime_error); +#endif } try { int option; @@ -59,7 +83,9 @@ class ArgumentParser { std::stringstream ss; ss << "Usage: " << argv[0] << " [options]\n" << "Options:\n" - << " -C Communicator {mpi, ucxx} (default: mpi)\n" + << " -C Communicator {" + << cudf_streaming::benchmarks::available_communicators() + << "} (default: " << comm_type << ")\n" << " -r Number of runs (default: 1)\n" << " -w Number of warmup runs (default: 0)\n" << " -c Number of columns in the input tables " @@ -78,23 +104,17 @@ class ArgumentParser { << " -x Enable memory profiler (default: disabled)\n" << " -h Display this help message\n"; if (rank == 0) { std::cerr << ss.str(); } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, 0)); - } else { - std::exit(0); - } + abort_or_exit(0); } break; case 'C': comm_type = std::string{optarg}; - if (!(comm_type == "mpi" || comm_type == "ucxx")) { + if (!cudf_streaming::benchmarks::is_communicator_available(comm_type)) { if (rank == 0) { - std::cerr << "-C (Communicator) must be one of {mpi, ucxx}" << std::endl; - } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); + std::cerr << "-C (Communicator) must be one of {" + << cudf_streaming::benchmarks::available_communicators() << "}" + << std::endl; } + abort_or_exit(-1); } break; case 'r': parse_integer(num_runs, optarg, 1); break; @@ -112,34 +132,20 @@ class ArgumentParser { "{cuda, pool, async, managed}" << std::endl; } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } + abort_or_exit(-1); } break; case 'l': parse_integer(device_mem_limit_mb, optarg); break; case 'L': pinned_mem_disable = true; break; case 'x': enable_memory_profiler = true; break; - case '?': - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } - break; + case '?': abort_or_exit(-1); break; default: RAPIDSMPF_FAIL("unknown option", std::invalid_argument); } } if (optind < argc) { RAPIDSMPF_FAIL("unknown option", std::invalid_argument); } } catch (std::exception const& e) { if (rank == 0) { std::cerr << "Error parsing arguments: " << e.what() << std::endl; } - if (use_mpi) { - RAPIDSMPF_MPI(MPI_Abort(MPI_COMM_WORLD, -1)); - } else { - std::exit(-1); - } + abort_or_exit(-1); } local_nbytes = num_columns * num_local_rows * num_local_partitions * sizeof(std::int32_t); @@ -185,7 +191,7 @@ class ArgumentParser { rapidsmpf::shuffler::PartID num_local_partitions{1}; rapidsmpf::shuffler::PartID num_output_partitions{1}; std::string rmm_mr{"pool"}; - std::string comm_type{"mpi"}; + std::string comm_type{cudf_streaming::benchmarks::default_communicator()}; std::uint64_t local_nbytes; std::uint64_t total_nbytes; bool enable_memory_profiler{false}; @@ -248,10 +254,22 @@ int main(int argc, char** argv) // and ucxx communicators when not using bootstrap mode. int provided = 0; if (!use_bootstrap) { +#ifdef CUDF_STREAMING_HAVE_MPI 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"); +#else + std::cerr << "Error: this build has no MPI support. Use UCXX bootstrap mode or build with MPI." + << std::endl; + return 1; +#endif + } else { +#ifndef CUDF_STREAMING_HAVE_UCXX + std::cerr << "Error: this build has no UCXX support. Bootstrap mode is unavailable." + << std::endl; + return 1; +#endif } ArgumentParser args{argc, argv, !use_bootstrap}; @@ -261,6 +279,7 @@ int main(int argc, char** argv) std::shared_ptr comm; if (args.comm_type == "mpi") { +#ifdef CUDF_STREAMING_HAVE_MPI if (use_bootstrap) { std::cerr << "Error: MPI communicator requires MPI initialization. Don't use with " "rrun or unset RRUN_RANK." @@ -269,15 +288,29 @@ int main(int argc, char** argv) } rapidsmpf::mpi::init(&argc, &argv); comm = std::make_shared(MPI_COMM_WORLD, options, progress_thread); +#else + std::cerr << "Error: MPI communicator is not available in this build." << std::endl; + return 1; +#endif } else if (args.comm_type == "ucxx") { +#ifdef CUDF_STREAMING_HAVE_UCXX if (use_bootstrap) { // Launched with rrun - use bootstrap backend comm = rapidsmpf::bootstrap::create_ucxx_comm( progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options); } else { +#ifdef CUDF_STREAMING_HAVE_MPI // Launched with mpirun - use MPI bootstrap comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); +#else + std::cerr << "Error: UCXX without MPI support requires bootstrap mode." << std::endl; + return 1; +#endif } +#else + std::cerr << "Error: UCXX communicator is not available in this build." << std::endl; + return 1; +#endif } else { std::cerr << "Error: Unknown communicator type: " << args.comm_type << std::endl; return 1; @@ -345,14 +378,22 @@ int main(int argc, char** argv) } if (!use_bootstrap) { +#ifdef CUDF_STREAMING_HAVE_MPI RAPIDSMPF_MPI(MPI_Barrier(MPI_COMM_WORLD)); +#else + RAPIDSMPF_FAIL("MPI barrier requested, but MPI support is not available in this build", + std::runtime_error); +#endif } else { +#ifdef CUDF_STREAMING_HAVE_UCXX auto ucxx = std::dynamic_pointer_cast(comm); - if (ucxx == nullptr) { - log.print("Expected UCXX communicator when using bootstrap mode"); - throw std::runtime_error{"Expected UCXX communicator when using bootstrap mode"}; - } + RAPIDSMPF_EXPECTS( + ucxx != nullptr, "Expected UCXX communicator when using bootstrap mode", std::runtime_error); ucxx->barrier(); +#else + RAPIDSMPF_FAIL("UCXX bootstrap barrier requested, but UCXX support is not available", + std::runtime_error); +#endif } { @@ -385,6 +426,8 @@ int main(int argc, char** argv) log.print(statistics->report({.header = "Statistics (of the last run):"})); } +#ifdef CUDF_STREAMING_HAVE_MPI if (!use_bootstrap) { RAPIDSMPF_MPI(MPI_Finalize()); } +#endif return 0; } diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/CMakeLists.txt b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/CMakeLists.txt index 629c86bf99b7..b4ca094026da 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/CMakeLists.txt +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/CMakeLists.txt @@ -22,8 +22,9 @@ set_target_properties( ) target_link_libraries( - cudf_streaming_ndsh PRIVATE cudf_streaming rapidsmpf::rapidsmpf cuco::cuco MPI::MPI_CXX - $ + cudf_streaming_ndsh + PRIVATE cudf_streaming rapidsmpf::rapidsmpf cuco::cuco cudf_streaming::optional_communication + $ ) set(CUDF_STREAMING_NDSH_QUERIES q01 q03 q04 q09 q21 bench_read) @@ -39,8 +40,8 @@ foreach(query IN ITEMS ${CUDF_STREAMING_NDSH_QUERIES}) CUDA_STANDARD_REQUIRED ON ) target_link_libraries( - ${query} PRIVATE cudf_streaming_ndsh cudf_streaming rapidsmpf::rapidsmpf MPI::MPI_CXX - $ + ${query} PRIVATE cudf_streaming_ndsh cudf_streaming rapidsmpf::rapidsmpf + cudf_streaming::optional_communication $ ) endforeach() diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/bench_read.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/bench_read.cpp index ff2f482e0767..84af904faa3d 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/bench_read.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/bench_read.cpp @@ -94,7 +94,7 @@ struct ProgramOptions { int num_iterations{2}; ///< Number of iterations of query to run int num_streams{16}; ///< Number of streams in stream pool rapidsmpf::ndsh::CommType comm_type{ - rapidsmpf::ndsh::CommType::UCXX}; ///< Type of communicator to create + rapidsmpf::ndsh::ProgramOptions{}.comm_type}; ///< Type of communicator to create cudf::size_type num_rows_per_chunk{100'000'000}; ///< Number of rows to produce per chunk read std::size_t num_producers{1}; ///< Number of simultaneous read_parquet chunk producers. std::size_t num_consumers{1}; ///< Number of simultaneous chunk consumers. @@ -107,11 +107,9 @@ ProgramOptions parse_arguments(int argc, char** argv) { ProgramOptions options; - static constexpr std::array(rapidsmpf::ndsh::CommType::MAX)> - comm_names{"single", "mpi", "ucxx"}; + auto const comm_names = rapidsmpf::ndsh::comm_type_names(); - auto print_usage = [&argv, &options]() { + auto print_usage = [&argv, &comm_names, &options]() { std::cerr << "Usage: " << argv[0] << " [options]\n" << "Options:\n" << " --num-streaming-threads Number of streaming threads (default: " @@ -128,7 +126,9 @@ ProgramOptions parse_arguments(int argc, char** argv) << options.num_producers << ")\n" << " --num-consumers Number of concurrent consumers (default: " << options.num_consumers << ")\n" - << " --comm-type Communicator type: single, mpi, ucxx " + << " --comm-type Communicator type: " + << rapidsmpf::ndsh::available_comm_types() + << " " "(default: " << comm_names[static_cast(options.comm_type)] << ")\n" << " --input-directory Input directory path (required)\n" @@ -276,17 +276,17 @@ ProgramOptions parse_arguments(int argc, char** argv) std::exit(1); } std::string_view const s{optarg}; - auto parsed = std::optional{}; - for (std::size_t i = 0; i < comm_names.size(); ++i) { - if (s == comm_names[i]) { - parsed = static_cast(i); - break; - } - } + auto parsed = rapidsmpf::ndsh::parse_comm_type(s); if (!parsed.has_value()) { std::cerr << "Error: invalid --comm-type '" << s << "' (expected: single, mpi, ucxx)\n"; std::exit(1); } + if (!rapidsmpf::ndsh::is_comm_type_available(*parsed)) { + std::cerr << "Error: communicator '" << s + << "' is not available in this build (available: " + << rapidsmpf::ndsh::available_comm_types() << ")\n"; + std::exit(1); + } options.comm_type = *parsed; break; } diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q01.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q01.cpp index 3a498ff105ff..4d08ef30a704 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q01.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q01.cpp @@ -27,9 +27,7 @@ #include #include -#include #include -#include #include #include #include diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp index 474ce717940c..d33072f88103 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp @@ -34,9 +34,7 @@ #include #include #include -#include #include -#include #include #include #include diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp index 379e1f283166..61283dddc691 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp @@ -32,9 +32,7 @@ #include #include #include -#include #include -#include #include #include #include diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q09.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q09.cpp index 29e341f9a6cf..cbc1429c3cf0 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q09.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q09.cpp @@ -31,9 +31,7 @@ #include #include -#include #include -#include #include #include #include diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp index 086e20e647e8..f8c60599db79 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp @@ -14,14 +14,10 @@ #include #include -#include #include -#include #include #include -#include #include -#include #include #include #include @@ -34,9 +30,24 @@ #include #include #include +#include #include #include +#ifdef CUDF_STREAMING_HAVE_MPI +#include +#include +#endif + +#ifdef CUDF_STREAMING_HAVE_UCXX +#include +#include +#endif + +#if defined(CUDF_STREAMING_HAVE_MPI) && defined(CUDF_STREAMING_HAVE_UCXX) +#include +#endif + namespace rapidsmpf::ndsh { namespace detail { std::vector list_parquet_files(std::string const root_path) @@ -89,6 +100,49 @@ std::map get_column_types(std::string const& input } // namespace detail +bool is_comm_type_available(CommType comm_type) +{ + switch (comm_type) { + case CommType::SINGLE: return true; + case CommType::MPI: +#ifdef CUDF_STREAMING_HAVE_MPI + return true; +#endif + break; + case CommType::UCXX: +#ifdef CUDF_STREAMING_HAVE_UCXX + return true; +#endif + break; + case CommType::MAX: break; + } + return false; +} + +std::string available_comm_types() +{ + auto const names = comm_type_names(); + std::stringstream out; + bool first = true; + for (std::size_t i = 0; i < static_cast(CommType::MAX); ++i) { + auto const comm_type = static_cast(i); + if (!is_comm_type_available(comm_type)) { continue; } + if (!first) { out << ", "; } + out << names[i]; + first = false; + } + return out.str(); +} + +std::optional parse_comm_type(std::string_view name) +{ + auto const names = comm_type_names(); + for (std::size_t i = 0; i < names.size(); ++i) { + if (name == names[i]) { return static_cast(i); } + } + return std::nullopt; +} + streaming::Actor sink_channel(std::shared_ptr ctx, std::shared_ptr ch) { @@ -156,19 +210,31 @@ std::pair, std::shared_ptr> cr std::shared_ptr comm; switch (arguments.comm_type) { case CommType::MPI: +#ifdef CUDF_STREAMING_HAVE_MPI RAPIDSMPF_EXPECTS(!bootstrap::is_running_with_rrun(), "Can't use MPI communicator with rrun"); mpi::init(nullptr, nullptr); comm = std::make_shared(MPI_COMM_WORLD, options, progress_thread); +#else + RAPIDSMPF_FAIL("MPI communicator is not available in this build", std::invalid_argument); +#endif break; case CommType::SINGLE: comm = std::make_shared(options, progress_thread); break; case CommType::UCXX: +#ifdef CUDF_STREAMING_HAVE_UCXX if (bootstrap::is_running_with_rrun()) { comm = bootstrap::create_ucxx_comm(progress_thread, bootstrap::BackendType::AUTO, options); } else { +#ifdef CUDF_STREAMING_HAVE_MPI mpi::init(nullptr, nullptr); comm = ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); +#else + RAPIDSMPF_FAIL("UCXX without MPI support requires bootstrap mode", std::invalid_argument); +#endif } +#else + RAPIDSMPF_FAIL("UCXX communicator is not available in this build", std::invalid_argument); +#endif break; default: RAPIDSMPF_FAIL("Unknown communicator type"); } @@ -187,10 +253,9 @@ ProgramOptions parse_arguments(int argc, char** argv) { ProgramOptions options; - static constexpr std::array(CommType::MAX)> comm_names{ - "single", "mpi", "ucxx"}; + auto const comm_names = comm_type_names(); - auto print_usage = [&argv, &options]() { + auto print_usage = [&argv, &comm_names, &options]() { std::cerr << "Usage: " << argv[0] << " [options]\n" << "Options:\n" << " --num-streaming-threads Number of streaming threads (default: " @@ -217,7 +282,8 @@ ProgramOptions parse_arguments(int argc, char** argv) ? std::to_string(options.periodic_spill.value().count()) : "None") << ")\n" - << " --comm-type Communicator type: single, mpi, ucxx " + << " --comm-type Communicator type: " << available_comm_types() + << " " "(default: " << comm_names[static_cast(options.comm_type)] << ")\n" << " --use-shuffle-join Use shuffle join (default: " @@ -318,14 +384,9 @@ ProgramOptions parse_arguments(int argc, char** argv) break; } case 10: { - std::string comm_type = optarg; - if (comm_type == "mpi") { - options.comm_type = CommType::MPI; - } else if (comm_type == "single") { - options.comm_type = CommType::SINGLE; - } else if (comm_type == "ucxx") { - options.comm_type = CommType::UCXX; - } else { + std::string_view comm_type = optarg; + auto parsed = parse_comm_type(comm_type); + if (!parsed.has_value()) { std::cerr << "Error: Invalid value for --comm-type: " << optarg << " (must be one of " << comm_names[0]; for (std::size_t i = 1; i < comm_names.size(); ++i) { @@ -335,6 +396,14 @@ ProgramOptions parse_arguments(int argc, char** argv) print_usage(); std::exit(1); } + if (!is_comm_type_available(*parsed)) { + std::cerr << "Error: communicator '" << comm_type + << "' is not available in this build (available: " << available_comm_types() + << ")\n\n"; + print_usage(); + std::exit(1); + } + options.comm_type = *parsed; break; } case 11: { @@ -370,4 +439,15 @@ ProgramOptions parse_arguments(int argc, char** argv) return options; } + +FinalizeMPI::~FinalizeMPI() noexcept +{ +#ifdef CUDF_STREAMING_HAVE_MPI + if (rapidsmpf::mpi::is_initialized()) { + int flag; + RAPIDSMPF_MPI(MPI_Finalized(&flag)); + if (!flag) { RAPIDSMPF_MPI(MPI_Finalize()); } + } +#endif +} } // namespace rapidsmpf::ndsh diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.hpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.hpp index cf9148f8adc4..886ae9e04f56 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.hpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.hpp @@ -16,9 +16,7 @@ #include #include -#include #include -#include #include #include #include @@ -26,10 +24,13 @@ #include #include +#include #include #include #include +#include #include +#include #include #include @@ -214,12 +215,30 @@ enum class CommType : std::uint8_t { MAX, ///< Max value }; +[[nodiscard]] constexpr std::array(CommType::MAX)> +comm_type_names() +{ + return {"single", "mpi", "ucxx"}; +} + +[[nodiscard]] bool is_comm_type_available(CommType comm_type); + +[[nodiscard]] std::string available_comm_types(); + +[[nodiscard]] std::optional parse_comm_type(std::string_view name); + ///< @brief Configuration options for the query struct ProgramOptions { - int num_streaming_threads{1}; ///< Number of streaming threads to use - int num_iterations{2}; ///< Number of iterations of query to run - int num_streams{16}; ///< Number of streams in stream pool + int num_streaming_threads{1}; ///< Number of streaming threads to use + int num_iterations{2}; ///< Number of iterations of query to run + int num_streams{16}; ///< Number of streams in stream pool +#ifdef CUDF_STREAMING_HAVE_UCXX CommType comm_type{CommType::UCXX}; ///< Type of communicator to create +#elif defined(CUDF_STREAMING_HAVE_MPI) + CommType comm_type{CommType::MPI}; ///< Type of communicator to create +#else + CommType comm_type{CommType::SINGLE}; ///< Type of communicator to create +#endif std::optional periodic_spill; ///< Duration between background periodic spilling checks cudf::size_type num_rows_per_chunk{100'000'000}; ///< Number of rows to produce per chunk read @@ -255,13 +274,6 @@ std::pair, std::shared_ptr> cr * @brief Finalize MPI when going out of scope. */ struct FinalizeMPI { - ~FinalizeMPI() noexcept - { - if (rapidsmpf::mpi::is_initialized()) { - int flag; - RAPIDSMPF_MPI(MPI_Finalized(&flag)); - if (!flag) { RAPIDSMPF_MPI(MPI_Finalize()); } - } - } + ~FinalizeMPI() noexcept; }; } // namespace rapidsmpf::ndsh diff --git a/cpp/libcudf_streaming/benchmarks/utils/comm.hpp b/cpp/libcudf_streaming/benchmarks/utils/comm.hpp new file mode 100644 index 000000000000..dbf6f7b93742 --- /dev/null +++ b/cpp/libcudf_streaming/benchmarks/utils/comm.hpp @@ -0,0 +1,48 @@ +/** + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights + * reserved. SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include + +namespace cudf_streaming::benchmarks { + +[[nodiscard]] inline bool is_communicator_available(std::string_view name) +{ +#ifdef CUDF_STREAMING_HAVE_MPI + if (name == "mpi") { return true; } +#endif +#ifdef CUDF_STREAMING_HAVE_UCXX + if (name == "ucxx") { return true; } +#endif + return false; +} + +[[nodiscard]] inline std::string available_communicators() +{ + std::string result; +#ifdef CUDF_STREAMING_HAVE_MPI + result += "mpi"; +#endif +#ifdef CUDF_STREAMING_HAVE_UCXX + if (!result.empty()) { result += ", "; } + result += "ucxx"; +#endif + return result.empty() ? "none" : result; +} + +[[nodiscard]] inline std::string default_communicator() +{ +#ifdef CUDF_STREAMING_HAVE_MPI + return "mpi"; +#elif defined(CUDF_STREAMING_HAVE_UCXX) + return "ucxx"; +#else + return {}; +#endif +} + +} // namespace cudf_streaming::benchmarks diff --git a/cpp/libcudf_streaming/cmake/ConfigureOptionalCommunication.cmake b/cpp/libcudf_streaming/cmake/ConfigureOptionalCommunication.cmake new file mode 100644 index 000000000000..1122e9618580 --- /dev/null +++ b/cpp/libcudf_streaming/cmake/ConfigureOptionalCommunication.cmake @@ -0,0 +1,114 @@ +# ============================================================================= +# cmake-format: off +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. +# SPDX-License-Identifier: Apache-2.0 +# cmake-format: on +# ============================================================================= + +include_guard(GLOBAL) + +set(CUDF_STREAMING_FOUND_MPI OFF) +set(CUDF_STREAMING_FOUND_UCXX OFF) +set(CUDF_STREAMING_HAVE_MPI OFF) +set(CUDF_STREAMING_HAVE_UCXX OFF) + +if(BUILD_BENCHMARKS OR BUILD_EXAMPLES) + rapids_find_package( + MPI QUIET + BUILD_EXPORT_SET cudf_streaming-exports + INSTALL_EXPORT_SET cudf_streaming-exports + ) + if(TARGET MPI::MPI_CXX) + set(CUDF_STREAMING_FOUND_MPI ON) + endif() +endif() + +if(BUILD_BENCHMARKS) + rapids_find_package( + ucxx QUIET + BUILD_EXPORT_SET cudf_streaming-exports + INSTALL_EXPORT_SET cudf_streaming-exports + ) + if(TARGET ucxx::ucxx) + set(CUDF_STREAMING_FOUND_UCXX ON) + endif() +endif() + +# Check whether rapidsmpf exports a compile definition for an optional feature. +function(cudf_streaming_rapidsmpf_exports_feature feature result) + set(has_feature OFF) + get_target_property( + rapidsmpf_compile_definitions rapidsmpf::rapidsmpf INTERFACE_COMPILE_DEFINITIONS + ) + if(rapidsmpf_compile_definitions) + foreach(definition IN LISTS rapidsmpf_compile_definitions) + if(definition STREQUAL "${feature}") + set(has_feature ON) + elseif(definition MATCHES "^\\$<\\$]*)>:${feature}>$") + set(feature_condition "${CMAKE_MATCH_1}") + if(feature_condition) + set(has_feature ON) + endif() + endif() + endforeach() + endif() + set(${result} + "${has_feature}" + PARENT_SCOPE + ) +endfunction() + +# Configure the optional communication target after rapidsmpf reports exported feature support. +function(cudf_streaming_configure_optional_communication) + set(CUDF_STREAMING_HAVE_MPI OFF) + set(CUDF_STREAMING_HAVE_UCXX OFF) + set(CUDF_STREAMING_HAVE_COMM OFF) + + cudf_streaming_rapidsmpf_exports_feature(RAPIDSMPF_HAVE_MPI rapidsmpf_have_mpi) + cudf_streaming_rapidsmpf_exports_feature(RAPIDSMPF_HAVE_UCXX rapidsmpf_have_ucxx) + + if(CUDF_STREAMING_FOUND_MPI AND rapidsmpf_have_mpi) + set(CUDF_STREAMING_HAVE_MPI ON) + endif() + if(CUDF_STREAMING_FOUND_UCXX AND rapidsmpf_have_ucxx) + set(CUDF_STREAMING_HAVE_UCXX ON) + endif() + if(CUDF_STREAMING_HAVE_MPI OR CUDF_STREAMING_HAVE_UCXX) + set(CUDF_STREAMING_HAVE_COMM ON) + endif() + + add_library(cudf_streaming_optional_communication INTERFACE) + target_compile_definitions( + cudf_streaming_optional_communication + INTERFACE $<$:CUDF_STREAMING_HAVE_MPI> + $<$:CUDF_STREAMING_HAVE_UCXX> + ) + if(CUDF_STREAMING_HAVE_MPI) + target_link_libraries(cudf_streaming_optional_communication INTERFACE MPI::MPI_CXX) + endif() + if(CUDF_STREAMING_HAVE_UCXX) + target_link_libraries(cudf_streaming_optional_communication INTERFACE ucxx::ucxx) + endif() + add_library(cudf_streaming::optional_communication ALIAS cudf_streaming_optional_communication) + + message(STATUS "CUDF_STREAMING: MPI support: ${CUDF_STREAMING_HAVE_MPI}") + message(STATUS "CUDF_STREAMING: UCXX support: ${CUDF_STREAMING_HAVE_UCXX}") + + set(CUDF_STREAMING_HAVE_MPI + "${CUDF_STREAMING_HAVE_MPI}" + PARENT_SCOPE + ) + set(CUDF_STREAMING_HAVE_UCXX + "${CUDF_STREAMING_HAVE_UCXX}" + PARENT_SCOPE + ) + set(CUDF_STREAMING_HAVE_COMM + "${CUDF_STREAMING_HAVE_COMM}" + PARENT_SCOPE + ) +endfunction() + +# Emit a standardized status message for targets disabled by unavailable optional dependencies. +function(cudf_streaming_skip_optional_target target reason) + message(STATUS "CUDF_STREAMING: Skipping ${target}: ${reason}") +endfunction() diff --git a/cpp/libcudf_streaming/cmake/thirdparty/get_rapidsmpf.cmake b/cpp/libcudf_streaming/cmake/thirdparty/get_rapidsmpf.cmake index 37a25b126e13..4f8516f2a111 100644 --- a/cpp/libcudf_streaming/cmake/thirdparty/get_rapidsmpf.cmake +++ b/cpp/libcudf_streaming/cmake/thirdparty/get_rapidsmpf.cmake @@ -8,9 +8,13 @@ # This function finds rapidsmpf and sets any additional necessary environment variables. function(find_and_configure_rapidsmpf VERSION) rapids_cmake_parse_version(MAJOR_MINOR ${VERSION} major_minor) - set(rapidsmpf_build_comm_support OFF) - if(BUILD_BENCHMARKS) - set(rapidsmpf_build_comm_support ON) + set(rapidsmpf_build_mpi_support OFF) + set(rapidsmpf_build_ucxx_support OFF) + if(CUDF_STREAMING_FOUND_MPI) + set(rapidsmpf_build_mpi_support ON) + endif() + if(CUDF_STREAMING_FOUND_UCXX) + set(rapidsmpf_build_ucxx_support ON) endif() rapids_cpm_find( rapidsmpf ${VERSION} @@ -20,8 +24,8 @@ function(find_and_configure_rapidsmpf VERSION) GIT_REPOSITORY https://github.com/rapidsai/rapidsmpf.git GIT_TAG "${RAPIDS_BRANCH}" GIT_SHALLOW TRUE SOURCE_SUBDIR cpp - OPTIONS "BUILD_MPI_SUPPORT ${rapidsmpf_build_comm_support}" - "BUILD_UCXX_SUPPORT ${rapidsmpf_build_comm_support}" + OPTIONS "BUILD_MPI_SUPPORT ${rapidsmpf_build_mpi_support}" + "BUILD_UCXX_SUPPORT ${rapidsmpf_build_ucxx_support}" "BUILD_SLURM_SUPPORT OFF" "BUILD_TESTS OFF" "BUILD_BENCHMARKS OFF" diff --git a/cpp/libcudf_streaming/examples/CMakeLists.txt b/cpp/libcudf_streaming/examples/CMakeLists.txt index 5d8a2ac7b053..bc957186e0df 100644 --- a/cpp/libcudf_streaming/examples/CMakeLists.txt +++ b/cpp/libcudf_streaming/examples/CMakeLists.txt @@ -5,25 +5,31 @@ # cmake-format: on # ============================================================================= -add_executable(example_shuffle "example_shuffle.cpp") -set_target_properties( - example_shuffle - PROPERTIES RUNTIME_OUTPUT_DIRECTORY "${CUDF_STREAMING_BINARY_DIR}/examples" - CXX_STANDARD 20 - CXX_STANDARD_REQUIRED ON - CUDA_STANDARD 20 - CUDA_STANDARD_REQUIRED ON -) -target_link_libraries( - example_shuffle PRIVATE cudf_streaming rapidsmpf::rapidsmpf $ - $ -) -target_sources( - example_shuffle PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/../benchmarks/utils/random_data.cu" -) -install( - TARGETS example_shuffle - COMPONENT examples - DESTINATION bin/examples/libcudf_streaming - EXCLUDE_FROM_ALL -) +if(CUDF_STREAMING_HAVE_MPI) + add_executable(example_shuffle "example_shuffle.cpp") + set_target_properties( + example_shuffle + PROPERTIES RUNTIME_OUTPUT_DIRECTORY "${CUDF_STREAMING_BINARY_DIR}/examples" + CXX_STANDARD 20 + CXX_STANDARD_REQUIRED ON + CUDA_STANDARD 20 + CUDA_STANDARD_REQUIRED ON + ) + target_link_libraries( + example_shuffle + PRIVATE cudf_streaming rapidsmpf::rapidsmpf cudf_streaming::optional_communication + $ + ) + target_compile_options(example_shuffle PRIVATE $<$:--expt-extended-lambda>) + target_sources( + example_shuffle PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/../benchmarks/utils/random_data.cu" + ) + install( + TARGETS example_shuffle + COMPONENT examples + DESTINATION bin/examples/libcudf_streaming + EXCLUDE_FROM_ALL + ) +else() + cudf_streaming_skip_optional_target(example_shuffle "requires MPI") +endif()