1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.channel.uring;
17
18 import io.netty.channel.IoHandle;
19 import io.netty.channel.IoHandler;
20 import io.netty.channel.IoHandlerContext;
21 import io.netty.channel.IoHandlerFactory;
22 import io.netty.channel.IoOps;
23 import io.netty.channel.IoRegistration;
24 import io.netty.channel.unix.Buffer;
25 import io.netty.channel.unix.Errors;
26 import io.netty.channel.unix.FileDescriptor;
27 import io.netty.channel.unix.IovArray;
28 import io.netty.util.collection.IntObjectHashMap;
29 import io.netty.util.collection.IntObjectMap;
30 import io.netty.util.concurrent.ThreadAwareExecutor;
31 import io.netty.util.internal.CleanableDirectBuffer;
32 import io.netty.util.internal.ObjectUtil;
33 import io.netty.util.internal.StringUtil;
34 import io.netty.util.internal.logging.InternalLogger;
35 import io.netty.util.internal.logging.InternalLoggerFactory;
36
37 import java.io.IOException;
38 import java.io.UncheckedIOException;
39 import java.nio.ByteBuffer;
40 import java.util.ArrayList;
41 import java.util.Collection;
42 import java.util.List;
43 import java.util.concurrent.TimeUnit;
44 import java.util.concurrent.atomic.AtomicBoolean;
45 import java.util.concurrent.atomic.AtomicInteger;
46
47 import static java.lang.Math.max;
48 import static java.lang.Math.min;
49 import static java.util.Objects.requireNonNull;
50
51
52
53
54 public final class IoUringIoHandler implements IoHandler {
55 private static final InternalLogger logger = InternalLoggerFactory.getInstance(IoUringIoHandler.class);
56 private static final int WAKEUP_CLOSED = 1 << 30;
57
58 private final RingBuffer ringBuffer;
59 private final IntObjectMap<IoUringBufferRing> registeredIoUringBufferRing;
60 private final IntObjectMap<DefaultIoUringIoRegistration> registrations;
61
62 private final byte[] inet4AddressArray = new byte[SockaddrIn.IPV4_ADDRESS_LENGTH];
63 private final byte[] inet6AddressArray = new byte[SockaddrIn.IPV6_ADDRESS_LENGTH];
64
65 private final AtomicBoolean eventfdAsyncNotify = new AtomicBoolean();
66 private final AtomicInteger wakeupWriters = new AtomicInteger();
67 private final FileDescriptor eventfd;
68 private final CleanableDirectBuffer eventfdReadBufCleanable;
69 private final ByteBuffer eventfdReadBuf;
70 private final long eventfdReadBufAddress;
71 private final CleanableDirectBuffer timeoutMemoryCleanable;
72 private final ByteBuffer timeoutMemory;
73 private final long timeoutMemoryAddress;
74 private final IovArray iovArray;
75 private final IovArrayReferenceCollector iovArrayReferenceCollector;
76 private final MsgHdrMemoryArray msgHdrMemoryArray;
77 private long eventfdReadSubmitted;
78 private boolean eventFdClosing;
79 private volatile boolean shuttingDown;
80 private boolean closeCompleted;
81 private final PendingOpMap pendingOps;
82 private int nextRegistrationId = 1;
83
84 private static final long INVALID_ID = 0;
85 private static final long EVENTFD_TOKEN = PendingOpMap.token(1);
86 private static final long RINGFD_TOKEN = PendingOpMap.token(2);
87 private static final int KERNEL_TIMESPEC_SIZE = 16;
88
89 private static final int KERNEL_TIMESPEC_TV_SEC_FIELD = 0;
90 private static final int KERNEL_TIMESPEC_TV_NSEC_FIELD = 8;
91
92 private final ThreadAwareExecutor executor;
93
94 IoUringIoHandler(ThreadAwareExecutor executor, IoUringIoHandlerConfig config) {
95
96 IoUring.ensureAvailability();
97 this.executor = requireNonNull(executor, "executor");
98 requireNonNull(config, "config");
99 int setupFlags = Native.setupFlags(config.singleIssuer());
100
101
102
103 int cqSize = 2 * config.getRingSize();
104 if (config.needSetupCqeSize()) {
105 assert IoUring.isSetupCqeSizeSupported();
106 setupFlags |= Native.IORING_SETUP_CQSIZE;
107 cqSize = config.getCqSize();
108 }
109 this.ringBuffer = Native.createRingBuffer(config.getRingSize(), cqSize, setupFlags);
110 if (IoUring.isRegisterIowqMaxWorkersSupported() && config.needRegisterIowqMaxWorker()) {
111 int maxBoundedWorker = Math.max(config.getMaxBoundedWorker(), 0);
112 int maxUnboundedWorker = Math.max(config.getMaxUnboundedWorker(), 0);
113 int result = Native.ioUringRegisterIoWqMaxWorkers(ringBuffer.fd(), maxBoundedWorker, maxUnboundedWorker);
114 if (result < 0) {
115
116 ringBuffer.close();
117 throw new UncheckedIOException(Errors.newIOException("io_uring_register", result));
118 }
119 }
120
121 registeredIoUringBufferRing = new IntObjectHashMap<>();
122 Collection<IoUringBufferRingConfig> bufferRingConfigs = config.getInternBufferRingConfigs();
123 if (bufferRingConfigs != null && !bufferRingConfigs.isEmpty()) {
124 for (IoUringBufferRingConfig bufferRingConfig : bufferRingConfigs) {
125 try {
126 IoUringBufferRing ring = newBufferRing(ringBuffer.fd(), bufferRingConfig);
127 registeredIoUringBufferRing.put(bufferRingConfig.bufferGroupId(), ring);
128 } catch (Errors.NativeIoException e) {
129 for (IoUringBufferRing bufferRing : registeredIoUringBufferRing.values()) {
130 bufferRing.close();
131 }
132
133 ringBuffer.close();
134 throw new UncheckedIOException(e);
135 }
136 }
137 }
138
139 registrations = new IntObjectHashMap<>();
140 pendingOps = new PendingOpMap(IoUring.DEFAULT_PENDING_OPS_INITIAL_CAPACITY);
141 eventfd = Native.newBlockingEventFd();
142 eventfdReadBufCleanable = Buffer.allocateDirectBufferWithNativeOrder(Long.BYTES);
143 eventfdReadBuf = eventfdReadBufCleanable.buffer();
144 eventfdReadBufAddress = Buffer.memoryAddress(eventfdReadBuf);
145 timeoutMemoryCleanable = Buffer.allocateDirectBufferWithNativeOrder(KERNEL_TIMESPEC_SIZE);
146 timeoutMemory = timeoutMemoryCleanable.buffer();
147 timeoutMemoryAddress = Buffer.memoryAddress(timeoutMemory);
148 iovArray = new IovArray(IoUring.NUM_ELEMENTS_IOVEC);
149 iovArrayReferenceCollector = new IovArrayReferenceCollector(iovArray);
150 msgHdrMemoryArray = new MsgHdrMemoryArray((short) 1024);
151 }
152
153 @Override
154 public void initialize() {
155 ringBuffer.enable();
156
157 for (IoUringBufferRing bufferRing : registeredIoUringBufferRing.values()) {
158 bufferRing.initialize();
159 }
160 }
161
162 @Override
163 public int run(IoHandlerContext context) {
164 if (closeCompleted) {
165 if (context.shouldReportActiveIoTime()) {
166 context.reportActiveIoTime(0);
167 }
168 return 0;
169 }
170 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
171 CompletionQueue completionQueue = ringBuffer.ioUringCompletionQueue();
172 if (!completionQueue.hasCompletions() && context.canBlock()) {
173 if (eventfdReadSubmitted == 0) {
174 submitEventFdRead();
175 }
176 long timeoutNanos = context.deadlineNanos() == -1 ? -1 : context.delayNanos(System.nanoTime());
177 submitAndWaitWithTimeout(submissionQueue, false, timeoutNanos);
178 } else {
179
180 submitAndClearNow(submissionQueue);
181 }
182
183 int ioCompletions;
184 if (context.shouldReportActiveIoTime()) {
185 long activeIoStartTimeNanos = System.nanoTime();
186 ioCompletions = processCompletionsAndHandleOverflow(submissionQueue, completionQueue, this::handle);
187 long activeIoEndTimeNanos = System.nanoTime();
188 context.reportActiveIoTime(activeIoEndTimeNanos - activeIoStartTimeNanos);
189 } else {
190 ioCompletions = processCompletionsAndHandleOverflow(submissionQueue, completionQueue, this::handle);
191 }
192 return ioCompletions;
193 }
194
195 int submitIfFullAndGetRemaining() {
196 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
197 if (submissionQueue.remaining() == 0) {
198 if (submitAndClearNow(submissionQueue) == 0) {
199 throw new IllegalStateException("Submission queue is full and no submissions were accepted");
200 }
201 }
202 return submissionQueue.remaining();
203 }
204
205 private boolean needSubmit(int sqFlags) {
206 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
207 return submissionQueue.count() > 0
208 || (sqFlags & (Native.IORING_SQ_CQ_OVERFLOW | Native.IORING_SQ_TASKRUN)) != 0;
209 }
210
211 private int processCompletionsAndHandleOverflow(SubmissionQueue submissionQueue, CompletionQueue completionQueue,
212 CompletionCallback callback) {
213 int ioCompletions = 0;
214 for (int i = 0; i < 128; i++) {
215 long packed = completionQueue.process(callback);
216 int total = (int) (packed >>> 32);
217 ioCompletions += (int) packed;
218 int sqFlags = submissionQueue.flags();
219 if ((sqFlags & Native.IORING_SQ_CQ_OVERFLOW) != 0) {
220 logger.warn("CompletionQueue overflow detected, consider increasing size: {} ",
221 completionQueue.ringEntries);
222 }
223 if (total == 0) {
224 if (!needSubmit(sqFlags)) {
225 break;
226 }
227 submitAndClearNow0(submissionQueue);
228 }
229 }
230 return ioCompletions;
231 }
232
233 private int submitAndClearNow(SubmissionQueue submissionQueue) {
234 if (needSubmit(submissionQueue.flags())) {
235 return submitAndClearNow0(submissionQueue);
236 }
237 return 0;
238 }
239
240 private int submitAndClearNow0(SubmissionQueue submissionQueue) {
241
242 int submitted = submissionQueue.submitAndGetNow();
243
244
245
246 iovArray.clear();
247 msgHdrMemoryArray.clear();
248 return submitted;
249 }
250
251 private static IoUringBufferRing newBufferRing(int ringFd, IoUringBufferRingConfig bufferRingConfig)
252 throws Errors.NativeIoException {
253 short bufferRingSize = bufferRingConfig.bufferRingSize();
254 short bufferGroupId = bufferRingConfig.bufferGroupId();
255 int flags = bufferRingConfig.isIncremental() ? Native.IOU_PBUF_RING_INC : 0;
256 long ioUringBufRingAddr = Native.ioUringRegisterBufRing(ringFd, bufferRingSize, bufferGroupId, flags);
257 if (ioUringBufRingAddr < 0) {
258 throw Errors.newIOException("ioUringRegisterBufRing", (int) ioUringBufRingAddr);
259 }
260 return new IoUringBufferRing(ringFd,
261 Buffer.wrapMemoryAddressWithNativeOrder(ioUringBufRingAddr, Native.ioUringBufRingSize(bufferRingSize)),
262 bufferRingSize, bufferRingConfig.batchSize(),
263 bufferGroupId, bufferRingConfig.isIncremental(), bufferRingConfig.allocator(),
264 bufferRingConfig.isBatchAllocation()
265 );
266 }
267
268 IoUringBufferRing findBufferRing(short bgId) {
269 IoUringBufferRing cached = registeredIoUringBufferRing.get(bgId);
270 if (cached != null) {
271 return cached;
272 }
273 throw new IllegalArgumentException(
274 String.format("Cant find bgId:%d, please register it in ioUringIoHandler", bgId)
275 );
276 }
277
278 private static void handleLoopException(Throwable throwable) {
279 logger.warn("Unexpected exception in the IO event loop.", throwable);
280
281
282
283 try {
284 Thread.sleep(100);
285 } catch (InterruptedException ignore) {
286
287 }
288 }
289
290 private boolean handle(int res, int flags, long udata, ByteBuffer extraCqeData) {
291 try {
292 if (udata == EVENTFD_TOKEN) {
293 handleEventFdRead();
294 return false;
295 }
296 if (udata == RINGFD_TOKEN) {
297 return false;
298 }
299 if (udata >= 0) {
300 handleFastPath(res, flags, udata, extraCqeData);
301 return true;
302 }
303 handleSlowPath(res, flags, udata, extraCqeData);
304 return true;
305 } catch (Error e) {
306 throw e;
307 } catch (Throwable throwable) {
308 handleLoopException(throwable);
309 return true;
310 }
311 }
312
313 private void handleFastPath(int res, int flags, long udata, ByteBuffer extraCqeData) {
314 int id = UserData.decodeId(udata);
315 byte op = UserData.decodeOp(udata);
316 long userData = UserData.decodeData(udata);
317 DefaultIoUringIoRegistration registration = registrations.get(id);
318 if (registration != null) {
319 traceCompletion(registration, id, op, res);
320 registration.handle(res, flags, op, userData, extraCqeData);
321 return;
322 }
323 if (logger.isDebugEnabled()) {
324 logger.debug("ignoring packed completion for unknown registration (registrationId={}, op={}, userData={},"
325 + " res={})",
326 id, Native.opToStr(op), userData, res);
327 }
328 }
329
330 private void handleSlowPath(int res, int flags, long udata, ByteBuffer extraCqeData) {
331 long sequence = PendingOpMap.tokenSequence(udata);
332 int slot = pendingOps.findSlot(udata);
333 if (slot != -1) {
334 int registrationId = pendingOps.registrationId(slot);
335 DefaultIoUringIoRegistration registration = registrations.get(registrationId);
336 byte op = pendingOps.op(slot);
337 long userData = pendingOps.userData(slot);
338
339
340 if ((flags & Native.IORING_CQE_F_MORE) == 0) {
341 pendingOps.release(slot);
342 }
343
344
345 if (registration != null) {
346 traceCompletion(registration, registrationId, op, res);
347 registration.handle(res, flags, op, userData, extraCqeData);
348 return;
349 }
350 if (logger.isDebugEnabled()) {
351 logger.debug("ignoring slow-path completion for missing registration (registrationId={}, seq={}, "
352 + "op={}, userData={}, res={})",
353 registrationId, sequence, Native.opToStr(op), userData, res);
354 }
355 return;
356 }
357 if (logger.isDebugEnabled()) {
358 logger.debug("ignoring slow-path completion for unknown sequence (seq={}, res={})", sequence, res);
359 }
360 }
361
362 private void traceCompletion(DefaultIoUringIoRegistration registration, int registrationId, byte op, int res) {
363 if (!logger.isTraceEnabled()) {
364 return;
365 }
366 int fd = registration.fd();
367 if (fd != -1) {
368 logger.trace("completed(ring {}): {}(fd={}, res={})",
369 ringBuffer.fd(), Native.opToStr(op), fd, res);
370 } else {
371 logger.trace("completed(ring {}): {}(registrationId={}, res={})",
372 ringBuffer.fd(), Native.opToStr(op), registrationId, res);
373 }
374 }
375
376 private void handleEventFdRead() {
377 eventfdReadSubmitted = 0;
378 if (!eventFdClosing) {
379 eventfdAsyncNotify.set(false);
380 submitEventFdRead();
381 }
382 }
383
384 private void submitEventFdRead() {
385 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
386 eventfdReadSubmitted = submissionQueue.addEventFdRead(
387 eventfd.intValue(), eventfdReadBufAddress, 0, 8, EVENTFD_TOKEN);
388 }
389
390 private int submitAndWaitWithTimeout(SubmissionQueue submissionQueue,
391 boolean linkTimeout, long timeoutNanoSeconds) {
392 if (timeoutNanoSeconds != -1) {
393
394
395 long seconds, nanoSeconds;
396 if (timeoutNanoSeconds == 0) {
397 seconds = 0;
398 nanoSeconds = 0;
399 } else {
400 seconds = (int) min(timeoutNanoSeconds / 1000000000L, Integer.MAX_VALUE);
401 nanoSeconds = (int) max(timeoutNanoSeconds - seconds * 1000000000L, 0);
402 }
403
404 timeoutMemory.putLong(KERNEL_TIMESPEC_TV_SEC_FIELD, seconds);
405 timeoutMemory.putLong(KERNEL_TIMESPEC_TV_NSEC_FIELD, nanoSeconds);
406 if (linkTimeout) {
407 submissionQueue.addLinkTimeout(timeoutMemoryAddress, RINGFD_TOKEN);
408 } else {
409 submissionQueue.addTimeout(timeoutMemoryAddress, RINGFD_TOKEN);
410 }
411 }
412 int submitted = submissionQueue.submitAndGet();
413
414
415 iovArray.clear();
416 msgHdrMemoryArray.clear();
417 return submitted;
418 }
419
420 @Override
421 public void prepareToDestroy() {
422 shuttingDown = true;
423 CompletionQueue completionQueue = ringBuffer.ioUringCompletionQueue();
424 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
425
426 List<DefaultIoUringIoRegistration> copy = new ArrayList<>(registrations.values());
427
428 for (DefaultIoUringIoRegistration registration: copy) {
429 registration.close();
430 }
431
432
433 Native.eventFdWrite(eventfd.intValue(), 1L);
434
435
436 submissionQueue.addNop((byte) Native.IOSQE_IO_DRAIN, RINGFD_TOKEN);
437
438
439 submissionQueue.submitAndGet();
440
441 while (completionQueue.hasCompletions()) {
442 processCompletionsAndHandleOverflow(submissionQueue, completionQueue, this::handle);
443 if (submissionQueue.count() > 0) {
444 submissionQueue.submitAndGetNow();
445 }
446 }
447 }
448
449 @Override
450 public void destroy() {
451 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
452 CompletionQueue completionQueue = ringBuffer.ioUringCompletionQueue();
453 drainEventFd();
454 if (submissionQueue.remaining() < 2) {
455
456
457 submissionQueue.submit();
458 }
459
460
461
462
463
464
465 submissionQueue.addNop((byte) (Native.IOSQE_IO_DRAIN | Native.IOSQE_LINK), RINGFD_TOKEN);
466
467 submitAndWaitWithTimeout(submissionQueue, true, TimeUnit.MILLISECONDS.toNanos(200));
468 completionQueue.process(this::handle);
469 for (IoUringBufferRing ioUringBufferRing : registeredIoUringBufferRing.values()) {
470 ioUringBufferRing.close();
471 }
472 completeRingClose();
473 }
474
475
476
477
478
479 private void drainEventFd() {
480 CompletionQueue completionQueue = ringBuffer.ioUringCompletionQueue();
481 SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
482 assert !eventFdClosing;
483 eventFdClosing = true;
484 boolean eventPending = eventfdAsyncNotify.getAndSet(true);
485 if (eventPending) {
486
487
488 while (eventfdReadSubmitted == 0) {
489 submitEventFdRead();
490 submissionQueue.submit();
491 }
492
493 class DrainFdEventCallback implements CompletionCallback {
494 boolean eventFdDrained;
495
496 @Override
497 public boolean handle(int res, int flags, long udata, ByteBuffer extraCqeData) {
498 if (udata == EVENTFD_TOKEN) {
499 eventFdDrained = true;
500 }
501 return IoUringIoHandler.this.handle(res, flags, udata, extraCqeData);
502 }
503 }
504 final DrainFdEventCallback handler = new DrainFdEventCallback();
505 completionQueue.process(handler);
506 while (!handler.eventFdDrained) {
507 submissionQueue.submitAndGet();
508 processCompletionsAndHandleOverflow(submissionQueue, completionQueue, handler);
509 }
510 }
511
512
513
514 if (eventfdReadSubmitted != 0) {
515 submissionQueue.addCancel(eventfdReadSubmitted, EVENTFD_TOKEN);
516 eventfdReadSubmitted = 0;
517 submissionQueue.submit();
518 }
519 }
520
521 private void completeRingClose() {
522 if (closeCompleted) {
523
524 return;
525 }
526 closeCompleted = true;
527 ringBuffer.close();
528 closeWakeupGate();
529 try {
530 eventfd.close();
531 } catch (IOException e) {
532 logger.warn("Failed to close eventfd", e);
533 }
534 eventfdReadBufCleanable.clean();
535 timeoutMemoryCleanable.clean();
536 iovArray.release();
537 msgHdrMemoryArray.release();
538 }
539
540 @Override
541 public IoRegistration register(IoHandle handle) throws Exception {
542 IoUringIoHandle ioHandle = cast(handle);
543 if (shuttingDown) {
544 throw new IllegalStateException("IoUringIoHandler is shutting down");
545 }
546 int startId = nextRegistrationId;
547 DefaultIoUringIoRegistration registration = new DefaultIoUringIoRegistration(executor, ioHandle);
548 for (;;) {
549 int id = nextRegistrationId();
550 DefaultIoUringIoRegistration old = registrations.put(id, registration);
551 if (old != null) {
552 assert old.handle != registration.handle;
553 registrations.put(id, old);
554 if (nextRegistrationId == startId) {
555 throw new IllegalStateException("registration id space exhausted");
556 }
557 } else {
558 registration.setId(id);
559 ioHandle.registered();
560 break;
561 }
562 }
563
564 return registration;
565 }
566
567 private int nextRegistrationId() {
568
569
570 int id = nextRegistrationId;
571 nextRegistrationId = id == Integer.MAX_VALUE ? 1 : id + 1;
572 return id;
573 }
574
575 private final class DefaultIoUringIoRegistration implements IoRegistration {
576 private final AtomicBoolean canceled = new AtomicBoolean();
577 private final ThreadAwareExecutor executor;
578 private final IoUringIoEvent event = new IoUringIoEvent(0, 0, (byte) 0, 0L);
579 final IoUringIoHandle handle;
580
581 private boolean removeLater;
582 private int outstandingCompletions;
583 private int id;
584
585 DefaultIoUringIoRegistration(ThreadAwareExecutor executor, IoUringIoHandle handle) {
586 this.executor = executor;
587 this.handle = handle;
588 }
589
590 void setId(int id) {
591 this.id = id;
592 }
593
594 @Override
595 public long submit(IoOps ops) {
596 IoUringIoOps ioOps = (IoUringIoOps) ops;
597 if (!isValid()) {
598 return INVALID_ID;
599 }
600 if ((ioOps.flags() & Native.IOSQE_CQE_SKIP_SUCCESS) != 0) {
601
602
603 throw new IllegalArgumentException("IOSQE_CQE_SKIP_SUCCESS not supported");
604 }
605 long userData = ioOps.userData();
606
607 if (canUseFastPath(userData)) {
608 long packedSeq = UserData.encode(id, ioOps.opcode(), (short) userData);
609 if (executor.isExecutorThread(Thread.currentThread())) {
610 submitFastPath0(ioOps, packedSeq);
611 } else {
612 executor.execute(() -> submitFastPath0(ioOps, packedSeq));
613 }
614 return packedSeq;
615 }
616 long token = pendingOps.nextToken();
617 if (executor.isExecutorThread(Thread.currentThread())) {
618 submitSlowPath0(ioOps, token, userData);
619 } else {
620 executor.execute(() -> submitSlowPath0(ioOps, token, userData));
621 }
622 return token;
623 }
624
625 private void submitFastPath0(IoUringIoOps ioOps, long seq) {
626 ringBuffer.ioUringSubmissionQueue().enqueueSqe(ioOps.opcode(), ioOps.flags(), ioOps.ioPrio(),
627 ioOps.fd(), ioOps.union1(), ioOps.union2(), ioOps.len(), ioOps.union3(), seq,
628 ioOps.union4(), ioOps.personality(), ioOps.union5(), ioOps.union6()
629 );
630 outstandingCompletions++;
631 }
632
633 private void submitSlowPath0(IoUringIoOps ioOps, long token, long userData) {
634 pendingOps.registerNormal(token, id, ioOps.opcode(), userData);
635 ringBuffer.ioUringSubmissionQueue().enqueueSqe(ioOps.opcode(), ioOps.flags(), ioOps.ioPrio(),
636 ioOps.fd(), ioOps.union1(), ioOps.union2(), ioOps.len(), ioOps.union3(), token,
637 ioOps.union4(), ioOps.personality(), ioOps.union5(), ioOps.union6()
638 );
639 outstandingCompletions++;
640 }
641
642 private boolean canUseFastPath(long userData) {
643 return ((short) userData) == userData;
644 }
645
646 private int fd() {
647 if (handle instanceof AbstractIoUringChannel) {
648 return ((AbstractIoUringChannel) handle).fd().intValue();
649 }
650 return -1;
651 }
652
653 @SuppressWarnings("unchecked")
654 @Override
655 public <T> T attachment() {
656 return (T) IoUringIoHandler.this;
657 }
658
659 @Override
660 public boolean isValid() {
661 return !canceled.get();
662 }
663
664 @Override
665 public boolean cancel() {
666 if (!canceled.compareAndSet(false, true)) {
667
668 return false;
669 }
670 if (executor.isExecutorThread(Thread.currentThread())) {
671 tryRemove();
672 } else {
673 executor.execute(this::tryRemove);
674 }
675 return true;
676 }
677
678 private void tryRemove() {
679 if (outstandingCompletions > 0) {
680
681
682 removeLater = true;
683 return;
684 }
685 remove();
686 }
687
688 private void remove() {
689 DefaultIoUringIoRegistration old = registrations.remove(id);
690 assert old == this;
691 handle.unregistered();
692 }
693
694 void close() {
695
696
697 assert executor.isExecutorThread(Thread.currentThread());
698 try {
699 handle.close();
700 } catch (Exception e) {
701 logger.debug("Exception during closing " + handle, e);
702 }
703 }
704
705 void handle(int res, int flags, byte op, long userData, ByteBuffer extraCqeData) {
706 event.update(res, flags, op, userData, extraCqeData);
707 handle.handle(this, event);
708
709
710 if ((flags & Native.IORING_CQE_F_MORE) == 0 && --outstandingCompletions == 0 && removeLater) {
711
712 removeLater = false;
713 remove();
714 }
715 }
716 }
717
718 private static IoUringIoHandle cast(IoHandle handle) {
719 if (handle instanceof IoUringIoHandle) {
720 return (IoUringIoHandle) handle;
721 }
722 throw new IllegalArgumentException("IoHandle of type " + StringUtil.simpleClassName(handle) + " not supported");
723 }
724
725 @Override
726 public void wakeup() {
727 if (!executor.isExecutorThread(Thread.currentThread()) &&
728 !eventfdAsyncNotify.getAndSet(true)) {
729
730
731
732
733 int s;
734 do {
735 s = wakeupWriters.get();
736 if ((s & WAKEUP_CLOSED) != 0) {
737 return;
738 }
739 } while (!wakeupWriters.compareAndSet(s, s + 1));
740 try {
741
742 Native.eventFdWrite(eventfd.intValue(), 1L);
743 } finally {
744 wakeupWriters.decrementAndGet();
745 }
746 }
747 }
748
749 private void closeWakeupGate() {
750 int s;
751 do {
752 s = wakeupWriters.get();
753 } while (!wakeupWriters.compareAndSet(s, s | WAKEUP_CLOSED));
754
755
756 while ((wakeupWriters.get() & ~WAKEUP_CLOSED) != 0) {
757 Thread.onSpinWait();
758 }
759 }
760
761 @Override
762 public boolean isCompatible(Class<? extends IoHandle> handleType) {
763 return IoUringIoHandle.class.isAssignableFrom(handleType);
764 }
765
766 IovArray iovArray() {
767 if (iovArray.isFull()) {
768
769 submitAndClearNow(ringBuffer.ioUringSubmissionQueue());
770 assert iovArray.count() == 0;
771 }
772 return iovArray;
773 }
774
775
776
777
778
779 IovArrayReferenceCollector iovArrayReferenceCollector() {
780 return iovArrayReferenceCollector;
781 }
782
783 MsgHdrMemoryArray msgHdrMemoryArray() {
784 if (msgHdrMemoryArray.isFull()) {
785
786 submitAndClearNow(ringBuffer.ioUringSubmissionQueue());
787 }
788 return msgHdrMemoryArray;
789 }
790
791
792
793
794 byte[] inet4AddressArray() {
795 return inet4AddressArray;
796 }
797
798
799
800
801 byte[] inet6AddressArray() {
802 return inet6AddressArray;
803 }
804
805
806
807
808
809
810 public static IoHandlerFactory newFactory() {
811 return newFactory(new IoUringIoHandlerConfig());
812 }
813
814
815
816
817
818
819
820
821 public static IoHandlerFactory newFactory(int ringSize) {
822 IoUringIoHandlerConfig configuration = new IoUringIoHandlerConfig();
823 configuration.setRingSize(ringSize);
824 return eventLoop -> new IoUringIoHandler(eventLoop, configuration);
825 }
826
827
828
829
830
831
832
833 public static IoHandlerFactory newFactory(IoUringIoHandlerConfig config) {
834 IoUring.ensureAvailability();
835 final IoUringIoHandlerConfig copy = ObjectUtil.checkNotNull(config, "config").verifyAndClone();
836 return new IoHandlerFactory() {
837 @Override
838 public IoHandler newHandler(ThreadAwareExecutor eventLoop) {
839 return new IoUringIoHandler(eventLoop, copy);
840 }
841
842 @Override
843 public boolean isChangingThreadSupported() {
844 return !copy.singleIssuer();
845 }
846 };
847 }
848 }