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_all_reduce_deter_pipeline_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceDeterPipelineExecutor::CollAllReduceDeterPipelineExecutor(
16 : const HcclDispatcher dispatcher,
17 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
19 : {
20 0 : DMAReduceFlag_ = true;
21 0 : }
22 :
23 0 : void CollAllReduceDeterPipelineExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 0 : }
27 :
28 0 : HcclResult CollAllReduceDeterPipelineExecutor::CalcStreamNum(u32& streamNum)
29 : {
30 0 : streamNum = topoAttr_.deviceNumPerAggregation + 3U; // (deviceNum - 1)机内 + 4Reduce + 1机间 - 1主流
31 0 : HCCL_INFO("[CollAllReduceDeterPipelineExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
32 : tag_.c_str(), streamNum);
33 0 : return HCCL_SUCCESS;
34 : }
35 :
36 0 : HcclResult CollAllReduceDeterPipelineExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
37 : {
38 0 : TransportMemType inputType = TransportMemType::RESERVED;
39 0 : TransportMemType outputType = TransportMemType::RESERVED;
40 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
41 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
42 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 0 : HcclResult CollAllReduceDeterPipelineExecutor::CalcLevel0CommInfo(TransportMemType inputType,
47 : TransportMemType outputType,
48 : std::vector<LevelNSubCommTransport>& opTransport)
49 : {
50 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_MESH);
51 0 : commParaInfo.meshSinglePlane = true;
52 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
53 0 : return HCCL_SUCCESS;
54 0 : }
55 :
56 0 : HcclResult CollAllReduceDeterPipelineExecutor::CalcLevel1CommInfo(TransportMemType inputType,
57 : TransportMemType outputType,
58 : std::vector<LevelNSubCommTransport>& opTransport)
59 : {
60 0 : CommParaInfo commParaInfo(COMM_LEVEL1, CommType::COMM_TAG_MESH);
61 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL1], inputType, outputType));
62 0 : return HCCL_SUCCESS;
63 0 : }
64 :
65 0 : HcclResult CollAllReduceDeterPipelineExecutor::CalcTransportMemType(TransportMemType &inputType,
66 : TransportMemType &outputType)
67 : {
68 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
69 0 : inputType = TransportMemType::CCL_INPUT;
70 0 : outputType = TransportMemType::CCL_OUTPUT;
71 : } else {
72 0 : inputType = TransportMemType::PARAM_INPUT;
73 0 : outputType = TransportMemType::PARAM_OUTPUT;
74 : }
75 0 : HCCL_INFO("[CollAllReduceDeterPipelineExecutor][CalcTransportMemType] tag[%s] inputType[%d], "
76 : "outputType[%d]", tag_.c_str(), inputType, outputType);
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : u64 CollAllReduceDeterPipelineExecutor::CalcCountPerSlice(const u64 &totalCount, const u32 &unitSize)
81 : {
82 0 : u64 sizePerBlock = (totalCount + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize * unitSize;
83 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN_910B);
84 0 : if (sizePerBlock * (topoAttr_.userRankSize - 1) < totalCount * unitSize) {
85 0 : return sizePerBlock;
86 : }
87 0 : sizePerBlock = (totalCount + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize * unitSize;
88 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN);
89 0 : return sizePerBlock;
90 : }
91 :
92 0 : HcclResult CollAllReduceDeterPipelineExecutor::RunLoopInner(OpParam ¶m, const ReduceType &reduceType, ExecMem &execMem)
93 : {
94 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
95 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
96 0 : HCCL_DEBUG("[CollAllReduceDeterPipelineExecutor][RunLoopInner]inputMem[%p][%llu], outputMem[%p][%llu], " \
97 : "intputPtr[%p], outputPtr[%p], curCount[%llu], curSize[%llu]",
98 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
99 : execMem.inputPtr, execMem.outputPtr, execMem.count, curSize);
100 0 : CHK_PRT_RET((execMem.count == 0),
101 : HCCL_ERROR("[CollAllReduceDeterPipelineExecutor][RunLoop]In OP_BASE curCount is zero."), HCCL_E_PARA);
102 :
103 : /* 设置子图复用标志 */
104 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
105 0 : bool hugeData = IsHugeData(curSize); // override
106 0 : bool smallData = IsSmallData(param.DataDes.count * unitSize, curSize); // override
107 0 : constexpr s64 HCCL_MEDIUM_COUNT_2_MB = 2 * 1024 * 1024;
108 0 : u64 sliceNum = (curSize / topoAttr_.userRankSize) < HCCL_MEDIUM_COUNT_2_MB ? 1 : 0 ;
109 0 : bool dataSplit = false;
110 0 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
111 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(autoSelectedAlgTypeLevel1,
112 : param.DataDes.dataType, reduceType, smallData, 1, hugeData, CopyPattern::ZCOPY, sliceNum,
113 : false, true, dataSplit, deterministic);
114 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey(), false));
115 :
116 0 : execMem.inputMem = DeviceMem::create(execMem.inputMem.ptr(), curSize);
117 0 : execMem.outputMem = DeviceMem::create(execMem.outputMem.ptr(), curSize);
118 :
119 : // 执行
120 0 : HcclResult ret = KernelRun(param, execMem);
121 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
122 : HCCL_ERROR("[CollAllReduceDeterPipelineExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], " \
123 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d], reduce op type[%d]",
124 : HCCL_ERROR_CODE(ret), param.tag.c_str(), execMem.inputMem.ptr(), execMem.outputMem.ptr(),
125 : execMem.count, param.DataDes.dataType, param.reduceType), ret);
126 :
127 0 : CHK_RET(LaunchTaskExtend(dispatcher_, const_cast<Stream &>(param.stream),
128 : const_cast<std::vector<Stream> &>(algResResp_->slaveStreams)));
129 0 : return ret;
130 : }
131 :
132 0 : HcclResult CollAllReduceDeterPipelineExecutor::PrepareDataSlice(const ExecMem &execMem, const u32 &unitSize,
133 : std::vector<Slice> &bufferSlices)
134 : {
135 0 : bufferSlices.resize(topoAttr_.userRankSize);
136 0 : u64 totalSize = execMem.count * unitSize;
137 0 : u64 sliceSize = CalcCountPerSlice(execMem.count, unitSize);
138 0 : for (u32 sliceIndex = 0; sliceIndex < topoAttr_.userRankSize; sliceIndex++) {
139 0 : bufferSlices[sliceIndex].size = totalSize > sliceSize ? sliceSize : totalSize;
140 0 : bufferSlices[sliceIndex].offset = sliceIndex * sliceSize;
141 0 : totalSize -= bufferSlices[sliceIndex].size;
142 0 : HCCL_DEBUG("[CollAllReduceDeterPipelineExecutor][PrepareDataSlice]tag[%s], buffer slice i[%u], "
143 : "size[%llu], offset[%llu], left size[%llu]",
144 : tag_.c_str(), bufferSlices[sliceIndex].size, bufferSlices[sliceIndex].offset, totalSize);
145 : }
146 0 : return HCCL_SUCCESS;
147 : }
148 :
149 0 : HcclResult CollAllReduceDeterPipelineExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
150 : {
151 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllReduceDeterPipelineExecutor][KernelRun] tag[%s], userRank[%u] starts.",
152 : tag_.c_str(), topoAttr_.userRank);
153 :
154 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
155 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
156 0 : u32 commIndex = level0CommInfo.localRank;
157 :
158 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
159 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
160 :
161 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
162 0 : std::vector<Slice> bufferSlices; // 数据分成ranksize份，每份的起始偏移和大小
163 0 : CHK_RET(PrepareDataSlice(execMem, unitSize, bufferSlices));
164 :
165 0 : CHK_RET(ActiveSlaveStreams(param.stream));
166 :
167 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
168 0 : TemplateType::TEMPLATE_ALL_REDUCE_MULTI_DETERMINISTIC_PIPELINE, dispatcher_);
169 0 : CHK_SMART_PTR_NULL(tempAlg);
170 :
171 0 : HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType,
172 0 : param.root, param.reduceType};
173 :
174 0 : CHK_RET(tempAlg->Prepare(&opInfo, execMem.inputMem, execMem.outputMem, execMem.count, bufferSlices, level0CommInfo,
175 : level1CommInfo, const_cast<Stream&>(param.stream), algResResp_->slaveStreams, algResResp_->notifiesMain,
176 : algResResp_->notifiesAux));
177 0 : CHK_RET(tempAlg->RunAsync());
178 :
179 0 : HCCL_INFO("[CollAllReduceDeterPipelineExecutor][KernelRun] tag[%s], userRank[%u] run success.",
180 : tag_.c_str(), topoAttr_.userRank);
181 0 : return HCCL_SUCCESS;
182 0 : }
183 :
184 : REGISTER_EXEC("AllReduceDeterPipelineExecutor", AllReduceDeterPipeline,
185 : CollAllReduceDeterPipelineExecutor);
186 :
187 : }
188 :
|
