Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 : #include "ub_memory_transport_mgr.h"
11 : #include "timeout_exception.h"
12 : #include "communicator_impl.h"
13 : #include "adapter_error_manager_pub.h"
14 :
15 : namespace Hccl {
16 :
17 : constexpr u32 AIV_TAG_BUF_INDEX = 1; // aiv tag buf的下标
18 : constexpr u32 AIV_OFFLOAD_TAG_BUF_INDEX = 2; // aiv offload tag buf的下标
19 199 : UbMemoryTransportMgr::UbMemoryTransportMgr(const CommunicatorImpl &communicator) : comm(&communicator)
20 : {
21 199 : }
22 :
23 397 : UbMemoryTransportMgr::~UbMemoryTransportMgr()
24 : {
25 199 : tempTransport.clear();
26 199 : ubMemLink2TransportMap.clear();
27 397 : }
28 3 : HcclResult UbMemoryTransportMgr::BatchCreateTransport(const std::vector<LinkData> &links)
29 : {
30 9 : HCCL_INFO("[%s] start", __func__);
31 4 : for (auto &link : links) {
32 1 : auto ret = CreateTransportByLink(link);
33 1 : if (ret != HcclResult::HCCL_SUCCESS) {
34 0 : HCCL_ERROR("[UbMemoryTransportMgr::%s] CreateTransportByLink fail link[%s]", __func__,
35 : link.Describe().c_str());
36 0 : return ret;
37 : }
38 : }
39 :
40 3 : return HcclResult::HCCL_SUCCESS;
41 : }
42 :
43 4 : std::vector<std::pair<RankId, RemoteIpcRmaBuffer *>> UbMemoryTransportMgr::GetRmtRankId2RmtIpcRmaBufList()
44 : {
45 12 : HCCL_INFO("[%s] start", __func__);
46 4 : std::vector<std::pair<RankId, RemoteIpcRmaBuffer *>> rankId2RmtIpcRmaBufList{};
47 :
48 4 : for (const auto &ubMemLink2TransportIter : ubMemLink2TransportMap) {
49 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
50 0 : auto rmtMemBuffer = ubMemLink2TransportIter.second->GetRmtMemBuffer(0);
51 0 : rankId2RmtIpcRmaBufList.push_back(std::make_pair(rmtRank, rmtMemBuffer));
52 : }
53 :
54 4 : return rankId2RmtIpcRmaBufList;
55 0 : }
56 :
57 0 : std::vector<std::pair<RankId, uintptr_t>> UbMemoryTransportMgr::GetAllRankId2AivTagBufAddrList()
58 : {
59 0 : HCCL_INFO("[%s] start", __func__);
60 0 : std::vector<std::pair<RankId, uintptr_t>> rankId2AivTagBufList{};
61 :
62 0 : for (const auto &ubMemLink2TransportIter : ubMemLink2TransportMap) {
63 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
64 0 : uintptr_t rmtAivTagufferAddr = ubMemLink2TransportIter.second->GetRmtMemBuffer(AIV_TAG_BUF_INDEX)->GetAddr();
65 0 : rankId2AivTagBufList.push_back(std::make_pair(rmtRank, rmtAivTagufferAddr));
66 : }
67 0 : rankId2AivTagBufList.push_back(std::make_pair(comm->GetMyRank(), comm->GetAivTagBuffer()->GetAddr()));
68 :
69 0 : return rankId2AivTagBufList;
70 0 : }
71 :
72 1 : std::vector<std::pair<RankId, uintptr_t>> UbMemoryTransportMgr::GetAllRankId2AivOffloadTagBufAddrList()
73 :
74 : {
75 3 : HCCL_INFO("[%s] start", __func__);
76 :
77 1 : std::vector<std::pair<RankId, uintptr_t>> rankId2AivOffloadTagBufList{};
78 :
79 1 : for (const auto &ubMemLink2TransportIter : ubMemLink2TransportMap) {
80 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
81 0 : uintptr_t rmtAivTagBufferAddr = ubMemLink2TransportIter.second->GetRmtMemBuffer(AIV_OFFLOAD_TAG_BUF_INDEX)->GetAddr();
82 0 : rankId2AivOffloadTagBufList.push_back(std::make_pair(rmtRank, rmtAivTagBufferAddr));
83 : }
84 1 : rankId2AivOffloadTagBufList.push_back(std::make_pair(comm->GetMyRank(), comm->GetAivOffloadTagBuffer()->GetAddr()));
85 :
86 1 : return rankId2AivOffloadTagBufList;
87 0 : }
88 :
89 1 : HcclResult UbMemoryTransportMgr::CreateTransportByLink(const LinkData &link)
90 : {
91 3 : HCCL_INFO("[%s] start", __func__);
92 1 : auto linkIter = ubMemLink2TransportMap.find(link);
93 1 : if (linkIter != ubMemLink2TransportMap.end()) {
94 0 : return HcclResult::HCCL_SUCCESS;
95 : }
96 : // 创建socket
97 1 : std::string socketTag = comm->GetEstablishLinkSocketTag();
98 1 : SocketConfig socketConfig(link.GetRemoteRankId(), link, socketTag);
99 1 : Socket *socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
100 1 : if (socket == nullptr) {
101 0 : HCCL_WARNING("[UbMemoryTransportMgr::%s] Fail to get socket via link %s, ", __func__, link.Describe().c_str());
102 :
103 0 : return HcclResult::HCCL_E_INTERNAL;
104 : }
105 :
106 : std::unique_ptr<UbMemoryTransport> transport = make_unique<UbMemoryTransport>(
107 1 : comm->GetCclBuffer(), comm->GetAivTagBuffer(), comm->GetAivOffloadTagBuffer(), socket, comm->GetDeviceLogicId());
108 :
109 1 : if (transport->Init() != HcclResult::HCCL_SUCCESS) {
110 0 : HCCL_ERROR("[UbMemoryTransportMgr][%s] transport init fail, link %s", __func__, link.Describe().c_str());
111 0 : return HCCL_E_INTERNAL;
112 : }
113 :
114 1 : tempTransport.emplace_back(link); // 插入TempTransport中表明Transport并未真正创建成功，需要等待握手确认
115 1 : ubMemLink2TransportMap[link] = std::move(transport);
116 1 : return HcclResult::HCCL_SUCCESS;
117 1 : }
118 1 : void UbMemoryTransportMgr::WaitTransportsReady(vector<std::pair<UbMemoryTransport *, LinkData>> &transports) const
119 : {
120 3 : HCCL_INFO("[%s] start", __func__);
121 :
122 1 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
123 1 : HcclUs startTime = std::chrono::steady_clock::now();
124 8 : while (!transports.empty()) {
125 14 : for (auto transIter = transports.begin(); transIter != transports.end();) {
126 7 : auto status = (*transIter).first->GetStatus();
127 7 : if (status == UbMemoryTransport::UBTransportStatus::READY) {
128 1 : transIter = transports.erase(transIter);
129 6 : } else if (status == UbMemoryTransport::UBTransportStatus::CONNECT_FAILED) {
130 0 : THROW<InternalException>(StringFormat("Invalid status occurs when creating transport connection %s!",
131 0 : (*transIter).first->Describe().c_str()));
132 6 : } else if (status == UbMemoryTransport::UBTransportStatus::SOCKET_TIMEOUT) {
133 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}),
134 : std::vector<std::string>({"UbMemoryTransport wait SOCKET_TIMEOUT."}));
135 0 : THROW<TimeoutException>(StringFormat("[UbMemoryTransportMgr][%s] [UbMemoryTransport]%s [LinkData]%s "
136 : "socket timeout, commId[%s], please check",
137 0 : __func__, (*transIter).first->Describe().c_str(),
138 0 : (*transIter).second.Describe().c_str(), comm->GetId().c_str()));
139 : } else {
140 6 : ++transIter;
141 : }
142 : }
143 :
144 7 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
145 : // 上报故障码EI0006
146 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}),
147 : std::vector<std::string>({"UbMemoryTransportMgr wait transports ready timeout."}));
148 0 : THROW<InternalException>("UbMemoryTransportMgr::WaitTransportReady timeout, commId[%s]", comm->GetId().c_str());
149 : }
150 : }
151 1 : }
152 :
153 1 : vector<std::pair<UbMemoryTransport *, LinkData>> UbMemoryTransportMgr::GetUnconfirmedTrans()
154 : {
155 3 : HCCL_INFO("[%s] start", __func__);
156 1 : if (tempTransport.size() == 0) {
157 0 : HCCL_WARNING("[UbMemoryTransportMgr::%s] UnConfirmedTrans does not exist, please check.", __func__);
158 0 : return vector<std::pair<UbMemoryTransport *, LinkData>>();
159 : }
160 :
161 1 : vector<std::pair<UbMemoryTransport *, LinkData>> unConfirmedTrans;
162 2 : for (const auto &linkId : tempTransport) {
163 1 : auto iterLink = ubMemLink2TransportMap.find(linkId);
164 1 : unConfirmedTrans.emplace_back(std::make_pair(iterLink->second.get(), linkId));
165 : }
166 1 : return unConfirmedTrans;
167 1 : }
168 :
169 1 : void UbMemoryTransportMgr::TransportsConnect()
170 : {
171 3 : HCCL_INFO("[%s] start", __func__);
172 : // transport建链
173 1 : vector<std::pair<UbMemoryTransport *, LinkData>> transLinkPairs = GetUnconfirmedTrans();
174 1 : auto op = comm->GetCurrentCollOperator();
175 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
176 3 : HCCL_INFO("[UbMemoryTransportMgr::TransportsConnect] accelerator[%s]", accelerator.Describe().c_str());
177 2 : for (auto &pair : transLinkPairs) {
178 1 : auto transport = pair.first;
179 1 : transport->SetLocalOpAcceState(accelerator);
180 1 : transport->SetHandshakeMsg(op->GetUniqueId());
181 :
182 3 : HCCL_INFO("[UbMemoryTransport::%s] transport=[%s]", __func__, transport->Describe().c_str());
183 3 : HCCL_INFO("[UbMemoryTransport::%s] links=[%s]", __func__, pair.second.Describe().c_str());
184 : }
185 :
186 : // 轮询Connect
187 1 : WaitTransportsReady(transLinkPairs);
188 :
189 1 : tempTransport.clear();
190 3 : HCCL_INFO("[UbMemoryTransport::%s] transports connect end.", __func__);
191 1 : }
192 : } // namespace Hccl
|
