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 "hccl_dispatcher_ctx.h"
12 : #include "dispatcher_ctx.h"
13 : #include "dispatcher_aicpu_pub.h"
14 : #include <unordered_map>
15 : #include "adapter_rts_common.h"
16 :
17 : // 多个通信域能并发跑通信算子，一个通信域只绑定一个dispatch_ctx线程变量
18 : // 若不用通信域绑定线程变量，需要创建默认dispatch_ctx
19 : static std::unordered_map<std::string, DispatcherCtxPtr> g_ctx;
20 : std::mutex g_mtx; // 考虑已有的universal_map，或读写锁
21 : thread_local DispatcherCtxPtr gDispatcherCtx = nullptr;
22 :
23 1068 : bool FindDispatcherByCommId(DispatcherCtxPtr* ctx, const char* commId)
24 : {
25 1068 : if (commId == nullptr) {
26 0 : HCCL_ERROR("[%s] find dispatcher fail, commId is nullptr", __func__);
27 0 : return false;
28 : }
29 1068 : std::lock_guard<std::mutex> lock(g_mtx);
30 1069 : std::string commIdkey = std::string(commId);
31 1069 : auto it = g_ctx.find(commIdkey);
32 1069 : if (it != g_ctx.end()) {
33 538 : *ctx = it->second;
34 538 : HCCL_INFO("[%s] commIdkey[%s] has been bound with ctx[%p]", __func__, commIdkey.c_str(), *ctx);
35 538 : return true;
36 : }
37 531 : *ctx = nullptr;
38 531 : HCCL_WARNING("[%s] commIdkey[%s] not found, ctx return nullptr", __func__, commIdkey.c_str());
39 531 : return false;
40 1069 : }
41 :
42 418 : bool DeleteDispatcherByCommId(const char* commId)
43 : {
44 418 : if (commId == nullptr) {
45 0 : HCCL_ERROR("[%s] delete dispatcher fail, commId is nullptr", __func__);
46 0 : return false;
47 : }
48 418 : std::lock_guard<std::mutex> lock(g_mtx);
49 418 : std::string commIdkey = std::string(commId);
50 418 : auto it = g_ctx.find(commIdkey);
51 418 : if (it != g_ctx.end()) {
52 418 : HCCL_INFO("[%s] ctx[%p] has been bound by commId[%s]", __func__, it->second, commIdkey.c_str());
53 418 : g_ctx.erase(it);
54 418 : return true;
55 : }
56 0 : HCCL_WARNING("[%s] ctx has not been bound by commId[%s]", __func__, commIdkey.c_str());
57 0 : return false;
58 418 : }
59 :
60 422 : HcclResult BindDispatcherCtxWithComm(DispatcherCtxPtr ctx, const char* commId)
61 : {
62 422 : CHK_PTR_NULL(commId);
63 422 : std::lock_guard<std::mutex> lock(g_mtx);
64 422 : std::string commIdkey = std::string(commId);
65 422 : auto it = g_ctx.find(commIdkey);
66 422 : if (it != g_ctx.end()) {
67 3 : HCCL_WARNING("[%s] commId[%s] has been bound", __func__, commIdkey.c_str());
68 3 : return HCCL_E_PARA;
69 : }
70 419 : g_ctx[commIdkey] = ctx;
71 419 : HCCL_INFO("[%s] ctx[%p] bind commId[%s] success", __func__, ctx, commIdkey.c_str());
72 419 : return HCCL_SUCCESS;
73 422 : }
74 :
75 423 : HcclResult CreateDispatcherCtx(DispatcherCtxPtr* ctx, u32 devPhyId, const char* commId)
76 : {
77 423 : CHK_PTR_NULL(commId);
78 423 : CHK_PRT_RET(devPhyId == INVALID_UINT, HCCL_ERROR("[CreateCtx] devPhyId invalid"), HCCL_E_PARA);
79 422 : CHK_PTR_NULL(ctx);
80 422 : hccl::DispatcherCtx* Ctx_tmp = new (std::nothrow) hccl::DispatcherCtx(devPhyId);
81 422 : CHK_PTR_NULL(Ctx_tmp);
82 : // 创建ctx，内部创建dispatcher和notify pool实例 目前没有pool
83 422 : HcclResult ret = Ctx_tmp->Init();
84 422 : if (ret != HCCL_SUCCESS) {
85 0 : delete Ctx_tmp;
86 0 : HCCL_ERROR("[CreateCtx] CTX init fail");
87 0 : return ret;
88 : }
89 :
90 422 : ret = BindDispatcherCtxWithComm(Ctx_tmp, commId);
91 : // 如果存在，销毁创建的DispatcherCtx，返回存在的DispatcherCtx
92 422 : if (ret != HCCL_SUCCESS) {
93 3 : Ctx_tmp->Destroy();
94 3 : delete Ctx_tmp;
95 : // 查找已有ctx
96 2 : if (!FindDispatcherByCommId(ctx, commId)) {
97 0 : HCCL_ERROR("[CreateCtx] Bind fail AND no existing ctx for commId[%s]", commId);
98 0 : return HCCL_E_NOT_FOUND; // 明确返回错误，而非SUCCESS
99 : }
100 3 : gDispatcherCtx = *ctx;
101 3 : HCCL_WARNING("[CreateCtx] CTX bind fail, reuse existing ctx[%p] commId[%s]", *ctx, commId);
102 3 : return HCCL_SUCCESS;
103 : }
104 419 : *ctx = Ctx_tmp;
105 419 : gDispatcherCtx = Ctx_tmp;
106 419 : HCCL_INFO("[CreateCtx] CTX create success, ctx[%p] commId[%s]", *ctx, commId);
107 419 : return HCCL_SUCCESS;
108 : }
109 :
110 121 : bool DeleteCommIdByDispatcherCtx(DispatcherCtxPtr ctx)
111 : {
112 121 : std::lock_guard<std::mutex> lock(g_mtx);
113 121 : for (const auto& pair : g_ctx) {
114 1 : if (pair.second == ctx) {
115 1 : HCCL_INFO("[%s] ctx[%p] bound with commId[%s], delete it", __func__, ctx, pair.first.c_str());
116 1 : g_ctx.erase(pair.first);
117 1 : return true;
118 : }
119 : }
120 120 : HCCL_WARNING("[%s] no commId bound with ctx[%p]", __func__, ctx);
121 120 : return false;
122 121 : }
123 :
124 : // 调用方有可能通过SetDispatcherCtx设置默认线程变量
125 : // 传入commId是为了快速索引并释放g_ctx中的内容
126 : // 不分成两个接口，防止重复释放
127 539 : HcclResult DestroyDispatcherCtx(DispatcherCtxPtr ctx, const char* commId)
128 : {
129 : static std::mutex deleteMutex_;
130 539 : const std::lock_guard<std::mutex> lock(deleteMutex_);
131 539 : CHK_PTR_NULL(commId);
132 539 : CHK_PTR_NULL(ctx);
133 539 : HCCL_INFO("[DestroyCtx] Destroy Ctx, ctx[%p] commId[%s]", ctx, commId);
134 539 : if (gDispatcherCtx == ctx) {
135 420 : gDispatcherCtx = nullptr;
136 : } else {
137 119 : HCCL_WARNING("[DestroyCtx] gDispatcherCtx[%p] and ctx[%p] do not match.", gDispatcherCtx, ctx);
138 : }
139 :
140 539 : DispatcherCtxPtr otherCtx = nullptr;
141 : // 若找到commId绑定的dispatch_ctx，解除绑定后销毁dispatch_ctx
142 : // 若找不到commId绑定的dispatch_ctx，查找map中
143 539 : if (LIKELY(FindDispatcherByCommId(&otherCtx, commId))) {
144 418 : DeleteDispatcherByCommId(commId);
145 : } else {
146 121 : bool hasFound = DeleteCommIdByDispatcherCtx(ctx);
147 121 : if (!hasFound) {
148 120 : HCCL_WARNING("[DestroyCtx] ctx[%p] not found by commId[%s], it may have been destroyed", ctx, commId);
149 120 : return HCCL_SUCCESS;
150 : }
151 1 : HCCL_WARNING("[DestroyCtx] ctx[%p] not found by commId[%s], just destroy", ctx, commId);
152 : }
153 419 : hccl::DispatcherCtx* Ctx_tmp = reinterpret_cast<hccl::DispatcherCtx*>(ctx);
154 419 : HcclResult ret = Ctx_tmp->Destroy();
155 419 : if (ret != HCCL_SUCCESS) {
156 0 : HCCL_ERROR("[DestroyCtx] CTX Destroy fail");
157 : }
158 419 : delete Ctx_tmp;
159 419 : ctx = nullptr;
160 419 : return ret;
161 539 : }
162 :
163 : // 设置当前的线程变量dispatcherCtx，必须先调用CreateDispatcherCtx
164 : // 同一个通信域不可以切换dispatcherCtx
165 0 : HcclResult SetDispatcherCtx(const DispatcherCtxPtr ctx)
166 : {
167 0 : HCCL_INFO("[%s], param: ctx[%p]", __func__, ctx);
168 0 : CHK_PTR_NULL(ctx);
169 0 : gDispatcherCtx = ctx;
170 0 : return HCCL_SUCCESS;
171 : }
172 :
173 : // 获取当前设置的线程变量dispatcherCtx，必须先调用CreateDispatcherCtx
174 22 : DispatcherCtxPtr GetDispatcherCtx(const char* commId)
175 : {
176 22 : if (UNLIKELY(commId == nullptr)) {
177 0 : HCCL_ERROR("[%s] get dispatcher fail, commId is nullptr", __func__);
178 0 : return nullptr;
179 : }
180 22 : HCCL_DEBUG("[%s], commId[%s]", __func__, commId);
181 22 : if (LIKELY((gDispatcherCtx != nullptr))) {
182 21 : HCCL_INFO("[%s], gDispatcherCtx[%p] exist, commId[%s]", __func__, gDispatcherCtx, commId);
183 21 : return gDispatcherCtx;
184 : }
185 : DispatcherCtxPtr ctx;
186 1 : if (FindDispatcherByCommId(&ctx, commId)) {
187 0 : HCCL_INFO("[%s], ctx[%p] found in g_ctx, commId[%s]", __func__, ctx, commId);
188 0 : gDispatcherCtx = ctx;
189 0 : return ctx;
190 : }
191 1 : return nullptr;
192 : }
193 :
194 0 : HcclResult SetDispatcherCtxOpIdx(u32 opRingBufferIdx)
195 : {
196 0 : HCCL_INFO("%s start, %u", __func__, opRingBufferIdx);
197 0 : hccl::DispatcherCtx* ctx_temp = reinterpret_cast<hccl::DispatcherCtx*>(GetDispatcherCtx());
198 0 : CHK_PTR_NULL(ctx_temp);
199 0 : hccl::DispatcherAiCpu* dispatcherPtr = reinterpret_cast<hccl::DispatcherAiCpu*>(ctx_temp->GetDispatcher());
200 0 : CHK_PTR_NULL(dispatcherPtr);
201 0 : dispatcherPtr->SetOpRingBufferIdx(opRingBufferIdx);
202 0 : return HCCL_SUCCESS;
203 : }
204 :
205 7 : HcclResult AcquireDispatcherCtx(DispatcherCtxPtr* ctx, const char* commId)
206 : {
207 7 : CHK_PTR_NULL(commId);
208 7 : DispatcherCtxPtr ctxPtr = GetDispatcherCtx(commId);
209 7 : if (ctxPtr != nullptr) {
210 7 : *ctx = ctxPtr;
211 7 : HCCL_INFO("[AcquireCtx] CTX get success, ctx[%p] commId[%s]", *ctx, commId);
212 7 : return HCCL_SUCCESS;
213 : }
214 0 : s32 deviceLogicId = 0;
215 0 : CHK_RET(hrtGetDevice(&deviceLogicId));
216 0 : u32 devPhyId = INVALID_UINT;
217 0 : CHK_RET(hrtGetDevicePhyIdByIndex(deviceLogicId, devPhyId));
218 0 : CHK_PRT_RET(devPhyId == INVALID_UINT, HCCL_ERROR("[CreateCtx] devPhyId invalid"), HCCL_E_PARA);
219 0 : CHK_PTR_NULL(ctx);
220 0 : hccl::DispatcherCtx* Ctx_tmp = new (std::nothrow) hccl::DispatcherCtx(devPhyId);
221 0 : CHK_PTR_NULL(Ctx_tmp);
222 0 : HcclResult ret = Ctx_tmp->Init();
223 0 : if (ret != HCCL_SUCCESS) {
224 0 : delete Ctx_tmp;
225 0 : HCCL_ERROR("[AcquireCtx] CTX init fail");
226 0 : return ret;
227 : }
228 :
229 : // 如果存在，销毁创建的DispatcherCtx，返回存在的DispatcherCtx
230 0 : if (BindDispatcherCtxWithComm(Ctx_tmp, commId) != HCCL_SUCCESS) {
231 0 : Ctx_tmp->Destroy();
232 0 : delete Ctx_tmp;
233 : // 查找已有ctx
234 0 : if (!FindDispatcherByCommId(ctx, commId)) {
235 0 : HCCL_ERROR("[AcquireCtx] Bind fail AND no existing ctx for commId[%s]", commId);
236 0 : return HCCL_E_NOT_FOUND; // 明确返回错误，而非SUCCESS
237 : }
238 0 : gDispatcherCtx = *ctx;
239 0 : HCCL_WARNING("[AcquireCtx] CTX bind fail, reuse existing ctx[%p] commId[%s]", *ctx, commId);
240 0 : return HCCL_SUCCESS;
241 : }
242 :
243 0 : *ctx = Ctx_tmp;
244 0 : gDispatcherCtx = Ctx_tmp;
245 0 : HCCL_INFO("[AcquireCtx] CTX create success, ctx[%p] commId[%s]", *ctx, commId);
246 0 : return HCCL_SUCCESS;
247 : }
|
