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_cache_manager.h"
12 :
13 : #include "comm_utils.h"
14 : #include "transport_pub.h"
15 : #include "dispatcher.h"
16 : #include "dispatcher_aicpu_pub.h"
17 : #include "log.h"
18 : #include "task_logic_info_pub.h"
19 : #include "profiling_manager_device.h"
20 : #include "alltoall_utils_pub.h"
21 :
22 : namespace hccl {
23 32 : AicpuCacheManager::AicpuCacheManager() {
24 32 : HCCL_RUN_INFO("Construct AicpuCacheManager complete.");
25 32 : }
26 :
27 31 : AicpuCacheManager::~AicpuCacheManager() {
28 : // 释放算子展开的动态缓存 (if any)
29 31 : if (opUnfoldCachePtr_ != nullptr) {
30 11 : delete opUnfoldCachePtr_;
31 11 : opUnfoldCachePtr_ = nullptr;
32 : }
33 :
34 31 : HCCL_RUN_INFO("Destruct AicpuCacheManager success!");
35 31 : }
36 :
37 11 : HcclResult AicpuCacheManager::InitOpUnfoldCache()
38 : {
39 : // 创建算子展开的动态缓存 (不区分单算子/图模式)
40 11 : HCCL_INFO("[AicpuCacheManager][InitOpUnfoldCache] create aicpu cache for operator unfolding");
41 11 : opUnfoldCachePtr_ = (new (std::nothrow) OpUnfoldCache());
42 11 : CHK_PTR_NULL(opUnfoldCachePtr_);
43 :
44 11 : return HCCL_SUCCESS;
45 : }
46 :
47 0 : HcclResult AicpuCacheManager::LookupOpUnfoldCache(const std::string& algName, const OpParam ¶m,
48 : const AlgResourceResponse &algResource, bool& needExecute, bool& isCacheMiss,
49 : Stream& mainStream, std::vector<Stream>& slaveStreams, void *dispatcherPtr, const bool isDeviceMode,
50 : const HcclTopoInfo& topoinfo, std::unique_ptr<TopoMatcher>& topoMatcherPtr, const AlgOpContext& algContext,
51 : std::shared_ptr<AicpuZeroCopyExchanger>& zeroCopyExchangerPtr, const HcclWorkflowMode workflowMode,
52 : const DeviceMem& tinySendRecvMem, std::function<HcclResult()> setProfStartCallback)
53 : {
54 0 : needExecute = true;
55 0 : isCacheMiss = false;
56 :
57 0 : CHK_PTR_NULL(opUnfoldCachePtr_);
58 :
59 : // Dump main stream and slave streams addr and id for debug
60 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] mainStream with streamId[%u] and id[%u]",
61 : mainStream.GetHcclStreamInfo().actualStreamId, mainStream.id());
62 0 : for (size_t i = 0; i < slaveStreams.size(); ++i) {
63 0 : Stream &slaveStream = slaveStreams[i];
64 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] %uth slaveStream with streamId[%u] and id[%u]",
65 : i, slaveStream.GetHcclStreamInfo().actualStreamId, slaveStream.id());
66 : }
67 :
68 : // 判断是否需要cache
69 0 : bool needCache = false;
70 0 : CHK_RET(NeedOpUnfoldCache(algName, param, algResource, isDeviceMode, topoinfo, topoMatcherPtr, algContext, workflowMode, needCache));
71 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] needCache[%u]", needCache);
72 :
73 : // Cacheable算子
74 0 : if (needCache) {
75 : // 将streams中已有的task强制下发, 放置cache缓存跟算子编排无关的SQE
76 : // 注意: cache miss需要先强制下发, 避免缓存和算子展开无关的SQE; cache hit也需要强制下发, 否则LaunchNewTask只会下发cache里的, 而不会下发stream里的
77 0 : CHK_PTR_NULL(dispatcherPtr);
78 0 : CHK_RET(LaunchTaskExtend(dispatcherPtr, mainStream, slaveStreams));
79 :
80 : // 准备key
81 0 : OpUnfoldKey opUnfoldKey;
82 0 : CHK_RET(GetOpUnfoldKey(param, opUnfoldKey, topoinfo, algContext, workflowMode));
83 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] prepare key[%s] for op-unfold cache", opUnfoldKey.GetKeyString().c_str());
84 :
85 : // 准备 memory ranges
86 0 : std::vector<OpUnfoldMemRange> userInputMemRanges;
87 0 : std::vector<OpUnfoldMemRange> userOutputMemRanges;
88 0 : CHK_RET(PrepareUserMemRanges(param, algResource, userInputMemRanges, userOutputMemRanges,
89 : topoinfo, zeroCopyExchangerPtr, workflowMode, tinySendRecvMem));
90 :
91 : // 查找算子展开的动态缓存
92 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] look up op-unfold cache for key %s", opUnfoldKey.GetKeyString().c_str());
93 0 : OpUnfoldCacheEntry *entryPtr = nullptr;
94 0 : CHK_RET(opUnfoldCachePtr_->FindEntry(opUnfoldKey, &entryPtr));
95 0 : if (entryPtr != nullptr) { // Cache hit
96 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] cache hit for key %s", opUnfoldKey.GetKeyString().c_str());
97 :
98 : // 判断是否为alltoallv算子
99 0 : CHK_PTR_NULL(dispatcherPtr);
100 0 : CHK_PTR_NULL(setProfStartCallback);
101 0 : const bool profL1Enable = dfx::ProfilingManager::GetProfL1State(); // SQE-level profiling info
102 0 : if (IsAlltoallvType(param.opType)) { // alltoallv类算子, 需要额外的offset信息
103 : // 准备offset信息
104 0 : AlltoallvSendRecvInfo alltoallvSendRecvInfo;
105 0 : CHK_RET(PrepareAlltoallvSendRecvInfo(param, alltoallvSendRecvInfo, topoinfo));
106 :
107 : // 刷新缓存的SQE并直接下发到RTSQ
108 : // 注意: AicpuCacheManager下dispatcher一定是DispatcherAicpu
109 0 : (void)setProfStartCallback(); // Keep consistent with cache miss (调用kfcHandler for kSetProfTimeStart)
110 0 : CHK_RET((reinterpret_cast<DispatcherAiCpu *>(dispatcherPtr))->LaunchNewTask(
111 : entryPtr, userInputMemRanges, userOutputMemRanges, mainStream, slaveStreams, profL1Enable,
112 : true, alltoallvMetadata_, alltoallvSendRecvInfo));
113 0 : } else { // 非V类算子, 无需offset信息
114 : // 刷新缓存的SQE并直接下发到RTSQ
115 : // 注意: AicpuCacheManager下dispatcher一定是DispatcherAicpu
116 0 : (void)setProfStartCallback(); // Keep consistent with cache miss (调用kfcHandler for kSetProfTimeStart)
117 0 : CHK_RET((reinterpret_cast<DispatcherAiCpu *>(dispatcherPtr))->LaunchNewTask(
118 : entryPtr, userInputMemRanges, userOutputMemRanges, mainStream, slaveStreams, profL1Enable,
119 : false, AlltoallvMetadata(), AlltoallvSendRecvInfo()));
120 : }
121 :
122 : // 不需要执行算子展开的具体编排
123 0 : needExecute = false;
124 : } else { // Cache miss
125 0 : HCCL_INFO("[AicpuCacheManager][LookupOpUnfoldCache] cache miss for key %s", opUnfoldKey.GetKeyString().c_str());
126 :
127 : // alltoallv类算子需要传入metadata, 用于HCCL input buffer的扫描, 判断SQE addr字段对应的rank id用于后续地址刷新
128 0 : bool isAlltoallv = false;
129 0 : const AlltoallvMetadata *alltoallvMetadataPtr = nullptr;
130 0 : if (IsAlltoallvType(param.opType)) {
131 0 : isAlltoallv = true;
132 0 : alltoallvMetadataPtr = &alltoallvMetadata_;
133 0 : CHK_PTR_NULL(alltoallvMetadataPtr);
134 : }
135 :
136 : // 设置launch context, enable DispatcherAicpu在下发SQE时去执行cache admission
137 : // 注意: AicpuCacheManager下dispatcher一定是DispatcherAicpu
138 0 : CHK_RET((reinterpret_cast<DispatcherAiCpu *>(dispatcherPtr))->SetLaunchContext(
139 : opUnfoldKey, opUnfoldCachePtr_, userInputMemRanges, userOutputMemRanges, isAlltoallv, alltoallvMetadataPtr));
140 :
141 0 : isCacheMiss = true;
142 : }
143 0 : }
144 :
145 0 : return HCCL_SUCCESS;
146 : }
147 :
148 0 : HcclResult AicpuCacheManager::PreProcessForCacheMiss(const OpParam ¶m, std::unique_ptr<CollExecutorBase> &executor)
149 : {
150 : // 第一个需要cache的alltoallv类算子
151 : // 注意: 只有当alltoallv的algName为"RunAlltoAllDirectFullmesh"时, 才会进入cache, 所以使用的一定是CollRunAlltoAllDirectFullmesh executor
152 0 : if (IsAlltoallvType(param.opType)) {
153 0 : CHK_PTR_NULL(executor.get());
154 0 : HCCL_INFO("[AicpuCacheManager][PreProcessForCacheMiss] mark NeedAlltoallvCache for CollRunAlltoAllDirectFullmesh");
155 0 : CHK_RET(executor->MarkNeedAlltoallvCache());
156 : }
157 :
158 0 : return HCCL_SUCCESS;
159 : }
160 :
161 0 : HcclResult AicpuCacheManager::PostProcessForCacheMiss(const OpParam ¶m, std::unique_ptr<CollExecutorBase> &executor,
162 : Stream& mainStream, std::vector<Stream>& slaveStreams, void* dispatcherPtr, const HcclTopoInfo& topoinfo,
163 : const AlgOpContext& algContext, const HcclWorkflowMode workflowMode)
164 : {
165 : // Cache miss会设置launch context to enable cache admission -> 需要清理launch context, DispatcherAicpu不会再admit当前算子后续展开的SQE
166 : // 注意: AicpuCacheManager下dispatcher一定是DispatcherAicpu
167 0 : CHK_RET((reinterpret_cast<DispatcherAiCpu *>(dispatcherPtr))->ClearLaunchContext());
168 :
169 : // 准备key
170 0 : OpUnfoldKey opUnfoldKey;
171 0 : CHK_RET(GetOpUnfoldKey(param, opUnfoldKey, topoinfo, algContext, workflowMode));
172 :
173 : // 校验cache entry (post cache miss前的orchestrate一定会add new cache entry)
174 0 : OpUnfoldCacheEntry *entryPtr = nullptr;
175 0 : CHK_PTR_NULL(opUnfoldCachePtr_);
176 0 : CHK_RET(opUnfoldCachePtr_->FindEntry(opUnfoldKey, &entryPtr));
177 0 : CHK_PTR_NULL(entryPtr); // Cache miss后刚刚admit的cache entry
178 :
179 : // 根据cache中的streamid计算是主流还是第几个从流
180 0 : HCCL_INFO("[AicpuCacheManager][PostProcessForCacheMiss] calculate stream seq idxes for a newly-admitted entry of key %s",
181 : opUnfoldKey.GetKeyString().c_str());
182 0 : CHK_RET(entryPtr->CalcStreamSeqIdxes(mainStream, slaveStreams));
183 :
184 : // 针对alltoallv类算子, cache miss后处理
185 0 : if (IsAlltoallvType(param.opType)) {
186 : // 第一个需要cache的alltoallv类算子
187 : // 注意: 同一个通信域下, alltoallv类算子展开得到的hcclOffsetDstRanksMap是相同的, 所以只需要初始化一次
188 0 : if (!isInitAlltoallvMetadata_) {
189 : // 获得hcclOffset-dstRanks mapping
190 : // 注意: 只有当alltoallv的algName为"RunAlltoAllDirectFullmesh"时, 才会进入cache, 所以使用的一定是CollRunAlltoAllDirectFullmesh executor
191 0 : CHK_PTR_NULL(executor.get());
192 0 : HCCL_INFO("[AicpuCacheManager][PostProcessForCacheMiss] get hcclOffset-dstRank mapping of key[%s] for CollRunAlltoAllDirectFullmesh",
193 : opUnfoldKey.GetKeyString().c_str());
194 0 : std::unordered_map<uint64_t, std::vector<uint32_t>> hcclOffsetDstRanksMap;
195 0 : CHK_RET(executor->GetHcclOffsetDstRanksMap(hcclOffsetDstRanksMap));
196 0 : for (std::unordered_map<uint64_t, std::vector<uint32_t>>::const_iterator mapIter = hcclOffsetDstRanksMap.cbegin();
197 0 : mapIter != hcclOffsetDstRanksMap.cend(); ++mapIter) {
198 0 : alltoallvMetadata_.hcclOffsetDstRanksIdxMap.emplace(mapIter->first, std::make_pair(mapIter->second, 0));
199 : }
200 :
201 : // Dump hcclOffset-dstRanks mapping
202 0 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
203 0 : for (std::unordered_map<uint64_t, std::vector<uint32_t>>::const_iterator mapIter = hcclOffsetDstRanksMap.cbegin();
204 0 : mapIter != hcclOffsetDstRanksMap.cend(); ++mapIter) {
205 0 : HCCL_INFO("[AicpuCacheManager][PostProcessForCacheMiss] hcclOffset[%llu]-dstRanks.size[%u]",
206 : mapIter->first, mapIter->second.size());
207 0 : for (uint32_t i = 0; i < mapIter->second.size(); ++i) {
208 0 : HCCL_INFO("[AicpuCacheManager][PostProcessForCacheMiss] dstRanks[%u]: %u", i, mapIter->second[i]);
209 : }
210 : }
211 : }
212 :
213 0 : isInitAlltoallvMetadata_ = true;
214 0 : }
215 :
216 : // 根据hcclOffset-dstRank mapping更新PrepareIntraData case下dstRefreshInfo中的rank id
217 0 : CHK_RET(entryPtr->UpdateRefreshAddrInfoForAlltoallv(topoinfo.userRank, alltoallvMetadata_));
218 : }
219 :
220 0 : return HCCL_SUCCESS;
221 : }
222 :
223 0 : HcclResult AicpuCacheManager::ClearOpUnfoldCacheEntry(const std::string& algName, const OpParam ¶m,
224 : const AlgResourceResponse& algResource, const bool isDeviceMode, const HcclTopoInfo& topoinfo,
225 : std::unique_ptr<TopoMatcher>& topoMatcherPtr, const AlgOpContext& algContext, const HcclWorkflowMode workflowMode)
226 : {
227 : // 清理当前aicpu算子对应的cache entry, 避免异常状态下, cache命中
228 0 : CHK_PTR_NULL(opUnfoldCachePtr_);
229 :
230 : // 判断是否需要cache
231 0 : bool needCache = false;
232 0 : CHK_RET(NeedOpUnfoldCache(algName, param, algResource, isDeviceMode, topoinfo, topoMatcherPtr, algContext, workflowMode, needCache));
233 0 : HCCL_INFO("[AicpuCacheManager][ClearOpUnfoldCacheEntry] needCache[%u]", needCache);
234 :
235 : // Cacheable算子
236 0 : if (needCache) {
237 : // 准备key
238 0 : OpUnfoldKey opUnfoldKey;
239 0 : CHK_RET(GetOpUnfoldKey(param, opUnfoldKey, topoinfo, algContext, workflowMode));
240 :
241 : // 清理cache entry if any
242 0 : HCCL_RUN_INFO("[AicpuCacheManager][ClearOpUnfoldCacheEntry] try to clear cache entry for key[%s]",
243 : opUnfoldKey.GetKeyString().c_str());
244 0 : CHK_RET(opUnfoldCachePtr_->ClearEntry(opUnfoldKey));
245 : }
246 :
247 : // 针对alltoallv算子的缓存进行metadata清理
248 : // 注意: 即使当前故障算子不是alltoallv类算子, 由于NS快恢可能会重新分配资源 (例如hccl input / notify/ signal),
249 : // 为了保证alltoallvMetadata_的正确性, 必须重新计算并初始化alltoallvMetadata_
250 0 : CHK_RET(ClearMetadataForFirstAlltoallv());
251 :
252 : // 清理alltoallv类算子的cache entry
253 : // 注意: 为了保证NS快恢/重执行后, 必定进入alltoallvMetadata_重新计算和初始化的流程, 需要清理与alltoallv类算子相关的entry,
254 : // 但对aicpu cache影响有限, 因为alltoallv类算子的cache entry数量有限 (只区分opType/isBigCount), 所以性能影响有限
255 0 : CHK_RET(opUnfoldCachePtr_->ClearEntryForAlltoallv());
256 :
257 0 : return HCCL_SUCCESS;
258 : }
259 :
260 0 : HcclResult AicpuCacheManager::ClearMetadataForFirstAlltoallv()
261 : {
262 : // 确保故障/重执行后第一次可能被cache的alltoallv算子仍然会重新计算/初始化metadata
263 0 : isCalcAlltoallvMetadata_ = false;
264 0 : isInitAlltoallvMetadata_ = false;
265 0 : alltoallvMetadata_.Clear();
266 :
267 0 : return HCCL_SUCCESS;
268 : }
269 :
270 0 : HcclResult AicpuCacheManager::NeedOpUnfoldCache(const std::string& algName, const OpParam ¶m,
271 : const AlgResourceResponse& algResource, const bool isDeviceMode, const HcclTopoInfo& topoinfo,
272 : std::unique_ptr<TopoMatcher>& topoMatcherPtr, const AlgOpContext& algContext, const HcclWorkflowMode workflowMode,
273 : bool& needCache) {
274 : // 初始化为不需要op-unfold cache
275 0 : needCache = false;
276 :
277 : // 检查cache容量
278 0 : if (opUnfoldCachePtr_->IsCacheFull()) {
279 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] cache is full, disable cache for current operator");
280 0 : return HCCL_SUCCESS;
281 : }
282 :
283 : // 校验环境变量
284 0 : if (param.aicpuCacheEnable == 0) {
285 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] disable aicpu cache for aicpuCacheEnable[%u]", param.aicpuCacheEnable);
286 0 : return HCCL_SUCCESS;
287 : }
288 :
289 : // 屏蔽MC2算子
290 0 : if (isDeviceMode) {
291 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] MC2 op is not supported for operator unfolding cache");
292 0 : return HCCL_SUCCESS;
293 : }
294 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] device mode is not MC2 op");
295 :
296 : // 判断当前通信域是否使用RDMA (例如跨超通信域), 使用则不cache (因为RoCE队列的WQE不可见)
297 0 : const std::unordered_map<u32, bool>& isUsedRdmaMap = topoinfo.isUsedRdmaMap;
298 0 : for (std::unordered_map<u32, bool>::const_iterator map_iter = isUsedRdmaMap.cbegin(); map_iter != isUsedRdmaMap.end(); ++map_iter) {
299 0 : if (map_iter->second) {
300 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] rank[%u] uses RDMA -> not supported for operator unfolding cache",
301 : map_iter->first);
302 0 : return HCCL_SUCCESS;
303 : }
304 : }
305 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] all ranks do not use RDMA");
306 :
307 : // 屏蔽inplace场景
308 0 : bool isInplace = false;
309 0 : CHK_RET(IsInplace(param, isInplace, topoinfo));
310 0 : if (isInplace) {
311 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] inplace case is not supported for operator unfolding cache");
312 0 : return HCCL_SUCCESS;
313 : }
314 :
315 : // 目前V类算子、batch类型算子、以及send/recv不考虑动态缓存 (使用白名单而非黑名单管理, 避免非预期算子进入cache机制)
316 : // 注意: 如果想要通过比较缓存刷新后的SQE与正常算子展开的SQE来debug, 可以将想要比较的算子从以下的cache白名单中移除, 重新打包运行
317 0 : const HcclCMDType opType = param.opType;
318 0 : if (opType == HcclCMDType::HCCL_CMD_BROADCAST ||
319 0 : opType == HcclCMDType::HCCL_CMD_REDUCE ||
320 0 : opType == HcclCMDType::HCCL_CMD_ALLGATHER ||
321 0 : opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER ||
322 0 : opType == HcclCMDType::HCCL_CMD_ALLTOALL ||
323 0 : opType == HcclCMDType::HCCL_CMD_SCATTER ||
324 : opType == HcclCMDType::HCCL_CMD_ALLREDUCE) { // 非V类算子
325 0 : if (algName == "RunAlltoAllVStaged" || algName == "RunAlltoAllVFullMesh") {
326 0 : HCCL_INFO("[AicpuCacheManager][%s] algName[%s] is not supported for unfolding cache", __func__, algName.c_str());
327 0 : return HCCL_SUCCESS;
328 : }
329 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] opType[%d] is supported for operator unfolding cache", opType);
330 0 : needCache = true;
331 0 : } else if (IsAlltoallvType(opType)) { // alltoallv类算子
332 : // 注意: 暂不支持alltoallv 图模式 / 存在强制单算子模式转换 (即图模式建链+单算子模式展开)
333 : // 原因: 这两个场景下, alltoallv每次执行会重新建链, 导致remote ranks' hccl buffer在本rank映射的虚拟地址发生变化;
334 : // 当前如果src是(local) user input, dst是local user output, 会当做LocalCopy根据recv offset进行src addr刷新;
335 : // 而图模式建链下, remote hccl input会作为remote user input传入cache, remote copy由原来的remote hccl input ->
336 : // local user output变成(remote) user input -> local user output, 需要根据hccl offset进行src addr刷新
337 : // 结论: 由于现网下基本不存在alltoallv图模式调用, 暂对该场景不使能aicpu cache;
338 : // 如果需要支持, 应当在fullmesh算法中拦截srcRank-hcclOffset的映射, 并通过AlltoallvMetadata传入cache;
339 : // 识别到remote user input -> local user output时, 判断为图模式下的RemoteCopy;
340 : // 根据remote user input baseaddr (即remote hccl input在本rank映射的VA) + hccl offset更新src addr
341 0 : if (workflowMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB || // 图模式
342 0 : param.aicpuCacheEnable > FORCE_OP_BASE_DELTA) { // 存在强制单算子模式转换 (即图模式建链+单算子模式展开)
343 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] graph mode[%u, %u > %u] is not supported for alltoallv's cache",
344 : workflowMode, param.aicpuCacheEnable, FORCE_OP_BASE_DELTA);
345 0 : return HCCL_SUCCESS;
346 : }
347 :
348 : // 注意: 假设CollRunAlltoAllDirectFullmesh一定只使用AlltoAllVDirectFullMesh作为algTemp
349 0 : if (algName != "RunAlltoAllDirectFullmesh") {
350 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] algName[%s] is not supported for alltoallv's cache", algName.c_str());
351 0 : return HCCL_SUCCESS;
352 : }
353 :
354 0 : if (!isCalcAlltoallvMetadata_) { // 当前通信域下第一次可能被cache的alltoallv, 需要计算相应metadata
355 0 : CHK_RET(CalcMetadataForFirstAlltoallv(algResource, isDeviceMode, topoinfo, topoMatcherPtr, algContext));
356 :
357 : // 后续不再重复计算alltoallv metadata
358 : // 注意: (i) 虽然alltoallvMetadata_中的相关mapping还未被初始化, 但isCalcAlltoallvMetadata_只是为了避免重复计算部分metadata
359 : // (ii) 参考CalcMetadataForFirstAlltoallv, 例如sdmaDataBlockSize, hcclInputMemRanges, notifyIdRankRflagMap等
360 : // 如果有故障发生:
361 : // (i) 发生在isCalcAlltoallvMetadata_ = true前, 则ClearOpUnfoldCacheEntry会重新计算sdmaDataBlockSize来判断是否需要清理cache entry
362 : // (ii) 发生在设置true后, 即使在PostProcessForCacheMiss初始化相关mapping前, 也不影响清理时needCache的判断 (只依赖sdmaDataBlockSize)
363 0 : isCalcAlltoallvMetadata_ = true;
364 : }
365 :
366 : // 判断是否为小数据量的alltoallv类算子
367 0 : bool isSmallData = false;
368 0 : CHK_RET(IsSmallDataAlltoallv(param, isSmallData, topoinfo));
369 0 : if (!isSmallData) {
370 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] large-data alltoallv[%u] is not supported for operator unfolding cache", opType);
371 0 : return HCCL_SUCCESS;
372 : }
373 :
374 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] small-data alltoallv[%u] is supported for operator unfolding cache", opType);
375 0 : needCache = true;
376 : } else {
377 0 : HCCL_INFO("[AicpuCacheManager][NeedOpUnfoldCache] opType[%d] is not supported for operator unfolding cache", opType);
378 0 : return HCCL_SUCCESS;
379 : }
380 :
381 : // 到这里needCache应该为true (如果为false则已经提前返回了)
382 0 : CHK_PRT_RET(!needCache, HCCL_ERROR("[AicpuCacheManager][NeedOpUnfoldCache] needCache should be true"), HCCL_E_INTERNAL);
383 :
384 0 : return HCCL_SUCCESS;
385 : }
386 :
387 0 : HcclResult AicpuCacheManager::IsInplace(const OpParam ¶m, bool& isInplace, const HcclTopoInfo& topoinfo)
388 : {
389 : // 准备input/output size
390 0 : HcclDataType sendType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
391 0 : HcclDataType recvType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
392 0 : uint64_t inputSize = 0;
393 0 : uint64_t outputSize = 0;
394 0 : CHK_RET(ParseOpParamForCache(param, sendType, recvType, inputSize, outputSize, topoinfo));
395 : UNUSED_PARAM(sendType);
396 : UNUSED_PARAM(recvType);
397 :
398 : // 注意: alltoall/alltoallv/alltoallvc可能存在inputSize/outputSize为0的情况, 导致不分配user input/output
399 : // 但会使用tinySendRecvMem_更新algResource.paramInput/OutputMem用于建链, 导致cache无法区分给定地址字段的地址类型
400 : // 参考aicpu_communicator.cc中的SetAlltoAllInputAndOutPutMem
401 0 : if (inputSize == 0 && outputSize == 0) {
402 0 : isInplace = true;
403 0 : HCCL_INFO("[AicpuCacheManager][IsInplace] inputSize[%u] is overlapping with outputSize[%u]",
404 : inputSize, outputSize);
405 0 : return HCCL_SUCCESS;
406 : }
407 :
408 0 : if (inputSize == 0 || outputSize == 0) {
409 0 : isInplace = false;
410 0 : HCCL_INFO("[AicpuCacheManager][IsInplace] inputSize[%u] is not overlapping with outputSize[%u]",
411 : inputSize, outputSize);
412 0 : return HCCL_SUCCESS;
413 : }
414 :
415 0 : const uint64_t inputStart = reinterpret_cast<uint64_t>(param.inputPtr);
416 0 : const uint64_t inputEnd = inputStart + inputSize - 1;
417 0 : const uint64_t outputStart = reinterpret_cast<uint64_t>(param.outputPtr);
418 0 : const uint64_t outputEnd = outputStart + outputSize - 1;
419 :
420 0 : if (inputStart <= outputEnd && outputStart <= inputEnd) {
421 0 : isInplace = true;
422 0 : HCCL_INFO("[AicpuCacheManager][IsInplace] input[0x%016llx, 0x%016llx] is overlapping with output[0x%016llx, 0x%016llx]",
423 : inputStart, inputEnd, outputStart, outputEnd);
424 : } else {
425 0 : isInplace = false;
426 0 : HCCL_INFO("[AicpuCacheManager][IsInplace] input[0x%016llx, 0x%016llx] is not overlapping with output[0x%016llx, 0x%016llx]",
427 : inputStart, inputEnd, outputStart, outputEnd);
428 : }
429 :
430 0 : return HCCL_SUCCESS;
431 : }
432 :
433 0 : bool AicpuCacheManager::IsAlltoallvType(const HcclCMDType opType)
434 : {
435 : // alltoallv/alltoallvc只是对上接口不同, 实际算法编排相同, 均视为alltoallv类型的算子
436 0 : if (opType == HcclCMDType::HCCL_CMD_ALLTOALLV || opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) {
437 0 : return true;
438 : }
439 0 : return false;
440 : }
441 :
442 0 : HcclResult AicpuCacheManager::IsSmallDataAlltoallv(const OpParam ¶m, bool& isSmallData, const HcclTopoInfo& topoinfo)
443 : {
444 : // 根据SDMA data block size判断是否需要cache
445 : // 参考coll_all_to_all_v_direct_fullmesh_executor.cc下的CollRunAlltoAllDirectFullmesh::GetLocalSendRecvInfoforAlltoallV
446 0 : const uint32_t rankSize = topoinfo.userRankSize;
447 0 : if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) { // alltoallv
448 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] check %u send/recv counts", rankSize);
449 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
450 0 : const uint64_t curSendCounts = *(static_cast<const uint64_t *>(param.All2AllDataDes.sendCounts) + tmpRank);
451 0 : const uint64_t curSendLength = curSendCounts * SIZE_TABLE[param.All2AllDataDes.sendType];
452 : // 如果curSendLength超过SDMA data block size (即alltoallv需要切step), 或者超过HCCL_SDMA_MAX_COUNT_4GB (即MemcpyAsync需要切split), 不做cache
453 0 : if (curSendLength > alltoallvMetadata_.sdmaDataBlockSize || curSendLength > HCCL_SDMA_MAX_COUNT_4GB) {
454 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] large-sdata alltoallv[%u]: userRank[%u] tmpRank[%u]"\
455 : "curSendLength[%u] sdmaDataBlockSize[%u] 4GB[%u]",
456 : param.opType, topoinfo.userRank, tmpRank, curSendLength, alltoallvMetadata_.sdmaDataBlockSize, HCCL_SDMA_MAX_COUNT_4GB);
457 0 : isSmallData = false;
458 0 : return HCCL_SUCCESS;
459 : }
460 :
461 0 : const uint64_t curRecvCounts = *(static_cast<const uint64_t *>(param.All2AllDataDes.recvCounts) + tmpRank);
462 0 : const uint64_t curRecvLength = curRecvCounts * SIZE_TABLE[param.All2AllDataDes.recvType];
463 : // 如果curRecvLength超过SDMA data block size (即alltoallv需要切step), 或者超过HCCL_SDMA_MAX_COUNT_4GB (即MemcpyAsync需要切split), 不做cache
464 0 : if (curRecvLength > alltoallvMetadata_.sdmaDataBlockSize || curRecvLength > HCCL_SDMA_MAX_COUNT_4GB) {
465 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] large-rdata alltoallv[%u]: userRank[%u] tmpRank[%u]"\
466 : "curRecvLength[%u] sdmaDataBlockSize[%u] 4GB[%u]",
467 : param.opType, topoinfo.userRank, tmpRank, curRecvLength, alltoallvMetadata_.sdmaDataBlockSize, HCCL_SDMA_MAX_COUNT_4GB);
468 0 : isSmallData = false;
469 0 : return HCCL_SUCCESS;
470 : }
471 : }
472 0 : } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) { // alltoallvc
473 0 : const uint32_t curRank = topoinfo.userRank;
474 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] check %u-size sendCountMatrix", rankSize);
475 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
476 0 : const uint64_t curSendCounts = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix) + curRank * rankSize + tmpRank); // sendCountMatrix[curRank][tmpRank]
477 0 : const uint64_t curSendLength = curSendCounts * SIZE_TABLE[param.All2AllDataDes.sendType];
478 : // 如果curSendLength超过SDMA data block size (即alltoallv需要切step), 或者超过HCCL_SDMA_MAX_COUNT_4GB (即MemcpyAsync需要切split), 不做cache
479 0 : if (curSendLength > alltoallvMetadata_.sdmaDataBlockSize || curSendLength > HCCL_SDMA_MAX_COUNT_4GB) {
480 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] large-sdata alltoallvc[%u]: userRank[%u] tmpRank[%u]"\
481 : "curSendLength[%u] sdmaDataBlockSize[%u] 4GB[%u]",
482 : param.opType, curRank, tmpRank, curSendLength, alltoallvMetadata_.sdmaDataBlockSize, HCCL_SDMA_MAX_COUNT_4GB);
483 0 : isSmallData = false;
484 0 : return HCCL_SUCCESS;
485 : }
486 :
487 0 : const uint64_t curRecvCounts = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix) + tmpRank * topoinfo.userRankSize + curRank); // sendCountMatrix[tmpRank][curRank]
488 0 : const uint64_t curRecvLength = curRecvCounts * SIZE_TABLE[param.All2AllDataDes.recvType];
489 : // 如果curRecvLength超过SDMA data block size (即alltoallv需要切step), 或者超过HCCL_SDMA_MAX_COUNT_4GB (即MemcpyAsync需要切split), 不做cache
490 0 : if (curRecvLength > alltoallvMetadata_.sdmaDataBlockSize || curRecvLength > HCCL_SDMA_MAX_COUNT_4GB) {
491 0 : HCCL_INFO("[AicpuCacheManager][IsSmallDataAlltoallv] large-rdata alltoallvc[%u]: userRank[%u] tmpRank[%u]"\
492 : "curRecvLength[%u] sdmaDataBlockSize[%u] 4GB[%u]",
493 : param.opType, topoinfo.userRank, tmpRank, curRecvLength, alltoallvMetadata_.sdmaDataBlockSize, HCCL_SDMA_MAX_COUNT_4GB);
494 0 : isSmallData = false;
495 0 : return HCCL_SUCCESS;
496 : }
497 : }
498 : } else {
499 0 : HCCL_ERROR("[AicpuCacheManager][IsSmallDataAlltoallv] invalid opType[%u] for alltoallv", param.opType);
500 0 : return HCCL_E_INTERNAL;
501 : }
502 :
503 0 : isSmallData = true;
504 0 : return HCCL_SUCCESS;
505 : }
506 :
507 0 : HcclResult AicpuCacheManager::CalcMetadataForFirstAlltoallv(const AlgResourceResponse& algResource,
508 : const bool isDeviceMode, const HcclTopoInfo& topoinfo,
509 : std::unique_ptr<TopoMatcher>& topoMatcherPtr, const AlgOpContext& algContext)
510 : {
511 0 : alltoallvMetadata_.Clear();
512 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] clear alltoallv metadata before calc.");
513 :
514 : // Part 1: 计算SDMA data block size, 用于根据数据量判断是否需要cache
515 : // 参考alltoallv_direct_fullmesh.cc下的AlltoAllVDirectFullMesh::Prepare (当前alltoallv算子只会使用direct full mesh算法)
516 :
517 : // 获取local pod中的device数量
518 : // 参考coll_all_to_all_v_direct_fullmesh_executor.cc下的CollRunAlltoAllDirectFullmesh::GetLocalSDMAGroupInfo
519 0 : uint32_t devNumInlocalPod = 0;
520 0 : uint32_t rankIdxInPod = 0;
521 0 : const bool isA2MultiModule = topoinfo.deviceType == DevType::DEV_TYPE_910B && !topoinfo.isSingleMeshAggregation;
522 0 : if (topoMatcherPtr->GetExternalInputInterHccsDisable() || isA2MultiModule) {
523 0 : CHK_RET(topoMatcherPtr->GetLocalServerRankSize(topoinfo.userRank, devNumInlocalPod, rankIdxInPod));
524 : } else {
525 0 : CHK_RET(topoMatcherPtr->GetLocalSuperPodRankSize(topoinfo.userRank, devNumInlocalPod, rankIdxInPod));
526 : }
527 0 : CHK_PRT_RET(devNumInlocalPod == INVALID_VALUE_RANKSIZE,
528 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] get local superPod total ranksize failed."),
529 : HCCL_E_PARA);
530 : UNUSED_PARAM(rankIdxInPod);
531 :
532 : // 计算SDMA在alltoallv下的最大并发数量
533 0 : const uint32_t sdmaConcurrentNum = (devNumInlocalPod > ALLTOALLV_DIRECT_FULLMESH_SDMA_CONCURRENT_SIZE) ?
534 0 : (ALLTOALLV_DIRECT_FULLMESH_SDMA_CONCURRENT_SIZE) : (devNumInlocalPod);
535 :
536 : // 注意: MC2算子不会进入cache, 不需要根据MC2 stepSize调整sdmaConcurrentNum
537 0 : CHK_PRT_RET(algContext.mc2Handler.stepSize > 0,
538 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] isDeviceMode[%u] mc2Handler.stepSize[%u]",
539 : isDeviceMode, algContext.mc2Handler.stepSize),
540 : HCCL_E_INTERNAL);
541 :
542 : // 计算SDMA data block大小
543 0 : constexpr uint32_t blockGroup = 2;
544 0 : alltoallvMetadata_.sdmaDataBlockSize = (algResource.cclInputMem.size() / std::max(1u, sdmaConcurrentNum * blockGroup));
545 :
546 : // 向下对齐到16k Byte
547 0 : if (alltoallvMetadata_.sdmaDataBlockSize > HCCL_MIN_SLICE_ALIGN_910B) {
548 0 : alltoallvMetadata_.sdmaDataBlockSize = (alltoallvMetadata_.sdmaDataBlockSize / HCCL_MIN_SLICE_ALIGN_910B) * HCCL_MIN_SLICE_ALIGN_910B;
549 : }
550 :
551 : // sdmaDataBlockSize应该大于0
552 0 : CHK_PRT_RET(alltoallvMetadata_.sdmaDataBlockSize== 0,
553 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] sdmaDataBlockSize is zero."),
554 : HCCL_E_INTERNAL);
555 :
556 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] first alltoallv, devNumInlocalPod[%u],"\
557 : "sdmaConcurrentNum[%u] cclInputSize[%u] sdmaDataBlockSize[%u]",
558 : devNumInlocalPod, sdmaConcurrentNum, algResource.cclInputMem.size(), alltoallvMetadata_.sdmaDataBlockSize);
559 :
560 : // Part 2: 计算每个rank的HCCL input buffer memory range
561 : // 参考alltoallv_direct_fullmesh.cc下的SDMAwithRemoteRankAndNotifyEnd, coll_all_to_all_v_direct_fullmesh_executor.cc下的KernelRun,
562 : // 和coll_native_executor_base.cc下的GetSubCommInfo
563 :
564 : // 初始化hcclInputMemRanges
565 0 : const uint32_t rankSize = topoinfo.userRankSize;
566 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] prepare %u hccl input memory ranges for op-unfold cache", rankSize);
567 0 : std::vector<OpUnfoldMemRange>& hcclInputMemRanges = alltoallvMetadata_.hcclInputMemRanges;
568 0 : hcclInputMemRanges.resize(rankSize);
569 :
570 : // 准备hccl input size
571 : // 来自于cclInputBuffer_ (size来自于HcclOpResParam commParam.winSize, 单位是bytes)
572 0 : const uint64_t hcclInputSize = algResource.cclInputMem.size();
573 :
574 : // 设置当前rank的hccl input memory range
575 0 : const uint32_t curRank = topoinfo.userRank; // NOTE: 不应该使用param.srcRank (某些算子始终为0)
576 0 : CHK_PRT_RET(curRank >= rankSize,
577 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] invalid curRank %u >= rankSize %u", curRank, rankSize),
578 : HCCL_E_INTERNAL);
579 0 : OpUnfoldMemRange& curHcclInputMemRange = hcclInputMemRanges[curRank];
580 0 : curHcclInputMemRange.isValid = true;
581 0 : curHcclInputMemRange.baseAddr = reinterpret_cast<uint64_t>(algResource.cclInputMem.ptr());
582 0 : curHcclInputMemRange.memSize = hcclInputSize; // NOTE: 不应该使用param.inputSize (user memory input size, 且alltoall类始终为0)
583 :
584 : // 设置其他rank的hccl input memory range
585 0 : const std::vector<LINK>& links = algResource.opTransportResponse[COMM_COMBINE_ORDER][COMM_INDEX_0].links;
586 0 : CHK_PRT_RET(links.size() != rankSize,
587 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] links.size[%u] != rankSize[%u]", links.size(), rankSize),
588 : HCCL_E_INTERNAL);
589 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
590 0 : if (tmpRank == curRank) {
591 0 : continue;
592 : }
593 :
594 : // 获取curRank与tmpRank之间的LINK
595 0 : const LINK& intraNeighboorTransport = links[tmpRank];
596 0 : CHK_PTR_NULL(intraNeighboorTransport);
597 :
598 : // 获取tmpRank的hccl input memory baseaddr
599 0 : void *tmpHcclInputBaseAddr = nullptr;
600 0 : CHK_RET(intraNeighboorTransport->GetRemoteMem(UserMemType::INPUT_MEM, &tmpHcclInputBaseAddr));
601 0 : CHK_PTR_NULL(tmpHcclInputBaseAddr);
602 :
603 : // 设置tmpRank对应的hccl input memory range
604 0 : OpUnfoldMemRange& tmpHcclInputMemRange = hcclInputMemRanges[tmpRank];
605 0 : tmpHcclInputMemRange.isValid = true;
606 0 : tmpHcclInputMemRange.baseAddr = reinterpret_cast<uint64_t>(tmpHcclInputBaseAddr);
607 0 : tmpHcclInputMemRange.memSize = hcclInputSize;
608 : }
609 :
610 : // 打印debug信息
611 0 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
612 0 : for (size_t rankId = 0; rankId < hcclInputMemRanges.size(); ++rankId) {
613 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] hcclInputMemRanges[%u] isValid: %d,"\
614 : "baseAddr: 0x%016llx, memSize: %llu",
615 : rankId, hcclInputMemRanges[rankId].isValid, hcclInputMemRanges[rankId].baseAddr, hcclInputMemRanges[rankId].memSize);
616 : }
617 : }
618 :
619 : // Part 3: 计算notifyId与remoteRank间的映射
620 :
621 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] prepare %u notify info for op-unfold cache", rankSize - 1);
622 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
623 0 : if (tmpRank == curRank) {
624 0 : continue;
625 : }
626 :
627 : // 获取curRank与tmpRank之间的LINK
628 0 : const LINK& intraNeighboorTransport = links[tmpRank];
629 0 : CHK_PTR_NULL(intraNeighboorTransport);
630 :
631 : // 获取RxAck相关的NotifyId (与recv count相关)
632 : HcclSignalInfo recvNotifyInfo;
633 0 : bool recvIsValid = false;
634 0 : CHK_RET(intraNeighboorTransport->GetSpecificNotify(recvNotifyInfo, recvIsValid, "localSendDone"));
635 0 : CHK_PRT_RET(!recvIsValid,
636 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] invalid localSendDoneNotify_"),
637 : HCCL_E_INTERNAL);
638 0 : const uint32_t recvNotifyId = static_cast<uint32_t>(recvNotifyInfo.resId);
639 0 : alltoallvMetadata_.notifyIdRankRflagMap.emplace(recvNotifyId, std::make_pair(tmpRank, true));
640 :
641 : // 获取RxDataSignal相关的NotifyId (与send count相关)
642 : HcclSignalInfo sendNotifyInfo;
643 0 : bool sendIsValid = false;
644 0 : CHK_RET(intraNeighboorTransport->GetSpecificNotify(sendNotifyInfo, sendIsValid, "localSendReady"));
645 0 : CHK_PRT_RET(!sendIsValid,
646 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] invalid localSendReadyNotify_"),
647 : HCCL_E_INTERNAL);
648 0 : const uint32_t sendNotifyId = static_cast<uint32_t>(sendNotifyInfo.resId);
649 0 : alltoallvMetadata_.notifyIdRankRflagMap.emplace(sendNotifyId, std::make_pair(tmpRank, false));
650 : }
651 :
652 : // Part 4: 计算signalAddr与remoteRank的映射
653 :
654 0 : HCCL_INFO("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] prepare %u signal info for op-unfold cache", rankSize - 1);
655 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
656 0 : if (tmpRank == curRank) {
657 0 : continue;
658 : }
659 :
660 : // 获取curRank与tmpRank之间的LINK
661 0 : const LINK& intraNeighboorTransport = links[tmpRank];
662 0 : CHK_PTR_NULL(intraNeighboorTransport);
663 :
664 : // 获取TxDataSignal相关的SignalAddr (与recv count相关)
665 : HcclSignalInfo recvNotifyInfo;
666 0 : bool recvIsValid = false;
667 0 : CHK_RET(intraNeighboorTransport->GetSpecificNotify(recvNotifyInfo, recvIsValid, "remoteSendReady"));
668 0 : CHK_PRT_RET(!recvIsValid,
669 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] invalid remoteSendReadyNotify_"),
670 : HCCL_E_INTERNAL);
671 0 : const uint64_t recvSignalAddr = recvNotifyInfo.addr;
672 0 : alltoallvMetadata_.signalAddrRankRflagMap.emplace(recvSignalAddr, std::make_pair(tmpRank, true));
673 :
674 : // 获取TxAck相关的SignalAddr (与send count相关)
675 : HcclSignalInfo sendNotifyInfo;
676 0 : bool sendIsValid = false;
677 0 : CHK_RET(intraNeighboorTransport->GetSpecificNotify(sendNotifyInfo, sendIsValid, "remoteSendDone"));
678 0 : CHK_PRT_RET(!sendIsValid,
679 : HCCL_ERROR("[AicpuCacheManager][CalcMetadataForFirstAlltoallv] invalid remoteSendDoneNotify_"),
680 : HCCL_E_INTERNAL);
681 0 : const uint64_t sendSignalAddr = sendNotifyInfo.addr;
682 0 : alltoallvMetadata_.signalAddrRankRflagMap.emplace(sendSignalAddr, std::make_pair(tmpRank, false));
683 : }
684 :
685 0 : return HCCL_SUCCESS;
686 : }
687 :
688 0 : HcclResult AicpuCacheManager::GetOpUnfoldKey(const OpParam ¶m, OpUnfoldKey& key, const HcclTopoInfo& topoinfo,
689 : const AlgOpContext& algContext, const HcclWorkflowMode workflowMode)
690 : {
691 : // 注意: 由于GetOpUnfoldKey前已经做过NeedOpUnfoldCache检查, 这里不再做重复检验
692 :
693 : // 准备sendType和inputSize
694 : // 注意: 如果是alltoallv类算子, sendType设置为RESERVED, inputSize设置为0, 保证即使dataType, sendCounts, recvCounts发生变化, 仍然能够缓存命中
695 0 : HcclDataType sendType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
696 0 : uint64_t inputSize = 0;
697 0 : if (!IsAlltoallvType(param.opType)) { // 非alltoallv类算子, 需要根据sendType和inputSize生成不同的key
698 0 : HcclDataType recvType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
699 0 : uint64_t outputSize = 0;
700 0 : CHK_RET(ParseOpParamForCache(param, sendType, recvType, inputSize, outputSize, topoinfo));
701 : UNUSED_PARAM(recvType);
702 : UNUSED_PARAM(outputSize);
703 : } else { // alltoallv类算子, 需要根据isBigCount生成不同的key, 决定SQE编排是否需要并发
704 0 : bool isBigCountForAlltoallv = false;
705 0 : CHK_RET(IsBigCountForAlltoallv(param, topoinfo, isBigCountForAlltoallv));
706 0 : if (isBigCountForAlltoallv) {
707 0 : inputSize = 1;
708 : } else {
709 0 : inputSize = 0;
710 : }
711 : }
712 :
713 : // 准备root: 仅 scatter / broadcast / reduce 三类算子的SQE模板会随root变化, 必须纳入key
714 : // 注意: 其他算子 (allgather/allreduce/alltoall/...) 在不同root下SQE模板相同, 固定为0即可
715 0 : uint32_t root = 0;
716 0 : if (param.opType == HcclCMDType::HCCL_CMD_SCATTER ||
717 0 : param.opType == HcclCMDType::HCCL_CMD_BROADCAST ||
718 0 : param.opType == HcclCMDType::HCCL_CMD_REDUCE) {
719 0 : root = param.root;
720 : }
721 :
722 : // 设置key for op-unfold cache
723 0 : CHK_RET(key.Init(param.opType, sendType, param.reduceType, param.isZeroCopy, inputSize,
724 : algContext.opRetryHandler.isInplacePreSync, workflowMode, param.isCapture, root));
725 :
726 0 : return HCCL_SUCCESS;
727 : }
728 :
729 0 : HcclResult AicpuCacheManager::IsBigCountForAlltoallv(const OpParam ¶m, const HcclTopoInfo& topoinfo,
730 : bool& isBigCount)
731 : {
732 0 : isBigCount = false;
733 0 : if (IsAlltoallvType(param.opType)) { // alltoallv类算子
734 : // 计算maxSendCount
735 : // 参考coll_all_to_all_v_direct_fullmesh_executor.cc中的GetLocalSendRecvInfoforAlltoallV
736 0 : const uint32_t curRank = topoinfo.userRank;
737 0 : const uint32_t rankSize = topoinfo.userRankSize;
738 0 : uint64_t maxSendCount = 0;
739 0 : for (size_t dstRank = 0; dstRank < rankSize; ++dstRank) {
740 : // 获得curRank -> dstRank的sendCount
741 0 : uint64_t curSendCount = 0;
742 0 : if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) { // alltoallv
743 0 : curSendCount = *(static_cast<const uint64_t *>(param.All2AllDataDes.sendCounts) + dstRank);
744 0 : } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) { // alltoallvc
745 0 : curSendCount = *(static_cast<const uint64_t *>(param.All2AllDataDes.sendCountMatrix)
746 0 : + curRank * rankSize + dstRank); // sendCountMatrix[curRank][dstRank]
747 : } else {
748 0 : HCCL_ERROR("[AicpuCacheManager][IsBigCountForAlltoallv] invalid opType[%u] for alltoallv", param.opType);
749 0 : return HCCL_E_INTERNAL;
750 : }
751 :
752 : // 更新maxSendCount
753 0 : if (curSendCount > maxSendCount) {
754 0 : maxSendCount = curSendCount;
755 : }
756 : }
757 :
758 : // 参考alltoallv_direct_fullmesh.cc中的Prepare()
759 0 : uint64_t maxSendLen = maxSendCount * SIZE_TABLE[param.All2AllDataDes.sendType];
760 0 : isBigCount = (maxSendLen > ALLTOALLV_DIRECT_FULLMESH_BIG_SIZE) ? true : false;
761 :
762 0 : HCCL_INFO("[AicpuCacheManager][IsBigCountForAlltoallv] maxSendCount[%llu] maxSendLen[%llu] isBigCount[%u]",
763 : maxSendCount, maxSendLen, isBigCount);
764 : }
765 :
766 0 : return HCCL_SUCCESS;
767 : }
768 :
769 0 : HcclResult AicpuCacheManager::PrepareUserMemRanges(const OpParam ¶m, const AlgResourceResponse &algResource,
770 : std::vector<OpUnfoldMemRange>& userInputMemRanges, std::vector<OpUnfoldMemRange>& userOutputMemRanges,
771 : const HcclTopoInfo& topoinfo, std::shared_ptr<AicpuZeroCopyExchanger>& zeroCopyExchangerPtr,
772 : const HcclWorkflowMode workflowMode, const DeviceMem& tinySendRecvMem)
773 : {
774 : // 注意: 由于PrepareUserMemRanges前已经做过NeedOpUnfoldCache检查, 这里不再做重复检验
775 :
776 0 : const uint32_t rankSize = topoinfo.userRankSize;
777 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] prepare %u user input/output memory ranges for op-unfold cache", rankSize);
778 :
779 : // 准备memory ranges
780 0 : userInputMemRanges.resize(rankSize);
781 0 : userOutputMemRanges.resize(rankSize);
782 :
783 : // 准备input/output size
784 0 : HcclDataType sendType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
785 0 : HcclDataType recvType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
786 0 : uint64_t inputSize = 0;
787 0 : uint64_t outputSize = 0;
788 0 : CHK_RET(ParseOpParamForCache(param, sendType, recvType, inputSize, outputSize, topoinfo));
789 :
790 : // 校验input/output size (应该在NeedOpUnfoldCache中被IsInplace拦截)
791 0 : CHK_PRT_RET(inputSize == 0 && outputSize == 0,
792 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] inputSize[%llu] outputSize[%llu]", inputSize, outputSize),
793 : HCCL_E_INTERNAL);
794 :
795 : // 当前只有alltoall/alltoallv/alltoallvc才会存在inputSize/outputSize为0的情况
796 : // 注意: 对于alltoall算子, outputSize一定等于inputSize, 所以不会存在inputSize/outputSize之一为0的情况
797 : // -> 这里其实只考虑alltoallv/alltoallvc
798 : // 参考aicpu_communicator.cc中的SetAlltoAllInputAndOutPutMem
799 0 : const HcclCMDType opType = param.opType;
800 0 : CHK_PRT_RET((inputSize == 0 || outputSize == 0) &&
801 : opType != HCCL_CMD_ALLTOALL && opType != HCCL_CMD_ALLTOALLV && opType != HCCL_CMD_ALLTOALLVC,
802 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] opType[%u] inputSize[%llu] outputSize[%llu]",
803 : opType, inputSize, outputSize),
804 : HCCL_E_INTERNAL);
805 :
806 : // 设置当前rank的input/output usermem addr
807 0 : const uint32_t curRank = topoinfo.userRank; // NOTE: 不应该使用param.srcRank (某些算子始终为0)
808 0 : CHK_PRT_RET(curRank >= rankSize,
809 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] invalid curRank %u >= rankSize %u", curRank, rankSize),
810 : HCCL_E_INTERNAL);
811 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] prepare user memory range of current rank %u", curRank);
812 0 : OpUnfoldMemRange& curUserInputMemRange = userInputMemRanges[curRank];
813 0 : curUserInputMemRange.isValid = true;
814 0 : if (inputSize == 0) { // 处理alltoallv/alltoallvc的corner case
815 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] use tinySendRecvMem[0x%016llx, %llu]"\
816 : "as local user input for opType[%u]",
817 : tinySendRecvMem.ptr(), tinySendRecvMem.size(), opType);
818 0 : curUserInputMemRange.baseAddr = reinterpret_cast<uint64_t>(tinySendRecvMem.ptr());
819 0 : curUserInputMemRange.memSize = tinySendRecvMem.size();
820 : } else {
821 0 : curUserInputMemRange.baseAddr = reinterpret_cast<uint64_t>(param.inputPtr);
822 0 : curUserInputMemRange.memSize = inputSize; // NOTE: 不应该使用param.inputSize (alltoall类始终为0)
823 : }
824 0 : OpUnfoldMemRange& curUserOutputMemRange = userOutputMemRanges[curRank];
825 0 : curUserOutputMemRange.isValid = true;
826 0 : if (outputSize == 0) { // 处理alltoallv/alltoallvc的corner case
827 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] use tinySendRecvMem[0x%016llx, %llu]"\
828 : "as local user output for opType[%u]",
829 : tinySendRecvMem.ptr(), tinySendRecvMem.size(), opType);
830 0 : curUserOutputMemRange.baseAddr = reinterpret_cast<uint64_t>(tinySendRecvMem.ptr());
831 0 : curUserOutputMemRange.memSize = tinySendRecvMem.size();
832 : } else {
833 0 : curUserOutputMemRange.baseAddr = reinterpret_cast<uint64_t>(param.outputPtr);
834 0 : curUserOutputMemRange.memSize = outputSize; // NOTE: 不应该使用param.outputSize (alltoall类始终为0)
835 : }
836 :
837 : // 针对zero copy, 设置remote rank的input/output usermem addr
838 0 : if (param.isZeroCopy) {
839 : // 注意: 只有非V类算子可能使用zero copy (因此假设remote ranks' input/output size与local rank相同)
840 : // 注意: 而V类算子一定是buffer copy, 只会存在local user/hccl <-> remote hccl之间的搬运 (否则PrepareRemoteUserMemRanges需要额外的输入作为remote ranks' input/output size)
841 0 : CHK_PRT_RET(opType == HCCL_CMD_ALLTOALLV || opType == HCCL_CMD_ALLTOALLVC || opType == HCCL_CMD_ALLGATHER_V ||
842 : opType == HCCL_CMD_REDUCE_SCATTER_V || opType == HCCL_CMD_HALF_ALLTOALLV,
843 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] opType[%u] should not use zero copy", opType),
844 : HCCL_E_INTERNAL);
845 :
846 : // 直接传入local rank's input/output size用于remote ranks' memory ranges
847 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] prepare user memory ranges of other remote ranks");
848 0 : CHK_PTR_NULL(zeroCopyExchangerPtr.get());
849 0 : CHK_RET(zeroCopyExchangerPtr->PrepareRemoteUserMemRanges(inputSize, outputSize, userInputMemRanges, userOutputMemRanges));
850 0 : } else if (workflowMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB || // 图模式
851 0 : param.aicpuCacheEnable > FORCE_OP_BASE_DELTA) { // 存在强制单算子模式转换 (即图模式建链+单算子模式展开)
852 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] check transport resource for potential user memory of remote ranks");
853 :
854 : // 遍历所有transport信息, 更新remote ranks' user input/output memory ranges
855 0 : for (size_t planeIdx = 0; planeIdx < algResource.opTransportResponse.size(); ++planeIdx) {
856 0 : const LevelNSubCommTransport& subCommTransport = algResource.opTransportResponse[planeIdx];
857 0 : for (size_t commIdx = 0; commIdx < subCommTransport.size(); ++commIdx) {
858 0 : const SingleSubCommTransport& commTransport = subCommTransport[commIdx];
859 :
860 : // 注意: 假设SingleSubCommTransport中的transportRequests和links是一一对应的
861 0 : const std::vector<TransportRequest>& transportRequests = commTransport.transportRequests;
862 0 : const std::vector<LINK>& links = commTransport.links;
863 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] planeIdx[%u] commIdx[%u] links.size[%u]", planeIdx, commIdx, links.size());
864 0 : CHK_PRT_RET(transportRequests.size() != links.size(),
865 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] transportRequests.size[%u] != links.size[%u]",
866 : transportRequests.size(), links.size()),
867 : HCCL_E_INTERNAL);
868 :
869 : // 遍历每个remote rank对应的link信息
870 0 : for (size_t reqIdx = 0; reqIdx < transportRequests.size(); ++reqIdx) {
871 0 : const TransportRequest& curReq = transportRequests[reqIdx];
872 0 : if (curReq.isValid) {
873 0 : if (curReq.remoteUserRank == curRank) { // 本rank无需从link获取user memory range
874 0 : continue;
875 0 : } else if (curReq.remoteUserRank == INVALID_VALUE_RANKID) { // 本rank无需从link获取user memory range
876 0 : continue;
877 : }
878 :
879 0 : CHK_PRT_RET(curReq.remoteUserRank >= rankSize,
880 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] invalid remoteRank %u >= rankSize %u",
881 : curReq.remoteUserRank, rankSize),
882 : HCCL_E_INTERNAL);
883 :
884 : // 获取curRank与remoteRank之间的LINK
885 0 : const LINK& curLink = links[reqIdx];
886 0 : CHK_PTR_NULL(curLink);
887 :
888 : // 获取user input memory range if any
889 0 : if (curReq.inputMemType == TransportMemType::PARAM_INPUT ||
890 0 : curReq.inputMemType == TransportMemType::CCL_INPUT) {
891 : // 获取remoteRank的user input memory baseaddr
892 0 : void *remoteUserInputBaseAddr = nullptr;
893 0 : CHK_RET(curLink->GetRemoteMem(UserMemType::INPUT_MEM, &remoteUserInputBaseAddr));
894 0 : CHK_PTR_NULL(remoteUserInputBaseAddr);
895 :
896 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] prepare user input of remoteRank[%u] for"\
897 : "graph mode; baseAddr[0x%016llx]", curReq.remoteUserRank, remoteUserInputBaseAddr);
898 :
899 : // 设置remoteRank对应的user input memory range
900 0 : OpUnfoldMemRange& remoteUserInputMemRange = userInputMemRanges[curReq.remoteUserRank];
901 0 : remoteUserInputMemRange.isValid = true;
902 0 : remoteUserInputMemRange.baseAddr = reinterpret_cast<uint64_t>(remoteUserInputBaseAddr);
903 0 : if (curReq.inputMemType == TransportMemType::PARAM_INPUT) { // user input
904 0 : remoteUserInputMemRange.memSize = inputSize;
905 0 : } else if (curReq.inputMemType == TransportMemType::CCL_INPUT) { // hccl input
906 0 : remoteUserInputMemRange.memSize = algResource.cclInputMem.size();
907 : } else {
908 0 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] invalid curReq.inputMemType[%u]",
909 : curReq.inputMemType);
910 0 : return HCCL_E_INTERNAL;
911 : }
912 : }
913 :
914 : // 获取user output memory range if any
915 0 : if (curReq.outputMemType == TransportMemType::PARAM_OUTPUT ||
916 0 : curReq.outputMemType == TransportMemType::CCL_OUTPUT) {
917 : // 获取remoteRank的user output memory baseaddr
918 0 : void *remoteUserOutputBaseAddr = nullptr;
919 0 : CHK_RET(curLink->GetRemoteMem(UserMemType::OUTPUT_MEM, &remoteUserOutputBaseAddr));
920 0 : CHK_PTR_NULL(remoteUserOutputBaseAddr);
921 :
922 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] prepare user output of remoteRank[%u] for"\
923 : "graph mode; baseAddr[0x%016llx]", curReq.remoteUserRank, remoteUserOutputBaseAddr);
924 :
925 : // 设置remoteRank对应的user output memory range
926 0 : OpUnfoldMemRange& remoteUserOutputMemRange = userOutputMemRanges[curReq.remoteUserRank];
927 0 : remoteUserOutputMemRange.isValid = true;
928 0 : remoteUserOutputMemRange.baseAddr = reinterpret_cast<uint64_t>(remoteUserOutputBaseAddr);
929 0 : if (curReq.outputMemType == TransportMemType::PARAM_OUTPUT) { // user output
930 0 : remoteUserOutputMemRange.memSize = outputSize;
931 0 : } else if (curReq.outputMemType == TransportMemType::CCL_OUTPUT) { // hccl output
932 0 : remoteUserOutputMemRange.memSize = algResource.cclOutputMem.size();
933 : } else {
934 0 : HCCL_ERROR("[AicpuCacheManager][PrepareUserMemRanges] invalid curReq.outputMemType[%u]",
935 : curReq.outputMemType);
936 0 : return HCCL_E_INTERNAL;
937 : }
938 : }
939 : } // curReq.isValid
940 : } // Each TransportRequest
941 : } // Each SingleSubCommTransport
942 : } // Each LevelNSubCommTransport
943 : }
944 :
945 : // 打印debug信息
946 0 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
947 0 : for (size_t rankId = 0; rankId < userInputMemRanges.size(); ++rankId) {
948 0 : const OpUnfoldMemRange& userInputMemRange = userInputMemRanges[rankId];
949 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] userInputMemRanges[%u] isValid: %d, baseAddr: 0x%016llx, memSize: %llu, endAddr: 0x%016llx",
950 : rankId, userInputMemRange.isValid, userInputMemRange.baseAddr, userInputMemRange.memSize, userInputMemRange.baseAddr + userInputMemRange.memSize);
951 :
952 0 : const OpUnfoldMemRange& userOutputMemRange = userOutputMemRanges[rankId];
953 0 : HCCL_INFO("[AicpuCacheManager][PrepareUserMemRanges] userOutputMemRanges[%u] isValid: %d, baseAddr: 0x%016llx, memSize: %llu, endAddr: 0x%016llx",
954 : rankId, userOutputMemRange.isValid, userOutputMemRange.baseAddr, userOutputMemRange.memSize, userOutputMemRange.baseAddr + userOutputMemRange.memSize);
955 : }
956 : }
957 :
958 0 : return HCCL_SUCCESS;
959 : }
960 :
961 0 : HcclResult AicpuCacheManager::ParseOpParamForCache(const OpParam ¶m, HcclDataType& sendType, HcclDataType& recvType,
962 : uint64_t& inputSize, uint64_t& outputSize, const HcclTopoInfo& topoinfo)
963 : {
964 : // 注意: 由于ParseOpParamForCache前已经做过NeedOpUnfoldCache检查, 这里不再做重复检验
965 :
966 0 : const HcclCMDType opType = param.opType;
967 0 : const uint32_t rankSize = topoinfo.userRankSize;
968 :
969 : // 准备data type和count
970 : // NOTE: 非V类算子 (DataRes), V类算子 (VDataDes), All2All类算子 (All2AllDataDes), batch类算子 (BatchSendRecvDataDes/BatchWriteDataDes)
971 0 : if (opType == HcclCMDType::HCCL_CMD_ALLTOALL) { // alltoall算子
972 : // 注意: sendType和recvType一定相同
973 0 : sendType = param.All2AllDataDes.sendType;
974 0 : recvType = param.All2AllDataDes.recvType;
975 :
976 : // 注意: 对于alltoall算子, inputSize和outputSize一定相同 (但不能直接使用param.input/outputSize, alltoall算子不会设置这两个字段)
977 0 : inputSize = param.All2AllDataDes.sendCount * rankSize * SIZE_TABLE[sendType];
978 0 : outputSize = inputSize; // 注意: 不能使用param.All2AllDataDes.recvCount * rankSize * SIZE_TABLE[recvType], 因为alltoall使用sendCount来表示send/recvCount, 而recvCount本身为0
979 0 : } else if (IsAlltoallvType(opType)) { // alltoallv类算子
980 : // 计算相应字段 (虽然GetOpUnfoldKey不需要, 但是PrepareUserMemRanges需要)
981 :
982 : // 注意: sendType和recvType一定相同
983 0 : sendType = param.All2AllDataDes.sendType;
984 0 : recvType = param.All2AllDataDes.recvType;
985 :
986 : // 注意: 对于alltoallv算子, inputSize和outputSize不一定相同 (但不能直接使用param.input/outputSize, alltoallv算子不会设置这两个字段)
987 : // 参考coll_all_to_all_v_direct_fullmesh_executor.cc下的CollRunAlltoAllDirectFullmesh::GetLocalSendRecvInfoforAlltoallV
988 0 : inputSize = 0;
989 0 : outputSize = 0;
990 0 : if (opType == HcclCMDType::HCCL_CMD_ALLTOALLV) { // alltoallv算子
991 0 : HCCL_INFO("[AicpuCacheManager][ParseOpParamForCache] sum %u send/recv counts for input/output size", rankSize);
992 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
993 : // curRank发送到tmpRank的数据量
994 0 : const uint64_t curSendCounts = *(static_cast<const u64 *>(param.All2AllDataDes.sendCounts) + tmpRank);
995 0 : const uint64_t curSendLength = curSendCounts * SIZE_TABLE[sendType];
996 0 : inputSize += curSendLength;
997 :
998 : // curRank从tmpRank接收的数据量
999 0 : const uint64_t curRecvCounts = *(static_cast<const u64 *>(param.All2AllDataDes.recvCounts) + tmpRank);
1000 0 : const uint64_t curRecvLength = curRecvCounts * SIZE_TABLE[recvType];
1001 0 : outputSize += curRecvLength;
1002 : }
1003 0 : } else if (opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) { // alltoallvc算子
1004 0 : const uint32_t curRank = topoinfo.userRank;
1005 0 : HCCL_INFO("[AicpuCacheManager][ParseOpParamForCache] sum %u-size sendCountMatrix for input/output size", rankSize);
1006 0 : for (uint32_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
1007 : // curRank发送到tmpRank的数据量
1008 0 : const uint64_t curSendCounts = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix)
1009 0 : + curRank * rankSize + tmpRank); // sendCountMatrix[curRank][tmpRank]
1010 0 : const uint64_t curSendLength = curSendCounts * SIZE_TABLE[sendType];
1011 0 : inputSize += curSendLength;
1012 :
1013 : // curRank从tmpRank接收到的数据量
1014 0 : const uint64_t curRecvCounts = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix)
1015 0 : + tmpRank * topoinfo.userRankSize + curRank); // sendCountMatrix[tmpRank][curRank]
1016 0 : const uint64_t curRecvLength = curRecvCounts * SIZE_TABLE[recvType];
1017 0 : outputSize += curRecvLength;
1018 : }
1019 : } else {
1020 0 : HCCL_ERROR("[AicpuCacheManager][ParseOpParamForCache] invalid opType[%u] for alltoallv", opType);
1021 0 : return HCCL_E_INTERNAL;
1022 : }
1023 : } else { // 非V类算子
1024 0 : sendType = param.DataDes.dataType;
1025 0 : recvType = param.DataDes.dataType;
1026 0 : inputSize = param.inputSize;
1027 0 : outputSize = param.outputSize;
1028 : }
1029 :
1030 0 : HCCL_DEBUG("[AicpuCacheManager][ParseOpParamForCache] opType[%u] rankSize[%u] sendType[%u] recvType[%u] inputSize[%u] outputSize[%u]",
1031 : opType, rankSize, sendType, recvType, inputSize, outputSize);
1032 :
1033 0 : return HCCL_SUCCESS;
1034 : }
1035 :
1036 0 : HcclResult AicpuCacheManager::PrepareAlltoallvSendRecvInfo(const OpParam& param, AlltoallvSendRecvInfo& alltoallvSendRecvInfo,
1037 : const HcclTopoInfo& topoinfo)
1038 : {
1039 0 : const uint32_t rankSize = topoinfo.userRankSize;
1040 0 : HCCL_INFO("[AicpuCacheManager][PrepareAlltoallvSendRecvInfo] prepare %u send/recv info", rankSize);
1041 :
1042 : // 准备send/recv data type
1043 0 : const HcclDataType sendType = param.All2AllDataDes.sendType;
1044 0 : alltoallvSendRecvInfo.sendType = sendType;
1045 0 : const HcclDataType recvType = param.All2AllDataDes.recvType;
1046 0 : alltoallvSendRecvInfo.recvType = recvType;
1047 :
1048 : // 初始化send/recv counts
1049 0 : alltoallvSendRecvInfo.sendCounts.resize(rankSize);
1050 0 : alltoallvSendRecvInfo.recvCounts.resize(rankSize);
1051 :
1052 : // 初始化send/recv offsets
1053 0 : alltoallvSendRecvInfo.sendOffsets.resize(rankSize);
1054 0 : alltoallvSendRecvInfo.recvOffsets.resize(rankSize);
1055 :
1056 : // 参考coll_all_to_all_v_direct_fullmesh_executor.cc中的GetLocalSendRecvInfoforAlltoallV
1057 0 : const uint32_t sendTypeSize = SIZE_TABLE[sendType]; // Size of sendType in units of bytes
1058 0 : const uint32_t recvTypeSize = SIZE_TABLE[recvType]; // Size of recvType in units of bytes
1059 0 : if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) { // alltoallv
1060 : // 准备send counts
1061 0 : for (size_t dstRank = 0; dstRank < rankSize; ++dstRank) {
1062 0 : alltoallvSendRecvInfo.sendCounts[dstRank] = *(static_cast<const u64 *>(param.All2AllDataDes.sendCounts) + dstRank);
1063 : }
1064 :
1065 : // 准备recv counts
1066 0 : for (size_t dstRank = 0; dstRank < rankSize; ++dstRank) {
1067 0 : alltoallvSendRecvInfo.recvCounts[dstRank] = *(static_cast<const u64 *>(param.All2AllDataDes.recvCounts) + dstRank);
1068 : }
1069 :
1070 : // 准备send offsets
1071 0 : for (size_t dstRank = 0; dstRank < rankSize; ++dstRank) {
1072 0 : const uint64_t curSendDispls = *(static_cast<const uint64_t *>(param.All2AllDataDes.sdispls) + dstRank);
1073 0 : alltoallvSendRecvInfo.sendOffsets[dstRank] = curSendDispls * sendTypeSize;
1074 : }
1075 :
1076 : // 准备recv offsets
1077 0 : for (size_t dstRank = 0; dstRank < rankSize; ++dstRank) {
1078 0 : const uint64_t curRecvDispls = *(static_cast<const uint64_t *>(param.All2AllDataDes.rdispls) + dstRank);
1079 0 : alltoallvSendRecvInfo.recvOffsets[dstRank] = curRecvDispls * recvTypeSize;
1080 : }
1081 0 : } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) { // alltoallvc
1082 0 : const uint32_t curRank = topoinfo.userRank;
1083 :
1084 : // 准备send counts
1085 0 : for (size_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
1086 0 : alltoallvSendRecvInfo.sendCounts[tmpRank] = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix)
1087 0 : + curRank * rankSize + tmpRank); // sendCountMatrix[curRank][tmpRank]
1088 : }
1089 :
1090 : // 准备recv counts
1091 0 : for (size_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
1092 0 : alltoallvSendRecvInfo.recvCounts[tmpRank] = *(static_cast<const u64 *>(param.All2AllDataDes.sendCountMatrix)
1093 0 : + tmpRank * topoinfo.userRankSize + curRank); // sendCountMatrix[tmpRank][curRank]
1094 : }
1095 :
1096 : // 准备send offsets
1097 0 : uint64_t curSendDispls = 0;
1098 0 : for (size_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
1099 0 : alltoallvSendRecvInfo.sendOffsets[tmpRank] = curSendDispls * sendTypeSize;
1100 0 : curSendDispls += alltoallvSendRecvInfo.sendCounts[tmpRank];
1101 : }
1102 :
1103 : // 准备recv offsets
1104 0 : uint64_t curRecvDispls = 0;
1105 0 : for (size_t tmpRank = 0; tmpRank < rankSize; ++tmpRank) {
1106 0 : alltoallvSendRecvInfo.recvOffsets[tmpRank] = curRecvDispls * recvTypeSize;
1107 0 : curRecvDispls += alltoallvSendRecvInfo.recvCounts[tmpRank];
1108 : }
1109 : } else {
1110 0 : HCCL_ERROR("[AicpuCacheManager][PrepareAlltoallvSendRecvInfo] invalid opType[%u] for alltoallv", param.opType);
1111 0 : return HCCL_E_INTERNAL;
1112 : }
1113 :
1114 0 : return HCCL_SUCCESS;
1115 : }
1116 :
1117 : } // namespace hccl
|
