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_reduce_plus_bcast_executor.h"
12 :
13 : namespace hccl {
14 :
15 1 : CollAllReduceReducePlusBcastExecutor::CollAllReduceReducePlusBcastExecutor(
16 1 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 1 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 1 : {}
19 :
20 1 : HcclResult CollAllReduceReducePlusBcastExecutor::CalcStreamNum(u32& streamNum)
21 : {
22 1 : streamNum = 0;
23 1 : HCCL_INFO("[CollAllReduceReducePlusBcastExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
24 1 : return HCCL_SUCCESS;
25 : }
26 :
27 1 : HcclResult CollAllReduceReducePlusBcastExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
28 : {
29 1 : TransportMemType inputType = TransportMemType::RESERVED;
30 1 : TransportMemType outputType = TransportMemType::RESERVED;
31 1 : CHK_RET(CalcTransportMemType(inputType, outputType));
32 1 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
33 1 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
34 1 : return HCCL_SUCCESS;
35 : }
36 :
37 : HcclResult
38 1 : CollAllReduceReducePlusBcastExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
39 : {
40 1 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
41 0 : inputType = TransportMemType::CCL_INPUT;
42 0 : outputType = TransportMemType::CCL_OUTPUT;
43 : } else {
44 1 : inputType = TransportMemType::PARAM_INPUT;
45 1 : outputType = TransportMemType::PARAM_OUTPUT;
46 : }
47 1 : HCCL_INFO(
48 : "[CollAllReduceReducePlusBcastExecutor][CalcTransportMemType]"
49 : "tag[%s] inputType[%d], outputType[%d]",
50 : tag_.c_str(), inputType, outputType);
51 1 : return HCCL_SUCCESS;
52 : }
53 :
54 1 : HcclResult CollAllReduceReducePlusBcastExecutor::CalcLevel0CommInfo(
55 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
56 : {
57 1 : HCCL_INFO("[CollAllReduceReducePlusBcastExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
58 1 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
59 1 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
60 1 : HCCL_INFO("[CollAllReduceReducePlusBcastExecutor][CalcLevel0CommInfo]tag[%s] Calc RingComm finish", tag_.c_str());
61 1 : return HCCL_SUCCESS;
62 1 : }
63 :
64 0 : bool CollAllReduceReducePlusBcastExecutor::IsHugeData(const u64 curSize)
65 : {
66 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
67 0 : || curSize > SDMA_SEND_MAX_SIZE;
68 0 : return hugeData;
69 : }
70 :
71 0 : bool CollAllReduceReducePlusBcastExecutor::IsSmallData([[maybe_unused]] const u64 totalSize, const u64 curSize)
72 : {
73 0 : bool smallData = IsAllReduceSmallData(curSize);
74 0 : return smallData;
75 : }
76 :
77 1 : HcclResult CollAllReduceReducePlusBcastExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
78 : {
79 1 : HCCL_CONFIG_INFO(
80 : HCCL_ALG, "[CollAllReduceReducePlusBcastExecutor][KernelRun] userRank[%u] starts.", topoAttr_.userRank);
81 1 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
82 :
83 1 : std::unique_ptr<AlgTemplateBase> reduceTempAlg;
84 2 : reduceTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
85 1 : TemplateType::TEMPLATE_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
86 1 : CHK_SMART_PTR_NULL(reduceTempAlg);
87 1 : CHK_RET(reduceTempAlg->Prepare(reduceAttr));
88 :
89 2 : std::vector<u32> nicRankList{0, 1};
90 3 : CHK_RET(reduceTempAlg->Prepare(
91 : execMem.inputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
92 : param.reduceType, 0, std::vector<Slice>(0), 0, nicRankList));
93 :
94 1 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
95 1 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
96 1 : CHK_RET(RunTemplate(reduceTempAlg, level0CommInfo));
97 :
98 : // AllReduce算子实现为input->output, 所以此处将reduce算子的结果从output拷贝到input
99 : HcclResult ret
100 1 : = HcclD2DMemcpyAsync(dispatcher_, execMem.inputMem, execMem.outputMem, const_cast<Stream&>(param.stream));
101 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("MemcpyAsync failed"), ret);
102 :
103 : // 执行server间allreduce
104 1 : if (topoAttr_.devicePhyId == 0) {
105 1 : std::unique_ptr<AlgTemplateBase> allreduceTempAlg = nullptr;
106 1 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
107 : allreduceTempAlg
108 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
109 0 : HCCL_INFO("AllReduce ring: using ring algo inter-server.");
110 0 : CHK_SMART_PTR_NULL(allreduceTempAlg);
111 0 : CHK_RET(allreduceTempAlg->Prepare(reduceAttr));
112 1 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
113 0 : u64 curSize = execMem.count * SIZE_TABLE[param.DataDes.dataType]; // 单位 byte
114 0 : HCCL_DEBUG(
115 : "AllReduce recursive hd: curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]", curSize,
116 : topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
117 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
118 0 : allreduceTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
119 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
120 : } else {
121 0 : allreduceTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
122 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
123 : }
124 0 : HCCL_INFO("AllReduce recursive hd: using nhr algo inter-server.");
125 0 : CHK_SMART_PTR_NULL(allreduceTempAlg);
126 0 : CHK_RET(allreduceTempAlg->Prepare(reduceAttr));
127 0 : allreduceTempAlg->CloseBarrier();
128 1 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
129 : allreduceTempAlg
130 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
131 0 : HCCL_INFO("AllReduce recursive hd: using nhr_v1 algo inter-server.");
132 0 : CHK_SMART_PTR_NULL(allreduceTempAlg);
133 0 : CHK_RET(allreduceTempAlg->Prepare(reduceAttr));
134 1 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
135 : allreduceTempAlg
136 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
137 0 : HCCL_INFO("AllReduce recursive hd: using nb algo inter-server.");
138 0 : CHK_SMART_PTR_NULL(allreduceTempAlg);
139 0 : CHK_RET(allreduceTempAlg->Prepare(reduceAttr));
140 : } else {
141 2 : allreduceTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
142 1 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
143 1 : HCCL_INFO("AllReduce recursive hd: using halving-doubling algo inter-server.");
144 1 : CHK_SMART_PTR_NULL(allreduceTempAlg);
145 1 : CHK_RET(allreduceTempAlg->Prepare(reduceAttr));
146 : }
147 :
148 1 : CHK_SMART_PTR_NULL(allreduceTempAlg);
149 3 : CHK_RET(allreduceTempAlg->Prepare(
150 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
151 : param.reduceType, 0, std::vector<Slice>(0), 0, nicRankList));
152 :
153 1 : CHK_RET(CheckCommSize(COMM_LEVEL1, COMM_INDEX_0 + 1));
154 1 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, COMM_INDEX_0);
155 1 : CHK_RET(RunTemplate(allreduceTempAlg, level1CommInfo));
156 1 : }
157 :
158 : // 执行server内broadcast
159 1 : std::unique_ptr<AlgTemplateBase> bcastTempAlg;
160 1 : bcastTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_BROADCAST_RING, dispatcher_);
161 1 : CHK_SMART_PTR_NULL(bcastTempAlg);
162 5 : CHK_RET(bcastTempAlg->Prepare(
163 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
164 : param.reduceType, 0));
165 1 : CHK_RET(RunTemplate(bcastTempAlg, level0CommInfo));
166 :
167 1 : return HCCL_SUCCESS;
168 1 : }
169 :
170 : REGISTER_EXEC("AllReduceReducePlusBcast", AllReduceReducePlusBcast, CollAllReduceReducePlusBcastExecutor);
171 :
172 : } // namespace hccl
|
