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 "notify_manager.h"
12 : #include "adapter_hal_pub.h"
13 : #include "device_capacity.h"
14 : #include "aicpu_launch_manager.h"
15 :
16 : namespace hccl {
17 : #ifndef CCL_KERNEL_AICPU
18 332 : NotifyManager::NotifyManager(std::string commId, aclrtBinHandle binHandle, const ManagerCallbacks& callbacks) :
19 332 : commId_(commId), binHandle_(binHandle), callbacks_(callbacks){}
20 : #endif
21 :
22 0 : HcclResult NotifyManager::InitNotifys(std::istringstream &iss, size_t notifyNum,
23 : std::vector<std::unique_ptr<LocalNotify>> &newNotifys)
24 : {
25 0 : newNotifys.reserve(newNotifys.size() + notifyNum);
26 0 : for (u32 idx = 0; idx < notifyNum; idx++) {
27 0 : std::unique_ptr<LocalNotify> notify;
28 0 : EXCEPTION_CATCH(notify = std::make_unique<LocalNotify>(), return HCCL_E_PTR);
29 : HcclSignalInfo notifyInfo;
30 0 : iss.read(reinterpret_cast<char_t *>(¬ifyInfo), sizeof(notifyInfo));
31 0 : CHK_RET(notify->Init(notifyInfo, NotifyLoadType::DEVICE_NOTIFY));
32 0 : newNotifys.emplace_back(std::move(notify));
33 0 : HCCL_INFO("[NotifyManager][Init]local notify init success, resId[%u], tsId:%d, devId[%u]",
34 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
35 0 : }
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 :
40 0 : HcclResult NotifyManager::ParseBinNotifys(const std::string& uniqueIdStr,
41 : std::vector<std::unique_ptr<LocalNotify>> &newNotifys)
42 : {
43 0 : bool isDeviceSid = false;
44 0 : CHK_RET(GetRunSideIsDevice(isDeviceSid));
45 : NotifyLoadType loadType;
46 0 : size_t notifyNum = 0;
47 0 : if (!isDeviceSid) {
48 0 : HCCL_ERROR("[NotifyManager][%s] not in deviceSide", __func__);
49 0 : return HCCL_E_NOT_SUPPORT;
50 : } else {
51 0 : CHK_PRT_RET(uniqueIdStr.empty(), HCCL_ERROR("[HcclThread][%s] uniqueIdStr is empty"), HCCL_E_INTERNAL);
52 0 : std::istringstream iss(uniqueIdStr);
53 0 : iss.read(reinterpret_cast<char_t *>(&loadType), sizeof(loadType));
54 0 : iss.read(reinterpret_cast<char_t *>(¬ifyNum), sizeof(notifyNum));
55 0 : CHK_RET(InitNotifys(iss, notifyNum, newNotifys));
56 0 : }
57 0 : HCCL_RUN_INFO("[NotifyManager][%s] recover success, notifyNum[%zu], notifyType[%d], uniqueIdSize[%zu]",
58 : __func__, notifyNum, loadType, uniqueIdStr.size());
59 0 : return HCCL_SUCCESS;
60 : }
61 :
62 : #ifndef CCL_KERNEL_AICPU
63 0 : std::string NotifyManager::GetBinNotifys(std::vector<std::unique_ptr<LocalNotify>> &newNotifys,
64 : const NotifyLoadType notifyType)
65 : {
66 0 : std::string uniqueIdStr;
67 0 : std::ostringstream oss;
68 0 : size_t notifyNum = newNotifys.size();
69 0 : oss.write(reinterpret_cast<const char_t *>(¬ifyType), sizeof(notifyType));
70 0 : oss.write(reinterpret_cast<const char_t *>(¬ifyNum), sizeof(notifyNum));
71 0 : HcclResult ret = HCCL_SUCCESS;
72 0 : for (u32 idx = 0; idx < notifyNum; idx++) {
73 : HcclSignalInfo notifyInfo;
74 0 : ret = newNotifys[idx]->GetNotifyData(notifyInfo);
75 0 : if (ret != HCCL_SUCCESS) {
76 0 : HCCL_ERROR("[NotifyManager][%s] GetNotifyData failed, ret[%d]", __func__, ret);
77 0 : std::string temp = std::string();
78 0 : return temp;
79 0 : }
80 0 : HCCL_INFO("[NotifyManager][%s] get local notify data success, resId[%u], tsId:%d, devId[%u]",
81 : __func__, notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
82 0 : oss.write(reinterpret_cast<const char_t *>(¬ifyInfo), sizeof(notifyInfo));
83 : }
84 0 : HCCL_RUN_INFO("[NotifyManager][%s] GetUniqueId success, notifyNum[%zu], notifyType[%u], uniqueId[%s]",
85 : __func__, notifyNum, notifyType, oss.str().c_str());
86 0 : uniqueIdStr = oss.str();
87 0 : return uniqueIdStr;
88 0 : }
89 :
90 0 : HcclResult NotifyManager::NotifyTypeToNotifyLoadType(::NotifyType notifyType, NotifyLoadType ¬ifyLoadType)
91 : {
92 0 : switch (notifyType) {
93 0 : case ::NOTIFY_TYPE_RTS_NOTIFY:
94 : case ::NOTIFY_TYPE_RTS_EVENT:
95 0 : notifyLoadType = NotifyLoadType::HOST_NOTIFY;
96 0 : break;
97 0 : case ::NOTIFY_TYPE_DEVICE_MEM:
98 0 : notifyLoadType = NotifyLoadType::DEVICE_NOTIFY;
99 0 : break;
100 0 : default:
101 0 : HCCL_ERROR("[NotifyManager] Unknown comm notifyType notifyLoadType: %d", notifyType);
102 0 : return HCCL_E_PARA;
103 : }
104 0 : return HCCL_SUCCESS;
105 : }
106 :
107 0 : HcclResult NotifyManager::HcclAllocNotify(CommEngine commEngine, ::NotifyType notifyType, uint32_t notifyNum,
108 : NotifyHandle **notifyHandleList)
109 : {
110 0 : std::lock_guard<std::mutex> lock(notifyMutex_);
111 0 : notifys_.reserve(notifys_.size() + notifyNum);
112 :
113 0 : std::vector<std::unique_ptr<LocalNotify>> newNotifys;
114 0 : newNotifys.reserve(notifyNum);
115 : NotifyLoadType notifyLoadType;
116 0 : CHK_PRT(NotifyTypeToNotifyLoadType(notifyType, notifyLoadType));
117 0 : bool isAicpu = (commEngine == CommEngine::COMM_ENGINE_AICPU || commEngine == CommEngine::COMM_ENGINE_AICPU_TS);
118 :
119 : // 构建 LocalNotify
120 0 : for (uint32_t i = 0; i < notifyNum; ++i) {
121 0 : std::unique_ptr<LocalNotify> notify;
122 0 : EXCEPTION_CATCH(notify = std::make_unique<LocalNotify>(), return HCCL_E_PTR);
123 0 : CHK_RET(notify->Init(notifyLoadType));
124 0 : if (Is310PDevice()) {
125 0 : CHK_RET(notify->SetIpc());
126 : }
127 0 : newNotifys.emplace_back(std::move(notify));
128 0 : }
129 :
130 0 : std::unique_ptr<NotifyHandle[]> handles;
131 0 : EXCEPTION_CATCH(handles = std::make_unique<NotifyHandle[]>(notifyNum), return HCCL_E_PTR);
132 0 : if (isAicpu) {
133 0 : if (!callbacks_.getAicpuCommState()) {
134 0 : HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
135 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
136 : HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
137 0 : callbacks_.setAicpuCommState(true);
138 : }
139 0 : CHK_RET(AicpuLaunchMgr::NotifyKernelLaunchAlloc(newNotifys, commId_, handles, binHandle_));
140 0 : for (uint32_t i = 0; i < notifyNum; ++i) {
141 0 : HCCL_INFO("[NotifyManager][%s] aicpu handles[%u] = [%llu]", __func__, i, handles[i]);
142 : }
143 : } else {
144 0 : for (uint32_t i = 0; i < notifyNum; ++i) {
145 0 : handles[i] = reinterpret_cast<NotifyHandle>(newNotifys[i].get());
146 0 : HCCL_INFO("[NotifyManager][%s] host handles[%u] = [%llu]", __func__, i, handles[i]);
147 : }
148 : }
149 0 : for (uint32_t i = 0; i < notifyNum; ++i) {
150 0 : LocalNotify *local = newNotifys[i].get();
151 0 : NotifyInfo info{commEngine, notifyType, isAicpu, handles[i]};
152 0 : notifysInfo_[local] = info;
153 : }
154 : // 插入到 notifys_ 尾部
155 0 : notifys_.insert(notifys_.end(), std::make_move_iterator(newNotifys.begin()),
156 : std::make_move_iterator(newNotifys.end()));
157 :
158 0 : handleBlocks_.push_back(std::move(handles));
159 0 : *notifyHandleList = handleBlocks_.back().get();
160 0 : return HCCL_SUCCESS;
161 0 : }
162 :
163 0 : HcclResult NotifyManager::HcommFreeNotify(uint32_t notifyNum, NotifyHandle *notifyHandleList)
164 : {
165 0 : std::lock_guard<std::mutex> lock(notifyMutex_);
166 :
167 0 : std::vector<LocalNotify*> localNotifys;
168 0 : localNotifys.reserve(notifyNum);
169 0 : std::vector<NotifyHandle> aicpuNotifys;
170 :
171 : // 1. 预扫描，判断是否为 AICPU，并收集 LocalNotify 指针
172 0 : for (uint32_t i = 0; i < notifyNum; ++i) {
173 0 : NotifyHandle handle = notifyHandleList[i];
174 0 : HCCL_INFO("[NotifyManager][%s] handles[%u] = [%llu]", __func__, i, handle);
175 0 : auto itInfo = std::find_if(notifysInfo_.begin(), notifysInfo_.end(),
176 0 : [handle](const auto &pair) {
177 0 : return pair.second.notifyHandle == handle;
178 : });
179 0 : if (itInfo == notifysInfo_.end()) {
180 0 : HCCL_RUN_WARNING("[NotifyManager][%s] handle[%llu] not found in notifysInfo_", __func__, handle);
181 0 : continue;
182 : }
183 0 : LocalNotify *localNotify = itInfo->first;
184 0 : const NotifyInfo &info = itInfo->second;
185 0 : if (info.isAicpu) {
186 0 : aicpuNotifys.push_back(handle);
187 : }
188 0 : localNotifys.push_back(localNotify);
189 : }
190 :
191 : // 2. 先释放 Device 侧（若失败则直接返回，不动 Host）
192 0 : bool hasAicpu = !aicpuNotifys.empty();
193 0 : if (hasAicpu) {
194 0 : HcclResult ret = AicpuLaunchMgr::NotifyKernelLaunchFree(aicpuNotifys, aicpuNotifys.size(), commId_, binHandle_);
195 0 : if (ret != HCCL_SUCCESS) {
196 0 : HCCL_ERROR("[NotifyManager][%s] NotifyKernelLaunchFree failed ret[%d], num[%zu], skip host erase",
197 : __func__, ret, aicpuNotifys.size());
198 0 : return ret; // 保留 Host 状态以便恢复
199 : }
200 : }
201 :
202 : // 3. 成功后再移除 Host 侧
203 0 : for (auto *localNotify : localNotifys) {
204 0 : notifysInfo_.erase(localNotify);
205 0 : auto it = std::find_if(notifys_.begin(), notifys_.end(),
206 0 : [localNotify](const std::unique_ptr<LocalNotify>& ptr) {
207 0 : return ptr.get() == localNotify;
208 : });
209 0 : if (it != notifys_.end()) {
210 0 : notifys_.erase(it);
211 : }
212 : }
213 :
214 : // 4. 删除对应的 handle block
215 0 : auto itBlock = std::find_if(handleBlocks_.begin(), handleBlocks_.end(),
216 0 : [notifyHandleList](const std::unique_ptr<NotifyHandle[]>& block) {
217 0 : return block.get() == notifyHandleList;
218 : });
219 0 : if (itBlock == handleBlocks_.end()) {
220 0 : HCCL_RUN_WARNING("[NotifyManager][%s] itBlock not found for notifyHandleList[%p]", __func__, notifyHandleList);
221 : } else {
222 0 : handleBlocks_.erase(itBlock);
223 : }
224 0 : return HCCL_SUCCESS;
225 0 : }
226 : #endif
227 : }
|
