testBufferLifecycleCorrectlyHandled() — netty Function Reference

Source Code

transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringSocketSendSzSendmsgZcTest.java lines 126–225

    private static void testBufferLifecycleCorrectlyHandled(Bootstrap cb, boolean multiple, Close remoteClose)
            throws Throwable {
        cb.handler(new ChannelInboundHandlerAdapter());
        // Force to use send_zc / sendmsg_zc if supported.
        cb.option(IoUringChannelOption.IO_URING_WRITE_ZERO_COPY_THRESHOLD, 0);
        if (remoteClose == Close.LOCAL) {
            // Configure TCP_USER_TIMEOUT to a small number so the buffers can be returned quickly.
            // See:
            // - https://man7.org/linux/man-pages/man7/tcp.7.html
            // - https://github.com/torvalds/linux/blob/v6.16/include/uapi/linux/tcp.h#L111
            cb.option(new IntegerUnixChannelOption("TCP_USER_TIMEOUT", 6, 18), 1000);
        }

        try (ServerSocket serverSocket = new ServerSocket()) {
            serverSocket.bind(new InetSocketAddress(0));
            ChannelFuture future = cb.connect(serverSocket.getLocalSocketAddress());
            final AtomicReference<Throwable> causeRef = new AtomicReference<>();

            try (Socket socket = serverSocket.accept()) {
                // We accept the socket but don't read data, this way we will not receive the second notification
                // for the send as we never see a TCP ack until we start reading.
                Channel channel = future.sync().channel();
                try {
                    final int numBuffers = multiple ? 2: 1;
                    CountDownLatch latch = new CountDownLatch(numBuffers);
                    int bufferSize = 1024 * 1024;
                    final ByteBuf buffer = channel.alloc().buffer(bufferSize);
                    future.addListener(f -> {
                        if (f.isSuccess()) {
                            ChannelFutureListener writeListener = f2 -> {
                                if (!f2.isSuccess()) {
                                    causeRef.compareAndSet(null, f2.cause());
                                }
                                latch.countDown();
                            };

                            buffer.writerIndex(buffer.capacity());
                            if (multiple) {
                                channel.write(buffer.readRetainedSlice(buffer.readableBytes() / 2))
                                        .addListener(writeListener);
                            }
                            channel.writeAndFlush(buffer)
                                    .addListener(writeListener);
                        } else {
                            buffer.release();
                            causeRef.set(f.cause());

                            for (int i = 0; i < numBuffers; i++) {
                                latch.countDown();
                            }
                        }
                    });
                    latch.await();
                    Throwable cause = causeRef.get();
                    if (cause != null) {
                        fail(cause);
                    }
                    // The buffer should still have a reference count of 1 as we did not receive the second notification
                    // yet as the remote peer did not start reading and did not close the socket yet.
                    if (multiple) {
                        assertEquals(numBuffers, buffer.refCnt());
                    } else {
                        assertEquals(numBuffers, buffer.refCnt());
                    }

                    switch (remoteClose) {
                        case REMOTE:
                            // Don't read any data but just close the socket. This should trigger the required
                            // notifications to release the buffers.
                            socket.close();
                            break;
                        case LOCAL:
                            // Don't read any data but just close the channel. Once we did not see an ack for the
                            // configured TCP_USER_TIMEOUT we will get the required notifications to release the buffers
                            channel.close().sync();
                            break;
                        case NONE:
                            // Let's read the bytes now so the buffer can be released again from the NIC.
                            try (InputStream stream = socket.getInputStream()) {
                                byte[] bytes = new byte[64 * 1024];
                                int r;