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