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