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