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 "mem_transport_manager.h"
12 : #include "rdma_handle_manager.h"
13 : #include "communicator_impl.h"
14 : #include "ub_mem_transport.h"
15 : #include "urma_direct_transport.h"
16 : #include "p2p_transport.h"
17 : #include "cnt_notify_res_helper.h"
18 : #include "timeout_exception.h"
19 : #include "local_rma_buf_manager.h"
20 : namespace Hccl {
21 479 : MemTransportManager::MemTransportManager(const CommunicatorImpl& communicator) : comm(&communicator) {}
22 :
23 943 : MemTransportManager::~MemTransportManager() {}
24 :
25 5 : std::vector<BaseLocalNotify*> MemTransportManager::GetNotifyVec(const LinkData& linkData) const
26 : {
27 5 : return comm->GetConnLocalNotifyManager().Get(linkData.GetRemoteRankId(), linkData);
28 : }
29 :
30 : const std::vector<BufferType> PIPE_BUFFER_TYPE = {BufferType::INPUT, BufferType::OUTPUT, BufferType::SCRATCH};
31 :
32 : std::vector<LocalRmaBuffer*>
33 5 : MemTransportManager::GetBufferVec(const std::string& opTag, const LinkData& linkData, OpMode opMode) const
34 : {
35 15 : HCCL_DEBUG("[MemTransportManager][%s] opMode[%s]", __func__, opMode.Describe().c_str());
36 5 : std::vector<LocalRmaBuffer*> result;
37 5 : if (opMode == OpMode::OPBASE) {
38 2 : result.push_back(nullptr); // 单算子 input/output 为null,
39 2 : result.push_back(nullptr);
40 2 : auto res = comm->GetLocalRmaBufManager().Get(opTag, linkData.GetLocalPort(), BufferType::SCRATCH);
41 2 : result.push_back(res);
42 : } else {
43 12 : for (auto& bufType : PIPE_BUFFER_TYPE) {
44 9 : auto res = comm->GetLocalRmaBufManager().Get(opTag, linkData.GetLocalPort(), bufType);
45 9 : result.push_back(res); // INPUT/OUTPUT/SCRATCH 都交换
46 : }
47 : }
48 5 : return result;
49 0 : }
50 :
51 6 : std::vector<RmaConnection*> MemTransportManager::GetConnVec(const std::string& opTag, const LinkData& linkData) const
52 : {
53 6 : std::vector<RmaConnection*> result;
54 6 : result.push_back(comm->GetRmaConnManager().Get(opTag, linkData));
55 6 : return result;
56 0 : }
57 :
58 2 : void MemTransportManager::CreateOpbasedUbMemTransport(
59 : BaseMemTransport::CommonLocRes& locRes, BaseMemTransport::Attribution& attr, const LinkData& linkData,
60 : const Socket& socket)
61 : {
62 2 : auto topicIdCntNotifyVecMap = comm->GetConnLocalCntNotifyManager().GetTopicIdCntNotifyMap(linkData.GetLocalPort());
63 : CntNotifyResHelper tool;
64 2 : BaseMemTransport::LocCntNotifyRes locCntNotifyRes = tool.GetCntNotifyRes(topicIdCntNotifyVecMap);
65 6 : HCCL_INFO("locCntNotifyRes=%s, linkData=%s", locCntNotifyRes.Describe().c_str(), linkData.Describe().c_str());
66 4 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(
67 2 : comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
68 :
69 : // DFX：注册transportCallBack, 用于信息保存
70 2 : auto transportCallBack = MemTransportCallback(linkData, comm->GetMirrorTaskManager());
71 : auto ubMemTransport
72 2 : = make_unique<UbMemTransport>(locRes, attr, linkData, socket, rdmaHandle, locCntNotifyRes, transportCallBack);
73 2 : opTagOpbasedMap[linkData] = std::move(ubMemTransport);
74 2 : }
75 :
76 2 : void MemTransportManager::CreateOffloadUbMemTransport(
77 : const string& opTag, BaseMemTransport::CommonLocRes& locRes, BaseMemTransport::Attribution& attr,
78 : const LinkData& linkData, const Socket& socket)
79 : {
80 2 : auto topicIdCntNotifyVecMap = comm->GetConnLocalCntNotifyManager().GetTopicIdCntNotifyMap(linkData.GetLocalPort());
81 : CntNotifyResHelper tool;
82 2 : BaseMemTransport::LocCntNotifyRes locCntNotifyRes = tool.GetCntNotifyRes(topicIdCntNotifyVecMap);
83 6 : HCCL_INFO("locCntNotifyRes=%s, linkData=%s", locCntNotifyRes.Describe().c_str(), linkData.Describe().c_str());
84 4 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(
85 2 : comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
86 :
87 : // DFX：注册transportCallBack, 用于信息保存
88 2 : auto transportCallBack = MemTransportCallback(linkData, comm->GetMirrorTaskManager());
89 : auto ubMemTransport
90 2 : = make_unique<UbMemTransport>(locRes, attr, linkData, socket, rdmaHandle, locCntNotifyRes, transportCallBack);
91 2 : opTagOffloadMap[opTag][linkData] = std::move(ubMemTransport);
92 2 : }
93 :
94 1 : BaseMemTransport* MemTransportManager::CreateOpbasedMemTransport(const LinkData& linkData)
95 : {
96 1 : auto op = comm->GetCurrentCollOperator();
97 3 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
98 3 : HCCL_INFO("Entry CreateMemTransport, opInfo=[%s]", CollOpToString(*op).c_str());
99 1 : BaseMemTransport::CommonLocRes locRes;
100 1 : locRes.notifyVec = GetNotifyVec(linkData);
101 3 : HCCL_INFO("link=%s get notifyVec OK", linkData.Describe().c_str());
102 :
103 : // buffer 来自localRmaBufferManager, input/output/scratch
104 1 : locRes.bufferVec = GetBufferVec(comm->GetId(), linkData, OpMode::OPBASE);
105 3 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
106 :
107 : // connection是一个，来自 RmaConnManager
108 1 : locRes.connVec = GetConnVec(comm->GetId(), linkData);
109 3 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
110 :
111 3 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
112 :
113 1 : BaseMemTransport::Attribution attr;
114 1 : attr.devicePhyId = linkData.GetLocalPort().GetId();
115 : // 握手消息定义，未来包括 cann版本号，rankTable CRC等字段
116 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
117 3 : HCCL_INFO("[MemTransportManager::%s] accelerator[%s]", __func__, accelerator.Describe().c_str());
118 1 : attr.opAcceState = accelerator;
119 1 : attr.handshakeMsg = op->GetUniqueId();
120 :
121 1 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
122 1 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
123 1 : if (socket == nullptr) {
124 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
125 : }
126 1 : if (linkData.GetType() == PortDeploymentType::P2P) {
127 0 : opTagOpbasedMap[linkData] = make_unique<P2PTransport>(locRes, attr, linkData, *socket);
128 1 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
129 1 : auto linkProtocol = linkData.GetLinkProtocol();
130 1 : if (linkProtocol == LinkProtocol::UB_CTP || linkProtocol == LinkProtocol::UB_TP
131 1 : || linkProtocol == LinkProtocol::UBOE || linkProtocol == LinkProtocol::UB_RTP) {
132 1 : CreateOpbasedUbMemTransport(locRes, attr, linkData, *socket);
133 : } else {
134 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
135 : }
136 : } else {
137 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
138 : }
139 :
140 1 : opTagOpbasedMap[linkData]->Establish();
141 :
142 1 : newOpbasedTransports[linkData] = 0;
143 :
144 3 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
145 3 : HCCL_INFO("create transport %s OK.", opTagOpbasedMap[linkData]->Describe().c_str());
146 :
147 2 : return opTagOpbasedMap[linkData].get();
148 1 : }
149 1 : BaseMemTransport* MemTransportManager::CreateOffloadMemTransport(const std::string& opTag, const LinkData& linkData)
150 : {
151 1 : auto op = comm->GetCurrentCollOperator();
152 3 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
153 3 : HCCL_INFO("Entry CreateMemTransport, opInfo=[%s]", CollOpToString(*op).c_str());
154 1 : BaseMemTransport::CommonLocRes locRes;
155 1 : locRes.notifyVec = GetNotifyVec(linkData);
156 3 : HCCL_INFO("link=%s get notifyVec OK", linkData.Describe().c_str());
157 :
158 : // buffer 来自localRmaBufferManager, input/output/scratch
159 1 : locRes.bufferVec = GetBufferVec(opTag, linkData, OpMode::OFFLOAD);
160 3 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
161 :
162 : // connection是一个，来自 RmaConnManager
163 1 : std::string tag = comm->GetOpAiCpuTSFeatureFlag() == true ? comm->GetId() : opTag; // 算子粒度
164 1 : locRes.connVec = GetConnVec(tag, linkData);
165 3 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
166 :
167 3 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
168 :
169 1 : BaseMemTransport::Attribution attr;
170 1 : attr.devicePhyId = linkData.GetLocalPort().GetId();
171 : // 握手消息定义，未来包括 cann版本号，rankTable CRC等字段
172 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
173 3 : HCCL_INFO("[MemTransportManager::%s] accelerator[%s]", __func__, accelerator.Describe().c_str());
174 1 : attr.opAcceState = accelerator;
175 1 : attr.handshakeMsg = op->GetUniqueId();
176 :
177 1 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
178 1 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
179 1 : if (socket == nullptr) {
180 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
181 : }
182 :
183 1 : if (linkData.GetType() == PortDeploymentType::P2P) {
184 0 : opTagOffloadMap[opTag][linkData] = make_unique<P2PTransport>(locRes, attr, linkData, *socket);
185 1 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
186 1 : auto linkProtocol = linkData.GetLinkProtocol();
187 1 : if (linkProtocol == LinkProtocol::UB_CTP || linkProtocol == LinkProtocol::UB_TP
188 1 : || linkProtocol == LinkProtocol::UBOE || linkProtocol == LinkProtocol::UB_RTP) {
189 1 : CreateOffloadUbMemTransport(opTag, locRes, attr, linkData, *socket);
190 : } else {
191 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
192 : }
193 : } else {
194 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
195 : }
196 :
197 1 : opTagOffloadMap[opTag][linkData]->Establish();
198 :
199 1 : newOffloadTransports[opTag][linkData] = 0;
200 :
201 3 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
202 3 : HCCL_INFO("create transport %s OK.", opTagOffloadMap[opTag][linkData]->Describe().c_str());
203 :
204 2 : return opTagOffloadMap[opTag][linkData].get();
205 1 : }
206 :
207 1 : void MemTransportManager::DumpNotReadyTransportsOpbased()
208 : {
209 3 : HCCL_ERROR("Dump opbased timeout transport info, transport size[%u]", newOpbasedTransports.size());
210 2 : for (auto linkIt = newOpbasedTransports.begin(); linkIt != newOpbasedTransports.end(); ++linkIt) {
211 1 : auto transportPtr = opTagOpbasedMap[linkIt->first].get();
212 3 : HCCL_ERROR("Transport info[%s]", transportPtr->Describe().c_str());
213 3 : HCCL_ERROR("Linkdata info[%s]", transportPtr->GetLinkDescInfo().c_str());
214 3 : HCCL_ERROR("Socket info[%s]", transportPtr->DescribeSocket().c_str());
215 : }
216 1 : }
217 :
218 1 : void MemTransportManager::DumpNotReadyTransportsOffload(const std::string& opTag)
219 : {
220 3 : HCCL_ERROR("Dump offload timeout transport info, transport size[%u]", newOffloadTransports[opTag].size());
221 2 : for (auto linkIt = newOffloadTransports[opTag].begin(); linkIt != newOffloadTransports[opTag].end(); ++linkIt) {
222 1 : auto transportPtr = opTagOffloadMap[opTag][linkIt->first].get();
223 3 : HCCL_ERROR("Transport info[%s]", transportPtr->Describe().c_str());
224 3 : HCCL_ERROR("Linkdata info[%s]", transportPtr->GetLinkDescInfo().c_str());
225 3 : HCCL_ERROR("Socket info[%s]", transportPtr->DescribeSocket().c_str());
226 : }
227 1 : }
228 :
229 0 : void MemTransportManager::DumpNotReadyTransportsUrma()
230 : {
231 0 : HCCL_RUN_INFO("[MemTransportManager][%s] start", __func__);
232 0 : for (auto& it : urmaDirectMap_) {
233 0 : auto status = it.second->GetStatus();
234 0 : if (status != TransportStatus::READY) {
235 0 : HCCL_INFO("linkData[%s] status[%s]", it.first.Describe().c_str(), status.Describe().c_str());
236 : }
237 : }
238 0 : }
239 :
240 11 : bool MemTransportManager::IsAllOpbasedTransportReady()
241 : {
242 11 : bool result = true;
243 : // 当前只针对新增的transports做资源交换和op校验
244 12 : for (auto linkIt = newOpbasedTransports.begin(); linkIt != newOpbasedTransports.end();) {
245 1 : auto status = opTagOpbasedMap[linkIt->first]->GetStatus();
246 1 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
247 1 : if (status == TransportStatus::SOCKET_TIMEOUT) {
248 0 : MACRO_THROW(
249 : TimeoutException,
250 : StringFormat(
251 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check", __func__,
252 : opTagOpbasedMap[linkIt->first]->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
253 : }
254 1 : result = false;
255 1 : ++linkIt;
256 : } else {
257 0 : HCCL_INFO("linkData[%s], status[%s].", linkIt->first.Describe().c_str(), status.Describe().c_str());
258 0 : linkIt = newOpbasedTransports.erase(linkIt);
259 : }
260 : }
261 11 : return result;
262 : }
263 :
264 1 : bool MemTransportManager::IsAllOneSidedTransportReady()
265 : {
266 1 : bool result = true;
267 : // 当前只针对新增的transports做资源交换和op校验
268 2 : for (auto linkIt = newOneSidedTransports.begin(); linkIt != newOneSidedTransports.end();) {
269 1 : auto status = oneSidedMap[linkIt->first]->GetStatus();
270 1 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
271 1 : result = false;
272 1 : ++linkIt;
273 : } else {
274 0 : HCCL_INFO("linkData[%s] status[%s]", linkIt->first.Describe().c_str(), status.Describe().c_str());
275 0 : linkIt = newOneSidedTransports.erase(linkIt);
276 : }
277 : }
278 1 : return result;
279 : }
280 :
281 9 : bool MemTransportManager::IsAllOffloadTransportReady(const std::string& opTag)
282 : {
283 9 : bool result = true;
284 : // 当前只针对新增的transports做资源交换和op校验
285 10 : for (auto linkIt = newOffloadTransports[opTag].begin(); linkIt != newOffloadTransports[opTag].end();) {
286 1 : auto status = opTagOffloadMap[opTag][linkIt->first]->GetStatus();
287 1 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
288 1 : if (status == TransportStatus::SOCKET_TIMEOUT) {
289 0 : MACRO_THROW(
290 : TimeoutException,
291 : StringFormat(
292 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check", __func__,
293 : opTagOffloadMap[opTag][linkIt->first]->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
294 : }
295 1 : result = false;
296 1 : ++linkIt;
297 : } else {
298 0 : HCCL_INFO(
299 : "opTag[%s] linkData[%s] status[%s]", opTag.c_str(), linkIt->first.Describe().c_str(),
300 : status.Describe().c_str());
301 0 : linkIt = newOffloadTransports[opTag].erase(linkIt);
302 : }
303 : }
304 9 : return result;
305 : }
306 :
307 3 : bool MemTransportManager::IsAllTransportReady()
308 : {
309 3 : bool result = true;
310 6 : for (auto& tagIt : opTagOffloadMap) {
311 6 : for (auto& it : tagIt.second) {
312 3 : auto status = it.second->GetStatus();
313 3 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
314 2 : if (status == TransportStatus::SOCKET_TIMEOUT) {
315 0 : MACRO_THROW(
316 : TimeoutException, StringFormat(
317 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check",
318 : __func__, it.second->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
319 : }
320 2 : result = false;
321 : }
322 : }
323 : }
324 3 : for (auto& it : opTagOpbasedMap) {
325 0 : auto status = it.second->GetStatus();
326 0 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
327 0 : if (status == TransportStatus::SOCKET_TIMEOUT) {
328 0 : MACRO_THROW(
329 : TimeoutException, StringFormat(
330 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check",
331 : __func__, it.second->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
332 : }
333 0 : result = false;
334 : }
335 : }
336 3 : for (auto& it : urmaDirectMap_) {
337 0 : auto status = it.second->GetStatus();
338 0 : if (status != TransportStatus::READY) { // 任意一个没有ready，结果为 false
339 0 : if (status == TransportStatus::SOCKET_TIMEOUT) {
340 0 : MACRO_THROW(
341 : TimeoutException, StringFormat(
342 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check",
343 : __func__, it.second->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
344 : }
345 0 : result = false;
346 : }
347 : }
348 3 : return result;
349 : }
350 :
351 6 : void MemTransportManager::BatchBuildOpbasedTransports(const vector<LinkData>& links)
352 : {
353 18 : HCCL_INFO("Batch build opbased transports start, link num is [%u]", links.size());
354 7 : for (auto& link : links) {
355 1 : if (opTagOpbasedMap.find(link) != opTagOpbasedMap.end()) {
356 0 : HCCL_WARNING("linkData=%s already exists, do not need to create transport", link.Describe().c_str());
357 0 : continue;
358 0 : }
359 1 : CreateOpbasedMemTransport(link);
360 : }
361 6 : }
362 :
363 6 : void MemTransportManager::BatchBuildOffloadTransports(const std::string& opTag, const vector<LinkData>& links)
364 : {
365 18 : HCCL_INFO("Batch build offload transports start, link num is [%u]", links.size());
366 7 : for (auto& link : links) {
367 1 : if (opTagOffloadMap.find(opTag) != opTagOffloadMap.end()
368 1 : && opTagOffloadMap[opTag].find(link) != opTagOffloadMap[opTag].end()) {
369 0 : HCCL_WARNING(
370 : "opTag=%s, linkData=%s already exists, do not need to create transport", opTag.c_str(),
371 : link.Describe().c_str());
372 0 : continue;
373 0 : }
374 1 : CreateOffloadMemTransport(opTag, link);
375 : }
376 6 : }
377 :
378 3 : BaseMemTransport* MemTransportManager::GetOpbasedTransport(const LinkData& linkData)
379 : {
380 3 : if (opTagOpbasedMap.find(linkData) == opTagOpbasedMap.end()) {
381 6 : HCCL_WARNING("GetOpbasedTransport, linkData=%s find transport is null", linkData.Describe().c_str());
382 2 : return nullptr;
383 : }
384 1 : return opTagOpbasedMap[linkData].get();
385 : }
386 :
387 3 : BaseMemTransport* MemTransportManager::GetOffloadTransport(const std::string& opTag, const LinkData& linkData)
388 : {
389 3 : if (opTagOffloadMap.find(opTag) == opTagOffloadMap.end()) {
390 3 : HCCL_WARNING(
391 : "GetOffloadTransport, opTag=%s, linkData=%s find transport is null", opTag.c_str(),
392 : linkData.Describe().c_str());
393 1 : return nullptr;
394 : }
395 2 : if (opTagOffloadMap[opTag].find(linkData) == opTagOffloadMap[opTag].end()) {
396 3 : HCCL_WARNING(
397 : "GetOffloadTransport, opTag=%s, linkData=%s find transport is null", opTag.c_str(),
398 : linkData.Describe().c_str());
399 1 : return nullptr;
400 : }
401 :
402 1 : return opTagOffloadMap[opTag][linkData].get();
403 : }
404 :
405 0 : BaseMemTransport* MemTransportManager::GetUrmaDirectTransport(const LinkData& linkData)
406 : {
407 0 : if (urmaDirectMap_.find(linkData) == urmaDirectMap_.end()) {
408 0 : HCCL_WARNING("GetUrmaDirectTransport, linkData=%s find transport is null", linkData.Describe().c_str());
409 0 : return nullptr;
410 : }
411 0 : return urmaDirectMap_[linkData].get();
412 : }
413 :
414 2 : std::vector<char> MemTransportManager::GetOneSidedPackedData()
415 : {
416 2 : if (!IsAllOneSidedTransportReady()) {
417 0 : std::string msg = StringFormat("status of some transports is not ready, please check.");
418 0 : THROW<InternalException>(msg);
419 0 : }
420 :
421 2 : std::vector<char> result;
422 2 : BinaryStream binaryStream;
423 2 : u32 mapSize = oneSidedMap.size();
424 2 : binaryStream << mapSize;
425 :
426 2 : if (mapSize == 0) {
427 0 : HCCL_WARNING("mem transport oneSidedMap is empty");
428 : }
429 :
430 4 : for (auto& it : oneSidedMap) {
431 2 : binaryStream << it.first.GetUniqueId();
432 2 : binaryStream << it.second->GetUniqueId();
433 6 : HCCL_INFO(
434 : "MemTransportManager::GetOneSidedPackedData: %s %s.", it.first.Describe().c_str(),
435 : it.second->Describe().c_str());
436 : }
437 :
438 2 : binaryStream.Dump(result);
439 2 : return result;
440 2 : }
441 :
442 1 : std::vector<HcclAiRMAWQ> MemTransportManager::GetUrmaWqs()
443 : {
444 1 : if (!IsAllTransportReady()) {
445 0 : std::string msg = StringFormat("status of some transports is not ready, please check.");
446 0 : THROW<InternalException>(msg);
447 0 : }
448 :
449 1 : std::vector<HcclAiRMAWQ> wqs;
450 1 : auto links = comm->GetFullMeshLinks();
451 3 : for (auto& link : links) {
452 2 : if (urmaDirectMap_.find(link) == urmaDirectMap_.end()) {
453 6 : HCCL_WARNING(
454 : "[MemTransportManager][GetUrmaWqs]GetUrmaDirectTransport, linkData=%s find transport is null",
455 : link.Describe().c_str());
456 2 : continue;
457 2 : }
458 0 : UrmaDirectTransport* urmaTransport = reinterpret_cast<UrmaDirectTransport*>(urmaDirectMap_[link].get());
459 :
460 0 : wqs.push_back(urmaTransport->GetAiRMAWQ());
461 0 : HCCL_INFO("MemTransportManager::GetUrmaWq: %s.", link.Describe().c_str());
462 : }
463 1 : return wqs;
464 1 : }
465 :
466 1 : std::vector<HcclAiRMACQ> MemTransportManager::GetUrmaCqs()
467 : {
468 1 : if (!IsAllTransportReady()) {
469 0 : std::string msg = StringFormat("status of some transports is not ready, please check.");
470 0 : THROW<InternalException>(msg);
471 0 : }
472 :
473 1 : std::vector<HcclAiRMACQ> cqs;
474 1 : auto links = comm->GetFullMeshLinks();
475 3 : for (auto& link : links) {
476 2 : if (urmaDirectMap_.find(link) == urmaDirectMap_.end()) {
477 6 : HCCL_WARNING(
478 : "[MemTransportManager][GetUrmaWqs]GetUrmaDirectTransport, linkData=%s find transport is null",
479 : link.Describe().c_str());
480 2 : continue;
481 2 : }
482 0 : UrmaDirectTransport* urmaTransport = reinterpret_cast<UrmaDirectTransport*>(urmaDirectMap_[link].get());
483 :
484 0 : cqs.push_back(urmaTransport->GetAiRMACQ());
485 0 : HCCL_INFO("MemTransportManager::GetUrmaCq: %s.", link.Describe().c_str());
486 : }
487 :
488 1 : return cqs;
489 1 : }
490 :
491 2 : std::vector<char> MemTransportManager::GetOpbasedPackedData()
492 : {
493 2 : if (!IsAllOpbasedTransportReady()) {
494 0 : std::string msg = StringFormat("status of some transports is not ready, please check.");
495 0 : THROW<InternalException>(msg);
496 0 : }
497 :
498 2 : std::vector<char> result;
499 2 : BinaryStream binaryStream;
500 2 : u32 mapSize = opTagOpbasedMap.size();
501 2 : binaryStream << mapSize;
502 :
503 2 : if (mapSize == 0) {
504 6 : HCCL_WARNING("mem transport opTagOpbasedMap is empty");
505 : }
506 :
507 2 : for (auto& it : opTagOpbasedMap) {
508 0 : binaryStream << it.first.GetUniqueId();
509 0 : binaryStream << it.second->GetUniqueId();
510 0 : HCCL_INFO("MemTransportManager::GetOpbasedPackedData: %s.", it.first.Describe().c_str());
511 : }
512 :
513 2 : binaryStream.Dump(result);
514 2 : return result;
515 2 : }
516 :
517 2 : std::vector<char> MemTransportManager::GetOffloadPackedData(const std::string& opTag)
518 : {
519 2 : if (!IsAllOffloadTransportReady(opTag)) {
520 : std::string msg
521 0 : = StringFormat("status of some transports is not ready, please check. opTag[%s]", opTag.c_str());
522 0 : THROW<InternalException>(msg);
523 0 : }
524 :
525 2 : std::vector<char> result;
526 2 : BinaryStream binaryStream;
527 2 : u32 mapSize = 0;
528 :
529 2 : auto transpMap = opTagOffloadMap.find(opTag);
530 2 : if (transpMap != opTagOffloadMap.end()) {
531 1 : mapSize = transpMap->second.size();
532 1 : binaryStream << mapSize;
533 2 : for (auto& it : transpMap->second) {
534 1 : binaryStream << it.first.GetUniqueId();
535 1 : binaryStream << it.second->GetUniqueId();
536 3 : HCCL_INFO("MemTransportManager::GetOffloadPackedData: %s.", it.first.Describe().c_str());
537 : }
538 : } else {
539 3 : HCCL_WARNING("mem transport opTagOffloadMap is empty for opTag[%s]", opTag.c_str());
540 1 : binaryStream << mapSize;
541 : }
542 :
543 2 : binaryStream.Dump(result);
544 2 : return result;
545 2 : }
546 :
547 1 : std::vector<char> MemTransportManager::GetPackedAllTransportData()
548 : {
549 : /* 打包的数据：
550 : {
551 : u32 opbasedMapSize
552 : 对opTagOpbasedMap里的每个pair:
553 : vector<char> Opbase linkdata
554 : vector<char> Opbase transport
555 : u32 opTagNum
556 : 对opTagOffloadMap里的每个opTag:
557 : vector<char> opTag
558 : u32 offloadMapSize
559 : 对opTagOffloadMap[opTag]里的每个pair:
560 : vector<char> Offload linkdata
561 : vector<char> Offload transport
562 : }
563 : */
564 :
565 1 : std::vector<char> result;
566 1 : BinaryStream binaryStream;
567 :
568 1 : u32 opbasedMapSize = opTagOpbasedMap.size();
569 3 : HCCL_INFO("GetPackedAllTransportData: opbasedMapSize=%u", opbasedMapSize);
570 1 : binaryStream << opbasedMapSize;
571 1 : for (auto& it : opTagOpbasedMap) {
572 0 : binaryStream << it.first.GetUniqueId();
573 0 : binaryStream << it.second->GetUniqueId();
574 : }
575 :
576 1 : u32 opTagNum = opTagOffloadMap.size();
577 3 : HCCL_INFO("GetPackedAllTransportData: opTagNum=%u", opTagNum);
578 1 : binaryStream << opTagNum;
579 1 : for (auto& opTagIt : opTagOffloadMap) {
580 0 : std::string opTag = opTagIt.first;
581 0 : std::vector<char> opTagVec(opTag.begin(), opTag.end());
582 0 : binaryStream << opTagVec;
583 0 : u32 offloadMapSize = opTagIt.second.size();
584 0 : binaryStream << offloadMapSize;
585 0 : for (auto& it : opTagIt.second) {
586 0 : binaryStream << it.first.GetUniqueId();
587 0 : binaryStream << it.second->GetUniqueId();
588 : }
589 0 : }
590 :
591 1 : binaryStream.Dump(result);
592 1 : return result;
593 1 : }
594 :
595 1 : BaseMemTransport* MemTransportManager::RecoverOpbasedMemTransport(const LinkData& linkData)
596 : {
597 3 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
598 1 : BaseMemTransport::CommonLocRes locRes;
599 1 : locRes.notifyVec = GetNotifyVec(linkData);
600 3 : HCCL_INFO("link=%s get notifyVec OK", linkData.Describe().c_str());
601 :
602 : // buffer 来自localRmaBufferManager, input/output/scratch
603 1 : locRes.bufferVec = GetBufferVec(comm->GetId(), linkData, OpMode::OPBASE);
604 3 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
605 :
606 : // connection是一个，来自 RmaConnManager
607 1 : locRes.connVec = GetConnVec(comm->GetId(), linkData);
608 3 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
609 :
610 3 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
611 :
612 1 : BaseMemTransport::Attribution attr;
613 1 : attr.devicePhyId = linkData.GetLocalPort().GetId();
614 :
615 1 : u32 crcValue{0};
616 3 : HCCL_INFO("[RecoverMemTransport]commptr=%p", comm);
617 :
618 1 : if (comm->IsWorldGroup()) {
619 : // 判断是否在框内
620 1 : if (comm->GetNeighboorRanks().find(linkData.GetRemoteRankId()) != comm->GetNeighboorRanks().end()) {
621 : // 在框内使用带LocalID的CRC值
622 1 : crcValue = comm->GetRanktableCrc(true);
623 : } else {
624 : // 不在框内使用不带LocalID的CRC值
625 0 : crcValue = comm->GetRanktableCrc(false);
626 : }
627 : }
628 :
629 : // 握手消息定义，包括 通信算子数目，rankTable CRC，通信步骤字段
630 1 : CollOperator op{};
631 2 : op.opTag = std::to_string(comm->GetCollOpIndex()) + "_" + std::to_string(crcValue) + "_"
632 3 : + std::to_string(comm->GetStep());
633 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
634 3 : HCCL_INFO("[MemTransportManager::CreateOpbasedMemTransport] accelerator[%s]", accelerator.Describe().c_str());
635 1 : attr.opAcceState = accelerator;
636 1 : attr.handshakeMsg = op.GetUniqueId();
637 :
638 1 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
639 1 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
640 1 : if (socket == nullptr) {
641 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
642 : }
643 1 : if (linkData.GetType() == PortDeploymentType::P2P) {
644 0 : opTagOpbasedMap[linkData] = make_unique<P2PTransport>(locRes, attr, linkData, *socket);
645 1 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
646 1 : auto linkProtocol = linkData.GetLinkProtocol();
647 1 : if (linkProtocol == LinkProtocol::UB_CTP || linkProtocol == LinkProtocol::UB_TP) {
648 1 : CreateOpbasedUbMemTransport(locRes, attr, linkData, *socket);
649 : } else {
650 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
651 : }
652 : } else {
653 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
654 : }
655 :
656 1 : opTagOpbasedMap[linkData]->Establish();
657 :
658 1 : newOpbasedTransports[linkData] = 0;
659 :
660 3 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
661 3 : HCCL_INFO("create transport %s OK.", opTagOpbasedMap[linkData]->Describe().c_str());
662 :
663 2 : return opTagOpbasedMap[linkData].get();
664 1 : }
665 :
666 1 : BaseMemTransport* MemTransportManager::RecoverOffloadMemTransport(const std::string& opTag, const LinkData& linkData)
667 : {
668 3 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
669 1 : BaseMemTransport::CommonLocRes locRes;
670 1 : locRes.notifyVec = GetNotifyVec(linkData);
671 3 : HCCL_INFO("link=%s get notifyVec OK", linkData.Describe().c_str());
672 :
673 : // buffer 来自localRmaBufferManager, input/output/scratch
674 1 : locRes.bufferVec = GetBufferVec(opTag, linkData, OpMode::OFFLOAD);
675 3 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
676 :
677 : // connection是一个，来自 RmaConnManager
678 1 : locRes.connVec = GetConnVec(opTag, linkData);
679 3 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
680 :
681 3 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
682 :
683 1 : BaseMemTransport::Attribution attr;
684 1 : attr.devicePhyId = linkData.GetLocalPort().GetId();
685 :
686 1 : u32 crcValue{0};
687 3 : HCCL_INFO("[RecoverMemTransport]commptr=%p", comm);
688 :
689 1 : if (comm->IsWorldGroup()) {
690 : // 判断是否在框内
691 1 : if (comm->GetNeighboorRanks().find(linkData.GetRemoteRankId()) != comm->GetNeighboorRanks().end()) {
692 : // 在框内使用带LocalID的CRC值
693 1 : crcValue = comm->GetRanktableCrc(true);
694 : } else {
695 : // 不在框内使用不带LocalID的CRC值
696 0 : crcValue = comm->GetRanktableCrc(false);
697 : }
698 : }
699 : // 握手消息定义，包括 通信算子数目，rankTable CRC，通信步骤字段
700 1 : CollOperator op{};
701 2 : op.opTag = std::to_string(comm->GetCollOpIndex()) + "_" + std::to_string(crcValue) + "_"
702 3 : + std::to_string(comm->GetStep());
703 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
704 3 : HCCL_INFO("[MemTransportManager::CreateOpbasedMemTransport] accelerator[%s]", accelerator.Describe().c_str());
705 1 : attr.opAcceState = accelerator;
706 1 : attr.handshakeMsg = op.GetUniqueId();
707 :
708 1 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
709 1 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
710 1 : if (socket == nullptr) {
711 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
712 : }
713 1 : if (linkData.GetType() == PortDeploymentType::P2P) {
714 0 : opTagOffloadMap[opTag][linkData] = make_unique<P2PTransport>(locRes, attr, linkData, *socket);
715 1 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
716 1 : auto linkProtocol = linkData.GetLinkProtocol();
717 1 : if (linkProtocol == LinkProtocol::UB_CTP || linkProtocol == LinkProtocol::UB_TP) {
718 1 : CreateOffloadUbMemTransport(opTag, locRes, attr, linkData, *socket);
719 : } else {
720 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
721 : }
722 : } else {
723 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
724 : }
725 :
726 1 : opTagOffloadMap[opTag][linkData]->Establish();
727 :
728 1 : newOffloadTransports[opTag][linkData] = 0;
729 :
730 3 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
731 3 : HCCL_INFO("create transport %s OK.", opTagOffloadMap[opTag][linkData]->Describe().c_str());
732 :
733 2 : return opTagOffloadMap[opTag][linkData].get();
734 1 : }
735 :
736 : // 功能说明：根据输入的CommID和LinkData信息，恢复单算子Tansport对象，并将通信域一致信息改为RecoverInfo
737 : // 输入说明：vector<LinkData> &links：linkData数据
738 1 : void MemTransportManager::BatchRecoverOpbasedTransports(const vector<LinkData>& links)
739 : {
740 3 : HCCL_INFO("BatchRecoverOpbasedTransports start, link num is [%u]", links.size());
741 2 : for (auto& link : links) {
742 : // 校验transport是否已经构建
743 1 : if (opTagOpbasedMap.find(link) != opTagOpbasedMap.end()) {
744 0 : HCCL_WARNING("linkData=%s already exists, do not need to create transport", link.Describe().c_str());
745 0 : continue;
746 0 : }
747 : // 创建transport
748 1 : RecoverOpbasedMemTransport(link);
749 : }
750 1 : }
751 :
752 : // 功能说明：根据输入的CommID和LinkData信息，恢复图模式Tansport对象，并将通信域一致信息改为RecoverInfo
753 : // 输入说明：vector<LinkData> &links：linkData数据
754 : // std::string &opTag：commId，通信域标记
755 1 : void MemTransportManager::BatchRecoverOffloadTransports(const std::string& opTag, const vector<LinkData>& links)
756 : {
757 3 : HCCL_INFO("BatchRecoverOffloadTransports start, link num is [%u]", links.size());
758 2 : for (auto& link : links) {
759 : // 校验transport是否已经构建
760 1 : if (opTagOffloadMap.find(opTag) != opTagOffloadMap.end()
761 1 : && opTagOffloadMap[opTag].find(link) != opTagOffloadMap[opTag].end()) {
762 0 : HCCL_WARNING(
763 : "opTag=%s, linkData=%s already exists, do not need to create transport", opTag.c_str(),
764 : link.Describe().c_str());
765 0 : continue;
766 0 : }
767 : // 创建transport
768 1 : RecoverOffloadMemTransport(opTag, link);
769 : }
770 1 : }
771 :
772 : // 功能说明：单算子场景，推动式建链，建链成功后，使用RankConsistent校验通信域一致性
773 2 : bool MemTransportManager::IsAllOpbasedTransportRecoveredReady()
774 : {
775 2 : bool isAllTransportRecoveredReady = true;
776 : // 当前只针对新增的transports做资源交换和op校验
777 4 : for (auto linkIt = newOpbasedTransports.begin(); linkIt != newOpbasedTransports.end();) {
778 : // 尝试建链
779 2 : auto status = opTagOpbasedMap[linkIt->first]->GetStatus();
780 2 : if (status != TransportStatus::READY) {
781 1 : if (status == TransportStatus::SOCKET_TIMEOUT) {
782 0 : MACRO_THROW(
783 : TimeoutException,
784 : StringFormat(
785 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check", __func__,
786 : opTagOpbasedMap[linkIt->first]->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
787 : }
788 : // 只要任意transport一个没有ready，整体建链结果为 false
789 1 : isAllTransportRecoveredReady = false;
790 1 : ++linkIt;
791 : } else {
792 3 : HCCL_INFO("linkData[%s], status[%s].", linkIt->first.Describe().c_str(), status.Describe().c_str());
793 1 : linkIt = newOpbasedTransports.erase(linkIt);
794 : }
795 : }
796 2 : return isAllTransportRecoveredReady;
797 : }
798 :
799 : // 功能说明：图模式场景，推动式建链，建链成功后，使用RankConsistent校验通信域一致性
800 : // 输入说明：std::string &opTag：commId，通信域标记
801 1 : bool MemTransportManager::IsAllOffloadTransportRecoveredReady(const std::string& opTag)
802 : {
803 1 : bool isAllTransportRecoveredReady = true;
804 : // 当前只针对新增的transports做资源交换和op校验
805 2 : for (auto linkIt = newOffloadTransports[opTag].begin(); linkIt != newOffloadTransports[opTag].end();) {
806 : // 尝试建链
807 1 : auto status = opTagOffloadMap[opTag][linkIt->first]->GetStatus();
808 1 : if (status != TransportStatus::READY) {
809 1 : if (status == TransportStatus::SOCKET_TIMEOUT) {
810 0 : MACRO_THROW(
811 : TimeoutException,
812 : StringFormat(
813 : "[MemTransportManager][%s] %s socket timeout, commId[%s], please check", __func__,
814 : opTagOffloadMap[opTag][linkIt->first]->GetLinkDescInfo().c_str(), comm->GetId().c_str()));
815 : }
816 : // 只要任意transport一个没有ready，整体建链结果为 false
817 1 : isAllTransportRecoveredReady = false;
818 1 : ++linkIt;
819 : } else {
820 0 : HCCL_INFO(
821 : "opTag[%s] linkData[%s] status[%s]", opTag.c_str(), linkIt->first.Describe().c_str(),
822 : status.Describe().c_str());
823 0 : linkIt = newOffloadTransports[opTag].erase(linkIt);
824 : }
825 : }
826 1 : return isAllTransportRecoveredReady;
827 : }
828 :
829 4 : void MemTransportManager::Clear()
830 : {
831 4 : opTagOpbasedMap.clear();
832 4 : std::vector<RmaConnection*> emptyVec;
833 5 : for (auto& offloadMapIt : opTagOffloadMap) {
834 2 : for (auto& memTransportMapIt : offloadMapIt.second) {
835 1 : memTransportMapIt.second->SetConnVec(emptyVec);
836 : }
837 : }
838 4 : }
839 :
840 2 : void MemTransportManager::UpdateOffloadTransports()
841 : {
842 6 : HCCL_INFO("[UpdateOffloadTransports] start, opTagOffloadMap size is [%u]", opTagOffloadMap.size());
843 3 : for (auto& it : opTagOffloadMap) {
844 1 : std::string opTag = it.first;
845 3 : HCCL_INFO("[UpdateOffloadTransports] start, opTag[%s]", opTag.c_str());
846 2 : for (auto& linkTransPair : it.second) {
847 1 : auto connectVec = GetConnVec(comm->GetId(), linkTransPair.first);
848 1 : linkTransPair.second->SetConnVec(connectVec);
849 1 : }
850 1 : }
851 2 : }
852 2 : BaseMemTransport* MemTransportManager::GetOneSidedTransport(const LinkData& linkData)
853 : {
854 2 : if (oneSidedMap.find(linkData) == oneSidedMap.end()) {
855 0 : HCCL_WARNING("GetOpbasedTransport, linkData=%s find transport is null", linkData.Describe().c_str());
856 0 : return nullptr;
857 : }
858 2 : return oneSidedMap[linkData].get();
859 : }
860 :
861 1 : void MemTransportManager::CreateOneSidedUbMemTransport(
862 : BaseMemTransport::CommonLocRes& locRes, BaseMemTransport::Attribution& attr, const LinkData& linkData,
863 : const Socket& socket)
864 : {
865 1 : auto topicIdCntNotifyVecMap = comm->GetConnLocalCntNotifyManager().GetTopicIdCntNotifyMap(linkData.GetLocalPort());
866 : CntNotifyResHelper tool;
867 1 : BaseMemTransport::LocCntNotifyRes locCntNotifyRes = tool.GetCntNotifyRes(topicIdCntNotifyVecMap);
868 3 : HCCL_INFO("locCntNotifyRes=%s, linkData=%s", locCntNotifyRes.Describe().c_str(), linkData.Describe().c_str());
869 2 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(
870 1 : comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
871 :
872 : // DFX：注册transportCallBack, 用于信息保存
873 1 : auto transportCallBack = MemTransportCallback(linkData, comm->GetMirrorTaskManager());
874 : auto ubMemTransport
875 1 : = make_unique<UbMemTransport>(locRes, attr, linkData, socket, rdmaHandle, locCntNotifyRes, transportCallBack);
876 3 : HCCL_INFO("[CreateOneSidedUbMemTransport] Add oneSidedMap");
877 1 : oneSidedMap[linkData] = std::move(ubMemTransport);
878 1 : }
879 :
880 1 : BaseMemTransport* MemTransportManager::CreateOneSidedTransport(const LinkData& linkData)
881 : {
882 3 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
883 1 : BaseMemTransport::CommonLocRes locRes;
884 1 : locRes.notifyVec = GetNotifyVec(linkData);
885 3 : HCCL_INFO("link=%s get notifyVec OK", linkData.Describe().c_str());
886 :
887 : // buffer 来自localRmaBufferManager, input/output/scratch
888 1 : locRes.bufferVec = GetBufferVec(comm->GetId(), linkData, OpMode::OFFLOAD);
889 3 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
890 :
891 : // connection是一个，来自 RmaConnManager
892 1 : locRes.connVec = GetConnVec(comm->GetId(), linkData);
893 3 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
894 :
895 3 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
896 :
897 1 : BaseMemTransport::Attribution attr;
898 1 : attr.devicePhyId = linkData.GetLocalPort().GetId();
899 :
900 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
901 3 : HCCL_INFO("[MemTransportManager::CreateOneSidedTransport] accelerator[%s]", accelerator.Describe().c_str());
902 1 : attr.opAcceState = accelerator;
903 :
904 1 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
905 1 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
906 1 : if (socket == nullptr) {
907 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
908 : }
909 1 : if (linkData.GetType() == PortDeploymentType::P2P) {
910 0 : oneSidedMap[linkData] = make_unique<P2PTransport>(locRes, attr, linkData, *socket);
911 1 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
912 1 : if (linkData.GetLinkProtocol() == LinkProtocol::UB_CTP || linkData.GetLinkProtocol() == LinkProtocol::UB_TP) {
913 3 : HCCL_INFO("CreateOneSidedUbMemTransport start");
914 1 : CreateOneSidedUbMemTransport(locRes, attr, linkData, *socket);
915 3 : HCCL_INFO("CreateOneSidedUbMemTransport end");
916 : } else {
917 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
918 : }
919 : } else {
920 0 : THROW<NullPtrException>(StringFormat("linkData=%s is error", linkData.Describe().c_str()));
921 : }
922 :
923 3 : HCCL_INFO("CreateOneSidedTransport Establish");
924 1 : oneSidedMap[linkData]->Establish();
925 :
926 3 : HCCL_INFO("CreateOneSidedTransport equal 0");
927 1 : newOneSidedTransports[linkData] = 0;
928 :
929 3 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
930 3 : HCCL_INFO("create transport %s OK", oneSidedMap[linkData]->Describe().c_str());
931 :
932 2 : return oneSidedMap[linkData].get();
933 1 : }
934 :
935 1 : void MemTransportManager::BatchBuildOneSidedTransports(const vector<LinkData>& links)
936 : {
937 3 : HCCL_INFO("Batch build opbased transports start, link num is [%u]", links.size());
938 2 : for (auto& link : links) {
939 1 : if (opTagOpbasedMap.find(link) != opTagOpbasedMap.end()) {
940 0 : HCCL_WARNING("linkData=%s already exists, do not need to create transport", link.Describe().c_str());
941 0 : continue;
942 0 : }
943 1 : CreateOneSidedTransport(link);
944 : }
945 1 : }
946 :
947 0 : void MemTransportManager::CreateUrmaDirectTransport(
948 : BaseMemTransport::CommonLocRes& locRes, BaseMemTransport::Attribution& attr, const LinkData& linkData,
949 : const Socket& socket)
950 : {
951 0 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(
952 0 : comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
953 :
954 : // DFX：注册transportCallBack, 用于信息保存
955 0 : auto transportCallBack = MemTransportCallback(linkData, comm->GetMirrorTaskManager());
956 0 : auto transport = make_unique<UrmaDirectTransport>(locRes, attr, linkData, socket, rdmaHandle, transportCallBack);
957 0 : HCCL_INFO("[CreateUrmaDirectTransport] Add urmaDirectMap_");
958 0 : urmaDirectMap_[linkData] = std::move(transport);
959 0 : }
960 :
961 0 : BaseMemTransport* MemTransportManager::CreateUrmaDirectTransport(const LinkData& linkData)
962 : {
963 0 : auto op = comm->GetCurrentCollOperator();
964 0 : HCCL_INFO("link=%s Entry CreateMemTransport", linkData.Describe().c_str());
965 0 : HCCL_INFO("Entry CreateMemTransport, opInfo=[%s]", CollOpToString(*op).c_str());
966 0 : BaseMemTransport::CommonLocRes locRes;
967 :
968 : // buffer 来自localRmaBufferManager, input/output/scratch
969 0 : locRes.bufferVec = GetBufferVec(GetAivUrmaBufferTag(comm->GetId()), linkData, OpMode::OPBASE);
970 0 : HCCL_INFO("link=%s get bufferVec OK", linkData.Describe().c_str());
971 :
972 : // connection是一个，来自 RmaConnManager
973 0 : locRes.connVec = GetConnVec(comm->GetId(), linkData);
974 0 : HCCL_INFO("link=%s get connVec OK", linkData.Describe().c_str());
975 :
976 0 : HCCL_INFO("locRes=%s", locRes.Describe().c_str());
977 :
978 0 : BaseMemTransport::Attribution attr;
979 0 : attr.devicePhyId = linkData.GetLocalPort().GetId();
980 : // 握手消息定义，未来包括 cann版本号，rankTable CRC等字段
981 0 : attr.handshakeMsg = op->GetUniqueId();
982 :
983 0 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, comm->GetEstablishLinkSocketTag());
984 0 : auto socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
985 0 : if (socket == nullptr) {
986 0 : throw std::runtime_error("CreateMemTransport GetConnectedSocket failed, socket is nullptr");
987 : }
988 :
989 0 : CreateUrmaDirectTransport(locRes, attr, linkData, *socket);
990 :
991 0 : urmaDirectMap_[linkData]->Establish();
992 :
993 0 : HCCL_INFO("link=%s OK.", linkData.Describe().c_str());
994 0 : HCCL_INFO("create transport %s OK.", urmaDirectMap_[linkData]->Describe().c_str());
995 :
996 0 : return urmaDirectMap_[linkData].get();
997 0 : }
998 :
999 0 : void MemTransportManager::BatchBuildUrmaDirectTransports(const vector<LinkData>& links)
1000 : {
1001 0 : HCCL_INFO("Batch build urma direct transports start, link num is [%u]", links.size());
1002 0 : for (auto& link : links) {
1003 0 : if (urmaDirectMap_.find(link) != urmaDirectMap_.end()) {
1004 0 : HCCL_WARNING("linkData=%s already exists, do not need to create transport", link.Describe().c_str());
1005 0 : continue;
1006 0 : }
1007 0 : CreateUrmaDirectTransport(link);
1008 : }
1009 0 : }
1010 :
1011 1 : HcclResult MemTransportManager::ClearOpTransport(const std::string& opTag)
1012 : {
1013 1 : if (opTagOffloadMap.find(opTag) == opTagOffloadMap.end()) {
1014 3 : HCCL_WARNING(
1015 : "[LocalRmaBufManager::%s] opTag[%s] Cannot find Transport in opTagOffloadMap.", __func__, opTag.c_str());
1016 : }
1017 1 : if (newOffloadTransports.find(opTag) == newOffloadTransports.end()) {
1018 3 : HCCL_WARNING(
1019 : "[LocalRmaBufManager::%s] opTag[%s] Cannot find Transport in newOffloadTransports.", __func__,
1020 : opTag.c_str());
1021 : }
1022 1 : opTagOffloadMap.erase(opTag);
1023 1 : newOffloadTransports.erase(opTag);
1024 1 : return HCCL_SUCCESS;
1025 : }
1026 :
1027 : } // namespace Hccl
|
