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_gatherv_for_310p_executor.h"
12 :
13 : #include <algorithm>
14 :
15 : namespace hccl {
16 0 : CollAllGatherVFor310PExecutor::CollAllGatherVFor310PExecutor(
17 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
18 0 : : CollAllGatherVExecutor(dispatcher, topoMatcher)
19 : {
20 0 : DMAReduceFlag_ = true;
21 0 : }
22 :
23 0 : HcclResult CollAllGatherVFor310PExecutor::CalcStreamNum(u32& streamNum)
24 : {
25 0 : streamNum = DMAReduceFlag_ ? 1 : 0;
26 0 : HCCL_INFO("[CollAllGatherVFor310PExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
27 :
28 0 : return HCCL_SUCCESS;
29 : }
30 :
31 0 : HcclResult CollAllGatherVFor310PExecutor::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 : return HCCL_SUCCESS;
38 : }
39 :
40 : HcclResult
41 0 : CollAllGatherVFor310PExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
42 : {
43 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
44 0 : inputType = TransportMemType::CCL_INPUT;
45 0 : outputType = TransportMemType::CCL_OUTPUT;
46 : } else {
47 0 : inputType = TransportMemType::PARAM_INPUT;
48 0 : outputType = TransportMemType::PARAM_OUTPUT;
49 : }
50 0 : HCCL_INFO(
51 : "[CollAllGatherVFor310PExecutor][CalcTransportMemType]"
52 : "tag[%s] inputType[%d], outputType[%d]",
53 : tag_.c_str(), inputType, outputType);
54 0 : return HCCL_SUCCESS;
55 : }
56 :
57 0 : HcclResult CollAllGatherVFor310PExecutor::CalcLevel0CommInfo(
58 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
59 : {
60 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
61 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
62 0 : return HCCL_SUCCESS;
63 0 : }
64 :
65 0 : u64 CollAllGatherVFor310PExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
66 : {
67 : // 中转内存单次最多能够接受的output count
68 0 : u64 maxCountPerLoop = cclBuffSize / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
69 :
70 0 : HCCL_INFO(
71 : "[CollAllGatherVExecutor][CalcLoopMaxCount]"
72 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.",
73 : maxCountPerLoop);
74 0 : return maxCountPerLoop;
75 : }
76 :
77 0 : bool CollAllGatherVFor310PExecutor::IsHugeData(const u64 curSize)
78 : {
79 0 : bool hugeData = curSize > HCCL_SMALL_COUNT_4_MB;
80 0 : return hugeData;
81 : }
82 :
83 0 : HcclResult CollAllGatherVFor310PExecutor::CalcCurCountsAndCurDispls(
84 : const u64 maxTotalCount, std::vector<u64>& countsLeft, std::vector<u64>& displs, std::vector<u64>& curCounts,
85 : std::vector<u64>& curDispls, bool& finished)
86 : {
87 0 : finished = true;
88 :
89 0 : curCounts.resize(countsLeft.size(), 0);
90 0 : curDispls.resize(displs.size(), 0);
91 :
92 : // 先设置本轮的displacements，等于入参displs
93 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
94 :
95 : // 分配好每个rank的counts
96 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
97 0 : const auto curCount = countsLeft[i] < maxTotalCount ? countsLeft[i] : maxTotalCount;
98 0 : curCounts[i] = curCount;
99 0 : countsLeft[i] -= curCount;
100 0 : displs[i] += curCount;
101 :
102 0 : if (countsLeft[i] != 0) {
103 0 : finished = false;
104 : }
105 : }
106 :
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 0 : HcclResult CollAllGatherVFor310PExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
111 : {
112 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherVFor310PExecutor][KernelRun] userRank[%u] starts.", topoAttr_.userRank);
113 0 : HcclDataType dataType = HCCL_DATA_TYPE_RESERVED;
114 0 : dataType = param.VDataDes.dataType;
115 0 : const u32 unitSize = SIZE_TABLE[dataType];
116 :
117 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
118 0 : SubCommInfo outerCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
119 :
120 0 : u32 rankSize = outerCommInfo.localRankSize;
121 :
122 : // allgatherv 计算slice，从外部传入
123 0 : std::vector<Slice> inputSlices;
124 0 : std::vector<Slice> outputSlices;
125 0 : const auto counts = static_cast<u64*>(param.VDataDes.counts);
126 0 : const auto displs = static_cast<u64*>(param.VDataDes.displs);
127 0 : u64 cclOffset = 0;
128 :
129 0 : for (u32 rank = 0; rank < rankSize; ++rank) {
130 0 : Slice cclslice;
131 0 : cclslice.offset = cclOffset;
132 0 : cclslice.size = counts[rank] * unitSize;
133 0 : cclOffset += cclslice.size;
134 0 : inputSlices.emplace_back(std::move(cclslice));
135 :
136 0 : Slice userslice;
137 0 : userslice.offset = displs[rank] * unitSize;
138 0 : userslice.size = counts[rank] * unitSize;
139 0 : outputSlices.emplace_back(std::move(userslice));
140 : }
141 :
142 0 : HcomCollOpInfo opInfo
143 0 : = {"", execMem.inputPtr, execMem.outputPtr, 0, param.VDataDes.dataType, param.root, param.reduceType, 0};
144 :
145 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
146 0 : if (!IsHugeData(cclOffset)) {
147 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
148 0 : TemplateType::TEMPLATE_ALL_GATHER_RING_DIRECT, dispatcher_);
149 0 : CHK_SMART_PTR_NULL(tempAlg);
150 0 : CHK_RET(tempAlg->Prepare(&opInfo, topoAttr_.userRank, outputSlices));
151 : } else {
152 0 : std::vector<u32> rankOrder(rankSize, 0);
153 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
154 0 : TemplateType::TEMPLATE_ALL_GATHER_RING_CONCURRENT_DIRECT, dispatcher_);
155 0 : CHK_SMART_PTR_NULL(tempAlg);
156 0 : CHK_RET(tempAlg->Prepare(
157 : &opInfo, topoAttr_.userRank, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
158 : rankOrder, outputSlices));
159 0 : }
160 :
161 0 : CHK_RET(tempAlg->Prepare(
162 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, dataType, param.stream, param.reduceType,
163 : 0, inputSlices));
164 :
165 0 : CHK_RET(tempAlg->RegisterProfiler(
166 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + outerCommInfo.localRank, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET,
167 : param.stream));
168 :
169 0 : CHK_RET(RunTemplate(tempAlg, outerCommInfo));
170 :
171 0 : HCCL_INFO("AllGatherV for 310P run success");
172 :
173 0 : return HCCL_SUCCESS;
174 0 : }
175 :
176 : REGISTER_EXEC("AllGatherVFor310PExecutor", AllGatherVFor310P, CollAllGatherVFor310PExecutor);
177 :
178 : } // namespace hccl
|
