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