@@ -29,15 +29,25 @@ InsTempAlltoAllMesh::~InsTempAlltoAllMesh() HcclResult InsTempAlltoAllMesh::CalcRes(AlgTempResReq &tempResReq) { - tempResReq.queNum = tempVTopo_[0].size(); + CHK_RET(CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq)); + + auto& linkReq = tempResReq.links; + for (auto resReqIter = linkReq.begin(); resReqIter != linkReq.end(); resReqIter++) { + auto remoteRank = resReqIter->first; + if (rank2PathNumMap_.find(remoteRank) == rank2PathNumMap_.end() || rank2PathNumMap_.at(remoteRank) == 0) { + HCCL_ERROR("No path to remoteRank[%u]", remoteRank); + return HcclResult::HCCL_E_INTERNAL; + } + resReqIter->second = rank2PathNumMap_.at(remoteRank); + maxPathNum = std::max(maxPathNum, rank2PathNumMap_.at(remoteRank)); // 每个rank的远端搬运最多需要pathNum条流 + } + tempResReq.queNum = 1 + maxPathNum * ALLTOALLV_DIRECT_FULLMESH_CONCURRENT_SIZE; tempResReq.streamNum = tempResReq.queNum; tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum); QId centerQ = 0; tempResReq.localWaitGroupCntNotify.emplace_back(centerQ, 0); tempResReq.localBcastPostCntNotify.emplace_back(centerQ, 0); - - CHK_RET(CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq)); HCCL_DEBUG("[InsTempAlltoAllMesh] Rank[%d], VtopoSize[%zu], requiredQue Num [%u].", myRank_, tempVTopo_[0].size(), tempResReq.queNum); return HcclResult::HCCL_SUCCESS; } @@ -226,16 +236,33 @@ HcclResult InsTempAlltoAllMesh::CalcCommRankSetforOneLoop(u32 roundIdx, const u3 HcclResult InsTempAlltoAllMesh::CopySendDataToScratch(u32 step, const std::vector &commRanks, std::unordered_map &sendSliceInfo, + const ResLinks &tempLinks, std::vector &queues) const { + HCCL_INFO("[InsTempAlltoAllMesh] CopySendDataToScratch"); + u32 queueId = 0; for(u32 i = 0 ; i < commRanks.size(); i++) { u32 remoteRank = commRanks[i]; + HCCL_INFO("remoteRank = %u", remoteRank); + u32 linkNum = rank2PathNumMap_.at(remoteRank); + std::vectorlinks = tempLinks.at(remoteRank); + std::vector dataSplitRate(linkNum); + CHK_RET(CalcDataSplitRateForLinks(links, dataSplitRate)); UsrData &currSendSliceInfo = sendSliceInfo[remoteRank]; - if (step < currSendSliceInfo.usrInSlices.size()) { - DataSlice &userInSlice = currSendSliceInfo.usrInSlices[step]; - DataSlice &sendScratchSlice = currSendSliceInfo.scratchInSlices[step]; - InsQuePtr queue = queues[i]; - CHK_RET(LocalCopy(queue, userInSlice, sendScratchSlice)); + for(u32 j = 0; j < linkNum; j++) + { + InsQuePtr queue = queues[queueId]; + queueId++; + HCCL_INFO("queueId=%u",queueId); + if (step < currSendSliceInfo.usrInSlices.size()) { + DataSlice &userInSliceAllLinks = currSendSliceInfo.usrInSlices[step]; + DataSlice &sendScratchSliceAllLinks = currSendSliceInfo.scratchInSlices[step]; + DataSlice userInSlice = CalcDataSliceForLinks(userInSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice sendScratchSlice = CalcDataSliceForLinks(sendScratchSliceAllLinks, dataSplitRate, j, dataType_); + + InsQuePtr queue = queues[i]; + CHK_RET(LocalCopy(queue, userInSlice, sendScratchSlice)); + } } } return HcclResult::HCCL_SUCCESS; @@ -250,57 +277,74 @@ HcclResult InsTempAlltoAllMesh::SendRecvData(u32 step, const std::vector &c CHK_PRT_RET(queues.empty(), HCCL_ERROR("[InsTempAlltoAllMesh][SendRecvData] empty queues"), HcclResult::HCCL_E_INTERNAL); CHK_PTR_NULL(queues[0]); - if (commRanks.size() > queues.size()) { - HCCL_ERROR("[InsTempAlltoAllMesh][SendRecvData] commRanks.size() > queues.size() is wrong"); + + if (commRanks.size() * maxPathNum > queues.size()) { + HCCL_ERROR("[InsTempAlltoAllMesh][SendRecvData] commRanks.size() * maxPathNum > queues.size() is wrong"); return HcclResult::HCCL_E_INTERNAL; } + uint64_t queuesId=0; for(u32 i = 0 ; i < commRanks.size(); i++) { s32 remoteRank = static_cast(commRanks[i]); - InsQuePtr queue = queues[i]; + u32 linkNum = rank2PathNumMap_.at(remoteRank); UsrData &currSendSliceInfo = sendSliceInfo[remoteRank]; UsrData &currReadSliceInfo = readSliceInfo[remoteRank]; - LinkData link = tempLinks.at(remoteRank)[0]; - + std::vectorlinks = tempLinks.at(remoteRank); + std::vector dataSplitRate(linkNum); + CHK_RET(CalcDataSplitRateForLinks(links, dataSplitRate)); std::vector &currSendSrcSlices = (dmaMode_ == DmaMode::GET) ? currSendSliceInfo.scratchInSlices : currSendSliceInfo.usrInSlices; std::vector &currSendDstSlices = (dmaMode_ == DmaMode::GET) ? currSendSliceInfo.usrOutSlices : currSendSliceInfo.scratchOutSlices; std::vector &currReadSrcSlices = (dmaMode_ == DmaMode::GET) ? currReadSliceInfo.scratchInSlices: currReadSliceInfo.usrInSlices; std::vector &currReadDstSlices = (dmaMode_ == DmaMode::GET) ? currReadSliceInfo.usrOutSlices : currReadSliceInfo.scratchOutSlices; + for(u32 j = 0; j < linkNum; j++) + { + InsQuePtr queue = queues[queuesId]; + queuesId++; - if (step < currSendSrcSlices.size() && step < currReadSrcSlices.size()) { - DataSlice &sendSrcSlice = currSendSrcSlices[step]; - DataSlice &sendDstSlice = currSendDstSlices[step]; - std::vector sendSrcSliceVec = {sendSrcSlice}; - std::vector sendDstSliceVec = {sendDstSlice}; - SlicesList sendDataSlice(sendSrcSliceVec, sendDstSliceVec); - DataSlice &recvSrcSlice = currReadSrcSlices[step]; - DataSlice &recvDstSlice = currReadDstSlices[step]; - std::vector recvSrcSliceVec = {recvSrcSlice}; - std::vector recvDstSliceVec = {recvDstSlice}; - SlicesList recvDataSlice(recvSrcSliceVec, recvDstSliceVec); - TxRxSlicesList sendRecvSlice(sendDataSlice, recvDataSlice); + LinkData link = tempLinks.at(remoteRank)[j]; + if (step < currSendSrcSlices.size() && step < currReadSrcSlices.size()) { + DataSlice &sendSrcSliceAllLinks = currSendSrcSlices[step]; + DataSlice &sendDstSliceAllLinks = currSendDstSlices[step]; + DataSlice sendSrcSlice = CalcDataSliceForLinks(sendSrcSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice sendDstSlice = CalcDataSliceForLinks(sendDstSliceAllLinks, dataSplitRate, j, dataType_); + std::vector sendSrcSliceVec = {sendSrcSlice}; + std::vector sendDstSliceVec = {sendDstSlice}; + SlicesList sendDataSlice(sendSrcSliceVec, sendDstSliceVec); + DataSlice &recvSrcSliceAllLinks = currReadSrcSlices[step]; + DataSlice &recvDstSliceAllLinks = currReadDstSlices[step]; + DataSlice recvSrcSlice = CalcDataSliceForLinks(recvSrcSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice recvDstSlice = CalcDataSliceForLinks(recvDstSliceAllLinks, dataSplitRate, j, dataType_); + std::vector recvSrcSliceVec = {recvSrcSlice}; + std::vector recvDstSliceVec = {recvDstSlice}; + SlicesList recvDataSlice(recvSrcSliceVec, recvDstSliceVec); + TxRxSlicesList sendRecvSlice(sendDataSlice, recvDataSlice); - TxRxLinks sendRecvLinks(link, link); - SendRecvInfo sendRecvInfo(sendRecvLinks, sendRecvSlice); - CHK_RET(SendRecv(sendRecvInfo, queue, 0, true, dmaMode_)); - HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run send and recv.", step, i, remoteRank); - } else if (step < currSendSrcSlices.size()) { - DataSlice &sendSrcSlice = currSendSrcSlices[step]; - DataSlice &sendDstSlice = currSendDstSlices[step]; - std::vector sendSrcSliceVec = {sendSrcSlice}; - std::vector sendDstSliceVec = {sendDstSlice}; - SlicesList sendDataSlice(sendSrcSliceVec, sendDstSliceVec); - DataInfo sendDataInfo(link, sendDataSlice); - CHK_RET(Send(sendDataInfo, queue)); - HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run send.", step, i, remoteRank); - } else if (step < currReadSrcSlices.size()) { - DataSlice &recvSrcSlice = currReadSrcSlices[step]; - DataSlice &recvDstSlice = currReadDstSlices[step]; - std::vector recvSrcSliceVec = {recvSrcSlice}; - std::vector recvDstSliceVec = {recvDstSlice}; - SlicesList recvDataSlice(recvSrcSliceVec, recvDstSliceVec); - DataInfo recvDataInfo(link, recvDataSlice); - CHK_RET(Recv(recvDataInfo, queue)); - HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run recv.", step, i, remoteRank); + TxRxLinks sendRecvLinks(link, link); + SendRecvInfo sendRecvInfo(sendRecvLinks, sendRecvSlice); + CHK_RET(SendRecv(sendRecvInfo, queue, 0, true, dmaMode_)); + HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run send and recv.", step, i, remoteRank); + } else if (step < currSendSrcSlices.size()) { + DataSlice &sendSrcSliceAllLinks = currSendSrcSlices[step]; + DataSlice &sendDstSliceAllLinks = currSendDstSlices[step]; + DataSlice sendSrcSlice = CalcDataSliceForLinks(sendSrcSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice sendDstSlice = CalcDataSliceForLinks(sendDstSliceAllLinks, dataSplitRate, j, dataType_); + std::vector sendSrcSliceVec = {sendSrcSlice}; + std::vector sendDstSliceVec = {sendDstSlice}; + SlicesList sendDataSlice(sendSrcSliceVec, sendDstSliceVec); + DataInfo sendDataInfo(link, sendDataSlice); + CHK_RET(Send(sendDataInfo, queue)); + HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run send.", step, i, remoteRank); + } else if (step < currReadSrcSlices.size()) { + DataSlice &recvSrcSliceAllLinks = currReadSrcSlices[step]; + DataSlice &recvDstSliceAllLinks = currReadDstSlices[step]; + DataSlice recvSrcSlice = CalcDataSliceForLinks(recvSrcSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice recvDstSlice = CalcDataSliceForLinks(recvDstSliceAllLinks, dataSplitRate, j, dataType_); + std::vector recvSrcSliceVec = {recvSrcSlice}; + std::vector recvDstSliceVec = {recvDstSlice}; + SlicesList recvDataSlice(recvSrcSliceVec, recvDstSliceVec); + DataInfo recvDataInfo(link, recvDataSlice); + CHK_RET(Recv(recvDataInfo, queue)); + HCCL_DEBUG("[InsTempAlltoAllMesh][SendRecvData] step[%u], commRank[%u], remoteRank[%d] run recv.", step, i, remoteRank); + } } } return HcclResult::HCCL_SUCCESS; @@ -308,16 +352,31 @@ HcclResult InsTempAlltoAllMesh::SendRecvData(u32 step, const std::vector &c HcclResult InsTempAlltoAllMesh::CopyRecvDataFromScratch(u32 step, const std::vector &commRanks, std::unordered_map &readSliceInfo, + const ResLinks &tempLinks, std::vector &queues) const { + HCCL_INFO("[InsTempAlltoAllMesh] CopyRecvDataFromScratch"); + u32 queueId = 0; for(u32 i = 0 ; i < commRanks.size(); i++) { u32 remoteRank = commRanks[i]; + HCCL_INFO("remoteRank = %u", remoteRank); + u32 linkNum = rank2PathNumMap_.at(remoteRank); + std::vectorlinks = tempLinks.at(remoteRank); + std::vector dataSplitRate(linkNum); + CHK_RET(CalcDataSplitRateForLinks(links, dataSplitRate)); UsrData &currReadSliceInfo = readSliceInfo[remoteRank]; - if (step < currReadSliceInfo.scratchOutSlices.size()) { - DataSlice &recvScratchSlice = currReadSliceInfo.scratchOutSlices[step]; - DataSlice &userOutSlice = currReadSliceInfo.usrOutSlices[step]; - InsQuePtr queue = queues[i]; - CHK_RET(LocalCopy(queue, recvScratchSlice, userOutSlice)); + for(u32 j = 0; j < linkNum; j++) + { + InsQuePtr queue = queues[queueId]; + queueId++; + HCCL_INFO("queueId=%u",queueId); + if (step < currReadSliceInfo.scratchOutSlices.size()) { + DataSlice &recvScratchSliceAllLinks = currReadSliceInfo.scratchOutSlices[step]; + DataSlice &userOutSliceAllLinks = currReadSliceInfo.usrOutSlices[step]; + DataSlice recvScratchSlice = CalcDataSliceForLinks(recvScratchSliceAllLinks, dataSplitRate, j, dataType_); + DataSlice userOutSlice = CalcDataSliceForLinks(userOutSliceAllLinks, dataSplitRate, j, dataType_); + CHK_RET(LocalCopy(queue, recvScratchSlice, userOutSlice)); + } } } return HcclResult::HCCL_SUCCESS; @@ -335,9 +394,9 @@ HcclResult InsTempAlltoAllMesh::RunSendRecvBufferLoop(u32 step, const std::vecto if (dmaMode_ == DmaMode::PUT) { CHK_RET(SendRecvData(step, commRanks, sendSliceInfoMap, recvSliceInfoMap, tempLinks, queues)); - CHK_RET(CopyRecvDataFromScratch(step, commRanks, recvSliceInfoMap, queues)); + CHK_RET(CopyRecvDataFromScratch(step, commRanks, recvSliceInfoMap, tempLinks, queues)); } else if (dmaMode_ == DmaMode::GET) { - CHK_RET(CopySendDataToScratch(step, commRanks, sendSliceInfoMap, queues)); + CHK_RET(CopySendDataToScratch(step, commRanks, sendSliceInfoMap, tempLinks, queues)); CHK_RET(SendRecvData(step, commRanks, sendSliceInfoMap, recvSliceInfoMap, tempLinks, queues)); }

hcomm

hcomm