diff --git a/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp b/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp index f8fa5fe19b2b..0fbab9c2777b 100644 --- a/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp +++ b/cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -516,6 +517,8 @@ int main(int argc, char** argv) auto& stat_enabled_mr = br->device_mr_adaptor(); rmm::mr::set_current_device_resource(stat_enabled_mr); + auto log = rapidsmpf::Logger::from_options(options); + std::shared_ptr comm; auto progress_thread = std::make_shared(stats); if (args.comm_type == "mpi") { @@ -527,7 +530,7 @@ int main(int argc, char** argv) return 1; } rapidsmpf::mpi::init(&argc, &argv); - comm = std::make_shared(MPI_COMM_WORLD, options, progress_thread); + comm = std::make_shared(MPI_COMM_WORLD, progress_thread, log); #else std::cerr << "Error: MPI communicator is not available in this build." << std::endl; return 1; @@ -537,11 +540,11 @@ int main(int argc, char** argv) if (use_bootstrap) { // Launched with rrun - use bootstrap backend comm = rapidsmpf::bootstrap::create_ucxx_comm( - progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options); + progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options, log); } else { #ifdef CUDF_STREAMING_HAVE_MPI // Launched with mpirun - use MPI bootstrap - comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); + comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread, log); #else std::cerr << "Error: UCXX without MPI support requires bootstrap mode." << std::endl; return 1; @@ -558,7 +561,6 @@ int main(int argc, char** argv) args.pprint(*comm); - auto& log = comm->logger(); rmm::cuda_stream_view stream = cudf::get_default_stream(); // Print benchmark/hardware info. diff --git a/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp b/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp index d15d5501db46..511b9a022135 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -276,6 +277,7 @@ int main(int argc, char** argv) // Initialize configuration options from environment variables. rapidsmpf::config::Options options{rapidsmpf::config::get_environment_variables()}; + auto log = rapidsmpf::Logger::from_options(options); auto progress_thread = std::make_shared(); std::shared_ptr comm; @@ -288,7 +290,7 @@ int main(int argc, char** argv) return 1; } rapidsmpf::mpi::init(&argc, &argv); - comm = std::make_shared(MPI_COMM_WORLD, options, progress_thread); + comm = std::make_shared(MPI_COMM_WORLD, progress_thread, log); #else std::cerr << "Error: MPI communicator is not available in this build." << std::endl; return 1; @@ -298,11 +300,11 @@ int main(int argc, char** argv) if (use_bootstrap) { // Launched with rrun - use bootstrap backend comm = rapidsmpf::bootstrap::create_ucxx_comm( - progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options); + progress_thread, rapidsmpf::bootstrap::BackendType::AUTO, options, log); } else { #ifdef CUDF_STREAMING_HAVE_MPI // Launched with mpirun - use MPI bootstrap - comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); + comm = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread, log); #else std::cerr << "Error: UCXX without MPI support requires bootstrap mode." << std::endl; return 1; @@ -344,7 +346,6 @@ int main(int argc, char** argv) auto& stat_enabled_mr = br->device_mr_adaptor(); rmm::mr::set_current_device_resource(stat_enabled_mr); - auto& log = *comm->logger(); rmm::cuda_stream_view stream = cudf::get_default_stream(); // Print benchmark/hardware info. @@ -361,7 +362,7 @@ int main(int argc, char** argv) ss << " PCI Bus ID: " << pci_bus_id.substr(0, pci_bus_id.find('\0')) << "\n"; ss << " Total Memory: " << rapidsmpf::format_nbytes(properties.totalGlobalMem, 0) << "\n"; ss << " Comm: " << *comm << "\n"; - log.print(ss.str()); + log->print(ss.str()); } auto ctx = std::make_shared(options, comm->logger(), br); @@ -378,7 +379,7 @@ int main(int argc, char** argv) << "/s | global throughput: " << rapidsmpf::format_nbytes(args.total_nbytes / elapsed) << "/s"; if (i < args.num_warmups) { ss << " (warmup run)"; } - log.print(ss.str()); + log->print(ss.str()); if (i >= args.num_warmups) { elapsed_vec.push_back(elapsed); } } @@ -417,18 +418,18 @@ int main(int argc, char** argv) << rapidsmpf::format_nbytes(record.total() / static_cast(total_num_runs)) << " (avg)"; } - log.print(ss.str()); + log->print(ss.str()); } auto statistics = ctx->statistics(); if (args.enable_memory_profiler) { - log.print(statistics->report({ + log->print(statistics->report({ .mr = stat_enabled_mr, .pinned_mr = pinned_mr, .header = "Statistics (of the last run):", })); } else { - log.print(statistics->report({.header = "Statistics (of the last run):"})); + log->print(statistics->report({.header = "Statistics (of the last run):"})); } #ifdef CUDF_STREAMING_HAVE_MPI diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp index fdd2237447b8..c90f56e78dae 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp +++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -206,6 +207,7 @@ std::pair, std::shared_ptr> cr auto environment = config::get_environment_variables(); environment["NUM_STREAMING_THREADS"] = std::to_string(arguments.num_streaming_threads); auto options = config::Options(environment); + auto log = Logger::from_options(options); auto progress_thread = std::make_shared(statistics); std::shared_ptr comm; switch (arguments.comm_type) { @@ -214,20 +216,21 @@ std::pair, std::shared_ptr> cr 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); + comm = std::make_shared(MPI_COMM_WORLD, progress_thread, log); #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::SINGLE: comm = std::make_shared(progress_thread, log); 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); + comm = + bootstrap::create_ucxx_comm(progress_thread, bootstrap::BackendType::AUTO, options, log); } else { #ifdef CUDF_STREAMING_HAVE_MPI mpi::init(nullptr, nullptr); - comm = ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread); + comm = ucxx::init_using_mpi(MPI_COMM_WORLD, options, progress_thread, log); #else RAPIDSMPF_FAIL("UCXX without MPI support requires bootstrap mode", std::invalid_argument); #endif @@ -238,13 +241,13 @@ std::pair, std::shared_ptr> cr break; default: RAPIDSMPF_FAIL("Unknown communicator type"); } - auto ctx = std::make_shared(options, comm->logger(), br); + auto ctx = std::make_shared(options, log, br); if (comm->rank() == 0) { - comm->logger()->print("Execution context on ", - comm->nranks(), - " ranks has ", - ctx->executor()->num_streaming_threads(), - " threads"); + log->print("Execution context on ", + comm->nranks(), + " ranks has ", + ctx->executor()->num_streaming_threads(), + " threads"); } return {ctx, comm}; } diff --git a/cpp/libcudf_streaming/examples/example_shuffle.cpp b/cpp/libcudf_streaming/examples/example_shuffle.cpp index eb861c34a227..06eb54343a1a 100644 --- a/cpp/libcudf_streaming/examples/example_shuffle.cpp +++ b/cpp/libcudf_streaming/examples/example_shuffle.cpp @@ -8,6 +8,7 @@ #include #include +#include #include #include #include @@ -35,14 +36,13 @@ int main(int argc, char** argv) // single progress thread may be used by multiple shufflers simultaneously. auto progress_thread = std::make_shared(stats); + auto log = rapidsmpf::Logger::from_options(options); + // Now we have to create a Communicator, which we will use throughout the // example. Multiple concurrent shuffles are possible on the same communicator by // providing differentiating "OpID" arguments. std::shared_ptr comm = - std::make_shared(MPI_COMM_WORLD, options, progress_thread); - - // The Communicator provides a logger. - auto& log = comm->logger(); + std::make_shared(MPI_COMM_WORLD, progress_thread, log); // We will use the same stream, memory, and buffer resource throughout the example. rmm::cuda_stream_view stream = cudf::get_default_stream(); diff --git a/cpp/libcudf_streaming/tests/main/mpi.cpp b/cpp/libcudf_streaming/tests/main/mpi.cpp index ecfc4ce1cb12..eedf76af8039 100644 --- a/cpp/libcudf_streaming/tests/main/mpi.cpp +++ b/cpp/libcudf_streaming/tests/main/mpi.cpp @@ -8,6 +8,7 @@ #include #include +#include #include #include #include @@ -29,8 +30,9 @@ class MPIEnvironment : public Environment { options_ = rapidsmpf::config::Options(rapidsmpf::config::get_environment_variables()); - comm_ = std::make_shared( - mpi_comm_, options_, std::make_shared()); + comm_ = std::make_shared(mpi_comm_, + std::make_shared(), + rapidsmpf::Logger::from_options(options_)); } void TearDown() override @@ -45,16 +47,15 @@ class MPIEnvironment : public Environment { std::shared_ptr split_comm() override { - // Initialize configuration options from environment variables. - rapidsmpf::config::Options options{rapidsmpf::config::get_environment_variables()}; - // Create the new split communicator int rank; RAPIDSMPF_MPI(MPI_Comm_rank(mpi_comm_, &rank)); MPI_Comm split_comm = MPI_COMM_NULL; RAPIDSMPF_MPI(MPI_Comm_split(mpi_comm_, rank, 0, &split_comm)); + auto new_logger = rapidsmpf::Logger::from_options(options_); + new_logger->set_name(std::to_string(rank)); return std::shared_ptr( - new rapidsmpf::MPI(split_comm, options, comm_->progress_thread()), + new rapidsmpf::MPI(split_comm, comm_->progress_thread(), std::move(new_logger)), // Don't leak the split handle. [comm = split_comm](rapidsmpf::MPI* x) mutable { delete x; diff --git a/cpp/libcudf_streaming/tests/main/single.cpp b/cpp/libcudf_streaming/tests/main/single.cpp index cf54b83eca5a..542db6f54861 100644 --- a/cpp/libcudf_streaming/tests/main/single.cpp +++ b/cpp/libcudf_streaming/tests/main/single.cpp @@ -7,6 +7,7 @@ #include +#include #include #include #include @@ -23,8 +24,8 @@ class SingleEnvironment : public Environment { void SetUp() override { options_ = rapidsmpf::config::Options(rapidsmpf::config::get_environment_variables()); - comm_ = - std::make_shared(options_, std::make_shared()); + comm_ = std::make_shared(std::make_shared(), + rapidsmpf::Logger::from_options(options_)); } void TearDown() override { comm_ = nullptr; } diff --git a/cpp/libcudf_streaming/tests/main/ucxx.cpp b/cpp/libcudf_streaming/tests/main/ucxx.cpp index 51ad19551ce9..b50c974947f4 100644 --- a/cpp/libcudf_streaming/tests/main/ucxx.cpp +++ b/cpp/libcudf_streaming/tests/main/ucxx.cpp @@ -10,6 +10,7 @@ #include #include +#include #include #include #include @@ -39,8 +40,10 @@ class UCXXEnvironment : public Environment { "didn't get the requested thread level support: MPI_THREAD_MULTIPLE"); options_ = rapidsmpf::config::Options(rapidsmpf::config::get_environment_variables()); - comm_ = rapidsmpf::ucxx::init_using_mpi( - MPI_COMM_WORLD, options_, std::make_shared()); + comm_ = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD, + options_, + std::make_shared(), + rapidsmpf::Logger::from_options(options_)); } void TearDown() override