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 397 : UbMemoryTransportMgr::~UbMemoryTransportMgr()
22 : {
23 199 : tempTransport.clear();
24 199 : ubMemLink2TransportMap.clear();
25 397 : }
26 3 : HcclResult UbMemoryTransportMgr::BatchCreateTransport(const std::vector<LinkData>& links)
27 : {
28 9 : HCCL_INFO("[%s] start", __func__);
29 4 : for (auto& link : links) {
30 1 : auto ret = CreateTransportByLink(link);
31 1 : if (ret != HcclResult::HCCL_SUCCESS) {
32 0 : HCCL_ERROR(
33 : "[UbMemoryTransportMgr::%s] CreateTransportByLink fail link[%s]", __func__, link.Describe().c_str());
34 0 : return ret;
35 : }
36 : }
37 :
38 3 : return HcclResult::HCCL_SUCCESS;
39 : }
40 :
41 4 : std::vector<std::pair<RankId, RemoteIpcRmaBuffer*>> UbMemoryTransportMgr::GetRmtRankId2RmtIpcRmaBufList()
42 : {
43 12 : HCCL_INFO("[%s] start", __func__);
44 4 : std::vector<std::pair<RankId, RemoteIpcRmaBuffer*>> rankId2RmtIpcRmaBufList{};
45 :
46 4 : for (const auto& ubMemLink2TransportIter : ubMemLink2TransportMap) {
47 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
48 0 : auto rmtMemBuffer = ubMemLink2TransportIter.second->GetRmtMemBuffer(0);
49 0 : rankId2RmtIpcRmaBufList.push_back(std::make_pair(rmtRank, rmtMemBuffer));
50 : }
51 :
52 4 : return rankId2RmtIpcRmaBufList;
53 0 : }
54 :
55 0 : std::vector<std::pair<RankId, uintptr_t>> UbMemoryTransportMgr::GetAllRankId2AivTagBufAddrList()
56 : {
57 0 : HCCL_INFO("[%s] start", __func__);
58 0 : std::vector<std::pair<RankId, uintptr_t>> rankId2AivTagBufList{};
59 :
60 0 : for (const auto& ubMemLink2TransportIter : ubMemLink2TransportMap) {
61 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
62 0 : uintptr_t rmtAivTagufferAddr = ubMemLink2TransportIter.second->GetRmtMemBuffer(AIV_TAG_BUF_INDEX)->GetAddr();
63 0 : rankId2AivTagBufList.push_back(std::make_pair(rmtRank, rmtAivTagufferAddr));
64 : }
65 0 : rankId2AivTagBufList.push_back(std::make_pair(comm->GetMyRank(), comm->GetAivTagBuffer()->GetAddr()));
66 :
67 0 : return rankId2AivTagBufList;
68 0 : }
69 :
70 1 : std::vector<std::pair<RankId, uintptr_t>> UbMemoryTransportMgr::GetAllRankId2AivOffloadTagBufAddrList()
71 :
72 : {
73 3 : HCCL_INFO("[%s] start", __func__);
74 :
75 1 : std::vector<std::pair<RankId, uintptr_t>> rankId2AivOffloadTagBufList{};
76 :
77 1 : for (const auto& ubMemLink2TransportIter : ubMemLink2TransportMap) {
78 0 : auto rmtRank = ubMemLink2TransportIter.first.GetRemoteRankId();
79 : uintptr_t rmtAivTagBufferAddr
80 0 : = ubMemLink2TransportIter.second->GetRmtMemBuffer(AIV_OFFLOAD_TAG_BUF_INDEX)->GetAddr();
81 0 : rankId2AivOffloadTagBufList.push_back(std::make_pair(rmtRank, rmtAivTagBufferAddr));
82 : }
83 1 : rankId2AivOffloadTagBufList.push_back(std::make_pair(comm->GetMyRank(), comm->GetAivOffloadTagBuffer()->GetAddr()));
84 :
85 1 : return rankId2AivOffloadTagBufList;
86 0 : }
87 :
88 1 : HcclResult UbMemoryTransportMgr::CreateTransportByLink(const LinkData& link)
89 : {
90 3 : HCCL_INFO("[%s] start", __func__);
91 1 : auto linkIter = ubMemLink2TransportMap.find(link);
92 1 : if (linkIter != ubMemLink2TransportMap.end()) {
93 0 : return HcclResult::HCCL_SUCCESS;
94 : }
95 : // 创建socket
96 1 : std::string socketTag = comm->GetEstablishLinkSocketTag();
97 1 : SocketConfig socketConfig(link.GetRemoteRankId(), link, socketTag);
98 1 : Socket* socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
99 1 : if (socket == nullptr) {
100 0 : HCCL_WARNING("[UbMemoryTransportMgr::%s] Fail to get socket via link %s, ", __func__, link.Describe().c_str());
101 :
102 0 : return HcclResult::HCCL_E_INTERNAL;
103 : }
104 :
105 : std::unique_ptr<UbMemoryTransport> transport = make_unique<UbMemoryTransport>(
106 2 : comm->GetCclBuffer(), comm->GetAivTagBuffer(), comm->GetAivOffloadTagBuffer(), socket,
107 2 : 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(
131 : "Invalid status occurs when creating transport connection %s!",
132 0 : (*transIter).first->Describe().c_str()));
133 6 : } else if (status == UbMemoryTransport::UBTransportStatus::SOCKET_TIMEOUT) {
134 0 : RPT_INPUT_ERR(
135 : true, "EI0006", std::vector<std::string>({"reason"}),
136 : std::vector<std::string>({"UbMemoryTransport wait SOCKET_TIMEOUT."}));
137 0 : THROW<TimeoutException>(StringFormat(
138 : "[UbMemoryTransportMgr][%s] [UbMemoryTransport]%s [LinkData]%s "
139 : "socket timeout, commId[%s], please check",
140 0 : __func__, (*transIter).first->Describe().c_str(), (*transIter).second.Describe().c_str(),
141 0 : comm->GetId().c_str()));
142 : } else {
143 6 : ++transIter;
144 : }
145 : }
146 :
147 7 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
148 : // 上报故障码EI0006
149 0 : RPT_INPUT_ERR(
150 : true, "EI0006", std::vector<std::string>({"reason"}),
151 : std::vector<std::string>({"UbMemoryTransportMgr wait transports ready timeout."}));
152 0 : THROW<InternalException>(
153 0 : "UbMemoryTransportMgr::WaitTransportReady timeout, commId[%s]", comm->GetId().c_str());
154 : }
155 : }
156 1 : }
157 :
158 1 : vector<std::pair<UbMemoryTransport*, LinkData>> UbMemoryTransportMgr::GetUnconfirmedTrans()
159 : {
160 3 : HCCL_INFO("[%s] start", __func__);
161 1 : if (tempTransport.size() == 0) {
162 0 : HCCL_WARNING("[UbMemoryTransportMgr::%s] UnConfirmedTrans does not exist, please check.", __func__);
163 0 : return vector<std::pair<UbMemoryTransport*, LinkData>>();
164 : }
165 :
166 1 : vector<std::pair<UbMemoryTransport*, LinkData>> unConfirmedTrans;
167 2 : for (const auto& linkId : tempTransport) {
168 1 : auto iterLink = ubMemLink2TransportMap.find(linkId);
169 1 : unConfirmedTrans.emplace_back(std::make_pair(iterLink->second.get(), linkId));
170 : }
171 1 : return unConfirmedTrans;
172 1 : }
173 :
174 1 : void UbMemoryTransportMgr::TransportsConnect()
175 : {
176 3 : HCCL_INFO("[%s] start", __func__);
177 : // transport建链
178 1 : vector<std::pair<UbMemoryTransport*, LinkData>> transLinkPairs = GetUnconfirmedTrans();
179 1 : auto op = comm->GetCurrentCollOperator();
180 1 : auto accelerator = comm->GetOpExecuteConfig().accState;
181 3 : HCCL_INFO("[UbMemoryTransportMgr::TransportsConnect] accelerator[%s]", accelerator.Describe().c_str());
182 2 : for (auto& pair : transLinkPairs) {
183 1 : auto transport = pair.first;
184 1 : transport->SetLocalOpAcceState(accelerator);
185 1 : transport->SetHandshakeMsg(op->GetUniqueId());
186 :
187 3 : HCCL_INFO("[UbMemoryTransport::%s] transport=[%s]", __func__, transport->Describe().c_str());
188 3 : HCCL_INFO("[UbMemoryTransport::%s] links=[%s]", __func__, pair.second.Describe().c_str());
189 : }
190 :
191 : // 轮询Connect
192 1 : WaitTransportsReady(transLinkPairs);
193 :
194 1 : tempTransport.clear();
195 3 : HCCL_INFO("[UbMemoryTransport::%s] transports connect end.", __func__);
196 1 : }
197 : } // namespace Hccl
|
