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