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 : #ifndef ALG_TEMPLATE_MULTI_DETER_PIPELINE_H
12 : #define ALG_TEMPLATE_MULTI_DETER_PIPELINE_H
13 :
14 : #include <vector>
15 : #include <memory>
16 : #include <hccl/hccl_types.h>
17 : #include "hccl/base.h"
18 : #include "externalinput_pub.h"
19 : #include "mem_device_pub.h"
20 : #include "stream_pub.h"
21 : #include "dispatcher.h"
22 : #include "alg_template_base_pub.h"
23 :
24 : namespace hccl {
25 : constexpr u32 STEP_OFFSET_TWO = 2;
26 : // 上游保证最多4条流做规约操作，4条做localreduce，其中1条做localreduce主流
27 : constexpr u32 MAX_REDUCE_STREAM_NUM = 4;
28 : constexpr u32 MIN_SERVER_NUM = 2;
29 : constexpr u32 MIN_INTRA_RANK_NUM = 3;
30 : constexpr u32 SECOND_TO_LAST = 2;
31 : constexpr u32 LOCAL_REDUCE_SERIIAL_ALG_SERVER_NUM = 2;
32 : constexpr u32 PARITY_BASE = 2;
33 : class MultiDeterPipeline : public AlgTemplateBase {
34 : public:
35 : explicit MultiDeterPipeline (const HcclDispatcher dispatcher);
36 : ~MultiDeterPipeline() override;
37 : HcclResult RunAsync() override;
38 : HcclResult RunAsyncReduceScatterPipeline();
39 : // ReduceScatterDeterPipeline
40 : HcclResult Prepare(HcomCollOpInfo *opInfo, DeviceMem &buffer, const u64 count,
41 : const u64 offset, const std::vector<Slice> &slices, const SubCommInfo &level0CommInfo,
42 : const SubCommInfo &level1CommInfo, Stream &mainStream, std::vector<Stream> &subStream,
43 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub) override;
44 :
45 : // AllReduceDeterPipeline
46 : HcclResult Prepare(HcomCollOpInfo *opInfo, DeviceMem &inBuffer, DeviceMem &outBuffer, const u64 count,
47 : const std::vector<Slice> &slices, const SubCommInfo &level0CommInfo,
48 : const SubCommInfo &level1CommInfo, Stream &mainStream, std::vector<Stream> &subStream,
49 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub) override;
50 : protected:
51 : HcclResult MainWaitSub(u32 begin, u32 end);
52 : HcclResult SubRecordMain(u32 begin, u32 end);
53 : HcclResult MainRecordSub(u32 begin, u32 end);
54 : HcclResult SubWaitMain(u32 begin, u32 end);
55 : // 根据step获取 机间或机内的rankId
56 0 : constexpr u32 GetPreRankIdByStep(u32 rankId, u32 rankSize, u32 step) {
57 0 : return (rankId + rankSize - step) % rankSize;
58 : }
59 :
60 0 : constexpr u32 GetNextRankIdByStep(u32 rankId, u32 rankSize, u32 step) {
61 0 : return (rankId + step) % rankSize;
62 : }
63 :
64 0 : inline u32 GetPreServerIdByStep(u32 step) {
65 0 : return GetPreRankIdByStep(serverId_, serverSize_, step);
66 : }
67 :
68 0 : inline u32 GetNextServerIdByStep(u32 step) {
69 0 : return GetNextRankIdByStep(serverId_, serverSize_, step);
70 : }
71 :
72 0 : inline u32 GetPreIntraRankIdByStep(u32 step) {
73 0 : return GetPreRankIdByStep(intraRankId_, intraRankSize_, step);
74 : }
75 :
76 0 : inline u32 GetNextIntraRankIdByStep(u32 step) {
77 0 : return GetNextRankIdByStep(intraRankId_, intraRankSize_, step);
78 : }
79 :
80 0 : inline u32 GetRankIdx(u32 serverId, u32 intraRankId) {
81 0 : return serverId * intraRankSize_ + intraRankId;
82 : }
83 : // 获取device内存部分
84 : virtual HcclResult GetRemoteCclbufferDeviceMem(u32 inputSliceIndex, LINK link,
85 : u32 outputSliceIndex, DeviceMem &remoteMem);
86 : virtual HcclResult GetLocalUserInDeviceMem(u32 rankIdInAllRanks, DeviceMem &locaMem);
87 : virtual HcclResult GetLocalUserOutDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem);
88 : virtual HcclResult GetLocalInCclbufferDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem, bool ifUseLastSize);
89 : virtual HcclResult GetLocalOutCclbufferDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem, bool ifUseLastSize);
90 :
91 : virtual HcclResult RunLocalCopy();
92 : virtual HcclResult RunIntraAlltoallPreSync(u32 step);
93 : HcclResult RunIntraAlltoall(u32 step);
94 : // LocalReduce内部函数
95 : HcclResult GroupTasksByStream(u32 activeCount, const std::vector<bool>& isReduceBlock,
96 : u32 retIndex, std::vector<std::vector<std::vector<std::pair<u32, u32>>>>& batchStreamTasks,
97 : std::vector<bool>& processed, std::vector<u32>& origIdxMap, u32& newActiveCount);
98 : HcclResult ExecuteStreamTasks(const std::vector<std::vector<std::pair<u32, u32>>>& streamTasks,
99 : const std::vector<DeviceMem>& validMem, std::vector<u32>& origIdxMap, bool useMainStream);
100 : virtual HcclResult BatchPostNotifyForStreams(const std::vector<std::vector<std::pair<u32, u32>>>& streamTasks,
101 : bool isStartPhase, bool useMainStream);
102 : void CompressActiveSet(std::vector<DeviceMem> &validMem, std::vector<bool> &isReduceBlock, std::vector<u32> &origIdxMap,
103 : const std::vector<bool> &processed, u32 &trackedTargetIdx, const u32 origRetIndex);
104 : HcclResult LocalReduce(std::vector<DeviceMem> &reduceMem, std::vector<bool> &isReduceBlock, u32 retIndex, bool useMainStream);
105 : virtual HcclResult RunIntraLocalReduce(u32 step);
106 : virtual HcclResult RunFinalReduce();
107 : // RDAM send部分
108 : virtual HcclResult RunInterSend(u32 step);
109 : // 主从流同步部分
110 : virtual HcclResult AlltoallSync(u32 step, bool isStartPhase);
111 : virtual HcclResult LocalReduceSync(u32 step, bool isStartPhase);
112 : HcclResult AlltoallLocalReduceSync(u32 step, bool isStartPhase);
113 : // local reduce串行算法
114 : HcclResult RunAsyncLocalReduceSerial();
115 : // 初始化部分
116 : void InitAlltoallRecvBlockIdxMap();
117 : HcclResult PrepareTopoInfo(const SubCommInfo &level0CommInfo, const SubCommInfo &level1CommInfo);
118 : virtual u64 GetLocalReduceSerialThresh() = 0;
119 :
120 : HcomCollOpInfo *opInfo_{nullptr};
121 :
122 : void* usrInMemPtr_ = nullptr;
123 : void* usrOutMemPtr_ = nullptr;
124 : u64 count_ = 0; // output中的数量
125 : u32 unitSize_ = 0;
126 : u64 curSize_ = 0;
127 : u64 memSliceSize_ = 0;
128 : u64 blockSize_ = 0;
129 : u64 bufferSize_ = 0;
130 : HcclReduceOp reductionOp_ = HcclReduceOp::HCCL_REDUCE_RESERVED;
131 : HcclDataType dataType_ = HcclDataType::HCCL_DATA_TYPE_RESERVED;
132 :
133 : std::vector<Stream> subStreams_;
134 : u32 subStreamNum_ = 0;
135 : Stream mainStream_;
136 :
137 : std::vector<std::shared_ptr<LocalNotify>> streamNotifyMain_;
138 : std::vector<std::shared_ptr<LocalNotify>> streamNotifySub_;
139 :
140 : u32 all2allStreamBegin_ = 0; // all2all专用
141 : u32 all2allStreamSize_ = 0;
142 : u32 reduceMainStreamIdx_ = 0;
143 : u32 reduceStreamBegin_ = 0; // local reduce专用
144 : u32 reduceStreamSize_ = 0;
145 : u32 intraRankSize_ = 0; // 机内
146 : u32 serverSize_ = 0; // 机间
147 : u32 intraRankId_ = 0; // 机内
148 : u32 serverId_ = 0; // 机间
149 : u64 offset_ = 0;
150 : u32 allSteps_ = 0;
151 : u64 eachRankCclbufferSize_ = 0;
152 :
153 : u32 userRankSize_ = 0;
154 : u32 userRank_ = 0;
155 : std::vector<std::vector<u32>> alltoallRecvBlockIdxMap_; // alltoall接收block的idx映射表
156 : // 本地cclbuffer的偏移
157 : // allreduce为了保证地址对齐，进行数据分块时除了最后一块数据
158 : // 其他分块都向上取HCCL_MIN_SLICE_ALIGN_910B倍数的大小，最后一块数据取剩余的大小。
159 : // reduce scatter每块大小相同向上直接HCCL_MIN_SLICE_ALIGN_910B取整。
160 : std::vector<Slice> slices_;
161 : std::vector<LINK> intraLinks_;
162 : std::vector<LINK> serverLinks_;
163 : };
164 : }
165 : #endif
|
