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 <algorithm>
12 : #include <future>
13 : #include <map>
14 : #include <string>
15 : #include <hccl/hccl_types.h>
16 :
17 : #include "hccl/base.h"
18 : #include "param_check_pub.h"
19 : #include "externalinput_pub.h"
20 : #include "../common/src/state_guard.h"
21 : #include "sal_pub.h"
22 : #include "profiling_manager_pub.h"
23 : #include "adapter_prof.h"
24 : #include "adapter_rts_common.h"
25 : #include "error_codes/rt_error_codes.h"
26 : #include "op_base.h"
27 : #include "hccl_group.h"
28 :
29 : using namespace std;
30 : using namespace hccl;
31 :
32 330 : HcclResult GetCaptureInfo(aclrtStream stream, aclmdlRICaptureStatus& captureStatus, uint64_t& modelId, bool& isCapture)
33 : {
34 330 : isCapture = false;
35 330 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
36 0 : HCCL_WARNING("[%s]Stream capture only support opbase mode!", __func__);
37 0 : return HCCL_SUCCESS;
38 : }
39 330 : aclmdlRI rtModel = nullptr;
40 330 : aclError ret = aclmdlRICaptureGetInfo(stream, &captureStatus, &rtModel);
41 330 : if (ret == ACL_ERROR_RT_FEATURE_NOT_SUPPORT) {
42 0 : HCCL_WARNING("[%s]Stream capture does not support!", __func__);
43 0 : return HCCL_SUCCESS;
44 : } else {
45 330 : CHK_PRT_RET(
46 : ret != ACL_SUCCESS, HCCL_ERROR("[%s]rtGet stream get capture status fail. return[%d]", __func__, ret),
47 : HCCL_E_RUNTIME);
48 : }
49 330 : if (captureStatus == ACL_MODEL_RI_CAPTURE_STATUS_ACTIVE) {
50 13 : isCapture = true;
51 : uint32_t mdlId;
52 13 : rtError_t rtRet = rtModelGetId(rtModel, &mdlId);
53 13 : CHK_PRT_RET(
54 : rtRet != RT_ERROR_NONE, HCCL_ERROR("[%s]rtGet stream get model id fail. return[%d]", __func__, rtRet),
55 : HCCL_E_RUNTIME);
56 13 : modelId = static_cast<uint64_t>(mdlId);
57 : }
58 :
59 330 : return HCCL_SUCCESS;
60 : }
61 :
62 29 : HcclResult HcclAllReduceInner(
63 : void* sendBuf, void* recvBuf, uint64_t count, HcclDataType dataType, HcclReduceOp op, HcclComm comm,
64 : aclrtStream stream)
65 : {
66 : // 入参合法性校验
67 29 : CHK_PRT_RET(count == 0, HCCL_WARNING("input count is 0, return AllReduce success"), HCCL_SUCCESS);
68 48 : RPT_INPUT_ERR(
69 : comm == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
70 : std::vector<std::string>({"HcclAllReduceInner", "nullptr", "comm", "non-null pointer"}));
71 28 : CHK_PTR_NULL(comm);
72 47 : RPT_INPUT_ERR(
73 : sendBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
74 : std::vector<std::string>({"HcclAllReduceInner", "nullptr", "sendBuf", "non-null pointer"}));
75 27 : CHK_PTR_NULL(sendBuf);
76 46 : RPT_INPUT_ERR(
77 : recvBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
78 : std::vector<std::string>({"HcclAllReduceInner", "nullptr", "recvBuf", "non-null pointer"}));
79 26 : CHK_PTR_NULL(recvBuf);
80 25 : RPT_INPUT_ERR(
81 : stream == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "value"}),
82 : std::vector<std::string>({"HcclAllReduceInner", "nullptr", "stream", "non-null pointer"}));
83 25 : CHK_PTR_NULL(stream);
84 :
85 25 : if (hcclGroupDepth > 0) {
86 : struct hcclOpInfo info;
87 0 : info.coll = HcclCMDType::HCCL_CMD_ALLREDUCE;
88 0 : info.sendbuff = sendBuf;
89 0 : info.recvbuff = recvBuf;
90 0 : info.sendCount = count;
91 0 : info.sendType = dataType;
92 0 : info.recvType = dataType;
93 0 : info.op = op;
94 0 : info.comm = comm;
95 0 : info.stream = stream;
96 0 : CHK_RET(taskAppend(comm, info));
97 0 : HCCL_INFO(
98 : "[HcclAllReduce] Finish taskAppend, count [%d] dataType [%s]", count, GetDataTypeEnumStr(dataType).c_str());
99 0 : return HCCL_SUCCESS;
100 : }
101 25 : HcclUs startut = TIME_NOW();
102 :
103 : bool isCapture;
104 25 : aclmdlRICaptureStatus captureStatus = aclmdlRICaptureStatus::ACL_MODEL_RI_CAPTURE_STATUS_NONE;
105 25 : uint64_t modelId = 0xFFFFFFFF;
106 25 : CHK_PRT(GetCaptureInfo(stream, captureStatus, modelId, isCapture));
107 25 : if (!isCapture) {
108 24 : HcclSetIfProfile();
109 : }
110 :
111 25 : uint64_t beginTime = hrtMsprofSysCycleTime();
112 :
113 25 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
114 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
115 : CHK_RET(HcclAllReduceV2(sendBuf, recvBuf, count, dataType, op, hcclComm->GetCommunicatorV2(), stream));
116 : return HCCL_SUCCESS;
117 : }());
118 25 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
119 25 : const std::lock_guard<std::mutex> lock(hcclComm->operatorlock_);
120 25 : StateGuard<hccl::hcclComm, HcclCommState> guard(hcclComm, HcclCommState::INUSE);
121 25 : s32 threadID = SalGetTid();
122 25 : ProfilingManagerPub::SetThreadCaptureStatus(threadID, isCapture);
123 : // 同通信域同算子复用tag
124 25 : const string tag = "AllReduce_" + hcclComm->GetIdentifier();
125 :
126 25 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParam(tag.c_str(), count, dataType, stream), tag.c_str());
127 :
128 72 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOp("HcclAllReduceInner", op), tag.c_str());
129 : DevType devType;
130 24 : CHK_RET(hrtGetDeviceType(devType));
131 24 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataType(dataType, op, devType), tag.c_str());
132 :
133 : /* 接口交互信息日志 */
134 : char stackLogBuffer[LOG_TMPBUF_SIZE];
135 23 : if (GetExternalInputHcclEnableEntryLog()) {
136 23 : s32 deviceLogicId = 0;
137 23 : CHK_RET(hrtGetDeviceRefresh(&deviceLogicId));
138 :
139 23 : u32 localRank = INVALID_VALUE_RANKID;
140 23 : CHK_RET_AND_PRINT_IDE(hcclComm->GetUserRank(localRank), tag.c_str());
141 :
142 23 : s32 streamId = 0;
143 23 : CHK_RET_AND_PRINT_IDE(hrtGetStreamId(stream, streamId), tag.c_str());
144 :
145 46 : s32 ret = snprintf_s(
146 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
147 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%llu], dataType[%s], op[%s], localRank[%u], streamId[%d], "
148 : "comm[%p], deviceLogicId[%d]",
149 69 : tag.c_str(), sendBuf, recvBuf, count, GetDataTypeEnumStr(dataType).c_str(), GetReduceOpEnumStr(op).c_str(),
150 : localRank, streamId, comm, deviceLogicId);
151 :
152 23 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
153 :
154 46 : std::string logInfo = "Entry-HcclAllReduceInner: " + std::string(stackLogBuffer) + ", capture status["
155 69 : + to_string(captureStatus) + "], model id[" + to_string(modelId) + "].";
156 23 : CHK_RET_AND_PRINT_IDE(hcclComm->SaveTraceInfo(logInfo), tag.c_str());
157 23 : }
158 :
159 23 : CHK_RET_AND_PRINT_IDE(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE), tag.c_str());
160 :
161 23 : CHK_RET_AND_PRINT_IDE(PrintMemoryAttr(sendBuf), tag.c_str());
162 :
163 23 : CHK_RET_AND_PRINT_IDE(PrintMemoryAttr(recvBuf), tag.c_str());
164 :
165 23 : CHK_RET_AND_PRINT_IDE(SetOverFlowAddr(hcclComm), tag.c_str());
166 23 : CHK_RET_AND_PRINT_IDE(hcclComm->AllReduceOutPlace(tag, sendBuf, recvBuf, count, dataType, op, stream), tag.c_str());
167 23 : CHK_RET(CallMsprofReportHostApi(hcclComm, HcclCMDType::HCCL_CMD_ALLREDUCE, beginTime, count, dataType, tag));
168 :
169 23 : if (!isCapture) {
170 22 : HcclResetIfProfile();
171 : }
172 23 : ProfilingManagerPub::DeleteThreadCaptureStatus(threadID);
173 :
174 23 : if (GetExternalInputHcclEnableEntryLog()) {
175 23 : HcclUs endut = TIME_NOW();
176 : /* 关键状态记录 */
177 : std::string endInfo
178 46 : = "HcclAllReduceInner:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us,"
179 69 : + std::string(stackLogBuffer);
180 23 : CHK_RET_AND_PRINT_IDE(hcclComm->SaveTraceInfo(endInfo), tag.c_str());
181 23 : }
182 :
183 23 : return HCCL_SUCCESS;
184 25 : }
185 :
186 2 : HcclResult HcclBarrier(HcclComm comm, aclrtStream stream)
187 : {
188 : // 入参合法性校验
189 2 : CHK_PTR_NULL(comm);
190 1 : CHK_PTR_NULL(stream);
191 0 : HcclUs startut = TIME_NOW();
192 : bool isCapture;
193 0 : aclmdlRICaptureStatus captureStatus = aclmdlRICaptureStatus::ACL_MODEL_RI_CAPTURE_STATUS_NONE;
194 0 : uint64_t modelId = 0xFFFFFFFF;
195 0 : CHK_PRT(GetCaptureInfo(stream, captureStatus, modelId, isCapture));
196 0 : if (!isCapture) {
197 0 : HcclSetIfProfile();
198 : }
199 0 : s32 threadID = SalGetTid();
200 0 : ProfilingManagerPub::SetThreadCaptureStatus(threadID, isCapture);
201 0 : uint64_t beginTime = hrtMsprofSysCycleTime();
202 0 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
203 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
204 : CHK_RET(HcclBarrierV2(hcclComm->GetCommunicatorV2(), stream));
205 : return HCCL_SUCCESS;
206 : }());
207 :
208 : // Allreduce入参定义
209 0 : HcclDataType dataType = HCCL_DATA_TYPE_FP32;
210 0 : HcclReduceOp op = HCCL_REDUCE_SUM;
211 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
212 0 : StateGuard<hccl::hcclComm, HcclCommState> guard(hcclComm, HcclCommState::INUSE);
213 : // 同通信域同算子复用tag
214 0 : const string tag = "AllReduce_" + hcclComm->GetIdentifier();
215 :
216 : /* 接口交互信息日志 */
217 : char stackLogBuffer[LOG_TMPBUF_SIZE];
218 0 : if (GetExternalInputHcclEnableEntryLog()) {
219 0 : s32 deviceLogicId = 0;
220 0 : CHK_RET(hrtGetDeviceRefresh(&deviceLogicId));
221 :
222 0 : u32 localRank = INVALID_VALUE_RANKID;
223 0 : CHK_RET_AND_PRINT_IDE(hcclComm->GetUserRank(localRank), tag.c_str());
224 :
225 0 : s32 streamId = 0;
226 0 : CHK_RET_AND_PRINT_IDE(hrtGetStreamId(stream, streamId), tag.c_str());
227 :
228 0 : s32 ret = snprintf_s(
229 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
230 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%d], dataType[%s], op[%s], localRank[%u], streamId[%d], "
231 : "deviceLogicId[%d]",
232 : tag.c_str(), hcclComm->barrierSendBuf, hcclComm->barrierRecvBuf, HCCL_BARRIER_DEFAULT_COUNT,
233 0 : GetDataTypeEnumStr(dataType).c_str(), GetReduceOpEnumStr(op).c_str(), localRank, streamId, deviceLogicId);
234 :
235 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
236 0 : std::string logInfo = "Entry-HcclBarrier:" + std::string(stackLogBuffer) + ", capture status["
237 0 : + to_string(captureStatus) + "], model id[" + to_string(modelId) + "].";
238 0 : CHK_RET_AND_PRINT_IDE(hcclComm->SaveTraceInfo(logInfo), tag.c_str());
239 0 : }
240 :
241 0 : CHK_RET_AND_PRINT_IDE(hcclComm->CreateBarrierMemory(), tag.c_str());
242 :
243 0 : CHK_RET_AND_PRINT_IDE(PrintMemoryAttr(hcclComm->barrierSendBuf), tag.c_str());
244 :
245 0 : CHK_RET_AND_PRINT_IDE(PrintMemoryAttr(hcclComm->barrierRecvBuf), tag.c_str());
246 :
247 0 : CHK_RET_AND_PRINT_IDE(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE), tag.c_str());
248 :
249 0 : CHK_RET_AND_PRINT_IDE(
250 : hcclComm->AllReduceOutPlace(
251 : tag, hcclComm->barrierSendBuf, hcclComm->barrierRecvBuf, HCCL_BARRIER_DEFAULT_COUNT, dataType, op, stream,
252 : SyncMode::UNLIMITED_TIMEWAITSYNCMODE),
253 : tag.c_str());
254 :
255 0 : CHK_RET(CallMsprofReportHostApi(
256 : hcclComm, HcclCMDType::HCCL_CMD_ALLREDUCE, beginTime, HCCL_BARRIER_DEFAULT_COUNT, dataType, tag));
257 0 : if (!isCapture) {
258 0 : HcclResetIfProfile();
259 : }
260 0 : ProfilingManagerPub::DeleteThreadCaptureStatus(threadID);
261 :
262 0 : if (GetExternalInputHcclEnableEntryLog()) {
263 0 : HcclUs endut = TIME_NOW();
264 : /* 关键状态记录 */
265 0 : std::string endInfo = "HcclBarrier:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
266 0 : + " us," + std::string(stackLogBuffer);
267 0 : CHK_RET_AND_PRINT_IDE(hcclComm->SaveTraceInfo(endInfo), tag.c_str());
268 0 : }
269 :
270 0 : return HCCL_SUCCESS;
271 0 : }
|
