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