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_reduce_scatter_mesh_graph_pipeline_executor.h"
12 :
13 : namespace hccl {
14 0 : CollReduceScatterMeshGraphPipelineExecutor::CollReduceScatterMeshGraphPipelineExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher)
16 0 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
17 0 : {}
18 :
19 0 : void CollReduceScatterMeshGraphPipelineExecutor::ParseParam(const OpParam ¶m)
20 : {
21 0 : tag_ = param.tag;
22 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
23 0 : }
24 :
25 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::CalcStreamNum(u32 &streamNum)
26 : {
27 0 : streamNum = topoAttr_.deviceNumPerAggregation;
28 0 : HCCL_INFO(
29 : "[CollReduceScatterMeshGraphPipelineExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
30 0 : return HCCL_SUCCESS;
31 : }
32 :
33 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport> &opTransport)
34 : {
35 0 : TransportMemType inputType = TransportMemType::RESERVED;
36 0 : TransportMemType outputType = TransportMemType::RESERVED;
37 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
38 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
39 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
40 0 : return HCCL_SUCCESS;
41 : }
42 :
43 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::CalcTransportMemType(
44 : TransportMemType &inputType, TransportMemType &outputType)
45 : {
46 0 : inputType = TransportMemType::PARAM_INPUT;
47 0 : outputType = TransportMemType::PARAM_OUTPUT;
48 0 : HCCL_INFO("[CollReduceScatterMeshGraphPipelineExecutor][CalcTransportMemType]tag[%s] inputType[%d], outputType[%d]",
49 : tag_.c_str(),
50 : inputType,
51 : outputType);
52 0 : return HCCL_SUCCESS;
53 : }
54 :
55 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::CalcLevel0CommInfo(
56 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport> &opTransport)
57 : {
58 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_MESH);
59 0 : commParaInfo.meshSinglePlane = true;
60 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
61 0 : return HCCL_SUCCESS;
62 0 : }
63 :
64 : // PipeLine模式下使用Ring算法
65 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::CalcLevel1CommInfo(
66 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport> &opTransport)
67 : {
68 0 : CommParaInfo commParaInfo(COMM_LEVEL1, CommType::COMM_TAG_RING_INNER);
69 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL1], inputType, outputType));
70 0 : return HCCL_SUCCESS;
71 0 : }
72 :
73 0 : HcclResult CollReduceScatterMeshGraphPipelineExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
74 : {
75 0 : HCCL_CONFIG_INFO(HCCL_ALG,
76 : "[CollReduceScatterMeshGraphPipelineExecutor][KernelRun] userRank[%u] starts.", topoAttr_.userRank);
77 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
78 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
79 0 : u32 commIndex = level0CommInfo.localRank;
80 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
81 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
82 :
83 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
84 0 : DeviceMem userInMem = DeviceMem::create(param.inputPtr, param.DataDes.count * unitSize);
85 0 : u64 reduceAttr = GetReduceAttr(userInMem, userInMem, param.DataDes.dataType, param.reduceType);
86 :
87 0 : CHK_RET(ActiveSlaveStreams(param.stream));
88 :
89 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
90 0 : TemplateType::TEMPLATE_REDUCESCATTER_GRAPH_PIPELINE, dispatcher_);
91 0 : CHK_SMART_PTR_NULL(tempAlg);
92 :
93 0 : HcomCollOpInfo opInfo = {"",
94 0 : execMem.inputPtr,
95 0 : execMem.outputPtr,
96 0 : param.DataDes.count,
97 0 : param.DataDes.dataType,
98 0 : param.root,
99 0 : param.reduceType};
100 :
101 0 : CHK_RET(tempAlg->Prepare(&opInfo,
102 : execMem.inputMem,
103 : param.DataDes.count,
104 : 0,
105 : 0,
106 : level0CommInfo,
107 : level1CommInfo,
108 : const_cast<Stream &>(param.stream),
109 : algResResp_->slaveStreams,
110 : algResResp_->notifiesMain,
111 : algResResp_->notifiesAux,
112 : reduceAttr));
113 0 : CHK_RET(tempAlg->RunAsync());
114 :
115 0 : return HCCL_SUCCESS;
116 0 : }
117 :
118 : REGISTER_EXEC("ReduceScatterMeshGraphPipelineExecutor", ReduceScatterMeshGraphPipeline,
119 : CollReduceScatterMeshGraphPipelineExecutor);
120 : } // namespace hccl
|