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.buffer.ByteBuf;
19 import io.netty.channel.AddressedEnvelope;
20 import io.netty.channel.ChannelException;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.channel.ChannelMetadata;
23 import io.netty.channel.ChannelOutboundBuffer;
24 import io.netty.channel.ChannelPipeline;
25 import io.netty.channel.ChannelPromise;
26 import io.netty.channel.DefaultAddressedEnvelope;
27 import io.netty.channel.IoRegistration;
28 import io.netty.channel.socket.DatagramChannel;
29 import io.netty.channel.socket.DatagramChannelConfig;
30 import io.netty.channel.socket.DatagramPacket;
31 import io.netty.channel.socket.SocketProtocolFamily;
32 import io.netty.channel.unix.Errors;
33 import io.netty.channel.unix.Errors.NativeIoException;
34 import io.netty.channel.unix.SegmentedDatagramPacket;
35 import io.netty.channel.unix.Socket;
36 import io.netty.util.UncheckedBooleanSupplier;
37 import io.netty.util.internal.ObjectUtil;
38 import io.netty.util.internal.StringUtil;
39 import io.netty.util.internal.SystemPropertyUtil;
40 import io.netty.util.internal.logging.InternalLogger;
41 import io.netty.util.internal.logging.InternalLoggerFactory;
42
43 import java.io.IOException;
44 import java.net.Inet4Address;
45 import java.net.InetAddress;
46 import java.net.InetSocketAddress;
47 import java.net.NetworkInterface;
48 import java.net.PortUnreachableException;
49 import java.net.SocketAddress;
50 import java.nio.channels.UnresolvedAddressException;
51
52 import static io.netty.channel.unix.Errors.ioResult;
53
54 public final class IoUringDatagramChannel extends AbstractIoUringChannel implements DatagramChannel {
55 private static final InternalLogger logger = InternalLoggerFactory.getInstance(IoUringDatagramChannel.class);
56 private static final boolean IP_MULTICAST_ALL =
57 SystemPropertyUtil.getBoolean("io.netty.channel.iouring.ipMulticastAll", false);
58 private static final ChannelMetadata METADATA = new ChannelMetadata(true, 16);
59 private static final String EXPECTED_TYPES =
60 " (expected: " + StringUtil.simpleClassName(DatagramPacket.class) + ", " +
61 StringUtil.simpleClassName(AddressedEnvelope.class) + '<' +
62 StringUtil.simpleClassName(ByteBuf.class) + ", " +
63 StringUtil.simpleClassName(InetSocketAddress.class) + ">, " +
64 StringUtil.simpleClassName(ByteBuf.class) + ')';
65
66 private final IoUringDatagramChannelConfig config;
67 private volatile boolean connected;
68
69 static {
70 if (logger.isDebugEnabled()) {
71 logger.debug("-Dio.netty.channel.iouring.ipMulticastAll: {}", IP_MULTICAST_ALL);
72 }
73 }
74
75
76
77
78
79
80
81 private final MsgHdrMemoryArray recvmsgHdrs = new MsgHdrMemoryArray((short) 256);
82 private final MsgHdrMemoryArray sendmsgHdrs = new MsgHdrMemoryArray((short) 256);
83 private final int[] sendmsgResArray = new int[sendmsgHdrs.capacity()];
84
85
86
87
88
89 public IoUringDatagramChannel() {
90 this(null);
91 }
92
93
94
95
96
97 public IoUringDatagramChannel(SocketProtocolFamily family) {
98 this(LinuxSocket.newSocketDgram(useIpv6(family)), false);
99 }
100
101 private static boolean useIpv6(SocketProtocolFamily family) {
102 if (family == null) {
103 return Socket.isIPv6Preferred();
104 }
105 return family == SocketProtocolFamily.INET6;
106 }
107
108
109
110
111
112 public IoUringDatagramChannel(int fd) {
113 this(new LinuxSocket(fd), true);
114 }
115
116 private IoUringDatagramChannel(LinuxSocket fd, boolean active) {
117
118 super(null, fd, active);
119
120
121 try {
122 fd.setIpMulticastAll(IP_MULTICAST_ALL);
123 } catch (IOException | ChannelException e) {
124 logger.debug("Failed to set IP_MULTICAST_ALL to {}", IP_MULTICAST_ALL, e);
125 }
126
127 config = new IoUringDatagramChannelConfig(this);
128 }
129
130 @Override
131 protected boolean isStreamSocket() {
132 return false;
133 }
134
135 @Override
136 public InetSocketAddress remoteAddress() {
137 return (InetSocketAddress) super.remoteAddress();
138 }
139
140 @Override
141 public InetSocketAddress localAddress() {
142 return (InetSocketAddress) super.localAddress();
143 }
144
145 @Override
146 public ChannelMetadata metadata() {
147 return METADATA;
148 }
149
150 @Override
151 public boolean isActive() {
152 return socket.isOpen() && (config.getActiveOnOpen() && isRegistered() || super.isActive());
153 }
154
155 @Override
156 public boolean isConnected() {
157 return connected;
158 }
159
160 @Override
161 public ChannelFuture joinGroup(InetAddress multicastAddress) {
162 return joinGroup(multicastAddress, newPromise());
163 }
164
165 @Override
166 public ChannelFuture joinGroup(InetAddress multicastAddress, ChannelPromise promise) {
167 try {
168 return joinGroup(
169 multicastAddress,
170 NetworkInterface.getByInetAddress(localAddress().getAddress()), null, promise);
171 } catch (IOException e) {
172 promise.setFailure(e);
173 }
174 return promise;
175 }
176
177 @Override
178 public ChannelFuture joinGroup(
179 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
180 return joinGroup(multicastAddress, networkInterface, newPromise());
181 }
182
183 @Override
184 public ChannelFuture joinGroup(
185 InetSocketAddress multicastAddress, NetworkInterface networkInterface,
186 ChannelPromise promise) {
187 return joinGroup(multicastAddress.getAddress(), networkInterface, null, promise);
188 }
189
190 @Override
191 public ChannelFuture joinGroup(
192 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
193 return joinGroup(multicastAddress, networkInterface, source, newPromise());
194 }
195
196 @Override
197 public ChannelFuture joinGroup(
198 final InetAddress multicastAddress, final NetworkInterface networkInterface,
199 final InetAddress source, final ChannelPromise promise) {
200
201 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
202 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
203
204 try {
205 socket.joinGroup(multicastAddress, networkInterface, source);
206 promise.setSuccess();
207 } catch (IOException e) {
208 promise.setFailure(e);
209 }
210 return promise;
211 }
212
213 @Override
214 public ChannelFuture leaveGroup(InetAddress multicastAddress) {
215 return leaveGroup(multicastAddress, newPromise());
216 }
217
218 @Override
219 public ChannelFuture leaveGroup(InetAddress multicastAddress, ChannelPromise promise) {
220 try {
221 return leaveGroup(
222 multicastAddress, NetworkInterface.getByInetAddress(localAddress().getAddress()), null, promise);
223 } catch (IOException e) {
224 promise.setFailure(e);
225 }
226 return promise;
227 }
228
229 @Override
230 public ChannelFuture leaveGroup(
231 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
232 return leaveGroup(multicastAddress, networkInterface, newPromise());
233 }
234
235 @Override
236 public ChannelFuture leaveGroup(
237 InetSocketAddress multicastAddress,
238 NetworkInterface networkInterface, ChannelPromise promise) {
239 return leaveGroup(multicastAddress.getAddress(), networkInterface, null, promise);
240 }
241
242 @Override
243 public ChannelFuture leaveGroup(
244 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
245 return leaveGroup(multicastAddress, networkInterface, source, newPromise());
246 }
247
248 @Override
249 public ChannelFuture leaveGroup(
250 final InetAddress multicastAddress, final NetworkInterface networkInterface, final InetAddress source,
251 final ChannelPromise promise) {
252 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
253 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
254
255 try {
256 socket.leaveGroup(multicastAddress, networkInterface, source);
257 promise.setSuccess();
258 } catch (IOException e) {
259 promise.setFailure(e);
260 }
261 return promise;
262 }
263
264 @Override
265 public ChannelFuture block(
266 InetAddress multicastAddress, NetworkInterface networkInterface,
267 InetAddress sourceToBlock) {
268 return block(multicastAddress, networkInterface, sourceToBlock, newPromise());
269 }
270
271 @Override
272 public ChannelFuture block(
273 final InetAddress multicastAddress, final NetworkInterface networkInterface,
274 final InetAddress sourceToBlock, final ChannelPromise promise) {
275 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
276 ObjectUtil.checkNotNull(sourceToBlock, "sourceToBlock");
277 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
278
279 promise.setFailure(new UnsupportedOperationException("Multicast not supported"));
280 return promise;
281 }
282
283 @Override
284 public ChannelFuture block(InetAddress multicastAddress, InetAddress sourceToBlock) {
285 return block(multicastAddress, sourceToBlock, newPromise());
286 }
287
288 @Override
289 public ChannelFuture block(
290 InetAddress multicastAddress, InetAddress sourceToBlock, ChannelPromise promise) {
291 try {
292 return block(
293 multicastAddress,
294 NetworkInterface.getByInetAddress(localAddress().getAddress()),
295 sourceToBlock, promise);
296 } catch (Throwable e) {
297 promise.setFailure(e);
298 }
299 return promise;
300 }
301
302 @Override
303 protected AbstractUnsafe newUnsafe() {
304 return new IoUringDatagramChannelUnsafe();
305 }
306
307 @Override
308 protected void doBind(SocketAddress localAddress) throws Exception {
309 if (localAddress instanceof InetSocketAddress) {
310 InetSocketAddress socketAddress = (InetSocketAddress) localAddress;
311 if (socketAddress.getAddress().isAnyLocalAddress() &&
312 socketAddress.getAddress() instanceof Inet4Address) {
313 if (socket.family() == SocketProtocolFamily.INET6) {
314 localAddress = new InetSocketAddress(LinuxSocket.INET6_ANY, socketAddress.getPort());
315 }
316 }
317 }
318 super.doBind(localAddress);
319 active = true;
320 }
321
322 private static void checkUnresolved(AddressedEnvelope<?, ?> envelope) {
323 if (envelope.recipient() instanceof InetSocketAddress
324 && (((InetSocketAddress) envelope.recipient()).isUnresolved())) {
325 throw new UnresolvedAddressException();
326 }
327 }
328
329 @Override
330 protected Object filterOutboundMessage(Object msg) {
331 if (msg instanceof DatagramPacket) {
332 DatagramPacket packet = (DatagramPacket) msg;
333 checkUnresolved(packet);
334 ByteBuf content = packet.content();
335 return !content.hasMemoryAddress() ?
336 packet.replace(newDirectBuffer(packet, content)) : msg;
337 }
338
339 if (msg instanceof ByteBuf) {
340 ByteBuf buf = (ByteBuf) msg;
341 return !buf.hasMemoryAddress()? newDirectBuffer(buf) : buf;
342 }
343
344 if (msg instanceof AddressedEnvelope) {
345 @SuppressWarnings("unchecked")
346 AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
347 checkUnresolved(e);
348 if (e.content() instanceof ByteBuf &&
349 (e.recipient() == null || e.recipient() instanceof InetSocketAddress)) {
350
351 ByteBuf content = (ByteBuf) e.content();
352 return !content.hasMemoryAddress()?
353 new DefaultAddressedEnvelope<>(
354 newDirectBuffer(e, content), (InetSocketAddress) e.recipient()) : e;
355 }
356 }
357
358 throw new UnsupportedOperationException(
359 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
360 }
361
362 @Override
363 public DatagramChannelConfig config() {
364 return config;
365 }
366
367 @Override
368 protected void doDisconnect() throws Exception {
369
370 socket.disconnect();
371 connected = active = false;
372
373 resetCachedAddresses();
374 }
375
376 @Override
377 protected void doClose() throws Exception {
378 super.doClose();
379 connected = false;
380 }
381
382 private final class IoUringDatagramChannelUnsafe extends AbstractUringUnsafe {
383 private final WriteProcessor writeProcessor = new WriteProcessor();
384
385 private ByteBuf readBuffer;
386
387 private final class WriteProcessor implements ChannelOutboundBuffer.MessageProcessor {
388 private int written;
389 @Override
390 public boolean processMessage(Object msg) {
391 if (scheduleWrite(msg, written == 0)) {
392 written++;
393 return true;
394 }
395 return false;
396 }
397
398 int write(ChannelOutboundBuffer in) {
399 written = 0;
400 try {
401 in.forEachFlushedMessage(this);
402 } catch (Exception e) {
403
404 throw new IllegalStateException(e);
405 }
406 return written;
407 }
408 }
409
410 @Override
411 protected void readComplete0(byte op, int res, int flags, short data, int outstanding) {
412 assert outstanding != -1 : "multi-shot not implemented yet";
413
414 final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
415 final ChannelPipeline pipeline = pipeline();
416 ByteBuf byteBuf = this.readBuffer;
417 assert byteBuf != null;
418 MsgHdrMemory hdr = recvmsgHdrs.hdr(data);
419
420 recvmsgHdrs.setId(data, MsgHdrMemoryArray.NO_ID);
421
422 try {
423 if (res < 0) {
424 if (res != Native.ERRNO_ECANCELED_NEGATIVE) {
425
426
427 allocHandle.lastBytesRead(ioResult("io_uring recvmsg", res));
428 }
429 } else {
430 allocHandle.lastBytesRead(res);
431 if (hdr.hasPort(IoUringDatagramChannel.this)) {
432 allocHandle.incMessagesRead(1);
433 DatagramPacket packet = hdr.get(
434 IoUringDatagramChannel.this, registration().attachment(), byteBuf, res);
435 pipeline.fireChannelRead(packet);
436 }
437 }
438 } catch (Throwable t) {
439 Throwable e = (connected && t instanceof NativeIoException) ?
440 translateForConnected((NativeIoException) t) : t;
441 pipeline.fireExceptionCaught(e);
442 }
443
444 if (outstanding == 0) {
445
446
447 this.readBuffer.release();
448 this.readBuffer = null;
449 recvmsgHdrs.clear();
450
451 if (res != Native.ERRNO_ECANCELED_NEGATIVE) {
452 if (allocHandle.lastBytesRead() > 0 &&
453 allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER) &&
454
455
456
457
458 (!IoUring.isCqeFSockNonEmptySupported() ||
459 (flags & Native.IORING_CQE_F_SOCK_NONEMPTY) != 0)) {
460
461 scheduleRead(false);
462 } else {
463
464 allocHandle.readComplete();
465 pipeline.fireChannelReadComplete();
466 }
467 }
468 }
469 }
470
471 @Override
472 protected int scheduleRead0(boolean first, boolean socketIsEmpty) {
473 final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
474 IoUringIoHandler ioUringIoHandler = registration().attachment();
475 int submissionQueueRemaining = ioUringIoHandler.submitIfFullAndGetRemaining();
476 ByteBuf byteBuf = allocHandle.allocate(alloc());
477 assert readBuffer == null;
478 readBuffer = byteBuf;
479
480 int writable = byteBuf.writableBytes();
481 allocHandle.attemptedBytesRead(writable);
482 int datagramSize = ((IoUringDatagramChannelConfig) config()).getMaxDatagramPayloadSize();
483
484 int numDatagram = datagramSize == 0 ? 1 : Math.max(1, byteBuf.writableBytes() / datagramSize);
485 numDatagram = Math.min(Math.min(submissionQueueRemaining, recvmsgHdrs.capacity()), numDatagram);
486 int scheduled = scheduleRecvmsg(byteBuf, numDatagram, datagramSize);
487 if (scheduled == 0) {
488
489
490 readBuffer = null;
491 byteBuf.release();
492 }
493 return scheduled;
494 }
495
496 private int scheduleRecvmsg(ByteBuf byteBuf, int numDatagram, int datagramSize) {
497 int writable = byteBuf.writableBytes();
498 long bufferAddress = IoUring.memoryAddress(byteBuf) + byteBuf.writerIndex();
499 if (numDatagram <= 1) {
500 return scheduleRecvmsg0(bufferAddress, writable, true, false) ? 1 : 0;
501 }
502 int i = 0;
503
504 for (; i < numDatagram && writable >= datagramSize; i++) {
505 if (!scheduleRecvmsg0(bufferAddress, datagramSize, i == 0, i + 1 < numDatagram)) {
506 break;
507 }
508 bufferAddress += datagramSize;
509 writable -= datagramSize;
510 }
511 return i;
512 }
513
514 private boolean scheduleRecvmsg0(long bufferAddress, int bufferLength, boolean first, boolean more) {
515 MsgHdrMemory msgHdrMemory = recvmsgHdrs.nextHdr();
516 if (msgHdrMemory == null) {
517
518 return false;
519 }
520 msgHdrMemory.set(socket, null, bufferAddress, bufferLength, (short) 0);
521
522 int fd = fd().intValue();
523 int msgFlags = first ? 0 : Native.MSG_DONTWAIT;
524 int sqeFlags = more ? Native.IOSQE_HARDLINK : 0;
525 IoRegistration registration = registration();
526
527
528 IoUringIoOps ops = IoUringIoOps.newRecvmsg(
529 fd, (byte) sqeFlags, msgFlags, msgHdrMemory.address(), msgHdrMemory.idx());
530 long id = registration.submit(ops);
531 if (id == 0) {
532
533 recvmsgHdrs.restoreNextHdr(msgHdrMemory);
534 return false;
535 }
536 recvmsgHdrs.setId(msgHdrMemory.idx(), id);
537 return true;
538 }
539
540 @Override
541 boolean writeComplete0(byte op, int res, int flags, short data, int outstanding) {
542 ChannelOutboundBuffer outboundBuffer = outboundBuffer();
543
544
545 sendmsgHdrs.setId(data, MsgHdrMemoryArray.NO_ID);
546 sendmsgResArray[data] = res;
547
548 if (outstanding == 0) {
549
550 boolean writtenSomething = false;
551 int numWritten = sendmsgHdrs.length();
552 sendmsgHdrs.clear();
553 for (int i = 0; i < numWritten; i++) {
554 writtenSomething |= removeFromOutboundBuffer(
555 outboundBuffer, sendmsgResArray[i], "io_uring sendmsg");
556 }
557 return writtenSomething;
558 }
559 return true;
560 }
561
562 private boolean removeFromOutboundBuffer(ChannelOutboundBuffer outboundBuffer, int res, String errormsg) {
563 if (res >= 0) {
564
565 return outboundBuffer.remove();
566 }
567 if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
568 return false;
569 }
570 try {
571 return ioResult(errormsg, res) != 0;
572 } catch (Throwable cause) {
573 Throwable e = (connected && cause instanceof NativeIoException) ?
574 translateForConnected((NativeIoException) cause) : cause;
575 return outboundBuffer.remove(e);
576 }
577 }
578
579 @Override
580 void connectComplete(byte op, int res, int flags, short data) {
581 if (res >= 0) {
582 connected = true;
583 }
584 super.connectComplete(op, res, flags, data);
585 }
586
587 @Override
588 protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
589 return writeProcessor.write(in);
590 }
591
592 @Override
593 protected int scheduleWriteSingle(Object msg) {
594 return scheduleWrite(msg, true) ? 1 : 0;
595 }
596
597 private boolean scheduleWrite(Object msg, boolean first) {
598 final ByteBuf data;
599 final InetSocketAddress remoteAddress;
600 final int segmentSize;
601 if (msg instanceof AddressedEnvelope) {
602 @SuppressWarnings("unchecked")
603 AddressedEnvelope<ByteBuf, InetSocketAddress> envelope =
604 (AddressedEnvelope<ByteBuf, InetSocketAddress>) msg;
605 data = envelope.content();
606 remoteAddress = envelope.recipient();
607 if (msg instanceof SegmentedDatagramPacket) {
608 segmentSize = ((SegmentedDatagramPacket) msg).segmentSize();
609 } else {
610 segmentSize = 0;
611 }
612 } else {
613 data = (ByteBuf) msg;
614 remoteAddress = (InetSocketAddress) remoteAddress();
615 segmentSize = 0;
616 }
617
618 long bufferAddress = IoUring.memoryAddress(data) + data.readerIndex();
619 return scheduleSendmsg(remoteAddress, bufferAddress, data.readableBytes(), segmentSize, first);
620 }
621
622 private boolean scheduleSendmsg(InetSocketAddress remoteAddress, long bufferAddress,
623 int bufferLength, int segmentSize, boolean first) {
624 MsgHdrMemory hdr = sendmsgHdrs.nextHdr();
625 if (hdr == null) {
626
627
628 return false;
629 }
630 hdr.set(socket, remoteAddress, bufferAddress, bufferLength, (short) segmentSize);
631
632 int fd = fd().intValue();
633 int msgFlags = first ? 0 : Native.MSG_DONTWAIT;
634 IoRegistration registration = registration();
635 IoUringIoOps ops = IoUringIoOps.newSendmsg(fd, (byte) 0, msgFlags, hdr.address(), hdr.idx());
636 long id = registration.submit(ops);
637 if (id == 0) {
638
639 sendmsgHdrs.restoreNextHdr(hdr);
640 return false;
641 }
642 sendmsgHdrs.setId(hdr.idx(), id);
643 return true;
644 }
645
646 @Override
647 public void unregistered() {
648 super.unregistered();
649 sendmsgHdrs.release();
650 recvmsgHdrs.release();
651 assert readBuffer == null;
652 }
653 }
654
655 private static IOException translateForConnected(NativeIoException e) {
656
657 if (e.expectedErr() == Errors.ERROR_ECONNREFUSED_NEGATIVE) {
658 PortUnreachableException error = new PortUnreachableException(e.getMessage());
659 error.initCause(e);
660 return error;
661 }
662 return e;
663 }
664
665
666
667
668
669
670 public static boolean isSegmentedDatagramPacketSupported() {
671 return IoUring.isAvailable();
672 }
673
674 @Override
675 protected void cancelOutstandingReads(IoRegistration registration, int numOutstandingReads) {
676 if (numOutstandingReads > 0) {
677 int canceled = cancel(registration, Native.IORING_OP_RECVMSG, recvmsgHdrs);
678 assert canceled == numOutstandingReads;
679 }
680 }
681
682 @Override
683 protected void cancelOutstandingWrites(IoRegistration registration, int numOutstandingWrites) {
684 if (numOutstandingWrites > 0) {
685 int canceled = cancel(registration, Native.IORING_OP_SENDMSG, sendmsgHdrs);
686 assert canceled == numOutstandingWrites;
687 }
688 }
689
690 private int cancel(IoRegistration registration, byte op, MsgHdrMemoryArray array) {
691 int cancelled = 0;
692 for (int idx = 0; idx < array.length(); idx++) {
693 long id = array.id(idx);
694 if (id == MsgHdrMemoryArray.NO_ID) {
695 continue;
696 }
697
698
699 IoUringIoOps ops = IoUringIoOps.newAsyncCancel((byte) 0, id, op);
700 registration.submit(ops);
701 cancelled++;
702 }
703 return cancelled;
704 }
705
706 @Override
707 protected boolean socketIsEmpty(int flags) {
708 return IoUring.isCqeFSockNonEmptySupported() && (flags & Native.IORING_CQE_F_SOCK_NONEMPTY) == 0;
709 }
710
711 @Override
712 boolean isPollInFirst() {
713 return false;
714 }
715 }