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