From bbc9492185346156d81461bde448889cb6d8dd03 Mon Sep 17 00:00:00 2001 From: Changho Hwang Date: Tue, 23 Jun 2026 16:26:49 +0000 Subject: [PATCH 1/2] Update port channel perf tests --- test/mp_unit/port_channel_tests.cu | 107 ++++++++++++++++++++--------- 1 file changed, 76 insertions(+), 31 deletions(-) diff --git a/test/mp_unit/port_channel_tests.cu b/test/mp_unit/port_channel_tests.cu index 47034cdb9..eec1760cf 100644 --- a/test/mp_unit/port_channel_tests.cu +++ b/test/mp_unit/port_channel_tests.cu @@ -629,22 +629,19 @@ PERF_TEST(PortChannelOneToOneTest, BandwidthIbHostNoAtomicMode) { static constexpr int kMaxQps = 4; __constant__ DeviceHandle gMultiQpPortChans[kMaxQps]; -// Multi-QP bandwidth kernel: barrier on QP 0 only, then putWithSignal on all QPs. -// Only one signal/wait pair is needed for sync between two GPU kernels. +// Multi-QP bandwidth kernel: one thread per QP, putWithSignal per QP, parallel waits. __global__ void kernelMultiQpBandwidth(int nElemPerChan, int nIters, int numQps) { - if (threadIdx.x != 0) return; + int q = threadIdx.x; + if (q >= numQps) return; for (int i = 0; i < nIters; i++) { - // Barrier on QP 0 only — syncs both ranks - gMultiQpPortChans[0].signal(); - gMultiQpPortChans[0].wait(); - // Data transfer: put on all QPs simultaneously - for (int q = 0; q < numQps; q++) { - gMultiQpPortChans[q].putWithSignal(0, nElemPerChan * sizeof(int)); - } - // Wait for all remote data arrivals - for (int q = 0; q < numQps; q++) { - gMultiQpPortChans[q].wait(); + if (q == 0) { + gMultiQpPortChans[0].signal(); + gMultiQpPortChans[0].wait(); } + __syncthreads(); + gMultiQpPortChans[q].putWithSignal(0, nElemPerChan * sizeof(int)); + gMultiQpPortChans[q].wait(); + __syncthreads(); } } @@ -715,15 +712,15 @@ void PortChannelOneToOneTest::testMultiQpBandwidth(IbMode ibMode, int numQps) { for (int nElemPerChan : {256, 16 * 1024, 256 * 1024, 1024 * 1024, 4 * 1024 * 1024, 16 * 1024 * 1024, 32 * 1024 * 1024}) { - int nIters = 10000; + int nIters = 200; // Warm-up - kernelMultiQpBandwidth<<<1, 1>>>(nElemPerChan, 10, numQps); + kernelMultiQpBandwidth<<<1, numQps>>>(nElemPerChan, 10, numQps); MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); communicator->bootstrap()->barrier(); // Measure mscclpp::Timer timer; - kernelMultiQpBandwidth<<<1, 1>>>(nElemPerChan, nIters, numQps); + kernelMultiQpBandwidth<<<1, numQps>>>(nElemPerChan, nIters, numQps); MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); double elapsedUs = timer.elapsed(); communicator->bootstrap()->barrier(); @@ -748,20 +745,46 @@ void PortChannelOneToOneTest::testMultiQpBandwidth(IbMode ibMode, int numQps) { for (auto& m : remoteMems) registeredMemories.push_back(m); } +PERF_TEST(PortChannelOneToOneTest, SingleQpBandwidthIbHostMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testMultiQpBandwidth(IbMode::Host, /*numQps=*/1); +} + +PERF_TEST(PortChannelOneToOneTest, TwoQpBandwidthIbHostMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testMultiQpBandwidth(IbMode::Host, /*numQps=*/2); +} + PERF_TEST(PortChannelOneToOneTest, MultiQpBandwidthIbHostMode) { REQUIRE_IBVERBS; REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); - for (int numQps : {1, 2, 4}) { - testMultiQpBandwidth(IbMode::Host, numQps); - } + testMultiQpBandwidth(IbMode::Host, /*numQps=*/4); +} + +PERF_TEST(PortChannelOneToOneTest, SingleQpBandwidthIbHostNoAtomicMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + testMultiQpBandwidth(IbMode::HostNoAtomic, /*numQps=*/1); +} + +PERF_TEST(PortChannelOneToOneTest, TwoQpBandwidthIbHostNoAtomicMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + testMultiQpBandwidth(IbMode::HostNoAtomic, /*numQps=*/2); +} + +PERF_TEST(PortChannelOneToOneTest, ThreeQpBandwidthIbHostNoAtomicMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + testMultiQpBandwidth(IbMode::HostNoAtomic, /*numQps=*/3); } PERF_TEST(PortChannelOneToOneTest, MultiQpBandwidthIbHostNoAtomicMode) { REQUIRE_IBVERBS; REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); - for (int numQps : {1, 2, 4}) { - testMultiQpBandwidth(IbMode::HostNoAtomic, numQps); - } + testMultiQpBandwidth(IbMode::HostNoAtomic, /*numQps=*/4); } // Multi-QP flush-stress kernel: one thread per QP, all calling putWithSignalAndFlush @@ -786,7 +809,7 @@ void PortChannelOneToOneTest::testMultiQpFlushStress(IbMode ibMode, int numQps) if (gEnv->rank >= numRanksToUse) return; const int rank = communicator->bootstrap()->getRank(); - const int maxElemPerChan = 64 * 1024; + const int maxElemPerChan = 8 * 1024 * 1024; std::vector> sendBuffs; std::vector localMems; @@ -805,8 +828,8 @@ void PortChannelOneToOneTest::testMultiQpFlushStress(IbMode ibMode, int numQps) const std::string qpLabel = std::to_string(numQps) + " QP" + (numQps > 1 ? "s" : ""); - for (int nElemPerChan : {256, 4 * 1024, 64 * 1024}) { - int nIters = 2000; + for (int nElemPerChan : {256, 4 * 1024, 64 * 1024, 256 * 1024, 1024 * 1024, 4 * 1024 * 1024, 8 * 1024 * 1024}) { + int nIters = (nElemPerChan >= 256 * 1024) ? 200 : 2000; kernelMultiQpFlushStress<<<1, numQps>>>(nElemPerChan, 10, numQps); MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); communicator->bootstrap()->barrier(); @@ -823,8 +846,10 @@ void PortChannelOneToOneTest::testMultiQpFlushStress(IbMode ibMode, int numQps) int bytesPerChan = nElemPerChan * (int)sizeof(int); std::string sizeLabel = (bytesPerChan >= 1024) ? (std::to_string(bytesPerChan / 1024) + " KB") : (std::to_string(bytesPerChan) + " B"); + double aggGbps = ((double)bytesPerChan * numQps) / usPerIter * 1e-3; // bytes/us = MB/s × 1e-3 = GB/s ::mscclpp::test::reportPerfResult(sizeLabel + " (" + qpLabel + ") per-iter", usPerIter, "us"); ::mscclpp::test::reportPerfResult(sizeLabel + " (" + qpLabel + ") per-iter/QP", usPerIterPerQp, "us"); + ::mscclpp::test::reportPerfResult(sizeLabel + " (" + qpLabel + ") aggregate", aggGbps, "GB/s"); } } @@ -834,20 +859,40 @@ void PortChannelOneToOneTest::testMultiQpFlushStress(IbMode ibMode, int numQps) for (auto& m : remoteMems) registeredMemories.push_back(m); } +PERF_TEST(PortChannelOneToOneTest, SingleQpFlushStressIbHostMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testMultiQpFlushStress(IbMode::Host, /*numQps=*/1); +} + +PERF_TEST(PortChannelOneToOneTest, TwoQpFlushStressIbHostMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testMultiQpFlushStress(IbMode::Host, /*numQps=*/2); +} + PERF_TEST(PortChannelOneToOneTest, MultiQpFlushStressIbHostMode) { REQUIRE_IBVERBS; REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); - for (int numQps : {1, 2, 4}) { - testMultiQpFlushStress(IbMode::Host, numQps); - } + testMultiQpFlushStress(IbMode::Host, /*numQps=*/4); +} + +PERF_TEST(PortChannelOneToOneTest, SingleQpFlushStressIbHostNoAtomicMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + testMultiQpFlushStress(IbMode::HostNoAtomic, /*numQps=*/1); +} + +PERF_TEST(PortChannelOneToOneTest, TwoQpFlushStressIbHostNoAtomicMode) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + testMultiQpFlushStress(IbMode::HostNoAtomic, /*numQps=*/2); } PERF_TEST(PortChannelOneToOneTest, MultiQpFlushStressIbHostNoAtomicMode) { REQUIRE_IBVERBS; REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); - for (int numQps : {1, 2, 4}) { - testMultiQpFlushStress(IbMode::HostNoAtomic, numQps); - } + testMultiQpFlushStress(IbMode::HostNoAtomic, /*numQps=*/4); } // Same-channel concurrent-flush kernel: N GPU threads on the same PortChannel each call From 4c624ce09e0f3a6f26b037245c1ac92bd9e0798e Mon Sep 17 00:00:00 2001 From: Changho Hwang Date: Tue, 23 Jun 2026 17:01:44 +0000 Subject: [PATCH 2/2] Apply IB CI fixes --- include/mscclpp/core.hpp | 2 +- src/core/ib.cc | 7 ++++--- src/core/include/ib.hpp | 1 + src/core/port_channel.cc | 14 ++++++++++++-- 4 files changed, 18 insertions(+), 6 deletions(-) diff --git a/include/mscclpp/core.hpp b/include/mscclpp/core.hpp index 45b56bcc0..0aa28bb8c 100644 --- a/include/mscclpp/core.hpp +++ b/include/mscclpp/core.hpp @@ -391,7 +391,7 @@ struct EndpointConfig { static constexpr int DefaultPort = -1; static constexpr int DefaultGidIndex = -1; static constexpr int DefaultMaxCqSize = 1024; - static constexpr int DefaultMaxCqPollNum = 1; + static constexpr int DefaultMaxCqPollNum = 256; static constexpr int DefaultMaxSendWr = 8192; static constexpr int DefaultMaxRecvWr = 16; static constexpr int DefaultMaxWrPerSend = 64; diff --git a/src/core/ib.cc b/src/core/ib.cc index 557f04268..3c6ba3e55 100644 --- a/src/core/ib.cc +++ b/src/core/ib.cc @@ -188,6 +188,7 @@ IbQp::IbQp(ibv_context* ctx, ibv_pd* pd, int portNum, int gidIndex, int maxSendC numStagedRecv_(0), numPostedSignaledSend_(0), numStagedSignaledSend_(0), + maxSendCqSize_(maxSendCqSize), maxSendCqPollNum_(maxSendCqPollNum), maxSendWr_(maxSendWr), maxWrPerSend_(maxWrPerSend), @@ -402,9 +403,9 @@ void IbQp::postSend() { numStagedSend_ = 0; numPostedSignaledSend_ += numStagedSignaledSend_; numStagedSignaledSend_ = 0; - if (numPostedSignaledSend_ + 4 > sendCq_->cqe) { - WARN(NET, "IB: CQ is almost full ( ", numPostedSignaledSend_, " / ", sendCq_->cqe, - " ). The connection needs to be flushed to prevent timeout errors."); + if (numPostedSignaledSend_ + 4 > maxSendCqSize_) { + WARN(NET, "IB: CQ is almost full (", numPostedSignaledSend_, " / ", maxSendCqSize_, + "). The connection needs to be flushed to prevent timeout errors."); } } diff --git a/src/core/include/ib.hpp b/src/core/include/ib.hpp index 36c5a2373..31a48a31b 100644 --- a/src/core/include/ib.hpp +++ b/src/core/include/ib.hpp @@ -124,6 +124,7 @@ class IbQp { int numPostedSignaledSend_; int numStagedSignaledSend_; + const int maxSendCqSize_; const int maxSendCqPollNum_; const int maxSendWr_; const int maxWrPerSend_; diff --git a/src/core/port_channel.cc b/src/core/port_channel.cc index 210b2eebe..0601ef84b 100644 --- a/src/core/port_channel.cc +++ b/src/core/port_channel.cc @@ -90,8 +90,8 @@ MSCCLPP_API_CPP void ProxyService::startProxy(bool blocking) { proxy_->start(blo MSCCLPP_API_CPP void ProxyService::stopProxy() { proxy_->stop(); - // Drain pending flushes. After a bounded loop, force-unblock any still-pending GPU - // waiters with a sentinel write (UINT64_MAX > any FIFO position). + // Drain pending TriggerSync flushes. After a bounded loop, force-unblock any still-pending + // GPU waiters with a sentinel write (UINT64_MAX > any FIFO position). for (int i = 0; i < 1000 && !pendingFlushPos_.empty(); ++i) { progressFlushes(); } @@ -102,6 +102,16 @@ MSCCLPP_API_CPP void ProxyService::stopProxy() { } pendingFlushPos_.clear(); } + // Drain any remaining in-flight signaled posts that the proxy issued but never sync-flushed. + for (auto& [conn, count] : inflightRequests_) { + if (count <= 0) continue; + try { + conn->flush(/*timeoutUsec=*/5'000'000); + } catch (const std::exception& e) { + WARN(CONN, "stopProxy: flush failed for a connection (continuing): ", e.what()); + } + } + inflightRequests_.clear(); } ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger trigger) {