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_send_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollSendExecutor::CollSendExecutor(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 CollSendExecutor::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 InputMem = algRes.paramInputMem;
31 0 : ret = RunTemplate(param, InputMem);
32 0 : } else {
33 0 : ret = RunLoop(param, algRes);
34 : }
35 :
36 0 : CHK_PRT_RET(
37 : ret != HCCL_SUCCESS,
38 : HCCL_ERROR(
39 : "[CollSendExecutor][Orchestrate]errNo[0x%016llx]send executor kernel run failed", HCCL_ERROR_CODE(ret)),
40 : ret);
41 :
42 0 : HCCL_INFO(
43 : "tag[%s] Send 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 CollSendExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
49 : {
50 0 : algResResp_ = &algRes;
51 :
52 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_COMBINE, 0);
53 0 : if (level1CommInfo.links.size() == 0) {
54 0 : HCCL_ERROR("[CollSendExecutor]links size is 0");
55 : }
56 0 : u32 localRank = level1CommInfo.localRank;
57 0 : u32 localRankSize = level1CommInfo.localRankSize;
58 :
59 0 : if (localRankSize == 1) {
60 0 : return HCCL_SUCCESS;
61 : }
62 0 : u32 ringNextRank = (localRank + 1) % localRankSize;
63 :
64 0 : adjInfo.dstRankNum = 1;
65 0 : NslbDpAdjInfo adjInfoStep = {};
66 0 : adjInfoStep.dstLocalRankId = ringNextRank;
67 0 : adjInfoStep.phaseId = 1;
68 0 : adjInfoStep.rev = 0;
69 0 : adjInfo.nsAdjInfo.push_back(adjInfoStep);
70 :
71 0 : HCCL_INFO("[nslbdp]GetAdjInfo localRank[%u], phaseId[%u].", localRank, ringNextRank);
72 0 : return HCCL_SUCCESS;
73 0 : }
74 :
75 0 : HcclResult CollSendExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
76 : {
77 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
78 0 : inputType = TransportMemType::CCL_INPUT;
79 0 : outputType = TransportMemType::CCL_INPUT;
80 : } else {
81 0 : inputType = TransportMemType::PARAM_INPUT;
82 0 : outputType = TransportMemType::PARAM_OUTPUT;
83 : }
84 0 : HCCL_INFO(
85 : "[CollSendExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(), inputType,
86 : outputType);
87 0 : return HCCL_SUCCESS;
88 : }
89 :
90 0 : HcclResult CollSendExecutor::CalcP2PCommInfo(
91 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport,
92 : u32 dstRank)
93 : {
94 0 : HCCL_INFO("[CollSendExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
95 0 : CommParaInfo commP2P(COMM_COMBINE, CommType::COMM_TAG_P2P);
96 0 : commP2P.peerUserRank = dstRank;
97 0 : CHK_RET(CalcCommPlaneInfo(tag_, commP2P, opTransport[COMM_COMBINE], inputType, outputType));
98 0 : HCCL_INFO("[CollSendExecutor][CalcLevel0CommInfo]tag[%s] Calc P2PComm finish", tag_.c_str());
99 0 : return HCCL_SUCCESS;
100 0 : }
101 :
102 0 : HcclResult CollSendExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport, u32 dstRank)
103 : {
104 0 : TransportMemType inputType = TransportMemType::RESERVED;
105 0 : TransportMemType outputType = TransportMemType::RESERVED;
106 0 : CalcTransportMemType(inputType, outputType);
107 0 : CHK_RET(CalcP2PCommInfo(inputType, outputType, opTransport, dstRank));
108 0 : return HCCL_SUCCESS;
109 : }
110 :
111 0 : HcclResult CollSendExecutor::CalcResRequest(const OpParam& param, AlgResourceRequest& resourceRequest)
112 : {
113 0 : ParseParam(param);
114 :
115 0 : u64 scratchMemSize = 0U;
116 0 : u32 streamNum = 0U;
117 0 : u32 notifyNum = 0U;
118 0 : u64 aivBufferRequest = 0U;
119 : std::vector<LevelNSubCommTransport> opTransport{
120 0 : std::vector<LevelNSubCommTransport>(static_cast<u32>(COMM_LEVEL_RESERVED))};
121 :
122 0 : CHK_RET(CalcCommInfo(opTransport, param.dstRank));
123 :
124 0 : CHK_RET(BuildResourceRequest(scratchMemSize, streamNum, notifyNum, aivBufferRequest, opTransport, resourceRequest));
125 0 : HCCL_INFO(
126 : "streamNum[%u], notifyNum[%u], sctrachMemSize[%llu], aivBufferRequest[%llu]", resourceRequest.streamNum,
127 : resourceRequest.notifyNum, resourceRequest.scratchMemSize, resourceRequest.aivBufferRequest);
128 : // 打印建链诉求
129 0 : PrintTransportRequest(resourceRequest);
130 0 : return HCCL_SUCCESS;
131 0 : }
132 :
133 0 : HcclResult CollSendExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
134 : {
135 : HcclResult ret;
136 :
137 0 : u64 commInputSize = algRes.cclInputMem.size();
138 :
139 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
140 0 : auto meta = HcclOpMetaInfo::GetOneForSend();
141 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
142 0 : CHK_PTR_NULL(curInputPtr);
143 :
144 0 : u64 inputOffset = 0;
145 0 : u64 countLeft = param.DataDes.count;
146 0 : while (countLeft > 0) {
147 0 : CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), meta.isEnableCache, meta.GetCacheKey()));
148 0 : curInputPtr += inputOffset;
149 :
150 0 : HCCL_DEBUG("SendOutPlace:inputOffset[%llu]", inputOffset);
151 0 : u64 curCount = ((countLeft * unitSize) > commInputSize) ? (commInputSize / unitSize) : countLeft;
152 0 : u64 curSize = curCount * unitSize; // 单位 byte
153 :
154 0 : HCCL_DEBUG("SendOutPlace:curInputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr, curCount, curSize);
155 :
156 : // 非A3场景，或A3的SDMA消减场景，send端不消减
157 0 : DeviceMem inCommMem(algRes.cclInputMem.ptr(), curSize);
158 0 : DeviceMem inMem(curInputPtr, curSize);
159 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, const_cast<Stream&>(param.stream)));
160 0 : HCCL_INFO(
161 : "[CollSendExecutor] inCommMem[%p]size[%lld], inMem[%p]size[%lld]", inCommMem.ptr(), inCommMem.size(),
162 : inMem.ptr(), inMem.size());
163 0 : ret = RunTemplate(param, inCommMem);
164 0 : CHK_PRT_RET(
165 : ret != HCCL_SUCCESS,
166 : HCCL_ERROR(
167 : "errNo[0x%016llx] SendOutPlace: send error, tag[%s], ptr[%p], count[%llu], dataType[%d]",
168 : HCCL_ERROR_CODE(ret), param.tag.c_str(), curInputPtr, curCount, param.DataDes.dataType),
169 : ret);
170 0 : HCCL_DEBUG("[CollSendExecutor][RunLoop]copy from user input to ccl input.");
171 :
172 0 : CHK_PRT_RET((curCount == 0), HCCL_ERROR("In OP_BASE curCount is zero"), HCCL_E_PARA);
173 0 : countLeft -= curCount;
174 0 : inputOffset = curSize;
175 :
176 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
177 0 : }
178 0 : return HCCL_SUCCESS;
179 : }
180 :
181 0 : HcclResult CollSendExecutor::RunTemplate(const OpParam& param, DeviceMem& inputMem)
182 : {
183 0 : SubCommInfo commInfo = GetSubCommInfo(COMM_COMBINE, 0);
184 0 : if (commInfo.links.size() == 0) {
185 0 : HCCL_ERROR("[CollSendExecutor]links size is 0");
186 : }
187 0 : LINK transportLink = commInfo.links[0];
188 :
189 0 : SendReceive sendExecutor(dispatcher_, transportLink);
190 0 : CHK_RET(sendExecutor.SendPrepare(inputMem, param.dstRank, param.stream));
191 0 : CHK_RET(sendExecutor.RegisterProfiler(0, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
192 :
193 0 : CHK_RET(sendExecutor.SendRunAsync());
194 :
195 0 : return HCCL_SUCCESS;
196 0 : }
197 :
198 : REGISTER_EXEC("SendExecutor", Send, CollSendExecutor);
199 :
200 : } // namespace hccl
|
