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 : #include "transport_ipc_mem.h"
12 : #include "log.h"
13 : #include "dispatcher_pub.h"
14 : #include "hccl_network.h"
15 :
16 : namespace hccl {
17 : using namespace std;
18 : using LocalIpcRmaBufferMgr = NetDevContext::LocalIpcRmaBufferMgr;
19 :
20 0 : TransportIpcMem::TransportIpcMem(const std::unique_ptr<NotifyPool> ¬ifyPool, const HcclNetDevCtx &netDevCtx,
21 0 : const HcclDispatcher &dispatcher, AttrInfo &attrInfo, bool aicpuUnfoldMode)
22 : : TransportMem(notifyPool, netDevCtx, dispatcher, attrInfo, aicpuUnfoldMode),
23 0 : sdid_(attrInfo.sdid), serverId_(attrInfo.serverId)
24 0 : {}
25 :
26 0 : TransportIpcMem::~TransportIpcMem()
27 0 : {}
28 :
29 0 : HcclResult TransportIpcMem::ExchangeMemDesc(
30 : const RmaMemDescs &localMemDescs, RmaMemDescs &remoteMemDescs, u32 &actualNumOfRemote)
31 : {
32 0 : return DoExchangeMemDesc(localMemDescs, remoteMemDescs, actualNumOfRemote);
33 : }
34 :
35 0 : HcclResult TransportIpcMem::EnableMemAccess(const RmaMemDesc &remoteMemDesc, RmaMem &remoteMem)
36 : {
37 0 : std::string tempDesc = RmaMemDescCopyToStr(remoteMemDesc);
38 :
39 : // 创建远程缓冲区，并进行反序列化、打开
40 0 : std::shared_ptr<RemoteIpcRmaBuffer> tempRemoteBufferPtr = nullptr;
41 0 : EXCEPTION_CATCH((tempRemoteBufferPtr = make_shared<RemoteIpcRmaBuffer>(netDevCtx_)), return HCCL_E_PARA);
42 0 : HcclResult ret = tempRemoteBufferPtr->Deserialize(tempDesc);
43 0 : CHK_PRT_RET((ret != HCCL_SUCCESS),
44 : HCCL_ERROR("[TransportIpcMem][EnableMemAccess]RemoteBuffer Deserialize failed."), ret);
45 :
46 0 : ret = tempRemoteBufferPtr->Open();
47 0 : CHK_PRT_RET((ret != HCCL_SUCCESS), HCCL_ERROR("[TransportIpcMem][EnableMemAccess]RemoteBuffer Open failed."), ret);
48 :
49 : BufferKey<uintptr_t, u64> tempKey(
50 0 : reinterpret_cast<uintptr_t>(tempRemoteBufferPtr->GetAddr()), tempRemoteBufferPtr->GetSize());
51 0 : auto resultPair = remoteIpcRmaBufferMgr_.Add(tempKey, tempRemoteBufferPtr);
52 0 : if (resultPair.first == remoteIpcRmaBufferMgr_.End()) {
53 : // 输入key与已有的内存重叠
54 0 : HCCL_ERROR("[TransportIpcMem][EnableMemAccess]The memory that is expected to enable"\
55 : " overlaps with the memory that has been enabled, please check params");
56 0 : return HCCL_E_INTERNAL;
57 : }
58 :
59 : // 已使能：输入key是表中某一最相近key的全集。 返回添加该key的迭代器，及false
60 : // 未使能：输入key是表中某一最相近key的空集。 返回添加成功的迭代器，及true
61 0 : std::string logInfo = resultPair.second ? "Enable memory access success!"
62 0 : : "Memory is already enabled, just increase the reference count.";
63 0 : HCCL_INFO("[TransportIpcMem][EnableMemAccess]:%s", logInfo.c_str());
64 :
65 : // 填充出参TransportRmaMem信息
66 0 : remoteMem.addr = tempRemoteBufferPtr->GetAddr();
67 0 : remoteMem.size = tempRemoteBufferPtr->GetSize();
68 0 : remoteMem.type = tempRemoteBufferPtr->GetMemType();
69 0 : return HCCL_SUCCESS;
70 0 : }
71 :
72 0 : HcclResult TransportIpcMem::DisableMemAccess(const RmaMemDesc &remoteMemDesc)
73 : {
74 : // 内存去使能管理
75 0 : std::string tempDesc = RmaMemDescCopyToStr(remoteMemDesc);
76 0 : std::shared_ptr<RemoteIpcRmaBuffer> tempRemoteBuffer = make_shared<RemoteIpcRmaBuffer>(netDevCtx_);
77 0 : HcclResult ret = tempRemoteBuffer->Deserialize(tempDesc);
78 0 : CHK_PRT_RET((ret != HCCL_SUCCESS),
79 : HCCL_ERROR("[TransportIpcMem][DisableMemAccess]RemoteBuffer Deserialize failed."), ret);
80 :
81 0 : ret = tempRemoteBuffer->Close();
82 0 : CHK_PRT_RET((ret != HCCL_SUCCESS), HCCL_ERROR("[TransportIpcMem][DisableMemAccess]RemoteBuffer Close failed."), ret);
83 :
84 : BufferKey<uintptr_t, u64> tempKey(
85 0 : reinterpret_cast<uintptr_t>(tempRemoteBuffer->GetAddr()), tempRemoteBuffer->GetSize());
86 : try {
87 0 : if (remoteIpcRmaBufferMgr_.Del(tempKey)) {
88 : // 删除成功：输入key是表中某一最相近key的全集，计数-1后为0，返回true
89 0 : HCCL_INFO("[TransportIpcMem][DisableMemAccess]Memory reference count is 0, disable memory access.");
90 : } else {
91 : // 删除失败：输入key是表中某一最相近key的全集，计数不为0（存在其他remoteRank使用），返回false
92 0 : HCCL_INFO("[TransportIpcMem][DisableMemAccess]Memory reference count is larger than 0"\
93 : "(used by other RemoteRank), do not disable memory.");
94 : }
95 0 : return HCCL_SUCCESS;
96 0 : } catch (std::out_of_range& e) {
97 0 : HCCL_ERROR("[TransportIpcMem][DisableMemAccess] catch RmaBufferMgr Del exception: %s", e.what());
98 0 : return HCCL_E_NOT_FOUND;
99 0 : }
100 0 : }
101 :
102 0 : HcclResult TransportIpcMem::FillRmaBufferSlice(const HcclBuf &localMem, const HcclBuf &remoteMem,
103 : RmaBufferSlice &localRmaBufferSlice, RmaBufferSlice &remoteRmaBufferSlice)
104 : {
105 0 : void* remoteAddr = remoteMem.addr;
106 0 : void* localAddr = localMem.addr;
107 0 : u64 byteSize = std::min(remoteMem.len, localMem.len);
108 : // local-handle还在map中获取，remote-hanle从外部传入
109 0 : auto localKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(localAddr), byteSize);
110 :
111 0 : NetDevContext *netDevCtx = static_cast<NetDevContext *>(netDevCtx_);
112 0 : std::shared_ptr<LocalIpcRmaBufferMgr> localRmaBufferMgr = netDevCtx->GetlocalIpcRmaBufferMgr();
113 0 : if (!localRmaBufferMgr) {
114 0 : HCCL_ERROR("[TransportIpcMem] can't get LocalIpcRmaBufferMgr");
115 0 : return HCCL_E_INTERNAL;
116 : }
117 :
118 0 : auto localBuffer = localRmaBufferMgr->Find(localKey);
119 0 : CHK_PRT_RET(!localBuffer.first,
120 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] Can't find localBuffer by key {%p, %llu}",
121 : localAddr, byteSize),
122 : HCCL_E_INTERNAL);
123 0 : CHK_PRT_RET(!localBuffer.second->GetAddr(),
124 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The addr of local Buffer or remote buffer is nullptr."),
125 : HCCL_E_NOT_FOUND);
126 0 : CHK_PRT_RET(!localBuffer.second->GetDevAddr(),
127 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The dev addr of local Buffer is nullptr."),
128 : HCCL_E_NOT_FOUND);
129 0 : CHK_RET(CheckHcclBuffer(localAddr, localBuffer.second.get()));
130 :
131 0 : RmaBuffer* remoteBuffer = static_cast<RmaBuffer*>(remoteMem.handle);
132 0 : CHK_PRT_RET(!remoteBuffer->GetDevAddr(),
133 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The dev addr of remote buffer is nullptr."),
134 : HCCL_E_NOT_FOUND);
135 0 : CHK_PRT_RET(!remoteBuffer->GetAddr(),
136 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The addr of remote buffer is nullptr."),
137 : HCCL_E_NOT_FOUND);
138 0 : CHK_RET(CheckHcclBuffer(remoteAddr, remoteBuffer));
139 :
140 0 : u64 localDataOffSet = static_cast<u8*>(localAddr) - static_cast<u8*>(localBuffer.second->GetAddr());
141 0 : u64 remoteDataOffSet = static_cast<u8*>(remoteAddr) - static_cast<u8*>(remoteBuffer->GetAddr());
142 0 : localRmaBufferSlice.addr = static_cast<void*>(static_cast<u8*>(localBuffer.second->GetDevAddr()) + localDataOffSet);
143 0 : localRmaBufferSlice.len = byteSize;
144 0 : localRmaBufferSlice.rmaBuffer = localBuffer.second;
145 0 : localRmaBufferSlice.memType = localBuffer.second->GetMemType();
146 0 : remoteRmaBufferSlice.addr =
147 0 : static_cast<void *>(static_cast<u8 *>(remoteBuffer->GetDevAddr()) + remoteDataOffSet);
148 0 : remoteRmaBufferSlice.len = byteSize;
149 0 : std::shared_ptr<RmaBuffer> temp(remoteBuffer, [](RmaBuffer* p){}); // 在外部进行删除操作，内部不能用智能指针进行生命周期管理
150 0 : remoteRmaBufferSlice.rmaBuffer = temp;
151 0 : remoteRmaBufferSlice.memType = remoteBuffer->GetMemType();
152 0 : HCCL_INFO("[TransportIpcMem][FillRmaBufferSlice] Local address before mapping is [%p], after mapping is [%p]."
153 : "Remote address before mapping is [%p], after mapping is [%p]. Datasize is [%llu].",
154 : localAddr, localRmaBufferSlice.addr, remoteAddr, remoteRmaBufferSlice.addr, byteSize);
155 0 : return HCCL_SUCCESS;
156 0 : }
157 :
158 0 : HcclResult TransportIpcMem::FillRmaBufferSlice(const RmaOpMem &localMem, const RmaOpMem &remoteMem,
159 : RmaBufferSlice &localRmaBufferSlice, RmaBufferSlice &remoteRmaBufferSlice)
160 : {
161 0 : void* remoteAddr = remoteMem.addr;
162 0 : void* localAddr = localMem.addr;
163 0 : u64 byteSize = std::min(remoteMem.size, localMem.size);
164 :
165 0 : auto localKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(localAddr), byteSize);
166 0 : auto remoteKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(remoteAddr), byteSize);
167 :
168 0 : NetDevContext *netDevCtx = static_cast<NetDevContext *>(netDevCtx_);
169 0 : std::shared_ptr<LocalIpcRmaBufferMgr> localRmaBufferMgr = netDevCtx->GetlocalIpcRmaBufferMgr();
170 0 : if (!localRmaBufferMgr) {
171 0 : HCCL_ERROR("[TransportIpcMem] can't get LocalIpcRmaBufferMgr");
172 0 : return HCCL_E_INTERNAL;
173 : }
174 :
175 0 : auto localBuffer = localRmaBufferMgr->Find(localKey);
176 0 : CHK_PRT_RET(!localBuffer.first,
177 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] Can't find localBuffer by key {%p, %llu}",
178 : localAddr, byteSize),
179 : HCCL_E_INTERNAL);
180 0 : CHK_PRT_RET(!localBuffer.second->GetAddr(),
181 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The addr of local Buffer or remote buffer is nullptr."),
182 : HCCL_E_NOT_FOUND);
183 0 : CHK_PRT_RET(!localBuffer.second->GetDevAddr(),
184 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The dev addr of local Buffer is nullptr."),
185 : HCCL_E_NOT_FOUND);
186 0 : CHK_RET(CheckHcclBuffer(localAddr, localBuffer.second.get()));
187 :
188 0 : auto remoteBuffer = remoteIpcRmaBufferMgr_.Find(remoteKey);
189 0 : CHK_PRT_RET(!remoteBuffer.first,
190 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] Can't find remoteBuffer by key {%p, %llu}",
191 : remoteAddr, byteSize),
192 : HCCL_E_INTERNAL);
193 0 : CHK_PRT_RET(!remoteBuffer.second->GetDevAddr(),
194 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The dev addr of remote buffer is nullptr."),
195 : HCCL_E_NOT_FOUND);
196 0 : CHK_PRT_RET(!remoteBuffer.second->GetAddr(),
197 : HCCL_ERROR("[TransportIpcMem][FillRmaBufferSlice] The addr of remote buffer is nullptr."),
198 : HCCL_E_NOT_FOUND);
199 0 : CHK_RET(CheckHcclBuffer(remoteAddr, remoteBuffer.second.get()));
200 :
201 0 : u64 localDataOffSet = static_cast<u8*>(localAddr) - static_cast<u8*>(localBuffer.second->GetAddr());
202 0 : u64 remoteDataOffSet = static_cast<u8*>(remoteAddr) - static_cast<u8*>(remoteBuffer.second->GetAddr());
203 0 : localRmaBufferSlice.addr = static_cast<void*>(static_cast<u8*>(localBuffer.second->GetDevAddr()) + localDataOffSet);
204 0 : localRmaBufferSlice.len = byteSize;
205 0 : localRmaBufferSlice.rmaBuffer = localBuffer.second;
206 0 : localRmaBufferSlice.memType = localBuffer.second->GetMemType();
207 0 : remoteRmaBufferSlice.addr =
208 0 : static_cast<void *>(static_cast<u8 *>(remoteBuffer.second->GetDevAddr()) + remoteDataOffSet);
209 0 : remoteRmaBufferSlice.len = byteSize;
210 0 : remoteRmaBufferSlice.rmaBuffer = remoteBuffer.second;
211 0 : remoteRmaBufferSlice.memType = remoteBuffer.second->GetMemType();
212 :
213 0 : HCCL_INFO("[TransportIpcMem][FillRmaBufferSlice] Local address before mapping is [%p], after mapping is [%p]."
214 : "Remote address before mapping is [%p], after mapping is [%p]. Datasize is [%llu].",
215 : localAddr, localRmaBufferSlice.addr, remoteAddr, remoteRmaBufferSlice.addr, byteSize);
216 0 : return HCCL_SUCCESS;
217 0 : }
218 :
219 :
220 0 : HcclResult TransportIpcMem::SetSocket(const std::shared_ptr<HcclSocket> &socket)
221 : {
222 0 : HCCL_INFO("TransportIpcMem doesn't need to set socket");
223 0 : return HCCL_SUCCESS;
224 : }
225 :
226 0 : HcclResult TransportIpcMem::Connect(s32 timeoutSec)
227 : {
228 0 : HCCL_INFO("TransportIpcMem doesn't need to connect socket");
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 0 : HcclResult TransportIpcMem::Write(
233 : const HcclBuf &remoteMem, const HcclBuf &localMem, const rtStream_t &stream)
234 : {
235 0 : CHK_PRT_RET((localMem.addr == nullptr) || (remoteMem.addr == nullptr),
236 : HCCL_ERROR("[TransportIpcMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
237 0 : CHK_PRT_RET((localMem.len == 0U) || (remoteMem.len == 0U),
238 : HCCL_ERROR(
239 : "[TransportIpcMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.len, remoteMem.len),
240 : HCCL_E_PARA);
241 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportIpcMem]stream is invalid"), HCCL_E_PARA);
242 :
243 0 : RmaBufferSlice localRmaBufferSlice{};
244 0 : RmaBufferSlice remoteRmaBufferSlice{};
245 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
246 0 : return TransportIpc(remoteRmaBufferSlice, localRmaBufferSlice, stream);
247 0 : }
248 :
249 0 : HcclResult TransportIpcMem::Write(
250 : const RmaOpMem &remoteMem, const RmaOpMem &localMem, const rtStream_t &stream)
251 : {
252 0 : CHK_PRT_RET((localMem.addr == nullptr) || (remoteMem.addr == nullptr),
253 : HCCL_ERROR("[TransportIpcMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
254 0 : CHK_PRT_RET((localMem.size == 0U) || (remoteMem.size == 0U),
255 : HCCL_ERROR(
256 : "[TransportIpcMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.size, remoteMem.size),
257 : HCCL_E_PARA);
258 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportIpcMem]stream is invalid"), HCCL_E_PARA);
259 :
260 0 : RmaBufferSlice localRmaBufferSlice{};
261 0 : RmaBufferSlice remoteRmaBufferSlice{};
262 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
263 0 : return TransportIpc(remoteRmaBufferSlice, localRmaBufferSlice, stream);
264 0 : }
265 :
266 0 : HcclResult TransportIpcMem::Read(
267 : const HcclBuf &localMem, const HcclBuf &remoteMem, const rtStream_t &stream)
268 : {
269 0 : CHK_PRT_RET((localMem.addr == nullptr) || (remoteMem.addr == nullptr),
270 : HCCL_ERROR("[TransportIpcMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
271 0 : CHK_PRT_RET((localMem.len == 0U) || (remoteMem.len == 0U),
272 : HCCL_ERROR(
273 : "[TransportIpcMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.len, remoteMem.len),
274 : HCCL_E_PARA);
275 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportIpcMem]stream is invalid"), HCCL_E_PARA);
276 :
277 0 : RmaBufferSlice localRmaBufferSlice{};
278 0 : RmaBufferSlice remoteRmaBufferSlice{};
279 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
280 0 : return TransportIpc(localRmaBufferSlice, remoteRmaBufferSlice, stream);
281 0 : }
282 :
283 0 : HcclResult TransportIpcMem::Read(
284 : const RmaOpMem &localMem, const RmaOpMem &remoteMem, const rtStream_t &stream)
285 : {
286 0 : CHK_PRT_RET((localMem.addr == nullptr) || (remoteMem.addr == nullptr),
287 : HCCL_ERROR("[TransportIpcMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
288 0 : CHK_PRT_RET((localMem.size == 0U) || (remoteMem.size == 0U),
289 : HCCL_ERROR(
290 : "[TransportIpcMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.size, remoteMem.size),
291 : HCCL_E_PARA);
292 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportIpcMem]stream is invalid"), HCCL_E_PARA);
293 :
294 0 : RmaBufferSlice localRmaBufferSlice{};
295 0 : RmaBufferSlice remoteRmaBufferSlice{};
296 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
297 0 : return TransportIpc(localRmaBufferSlice, remoteRmaBufferSlice, stream);
298 0 : }
299 :
300 0 : HcclResult TransportIpcMem::AddOpFence(const rtStream_t &stream)
301 : {
302 0 : HCCL_INFO("TransportIpcMem doesn't need to add op fence");
303 0 : return HCCL_SUCCESS;
304 : }
305 :
306 0 : HcclResult TransportIpcMem::GetMemInfo(u32 &lkey, u32 &rkey, HcclBuf &localMem, HcclBuf &remoteMem)
307 : {
308 0 : CHK_PRT_RET((localMem.addr == nullptr) || (remoteMem.addr == nullptr),
309 : HCCL_ERROR("[TransportIpcMem] localMem addr[%p] or remoteMem addr[%p] is invalid",
310 : localMem.addr, remoteMem.addr), HCCL_E_PARA);
311 0 : CHK_PRT_RET((localMem.len == 0U) || (remoteMem.len == 0U),
312 : HCCL_ERROR("[TransportIpcMem] localMem size[%llu] or remoteMem size[%llu]is invalid",
313 : localMem.len, remoteMem.len),
314 : HCCL_E_PARA);
315 :
316 0 : RmaBufferSlice localRmaBufferSlice{};
317 0 : RmaBufferSlice remoteRmaBufferSlice{};
318 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
319 0 : lkey = 0U;
320 0 : localMem.addr = localRmaBufferSlice.addr;
321 0 : localMem.len = localRmaBufferSlice.len;
322 :
323 0 : rkey = 0U;
324 0 : remoteMem.addr = remoteRmaBufferSlice.addr;
325 0 : remoteMem.len = remoteRmaBufferSlice.len;
326 :
327 0 : return HCCL_SUCCESS;
328 0 : }
329 :
330 0 : HcclResult TransportIpcMem::GetTransInfo(HcclQpInfoV2 &qpInfo, u32 *lkey, u32 *rkey, HcclBuf *localMem,
331 : HcclBuf *remoteMem, u32 num)
332 : {
333 0 : CHK_PTR_NULL(lkey);
334 0 : CHK_PTR_NULL(rkey);
335 0 : CHK_PTR_NULL(localMem);
336 0 : CHK_PTR_NULL(remoteMem);
337 0 : CHK_PRT_RET(num == 0, HCCL_ERROR("[GetTransInfo] mem num should not be zero"), HCCL_E_PARA);
338 :
339 : // GetTransInfo为TransportMem对外接口，TransportRoceMem将最后一个localMem用来反OpFence
340 : // Ipc mem无需Opfence，最后一个rkey/rkey/localMem/remoteMem空着
341 0 : for (u32 i = 0; i < num - 1; ++i) {
342 0 : HcclResult ret = GetMemInfo(lkey[i], rkey[i], localMem[i], remoteMem[i]);
343 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[GetTransInfo] failed at index[%u], localAddr[%p,%llu], "
344 : "remoteAddr[%p,%llu]", i, localMem[i].addr, localMem[i].len, remoteMem[i].addr, remoteMem[i].len), ret);
345 : }
346 0 : return HCCL_SUCCESS;
347 : }
348 :
349 0 : HcclResult TransportIpcMem::WaitOpFence(const rtStream_t &stream)
350 : {
351 0 : HCCL_DEBUG("TransportIpcMem doesn't need to wait fence");
352 0 : return HCCL_SUCCESS;
353 : }
354 :
355 0 : HcclResult TransportIpcMem::BatchWrite(const std::vector<MemDetails> &remoteMems,
356 : const std::vector<MemDetails> &localMems, Stream &stream)
357 : {
358 0 : HCCL_ERROR("TransportIpcMem doesn't support BatchWrite");
359 0 : return HCCL_E_NOT_SUPPORT;
360 : }
361 :
362 0 : HcclResult TransportIpcMem::BatchRead(const std::vector<MemDetails> &localMems,
363 : const std::vector<MemDetails> &remoteMems, Stream &stream)
364 : {
365 0 : HCCL_ERROR("TransportIpcMem doesn't support BatchRead");
366 0 : return HCCL_E_NOT_SUPPORT;
367 : }
368 :
369 0 : HcclResult TransportIpcMem::AddOpFence(const MemDetails &localFenceMem, const MemDetails &remoteFenceMem,
370 : Stream &stream)
371 : {
372 0 : HCCL_ERROR("TransportIpcMem doesn't support AICPU AddOpFence");
373 0 : return HCCL_E_NOT_SUPPORT;
374 : }
375 :
376 0 : HcclResult TransportIpcMem::TransportIpc(
377 : const RmaBufferSlice &dstRmaBufferSlice, const RmaBufferSlice &srcRmaBufferSlice, const rtStream_t &stream)
378 : {
379 0 : CHK_PTR_NULL(dstRmaBufferSlice.addr);
380 0 : CHK_PTR_NULL(srcRmaBufferSlice.addr);
381 0 : Stream hcclStream(stream);
382 0 : DispatcherPub* dispatcher = reinterpret_cast<DispatcherPub*>(dispatcher_);
383 0 : if (dstRmaBufferSlice.memType == RmaMemType::HOST ||
384 0 : srcRmaBufferSlice.memType == RmaMemType::HOST) {
385 0 : CHK_RET(dispatcher->MemcpyAsyncWithoutCheckKind(dstRmaBufferSlice.addr, dstRmaBufferSlice.len,
386 : srcRmaBufferSlice.addr, srcRmaBufferSlice.len, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE,
387 : hcclStream, remoteRankId_, hccl::LinkType::LINK_HCCS));
388 0 : } else {
389 0 : CHK_RET(dispatcher->MemcpyAsync(dstRmaBufferSlice.addr, dstRmaBufferSlice.len, srcRmaBufferSlice.addr,
390 : srcRmaBufferSlice.len, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, hcclStream,
391 : remoteRankId_, hccl::LinkType::LINK_HCCS));
392 : }
393 0 : return HCCL_SUCCESS;
394 0 : }
395 : } // namespace hccl
|
