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 "all_gather_v_pipeline.h"
12 : #include "alg_template_register.h"
13 :
14 : constexpr u32 STEP_OFFSET_TWO = 2;
15 :
16 : namespace hccl {
17 0 : AllGatherVPipeline::AllGatherVPipeline(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
18 :
19 0 : AllGatherVPipeline::~AllGatherVPipeline() {}
20 0 : HcclResult AllGatherVPipeline::Prepare(
21 : HcomCollOpInfo* opInfo, u32 userRank, u64& count, DeviceMem& cclBufferPartOne, DeviceMem& cclBufferPartTwo,
22 : SubCommInfo& level0CommInfo, SubCommInfo& level1CommInfo, Stream& mainStream, std::vector<Stream>& subStream,
23 : std::vector<std::shared_ptr<LocalNotify>>& notifyMain, std::vector<std::shared_ptr<LocalNotify>>& notifySub,
24 : std::vector<Slice>& userOutSlice)
25 : {
26 0 : opInfo_ = opInfo; // 这里面的count是总count
27 0 : memSliceCount_ = count; // 这里的count是单次loop的count
28 0 : userRank_ = userRank;
29 :
30 0 : usrInMemAddr_ = opInfo_->inputAddr;
31 0 : usrOutMemAddr_ = opInfo_->outputAddr;
32 :
33 : // needed resource
34 : // stream: 1 * mainStream + n * subStream
35 : // mem: usrInMem, usrOutMem, DMAMem
36 : // interNotify, streamNotify
37 :
38 : // stream
39 : // mainStream负责locMemCpy以及subStream同步控制
40 0 : stream_ = mainStream;
41 : // subStream负责:
42 : // streamId[0]: inter执行
43 : // streamId[1:intraRankSize]: intraRankSize-1个intra执行
44 0 : subStream_ = subStream;
45 :
46 : // DMAMem + interNotify from Link
47 0 : intraRankSize_ = level0CommInfo.localRankSize;
48 0 : interRankSize_ = level1CommInfo.localRankSize;
49 0 : intraRankId_ = level0CommInfo.localRank;
50 0 : interRankId_ = level1CommInfo.localRank;
51 0 : intraLinks_ = level0CommInfo.links;
52 0 : interLinks_ = level1CommInfo.links;
53 :
54 : // streamNotify, size: n
55 0 : streamNotifyMain_ = notifyMain;
56 0 : if (streamNotifyMain_.size() < intraRankSize_) {
57 0 : HCCL_ERROR(
58 : "[AllGatherVPipeline][Prepare]rank[%u] streamNotifyMain_ size[%u] error, is smaller than,"
59 : "intraRankSize_[%u]",
60 : userRank_, streamNotifyMain_.size(), intraRankSize_);
61 0 : return HCCL_E_INTERNAL;
62 : }
63 0 : streamNotifySub_ = notifySub;
64 0 : if (streamNotifySub_.size() < intraRankSize_) {
65 0 : HCCL_ERROR(
66 : "[AllGatherVPipeline][Prepare]rank[%u] streamNotifySub_ size[%u] error, is smaller than, "
67 : "intraRankSize_[%u]",
68 : userRank_, streamNotifySub_.size(), intraRankSize_);
69 0 : return HCCL_E_INTERNAL;
70 : }
71 :
72 : // 128byte align offset
73 0 : dmaMem_.push_back(cclBufferPartOne);
74 0 : dmaMem_.push_back(cclBufferPartTwo);
75 :
76 0 : HCCL_INFO(
77 : "[AllGatherVPipeline][Prepare]streamNum[%zu], streamNotifyMainNum[%zu], streamNotifySubNum[%zu]",
78 : subStream_.size(), streamNotifyMain_.size(), streamNotifySub_.size());
79 0 : HCCL_INFO(
80 : "[AllGatherVPipeline][Prepare]interLinksNum[%zu], intraLinksNum[%zu]", interLinks_.size(), intraLinks_.size());
81 0 : userMemSlice_ = std::move(userOutSlice);
82 :
83 0 : return HCCL_SUCCESS;
84 : }
85 :
86 0 : HcclResult AllGatherVPipeline::MainWaitSub()
87 : {
88 0 : u32 subStreamNum = intraRankSize_;
89 0 : for (u32 signalIndex = 0; signalIndex < subStreamNum; signalIndex++) {
90 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, streamNotifyMain_[signalIndex], INVALID_VALUE_STAGE));
91 : }
92 0 : return HCCL_SUCCESS;
93 : }
94 :
95 0 : HcclResult AllGatherVPipeline::SubRecordMain()
96 : {
97 0 : u32 subStreamNum = intraRankSize_;
98 0 : for (u32 streamIndex = 0; streamIndex < subStreamNum; streamIndex++) {
99 0 : CHK_RET(LocalNotify::Post(
100 : subStream_[streamIndex], dispatcher_, streamNotifyMain_[streamIndex], INVALID_VALUE_STAGE));
101 : }
102 0 : return HCCL_SUCCESS;
103 : }
104 :
105 0 : HcclResult AllGatherVPipeline::MainRecordSub()
106 : {
107 0 : u32 subStreamNum = intraRankSize_;
108 0 : for (u32 signalIndex = 0; signalIndex < subStreamNum; signalIndex++) {
109 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, streamNotifySub_[signalIndex], INVALID_VALUE_STAGE));
110 : }
111 0 : return HCCL_SUCCESS;
112 : }
113 :
114 0 : HcclResult AllGatherVPipeline::SubWaitMain()
115 : {
116 0 : u32 subStreamNum = intraRankSize_;
117 0 : for (u32 streamIndex = 0; streamIndex < subStreamNum; streamIndex++) {
118 0 : CHK_RET(LocalNotify::Wait(
119 : subStream_[streamIndex], dispatcher_, streamNotifySub_[streamIndex], INVALID_VALUE_STAGE));
120 : }
121 0 : return HCCL_SUCCESS;
122 : }
123 :
124 0 : HcclResult AllGatherVPipeline::RunAsync()
125 : {
126 0 : HCCL_INFO("[AllGatherVPipeline][RunAsync]AllGatherRingMesh starts groupRankId[%u] ", userRank_);
127 : // inter ring algo
128 0 : u32 prevInterRankId = (interRankId_ - 1 + interRankSize_) % interRankSize_;
129 0 : u32 nextInterRankId = (interRankId_ + 1) % interRankSize_;
130 0 : LINK prevInterLink = interLinks_[prevInterRankId];
131 0 : LINK nextInterLink = interLinks_[nextInterRankId];
132 :
133 0 : u32 unitSize = SIZE_TABLE[opInfo_->dataType];
134 :
135 : // 仅使用两块DMAMem，为了方便切换使用
136 0 : u32 dmaMemSliceId = 0;
137 0 : u32 dmaMemSliceNum = dmaMem_.size();
138 :
139 : // step 0前置操作 : 所有卡本地数据从userIn-->DMAIn
140 0 : DeviceMem locSrc = DeviceMem::create(usrInMemAddr_, memSliceCount_ * unitSize);
141 0 : u64 localOffset = userMemSlice_[userRank_].offset % HCCL_MIN_SLICE_ALIGN_910B;
142 : DeviceMem locDMAInMem
143 0 : = DeviceMem::create(static_cast<u8*>(dmaMem_[dmaMemSliceId].ptr()) + localOffset, memSliceCount_ * unitSize);
144 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, locDMAInMem, locSrc, stream_));
145 :
146 0 : for (u32 step = 0; step < interRankSize_; step++) {
147 : // 主从流同步
148 0 : CHK_RET(MainRecordSub());
149 0 : CHK_RET(SubWaitMain());
150 :
151 : // 数据搬运及后同步
152 0 : u32 srcDMAMemSliceId = dmaMemSliceId;
153 0 : dmaMemSliceId = (dmaMemSliceId + 1) % dmaMemSliceNum;
154 0 : u32 dstDMAMemSliceId = dmaMemSliceId;
155 :
156 0 : u64 serverRankOffset = intraRankId_ + (interRankId_ + interRankSize_ - step) % interRankSize_ * intraRankSize_;
157 0 : u64 serverOffsetByte = userMemSlice_[serverRankOffset].offset % HCCL_MIN_SLICE_ALIGN_910B;
158 0 : u64 readRemoteOffset
159 0 : = intraRankId_
160 0 : + (prevInterRankId + interRankSize_ - step) % interRankSize_ * intraRankSize_; // sever间前通信rank偏移
161 0 : u64 readRemoteOffsetByte = userMemSlice_[readRemoteOffset].offset % HCCL_MIN_SLICE_ALIGN_910B;
162 0 : if (step < interRankSize_ - 1) {
163 0 : CHK_RET(prevInterLink->TxAck(subStream_[0])); // AckRecord
164 0 : CHK_RET(nextInterLink->RxAck(subStream_[0])); // AckWait
165 : // RdmaSend + Record 或 PCIE::Record
166 0 : CHK_RET(nextInterLink->TxAsync(
167 : (dstDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM), serverOffsetByte,
168 : static_cast<u8*>(dmaMem_[srcDMAMemSliceId].ptr()) + serverOffsetByte,
169 : userMemSlice_[serverRankOffset].size, subStream_[0]));
170 0 : HCCL_DEBUG(
171 : "[AllGatherVPipeline][RunAsync] local rank[%u] localOffset[%llu]tx with remoteRank[%u],"
172 : "remoteOffset[%llu] with slice[%llu]",
173 : userRank_, serverOffsetByte, nextInterRankId, serverOffsetByte, userMemSlice_[serverRankOffset].size);
174 : // 对于RDM RxAsync，内存属性入参无效 RDMA::Wait
175 : // 对于PCIE，需设置内存属性 PCIE::Read + Record
176 0 : CHK_RET(prevInterLink->RxAsync(
177 : (srcDMAMemSliceId == 0 ? UserMemType::INPUT_MEM : UserMemType::OUTPUT_MEM), readRemoteOffsetByte,
178 : static_cast<u8*>(dmaMem_[dstDMAMemSliceId].ptr()) + readRemoteOffsetByte,
179 : userMemSlice_[readRemoteOffset].size, subStream_[0])); // wait
180 0 : HCCL_DEBUG(
181 : "[AllGatherVPipeline][RunAsync]read local rank[%u] localOffset[%llu]tx with remoteRank[%u],"
182 : "remoteOffset[%llu] with slice[%llu]",
183 : userRank_, readRemoteOffsetByte, readRemoteOffset, readRemoteOffsetByte,
184 : userMemSlice_[readRemoteOffset].size);
185 0 : CHK_RET(prevInterLink->PostFinAck(subStream_[0]));
186 0 : CHK_RET(nextInterLink->WaitFinAck(subStream_[0]));
187 : // inter的最后一步需要barrier确保数据发完
188 0 : if (step == interRankSize_ - STEP_OFFSET_TWO) {
189 0 : CHK_RET(ExecuteBarrier(prevInterLink, nextInterLink, subStream_[0]));
190 : }
191 : }
192 :
193 0 : for (u32 i = 1; i < intraRankSize_; i++) {
194 0 : u32 remIntraRankId = (intraRankId_ + i) % intraRankSize_;
195 0 : CHK_RET(intraLinks_[remIntraRankId]->TxAck(subStream_[i])); // ackrecord
196 0 : CHK_RET(intraLinks_[remIntraRankId]->RxAck(subStream_[i]));
197 0 : void* remDMAMemPtr = nullptr;
198 :
199 : // 从对端的cclbuffer读到本端的userout，因此偏移值是不一样的
200 0 : CHK_RET(intraLinks_[remIntraRankId]->GetRemoteMem(
201 : srcDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM, &remDMAMemPtr));
202 0 : u32 remUserRankId
203 0 : = (interRankId_ - step + interRankSize_) % interRankSize_ * intraRankSize_ + remIntraRankId;
204 0 : void* dstAddr = static_cast<u8*>(usrOutMemAddr_) + userMemSlice_[remUserRankId].offset;
205 :
206 0 : u64 remoteOffsetByte = userMemSlice_[remUserRankId].offset % HCCL_MIN_SLICE_ALIGN_910B;
207 : DeviceMem src = DeviceMem::create(
208 0 : static_cast<u8*>(remDMAMemPtr) + remoteOffsetByte, userMemSlice_[remUserRankId].size);
209 0 : DeviceMem dst = DeviceMem::create(dstAddr, userMemSlice_[remUserRankId].size);
210 0 : CHK_RET(HcclD2DMemcpyAsync(
211 : dispatcher_, dst, src, subStream_[i], intraLinks_[remIntraRankId]->GetRemoteRank(),
212 : intraLinks_[remIntraRankId]->GetLinkType()));
213 0 : CHK_RET(intraLinks_[remIntraRankId]->TxDataSignal(subStream_[i])); // data record
214 0 : CHK_RET(intraLinks_[remIntraRankId]->RxDataSignal(subStream_[i])); // data wait
215 0 : }
216 :
217 0 : CHK_RET(SubRecordMain());
218 0 : CHK_RET(MainWaitSub());
219 :
220 0 : void* dstAddr = static_cast<u8*>(usrOutMemAddr_) + userMemSlice_[serverRankOffset].offset;
221 0 : DeviceMem locDst = DeviceMem::create(dstAddr, userMemSlice_[serverRankOffset].size);
222 : DeviceMem srcMem = DeviceMem::create(
223 0 : static_cast<u8*>(dmaMem_[srcDMAMemSliceId].ptr()) + serverOffsetByte, userMemSlice_[serverRankOffset].size);
224 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, locDst, srcMem, stream_));
225 0 : }
226 :
227 0 : HCCL_INFO("[AllGatherVPipeline][RunAsync]AllGatherRingMesh finished groupRankId[%u] ", userRank_);
228 0 : return HCCL_SUCCESS;
229 0 : }
230 :
231 0 : HcclResult AllGatherVPipeline::GetNslbAdjInfo(
232 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
233 : {
234 0 : u32 ringNextRank = (rank + 1) % rankSize;
235 0 : LINK nslbNext = links[ringNextRank];
236 :
237 : // Pipeline 步长合并 等同于 ring
238 0 : NslbDpAdjInfo adjInfoStep = {};
239 0 : nslbAdjInfo.dstRankNum = 1;
240 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
241 0 : adjInfoStep.phaseId = 1;
242 0 : adjInfoStep.rev = 0;
243 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
244 :
245 0 : return HCCL_SUCCESS;
246 0 : }
247 :
248 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_V_PIPELINE, AllGatherVPipeline);
249 : } // namespace hccl
|
