IoUringBufferRingTest Class — netty Architecture

Source Code

transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringBufferRingTest.java lines 47–300

public class IoUringBufferRingTest {
    @BeforeAll
    public static void loadJNI() {
        assumeTrue(IoUring.isAvailable());
        assumeTrue(IoUring.isRegisterBufferRingSupported());
    }

    @Test
    public void testRegister() {
        // using cqeSize on purpose NOT a power of 2
        RingBuffer ringBuffer = Native.createRingBuffer(8, 15, 0);
        try {
            int ringFd = ringBuffer.fd();
            long ioUringBufRingAddr = Native.ioUringRegisterBufRing(ringFd, 4, (short) 1, 0);
            assumeTrue(
                    ioUringBufRingAddr > 0,
                    "ioUringSetupBufRing result must great than 0, but now result is " + ioUringBufRingAddr);
            int freeRes = Native.ioUringUnRegisterBufRing(ringFd, ioUringBufRingAddr, 4, (short) 1);
            assertEquals(
                    0,
                    freeRes,
                    "ioUringFreeBufRing result must be 0, but now result is " + freeRes
            );
            // let io_uring to "fix" it
            assertEquals(16, ringBuffer.ioUringCompletionQueue().ringCapacity);
        } finally {
            ringBuffer.close();
        }
    }

    private static ByteBuf unwrapLeakAware(ByteBuf buf) {
        // If its a sub-type of WrappedByteBuf we know its because it was wrapped for leak-detection.
        if (buf instanceof WrappedByteBuf) {
            return buf.unwrap();
        }
        return buf;
    }

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testProviderBufferRead(boolean incremental) throws InterruptedException {
        if (incremental) {
            assumeTrue(IoUring.isRegisterBufferRingIncSupported());
        }
        final BlockingQueue<ByteBuf> bufferSyncer = new LinkedBlockingQueue<>();
        IoUringIoHandlerConfig ioUringIoHandlerConfiguration = new IoUringIoHandlerConfig();
        IoUringBufferRingConfig bufferRingConfig =
                IoUringBufferRingConfig.builder()
                        .bufferGroupId((short) 1)
                        .bufferRingSize((short) 2)
                        .batchSize(2).incremental(incremental)
                        .allocator(new IoUringFixedBufferRingAllocator(1024))
                        .batchAllocation(false)
                        .build();

        IoUringBufferRingConfig bufferRingConfig1 =
                IoUringBufferRingConfig.builder()
                        .bufferGroupId((short) 2)
                        .bufferRingSize((short) 16)
                        .batchSize(8)
                        .incremental(incremental)
                        .allocator(new IoUringFixedBufferRingAllocator(1024))
                        .batchAllocation(true)
                        .build();
        ioUringIoHandlerConfiguration.setBufferRingConfig(bufferRingConfig, bufferRingConfig1);

        MultiThreadIoEventLoopGroup group = new MultiThreadIoEventLoopGroup(1,
                IoUringIoHandler.newFactory(ioUringIoHandlerConfiguration)
        );
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.channel(IoUringServerSocketChannel.class);

        String randomString = UUID.randomUUID().toString();
        int randomStringLength = randomString.length();

        ArrayBlockingQueue<IoUringBufferRingExhaustedEvent> eventSyncer = new ArrayBlockingQueue<>(1);

        Channel serverChannel = serverBootstrap.group(group)
                .childHandler(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) {