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 "send_receive.h"
12 :
13 : namespace hccl {
14 0 : SendReceive::SendReceive(
15 : const HcclDispatcher dispatcher, const std::shared_ptr<Transport>& link, const u32 peerRank, const u64 chunkNum,
16 0 : bool retryEnable)
17 : : AlgTemplateBase(dispatcher),
18 0 : transLink_(link),
19 0 : peerRank_(peerRank),
20 0 : chunkSize_(chunkNum),
21 0 : retryEnable_(retryEnable)
22 0 : {}
23 :
24 0 : SendReceive::~SendReceive() {}
25 :
26 0 : HcclResult SendReceive::SendPrepare(const DeviceMem& inputMem, const u32 destRank, const Stream& stream)
27 : {
28 : /* 参数赋值 */
29 0 : inputMem_ = inputMem;
30 0 : stream_ = stream;
31 0 : peerRank_ = destRank;
32 0 : return HCCL_SUCCESS;
33 : }
34 :
35 0 : HcclResult SendReceive::ReceivePrepare(const DeviceMem& outputMem, const u32 srcRank, const Stream& stream)
36 : {
37 : /* 参数赋值 */
38 0 : outputMem_ = outputMem;
39 0 : stream_ = stream;
40 0 : peerRank_ = srcRank;
41 :
42 0 : return HCCL_SUCCESS;
43 : }
44 :
45 0 : HcclResult SendReceive::SendRunAsync()
46 : {
47 0 : if (!inputMem_) {
48 0 : HCCL_ERROR("[SendReceive][SendRunAsync]SendRunAsync inputmem is null");
49 0 : return HCCL_E_PTR;
50 : }
51 :
52 0 : CHK_SMART_PTR_NULL(transLink_);
53 0 : u64 sizePerRound = 0;
54 0 : u64 sizePerSlice = chunkSize_;
55 0 : u64 length = inputMem_.size();
56 0 : u64 offset = 0;
57 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
58 0 : HcclResult ret = transLink_->TxAck(stream_);
59 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][SendRunAsync]tx ack run failed"), ret);
60 0 : ret = transLink_->RxAck(stream_);
61 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][SendRunAsync]rx ack run failed"), ret);
62 :
63 0 : offset += sizePerRound;
64 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
65 0 : void* localAddr = static_cast<u8*>(inputMem_.ptr()) + offset;
66 0 : HCCL_DEBUG("tx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
67 :
68 0 : ret = transLink_->TxAsync(UserMemType::OUTPUT_MEM, offset, localAddr, sizePerRound, stream_);
69 0 : CHK_PRT_RET(
70 : ret != HCCL_SUCCESS,
71 : HCCL_ERROR(
72 : "[SendReceive][SendRunAsync]tx async offset[%llu] "
73 : "size[%llu] failed",
74 : offset, sizePerRound),
75 : ret);
76 :
77 0 : ret = transLink_->RxAsync(UserMemType::OUTPUT_MEM, 0, nullptr, 0, stream_);
78 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][ReceiveRunAsync]tx async failed"), ret);
79 :
80 0 : ret = transLink_->RxWaitDone(stream_);
81 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][SendRunAsync]RxWaitDone failed"), ret);
82 0 : ret = transLink_->TxWaitDone(stream_);
83 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][SendRunAsync]TxWaitDone failed"), ret);
84 : }
85 0 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult SendReceive::ReceiveRunAsync()
89 : {
90 0 : if (!outputMem_) {
91 0 : HCCL_ERROR("[SendReceive][ReceiveRunAsync]ReceiveRunAsync outputmem is null");
92 0 : return HCCL_E_PTR;
93 : }
94 0 : CHK_SMART_PTR_NULL(transLink_);
95 :
96 0 : HCCL_DEBUG("[SendReceive][ReceiveRunAsync]ReceiveRunAsync begins");
97 0 : u64 sizePerRound = 0;
98 0 : u64 sizePerSlice = chunkSize_;
99 0 : u64 length = outputMem_.size();
100 0 : u64 offset = 0;
101 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
102 0 : HcclResult ret = transLink_->TxAck(stream_);
103 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][ReceiveRunAsync]tx ack failed"), ret);
104 0 : ret = transLink_->RxAck(stream_);
105 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][ReceiveRunAsync]rx ack failed"), ret);
106 :
107 0 : offset += sizePerRound;
108 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
109 0 : void* localAddr = static_cast<u8*>(outputMem_.ptr()) + offset;
110 0 : HCCL_DEBUG("rx async outputmem's offset[%llu] size[%llu]", offset, sizePerRound);
111 :
112 0 : ret = transLink_->RxAsync(UserMemType::OUTPUT_MEM, offset, localAddr, sizePerRound, stream_);
113 0 : CHK_PRT_RET(
114 : ret != HCCL_SUCCESS,
115 : HCCL_ERROR(
116 : "[SendReceive][ReceiveRunAsync]rx async with offset[%llu] "
117 : "size[%llu] failed",
118 : offset, sizePerRound),
119 : ret);
120 :
121 0 : ret = transLink_->TxAsync(UserMemType::OUTPUT_MEM, 0, nullptr, 0, stream_);
122 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][ReceiveRunAsync]tx async failed"), ret);
123 : }
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 0 : HcclResult SendReceive::BatchSendRunAsync()
128 : {
129 0 : if (!inputMem_) {
130 0 : HCCL_ERROR("[SendReceive][BatchSendRunAsync] BatchSendRunAsync inputmem is null");
131 0 : return HCCL_E_PTR;
132 : }
133 :
134 0 : CHK_SMART_PTR_NULL(transLink_);
135 0 : u64 sizePerRound = 0;
136 0 : u64 sizePerSlice = chunkSize_;
137 0 : u64 length = inputMem_.size();
138 0 : u64 offset = 0;
139 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
140 0 : HcclResult ret = transLink_->TxPrepare(stream_);
141 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][BatchSendRunAsync]tx ack run failed"), ret);
142 :
143 0 : offset += sizePerRound;
144 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
145 0 : void* localAddr = static_cast<u8*>(inputMem_.ptr()) + offset;
146 0 : HCCL_DEBUG("tx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
147 :
148 0 : ret = transLink_->TxData(UserMemType::OUTPUT_MEM, offset, localAddr, sizePerRound, stream_);
149 0 : CHK_PRT_RET(
150 : ret != HCCL_SUCCESS,
151 : HCCL_ERROR(
152 : "[SendReceive][BatchSendRunAsync]tx async offset[%llu] "
153 : "size[%llu] failed",
154 : offset, sizePerRound),
155 : ret);
156 :
157 0 : ret = transLink_->TxDone(stream_);
158 :
159 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][BatchSendRunAsync]TxWaitDone failed"), ret);
160 : }
161 0 : return HCCL_SUCCESS;
162 : }
163 :
164 0 : HcclResult SendReceive::BatchReceiveRunAsync()
165 : {
166 0 : if (!outputMem_) {
167 0 : HCCL_ERROR("[SendReceive][ReceiveRunAsync]ReceiveRunAsync outputmem is null");
168 0 : return HCCL_E_PTR;
169 : }
170 0 : CHK_SMART_PTR_NULL(transLink_);
171 :
172 0 : u64 sizePerRound = 0;
173 0 : u64 sizePerSlice = chunkSize_;
174 0 : u64 length = outputMem_.size();
175 0 : u64 offset = 0;
176 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
177 0 : HcclResult ret = transLink_->RxPrepare(stream_);
178 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendReceive][BatchReceiveRunAsync]rx ack failed"), ret);
179 :
180 0 : offset += sizePerRound;
181 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
182 0 : void* localAddr = static_cast<u8*>(outputMem_.ptr()) + offset;
183 0 : HCCL_DEBUG("rx async outputmem's offset[%llu] size[%llu]", offset, sizePerRound);
184 :
185 0 : ret = transLink_->RxData(UserMemType::INPUT_MEM, offset, localAddr, sizePerRound, stream_);
186 0 : CHK_PRT_RET(
187 : ret != HCCL_SUCCESS,
188 : HCCL_ERROR(
189 : "[SendReceive][BatchReceiveRunAsync]rx async with offset[%llu] "
190 : "size[%llu] failed",
191 : offset, sizePerRound),
192 : ret);
193 :
194 0 : ret = transLink_->RxDone(stream_);
195 0 : CHK_PRT_RET(
196 : ret != HCCL_SUCCESS,
197 : HCCL_ERROR(
198 : "[SendReceive][BatchReceiveRunAsync]TxDataSignal offset[%llu]"
199 : "size[%llu] failed",
200 : offset, sizePerRound),
201 : ret);
202 : }
203 0 : return HCCL_SUCCESS;
204 : }
205 : } // namespace hccl
|
