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