EpollDatagramChannelUnsafe Class — netty Architecture

Source Code

transport-classes-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java lines 542–613

    final class EpollDatagramChannelUnsafe extends AbstractEpollUnsafe {

        @Override
        void epollInReady() {
            assert eventLoop().inEventLoop();
            EpollDatagramChannelConfig config = config();
            if (shouldBreakEpollInReady(config)) {
                clearEpollIn0();
                return;
            }
            final EpollRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            allocHandle.reset(config);

            Throwable exception = null;
            try {
                try {
                    boolean connected = isConnected();
                    do {
                        final boolean read;
                        int datagramSize = config().getMaxDatagramPayloadSize();

                        ByteBuf byteBuf = allocHandle.allocate(allocator);
                        // Only try to use recvmmsg if its really supported by the running system.
                        int numDatagram = Native.IS_SUPPORTING_RECVMMSG ?
                                datagramSize == 0 ? 1 : byteBuf.writableBytes() / datagramSize :
                                0;
                        try {
                            if (numDatagram <= 1) {
                                if (!connected || config.isUdpGro()) {
                                    read = recvmsg(allocHandle, cleanDatagramPacketArray(), byteBuf);
                                } else {
                                    read = connectedRead(allocHandle, byteBuf, datagramSize);
                                }
                            } else {
                                // Try to use scattering reads via recvmmsg(...) syscall.
                                read = scatteringRead(allocHandle, cleanDatagramPacketArray(),
                                        byteBuf, datagramSize, numDatagram);
                            }
                        } catch (NativeIoException e) {
                            if (connected) {
                                throw translateForConnected(e);
                            }
                            throw e;
                        }

                        if (read) {
                            readPending = false;
                        } else {
                            break;
                        }
                    // We use the TRUE_SUPPLIER as it is also ok to read less then what we did try to read (as long
                    // as we read anything).
                    } while (allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER));
                } catch (Throwable t) {
                    exception = t;
                }

                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();

                if (exception != null) {
                    pipeline.fireExceptionCaught(exception);
                }
            } finally {
                if (shouldStopReading(config)) {
                    clearEpollIn();
                }
            }
        }
    }