testProviderBufferRead() — netty Function Reference

Source Code

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

    @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) {
                        bufferSyncer.offer((ByteBuf) msg);
                    }

                    @Override
                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                        if (evt instanceof IoUringBufferRingExhaustedEvent) {
                            eventSyncer.add((IoUringBufferRingExhaustedEvent) evt);
                        }
                    }
                })
                .childOption(IoUringChannelOption.IO_URING_BUFFER_GROUP_ID, bufferRingConfig.bufferGroupId())
                .bind(NetUtil.LOCALHOST, 0)
                .syncUninterruptibly().channel();

        Bootstrap clientBoostrap = new Bootstrap();
        clientBoostrap.group(group)
                .channel(IoUringSocketChannel.class)
                .handler(new ChannelInboundHandlerAdapter());
        ChannelFuture channelFuture = clientBoostrap.connect(serverChannel.localAddress()).syncUninterruptibly();
        assumeTrue(channelFuture.isSuccess());
        Channel clientChannel = channelFuture.channel();

        //is provider buffer read?
        ByteBuf writeBuffer = Unpooled.directBuffer(randomStringLength);
        ByteBufUtil.writeAscii(writeBuffer, randomString);
        ByteBuf userspaceIoUringBufferElement1 = sendAndRecvMessage(clientChannel, writeBuffer, bufferSyncer);
        ByteBuf userspaceIoUringBufferElement2 = sendAndRecvMessage(clientChannel, writeBuffer, bufferSyncer);
        ByteBuf readBuffer = sendAndRecvMessage(clientChannel, writeBuffer, bufferSyncer);
        readBuffer.release();

        // Now we release the buffer and so put it back into the buffer ring.
        userspaceIoUringBufferElement1.release();
        userspaceIoUringBufferElement2.release();

        readBuffer = sendAndRecvMessage(clientChannel, writeBuffer, bufferSyncer);
        readBuffer.release();

        // The next buffer is expected to be provided out of the ring again.