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 "order_launch.h"
12 : #include "log.h"
13 : #include "acl/acl_rt.h"
14 : #include "config_log.h"
15 : #include "hccl_types.h"
16 : #include "adapter_rts_common.h"
17 :
18 : namespace hccl {
19 1330 : OrderLaunch& OrderLaunch::GetInstance(s32 deviceLogicID)
20 : {
21 2110 : static OrderLaunch orderLaunch[MAX_MODULE_DEVICE_NUM];
22 1330 : if (static_cast<u32>(deviceLogicID) >= MAX_MODULE_DEVICE_NUM) {
23 273 : HCCL_WARNING("[OrderLaunch][GetInstance]Invalid deviceLogicID[%d]", deviceLogicID);
24 273 : return orderLaunch[0];
25 : }
26 1057 : HCCL_DEBUG("[OrderLaunch][GetInstance]Valid deviceLogicID[%d]", deviceLogicID);
27 1057 : return orderLaunch[deviceLogicID];
28 : }
29 :
30 780 : OrderLaunch::OrderLaunch() : initialized_(true) {}
31 :
32 780 : OrderLaunch::~OrderLaunch()
33 : {
34 780 : std::unique_lock<std::mutex> mapLock(streamMutex_);
35 780 : initialized_ = false;
36 780 : groupCtxMap_.clear();
37 780 : DestroyRes();
38 780 : }
39 :
40 780 : void OrderLaunch::DestroyRes()
41 : {
42 780 : for (auto& entry : contextResMgrMap_) {
43 0 : entry.second.DestroyResources();
44 : }
45 780 : contextResMgrMap_.clear();
46 780 : hcomStreamMap_.clear();
47 780 : }
48 :
49 523 : HcclResult OrderLaunch::RegisterOrderLaunch(const std::string& group)
50 : {
51 523 : std::unique_lock<std::mutex> mapLock(streamMutex_);
52 523 : if (groupCtxMap_.find(group) != groupCtxMap_.end()) {
53 1 : HCCL_WARNING("%s skip, group[%s] has already been registered", __func__, group.c_str());
54 1 : return HCCL_SUCCESS;
55 : }
56 : // 只记录group，context暂不赋值，只在算子下发阶段对context赋值
57 522 : groupCtxMap_.insert({group, INVALID_U64});
58 522 : HCCL_INFO("%s success, group[%s]", __func__, group.c_str());
59 522 : return HCCL_SUCCESS;
60 523 : }
61 :
62 : /**
63 : * @brief 从order launch系统注销group
64 : * 注销group时，会清理group与context的映射关系。
65 : * 只有当context下没有其他group时，才会清理context对应的资源。
66 : */
67 807 : HcclResult OrderLaunch::UnRegisterOrderLaunch(const std::string& group)
68 : {
69 807 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("OrderLaunch has been destroyed"), HCCL_SUCCESS);
70 807 : std::unique_lock<std::mutex> mapLock(streamMutex_);
71 807 : auto it = groupCtxMap_.find(group);
72 805 : if (it == groupCtxMap_.end()) {
73 285 : HCCL_WARNING("%s skip, group[%s] has not been registered", __func__, group.c_str());
74 285 : return HCCL_SUCCESS;
75 : }
76 :
77 519 : u64 context = it->second;
78 521 : HCCL_INFO(
79 : "[OrderLaunch][UnRegisterOrderLaunch] group[%s] context[0x%llx], contextGroupsMap_.size[%zu]", group.c_str(),
80 : context, contextGroupsMap_.size());
81 522 : if (contextGroupsMap_.find(context) != contextGroupsMap_.end()) {
82 0 : contextGroupsMap_[context].erase(group);
83 0 : if (contextGroupsMap_[context].empty()) {
84 0 : contextGroupsMap_.erase(context);
85 0 : if (contextResMgrMap_.find(context) != contextResMgrMap_.end()) {
86 0 : contextResMgrMap_[context].DestroyResources();
87 0 : contextResMgrMap_.erase(context);
88 : }
89 0 : HCCL_INFO("%s contextGroupsMap_ erase context[0x%llx]", __func__, context);
90 : }
91 : }
92 :
93 519 : groupCtxMap_.erase(it);
94 520 : HCCL_INFO("%s success, group[%s]", __func__, group.c_str());
95 522 : return HCCL_SUCCESS;
96 807 : }
97 :
98 : /**
99 : * @brief 设置图模式使用的HCOM stream
100 : * 图模式下，通信使用的附属从流预先设置到hcomStreamMap_中
101 : */
102 0 : HcclResult OrderLaunch::SetHcomStream(u32 graphId, const Stream& hcomAttachedStream)
103 : {
104 0 : std::unique_lock<std::mutex> mapLock(streamMutex_);
105 0 : hcomStreamMap_[graphId] = hcomAttachedStream;
106 0 : HCCL_INFO("%s success, graphId[%u], hcomStreamId[%u]", __func__, graphId, hcomAttachedStream.id());
107 0 : return HCCL_SUCCESS;
108 0 : }
109 :
110 : /**
111 : * @brief 初始化group的context映射关系
112 : *
113 : * 获取当前线程的context，建立以下映射关系：
114 : * - groupCtxMap_[group] = currentContext
115 : * - contextGroupsMap_[currentContext] 包含 group
116 : * - contextResMgrMap_[currentContext] 包含该context的资源管理器
117 : */
118 0 : HcclResult OrderLaunch::InitGroupCtx(const std::string& group)
119 : {
120 0 : u64 currentContext = INVALID_U64;
121 0 : HcclResult ret = GetCurrentContext(currentContext);
122 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]GetCurrentContext failed, ret[%d]", __func__, ret), ret);
123 :
124 0 : if (contextResMgrMap_.find(currentContext) == contextResMgrMap_.end()) {
125 0 : contextResMgrMap_[currentContext] = OrderLaunchResMgr();
126 0 : HCCL_INFO("[OrderLaunch][InitGroupCtx] created new OrderLaunchResMgr for context[0x%llx]", currentContext);
127 : }
128 :
129 0 : auto& resMgr = contextResMgrMap_[currentContext];
130 0 : if (!resMgr.contextInitialized) {
131 0 : resMgr.MarkContextInitialized(currentContext);
132 : }
133 :
134 0 : groupCtxMap_[group] = currentContext;
135 0 : contextGroupsMap_[currentContext].insert(group);
136 :
137 0 : HCCL_RUN_INFO("[%s]group[%s] init or update context[0x%llx]", __func__, group.c_str(), currentContext);
138 0 : return HCCL_SUCCESS;
139 : }
140 :
141 : // aclgraph模式下，先在kernel stream上写record，再在上order stream写wait；解order stream的wait
142 0 : HcclResult OrderLaunch::AclgraphLaunchInOrderToOrderStream(
143 : std::string& group, const Stream& kernelStream, std::shared_ptr<LocalNotify> notify0,
144 : std::shared_ptr<LocalNotify> notify1, u32 timeOut, HcclRtEvent event)
145 : {
146 0 : std::unique_lock<std::mutex> mapLock(streamMutex_);
147 : // group未注册过，或者未记录过算子下发阶段的线程context
148 0 : if (groupCtxMap_.find(group) == groupCtxMap_.end() || groupCtxMap_[group] == INVALID_U64) {
149 0 : CHK_RET(InitGroupCtx(group));
150 : }
151 :
152 0 : u64 context = groupCtxMap_[group];
153 0 : Stream& aclgraphStream = contextResMgrMap_[context].aclgraphStream;
154 0 : EnsureOrderStreamForGroup(group, context, aclgraphStream); // aclgraph控制流
155 :
156 0 : aclError ret = ACL_SUCCESS;
157 : // kernelStream -> aclgraphStream
158 0 : ret = aclrtRecordEvent(event, kernelStream.ptr());
159 0 : CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR("[%s]aclrtRecordEvent failed, ret[%d]", __func__, ret), HCCL_E_RUNTIME);
160 0 : HCCL_CONFIG_INFO(HCCL_TASK, "[%s]aclrtRecordEvent para: kernelStreamId[%d]", __func__, kernelStream.id());
161 :
162 0 : ret = aclrtStreamWaitEvent(aclgraphStream.ptr(), event);
163 0 : CHK_PRT_RET(
164 : ret != ACL_SUCCESS, HCCL_ERROR("[%s]aclrtStreamWaitEvent failed, ret[%d]", __func__, ret), HCCL_E_RUNTIME);
165 0 : HCCL_CONFIG_INFO(HCCL_TASK, "[%s]aclrtStreamWaitEvent para: orderStreamId[%d]", __func__, aclgraphStream.id());
166 :
167 0 : HCCL_INFO(
168 : "[%s] group[%s], kernelStreamId[%u], orderStreamId[%u], context[0x%llx]", __func__, group.c_str(),
169 : kernelStream.id(), aclgraphStream.id(), context);
170 0 : CHK_RET(LaunchInOrder(group, kernelStream, aclgraphStream, notify0, notify1, timeOut));
171 0 : return HCCL_SUCCESS;
172 0 : }
173 :
174 : /**
175 : * @brief ACLGRAPH模式第二步：在order stream上record事件并解kernel stream的wait
176 : * 执行流程：
177 : * 1. 在order stream上record事件
178 : * 2. 在kernel stream上wait该事件，解开kernel stream的阻塞
179 : */
180 : HcclResult
181 0 : OrderLaunch::AclgraphLaunchInOrderToKernelStream(std::string& group, const Stream& kernelStream, HcclRtEvent event)
182 : {
183 0 : std::unique_lock<std::mutex> mapLock(streamMutex_);
184 :
185 0 : auto ctxIt = groupCtxMap_.find(group);
186 0 : CHK_PRT_RET(
187 : ctxIt == groupCtxMap_.end(), HCCL_ERROR("[%s]fail, group[%s] is not in groupCtxMap_", __func__, group.c_str()),
188 : HCCL_E_NOT_FOUND);
189 :
190 0 : u64 context = ctxIt->second;
191 0 : if (contextResMgrMap_.find(context) == contextResMgrMap_.end()) {
192 0 : HCCL_ERROR("[%s]fail, context[0x%llx] is not in contextResMgrMap_", __func__, context);
193 0 : return HCCL_E_NOT_FOUND;
194 : }
195 :
196 0 : Stream& aclgraphStream = contextResMgrMap_[context].aclgraphStream;
197 :
198 0 : aclError ret = ACL_SUCCESS;
199 0 : ret = aclrtRecordEvent(event, aclgraphStream.ptr());
200 0 : CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR("[%s]aclrtRecordEvent failed, ret[%d]", __func__, ret), HCCL_E_RUNTIME);
201 0 : HCCL_CONFIG_INFO(HCCL_TASK, "[%s]aclrtRecordEvent para: orderStreamId[%d]", __func__, aclgraphStream.id());
202 :
203 0 : ret = aclrtStreamWaitEvent(kernelStream.ptr(), event);
204 0 : CHK_PRT_RET(
205 : ret != ACL_SUCCESS, HCCL_ERROR("[%s]aclrtStreamWaitEvent failed, ret[%d]", __func__, ret), HCCL_E_RUNTIME);
206 0 : HCCL_CONFIG_INFO(HCCL_TASK, "[%s]aclrtStreamWaitEvent para: kernelStreamId[%d]", __func__, kernelStream.id());
207 :
208 0 : HCCL_INFO(
209 : "[%s] group[%s], kernelStreamId[%u], orderStreamId[%u], context[0x%llx]", __func__, group.c_str(),
210 : kernelStream.id(), aclgraphStream.id(), ctxIt->second);
211 0 : return HCCL_SUCCESS;
212 0 : }
213 :
214 0 : HcclResult OrderLaunch::OpbaseLaunchInOrder(
215 : std::string& group, const Stream& kernelStream, std::shared_ptr<LocalNotify> notify0,
216 : std::shared_ptr<LocalNotify> notify1, u32 timeOut)
217 : {
218 0 : std::unique_lock<std::mutex> mapLock(streamMutex_);
219 0 : HCCL_INFO(
220 : "[OrderLaunch][OpbaseLaunchInOrder] group[%s], kernelStreamId[%u], timeOut[%d ms]", group.c_str(),
221 : kernelStream.id(), timeOut);
222 : // group未注册过，或者未记录过算子下发阶段的线程context
223 0 : if (groupCtxMap_.find(group) == groupCtxMap_.end() || groupCtxMap_[group] == INVALID_U64) {
224 0 : CHK_RET(InitGroupCtx(group));
225 : }
226 :
227 0 : u64 context = groupCtxMap_[group];
228 0 : Stream& opbaseStream = contextResMgrMap_[context].opbaseStream;
229 0 : EnsureOrderStreamForGroup(group, context, opbaseStream); // 单算子控制流
230 :
231 0 : HCCL_INFO(
232 : "[%s] group[%s], kernelStreamId[%u], orderStreamId[%u], context[0x%llx]", __func__, group.c_str(),
233 : kernelStream.id(), opbaseStream.id(), context);
234 0 : CHK_RET(LaunchInOrder(group, kernelStream, opbaseStream, notify0, notify1, timeOut));
235 0 : return HCCL_SUCCESS;
236 0 : }
237 :
238 : /**
239 : * @brief 图模式下的按序下发
240 : *
241 : * 图模式下，使用预先设置的hcomAttachedStream作为order stream
242 : */
243 0 : HcclResult OrderLaunch::HcomLaunchInOrder(
244 : std::string& group, const Stream& kernelStream, u32 graphId, std::shared_ptr<LocalNotify> notify0,
245 : std::shared_ptr<LocalNotify> notify1, u32 timeOut)
246 : {
247 0 : std::unique_lock<std::mutex> mapLock(streamMutex_);
248 0 : Stream hostOrderStream;
249 0 : if (hcomStreamMap_.find(graphId) == hcomStreamMap_.end()) {
250 0 : HCCL_ERROR("[%s] graphId[%u] group[%s] stream not found", __func__, graphId, group.c_str());
251 0 : return HCCL_E_NOT_FOUND;
252 : }
253 0 : hostOrderStream = hcomStreamMap_[graphId];
254 0 : CHK_PTR_NULL(hostOrderStream.ptr());
255 0 : HCCL_INFO("[%s] group[%s], graphId[%u], streamId[%u]", __func__, group.c_str(), graphId, hostOrderStream.id());
256 0 : CHK_RET(LaunchInOrder(group, kernelStream, hostOrderStream, notify0, notify1, timeOut));
257 0 : return HCCL_SUCCESS;
258 0 : }
259 :
260 : /**
261 : * @brief 使用notify机制实现stream间的按序下发
262 : * 1. wait notify0 on kernelStream - 等待kernel stream上的算子完成
263 : * 2. record notify0 on hostOrderStream - 在order stream上record notify
264 : * 3. wait notify1 on hostOrderStream - 等待其他算子在order stream上完成
265 : */
266 0 : HcclResult OrderLaunch::LaunchInOrder(
267 : [[maybe_unused]] std::string& group, const Stream& kernelStream, const Stream& hostOrderStream,
268 : std::shared_ptr<LocalNotify> notify0, std::shared_ptr<LocalNotify> notify1, u32 timeOut)
269 : {
270 0 : CHK_SMART_PTR_NULL(notify0);
271 0 : CHK_SMART_PTR_NULL(notify1);
272 0 : aclError ret = ACL_SUCCESS;
273 0 : ret = aclrtWaitAndResetNotify(notify0->ptr(), kernelStream.ptr(), timeOut);
274 0 : CHK_PRT_RET(
275 : ret != ACL_SUCCESS,
276 : HCCL_ERROR(
277 : "[%s] aclrtWaitAndResetNotify failed, ret[%d], notifyId[%u], streamId[%d], timeOut[%d s]", __func__, ret,
278 : notify0->notifyId_, kernelStream.id(), timeOut),
279 : HCCL_E_RUNTIME);
280 0 : HCCL_CONFIG_INFO(
281 : HCCL_TASK, "[%s] aclrtWaitAndResetNotify para: notifyId[%u], streamId[%d], timeOut[%d s]", __func__,
282 : notify0->notifyId_, kernelStream.id(), timeOut);
283 :
284 0 : ret = aclrtRecordNotify(notify0->ptr(), hostOrderStream.ptr());
285 0 : CHK_PRT_RET(
286 : ret != ACL_SUCCESS,
287 : HCCL_ERROR(
288 : "[%s] aclrtRecordNotify failed, ret[%d], notifyId[%u], streamId[%d]", __func__, ret, notify0->notifyId_,
289 : hostOrderStream.id()),
290 : HCCL_E_RUNTIME);
291 0 : HCCL_CONFIG_INFO(
292 : HCCL_TASK, "[%s] aclrtRecordNotify para: notifyId[%u], streamId[%d]", __func__, notify0->notifyId_,
293 : hostOrderStream.id());
294 :
295 0 : ret = aclrtWaitAndResetNotify(notify1->ptr(), hostOrderStream.ptr(), timeOut);
296 0 : CHK_PRT_RET(
297 : ret != ACL_SUCCESS,
298 : HCCL_ERROR(
299 : "[%s] aclrtWaitAndResetNotify failed, ret[%d], notifyId[%u], streamId[%d], timeOut[%d s]", __func__, ret,
300 : notify1->notifyId_, hostOrderStream.id(), timeOut),
301 : HCCL_E_RUNTIME);
302 0 : HCCL_CONFIG_INFO(
303 : HCCL_TASK, "[%s] aclrtWaitAndResetNotify para: notifyId[%u], streamId[%d], timeOut[%d s]", __func__,
304 : notify1->notifyId_, hostOrderStream.id(), timeOut);
305 0 : return HCCL_SUCCESS;
306 : }
307 :
308 0 : HcclResult OrderLaunch::EnsureOrderStreamForGroup(std::string& group, u64 context, Stream& orderStream)
309 : {
310 0 : auto it = groupCtxMap_.find(group);
311 0 : if (it == groupCtxMap_.end()) {
312 0 : HCCL_ERROR("[%s] group[%s] not found", __func__, group.c_str());
313 0 : return HCCL_E_PARA;
314 : }
315 :
316 0 : if (contextResMgrMap_.find(context) == contextResMgrMap_.end()) {
317 0 : HCCL_ERROR("[%s] context[0x%llx] not found for group[%s]", __func__, context, group.c_str());
318 0 : return HCCL_E_PARA;
319 : }
320 :
321 0 : auto& groupCtxRes = contextResMgrMap_[context];
322 :
323 0 : if (orderStream.ptr() == nullptr) {
324 0 : HCCL_INFO(
325 : "[OrderLaunch][EnsureOrderStreamForGroup] creating new order stream for group[%s], context[0x%llx]",
326 : group.c_str(), context);
327 : // Stream 尚未创建，基于传入的 context 创建
328 : // 创建新的order stream; streamMode = 1 使能遇错即停，避免出错后流卡住不退
329 0 : constexpr u32 streamMode = 1;
330 0 : orderStream = Stream(StreamType::STREAM_TYPE_ONLINE);
331 0 : CHK_RET(hrtStreamSetMode(orderStream.ptr(), streamMode));
332 0 : HCCL_INFO(
333 : "[OrderLaunch] Created new order stream with id[%d] with context [0x%llx]", orderStream.id(), context);
334 :
335 : // 对group->contextResource的映射关系进行更新
336 0 : if (!groupCtxRes.contextInitialized) {
337 0 : groupCtxRes.MarkContextInitialized(context);
338 : } else {
339 0 : groupCtxRes.UpdateContext(context);
340 : }
341 : // 对context->group的映射关系进行更新
342 0 : contextGroupsMap_[context].insert(group);
343 0 : HCCL_INFO(
344 : "[OrderLaunch] Added group[%s] to context [0x%llx] with order stream id[%d]", group.c_str(), context,
345 : orderStream.id());
346 : } else {
347 0 : HCCL_INFO(
348 : "[OrderLaunch][EnsureOrderStreamForGroup] order stream already exists, group[%s], context[0x%llx], "
349 : "streamId[%d]",
350 : group.c_str(), context, orderStream.id());
351 : }
352 0 : return HCCL_SUCCESS;
353 : }
354 :
355 : /**
356 : * @brief 获取当前线程的context
357 : * 通过hrtCtxGetCurrent接口获取当前线程关联的runtime context
358 : */
359 0 : HcclResult OrderLaunch::GetCurrentContext(u64& currentContext)
360 : {
361 0 : HcclRtContext rtCtx = nullptr;
362 0 : CHK_RET(hrtCtxGetCurrent(&rtCtx));
363 0 : currentContext = reinterpret_cast<u64>(rtCtx);
364 :
365 0 : if (currentContext == INVALID_U64) {
366 0 : HCCL_ERROR("[%s] GetCurrentContext failed", __func__);
367 0 : return HCCL_E_RUNTIME;
368 : }
369 :
370 0 : return HCCL_SUCCESS;
371 : }
372 : } // namespace hccl
|
