Repository: netty/netty
Issue: HttpObjectAggregator could possibly leak in some conditions

Hi,

I am getting messages about buffer leaks in netty server very rarely, like once in 4-6 hours. And it looks like from advanced logging info that HttpObjectAggregator could not release buffer sometimes. My assumption - it could be related to closing idle connection while HttpObjectAggregator has something cached inside and HttpObjectAggregator doesn't release it correctly.

This is how my pipeline looks like:

```
        p.addLast("idleState", new IdleStateHandler(0, 0, idleChannelTimeOutSec));
        p.addLast("httpCodec", new HttpServerCodec());
        p.addLast("unGzip", new HttpContentDecompressor());
        p.addLast("httpAggregator", new HttpObjectAggregator(64 * 1024));
        p.addLast("handler", mainHandler);
```

mainHandler has

```
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            ...
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
```

This is how cleaned up version of log looks like:

```
[03 Aug 2016 18:44:55,952][ERROR][ResourceLeakDetector,httpServerWorker-4-1] - LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 13
#13:
    io.netty.buffer.AdvancedLeakAwareByteBuf.release(AdvancedLeakAwareByteBuf.java:61)
    io.netty.handler.codec.http.DefaultHttpContent.release(DefaultHttpContent.java:72)
    io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:59)
    io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:90)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:435)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:293)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:267)
    io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:250)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1279)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:889)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:572)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:513)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:427)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:399)
    io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:136)
    io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:145)
    java.lang.Thread.run(Thread.java:745)

#10:
    io.netty.buffer.AdvancedLeakAwareByteBuf.retain(AdvancedLeakAwareByteBuf.java:731)
    io.netty.handler.codec.http.DefaultHttpContent.retain(DefaultHttpContent.java:60)
    io.netty.handler.codec.http.HttpObjectAggregator.decode(HttpObjectAggregator.java:225)
    io.netty.handler.codec.http.HttpObjectAggregator.decode(HttpObjectAggregator.java:57)
    io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:88)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:435)
    io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:293)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:267)
    io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:250)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1279)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:889)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:572)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:513)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:427)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:399)
    io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:136)
    io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:145)
    java.lang.Thread.run(Thread.java:745)
#9:
    io.netty.buffer.AdvancedLeakAwareByteBuf.release(AdvancedLeakAwareByteBuf.java:61)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:256)
    io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:250)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1279)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:889)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:572)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:513)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:427)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:399)
    io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:136)
    io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:145)
    java.lang.Thread.run(Thread.java:745)
#8:
    io.netty.buffer.AdvancedLeakAwareByteBuf.retain(AdvancedLeakAwareByteBuf.java:731)
    io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:347)
    io.netty.handler.codec.http.HttpClientCodec$Decoder.decode(HttpClientCodec.java:152)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
    io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:250)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1279)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:889)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:572)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:513)
    io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:427)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:399)
    io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:136)
    io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:145)
    java.lang.Thread.run(Thread.java:745)
```

Full advanced logs: [log1.txt](https://github.com/netty/netty/files/401905/log1.txt) [log2.txt](https://github.com/netty/netty/files/401904/log2.txt).

If this info would not be enough for troubleshooting I would try to write a handler that will analyse/log this situation more deeply when I get a chance.
Or, please, let me know if it looks like a mistake in user code, but for me it doesn't.

Netty version 4.0.38


Issue comments:
thanks for reporting @Spikhalskiy ! will look soon

@Spikhalskiy - do you get any requests of type 100? also do you get any requests that exceed the maximum content length? are you able to reproduce the conditions that cause the leak?

> do you get any requests of type 100?

Yes, we have some % of requests of type "100 Continue".

> do you get any requests that exceed the maximum content length?

No, we don't get TooLongFrameException in mainHandler.

> are you able to reproduce the conditions that cause the leak?

No, I don't know which specific request or condition causing this leak. If it's hard to get into this problem with already provided info I would think about a way of collecting more data.

BTW, more info, that could be related. We had a bug that we returned

```
if (HttpHeaders.is100ContinueExpected(request)) {
    send100Continue(ctx);
}
```

in the mainHandler AFTER HttpObjectAggregator, which already does this. And looks like our clients failed after receiving this second "continue". After fixing this bug (removing this code from mainHandler) I didn't see leaks anymore, so it looks like something possibly related, or just by chance and I will see leak soon (this bug fixed just yesterday).

On the same time, I don't get how it could be related because we sent this second continue in mainHandler, which have release() inside channelRead and we should see this release in advanced logs, but we don't. So, it shouldn't be that cases when we send this second continue.

Just a stream of thoughts, that probably could be useful for you.

The `HttpObjectAggregator` owns the responsibility to send a 100 continue response (because it aggregates and knows the limit on how much data you want to accept). Your peer may get confused if you send 2 continue responses. I will throw together a PR which tightens up the release related stuff in `HttpObjectAggregator` and ping you when ready.

@Scottmitch yep, I already got it and bug with double continue fixed now in our server. Thanks for your work, waiting.

Closed by https://github.com/netty/netty/pull/5676


Pull request review comments:
This rework doesn't make too much sense because this invocation will always fall to code

```
    private void AbstractChannelHandlerContext#invokeChannelInactive() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelInactive(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelInactive();
        }
    }
```

So, looks like the current call `invokeChannelInactive` here is safe in terms of throwing exceptions.

super#handlerRemoved is empty, so should be safe without `try {} catch` too.

practically there should be no difference, but there is no need to make assumptions about what the super class does.


Patch that resolved the issue:
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectAggregator.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectAggregator.java
index bebe52a3995..2a530bbe8fb 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectAggregator.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectAggregator.java
@@ -99,7 +99,6 @@ public class HttpObjectAggregator extends MessageToMessageDecoder<HttpObject> {
 
     private final int maxContentLength;
     private AggregatedFullHttpMessage currentMessage;
-    private boolean tooLongFrameFound;
     private final boolean closeOnExpectationFailed;
 
     private int maxCumulationBufferComponents = DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS;
@@ -168,18 +167,17 @@ public final void setMaxCumulationBufferComponents(int maxCumulationBufferCompon
 
     @Override
     protected void decode(final ChannelHandlerContext ctx, HttpObject msg, List<Object> out) throws Exception {
-        AggregatedFullHttpMessage currentMessage = this.currentMessage;
-
         if (msg instanceof HttpMessage) {
-            tooLongFrameFound = false;
-            assert currentMessage == null;
-
+            if (currentMessage != null) {
+                currentMessage.release();
+                currentMessage = null;
+                throw new IllegalStateException("Start of new message received before existing message completed.");
+            }
             HttpMessage m = (HttpMessage) msg;
 
             // Handle the 'Expect: 100-continue' header if necessary.
             if (is100ContinueExpected(m)) {
                 if (HttpHeaders.getContentLength(m, 0) > maxContentLength) {
-                    tooLongFrameFound = true;
                     final ChannelFuture future = ctx.writeAndFlush(EXPECTATION_FAILED.duplicate().retain());
                     future.addListener(new ChannelFutureListener() {
                         @Override
@@ -208,18 +206,16 @@ public void operationComplete(ChannelFuture future) throws Exception {
             if (!m.getDecoderResult().isSuccess()) {
                 removeTransferEncodingChunked(m);
                 out.add(toFullMessage(m));
-                this.currentMessage = null;
                 return;
             }
             if (msg instanceof HttpRequest) {
                 HttpRequest header = (HttpRequest) msg;
-                this.currentMessage = currentMessage = new AggregatedFullHttpRequest(
+                currentMessage = new AggregatedFullHttpRequest(
                         header, ctx.alloc().compositeBuffer(maxCumulationBufferComponents), null);
             } else if (msg instanceof HttpResponse) {
                 HttpResponse header = (HttpResponse) msg;
-                this.currentMessage = currentMessage = new AggregatedFullHttpResponse(
-                        header,
-                        Unpooled.compositeBuffer(maxCumulationBufferComponents), null);
+                currentMessage = new AggregatedFullHttpResponse(
+                        header, ctx.alloc().compositeBuffer(maxCumulationBufferComponents), null);
             } else {
                 throw new Error();
             }
@@ -227,25 +223,20 @@ public void operationComplete(ChannelFuture future) throws Exception {
             // A streamed message - initialize the cumulative buffer, and wait for incoming chunks.
             removeTransferEncodingChunked(currentMessage);
         } else if (msg instanceof HttpContent) {
-            if (tooLongFrameFound) {
-                if (msg instanceof LastHttpContent) {
-                    this.currentMessage = null;
-                }
-                // already detect the too long frame so just discard the content
+            if (currentMessage == null) {
+                // it is possible that a TooLongFrameException was already thrown but we can still discard data
+                // until the begging of the next request/response.
                 return;
             }
-            assert currentMessage != null;
 
             // Merge the received chunk into the content of the current message.
             HttpContent chunk = (HttpContent) msg;
             CompositeByteBuf content = (CompositeByteBuf) currentMessage.content();
 
             if (content.readableBytes() > maxContentLength - chunk.content().readableBytes()) {
-                tooLongFrameFound = true;
-
                 // release current message to prevent leaks
                 currentMessage.release();
-                this.currentMessage = null;
+                currentMessage = null;
 
                 throw new TooLongFrameException(
                         "HTTP content length exceeded " + maxContentLength +
@@ -254,9 +245,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
 
             // Append the content of the chunk
             if (chunk.content().isReadable()) {
-                chunk.retain();
-                content.addComponent(chunk.content());
-                content.writerIndex(content.writerIndex() + chunk.content().readableBytes());
+                content.addComponent(true, chunk.content().retain());
             }
 
             final boolean last;
@@ -269,8 +258,6 @@ public void operationComplete(ChannelFuture future) throws Exception {
             }
 
             if (last) {
-                this.currentMessage = null;
-
                 // Merge trailing headers into the message.
                 if (chunk instanceof LastHttpContent) {
                     LastHttpContent trailer = (LastHttpContent) chunk;
@@ -290,7 +277,9 @@ public void operationComplete(ChannelFuture future) throws Exception {
                             Names.CONTENT_LENGTH,
                             String.valueOf(content.readableBytes()));
                 }
-                // All done
+                // Set our currentMessage member variable to null in case adding to out will cause re-entry.
+                AggregatedFullHttpMessage currentMessage = this.currentMessage;
+                this.currentMessage = null;
                 out.add(currentMessage);
             }
         } else {
@@ -300,12 +289,11 @@ public void operationComplete(ChannelFuture future) throws Exception {
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        super.channelInactive(ctx);
-
-        // release current message if it is not null as it may be a left-over
-        if (currentMessage != null) {
-            currentMessage.release();
-            currentMessage = null;
+        try {
+            super.channelInactive(ctx);
+        } finally {
+            // release current message if it is not null as it may be a left-over
+            releaseCurrentMessage();
         }
     }
 
@@ -316,9 +304,16 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
-        super.handlerRemoved(ctx);
-        // release current message if it is not null as it may be a left-over as there is not much more we can do in
-        // this case
+        try {
+            super.handlerRemoved(ctx);
+        } finally {
+            // release current message if it is not null as it may be a left-over as there is not much more we can do in
+            // this case
+            releaseCurrentMessage();
+        }
+    }
+
+    private void releaseCurrentMessage() {
         if (currentMessage != null) {
             currentMessage.release();
             currentMessage = null;


Test patch: