AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

NCCL 2.27.3-1

Browse Source 
Symmetric memory API and symmetric kernels
 * Redesign from the ground up, enabling major latency and bandwidth
   improvements.
 * Add new API calls to register user-allocated memory among communicator
   ranks into a NCCL window: ncclCommWindowRegister() and
   ncclCommWindowDeregister(). The calls currently support symmetric
   registration for P2P and NVLS, and require VMM memory buffers (i.e.,
   CUMEM must be operational).
 * Implement specialized kernels taking advantage of symmetrically
   registered memory, with performance gains expected particularly for
   small to medium message sizes.
 * The kernels support 32 bit floating point types and smaller, and sum as
   the reduction operator, with no more than one collective operation per
   group.
 * Floating point summation is always done in fp32 accumulators (with the
   exception of fp8 on NVLS, where it uses fp16 inside the switch). Thus,
   the accuracy with fp8 and fp16 data types should be much improved.
 * This initial implementation supports non-network communicators only (P2P
   and NVLS transports).
 * To explore this functionality users need to use the new memory
   registration API calls with the NCCL_WIN_COLL_SYMMETRIC flag and all
   ranks of a communicator must pass buffers at the same offset in the same
   registration when invoking a collective NCCL operation.

Add support for DGX Spark.

Add support for DirectNIC (CX8) to the internal IB plugin.

Add a new ncclCommShrink() API call
 * It is a non-collective call similar to ncclCommSplit(), which makes it
   possible to exclude some (possibly unresponsive) ranks from the parent
   communicator.

Add support for loading multiple network plugins
 * This enables the creation of generic containers that can work across a
   range of providers.
 * Allow NCCL_NET_PLUGIN to accept a comma-separated list of plugins to
   load.

NVLink SHARP (NVLS) improvements
 * Implement NVLS+IB SHARP support for AllGather and ReduceScatter with
   user buffer registration. This improves performance and reduces the
   number of CTAs needed to achieve peak bandwidth.
 * Gracefully fall back by default to other transports if NVLS
   initialization fails (the old behavior of returning an error code from a
   NCCL call can be preserved by setting NCCL_NVLS_ENABLE=1).
 * Decrease the NVLS channel count to 24 on Blackwell systems with multiple
   NVLink domains per communicator.
 * Enable fine-tuning of NCCL behavior per communicator using new
   "ncclConfig_t" members "collnetEnable", "CTAPolicy", and "nvlsCTAs".

Profiler improvements
 * Extend the init function by adding communicator name, comm id (hash),
   rank, number of ranks, number of nodes, and the NCCL log function to the
   argument list. This makes the name and the comm id available to all
   events in the communicator without explicitly passing them to each
   individual event. Add the communicator id and rank to the profiler trace
   filename. Now, the communicator name can be set via a new "ncclConfig_t"
   member "commName".
 * Improve the accuracy of the GPU kernel events by providing GPU-generated
   timestamps for the start and stop of every NCCL operation.
 * Harmonize proxy events, removing overlaps between ProxyOp and ProxyStep
   states.
 * Add support for network-defined event updates (through
   "recordEventState").
 * Report the correct number of channels used by every collective/p2p
   operation (used to be set to nMaxChannels for collectives and absent for
   p2ps).
 * Fix the logic on proxyCtrl Idle/Active events (Issue #1162).
 * Fix an issue where the network proxy profiler could lose track of an
   event identifier (Issue #1682).
 * Improve the backward compatibility with plugins older than v4.
 * Ensure that the work counters are 0-initialized.
 * Fix a potential race condition in the network profiler that could result
   in an event being linked to a wrong parent.

MNNVL improvements
 * Increase to 16 the number of NICs used to communicate between MNNVL
   domains on GB200 systems, to optimize the performance of collective
   operations.
 * Add support for more complex MNNVL topologies with up to 32 NICs per
   node.
 * If the MNNVL fabric initialization was unsuccessful, NCCL will now fail
   by default, so as to avoid inadvertently falling back to a potentially
   much slower network transport. Such failures are typically due to a
   misconfigured IMEX support on the system. To continue without MNNVL,
   restart the job with NCCL_MNNVL_ENABLE=0.
 * Fix a potential hang in alltoall-like communication patterns at a scale
   of over 80 ranks.
 * Make NCCL_P2P_DISABLE=1 imply NCCL_MNNVL_ENABLE=0 (so the latter no
   longer needs to be specified on MNNVL systems).
 * Fix an initialization failure when NCCL_TOPO_FILE is used on MNNVL
   systems.
 * Fix the graph search to exclude non-local NICs.
 * Fix the SHM transport to use fabric handles on MNNVL systems.

NIC Fusion improvements
 * Disable the creation of fused NICs for physical devices that haven't
   been merged.
 * Flatten multiple ports to a single PCI device within the internal IB
   plugin and reparent dual-port NICs under the first PCI parent. If the
   parent is not a PCI switch, PCI devices for fused NICs won't be
   duplicated.
 * Route traffic on GB200-CX8 systems through DirectNIC, not the host
   interface.

Improve support for platforms with C2C connectivity (e.g., GB200)
 * Enable GPUDirect RDMA for the NICs by default.
 * Add support for P2C (PXN over C2C) and the LL128 protocol.

Extend NCCL fault tolerance in multithreaded scenarios
 * Support the creation of multiple nonblocking communicators within a
   single group and polling in parallel for the completion using multiple
   threads (one per communicator).

Enable ncclImplicitOrderLaunch for CUDA 12.9+
 * This can potentially speed up NCCL_IMPLICIT_LAUNCH_ORDER.

Improve the netSocket transport latency and control
 * Provide finer control over the size of the socket send/receive buffers,
   the task size, and the number of sockets that a single peer can open.
 * Add support for the inlining of small messages behind the header when
   using multiple sockets per connection.

Improve the readability of the CPU affinity in the debug output
 * Print it as a range string rather than a bitmask.

Fix a potential race condition in graph execution
 * A contention could arise when mixing graph and non-graph execution.

Improve PXN connection code
 * Avoid duplicate and unused connections.

RAS fixes
 * Fix a memory corruption at job termination time in case of a previously
   failed initialization of a RAS socket connection.
 * Fix a race condition leading to a crash when generating a RAS report
   during communicator initialization (Issues #1669, #1718).
 * Fix a potential race condition when gathering data for a RAS status
   report.

Fix a potential memory corruption in ncclCommSplit()
 * Memory could get corrupted when resource sharing was in use and the size
   of the NVLink domain in the new communicator was smaller than in the old
   one.

Fix asynchronous graph upload
 * Fix a small memory leak.
 * Fix oversychronization.

Add a check for out-of-memory conditions in ncclMemAlloc()

Clean up the NCCL socket code
 * accept() will retry also if just reading the magic failed (Issue #1613).
 * connect() will retry also if poll() did not return a POLLOUT event
   (Issue #1618).
 * Add error checking in a few instances (Issue #1539).
 * Fix the loop condition in ncclFindInterfaceMatchSubnet() (Issue #1574).
 * Clean up the debug output, downgrading WARN messages to INFO in
   non-critical cases, and printing the peer's address where relevant.

Switch NCCL_DEBUG_FILE to line buffering
 * This should help avoid mixed-up partial output lines in multithreaded
   cases.

Other minor fixes
 * Improve the checks for buffer overflows in the graph code (Issue #1585).
 * Extend logging and state clearing to all four events in the internal IB
   plugin (Issue #1650).
 * Fix the error path in case IB communication is not ready (Issue #1489).
 * Add ECE logging for IB fabric.
 * Fix various minor issues in the graph module (Issue #1635).
 * Clean up the debug output in the graph code, downgrading WARN messages
   to INFO in non-critical cases.
 * Add a missing argument to a directSend() call (Issue #1628).
 * Remove duplicate code in sendProxySetup() (Issue #1420).
 * Fix the order of arguments of cudaDeviceCanAccessPeer() (Issue #1507).
 * Fix compiler warnings with GCC 14.
 * Fix a typo in a comment (Issue #1236).
This commit is contained in:
Kamil Iskra
2025-05-29 20:56:40 -07:00
parent 8171af656b
commit 72d2432094
99 changed files with 7223 additions and 2029 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/transport/coll_net.cc
+7 -9
View File
@@ -13,6 +13,7 @@
#include "assert.h"
#include "bootstrap.h"
#include "channel.h"
#include "register_inline.h"
int64_t ncclParamGdrCopySyncEnable();
int64_t ncclParamGdrCopyFlushEnable();
@@ -1188,7 +1189,7 @@ static ncclResult_t collnetRegisterBuffer(struct ncclComm* comm, const void* use
goto exit;
} else {
/* start register collnet buffer */
struct collnetRegInfo info = { regRecord->addr, regRecord->pages * comm->regCache.pageSize };
struct collnetRegInfo info = { regRecord->begAddr, regRecord->endAddr - regRecord->begAddr };
void* handle = NULL;
struct ncclConnInfo* conn = (type == collNetRecv) ? &comm->channels[0].peers[comm->nRanks]->recv[type].conn : &comm->channels[0].peers[comm->nRanks]->send[type].conn;
@@ -1389,7 +1390,7 @@ ncclResult_t ncclCollNetChainBufferSetup(ncclComm_t comm) {
ncclResult_t ret = ncclSuccess;
char line[1024];
if (comm->collNetSupport == 0) goto exit;
if (comm->config.collnetEnable == 0) goto exit;
// Connect Collnet + chain
for (int c = 0; c < comm->nChannels; c++) {
struct ncclChannel* channel = comm->channels + c;
@@ -1421,7 +1422,7 @@ fail:
ncclResult_t ncclCollNetDirectBufferSetup(ncclComm_t comm) {
ncclResult_t ret = ncclSuccess;
if (comm->collNetSupport == 0) goto exit;
if (comm->config.collnetEnable == 0) goto exit;
// Connect intra-node CollNet + Direct
for (int c = 0; c < comm->nChannels; c++) {
@@ -1498,8 +1499,8 @@ ncclResult_t ncclCollNetSetup(ncclComm_t comm, ncclComm_t parent, struct ncclTop
comm->collNetHeads = headsUnique;
comm->collNetHeadsNum = nHeadsUnique;
if (parent && parent->collNetSupport && parent->nNodes == comm->nNodes) {
if (!parent->config.splitShare) {
if (parent && parent->config.collnetEnable && parent->nNodes == comm->nNodes) {
if (!parent->shareResources) {
collNetSetupFail = 1;
goto fail;
}
@@ -1547,9 +1548,6 @@ ncclResult_t ncclCollNetSetup(ncclComm_t comm, ncclComm_t parent, struct ncclTop
NCCLCHECKGOTO(collNetInitRailRankMap(comm), ret, fail);
} else {
/* TODO: CX-6 and CX-7 both do not support multiple sharp resources per process, if child comm cannot
* share the sharp resource from parent, we cannot use sharp in this case. This restriction might be
* lifted by sharp plugin/IB hardware in the future. */
collNetSetupFail = 1;
if (comm->rank == 0) {
WARN("Child comms (nRanks %d) fails to share parent comms (nRanks %d) sharp resources", comm->nRanks, parent->nRanks);
@@ -1629,7 +1627,7 @@ exit:
return ret;
fail:
ncclTransportCollNetFree(comm);
comm->collNetSupport = 0;
comm->config.collnetEnable = 0;
goto exit;
}
src/transport/net.cc
+29 -23
View File
@@ -16,6 +16,7 @@
#include "transport.h"
#include "shm.h"
#include <assert.h>
#include "register_inline.h"
static_assert(sizeof(ncclNetHandle_t) <= CONNECT_SIZE, "NET Connect info is too large");
@@ -629,8 +630,6 @@ static ncclResult_t sendProxySetup(struct ncclProxyConnection* connection, struc
resources->netDeviceVersion = props.netDeviceVersion;
resources->netDeviceType = props.netDeviceType;
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)) {
@@ -732,7 +731,14 @@ static ncclResult_t sendProxyConnect(struct ncclProxyConnection* connection, str
NCCLCHECK(ncclCalloc(progressState->netComms + resources->netDev, proxyState->tpnRanks));
}
struct ncclSharedNetComms* comms = progressState->netComms[resources->netDev] + resources->tpRemoteRank;
if (comms->sendComm[resources->channelId] == NULL) ret = proxyState->ncclNet->connect(resources->netDev, &commConfig, req->handle, comms->sendComm + resources->channelId, &resources->netDeviceHandle);
// 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)) {
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 {
@@ -886,7 +892,15 @@ static ncclResult_t recvProxyConnect(struct ncclProxyConnection* connection, str
NCCLCHECK(ncclCalloc(progressState->netComms + resources->netDev, proxyState->tpnRanks));
}
struct ncclSharedNetComms* comms = progressState->netComms[resources->netDev] + resources->tpRemoteProxyRank;
if (comms->recvComm[resources->channelId] == NULL) ret = proxyState->ncclNet->accept(resources->netListenComm, comms->recvComm+resources->channelId, &resources->netDeviceHandle);
// 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)) {
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 {
@@ -1101,7 +1115,7 @@ static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct
// Set step base for next op
resources->step = sub->base + sub->nsteps;
sub->posted = sub->transmitted = sub->done = 0;
ncclProfilerStartSendProxyOpEvent(s, args);
ncclProfilerRecordProxyOpEventState(s, args, ncclProfilerProxyOpInProgress_v4);
if (!sub->reg)
sub->sendMhandle = resources->mhandles[args->protocol];
}
@@ -1140,7 +1154,6 @@ static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct
} else {
sub->posted += args->sliceSteps;
}
ncclProfilerRecordProxyOpEventState(s, args, sub->posted, sub->transSize, ncclProfilerProxyOpSendPosted);
ncclProfilerRecordProxyStepEventState(s, args, postedStepId, ncclProfilerProxyStepSendGPUWait);
args->idle = 0;
continue;
@@ -1188,18 +1201,17 @@ static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct
}
}
if (ready) {
ncclProfilerRecordProxyOpEventState(s, args, sub->transmitted+args->sliceSteps, sub->transSize, ncclProfilerProxyOpSendRemFifoWait);
ncclProfilerRecordProxyStepEventState(s, args, transmittedStepId, ncclProfilerProxyStepSendPeerWait_v4);
// 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]
NCCLCHECK(proxyState->ncclNet->isend(resources->netSendComm, buff, size, resources->tpRank, sub->sendMhandle, sub, sub->requests+buffSlot));
void* phandle = &sub->pHandles[DIVUP(transmittedStepId, args->sliceSteps)%NCCL_STEPS];
NCCLCHECK(proxyState->ncclNet->isend(resources->netSendComm, buff, size, resources->tpRank, sub->sendMhandle, phandle, sub->requests+buffSlot));
if (sub->requests[buffSlot] != NULL) {
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->transSize = size;
sub->transmitted += args->sliceSteps;
sub->profilerSteps++;
ncclProfilerRecordProxyOpEventState(s, args, sub->transmitted, sub->transSize, ncclProfilerProxyOpSendTransmitted);
ncclProfilerRecordProxyStepEventState(s, args, transmittedStepId, ncclProfilerProxyStepSendWait);
args->idle = 0;
continue;
@@ -1220,7 +1232,6 @@ static ncclResult_t sendProxyProgress(struct ncclProxyState* proxyState, struct
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);
ncclProfilerRecordProxyOpEventState(s, args, sub->done, sub->transSize, ncclProfilerProxyOpSendDone);
if (resources->shared == 0) {
volatile uint64_t* sendHead = resources->gdcSync ? resources->gdcSync : &resources->sendMem->head;
@@ -1282,7 +1293,7 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
sub->posted = sub->received = sub->transmitted = sub->done = 0;
sub->regBufferReady = 0;
for (int i=0; i<groupSize; i++) sub[-i].groupSize = groupSize;
ncclProfilerStartRecvProxyOpEvent(s, args);
ncclProfilerRecordProxyOpEventState(s, args, ncclProfilerProxyOpInProgress_v4);
if (!sub->reg)
sub->recvMhandle = resources->mhandles[args->protocol];
}
@@ -1343,7 +1354,7 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
if (sub->nbytes < sizes[subCount]) sizes[subCount] = sub->nbytes;
tags[subCount] = resources->tpRemoteRank;
mhandles[subCount] = sub->recvMhandle;
phandles[subCount] = sub;
phandles[subCount] = &sub->pHandles[DIVUP(postedStepId, args->sliceSteps)%NCCL_STEPS];
subCount++;
}
}
@@ -1362,8 +1373,6 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
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]);
sub->posted += args->sliceSteps;
sub->profilerSteps++;
ncclProfilerRecordProxyOpEventState(s+i, args, sub->posted, sub->transSize, ncclProfilerProxyOpRecvPosted);
ncclProfilerRecordProxyStepEventState(s+i, args, postedStepId, ncclProfilerProxyStepRecvWait);
}
args->idle = 0;
@@ -1393,9 +1402,8 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
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->transSize = sizes[i];
sub->received += args->sliceSteps;
ncclProfilerRecordProxyOpEventState(s+i, args, sub->received, sub->transSize, ncclProfilerProxyOpRecvReceived);
ncclProfilerRecordProxyStepEventState(s+i, args, receivedStepId, ncclProfilerProxyStepRecvFlushWait);
if (step < sub->nsteps) {
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
@@ -1459,7 +1467,6 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
int transmittedStepId = sub->transmitted;
sub->transmitted += args->sliceSteps;
ncclProfilerRecordProxyOpEventState(s+i, args, sub->transmitted, sub->transSize, ncclProfilerProxyOpRecvTransmitted);
ncclProfilerRecordProxyStepEventState(s+i, args, transmittedStepId, ncclProfilerProxyStepRecvGPUWait);
if (step < sub->nsteps) {
__sync_synchronize();
@@ -1479,7 +1486,6 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
struct ncclProxySubArgs* subGroup = args->subs+s;
for (int i=0; i<subGroup->groupSize; i++) {
struct ncclProxySubArgs* sub = subGroup + i;
int doneStepId = sub->done;
if (sub->done == sub->nsteps) continue;
if (sub->transmitted > sub->done) {
struct recvNetResources* resources = (struct recvNetResources*) (sub->connection->transportResources);
@@ -1494,9 +1500,9 @@ static ncclResult_t recvProxyProgress(struct ncclProxyState* proxyState, struct
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);
ncclProfilerRecordProxyOpEventState(s+i, args, sub->done, sub->transSize, ncclProfilerProxyOpRecvDone);
args->idle = 0;
if (sub->done == sub->nsteps) {
args->done++;
@@ -1547,9 +1553,9 @@ static ncclResult_t netRegisterBuffer(ncclComm* comm, const void* userbuff, size
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->addr, regRecord->pages * comm->regCache.pageSize, 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->addr, regRecord->pages * comm->regCache.pageSize };
struct netRegInfo info = { regRecord->begAddr, regRecord->endAddr - regRecord->begAddr };
void* handle = NULL;
if (peerConn->conn.flags & NCCL_DIRECT_NIC) {
src/transport/net_ib.cc
+197 -105
View File
@@ -25,8 +25,10 @@
#include "timer.h"
#include "ibvwrap.h"
#include "mlx5/mlx5dvwrap.h"
#define MAXNAMESIZE 64
#define MAXSUFFIXSIZE 16
#define MAXNAMESIZE (64 + MAXSUFFIXSIZE)
static char ncclIbIfName[MAX_IF_NAME_SIZE+1];
static union ncclSocketAddress ncclIbIfAddr;
@@ -55,6 +57,17 @@ struct ncclIbStats {
int fatalErrorCount;
};
enum ncclIbProvider {
IB_PROVIDER_NONE = 0,
IB_PROVIDER_MLX5 = 1,
IB_PROVIDER_MAX = 2,
};
const char* ibProviderName[] = {
"None",
"Mlx5",
};
static int ncclNIbDevs = -1;
struct alignas(64) ncclIbDev {
pthread_mutex_t lock;
@@ -77,6 +90,12 @@ struct alignas(64) ncclIbDev {
struct ibv_port_attr portAttr;
struct ncclIbStats stats;
int dmaBufSupported;
enum ncclIbProvider ibProvider;
union {
struct {
int dataDirect;
} mlx5;
} capsProvider;
};
#define MAX_IB_DEVS 32
@@ -106,6 +125,7 @@ NCCL_PARAM(IbAdaptiveRouting, "IB_ADAPTIVE_ROUTING", -2);
NCCL_PARAM(IbFifoTc, "IB_FIFO_TC", -1);
NCCL_PARAM(IbAsyncEvents,"IB_RETURN_ASYNC_EVENTS",1);
NCCL_PARAM(IbEceEnable,"IB_ECE_ENABLE",1);
NCCL_PARAM(IbDataDirect,"IB_DATA_DIRECT",1);
static ncclResult_t ncclIbStatsInit(struct ncclIbStats* stat) {
__atomic_store_n(&stat->fatalErrorCount, 0, __ATOMIC_RELAXED);
@@ -451,6 +471,10 @@ static ncclResult_t ncclIbGetPciPath(char* devName, char** path, int* realPort)
if (p == NULL) {
WARN("Could not find real path of %s (%s)", devName, devicePath);
} else {
// Merge multi-port NICs into the same PCI device
p[strlen(p)-1] = '0';
// Also merge virtual functions (VF) into the same device
if (ncclParamIbMergeVfs()) p[strlen(p)-3] = p[strlen(p)-4] = '0';
// Keep the real port aside (the ibv port is always 1 on recent cards)
*realPort = 0;
for (int d=0; d<ncclNIbDevs; d++) {
@@ -495,9 +519,29 @@ static int ncclIbRelaxedOrderingCapable(void) {
return r == ncclInternalError ? 0 : 1;
}
static bool ncclMlx5dvDmaBufCapable(ibv_context *context){
ncclResult_t res;
int dev_fail = 0;
struct ibv_pd* pd;
NCCLCHECKGOTO(wrap_ibv_alloc_pd(&pd, context), res, failure);
// Test kernel DMA-BUF support with a dummy call (fd=-1)
(void)wrap_direct_ibv_reg_dmabuf_mr(pd, 0ULL /*offset*/, 0ULL /*len*/, 0ULL /*iova*/, -1 /*fd*/, 0 /*flags*/);
// ibv_reg_dmabuf_mr() will fail with EOPNOTSUPP/EPROTONOSUPPORT if not supported (EBADF otherwise)
(void)wrap_direct_mlx5dv_reg_dmabuf_mr(pd, 0ULL /*offset*/, 0ULL /*len*/, 0ULL /*iova*/, -1 /*fd*/, 0 /*flags*/, 0 /* mlx5 flags*/);
// mlx5dv_reg_dmabuf_mr() will fail with EOPNOTSUPP/EPROTONOSUPPORT if not supported (EBADF otherwise)
dev_fail |= (errno == EOPNOTSUPP) || (errno == EPROTONOSUPPORT);
NCCLCHECKGOTO(wrap_ibv_dealloc_pd(pd), res, failure);
// stop the search and goto failure
if (dev_fail) goto failure;
return true;
failure:
return false;
}
ncclResult_t ncclIbMakeVDeviceInternal(int* d, ncclNetVDeviceProps_t* props) {
if (ncclParamIbMergeNics() == 0 && props->ndevs > 1) {
WARN("NET/IB : Trying to merge multiple devices together when NCCL_IB_MERGE_NICS=0. Please enable it or disable device merging in NCCL.");
INFO(NCCL_NET, "NET/IB : Skipping makeVDevice, NCCL_IB_MERGE_NICS=0");
return ncclInvalidUsage;
}
@@ -565,14 +609,17 @@ ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction, ncclProfilerCallback_t pr
if (ncclParamIbDisable()) return ncclInternalError;
static int shownIbHcaEnv = 0;
if(wrap_ibv_symbols() != ncclSuccess) { return ncclInternalError; }
if(wrap_mlx5dv_symbols() != ncclSuccess) { INFO(NCCL_NET, "NET/IB : Failed to open mlx5dv symbols. Advance features like CX-8 Direct-NIC will be disabled."); }
if (ncclNIbDevs == -1) {
pthread_mutex_lock(&ncclIbLock);
wrap_ibv_fork_init();
if (ncclNIbDevs == -1) {
int nIpIfs = 0;
ncclNIbDevs = 0;
ncclNMergedIbDevs = 0;
if (ncclFindInterfaces(ncclIbIfName, &ncclIbIfAddr, MAX_IF_NAME_SIZE, 1) != 1) {
NCCLCHECK(ncclFindInterfaces(ncclIbIfName, &ncclIbIfAddr, MAX_IF_NAME_SIZE, 1, &nIpIfs));
if (nIpIfs != 1) {
WARN("NET/IB : No IP interface found.");
ret = ncclInternalError;
goto fail;
@@ -600,6 +647,17 @@ ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction, ncclProfilerCallback_t pr
WARN("NET/IB : Unable to open device %s", devices[d]->name);
continue;
}
enum ncclIbProvider ibProvider = IB_PROVIDER_NONE;
char dataDirectDevicePath[PATH_MAX];
int dataDirectSupported = 0;
if (wrap_mlx5dv_is_supported(devices[d])) {
ibProvider = IB_PROVIDER_MLX5;
snprintf(dataDirectDevicePath, PATH_MAX, "/sys");
if((ncclMlx5dvDmaBufCapable(context)) && (wrap_mlx5dv_get_data_direct_sysfs_path(context, dataDirectDevicePath + 4, PATH_MAX - 4) == ncclSuccess)) {
INFO(NCCL_NET, "Data Direct DMA Interface is detected for device:%s", devices[d]->name);
if(ncclParamIbDataDirect()) dataDirectSupported = 1;
}
}
int nPorts = 0;
struct ibv_device_attr devAttr;
memset(&devAttr, 0, sizeof(devAttr));
@@ -609,58 +667,69 @@ ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction, ncclProfilerCallback_t pr
continue;
}
for (int port_num = 1; port_num <= devAttr.phys_port_cnt; port_num++) {
struct ibv_port_attr portAttr;
if (ncclSuccess != wrap_ibv_query_port(context, port_num, &portAttr)) {
WARN("NET/IB : Unable to query port_num %d", port_num);
continue;
for (int dataDirect = 0; dataDirect < 1 + dataDirectSupported; ++dataDirect) {
struct ibv_port_attr portAttr;
if (ncclSuccess != wrap_ibv_query_port(context, port_num, &portAttr)) {
WARN("NET/IB : Unable to query port_num %d", port_num);
continue;
}
if (portAttr.state != IBV_PORT_ACTIVE) continue;
if (portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND
&& portAttr.link_layer != IBV_LINK_LAYER_ETHERNET) continue;
// check against user specified HCAs/ports
if (! (matchIfList(devices[d]->name, port_num, userIfs, nUserIfs, searchExact) ^ searchNot)) {
continue;
}
pthread_mutex_init(&ncclIbDevs[ncclNIbDevs].lock, NULL);
ncclIbDevs[ncclNIbDevs].device = d;
ncclIbDevs[ncclNIbDevs].ibProvider = ibProvider;
ncclIbDevs[ncclNIbDevs].guid = devAttr.sys_image_guid;
ncclIbDevs[ncclNIbDevs].portAttr = portAttr;
ncclIbDevs[ncclNIbDevs].portNum = port_num;
ncclIbDevs[ncclNIbDevs].link = portAttr.link_layer;
ncclIbDevs[ncclNIbDevs].speed = ncclIbSpeed(portAttr.active_speed) * ncclIbWidth(portAttr.active_width);
ncclIbDevs[ncclNIbDevs].context = context;
ncclIbDevs[ncclNIbDevs].pdRefs = 0;
ncclIbDevs[ncclNIbDevs].pd = NULL;
if (!dataDirect) {
strncpy(ncclIbDevs[ncclNIbDevs].devName, devices[d]->name, MAXNAMESIZE);
NCCLCHECKGOTO(ncclIbGetPciPath(ncclIbDevs[ncclNIbDevs].devName, &ncclIbDevs[ncclNIbDevs].pciPath, &ncclIbDevs[ncclNIbDevs].realPort), ret, fail);
}
else {
snprintf(ncclIbDevs[ncclNIbDevs].devName, MAXNAMESIZE, "%s_dma", devices[d]->name);
NCCLCHECK(ncclCalloc(&ncclIbDevs[ncclNIbDevs].pciPath, PATH_MAX));
strncpy(ncclIbDevs[ncclNIbDevs].pciPath, dataDirectDevicePath, PATH_MAX);
ncclIbDevs[ncclNIbDevs].capsProvider.mlx5.dataDirect = 1;
}
ncclIbDevs[ncclNIbDevs].maxQp = devAttr.max_qp;
ncclIbDevs[ncclNIbDevs].mrCache.capacity = 0;
ncclIbDevs[ncclNIbDevs].mrCache.population = 0;
ncclIbDevs[ncclNIbDevs].mrCache.slots = NULL;
NCCLCHECK(ncclIbStatsInit(&ncclIbDevs[ncclNIbDevs].stats));
// Enable ADAPTIVE_ROUTING by default on IB networks
// But allow it to be overloaded by an env parameter
ncclIbDevs[ncclNIbDevs].ar = (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) ? 1 : 0;
if (ncclParamIbAdaptiveRouting() != -2) ncclIbDevs[ncclNIbDevs].ar = ncclParamIbAdaptiveRouting();
INFO(NCCL_NET,"NET/IB: [%d] %s:%s:%d/%s provider=%s speed=%d context=%p pciPath=%s ar=%d", d, devices[d]->name, devices[d]->dev_name, ncclIbDevs[ncclNIbDevs].portNum,
NCCL_IB_LLSTR(portAttr.link_layer), ibProviderName[ncclIbDevs[ncclNIbDevs].ibProvider], ncclIbDevs[ncclNIbDevs].speed, context, ncclIbDevs[ncclNIbDevs].pciPath, ncclIbDevs[ncclNIbDevs].ar);
PTHREADCHECKGOTO(pthread_create(&ncclIbAsyncThread, NULL, ncclIbAsyncThreadMain, ncclIbDevs + ncclNIbDevs), "pthread_create", ret, fail);
ncclSetThreadName(ncclIbAsyncThread, "NCCL IbAsync %2d", ncclNIbDevs);
PTHREADCHECKGOTO(pthread_detach(ncclIbAsyncThread), "pthread_detach", ret, fail); // will not be pthread_join()'d
// Add this plain physical device to the list of virtual devices
int vDev;
ncclNetVDeviceProps_t vProps = {0};
vProps.ndevs = 1;
vProps.devs[0] = ncclNIbDevs;
NCCLCHECK(ncclIbMakeVDeviceInternal(&vDev, &vProps));
ncclNIbDevs++;
nPorts++;
}
if (portAttr.state != IBV_PORT_ACTIVE) continue;
if (portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND
&& portAttr.link_layer != IBV_LINK_LAYER_ETHERNET) continue;
// check against user specified HCAs/ports
if (! (matchIfList(devices[d]->name, port_num, userIfs, nUserIfs, searchExact) ^ searchNot)) {
continue;
}
pthread_mutex_init(&ncclIbDevs[ncclNIbDevs].lock, NULL);
ncclIbDevs[ncclNIbDevs].device = d;
ncclIbDevs[ncclNIbDevs].guid = devAttr.sys_image_guid;
ncclIbDevs[ncclNIbDevs].portAttr = portAttr;
ncclIbDevs[ncclNIbDevs].portNum = port_num;
ncclIbDevs[ncclNIbDevs].link = portAttr.link_layer;
ncclIbDevs[ncclNIbDevs].speed = ncclIbSpeed(portAttr.active_speed) * ncclIbWidth(portAttr.active_width);
ncclIbDevs[ncclNIbDevs].context = context;
ncclIbDevs[ncclNIbDevs].pdRefs = 0;
ncclIbDevs[ncclNIbDevs].pd = NULL;
strncpy(ncclIbDevs[ncclNIbDevs].devName, devices[d]->name, MAXNAMESIZE);
NCCLCHECKGOTO(ncclIbGetPciPath(ncclIbDevs[ncclNIbDevs].devName, &ncclIbDevs[ncclNIbDevs].pciPath, &ncclIbDevs[ncclNIbDevs].realPort), ret, fail);
ncclIbDevs[ncclNIbDevs].maxQp = devAttr.max_qp;
ncclIbDevs[ncclNIbDevs].mrCache.capacity = 0;
ncclIbDevs[ncclNIbDevs].mrCache.population = 0;
ncclIbDevs[ncclNIbDevs].mrCache.slots = NULL;
NCCLCHECK(ncclIbStatsInit(&ncclIbDevs[ncclNIbDevs].stats));
// Enable ADAPTIVE_ROUTING by default on IB networks
// But allow it to be overloaded by an env parameter
ncclIbDevs[ncclNIbDevs].ar = (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) ? 1 : 0;
if (ncclParamIbAdaptiveRouting() != -2) ncclIbDevs[ncclNIbDevs].ar = ncclParamIbAdaptiveRouting();
TRACE(NCCL_NET,"NET/IB: [%d] %s:%s:%d/%s speed=%d context=%p pciPath=%s ar=%d", d, devices[d]->name, devices[d]->dev_name, ncclIbDevs[ncclNIbDevs].portNum,
NCCL_IB_LLSTR(portAttr.link_layer), ncclIbDevs[ncclNIbDevs].speed, context, ncclIbDevs[ncclNIbDevs].pciPath, ncclIbDevs[ncclNIbDevs].ar);
PTHREADCHECKGOTO(pthread_create(&ncclIbAsyncThread, NULL, ncclIbAsyncThreadMain, ncclIbDevs + ncclNIbDevs), "pthread_create", ret, fail);
ncclSetThreadName(ncclIbAsyncThread, "NCCL IbAsync %2d", ncclNIbDevs);
PTHREADCHECKGOTO(pthread_detach(ncclIbAsyncThread), "pthread_detach", ret, fail); // will not be pthread_join()'d
// Add this plain physical device to the list of virtual devices
int vDev;
ncclNetVDeviceProps_t vProps = {0};
vProps.ndevs = 1;
vProps.devs[0] = ncclNIbDevs;
NCCLCHECK(ncclIbMakeVDeviceInternal(&vDev, &vProps));
ncclNIbDevs++;
nPorts++;
}
if (nPorts == 0 && ncclSuccess != wrap_ibv_close_device(context)) { ret = ncclInternalError; goto fail; }
}
@@ -779,6 +848,9 @@ ncclResult_t ncclIbGetPhysProperties(int dev, ncclNetProperties_t* props) {
props->ptrSupport |= NCCL_PTR_DMABUF; // GDR support via DMA-BUF
}
props->forceFlush = 0;
if (ibDev->capsProvider.mlx5.dataDirect) {
props->forceFlush = 1;
}
props->latency = 0; // Not set
props->port = ibDev->portNum + ibDev->realPort;
props->maxComms = ibDev->maxQp;
@@ -893,6 +965,7 @@ struct ncclProfilerInfo {
int qpIndex[MAX_QPS_PER_REQ];
int nEventHandles;
ncclProfilerNetIbDescr_v1_t data;
void* pHandle;
};
struct ncclIbRequest {
@@ -1312,23 +1385,27 @@ ib_recv_dev_list:
devInfo->gid.global.interface_id = commDev->base.gidInfo.localGid.global.interface_id;
// info logging
if (devInfo->link_layer == IBV_LINK_LAYER_INFINIBAND) { // IB
for (int q = 0; q < comm->base.nqps; q++) {
// Print just the QPs for this dev
if (comm->base.qps[q].devIndex == i)
for (int q = 0; q < comm->base.nqps; q++) {
// Print just the QPs for this dev
if (comm->base.qps[q].devIndex == i) {
if (devInfo->link_layer == IBV_LINK_LAYER_INFINIBAND) { // IB
INFO(NCCL_NET,"NET/IB: %s %d IbDev %d Port %d qpn %d mtu %d LID %d subnet-prefix %lu FLID %d fifoRkey=0x%x fifoLkey=0x%x",
comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev",
dev, commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu, devInfo->lid,
devInfo->gid.global.subnet_prefix, ncclIbExtractFlid(&devInfo->gid), devInfo->fifoRkey, commDev->fifoMr->lkey);
}
} else { // RoCE
for (int q = 0; q < comm->base.nqps; q++) {
// Print just the QPs for this dev
if (comm->base.qps[q].devIndex == i)
INFO(NCCL_NET,"NET/IB: %s %d IbDev %d Port %d qpn %d mtu %d query_ece={supported=%d, vendor_id=0x%x, options=0x%x, comp_mask=0x%x} GID %ld (%lX/%lX) fifoRkey=0x%x fifoLkey=0x%x",
comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev", dev,
commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu, meta.qpInfo[q].ece_supported, meta.qpInfo[q].ece.vendor_id, meta.qpInfo[q].ece.options, meta.qpInfo[q].ece.comp_mask, (int64_t)commDev->base.gidInfo.localGidIndex,
devInfo->gid.global.subnet_prefix, devInfo->gid.global.interface_id, devInfo->fifoRkey, commDev->fifoMr->lkey);
comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev",
dev, commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu, devInfo->lid,
devInfo->gid.global.subnet_prefix, ncclIbExtractFlid(&devInfo->gid), devInfo->fifoRkey, commDev->fifoMr->lkey);
} else { // RoCE
INFO(NCCL_NET,"NET/IB: %s %d IbDev %d Port %d qpn %d mtu %d GID %ld (%lX/%lX) fifoRkey=0x%x fifoLkey=0x%x",
comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev", dev,
commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu,
(int64_t)commDev->base.gidInfo.localGidIndex,
devInfo->gid.global.subnet_prefix, devInfo->gid.global.interface_id, devInfo->fifoRkey, commDev->fifoMr->lkey);
}
// Log ECE info
if (meta.qpInfo[q].ece_supported) {
INFO(NCCL_NET,"NET/IB: IbDev %d Port %d qpn %d query_ece={supported=%d, vendor_id=0x%x, options=0x%x, comp_mask=0x%x}",
commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn,
meta.qpInfo[q].ece_supported, meta.qpInfo[q].ece.vendor_id, meta.qpInfo[q].ece.options, meta.qpInfo[q].ece.comp_mask);
}
}
}
if (link_layer == IBV_LINK_LAYER_UNSPECIFIED) link_layer = devInfo->link_layer;
@@ -1406,8 +1483,14 @@ ib_connect:
ncclIbSendCommDev* commDev = comm->devs + devIndex;
struct ibv_qp* qp = comm->base.qps[q].qp;
if (remQpInfo->ece_supported)
if (remQpInfo->ece_supported) {
struct ncclIbQp* nqp = comm->base.qps + q;
int ibDevN = comm->devs[nqp->devIndex].base.ibDevN;
struct ncclIbDev* ibDev = ncclIbDevs + ibDevN;
INFO(NCCL_NET,"NET/IB: IbDev %d Port %d qpn %d set_ece={supported=%d, vendor_id=0x%x, options=0x%x, comp_mask=0x%x}",
ibDevN, ibDev->portNum, qp->qp_num, remMeta.qpInfo[q].ece_supported, remMeta.qpInfo[q].ece.vendor_id, remMeta.qpInfo[q].ece.options, remMeta.qpInfo[q].ece.comp_mask);
NCCLCHECKGOTO(wrap_ibv_set_ece(qp, &remQpInfo->ece, &remQpInfo->ece_supported), ret, fail);
}
ncclIbDev* ibDev = ncclIbDevs + commDev->base.ibDevN;
remDevInfo->mtu = std::min(remDevInfo->mtu, ibDev->portAttr.active_mtu);
@@ -1415,16 +1498,6 @@ ib_connect:
NCCLCHECKGOTO(ncclIbRtsQp(qp), ret, fail);
}
if (link_layer == IBV_LINK_LAYER_ETHERNET ) { // RoCE
for (int q = 0; q < comm->base.nqps; q++) {
struct ncclIbQp* qp = comm->base.qps + q;
int ibDevN = comm->devs[qp->devIndex].base.ibDevN;
struct ncclIbDev* ibDev = ncclIbDevs + ibDevN;
INFO(NCCL_NET,"NET/IB: IbDev %d Port %d qpn %d set_ece={supported=%d, vendor_id=0x%x, options=0x%x, comp_mask=0x%x}",
ibDevN, ibDev->portNum, remMeta.qpInfo[q].qpn, remMeta.qpInfo[q].ece_supported, remMeta.qpInfo[q].ece.vendor_id, remMeta.qpInfo[q].ece.options, remMeta.qpInfo[q].ece.comp_mask);
}
}
comm->base.nDataQps = std::max(comm->base.vProps.ndevs, comm->base.nRemDevs);
comm->base.ready = 1;
@@ -1750,9 +1823,8 @@ ncclResult_t ncclIbGetRequest(struct ncclIbNetCommBase* base, struct ncclIbReque
if (r->type == NCCL_NET_IB_REQ_UNUSED) {
r->base = base;
r->sock = NULL;
r->devBases[0] = NULL;
r->devBases[1] = NULL;
r->events[0] = r->events[1] = 0;
memset(r->devBases, 0, sizeof(r->devBases));
memset(r->events, 0, sizeof(r->events));
*req = r;
return ncclSuccess;
}
@@ -1789,7 +1861,11 @@ ncclResult_t ncclIbRegMrDmaBufInternal(ncclIbNetCommDevBase* base, void* data, s
if (ncclIbRelaxedOrderingEnabled) flags |= IBV_ACCESS_RELAXED_ORDERING;
if (fd != -1) {
/* DMA-BUF support */
NCCLCHECKGOTO(wrap_ibv_reg_dmabuf_mr(&mr, base->pd, offset, pages*pageSize, addr, fd, flags), res, returning);
if (!ncclIbDevs[base->ibDevN].capsProvider.mlx5.dataDirect) {
NCCLCHECKGOTO(wrap_ibv_reg_dmabuf_mr(&mr, base->pd, offset, pages*pageSize, addr, fd, flags), res, returning);
} else {
NCCLCHECKGOTO(wrap_mlx5dv_reg_dmabuf_mr(&mr, base->pd, offset, pages*pageSize, addr, fd, flags, MLX5DV_REG_DMABUF_ACCESS_DATA_DIRECT), res, returning);
}
} else {
if (ncclIbRelaxedOrderingEnabled) {
// Use IBVERBS_1.8 API - needed for IBV_ACCESS_RELAXED_ORDERING support
@@ -1897,7 +1973,7 @@ ncclResult_t ncclIbDeregMr(void* comm, void* mhandle) {
NCCL_PARAM(IbSplitDataOnQps, "IB_SPLIT_DATA_ON_QPS", 0);
ncclResult_t ncclIbMultiSend(struct ncclIbSendComm* comm, int slot, void* pHandle) {
ncclResult_t ncclIbMultiSend(struct ncclIbSendComm* comm, int slot) {
struct ncclIbRequest** reqs = comm->fifoReqs[slot];
volatile struct ncclIbSendFifo* slots = comm->fifo[slot];
int nreqs = slots[0].nreqs;
@@ -1989,19 +2065,21 @@ ncclResult_t ncclIbMultiSend(struct ncclIbSendComm* comm, int slot, void* pHandl
struct ibv_send_wr* bad_wr;
#ifdef NCCL_ENABLE_NET_PROFILING
// QP profiling loop
for (int r=0; r<nreqs && pHandle; r++) {
for (int r=0; r<nreqs; r++) {
// Store comm qpIndex for this request
int nEventHandles = reqs[r]->pInfo[0].nEventHandles;
reqs[r]->pInfo[0].qpIndex[nEventHandles%MAX_QPS_PER_REQ] = qpIndex;
assert(nEventHandles < MAX_QPS_PER_REQ);
reqs[r]->pInfo[0].qpIndex[nEventHandles] = qpIndex;
// Store info for profiler
int pluginId = NCCL_PROFILER_NET_TYPE_IB | NCCL_PROFILER_NET_IB_VER;
int64_t pluginId = NCCL_PROFILER_NET_TYPE_IB | NCCL_PROFILER_NET_IB_VER;
reqs[r]->pInfo[0].data.type = ncclProfileQp;
reqs[r]->pInfo[0].data.qp.device = devIndex;
reqs[r]->pInfo[0].data.qp.wr_id = comm->wrs[r].wr_id;
reqs[r]->pInfo[0].data.qp.opcode = comm->wrs[r].opcode;
reqs[r]->pInfo[0].data.qp.qpNum = qp->qp->qp_num;
reqs[r]->pInfo[0].data.qp.length = comm->sges[r].length;
NCCLCHECK(ncclProfilerFunction(&reqs[r]->pInfo[0].qpEventHandles[nEventHandles%MAX_QPS_PER_REQ], 0, pHandle, pluginId, &reqs[r]->pInfo[0].data));
void* pHandle = reqs[r]->pInfo[0].pHandle;
NCCLCHECK(ncclProfilerFunction(&reqs[r]->pInfo[0].qpEventHandles[nEventHandles], ncclProfilerNetEventStart, pHandle, pluginId, &reqs[r]->pInfo[0].data));
reqs[r]->pInfo[0].nEventHandles++;
}
#endif
@@ -2023,8 +2101,11 @@ ncclResult_t ncclIbMultiSend(struct ncclIbSendComm* comm, int slot, void* pHandl
ncclResult_t ncclIbIsend(void* sendComm, void* data, size_t size, int tag, void* mhandle, void* phandle, void** request) {
struct ncclIbSendComm* comm = (struct ncclIbSendComm*)sendComm;
if (comm->base.ready == 0) { WARN("NET/IB: ncclIbIsend() called when comm->base.ready == 0"); return ncclInternalError; }
if (comm->base.ready == 0) { *request = NULL; return ncclSuccess; }
if (comm->base.ready == 0) {
WARN("NET/IB: ncclIbIsend() called when comm->base.ready == 0");
*request = NULL;
return ncclInternalError;
}
NCCLCHECK(ncclIbStatsCheckFatalCount(&comm->base.stats,__func__));
struct ncclIbMrHandle* mhandleWrapper = (struct ncclIbMrHandle*) mhandle;
@@ -2065,6 +2146,9 @@ ncclResult_t ncclIbIsend(void* sendComm, void* data, size_t size, int tag, void*
req->send.size = size;
req->send.data = data;
req->send.offset = 0;
#ifdef NCCL_ENABLE_NET_PROFILING
req->pInfo[0].pHandle = phandle;
#endif
// Populate events
int nEvents = ncclParamIbSplitDataOnQps() ? comm->base.nqps : comm->base.nDataQps;
@@ -2094,7 +2178,7 @@ ncclResult_t ncclIbIsend(void* sendComm, void* data, size_t size, int tag, void*
}
TIME_START(0);
NCCLCHECK(ncclIbMultiSend(comm, slot, phandle));
NCCLCHECK(ncclIbMultiSend(comm, slot));
// Clear slots[0]->nreqs, as well as other fields to help debugging and sanity checks
memset((void*)slots, 0, sizeof(struct ncclIbSendFifo));
@@ -2187,8 +2271,11 @@ ncclResult_t ncclIbPostFifo(struct ncclIbRecvComm* comm, int n, void** data, siz
ncclResult_t ncclIbIrecv(void* recvComm, int n, void** data, size_t* sizes, int* tags, void** mhandles, void** phandles, void** request) {
struct ncclIbRecvComm* comm = (struct ncclIbRecvComm*)recvComm;
if (comm->base.ready == 0) { WARN("NET/IB: ncclIbIrecv() called when comm->base.ready == 0"); return ncclInternalError; }
if (comm->base.ready == 0) { *request = NULL; return ncclSuccess; }
if (comm->base.ready == 0) {
WARN("NET/IB: ncclIbIrecv() called when comm->base.ready == 0");
*request = NULL;
return ncclInternalError;
}
if (n > NCCL_NET_IB_MAX_RECVS) return ncclInternalError;
NCCLCHECK(ncclIbStatsCheckFatalCount(&comm->base.stats,__func__));
@@ -2222,14 +2309,17 @@ ncclResult_t ncclIbIrecv(void* recvComm, int n, void** data, size_t* sizes, int*
ncclIbAddEvent(req, qp->devIndex, &comm->devs[qp->devIndex].base);
#ifdef NCCL_ENABLE_NET_PROFILING
// Start a QP event for every request in the multirecv and every qp
for (int r = 0; r < n && phandles; r++) {
for (int r = 0; r < n; r++) {
int nEventHandles = req->pInfo[r].nEventHandles;
assert(nEventHandles < MAX_QPS_PER_REQ);
req->pInfo[r].qpIndex[nEventHandles] = comm->base.qpIndex;
// Store info for profiler
int pluginId = NCCL_PROFILER_NET_TYPE_IB | NCCL_PROFILER_NET_IB_VER;
int64_t pluginId = NCCL_PROFILER_NET_TYPE_IB | NCCL_PROFILER_NET_IB_VER;
req->pInfo[r].data.type = ncclProfileQp;
req->pInfo[r].data.qp.device = qp->devIndex;
req->pInfo[r].data.qp.wr_id = wr.wr_id;
req->pInfo[r].data.qp.qpNum = qp->qp->qp_num;
NCCLCHECK(ncclProfilerFunction(&req->pInfo[r].qpEventHandles[i], 0, phandles[r], pluginId, &req->pInfo[r].data));
NCCLCHECK(ncclProfilerFunction(&req->pInfo[r].qpEventHandles[nEventHandles], ncclProfilerNetEventStart, phandles[r], pluginId, &req->pInfo[r].data));
req->pInfo[r].nEventHandles++;
}
#endif
@@ -2311,7 +2401,7 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) {
sizes[i] = r->recv.sizes[i];
#ifdef NCCL_ENABLE_NET_PROFILING
for (int j = 0; j < r->pInfo[i].nEventHandles; j++) {
NCCLCHECK(ncclProfilerFunction(&r->pInfo[i].qpEventHandles[j], 1, NULL, 0, NULL));
NCCLCHECK(ncclProfilerFunction(&r->pInfo[i].qpEventHandles[j], ncclProfilerNetEventStop, NULL, 0, NULL));
}
#endif
}
@@ -2320,7 +2410,7 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) {
sizes[0] = r->send.size;
#ifdef NCCL_ENABLE_NET_PROFILING
for (int j = 0; j < r->pInfo[0].nEventHandles; j++) {
NCCLCHECK(ncclProfilerFunction(&r->pInfo[0].qpEventHandles[j], 1, NULL, 0, NULL));
NCCLCHECK(ncclProfilerFunction(&r->pInfo[0].qpEventHandles[j], ncclProfilerNetEventStop, NULL, 0, NULL));
}
#endif
}
@@ -2368,20 +2458,21 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) {
#ifdef ENABLE_TRACE
char line[SOCKET_NAME_MAXLEN+1];
TRACE(NCCL_NET, "Got completion from peer %s with status=%d opcode=%d len=%u wr_id=%lu r=%p type=%d events={%d,%d}, i=%d",
ncclSocketToString(&addr, line), wc->status, wc->opcode,wc->byte_len, wc->wr_id, req, req->type, req->events[0], req->events[1], i);
TRACE(NCCL_NET, "Got completion from peer %s with status=%d opcode=%d len=%u wr_id=%lu r=%p type=%d events={%d,%d,%d,%d}, i=%d",
ncclSocketToString(&addr, line), wc->status, wc->opcode,wc->byte_len, wc->wr_id, req, req->type, req->events[0], req->events[1], req->events[2], req->events[3], i);
#endif
if (req && req->type == NCCL_NET_IB_REQ_SEND) {
for (int j = 0; j < req->nreqs; j++) {
struct ncclIbRequest* sendReq = r->base->reqs+((wc->wr_id >> (j*8)) & 0xff);
if ((sendReq->events[i] <= 0)) {
WARN("NET/IB: sendReq(%p)->events={%d,%d}, i=%d, j=%d <= 0", sendReq, sendReq->events[0], sendReq->events[1], i, j);
WARN("NET/IB: sendReq(%p)->events={%d,%d,%d,%d}, i=%d, j=%d <= 0", sendReq, sendReq->events[0], sendReq->events[1], sendReq->events[2], sendReq->events[3], i, j);
return ncclInternalError;
}
sendReq->events[i]--;
#ifdef NCCL_ENABLE_NET_PROFILING
// Stop Qp event for sendReq
NCCLCHECK(ncclProfilerFunction(&sendReq->pInfo[j].qpEventHandles[getReqQpIndex(sendReq, j, wc->qp_num)], 1, NULL, 0, NULL));
int qpIndex = getReqQpIndex(sendReq, j, wc->qp_num);
NCCLCHECK(ncclProfilerFunction(&sendReq->pInfo[j].qpEventHandles[qpIndex], ncclProfilerNetEventStop, NULL, 0, NULL));
#endif
}
} else {
@@ -2398,7 +2489,8 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) {
#ifdef NCCL_ENABLE_NET_PROFILING
// Stop Qp event for workFifo
for (int j = 0; j < req->nreqs; j++) {
NCCLCHECK(ncclProfilerFunction(&req->pInfo[j].qpEventHandles[getReqQpIndex(req, j, wc->qp_num)], 1, NULL, 0, NULL));
int qpIndex = getReqQpIndex(req, j, wc->qp_num);
NCCLCHECK(ncclProfilerFunction(&req->pInfo[j].qpEventHandles[qpIndex], ncclProfilerNetEventStop, NULL, 0, NULL));
}
#endif
}
src/transport/net_socket.cc
+64 -35
View File
@@ -45,7 +45,7 @@ ncclResult_t ncclNetSocketInit(ncclDebugLogger_t logFunction, ncclProfilerCallba
if (ncclNetIfs == -1) {
char names[MAX_IF_NAME_SIZE*MAX_IFS];
union ncclSocketAddress addrs[MAX_IFS];
ncclNetIfs = ncclFindInterfaces(names, addrs, MAX_IF_NAME_SIZE, MAX_IFS);
NCCLCHECK(ncclFindInterfaces(names, addrs, MAX_IF_NAME_SIZE, MAX_IFS, &ncclNetIfs));
if (ncclNetIfs <= 0) {
WARN("NET/Socket : no interface found");
pthread_mutex_unlock(&ncclNetSocketLock);
@@ -124,8 +124,9 @@ ncclResult_t ncclNetSocketGetProperties(int dev, ncclNetProperties_t* props) {
#define MAX_SOCKETS 64
#define MAX_THREADS 16
#define MAX_REQUESTS NCCL_NET_MAX_REQUESTS
#define MIN_CHUNKSIZE (64*1024)
NCCL_PARAM(SocketInlineSize, "SOCKET_INLINE", /*128 B=*/1 << 7);
NCCL_PARAM(SocketMinTaskSize, "SOCKET_MIN_TASKSIZE", /*64 kiB=*/1 << 16);
NCCL_PARAM(SocketNsocksPerThread, "NSOCKS_PERTHREAD", -2);
NCCL_PARAM(SocketNthreads, "SOCKET_NTHREADS", -2);
@@ -171,6 +172,7 @@ struct ncclNetSocketRequest {
int op;
void* data;
int size;
void* inlineData;
struct ncclSocket* ctrlSock;
int offset;
int used;
@@ -211,6 +213,7 @@ struct ncclNetSocketComm {
int nSocks;
int nThreads;
int nextSock;
void* inlineData;
struct ncclNetSocketRequest requests[MAX_REQUESTS];
pthread_t helperThread[MAX_THREADS];
struct ncclNetSocketThreadResources threadResources[MAX_THREADS];
@@ -241,13 +244,13 @@ void* persistentSocketThread(void *args_) {
data.sock.fd = r->sock->fd;
data.sock.op = r->op;
data.sock.length = r->size;
ncclProfilerFunction(&eHandle[i+j], 0, resource->pInfo->pHandle, NCCL_PROFILER_NET_TYPE_SOCK | 1, &data);
ncclProfilerFunction(&eHandle[i+j], ncclProfilerNetEventStart, resource->pInfo->pHandle, NCCL_PROFILER_NET_TYPE_SOCK | 1, &data);
}
#endif
r->result = ncclSocketProgress(r->op, r->sock, r->data, r->size, &r->offset);
if (r->result != ncclSuccess) {
#ifdef NCCL_ENABLE_NET_PROFILING
ncclProfilerFunction(&eHandle[i+j], 1, NULL, 0, NULL);
ncclProfilerFunction(&eHandle[i+j], ncclProfilerNetEventStop, NULL, 0, NULL);
eHandle[i+j] = NULL;
#endif
WARN("NET/Socket : socket progress error");
@@ -257,7 +260,7 @@ void* persistentSocketThread(void *args_) {
if (r->offset < r->size) repeat = 1;
#ifdef NCCL_ENABLE_NET_PROFILING
if (repeat == 0) {
ncclProfilerFunction(&eHandle[i+j], 1, NULL, 0, NULL);
ncclProfilerFunction(&eHandle[i+j], ncclProfilerNetEventStop, NULL, 0, NULL);
eHandle[i+j] = NULL;
}
#endif
@@ -360,6 +363,7 @@ fail:
goto exit;
}
#define SOCKET_CTRL_SIZE (sizeof(int))
ncclResult_t ncclNetSocketConnect(int dev, ncclNetCommConfig_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** /*sendDevComm*/) {
if (dev < 0 || dev >= ncclNetIfs) { // data transfer socket is based on specified dev
return ncclInternalError;
@@ -401,6 +405,7 @@ socket_send:
NCCLCHECK(ncclSocketProgress(NCCL_SOCKET_SEND, sock, &i, sizeof(uint8_t), &done));
if (done == 0) return ncclSuccess;
}
NCCLCHECK(ncclCalloc(&comm->inlineData, MAX_REQUESTS * (SOCKET_CTRL_SIZE + ncclParamSocketInlineSize())));
*sendComm = comm;
return ncclSuccess;
}
@@ -449,6 +454,7 @@ socket_recv:
memcpy(rComm->socks+sendSockIdx, sock, sizeof(struct ncclSocket));
free(sock);
}
NCCLCHECK(ncclCalloc(&rComm->inlineData, MAX_REQUESTS * (SOCKET_CTRL_SIZE + ncclParamSocketInlineSize())));
*recvComm = rComm;
/* reset lComm state */
@@ -470,6 +476,7 @@ ncclResult_t ncclNetSocketGetRequest(struct ncclNetSocketComm* comm, int op, voi
r->used = 1;
r->comm = comm;
r->nSubs = 0;
r->inlineData = (uint8_t*)comm->inlineData + i * (SOCKET_CTRL_SIZE + ncclParamSocketInlineSize());
*req = r;
return ncclSuccess;
}
@@ -520,6 +527,9 @@ ncclResult_t ncclNetSocketGetTask(struct ncclNetSocketComm* comm, struct ncclPro
return ncclInternalError;
}
// if the dataSize is smaller than the inline size, return the inline size; if not, return 0 to avoid the extra copy.
static int ncclNetSocketInlineSize(int dataSize) { return (dataSize <= ncclParamSocketInlineSize()) ? dataSize : 0; }
ncclResult_t ncclNetSocketTest(void* request, int* done, int* size) {
*done = 0;
struct ncclNetSocketRequest *r = (struct ncclNetSocketRequest*)request;
@@ -527,37 +537,55 @@ ncclResult_t ncclNetSocketTest(void* request, int* done, int* size) {
WARN("NET/Socket : test called with NULL request");
return ncclInternalError;
}
if (r->used == 1) { /* try to send/recv size */
int data = r->size;
int offset = 0;
NCCLCHECK(ncclSocketProgress(r->op, r->ctrlSock, &data, sizeof(int), &offset));
if (offset == 0) return ncclSuccess; /* Not ready -- retry later */
// Not sure we could ever receive less than 4 bytes, but just in case ...
if (offset < sizeof(int)) NCCLCHECK(ncclSocketWait(r->op, r->ctrlSock, &data, sizeof(int), &offset));
// Check size is less or equal to the size provided by the user
if (r->op == NCCL_SOCKET_RECV && data > r->size) {
char line[SOCKET_NAME_MAXLEN+1];
union ncclSocketAddress addr;
NCCLCHECK(ncclSocketGetAddr(r->ctrlSock, &addr));
WARN("NET/Socket : peer %s message truncated : receiving %d bytes instead of %d. If you believe your socket network is in healthy state, \
there may be a mismatch in collective sizes or environment settings (e.g. NCCL_PROTO, NCCL_ALGO) between ranks",
ncclSocketToString(&addr, line), data, r->size);
return ncclInvalidUsage;
if (r->used == 1) { /* try to send/recv size (+ inline data if any) */
int msgSize;
uint8_t* msg = (uint8_t*)r->inlineData;
if (r->op == NCCL_SOCKET_SEND) {
// sender side has the right data size, copy size info + inline data to the buffer
int inlineSize = ncclNetSocketInlineSize(r->size);
msgSize = inlineSize + SOCKET_CTRL_SIZE;
memcpy(msg, &r->size, SOCKET_CTRL_SIZE);
if (inlineSize > 0) memcpy(msg + SOCKET_CTRL_SIZE, r->data, inlineSize);
} else {
// receiver side doesn't have the right data size, wait for the sender to send it
int sizeOffset = 0, senderSize = 0;
while (sizeOffset < SOCKET_CTRL_SIZE) {
NCCLCHECK(ncclSocketProgress(r->op, r->ctrlSock, msg, SOCKET_CTRL_SIZE, &sizeOffset));
if (sizeOffset == 0) return ncclSuccess; /* not ready yet*/
}
memcpy(&senderSize, msg, SOCKET_CTRL_SIZE);
if (senderSize > r->size) {
char line[SOCKET_NAME_MAXLEN + 1];
union ncclSocketAddress addr;
NCCLCHECK(ncclSocketGetAddr(r->ctrlSock, &addr));
WARN("NET/Socket : peer %s message truncated : receiving %d bytes instead of %d. If you believe your socket network is in a healthy state, "
"there may be a mismatch in collective sizes or environment settings (e.g. NCCL_PROTO, NCCL_ALGO) between ranks",
ncclSocketToString(&addr, line), senderSize, r->size);
return ncclInvalidUsage;
}
// copy to the data buffer if we have received some inline data already
int receivedInline = sizeOffset - SOCKET_CTRL_SIZE;
if (receivedInline > 0) memcpy(r->data, msg + SOCKET_CTRL_SIZE, receivedInline);
// from the actual size, extract the remaining inline size to be received and redirect the msg buffer to the user data
r->size = senderSize;
msgSize = ncclNetSocketInlineSize(r->size) - receivedInline;
msg = (uint8_t*)r->data + receivedInline;
}
r->size = data;
r->offset = 0;
r->used = 2; // done exchanging size
// divide into subtasks
int chunkOffset = 0, i = 0;
int offset = 0;
while (offset < msgSize) {
NCCLCHECK(ncclSocketProgress(r->op, r->ctrlSock, msg, msgSize, &offset));
if (offset == 0) return ncclSuccess; /* not ready yet*/
}
// done exchanging sizes, r->size now contains the actual size
r->used = 2;
r->offset = ncclNetSocketInlineSize(r->size);
int chunkOffset = r->offset, i = 0;
if (r->comm->nSocks > 0) {
// each request can be divided up to nSocks tasks
int taskSize = std::max(MIN_CHUNKSIZE, DIVUP(r->size, r->comm->nSocks));
// each request can be divided up to nSocks tasks, we use the size left to transfer
int taskSize = std::max((int)ncclParamSocketMinTaskSize(), DIVUP(r->size - r->offset, r->comm->nSocks));
while (chunkOffset < r->size) {
int chunkSize = std::min(taskSize, r->size-chunkOffset);
NCCLCHECK(ncclNetSocketGetTask(r->comm, &r->pInfo, r->op, (char*)(r->data)+chunkOffset, chunkSize, r->tasks+i++));
int chunkSize = std::min(taskSize, r->size - chunkOffset);
NCCLCHECK(ncclNetSocketGetTask(r->comm, &r->pInfo, r->op, (char*)(r->data) + chunkOffset, chunkSize, r->tasks + i++));
chunkOffset += chunkSize;
}
}
@@ -588,7 +616,7 @@ ncclResult_t ncclNetSocketTest(void* request, int* done, int* size) {
data.sock.fd = r->ctrlSock->fd;
data.sock.op = r->op;
data.sock.length = r->size;
ncclProfilerFunction(&r->pInfo.eHandle, 0, r->pInfo.pHandle, NCCL_PROFILER_NET_TYPE_SOCK | 1, &data);
ncclProfilerFunction(&r->pInfo.eHandle, ncclProfilerNetEventStart, r->pInfo.pHandle, NCCL_PROFILER_NET_TYPE_SOCK | 1, &data);
}
#endif
if (r->offset < r->size) {
@@ -599,7 +627,7 @@ ncclResult_t ncclNetSocketTest(void* request, int* done, int* size) {
*done = 1;
r->used = 0;
#ifdef NCCL_ENABLE_NET_PROFILING
ncclProfilerFunction(&r->pInfo.eHandle, 1, NULL, 0, NULL);
ncclProfilerFunction(&r->pInfo.eHandle, ncclProfilerNetEventStop, NULL, 0, NULL);
r->pInfo.eHandle = NULL;
#endif
}
@@ -673,6 +701,7 @@ ncclResult_t ncclNetSocketClose(void* opaqueComm) {
NCCLCHECK(ncclSocketReady(&comm->socks[i], &ready));
if (ready) NCCLCHECK(ncclSocketClose(&comm->socks[i]));
}
if(comm->inlineData) free(comm->inlineData);
free(comm);
}
return ncclSuccess;
src/transport/nvls.cc
+211 -49
View File
@@ -13,6 +13,7 @@
#include "enqueue.h"
#include "register.h"
#include "transport.h"
#include "register_inline.h"
#if CUDART_VERSION >= 12010
@@ -109,7 +110,9 @@ ncclResult_t nvlsGroupUnbind(struct ncclComm *comm, size_t size, CUmemGenericAll
}
ncclResult_t ncclNvlsDeregBuffer(struct ncclComm* comm, CUmemGenericAllocationHandle *mcHandler, CUdeviceptr ptr, int dev, size_t ucsize, size_t mcsize) {
CUCHECK(cuMulticastUnbind(*mcHandler, dev, 0/*mcOffset*/, ucsize));
// unbind can trigger RM error if buffer is freed already by users
// however, it is safe to ignore the error, and unbind will succeed anyway
CUCALL(cuMulticastUnbind(*mcHandler, dev, 0/*mcOffset*/, ucsize));
CUCHECK(cuMemUnmap(ptr, mcsize));
CUCHECK(cuMemAddressFree(ptr, mcsize));
CUCHECK(cuMemRelease(*mcHandler));
@@ -143,9 +146,9 @@ ncclResult_t nvlsGroupUnmapMem(struct ncclComm *comm, size_t ucsize, void* ucptr
#define NVLS_MEM_ALIGN_SIZE (1 << 21)
#define NVLS_NCHANNELS_SM90 16
#define NVLS_NCHANNELS_SM100 32
#define NVLS_NCHANNELS_SM100_NVL 24
NCCL_PARAM(NvlsEnable, "NVLS_ENABLE", 2);
NCCL_PARAM(NvlsChannels, "NVLS_NCHANNELS", -2);
NCCL_PARAM(NvlsChunkSize, "NVLS_CHUNKSIZE", 128*1024);
ncclResult_t ncclNvlsInit(struct ncclComm* comm) {
@@ -171,12 +174,31 @@ ncclResult_t ncclNvlsInit(struct ncclComm* comm) {
comm->nvlsSupport = 1;
}
INFO(NCCL_INIT, "NVLS multicast support is %savailable on dev %d", comm->nvlsSupport ? "" : "not ", dev);
if (comm->nvlsSupport) {
int channels = (comm->compCap >= 100) ? NVLS_NCHANNELS_SM100 : NVLS_NCHANNELS_SM90;
if (ncclParamNvlsChannels() >= 0) channels = ncclParamNvlsChannels();
int channels;
if (comm->compCap >= 100) {
// Use a reduced number of channels for single node/MNNVL domain on Blackwell.
// comm->nNodes is not yet initialized at this point so we need to use other data.
bool multiNode;
if (comm->MNNVL) {
multiNode = (comm->clique.size < comm->nRanks);
} else {
int i;
for (i = 1; i < comm->nRanks; i++) {
if (comm->peerInfo[i].hostHash != comm->peerInfo[0].hostHash)
break;
}
multiNode = (i < comm->nRanks);
}
channels = (multiNode ? NVLS_NCHANNELS_SM100 : NVLS_NCHANNELS_SM100_NVL);
} else {
channels = NVLS_NCHANNELS_SM90;
}
if (comm->config.nvlsCTAs != NCCL_CONFIG_UNDEF_INT) channels = comm->config.nvlsCTAs;
comm->nvlsChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, channels));
}
INFO(NCCL_INIT, "NVLS multicast support is %savailable on dev %d (NVLS_NCHANNELS %d)",
comm->nvlsSupport ? "" : "not ", dev, comm->nvlsChannels);
return ncclSuccess;
}
@@ -242,16 +264,33 @@ static ncclResult_t nvlsAllocateMem(struct ncclComm* comm, const CUmemAccessDesc
CUCHECKGOTO(cuMemAddressReserve((CUdeviceptr*)ucptr, ucsize, ucgran, 0U, 0), ret, fail);
// Alloc local physical mem for this NVLS group
CUCHECKGOTO(cuMemCreate(ucHandle, ucsize, &ucprop, 0), ret, fail);
CUCHECKGOTO(cuMemMap((CUdeviceptr)*ucptr, ucsize, 0, *ucHandle, 0), ret, fail);
CUCHECKGOTO(cuMemSetAccess((CUdeviceptr)*ucptr, ucsize, desc, 1), ret, fail);
CUDACHECKGOTO(cudaMemset(*ucptr, 0, ucsize), ret, fail);
CUCHECKGOTO(cuMemCreate(ucHandle, ucsize, &ucprop, 0), ret, fail1);
CUCHECKGOTO(cuMemMap((CUdeviceptr)*ucptr, ucsize, 0, *ucHandle, 0), ret, fail2);
CUCHECKGOTO(cuMemSetAccess((CUdeviceptr)*ucptr, ucsize, desc, 1), ret, fail3);
CUDACHECKGOTO(cudaMemset(*ucptr, 0, ucsize), ret, fail3);
// intra-node barrier to mitigate the possible hang in cuMulticastBindMem during abort
NCCLCHECKGOTO(bootstrapIntraNodeBarrier(comm->bootstrap, comm->localRankToRank, comm->localRank, comm->localRanks, comm->localRankToRank[0]), ret, fail);
NCCLCHECKGOTO(bootstrapIntraNodeBarrier(comm->bootstrap, comm->localRankToRank, comm->localRank, comm->localRanks, comm->localRankToRank[0]), ret, fail3);
// Bind physical memory to the Multicast group
// NB: It will block until all ranks have been added to the Group
CUCHECKGOTO(cuMulticastBindMem(*mcHandle, 0/*mcOffset*/, *ucHandle, 0/*memOffset*/, ucsize, 0/*flags*/), ret, fail);
// This is where we normally see issues if the system NVLS/Multicast support is broken
{
CUresult err = CUPFN(cuMulticastBindMem(*mcHandle, 0/*mcOffset*/, *ucHandle, 0/*memOffset*/, ucsize, 0/*flags*/));
if (err != CUDA_SUCCESS) {
const char *errStr; \
(void) pfn_cuGetErrorString(err, &errStr); \
if (ncclParamNvlsEnable() == 1) {
// Fail the job as NVLS support is not available
WARN("Failed to bind NVLink SHARP (NVLS) Multicast memory of size %ld : CUDA error %d '%s'.\nThis is usually caused by a system or configuration error in the Fabric Manager or NVSwitches.\nDo not force-enable NVLS (NCCL_NVLS_ENABLE=1) if you wish to avoid this error in the future.", ucsize, err, errStr );
ret = ncclUnhandledCudaError;
} else {
// Continue without NVLS support (returns ncclSuccess)
INFO(NCCL_INIT|NCCL_NVLS, "Failed to bind NVLink SHARP (NVLS) Multicast memory of size %ld : CUDA error %d '%s'. Proceeding without NVLS support.", ucsize, err, errStr);
}
comm->nvlsSupport = comm->nvlsChannels = 0;
goto fail3;
}
}
// Map mc virtual address
CUCHECKGOTO(cuMemAddressReserve((CUdeviceptr*)mcptr, mcsize, mcgran, 0U, 0), ret, fail);
@@ -263,6 +302,12 @@ static ncclResult_t nvlsAllocateMem(struct ncclComm* comm, const CUmemAccessDesc
exit:
return ret;
fail3:
CUCHECK(cuMemUnmap((CUdeviceptr)*ucptr, ucsize));
fail2:
CUCHECK(cuMemRelease(*ucHandle));
fail1:
CUCHECK(cuMemAddressFree((CUdeviceptr)*ucptr, ucsize));
fail:
if (allocMcHandle && *mcptr == NULL && *ucptr == NULL) CUCHECK(cuMemRelease(*mcHandle));
goto exit;
@@ -291,8 +336,8 @@ ncclResult_t ncclNvlsBufferSetup(struct ncclComm* comm) {
nvlsPerRankSize = nChannels * 2 * buffSize;
nvlsTotalSize = nvlsPerRankSize * nHeads;
INFO(NCCL_INIT | NCCL_NVLS, "NVLS comm %p headRank %d nHeads %d buffSize %zu nvlsPerRankSize %zu nvlsTotalSize %zu",
comm, headRank, nHeads, buffSize, nvlsPerRankSize, nvlsTotalSize);
INFO(NCCL_INIT | NCCL_NVLS, "NVLS comm %p headRank %d nHeads %d nvlsRanks %d buffSize %zu nvlsPerRankSize %zu nvlsTotalSize %zu",
comm, headRank, nHeads, comm->localRanks, buffSize, nvlsPerRankSize, nvlsTotalSize);
NCCLCHECKGOTO(nvlsAllocateMem(comm, &resources->accessDesc, nvlsTotalSize, &resources->ucBuffHandle, &resources->mcBuffHandle, (void**)&resources->ucBuff, (void**)&resources->mcBuff, &resources->buffUCSize, &resources->buffMCSize), res, fail);
@@ -338,32 +383,10 @@ ncclResult_t ncclNvlsSetup(struct ncclComm* comm, struct ncclComm* parent) {
size_t typeSize;
char shmPath[sizeof("/dev/shm/nccl-XXXXXX")];
uintptr_t *nvlsShmem = NULL;
bool nvlsShare = parent && parent->nvlsSupport && parent->config.splitShare;
int nHeads = comm->channels[0].nvls.nHeads;
bool nvlsShare = parent && parent->nvlsSupport && parent->shareResources && parent->localRanks == comm->localRanks;
if (comm->nvlsSupport == 0 || comm->nvlsChannels == 0) return ncclSuccess;
if (nvlsShare && parent->channels[0].nvls.nHeads == nHeads) {
for (int ch = 0; ch < nHeads; ++ch) {
bool find = false;
for (int h = 0; h < parent->channels[0].nvls.nHeads; ++h) {
if (comm->nvlsHeads[ch] == parent->nvlsHeads[h]) {
// find the head
find = true;
break;
}
}
if (find == false) {
nvlsShare = false;
goto setup;
}
}
nvlsShare = true;
} else {
nvlsShare = false;
}
setup:
comm->nvlsChunkSize = ncclParamNvlsChunkSize();
if (nvlsShare) {
/* reuse NVLS resources */
@@ -387,9 +410,10 @@ setup:
comm->nvlsResources->inited = false;
comm->nvlsResources->refCount = 1;
comm->nvlsResources->nChannels = comm->nvlsChannels;
comm->nvlsResources->nHeads = nHeads;
resources = comm->nvlsResources;
if (parent && parent->nvlsSupport && parent->config.splitShare) {
if (parent && parent->nvlsSupport && parent->shareResources) {
/* ranks on other nodes might share the NVLS resources, we need to cap nvlsChannels
* to make sure nvlsChannels match for each rank. */
comm->nvlsChannels = std::min(comm->nvlsChannels, parent->nvlsResources->nChannels);
@@ -529,9 +553,9 @@ ncclResult_t tryRegisterBuffer(struct ncclComm *comm, uintptr_t userBuff, size_t
if (userBuff) {
NCCLCHECKGOTO(ncclRegFind(comm, (void*)userBuff, buffSize, &regRecord), ret, fail);
if (regRecord) {
CUDACHECKGOTO(cudaPointerGetAttributes(&attr, (void*)regRecord->addr), ret, fail);
CUDACHECKGOTO(cudaPointerGetAttributes(&attr, (void*)regRecord->begAddr), ret, fail);
if (attr.type == cudaMemoryTypeDevice) {
size_t regSize = regRecord->pages * comm->regCache.pageSize;
size_t regSize = regRecord->endAddr - regRecord->begAddr;
memset(&mcprop, 0, sizeof(CUmulticastObjectProp));
mcprop.numDevices = comm->localRanks;
mcprop.handleTypes = ncclCuMemHandleType;
@@ -546,8 +570,7 @@ ncclResult_t tryRegisterBuffer(struct ncclComm *comm, uintptr_t userBuff, size_t
ucprop.requestedHandleTypes = ncclCuMemHandleType;
CUCHECKGOTO(cuMemGetAllocationGranularity(&ucgran, &ucprop, CU_MEM_ALLOC_GRANULARITY_RECOMMENDED), ret, fail);
CUCHECKGOTO(cuMemGetAddressRange((CUdeviceptr*)&regRecord->baseAddr, &regRecord->baseSize, (CUdeviceptr)regRecord->addr), ret, fail);
if (regRecord->addr % ucgran == 0) {
if (regRecord->begAddr % ucgran == 0) {
if (regSize % ucgran != 0) {
regRecord->regUCSize = ALIGN_SIZE(regSize, ucgran);
} else {
@@ -555,7 +578,7 @@ ncclResult_t tryRegisterBuffer(struct ncclComm *comm, uintptr_t userBuff, size_t
}
regRecord->state |= NVLS_REG_POSSIBLE;
memcpy(&regData[comm->localRank].reg, regRecord, sizeof(struct ncclReg));
regData[comm->localRank].offset = userBuff - regRecord->addr;
regData[comm->localRank].offset = userBuff - regRecord->begAddr;
}
}
@@ -595,7 +618,7 @@ ncclResult_t tryRegisterBuffer(struct ncclComm *comm, uintptr_t userBuff, size_t
// Coverity complains that regRecord could be NULL. That won't in practice be the case because we've already checked
// (regData[i].reg.state & NVLS_REG_POSSIBLE) of all local ranks, which would catch it and bail out.
// coverity[var_deref_op]
CUCHECKGOTO(cuMulticastBindAddr(mcHandle, 0, (CUdeviceptr)regRecord->addr, ucsize, 0), ret, fail);
CUCHECKGOTO(cuMulticastBindAddr(mcHandle, 0, (CUdeviceptr)regRecord->begAddr, ucsize, 0), ret, fail);
// Create a VA for the NVLS
CUCHECKGOTO(cuMemAddressReserve(&regPtr, mcsize, mcgran, 0U, 0), ret, fail);
@@ -610,7 +633,7 @@ ncclResult_t tryRegisterBuffer(struct ncclComm *comm, uintptr_t userBuff, size_t
regRecord->mcHandle = mcHandle;
regRecord->state |= NVLS_REG_COMPLETE;
/* get all buffer addresses */
regRecord->caddrs[comm->localRank] = regRecord->addr;
regRecord->caddrs[comm->localRank] = regRecord->begAddr;
NCCLCHECKGOTO(ncclShmemAllgather(comm, &comm->nvlsResources->nvlsShmem, regRecord->caddrs + comm->localRank, regRecord->caddrs, sizeof(uintptr_t)), ret, fail);
/* Although registration is done, we still need to check whether the offsets are same among ranks. */
@@ -642,23 +665,23 @@ static ncclResult_t nvlsRegisterBuffer(struct ncclComm *comm, const void *sendbu
if (sendRegRecord) {
memcpy(&regData[comm->localRank * 2].reg, sendRegRecord, sizeof(struct ncclReg));
regData[comm->localRank * 2].offset = (uintptr_t)sendbuff - sendRegRecord->addr;
regData[comm->localRank * 2].offset = (uintptr_t)sendbuff - sendRegRecord->begAddr;
}
if (recvRegRecord) {
memcpy(&regData[comm->localRank * 2 + 1].reg, recvRegRecord, sizeof(struct ncclReg));
regData[comm->localRank * 2 + 1].offset = (uintptr_t)recvbuff - recvRegRecord->addr;
regData[comm->localRank * 2 + 1].offset = (uintptr_t)recvbuff - recvRegRecord->begAddr;
}
NCCLCHECKGOTO(ncclShmemAllgather(comm, &comm->nvlsResources->nvlsShmem, regData + comm->localRank * 2, regData, sizeof(struct localRegData) * 2), ret, fail);
/* first check whether all local ranks find their registered buffer */
for (int i = 0; i < comm->localRanks; ++i) {
if ((regData[i * 2].reg.state & NVLS_REG_COMPLETE) == 0 || regData[comm->localRank * 2].reg.caddrs[i] != regData[i * 2].reg.addr) {
if ((regData[i * 2].reg.state & NVLS_REG_COMPLETE) == 0 || regData[comm->localRank * 2].reg.caddrs[i] != regData[i * 2].reg.begAddr) {
sendNeedReg = true;
}
if ((regData[i * 2 + 1].reg.state & NVLS_REG_COMPLETE) == 0 || regData[comm->localRank * 2 + 1].reg.caddrs[i] != regData[i * 2 + 1].reg.addr) {
if ((regData[i * 2 + 1].reg.state & NVLS_REG_COMPLETE) == 0 || regData[comm->localRank * 2 + 1].reg.caddrs[i] != regData[i * 2 + 1].reg.begAddr) {
recvNeedReg = true;
}
@@ -787,7 +810,7 @@ ncclResult_t ncclNvlsGraphRegisterBuffer(
NCCLCHECK(ncclCommGraphRegister(comm, baseRecv, baseRecvSize, (void**)&recvRegRecord));
}
NCCLCHECK(nvlsRegisterBuffer(comm, baseSend, baseRecv, baseSendSize, baseRecvSize, sendRegRecord, recvRegRecord, outRegBufUsed, outRegBufSend, outRegBufRecv));
NCCLCHECK(nvlsRegisterBuffer(comm, sendbuff, recvbuff, sendbuffSize, recvbuffSize, sendRegRecord, recvRegRecord, outRegBufUsed, outRegBufSend, outRegBufRecv));
if (*outRegBufUsed) {
if (sendRegRecord) {
@@ -815,6 +838,124 @@ ncclResult_t ncclNvlsGraphRegisterBuffer(
return ncclSuccess;
}
ncclResult_t ncclNvlsSymmetricInit(struct ncclComm* comm) {
ncclResult_t ret = ncclSuccess;
if (comm && comm->nvlsSupport) {
CUmulticastObjectProp mcprop = {};
CUmemGenericAllocationHandle mcHandle;
char shareableHandle[NVLS_HANDLE_SIZE];
CUmemAccessDesc accessDesc = {};
mcprop.numDevices = comm->localRanks;
mcprop.handleTypes = ncclCuMemHandleType;
mcprop.flags = 0;
mcprop.size = comm->baseStride;
if (comm->localRank == 0) {
NCCLCHECKGOTO(nvlsGroupCreate(comm, &mcprop, comm->localRank, comm->localRanks, &mcHandle, shareableHandle), ret, fail);
NCCLCHECKGOTO(bootstrapIntraNodeBroadcast(comm->bootstrap, comm->localRankToRank, comm->localRank, comm->localRanks, 0, shareableHandle, NVLS_HANDLE_SIZE), ret, fail);
} else {
NCCLCHECKGOTO(bootstrapIntraNodeBroadcast(comm->bootstrap, comm->localRankToRank, comm->localRank, comm->localRanks, 0, shareableHandle, NVLS_HANDLE_SIZE), ret, fail);
NCCLCHECKGOTO(nvlsGroupConnect(comm, shareableHandle, comm->localRankToRank[0], &mcHandle), ret, fail);
}
CUCHECKGOTO(cuMulticastAddDevice(mcHandle, comm->cudaDev), ret, fail);
CUCHECKGOTO(cuMemAddressReserve((CUdeviceptr*)&comm->baseMCSymPtr, comm->baseStride, NCCL_MAX_PAGE_SIZE, 0, 0), ret, fail);
CUCHECKGOTO(cuMemMap((CUdeviceptr)comm->baseMCSymPtr, comm->baseStride, 0, mcHandle, 0), ret, fail);
accessDesc.location.type = CU_MEM_LOCATION_TYPE_DEVICE;
accessDesc.location.id = comm->cudaDev;
accessDesc.flags = CU_MEM_ACCESS_FLAGS_PROT_READWRITE;
CUCHECKGOTO(cuMemSetAccess((CUdeviceptr)comm->baseMCSymPtr, comm->baseStride, &accessDesc, 1), ret, fail);
comm->symMCHandle = mcHandle;
}
exit:
return ret;
fail:
goto exit;
}
ncclResult_t ncclNvlsSymmetricFinalize(struct ncclComm* comm) {
ncclResult_t ret = ncclSuccess;
if (comm && comm->nvlsSupport && comm->baseMCSymPtr) {
CUCHECKGOTO(cuMemUnmap((CUdeviceptr)comm->baseMCSymPtr, comm->baseStride), ret, fail);
CUCHECKGOTO(cuMemAddressFree((CUdeviceptr)comm->baseMCSymPtr, comm->baseStride), ret, fail);
CUCHECKGOTO(cuMemRelease(comm->symMCHandle), ret, fail);
}
exit:
return ret;
fail:
goto exit;
}
ncclResult_t ncclNvlsSymmetricMap(struct ncclComm* comm, size_t offset, size_t ucsize, void* ucaddr) {
ncclResult_t ret = ncclSuccess;
assert((uintptr_t)ucaddr % NCCL_REC_PAGE_SIZE == 0 && ucsize % NCCL_REC_PAGE_SIZE == 0);
if (comm && comm->nvlsSupport && ucaddr && ucsize > 0) {
CUCHECKGOTO(cuMulticastBindAddr(comm->symMCHandle, offset, (CUdeviceptr)ucaddr, ucsize, 0), ret, fail);
INFO(NCCL_ALLOC, "NVLS symmetric alloc mc buffer ptr %p offset %ld UC addr %p UC size %ld symAllocHead %ld", comm->baseMCSymPtr + offset, offset, ucaddr, ucsize, comm->symAllocHead);
}
exit:
return ret;
fail:
goto exit;
}
ncclResult_t ncclNvlsSymmetricFree(struct ncclComm* comm, size_t ucsize, void* ucaddr) {
ncclResult_t ret = ncclSuccess;
if (comm && comm->nvlsSupport && ucaddr && ucsize > 0) {
size_t offset = (size_t)ucaddr - ((size_t)comm->baseUCSymPtr + comm->localRank * comm->baseStride);
CUCHECKGOTO(cuMulticastUnbind(comm->symMCHandle, comm->cudaDev, offset, ucsize), ret, fail);
}
exit:
return ret;
fail:
goto exit;
}
ncclResult_t ncclNvlsRegResourcesQuery(struct ncclComm* comm, struct ncclTaskColl* info, int* recChannels) {
int factor;
ncclResult_t ret = ncclSuccess;
if (comm->nNodes == 1) {
if (info->func == ncclFuncReduceScatter) {
factor = (comm->compCap >= 100 ? 6 : 5) * 8;
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else if (info->func == ncclFuncAllGather) {
factor = 4 * 8;
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else if (info->func == ncclFuncAllReduce) {
if (comm->compCap >= 100) {
factor = 8 * 8;
} else {
factor = 4 * 8;
}
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else {
goto fail;
}
} else {
// Further tweaks for Blackwell with NVLS registered buffers
if (info->func == ncclFuncReduceScatter) {
factor = (comm->bandwidths[ncclFuncReduceScatter][NCCL_ALGO_NVLS][NCCL_PROTO_SIMPLE] > 400 ? 7 : 6) * 8;
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else if (info->func == ncclFuncAllGather) {
factor = 6 * 8;
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else if (info->func == ncclFuncAllReduce) {
factor = (comm->compCap >= 100 ? 7 : 6) * 8;
*recChannels = std::max(comm->config.minCTAs, std::min(comm->config.maxCTAs, DIVUP(factor, comm->nvlsResources->nHeads)));
} else {
goto fail;
}
}
exit:
return ret;
fail:
ret = ncclInvalidArgument;
goto exit;
}
#else
/*
@@ -860,4 +1001,25 @@ ncclResult_t ncclNvlsDeregBuffer(struct ncclComm* comm, CUmemGenericAllocationHa
return ncclSuccess;
}
ncclResult_t ncclNvlsSymmetricInit(struct ncclComm* comm) {
return ncclSuccess;
}
ncclResult_t ncclNvlsSymmetricMap(struct ncclComm* comm, size_t offset, size_t ucsize, void* ucaddr) {
return ncclSuccess;
}
ncclResult_t ncclNvlsSymmetricFree(struct ncclComm* comm, size_t ucsize, void* ucaddr) {
return ncclSuccess;
}
ncclResult_t ncclNvlsSymmetricFinalize(struct ncclComm* comm) {
return ncclSuccess;
}
ncclResult_t ncclNvlsRegResourcesQuery(struct ncclComm* comm, struct ncclTaskColl* info, int* recChannels) {
*recChannels = 0;
return ncclSuccess;
}
#endif /* CUDA_VERSION >= 12010 */
src/transport/p2p.cc
+88 -5
View File
@@ -12,6 +12,7 @@
#include "transport.h"
#include <assert.h>
#include "shm.h"
#include "register_inline.h"
enum p2pType { P2P_DIRECT, P2P_INTERMEDIATE, P2P_IPC, P2P_CUMEM };
@@ -826,7 +827,7 @@ ncclResult_t ret = ncclSuccess;
// We already have IPC info for peerLocalRank, no need to register it, we can reuse it
*regBufFlag = 1;
if (isLegacyIpc) *isLegacyIpc = regRecord->ipcInfos[peerLocalRank]->impInfo.legacyIpcCap;
INFO(NCCL_REG, "rank %d - IPC reuse buffer %p size %ld (baseAddr %p size %ld) to peer %d regAddr %p", comm->rank, userbuff, buffSize, (void*)regRecord->addr, regRecord->pages * comm->regCache.pageSize, peerRank, regRecord->ipcInfos[peerLocalRank]->impInfo.rmtRegAddr);
INFO(NCCL_REG, "rank %d - IPC reuse buffer %p size %ld (baseAddr %p size %ld) to peer %d regAddr %p", comm->rank, userbuff, buffSize, (void*)regRecord->begAddr, regRecord->endAddr - regRecord->begAddr, peerRank, regRecord->ipcInfos[peerLocalRank]->impInfo.rmtRegAddr);
} else {
// Register buffer with peerLocalRank
struct ncclProxyConnector* proxyConn = NULL;
@@ -885,11 +886,11 @@ ncclResult_t ret = ncclSuccess;
void* rmtRegAddr = NULL;
ipcInfo.size = baseSize;
ipcInfo.offset = regRecord->addr - (uintptr_t)baseAddr;
ipcInfo.offset = regRecord->begAddr - (uintptr_t)baseAddr;
// Now ipcInfo contains all necessary registration info. Start to register buffer on proxy side
// and get the remote register address back.
if (proxyConn) {
INFO(NCCL_REG, "rank %d - IPC registering buffer %p size %ld (baseAddr %p size %ld) to peer %d", comm->rank, userbuff, buffSize, (void*)regRecord->addr, ipcInfo.size, peerRank);
INFO(NCCL_REG, "rank %d - IPC registering buffer %p size %ld (baseAddr %p size %ld) to peer %d", comm->rank, userbuff, buffSize, (void*)regRecord->begAddr, ipcInfo.size, peerRank);
NCCLCHECKGOTO(ncclProxyCallBlocking(comm, proxyConn, ncclProxyMsgRegister, &ipcInfo, sizeof(p2pIpcExpInfo), &rmtRegAddr, sizeof(void*)), ret, fail);
}
if (rmtRegAddr) {
@@ -909,7 +910,7 @@ ncclResult_t ret = ncclSuccess;
regRecord->regIpcAddrs.hostPeerRmtAddrs[peerLocalRank] = (uintptr_t)rmtRegAddr;
needUpdate = true;
*regBufFlag = 1;
INFO(NCCL_REG, "rank %d - IPC registered buffer %p size %ld (baseAddr %p size %ld) to peer %d regAddr %p offsetOut %ld", comm->rank, userbuff, buffSize, (void*)regRecord->addr, ipcInfo.size, peerRank, rmtRegAddr, (uintptr_t)userbuff - regRecord->addr);
INFO(NCCL_REG, "rank %d - IPC register buffer %p size %ld (baseAddr %p size %ld) to peer %d regAddr %p offsetOut %ld", comm->rank, userbuff, buffSize, (void*)regRecord->begAddr, ipcInfo.size, peerRank, rmtRegAddr, (uintptr_t)userbuff - regRecord->begAddr);
}
}
}
@@ -935,7 +936,7 @@ ncclResult_t ret = ncclSuccess;
// p2p always returns remote addr here since remote buffer addr is passed in ncclDevWorkP2p struct
peerRmtAddrs = (uintptr_t*)regRecord->regIpcAddrs.hostPeerRmtAddrs[peerLocalRank];
}
*offsetOut = (uintptr_t)userbuff - regRecord->addr;
*offsetOut = (uintptr_t)userbuff - regRecord->begAddr;
*peerRmtAddrsOut = peerRmtAddrs;
}
}
@@ -1117,6 +1118,88 @@ fail:
goto exit;
}
ncclResult_t ncclIpcSymmetricInit(struct ncclComm* comm) {
CUCHECK(cuMemAddressReserve((CUdeviceptr*)&comm->baseUCSymPtr, comm->baseStride * comm->localRanks, NCCL_MAX_PAGE_SIZE, 0, 0));
return ncclSuccess;
}
ncclResult_t ncclIpcSymmetricFinalize(struct ncclComm* comm) {
if (comm->baseUCSymPtr) {
CUCHECK(cuMemAddressFree((CUdeviceptr)comm->baseUCSymPtr, comm->baseStride * comm->localRanks));
}
return ncclSuccess;
}
ncclResult_t ncclIpcSymmetricMap(struct ncclComm* comm, size_t offset, size_t size, CUmemGenericAllocationHandle memHandle, void** symPtr) {
ncclResult_t ret = ncclSuccess;
CUmemGenericAllocationHandle impHandle;
int impFd = -1;
ncclCuDesc* desc = NULL;
CUmemAccessDesc accessDesc = {};
assert(offset % NCCL_REC_PAGE_SIZE == 0 && size % NCCL_REC_PAGE_SIZE == 0);
NCCLCHECKGOTO(ncclCalloc(&desc, comm->localRanks), ret, fail);
if (ncclCuMemHandleType == CU_MEM_HANDLE_TYPE_POSIX_FILE_DESCRIPTOR) {
memcpy(&desc[comm->localRank].data, &memHandle, sizeof(CUmemGenericAllocationHandle));
} else {
CUCHECKGOTO(cuMemExportToShareableHandle(&desc[comm->localRank].handle, memHandle, ncclCuMemHandleType, 0), ret, fail);
}
NCCLCHECKGOTO(bootstrapIntraNodeAllGather(comm->bootstrap, comm->localRankToRank, comm->localRank, comm->localRanks, desc, sizeof(ncclCuDesc)), ret, fail);
// start mapping
accessDesc.location.type = CU_MEM_LOCATION_TYPE_DEVICE;
accessDesc.location.id = comm->cudaDev;
accessDesc.flags = CU_MEM_ACCESS_FLAGS_PROT_READWRITE;
for (int r = 0; r < comm->localRanks; ++r) {
CUdeviceptr maddr;
if (r == comm->localRank) {
impHandle = memHandle;
} else {
if (ncclCuMemHandleType == CU_MEM_HANDLE_TYPE_POSIX_FILE_DESCRIPTOR) {
impFd = -1;
NCCLCHECKGOTO(ncclProxyClientGetFdBlocking(comm, comm->localRankToRank[r], &desc[r].data, &impFd), ret, fail);
CUCHECKGOTO(cuMemImportFromShareableHandle(&impHandle, (void*)(uintptr_t)impFd, ncclCuMemHandleType), ret, fail);
SYSCHECKGOTO(close(impFd), "close", ret, fail);
} else {
CUCHECKGOTO(cuMemImportFromShareableHandle(&impHandle, (void*)&desc[r].handle, ncclCuMemHandleType), ret, fail);
}
}
maddr = (CUdeviceptr)(comm->baseUCSymPtr + (size_t)r * comm->baseStride + offset);
CUCHECKGOTO(cuMemMap(maddr, size, 0, impHandle, 0), ret, fail);
CUCHECKGOTO(cuMemSetAccess(maddr, size, &accessDesc, 1), ret, fail);
if (r == comm->localRank) {
*symPtr = (void*)maddr;
} else {
CUCHECKGOTO(cuMemRelease(impHandle), ret, fail);
}
}
INFO(NCCL_ALLOC, "IPC symmetric alloc buffer %p offset %ld size %ld symAllocHead %ld", *symPtr, offset, size, comm->symAllocHead);
exit:
free(desc);
return ret;
fail:
goto exit;
}
ncclResult_t ncclIpcSymmetricFree(struct ncclComm* comm, size_t size, void* symPtr) {
ncclResult_t ret = ncclSuccess;
if (comm && symPtr && size > 0) {
size_t offset = (size_t)symPtr - ((size_t)comm->baseUCSymPtr + comm->localRank * comm->baseStride);
for (int r = 0; r < comm->localRanks; ++r) {
CUdeviceptr peerAddr = (CUdeviceptr)(comm->baseUCSymPtr + r * comm->baseStride + offset);
CUCHECKGOTO(cuMemUnmap(peerAddr, size), ret, fail);
}
}
exit:
return ret;
fail:
goto exit;
}
struct ncclTransport p2pTransport = {
"P2P",
p2pCanConnect,
src/transport/profiler.cc
+7 -6
View File
@@ -6,6 +6,7 @@
#include "transport.h"
#include "proxy.h"
#include "profiler.h"
#include "device.h"
static ncclResult_t profilerProxyConnect(struct ncclProxyConnection* connection, struct ncclProxyState* proxyState, void* reqBuff, int reqSize, void* respBuff, int respSize, int* done) {
connection->proxyAppendPtr = &connection->proxyAppend;
@@ -29,15 +30,15 @@ static ncclResult_t profilerProxyProgress(struct ncclProxyState* proxyState, str
if (args->state == ncclProxyOpProgress) {
for (int s = 0; s < args->nsubs; s++) {
struct ncclProxySubArgs* sub = args->subs + s;
uint64_t* workStarted = (uint64_t *)sub->sendbuff;
uint64_t* workCompleted = (uint64_t *)sub->recvbuff;
if (sub->posted < sub->nsteps && sub->base <= workStarted[sub->channelId]) {
ncclProfilerStartKernelChEvent(args, s);
struct ncclDevProfiler* workStarted = (struct ncclDevProfiler *)sub->sendbuff;
struct ncclDevProfiler* workCompleted = (struct ncclDevProfiler *)sub->recvbuff;
if (sub->posted < sub->nsteps && sub->base <= workStarted[sub->channelId].data[sub->base%MAX_PROFILER_EVENTS_PER_CHANNEL].counter) {
ncclProfilerStartKernelChEvent(args, s, workStarted[sub->channelId].data[sub->base%MAX_PROFILER_EVENTS_PER_CHANNEL].timestamp);
sub->posted = sub->nsteps;
continue; // allow events on every channel to start
}
if (sub->transmitted < sub->nsteps && sub->base <= workCompleted[sub->channelId]) {
ncclProfilerStopKernelChEvent(args, s);
if (sub->transmitted < sub->nsteps && sub->base <= workCompleted[sub->channelId].data[sub->base%MAX_PROFILER_EVENTS_PER_CHANNEL].counter) {
ncclProfilerStopKernelChEvent(args, s, workCompleted[sub->channelId].data[sub->base%MAX_PROFILER_EVENTS_PER_CHANNEL].timestamp);
sub->transmitted = sub->nsteps;
args->done++;
}
src/transport/shm.cc
+1 -1
View File
@@ -10,7 +10,7 @@
#include "transport.h"
#define SHM_PATH_MAX 128
#define SHM_HANDLE_TYPE CU_MEM_HANDLE_TYPE_POSIX_FILE_DESCRIPTOR
#define SHM_HANDLE_TYPE ncclCuMemHandleType
struct shmBuffInfo {
void *hptr;