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 "all_gather_slim_ring.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : AllGatherSlimRing::AllGatherSlimRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher)
16 : {
17 0 : }
18 :
19 0 : AllGatherSlimRing::~AllGatherSlimRing()
20 : {
21 0 : }
22 :
23 0 : HcclResult AllGatherSlimRing::TxVector(const LINK &link, const std::vector<Slice> &txSlices)
24 : {
25 0 : std::vector<TxMemoryInfo> txMems;
26 0 : for (const Slice &txSlice : txSlices) {
27 0 : DeviceMem srcMem = outputMem_.range(txSlice.offset, txSlice.size);
28 0 : HCCL_DEBUG("tx srcMem[%p] range[%llu] size[%llu] ", srcMem.ptr(), txSlice.offset, txSlice.size);
29 0 : txMems.emplace_back(TxMemoryInfo{UserMemType::OUTPUT_MEM, txSlice.offset + baseOffset_,
30 0 : srcMem.ptr(), txSlice.size});
31 0 : }
32 0 : CHK_RET(link->Post(notifyIdx_, stream_));
33 0 : return HCCL_SUCCESS;
34 0 : }
35 :
36 0 : HcclResult AllGatherSlimRing::RxVector(const LINK &link, const std::vector<Slice> &rxSlices)
37 : {
38 0 : std::vector<RxMemoryInfo> rxMems;
39 0 : for (const Slice &rxSlice : rxSlices) {
40 0 : DeviceMem dstMem = outputMem_.range(rxSlice.offset, rxSlice.size);
41 0 : HCCL_DEBUG("rx dstMem[%p] range[%llu], size[%llu] ", dstMem.ptr(),
42 : rxSlice.offset, rxSlice.size);
43 0 : rxMems.emplace_back(RxMemoryInfo{UserMemType::OUTPUT_MEM, rxSlice.offset + baseOffset_,
44 0 : dstMem.ptr(), rxSlice.size});
45 0 : }
46 0 : CHK_RET(link->Wait(notifyIdx_, stream_));
47 0 : for (auto& mem : rxMems) {
48 0 : CHK_PTR_NULL(mem.dst);
49 0 : void *srcMemPtr = nullptr;
50 0 : CHK_RET(link->GetRemoteMem(mem.srcMemType, &srcMemPtr));
51 :
52 0 : DeviceMem srcDevMem(static_cast<s8 *>(srcMemPtr) + mem.srcOffset, mem.len);
53 0 : DeviceMem dstDevMem(static_cast<s8 *>(mem.dst), mem.len);
54 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstDevMem, srcDevMem, stream_, link->GetRemoteRank(),
55 : link->GetLinkType()));
56 0 : }
57 0 : return HCCL_SUCCESS;
58 0 : }
59 :
60 0 : HcclResult AllGatherSlimRing::InitSlice(std::vector<Slice>& inputSlices, u32 rank,
61 : u32 rankSize, u32 unitSize){
62 0 : if (slices_.size() == 0) {
63 0 : slices_.resize(rankSize);
64 0 : inputSlices.resize(rankSize);
65 0 : u64 sliceSize = count_ * unitSize;
66 0 : for (u32 i = 0; i < rankSize; i++) {
67 0 : slices_[i].size = sliceSize;
68 0 : slices_[i].offset = sliceSize * i;
69 0 : inputSlices[i].size = sliceSize;
70 0 : inputSlices[i].offset = (inputMem_.size() < outputMem_.size()) ? 0 : (sliceSize * i);
71 0 : HCCL_DEBUG("rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu", \
72 : rank, i, slices_[i].offset, i, slices_[i].size);
73 : }
74 : }
75 0 : return HCCL_SUCCESS;
76 : }
77 :
78 : // 服务器间allgather的入口函数
79 0 : HcclResult AllGatherSlimRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
80 : {
81 0 : CHK_SMART_PTR_NULL(dispatcher_);
82 0 : CHK_PTR_NULL(stream_.ptr());
83 0 : HCCL_INFO("AllGatherSlimRing run_async rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", \
84 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
85 :
86 0 : if (rankSize == 1) {
87 0 : if (inputMem_ != outputMem_) {
88 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_));
89 : }
90 0 : HCCL_DEBUG("[AllGatherSlimRing]rankSize is 1, RunAsync success");
91 0 : return HCCL_SUCCESS;
92 : }
93 :
94 : // 获取ring algorithm所需的通信连接
95 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
96 0 : u32 ringNextRank = (rank + 1) % rankSize;
97 :
98 0 : if (links.size() < rankSize) {
99 0 : HCCL_ERROR("[AllGatherSlimRing][RunAsync]rank[%u] linkSize is less than rankSize", rank);
100 0 : return HCCL_E_INTERNAL;
101 : }
102 :
103 0 : linkLeft_ = links[ringPrevRank];
104 0 : CHK_SMART_PTR_NULL(linkLeft_);
105 :
106 0 : linkRight_ = links[ringNextRank];
107 0 : CHK_SMART_PTR_NULL(linkRight_);
108 :
109 0 : u32 unitSize = DataUnitSize(dataType_);
110 0 : if (unitSize == 0) {
111 0 : HCCL_ERROR("[AllGatherSlimRing][RunAsync]unitSize is zero");
112 0 : return HCCL_E_INTERNAL;
113 : }
114 :
115 0 : std::vector<Slice> inputSlices(slices_);
116 0 : InitSlice(inputSlices, rank, rankSize, unitSize);
117 :
118 : // 双buffer下, 先将input拷贝到output的合适位置
119 0 : if (inputMem_ != outputMem_) {
120 0 : DeviceMem dst = outputMem_.range(slices_[rank].offset, slices_[rank].size);
121 0 : DeviceMem src = inputMem_.range(inputSlices[rank].offset, inputSlices[rank].size);
122 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
123 0 : }
124 :
125 : // 运行all-gather, ring算法
126 : // 单环场景下 nicRankList_ 长度默认为 8。
127 : // 多环场景下 nicRankList_ 长度为网口数量。此时若 rankSize != nicRankList_ 则为网口裁剪场景
128 0 : if (rankSize != HCCL_NIC_MAX_NUM || nicRankList_.size() == HCCL_NIC_MAX_NUM) {
129 : // 非网口裁剪场景:
130 0 : CHK_RET(RunAllGather(rank, rankSize, slices_));
131 : }
132 :
133 0 : if (barrierSwitchOn_) {
134 : // 执行barrier，保证数据收发完成
135 0 : CHK_RET(ExecuteBarrier(linkRight_, linkLeft_, notifyIdx_));
136 0 : notifyIdx_++;
137 : }
138 :
139 0 : HCCL_INFO("AllGatherSlimRing finished: rank[%u] end", rank);
140 0 : return HCCL_SUCCESS;
141 0 : }
142 :
143 0 : HcclResult AllGatherSlimRing::RunAllGather(u32 rank, u32 rankSize, const std::vector<Slice> &outputSlices)
144 : {
145 0 : if (outputSlices.size() < rankSize) {
146 0 : HCCL_ERROR("[Run][AllGather]rank[%u] OutputSlice Size is less than rank size.", rank);
147 0 : return HCCL_E_INTERNAL;
148 : }
149 0 : HcclResult ret = HCCL_SUCCESS;
150 :
151 : // 首次传输，将本rank的数据发送到下游
152 0 : u32 sliceSize = outputSlices.size() / rankSize;
153 0 : u32 rxSliceIndex = ForwordRank(rank, rankSize, 1);
154 0 : u32 txSliceIndex = rank;
155 0 : for (u32 i = 0; i < rankSize - 1; i++) {
156 : // reduce目的操作
157 0 : std::vector<Slice> rxSegsSlice;
158 0 : std::vector<Slice> txSegsSlice;
159 0 : for (u32 j = 0; j < sliceSize; j++) {
160 0 : txSegsSlice.push_back(outputSlices[txSliceIndex * sliceSize + j]);
161 0 : rxSegsSlice.push_back(outputSlices[rxSliceIndex * sliceSize + j]);
162 : }
163 0 : ret = TxVector(linkRight_, txSegsSlice);
164 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
165 : HCCL_ERROR("[Run][AllGather]rank[%u] round[%u] Right Link tx outputSlices[%u] "\
166 : "Failed", rank, i, txSliceIndex), ret);
167 :
168 : // reduce源操作
169 0 : ret = RxVector(linkLeft_, rxSegsSlice);
170 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
171 : HCCL_ERROR("[Run][AllGather]rank[%u] round[%u] Left Link rx outputSlices[%u] "\
172 : "Failed", rank, i, rxSliceIndex), ret);
173 :
174 : // 末尾传输, 只接收一次, 不用再次发送
175 0 : txSliceIndex = ForwordRank(txSliceIndex, rankSize, 1);
176 0 : rxSliceIndex = ForwordRank(rxSliceIndex, rankSize, 1);
177 :
178 0 : notifyIdx_++;
179 0 : }
180 0 : return HCCL_SUCCESS;
181 : }
182 :
183 0 : HcclResult AllGatherSlimRing::SetNotifyIdx(u32 notifyIdx)
184 : {
185 0 : notifyIdx_ = notifyIdx;
186 0 : return HCCL_SUCCESS;
187 : }
188 :
189 0 : HcclResult AllGatherSlimRing::GetNotifyIdx(u32 ¬ifyIdx)
190 : {
191 0 : notifyIdx = notifyIdx_;
192 0 : return HCCL_SUCCESS;
193 : }
194 :
195 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_SLIM_RING, AllGatherSlimRing);
196 : } // namespace hccl
|
