feat: add point-to-point communication for mccl

CMakeLists.txt

@@ -1,6 +1,8 @@
 cmake_minimum_required(VERSION 3.18)
 project(InfiniCCL VERSION 0.1.0 LANGUAGES C CXX)
 
+include(CTest)
+
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
@@ -507,6 +509,9 @@ add_subdirectory(src)
 if(BUILD_EXAMPLES)
     add_subdirectory(examples)
 endif()
+if(BUILD_TESTING)
+    add_subdirectory(tests)
+endif()
 
 # =========================================================
 # --- Installation ---

examples/ccl/send_recv.cc

@@ -0,0 +1,194 @@
+/**
+ * InfiniCCL Example: Thread-per-GPU Single-Node Send/Recv
+ *
+ * This example transfers data from GPU 0 to GPU 1 through InfiniCCL's native
+ * CCL backend without an MPI launcher.
+ */
+
+#include <unistd.h>
+
+#include <atomic>
+#include <charconv>
+#include <cstdlib>
+#include <iostream>
+#include <string_view>
+#include <thread>
+#include <vector>
+
+#include "backend_manifest.h"
+#include "infiniccl.h"
+#include "utils.h"
+
+using namespace infini::ccl;
+
+namespace {
+
+constexpr int kRankCount = 2;
+constexpr int kSender = 0;
+constexpr int kReceiver = 1;
+constexpr float kSendValue = 7.0f;
+
+struct ThreadArgs {
+  int rank;
+  infinicclUniqueId id;
+  size_t num_elements;
+  int warmup_iter;
+  int profile_iter;
+  std::atomic_bool* all_correct;
+};
+
+template <typename T>
+bool ParseNumber(const char* text, T* value) {
+  const std::string_view input{text};
+  T parsed{};
+  const auto [end, error] =
+      std::from_chars(input.data(), input.data() + input.size(), parsed);
+
+  if (error != std::errc{} || end != input.data() + input.size()) {
+    return false;
+  }
+
+  *value = parsed;
+  return true;
+}
+
+void WorkerThread(ThreadArgs args) {
+  constexpr Device::Type kDevType =
+      ListGetBest<DevicePriority>(EnabledDevices{});
+  using Rt = Runtime<kDevType>;
+
+  CHECK_RT(Rt, Rt::SetDevice(args.rank));
+
+  infinicclComm_t comm = nullptr;
+  CHECK_INFINI(infinicclCommInitRank(&comm, kRankCount, args.id, args.rank));
+
+  std::vector<float> host_buffer(args.num_elements,
+                                 args.rank == kSender ? kSendValue : 0.0f);
+  float* device_buffer = nullptr;
+  const size_t total_bytes = args.num_elements * sizeof(float);
+
+  CHECK_RT(Rt,
+           Rt::Malloc(reinterpret_cast<void**>(&device_buffer), total_bytes));
+  CHECK_RT(Rt, Rt::Memcpy(device_buffer, host_buffer.data(), total_bytes,
+                          Rt::MemcpyHostToDevice));
+  CHECK_RT(Rt, Rt::StreamSynchronize(nullptr));
+
+  auto exchange = [&]() {
+    if (args.rank == kSender) {
+      return infinicclSend(device_buffer, args.num_elements, infinicclFloat32,
+                           kReceiver, comm, nullptr);
+    }
+    return infinicclRecv(device_buffer, args.num_elements, infinicclFloat32,
+                         kSender, comm, nullptr);
+  };
+
+  for (int i = 0; i < args.warmup_iter; ++i) {
+    CHECK_INFINI(exchange());
+  }
+  CHECK_RT(Rt, Rt::StreamSynchronize(nullptr));
+
+  Timer timer;
+  for (int i = 0; i < args.profile_iter; ++i) {
+    CHECK_INFINI(exchange());
+  }
+  CHECK_RT(Rt, Rt::StreamSynchronize(nullptr));
+  const double elapsed =
+      timer.ElapsedMs() / static_cast<double>(args.profile_iter);
+
+  if (args.rank == kReceiver) {
+    CHECK_RT(Rt, Rt::Memcpy(host_buffer.data(), device_buffer, total_bytes,
+                            Rt::MemcpyDeviceToHost));
+    const bool correct =
+        Validator::ValidateResult(host_buffer.data(), args.num_elements,
+                                  kSendValue, kSender, true, "Send/Recv");
+    if (!correct) {
+      args.all_correct->store(false, std::memory_order_relaxed);
+    }
+  } else {
+    std::cout << "\n=== Single-Node Threaded Send/Recv Results ==="
+              << std::endl;
+    Metrics metrics{elapsed, total_bytes, kRankCount};
+    metrics.Print();
+  }
+
+  CHECK_RT(Rt, Rt::Free(device_buffer));
+  CHECK_INFINI(infinicclCommDestroy(comm));
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+  size_t num_elements = 1 << 20;
+  int warmup_iterations = 2;
+  int profile_iterations = 20;
+
+  int opt;
+  while ((opt = getopt(argc, argv, "n:w:p:h")) != -1) {
+    switch (opt) {
+      case 'n':
+        if (!ParseNumber(optarg, &num_elements)) {
+          std::cerr << "Invalid value for `-n`." << std::endl;
+          return EXIT_FAILURE;
+        }
+        break;
+      case 'w':
+        if (!ParseNumber(optarg, &warmup_iterations)) {
+          std::cerr << "Invalid value for `-w`." << std::endl;
+          return EXIT_FAILURE;
+        }
+        break;
+      case 'p':
+        if (!ParseNumber(optarg, &profile_iterations)) {
+          std::cerr << "Invalid value for `-p`." << std::endl;
+          return EXIT_FAILURE;
+        }
+        break;
+      case 'h':
+        std::cout << "Usage: " << argv[0] << " [options]\n"
+                  << "  -n <elements>     Elements to transfer (default: "
+                  << (1 << 20) << ")\n"
+                  << "  -w <iterations>   Warm-up iterations (default: 2)\n"
+                  << "  -p <iterations>   Profile iterations (default: 20)\n";
+        return EXIT_SUCCESS;
+      default:
+        return EXIT_FAILURE;
+    }
+  }
+
+  if (num_elements == 0 || warmup_iterations < 0 ||
+      profile_iterations <= 0) {
+    std::cerr << "Elements and profile iterations must be positive; warm-up "
+                 "iterations must be non-negative."
+              << std::endl;
+    return EXIT_FAILURE;
+  }
+
+  infinicclUniqueId shared_id;
+  CHECK_INFINI(infinicclGetUniqueId(&shared_id));
+
+  std::atomic_bool all_correct{true};
+  std::vector<std::thread> threads;
+  threads.reserve(kRankCount);
+
+  for (int rank = 0; rank < kRankCount; ++rank) {
+    ThreadArgs args{rank,
+                    shared_id,
+                    num_elements,
+                    warmup_iterations,
+                    profile_iterations,
+                    &all_correct};
+    threads.emplace_back(WorkerThread, args);
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+
+  if (!all_correct.load(std::memory_order_relaxed)) {
+    std::cerr << "Send/Recv validation failed." << std::endl;
+    return EXIT_FAILURE;
+  }
+
+  std::cout << "Send/Recv validation passed." << std::endl;
+  return EXIT_SUCCESS;
+}

src/CMakeLists.txt

@@ -16,6 +16,21 @@ set(AUTOGEN_WARNING [[/*
 
 file(GLOB CORE_SRCS "*.cc")
 file(GLOB_RECURSE BASE_IMPL_SRCS "base/*.cc")
+file(GLOB_RECURSE BRIDGE_DEP_HEADERS CONFIGURE_DEPENDS
+    "base/*.h"
+    "backends/*.h"
+    "devices/*.h"
+    "devices/*.cuh"
+)
+
+string(REPLACE ";" "\n" BRIDGE_DEPENDENCY_CONTENT "${BRIDGE_DEP_HEADERS}")
+set(BRIDGE_DEPENDENCY_MANIFEST
+    "${CMAKE_CURRENT_BINARY_DIR}/bridge_dependencies.txt"
+)
+file(GENERATE
+    OUTPUT "${BRIDGE_DEPENDENCY_MANIFEST}"
+    CONTENT "${BRIDGE_DEPENDENCY_CONTENT}\n"
+)
 
 target_sources(infiniccl PRIVATE
     ${CORE_SRCS}
@@ -298,7 +313,9 @@ add_custom_command(
             "${BACK_STR}"
     DEPENDS "${PROJECT_SOURCE_DIR}/scripts/gen_bridge.py"
             "${PROJECT_SOURCE_DIR}/include/comm.h"
+            "${BRIDGE_DEPENDENCY_MANIFEST}"
             ${BASE_IMPL_SRCS}
+            ${BRIDGE_DEP_HEADERS}
     VERBATIM
     COMMENT "Generating InfiniCCL bridge and manifest files for Devices: [${DEV_STR}] Backends: [${BACK_STR}]..."
 )

src/backends/ccl/common/impl/recv.h

@@ -0,0 +1,43 @@
+#ifndef INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_RECV_H_
+#define INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_RECV_H_
+
+#include "backends/ccl/common/api.h"
+#include "backends/ccl/common/comm_instance.h"
+#include "base/recv.h"
+#include "communicator.h"
+
+namespace infini::ccl {
+
+template <BackendType backend, Device::Type device>
+class CclRecvImpl {
+ public:
+  static ReturnStatus Apply(void* recv_buff, size_t count, DataType data_type,
+                            int peer, Communicator* comm, void* stream) {
+    using Api = CclApi<backend, device>;
+    using TypeMap = CclTypeMap<backend, device>;
+    using CommInstance = CclCommInstance<Api>;
+
+    if (!comm || !comm->intra_comm() || comm->intra_comm_backend() != backend ||
+        comm->device_type() != device) {
+      return ReturnStatus::kInternalError;
+    }
+
+    auto* intra = static_cast<CommInstance*>(comm->intra_comm());
+    if (!intra->handle) {
+      return ReturnStatus::kInternalError;
+    }
+
+    typename Api::DataType ccl_type{};
+    if (!TypeMap::ToBackendDataType(data_type, &ccl_type)) {
+      return ReturnStatus::kNotSupported;
+    }
+
+    return Api::Check(
+        Api::Recv(recv_buff, count, ccl_type, peer, intra->handle,
+                  reinterpret_cast<typename Api::Stream>(stream)));
+  }
+};
+
+}  // namespace infini::ccl
+
+#endif  // INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_RECV_H_

src/backends/ccl/common/impl/send.h

@@ -0,0 +1,44 @@
+#ifndef INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_SEND_H_
+#define INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_SEND_H_
+
+#include "backends/ccl/common/api.h"
+#include "backends/ccl/common/comm_instance.h"
+#include "base/send.h"
+#include "communicator.h"
+
+namespace infini::ccl {
+
+template <BackendType backend, Device::Type device>
+class CclSendImpl {
+ public:
+  static ReturnStatus Apply(const void* send_buff, size_t count,
+                            DataType data_type, int peer, Communicator* comm,
+                            void* stream) {
+    using Api = CclApi<backend, device>;
+    using TypeMap = CclTypeMap<backend, device>;
+    using CommInstance = CclCommInstance<Api>;
+
+    if (!comm || !comm->intra_comm() || comm->intra_comm_backend() != backend ||
+        comm->device_type() != device) {
+      return ReturnStatus::kInternalError;
+    }
+
+    auto* intra = static_cast<CommInstance*>(comm->intra_comm());
+    if (!intra->handle) {
+      return ReturnStatus::kInternalError;
+    }
+
+    typename Api::DataType ccl_type{};
+    if (!TypeMap::ToBackendDataType(data_type, &ccl_type)) {
+      return ReturnStatus::kNotSupported;
+    }
+
+    return Api::Check(
+        Api::Send(send_buff, count, ccl_type, peer, intra->handle,
+                  reinterpret_cast<typename Api::Stream>(stream)));
+  }
+};
+
+}  // namespace infini::ccl
+
+#endif  // INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_SEND_H_

src/backends/ccl/mccl/api.h

@@ -35,20 +35,30 @@ struct McclApi {
     return ReturnStatus::kSuccess;
   }
 
-  static Result GetUniqueId(UniqueId *id) { return mcclGetUniqueId(id); }
+  static Result GetUniqueId(UniqueId* id) { return mcclGetUniqueId(id); }
 
-  static Result CommInitRank(Comm *comm, int nranks, UniqueId id, int rank) {
+  static Result CommInitRank(Comm* comm, int nranks, UniqueId id, int rank) {
     return mcclCommInitRank(comm, nranks, id, rank);
   }
 
   static Result CommDestroy(Comm comm) { return mcclCommDestroy(comm); }
 
-  static Result AllReduce(const void *send_buff, void *recv_buff, size_t count,
+  static Result AllReduce(const void* send_buff, void* recv_buff, size_t count,
                           DataType data_type, RedOp op, Comm comm,
                           Stream stream) {
     return mcclAllReduce(send_buff, recv_buff, count, data_type, op, comm,
                          stream);
   }
+
+  static Result Send(const void* send_buff, size_t count, DataType data_type,
+                     int peer, Comm comm, Stream stream) {
+    return mcclSend(send_buff, count, data_type, peer, comm, stream);
+  }
+
+  static Result Recv(void* recv_buff, size_t count, DataType data_type,
+                     int peer, Comm comm, Stream stream) {
+    return mcclRecv(recv_buff, count, data_type, peer, comm, stream);
+  }
 };
 
 }  // namespace infini::ccl

src/backends/ccl/mccl/impl/recv.h

@@ -0,0 +1,17 @@
+#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_RECV_H_
+#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_RECV_H_
+
+#include "backends/ccl/common/impl/recv.h"
+
+namespace infini::ccl {
+
+template <Device::Type device>
+class RecvImpl<BackendType::kMccl, device>
+    : public CclRecvImpl<BackendType::kMccl, device> {};
+
+template <>
+struct BackendEnabled<Recv, BackendType::kMccl> : std::true_type {};
+
+}  // namespace infini::ccl
+
+#endif  // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_RECV_H_