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 : #include <chrono>
11 : #include "ub_conn_lite.h"
12 : #include "log.h"
13 : #include "exception_util.h"
14 : #include "udma_data_struct.h"
15 : #include "internal_exception.h"
16 : #include "string_util.h"
17 : #include "binary_stream.h"
18 : #include "data_type.h"
19 :
20 : constexpr u32 MAX_LOG_TIMEOUT_MS = 500;
21 : namespace Hccl {
22 : constexpr u32 ADDR_BIT_OFFSET = 32;
23 : constexpr u32 SQE_SIZE_128 = 128;
24 : constexpr u32 SQE_SIZE_64 = 64;
25 : constexpr u32 SQE_INLINE_DATA_SIZE = 16;
26 : constexpr u32 RAW_SIZE = 16;
27 : constexpr u32 RMT_EID_BYTE_SIZE = 16;
28 : constexpr u32 PI_NUM_TWO = 2;
29 : constexpr u32 WRITE_WITH_NOTIFY_OPCODE = 0x5;
30 : constexpr u32 ADDR_BIT_LOW = 0xffffffff;
31 : constexpr u32 UB_DMA_MAX_READ_WEITE_SIZE = 256 * 1024 * 1024; // Byte, UB协议一次传输的最大size
32 : constexpr u32 UB_RELAX_ORDER = 0x1; // Relax Order表示当前SQE与后续Strong Order SQE有保序要求
33 : constexpr u32 UB_STRONG_ORDER = 0x2; // Strong Order表示当前SQE有保序要求，该SQE不能超越前面的Relax Order SQE
34 :
35 : static std::map<DataType, u32> g_ubmaDataTypeMap
36 : = {{DataType::INT8, 0x0}, {DataType::INT16, 0x1}, {DataType::INT32, 0x2}, {DataType::UINT8, 0x3},
37 : {DataType::UINT16, 0x4}, {DataType::UINT32, 0x5}, {DataType::FP16, 0x6}, {DataType::FP32, 0x7},
38 : {DataType::BFP16, 0x8}, {DataType::BF16_SAT, 0x9}};
39 :
40 : static std::map<ReduceOp, u32> g_ubmaDataOpMap = {{ReduceOp::SUM, 0xA}, {ReduceOp::MAX, 0x8}, {ReduceOp::MIN, 0x9}};
41 :
42 43 : void UbConnLite::FillCommSqe(UdmaSqeCommon *sqe, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg, u32 opCode,
43 : SlicePosition slicePos)
44 : {
45 43 : u32 cqeEn = (cfg.cqeEn && (slicePos == SlicePosition::LAST || slicePos == SlicePosition::ONLY)) ? 1 : 0;
46 43 : sqe->cqe = cqeEn;
47 43 : sqe->owner = (pi == (sqDepth_ - 1)) ? 1 : 0;
48 43 : sqe->opcode = opCode;
49 43 : sqe->tpn = tpn_;
50 :
51 : // 当前片是ONLY片(只有一片的情况)和最后一片的情况，全严格保序
52 43 : if (slicePos == SlicePosition::ONLY || slicePos == SlicePosition::LAST) {
53 33 : sqe->placeOdr = UB_STRONG_ORDER;
54 33 : sqe->compOrder = 1;
55 33 : sqe->fence = 1;
56 : } else {
57 : // 中间片写死配置，第一片由全局cfg配置
58 10 : sqe->placeOdr = (slicePos == SlicePosition::MIDDLE) ? UB_RELAX_ORDER : cfg.placeOdr;
59 10 : sqe->compOrder = (slicePos == SlicePosition::MIDDLE) ? 0 : cfg.compOrder;
60 10 : sqe->fence = (slicePos == SlicePosition::MIDDLE) ? 0 : cfg.fence;
61 : }
62 :
63 43 : sqe->se = 1; // 表示是否使能solicited event
64 43 : sqe->rmtJettyType = 1; // 00 JFR 01:JETTY 10:jettyGroup 11:reserved
65 43 : s32 ret = memcpy_sp(sqe->rmtEid, RMT_EID_BYTE_SIZE, rmtEid_.raw, RAW_SIZE);
66 43 : if (UNLIKELY(ret != 0)) {
67 0 : HCCL_ERROR("UbConnLite::FillCommSqe FillCommSqe memcpy failed, ret=%d", ret);
68 0 : THROW<InternalException>(StringFormat("UbConnLite::FillCommSqe memcpy_sp failed, ret = %d", ret));
69 : }
70 :
71 43 : sqe->sgeNum = 1;
72 43 : sqe->targetHint = 0;
73 43 : sqe->rmtObjId = rmt.GetTokenId();
74 43 : sqe->tokenEn = 1;
75 43 : sqe->rmtTokenValue = rmt.GetTokenValue();
76 43 : sqe->rmtAddrLow = rmt.GetAddr() & ADDR_BIT_LOW;
77 43 : sqe->rmtAddrHigh = rmt.GetAddr() >> ADDR_BIT_OFFSET;
78 129 : HCCL_INFO("UbConnLite FillCommSqe UdmaSqeCommon slicePos[%d] sqe->cqe = %u, sqe->owner = %u sqe->opcode = %u, "
79 : "sqe->tpn = %u, sqe->rmtObjId = %u, sqe->rmtAddrLow = %u, sqe->rmtAddrHigh = %u, sqe->placeOdr = %u, "
80 : "sqe->compOrder = %u, sqe->fence = %u", slicePos, sqe->cqe, sqe->owner, sqe->opcode, sqe->tpn,
81 : sqe->rmtObjId, sqe->rmtAddrLow, sqe->rmtAddrHigh, sqe->placeOdr, sqe->compOrder, sqe->fence);
82 43 : }
83 :
84 23 : void UbConnLite::FillCommSqeReduceInfo(UdmaSqeCommon &sqeComm, ReduceOp reduceOp, DataType dataType, u32 udfType) const
85 : {
86 69 : HCCL_INFO("[UbConnLite::%s] start", __func__);
87 :
88 23 : sqeComm.inlinedata.udfData.udfType = udfType; // 0代表inline reduce
89 :
90 23 : if ((g_ubmaDataOpMap.find(reduceOp) != g_ubmaDataOpMap.end()) && (g_ubmaDataTypeMap.find(dataType) != g_ubmaDataTypeMap.end())) {
91 23 : sqeComm.inlinedata.udfData.reduceOp = g_ubmaDataOpMap.at(reduceOp);
92 23 : sqeComm.inlinedata.udfData.reduceType = g_ubmaDataTypeMap.at(dataType);
93 : } else {
94 0 : THROW<InvalidParamsException>(StringFormat("%s reduceOp[%s] or type[%s] is not supported.", __func__, reduceOp.Describe().c_str(), dataType.Describe().c_str()));
95 : }
96 :
97 : // udf字段是否有效
98 23 : sqeComm.udfFlag = 1;
99 :
100 69 : HCCL_INFO("[UbConnLite::%s] end, reduceOp[%s], reduceType[%s]", __func__, reduceOp.Describe().c_str(),
101 : dataType.Describe().c_str());
102 23 : }
103 :
104 10 : void UbConnLite::ProcessSlices(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, u32 maxSliceSize,
105 : std::function<void(const RmaBufSliceLite &, const RmtRmaBufSliceLite &, SlicePosition)> processOneSlice,
106 : DataType dataType) const
107 : {
108 : (void)dataType;
109 : // reduce操作需要保证切片大小是数据类型大小的整数倍
110 10 : u64 sliceSize = static_cast<u64>(maxSliceSize);
111 :
112 10 : u64 locBufSize = loc.GetSize();
113 10 : u64 sliceNum = locBufSize / sliceSize;
114 10 : u64 lastSliceSize = locBufSize % sliceSize;
115 :
116 10 : u64 totalSize = sliceNum * sliceSize;
117 :
118 10 : if (UNLIKELY(loc.GetAddr() > UINT64_MAX - totalSize || rmt.GetAddr() > UINT64_MAX - totalSize)) {
119 0 : THROW<InternalException>("integer overflow occurs");
120 : }
121 14 : for (u64 sliceIdx = 0; sliceIdx < sliceNum; sliceIdx++) {
122 4 : u64 offset = sliceIdx * sliceSize;
123 4 : u64 locAddr = loc.GetAddr() + offset;
124 4 : u64 rmtAddr = rmt.GetAddr() + offset;
125 :
126 12 : HCCL_INFO("[UbConnLite::%s] Slice[%llu]: offset=0x%llx, locAddr=0x%llx, rmtAddr=0x%llx, size=0x%llx",
127 : __func__, sliceIdx, offset, locAddr, rmtAddr, sliceSize);
128 :
129 4 : RmaBufSliceLite locSlice(locAddr, sliceSize, 0, loc.GetTokenId());
130 :
131 : RmtRmaBufSliceLite rmtSlice(rmtAddr, sliceSize, 0, rmt.GetTokenId(),
132 4 : rmt.GetTokenValue(), UINT32_MAX);
133 : SlicePosition slicePos;
134 4 : slicePos = (sliceIdx == 0) ? SlicePosition::FIRST : SlicePosition::MIDDLE;
135 4 : if ((sliceIdx == sliceNum - 1) && lastSliceSize == 0) {
136 : // SlicePosition::ONLY表示既是首片又是尾片的情况，只有一片的情况
137 0 : slicePos = (sliceIdx == 0) ? SlicePosition::ONLY : SlicePosition::LAST;
138 : }
139 4 : processOneSlice(locSlice, rmtSlice, slicePos);
140 : }
141 :
142 10 : if (lastSliceSize > 0) {
143 10 : RmaBufSliceLite lastLocSlice(loc.GetAddr() + sliceNum * sliceSize, lastSliceSize, 0, loc.GetTokenId());
144 :
145 10 : RmtRmaBufSliceLite lastRmtSlice(rmt.GetAddr() + sliceNum * sliceSize, lastSliceSize, 0, rmt.GetTokenId(),
146 20 : rmt.GetTokenValue(), UINT32_MAX);
147 : SlicePosition slicePos;
148 10 : slicePos = (sliceNum == 0) ? SlicePosition::ONLY : SlicePosition::LAST;
149 10 : processOneSlice(lastLocSlice, lastRmtSlice, slicePos);
150 10 : sliceNum++;
151 : }
152 :
153 30 : HCCL_INFO("[UbConnLite::%s] end, locBufSize[%u], sliceNUm[%u], sliceSize[%u], lastSliceSize[%u]", __func__,
154 : locBufSize, sliceNum, sliceSize, lastSliceSize);
155 10 : }
156 :
157 18 : void UbConnLite::ProcessSlicesWithNotify(
158 : const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, u32 maxSliceSize,
159 : std::function<void(const RmaBufSliceLite &, const RmtRmaBufSliceLite &, SlicePosition)> processOneSlice,
160 : std::function<void(const RmaBufSliceLite &, const RmtRmaBufSliceLite &, SlicePosition)> processOneSliceWithNotify,
161 : DataType dataType) const
162 : {
163 54 : HCCL_INFO("[UbConnLite::%s] start", __func__);
164 :
165 : // reduce操作需要保证切片大小是数据类型大小的整数倍
166 18 : u32 sliceSize = maxSliceSize;
167 18 : if (dataType != DataType::INVALID) {
168 12 : u32 dataTypeSize = DATA_TYPE_SIZE_MAP.at(dataType);
169 12 : sliceSize = maxSliceSize / dataTypeSize * dataTypeSize;
170 : }
171 :
172 18 : u32 locBufSize = loc.GetSize();
173 18 : u32 sliceNum = locBufSize / sliceSize;
174 18 : u32 lastSliceSize = locBufSize % sliceSize;
175 18 : if (sliceNum > 0 && lastSliceSize == 0) {
176 0 : sliceNum--;
177 0 : lastSliceSize = sliceSize;
178 : }
179 18 : u64 totalSize = static_cast<u64>(sliceNum) * static_cast<u64>(sliceSize);
180 18 : if (UNLIKELY(loc.GetAddr() > UINT64_MAX - totalSize || rmt.GetAddr() > UINT64_MAX - totalSize)) {
181 0 : THROW<InternalException>("integer overflow occurs");
182 : }
183 24 : for (u32 sliceIdx = 0; sliceIdx < sliceNum; sliceIdx++) {
184 6 : RmaBufSliceLite locSlice(loc.GetAddr() + sliceIdx * sliceSize, sliceSize, 0, loc.GetTokenId());
185 :
186 12 : RmtRmaBufSliceLite rmtSlice(rmt.GetAddr() + sliceIdx * sliceSize, sliceSize, 0, rmt.GetTokenId(),
187 6 : rmt.GetTokenValue(), UINT32_MAX);
188 : SlicePosition slicePos;
189 6 : slicePos = (sliceIdx == 0) ? SlicePosition::FIRST : SlicePosition::MIDDLE;
190 6 : processOneSlice(locSlice, rmtSlice, slicePos);
191 : }
192 :
193 18 : if (lastSliceSize > 0) {
194 18 : RmaBufSliceLite lastLocSlice(loc.GetAddr() + sliceNum * sliceSize, lastSliceSize, 0, loc.GetTokenId());
195 :
196 36 : RmtRmaBufSliceLite lastRmtSlice(rmt.GetAddr() + sliceNum * sliceSize, lastSliceSize, 0, rmt.GetTokenId(),
197 18 : rmt.GetTokenValue(), UINT32_MAX);
198 : SlicePosition slicePos;
199 18 : slicePos = (sliceNum == 0) ? SlicePosition::ONLY : SlicePosition::LAST;
200 18 : processOneSliceWithNotify(lastLocSlice, lastRmtSlice, slicePos);
201 : }
202 :
203 48 : HCCL_INFO("[UbConnLite::%s] end, locBufSize[%u], sliceNUm[%u], sliceSize[%u], lastSliceSize[%u]", __func__,
204 : locBufSize, sliceNum, sliceSize, lastSliceSize);
205 16 : }
206 :
207 20 : void UbConnLite::FillOneSqeWrite(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
208 : UdmaSqeWrite *sqe, UdmaSqOpcode opCode, SlicePosition slicePos)
209 : {
210 60 : HCCL_INFO("[UbConnLite::%s] start, loc size[%llu]", __func__, loc.GetSize());
211 :
212 20 : sqe->comm.inlineEn = 0;
213 20 : FillCommSqe(&(sqe->comm), rmt, cfg, opCode, slicePos);
214 20 : FillLocalSgeSqe(&(sqe->u.sge), loc);
215 20 : if (sqe->u.sge.length == 0) {
216 0 : sqe->comm.sgeNum = 0;
217 : }
218 :
219 60 : HCCL_INFO("[UbConnLite::%s] end", __func__);
220 20 : }
221 :
222 21 : void UbConnLite::ProcessOneWqe(UdmaSqeWrite *sqe, UdmaSqOpcode opCode, const StreamLite &stream)
223 : {
224 : (void)stream;
225 63 : HCCL_INFO("[UbConnLite::%s] start, opCode[%s]", __func__, opCode.Describe().c_str());
226 :
227 : // sqOffset是用于计算Ubjetty中下wqe位置的偏移，小于sqDepth
228 21 : u32 sqOffset = pi % sqDepth_;
229 21 : if (sqOffset < sqDepth_ && (sqOffset + 1) >= sqDepth_) {
230 5 : piDetourCount++;
231 : }
232 : // pi维护用于传入DB Send用于Rtsq 敲door bell，要求u16数据结构并且自然增长
233 21 : pi = pi + 1;
234 :
235 : // 写wqe到va
236 21 : u8 *va = reinterpret_cast<u8 *>(sqVa_ + sqOffset * SQE_SIZE_64);
237 21 : if (!dwqeCacheLocked_) {
238 21 : auto ret = memcpy_sp(va, SQE_SIZE_64, sqe, SQE_SIZE_64);
239 21 : if (UNLIKELY(ret != 0)) {
240 0 : THROW<InternalException>(StringFormat("[UbConnLite::%s] memcpy_sp failed, ret = %d", __func__, ret));
241 : }
242 : }
243 :
244 63 : HCCL_INFO("[UbConnLite::%s] end, pi[%u], ci[%u]", __func__, pi, ci);
245 21 : }
246 :
247 18 : void UbConnLite::ProcessOneWqeWithNotify(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt,
248 : const SqeConfigLite &cfg, UdmaSqeWriteWithNotify *sqe,
249 : const RmtRmaBufSliceLite ¬ify, u64 notifyData, u32 opCode,
250 : SlicePosition slicePos, const StreamLite &stream)
251 : {
252 : (void)stream;
253 54 : HCCL_INFO("[UbConnLite::%s] start, locSize[%u], opCode[%u]", __func__, loc.GetSize(), opCode);
254 :
255 : // sqOffset是用于计算Ubjetty中下wqe位置的偏移，小于sqDepth
256 18 : u32 sqOffset = pi % sqDepth_;
257 18 : if (sqOffset < sqDepth_ && (sqOffset + PI_NUM_TWO) >= sqDepth_) {
258 3 : piDetourCount++;
259 : }
260 : // pi维护用于传入DB Send用于Rtsq 敲door bell，要求u16数据结构并且自然增长
261 18 : pi = pi + PI_NUM_TWO;
262 : // 填充sqe
263 18 : sqe->comm.inlineEn = 0;
264 18 : FillCommSqe(&(sqe->comm), rmt, cfg, WRITE_WITH_NOTIFY_OPCODE, slicePos);
265 18 : FillNotifySqe(&(sqe->notify), notify, notifyData);
266 18 : FillLocalSgeSqe(&(sqe->localU.sge), loc);
267 18 : if (sqe->localU.sge.length == 0) {
268 0 : sqe->comm.sgeNum = 0;
269 : }
270 18 : sqe->rsv1 = 0;
271 18 : sqe->rsv2 = 0;
272 :
273 18 : u8 *va = reinterpret_cast<u8 *>((sqVa_) + sqOffset * SQE_SIZE_64);
274 18 : if (!dwqeCacheLocked_) {
275 : // 带notify的wqe是96字节, 需要占用两个wqebb, 实际占用128字节
276 18 : if (sqOffset == sqDepth_ - 1) {
277 2 : MemorySetAndCopy(va, SQE_SIZE_64, sqe);
278 1 : va = reinterpret_cast<u8 *>(sqVa_);
279 1 : MemorySetAndCopy(va, SQE_SIZE_64, reinterpret_cast<u8 *>(sqe) + SQE_SIZE_64);
280 : } else {
281 16 : MemorySetAndCopy(va, SQE_SIZE_128, sqe);
282 : }
283 : }
284 :
285 48 : HCCL_INFO("[UbConnLite::%s] end, pi[%u], ci[%u]", __func__, pi, ci);
286 16 : }
287 :
288 19 : void UbConnLite::MemorySetAndCopy(u8 *va, u32 sqeSize, void *sqe)
289 : {
290 19 : auto ret = memset_s(va, sqeSize, 0, sqeSize);
291 19 : if (UNLIKELY(ret != 0)) {
292 4 : THROW<InternalException>(StringFormat("[UbConnLite::%s] memset fail, ret = %d", __func__, ret));
293 : }
294 17 : ret = memcpy_sp(va, sqeSize, sqe, sqeSize);
295 17 : if (UNLIKELY(ret != 0)) {
296 0 : THROW<InternalException>(StringFormat("[UbConnLite::%s] not support this op type yet.", __func__));
297 : }
298 17 : }
299 :
300 2 : void UbConnLite::Read(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
301 : const StreamLite &stream, ConnLiteOperationOut &out)
302 : {
303 6 : HCCL_INFO("[UbConnLite::%s] start", __func__);
304 :
305 2 : ProcessSlices(loc, rmt, maxReadSize,
306 2 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
307 3 : UdmaSqeWrite sqe{};
308 3 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_READ, slicePos);
309 3 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_READ, stream);
310 3 : });
311 :
312 2 : out.pi = pi;
313 6 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
314 2 : }
315 :
316 2 : void UbConnLite::ReadReduce(ReduceIn reduceIn, const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt,
317 : const StreamLite &stream, const SqeConfigLite &cfg, ConnLiteOperationOut &out)
318 : {
319 6 : HCCL_INFO("[UbConnLite::%s] start", __func__);
320 :
321 2 : ProcessSlices(loc, rmt, maxReadSize,
322 2 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
323 3 : UdmaSqeWrite sqe{};
324 3 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_READ, slicePos);
325 3 : FillCommSqeReduceInfo(sqe.comm, reduceIn.reduceOp, reduceIn.dataType);
326 3 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_READ, stream);
327 3 : },
328 : reduceIn.dataType);
329 :
330 2 : out.pi = pi;
331 6 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
332 2 : }
333 :
334 4 : void UbConnLite::Write(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
335 : const StreamLite &stream, ConnLiteOperationOut &out)
336 : {
337 12 : HCCL_INFO("[UbConnLite::%s] start, loc size = %llu", __func__, loc.GetSize());
338 :
339 4 : ProcessSlices(loc, rmt, maxWriteSize,
340 4 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
341 5 : UdmaSqeWrite sqe{};
342 5 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_WRITE, slicePos);
343 5 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
344 5 : });
345 :
346 4 : out.pi = pi;
347 12 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
348 4 : }
349 :
350 1 : void UbConnLite::InlineWrite(const u8 *data, u16 size, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
351 : const StreamLite &stream, ConnLiteOperationOut &out)
352 : {
353 3 : HCCL_INFO("[UbConnLite::%s] start", __func__);
354 :
355 : // 构造sqe
356 1 : UdmaSqeWrite sqe{};
357 1 : sqe.comm.inlineEn = 1;
358 1 : sqe.comm.inlineMsgLen = size;
359 1 : FillCommSqe(&(sqe.comm), rmt, cfg, UdmaSqOpcode::UDMA_OPC_WRITE);
360 1 : auto ret = memcpy_sp(sqe.u.inlineData.data, SQE_INLINE_DATA_SIZE, data, size);
361 1 : if (UNLIKELY(ret != 0)) {
362 0 : THROW<InternalException>(StringFormat("[UbConnLite::%s] not support this op type yet.", __func__));
363 : }
364 :
365 : // 写wqe到va
366 1 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
367 :
368 1 : out.pi = pi;
369 3 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, ConnLiteOperationOut.datasize = %u, conn[%s]",
370 : __func__, out.pi, out.dataSize, Describe().c_str());
371 1 : }
372 :
373 18 : void UbConnLite::FillNotifySqe(struct UdmaSqeNotify *sqe, const RmtRmaBufSliceLite ¬ify, u64 notifyData) const
374 : {
375 18 : sqe->notifyTokenId = notify.GetTokenId();
376 18 : sqe->notifyTokenValue = notify.GetTokenValue();
377 18 : sqe->notifyAddrLow = notify.GetAddr() & ADDR_BIT_LOW;
378 18 : sqe->notifyAddrHigh = notify.GetAddr() >> ADDR_BIT_OFFSET;
379 18 : sqe->notifyDataLow = notifyData & ADDR_BIT_LOW;
380 18 : sqe->notifyDataHigh = notifyData >> ADDR_BIT_OFFSET;
381 54 : HCCL_INFO("UbConnLite FillNotifySqe sqe->notifyAddrLow = %u "
382 : "sqe->notifyAddrHigh = %u, sqe->notifyDataLow = %u, sqe->notifyDataHigh = %u",
383 : sqe->notifyAddrLow, sqe->notifyAddrHigh, sqe->notifyDataLow, sqe->notifyDataHigh);
384 18 : }
385 :
386 42 : void UbConnLite::FillLocalSgeSqe(UdmaNormalSge *sqe, const RmaBufSliceLite &loc) const
387 : {
388 42 : sqe->length = loc.GetSize();
389 42 : sqe->tokenId = loc.GetTokenId();
390 42 : sqe->dataAddrLow = loc.GetAddr() & ADDR_BIT_LOW;
391 42 : sqe->dataAddrHigh = loc.GetAddr() >> ADDR_BIT_OFFSET;
392 126 : HCCL_INFO("UbConnLite FillLocalSgeSqe sqe->length = %u, sqe->dataAddrLow = %u "
393 : "sqe->dataAddrHigh = %u",
394 : sqe->length, sqe->dataAddrLow, sqe->dataAddrHigh);
395 42 : }
396 :
397 2 : void UbConnLite::WriteReduce(DataType dataType, ReduceOp reduceOp, const RmaBufSliceLite &loc,
398 : const StreamLite &stream, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
399 : ConnLiteOperationOut &out)
400 : {
401 6 : HCCL_INFO("[UbConnLite::%s] start, dataType = %u, reduceOp %u, loc.addr = %llu, "
402 : "rmt.addr = %llu, cfg.cqeEn = %u, out.pi = %u",
403 : __func__, dataType, reduceOp, loc.GetAddr(), rmt.GetAddr(), cfg.cqeEn, out.pi);
404 :
405 2 : ProcessSlices(loc, rmt, maxWriteSize,
406 2 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
407 3 : UdmaSqeWrite sqe{};
408 3 : FillCommSqeReduceInfo(sqe.comm, reduceOp, dataType);
409 3 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_WRITE, slicePos);
410 3 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
411 3 : },
412 : dataType);
413 :
414 2 : out.pi = pi;
415 6 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
416 2 : }
417 :
418 6 : void UbConnLite::WriteWithNotify(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
419 : ConnLiteOperationOut &out, const RmtRmaBufSliceLite ¬ify, const StreamLite &stream,
420 : u64 notifyData)
421 : {
422 18 : HCCL_INFO("[UbConnLite::%s] start", __func__);
423 :
424 10 : ProcessSlicesWithNotify(
425 : loc, rmt, maxWriteSize,
426 12 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
427 1 : UdmaSqeWrite sqe{};
428 1 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_WRITE, slicePos);
429 1 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
430 1 : },
431 8 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
432 6 : UdmaSqeWriteWithNotify sqe{};
433 6 : ProcessOneWqeWithNotify(locSlice, rmtSlice, cfg, &sqe, notify, notifyData,
434 : WRITE_WITH_NOTIFY_OPCODE, slicePos, stream);
435 4 : });
436 :
437 4 : out.pi = pi;
438 12 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
439 4 : }
440 :
441 12 : void UbConnLite::WriteReduceWithNotify(DataType dataType, ReduceOp reduceOp, const RmaBufSliceLite &loc,
442 : const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg, const StreamLite &stream,
443 : ConnLiteOperationOut &out, const RmtRmaBufSliceLite ¬ify, u64 notifyData)
444 : {
445 36 : HCCL_INFO("[UbConnLite::%s] start", __func__);
446 :
447 12 : ProcessSlicesWithNotify(
448 : loc, rmt, maxWriteSize,
449 24 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
450 5 : UdmaSqeWrite sqe{};
451 5 : FillCommSqeReduceInfo(sqe.comm, reduceOp, dataType);
452 5 : FillOneSqeWrite(locSlice, rmtSlice, cfg, &sqe, UdmaSqOpcode::UDMA_OPC_WRITE, slicePos);
453 5 : ProcessOneWqe(&sqe, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
454 5 : },
455 12 : [&](const RmaBufSliceLite &locSlice, const RmtRmaBufSliceLite &rmtSlice, SlicePosition slicePos) {
456 12 : UdmaSqeWriteWithNotify sqe{};
457 12 : FillCommSqeReduceInfo(sqe.comm, reduceOp, dataType);
458 12 : ProcessOneWqeWithNotify(locSlice, rmtSlice, cfg, &sqe, notify, notifyData,
459 : WRITE_WITH_NOTIFY_OPCODE, slicePos, stream);
460 12 : },
461 : dataType);
462 :
463 12 : out.pi = pi;
464 36 : HCCL_INFO("[UbConnLite::%s] end, ConnLiteOperationOut.pi = %u, conn[%s]", __func__, out.pi, Describe().c_str());
465 12 : }
466 :
467 4 : void UbConnLite::CustomizeSqeByOneSidedComm(UdmaSqeCommon *sqe, bool isLastWqe) const
468 : {
469 : /* 表示SQE是否需要上报CQE:为1表示此SQE需要上报CQE，为0表示不需要 */
470 4 : sqe->cqe = isLastWqe;
471 :
472 : /* 2’b00:No order，表示当前报文与其他报文无保序要求
473 : 2’b01:Relax Order，表示当前报文与后续的Strong Order报文有保序要求，strong order报文不能超越relax order报文执行。
474 : 2’b10：Strong Order，表示当前报文有保序要求，该报文与前面的Relax Order报文有保序要求。
475 : 2’b11：Reserved。
476 : */
477 4 : sqe->placeOdr = (isLastWqe == true ? 0x02 : 0x01);
478 :
479 : /* ODR[2]表示请求报文在目的端的completion order属性，表示当前报文和前面报文是否存在completion序：
480 : 1’b0 :no order，表示当前报文和前面报文没有completion序要求，报文对应的CQE可以乱序上报。
481 : 1’b1 :表示当前报文和前面报文有completion序要求，报文对应的CQE需要保序上报
482 : */
483 4 : sqe->compOrder = 1;
484 :
485 : /* 表示是否使能fence保序。为1时表示使能，为0时表示不使能。对于send/write/atomic SQE
486 : 当fence为1时需要等待前面所有read和Atomic完成才开始执行，即等待前面发出的read或Atomic接收到所有response
487 : */
488 4 : sqe->fence = (isLastWqe == true ? 0x01 : 0x00);
489 :
490 12 : HCCL_INFO(
491 : "UbConnLite CustomizeSqeByOneSidedComm sqe->cqe =%u, sqe->placeOdr = %u sqe->compOrder =%u, sqe->fence = %u",
492 : sqe->cqe, sqe->placeOdr, sqe->compOrder, sqe->fence);
493 4 : }
494 :
495 4 : void UbConnLite::FillBatchOneWqe(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
496 : bool isLastWqe, u32 opCode, const StreamLite &stream)
497 : {
498 : (void)stream;
499 12 : HCCL_INFO("UbConnLite FillBatchOneWqe start, loc[%s], rmt[%s]", loc.Describe().c_str(), rmt.Describe().c_str());
500 :
501 4 : u32 sqOffset = pi % sqDepth_;
502 4 : pi = pi + 1;
503 4 : if (UNLIKELY(pi > sqDepth_)) {
504 0 : pi = pi % sqDepth_;
505 : }
506 :
507 : // 写入wqe数据到out.data
508 4 : UdmaSqeWrite sqe{};
509 4 : sqe.comm.inlineEn = 0;
510 4 : FillCommSqe(&(sqe.comm), rmt, cfg, opCode);
511 4 : FillLocalSgeSqe(&(sqe.u.sge), loc);
512 :
513 4 : if (UNLIKELY(sqe.u.sge.length == 0)) {
514 0 : sqe.comm.sgeNum = 0;
515 : }
516 :
517 4 : CustomizeSqeByOneSidedComm(&(sqe.comm), isLastWqe);
518 :
519 12 : HCCL_INFO("UbConnLite BatchWrite cp data to va %llu, pi %u", sqVa_, pi);
520 4 : u8 *va = reinterpret_cast<u8 *>(sqVa_ + sqOffset * SQE_SIZE_64);
521 4 : if (dwqeCacheLocked_ == false) {
522 4 : auto ret = memcpy_sp(va, SQE_SIZE_64, &sqe, sizeof(UdmaSqeWrite));
523 4 : if (UNLIKELY(ret != 0)) {
524 0 : HCCL_ERROR("UbConnLite::BatchWrite FillCommSqe memcpy failed, ret=%d", ret);
525 0 : THROW<InternalException>(StringFormat("UbConnLite::BatchWrite memcpy_sp failed, ret = %d", ret));
526 : }
527 : }
528 12 : HCCL_INFO("UbConnLite BatchWrite cp data to va end va(%p)", va);
529 4 : }
530 :
531 2 : void UbConnLite::BatchProcessOneSlice(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg,
532 : u32 maxSliceSize, bool isLastSlice, u32 opCode, const StreamLite &stream)
533 : {
534 2 : u64 dataSize = loc.GetSize();
535 : // 按照UDMA能力切分数据
536 : bool isLastWqe;
537 2 : u64 offset = 0;
538 :
539 : // 使用整数除法和取余运算优化循环
540 2 : u64 numIterations = dataSize / maxSliceSize;
541 2 : u64 remainingSize = dataSize % maxSliceSize;
542 :
543 4 : for (u64 i = 0; i < numIterations; ++i) {
544 2 : isLastWqe = false;
545 2 : if ((remainingSize == 0) && (i == numIterations - 1) && isLastSlice) {
546 0 : isLastWqe = true;
547 : }
548 :
549 : // 构造本次wqe的log和rmt RmaBufSilce
550 2 : RmaBufSliceLite locTmp(loc.GetAddr() + offset, UB_DMA_MAX_READ_WEITE_SIZE, loc.GetLkey(), loc.GetTokenId());
551 2 : RmtRmaBufSliceLite rmtTmp(rmt.GetAddr() + offset, UB_DMA_MAX_READ_WEITE_SIZE, rmt.GetRkey(), rmt.GetTokenId(),
552 4 : rmt.GetTokenValue(), UINT32_MAX);
553 :
554 2 : FillBatchOneWqe(locTmp, rmtTmp, cfg, isLastWqe, opCode, stream);
555 :
556 2 : offset += UB_DMA_MAX_READ_WEITE_SIZE;
557 : }
558 :
559 : // 处理剩余的数据
560 2 : if (remainingSize > 0 && isLastSlice) {
561 2 : isLastWqe = true;
562 :
563 2 : RmaBufSliceLite locTmp(loc.GetAddr() + offset, remainingSize, loc.GetLkey(), loc.GetTokenId());
564 2 : RmtRmaBufSliceLite rmtTmp(rmt.GetAddr() + offset, remainingSize, rmt.GetRkey(), rmt.GetTokenId(),
565 4 : rmt.GetTokenValue(), UINT32_MAX);
566 2 : FillBatchOneWqe(locTmp, rmtTmp, cfg, isLastWqe, opCode, stream);
567 : }
568 2 : }
569 :
570 2 : void UbConnLite::BatchCommDataProcess(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
571 : const SqeConfigLite &cfg, u32 maxSliceSize, u32 opCode, const StreamLite &stream)
572 : {
573 2 : u64 siliceSize = loc.size();
574 : // 按照UDMA能力切分数据, 组装wqe
575 4 : for (u64 i = 0; i < siliceSize; i++) {
576 2 : BatchProcessOneSlice(loc[i], rmt[i], cfg, maxSliceSize, (i == (siliceSize - 1)), opCode, stream);
577 : }
578 :
579 2 : return;
580 : }
581 :
582 1 : void UbConnLite::BatchOneSidedRead(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
583 : const SqeConfigLite &cfg, const StreamLite &stream, ConnLiteOperationOut &out)
584 : {
585 : // 按照UDMA能力切分数据, 组装wqe
586 1 : BatchCommDataProcess(loc, rmt, cfg, maxReadSize, UdmaSqOpcode::UDMA_OPC_READ, stream);
587 :
588 : // 更新connlite的输出信息
589 1 : out.pi = pi;
590 3 : HCCL_INFO("UbConnLite BatchRead end, out.pi = %u", out.pi);
591 1 : }
592 :
593 1 : void UbConnLite::BatchOneSidedWrite(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
594 : const SqeConfigLite &cfg, const StreamLite &stream, ConnLiteOperationOut &out)
595 : {
596 : // 按照UDMA能力切分数据, 组装wqe
597 1 : BatchCommDataProcess(loc, rmt, cfg, maxWriteSize, UdmaSqOpcode::UDMA_OPC_WRITE, stream);
598 :
599 : // 更新connlite的输出信息
600 1 : out.pi = pi;
601 3 : HCCL_INFO("UbConnLite BatchWrite end, out.pi = %u", out.pi);
602 1 : }
603 :
604 32 : std::string UbConnLite::Describe()
605 : {
606 : return StringFormat("UbConnLite[dieId=%u, funcId=%u, jettyId=%u, dbAddr=0x%llx, sqVa=0x%llx, sqDepth=%u, "
607 : "jfcPollMode=%u, tpn=%u, dwqeCacheLocked=%d, locEid=%s, rmtEid=%s,jettyPi=%u, jettyCi=%u]",
608 32 : dieId_, funcId_, jettyId_, dbAddr_, sqVa_, sqDepth_, jfcPollMode_, tpn_, dwqeCacheLocked_,
609 64 : Bytes2hex(locEid_.raw, sizeof(locEid_.raw)).c_str(), Bytes2hex(rmtEid_.raw, sizeof(rmtEid_.raw)).c_str(),
610 96 : pi, ci);
611 : }
612 :
613 : constexpr uint32_t UB_WQE_NUM_PER_SQE = 4; // URMA约束每个SQE包含4个WQEBB
614 2 : UbConnLite::UbConnLite(const UbConnLiteParam &liteParam)
615 : {
616 4 : HCCL_INFO("[UbConnLite::%s] liteParam[%s]", __func__, liteParam.Describe().c_str());
617 2 : dieId_ = liteParam.dieId;
618 2 : funcId_ = liteParam.funcId;
619 2 : jettyId_ = liteParam.jettyId;
620 2 : dbAddr_ = liteParam.dbAddr;
621 2 : sqVa_ = liteParam.sqVa;
622 : // host侧创建jetty指定的sqDepth为sqeBBNum,device侧需要感知wqebbnum,URMA约束每个SQE包含4个WQEBB
623 2 : sqDepth_ = liteParam.sqDepth * UB_WQE_NUM_PER_SQE;
624 2 : dwqeCacheLocked_ = liteParam.dwqeCacheLocked;
625 2 : jfcPollMode_ = liteParam.jfcPollMode;
626 2 : tpn_ = liteParam.tpn;
627 :
628 2 : maxReadSize = liteParam.maxReadSize;
629 2 : maxWriteSize = liteParam.maxWriteSize;
630 :
631 2 : (void)memcpy_sp(rmtEid_.raw, URMA_EID_LEN, liteParam.rmtEid.raw, URMA_EID_LEN);
632 2 : (void)memcpy_sp(locEid_.raw, URMA_EID_LEN, liteParam.locEid.raw, URMA_EID_LEN);
633 4 : HCCL_INFO("%s", Describe().c_str());
634 2 : }
635 :
636 17 : UbConnLite::UbConnLite(const UbJettyLiteId &id, const UbJettyLiteAttr &attr, const Eid &rmtInfo)
637 : : RmaConnLite(id, attr, rmtInfo),
638 17 : maxReadSize(UB_DMA_MAX_READ_WEITE_SIZE),
639 17 : maxWriteSize(UB_DMA_MAX_READ_WEITE_SIZE)
640 : {
641 17 : }
642 :
643 2 : std::string UbConnLiteParam::Describe() const
644 : {
645 : return StringFormat("UbConnLiteParam[dieId=%u, funcId=%u, jettyId=%u, dbAddr=0x%llx, sqVa=0x%llx, sqDepth=%u, "
646 : "jfcPollMode=%u, tpn=%u, dwqeCacheLocked=%d, sqCiAddr=0x%llx, rmtEid=%s, localEid=%s, "
647 : "maxReadSize=%u, maxWriteSize=%u]",
648 2 : dieId, funcId, jettyId, dbAddr, sqVa, sqDepth, jfcPollMode, tpn, dwqeCacheLocked, sqCiAddr,
649 4 : Bytes2hex(rmtEid.raw, sizeof(rmtEid.raw)).c_str(), Bytes2hex(locEid.raw, sizeof(locEid.raw)).c_str(),
650 6 : maxReadSize, maxWriteSize);
651 : }
652 :
653 7 : UbConnLiteParam::UbConnLiteParam(std::vector<char> &uniqueId)
654 : {
655 7 : BinaryStream binaryStream(uniqueId);
656 7 : binaryStream >> dieId;
657 7 : binaryStream >> funcId;
658 7 : binaryStream >> jettyId;
659 :
660 7 : binaryStream >> jfcPollMode;
661 7 : binaryStream >> dwqeCacheLocked;
662 7 : binaryStream >> dbAddr;
663 7 : binaryStream >> sqCiAddr;
664 7 : binaryStream >> sqVa;
665 7 : binaryStream >> sqDepth;
666 7 : binaryStream >> tpn;
667 7 : binaryStream >> rmtEid.raw;
668 7 : binaryStream >> locEid.raw;
669 7 : binaryStream >> maxReadSize;
670 7 : binaryStream >> maxWriteSize;
671 :
672 7 : static auto lastPrintTime = std::chrono::steady_clock::now();
673 7 : const auto now = std::chrono::steady_clock::now();
674 7 : const auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(now - lastPrintTime).count();
675 7 : if (UNLIKELY(duration >= MAX_LOG_TIMEOUT_MS)) {
676 0 : HCCL_INFO("%s", Describe().c_str());
677 0 : lastPrintTime = now;
678 : }
679 17 : HCCL_INFO("[UbConnLiteParam::%s] locEid[%s], rmtEid[%s]", __func__, locEid.Describe().c_str(), rmtEid.Describe().c_str());
680 7 : }
681 :
682 : } // namespace Hccl
|
