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 : #ifndef TRANSPORT_PUB_H
12 : #define TRANSPORT_PUB_H
13 :
14 : #include <initializer_list>
15 : #include <vector>
16 : #include <hccl/hccl_types.h>
17 : #include "hccl_common.h"
18 : #include "sal_pub.h"
19 : #include "adapter_pub.h"
20 : #include "stream_pub.h"
21 : #include "dispatcher.h"
22 : #include "hccl_socket.h"
23 : #include "notify_pool.h"
24 : #include "local_notify.h"
25 : #include "remote_notify.h"
26 : #include "hccl_mem_defs.h"
27 : #include "hcomm_primitives.h"
28 :
29 : enum class DBMode : s32 { INVALID_DB = -1, HW_DB = 0, SW_DB };
30 :
31 : struct HcclAiRMAWQ {
32 : u32 wqn;
33 : u64 bufAddr;
34 : u32 wqeSize;
35 : u32 depth;
36 : u64 headAddr;
37 : u64 tailAddr;
38 : DBMode dbMode; // 0-hw/1-sw
39 : u64 dbAddr;
40 : u32 sl;
41 0 : HcclAiRMAWQ()
42 0 : : wqn(0),
43 0 : bufAddr(0),
44 0 : wqeSize(0),
45 0 : depth(0),
46 0 : headAddr(0),
47 0 : tailAddr(0),
48 0 : dbMode(DBMode::INVALID_DB),
49 0 : dbAddr(0),
50 0 : sl(0)
51 0 : {}
52 : };
53 :
54 : struct HcclAiRMACQ {
55 : u32 cqn;
56 : u64 bufAddr;
57 : u32 cqeSize;
58 : u32 depth;
59 : u64 headAddr;
60 : u64 tailAddr;
61 : DBMode dbMode; // 0-hw/1-sw
62 : u64 dbAddr;
63 0 : HcclAiRMACQ()
64 0 : : cqn(0),
65 0 : bufAddr(0),
66 0 : cqeSize(0),
67 0 : depth(0),
68 0 : headAddr(0),
69 0 : tailAddr(0),
70 0 : dbMode(DBMode::INVALID_DB),
71 0 : dbAddr(0)
72 0 : {}
73 : };
74 :
75 : struct HcclAiRMAQueueInfo {
76 : struct HcclAiRMAWQ sq;
77 : struct HcclAiRMAWQ rq;
78 : struct HcclAiRMACQ scq;
79 : struct HcclAiRMACQ rcq;
80 : };
81 :
82 : #pragma pack(push)
83 : #pragma pack(4)
84 : struct HcclQpInfoV2 {
85 : u64 qpPtr;
86 : u32 sqIndex;
87 : u32 dbIndex;
88 : u16 retryCnt{0};
89 : u16 retryTime{0};
90 :
91 677 : HcclQpInfoV2() : qpPtr(0), sqIndex(0), dbIndex(0), retryCnt(0), retryTime(0) {}
92 12 : HcclQpInfoV2(const HcclQpInfoV2& other)
93 12 : : qpPtr(other.qpPtr),
94 12 : sqIndex(other.sqIndex),
95 12 : dbIndex(other.dbIndex),
96 12 : retryCnt(other.retryCnt),
97 12 : retryTime(other.retryTime)
98 12 : {}
99 : HcclQpInfoV2(HcclQpInfoV2&& other)
100 : : qpPtr(other.qpPtr),
101 : sqIndex(other.sqIndex),
102 : dbIndex(other.dbIndex),
103 : retryCnt(other.retryCnt),
104 : retryTime(other.retryTime)
105 : {}
106 6 : HcclQpInfoV2& operator=(const HcclQpInfoV2& other)
107 : {
108 6 : if (&other != this) {
109 6 : qpPtr = other.qpPtr;
110 6 : sqIndex = other.sqIndex;
111 6 : dbIndex = other.dbIndex;
112 6 : retryCnt = other.retryCnt;
113 6 : retryTime = other.retryTime;
114 : }
115 6 : return *this;
116 : }
117 99 : HcclQpInfoV2& operator=(HcclQpInfoV2&& other)
118 : {
119 99 : if (&other != this) {
120 99 : qpPtr = other.qpPtr;
121 99 : sqIndex = other.sqIndex;
122 99 : dbIndex = other.dbIndex;
123 99 : retryCnt = other.retryCnt;
124 99 : retryTime = other.retryTime;
125 : }
126 99 : return *this;
127 : }
128 : };
129 : #pragma pack(pop)
130 :
131 : struct AddrKey {
132 : u64 addr = 0;
133 : u32 key = 0;
134 : u32 notifyId = INVALID_UINT;
135 : };
136 :
137 : /**
138 : * AICPU TS RoCE：下发 device 的 MR 元数据；亦为 TransportDeviceIbverbs 区间查表 value。
139 : * addr：同 RmaBuffer::GetAddr()（HOST 为主机 VA，DEVICE 为设备 VA）。
140 : * devAddr：MR 用设备 VA（GetDevAddr()）；HOST 映射后与 addr 不同，DEVICE 时与 addr 相同。
141 : */
142 : struct RoceMemDetails {
143 : u64 addr = 0;
144 : u64 devAddr = 0;
145 : u64 size = 0;
146 : u32 key = 0;
147 : };
148 :
149 : struct MemDetails {
150 : u64 size = 0;
151 : u64 addr = 0;
152 : u32 key = 0;
153 :
154 622232 : MemDetails() {}
155 20 : MemDetails(const MemDetails& that) : size(that.size), addr(that.addr), key(that.key) {}
156 :
157 : MemDetails(MemDetails&& that) : size(that.size), addr(that.addr), key(that.key) {}
158 0 : MemDetails& operator=(const MemDetails& that)
159 : {
160 0 : if (&that != this) {
161 0 : size = that.size;
162 0 : addr = that.addr;
163 0 : key = that.key;
164 : }
165 0 : return *this;
166 : }
167 :
168 : MemDetails operator=(MemDetails&& that)
169 : {
170 : if (&that != this) {
171 : size = that.size;
172 : addr = that.addr;
173 : key = that.key;
174 : }
175 : return *this;
176 : }
177 : };
178 :
179 : namespace hccl {
180 :
181 : class TransportBase;
182 : struct RxMemoryInfo;
183 : struct TxMemoryInfo;
184 : enum class UserMemType;
185 : class DeviceMem;
186 : class MemNameRepository;
187 :
188 : enum class MachineType { MACHINE_SERVER_TYPE, MACHINE_CLIENT_TYPE, MACHINE_RESERVED_TYPE };
189 : enum class LinkMode { LINK_SIMPLEX_MODE, LINK_DUPLEX_MODE, LINK_RESERVED_MODE };
190 :
191 : // signal record 使用的value的内存信息
192 : // 使用SDMDA或者RDMA进行notify record时需要将该内存copy到远端 notify 寄存器
193 : using HcclSignalRecordBuff = struct HcclSignalRecordBuffDef {
194 : u64 address{0}; // signal 地址
195 : u64 length{0}; // signal 长度
196 : };
197 :
198 : constexpr u32 HCCL_TRANSPORT_RELATIONSHIP_SAME_CHIP = 0x1U << 0; // transport 的两端rank位于同一个NPU芯片内
199 : constexpr u32 HCCL_TRANSPORT_RELATIONSHIP_SAME_SERVER = 0x1U << 1; // transport 的两端rank位于同一个服务器内
200 : constexpr u32 HCCL_TRANSPORT_RELATIONSHIP_SAME_SUPERPOD = 0x1U << 2; // transport 的两端rank位于同一个超节点内
201 :
202 : // transport 使用的基础信息，包括链路类型、和远端的位置关系等
203 : using TransportAttr = struct TransportAttrDef {
204 : hccl::LinkType linkType{hccl::LinkType::LINK_RESERVED}; // 链路类型，HCCS,
205 : u32 relationship{0}; // 和remote的位置关系{同芯片，同节点，跨节点}
206 : HcclSignalRecordBuff signalRecordBuff;
207 : };
208 :
209 : constexpr u64 MAX_EXCHANGE_DATA_LEN = 2ULL * 1024 * 1024; // 自定义交换数据限制2MB
210 :
211 : // 传参数的时候都填充，link自己使用的时候区分
212 : // RDMA: machineType+serverId+local_rank_id+remote_rank_id+collectiveId
213 : // TCP: machineType+serverId+local_rank_id+remote_rank_id+collectiveId
214 : // PCIE: local_rank_id+remote_rank_id+collectiveId+localDeviceId+remoteDeviceId
215 : using MachinePara = struct TagMachinePara {
216 : public:
217 : MachineType machineType{MachineType::MACHINE_RESERVED_TYPE}; // client或者server
218 : LinkMode linkMode{LinkMode::LINK_RESERVED_MODE};
219 : std::string collectiveId{""}; // 本节点所在的通信域ID
220 : std::string tag{""};
221 : std::string serverId; // 本端server id
222 :
223 : HcclIpAddress localIpAddr; // 本端rank ip
224 : HcclIpAddress remoteIpAddr; // 对端rank ip
225 :
226 : u32 localSocketPort;
227 : u32 remoteSocketPort;
228 :
229 : s32 localDeviceId{-1}; // 本端device physical id
230 : s32 remoteDeviceId{-1}; // 对端device physical id
231 : s32 deviceLogicId{0};
232 :
233 : u32 localUserrank{INVALID_VALUE_RANKID}; // 本端user rank
234 : u32 remoteUserrank{INVALID_VALUE_RANKID}; // 对端user rank
235 :
236 : u32 localWorldRank{INVALID_VALUE_RANKID}; // 本端world group rank
237 : u32 remoteWorldRank{INVALID_VALUE_RANKID}; // 对端world group rank
238 :
239 : NICDeployment nicDeploy{NICDeployment::NIC_DEPLOYMENT_DEVICE};
240 : DevType deviceType{DevType::DEV_TYPE_COUNT};
241 :
242 : std::vector<std::shared_ptr<HcclSocket>> sockets;
243 : std::vector<u8> exchangeInfo; // 自定义交换数据,限制MAX_EXCHANGE_DATA_LEN = 2MB
244 :
245 : DeviceMem inputMem{DeviceMem()};
246 : DeviceMem outputMem{DeviceMem()};
247 : std::vector<DeviceMem> mem{};
248 :
249 : // 自定义算子交换内存
250 : std::vector<DeviceMem> userDeviceMem{};
251 : std::vector<HostMem> userHostMem{};
252 : bool isIndOp{false};
253 :
254 : // link特性位图: bit0:0x1支持WRITE操作（源端发起数据传输，优选项）
255 : // bit1:0x2支持READ操作（目的端发起数据传输）。如果同时支持link优先选用目的端发起数据传输
256 : u64 linkAttribute{0x1}; // 初始设置为WRITE操作，从源端发起数据传输;
257 :
258 : bool supportDataReceivedAck{false};
259 : bool isAicpuModeEn{false};
260 : std::vector<std::uint16_t> srcPorts; // 多qp配置的源端口号
261 : u32 notifyNum{0};
262 : QPMode qpMode{QPMode::INVALID}; // 是否为普通QP模式
263 : u32 tc{HCCL_COMM_TRAFFIC_CLASS_CONFIG_NOT_SET};
264 : u32 sl{HCCL_COMM_SERVICE_LEVEL_CONFIG_NOT_SET};
265 : LinkTypeInServer specifyLink{LinkTypeInServer::RESERVED_LINK_TYPE}; // 指定链路类型
266 : bool enableAtomicWrite{false}; // 使能atomicWrite
267 : QueueDepthAttr queueDepthAttr{}; // QP深度配置
268 : bool userMemEnable{true};
269 : bool drainEnable{false};
270 : // DispatcherCtxPtr；设备侧 TS Roce 等场景传入，WriteCommon 内写入线程局部 dispatcher
271 : void* dctxPtr{nullptr};
272 : bool isNewOneSide{false};
273 : u32 localBufSize{0};
274 : u32 remoteBufSize{0};
275 : HcclMemEx* localBufMem{nullptr};
276 : HcclMemEx* remoteBufMem{nullptr};
277 750 : TagMachinePara() {}
278 :
279 138 : TagMachinePara(const struct TagMachinePara& that)
280 690 : {
281 138 : machineType = (that.machineType);
282 138 : linkMode = (that.linkMode);
283 138 : serverId = (that.serverId);
284 138 : localIpAddr = (that.localIpAddr);
285 138 : remoteIpAddr = (that.remoteIpAddr);
286 138 : localDeviceId = (that.localDeviceId);
287 138 : remoteDeviceId = (that.remoteDeviceId);
288 138 : localUserrank = (that.localUserrank);
289 138 : remoteUserrank = (that.remoteUserrank);
290 138 : localWorldRank = (that.localWorldRank);
291 138 : remoteWorldRank = (that.remoteWorldRank);
292 138 : collectiveId = (that.collectiveId);
293 138 : deviceType = (that.deviceType);
294 138 : tag = (that.tag);
295 138 : inputMem = (that.inputMem);
296 138 : outputMem = (that.outputMem);
297 138 : mem = (that.mem);
298 138 : userDeviceMem = (that.userDeviceMem);
299 138 : userHostMem = (that.userHostMem);
300 138 : isIndOp = (that.isIndOp);
301 138 : linkAttribute = (that.linkAttribute);
302 138 : sockets = (that.sockets);
303 138 : exchangeInfo = (that.exchangeInfo);
304 138 : supportDataReceivedAck = (that.supportDataReceivedAck);
305 138 : nicDeploy = (that.nicDeploy);
306 138 : localSocketPort = that.localSocketPort;
307 138 : remoteSocketPort = that.remoteSocketPort;
308 138 : isAicpuModeEn = that.isAicpuModeEn;
309 138 : deviceLogicId = that.deviceLogicId;
310 138 : srcPorts = that.srcPorts;
311 138 : notifyNum = that.notifyNum;
312 138 : qpMode = that.qpMode;
313 138 : tc = that.tc;
314 138 : sl = that.sl;
315 138 : specifyLink = that.specifyLink;
316 138 : enableAtomicWrite = that.enableAtomicWrite;
317 138 : queueDepthAttr = that.queueDepthAttr;
318 138 : userMemEnable = that.userMemEnable;
319 138 : drainEnable = that.drainEnable;
320 138 : dctxPtr = that.dctxPtr;
321 138 : isNewOneSide = (that.isNewOneSide);
322 138 : localBufSize = (that.localBufSize);
323 138 : remoteBufSize = (that.remoteBufSize);
324 138 : localBufMem = (that.localBufMem);
325 138 : remoteBufMem = (that.remoteBufMem);
326 138 : }
327 :
328 : struct TagMachinePara& operator=(struct TagMachinePara& that)
329 : {
330 : if (&that != this) {
331 : machineType = (that.machineType);
332 : linkMode = (that.linkMode);
333 : serverId = (that.serverId);
334 : localIpAddr = (that.localIpAddr);
335 : remoteIpAddr = (that.remoteIpAddr);
336 : localDeviceId = (that.localDeviceId);
337 : remoteDeviceId = (that.remoteDeviceId);
338 : localUserrank = (that.localUserrank);
339 : remoteUserrank = (that.remoteUserrank);
340 : localWorldRank = (that.localWorldRank);
341 : remoteWorldRank = (that.remoteWorldRank);
342 : collectiveId = (that.collectiveId);
343 : deviceType = (that.deviceType);
344 : tag = (that.tag);
345 : inputMem = (that.inputMem);
346 : outputMem = (that.outputMem);
347 : mem = (that.mem);
348 : userDeviceMem = (that.userDeviceMem);
349 : userHostMem = (that.userHostMem);
350 : isIndOp = (that.isIndOp);
351 : linkAttribute = (that.linkAttribute);
352 : sockets = (that.sockets);
353 : exchangeInfo = (that.exchangeInfo);
354 : supportDataReceivedAck = (that.supportDataReceivedAck);
355 : localSocketPort = that.localSocketPort;
356 : remoteSocketPort = that.remoteSocketPort;
357 : isAicpuModeEn = that.isAicpuModeEn;
358 : deviceLogicId = that.deviceLogicId;
359 : srcPorts = that.srcPorts;
360 : notifyNum = that.notifyNum;
361 : qpMode = that.qpMode;
362 : tc = that.tc;
363 : sl = that.sl;
364 : specifyLink = that.specifyLink;
365 : enableAtomicWrite = that.enableAtomicWrite;
366 : queueDepthAttr = that.queueDepthAttr;
367 : userMemEnable = that.userMemEnable;
368 : drainEnable = that.drainEnable;
369 : dctxPtr = that.dctxPtr;
370 : isNewOneSide = (that.isNewOneSide);
371 : localBufSize = (that.localBufSize);
372 : remoteBufSize = (that.remoteBufSize);
373 : localBufMem = (that.localBufMem);
374 : remoteBufMem = (that.remoteBufMem);
375 : }
376 :
377 : return *this;
378 : }
379 : };
380 :
381 : struct TransportPara {
382 : std::chrono::milliseconds timeout;
383 : NICDeployment nicDeploy;
384 : u32 localDieID;
385 : u32 dstDieID;
386 : HcclIpAddress* selfIp;
387 : HcclIpAddress* peerIp;
388 : u32 peerPort;
389 : u32 selfPort;
390 : u32 index;
391 : bool isRootRank;
392 : u32 devLogicId;
393 : u32 proxyDevLogicId;
394 : s32 qpMode = 0;
395 : bool isHdcMode = false;
396 : bool remoteIsHdc = false;
397 : bool isESPs = false;
398 : bool virtualFlag = false;
399 : };
400 :
401 : struct TransportDeviceNormalData {
402 : MemDetails remoteInputMem{};
403 : MemDetails remoteOutputMem{};
404 : MemDetails localInputMem{};
405 : MemDetails localOutputMem{};
406 : struct HcclQpInfoV2 qpInfo {};
407 : QPMode qpMode{QPMode::INVALID};
408 0 : void Print()
409 : {
410 0 : HCCL_DEBUG(
411 : "remoteInputMem: addr[%llu], size[%llu], key[%u]", remoteInputMem.addr, remoteInputMem.size,
412 : remoteInputMem.key);
413 0 : HCCL_DEBUG(
414 : "remoteOutputMem: addr[%llu], size[%llu], key[%u]", remoteOutputMem.addr, remoteOutputMem.size,
415 : remoteOutputMem.key);
416 0 : HCCL_DEBUG(
417 : "remoteInputMem: addr[%llu], size[%llu], key[%u]", localInputMem.addr, localInputMem.size,
418 : localInputMem.key);
419 0 : HCCL_DEBUG(
420 : "remoteOutputMem: addr[%llu], size[%llu], key[%u]", localOutputMem.addr, localOutputMem.size,
421 : localOutputMem.key);
422 0 : HCCL_DEBUG("qpInfo: qpPtr[%llu], sqIndex[%u], dbIndex[%u]", qpInfo.qpPtr, qpInfo.sqIndex, qpInfo.dbIndex);
423 0 : HCCL_DEBUG("qpMode[%d]", static_cast<int32_t>(qpMode));
424 0 : return;
425 : }
426 : };
427 :
428 : struct TransportDeviceP2pData {
429 : void* inputBufferPtr;
430 : void* outputBufferPtr;
431 : std::shared_ptr<LocalNotify> ipcPreWaitNotify;
432 : std::shared_ptr<LocalNotify> ipcPostWaitNotify;
433 : std::vector<std::shared_ptr<LocalNotify>> userLocalNotify;
434 : std::shared_ptr<RemoteNotify> ipcPreRecordNotify;
435 : std::shared_ptr<RemoteNotify> ipcPostRecordNotify;
436 : std::vector<std::shared_ptr<RemoteNotify>> userRemoteNotify;
437 : TransportAttr transportAttr;
438 :
439 23 : TransportDeviceP2pData() {}
440 : TransportDeviceP2pData(
441 : void* inputBufferPtr, void* outputBufferPtr, std::shared_ptr<LocalNotify> ipcPreWaitNotify,
442 : std::shared_ptr<LocalNotify> ipcPostWaitNotify, std::vector<std::shared_ptr<LocalNotify>> userLocalNotify,
443 : std::shared_ptr<RemoteNotify> ipcPreRecordNotify, std::shared_ptr<RemoteNotify> ipcPostRecordNotify,
444 : std::vector<std::shared_ptr<RemoteNotify>> userRemoteNotify, TransportAttr& transportAttr)
445 : : inputBufferPtr(inputBufferPtr),
446 : outputBufferPtr(outputBufferPtr),
447 : ipcPreWaitNotify(ipcPreWaitNotify),
448 : ipcPostWaitNotify(ipcPostWaitNotify),
449 : userLocalNotify(userLocalNotify),
450 : ipcPreRecordNotify(ipcPreRecordNotify),
451 : ipcPostRecordNotify(ipcPostRecordNotify),
452 : userRemoteNotify(userRemoteNotify),
453 : transportAttr(transportAttr)
454 : {}
455 : };
456 :
457 : struct TransportDeviceIbverbsData {
458 : void* inputBufferPtr;
459 : void* outputBufferPtr;
460 : MemDetails localInputMem;
461 : MemDetails localOutputMem;
462 : std::shared_ptr<LocalNotify> ackNotify;
463 : std::shared_ptr<LocalNotify> dataAckNotify;
464 : std::shared_ptr<LocalNotify> dataNotify;
465 : std::vector<std::vector<std::shared_ptr<LocalNotify>>> userLocalNotify;
466 : uint64_t localNotifyValueAddr;
467 : AddrKey remoteAckNotifyDetails;
468 : AddrKey remoteDataNotifyDetails;
469 : AddrKey remoteDataAckNotifyDetails;
470 : std::vector<std::vector<AddrKey>> userRemoteNotifyDetails;
471 : uint32_t notifyValueKey;
472 : std::vector<struct HcclQpInfoV2> qpInfo;
473 : uint32_t remoteInputKey;
474 : uint32_t remoteOutputKey;
475 : uint32_t notifySize;
476 : u32 multiQpThreshold;
477 : u32 qpsPerConnection;
478 : bool useAtomicWrite = false;
479 : std::vector<RoceMemDetails> localRoceMemDetailsList;
480 : std::vector<RoceMemDetails> remoteRoceMemDetailsList;
481 : bool useMemDetailsMgr{false};
482 : uint64_t remoteNotifyValueAddr{0}; // 对端 NOTIFY_SRC_MEM 地址 (Fence Read 读取源)
483 : uint32_t remoteNotifyValueKey{0}; // 对端 notify rkey
484 : uint64_t localDataNotifyAddr{0}; // 本端dataNotify addr
485 : uint32_t localDataNotifyKey{0}; // 本端dataNotifyKey
486 30 : TransportDeviceIbverbsData() {}
487 : TransportDeviceIbverbsData(
488 : void* inputBufferPtr, void* outputBufferPtr, MemDetails localInputMem, MemDetails localOutputMem,
489 : std::shared_ptr<LocalNotify> ackNotify, std::shared_ptr<LocalNotify> dataAckNotify,
490 : std::shared_ptr<LocalNotify> dataNotify, std::vector<std::vector<std::shared_ptr<LocalNotify>>> userLocalNotify,
491 : uint64_t localNotifyValueAddr, AddrKey remoteAckNotifyDetails, AddrKey remoteDataNotifyDetails,
492 : AddrKey remoteDataAckNotifyDetails, std::vector<std::vector<AddrKey>> userRemoteNotifyDetails,
493 : uint32_t notifyValueKey, std::vector<struct HcclQpInfoV2> qpInfo, uint32_t remoteInputKey,
494 : uint32_t remoteOutputKey, uint32_t notifySize, u32 multiQpThreshold, u32 qpsPerConnection, bool useAtomicWrite,
495 : uint64_t remoteNotifyValueAddr, uint32_t remoteNotifyValueKey, uint64_t localDataNotifyAddr,
496 : uint32_t localDataNotifyKey)
497 : : inputBufferPtr(inputBufferPtr),
498 : outputBufferPtr(outputBufferPtr),
499 : localInputMem(localInputMem),
500 : localOutputMem(localOutputMem),
501 : ackNotify(ackNotify),
502 : dataAckNotify(dataAckNotify),
503 : dataNotify(dataNotify),
504 : userLocalNotify(userLocalNotify),
505 : localNotifyValueAddr(localNotifyValueAddr),
506 : remoteAckNotifyDetails(remoteAckNotifyDetails),
507 : remoteDataNotifyDetails(remoteDataNotifyDetails),
508 : remoteDataAckNotifyDetails(remoteDataAckNotifyDetails),
509 : userRemoteNotifyDetails(userRemoteNotifyDetails),
510 : notifyValueKey(notifyValueKey),
511 : qpInfo(qpInfo),
512 : remoteInputKey(remoteInputKey),
513 : remoteOutputKey(remoteOutputKey),
514 : notifySize(notifySize),
515 : multiQpThreshold(multiQpThreshold),
516 : qpsPerConnection(qpsPerConnection),
517 : useAtomicWrite(useAtomicWrite),
518 : remoteNotifyValueAddr(remoteNotifyValueAddr),
519 : remoteNotifyValueKey(remoteNotifyValueKey),
520 : localDataNotifyAddr(localDataNotifyAddr),
521 : localDataNotifyKey(localDataNotifyKey)
522 : {}
523 :
524 10 : TransportDeviceIbverbsData(const TransportDeviceIbverbsData& that)
525 10 : : inputBufferPtr(that.inputBufferPtr),
526 10 : outputBufferPtr(that.outputBufferPtr),
527 10 : localInputMem(that.localInputMem),
528 10 : localOutputMem(that.localOutputMem),
529 10 : ackNotify(that.ackNotify),
530 10 : dataAckNotify(that.dataAckNotify),
531 10 : dataNotify(that.dataNotify),
532 10 : userLocalNotify(that.userLocalNotify),
533 10 : localNotifyValueAddr(that.localNotifyValueAddr),
534 10 : remoteAckNotifyDetails(that.remoteAckNotifyDetails),
535 10 : remoteDataNotifyDetails(that.remoteDataNotifyDetails),
536 10 : remoteDataAckNotifyDetails(that.remoteDataAckNotifyDetails),
537 10 : userRemoteNotifyDetails(that.userRemoteNotifyDetails),
538 10 : notifyValueKey(that.notifyValueKey),
539 10 : qpInfo(that.qpInfo),
540 10 : remoteInputKey(that.remoteInputKey),
541 10 : remoteOutputKey(that.remoteOutputKey),
542 10 : notifySize(that.notifySize),
543 10 : multiQpThreshold(that.multiQpThreshold),
544 10 : qpsPerConnection(that.qpsPerConnection),
545 10 : useAtomicWrite(that.useAtomicWrite),
546 10 : localRoceMemDetailsList(that.localRoceMemDetailsList),
547 10 : remoteRoceMemDetailsList(that.remoteRoceMemDetailsList),
548 10 : useMemDetailsMgr(that.useMemDetailsMgr),
549 10 : remoteNotifyValueAddr(that.remoteNotifyValueAddr),
550 10 : remoteNotifyValueKey(that.remoteNotifyValueKey),
551 10 : localDataNotifyAddr(that.localDataNotifyAddr),
552 10 : localDataNotifyKey(that.localDataNotifyKey)
553 10 : {}
554 : };
555 3 : using CqeInfo = struct tagCqeInfo {
556 : struct timeval time;
557 : uint32_t status = 0;
558 : HcclIpAddress remoteIp;
559 : char reserved[32];
560 7 : tagCqeInfo() {}
561 1 : tagCqeInfo(const struct timeval& time, const uint32_t status, const HcclIpAddress& remoteIp)
562 1 : : time(time),
563 1 : status(status),
564 1 : remoteIp(remoteIp)
565 1 : {}
566 : };
567 :
568 : class Transport {
569 : public:
570 : struct Buffer {
571 : const void* addr{nullptr};
572 : u64 size{0};
573 :
574 6 : Buffer() : addr(nullptr), size(0) {}
575 0 : Buffer(const void* addr, u64 size) : addr(addr), size(size) {}
576 : };
577 :
578 3 : Transport() {};
579 92 : explicit Transport(TransportBase* pimpl) : pimpl_(pimpl) {};
580 : Transport(
581 : TransportType type, TransportPara& para, const HcclDispatcher dispatcher,
582 : const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
583 : const TransportDeviceP2pData& transDevP2pData = TransportDeviceP2pData(),
584 : const TransportDeviceIbverbsData& transDevIbverbsData = TransportDeviceIbverbsData());
585 :
586 : ~Transport();
587 :
588 : HcclResult Stop();
589 : HcclResult Resume();
590 : HcclResult Init();
591 : HcclResult DeInit();
592 :
593 : HcclResult TxDataSignal(Stream& stream);
594 : HcclResult RxDataSignal(Stream& stream);
595 :
596 : HcclResult TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream);
597 : HcclResult TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream);
598 :
599 : HcclResult TxWithReduce(
600 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, const HcclDataType datatype,
601 : HcclReduceOp redOp, Stream& stream);
602 : HcclResult TxWithReduce(
603 : const std::vector<TxMemoryInfo>& txWithReduceMems, const HcclDataType datatype, HcclReduceOp redOp,
604 : Stream& stream);
605 : HcclResult RxWithReduce(
606 : UserMemType recvSrcMemType, u64 recvSrcOffset, void* recvDst, u64 recvLen, void* reduceSrc, void* reduceDst,
607 : u64 reduceDataCount, HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream& stream, const u64 reduceAttr);
608 : HcclResult RxWithReduce(
609 : const std::vector<RxWithReduceMemoryInfo>& rxWithReduceMems, HcclDataType reduceDatatype, HcclReduceOp reduceOp,
610 : Stream& stream, const u64 reduceAttr);
611 : bool IsSupportTransportWithReduce();
612 :
613 : HcclResult RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream);
614 : HcclResult RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream);
615 : HcclResult DataReceivedAck(Stream& stream);
616 :
617 : HcclResult TxAck(Stream& stream);
618 : HcclResult RxAck(Stream& stream);
619 :
620 : HcclResult TxPrepare(Stream& stream);
621 : HcclResult RxPrepare(Stream& stream);
622 :
623 : HcclResult TxDone(Stream& stream);
624 : HcclResult RxDone(Stream& stream);
625 :
626 : HcclResult TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream);
627 : HcclResult RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream);
628 :
629 : // 保证send语义完成
630 : HcclResult TxWaitDone(Stream& stream);
631 : // 保证recv语义完成
632 : HcclResult RxWaitDone(Stream& stream);
633 : // TxWaitDone、RxWaitDone共同出现保证sendrecv语义完成
634 :
635 : HcclResult Post(u32 notifyIdx, Stream& stream);
636 : HcclResult Wait(u32 notifyIdx, Stream& stream, const u32 timeOut = NOTIFY_INVALID_WAIT_TIME);
637 :
638 : u32 GetNotifyNum();
639 : HcclResult GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t* memNum, HcclMemType memType);
640 : HcclResult GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum);
641 : HcclResult GetLocalNotify(std::vector<HcclSignalInfo>& localNotify);
642 : HcclResult GetRemoteNotify(std::vector<HcclSignalInfo>& localNotify);
643 : HcclResult GetRemoteMem(UserMemType memType, void** remotePtr);
644 : HcclResult GetRemoteMem(std::vector<void*>* remotePtrVec);
645 : HcclResult GetRemoteMemKey(UserMemType memType, uint32_t* remoteMemKey);
646 : HcclResult GetLocalRdmaNotify(std::vector<HcclSignalInfo>& rdmaNotify);
647 : HcclResult GetDrainLocalDataNotify(void*& localAddr, uint32_t& lkey, HcclSignalInfo& dataNotify);
648 : HcclResult GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey>& rdmaNotifyAddr);
649 : HcclResult GetLocalNotifyValueAddrKey(std::vector<AddrKey>& notifyValue);
650 : HcclResult GetLocalMemDetails(UserMemType memType, MemDetails& memDetails);
651 : HcclResult GetAiQpInfo(std::vector<HcclQpInfoV2>& aiQpInfo);
652 : HcclResult GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo>& aiRMAQueueInfo);
653 : HcclResult GetTransportId(u32& id);
654 : HcclResult GetChipId(s64& chipId);
655 : virtual HcclResult GetRemoteMemSize(UserMemType memType, u64& size);
656 : HcclResult GetTxAckDevNotifyInfo(HcclSignalInfo& notifyInfo);
657 : HcclResult GetRxAckDevNotifyInfo(HcclSignalInfo& notifyInfo);
658 : HcclResult GetTxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo);
659 : HcclResult GetRxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo);
660 :
661 : hccl::LinkType GetLinkType() const;
662 : bool IsSpInlineReduce() const;
663 : bool GetSupportDataReceivedAck() const;
664 : void SetSupportDataReceivedAck(bool supportDataReceivedAck);
665 : u32 GetRemoteRank();
666 :
667 : HcclResult ConnectAsync(u32& status);
668 : HcclResult ConnectQuerry(u32& status);
669 : void Break();
670 :
671 : void EnableUseOneDoorbell();
672 :
673 : bool GetUseOneDoorbellValue();
674 :
675 : HcclResult GetTransportAttr(TransportAttr& attr);
676 :
677 : HcclResult TxEnv(const void* ptr, const u64 len, Stream& stream);
678 : HcclResult RxEnv(Stream& stream);
679 : bool IsTransportRoce();
680 :
681 : HcclResult WriteAsync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream);
682 : HcclResult WriteSync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream);
683 :
684 : HcclResult WriteReduceAsync(
685 : struct Buffer& remoteBuf, struct Buffer& localBuf, const HcclDataType datatype, HcclReduceOp redOp,
686 : Stream& stream);
687 :
688 : HcclResult ReadAsync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream);
689 : HcclResult ReadSync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream);
690 : HcclResult ReadReduceSync(
691 : struct Buffer& localBuf, struct Buffer& remoteBuf, const HcclDataType datatype, HcclReduceOp redOp,
692 : Stream& stream);
693 :
694 : HcclResult BatchTransferAsync(const HcommBatchTransferDesc* transferDescs, uint32_t descNum, Stream& stream);
695 :
696 : HcclResult PostReady(Stream& stream);
697 : HcclResult WaitReady(Stream& stream);
698 :
699 : HcclResult PostFin(Stream& stream);
700 : HcclResult WaitFin(Stream& stream);
701 :
702 : HcclResult PostFinAck(Stream& stream);
703 : HcclResult WaitFinAck(Stream& stream);
704 :
705 : HcclResult SetStopFlag(bool value);
706 : HcclResult Fence();
707 : HcclResult UpdateRemoteAddr(void* remoteIn, void* remoteOut);
708 : HcclResult Drain(Stream& stream);
709 : HcclResult InitDrainNotifyInfo();
710 : HcclResult GetDrainRemSrcMem(void*& remoteAddr, uint32_t& remoteKey, uint32_t& size);
711 : static HcclResult
712 : GetTransportErrorCqe(const HcclNetDevCtx netDevCtx, std::vector<std::pair<Transport*, CqeInfo>>& infos, u32& num);
713 1 : inline TransportType GetTransportType() const { return type_; }
714 :
715 : std::vector<u8> GetExchangeInfo();
716 : static HcclResult HcclBatchRead(
717 : const TransportDeviceNormalData& ibvData, struct MemDetails* localMems, struct MemDetails* remoteMems,
718 : u32 memNum, u64& dbInfo);
719 : static HcclResult HcclBatchWrite(
720 : const TransportDeviceNormalData& ibvData, struct MemDetails* localMems, struct MemDetails* remoteMems,
721 : u32 memNum, u64& dbInfo);
722 : static HcclResult SetDeviceUnavailable(u32 deviceId);
723 :
724 : bool GetIsUseAtomicWrite();
725 :
726 : HcclResult GetSpecificNotify(HcclSignalInfo& notifyInfo, bool& isValid, const std::string& notifyName);
727 :
728 : private:
729 : void CreateTransportRoce(
730 : TransportType type, TransportPara& para, const HcclDispatcher dispatcherPtr,
731 : const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara);
732 : TransportBase* pimpl_ = nullptr;
733 : const TransportType type_ = TransportType::TRANS_TYPE_RESERVED;
734 :
735 : static std::mutex mapMutex_;
736 : static std::unordered_map<TransportBase*, Transport*> transportMap_;
737 : };
738 :
739 : using LINK = std::shared_ptr<Transport>;
740 : } // namespace hccl
741 :
742 : #endif /* TRANSPORT_BASE_H */
|
