/************************************************************************* * Copyright (c) 2015-2023, NVIDIA CORPORATION. All rights reserved. * * See LICENSE.txt for license information ************************************************************************/ #include "argcheck.h" // Need some checks here since we access comm #include "collectives.h" #include "enqueue.h" #include "graph/topo.h" #include "nccl.h" #include "api_trace.h" #include "msccl/msccl_lifecycle.h" const char* ncclFuncToString(ncclFunc_t fn) { switch (fn) { case ncclFuncAllGather: return "AllGather"; case ncclFuncAllReduce: return "AllReduce"; case ncclFuncBroadcast: return "Broadcast"; case ncclFuncRecv: return "Recv"; case ncclFuncReduce: return "Reduce"; case ncclFuncReduceScatter: return "ReduceScatter"; case ncclFuncSendRecv: return "SendRecv"; case ncclFuncSend: return "Send"; default: return "Invalid"; } } const char* ncclDevRedOpToString(ncclDevRedOp_t op) { switch (op) { case ncclDevSum: return "Sum"; case ncclDevProd: return "Prod"; case ncclDevMinMax: return "MinMax"; case ncclDevPreMulSum: return "PreMulSum"; case ncclDevSumPostDiv: return "SumPostDiv"; default: return "Unknown"; } } const char* ncclDatatypeToString(ncclDataType_t type) { switch (type) { case ncclInt8: return "ncclInt8"; case ncclInt32: return "ncclInt32"; case ncclUint32: return "ncclUint32"; case ncclInt64: return "ncclInt64"; case ncclUint64: return "ncclUint64"; #if defined(RCCL_FLOAT8) case ncclFp8E4M3: return "ncclFp8E4M3"; case ncclFp8E5M2: return "ncclFp8E5M2"; #endif case ncclFloat16: return "ncclFloat16"; case ncclFloat32: return "ncclFloat32"; case ncclFloat64: return "ncclFloat64"; #if defined(RCCL_BFLOAT16) case ncclBfloat16: return "ncclBfloat16"; #endif default: return "Unknown"; } } const char* ncclAlgoToString(int algo) { switch (algo) { case NCCL_ALGO_TREE: return "TREE"; case NCCL_ALGO_RING: return "RING"; case NCCL_ALGO_COLLNET_DIRECT: return "COLLNET_DIRECT"; case NCCL_ALGO_COLLNET_CHAIN: return "COLLNET_CHAIN"; case NCCL_ALGO_NVLS: return "NVLS"; case NCCL_ALGO_NVLS_TREE: return "NVLS_TREE"; default: return "Unknown"; } } const char* ncclProtoToString(int proto) { switch (proto) { case NCCL_PROTO_LL: return "LL"; case NCCL_PROTO_LL128: return "LL128"; case NCCL_PROTO_SIMPLE: return "SIMPLE"; default: return "Unknown"; } } NCCL_API(ncclResult_t, ncclAllGather, const void* sendbuff, void* recvbuff, size_t sendcount, ncclDataType_t datatype, ncclComm_t comm, cudaStream_t stream); ncclResult_t ncclAllGather_impl(const void* sendbuff, void* recvbuff, size_t sendcount, ncclDataType_t datatype, ncclComm_t comm, cudaStream_t stream) { struct NvtxParamsAllGather { size_t bytes; ncclDataType_t datatype; }; // Just pass the size of one message and not the total bytes sent/received. constexpr nvtxPayloadSchemaEntry_t AllGatherSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Message size [bytes]"}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsAllGather, datatype)} }; NvtxParamsAllGather payload{sendcount * ncclTypeSize(datatype), datatype}; NVTX3_FUNC_WITH_PARAMS(AllGather, AllGatherSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, sendcount, datatype, 0, 0, ncclSum, mscclFuncAllGather, comm, stream); } struct ncclInfo info = { ncclFuncAllGather, "AllGather", sendbuff, recvbuff, sendcount, datatype, ncclSum, 0, comm, stream, /* Args */ ALLGATHER_CHUNKSTEPS, ALLGATHER_SLICESTEPS }; NCCLCHECK(ncclEnqueueCheck(&info)); return ncclSuccess; } NCCL_API(ncclResult_t, ncclAllReduce, const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclRedOp_t op, ncclComm* comm, cudaStream_t stream); ncclResult_t ncclAllReduce_impl(const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclRedOp_t op, ncclComm* comm, cudaStream_t stream) { struct NvtxParamsAllReduce { size_t bytes; ncclRedOp_t op; ncclDataType_t datatype; }; // Just pass the size of one message and not the total bytes sent/received. static constexpr nvtxPayloadSchemaEntry_t AllReduceSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Message size [bytes]"}, {0, NVTX_PAYLOAD_ENTRY_NCCL_REDOP, "Reduction operation", nullptr, 0, offsetof(NvtxParamsAllReduce, op)}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsAllReduce, datatype)} }; NvtxParamsAllReduce payload{count * ncclTypeSize(datatype), op, datatype}; NVTX3_FUNC_WITH_PARAMS(AllReduce, AllReduceSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, count, datatype, 0, 0, op, mscclFuncAllReduce, comm, stream); } struct ncclInfo info = { ncclFuncAllReduce, "AllReduce", sendbuff, recvbuff, count, datatype, op, 0, comm, stream, /* Args */ ALLREDUCE_CHUNKSTEPS, ALLREDUCE_SLICESTEPS }; NCCLCHECK(ncclEnqueueCheck(&info)); return ncclSuccess; } RCCL_PARAM(AllToAllPivotEnable, "ALL_TO_ALL_PIVOT_ENABLE", 0); NCCL_API(ncclResult_t, ncclAllToAll, const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream); ncclResult_t ncclAllToAll_impl(const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream) { struct NvtxParamsAllToAll { size_t bytes; ncclDataType_t datatype; }; // Just pass the size of one message and not the total bytes sent/received. constexpr nvtxPayloadSchemaEntry_t AllToAllSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Message size [bytes]"}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsAllToAll, datatype)} }; NvtxParamsAllToAll payload{count * ncclTypeSize(datatype), datatype}; NVTX3_FUNC_WITH_PARAMS(AllToAll, AllToAllSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, count, datatype, 0, 0, ncclSum, mscclFuncAllToAll, comm, stream); } size_t rankOffset = count * ncclTypeSize(datatype); size_t rankAlign = rankOffset & ((~rankOffset) + 1); // Determine Pivot A2A support now that we know number of channels if (comm->topo->pivotA2AEnabled && comm->nChannels >= comm->topo->pivotA2ANumBiRings * 2 && rankOffset >= 744 * 1024 && rankAlign != 4 && rcclParamAllToAllPivotEnable()) { struct ncclInfo info = { ncclFuncAllToAllPivot, "AllToAllPivot", sendbuff, recvbuff, count, datatype, ncclSum, 0, comm, stream, /* Args */ ALLTOALL_PIVOT_CHUNKSTEPS, ALLTOALL_PIVOT_SLICESTEPS }; return ncclEnqueueCheck(&info); } else { int nRanks; NCCLCHECK(ncclCommCount(comm, &nRanks)); if (count == 0) return ncclSuccess; NCCLCHECK(ncclGroupStart()); for (int r=0; rrank] * ncclTypeSize(datatype), recvcounts[comm->rank] * ncclTypeSize(datatype), datatype}; NVTX3_FUNC_WITH_PARAMS(AllToAllv, AllToAllvSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, sendcounts, sdispls, recvbuff, recvcounts, rdispls, 0, datatype, 0, 0, ncclSum, mscclFuncAllToAllv, comm, stream); } int nRanks; NCCLCHECK(ncclCommCount(comm, &nRanks)); NCCLCHECK(ncclGroupStart()); for (int r=0; rrank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, count, datatype, root, 0, ncclSum, mscclFuncBroadcast, comm, stream); } struct ncclInfo info = { ncclFuncBroadcast, "Broadcast", sendbuff, recvbuff, count, datatype, ncclSum, root, comm, stream, /* Args */ BROADCAST_CHUNKSTEPS, BROADCAST_SLICESTEPS }; NCCLCHECK(ncclEnqueueCheck(&info)); return ncclSuccess; } /* Deprecated original "in place" function, similar to MPI */ NCCL_API(ncclResult_t, ncclBcast, void* buff, size_t count, ncclDataType_t datatype, int root, ncclComm_t comm, cudaStream_t stream); ncclResult_t ncclBcast(void* buff, size_t count, ncclDataType_t datatype, int root, ncclComm_t comm, cudaStream_t stream) { NCCLCHECK(ncclBroadcast(buff, buff, count, datatype, root, comm, stream)); return ncclSuccess; } NCCL_API(ncclResult_t, ncclGather, const void* sendbuff, void* recvbuff, size_t sendcount, ncclDataType_t datatype, int root, ncclComm_t comm, hipStream_t stream); ncclResult_t ncclGather_impl(const void* sendbuff, void* recvbuff, size_t sendcount, ncclDataType_t datatype, int root, ncclComm_t comm, hipStream_t stream) { struct NvtxParamsGather { size_t bytes; int root; ncclDataType_t datatype; }; constexpr nvtxPayloadSchemaEntry_t GatherSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Bytes"}, {0, NVTX_PAYLOAD_ENTRY_TYPE_INT, "Root", nullptr, 0, offsetof(NvtxParamsGather, root)}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsGather, datatype)} }; NvtxParamsGather payload{sendcount * ncclTypeSize(datatype), root, datatype}; NVTX3_FUNC_WITH_PARAMS(Gather, GatherSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, sendcount, datatype, root, 0, ncclSum, mscclFuncGather, comm, stream); } int nRanks; NCCLCHECK(ncclCommCount(comm, &nRanks)); size_t rankOffset = sendcount * ncclTypeSize(datatype); if (sendcount == 0) return ncclSuccess; int rank; NCCLCHECK(ncclCommUserRank(comm, &rank)); NCCLCHECK(ncclGroupStart()); if (rank == root) { for (int r=0; rrank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, count, datatype, root, 0, op, mscclFuncReduce, comm, stream); } struct ncclInfo info = { ncclFuncReduce, "Reduce", sendbuff, recvbuff, count, datatype, op, root, comm, stream, /* Args */ REDUCE_CHUNKSTEPS, REDUCE_SLICESTEPS }; NCCLCHECK(ncclEnqueueCheck(&info)); return ncclSuccess; } NCCL_API(ncclResult_t, ncclReduceScatter, const void* sendbuff, void* recvbuff, size_t recvcount, ncclDataType_t datatype, ncclRedOp_t op, ncclComm* comm, cudaStream_t stream); ncclResult_t ncclReduceScatter_impl(const void* sendbuff, void* recvbuff, size_t recvcount, ncclDataType_t datatype, ncclRedOp_t op, ncclComm* comm, cudaStream_t stream) { struct NvtxParamsReduceScatter { size_t bytes; ncclRedOp_t op; ncclDataType_t datatype; }; constexpr nvtxPayloadSchemaEntry_t ReduceScatterSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Message size [bytes]"}, {0, NVTX_PAYLOAD_ENTRY_NCCL_REDOP, "Reduction operation", nullptr, 0, offsetof(NvtxParamsReduceScatter, op)}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsReduceScatter, datatype)} }; NvtxParamsReduceScatter payload{recvcount * ncclTypeSize(datatype), op, datatype}; NVTX3_FUNC_WITH_PARAMS(ReduceScatter, ReduceScatterSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, recvcount, datatype, 0, 0, op, mscclFuncReduceScatter, comm, stream); } struct ncclInfo info = { ncclFuncReduceScatter, "ReduceScatter", sendbuff, recvbuff, recvcount, datatype, op, 0, comm, stream, /* Args */ REDUCESCATTER_CHUNKSTEPS, REDUCESCATTER_SLICESTEPS }; NCCLCHECK(ncclEnqueueCheck(&info)); return ncclSuccess; } NCCL_API(ncclResult_t, ncclScatter, const void* sendbuff, void* recvbuff, size_t recvcount, ncclDataType_t datatype, int root, ncclComm_t comm, hipStream_t stream); ncclResult_t ncclScatter_impl(const void* sendbuff, void* recvbuff, size_t recvcount, ncclDataType_t datatype, int root, ncclComm_t comm, hipStream_t stream) { struct NvtxParamsScatter { size_t bytes; int root; ncclDataType_t datatype; }; constexpr nvtxPayloadSchemaEntry_t ScatterSchema[] = { {0, NVTX_PAYLOAD_ENTRY_TYPE_SIZE, "Bytes"}, {0, NVTX_PAYLOAD_ENTRY_TYPE_INT, "Root", nullptr, 0, offsetof(NvtxParamsScatter, root)}, {0, NVTX_PAYLOAD_ENTRY_TYPE_DATATYPE, "Data type", nullptr, 0, offsetof(NvtxParamsScatter, datatype)} }; NvtxParamsScatter payload{recvcount * ncclTypeSize(datatype), root, datatype}; NVTX3_FUNC_WITH_PARAMS(Scatter, ScatterSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, recvbuff, nullptr, nullptr, recvcount, datatype, root, 0, ncclSum, mscclFuncScatter, comm, stream); } int nRanks; NCCLCHECK(ncclCommCount(comm, &nRanks)); size_t rankOffset = recvcount * ncclTypeSize(datatype); if (recvcount == 0) return ncclSuccess; int rank; NCCLCHECK(ncclCommUserRank(comm, &rank)); NCCLCHECK(ncclGroupStart()); if (rank == root) { for (int r=0; rrank) && !mscclIsCaller()) { return mscclEnqueueCheck( sendbuff, nullptr, nullptr, nullptr, nullptr, nullptr, count, datatype, 0, peer, ncclSum, mscclFuncSend, comm, stream); } struct ncclInfo info = { ncclFuncSend, "Send", NULL, (void*)sendbuff, count, datatype, ncclSum, peer, comm, stream, /* Args */ 1, 1 }; ncclResult_t ret; NCCLCHECK(ncclGroupStart()); NCCLCHECKGOTO(ncclEnqueueCheck(&info), ret, exit); exit: NCCLCHECK(ncclGroupEnd()); return ret; } NCCL_API(ncclResult_t, ncclRecv, void* recvbuff, size_t count, ncclDataType_t datatype, int peer, ncclComm_t comm, cudaStream_t stream); ncclResult_t ncclRecv_impl(void* recvbuff, size_t count, ncclDataType_t datatype, int peer, ncclComm_t comm, cudaStream_t stream) { NvtxParamsSendRecv payload{count * ncclTypeSize(datatype), peer, datatype}; NVTX3_FUNC_WITH_PARAMS(Recv, SendRecvSchema, payload) if (mscclAvailable(comm->rank) && !mscclIsCaller()) { return mscclEnqueueCheck( nullptr, nullptr, nullptr, recvbuff, nullptr, nullptr, count, datatype, 0, peer, ncclSum, mscclFuncRecv, comm, stream); } struct ncclInfo info = { ncclFuncRecv, "Recv", NULL, recvbuff, count, datatype, ncclSum, peer, comm, stream, /* Args */ 1, 1 }; ncclResult_t ret; NCCLCHECK(ncclGroupStart()); NCCLCHECKGOTO(ncclEnqueueCheck(&info), ret, exit); exit: NCCLCHECK(ncclGroupEnd()); return ret; }