diff --git a/include/mscclpp/core.hpp b/include/mscclpp/core.hpp index a0c9f749..b990e210 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 4670f47c..d6304138 100644 --- a/include/mscclpp/fifo_device.hpp +++ b/include/mscclpp/fifo_device.hpp @@ -15,10 +15,21 @@ 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; 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. +constexpr TriggerType TriggerAccumulate = 6; // Add a value to remote memory. +// 7 is unassigned. constexpr unsigned int TriggerBitsSize = 32; constexpr unsigned int TriggerBitsOffset = 32; @@ -29,6 +40,8 @@ constexpr unsigned int TriggerBitsSemaphoreId = 10; // there. See FifoDeviceHandle::push(). constexpr unsigned int TriggerBitsFifoReserved = 1; +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. /// Most significant bit of snd is reserved. diff --git a/include/mscclpp/port_channel.hpp b/include/mscclpp/port_channel.hpp index 18d67524..b8d5f850 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 fd575b4c..a6ce3632 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,13 +118,31 @@ 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); } + /// 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(); } @@ -149,7 +166,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 +174,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 +187,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,12 +202,19 @@ 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) { 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 00000000..e2730f26 --- /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 e01c4a6e..32b4ff4f 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 eda4b3ef..1d94a06c 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. @@ -94,7 +96,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_; }; @@ -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 42d03db1..54a6db64 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 0601ef84..6cb0dacd 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,55 @@ 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++; + }; + 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) { + 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; + case TriggerAccumulate: + accumulate(); + 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/mp_unit_tests.hpp b/test/mp_unit/mp_unit_tests.hpp index 8654ccc9..0b1e18d9 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 eec1760c..fd532693 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]; @@ -897,7 +1208,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 +1247,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); @@ -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); }