From 6363f6f5f5f157dc547fd9f57e0aeaa9303e21cc Mon Sep 17 00:00:00 2001 From: Changho Hwang Date: Thu, 6 Aug 2026 08:11:10 +0000 Subject: [PATCH 1/2] Encode trigger types as opcodes instead of flags `TriggerData`, `TriggerFlag`, and `TriggerSync` were combinable bits, but the device API only ever produced five of the seven combinations, and the proxy tested them with three independent masks. Replace them with one opcode per operation the API can express: TriggerNone = 0 TriggerPut = 1 TriggerSignal = 2 TriggerFlush = 3 TriggerPutWithSignal = 4 TriggerPutWithSignalAndFlush = 5 Seven values fit the existing 3-bit field with one spare. `handleTrigger` becomes a switch that names one operation per case and warns on anything unknown, so a combination nothing emits is now unrepresentable, and a trigger whose type field is unset is no longer a valid operation. The old names are removed rather than redefined. Code that composed them with `|`, or tested them with `&`, must fail to compile: `TriggerFlush` is 3, which has the old `TriggerData` bit set, so a mask test would silently misbehave. No behavior change. mp_unit_tests 29/29 and unit_tests 34/34 on H200. --- include/mscclpp/fifo_device.hpp | 18 +++++++++-- include/mscclpp/port_channel.hpp | 2 +- include/mscclpp/port_channel_device.hpp | 41 ++++++++++++------------ src/core/include/connection.hpp | 2 +- src/core/port_channel.cc | 42 +++++++++++++++++++------ test/mp_unit/port_channel_tests.cu | 4 +-- 6 files changed, 72 insertions(+), 37 deletions(-) diff --git a/include/mscclpp/fifo_device.hpp b/include/mscclpp/fifo_device.hpp index 4670f47c9..192672c75 100644 --- a/include/mscclpp/fifo_device.hpp +++ b/include/mscclpp/fifo_device.hpp @@ -15,10 +15,20 @@ namespace mscclpp { +/// Operation that a trigger asks the proxy to perform. +/// +/// These are opcodes, not flags: compare one by equality, and never combine two. The encoding +/// enumerates the combinations the device API can produce rather than composing them, so a +/// combination nothing emits cannot be expressed, and a trigger whose type field is unset is not +/// a valid operation. using TriggerType = uint64_t; -constexpr TriggerType TriggerData = 0x1; // Trigger a data transfer. -constexpr TriggerType TriggerFlag = 0x2; // Trigger a signaling. -constexpr TriggerType TriggerSync = 0x4; // Trigger a flush. +constexpr TriggerType TriggerNone = 0; // Not an operation. Unwritten or malformed. +constexpr TriggerType TriggerPut = 1; // Transfer data. +constexpr TriggerType TriggerSignal = 2; // Signal the remote semaphore. +constexpr TriggerType TriggerFlush = 3; // Flush the connection. +constexpr TriggerType TriggerPutWithSignal = 4; // Transfer data, then signal. +constexpr TriggerType TriggerPutWithSignalAndFlush = 5; // Transfer data, signal, then flush. +// 6 and 7 are unassigned. constexpr unsigned int TriggerBitsSize = 32; constexpr unsigned int TriggerBitsOffset = 32; @@ -29,6 +39,8 @@ constexpr unsigned int TriggerBitsSemaphoreId = 10; // there. See FifoDeviceHandle::push(). constexpr unsigned int TriggerBitsFifoReserved = 1; +static_assert(TriggerPutWithSignalAndFlush < (1ULL << TriggerBitsType), "trigger opcodes must fit in the type field"); + /// Pair of 64-bit unsigned integers used as a trigger for the proxy. /// Used as a work element in the concurrent FIFO. /// Most significant bit of snd is reserved. diff --git a/include/mscclpp/port_channel.hpp b/include/mscclpp/port_channel.hpp index 18d67524e..b8d5f8501 100644 --- a/include/mscclpp/port_channel.hpp +++ b/include/mscclpp/port_channel.hpp @@ -84,7 +84,7 @@ class ProxyService : public BaseProxyService { std::vector memories_; std::shared_ptr proxy_; std::unordered_map, int> inflightRequests_; - // Latest pending TriggerSync FIFO position per connection. Proxy publishes pos+1 to the + // Latest pending TriggerFlush FIFO position per connection. Proxy publishes pos+1 to the // connection's gpuFlushDonePos_ when the CQ drains, then erases the entry. std::unordered_map, uint64_t> pendingFlushPos_; diff --git a/include/mscclpp/port_channel_device.hpp b/include/mscclpp/port_channel_device.hpp index fd575b4c6..9cf9c8e8c 100644 --- a/include/mscclpp/port_channel_device.hpp +++ b/include/mscclpp/port_channel_device.hpp @@ -19,7 +19,7 @@ using MemoryId = uint32_t; namespace detail { #if defined(MSCCLPP_DEVICE_COMPILE) -/// Wait until the proxy has processed and drained the TriggerSync at FIFO position `fifoPos`. +/// Wait until the proxy has processed and drained the TriggerFlush at FIFO position `fifoPos`. /// The proxy publishes `flushDonePos = latestCompletedPos + 1` when the CQ drains, so the /// wait condition `flushDonePos > fifoPos` is satisfied exactly when our own request has /// been completed. Using the FIFO push position as the wait target couples the wait to the @@ -51,7 +51,7 @@ struct BasePortChannelDeviceHandle { : semaphoreId_(semaphoreId), semaphore_(semaphore), fifo_(fifo), flushDonePos_(flushDonePos) {} #if defined(MSCCLPP_DEVICE_COMPILE) - /// Push a TriggerData to the FIFO. + /// Push a TriggerPut to the FIFO. /// @param dstId The ID of destination memory region. /// @param dstOffset The offset into the destination memory region. /// @param srcId The ID of source memory region. @@ -59,10 +59,10 @@ struct BasePortChannelDeviceHandle { /// @param size The size of the transfer. MSCCLPP_DEVICE_INLINE void put(MemoryId dstId, uint64_t dstOffset, MemoryId srcId, uint64_t srcOffset, uint64_t size) { - fifo_.push({TriggerData, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); + fifo_.push({TriggerPut, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); } - /// Push a TriggerData to the FIFO. + /// Push a TriggerPut to the FIFO. /// @param dstId The ID of destination memory region. /// @param srcId The ID of source memory region. /// @param offset The common offset into the destination and source memory regions. @@ -71,10 +71,10 @@ struct BasePortChannelDeviceHandle { put(dstId, offset, srcId, offset, size); } - /// Push a TriggerFlag to the FIFO. - MSCCLPP_DEVICE_INLINE void signal() { fifo_.push({TriggerFlag, 0, 0, 0, 0, 0, semaphoreId_}); } + /// Push a TriggerSignal to the FIFO. + MSCCLPP_DEVICE_INLINE void signal() { fifo_.push({TriggerSignal, 0, 0, 0, 0, 0, semaphoreId_}); } - /// Push a TriggerData and a TriggerFlag at the same time to the FIFO. + /// Push a TriggerPutWithSignal to the FIFO. /// @param dstId The ID of destination memory region. /// @param dstOffset The offset into the destination memory region. /// @param srcId The ID of source memory region. @@ -82,10 +82,10 @@ struct BasePortChannelDeviceHandle { /// @param size The size of the transfer. MSCCLPP_DEVICE_INLINE void putWithSignal(MemoryId dstId, uint64_t dstOffset, MemoryId srcId, uint64_t srcOffset, uint64_t size) { - fifo_.push({TriggerData | TriggerFlag, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); + fifo_.push({TriggerPutWithSignal, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); } - /// Push a TriggerData and a TriggerFlag at the same time to the FIFO. + /// Push a TriggerPutWithSignal to the FIFO. /// @param dstId The ID of destination memory region. /// @param srcId The ID of source memory region. /// @param offset The common offset into the destination and source memory regions. @@ -94,7 +94,7 @@ struct BasePortChannelDeviceHandle { putWithSignal(dstId, offset, srcId, offset, size); } - /// Push a TriggerData, a TriggerFlag, and a TriggerSync at the same time to the FIFO. + /// Push a TriggerPutWithSignalAndFlush to the FIFO. /// @param dstId The ID of destination memory region. /// @param dstOffset The offset into the destination memory region. /// @param srcId The ID of source memory region. @@ -103,12 +103,11 @@ struct BasePortChannelDeviceHandle { /// @param maxSpinCount The maximum number of spin counts before asserting. Never assert if negative. MSCCLPP_DEVICE_INLINE void putWithSignalAndFlush(MemoryId dstId, uint64_t dstOffset, MemoryId srcId, uint64_t srcOffset, uint64_t size, int64_t maxSpinCount = 1000000) { - uint64_t pos = - fifo_.push({TriggerData | TriggerFlag | TriggerSync, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); + uint64_t pos = fifo_.push({TriggerPutWithSignalAndFlush, dstId, dstOffset, srcId, srcOffset, size, semaphoreId_}); detail::waitFlush(flushDonePos_, pos, maxSpinCount); } - /// Push a TriggerData, a TriggerFlag, and a TriggerSync at the same time to the FIFO. + /// Push a TriggerPutWithSignalAndFlush to the FIFO. /// @param dstId The ID of destination memory region. /// @param srcId The ID of source memory region. /// @param offset The common offset into the destination and source memory regions. @@ -119,10 +118,10 @@ struct BasePortChannelDeviceHandle { putWithSignalAndFlush(dstId, offset, srcId, offset, size, maxSpinCount); } - /// Push a TriggerSync to the FIFO. + /// Push a TriggerFlush to the FIFO. /// @param maxSpinCount The maximum number of spin counts before asserting. Never assert if negative. MSCCLPP_DEVICE_INLINE void flush(int64_t maxSpinCount = 1000000) { - uint64_t pos = fifo_.push({TriggerSync, 0, 0, 0, 0, 0, semaphoreId_}); + uint64_t pos = fifo_.push({TriggerFlush, 0, 0, 0, 0, 0, semaphoreId_}); detail::waitFlush(flushDonePos_, pos, maxSpinCount); } @@ -149,7 +148,7 @@ struct PortChannelDeviceHandle : public BasePortChannelDeviceHandle { : BasePortChannelDeviceHandle(semaphoreId, semaphore, fifo, flushDonePos), dst_(dst), src_(src) {} #if defined(MSCCLPP_DEVICE_COMPILE) - /// Push a TriggerData to the FIFO. + /// Push a TriggerPut to the FIFO. /// @param dstOffset The offset into the destination memory region. /// @param srcOffset The offset into the source memory region. /// @param size The size of the transfer. @@ -157,12 +156,12 @@ struct PortChannelDeviceHandle : public BasePortChannelDeviceHandle { BasePortChannelDeviceHandle::put(dst_, dstOffset, src_, srcOffset, size); } - /// Push a TriggerData to the FIFO. + /// Push a TriggerPut to the FIFO. /// @param offset The common offset into the destination and source memory regions. /// @param size The size of the transfer. MSCCLPP_DEVICE_INLINE void put(uint64_t offset, uint64_t size) { put(offset, offset, size); } - /// Push a TriggerData and a TriggerFlag at the same time to the FIFO. + /// Push a TriggerPutWithSignal to the FIFO. /// @param dstOffset The offset into the destination memory region. /// @param srcOffset The offset into the source memory region. /// @param size The size of the transfer. @@ -170,12 +169,12 @@ struct PortChannelDeviceHandle : public BasePortChannelDeviceHandle { BasePortChannelDeviceHandle::putWithSignal(dst_, dstOffset, src_, srcOffset, size); } - /// Push a TriggerData and a TriggerFlag at the same time to the FIFO. + /// Push a TriggerPutWithSignal to the FIFO. /// @param offset The common offset into the destination and source memory regions. /// @param size The size of the transfer. MSCCLPP_DEVICE_INLINE void putWithSignal(uint64_t offset, uint64_t size) { putWithSignal(offset, offset, size); } - /// Push a TriggerData, a TriggerFlag, and a TriggerSync at the same time to the FIFO. + /// Push a TriggerPutWithSignalAndFlush to the FIFO. /// @param dstOffset The offset into the destination memory region. /// @param srcOffset The offset into the source memory region. /// @param size The size of the transfer. @@ -185,7 +184,7 @@ struct PortChannelDeviceHandle : public BasePortChannelDeviceHandle { BasePortChannelDeviceHandle::putWithSignalAndFlush(dst_, dstOffset, src_, srcOffset, size, maxSpinCount); } - /// Push a TriggerData, a TriggerFlag, and a TriggerSync at the same time to the FIFO. + /// Push a TriggerPutWithSignalAndFlush to the FIFO. /// @param offset The common offset into the destination and source memory regions. /// @param size The size of the transfer. MSCCLPP_DEVICE_INLINE void putWithSignalAndFlush(uint64_t offset, uint64_t size) { diff --git a/src/core/include/connection.hpp b/src/core/include/connection.hpp index eda4b3ef1..bc18b3205 100644 --- a/src/core/include/connection.hpp +++ b/src/core/include/connection.hpp @@ -94,7 +94,7 @@ class BaseConnection { int maxWriteQueueSize_; // GPU-visible flush-done position (host-pinned memory). ProxyService writes one past the - // highest FIFO position whose TriggerSync request has fully completed on this connection + // highest FIFO position whose TriggerFlush request has fully completed on this connection // (CQ drained for IB, synchronous flush() returned for non-IB). std::shared_ptr gpuFlushDonePos_; }; diff --git a/src/core/port_channel.cc b/src/core/port_channel.cc index 0601ef84b..40de48005 100644 --- a/src/core/port_channel.cc +++ b/src/core/port_channel.cc @@ -90,7 +90,7 @@ MSCCLPP_API_CPP void ProxyService::startProxy(bool blocking) { proxy_->start(blo MSCCLPP_API_CPP void ProxyService::stopProxy() { proxy_->stop(); - // Drain pending TriggerSync flushes. After a bounded loop, force-unblock any still-pending + // Drain pending TriggerFlush operations. 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(); @@ -126,22 +126,46 @@ ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger trigger) { int maxWriteQueueSize = conn.getMaxWriteQueueSize(); auto& numRequests = inflightRequests_[conn.impl_]; - if (trigger.fields.type & TriggerData) { + auto put = [&]() { RegisteredMemory& dst = memories_[trigger.fields.dstMemoryId]; RegisteredMemory& src = memories_[trigger.fields.srcMemoryId]; conn.write(dst, trigger.fields.dstOffset, src, trigger.fields.srcOffset, trigger.fields.size); numRequests++; - } - - if (trigger.fields.type & TriggerFlag) { + }; + auto signal = [&]() { semaphore->signal(); numRequests++; + }; + + bool flushRequested = false; + switch (trigger.fields.type) { + case TriggerPut: + put(); + break; + case TriggerSignal: + signal(); + break; + case TriggerFlush: + flushRequested = true; + break; + case TriggerPutWithSignal: + put(); + signal(); + break; + case TriggerPutWithSignalAndFlush: + put(); + signal(); + flushRequested = true; + break; + default: + WARN(CONN, "unknown trigger opcode ", uint64_t(trigger.fields.type), ", ignoring the trigger"); + return ProxyHandlerResult::Continue; } - if (trigger.fields.type & TriggerSync) { - // Record this TriggerSync's FIFO position. The GPU caller is spinning on - // flushDonePos_ > pos; progressFlushes() will publish pos+1 once the CQ drains. - // Later TriggerSyncs on the same conn overwrite — CQ drain completes them all at once. + if (flushRequested) { + // Record this flush's FIFO position. The GPU caller is spinning on flushDonePos_ > pos; + // progressFlushes() publishes pos+1 once the CQ drains. A later flush on the same connection + // overwrites this entry, and the CQ drain completes them all at once. conn.impl_->requestFlush(); pendingFlushPos_[conn.impl_] = pos; numRequests = 0; diff --git a/test/mp_unit/port_channel_tests.cu b/test/mp_unit/port_channel_tests.cu index eec1760cf..265729ce1 100644 --- a/test/mp_unit/port_channel_tests.cu +++ b/test/mp_unit/port_channel_tests.cu @@ -897,7 +897,7 @@ PERF_TEST(PortChannelOneToOneTest, MultiQpFlushStressIbHostNoAtomicMode) { // Same-channel concurrent-flush kernel: N GPU threads on the same PortChannel each call // putWithSignalAndFlush in lockstep. Stresses the FIFO-position-based wait target so that -// each caller waits on its own TriggerSync rather than on a globally-incrementing counter +// each caller waits on its own TriggerFlush rather than on a globally-incrementing counter // that could be assigned out-of-order relative to the FIFO push order. __constant__ DeviceHandle gSingleChanForConcurrentFlush; @@ -936,7 +936,7 @@ void PortChannelOneToOneTest::testSameChanConcurrentFlush(IbMode ibMode) { communicator->bootstrap()->barrier(); // Measure: a successful completion (no deadlock, no CQ error) validates that each - // concurrent-flush caller waited on its own TriggerSync (not someone else's earlier one). + // concurrent-flush caller waited on its own TriggerFlush (not someone else's earlier one). const int nIters = 500; mscclpp::Timer timer; kernelSameChanConcurrentFlush<<<1, nThreads>>>(nIters); From ccab6d7b3802373e4ddb9b3761b3b2e8aaace36b Mon Sep 17 00:00:00 2001 From: Changho Hwang Date: Thu, 6 Aug 2026 08:22:07 +0000 Subject: [PATCH 2/2] Add `accumulate` method for `PortChannel` Add `Connection::accumulate()` and `PortChannel::accumulate()`, which add a 64-bit signed value to remote memory. Unlike `updateAndSync()`, the caller needs to know only its own contribution, not the destination's current value. Carried by the `TriggerAccumulate` opcode, with the operand in the trigger's first word. FIFO readiness now comes from the lap-parity commit bit, so a zero operand traverses the proxy like any other trigger; transport tests interleave zero and nonzero accumulates to cover that path. The addition has to be a real read-modify-write at the destination, which not every transport can do: - IB: any number of writers, via RDMA fetch-and-add. Rejected in no-atomic mode, where the device has no RDMA atomics. - Ethernet: any number of remote writers. The receiving process performs the update, and its connections are serialized against each other; without that a 7-writer fan-in loses ~74% of updates. - CudaIpc on ROCm: any number of writers. The proxy runs a kernel on the connection's stream, which also orders it against a following signal. A kernel there runs even while the caller's kernel spins. - CudaIpc on CUDA: not supported, throws InvalidUsage. The host cannot read-modify-write device memory, and a proxy-launched kernel either cannot be scheduled while the caller's kernel spins (same context) or costs 2391 us per operation (separate context), against 19 us for a plain remote store. Tested on H200 (CUDA 12.9) and MI300X (ROCm 7.2), including an 8-rank fan-in test with 7 concurrent writers into one address. --- include/mscclpp/core.hpp | 25 ++ include/mscclpp/fifo_device.hpp | 7 +- include/mscclpp/port_channel_device.hpp | 25 ++ src/core/accumulate_kernel.cu | 33 ++ src/core/connection.cc | 127 ++++++- src/core/include/connection.hpp | 5 + src/core/include/context.hpp | 6 + src/core/port_channel.cc | 9 + test/mp_unit/mp_unit_tests.hpp | 19 + test/mp_unit/port_channel_tests.cu | 449 ++++++++++++++++++++++++ 10 files changed, 691 insertions(+), 14 deletions(-) create mode 100644 src/core/accumulate_kernel.cu diff --git a/include/mscclpp/core.hpp b/include/mscclpp/core.hpp index a0c9f7494..b990e210e 100644 --- a/include/mscclpp/core.hpp +++ b/include/mscclpp/core.hpp @@ -649,6 +649,31 @@ class Connection { /// @param newValue The new value to write. void updateAndSync(RegisteredMemory dst, uint64_t dstOffset, uint64_t* src, uint64_t newValue); + /// Add a value to a 64-bit integer in a destination RegisteredMemory. + /// + /// The caller supplies only its own contribution, unlike updateAndSync(), which needs the + /// destination's current value. Addition commutes, so arrival order does not matter. + /// + /// The addition must be a real read-modify-write at the destination, so how many concurrent + /// writers one address allows depends on the transport: + /// + /// - IB: any number of writers, via NIC atomic fetch-and-add. Throws in no-atomic mode, where + /// the device has no RDMA atomics. + /// - Ethernet: any number of remote writers. The receiving process does the update and + /// serializes its connections. The destination GPU must not write the address concurrently; + /// such a write is lost inside the read-modify-write window. + /// - CudaIpc on ROCm: any number of writers. The proxy runs a kernel, which a caller kernel + /// does not block. + /// - CudaIpc on CUDA: throws. The host cannot read-modify-write device memory, and a + /// proxy-launched kernel cannot run while the caller's kernel waits. Use a device-side atomic + /// on peer memory reached through a MemoryChannel. + /// + /// @param dst The destination RegisteredMemory. + /// @param dstOffset The offset in bytes from the start of the destination RegisteredMemory. + /// @param value The 64-bit signed value to add. + /// @throws Error with ErrorCode::InvalidUsage if the transport cannot accumulate. + void accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value); + /// Flush any pending writes to the remote process. /// @param timeoutUsec Timeout in microseconds. Default: -1 (no timeout) void flush(int64_t timeoutUsec = -1); diff --git a/include/mscclpp/fifo_device.hpp b/include/mscclpp/fifo_device.hpp index 192672c75..d6304138d 100644 --- a/include/mscclpp/fifo_device.hpp +++ b/include/mscclpp/fifo_device.hpp @@ -22,13 +22,14 @@ namespace mscclpp { /// combination nothing emits cannot be expressed, and a trigger whose type field is unset is not /// a valid operation. using TriggerType = uint64_t; -constexpr TriggerType TriggerNone = 0; // Not an operation. Unwritten or malformed. +constexpr TriggerType TriggerNone = 0; // Not an operation; invalid for ProxyService. constexpr TriggerType TriggerPut = 1; // Transfer data. constexpr TriggerType TriggerSignal = 2; // Signal the remote semaphore. constexpr TriggerType TriggerFlush = 3; // Flush the connection. constexpr TriggerType TriggerPutWithSignal = 4; // Transfer data, then signal. constexpr TriggerType TriggerPutWithSignalAndFlush = 5; // Transfer data, signal, then flush. -// 6 and 7 are unassigned. +constexpr TriggerType TriggerAccumulate = 6; // Add a value to remote memory. +// 7 is unassigned. constexpr unsigned int TriggerBitsSize = 32; constexpr unsigned int TriggerBitsOffset = 32; @@ -39,7 +40,7 @@ constexpr unsigned int TriggerBitsSemaphoreId = 10; // there. See FifoDeviceHandle::push(). constexpr unsigned int TriggerBitsFifoReserved = 1; -static_assert(TriggerPutWithSignalAndFlush < (1ULL << TriggerBitsType), "trigger opcodes must fit in the type field"); +static_assert(TriggerAccumulate < (1ULL << TriggerBitsType), "trigger opcodes must fit in the type field"); /// Pair of 64-bit unsigned integers used as a trigger for the proxy. /// Used as a work element in the concurrent FIFO. diff --git a/include/mscclpp/port_channel_device.hpp b/include/mscclpp/port_channel_device.hpp index 9cf9c8e8c..a6ce3632f 100644 --- a/include/mscclpp/port_channel_device.hpp +++ b/include/mscclpp/port_channel_device.hpp @@ -125,6 +125,24 @@ struct BasePortChannelDeviceHandle { detail::waitFlush(flushDonePos_, pos, maxSpinCount); } + /// Push an accumulate trigger to the FIFO: add a 64-bit value to remote memory. + /// Connection::accumulate() documents how many concurrent writers each transport allows. + /// @param dstId The ID of destination memory region. + /// @param dstOffset The offset into the destination memory region. + /// @param value The 64-bit signed value to add. + MSCCLPP_DEVICE_INLINE void accumulate(MemoryId dstId, uint64_t dstOffset, int64_t value) { + ProxyTrigger trigger; + // The operand occupies fst, spanning the size and srcOffset fields. + trigger.fst = static_cast(value); + // snd carries dstOffset, dstMemoryId, the opcode, and semaphoreId. + trigger.snd = 0; + trigger.fields.dstOffset = dstOffset; + trigger.fields.dstMemoryId = dstId; + trigger.fields.type = TriggerAccumulate; + trigger.fields.semaphoreId = semaphoreId_; + fifo_.push(trigger); + } + /// Check if the port channel has been signaled. /// @return true if the port channel has been signaled. MSCCLPP_DEVICE_INLINE bool poll() { return semaphore_.poll(); } @@ -190,6 +208,13 @@ struct PortChannelDeviceHandle : public BasePortChannelDeviceHandle { MSCCLPP_DEVICE_INLINE void putWithSignalAndFlush(uint64_t offset, uint64_t size) { putWithSignalAndFlush(offset, offset, size); } + /// Push an accumulate trigger to the FIFO: add a 64-bit value to the destination memory. + /// See Connection::accumulate() for transport support. + /// @param dstOffset The offset into the destination memory region. + /// @param value The 64-bit signed value to add. + MSCCLPP_DEVICE_INLINE void accumulate(uint64_t dstOffset, int64_t value) { + BasePortChannelDeviceHandle::accumulate(dst_, dstOffset, value); + } #endif // defined(MSCCLPP_DEVICE_COMPILE) }; diff --git a/src/core/accumulate_kernel.cu b/src/core/accumulate_kernel.cu new file mode 100644 index 000000000..e2730f264 --- /dev/null +++ b/src/core/accumulate_kernel.cu @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#include + +#if defined(MSCCLPP_USE_ROCM) + +#include +#include + +#include "context.hpp" + +namespace mscclpp { + +// System-scope atomic add on a signed 64-bit value. +__global__ void accumulateI64Kernel(int64_t* dst, int64_t value) { + (void)atomicFetchAdd(dst, value, memoryOrderRelaxed); +} + +void CudaIpcStream::accumulate(int64_t* dst, int64_t value) { + CudaDeviceGuard deviceGuard(deviceId_); + setStreamIfNeeded(); + // Submit to this connection's stream, which orders the add ahead of any signal or flush that + // follows. On ROCm a kernel runs while the caller's kernel occupies the GPU, so the proxy does + // not wait for the caller. + accumulateI64Kernel<<<1, 1, 0, *stream_>>>(dst, value); + MSCCLPP_CUDATHROW(cudaGetLastError()); + dirty_ = true; +} + +} // namespace mscclpp + +#endif // defined(MSCCLPP_USE_ROCM) diff --git a/src/core/connection.cc b/src/core/connection.cc index e01c4a6ea..32b4ff4f5 100644 --- a/src/core/connection.cc +++ b/src/core/connection.cc @@ -9,6 +9,7 @@ #include #include +#include #include #include @@ -68,6 +69,10 @@ MSCCLPP_API_CPP void Connection::updateAndSync(RegisteredMemory dst, uint64_t ds impl_->updateAndSync(dst, dstOffset, src, newValue); } +MSCCLPP_API_CPP void Connection::accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) { + impl_->accumulate(dst, dstOffset, value); +} + MSCCLPP_API_CPP void Connection::flush(int64_t timeoutUsec) { impl_->flush(timeoutUsec); } MSCCLPP_API_CPP Transport Connection::transport() const { return impl_->transport(); } @@ -194,6 +199,31 @@ void CudaIpcConnection::flush(int64_t timeoutUsec) { #endif } +void CudaIpcConnection::accumulate([[maybe_unused]] RegisteredMemory dst, [[maybe_unused]] uint64_t dstOffset, + [[maybe_unused]] int64_t value) { +#if defined(MSCCLPP_USE_ROCM) + validateTransport(dst, remoteTransport()); + // A kernel on this connection's stream performs the addition, a real read-modify-write, so + // writers in any number of processes may target one address. The host cannot do it instead: + // GPU memory is host-accessible only to the process that allocated it, and the proxy holds an + // IPC-imported mapping, which is device-only. + int64_t* dstPtr = reinterpret_cast(reinterpret_cast(dst.data()) + dstOffset); + stream_->accumulate(dstPtr, value); + INFO(CONN, "CudaIpcConnection accumulate: dst ", dstPtr, ", value ", value); +#else + // The host reaches device memory only through the copy engines, which move a value but cannot + // add to one, and a host-side read-modify-write is not atomic. A kernel is atomic but unusable: + // in the caller's context it does not start until the caller's kernel finishes, deadlocking any + // caller that spins on the result; in a separate context it costs 2391 us per operation against + // 19 us for a plain remote store. ROCm has neither limit. + THROW(CONN, Error, ErrorCode::InvalidUsage, + "accumulate is not supported over CudaIpc on CUDA: the host cannot atomically " + "read-modify-write device memory, and a proxy-launched kernel cannot run while the " + "caller's kernel waits. Use a device-side atomic on peer memory reached through a " + "MemoryChannel instead"); +#endif // defined(MSCCLPP_USE_ROCM) +} + // IBConnection void IBConnection::recvThreadFunc() { @@ -492,6 +522,24 @@ void IBConnection::flush(int64_t timeoutUsec) { #endif } +void IBConnection::accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) { + validateTransport(dst, remoteTransport()); + auto dstTransportInfo = getImpl(dst).getTransportInfo(remoteTransport()); + if (dstTransportInfo.ibLocal) { + THROW(CONN, Error, ErrorCode::InvalidUsage, "dst is local, which is not supported"); + } + auto dstMrInfo = dstTransportInfo.ibMrInfo; + + if (ibNoAtomic_) { + THROW(CONN, Error, ErrorCode::InvalidUsage, "accumulate is not supported in IB no-atomic mode"); + } + + qp_.lock()->stageSendAtomicAdd(atomicSrcTransportInfo_.ibMr, dstMrInfo, /*wrId=*/0, dstOffset, + static_cast(value), /*signaled=*/true); + qp_.lock()->postSend(); + INFO(CONN, "IBConnection accumulate: dst ", (uint8_t*)dstMrInfo.addr + dstOffset, ", value ", value); +} + void IBConnection::requestFlush() { // No-op: IB sends were already posted by prior conn.write() calls in handleTrigger. // progressFlush() drives completion by polling the send CQ. @@ -666,13 +714,48 @@ void EthernetConnection::flush(int64_t) { #endif } +// Serializes the receive-side read-modify-write of accumulate() across this process's Ethernet +// connections. Ethernet costs ~136 us per operation, so the contention is irrelevant. +static std::mutex& accumulateMutex() { + static std::mutex mtx; + return mtx; +} + +void EthernetConnection::accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) { + validateTransport(dst, remoteTransport()); + + // Wire format matches write(): [dstPtr(8B)] [size(8B)] [data(size B)]. The MSB of size marks + // the message as an accumulate. + uint64_t* dstPtr = reinterpret_cast(reinterpret_cast(dst.originalDataPtr()) + dstOffset); + constexpr uint64_t accumulateFlag = uint64_t{1} << uint64_t{63}; + uint64_t dataSize = sizeof(uint64_t) | accumulateFlag; + uint64_t messageSize = 0; + + char* dstPtrBytes = reinterpret_cast(&dstPtr); + std::copy(dstPtrBytes, dstPtrBytes + sizeof(dstPtr), sendBuffer_.data() + messageSize); + messageSize += sizeof(dstPtr); + + char* sizeBytes = reinterpret_cast(&dataSize); + std::copy(sizeBytes, sizeBytes + sizeof(dataSize), sendBuffer_.data() + messageSize); + messageSize += sizeof(dataSize); + + char* valueBytes = reinterpret_cast(&value); + std::copy(valueBytes, valueBytes + sizeof(value), sendBuffer_.data() + messageSize); + messageSize += sizeof(value); + + sendSocket_->send(sendBuffer_.data(), messageSize); + + INFO(CONN, "EthernetConnection accumulate: dst ", dstPtr, ", value ", value); +} + void EthernetConnection::recvMessages() { - // Declarating Variables + // Declaring Variables char* ptr; uint64_t size; uint64_t recvSize; int closed = 0; bool received = true; + constexpr uint64_t accumulateFlag = uint64_t{1} << uint64_t{63}; // Receiving Messages Until Connection is Closed while (recvSocket_->getState() != SocketStateClosed) { @@ -684,10 +767,15 @@ void EthernetConnection::recvMessages() { if (closed == 0) recvSocket_->recvUntilEnd(&ptr, sizeof(char*), &closed); received &= !closed; - // Receiving data size + // Receiving data size (MSB may indicate accumulate) if (closed == 0) recvSocket_->recvUntilEnd(&size, sizeof(uint64_t), &closed); received &= !closed; + bool isAccumulate = (size & accumulateFlag) != 0; + if (isAccumulate) { + size &= ~accumulateFlag; // Strip the flag to get the data size. + } + #if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_CONN_ETH_RECV_META_EXIT) NpKit::CollectCpuEvent(NPKIT_EVENT_CONN_ETH_RECV_META_EXIT, uint32_t(size), 0, *NpKit::GetCpuTimestamp(), 1); #endif @@ -696,16 +784,33 @@ void EthernetConnection::recvMessages() { NpKit::CollectCpuEvent(NPKIT_EVENT_CONN_ETH_RECV_DATA_ENTRY, uint32_t(size), 0, *NpKit::GetCpuTimestamp(), 1); #endif - // Receiving Data and Copying Data yo GPU - recvSize = 0; - while (recvSize < size && closed == 0) { - uint64_t messageSize = std::min(recvBufferSize_, (size - recvSize) / sizeof(char)) * sizeof(char); - recvSocket_->recvUntilEnd(recvBuffer_.data(), messageSize, &closed); + if (isAccumulate && received && size == sizeof(int64_t)) { + // Accumulate: receive the operand, then read, add, and write back. + int64_t addValue; + recvSocket_->recvUntilEnd(&addValue, sizeof(int64_t), &closed); received &= !closed; - - if (received) - mscclpp::gpuMemcpy(ptr + (recvSize / sizeof(char)), recvBuffer_.data(), messageSize, cudaMemcpyHostToDevice); - recvSize += messageSize; + if (received) { + // Every peer terminates its socket in this process, so several recv threads can be here + // at once for one address. The read-modify-write below is not atomic, so serialize it + // against the other recv threads. + const std::lock_guard lock(accumulateMutex()); + int64_t current; + mscclpp::gpuMemcpy(reinterpret_cast(¤t), ptr, sizeof(int64_t), cudaMemcpyDeviceToHost); + current += addValue; + mscclpp::gpuMemcpy(ptr, reinterpret_cast(¤t), sizeof(int64_t), cudaMemcpyHostToDevice); + } + } else { + // Regular write: receive data and copy to GPU + recvSize = 0; + while (recvSize < size && closed == 0) { + uint64_t messageSize = std::min(recvBufferSize_, (size - recvSize) / sizeof(char)) * sizeof(char); + recvSocket_->recvUntilEnd(recvBuffer_.data(), messageSize, &closed); + received &= !closed; + + if (received) + mscclpp::gpuMemcpy(ptr + (recvSize / sizeof(char)), recvBuffer_.data(), messageSize, cudaMemcpyHostToDevice); + recvSize += messageSize; + } } #if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_CONN_ETH_RECV_DATA_EXIT) diff --git a/src/core/include/connection.hpp b/src/core/include/connection.hpp index bc18b3205..1d94a06cd 100644 --- a/src/core/include/connection.hpp +++ b/src/core/include/connection.hpp @@ -35,6 +35,8 @@ class BaseConnection { virtual void updateAndSync(RegisteredMemory dst, uint64_t dstOffset, uint64_t* src, uint64_t newValue) = 0; + virtual void accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) = 0; + virtual void flush(int64_t timeoutUsec = -1) = 0; /// Start signal forwarding to the given memory address. @@ -113,6 +115,7 @@ class CudaIpcConnection : public BaseConnection { void write(RegisteredMemory dst, uint64_t dstOffset, RegisteredMemory src, uint64_t srcOffset, uint64_t size) override; void updateAndSync(RegisteredMemory dst, uint64_t dstOffset, uint64_t* src, uint64_t newValue) override; + void accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) override; void flush(int64_t timeoutUsec) override; }; @@ -169,6 +172,7 @@ class IBConnection : public BaseConnection { void write(RegisteredMemory dst, uint64_t dstOffset, RegisteredMemory src, uint64_t srcOffset, uint64_t size) override; void updateAndSync(RegisteredMemory dst, uint64_t dstOffset, uint64_t* src, uint64_t newValue) override; + void accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) override; void flush(int64_t timeoutUsec) override; @@ -203,6 +207,7 @@ class EthernetConnection : public BaseConnection { void write(RegisteredMemory dst, uint64_t dstOffset, RegisteredMemory src, uint64_t srcOffset, uint64_t size) override; void updateAndSync(RegisteredMemory dst, uint64_t dstOffset, uint64_t* src, uint64_t newValue) override; + void accumulate(RegisteredMemory dst, uint64_t dstOffset, int64_t value) override; void flush(int64_t timeoutUsec) override; }; diff --git a/src/core/include/context.hpp b/src/core/include/context.hpp index 42d03db15..54a6db64c 100644 --- a/src/core/include/context.hpp +++ b/src/core/include/context.hpp @@ -28,6 +28,12 @@ class CudaIpcStream { void memcpyH2D(void* dst, const void* src, size_t nbytes); +#if defined(MSCCLPP_USE_ROCM) + /// Add a value to a 64-bit integer in peer memory, with a kernel on this stream. ROCm only: + /// on CUDA such a kernel cannot be scheduled while the caller's kernel spins. + void accumulate(int64_t* dst, int64_t value); +#endif // defined(MSCCLPP_USE_ROCM) + void sync(); operator cudaStream_t() const { return *stream_; } diff --git a/src/core/port_channel.cc b/src/core/port_channel.cc index 40de48005..6cb0dacdc 100644 --- a/src/core/port_channel.cc +++ b/src/core/port_channel.cc @@ -136,6 +136,12 @@ ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger trigger) { semaphore->signal(); numRequests++; }; + auto accumulate = [&]() { + RegisteredMemory& dst = memories_[trigger.fields.dstMemoryId]; + // The operand is the full fst word, spanning the size and srcOffset fields. + conn.accumulate(dst, trigger.fields.dstOffset, static_cast(trigger.fst)); + numRequests++; + }; bool flushRequested = false; switch (trigger.fields.type) { @@ -157,6 +163,9 @@ ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger trigger) { signal(); flushRequested = true; break; + case TriggerAccumulate: + accumulate(); + break; default: WARN(CONN, "unknown trigger opcode ", uint64_t(trigger.fields.type), ", ignoring the trigger"); return ProxyHandlerResult::Continue; diff --git a/test/mp_unit/mp_unit_tests.hpp b/test/mp_unit/mp_unit_tests.hpp index 8654ccc91..0b1e18d92 100644 --- a/test/mp_unit/mp_unit_tests.hpp +++ b/test/mp_unit/mp_unit_tests.hpp @@ -141,6 +141,22 @@ using DeviceHandle = mscclpp::DeviceHandle; using IbMode = mscclpp::EndpointConfig::Ib::Mode; +// Fan-in: every rank other than 0 accumulates into rank 0's counter, so the destination has many +// concurrent writers. An unserialized read-modify-write loses updates here, which a one-to-one +// test cannot show. +// +// Covers every transport that allows concurrent writers: IB, Ethernet, and CudaIpc on ROCm. See +// Connection::accumulate(). +class PortChannelFanInTest : public CommunicatorTestBase { + protected: + void SetUp() override; + void TearDown() override; + + void testFanIn(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode = IbMode::Default); + + std::shared_ptr proxyService; +}; + class PortChannelOneToOneTest : public CommunicatorTestBase { protected: struct PingPongTestParams { @@ -161,6 +177,9 @@ class PortChannelOneToOneTest : public CommunicatorTestBase { void testPingPongPerf(PingPongTestParams params); void testPacketPingPong(bool useIbOnly, IbMode ibMode = IbMode::Default); void testPacketPingPongPerf(bool useIbOnly, IbMode ibMode = IbMode::Default); + void testAccumulate(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode = IbMode::Default); + void testAccumulateSigned(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode = IbMode::Default); + void testAccumulateZero(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode = IbMode::Default); void testBandwidth(PingPongTestParams params); void setupMultiQpChannels(int numQps, size_t elemsPerChan, IbMode ibMode, int tagBase, std::vector>& sendBuffs, diff --git a/test/mp_unit/port_channel_tests.cu b/test/mp_unit/port_channel_tests.cu index 265729ce1..fd5326934 100644 --- a/test/mp_unit/port_channel_tests.cu +++ b/test/mp_unit/port_channel_tests.cu @@ -1,8 +1,10 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +#include #include #include +#include #include "gdr.hpp" #include "mp_unit_tests.hpp" @@ -626,6 +628,315 @@ PERF_TEST(PortChannelOneToOneTest, BandwidthIbHostNoAtomicMode) { .useIPC = false, .useIB = true, .useEthernet = false, .waitWithPoll = false, .ibMode = IbMode::HostNoAtomic}); } +// Larger than 32 bits, so a truncated operand shows up as a wrong sum. The operand spans the +// size and srcOffset fields of ProxyTrigger::fst. +static constexpr int64_t kAccumulateValue = (int64_t{1} << 32) + 1; + +// Every block accumulates kAccumulateValue into the remote buffer, then block 0 signals, flushes, +// and waits. Both ranks do this at once, so each iteration carries numBlocks accumulates in each +// direction. +// +// After wait(), the local buffer must hold every add the remote issued before its signal. That is +// the ordering check: the proxy must finish an accumulate before the signal that follows it on the +// same connection. A fire-and-forget accumulate fails it. +// +// The check is a range, not an equality. The remote resumes on our signal, which we send before +// waiting, so it may already have issued one more iteration of adds — and no more than one, +// because its next wait() needs a signal we have not sent. +__global__ void kernelPortChannelAccumulateConcurrent(int64_t* localBuff, int nTries, mscclpp::DeviceSyncer* syncer, + int* ret) { + DeviceHandle& portChan = gChannelOneToOneTestConstPortChans; + const int numBlocks = gridDim.x; + + for (int iter = 0; iter < nTries; iter++) { + // Step 1: every block accumulates into the remote buffer. + portChan.accumulate(0, kAccumulateValue); + + // Step 2: grid barrier, so every block has pushed its accumulate. + syncer->sync(numBlocks); + + // Step 3: block 0 signals that this rank's adds are done, then waits for the remote. + if (blockIdx.x == 0) { + portChan.signal(); + portChan.flush(); + portChan.wait(); + + // Step 4: the remote's adds for this iteration have all landed. + const int64_t perIter = (int64_t)numBlocks * kAccumulateValue; + int64_t lowerBound = (int64_t)(iter + 1) * perIter; + int64_t observed = *(volatile int64_t*)localBuff; + if (observed < lowerBound || observed > lowerBound + perIter) { + printf("iter %d: buff = %lld, expected %lld..%lld\n", iter, (long long)observed, (long long)lowerBound, + (long long)(lowerBound + perIter)); + *ret = 1; + } + } + + // Step 5: grid barrier, so signal and wait finish before the next iteration. + syncer->sync(numBlocks); + } +} + +// Negative operands. Rank 0 adds +kAccumulateValue nTries times and rank 1 adds -kAccumulateValue +// nTries times, each into the peer's buffer, which starts at 0. Each rank ends holding the peer's +// signed sum. +__global__ void kernelPortChannelAccumulateSigned(int64_t* localBuff, int nTries, int rank, int* ret) { + DeviceHandle& portChan = gChannelOneToOneTestConstPortChans; + if (threadIdx.x != 0 || blockIdx.x != 0) return; + + const int64_t value = (rank == 0) ? kAccumulateValue : -kAccumulateValue; + for (int iter = 0; iter < nTries; iter++) { + portChan.accumulate(0, value); + } + portChan.signal(); + portChan.flush(); + portChan.wait(); + + int64_t expected = (int64_t)nTries * ((rank == 0) ? -kAccumulateValue : kAccumulateValue); + int64_t observed = *(volatile int64_t*)localBuff; + if (observed != expected) { + printf("buff = %lld, expected = %lld\n", (long long)observed, (long long)expected); + *ret = 1; + } +} + +void PortChannelOneToOneTest::testAccumulate(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode) { + if (gEnv->rank >= numRanksToUse) return; + + const int nElem = 1; + const int numBlocks = 64; + const int nTries = 50; + + std::vector portChannels; + auto buff = mscclpp::GpuBuffer(nElem); + MSCCLPP_CUDATHROW(cudaMemset(buff.memory().get(), 0, nElem * sizeof(int64_t))); + + setupMeshConnections(portChannels, useIPC, useIb, useEthernet, buff.memory().get(), nElem * sizeof(int64_t), nullptr, + 0, ibMode); + + ASSERT_EQ(portChannels.size(), 1); + + std::vector> portChannelHandles; + for (auto& ch : portChannels) portChannelHandles.push_back(ch.deviceHandle()); + + MSCCLPP_CUDATHROW(cudaMemcpyToSymbol(gChannelOneToOneTestConstPortChans, portChannelHandles.data(), + sizeof(DeviceHandle))); + + // Grid barrier, zero-initialized in device memory. + auto syncer = mscclpp::detail::gpuCallocShared(); + + proxyService->startProxy(); + + auto ret = mscclpp::detail::gpuCallocHostShared(); + *ret = 0; + + kernelPortChannelAccumulateConcurrent<<>>(buff.memory().get(), nTries, syncer.get(), ret.get()); + MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); + + EXPECT_EQ(*ret, 0); + + proxyService->stopProxy(); +} + +// Zero and non-zero operands interleave. This verifies that a zero-valued trigger traverses the +// FIFO, where readiness is independent of payload, without changing the accumulated total. +__global__ void kernelPortChannelAccumulateZero(int64_t* localBuff, int nTries, int* ret) { + DeviceHandle& portChan = gChannelOneToOneTestConstPortChans; + if (threadIdx.x != 0 || blockIdx.x != 0) return; + + for (int iter = 0; iter < nTries; iter++) { + portChan.accumulate(0, 0); + portChan.accumulate(0, kAccumulateValue); + portChan.accumulate(0, 0); + } + portChan.signal(); + portChan.flush(); + portChan.wait(); + + int64_t expected = (int64_t)nTries * kAccumulateValue; + int64_t observed = *(volatile int64_t*)localBuff; + if (observed != expected) { + printf("buff = %lld, expected = %lld\n", (long long)observed, (long long)expected); + *ret = 1; + } +} + +void PortChannelOneToOneTest::testAccumulateZero(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode) { + if (gEnv->rank >= numRanksToUse) return; + + const int nTries = 50; + + std::vector portChannels; + auto buff = mscclpp::GpuBuffer(1); + MSCCLPP_CUDATHROW(cudaMemset(buff.memory().get(), 0, sizeof(int64_t))); + + setupMeshConnections(portChannels, useIPC, useIb, useEthernet, buff.memory().get(), sizeof(int64_t), nullptr, 0, + ibMode); + ASSERT_EQ(portChannels.size(), 1); + + std::vector> portChannelHandles; + for (auto& ch : portChannels) portChannelHandles.push_back(ch.deviceHandle()); + MSCCLPP_CUDATHROW(cudaMemcpyToSymbol(gChannelOneToOneTestConstPortChans, portChannelHandles.data(), + sizeof(DeviceHandle))); + + proxyService->startProxy(); + + auto ret = mscclpp::detail::gpuCallocHostShared(); + *ret = 0; + + kernelPortChannelAccumulateZero<<<1, 1>>>(buff.memory().get(), nTries, ret.get()); + MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); + + EXPECT_EQ(*ret, 0); + + proxyService->stopProxy(); +} + +void PortChannelOneToOneTest::testAccumulateSigned(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode) { + if (gEnv->rank >= numRanksToUse) return; + + const int nTries = 100; + + std::vector portChannels; + auto buff = mscclpp::GpuBuffer(1); + MSCCLPP_CUDATHROW(cudaMemset(buff.memory().get(), 0, sizeof(int64_t))); + + setupMeshConnections(portChannels, useIPC, useIb, useEthernet, buff.memory().get(), sizeof(int64_t), nullptr, 0, + ibMode); + + ASSERT_EQ(portChannels.size(), 1); + + std::vector> portChannelHandles; + for (auto& ch : portChannels) portChannelHandles.push_back(ch.deviceHandle()); + + MSCCLPP_CUDATHROW(cudaMemcpyToSymbol(gChannelOneToOneTestConstPortChans, portChannelHandles.data(), + sizeof(DeviceHandle))); + + proxyService->startProxy(); + + auto ret = mscclpp::detail::gpuCallocHostShared(); + *ret = 0; + + kernelPortChannelAccumulateSigned<<<1, 1>>>(buff.memory().get(), nTries, gEnv->rank, ret.get()); + MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); + + EXPECT_EQ(*ret, 0); + + proxyService->stopProxy(); +} + +// CudaIpc accumulate needs an atomic read-modify-write of peer memory. On ROCm the proxy runs a +// kernel, which the caller's kernel does not block. On CUDA it can do neither, so the call throws. +#if defined(__HIP_PLATFORM_AMD__) + +TEST(PortChannelOneToOneTest, Accumulate) { + REQUIRE_CUDA_IPC_AVAILABLE; + testAccumulate(true, false, false); +} + +TEST(PortChannelOneToOneTest, AccumulateSigned) { + REQUIRE_CUDA_IPC_AVAILABLE; + testAccumulateSigned(true, false, false); +} + +TEST(PortChannelOneToOneTest, AccumulateZero) { + REQUIRE_CUDA_IPC_AVAILABLE; + testAccumulateZero(true, false, false); +} + +#else // !defined(__HIP_PLATFORM_AMD__) + +TEST(PortChannelOneToOneTest, AccumulateCudaIpcRejected) { + REQUIRE_CUDA_IPC_AVAILABLE; + if (gEnv->rank >= numRanksToUse) return; + + const int peer = 1 - gEnv->rank; + auto buff = mscclpp::GpuBuffer(1).memory(); + mscclpp::RegisteredMemory localMem; + mscclpp::RegisteredMemory remoteMem; + + mscclpp::EndpointConfig cfg; + cfg.transport = mscclpp::Transport::CudaIpc; + + auto connFuture = communicator->connect(cfg, peer); + localMem = communicator->registerMemory(buff.get(), sizeof(int64_t), mscclpp::Transport::CudaIpc); + communicator->sendMemory(localMem, peer, /*tag=*/78); + auto remoteFuture = communicator->recvMemory(peer, /*tag=*/78); + + auto conn = connFuture.get(); + remoteMem = remoteFuture.get(); + registeredMemories.push_back(localMem); + + try { + conn.accumulate(remoteMem, 0, 1); + FAIL() << "Expected accumulate over CudaIpc to throw InvalidUsage on CUDA"; + } catch (const mscclpp::Error& e) { + EXPECT_TRUE(e.getErrorCode() == mscclpp::ErrorCode::InvalidUsage); + } + + communicator->bootstrap()->barrier(); +} + +#endif // !defined(__HIP_PLATFORM_AMD__) + +TEST(PortChannelOneToOneTest, AccumulateIb) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testAccumulate(false, true, false, IbMode::Host); +} + +TEST(PortChannelOneToOneTest, AccumulateEthernet) { testAccumulate(false, false, true); } + +TEST(PortChannelOneToOneTest, AccumulateSignedIb) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testAccumulateSigned(false, true, false, IbMode::Host); +} + +TEST(PortChannelOneToOneTest, AccumulateSignedEthernet) { testAccumulateSigned(false, false, true); } + +TEST(PortChannelOneToOneTest, AccumulateZeroIb) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testAccumulateZero(false, true, false, IbMode::Host); +} + +TEST(PortChannelOneToOneTest, AccumulateZeroEthernet) { testAccumulateZero(false, false, true); } + +TEST(PortChannelOneToOneTest, AccumulateIbHostNoAtomicRejected) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::HostNoAtomic); + if (gEnv->rank >= numRanksToUse) return; + + const int peer = 1 - gEnv->rank; + auto buff = mscclpp::GpuBuffer(1).memory(); + mscclpp::RegisteredMemory localMem; + mscclpp::RegisteredMemory remoteMem; + + mscclpp::EndpointConfig cfg; + cfg.transport = ibTransport; + cfg.ib.gidIndex = std::stoi(gEnv->args["ib_gid_index"]); + cfg.ib.mode = IbMode::HostNoAtomic; + + auto connFuture = communicator->connect(cfg, peer); + localMem = communicator->registerMemory(buff.get(), sizeof(int64_t), ibTransport); + communicator->sendMemory(localMem, peer, /*tag=*/77); + auto remoteFuture = communicator->recvMemory(peer, /*tag=*/77); + + auto conn = connFuture.get(); + remoteMem = remoteFuture.get(); + registeredMemories.push_back(localMem); + + try { + conn.accumulate(remoteMem, 0, 1); + FAIL() << "Expected accumulate in IB HostNoAtomic mode to throw InvalidUsage"; + } catch (const mscclpp::Error& e) { + EXPECT_TRUE(e.getErrorCode() == mscclpp::ErrorCode::InvalidUsage); + } + + communicator->bootstrap()->barrier(); +} + static constexpr int kMaxQps = 4; __constant__ DeviceHandle gMultiQpPortChans[kMaxQps]; @@ -959,3 +1270,141 @@ TEST(PortChannelOneToOneTest, SameChanConcurrentFlushIbHostMode) { REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); testSameChanConcurrentFlush(IbMode::Host); } + +void PortChannelFanInTest::SetUp() { + CommunicatorTestBase::SetUp(); + proxyService = std::make_shared(); +} + +void PortChannelFanInTest::TearDown() { CommunicatorTestBase::TearDown(); } + +// Each rank other than 0 pushes nTries accumulates at rank 0's single counter. +__global__ void kernelFanInAccumulate(int nTries) { + DeviceHandle& portChan = gChannelOneToOneTestConstPortChans; + if (threadIdx.x != 0 || blockIdx.x != 0) return; + for (int i = 0; i < nTries; i++) { + portChan.accumulate(0, kAccumulateValue); + } + portChan.flush(); +} + +void PortChannelFanInTest::testFanIn(bool useIPC, bool useIb, bool useEthernet, IbMode ibMode) { + const int worldSize = communicator->bootstrap()->getNranks(); + const int rank = communicator->bootstrap()->getRank(); + if (worldSize < 3) { + SKIP_TEST() << "Fan-in test needs at least 3 ranks to have more than one writer."; + return; + } + const int nTries = 200; + + auto buff = mscclpp::GpuBuffer(1); + MSCCLPP_CUDATHROW(cudaMemset(buff.memory().get(), 0, sizeof(int64_t))); + + // Rank 0 is the target; every other rank connects to it. + mscclpp::TransportFlags transport; + if (useIPC) transport |= mscclpp::Transport::CudaIpc; + if (useIb) transport |= ibTransport; + if (useEthernet) transport |= mscclpp::Transport::Ethernet; + + mscclpp::EndpointConfig cfg; + if (useIPC) { + cfg.transport = mscclpp::Transport::CudaIpc; + } else if (useIb) { + cfg.transport = ibTransport; + cfg.ib.gidIndex = std::stoi(gEnv->args["ib_gid_index"]); + cfg.ib.mode = ibMode; + } else { + cfg.transport = mscclpp::Transport::Ethernet; + } + + mscclpp::RegisteredMemory localMem = communicator->registerMemory(buff.memory().get(), sizeof(int64_t), transport); + registeredMemories.push_back(localMem); + + std::vector> connFutures(worldSize); + std::vector> remoteMemFutures(worldSize); + if (rank == 0) { + for (int r = 1; r < worldSize; r++) { + connFutures[r] = communicator->connect(cfg, r); + communicator->sendMemory(localMem, r); + remoteMemFutures[r] = communicator->recvMemory(r); + } + } else { + connFutures[0] = communicator->connect(cfg, 0); + communicator->sendMemory(localMem, 0); + remoteMemFutures[0] = communicator->recvMemory(0); + } + + std::vector portChannels; + if (rank == 0) { + for (int r = 1; r < worldSize; r++) { + auto sema = communicator->buildSemaphore(connFutures[r].get(), r).get(); + mscclpp::SemaphoreId cid = proxyService->addSemaphore(sema); + portChannels.emplace_back(proxyService->portChannel(cid, proxyService->addMemory(remoteMemFutures[r].get()), + proxyService->addMemory(localMem))); + registeredMemories.push_back(remoteMemFutures[r].get()); + } + } else { + auto sema = communicator->buildSemaphore(connFutures[0].get(), 0).get(); + mscclpp::SemaphoreId cid = proxyService->addSemaphore(sema); + portChannels.emplace_back(proxyService->portChannel(cid, proxyService->addMemory(remoteMemFutures[0].get()), + proxyService->addMemory(localMem))); + registeredMemories.push_back(remoteMemFutures[0].get()); + } + + proxyService->startProxy(); + + if (rank != 0) { + std::vector> handles; + handles.push_back(portChannels[0].deviceHandle()); + MSCCLPP_CUDATHROW(cudaMemcpyToSymbol(gChannelOneToOneTestConstPortChans, handles.data(), + sizeof(DeviceHandle))); + kernelFanInAccumulate<<<1, 1>>>(nTries); + MSCCLPP_CUDATHROW(cudaDeviceSynchronize()); + } + + communicator->bootstrap()->barrier(); + + if (rank == 0) { + // Poll until the total stops changing rather than sleeping a fixed time, so slow arrival is + // not mistaken for a lost update. EthernetConnection::flush() is a no-op, so a sender cannot + // tell when the receiver applied its updates. + const int64_t expected = (int64_t)(worldSize - 1) * nTries * kAccumulateValue; + int64_t observed = 0; + int64_t previous = -1; + int stableRounds = 0; + for (int i = 0; i < 600 && observed != expected; i++) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + mscclpp::gpuMemcpy(reinterpret_cast(&observed), reinterpret_cast(buff.memory().get()), + sizeof(int64_t), cudaMemcpyDeviceToHost); + stableRounds = (observed == previous) ? stableRounds + 1 : 0; + previous = observed; + if (stableRounds >= 50) break; // 5 s with no progress: treat as final + } + if (observed != expected) { + std::cout << "fan-in lost " << (expected - observed) / kAccumulateValue << " of " + << (int64_t)(worldSize - 1) * nTries << " accumulates" << std::endl; + } + EXPECT_EQ(observed, expected); + } + + communicator->bootstrap()->barrier(); + proxyService->stopProxy(); + communicator->bootstrap()->barrier(); +} + +#if defined(__HIP_PLATFORM_AMD__) +// CudaIpc supports many writers on ROCm: the kernel is a real read-modify-write, so writers in +// separate processes do not lose updates. +TEST(PortChannelFanInTest, Accumulate) { + REQUIRE_CUDA_IPC_AVAILABLE; + testFanIn(true, false, false); +} +#endif // defined(__HIP_PLATFORM_AMD__) + +TEST(PortChannelFanInTest, AccumulateIb) { + REQUIRE_IBVERBS; + REQUIRE_GDR_FOR_IB_MODE(IbMode::Host); + testFanIn(false, true, false, IbMode::Host); +} + +TEST(PortChannelFanInTest, AccumulateEthernet) { testFanIn(false, false, true); }