testProviderBufferReadWithRecvsendBundle() — netty Function Reference

Source Code

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

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    @EnabledIf("recvsendBundleEnabled")
    public void testProviderBufferReadWithRecvsendBundle(boolean incremental) throws InterruptedException {
        // See https://lore.kernel.org/io-uring/184f9f92-a682-4205-a15d-89e18f664502@kernel.dk/T/#u
        assumeTrue(IoUring.isRecvMultishotEnabled(),
                "Only yields expected test results when using multishot atm");
        if (incremental) {
            assumeTrue(IoUring.isRegisterBufferRingIncSupported());
        }
        int bufferRingChunkSize = 8;
        IoUringIoHandlerConfig ioUringIoHandlerConfiguration = new IoUringIoHandlerConfig();
        IoUringBufferRingConfig bufferRingConfig = new IoUringBufferRingConfig(
                // let's use a small chunkSize so we are sure a recv will span multiple buffers.
                (short) 1, (short) 16, 8, 16 * 16,
                incremental, new IoUringFixedBufferRingAllocator(bufferRingChunkSize));

        ioUringIoHandlerConfiguration.setBufferRingConfig(bufferRingConfig);

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

        final BlockingQueue<ByteBuf> buffers = new LinkedBlockingQueue<>();
        Channel serverChannel = serverBootstrap.group(group)
                .childHandler(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) {
                        buffers.offer((ByteBuf) msg);
                    }
                })
                .childOption(IoUringChannelOption.IO_URING_BUFFER_GROUP_ID, (short) 1)
                .bind(new InetSocketAddress(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();

        // Create a buffer that will span multiple buffers that are used out of the buffer ring.
        ByteBuf writeBuffer = Unpooled.directBuffer(bufferRingChunkSize * 16);
        CompositeByteBuf received = Unpooled.compositeBuffer();
        try {
            // Fill the buffer with something so we can assert if the received bytes are the same.
            for (int i = 0; i < writeBuffer.capacity(); i++) {
                writeBuffer.writeByte((byte) i);
            }
            clientChannel.writeAndFlush(writeBuffer.retainedDuplicate()).syncUninterruptibly();

            // Aggregate all received buffers until we received everything.
            do {
                ByteBuf buffer = buffers.take();
                received.addComponent(true, buffer);
            } while (received.readableBytes() != writeBuffer.readableBytes());

            assertEquals(writeBuffer, received);
            serverChannel.close().syncUninterruptibly();
            clientChannel.close().syncUninterruptibly();
            group.shutdownGracefully();
            assertTrue(buffers.isEmpty());
        } finally {
            writeBuffer.release();
            received.release();
        }
    }