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 "broadcast_ring.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 1 : BroadcastRing::BroadcastRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
16 :
17 2 : BroadcastRing::~BroadcastRing() {}
18 :
19 : HcclResult
20 0 : BroadcastRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<std::shared_ptr<Transport>>& links)
21 : {
22 0 : CHK_SMART_PTR_NULL(dispatcher_);
23 0 : CHK_PTR_NULL(stream_.ptr());
24 0 : HCCL_INFO("BroadcastRing run: rank[%u] totalrank[%u] count[%llu]", rank, rankSize, count_);
25 :
26 0 : if (rankSize == 1) {
27 0 : return HCCL_SUCCESS;
28 : }
29 :
30 : // 获取ring algorithm所需的通信连接
31 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
32 0 : u32 ringNextRank = (rank + 1) % rankSize;
33 :
34 0 : if (links.size() < rankSize) {
35 0 : HCCL_ERROR("[BroadcastRing][RunAsync]rank[%u] linksize[%llu] is less than rank size", rank, links.size());
36 0 : return HCCL_E_INTERNAL;
37 : }
38 0 : linkLeft_ = links[ringPrevRank];
39 0 : CHK_SMART_PTR_NULL(linkLeft_);
40 :
41 0 : linkRight_ = links[ringNextRank];
42 0 : CHK_SMART_PTR_NULL(linkRight_);
43 :
44 0 : u32 unitSize = DataUnitSize(dataType_);
45 0 : if (unitSize == 0) {
46 0 : HCCL_ERROR("[BroadcastRing][RunAsync]rank[%u] unit data size is zero", rank);
47 0 : return HCCL_E_INTERNAL;
48 : }
49 :
50 0 : if (rank == root_) {
51 0 : CHK_PRT_RET(!inputMem_, HCCL_ERROR("[BroadcastRing][RunAsync]rank[%u] inputmem is null", rank), HCCL_E_PTR);
52 0 : scratch_ = DeviceMem::create(inputMem_.ptr(), inputMem_.size());
53 : } else {
54 0 : CHK_PRT_RET(!outputMem_, HCCL_ERROR("[BroadcastRing][RunAsync]rank[%u] outputmem is null", rank), HCCL_E_PTR);
55 0 : scratch_ = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
56 : }
57 0 : HCCL_DEBUG("root[%u] scratch[%p] memsize[%llu]", root_, scratch_.ptr(), scratch_.size());
58 :
59 : // 所有的数据平均到每个rank上
60 0 : u64 sizeAvg = ((count_ + rankSize - 1) / rankSize) * unitSize;
61 :
62 0 : u64 sizePerSlice = AlgTemplateBase::RoundUpWithDivisor(sizeAvg, HCCL_MIN_SLICE_ALIGN);
63 0 : u64 sizePerRound = 0;
64 0 : HCCL_DEBUG(
65 : "bcast total count[%llu] sizeAverage[%llu] sizePerSlice after aligns[%llu]", count_, sizeAvg, sizePerSlice);
66 0 : CHK_RET(linkLeft_->TxAck(stream_));
67 0 : CHK_RET(linkRight_->RxAck(stream_));
68 :
69 0 : HcclResult ret = HCCL_SUCCESS;
70 0 : if (rank == root_) { // root节点，数据发送下一个节点
71 0 : DeviceMem localSrc;
72 :
73 0 : u64 sizeResidue = count_ * unitSize;
74 0 : for (u32 round = 0; round < rankSize; round++, sizeResidue -= sizePerRound) { // 固定循环次数，避免子图复用出错
75 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
76 0 : localSrc = scratch_.range(count_ * unitSize - sizeResidue, sizePerRound);
77 :
78 : // 数据向下一个rank发送
79 0 : u64 dstOffset = count_ * unitSize - sizeResidue + baseOffset_;
80 0 : ret = linkRight_->TxAsync(UserMemType::OUTPUT_MEM, dstOffset, localSrc.ptr(), sizePerRound, stream_);
81 0 : CHK_PRT_RET(
82 : ret != HCCL_SUCCESS,
83 : HCCL_ERROR(
84 : "[BroadcastRing][RunAsync]root[%u] right link tx async srcmem "
85 : "[%p] to offset[%llu] failed",
86 : rank, localSrc.ptr(), dstOffset),
87 : ret);
88 0 : HCCL_DEBUG(
89 : "root rank[%u] send scratchmem[%p] to offset[%llu] sendsize[%llu]", rank, localSrc.ptr(), dstOffset,
90 : sizePerRound);
91 :
92 : // 等待后一节点同步信号
93 0 : CHK_RET(linkRight_->RxAck(stream_));
94 0 : ret = linkRight_->TxWaitDone(stream_);
95 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[BroadcastRing][RunAsync]TxWaitDone failed"), ret);
96 : }
97 0 : } else if (ringNextRank == root_) { // 最后一个节点，只接收数据
98 0 : DeviceMem localSrc;
99 :
100 0 : u64 sizeResidue = count_ * unitSize;
101 0 : for (u32 round = 0; round < rankSize; round++, sizeResidue -= sizePerRound) { // 固定循环次数，避免子图复用出错
102 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
103 0 : localSrc = scratch_.range(count_ * unitSize - sizeResidue, sizePerRound);
104 : // 从前一节点接收数据
105 0 : u64 dstOffset = count_ * unitSize - sizeResidue + baseOffset_;
106 0 : ret = linkLeft_->RxAsync(UserMemType::OUTPUT_MEM, dstOffset, localSrc.ptr(), sizePerRound, stream_);
107 0 : CHK_PRT_RET(
108 : ret != HCCL_SUCCESS,
109 : HCCL_ERROR(
110 : "[BroadcastRing][RunAsync]rank[%u] rx data from offset[%llu] "
111 : "with localmem[%p] failed",
112 : rank, dstOffset, localSrc.ptr()),
113 : ret);
114 0 : HCCL_DEBUG(
115 : "last rank[%u] rx_sync from range[%llu] with localmem[%p] size:[%llu] ", rank, dstOffset,
116 : localSrc.ptr(), sizePerRound);
117 :
118 : // 给前一节点发送同步
119 0 : ret = linkLeft_->TxAck(stream_);
120 0 : CHK_PRT_RET(
121 : ret != HCCL_SUCCESS, HCCL_ERROR("[BroadcastRing][RunAsync]rank[%u] left link tx ack failed", rank),
122 : ret);
123 0 : ret = linkLeft_->RxWaitDone(stream_);
124 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[BroadcastRing][RunAsync]RxWaitDone failed"), ret);
125 : }
126 0 : } else { // 非root节点或者尾节点，先接收来自前一节点的数据，再发送至下一结点
127 0 : DeviceMem localSrc;
128 :
129 0 : u64 sizeResidue = count_ * unitSize;
130 0 : for (u32 round = 0; round < rankSize; round++, sizeResidue -= sizePerRound) { // 固定循环次数，避免子图复用出错
131 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
132 : // 需要从前一节点接收数据
133 0 : localSrc = scratch_.range(count_ * unitSize - sizeResidue, sizePerRound);
134 0 : u64 dstOffset = count_ * unitSize - sizeResidue + baseOffset_;
135 0 : ret = linkLeft_->RxAsync(UserMemType::OUTPUT_MEM, dstOffset, localSrc.ptr(), sizePerRound, stream_);
136 0 : CHK_PRT_RET(
137 : ret != HCCL_SUCCESS,
138 : HCCL_ERROR(
139 : "[BroadcastRing][RunAsync]rank[%u] left link rx sync from offset[%llu] with "
140 : "localmem[%p] failed",
141 : rank, dstOffset, localSrc.ptr()),
142 : ret);
143 0 : HCCL_DEBUG(
144 : "rank[%u] rx_sync from range[%u] with localmem[%p] size[%llu] ", rank, dstOffset, localSrc.ptr(),
145 : sizePerRound);
146 :
147 : // 给前一节点发送同步
148 0 : CHK_RET(linkLeft_->TxAck(stream_));
149 0 : ret = linkLeft_->RxWaitDone(stream_);
150 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[BroadcastRing][RunAsync]RxWaitDone failed"), ret);
151 :
152 : // 数据向下一个节点发送
153 0 : CHK_RET(linkRight_->TxAsync(UserMemType::OUTPUT_MEM, dstOffset, localSrc.ptr(), sizePerRound, stream_));
154 :
155 0 : HCCL_DEBUG(
156 : "rank[%u] tx_sync from localmem[%p] to offset[%llu] size[%llu] ", rank, localSrc.ptr(), dstOffset,
157 : sizePerRound);
158 : // 等待后一节点同步信号
159 0 : CHK_RET(linkRight_->RxAck(stream_));
160 0 : ret = linkRight_->TxWaitDone(stream_);
161 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[BroadcastRing][RunAsync]TxWaitDone failed"), ret);
162 : }
163 0 : }
164 0 : CHK_RET(linkRight_->TxDataSignal(stream_));
165 :
166 0 : CHK_RET(linkLeft_->RxDataSignal(stream_));
167 0 : HCCL_INFO("BroadcastRing finished: rank[%u] end count[%llu]", rank, count_);
168 0 : return HCCL_SUCCESS;
169 : }
170 0 : HcclResult BroadcastRing::GetNslbAdjInfo(
171 : [[maybe_unused]] const u32 rank, [[maybe_unused]] const u32 rankSize,
172 : [[maybe_unused]] const std::vector<LINK>& links, [[maybe_unused]] AdjInfo& nslbAdjInfo)
173 : {
174 0 : return HCCL_SUCCESS;
175 : }
176 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_BROADCAST_RING, BroadcastRing);
177 : } // namespace hccl
|
