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 "aicpu_zero_copy_exchanger.h"
12 : #include "ascend_hal.h"
13 : #include "sal_pub.h"
14 :
15 : namespace hccl {
16 : ZeroCopyAddressMgr AicpuZeroCopyExchanger::globalAddrMgr_;
17 :
18 11 : AicpuZeroCopyExchanger::AicpuZeroCopyExchanger(
19 : u32 rank, u32 rankSize, const HcclOpResParam* resParam, std::function<bool()> needStop, u32 timeoutSec,
20 11 : u32 deviceNumPerAggregation, u32 taskMonitorInterval)
21 11 : : rankId_(rank),
22 11 : rankSize_(rankSize),
23 11 : resParam_(resParam),
24 11 : needStop_(needStop),
25 11 : timeoutSec_(timeoutSec),
26 11 : deviceNumPerAggregation_(deviceNumPerAggregation),
27 11 : taskMonitorInterval_(taskMonitorInterval)
28 : {
29 11 : HCCL_INFO("Construct AicpuZeroCopyExchanger complete.");
30 11 : }
31 :
32 11 : AicpuZeroCopyExchanger::~AicpuZeroCopyExchanger() {}
33 :
34 0 : HcclResult AicpuZeroCopyExchanger::ExchangeAddress(
35 : const std::string& tag, void* localInput, void* localOutput, AlgResourceResponse* algResResponse)
36 : {
37 0 : if (localInput == nullptr || localOutput == nullptr || algResResponse == nullptr) {
38 0 : HCCL_ERROR("[AicpuZeroCopyExchanger][ExchangeAddress] Invalid input params, maybe nullptr");
39 0 : return HCCL_E_PARA;
40 : }
41 :
42 0 : CHK_PRT_RET(
43 : needStop_ == nullptr, HCCL_ERROR("[AicpuZeroCopyExchanger][ExchangeAddress] needStop function is nullptr"),
44 : HCCL_E_PARA);
45 0 : HcclUs startut = TIME_NOW();
46 0 : HCCL_INFO(
47 : "[AicpuZeroCopyExchanger][ExchangeAddress] rank[%u] input[%p] output[%p]", rankId_, localInput, localOutput);
48 0 : CHK_RET(PrepareTagRes(tag, algResResponse->opTransportResponse));
49 0 : CHK_PTR_NULL(current_);
50 :
51 0 : if (!IsAllIpcAddressValid()) {
52 0 : HCCL_ERROR("[AicpuZeroCopyExchanger][ExchangeAddress] may some ipc address invalid");
53 0 : return HCCL_E_PARA;
54 : }
55 :
56 0 : CHK_RET(BatchSetLocalAddrToRemote(localInput, localOutput));
57 :
58 0 : HcclResult ret = GetRemoteAddr();
59 0 : if (ret != HCCL_SUCCESS) {
60 0 : HCCL_ERROR("[AicpuZeroCopyExchanger][ExchangeAddress] tag[%s], rank[%u]", tag.c_str(), rankId_);
61 0 : return ret;
62 : }
63 :
64 0 : CHK_RET(UpdateTransportAddress());
65 0 : HcclUs endut = TIME_NOW();
66 0 : auto timeVal = DURATION_US(endut - startut).count();
67 0 : constexpr u64 MS_TO_US = 1000;
68 0 : if (taskMonitorInterval_ != 0 && static_cast<u64>(timeVal) >= taskMonitorInterval_ * MS_TO_US) {
69 0 : std::string endInfo;
70 0 : const int kLogMessageBufferSize = 100;
71 0 : endInfo.reserve(kLogMessageBufferSize);
72 0 : endInfo = "task time: " + std::to_string(timeVal) + " us," + "taskMonitor"
73 0 : + std::to_string(taskMonitorInterval_ * MS_TO_US) + " us";
74 0 : HCCL_RUN_INFO("[ExchangeAddress] %s, %s", tag.c_str(), endInfo.c_str());
75 0 : }
76 0 : return HCCL_SUCCESS;
77 : }
78 :
79 0 : HcclResult AicpuZeroCopyExchanger::PrepareRemoteUserMemRanges(
80 : const uint64_t inputSize, const uint64_t outputSize, std::vector<OpUnfoldMemRange>& userInputMemRanges,
81 : std::vector<OpUnfoldMemRange>& userOutputMemRanges) const
82 : {
83 : // 注意: 不能直接使用inAddrs_和outAddrs_, 保存的是remote ranks' user input/output memory在远端的virtual addr
84 : // 需要使用current_->links中的input/output memory, 才是remote ranks' user input/output memory在本端的virtual addr
85 :
86 0 : HCCL_INFO("[AicpuZeroCopyExchanger][PrepareRemoteUserMemRanges] prepare remote input/output memory ranges");
87 :
88 0 : const uint32_t rankSize = userInputMemRanges.size(); // 获取通信域内的rank数量
89 0 : const std::vector<LINK>& links = current_->links;
90 0 : for (size_t linkIdx = 0; linkIdx < links.size(); ++linkIdx) {
91 0 : const LINK& curLink = links[linkIdx];
92 :
93 : // 对端在通信域内的rank id
94 0 : const uint32_t remoteRank = curLink->GetRemoteRank();
95 0 : CHK_PRT_RET(
96 : remoteRank >= rankSize,
97 : HCCL_ERROR(
98 : "[AicpuZeroCopyExchanger][PrepareRemoteUserMemRanges] remoteRank %u >= rankSize %u", remoteRank,
99 : rankSize),
100 : HCCL_E_INTERNAL);
101 :
102 0 : HCCL_INFO(
103 : "[AicpuZeroCopyExchanger][PrepareRemoteUserMemRanges] prepare memory range of remote rank %u", remoteRank);
104 :
105 : // 获取remote user input memory addr
106 0 : void* remoteUserInputBaseAddr = nullptr;
107 0 : CHK_RET(curLink->GetRemoteMem(UserMemType::INPUT_MEM, &remoteUserInputBaseAddr));
108 0 : CHK_PTR_NULL(remoteUserInputBaseAddr);
109 :
110 : // 更新remote user input memory range
111 0 : OpUnfoldMemRange& remoteInputMemRange = userInputMemRanges[remoteRank];
112 0 : remoteInputMemRange.isValid = true;
113 0 : remoteInputMemRange.baseAddr = reinterpret_cast<uint64_t>(remoteUserInputBaseAddr);
114 0 : remoteInputMemRange.memSize = inputSize;
115 :
116 : // 获取remote user output memory addr
117 0 : void* remoteUserOutputBaseAddr = nullptr;
118 0 : CHK_RET(curLink->GetRemoteMem(UserMemType::OUTPUT_MEM, &remoteUserOutputBaseAddr));
119 0 : CHK_PTR_NULL(remoteUserOutputBaseAddr);
120 :
121 : // 更新remote user output memory range
122 0 : OpUnfoldMemRange& remoteOutputMemRange = userOutputMemRanges[remoteRank];
123 0 : remoteOutputMemRange.isValid = true;
124 0 : remoteOutputMemRange.baseAddr = reinterpret_cast<uint64_t>(remoteUserOutputBaseAddr);
125 0 : remoteOutputMemRange.memSize = outputSize;
126 : }
127 :
128 0 : return HCCL_SUCCESS;
129 : }
130 :
131 0 : bool AicpuZeroCopyExchanger::IsAllIpcAddressValid()
132 : {
133 : // 目前只判断所有的共享内存是否Ok，映射部分校验放到后面check
134 0 : if (resParam_->zeroCopyIpcPtrs[rankId_ % deviceNumPerAggregation_] == 0) {
135 0 : HCCL_ERROR("[AicpuZeroCopyExchanger][IsAllIpcAddressValid] self rank %u ipc addrs is nullptr", rankId_);
136 0 : return false;
137 : }
138 :
139 0 : for (auto rank : current_->remoteRanks) {
140 0 : CHK_PRT_RET(
141 : resParam_->zeroCopyIpcPtrs[rank % deviceNumPerAggregation_] == 0,
142 : HCCL_ERROR("[AicpuZeroCopyExchanger][IsAllIpcAddressValid] rank %u ipc addrs is nullptr", rank), false);
143 : }
144 :
145 0 : return true;
146 : }
147 :
148 0 : bool AicpuZeroCopyExchanger::IsSupportZeroCopyLinkType(LinkType linkType)
149 : {
150 0 : return linkType == LinkType::LINK_HCCS || linkType == LinkType::LINK_SIO || linkType == LinkType::LINK_HCCS_SW;
151 : }
152 :
153 0 : HcclResult AicpuZeroCopyExchanger::TryToRead(FlagData& data, u64& in, u64& out)
154 : {
155 0 : u64 flag = data.flag;
156 0 : CHK_PRT_RET(
157 : flag != INVALID_DATA && flag != VALID_DATA,
158 : HCCL_ERROR("[AicpuZeroCopyExchanger][TryToRead] flag is [%lu] corruption", flag), HCCL_E_INTERNAL);
159 :
160 : // 必须是有效的才能读
161 0 : if (data.flag != VALID_DATA) {
162 0 : return HCCL_E_AGAIN;
163 : }
164 :
165 : // 先读取数据，再修改flag
166 0 : in = data.inAddr;
167 0 : out = data.outAddr;
168 0 : if (in == 0 || out == 0) {
169 0 : return HCCL_E_AGAIN;
170 : }
171 :
172 0 : MemFence();
173 :
174 0 : data.flag = INVALID_DATA;
175 0 : data.inAddr = 0;
176 0 : data.outAddr = 0;
177 :
178 0 : MemFence();
179 :
180 0 : return HCCL_SUCCESS;
181 : }
182 :
183 0 : HcclResult AicpuZeroCopyExchanger::GetRemoteRanks(TagRes& tagRes, OpCommTransport& opTransportResponse)
184 : {
185 0 : CHK_PRT_RET(
186 : opTransportResponse.size() == 0,
187 : HCCL_ERROR("[AicpuZeroCopyExchanger][GetRemoteRanks] opTransportResponse size is 0"), HCCL_E_PARA);
188 : // 先清空已有的数据
189 0 : tagRes.remoteRanks.clear();
190 0 : tagRes.links.clear();
191 :
192 0 : for (auto& singleSubCommTransport : opTransportResponse[COMM_LEVEL0]) {
193 0 : for (u64 i = 0; i < singleSubCommTransport.links.size(); ++i) {
194 0 : LINK link = singleSubCommTransport.links[i];
195 0 : if (link == nullptr || !singleSubCommTransport.transportRequests[i].isValid
196 0 : || !IsSupportZeroCopyLinkType(link->GetLinkType())) {
197 : // 无效或者不支持的链路
198 0 : continue;
199 : }
200 0 : tagRes.remoteRanks.insert(link->GetRemoteRank());
201 0 : tagRes.links.emplace_back(link);
202 0 : }
203 : }
204 :
205 : // 校验交换地址的buffer长度是足够，目前是固定使用16个
206 : u32 maxDeviceNum;
207 0 : CHK_RET(GetMaxDevNum(maxDeviceNum));
208 0 : u64 actualUseLen = maxDeviceNum * sizeof(FlagData);
209 0 : CHK_PRT_RET(
210 : actualUseLen > ZERO_COPY_IPC_BUFFER_LENGTH,
211 : HCCL_ERROR(
212 : "[AicpuZeroCopyExchanger][GetRemoteRanks] invalid ipc buffer length [%lu] max [%lu]", actualUseLen,
213 : ZERO_COPY_IPC_BUFFER_LENGTH),
214 : HCCL_E_PARA);
215 :
216 0 : return HCCL_SUCCESS;
217 : }
218 :
219 0 : HcclResult AicpuZeroCopyExchanger::PrepareTagRes(const std::string& tag, OpCommTransport& opTransportResponse)
220 : {
221 : // 清理一下当前正在使用的tag资源
222 0 : current_ = nullptr;
223 :
224 : // 查找是否已经配置过
225 0 : HCCL_INFO("[%s] tag[%s]", __func__, tag.c_str());
226 0 : auto it = tagRes_.find(tag);
227 0 : if (it != tagRes_.end()) {
228 0 : current_ = &it->second;
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 0 : TagRes tagRes;
233 0 : CHK_RET(GetRemoteRanks(tagRes, opTransportResponse));
234 :
235 0 : tagRes_[tag] = tagRes;
236 0 : current_ = &tagRes_[tag];
237 :
238 : // 初始化batchSdma的数据
239 0 : auto peerCount = current_->remoteRanks.size();
240 0 : current_->remotePtrs.resize(peerCount, nullptr);
241 0 : current_->selfPtrs.resize(peerCount, nullptr);
242 0 : current_->selfData.resize(peerCount);
243 0 : current_->sizes.resize(peerCount, sizeof(FlagData));
244 0 : current_->rankIds.resize(peerCount);
245 :
246 : // 准备batch sdma的输入输出地址
247 0 : int index = 0;
248 0 : for (auto remoteRank : current_->remoteRanks) {
249 0 : FlagData* datas
250 0 : = reinterpret_cast<FlagData*>(resParam_->zeroCopyIpcPtrs[remoteRank % deviceNumPerAggregation_]);
251 0 : current_->remotePtrs[index] = &datas[rankId_ % deviceNumPerAggregation_];
252 0 : current_->selfPtrs[index] = ¤t_->selfData[index];
253 0 : current_->rankIds[index] = remoteRank;
254 0 : ++index;
255 : }
256 :
257 0 : return HCCL_SUCCESS;
258 0 : }
259 :
260 0 : HcclResult AicpuZeroCopyExchanger::GetRemoteAddr()
261 : {
262 : // 遍历所有对端，读取出自己所拥有的地址即可
263 0 : std::set<u32> doneRanks;
264 0 : HcclResult ret = HCCL_SUCCESS;
265 :
266 0 : auto startTime = std::chrono::steady_clock::now();
267 0 : auto timeout = std::chrono::seconds(timeoutSec_);
268 0 : while (doneRanks.size() < current_->remoteRanks.size()) {
269 0 : CHK_PRT_RET(
270 : needStop_(), HCCL_ERROR("AicpuZeroCopyExchanger][GetRemoteAddr] we need stop now"), HCCL_E_SUSPENDING);
271 0 : for (auto remoteRank : current_->remoteRanks) {
272 0 : CHK_PRT_RET(
273 : ((std::chrono::steady_clock::now() - startTime) > timeout && timeoutSec_ != 0),
274 : HCCL_ERROR(
275 : "[AicpuZeroCopyExchanger][GetRemoteAddr] get remote addr timeout [%ld s], %s", timeout,
276 : DumpLinkInfo(doneRanks).c_str()),
277 : HCCL_E_TIMEOUT);
278 :
279 0 : if (doneRanks.find(remoteRank) != doneRanks.end()) {
280 0 : continue;
281 : }
282 :
283 0 : FlagData* datas
284 0 : = reinterpret_cast<FlagData*>(resParam_->zeroCopyIpcPtrs[rankId_ % deviceNumPerAggregation_]);
285 0 : ret = TryToRead(
286 0 : datas[remoteRank % deviceNumPerAggregation_], inAddrs_[remoteRank % deviceNumPerAggregation_],
287 0 : outAddrs_[remoteRank % deviceNumPerAggregation_]);
288 0 : if (ret == HCCL_E_AGAIN) {
289 0 : continue;
290 0 : } else if (ret == HCCL_SUCCESS) {
291 0 : HCCL_INFO(
292 : "[AicpuZeroCopyExchanger][GetRemoteAddr] success read from rank[%u], remoteInput[0x%lx] "
293 : "remoteOutput[0x%lx]",
294 : remoteRank, inAddrs_[remoteRank % deviceNumPerAggregation_],
295 : outAddrs_[remoteRank % deviceNumPerAggregation_]);
296 0 : doneRanks.insert(remoteRank);
297 : } else {
298 0 : HCCL_ERROR(
299 : "[AicpuZeroCopyExchanger][GetRemoteAddr] failed read from rank[%u] ipcPtr[%p] data[%p]", remoteRank,
300 : datas, &datas[remoteRank % deviceNumPerAggregation_]);
301 0 : return ret;
302 : }
303 : }
304 : }
305 :
306 0 : return HCCL_SUCCESS;
307 0 : }
308 :
309 0 : HcclResult AicpuZeroCopyExchanger::BatchSetLocalAddrToRemote(void* in, void* out)
310 : {
311 0 : CHK_PTR_NULL(in);
312 0 : CHK_PTR_NULL(out);
313 0 : CHK_PTR_NULL(current_);
314 :
315 0 : size_t peerCount = current_->remoteRanks.size();
316 0 : auto startTime = std::chrono::steady_clock::now();
317 0 : auto timeout = std::chrono::seconds(timeoutSec_);
318 0 : std::set<u32> doneRanks;
319 : while (true) {
320 0 : CHK_PRT_RET(
321 : needStop_(), HCCL_ERROR("AicpuZeroCopyExchanger][BatchSetLocalAddrToRemote] we need stop now"),
322 : HCCL_E_SUSPENDING);
323 :
324 0 : CHK_PRT_RET(
325 : ((std::chrono::steady_clock::now() - startTime) > timeout && timeoutSec_ != 0),
326 : HCCL_ERROR(
327 : "[AicpuZeroCopyExchanger][BatchSetLocalAddrToRemote] Set to remote addr timeout [%ld s], %s", timeout,
328 : DumpLinkInfo(doneRanks).c_str()),
329 : HCCL_E_TIMEOUT);
330 :
331 0 : DVresult ret = halSdmaBatchCopy(
332 0 : current_->selfPtrs.data(), current_->remotePtrs.data(), current_->sizes.data(), peerCount);
333 0 : CHK_PRT_RET(
334 : ret != 0,
335 : HCCL_ERROR(
336 : "[[AicpuZeroCopyExchanger][BatchSetLocalAddrToRemote] Batch get remote "
337 : "failed, ret[%u]",
338 : ret),
339 : HCCL_E_INTERNAL);
340 :
341 0 : size_t readyCount = 0;
342 0 : for (u64 i = 0; i < peerCount; ++i) {
343 0 : FlagData* data = reinterpret_cast<FlagData*>(current_->selfPtrs[i]);
344 0 : u64 flag = data->flag;
345 0 : CHK_PRT_RET(
346 : flag != INVALID_DATA && flag != VALID_DATA,
347 : HCCL_ERROR(
348 : "[AicpuZeroCopyExchanger][BatchSetLocalAddrToRemote] rank[%lu]'s flag is [%lu] corruption",
349 : current_->rankIds[i], flag),
350 : HCCL_E_INTERNAL);
351 :
352 0 : if (flag != INVALID_DATA) {
353 0 : break;
354 : }
355 0 : readyCount++;
356 0 : data->inAddr = reinterpret_cast<u64>(in);
357 0 : data->outAddr = reinterpret_cast<u64>(out);
358 0 : data->flag = VALID_DATA;
359 0 : doneRanks.insert(current_->rankIds[i]);
360 : }
361 :
362 0 : if (readyCount != peerCount) {
363 0 : continue;
364 : }
365 :
366 0 : ret = halSdmaBatchCopy(
367 0 : current_->remotePtrs.data(), current_->selfPtrs.data(), current_->sizes.data(), peerCount);
368 0 : CHK_PRT_RET(
369 : ret != 0,
370 : HCCL_ERROR(
371 : "[[AicpuZeroCopyExchanger][BatchSetLocalAddrToRemote] Batch get remote "
372 : "failed, ret[%u]",
373 : ret),
374 : HCCL_E_INTERNAL);
375 0 : break;
376 0 : }
377 :
378 0 : return HCCL_SUCCESS;
379 0 : }
380 :
381 0 : HcclResult AicpuZeroCopyExchanger::UpdateTransportAddress()
382 : {
383 0 : u32* head = reinterpret_cast<u32*>(resParam_->zeroCopyHeadPtr);
384 0 : u32* tail = reinterpret_cast<u32*>(resParam_->zeroCopyTailPtr);
385 0 : ZeroCopyRingBufferItem* ringBuffer = reinterpret_cast<ZeroCopyRingBufferItem*>(resParam_->zeroCopyRingBuffer);
386 :
387 0 : CHK_PRT_RET(
388 : head == nullptr || tail == nullptr || ringBuffer == nullptr,
389 : HCCL_ERROR("[AicpuZeroCopyExchanger][UpdateTransportAddress] ring buffer ptr is nullptr"), HCCL_E_INTERNAL);
390 :
391 : // RingBuffer中有东西，所以先去处理一下，更新一下mgr的值
392 0 : if (*head != *tail) {
393 0 : CHK_RET(globalAddrMgr_.ProcessRingBuffer(ringBuffer, head, tail));
394 : }
395 :
396 0 : u64 remoteIns[MAX_MODULE_DEVICE_NUM]{};
397 0 : u64 remoteOuts[MAX_MODULE_DEVICE_NUM]{};
398 0 : for (auto remoteRank : current_->remoteRanks) {
399 0 : u32 devicePhyId = resParam_->zeroCopyDevicePhyId[remoteRank % deviceNumPerAggregation_];
400 :
401 : // remote in addr
402 0 : LocalIpc2RemoteAddr inMapAddr;
403 0 : CHK_RET(globalAddrMgr_.GetLocalIpc2RemoteAddr(
404 : devicePhyId, reinterpret_cast<void*>(inAddrs_[remoteRank % deviceNumPerAggregation_]), inMapAddr));
405 0 : remoteIns[remoteRank % deviceNumPerAggregation_]
406 0 : = inMapAddr.localIpcAddr + (inAddrs_[remoteRank % deviceNumPerAggregation_] - inMapAddr.remoteAddr);
407 0 : CHK_PRT_RET(
408 : !globalAddrMgr_.IsActivateCommMemoryAddr(
409 : reinterpret_cast<void*>(remoteIns[remoteRank % deviceNumPerAggregation_]), 1),
410 : HCCL_ERROR(
411 : "[AicpuZeroCopyExchanger][UpdateTransportAddress] rank[%u] ptr[0x%lx] is not activate", remoteRank,
412 : remoteIns[remoteRank % deviceNumPerAggregation_]),
413 : HCCL_E_PARA);
414 :
415 : // remote out addr
416 0 : LocalIpc2RemoteAddr outMapAddr;
417 0 : CHK_RET(globalAddrMgr_.GetLocalIpc2RemoteAddr(
418 : devicePhyId, reinterpret_cast<void*>(outAddrs_[remoteRank % deviceNumPerAggregation_]), outMapAddr));
419 0 : remoteOuts[remoteRank % deviceNumPerAggregation_]
420 0 : = outMapAddr.localIpcAddr + (outAddrs_[remoteRank % deviceNumPerAggregation_] - outMapAddr.remoteAddr);
421 0 : CHK_PRT_RET(
422 : !globalAddrMgr_.IsActivateCommMemoryAddr(
423 : reinterpret_cast<void*>(remoteOuts[remoteRank % deviceNumPerAggregation_]), 1),
424 : HCCL_ERROR(
425 : "[AicpuZeroCopyExchanger][UpdateTransportAddress] rank[%u] ptr[0x%lx] is not activate", remoteRank,
426 : remoteOuts[remoteRank % deviceNumPerAggregation_]),
427 : HCCL_E_PARA);
428 :
429 0 : HCCL_INFO(
430 : "[AicpuZeroCopyExchanger][UpdateTransportAddress] remoteRank[%u] localInBase[0x%lx] remoteInBase[0x%lx] "
431 : "remoteIn [0x%lx] localOutBase [0x%lx] remoteOutBase [0x%lx] remoteOut [0x%lx]",
432 : remoteRank, inMapAddr.localIpcAddr, inMapAddr.remoteAddr, remoteIns[remoteRank % deviceNumPerAggregation_],
433 : outMapAddr.localIpcAddr, outMapAddr.remoteAddr, remoteOuts[remoteRank % deviceNumPerAggregation_]);
434 : }
435 :
436 : // 因此同一个对端可能有多条p2p链路
437 0 : for (auto& link : current_->links) {
438 0 : u32 remoteRank = link->GetRemoteRank();
439 0 : void* remoteIn = reinterpret_cast<void*>(remoteIns[remoteRank % deviceNumPerAggregation_]);
440 0 : void* remoteOut = reinterpret_cast<void*>(remoteOuts[remoteRank % deviceNumPerAggregation_]);
441 :
442 0 : CHK_PRT_RET(
443 : remoteIn == nullptr || remoteOut == nullptr,
444 : HCCL_ERROR(
445 : "[AicpuZeroCopyExchanger][UpdateTransportAddress] remoteRank in[%p] out[%p] is invalid", remoteIn,
446 : remoteOut),
447 : HCCL_E_INTERNAL);
448 0 : CHK_RET(link->UpdateRemoteAddr(remoteIn, remoteOut));
449 : }
450 :
451 0 : return HCCL_SUCCESS;
452 : }
453 :
454 0 : std::string AicpuZeroCopyExchanger::DumpLinkInfo(std::set<u32>& doneRanks)
455 : {
456 0 : std::string msg = "Expect:[";
457 0 : for (auto remoteRank : current_->remoteRanks) {
458 0 : msg += std::to_string(remoteRank) + " ";
459 : }
460 :
461 0 : msg += "] actual:[";
462 0 : for (auto remoteRank : doneRanks) {
463 0 : msg += std::to_string(remoteRank) + " ";
464 : }
465 0 : msg += "]";
466 :
467 0 : return msg;
468 0 : }
469 :
470 : } // namespace hccl
|
