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_deter_executor.h"
12 :
13 : namespace hccl {
14 :
15 5 : CollReduceScatterDeterExecutor::CollReduceScatterDeterExecutor(
16 5 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 5 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
18 : {
19 8 : desc_.deterministic = 1;
20 8 : DMAReduceFlag_ = true;
21 8 : CCLMemSlice_ = false;
22 8 : }
23 :
24 5 : void CollReduceScatterDeterExecutor::ParseParam(const OpParam& param)
25 : {
26 5 : tag_ = param.tag;
27 :
28 : // 是否需要scratch memory 选中确定性计算Executor，其他条件必定满足，只需区分是否为图模式
29 8 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
30 :
31 : // 记录图模式总数据量
32 8 : totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
33 8 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
34 8 : }
35 :
36 6 : HcclResult CollReduceScatterDeterExecutor::CalcScratchMemSize(u64& scratchMemSize)
37 : {
38 6 : if (scratchMemFlag_) { // 确定性计算只有图模式需要scratch memory
39 4 : scratchMemSize = totalSize_;
40 : } else {
41 2 : scratchMemSize = 0U;
42 : }
43 6 : HCCL_INFO(
44 : "[CollReduceScatterDeterExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]", tag_.c_str(),
45 : scratchMemSize);
46 8 : return HCCL_SUCCESS;
47 : }
48 :
49 8 : HcclResult CollReduceScatterDeterExecutor::CalcStreamNum(u32& streamNum)
50 : {
51 8 : u32 totalStreamNum = 0U;
52 8 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
53 4 : totalStreamNum = topoAttr_.deviceNumPerAggregation - 1U;
54 : } else {
55 4 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
56 : }
57 8 : streamNum = totalStreamNum - 1U;
58 8 : const u32 subStreamNum = 3;
59 8 : if (topoAttr_.serverNum != 1) {
60 0 : streamNum = subStreamNum;
61 : }
62 8 : HCCL_INFO("[CollReduceScatterDeterExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
63 7 : return HCCL_SUCCESS;
64 : }
65 :
66 8 : HcclResult CollReduceScatterDeterExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
67 : {
68 8 : TransportMemType inputType = TransportMemType::RESERVED;
69 8 : TransportMemType outputType = TransportMemType::RESERVED;
70 8 : CHK_RET(CalcTransportMemType(inputType, outputType));
71 8 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
72 5 : return HCCL_SUCCESS;
73 : }
74 :
75 : HcclResult
76 8 : CollReduceScatterDeterExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
77 : {
78 8 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
79 4 : inputType = TransportMemType::CCL_INPUT;
80 4 : if (scratchMemFlag_) {
81 0 : outputType = TransportMemType::SCRATCH;
82 : } else {
83 4 : outputType = TransportMemType::CCL_OUTPUT;
84 : }
85 : } else {
86 4 : inputType = TransportMemType::PARAM_INPUT;
87 4 : if (scratchMemFlag_) {
88 4 : outputType = TransportMemType::SCRATCH;
89 : } else {
90 0 : outputType = TransportMemType::PARAM_OUTPUT;
91 : }
92 : }
93 8 : HCCL_INFO(
94 : "[CollReduceScatterDeterExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(),
95 : inputType, outputType);
96 8 : return HCCL_SUCCESS;
97 : }
98 :
99 8 : HcclResult CollReduceScatterDeterExecutor::CalcLevel0CommInfo(
100 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
101 : {
102 8 : if (topoAttr_.serverNum == 1) {
103 8 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
104 6 : commParaLevel0.meshSinglePlane = true;
105 6 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
106 5 : } else {
107 0 : CommParaInfo commCombinePara(COMM_COMBINE_ORDER, CommType::COMM_TAG_MESH);
108 0 : CHK_RET(CalcCommPlaneInfo(tag_, commCombinePara, opTransport[COMM_COMBINE_ORDER], inputType, outputType));
109 0 : }
110 5 : return HCCL_SUCCESS;
111 : }
112 :
113 0 : u64 CollReduceScatterDeterExecutor::CalcLoopMaxCount(const u32 unitSize)
114 : {
115 : u64 maxCountPerLoop;
116 0 : bool isLocalReduce91073 = ((((topoAttr_.userRankSize & (topoAttr_.userRankSize - 1)) != 0) || aicpuUnfoldMode_
117 0 : || (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB))
118 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910_93))
119 0 : && (topoAttr_.serverNum == 1);
120 :
121 0 : bool isLocalReduce910B = ((totalSize_ > HCCL_SMALL_COUNT_32_KB)
122 0 : || (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB)
123 0 : || ((topoAttr_.deviceNumPerAggregation != DEVICE_EIGHT)
124 0 : && (topoAttr_.deviceNumPerAggregation != DEVICE_FOUR)))
125 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910B);
126 0 : if (isLocalReduce91073 || isLocalReduce910B) {
127 0 : maxCountPerLoop = (inCCLbufferSize_ - HCCL_MIN_SLICE_ALIGN_910B * topoAttr_.deviceNumPerAggregation) / unitSize
128 0 : / (topoAttr_.deviceNumPerAggregation - 1);
129 0 : maxCountPerLoop = maxCountPerLoop / HCCL_MIN_SLICE_ALIGN_910B;
130 0 : maxCountPerLoop = maxCountPerLoop * HCCL_MIN_SLICE_ALIGN_910B;
131 : } else {
132 0 : const u32 base = 2;
133 0 : maxCountPerLoop = inCCLbufferSize_ * base / topoAttr_.userRankSize;
134 : }
135 0 : return maxCountPerLoop;
136 : }
137 :
138 0 : bool CollReduceScatterDeterExecutor::IsHugeData(const u64 curSize, [[maybe_unused]] OpParam* param)
139 : {
140 : // 只有server内通信，多QP哈希散列下不刷新子图
141 0 : bool hugeData = curSize > SDMA_SEND_MAX_SIZE;
142 0 : return hugeData;
143 : }
144 :
145 0 : bool CollReduceScatterDeterExecutor::IsSmallData(const u64 totalSize, [[maybe_unused]] const u64 curSize)
146 : {
147 0 : bool smallData = false;
148 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
149 0 : smallData = true;
150 : } else {
151 0 : smallData = totalSize <= HCCL_SMALL_COUNT_32_KB;
152 : }
153 0 : return smallData;
154 : }
155 :
156 0 : HcclResult CollReduceScatterDeterExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
157 : {
158 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] userRank[%u] starts.", __func__, topoAttr_.userRank);
159 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
160 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
161 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg;
162 0 : CommPlane commPlane = COMM_LEVEL0;
163 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93 && topoAttr_.serverNum != 1) {
164 0 : commPlane = COMM_COMBINE_ORDER;
165 : }
166 :
167 0 : CHK_RET(CheckCommSize(commPlane, COMM_INDEX_0 + 1));
168 0 : SubCommInfo level0CommInfo = GetSubCommInfo(commPlane, COMM_INDEX_0);
169 :
170 0 : CHK_RET(ActiveSlaveStreams(param.stream));
171 :
172 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
173 0 : HcomCollOpInfo opInfo = {"",
174 0 : execMem.inputPtr,
175 0 : execMem.outputPtr,
176 0 : param.DataDes.count,
177 0 : param.DataDes.dataType,
178 0 : param.root,
179 0 : param.reduceType,
180 0 : 0};
181 :
182 0 : bool isLocalReduce91073 = ((((topoAttr_.userRankSize & (topoAttr_.userRankSize - 1)) != 0) || aicpuUnfoldMode_
183 0 : || (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB))
184 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910_93))
185 0 : && (topoAttr_.serverNum == 1);
186 :
187 0 : bool isLocalReduce910B = ((param.DataDes.count * unitSize > HCCL_SMALL_COUNT_32_KB)
188 0 : || (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB)
189 0 : || ((topoAttr_.deviceNumPerAggregation != DEVICE_EIGHT)
190 0 : && (topoAttr_.deviceNumPerAggregation != DEVICE_FOUR)))
191 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910B);
192 :
193 0 : if (isLocalReduce91073 || isLocalReduce910B) {
194 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
195 0 : TemplateType::TEMPLATE_REDUCESCATTER_LOCAL_REDUCE, dispatcher_);
196 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_LOCAL_REDUCE in COMM_COMBINE_ORDER", __func__);
197 0 : } else {
198 : level0TempAlg
199 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_HDSTAGE, dispatcher_);
200 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_HDSTAGE in COMM_COMBINE_ORDER", __func__);
201 : }
202 :
203 0 : CHK_SMART_PTR_NULL(level0TempAlg);
204 0 : CHK_RET(level0TempAlg->Prepare(
205 : execMem.inputMem, execMem.scratchMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
206 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0, reduceAttr, algResResp_->slaveStreams,
207 : algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, &opInfo));
208 :
209 0 : CHK_RET(level0TempAlg->RegisterProfiler(
210 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
211 : HCCL_EXEC_STEP_NOT_SET, param.stream));
212 :
213 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
214 0 : HCCL_INFO("ReduceScatter mesh deter run success");
215 0 : return HCCL_SUCCESS;
216 0 : }
217 :
218 : REGISTER_EXEC("ReduceScatterDeterExecutor", ReduceScatterDeter, CollReduceScatterDeterExecutor);
219 : } // namespace hccl
|
