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 "alltoallv_pairwise.h"
12 : #include "alg_template_register.h"
13 :
14 :
15 : namespace hccl {
16 0 : AlltoAllVPairWise::AlltoAllVPairWise(const HcclDispatcher dispatcher)
17 0 : : AlgTemplateBase(dispatcher)
18 : {
19 0 : }
20 :
21 0 : AlltoAllVPairWise::~AlltoAllVPairWise() {}
22 :
23 0 : HcclResult AlltoAllVPairWise::Prepare(AlltoAllVBufferInfo& sendBuffer, AlltoAllVBufferInfo& recvBuffer,
24 : bool isAlltoAllZCopyMode, const Stream &stream, HcclWorkflowMode workMode,
25 : std::map<u32, std::vector<u64>> &rankSendDisplsMap,
26 : std::map<u32, std::vector<u64>> &rankRecvDisplsMap)
27 : {
28 0 : DeviceMem scratchInputMem = DeviceMem();
29 0 : DeviceMem scratchOutputMem = DeviceMem();
30 0 : CHK_RET(AlltoAllVPairWise::Prepare(sendBuffer, recvBuffer, scratchInputMem, scratchOutputMem,
31 : isAlltoAllZCopyMode, stream, workMode,
32 : rankSendDisplsMap, rankRecvDisplsMap));
33 0 : return HCCL_SUCCESS;
34 0 : }
35 :
36 0 : HcclResult AlltoAllVPairWise::Prepare(AlltoAllVBufferInfo &sendBuffer, AlltoAllVBufferInfo &recvBuffer,
37 : DeviceMem &scratchInputMem, DeviceMem &scratchOutputMem, bool isAlltoAllZCopyMode, const Stream &stream,
38 : HcclWorkflowMode workMode, std::map<u32, std::vector<u64>> &rankSendDisplsMap,
39 : std::map<u32, std::vector<u64>> &rankRecvDisplsMap)
40 : {
41 0 : HCCL_INFO("[AlltoAllVPairWise][Prepare] Begin");
42 0 : scratchMemSize_ = 0;
43 0 : sendDataUnitBytes_ = 0;
44 0 : recvDataUnitBytes_ = 0;
45 0 : isAlltoAllZCopyMode_ = isAlltoAllZCopyMode;
46 0 : workMode_ = workMode;
47 0 : rankSendDisplsMapPtr_ = &rankSendDisplsMap;
48 0 : rankRecvDisplsMapPtr_ = &rankRecvDisplsMap;
49 :
50 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
51 0 : CHK_PRT_RET((!isAlltoAllZCopyMode_ && scratchInputMem.size() != scratchOutputMem.size()),
52 : HCCL_ERROR("[AlltoAllVPairWise][Prepare]scratchInputMem and scratchOutputMem should be the same size, "
53 : "ScratchInputMem[%llu] ScratchOutputMem[%llu]", scratchInputMem.size(), scratchOutputMem.size()),
54 : HCCL_E_MEMORY);
55 :
56 0 : CHK_PRT_RET(scratchInputMem.size() == 0 || scratchOutputMem.size() == 0,
57 : HCCL_ERROR("[AlltoAllVPairWise][Prepare] invilad scratchMemSize[%llu]", scratchInputMem.size()),
58 : HCCL_E_PARA);
59 0 : scratchInputMem_ = scratchInputMem;
60 0 : scratchOutputMem_ = scratchOutputMem;
61 0 : scratchMemSize_ = scratchInputMem.size();
62 : }
63 :
64 0 : sendBuffer_ = sendBuffer;
65 0 : recvBuffer_ = recvBuffer;
66 0 : stream_ = stream;
67 :
68 0 : CHK_RET(SalGetDataTypeSize(sendBuffer_.dataType, sendDataUnitBytes_));
69 0 : CHK_RET(SalGetDataTypeSize(recvBuffer_.dataType, recvDataUnitBytes_));
70 :
71 0 : return HCCL_SUCCESS;
72 : }
73 :
74 0 : HcclResult AlltoAllVPairWise::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
75 : {
76 0 : HCCL_INFO("[AlltoAllVPairWise][RunAsync]: rank[%u] transportSize[%llu]", rank, links.size());
77 0 : CHK_SMART_PTR_NULL(dispatcher_);
78 0 : CHK_PTR_NULL(stream_.ptr());
79 :
80 0 : CHK_PRT_RET(rankSize == 0, HCCL_ERROR("[AlltoAllVPairWise][Prepare] invilad rankSize[%u]", rankSize), HCCL_E_PARA);
81 :
82 0 : CHK_PRT_RET(rankSize != links.size(),
83 : HCCL_ERROR("[AlltoAllVPairWise][RunAsync]: rankSize[%u] and transport size[%llu] do not match", rankSize,
84 : links.size()),
85 : HCCL_E_PARA);
86 :
87 0 : CHK_RET(LocalCopy(rank));
88 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
89 0 : !isAlltoAllZCopyMode_) { // 单算子 && BCopy模式
90 0 : CHK_RET(RunBCopyAlltoAll(rank, rankSize, links));
91 0 : } else {
92 0 : CHK_RET(RunZCopyAlltoAll(rank, rankSize, links));
93 : }
94 0 : return HCCL_SUCCESS;
95 : }
96 :
97 : // 从本rank的sendbuffer拷贝到本rank的recvbuffer
98 0 : HcclResult AlltoAllVPairWise::LocalCopy(const u32 rank)
99 : {
100 0 : DeviceMem dstMem = recvBuffer_.mem.range(recvDataUnitBytes_ * recvBuffer_.displs[rank],
101 0 : recvBuffer_.counts[rank] * recvDataUnitBytes_);
102 0 : DeviceMem srcMem = sendBuffer_.mem.range(sendDataUnitBytes_ * sendBuffer_.displs[rank],
103 0 : sendBuffer_.counts[rank] * sendDataUnitBytes_);
104 0 : HCCL_DEBUG("[AlltoAllVPairWise][LocalCopy]: Rank[%u] destAddr[%p], destMax[%llu], srcAddr[%p], size[%llu]",
105 : rank, dstMem.ptr(), dstMem.size(), srcMem.ptr(), srcMem.size());
106 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream_));
107 :
108 0 : return HCCL_SUCCESS;
109 0 : }
110 :
111 0 : HcclResult AlltoAllVPairWise::RunBCopyAlltoAll(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
112 : {
113 0 : for (u32 i = 1; i < rankSize; i++) {
114 0 : u32 prevRank = (rank + rankSize - i) % rankSize;
115 0 : u32 nextRank = (rank + i) % rankSize;
116 0 : std::shared_ptr<Transport> prevTransport = links[prevRank];
117 0 : std::shared_ptr<Transport> nextTransport = links[nextRank];
118 :
119 0 : CHK_SMART_PTR_NULL(prevTransport);
120 0 : CHK_SMART_PTR_NULL(nextTransport);
121 :
122 0 : HCCL_DEBUG("[AlltoAllVPairWise][RunBCopyAlltoAll]: prevRank[%u] nextRank[%u], step[%u]", prevRank, nextRank, i);
123 :
124 0 : u64 sendBytes = sendBuffer_.counts[nextRank] * sendDataUnitBytes_;
125 0 : u64 recvBytes = recvBuffer_.counts[prevRank] * recvDataUnitBytes_;
126 :
127 0 : u64 sendDispBytes = sendBuffer_.displs[nextRank] * sendDataUnitBytes_;
128 0 : u64 recvDispBytes = recvBuffer_.displs[prevRank] * recvDataUnitBytes_;
129 :
130 : // scratchMemSize_ 的合法性已经在 Prepare 函数中校验
131 0 : u32 sendTimes = (sendBytes / scratchMemSize_) + ((sendBytes % scratchMemSize_) == 0 ? 0 : 1);
132 0 : u32 recvTimes = (recvBytes / scratchMemSize_) + ((recvBytes % scratchMemSize_) == 0 ? 0 : 1);
133 :
134 0 : HCCL_DEBUG("[AlltoAllVPairWise][RunBCopyAlltoAll]: rank[%u] "\
135 : "sendTimes[%u] recvTimes[%u] sendBytes[%llu] recvBytes[%llu] scratchMemSize_[%llu]",
136 : rank, sendTimes, recvTimes, sendBytes, recvBytes, scratchMemSize_);
137 :
138 0 : u32 curSendTime = 0;
139 0 : u32 curRecvTime = 0;
140 0 : while (sendTimes != 0 || recvTimes != 0) {
141 : u8 *sendAddr =
142 0 : reinterpret_cast<u8 *>(sendBuffer_.mem.ptr()) + sendDispBytes + curSendTime * scratchMemSize_;
143 : u8 *recvAddr =
144 0 : reinterpret_cast<u8 *>(recvBuffer_.mem.ptr()) + recvDispBytes + curRecvTime * scratchMemSize_;
145 0 : u64 curSendBytes = 0;
146 0 : u64 curRecvBytes = 0;
147 0 : CHK_RET(CalcSendRecvCounts(sendTimes, curSendTime, sendBytes, curSendBytes));
148 0 : CHK_RET(CalcSendRecvCounts(recvTimes, curRecvTime, recvBytes, curRecvBytes));
149 :
150 0 : HCCL_DEBUG("[AlltoAllVPairWise][RunBCopyAlltoAll]: "\
151 : "curSendTime[%llu] curRecvTime[%llu] curSendBytes[%llu] curRecvBytes[%llu]",
152 : curSendTime, curRecvTime, curSendBytes, curRecvBytes);
153 :
154 0 : HcclResult ret = SendRecv(curSendBytes, curRecvBytes, sendAddr, recvAddr, prevTransport, nextTransport);
155 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
156 : HCCL_ERROR("[AlltoAllVPairWise][RunBCopyAlltoAll]: errNo[0x%016llx] "\
157 : "curSendBytes[%llu] curRecvBytes[%llu] sendAddr[%p] recvAddr[%p]",
158 : HCCL_ERROR_CODE(ret), curSendBytes, curRecvBytes, sendAddr, recvAddr),
159 : ret);
160 :
161 0 : curSendTime = curSendBytes != 0 ? curSendTime + 1 : curSendTime;
162 0 : curRecvTime = curRecvBytes != 0 ? curRecvTime + 1 : curRecvTime;
163 0 : if (curSendTime == sendTimes && curRecvTime == recvTimes) {
164 0 : break;
165 : }
166 : }
167 0 : }
168 :
169 0 : return HCCL_SUCCESS;
170 : }
171 :
172 0 : HcclResult AlltoAllVPairWise::CalcSendRecvCounts(u32 times, u32 curTime, u64 totalBytes, u64 &curBytes) const
173 : {
174 0 : if (times == 0) { // 不需要发送
175 0 : curBytes = 0;
176 0 : } else if (times == 1 && curTime == times - 1) { // 只发一次
177 0 : curBytes = totalBytes;
178 0 : } else if (times > 1 && totalBytes % scratchMemSize_ == 0 && curTime < times) {
179 0 : curBytes = scratchMemSize_;
180 0 : } else if (times > 1 && totalBytes % scratchMemSize_ != 0 && curTime < times - 1) {
181 0 : curBytes = scratchMemSize_;
182 0 : } else if (times > 1 && totalBytes % scratchMemSize_ != 0 && curTime == times - 1) {
183 0 : curBytes = totalBytes % scratchMemSize_;
184 : } else {
185 0 : curBytes = 0;
186 : }
187 0 : return HCCL_SUCCESS;
188 : }
189 :
190 0 : HcclResult AlltoAllVPairWise::SendRecv(u64 curSendBytes, u64 curRecvBytes, u8 *sendAddr, u8 *recvAddr,
191 : std::shared_ptr<Transport> prevTransport, std::shared_ptr<Transport> nextTransport)
192 : {
193 0 : if (curRecvBytes > 0) {
194 0 : CHK_RET(prevTransport->TxAck(stream_)); // transport sync record
195 : }
196 0 : if (curSendBytes > 0) {
197 0 : CHK_RET(nextTransport->RxAck(stream_)); // transport sync wait
198 0 : DeviceMem srcMem1 = DeviceMem::create(sendAddr, curSendBytes);
199 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, scratchInputMem_, srcMem1, stream_));
200 : // send payload + notify
201 0 : CHK_RET(nextTransport->TxAsync(UserMemType::OUTPUT_MEM, 0, scratchInputMem_.ptr(), curSendBytes, stream_));
202 0 : }
203 0 : if (curRecvBytes > 0) {
204 0 : CHK_RET(prevTransport->RxAsync(UserMemType::INPUT_MEM, 0, scratchOutputMem_.ptr(), curRecvBytes, stream_));
205 0 : DeviceMem dstMem = DeviceMem::create(recvAddr, curRecvBytes);
206 0 : DeviceMem srcMem = scratchOutputMem_.range(0, curRecvBytes);
207 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream_));
208 0 : CHK_RET(prevTransport->TxAck(stream_)); // record
209 0 : }
210 0 : if (curSendBytes > 0) {
211 0 : CHK_RET(nextTransport->RxAck(stream_)); // wait
212 0 : CHK_RET(nextTransport->TxDataSignal(stream_)); // record
213 : }
214 0 : if (curRecvBytes > 0) {
215 0 : CHK_RET(prevTransport->RxDataSignal(stream_)); // wait
216 0 : CHK_RET(prevTransport->RxWaitDone(stream_));
217 : }
218 0 : if (curSendBytes > 0) {
219 0 : CHK_RET(nextTransport->TxWaitDone(stream_));
220 : }
221 0 : return HCCL_SUCCESS;
222 : }
223 :
224 0 : HcclResult AlltoAllVPairWise::SendRecv(TxMemoryInfo txMemoryInfo, RxMemoryInfo rxMemoryInfo,
225 : std::shared_ptr<Transport> prevTransport, std::shared_ptr<Transport> nextTransport)
226 : {
227 : // send payload + notify
228 0 : CHK_RET(nextTransport->TxAsync(txMemoryInfo.dstMemType, txMemoryInfo.dstOffset, txMemoryInfo.src,
229 : txMemoryInfo.len, stream_));
230 0 : CHK_RET(prevTransport->RxAsync(rxMemoryInfo.srcMemType, rxMemoryInfo.srcOffset, rxMemoryInfo.dst,
231 : rxMemoryInfo.len, stream_));
232 0 : CHK_RET(prevTransport->TxAck(stream_)); // record
233 0 : CHK_RET(nextTransport->RxAck(stream_)); // wait
234 0 : CHK_RET(nextTransport->TxDataSignal(stream_)); // record
235 0 : CHK_RET(prevTransport->RxDataSignal(stream_)); // wait
236 0 : CHK_RET(prevTransport->RxWaitDone(stream_));
237 0 : CHK_RET(nextTransport->TxWaitDone(stream_));
238 0 : return HCCL_SUCCESS;
239 : }
240 :
241 0 : HcclResult AlltoAllVPairWise::RunZCopyAlltoAll(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
242 : {
243 0 : for (u32 i = 1; i < rankSize; i++) {
244 0 : u32 prevRank = (rank + rankSize - i) % rankSize;
245 0 : u32 nextRank = (rank + i) % rankSize;
246 0 : std::shared_ptr<Transport> prevTransport = links[prevRank];
247 0 : std::shared_ptr<Transport> nextTransport = links[nextRank];
248 :
249 0 : CHK_SMART_PTR_NULL(prevTransport);
250 0 : CHK_SMART_PTR_NULL(nextTransport);
251 :
252 0 : HCCL_DEBUG("[AlltoAllVPairWise][RunZCopyAlltoAll]: prevRank[%u] nextRank[%u], step[%u]", prevRank, nextRank, i);
253 :
254 0 : CHK_RET(prevTransport->TxAck(stream_)); // transport sync record
255 0 : CHK_RET(nextTransport->RxAck(stream_)); // transport sync wait
256 :
257 0 : u64 sendBytes = sendBuffer_.counts[nextRank] * sendDataUnitBytes_;
258 0 : u64 recvBytes = recvBuffer_.counts[prevRank] * recvDataUnitBytes_;
259 0 : u64 sendDispBytes = sendBuffer_.displs[nextRank] * sendDataUnitBytes_;
260 0 : u64 recvDispBytes = recvBuffer_.displs[prevRank] * recvDataUnitBytes_;
261 0 : u8 *sendAddr = reinterpret_cast<u8 *>(sendBuffer_.mem.ptr()) + sendDispBytes;
262 0 : u8 *recvAddr = reinterpret_cast<u8 *>(recvBuffer_.mem.ptr()) + recvDispBytes;
263 :
264 0 : u64 dstOffset = rankRecvDisplsMapPtr_->at(nextRank)[rank];
265 0 : u64 srcOffset = rankSendDisplsMapPtr_->at(prevRank)[rank];
266 :
267 0 : TxMemoryInfo txMemoryInfo{UserMemType::OUTPUT_MEM, dstOffset, sendAddr, sendBytes};
268 0 : RxMemoryInfo rxMemoryInfo{UserMemType::INPUT_MEM, srcOffset, recvAddr, recvBytes};
269 :
270 0 : HCCL_DEBUG("[AlltoAllVPairWise][RunZCopyAlltoAll]: sendBytes[%llu] recvBytes[%llu] sendDispBytes[%llu]" \
271 : " dstOffset[%llu]", sendBytes, recvBytes, sendDispBytes, dstOffset);
272 0 : HcclResult ret = SendRecv(txMemoryInfo, rxMemoryInfo, prevTransport, nextTransport);
273 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
274 : HCCL_ERROR("[AlltoAllVPairWise][RunZCopyAlltoAll]errNo[0x%016llx] "\
275 : "sendBytes[%llu] recvBytes[%llu] sendAddr[%p] dstOffset[%llu]",
276 : HCCL_ERROR_CODE(ret), sendBytes, recvBytes, sendAddr, dstOffset),
277 : ret);
278 0 : }
279 :
280 0 : return HCCL_SUCCESS;
281 : }
282 0 : HcclResult AlltoAllVPairWise::GetNslbAdjInfo(const u32 rank, const u32 rankSize,
283 : const std::vector<LINK> &links, AdjInfo& nslbAdjInfo)
284 : {
285 : (void) links;
286 0 : for (u32 i = 1; i < rankSize; i++) {
287 0 : u32 nextRank = (rank + i) % rankSize;
288 0 : if (i < NSLBDP_PAIRWISE_MAXPHASE) {
289 0 : NslbDpAdjInfo adjInfoStep = {0};
290 0 : adjInfoStep.dstLocalRankId = nextRank;
291 0 : adjInfoStep.phaseId = i;
292 0 : adjInfoStep.rev = 0;
293 0 : HCCL_INFO("AlltoAllVPairWise-nslb: adjInfoStep.phaseId[%u], remoteuserRank[%u]", adjInfoStep.phaseId, nextRank);
294 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
295 : }
296 : }
297 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
298 :
299 0 : return HCCL_SUCCESS;
300 : }
301 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_2_ALL_V_PAIRWISE, AlltoAllVPairWise);
302 : } // namespace hccl
|
