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_mesh_opbase_pipeline_executor.h"
12 :
13 : namespace hccl {
14 : // 准入条件: pipeLine && 910B && 单算子 && sdmaReduce && rdmaReduce && 多Mesh && MeshTopo && 非确定性
15 0 : CollAllReduceMeshOpbasePipelineExecutor::CollAllReduceMeshOpbasePipelineExecutor(
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 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::CalcStreamNum(u32& streamNum)
23 : {
24 0 : u32 totalStreamNum = topoAttr_.deviceNumPerAggregation + 1U;
25 0 : streamNum = totalStreamNum - 1U;
26 0 : HCCL_INFO(
27 : "[CollAllReduceMeshOpbasePipelineExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
28 0 : return HCCL_SUCCESS;
29 : }
30 :
31 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::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 : HcclResult
42 0 : CollAllReduceMeshOpbasePipelineExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
43 : {
44 0 : inputType = TransportMemType::CCL_INPUT;
45 0 : outputType = TransportMemType::CCL_OUTPUT;
46 0 : HCCL_INFO(
47 : "[CollAllReduceMeshOpbasePipelineExecutor][CalcTransportMemType]"
48 : "tag[%s] inputType[%d], outputType[%d]",
49 : tag_.c_str(), inputType, outputType);
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::CalcLevel0CommInfo(
54 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
55 : {
56 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_MESH);
57 0 : commParaInfo.meshSinglePlane = true;
58 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
59 0 : return HCCL_SUCCESS;
60 0 : }
61 :
62 : // PipeLine模式下使用Ring算法
63 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::CalcLevel1CommInfo(
64 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
65 : {
66 0 : CommParaInfo commParaInfo(COMM_LEVEL1, CommType::COMM_TAG_RING_INNER);
67 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL1], inputType, outputType));
68 0 : return HCCL_SUCCESS;
69 0 : }
70 :
71 0 : u64 CollAllReduceMeshOpbasePipelineExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
72 : {
73 0 : if (cclBuffSize <= HCCL_MIN_SLICE_ALIGN_910B) {
74 0 : return 0;
75 : }
76 0 : u64 maxCountPerLoop = (cclBuffSize - HCCL_MIN_SLICE_ALIGN_910B) / unitSize * topoAttr_.userRankSize;
77 0 : return maxCountPerLoop;
78 : }
79 :
80 0 : bool CollAllReduceMeshOpbasePipelineExecutor::IsHugeData(const u64 curSize)
81 : {
82 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
83 0 : || curSize > SDMA_SEND_MAX_SIZE;
84 0 : return hugeData;
85 : }
86 :
87 0 : bool CollAllReduceMeshOpbasePipelineExecutor::IsSmallData(const u64 totalSize, const u64 curSize) { return false; }
88 :
89 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
90 : {
91 0 : HCCL_CONFIG_INFO(
92 : HCCL_ALG, "[CollAllReduceMeshOpbasePipelineExecutor][Run]CollAllReduceMeshOpbasePipelineExecutor begins.");
93 :
94 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
95 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
96 0 : u32 commIndex = level0CommInfo.localRank;
97 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
98 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
99 :
100 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
101 :
102 0 : HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, execMem.count, param.DataDes.dataType,
103 0 : param.root, param.reduceType};
104 :
105 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
106 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
107 0 : TemplateType::TEMPLATE_ALL_REDUCE_OPBASE_PIPELINE, dispatcher_);
108 0 : CHK_SMART_PTR_NULL(tempAlg);
109 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
110 0 : CHK_RET(tempAlg->Prepare(
111 : &opInfo, execMem.inputMem, execMem.outputMem, execMem.count, level1CommInfo, level0CommInfo,
112 : const_cast<Stream&>(param.stream), algResResp_->slaveStreams, algResResp_->notifiesMain,
113 : algResResp_->notifiesAux));
114 0 : CHK_RET(tempAlg->RunAsync());
115 0 : return HCCL_SUCCESS;
116 0 : }
117 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::Getlevel1CommRank(SubCommInfo& level1CommInfo)
118 : {
119 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
120 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
121 0 : u32 ringNum
122 0 : = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? LEVEL0_PLANE_NUM_IN_8PRING : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
123 0 : u32 commIndex = (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) ? topoAttr_.devicePhyId : level0CommInfo.localRank;
124 :
125 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
126 0 : level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
127 :
128 0 : return HCCL_SUCCESS;
129 0 : }
130 :
131 0 : HcclResult CollAllReduceMeshOpbasePipelineExecutor::SelectTempAlg(
132 : std::unique_ptr<AlgTemplateBase>& level1TempAlg, u32 level1RankSize)
133 : {
134 0 : if (level1RankSize > 1) {
135 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
136 0 : TemplateType::TEMPLATE_ALL_REDUCE_OPBASE_PIPELINE, dispatcher_);
137 0 : CHK_SMART_PTR_NULL(level1TempAlg);
138 0 : return HCCL_SUCCESS;
139 : }
140 0 : return HCCL_E_UNAVAIL;
141 : }
142 : REGISTER_EXEC(
143 : "AllReduceMeshOpbasePipelineExecutor", AllReduceMeshOpbasePipeline, CollAllReduceMeshOpbasePipelineExecutor);
144 :
145 : } // namespace hccl
|
