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