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