@@ -1291,8 +1291,6 @@ namespace hccl for (u32 i = 0; i < iterServ->second.size(); i++) { if (iterServ->second[i].deviceInfo.devicePhyId != HOST_DEVICE_ID) { enableP2PDevices_.push_back(iterServ->second[i].deviceInfo.devicePhyId); - HCCL_INFO("[InitPreResource]Insert all devices in the current server[%s] into the enablep2p queue, devicePhyId[%d]", - iterServ->second[i].serverId.c_str(), iterServ->second[i].deviceInfo.devicePhyId); } } if (deviceType_ != DevType::DEV_TYPE_310P3 && !isStandardCard_) { @@ -2085,160 +2083,6 @@ namespace hccl return cclBufferManager_.InitCCLbuffer(inCCLbufferSize, outCCLbufferSize); } - HcclResult HcclCommunicator::SetGroupMainStream(HcclRtStream sendRecvMainStream){ - Stream streamObj(sendRecvMainStream); - groupSendRecvMainStream = streamObj; - HCCL_INFO("groupSendRecvMainStream[%p], id[%d]", groupSendRecvMainStream, groupSendRecvMainStream.id()); - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::CreateGroupSendNotifies(){ - u32 sendStreamNum = std::min(nSend, MAX_CONCURRENT); - if (groupSendNotifies.size() >= GROUP_SYNC_NOTIFY_NUM * sendStreamNum) { - return HCCL_SUCCESS; - } - u32 currentSize = groupSendNotifies.size(); - groupSendNotifies.resize(GROUP_SYNC_NOTIFY_NUM * sendStreamNum); - for (u32 i = currentSize; i < GROUP_SYNC_NOTIFY_NUM * sendStreamNum; i++){ - EXECEPTION_CATCH((groupSendNotifies[i] = std::make_shared()), return HCCL_E_PTR); - CHK_RET(groupSendNotifies[i]->Init(NotifyLoadType::HOST_NOTIFY)); - CHK_RET(groupSendNotifies[i]->SetIpc()); - HCCL_INFO("Create Send Notify[%d]", i); - } - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::CreateGroupRecvNotifies(){ - u32 recvStreamNum = std::min(nRecv, MAX_CONCURRENT); - if (groupRecvNotifies.size() >= GROUP_SYNC_NOTIFY_NUM * recvStreamNum) { - return HCCL_SUCCESS; - } - u32 currentSize = groupRecvNotifies.size(); - groupRecvNotifies.resize(GROUP_SYNC_NOTIFY_NUM * recvStreamNum); - for (u32 i = currentSize; i < GROUP_SYNC_NOTIFY_NUM * recvStreamNum; i++){ - EXECEPTION_CATCH((groupRecvNotifies[i] = std::make_shared()), return HCCL_E_PTR); - CHK_RET(groupRecvNotifies[i]->Init(NotifyLoadType::HOST_NOTIFY)); - CHK_RET(groupRecvNotifies[i]->SetIpc()); - HCCL_INFO("Create Recv Notify[%d]", i); - } - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::CreateGroupSendStreams(){ - u32 sendStreamNum = std::min(nSend, MAX_CONCURRENT); - if (groupSendStreams.size() >= sendStreamNum){ - return HCCL_SUCCESS; - } - - u32 currentSize = groupSendStreams.size(); - groupSendStreams.resize(sendStreamNum); - for (u32 i = currentSize; i < sendStreamNum; i++){ - groupSendStreams[i] = Stream(StreamType::STREAM_TYPE_ONLINE); - } - - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::CreateGroupRecvStreams(){ - u32 recvStreamNum = std::min(nRecv, MAX_CONCURRENT); - if (groupRecvStreams.size() >= recvStreamNum){ - return HCCL_SUCCESS; - } - - u32 currentSize = groupRecvStreams.size(); - groupRecvStreams.resize(recvStreamNum); - for (u32 i = currentSize; i < recvStreamNum; i++){ - groupRecvStreams[i] = Stream(StreamType::STREAM_TYPE_ONLINE); - } - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::GroupSyncMainstream(std::unordered_map> &sendIdx2Byte, std::unordered_map> &recvIdx2Byte){ - HcclUs startutime = TIME_NOW(); - u64 sliceSize; - CHK_RET(GetSliceSize(sliceSize)); - u32 sendStreamNum = std::min(nSend, MAX_CONCURRENT); - u32 recvStreamNum = std::min(nRecv, MAX_CONCURRENT); - std::unordered_map sendStreamIters; // ，这决定了需要同步的次数 - std::unordered_map recvStreamIters; // - HCCL_INFO("sendStreamNum[%d], recvStreamNum[%d], sendIdx2Byte.size[%d], recvIdx2Byte.size[%d]", sendStreamNum, recvStreamNum, sendIdx2Byte.size(), recvIdx2Byte.size()); - for (auto &it : sendIdx2Byte){ - for (u32 i = 0; i < it.second.size(); i++){ - sendStreamIters[it.first % sendStreamNum] += (it.second[i] + sliceSize - 1) / sliceSize; //取上整 - HCCL_INFO("sendStreamIters[%d] = %d, sendIdx[%d] bytes[%lld], sliceSize[%lld]", it.first % sendStreamNum, sendStreamIters[it.first % sendStreamNum], it.first, it.second[i], sliceSize); - } - } - for (auto &it : recvIdx2Byte){ - for (u32 i = 0; i < it.second.size(); i++){ - recvStreamIters[it.first % recvStreamNum] += (it.second[i] + sliceSize - 1) / sliceSize; //取上整 - HCCL_INFO("recvStreamIters[%d] = %d, recvIdx[%d] bytes[%lld], sliceSize[%lld]", it.first % recvStreamNum, recvStreamIters[it.first % recvStreamNum], it.first, it.second[i], sliceSize); - } - } - u32 emptySend = 0; - u32 emptyRecv = 0; - // one round(while loop) corresponds one local copy - // one round dispatches M+N Wait, M+N Record to the Mainstream - // M = the counts of sendStream which still has local copy tasks, N = the counts of recvStream which still has local copy tasks - while ((emptySend < sendStreamNum || sendStreamNum == 0) && (emptyRecv < recvStreamNum || recvStreamNum == 0) && sendStreamNum + recvStreamNum > 0) { - std::vector aliveSendTasks; // the size should be M - std::vector aliveRecvTasks; // the size should be N - for (auto it = sendStreamIters.begin(); it != sendStreamIters.end(); ) { - if (it->second > 0) { - aliveSendTasks.push_back(it->first); - it->second--; - ++it; - } else { - emptySend++; - it = sendStreamIters.erase(it); // 删除并返回下一个迭代器，下次进入while循环就不会遍历到这个消耗完了的stream - } - } - - for (auto it = recvStreamIters.begin(); it != recvStreamIters.end(); ) { - if (it->second > 0) { - aliveRecvTasks.push_back(it->first); - it->second--; - ++it; - } else { - emptyRecv++; - it = recvStreamIters.erase(it); // 删除并返回下一个迭代器，下次进入while循环就不会遍历到这个消耗完了的stream - } - } - - for (u32 sendStreamId : aliveSendTasks){ - CHK_RET(LocalNotify::Wait(groupSendRecvMainStream, dispatcher_, groupSendNotifies[GROUP_SYNC_NOTIFY_NUM * sendStreamId], INVALID_VALUE_STAGE)); - HCCL_INFO("Mainstream[%d] Wait, thread id[%d]", groupSendRecvMainStream.id(), std::this_thread::get_id()); - } - for (u32 recvStreamId : aliveRecvTasks){ - CHK_RET(LocalNotify::Wait(groupSendRecvMainStream, dispatcher_, groupRecvNotifies[GROUP_SYNC_NOTIFY_NUM * recvStreamId], INVALID_VALUE_STAGE)); - HCCL_INFO("Mainstream[%d] Wait, thread id[%d]", groupSendRecvMainStream.id(), std::this_thread::get_id()); - } - - for (u32 sendStreamId : aliveSendTasks){ - CHK_RET(LocalNotify::Post(groupSendRecvMainStream, dispatcher_, groupSendNotifies[GROUP_SYNC_NOTIFY_NUM * sendStreamId + 1], INVALID_VALUE_STAGE)); - HCCL_INFO("Mainstream[%d] Post, thread id[%d]", groupSendRecvMainStream.id(), std::this_thread::get_id()); - } - for (u32 recvStreamId : aliveRecvTasks){ - CHK_RET(LocalNotify::Post(groupSendRecvMainStream, dispatcher_, groupRecvNotifies[GROUP_SYNC_NOTIFY_NUM * recvStreamId + 1], INVALID_VALUE_STAGE)); - HCCL_INFO("Mainstream[%d] Post, thread id[%d]", groupSendRecvMainStream.id(), std::this_thread::get_id()); - } - } - HCCL_RUN_INFO("[GroupSyncMainstream] take time [%lld]us.", DURATION_US(TIME_NOW() - startutime)); - - return HCCL_SUCCESS; - } - - HcclResult HcclCommunicator::GroupSubstreamsSync(){ - for(auto substream : groupSendStreams){ - hcclStreamSynchronize(substream.ptr()); - } - - for(auto substream : groupRecvStreams){ - hcclStreamSynchronize(substream.ptr()); - } - - return HCCL_SUCCESS; - } - u32 HcclCommunicator::GetLocalNicPort(NicType nicType) { u32 port = HCCL_INVALID_PORT; @@ -3952,6 +3796,10 @@ namespace hccl opParam.BatchSendRecvDataDes.sendRecvItemsPtr = sendRecvItemsPtr; opParam.BatchSendRecvDataDes.itemNum = itemNum; opParam.opType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV; + opParam.isGroupMode = isGroupMode_; + if (isGroupMode_) { + opParam.aicpuUnfoldMode = true; // A2的GroupSendRecv也走aicpu模式 + } CHK_RET(ExecOp(HcclCMDType::HCCL_CMD_BATCH_SEND_RECV, opParam)); @@ -4050,17 +3898,6 @@ namespace hccl opParam.dstRank = destRank; opParam.opType = HcclCMDType::HCCL_CMD_SEND; opParam.localGroupRank = userRank_; - opParam.isGroupMode = GetGroupMode(); - if (opParam.isGroupMode) { - opParam.aicpuUnfoldMode = false; - opParam.aicpuCacheEnable = 0; - GetNSend(opParam.nSend); - GetNRecv(opParam.nRecv); - GetSendIndex(opParam.iSend); - GetRecvIndex(opParam.iRecv); - opParam.stream = groupSendStreams[opParam.iSend % std::min(opParam.nSend, MAX_CONCURRENT)]; // opParam.stream修改为执行流，也就是从流，目的是为了减少代码修改 - HCCL_INFO("SendSubstream Id [%d], destRank[%d], opParam.iSend[%d]", opParam.stream.id(), destRank, opParam.iSend); - } CHK_RET(ExecOp(HcclCMDType::HCCL_CMD_SEND, opParam)); return HCCL_SUCCESS; @@ -4158,17 +3995,6 @@ namespace hccl opParam.srcRank = srcRank; opParam.opType = HcclCMDType::HCCL_CMD_RECEIVE; opParam.localGroupRank = userRank_; - opParam.isGroupMode = GetGroupMode(); - if (opParam.isGroupMode) { - opParam.aicpuUnfoldMode = false; - opParam.aicpuCacheEnable = 0; - GetNSend(opParam.nSend); - GetNRecv(opParam.nRecv); - GetSendIndex(opParam.iSend); - GetRecvIndex(opParam.iRecv); - opParam.stream = groupRecvStreams[opParam.iRecv % std::min(opParam.nRecv, MAX_CONCURRENT)]; // opParam.stream修改为执行流，也就是从流，目的是为了减少代码修改 - HCCL_INFO("RecvSubstream Id [%d], srcRank[%d], opParam.iRecv[%d]", opParam.stream.id(), srcRank, opParam.iRecv); - } CHK_RET(ExecOp(HcclCMDType::HCCL_CMD_RECEIVE, opParam)); return HCCL_SUCCESS; @@ -4501,6 +4327,7 @@ namespace hccl CHK_RET(RecordOpPara(opType, opParam)); CHK_RET(IncreAllocLink(newTag, opParam, resRequest, resMap_[newTag])); CHK_RET(RankConsistentcyChecker::GetInstance().DelOpPara(opParam.tag)); + opParam.needIncreLink = true; } } @@ -6407,40 +6234,6 @@ namespace hccl algResResponse.cclInputMem = cclBufferManager_.GetInCCLbuffer(); algResResponse.cclOutputMem = cclBufferManager_.GetOutCCLbuffer(); - if (GetGroupMode() && (opType == HcclCMDType::HCCL_CMD_SEND || opType == HcclCMDType::HCCL_CMD_RECEIVE)){// 做CCLBuffer的切分，stream资源、notify资源的分配 - if (opType == HcclCMDType::HCCL_CMD_SEND) { - u32 iSend; - u32 nSend; - CHK_RET(GetSendIndex(iSend)); - CHK_RET(GetNSend(nSend)); - u32 sendStreamNum = std::min(nSend, MAX_CONCURRENT); - // 流分配和Notify分配 - algResResponse.notifiesMain.push_back(groupSendNotifies[GROUP_SYNC_NOTIFY_NUM * (iSend % sendStreamNum)]); - algResResponse.notifiesAux.push_back(groupSendNotifies[GROUP_SYNC_NOTIFY_NUM * (iSend % sendStreamNum) + 1]); - // CCL Buffer分配 - DeviceMem cclin = cclBufferManager_.GetInCCLbuffer(); - u64 sliceSize; - CHK_RET(GetSliceSize(sliceSize)); - algResResponse.cclInputMem = cclin.range(iSend % bufferSliceNum * sliceSize, sliceSize); - HCCL_INFO("Send[%d] algResResponse.cclInputMem[%p], size[%d]", iSend, algResResponse.cclInputMem.ptr(), algResResponse.cclInputMem.size()); - } - else { - u32 iRecv; - u32 nRecv; - CHK_RET(GetRecvIndex(iRecv)); - CHK_RET(GetNRecv(nRecv)); - u32 recvStreamNum = std::min(nRecv, MAX_CONCURRENT); - // 流分配和Notify分配 - algResResponse.notifiesMain.push_back(groupRecvNotifies[GROUP_SYNC_NOTIFY_NUM * (iRecv % recvStreamNum)]); - algResResponse.notifiesAux.push_back(groupRecvNotifies[GROUP_SYNC_NOTIFY_NUM * (iRecv % recvStreamNum) + 1]); - // CCL Buffer分配 - DeviceMem cclout = cclBufferManager_.GetOutCCLbuffer(); - u64 sliceSize; - CHK_RET(GetSliceSize(sliceSize)); - algResResponse.cclOutputMem = cclout.range(iRecv % bufferSliceNum * sliceSize, sliceSize); - HCCL_INFO("Recv[%d] algResResponse.cclOutputMem[%p], size[%d]", iRecv, algResResponse.cclOutputMem.ptr(), algResResponse.cclOutputMem.size()); - } - } DeviceMem expMem = cclBufferManager_.GetCommCCLBuffer(); // 获取拓展内存 if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLV || opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC || opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALL) { DeviceMem tinySendRecvMem; @@ -7257,6 +7050,7 @@ namespace hccl opTilingData->isCapture = opParam.isCapture; opTilingData->orderLaunchMode = GetOrderLaunchMode(opParam.isCapture); opTilingData->isSymmetricMemory = opParam.supportSymmetricMemory; + opTilingData->needIncreLink = opParam.needIncreLink; // 有没有存在对应的Notify CHK_RET(InitAndCheckAicpuOrderNotify(opTilingData->orderLaunchMode)); CHK_RET(BuildHierarchicalAlgOption(opTilingData->ahcConfInfo)); @@ -7295,10 +7089,14 @@ namespace hccl CHK_PTR_NULL(opParam.BatchSendRecvDataDes.sendRecvItemsPtr + i); batchSendRecvDataPtr->batchSendRecvItem[i] = *(opParam.BatchSendRecvDataDes.sendRecvItemsPtr + i); } - u8 *isDirectRemoteRankPtr = reinterpret_cast(batchSendRecvDataPtr->batchSendRecvItem + opParam.BatchSendRecvDataDes.itemNum); - for (u32 i = 0; i < userRankSize_; i++) { - CHK_PTR_NULL(isDirectRemoteRankPtr + i); - isDirectRemoteRankPtr[i] = *(opParam.BatchSendRecvDataDes.isDirectRemoteRank + i); + if (deviceType_ == DevType::DEV_TYPE_910B && isGroupMode_) { + // 如果是A2的GroupSendRecv则跳过下面这段 + } else { + u8 *isDirectRemoteRankPtr = reinterpret_cast(batchSendRecvDataPtr->batchSendRecvItem + opParam.BatchSendRecvDataDes.itemNum); + for (u32 i = 0; i < userRankSize_; i++) { + CHK_PTR_NULL(isDirectRemoteRankPtr + i); + isDirectRemoteRankPtr[i] = *(opParam.BatchSendRecvDataDes.isDirectRemoteRank + i); + } } } else if (opType == HcclCMDType::HCCL_CMD_ALLTOALL) { CHK_RET(SetDynamicTilingDataAlltoall(opParam, dynamicDataMem));

hcomm

hcomm