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