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 <algorithm>
12 : #include "device_capacity.h"
13 : #include "sal_pub.h"
14 : #include "adapter_hal.h"
15 : #include "dlhal_function.h"
16 : #include "adapter_rts.h"
17 : #include "notify_pool_impl.h"
18 :
19 : namespace hccl {
20 : constexpr u32 NOTIFY_NORMAL = 0; // 常规算子
21 : constexpr u32 NOTIFY_A2A = 1; // alltoall
22 : constexpr u32 NOTIFY_ALIGN = 2; // atomic write，需要8字节对齐的notify，暂不支持alltoall
23 :
24 : const std::string HCCL_ALLTOALL = "ALLTOALL";
25 :
26 : // notify的offset对齐标准
27 : constexpr u32 NOTIFY_OFFSET_ALIGN_EIGHT = 8;
28 : constexpr u32 NOTIFY_OFFSET_ALIGN_FOUR = 4;
29 :
30 524 : NotifyPoolImpl::NotifyPoolImpl(const s32 devicePhyId)
31 524 : : devicePhyId_(devicePhyId)
32 : {
33 524 : }
34 :
35 524 : NotifyPoolImpl::~NotifyPoolImpl()
36 : {
37 524 : HcclResult ret = Destroy();
38 524 : if (ret != HCCL_SUCCESS) {
39 0 : HCCL_WARNING("destroy NotifyPoolImpl resources failed, ret[%d]", ret);
40 : }
41 524 : }
42 :
43 524 : HcclResult NotifyPoolImpl::Init()
44 : {
45 : #ifndef HCCD
46 524 : CHK_RET(SalGetBareTgid(&pid_)); // 当前进程id
47 : #else
48 : s32 psPid = 0;
49 : hrtDrvDeviceGetBareTgid(psPid);
50 : pid_ = psPid;
51 : #endif
52 524 : return HCCL_SUCCESS;
53 : }
54 :
55 524 : HcclResult NotifyPoolImpl::Destroy()
56 : {
57 524 : HCCL_INFO("NotifyPoolImpl Destroy.");
58 2096 : for (u32 index = 0; index < NOTIFY_RES_MGR_NUM; ++index) {
59 1572 : CHK_RET(DestroyRegisteredOpMap(index));
60 1572 : CHK_RET(DestroyNotifyPoolIPCAsignedMap(index));
61 1572 : CHK_RET(DestroyNotifyPoolDevIPCAsignedMap(index));
62 1572 : CHK_RET(DestroyNotifyPoolNoIPCAsignedMap(index));
63 1572 : CHK_RET(DestroyNotifyPoolDevNoIPCAsignedMap(index));
64 : }
65 524 : HCCL_INFO("NotifyPoolImpl Destroy success.");
66 524 : return HCCL_SUCCESS;
67 : }
68 :
69 1 : HcclResult NotifyPoolImpl::CreateNotify(std::shared_ptr<LocalIpcNotify> &localNotify, const s32 localDeviceId,
70 : const s32 remoteDeviceId, const NotifyLoadType type, bool withIpc, s64 recvId, u32 offsetAlignSize)
71 : {
72 1 : std::vector<std::shared_ptr<LocalIpcNotify>> tmpNotifys;
73 1 : HcclResult ret = HCCL_SUCCESS;
74 1 : bool errorFlag = false;
75 : while (true) {
76 1 : std::shared_ptr<LocalIpcNotify> tmpNotify;
77 1 : EXCEPTION_CATCH((tmpNotify = std::make_shared<LocalIpcNotify>()), errorFlag = true);
78 1 : CHK_PRT_BREAK(!(tmpNotify) || errorFlag, HCCL_ERROR("[NotifyPoolImpl][CreateNotify]create notify failed, "
79 : "errorFlag[%d]", errorFlag), errorFlag = true);
80 :
81 1 : ret = tmpNotify->Init(localDeviceId, remoteDeviceId, type);
82 1 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[NotifyPoolImpl][CreateNotify]localNotify init failed, "
83 : "ret[%d]", ret), errorFlag = true);
84 :
85 : // 申请到的notify offset不满足要求，再次申请
86 1 : bool isAligned = false;
87 1 : ret = IsNotifyOffsetAligned(tmpNotify, offsetAlignSize, isAligned);
88 1 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[NotifyPoolImpl][CreateNotify]IsNotifyOffsetAligned failed, "
89 : "ret[%d]", ret), errorFlag = true);
90 1 : if (!isAligned) {
91 0 : tmpNotifys.push_back(tmpNotify);
92 0 : HCCL_DEBUG("CreateNotify id[%u] offset[%llu] is not support atomic write, create again",
93 : tmpNotify->notifyId_, tmpNotify->offset);
94 0 : continue;
95 : }
96 :
97 1 : localNotify = tmpNotify;
98 1 : HCCL_DEBUG("withIpc[%d], Is310PDevice[%d] recvId[%lld]", withIpc, Is310PDevice(), recvId);
99 1 : if (withIpc || Is310PDevice()) {
100 0 : ret = localNotify->SetIpc();
101 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[NotifyPoolImpl][CreateNotify]localNotify set ipc failed, "
102 : "ret[%d]", ret), errorFlag = true);
103 : }
104 :
105 1 : if (withIpc) {
106 0 : HCCL_DEBUG("withIpc[%d], Is310PDevice[%d] recvId[%lld]", withIpc, Is310PDevice(), recvId);
107 0 : ret = localNotify->Grant(recvId);
108 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[NotifyPoolImpl][CreateNotify]localNotify grant failed, "
109 : "ret[%d]", ret), errorFlag = true);
110 : }
111 1 : break;
112 1 : }
113 :
114 1 : tmpNotifys.clear(); // 释放不满足要求的notify
115 1 : if (errorFlag) {
116 0 : HCCL_ERROR("[NotifyPoolImpl][CreateNotify]localNotify create failed ,ret[%d]", ret);
117 0 : localNotify = nullptr;
118 0 : return ret;
119 : }
120 :
121 1 : return HCCL_SUCCESS;
122 1 : }
123 :
124 1572 : HcclResult NotifyPoolImpl::DestroyNotifyPoolIPCAsignedMap(u32 index)
125 : {
126 1572 : std::unique_lock<std::mutex> lock(notifyResMgr_[index].notifyPoolIPCAsignedMutex);
127 1572 : auto ¬ifyPoolIPCAsignedMap = notifyResMgr_[index].notifyPoolIPCAsignedMap;
128 1572 : for (auto iter = notifyPoolIPCAsignedMap.begin(); iter != notifyPoolIPCAsignedMap.end(); iter++) {
129 0 : for (auto &it : iter->second) {
130 0 : if (DestroyNotify(it) != HCCL_SUCCESS) {
131 0 : HCCL_ERROR("[%s]DestroyNotify failed", __func__);
132 : }
133 : }
134 : }
135 1572 : notifyPoolIPCAsignedMap.clear();
136 1572 : HCCL_INFO("%s index[%u] success", __func__, index);
137 1572 : return HCCL_SUCCESS;
138 1572 : }
139 :
140 1572 : HcclResult NotifyPoolImpl::DestroyNotifyPoolDevIPCAsignedMap(u32 index)
141 : {
142 1572 : std::unique_lock<std::mutex> lock(notifyResMgr_[index].notifyPoolIPCAsignedMutex);
143 1572 : auto ¬ifyPoolDevIPCAsignedMap = notifyResMgr_[index].notifyPoolDevIPCAsignedMap;
144 1572 : for (auto iter = notifyPoolDevIPCAsignedMap.begin(); iter != notifyPoolDevIPCAsignedMap.end(); iter++) {
145 0 : for (auto &it : iter->second) {
146 0 : if (DestroyNotify(it) != HCCL_SUCCESS) {
147 0 : HCCL_ERROR("[%s]DestroyNotify failed", __func__);
148 : }
149 : }
150 : }
151 1572 : notifyPoolDevIPCAsignedMap.clear();
152 1572 : HCCL_INFO("%s index[%u] success", __func__, index);
153 1572 : return HCCL_SUCCESS;
154 1572 : }
155 :
156 1572 : HcclResult NotifyPoolImpl::DestroyNotifyPoolNoIPCAsignedMap(u32 index)
157 : {
158 1572 : std::unique_lock<std::mutex> lock(notifyResMgr_[index].notifyPoolNoIPCAsignedMutex);
159 1572 : auto ¬ifyPoolNoIPCAsignedMap = notifyResMgr_[index].notifyPoolNoIPCAsignedMap;
160 1572 : for (auto iter = notifyPoolNoIPCAsignedMap.begin(); iter != notifyPoolNoIPCAsignedMap.end(); iter++) {
161 0 : for (auto &it : iter->second) {
162 0 : if (DestroyNotify(it) != HCCL_SUCCESS) {
163 0 : HCCL_ERROR("[%s]DestroyNotify failed", __func__);
164 : }
165 : }
166 : }
167 1572 : notifyPoolNoIPCAsignedMap.clear();
168 1572 : HCCL_INFO("%s index[%u] success", __func__, index);
169 1572 : return HCCL_SUCCESS;
170 1572 : }
171 :
172 1572 : HcclResult NotifyPoolImpl::DestroyNotifyPoolDevNoIPCAsignedMap(u32 index)
173 : {
174 1572 : std::unique_lock<std::mutex> lock(notifyResMgr_[index].notifyPoolNoIPCAsignedMutex);
175 1572 : auto ¬ifyPoolDevNoIPCAsignedMap = notifyResMgr_[index].notifyPoolDevNoIPCAsignedMap;
176 1573 : for (auto iter = notifyPoolDevNoIPCAsignedMap.begin(); iter != notifyPoolDevNoIPCAsignedMap.end(); iter++) {
177 2 : for (auto &it : iter->second) {
178 1 : if (DestroyNotify(it) != HCCL_SUCCESS) {
179 0 : HCCL_ERROR("[%s]DestroyNotify failed", __func__);
180 : }
181 : }
182 : }
183 1572 : notifyPoolDevNoIPCAsignedMap.clear();
184 1572 : HCCL_INFO("%s index[%u] success", __func__, index);
185 1572 : return HCCL_SUCCESS;
186 1572 : }
187 :
188 1 : HcclResult NotifyPoolImpl::DestroyNotify(std::shared_ptr<LocalIpcNotify> &localNotify)
189 : {
190 1 : CHK_PTR_NULL(localNotify);
191 1 : CHK_RET(localNotify->Destroy());
192 1 : return HCCL_SUCCESS;
193 : }
194 212 : HcclResult NotifyPoolImpl::RegisterOpMap(
195 : const std::string &tag, std::map<std::string, NotifyPoolIndicator> ®isteredOpMap)
196 : {
197 212 : auto iterTag = registeredOpMap.find(tag);
198 209 : if (iterTag == registeredOpMap.end()) {
199 204 : NotifyPoolIndicator indicator;
200 204 : registeredOpMap.insert(std::make_pair(tag, indicator));
201 205 : } else {
202 0 : HCCL_ERROR(
203 : "[NotifyPoolImpl][RegisterOp]register op to the notify pool failed, tag[%s] has existed", tag.c_str());
204 0 : return HCCL_E_PARA;
205 : }
206 202 : return HCCL_SUCCESS;
207 : }
208 :
209 56 : HcclResult NotifyPoolImpl::RegisterOp(const std::string &tag)
210 : {
211 56 : std::string upTag = tag;
212 56 : std::transform(upTag.begin(), upTag.end(), upTag.begin(), ::toupper);
213 55 : bool hasAlltoAll = upTag.find(HCCL_ALLTOALL) != std::string::npos;
214 56 : HCCL_INFO("RegisterOp hasAlltoAll[%d]", hasAlltoAll);
215 56 : u32 notifyResIdx = hasAlltoAll ? NOTIFY_A2A : NOTIFY_NORMAL;
216 56 : auto tagDev = "Dev_" + tag;
217 :
218 : /* 此处可能会与并发，加锁 */
219 56 : std::unique_lock<std::mutex> lock1(notifyResMgr_[notifyResIdx].registeredOpMapMutex);
220 56 : CHK_RET(RegisterOpMap(tag, notifyResMgr_[notifyResIdx].registeredOpMap));
221 50 : CHK_RET(RegisterOpMap(tagDev, notifyResMgr_[notifyResIdx].registeredOpMap));
222 51 : lock1.unlock();
223 :
224 : // ALIGN的资源池不支持alltoall算子
225 56 : if (!hasAlltoAll) {
226 56 : std::unique_lock<std::mutex> lock2(notifyResMgr_[NOTIFY_ALIGN].registeredOpMapMutex);
227 55 : CHK_RET(RegisterOpMap(tag, notifyResMgr_[NOTIFY_ALIGN].registeredOpMap));
228 52 : CHK_RET(RegisterOpMap(tagDev, notifyResMgr_[NOTIFY_ALIGN].registeredOpMap));
229 54 : lock2.unlock();
230 55 : }
231 55 : HCCL_INFO("register op[%s] to the notify pool success.", tag.c_str());
232 56 : return HCCL_SUCCESS;
233 56 : }
234 :
235 4 : HcclResult NotifyPoolImpl::UnregisterOpMap(const std::string &tag,
236 : std::map<std::string, NotifyPoolIndicator> ®isteredOpMap)
237 : {
238 4 : auto iterTag = registeredOpMap.find(tag);
239 4 : if (iterTag == registeredOpMap.end()) {
240 0 : HCCL_ERROR("[NotifyPoolImpl][UnregisterOp]unregister op from the notify pool failed, tag[%s] has unregistered",
241 : tag.c_str());
242 0 : return HCCL_E_PARA;
243 : } else {
244 4 : registeredOpMap.erase(tag);
245 : }
246 4 : return HCCL_SUCCESS;
247 : }
248 :
249 1 : HcclResult NotifyPoolImpl::UnregisterOp(const std::string &tag)
250 : {
251 1 : std::string upTag = tag;
252 1 : std::transform(upTag.begin(), upTag.end(), upTag.begin(), ::toupper);
253 1 : bool hasAlltoAll = upTag.find(HCCL_ALLTOALL) != std::string::npos;
254 1 : HCCL_INFO("UnregisterOp hasAlltoAll[%d]", hasAlltoAll);
255 1 : u32 notifyResIdx = hasAlltoAll ? NOTIFY_A2A : NOTIFY_NORMAL;
256 1 : auto tagDev = "Dev_" + tag;
257 :
258 : /* 此处可能会与并发，加锁 */
259 1 : std::unique_lock<std::mutex> lock1(notifyResMgr_[notifyResIdx].registeredOpMapMutex);
260 1 : CHK_RET(UnregisterOpMap(tag, notifyResMgr_[notifyResIdx].registeredOpMap));
261 1 : CHK_RET(UnregisterOpMap(tagDev, notifyResMgr_[notifyResIdx].registeredOpMap));
262 1 : lock1.unlock();
263 :
264 1 : if (!hasAlltoAll) {
265 1 : std::unique_lock<std::mutex> lock2(notifyResMgr_[NOTIFY_ALIGN].registeredOpMapMutex);
266 1 : CHK_RET(UnregisterOpMap(tag, notifyResMgr_[NOTIFY_ALIGN].registeredOpMap));
267 1 : CHK_RET(UnregisterOpMap(tagDev, notifyResMgr_[NOTIFY_ALIGN].registeredOpMap));
268 1 : lock2.unlock();
269 1 : }
270 1 : HCCL_INFO("unregister op[%s] from the notify pool success.", tag.c_str());
271 1 : return HCCL_SUCCESS;
272 1 : }
273 :
274 1572 : HcclResult NotifyPoolImpl::DestroyRegisteredOpMap(u32 index)
275 : {
276 1572 : std::unique_lock<std::mutex> lock(notifyResMgr_[index].registeredOpMapMutex);
277 1572 : notifyResMgr_[index].registeredOpMap.clear();
278 1572 : HCCL_INFO("%s index[%u] success", __func__, index);
279 1572 : return HCCL_SUCCESS;
280 1572 : }
281 :
282 1 : HcclResult NotifyPoolImpl::IsNotifyOffsetAligned(std::shared_ptr<LocalIpcNotify> &localNotify, u32 offsetAlignSize,
283 : bool &isAligned)
284 : {
285 1 : if (offsetAlignSize == INVALID_UINT) {
286 1 : isAligned = true;
287 0 : } else if (offsetAlignSize == NOTIFY_OFFSET_ALIGN_EIGHT) { // offset按照8byte对齐
288 0 : isAligned = (localNotify->offset % NOTIFY_OFFSET_ALIGN_EIGHT == 0);
289 0 : } else if (offsetAlignSize == NOTIFY_OFFSET_ALIGN_FOUR) { // offset按照4byte对齐
290 0 : isAligned = (localNotify->offset % NOTIFY_OFFSET_ALIGN_FOUR == 0);
291 : } else {
292 0 : HCCL_ERROR("IsNotifyOffsetAligned offsetAlignSize[%u] is invalid, only support 4 or 8", offsetAlignSize);
293 0 : return HCCL_E_PARA;
294 : }
295 1 : return HCCL_SUCCESS;
296 : }
297 :
298 0 : HcclResult NotifyPoolImpl::AllocIpc(const std::string &tag, s64 remote, s64 recvId,
299 : const s32 localDeviceId, const s32 remoteDeviceId, const NotifyLoadType type,
300 : std::shared_ptr<LocalIpcNotify> &localNotify, std::mutex ®isteredOpMapMutex,
301 : std::map<std::string, NotifyPoolIndicator> ®isteredOpMap, std::mutex ¬ifyPoolIPCAsignedMapMutex,
302 : std::map<s64, NotifyPoolIPCSub> ¬ifyPoolIPCAsignedMap, u32 offsetAlignSize)
303 : {
304 : /* 此处可能会与并发，加锁 */
305 0 : std::unique_lock<std::mutex> lock(registeredOpMapMutex);
306 0 : auto iterTag = registeredOpMap.find(tag);
307 0 : CHK_PRT_RET(iterTag == registeredOpMap.end(), HCCL_ERROR("[NotifyPoolImpl][Alloc]tag[%s] is not registered.",
308 : tag.c_str()), HCCL_E_PARA);
309 0 : auto iterIdx = iterTag->second.notifyPoolIPC.find(remote);
310 0 : if (iterIdx == iterTag->second.notifyPoolIPC.end()) {
311 0 : iterTag->second.notifyPoolIPC.insert({remote, 0});
312 0 : iterIdx = iterTag->second.notifyPoolIPC.find(remote);
313 0 : CHK_PRT_RET(iterIdx == iterTag->second.notifyPoolIPC.end(), HCCL_ERROR("[NotifyPoolImpl][Alloc]remote[%d] "
314 : "is not found.", remote), HCCL_E_PARA);
315 : }
316 :
317 0 : std::unique_lock<std::mutex> lockIPC(notifyPoolIPCAsignedMapMutex);
318 0 : auto iterRemoteDev = notifyPoolIPCAsignedMap.find(remote);
319 0 : if (iterRemoteDev != notifyPoolIPCAsignedMap.end()) {
320 : // 从资源池中遍历获取offset按照要求对齐的notify
321 0 : while (iterIdx->second < iterRemoteDev->second.size()) {
322 0 : bool isAligned = false;
323 0 : CHK_RET(IsNotifyOffsetAligned(iterRemoteDev->second[iterIdx->second], offsetAlignSize, isAligned));
324 0 : if (isAligned) {
325 0 : break; // notify满足对齐要求，退出循环
326 : }
327 0 : iterIdx->second++; // 继续遍历
328 : }
329 :
330 0 : if (iterIdx->second >= iterRemoteDev->second.size()) {
331 0 : CHK_RET(CreateNotify(localNotify, localDeviceId, remoteDeviceId, type, true, recvId, offsetAlignSize));
332 :
333 0 : iterRemoteDev->second.push_back(localNotify);
334 0 : HCCL_INFO("create one notify in notify pool(ipc):tag[%s] remote[%d] total[%zu].", tag.c_str(), remote, \
335 : iterRemoteDev->second.size());
336 : } else {
337 0 : localNotify = iterRemoteDev->second[iterIdx->second];
338 0 : CHK_SMART_PTR_NULL(localNotify);
339 0 : CHK_RET(localNotify->Grant(recvId));
340 0 : HCCL_INFO("create one notify in notify pool(ipc):tag[%s] remote[%d] total[%zu].", tag.c_str(), remote, \
341 : iterRemoteDev->second.size());
342 : }
343 : } else {
344 0 : CHK_RET(CreateNotify(localNotify, localDeviceId, remoteDeviceId, type, true, recvId, offsetAlignSize));
345 :
346 0 : NotifyPoolIPCSub tmpVec{{localNotify}};
347 0 : notifyPoolIPCAsignedMap.insert(std::make_pair(remote, tmpVec));
348 0 : HCCL_INFO("create one notify in notify pool(ipc):tag[%s] remote[%d] total[%zu].", tag.c_str(), remote, \
349 : tmpVec.size());
350 0 : }
351 0 : iterIdx->second++;
352 :
353 0 : HCCL_INFO("notify pool ipc alloc: tag[%s] remote[%d] used[%u]", tag.c_str(), remote, \
354 : iterIdx->second);
355 0 : return HCCL_SUCCESS;
356 0 : }
357 :
358 1 : u32 NotifyPoolImpl::GetNotifyResIdx(const std::string &tag, u32 offsetAlignSize)
359 : {
360 1 : std::string upTag = tag;
361 1 : std::transform(upTag.begin(), upTag.end(), upTag.begin(), ::toupper);
362 1 : bool hasAlltoAll = upTag.find(HCCL_ALLTOALL) != std::string::npos;
363 1 : u32 res = 0;
364 : // alltoall算子不需要使用atomic write，因此固定使用A2A的资源池
365 1 : if (hasAlltoAll) {
366 0 : res = NOTIFY_A2A;
367 1 : } else if (offsetAlignSize == NOTIFY_OFFSET_ALIGN_EIGHT) {
368 0 : res = NOTIFY_ALIGN;
369 : } else {
370 1 : res = NOTIFY_NORMAL;
371 : }
372 1 : HCCL_INFO("%s tag[%s], offsetAlignSize[%u], res[%u]", __func__, tag.c_str(), offsetAlignSize, res);
373 1 : return res;
374 1 : }
375 :
376 0 : HcclResult NotifyPoolImpl::Alloc(const std::string &tag, s64 remote, s64 recvId, const s32 localDeviceId,
377 : const s32 remoteDeviceId, const NotifyLoadType type, std::shared_ptr<LocalIpcNotify> &localNotify,
378 : u32 offsetAlignSize)
379 : {
380 0 : u32 notifyResIdx = GetNotifyResIdx(tag, offsetAlignSize);
381 0 : HCCL_INFO("[NotifyPoolImpl][Alloc]notifyResIdx[%u], tag[%s], remote[%lld], recvId[%lld], "\
382 : "localDeviceId[%d], remoteDeviceId[%d], type[%d], offsetAlignSize[%u]",
383 : notifyResIdx, tag.c_str(), remote, recvId, localDeviceId, remoteDeviceId, type, offsetAlignSize);
384 :
385 0 : std::mutex ®isteredOpMapMutex = notifyResMgr_[notifyResIdx].registeredOpMapMutex;
386 0 : std::mutex ¬ifyPoolIPCAsignedMapMutex = notifyResMgr_[notifyResIdx].notifyPoolIPCAsignedMutex;
387 0 : std::map<std::string, NotifyPoolIndicator> ®isteredOpMap = notifyResMgr_[notifyResIdx].registeredOpMap;
388 0 : if (type == NotifyLoadType::HOST_NOTIFY) {
389 0 : std::map<s64, NotifyPoolIPCSub> ¬ifyPoolIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolIPCAsignedMap;
390 0 : CHK_RET(AllocIpc(tag, remote, recvId, localDeviceId, remoteDeviceId, type, localNotify,
391 : registeredOpMapMutex, registeredOpMap, notifyPoolIPCAsignedMapMutex, notifyPoolIPCAsignedMap,
392 : offsetAlignSize));
393 0 : } else if (type == NotifyLoadType::DEVICE_NOTIFY) { // 申请device上使用的notify资源
394 0 : std::map<s64, NotifyPoolIPCSub> ¬ifyPoolIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolDevIPCAsignedMap;
395 0 : auto tagDev = "Dev_" + tag;
396 0 : CHK_RET(AllocIpc(tagDev, remote, recvId, localDeviceId, remoteDeviceId, type, localNotify,
397 : registeredOpMapMutex, registeredOpMap, notifyPoolIPCAsignedMapMutex, notifyPoolIPCAsignedMap,
398 : offsetAlignSize));
399 0 : }
400 0 : return HCCL_SUCCESS;
401 : }
402 :
403 1 : HcclResult NotifyPoolImpl::AllocNoIpc(const std::string &tag, s64 remote, const s32 deviceId,
404 : const NotifyLoadType type, std::shared_ptr<LocalIpcNotify> &localNotify, std::mutex ®isteredOpMapMutex,
405 : std::map<std::string, NotifyPoolIndicator> ®isteredOpMap, std::mutex ¬ifyPoolNoIPCAsignedMapMutex,
406 : std::map<s64, NotifyPoolNoIPCSub> ¬ifyPoolNoIPCAsignedMap, u32 offsetAlignSize)
407 : {
408 : /* 此处可能会与并发，加锁 */
409 1 : std::unique_lock<std::mutex> lock(registeredOpMapMutex);
410 :
411 1 : auto iterTag = registeredOpMap.find(tag);
412 1 : CHK_PRT_RET(iterTag == registeredOpMap.end(), HCCL_ERROR("[NotifyPool][Alloc]tag[%s] is not registered.",
413 : tag.c_str()), HCCL_E_PARA);
414 :
415 1 : auto iterIdx = iterTag->second.notifyPoolNoIPC.find(remote);
416 1 : if (iterIdx == iterTag->second.notifyPoolNoIPC.end()) {
417 1 : iterTag->second.notifyPoolNoIPC.insert({remote, 0});
418 1 : iterIdx = iterTag->second.notifyPoolNoIPC.find(remote);
419 1 : CHK_PRT_RET(iterIdx == iterTag->second.notifyPoolNoIPC.end(), HCCL_ERROR("[NotifyPool][Alloc]remote[%lld] is "\
420 : "not found.", remote), HCCL_E_PARA);
421 : }
422 :
423 1 : std::unique_lock<std::mutex> lockNoIPC(notifyPoolNoIPCAsignedMapMutex);
424 1 : auto iterRemoteDev = notifyPoolNoIPCAsignedMap.find(remote);
425 1 : if (iterRemoteDev != notifyPoolNoIPCAsignedMap.end()) {
426 : // 从资源池中获取offset按照要求对齐的notify
427 0 : while (iterIdx->second < iterRemoteDev->second.size()) {
428 0 : bool isAligned = false;
429 0 : CHK_RET(IsNotifyOffsetAligned(iterRemoteDev->second[iterIdx->second], offsetAlignSize, isAligned));
430 0 : if (isAligned) {
431 0 : break; // notify满足对齐要求，退出循环
432 : }
433 0 : iterIdx->second++; // 继续遍历
434 : }
435 :
436 0 : if (iterIdx->second >= iterRemoteDev->second.size()) {
437 0 : CHK_RET(CreateNotify(localNotify, deviceId, deviceId, type, false, -1, offsetAlignSize));
438 0 : iterRemoteDev->second.push_back(localNotify);
439 0 : HCCL_INFO("create one notify in notify pool(no ipc):tag[%s] remote[%lld] total[%zu].",
440 : tag.c_str(), remote, iterRemoteDev->second.size());
441 : } else {
442 0 : localNotify = iterRemoteDev->second[iterIdx->second];
443 0 : CHK_SMART_PTR_NULL(localNotify);
444 : }
445 : } else {
446 1 : CHK_RET(CreateNotify(localNotify, deviceId, deviceId, type, false, -1, offsetAlignSize));
447 :
448 3 : NotifyPoolNoIPCSub tmpVec{localNotify};
449 1 : notifyPoolNoIPCAsignedMap.insert(std::make_pair(remote, tmpVec));
450 1 : HCCL_INFO("create one notify in notify pool(no ipc):tag[%s] remote[%lld] total[%zu].", tag.c_str(), remote,
451 : tmpVec.size());
452 1 : }
453 1 : iterIdx->second++;
454 :
455 1 : HCCL_INFO("notify pool no ipc alloc: tag[%s] remote[%lld] used[%u]", tag.c_str(), remote,
456 : iterIdx->second);
457 1 : return HCCL_SUCCESS;
458 2 : }
459 :
460 1 : HcclResult NotifyPoolImpl::Alloc(const std::string &tag, s64 remote, const s32 deviceId, const NotifyLoadType type,
461 : std::shared_ptr<LocalIpcNotify> &localNotify, u32 offsetAlignSize)
462 : {
463 1 : u32 notifyResIdx = GetNotifyResIdx(tag, offsetAlignSize);
464 1 : HCCL_INFO("[NotifyPoolImpl][Alloc]notifyResIdx[%u], tag[%s], remote[%lld], deviceId[%d], type[%d], "\
465 : "offsetAlignSize[%u]", notifyResIdx, tag.c_str(), remote, deviceId, type, offsetAlignSize);
466 :
467 1 : std::mutex ®isteredOpMapMutex = notifyResMgr_[notifyResIdx].registeredOpMapMutex;
468 1 : std::mutex ¬ifyPoolNoIPCAsignedMapMutex = notifyResMgr_[notifyResIdx].notifyPoolNoIPCAsignedMutex;
469 1 : std::map<std::string, NotifyPoolIndicator> ®isteredOpMap = notifyResMgr_[notifyResIdx].registeredOpMap;
470 1 : if (type == NotifyLoadType::HOST_NOTIFY) {
471 0 : std::map<s64, NotifyPoolNoIPCSub> ¬ifyPoolNoIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolNoIPCAsignedMap;
472 0 : CHK_RET(AllocNoIpc(tag, remote, deviceId, type, localNotify, registeredOpMapMutex, registeredOpMap,
473 : notifyPoolNoIPCAsignedMapMutex, notifyPoolNoIPCAsignedMap, offsetAlignSize));
474 1 : } else if (type == NotifyLoadType::DEVICE_NOTIFY) { // 申请device上使用的notify资源
475 1 : std::map<s64, NotifyPoolNoIPCSub> ¬ifyPoolNoIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolDevNoIPCAsignedMap;
476 1 : auto tagDev = "Dev_" + tag;
477 1 : CHK_RET(AllocNoIpc(tagDev, remote, deviceId, type, localNotify, registeredOpMapMutex, registeredOpMap,
478 : notifyPoolNoIPCAsignedMapMutex, notifyPoolNoIPCAsignedMap, offsetAlignSize));
479 1 : }
480 1 : return HCCL_SUCCESS;
481 : }
482 :
483 1 : HcclResult NotifyPoolImpl::Alloc(const std::string &tag, const RemoteRankInfo &info, const NotifyLoadType type,
484 : std::shared_ptr<LocalIpcNotify> &localNotify, u32 offsetAlignSize)
485 : {
486 1 : HCCL_DEBUG("[Alloc][IpcNotify]localPid[%016llx], remotePid[%016llx], localDeviceId[%d], remoteDeviceId[%d], "
487 : "remoteSdid[%x], offsetAlignSize[%u]", pid_, info.remotePid, devicePhyId_, info.remoteDeviceId, info.remoteSdid,
488 : offsetAlignSize);
489 : // 主从流下标为-1、rdma notify下标是remoteRank
490 1 : if (pid_ == info.remotePid && devicePhyId_ == info.remoteDeviceId &&
491 1 : info.remoteSdid == INVALID_INT) {
492 1 : CHK_RET(Alloc(tag, info.remoteRank, info.remoteDeviceId, type, localNotify, offsetAlignSize));
493 1 : } else {
494 : // 统一使用remoteRank作为notifyPoolMap下标区分
495 0 : s64 remoteRank = static_cast<s64>(info.remoteRank);
496 : // 将(s32)SDID和(s32)Pid拼接成s64作为标志位, 高32位为SDID, 低32位为pid
497 0 : s64 recvId =
498 0 : ((static_cast<s64>(info.remoteSdid) & 0xFFFFFFFF) << 32) | (static_cast<s64>(info.remotePid) & 0xFFFFFFFF);
499 0 : HCCL_INFO("[Alloc][IpcNotify]recvSdid[%016llx], recvPid[%016llx], remoteRank[%u]",
500 : info.remoteSdid, info.remotePid, info.remoteRank);
501 0 : CHK_RET(Alloc(tag, remoteRank, recvId, devicePhyId_, info.remoteDeviceId, type, localNotify, offsetAlignSize));
502 : }
503 :
504 1 : return HCCL_SUCCESS;
505 : }
506 :
507 0 : HcclResult NotifyPoolImpl::ResetNotifyForDestRank(s64 destRank) {
508 : // send/recv场景需要单点重置和对端的notify
509 0 : std::vector<u32> notifyResIdxs = {NOTIFY_NORMAL, NOTIFY_ALIGN};
510 0 : for (u32 notifyResIdx : notifyResIdxs) {
511 0 : std::unique_lock<std::mutex> lockIPC(notifyResMgr_[notifyResIdx].notifyPoolIPCAsignedMutex);
512 0 : const auto ¬ifyPoolDevIPCAsignedIt = notifyResMgr_[notifyResIdx].notifyPoolDevIPCAsignedMap.find(destRank);
513 0 : if (notifyPoolDevIPCAsignedIt == notifyResMgr_[notifyResIdx].notifyPoolDevIPCAsignedMap.end()) {
514 0 : HCCL_RUN_INFO("[ResetNotify]remoteRank[%d] is not in notifyPoolDevIPCAsignedMap, notifyResIdx[%u]",
515 : destRank, notifyResIdx);
516 : } else {
517 0 : HCCL_RUN_INFO("[ResetNotify]reset notifyPoolDevIPCAsignedIt remoteRank[%d], notifyResIdx[%u]",
518 : destRank, notifyResIdx);
519 0 : for (auto &it : notifyPoolDevIPCAsignedIt->second) {
520 0 : CHK_RET(hrtNotifyReset(it->ptr()));
521 : }
522 : }
523 0 : lockIPC.unlock();
524 :
525 0 : std::unique_lock<std::mutex> lockNoIPC(notifyResMgr_[notifyResIdx].notifyPoolNoIPCAsignedMutex);
526 0 : const auto ¬ifyPoolDevNoIPCAsignedIt = notifyResMgr_[notifyResIdx].notifyPoolDevNoIPCAsignedMap.find(destRank);
527 0 : if (notifyPoolDevNoIPCAsignedIt == notifyResMgr_[notifyResIdx].notifyPoolDevNoIPCAsignedMap.end()) {
528 0 : HCCL_RUN_INFO("[ResetNotify]remoteRank[%d] is not in notifyPoolDevNoIPCAsignedMap, notifyResIdx[%u]",
529 : destRank, notifyResIdx);
530 : } else {
531 0 : HCCL_RUN_INFO("[ResetNotify]reset notifyPoolDevNoIPCAsignedIt remoteRank[%d], notifyResIdx[%u]",
532 : destRank, notifyResIdx);
533 0 : for (auto &it : notifyPoolDevNoIPCAsignedIt->second) {
534 0 : CHK_RET(hrtNotifyReset(it->ptr()));
535 : }
536 : }
537 0 : lockNoIPC.unlock();
538 0 : }
539 0 : return HCCL_SUCCESS;
540 0 : }
541 :
542 0 : HcclResult NotifyPoolImpl::ResetNotify()
543 : {
544 0 : HCCL_DEBUG("NotifyPoolImpl ResetNotify");
545 0 : for (u32 notifyResIdx = 0; notifyResIdx < NOTIFY_RES_MGR_NUM; ++notifyResIdx) {
546 0 : HCCL_INFO("NotifyPoolImpl ResetNotify, notifyResIdx[%u]", notifyResIdx);
547 0 : std::unique_lock<std::mutex> lockIPC(notifyResMgr_[notifyResIdx].notifyPoolIPCAsignedMutex);
548 0 : auto ¬ifyPoolDevIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolDevIPCAsignedMap;
549 0 : for (auto iter = notifyPoolDevIPCAsignedMap.begin(); iter != notifyPoolDevIPCAsignedMap.end(); iter++) {
550 0 : for (auto &it : iter->second) {
551 0 : CHK_RET(hrtNotifyReset(it->ptr()));
552 : }
553 : }
554 0 : lockIPC.unlock();
555 :
556 0 : std::unique_lock<std::mutex> lockNoIPC(notifyResMgr_[notifyResIdx].notifyPoolNoIPCAsignedMutex);
557 0 : auto ¬ifyPoolDevNoIPCAsignedMap = notifyResMgr_[notifyResIdx].notifyPoolDevNoIPCAsignedMap;
558 0 : for (auto iter = notifyPoolDevNoIPCAsignedMap.begin(); iter != notifyPoolDevNoIPCAsignedMap.end(); iter++) {
559 0 : for (auto &it : iter->second) {
560 0 : CHK_RET(hrtNotifyReset(it->ptr()));
561 : }
562 : }
563 0 : lockNoIPC.unlock();
564 0 : }
565 0 : return HCCL_SUCCESS;
566 : }
567 : } // namespace hccl
|
