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 "transport_base.h"
12 : #include "adapter_rts.h"
13 : #include "externalinput_pub.h"
14 : #include "device_capacity.h"
15 : #include "new/hccl_dispatcher_ctx.h"
16 : #include "dispatcher_ctx.h"
17 :
18 : namespace hccl {
19 : struct SuperPodInfo {
20 : s32 pid = 0;
21 : s32 sdid = INVALID_INT; // super Pod device id
22 : s32 serverPhyIdx = INVALID_INT; // 超节点server id
23 : };
24 :
25 137 : TransportBase::TransportBase(DispatcherPub *dispatcher,
26 : const std::unique_ptr<NotifyPool> ¬ifyPool,
27 : MachinePara &machinePara,
28 137 : std::chrono::milliseconds timeout)
29 137 : : exchangeDataTotalSize_(0),
30 137 : dispatcher_(dispatcher), notifyPool_(notifyPool), defaultSocket_(nullptr), machinePara_(machinePara),
31 274 : timeout_(timeout), recvPid_(0), recvSdid_(INVALID_INT),
32 137 : nicDeploy_(NICDeployment::NIC_DEPLOYMENT_RESERVED),
33 274 : useOneDoorbell_(false), notifyNum_(machinePara.notifyNum)
34 : {
35 137 : if (machinePara_.sockets.size() > 0) {
36 2 : defaultSocket_ = machinePara_.sockets[0];
37 : }
38 137 : }
39 :
40 221 : TransportBase::~TransportBase()
41 : {
42 221 : }
43 :
44 83 : HcclResult TransportBase::Init()
45 : {
46 83 : CHK_SMART_PTR_NULL(dispatcher_);
47 83 : CHK_RET(CheckExchangeData());
48 :
49 83 : return HCCL_SUCCESS;
50 : }
51 :
52 :
53 7 : HcclResult TransportBase::CheckDeviceId()
54 : {
55 : u32 maxDeviceNum;
56 7 : CHK_RET(GetMaxDevNum(maxDeviceNum));
57 7 : bool invalidDevId =
58 7 : machinePara_.deviceLogicId < 0 || (static_cast<u32>(machinePara_.deviceLogicId) >= maxDeviceNum);
59 7 : CHK_PRT_RET(invalidDevId,
60 : HCCL_ERROR("[TransportBase][CheckDeviceId] deviceLogicId[%d] is invalid", machinePara_.deviceLogicId),
61 : HCCL_E_INTERNAL);
62 7 : return HCCL_SUCCESS;
63 : }
64 :
65 0 : HcclResult TransportBase::DeInit()
66 : {
67 0 : return HCCL_SUCCESS;
68 : }
69 :
70 0 : HcclResult TransportBase::Stop()
71 : {
72 0 : return HCCL_SUCCESS;
73 : }
74 :
75 0 : HcclResult TransportBase::Resume()
76 : {
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : TransportAttr TransportBase::GetTransportAttr()
81 : {
82 0 : return transportAttr_;
83 : }
84 :
85 0 : HcclResult TransportBase::TxDataSignal(Stream &stream)
86 : {
87 : static_cast<void>(stream);
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 9 : HcclResult TransportBase::RxDataSignal(Stream &stream)
92 : {
93 : static_cast<void>(stream);
94 9 : return HCCL_SUCCESS;
95 : }
96 :
97 0 : HcclResult TransportBase::TxData(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len, Stream &stream)
98 : {
99 : static_cast<void>(dstMemType);
100 : static_cast<void>(dstOffset);
101 : static_cast<void>(src);
102 : static_cast<void>(len);
103 : static_cast<void>(stream);
104 0 : return HCCL_SUCCESS;
105 : }
106 :
107 0 : HcclResult TransportBase::RxData(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream)
108 : {
109 : static_cast<void>(srcMemType);
110 : static_cast<void>(srcOffset);
111 : static_cast<void>(dst);
112 : static_cast<void>(len);
113 : static_cast<void>(stream);
114 0 : return HCCL_SUCCESS;
115 : }
116 :
117 9 : HcclResult TransportBase::TxAsync(UserMemType dstMemType, u64 dstOffset, const void *src,
118 : u64 len, Stream &stream)
119 : {
120 : static_cast<void>(dstMemType);
121 : static_cast<void>(dstOffset);
122 : static_cast<void>(src);
123 : static_cast<void>(len);
124 : static_cast<void>(stream);
125 9 : return HCCL_SUCCESS;
126 : }
127 :
128 0 : HcclResult TransportBase::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream &stream)
129 : {
130 : static_cast<void>(txMems);
131 : static_cast<void>(stream);
132 0 : return HCCL_SUCCESS;
133 : }
134 :
135 9 : HcclResult TransportBase::RxAsync(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream)
136 : {
137 : static_cast<void>(srcMemType);
138 : static_cast<void>(srcOffset);
139 : static_cast<void>(dst);
140 : static_cast<void>(len);
141 : static_cast<void>(stream);
142 9 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : HcclResult TransportBase::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream &stream)
146 : {
147 : static_cast<void>(rxMems);
148 : static_cast<void>(stream);
149 0 : return HCCL_SUCCESS;
150 : }
151 :
152 0 : HcclResult TransportBase::DataReceivedAck(Stream &stream)
153 : {
154 : static_cast<void>(stream);
155 0 : return HCCL_SUCCESS;
156 : }
157 :
158 18 : HcclResult TransportBase::TxAck(Stream &stream)
159 : {
160 : static_cast<void>(stream);
161 18 : return HCCL_SUCCESS;
162 : }
163 :
164 9 : HcclResult TransportBase::RxAck(Stream &stream)
165 : {
166 : static_cast<void>(stream);
167 9 : return HCCL_SUCCESS;
168 : }
169 :
170 0 : HcclResult TransportBase::TxPrepare(Stream &stream)
171 : {
172 : static_cast<void>(stream);
173 0 : return HCCL_SUCCESS;
174 : }
175 :
176 0 : HcclResult TransportBase::RxPrepare(Stream &stream)
177 : {
178 : static_cast<void>(stream);
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : HcclResult TransportBase::TxDone(Stream &stream)
183 : {
184 : static_cast<void>(stream);
185 0 : return HCCL_SUCCESS;
186 : }
187 :
188 0 : HcclResult TransportBase::RxDone(Stream &stream)
189 : {
190 : static_cast<void>(stream);
191 0 : return HCCL_SUCCESS;
192 : }
193 :
194 0 : HcclResult TransportBase::TxWaitDone(Stream &stream)
195 : {
196 : static_cast<void>(stream);
197 0 : return HCCL_SUCCESS;
198 : }
199 :
200 0 : HcclResult TransportBase::RxWaitDone(Stream &stream)
201 : {
202 : static_cast<void>(stream);
203 0 : return HCCL_SUCCESS;
204 : }
205 :
206 0 : HcclResult TransportBase::Post(u32 notifyIdx, Stream &stream)
207 : {
208 : static_cast<void>(notifyIdx);
209 : static_cast<void>(stream);
210 0 : return HCCL_E_NOT_SUPPORT;
211 : }
212 :
213 0 : HcclResult TransportBase::Wait(u32 notifyIdx, Stream &stream, const u32 timeOut)
214 : {
215 : static_cast<void>(notifyIdx);
216 : static_cast<void>(stream);
217 : static_cast<void>(timeOut);
218 0 : return HCCL_E_NOT_SUPPORT;
219 : }
220 :
221 :
222 0 : HcclResult TransportBase::TxEnv(const void *ptr, const u64 len, Stream &stream)
223 : {
224 0 : return HCCL_SUCCESS;
225 : }
226 :
227 0 : HcclResult TransportBase::RxEnv(Stream &stream)
228 : {
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 :
233 0 : HcclResult TransportBase::TxWithReduce(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len,
234 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
235 : {
236 : static_cast<void>(dstMemType);
237 : static_cast<void>(dstOffset);
238 : static_cast<void>(src);
239 : static_cast<void>(len);
240 : static_cast<void>(datatype);
241 : static_cast<void>(redOp);
242 : static_cast<void>(stream);
243 0 : return HCCL_SUCCESS;
244 : }
245 :
246 0 : HcclResult TransportBase::TxWithReduce(const std::vector<TxMemoryInfo>& txWithReduceMems,
247 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
248 : {
249 : static_cast<void>(txWithReduceMems);
250 : static_cast<void>(datatype);
251 : static_cast<void>(redOp);
252 : static_cast<void>(stream);
253 0 : return HCCL_SUCCESS;
254 : }
255 :
256 0 : HcclResult TransportBase::RxWithReduce(UserMemType recvSrcMemType, u64 recvSrcOffset, void *recvDst, u64 recvLen,
257 : void *reduceSrc, void *reduceDst, u64 reduceDataCount, HcclDataType reduceDatatype,
258 : HcclReduceOp reduceOp, Stream &stream, const u64 reduceAttr)
259 : {
260 : static_cast<void>(recvSrcMemType);
261 : static_cast<void>(recvSrcOffset);
262 : static_cast<void>(recvDst);
263 : static_cast<void>(recvLen);
264 : static_cast<void>(reduceSrc);
265 : static_cast<void>(reduceDst);
266 : static_cast<void>(reduceDataCount);
267 : static_cast<void>(reduceDatatype);
268 : static_cast<void>(reduceOp);
269 : static_cast<void>(stream);
270 : static_cast<void>(reduceAttr);
271 0 : return HCCL_SUCCESS;
272 : }
273 :
274 0 : HcclResult TransportBase::RxWithReduce(const std::vector<RxWithReduceMemoryInfo> &rxWithReduceMems,
275 : HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream &stream, const u64 reduceAttr)
276 : {
277 : static_cast<void>(rxWithReduceMems);
278 : static_cast<void>(reduceDatatype);
279 : static_cast<void>(reduceOp);
280 : static_cast<void>(stream);
281 : static_cast<void>(reduceAttr);
282 0 : return HCCL_SUCCESS;
283 : }
284 :
285 18 : bool TransportBase::IsSupportTransportWithReduce()
286 : {
287 18 : return false;
288 : }
289 :
290 0 : HcclResult TransportBase::GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t *memNum, HcclMemType memType)
291 : {
292 : static_cast<void>(remoteMem);
293 : static_cast<void>(memNum);
294 0 : return HCCL_E_PARA;
295 : }
296 :
297 0 : HcclResult TransportBase::GetIndOpRemoteMem(HcclMem **remoteMem, uint32_t *memNum)
298 : {
299 : static_cast<void>(remoteMem);
300 : static_cast<void>(memNum);
301 0 : return HCCL_E_PARA;
302 : }
303 :
304 0 : HcclResult TransportBase::GetRemoteMem(UserMemType memType, void **remotePtr)
305 : {
306 : static_cast<void>(memType);
307 : static_cast<void>(remotePtr);
308 0 : return HCCL_E_PARA;
309 : }
310 :
311 0 : HcclResult TransportBase::GetRemoteMem(std::vector<void *> *remotePtrVec)
312 : {
313 : static_cast<void>(remotePtrVec);
314 0 : return HCCL_SUCCESS;
315 : }
316 :
317 0 : HcclResult TransportBase::GetRemoteMemKey(UserMemType memType, uint32_t *remoteMemKey)
318 : {
319 : static_cast<void>(memType);
320 : static_cast<void>(remoteMemKey);
321 0 : return HCCL_E_PARA;
322 : }
323 :
324 0 : HcclResult TransportBase::GetRemoteMemSize(UserMemType memType, u64 &size)
325 : {
326 : static_cast<void>(memType);
327 : static_cast<void>(size);
328 0 : return HCCL_E_PARA;
329 : }
330 :
331 0 : HcclResult TransportBase::GetLocalRdmaNotify(std::vector<HcclSignalInfo> &rdmaNotify)
332 : {
333 : static_cast<void>(rdmaNotify);
334 0 : return HCCL_E_PARA;
335 : }
336 :
337 0 : HcclResult TransportBase::GetDrainLocalDataNotify(void* &localAddr, uint32_t& lkey, HcclSignalInfo &dataNotify)
338 : {
339 : static_cast<void>(localAddr);
340 : static_cast<void>(lkey);
341 : static_cast<void>(dataNotify);
342 0 : return HCCL_E_NOT_SUPPORT;
343 : }
344 :
345 0 : HcclResult TransportBase::GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey> &rdmaNotifyAddr)
346 : {
347 : static_cast<void>(rdmaNotifyAddr);
348 0 : return HCCL_E_PARA;
349 : }
350 :
351 0 : HcclResult TransportBase::GetLocalNotifyValueAddrKey(std::vector<AddrKey> ¬ifyValue)
352 : {
353 : static_cast<void>(notifyValue);
354 0 : return HCCL_E_PARA;
355 : }
356 :
357 0 : HcclResult TransportBase::GetLocalMemDetails(UserMemType memType, MemDetails &memDetails)
358 : {
359 : static_cast<void>(memType);
360 : static_cast<void>(memDetails);
361 0 : return HCCL_E_PARA;
362 : }
363 :
364 0 : HcclResult TransportBase::GetLocalNotify(std::vector<HcclSignalInfo> &localNotify)
365 : {
366 : static_cast<void>(localNotify);
367 0 : return HCCL_E_PARA;
368 : }
369 :
370 0 : HcclResult TransportBase::GetRemoteNotify(std::vector<HcclSignalInfo> &localNotify)
371 : {
372 : static_cast<void>(localNotify);
373 0 : return HCCL_E_PARA;
374 : }
375 :
376 0 : HcclResult TransportBase::GetAiQpInfo(std::vector<HcclQpInfoV2> &aiQpInfo)
377 : {
378 : static_cast<void>(aiQpInfo);
379 0 : return HCCL_E_PARA;
380 : }
381 0 : HcclResult TransportBase::GetTransportId(u32 &id)
382 : {
383 : static_cast<void>(id);
384 0 : return HCCL_E_PARA;
385 : }
386 :
387 0 : HcclResult TransportBase::GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo> &aiRMAQueueInfo)
388 : {
389 : static_cast<void>(aiRMAQueueInfo);
390 0 : return HCCL_E_PARA;
391 : }
392 :
393 0 : HcclResult TransportBase::FillExchangeDataTotalSize()
394 : {
395 0 : exchangeDataTotalSize_ = 0;
396 0 : return HCCL_E_PARA; // this function should not be called in normal process
397 : }
398 :
399 0 : HcclResult TransportBase::ConstructExchangeForSend()
400 : {
401 0 : return HCCL_E_PARA; // this function should not be called in normal process
402 : }
403 :
404 0 : HcclResult TransportBase::ParseReceivedExchangeData()
405 : {
406 0 : return HCCL_E_PARA; // this function should not be called in normal process
407 : }
408 :
409 0 : HcclResult TransportBase::GetChipId(s64 &chipId)
410 : {
411 0 : CHK_RET(hrtGetDeviceInfo(machinePara_.deviceLogicId, HcclRtDeviceModuleType::HCCL_RT_MODULE_TYPE_SYSTEM,
412 : HcclRtDeviceInfoType::HCCL_INFO_TYPE_PHY_CHIP_ID, chipId));
413 0 : HCCL_DEBUG("[GetChipId]chipId: %ld", chipId);
414 0 : return HCCL_SUCCESS;
415 : }
416 :
417 2 : HcclResult TransportBase::ExchangeTgidMesg()
418 : {
419 2 : SuperPodInfo sendInfo;
420 2 : CHK_RET(SalGetBareTgid(&sendInfo.pid)); // 当前进程id
421 2 : if (machinePara_.deviceType == DevType::DEV_TYPE_910_93) {
422 0 : s64 sdid = 0;
423 0 : CHK_RET(hrtGetDeviceInfo(machinePara_.deviceLogicId, HcclRtDeviceModuleType::HCCL_RT_MODULE_TYPE_SYSTEM,
424 : HcclRtDeviceInfoType::HCCL_INFO_TYPE_SDID, sdid));
425 0 : sendInfo.sdid = static_cast<s32>(sdid);
426 :
427 0 : s64 serverPhyIdx = 0;
428 0 : CHK_RET(hrtGetDeviceInfo(machinePara_.deviceLogicId, HcclRtDeviceModuleType::HCCL_RT_MODULE_TYPE_SYSTEM,
429 : HcclRtDeviceInfoType::HCCL_INFO_TYPE_SERVER_ID, serverPhyIdx));
430 0 : sendInfo.serverPhyIdx = static_cast<s32>(serverPhyIdx);
431 : }
432 :
433 2 : HcclResult ret = HCCL_SUCCESS;
434 2 : CHK_SMART_PTR_NULL(defaultSocket_);
435 2 : ret = defaultSocket_->Send(reinterpret_cast<u8*>(&sendInfo), sizeof(SuperPodInfo));
436 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
437 : HCCL_ERROR("[Exchange][TgidMesg]errNo[0x%016llx] In exchange tgid mesg, send pid failed. "\
438 : "remote userrank[%u] pid[%d] sdid[%016llx] local rank[%u]", HCCL_ERROR_CODE(ret),
439 : machinePara_.remoteUserrank, sendInfo.pid, sendInfo.sdid, machinePara_.localUserrank), ret);
440 :
441 2 : SuperPodInfo recvInfo = {};
442 2 : ret = defaultSocket_->Recv(reinterpret_cast<u8*>(&recvInfo), sizeof(SuperPodInfo));
443 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
444 : HCCL_ERROR("[Exchange][TgidMesg]errNo[0x%016llx] In exchange tgid mesg, recv pid failed. "\
445 : "remote userrank[%u] pid[%d] sdid[%016llx] local rank[%u]", HCCL_ERROR_CODE(ret),
446 : machinePara_.remoteUserrank, recvInfo.pid, recvInfo.sdid, machinePara_.localUserrank), ret);
447 :
448 2 : recvPid_ = recvInfo.pid;
449 : // sdid同时满足以下条件时使用: 1.跨server场景 2.使能HCCS 3.超节点内(默认满足, 链路选择时保证)
450 4 : recvSdid_ = (sendInfo.serverPhyIdx != recvInfo.serverPhyIdx &&
451 2 : !GetExternalInputInterHccsDisable()) ? recvInfo.sdid : INVALID_INT;
452 2 : HCCL_INFO("[Exchange][TgidMesg]local: rank[%u], pid[%d], sdid[%016llx], serverPhyIdx[%016llx], "\
453 : "remote: rank[%u], pid[%d], sdid[%016llx], serverPhyIdx[%016llx], recvSdid[%016llx]",
454 : machinePara_.localUserrank, sendInfo.pid, sendInfo.sdid, sendInfo.serverPhyIdx,
455 : machinePara_.remoteUserrank, recvInfo.pid, recvInfo.sdid, recvInfo.serverPhyIdx, recvSdid_);
456 :
457 2 : return HCCL_SUCCESS;
458 : }
459 :
460 0 : HcclResult TransportBase::SendNotifyReadyMesg()
461 : {
462 0 : HCCL_DEBUG("[Send][NotifyReadyMesg]recvSDID[%016llx], remoteRank[%016llx], recvPid[%016llx]",
463 : recvSdid_, machinePara_.remoteUserrank, recvPid_);
464 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
465 0 : CHK_SMART_PTR_NULL(notifyPool_);
466 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendReadyNotify_));
467 :
468 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
469 0 : CHK_RET(localSendReadyNotify_->Serialize(data));
470 0 : CHK_SMART_PTR_NULL(defaultSocket_);
471 0 : HcclResult ret = defaultSocket_->Send(&data[0], data.size());
472 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
473 : HCCL_ERROR("[Send][IpcNotifyReadyMesg]errNo[0x%016llx]In send notify ready mesg, send read msg failed. remote "
474 : "userrank[%u] notify locak rank[%u]",
475 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
476 : ret);
477 :
478 0 : HCCL_DEBUG("local_send_ready_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
479 : machinePara_.remoteUserrank);
480 0 : return HCCL_SUCCESS;
481 0 : }
482 :
483 0 : HcclResult TransportBase::SendNotifyDoneMesg()
484 : {
485 0 : HCCL_DEBUG("[Send][NotifyDoneMesg]recvSDID[%016llx], remoteRank[%016llx], recvPid[%016llx]",
486 : recvSdid_, machinePara_.remoteUserrank, recvPid_);
487 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
488 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendDoneNotify_));
489 :
490 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
491 0 : CHK_RET(localSendDoneNotify_->Serialize(data));
492 0 : CHK_SMART_PTR_NULL(defaultSocket_);
493 0 : HcclResult ret = defaultSocket_->Send(&data[0], data.size());
494 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
495 : HCCL_ERROR("[Send][IpcNotifyDoneMesg]errNo[0x%016llx] In send notify done mesg, send done msg "\
496 : "failed. remote userrank[%u] local rank[%u]", HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank,
497 : machinePara_.localUserrank), ret);
498 :
499 0 : HCCL_DEBUG("send_done_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
500 : machinePara_.remoteUserrank);
501 0 : return HCCL_SUCCESS;
502 0 : }
503 :
504 0 : HcclResult TransportBase::SendDeviceIpcNotifyReadyMesg()
505 : {
506 0 : HCCL_DEBUG("[Send][DeviceIpcNotifyReadyMesg]recvSDID[%016llx], remoteRank[%016llx], recvPid[%016llx]",
507 : recvSdid_, machinePara_.remoteUserrank, recvPid_);
508 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
509 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendReadyDeviceNotify_, NotifyLoadType::DEVICE_NOTIFY));
510 :
511 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
512 0 : CHK_RET(localSendReadyDeviceNotify_->Serialize(data));
513 0 : CHK_SMART_PTR_NULL(defaultSocket_);
514 0 : HcclResult ret = defaultSocket_->Send(&data[0], data.size());
515 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
516 : HCCL_ERROR("[Send][IpcNotifyReadyMesg]errNo[0x%016llx]In send notify ready mesg, send read msg failed. remote "
517 : "userrank[%u] notify locak rank[%u]",
518 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
519 : ret);
520 :
521 0 : HCCL_DEBUG("send_device_ready_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
522 : machinePara_.remoteUserrank);
523 0 : return HCCL_SUCCESS;
524 0 : }
525 :
526 0 : HcclResult TransportBase::SendDeviceIpcNotifyDoneMesg()
527 : {
528 0 : HCCL_DEBUG("[Send][DeviceIpcNotifyDoneMesg]recvSDID[%016llx], remoteRank[%016llx], recvPid[%016llx]",
529 : recvSdid_, machinePara_.remoteUserrank, recvPid_);
530 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
531 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendDoneDeviceNotify_, NotifyLoadType::DEVICE_NOTIFY));
532 :
533 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
534 0 : CHK_RET(localSendDoneDeviceNotify_->Serialize(data));
535 0 : CHK_SMART_PTR_NULL(defaultSocket_);
536 0 : HcclResult ret = defaultSocket_->Send(&data[0], data.size());
537 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
538 : HCCL_ERROR("[Send][IpcNotifyReadyMesg]errNo[0x%016llx]In send notify ready mesg, send read msg failed. remote "
539 : "userrank[%u] notify locak rank[%u]",
540 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
541 : ret);
542 :
543 0 : HCCL_DEBUG("send_device_done_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
544 : machinePara_.remoteUserrank);
545 0 : return HCCL_SUCCESS;
546 0 : }
547 :
548 0 : HcclResult TransportBase::RecvNotifyReadyMesg()
549 : {
550 : // 获取ready notify data
551 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
552 0 : CHK_SMART_PTR_NULL(defaultSocket_);
553 0 : HcclResult ret = defaultSocket_->Recv(&data[0], NOTIFY_INFO_LENGTH);
554 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
555 : HCCL_ERROR("[Recv][NotifyReadyMesg]errNo[0x%016llx]receive remote send ready notify data failed. remote "
556 : "user rank[%u], receive local rank[%u]",
557 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
558 : ret);
559 :
560 0 : CHK_RET(OpenRemoteNotify(data, remoteSendReadyNotify_));
561 0 : return HCCL_SUCCESS;
562 0 : }
563 :
564 0 : HcclResult TransportBase::RecvNotifyDoneMesg()
565 : {
566 : // 获取done notify data
567 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
568 0 : CHK_SMART_PTR_NULL(defaultSocket_);
569 0 : HcclResult ret = defaultSocket_->Recv(&data[0], NOTIFY_INFO_LENGTH);
570 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
571 : HCCL_ERROR("[Recv][RecvNotifyDoneMesg]errNo[0x%016llx]receive remote send ready notify data failed. remote "
572 : "user rank[%u], receive local rank[%u]",
573 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
574 : ret);
575 0 : HCCL_DEBUG("send_done_notify rank[%u] receive from rank[%u]", machinePara_.localUserrank,
576 : machinePara_.remoteUserrank);
577 :
578 0 : CHK_RET(OpenRemoteNotify(data, remoteSendDoneNotify_));
579 :
580 0 : HCCL_DEBUG("remote_send_done_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
581 : machinePara_.remoteUserrank);
582 :
583 0 : return HCCL_SUCCESS;
584 0 : }
585 :
586 0 : HcclResult TransportBase::RecvDeviceIpcNotifyReadyMesg()
587 : {
588 : // 获取ready notify data
589 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
590 0 : CHK_SMART_PTR_NULL(defaultSocket_);
591 0 : HcclResult ret = defaultSocket_->Recv(&data[0], NOTIFY_INFO_LENGTH);
592 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
593 : HCCL_ERROR("[Recv][DeviceIpcNotifyReadyMesg]errNo[0x%016llx]receive remote send ready notify data failed. "
594 : "remote user rank[%u], receive local rank[%u]",
595 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
596 : ret);
597 0 : HCCL_DEBUG("send_ready_device_notify rank[%u] receive from rank[%u]", machinePara_.localUserrank,
598 : machinePara_.remoteUserrank);
599 :
600 0 : CHK_RET(OpenRemoteNotify(data, remoteSendReadyDeviceNotify_));
601 :
602 0 : HCCL_DEBUG("remote_send_ready_device_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
603 : machinePara_.remoteUserrank);
604 0 : return HCCL_SUCCESS;
605 0 : }
606 :
607 0 : HcclResult TransportBase::RecvDeviceIpcNotifyDoneMesg()
608 : {
609 : // 获取ready notify data
610 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
611 0 : CHK_SMART_PTR_NULL(defaultSocket_);
612 0 : HcclResult ret = defaultSocket_->Recv(&data[0], NOTIFY_INFO_LENGTH);
613 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
614 : HCCL_ERROR("[Recv][DeviceIpcNotifyDoneMesg]errNo[0x%016llx]receive remote send ready notify data failed. remote"
615 : " user rank[%u], receive local rank[%u]",
616 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
617 : ret);
618 0 : HCCL_DEBUG("send_done_device_notify rank[%u] receive from rank[%u]", machinePara_.localUserrank,
619 : machinePara_.remoteUserrank);
620 :
621 0 : CHK_RET(OpenRemoteNotify(data, remoteSendDoneDeviceNotify_));
622 :
623 0 : HCCL_DEBUG("remote_send_done_device_notify send rank[%u] to rank[%u]", machinePara_.localUserrank,
624 : machinePara_.remoteUserrank);
625 0 : return HCCL_SUCCESS;
626 0 : }
627 :
628 0 : HcclResult TransportBase::CheckLinkStatus()
629 : {
630 : HcclResult ret;
631 : /* link状态 */
632 0 : std::string localLinkStatus = "true";
633 0 : CHK_SMART_PTR_NULL(defaultSocket_);
634 0 : ret = defaultSocket_->Send(localLinkStatus);
635 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
636 : HCCL_ERROR("[Check][LinkStatus]errNo[0x%016llx]In check link status, send link status failed. "\
637 : "remote userrank[%u] local rank[%u]", HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank,
638 : machinePara_.localUserrank), ret);
639 :
640 0 : HCCL_DEBUG("local_link_status send rank[%u] to rank[%u] message[%s]", machinePara_.localUserrank,
641 : machinePara_.remoteUserrank, localLinkStatus.c_str());
642 :
643 : // 获取remote_link_status
644 0 : std::string remoteLinkStatus;
645 0 : CHK_SMART_PTR_NULL(defaultSocket_);
646 0 : ret = defaultSocket_->Recv(remoteLinkStatus);
647 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
648 : HCCL_ERROR("[Check][LinkStatus]errNo[0x%016llx]In check link status, receive remote link status failed. "\
649 : "remote user rank[%u] local rank[%u]", HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank,
650 : machinePara_.localUserrank), ret);
651 :
652 0 : HCCL_DEBUG("remote_link_status rank[%u] receive from rank[%u] message[%s]", machinePara_.localUserrank,
653 : machinePara_.remoteUserrank, remoteLinkStatus.c_str());
654 0 : return HCCL_SUCCESS;
655 0 : }
656 :
657 2 : HcclResult TransportBase::CheckLinkMode()
658 : {
659 4 : bool bErr = (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE) &&
660 2 : (machinePara_.linkMode != LinkMode::LINK_DUPLEX_MODE);
661 2 : CHK_PRT_RET(bErr, \
662 : HCCL_ERROR("[Check][LinkMode]errNo[0x%016llx] check LinkMode[%d] fail", HCCL_ERROR_CODE(HCCL_E_PARA),
663 : machinePara_.linkMode), HCCL_E_PARA);
664 2 : return HCCL_SUCCESS;
665 : }
666 :
667 0 : HcclResult TransportBase::LinkSendNotifyMesg()
668 : {
669 0 : s32 sendPid = 0;
670 0 : CHK_RET(SalGetBareTgid(&sendPid)); // 当前进程id
671 0 : HCCL_INFO("LinkSendNotifyMesg, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
672 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
673 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
674 : /* 发送IPC notify Ready 信息 */
675 0 : CHK_RET(SendNotifyReadyMesg());
676 : }
677 :
678 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
679 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
680 : /* 发送IPC notify Done 信息 */
681 0 : CHK_RET(SendNotifyDoneMesg());
682 : }
683 :
684 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
685 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) &&
686 0 : machinePara_.isAicpuModeEn == true) {
687 : /* 发送Device上使用的IPC notify ready信息 */
688 0 : CHK_RET(SendDeviceIpcNotifyReadyMesg());
689 : }
690 :
691 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
692 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) &&
693 0 : machinePara_.isAicpuModeEn == true) {
694 : /* 发送Device上使用的IPC notify ready信息 */
695 0 : CHK_RET(SendDeviceIpcNotifyDoneMesg());
696 : }
697 0 : return HCCL_SUCCESS;
698 : }
699 :
700 0 : HcclResult TransportBase::LinkRecvNotifyMesg()
701 : {
702 0 : s32 sendPid = 0;
703 0 : CHK_RET(SalGetBareTgid(&sendPid)); // 当前进程id
704 0 : HCCL_INFO("LinkRecvNotifyMesg, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
705 :
706 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
707 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
708 : /* 接收IPC ready 信息 */
709 0 : CHK_RET(RecvNotifyReadyMesg());
710 : }
711 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
712 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
713 : /* 接收IPC ready 信息 */
714 0 : CHK_RET(RecvNotifyDoneMesg());
715 : }
716 :
717 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
718 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) &&
719 0 : machinePara_.isAicpuModeEn == true) {
720 : /* 接收IPC ready 信息 */
721 0 : CHK_RET(RecvDeviceIpcNotifyReadyMesg());
722 : }
723 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
724 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) &&
725 0 : machinePara_.isAicpuModeEn == true) {
726 : /* 接收IPC ready 信息 */
727 0 : CHK_RET(RecvDeviceIpcNotifyDoneMesg());
728 : }
729 0 : return HCCL_SUCCESS;
730 : }
731 :
732 2 : HcclResult TransportBase::SetNotify()
733 : {
734 : HcclSignalInfo signalInfo;
735 2 : CHK_PTR_NULL(remoteSendReadyNotify_);
736 2 : CHK_PTR_NULL(remoteSendDoneNotify_);
737 2 : CHK_PTR_NULL(localSendReadyNotify_);
738 2 : CHK_PTR_NULL(localSendDoneNotify_);
739 :
740 2 : CHK_RET(remoteSendReadyNotify_->GetNotifyData(signalInfo));
741 2 : remoteSendReadyAddress_ = signalInfo.addr;
742 :
743 2 : CHK_RET(remoteSendDoneNotify_->GetNotifyData(signalInfo));
744 2 : remoteSendDoneAddress_ = signalInfo.addr;
745 :
746 2 : remoteSendReadyNotify_->GetNotifyOffset(remoteSendReadyOffset_);
747 2 : remoteSendDoneNotify_->GetNotifyOffset(remoteSendDoneOffset_);
748 :
749 2 : bool bRet = !(notifyNum_ == userLocalNotify_.size() && notifyNum_ == userRemoteNotify_.size());
750 2 : CHK_PRT_RET(bRet,
751 : HCCL_ERROR("[TransportBase][SetNotify]NotifyNumber of userLocalNotify_/userRemoteNotify_ doesn't equal to notifyNum_[%u]", \
752 : notifyNum_), HCCL_E_INTERNAL);
753 :
754 2 : for (u32 i = 0; i < notifyNum_; i++) {
755 0 : CHK_PTR_NULL(userLocalNotify_[i]);
756 0 : CHK_PTR_NULL(userRemoteNotify_[i]);
757 0 : CHK_RET(userRemoteNotify_[i]->GetNotifyData(signalInfo));
758 0 : userRemoteNotifyAddr_[i] = signalInfo.addr;
759 0 : userRemoteNotify_[i]->GetNotifyOffset(userRemoteNotifyOffset_[i]);
760 : }
761 :
762 2 : return HCCL_SUCCESS;
763 : }
764 :
765 5 : HcclResult TransportBase::SignalInit(const std::shared_ptr<LocalNotify> ¬ify,
766 : std::shared_ptr<LocalIpcNotify> &ipcNotify)
767 : {
768 5 : CHK_SMART_PTR_NULL(notify);
769 : HcclSignalInfo signalInfo;
770 5 : CHK_RET(notify->GetNotifyData(signalInfo));
771 5 : EXCEPTION_CATCH((ipcNotify = std::make_shared<LocalIpcNotify>()), return HCCL_E_PTR);
772 5 : CHK_RET(ipcNotify->Init(signalInfo, NotifyLoadType::DEVICE_NOTIFY));
773 5 : HCCL_INFO("%s notifyId_ [%u]", __func__, ipcNotify->notifyId_);
774 5 : return HCCL_SUCCESS;
775 : }
776 :
777 2 : HcclResult TransportBase::SetNotifyPtr(const TransportDeviceP2pData &transDevP2pData)
778 : {
779 2 : CHK_RET(SignalInit(transDevP2pData.ipcPreWaitNotify, localSendReadyNotify_));
780 2 : CHK_RET(SignalInit(transDevP2pData.ipcPostWaitNotify, localSendDoneNotify_));
781 2 : remoteSendReadyNotify_ = transDevP2pData.ipcPreRecordNotify;
782 2 : remoteSendDoneNotify_ = transDevP2pData.ipcPostRecordNotify;
783 :
784 : // 校验notifyNum_数量
785 6 : bool bRet = !(notifyNum_ == transDevP2pData.userLocalNotify.size() && notifyNum_ == transDevP2pData.userRemoteNotify.size() &&
786 4 : notifyNum_ == userLocalNotify_.size() && notifyNum_ == userRemoteNotify_.size());
787 2 : CHK_PRT_RET(bRet,
788 : HCCL_ERROR("[TransportBase][SetNotifyPtr]NotifyNum of userLocalNotify/userRemoteNotify doesn't equal to notifyNum_[%u]", \
789 : notifyNum_), HCCL_E_INTERNAL);
790 :
791 2 : for (u32 i = 0; i < notifyNum_; i++) {
792 0 : CHK_RET(SignalInit(transDevP2pData.userLocalNotify[i], userLocalNotify_[i]));
793 0 : userRemoteNotify_[i] = transDevP2pData.userRemoteNotify[i];
794 : }
795 :
796 2 : return HCCL_SUCCESS;
797 : }
798 :
799 4 : void TransportBase::DestroyHostSignal()
800 : {
801 4 : s32 sendPid = 0;
802 4 : SalGetBareTgid(&sendPid); // 当前进程id
803 4 : HCCL_INFO("SignalDestroy, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
804 :
805 4 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
806 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
807 4 : if ((remoteSendReadyNotify_ != nullptr)) {
808 2 : remoteSendReadyNotify_->Close();
809 2 : remoteSendReadyNotify_ = nullptr;
810 : }
811 : /* 销毁creat的signal资源 */
812 4 : localSendDoneNotify_ = nullptr;
813 : }
814 4 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
815 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
816 : /* 关闭open的signal资源, destroy支持close */
817 4 : if ((remoteSendDoneNotify_ != nullptr)) {
818 2 : remoteSendDoneNotify_->Close();
819 2 : remoteSendDoneNotify_ = nullptr;
820 : }
821 4 : localSendReadyNotify_ = nullptr;
822 : }
823 4 : }
824 :
825 4 : void TransportBase::DestroyDeviceSignal()
826 : {
827 4 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
828 0 : machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) &&
829 4 : machinePara_.isAicpuModeEn == true) {
830 0 : if ((remoteSendReadyDeviceNotify_ != nullptr)) {
831 0 : remoteSendReadyDeviceNotify_->Close();
832 0 : remoteSendReadyDeviceNotify_ = nullptr;
833 : }
834 : /* 销毁creat的signal资源 */
835 0 : localSendDoneDeviceNotify_ = nullptr;
836 : }
837 4 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE ||
838 0 : machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) &&
839 4 : machinePara_.isAicpuModeEn == true) {
840 : /* 关闭open的signal资源, destroy支持close */
841 0 : if ((remoteSendDoneDeviceNotify_ != nullptr)) {
842 0 : remoteSendDoneDeviceNotify_->Close();
843 0 : remoteSendDoneDeviceNotify_ = nullptr;
844 : }
845 0 : localSendReadyDeviceNotify_ = nullptr;
846 : }
847 4 : }
848 :
849 4 : void TransportBase::SignalDestroy()
850 : {
851 4 : DestroyHostSignal();
852 4 : DestroyDeviceSignal();
853 4 : }
854 :
855 1 : HcclResult TransportBase::GetTxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
856 : {
857 1 : CHK_SMART_PTR_NULL(remoteSendDoneDeviceNotify_);
858 1 : CHK_RET(remoteSendDoneDeviceNotify_->GetNotifyData(notifyInfo));
859 :
860 1 : return HCCL_SUCCESS;
861 : }
862 :
863 1 : HcclResult TransportBase::GetRxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
864 : {
865 1 : CHK_SMART_PTR_NULL(localSendDoneDeviceNotify_);
866 1 : CHK_RET(localSendDoneDeviceNotify_->GetNotifyData(notifyInfo));
867 :
868 1 : return HCCL_SUCCESS;
869 : }
870 :
871 1 : HcclResult TransportBase::GetTxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
872 : {
873 1 : CHK_SMART_PTR_NULL(remoteSendReadyDeviceNotify_);
874 1 : CHK_RET(remoteSendReadyDeviceNotify_->GetNotifyData(notifyInfo));
875 :
876 1 : return HCCL_SUCCESS;
877 : }
878 :
879 1 : HcclResult TransportBase::GetRxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
880 : {
881 1 : CHK_SMART_PTR_NULL(localSendReadyDeviceNotify_);
882 1 : CHK_RET(localSendReadyDeviceNotify_->GetNotifyData(notifyInfo));
883 :
884 1 : return HCCL_SUCCESS;
885 : }
886 :
887 2 : HcclResult TransportBase::ConstructExchangeDataForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
888 : {
889 2 : u64 dataLength = machinePara_.exchangeInfo.size();
890 2 : if (dataLength == 0) {
891 0 : HCCL_DEBUG("[Construct][ExchangeData]exchangeInfo size is 0.");
892 0 : return HCCL_SUCCESS;
893 : }
894 :
895 2 : HCCL_DEBUG("[Construct][ExchangeData]exchangeInfo size[%llu].", dataLength);
896 2 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &machinePara_.exchangeInfo[0], dataLength));
897 2 : exchangeDataPtr += dataLength;
898 2 : exchangeDataBlankSize -= dataLength;
899 2 : return HCCL_SUCCESS;
900 : }
901 :
902 2 : HcclResult TransportBase::ParseExchangeData(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
903 : {
904 2 : u64 dataLength = machinePara_.exchangeInfo.size();
905 2 : if (dataLength == 0) {
906 0 : HCCL_DEBUG("[Parse][ExchangeData]exchangeInfo size is 0.");
907 0 : return HCCL_SUCCESS;
908 : }
909 2 : exchangeMsg_.resize(dataLength);
910 2 : CHK_SAFETY_FUNC_RET(memcpy_s(&exchangeMsg_[0], exchangeMsg_.size(), exchangeDataPtr, dataLength));
911 2 : exchangeDataPtr += dataLength;
912 2 : exchangeDataBlankSize -= dataLength;
913 :
914 2 : return HCCL_SUCCESS;
915 : }
916 :
917 0 : HcclResult TransportBase::SendExchangeData(void)
918 : {
919 0 : u64 dataLength = machinePara_.exchangeInfo.size();
920 0 : if (dataLength == 0) {
921 0 : HCCL_DEBUG("[Send][ExchangeData]exchangeInfo size is 0.");
922 0 : return HCCL_SUCCESS;
923 : }
924 :
925 0 : HCCL_DEBUG("[Send][ExchangeData]exchangeInfo size[%llu].", dataLength);
926 0 : CHK_SMART_PTR_NULL(defaultSocket_);
927 0 : HcclResult ret = defaultSocket_->Send(machinePara_.exchangeInfo.data(), dataLength);
928 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
929 : HCCL_ERROR("[Send][ExchangeData]failed to send custom exchange data size [%llu].",
930 : dataLength), ret);
931 :
932 0 : return HCCL_SUCCESS;
933 : }
934 0 : HcclResult TransportBase::RecvAndCheckExchangeData(void)
935 : {
936 0 : u64 dataLength = machinePara_.exchangeInfo.size();
937 0 : if (dataLength == 0) {
938 0 : HCCL_DEBUG("[Check][ExchangeData]exchangeInfo size is 0.");
939 0 : return HCCL_SUCCESS;
940 : }
941 0 : exchangeMsg_.resize(dataLength);
942 :
943 0 : CHK_SMART_PTR_NULL(defaultSocket_);
944 0 : HcclResult ret = defaultSocket_->Recv(exchangeMsg_.data(), dataLength);
945 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
946 : HCCL_ERROR("[Check][ExchangeData]failed to recv custom exchange data size [%llu].",
947 : dataLength), ret);
948 :
949 0 : return HCCL_SUCCESS;
950 : }
951 :
952 0 : HcclResult TransportBase::OpenRemoteNotify(const std::vector<u8>& byteVector,
953 : std::shared_ptr<RemoteNotify> &remoteNotify)
954 : {
955 0 : EXCEPTION_CATCH((remoteNotify = std::make_shared<RemoteNotify>()), return HCCL_E_PTR);
956 0 : CHK_SMART_PTR_NULL(remoteNotify);
957 :
958 0 : HcclResult ret = HCCL_SUCCESS;
959 0 : bool errorFlag = false;
960 : do {
961 0 : ret = remoteNotify->Init(byteVector);
962 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[TransportBase][OpenRemoteNotify]remoteNotify init failed, "
963 : "ret[%d]", ret), errorFlag = true);
964 :
965 0 : ret = remoteNotify->Open();
966 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[TransportBase][OpenRemoteNotify]remoteNotify open failed, "
967 : "ret[%d]", ret), errorFlag = true);
968 : } while (0);
969 :
970 0 : if (errorFlag) {
971 0 : HCCL_ERROR("[TransportBase][OpenRemoteNotify]remoteNotify open failed ,ret[%d]", ret);
972 0 : remoteNotify = nullptr;
973 0 : return ret;
974 : }
975 0 : return HCCL_SUCCESS;
976 : }
977 :
978 0 : HcclResult TransportBase::PostReady(Stream &stream)
979 : {
980 : static_cast<void>(stream);
981 0 : return HCCL_SUCCESS;
982 : }
983 :
984 0 : HcclResult TransportBase::WaitReady(Stream &stream)
985 : {
986 : static_cast<void>(stream);
987 0 : return HCCL_SUCCESS;
988 : }
989 :
990 0 : HcclResult TransportBase::PostFin(Stream &stream)
991 : {
992 : static_cast<void>(stream);
993 0 : return HCCL_SUCCESS;
994 : }
995 :
996 0 : HcclResult TransportBase::WaitFin(Stream &stream)
997 : {
998 : static_cast<void>(stream);
999 0 : return HCCL_SUCCESS;
1000 : }
1001 :
1002 9 : HcclResult TransportBase::PostFinAck(Stream &stream)
1003 : {
1004 : static_cast<void>(stream);
1005 9 : return HCCL_SUCCESS;
1006 : }
1007 :
1008 9 : HcclResult TransportBase::WaitFinAck(Stream &stream)
1009 : {
1010 : static_cast<void>(stream);
1011 9 : return HCCL_SUCCESS;
1012 : }
1013 :
1014 0 : HcclResult TransportBase::SetStopFlag(bool value)
1015 : {
1016 0 : stopFlag_.store(value);
1017 0 : return HCCL_SUCCESS;
1018 : }
1019 :
1020 0 : bool TransportBase::GetStopFlag()
1021 : {
1022 0 : return stopFlag_.load();
1023 : }
1024 :
1025 0 : HcclResult TransportBase::UpdateRemoteAddr(void *remoteIn, void *remoteOut)
1026 : {
1027 : static_cast<void>(remoteIn);
1028 : static_cast<void>(remoteOut);
1029 0 : return HCCL_E_NOT_SUPPORT;
1030 : }
1031 :
1032 0 : HcclResult TransportBase::WriteAsync(
1033 : struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream)
1034 : {
1035 : static_cast<void>(remoteBuf);
1036 : static_cast<void>(localBuf);
1037 : static_cast<void>(stream);
1038 0 : return HCCL_E_NOT_SUPPORT;
1039 : }
1040 :
1041 0 : HcclResult TransportBase::WriteSync(
1042 : struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream)
1043 : {
1044 : static_cast<void>(remoteBuf);
1045 : static_cast<void>(localBuf);
1046 : static_cast<void>(stream);
1047 0 : return HCCL_E_NOT_SUPPORT;
1048 : }
1049 :
1050 0 : HcclResult TransportBase::WriteReduceAsync(struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf,
1051 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
1052 : {
1053 : static_cast<void>(remoteBuf);
1054 : static_cast<void>(localBuf);
1055 : static_cast<void>(datatype);
1056 : static_cast<void>(redOp);
1057 : static_cast<void>(stream);
1058 :
1059 0 : return HCCL_E_NOT_SUPPORT;
1060 : }
1061 :
1062 0 : HcclResult TransportBase::ReadAsync(
1063 : struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream)
1064 : {
1065 : static_cast<void>(localBuf);
1066 : static_cast<void>(remoteBuf);
1067 : static_cast<void>(stream);
1068 0 : return HCCL_E_NOT_SUPPORT;
1069 : }
1070 :
1071 0 : HcclResult TransportBase::ReadSync(
1072 : struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream)
1073 : {
1074 : static_cast<void>(localBuf);
1075 : static_cast<void>(remoteBuf);
1076 : static_cast<void>(stream);
1077 0 : return HCCL_E_NOT_SUPPORT;
1078 : }
1079 :
1080 0 : HcclResult TransportBase::ReadReduceSync(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf,
1081 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
1082 : {
1083 : static_cast<void>(remoteBuf);
1084 : static_cast<void>(localBuf);
1085 : static_cast<void>(datatype);
1086 : static_cast<void>(redOp);
1087 : static_cast<void>(stream);
1088 :
1089 0 : return HCCL_E_NOT_SUPPORT;
1090 : }
1091 :
1092 0 : HcclResult TransportBase::BatchTransferAsync(
1093 : const HcommBatchTransferDesc *transferDescs, uint32_t descNum, Stream &stream)
1094 : {
1095 : static_cast<void>(transferDescs);
1096 : static_cast<void>(descNum);
1097 : static_cast<void>(stream);
1098 0 : return HCCL_E_NOT_SUPPORT;
1099 : }
1100 :
1101 0 : HcclResult TransportBase::Fence()
1102 : {
1103 0 : return HCCL_E_NOT_SUPPORT;
1104 : }
1105 :
1106 0 : HcclResult TransportBase::Drain(Stream &stream)
1107 : {
1108 : static_cast<void>(stream);
1109 0 : return HCCL_E_NOT_SUPPORT;
1110 : }
1111 :
1112 0 : HcclResult TransportBase::InitDrainNotifyInfo()
1113 : {
1114 0 : return HCCL_E_NOT_SUPPORT;
1115 : }
1116 :
1117 0 : HcclResult TransportBase::GetDrainRemSrcMem(void* &remoteAddr, uint32_t &remoteKey, uint32_t &size)
1118 : {
1119 : (void)remoteAddr;
1120 : (void)remoteKey;
1121 : (void)size;
1122 0 : return HCCL_E_NOT_SUPPORT;
1123 : }
1124 : } // namespace hccl
|
