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_P2P_PUB_H
12 : #define TRANSPORT_P2P_PUB_H
13 :
14 : #include <sstream>
15 :
16 : #include "transport_base_pub.h"
17 : #include "rma_buffer_mgr.h"
18 :
19 : namespace hccl {
20 : using HcclMemExMgr = hcomm::RmaBufferMgr<hccl::BufferKey<uintptr_t, u64>, std::shared_ptr<HcclMemEx>>;
21 :
22 : typedef enum {
23 : EX_IPCMEN_SIZE = 0, /**< ipcMenSize */
24 : EX_NOTIFY_SIZE = 1, /**< notifySize */
25 : EX_EXDATA_SIZE = 2 /**< exDataSize */
26 : } ExInfoType;
27 : struct ExchangeInfoSize {
28 : u32 ipcMenSize;
29 : u32 notifySize;
30 : u32 exDataSize;
31 : u32 indOpMemSize; // 独立算子内存两端大小不同，不参与比较
32 :
33 2 : bool compare(ExchangeInfoSize& that)
34 : {
35 2 : if ((this->ipcMenSize != that.ipcMenSize) || (this->notifySize != that.notifySize)
36 2 : || (this->exDataSize != that.exDataSize)) {
37 0 : return false;
38 : }
39 2 : return true;
40 : }
41 : };
42 :
43 : class TransportP2p : public TransportBase {
44 : public:
45 : explicit TransportP2p(
46 : DispatcherPub* dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
47 : std::chrono::milliseconds timeout);
48 : ~TransportP2p() override;
49 :
50 : HcclResult Init() override;
51 :
52 : HcclResult TxDataSignal(Stream& stream) override;
53 : HcclResult RxDataSignal(Stream& stream) override;
54 :
55 : HcclResult TxAck(Stream& stream) override;
56 : HcclResult RxAck(Stream& stream) override;
57 :
58 : HcclResult TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream) override;
59 : HcclResult TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream) override;
60 :
61 : HcclResult RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream) override;
62 : HcclResult RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream) override;
63 :
64 : HcclResult DataReceivedAck(Stream& stream) override;
65 :
66 : HcclResult TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream) override;
67 :
68 : HcclResult RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream) override;
69 :
70 : HcclResult TxPrepare(Stream& stream) override;
71 : HcclResult RxPrepare(Stream& stream) override;
72 :
73 : HcclResult TxDone(Stream& stream) override;
74 : HcclResult RxDone(Stream& stream) override;
75 :
76 : HcclResult GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum) override;
77 : HcclResult GetRemoteMem(UserMemType memType, void** remotePtr) override;
78 : HcclResult GetRemoteMem(std::vector<void*>* remotePtrVec) override;
79 :
80 : HcclResult GetRemoteMemSize(UserMemType memType, u64& size) override;
81 :
82 : HcclResult WriteAsync(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream);
83 : HcclResult WriteAsyncEx(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream);
84 : HcclResult WriteSync(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream);
85 : HcclResult WriteReduceAsync(
86 : struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, const HcclDataType datatype,
87 : HcclReduceOp redOp, Stream& stream) override;
88 :
89 : HcclResult ReadAsync(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream);
90 : HcclResult ReadAsyncEx(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream);
91 : HcclResult ReadSync(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream);
92 : HcclResult ReadReduceSync(
93 : struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, const HcclDataType datatype,
94 : HcclReduceOp redOp, Stream& stream);
95 :
96 : HcclResult PostReady(Stream& stream);
97 : HcclResult WaitReady(Stream& stream);
98 :
99 : HcclResult PostFin(Stream& stream);
100 : HcclResult WaitFin(Stream& stream);
101 :
102 : HcclResult Post(u32 notifyIdx, Stream& stream) override;
103 : HcclResult Wait(u32 notifyIdx, Stream& stream, const u32 timeOut = NOTIFY_INVALID_WAIT_TIME) override;
104 :
105 : protected:
106 : HcclResult FillExchangeDataTotalSize() override;
107 :
108 : HcclResult ConstructExchangeForSend() override;
109 : HcclResult ConstructIpcMemInfoForSend(void* ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
110 : HcclResult ConstructIntraProcMemInfoForSend(void* ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
111 : HcclResult ConstructNumInfoForSend(u64 num, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
112 :
113 : HcclResult ConstructNotifyInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
114 : HcclResult ConstructNotifyVectorInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
115 : HcclResult ConstructDataLenForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
116 :
117 : HcclResult ParseReceivedExchangeData() override;
118 : HcclResult ParseIpcMemInfo(
119 : void** memPtr, u64& size, u8* memName, u64& offset, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
120 : HcclResult ParseIntraProcMemInfo(u64* addr, u64* size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
121 : HcclResult ParseNotifyInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
122 : HcclResult ParseNotifyInfoEx(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
123 : HcclResult ParseNotifyVectorInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
124 : HcclResult ParseCheckDataLen(ExchangeInfoSize& remoteInfoSize, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
125 : HcclResult ParseMemNumInfo(u64& memNum, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
126 :
127 : HcclResult GetLocalNotify(std::vector<HcclSignalInfo>& localNotify) override;
128 : HcclResult GetRemoteNotify(std::vector<HcclSignalInfo>& localNotify) override;
129 :
130 : HcclResult WaitPeerMemConfig(void** memPtr, const u8* memName, uint64_t size, u64 offset);
131 : HcclResult ExchangeMemAndNotifyMesg();
132 : HcclResult SendIpcMemMesg(void* ptr, u64 size) const;
133 : HcclResult RecvIpcMemMesg(void** memPtr, u8* memName, u64& offset);
134 : HcclResult SendMemMesgWithoutIpc(void* ptr, u64 size) const;
135 : HcclResult RecvMemMesgWithoutIpc(u64& addr, u8* memName, u64& offset);
136 : HcclResult ExchangeMemAndNotifyWithIpc();
137 : HcclResult ExchangeMemAndNotifyWithoutIpc();
138 : virtual HcclResult SignalRecord(
139 : std::shared_ptr<RemoteNotify>& remoteSignal, u64 remoteSignalAddr, u64 remoteSignalOffset, Stream& stream);
140 : void SetTransportRelationship();
141 : HcclResult SetLinkType();
142 : HcclResult CreateNotifyValueBuffer();
143 : HcclResult
144 : SumCheckSizeAndConsisten(ExInfoType exInfoType, u32 rightInfoSize, u64& blankSizeRecord, u64 exchangeDataBlankSize);
145 : void SetUseSdmaToSignalRecord();
146 :
147 : void* remoteInputPtr_;
148 : void* remoteOutputPtr_;
149 : std::vector<void*> remoteIpcMemPtrVector_;
150 : std::vector<void*> remoteIndOpHostMemPtrVector_;
151 : std::vector<void*> remoteIndOpDeviceMemPtrVector_;
152 : u64 remoteInputSize_;
153 : u64 remoteOutputSize_;
154 : std::vector<u64> remoteIpcMemSizeVector_;
155 : std::vector<u64> remoteIndOpHostMemSizeVector_;
156 : std::vector<u64> remoteIndOpDeviceMemSizeVector_;
157 : u64 remoteOutputOffsetValue_;
158 : u64 remoteInputOffsetValue_;
159 : std::vector<u64> remoteIpcMemOffsetValueVector_;
160 : std::vector<u64> remoteIndOpHostMemOffsetValueVector_;
161 : std::vector<u64> remoteIndOpDeviceMemOffsetValueVector_;
162 :
163 : bool useSdmaToSignalRecord_{false};
164 :
165 : HcclResult ReplaceMemAddr(
166 : Transport::Buffer& localMem, Transport::Buffer& remoteMem, Transport::Buffer& newLocalMem,
167 : Transport::Buffer& newRemoteMem, bool& isLocalHostAddr, bool& isRemoteHostAddr);
168 : HcclResult InitHcclMemExMgrWithMem(HcclMemEx* bufMem, u32 bufSize, HcclMemExMgr& hcommMemExMgr);
169 : HcclResult InitHcclMemExMgr(MachinePara& machinePara);
170 : HcclMemExMgr localHcclMemExMgr_;
171 : HcclMemExMgr remoteHcclMemExMgr_;
172 :
173 : private:
174 : HcclResult ParseSpecifyLink(LinkTypeInServer& linkType);
175 : void SetMemIncludeFlag();
176 : HcclResult ConstructMemIncludeInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
177 : HcclResult ParseMemIncludeInfo(void** memPtr, u64& size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
178 : SecIpcName_t remoteOutputMemName_;
179 : SecIpcName_t remoteInputMemName_;
180 : std::vector<SecIpcName_t> remoteIpcMemNameVector_;
181 : std::vector<SecIpcName_t> remoteIndOpHostMemNameVector_;
182 : std::vector<SecIpcName_t> remoteIndOpDeviceMemNameVector_;
183 : static std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> notifyValueMem_;
184 : static std::array<std::mutex, MAX_MODULE_DEVICE_NUM> notifyValueMutex_;
185 : const u64 notifyValueSize_{LARGE_PAGE_MEMORY_MIN_SIZE}; // 避免申请小页内存。最小2*1024*1024
186 : static std::array<Referenced, MAX_MODULE_DEVICE_NUM> instanceRef_; // 实例计数，用于释放静态资源
187 : ExchangeInfoSize exchangeInfoSize_{};
188 : bool isSioToHccs_{false}; // 是否是sio->hccs的链路
189 : bool isMemInclude_{false}; // input output是否在machinePara_.mem[0]范围内
190 : };
191 : } // namespace hccl
192 :
193 : #endif /* TRANSPORT_P2P_PUB_H */
|
