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