Karthikeyan Arumugam
bb599d8ed7
* Added support for AMD ROCm net-ib alongside vanilla net-ib, with auto-generation to detect conflicts early during NCCL sync and enable future customizations.
* Integrated AMD AINIC support in RCCL for out-of-the-box usage, leveraging performance improvements by default, channel pinning for optimal pipeline performance, and extended support for 32B in-line CTS messages.
* Implemented internal derivation of AINIC-specific flags when RCCL AINIC environment parameter is set, and checks before initializing AINIC net-ib methods.
* Included snapshot of auto-generated ROCm net-ib file (src/transport/net_ib_rocm.cc) for reference.
* Fixed typos in RCCL param API (RCCL_AINIC_ROCE) and dlclose.
* Updated plugin loading logic:
* Load internal ROCmIB plugin only when NCCL_NET_PLUGIN is not set.
* Load default internal net-ib only when not AINIC and no external plugin env is set.
[ROCm/rccl commit: 9f4651f20f]
2115 lines
95 KiB
C++
2115 lines
95 KiB
C++
/*************************************************************************
|
|
* Copyright (c) 2016-2022, NVIDIA CORPORATION. All rights reserved.
|
|
* Modifications Copyright (c) 2019-2023 Advanced Micro Devices, Inc. All rights reserved.
|
|
* Modifications Copyright (c) Microsoft Corporation. Licensed under the MIT License.
|
|
*
|
|
* See LICENSE.txt for license information
|
|
************************************************************************/
|
|
|
|
#include "comm.h"
|
|
#include "net.h"
|
|
#include "graph.h"
|
|
#include "proxy.h"
|
|
#include "collectives.h"
|
|
#include "gdrwrap.h"
|
|
#include "shmutils.h"
|
|
#include "p2p.h"
|
|
#include "profiler.h"
|
|
#include "transport.h"
|
|
#include "shm.h"
|
|
#include <assert.h>
|
|
#include "graph.h"
|
|
#include "graph/topo.h"
|
|
#include "nccl_net.h"
|
|
#include "register_inline.h"
|
|
#if defined(ENABLE_NPKIT)
|
|
#include "npkit/npkit.h"
|
|
#endif
|
|
#include "msccl/msccl_lifecycle.h"
|
|
|
|
static_assert(sizeof(ncclNetHandle_t) <= CONNECT_SIZE, "NET Connect info is too large");
|
|
|
|
#define NCCL_NET_MAP_HOSTMEM 0
|
|
#define NCCL_NET_MAP_DEVMEM 1
|
|
#define NCCL_NET_MAP_SHARED_HOSTMEM 2
|
|
#define NCCL_NET_MAP_SHARED_DEVMEM 3
|
|
#define NCCL_NET_MAP_GDCMEM 4
|
|
#define NCCL_NET_MAP_MEMS 5
|
|
|
|
#define NCCL_NET_MAP_MASK_DEVMEM 0x40000000
|
|
#define NCCL_NET_MAP_MASK_SHARED 0x80000000
|
|
#define NCCL_NET_MAP_MASK_USED 0x20000000
|
|
#define NCCL_NET_MAP_MASK_OFFSET 0x1fffffff
|
|
|
|
#define NCCL_NET_MAP_OFFSET_BANK(mapStruct, offsetName) \
|
|
((mapStruct)->offsets.offsetName >> 30)
|
|
|
|
#define NCCL_NET_MAP_OFFSET_NULL(mapStruct, offsetName) \
|
|
(((mapStruct)->offsets.offsetName >> 29) == 0)
|
|
|
|
#define NCCL_NET_MAP_GET_POINTER(mapStruct, cpuOrGpu, offsetName) \
|
|
(NCCL_NET_MAP_OFFSET_NULL(mapStruct, offsetName) ? NULL : \
|
|
(mapStruct)->mems[NCCL_NET_MAP_OFFSET_BANK(mapStruct, offsetName)].cpuOrGpu##Ptr + ((mapStruct)->offsets.offsetName & NCCL_NET_MAP_MASK_OFFSET))
|
|
|
|
#define NCCL_NET_MAP_DEV_MEM(mapStruct, offsetName) \
|
|
(((mapStruct)->offsets.offsetName & NCCL_NET_MAP_MASK_DEVMEM) != 0)
|
|
|
|
#define NCCL_NET_MAP_ADD_POINTER(mapStruct, shared, dev, memSize, offsetName) do { \
|
|
int bank = NCCL_NET_MAP_MASK_USED + (dev)*NCCL_NET_MAP_MASK_DEVMEM + (shared)*NCCL_NET_MAP_MASK_SHARED; \
|
|
if ((shared) == 0) { \
|
|
if (dev) { \
|
|
(mapStruct)->offsets.offsetName = bank + (mapStruct)->mems[NCCL_NET_MAP_DEVMEM].size; \
|
|
(mapStruct)->mems[NCCL_NET_MAP_DEVMEM].size += memSize; \
|
|
} else { \
|
|
(mapStruct)->offsets.offsetName = bank + (mapStruct)->mems[NCCL_NET_MAP_HOSTMEM].size; \
|
|
(mapStruct)->mems[NCCL_NET_MAP_HOSTMEM].size += memSize; \
|
|
} \
|
|
} else { \
|
|
(mapStruct)->offsets.offsetName = bank; \
|
|
} \
|
|
} while (0);
|
|
|
|
struct connectMapMem{
|
|
char* gpuPtr;
|
|
char* cpuPtr;
|
|
int size;
|
|
ncclIpcDesc ipcDesc;
|
|
ncclShmIpcDesc_t attachDesc;
|
|
ncclShmIpcDesc_t createDesc;
|
|
};
|
|
|
|
struct connectMap {
|
|
int sameProcess;
|
|
int shared;
|
|
int cudaDev;
|
|
// First 3 bits of offsets determine the mem bank. 001 is host mem, 011 is dev mem, 101 is shared host mem and 111 is shared dev mem.
|
|
struct connectMapMem mems[NCCL_NET_MAP_MEMS];
|
|
// Offsets. 3 MSBs indicate mem bank, 111 indicates NULL.
|
|
struct {
|
|
uint32_t sendMem;
|
|
uint32_t recvMem;
|
|
uint32_t buffs[NCCL_NUM_PROTOCOLS];
|
|
} offsets;
|
|
};
|
|
|
|
struct sendNetResources {
|
|
struct connectMap map;
|
|
void* netSendComm;
|
|
struct ncclSendMem* sendMem;
|
|
struct ncclRecvMem* recvMem;
|
|
|
|
int tpRank;
|
|
int tpLocalRank;
|
|
int tpRemoteRank;
|
|
int netDev;
|
|
enum ncclTopoGdrMode useGdr;
|
|
int useDmaBuf;
|
|
int maxRecvs;
|
|
uint64_t* gdcSync;
|
|
void* gdrDesc;
|
|
int shared;
|
|
int channelId;
|
|
int connIndex;
|
|
char* buffers[NCCL_NUM_PROTOCOLS];
|
|
int buffSizes[NCCL_NUM_PROTOCOLS];
|
|
void* mhandles[NCCL_NUM_PROTOCOLS];
|
|
uint64_t step;
|
|
uint64_t llLastCleaning;
|
|
int netDeviceVersion;
|
|
ncclNetDeviceType netDeviceType;
|
|
ncclNetDeviceHandle_t* netDeviceHandle;
|
|
size_t maxP2pBytes;
|
|
volatile uint32_t* curr_hdp_reg; // Curr GPU in ring (for rdma transport use only)
|
|
int isP2p;
|
|
};
|
|
|
|
struct recvNetResources {
|
|
struct connectMap map;
|
|
void* netListenComm;
|
|
void* netRecvComm;
|
|
struct ncclSendMem* sendMem;
|
|
struct ncclRecvMem* recvMem;
|
|
|
|
int tpRank;
|
|
int tpLocalRank;
|
|
int tpRemoteRank;
|
|
int tpRemoteProxyRank;
|
|
int netDev;
|
|
enum ncclTopoGdrMode useGdr;
|
|
int useDmaBuf;
|
|
int needFlush;
|
|
int maxRecvs;
|
|
uint64_t* gdcSync;
|
|
uint64_t* gdcFlush;
|
|
void* gdrDesc;
|
|
int shared;
|
|
int channelId;
|
|
int connIndex;
|
|
char* buffers[NCCL_NUM_PROTOCOLS];
|
|
int buffSizes[NCCL_NUM_PROTOCOLS];
|
|
void* mhandles[NCCL_NUM_PROTOCOLS];
|
|
uint64_t step;
|
|
uint64_t llLastCleaning;
|
|
int netDeviceVersion;
|
|
ncclNetDeviceType netDeviceType;
|
|
ncclNetDeviceHandle_t* netDeviceHandle;
|
|
size_t maxP2pBytes;
|
|
volatile uint32_t* curr_hdp_reg; // Curr GPU in ring (for rdma transport use only)
|
|
};
|
|
|
|
struct netRegInfo {
|
|
uintptr_t buffer;
|
|
size_t size;
|
|
};
|
|
|
|
/* Determine if two peers can communicate with NET */
|
|
static ncclResult_t canConnect(int* ret, struct ncclComm* comm, struct ncclTopoGraph* graph, struct ncclPeerInfo* info1, struct ncclPeerInfo* info2) {
|
|
*ret = 1;
|
|
#if defined(__HIP_PLATFORM_AMD__) || defined(__HIPCC__)
|
|
#else
|
|
if (info1->hostHash == info2->hostHash) {
|
|
// If on the same host, check intra-node net is not disabled.
|
|
NCCLCHECK(ncclTopoCheckNet(comm->topo, info1->rank, info2->rank, ret));
|
|
}
|
|
#endif
|
|
return ncclSuccess;
|
|
}
|
|
|
|
NCCL_PARAM(NetSharedBuffers, "NET_SHARED_BUFFERS", -2);
|
|
NCCL_PARAM(NetSharedComms, "NET_SHARED_COMMS", 1);
|
|
|
|
#if defined(HIP_CONTIGUOUS_MEMORY)
|
|
RCCL_PARAM(NetContiguousMem, "NET_CONTIGUOUS_MEM", 0);
|
|
#endif
|
|
|
|
struct setupReq {
|
|
int tpRank;
|
|
int tpLocalRank;
|
|
int tpRemoteRank;
|
|
int shared;
|
|
int netDev;
|
|
enum ncclTopoGdrMode useGdr;
|
|
int needFlush;
|
|
int channelId;
|
|
int connIndex;
|
|
uint32_t* curr_hdp_reg;
|
|
int isP2p;
|
|
};
|
|
|
|
NCCL_PARAM(NetOptionalRecvCompletion, "NET_OPTIONAL_RECV_COMPLETION", 1);
|
|
RCCL_PARAM(AinicRoce, "AINIC_ROCE", 0);
|
|
|
|
static_assert(sizeof(ncclNetHandle_t) + sizeof(int) <= CONNECT_SIZE, "Not large enough ncclConnect to hold ncclNetHandle_t and useGdr flag");
|
|
// Forward declaration
|
|
static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct ncclProxyArgs* args);
|
|
|
|
// Returns the flags to be used by a call to cuMemGetHandleForAddressRange.
|
|
static inline int getHandleForAddressRangeFlags(ncclTopoGdrMode useGdr) {
|
|
int flags = 0;
|
|
#if CUDA_VERSION >= 12080
|
|
// Force mapping on PCIe on systems with both PCI and C2C attachments.
|
|
if (useGdr == ncclTopoGdrModePci) flags = CU_MEM_RANGE_FLAG_DMA_BUF_MAPPING_TYPE_PCIE;
|
|
#endif
|
|
return flags;
|
|
}
|
|
|
|
/* Determine if we will use this transport for this peer and return connect
|
|
* information for this peer */
|
|
static ncclResult_t sendSetup(struct ncclComm* comm, struct ncclTopoGraph* graph, struct ncclPeerInfo* myInfo, struct ncclPeerInfo* peerInfo, struct ncclConnect* connectInfo, struct ncclConnector* send, int channelId, int connIndex) {
|
|
struct setupReq req = { 0 };
|
|
|
|
send->conn.shared = req.shared = (graph || connIndex == 0 || mscclIsCaller()) ? 0 : ncclParamNetSharedBuffers() != -2 ? ncclParamNetSharedBuffers() : 1;
|
|
req.channelId = channelId;
|
|
req.connIndex = connIndex;
|
|
req.curr_hdp_reg = 0;
|
|
req.netDev = -1;
|
|
// Determine if this is a P2P connection or not based on the graph pointer
|
|
if(graph == NULL) {
|
|
req.isP2p = 1;
|
|
} else {
|
|
req.isP2p = 0;
|
|
}
|
|
|
|
int proxyRank = myInfo->rank;
|
|
int64_t netId;
|
|
if (connIndex == NCCL_CONN_IDX_P2P_NET) NCCLCHECK(ncclTopoGetIntraNetDev(comm->topo, myInfo->rank, graph, channelId, 1, &netId, &req.netDev));
|
|
if (req.netDev < 0) NCCLCHECK(ncclTopoGetNetDev(comm, myInfo->rank, graph, channelId, peerInfo->rank, &netId, &req.netDev, &proxyRank));
|
|
NCCLCHECK(ncclTopoCheckGdr(comm->topo, myInfo->rank, netId, 1, &req.useGdr));
|
|
send->conn.flags |= req.useGdr ? NCCL_DIRECT_NIC : 0;
|
|
if (!req.useGdr && connIndex == 0) comm->useGdr = 0;
|
|
if (proxyRank != myInfo->rank && connIndex == 0) comm->useNetPXN = true;
|
|
if (req.useGdr && !IsArchMatch(comm->topo->nodes[GPU].nodes[0].gpu.gcn, "gfx90a") && !IsArchMatch(comm->topo->nodes[GPU].nodes[0].gpu.gcn, "gfx942") && !IsArchMatch(comm->topo->nodes[GPU].nodes[0].gpu.gcn, "gfx950")) {
|
|
CUDACHECK(hipDeviceGetAttribute((int*)&req.curr_hdp_reg, hipDeviceAttributeHdpMemFlushCntl, myInfo->cudaDev));
|
|
send->conn.curr_hdp_reg = req.curr_hdp_reg;
|
|
}
|
|
|
|
NCCLCHECK(ncclProxyConnect(comm, TRANSPORT_NET, 1, proxyRank, &send->proxyConn));
|
|
req.tpLocalRank = comm->topParentLocalRanks[comm->localRank];
|
|
req.tpRank = comm->topParentRanks[myInfo->rank];
|
|
req.tpRemoteRank = comm->topParentRanks[peerInfo->rank];
|
|
NCCLCHECK(ncclProxyCallBlocking(comm, &send->proxyConn, ncclProxyMsgSetup, &req, sizeof(req), NULL, 0));
|
|
|
|
if (proxyRank == myInfo->rank) {
|
|
INFO(NCCL_INIT|NCCL_NET,"Channel %02d/%d : %d[%d] -> %d[%d] [send] via NET/%s/%d%s%s%s comm %p nRanks %02d", channelId, connIndex, myInfo->rank, myInfo->nvmlDev, peerInfo->rank, peerInfo->nvmlDev, comm->ncclNet->name, req.netDev,
|
|
req.useGdr ? "/GDRDMA" : "", req.useGdr==ncclTopoGdrModePci ? "(PCI)" : "",
|
|
req.shared ? "/Shared" : "", comm, comm->nRanks);
|
|
} else {
|
|
INFO(NCCL_INIT|NCCL_NET,"Channel %02d/%d : %d[%d] -> %d[%d] [send] via NET/%s/%d(%d)%s%s%s comm %p nRanks %02d", channelId, connIndex, myInfo->rank, myInfo->nvmlDev, peerInfo->rank, peerInfo->nvmlDev, comm->ncclNet->name, req.netDev,
|
|
proxyRank,
|
|
req.useGdr ? "/GDRDMA" : "", req.useGdr==ncclTopoGdrModePci ? "(PCI)" : "",
|
|
req.shared ? "/Shared" : "", comm, comm->nRanks);
|
|
}
|
|
*((int*)connectInfo) = comm->topParentRanks[proxyRank];
|
|
memcpy((uint8_t*)connectInfo + sizeof(ncclNetHandle_t), &req.useGdr, sizeof(int));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
// GDRCOPY support: TAIL_ENABLE When enabled locates the RX proxy tail in CUDA memory
|
|
NCCL_PARAM(GdrCopySyncEnable, "GDRCOPY_SYNC_ENABLE", 1);
|
|
// GDRCOPY support: FLUSH_ENABLE When enabled uses a PCI-E read to flush GDRDMA buffers
|
|
NCCL_PARAM(GdrCopyFlushEnable, "GDRCOPY_FLUSH_ENABLE", 0);
|
|
|
|
/* Setup recv connector */
|
|
static ncclResult_t recvSetup(struct ncclComm* comm, struct ncclTopoGraph* graph, struct ncclPeerInfo* myInfo, struct ncclPeerInfo* peerInfo, struct ncclConnect* connectInfo, struct ncclConnector* recv, int channelId, int connIndex) {
|
|
struct setupReq req = { 0 };
|
|
|
|
recv->conn.shared = req.shared = (graph || connIndex == 0 || mscclIsCaller()) ? 0 : ncclParamNetSharedBuffers() != -2 ? ncclParamNetSharedBuffers() : 1;
|
|
req.channelId = channelId;
|
|
req.connIndex = connIndex;
|
|
req.netDev = -1;
|
|
|
|
// Use myInfo->rank as the receiver uses its own NIC
|
|
int proxyRank = myInfo->rank;
|
|
int64_t netId;
|
|
if (connIndex == NCCL_CONN_IDX_P2P_NET) NCCLCHECK(ncclTopoGetIntraNetDev(comm->topo, myInfo->rank, graph, channelId, 0, &netId, &req.netDev));
|
|
if (req.netDev < 0) NCCLCHECK(ncclTopoGetNetDev(comm, myInfo->rank, graph, channelId, myInfo->rank, &netId, &req.netDev, &proxyRank));
|
|
NCCLCHECK(ncclTopoCheckGdr(comm->topo, myInfo->rank, netId, 0, &req.useGdr));
|
|
recv->conn.flags |= req.useGdr ? NCCL_DIRECT_NIC : 0;
|
|
if (!req.useGdr && connIndex == 0) comm->useGdr = 0;
|
|
|
|
// Determine whether we need to flush the GDR buffer on recv or not
|
|
if (req.useGdr) {
|
|
NCCLCHECK(ncclTopoNeedFlush(comm, netId, req.netDev, myInfo->rank, &req.needFlush));
|
|
CUDACHECK(hipDeviceGetAttribute((int*)&req.curr_hdp_reg, hipDeviceAttributeHdpMemFlushCntl, myInfo->cudaDev));
|
|
recv->conn.curr_hdp_reg = req.curr_hdp_reg;
|
|
}
|
|
|
|
// We don't support PXN on receive yet
|
|
NCCLCHECK(ncclProxyConnect(comm, TRANSPORT_NET, 0, myInfo->rank, &recv->proxyConn));
|
|
|
|
req.tpLocalRank = comm->topParentLocalRanks[comm->localRank];
|
|
req.tpRank = comm->topParentRanks[myInfo->rank];
|
|
req.tpRemoteRank = comm->topParentRanks[peerInfo->rank];
|
|
NCCLCHECK(ncclProxyCallBlocking(comm, &recv->proxyConn, ncclProxyMsgSetup, &req, sizeof(req), connectInfo, sizeof(ncclNetHandle_t)));
|
|
memcpy((uint8_t*)connectInfo + sizeof(ncclNetHandle_t), &req.useGdr, sizeof(int));
|
|
INFO(NCCL_INIT|NCCL_NET,"Channel %02d/%d : %d[%d] -> %d[%d] [receive] via NET/%s/%d%s%s%s comm %p nRanks %02d", channelId, connIndex, peerInfo->rank, peerInfo->nvmlDev, myInfo->rank, myInfo->nvmlDev, comm->ncclNet->name, req.netDev,
|
|
req.useGdr ? "/GDRDMA" : "", req.useGdr==ncclTopoGdrModePci ? "(PCI)" : "",
|
|
req.shared ? "/Shared" : "", comm, comm->nRanks);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t netMapShm(struct ncclComm *comm, struct ncclProxyConnector* proxyConn, struct connectMapMem* mem) {
|
|
NCCLCHECK(ncclShmImportShareableBuffer(comm, proxyConn->rank, &mem->createDesc, (void**)&mem->cpuPtr, (void**)&mem->gpuPtr, &mem->attachDesc));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t netCreateShm(struct ncclProxyState* proxyState, struct connectMapMem* mem) {
|
|
NCCLCHECK(ncclShmAllocateShareableBuffer(mem->size, false, &mem->createDesc, (void**)&mem->cpuPtr, (void**)&mem->gpuPtr));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t netDumpMap(struct connectMap* map) {
|
|
printf("Dump map same process %d shared %d\n", map->sameProcess, map->shared);
|
|
struct connectMapMem *mem = map->mems+NCCL_NET_MAP_HOSTMEM;
|
|
printf("Mem 0: Host mem (%x B) CPU %p GPU %p\n", mem->size, mem->cpuPtr, mem->gpuPtr);
|
|
mem = map->mems+NCCL_NET_MAP_DEVMEM;
|
|
printf("Mem 1: Vid mem (%x B) CPU %p GPU %p\n", mem->size, mem->cpuPtr, mem->gpuPtr);
|
|
mem = map->mems+NCCL_NET_MAP_SHARED_HOSTMEM;
|
|
printf("Mem 2: Shared Host mem (%x B) CPU %p GPU %p\n", mem->size, mem->cpuPtr, mem->gpuPtr);
|
|
mem = map->mems+NCCL_NET_MAP_SHARED_DEVMEM;
|
|
printf("Mem 3: Shared Vid mem (%x B) CPU %p GPU %p\n", mem->size, mem->cpuPtr, mem->gpuPtr);
|
|
printf("SendMem -> Used %d Bank %d Offset %x, cpu %p gpu %p\n",
|
|
map->offsets.sendMem & NCCL_NET_MAP_MASK_USED ? 1 : 0,
|
|
NCCL_NET_MAP_OFFSET_BANK(map, sendMem), map->offsets.sendMem & NCCL_NET_MAP_MASK_OFFSET,
|
|
NCCL_NET_MAP_GET_POINTER(map, cpu, sendMem), NCCL_NET_MAP_GET_POINTER(map, gpu, sendMem));
|
|
printf("RecvMem -> Used %d Bank %d Offset %x, cpu %p gpu %p\n",
|
|
map->offsets.recvMem & NCCL_NET_MAP_MASK_USED ? 1 : 0,
|
|
NCCL_NET_MAP_OFFSET_BANK(map, recvMem), map->offsets.recvMem & NCCL_NET_MAP_MASK_OFFSET,
|
|
NCCL_NET_MAP_GET_POINTER(map, cpu, recvMem), NCCL_NET_MAP_GET_POINTER(map, gpu, recvMem));
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
printf("Proto %d -> Used %d Bank %d Offset %x, cpu %p, gpu %p\n", p,
|
|
map->offsets.buffs[p] & NCCL_NET_MAP_MASK_USED ? 1 : 0,
|
|
NCCL_NET_MAP_OFFSET_BANK(map, buffs[p]), map->offsets.buffs[p] & NCCL_NET_MAP_MASK_OFFSET,
|
|
NCCL_NET_MAP_GET_POINTER(map, cpu, buffs[p]), NCCL_NET_MAP_GET_POINTER(map, gpu, buffs[p]));
|
|
}
|
|
printf("End of dump\n");
|
|
return ncclSuccess;
|
|
}
|
|
|
|
struct netSendConnectArgs {
|
|
ncclNetHandle_t handle;
|
|
int trafficClass;
|
|
};
|
|
|
|
struct netRecvConnectArgs {
|
|
int proxyRank;
|
|
};
|
|
|
|
static ncclResult_t sendConnect(struct ncclComm* comm, struct ncclConnect* connectInfo, int nranks, int rank, struct ncclConnector* send) {
|
|
struct connectMap* map = (connectMap*) send->transportResources;
|
|
void* opId;
|
|
int recvUseGdr;
|
|
|
|
memcpy(&recvUseGdr, (uint8_t*)connectInfo + sizeof(ncclNetHandle_t), sizeof(int));
|
|
if (!recvUseGdr) send->conn.flags &= ~NCCL_DIRECT_NIC;
|
|
|
|
// map isn't allocated thus this op hasn't been submitted yet
|
|
if (!map) {
|
|
// Setup device pointers
|
|
NCCLCHECK(ncclCalloc(&map, 1));
|
|
send->transportResources = map;
|
|
opId = send;
|
|
INFO(NCCL_PROXY, "sendConnect ncclProxyCallAsync opId=%p", opId);
|
|
netSendConnectArgs args = {{},0};
|
|
memcpy(&args.handle, connectInfo, sizeof(ncclNetHandle_t));
|
|
args.trafficClass = comm->config.trafficClass;
|
|
NCCLCHECK(ncclProxyCallAsync(comm, &send->proxyConn, ncclProxyMsgConnect, &args, sizeof(netSendConnectArgs), sizeof(struct connectMap), opId));
|
|
} else {
|
|
opId = send;
|
|
}
|
|
|
|
ncclResult_t ret;
|
|
ret = ncclPollProxyResponse(comm, &send->proxyConn, map, opId);
|
|
if (ret != ncclSuccess) {
|
|
if (ret != ncclInProgress) {
|
|
free(map);
|
|
send->transportResources = NULL;
|
|
}
|
|
return ret;
|
|
}
|
|
INFO(NCCL_PROXY, "sendConnect ncclPollProxyResponse opId=%p", opId);
|
|
|
|
if (map->sameProcess && !ncclCuMemEnable()) {
|
|
if (map->cudaDev != comm->cudaDev) {
|
|
// Enable P2P access for Legacy IPC
|
|
cudaError_t err = cudaDeviceEnablePeerAccess(map->cudaDev, 0);
|
|
if (err == cudaErrorPeerAccessAlreadyEnabled) {
|
|
cudaGetLastError();
|
|
} else if (err != cudaSuccess) {
|
|
WARN("failed to peer with device %d: %d %s", map->cudaDev, err, cudaGetErrorString(err));
|
|
return ncclInternalError;
|
|
}
|
|
}
|
|
} else if (!(map->sameProcess && map->cudaDev == comm->cudaDev)) {
|
|
if (!map->sameProcess) NCCLCHECK(netMapShm(comm, &send->proxyConn, map->mems + NCCL_NET_MAP_HOSTMEM));
|
|
if (map->mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr = NULL;
|
|
NCCLCHECK(ncclP2pImportShareableBuffer(comm, send->proxyConn.rank,
|
|
map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
&map->mems[NCCL_NET_MAP_DEVMEM].ipcDesc,
|
|
(void**)&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr));
|
|
map->mems[NCCL_NET_MAP_DEVMEM].cpuPtr = NULL;
|
|
}
|
|
if (map->mems[NCCL_NET_MAP_SHARED_DEVMEM].size) {
|
|
void** sharedDevMemPtr = comm->proxyState->sharedDevMems + send->proxyConn.tpLocalRank;
|
|
if (*sharedDevMemPtr == NULL) {
|
|
map->mems[NCCL_NET_MAP_SHARED_DEVMEM].gpuPtr = NULL;
|
|
NCCLCHECK(ncclP2pImportShareableBuffer(comm, send->proxyConn.rank,
|
|
map->mems[NCCL_NET_MAP_SHARED_DEVMEM].size,
|
|
&map->mems[NCCL_NET_MAP_SHARED_DEVMEM].ipcDesc,
|
|
sharedDevMemPtr));
|
|
}
|
|
map->mems[NCCL_NET_MAP_SHARED_DEVMEM].gpuPtr = (char*)(*sharedDevMemPtr);
|
|
map->mems[NCCL_NET_MAP_SHARED_DEVMEM].cpuPtr = NULL;
|
|
}
|
|
}
|
|
//NCCLCHECK(netDumpMap(map));
|
|
|
|
struct ncclSendMem *sendMem = (struct ncclSendMem*) NCCL_NET_MAP_GET_POINTER(map, gpu, sendMem);
|
|
void* gdcMem = map->mems[NCCL_NET_MAP_GDCMEM].gpuPtr;
|
|
send->conn.head = gdcMem ? (uint64_t*)gdcMem : &sendMem->head;
|
|
|
|
struct ncclRecvMem *recvMem = (struct ncclRecvMem*) NCCL_NET_MAP_GET_POINTER(map, gpu, recvMem);
|
|
send->conn.tail = &recvMem->tail;
|
|
send->conn.stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE]/NCCL_STEPS;
|
|
send->conn.connFifo = recvMem->connFifo;
|
|
struct ncclRecvMem *recvMemCpu = (struct ncclRecvMem*) NCCL_NET_MAP_GET_POINTER(map, cpu, recvMem);
|
|
for (int i=0; i<NCCL_STEPS; i++) {
|
|
recvMemCpu->connFifo[i].offset = -1;
|
|
recvMemCpu->connFifo[i].mode = map->shared ? NCCL_MODE_OFFSET : NCCL_MODE_NORMAL;
|
|
}
|
|
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++)
|
|
send->conn.buffs[p] = NCCL_NET_MAP_GET_POINTER(map, gpu, buffs[p]);
|
|
|
|
if (send->proxyConn.sameProcess) {
|
|
if (send->proxyConn.connection->netDeviceHandle) {
|
|
send->conn.netDeviceHandle = *send->proxyConn.connection->netDeviceHandle;
|
|
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++)
|
|
send->conn.mhandles[p] = send->proxyConn.connection->mhandles[p];
|
|
}
|
|
|
|
if (send->proxyConn.connection->needsProxyProgress) {
|
|
send->proxyConn.proxyProgress = sendProxyProgress;
|
|
} else {
|
|
send->proxyConn.proxyProgress = NULL;
|
|
}
|
|
} else {
|
|
send->proxyConn.proxyProgress = sendProxyProgress;
|
|
}
|
|
|
|
return ncclSuccess;
|
|
}
|
|
|
|
// Forward declare
|
|
static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct ncclProxyArgs* args);
|
|
|
|
/* Connect to this peer */
|
|
static ncclResult_t recvConnect(struct ncclComm* comm, struct ncclConnect* connectInfo, int nranks, int rank, struct ncclConnector* recv) {
|
|
struct connectMap* map = (connectMap*) recv->transportResources;
|
|
void* opId;
|
|
int sendUseGdr;
|
|
|
|
memcpy(&sendUseGdr, (uint8_t*)connectInfo + sizeof(ncclNetHandle_t), sizeof(int));
|
|
if (!sendUseGdr) recv->conn.flags &= ~NCCL_DIRECT_NIC;
|
|
|
|
if (!map) {
|
|
NCCLCHECK(ncclCalloc(&map, 1));
|
|
recv->transportResources = map;
|
|
// Use recv connector as unique identifier
|
|
opId = recv;
|
|
INFO(NCCL_PROXY, "recvConnect ncclProxyCallAsync opId=%p &recv->proxyConn=%p connectInfo=%p",
|
|
opId, &recv->proxyConn, connectInfo);
|
|
netRecvConnectArgs args = {0};
|
|
args.proxyRank = *((int*)connectInfo);
|
|
NCCLCHECK(ncclProxyCallAsync(comm, &recv->proxyConn, ncclProxyMsgConnect, &args, sizeof(netRecvConnectArgs), sizeof(struct connectMap), opId));
|
|
} else {
|
|
opId = recv;
|
|
}
|
|
|
|
ncclResult_t ret;
|
|
NCCLCHECK(ret = ncclPollProxyResponse(comm, &recv->proxyConn, map, opId));
|
|
if (ret != ncclSuccess) {
|
|
if (ret != ncclInProgress) {
|
|
free(map);
|
|
recv->transportResources = NULL;
|
|
}
|
|
return ret;
|
|
}
|
|
INFO(NCCL_PROXY, "recvConnect ncclPollProxyResponse opId=%p", opId);
|
|
//NCCLCHECK(netDumpMap(map));
|
|
|
|
struct ncclSendMem *sendMem = (struct ncclSendMem*) NCCL_NET_MAP_GET_POINTER(map, gpu, sendMem);
|
|
recv->conn.head = &sendMem->head;
|
|
|
|
struct ncclRecvMem *recvMem = (struct ncclRecvMem*) NCCL_NET_MAP_GET_POINTER(map, gpu, recvMem);
|
|
void* gdcMem = map->mems[NCCL_NET_MAP_GDCMEM].gpuPtr;
|
|
recv->conn.tail = gdcMem ? (uint64_t*)gdcMem : &recvMem->tail;
|
|
recv->conn.stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE]/NCCL_STEPS;
|
|
recv->conn.connFifo = recvMem->connFifo;
|
|
// Only fuse P2P buffers, continue to allocate dedicated buffers for ring/tree
|
|
for (int i=0; i<NCCL_STEPS; i++) {
|
|
recvMem->connFifo[i].mode = map->shared ? NCCL_MODE_OFFSET : NCCL_MODE_NORMAL;
|
|
}
|
|
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++)
|
|
recv->conn.buffs[p] = NCCL_NET_MAP_GET_POINTER(map, gpu, buffs[p]);
|
|
|
|
if (recv->proxyConn.sameProcess) {
|
|
if (recv->proxyConn.connection->netDeviceHandle) {
|
|
recv->conn.netDeviceHandle = *recv->proxyConn.connection->netDeviceHandle;
|
|
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++)
|
|
recv->conn.mhandles[p] = recv->proxyConn.connection->mhandles[p];
|
|
}
|
|
|
|
if (recv->proxyConn.connection->needsProxyProgress) {
|
|
recv->proxyConn.proxyProgress = recvProxyProgress;
|
|
} else {
|
|
recv->proxyConn.proxyProgress = NULL;
|
|
}
|
|
} else {
|
|
recv->proxyConn.proxyProgress = recvProxyProgress;
|
|
}
|
|
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sendFree(struct ncclConnector* send) {
|
|
struct connectMap* map = (struct connectMap*)(send->transportResources);
|
|
if (map) {
|
|
int cudaDev;
|
|
CUDACHECK(cudaGetDevice(&cudaDev));
|
|
if (map->cudaDev != cudaDev && map->mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
if (ncclCuMemEnable()) {
|
|
// cuMem API support
|
|
NCCLCHECK(ncclP2pFreeShareableBuffer(&map->mems[NCCL_NET_MAP_DEVMEM].ipcDesc));
|
|
NCCLCHECK(ncclCuMemFree(map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr));
|
|
} else {
|
|
// Legacy CUDA IPC support
|
|
CUDACHECK(cudaIpcCloseMemHandle(map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr));
|
|
}
|
|
}
|
|
if (!map->sameProcess) {
|
|
NCCLCHECK(ncclShmIpcClose(&map->mems[NCCL_NET_MAP_HOSTMEM].attachDesc));
|
|
}
|
|
free(map);
|
|
}
|
|
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t recvFree(struct ncclConnector* recv) {
|
|
if (recv->transportResources) free(recv->transportResources);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
#define NCCL_SHARED_STEPS 16
|
|
static ncclResult_t sharedNetBuffersInit(struct ncclProxyState* proxyState, int cuda, int tpLocalRank, int type, int sameProcess,
|
|
int nChannels, char** gpuPtr, char** cpuPtr, int* size, ncclIpcDesc *ipcDesc) {
|
|
if (cuda == 0 && sameProcess == 0) {
|
|
WARN("PXN should not use host buffers for data");
|
|
return ncclInternalError;
|
|
}
|
|
struct ncclProxyProgressState* progressState = &proxyState->progressState;
|
|
if (progressState->localPeers == NULL) {
|
|
NCCLCHECK(ncclCalloc(&progressState->localPeers, proxyState->tpLocalnRanks));
|
|
}
|
|
struct ncclProxyPeer** localPeers = progressState->localPeers;
|
|
if (localPeers[tpLocalRank] == NULL) {
|
|
NCCLCHECK(ncclCalloc(localPeers + tpLocalRank, 1));
|
|
}
|
|
struct ncclProxyPeer* peer = localPeers[tpLocalRank];
|
|
struct ncclProxySharedP2p* state = type == 0 ? &peer->send : &peer->recv;
|
|
state->refcount++;
|
|
if (state->size == 0) {
|
|
state->size = (int64_t)nChannels * NCCL_SHARED_STEPS * proxyState->p2pChunkSize;
|
|
|
|
}
|
|
|
|
if (size) *size = state->size;
|
|
|
|
if (cuda && state->cudaBuff == NULL) {
|
|
if (sameProcess == 0 || ncclCuMemEnable()) {
|
|
NCCLCHECK(ncclP2pAllocateShareableBuffer(state->size, 0, &state->ipcDesc, (void**)&state->cudaBuff));
|
|
} else {
|
|
#if defined(HIP_UNCACHED_MEMORY)
|
|
#if defined(HIP_CONTIGUOUS_MEMORY)
|
|
NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, state->size,
|
|
cuda ? (rcclParamNetContiguousMem() ? hipDeviceMallocContiguous : hipDeviceMallocUncached) : hipDeviceMallocDefault));
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, state->size,
|
|
cuda ? hipDeviceMallocUncached : hipDeviceMallocDefault));
|
|
#endif
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, state->size,
|
|
cuda ? hipDeviceMallocFinegrained : hipDeviceMallocDefault));
|
|
#endif
|
|
}
|
|
}
|
|
if (!cuda && state->hostBuff == NULL) {
|
|
NCCLCHECK(ncclCudaHostCalloc(&state->hostBuff, state->size));
|
|
}
|
|
if (cpuPtr) *cpuPtr = cuda ? state->cudaBuff : state->hostBuff;
|
|
if (gpuPtr) *gpuPtr = (cpuPtr && sameProcess) ? *cpuPtr : NULL;
|
|
if (ipcDesc) memcpy(ipcDesc, &state->ipcDesc, sizeof(state->ipcDesc));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sharedBuffersGet(struct ncclProxyState* proxyState, int channel, int slot, int* offset, size_t* size) {
|
|
// Use different pools for different channels and also separate send/recv.
|
|
int globalSlot = (channel*NCCL_SHARED_STEPS)+slot;
|
|
*offset = proxyState->p2pChunkSize * globalSlot;
|
|
if (size) *size = proxyState->p2pChunkSize;
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sharedNetBuffersDestroy(struct ncclProxyState* proxyState, int tpLocalRank, int type, struct ncclProxyConnection* connection) {
|
|
if (proxyState->progressState.localPeers == NULL) NCCLCHECK(ncclInternalError);
|
|
struct ncclProxyPeer* peer = proxyState->progressState.localPeers[tpLocalRank];
|
|
if (peer == NULL) NCCLCHECK(ncclInternalError);
|
|
struct ncclProxySharedP2p* state = type == 0 ? &peer->send : &peer->recv;
|
|
if (state->size == 0) NCCLCHECK(ncclInternalError);
|
|
if (ncclAtomicRefCountDecrement(&state->refcount) == 0) {
|
|
if (state->cudaBuff) {
|
|
if (!connection->sameProcess || ncclCuMemEnable()) {
|
|
NCCLCHECK(ncclP2pFreeShareableBuffer(&state->ipcDesc));
|
|
}
|
|
NCCLCHECK(ncclCudaFree(state->cudaBuff));
|
|
}
|
|
if (state->hostBuff) NCCLCHECK(ncclCudaHostFree(state->hostBuff));
|
|
}
|
|
|
|
if (peer->send.refcount || peer->recv.refcount) return ncclSuccess;
|
|
|
|
free(peer);
|
|
proxyState->progressState.localPeers[tpLocalRank] = NULL;
|
|
for (int r = 0; r < proxyState->tpLocalnRanks; r++) {
|
|
if (proxyState->progressState.localPeers[r]) return ncclSuccess;
|
|
}
|
|
// All peers are freed, free array
|
|
free(proxyState->progressState.localPeers);
|
|
proxyState->progressState.localPeers = NULL;
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t proxySharedInit(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, int nChannels) {
|
|
static int is_wsl2 = -1;
|
|
if (is_wsl2 == -1)
|
|
is_wsl2 = (access("/dev/dxg", F_OK) == -1) ? 0 : 1;
|
|
NCCLCHECK(sharedNetBuffersInit(proxyState, is_wsl2 == 0 ? 1 : 0, connection->tpLocalRank, 0, connection->sameProcess, nChannels, NULL, NULL, NULL, NULL));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sendProxySetup(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
struct setupReq* req = (struct setupReq*) reqBuff;
|
|
if (reqSize != sizeof(struct setupReq)) return ncclInternalError;
|
|
|
|
struct sendNetResources* resources;
|
|
NCCLCHECK(ncclCalloc(&resources, 1));
|
|
connection->transportResources = resources;
|
|
|
|
resources->tpRank = req->tpRank;
|
|
resources->tpLocalRank = req->tpLocalRank;
|
|
resources->tpRemoteRank = req->tpRemoteRank;
|
|
resources->netDev = req->netDev;
|
|
resources->shared = connection->shared = req->shared;
|
|
resources->useGdr = req->useGdr;
|
|
resources->channelId = req->channelId;
|
|
resources->connIndex = req->connIndex;
|
|
resources->curr_hdp_reg = req->curr_hdp_reg;
|
|
resources->isP2p = req->isP2p;
|
|
ncclNetProperties_t props;
|
|
NCCLCHECK(proxyState->ncclNet->getProperties(req->netDev, &props));
|
|
/* DMA-BUF support */
|
|
resources->useDmaBuf = resources->useGdr && proxyState->dmaBufSupport && (props.ptrSupport & NCCL_PTR_DMABUF);
|
|
resources->maxRecvs = props.maxRecvs;
|
|
resources->netDeviceVersion = props.netDeviceVersion;
|
|
resources->netDeviceType = props.netDeviceType;
|
|
|
|
/* point-to-point size limits*/
|
|
resources->maxP2pBytes = props.maxP2pBytes;
|
|
if((resources->maxP2pBytes <= 0) || (resources->maxP2pBytes > NCCL_MAX_NET_SIZE_BYTES)) {
|
|
WARN("sendProxySetup: net plugin returned invalid value for maxP2pBytes %ld \
|
|
[allowed range: %ld - %ld] \n", resources->maxP2pBytes, 0L, NCCL_MAX_NET_SIZE_BYTES);
|
|
return ncclInternalError;
|
|
}
|
|
|
|
// We don't return any data
|
|
if (respSize != 0) return ncclInternalError;
|
|
*done = 1;
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t recvProxySetup(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
struct setupReq* req = (struct setupReq*) reqBuff;
|
|
if (reqSize != sizeof(struct setupReq)) return ncclInternalError;
|
|
|
|
struct recvNetResources* resources;
|
|
NCCLCHECK(ncclCalloc(&resources, 1));
|
|
connection->transportResources = resources;
|
|
|
|
resources->tpRank = req->tpRank;
|
|
resources->tpLocalRank = req->tpLocalRank;
|
|
resources->tpRemoteRank = req->tpRemoteRank;
|
|
resources->netDev = req->netDev;
|
|
resources->shared = connection->shared = req->shared;
|
|
resources->useGdr = req->useGdr;
|
|
resources->needFlush = req->needFlush;
|
|
resources->channelId = req->channelId;
|
|
resources->connIndex = req->connIndex;
|
|
resources->curr_hdp_reg = req->curr_hdp_reg;
|
|
ncclNetProperties_t props;
|
|
NCCLCHECK(proxyState->ncclNet->getProperties(req->netDev, &props));
|
|
/* DMA-BUF support */
|
|
resources->useDmaBuf = resources->useGdr && proxyState->dmaBufSupport && (props.ptrSupport & NCCL_PTR_DMABUF);
|
|
resources->maxRecvs = props.maxRecvs;
|
|
resources->netDeviceVersion = props.netDeviceVersion;
|
|
resources->netDeviceType = props.netDeviceType;
|
|
/* point-to-point size limits*/
|
|
resources->maxP2pBytes = props.maxP2pBytes;
|
|
if((resources->maxP2pBytes <= 0) || (resources->maxP2pBytes > NCCL_MAX_NET_SIZE_BYTES)) {
|
|
WARN("recvProxySetup: net plugin returned invalid value for maxP2pBytes %ld \
|
|
[allowed range: %ld - %ld] \n", resources->maxP2pBytes, 0L, NCCL_MAX_NET_SIZE_BYTES);
|
|
return ncclInternalError;
|
|
}
|
|
|
|
if (respSize != sizeof(ncclNetHandle_t)) return ncclInternalError;
|
|
NCCLCHECK(proxyState->ncclNet->listen(req->netDev, respBuff, &resources->netListenComm));
|
|
*done = 1;
|
|
|
|
return ncclSuccess;
|
|
}
|
|
|
|
// This function embeds plugin-specific rules given the current versions
|
|
static ncclResult_t ncclNetGetDeviceHandle(ncclNetDeviceType type, int version, bool isRecv, ncclNetDeviceHandle_t** handle) {
|
|
bool needsDeviceHandle = false;
|
|
|
|
if (type == NCCL_NET_DEVICE_UNPACK) {
|
|
if (version == NCCL_NET_DEVICE_UNPACK_VERSION && isRecv) {
|
|
needsDeviceHandle = true;
|
|
}
|
|
}
|
|
|
|
// Don't re-alloc netDeviceHandles
|
|
if (needsDeviceHandle && (*handle == NULL)) {
|
|
NCCLCHECK(ncclCalloc(handle, 1));
|
|
(*handle)->netDeviceType = type;
|
|
(*handle)->netDeviceVersion = version;
|
|
} else if (!needsDeviceHandle) {
|
|
*handle = NULL;
|
|
}
|
|
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sendProxyConnect(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
ncclNet_ctxt_t ncclNetCtxt = {};
|
|
struct sendNetResources* resources = (struct sendNetResources*)(connection->transportResources);
|
|
ncclNetCommConfig_t commConfig = {0};
|
|
bool rcclAinicRoce = ((rcclParamAinicRoce() == 1) ? true : false);
|
|
if (reqSize != sizeof(netSendConnectArgs)) return ncclInternalError;
|
|
ncclResult_t ret = ncclSuccess;
|
|
netSendConnectArgs* req = (netSendConnectArgs*) reqBuff;
|
|
commConfig.trafficClass = req->trafficClass == NCCL_CONFIG_UNDEF_INT ? NCCL_NET_TRAFFIC_CLASS_UNDEF : req->trafficClass;
|
|
NCCLCHECK(ncclNetGetDeviceHandle(resources->netDeviceType, resources->netDeviceVersion, false /*isRecv*/, &resources->netDeviceHandle));
|
|
|
|
// Only call rcclNetP2pPolicy for ncclNetIb
|
|
if (proxyState->ncclNet == &ncclNetIb) {
|
|
NCCLCHECK(rcclNetP2pPolicy(req->handle, resources->isP2p));
|
|
}
|
|
|
|
if (resources->shared) {
|
|
// Shared buffers
|
|
struct ncclProxyProgressState* progressState = &proxyState->progressState;
|
|
if (progressState->localPeers == NULL) {
|
|
NCCLCHECK(ncclCalloc(&progressState->localPeers, proxyState->tpLocalnRanks));
|
|
}
|
|
struct ncclProxyPeer** localPeers = progressState->localPeers;
|
|
if (localPeers[resources->tpLocalRank] == NULL) {
|
|
NCCLCHECK(ncclCalloc(localPeers + resources->tpLocalRank, 1));
|
|
}
|
|
connection->proxyAppendPtr = localPeers[resources->tpLocalRank]->send.proxyAppend + resources->channelId;
|
|
|
|
if (resources->maxRecvs > 1 && ncclParamNetSharedComms()) {
|
|
// Connect or reuse connection for a netdev/remote rank.
|
|
if (progressState->netComms[resources->netDev] == NULL) {
|
|
NCCLCHECK(ncclCalloc(progressState->netComms + resources->netDev, proxyState->tpnRanks));
|
|
}
|
|
struct ncclSharedNetComms* comms = progressState->netComms[resources->netDev] + resources->tpRemoteRank;
|
|
// let only one localrank connect to a tpRemoteRank to avoid duplicate connections
|
|
if (comms->activeConnect[resources->channelId] == 0)
|
|
comms->activeConnect[resources->channelId] = (resources->tpLocalRank + 1);
|
|
if (comms->sendComm[resources->channelId] == NULL
|
|
&& comms->activeConnect[resources->channelId] == (resources->tpLocalRank + 1)) {
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle,
|
|
comms->sendComm + resources->channelId, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle,
|
|
comms->sendComm + resources->channelId, &resources->netDeviceHandle);
|
|
}
|
|
}
|
|
resources->netSendComm = comms->sendComm[resources->channelId];
|
|
if (comms->sendComm[resources->channelId]) comms->sendRefCount[resources->channelId]++;
|
|
} else {
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle, &resources->netSendComm, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle, &resources->netSendComm, &resources->netDeviceHandle);
|
|
}
|
|
}
|
|
} else {
|
|
// Connect to remote peer
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle, &resources->netSendComm, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle, &resources->netSendComm, &resources->netDeviceHandle);
|
|
}
|
|
connection->proxyAppendPtr = &connection->proxyAppend;
|
|
}
|
|
|
|
NCCLCHECK(ret);
|
|
if (resources->netSendComm == NULL) {
|
|
*done = 0;
|
|
return ncclInProgress;
|
|
}
|
|
*done = 1;
|
|
|
|
if (resources->netDeviceHandle) {
|
|
connection->netDeviceHandle = resources->netDeviceHandle;
|
|
connection->needsProxyProgress = connection->netDeviceHandle->needsProxyProgress;
|
|
} else {
|
|
connection->needsProxyProgress = 1;
|
|
}
|
|
|
|
// Create structures
|
|
struct connectMap* map = &resources->map;
|
|
map->sameProcess = connection->sameProcess;
|
|
map->shared = resources->shared;
|
|
CUDACHECK(cudaGetDevice(&map->cudaDev));
|
|
|
|
if (resources->shared == 0) { // Only allocate dedicated buffers for ring/tree, not for p2p
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, p!= NCCL_PROTO_LL && resources->useGdr ? 1 : 0, proxyState->buffSizes[p], buffs[p]);
|
|
resources->buffSizes[p] = proxyState->buffSizes[p];
|
|
}
|
|
} else {
|
|
// Get shared buffers
|
|
int bank = resources->useGdr ? NCCL_NET_MAP_SHARED_DEVMEM : NCCL_NET_MAP_SHARED_HOSTMEM;
|
|
struct connectMapMem* mapMem = map->mems+bank;
|
|
NCCLCHECK(sharedNetBuffersInit(
|
|
proxyState, resources->useGdr, resources->tpLocalRank, 0, map->sameProcess, proxyState->p2pnChannels,
|
|
&mapMem->gpuPtr, &mapMem->cpuPtr, &mapMem->size, &mapMem->ipcDesc));
|
|
resources->buffSizes[NCCL_PROTO_SIMPLE] = mapMem->size;
|
|
|
|
if (proxyState->allocP2pNetLLBuffers) {
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0 /*p == NCCL_PROTO_LL*/, proxyState->buffSizes[NCCL_PROTO_LL], buffs[NCCL_PROTO_LL]);
|
|
resources->buffSizes[NCCL_PROTO_LL] = proxyState->buffSizes[NCCL_PROTO_LL];
|
|
}
|
|
|
|
NCCL_NET_MAP_ADD_POINTER(map, 1, resources->useGdr ? 1 : 0, mapMem->size, buffs[NCCL_PROTO_SIMPLE]);
|
|
}
|
|
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0, sizeof(struct ncclSendMem), sendMem);
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0, sizeof(struct ncclRecvMem), recvMem);
|
|
|
|
if (map->mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
if (resources->shared == 0) {
|
|
if (!map->sameProcess || ncclCuMemEnable()) {
|
|
ALIGN_SIZE(map->mems[NCCL_NET_MAP_DEVMEM].size, CUDA_IPC_MIN);
|
|
NCCLCHECK(ncclP2pAllocateShareableBuffer(map->mems[NCCL_NET_MAP_DEVMEM].size, 0, &map->mems[NCCL_NET_MAP_DEVMEM].ipcDesc,
|
|
(void**)&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr));
|
|
} else {
|
|
#if defined(HIP_UNCACHED_MEMORY)
|
|
#if defined(HIP_CONTIGUOUS_MEMORY)
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? (rcclParamNetContiguousMem() ? hipDeviceMallocContiguous : hipDeviceMallocUncached) : hipDeviceMallocDefault));
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? hipDeviceMallocUncached : hipDeviceMallocDefault));
|
|
#endif
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? hipDeviceMallocFinegrained : hipDeviceMallocDefault));
|
|
#endif
|
|
}
|
|
map->mems[NCCL_NET_MAP_DEVMEM].cpuPtr = map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr;
|
|
}
|
|
}
|
|
if (map->sameProcess) {
|
|
NCCLCHECK(ncclCudaHostCalloc(&map->mems[NCCL_NET_MAP_HOSTMEM].cpuPtr, map->mems[NCCL_NET_MAP_HOSTMEM].size));
|
|
map->mems[NCCL_NET_MAP_HOSTMEM].gpuPtr = map->mems[NCCL_NET_MAP_HOSTMEM].cpuPtr;
|
|
} else {
|
|
NCCLCHECK(netCreateShm(proxyState, map->mems+NCCL_NET_MAP_HOSTMEM));
|
|
void* sendMem = (void*)NCCL_NET_MAP_GET_POINTER(map, cpu, sendMem);
|
|
void* recvMem = (void*)NCCL_NET_MAP_GET_POINTER(map, cpu, recvMem);
|
|
memset(sendMem, 0, sizeof(struct ncclSendMem));
|
|
memset(recvMem, 0, sizeof(struct ncclRecvMem));
|
|
}
|
|
if (ncclGdrCopy && map->sameProcess && ncclParamGdrCopySyncEnable()) {
|
|
uint64_t *cpuPtr, *gpuPtr;
|
|
NCCLCHECK(ncclGdrCudaCalloc(&cpuPtr, &gpuPtr, 1, &resources->gdrDesc));
|
|
|
|
resources->gdcSync = cpuPtr;
|
|
struct connectMapMem* gdcMem = map->mems+NCCL_NET_MAP_GDCMEM;
|
|
gdcMem->cpuPtr = (char*)cpuPtr;
|
|
gdcMem->gpuPtr = (char*)gpuPtr;
|
|
gdcMem->size = sizeof(uint64_t); // sendMem->head
|
|
}
|
|
|
|
resources->sendMem = (struct ncclSendMem*) NCCL_NET_MAP_GET_POINTER(map, cpu, sendMem);
|
|
resources->recvMem = (struct ncclRecvMem*) NCCL_NET_MAP_GET_POINTER(map, cpu, recvMem);
|
|
|
|
// Don't give credits yet in shared mode.
|
|
(resources->gdcSync ? *resources->gdcSync : resources->sendMem->head) =
|
|
(map->shared ? -NCCL_STEPS : 0);
|
|
for (int i=0; i<NCCL_STEPS; i++) resources->recvMem->connFifo[i].size = -1;
|
|
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
resources->buffers[p] = NCCL_NET_MAP_GET_POINTER(map, cpu, buffs[p]);
|
|
if (resources->buffers[p]) {
|
|
#if CUDA_VERSION >= 11070
|
|
/* DMA-BUF support */
|
|
int type = NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST;
|
|
if (type == NCCL_PTR_CUDA && resources->useDmaBuf) {
|
|
int dmabuf_fd;
|
|
CUCHECK(cuMemGetHandleForAddressRange((void *)&dmabuf_fd, (CUdeviceptr)resources->buffers[p], resources->buffSizes[p], CU_MEM_RANGE_HANDLE_TYPE_DMA_BUF_FD, getHandleForAddressRangeFlags(resources->useGdr)));
|
|
NCCLCHECK(proxyState->ncclNet->regMrDmaBuf(resources->netSendComm, resources->buffers[p], resources->buffSizes[p], type, 0ULL, dmabuf_fd, &resources->mhandles[p]));
|
|
(void)close(dmabuf_fd);
|
|
} else // FALL-THROUGH to nv_peermem GDR path
|
|
#else
|
|
/* DMA-BUF support */
|
|
int type = NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST;
|
|
if (type == NCCL_PTR_CUDA && proxyState->dmaBufSupport && pfn_hsa_amd_portable_export_dmabuf) {
|
|
int dmabuf_fd;
|
|
uint64_t offset;
|
|
HSACHECK(hsa_amd_portable_export_dmabuf((const void*)resources->buffers[p], resources->buffSizes[p], &dmabuf_fd, &offset));
|
|
NCCLCHECK(proxyState->ncclNet->regMrDmaBuf(resources->netSendComm, resources->buffers[p], resources->buffSizes[p], type, offset, dmabuf_fd, &resources->mhandles[p]));
|
|
(void)close(dmabuf_fd);
|
|
TRACE(NCCL_INIT|NCCL_NET, "hsa_amd_portable_export_dmabuf buffer %p size %d handle %x offset %ld",
|
|
(const void*)resources->buffers[p], resources->buffSizes[p], dmabuf_fd, offset);
|
|
} else // FALL-THROUGH to nv_peermem GDR path
|
|
#endif
|
|
{
|
|
NCCLCHECK(proxyState->ncclNet->regMr(resources->netSendComm, resources->buffers[p], resources->buffSizes[p], NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST, &resources->mhandles[p]));
|
|
}
|
|
|
|
// Copy the mhandle dptr, if implemented
|
|
if (resources->netDeviceHandle && proxyState->ncclNet->getDeviceMr)
|
|
NCCLCHECK(proxyState->ncclNet->getDeviceMr(resources->netSendComm, resources->mhandles[p], &connection->mhandles[p]));
|
|
}
|
|
}
|
|
|
|
//NCCLCHECK(netDumpMap(map));
|
|
if (respSize != sizeof(struct connectMap)) return ncclInternalError;
|
|
memcpy(respBuff, map, sizeof(struct connectMap));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t recvProxyConnect(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
if (reqSize != sizeof(netRecvConnectArgs)) return ncclInternalError;
|
|
struct recvNetResources* resources = (struct recvNetResources*)(connection->transportResources);
|
|
netRecvConnectArgs* req = (netRecvConnectArgs*) reqBuff;
|
|
resources->tpRemoteProxyRank = req->proxyRank;
|
|
ncclResult_t ret = ncclSuccess;
|
|
ncclNet_ctxt_t ncclNetCtxt = {};
|
|
bool rcclAinicRoce = ((rcclParamAinicRoce() == 1) ? true : false);
|
|
|
|
NCCLCHECK(ncclNetGetDeviceHandle(resources->netDeviceType, resources->netDeviceVersion, true /*isRecv*/, &resources->netDeviceHandle));
|
|
// Finish connection establishment from remote peer
|
|
if (resources->shared) {
|
|
// Shared buffers
|
|
struct ncclProxyProgressState* progressState = &proxyState->progressState;
|
|
if (progressState->localPeers == NULL) {
|
|
NCCLCHECK(ncclCalloc(&progressState->localPeers, proxyState->tpLocalnRanks));
|
|
}
|
|
struct ncclProxyPeer** localPeers = progressState->localPeers;
|
|
if (localPeers[resources->tpLocalRank] == NULL) {
|
|
NCCLCHECK(ncclCalloc(localPeers + resources->tpLocalRank, 1));
|
|
}
|
|
connection->proxyAppendPtr = localPeers[resources->tpLocalRank]->recv.proxyAppend + resources->channelId;
|
|
|
|
if (resources->maxRecvs > 1 && ncclParamNetSharedComms()) {
|
|
// Connect or reuse connection for a netdev/remote rank.
|
|
if (progressState->netComms[resources->netDev] == NULL) {
|
|
NCCLCHECK(ncclCalloc(progressState->netComms + resources->netDev, proxyState->tpnRanks));
|
|
}
|
|
struct ncclSharedNetComms* comms = progressState->netComms[resources->netDev] + resources->tpRemoteProxyRank;
|
|
// reuse handle to for netdev/remote rank to avoid duplicate connections
|
|
if (comms->activeAccept[resources->channelId] == 0)
|
|
comms->activeAccept[resources->channelId] = (resources->tpLocalRank + 1);
|
|
//try connecting while comm is null
|
|
if (comms->recvComm[resources->channelId] == NULL
|
|
&& comms->activeAccept[resources->channelId] == (resources->tpLocalRank + 1)) {
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm,
|
|
comms->recvComm+resources->channelId, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm,
|
|
comms->recvComm+resources->channelId, &resources->netDeviceHandle);
|
|
}
|
|
}
|
|
resources->netRecvComm = comms->recvComm[resources->channelId];
|
|
if (comms->recvComm[resources->channelId]) comms->recvRefCount[resources->channelId]++;
|
|
} else {
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm, &resources->netRecvComm, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm, &resources->netRecvComm, &resources->netDeviceHandle);
|
|
}
|
|
}
|
|
} else {
|
|
// Connect to remote peer
|
|
if (rcclAinicRoce) {
|
|
ncclNetCtxt.chId = resources->channelId;
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm, &resources->netRecvComm, (ncclNetDeviceHandle_t **)&ncclNetCtxt);
|
|
} else {
|
|
ret = proxyState->ncclNet->accept(resources->netListenComm, &resources->netRecvComm, &resources->netDeviceHandle);
|
|
}
|
|
connection->proxyAppendPtr = &connection->proxyAppend;
|
|
}
|
|
|
|
NCCLCHECK(ret);
|
|
if (resources->netRecvComm == NULL) {
|
|
*done = 0;
|
|
return ncclInProgress;
|
|
}
|
|
*done = 1;
|
|
|
|
if (resources->netDeviceHandle) {
|
|
connection->netDeviceHandle = resources->netDeviceHandle;
|
|
connection->needsProxyProgress = connection->netDeviceHandle->needsProxyProgress;
|
|
} else {
|
|
connection->needsProxyProgress = 1;
|
|
}
|
|
|
|
NCCLCHECK(proxyState->ncclNet->closeListen(resources->netListenComm));
|
|
|
|
// Create structures
|
|
struct connectMap* map = &resources->map;
|
|
map->sameProcess = connection->sameProcess;
|
|
if (map->sameProcess == 0) return ncclInternalError; // We don't support remote proxy for recv
|
|
map->shared = resources->shared;
|
|
|
|
if (resources->shared == 0) { // Only allocate dedicated buffers for ring/tree, not for p2p
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, resources->useGdr ? 1 : 0, proxyState->buffSizes[p], buffs[p]);
|
|
resources->buffSizes[p] = proxyState->buffSizes[p];
|
|
}
|
|
} else {
|
|
// Get shared buffers
|
|
int bank = resources->useGdr ? NCCL_NET_MAP_SHARED_DEVMEM : NCCL_NET_MAP_SHARED_HOSTMEM;
|
|
struct connectMapMem* mapMem = map->mems+bank;
|
|
NCCLCHECK(sharedNetBuffersInit(
|
|
proxyState, resources->useGdr, resources->tpLocalRank, 1, 1, proxyState->p2pnChannels,
|
|
&mapMem->gpuPtr, &mapMem->cpuPtr, &mapMem->size, NULL));
|
|
resources->buffSizes[NCCL_PROTO_SIMPLE] = mapMem->size;
|
|
NCCL_NET_MAP_ADD_POINTER(map, 1, resources->useGdr ? 1 : 0, mapMem->size, buffs[NCCL_PROTO_SIMPLE]);
|
|
}
|
|
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0, sizeof(struct ncclSendMem), sendMem);
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0, sizeof(struct ncclRecvMem), recvMem);
|
|
|
|
if (proxyState->allocP2pNetLLBuffers) {
|
|
NCCL_NET_MAP_ADD_POINTER(map, 0, 0 /*devMem*/, proxyState->buffSizes[NCCL_PROTO_LL], buffs[NCCL_PROTO_LL]);
|
|
resources->buffSizes[NCCL_PROTO_LL] = proxyState->buffSizes[NCCL_PROTO_LL];
|
|
}
|
|
|
|
if (map->mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
if (resources->shared == 0) {
|
|
if (ncclCuMemEnable()) {
|
|
NCCLCHECK(ncclP2pAllocateShareableBuffer(map->mems[NCCL_NET_MAP_DEVMEM].size, 0, &map->mems[NCCL_NET_MAP_DEVMEM].ipcDesc,
|
|
(void**)&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr));
|
|
} else {
|
|
#if defined(HIP_UNCACHED_MEMORY)
|
|
#if defined(HIP_CONTIGUOUS_MEMORY)
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? (rcclParamNetContiguousMem() ? hipDeviceMallocContiguous : hipDeviceMallocUncached) : hipDeviceMallocDefault));
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? hipDeviceMallocUncached : hipDeviceMallocDefault));
|
|
#endif
|
|
#else
|
|
NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size,
|
|
resources->useGdr ? hipDeviceMallocFinegrained : hipDeviceMallocDefault));
|
|
#endif
|
|
}
|
|
map->mems[NCCL_NET_MAP_DEVMEM].cpuPtr = map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr;
|
|
}
|
|
}
|
|
NCCLCHECK(ncclCudaHostCalloc(&map->mems[NCCL_NET_MAP_HOSTMEM].cpuPtr, map->mems[NCCL_NET_MAP_HOSTMEM].size));
|
|
map->mems[NCCL_NET_MAP_HOSTMEM].gpuPtr = map->mems[NCCL_NET_MAP_HOSTMEM].cpuPtr;
|
|
if (ncclGdrCopy && map->sameProcess) {
|
|
uint64_t *cpuPtr, *gpuPtr;
|
|
NCCLCHECK(ncclGdrCudaCalloc(&cpuPtr, &gpuPtr, 2, &resources->gdrDesc));
|
|
|
|
if (ncclParamGdrCopySyncEnable()) {
|
|
resources->gdcSync = cpuPtr;
|
|
struct connectMapMem* gdcMem = map->mems+NCCL_NET_MAP_GDCMEM;
|
|
gdcMem->cpuPtr = (char*)cpuPtr;
|
|
gdcMem->gpuPtr = (char*)gpuPtr;
|
|
gdcMem->size = sizeof(uint64_t);
|
|
}
|
|
if (ncclParamGdrCopyFlushEnable()) resources->gdcFlush = cpuPtr + 1;
|
|
}
|
|
|
|
resources->sendMem = (struct ncclSendMem*) NCCL_NET_MAP_GET_POINTER(map, cpu, sendMem);
|
|
resources->recvMem = (struct ncclRecvMem*) NCCL_NET_MAP_GET_POINTER(map, cpu, recvMem);
|
|
for (int i = 0; i < NCCL_STEPS; i++) resources->recvMem->connFifo[i].size = -1;
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
resources->buffers[p] = NCCL_NET_MAP_GET_POINTER(map, cpu, buffs[p]);
|
|
if (resources->buffers[p]) {
|
|
#if CUDA_VERSION >= 11070
|
|
/* DMA-BUF support */
|
|
int type = NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST;
|
|
if (type == NCCL_PTR_CUDA && resources->useDmaBuf) {
|
|
int dmabuf_fd;
|
|
CUCHECK(cuMemGetHandleForAddressRange((void *)&dmabuf_fd, (CUdeviceptr)resources->buffers[p], resources->buffSizes[p], CU_MEM_RANGE_HANDLE_TYPE_DMA_BUF_FD, getHandleForAddressRangeFlags(resources->useGdr)));
|
|
NCCLCHECK(proxyState->ncclNet->regMrDmaBuf(resources->netRecvComm, resources->buffers[p], resources->buffSizes[p], type, 0ULL, dmabuf_fd, &resources->mhandles[p]));
|
|
(void)close(dmabuf_fd);
|
|
} else // FALL-THROUGH to nv_peermem GDR path
|
|
#else
|
|
/* DMA-BUF support */
|
|
int type = NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST;
|
|
if (type == NCCL_PTR_CUDA && proxyState->dmaBufSupport && pfn_hsa_amd_portable_export_dmabuf) {
|
|
int dmabuf_fd;
|
|
uint64_t offset;
|
|
HSACHECK(hsa_amd_portable_export_dmabuf((const void*)resources->buffers[p], resources->buffSizes[p], &dmabuf_fd, &offset));
|
|
NCCLCHECK(proxyState->ncclNet->regMrDmaBuf(resources->netRecvComm, resources->buffers[p], resources->buffSizes[p], type, offset, dmabuf_fd, &resources->mhandles[p]));
|
|
(void)close(dmabuf_fd);
|
|
TRACE(NCCL_INIT|NCCL_NET, "hsa_amd_portable_export_dmabuf buffer %p size %d handle %x offset %ld",
|
|
(const void*)resources->buffers[p], resources->buffSizes[p], dmabuf_fd, offset);
|
|
} else // FALL-THROUGH to nv_peermem GDR path
|
|
#endif
|
|
{
|
|
NCCLCHECK(proxyState->ncclNet->regMr(resources->netRecvComm, resources->buffers[p], resources->buffSizes[p], NCCL_NET_MAP_DEV_MEM(map, buffs[p]) ? NCCL_PTR_CUDA : NCCL_PTR_HOST, &resources->mhandles[p]));
|
|
}
|
|
|
|
// Copy the mhandle dptr
|
|
if (resources->netDeviceType != NCCL_NET_DEVICE_HOST && proxyState->ncclNet->getDeviceMr)
|
|
NCCLCHECK(proxyState->ncclNet->getDeviceMr(resources->netRecvComm, resources->mhandles[p], &connection->mhandles[p]));
|
|
}
|
|
}
|
|
|
|
//NCCLCHECK(netDumpMap(map));
|
|
if (respSize != sizeof(struct connectMap)) return ncclInternalError;
|
|
memcpy(respBuff, map, sizeof(struct connectMap));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t sendProxyFree(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState) {
|
|
struct sendNetResources* resources = (struct sendNetResources*)(connection->transportResources);
|
|
if (connection->state == connSharedInitialized) { // NVB Preconnect
|
|
NCCLCHECK(sharedNetBuffersDestroy(proxyState, connection->tpLocalRank, 0, connection));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
if (connection->state == connConnected) {
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
if (resources->buffers[p]) {
|
|
NCCLCHECK(proxyState->ncclNet->deregMr(resources->netSendComm, resources->mhandles[p]));
|
|
}
|
|
}
|
|
struct connectMapMem* mems = resources->map.mems;
|
|
if (resources->map.sameProcess) {
|
|
NCCLCHECK(ncclCudaHostFree(mems[NCCL_NET_MAP_HOSTMEM].cpuPtr));
|
|
} else {
|
|
NCCLCHECK(ncclShmIpcClose(&mems[NCCL_NET_MAP_HOSTMEM].createDesc));
|
|
}
|
|
NCCLCHECK(ncclCudaFree(mems[NCCL_NET_MAP_DEVMEM].cpuPtr));
|
|
if (!resources->map.sameProcess || ncclCuMemEnable()) {
|
|
// cuMem API support
|
|
if (mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
NCCLCHECK(ncclP2pFreeShareableBuffer(&mems[NCCL_NET_MAP_DEVMEM].ipcDesc));
|
|
}
|
|
}
|
|
if (mems[NCCL_NET_MAP_GDCMEM].cpuPtr) NCCLCHECK(ncclGdrCudaFree(resources->gdrDesc));
|
|
if (resources->shared) {
|
|
NCCLCHECK(sharedNetBuffersDestroy(proxyState, resources->tpLocalRank, 0, connection));
|
|
if (resources->maxRecvs > 1 && ncclParamNetSharedComms()) {
|
|
struct ncclSharedNetComms* comms = proxyState->progressState.netComms[resources->netDev]+resources->tpRemoteRank;
|
|
comms->sendRefCount[resources->channelId]--;
|
|
if (comms->sendRefCount[resources->channelId] == 0) NCCLCHECK(proxyState->ncclNet->closeSend(comms->sendComm[resources->channelId]));
|
|
} else {
|
|
NCCLCHECK(proxyState->ncclNet->closeSend(resources->netSendComm));
|
|
}
|
|
} else {
|
|
NCCLCHECK(proxyState->ncclNet->closeSend(resources->netSendComm));
|
|
}
|
|
}
|
|
|
|
if (resources) free(resources);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t recvProxyFree(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState) {
|
|
struct recvNetResources* resources = (struct recvNetResources*)(connection->transportResources);
|
|
if (connection->state == connSharedInitialized) { // NVB Preconnect
|
|
NCCLCHECK(sharedNetBuffersDestroy(proxyState, connection->tpLocalRank, 1, connection));
|
|
return ncclSuccess;
|
|
}
|
|
|
|
if (connection->state == connConnected) {
|
|
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
|
|
if (resources->buffers[p]) {
|
|
NCCLCHECK(proxyState->ncclNet->deregMr(resources->netRecvComm, resources->mhandles[p]));
|
|
}
|
|
}
|
|
struct connectMapMem* mems = resources->map.mems;
|
|
NCCLCHECK(ncclCudaHostFree(mems[NCCL_NET_MAP_HOSTMEM].cpuPtr));
|
|
NCCLCHECK(ncclCudaFree(mems[NCCL_NET_MAP_DEVMEM].cpuPtr));
|
|
if (!resources->map.sameProcess || ncclCuMemEnable()) {
|
|
// cuMem API support
|
|
if (mems[NCCL_NET_MAP_DEVMEM].size) {
|
|
NCCLCHECK(ncclP2pFreeShareableBuffer(&mems[NCCL_NET_MAP_DEVMEM].ipcDesc));
|
|
}
|
|
}
|
|
if (mems[NCCL_NET_MAP_GDCMEM].cpuPtr) NCCLCHECK(ncclGdrCudaFree(resources->gdrDesc));
|
|
if (resources->shared) {
|
|
NCCLCHECK(sharedNetBuffersDestroy(proxyState, resources->tpLocalRank, 1, connection));
|
|
if (resources->maxRecvs > 1 && ncclParamNetSharedComms()) {
|
|
struct ncclSharedNetComms* comms = proxyState->progressState.netComms[resources->netDev] + resources->tpRemoteProxyRank;
|
|
comms->recvRefCount[resources->channelId]--;
|
|
if (comms->recvRefCount[resources->channelId] == 0) NCCLCHECK(proxyState->ncclNet->closeRecv(comms->recvComm[resources->channelId]));
|
|
} else {
|
|
NCCLCHECK(proxyState->ncclNet->closeRecv(resources->netRecvComm));
|
|
}
|
|
} else {
|
|
NCCLCHECK(proxyState->ncclNet->closeRecv(resources->netRecvComm));
|
|
}
|
|
}
|
|
|
|
if (resources) free(resources);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static_assert(NCCL_STEPS <= NCCL_NET_MAX_REQUESTS, "Not enough net requests to cover for steps");
|
|
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
static int g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
|
|
static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct ncclProxyArgs* args) {
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt++;
|
|
#endif
|
|
if (args->state == ncclProxyOpReady) {
|
|
for (int s=0; s<args->nsubs; s++) {
|
|
struct ncclProxySubArgs* sub = args->subs+s;
|
|
struct sendNetResources* resources = (struct sendNetResources*) (sub->connection->transportResources);
|
|
// Round to next multiple of sliceSteps
|
|
sub->base = ROUNDUP(resources->step, args->chunkSteps);
|
|
// Set step base for next op
|
|
resources->step = sub->base + sub->nsteps;
|
|
sub->posted = sub->transmitted = sub->done = 0;
|
|
ncclProfilerRecordProxyOpEventState(s, args, ncclProfilerProxyOpInProgress_v4);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->addNewProxyOp(
|
|
sub->traceKey,
|
|
sub->traceInfo,
|
|
facebook_rccl::ProxyOpType::SEND,
|
|
sub->channelId,
|
|
sub->nsteps,
|
|
sub->nbytes,
|
|
sub->peer);
|
|
}
|
|
if (!sub->reg)
|
|
sub->sendMhandle = resources->mhandles[args->protocol];
|
|
}
|
|
args->state = ncclProxyOpProgress;
|
|
args->hdp_flushed = 0;
|
|
}
|
|
args->idle = 1;
|
|
if (args->state == ncclProxyOpProgress) {
|
|
int p = args->protocol;
|
|
int maxDepth = std::min(NCCL_STEPS, NCCL_SHARED_STEPS/args->nsubs);
|
|
for (int s=0; s<args->nsubs; s++) {
|
|
struct ncclProxySubArgs* sub = args->subs+s;
|
|
int postedStepId = sub->posted;
|
|
int transmittedStepId = sub->transmitted;
|
|
int doneStepId = sub->done;
|
|
if (sub->done == sub->nsteps) continue;
|
|
struct sendNetResources* resources = (struct sendNetResources*) (sub->connection->transportResources);
|
|
volatile struct ncclConnFifo* connFifo = (volatile struct ncclConnFifo*)resources->recvMem->connFifo;
|
|
int stepSize = resources->buffSizes[p] / NCCL_STEPS;
|
|
char* localBuff = NCCL_NET_MAP_GET_POINTER(&resources->map, cpu, buffs[p]);
|
|
// Post buffers to the GPU
|
|
if (sub->posted < sub->nsteps && sub->posted < sub->done + maxDepth) {
|
|
ncclProfilerStartSendProxyStepEvent(s, args, postedStepId);
|
|
int buffSlot = (sub->base+sub->posted)%NCCL_STEPS;
|
|
if (resources->shared) {
|
|
if (!sub->reg) {
|
|
int sharedBuffSlot = sub->posted%maxDepth;
|
|
int offset;
|
|
NCCLCHECK(sharedBuffersGet(proxyState, sub->channelId, sharedBuffSlot*args->nsubs+s, &offset, NULL));
|
|
__atomic_store_n(&resources->recvMem->connFifo[buffSlot].offset, offset, __ATOMIC_RELAXED);
|
|
__sync_synchronize();
|
|
}
|
|
volatile uint64_t* sendHead = resources->gdcSync ? resources->gdcSync : &resources->sendMem->head;
|
|
sub->posted += args->sliceSteps;
|
|
*sendHead = sub->base + sub->posted - NCCL_STEPS;
|
|
if (resources->gdcSync) wc_store_fence(); // Flush out WC write
|
|
} else {
|
|
sub->posted += args->sliceSteps;
|
|
}
|
|
ncclProfilerRecordProxyStepEventState(s, args, postedStepId, ncclProfilerProxyStepSendGPUWait);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::POSTED, sub->posted);
|
|
proxyState->proxyTrace->setProxyOpTimestamp(sub->traceKey, facebook_rccl::ProxyCounterTypes::POSTED);
|
|
}
|
|
args->idle = 0;
|
|
continue;
|
|
}
|
|
// Check whether we received data from the GPU and send it to the network
|
|
if (sub->transmitted < sub->posted && sub->transmitted < sub->done + NCCL_STEPS) {
|
|
int buffSlot = (sub->base+sub->transmitted)%NCCL_STEPS;
|
|
volatile uint64_t* recvTail = &resources->recvMem->tail;
|
|
uint64_t tail = sub->base + sub->transmitted;
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::RECV_TAIL, *recvTail);
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::TAIL_OR_HEAD, tail);
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::FIFO_SZ_OR_HEAD_CACHE, connFifo[buffSlot].size);
|
|
}
|
|
|
|
if (connFifo[buffSlot].size != -1 && (*recvTail > tail || p == NCCL_PROTO_LL)) {
|
|
// We have something to receive, let's check if it's completely ready.
|
|
int size = connFifo[buffSlot].size;
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_SEND_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_SEND_EXIT)
|
|
sub->npKitSizesFifo[buffSlot] = size;
|
|
#endif
|
|
bool shared = (p == NCCL_PROTO_SIMPLE) && resources->shared;
|
|
char* buff = shared ? localBuff+connFifo[buffSlot].offset : localBuff+buffSlot*stepSize;
|
|
int ready = 1;
|
|
if (p == NCCL_PROTO_LL128) {
|
|
ready = resources->useGdr;
|
|
if (!ready) {
|
|
// When data is in sysmem, we need to wait until all flags are correct since the GPU only
|
|
// called threadfence()
|
|
uint64_t flag = sub->base+sub->transmitted+1;
|
|
int nFifoLines = DIVUP(connFifo[buffSlot].size, sizeof(uint64_t)*NCCL_LL128_LINEELEMS);
|
|
volatile uint64_t* lines = (volatile uint64_t*)buff;
|
|
ready = 1;
|
|
for (int i=0; i<nFifoLines; i++) {
|
|
if (lines[i*NCCL_LL128_LINEELEMS+NCCL_LL128_DATAELEMS] != flag) { ready = 0; break; }
|
|
}
|
|
}
|
|
} else if (p == NCCL_PROTO_LL) {
|
|
uint32_t flag = NCCL_LL_FLAG(sub->base+sub->transmitted+1);
|
|
int nFifoLines = DIVUP(size, sizeof(union ncclLLFifoLine));
|
|
union ncclLLFifoLine* lines = (union ncclLLFifoLine*)buff;
|
|
for (int i=0; i<nFifoLines; i++) {
|
|
volatile uint32_t *f1 = &lines[i].flag1;
|
|
volatile uint32_t *f2 = &lines[i].flag2;
|
|
if (f1[0] != flag || f2[0] != flag) { ready = 0; break; }
|
|
}
|
|
} else if (p == NCCL_PROTO_SIMPLE) {
|
|
if (resources->shared) {
|
|
buff = sub->reg ? (char*)sub->sendbuff + sub->transmitted * NCCL_MAX_NET_SIZE : localBuff + resources->recvMem->connFifo[buffSlot].offset;
|
|
} else if (sub->reg) {
|
|
size_t sendSize;
|
|
sub->ringAlgo->getNextSendAddr(sub->transmitted, (uint8_t**)&buff, &sendSize, &sub->sendMhandle);
|
|
assert(sendSize == size);
|
|
}
|
|
}
|
|
if (ready) {
|
|
// flush HDP if not done
|
|
if (resources->curr_hdp_reg && args->hdp_flushed < *recvTail) {
|
|
args->hdp_flushed = *recvTail;
|
|
*resources->curr_hdp_reg = 1;
|
|
}
|
|
ncclProfilerRecordProxyStepEventState(s, args, transmittedStepId, ncclProfilerProxyStepSendPeerWait_v4);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::KERNEL_COPY_READY, sub->reg ? 1: sub->transmitted + args->sliceSteps);
|
|
}
|
|
// Data is ready, try to send.
|
|
// Coverity complains about the size here as pointing to an out-of-scope temporary. Which is nonsense,
|
|
// since size is a plain integer.
|
|
// coverity[use_invalid:FALSE]
|
|
void* phandle = &sub->pHandles[DIVUP(transmittedStepId, args->sliceSteps)%NCCL_STEPS];
|
|
void **requestPtr = sub->requests+buffSlot;
|
|
// for LL/LL128 protocols, completion event for write operation is not needed on the receiver side as
|
|
// the LL flags are actively polled to detect if full data is received or not, so this hint can be used
|
|
// by network plugin to optimize the transport for LL/LL128
|
|
bool ignoreCompletion = ncclParamNetOptionalRecvCompletion() && ((args->protocol == NCCL_PROTO_LL128) || (args->protocol == NCCL_PROTO_LL));
|
|
if (ignoreCompletion) *requestPtr = (void *)NCCL_NET_OPTIONAL_RECV_COMPLETION;
|
|
NCCLCHECK(proxyState->ncclNet->isend(resources->netSendComm, buff, size, resources->tpRank, sub->sendMhandle, phandle, requestPtr));
|
|
if (*requestPtr != NULL) {
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->setProxyOpTimestamp(sub->traceKey, facebook_rccl::ProxyCounterTypes::KERNEL_COPY_READY);
|
|
}
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_SEND_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_SEND_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_SEND_ENTRY,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
size,
|
|
#endif
|
|
uint64_t(sub->requests+buffSlot)/sizeof(void*),
|
|
*(volatile uint64_t*)NpKit::GetCpuTimestamp(), sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
sub->timestamp[buffSlot] = 0;
|
|
#endif
|
|
TRACE(NCCL_NET, "sendProxy [%ld/%d/%d] Isend posted, req %p, buff %p, size %d, proto %d, myRank %d, channelId %d, mhandle %p", sub->transmitted, buffSlot, sub->nsteps, sub->requests[buffSlot], buff, size, p, proxyState->tpRank, sub->channelId, sub->sendMhandle);
|
|
sub->transSize = size;
|
|
sub->transmitted += args->sliceSteps;
|
|
ncclProfilerRecordProxyStepEventState(s, args, transmittedStepId, ncclProfilerProxyStepSendWait);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::TRANSMITTED, sub->transmitted);
|
|
}
|
|
args->idle = 0;
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// Check whether the network has completed some send operations.
|
|
if (sub->done < sub->transmitted) {
|
|
int done;
|
|
int size;
|
|
int buffSlot = (sub->base+sub->done)%NCCL_STEPS;
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_SEND_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_SEND_EXIT)
|
|
if (sub->timestamp[buffSlot] == 0)
|
|
sub->timestamp[buffSlot] = *(volatile uint64_t*)NpKit::GetCpuTimestamp();
|
|
#endif
|
|
NCCLCHECK(proxyState->ncclNet->test(sub->requests[buffSlot], &done, &size));
|
|
if (done) {
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_SEND_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_SEND_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_SEND_EXIT,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
sub->npKitSizesFifo[buffSlot],
|
|
#endif
|
|
uint64_t(sub->requests+buffSlot)/sizeof(void*),
|
|
sub->timestamp[buffSlot], sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
#endif
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_TEST_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_TEST_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_TEST_ENTRY,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
sub->npKitSizesFifo[buffSlot],
|
|
#endif
|
|
uint64_t(sub->requests+buffSlot)/sizeof(void*),
|
|
sub->timestamp[buffSlot], sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
#endif
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_TEST_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_TEST_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_TEST_EXIT,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
sub->npKitSizesFifo[buffSlot],
|
|
#endif
|
|
uint64_t(sub->requests+buffSlot)/sizeof(void*),
|
|
*(volatile uint64_t*)NpKit::GetCpuTimestamp(), sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
#endif
|
|
// Make sure size is reset to -1 before we update the head.
|
|
connFifo[buffSlot].size = -1;
|
|
__sync_synchronize();
|
|
TRACE(NCCL_NET, "sendProxy [%ld/%d/%d] request %p done", sub->done, buffSlot, sub->nsteps, sub->requests[buffSlot]);
|
|
sub->done += args->sliceSteps;
|
|
ncclProfilerStopProxyStepEvent(s, args, doneStepId);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::DONE, sub->done);
|
|
}
|
|
if (resources->shared == 0) {
|
|
volatile uint64_t* sendHead = resources->gdcSync ? resources->gdcSync : &resources->sendMem->head;
|
|
*sendHead = sub->base + sub->done;
|
|
if (resources->gdcSync) wc_store_fence(); // Flush out WC write
|
|
}
|
|
args->idle = 0;
|
|
if (sub->done == sub->nsteps) {
|
|
args->done++;
|
|
if (sub->ringAlgo && sub->ringAlgo->decRefCount() == 0) delete sub->ringAlgo;
|
|
sub->ringAlgo = NULL;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (args->done == args->nsubs) {
|
|
for (int s=0; s<args->nsubs; s++) {
|
|
ncclProfilerStopProxyOpEvent(s, args);
|
|
}
|
|
args->state = ncclProxyOpNone;
|
|
}
|
|
}
|
|
return ncclSuccess;
|
|
}
|
|
|
|
RCCL_PARAM(NetHdpFlush, "NET_HDP_FLUSH", 0);
|
|
|
|
static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct ncclProxyArgs* args) {
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt++;
|
|
#endif
|
|
if (args->state == ncclProxyOpReady) {
|
|
// Initialize subs and group them by same recvComm.
|
|
void* recvComm;
|
|
int groupSize = 0;
|
|
int maxRecvs = 1;
|
|
for (int s=0; s<args->nsubs; s++) {
|
|
struct ncclProxySubArgs* sub = args->subs+s;
|
|
if (groupSize == maxRecvs) {
|
|
groupSize = 0;
|
|
} else if (s>0) { // Find next sub with the same recvComm
|
|
int next;
|
|
for (next=s; next<args->nsubs; next++) {
|
|
struct recvNetResources* nextRes = (struct recvNetResources*) (args->subs[next].connection->transportResources);
|
|
if (nextRes->netRecvComm == recvComm) break;
|
|
}
|
|
if (next == args->nsubs) { // Not found
|
|
groupSize = 0;
|
|
} else if (s != next) { // We found a sub later with the same recvComm ; swap subs
|
|
struct ncclProxySubArgs temp;
|
|
memcpy(&temp, sub, sizeof(struct ncclProxySubArgs));
|
|
memcpy(sub, args->subs+next, sizeof(struct ncclProxySubArgs));
|
|
memcpy(args->subs+next, &temp, sizeof(struct ncclProxySubArgs));
|
|
}
|
|
}
|
|
groupSize++;
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
maxRecvs = resources->maxRecvs;
|
|
recvComm = resources->netRecvComm;
|
|
// Round to next multiple of sliceSteps
|
|
sub->base = ROUNDUP(resources->step, args->chunkSteps);
|
|
// Set step base for next op
|
|
resources->step = sub->base + sub->nsteps;
|
|
sub->posted = sub->received = sub->transmitted = sub->done = 0;
|
|
sub->regBufferReady = 0;
|
|
for (int i=0; i<groupSize; i++) sub[-i].groupSize = groupSize;
|
|
ncclProfilerRecordProxyOpEventState(s, args, ncclProfilerProxyOpInProgress_v4);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->addNewProxyOp(
|
|
sub->traceKey,
|
|
sub->traceInfo,
|
|
facebook_rccl::ProxyOpType::RECV,
|
|
sub->channelId,
|
|
sub->nsteps,
|
|
sub->nbytes,
|
|
sub->peer);
|
|
}
|
|
if (!sub->reg)
|
|
sub->recvMhandle = resources->mhandles[args->protocol];
|
|
}
|
|
args->state = ncclProxyOpProgress;
|
|
}
|
|
args->idle = 1;
|
|
if (args->state == ncclProxyOpProgress) {
|
|
int p = args->protocol;
|
|
int maxDepth = std::min(NCCL_STEPS, NCCL_SHARED_STEPS/args->nsubs);
|
|
for (int s=0; s<args->nsubs; s+=args->subs[s].groupSize) {
|
|
struct ncclProxySubArgs* subGroup = args->subs+s;
|
|
int subCount = 0;
|
|
void* ptrs[NCCL_PROXY_MAX_SUBS];
|
|
size_t sizes[NCCL_PROXY_MAX_SUBS];
|
|
int tags[NCCL_PROXY_MAX_SUBS];
|
|
void* mhandles[NCCL_PROXY_MAX_SUBS];
|
|
void* phandles[NCCL_PROXY_MAX_SUBS];
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
int postedStepId = sub->posted;
|
|
if (sub->posted < sub->nsteps) {
|
|
if (sub->posted >= sub->done + maxDepth) { subCount = 0; break; }
|
|
ncclProfilerStartRecvProxyStepEvent(s+i, args, postedStepId);
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
int stepSize = resources->buffSizes[p] / NCCL_STEPS;
|
|
char* localBuff = NCCL_NET_MAP_GET_POINTER(&resources->map, cpu, buffs[p]);
|
|
int buffSlot = (sub->base+sub->posted)%NCCL_STEPS;
|
|
volatile struct ncclConnFifo* connFifo = (volatile struct ncclConnFifo*)resources->recvMem->connFifo;
|
|
if (p == NCCL_PROTO_SIMPLE) {
|
|
if (resources->shared) {
|
|
if (sub->reg) {
|
|
// Wait until CUDA kernel has started before we access the user buffer directly.
|
|
if (!sub->regBufferReady && connFifo[sub->base % NCCL_STEPS].size == -1) continue;
|
|
sub->regBufferReady = 1;
|
|
ptrs[subCount] = sub->recvbuff + sub->posted * NCCL_MAX_NET_SIZE;
|
|
sizes[subCount] = std::min(NCCL_MAX_NET_SIZE, (ssize_t)(sub->nbytes - sub->posted * NCCL_MAX_NET_SIZE));
|
|
} else {
|
|
int sharedBuffSlot = sub->posted % maxDepth;
|
|
int offset;
|
|
NCCLCHECK(sharedBuffersGet(proxyState, sub->channelId, sharedBuffSlot * args->nsubs + s + i, &offset, sizes + subCount));
|
|
__atomic_store_n(&connFifo[buffSlot].offset, offset, __ATOMIC_RELAXED);
|
|
ptrs[subCount] = localBuff + offset;
|
|
}
|
|
} else {
|
|
if (sub->reg) {
|
|
if (!sub->regBufferReady && connFifo[sub->base % NCCL_STEPS].size == -1) continue;
|
|
sub->regBufferReady = 1;
|
|
sub->ringAlgo->getNextRecvAddr(sub->posted, (uint8_t**)&ptrs[subCount], &sizes[subCount], &sub->recvMhandle);
|
|
} else {
|
|
ptrs[subCount] = localBuff + buffSlot * stepSize;
|
|
sizes[subCount] = stepSize * args->sliceSteps;
|
|
}
|
|
}
|
|
} else {
|
|
ptrs[subCount] = localBuff+buffSlot*stepSize;
|
|
sizes[subCount] = stepSize*args->sliceSteps;
|
|
}
|
|
if (sub->nbytes < sizes[subCount]) sizes[subCount] = sub->nbytes;
|
|
tags[subCount] = resources->tpRemoteRank;
|
|
mhandles[subCount] = sub->recvMhandle;
|
|
phandles[subCount] = &sub->pHandles[DIVUP(postedStepId, args->sliceSteps)%NCCL_STEPS];
|
|
subCount++;
|
|
}
|
|
}
|
|
if (subCount) {
|
|
uint64_t step = subGroup->posted;
|
|
struct recvNetResources* resources = (struct recvNetResources*) (subGroup->connection->transportResources);
|
|
void** requestPtr = subGroup->requests+(step%NCCL_STEPS);
|
|
bool ignoreCompletion = ncclParamNetOptionalRecvCompletion() && ((args->protocol == NCCL_PROTO_LL128) || (args->protocol == NCCL_PROTO_LL)) && (subCount == 1);
|
|
if (ignoreCompletion) *requestPtr = (void *)NCCL_NET_OPTIONAL_RECV_COMPLETION;
|
|
NCCLCHECK(proxyState->ncclNet->irecv(resources->netRecvComm, subCount, ptrs, sizes, tags, mhandles, phandles, requestPtr));
|
|
if (*requestPtr) {
|
|
subGroup->recvRequestsCache[step%NCCL_STEPS] = *requestPtr;
|
|
subGroup->recvRequestsSubCount = subCount;
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup+i;
|
|
int postedStepId = sub->posted;
|
|
TRACE(NCCL_NET, "recvProxy [%ld/%ld/%d] Irecv posted, buff %p, size %ld, myRank %d, channelId %d, mhandle %p", sub->posted, (sub->base + sub->posted) % NCCL_STEPS, sub->nsteps, ptrs[i], sizes[i], proxyState->tpRank, sub->channelId, mhandles[i]);
|
|
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_RECV_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_RECV_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_RECV_ENTRY,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
sizes[i],
|
|
#endif
|
|
uint64_t(sub->requests+(step%NCCL_STEPS))/sizeof(void*),
|
|
*(volatile uint64_t*)NpKit::GetCpuTimestamp(), sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
#endif
|
|
|
|
sub->posted += args->sliceSteps;
|
|
ncclProfilerRecordProxyStepEventState(s+i, args, postedStepId, ncclProfilerProxyStepRecvWait);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(
|
|
sub->traceKey, facebook_rccl::ProxyCounterTypes::POSTED, sub->posted);
|
|
}
|
|
}
|
|
args->idle = 0;
|
|
}
|
|
}
|
|
}
|
|
if (args->idle == 0) return ncclSuccess;
|
|
|
|
for (int s=0; s<args->nsubs; s+=args->subs[s].groupSize) {
|
|
struct ncclProxySubArgs* subGroup = args->subs+s;
|
|
if (subGroup->posted > subGroup->received) {
|
|
uint64_t step = subGroup->received;
|
|
int done;
|
|
void* ptrs[NCCL_PROXY_MAX_SUBS];
|
|
int sizes[NCCL_PROXY_MAX_SUBS];
|
|
void* mhandles[NCCL_PROXY_MAX_SUBS];
|
|
for (int i=0; i<NCCL_PROXY_MAX_SUBS; i++) sizes[i] = 0;
|
|
NCCLCHECK(proxyState->ncclNet->test(subGroup->requests[step%NCCL_STEPS], &done, sizes));
|
|
if (done) {
|
|
int needFlush = 0;
|
|
int totalSize = 0;
|
|
for (int i=0; i<NCCL_PROXY_MAX_SUBS; i++) totalSize += sizes[i];
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
|
|
#if defined(ENABLE_NPKIT) && defined(ENABLE_NPKIT_EVENT_NET_RECV_ENTRY) && defined(ENABLE_NPKIT_EVENT_NET_RECV_EXIT)
|
|
NpKit::CollectCpuEvent(
|
|
NPKIT_EVENT_NET_RECV_EXIT,
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt,
|
|
#else
|
|
sizes[i],
|
|
#endif
|
|
uint64_t(sub->requests+(step%NCCL_STEPS))/sizeof(void*),
|
|
*(volatile uint64_t*)NpKit::GetCpuTimestamp(), sub->channelId);
|
|
#if defined(ENABLE_NPKIT_NET_COLLECT_POLL_CNT)
|
|
g_npkit_net_poll_cnt = 0;
|
|
#endif
|
|
#endif
|
|
|
|
int receivedStepId = sub->received;
|
|
int buffSlot = (sub->base + sub->received) % NCCL_STEPS;
|
|
struct recvNetResources* resources = (struct recvNetResources*)(sub->connection->transportResources);
|
|
volatile struct ncclConnFifo* connFifo = (volatile struct ncclConnFifo*)resources->recvMem->connFifo;
|
|
connFifo[buffSlot].size = -1;
|
|
sub->transSize = sizes[i];
|
|
sub->received += args->sliceSteps;
|
|
ncclProfilerRecordProxyStepEventState(s+i, args, receivedStepId, ncclProfilerProxyStepRecvFlushWait);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::RECEIVED, sub->received);
|
|
}
|
|
if (step < sub->nsteps) {
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
if (resources->useGdr) needFlush |= resources->needFlush;
|
|
}
|
|
}
|
|
subGroup->requests[step%NCCL_STEPS] = NULL;
|
|
if (totalSize > 0 && p == NCCL_PROTO_SIMPLE && needFlush) {
|
|
// GDRCOPY support
|
|
struct recvNetResources* resources = (struct recvNetResources*) (subGroup->connection->transportResources);
|
|
if (rcclParamNetHdpFlush() && resources->curr_hdp_reg) {
|
|
static bool once = true;
|
|
*resources->curr_hdp_reg = 0x1;
|
|
__sync_synchronize();
|
|
if (once) {
|
|
once = false;
|
|
INFO(NCCL_INIT, "%s: flushed HDP %p", __func__, resources->curr_hdp_reg);
|
|
}
|
|
}
|
|
if (resources->gdcFlush) {
|
|
#if defined (__x86_64__)
|
|
// Force a PCI-E read from GPU memory
|
|
static bool once = true;
|
|
asm volatile ("mov (%0), %%eax" :: "l"(resources->gdcFlush) : "%eax");
|
|
if (once) {
|
|
once = false;
|
|
INFO(NCCL_INIT, "%s: issued GDC flush", __func__);
|
|
}
|
|
#else
|
|
WARN("NET: GDR Flush only supported on x86_64");
|
|
return ncclInternalError;
|
|
#endif
|
|
} else {
|
|
int subCount = 0;
|
|
static bool once = true;
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
if (step < sub->nsteps) {
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
int stepSize = resources->buffSizes[p] / NCCL_STEPS;
|
|
char* localBuff = NCCL_NET_MAP_GET_POINTER(&resources->map, cpu, buffs[p]);
|
|
int buffSlot = (sub->base+sub->received-args->sliceSteps)%NCCL_STEPS;
|
|
if (resources->shared) {
|
|
ptrs[subCount] = sub->reg ? (char*)sub->recvbuff + step * NCCL_MAX_NET_SIZE : localBuff + resources->recvMem->connFifo[buffSlot].offset;
|
|
} else {
|
|
if (sub->reg) {
|
|
sub->ringAlgo->getNextRecvAddr(step, (uint8_t**)&ptrs[subCount], NULL, &sub->recvMhandle);
|
|
} else {
|
|
ptrs[subCount] = localBuff + buffSlot * stepSize;
|
|
}
|
|
}
|
|
mhandles[subCount] = sub->recvMhandle;
|
|
subCount++;
|
|
}
|
|
}
|
|
struct recvNetResources* resources = (struct recvNetResources*) (subGroup->connection->transportResources);
|
|
NCCLCHECK(proxyState->ncclNet->iflush(resources->netRecvComm, subCount, ptrs, sizes, mhandles, subGroup->requests+(step%NCCL_STEPS)));
|
|
// [Added-comment] iflush has been posted for this subGroup at `step`
|
|
if (subGroup->requests[step%NCCL_STEPS]){
|
|
for(int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::FLUSHED, sub->received);
|
|
}
|
|
}
|
|
}
|
|
if (once) {
|
|
once = false;
|
|
INFO(NCCL_INIT, "%s: issued GDR flush", __func__);
|
|
}
|
|
}
|
|
}
|
|
args->idle = 0;
|
|
}
|
|
}
|
|
}
|
|
if (args->idle == 0) return ncclSuccess;
|
|
|
|
for (int s=0; s<args->nsubs; s+=args->subs[s].groupSize) {
|
|
struct ncclProxySubArgs* subGroup = args->subs+s;
|
|
if (subGroup->received > subGroup->transmitted) {
|
|
uint64_t step = subGroup->transmitted;
|
|
int done = 1;
|
|
void* request = subGroup->requests[step%NCCL_STEPS];
|
|
if (request) NCCLCHECK(proxyState->ncclNet->test(request, &done, NULL));
|
|
if (done) {
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
int transmittedStepId = sub->transmitted;
|
|
|
|
sub->transmitted += args->sliceSteps;
|
|
ncclProfilerRecordProxyStepEventState(s+i, args, transmittedStepId, ncclProfilerProxyStepRecvGPUWait);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::TRANSMITTED, sub->transmitted);
|
|
}
|
|
if (step < sub->nsteps) {
|
|
__sync_synchronize();
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
volatile uint64_t* recvTail = resources->gdcSync ? resources->gdcSync : &resources->recvMem->tail;
|
|
*recvTail = sub->base + sub->transmitted;
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::RECV_TAIL, *recvTail);
|
|
}
|
|
if (resources->gdcSync) wc_store_fence(); // Flush out WC write
|
|
}
|
|
}
|
|
args->idle = 0;
|
|
}
|
|
}
|
|
}
|
|
if (args->idle == 0) return ncclSuccess;
|
|
|
|
for (int s=0; s<args->nsubs; s+=args->subs[s].groupSize) {
|
|
struct ncclProxySubArgs* subGroup = args->subs+s;
|
|
for (int i=0; i<subGroup->groupSize; i++) {
|
|
struct ncclProxySubArgs* sub = subGroup + i;
|
|
if (sub->done == sub->nsteps) continue;
|
|
if (sub->transmitted > sub->done) {
|
|
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
|
|
volatile uint64_t* sendHead = &resources->sendMem->head;
|
|
uint64_t done = *sendHead;
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::TAIL_OR_HEAD, done);
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey, facebook_rccl::ProxyCounterTypes::FIFO_SZ_OR_HEAD_CACHE, sub->base + sub->done);
|
|
}
|
|
while (done > sub->base + sub->done &&
|
|
// LL and LL128 can acknowledge 0-bytes send before they even happen. Don't go past what we transmitted.
|
|
sub->transmitted > sub->done) {
|
|
if (subGroup->recvRequestsCache[sub->done%NCCL_STEPS]) {
|
|
// the multirecv requests are only cached in the first sub.
|
|
if (proxyState->ncclNet->irecvConsumed)
|
|
NCCLCHECK(proxyState->ncclNet->irecvConsumed(resources->netRecvComm, subGroup->recvRequestsSubCount, subGroup->recvRequestsCache[sub->done%NCCL_STEPS]));
|
|
subGroup->recvRequestsCache[sub->done%NCCL_STEPS] = NULL;
|
|
}
|
|
int doneStepId = sub->done;
|
|
sub->done += args->sliceSteps;
|
|
ncclProfilerStopProxyStepEvent(s+i, args, doneStepId);
|
|
if (proxyState->proxyTrace) {
|
|
proxyState->proxyTrace->updateProxyOpCounter(sub->traceKey,
|
|
facebook_rccl::ProxyCounterTypes::DONE, sub->done);
|
|
}
|
|
args->idle = 0;
|
|
if (sub->done == sub->nsteps) {
|
|
args->done++;
|
|
if (sub->ringAlgo && sub->ringAlgo->decRefCount() == 0) delete sub->ringAlgo;
|
|
sub->ringAlgo = NULL;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (args->done == args->nsubs) {
|
|
args->state = ncclProxyOpNone;
|
|
for (int s=0; s<args->nsubs; s++) {
|
|
ncclProfilerStopProxyOpEvent(s, args);
|
|
}
|
|
}
|
|
}
|
|
return ncclSuccess;
|
|
}
|
|
|
|
ncclResult_t ncclNetDeregBuffer(struct ncclComm* comm, struct ncclProxyConnector* proxyConn, void* handle) {
|
|
NCCLCHECK(ncclProxyCallBlocking(comm, proxyConn, ncclProxyMsgDeregister, &handle, sizeof(void*), NULL, 0));
|
|
INFO(NCCL_REG, "rank %d - deregistered net buffer handle %p", comm->rank, handle);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t netRegisterBuffer(ncclComm* comm, const void* userbuff, size_t buffSize, struct ncclConnector** peerConns, int nPeers, struct ncclReg* regRecord, int* outRegBufFlag, void** outHandle) {
|
|
ncclResult_t ret = ncclSuccess;
|
|
int gdrFlag = 1;
|
|
|
|
if (regRecord) {
|
|
for (int p = 0; p < nPeers; ++p) {
|
|
struct ncclConnector* peerConn = peerConns[p];
|
|
struct ncclProxyConnector* peerProxyConn = NULL;
|
|
struct ncclRegNetHandles* netHandle = NULL;
|
|
bool found = false;
|
|
if (peerConn == NULL) continue;
|
|
peerProxyConn = &peerConn->proxyConn;
|
|
netHandle = regRecord->netHandleHead;
|
|
while (netHandle) {
|
|
if (netHandle->proxyConn == peerProxyConn) {
|
|
found = true;
|
|
break;
|
|
}
|
|
netHandle = netHandle->next;
|
|
}
|
|
if (found) {
|
|
*outRegBufFlag = 1;
|
|
outHandle[p] = netHandle->handle;
|
|
INFO(NCCL_REG, "rank %d - NET reuse buffer %p size %ld (baseAddr %p size %ld) handle %p", comm->rank, userbuff, buffSize, (void*)regRecord->begAddr, regRecord->endAddr - regRecord->begAddr, netHandle->handle);
|
|
} else {
|
|
struct netRegInfo info = { regRecord->begAddr, regRecord->endAddr - regRecord->begAddr };
|
|
void* handle = NULL;
|
|
|
|
if (peerConn->conn.flags & NCCL_DIRECT_NIC) {
|
|
NCCLCHECKGOTO(ncclProxyCallBlocking(comm, peerProxyConn, ncclProxyMsgRegister, &info, sizeof(struct netRegInfo), &handle, sizeof(void*)), ret, fail);
|
|
if (handle) {
|
|
struct ncclRegNetHandles* netHandle;
|
|
regRecord->state |= NET_REG_COMPLETE;
|
|
NCCLCHECK(ncclCalloc(&netHandle, 1));
|
|
netHandle->handle = handle;
|
|
netHandle->proxyConn = peerProxyConn;
|
|
netHandle->next = regRecord->netHandleHead;
|
|
regRecord->netHandleHead = netHandle;
|
|
outHandle[p] = handle;
|
|
*outRegBufFlag = 1;
|
|
INFO(NCCL_REG, "rank %d - NET register userbuff %p (handle %p), buffSize %ld", comm->rank, userbuff, handle, buffSize);
|
|
} else {
|
|
goto fail;
|
|
}
|
|
} else {
|
|
gdrFlag = 0;
|
|
goto fail;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
exit:
|
|
return ret;
|
|
fail:
|
|
*outRegBufFlag = 0;
|
|
INFO(NCCL_REG, "rank %d failed to NET register userbuff %p buffSize %ld GDR flag %d", comm->rank, userbuff, buffSize, gdrFlag);
|
|
goto exit;
|
|
}
|
|
|
|
ncclResult_t ncclNetLocalRegisterBuffer(ncclComm* comm, const void* userbuff, size_t buffSize, struct ncclConnector** peerConns, int nPeers, int* outRegBufFlag, void** outHandle) {
|
|
ncclResult_t ret = ncclSuccess;
|
|
struct ncclReg *regRecord = NULL;
|
|
bool isValid = false;
|
|
|
|
*outRegBufFlag = 0;
|
|
if (comm && userbuff && buffSize > 0 && nPeers > 0) {
|
|
NCCLCHECKGOTO(ncclRegFind(comm, userbuff, buffSize, ®Record), ret, fail);
|
|
NCCLCHECKGOTO(ncclRegLocalIsValid(regRecord, &isValid), ret, fail);
|
|
if (isValid)
|
|
NCCLCHECKGOTO(netRegisterBuffer(comm, userbuff, buffSize, peerConns, nPeers, regRecord, outRegBufFlag, outHandle), ret, fail);
|
|
}
|
|
|
|
exit:
|
|
return ret;
|
|
fail:
|
|
*outRegBufFlag = 0;
|
|
goto exit;
|
|
}
|
|
|
|
struct ncclNetCleanupCallback {
|
|
struct ncclCommCallback base;
|
|
struct ncclComm *comm;
|
|
struct ncclReg *reg;
|
|
};
|
|
|
|
static ncclResult_t cleanupNet(struct ncclComm* comm, struct ncclCommCallback* cb) {
|
|
struct ncclNetCleanupCallback* obj = (struct ncclNetCleanupCallback*)cb;
|
|
NCCLCHECK(ncclCommGraphDeregister(obj->comm, obj->reg));
|
|
free(obj);
|
|
return ncclSuccess;
|
|
}
|
|
|
|
ncclResult_t ncclNetGraphRegisterBuffer(ncclComm* comm, const void* userbuff, size_t buffSize, struct ncclConnector** peerConns, int nPeers, int* outRegBufFlag, void** outHandle, struct ncclIntruQueue<struct ncclCommCallback, &ncclCommCallback::next>* cleanupQueue, int* nCleanupQueueElts) {
|
|
ncclResult_t ret = ncclSuccess;
|
|
struct ncclNetCleanupCallback *record = NULL;
|
|
struct ncclReg *regRecord = NULL;
|
|
void *baseSend;
|
|
size_t baseSendSize;
|
|
|
|
*outRegBufFlag = 0;
|
|
if (comm && userbuff && buffSize > 0 && nPeers > 0) {
|
|
CUDACHECKGOTO(cuMemGetAddressRange((CUdeviceptr *)&baseSend, &baseSendSize, (CUdeviceptr)userbuff), ret, fail);
|
|
NCCLCHECKGOTO(ncclCommGraphRegister(comm, baseSend, baseSendSize, (void**)®Record), ret, fail);
|
|
NCCLCHECKGOTO(netRegisterBuffer(comm, userbuff, buffSize, peerConns, nPeers, regRecord, outRegBufFlag, outHandle), ret, fail);
|
|
if (*outRegBufFlag) {
|
|
NCCLCHECKGOTO(ncclCalloc(&record, 1), ret, fail);
|
|
record->base.fn = cleanupNet;
|
|
record->comm = comm;
|
|
record->reg = regRecord;
|
|
ncclIntruQueueEnqueue(cleanupQueue, (struct ncclCommCallback*)record);
|
|
if (nCleanupQueueElts) *nCleanupQueueElts += 1;
|
|
} else {
|
|
NCCLCHECKGOTO(ncclCommGraphDeregister(comm, regRecord), ret, fail);
|
|
}
|
|
}
|
|
exit:
|
|
return ret;
|
|
fail:
|
|
*outRegBufFlag = 0;
|
|
goto exit;
|
|
}
|
|
|
|
static ncclResult_t sendProxyRegBuffer(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
void* handle;
|
|
struct netRegInfo* info = (struct netRegInfo*)reqBuff;
|
|
struct sendNetResources* resources = (struct sendNetResources*)(connection->transportResources);
|
|
ncclResult_t ret = ncclSuccess;
|
|
bool needReg = true;
|
|
|
|
assert(reqSize == sizeof(struct netRegInfo));
|
|
assert(respSize == sizeof(void*));
|
|
|
|
#if CUDART_VERSION >= 11070
|
|
/* DMA-BUF support */
|
|
if (resources->useDmaBuf) {
|
|
int dmabuf_fd;
|
|
CUCHECKGOTO(cuMemGetHandleForAddressRange((void*)&dmabuf_fd, (CUdeviceptr)info->buffer, info->size, CU_MEM_RANGE_HANDLE_TYPE_DMA_BUF_FD, getHandleForAddressRangeFlags(resources->useGdr)), ret, peermem);
|
|
NCCLCHECKGOTO(proxyState->ncclNet->regMrDmaBuf(resources->netSendComm, (void*)info->buffer, info->size, NCCL_PTR_CUDA, 0ULL, dmabuf_fd, &handle), ret, peermem);
|
|
(void)close(dmabuf_fd);
|
|
needReg = false;
|
|
}
|
|
peermem:
|
|
#endif
|
|
if (needReg) {
|
|
NCCLCHECKGOTO(proxyState->ncclNet->regMr(resources->netSendComm, (void*)info->buffer, info->size, NCCL_PTR_CUDA, &handle), ret, fail);
|
|
}
|
|
|
|
exit:
|
|
memcpy(respBuff, (void*)&handle, sizeof(void*));
|
|
*done = 1;
|
|
return ncclSuccess;
|
|
fail:
|
|
handle = NULL;
|
|
goto exit;
|
|
}
|
|
|
|
static ncclResult_t recvProxyRegBuffer(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
|
|
void* handle;
|
|
struct netRegInfo* info = (struct netRegInfo*)reqBuff;
|
|
struct recvNetResources* resources = (struct recvNetResources*)(connection->transportResources);
|
|
ncclResult_t ret = ncclSuccess;
|
|
bool needReg = true;
|
|
|
|
assert(reqSize == sizeof(struct netRegInfo));
|
|
assert(respSize == sizeof(void*));
|
|
|
|
#if CUDART_VERSION >= 11070
|
|
/* DMA-BUF support */
|
|
if (resources->useDmaBuf) {
|
|
int dmabuf_fd;
|
|
CUCHECKGOTO(cuMemGetHandleForAddressRange((void*)&dmabuf_fd, (CUdeviceptr)info->buffer, info->size, CU_MEM_RANGE_HANDLE_TYPE_DMA_BUF_FD, getHandleForAddressRangeFlags(resources->useGdr)), ret, peermem);
|
|
NCCLCHECKGOTO(proxyState->ncclNet->regMrDmaBuf(resources->netRecvComm, (void*)info->buffer, info->size, NCCL_PTR_CUDA, 0ULL, dmabuf_fd, &handle), ret, peermem);
|
|
(void)close(dmabuf_fd);
|
|
needReg = false;
|
|
}
|
|
peermem:
|
|
#endif
|
|
if (needReg) {
|
|
NCCLCHECKGOTO(proxyState->ncclNet->regMr(resources->netRecvComm, (void*)info->buffer, info->size, NCCL_PTR_CUDA, &handle), ret, fail);
|
|
}
|
|
|
|
exit:
|
|
memcpy(respBuff, (void*)&handle, sizeof(void*));
|
|
*done = 1;
|
|
return ncclSuccess;
|
|
fail:
|
|
handle = NULL;
|
|
goto exit;
|
|
}
|
|
|
|
static ncclResult_t sendProxyDeregBuffer(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, int* done) {
|
|
void* handle;
|
|
struct sendNetResources* resources = (struct sendNetResources*)(connection->transportResources);
|
|
|
|
assert(reqSize == sizeof(void*));
|
|
memcpy(&handle, reqBuff, sizeof(void*));
|
|
NCCLCHECK(proxyState->ncclNet->deregMr(resources->netSendComm, handle));
|
|
*done = 1;
|
|
return ncclSuccess;
|
|
}
|
|
|
|
static ncclResult_t recvProxyDeregBuffer(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, int* done) {
|
|
void* handle;
|
|
struct recvNetResources* resources = (struct recvNetResources*)(connection->transportResources);
|
|
|
|
assert(reqSize == sizeof(void*));
|
|
memcpy(&handle, reqBuff, sizeof(void*));
|
|
NCCLCHECK(proxyState->ncclNet->deregMr(resources->netRecvComm, handle));
|
|
*done = 1;
|
|
return ncclSuccess;
|
|
}
|
|
|
|
struct ncclTransport netTransport = {
|
|
"NET",
|
|
canConnect,
|
|
{ sendSetup, sendConnect, sendFree, proxySharedInit, sendProxySetup, sendProxyConnect, sendProxyFree, sendProxyProgress, sendProxyRegBuffer, sendProxyDeregBuffer },
|
|
{ recvSetup, recvConnect, recvFree, proxySharedInit, recvProxySetup, recvProxyConnect, recvProxyFree, recvProxyProgress, recvProxyRegBuffer, recvProxyDeregBuffer }
|
|
};
|