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