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, const 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(ptr, leftSize, sendMgr.reqDatas_[0]->data(), IPC_MEMORY_EXCHANGE_LENGTH)
266 0 : == HCCL_SUCCESS) {
267 0 : sendMgr.hasReq_[0] = false;
268 0 : sendMgr.currIndex_ = 1;
269 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH + IPC_MEMORY_EXCHANGE_LENGTH;
270 : } else {
271 0 : sendMgr.currIndex_ = 0;
272 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH;
273 : }
274 : } else {
275 0 : sendMgr.currIndex_ = sendMgr.hasReq_[0] ? 0 : 1;
276 0 : sendMgr.reqDataSize_ = IPC_MEMORY_EXCHANGE_LENGTH;
277 : }
278 : }
279 0 : const std::vector<u8>* req = sendMgr.reqDatas_[sendMgr.currIndex_];
280 0 : sendMgr.lastSendSize_ = 0; // 用于ra上报发送的数据量
281 0 : ret = socket->SendAsync(
282 0 : req->data() + sendMgr.sentSize_, sendMgr.reqDataSize_ - sendMgr.sentSize_, &sendMgr.lastSendSize_,
283 : &sendMgr.lastSendHandle_);
284 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) { // 发送失败的场景
285 0 : RequestType requestType = *reinterpret_cast<const RequestType*>(req->data());
286 0 : HCCL_ERROR(
287 : "[ZeroCopyMemoryAgent][RequestBatchSendAsync] failed, ret[%d] remote[%u] requestType[%s] "
288 : "sentSize[%llu]",
289 : ret, kv.first, GetReadableRequestType(requestType), sendMgr.sentSize_);
290 : }
291 : }
292 0 : }
293 :
294 0 : void ZeroCopyMemoryAgent::CheckBatchSendAsyncResult()
295 : {
296 : HcclResult ret;
297 : HcclResult lastSendRet;
298 0 : for (auto& kv : sendMgrs_) {
299 0 : auto& sendMgr = kv.second;
300 0 : if (sendMgr.lastSendHandle_ == nullptr) { // 没有正在执行的异步send
301 0 : continue;
302 : }
303 :
304 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
305 0 : HCCL_ERROR(
306 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult] remote[%u] not found in"
307 : "mapDevPhyIdconnectedSockets_",
308 : kv.first);
309 0 : continue;
310 : }
311 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
312 0 : ret = socket->GetAsyncReqResult(sendMgr.lastSendHandle_, lastSendRet);
313 0 : if (ret != HCCL_SUCCESS) {
314 0 : CHK_PRT_CONT(
315 : ret != HCCL_E_AGAIN,
316 : HCCL_ERROR(
317 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]GetAsyncReqResult failed, ret[%d] remote[%u]", ret,
318 : kv.first));
319 0 : continue;
320 : }
321 :
322 0 : sendMgr.lastSendHandle_ = nullptr;
323 0 : if ((lastSendRet != HCCL_SUCCESS) && (sendMgr.lastSendSize_ == 0)) {
324 0 : CHK_PRT_CONT(
325 : lastSendRet != HCCL_E_AGAIN, HCCL_ERROR(
326 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]SendAsync failed, "
327 : "result[%d] remote[%u] sentSize[%llu]",
328 : lastSendRet, kv.first, sendMgr.sentSize_));
329 0 : continue;
330 : }
331 :
332 0 : sendMgr.sentSize_ += sendMgr.lastSendSize_; // 下次从中断的地方开始重发
333 0 : if (sendMgr.sentSize_ == sendMgr.reqDataSize_) { // request发送完成
334 0 : sendMgr.sentSize_ = 0;
335 0 : sendMgr.hasReq_[sendMgr.currIndex_] = false;
336 0 : HCCL_DEBUG(
337 : "[ZeroCopyMemoryAgent][CheckBatchSendAsyncResult]SendAsync success, requestType[%s] remote[%u]",
338 : GetReadableRequestType(
339 : *reinterpret_cast<const RequestType*>(sendMgr.reqDatas_[sendMgr.currIndex_]->data())),
340 : kv.first);
341 : }
342 : }
343 0 : }
344 :
345 0 : void ZeroCopyMemoryAgent::RequestBatchRecvAsync()
346 : {
347 : HcclResult ret;
348 0 : for (auto& kv : recvMgrs_) {
349 0 : auto& recvMgr = kv.second;
350 0 : if ((recvMgr.lastRecvHandle_ != nullptr) || // 前回接收未完成
351 0 : ((receivedBarrierClose_.count(kv.first) != 0) && (receivedBarrierCloseAck_.count(kv.first) != 0))) {
352 : // 该socket已经收到BarrierClose与BarrierCloseAck报文，因此不允许再进行其他数据接收了
353 0 : continue;
354 : }
355 :
356 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
357 0 : HCCL_ERROR(
358 : "[ZeroCopyMemoryAgent][RequestBatchRecvAsync] remote[%u] not found in"
359 : "mapDevPhyIdconnectedSockets_",
360 : kv.first);
361 0 : continue;
362 : }
363 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
364 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.recvIndex_];
365 0 : recvMgr.lastRecvSize_ = 0; // 用于ra上报接收的数据量
366 0 : ret = socket->RecvAsync(
367 0 : req.data() + recvMgr.receivedSize_, IPC_MEMORY_EXCHANGE_LENGTH - recvMgr.receivedSize_,
368 : &recvMgr.lastRecvSize_, &recvMgr.lastRecvHandle_);
369 0 : CHK_PRT_CONT(
370 : (ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN),
371 : HCCL_ERROR(
372 : "[ZeroCopyMemoryAgent][RequestBatchRecvAsync] RecvAsync failed, ret[%d] remote[%u] receivedSize[%llu]",
373 : ret, kv.first, recvMgr.receivedSize_));
374 : }
375 0 : }
376 :
377 0 : void ZeroCopyMemoryAgent::CheckBatchRecvAsyncResult()
378 : {
379 : HcclResult ret;
380 : HcclResult lastRecvRet;
381 0 : for (auto& kv : recvMgrs_) {
382 0 : auto& recvMgr = kv.second;
383 0 : if (recvMgr.lastRecvHandle_ == nullptr) { // 没有正在异步接收
384 0 : continue;
385 : }
386 :
387 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
388 0 : HCCL_ERROR(
389 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] remote[%u] not found in"
390 : "mapDevPhyIdconnectedSockets_",
391 : kv.first);
392 0 : continue;
393 : }
394 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
395 0 : ret = socket->GetAsyncReqResult(recvMgr.lastRecvHandle_, lastRecvRet);
396 0 : if (ret != HCCL_SUCCESS) {
397 0 : CHK_PRT_CONT(
398 : ret != HCCL_E_AGAIN,
399 : HCCL_ERROR(
400 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] GetAsyncReqResult failed, ret[%d] remote[%u]",
401 : ret, kv.first));
402 0 : continue;
403 : }
404 :
405 0 : recvMgr.lastRecvHandle_ = nullptr;
406 0 : if ((lastRecvRet != HCCL_SUCCESS) && (recvMgr.lastRecvSize_ == 0)) {
407 0 : CHK_PRT_CONT(
408 : lastRecvRet != HCCL_E_AGAIN, HCCL_WARNING(
409 : "[ZeroCopyMemoryAgent][CheckBatchRecvAsyncResult] RecvAsync failed, "
410 : "result[%d] remote[%u] lastRecvSize[%llu]",
411 : lastRecvRet, kv.first, recvMgr.lastRecvSize_));
412 0 : continue;
413 : }
414 :
415 0 : recvMgr.receivedSize_ += recvMgr.lastRecvSize_;
416 0 : if (recvMgr.receivedSize_ == IPC_MEMORY_EXCHANGE_LENGTH) {
417 0 : recvMgr.receivedSize_ = 0;
418 0 : RecvRequest(recvMgr, kv.first);
419 0 : ioRecvWaiting_ = true; // 后面高概率还有数据要收（ack与request合并场景），loop不等待
420 : } else {
421 : // request没收全，loop不等待
422 0 : ioRecvWaiting_ = (recvMgr.receivedSize_ > 0);
423 : }
424 : }
425 0 : }
426 :
427 0 : inline void ZeroCopyMemoryAgent::RecvRequest(ZeroCopyMemoryAgentRecvMgr& recvMgr, u32 remoteDevicePhyId)
428 : {
429 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.recvIndex_];
430 0 : RequestType requestType = *reinterpret_cast<RequestType*>(req.data());
431 0 : HCCL_DEBUG(
432 : "[ZeroCopyMemoryAgent][RecvRequest] recv requestType[%s] remote[%u]", GetReadableRequestType(requestType),
433 : remoteDevicePhyId);
434 :
435 0 : if (IsAckRequestType(requestType)) { // 收到ACK时，直接优先处理
436 0 : u32 remoteRank = mapDevPhyId2RankId_[remoteDevicePhyId];
437 0 : CHK_PRT_CONT(
438 : ParseReceivedRequest(req, remoteRank) != HCCL_SUCCESS,
439 : HCCL_ERROR(
440 : "[ZeroCopyMemoryAgent][ParseReceivedRequest] failed requestType[%s] remote[%u]",
441 : GetReadableRequestType(requestType), remoteDevicePhyId));
442 0 : return;
443 : }
444 :
445 0 : recvMgr.recvIndex_ = (recvMgr.recvIndex_ + 1) % ZERO_COPY_MEMORY_AGENT_RECV_QUEUE_SIZE; // 准备下一次接收
446 0 : hasReceivedRequest_ = true;
447 : }
448 :
449 0 : void ZeroCopyMemoryAgent::ParseReceivedRequests()
450 : {
451 0 : if (!hasReceivedRequest_) {
452 0 : return;
453 : }
454 0 : hasReceivedRequest_ = false;
455 :
456 0 : for (auto& kv : recvMgrs_) {
457 0 : u32 remoteRank = mapDevPhyId2RankId_[kv.first];
458 0 : auto& recvMgr = kv.second;
459 0 : while (recvMgr.praseIndex_ != recvMgr.recvIndex_) {
460 0 : std::vector<u8>& req = recvMgr.receivedData_[recvMgr.praseIndex_];
461 0 : CHK_PRT_CONT(
462 : ParseReceivedRequest(req, remoteRank) != HCCL_SUCCESS,
463 : HCCL_ERROR(
464 : "[ZeroCopyMemoryAgent][ParseReceivedRequest] failed prase requestType[%s] remote[%u]",
465 : GetReadableRequestType(*reinterpret_cast<RequestType*>(req.data())), kv.first));
466 0 : recvMgr.praseIndex_++;
467 0 : if (recvMgr.praseIndex_ == ZERO_COPY_MEMORY_AGENT_RECV_QUEUE_SIZE) {
468 0 : recvMgr.praseIndex_ = 0;
469 : }
470 : }
471 : }
472 : }
473 :
474 0 : void ZeroCopyMemoryAgent::RequestBatchRecvSync()
475 : {
476 : HcclResult ret;
477 0 : for (auto& kv : recvMgrs_) {
478 0 : auto& recvMgr = kv.second;
479 0 : if ((receivedBarrierClose_.count(kv.first) != 0) && (receivedBarrierCloseAck_.count(kv.first) != 0)) {
480 : // 该socket已经收到BarrierClose与BarrierCloseAck报文，因此不允许再进行其他数据接收了
481 0 : continue;
482 : }
483 :
484 0 : if (mapDevPhyIdconnectedSockets_.find(kv.first) == mapDevPhyIdconnectedSockets_.end()) {
485 0 : HCCL_ERROR(
486 : "[ZeroCopyMemoryAgent][RequestBatchRecvSync] remote[%u] not found in"
487 : "mapDevPhyIdconnectedSockets_",
488 : kv.first);
489 0 : continue;
490 : }
491 0 : auto& socket = mapDevPhyIdconnectedSockets_[kv.first];
492 0 : std::vector<u8>& req = recvMgr.receivedData_[0];
493 0 : recvMgr.lastRecvSize_ = 0;
494 0 : ret = socket->IRecv(
495 0 : req.data() + recvMgr.receivedSize_, IPC_MEMORY_EXCHANGE_LENGTH - recvMgr.receivedSize_,
496 0 : recvMgr.lastRecvSize_);
497 0 : CHK_PRT_CONT(
498 : (ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN),
499 : HCCL_ERROR(
500 : "[ZeroCopyMemoryAgent][RequestBatchRecvSync] IRecv failed, ret[%d] remote[%u] receivedSize[%llu]", ret,
501 : kv.first, recvMgr.receivedSize_));
502 :
503 0 : recvMgr.receivedSize_ += recvMgr.lastRecvSize_;
504 0 : if (recvMgr.receivedSize_ == IPC_MEMORY_EXCHANGE_LENGTH) {
505 0 : recvMgr.receivedSize_ = 0;
506 0 : ret = ParseReceivedRequest(req, mapDevPhyId2RankId_[kv.first]);
507 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS, HCCL_ERROR("[ZeroCopyMemoryAgent][ParseReceivedRequest] failed"));
508 : }
509 : }
510 0 : }
511 :
512 0 : void ZeroCopyMemoryAgent::InnerThread()
513 : {
514 : // 新线程，更新一下使用的设备
515 0 : if (hrtSetDevice(deviceLogicId_) != HCCL_SUCCESS) {
516 0 : HCCL_ERROR("[ZeroCopyMemoryAgent][InnerThread] set device failed");
517 0 : return;
518 : }
519 :
520 0 : while (threadRun_) {
521 0 : CheckSnapshotStatus();
522 0 : if (isPaused_) {
523 0 : SaluSleep(USLEEP_ONE_THOUSAND);
524 0 : continue;
525 : }
526 :
527 0 : if (isSocketSupportAsync_) {
528 0 : CheckBatchSendAsyncResult();
529 0 : RequestBatchSendAsync();
530 :
531 0 : CheckBatchRecvAsyncResult();
532 0 : RequestBatchRecvAsync();
533 :
534 0 : ParseReceivedRequests();
535 :
536 0 : std::unique_lock<std::mutex> lock(sendMutex_);
537 0 : if (!ioRecvWaiting_ && !hasSendRequest_) {
538 0 : sendCv_.wait_for(lock, std::chrono::microseconds(INNER_THREAD_LOOP_US));
539 : }
540 0 : hasSendRequest_ = false;
541 0 : ioRecvWaiting_ = false;
542 0 : } else {
543 0 : RequestBatchRecvSync();
544 0 : SaluSleep(USLEEP_ONE_THOUSAND);
545 : }
546 : }
547 :
548 0 : if (hrtResetDevice(deviceLogicId_) != HCCL_SUCCESS) {
549 0 : HCCL_ERROR("[ZeroCopyMemoryAgent][InnerThread] reset device failed");
550 0 : return;
551 : }
552 : }
553 :
554 0 : HcclResult ZeroCopyMemoryAgent::SetRemoteTgid()
555 : {
556 0 : if (remotePids_.size() == mapDevPhyIdconnectedSockets_.size()) {
557 0 : HCCL_INFO("[ZeroCopyMemoryAgent][SetRemoteTgid] tgid exchange is ok");
558 0 : return HCCL_SUCCESS;
559 : }
560 0 : remotePids_.clear();
561 :
562 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
563 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
564 :
565 0 : RequestType requestType = RequestType::SET_REMOTE_BARE_TGID;
566 :
567 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
568 :
569 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
570 :
571 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
572 :
573 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::SET_REMOTE_BARE_TGID_ACK));
574 0 : if (remotePids_.size() != mapDevPhyIdconnectedSockets_.size()) {
575 0 : HCCL_ERROR(
576 : "[ZeroCopyMemoryAgent][SetRemoteTgid] tgid exchange failed recv pids count[%lu]", remotePids_.size());
577 0 : return HCCL_E_INTERNAL;
578 : }
579 0 : return HCCL_SUCCESS;
580 : }
581 :
582 0 : HcclResult ZeroCopyMemoryAgent::DeInit()
583 : {
584 0 : CHK_PRT_RET(isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][DeInit] single rank communicator"), HCCL_SUCCESS);
585 0 : std::unique_lock<std::mutex> lock(commRefCntLock_);
586 0 : if (!ZeroCopyMemoryAgent::IsAddressMgrInited()) {
587 0 : HCCL_ERROR(
588 : "[ZeroCopyMemoryAgent][%s]addressMgr_ is nullptr, no need to deinit. local rank[u32]", __func__, userRank_);
589 0 : return HCCL_E_INTERNAL;
590 : }
591 0 : threadRun_ = false;
592 0 : if (innerThread_) {
593 0 : if (innerThread_->joinable()) {
594 0 : innerThread_->join(); // 等待线程执行后释放资源
595 : }
596 : }
597 0 : innerThread_ = nullptr;
598 :
599 0 : if (vnicPortCtx_ != nullptr) {
600 0 : HcclNetCloseDev(vnicPortCtx_);
601 0 : vnicPortCtx_ = nullptr;
602 : }
603 0 : CHK_RET(addressMgr_->DecreCommRefCnt());
604 0 : if (addressMgr_->GetCommRefCnt() == 0) {
605 0 : addressMgr_.reset();
606 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][%s]Release addressMgr_", __func__);
607 : }
608 0 : return HCCL_SUCCESS;
609 0 : }
610 :
611 0 : HcclResult ZeroCopyMemoryAgent::SetMemoryRange(void* virPtr, size_t size, size_t alignment, uint64_t flags)
612 : {
613 0 : CHK_PRT_RET(
614 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][SetMemoryRange] single rank communicator"), HCCL_SUCCESS);
615 0 : CHK_PRT_RET(
616 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
617 : HCCL_ERROR(
618 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
619 : "is not init.",
620 : __func__),
621 : HCCL_E_INTERNAL);
622 0 : CHK_PRT_RET(
623 : addressMgr_->SetMemoryRange(devicePhyId_, virPtr, size) != HCCL_SUCCESS,
624 : HCCL_ERROR(
625 : "[ZeroCopyMemoryAgent][SetMemoryRange] invalid set ptr[%p] size[%lu] alignment[%lu] flags[%lu]", virPtr,
626 : size, alignment, flags),
627 : HCCL_E_PARA);
628 :
629 0 : HCCL_INFO(
630 : "[ZeroCopyMemoryAgent][SetMemoryRange] basePtr[%p] size[%lu] alignment[%lu] flag[%lu]", virPtr, size, alignment,
631 : flags);
632 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
633 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
634 :
635 0 : RequestType requestType = RequestType::SET_MEMORY_RANGE;
636 :
637 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
638 :
639 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
640 :
641 0 : u64 addr = reinterpret_cast<u64>(virPtr);
642 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
643 :
644 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, size));
645 :
646 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, alignment));
647 :
648 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, flags));
649 :
650 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
651 :
652 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::SET_MEMORY_RANGE_ACK));
653 0 : return HCCL_SUCCESS;
654 : }
655 :
656 0 : HcclResult ZeroCopyMemoryAgent::UnsetMemoryRange(void* virPtr)
657 : {
658 0 : CHK_PRT_RET(
659 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][UnsetMemoryRange] single rank communicator"), HCCL_SUCCESS);
660 0 : CHK_PRT_RET(
661 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
662 : HCCL_ERROR(
663 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
664 : "is not init.",
665 : __func__),
666 : HCCL_E_INTERNAL);
667 0 : CHK_PRT_RET(
668 : !addressMgr_->IsAddressSet(devicePhyId_, virPtr),
669 : HCCL_ERROR("[ZeroCopyMemoryAgent][UnsetMemoryRange] ptr[%p] is not set memory", virPtr), HCCL_E_PARA);
670 0 : CHK_RET(addressMgr_->UnsetMemoryRange(devicePhyId_, virPtr));
671 :
672 0 : HCCL_INFO("[ZeroCopyMemoryAgent][UnsetMemoryRange] basePtr[%p]", virPtr);
673 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
674 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
675 :
676 0 : RequestType requestType = RequestType::UNSET_MEMORY_RANGE;
677 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
678 :
679 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
680 :
681 0 : u64 addr = reinterpret_cast<u64>(virPtr);
682 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
683 :
684 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
685 :
686 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::UNSET_MEMORY_RANGE_ACK));
687 0 : return HCCL_SUCCESS;
688 : }
689 :
690 : HcclResult
691 0 : ZeroCopyMemoryAgent::ActivateCommMemory(void* virPtr, size_t size, size_t offset, void* memHandle, uint64_t flags)
692 : {
693 0 : CHK_PRT_RET(
694 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][ActivateCommMemory] single rank communicator"), HCCL_SUCCESS);
695 0 : CHK_PRT_RET(
696 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
697 : HCCL_ERROR(
698 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
699 : "is not init.",
700 : __func__),
701 : HCCL_E_INTERNAL);
702 0 : CHK_PRT_RET(
703 : !addressMgr_->IsInSetAddressRange(devicePhyId_, virPtr, size),
704 : HCCL_ERROR(
705 : "[ZeroCopyMemoryAgent][ActivateCommMemory] input ptr[%p] size[%lu] is not in set address range", virPtr,
706 : size),
707 : HCCL_E_PARA);
708 0 : CHK_PRT_RET(
709 : addressMgr_->IsOverlapWithActivateAddr(virPtr, size),
710 : HCCL_ERROR(
711 : "[ZeroCopyMemoryAgent][ActivateCommMemory] input ptr[%p] size[%lu] overlap with activate memory", virPtr,
712 : size),
713 : HCCL_E_PARA);
714 :
715 0 : HCCL_INFO(
716 : "[ZeroCopyMemoryAgent][ActivateCommMemory] virPtr[%p] size[%lu] offset[%lu] memHandle[%p], flags[%lu]", virPtr,
717 : size, offset, memHandle, flags);
718 0 : CHK_RET(SetRemoteTgid());
719 :
720 : uint64_t shareableHandle;
721 0 : aclrtMemHandleType handleType = ACL_MEM_HANDLE_TYPE_NONE;
722 0 : aclError ret = ACL_SUCCESS;
723 0 : ret = aclrtMemExportToShareableHandle(memHandle, handleType, 0, &shareableHandle);
724 0 : CHK_PRT_RET(
725 : ret != ACL_SUCCESS,
726 : HCCL_ERROR(
727 : "[ZeroCopyMemoryAgent][ActivateCommMemory] aclrtMemExportToShareableHandle handle[%p] type[%d] flags[%llu] "
728 : "failed, ret[%d]",
729 : memHandle, handleType, 0, ret),
730 : HCCL_E_RUNTIME);
731 0 : ret = aclrtMemSetPidToShareableHandle(shareableHandle, remotePids_.data(), remotePids_.size());
732 0 : CHK_PRT_RET(
733 : ret != ACL_SUCCESS,
734 : HCCL_ERROR(
735 : "[ZeroCopyMemoryAgent][ActivateCommMemory] aclrtMemSetPidToShareableHandle shareableHandl[%llu]",
736 : " failed, ret[%d]", shareableHandle, ret),
737 : HCCL_E_RUNTIME);
738 :
739 0 : HCCL_INFO(
740 : "[ZeroCopyMemoryAgent][ActivateCommMemory] dev[%u] export shareableHandle[%lu]", devicePhyId_, shareableHandle);
741 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
742 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
743 :
744 0 : RequestType requestType = RequestType::ACTIVATE_COMM_MEMORY;
745 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
746 :
747 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
748 :
749 0 : u64 addr = reinterpret_cast<u64>(virPtr);
750 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
751 :
752 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, size));
753 :
754 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, offset));
755 :
756 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, shareableHandle));
757 :
758 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, flags));
759 :
760 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
761 :
762 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::ACTIVATE_COMM_MEMORY_ACK));
763 0 : CHK_RET(addressMgr_->ActivateCommMemoryAddr(virPtr, size));
764 :
765 0 : return HCCL_SUCCESS;
766 : }
767 :
768 0 : HcclResult ZeroCopyMemoryAgent::DeactivateCommMemory(void* virPtr)
769 : {
770 0 : CHK_PRT_RET(
771 : isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][DeactivateCommMemory] single rank communicator"), HCCL_SUCCESS);
772 0 : CHK_PRT_RET(
773 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
774 : HCCL_ERROR(
775 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
776 : "is not init.",
777 : __func__),
778 : HCCL_E_INTERNAL);
779 0 : CHK_PRT_RET(
780 : !addressMgr_->IsActivateCommMemoryAddr(virPtr, 1),
781 : HCCL_ERROR("[ZeroCopyMemoryAgent][DeactivateCommMemory] input ptr[%p] is not activate", virPtr), HCCL_E_PARA);
782 :
783 0 : HCCL_INFO("[ZeroCopyMemoryAgent][DeactivateCommMemory] virPtr[%p]", virPtr);
784 0 : CHK_RET(addressMgr_->DeactivateCommMemoryAddr(virPtr));
785 :
786 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
787 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
788 :
789 0 : RequestType requestType = RequestType::DEACTIVATE_COMM_MEMORY;
790 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
791 :
792 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
793 :
794 0 : u64 addr = reinterpret_cast<u64>(virPtr);
795 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, addr));
796 :
797 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
798 :
799 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::DEACTIVATE_COMM_MEMORY_ACK));
800 0 : return HCCL_SUCCESS;
801 : }
802 :
803 0 : HcclResult ZeroCopyMemoryAgent::BarrierClose()
804 : {
805 0 : CHK_PRT_RET(isSingleRank_, HCCL_INFO("[ZeroCopyMemoryAgent][BarrierClose] single rank communicator"), HCCL_SUCCESS);
806 :
807 0 : HCCL_RUN_INFO("[ZeroCopyMemoryAgent][BarrierClose] [%s] ready to barrier close", identifier_.c_str());
808 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
809 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
810 :
811 0 : RequestType requestType = RequestType::BARRIER_CLOSE;
812 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, requestType));
813 0 : CHK_RET(ConstructData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId_));
814 :
815 0 : CHK_RET(SendRequest(requestType, exchangeDataForSend_));
816 :
817 0 : CHK_RET(WaitForAllRemoteComplete(RequestType::BARRIER_CLOSE_ACK));
818 :
819 0 : return HCCL_SUCCESS;
820 : }
821 :
822 40 : bool ZeroCopyMemoryAgent::IsActivateCommMemoryAddr(void* virPtr, u64 length)
823 : {
824 40 : if (!ZeroCopyMemoryAgent::IsAddressMgrInited()) {
825 40 : HCCL_INFO("[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent is not init.", __func__);
826 40 : return false;
827 : }
828 0 : return addressMgr_->IsActivateCommMemoryAddr(virPtr, length);
829 : }
830 :
831 0 : HcclResult ZeroCopyMemoryAgent::GetRingBufferAddr(u64& bufferPtr, u64& headPtr, u64& tailPtr)
832 : {
833 0 : CHK_PRT_RET(
834 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
835 : HCCL_ERROR(
836 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
837 : "is not init.",
838 : __func__),
839 : HCCL_E_INTERNAL);
840 0 : addressMgr_->GetRingBufferAddr(bufferPtr, headPtr, tailPtr);
841 0 : return HCCL_SUCCESS;
842 : }
843 :
844 40 : bool ZeroCopyMemoryAgent::IsAddressMgrInited() { return addressMgr_ != nullptr; }
845 :
846 0 : HcclResult ZeroCopyMemoryAgent::WaitForAllRemoteComplete(RequestType requestType)
847 : {
848 0 : bool useBarrier = NeedBarrier(requestType);
849 0 : if (useBarrier) {
850 0 : reqMsgDeliverCnt_++;
851 : }
852 :
853 0 : u32 expectedNum = mapDevPhyIdconnectedSockets_.size();
854 0 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
855 0 : std::unique_lock<std::mutex> lock(dfxMutex_);
856 0 : waitCompleteCv_.wait_for(lock, timeout);
857 0 : if ((reqMsgCounter_[static_cast<int>(requestType)] == expectedNum)
858 0 : && (!useBarrier || (useBarrier && reqMsgDeliverCnt_ <= reqMsgFinishCnt_))) {
859 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
860 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].clear();
861 0 : return HCCL_SUCCESS;
862 : }
863 :
864 0 : HCCL_ERROR(
865 : "[Wait][RemoteComplete %s] dev[%u] errNo[0x%016llx] timeout[%d s] completeCount[%u] %s",
866 : GetReadableRequestType(requestType), devicePhyId_, HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER), timeout,
867 : reqMsgCounter_[static_cast<int>(requestType)].load(), DumpFinishInfo(requestType).c_str());
868 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
869 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].clear();
870 0 : return HCCL_E_TCP_TRANSFER;
871 0 : }
872 :
873 0 : HcclResult ZeroCopyMemoryAgent::ParseSetMemoryRange(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
874 : {
875 0 : CHK_PRT_RET(
876 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
877 : HCCL_ERROR(
878 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
879 : "is not init.",
880 : __func__),
881 : HCCL_E_INTERNAL);
882 : u32 devicePhyId;
883 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
884 :
885 : u64 addr;
886 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
887 :
888 : size_t size;
889 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, size));
890 :
891 : size_t alignment;
892 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, alignment));
893 :
894 : uint64_t flags;
895 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, flags));
896 :
897 : u32 maxDeviceNum;
898 0 : CHK_RET(GetMaxDevNum(maxDeviceNum));
899 0 : CHK_PRT_RET(
900 : devicePhyId >= maxDeviceNum,
901 : HCCL_ERROR(
902 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] devicePhyId[%u] is exceed max device num[%u]", devicePhyId,
903 : maxDeviceNum),
904 : HCCL_E_PARA);
905 :
906 0 : void* remoteAddrBase = reinterpret_cast<void*>(addr);
907 0 : CHK_PRT_RET(
908 : addressMgr_->IsAddressSet(devicePhyId, remoteAddrBase),
909 : HCCL_ERROR(
910 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] devicePhyId[%u] had set addr [%p]", devicePhyId,
911 : remoteAddrBase),
912 : HCCL_E_PARA);
913 :
914 0 : void* devPtr = nullptr;
915 0 : void* devAddr = nullptr;
916 0 : aclError ret = aclrtReserveMemAddress(&devPtr, size, alignment, devAddr, flags);
917 0 : CHK_PRT_RET(
918 : ret != ACL_SUCCESS,
919 : HCCL_ERROR(
920 : "[ZeroCopyMemoryAgent][ParseSetMemoryRange] rtReserve Memory failed, "
921 : "return[%d], devPtr[%p] size[%llu] alignment[%llu] devAddr[%p] flags[%llu]",
922 : ret, devPtr, size, alignment, devAddr, flags),
923 : HCCL_E_RUNTIME);
924 :
925 0 : CHK_RET(addressMgr_->AddLocalIpc2RemoteAddr(devicePhyId, devPtr, reinterpret_cast<void*>(addr), size));
926 :
927 0 : CHK_RET(SendAckAfterParse(RequestType::SET_MEMORY_RANGE, RequestType::SET_MEMORY_RANGE_ACK, devicePhyId));
928 :
929 0 : return HCCL_SUCCESS;
930 : }
931 :
932 0 : HcclResult ZeroCopyMemoryAgent::SendAckAfterParse(
933 : RequestType requestType, RequestType ackType, u32 remoteDevicePhyId, void* extraData, u64 extraDataLen)
934 : {
935 0 : u8* exchangeDataAckPtr = exchangeDataForAck_[remoteDevicePhyId].data();
936 0 : u32 exchangeDataAckBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
937 :
938 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, ackType));
939 :
940 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, devicePhyId_));
941 :
942 0 : if (extraData != nullptr && extraDataLen != 0) {
943 0 : CHK_RET(ConstructData(exchangeDataAckPtr, exchangeDataAckBlankSize, extraData, extraDataLen));
944 : }
945 :
946 : // 不需要进行barrier，那么我们每处理一个请求就回复一个请求
947 0 : if (!NeedBarrier(requestType)) {
948 0 : CHK_PRT_RET(
949 : SendRequest(ackType, exchangeDataForAck_[remoteDevicePhyId], remoteDevicePhyId) != HCCL_SUCCESS,
950 : HCCL_WARNING("[ZeroCopyMemoryAgent][SendAckAfterParse] failed, remote[%u]", remoteDevicePhyId),
951 : HCCL_E_INTERNAL);
952 0 : return HCCL_SUCCESS;
953 : }
954 :
955 : // 需要进行barrier的请求，我们先统计一下收到的请求数目，等于链接数才算收完所有
956 0 : u32 expectedNum = mapDevPhyIdconnectedSockets_.size();
957 0 : u32 counter = ++reqMsgCounter_[static_cast<int>(requestType)];
958 0 : HCCL_INFO(
959 : "[ZeroCopyMemoryAgent][SendAckAfterParse] requestType[%d] counter %u expect %u", requestType, counter,
960 : expectedNum);
961 0 : if (counter < expectedNum) {
962 0 : return HCCL_SUCCESS;
963 : } else {
964 0 : reqMsgCounter_[static_cast<int>(requestType)] = 0;
965 0 : reqMsgFinishCnt_++;
966 :
967 : // 我们统一将所有的请求一次性都发送过去
968 0 : CHK_PRT_RET(
969 : SendRequest(ackType, exchangeDataForAck_[remoteDevicePhyId]) != HCCL_SUCCESS,
970 : HCCL_WARNING("[ZeroCopyMemoryAgent][SendAckAfterParse] failed, remote[all]"), HCCL_E_INTERNAL);
971 : }
972 :
973 0 : return HCCL_SUCCESS;
974 : }
975 :
976 0 : HcclResult ZeroCopyMemoryAgent::ParseRemoteAck(RequestType requestType, u32 remoteRank)
977 : {
978 0 : bool useBarrier = NeedBarrier(requestType);
979 0 : std::unique_lock<std::mutex> dfxLock(dfxMutex_);
980 0 : reqMsgFinishedRanks_[static_cast<int>(requestType)].insert(remoteRank);
981 0 : u32 counter = ++reqMsgCounter_[static_cast<int>(requestType)];
982 0 : if ((counter == mapDevPhyIdconnectedSockets_.size())
983 0 : && (!useBarrier || (useBarrier && reqMsgDeliverCnt_ <= reqMsgFinishCnt_))) {
984 0 : waitCompleteCv_.notify_all();
985 : }
986 0 : return HCCL_SUCCESS;
987 0 : }
988 :
989 0 : HcclResult ZeroCopyMemoryAgent::ParseUnsetMemoryRange(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
990 : {
991 0 : CHK_PRT_RET(
992 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
993 : HCCL_ERROR(
994 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
995 : "is not init.",
996 : __func__),
997 : HCCL_E_INTERNAL);
998 : u32 devicePhyId;
999 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1000 :
1001 : u64 addr;
1002 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1003 :
1004 0 : LocalIpc2RemoteAddr mapAddr;
1005 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1006 0 : CHK_PRT_RET(
1007 : addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS,
1008 : HCCL_ERROR(
1009 : "[ZeroCopyMemoryAgent][ParseUnsetMemoryRange] device[%u] not set addr [%p]", devicePhyId, remoteAddr),
1010 : HCCL_E_PARA);
1011 0 : CHK_RET(addressMgr_->DelLocalIpc2RemoteAddr(devicePhyId, reinterpret_cast<void*>(mapAddr.remoteAddr)));
1012 :
1013 0 : void* devPtr = reinterpret_cast<void*>(mapAddr.localIpcAddr);
1014 0 : aclError ret = aclrtReleaseMemAddress(devPtr);
1015 0 : CHK_PRT_RET(
1016 : ret != ACL_SUCCESS,
1017 : HCCL_ERROR(
1018 : "[ZeroCopyMemoryAgent][ParseUnsetMemoryRange]rtRelease Memory failed, "
1019 : "return[%d], devPtr[%p]",
1020 : ret, devPtr),
1021 : HCCL_E_RUNTIME);
1022 :
1023 0 : CHK_RET(SendAckAfterParse(RequestType::UNSET_MEMORY_RANGE, RequestType::UNSET_MEMORY_RANGE_ACK, devicePhyId));
1024 0 : return HCCL_SUCCESS;
1025 : }
1026 :
1027 0 : HcclResult ZeroCopyMemoryAgent::ParseBareTgid(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1028 : {
1029 : u32 devicePhyId;
1030 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1031 :
1032 : // 获取本端的ack，然后通过ack返回给对端
1033 0 : int32_t tgid = 0;
1034 0 : aclError ret = aclrtDeviceGetBareTgid(&tgid);
1035 0 : CHK_PRT_RET(
1036 : ret != ACL_SUCCESS, HCCL_ERROR("[ZeroCopyMemoryAgent][ParseBareTgid] get tgid failed, ret[%d]", ret),
1037 : HCCL_E_RUNTIME);
1038 :
1039 0 : HCCL_INFO(
1040 : "[ZeroCopyMemoryAgent][ParseBareTgid] dev[%u] tgid[%d] to remoteDev[%u]", devicePhyId_, tgid, devicePhyId);
1041 0 : CHK_RET(SendAckAfterParse(
1042 : RequestType::SET_REMOTE_BARE_TGID, RequestType::SET_REMOTE_BARE_TGID_ACK, devicePhyId, &tgid, sizeof(tgid)));
1043 0 : return HCCL_SUCCESS;
1044 : }
1045 :
1046 0 : HcclResult ZeroCopyMemoryAgent::ParseBareTgidAck(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1047 : {
1048 : u32 devicePhyId;
1049 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1050 :
1051 : u32 tgid;
1052 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, tgid));
1053 :
1054 0 : HCCL_INFO("[ZeroCopyMemoryAgent][ParseBareTgidAck] recv dev[%u] tgid[%u]", devicePhyId, tgid);
1055 0 : remotePids_.emplace_back(tgid);
1056 0 : return HCCL_SUCCESS;
1057 : }
1058 :
1059 0 : HcclResult ZeroCopyMemoryAgent::ParseBarrierCloseAck(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1060 : {
1061 : u32 devicePhyId;
1062 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1063 :
1064 : u32 tgid;
1065 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, tgid));
1066 :
1067 0 : receivedBarrierCloseAck_.insert(devicePhyId);
1068 0 : HCCL_RUN_INFO(
1069 : "[ZeroCopyMemoryAgent][ParseBarrierCloseAck] [%s] recv dev[%u] barrier close ack, so we stop this socket's "
1070 : "recv",
1071 : identifier_.c_str(), devicePhyId, tgid);
1072 0 : return HCCL_SUCCESS;
1073 : }
1074 :
1075 0 : HcclResult ZeroCopyMemoryAgent::ParseActivateCommMemory(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1076 : {
1077 0 : CHK_PRT_RET(
1078 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
1079 : HCCL_ERROR(
1080 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
1081 : "is not init.",
1082 : __func__),
1083 : HCCL_E_INTERNAL);
1084 : u32 devicePhyId;
1085 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1086 :
1087 : u64 addr;
1088 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1089 :
1090 : size_t size;
1091 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, size));
1092 :
1093 : size_t offset;
1094 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, offset));
1095 :
1096 : size_t shareableHandle;
1097 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, shareableHandle));
1098 :
1099 : size_t flags;
1100 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, flags));
1101 :
1102 0 : LocalIpc2RemoteAddr mapAddr;
1103 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1104 0 : CHK_PRT_RET(
1105 : (addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS),
1106 : HCCL_ERROR(
1107 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] address may not be reserved in device[%u]", devicePhyId),
1108 : HCCL_E_PARA);
1109 :
1110 0 : HCCL_INFO(
1111 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] prepare import from dev[%u] shareableHandle[%llu]", devicePhyId,
1112 : shareableHandle);
1113 0 : u64 actualAddr = mapAddr.localIpcAddr + (addr - mapAddr.remoteAddr);
1114 0 : void* devPtr = reinterpret_cast<void*>(actualAddr);
1115 0 : CHK_PRT_RET(
1116 : actualAddr + size > mapAddr.localIpcAddr + mapAddr.length,
1117 : HCCL_ERROR(
1118 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] remote addr[0x%lx] size[%llu] exceed memory range", addr,
1119 : size),
1120 : HCCL_E_PARA);
1121 0 : CHK_PRT_RET(
1122 : addressMgr_->IsOverlapWithActivateAddr(devPtr, size),
1123 : HCCL_ERROR(
1124 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] remote addr[0x%lx] size[%llu] devPtr[%p] is overlap", addr,
1125 : size, devPtr),
1126 : HCCL_E_PARA);
1127 :
1128 0 : aclError ret = ACL_SUCCESS;
1129 0 : void* pHandle = nullptr;
1130 0 : CHK_RET(addressMgr_->ActivateCommMemoryAddr(devPtr, size));
1131 0 : ret = aclrtMemImportFromShareableHandle(shareableHandle, deviceLogicId_, &pHandle);
1132 0 : CHK_PRT_RET(
1133 : ret != ACL_SUCCESS,
1134 : HCCL_ERROR(
1135 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] import shareableHandle[%llu] dev[%d] failed, ret[%d]",
1136 : shareableHandle, deviceLogicId_, ret),
1137 : HCCL_E_RUNTIME);
1138 :
1139 0 : ret = aclrtMapMem(devPtr, size, offset, pHandle, flags);
1140 0 : CHK_PRT_RET(
1141 : ret != ACL_SUCCESS,
1142 : HCCL_ERROR(
1143 : "[ZeroCopyMemoryAgent][ParseActivateCommMemory] map dev[%p] size[%llu] offset[%llu] handle[%p]",
1144 : " flag[%llu] failed, ret[%d]", devPtr, size, offset, pHandle, flags, ret),
1145 : HCCL_E_RUNTIME);
1146 :
1147 0 : CHK_RET(addressMgr_->AddRemoteImportAddr(devPtr, pHandle));
1148 :
1149 0 : CHK_RET(SendAckAfterParse(RequestType::ACTIVATE_COMM_MEMORY, RequestType::ACTIVATE_COMM_MEMORY_ACK, devicePhyId));
1150 :
1151 0 : return HCCL_SUCCESS;
1152 : }
1153 :
1154 0 : HcclResult ZeroCopyMemoryAgent::ParseDeactivateCommMemory(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1155 : {
1156 0 : CHK_PRT_RET(
1157 : !ZeroCopyMemoryAgent::IsAddressMgrInited(),
1158 : HCCL_ERROR(
1159 : "[ZeroCopyMemoryAgent][%s]ZeroCopyMemoryAgent "
1160 : "is not init.",
1161 : __func__),
1162 : HCCL_E_INTERNAL);
1163 : u32 devicePhyId;
1164 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1165 :
1166 : u64 addr;
1167 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, addr));
1168 :
1169 0 : LocalIpc2RemoteAddr mapAddr;
1170 0 : void* remoteAddr = reinterpret_cast<void*>(addr);
1171 0 : CHK_PRT_RET(
1172 : (addressMgr_->GetLocalIpc2RemoteAddr(devicePhyId, remoteAddr, mapAddr) != HCCL_SUCCESS),
1173 : HCCL_ERROR(
1174 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] address [%p] not be set in device[%u]", remoteAddr,
1175 : devicePhyId),
1176 : HCCL_E_PARA);
1177 :
1178 0 : u64 actualAddr = mapAddr.localIpcAddr + (addr - mapAddr.remoteAddr);
1179 0 : void* devPtr = reinterpret_cast<void*>(actualAddr);
1180 0 : CHK_RET(addressMgr_->DeactivateCommMemoryAddr(devPtr));
1181 :
1182 0 : void* handle = nullptr;
1183 0 : CHK_RET(addressMgr_->GetRemoteImportAddr(devPtr, handle));
1184 :
1185 0 : aclError ret = ACL_SUCCESS;
1186 0 : ret = aclrtUnmapMem(devPtr);
1187 0 : CHK_PRT_RET(
1188 : ret != ACL_SUCCESS,
1189 : HCCL_ERROR(
1190 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] aclrtUnmapMem dev[%p] failed, ret[%d]", devPtr, ret),
1191 : HCCL_E_RUNTIME);
1192 0 : ret = aclrtFreePhysical(handle);
1193 0 : CHK_PRT_RET(
1194 : ret != ACL_SUCCESS,
1195 : HCCL_ERROR(
1196 : "[ZeroCopyMemoryAgent][ParseDeactivateCommMemory] aclrtFreePhysical handle[%p] failed, ret[%d]", handle,
1197 : ret),
1198 : HCCL_E_RUNTIME);
1199 :
1200 0 : CHK_RET(addressMgr_->DelRemoteImportAddr(devPtr));
1201 :
1202 0 : CHK_RET(
1203 : SendAckAfterParse(RequestType::DEACTIVATE_COMM_MEMORY, RequestType::DEACTIVATE_COMM_MEMORY_ACK, devicePhyId));
1204 :
1205 0 : return HCCL_SUCCESS;
1206 : }
1207 :
1208 0 : HcclResult ZeroCopyMemoryAgent::ParseBarrierClose(u8*& exchangeDataPtr, u32& exchangeDataBlankSize)
1209 : {
1210 : u32 devicePhyId;
1211 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, devicePhyId));
1212 0 : HCCL_INFO("[ZeroCopyMemoryAgent][ParseBarrierClose] recv dev[%u] barrier close", devicePhyId);
1213 :
1214 0 : receivedBarrierClose_.insert(devicePhyId);
1215 0 : CHK_RET(SendAckAfterParse(RequestType::BARRIER_CLOSE, RequestType::BARRIER_CLOSE_ACK, devicePhyId));
1216 0 : return HCCL_SUCCESS;
1217 : }
1218 :
1219 0 : HcclResult ZeroCopyMemoryAgent::ParseReceivedRequest(std::vector<u8>& receivedData, u32 remoteRank)
1220 : {
1221 0 : u8* exchangeDataPtr = receivedData.data();
1222 0 : u32 exchangeDataBlankSize = IPC_MEMORY_EXCHANGE_LENGTH;
1223 :
1224 : RequestType requestType;
1225 0 : CHK_RET(ParseData(exchangeDataPtr, exchangeDataBlankSize, requestType));
1226 :
1227 0 : HcclResult ret = HCCL_SUCCESS;
1228 0 : switch (requestType) {
1229 0 : case RequestType::SET_MEMORY_RANGE:
1230 0 : ret = ParseSetMemoryRange(exchangeDataPtr, exchangeDataBlankSize);
1231 0 : break;
1232 0 : case RequestType::UNSET_MEMORY_RANGE:
1233 0 : ret = ParseUnsetMemoryRange(exchangeDataPtr, exchangeDataBlankSize);
1234 0 : break;
1235 0 : case RequestType::ACTIVATE_COMM_MEMORY:
1236 0 : ret = ParseActivateCommMemory(exchangeDataPtr, exchangeDataBlankSize);
1237 0 : break;
1238 0 : case RequestType::DEACTIVATE_COMM_MEMORY:
1239 0 : ret = ParseDeactivateCommMemory(exchangeDataPtr, exchangeDataBlankSize);
1240 0 : break;
1241 0 : case RequestType::SET_REMOTE_BARE_TGID:
1242 0 : ret = ParseBareTgid(exchangeDataPtr, exchangeDataBlankSize);
1243 0 : break;
1244 0 : case RequestType::BARRIER_CLOSE:
1245 0 : ret = ParseBarrierClose(exchangeDataPtr, exchangeDataBlankSize);
1246 0 : break;
1247 0 : case RequestType::SET_REMOTE_BARE_TGID_ACK:
1248 0 : ret = ParseBareTgidAck(exchangeDataPtr, exchangeDataBlankSize);
1249 0 : ParseRemoteAck(requestType, remoteRank);
1250 0 : break;
1251 0 : case RequestType::SET_MEMORY_RANGE_ACK:
1252 : case RequestType::UNSET_MEMORY_RANGE_ACK:
1253 : case RequestType::ACTIVATE_COMM_MEMORY_ACK:
1254 : case RequestType::DEACTIVATE_COMM_MEMORY_ACK:
1255 0 : ParseRemoteAck(requestType, remoteRank);
1256 0 : break;
1257 0 : case RequestType::BARRIER_CLOSE_ACK:
1258 0 : ret = ParseBarrierCloseAck(exchangeDataPtr, exchangeDataBlankSize);
1259 0 : ParseRemoteAck(requestType, remoteRank);
1260 0 : break;
1261 0 : default:
1262 0 : HCCL_ERROR("[Parse][ReceivedRequest] invalid RequestType[%d]", requestType);
1263 0 : ret = HCCL_E_INTERNAL;
1264 0 : break;
1265 : }
1266 0 : return ret;
1267 : }
1268 :
1269 0 : std::string ZeroCopyMemoryAgent::DumpFinishInfo(RequestType requestType)
1270 : {
1271 0 : auto& finishedRanks = reqMsgFinishedRanks_[static_cast<int>(requestType)];
1272 :
1273 0 : std::string msg = "Expect [";
1274 0 : for (auto& info : rankInfoList_) {
1275 0 : msg += std::to_string(info.userRank) + " ";
1276 : }
1277 :
1278 0 : msg += "] Actual [";
1279 0 : for (auto& rank : finishedRanks) {
1280 0 : msg += std::to_string(rank) + " ";
1281 : }
1282 :
1283 0 : msg += "]";
1284 0 : finishedRanks.clear();
1285 :
1286 0 : return msg;
1287 0 : }
1288 :
1289 0 : bool ZeroCopyMemoryAgent::IsPaused() const { return !threadRun_ || isPaused_; }
1290 :
1291 0 : bool ZeroCopyMemoryAgent::IsResumed() const { return !threadRun_ || !isPaused_; }
1292 :
1293 0 : void ZeroCopyMemoryAgent::CheckSnapshotStatus()
1294 : {
1295 0 : auto snapshotStatus = SnapshotControl::GetInstance(deviceLogicId_).GetStatus();
1296 0 : if (isPaused_ && snapshotStatus == SnapshotStatus::POST_SNAPSHOT) {
1297 0 : isPaused_ = false;
1298 0 : HCCL_RUN_INFO(
1299 : "[ZeroCopyMemoryAgent][CheckSnapshotStatus] detect snapshot post-processing, "
1300 : "zero-copy memory agent is resumed, deviceLogicId[%d].",
1301 : deviceLogicId_);
1302 0 : } else if (!isPaused_ && snapshotStatus == SnapshotStatus::PRE_SNAPSHOT) {
1303 0 : isPaused_ = true;
1304 0 : HCCL_RUN_INFO(
1305 : "[ZeroCopyMemoryAgent][CheckSnapshotStatus] detect snapshot pre-processing, "
1306 : "zero-copy memory agent is paused, deviceLogicId[%d].",
1307 : deviceLogicId_);
1308 : }
1309 0 : }
1310 :
1311 : } // namespace hccl
|
