Is flush() necessary for releasing pooled buffers?

[BaurzhanSakhariev]

Hi, I'm investigating a buffer leak and was wondering if I understood some parts
of the https://netty.io/wiki/new-and-noteworthy-in-4.0.html correctly:

Afaik, Netty takes care of releasing buffers written to a channel.
But I'm not sure if flush() is needed to ensure such behavior.

From https://netty.io/wiki/new-and-noteworthy-in-4.0.html#write-does-not-flush-automatically

write() operation does not flush automatically

you must be very careful not to forget to call ctx.flush() after writing something.

From https://netty.io/wiki/new-and-noteworthy-in-4.0.html#pooled-buffers

Once a ByteBuf is written to the remote peer it will be returned automatically to the pool it originated from.

When an outbound (a.k.a. downstream) message reaches at the beginning of the pipeline, Netty will release it after writing it out.

This reads to me like we need to actually send data to release the message and return ByteBuf back to the pool, is it correct?

I.e, I have a handler that is using a pooled buffer and don't use neither write() + flush() nor writeAndFlush()

ByteBuf buffer = channel.alloc().buffer(5);
buffer.writeByte(1);
buffer.writeInt(4);
channel.write(buffer);

and other outbound handlers in the pipeline also use plain write() without any flushing. My current thinking is that channel.write has to be replaced with channel.writeAndFlush.

Leak report:

Expecting empty but was: ["LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.buffer.AdvancedLeakAwareByteBuf.writeInt(AdvancedLeakAwareByteBuf.java:563)
	...
#2:
	io.netty.buffer.AdvancedLeakAwareByteBuf.writeByte(AdvancedLeakAwareByteBuf.java:545)
	...
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:410)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
	io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:116)

[yawkat]

Pretty sure it's not necessary, a normal write is enough. Your issue is elsewhere.

[BaurzhanSakhariev]

Posting the whole leak report

Expecting empty but was: ["LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.buffer.AdvancedLeakAwareByteBuf.writeInt(AdvancedLeakAwareByteBuf.java:563)
	io.crate.protocols.postgres.Messages.sendShortMsg(Messages.java:513)
	io.crate.protocols.postgres.Messages.sendBindComplete(Messages.java:487)
	io.crate.protocols.postgres.PostgresWireProtocol.handleBindMessage(PostgresWireProtocol.java:645)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchMessage(PostgresWireProtocol.java:348)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchState(PostgresWireProtocol.java:329)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:297)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:281)
	io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:333)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:455)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
	io.crate.protocols.postgres.PgDecoder.channelRead(PgDecoder.java:116)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
	org.elasticsearch.transport.netty4.Netty4InboundStatsHandler.channelRead(Netty4InboundStatsHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	java.base/java.lang.Thread.run(Thread.java:1575)
#2:
	io.netty.buffer.AdvancedLeakAwareByteBuf.writeByte(AdvancedLeakAwareByteBuf.java:545)
	io.crate.protocols.postgres.Messages.sendShortMsg(Messages.java:512)
	io.crate.protocols.postgres.Messages.sendBindComplete(Messages.java:487)
	io.crate.protocols.postgres.PostgresWireProtocol.handleBindMessage(PostgresWireProtocol.java:645)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchMessage(PostgresWireProtocol.java:348)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchState(PostgresWireProtocol.java:329)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:297)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:281)
	io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:333)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:455)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
	io.crate.protocols.postgres.PgDecoder.channelRead(PgDecoder.java:116)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
	org.elasticsearch.transport.netty4.Netty4InboundStatsHandler.channelRead(Netty4InboundStatsHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	java.base/java.lang.Thread.run(Thread.java:1575)
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:410)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
	io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:116)
	io.crate.protocols.postgres.Messages.sendShortMsg(Messages.java:511)
	io.crate.protocols.postgres.Messages.sendBindComplete(Messages.java:487)
	io.crate.protocols.postgres.PostgresWireProtocol.handleBindMessage(PostgresWireProtocol.java:645)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchMessage(PostgresWireProtocol.java:348)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.dispatchState(PostgresWireProtocol.java:329)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:297)
	io.crate.protocols.postgres.PostgresWireProtocol$MessageHandler.channelRead0(PostgresWireProtocol.java:281)
	io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:333)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:455)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
	io.crate.protocols.postgres.PgDecoder.channelRead(PgDecoder.java:116)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
	org.elasticsearch.transport.netty4.Netty4InboundStatsHandler.channelRead(Netty4InboundStatsHandler.java:99)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)

Method appearing in the leak report (it's part of an InboundHandler that sends back responses)

private static ChannelFuture sendShortMsg(Channel channel, char msgType, final String traceLogMsg) {
        ByteBuf buffer = channel.alloc().buffer(5);
        buffer.writeByte(msgType);
        buffer.writeInt(4);

        ChannelFuture channelFuture = channel.write(buffer);
        if (LOGGER.isTraceEnabled()) {
            channelFuture.addListener(_ -> LOGGER.trace(traceLogMsg));
        }
        return channelFuture;
    }

The only other outbound handler (comes before a message handler, ie last handler to process outbound messages):

SomeHandler extends ChannelOutboundHandlerAdapter...

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof ByteBuf bb) {
        statsTracker.incrementBytesSent(bb.readableBytes());
    } else {
        logger.warn("Message received is: {} and not a ByteBuf, cannot track sent bytes or message count",
            msg.getClass().getCanonicalName());
    }
    ctx.write(msg, promise);
}

[BaurzhanSakhariev]

which means it either has been flushed already or the channel has been closed

Thanks for the pointer.
I didn't find in docs how pending writes are processed on a channel closure, maybe suddenly dropped connection in combination with an unflushed buffers leaks it.

[BaurzhanSakhariev]

A follow up question here - afaik,channel.close() fails all pending writes.

I have an exceptionCaught override that closes app specific resource and closes a Channel.

Should I use write-empty-buffer-and-close-when-done idiom to ensure that we close only when all writes are done?
See #9862, #9772 (comment) and #13072

I mean replacing ctx.channel().close() with ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);

I understand that failing pending writes basically takes care of releasing buffers (?) so I think channel.close is good enough but maybe I'm missing something.

 @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
    ... close app specific resources
      ctx.channel().close()  -> fails all pending writes so no leak here?
      // ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); 
  }

Thanks!

[BaurzhanSakhariev]

Turns out we faced some regression on 4.1.125: our tests were flaky on 4.1.116, upgrade to 4.1.124 helped to get rid of buf-leak flakiness (so issue resolved somewhere in between) and then with 4.1.125 tests again became flaky.

Not sure which change in 4.1.125 is causing troubles, many 4.1.125 commits are about HTTP, irrelevant for this particular leak report.

Closing this, we will stick to version 4.1.124 for now.
crate/crate#18340 (comment)