io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1 when using ByteBufPayload

[rasentry]

Actual Behavior

Provided code throws IllegalReferenceCountException when I use ByteBufPayload and publish result on different thread.
There is no exception when I use DefaultPayload.create("World".getBytes()) insted of ByteBufPayload.

Steps to Reproduce

    @Test
    public void refErrorTest() {
        CloseableChannel server = RSocketServer.create((setup, sendingRSocket) -> Mono.just(new RSocket() {
            @Override
            public Mono<Payload> requestResponse(Payload payload) {
                return Mono.just(DefaultPayload.create("Hello, " + payload.getDataUtf8()));
            }
        }))
                .bind(TcpServerTransport.create(TcpServer.create().host("localhost").port(7890)))
                .block();

        RSocket client = RSocketConnector.create()
                .payloadDecoder(PayloadDecoder.ZERO_COPY)
                .connect(() -> TcpClientTransport.create(TcpClient.create().host("localhost").port(7890)))
                .block();

        var mrrMono = client.requestResponse(ByteBufPayload.create(ByteBuffer.wrap("World".getBytes())))
                .publishOn(Schedulers.parallel())//TODO if remove this line code works as expected
                .map(payload -> {
                    var dataBuf = payload.sliceData();
                    payload.release();
                    return dataBuf.toString(StandardCharsets.UTF_8);
                })
                .flatMap(d -> Mono.error(new IllegalArgumentException("oops")));

        StepVerifier.create(mrrMono)
                .verifyError();

        client.dispose();
        server.dispose();
    }

Your Environment

• RSocket version(s) used: 1.0.1
• Other relevant libraries versions (eg. netty, ...): 4.1.51.Final
• Platform (eg. JVM version (javar -version): java 11.0.3 2019-04-16 LTS
• OS and version (eg uname -a): windows 10

[OlegDokuka]

Hi @rasentry!

Thanks for asking that!

The first problem here is that DefaultPayload and ByteBufPayload has different refcount management. The DefayltPayload is an unpooled payload, so calling the release method makes no effect on it. In contrast, calling the ByteBufPayload#release method releases all resources used by this Payload.

It means that if there are data and metadata, these content will not be available anymore after calling the release on the ByteBufPayload.

That said, if you sliced data and then called release, the slice will be released as well and the content of the slice will not be accessible anymore.

var dataBuf = payload.sliceData();
payload.release(); // after this dataBuf is freed as well
return dataBuf.toString(StandardCharsets.UTF_8); // here you observe error

To fix that you may defer the release to the point when content is read. For example:

var dataBuf = payload.sliceData();
var content = dataBuf.toString(StandardCharsets.UTF_8);

payload.release();
                    
return content;

The above should fix the problem.

Let me know if you experience something different than that,

Cheers,
Oleh

[rasentry]

Thank you for your attention.
I have tried your suggestion but unfortunately id doesn't work, there is still the same exception.
I want you to notice the line

.publishOn(Schedulers.parallel())//TODO if remove this line code works as expected

if I remove that line error disappear

[OlegDokuka]

Indeed. This is a bug. Thank you for finding that! Appreciate this!

[OlegDokuka]

fixed in #894 should be available started from 1.0.2