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