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