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 : u32 retryCnt{INVALID_UINT};
266 : u32 retryInterval{INVALID_UINT};
267 : LinkTypeInServer specifyLink{LinkTypeInServer::RESERVED_LINK_TYPE}; // 指定链路类型
268 : bool enableAtomicWrite{false}; // 使能atomicWrite
269 : QueueDepthAttr queueDepthAttr{}; // QP深度配置
270 : bool userMemEnable{true};
271 : bool drainEnable{false};
272 : // DispatcherCtxPtr；设备侧 TS Roce 等场景传入，WriteCommon 内写入线程局部 dispatcher
273 : void* dctxPtr{nullptr};
274 : bool isNewOneSide{false};
275 : u32 localBufSize{0};
276 : u32 remoteBufSize{0};
277 : HcclMemEx* localBufMem{nullptr};
278 : HcclMemEx* remoteBufMem{nullptr};
279 747 : TagMachinePara() {}
280 :
281 138 : TagMachinePara(const struct TagMachinePara& that)
282 690 : {
283 138 : machineType = (that.machineType);
284 138 : linkMode = (that.linkMode);
285 138 : serverId = (that.serverId);
286 138 : localIpAddr = (that.localIpAddr);
287 138 : remoteIpAddr = (that.remoteIpAddr);
288 138 : localDeviceId = (that.localDeviceId);
289 138 : remoteDeviceId = (that.remoteDeviceId);
290 138 : localUserrank = (that.localUserrank);
291 138 : remoteUserrank = (that.remoteUserrank);
292 138 : localWorldRank = (that.localWorldRank);
293 138 : remoteWorldRank = (that.remoteWorldRank);
294 138 : collectiveId = (that.collectiveId);
295 138 : deviceType = (that.deviceType);
296 138 : tag = (that.tag);
297 138 : inputMem = (that.inputMem);
298 138 : outputMem = (that.outputMem);
299 138 : mem = (that.mem);
300 138 : userDeviceMem = (that.userDeviceMem);
301 138 : userHostMem = (that.userHostMem);
302 138 : isIndOp = (that.isIndOp);
303 138 : linkAttribute = (that.linkAttribute);
304 138 : sockets = (that.sockets);
305 138 : exchangeInfo = (that.exchangeInfo);
306 138 : supportDataReceivedAck = (that.supportDataReceivedAck);
307 138 : nicDeploy = (that.nicDeploy);
308 138 : localSocketPort = that.localSocketPort;
309 138 : remoteSocketPort = that.remoteSocketPort;
310 138 : isAicpuModeEn = that.isAicpuModeEn;
311 138 : deviceLogicId = that.deviceLogicId;
312 138 : srcPorts = that.srcPorts;
313 138 : notifyNum = that.notifyNum;
314 138 : qpMode = that.qpMode;
315 138 : tc = that.tc;
316 138 : sl = that.sl;
317 138 : retryCnt = that.retryCnt;
318 138 : retryInterval = that.retryInterval;
319 138 : specifyLink = that.specifyLink;
320 138 : enableAtomicWrite = that.enableAtomicWrite;
321 138 : queueDepthAttr = that.queueDepthAttr;
322 138 : userMemEnable = that.userMemEnable;
323 138 : drainEnable = that.drainEnable;
324 138 : dctxPtr = that.dctxPtr;
325 138 : isNewOneSide = (that.isNewOneSide);
326 138 : localBufSize = (that.localBufSize);
327 138 : remoteBufSize = (that.remoteBufSize);
328 138 : localBufMem = (that.localBufMem);
329 138 : remoteBufMem = (that.remoteBufMem);
330 138 : }
331 :
332 : struct TagMachinePara& operator=(struct TagMachinePara& that)
333 : {
334 : if (&that != this) {
335 : machineType = (that.machineType);
336 : linkMode = (that.linkMode);
337 : serverId = (that.serverId);
338 : localIpAddr = (that.localIpAddr);
339 : remoteIpAddr = (that.remoteIpAddr);
340 : localDeviceId = (that.localDeviceId);
341 : remoteDeviceId = (that.remoteDeviceId);
342 : localUserrank = (that.localUserrank);
343 : remoteUserrank = (that.remoteUserrank);
344 : localWorldRank = (that.localWorldRank);
345 : remoteWorldRank = (that.remoteWorldRank);
346 : collectiveId = (that.collectiveId);
347 : deviceType = (that.deviceType);
348 : tag = (that.tag);
349 : inputMem = (that.inputMem);
350 : outputMem = (that.outputMem);
351 : mem = (that.mem);
352 : userDeviceMem = (that.userDeviceMem);
353 : userHostMem = (that.userHostMem);
354 : isIndOp = (that.isIndOp);
355 : linkAttribute = (that.linkAttribute);
356 : sockets = (that.sockets);
357 : exchangeInfo = (that.exchangeInfo);
358 : supportDataReceivedAck = (that.supportDataReceivedAck);
359 : localSocketPort = that.localSocketPort;
360 : remoteSocketPort = that.remoteSocketPort;
361 : isAicpuModeEn = that.isAicpuModeEn;
362 : deviceLogicId = that.deviceLogicId;
363 : srcPorts = that.srcPorts;
364 : notifyNum = that.notifyNum;
365 : qpMode = that.qpMode;
366 : tc = that.tc;
367 : sl = that.sl;
368 : retryCnt = that.retryCnt;
369 : retryInterval = that.retryInterval;
370 : specifyLink = that.specifyLink;
371 : enableAtomicWrite = that.enableAtomicWrite;
372 : queueDepthAttr = that.queueDepthAttr;
373 : userMemEnable = that.userMemEnable;
374 : drainEnable = that.drainEnable;
375 : dctxPtr = that.dctxPtr;
376 : isNewOneSide = (that.isNewOneSide);
377 : localBufSize = (that.localBufSize);
378 : remoteBufSize = (that.remoteBufSize);
379 : localBufMem = (that.localBufMem);
380 : remoteBufMem = (that.remoteBufMem);
381 : }
382 :
383 : return *this;
384 : }
385 : };
386 :
387 : struct TransportPara {
388 : std::chrono::milliseconds timeout;
389 : NICDeployment nicDeploy;
390 : u32 localDieID;
391 : u32 dstDieID;
392 : HcclIpAddress* selfIp;
393 : HcclIpAddress* peerIp;
394 : u32 peerPort;
395 : u32 selfPort;
396 : u32 index;
397 : bool isRootRank;
398 : u32 devLogicId;
399 : u32 proxyDevLogicId;
400 : s32 qpMode = 0;
401 : bool isHdcMode = false;
402 : bool remoteIsHdc = false;
403 : bool isESPs = false;
404 : bool virtualFlag = false;
405 : };
406 :
407 : struct TransportDeviceNormalData {
408 : MemDetails remoteInputMem{};
409 : MemDetails remoteOutputMem{};
410 : MemDetails localInputMem{};
411 : MemDetails localOutputMem{};
412 : struct HcclQpInfoV2 qpInfo {};
413 : QPMode qpMode{QPMode::INVALID};
414 0 : void Print()
415 : {
416 0 : HCCL_DEBUG(
417 : "remoteInputMem: addr[%llu], size[%llu], key[%u]", remoteInputMem.addr, remoteInputMem.size,
418 : remoteInputMem.key);
419 0 : HCCL_DEBUG(
420 : "remoteOutputMem: addr[%llu], size[%llu], key[%u]", remoteOutputMem.addr, remoteOutputMem.size,
421 : remoteOutputMem.key);
422 0 : HCCL_DEBUG(
423 : "remoteInputMem: addr[%llu], size[%llu], key[%u]", localInputMem.addr, localInputMem.size,
424 : localInputMem.key);
425 0 : HCCL_DEBUG(
426 : "remoteOutputMem: addr[%llu], size[%llu], key[%u]", localOutputMem.addr, localOutputMem.size,
427 : localOutputMem.key);
428 0 : HCCL_DEBUG("qpInfo: qpPtr[%llu], sqIndex[%u], dbIndex[%u]", qpInfo.qpPtr, qpInfo.sqIndex, qpInfo.dbIndex);
429 0 : HCCL_DEBUG("qpMode[%d]", static_cast<int32_t>(qpMode));
430 0 : return;
431 : }
432 : };
433 :
434 : struct TransportDeviceP2pData {
435 : void* inputBufferPtr;
436 : void* outputBufferPtr;
437 : std::shared_ptr<LocalNotify> ipcPreWaitNotify;
438 : std::shared_ptr<LocalNotify> ipcPostWaitNotify;
439 : std::vector<std::shared_ptr<LocalNotify>> userLocalNotify;
440 : std::shared_ptr<RemoteNotify> ipcPreRecordNotify;
441 : std::shared_ptr<RemoteNotify> ipcPostRecordNotify;
442 : std::vector<std::shared_ptr<RemoteNotify>> userRemoteNotify;
443 : TransportAttr transportAttr;
444 :
445 23 : TransportDeviceP2pData() {}
446 : TransportDeviceP2pData(
447 : void* inputBufferPtr, void* outputBufferPtr, std::shared_ptr<LocalNotify> ipcPreWaitNotify,
448 : std::shared_ptr<LocalNotify> ipcPostWaitNotify, std::vector<std::shared_ptr<LocalNotify>> userLocalNotify,
449 : std::shared_ptr<RemoteNotify> ipcPreRecordNotify, std::shared_ptr<RemoteNotify> ipcPostRecordNotify,
450 : std::vector<std::shared_ptr<RemoteNotify>> userRemoteNotify, TransportAttr& transportAttr)
451 : : inputBufferPtr(inputBufferPtr),
452 : outputBufferPtr(outputBufferPtr),
453 : ipcPreWaitNotify(ipcPreWaitNotify),
454 : ipcPostWaitNotify(ipcPostWaitNotify),
455 : userLocalNotify(userLocalNotify),
456 : ipcPreRecordNotify(ipcPreRecordNotify),
457 : ipcPostRecordNotify(ipcPostRecordNotify),
458 : userRemoteNotify(userRemoteNotify),
459 : transportAttr(transportAttr)
460 : {}
461 : };
462 :
463 : struct TransportDeviceIbverbsData {
464 : void* inputBufferPtr;
465 : void* outputBufferPtr;
466 : MemDetails localInputMem;
467 : MemDetails localOutputMem;
468 : std::shared_ptr<LocalNotify> ackNotify;
469 : std::shared_ptr<LocalNotify> dataAckNotify;
470 : std::shared_ptr<LocalNotify> dataNotify;
471 : std::vector<std::vector<std::shared_ptr<LocalNotify>>> userLocalNotify;
472 : uint64_t localNotifyValueAddr;
473 : AddrKey remoteAckNotifyDetails;
474 : AddrKey remoteDataNotifyDetails;
475 : AddrKey remoteDataAckNotifyDetails;
476 : std::vector<std::vector<AddrKey>> userRemoteNotifyDetails;
477 : uint32_t notifyValueKey;
478 : std::vector<struct HcclQpInfoV2> qpInfo;
479 : uint32_t remoteInputKey;
480 : uint32_t remoteOutputKey;
481 : uint32_t notifySize;
482 : u32 multiQpThreshold;
483 : u32 qpsPerConnection;
484 : bool useAtomicWrite = false;
485 : std::vector<RoceMemDetails> localRoceMemDetailsList;
486 : std::vector<RoceMemDetails> remoteRoceMemDetailsList;
487 : bool useMemDetailsMgr{false};
488 : uint64_t remoteNotifyValueAddr{0}; // 对端 NOTIFY_SRC_MEM 地址 (Fence Read 读取源)
489 : uint32_t remoteNotifyValueKey{0}; // 对端 notify rkey
490 : uint64_t localDataNotifyAddr{0}; // 本端dataNotify addr
491 : uint32_t localDataNotifyKey{0}; // 本端dataNotifyKey
492 30 : TransportDeviceIbverbsData() {}
493 : TransportDeviceIbverbsData(
494 : void* inputBufferPtr, void* outputBufferPtr, MemDetails localInputMem, MemDetails localOutputMem,
495 : std::shared_ptr<LocalNotify> ackNotify, std::shared_ptr<LocalNotify> dataAckNotify,
496 : std::shared_ptr<LocalNotify> dataNotify, std::vector<std::vector<std::shared_ptr<LocalNotify>>> userLocalNotify,
497 : uint64_t localNotifyValueAddr, AddrKey remoteAckNotifyDetails, AddrKey remoteDataNotifyDetails,
498 : AddrKey remoteDataAckNotifyDetails, std::vector<std::vector<AddrKey>> userRemoteNotifyDetails,
499 : uint32_t notifyValueKey, std::vector<struct HcclQpInfoV2> qpInfo, uint32_t remoteInputKey,
500 : uint32_t remoteOutputKey, uint32_t notifySize, u32 multiQpThreshold, u32 qpsPerConnection, bool useAtomicWrite,
501 : uint64_t remoteNotifyValueAddr, uint32_t remoteNotifyValueKey, uint64_t localDataNotifyAddr,
502 : uint32_t localDataNotifyKey)
503 : : inputBufferPtr(inputBufferPtr),
504 : outputBufferPtr(outputBufferPtr),
505 : localInputMem(localInputMem),
506 : localOutputMem(localOutputMem),
507 : ackNotify(ackNotify),
508 : dataAckNotify(dataAckNotify),
509 : dataNotify(dataNotify),
510 : userLocalNotify(userLocalNotify),
511 : localNotifyValueAddr(localNotifyValueAddr),
512 : remoteAckNotifyDetails(remoteAckNotifyDetails),
513 : remoteDataNotifyDetails(remoteDataNotifyDetails),
514 : remoteDataAckNotifyDetails(remoteDataAckNotifyDetails),
515 : userRemoteNotifyDetails(userRemoteNotifyDetails),
516 : notifyValueKey(notifyValueKey),
517 : qpInfo(qpInfo),
518 : remoteInputKey(remoteInputKey),
519 : remoteOutputKey(remoteOutputKey),
520 : notifySize(notifySize),
521 : multiQpThreshold(multiQpThreshold),
522 : qpsPerConnection(qpsPerConnection),
523 : useAtomicWrite(useAtomicWrite),
524 : remoteNotifyValueAddr(remoteNotifyValueAddr),
525 : remoteNotifyValueKey(remoteNotifyValueKey),
526 : localDataNotifyAddr(localDataNotifyAddr),
527 : localDataNotifyKey(localDataNotifyKey)
528 : {}
529 :
530 10 : TransportDeviceIbverbsData(const TransportDeviceIbverbsData& that)
531 10 : : inputBufferPtr(that.inputBufferPtr),
532 10 : outputBufferPtr(that.outputBufferPtr),
533 10 : localInputMem(that.localInputMem),
534 10 : localOutputMem(that.localOutputMem),
535 10 : ackNotify(that.ackNotify),
536 10 : dataAckNotify(that.dataAckNotify),
537 10 : dataNotify(that.dataNotify),
538 10 : userLocalNotify(that.userLocalNotify),
539 10 : localNotifyValueAddr(that.localNotifyValueAddr),
540 10 : remoteAckNotifyDetails(that.remoteAckNotifyDetails),
541 10 : remoteDataNotifyDetails(that.remoteDataNotifyDetails),
542 10 : remoteDataAckNotifyDetails(that.remoteDataAckNotifyDetails),
543 10 : userRemoteNotifyDetails(that.userRemoteNotifyDetails),
544 10 : notifyValueKey(that.notifyValueKey),
545 10 : qpInfo(that.qpInfo),
546 10 : remoteInputKey(that.remoteInputKey),
547 10 : remoteOutputKey(that.remoteOutputKey),
548 10 : notifySize(that.notifySize),
549 10 : multiQpThreshold(that.multiQpThreshold),
550 10 : qpsPerConnection(that.qpsPerConnection),
551 10 : useAtomicWrite(that.useAtomicWrite),
552 10 : localRoceMemDetailsList(that.localRoceMemDetailsList),
553 10 : remoteRoceMemDetailsList(that.remoteRoceMemDetailsList),
554 10 : useMemDetailsMgr(that.useMemDetailsMgr),
555 10 : remoteNotifyValueAddr(that.remoteNotifyValueAddr),
556 10 : remoteNotifyValueKey(that.remoteNotifyValueKey),
557 10 : localDataNotifyAddr(that.localDataNotifyAddr),
558 10 : localDataNotifyKey(that.localDataNotifyKey)
559 10 : {}
560 : };
561 3 : using CqeInfo = struct tagCqeInfo {
562 : struct timeval time;
563 : uint32_t status = 0;
564 : HcclIpAddress remoteIp;
565 : char reserved[32];
566 7 : tagCqeInfo() {}
567 1 : tagCqeInfo(const struct timeval& time, const uint32_t status, const HcclIpAddress& remoteIp)
568 1 : : time(time),
569 1 : status(status),
570 1 : remoteIp(remoteIp)
571 1 : {}
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(
587 : TransportType type, TransportPara& para, const HcclDispatcher dispatcher,
588 : const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
589 : const TransportDeviceP2pData& transDevP2pData = TransportDeviceP2pData(),
590 : const TransportDeviceIbverbsData& transDevIbverbsData = TransportDeviceIbverbsData());
591 :
592 : ~Transport();
593 :
594 : HcclResult Stop();
595 : HcclResult Resume();
596 : HcclResult Init();
597 : HcclResult DeInit();
598 :
599 : HcclResult TxDataSignal(Stream& stream);
600 : HcclResult RxDataSignal(Stream& stream);
601 :
602 : HcclResult TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream);
603 : HcclResult TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream);
604 :
605 : HcclResult TxWithReduce(
606 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, const HcclDataType datatype,
607 : HcclReduceOp redOp, Stream& stream);
608 : HcclResult TxWithReduce(
609 : const std::vector<TxMemoryInfo>& txWithReduceMems, const HcclDataType datatype, HcclReduceOp redOp,
610 : Stream& stream);
611 : HcclResult RxWithReduce(
612 : UserMemType recvSrcMemType, u64 recvSrcOffset, void* recvDst, u64 recvLen, void* reduceSrc, void* reduceDst,
613 : u64 reduceDataCount, HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream& stream, const u64 reduceAttr);
614 : HcclResult RxWithReduce(
615 : const std::vector<RxWithReduceMemoryInfo>& rxWithReduceMems, HcclDataType reduceDatatype, HcclReduceOp reduceOp,
616 : Stream& stream, const u64 reduceAttr);
617 : bool IsSupportTransportWithReduce();
618 :
619 : HcclResult RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream);
620 : HcclResult RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream);
621 : HcclResult DataReceivedAck(Stream& stream);
622 :
623 : HcclResult TxAck(Stream& stream);
624 : HcclResult RxAck(Stream& stream);
625 :
626 : HcclResult TxPrepare(Stream& stream);
627 : HcclResult RxPrepare(Stream& stream);
628 :
629 : HcclResult TxDone(Stream& stream);
630 : HcclResult RxDone(Stream& stream);
631 :
632 : HcclResult TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream);
633 : HcclResult RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream);
634 :
635 : // 保证send语义完成
636 : HcclResult TxWaitDone(Stream& stream);
637 : // 保证recv语义完成
638 : HcclResult RxWaitDone(Stream& stream);
639 : // TxWaitDone、RxWaitDone共同出现保证sendrecv语义完成
640 :
641 : HcclResult Post(u32 notifyIdx, Stream& stream);
642 : HcclResult Wait(u32 notifyIdx, Stream& stream, const u32 timeOut = NOTIFY_INVALID_WAIT_TIME);
643 :
644 : u32 GetNotifyNum();
645 : HcclResult GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t* memNum, HcclMemType memType);
646 : HcclResult GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum);
647 : HcclResult GetLocalNotify(std::vector<HcclSignalInfo>& localNotify);
648 : HcclResult GetRemoteNotify(std::vector<HcclSignalInfo>& localNotify);
649 : HcclResult GetRemoteMem(UserMemType memType, void** remotePtr);
650 : HcclResult GetRemoteMem(std::vector<void*>* remotePtrVec);
651 : HcclResult GetRemoteMemKey(UserMemType memType, uint32_t* remoteMemKey);
652 : HcclResult GetLocalRdmaNotify(std::vector<HcclSignalInfo>& rdmaNotify);
653 : HcclResult GetDrainLocalDataNotify(void*& localAddr, uint32_t& lkey, HcclSignalInfo& dataNotify);
654 : HcclResult GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey>& rdmaNotifyAddr);
655 : HcclResult GetLocalNotifyValueAddrKey(std::vector<AddrKey>& notifyValue);
656 : HcclResult GetLocalMemDetails(UserMemType memType, MemDetails& memDetails);
657 : HcclResult GetAiQpInfo(std::vector<HcclQpInfoV2>& aiQpInfo);
658 : HcclResult GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo>& aiRMAQueueInfo);
659 : HcclResult GetTransportId(u32& id);
660 : HcclResult GetChipId(s64& chipId);
661 : virtual HcclResult GetRemoteMemSize(UserMemType memType, u64& size);
662 : HcclResult GetTxAckDevNotifyInfo(HcclSignalInfo& notifyInfo);
663 : HcclResult GetRxAckDevNotifyInfo(HcclSignalInfo& notifyInfo);
664 : HcclResult GetTxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo);
665 : HcclResult GetRxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo);
666 :
667 : hccl::LinkType GetLinkType() const;
668 : bool IsSpInlineReduce() const;
669 : bool GetSupportDataReceivedAck() const;
670 : void SetSupportDataReceivedAck(bool supportDataReceivedAck);
671 : u32 GetRemoteRank();
672 :
673 : HcclResult ConnectAsync(u32& status);
674 : HcclResult ConnectQuerry(u32& status);
675 : void Break();
676 :
677 : void EnableUseOneDoorbell();
678 :
679 : bool GetUseOneDoorbellValue();
680 :
681 : HcclResult GetTransportAttr(TransportAttr& attr);
682 :
683 : HcclResult TxEnv(const void* ptr, const u64 len, Stream& stream);
684 : HcclResult RxEnv(Stream& stream);
685 : bool IsTransportRoce();
686 :
687 : HcclResult WriteAsync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream);
688 : HcclResult WriteSync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream);
689 :
690 : HcclResult WriteReduceAsync(
691 : struct Buffer& remoteBuf, struct Buffer& localBuf, const HcclDataType datatype, HcclReduceOp redOp,
692 : Stream& stream);
693 :
694 : HcclResult ReadAsync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream);
695 : HcclResult ReadSync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream);
696 : HcclResult ReadReduceSync(
697 : struct Buffer& localBuf, struct Buffer& remoteBuf, const HcclDataType datatype, HcclReduceOp redOp,
698 : Stream& stream);
699 :
700 : HcclResult BatchTransferAsync(const HcommBatchTransferDesc* transferDescs, uint32_t descNum, Stream& stream);
701 :
702 : HcclResult PostReady(Stream& stream);
703 : HcclResult WaitReady(Stream& stream);
704 :
705 : HcclResult PostFin(Stream& stream);
706 : HcclResult WaitFin(Stream& stream);
707 :
708 : HcclResult PostFinAck(Stream& stream);
709 : HcclResult WaitFinAck(Stream& stream);
710 :
711 : HcclResult SetStopFlag(bool value);
712 : HcclResult Fence();
713 : HcclResult UpdateRemoteAddr(void* remoteIn, void* remoteOut);
714 : HcclResult Drain(Stream& stream);
715 : HcclResult InitDrainNotifyInfo();
716 : HcclResult GetDrainRemSrcMem(void*& remoteAddr, uint32_t& remoteKey, uint32_t& size);
717 : static HcclResult
718 : GetTransportErrorCqe(const HcclNetDevCtx netDevCtx, std::vector<std::pair<Transport*, CqeInfo>>& infos, u32& num);
719 1 : inline TransportType GetTransportType() const { return type_; }
720 :
721 : std::vector<u8> GetExchangeInfo();
722 : static HcclResult HcclBatchRead(
723 : const TransportDeviceNormalData& ibvData, struct MemDetails* localMems, struct MemDetails* remoteMems,
724 : u32 memNum, u64& dbInfo);
725 : static HcclResult HcclBatchWrite(
726 : const TransportDeviceNormalData& ibvData, struct MemDetails* localMems, struct MemDetails* remoteMems,
727 : u32 memNum, u64& dbInfo);
728 : static HcclResult SetDeviceUnavailable(u32 deviceId);
729 :
730 : bool GetIsUseAtomicWrite();
731 :
732 : HcclResult GetSpecificNotify(HcclSignalInfo& notifyInfo, bool& isValid, const std::string& notifyName);
733 :
734 : private:
735 : void CreateTransportRoce(
736 : TransportType type, TransportPara& para, const HcclDispatcher dispatcherPtr,
737 : const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara);
738 : TransportBase* pimpl_ = nullptr;
739 : const TransportType type_ = TransportType::TRANS_TYPE_RESERVED;
740 :
741 : static std::mutex mapMutex_;
742 : static std::unordered_map<TransportBase*, Transport*> transportMap_;
743 : };
744 :
745 : using LINK = std::shared_ptr<Transport>;
746 : } // namespace hccl
747 :
748 : #endif /* TRANSPORT_BASE_H */
|
