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