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