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