Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "allltoall_pipeline_base.h"
12 :
13 : namespace hccl {
14 0 : AlltoallPipelineBase::AlltoallPipelineBase(
15 0 : const HcclDispatcher dispatcher): AlgTemplateBase(dispatcher)
16 0 : {}
17 :
18 0 : AlltoallPipelineBase::~AlltoallPipelineBase() {}
19 :
20 0 : HcclResult AlltoallPipelineBase::Prepare(u32 userRank, A2aPipelineMemory A2aPipelineMemory,
21 : const SubCommInfo &level0CommInfo, const SubCommInfo &level1CommInfo,
22 : Stream &mainStream, std::vector<Stream> &subStream,
23 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub,
24 : std::vector<SendRecvInfo> &allMeshAggregationSendRecvInfo, HcclWorkflowMode workMode)
25 : {
26 0 : allMeshAggregationSendRecvInfo_ = &allMeshAggregationSendRecvInfo;
27 0 : workMode_ = workMode;
28 :
29 0 : localSendRecvInfo_ = (*allMeshAggregationSendRecvInfo_)[userRank];
30 :
31 0 : inputMem_ = A2aPipelineMemory.userInput;
32 0 : outputMem_ = A2aPipelineMemory.userOutput;
33 0 : scratchMem_ = A2aPipelineMemory.scratchMem;
34 0 : cclIn_ = A2aPipelineMemory.cclInBuffer;
35 0 : cclOut_ = A2aPipelineMemory.cclOutBuffer;
36 :
37 0 : intraRankSize_ = level0CommInfo.localRankSize;
38 0 : interRankSize_ = level1CommInfo.localRankSize;
39 0 : groupRankSize_ = intraRankSize_ * interRankSize_;
40 :
41 0 : userRank_ = userRank;
42 0 : intraRankId_ = level0CommInfo.localRank;
43 0 : interRankId_ = level1CommInfo.localRank;
44 :
45 0 : meshRankStart_ = userRank - intraRankId_;
46 0 : meshRankEnd_ = meshRankStart_ + intraRankSize_ - 1;
47 :
48 0 : mainStream_ = mainStream;
49 0 : subStream_ = subStream;
50 0 : streamNotifyMain_ = notifyMain;
51 0 : streamNotifySub_ = notifySub;
52 :
53 0 : intraLinks_ = level0CommInfo.links;
54 0 : interLinks_ = level1CommInfo.links;
55 :
56 0 : HCCL_DEBUG("[AlltoallPipelineBase]streamNum[%u], streamNotifyMainNum[%u], streamNotifySubNum[%u]",
57 : subStream_.size(), streamNotifyMain_.size(), streamNotifySub_.size());
58 0 : HCCL_DEBUG("[AlltoallPipelineBase]interLinksNum[%u], intraLinksNum[%u]", interLinks_.size(), intraLinks_.size());
59 :
60 0 : return HCCL_SUCCESS;
61 : }
62 :
63 0 : HcclResult AlltoallPipelineBase::CheckResourceValid()
64 : {
65 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
66 0 : CHK_PRT_RET(cclIn_.size() != cclOut_.size(),
67 : HCCL_ERROR("[AlltoallPipelineBase][CheckResourceValid] cclIn mem and cclOut mem should be the same size, "
68 : "ScratchInputMem[%llu] ScratchOutputMem[%llu]", cclIn_.size(), cclOut_.size()),
69 : HCCL_E_MEMORY);
70 : }
71 0 : CHK_PRT_RET(subStream_.size() < intraRankSize_ || streamNotifyMain_.size() < intraRankSize_ ||
72 : streamNotifySub_.size() < intraRankSize_, HCCL_DEBUG("[AlltoallPipelineBase][CheckResourceValid] "
73 : "stream resource not enough, num sub stream[%llu], num notify main signal[%llu] num notify sub signal[%llu], "
74 : "should be more than or equal to intraRankSize %llu", subStream_.size(), streamNotifyMain_.size(),
75 : streamNotifySub_.size(), intraRankSize_), HCCL_E_UNAVAIL);
76 0 : return HCCL_SUCCESS;
77 : }
78 :
79 : // alltoall 系列算法抽象行为应该都可以分为第一次发送前的数据准备，中间的每一步同步发送，以及本地数据搬移
80 0 : HcclResult AlltoallPipelineBase::RunAsync()
81 : {
82 0 : CHK_RET(CheckResourceValid());
83 0 : CHK_RET(PreProcess());
84 0 : for (u32 step = 0, numStep = CalcInterNumSteps(); step < numStep; step++) {
85 0 : CHK_RET(PipelineSend(step, step == (numStep - 1)));
86 : }
87 0 : CHK_RET(PostProcess());
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 0 : std::string AlltoallPipelineBase::GetCurrClassName()
92 : {
93 0 : std::string className = typeid(*this).name();
94 0 : if (className.find("class") != className.npos) {
95 0 : size_t classNamePrefixLen = 6;
96 0 : className = className.substr(classNamePrefixLen);
97 : }
98 0 : return className;
99 0 : }
100 :
101 0 : std::string AlltoallPipelineBase::GetStreamIndexString()
102 : {
103 0 : std::string res = "";
104 0 : for (auto& info : intraStreamInfo_) {
105 0 : res += std::to_string(info.first) + ", ";
106 : }
107 0 : return res;
108 0 : }
109 :
110 0 : HcclResult AlltoallPipelineBase::NotifyInterStreamStart()
111 : {
112 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySub_[intraRankId_],
113 : INVALID_VALUE_STAGE));
114 0 : CHK_RET(LocalNotify::Wait(subStream_[intraRankId_], dispatcher_, streamNotifySub_[intraRankId_],
115 : INVALID_VALUE_STAGE));
116 0 : HCCL_DEBUG("[%s][NotifyInterStreamStart] userRank %u, interRank %u, "
117 : "intraRank %u, main stream notify sdma stream %s", GetCurrClassName().c_str(),
118 : userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
119 0 : return HCCL_SUCCESS;
120 : }
121 :
122 0 : HcclResult AlltoallPipelineBase::WaitInterStreamFinish()
123 : {
124 0 : CHK_RET(LocalNotify::Post(subStream_[intraRankId_], dispatcher_, streamNotifyMain_[intraRankId_],
125 : INVALID_VALUE_STAGE));
126 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMain_[intraRankId_],
127 : INVALID_VALUE_STAGE));
128 0 : HCCL_DEBUG("[%s][WaitInterStreamFinish] userRank %u, interRank %u, intraRank %u, "
129 : "main stream notify sdma stream %s", GetCurrClassName().c_str(), userRank_, interRankId_,
130 : intraRankId_, GetStreamIndexString().c_str());
131 0 : return HCCL_SUCCESS;
132 : }
133 :
134 : // 主流只需要通知当前子步骤需要收发数据的 SDMA 流，减少同步开销
135 0 : HcclResult AlltoallPipelineBase::NotifyIntraStreamStart()
136 : {
137 0 : for (auto& sdmaInfo : intraStreamInfo_) {
138 0 : u32 streamIndex = sdmaInfo.first;
139 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySub_[streamIndex],
140 : INVALID_VALUE_STAGE));
141 0 : CHK_RET(LocalNotify::Wait(subStream_[streamIndex], dispatcher_, streamNotifySub_[streamIndex],
142 : INVALID_VALUE_STAGE));
143 : }
144 0 : HCCL_DEBUG("[%s][NotifyIntraStreamStart] userRank %u, interRank %u, "
145 : "intraRank %u, main stream notify sdma stream %s", GetCurrClassName().c_str(),
146 : userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
147 0 : return HCCL_SUCCESS;
148 : }
149 :
150 0 : HcclResult AlltoallPipelineBase::WaitIntraStreamFinish()
151 : {
152 0 : for (auto& sdmaInfo : intraStreamInfo_) {
153 0 : u32 streamIndex = sdmaInfo.first;
154 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMain_[streamIndex],
155 : INVALID_VALUE_STAGE));
156 0 : CHK_RET(LocalNotify::Post(subStream_[streamIndex], dispatcher_, streamNotifyMain_[streamIndex],
157 : INVALID_VALUE_STAGE));
158 : }
159 0 : HCCL_DEBUG("[%s][WaitIntraStreamFinish] userRank %u, interRank %u, "
160 : "intraRank %u, main stream wait sdma stream %s", GetCurrClassName().c_str(), userRank_,
161 : interRankId_, intraRankId_, GetStreamIndexString().c_str());
162 0 : return HCCL_SUCCESS;
163 : }
164 :
165 0 : HcclResult AlltoallPipelineBase::GetNslbAdjInfo(const u32 rank, const u32 rankSize,
166 : const std::vector<LINK> &links, AdjInfo& nslbAdjInfo)
167 : {
168 0 : u32 numStep = rankSize - 1;
169 :
170 0 : for (u32 step = 0; step < numStep; step++) {
171 0 : u32 nextRank = (rank + 1 + step) % rankSize;
172 0 : LINK nslbNext = links[nextRank];
173 0 : CHK_SMART_PTR_NULL(nslbNext);
174 0 : NslbDpAdjInfo nextInfoStep = {0};
175 0 : nextInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
176 0 : nextInfoStep.phaseId = step + 1;
177 0 : nextInfoStep.rev = 0;
178 0 : nslbAdjInfo.nsAdjInfo.push_back(nextInfoStep);
179 0 : }
180 :
181 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
182 0 : return HCCL_SUCCESS;
183 : }
184 : } // namespace hccl
|
