1 /*
2 * Copyright 2025 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.channel.uring;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.channel.unix.Buffer;
20
21 import java.lang.invoke.MethodHandles;
22 import java.lang.invoke.VarHandle;
23 import java.nio.ByteBuffer;
24 import java.nio.ByteOrder;
25 import java.util.Arrays;
26 import java.util.function.Consumer;
27
28 final class IoUringBufferRing {
29 private static final VarHandle SHORT_HANDLE =
30 MethodHandles.byteBufferViewVarHandle(short[].class, ByteOrder.nativeOrder());
31 private final ByteBuffer ioUringBufRing;
32 private final int tailFieldPosition;
33 private final short entries;
34 private final short mask;
35 private final short bufferGroupId;
36 private final int ringFd;
37 private final ByteBuf[] buffers;
38 private final IoUringBufferRingAllocator allocator;
39 private final boolean batchAllocation;
40 private final IoUringBufferRingExhaustedEvent exhaustedEvent;
41 private final RingConsumer ringConsumer;
42 private final boolean incremental;
43 private final int batchSize;
44 private boolean corrupted;
45 private boolean closed;
46 private int usableBuffers;
47 private int allocatedBuffers;
48 private boolean needExpand;
49 private short lastGeneratedBid;
50
51 IoUringBufferRing(int ringFd, ByteBuffer ioUringBufRing,
52 short entries, int batchSize, short bufferGroupId, boolean incremental,
53 IoUringBufferRingAllocator allocator, boolean batchAllocation) {
54 assert entries % 2 == 0;
55 assert batchSize % 2 == 0;
56 this.batchSize = batchSize;
57 this.ioUringBufRing = ioUringBufRing;
58 this.tailFieldPosition = Native.IO_URING_BUFFER_RING_TAIL;
59 this.entries = entries;
60 this.mask = (short) (entries - 1);
61 this.bufferGroupId = bufferGroupId;
62 this.ringFd = ringFd;
63 this.buffers = new ByteBuf[entries];
64 this.incremental = incremental;
65 this.allocator = allocator;
66 this.batchAllocation = batchAllocation;
67 this.ringConsumer = new RingConsumer();
68 this.exhaustedEvent = new IoUringBufferRingExhaustedEvent(bufferGroupId);
69 }
70
71 boolean isUsable() {
72 return !closed && !corrupted;
73 }
74
75 void initialize() {
76 // We already validated that batchSize is <= ring length.
77 fill((short) 0, batchSize);
78 allocatedBuffers = batchSize;
79 }
80
81 private final class RingConsumer implements Consumer<ByteBuf> {
82 private int expectedBuffers;
83 private short num;
84 private short bid;
85 private short oldTail;
86
87 short fill(short startBid, int numBuffers) {
88 // Fetch the tail once before allocate the batch.
89 oldTail = (short) SHORT_HANDLE.get(ioUringBufRing, tailFieldPosition);
90
91 // At the moment we always start with bid 0 and so num and bid is the same. As this is more of an
92 // implementation detail it is better to still keep both separated.
93 this.num = 0;
94 this.bid = startBid;
95 this.expectedBuffers = numBuffers;
96 try {
97 if (batchAllocation) {
98 allocator.allocateBatch(this, numBuffers);
99 } else {
100 for (int i = 0; i < numBuffers; i++) {
101 add(oldTail, bid++, num++, allocator.allocate());
102 }
103 }
104 } catch (Throwable t) {
105 corrupted = true;
106 for (int i = 0; i < buffers.length; i++) {
107 ByteBuf buffer = buffers[i];
108 if (buffer != null) {
109 buffer.release();
110 buffers[i] = null;
111 }
112 }
113 throw t;
114 }
115 // Now advanced the tail by the number of buffers that we just added.
116 SHORT_HANDLE.setRelease(ioUringBufRing, tailFieldPosition, (short) (oldTail + num));
117
118 return (short) (bid - 1);
119 }
120
121 void fill(short bid) {
122 short tail = (short) SHORT_HANDLE.get(ioUringBufRing, tailFieldPosition);
123 add(tail, bid, 0, allocator.allocate());
124 // Now advanced the tail by one
125 SHORT_HANDLE.setRelease(ioUringBufRing, tailFieldPosition, (short) (tail + 1));
126 }
127
128 @Override
129 public void accept(ByteBuf byteBuf) {
130 if (corrupted || closed) {
131 byteBuf.release();
132 throw new IllegalStateException("Already closed");
133 }
134 if (expectedBuffers == num) {
135 byteBuf.release();
136 throw new IllegalStateException("Produced too many buffers");
137 }
138 add(oldTail, bid++, num++, byteBuf);
139 }
140
141 private void add(int tail, short bid, int offset, ByteBuf byteBuf) {
142 short ringIndex = (short) ((tail + offset) & mask);
143 assert buffers[bid] == null;
144
145 long memoryAddress = IoUring.memoryAddress(byteBuf) + byteBuf.writerIndex();
146 int writable = byteBuf.writableBytes();
147
148 // see:
149 // https://github.com/axboe/liburing/
150 // blob/19134a8fffd406b22595a5813a3e319c19630ac9/src/include/liburing.h#L1561
151 int position = Native.SIZEOF_IOURING_BUF * ringIndex;
152 ioUringBufRing.putLong(position + Native.IOURING_BUFFER_OFFSETOF_ADDR, memoryAddress);
153 ioUringBufRing.putInt(position + Native.IOURING_BUFFER_OFFSETOF_LEN, writable);
154 ioUringBufRing.putShort(position + Native.IOURING_BUFFER_OFFSETOF_BID, bid);
155
156 buffers[bid] = byteBuf;
157 }
158 }
159
160 /**
161 * Try to expand by adding more buffers to the ring if there is any space left, this will be done lazy.
162 *
163 * @return {@code true} if we can expand the number of buffers in the ring, {@code false} otherwise.
164 */
165 boolean expand() {
166 needExpand = true;
167 return allocatedBuffers < buffers.length;
168 }
169
170 private void fill(short startBid, int buffers) {
171 if (corrupted || closed) {
172 return;
173 }
174 assert buffers % 2 == 0;
175 lastGeneratedBid = ringConsumer.fill(startBid, buffers);
176 usableBuffers += buffers;
177 }
178
179 private void fill(short bid) {
180 if (corrupted || closed) {
181 return;
182 }
183 ringConsumer.fill(bid);
184 usableBuffers++;
185 }
186
187 /**
188 * @return the {@link IoUringBufferRingExhaustedEvent} that should be used to signal that there were no buffers
189 * left for this buffer ring.
190 */
191 IoUringBufferRingExhaustedEvent getExhaustedEvent() {
192 return exhaustedEvent;
193 }
194
195 /**
196 * Return the amount of bytes that we attempted to read for the given id.
197 * This method must be called before {@link #useBuffer(short, int, boolean)}.
198 *
199 * @param bid the id of the buffer.
200 * @return the attempted bytes.
201 */
202 int attemptedBytesRead(short bid) {
203 return buffers[bid].writableBytes();
204 }
205
206 private int calculateNextBufferBatch() {
207 return Math.min(batchSize, entries - allocatedBuffers);
208 }
209
210 /**
211 * Use the buffer for the given buffer id. The returned {@link ByteBuf} must be released once not used anymore.
212 *
213 * @param bid the id of the buffer
214 * @param read the number of bytes that could be read. This value might be larger then what a single
215 * {@link ByteBuf} can hold. Because of this, the caller should call
216 * @link #useBuffer(short, int, boolean)} in a loop (obtaining the next bid to use by calling
217 * {@link #nextBid(short, int)} with an {@link #allocatedBuffers()} snapshot taken before
218 * this call) until all buffers could be obtained.
219 * @return the buffer.
220 */
221 ByteBuf useBuffer(short bid, int read, boolean more) {
222 assert read > 0;
223 ByteBuf byteBuf = buffers[bid];
224
225 allocator.lastBytesRead(byteBuf.writableBytes(), read);
226 // The kernel might report a bundle total (RECVSEND_BUNDLE) that spans more than this single ring
227 // entry can hold. Clamp to what this entry can actually provide; the caller loops over the
228 // remainder using the next bid (see nextBid(short, int)).
229 int len = Math.min(read, byteBuf.writableBytes());
230 // We always slice so the user will not mess up things later.
231 ByteBuf buffer = byteBuf.retainedSlice(byteBuf.writerIndex(), len);
232 byteBuf.writerIndex(byteBuf.writerIndex() + len);
233
234 if (incremental && more && byteBuf.isWritable()) {
235 // The buffer will be used later again, just slice out what we did read so far.
236 return buffer;
237 }
238
239 // The buffer is considered to be used, null out the slot.
240 buffers[bid] = null;
241 byteBuf.release();
242 if (--usableBuffers == 0) {
243 int numBuffers = allocatedBuffers;
244 if (needExpand) {
245 // We did get a signal that our buffer ring did not have enough buffers, let's see if we
246 // can grow it.
247 needExpand = false;
248 numBuffers += calculateNextBufferBatch();
249 }
250 fill((short) 0, numBuffers);
251 allocatedBuffers = numBuffers;
252 assert allocatedBuffers % 2 == 0;
253 } else if (!batchAllocation) {
254 // If we don'T do bulk allocations to refill the buffer ring we need to fill in the just used bid again
255 // if we didn't get a signal that we need expansion.
256 fill(bid);
257
258 if (needExpand && lastGeneratedBid == bid) {
259 // We did get a signal that our buffer ring did not have enough buffers and we just did add the last
260 // generated bid at the tail of the ring. Now its safe to grow the buffer ring and still guarantee
261 // sequential ordering which is needed for our RECVSEND_BUNDLE implementation.
262 needExpand = false;
263 int numBuffers = calculateNextBufferBatch();
264 fill((short) (bid + 1), numBuffers);
265 allocatedBuffers += numBuffers;
266 assert allocatedBuffers % 2 == 0;
267 }
268 }
269 return buffer;
270 }
271
272 /**
273 * The number of buffers that are currently allocated (and so posted) in this buffer ring. This value can
274 * change every time {@link #useBuffer(short, int, boolean)} is called as it might grow the ring. Because of
275 * this, callers that need to walk a sequence of consecutive buffer ids that belong to the same completion
276 * (for example a {@code IORING_RECVSEND_BUNDLE} completion) must capture this value <strong>before</strong>
277 * calling {@link #useBuffer(short, int, boolean)} and pass it to {@link #nextBid(short, int)} afterwards,
278 * so the wrap-around is computed using the size of the ring as the kernel observed it when it produced the
279 * bundle, not the (possibly already grown) size afterwards.
280 *
281 * @return the number of allocated buffers.
282 */
283 int allocatedBuffers() {
284 return allocatedBuffers;
285 }
286
287 /**
288 * Return the next buffer id that follows {@code bid} in ring order, wrapping around after
289 * {@code allocatedBuffers} ids.
290 *
291 * @param bid the current buffer id.
292 * @param allocatedBuffers the number of allocated buffers to use for the wrap-around computation. This should
293 * be a value obtained from {@link #allocatedBuffers()} <strong>before</strong> the
294 * corresponding {@link #useBuffer(short, int, boolean)} call for {@code bid}, as that
295 * call might have grown the ring already.
296 * @return the next buffer id.
297 */
298 short nextBid(short bid, int allocatedBuffers) {
299 // Use the real modulo (and not a bitmask) as allocatedBuffers is not guaranteed to be a power of two.
300 return (short) ((bid + 1) % allocatedBuffers);
301 }
302
303 /**
304 * The group id that is assigned to this buffer ring.
305 *
306 * @return group id.
307 */
308 short bufferGroupId() {
309 return bufferGroupId;
310 }
311
312 /**
313 * Close this {@link IoUringBufferRing}, using it after this method is called will lead to undefined behaviour.
314 */
315 void close() {
316 if (closed) {
317 return;
318 }
319 closed = true;
320 Native.ioUringUnRegisterBufRing(ringFd, Buffer.memoryAddress(ioUringBufRing), entries, bufferGroupId);
321 for (ByteBuf byteBuf : buffers) {
322 if (byteBuf != null) {
323 byteBuf.release();
324 }
325 }
326 Arrays.fill(buffers, null);
327 }
328 }