From 77dcdddeaa388b198e1b249d0a5131cd2dee8318 Mon Sep 17 00:00:00 2001 From: lqinfdim <183612562+lqinfdim@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:37:29 +0800 Subject: [PATCH] feat: add collective communication for mccl --- CMakeLists.txt | 4 + examples/ccl/collectives.cc | 160 +++++++++++++++++++ src/backends/ccl/common/impl/collectives.h | 161 ++++++++++++++++++++ src/backends/ccl/mccl/api.h | 45 ++++++ src/backends/ccl/mccl/impl/all_gather.h | 18 +++ src/backends/ccl/mccl/impl/all_to_all.h | 18 +++ src/backends/ccl/mccl/impl/broadcast.h | 18 +++ src/backends/ccl/mccl/impl/gather.h | 18 +++ src/backends/ccl/mccl/impl/reduce.h | 18 +++ src/backends/ccl/mccl/impl/reduce_scatter.h | 18 +++ src/backends/ccl/mccl/impl/scatter.h | 18 +++ tests/CMakeLists.txt | 19 +++ tests/ccl_collectives_impl.cc | 89 +++++++++++ 13 files changed, 604 insertions(+) create mode 100644 examples/ccl/collectives.cc create mode 100644 src/backends/ccl/common/impl/collectives.h create mode 100644 src/backends/ccl/mccl/impl/all_gather.h create mode 100644 src/backends/ccl/mccl/impl/all_to_all.h create mode 100644 src/backends/ccl/mccl/impl/broadcast.h create mode 100644 src/backends/ccl/mccl/impl/gather.h create mode 100644 src/backends/ccl/mccl/impl/reduce.h create mode 100644 src/backends/ccl/mccl/impl/reduce_scatter.h create mode 100644 src/backends/ccl/mccl/impl/scatter.h create mode 100644 tests/CMakeLists.txt create mode 100644 tests/ccl_collectives_impl.cc diff --git a/CMakeLists.txt b/CMakeLists.txt index ebaca2c..5288092 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,5 +1,6 @@ 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) @@ -23,6 +24,9 @@ option(WITH_OMPI "Enable OpenMPI backend" OFF) option(WITH_MPICH "Enable MPICH backend" OFF) option(WITH_NCCL "Enable NCCL backend" OFF) option(WITH_MCCL "Enable MCCL backend" OFF) +if(BUILD_TESTING) + add_subdirectory(tests) +endif() # ========================================================= # --- MISC. BUILD OPTIONS --- diff --git a/examples/ccl/collectives.cc b/examples/ccl/collectives.cc new file mode 100644 index 0000000..9f51ede --- /dev/null +++ b/examples/ccl/collectives.cc @@ -0,0 +1,160 @@ +/** + * InfiniCCL Example: Thread-per-GPU Single-Node MCCL Broadcast + * + * This example demonstrates spawning one CPU thread per GPU on a single node, + * generating a shared UniqueID, and performing a Broadcast entirely via + * InfiniCCL's native CCL backend. + * + * Note: to properly run this example, you should specify `--launcher none` + * since it requires no MPI process spawning. + */ + +#include + +#include +#include +#include +#include + +// Public API +#include "infiniccl.h" + +// Example-Specific Utilities +#include "utils.h" + +// Internal Headers (Accessible via example-specific include paths, technically +// not public APIs) +#include "backend_manifest.h" + +using namespace infini::ccl; + +// Structure to pass execution data to each GPU worker thread. +struct ThreadArgs { + int rank; + int size; + infinicclUniqueId id; + size_t num_elements; + int warmup_iter; + int profile_iter; +}; + +// Worker function executed by each CPU thread. +void WorkerThread(ThreadArgs args) { + constexpr Device::Type kDevType = + ListGetBest(EnabledDevices{}); + using Rt = Runtime; + + // Bind this specific CPU thread to its designated local GPU device + // In a thread-per-GPU model, `local_rank` is exactly the thread's rank ID. + int local_device_id = args.rank; + CHECK_RT(Rt, Rt::SetDevice(local_device_id)); + + infinicclComm_t comm = nullptr; + CHECK_INFINI(infinicclCommInitRank(&comm, args.size, args.id, args.rank)); + + // Prepare Host Data Structures + std::vector h_send(args.num_elements); + std::vector h_recv(args.num_elements, 0.0f); + + // Each rank provides its own (rank + 1) as data. + for (size_t i = 0; i < args.num_elements; i++) { + h_send[i] = static_cast(args.rank + 1); + } + + // Allocate GPU Memory using InfiniCCL's Runtime abstraction layer. + float *d_send = nullptr; + float *d_recv = nullptr; + size_t total_bytes = args.num_elements * sizeof(float); + + CHECK_RT(Rt, Rt::Malloc((void **)&d_send, total_bytes)); + CHECK_RT(Rt, Rt::Malloc((void **)&d_recv, total_bytes)); + + CHECK_RT(Rt, Rt::Memcpy(d_send, h_send.data(), total_bytes, + Rt::MemcpyHostToDevice)); + CHECK_RT(Rt, Rt::Memcpy(d_recv, h_recv.data(), total_bytes, + Rt::MemcpyHostToDevice)); + + CHECK_RT(Rt, Rt::StreamSynchronize(nullptr)); + + // Warm-up Iterations + for (int i = 0; i < args.warmup_iter; ++i) { + CHECK_INFINI(infinicclBroadcast(d_send, d_recv, args.num_elements, + infinicclFloat32, 0, comm, + nullptr)); + } + CHECK_RT(Rt, Rt::StreamSynchronize(nullptr)); + + // Profiling Iterations + Timer timer; + for (int i = 0; i < args.profile_iter; i++) { + CHECK_INFINI(infinicclBroadcast(d_send, d_recv, args.num_elements, + infinicclFloat32, 0, comm, + nullptr)); + } + CHECK_RT(Rt, Rt::StreamSynchronize(nullptr)); + double elapsed = timer.ElapsedMs() / static_cast(args.profile_iter); + + // Copy broadcast data back from device to host. + CHECK_RT(Rt, Rt::Memcpy(h_recv.data(), d_recv, total_bytes, + Rt::MemcpyDeviceToHost)); + + // Result Validation + constexpr float expected = 1.0f; + Validator::ValidateResult(h_recv.data(), args.num_elements, expected, + args.rank, true, "Broadcast"); + + // Metrics Reporting (Only Rank 0) + if (args.rank == 0) { + std::cout << "\n=== Single-Node Threaded MCCL Broadcast Results ===" + << std::endl; + std::cout << "Data size: " << args.num_elements << " floats (" + << total_bytes / 1024 / 1024 << " MB)" << std::endl; + Metrics metrics{elapsed, total_bytes, args.size}; + metrics.Print(); + } + + // Cleanup local rank resources. + CHECK_RT(Rt, Rt::Free(d_send)); + CHECK_RT(Rt, Rt::Free(d_recv)); + CHECK_INFINI(infinicclCommDestroy(comm)); +} + +int main(int argc, char **argv) { + int num_gpus = 8; + int warmup_iters = 1; + int profile_iters = 20; + size_t num_elements = 1 << 25; + + (void)argc; + (void)argv; + + char hostname[256]; + gethostname(hostname, sizeof(hostname)); + std::cout << "[Main Process] Host: " << hostname + << " | Target GPUs: " << num_gpus << std::endl; + + infinicclUniqueId shared_id; + CHECK_INFINI(infinicclGetUniqueId(&shared_id)); + + // Spawn CPU thread pool. + std::vector threads; + threads.reserve(num_gpus); + + for (int rank = 0; rank < num_gpus; ++rank) { + ThreadArgs args{rank, num_gpus, shared_id, + num_elements, warmup_iters, profile_iters}; + threads.emplace_back(WorkerThread, args); + } + + // Await execution completion across all threads. + for (auto &t : threads) { + if (t.joinable()) { + t.join(); + } + } + + std::cout + << "[Main Process] All worker threads joined. InfiniCCL finalized safely." + << std::endl; + return EXIT_SUCCESS; +} diff --git a/src/backends/ccl/common/impl/collectives.h b/src/backends/ccl/common/impl/collectives.h new file mode 100644 index 0000000..39b1a0c --- /dev/null +++ b/src/backends/ccl/common/impl/collectives.h @@ -0,0 +1,161 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_COLLECTIVES_H_ +#define INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_COLLECTIVES_H_ + +#include "backends/ccl/common/api.h" +#include "device.h" +#include "traits.h" +#include "data_type_impl.h" +#include "return_status_impl.h" +#include "base/all_gather.h" +#include "base/all_reduce.h" +#include "base/all_to_all.h" +#include "base/broadcast.h" +#include "base/gather.h" +#include "base/reduce.h" +#include "base/reduce_scatter.h" +#include "base/scatter.h" +#include "backends/ccl/common/comm_instance.h" +#include "communicator.h" + +namespace infini::ccl { + +template +struct CclCollective { + using Api = CclApi; + using TypeMap = CclTypeMap; + using CommInstance = CclCommInstance; + + static CommInstance *Get(Communicator *comm) { + auto *instance = static_cast(comm->intra_comm()); + return instance && instance->handle ? instance : nullptr; + } + + static bool ToDataType(DataType value, typename Api::DataType *result) { + return TypeMap::ToBackendDataType(value, result); + } + + static bool ToReduction(ReductionOpType value, typename Api::RedOp *result) { + return TypeMap::ToBackendRedOp(value, result); + } +}; + +template +struct CclBroadcastImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, int root, Communicator *comm, + void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type)) return ReturnStatus::kNotSupported; + return C::Api::Check(C::Api::Broadcast(send_buff, recv_buff, count, + backend_type, root, instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclReduceImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, ReductionOpType op, int root, + Communicator *comm, void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + typename C::Api::RedOp backend_op{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type) || !C::ToReduction(op, &backend_op)) { + return ReturnStatus::kNotSupported; + } + return C::Api::Check(C::Api::Reduce(send_buff, recv_buff, count, backend_type, + backend_op, root, instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclReduceScatterImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, ReductionOpType op, Communicator *comm, + void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + typename C::Api::RedOp backend_op{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type) || !C::ToReduction(op, &backend_op)) { + return ReturnStatus::kNotSupported; + } + return C::Api::Check(C::Api::ReduceScatter(send_buff, recv_buff, count, + backend_type, backend_op, + instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclAllGatherImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, Communicator *comm, void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type)) return ReturnStatus::kNotSupported; + return C::Api::Check(C::Api::AllGather(send_buff, recv_buff, count, + backend_type, instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclGatherImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, int root, Communicator *comm, + void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type)) return ReturnStatus::kNotSupported; + return C::Api::Check(C::Api::Gather(send_buff, recv_buff, count, backend_type, + root, instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclScatterImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, int root, Communicator *comm, + void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type)) return ReturnStatus::kNotSupported; + return C::Api::Check(C::Api::Scatter(send_buff, recv_buff, count, backend_type, + root, instance->handle, + reinterpret_cast(stream))); + } +}; + +template +struct CclAllToAllImpl { + static ReturnStatus Apply(const void *send_buff, void *recv_buff, size_t count, + DataType type, Communicator *comm, void *stream) { + using C = CclCollective; + auto *instance = C::Get(comm); + typename C::Api::DataType backend_type{}; + if (!instance) return ReturnStatus::kInternalError; + if (!C::ToDataType(type, &backend_type)) return ReturnStatus::kNotSupported; + return C::Api::Check(C::Api::AllToAll(send_buff, recv_buff, count, + backend_type, instance->handle, + reinterpret_cast(stream))); + } +}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_COMMON_IMPL_COLLECTIVES_H_ diff --git a/src/backends/ccl/mccl/api.h b/src/backends/ccl/mccl/api.h index a5d2bcd..39de7f2 100644 --- a/src/backends/ccl/mccl/api.h +++ b/src/backends/ccl/mccl/api.h @@ -49,6 +49,51 @@ struct McclApi { return mcclAllReduce(send_buff, recv_buff, count, data_type, op, comm, stream); } + static Result Broadcast(const void *send_buff, void *recv_buff, size_t count, + DataType data_type, int root, Comm comm, + Stream stream) { + return mcclBroadcast(send_buff, recv_buff, count, data_type, root, comm, + stream); + } + + static Result Reduce(const void *send_buff, void *recv_buff, size_t count, + DataType data_type, RedOp op, int root, Comm comm, + Stream stream) { + return mcclReduce(send_buff, recv_buff, count, data_type, op, root, comm, + stream); + } + + static Result ReduceScatter(const void *send_buff, void *recv_buff, + size_t recv_count, DataType data_type, RedOp op, + Comm comm, Stream stream) { + return mcclReduceScatter(send_buff, recv_buff, recv_count, data_type, op, + comm, stream); + } + + static Result AllGather(const void *send_buff, void *recv_buff, + size_t send_count, DataType data_type, Comm comm, + Stream stream) { + return mcclAllGather(send_buff, recv_buff, send_count, data_type, comm, + stream); + } + + static Result Gather(const void *send_buff, void *recv_buff, size_t send_count, + DataType data_type, int root, Comm comm, Stream stream) { + return mcclGather(send_buff, recv_buff, send_count, data_type, root, comm, + stream); + } + + static Result Scatter(const void *send_buff, void *recv_buff, + size_t recv_count, DataType data_type, int root, + Comm comm, Stream stream) { + return mcclScatter(send_buff, recv_buff, recv_count, data_type, root, comm, + stream); + } + + static Result AllToAll(const void *send_buff, void *recv_buff, size_t count, + DataType data_type, Comm comm, Stream stream) { + return mcclAllToAll(send_buff, recv_buff, count, data_type, comm, stream); + } }; } // namespace infini::ccl diff --git a/src/backends/ccl/mccl/impl/all_gather.h b/src/backends/ccl/mccl/impl/all_gather.h new file mode 100644 index 0000000..d003414 --- /dev/null +++ b/src/backends/ccl/mccl/impl/all_gather.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_GATHER_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_GATHER_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/all_gather.h" + +namespace infini::ccl { + +template +struct AllGatherImpl + : CclAllGatherImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_GATHER_H_ diff --git a/src/backends/ccl/mccl/impl/all_to_all.h b/src/backends/ccl/mccl/impl/all_to_all.h new file mode 100644 index 0000000..27ffb08 --- /dev/null +++ b/src/backends/ccl/mccl/impl/all_to_all.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_TO_ALL_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_TO_ALL_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/all_to_all.h" + +namespace infini::ccl { + +template +struct AllToAllImpl + : CclAllToAllImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_ALL_TO_ALL_H_ diff --git a/src/backends/ccl/mccl/impl/broadcast.h b/src/backends/ccl/mccl/impl/broadcast.h new file mode 100644 index 0000000..d4ef14a --- /dev/null +++ b/src/backends/ccl/mccl/impl/broadcast.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_BROADCAST_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_BROADCAST_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/broadcast.h" + +namespace infini::ccl { + +template +struct BroadcastImpl + : CclBroadcastImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_BROADCAST_H_ diff --git a/src/backends/ccl/mccl/impl/gather.h b/src/backends/ccl/mccl/impl/gather.h new file mode 100644 index 0000000..672784a --- /dev/null +++ b/src/backends/ccl/mccl/impl/gather.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_GATHER_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_GATHER_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/gather.h" + +namespace infini::ccl { + +template +struct GatherImpl + : CclGatherImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_GATHER_H_ diff --git a/src/backends/ccl/mccl/impl/reduce.h b/src/backends/ccl/mccl/impl/reduce.h new file mode 100644 index 0000000..3a23cfd --- /dev/null +++ b/src/backends/ccl/mccl/impl/reduce.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/reduce.h" + +namespace infini::ccl { + +template +struct ReduceImpl + : CclReduceImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_H_ diff --git a/src/backends/ccl/mccl/impl/reduce_scatter.h b/src/backends/ccl/mccl/impl/reduce_scatter.h new file mode 100644 index 0000000..522ef5f --- /dev/null +++ b/src/backends/ccl/mccl/impl/reduce_scatter.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_SCATTER_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_SCATTER_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/reduce_scatter.h" + +namespace infini::ccl { + +template +struct ReduceScatterImpl + : CclReduceScatterImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_REDUCE_SCATTER_H_ diff --git a/src/backends/ccl/mccl/impl/scatter.h b/src/backends/ccl/mccl/impl/scatter.h new file mode 100644 index 0000000..59d3b8d --- /dev/null +++ b/src/backends/ccl/mccl/impl/scatter.h @@ -0,0 +1,18 @@ +#ifndef INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_SCATTER_H_ +#define INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_SCATTER_H_ + +#include "backends/ccl/common/impl/collectives.h" +#include "base/scatter.h" + +namespace infini::ccl { + +template +struct ScatterImpl + : CclScatterImpl {}; + +template <> +struct BackendEnabled : std::true_type {}; + +} // namespace infini::ccl + +#endif // INFINI_CCL_BACKENDS_CCL_MCCL_IMPL_SCATTER_H_ diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt new file mode 100644 index 0000000..ad0bf37 --- /dev/null +++ b/tests/CMakeLists.txt @@ -0,0 +1,19 @@ +if(NOT WITH_MCCL) + return() +endif() + +add_executable(ccl_collectives_impl_test + ccl_collectives_impl.cc +) + +target_include_directories(ccl_collectives_impl_test PRIVATE + "${PROJECT_SOURCE_DIR}/include" + "${PROJECT_SOURCE_DIR}/src" + "${PROJECT_BINARY_DIR}/src" +) +target_compile_features(ccl_collectives_impl_test PRIVATE cxx_std_17) + +add_test( + NAME ccl_collectives_impl + COMMAND ccl_collectives_impl_test +) diff --git a/tests/ccl_collectives_impl.cc b/tests/ccl_collectives_impl.cc new file mode 100644 index 0000000..8bf0d4c --- /dev/null +++ b/tests/ccl_collectives_impl.cc @@ -0,0 +1,89 @@ +#include +#include +#include + +#include "backends/ccl/common/impl/collectives.h" + +namespace infini::ccl { + +struct FakeMcclApi { + static constexpr BackendType kBackendType = BackendType::kMccl; + using Comm = int; + using DataType = int; + using RedOp = int; + using Result = int; + using Stream = void*; + static constexpr Result kSuccess = 0; + static constexpr Result kFailure = 1; + struct Call { const void* send; void* recv; size_t count; DataType type; RedOp op; int root; Comm comm; Stream stream; }; + static Call call; + static Result next_result; + static Result CommDestroy(Comm) { return kSuccess; } + static void Reset() { call = {}; next_result = kSuccess; } + static ReturnStatus Check(Result result) { return result == kSuccess ? ReturnStatus::kSuccess : ReturnStatus::kSystemError; } + static Result Broadcast(const void* s, void* r, size_t n, DataType t, int root, Comm c, Stream x) { call = {s,r,n,t,0,root,c,x}; return next_result; } + static Result Reduce(const void* s, void* r, size_t n, DataType t, RedOp o, int root, Comm c, Stream x) { call = {s,r,n,t,o,root,c,x}; return next_result; } + static Result ReduceScatter(const void* s, void* r, size_t n, DataType t, RedOp o, Comm c, Stream x) { call = {s,r,n,t,o,-1,c,x}; return next_result; } + static Result AllGather(const void* s, void* r, size_t n, DataType t, Comm c, Stream x) { call = {s,r,n,t,0,-1,c,x}; return next_result; } + static Result Gather(const void* s, void* r, size_t n, DataType t, int root, Comm c, Stream x) { call = {s,r,n,t,0,root,c,x}; return next_result; } + static Result Scatter(const void* s, void* r, size_t n, DataType t, int root, Comm c, Stream x) { call = {s,r,n,t,0,root,c,x}; return next_result; } + static Result AllToAll(const void* s, void* r, size_t n, DataType t, Comm c, Stream x) { call = {s,r,n,t,0,-1,c,x}; return next_result; } +}; +FakeMcclApi::Call FakeMcclApi::call{}; +FakeMcclApi::Result FakeMcclApi::next_result = FakeMcclApi::kSuccess; + +template <> struct CclApi : FakeMcclApi {}; +template <> struct CclTypeMap { + static bool ToBackendDataType(DataType type, int* result) { if (type == DataType::kUInt16) return false; *result = 100 + static_cast(type); return true; } + static bool ToBackendRedOp(ReductionOpType op, int* result) { if (op == ReductionOpType::kNumRedOps) return false; *result = 200 + static_cast(op); return true; } +}; +using Api = CclApi; +using Instance = CclCommInstance; +std::unique_ptr MakeComm() { auto result = std::make_unique(); result->handle = 7; return result; } + +bool ForwardAllCollectives() { + FakeMcclApi::Reset(); + Communicator comm(Device::Type::kMetax, 0); comm.set_intra_comm(MakeComm()); + int send = 1, recv = 0; void* stream = reinterpret_cast(0x1234); + using B = CclBroadcastImpl; + using R = CclReduceImpl; + using S = CclReduceScatterImpl; + using G = CclAllGatherImpl; + using H = CclGatherImpl; + using T = CclScatterImpl; + using A = CclAllToAllImpl; + if (B::Apply(&send,&recv,3,DataType::kFloat32,1,&comm,stream) != ReturnStatus::kSuccess || FakeMcclApi::call.root != 1) return false; + if (R::Apply(&send,&recv,4,DataType::kInt32,ReductionOpType::kSum,0,&comm,stream) != ReturnStatus::kSuccess || FakeMcclApi::call.op != 200) return false; + if (S::Apply(&send,&recv,5,DataType::kFloat32,ReductionOpType::kMax,&comm,stream) != ReturnStatus::kSuccess) return false; + if (G::Apply(&send,&recv,6,DataType::kFloat32,&comm,stream) != ReturnStatus::kSuccess) return false; + if (H::Apply(&send,&recv,7,DataType::kFloat32,0,&comm,stream) != ReturnStatus::kSuccess) return false; + if (T::Apply(&send,&recv,8,DataType::kFloat32,0,&comm,stream) != ReturnStatus::kSuccess) return false; + return A::Apply(&send,&recv,9,DataType::kFloat32,&comm,stream) == ReturnStatus::kSuccess; +} + +bool RejectUnsupportedAndPropagateError() { + FakeMcclApi::Reset(); Communicator comm(Device::Type::kMetax, 0); comm.set_intra_comm(MakeComm()); + int send = 1, recv = 0; + using B = CclBroadcastImpl; + if (B::Apply(&send,&recv,1,DataType::kUInt16,0,&comm,nullptr) != ReturnStatus::kNotSupported || FakeMcclApi::call.send != nullptr) return false; + FakeMcclApi::next_result = FakeMcclApi::kFailure; + return B::Apply(&send,&recv,1,DataType::kFloat32,0,&comm,nullptr) == ReturnStatus::kSystemError; +} + +bool RejectInvalidCommunicator() { + int send = 1, recv = 0; + using A = CclAllToAllImpl; + Communicator comm(Device::Type::kMetax, 0); + return A::Apply(&send,&recv,1,DataType::kFloat32,&comm,nullptr) == ReturnStatus::kInternalError; +} + +bool Run(const char* name, bool (*test)()) { if (test()) return true; std::cerr << "FAILED: " << name << std::endl; return false; } +} // namespace infini::ccl + +int main() { + using namespace infini::ccl; + bool passed = Run("forward all collectives", ForwardAllCollectives); + passed = Run("unsupported type and backend error", RejectUnsupportedAndPropagateError) && passed; + passed = Run("invalid communicator", RejectInvalidCommunicator) && passed; + return passed ? EXIT_SUCCESS : EXIT_FAILURE; +}