From 43f2249c8c7c18445fc16143ce9e4a2b47d3c794 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:23:13 -0700 Subject: [PATCH 01/11] test(rccl): add host tests for nccl::utility::add4G Pin the add4G contract used by the LSA peer pointer math: the delta lands in the high word only, the low word is never touched, the high word wraps mod 2^32, and pointer bases round-trip. Register Add4G.* in the fixtures quick and standard categories and in the precheckin BitOps filter. --- projects/rccl/test/BitOpsTests.cpp | 43 +++++++++++++++++++ .../rccl/test/test_categories_fixtures.yaml | 2 + .../test_runner/configs/ci-precheckin.json | 2 +- 3 files changed, 46 insertions(+), 1 deletion(-) diff --git a/projects/rccl/test/BitOpsTests.cpp b/projects/rccl/test/BitOpsTests.cpp index d372717afa66..13ce0435ff28 100644 --- a/projects/rccl/test/BitOpsTests.cpp +++ b/projects/rccl/test/BitOpsTests.cpp @@ -5,8 +5,14 @@ ************************************************************************/ #include "bitops.h" + +#include +#include + #include "gtest/gtest.h" +#include "nccl_device/utility.h" + namespace RcclUnitTesting { @@ -279,4 +285,41 @@ TYPED_TEST(BitOpsTemplateUnsignedTestsFixture, reverseBitsSuccess) { EXPECT_EQ(ret, 16384); } +// One 4 GiB stride, the unit of add4G's delta. +constexpr uint64_t k4G = 1ULL << 32; + +// add4G backs the LSA peer pointer math; always qualify, since bitops.h declares a same-signature ::add4G. +TEST(Add4G, AddsDeltaTimes4G) { + constexpr uint64_t kBase = 0x00000002ABCDEF01ULL; + EXPECT_EQ(nccl::utility::add4G(kBase, 0), kBase); + EXPECT_EQ(nccl::utility::add4G(kBase, 1), kBase + k4G); + EXPECT_EQ(nccl::utility::add4G(kBase, 3), kBase + 3 * k4G); + EXPECT_EQ(nccl::utility::add4G(kBase, -1), kBase - k4G); + EXPECT_EQ(nccl::utility::add4G(kBase, -2), kBase - 2 * k4G); +} + +TEST(Add4G, LowWordUntouchedHighWordWraps) { + EXPECT_EQ(nccl::utility::add4G(0x00000001FFFFFFFFULL, 1), 0x00000002FFFFFFFFULL); + EXPECT_EQ(nccl::utility::add4G(0xFFFFFFFF00000005ULL, 1), 0x0000000000000005ULL); + EXPECT_EQ(nccl::utility::add4G(0x0000000000000005ULL, -1), 0xFFFFFFFF00000005ULL); + EXPECT_EQ(nccl::utility::add4G(0ULL, INT_MAX), 0x7FFFFFFF00000000ULL); + EXPECT_EQ(nccl::utility::add4G(0ULL, INT_MIN), 0x8000000000000000ULL); +} + +TEST(Add4G, PointerBase) { + constexpr uintptr_t kAddr = 0x00007F0012345000ULL; + constexpr int kStride4G = 2; + constexpr int kNumRanks = 4; + char* cbase = reinterpret_cast(kAddr); + void* vbase = reinterpret_cast(kAddr); + for (int rank = 0; rank < kNumRanks; ++rank) { + const int delta4G = rank * kStride4G; + const uintptr_t expected = kAddr + static_cast(delta4G) * k4G; + EXPECT_EQ(reinterpret_cast(nccl::utility::add4G(cbase, delta4G)), expected); + EXPECT_EQ(reinterpret_cast(nccl::utility::add4G(vbase, delta4G)), expected); + } + char* top = nccl::utility::add4G(cbase, 3 * kStride4G); + EXPECT_EQ(nccl::utility::add4G(top, -3 * kStride4G), cbase); +} + } \ No newline at end of file diff --git a/projects/rccl/test/test_categories_fixtures.yaml b/projects/rccl/test/test_categories_fixtures.yaml index 4066371bbafc..dce70e279aac 100644 --- a/projects/rccl/test/test_categories_fixtures.yaml +++ b/projects/rccl/test/test_categories_fixtures.yaml @@ -21,6 +21,7 @@ test_categories: - "u32fpDecode.*" - "getHash.*" - "eatHash.*" + - "Add4G.*" - "BitOpsTemplateAllIntTestsFixture.*" - "BitOpsTemplateUnsignedTestsFixture.*" - "CommTests.*" @@ -78,6 +79,7 @@ test_categories: - "u32fpDecode.*" - "getHash.*" - "eatHash.*" + - "Add4G.*" - "RcclEpWaveTest.*" - "RcclEpLayoutTest.*" exclude_windows: diff --git a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json index f980042e97d9..858522d59889 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json +++ b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json @@ -129,7 +129,7 @@ "NCCL_DEBUG": "INFO" }, "tests": [ - {"name": "BitOps", "description": "Bit operation utility tests (macros and typed fixtures)", "test_filter": "ALIGN_*:DIVUP:ROUNDUP:u32fp*:*Hash:BitOpsTemplate*"}, + {"name": "BitOps", "description": "Bit operation utility tests (macros and typed fixtures)", "test_filter": "ALIGN_*:DIVUP:ROUNDUP:u32fp*:*Hash:BitOpsTemplate*:Add4G.*"}, {"name": "EnqueueCountTests", "description": "Enqueue send/recv count helpers", "test_filter": "EnqueueCountTests.*"}, {"name": "MiscTests.Sorter", "description": "Task coll sorter unit test", "test_filter": "MiscTests.Sorter"}, {"name": "VersionInfoTests", "description": "HIP/ROCm version string formatting tests", "test_filter": "VersionInfoTests.*"}, From a66984e63cbd441700b93d7bf6cb16234b4a5407 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:23:58 -0700 Subject: [PATCH 02/11] test(rccl): add Recorder RedOp record JSON test Cover the RedOp record() overload in recorder.cc: the PreMulSum branch must serialise scalar, datatype and residence, and RedOpDestroy must serialise op and comm. Register the test in the test_runner configs. --- projects/rccl/test/_RecorderTests.cpp | 65 +++++++++++++++++++ .../test_runner/configs/ci-precheckin.json | 1 + .../configs/mi300_mellanox_ib_func.json | 5 ++ .../configs/mi300x_mellanox_ib.json | 1 + .../configs/mi355x_thor2_roce.json | 5 ++ .../configs/mi355x_thor2_roce_func.json | 5 ++ .../test_runner/configs/mi455_ainic_roce.json | 5 ++ .../configs/mi455_ainic_roce_func.json | 5 ++ 8 files changed, 92 insertions(+) diff --git a/projects/rccl/test/_RecorderTests.cpp b/projects/rccl/test/_RecorderTests.cpp index dc1d81a72eab..3a406970627f 100644 --- a/projects/rccl/test/_RecorderTests.cpp +++ b/projects/rccl/test/_RecorderTests.cpp @@ -11,6 +11,7 @@ #include "comm.h" #include "common/ProcessIsolatedTestRunner.hpp" +#include #include #include @@ -69,6 +70,70 @@ namespace RcclUnitTesting ); } + /** + * \brief Verify the RedOp record() overload writes PreMulSum scalar/datatype/residence and Destroy op/comm to json + * ******************************************************************************************/ + TEST(Recorder, RedOpRecordJson) + { + RUN_ISOLATED_TEST_WITH_ENV( + "RedOpRecordJson", + []() + { + int array[] = {2, 3, 5}; + ncclComm comm{.nRanks = 1, .localRank = 1, .localRankToRank = array, .opCount = 8, + .planner = {.nTasksColl = 13, .nTasksP2p = 21}}; + float scalar = 2.0f; + // Non-zero datatype and residence so a dropped PreMulSum branch is not hidden by the zero-initialised defaults. + ASSERT_EQ(ncclSuccess, rccl::Recorder::instance().record(rccl::rrRedOpCreatePreMulSum, ncclMax, &comm, + ncclFloat32, ncclScalarHostImmediate, &scalar)); + ASSERT_EQ(ncclSuccess, rccl::Recorder::instance().record(rccl::rrRedOpDestroy, ncclMax, &comm)); + + char line[4096]; + gethostname(line, sizeof(line)); + std::string filename = "/tmp/test_redop." + std::to_string(getpid()) + "." + std::string(line) + ".json"; + std::ifstream fp(filename); + ASSERT_TRUE(fp.is_open()) << "Recorder did not create expected file: " << filename; + fp.getline(line, sizeof(line)); // line 1: "{" + fp.getline(line, sizeof(line)); // line 2: " version : 1," + fp.getline(line, sizeof(line)); // line 3: RedOpCreatePreMulSum + std::string create(line); + fp.getline(line, sizeof(line)); // line 4: RedOpDestroy + std::string destroy(line); + fp.close(); + std::remove(filename.c_str()); + + // parseJsonEntry is not used: its RedOp sscanf calls pass fewer pointers than conversions (UB) inside assert(). + size_t pos = create.find("RedOpCreatePreMulSum : ["); + ASSERT_NE(pos, std::string::npos) << "raw line: " << create; + void* gotScalar = nullptr; + void* gotComm = nullptr; + int gotDatatype = -1; + int gotOp = -1; + int gotResidence = -1; + constexpr const char* kCreateFmt = + "RedOpCreatePreMulSum : [scalar : %p, datatype : %d, op : %d, residence : %d, comm : %p,"; + ASSERT_EQ(5, std::sscanf(create.c_str() + pos, kCreateFmt, &gotScalar, &gotDatatype, &gotOp, &gotResidence, + &gotComm)) + << "raw line: " << create; + EXPECT_EQ(gotScalar, static_cast(&scalar)); + EXPECT_EQ(gotDatatype, ncclFloat32); + EXPECT_EQ(gotOp, ncclMax); + EXPECT_EQ(gotResidence, ncclScalarHostImmediate); + EXPECT_EQ(gotComm, static_cast(&comm)); + + pos = destroy.find("RedOpDestroy : ["); + ASSERT_NE(pos, std::string::npos) << "raw line: " << destroy; + gotOp = -1; + gotComm = nullptr; + ASSERT_EQ(2, std::sscanf(destroy.c_str() + pos, "RedOpDestroy : [op : %d, comm : %p,", &gotOp, &gotComm)) + << "raw line: " << destroy; + EXPECT_EQ(gotOp, ncclMax); + EXPECT_EQ(gotComm, static_cast(&comm)); + }, + {{"RCCL_REPLAY_FILE", "/tmp/test_redop.json"}} + ); + } + /** * \brief Verify RCCL Recorder's integrity in multithread context by comparing Recorder * instance across different threads. diff --git a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json index 858522d59889..85eb4c927647 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json +++ b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json @@ -85,6 +85,7 @@ {"name": "Recorder.ParseBinary", "description": "Recorder parse binary tests", "test_filter": "Recorder.ParseBinary", "timeout": 60}, {"name": "Recorder.ParseJson", "description": "Recorder parse JSON tests", "test_filter": "Recorder.ParseJson", "timeout": 60}, {"name": "Recorder.VerifyMultithread", "description": "Recorder multithread verification tests", "test_filter": "Recorder.VerifyMultithread", "timeout": 60}, + {"name": "Recorder.RedOpRecordJson", "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson", "timeout": 60}, {"name": "ProxyTraceTestFixture.nonEmptySingleton", "description": "ProxyTrace non-empty singleton test", "test_filter": "ProxyTraceTestFixture.nonEmptySingleton", "timeout": 60}, {"name": "ProxyTraceTestFixture.addTraceOp", "description": "ProxyTrace add trace operation test", "test_filter": "ProxyTraceTestFixture.addTraceOp", "timeout": 60}, {"name": "ProxyTraceTestFixture.getMapSizeMB", "description": "ProxyTrace get map size test", "test_filter": "ProxyTraceTestFixture.getMapSizeMB", "timeout": 60}, diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json index a6dacc15a3eb..493eedda8db5 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json @@ -1828,6 +1828,11 @@ "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread" }, + { + "name": "Recorder.RedOpRecordJson", + "description": "Recorder RedOp record JSON output", + "test_filter": "Recorder.RedOpRecordJson" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json index 39c8b0288d6d..560c2bd3693a 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json @@ -1195,6 +1195,7 @@ {"name": "Recorder.ParseBinary", "description": "Recorder parse binary trace", "test_filter": "Recorder.ParseBinary"}, {"name": "Recorder.ParseJson", "description": "Recorder parse JSON trace", "test_filter": "Recorder.ParseJson"}, {"name": "Recorder.VerifyMultithread", "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread"}, + {"name": "Recorder.RedOpRecordJson", "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson"}, {"name": "IbvSymbolsEnv.DefaultLoadsLibrary", "description": "NCCL_IBVERBS_LIB unset loads default libibverbs", "test_filter": "IbvSymbolsEnv.DefaultLoadsLibrary"}, {"name": "IbvSymbolsEnv.EnvOverrideValidSoname", "description": "NCCL_IBVERBS_LIB override to a valid soname is honored", "test_filter": "IbvSymbolsEnv.EnvOverrideValidSoname"}, {"name": "IbvSymbolsEnv.BogusPathFallsBackToDefault", "description": "NCCL_IBVERBS_LIB bogus path falls back to default libibverbs", "test_filter": "IbvSymbolsEnv.BogusPathFallsBackToDefault"}, diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json index eb3f9ace78b3..46b968edf3b6 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json @@ -1422,6 +1422,11 @@ "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread" }, + { + "name": "Recorder.RedOpRecordJson", + "description": "Recorder RedOp record JSON output", + "test_filter": "Recorder.RedOpRecordJson" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json index f324b25cf683..5e53129a6129 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json @@ -1878,6 +1878,11 @@ "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread" }, + { + "name": "Recorder.RedOpRecordJson", + "description": "Recorder RedOp record JSON output", + "test_filter": "Recorder.RedOpRecordJson" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json index 71b24957a23c..b3d7c47e1d40 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json @@ -1965,6 +1965,11 @@ "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread" }, + { + "name": "Recorder.RedOpRecordJson", + "description": "Recorder RedOp record JSON output", + "test_filter": "Recorder.RedOpRecordJson" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json index 705fb6c41f37..17c3afb09e74 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json @@ -2019,6 +2019,11 @@ "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread" }, + { + "name": "Recorder.RedOpRecordJson", + "description": "Recorder RedOp record JSON output", + "test_filter": "Recorder.RedOpRecordJson" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", From 374120c1b1ef6e223a909aa5a3287e66f18ab880 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:23:58 -0700 Subject: [PATCH 03/11] test(rccl): cover negative lsaRank in ncclDevrGetLsaRankPtr Add flat-VA and IPC-table cases asserting a negative lsaRank returns ncclInvalidArgument instead of resolving below the flat base or reading ipcPeerPtrs[-1]. --- projects/rccl/test/host/dev-runtime-test.cc | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/projects/rccl/test/host/dev-runtime-test.cc b/projects/rccl/test/host/dev-runtime-test.cc index 9d29a45efdfb..d06f042e61e4 100644 --- a/projects/rccl/test/host/dev-runtime-test.cc +++ b/projects/rccl/test/host/dev-runtime-test.cc @@ -4321,6 +4321,13 @@ TEST_F(DevrGetLsaRankPtrTest, RankPastTeam_ReturnsInvalidArgument) { EXPECT_EQ(ncclDevrGetLsaRankPtr(comm, &win, 0, comm->devrState.lsaSize, &out), ncclInvalidArgument); } +// Boundary: a negative rank; a missing check would resolve below the flat base, outside every rank's slot. +TEST_F(DevrGetLsaRankPtrTest, NegativeRank_ReturnsInvalidArgument) { + void* out = nullptr; + EXPECT_EQ(ncclDevrGetLsaRankPtr(comm, &win, 0, -1, &out), ncclInvalidArgument); + EXPECT_EQ(out, nullptr); +} + // Branch: a non-symmetric window targeting ourselves resolves against the local // window base, not the flat space -- which does not apply to it. TEST_F(DevrGetLsaRankPtrTest, NonSymmetricSelfTarget_UsesLocalBase) { @@ -4368,6 +4375,19 @@ TEST_F(DevrGetLsaRankPtrTest, IpcWindowRankPastTable_ReturnsInvalidArgument) { EXPECT_EQ(ncclDevrGetLsaRankPtr(comm, &win, 0, 5, &out), ncclInvalidArgument); } +// Boundary: a negative rank must not index the IPC table; a mapped sentinel at [-1] makes a missing check succeed. +TEST_F(DevrGetLsaRankPtrTest, IpcWindowNegativeRank_ReturnsInvalidArgument) { + comm->symmetricSupport = 0; + peerPtrs = {reinterpret_cast(0x5000), reinterpret_cast(0xA000), reinterpret_cast(0xB000), + reinterpret_cast(0xC000)}; + win.ipcPeerPtrs = peerPtrs.data() + 1; + win.ipcPeerCount = 3; + + void* out = nullptr; + EXPECT_EQ(ncclDevrGetLsaRankPtr(comm, &win, 0, -1, &out), ncclInvalidArgument); + EXPECT_EQ(out, nullptr); +} + // --------------------------------------------------------------------------- // The public device-pointer accessors. All four resolve the device window to From 86499123560931fc2ec2337e71ff8f32e0e17e91 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:24:41 -0700 Subject: [PATCH 04/11] test(rccl): add ncclProxyCallBlocking socketpair tests Drive ncclProxyCallBlocking against a fake proxy on a socketpair and check four paths: a delayed reply is polled until it arrives, a proxy error result ends the poll unchanged, a raised abort flag ends a wait on a silent proxy, and a send failure returns without polling. --- projects/rccl/test/ProxyTests.cpp | 348 ++++++++++++++++++++++++++++++ 1 file changed, 348 insertions(+) diff --git a/projects/rccl/test/ProxyTests.cpp b/projects/rccl/test/ProxyTests.cpp index ac846032b4e7..5f304fc1fb38 100644 --- a/projects/rccl/test/ProxyTests.cpp +++ b/projects/rccl/test/ProxyTests.cpp @@ -13,9 +13,11 @@ #include "socket.h" #define ENABLE_TIMER 0 #include +#include #include #include #include +#include #include #include #include @@ -24,9 +26,13 @@ #include #include +#include +#include #include #include #include +#include +#include #include "common/ErrCode.hpp" #include "common/ProcessIsolatedTestRunner.hpp" @@ -317,6 +323,348 @@ TEST(ProxyTests, ncclProxyCallBlockingUDS) TEST_INFO("[ProxyTests] Test Complete"); } +namespace +{ +constexpr int kProxyCallConnId = 7; +constexpr auto kProxyCallTimeout = std::chrono::seconds(30); +constexpr auto kProxyCallReplyDelay = std::chrono::milliseconds(100); +constexpr auto kNoReplyDelay = std::chrono::milliseconds(0); + +// Loops recv until len bytes arrive; false on EOF or error. +bool ReadFull(int fd, void* buf, size_t len) +{ + char* p = static_cast(buf); + while(len > 0) + { + ssize_t n = recv(fd, p, len, 0); + if(n <= 0) + { + return false; + } + p += n; + len -= static_cast(n); + } + return true; +} + +// Loops send until len bytes are written; false on error. +bool WriteFull(int fd, const void* buf, size_t len) +{ + const char* p = static_cast(buf); + while(len > 0) + { + ssize_t n = send(fd, p, len, MSG_NOSIGNAL); + if(n <= 0) + { + return false; + } + p += n; + len -= static_cast(n); + } + return true; +} + +// One RPC as ncclProxyCallAsync puts it on the wire, decoded by the fake proxy. +struct ProxyRpcRequest +{ + int type = -1; + int connId = -1; + int reqSize = -1; + int respSize = -1; + std::vector req; + void* opId = nullptr; +}; + +bool ReadProxyRpcRequest(int fd, ProxyRpcRequest* out) +{ + if(!ReadFull(fd, &out->type, sizeof(out->type)) + || !ReadFull(fd, &out->connId, sizeof(out->connId)) + || !ReadFull(fd, &out->reqSize, sizeof(out->reqSize)) + || !ReadFull(fd, &out->respSize, sizeof(out->respSize))) + { + return false; + } + if(out->reqSize < 0) + { + return false; + } + out->req.resize(static_cast(out->reqSize)); + if(out->reqSize > 0 && !ReadFull(fd, out->req.data(), out->req.size())) + { + return false; + } + return ReadFull(fd, &out->opId, sizeof(out->opId)); +} + +// Checks the fixed RPC header fields every ncclProxyCallBlocking request must carry. +void ExpectProxyRpcHeader(const ProxyRpcRequest& seen, int reqSize, int respSize) +{ + EXPECT_EQ(seen.type, ncclProxyMsgSetup); + EXPECT_EQ(seen.connId, kProxyCallConnId); + EXPECT_EQ(seen.reqSize, reqSize); + EXPECT_EQ(seen.respSize, respSize); + EXPECT_NE(seen.opId, nullptr); +} + +// Expects fd to have no unread bytes, so the client sent nothing beyond what was already consumed. +void ExpectNoPendingBytes(int fd) +{ + char byte = 0; + ssize_t n = recv(fd, &byte, 1, MSG_DONTWAIT); + int recvErr = errno; + EXPECT_EQ(n, -1); + EXPECT_TRUE(recvErr == EAGAIN || recvErr == EWOULDBLOCK) << "errno " << recvErr; +} + +// Client side of ncclProxyCallBlocking: a socketpair end wrapped as a Ready ncclSocket on a heap comm. +class ProxyCallHarness +{ +public: + ProxyCallHarness() = default; + ProxyCallHarness(const ProxyCallHarness&) = delete; + ProxyCallHarness& operator=(const ProxyCallHarness&) = delete; + + ~ProxyCallHarness() + { + // Unblocks a fake proxy still waiting in recv before the fds go away. + if(fds_[0] >= 0) + { + shutdown(fds_[0], SHUT_RDWR); + } + JoinProxy(); + for(int fd : fds_) + { + if(fd >= 0) + { + close(fd); + } + } + } + + bool Init() + { + if(socketpair(AF_UNIX, SOCK_STREAM, 0, fds_) != 0) + { + return false; + } + // sock_.abortFlag stays null, unlike production; the poll loop's abort check reads comm_->abortFlag instead. + sock_ = ncclSocket{}; + sock_.socketDescriptor = fds_[0]; + sock_.state = ncclSocketStateReady; + proxyState_ = std::make_unique(); + proxyState_->peerSocks = &sock_; + comm_ = std::make_unique(); + comm_->proxyState = proxyState_.get(); + comm_->abortFlag = &abortFlag_; + conn_ = ncclProxyConnector{}; + conn_.tpLocalRank = 0; + conn_.connId = kProxyCallConnId; + return true; + } + + // Plays the proxy for one RPC: reads the request, waits delay, then replies with res and resp. + void StartReplyingProxy(ncclResult_t res, std::vector resp, std::chrono::milliseconds delay) + { + int peer = fds_[1]; + proxy_ = std::thread([this, res, resp = std::move(resp), delay, peer]() { + readOk_ = ReadProxyRpcRequest(peer, &seen_); + if(readOk_) + { + std::this_thread::sleep_for(delay); + ncclProxyRpcResponseHeader hdr = {seen_.opId, res, static_cast(resp.size())}; + writeOk_ = WriteFull(peer, &hdr, sizeof(hdr)) + && (resp.empty() || WriteFull(peer, resp.data(), resp.size())); + } + if(!readOk_ || !writeOk_) + { + // EOF makes the client's poll fail fast instead of waiting for the test timeout. + shutdown(peer, SHUT_RDWR); + } + }); + } + + // Plays a proxy that reads the request, then raises the comm abort flag instead of replying. + void StartAbortingProxy() + { + int peer = fds_[1]; + proxy_ = std::thread([this, peer]() { + readOk_ = ReadProxyRpcRequest(peer, &seen_); + __atomic_store_n(&abortFlag_, 1, __ATOMIC_RELEASE); + }); + } + + void JoinProxy() + { + if(proxy_.joinable()) + { + proxy_.join(); + } + } + + ncclResult_t Call(int type, void* req, int reqSize, void* resp, int respSize) + { + return ncclProxyCallBlocking(comm_.get(), &conn_, type, req, reqSize, resp, respSize); + } + + ncclSocket& sock() + { + return sock_; + } + + ncclProxyState* proxyState() + { + return proxyState_.get(); + } + + int clientFd() const + { + return fds_[0]; + } + + int peerFd() const + { + return fds_[1]; + } + + // Valid only after JoinProxy(). + const ProxyRpcRequest& seen() const + { + return seen_; + } + + bool readOk() const + { + return readOk_; + } + + bool writeOk() const + { + return writeOk_; + } + +private: + int fds_[2] = {-1, -1}; + uint32_t abortFlag_ = 0; + ncclSocket sock_{}; + ncclProxyConnector conn_{}; + std::unique_ptr proxyState_; + std::unique_ptr comm_; + ProxyRpcRequest seen_; + bool readOk_ = false; + bool writeOk_ = false; + std::thread proxy_; +}; +} // namespace + +// Delayed reply forces InProgress polls; the call must keep polling and return the proxy's payload. +TEST(ProxyTests, ProxyCallBlockingWaitsForDelayedResponse) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingWaitsForDelayedResponse", + []() { + constexpr uint64_t kReqPayload = 0x1122334455667788ULL; + constexpr uint64_t kRespPayload = 0xA5A5C3C3F00F0FF0ULL; + ProxyCallHarness harness; + ASSERT_TRUE(harness.Init()); + std::vector reply(sizeof(kRespPayload)); + memcpy(reply.data(), &kRespPayload, sizeof(kRespPayload)); + harness.StartReplyingProxy(ncclSuccess, std::move(reply), kProxyCallReplyDelay); + + uint64_t req = kReqPayload; + uint64_t resp = 0; + ncclResult_t res = harness.Call(ncclProxyMsgSetup, &req, sizeof(req), &resp, sizeof(resp)); + harness.JoinProxy(); + + EXPECT_EQ(res, ncclSuccess); + EXPECT_EQ(resp, kRespPayload); + ASSERT_TRUE(harness.readOk()); + EXPECT_TRUE(harness.writeOk()); + const ProxyRpcRequest& seen = harness.seen(); + ExpectProxyRpcHeader(seen, static_cast(sizeof(req)), static_cast(sizeof(resp))); + ASSERT_EQ(seen.req.size(), sizeof(req)); + uint64_t sentReq = 0; + memcpy(&sentReq, seen.req.data(), sizeof(req)); + EXPECT_EQ(sentReq, kReqPayload); + EXPECT_EQ(harness.proxyState()->expectedResponses, nullptr); + }) + .withTimeout(kProxyCallTimeout)); +} + +// A non-InProgress error from the proxy must end the poll loop and be returned unchanged. +TEST(ProxyTests, ProxyCallBlockingReturnsProxyErrorResult) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingReturnsProxyErrorResult", + []() { + ProxyCallHarness harness; + ASSERT_TRUE(harness.Init()); + harness.StartReplyingProxy(ncclInvalidUsage, {}, kNoReplyDelay); + + ncclResult_t res = harness.Call(ncclProxyMsgSetup, nullptr, 0, nullptr, 0); + harness.JoinProxy(); + + EXPECT_EQ(res, ncclInvalidUsage); + ASSERT_TRUE(harness.readOk()); + EXPECT_TRUE(harness.writeOk()); + ExpectProxyRpcHeader(harness.seen(), 0, 0); + EXPECT_EQ(harness.proxyState()->expectedResponses, nullptr); + }) + .withTimeout(kProxyCallTimeout)); +} + +// Abort while waiting is the only exit from a proxy that never replies; the call must return, not hang. +TEST(ProxyTests, ProxyCallBlockingReturnsOnAbortWhileWaiting) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingReturnsOnAbortWhileWaiting", + []() { + ProxyCallHarness harness; + ASSERT_TRUE(harness.Init()); + harness.StartAbortingProxy(); + + ncclResult_t res = harness.Call(ncclProxyMsgSetup, nullptr, 0, nullptr, 0); + harness.JoinProxy(); + + EXPECT_EQ(res, ncclInternalError); + ASSERT_TRUE(harness.readOk()); + ExpectProxyRpcHeader(harness.seen(), 0, 0); + // The proxy consumed exactly one request; a resend while polling would leave bytes here. + ExpectNoPendingBytes(harness.peerFd()); + }) + .withTimeout(kProxyCallTimeout)); +} + +// A send failure must return straight away without polling for a reply that will never come. +TEST(ProxyTests, ProxyCallBlockingReturnsSendErrorWithoutPolling) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingReturnsSendErrorWithoutPolling", + []() { + ProxyCallHarness harness; + ASSERT_TRUE(harness.Init()); + // A poll would consume this foreign reply and fail in expectedProxyResponseStore, so it must remain. + const ncclProxyRpcResponseHeader sentinel + = {reinterpret_cast(uintptr_t{0x5E471E1}), ncclSuccess, 0}; + ASSERT_TRUE(WriteFull(harness.peerFd(), &sentinel, sizeof(sentinel))); + harness.sock().state = ncclSocketStateTerminating; + + ncclResult_t res = harness.Call(ncclProxyMsgSetup, nullptr, 0, nullptr, 0); + + EXPECT_EQ(res, ncclInternalError); + const int clientFd = harness.clientFd(); + ncclProxyRpcResponseHeader pending = {}; + ssize_t pendingLen = recv(clientFd, &pending, sizeof(pending), MSG_DONTWAIT); + EXPECT_EQ(pendingLen, static_cast(sizeof(pending))); + EXPECT_EQ(pending.opId, sentinel.opId); + ExpectNoPendingBytes(harness.peerFd()); + }) + .withTimeout(kProxyCallTimeout)); +} + TEST(ProxyTests, ncclProxyClientGetFdBlocking) { RUN_ISOLATED_TEST( From a39e9d0fc634bbd56cf6dbdc60594534a46b1173 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:27:09 -0700 Subject: [PATCH 05/11] test(rccl): pin proxy response overrun and stale response defects Add two inverted pins for current ncclProxyCallBlocking behaviour: an oversized matching reply is received past the caller's respBuff, and an error exit leaves its entry on proxyState->expectedResponses. Each pin carries the post-fix assertions commented out, so the fix PR flips them instead of writing new tests. --- projects/rccl/test/ProxyTests.cpp | 104 ++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/projects/rccl/test/ProxyTests.cpp b/projects/rccl/test/ProxyTests.cpp index 5f304fc1fb38..1f758e46b51f 100644 --- a/projects/rccl/test/ProxyTests.cpp +++ b/projects/rccl/test/ProxyTests.cpp @@ -24,9 +24,11 @@ #include #include +#include #include #include #include +#include #include #include #include @@ -432,6 +434,14 @@ class ProxyCallHarness shutdown(fds_[0], SHUT_RDWR); } JoinProxy(); + // Mirrors proxy.cc's static expectedProxyResponseFree, so entries a failed call left queued do not leak. + while(proxyState_ != nullptr && proxyState_->expectedResponses != nullptr) + { + ncclExpectedProxyResponse* elem = proxyState_->expectedResponses; + proxyState_->expectedResponses = elem->next; + free(elem->respBuff); + free(elem); + } for(int fd : fds_) { if(fd >= 0) @@ -554,6 +564,19 @@ class ProxyCallHarness bool writeOk_ = false; std::thread proxy_; }; + +// Returns the expected-response entry stored under opId, or nullptr; compares pointer values, never dereferences opId. +const ncclExpectedProxyResponse* FindExpectedResponse(const ncclProxyState* state, const void* opId) +{ + for(const ncclExpectedProxyResponse* elem = state->expectedResponses; elem != nullptr; elem = elem->next) + { + if(elem->opId == opId) + { + return elem; + } + } + return nullptr; +} } // namespace // Delayed reply forces InProgress polls; the call must keep polling and return the proxy's payload. @@ -665,6 +688,87 @@ TEST(ProxyTests, ProxyCallBlockingReturnsSendErrorWithoutPolling) .withTimeout(kProxyCallTimeout)); } +// Known defect: a matching reply larger than respSize overruns respBuff; real guard bytes past respSize absorb it. +TEST(ProxyTests, ProxyCallBlockingOversizedReplyOverrunsRespBuff_PinsKnownDefect) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingOversizedReplyOverrunsRespBuff_PinsKnownDefect", + []() { + constexpr size_t kRespSize = 8; + constexpr size_t kGuardBytes = 16; + constexpr size_t kOverrun = 8; + constexpr unsigned char kGuard = 0xEE; + constexpr unsigned char kReplyByte = 0x5A; + static_assert(kOverrun < kGuardBytes, "the overrun must stay inside the guard, short of its end"); + ProxyCallHarness harness; + ASSERT_TRUE(harness.Init()); + harness.StartReplyingProxy(ncclSuccess, std::vector(kRespSize + kOverrun, kReplyByte), + kNoReplyDelay); + + std::vector respBuff(kRespSize + kGuardBytes, kGuard); + ncclResult_t res + = harness.Call(ncclProxyMsgSetup, nullptr, 0, respBuff.data(), static_cast(kRespSize)); + harness.JoinProxy(); + + ASSERT_TRUE(harness.readOk()); + EXPECT_TRUE(harness.writeOk()); + ExpectProxyRpcHeader(harness.seen(), 0, static_cast(kRespSize)); + // Pin: reply bytes past respSize land in the guard. res is only logged: a fix may reject or truncate. + std::vector overrun(kRespSize + kGuardBytes, kGuard); + std::fill_n(overrun.begin(), kRespSize + kOverrun, kReplyByte); + EXPECT_EQ(respBuff, overrun) << "res " << res + << ": respBuff guard intact, the overrun was fixed, so delete this pin" + << " and uncomment the check below"; + // When this test starts failing because the bug was fixed, delete the pin above and uncomment this. + // const std::vector guard(respBuff.begin() + kRespSize, respBuff.end()); + // EXPECT_EQ(guard, std::vector(kGuardBytes, kGuard)) << "res " << res; + }) + .withTimeout(kProxyCallTimeout)); +} + +// Known defect: ncclPollProxyResponse local error returns leave an entry keyed by the opId ncclProxyCallBlocking frees. +TEST(ProxyTests, ProxyCallBlockingErrorExitLeavesStaleExpectedResponse_PinsKnownDefect) +{ + RUN_ISOLATED_TESTS( + ProcessIsolatedTestRunner::TestConfig( + "ProxyCallBlockingErrorExitLeavesStaleExpectedResponse_PinsKnownDefect", + []() { + ProxyCallHarness aborted; + ASSERT_TRUE(aborted.Init()); + aborted.StartAbortingProxy(); + EXPECT_EQ(aborted.Call(ncclProxyMsgSetup, nullptr, 0, nullptr, 0), ncclInternalError); + aborted.JoinProxy(); + ASSERT_TRUE(aborted.readOk()); + ExpectProxyRpcHeader(aborted.seen(), 0, 0); + + // A correctly sized matching reply with no respBuff hits the NULL respBuff return, not the overrun. + constexpr int kNullRespSize = static_cast(sizeof(uint64_t)); + ProxyCallHarness nullResp; + ASSERT_TRUE(nullResp.Init()); + nullResp.StartReplyingProxy(ncclSuccess, std::vector(kNullRespSize), kNoReplyDelay); + EXPECT_EQ(nullResp.Call(ncclProxyMsgSetup, nullptr, 0, nullptr, kNullRespSize), ncclInternalError); + nullResp.JoinProxy(); + ASSERT_TRUE(nullResp.readOk()); + ExpectProxyRpcHeader(nullResp.seen(), 0, kNullRespSize); + + // seen().opId is the value ncclProxyCallBlocking sent and then freed before returning. + const ncclProxyState* abortedState = aborted.proxyState(); + const ncclProxyState* nullRespState = nullResp.proxyState(); + // Pin: each error exit left the entry for its freed opId queued. + EXPECT_NE(FindExpectedResponse(abortedState, aborted.seen().opId), nullptr) + << "abort exit no longer leaks: delete this pin and uncomment the check below"; + EXPECT_NE(FindExpectedResponse(nullRespState, nullResp.seen().opId), nullptr) + << "NULL respBuff exit no longer leaks: delete this pin and uncomment the check below"; + // When this test starts failing because the bug was fixed, delete the pin above and uncomment this. + // EXPECT_EQ(FindExpectedResponse(abortedState, aborted.seen().opId), nullptr); + // EXPECT_EQ(FindExpectedResponse(nullRespState, nullResp.seen().opId), nullptr); + // EXPECT_EQ(abortedState->expectedResponses, nullptr); + // EXPECT_EQ(nullRespState->expectedResponses, nullptr); + }) + .withTimeout(kProxyCallTimeout)); +} + TEST(ProxyTests, ncclProxyClientGetFdBlocking) { RUN_ISOLATED_TEST( From 9797703f0c4e634b2b75cdd4030df56f94e67675 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:36:59 -0700 Subject: [PATCH 06/11] test(rccl): pin ReduceScatter_LL with position-dependent data Force NCCL_SYM_KERNEL=ReduceScatter_LL, the sole caller of ncclLLA2ASession::send, and check a per-(rank, index) unique float sum so swapped LL data words or misplaced slots fail the oracle. Counts cover the aligned 8-byte and the bounded odd paths. Run on 2 to 8 ranks of one node. Allocate the symmetric windows once and skip collectively when symmetric memory or deviceApiSupport is missing, so the skip names the real reason rather than the ncclInvalidUsage the ineligible forced kernel would return. Only the first count may skip on ncclInvalidUsage, and the per-count checks are MPI-collective so a rank-local failure stops every rank together. Register the test in the mi300x device_api_mpi_lsa config, which already runs 2 co-located ranks with NCCL_CUMEM_ENABLE=1. --- .../rccl/test/SymmetricKernelMPITests.cpp | 104 ++++++++++++++++++ .../configs/mi300x_mellanox_ib.json | 7 +- 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/projects/rccl/test/SymmetricKernelMPITests.cpp b/projects/rccl/test/SymmetricKernelMPITests.cpp index 1f76af6b6ee7..069e37a382b4 100644 --- a/projects/rccl/test/SymmetricKernelMPITests.cpp +++ b/projects/rccl/test/SymmetricKernelMPITests.cpp @@ -27,8 +27,10 @@ #include "MPITestBase.hpp" #include "SymmetricBufferHelpers.hpp" #include "TestChecks.hpp" +#include "nccl_device.h" #include "rccl/rccl.h" +#include #include #include #include @@ -507,4 +509,106 @@ TEST_F(SymmetricKernelCorruptionTest, GroupedOps_Sub8ByteAlignment) } } +// =========================================================================== +// Test group 3: LL all-to-all session data placement +// +// Forces ReduceScatter_LL, the only caller of ncclLLA2ASession::send, and +// sends values unique per (rank, index) so a lane or slot permutation in the +// send path lands a wrong value in a checked output element. +// =========================================================================== + +// A slot shift that leaves a slot unwritten hangs recvReduce instead of failing; only the runner timeout shows it. +TEST_F(SymmetricKernelCorruptionTest, ReduceScatterLL_PositionDependentData) +{ + // ReduceScatter_LL needs a single LSA team: at most 8 ranks on one node. + constexpr int kMaxRanks = 8; + if(!validateTestPrerequisites(2, kMaxRanks, false, 1, 1)) + { + GTEST_SKIP() << "Need 2 to " << kMaxRanks << " MPI ranks on a single node"; + } + + // Read at communicator init, so the guard must outlive createTestCommunicator. + MPIHelpers::MpiEnvGuard symKernelGuard("NCCL_SYM_KERNEL", "ReduceScatter_LL"); + ASSERT_EQ(ncclSuccess, createTestCommunicator()); + + int rank{}; + int nRanks{}; + ASSERT_EQ(ncclSuccess, ncclCommUserRank(getActiveCommunicator(), &rank)); + ASSERT_EQ(ncclSuccess, ncclCommCount(getActiveCommunicator(), &nRanks)); + + // Even counts take the aligned 8-byte path, odd ones the bounded path; 512 floats is one LL iteration per block. + const std::vector counts = {1, 2, 7, 512, 1536, 1537, 1538, 4096, 4097}; + const size_t maxCount = *std::max_element(counts.begin(), counts.end()); + + // value(s, i) = i * 8 + s is injective; the largest sum over 8 ranks is about 2^21, so float addition is exact. + auto value = [](int src, size_t idx) { + return static_cast(idx * kMaxRanks + static_cast(src)); + }; + + // Window registration is collective, so allocate once and agree on the outcome before any rank can skip. + SymBuf sendSym; + SymBuf recvSym; + int allocOk = (tryAllocSymBuf(maxCount * nRanks * sizeof(float), sendSym) && + tryAllocSymBuf(maxCount * sizeof(float), recvSym)) + ? 1 + : 0; + ASSERT_EQ(MPI_SUCCESS, MPI_Allreduce(MPI_IN_PLACE, &allocOk, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD)); + if(!allocOk) + { + GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + } + + // Without symmetricSupport the forced kernel is ineligible; skip with that reason instead of on ncclInvalidUsage. + ncclCommProperties_t props = NCCL_COMM_PROPERTIES_INITIALIZER; + ASSERT_MPI_EQ(ncclSuccess, ncclCommQueryProperties(getActiveCommunicator(), &props)); + int symmetric = props.deviceApiSupport ? 1 : 0; + ASSERT_EQ(MPI_SUCCESS, MPI_Allreduce(MPI_IN_PLACE, &symmetric, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD)); + if(!symmetric) + { + GTEST_SKIP() << "Symmetric kernels unavailable (symmetricSupport off, e.g. cuMem needs Linux >= 6.8)"; + } + + for(size_t recvCount : counts) + { + size_t sendCount = recvCount * static_cast(nRanks); + + ASSERT_MPI_EQ(hipSuccess, + initializeBufferWithPattern( + sendSym.ptr, sendCount, + [rank, &value](size_t i) { return value(rank, i); })); + ASSERT_MPI_EQ(hipSuccess, zeroInitializeBuffer(recvSym.ptr, recvCount)); + + // Tuning is comm-wide, so every rank sees ncclInvalidUsage together; only the first count may skip on it. + ncclResult_t res = ncclReduceScatter(sendSym.ptr, recvSym.ptr, recvCount, ncclFloat, ncclSum, + getActiveCommunicator(), getActiveStream()); + if(res == ncclInvalidUsage && recvCount == counts.front()) + { + GTEST_SKIP() << "ReduceScatter_LL symmetric kernel not eligible on this topology"; + } + ASSERT_MPI_EQ(ncclSuccess, res); + ASSERT_MPI_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); + + size_t base = static_cast(rank) * recvCount; + auto expected = [nRanks, base, &value](size_t j) { + float sum = 0.0f; + for(int s = 0; s < nRanks; ++s) + { + sum += value(s, base + j); + } + return sum; + }; + + size_t errIdx{}; + float expVal{}, actVal{}; + bool ok = verifyBufferData(recvSym.ptr, recvCount, expected, 0, 1e-5, &errIdx, &expVal, &actVal); + if(!ok) + { + ADD_FAILURE() << "ReduceScatter_LL mismatch at recvCount=" << recvCount << " index=" << errIdx + << " expected=" << expVal << " got=" << actVal; + } + // Collective so a rank-local mismatch stops every rank instead of leaving peers in the next LL kernel. + ASSERT_MPI_TRUE(ok); + } +} + #endif // MPI_TESTS_ENABLED diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json index 560c2bd3693a..434d95f40411 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json @@ -4261,6 +4261,11 @@ "name": "DeviceApiMPI_GroupD_CustomRedOp", "description": "Group D: ncclLsaReduceLsaCopy with test-local Min/Max functors (OpSum-only in headers)", "test_filter": "DeviceApiMPITests.Lsa_ReduceLsaCopy_OpMax:DeviceApiMPITests.Lsa_ReduceLsaCopy_OpMin" + }, + { + "name": "SymmetricKernel_ReduceScatterLL_PositionDependentData", + "description": "Forced ReduceScatter_LL symmetric kernel with per-(rank, index) unique data; skips when the kernel is ineligible", + "test_filter": "SymmetricKernelCorruptionTest.ReduceScatterLL_PositionDependentData" } ] }, @@ -5024,7 +5029,7 @@ { "name": "Device API MPI - LSA Reduce/Copy", - "description": "DeviceApiMPITests memory/LSA pointer, barrier, reduce/copy, and custom RedOp coverage on 2 co-located ranks (NCCL_CUMEM_ENABLE=1)", + "description": "DeviceApiMPITests memory/LSA pointer, barrier, reduce/copy, and custom RedOp coverage, plus the ReduceScatter_LL placement test, on 2 co-located ranks (NCCL_CUMEM_ENABLE=1)", "config": "device_api_mpi_lsa", "enabled": true }, From abca9a911b2bf6edca642155c45f8de8d9334c3f Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Wed, 30 Sep 2026 22:37:41 -0700 Subject: [PATCH 07/11] test(rccl): add LSA barrier epoch persistence MPI test Barrier_Lsa_EpochPersistsAcrossLaunches checks that each launch's LSA barrier session starts from the epoch the previous session's destructor saved, so the unicast barrier keeps ordering a cross-rank handoff over repeated launches. --- projects/rccl/test/DeviceApiMPITests.cpp | 90 +++++++++++++++++++++++- 1 file changed, 89 insertions(+), 1 deletion(-) diff --git a/projects/rccl/test/DeviceApiMPITests.cpp b/projects/rccl/test/DeviceApiMPITests.cpp index d67422727428..4ed95e9801a6 100644 --- a/projects/rccl/test/DeviceApiMPITests.cpp +++ b/projects/rccl/test/DeviceApiMPITests.cpp @@ -805,10 +805,14 @@ TEST_F(DeviceApiMPITests, Pointer_Host_GetLsaDevicePointer) { ASSERT_EQ(hostPeerPtr, hostPeerViaWorld); } +// startEpoch, when non-null, receives the session's starting epoch so the host can track it across launches. __global__ void barrierSyncHandoffKernel(ncclWindow_t win, int dstLsaPeer, int writeVal, int* readBack, - ncclDevComm devComm, int mode) { + ncclDevComm devComm, int mode, uint32_t* startEpoch = nullptr) { ncclCoopCta coop = ncclCoopCta(); ncclLsaBarrierSession bar(coop, devComm, ncclTeamLsa(devComm), devComm.lsaBarrier, blockIdx.x); + if (startEpoch != nullptr && threadIdx.x == 0) { + startEpoch[0] = bar.epoch; + } bar.sync(coop, cuda::memory_order_acquire); if (mode == 0 && threadIdx.x == 0) { int* peer = static_cast(ncclGetLsaPointer(win, 0, dstLsaPeer)); @@ -871,6 +875,90 @@ TEST_F(DeviceApiMPITests, Barrier_Lsa_SyncOrdering) { ASSERT_MPI_EQ_ON_RANK(rank, lsaBase + dstLsaPeer, kValue, hostRead); } +// The session destructor saves its epoch; without it the next launch restarts from a stale epoch and its waits no-op. +TEST_F(DeviceApiMPITests, Barrier_Lsa_EpochPersistsAcrossLaunches) { + if (auto reason = lsaSkipReason(); !reason.empty()) { + GTEST_SKIP() << reason; + } + if (!validateTestPrerequisites(/*min_processes=*/2, /*max_processes=*/8)) { + GTEST_SKIP() << "Requires 2-8 ranks"; + } + + ASSERT_EQ(ncclSuccess, createTestCommunicator()); + ncclComm_t comm = getActiveCommunicator(); + hipStream_t stream = getActiveStream(); + + ncclDevCommRequirements reqs = NCCL_DEV_COMM_REQUIREMENTS_INITIALIZER; + reqs.lsaBarrierCount = 1; + ncclDevComm devComm{}; + ASSERT_MPI_EQ(ncclSuccess, ncclDevCommCreate(comm, &reqs, &devComm)); + auto devCommCleanup = makeScopeGuard([&]() { (void)ncclDevCommDestroy(comm, &devComm); }); + + int rank = -1; + ncclCommUserRank(comm, &rank); + const int lsaSize = nodeLocalRanks(); + const int lsaBase = lsaBaseRank(rank, lsaSize); + const int myLsa = rank - lsaBase; + + void* dWin = nullptr; + ASSERT_MPI_EQ(ncclSuccess, ncclMemAlloc(&dWin, sizeof(int))); + auto memCleanup = makeScopeGuard([&]() { + if (dWin) { + (void)ncclMemFree(dWin); + } + }); + ncclWindow_t win = nullptr; + ASSERT_MPI_EQ(ncclSuccess, ncclCommWindowRegister(comm, dWin, sizeof(int), &win, NCCL_WIN_COLL_SYMMETRIC)); + auto winCleanup = makeScopeGuard([&]() { + if (win) { + (void)ncclCommWindowDeregister(comm, win); + } + }); + ASSERT_MPI_EQ(hipSuccess, hipMemset(dWin, 0, sizeof(int))); + + int* dRead = nullptr; + ASSERT_MPI_EQ(hipSuccess, hipMalloc(&dRead, sizeof(int))); + auto readCleanup = makeScopeGuard([&]() { + if (dRead) { + (void)hipFree(dRead); + } + }); + uint32_t* dEpoch = nullptr; + ASSERT_MPI_EQ(hipSuccess, hipMalloc(&dEpoch, sizeof(uint32_t))); + auto epochCleanup = makeScopeGuard([&]() { + if (dEpoch) { + (void)hipFree(dEpoch); + } + }); + + constexpr int kLaunches = 3; + constexpr uint32_t kSyncsPerLaunch = 2; // non-multimem: each sync advances the session epoch by one + constexpr int kValueBase = 4242; + const int dstLsaPeer = 1; // rank 0 writes into LSA peer 1's slot (lsaSize >= 2) + const int mode = (myLsa == 0) ? 0 : (myLsa == dstLsaPeer) ? 1 : 2; + constexpr uint32_t kInitialEpoch = 0; // ncclDevCommCreate zeroes the barrier state in its resource window + for (int launch = 0; launch < kLaunches; ++launch) { + const int writeVal = kValueBase + launch; // distinct per launch so a stale window value cannot satisfy the check + ASSERT_MPI_EQ(hipSuccess, hipMemset(dRead, 0, sizeof(int))); + ASSERT_MPI_EQ(hipSuccess, hipMemset(dEpoch, 0xff, sizeof(uint32_t))); // unwritten marker, never a valid epoch here + MPI_Barrier(MPI_COMM_WORLD); + barrierSyncHandoffKernel<<<1, 1, 0, stream>>>(win, dstLsaPeer, writeVal, dRead, devComm, mode, dEpoch); + ASSERT_MPI_EQ(hipSuccess, hipStreamSynchronize(stream)); + + uint32_t startEpoch = 0; + ASSERT_MPI_EQ(hipSuccess, hipMemcpy(&startEpoch, dEpoch, sizeof(uint32_t), hipMemcpyDeviceToHost)); + ASSERT_MPI_EQ(kInitialEpoch + kSyncsPerLaunch * static_cast(launch), startEpoch); + + int hostRead = 0; + const hipError_t hostMemcpyErr = + (rank == lsaBase + dstLsaPeer) + ? hipMemcpy(&hostRead, dRead, sizeof(int), hipMemcpyDeviceToHost) + : hipSuccess; + ASSERT_MPI_HIP_OK_ON_RANK(rank, lsaBase + dstLsaPeer, hostMemcpyErr); + ASSERT_MPI_EQ_ON_RANK(rank, lsaBase + dstLsaPeer, writeVal, hostRead); + } +} + __global__ void barrierArriveWaitKernel(ncclWindow_t win, int dstLsaPeer, int writeVal, int* readBack, ncclDevComm devComm, int mode) { ncclCoopCta coop = ncclCoopCta(); From 2d1ec1f46c607f2bfc80d2541f50acd82822e41a Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Thu, 1 Oct 2026 11:56:24 -0700 Subject: [PATCH 08/11] test(rccl): address ARGUS review on misc tests Fix the BitOps precheckin filter: DIVUP, ROUNDUP and *Hash matched no Suite.Test name, so four suites never ran. Use the Suite.* form. Make the ReduceScatter_LL ncclInvalidUsage skip collective through mpiCoordinatedSkipReason so a diverging verdict cannot hang peers. --- projects/rccl/test/SymmetricKernelMPITests.cpp | 10 +++++++--- .../scripts/test_runner/configs/ci-precheckin.json | 2 +- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/projects/rccl/test/SymmetricKernelMPITests.cpp b/projects/rccl/test/SymmetricKernelMPITests.cpp index 069e37a382b4..bcaf01e9e0a9 100644 --- a/projects/rccl/test/SymmetricKernelMPITests.cpp +++ b/projects/rccl/test/SymmetricKernelMPITests.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #ifdef MPI_TESTS_ENABLED @@ -578,12 +579,15 @@ TEST_F(SymmetricKernelCorruptionTest, ReduceScatterLL_PositionDependentData) [rank, &value](size_t i) { return value(rank, i); })); ASSERT_MPI_EQ(hipSuccess, zeroInitializeBuffer(recvSym.ptr, recvCount)); - // Tuning is comm-wide, so every rank sees ncclInvalidUsage together; only the first count may skip on it. + // Only the first count may skip on ncclInvalidUsage; agree on it collectively so diverging verdicts cannot hang. ncclResult_t res = ncclReduceScatter(sendSym.ptr, recvSym.ptr, recvCount, ncclFloat, ncclSum, getActiveCommunicator(), getActiveStream()); - if(res == ncclInvalidUsage && recvCount == counts.front()) + const std::string ineligible + = mpiCoordinatedSkipReason(res == ncclInvalidUsage && recvCount == counts.front(), + "ReduceScatter_LL symmetric kernel not eligible on this topology"); + if(!ineligible.empty()) { - GTEST_SKIP() << "ReduceScatter_LL symmetric kernel not eligible on this topology"; + GTEST_SKIP() << ineligible; } ASSERT_MPI_EQ(ncclSuccess, res); ASSERT_MPI_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); diff --git a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json index 85eb4c927647..5c81618f52a1 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json +++ b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json @@ -130,7 +130,7 @@ "NCCL_DEBUG": "INFO" }, "tests": [ - {"name": "BitOps", "description": "Bit operation utility tests (macros and typed fixtures)", "test_filter": "ALIGN_*:DIVUP:ROUNDUP:u32fp*:*Hash:BitOpsTemplate*:Add4G.*"}, + {"name": "BitOps", "description": "Bit operation utility tests (macros and typed fixtures)", "test_filter": "ALIGN_*:DIVUP.*:ROUNDUP.*:u32fp*:getHash.*:eatHash.*:BitOpsTemplate*:Add4G.*"}, {"name": "EnqueueCountTests", "description": "Enqueue send/recv count helpers", "test_filter": "EnqueueCountTests.*"}, {"name": "MiscTests.Sorter", "description": "Task coll sorter unit test", "test_filter": "MiscTests.Sorter"}, {"name": "VersionInfoTests", "description": "HIP/ROCm version string formatting tests", "test_filter": "VersionInfoTests.*"}, From 3faa69766072eaf467850472187eda9acce83fe8 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Thu, 1 Oct 2026 13:18:14 -0700 Subject: [PATCH 09/11] test(rccl): drain ReduceScatter_LL launch before the skip vote Synchronize the stream after a successful ncclReduceScatter before the coordinated ineligibility vote, so a skip can never return into ~SymBuf while a kernel still reads the window it deregisters and frees. --- projects/rccl/test/SymmetricKernelMPITests.cpp | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/projects/rccl/test/SymmetricKernelMPITests.cpp b/projects/rccl/test/SymmetricKernelMPITests.cpp index bcaf01e9e0a9..93da002ebd4a 100644 --- a/projects/rccl/test/SymmetricKernelMPITests.cpp +++ b/projects/rccl/test/SymmetricKernelMPITests.cpp @@ -579,9 +579,11 @@ TEST_F(SymmetricKernelCorruptionTest, ReduceScatterLL_PositionDependentData) [rank, &value](size_t i) { return value(rank, i); })); ASSERT_MPI_EQ(hipSuccess, zeroInitializeBuffer(recvSym.ptr, recvCount)); - // Only the first count may skip on ncclInvalidUsage; agree on it collectively so diverging verdicts cannot hang. + // Only the first count may skip on ncclInvalidUsage; agree collectively so diverging verdicts cannot hang. ncclResult_t res = ncclReduceScatter(sendSym.ptr, recvSym.ptr, recvCount, ncclFloat, ncclSum, getActiveCommunicator(), getActiveStream()); + // Drain before the vote so a skip never leaves a kernel reading a window that ~SymBuf deregisters and frees. + hipError_t syncErr = (res == ncclSuccess) ? hipStreamSynchronize(getActiveStream()) : hipSuccess; const std::string ineligible = mpiCoordinatedSkipReason(res == ncclInvalidUsage && recvCount == counts.front(), "ReduceScatter_LL symmetric kernel not eligible on this topology"); @@ -590,7 +592,7 @@ TEST_F(SymmetricKernelCorruptionTest, ReduceScatterLL_PositionDependentData) GTEST_SKIP() << ineligible; } ASSERT_MPI_EQ(ncclSuccess, res); - ASSERT_MPI_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); + ASSERT_MPI_EQ(hipSuccess, syncErr); size_t base = static_cast(rank) * recvCount; auto expected = [nRanks, base, &value](size_t j) { From 8432dc26b670e0e0978e4689a510a636570e2883 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Thu, 1 Oct 2026 15:24:47 -0700 Subject: [PATCH 10/11] test(rccl): coordinate skip votes in SymmetricKernelCorruptionTest Every case skipped on its own rank's tryAllocSymBuf result, and the short-circuited allocation chain let a failing rank stop registering windows while its peers kept going. Add allocSymBufsSkipReason, which registers every buffer regardless and then votes through mpiCoordinatedSkipReason, and use it in all seven cases. --- .../rccl/test/SymmetricKernelMPITests.cpp | 76 +++++++++++-------- 1 file changed, 45 insertions(+), 31 deletions(-) diff --git a/projects/rccl/test/SymmetricKernelMPITests.cpp b/projects/rccl/test/SymmetricKernelMPITests.cpp index 93da002ebd4a..61cbe354c266 100644 --- a/projects/rccl/test/SymmetricKernelMPITests.cpp +++ b/projects/rccl/test/SymmetricKernelMPITests.cpp @@ -33,8 +33,10 @@ #include #include #include +#include #include #include +#include #include #ifdef MPI_TESTS_ENABLED @@ -75,6 +77,17 @@ class SymmetricKernelCorruptionTest : public MPITestBase { return allocSymBuf(bytes, sb) == ncclSuccess; } + + // Registers every buffer even after a failure, so all ranks issue the same collective registrations, then votes. + std::string allocSymBufsSkipReason(std::initializer_list> bufs) + { + bool ok = true; + for(const auto& [bytes, sb] : bufs) + { + ok = tryAllocSymBuf(bytes, *sb) && ok; + } + return mpiCoordinatedSkipReason(!ok, "Symmetric memory not available (VMM/cuMem unsupported)"); + } }; // =========================================================================== @@ -110,10 +123,10 @@ TEST_F(SymmetricKernelCorruptionTest, AllGather_Sub8ByteAlignment) size_t recvBytes = count * static_cast(nRanks) * sizeof(float); SymBuf sendSym, recvSym; - if(!tryAllocSymBuf(sendBytes, sendSym) || - !tryAllocSymBuf(recvBytes, recvSym)) + const std::string noSym = allocSymBufsSkipReason({{sendBytes, &sendSym}, {recvBytes, &recvSym}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } ASSERT_EQ(hipSuccess, @@ -166,10 +179,10 @@ TEST_F(SymmetricKernelCorruptionTest, ReduceScatter_Sub8ByteAlignment) size_t recvBytes = recvCount * sizeof(float); SymBuf sendSym, recvSym; - if(!tryAllocSymBuf(sendBytes, sendSym) || - !tryAllocSymBuf(recvBytes, recvSym)) + const std::string noSym = allocSymBufsSkipReason({{sendBytes, &sendSym}, {recvBytes, &recvSym}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } // Each rank fills its send buffer with float(rank + 1) at every position. @@ -242,10 +255,11 @@ TEST_F(SymmetricKernelCorruptionTest, GroupedAllGather_VaryingSizes) sendBufs[i] = std::make_unique(); recvBufs[i] = std::make_unique(); - if(!tryAllocSymBuf(sendBytes, *sendBufs[i]) || - !tryAllocSymBuf(recvBytes, *recvBufs[i])) + const std::string noSym + = allocSymBufsSkipReason({{sendBytes, sendBufs[i].get()}, {recvBytes, recvBufs[i].get()}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } ASSERT_EQ(hipSuccess, @@ -314,10 +328,11 @@ TEST_F(SymmetricKernelCorruptionTest, GroupedReduceScatter_VaryingSizes) sendBufs[i] = std::make_unique(); recvBufs[i] = std::make_unique(); - if(!tryAllocSymBuf(sendBytes, *sendBufs[i]) || - !tryAllocSymBuf(recvBytes, *recvBufs[i])) + const std::string noSym + = allocSymBufsSkipReason({{sendBytes, sendBufs[i].get()}, {recvBytes, recvBufs[i].get()}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } ASSERT_EQ(hipSuccess, @@ -374,12 +389,13 @@ TEST_F(SymmetricKernelCorruptionTest, GroupedMixed_AllGatherAndReduceScatter) const size_t rsSendCount = rsRecvCount * static_cast(nRanks); SymBuf agSend, agRecv, rsSend, rsRecv; - if(!tryAllocSymBuf(agCount * sizeof(float), agSend) || - !tryAllocSymBuf(agCount * nRanks * sizeof(float), agRecv) || - !tryAllocSymBuf(rsSendCount * sizeof(float), rsSend) || - !tryAllocSymBuf(rsRecvCount * sizeof(float), rsRecv)) + const std::string noSym = allocSymBufsSkipReason({{agCount * sizeof(float), &agSend}, + {agCount * nRanks * sizeof(float), &agRecv}, + {rsSendCount * sizeof(float), &rsSend}, + {rsRecvCount * sizeof(float), &rsRecv}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } ASSERT_EQ(hipSuccess, @@ -452,12 +468,13 @@ TEST_F(SymmetricKernelCorruptionTest, GroupedOps_Sub8ByteAlignment) const size_t rsSendCount = rsRecvCount * static_cast(nRanks); SymBuf agSend, agRecv, rsSend, rsRecv; - if(!tryAllocSymBuf(agCount * sizeof(float), agSend) || - !tryAllocSymBuf(agCount * nRanks * sizeof(float), agRecv) || - !tryAllocSymBuf(rsSendCount * sizeof(float), rsSend) || - !tryAllocSymBuf(rsRecvCount * sizeof(float), rsRecv)) + const std::string noSym = allocSymBufsSkipReason({{agCount * sizeof(float), &agSend}, + {agCount * nRanks * sizeof(float), &agRecv}, + {rsSendCount * sizeof(float), &rsSend}, + {rsRecvCount * sizeof(float), &rsRecv}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } ASSERT_EQ(hipSuccess, @@ -547,16 +564,13 @@ TEST_F(SymmetricKernelCorruptionTest, ReduceScatterLL_PositionDependentData) }; // Window registration is collective, so allocate once and agree on the outcome before any rank can skip. - SymBuf sendSym; - SymBuf recvSym; - int allocOk = (tryAllocSymBuf(maxCount * nRanks * sizeof(float), sendSym) && - tryAllocSymBuf(maxCount * sizeof(float), recvSym)) - ? 1 - : 0; - ASSERT_EQ(MPI_SUCCESS, MPI_Allreduce(MPI_IN_PLACE, &allocOk, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD)); - if(!allocOk) + SymBuf sendSym; + SymBuf recvSym; + const std::string noSym + = allocSymBufsSkipReason({{maxCount * nRanks * sizeof(float), &sendSym}, {maxCount * sizeof(float), &recvSym}}); + if(!noSym.empty()) { - GTEST_SKIP() << "Symmetric memory not available (VMM/cuMem unsupported)"; + GTEST_SKIP() << noSym; } // Without symmetricSupport the forced kernel is ineligible; skip with that reason instead of on ncclInvalidUsage. From e8fdafb98531eb5b4aecc1d4f1079a3a5f1e7ba1 Mon Sep 17 00:00:00 2001 From: Praneeth Vallem Date: Fri, 2 Oct 2026 13:55:46 -0700 Subject: [PATCH 11/11] test(rccl): add binary-mode Recorder RedOp record tests (#12704) Only the JSON path of Recorder RedOp records is tested, so raw rcclApiCall fields written in binary mode can regress unseen. Fix: read both RedOp records back from the binary dump and check each field; stacked on #12641. JIRA ID: AICOMRCCL-2699 --- projects/rccl/test/_RecorderTests.cpp | 62 +++++++++++++++++++ .../test_runner/configs/ci-precheckin.json | 1 + .../configs/mi300_mellanox_ib_func.json | 5 ++ .../configs/mi300x_mellanox_ib.json | 1 + .../configs/mi355x_thor2_roce.json | 5 ++ .../configs/mi355x_thor2_roce_func.json | 5 ++ .../test_runner/configs/mi455_ainic_roce.json | 5 ++ .../configs/mi455_ainic_roce_func.json | 5 ++ 8 files changed, 89 insertions(+) diff --git a/projects/rccl/test/_RecorderTests.cpp b/projects/rccl/test/_RecorderTests.cpp index 3a406970627f..0379ac5a0fea 100644 --- a/projects/rccl/test/_RecorderTests.cpp +++ b/projects/rccl/test/_RecorderTests.cpp @@ -13,6 +13,7 @@ #include #include +#include #include #define HIPCALL(cmd) \ @@ -134,6 +135,67 @@ namespace RcclUnitTesting ); } + /** + * \brief Verify binary mode writes PreMulSum scalar/datatype/residence as sendbuff/datatype/root; Destroy drops them + * ******************************************************************************************/ + TEST(Recorder, RedOpRecordBinary) + { + RUN_ISOLATED_TEST_WITH_ENV( + "RedOpRecordBinary", + []() + { + int ranks[] = {2, 3, 5}; + auto comm = std::make_unique(); + comm->nRanks = 4; + comm->localRank = 1; + comm->localRankToRank = ranks; + float scalar = 2.0f; + double ignored = 4.0; + // Non-default datatype and residence, so a dropped or inverted PreMulSum guard cannot match the defaults. + ASSERT_EQ(ncclSuccess, rccl::Recorder::instance().record(rccl::rrRedOpCreatePreMulSum, ncclMax, comm.get(), + ncclFloat32, ncclScalarHostImmediate, &scalar)); + // Destroy gets non-default scalar arguments too, so a guard that always copies them is caught. + ASSERT_EQ(ncclSuccess, rccl::Recorder::instance().record(rccl::rrRedOpDestroy, ncclMax, comm.get(), + ncclFloat64, ncclScalarHostImmediate, &ignored)); + + char host[256]; + gethostname(host, sizeof(host)); + std::string filename = "/tmp/test_redop." + std::to_string(getpid()) + "." + std::string(host) + ".bin"; + std::ifstream fp(filename, std::ios::binary); + ASSERT_TRUE(fp.is_open()) << "Recorder did not create expected file: " << filename; + // One spare slot so a stray extra record shows up in bytesRead. + rccl::rcclApiCall calls[3]; + fp.read(reinterpret_cast(calls), sizeof(calls)); + std::streamsize bytesRead = fp.gcount(); + fp.close(); + std::remove(filename.c_str()); + // Binary mode writes exactly one raw struct per call and no header. + ASSERT_EQ(bytesRead, static_cast(2 * sizeof(rccl::rcclApiCall))); + + const rccl::rcclApiCall& create = calls[0]; + EXPECT_EQ(create.type, rccl::rrRedOpCreatePreMulSum); + EXPECT_EQ(create.op, ncclMax); + EXPECT_EQ(create.comm, comm.get()); + // nRanks and globalRank are taken from the comm and only reach disk in binary mode. + EXPECT_EQ(create.nRanks, 4); + EXPECT_EQ(create.globalRank, 3); + EXPECT_EQ(create.sendbuff, static_cast(&scalar)) << "sendbuff must come from scalar"; + EXPECT_EQ(create.datatype, ncclFloat32) << "datatype must come from the datatype argument"; + EXPECT_EQ(create.root, static_cast(ncclScalarHostImmediate)) << "root must come from residence"; + + const rccl::rcclApiCall& destroy = calls[1]; + EXPECT_EQ(destroy.type, rccl::rrRedOpDestroy); + EXPECT_EQ(destroy.op, ncclMax); + EXPECT_EQ(destroy.comm, comm.get()); + // record() builds from the aggregate ncclInfo{.op, .comm}, so fields Destroy must not set are zero. + EXPECT_EQ(destroy.sendbuff, nullptr) << "Destroy must not record the scalar"; + EXPECT_EQ(destroy.datatype, ncclInt8) << "Destroy must not record the datatype"; + EXPECT_EQ(destroy.root, 0) << "Destroy must not record the residence"; + }, + {{"RCCL_REPLAY_FILE", "/tmp/test_redop.bin"}} + ); + } + /** * \brief Verify RCCL Recorder's integrity in multithread context by comparing Recorder * instance across different threads. diff --git a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json index 5c81618f52a1..741d4954ddef 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json +++ b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json @@ -86,6 +86,7 @@ {"name": "Recorder.ParseJson", "description": "Recorder parse JSON tests", "test_filter": "Recorder.ParseJson", "timeout": 60}, {"name": "Recorder.VerifyMultithread", "description": "Recorder multithread verification tests", "test_filter": "Recorder.VerifyMultithread", "timeout": 60}, {"name": "Recorder.RedOpRecordJson", "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson", "timeout": 60}, + {"name": "Recorder.RedOpRecordBinary", "description": "Recorder RedOp record binary output", "test_filter": "Recorder.RedOpRecordBinary", "timeout": 60}, {"name": "ProxyTraceTestFixture.nonEmptySingleton", "description": "ProxyTrace non-empty singleton test", "test_filter": "ProxyTraceTestFixture.nonEmptySingleton", "timeout": 60}, {"name": "ProxyTraceTestFixture.addTraceOp", "description": "ProxyTrace add trace operation test", "test_filter": "ProxyTraceTestFixture.addTraceOp", "timeout": 60}, {"name": "ProxyTraceTestFixture.getMapSizeMB", "description": "ProxyTrace get map size test", "test_filter": "ProxyTraceTestFixture.getMapSizeMB", "timeout": 60}, diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json index 493eedda8db5..723db355e34f 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json @@ -1833,6 +1833,11 @@ "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson" }, + { + "name": "Recorder.RedOpRecordBinary", + "description": "Recorder RedOp record binary output", + "test_filter": "Recorder.RedOpRecordBinary" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json index 434d95f40411..18c560f58e55 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json @@ -1196,6 +1196,7 @@ {"name": "Recorder.ParseJson", "description": "Recorder parse JSON trace", "test_filter": "Recorder.ParseJson"}, {"name": "Recorder.VerifyMultithread", "description": "Recorder multithreaded verification", "test_filter": "Recorder.VerifyMultithread"}, {"name": "Recorder.RedOpRecordJson", "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson"}, + {"name": "Recorder.RedOpRecordBinary", "description": "Recorder RedOp record binary output", "test_filter": "Recorder.RedOpRecordBinary"}, {"name": "IbvSymbolsEnv.DefaultLoadsLibrary", "description": "NCCL_IBVERBS_LIB unset loads default libibverbs", "test_filter": "IbvSymbolsEnv.DefaultLoadsLibrary"}, {"name": "IbvSymbolsEnv.EnvOverrideValidSoname", "description": "NCCL_IBVERBS_LIB override to a valid soname is honored", "test_filter": "IbvSymbolsEnv.EnvOverrideValidSoname"}, {"name": "IbvSymbolsEnv.BogusPathFallsBackToDefault", "description": "NCCL_IBVERBS_LIB bogus path falls back to default libibverbs", "test_filter": "IbvSymbolsEnv.BogusPathFallsBackToDefault"}, diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json index 46b968edf3b6..aaa08a78e799 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json @@ -1427,6 +1427,11 @@ "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson" }, + { + "name": "Recorder.RedOpRecordBinary", + "description": "Recorder RedOp record binary output", + "test_filter": "Recorder.RedOpRecordBinary" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json index 5e53129a6129..f735df4d386f 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json @@ -1883,6 +1883,11 @@ "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson" }, + { + "name": "Recorder.RedOpRecordBinary", + "description": "Recorder RedOp record binary output", + "test_filter": "Recorder.RedOpRecordBinary" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json index b3d7c47e1d40..0fb4550416dd 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json @@ -1970,6 +1970,11 @@ "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson" }, + { + "name": "Recorder.RedOpRecordBinary", + "description": "Recorder RedOp record binary output", + "test_filter": "Recorder.RedOpRecordBinary" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json index 17c3afb09e74..2f2058c24994 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json @@ -2024,6 +2024,11 @@ "description": "Recorder RedOp record JSON output", "test_filter": "Recorder.RedOpRecordJson" }, + { + "name": "Recorder.RedOpRecordBinary", + "description": "Recorder RedOp record binary output", + "test_filter": "Recorder.RedOpRecordBinary" + }, { "name": "ProxyTrace.nonEmptySingleton", "description": "Proxy trace non-empty singleton",