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 "coll_reduce_scatter_comm_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 6 : CollReduceScatterCommExecutor::CollReduceScatterCommExecutor(const HcclDispatcher dispatcher,
17 6 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 6 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
19 : {
20 6 : desc_.deterministic = 1;
21 6 : DMAReduceFlag_ = false;
22 6 : }
23 :
24 12 : void CollReduceScatterCommExecutor::ParseParam(const OpParam& param)
25 : {
26 12 : tag_ = param.tag;
27 :
28 12 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
29 0 : scratchMemFlag_ = false;
30 : } else {
31 12 : scratchMemFlag_ = true;
32 : }
33 :
34 : // 记录图模式总数据量
35 12 : totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
36 12 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
37 12 : }
38 :
39 6 : HcclResult CollReduceScatterCommExecutor::CalcScratchMemSize(u64& scratchMemSize)
40 : {
41 6 : if (scratchMemFlag_) {
42 6 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
43 0 : scratchMemSize = inCCLbufferSize_;
44 : } else {
45 6 : scratchMemSize = totalSize_;
46 : }
47 : } else {
48 0 : scratchMemSize = 0U;
49 : }
50 :
51 6 : HCCL_INFO("[CollReduceScatterCommExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]",
52 : tag_.c_str(), scratchMemSize);
53 6 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : bool CollReduceScatterCommExecutor::IsSmallData(const u64 totalSize, const u64 curSize)
57 : {
58 0 : return topoAttr_.deviceType == DevType::DEV_TYPE_910_93;
59 : }
60 :
61 6 : HcclResult CollReduceScatterCommExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
62 : {
63 6 : TransportMemType inputType = TransportMemType::RESERVED;
64 6 : TransportMemType outputType = TransportMemType::RESERVED;
65 6 : CHK_RET(CalcTransportMemType(inputType, outputType));
66 6 : CHK_RET(CalcCombinedCommInfo(inputType, outputType, opTransport));
67 6 : return HCCL_SUCCESS;
68 : }
69 :
70 6 : HcclResult CollReduceScatterCommExecutor::CalcTransportMemType(TransportMemType &inputType,
71 : TransportMemType &outputType)
72 : {
73 6 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
74 0 : inputType = TransportMemType::CCL_INPUT;
75 0 : if (scratchMemFlag_) {
76 0 : outputType = TransportMemType::SCRATCH;
77 : } else {
78 0 : outputType = TransportMemType::CCL_OUTPUT;
79 : }
80 : } else {
81 6 : inputType = TransportMemType::PARAM_INPUT;
82 6 : if (scratchMemFlag_) {
83 6 : outputType = TransportMemType::SCRATCH;
84 : } else {
85 0 : outputType = TransportMemType::PARAM_OUTPUT;
86 : }
87 : }
88 6 : HCCL_INFO("[CollReduceScatterCommExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
89 : tag_.c_str(), inputType, outputType);
90 6 : return HCCL_SUCCESS;
91 : }
92 :
93 6 : HcclResult CollReduceScatterCommExecutor::CalcCombinedCommInfo(TransportMemType inputType,
94 : TransportMemType outputType,
95 : std::vector<LevelNSubCommTransport>& opTransport)
96 : {
97 6 : CommPlane commPlane = COMM_COMBINE_ORDER;
98 :
99 6 : CommParaInfo commParaInfo(commPlane, CommType::COMM_TAG_MAX);
100 :
101 6 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
102 0 : commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
103 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
104 0 : commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1;
105 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
106 0 : commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
107 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
108 0 : commParaInfo.commType = CommType::COMM_TAG_HALVING_DOUBLING;
109 : } else {
110 6 : commParaInfo.commType = CommType::COMM_TAG_RING_INNER;
111 : }
112 6 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[commPlane], inputType, outputType));
113 :
114 6 : return HCCL_SUCCESS;
115 6 : }
116 :
117 0 : u64 CollReduceScatterCommExecutor::CalcLoopMaxCount(const u32 unitSize)
118 : {
119 : // 中转内存单次最多能够接受的output count
120 0 : u64 maxCountPerLoop = inCCLbufferSize_ / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN
121 0 : * HCCL_MIN_SLICE_ALIGN / unitSize;
122 0 : return maxCountPerLoop;
123 : }
124 :
125 0 : bool CollReduceScatterCommExecutor::IsHugeData(const u64 curSize, OpParam *param)
126 : {
127 0 : bool hugeData = (curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE) ||
128 : (curSize > SDMA_SEND_MAX_SIZE);
129 0 : return hugeData;
130 : }
131 :
132 6 : HcclResult CollReduceScatterCommExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
133 : {
134 6 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] userRank[%u] starts.", __func__, topoAttr_.userRank);
135 6 : CommPlane commPlane = COMM_COMBINE_ORDER;
136 :
137 6 : CHK_RET(CheckCommSize(commPlane, COMM_INDEX_0 + 1));
138 6 : SubCommInfo combinedCommInfo = GetSubCommInfo(commPlane, COMM_INDEX_0);
139 :
140 6 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
141 :
142 : // 构造ring algorithm对应的reduce-scatter实例
143 6 : std::unique_ptr<AlgTemplateBase> tempAlg;
144 6 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
145 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
146 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
147 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_COMBINE_ORDER", __func__);
148 0 : CHK_SMART_PTR_NULL(tempAlg);
149 0 : CHK_RET(tempAlg->Prepare(reduceAttr, false));
150 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count,
151 : param.DataDes.dataType, param.stream, param.reduceType));
152 0 : if (topoAttr_.deviceType != DevType::DEV_TYPE_910_93 || algoAttr_.isSupportAtomicWrite) {
153 0 : tempAlg->CloseBarrier();
154 : }
155 0 : CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
156 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
157 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
158 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR_V1, dispatcher_);
159 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR_V1 in COMM_COMBINE_ORDER", __func__);
160 0 : CHK_SMART_PTR_NULL(tempAlg);
161 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
162 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count,
163 : param.DataDes.dataType, param.stream, param.reduceType));
164 0 : CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
165 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
166 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
167 0 : TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
168 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_COMBINE_ORDER", __func__);
169 0 : CHK_SMART_PTR_NULL(tempAlg);
170 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
171 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count,
172 : param.DataDes.dataType, param.stream, param.reduceType));
173 0 : CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
174 6 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
175 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
176 0 : TemplateType::TEMPLATE_REDUCESCATTER_RECURSIVE_HD, dispatcher_);
177 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RECURSIVE_HD in COMM_COMBINE_ORDER", __func__);
178 0 : CHK_SMART_PTR_NULL(tempAlg);
179 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
180 0 : DeviceMem scratchMem = execMem.scratchMem.range(0, execMem.inputMem.size());
181 0 : u64 inputDataCount = execMem.inputMem.size() / SIZE_TABLE[param.DataDes.dataType];
182 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.inputMem, scratchMem, inputDataCount,
183 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0)));
184 0 : CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
185 0 : u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
186 0 : DeviceMem srcMem = execMem.inputMem.range(dataSize * topoAttr_.userRank, dataSize);
187 0 : DeviceMem dstMem = execMem.outputMem.range(0, dataSize);
188 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
189 0 : } else {
190 12 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
191 6 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
192 6 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_COMBINE_ORDER", __func__);
193 6 : CHK_SMART_PTR_NULL(tempAlg);
194 6 : CHK_RET(tempAlg->Prepare(reduceAttr));
195 30 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
196 : param.DataDes.dataType, param.stream, param.reduceType));
197 6 : CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
198 : // 将cclInBuffer中与userRank_对应的部分拷贝至cclOutBuffer
199 6 : u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
200 6 : DeviceMem srcMem = execMem.inputMem.range(dataSize * topoAttr_.userRank, dataSize);
201 6 : DeviceMem dstMem = execMem.outputMem.range(0, dataSize);
202 6 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
203 6 : }
204 6 : return HCCL_SUCCESS;
205 6 : }
206 :
207 : REGISTER_EXEC("ReduceScatterComm", ReduceScatterComm, CollReduceScatterCommExecutor);
208 : }
|
