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 "hccl_one_sided_conn.h"
12 : #include "connections_builder.h"
13 : #include "hccl_net_dev.h"
14 : #include "hccl_mem.h"
15 : #include "communicator_impl.h"
16 : #include "transport_urma_mem.h"
17 : namespace Hccl {
18 : using namespace std;
19 :
20 5 : HcclOneSidedConn::HcclOneSidedConn(CommunicatorImpl *comm, LinkData linkData) : comm_(comm), linkData_(linkData)
21 : {
22 5 : }
23 :
24 5 : HcclOneSidedConn::~HcclOneSidedConn()
25 : {
26 6 : for (const auto &pair : desc2netDevMap_) {
27 1 : const HcclNetDev &hcclNetDev = pair.second;
28 1 : HcclResult ret = HcclNetDevClose(hcclNetDev);
29 1 : if (ret != HCCL_SUCCESS) {
30 0 : HCCL_ERROR("[HcclOneSidedConn][~HcclOneSidedConn]HcclNetDevClose failed, descStr[%s], ret[%d].",
31 : pair.first.c_str(), ret);
32 : }
33 : }
34 5 : }
35 1 : HcclResult HcclOneSidedConn::Connect(const std::string &commId)
36 : {
37 3 : HCCL_INFO("[HcclOneSidedConn]Connect start");
38 :
39 : // Socket/RmaConnection建链
40 1 : vector<LinkData> links;
41 1 : links.push_back(linkData_);
42 1 : comm_->GetSocketManager().BatchCreateSockets(links);
43 1 : make_unique<ConnectionsBuilder>(*comm_)->BatchBuild(comm_->GetId(), links);
44 1 : comm_->GetMemTransportManager()->BatchBuildOneSidedTransports(links);
45 :
46 : // Transport粒度申请notify，aicpu76行那个
47 2 : for (auto &link : links) {
48 1 : comm_->GetConnLocalNotifyManager().ApplyFor(link.GetRemoteRankId(), link);
49 : }
50 :
51 : // 推动式建链
52 1 : WaitOneSidedTransportReady();
53 :
54 : // 保存socket
55 1 : SocketConfig socketConfig(linkData_.GetRemoteRankId(), linkData_, comm_->GetEstablishLinkSocketTag());
56 1 : socket_ = comm_->GetSocketManager().GetConnectedSocket(socketConfig);
57 1 : if (socket_ == nullptr) {
58 0 : HCCL_ERROR("[HcclOneSidedConn]socket_ is nullptr");
59 0 : return HCCL_E_PTR;
60 : }
61 :
62 : // 创建TransportUrmaMem并保存
63 2 : transportMemPtr_ = make_shared<TransportUrmaMem>(comm_->GetMemTransportManager()->GetOneSidedTransport(linkData_),
64 2 : remoteHcclBufMgr_);
65 1 : return HCCL_SUCCESS;
66 1 : }
67 :
68 1 : void HcclOneSidedConn::WaitOneSidedTransportReady()
69 : {
70 1 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
71 1 : HcclUs startTime = std::chrono::steady_clock::now();
72 : while (true) {
73 1 : if (comm_->GetMemTransportManager()->IsAllOneSidedTransportReady()) {
74 1 : break;
75 : }
76 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
77 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}),
78 : std::vector<std::string>({"WaitOneSidedTransportReady timeout, SOCKET_TIMEOUT."}));
79 0 : THROW<InternalException>("WaitOneSidedTransportReady timeout.");
80 : }
81 : }
82 1 : }
83 :
84 1 : HcclResult HcclOneSidedConn::SendLocalMemDesc(const HcclMemDescs &localMemDescs)
85 : {
86 3 : HCCL_INFO("[HcclOneSidedConn]SendLocalMemDesc start");
87 1 : socket_->Send((u8 *)(&localMemDescs.arrayLength), sizeof(u32));
88 3 : HCCL_INFO("send localMemDescs.arrayLength:%u", localMemDescs.arrayLength);
89 1 : if (localMemDescs.arrayLength == 0) {
90 0 : HCCL_INFO("localMemDescs.arrayLength[%u], no need to send data", localMemDescs.arrayLength);
91 : } else {
92 3 : HCCL_INFO("send descSize:%u", localMemDescs.arrayLength * sizeof(HcclMemDesc));
93 1 : if (static_cast<u64>(localMemDescs.arrayLength) > static_cast<u64>(UINT32_MAX) / sizeof(HcclMemDesc)) {
94 0 : THROW<InternalException>("integer overflow occurs");
95 : }
96 1 : socket_->Send((u8 *)(localMemDescs.array), localMemDescs.arrayLength * sizeof(HcclMemDesc));
97 : }
98 1 : return HCCL_SUCCESS;
99 : }
100 :
101 2 : HcclResult HcclOneSidedConn::ReceiveRemoteMemDesc(HcclMemDescs &remoteMemDescs, u32 &actualNumOfRemote)
102 : {
103 6 : HCCL_INFO("[HcclOneSidedConn]ReceiveRemoteMemDesc start");
104 2 : socket_->Recv((u8 *)(&actualNumOfRemote), sizeof(u32));
105 2 : remoteMemDescs.arrayLength = actualNumOfRemote;
106 6 : HCCL_INFO("receive actualNumOfRemote:%u", actualNumOfRemote);
107 2 : if (remoteMemDescs.arrayLength == 0) {
108 3 : HCCL_INFO("actualNumOfRemote[%u], no need to receive data", remoteMemDescs.arrayLength);
109 : } else {
110 1 : if (remoteMemDescs.array == nullptr) {
111 3 : HCCL_ERROR("[HcclOneSidedConn]remoteMemDescs.array is nullptr but actualNumOfRemote[%u] > 0", actualNumOfRemote);
112 1 : return HCCL_E_PTR;
113 : }
114 0 : HCCL_INFO("receive descSize:%u", actualNumOfRemote * sizeof(HcclMemDesc));
115 0 : socket_->Recv((u8 *)remoteMemDescs.array, actualNumOfRemote * sizeof(HcclMemDesc));
116 : }
117 1 : return HCCL_SUCCESS;
118 : }
119 :
120 1 : HcclResult HcclOneSidedConn::ExchangeMemDesc(const HcclMemDescs &localMemDescs, HcclMemDescs &remoteMemDescs,
121 : u32 &actualNumOfRemote)
122 : {
123 3 : HCCL_INFO("[HcclOneSidedConn]ExchangeMemDesc start");
124 1 : CHK_PRT_RET(
125 : (localMemDescs.array == nullptr) && (remoteMemDescs.array == nullptr),
126 : HCCL_ERROR(
127 : "[HcclOneSidedConn]localMemDesc array and remoteMemDesc array are both nullptr, do not need to exchange"),
128 : HCCL_E_PARA);
129 1 : CHK_PRT_RET((localMemDescs.arrayLength == 0) && (remoteMemDescs.arrayLength == 0),
130 : HCCL_ERROR("[HcclOneSidedConn]localMemDesc arrayLength = %u and remoteMemDescs arrayLength = %u , do "
131 : "not need to exchange",
132 : localMemDescs.arrayLength, remoteMemDescs.arrayLength),
133 : HCCL_E_PARA);
134 :
135 1 : if (socket_ == nullptr) {
136 1 : CHK_RET(Connect(comm_->GetId()));
137 : }
138 :
139 1 : if (socket_->GetRole() == SocketRole::CLIENT) {
140 : // 先收后发
141 0 : CHK_RET(ReceiveRemoteMemDesc(remoteMemDescs, actualNumOfRemote));
142 0 : CHK_RET(SendLocalMemDesc(localMemDescs));
143 : } else {
144 : // 先发后收
145 1 : CHK_RET(SendLocalMemDesc(localMemDescs));
146 1 : CHK_RET(ReceiveRemoteMemDesc(remoteMemDescs, actualNumOfRemote));
147 : }
148 :
149 : // 校验remoteDescs中的remoteRankId和conn对象中保存的localRankId是否一样
150 1 : for (u32 i = 0; i < actualNumOfRemote; i++) {
151 0 : CHK_PTR_NULL(remoteMemDescs.array);
152 0 : const RmaMemDesc *remoteRmaMemDesc = reinterpret_cast<const RmaMemDesc *>(remoteMemDescs.array[i].desc);
153 0 : CHK_PTR_NULL(remoteRmaMemDesc);
154 0 : RankId tempRankId = remoteRmaMemDesc->remoteRankId;
155 0 : HCCL_INFO("[TransportMem][ExchangeMemDesc]tempRankId:%u, userRank:%u", tempRankId, comm_->GetMyRank());
156 0 : if (tempRankId != comm_->GetMyRank()) {
157 0 : HCCL_ERROR("[TransportMem][ExchangeMemDesc]localRank[%u] receive remoteMemDesc from wrong localRank[%u], "
158 : "connection is for localRank[%u]",
159 : comm_->GetMyRank(), tempRankId, comm_->GetMyRank());
160 0 : return HCCL_E_INTERNAL;
161 : }
162 : }
163 :
164 1 : return HCCL_SUCCESS;
165 : }
166 :
167 2 : HcclResult HcclOneSidedConn::EnableMemAccess(const HcclMemDesc &remoteMemDesc, HcclMem &remoteMem)
168 : {
169 6 : HCCL_INFO("[HcclOneSidedConn]EnableMemAccess start");
170 : // 反序列化remoteMemDesc
171 2 : const RmaMemDesc *remoteRmaMemDesc = reinterpret_cast<const RmaMemDesc *>(remoteMemDesc.desc);
172 2 : std::vector<char> tempDesc(TRANSPORT_EMD_ESC_SIZE);
173 2 : tempDesc.assign(remoteRmaMemDesc->memDesc, remoteRmaMemDesc->memDesc + TRANSPORT_EMD_ESC_SIZE);
174 2 : ExchangeUbBufferDto dto;
175 2 : BinaryStream remoteRdmaRmaBufferStream(tempDesc);
176 2 : dto.Deserialize(remoteRdmaRmaBufferStream);
177 :
178 : // 导入内存描述符
179 2 : shared_ptr<HcclBuf> outBuf = make_shared<HcclBuf>();
180 2 : string tempStr = string(remoteRmaMemDesc->memDesc, TRANSPORT_EMD_ESC_SIZE);
181 2 : auto iter = desc2HcclBufMapRemoteUb_.find(tempStr);
182 2 : if (iter != desc2HcclBufMapRemoteUb_.end()) {
183 0 : outBuf = iter->second;
184 : } else {
185 2 : HcclNetDevInfos info;
186 2 : info.addr.protoType = HcclNetDevice::ConvertHcclProtoToLinkProto(linkData_.GetLocalPort().GetProto());
187 2 : info.addr.type = HCCL_ADDR_TYPE_IP_V4;
188 2 : info.netdevDeployment = HcclNetDevice::ConvertDeploymentType(linkData_.GetLocalPort().GetType());
189 2 : info.devicePhyId = comm_->GetDevicePhyId();
190 2 : info.addr.addr = linkData_.GetLocalPort().GetAddr().GetBinaryAddress().addr;
191 : HcclNetDev netDev;
192 2 : HcclResult ret = HcclNetDevOpen(&info, &netDev);
193 2 : if (ret != HCCL_SUCCESS) {
194 3 : HCCL_ERROR("[HcclOneSidedConn][EnableMemAccess]HcclNetDevOpen failed, ret[%d]", ret);
195 1 : return ret;
196 : }
197 1 : desc2netDevMap_.emplace(tempStr, netDev);
198 1 : ret = HcclMemImport(remoteRmaMemDesc->memDesc, TRANSPORT_EMD_ESC_SIZE, true, outBuf.get(), netDev);
199 1 : if (ret != HCCL_SUCCESS) {
200 0 : HCCL_ERROR("[HcclOneSidedConn][EnableMemAccess]EnableMemAccess failed, ret [%d]", ret);
201 0 : return ret;
202 : }
203 : }
204 :
205 : // 填充remoteMem
206 1 : remoteMem.type = static_cast<HcclMemType>(dto.memType);
207 1 : remoteMem.addr = outBuf->addr;
208 1 : remoteMem.size = outBuf->len;
209 :
210 : // 添加计数器
211 1 : BufferKey<uintptr_t, u64> tempKey(reinterpret_cast<uintptr_t>(outBuf->addr), outBuf->len);
212 1 : auto resultPair = remoteHcclBufMgr_.Add(tempKey, outBuf);
213 1 : if (resultPair.first == remoteHcclBufMgr_.End()) {
214 0 : HCCL_ERROR("[HcclOneSidedConn][EnableMemAccess]The memory overlaps with the memory has been enabled");
215 0 : return HCCL_E_INTERNAL;
216 : }
217 :
218 : // 存储remoteHcclBuf
219 1 : desc2HcclBufMapRemoteUb_.emplace(tempStr, outBuf);
220 3 : HCCL_INFO("[HcclOneSidedConn][EnableMemAccess] Enable memory access success.");
221 1 : return HCCL_SUCCESS;
222 2 : }
223 :
224 3 : HcclResult HcclOneSidedConn::DisableMemAccess(const HcclMemDesc &remoteMemDesc)
225 : {
226 9 : HCCL_INFO("[HcclOneSidedConn]DisableMemAccess start");
227 : // 将HcclMemDesc转化为RmaMemDesc
228 3 : const RmaMemDesc *remoteRmaMemDesc = reinterpret_cast<const RmaMemDesc *>(remoteMemDesc.desc);
229 3 : string tempStr = string(remoteRmaMemDesc->memDesc, TRANSPORT_EMD_ESC_SIZE);
230 3 : auto it = desc2HcclBufMapRemoteUb_.find(tempStr);
231 3 : if (it == desc2HcclBufMapRemoteUb_.end()) {
232 6 : HCCL_ERROR("[HcclOneSidedConn][DisableMemAccess]Can't find hcclmem by key.");
233 2 : return HCCL_E_INTERNAL;
234 : }
235 :
236 : // 计数器删除HcclBuf
237 1 : HcclBuf *buf = it->second.get();
238 1 : BufferKey<uintptr_t, u64> tempKey(reinterpret_cast<uintptr_t>(it->second->addr), it->second->len);
239 : // 删除成功：输入key是表中某一最相近key的全集，计数-1后为0，返回true
240 : // 删除失败：输入key是表中某一最相近key的全集，计数-1后不为0（说明存在其他remoteRank使用），返回false
241 1 : auto resultPair = remoteHcclBufMgr_.Del(tempKey);
242 1 : if (resultPair) {
243 1 : HcclResult ret = HcclMemClose(buf);
244 1 : if (ret != HCCL_SUCCESS) {
245 0 : HCCL_ERROR("[HcclOneSidedConn][DisableMemAccess]Close remote memory failed. ret[%d]", ret);
246 0 : return HCCL_E_INTERNAL;
247 : }
248 3 : desc2HcclBufMapRemoteUb_.erase(remoteRmaMemDesc->memDesc);
249 : }
250 3 : HCCL_INFO("[HcclOneSidedConn][DisableMemAccess] Disable memory access success.");
251 1 : return HCCL_SUCCESS;
252 3 : }
253 :
254 2 : HcclResult HcclOneSidedConn::BatchBufferSlice(const HcclOneSideOpDesc *oneSideDescs, u32 descNum,
255 : vector<HcclAicpuLocBufLite> &hostBatchPutGetLocalBufferSliceBufs,
256 : vector<HcclAicpuLocBufLite> &hostBatchPutGetRemoteBufferSliceBufs)
257 : {
258 4 : RmaBufferSlice localRmaBufferSlice[descNum] = {};
259 4 : RmtRmaBufferSlice remoteRmaBufferSlice[descNum] = {};
260 :
261 2 : if (transportMemPtr_ != nullptr) {
262 2 : CHK_RET(transportMemPtr_->BatchBufferSlice(oneSideDescs, descNum, localRmaBufferSlice, remoteRmaBufferSlice));
263 : } else {
264 0 : THROW<InternalException>("transportMemPtr is nullptr");
265 : }
266 :
267 4 : for (u32 i = 0; i < descNum; i++) {
268 2 : hostBatchPutGetLocalBufferSliceBufs[i].addr = localRmaBufferSlice[i].addr;
269 2 : hostBatchPutGetLocalBufferSliceBufs[i].size = localRmaBufferSlice[i].size;
270 2 : hostBatchPutGetLocalBufferSliceBufs[i].tokenId
271 2 : = static_cast<LocalUbRmaBuffer *>(localRmaBufferSlice[i].buf)->GetTokenId();
272 2 : hostBatchPutGetLocalBufferSliceBufs[i].tokenValue
273 2 : = static_cast<LocalUbRmaBuffer *>(localRmaBufferSlice[i].buf)->GetTokenValue();
274 6 : HCCL_INFO("hostBatchPutGetLocalBufferSliceBufs, addr=0x%llx, size=0x%llx",
275 : hostBatchPutGetLocalBufferSliceBufs[i].addr, hostBatchPutGetLocalBufferSliceBufs[i].size);
276 :
277 2 : hostBatchPutGetRemoteBufferSliceBufs[i].addr = remoteRmaBufferSlice[i].addr;
278 2 : hostBatchPutGetRemoteBufferSliceBufs[i].size = remoteRmaBufferSlice[i].size;
279 2 : hostBatchPutGetRemoteBufferSliceBufs[i].tokenId
280 2 : = static_cast<RemoteUbRmaBuffer *>(remoteRmaBufferSlice[i].buf)->GetTokenId();
281 2 : hostBatchPutGetRemoteBufferSliceBufs[i].tokenValue
282 2 : = static_cast<RemoteUbRmaBuffer *>(remoteRmaBufferSlice[i].buf)->GetTokenValue();
283 6 : HCCL_INFO("hostBatchPutGetRemoteBufferSliceBufs, addr=0x%llx, size=0x%llx",
284 : hostBatchPutGetRemoteBufferSliceBufs[i].addr, hostBatchPutGetRemoteBufferSliceBufs[i].size);
285 : }
286 2 : return HCCL_SUCCESS;
287 2 : }
288 : } // namespace Hccl
|
