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_receive_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollReceiveExecutor::CollReceiveExecutor(const HcclDispatcher dispatcher,
16 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
17 0 : : CollNativeExecutorBase(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
20 0 : }
21 :
22 0 : HcclResult CollReceiveExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
23 : {
24 0 : HcclUs startut = TIME_NOW();
25 0 : tag_ = param.tag;
26 0 : algResResp_ = &algRes;
27 :
28 0 : HcclResult ret = HCCL_SUCCESS;
29 : // 图模式场景下不需要Loop
30 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
31 0 : DeviceMem outputMem = algRes.paramOutputMem;
32 0 : ret = RunTemplate(param, outputMem);
33 0 : } else {
34 0 : ret = RunLoop(param, algRes);
35 : }
36 :
37 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
38 : HCCL_ERROR("[CollReceiveExecutor][Orchestrate]errNo[0x%016llx]recv executor kernel run failed",
39 : HCCL_ERROR_CODE(ret)), ret);
40 :
41 0 : HCCL_INFO("tag[%s], Receive Executor orchestrate success, take time [%lld]us.",
42 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 0 : HcclResult CollReceiveExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
47 : {
48 : (void) algRes;
49 : (void) adjInfo;
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult CollReceiveExecutor::CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType)
54 : {
55 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
56 0 : inputType = TransportMemType::CCL_OUTPUT;
57 0 : outputType = TransportMemType::CCL_OUTPUT;
58 : } else {
59 0 : inputType = TransportMemType::PARAM_INPUT;
60 0 : outputType = TransportMemType::PARAM_OUTPUT;
61 : }
62 0 : HCCL_INFO("[CollRecvExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
63 : tag_.c_str(), inputType, outputType);
64 0 : return HCCL_SUCCESS;
65 : }
66 :
67 0 : HcclResult CollReceiveExecutor::CalcP2PCommInfo(TransportMemType inputType,
68 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport, u32 srcRank)
69 : {
70 0 : HCCL_INFO("[CollRecvExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
71 0 : CommParaInfo commP2P(COMM_COMBINE, CommType::COMM_TAG_P2P);
72 0 : commP2P.peerUserRank = srcRank;
73 0 : CHK_RET(CalcCommPlaneInfo(tag_, commP2P, opTransport[COMM_COMBINE], inputType, outputType));
74 0 : HCCL_INFO("[CollRecvExecutor][CalcLevel0CommInfo]tag[%s] Calc RingComm finish", tag_.c_str());
75 0 : return HCCL_SUCCESS;
76 0 : }
77 :
78 0 : HcclResult CollReceiveExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport, u32 srcRank)
79 : {
80 0 : TransportMemType inputType = TransportMemType::RESERVED;
81 0 : TransportMemType outputType = TransportMemType::RESERVED;
82 0 : CalcTransportMemType(inputType, outputType);
83 0 : CalcP2PCommInfo(inputType, outputType, opTransport, srcRank);
84 0 : return HCCL_SUCCESS;
85 : }
86 :
87 0 : HcclResult CollReceiveExecutor::CalcResRequest(const OpParam& param, AlgResourceRequest& resourceRequest)
88 : {
89 0 : ParseParam(param);
90 :
91 0 : u64 scratchMemSize = 0U;
92 0 : u32 streamNum = 0U;
93 0 : u32 notifyNum = 0U;
94 0 : u64 aivBufferRequest = 0U;
95 : std::vector<LevelNSubCommTransport> opTransport {
96 0 : std::vector<LevelNSubCommTransport>(static_cast<u32>(COMM_LEVEL_RESERVED))
97 0 : };
98 :
99 0 : CHK_RET(CalcCommInfo(opTransport, param.srcRank));
100 :
101 0 : CHK_RET(BuildResourceRequest(scratchMemSize, streamNum, notifyNum, aivBufferRequest, opTransport, resourceRequest));
102 0 : HCCL_INFO("streamNum[%u], notifyNum[%u], sctrachMemSize[%llu], aivBufferRequest[%llu]",
103 : resourceRequest.streamNum, resourceRequest.notifyNum, resourceRequest.scratchMemSize,
104 : resourceRequest.aivBufferRequest);
105 : // 打印建链诉求
106 0 : PrintTransportRequest(resourceRequest);
107 0 : return HCCL_SUCCESS;
108 0 : }
109 :
110 0 : HcclResult CollReceiveExecutor::RunLoop(OpParam ¶m, AlgResourceResponse &algRes)
111 : {
112 : HcclResult ret;
113 :
114 0 : u64 commOutputSize = algRes.cclOutputMem.size();
115 :
116 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
117 :
118 0 : auto meta = HcclOpMetaInfo::GetOneForRecieve();
119 0 : u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
120 0 : CHK_PTR_NULL(curOutputPtr);
121 :
122 0 : u64 outputOffset = 0;
123 0 : u64 countLeft = param.DataDes.count;
124 0 : while (countLeft > 0) {
125 0 : CHK_RET(InitTask(dispatcher_, param.stream, meta.isEnableCache, meta.GetCacheKey()));
126 0 : curOutputPtr += outputOffset;
127 0 : HCCL_DEBUG("RecvOutPlace:outputOffset[%llu]", outputOffset);
128 0 : u64 curCount = ((countLeft * unitSize) > commOutputSize) ? (commOutputSize / unitSize) : countLeft;
129 0 : u64 curSize = curCount * unitSize; // 单位 byte
130 0 : HCCL_DEBUG("RecvOutPlace:curOutputPtr[%p], curCount[%llu], curSize[%llu]", curOutputPtr, curCount, curSize);
131 :
132 0 : if(topoAttr_.deviceType != DevType::DEV_TYPE_910_93 || topoAttr_.isDiffDeviceType ||
133 0 : (topoAttr_.superPodNum > 1 || (topoAttr_.moduleNum > 1 && static_cast<bool>(topoMatcher_->GetExternalInputInterHccsDisable())))) {
134 : // 非A3场景不做DMA消减；A3的RDMA场景，也不做DMA消减
135 0 : DeviceMem outCommMem(algRes.cclOutputMem.ptr(), curSize);
136 0 : DeviceMem outMem(curOutputPtr, curSize);
137 0 : ret = RunTemplate(param, outCommMem);
138 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
139 : HCCL_ERROR("errNo[0x%016llx] RecvOutPlace: recv error, tag[%s], ptr[%p], count[%llu], dataType[%d]",
140 : HCCL_ERROR_CODE(ret), param.tag.c_str(), curOutputPtr, curCount, param.DataDes.dataType),
141 : ret);
142 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outMem, outCommMem, param.stream));
143 0 : HCCL_DEBUG("[CollReceiveExecutor][RunLoop]copy from ccl output to user output.");
144 0 : } else {
145 : // A3的SDMA场景，recv端做DMA消减
146 0 : DeviceMem outMem(curOutputPtr, curSize);
147 0 : ret = RunTemplate(param, outMem);
148 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
149 : HCCL_ERROR("errNo[0x%016llx] RecvOutPlace: recv error, tag[%s], ptr[%p], count[%llu], dataType[%d]",
150 : HCCL_ERROR_CODE(ret), param.tag.c_str(), curOutputPtr, curCount, param.DataDes.dataType),
151 : ret);
152 0 : HCCL_DEBUG("[CollReceiveExecutor][RunLoop]copy from ccl input to user output.");
153 0 : }
154 :
155 0 : CHK_PRT_RET((curCount == 0), HCCL_ERROR("In OP_BASE curCount is zero"), HCCL_E_PARA);
156 0 : countLeft -= curCount;
157 0 : outputOffset = curSize;
158 :
159 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
160 : }
161 0 : return HCCL_SUCCESS;
162 : }
163 :
164 0 : HcclResult CollReceiveExecutor::RunTemplate(const OpParam ¶m, DeviceMem &outputMem)
165 : {
166 0 : SubCommInfo commInfo = GetSubCommInfo(COMM_COMBINE, 0);
167 0 : if (commInfo.links.size() == 0) {
168 0 : HCCL_ERROR("[CollReceiveExecutor]links size is 0");
169 : }
170 0 : LINK transportLink = commInfo.links[0];
171 :
172 0 : SendReceive ReceiveExecutor(dispatcher_, transportLink);
173 0 : CHK_RET(ReceiveExecutor.ReceivePrepare(outputMem, param.srcRank, param.stream));
174 0 : CHK_RET(ReceiveExecutor.RegisterProfiler(0, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
175 0 : CHK_RET(ReceiveExecutor.ReceiveRunAsync());
176 :
177 0 : return HCCL_SUCCESS;
178 0 : }
179 :
180 : REGISTER_EXEC("ReceiveExecutor", Receive, CollReceiveExecutor);
181 :
182 : } // namespace hccl
|
