Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "nic_plugin_dispatcher.h"
12 :
13 : #include <chrono>
14 : #include <cstddef>
15 : #include <vector>
16 :
17 : #include "env_config/env_config.h"
18 : #include "nic_plugin_manager.h"
19 : #include "param_check_pub.h"
20 :
21 : namespace hcomm {
22 : namespace {
23 99 : PluginEndpointCtx *GetPluginEndpointCtx(EndpointHandle handle, bool &handled)
24 : {
25 99 : handled = IsPluginEndpoint(handle);
26 99 : return handled ? PLUGIN_EP_CTX(handle) : nullptr;
27 : }
28 :
29 141 : PluginChannelCtx *GetPluginChannelCtx(ChannelHandle handle, bool &handled)
30 : {
31 141 : handled = IsPluginChannel(handle);
32 141 : return handled ? PLUGIN_CH_CTX(handle) : nullptr;
33 : }
34 :
35 2 : void DestroyCreatedPluginChannels(ChannelHandle *channels, uint32_t channelNum)
36 : {
37 2 : if (channels == nullptr) {
38 0 : return;
39 : }
40 4 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
41 2 : if (channels[idx] == 0) {
42 0 : continue;
43 : }
44 2 : bool handled = false;
45 2 : (void)PluginChannelDestroy(channels[idx], handled);
46 2 : channels[idx] = 0;
47 : }
48 : }
49 :
50 3 : HcommResult ConnectPluginChannels(ChannelHandle *channels, uint32_t channelNum)
51 : {
52 3 : CHK_PTR_NULL(channels);
53 3 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]",
54 : __func__, channelNum), HCCL_E_PARA);
55 :
56 3 : const auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
57 3 : const auto startTime = std::chrono::steady_clock::now();
58 3 : std::vector<int32_t> statusVec(channelNum, 1);
59 :
60 : while (true) {
61 9 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
62 5 : bool handled = false;
63 5 : HcommResult ret = PluginChannelGetStatus(channels[idx], &statusVec[idx], handled);
64 5 : if (ret != HCCL_SUCCESS) {
65 1 : HCCL_ERROR("[%s] PluginChannelGetStatus failed, ret[%d].", __func__, ret);
66 1 : return ret;
67 : }
68 4 : CHK_PRT_RET(!handled, HCCL_ERROR("[%s] channel[%u] is not plugin channel.", __func__, idx),
69 : HCCL_E_PARA);
70 : }
71 :
72 4 : bool allReady = true;
73 5 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
74 4 : if (statusVec[idx] != 0) {
75 3 : allReady = false;
76 3 : break;
77 : }
78 : }
79 4 : if (allReady) {
80 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
81 1 : return HCCL_SUCCESS;
82 : }
83 :
84 3 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
85 1 : HCCL_ERROR("[%s] plugin channel connect timeout.", __func__);
86 1 : return HCCL_E_TIMEOUT;
87 : }
88 2 : }
89 3 : }
90 :
91 : #define CHK_PLUGIN_ENDPOINT_OP(ctx, op) do { \
92 : CHK_PTR_NULL(ctx); \
93 : CHK_PTR_NULL((ctx)->ops); \
94 : if (!IsEndpointOpAvailable((ctx)->ops, offsetof(HcommNicEndpointOps, op), sizeof((ctx)->ops->op))) { \
95 : return UnsupportedPluginOp(__func__);\
96 : } \
97 : if ((ctx)->ops->op == nullptr) { \
98 : return UnsupportedPluginOp(__func__);\
99 : } \
100 : } while (0)
101 :
102 : #define CHK_PLUGIN_CHANNEL_OP(ctx, op) do { \
103 : CHK_PTR_NULL(ctx); \
104 : CHK_PTR_NULL((ctx)->ops); \
105 : if (!IsChannelOpAvailable((ctx)->ops, offsetof(HcommNicChannelOps, op), sizeof((ctx)->ops->op))) { \
106 : return UnsupportedPluginOp(__func__);\
107 : } \
108 : if ((ctx)->ops->op == nullptr) { \
109 : return UnsupportedPluginOp(__func__);\
110 : } \
111 : } while (0)
112 :
113 : #define CHK_PLUGIN_CHANNEL_OP_NAME(ctx, op, opName) do { \
114 : CHK_PTR_NULL(ctx); \
115 : CHK_PTR_NULL((ctx)->ops); \
116 : if (!IsChannelOpAvailable((ctx)->ops, offsetof(HcommNicChannelOps, op), sizeof((ctx)->ops->op))) { \
117 : return UnsupportedPluginOp(opName); \
118 : } \
119 : if ((ctx)->ops->op == nullptr) { \
120 : return UnsupportedPluginOp(opName); \
121 : } \
122 : } while (0)
123 :
124 : #define DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, op, args) do { \
125 : PluginEndpointCtx *ctx = GetPluginEndpointCtx((handle), (handled)); \
126 : if (!(handled)) { \
127 : return HCCL_SUCCESS; \
128 : } \
129 : CHK_PLUGIN_ENDPOINT_OP(ctx, op); \
130 : return ctx->ops->op args; \
131 : } while (0)
132 :
133 : #define DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, op, args) do { \
134 : PluginChannelCtx *ctx = GetPluginChannelCtx((handle), (handled)); \
135 : if (!(handled)) { \
136 : return HCCL_SUCCESS; \
137 : } \
138 : CHK_PLUGIN_CHANNEL_OP(ctx, op); \
139 : return ctx->ops->op args; \
140 : } while (0)
141 :
142 : #define DISPATCH_PLUGIN_CHANNEL_OP_NAME(handle, handled, op, opName, args) do { \
143 : PluginChannelCtx *ctx = GetPluginChannelCtx((handle), (handled)); \
144 : if (!(handled)) { \
145 : return HCCL_SUCCESS; \
146 : } \
147 : CHK_PLUGIN_CHANNEL_OP_NAME(ctx, op, opName); \
148 : return ctx->ops->op args; \
149 : } while (0)
150 :
151 4 : HcommResult UnsupportedPluginChannelOp(ChannelHandle handle, bool &handled, const char *opName)
152 : {
153 4 : handled = IsPluginChannel(handle);
154 4 : if (!handled) {
155 0 : return HCCL_SUCCESS;
156 : }
157 4 : return UnsupportedPluginOp(opName);
158 : }
159 :
160 2 : HcommResult UnsupportedPluginChannelReduceOp(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
161 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, bool &handled, const char *opName)
162 : {
163 : (void)thread;
164 : (void)dst;
165 : (void)src;
166 : (void)count;
167 : (void)dataType;
168 : (void)reduceOp;
169 2 : return UnsupportedPluginChannelOp(handle, handled, opName);
170 : }
171 :
172 10 : HcommResult DispatchPluginChannelWriteNbi(ChannelHandle handle, void *dst, const void *src, uint64_t len,
173 : bool &handled, const char *opName)
174 : {
175 10 : DISPATCH_PLUGIN_CHANNEL_OP_NAME(handle, handled, writeNbi, opName, (ctx->ctx, dst, src, len));
176 : }
177 :
178 8 : HcommResult DispatchPluginChannelWriteWithNotifyNbi(ChannelHandle handle, void *dst, const void *src, uint64_t len,
179 : uint32_t remoteNotifyIdx, bool &handled, const char *opName)
180 : {
181 8 : DISPATCH_PLUGIN_CHANNEL_OP_NAME(handle, handled, writeWithNotifyNbi, opName,
182 : (ctx->ctx, dst, src, len, remoteNotifyIdx));
183 : }
184 :
185 8 : HcommResult DispatchPluginChannelReadNbi(ChannelHandle handle, void *dst, const void *src, uint64_t len,
186 : bool &handled, const char *opName)
187 : {
188 8 : DISPATCH_PLUGIN_CHANNEL_OP_NAME(handle, handled, readNbi, opName, (ctx->ctx, dst, src, len));
189 : }
190 : } // namespace
191 :
192 156 : bool IsPluginEndpoint(EndpointHandle handle)
193 : {
194 156 : return IS_PLUGIN_HANDLE(handle);
195 : }
196 :
197 164 : bool IsPluginChannel(ChannelHandle handle)
198 : {
199 164 : return IS_PLUGIN_HANDLE(handle);
200 : }
201 :
202 65 : HcommResult PluginEndpointCreate(const EndpointDesc *endpoint, EndpointHandle *endpointHandle, bool &handled)
203 : {
204 65 : handled = false;
205 65 : CHK_PTR_NULL(endpoint);
206 65 : CHK_PTR_NULL(endpointHandle);
207 65 : if (endpoint->loc.locType != ENDPOINT_LOC_TYPE_HOST || FindHostNicPlugin(endpoint->protocol) == nullptr) {
208 64 : return HCCL_SUCCESS;
209 : }
210 1 : handled = true;
211 1 : return CreatePluginEndpoint(endpoint, endpointHandle);
212 : }
213 :
214 18 : HcommResult PluginEndpointGet(EndpointHandle handle, void **endpoint, bool &handled)
215 : {
216 18 : PluginEndpointCtx *ctx = GetPluginEndpointCtx(handle, handled);
217 18 : if (!handled) {
218 17 : return HCCL_SUCCESS;
219 : }
220 1 : CHK_PTR_NULL(ctx);
221 1 : *endpoint = ctx->ctx;
222 1 : return HCCL_SUCCESS;
223 : }
224 :
225 32 : HcommResult PluginEndpointDestroy(EndpointHandle handle, bool &handled)
226 : {
227 32 : handled = IsPluginEndpoint(handle);
228 32 : return handled ? DestroyPluginEndpoint(handle) : HCCL_SUCCESS;
229 : }
230 :
231 34 : HcommResult PluginMemReg(EndpointHandle handle, const char *memTag,
232 : const CommMem *mem, HcommMemHandle *memHandle, bool &handled)
233 : {
234 34 : DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, registerMemory,
235 : (ctx->ctx, mem, memTag, reinterpret_cast<void **>(memHandle)));
236 : }
237 :
238 34 : HcommResult PluginMemUnreg(EndpointHandle handle, HcommMemHandle memHandle, bool &handled)
239 : {
240 34 : DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, unregisterMemory, (ctx->ctx, memHandle));
241 : }
242 :
243 5 : HcommResult PluginMemExport(EndpointHandle handle, HcommMemHandle memHandle,
244 : void **memDesc, uint32_t *memDescLen, bool &handled)
245 : {
246 5 : DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, memoryExport, (ctx->ctx, memHandle, memDesc, memDescLen));
247 : }
248 :
249 4 : HcommResult PluginMemImport(EndpointHandle handle, const void *memDesc, uint32_t descLen,
250 : CommMem *outMem, bool &handled)
251 : {
252 4 : DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, memoryImport, (ctx->ctx, memDesc, descLen, outMem));
253 : }
254 :
255 4 : HcommResult PluginMemUnimport(EndpointHandle handle, const void *memDesc, uint32_t descLen, bool &handled)
256 : {
257 4 : DISPATCH_PLUGIN_ENDPOINT_OP(handle, handled, memoryUnimport, (ctx->ctx, memDesc, descLen));
258 : }
259 :
260 3 : HcommResult PluginChannelCreate(EndpointHandle endpointHandle, const HcommChannelDesc *channelDesc,
261 : ChannelHandle *channelHandle)
262 : {
263 3 : return CreatePluginChannel(endpointHandle, channelDesc, channelHandle);
264 : }
265 :
266 21 : HcommResult PluginChannelCreate(EndpointHandle endpointHandle, CommEngine engine,
267 : const HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *channels, bool &handled)
268 : {
269 21 : handled = IsPluginEndpoint(endpointHandle);
270 21 : if (!handled) {
271 17 : return HCCL_SUCCESS;
272 : }
273 4 : CHK_PTR_NULL(channelDescs);
274 4 : CHK_PTR_NULL(channels);
275 4 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]",
276 : __func__, channelNum), HCCL_E_PARA);
277 4 : CHK_PRT_RET(engine != COMM_ENGINE_CPU,
278 : HCCL_ERROR("[%s] nic plugin endpoint only supports COMM_ENGINE_CPU, engine[%d].", __func__, engine),
279 : HCCL_E_NOT_SUPPORT);
280 :
281 6 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
282 3 : HcommResult ret = PluginChannelCreate(endpointHandle, &channelDescs[idx], &channels[idx]);
283 3 : if (ret != HCCL_SUCCESS) {
284 0 : DestroyCreatedPluginChannels(channels, idx);
285 0 : return ret;
286 : }
287 : }
288 :
289 3 : HcommResult ret = ConnectPluginChannels(channels, channelNum);
290 3 : if (ret != HCCL_SUCCESS) {
291 2 : DestroyCreatedPluginChannels(channels, channelNum);
292 2 : return ret;
293 : }
294 1 : return HCCL_SUCCESS;
295 : }
296 :
297 42 : HcommResult PluginChannelGet(ChannelHandle handle, void **channel, bool &handled)
298 : {
299 42 : PluginChannelCtx *ctx = GetPluginChannelCtx(handle, handled);
300 42 : if (!handled) {
301 42 : return HCCL_SUCCESS;
302 : }
303 0 : CHK_PTR_NULL(ctx);
304 0 : *channel = ctx->ctx;
305 0 : return HCCL_SUCCESS;
306 : }
307 :
308 35 : HcommResult PluginChannelGetStatus(ChannelHandle handle, int32_t *status, bool &handled)
309 : {
310 35 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, getStatus, (ctx->ctx, status));
311 : }
312 :
313 1 : HcommResult PluginChannelGetNotifyNum(ChannelHandle handle, uint32_t *notifyNum, bool &handled)
314 : {
315 1 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, getNotifyNum, (ctx->ctx, notifyNum));
316 : }
317 :
318 19 : HcommResult PluginChannelDestroy(ChannelHandle handle, bool &handled)
319 : {
320 19 : handled = IsPluginChannel(handle);
321 19 : return handled ? DestroyPluginChannel(handle) : HCCL_SUCCESS;
322 : }
323 :
324 0 : HcommResult PluginChannelUpdateMemInfo(ChannelHandle handle, HcommMemHandle *memHandles, uint32_t memHandleNum,
325 : bool &handled)
326 : {
327 0 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, updateMemInfo, (ctx->ctx, memHandles, memHandleNum));
328 : }
329 :
330 7 : HcommResult PluginChannelGetRemoteMems(ChannelHandle handle, uint32_t *memNum, CommMem **remoteMem,
331 : char ***memInfos, bool &handled)
332 : {
333 7 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, getUserRemoteMem, (ctx->ctx, remoteMem, memInfos, memNum));
334 : }
335 :
336 2 : HcommResult PluginChannelWrite(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src, uint64_t len,
337 : bool &handled)
338 : {
339 : (void)thread;
340 2 : return DispatchPluginChannelWriteNbi(handle, dst, src, len, handled, __func__);
341 : }
342 :
343 1 : HcommResult PluginChannelBatchTransfer(ChannelHandle handle, ThreadHandle thread,
344 : const HcommBatchTransferDesc *transferDescs, uint32_t transferDescNum, bool &handled)
345 : {
346 : (void)thread;
347 : (void)transferDescs;
348 : (void)transferDescNum;
349 1 : return UnsupportedPluginChannelOp(handle, handled, __func__);
350 : }
351 :
352 1 : HcommResult PluginChannelWriteReduce(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
353 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, bool &handled)
354 : {
355 1 : return UnsupportedPluginChannelReduceOp(handle, thread, dst, src, count, dataType, reduceOp, handled, __func__);
356 : }
357 :
358 1 : HcommResult PluginChannelWriteWithNotify(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
359 : uint64_t len, uint32_t remoteNotifyIdx, bool &handled)
360 : {
361 : (void)thread;
362 1 : return DispatchPluginChannelWriteWithNotifyNbi(handle, dst, src, len, remoteNotifyIdx, handled, __func__);
363 : }
364 :
365 1 : HcommResult PluginChannelWriteReduceWithNotify(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
366 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, uint32_t remoteNotifyIdx, bool &handled)
367 : {
368 : (void)thread;
369 : (void)dst;
370 : (void)src;
371 : (void)count;
372 : (void)dataType;
373 : (void)reduceOp;
374 : (void)remoteNotifyIdx;
375 1 : return UnsupportedPluginChannelOp(handle, handled, __func__);
376 : }
377 :
378 1 : HcommResult PluginChannelRead(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src, uint64_t len,
379 : bool &handled)
380 : {
381 : (void)thread;
382 1 : return DispatchPluginChannelReadNbi(handle, dst, src, len, handled, __func__);
383 : }
384 :
385 1 : HcommResult PluginChannelReadReduce(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
386 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, bool &handled)
387 : {
388 1 : return UnsupportedPluginChannelReduceOp(handle, thread, dst, src, count, dataType, reduceOp, handled, __func__);
389 : }
390 :
391 8 : HcommResult PluginChannelWriteNbi(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src, uint64_t len,
392 : bool &handled)
393 : {
394 : (void)thread;
395 8 : return DispatchPluginChannelWriteNbi(handle, dst, src, len, handled, __func__);
396 : }
397 :
398 7 : HcommResult PluginChannelWriteWithNotifyNbi(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src,
399 : uint64_t len, uint32_t remoteNotifyIdx, bool &handled)
400 : {
401 : (void)thread;
402 7 : return DispatchPluginChannelWriteWithNotifyNbi(handle, dst, src, len, remoteNotifyIdx, handled, __func__);
403 : }
404 :
405 7 : HcommResult PluginChannelReadNbi(ChannelHandle handle, ThreadHandle thread, void *dst, const void *src, uint64_t len,
406 : bool &handled)
407 : {
408 : (void)thread;
409 7 : return DispatchPluginChannelReadNbi(handle, dst, src, len, handled, __func__);
410 : }
411 :
412 11 : HcommResult PluginChannelNotifyRecord(ChannelHandle handle, ThreadHandle thread, uint32_t remoteNotifyIdx,
413 : bool &handled)
414 : {
415 : (void)thread;
416 11 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, notifyRecord, (ctx->ctx, remoteNotifyIdx));
417 : }
418 :
419 10 : HcommResult PluginChannelNotifyWait(ChannelHandle handle, ThreadHandle thread, uint32_t localNotifyIdx,
420 : uint32_t timeOut, bool &handled)
421 : {
422 : (void)thread;
423 10 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, notifyWait, (ctx->ctx, localNotifyIdx, timeOut));
424 : }
425 :
426 9 : HcommResult PluginChannelFence(ChannelHandle handle, ThreadHandle thread, bool &handled)
427 : {
428 : (void)thread;
429 9 : DISPATCH_PLUGIN_CHANNEL_OP(handle, handled, fence, (ctx->ctx));
430 : }
431 :
432 : } // namespace hcomm
|
