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