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