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 "reduce_scatter_hccs_sio.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 : using namespace std;
16 :
17 0 : ReduceScatterHccsSio::ReduceScatterHccsSio(const HcclDispatcher dispatcher)
18 0 : : AlgTemplateBase(dispatcher)
19 0 : {}
20 :
21 0 : ReduceScatterHccsSio::~ReduceScatterHccsSio() {}
22 :
23 0 : HcclResult ReduceScatterHccsSio::Prepare(DeviceMem &inputMem, DeviceMem &outputMem, DeviceMem &scratchMem,
24 : const u64 count, const HcclDataType dataType, const Stream &stream,
25 : const HcclReduceOp reductionOp, const u32 root,
26 : const u64 baseOffset,
27 : const u64 reduceAttrBitMap, std::vector<Stream> &meshStreams,
28 : std::vector<std::shared_ptr<LocalNotify>> &meshSignal,
29 : std::vector<std::shared_ptr<LocalNotify>> &meshSignalAux,
30 : u32 userRank, SubCommInfo subCommInfoHccs, SubCommInfo subCommInfoSio, HcomCollOpInfo *opInfo)
31 : {
32 0 : reduceAttr_ = reduceAttrBitMap;
33 0 : userRank_ = userRank;
34 0 : meshStreams_ = meshStreams;
35 0 : meshSignalPtr_ = &meshSignal;
36 0 : meshSignalAuxPtr_ = &meshSignalAux;
37 0 : subCommInfoHccs_ = subCommInfoHccs;
38 0 : subCommInfoSio_ = subCommInfoSio;
39 0 : std::vector<Slice> slices;
40 0 : opInfo_ = opInfo;
41 0 : return AlgTemplateBase::Prepare(inputMem, outputMem, scratchMem, count, dataType, stream, reductionOp,
42 0 : root, slices, baseOffset);
43 0 : }
44 :
45 0 : HcclResult ReduceScatterHccsSio::MainRecordSub()
46 : {
47 0 : for (u32 signalIndex = 0; signalIndex < meshSignalAuxPtr_->size(); signalIndex++) {
48 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, (*meshSignalAuxPtr_)[signalIndex],
49 : profilerInput_.stage));
50 : }
51 0 : return HCCL_SUCCESS;
52 : }
53 :
54 0 : HcclResult ReduceScatterHccsSio::SubWaitMain()
55 : {
56 0 : for (u32 streamIndex = 0; streamIndex < meshSignalAuxPtr_->size(); streamIndex++) {
57 0 : CHK_RET(LocalNotify::Wait(meshStreams_[streamIndex], dispatcher_,
58 : (*meshSignalAuxPtr_)[streamIndex], profilerInput_.stage));
59 : }
60 0 : return HCCL_SUCCESS;
61 : }
62 :
63 0 : HcclResult ReduceScatterHccsSio::MainWaitSub()
64 : {
65 0 : for (u32 signalIndex = 0; signalIndex < meshSignalPtr_->size(); signalIndex++) {
66 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, (*meshSignalPtr_)[signalIndex], profilerInput_.stage));
67 : }
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 0 : HcclResult ReduceScatterHccsSio::SubRecordMain()
72 : {
73 0 : for (u32 streamIndex = 0; streamIndex < meshSignalPtr_->size(); streamIndex++) {
74 0 : CHK_RET(LocalNotify::Post(meshStreams_[streamIndex], dispatcher_, (*meshSignalPtr_)[streamIndex],
75 : profilerInput_.stage));
76 : }
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : HcclResult ReduceScatterHccsSio::RunInterDieOpBase(
81 : const u32 rank, const LINK &link, const u32 srcDMAMemSliceId, u32 unitSize)
82 : {
83 : (void) rank;
84 0 : void *remMemPtr = nullptr;
85 : // 获取远端的commoutMem
86 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
87 : static u32 HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR = 2;
88 0 : DeviceMem src;
89 0 : DeviceMem dst;
90 : uint64_t countpeace;
91 0 : if (srcDMAMemSliceId == 0) {
92 0 : src = DeviceMem::create(static_cast<u8 *>(remMemPtr), count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize);
93 0 : dst = DeviceMem::create(
94 0 : static_cast<u8 *>(opInfo_->outputAddr), count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize);
95 0 : countpeace = count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR;
96 : } else {
97 0 : src = DeviceMem::create(static_cast<u8 *>(remMemPtr) + count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize,
98 0 : (count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR) * unitSize);
99 0 : dst = DeviceMem::create(
100 0 : static_cast<u8 *>(opInfo_->outputAddr) + count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize,
101 0 : (count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR) * unitSize);
102 0 : countpeace = count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR;
103 : }
104 : // 通过对应链路写给对端的usrout
105 0 : CHK_RET(HcclReduceAsync(dispatcher_,
106 : static_cast<void *>(src.ptr()),
107 : countpeace,
108 : dataType_,
109 : reductionOp_,
110 : meshStreams_[srcDMAMemSliceId],
111 : static_cast<void *>(dst.ptr()),
112 : link->GetRemoteRank(),
113 : link->GetLinkType(),
114 : INLINE_REDUCE_BIT));
115 :
116 0 : return HCCL_SUCCESS;
117 0 : }
118 :
119 0 : HcclResult ReduceScatterHccsSio::RunInterDieOffload(
120 : const u32 rank, const LINK &link, const u32 srcDMAMemSliceId, u32 unitSize)
121 : {
122 0 : void *remMemPtr = nullptr;
123 : // 获取远端的commoutMem
124 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
125 : static u32 HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR = 2;
126 0 : DeviceMem src;
127 0 : DeviceMem dst;
128 0 : uint64_t countpeace = 0;
129 0 : if (srcDMAMemSliceId == 0) {
130 0 : src = DeviceMem::create(static_cast<u8 *>(remMemPtr) + rank * count_ * unitSize,
131 0 : count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize);
132 0 : dst = DeviceMem::create(
133 0 : static_cast<u8 *>(outputMem_.ptr()), count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize);
134 0 : countpeace = count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR;
135 : } else {
136 0 : src = DeviceMem::create(static_cast<u8 *>(remMemPtr) + rank * count_ * unitSize +
137 0 : count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize,
138 0 : (count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR) * unitSize);
139 0 : dst = DeviceMem::create(
140 0 : static_cast<u8 *>(outputMem_.ptr()) + count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR * unitSize,
141 0 : (count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR) * unitSize);
142 0 : countpeace = count_ - count_ / HCCL_REDUCE_SCATTER_SIO_SPLIT_FACTOR;
143 : }
144 : // 通过对应链路写给对端的usrout
145 0 : CHK_RET(HcclReduceAsync(dispatcher_,
146 : static_cast<void *>(src.ptr()),
147 : countpeace,
148 : dataType_,
149 : reductionOp_,
150 : meshStreams_[srcDMAMemSliceId],
151 : static_cast<void *>(dst.ptr()),
152 : link->GetRemoteRank(),
153 : link->GetLinkType(),
154 : INLINE_REDUCE_BIT));
155 :
156 0 : return HCCL_SUCCESS;
157 0 : }
158 :
159 0 : HcclResult ReduceScatterHccsSio::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
160 : {
161 0 : HCCL_INFO("ReduceScatterHccsSio run: rank[%u] totalrank[%u] inputMem[%p] outputMem[%p] count[%llu]", rank,
162 : rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
163 :
164 : // 数据准备
165 0 : u32 unitSize = SIZE_TABLE[dataType_];
166 0 : u64 totalCnt = opInfo_->count;
167 0 : u32 srcDMAMemSliceId = 0;
168 :
169 0 : DeviceMem src;
170 0 : DeviceMem dst;
171 :
172 0 : u32 dstRank = rank + (rank % 2 == 0 ? (1) : (-1));
173 0 : src = DeviceMem::create(static_cast<u8 *>(inputMem_.ptr()) + rank * count_ * unitSize, count_ * unitSize);
174 0 : dst = DeviceMem::create(static_cast<u8 *>(outputMem_.ptr()), count_ * unitSize);
175 :
176 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
177 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
178 : } else {
179 : //usrin到usrout
180 0 : src = DeviceMem::create(static_cast<u8 *>(opInfo_->inputAddr) + rank * totalCnt * unitSize, count_ * unitSize);
181 0 : dst = DeviceMem::create(static_cast<u8 *>(opInfo_->outputAddr), count_ * unitSize);
182 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
183 : //usrin到cclbuffer
184 0 : src = DeviceMem::create(static_cast<u8 *>(opInfo_->inputAddr) + dstRank * totalCnt * unitSize, count_ * unitSize);
185 0 : dst = DeviceMem::create(static_cast<u8 *>(inputMem_.ptr()), count_ * unitSize);
186 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
187 : }
188 :
189 0 : CHK_RET(MainRecordSub());
190 0 : CHK_RET(SubWaitMain());
191 :
192 : // 每个stream只负责一个对端的交互
193 :
194 0 : const LINK &dstLinkHccs = subCommInfoHccs_.links[dstRank];
195 0 : const LINK &dstLinkSio = subCommInfoSio_.links[dstRank];
196 0 : CHK_RET(dstLinkHccs->TxAck(meshStreams_[srcDMAMemSliceId]));
197 0 : CHK_RET(dstLinkHccs->RxAck(meshStreams_[srcDMAMemSliceId]));
198 0 : CHK_RET(dstLinkSio->TxAck(meshStreams_[srcDMAMemSliceId + 1]));
199 0 : CHK_RET(dstLinkSio->RxAck(meshStreams_[srcDMAMemSliceId + 1]));
200 :
201 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
202 : // 两条链路inline执行 reduce
203 0 : CHK_RET(RunInterDieOffload(rank, dstLinkHccs, srcDMAMemSliceId, unitSize));
204 :
205 0 : CHK_RET(RunInterDieOffload(rank, dstLinkSio, srcDMAMemSliceId + 1, unitSize));
206 : } else { // 单算子模式
207 : // usrout 读取对端的cclin
208 :
209 : // inline执行notice reduce
210 0 : CHK_RET(RunInterDieOpBase(rank, dstLinkHccs, srcDMAMemSliceId, unitSize));
211 :
212 0 : CHK_RET(RunInterDieOpBase(rank, dstLinkSio, srcDMAMemSliceId + 1, unitSize));
213 : }
214 :
215 0 : CHK_RET(dstLinkHccs->TxDataSignal(meshStreams_[0]));
216 0 : CHK_RET(dstLinkHccs->RxDataSignal(meshStreams_[0]));
217 0 : CHK_RET(dstLinkSio->TxDataSignal(meshStreams_[1]));
218 0 : CHK_RET(dstLinkSio->RxDataSignal(meshStreams_[1]));
219 :
220 0 : CHK_RET(SubRecordMain());
221 0 : CHK_RET(MainWaitSub());
222 :
223 0 : HCCL_INFO("ReduceScatterHccsSio finished: rank[%u]", rank);
224 0 : return HCCL_SUCCESS;
225 0 : }
226 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCESCATTER_HCCS_SIO, ReduceScatterHccsSio);
227 : }
|
