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