Rapidsmpf logger API changes

cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp

@@ -8,6 +8,7 @@
 #include <rapidsmpf/bootstrap/bootstrap.hpp>
 #include <rapidsmpf/bootstrap/utils.hpp>
 #include <rapidsmpf/communicator/communicator.hpp>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/error.hpp>
 #include <rapidsmpf/memory/spill.hpp>
 #include <rapidsmpf/nvtx.hpp>
@@ -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<rapidsmpf::Communicator> comm;
   auto progress_thread = std::make_shared<rapidsmpf::ProgressThread>(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<rapidsmpf::MPI>(MPI_COMM_WORLD, options, progress_thread);
+    comm = std::make_shared<rapidsmpf::MPI>(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.

cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp

@@ -15,6 +15,7 @@
 #include <rapidsmpf/bootstrap/bootstrap.hpp>
 #include <rapidsmpf/bootstrap/utils.hpp>
 #include <rapidsmpf/communicator/communicator.hpp>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/error.hpp>
 #include <rapidsmpf/nvtx.hpp>
 #include <rapidsmpf/shuffler/shuffler.hpp>
@@ -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<rapidsmpf::ProgressThread>();
 
   std::shared_ptr<rapidsmpf::Communicator> comm;
@@ -288,7 +290,7 @@ int main(int argc, char** argv)
       return 1;
     }
     rapidsmpf::mpi::init(&argc, &argv);
-    comm = std::make_shared<rapidsmpf::MPI>(MPI_COMM_WORLD, options, progress_thread);
+    comm = std::make_shared<rapidsmpf::MPI>(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<rapidsmpf::streaming::Context>(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<std::int64_t>(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

cpp/libcudf_streaming/benchmarks/streaming/ndsh/utils.cpp

@@ -18,6 +18,7 @@
 #include <rapidsmpf/bootstrap/bootstrap.hpp>
 #include <rapidsmpf/bootstrap/utils.hpp>
 #include <rapidsmpf/communicator/communicator.hpp>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/communicator/single.hpp>
 #include <rapidsmpf/config.hpp>
 #include <rapidsmpf/error.hpp>
@@ -206,6 +207,7 @@ std::pair<std::shared_ptr<streaming::Context>, std::shared_ptr<Communicator>> 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<rapidsmpf::ProgressThread>(statistics);
   std::shared_ptr<Communicator> comm;
   switch (arguments.comm_type) {
@@ -214,20 +216,21 @@ std::pair<std::shared_ptr<streaming::Context>, std::shared_ptr<Communicator>> cr
       RAPIDSMPF_EXPECTS(!bootstrap::is_running_with_rrun(), "Can't use MPI communicator with rrun");
       mpi::init(nullptr, nullptr);
 
-      comm = std::make_shared<MPI>(MPI_COMM_WORLD, options, progress_thread);
+      comm = std::make_shared<MPI>(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<Single>(options, progress_thread); break;
+    case CommType::SINGLE: comm = std::make_shared<Single>(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<streaming::Context>, std::shared_ptr<Communicator>> cr
       break;
     default: RAPIDSMPF_FAIL("Unknown communicator type");
   }
-  auto ctx = std::make_shared<streaming::Context>(options, comm->logger(), br);
+  auto ctx = std::make_shared<streaming::Context>(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};
 }

cpp/libcudf_streaming/examples/example_shuffle.cpp

@@ -8,6 +8,7 @@
 #include <cudf_streaming/partition_utils.hpp>
 
 #include <mpi.h>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/communicator/mpi.hpp>
 #include <rapidsmpf/error.hpp>
 #include <rapidsmpf/memory/packed_data.hpp>
@@ -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<rapidsmpf::ProgressThread>(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<rapidsmpf::Communicator> comm =
-    std::make_shared<rapidsmpf::MPI>(MPI_COMM_WORLD, options, progress_thread);
-
-  // The Communicator provides a logger.
-  auto& log = comm->logger();
+    std::make_shared<rapidsmpf::MPI>(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();

cpp/libcudf_streaming/tests/main/mpi.cpp

@@ -8,6 +8,7 @@
 #include <cudf_test/cudf_gtest.hpp>
 
 #include <mpi.h>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/communicator/mpi.hpp>
 #include <rapidsmpf/config.hpp>
 #include <rapidsmpf/progress_thread.hpp>
@@ -29,8 +30,9 @@ class MPIEnvironment : public Environment {
 
     options_ = rapidsmpf::config::Options(rapidsmpf::config::get_environment_variables());
 
-    comm_ = std::make_shared<rapidsmpf::MPI>(
-      mpi_comm_, options_, std::make_shared<rapidsmpf::ProgressThread>());
+    comm_ = std::make_shared<rapidsmpf::MPI>(mpi_comm_,
+                                             std::make_shared<rapidsmpf::ProgressThread>(),
+                                             rapidsmpf::Logger::from_options(options_));
   }
 
   void TearDown() override
@@ -45,16 +47,15 @@ class MPIEnvironment : public Environment {
 
   std::shared_ptr<rapidsmpf::Communicator> 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<rapidsmpf::MPI>(
-      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;

cpp/libcudf_streaming/tests/main/single.cpp

@@ -7,6 +7,7 @@
 
 #include <cudf_test/cudf_gtest.hpp>
 
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/communicator/single.hpp>
 #include <rapidsmpf/config.hpp>
 #include <rapidsmpf/progress_thread.hpp>
@@ -23,8 +24,8 @@ class SingleEnvironment : public Environment {
   void SetUp() override
   {
     options_ = rapidsmpf::config::Options(rapidsmpf::config::get_environment_variables());
-    comm_ =
-      std::make_shared<rapidsmpf::Single>(options_, std::make_shared<rapidsmpf::ProgressThread>());
+    comm_    = std::make_shared<rapidsmpf::Single>(std::make_shared<rapidsmpf::ProgressThread>(),
+                                                rapidsmpf::Logger::from_options(options_));
   }
 
   void TearDown() override { comm_ = nullptr; }

cpp/libcudf_streaming/tests/main/ucxx.cpp

@@ -10,6 +10,7 @@
 #include <cuda_runtime_api.h>
 
 #include <mpi.h>
+#include <rapidsmpf/communicator/logger.hpp>
 #include <rapidsmpf/communicator/mpi.hpp>
 #include <rapidsmpf/communicator/ucxx_utils.hpp>
 #include <rapidsmpf/config.hpp>
@@ -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<rapidsmpf::ProgressThread>());
+    comm_    = rapidsmpf::ucxx::init_using_mpi(MPI_COMM_WORLD,
+                                            options_,
+                                            std::make_shared<rapidsmpf::ProgressThread>(),
+                                            rapidsmpf::Logger::from_options(options_));
   }
 
   void TearDown() override