Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 :
12 : #ifndef RDMA_BASE_VENDOR_OPS_H
13 : #define RDMA_BASE_VENDOR_OPS_H
14 :
15 : #include <chrono>
16 : #include <cstddef>
17 : #include <cstdint>
18 : #include <vector>
19 : #include <securec.h>
20 :
21 : #include "exception_util.h"
22 : #include "internal_exception.h"
23 : #include "log.h"
24 : #include "rma_buf_slice_lite.h" // RmaBufSliceLite
25 : #include "rmt_rma_buf_slice_lite.h" // RmtRmaBufSliceLite
26 : #include "data_type.h" // DataType
27 : #include "reduce_op.h" // ReduceOp
28 :
29 : namespace Hccl {
30 :
31 : struct SqeConfigLite;
32 :
33 : struct RdmaSqContextLite {
34 : uint32_t qpn;
35 : uint64_t sqVa;
36 : uint32_t wqeSize;
37 : uint32_t depth;
38 : uint64_t headAddr;
39 : uint64_t tailAddr;
40 : uint64_t dbHwVa;
41 : uint64_t dbSwVa;
42 : uint8_t sl;
43 : uint8_t mtuShift;
44 : };
45 :
46 : struct RdmaCqContextLite {
47 : uint32_t cqn;
48 : uint64_t cqVa;
49 : uint32_t cqeSize;
50 : uint32_t cqDepth;
51 : uint64_t headAddr;
52 : uint64_t tailAddr;
53 : uint64_t dbHwVa;
54 : uint64_t dbSwVa;
55 : };
56 :
57 : // Necessary helper funcs
58 : constexpr uint32_t BITS_1BYTE = 8;
59 : constexpr uint32_t BITS_3BYTE = 24;
60 : constexpr uint32_t BITS_5BYTE = 40;
61 : constexpr uint32_t BITS_7BYTE = 56;
62 :
63 33 : inline uint16_t Htons16(uint16_t x) {
64 33 : return (((x & 0xffULL) << BITS_1BYTE) | ((x & 0xff00ULL) >> BITS_1BYTE));
65 : }
66 :
67 86 : inline uint32_t Htonl32(uint32_t x) {
68 86 : return ((x & 0x000000ffU) << BITS_3BYTE) | ((x & 0x0000ff00U) << BITS_1BYTE) |
69 86 : ((x & 0x00ff0000U) >> BITS_1BYTE) | ((x & 0xff000000U) >> BITS_3BYTE);
70 : }
71 :
72 22 : inline uint64_t Htonll64(uint64_t x) {
73 22 : return ((x & 0x00000000000000ffULL) << BITS_7BYTE) |
74 22 : ((x & 0x000000000000ff00ULL) << BITS_5BYTE) |
75 22 : ((x & 0x0000000000ff0000ULL) << BITS_3BYTE) |
76 22 : ((x & 0x00000000ff000000ULL) << BITS_1BYTE) |
77 22 : ((x & 0x000000ff00000000ULL) >> BITS_1BYTE) |
78 22 : ((x & 0x0000ff0000000000ULL) >> BITS_3BYTE) |
79 22 : ((x & 0x00ff000000000000ULL) >> BITS_5BYTE) |
80 22 : ((x & 0xff00000000000000ULL) >> BITS_7BYTE);
81 : }
82 :
83 : enum class CqPollStatus : int32_t {
84 : SUCCESS = 0, /* indicate poll once cqe successfully; */
85 : EMPTY = -1, /* indicate the cq is empty when poll cq; */
86 : ERROR = -2, /* indicate the error when poll cq; */
87 : REROLL = -3, /* 返回到repoll标识 */
88 : CONTINUE = 1, /* indicate continue the process */
89 : };
90 :
91 : class RdmaBaseOps {
92 : public:
93 39 : RdmaBaseOps(RdmaSqContextLite *sqContext, RdmaCqContextLite *cqContext)
94 39 : : sqContext_(sqContext), cqContext_(cqContext) {}
95 39 : virtual ~RdmaBaseOps() = default;
96 :
97 : // 上层接口，不关心具体vendor类型
98 3 : HcclResult Read(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg)
99 : {
100 : // Read需要占用1个wr位置, 确定Sq存在空位
101 3 : constexpr int ReadWqeCount = 1;
102 3 : CHK_RET(WaitSqFree(ReadWqeCount));
103 :
104 : // 进入vendor特有wqe组装接口, 组装完wqe直接下发, opCode不支持直接返回HCCL_E_NOT_SUPPORT
105 3 : CHK_RET(BuildReadWqe(loc, rmt, cfg));
106 :
107 : // 更新Sq队列PI值
108 3 : CHK_RET(UpdateSqPI());
109 :
110 3 : return HCCL_SUCCESS;
111 : }
112 :
113 6 : HcclResult Write(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg)
114 : {
115 : // Write需要占用1个wr位置, 确定Sq存在空位
116 6 : constexpr int WriteWqeCount = 1;
117 6 : CHK_RET(WaitSqFree(WriteWqeCount));
118 :
119 : // 进入vendor特有wqe组装接口, 组装完wqe直接下发
120 6 : CHK_RET(BuildWriteWqe(loc, rmt, cfg));
121 :
122 : // 更新Sq队列PI值
123 6 : CHK_RET(UpdateSqPI());
124 :
125 6 : return HCCL_SUCCESS;
126 : }
127 :
128 2 : HcclResult WriteReduce(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg, DataType dataType, ReduceOp reduceOp)
129 : {
130 : // Inline Reduce Write需要占用1个wr位置, 确定Sq存在空位
131 2 : constexpr int WriteReduceWqeCount = 1;
132 2 : CHK_RET(WaitSqFree(WriteReduceWqeCount));
133 :
134 : // 进入vendor特有wqe组装接口, 组装完wqe直接下发
135 2 : CHK_RET(BuildWriteReduceWqe(loc, rmt, cfg, dataType, reduceOp));
136 :
137 : // 更新Sq队列PI值
138 2 : CHK_RET(UpdateSqPI());
139 :
140 2 : return HCCL_SUCCESS;
141 : }
142 :
143 3 : HcclResult PollCq(int32_t numEntries, int32_t timeOut, std::vector<int32_t> &errList)
144 : {
145 3 : auto timeLimit = std::chrono::milliseconds(timeOut);
146 3 : auto startTime = std::chrono::steady_clock::now();
147 :
148 3 : int32_t totalPollNum = 0;
149 3 : int32_t ret = 0;
150 :
151 4 : while (totalPollNum < numEntries) {
152 : // 逐一Poll Cq，并处理cqe
153 3 : ret = PollCqImpl(numEntries - totalPollNum, errList);
154 3 : if (ret == static_cast<int32_t>(CqPollStatus::ERROR)) {
155 3 : HCCL_ERROR("[RdmaBaseOps::%s][Poll cq] Poll Cq Error.", __func__);
156 1 : return HCCL_E_REMOTE;
157 : }
158 :
159 2 : if (ret > 0) {
160 : // Update cqe in this loop
161 1 : totalPollNum += ret;
162 :
163 : // Update Sq Tail
164 1 : sqTail_ += ret;
165 :
166 : // continue poll cqe
167 1 : continue;
168 : }
169 :
170 1 : if ((std::chrono::steady_clock::now() - startTime) > timeLimit) {
171 3 : HCCL_ERROR("[RdmaBaseOps::%s][Poll cq] Poll Cq timeout, expected[%d], actual[%d], lastRet[%d]",
172 : __func__, numEntries, totalPollNum, ret);
173 1 : return HCCL_E_TIMEOUT;
174 : }
175 : }
176 :
177 3 : HCCL_INFO("[RdmaBaseOps::%s][Poll cq] Poll Cq success, expected[%d], actual[%d]",
178 : __func__, numEntries, totalPollNum);
179 1 : return HCCL_SUCCESS;
180 : }
181 :
182 : // 准备Doorbell(厂商实现)
183 : virtual HcclResult BuildDoorbell(u64 &dbAddr, u64 &dbValue) = 0;
184 :
185 : // 准备CqDoorbell(厂商实现)
186 : virtual HcclResult BuildCqDoorbell(u64 &dbAddr, u64 &dbValue) = 0;
187 :
188 : protected:
189 : // 软件侧只维护Sq PI，Sq CI由硬件维护
190 : u32 sqHead_{0};
191 : u32 sqTail_{0};
192 :
193 : u32 cqHead_{0};
194 : u32 cqTail_{0};
195 : bool cqDbFlush_ = false;
196 :
197 : RdmaSqContextLite *sqContext_;
198 : RdmaCqContextLite *cqContext_;
199 :
200 : // 默认超时时间 30 ms
201 : const std::chrono::milliseconds timeout_ = std::chrono::milliseconds(30U);
202 :
203 : // vendor扩展点: 每个原子op一个虚函数
204 : // 默认 NOT_SUPPORT, 各个vendor 只重写自己支持的
205 0 : virtual HcclResult BuildReadWqe(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg) {
206 : (void)loc;
207 : (void)rmt;
208 : (void)cfg;
209 0 : HCCL_ERROR("[RdmaBaseOps::%s] This Backend Not support Read Now.", __func__);
210 0 : return HCCL_E_NOT_SUPPORT;
211 : }
212 :
213 0 : virtual HcclResult BuildWriteWqe(const RmaBufSliceLite &loc, const RmtRmaBufSliceLite &rmt, const SqeConfigLite &cfg) {
214 : (void)loc;
215 : (void)rmt;
216 : (void)cfg;
217 0 : HCCL_ERROR("[RdmaBaseOps::%s] This Backend Not support Write Now.", __func__);
218 0 : return HCCL_E_NOT_SUPPORT;
219 : }
220 :
221 0 : virtual HcclResult BuildWriteReduceWqe(const RmaBufSliceLite &locNotify, const RmtRmaBufSliceLite ¬ify, const SqeConfigLite &cfg,
222 : DataType dataType, ReduceOp reduceOp) {
223 : (void)locNotify;
224 : (void)notify;
225 : (void)cfg;
226 : (void)dataType;
227 : (void)reduceOp;
228 0 : HCCL_ERROR("[RdmaBaseOps::%s] This Backend Not support WriteReduce Now.", __func__);
229 0 : return HCCL_E_NOT_SUPPORT;
230 : }
231 :
232 0 : virtual HcclResult WriteInvalidWqebb(uint32_t nextIdx) {
233 : (void)nextIdx;
234 0 : return HCCL_SUCCESS;
235 : }
236 :
237 0 : virtual int32_t PollCqImpl(int32_t numEntries, std::vector<int32_t> &errList) {
238 : (void)numEntries;
239 0 : HCCL_ERROR("[RdmaBaseOps::%s] This Backend Not support PollCq Now.", __func__);
240 0 : return HCCL_E_NOT_SUPPORT;
241 : }
242 :
243 : // 搬运wqe(通用实现), 把 Wqe 写到 SQ
244 0 : HcclResult CommitWqe(const void *wqe, uint32_t wqeSize)
245 : {
246 0 : HCCL_INFO("[RdmaBaseOps::%s] Memcpy wqe start, Now SQ PI: [%u]", __func__, sqHead_);
247 :
248 : // 写wqe到va
249 0 : auto sqDepth = sqContext_->depth;
250 0 : uint32_t sqPIMask = sqDepth - 1;
251 0 : u8 *va = reinterpret_cast<u8 *>(sqContext_->sqVa + (sqHead_ & sqPIMask) * wqeSize);
252 :
253 0 : HCCL_INFO("[RdmaBaseOps][Wqe Write] before copy, sqHead[%u], slot[%u], sqVa[0x%llx], dst[0x%llx], size[%u]",
254 : sqHead_, sqHead_ & sqPIMask,
255 : static_cast<unsigned long long>(sqContext_->sqVa),
256 : reinterpret_cast<unsigned long long>(va),
257 : wqeSize);
258 :
259 0 : auto ret = memcpy_sp(va, wqeSize, wqe, wqeSize);
260 0 : if (UNLIKELY(ret != 0)) {
261 0 : THROW<InternalException>(StringFormat("[RdmaBaseOps::%s] memcpy_s failed, ret = %d", __func__, ret));
262 : }
263 :
264 : // pi维护用于传入DB Send用于Rtsq 敲door bell
265 0 : sqHead_ = sqHead_ + 1;
266 :
267 : // Write InValid Wqebb
268 0 : CHK_RET(WriteInvalidWqebb(sqHead_));
269 :
270 0 : HCCL_INFO("[RdmaBaseOps::%s] Memcpy wqe end, Now SQ PI: [%u]", __func__, sqHead_);
271 0 : return HCCL_SUCCESS;
272 : }
273 :
274 0 : HcclResult WaitSqFree(uint32_t wqeNum) {
275 : // wq_overflow
276 0 : bool timeOutFlag = false;
277 0 : auto startTime = std::chrono::steady_clock::now();
278 :
279 0 : HCCL_INFO("[RdmaBaseOps::%s] Operate: sqTail = %u", __func__, sqTail_);
280 0 : while (!timeOutFlag) {
281 : // sq 队列能放下，直接成功返回
282 0 : if (static_cast<uint32_t>(sqHead_ - sqTail_ + wqeNum) <= sqContext_->depth) {
283 0 : return HCCL_SUCCESS;
284 : }
285 :
286 0 : timeOutFlag = (std::chrono::steady_clock::now() - startTime) > timeout_;
287 : }
288 :
289 : // 超时处理
290 0 : HCCL_ERROR("[RdmaBaseOps::%s] Sq is Full !! Operate: sqTail = %u Failed. ", __func__, sqTail_);
291 0 : return HCCL_E_TIMEOUT;
292 : }
293 :
294 : // 将PI更新到硬件可见地址
295 0 : HcclResult UpdateSqPI() {
296 : // 更新Sq PI指针
297 0 : uint32_t sqHeadNum = Htonl32(sqHead_);
298 :
299 0 : HCCL_INFO("[RdmaBaseOps][Wqe Write] write soft PI, sqHead host[%u], dbSwVa[0x%llx]",
300 : sqHead_, static_cast<unsigned long long>(sqContext_->dbSwVa));
301 :
302 0 : auto status = memcpy_sp(reinterpret_cast<void *>(sqContext_->dbSwVa), sizeof(uint32_t), &sqHeadNum,
303 : sizeof(uint32_t));
304 0 : if (UNLIKELY(status != 0)) {
305 0 : THROW<InternalException>(StringFormat("[RdmaBaseOps::%s] Ring Sw DB failed, ret = %d", __func__, status));
306 : }
307 0 : return HCCL_SUCCESS;
308 : }
309 : };
310 :
311 : }
312 : #endif // RDMA_BASE_VENDOR_OPS_H
|
