IoUringIoHandler Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringIoHandler.java lines 53–704

public final class IoUringIoHandler implements IoHandler {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(IoUringIoHandler.class);

    private final RingBuffer ringBuffer;
    private final IntObjectMap<IoUringBufferRing> registeredIoUringBufferRing;
    private final IntObjectMap<DefaultIoUringIoRegistration> registrations;
    // The maximum number of bytes for an InetAddress / Inet6Address
    private final byte[] inet4AddressArray = new byte[SockaddrIn.IPV4_ADDRESS_LENGTH];
    private final byte[] inet6AddressArray = new byte[SockaddrIn.IPV6_ADDRESS_LENGTH];

    private final AtomicBoolean eventfdAsyncNotify = new AtomicBoolean();
    private final FileDescriptor eventfd;
    private final CleanableDirectBuffer eventfdReadBufCleanable;
    private final ByteBuffer eventfdReadBuf;
    private final long eventfdReadBufAddress;
    private final CleanableDirectBuffer timeoutMemoryCleanable;
    private final ByteBuffer timeoutMemory;
    private final long timeoutMemoryAddress;
    private final IovArray iovArray;
    private final MsgHdrMemoryArray msgHdrMemoryArray;
    private long eventfdReadSubmitted;
    private boolean eventFdClosing;
    private volatile boolean shuttingDown;
    private boolean closeCompleted;
    private int nextRegistrationId = Integer.MIN_VALUE;

    // these two ids are used internally any so can't be used by nextRegistrationId().
    private static final int EVENTFD_ID = Integer.MAX_VALUE;
    private static final int RINGFD_ID = EVENTFD_ID - 1;
    private static final int INVALID_ID = 0;

    private static final int KERNEL_TIMESPEC_SIZE = 16; //__kernel_timespec

    private static final int KERNEL_TIMESPEC_TV_SEC_FIELD = 0;
    private static final int KERNEL_TIMESPEC_TV_NSEC_FIELD = 8;

    private final ThreadAwareExecutor executor;

    IoUringIoHandler(ThreadAwareExecutor executor, IoUringIoHandlerConfig config) {
        // Ensure that we load all native bits as otherwise it may fail when try to use native methods in IovArray
        IoUring.ensureAvailability();
        this.executor = requireNonNull(executor, "executor");
        requireNonNull(config, "config");
        int setupFlags = Native.setupFlags(config.singleIssuer());

        //The default cq size is always twice the ringSize.
        // It only makes sense when the user actually specifies the cq ring size.
        int cqSize = 2 * config.getRingSize();
        if (config.needSetupCqeSize()) {
            assert IoUring.isSetupCqeSizeSupported();
            setupFlags |= Native.IORING_SETUP_CQSIZE;
            cqSize = config.getCqSize();
        }
        this.ringBuffer = Native.createRingBuffer(config.getRingSize(), cqSize, setupFlags);
        if (IoUring.isRegisterIowqMaxWorkersSupported() && config.needRegisterIowqMaxWorker()) {
            int maxBoundedWorker = Math.max(config.getMaxBoundedWorker(), 0);
            int maxUnboundedWorker = Math.max(config.getMaxUnboundedWorker(), 0);
            int result = Native.ioUringRegisterIoWqMaxWorkers(ringBuffer.fd(), maxBoundedWorker, maxUnboundedWorker);
            if (result < 0) {
                // Close ringBuffer before throwing to ensure we release all memory on failure.
                ringBuffer.close();
                throw new UncheckedIOException(Errors.newIOException("io_uring_register", result));
            }
        }

        registeredIoUringBufferRing = new IntObjectHashMap<>();
        Collection<IoUringBufferRingConfig> bufferRingConfigs = config.getInternBufferRingConfigs();
        if (bufferRingConfigs != null && !bufferRingConfigs.isEmpty()) {
            for (IoUringBufferRingConfig bufferRingConfig : bufferRingConfigs) {
                try {
                    IoUringBufferRing ring = newBufferRing(ringBuffer.fd(), bufferRingConfig);
                    registeredIoUringBufferRing.put(bufferRingConfig.bufferGroupId(), ring);
                } catch (Errors.NativeIoException e) {
                    for (IoUringBufferRing bufferRing : registeredIoUringBufferRing.values()) {
                        bufferRing.close();
                    }
                    // Close ringBuffer before throwing to ensure we release all memory on failure.
                    ringBuffer.close();
                    throw new UncheckedIOException(e);
                }
            }