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