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