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_HETEROG_PUB_H
12 : #define TRANSPORT_HETEROG_PUB_H
13 : #include <unordered_map>
14 : #include <atomic>
15 : #include <functional>
16 : #include <vector>
17 : #include <queue>
18 : #include <stack>
19 : #include "memory_alloc_ring.h"
20 : #include "private_types.h"
21 : #include "local_ipc_notify.h"
22 :
23 : namespace hccl {
24 : enum class ConnState {
25 : CONN_STATE_IDLE,
26 : CONN_STATE_CONNECT_CHECK_SOCKET,
27 : CONN_STATE_GET_CHECK_SOCKET,
28 : CONN_STATE_SEND_CF,
29 : CONN_STATE_RECV_CF,
30 : CONN_STATE_CHECK_CF,
31 : CONN_STATE_CONNECT_ALL_SOCKET,
32 : CONN_STATE_GET_ALL_SOCKET,
33 : CONN_STATE_CONNECT_QP,
34 : CONN_STATE_GET_QP,
35 : CONN_STATE_SEND_STATUS,
36 : CONN_STATE_RECV_STATUS,
37 : CONN_STATE_FLUSH_QUEUE,
38 : CONN_STATE_GET_TAG_QP_ATTR,
39 : CONN_STATE_SEND_TAG_QP_INFO,
40 : CONN_STATE_RECV_TAG_QP_INFO,
41 : CONN_STATE_MODIFY_TAG_QP,
42 : CONN_STATE_GET_DATA_QP_ATTR,
43 : CONN_STATE_SEND_DATA_QP_INFO,
44 : CONN_STATE_RECV_DATA_QP_INFO,
45 : CONN_STATE_MODIFY_DATA_QP,
46 : CONN_STATE_COMPLETE
47 : };
48 :
49 : enum class RdmaNotifyOp {
50 : SEND_NOTIFY,
51 : RECV_NOTIFY,
52 : NUM
53 : };
54 : constexpr u32 HETEROG_MAX_FRAME_LEN = 128;
55 : struct InitInfo {
56 : s32 protocolType = 0; // 0:ROCE; 1:TCP
57 : u32 role = 0;
58 : u32 signal = 0;
59 : u8 checkFrame[HETEROG_MAX_FRAME_LEN] = {0};
60 : std::vector<SocketInfoT> socketInfo;
61 : std::vector<SocketConnectInfoT> socketConnInfo;
62 : };
63 :
64 : struct InitStateMachine {
65 : InitInfo locInitInfo;
66 : InitInfo remInitInfo;
67 : u64 size = 0;
68 : u64 completeSize = 0;
69 : u32 socketNum = 0;
70 : u32 completeNum = 0;
71 : };
72 :
73 : static constexpr u32 SYNC_SIGNAL = 0xFFFFFFFF;
74 : constexpr u32 HCCL_POLL_CQ_DEPTH = 32;
75 :
76 : struct TransportEndPointInfoHash {
77 0 : std::size_t operator () (const TransportEndPointInfo &t) const
78 : {
79 0 : return std::hash<u32>()(t.commId) ^ std::hash<u32>()(t.rank) ^ std::hash<u32>()(t.tag);
80 : }
81 : };
82 :
83 : using HcclReceivedEnvelope =
84 : std::unordered_map<TransportEndPointInfo, std::queue<HcclEnvelopeSummary>, TransportEndPointInfoHash>;
85 :
86 : class TransportHeterog {
87 : public:
88 : explicit TransportHeterog(const std::string &tag, HcclIpAddress &selfIp, HcclIpAddress &peerIp, u32 peerPort,
89 : u32 selfPort, const TransportResourceInfo &transportResourceInfo);
90 : explicit TransportHeterog(const TransportResourceInfo &transportResourceInfo);
91 : virtual ~TransportHeterog();
92 : virtual HcclResult Init() = 0;
93 : virtual HcclResult Init(u32 localUserRank, u32 remoteUserRank);
94 : virtual HcclResult Init(SocketInfoT &socketInfo, RdmaHandle rdmaHandle, MrHandle mrHandle);
95 : virtual HcclResult Deinit() = 0;
96 : virtual HcclResult Isend(const TransData &sendData, const TransportEndPointParam &epParam,
97 : HcclRequestInfo *&request) = 0;
98 : virtual HcclResult Send(const TransData &sendData, const TransportEndPointParam &epParam) = 0;
99 : virtual HcclResult Improbe(const TransportEndPointParam &epParam, s32 &matched, HcclMessageInfo *&msg,
100 : HcclStatus &status) = 0;
101 : virtual HcclResult Imrecv(const TransData &recvData, HcclMessageInfo &msg, HcclRequestInfo *&request) = 0;
102 : virtual HcclResult Test(HcclRequestInfo &request, s32 &flag, HcclStatus &compState) = 0;
103 : virtual HcclResult Improbe(const TransportEndPointParam &epParam, s32 &matched, HcclMessageInfo *&msg,
104 : HcclStatus &status, bool &flag);
105 : virtual HcclResult Imrecv(const TransData &recvData, HcclMessageInfo &msg, HcclRequestInfo *&request,
106 : bool flag, bool needRecordFlag);
107 : virtual HcclResult ImrecvScatter(void *buf[], int count[], int bufCount, HcclDataType datatype,
108 : HcclMessageInfo &msg, HcclRequestInfo *&request);
109 : HcclResult SetDeviceIndex(s32 index);
110 : u32 GetRecvEnvelopNum();
111 : void AddRecvEnvelopNum();
112 : void SubRecvEnvelopNum();
113 : virtual HcclResult BlockSend(const TransData &sendData, const TransportEndPointParam &epParam,
114 : HcclRequestInfo *&request, s32 waitTimeOut);
115 : virtual HcclResult BlockRecv(const TransData &recvData, bool matched,
116 : TransportHeterog *&transport, s32 waitTimeOut, s32 waitPayloadTimeOut);
117 : HcclResult CheckAndPushBuildLink();
118 : HcclResult WaitBuildLinkComplete();
119 : virtual HcclResult Iwrite(const TransData &sendData, const HcclEnvelope &envelope, HcclRequestInfo *&request);
120 : virtual HcclResult GetRemoteIsendDoneSignal(std::shared_ptr<LocalIpcNotify> &signal);
121 : virtual HcclResult GetRemoteImrecvDoneSignal(std::shared_ptr<LocalIpcNotify> &signal);
122 :
123 : ConnState GetState();
124 : virtual void GetLinkTag(std::string &tag);
125 : void SetForceClose();
126 : HcclResult SocketSend(const FdHandle fdHandle, void *data, u64 size, u64 &sentSize, bool &completed);
127 : HcclResult SocketRecv(const FdHandle fdHandle, void *data, u64 size, u64 &recvSize, bool &completed);
128 : static void RecordRankTableCrc(const u32 crcValue);
129 :
130 : protected:
131 : HcclResult CheckRecvMsgAndRequestBuffer();
132 : HcclResult GenerateSendRequest(const TransData &sendData, const TransportEndPointParam &epParam,
133 : HcclRequestInfo *&request);
134 : HcclResult GenerateRecvRequest(const TransData &recvData, const HcclMessageInfo &msg, HcclRequestInfo *&request);
135 : HcclResult GenerateRecvScatterRequest(const HcclMessageInfo &msg, HcclRequestInfo *&request);
136 : HcclResult FreeRequest(HcclRequestInfo &request) const;
137 : HcclResult CheckTransportEndPointInfo(const TransportEndPointInfo &epInfo,
138 : const TransportEndPointInfo &epInfoCheck) const;
139 : HcclResult CheckRecvEnvelope(const TransData &recvDataCheck, const HcclEnvelopeSummary &envelope);
140 : HcclResult CheckRecvScatterEnvelope(void *buf[], int count[], int bufCount, HcclDataType datatype,
141 : const HcclEnvelopeSummary &envelope);
142 : HcclResult GenerateRecvMessage(HcclEnvelopeSummary &recvEnvelope, HcclMessageInfo *&msg, HcclStatus &status);
143 : HcclResult FreeRecvMessage(HcclMessageInfo &msg) const;
144 : HcclResult ProbeNothing(s32 &flag, HcclMessageInfo *&msg, HcclStatus &status) const;
145 : HcclResult ConnectSocket(SocketConnectInfoT conn[], u32 num, bool &completed);
146 : HcclResult GetSocket(u32 role, struct SocketInfoT info[], u32 num, u32 &connectedNum, bool &completed);
147 : HcclResult SocketClose();
148 : HcclResult CheckConsistentFrame();
149 : HcclResult ConnectAsync();
150 : HcclResult PrepareSocketInfo(s32 type, s32 linkNum, const std::string &clientTag, const std::string &serverTag);
151 : HcclResult InitTransportConnect(s32 type, s32 linkNum);
152 : HcclResult InitTransportConnect(s32 type, u32 role, s32 linkNum, u32 tag);
153 : HcclResult AddSocketWhiteList(std::string& tag);
154 : HcclResult TryTransition(HcclResult ret, bool completed, ConnState nextState);
155 : virtual HcclResult EnterStateProcess(ConnState nextState) = 0;
156 : virtual HcclResult LoopStateProcess() = 0;
157 :
158 : const std::string transTag_;
159 : SocketHandle nicSocketHandle_;
160 : HcclIpAddress selfIp_;
161 : HcclIpAddress peerIp_;
162 : u32 peerPort_;
163 : u32 selfPort_;
164 : struct InitStateMachine initSM_{};
165 : std::atomic<ConnState> connState_{ ConnState::CONN_STATE_IDLE };
166 : const std::unique_ptr<LocklessRingMemoryAllocate<HcclMessageInfo>> &pMsgInfosMem_;
167 : const std::unique_ptr<LocklessRingMemoryAllocate<HcclRequestInfo>> &pReqInfosMem_;
168 : s32 index_ = 0;
169 : u32 recvEnvelopNum_;
170 : bool isHdcMode_ = false;
171 : u32 localRank_ = 0;
172 : u32 remoteRank_ = 0;
173 : bool remoteIsHdc_ = false; // 连接对端为310时，为false
174 : bool isESMode_ = false;
175 : bool forceClose_ = false; // 设置socket batch close时是否为强制关闭，而非超时“优雅”关闭
176 :
177 : static std::atomic<u32> rankTableCrc_;
178 : };
179 : } // namespace hccl
180 : #endif
|
