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 : #include "ub_mem_transport.h"
11 : #include "serializable.h"
12 : #include "exchange_ub_buffer_dto.h"
13 : #include "exchange_ub_conn_dto.h"
14 : #include "local_ub_rma_buffer.h"
15 : #include "dev_capability.h"
16 : #include "dev_buffer.h"
17 : #include "../../common/dlprof_func_v2.h"
18 : #include "user_remote_mem_getter.h"
19 : #include "exception_util.h"
20 : #include "env_config/env_config_v2.h"
21 :
22 : namespace Hccl {
23 : constexpr u32 FINISH_MSG_SIZE = 128;
24 : constexpr char_t FINISH_MSG[FINISH_MSG_SIZE] = "Ub Comm Pipe ready!";
25 : constexpr u32 ONE_MILLISECOND_OF_USLEEP = 1000;
26 :
27 109 : UbMemTransport::UbMemTransport(
28 : CommonLocRes& commonLocRes, Attribution& attr, const LinkData& linkData, const Socket& socket,
29 109 : RdmaHandle rdmaHandle1, LocCntNotifyRes& locCntNotifyRes1, bool isRecvFirst)
30 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::UB),
31 109 : rdmaHandle(rdmaHandle1),
32 109 : locCntNotifyRes(locCntNotifyRes1),
33 109 : isRecvFirst_(isRecvFirst)
34 : {
35 327 : HCCL_INFO("source: %s", locCntNotifyRes.Describe().c_str());
36 109 : }
37 :
38 8 : UbMemTransport::UbMemTransport(
39 : CommonLocRes& commonLocRes, Attribution& attr, const LinkData& linkData, const Socket& socket,
40 : RdmaHandle rdmaHandle1, LocCntNotifyRes& locCntNotifyRes1,
41 8 : std::function<void(u32 streamId, u32 taskId, const TaskParam& taskParam)> callback)
42 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::UB, callback),
43 8 : rdmaHandle(rdmaHandle1),
44 8 : locCntNotifyRes(locCntNotifyRes1)
45 : {
46 24 : HCCL_INFO("source: %s", locCntNotifyRes.Describe().c_str());
47 8 : }
48 :
49 10 : std::string UbMemTransport::Describe() const
50 : {
51 : string msg = StringFormat(
52 20 : "UbMemTransport=[commonLocRes=%s, locCntNotifyRes=%s, ubStatus=%s, ", commonLocRes.Describe().c_str(),
53 30 : locCntNotifyRes.Describe().c_str(), ubStatus.Describe().c_str());
54 10 : msg += StringFormat("exchangeDataSize=%u, ", exchangeDataSize);
55 10 : msg += StringFormat("rmtNotifyNum=%zu, rmtCntNotifyVecNum=%zu]", rmtNotifyVec.size(), rmtCntNotifyVec.size());
56 10 : return msg;
57 0 : }
58 :
59 1 : HcclResult UbMemTransport::BuildDrainResource()
60 : {
61 : // notify作为read的落点
62 1 : if (drainNotify_ == nullptr) {
63 1 : bool devUsed = true;
64 1 : EXCEPTION_CATCH(drainNotify_ = std::make_unique<Hccl::UbLocalNotify>(rdmaHandle, devUsed), return HCCL_E_PTR);
65 3 : HCCL_INFO("[UbMemTransport][%s] drain notify created: %s", __func__, drainNotify_->Describe().c_str());
66 : }
67 :
68 : // 常量1内存供远端读取
69 1 : if (drainBuffer_ == nullptr) {
70 1 : u32 notifySize = Hccl::DevCapability::GetInstance().GetNotifySize();
71 :
72 1 : std::shared_ptr<Hccl::DevBuffer> constMem;
73 1 : EXCEPTION_CATCH(constMem = std::make_shared<Hccl::DevBuffer>(notifySize), return HCCL_E_PTR);
74 :
75 1 : Hccl::HrtMemcpy(
76 2 : reinterpret_cast<void*>(constMem->GetAddr()), constMem->GetSize(), &NORMAL_NOTIFY_VAL,
77 : sizeof(NORMAL_NOTIFY_VAL), RT_MEMCPY_HOST_TO_DEVICE);
78 :
79 1 : EXCEPTION_CATCH(
80 : drainBuffer_ = std::make_unique<Hccl::LocalUbRmaBuffer>(constMem, rdmaHandle), return HCCL_E_PTR);
81 3 : HCCL_INFO(
82 : "[UbMemTransport][%s] drain buffer created: addr[0x%llx], size[%zu]", __func__,
83 : static_cast<unsigned long long>(drainBuffer_->GetAddr()), drainBuffer_->GetSize());
84 1 : }
85 :
86 1 : return HCCL_SUCCESS;
87 : }
88 :
89 0 : HcclResult UbMemTransport::Describe(std::string& dfxMsg)
90 : {
91 0 : HCCL_INFO("UbMemTransport Describe connNum[%u]", connNum);
92 0 : for (u32 i = 0; i < connNum; i++) {
93 0 : CHK_RET(commonLocRes.connVec[i]->Describe(dfxMsg));
94 : }
95 0 : return HCCL_SUCCESS;
96 : }
97 :
98 8 : MemoryBuffer UbMemTransport::GetLocMemBuffer(const RmaBufferSlice& locSlice) const
99 : {
100 8 : return MemoryBuffer(locSlice.addr, locSlice.size, locSlice.buf->GetMemHandle());
101 : }
102 :
103 8 : MemoryBuffer UbMemTransport::GetRmtMemBuffer(const RmtRmaBufferSlice& rmtSlice) const
104 : {
105 8 : return MemoryBuffer(rmtSlice.addr, rmtSlice.size, rmtSlice.buf->GetMemHandle());
106 : }
107 :
108 5 : MemoryBuffer UbMemTransport::GetRmtNotifyMemBuffer(u32 index)
109 : {
110 : return MemoryBuffer(
111 5 : rmtNotifyVec[index]->GetAddr(), rmtNotifyVec[index]->GetSize(), rmtNotifyVec[index]->GetMemHandle());
112 : }
113 :
114 4 : MemoryBuffer UbMemTransport::GetRmtCntNotifyMemBuffer(const WithNotifyIn& withNotify)
115 : {
116 4 : auto index = withNotify.index_;
117 : return MemoryBuffer(
118 4 : rmtCntNotifyVec[index]->GetAddr(), rmtCntNotifyVec[index]->GetSize(), rmtCntNotifyVec[index]->GetMemHandle());
119 : }
120 :
121 5 : static void SubmitTask(const TaskUbDbSend& ubSend, const Stream& stream)
122 : {
123 15 : HCCL_INFO("SubmitTask UbDbSend ");
124 : HrtUbDbInfo info;
125 5 : info.dbNum = 1;
126 5 : info.wrCqe = 0; // 默认值是0 不会cqe 如果传1，驱动分发，会给hccl cqe，用于维护ci指针。
127 5 : info.info[0].functionId = ubSend.GetFuncId();
128 5 : info.info[0].dieId = ubSend.GetDieId();
129 5 : info.info[0].jettyId = ubSend.GetJettyId();
130 5 : info.info[0].piValue = ubSend.GetPiVal();
131 5 : HrtUbDbSend(info, stream.GetPtr());
132 0 : }
133 :
134 1 : static void SubmitTask(const TaskUbDirectSend& ubDirectSend, const Stream& stream)
135 : {
136 3 : HCCL_INFO("SubmitTask UbDirectSend");
137 1 : if (ubDirectSend.GetDwqeSize() != DWQE_SIZE_64 && ubDirectSend.GetDwqeSize() != DWQE_SIZE_128) {
138 : std::string msg
139 0 : = StringFormat("dwqe size is not valid, cannot submit task, dwqeSize=%u", ubDirectSend.GetDwqeSize());
140 0 : THROW<InternalException>(msg);
141 0 : }
142 : HrtUbWqeInfo info;
143 1 : info.wrCqe = 0;
144 1 : info.functionId = ubDirectSend.GetFuncId();
145 1 : info.dieId = ubDirectSend.GetDieId();
146 1 : info.jettyId = ubDirectSend.GetJettyId();
147 1 : info.wqe = const_cast<u8*>(ubDirectSend.GetDwqePtr());
148 1 : info.wqePtrLen = ubDirectSend.GetDwqeSize();
149 1 : info.wqeSize = info.wqePtrLen == DWQE_SIZE_64 ? 0 : 1;
150 1 : HrtUbDirectSend(info, stream.GetPtr());
151 0 : }
152 :
153 5 : static void SubmitTask(const TaskWriteValue& taskWriteValue, const Stream& stream)
154 : {
155 15 : HCCL_INFO("begin HrtWriteValue");
156 5 : HrtWriteValue(taskWriteValue.GetDbAddr(), taskWriteValue.GetPiVal(), stream.GetPtr());
157 0 : HCCL_INFO("finished HrtWriteValue");
158 0 : }
159 :
160 : template <typename TaskType>
161 3 : std::function<void(const BaseTask&, const Stream&)> GetSubmitUbTaskFunction()
162 : {
163 14 : return [](const BaseTask& task, const Stream& stream) {
164 11 : SubmitTask(static_cast<const TaskType&>(task), stream);
165 3 : };
166 : }
167 :
168 : std::map<TaskType, std::function<void(const BaseTask&, const Stream&)>> g_ubTaskSubmitRuleMap
169 : = {{TaskType::UB_SEND, GetSubmitUbTaskFunction<TaskUbDbSend>()},
170 : {TaskType::UB_DIRECT_SEND, GetSubmitUbTaskFunction<TaskUbDirectSend>()},
171 : {TaskType::WRITE_VALUE, GetSubmitUbTaskFunction<TaskWriteValue>()}};
172 :
173 13 : static void SubmitUbTask(unique_ptr<BaseTask> task, const Stream& stream)
174 : {
175 13 : if (task != nullptr) {
176 11 : g_ubTaskSubmitRuleMap.at(task->GetType())(*task.get(), stream);
177 : }
178 2 : }
179 :
180 5 : void UbMemTransport::SubmitNotify(const MemoryBuffer& rmtNotify, u64 data, const Stream& stream)
181 : {
182 5 : SqeConfig config;
183 10 : SubmitUbTask(commonLocRes.connVec[0]->PrepareInlineWrite(rmtNotify, data, config), stream);
184 0 : }
185 :
186 1 : void UbMemTransport::Post(u32 index, const Stream& stream)
187 : {
188 1 : TaskParam taskParam{};
189 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
190 :
191 1 : SubmitNotify(GetRmtNotifyMemBuffer(index), NORMAL_NOTIFY_VAL, stream);
192 :
193 0 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
194 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
195 : ;
196 0 : taskParam.taskPara.Notify.notifyID = rmtNotifyVec[index]->GetAddr();
197 0 : taskParam.taskPara.Notify.value = NORMAL_NOTIFY_VAL;
198 :
199 0 : SaveDfxTaskInfo(taskParam);
200 1 : }
201 :
202 1 : void UbMemTransport::Wait(u32 index, const Stream& stream, u32 timeout)
203 : {
204 1 : TaskParam taskParam{};
205 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
206 :
207 1 : commonLocRes.notifyVec[index]->Wait(stream, timeout);
208 :
209 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
210 1 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
211 1 : taskParam.taskPara.Notify.notifyID = commonLocRes.notifyVec[index]->GetNotify()->GetId();
212 1 : taskParam.taskPara.Notify.value = NORMAL_NOTIFY_VAL;
213 :
214 1 : SaveDfxTaskInfo(taskParam);
215 1 : }
216 :
217 1 : void UbMemTransport::Read(const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const Stream& stream)
218 : {
219 1 : TaskParam taskParam{};
220 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
221 :
222 1 : SqeConfig config;
223 1 : config.wqeMode = WqeMode::DWQE;
224 1 : SubmitUbTask(
225 3 : commonLocRes.connVec[0]->PrepareRead(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config), stream);
226 :
227 0 : taskParam.taskType = TaskParamType::TASK_RDMA;
228 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
229 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
230 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
231 0 : taskParam.taskPara.DMA.size = rmtSlice.size;
232 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
233 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
234 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
235 0 : SaveDfxTaskInfo(taskParam);
236 1 : }
237 :
238 1 : void UbMemTransport::ReadReduce(
239 : const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const ReduceIn& reduceIn, const Stream& stream)
240 : {
241 1 : TaskParam taskParam{};
242 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
243 :
244 1 : SqeConfig config;
245 1 : config.wqeMode = WqeMode::DWQE;
246 1 : SubmitUbTask(
247 4 : commonLocRes.connVec[0]->PrepareReadReduce(
248 2 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn.dataType, reduceIn.reduceOp, config),
249 : stream);
250 :
251 0 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
252 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
253 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
254 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
255 0 : taskParam.taskPara.DMA.size = rmtSlice.size;
256 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
257 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
258 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
259 :
260 0 : SaveDfxTaskInfo(taskParam);
261 1 : }
262 :
263 1 : void UbMemTransport::Write(const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const Stream& stream)
264 : {
265 1 : TaskParam taskParam{};
266 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
267 :
268 1 : SqeConfig config;
269 1 : config.wqeMode = WqeMode::DWQE;
270 1 : SubmitUbTask(
271 3 : commonLocRes.connVec[0]->PrepareWrite(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config), stream);
272 0 : taskParam.taskType = TaskParamType::TASK_RDMA;
273 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
274 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
275 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
276 0 : taskParam.taskPara.DMA.size = locSlice.size;
277 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
278 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
279 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
280 :
281 0 : SaveDfxTaskInfo(taskParam);
282 1 : }
283 :
284 1 : void UbMemTransport::WriteReduce(
285 : const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const ReduceIn& reduceIn, const Stream& stream)
286 : {
287 1 : TaskParam taskParam{};
288 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
289 :
290 1 : SqeConfig config;
291 1 : config.wqeMode = WqeMode::DWQE;
292 1 : SubmitUbTask(
293 4 : commonLocRes.connVec[0]->PrepareWriteReduce(
294 2 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn.dataType, reduceIn.reduceOp, config),
295 : stream);
296 :
297 0 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
298 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
299 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
300 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
301 0 : taskParam.taskPara.DMA.size = locSlice.size;
302 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
303 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
304 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
305 :
306 0 : SaveDfxTaskInfo(taskParam);
307 1 : }
308 :
309 5 : void UbMemTransport::WriteWithNotify(
310 : const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const WithNotifyIn& withNotify,
311 : const Stream& stream)
312 : {
313 5 : if (locSlice.size == 0) {
314 2 : return SubmitWriteEmptyWithNotify(withNotify, stream);
315 : }
316 :
317 3 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
318 1 : return SubmitWriteWithNotify(
319 2 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), NORMAL_NOTIFY_VAL,
320 2 : GetRmtNotifyMemBuffer(withNotify.index_), stream);
321 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
322 1 : return SubmitWriteWithNotify(
323 2 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), withNotify.userData_,
324 2 : GetRmtCntNotifyMemBuffer(withNotify), stream);
325 : } else {
326 1 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
327 1 : THROW<InternalException>(msg);
328 1 : }
329 : }
330 :
331 5 : void UbMemTransport::WriteReduceWithNotify(
332 : const RmaBufferSlice& locSlice, const RmtRmaBufferSlice& rmtSlice, const ReduceIn& reduceIn,
333 : const WithNotifyIn& withNotify, const Stream& stream)
334 : {
335 5 : if (locSlice.size == 0) {
336 2 : return SubmitWriteEmptyWithNotify(withNotify, stream);
337 : }
338 :
339 3 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
340 1 : SubmitWriteReduceWithNotify(
341 1 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn, NORMAL_NOTIFY_VAL,
342 2 : GetRmtNotifyMemBuffer(withNotify.index_), stream);
343 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
344 1 : SubmitWriteReduceWithNotify(
345 1 : GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn, withNotify.userData_,
346 2 : GetRmtCntNotifyMemBuffer(withNotify), stream);
347 : } else {
348 1 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
349 1 : THROW<InternalException>(msg);
350 1 : }
351 : }
352 :
353 4 : void UbMemTransport::SubmitWriteEmptyWithNotify(const WithNotifyIn& withNotify, const Stream& stream)
354 : {
355 4 : TaskParam taskParam{};
356 4 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
357 4 : u32 value = NORMAL_NOTIFY_VAL;
358 :
359 4 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
360 2 : SubmitNotify(GetRmtNotifyMemBuffer(withNotify.index_), NORMAL_NOTIFY_VAL, stream);
361 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
362 2 : SubmitNotify(GetRmtCntNotifyMemBuffer(withNotify), withNotify.userData_, stream);
363 0 : value = withNotify.userData_;
364 : } else {
365 0 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
366 0 : THROW<InternalException>(msg);
367 0 : }
368 :
369 0 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
370 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
371 0 : taskParam.taskPara.Notify.notifyID = INVALID_VALUE_NOTIFYID;
372 0 : taskParam.taskPara.Notify.value = value;
373 :
374 0 : SaveDfxTaskInfo(taskParam);
375 4 : }
376 :
377 2 : void UbMemTransport::SubmitWriteWithNotify(
378 : const MemoryBuffer& rmt, const MemoryBuffer& loc, u64 data, const MemoryBuffer& rmtNotify, const Stream& stream)
379 : {
380 2 : TaskParam taskParam{};
381 2 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
382 :
383 2 : SqeConfig config;
384 2 : config.wqeMode = WqeMode::DWQE;
385 4 : SubmitUbTask(commonLocRes.connVec[0]->PrepareWriteWithNotify(rmt, loc, data, rmtNotify, config), stream);
386 :
387 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
388 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
389 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(loc.addr);
390 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmt.addr);
391 0 : taskParam.taskPara.DMA.size = loc.size;
392 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
393 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
394 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
395 :
396 0 : SaveDfxTaskInfo(taskParam);
397 2 : }
398 :
399 2 : void UbMemTransport::SubmitWriteReduceWithNotify(
400 : const MemoryBuffer& rmt, const MemoryBuffer& loc, const ReduceIn& reduceIn, u64 data, const MemoryBuffer& rmtNotify,
401 : const Stream& stream)
402 : {
403 2 : TaskParam taskParam{};
404 2 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
405 :
406 2 : SqeConfig config;
407 2 : config.wqeMode = WqeMode::DWQE;
408 2 : SubmitUbTask(
409 4 : commonLocRes.connVec[0]->PrepareWriteReduceWithNotify(
410 : rmt, loc, reduceIn.dataType, reduceIn.reduceOp, data, rmtNotify, config),
411 : stream);
412 :
413 2 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
414 2 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
415 2 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(loc.addr);
416 2 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmt.addr);
417 2 : taskParam.taskPara.DMA.size = loc.size;
418 2 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
419 2 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
420 2 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
421 :
422 2 : SaveDfxTaskInfo(taskParam);
423 2 : }
424 :
425 5 : bool UbMemTransport::IsResReady()
426 : {
427 7 : for (auto& it : commonLocRes.connVec) {
428 3 : CHECK_NULLPTR(it, StringFormat("[UbMemTransport::%s] failed, connection pointer is nullptr", __func__));
429 :
430 3 : RmaConnType connType = it->GetRmaConnType();
431 3 : if (connType != RmaConnType::UB) {
432 0 : THROW<InternalException>(
433 0 : "[UbMemTransport::%s] connection type[%s] is not ub", __func__, connType.Describe().c_str());
434 : }
435 :
436 3 : auto status = it->GetStatus();
437 3 : if (status != RmaConnStatus::EXCHANGEABLE && status != RmaConnStatus::READY) {
438 1 : return false;
439 : }
440 : }
441 :
442 12 : HCCL_INFO("[UbMemTransport::IsResReady] all resources ready.");
443 4 : return true;
444 : }
445 :
446 4 : bool UbMemTransport::IsConnsReady()
447 : {
448 4 : for (u32 i = 0; i < connNum; i++) {
449 1 : if (commonLocRes.connVec[i]->GetStatus() != RmaConnStatus::READY) {
450 1 : return false;
451 : }
452 : }
453 9 : HCCL_INFO("conns are ready.");
454 3 : return true;
455 : }
456 :
457 15 : HcclResult UbMemTransport::StatusMachine()
458 : {
459 18 : TRY_CATCH_RETURN(
460 : if (socket == nullptr) {
461 : HCCL_ERROR("[UbMemTransport][StatusMachine]socket is nullptr, please check");
462 : return HcclResult::HCCL_E_INTERNAL;
463 : } SocketStatus socketStatus
464 : = isHost_ ? socket->GetStatus() : socket->GetAsyncStatus();
465 : if (socketStatus == Hccl::SocketStatus::INIT || socketStatus == Hccl::SocketStatus::TIMEOUT) {
466 : HCCL_ERROR("[UbMemTransport][StatusMachine]socket timeout or no link, please check");
467 : return HcclResult::HCCL_E_INTERNAL;
468 : }
469 :
470 : if (socketStatus != Hccl::SocketStatus::OK) {
471 : SaluSleep(ONE_MILLISECOND_OF_USLEEP); // 防止get sockets冲高CtrlCPU
472 : return HcclResult::HCCL_SUCCESS; // 操作成功，保持当前状态
473 : } switch (ubStatus) {
474 : case UbStatus::INIT:
475 : CHK_RET(HandleInitStatus());
476 : break;
477 : case UbStatus::SEND_DATA:
478 : CHK_RET(HandleSendAllStatus());
479 : break;
480 : case UbStatus::RECV_SIZE:
481 : CHK_RET(HandleRecvSizeStatus());
482 : break;
483 : case UbStatus::RECV_DATA:
484 : CHK_RET(HandleRecvDataStatus());
485 : break;
486 : case UbStatus::PROCESS_DATA:
487 : CHK_RET(HandleProcessDataStatus());
488 : break;
489 : case UbStatus::SEND_FIN:
490 : CHK_RET(HandleSendFinStatus());
491 : break;
492 : case UbStatus::RECV_FIN:
493 : CHK_RET(HandleRecvFinStatus());
494 : break;
495 : case UbStatus::SET_READY:
496 : CHK_RET(HandleSetReadyStatus());
497 : break;
498 : default:
499 : break;
500 : });
501 14 : return HCCL_SUCCESS;
502 : }
503 :
504 4 : HcclResult UbMemTransport::HandleInitStatus()
505 : {
506 4 : ubStatus = isRecvFirst_ ? UbStatus::RECV_SIZE : UbStatus::SEND_DATA;
507 4 : baseStatus = TransportStatus::SOCKET_OK;
508 4 : return HCCL_SUCCESS;
509 : }
510 :
511 4 : HcclResult UbMemTransport::HandleSendAllStatus()
512 : {
513 4 : if (IsResReady()) {
514 3 : CHK_RET(SendAll());
515 3 : ubStatus = isRecvFirst_ ? UbStatus::PROCESS_DATA : UbStatus::RECV_SIZE;
516 : }
517 4 : return HCCL_SUCCESS;
518 : }
519 :
520 5 : HcclResult UbMemTransport::HandleRecvSizeStatus()
521 : {
522 8 : CHK_RET(RecvDataSize());
523 4 : ubStatus = UbStatus::RECV_DATA;
524 4 : return HCCL_SUCCESS;
525 : }
526 :
527 5 : HcclResult UbMemTransport::HandleRecvDataStatus()
528 : {
529 8 : CHK_RET(RecvExchangeData());
530 4 : ubStatus = isRecvFirst_ ? UbStatus::SEND_DATA : UbStatus::PROCESS_DATA;
531 4 : return HCCL_SUCCESS;
532 : }
533 :
534 5 : HcclResult UbMemTransport::HandleProcessDataStatus()
535 : {
536 5 : bool needSendFinish = false;
537 8 : CHK_RET(RecvDataProcess(needSendFinish));
538 4 : if (needSendFinish) {
539 2 : ubStatus = UbStatus::SEND_FIN;
540 : } else {
541 2 : SetBaseStatusReady();
542 2 : ubStatus = UbStatus::READY;
543 : }
544 4 : return HCCL_SUCCESS;
545 : }
546 :
547 4 : HcclResult UbMemTransport::HandleSendFinStatus()
548 : {
549 4 : if (IsConnsReady()) {
550 6 : CHK_RET(SendFinish());
551 2 : ubStatus = UbStatus::RECV_FIN;
552 : }
553 3 : return HCCL_SUCCESS;
554 : }
555 :
556 3 : HcclResult UbMemTransport::HandleRecvFinStatus()
557 : {
558 6 : CHK_RET(RecvFinish());
559 2 : ubStatus = UbStatus::SET_READY;
560 2 : return HCCL_SUCCESS;
561 : }
562 :
563 2 : HcclResult UbMemTransport::HandleSetReadyStatus()
564 : {
565 2 : SetBaseStatusReady();
566 2 : ubStatus = UbStatus::READY;
567 2 : return HCCL_SUCCESS;
568 : }
569 :
570 13 : TransportStatus UbMemTransport::GetStatus()
571 : {
572 26 : if (baseStatus == TransportStatus::READY || baseStatus == TransportStatus::CONNECT_FAILED
573 26 : || baseStatus == TransportStatus::SOCKET_TIMEOUT) {
574 0 : return baseStatus;
575 : }
576 :
577 13 : HcclResult ret = StatusMachine();
578 13 : if (ret != HCCL_SUCCESS) {
579 0 : HCCL_ERROR("[UbMemTransport::GetStatus] StatusMachine failed, ret=%d", ret);
580 0 : baseStatus = TransportStatus::CONNECT_FAILED;
581 : }
582 13 : return baseStatus;
583 : }
584 :
585 1 : HcclResult UbMemTransport::SendAll()
586 : {
587 1 : notifyNum = commonLocRes.notifyVec.size();
588 1 : bufferNum = commonLocRes.bufferVec.size();
589 1 : connNum = commonLocRes.connVec.size();
590 1 : cntNotifyNum = locCntNotifyRes.vec.size();
591 :
592 1 : cntNotifyDescSize = locCntNotifyRes.desc.size();
593 :
594 3 : HCCL_INFO(
595 : "notifyNum=%u, bufferNum=%u, connNum=%u, cntNotifyNum=%u, cntNotifyDescSize=%u, isHost_[%d]", notifyNum,
596 : bufferNum, connNum, cntNotifyNum, cntNotifyDescSize, isHost_);
597 :
598 1 : BinaryStream binaryStream;
599 1 : HandshakeMsgPack(binaryStream);
600 1 : NotifyVecPack(binaryStream);
601 1 : BufferVecPack(binaryStream, commonLocRes.bufferVec);
602 1 : CntNotifyVecPack(binaryStream);
603 1 : CntNotifyDescPack(binaryStream);
604 1 : CHK_RET(DrainBufPack(binaryStream));
605 1 : ConnVecPack(binaryStream);
606 :
607 1 : sendDataPack_.resize(sizeof(u32));
608 1 : binaryStream.Dump(sendDataPack_);
609 1 : u32 dataSize = sendDataPack_.size() - sizeof(u32);
610 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sendDataPack_.data(), sizeof(u32), &dataSize, sizeof(u32)));
611 :
612 1 : bool ret = false;
613 1 : if (isHost_) {
614 0 : ret = socket->Send(sendDataPack_.data(), sendDataPack_.size());
615 : } else {
616 1 : socket->SendAsync(sendDataPack_.data(), sendDataPack_.size());
617 1 : ret = true;
618 : }
619 1 : if (!ret) {
620 0 : HCCL_ERROR("[UbMemTransport::SendAll] Send failed");
621 0 : return HCCL_E_INTERNAL;
622 : }
623 3 : HCCL_INFO("[UbMemTransport::%s] Send size[%zu] of data success.", __func__, sendDataPack_.size());
624 1 : return HCCL_SUCCESS;
625 1 : }
626 :
627 1 : HcclResult UbMemTransport::RecvDataSize()
628 : {
629 : // 接收数据包尺寸
630 1 : bool ret = false;
631 3 : HCCL_DEBUG("Starting to recv message size[%zu] bytes, isHost_[%d]", sizeof(exchangeDataSize), isHost_);
632 1 : if (isHost_) {
633 0 : ret = socket->Recv(&exchangeDataSize, sizeof(exchangeDataSize));
634 : } else {
635 1 : socket->RecvAsync(reinterpret_cast<u8*>(&exchangeDataSize), sizeof(exchangeDataSize));
636 1 : ret = true;
637 : }
638 1 : if (!ret) {
639 0 : HCCL_ERROR("[UbMemTransport::RecvDataSize] Recv size failed");
640 0 : return HCCL_E_INTERNAL;
641 : }
642 3 : HCCL_INFO(
643 : "[UbMemTransport::%s] Receive size[%u] of data success. [%zu] bytes received.", __func__, exchangeDataSize,
644 : sizeof(exchangeDataSize));
645 1 : return HCCL_SUCCESS;
646 : }
647 :
648 1 : HcclResult UbMemTransport::SendExchangeData()
649 : {
650 1 : bool ret = false;
651 3 : HCCL_DEBUG("Starting to send message size[%zu] bytes, isHost_[%d]", sendData.size(), isHost_);
652 1 : if (isHost_) {
653 0 : ret = socket->Send(sendData.data(), sendData.size());
654 : } else {
655 1 : socket->SendAsync(sendData.data(), sendData.size());
656 1 : ret = true;
657 : }
658 1 : if (!ret) {
659 0 : HCCL_ERROR("[UbMemTransport::SendExchangeData] Send data failed");
660 0 : return HCCL_E_INTERNAL;
661 : }
662 3 : HCCL_INFO("send data %s, size=%zu", GetLinkDescInfo().c_str(), sendData.size());
663 1 : return HCCL_SUCCESS;
664 : }
665 :
666 0 : HcclResult UbMemTransport::RecvExchangeData()
667 : {
668 0 : recvData.resize(exchangeDataSize);
669 0 : bool ret = false;
670 0 : HCCL_DEBUG("Starting to recv message size[%zu] bytes, isHost_[%d]", recvData.size(), isHost_);
671 0 : if (isHost_) {
672 0 : ret = socket->Recv(recvData.data(), recvData.size());
673 : } else {
674 0 : socket->RecvAsync(reinterpret_cast<u8*>(recvData.data()), recvData.size());
675 0 : ret = true;
676 : }
677 0 : if (!ret) {
678 0 : HCCL_ERROR("[UbMemTransport::RecvExchangeData] Recv data failed");
679 0 : return HCCL_E_INTERNAL;
680 : }
681 :
682 0 : HCCL_INFO("recv data %s, size=%zu", GetLinkDescInfo().c_str(), recvData.size());
683 0 : return HCCL_SUCCESS;
684 : }
685 :
686 0 : HcclResult UbMemTransport::RecvDataProcess(bool& needSendFinish)
687 : {
688 0 : HCCL_INFO(
689 : "RecvDataProcess: link=%s, size=%zu, exchangeDataSize=%u", GetLinkDescInfo().c_str(), recvData.size(),
690 : exchangeDataSize);
691 0 : BinaryStream binaryStream(recvData);
692 0 : HcclResult ret = HandshakeMsgUnpack(binaryStream);
693 0 : if (ret != HCCL_SUCCESS) {
694 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] HandshakeMsgUnpack failed, ret=%d", ret);
695 0 : return ret;
696 : }
697 :
698 0 : ret = RmtBufferVecUnpackProc(notifyNum, binaryStream, rmtNotifyVec, UbRmtBufType::NOTIFY);
699 0 : if (ret != HCCL_SUCCESS) {
700 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc notify failed, ret=%d", ret);
701 0 : return ret;
702 : }
703 :
704 0 : ret = RmtBufferVecUnpackProc(bufferNum, binaryStream, rmtBufferVec, UbRmtBufType::BUFFER);
705 0 : if (ret != HCCL_SUCCESS) {
706 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc buffer failed, ret=%d", ret);
707 0 : return ret;
708 : }
709 :
710 0 : ret = RmtBufferVecUnpackProc(cntNotifyNum, binaryStream, rmtCntNotifyVec, UbRmtBufType::CNT_NOTIFY);
711 0 : if (ret != HCCL_SUCCESS) {
712 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc cntNotify failed, ret=%d", ret);
713 0 : return ret;
714 : }
715 :
716 0 : ret = CntNotifyDescUnpack(binaryStream);
717 0 : if (ret != HCCL_SUCCESS) {
718 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] CntNotifyDescUnpack failed, ret=%d", ret);
719 0 : return ret;
720 : }
721 :
722 0 : ret = DrainBufUnpack(binaryStream);
723 0 : if (ret != HCCL_SUCCESS) {
724 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] DrainBufUnpack failed, ret=%d", ret);
725 0 : return ret;
726 : }
727 :
728 0 : ret = ConnVecUnpackProc(binaryStream, needSendFinish);
729 0 : if (ret != HCCL_SUCCESS) {
730 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] ConnVecUnpackProc failed, ret=%d", ret);
731 0 : return ret;
732 : }
733 :
734 0 : return HCCL_SUCCESS;
735 0 : }
736 :
737 6 : void UbMemTransport::BufferVecPack(BinaryStream& binaryStream, std::vector<LocalRmaBuffer*>& bufferVec)
738 : {
739 6 : binaryStream << static_cast<u32>(bufferVec.size());
740 18 : HCCL_INFO("start pack %s bufferVec", transportType.Describe().c_str());
741 6 : u32 pos = 0;
742 13 : for (auto& it : bufferVec) {
743 7 : binaryStream << pos;
744 7 : if (it != nullptr) { // 非空的buffer，从buffer中获取 dto
745 7 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
746 7 : dto->Serialize(binaryStream);
747 21 : HCCL_INFO("pack buffer pos=%u dto %s", pos, dto->Describe().c_str());
748 7 : } else { // 空的buffer，dto所有字段为0(size=0)
749 0 : ExchangeUbBufferDto exchangeDto;
750 0 : exchangeDto.Serialize(binaryStream);
751 0 : HCCL_INFO("pack buffer pos=%u, dto is null %s", pos, exchangeDto.Describe().c_str());
752 0 : }
753 7 : pos++;
754 : }
755 6 : }
756 :
757 1 : void UbMemTransport::CntNotifyVecPack(BinaryStream& binaryStream)
758 : {
759 1 : binaryStream << cntNotifyNum;
760 3 : HCCL_INFO("pack UB cntNotify num=%u, %s", cntNotifyNum, GetLinkDescInfo().c_str());
761 1 : u32 pos = 0;
762 2 : for (auto& it : locCntNotifyRes.vec) {
763 1 : binaryStream << pos;
764 1 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
765 1 : dto->Serialize(binaryStream);
766 3 : HCCL_INFO("pack cntNotify pos=%u, dto %s", pos, dto->Describe().c_str());
767 1 : pos++;
768 1 : }
769 1 : }
770 :
771 1 : void UbMemTransport::CntNotifyDescPack(BinaryStream& binaryStream)
772 : {
773 1 : binaryStream << cntNotifyDescSize;
774 3 : HCCL_INFO("pack cntNotify desc size=%u %s", cntNotifyDescSize, GetLinkDescInfo().c_str());
775 3 : HCCL_INFO("pack cntNotify desc =%s", Bytes2hex(locCntNotifyRes.desc.data(), locCntNotifyRes.desc.size()).c_str());
776 3 : for (auto& it : locCntNotifyRes.desc) {
777 2 : binaryStream << it;
778 : }
779 1 : }
780 :
781 1 : HcclResult UbMemTransport::DrainBufPack(BinaryStream& binaryStream)
782 : {
783 : // 打包交换信息前，进行资源创建
784 1 : CHK_RET(BuildDrainResource());
785 :
786 : // 只需交换 常量buffer信息 供远端读
787 3 : HCCL_INFO("start pack drain buffer");
788 1 : if (drainBuffer_ != nullptr) { // 非空的buffer，从buffer中获取 dto
789 1 : std::unique_ptr<Serializable> dto = drainBuffer_->GetExchangeDto();
790 1 : dto->Serialize(binaryStream);
791 3 : HCCL_INFO("pack drain buffer dto %s", dto->Describe().c_str());
792 1 : } else { // 空的buffer，dto所有字段为0(size=0)
793 0 : ExchangeUbBufferDto exchangeDto;
794 0 : exchangeDto.Serialize(binaryStream);
795 0 : HCCL_INFO("pack drain buffer, dto is null %s", exchangeDto.Describe().c_str());
796 0 : }
797 :
798 1 : return HCCL_SUCCESS;
799 : }
800 :
801 0 : HcclResult UbMemTransport::DrainBufUnpack(BinaryStream& binaryStream)
802 : {
803 0 : HCCL_INFO("start unpack drain buffer");
804 0 : ExchangeUbBufferDto dto;
805 0 : dto.Deserialize(binaryStream);
806 :
807 0 : if (dto.size == 0) {
808 0 : rmtDrainBuffer_ = nullptr;
809 0 : HCCL_WARNING("unpack drain buffer dto is null");
810 : } else {
811 0 : rmtDrainBuffer_ = std::make_unique<RemoteUbRmaBuffer>(rdmaHandle, dto);
812 0 : HCCL_INFO("unpack drain buffer rmtDrainBuffer=%s", rmtDrainBuffer_->Describe().c_str());
813 : }
814 :
815 0 : return HCCL_SUCCESS;
816 0 : }
817 :
818 0 : HcclResult UbMemTransport::CntNotifyDescUnpack(BinaryStream& binaryStream)
819 : {
820 : u32 descSize;
821 0 : binaryStream >> descSize;
822 0 : if (descSize != cntNotifyDescSize) {
823 0 : HCCL_ERROR(
824 : "[UbMemTransport::CntNotifyDescUnpack] size=%u is not equal to rmtNum=%u", descSize, cntNotifyDescSize);
825 0 : return HCCL_E_PARA;
826 : }
827 0 : rmtCntNotifyDesc.clear();
828 0 : u32 pos = 0;
829 0 : for (pos = 0; pos < descSize; pos++) {
830 : char c;
831 0 : binaryStream >> c;
832 0 : rmtCntNotifyDesc.push_back(c);
833 : }
834 0 : HCCL_INFO("unpack cntNotify Desc=%s", Bytes2hex(rmtCntNotifyDesc.data(), rmtCntNotifyDesc.size()).c_str());
835 0 : return HCCL_SUCCESS;
836 : }
837 :
838 3 : HcclResult UbMemTransport::RmtBufferVecUnpackProc(
839 : u32 locNum, BinaryStream& binaryStream, RemoteBufferVec& bufferVec, UbRmtBufType type)
840 : {
841 : u32 rmtNum;
842 3 : binaryStream >> rmtNum;
843 3 : if (UNLIKELY(type == UbRmtBufType::BUFFER && rmtNum > MAX_BUFFER_NUM)) {
844 0 : HCCL_ERROR("[UbMemTransport][RmtBufferVecUnpackProc] rmtNum[%u] exceeds limit[%u]", rmtNum, MAX_BUFFER_NUM);
845 0 : return HCCL_E_PARA;
846 : }
847 :
848 : // 允许本端和远端交换内存数量不一致
849 9 : HCCL_INFO("unpack %s %s, locNum=%u, rmtNum=%u", type.Describe().c_str(), GetLinkDescInfo().c_str(), locNum, rmtNum);
850 :
851 7 : for (u32 i = 0; i < rmtNum; i++) {
852 : u32 pos;
853 4 : binaryStream >> pos;
854 4 : ExchangeUbBufferDto dto;
855 4 : dto.Deserialize(binaryStream);
856 4 : if (bufferVec.size() > pos) {
857 : // 对于之前已经加过的资源，无需追加
858 0 : continue;
859 : }
860 :
861 12 : HCCL_INFO("unpack %s pos=%u, dto %s", type.Describe().c_str(), pos, dto.Describe().c_str());
862 4 : if (dto.size == 0) { // size为0，则为 remote 空buffer
863 0 : HCCL_INFO("unpack nullptr, pos=%u", pos);
864 0 : bufferVec.push_back(nullptr);
865 0 : FillRmtRmaBufferVec(nullptr, type);
866 : } else { // size非0，则构造一个remote buffer
867 4 : bufferVec.push_back(make_unique<RemoteUbRmaBuffer>(rdmaHandle, dto));
868 4 : FillRmtRmaBufferVec(bufferVec.back().get(), type);
869 12 : HCCL_INFO("unpack buffer pos=%u, rmtRmaBuffer=%s", pos, bufferVec.back()->Describe().c_str());
870 : }
871 4 : }
872 :
873 3 : return HCCL_SUCCESS;
874 : }
875 :
876 1 : HcclResult UbMemTransport::ConnVecUnpackProc(BinaryStream& binaryStream, bool& needSendFinish)
877 : {
878 : u32 rmtConnNum;
879 1 : binaryStream >> rmtConnNum;
880 3 : HCCL_INFO("start unpack conn %s connNum=%u, rmtConnNum=%u", GetLinkDescInfo().c_str(), connNum, rmtConnNum);
881 1 : if (connNum != rmtConnNum) {
882 0 : HCCL_ERROR("[UbMemTransport::ConnVecUnpackProc] connNum=%u is not equal to rmtConnNum=%u", connNum, rmtConnNum);
883 0 : return HCCL_E_PARA;
884 : }
885 :
886 1 : needSendFinish = false; // 不需要发送 finish
887 2 : for (u32 i = 0; i < rmtConnNum; i++) {
888 : u32 pos;
889 1 : binaryStream >> pos;
890 1 : ExchangeUbConnDto rmtDto;
891 1 : rmtDto.Deserialize(binaryStream);
892 3 : HCCL_INFO("unpack connection pos=%u dto %s", pos, rmtDto.Describe().c_str());
893 1 : if (commonLocRes.connVec[i]->GetStatus() != RmaConnStatus::READY) {
894 0 : HCCL_INFO(
895 : "parse and import pos=%u, rmt dto to connection[%s]", pos, commonLocRes.connVec[i]->Describe().c_str());
896 0 : commonLocRes.connVec[i]->ParseRmtExchangeDto(rmtDto);
897 0 : commonLocRes.connVec[i]->ImportRmtDto();
898 0 : needSendFinish = true; // connection 建链，需要发送finish
899 : }
900 1 : }
901 1 : return HCCL_SUCCESS;
902 : }
903 :
904 4 : void UbMemTransport::FillRmtRmaBufferVec(RemoteRmaBuffer* rmaBuffer, UbRmtBufType type)
905 : {
906 4 : if (type == UbRmtBufType::BUFFER) {
907 4 : rmtRmaBufferVec.push_back(rmaBuffer);
908 : }
909 4 : }
910 :
911 2 : HcclResult UbMemTransport::SendFinish()
912 : {
913 6 : HCCL_INFO("start send Finish Msg %s [%s], isHost_[%d]", GetLinkDescInfo().c_str(), FINISH_MSG, isHost_);
914 2 : sendFinishMsg = std::vector<char>(FINISH_MSG, FINISH_MSG + FINISH_MSG_SIZE);
915 2 : bool ret = false;
916 2 : if (isHost_) {
917 0 : ret = socket->Send(sendFinishMsg.data(), FINISH_MSG_SIZE);
918 : } else {
919 2 : socket->SendAsync(sendFinishMsg.data(), FINISH_MSG_SIZE);
920 1 : ret = true;
921 : }
922 1 : if (!ret) {
923 0 : HCCL_ERROR("[UbMemTransport::SendFinish] Send finish msg failed");
924 0 : return HCCL_E_INTERNAL;
925 : }
926 3 : HCCL_INFO("end send Finish Msg %s [%s]", GetLinkDescInfo().c_str(), FINISH_MSG);
927 1 : return HCCL_SUCCESS;
928 : }
929 :
930 1 : HcclResult UbMemTransport::RecvFinish()
931 : {
932 1 : recvFinishMsg.resize(FINISH_MSG_SIZE);
933 3 : HCCL_INFO("start recv Finish Msg %s [%s], isHost_[%d]", GetLinkDescInfo().c_str(), FINISH_MSG, isHost_);
934 1 : bool ret = false;
935 1 : if (isHost_) {
936 0 : ret = socket->Recv(recvFinishMsg.data(), FINISH_MSG_SIZE);
937 : } else {
938 1 : socket->RecvAsync(reinterpret_cast<u8*>(recvFinishMsg.data()), FINISH_MSG_SIZE);
939 1 : ret = true;
940 : }
941 1 : if (!ret) {
942 0 : HCCL_ERROR("[UbMemTransport::RecvFinish] Recv finish msg failed");
943 0 : return HCCL_E_INTERNAL;
944 : }
945 3 : HCCL_INFO("end recv Finish Msg %s [%s]", GetLinkDescInfo().c_str(), FINISH_MSG);
946 1 : return HCCL_SUCCESS;
947 : }
948 :
949 49 : std::vector<char> UbMemTransport::GetUniqueId()
950 : {
951 49 : if (baseStatus != TransportStatus::READY) {
952 4 : MACRO_THROW(InternalException, StringFormat("transport status is not ready, please check"));
953 : }
954 48 : u32 type = static_cast<u32>(transportType);
955 48 : BinaryStream binaryStream;
956 48 : binaryStream << type;
957 48 : binaryStream << notifyNum;
958 48 : binaryStream << bufferNum;
959 48 : binaryStream << static_cast<u32>(rmtBufferVec.size());
960 48 : binaryStream << connNum;
961 :
962 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
963 48 : auto notifyUniqueIds = GetNotifyUniqueIds();
964 48 : binaryStream << notifyUniqueIds;
965 :
966 48 : auto rmtNotifyUniqueIds = GetRmtBufferUniqueIds(rmtNotifyVec, UbRmtBufType::NOTIFY);
967 48 : binaryStream << rmtNotifyUniqueIds;
968 :
969 48 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds(rmtBufferVec, UbRmtBufType::BUFFER);
970 48 : binaryStream << rmtBufferUniqueIds;
971 :
972 48 : auto connUniqueIds = GetConnUniqueIds();
973 48 : binaryStream << connUniqueIds;
974 :
975 48 : std::vector<char> result;
976 48 : binaryStream.Dump(result);
977 48 : return result;
978 48 : }
979 :
980 0 : std::vector<char> UbMemTransport::GetUniqueIdV2()
981 : {
982 0 : if (baseStatus != TransportStatus::READY) {
983 0 : MACRO_THROW(
984 : InternalException,
985 : StringFormat("transport status[%d] is not ready[%d], please check.", baseStatus, TransportStatus::READY));
986 : }
987 0 : u32 type = static_cast<u32>(transportType);
988 0 : BinaryStream binaryStream;
989 0 : binaryStream << type;
990 0 : binaryStream << notifyNum;
991 0 : binaryStream << bufferNum;
992 0 : binaryStream << static_cast<u32>(rmtBufferVec.size());
993 0 : binaryStream << connNum;
994 :
995 0 : auto notifyUniqueIds = GetNotifyUniqueIds();
996 0 : binaryStream << notifyUniqueIds;
997 :
998 0 : auto rmtNotifyUniqueIds = GetRmtBufferUniqueIds(rmtNotifyVec, UbRmtBufType::NOTIFY);
999 0 : binaryStream << rmtNotifyUniqueIds;
1000 :
1001 0 : for (auto& it : commonLocRes.bufferVec) {
1002 0 : locBufferVec.emplace_back(reinterpret_cast<LocalUbRmaBuffer*>(it));
1003 : }
1004 :
1005 0 : auto locBufferUniqueIds = GetLocBufferUniqueIds(locBufferVec, UbRmtBufType::BUFFER);
1006 0 : binaryStream << locBufferUniqueIds;
1007 :
1008 0 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds(rmtBufferVec, UbRmtBufType::BUFFER);
1009 0 : binaryStream << rmtBufferUniqueIds;
1010 :
1011 0 : auto drainUniqueIds = GetDrainUniqueIds();
1012 0 : binaryStream << drainUniqueIds;
1013 :
1014 0 : auto connUniqueIds = GetConnUniqueIds();
1015 0 : binaryStream << connUniqueIds;
1016 :
1017 0 : std::vector<char> result;
1018 0 : binaryStream.Dump(result);
1019 0 : return result;
1020 0 : }
1021 :
1022 0 : std::vector<char> UbMemTransport::PackConnData()
1023 : {
1024 0 : if (baseStatus != TransportStatus::READY) {
1025 0 : MACRO_THROW(
1026 : InternalException,
1027 : StringFormat("transport status[%d] is not ready[%d], please check.", baseStatus, TransportStatus::READY));
1028 : }
1029 0 : u32 type = static_cast<u32>(transportType);
1030 0 : BinaryStream binaryStream;
1031 0 : binaryStream << type;
1032 0 : binaryStream << connNum;
1033 :
1034 0 : auto connUniqueIds = GetConnUniqueIds();
1035 0 : binaryStream << connUniqueIds;
1036 :
1037 0 : std::vector<char> result;
1038 0 : binaryStream.Dump(result);
1039 0 : return result;
1040 0 : }
1041 :
1042 : std::vector<char>
1043 0 : UbMemTransport::GetSingleRmtBufferUniqueId(u64 addr, u64 size, u32 tokenId, u32 tokenValue, u32 notifyId) const
1044 : {
1045 0 : BinaryStream binaryStream;
1046 0 : binaryStream << addr;
1047 0 : binaryStream << size;
1048 0 : binaryStream << tokenId;
1049 0 : binaryStream << tokenValue;
1050 0 : binaryStream << notifyId;
1051 0 : HCCL_INFO("UbMemTransport RmtBuffer[addr=0x%llx, size=0x%llx, notifyId=%u]", addr, size, notifyId);
1052 0 : std::vector<char> result;
1053 0 : binaryStream.Dump(result);
1054 0 : return result;
1055 0 : }
1056 :
1057 48 : std::vector<char> UbMemTransport::GetNotifyUniqueIds()
1058 : {
1059 144 : HCCL_INFO("start packing all notify uniqueIds");
1060 48 : std::vector<char> result(0);
1061 94 : for (auto& it : commonLocRes.notifyVec) {
1062 138 : HCCL_INFO("ubMemTransport Notify %s", it->Describe().c_str());
1063 46 : auto uniqueId = it->GetUniqueId();
1064 46 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1065 46 : }
1066 48 : return result;
1067 0 : }
1068 :
1069 0 : std::vector<char> UbMemTransport::GetDrainUniqueIds() const
1070 : {
1071 0 : HCCL_INFO("start packing drain resources uniqueIds");
1072 0 : std::vector<char> result(0);
1073 0 : std::vector<char> uniqueId;
1074 :
1075 : // pack drain notify
1076 0 : if (drainNotify_ != nullptr) {
1077 0 : auto dto = drainNotify_->GetExchangeDto();
1078 0 : ExchangeUbBufferDto* rawDto = static_cast<ExchangeUbBufferDto*>(dto.get());
1079 0 : uniqueId = GetSingleRmtBufferUniqueId(
1080 0 : rawDto->addr, rawDto->size, rawDto->tokenId, rawDto->tokenValue, rawDto->notifyId);
1081 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, %s", drainNotify_->Describe().c_str());
1082 0 : } else {
1083 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1084 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, drainNotify_ null buffer");
1085 : }
1086 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1087 :
1088 : // pack drain rmtMem
1089 0 : if (rmtDrainBuffer_ != nullptr) {
1090 0 : uniqueId = GetSingleRmtBufferUniqueId(
1091 0 : rmtDrainBuffer_->GetAddr(), rmtDrainBuffer_->GetSize(), rmtDrainBuffer_->GetTokenId(),
1092 0 : rmtDrainBuffer_->GetTokenValue(), rmtDrainBuffer_->GetNotifyId());
1093 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, %s", rmtDrainBuffer_->Describe().c_str());
1094 : } else {
1095 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1096 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, rmtDrainBuffer_ null buffer");
1097 : }
1098 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1099 :
1100 0 : return result;
1101 0 : }
1102 :
1103 96 : std::vector<char> UbMemTransport::GetRmtBufferUniqueIds(RemoteBufferVec& bufferVec, UbRmtBufType type) const
1104 : {
1105 288 : HCCL_INFO("start packing all remote buffer %s uniqueIds", type.Describe().c_str());
1106 96 : std::vector<char> result(0);
1107 96 : for (auto& it : bufferVec) {
1108 0 : std::vector<char> uniqueId;
1109 0 : if (it != nullptr) {
1110 0 : uniqueId = GetSingleRmtBufferUniqueId(
1111 0 : it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue(), it->GetNotifyId());
1112 0 : HCCL_INFO("UbMemTransport::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
1113 : } else {
1114 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1115 0 : HCCL_INFO("UbMemTransport::GetRmtBufferUniqueIds, null buffer");
1116 : }
1117 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1118 0 : }
1119 96 : return result;
1120 0 : }
1121 :
1122 0 : std::vector<char> UbMemTransport::GetLocBufferUniqueIds(LocalBufferVec& bufferVec, UbRmtBufType type) const
1123 : {
1124 0 : HCCL_INFO("start packing all local buffer %s uniqueIds", type.Describe().c_str());
1125 0 : std::vector<char> result(0);
1126 0 : for (auto& it : bufferVec) {
1127 0 : std::vector<char> uniqueId;
1128 0 : if (it != nullptr) {
1129 0 : uniqueId = GetSingleRmtBufferUniqueId(
1130 0 : it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue(), UINT32_MAX);
1131 0 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, %s", it->Describe().c_str());
1132 : } else {
1133 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1134 0 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, null buffer");
1135 : }
1136 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1137 0 : }
1138 0 : return result;
1139 0 : }
1140 :
1141 48 : std::vector<char> UbMemTransport::GetConnUniqueIds()
1142 : {
1143 144 : HCCL_INFO("start packing all conn uniqueIds");
1144 48 : std::vector<char> result(0);
1145 94 : for (auto& it : commonLocRes.connVec) {
1146 138 : HCCL_INFO("[UbMemTransport::%s] conn[%s]", __func__, it->Describe().c_str());
1147 46 : auto uniqueId = it->GetUniqueId();
1148 46 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1149 46 : }
1150 48 : return result;
1151 0 : }
1152 :
1153 3 : void UbMemTransport::SaveDfxTaskInfo(const TaskParam& taskParam)
1154 : {
1155 : u32 taskId;
1156 : u32 streamId;
1157 3 : HrtGetTaskIdAndStreamID(taskId, streamId);
1158 :
1159 3 : callback(streamId, taskId, taskParam);
1160 3 : }
1161 :
1162 2 : HcclResult UbMemTransport::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
1163 : {
1164 2 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
1165 : Hccl::RemoteMemCtx<std::unique_ptr<RemoteUbRmaBuffer>> remoteMemCtx{
1166 2 : cacheValid_, rmtBufferVec, remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
1167 2 : CHK_RET(GetRemoteUserMems(remoteMemCtx));
1168 2 : return HCCL_SUCCESS;
1169 2 : }
1170 :
1171 5 : HcclResult UbMemTransport::CheckSocketStatus(std::string socketOpreator)
1172 : {
1173 5 : CHK_PTR_NULL(socket);
1174 5 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
1175 5 : auto startTime = std::chrono::steady_clock::now();
1176 5 : uint32_t retryCount = 0;
1177 : while (true) {
1178 5 : SocketStatus socketStatus = socket->GetAsyncStatus();
1179 5 : if (socketStatus == SocketStatus::OK) {
1180 : auto elapsed
1181 4 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1182 4 : .count();
1183 12 : HCCL_INFO(
1184 : "[UbMemTransport][%s] socket transport operation[%s] success, elapsed[%lld]ms, retryCount[%u]",
1185 : __func__, socketOpreator.c_str(), elapsed, retryCount);
1186 4 : break;
1187 : }
1188 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout || socketStatus == Hccl::SocketStatus::TIMEOUT) {
1189 : auto elapsed
1190 1 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1191 1 : .count();
1192 3 : HCCL_ERROR(
1193 : "[UbMemTransport][%s] socket transport operation[%s] timeout after %lld sec, elapsed[%lld]ms, "
1194 : "retryCount[%u]",
1195 : __func__, socketOpreator.c_str(), timeout, elapsed, retryCount);
1196 1 : return HCCL_E_TIMEOUT;
1197 : }
1198 0 : retryCount++;
1199 0 : }
1200 4 : return HCCL_SUCCESS;
1201 : }
1202 :
1203 3 : HcclResult UbMemTransport::UpdateMemInfo(std::vector<LocalRmaBuffer*>& bufferVecTemp)
1204 : {
1205 3 : if (bufferVecTemp.size() == 0) {
1206 3 : HCCL_WARNING("[UbMemTransport][UpdateMemInfo] bufferNum is 0.");
1207 1 : return HCCL_SUCCESS;
1208 : }
1209 6 : HCCL_INFO("[UbMemTransport][UpdateMemInfo] bufferNum[%zu]", bufferVecTemp.size());
1210 2 : sendData.clear();
1211 2 : BinaryStream sendStream;
1212 2 : std::vector<std::unique_ptr<RemoteUbRmaBuffer>> rmtBufferTemp{};
1213 84 : TRY_CATCH_RETURN([&]() -> void {
1214 : BufferVecPack(sendStream, bufferVecTemp);
1215 : sendStream.Dump(sendData);
1216 : u32 sendSize = sendData.size();
1217 : socket->SendAsync(&sendSize, sizeof(sendSize));
1218 : HCCL_INFO(
1219 : "[UbMemTransport][UpdateMemInfo] Send size[%u] of data success. [%zu] bytes sent.", sendSize,
1220 : sizeof(sendSize));
1221 : HcclResult result = CheckSocketStatus("SendDataSize");
1222 : CHK_RET_THROW(
1223 : InternalException, StringFormat("[UbMemTransport][UpdateMemInfo] failed to send dataSize."), result);
1224 : RecvDataSize();
1225 : result = CheckSocketStatus("RecvDataSize");
1226 : CHK_RET_THROW(
1227 : InternalException, StringFormat("[UbMemTransport][UpdateMemInfo] failed to receive dataSize."), result);
1228 : SendExchangeData();
1229 : result = CheckSocketStatus("SendExchangeData");
1230 : CHK_RET_THROW(InternalException, StringFormat("[UbMemTransport][UpdateMemInfo] failed to send data."), result);
1231 : RecvExchangeData();
1232 : result = CheckSocketStatus("RecvExchangeData");
1233 : CHK_RET_THROW(
1234 : InternalException, StringFormat("[UbMemTransport][UpdateMemInfo] failed to receive data."), result);
1235 : BinaryStream recvStream(recvData);
1236 : RmtBufferVecUnpackProc(bufferNum, recvStream, rmtBufferTemp, UbRmtBufType::BUFFER);
1237 : }());
1238 2 : rmtBufferVec.insert(
1239 1 : rmtBufferVec.end(), std::make_move_iterator(rmtBufferTemp.begin()),
1240 : std::make_move_iterator(rmtBufferTemp.end()));
1241 1 : commonLocRes.bufferVec.insert(commonLocRes.bufferVec.end(), bufferVecTemp.begin(), bufferVecTemp.end());
1242 1 : cacheValid_ = false;
1243 1 : return HCCL_SUCCESS;
1244 2 : }
1245 :
1246 0 : HcclResult UbMemTransport::Init()
1247 : {
1248 0 : for (auto& ubConn : commonLocRes.connVec) {
1249 0 : TRY_CATCH_RETURN(ubConn->Connect());
1250 : }
1251 :
1252 0 : return HCCL_SUCCESS;
1253 : }
1254 :
1255 0 : HcclResult UbMemTransport::DeInit() const
1256 : {
1257 0 : socket->Destroy();
1258 0 : return HCCL_SUCCESS;
1259 : }
1260 :
1261 0 : HcclResult UbMemTransport::GetRemoteSeg(const void* addr, u64 len, u64* seg)
1262 : {
1263 0 : if (rmtBufferVec.empty()) {
1264 0 : HCCL_ERROR("[UbMemTransport::%s] rmtBufferVec is empty.", __func__);
1265 0 : return HCCL_E_INTERNAL;
1266 : }
1267 :
1268 0 : bool isAddrInRange = false;
1269 0 : for (auto& it : rmtBufferVec) {
1270 0 : Buffer iterBuf(it->GetAddr(), it->GetSize());
1271 0 : if (iterBuf.Contains(reinterpret_cast<uintptr_t>(addr), len)) {
1272 0 : *seg = it->GetSegVa();
1273 0 : isAddrInRange = true;
1274 0 : break;
1275 : }
1276 0 : }
1277 :
1278 0 : if (!isAddrInRange) {
1279 0 : return HCCL_E_INTERNAL;
1280 : }
1281 0 : return HCCL_SUCCESS;
1282 : }
1283 :
1284 : } // namespace Hccl
|
