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(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
15 :
16 0 : AlltoallPipelineBase::~AlltoallPipelineBase() {}
17 :
18 0 : HcclResult AlltoallPipelineBase::Prepare(
19 : u32 userRank, A2aPipelineMemory A2aPipelineMemory, const SubCommInfo& level0CommInfo,
20 : const SubCommInfo& level1CommInfo, Stream& mainStream, std::vector<Stream>& subStream,
21 : std::vector<std::shared_ptr<LocalNotify>>& notifyMain, std::vector<std::shared_ptr<LocalNotify>>& notifySub,
22 : std::vector<SendRecvInfo>& allMeshAggregationSendRecvInfo, HcclWorkflowMode workMode)
23 : {
24 0 : allMeshAggregationSendRecvInfo_ = &allMeshAggregationSendRecvInfo;
25 0 : workMode_ = workMode;
26 :
27 0 : localSendRecvInfo_ = (*allMeshAggregationSendRecvInfo_)[userRank];
28 :
29 0 : inputMem_ = A2aPipelineMemory.userInput;
30 0 : outputMem_ = A2aPipelineMemory.userOutput;
31 0 : scratchMem_ = A2aPipelineMemory.scratchMem;
32 0 : cclIn_ = A2aPipelineMemory.cclInBuffer;
33 0 : cclOut_ = A2aPipelineMemory.cclOutBuffer;
34 :
35 0 : intraRankSize_ = level0CommInfo.localRankSize;
36 0 : interRankSize_ = level1CommInfo.localRankSize;
37 0 : groupRankSize_ = intraRankSize_ * interRankSize_;
38 :
39 0 : userRank_ = userRank;
40 0 : intraRankId_ = level0CommInfo.localRank;
41 0 : interRankId_ = level1CommInfo.localRank;
42 :
43 0 : meshRankStart_ = userRank - intraRankId_;
44 0 : meshRankEnd_ = meshRankStart_ + intraRankSize_ - 1;
45 :
46 0 : mainStream_ = mainStream;
47 0 : subStream_ = subStream;
48 0 : streamNotifyMain_ = notifyMain;
49 0 : streamNotifySub_ = notifySub;
50 :
51 0 : intraLinks_ = level0CommInfo.links;
52 0 : interLinks_ = level1CommInfo.links;
53 :
54 0 : HCCL_DEBUG(
55 : "[AlltoallPipelineBase]streamNum[%u], streamNotifyMainNum[%u], streamNotifySubNum[%u]", subStream_.size(),
56 : streamNotifyMain_.size(), streamNotifySub_.size());
57 0 : HCCL_DEBUG("[AlltoallPipelineBase]interLinksNum[%u], intraLinksNum[%u]", interLinks_.size(), intraLinks_.size());
58 :
59 0 : return HCCL_SUCCESS;
60 : }
61 :
62 0 : HcclResult AlltoallPipelineBase::CheckResourceValid()
63 : {
64 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
65 0 : CHK_PRT_RET(
66 : cclIn_.size() != cclOut_.size(),
67 : HCCL_ERROR(
68 : "[AlltoallPipelineBase][CheckResourceValid] cclIn mem and cclOut mem should be the same size, "
69 : "ScratchInputMem[%llu] ScratchOutputMem[%llu]",
70 : cclIn_.size(), cclOut_.size()),
71 : HCCL_E_MEMORY);
72 : }
73 0 : CHK_PRT_RET(
74 : subStream_.size() < intraRankSize_ || streamNotifyMain_.size() < intraRankSize_
75 : || streamNotifySub_.size() < intraRankSize_,
76 : HCCL_DEBUG(
77 : "[AlltoallPipelineBase][CheckResourceValid] "
78 : "stream resource not enough, num sub stream[%llu], num notify main signal[%llu] num notify sub "
79 : "signal[%llu], "
80 : "should be more than or equal to intraRankSize %llu",
81 : subStream_.size(), streamNotifyMain_.size(), streamNotifySub_.size(), intraRankSize_),
82 : HCCL_E_UNAVAIL);
83 0 : return HCCL_SUCCESS;
84 : }
85 :
86 : // alltoall 系列算法抽象行为应该都可以分为第一次发送前的数据准备，中间的每一步同步发送，以及本地数据搬移
87 0 : HcclResult AlltoallPipelineBase::RunAsync()
88 : {
89 0 : CHK_RET(CheckResourceValid());
90 0 : CHK_RET(PreProcess());
91 0 : for (u32 step = 0, numStep = CalcInterNumSteps(); step < numStep; step++) {
92 0 : CHK_RET(PipelineSend(step, step == (numStep - 1)));
93 : }
94 0 : CHK_RET(PostProcess());
95 0 : return HCCL_SUCCESS;
96 : }
97 :
98 0 : std::string AlltoallPipelineBase::GetCurrClassName()
99 : {
100 0 : std::string className = typeid(*this).name();
101 0 : if (className.find("class") != className.npos) {
102 0 : size_t classNamePrefixLen = 6;
103 0 : className = className.substr(classNamePrefixLen);
104 : }
105 0 : return className;
106 0 : }
107 :
108 0 : std::string AlltoallPipelineBase::GetStreamIndexString()
109 : {
110 0 : std::string res = "";
111 0 : for (auto& info : intraStreamInfo_) {
112 0 : res += std::to_string(info.first) + ", ";
113 : }
114 0 : return res;
115 0 : }
116 :
117 0 : HcclResult AlltoallPipelineBase::NotifyInterStreamStart()
118 : {
119 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySub_[intraRankId_], INVALID_VALUE_STAGE));
120 0 : CHK_RET(
121 : LocalNotify::Wait(subStream_[intraRankId_], dispatcher_, streamNotifySub_[intraRankId_], INVALID_VALUE_STAGE));
122 0 : HCCL_DEBUG(
123 : "[%s][NotifyInterStreamStart] userRank %u, interRank %u, "
124 : "intraRank %u, main stream notify sdma stream %s",
125 : GetCurrClassName().c_str(), userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
126 0 : return HCCL_SUCCESS;
127 : }
128 :
129 0 : HcclResult AlltoallPipelineBase::WaitInterStreamFinish()
130 : {
131 0 : CHK_RET(
132 : LocalNotify::Post(subStream_[intraRankId_], dispatcher_, streamNotifyMain_[intraRankId_], INVALID_VALUE_STAGE));
133 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMain_[intraRankId_], INVALID_VALUE_STAGE));
134 0 : HCCL_DEBUG(
135 : "[%s][WaitInterStreamFinish] userRank %u, interRank %u, intraRank %u, "
136 : "main stream notify sdma stream %s",
137 : GetCurrClassName().c_str(), userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
138 0 : return HCCL_SUCCESS;
139 : }
140 :
141 : // 主流只需要通知当前子步骤需要收发数据的 SDMA 流，减少同步开销
142 0 : HcclResult AlltoallPipelineBase::NotifyIntraStreamStart()
143 : {
144 0 : for (auto& sdmaInfo : intraStreamInfo_) {
145 0 : u32 streamIndex = sdmaInfo.first;
146 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySub_[streamIndex], INVALID_VALUE_STAGE));
147 0 : CHK_RET(LocalNotify::Wait(
148 : subStream_[streamIndex], dispatcher_, streamNotifySub_[streamIndex], INVALID_VALUE_STAGE));
149 : }
150 0 : HCCL_DEBUG(
151 : "[%s][NotifyIntraStreamStart] userRank %u, interRank %u, "
152 : "intraRank %u, main stream notify sdma stream %s",
153 : GetCurrClassName().c_str(), userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 0 : HcclResult AlltoallPipelineBase::WaitIntraStreamFinish()
158 : {
159 0 : for (auto& sdmaInfo : intraStreamInfo_) {
160 0 : u32 streamIndex = sdmaInfo.first;
161 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMain_[streamIndex], INVALID_VALUE_STAGE));
162 0 : CHK_RET(LocalNotify::Post(
163 : subStream_[streamIndex], dispatcher_, streamNotifyMain_[streamIndex], INVALID_VALUE_STAGE));
164 : }
165 0 : HCCL_DEBUG(
166 : "[%s][WaitIntraStreamFinish] userRank %u, interRank %u, "
167 : "intraRank %u, main stream wait sdma stream %s",
168 : GetCurrClassName().c_str(), userRank_, interRankId_, intraRankId_, GetStreamIndexString().c_str());
169 0 : return HCCL_SUCCESS;
170 : }
171 :
172 0 : HcclResult AlltoallPipelineBase::GetNslbAdjInfo(
173 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
174 : {
175 0 : u32 numStep = rankSize - 1;
176 :
177 0 : for (u32 step = 0; step < numStep; step++) {
178 0 : u32 nextRank = (rank + 1 + step) % rankSize;
179 0 : LINK nslbNext = links[nextRank];
180 0 : CHK_SMART_PTR_NULL(nslbNext);
181 0 : NslbDpAdjInfo nextInfoStep = {};
182 0 : nextInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
183 0 : nextInfoStep.phaseId = step + 1;
184 0 : nextInfoStep.rev = 0;
185 0 : nslbAdjInfo.nsAdjInfo.push_back(nextInfoStep);
186 0 : }
187 :
188 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
189 0 : return HCCL_SUCCESS;
190 : }
191 : } // namespace hccl
|
