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