Add MSCCL Support (#658)
* Add MSCCL support
* Add alignment and message size checking
* Fix nRanks checking, in-place and out-of-place tests and group call handling
* Fix hipGraph unit test
* Change MSCCL init warning to INFO
* Revise license info
[ROCm/rccl commit: adafc0f759]
이 커밋은 다음에 포함됨:
Ziyue Yang
GitHub
@@ -0,0 +1,333 @@
|
||||
/*************************************************************************
|
||||
* Copyright (c) Microsoft Corporation.
|
||||
* Licensed under the MIT License.
|
||||
************************************************************************/
|
||||
|
||||
#include <atomic>
|
||||
|
||||
#include "alloc.h"
|
||||
#include "checks.h"
|
||||
|
||||
#include "msccl/msccl_lifecycle.h"
|
||||
#include "msccl/msccl_parser.h"
|
||||
#include "msccl/msccl_setup.h"
|
||||
#include "msccl/msccl_status.h"
|
||||
|
||||
RCCL_PARAM(MscclEnabled, "MSCCL_ENABLE", 0);
|
||||
static const char* mscclAlgoFilePathEnv = "MSCCL_ALGO_FILE_PATH";
|
||||
static std::atomic<bool> mscclInitialized;
|
||||
static bool mscclSchedulerTriedLoadAlgo = false;
|
||||
|
||||
bool mscclEnabled() {
|
||||
return rcclParamMscclEnabled();
|
||||
}
|
||||
|
||||
static bool mscclIsCallerFlag = false;
|
||||
|
||||
void mscclSetIsCallerFlag() {
|
||||
mscclIsCallerFlag = true;
|
||||
}
|
||||
|
||||
void mscclClearIsCallerFlag() {
|
||||
mscclIsCallerFlag = false;
|
||||
}
|
||||
|
||||
bool mscclIsCaller() {
|
||||
return mscclIsCallerFlag;
|
||||
}
|
||||
|
||||
bool mscclAvailable() {
|
||||
return mscclEnabled() && mscclInitialized.load(std::memory_order_acquire);
|
||||
}
|
||||
|
||||
ncclResult_t mscclInit(ncclComm_t comm) {
|
||||
if (comm->intraRanks > 1) {
|
||||
mscclInitialized.store(false, std::memory_order_release);
|
||||
INFO(NCCL_INIT, "MSCCL doesn't support multiple GPUs in one process and is not available");
|
||||
return ncclSuccess;
|
||||
} else {
|
||||
mscclInitialized.store(true, std::memory_order_release);
|
||||
}
|
||||
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
status.scratchBuffer = nullptr;
|
||||
status.scratchBufferSize = 0;
|
||||
status.rank = comm->rank;
|
||||
status.workIndex = 1;
|
||||
status.freeAlgoHandles.resize(MSCCL_MAX_NUM_ALGOS);
|
||||
for (int i = 0; i < MSCCL_MAX_NUM_ALGOS; i++) {
|
||||
status.freeAlgoHandles[i] = MSCCL_MAX_NUM_ALGOS - i - 1;
|
||||
}
|
||||
NCCLCHECK(ncclCudaCalloc(&status.syncFlags, MSCCL_MAX_NUM_THREAD_BLOCKS));
|
||||
status.groupStatus = mscclNoGroup;
|
||||
status.groupDepth = 0;
|
||||
mscclSchedulerTriedLoadAlgo = false;
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclGroupStart() {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
status.groupDepth++;
|
||||
if (status.groupStatus == mscclNoGroup) {
|
||||
status.groupStatus = mscclGroupSupportedOp;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t mscclScheduler(struct mscclSchedulerParam* param) {
|
||||
static bool algoAvailable = false;
|
||||
static mscclAlgoHandle_t loadedAlgoHandle;
|
||||
static mscclAlgo* loadedHostAlgo = nullptr;
|
||||
|
||||
param->scheduled = false;
|
||||
|
||||
if (!mscclSchedulerTriedLoadAlgo) {
|
||||
mscclSchedulerTriedLoadAlgo = true;
|
||||
const char* mscclAlgoFilePath = getenv(mscclAlgoFilePathEnv);
|
||||
if (mscclAlgoFilePath != nullptr) {
|
||||
NCCLCHECK(mscclLoadAlgo(mscclAlgoFilePath, &loadedAlgoHandle));
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
loadedHostAlgo = status.hostAlgos[loadedAlgoHandle];
|
||||
algoAvailable = true;
|
||||
}
|
||||
}
|
||||
if (!algoAvailable) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
bool mscclAlgoFuncIsValid = loadedHostAlgo->func == param->func;
|
||||
if (!mscclAlgoFuncIsValid) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
bool numGpusIsValid = loadedHostAlgo->nRanks == param->comm->nRanks;
|
||||
if (!numGpusIsValid) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
size_t nBytes = param->count * ncclTypeSize(param->dataType) * loadedHostAlgo->sizeMultiplier;
|
||||
bool msgSizeIsValid =
|
||||
param->count > 0 && (param->count % loadedHostAlgo->nChunksPerLoop) == 0 &&
|
||||
nBytes >= loadedHostAlgo->minBytes &&
|
||||
(loadedHostAlgo->maxBytes == 0 || nBytes <= loadedHostAlgo->maxBytes);
|
||||
if (!msgSizeIsValid) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
bool isInPlace = false;
|
||||
if (param->func == mscclFuncReduce ||
|
||||
param->func == mscclFuncBroadcast ||
|
||||
param->func == mscclFuncAllReduce ||
|
||||
param->func == mscclFuncAllToAll ||
|
||||
param->func == mscclFuncAllToAllv) {
|
||||
isInPlace = param->sendBuff == param->recvBuff;
|
||||
} else if (param->func == mscclFuncAllGather ||
|
||||
param->func == mscclFuncGather) {
|
||||
isInPlace = (char*)param->sendBuff == (char*)param->recvBuff + param->comm->rank * param->count * ncclTypeSize(param->dataType);
|
||||
} else if (param->func == mscclFuncReduceScatter ||
|
||||
param->func == mscclFuncScatter) {
|
||||
isInPlace = (char*)param->recvBuff == (char*)param->sendBuff + param->comm->rank * param->count * ncclTypeSize(param->dataType);
|
||||
}
|
||||
bool inPlaceOutOfPlaceIsValid = isInPlace ? loadedHostAlgo->inPlace : loadedHostAlgo->outOfPlace;
|
||||
if (!inPlaceOutOfPlaceIsValid) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
param->handle = loadedAlgoHandle;
|
||||
param->scheduled = true;
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t mscclSetSchedulerParam(
|
||||
const void* sendBuff, const size_t sendCounts[], const size_t sDisPls[],
|
||||
void* recvBuff, const size_t recvCounts[], const size_t rDisPls[],
|
||||
size_t count, ncclDataType_t dataType, int root, int peer, ncclRedOp_t op,
|
||||
mscclFunc_t func, ncclComm_t comm, hipStream_t stream,
|
||||
struct mscclSchedulerParam* param) {
|
||||
param->sendBuff = sendBuff;
|
||||
param->sendCounts = sendCounts;
|
||||
param->sDisPls = sDisPls;
|
||||
param->recvBuff = recvBuff;
|
||||
param->recvCounts = recvCounts;
|
||||
param->rDisPls = rDisPls;
|
||||
param->count = count;
|
||||
param->dataType = dataType;
|
||||
param->root = root;
|
||||
param->peer = peer;
|
||||
param->op = op;
|
||||
param->func = func;
|
||||
param->comm = comm;
|
||||
param->stream = stream;
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t mscclSaveCountsAndDispls(struct mscclSchedulerParam* param) {
|
||||
if (param->sendCounts) {
|
||||
param->savedSendCounts.assign(param->sendCounts, param->sendCounts + param->comm->nRanks);
|
||||
param->sendCounts = param->savedSendCounts.data();
|
||||
param->savedSDisPls.assign(param->sDisPls, param->sDisPls + param->comm->nRanks);
|
||||
param->sDisPls = param->savedSDisPls.data();
|
||||
param->savedRecvCounts.assign(param->recvCounts, param->recvCounts + param->comm->nRanks);
|
||||
param->recvCounts = param->savedRecvCounts.data();
|
||||
param->savedRDisPls.assign(param->rDisPls, param->rDisPls + param->comm->nRanks);
|
||||
param->rDisPls = param->savedRDisPls.data();
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t mscclRunSavedParams() {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
for (auto& param : status.savedSchedulerParams) {
|
||||
NCCLCHECK(mscclRunAlgo(
|
||||
param.sendBuff, param.sendCounts, param.sDisPls,
|
||||
param.recvBuff, param.recvCounts, param.rDisPls,
|
||||
param.count, param.dataType, param.root, param.peer, param.op, param.handle, param.comm, param.stream));
|
||||
}
|
||||
status.savedSchedulerParams.clear();
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t mscclFallBackSavedParams() {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
mscclSetIsCallerFlag();
|
||||
for (auto& param : status.savedSchedulerParams) {
|
||||
switch (param.func) {
|
||||
case mscclFuncReduce:
|
||||
NCCLCHECK(ncclReduce(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.op, param.root, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncBroadcast:
|
||||
NCCLCHECK(ncclBroadcast(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.root, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncAllReduce:
|
||||
NCCLCHECK(ncclAllReduce(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.op, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncReduceScatter:
|
||||
NCCLCHECK(ncclReduceScatter(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.op, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncAllGather:
|
||||
NCCLCHECK(ncclAllGather(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncSend:
|
||||
NCCLCHECK(ncclSend(param.sendBuff, param.count, param.dataType,
|
||||
param.peer, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncRecv:
|
||||
NCCLCHECK(ncclRecv(param.recvBuff, param.count, param.dataType,
|
||||
param.peer, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncGather:
|
||||
NCCLCHECK(ncclGather(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.root, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncScatter:
|
||||
NCCLCHECK(ncclScatter(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.root, param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncAllToAll:
|
||||
NCCLCHECK(ncclAllToAll(param.sendBuff, param.recvBuff, param.count, param.dataType,
|
||||
param.comm, param.stream));
|
||||
break;
|
||||
case mscclFuncAllToAllv:
|
||||
NCCLCHECK(ncclAllToAllv(
|
||||
param.sendBuff, param.sendCounts, param.sDisPls,
|
||||
param.recvBuff, param.recvCounts, param.rDisPls,
|
||||
param.dataType, param.comm, param.stream));
|
||||
break;
|
||||
default:
|
||||
WARN("Invalid MSCCL function type in saved parameter");
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
}
|
||||
mscclClearIsCallerFlag();
|
||||
status.savedSchedulerParams.clear();
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclEnqueueCheck(
|
||||
const void* sendBuff, const size_t sendCounts[], const size_t sDisPls[],
|
||||
void* recvBuff, const size_t recvCounts[], const size_t rDisPls[],
|
||||
size_t count, ncclDataType_t dataType, int root, int peer, ncclRedOp_t op,
|
||||
mscclFunc_t func, ncclComm_t comm, hipStream_t stream) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
hipStreamCaptureStatus captureStatus;
|
||||
unsigned long long pid;
|
||||
|
||||
status.savedSchedulerParams.push_back({});
|
||||
NCCLCHECK(mscclSetSchedulerParam(
|
||||
sendBuff, sendCounts, sDisPls, recvBuff, recvCounts, rDisPls,
|
||||
count, dataType, root, peer, op, func, comm, stream,
|
||||
&status.savedSchedulerParams.back()));
|
||||
|
||||
switch (status.groupStatus) {
|
||||
case mscclNoGroup:
|
||||
CUDACHECK(hipStreamGetCaptureInfo(stream, &captureStatus, &pid));
|
||||
if (captureStatus == hipStreamCaptureStatusNone) {
|
||||
NCCLCHECK(mscclScheduler(&status.savedSchedulerParams.back()));
|
||||
if (status.savedSchedulerParams.back().scheduled) {
|
||||
NCCLCHECK(mscclRunSavedParams());
|
||||
break;
|
||||
}
|
||||
}
|
||||
NCCLCHECK(mscclFallBackSavedParams());
|
||||
break;
|
||||
case mscclGroupSupportedOp:
|
||||
CUDACHECK(hipStreamGetCaptureInfo(stream, &captureStatus, &pid));
|
||||
if (captureStatus == hipStreamCaptureStatusNone) {
|
||||
NCCLCHECK(mscclScheduler(&status.savedSchedulerParams.back()));
|
||||
if (status.savedSchedulerParams.back().scheduled) {
|
||||
// Only save counts and displs when there is suitable MSCCL algorithm for this
|
||||
NCCLCHECK(mscclSaveCountsAndDispls(&status.savedSchedulerParams.back()));
|
||||
break;
|
||||
}
|
||||
}
|
||||
NCCLCHECK(mscclFallBackSavedParams());
|
||||
break;
|
||||
case mscclGroupUnsupportedOp:
|
||||
NCCLCHECK(mscclFallBackSavedParams());
|
||||
break;
|
||||
default:
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclGroupEnd() {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
status.groupDepth--;
|
||||
if (status.groupDepth == 0) {
|
||||
if (status.groupStatus == mscclGroupSupportedOp) {
|
||||
NCCLCHECK(mscclRunSavedParams());
|
||||
}
|
||||
status.groupStatus = mscclNoGroup;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclTeardown() {
|
||||
if (!mscclInitialized.load(std::memory_order_acquire)) {
|
||||
return ncclSuccess;
|
||||
}
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
for (auto &p : status.hostAlgos) {
|
||||
free(p.second);
|
||||
status.freeAlgoHandles.push_back(p.first);
|
||||
}
|
||||
for (auto &p : status.devAlgos) {
|
||||
CUDACHECK(hipFree(p.second));
|
||||
}
|
||||
CUDACHECK(hipFree(status.scratchBuffer));
|
||||
CUDACHECK(hipFree(status.syncFlags));
|
||||
status.hostAlgos.clear();
|
||||
status.devAlgos.clear();
|
||||
status.freeAlgoHandles.clear();
|
||||
status.scratchBuffer = nullptr;
|
||||
status.scratchBufferSize = 0;
|
||||
status.workIndex = 1;
|
||||
mscclInitialized.store(false, std::memory_order_release);
|
||||
return ncclSuccess;
|
||||
}
|
||||
@@ -0,0 +1,703 @@
|
||||
/*************************************************************************
|
||||
* Copyright (c) 2019-2022, NVIDIA CORPORATION. All rights reserved.
|
||||
* Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved.
|
||||
* Modifications Copyright (c) Microsoft Corporation. Licensed under the MIT License.
|
||||
*
|
||||
* See LICENSE.txt for license information
|
||||
************************************************************************/
|
||||
|
||||
#include <sys/types.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h>
|
||||
#include <ctype.h>
|
||||
#include "core.h"
|
||||
#include "collectives.h"
|
||||
#include "msccl/msccl_parser.h"
|
||||
|
||||
ncclResult_t mscclXmlGetChar(FILE* file, char* c) {
|
||||
if (fread(c, 1, 1, file) == 0) {
|
||||
WARN("XML Parse : Unexpected EOF");
|
||||
return ncclInternalError;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclXmlGetValue(FILE* file, char* value, char* last) {
|
||||
char c;
|
||||
NCCLCHECK(mscclXmlGetChar(file, &c));
|
||||
if (c != '"' && c != '\'') {
|
||||
#if INT_OK
|
||||
int o = 0;
|
||||
do {
|
||||
value[o++] = c;
|
||||
NCCLCHECK(mscclXmlGetChar(file, &c));
|
||||
} while (c >= '0' && c <= '9');
|
||||
value[o] = '\0';
|
||||
*last = c;
|
||||
return ncclSuccess;
|
||||
#else
|
||||
WARN("XML Parse : Expected (double) quote.");
|
||||
return ncclInternalError;
|
||||
#endif
|
||||
}
|
||||
int o = 0;
|
||||
do {
|
||||
NCCLCHECK(mscclXmlGetChar(file, &c));
|
||||
value[o++] = c;
|
||||
} while (c != '"');
|
||||
value[o-1] = '\0';
|
||||
NCCLCHECK(mscclXmlGetChar(file, last));
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclXmlGetToken(FILE* file, char* name, char* value, char* last) {
|
||||
char c;
|
||||
char* ptr = name;
|
||||
int o = 0;
|
||||
do {
|
||||
NCCLCHECK(mscclXmlGetChar(file, &c));
|
||||
if (c == '=') {
|
||||
ptr[o] = '\0';
|
||||
if (value == NULL) {
|
||||
WARN("XML Parse : Unexpected value with name %s", ptr);
|
||||
return ncclInternalError;
|
||||
}
|
||||
return mscclXmlGetValue(file, value, last);
|
||||
}
|
||||
ptr[o] = c;
|
||||
if (o == MAX_STR_LEN-1) {
|
||||
ptr[o] = '\0';
|
||||
WARN("Error : name %s too long (max %d)", ptr, MAX_STR_LEN);
|
||||
return ncclInternalError;
|
||||
}
|
||||
o++;
|
||||
} while (c != ' ' && c != '>' && c != '/' && c != '\n' && c != '\r');
|
||||
ptr[o-1] = '\0';
|
||||
*last = c;
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
// Shift the 3-chars string by one char and append c at the end
|
||||
#define SHIFT_APPEND(s, c) do { s[0]=s[1]; s[1]=s[2]; s[2]=c; } while(0)
|
||||
ncclResult_t mscclXmlSkipComment(FILE* file, char* start, char next) {
|
||||
// Start from something neutral with \0 at the end.
|
||||
char end[4] = "...";
|
||||
|
||||
// Inject all trailing chars from previous reads. We don't need
|
||||
// to check for --> here because there cannot be a > in the name.
|
||||
for (int i=0; i<strlen(start); i++) SHIFT_APPEND(end, start[i]);
|
||||
SHIFT_APPEND(end, next);
|
||||
|
||||
// Stop when we find "-->"
|
||||
while (strcmp(end, "-->") != 0) {
|
||||
int c;
|
||||
if (fread(&c, 1, 1, file) != 1) {
|
||||
WARN("XML Parse error : unterminated comment");
|
||||
return ncclInternalError;
|
||||
}
|
||||
SHIFT_APPEND(end, c);
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclXmlGetNode(FILE* file, struct mscclXmlNode* node) {
|
||||
node->type = NODE_TYPE_NONE;
|
||||
char c = ' ';
|
||||
while (c == ' ' || c == '\n' || c == '\r') {
|
||||
if (fread(&c, 1, 1, file) == 0) return ncclSuccess;
|
||||
}
|
||||
if (c != '<') {
|
||||
WARN("XML Parse error : expecting '<', got '%c'", c);
|
||||
return ncclInternalError;
|
||||
}
|
||||
// Read XML element name
|
||||
NCCLCHECK(mscclXmlGetToken(file, node->name, NULL, &c));
|
||||
|
||||
// Check for comments
|
||||
if (strncmp(node->name, "!--", 3) == 0) {
|
||||
NCCLCHECK(mscclXmlSkipComment(file, node->name+3, c));
|
||||
return mscclXmlGetNode(file, node);
|
||||
}
|
||||
|
||||
// Check for closing tag
|
||||
if (node->name[0] == '\0' && c == '/') {
|
||||
node->type = NODE_TYPE_CLOSE;
|
||||
// Re-read the name, we got '/' in the first call
|
||||
NCCLCHECK(mscclXmlGetToken(file, node->name, NULL, &c));
|
||||
if (c != '>') {
|
||||
WARN("XML Parse error : unexpected trailing %c in closing tag %s", c, node->name);
|
||||
return ncclInternalError;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
node->type = NODE_TYPE_OPEN;
|
||||
|
||||
// Get Attributes
|
||||
int a = 0;
|
||||
while (c == ' ') {
|
||||
NCCLCHECK(mscclXmlGetToken(file, node->attrs[a].key, node->attrs[a].value, &c));
|
||||
if (a == MAX_ATTR_COUNT) {
|
||||
INFO(NCCL_GRAPH, "XML Parse : Ignoring extra attributes (max %d)", MAX_ATTR_COUNT);
|
||||
// Actually we need to still consume the extra attributes so we have an extra one.
|
||||
} else a++;
|
||||
}
|
||||
node->nAttrs = a;
|
||||
if (c == '/') {
|
||||
node->type = NODE_TYPE_SINGLE;
|
||||
char str[MAX_STR_LEN];
|
||||
NCCLCHECK(mscclXmlGetToken(file, str, NULL, &c));
|
||||
}
|
||||
if (c != '>') {
|
||||
WARN("XML Parse : expected >, got '%c'", c);
|
||||
return ncclInternalError;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
typedef ncclResult_t (*mscclXmlHandlerFunc_t)(FILE*, struct mscclXml*, struct mscclXmlNode*);
|
||||
|
||||
struct mscclXmlHandler {
|
||||
const char * name;
|
||||
mscclXmlHandlerFunc_t func;
|
||||
};
|
||||
|
||||
ncclResult_t mscclXmlLoadSub(FILE* file, struct mscclXml* xml, struct mscclXmlNode* head, struct mscclXmlHandler handlers[], int nHandlers) {
|
||||
if (head && head->type == NODE_TYPE_SINGLE) return ncclSuccess;
|
||||
while (1) {
|
||||
if (xml->maxIndex == MAX_NODES) {
|
||||
WARN("Error : XML parser is limited to 1024 nodes");
|
||||
return ncclInternalError;
|
||||
}
|
||||
struct mscclXmlNode* node = xml->nodes+xml->maxIndex;
|
||||
memset(node, 0, sizeof(struct mscclXmlNode));
|
||||
NCCLCHECK(mscclXmlGetNode(file, node));
|
||||
if (node->type == NODE_TYPE_NONE) {
|
||||
if (head) {
|
||||
WARN("XML Parse : unterminated %s", head->name);
|
||||
return ncclInternalError;
|
||||
} else {
|
||||
// All done
|
||||
return ncclSuccess;
|
||||
}
|
||||
}
|
||||
if (head && node->type == NODE_TYPE_CLOSE) {
|
||||
if (strcmp(node->name, head->name) != 0) {
|
||||
WARN("XML Mismatch : %s / %s", head->name, node->name);
|
||||
return ncclInternalError;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
int found = 0;
|
||||
for (int h=0; h<nHandlers; h++) {
|
||||
if (strcmp(node->name, handlers[h].name) == 0) {
|
||||
if (head) head->subs[head->nSubs++] = node;
|
||||
node->parent = head;
|
||||
node->nSubs = 0;
|
||||
xml->maxIndex++;
|
||||
NCCLCHECK(handlers[h].func(file, xml, node));
|
||||
found = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!found) {
|
||||
if (nHandlers) INFO(NCCL_GRAPH, "Ignoring element %s", node->name);
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xml, node, NULL, 0));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ncclResult_t mscclAlgoXmlStep(FILE* file, struct mscclXml* xml, struct mscclXmlNode* head) {
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xml, head, NULL, 1));
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclAlgoXmlThreadBlock(FILE* file, struct mscclXml* xmlGraph, struct mscclXmlNode* head) {
|
||||
struct mscclXmlHandler handlers[] = { { "step", mscclAlgoXmlStep } };
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xmlGraph, head, handlers, 1));
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static int currentRank;
|
||||
|
||||
ncclResult_t mscclAlgoXmlGpu(FILE* file, struct mscclXml* xmlGraph, struct mscclXmlNode* head) {
|
||||
int thisrank;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(head, "id", &thisrank));
|
||||
if (thisrank == currentRank) {
|
||||
struct mscclXmlHandler handlers[] = { { "tb", mscclAlgoXmlThreadBlock } };
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xmlGraph, head, handlers, 1));
|
||||
} else {
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xmlGraph, head, NULL, 0));
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclAlgoXmlAlgo(FILE* file, struct mscclXml* xmlGraph, struct mscclXmlNode* head) {
|
||||
struct mscclXmlHandler handlers[] = { { "gpu", mscclAlgoXmlGpu } };
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xmlGraph, head, handlers, 1));
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclAlgoXmlLoad(const char* xmlFilePath, struct mscclXml* xml, int rank) {
|
||||
currentRank = rank;
|
||||
FILE* file = fopen(xmlFilePath, "r");
|
||||
if (file == NULL) {
|
||||
WARN("Could not open MSCCL XML algorithm file %s : %s", xmlFilePath, strerror(errno));
|
||||
return ncclSystemError;
|
||||
}
|
||||
struct mscclXmlHandler handlers[] = { { "algo", mscclAlgoXmlAlgo } };
|
||||
xml->maxIndex = 0;
|
||||
NCCLCHECK(mscclXmlLoadSub(file, xml, NULL, handlers, 1));
|
||||
fclose(file);
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclGetBufferType(const char* str, uint8_t* output) {
|
||||
if (strcmp(str, "i") == 0) {
|
||||
*output = MSCCL_INPUT_BUFFER;
|
||||
} else if (strcmp(str, "o") == 0) {
|
||||
*output = MSCCL_OUTPUT_BUFFER;
|
||||
} else if (strcmp(str, "s") == 0) {
|
||||
*output = MSCCL_SCRATCH_BUFFER;
|
||||
} else {
|
||||
WARN("type of buffer is not supported: %s", str);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclCheckBufferBounds(int bufferType, int offset, int nInputChunks, int nOutputChunks, int nScratchChunks) {
|
||||
if (bufferType == MSCCL_INPUT_BUFFER) {
|
||||
if (offset < -1 || offset >= nInputChunks) {
|
||||
WARN("Incorrect offset set for input buffer: offset: %d maximum allowed: %d", offset, nInputChunks);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
} else if (bufferType == MSCCL_OUTPUT_BUFFER) {
|
||||
if (offset < -1 || offset >= nOutputChunks) {
|
||||
WARN("Incorrect offset set for output buffer: offset: %d maximum allowed: %d", offset, nOutputChunks);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
} else if (bufferType == MSCCL_SCRATCH_BUFFER) {
|
||||
if (offset < -1 || offset >= nScratchChunks) {
|
||||
WARN("Incorrect offset set for scratch buffer: offset: %d maximum allowed: %d", offset, nScratchChunks);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclProtocolStrToId(const char *protocol, int *protocolId) {
|
||||
if (strcmp(protocol, "Simple") == 0) {
|
||||
*protocolId = NCCL_PROTO_SIMPLE;
|
||||
} else if (strcmp(protocol, "LL128") == 0) {
|
||||
*protocolId = NCCL_PROTO_LL128;
|
||||
} else if (strcmp(protocol, "LL") == 0) {
|
||||
*protocolId = NCCL_PROTO_LL;
|
||||
} else {
|
||||
WARN("MSCCL: protocol %s is not supported.", protocol);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclGetAlgoFromXmlFile(const char* str, struct mscclAlgo* algo, int rank) {
|
||||
struct mscclXml* xml;
|
||||
NCCLCHECK(ncclCalloc(&xml, 1));
|
||||
NCCLCHECK(mscclAlgoXmlLoad(str, xml, rank));
|
||||
|
||||
// zeroing out all entries.
|
||||
memset(algo, 0, sizeof(struct mscclAlgo));
|
||||
struct mscclXmlNode* topNode;
|
||||
NCCLCHECK(mscclXmlFindTag(xml, "algo", &topNode));
|
||||
|
||||
int nChunksPerLoop;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(topNode, "nchunksperloop", &nChunksPerLoop));
|
||||
algo->nChunksPerLoop = nChunksPerLoop;
|
||||
|
||||
int nChannels;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(topNode, "nchannels", &nChannels));
|
||||
algo->nChannels = nChannels;
|
||||
|
||||
int nGpus;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(topNode, "ngpus", &nGpus));
|
||||
algo->nRanks = nGpus;
|
||||
|
||||
const char* protocol;
|
||||
NCCLCHECK(mscclXmlGetAttrStr(topNode, "proto", &protocol));
|
||||
NCCLCHECK(mscclProtocolStrToId(protocol, &algo->protocol));
|
||||
|
||||
algo->sizeMultiplier = 1;
|
||||
algo->chunkSteps = MSCCL_CHUNKSTEPS;
|
||||
algo->sliceSteps = MSCCL_SLICESTEPS;
|
||||
const char* coll;
|
||||
NCCLCHECK(mscclXmlGetAttrStr(topNode, "coll", &coll));
|
||||
if (strcmp(coll, "reduce") == 0) {
|
||||
algo->chunkSteps = REDUCE_CHUNKSTEPS;
|
||||
algo->sliceSteps = REDUCE_SLICESTEPS;
|
||||
algo->func = mscclFuncReduce;
|
||||
} else if (strcmp(coll, "broadcast") == 0) {
|
||||
algo->chunkSteps = BROADCAST_CHUNKSTEPS;
|
||||
algo->sliceSteps = BROADCAST_SLICESTEPS;
|
||||
algo->func = mscclFuncBroadcast;
|
||||
} else if (strcmp(coll, "allreduce") == 0) {
|
||||
algo->chunkSteps = ALLREDUCE_CHUNKSTEPS;
|
||||
algo->sliceSteps = ALLREDUCE_SLICESTEPS;
|
||||
algo->func = mscclFuncAllReduce;
|
||||
} else if (strcmp(coll, "reducescatter") == 0) {
|
||||
algo->sizeMultiplier = nGpus;
|
||||
algo->chunkSteps = REDUCESCATTER_CHUNKSTEPS;
|
||||
algo->sliceSteps = REDUCESCATTER_SLICESTEPS;
|
||||
algo->func = mscclFuncReduceScatter;
|
||||
} else if (strcmp(coll, "allgather") == 0) {
|
||||
algo->sizeMultiplier = nGpus;
|
||||
algo->chunkSteps = ALLGATHER_CHUNKSTEPS;
|
||||
algo->sliceSteps = ALLGATHER_SLICESTEPS;
|
||||
algo->func = mscclFuncAllGather;
|
||||
} else if (strcmp(coll, "send") == 0) {
|
||||
algo->func = mscclFuncSend;
|
||||
} else if (strcmp(coll, "recv") == 0) {
|
||||
algo->func = mscclFuncRecv;
|
||||
} else if (strcmp(coll, "gather") == 0) {
|
||||
algo->func = mscclFuncGather;
|
||||
} else if (strcmp(coll, "scatter") == 0) {
|
||||
algo->func = mscclFuncScatter;
|
||||
} else if (strcmp(coll, "alltoall") == 0) {
|
||||
algo->sizeMultiplier = nGpus;
|
||||
algo->func = mscclFuncAllToAll;
|
||||
} else if (strcmp(coll, "alltoallv") == 0) {
|
||||
algo->func = mscclFuncAllToAllv;
|
||||
} else {
|
||||
WARN("MSCCL: unsupported collective: %s", coll);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
|
||||
int64_t minBytes;
|
||||
NCCLCHECK(mscclXmlGetAttrInt64(topNode, "minBytes", &minBytes));
|
||||
algo->minBytes = minBytes;
|
||||
|
||||
int64_t maxBytes;
|
||||
NCCLCHECK(mscclXmlGetAttrInt64(topNode, "maxBytes", &maxBytes));
|
||||
algo->maxBytes = maxBytes;
|
||||
|
||||
int inplace;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(topNode, "inplace", &inplace));
|
||||
algo->inPlace = (bool)inplace;
|
||||
|
||||
int outofplace;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(topNode, "outofplace", &outofplace));
|
||||
algo->outOfPlace = (bool)outofplace;
|
||||
|
||||
algo->hasReduce = false;
|
||||
|
||||
for (int s=0; s<topNode->nSubs; s++) {
|
||||
struct mscclXmlNode* node = topNode->subs[s];
|
||||
if (strcmp(node->name, "gpu") == 0) {
|
||||
int blockExists[MSCCL_MAX_NUM_THREAD_BLOCKS];
|
||||
memset(blockExists, 0, sizeof(int[MSCCL_MAX_NUM_THREAD_BLOCKS]));
|
||||
int id, nScratchChunks, nInputChunks, nOutputChunks;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(node, "id", &id));
|
||||
if (id == rank) {
|
||||
NCCLCHECK(mscclXmlGetAttrInt(node, "i_chunks", &nInputChunks));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(node, "o_chunks", &nOutputChunks));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(node, "s_chunks", &nScratchChunks));
|
||||
if (nScratchChunks < 0) {
|
||||
WARN("MSCCL: nScratchChunks must be not negative. nScratchChunks: %d", nScratchChunks);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
algo->nScratchChunks = nScratchChunks;
|
||||
for (int t=0; t<node->nSubs; t++) {
|
||||
struct mscclXmlNode* threadBlockNode = node->subs[t];
|
||||
if (strcmp(threadBlockNode->name, "tb") == 0) {
|
||||
int bid, recvPeer, sendPeer, channelId;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(threadBlockNode, "id", &bid));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(threadBlockNode, "recv", &recvPeer));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(threadBlockNode, "send", &sendPeer));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(threadBlockNode, "chan", &channelId));
|
||||
if (bid < 0) {
|
||||
WARN("MSCCL: bid must be not negative. bid: %d", bid);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
if (bid >= MSCCL_MAX_NUM_THREAD_BLOCKS) {
|
||||
WARN("MSCCL: too many thread blocks are requested. Max thread blocks: %d", MSCCL_MAX_NUM_THREAD_BLOCKS);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
if (blockExists[bid]) {
|
||||
WARN("MSCCL: duplicate thread block id %d for MSCCL", bid);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
blockExists[bid] = 1;
|
||||
|
||||
if (recvPeer == id || sendPeer == id) {
|
||||
WARN("MSCCL: peer (%d,%d) and gpu id (%d) must be different", recvPeer, sendPeer, id);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
struct mscclThreadBlock* sTB = &algo->mscclTBs[bid];
|
||||
sTB->nSteps = 0;
|
||||
if (recvPeer < -1 || sendPeer < -1) {
|
||||
WARN("MSCCL: wrong recvPeer (%d) or sendPeer (%d) in thread block %d on gpu %d", recvPeer, sendPeer, bid, id);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
|
||||
if (recvPeer == id || sendPeer == id) {
|
||||
WARN("MSCCL: recvPeer (%d) or sendPeer (%d) for thread block %d cannot be gpu %d", recvPeer, sendPeer, bid, id);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
|
||||
sTB->recvPeer = recvPeer;
|
||||
sTB->sendPeer = sendPeer;
|
||||
if (channelId < 0 || channelId > MAXCHANNELS) {
|
||||
WARN("MSCCL: threadblock %d on GPU %d has an invalid channel %d", bid, id, channelId);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
sTB->channelId = channelId;
|
||||
|
||||
// setting the summary of the msccl algorithm in msccl channels
|
||||
mscclChannelInfo* mscclChannel = &algo->mscclChannels[sTB->channelId];
|
||||
|
||||
int numDependencies = 0;
|
||||
int oldDependencePointer = 0; // Indicator of where the dependencies started for nop
|
||||
|
||||
int oldReductionDstBuffer = -1; // Indicator of last reduction buffer name; -1 means that last one wasn't a compatible reduction
|
||||
int oldReductionDstOffset = -1; // Indicator of last reduction buffer index
|
||||
int oldReductionSrcBuffer = -1; //
|
||||
int numReductions = 0;
|
||||
|
||||
int numTransfers = 0;
|
||||
for (int st=0; st<threadBlockNode->nSubs; st++) {
|
||||
struct mscclXmlNode* stepNode = threadBlockNode->subs[st];
|
||||
if (strcmp(stepNode->name, "step") == 0) {
|
||||
int s, srcOffset, dstOffset, dependBid, dependStep, hasDependence, count;
|
||||
const char* srcBuffer, * dstBuffer, * type;
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "s", &s));
|
||||
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "srcoff", &srcOffset));
|
||||
NCCLCHECK(mscclXmlGetAttrStr(stepNode, "srcbuf", &srcBuffer));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "dstoff", &dstOffset));
|
||||
NCCLCHECK(mscclXmlGetAttrStr(stepNode, "dstbuf", &dstBuffer));
|
||||
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "cnt", &count));
|
||||
NCCLCHECK(mscclXmlGetAttrStr(stepNode, "type", &type));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "depid", &dependBid));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "deps", &dependStep));
|
||||
NCCLCHECK(mscclXmlGetAttrInt(stepNode, "hasdep", &hasDependence));
|
||||
|
||||
if (s >= MSCCL_MAX_NUM_STEPS){
|
||||
WARN("MSCCL: too many steps are requested. Max number of steps: %d, requested: %d", MSCCL_MAX_NUM_STEPS, s+1);
|
||||
return ncclInternalError;
|
||||
}
|
||||
if (s < 0){
|
||||
WARN("MSCCL: step must be positive: step %d", s);
|
||||
return ncclInternalError;
|
||||
}
|
||||
|
||||
int hasSend = 0;
|
||||
int hasRecv = 0;
|
||||
int checkSrc = 0;
|
||||
int checkDst = 0;
|
||||
int transferType = -1; // -1 indicate a nop
|
||||
if (strcmp(type, "s") == 0) {
|
||||
transferType = MSCCL_SEND;
|
||||
hasSend = 1;
|
||||
checkSrc = 1;
|
||||
} else if (strcmp(type, "r") == 0) {
|
||||
transferType = MSCCL_RECV;
|
||||
hasRecv = 1;
|
||||
checkDst = 1;
|
||||
} else if (strcmp(type, "rcs") == 0) {
|
||||
transferType = MSCCL_RECV_COPY_SEND;
|
||||
hasSend = 1;
|
||||
hasRecv = 1;
|
||||
checkDst = 1;
|
||||
} else if (strcmp(type, "rrs") == 0) {
|
||||
transferType = MSCCL_RECV_REDUCE_SEND;
|
||||
hasSend = 1;
|
||||
hasRecv = 1;
|
||||
checkSrc = 1;
|
||||
algo->hasReduce = true;
|
||||
} else if (strcmp(type, "rrc") == 0) {
|
||||
transferType = MSCCL_RECV_REDUCE_COPY;
|
||||
hasRecv = 1;
|
||||
algo->hasReduce = true;
|
||||
} else if (strcmp(type, "rrcs") == 0) {
|
||||
transferType = MSCCL_RECV_REDUCE_COPY_SEND;
|
||||
hasRecv = 1;
|
||||
hasSend = 1;
|
||||
checkSrc = 1;
|
||||
checkDst = 1;
|
||||
algo->hasReduce = true;
|
||||
} else if (strcmp(type, "cpy") == 0) {
|
||||
transferType = MSCCL_LOCAL_COPY;
|
||||
checkSrc = 1;
|
||||
checkDst = 1;
|
||||
} else if (strcmp(type, "re") == 0) {
|
||||
transferType = MSCCL_REDUCE;
|
||||
checkSrc = 1;
|
||||
checkDst = 1;
|
||||
algo->hasReduce = true;
|
||||
} else if (strcmp(type, "nop") == 0) {
|
||||
transferType = -1;
|
||||
} else {
|
||||
WARN("MSCCL: type of transfer is not supported: %s", type);
|
||||
return ncclInternalError;
|
||||
}
|
||||
|
||||
if (dependBid >= 0) {
|
||||
sTB->dependentBid[numDependencies] = dependBid;
|
||||
sTB->dependentStep[numDependencies] = dependStep;
|
||||
numDependencies++;
|
||||
}
|
||||
|
||||
uint8_t srcBufferInt = 0;
|
||||
uint8_t dstBufferInt = 0;
|
||||
NCCLCHECK(mscclGetBufferType(srcBuffer, &srcBufferInt));
|
||||
NCCLCHECK(mscclGetBufferType(dstBuffer, &dstBufferInt));
|
||||
|
||||
int continuationOfReductions = 0;
|
||||
// Analyze to see if this is in the same list of reductions for them to be chained
|
||||
if (transferType == MSCCL_REDUCE) {
|
||||
if (oldReductionDstBuffer == dstBufferInt && oldReductionDstOffset == dstOffset && oldReductionSrcBuffer == srcBufferInt && dependBid == -1) {
|
||||
numTransfers--; // reuse the same transfer
|
||||
continuationOfReductions = 1;
|
||||
} else {
|
||||
oldReductionDstBuffer = -1;
|
||||
oldReductionDstOffset = -1;
|
||||
}
|
||||
}
|
||||
|
||||
if (transferType != -1) {
|
||||
struct mscclTransmission* mscclTran = &sTB->transmissions[numTransfers];
|
||||
mscclTran->type = transferType;
|
||||
mscclTran->srcOffset = srcOffset;
|
||||
mscclTran->srcBuffer = srcBufferInt;
|
||||
mscclTran->srcOffset = srcOffset;
|
||||
mscclTran->dstBuffer = dstBufferInt;
|
||||
mscclTran->dstOffset = dstOffset;
|
||||
|
||||
if (count < 0 || count >= MSCCL_MAX_COUNT){
|
||||
WARN("MSCCL: count (%d) must be positive and less than %d", count, MSCCL_MAX_COUNT);
|
||||
return ncclInternalError;
|
||||
}
|
||||
|
||||
mscclTran->count = count;
|
||||
|
||||
if (hasSend) {
|
||||
if (sendPeer < 0) {
|
||||
WARN("MSCCL: there is a send in thread block %d on GPU %d without a sendPeer.", bid, id);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
if (mscclChannel->nSendPeers >= MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL) {
|
||||
WARN("MSCCL: too many sends per channel. Max allowed %d", MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
|
||||
struct mscclChannelPeerInfo* sendPeerInfo = &mscclChannel->sendPeerInfo[mscclChannel->nSendPeers];
|
||||
sendPeerInfo->nTransmissionsOfCount[count]++;
|
||||
}
|
||||
if (hasRecv) {
|
||||
if (recvPeer < 0) {
|
||||
WARN("MSCCL: there is a recv in thread block %d on GPU %d without a recvPeer.", bid, id);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
if (mscclChannel->nRecvPeers >= MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL) {
|
||||
WARN("MSCCL: too many recvs per channel. Max allowed %d", MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
struct mscclChannelPeerInfo* recvPeerInfo = &mscclChannel->recvPeerInfo[mscclChannel->nRecvPeers];
|
||||
recvPeerInfo->nTransmissionsOfCount[count]++;
|
||||
}
|
||||
|
||||
if (checkSrc) NCCLCHECK(mscclCheckBufferBounds(mscclTran->srcBuffer, mscclTran->srcOffset, nInputChunks, nOutputChunks, nScratchChunks));
|
||||
if (checkDst) NCCLCHECK(mscclCheckBufferBounds(mscclTran->dstBuffer, mscclTran->dstOffset, nInputChunks, nOutputChunks, nScratchChunks));
|
||||
|
||||
if (!continuationOfReductions) {
|
||||
mscclTran->dependencePointer = oldDependencePointer;
|
||||
mscclTran->numDependencies = numDependencies - oldDependencePointer;
|
||||
if (mscclTran->numDependencies > 0 && dependBid < 0) {
|
||||
WARN("MSCCL: when there is a chain of dependencies, the last reduction must be a part of the first immediate instruction. Detected for GPU %d, thread block %d, and step %d. XML will be ignored.", id, bid, s);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
oldDependencePointer = numDependencies;
|
||||
}
|
||||
|
||||
// reduction related pointers
|
||||
if (transferType != MSCCL_REDUCE) {
|
||||
oldReductionDstBuffer = -1;
|
||||
oldReductionDstOffset = -1;
|
||||
oldReductionSrcBuffer = -1;
|
||||
} else {
|
||||
if (oldReductionDstBuffer == -1) { // if this is the first reduction
|
||||
mscclTran->reductionPointer = numReductions;
|
||||
}
|
||||
sTB->reductionSrcOffsets[numReductions] = mscclTran->srcOffset;
|
||||
numReductions++;
|
||||
mscclTran->numReductions = numReductions - mscclTran->reductionPointer;
|
||||
|
||||
if (hasDependence || numReductions == MSCCL_MAX_REDUCE_FUSION) {
|
||||
oldReductionDstBuffer = -1;
|
||||
oldReductionDstOffset = -1;
|
||||
} else {
|
||||
oldReductionDstBuffer = mscclTran->dstBuffer;
|
||||
oldReductionDstOffset = mscclTran->dstOffset;
|
||||
oldReductionSrcBuffer = mscclTran->srcBuffer;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
if (hasDependence != 0 && hasDependence != 1) {
|
||||
WARN("MSCCL: hasDependence needs to be 0 or 1, but it was %d", hasDependence);
|
||||
return ncclInternalError;
|
||||
}
|
||||
mscclTran->hasDependence = hasDependence;
|
||||
|
||||
numTransfers++;
|
||||
sTB->nSteps = numTransfers;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// finish up mscclChannel calculation
|
||||
|
||||
for (int c = 0; c < MSCCL_MAX_COUNT; c++) {
|
||||
struct mscclChannelPeerInfo* sendPeer = &mscclChannel->sendPeerInfo[mscclChannel->nSendPeers];
|
||||
if (sendPeer->nTransmissionsOfCount[c] > 0) {
|
||||
sendPeer->existingCounts[sendPeer->nExistingCounts] = c;
|
||||
sendPeer->nExistingCounts++;
|
||||
}
|
||||
struct mscclChannelPeerInfo* recvPeer = &mscclChannel->recvPeerInfo[mscclChannel->nRecvPeers];
|
||||
if (recvPeer->nTransmissionsOfCount[c] > 0) {
|
||||
recvPeer->existingCounts[recvPeer->nExistingCounts] = c;
|
||||
recvPeer->nExistingCounts++;
|
||||
}
|
||||
}
|
||||
|
||||
if (sTB->sendPeer >= 0) {
|
||||
mscclChannel->sendPeerInfo[mscclChannel->nSendPeers].peer = sTB->sendPeer;
|
||||
mscclChannel->nSendPeers++;
|
||||
}
|
||||
if (sTB->recvPeer >= 0) {
|
||||
mscclChannel->recvPeerInfo[mscclChannel->nRecvPeers].peer = sTB->recvPeer;
|
||||
mscclChannel->nRecvPeers++;
|
||||
}
|
||||
}
|
||||
}
|
||||
// make sure that thread blocks are in order. Something like 0, 2, 3 is not allowed.
|
||||
if (blockExists[0] == 1) {
|
||||
algo->nBlocks = 1;
|
||||
}
|
||||
for (int i = 1; i < MSCCL_MAX_NUM_THREAD_BLOCKS; i++) {
|
||||
if (blockExists[i] == 1 && blockExists[i-1] == 0) {
|
||||
WARN("MSCCL: thread block %d is missing", i);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
if (blockExists[i] == 1) {
|
||||
algo->nBlocks = i+1;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
free(xml);
|
||||
return ncclSuccess;
|
||||
}
|
||||
@@ -0,0 +1,284 @@
|
||||
/*************************************************************************
|
||||
* Copyright (c) Microsoft Corporation.
|
||||
* Licensed under the MIT License.
|
||||
************************************************************************/
|
||||
|
||||
#include "checks.h"
|
||||
#include "collectives.h"
|
||||
#include "proxy.h"
|
||||
#include "transport.h"
|
||||
|
||||
#include "msccl/msccl_lifecycle.h"
|
||||
#include "msccl/msccl_kernel.h"
|
||||
#include "msccl/msccl_setup.h"
|
||||
#include "msccl/msccl_status.h"
|
||||
|
||||
ncclResult_t mscclSetupCount(struct mscclAlgo* hostAlgo, ncclComm_t comm, size_t count, ncclDataType_t dataType) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
status.stepSize = comm->buffSizes[hostAlgo->protocol] / NCCL_STEPS;
|
||||
status.chunkSteps = hostAlgo->protocol == NCCL_PROTO_SIMPLE ? hostAlgo->chunkSteps : 1;
|
||||
status.sliceSteps = hostAlgo->protocol == NCCL_PROTO_SIMPLE ? hostAlgo->sliceSteps : 1;
|
||||
status.chunkSize = status.stepSize * status.chunkSteps;
|
||||
status.chunkEffectiveSize = status.chunkSize;
|
||||
if (hostAlgo->protocol == NCCL_PROTO_LL) status.chunkEffectiveSize /= 2;
|
||||
if (hostAlgo->protocol == NCCL_PROTO_LL128) status.chunkEffectiveSize = (status.chunkSize / NCCL_LL128_LINEELEMS) * NCCL_LL128_DATAELEMS;
|
||||
status.dataType = dataType;
|
||||
status.nBytes = count * ncclTypeSize(status.dataType) * hostAlgo->sizeMultiplier;
|
||||
status.maxAllowedCount = std::max((uint32_t)1, (uint32_t)(status.chunkEffectiveSize / DIVUP(status.nBytes, (size_t)(hostAlgo->nChunksPerLoop))));
|
||||
if (status.maxAllowedCount == 0){
|
||||
WARN("MSCCL: something went wrong. Max allowed count is 0\n");
|
||||
return ncclInternalError;
|
||||
}
|
||||
if (status.maxAllowedCount >= MSCCL_MAX_COUNT) {
|
||||
status.maxAllowedCount = MSCCL_MAX_COUNT - 1;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclSetupScratch(struct mscclAlgo* hostAlgo, hipStream_t stream) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
size_t sizeNeeded = (status.nBytes * (size_t)(hostAlgo->nScratchChunks)) / (size_t)(hostAlgo->nChunksPerLoop);
|
||||
if (sizeNeeded > status.scratchBufferSize){
|
||||
CUDACHECK(hipStreamSynchronize(stream));
|
||||
CUDACHECK(hipFree(status.scratchBuffer));
|
||||
NCCLCHECK(ncclCudaCalloc((char**)&status.scratchBuffer, sizeNeeded));
|
||||
status.scratchBufferSize = sizeNeeded;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclSetupSyncFlags(hipStream_t stream) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
if (status.workIndex > (1ULL << (8*sizeof(status.workIndex))) - 2 * NCCL_MAX_OPS - 1) {
|
||||
CUDACHECK(hipMemsetAsync(status.syncFlags, 0, sizeof(struct mscclFlag) * MSCCL_MAX_NUM_THREAD_BLOCKS, stream));
|
||||
status.workIndex = 1; // setting the workIndex back to 1 for next iterations
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclSetupConnections(struct mscclAlgo* hostAlgo, ncclComm_t comm) {
|
||||
// Check whether there is enough channels
|
||||
if (hostAlgo->nChannels > comm->nChannels) {
|
||||
WARN("MSCCL: number of channels available (%d) less than required (%d)", comm->nChannels, hostAlgo->nChannels);
|
||||
return ncclInvalidUsage;
|
||||
}
|
||||
|
||||
// Flag MSCCL connections
|
||||
for (int i = 0; i < hostAlgo->nChannels; i++) {
|
||||
struct mscclChannelInfo* mCh = hostAlgo->mscclChannels + i;
|
||||
|
||||
int sendPeers[MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL];
|
||||
for (int p = 0; p < mCh->nSendPeers; p++) {
|
||||
sendPeers[p] = mCh->sendPeerInfo[p].peer;
|
||||
}
|
||||
|
||||
int recvPeers[MSCCL_MAX_NUM_THREAD_BLOCKS_PER_CHANNEL];
|
||||
for (int p = 0; p < mCh->nRecvPeers; p++) {
|
||||
recvPeers[p] = mCh->recvPeerInfo[p].peer;
|
||||
}
|
||||
|
||||
NCCLCHECK(ncclTransportP2pConnect(comm, i, mCh->nRecvPeers, recvPeers, mCh->nSendPeers, sendPeers, 0 /*connIndex*/));
|
||||
}
|
||||
|
||||
// Connect MSCCL connections
|
||||
mscclSetIsCallerFlag();
|
||||
NCCLCHECK(ncclTransportP2pSetup(comm, NULL, 0));
|
||||
mscclClearIsCallerFlag();
|
||||
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
ncclResult_t mscclSetupProxy(struct mscclAlgo* hostAlgo, ncclComm_t comm) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
struct ncclProxyOp proxyOp = {};
|
||||
proxyOp.connIndex = 0;
|
||||
proxyOp.sliceSteps = status.sliceSteps;
|
||||
proxyOp.chunkSteps = status.chunkSteps;
|
||||
proxyOp.chunkSize = status.chunkSize;
|
||||
proxyOp.protocol = hostAlgo->protocol;
|
||||
proxyOp.dtype = status.dataType;
|
||||
proxyOp.redOp = 0;
|
||||
proxyOp.pattern = 0;
|
||||
proxyOp.root = 0;
|
||||
proxyOp.nbytes = status.stepSize*proxyOp.sliceSteps;
|
||||
proxyOp.opCount = comm->collOpCount;
|
||||
int nLoops = (int)(DIVUP(status.nBytes, (size_t)((size_t)hostAlgo->nChunksPerLoop*(size_t)status.chunkEffectiveSize)));
|
||||
int nLoopsChunkSteps = nLoops * status.chunkSteps;
|
||||
for (int ch = 0; ch < hostAlgo->nChannels; ch++) {
|
||||
proxyOp.channelId = ch;
|
||||
struct mscclChannelInfo* mscclChannel = hostAlgo->mscclChannels + ch;
|
||||
struct ncclChannel* ncclChannel = comm->channels + ch;
|
||||
for (int i = 0; i < mscclChannel->nRecvPeers; i++){
|
||||
struct mscclChannelPeerInfo* recvPeer = mscclChannel->recvPeerInfo + i;
|
||||
int nRecvs = 0;
|
||||
for (int j = 0; j < recvPeer->nExistingCounts; j++){
|
||||
int c = recvPeer->existingCounts[j];
|
||||
int nStepsInCount = DIVUP(c+1, status.maxAllowedCount);
|
||||
nRecvs += recvPeer->nTransmissionsOfCount[c] * nStepsInCount;
|
||||
}
|
||||
proxyOp.nsteps = nLoopsChunkSteps * nRecvs;
|
||||
if (proxyOp.nsteps > 0) {
|
||||
NCCLCHECK(mscclSaveProxy(ncclChannel, proxyRecv, recvPeer->peer, &proxyOp, 0));
|
||||
}
|
||||
}
|
||||
for (int i=0; i<mscclChannel->nSendPeers; i++){
|
||||
struct mscclChannelPeerInfo* sendPeer = &mscclChannel->sendPeerInfo[i];
|
||||
int nSends = 0;
|
||||
for (int j = 0; j < sendPeer->nExistingCounts; j++){
|
||||
int c = sendPeer->existingCounts[j];
|
||||
int nStepsInCount = DIVUP(c+1, status.maxAllowedCount);
|
||||
nSends += sendPeer->nTransmissionsOfCount[c] * nStepsInCount;
|
||||
}
|
||||
proxyOp.nsteps = nLoopsChunkSteps * nSends;
|
||||
if (proxyOp.nsteps > 0) {
|
||||
NCCLCHECK(mscclSaveProxy(ncclChannel, proxySend, sendPeer->peer, &proxyOp, 0));
|
||||
}
|
||||
}
|
||||
}
|
||||
NCCLCHECK(ncclProxyStart(comm));
|
||||
comm->collOpCount++;
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
static ncclResult_t hostToDevRedOp(
|
||||
ncclDevRedOpFull *opFull, ncclRedOp_t op, ncclDataType_t datatype, ncclComm *comm
|
||||
) {
|
||||
union {
|
||||
int8_t i8;
|
||||
uint8_t u8;
|
||||
int32_t i32;
|
||||
uint32_t u32;
|
||||
int64_t i64;
|
||||
uint64_t u64;
|
||||
half f16;
|
||||
#if defined(RCCL_BFLOAT16)
|
||||
rccl_bfloat16 bf16;
|
||||
#endif
|
||||
float f32;
|
||||
double f64;
|
||||
void *ptr;
|
||||
};
|
||||
u64 = 0;
|
||||
opFull->scalarArgIsPtr = false;
|
||||
switch (int(op)) {
|
||||
case ncclSum: opFull->op = ncclDevSum; break;
|
||||
case ncclProd: opFull->op = ncclDevProd; break;
|
||||
case ncclMax: opFull->op = ncclDevMax; break;
|
||||
case ncclMin: opFull->op = ncclDevMin; break;
|
||||
case ncclAvg:
|
||||
switch ((int)datatype) {
|
||||
case ncclInt8: case ncclInt32: case ncclInt64:
|
||||
case ncclUint8: case ncclUint32: case ncclUint64:
|
||||
opFull->op = ncclDevSumPostDiv;
|
||||
u64 = comm->nRanks;
|
||||
break;
|
||||
case ncclFloat16:
|
||||
opFull->op = ncclDevPreMulSum;
|
||||
f16 = __float2half(float(1.0/comm->nRanks)); // __double2half not supported pre CUDA 11.x
|
||||
break;
|
||||
#if defined(RCCL_BFLOAT16)
|
||||
case ncclBfloat16:
|
||||
opFull->op = ncclDevPreMulSum;
|
||||
bf16 = (rccl_bfloat16)(float(1.0/comm->nRanks));
|
||||
break;
|
||||
#endif
|
||||
case ncclFloat32:
|
||||
opFull->op = ncclDevPreMulSum;
|
||||
f32 = float(1.0/comm->nRanks);
|
||||
break;
|
||||
case ncclFloat64:
|
||||
opFull->op = ncclDevPreMulSum;
|
||||
f64 = 1.0/comm->nRanks;
|
||||
break;
|
||||
}
|
||||
opFull->scalarArgIsPtr = false;
|
||||
opFull->scalarArg = u64;
|
||||
break;
|
||||
default: // user created
|
||||
int ix = int(ncclUserRedOpMangle(comm, op)) - int(ncclNumOps);
|
||||
ncclUserRedOp *user = &comm->userRedOps[ix];
|
||||
if (datatype != user->datatype) {
|
||||
WARN("Data type supplied to user-created ncclRedOp_t does not match type "
|
||||
"given to reduction operation");
|
||||
return ncclInvalidArgument;
|
||||
}
|
||||
*opFull = user->opFull;
|
||||
break;
|
||||
}
|
||||
return ncclSuccess;
|
||||
}
|
||||
|
||||
#define MSCCL_KERNEL_ENTRY_DEVREDOP_NULL() \
|
||||
nullptr, \
|
||||
nullptr, \
|
||||
nullptr
|
||||
|
||||
#define MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, type) \
|
||||
(void *)MSCCL_KERNEL_ENTRY_NAME(devredop, type, LL), \
|
||||
(void *)MSCCL_KERNEL_ENTRY_NAME(devredop, type, LL128), \
|
||||
(void *)MSCCL_KERNEL_ENTRY_NAME(devredop, type, Simple)
|
||||
|
||||
#define MSCCL_KERNEL_ENTRY_DEVREDOP(devredop) \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int8_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint8_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int32_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint32_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int64_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint64_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, half), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, float), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, double), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, rccl_bfloat16)
|
||||
|
||||
#define MSCCL_KERNEL_ENTRY_DEVREDOP_NOFLOAT(devredop) \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int8_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint8_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int32_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint32_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, int64_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_TYPE(devredop, uint64_t), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_NULL(), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_NULL(), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_NULL(), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_NULL()
|
||||
|
||||
#define MSCCL_KERNEL_ENTRY() \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP(Sum), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP(Prod), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP(Min), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP(Max), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP(PreMulSum), \
|
||||
MSCCL_KERNEL_ENTRY_DEVREDOP_NOFLOAT(SumPostDiv)
|
||||
|
||||
void* mscclKernelEntries[ncclNumDevRedOps * ncclNumTypes * NCCL_NUM_PROTOCOLS] = {
|
||||
MSCCL_KERNEL_ENTRY()
|
||||
};
|
||||
|
||||
ncclResult_t mscclSetupKernel(const void* sendBuff, void* recvBuff, size_t count,
|
||||
ncclDataType_t dataType, ncclRedOp_t op, struct mscclAlgo* hostAlgo, struct mscclAlgo* devAlgo,
|
||||
ncclComm_t comm, hipStream_t stream) {
|
||||
mscclStatus& status = mscclGetStatus();
|
||||
dim3 grid = {(uint32_t)hostAlgo->nBlocks, 1, 1};
|
||||
dim3 block = {NCCL_MAX_NTHREADS, 1, 1};
|
||||
ncclDevRedOpFull opFull;
|
||||
NCCLCHECK(hostToDevRedOp(&opFull, op, dataType, comm));
|
||||
|
||||
mscclWork work;
|
||||
work.syncFlags = status.syncFlags;
|
||||
work.scratchBuffer = status.scratchBuffer;
|
||||
work.sendBuff = sendBuff;
|
||||
work.recvBuff = recvBuff;
|
||||
work.count = count * hostAlgo->sizeMultiplier; // count is sum of all ranks in MSCCL kernel
|
||||
work.redOpArg = opFull.scalarArg;
|
||||
work.workIndex = status.workIndex;
|
||||
work.nChunksPerLoop = hostAlgo->nChunksPerLoop;
|
||||
work.maxAllowedCount = status.maxAllowedCount;
|
||||
work.hasReduce = hostAlgo->hasReduce;
|
||||
work.redOpArgIsPtr = opFull.scalarArgIsPtr;
|
||||
|
||||
void *args[3] = {&comm->devComm, &devAlgo, &work};
|
||||
void *func = mscclKernelEntries[(opFull.op * ncclNumTypes + dataType) * NCCL_NUM_PROTOCOLS + hostAlgo->protocol];
|
||||
CUDACHECK(hipExtLaunchKernel(func, grid, block, args, 0, stream, NULL, NULL,0));
|
||||
status.workIndex++;
|
||||
return ncclSuccess;
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
/*************************************************************************
|
||||
* Copyright (c) Microsoft Corporation.
|
||||
* Licensed under the MIT License.
|
||||
************************************************************************/
|
||||
|
||||
#include "msccl/msccl_status.h"
|
||||
|
||||
mscclStatus& mscclGetStatus() {
|
||||
static mscclStatus status;
|
||||
return status;
|
||||
}
|
||||
새 이슈에서 참조
사용자 차단