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