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 = hccl::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 : {
29 : u32 ipcMenSize;
30 : u32 notifySize;
31 : u32 exDataSize;
32 : u32 indOpMemSize; // 独立算子内存两端大小不同，不参与比较
33 :
34 2 : bool compare(ExchangeInfoSize &that) {
35 2 : if ((this->ipcMenSize != that.ipcMenSize) ||
36 2 : (this->notifySize != that.notifySize) ||
37 2 : (this->exDataSize != that.exDataSize)) {
38 0 : return false;
39 : }
40 2 : return true;
41 : }
42 : };
43 :
44 : class TransportP2p : public TransportBase {
45 : public:
46 : explicit TransportP2p(DispatcherPub *dispatcher,
47 : const std::unique_ptr<NotifyPool> ¬ifyPool,
48 : MachinePara &machinePara,
49 : std::chrono::milliseconds timeout);
50 : ~TransportP2p() override;
51 :
52 : HcclResult Init() override;
53 :
54 : HcclResult TxDataSignal(Stream &stream) override;
55 : HcclResult RxDataSignal(Stream &stream) override;
56 :
57 : HcclResult TxAck(Stream &stream) override;
58 : HcclResult RxAck(Stream &stream) override;
59 :
60 : HcclResult TxAsync(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len,
61 : Stream &stream) override;
62 : HcclResult TxAsync(std::vector<TxMemoryInfo>& txMems, Stream &stream) override;
63 :
64 : HcclResult RxAsync(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len,
65 : Stream &stream) override;
66 : HcclResult RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream &stream) override;
67 :
68 : HcclResult DataReceivedAck(Stream &stream) override;
69 :
70 : HcclResult TxData(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len, Stream &stream) override;
71 :
72 : HcclResult RxData(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream) override;
73 :
74 : HcclResult TxPrepare(Stream &stream) override;
75 : HcclResult RxPrepare(Stream &stream) override;
76 :
77 : HcclResult TxDone(Stream &stream) override;
78 : HcclResult RxDone(Stream &stream) override;
79 :
80 : HcclResult GetIndOpRemoteMem(HcclMem **remoteMem, uint32_t *memNum) override;
81 : HcclResult GetRemoteMem(UserMemType memType, void **remotePtr) override;
82 : HcclResult GetRemoteMem(std::vector<void *> *remotePtrVec) override;
83 :
84 : HcclResult GetRemoteMemSize(UserMemType memType, u64 &size) override;
85 :
86 : HcclResult WriteAsync(struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream);
87 : HcclResult WriteAsyncEx(struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream);
88 : HcclResult WriteSync(struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream);
89 : HcclResult WriteReduceAsync(struct Transport::Buffer &remoteBuf,
90 : struct Transport::Buffer &localBuf, const HcclDataType datatype, HcclReduceOp redOp, Stream &stream) override;
91 :
92 : HcclResult ReadAsync(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream);
93 : HcclResult ReadAsyncEx(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream);
94 : HcclResult ReadSync(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream);
95 : HcclResult ReadReduceSync(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf,
96 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream);
97 :
98 : HcclResult PostReady(Stream &stream);
99 : HcclResult WaitReady(Stream &stream);
100 :
101 : HcclResult PostFin(Stream &stream);
102 : HcclResult WaitFin(Stream &stream);
103 :
104 : HcclResult Post(u32 notifyIdx, Stream &stream) override;
105 : HcclResult Wait(u32 notifyIdx, Stream &stream, const u32 timeOut = NOTIFY_INVALID_WAIT_TIME) override;
106 :
107 : protected:
108 : HcclResult FillExchangeDataTotalSize() override;
109 :
110 : HcclResult ConstructExchangeForSend() override;
111 : HcclResult ConstructIpcMemInfoForSend(void *ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
112 : HcclResult ConstructIntraProcMemInfoForSend(void *ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
113 : HcclResult ConstructNumInfoForSend(u64 num, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
114 :
115 : HcclResult ConstructNotifyInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
116 : HcclResult ConstructNotifyVectorInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
117 : HcclResult ConstructDataLenForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
118 :
119 : HcclResult ParseReceivedExchangeData() override;
120 : HcclResult ParseIpcMemInfo(void **memPtr, u64 &size, u8 *memName, u64 &offset, u8 *&exchangeDataPtr,
121 : u64 &exchangeDataBlankSize);
122 : HcclResult ParseIntraProcMemInfo(u64* addr, u64* size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
123 : HcclResult ParseNotifyInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
124 : HcclResult ParseNotifyInfoEx(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
125 : HcclResult ParseNotifyVectorInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
126 : HcclResult ParseCheckDataLen(ExchangeInfoSize &remoteInfoSize, u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
127 : HcclResult ParseMemNumInfo(u64 &memNum, u8 *&exchangeDataPtr, u64 &exchangeDataBlankSize);
128 :
129 : HcclResult GetLocalNotify(std::vector<HcclSignalInfo> &localNotify) override;
130 : HcclResult GetRemoteNotify(std::vector<HcclSignalInfo> &localNotify) override;
131 :
132 : HcclResult WaitPeerMemConfig(void **memPtr, const u8 *memName, uint64_t size, u64 offset);
133 : HcclResult ExchangeMemAndNotifyMesg();
134 : HcclResult SendIpcMemMesg(void *ptr, u64 size) const;
135 : HcclResult RecvIpcMemMesg(void **memPtr, u8 *memName, u64 &offset);
136 : HcclResult SendMemMesgWithoutIpc(void *ptr, u64 size) const;
137 : HcclResult RecvMemMesgWithoutIpc(u64 &addr, u8 *memName, u64 &offset);
138 : HcclResult ExchangeMemAndNotifyWithIpc();
139 : HcclResult ExchangeMemAndNotifyWithoutIpc();
140 : virtual HcclResult SignalRecord(std::shared_ptr<RemoteNotify> &remoteSignal, u64 remoteSignalAddr, u64 remoteSignalOffset, Stream &stream);
141 : void SetTransportRelationship();
142 : HcclResult SetLinkType();
143 : HcclResult CreateNotifyValueBuffer();
144 : HcclResult SumCheckSizeAndConsisten(ExInfoType exInfoType, u32 rightInfoSize, u64 &blankSizeRecord,
145 : u64 exchangeDataBlankSize);
146 : void SetUseSdmaToSignalRecord();
147 :
148 : void *remoteInputPtr_;
149 : void *remoteOutputPtr_;
150 : std::vector<void*> remoteIpcMemPtrVector_;
151 : std::vector<void*> remoteIndOpHostMemPtrVector_;
152 : std::vector<void*> remoteIndOpDeviceMemPtrVector_;
153 : u64 remoteInputSize_;
154 : u64 remoteOutputSize_;
155 : std::vector<u64> remoteIpcMemSizeVector_;
156 : std::vector<u64> remoteIndOpHostMemSizeVector_;
157 : std::vector<u64> remoteIndOpDeviceMemSizeVector_;
158 : u64 remoteOutputOffsetValue_;
159 : u64 remoteInputOffsetValue_;
160 : std::vector<u64> remoteIpcMemOffsetValueVector_;
161 : std::vector<u64> remoteIndOpHostMemOffsetValueVector_;
162 : std::vector<u64> remoteIndOpDeviceMemOffsetValueVector_;
163 :
164 : bool useSdmaToSignalRecord_{false};
165 :
166 : HcclResult ReplaceMemAddr(Transport::Buffer &localMem, Transport::Buffer &remoteMem,
167 : Transport::Buffer &newLocalMem, 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_ {0};
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 */
|
