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 <math.h>
12 : #include "alg_template_register.h"
13 : #include "all_reduce_doubling_direct.h"
14 :
15 : namespace hccl {
16 :
17 : // Doubling算法实现AllReduce，只用于server内通信
18 0 : AllReduceDoublingDirect ::AllReduceDoublingDirect(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
19 :
20 0 : AllReduceDoublingDirect::~AllReduceDoublingDirect() {}
21 :
22 0 : HcclResult AllReduceDoublingDirect::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo* opInfo)
23 : {
24 0 : reduceAttr_ = reduceAttrBitMap;
25 0 : opInfo_ = opInfo;
26 0 : return HCCL_SUCCESS;
27 : }
28 :
29 0 : HcclResult AllReduceDoublingDirect::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
30 : {
31 0 : HCCL_INFO(
32 : "[AllReduceDoublingDirect] runAsync rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", rank,
33 : rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
34 :
35 : // 基本的检查
36 0 : CHK_RET(SimpleCheck(rank, rankSize, links));
37 :
38 : // 判断 ranksize == 1 场景，把数据从userIn直接拷到userOut
39 0 : u64 totalSize = count_ * SIZE_TABLE[dataType_];
40 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, totalSize);
41 0 : DeviceMem userMemOut = DeviceMem::create(opInfo_->outputAddr, totalSize);
42 0 : if (rankSize == 1) {
43 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, userMemOut, userMemIn, stream_));
44 0 : return HCCL_SUCCESS;
45 : }
46 :
47 : // 设置Slices
48 0 : if (slices_.size() != 0) {
49 0 : HCCL_WARNING("[AllReduceDoublingDirect] slices_ will be not used in executor.");
50 : }
51 :
52 : // 执行算法
53 0 : CHK_RET(RunAllReduce(rank, rankSize, links));
54 :
55 0 : HCCL_INFO("[AllReduceDoublingDirect] finished: rank[%u] ranksize[%u]", rank, rankSize);
56 0 : return HCCL_SUCCESS;
57 0 : }
58 :
59 0 : HcclResult AllReduceDoublingDirect::RunAllReduce(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
60 : {
61 0 : u32 nSteps = static_cast<u32>(log2(rankSize));
62 0 : u64 totalSize = count_ * SIZE_TABLE[dataType_];
63 0 : DeviceMem commMemIn = DeviceMem::create(inputMem_.ptr(), totalSize);
64 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), totalSize);
65 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, totalSize);
66 0 : DeviceMem userMemOut = DeviceMem::create(opInfo_->outputAddr, totalSize);
67 :
68 0 : DeviceMem src;
69 0 : DeviceMem dst;
70 :
71 : // 第一步：把本端的数据从userIn拷到cclIn
72 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, commMemIn, userMemIn, stream_));
73 :
74 0 : u32 neighbor = rank ^ (1 << 0);
75 0 : CHK_PTR_NULL(links[neighbor]);
76 :
77 0 : const u32 ONE_STEPS = 1;
78 0 : if (nSteps == ONE_STEPS) {
79 : // 把本端的数据从userIn拷到userOut
80 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, userMemOut, userMemIn, stream_));
81 : // 把数据从远端的cclIn读到本端的userOut
82 0 : void* remMemPtr = nullptr;
83 0 : CHK_RET(links[neighbor]->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
84 0 : src = DeviceMem::create(static_cast<u8*>(remMemPtr), totalSize);
85 0 : CHK_RET(RunInlineReduce(userMemOut, src, links[neighbor]));
86 0 : return HCCL_SUCCESS;
87 : }
88 : // 把数据从本端的userIn写到对端的cclIn
89 0 : void* remMemPtr = nullptr;
90 0 : CHK_RET(links[neighbor]->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
91 0 : dst = DeviceMem::create(static_cast<u8*>(remMemPtr), totalSize);
92 0 : CHK_RET(RunInlineReduce(dst, userMemIn, links[neighbor]));
93 :
94 : // 只需要一块cclbuffer
95 0 : const u32 TWO_STEPS = 2;
96 0 : if (nSteps == TWO_STEPS) {
97 : // 把数据从本端的cclIn拷到本端的userOut
98 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, userMemOut, commMemIn, stream_));
99 :
100 : // 从远端的cclIn读到本端的userOut
101 0 : neighbor = rank ^ (1 << 1);
102 0 : CHK_PTR_NULL(links[neighbor]);
103 0 : CHK_RET(links[neighbor]->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
104 0 : src = DeviceMem::create(static_cast<u8*>(remMemPtr), totalSize);
105 0 : CHK_RET(RunInlineReduce(userMemOut, src, links[neighbor]));
106 : }
107 : // 必须有两块memory
108 0 : const u32 THREE_STEPS = 3;
109 0 : if (nSteps == THREE_STEPS) {
110 : // 把数据从本端的cclIn拷到本端的cclOut
111 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, commMemOut, commMemIn, stream_));
112 :
113 : // 从远端的cclIn读到本端的cclOut
114 0 : neighbor = rank ^ (1 << 1);
115 0 : CHK_PTR_NULL(links[neighbor]);
116 0 : CHK_RET(links[neighbor]->GetRemoteMem(UserMemType::INPUT_MEM, &remMemPtr));
117 0 : src = DeviceMem::create(static_cast<u8*>(remMemPtr), totalSize);
118 0 : CHK_RET(RunInlineReduce(commMemOut, src, links[neighbor]));
119 :
120 : // 把数据从本端的cclOut拷到本端的userOut
121 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, userMemOut, commMemOut, stream_));
122 :
123 : // 最后一步：从远端的cclOut读到本端的userOut
124 0 : const u32 LAST_STEPS = 2;
125 0 : neighbor = rank ^ (1 << LAST_STEPS);
126 0 : CHK_PTR_NULL(links[neighbor]);
127 0 : CHK_RET(links[neighbor]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
128 0 : src = DeviceMem::create(static_cast<u8*>(remMemPtr), totalSize);
129 0 : CHK_RET(RunInlineReduce(userMemOut, src, links[neighbor]));
130 : }
131 0 : return HCCL_SUCCESS;
132 0 : }
133 :
134 0 : HcclResult AllReduceDoublingDirect::RunInlineReduce(hccl::DeviceMem& dst, const hccl::DeviceMem& src, const LINK& link)
135 : {
136 0 : CHK_RET(link->TxAck(stream_));
137 0 : CHK_RET(link->RxAck(stream_));
138 :
139 0 : CHK_RET(HcclReduceAsync(
140 : dispatcher_, static_cast<void*>(src.ptr()), count_, dataType_, reductionOp_, stream_,
141 : static_cast<void*>(dst.ptr()), link->GetRemoteRank(), link->GetLinkType(), INLINE_REDUCE_BIT));
142 :
143 0 : CHK_RET(link->TxDataSignal(stream_));
144 0 : CHK_RET(link->RxDataSignal(stream_));
145 :
146 0 : HCCL_INFO("[AllReduceDoublingDirect] RunInlineReduce finished");
147 0 : return HCCL_SUCCESS;
148 : }
149 :
150 0 : HcclResult AllReduceDoublingDirect::SimpleCheck(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
151 : {
152 : // 判断stream, dispatcher是否为空
153 0 : CHK_SMART_PTR_NULL(dispatcher_);
154 0 : CHK_PTR_NULL(stream_.ptr());
155 :
156 : // 当前只支持ranksize <= 8
157 0 : CHK_PRT_RET(rankSize > 8, HCCL_ERROR("[AllReduceDoublingDirect] only support rankSize <= 8"), HCCL_E_PTR);
158 :
159 : // 判断Memory是否为空
160 0 : CHK_PRT_RET(!inputMem_, HCCL_ERROR("[AllReduceDoublingDirect] rank[%u] inputmem is null", rank), HCCL_E_PTR);
161 0 : CHK_PRT_RET(!outputMem_, HCCL_ERROR("[AllReduceDoublingDirect] rank[%u] outputmem is null", rank), HCCL_E_PTR);
162 0 : CHK_PRT_RET(
163 : inputMem_ == outputMem_, HCCL_ERROR("rank[%u] inputMem and outputMem should be different", rank), HCCL_E_PARA);
164 :
165 : // 判断links数量是否正确
166 0 : CHK_PRT_RET(
167 : links.size() < rankSize,
168 : HCCL_ERROR(
169 : "[AllReduceDoublingDirect] rank[%u] link size[%llu] is less than "
170 : "rank size[%u]",
171 : rank, links.size(), rankSize),
172 : HCCL_E_PARA);
173 :
174 : // 判断rankSize是否为2的幂次
175 0 : CHK_PRT_RET(
176 : (rankSize & (rankSize - 1)) != 0,
177 : HCCL_ERROR(
178 : "[AllReduceDoublingDirect] rankSize must be power of 2, "
179 : "but get rankSize=%u",
180 : rankSize),
181 : HCCL_E_PARA);
182 0 : return HCCL_SUCCESS;
183 : }
184 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING_DIRECT, AllReduceDoublingDirect);
185 : } // namespace hccl
|
