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 "zero_copy_memory_agent.h"
12 : #include <string>
13 : #include "acl/acl_rt.h"
14 : #include "hccl_network_pub.h"
15 : #include "adapter_hccp_common.h"
16 : #include "adapter_rts_common.h"
17 : #include "snapshot_control.h"
18 :
19 : namespace hccl {
20 : using namespace std;
21 :
22 : const string STR_IPC_MEM_EXCHANGE = "IpcMemExchange";
23 : constexpr u32 IPC_MEMORY_EXCHANGE_LENGTH = 64; // Bytes
24 : constexpr u32 USLEEP_ONE_THOUSAND = 1000;
25 : constexpr int INNER_THREAD_LOOP_US = 500;
26 :
27 : std::unique_ptr<ZeroCopyAddressMgr> ZeroCopyMemoryAgent::addressMgr_ = nullptr;
28 :
29 : template <typename T>
30 0 : HcclResult ConstructData(u8*& exchangeDataPtr, u32& exchangeDataBlankSize, T& value)
31 : {
32 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &value, sizeof(T)));
33 0 : exchangeDataPtr += sizeof(T);
34 0 : exchangeDataBlankSize -= sizeof(T);
35 0 : return HCCL_SUCCESS;
36 : }
37 :
38 : /* copy 变长数据 */
39 0 : HcclResult ConstructData(u8*& exchangeDataPtr, u32& exchangeDataBlankSize, void* ptr, size_t len)
40 : {
41 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, ptr, len));
42 0 : exchangeDataPtr += len;
43 0 : exchangeDataBlankSize -= len;
44 0 : return HCCL_SUCCESS;
45 : }
46 :
47 : template <typename T>
48 0 : HcclResult ParseData(u8*& exchangeDataPtr, u32& exchangeDataBlankSize, T& value)
49 : {
50 0 : CHK_PRT_RET(
51 : exchangeDataBlankSize < sizeof(T),
52 : HCCL_ERROR("[ParseData] blankSize is [%u] less than [%lu]", exchangeDataBlankSize, sizeof(T)), HCCL_E_INTERNAL);
53 :
54 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&value, sizeof(T), exchangeDataPtr, sizeof(T)));
55 0 : exchangeDataPtr += sizeof(T);
56 0 : exchangeDataBlankSize -= sizeof(T);
57 0 : return HCCL_SUCCESS;
58 : }
59 :
60 0 : ZeroCopyMemoryAgent::ZeroCopyMemoryAgent(
61 : const std::unique_ptr<HcclSocketManager>& socketManager, u32 devicePhyId, s32 deviceLogicId,
62 : const HcclIpAddress& localVnicIp, const std::vector<RankInfo>& rankInfoList, RankId userRank, bool useSuperPodMode,
63 0 : const std::string& identifier)
64 0 : : initiated_(false),
65 0 : socketManager_(socketManager),
66 0 : devicePhyId_(devicePhyId),
67 0 : deviceLogicId_(deviceLogicId),
68 0 : localVnicIp_(localVnicIp),
69 0 : rankInfoList_(rankInfoList),
70 0 : userRank_(userRank),
71 0 : rankSize_(rankInfoList.size()),
72 0 : useSuperPodMode_(useSuperPodMode),
73 0 : identifier_(identifier)
74 0 : {}
75 :
76 : // 创建vnic socket连接，启动recv 接收线程
77 : // 每个rank 都启动listen，并且都和对端connect
78 0 : HcclResult ZeroCopyMemoryAgent::Init()
79 : {
80 0 : isSingleRank_ = (rankInfoList_.size() == 1);
81 0 : CHK_PRT_RET(isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][Init] single rank communicator"), HCCL_SUCCESS);
82 0 : std::unique_lock<std::mutex> lock(commRefCntLock_);
83 :
84 0 : if (!ZeroCopyMemoryAgent::IsAddressMgrInited()) {
85 0 : addressMgr_ = std::make_unique<ZeroCopyAddressMgr>();
86 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][%s]init addressMgr_ success.", __func__);
87 : }
88 0 : CHK_RET(addressMgr_->IncreCommRefCnt());
89 :
90 0 : CHK_RET(EstablishSockets());
91 :
92 0 : exchangeDataForSend_.resize(IPC_MEMORY_EXCHANGE_LENGTH * ZERO_COPY_MEMORY_AGENT_SEND_QUEUE_SIZE, 0);
93 0 : for (const auto& kv : mapDevPhyIdconnectedSockets_) {
94 0 : exchangeDataForAck_[kv.first].resize(IPC_MEMORY_EXCHANGE_LENGTH, 0);
95 0 : sendMgrs_[kv.first].reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH;
96 0 : recvMgrs_[kv.first].receivedData_.resize(
97 0 : ZERO_COPY_MEMORY_AGENT_RECV_QUEUE_SIZE, std::vector<u8>(IPC_MEMORY_EXCHANGE_LENGTH, 0));
98 : }
99 :
100 0 : CHK_RET(InitInnerThread());
101 :
102 0 : return HCCL_SUCCESS;
103 0 : }
104 :
105 0 : HcclResult ZeroCopyMemoryAgent::InitInnerThread()
106 : {
107 0 : threadRun_ = true;
108 0 : innerThread_.reset(new (std::nothrow) std::thread(&ZeroCopyMemoryAgent::InnerThread, std::ref(*this)));
109 0 : CHK_SMART_PTR_NULL(innerThread_);
110 0 : return HCCL_SUCCESS;
111 : }
112 :
113 0 : HcclResult ZeroCopyMemoryAgent::EstablishSockets()
114 : {
115 0 : CHK_PRT_RET((vnicPortCtx_ != nullptr), HCCL_ERROR("[ZeroCopyMemoryAgent][Init] already initd"), HCCL_E_PARA);
116 0 : CHK_RET(HcclNetOpenDev(&vnicPortCtx_, NicType::VNIC_TYPE, devicePhyId_, deviceLogicId_, localVnicIp_));
117 0 : CHK_PTR_NULL(vnicPortCtx_);
118 :
119 0 : isSocketSupportAsync_ = HcclSocket::IsSupportAsync();
120 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][Init] isSocketSupportAsync[%d]", isSocketSupportAsync_);
121 :
122 0 : for (size_t i = 0; i < rankInfoList_.size(); i++) {
123 0 : if (rankInfoList_[i].devicePhyId == static_cast<s32>(devicePhyId_)) {
124 0 : continue;
125 : }
126 0 : HcclRankLinkInfo remoteLinkInfo;
127 0 : RankInfo dstRankInfo = rankInfoList_[i];
128 0 : remoteLinkInfo.userRank = dstRankInfo.userRank;
129 0 : remoteLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
130 0 : remoteLinkInfo.ip = HcclIpAddress(dstRankInfo.devicePhyId);
131 0 : if (useSuperPodMode_) {
132 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
133 : devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_SDID, dstRankInfo.superDeviceId, remoteLinkInfo.ip));
134 : } else {
135 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
136 : devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, dstRankInfo.devicePhyId, remoteLinkInfo.ip));
137 : }
138 : // 通信域未分配端口则使用默认端口
139 : remoteLinkInfo.port
140 0 : = dstRankInfo.deviceVnicPort == HCCL_INVALID_PORT ? HETEROG_CCL_PORT : dstRankInfo.deviceVnicPort;
141 0 : remoteLinkInfo.socketsPerLink = 1;
142 0 : string newTag = GenerateSocketTag(devicePhyId_, rankInfoList_[i].devicePhyId);
143 0 : std::vector<std::shared_ptr<HcclSocket>> tmpSockets;
144 : HcclResult ret
145 0 : = socketManager_->CreateSingleLinkSocket(newTag, vnicPortCtx_, remoteLinkInfo, tmpSockets, false, true);
146 0 : CHK_PRT_RET(
147 : ret != HCCL_SUCCESS,
148 : HCCL_ERROR(
149 : "[Create][DestSockets]Create single link sockets failed, "
150 : "local rank[%u], remote rank[%u]",
151 : userRank_, i),
152 : ret);
153 0 : if (tmpSockets.size() != 1) {
154 0 : HCCL_ERROR(
155 : "[ZeroCopyMemoryAgent][CreateVnic] socket number[%llu] is not 1 as expected!", tmpSockets.size());
156 0 : return HCCL_E_INTERNAL;
157 : }
158 : // 设置强制断链为关闭，避免进程退出时recv失败
159 0 : tmpSockets[0]->SetForceClose(false);
160 0 : mapDevPhyIdconnectedSockets_[remoteLinkInfo.devicePhyId] = (tmpSockets[0]);
161 0 : mapDevPhyId2RankId_[remoteLinkInfo.devicePhyId] = remoteLinkInfo.userRank;
162 0 : }
163 :
164 0 : for (const auto& kv : mapDevPhyIdconnectedSockets_) {
165 0 : CHK_PRT_RET(
166 : socketManager_->WaitLinkEstablish(kv.second) != HCCL_SUCCESS,
167 : HCCL_ERROR(
168 : "[ZeroCopyMemoryAgent][EstablishSockets] tag[%s] socket establish failed", kv.second->GetTag().c_str()),
169 : HCCL_E_INTERNAL);
170 : }
171 0 : return HCCL_SUCCESS;
172 : }
173 :
174 0 : std::string ZeroCopyMemoryAgent::GenerateSocketTag(u32 localRank, u32 remoteRank)
175 : {
176 0 : u32 small = localRank;
177 0 : u32 large = remoteRank;
178 :
179 0 : if (localRank > remoteRank) {
180 0 : small = remoteRank;
181 0 : large = localRank;
182 : }
183 :
184 : // Socket构造规则：前缀 + identifier + small + large
185 : std::string tag
186 0 : = STR_IPC_MEM_EXCHANGE + "_" + identifier_ + "_" + std::to_string(small) + ":" + std::to_string(large);
187 0 : return tag;
188 : }
189 :
190 0 : HcclResult ZeroCopyMemoryAgent::SendRequestSync(RequestType requestType, const std::vector<u8>& req, u32 remoteDevPhyId)
191 : {
192 : HcclResult ret;
193 0 : if (remoteDevPhyId != INVALID_VALUE_RANKID) {
194 0 : std::unique_lock<std::mutex> lock(sendMutex_); // send 存在多线调用，需要锁保护
195 0 : ret = mapDevPhyIdconnectedSockets_[remoteDevPhyId]->Send(req.data(), IPC_MEMORY_EXCHANGE_LENGTH);
196 0 : CHK_PRT_RET(
197 : ret != HCCL_SUCCESS,
198 : HCCL_ERROR(
199 : "[ZeroCopyMemoryAgent][SendRequestSync] Send %s to remote[%u] failed",
200 : GetReadableRequestType(requestType), remoteDevPhyId),
201 : HCCL_E_INTERNAL);
202 0 : return HCCL_SUCCESS;
203 0 : }
204 :
205 0 : std::unique_lock<std::mutex> lock(sendMutex_);
206 0 : for (const auto& kv : mapDevPhyIdconnectedSockets_) {
207 0 : CHK_PRT_RET(
208 : kv.second->Send(req.data(), IPC_MEMORY_EXCHANGE_LENGTH) != HCCL_SUCCESS,
209 : HCCL_ERROR(
210 : "[ZeroCopyMemoryAgent][SendRequestSync] Send %s to remote[%u] failed",
211 : GetReadableRequestType(requestType), kv.first),
212 : HCCL_E_INTERNAL);
213 : }
214 0 : return HCCL_SUCCESS;
215 0 : }
216 :
217 0 : HcclResult ZeroCopyMemoryAgent::SendRequest(RequestType requestType, const std::vector<u8>& req, u32 remoteDevPhyId)
218 : {
219 0 : HCCL_INFO(
220 : "[ZeroCopyMemoryAgent][SendRequest] requestType[%s] remote[%u]", GetReadableRequestType(requestType),
221 : remoteDevPhyId);
222 :
223 0 : if (!isSocketSupportAsync_) { // socket不支持异步收发的场景
224 0 : return SendRequestSync(requestType, req, remoteDevPhyId);
225 : }
226 :
227 0 : bool isAck = IsAckRequestType(requestType);
228 0 : if (remoteDevPhyId != INVALID_VALUE_RANKID) {
229 0 : sendMgrs_[remoteDevPhyId].AddRequest(isAck, req);
230 : } else {
231 0 : for (auto& kv : sendMgrs_) {
232 0 : kv.second.AddRequest(isAck, req);
233 : }
234 : }
235 :
236 : // 唤醒内部io线程
237 0 : std::unique_lock<std::mutex> lock(sendMutex_);
238 0 : hasSendRequest_ = true;
239 0 : sendCv_.notify_all();
240 0 : return HCCL_SUCCESS;
241 0 : }
242 :
243 0 : void ZeroCopyMemoryAgent::RequestBatchSendAsync()
244 : {
245 : HcclResult ret;
246 0 : for (auto& kv : sendMgrs_) {
247 0 : auto& sendMgr = kv.second;
248 0 : if ((sendMgr.lastSendHandle_ != nullptr) || (!sendMgr.hasReq_[0] && !sendMgr.hasReq_[1])) {
249 : // 前回发送未完成 或者 没有待发送的数据
250 0 : continue;
251 : }
252 :
253 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
254 0 : HCCL_ERROR(
255 : "[ZeroCopyMemoryAgent][RequestBatchSendAsync] remote[%u] not found in"
256 : "mapDevPhyIdconnectedSockets_",
257 : kv.first);
258 0 : continue;
259 : }
260 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
261 0 : if (sendMgr.sentSize_ == 0) { // 非断点续传
262 0 : if (sendMgr.hasReq_[0] && sendMgr.hasReq_[1]) { // 合并发送
263 0 : u8* ptr = const_cast<u8*>(sendMgr.reqDatas_[1]->data()) + IPC_MEMORY_EXCHANGE_LENGTH;
264 0 : u32 leftSize = IPC_MEMORY_EXCHANGE_LENGTH;
265 0 : if (ConstructData(
266 0 : ptr, leftSize, const_cast<u8*>(sendMgr.reqDatas_[0]->data()), IPC_MEMORY_EXCHANGE_LENGTH)
267 0 : == HCCL_SUCCESS) {
268 0 : sendMgr.hasReq_[0] = false;
269 0 : sendMgr.currIndex_ = 1;
270 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH + IPC_MEMORY_EXCHANGE_LENGTH;
271 : } else {
272 0 : sendMgr.currIndex_ = 0;
273 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH;
274 : }
275 : } else {
276 0 : sendMgr.currIndex_ = sendMgr.hasReq_[0] ? 0 : 1;
277 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH;
278 : }
279 : }
280 0 : const std::vector<u8>* req = sendMgr.reqDatas_[sendMgr.currIndex_];
281 0 : sendMgr.lastSendSize_ = 0; // 用于ra上报发送的数据量
282 0 : ret = socket->SendAsync(
283 0 : req->data() + sendMgr.sentSize_, sendMgr.reqDataSize_ - sendMgr.sentSize_, &sendMgr.lastSendSize_,
284 : &sendMgr.lastSendHandle_);
285 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) { // 发送失败的场景
286 0 : RequestType requestType = *reinterpret_cast<const RequestType*>(req->data());
287 0 : HCCL_ERROR(
288 : "[ZeroCopyMemoryAgent][RequestBatchSendAsync] failed, ret[%d] remote[%u] requestType[%s] "
289 : "sentSize[%llu]",
290 : ret, kv.first, GetReadableRequestType(requestType), sendMgr.sentSize_);
291 : }
292 : }
293 0 : }
294 :
295 0 : void ZeroCopyMemoryAgent::CheckBatchSendAsyncResult()
296 : {
297 : HcclResult ret;
298 : HcclResult lastSendRet;
299 0 : for (auto& kv : sendMgrs_) {
300 0 : auto& sendMgr = kv.second;
301 0 : if (sendMgr.lastSendHandle_ == nullptr) { // 没有正在执行的异步send
302 0 : continue;
303 : }
304 :
305 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
306 0 : HCCL_ERROR(
307 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult] remote[%u] not found in"
308 : "mapDevPhyIdconnectedSockets_",
309 : kv.first);
310 0 : continue;
311 : }
312 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
313 0 : ret = socket->GetAsyncReqResult(sendMgr.lastSendHandle_, lastSendRet);
314 0 : if (ret != HCCL_SUCCESS) {
315 0 : CHK_PRT_CONT(
316 : ret != HCCL_E_AGAIN,
317 : HCCL_ERROR(
318 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]GetAsyncReqResult failed, ret[%d] remote[%u]", ret,
319 : kv.first));
320 0 : continue;
321 : }
322 :
323 0 : sendMgr.lastSendHandle_ = nullptr;
324 0 : if ((lastSendRet != HCCL_SUCCESS) && (sendMgr.lastSendSize_ == 0)) {
325 0 : CHK_PRT_CONT(
326 : lastSendRet != HCCL_E_AGAIN, HCCL_ERROR(
327 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]SendAsync failed, "
328 : "result[%d] remote[%u] sentSize[%llu]",
329 : lastSendRet, kv.first, sendMgr.sentSize_));
330 0 : continue;
331 : }
332 :
333 0 : sendMgr.sentSize_ += sendMgr.lastSendSize_; // 下次从中断的地方开始重发
334 0 : if (sendMgr.sentSize_ == sendMgr.reqDataSize_) { // request发送完成
335 0 : sendMgr.sentSize_ = 0;
336 0 : sendMgr.hasReq_[sendMgr.currIndex_] = false;
337 0 : HCCL_DEBUG(
338 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]SendAsync success, requestType[%s] remote[%u]",
339 : GetReadableRequestType(
340 : *reinterpret_cast<const RequestType*>(sendMgr.reqDatas_[sendMgr.currIndex_]->data())),
341 : kv.first);
342 : }
343 : }
344 0 : }
345 :
346 0 : void ZeroCopyMemoryAgent::RequestBatchRecvAsync()
347 : {
348 : HcclResult ret;
349 0 : for (auto& kv : recvMgrs_) {
350 0 : auto& recvMgr = kv.second;
351 0 : if ((recvMgr.lastRecvHandle_ != nullptr) || // 前回接收未完成
352 0 : ((receivedBarrierClose_.count(kv.first) != 0) && (receivedBarrierCloseAck_.count(kv.first) != 0))) {
353 : // 该socket已经收到BarrierClose与BarrierCloseAck报文，因此不允许再进行其他数据接收了
354 0 : continue;
355 : }
356 :
357 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
358 0 : HCCL_ERROR(
359 : "[ZeroCopyMemoryAgent][RequestBatchRecvAsync] remote[%u] not found in"
360 : "mapDevPhyIdconnectedSockets_",
361 : kv.first);
362 0 : continue;
363 : }
364 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
365 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.recvIndex_];
366 0 : recvMgr.lastRecvSize_ = 0; // 用于ra上报接收的数据量
367 0 : ret = socket->RecvAsync(
368 0 : req.data() + recvMgr.receivedSize_, IPC_MEMORY_EXCHANGE_LENGTH - recvMgr.receivedSize_,
369 : &recvMgr.lastRecvSize_, &recvMgr.lastRecvHandle_);
370 0 : CHK_PRT_CONT(
371 : (ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN),
372 : HCCL_ERROR(
373 : "[ZeroCopyMemoryAgent][RequestBatchRecvAsync] RecvAsync failed, ret[%d] remote[%u] receivedSize[%llu]",
374 : ret, kv.first, recvMgr.receivedSize_));
375 : }
376 0 : }
377 :
378 0 : void ZeroCopyMemoryAgent::CheckBatchRecvAsyncResult()
379 : {
380 : HcclResult ret;
381 : HcclResult lastRecvRet;
382 0 : for (auto& kv : recvMgrs_) {
383 0 : auto& recvMgr = kv.second;
384 0 : if (recvMgr.lastRecvHandle_ == nullptr) { // 没有正在异步接收
385 0 : continue;
386 : }
387 :
388 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
389 0 : HCCL_ERROR(
390 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] remote[%u] not found in"
391 : "mapDevPhyIdconnectedSockets_",
392 : kv.first);
393 0 : continue;
394 : }
395 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
396 0 : ret = socket->GetAsyncReqResult(recvMgr.lastRecvHandle_, lastRecvRet);
397 0 : if (ret != HCCL_SUCCESS) {
398 0 : CHK_PRT_CONT(
399 : ret != HCCL_E_AGAIN,
400 : HCCL_ERROR(
401 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] GetAsyncReqResult failed, ret[%d] remote[%u]",
402 : ret, kv.first));
403 0 : continue;
404 : }
405 :
406 0 : recvMgr.lastRecvHandle_ = nullptr;
407 0 : if ((lastRecvRet != HCCL_SUCCESS) && (recvMgr.lastRecvSize_ == 0)) {
408 0 : CHK_PRT_CONT(
409 : lastRecvRet != HCCL_E_AGAIN, HCCL_WARNING(
410 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] RecvAsync failed, "
411 : "result[%d] remote[%u] lastRecvSize[%llu]",
412 : lastRecvRet, kv.first, recvMgr.lastRecvSize_));
413 0 : continue;
414 : }
415 :
416 0 : recvMgr.receivedSize_ += recvMgr.lastRecvSize_;
417 0 : if (recvMgr.receivedSize_ == IPC_MEMORY_EXCHANGE_LENGTH) {
418 0 : recvMgr.receivedSize_ = 0;
419 0 : RecvRequest(recvMgr, kv.first);
420 0 : ioRecvWaiting_ = true; // 后面高概率还有数据要收（ack与request合并场景），loop不等待
421 : } else {
422 : // request没收全，loop不等待
423 0 : ioRecvWaiting_ = (recvMgr.receivedSize_ > 0);
424 : }
425 : }
426 0 : }
427 :
428 0 : inline void ZeroCopyMemoryAgent::RecvRequest(ZeroCopyMemoryAgentRecvMgr& recvMgr, u32 remoteDevicePhyId)
429 : {
430 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.recvIndex_];
431 0 : RequestType requestType = *reinterpret_cast<RequestType*>(req.data());
432 0 : HCCL_DEBUG(
433 : "[ZeroCopyMemoryAgent][RecvRequest] recv requestType[%s] remote[%u]", GetReadableRequestType(requestType),
434 : remoteDevicePhyId);
435 :
436 0 : if (IsAckRequestType(requestType)) { // 收到ACK时，直接优先处理
437 0 : u32 remoteRank = mapDevPhyId2RankId_[remoteDevicePhyId];
438 0 : CHK_PRT_CONT(
439 : ParseReceivedRequest(req, remoteRank) != HCCL_SUCCESS,
440 : HCCL_ERROR(
441 : "[ZeroCopyMemoryAgent][ParseReceivedRequest] failed requestType[%s] remote[%u]",
442 : GetReadableRequestType(requestType), remoteDevicePhyId));
443 0 : return;
444 : }
445 :
446 0 : recvMgr.recvIndex_ = (recvMgr.recvIndex_ + 1) % ZERO_COPY_MEMORY_AGENT_RECV_QUEUE_SIZE; // 准备下一次接收
447 0 : hasReceivedRequest_ = true;
448 : }
449 :
450 0 : void ZeroCopyMemoryAgent::ParseReceivedRequests()
451 : {
452 0 : if (!hasReceivedRequest_) {
453 0 : return;
454 : }
455 0 : hasReceivedRequest_ = false;
456 :
457 0 : for (auto& kv : recvMgrs_) {
458 0 : u32 remoteRank = mapDevPhyId2RankId_[kv.first];
459 0 : auto& recvMgr = kv.second;
460 0 : while (recvMgr.praseIndex_ != recvMgr.recvIndex_) {
461 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.praseIndex_];
462 0 : CHK_PRT_CONT(
463 : ParseReceivedRequest(req, remoteRank) != HCCL_SUCCESS,
464 : HCCL_ERROR(
465 : "[ZeroCopyMemoryAgent][ParseReceivedRequest] failed prase requestType[%s] remote[%u]",
466 : GetReadableRequestType(*reinterpret_cast<RequestType*>(req.data())), kv.first));
467 0 : recvMgr.praseIndex_++;
468 0 : if (recvMgr.praseIndex_ == ZERO_COPY_MEMORY_AGENT_RECV_QUEUE_SIZE) {
469 0 : recvMgr.praseIndex_ = 0;
470 : }
471 : }
472 : }
473 : }
474 :
475 0 : void ZeroCopyMemoryAgent::RequestBatchRecvSync()
476 : {
477 : HcclResult ret;
478 0 : for (auto& kv : recvMgrs_) {
479 0 : auto& recvMgr = kv.second;
480 0 : if ((receivedBarrierClose_.count(kv.first) != 0) && (receivedBarrierCloseAck_.count(kv.first) != 0)) {
481 : // 该socket已经收到BarrierClose与BarrierCloseAck报文，因此不允许再进行其他数据接收了
482 0 : continue;
483 : }
484 :
485 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
486 0 : HCCL_ERROR(
487 : "[ZeroCopyMemoryAgent][RequestBatchRecvSync] remote[%u] not found in"
488 : "mapDevPhyIdconnectedSockets_",
489 : kv.first);
490 0 : continue;
491 : }
492 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
493 0 : std::vector<u8>& req = recvMgr.receivedData_[0];
494 0 : recvMgr.lastRecvSize_ = 0;
495 0 : ret = socket->IRecv(
496 0 : req.data() + recvMgr.receivedSize_, IPC_MEMORY_EXCHANGE_LENGTH - recvMgr.receivedSize_,
497 0 : recvMgr.lastRecvSize_);
498 0 : CHK_PRT_CONT(
499 : (ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN),
500 : HCCL_ERROR(
501 : "[ZeroCopyMemoryAgent][RequestBatchRecvSync] IRecv failed, ret[%d] remote[%u] receivedSize[%llu]", ret,
502 : kv.first, recvMgr.receivedSize_));
503 :
504 0 : recvMgr.receivedSize_ += recvMgr.lastRecvSize_;
505 0 : if (recvMgr.receivedSize_ == IPC_MEMORY_EXCHANGE_LENGTH) {
506 0 : recvMgr.receivedSize_ = 0;
507 0 : ret = ParseReceivedRequest(req, mapDevPhyId2RankId_[kv.first]);
508 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS, HCCL_ERROR("[ZeroCopyMemoryAgent][ParseReceivedRequest] failed"));
509 : }
510 : }
511 0 : }
512 :
513 0 : void ZeroCopyMemoryAgent::InnerThread()
514 : {
515 : // 新线程，更新一下使用的设备
516 0 : if (hrtSetDevice(deviceLogicId_) != HCCL_SUCCESS) {
517 0 : HCCL_ERROR("[ZeroCopyMemoryAgent][InnerThread] set device failed");
518 0 : return;
519 : }
520 :
521 0 : while (threadRun_) {
522 0 : CheckSnapshotStatus();
523 0 : if (isPaused_) {
524 0 : SaluSleep(USLEEP_ONE_THOUSAND);
525 0 : continue;
526 : }
527 :
528 0 : if (isSocketSupportAsync_) {
529 0 : CheckBatchSendAsyncResult();
530 0 : RequestBatchSendAsync();
531 :
532 0 : CheckBatchRecvAsyncResult();
533 0 : RequestBatchRecvAsync();
534 :
535 0 : ParseReceivedRequests();
536 :
537 0 : std::unique_lock<std::mutex> lock(sendMutex_);
538 0 : if (!ioRecvWaiting_ && !hasSendRequest_) {
539 0 : sendCv_.wait_for(lock, std::chrono::microseconds(INNER_THREAD_LOOP_US));
540 : }
541 0 : hasSendRequest_ = false;
542 0 : ioRecvWaiting_ = false;
543 0 : } else {
544 0 : RequestBatchRecvSync();
545 0 : SaluSleep(USLEEP_ONE_THOUSAND);
546 : }
547 : }
548 :
549 0 : if (hrtResetDevice(deviceLogicId_) != HCCL_SUCCESS) {
550 0 : HCCL_ERROR("[ZeroCopyMemoryAgent][InnerThread] reset device failed");
551 0 : return;
552 : }
553 : }
554 :
555 0 : HcclResult ZeroCopyMemoryAgent::SetRemoteTgid()
556 : {
557 0 : if (remotePids_.size() == mapDevPhyIdconnectedSockets_.size()) {
558 0 : HCCL_INFO("[ZeroCopyMemoryAgent][SetRemoteTgid] tgid exchange is ok");
559 0 : return HCCL_SUCCESS;
560 : }
561 0 : remotePids_.clear();
562 :
563 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
564 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
565 :
566 0 : RequestType requestType = RequestType::SET_REMOTE_BARE_TGID;
567 :
568 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
569 :
570 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
571 :
572 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
573 :
574 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::SET_REMOTE_BARE_TGID_ACK));
575 0 : if (remotePids_.size() != mapDevPhyIdconnectedSockets_.size()) {
576 0 : HCCL_ERROR(
577 : "[ZeroCopyMemoryAgent][SetRemoteTgid] tgid exchange failed recv pids count[%lu]", remotePids_.size());
578 0 : return HCCL_E_INTERNAL;
579 : }
580 0 : return HCCL_SUCCESS;
581 : }
582 :
583 0 : HcclResult ZeroCopyMemoryAgent::DeInit()
584 : {
585 0 : CHK_PRT_RET(isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][DeInit] single rank communicator"), HCCL_SUCCESS);
586 0 : std::unique_lock<std::mutex> lock(commRefCntLock_);
587 0 : if (!ZeroCopyMemoryAgent::IsAddressMgrInited()) {
588 0 : HCCL_ERROR(
589 : "[ZeroCopyMemoryAgent][%s]addressMgr_ is nullptr, no need to deinit. local rank[u32]", __func__, userRank_);
590 0 : return HCCL_E_INTERNAL;
591 : }
592 0 : threadRun_ = false;
593 0 : if (innerThread_) {
594 0 : if (innerThread_->joinable()) {
595 0 : innerThread_->join(); // 等待线程执行后释放资源
596 : }
597 : }
598 0 : innerThread_ = nullptr;
599 :
600 0 : if (vnicPortCtx_ != nullptr) {
601 0 : HcclNetCloseDev(vnicPortCtx_);
602 0 : vnicPortCtx_ = nullptr;
603 : }
604 0 : CHK_RET(addressMgr_->DecreCommRefCnt());
605 0 : if (addressMgr_->GetCommRefCnt() == 0) {
606 0 : addressMgr_.reset();
607 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][%s]Release addressMgr_", __func__);
608 : }
609 0 : return HCCL_SUCCESS;
610 0 : }
611 :
612 0 : HcclResult ZeroCopyMemoryAgent::SetMemoryRange(void* virPtr, size_t size, size_t alignment, uint64_t flags)
613 : {
614 0 : CHK_PRT_RET(
615 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][SetMemoryRange] single rank communicator"), HCCL_SUCCESS);
616 0 : CHK_PRT_RET(
617 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
618 : HCCL_ERROR(
619 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
620 : "is not init.",
621 : __func__),
622 : HCCL_E_INTERNAL);
623 0 : CHK_PRT_RET(
624 : addressMgr_->SetMemoryRange(devicePhyId_, virPtr, size) != HCCL_SUCCESS,
625 : HCCL_ERROR(
626 : "[ZeroCopyMemoryAgent][SetMemoryRange] invalid set ptr[%p] size[%lu] alignment[%lu] flags[%lu]", virPtr,
627 : size, alignment, flags),
628 : HCCL_E_PARA);
629 :
630 0 : HCCL_INFO(
631 : "[ZeroCopyMemoryAgent][SetMemoryRange] basePtr[%p] size[%lu] alignment[%lu] flag[%lu]", virPtr, size, alignment,
632 : flags);
633 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
634 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
635 :
636 0 : RequestType requestType = RequestType::SET_MEMORY_RANGE;
637 :
638 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
639 :
640 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
641 :
642 0 : u64 addr = reinterpret_cast<u64>(virPtr);
643 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
644 :
645 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, size));
646 :
647 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, alignment));
648 :
649 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, flags));
650 :
651 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
652 :
653 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::SET_MEMORY_RANGE_ACK));
654 0 : return HCCL_SUCCESS;
655 : }
656 :
657 0 : HcclResult ZeroCopyMemoryAgent::UnsetMemoryRange(void* virPtr)
658 : {
659 0 : CHK_PRT_RET(
660 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][UnsetMemoryRange] single rank communicator"), HCCL_SUCCESS);
661 0 : CHK_PRT_RET(
662 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
663 : HCCL_ERROR(
664 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
665 : "is not init.",
666 : __func__),
667 : HCCL_E_INTERNAL);
668 0 : CHK_PRT_RET(
669 : !addressMgr_->IsAddressSet(devicePhyId_, virPtr),
670 : HCCL_ERROR("[ZeroCopyMemoryAgent][UnsetMemoryRange] ptr[%p] is not set memory", virPtr), HCCL_E_PARA);
671 0 : CHK_RET(addressMgr_->UnsetMemoryRange(devicePhyId_, virPtr));
672 :
673 0 : HCCL_INFO("[ZeroCopyMemoryAgent][UnsetMemoryRange] basePtr[%p]", virPtr);
674 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
675 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
676 :
677 0 : RequestType requestType = RequestType::UNSET_MEMORY_RANGE;
678 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
679 :
680 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
681 :
682 0 : u64 addr = reinterpret_cast<u64>(virPtr);
683 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
684 :
685 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
686 :
687 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::UNSET_MEMORY_RANGE_ACK));
688 0 : return HCCL_SUCCESS;
689 : }
690 :
691 : HcclResult
692 0 : ZeroCopyMemoryAgent::ActivateCommMemory(void* virPtr, size_t size, size_t offset, void* memHandle, uint64_t flags)
693 : {
694 0 : CHK_PRT_RET(
695 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][ActivateCommMemory] single rank communicator"), HCCL_SUCCESS);
696 0 : CHK_PRT_RET(
697 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
698 : HCCL_ERROR(
699 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
700 : "is not init.",
701 : __func__),
702 : HCCL_E_INTERNAL);
703 0 : CHK_PRT_RET(
704 : !addressMgr_->IsInSetAddressRange(devicePhyId_, virPtr, size),
705 : HCCL_ERROR(
706 : "[ZeroCopyMemoryAgent][ActivateCommMemory] input ptr[%p] size[%lu] is not in set address range", virPtr,
707 : size),
708 : HCCL_E_PARA);
709 0 : CHK_PRT_RET(
710 : addressMgr_->IsOverlapWithActivateAddr(virPtr, size),
711 : HCCL_ERROR(
712 : "[ZeroCopyMemoryAgent][ActivateCommMemory] input ptr[%p] size[%lu] overlap with activate memory", virPtr,
713 : size),
714 : HCCL_E_PARA);
715 :
716 0 : HCCL_INFO(
717 : "[ZeroCopyMemoryAgent][ActivateCommMemory] virPtr[%p] size[%lu] offset[%lu] memHandle[%p], flags[%lu]", virPtr,
718 : size, offset, memHandle, flags);
719 0 : CHK_RET(SetRemoteTgid());
720 :
721 : uint64_t shareableHandle;
722 0 : aclrtMemHandleType handleType = ACL_MEM_HANDLE_TYPE_NONE;
723 0 : aclError ret = ACL_SUCCESS;
724 0 : ret = aclrtMemExportToShareableHandle(memHandle, handleType, 0, &shareableHandle);
725 0 : CHK_PRT_RET(
726 : ret != ACL_SUCCESS,
727 : HCCL_ERROR(
728 : "[ZeroCopyMemoryAgent][ActivateCommMemory] aclrtMemExportToShareableHandle handle[%p] type[%d] flags[%llu] "
729 : "failed, ret[%d]",
730 : memHandle, handleType, 0, ret),
731 : HCCL_E_RUNTIME);
732 0 : ret = aclrtMemSetPidToShareableHandle(shareableHandle, remotePids_.data(), remotePids_.size());
733 0 : CHK_PRT_RET(
734 : ret != ACL_SUCCESS,
735 : HCCL_ERROR(
736 : "[ZeroCopyMemoryAgent][ActivateCommMemory] aclrtMemSetPidToShareableHandle shareableHandl[%llu]",
737 : " failed, ret[%d]", shareableHandle, ret),
738 : HCCL_E_RUNTIME);
739 :
740 0 : HCCL_INFO(
741 : "[ZeroCopyMemoryAgent][ActivateCommMemory] dev[%u] export shareableHandle[%lu]", devicePhyId_, shareableHandle);
742 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
743 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
744 :
745 0 : RequestType requestType = RequestType::ACTIVATE_COMM_MEMORY;
746 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
747 :
748 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
749 :
750 0 : u64 addr = reinterpret_cast<u64>(virPtr);
751 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
752 :
753 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, size));
754 :
755 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, offset));
756 :
757 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, shareableHandle));
758 :
759 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, flags));
760 :
761 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
762 :
763 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::ACTIVATE_COMM_MEMORY_ACK));
764 0 : CHK_RET(addressMgr_->ActivateCommMemoryAddr(virPtr, size));
765 :
766 0 : return HCCL_SUCCESS;
767 : }
768 :
769 0 : HcclResult ZeroCopyMemoryAgent::DeactivateCommMemory(void* virPtr)
770 : {
771 0 : CHK_PRT_RET(
772 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][DeactivateCommMemory] single rank communicator"), HCCL_SUCCESS);
773 0 : CHK_PRT_RET(
774 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
775 : HCCL_ERROR(
776 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
777 : "is not init.",
778 : __func__),
779 : HCCL_E_INTERNAL);
780 0 : CHK_PRT_RET(
781 : !addressMgr_->IsActivateCommMemoryAddr(virPtr, 1),
782 : HCCL_ERROR("[ZeroCopyMemoryAgent][DeactivateCommMemory] input ptr[%p] is not activate", virPtr), HCCL_E_PARA);
783 :
784 0 : HCCL_INFO("[ZeroCopyMemoryAgent][DeactivateCommMemory] virPtr[%p]", virPtr);
785 0 : CHK_RET(addressMgr_->DeactivateCommMemoryAddr(virPtr));
786 :
787 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
788 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
789 :
790 0 : RequestType requestType = RequestType::DEACTIVATE_COMM_MEMORY;
791 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
792 :
793 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
794 :
795 0 : u64 addr = reinterpret_cast<u64>(virPtr);
796 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
797 :
798 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
799 :
800 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::DEACTIVATE_COMM_MEMORY_ACK));
801 0 : return HCCL_SUCCESS;
802 : }
803 :
804 0 : HcclResult ZeroCopyMemoryAgent::BarrierClose()
805 : {
806 0 : CHK_PRT_RET(isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][BarrierClose] single rank communicator"), HCCL_SUCCESS);
807 :
808 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][BarrierClose] [%s] ready to barrier close", identifier_.c_str());
809 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
810 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
811 :
812 0 : RequestType requestType = RequestType::BARRIER_CLOSE;
813 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
814 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
815 :
816 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
817 :
818 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::BARRIER_CLOSE_ACK));
819 :
820 0 : return HCCL_SUCCESS;
821 : }
822 :
823 40 : bool ZeroCopyMemoryAgent::IsActivateCommMemoryAddr(void* virPtr, u64 length)
824 : {
825 40 : if (!ZeroCopyMemoryAgent::IsAddressMgrInited()) {
826 40 : HCCL_INFO("[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent is not init.", __func__);
827 40 : return false;
828 : }
829 0 : return addressMgr_->IsActivateCommMemoryAddr(virPtr, length);
830 : }
831 :
832 0 : HcclResult ZeroCopyMemoryAgent::GetRingBufferAddr(u64& bufferPtr, u64& headPtr, u64& tailPtr)
833 : {
834 0 : CHK_PRT_RET(
835 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
836 : HCCL_ERROR(
837 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
838 : "is not init.",
839 : __func__),
840 : HCCL_E_INTERNAL);
841 0 : addressMgr_->GetRingBufferAddr(bufferPtr, headPtr, tailPtr);
842 0 : return HCCL_SUCCESS;
843 : }
844 :
845 40 : bool ZeroCopyMemoryAgent::IsAddressMgrInited() { return addressMgr_ != nullptr; }
846 :
847 0 : HcclResult ZeroCopyMemoryAgent::WaitForAllRemoteComplete(RequestType requestType)
848 : {
849 0 : bool useBarrier = NeedBarrier(requestType);
850 0 : if (useBarrier) {
851 0 : reqMsgDeliverCnt_++;
852 : }
853 :
854 0 : u32 expectedNum = mapDevPhyIdconnectedSockets_.size();
855 0 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
856 0 : std::unique_lock<std::mutex> lock(dfxMutex_);
857 0 : waitCompleteCv_.wait_for(lock, timeout);
858 0 : if ((reqMsgCounter_[static_cast<int>(requestType)] == expectedNum)
859 0 : && (!useBarrier || (useBarrier && reqMsgDeliverCnt_ <= reqMsgFinishCnt_))) {
860 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
861 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].clear();
862 0 : return HCCL_SUCCESS;
863 : }
864 :
865 0 : HCCL_ERROR(
866 : "[Wait][RemoteComplete %s] dev[%u] errNo[0x%016llx] timeout[%d s] completeCount[%u] %s",
867 : GetReadableRequestType(requestType), devicePhyId_, HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER), timeout,
868 : reqMsgCounter_[static_cast<int>(requestType)].load(), DumpFinishInfo(requestType).c_str());
869 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
870 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].clear();
871 0 : return HCCL_E_TCP_TRANSFER;
872 0 : }
873 :
874 0 : HcclResult ZeroCopyMemoryAgent::ParseSetMemoryRange(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
875 : {
876 0 : CHK_PRT_RET(
877 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
878 : HCCL_ERROR(
879 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
880 : "is not init.",
881 : __func__),
882 : HCCL_E_INTERNAL);
883 : u32 devicePhyId;
884 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
885 :
886 : u64 addr;
887 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
888 :
889 : size_t size;
890 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, size));
891 :
892 : size_t alignment;
893 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, alignment));
894 :
895 : uint64_t flags;
896 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, flags));
897 :
898 : u32 maxDeviceNum;
899 0 : CHK_RET(GetMaxDevNum(maxDeviceNum));
900 0 : CHK_PRT_RET(
901 : devicePhyId >= maxDeviceNum,
902 : HCCL_ERROR(
903 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] devicePhyId[%u] is exceed max device num[%u]", devicePhyId,
904 : maxDeviceNum),
905 : HCCL_E_PARA);
906 :
907 0 : void* remoteAddrBase = reinterpret_cast<void*>(addr);
908 0 : CHK_PRT_RET(
909 : addressMgr_->IsAddressSet(devicePhyId, remoteAddrBase),
910 : HCCL_ERROR(
911 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] devicePhyId[%u] had set addr [%p]", devicePhyId,
912 : remoteAddrBase),
913 : HCCL_E_PARA);
914 :
915 0 : void* devPtr = nullptr;
916 0 : void* devAddr = nullptr;
917 0 : aclError ret = aclrtReserveMemAddress(&devPtr, size, alignment, devAddr, flags);
918 0 : CHK_PRT_RET(
919 : ret != ACL_SUCCESS,
920 : HCCL_ERROR(
921 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] rtReserve Memory failed, "
922 : "return[%d], devPtr[%p] size[%llu] alignment[%llu] devAddr[%p] flags[%llu]",
923 : ret, devPtr, size, alignment, devAddr, flags),
924 : HCCL_E_RUNTIME);
925 :
926 0 : CHK_RET(addressMgr_->AddLocalIpc2RemoteAddr(devicePhyId, devPtr, reinterpret_cast<void*>(addr), size));
927 :
928 0 : CHK_RET(SendAckAfterParse(RequestType::SET_MEMORY_RANGE, RequestType::SET_MEMORY_RANGE_ACK, devicePhyId));
929 :
930 0 : return HCCL_SUCCESS;
931 : }
932 :
933 0 : HcclResult ZeroCopyMemoryAgent::SendAckAfterParse(
934 : RequestType requestType, RequestType ackType, u32 remoteDevicePhyId, void* extraData, u64 extraDataLen)
935 : {
936 0 : u8* exchangeDataAckPtr = exchangeDataForAck_[remoteDevicePhyId].data();
937 0 : u32 exchangeDataAckBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
938 :
939 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, ackType));
940 :
941 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, devicePhyId_));
942 :
943 0 : if (extraData != nullptr && extraDataLen != 0) {
944 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, extraData, extraDataLen));
945 : }
946 :
947 : // 不需要进行barrier，那么我们每处理一个请求就回复一个请求
948 0 : if (!NeedBarrier(requestType)) {
949 0 : CHK_PRT_RET(
950 : SendRequest(ackType, exchangeDataForAck_[remoteDevicePhyId], remoteDevicePhyId) != HCCL_SUCCESS,
951 : HCCL_WARNING("[ZeroCopyMemoryAgent][SendAckAfterParse] failed, remote[%u]", remoteDevicePhyId),
952 : HCCL_E_INTERNAL);
953 0 : return HCCL_SUCCESS;
954 : }
955 :
956 : // 需要进行barrier的请求，我们先统计一下收到的请求数目，等于链接数才算收完所有
957 0 : u32 expectedNum = mapDevPhyIdconnectedSockets_.size();
958 0 : u32 counter = ++reqMsgCounter_[static_cast<int>(requestType)];
959 0 : HCCL_INFO(
960 : "[ZeroCopyMemoryAgent][SendAckAfterParse] requestType[%d] counter %u expect %u", requestType, counter,
961 : expectedNum);
962 0 : if (counter < expectedNum) {
963 0 : return HCCL_SUCCESS;
964 : } else {
965 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
966 0 : reqMsgFinishCnt_++;
967 :
968 : // 我们统一将所有的请求一次性都发送过去
969 0 : CHK_PRT_RET(
970 : SendRequest(ackType, exchangeDataForAck_[remoteDevicePhyId]) != HCCL_SUCCESS,
971 : HCCL_WARNING("[ZeroCopyMemoryAgent][SendAckAfterParse] failed, remote[all]"), HCCL_E_INTERNAL);
972 : }
973 :
974 0 : return HCCL_SUCCESS;
975 : }
976 :
977 0 : HcclResult ZeroCopyMemoryAgent::ParseRemoteAck(RequestType requestType, u32 remoteRank)
978 : {
979 0 : bool useBarrier = NeedBarrier(requestType);
980 0 : std::unique_lock<std::mutex> dfxLock(dfxMutex_);
981 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].insert(remoteRank);
982 0 : u32 counter = ++reqMsgCounter_[static_cast<int>(requestType)];
983 0 : if ((counter == mapDevPhyIdconnectedSockets_.size())
984 0 : && (!useBarrier || (useBarrier && reqMsgDeliverCnt_ <= reqMsgFinishCnt_))) {
985 0 : waitCompleteCv_.notify_all();
986 : }
987 0 : return HCCL_SUCCESS;
988 0 : }
989 :
990 0 : HcclResult ZeroCopyMemoryAgent::ParseUnsetMemoryRange(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
991 : {
992 0 : CHK_PRT_RET(
993 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
994 : HCCL_ERROR(
995 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
996 : "is not init.",
997 : __func__),
998 : HCCL_E_INTERNAL);
999 : u32 devicePhyId;
1000 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1001 :
1002 : u64 addr;
1003 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1004 :
1005 0 : LocalIpc2RemoteAddr mapAddr;
1006 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1007 0 : CHK_PRT_RET(
1008 : addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS,
1009 : HCCL_ERROR(
1010 : "[ZeroCopyMemoryAgent][ParseUnsetMemoryRange] device[%u] not set addr [%p]", devicePhyId, remoteAddr),
1011 : HCCL_E_PARA);
1012 0 : CHK_RET(addressMgr_->DelLocalIpc2RemoteAddr(devicePhyId, reinterpret_cast<void*>(mapAddr.remoteAddr)));
1013 :
1014 0 : void* devPtr = reinterpret_cast<void*>(mapAddr.localIpcAddr);
1015 0 : aclError ret = aclrtReleaseMemAddress(devPtr);
1016 0 : CHK_PRT_RET(
1017 : ret != ACL_SUCCESS,
1018 : HCCL_ERROR(
1019 : "[ZeroCopyMemoryAgent][ParseUnsetMemoryRange]rtRelease Memory failed, "
1020 : "return[%d], devPtr[%p]",
1021 : ret, devPtr),
1022 : HCCL_E_RUNTIME);
1023 :
1024 0 : CHK_RET(SendAckAfterParse(RequestType::UNSET_MEMORY_RANGE, RequestType::UNSET_MEMORY_RANGE_ACK, devicePhyId));
1025 0 : return HCCL_SUCCESS;
1026 : }
1027 :
1028 0 : HcclResult ZeroCopyMemoryAgent::ParseBareTgid(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1029 : {
1030 : u32 devicePhyId;
1031 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1032 :
1033 : // 获取本端的ack，然后通过ack返回给对端
1034 0 : int32_t tgid = 0;
1035 0 : aclError ret = aclrtDeviceGetBareTgid(&tgid);
1036 0 : CHK_PRT_RET(
1037 : ret != ACL_SUCCESS, HCCL_ERROR("[ZeroCopyMemoryAgent][ParseBareTgid] get tgid failed, ret[%d]", ret),
1038 : HCCL_E_RUNTIME);
1039 :
1040 0 : HCCL_INFO(
1041 : "[ZeroCopyMemoryAgent][ParseBareTgid] dev[%u] tgid[%d] to remoteDev[%u]", devicePhyId_, tgid, devicePhyId);
1042 0 : CHK_RET(SendAckAfterParse(
1043 : RequestType::SET_REMOTE_BARE_TGID, RequestType::SET_REMOTE_BARE_TGID_ACK, devicePhyId, &tgid, sizeof(tgid)));
1044 0 : return HCCL_SUCCESS;
1045 : }
1046 :
1047 0 : HcclResult ZeroCopyMemoryAgent::ParseBareTgidAck(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1048 : {
1049 : u32 devicePhyId;
1050 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1051 :
1052 : u32 tgid;
1053 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, tgid));
1054 :
1055 0 : HCCL_INFO("[ZeroCopyMemoryAgent][ParseBareTgidAck] recv dev[%u] tgid[%u]", devicePhyId, tgid);
1056 0 : remotePids_.emplace_back(tgid);
1057 0 : return HCCL_SUCCESS;
1058 : }
1059 :
1060 0 : HcclResult ZeroCopyMemoryAgent::ParseBarrierCloseAck(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1061 : {
1062 : u32 devicePhyId;
1063 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1064 :
1065 : u32 tgid;
1066 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, tgid));
1067 :
1068 0 : receivedBarrierCloseAck_.insert(devicePhyId);
1069 0 : HCCL_RUN_INFO(
1070 : "[ZeroCopyMemoryAgent][ParseBarrierCloseAck] [%s] recv dev[%u] barrier close ack, so we stop this socket's "
1071 : "recv",
1072 : identifier_.c_str(), devicePhyId, tgid);
1073 0 : return HCCL_SUCCESS;
1074 : }
1075 :
1076 0 : HcclResult ZeroCopyMemoryAgent::ParseActivateCommMemory(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1077 : {
1078 0 : CHK_PRT_RET(
1079 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
1080 : HCCL_ERROR(
1081 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
1082 : "is not init.",
1083 : __func__),
1084 : HCCL_E_INTERNAL);
1085 : u32 devicePhyId;
1086 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1087 :
1088 : u64 addr;
1089 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1090 :
1091 : size_t size;
1092 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, size));
1093 :
1094 : size_t offset;
1095 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, offset));
1096 :
1097 : size_t shareableHandle;
1098 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, shareableHandle));
1099 :
1100 : size_t flags;
1101 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, flags));
1102 :
1103 0 : LocalIpc2RemoteAddr mapAddr;
1104 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1105 0 : CHK_PRT_RET(
1106 : (addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS),
1107 : HCCL_ERROR(
1108 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] address may not be reserved in device[%u]", devicePhyId),
1109 : HCCL_E_PARA);
1110 :
1111 0 : HCCL_INFO(
1112 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] prepare import from dev[%u] shareableHandle[%llu]", devicePhyId,
1113 : shareableHandle);
1114 0 : u64 actualAddr = mapAddr.localIpcAddr + (addr - mapAddr.remoteAddr);
1115 0 : void* devPtr = reinterpret_cast<void*>(actualAddr);
1116 0 : CHK_PRT_RET(
1117 : actualAddr + size > mapAddr.localIpcAddr + mapAddr.length,
1118 : HCCL_ERROR(
1119 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] remote addr[0x%lx] size[%llu] exceed memory range", addr,
1120 : size),
1121 : HCCL_E_PARA);
1122 0 : CHK_PRT_RET(
1123 : addressMgr_->IsOverlapWithActivateAddr(devPtr, size),
1124 : HCCL_ERROR(
1125 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] remote addr[0x%lx] size[%llu] devPtr[%p] is overlap", addr,
1126 : size, devPtr),
1127 : HCCL_E_PARA);
1128 :
1129 0 : aclError ret = ACL_SUCCESS;
1130 0 : void* pHandle = nullptr;
1131 0 : CHK_RET(addressMgr_->ActivateCommMemoryAddr(devPtr, size));
1132 0 : ret = aclrtMemImportFromShareableHandle(shareableHandle, deviceLogicId_, &pHandle);
1133 0 : CHK_PRT_RET(
1134 : ret != ACL_SUCCESS,
1135 : HCCL_ERROR(
1136 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] import shareableHandle[%llu] dev[%d] failed, ret[%d]",
1137 : shareableHandle, deviceLogicId_, ret),
1138 : HCCL_E_RUNTIME);
1139 :
1140 0 : ret = aclrtMapMem(devPtr, size, offset, pHandle, flags);
1141 0 : CHK_PRT_RET(
1142 : ret != ACL_SUCCESS,
1143 : HCCL_ERROR(
1144 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] map dev[%p] size[%llu] offset[%llu] handle[%p]",
1145 : " flag[%llu] failed, ret[%d]", devPtr, size, offset, pHandle, flags, ret),
1146 : HCCL_E_RUNTIME);
1147 :
1148 0 : CHK_RET(addressMgr_->AddRemoteImportAddr(devPtr, pHandle));
1149 :
1150 0 : CHK_RET(SendAckAfterParse(RequestType::ACTIVATE_COMM_MEMORY, RequestType::ACTIVATE_COMM_MEMORY_ACK, devicePhyId));
1151 :
1152 0 : return HCCL_SUCCESS;
1153 : }
1154 :
1155 0 : HcclResult ZeroCopyMemoryAgent::ParseDeactivateCommMemory(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1156 : {
1157 0 : CHK_PRT_RET(
1158 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
1159 : HCCL_ERROR(
1160 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
1161 : "is not init.",
1162 : __func__),
1163 : HCCL_E_INTERNAL);
1164 : u32 devicePhyId;
1165 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1166 :
1167 : u64 addr;
1168 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1169 :
1170 0 : LocalIpc2RemoteAddr mapAddr;
1171 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1172 0 : CHK_PRT_RET(
1173 : (addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS),
1174 : HCCL_ERROR(
1175 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] address [%p] not be set in device[%u]", remoteAddr,
1176 : devicePhyId),
1177 : HCCL_E_PARA);
1178 :
1179 0 : u64 actualAddr = mapAddr.localIpcAddr + (addr - mapAddr.remoteAddr);
1180 0 : void* devPtr = reinterpret_cast<void*>(actualAddr);
1181 0 : CHK_RET(addressMgr_->DeactivateCommMemoryAddr(devPtr));
1182 :
1183 0 : void* handle = nullptr;
1184 0 : CHK_RET(addressMgr_->GetRemoteImportAddr(devPtr, handle));
1185 :
1186 0 : aclError ret = ACL_SUCCESS;
1187 0 : ret = aclrtUnmapMem(devPtr);
1188 0 : CHK_PRT_RET(
1189 : ret != ACL_SUCCESS,
1190 : HCCL_ERROR(
1191 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] aclrtUnmapMem dev[%p] failed, ret[%d]", devPtr, ret),
1192 : HCCL_E_RUNTIME);
1193 0 : ret = aclrtFreePhysical(handle);
1194 0 : CHK_PRT_RET(
1195 : ret != ACL_SUCCESS,
1196 : HCCL_ERROR(
1197 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] aclrtFreePhysical handle[%p] failed, ret[%d]", handle,
1198 : ret),
1199 : HCCL_E_RUNTIME);
1200 :
1201 0 : CHK_RET(addressMgr_->DelRemoteImportAddr(devPtr));
1202 :
1203 0 : CHK_RET(
1204 : SendAckAfterParse(RequestType::DEACTIVATE_COMM_MEMORY, RequestType::DEACTIVATE_COMM_MEMORY_ACK, devicePhyId));
1205 :
1206 0 : return HCCL_SUCCESS;
1207 : }
1208 :
1209 0 : HcclResult ZeroCopyMemoryAgent::ParseBarrierClose(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1210 : {
1211 : u32 devicePhyId;
1212 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1213 0 : HCCL_INFO("[ZeroCopyMemoryAgent][ParseBarrierClose] recv dev[%u] barrier close", devicePhyId);
1214 :
1215 0 : receivedBarrierClose_.insert(devicePhyId);
1216 0 : CHK_RET(SendAckAfterParse(RequestType::BARRIER_CLOSE, RequestType::BARRIER_CLOSE_ACK, devicePhyId));
1217 0 : return HCCL_SUCCESS;
1218 : }
1219 :
1220 0 : HcclResult ZeroCopyMemoryAgent::ParseReceivedRequest(std::vector<u8>& receivedData, u32 remoteRank)
1221 : {
1222 0 : u8* exchangeDataPtr = receivedData.data();
1223 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
1224 :
1225 : RequestType requestType;
1226 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, requestType));
1227 :
1228 0 : HcclResult ret = HCCL_SUCCESS;
1229 0 : switch (requestType) {
1230 0 : case RequestType::SET_MEMORY_RANGE:
1231 0 : ret = ParseSetMemoryRange(exchangeDataPtr, exchangeDataBlankSize);
1232 0 : break;
1233 0 : case RequestType::UNSET_MEMORY_RANGE:
1234 0 : ret = ParseUnsetMemoryRange(exchangeDataPtr, exchangeDataBlankSize);
1235 0 : break;
1236 0 : case RequestType::ACTIVATE_COMM_MEMORY:
1237 0 : ret = ParseActivateCommMemory(exchangeDataPtr, exchangeDataBlankSize);
1238 0 : break;
1239 0 : case RequestType::DEACTIVATE_COMM_MEMORY:
1240 0 : ret = ParseDeactivateCommMemory(exchangeDataPtr, exchangeDataBlankSize);
1241 0 : break;
1242 0 : case RequestType::SET_REMOTE_BARE_TGID:
1243 0 : ret = ParseBareTgid(exchangeDataPtr, exchangeDataBlankSize);
1244 0 : break;
1245 0 : case RequestType::BARRIER_CLOSE:
1246 0 : ret = ParseBarrierClose(exchangeDataPtr, exchangeDataBlankSize);
1247 0 : break;
1248 0 : case RequestType::SET_REMOTE_BARE_TGID_ACK:
1249 0 : ret = ParseBareTgidAck(exchangeDataPtr, exchangeDataBlankSize);
1250 0 : ParseRemoteAck(requestType, remoteRank);
1251 0 : break;
1252 0 : case RequestType::SET_MEMORY_RANGE_ACK:
1253 : case RequestType::UNSET_MEMORY_RANGE_ACK:
1254 : case RequestType::ACTIVATE_COMM_MEMORY_ACK:
1255 : case RequestType::DEACTIVATE_COMM_MEMORY_ACK:
1256 0 : ParseRemoteAck(requestType, remoteRank);
1257 0 : break;
1258 0 : case RequestType::BARRIER_CLOSE_ACK:
1259 0 : ret = ParseBarrierCloseAck(exchangeDataPtr, exchangeDataBlankSize);
1260 0 : ParseRemoteAck(requestType, remoteRank);
1261 0 : break;
1262 0 : default:
1263 0 : HCCL_ERROR("[Parse][ReceivedRequest] invalid RequestType[%d]", requestType);
1264 0 : ret = HCCL_E_INTERNAL;
1265 0 : break;
1266 : }
1267 0 : return ret;
1268 : }
1269 :
1270 0 : std::string ZeroCopyMemoryAgent::DumpFinishInfo(RequestType requestType)
1271 : {
1272 0 : auto& finishedRanks = reqMsgFinishedRanks_[static_cast<int>(requestType)];
1273 :
1274 0 : std::string msg = "Expect [";
1275 0 : for (auto& info : rankInfoList_) {
1276 0 : msg += std::to_string(info.userRank) + " ";
1277 : }
1278 :
1279 0 : msg += "] Actual [";
1280 0 : for (auto& rank : finishedRanks) {
1281 0 : msg += std::to_string(rank) + " ";
1282 : }
1283 :
1284 0 : msg += "]";
1285 0 : finishedRanks.clear();
1286 :
1287 0 : return msg;
1288 0 : }
1289 :
1290 0 : bool ZeroCopyMemoryAgent::IsPaused() const { return !threadRun_ || isPaused_; }
1291 :
1292 0 : bool ZeroCopyMemoryAgent::IsResumed() const { return !threadRun_ || !isPaused_; }
1293 :
1294 0 : void ZeroCopyMemoryAgent::CheckSnapshotStatus()
1295 : {
1296 0 : auto snapshotStatus = SnapshotControl::GetInstance(deviceLogicId_).GetStatus();
1297 0 : if (isPaused_ && snapshotStatus == SnapshotStatus::POST_SNAPSHOT) {
1298 0 : isPaused_ = false;
1299 0 : HCCL_RUN_INFO(
1300 : "[ZeroCopyMemoryAgent][CheckSnapshotStatus] detect snapshot post-processing, "
1301 : "zero-copy memory agent is resumed, deviceLogicId[%d].",
1302 : deviceLogicId_);
1303 0 : } else if (!isPaused_ && snapshotStatus == SnapshotStatus::PRE_SNAPSHOT) {
1304 0 : isPaused_ = true;
1305 0 : HCCL_RUN_INFO(
1306 : "[ZeroCopyMemoryAgent][CheckSnapshotStatus] detect snapshot pre-processing, "
1307 : "zero-copy memory agent is paused, deviceLogicId[%d].",
1308 : deviceLogicId_);
1309 : }
1310 0 : }
1311 :
1312 : } // namespace hccl
|
