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