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