Raise exception instead of deadlocking based on minimum elem count validation for ctring AllReduce

Summary:
The `ctring` algorithm for the `AllReduce` collective has a bug that causes it to hang when the message size is too small. Based on my empirical testing, this appears to occur when the message size is below 8 bytes.

## Exception Details
When the message size is less than 8 bytes, the code:
1.  Throws an `std::runtime_error` with a clear, informative error message
2.  Logs the error using `XLOGF(ERR, ...)` before throwing
3.  Provides actionable guidance to the user (use a larger message size or different algorithm)

## Error Message
The error message includes:
*   The actual message size received
*   The count and datatype size for debugging
*   Clear explanation that this would cause a hang
*   Suggestions for remediation

Ultimately, this is a short-term fix that will prevent users from experiencing a hang when using the `ctring` `AllReduce` algorithm with message sizes below 8 bytes. Instead, they'll get a clear error message explaining the issue and how to work around it.

Reviewed By: arttianezhu

Differential Revision: D84266985

fbshipit-source-id: ecc340ebaa1bacbce28662d9193f73ed2780c180

Author: Ben Carver

comms/ctran/algos/AllReduce/AllReduceRing.cc

@@ -737,6 +737,35 @@ commResult_t ctranAllReduceRing(
     CtranComm* comm,
     cudaStream_t stream,
     std::optional<std::chrono::milliseconds> timeout) {
+  // Check for minimum message size requirement for ctring algorithm.
+  // The ctring algorithm uses a ring-based approach that shards data across all
+  // ranks. Each rank must have at least one element in its shard to avoid empty
+  // chunk transfers that can lead to synchronization deadlocks. Therefore, we
+  // need at least nRanks elements.
+  const auto& statex = comm->statex_.get();
+  const auto nRanks = statex->nRanks();
+  const auto rank = statex->rank();
+  const size_t typeSize = static_cast<size_t>(commTypeSize(datatype));
+  const size_t minRequiredElements = nRanks;
+  const size_t minRequiredBytes = minRequiredElements * typeSize;
+
+  if (count < minRequiredElements) {
+    std::string errorMsg = fmt::format(
+        "ctring algorithm requires at least {} elements ({} bytes) for {} ranks, "
+        "but rank {} got {} elements ({} bytes) with datatype size={} bytes. "
+        "Each rank needs at least one element per shard. "
+        "Please use a larger message size or a different allreduce algorithm (e.g., ctdirect).",
+        minRequiredElements,
+        minRequiredBytes,
+        nRanks,
+        rank,
+        count,
+        count * typeSize,
+        typeSize);
+    CLOGF(ERR, "{}", errorMsg);
+    throw ctran::utils::Exception(errorMsg, commInvalidArgument);
+  }
+
   auto opCount = comm->ctran_->getOpCount();
   CTRAN_REDCOLL_INFO(
       allReduceAlgoName(ctran::allreduce::ring::myAlgo),
@@ -752,10 +781,6 @@ commResult_t ctranAllReduceRing(
   std::vector<std::unique_ptr<struct OpElem>> opGroup;
   std::unique_ptr<struct OpElem> op;
 
-  const auto& statex = comm->statex_.get();
-  const auto nRanks = statex->nRanks();
-  const auto rank = statex->rank();
-
   FB_CHECKTHROW(
       typeToFunc.contains(std::make_pair(datatype, redOp)),
       "typeToFunc does not contain datatype {} with op {}",

comms/ctran/tests/CtranAllReduceTest.cc

@@ -14,9 +14,17 @@
 #include "comms/ctran/tests/CtranStandaloneUTUtils.h"
 #include "comms/utils/cvars/nccl_cvars.h"
 
+#include "comms/ctran/algos/AllReduce/AllReduceImpl.h"
+
 namespace ctran::testing {
 
 using AllReduceTestParam = std::tuple<std::string, enum NCCL_ALLREDUCE_ALGO>;
+using AllReduceMinMsgSizeTestParam = std::tuple<size_t, commDataType_t>;
+
+enum class CtranAllReduceRingMinSizeTestOpt {
+  expect_sufficient,
+  expect_insufficient,
+};
 
 class CtranAllReduceTest
     : public CtranStandaloneMultiRankBaseTest,
@@ -250,4 +258,156 @@ INSTANTIATE_TEST_SUITE_P(
       return std::get<0>(info.param);
     });
 
+// Test fixture for ctring minimum message size validation
+class CtranAllReduceRingMinSizeTest
+    : public CtranStandaloneMultiRankBaseTest,
+      public ::testing::WithParamInterface<AllReduceMinMsgSizeTestParam> {
+ protected:
+  static constexpr int kDefaultNumRanks = 4;
+  static_assert(kDefaultNumRanks % 2 == 0);
+  static constexpr commRedOp_t kReduceOpType = commSum;
+
+  void SetUp() override {
+    setenv("NCCL_COMM_STATE_DEBUG_TOPO", "nolocal", 1);
+    setenv("NCCL_IGNORE_TOPO_LOAD_FAILURE", "1", 1);
+    CtranStandaloneMultiRankBaseTest::SetUp();
+  }
+
+  void startWorkers(int numRanks = kDefaultNumRanks) {
+    std::vector<std::shared_ptr<::ctran::utils::Abort>> aborts;
+    aborts.reserve(numRanks);
+    for (int i = 0; i < numRanks; ++i) {
+      aborts.push_back(ctran::utils::createAbort(/*enabled=*/true));
+    }
+    CtranStandaloneMultiRankBaseTest::startWorkers(numRanks, /*aborts=*/aborts);
+  }
+
+  void runTest(
+      size_t count,
+      commDataType_t dt,
+      enum CtranAllReduceRingMinSizeTestOpt testOpt,
+      int numRanks = kDefaultNumRanks) {
+    startWorkers(numRanks);
+    for (int rank = 0; rank < numRanks; ++rank) {
+      run(rank, [this, count, dt, testOpt](PerRankState& state) {
+        ASSERT_TRUE(ctranAllReduceSupport(state.ctranComm.get()));
+
+        size_t bufferSize = count * commTypeSize(dt);
+        if (bufferSize < CTRAN_MIN_REGISTRATION_SIZE) {
+          bufferSize = CTRAN_MIN_REGISTRATION_SIZE;
+        }
+
+        void* srcHandle;
+        void* dstHandle;
+        ASSERT_EQ(
+            commSuccess,
+            state.ctranComm->ctran_->commRegister(
+                state.srcBuffer, bufferSize, &srcHandle));
+        ASSERT_EQ(
+            commSuccess,
+            state.ctranComm->ctran_->commRegister(
+                state.dstBuffer, bufferSize, &dstHandle));
+
+        if (testOpt == CtranAllReduceRingMinSizeTestOpt::expect_sufficient) {
+          // Should not throw when count >= nRanks
+          EXPECT_NO_THROW({
+            auto res = ctranAllReduceRing(
+                state.srcBuffer,
+                state.dstBuffer,
+                count,
+                dt,
+                kReduceOpType,
+                state.ctranComm.get(),
+                state.stream);
+            EXPECT_EQ(res, commSuccess);
+          });
+        } else {
+          // Expect ctran::utils::Exception when count < nRanks
+          EXPECT_THROW(
+              {
+                ctranAllReduceRing(
+                    state.srcBuffer,
+                    state.dstBuffer,
+                    count,
+                    dt,
+                    kReduceOpType,
+                    state.ctranComm.get(),
+                    state.stream);
+              },
+              ctran::utils::Exception);
+        }
+
+        // ensure async execution completion and no error
+        EXPECT_EQ(cudaSuccess, cudaStreamSynchronize(state.stream));
+
+        // deregistering will happen after streamSync below
+        ASSERT_EQ(
+            commSuccess, state.ctranComm->ctran_->commDeregister(dstHandle));
+        ASSERT_EQ(
+            commSuccess, state.ctranComm->ctran_->commDeregister(srcHandle));
+      });
+    }
+  }
+};
+
+TEST_P(CtranAllReduceRingMinSizeTest, InsufficientElements_1Element) {
+  auto [numRanks, dt] = GetParam();
+  ASSERT_FALSE(numRanks <= 1) << "Need at least 2 ranks for this test";
+  runTest(
+      1, dt, CtranAllReduceRingMinSizeTestOpt::expect_insufficient, numRanks);
+}
+
+TEST_P(CtranAllReduceRingMinSizeTest, InsufficientElements_NRanksMinus1) {
+  auto [numRanks, dt] = GetParam();
+  ASSERT_FALSE(numRanks <= 1) << "Need at least 2 ranks for this test";
+  runTest(
+      numRanks - 1,
+      dt,
+      CtranAllReduceRingMinSizeTestOpt::expect_insufficient,
+      numRanks);
+}
+
+TEST_P(CtranAllReduceRingMinSizeTest, SufficientElements_ExactlyNRanks) {
+  auto [numRanks, dt] = GetParam();
+  XLOG(INFO) << "SufficientElements_ExactlyNRanks: numRanks: " << numRanks
+             << ", dt: " << dt;
+  runTest(
+      numRanks,
+      dt,
+      CtranAllReduceRingMinSizeTestOpt::expect_sufficient,
+      numRanks);
+}
+
+TEST_P(CtranAllReduceRingMinSizeTest, SufficientElements_NRanksPlus1) {
+  auto [numRanks, dt] = GetParam();
+  runTest(
+      numRanks + 1,
+      dt,
+      CtranAllReduceRingMinSizeTestOpt::expect_sufficient,
+      numRanks);
+}
+
+TEST_P(CtranAllReduceRingMinSizeTest, SufficientElements_LargeMessage) {
+  auto [numRanks, dt] = GetParam();
+  runTest(
+      1024, dt, CtranAllReduceRingMinSizeTestOpt::expect_sufficient, numRanks);
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    AllDataTypes,
+    CtranAllReduceRingMinSizeTest,
+    ::testing::Values(
+        std::make_tuple<>(2, commFloat),
+        std::make_tuple<>(2, commInt32),
+        std::make_tuple<>(2, commInt8),
+        std::make_tuple<>(4, commFloat),
+        std::make_tuple<>(4, commInt32),
+        std::make_tuple<>(4, commInt8),
+        std::make_tuple<>(6, commFloat),
+        std::make_tuple<>(6, commInt32),
+        std::make_tuple<>(6, commInt8),
+        std::make_tuple<>(8, commFloat),
+        std::make_tuple<>(8, commInt32),
+        std::make_tuple<>(8, commInt8)));
+
 } // namespace ctran::testing

comms/ctran/tests/CtranStandaloneUTUtils.cc

@@ -56,7 +56,7 @@ void initOnce() {
 } // namespace
 
 void CtranStandaloneBaseTest::setupBase() {
-  setenv("NCCL_CTRAN_ENABLE", "INFO", 1);
+  setenv("NCCL_CTRAN_ENABLE", "1", 1);
   setenv("NCCL_DEBUG", "INFO", 1);
 
   rank = 0;