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 "alg_template_register.h"
12 : #include "reduce_ring.h"
13 :
14 : namespace hccl {
15 1 : ReduceRing::ReduceRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
16 :
17 2 : ReduceRing::~ReduceRing() {}
18 :
19 1 : HcclResult ReduceRing::Prepare(u64 reduceAttrBitMap, [[maybe_unused]] HcomCollOpInfo* opInfo)
20 : {
21 1 : reduceAttr_ = reduceAttrBitMap;
22 1 : return HCCL_SUCCESS;
23 : }
24 :
25 : // reduce算法的入口函数
26 : HcclResult
27 0 : ReduceRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<std::shared_ptr<Transport>>& links)
28 : {
29 0 : CHK_SMART_PTR_NULL(dispatcher_);
30 0 : CHK_PTR_NULL(stream_.ptr());
31 0 : bool bRetNull = (!outputMem_ || !inputMem_);
32 0 : CHK_PRT_RET(
33 : bRetNull, HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] inputmem or outputmem is null", rank), HCCL_E_PARA);
34 :
35 0 : HcclResult ret = HCCL_SUCCESS;
36 0 : HCCL_INFO(
37 : "ReduceRing run: rank[%u] totalrank[%u] root[%u] inputmem[%p] output[%p] count[%llu]", rank, rankSize, root_,
38 : inputMem_.ptr(), outputMem_.ptr(), count_);
39 :
40 : // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
41 0 : if (rankSize == 1) {
42 0 : if (inputMem_ != outputMem_) {
43 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
44 : }
45 0 : return ret;
46 : }
47 :
48 0 : HCCL_DEBUG("[ReduceRing][RunAsync]rankSize is %u", rankSize);
49 : // 创建reducer & sender
50 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
51 0 : CHK_SMART_PTR_NULL(senderInfo_);
52 :
53 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
54 0 : CHK_SMART_PTR_NULL(reducerInfo_);
55 :
56 : // 获取ring algorithm所需的通信连接
57 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
58 0 : u32 ringNextRank = (rank + 1) % rankSize;
59 :
60 0 : if (links.size() < rankSize) {
61 0 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] Link size is less than rank size", rank);
62 0 : return HCCL_E_INTERNAL;
63 : }
64 :
65 0 : linkLeft_ = links[ringPrevRank];
66 0 : CHK_SMART_PTR_NULL(linkLeft_);
67 :
68 0 : linkRight_ = links[ringNextRank];
69 0 : CHK_SMART_PTR_NULL(linkRight_);
70 :
71 0 : scratch_ = DeviceMem::create(inputMem_.ptr(), inputMem_.size());
72 :
73 0 : u32 dataSize = DataUnitSize(dataType_);
74 0 : if (dataSize == 0) {
75 0 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] unit data size is zero", rank);
76 0 : return HCCL_E_INTERNAL;
77 : }
78 :
79 : // 计算以chunk_size_为最大处理单元时，能够处理的最大数据个数
80 : // 每轮需要操作的数据个数
81 0 : CHK_RET(linkLeft_->TxAck(stream_));
82 0 : CHK_RET(linkRight_->RxAck(stream_));
83 0 : u64 length = count_ * dataSize;
84 :
85 0 : if (rank == root_) {
86 : // root节点只接收数据
87 0 : DeviceMem localSrc = scratch_.range(0, length);
88 0 : DeviceMem dst = outputMem_.range(0, length);
89 0 : HCCL_DEBUG("rank [%u] recv data offset[%llu] size[%llu] reduce", rank, 0, length);
90 :
91 : // 需要从前一节点接收数据,替换reducer接口
92 0 : ret = reducerInfo_->run(
93 0 : dispatcher_, linkLeft_, baseOffset_, localSrc, dst, dst, stream_, DstMemType::RESULT_OUTPUT_MEM);
94 0 : CHK_PRT_RET(
95 : ret != HCCL_SUCCESS,
96 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] reduce data offset[%llu] size[%llu]", rank, 0, length), ret);
97 :
98 : // 给前一节点发送同步
99 0 : CHK_RET(linkLeft_->TxAck(stream_));
100 0 : CHK_RET(linkLeft_->RxWaitDone(stream_));
101 0 : } else if (ringPrevRank == root_) {
102 : // 本rank的前一节点是root节点，本rank数据拷贝到下一rank，不做reduce操作
103 : // 需要向下一节点拷贝的数据
104 0 : DeviceMem localSrc = scratch_.range(0, length);
105 :
106 : // 数据拷贝和向下一节点发送
107 0 : HCCL_DEBUG("rank [%u] send offset[%llu] size[%llu]", rank, 0, length);
108 :
109 0 : ret = senderInfo_->run(linkRight_, baseOffset_, localSrc, stream_);
110 0 : CHK_PRT_RET(
111 : ret != HCCL_SUCCESS,
112 : HCCL_ERROR(
113 : "[ReduceRing][RunAsync]rank[%u] send scratch offset[%llu] size[%llu] "
114 : "failed",
115 : rank, baseOffset_, length),
116 : ret);
117 :
118 : // 等待后一节点同步信号
119 0 : CHK_RET(linkRight_->RxAck(stream_));
120 0 : CHK_RET(linkRight_->TxWaitDone(stream_));
121 0 : } else {
122 : // 其余节点，先接收数据，和自身数据进行reduce操作，结果放入tx中，发送至下一节点
123 : // 剩余需要处理的数据大于满chunk size时，以chunksize为处理单位，否则直接处理剩余数据
124 :
125 : // 接收到的数据和scratch数据运算后，放入output
126 0 : DeviceMem localSrc = scratch_.range(0, length);
127 0 : DeviceMem dst = outputMem_.range(0, length);
128 :
129 : // 用reduce接口封装
130 0 : HCCL_DEBUG("rank[%u] recv data reduce offset[%llu] size[%llu]", rank, 0, length);
131 :
132 0 : ret = reducerInfo_->run(
133 0 : dispatcher_, linkLeft_, baseOffset_, localSrc, localSrc, dst, stream_, DstMemType::RESULT_INPUT_MEM);
134 0 : CHK_PRT_RET(
135 : ret != HCCL_SUCCESS,
136 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] reducer offset[%llu] size[%llu] failed", rank, 0, length), ret);
137 :
138 : // 给前一节点发送同步
139 0 : CHK_RET(linkLeft_->TxAck(stream_));
140 0 : CHK_RET(linkLeft_->RxWaitDone(stream_));
141 :
142 : // tx数据向下一个节点发送
143 : // 需要再封装接口，只把数据发到tx_mem,send_only
144 0 : HCCL_DEBUG("rank[%u] send localSrc offset[%llu] size[%llu]", rank, 0, length);
145 :
146 0 : ret = senderInfo_->run(linkRight_, baseOffset_, localSrc, stream_);
147 0 : CHK_PRT_RET(
148 : ret != HCCL_SUCCESS,
149 : HCCL_ERROR("[ReduceRing][RunAsync]rank[%u] sender offset[%llu] failed", rank, baseOffset_), ret);
150 :
151 : // 等待后一节点同步信号
152 0 : CHK_RET(linkRight_->RxAck(stream_));
153 0 : CHK_RET(linkRight_->TxWaitDone(stream_));
154 0 : }
155 0 : CHK_RET(linkRight_->TxDataSignal(stream_));
156 0 : CHK_RET(linkLeft_->RxDataSignal(stream_));
157 0 : HCCL_INFO("ReduceRing finished: rank[%u]", rank);
158 0 : return HCCL_SUCCESS;
159 : }
160 : HcclResult
161 0 : ReduceRing::GetNslbAdjInfo(const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
162 : {
163 0 : if (rankSize == 1) {
164 0 : return HCCL_E_NOT_SUPPORT;
165 : }
166 0 : u32 ringNextRank = (rank + 1) % rankSize;
167 0 : LINK nslbNext = links[ringNextRank];
168 :
169 0 : NslbDpAdjInfo adjInfoStep = {};
170 0 : nslbAdjInfo.dstRankNum = 1;
171 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
172 0 : adjInfoStep.phaseId = 1;
173 0 : adjInfoStep.rev = 0;
174 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
175 :
176 0 : return HCCL_SUCCESS;
177 0 : }
178 :
179 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCE_RING, ReduceRing);
180 : } // namespace hccl
|
