Repository: netty/netty
Issue: Reduce the memory consumption of ChannelOutboundBuffer
Labels: improvement

The initial capacity of a `ChannelOutboundBuffer` is 32, and it means each `ChannelOutboundBuffer` will end up with 32 entry objects as a channel lives on, even if the channel's maximum number of pending write requests is always 1.

We could do a few things to fix this:
- Reduce the initial capacity to a smaller value such as 4.
- Recycle entry objects or do something similar (we don't need thread local here obviously.)

http://stackoverflow.com/questions/24726973/netty-4-0-19-final-memory-leak-with-io-netty-channel-channeloutboundbufferentry


Issue comments:
[This build](https://secure.motd.kr/jenkins/job/netty-4.0/4803/) reduces the initial capacity of `ChannelOutboundBuffer` to 4. (commit: 7e362277b93bbf4d59fd80485f3fe534a3ed8bba)

@trustin I think we can close this ?

Nope. The second item was not resolved yet.

Ah got it

> Am 23.07.2014 um 22:44 schrieb Trustin Lee notifications@github.com:
> 
> Nope. The second item was not resolved yet.
> 
> —
> Reply to this email directly or view it on GitHub.

Another thing we could do is to make the DefaultChannelPromise an entry element by itself and just store it. The downside would be that we need to wrap any other ChsnnelPromise impls.

Just thinking out loud...

> Am 23.07.2014 um 22:44 schrieb Trustin Lee notifications@github.com:
> 
> Nope. The second item was not resolved yet.
> 
> —
> Reply to this email directly or view it on GitHub.

I had to switch to Netty 3 to avoid this memory usage problem. I'm not sure what the difference is, but with Netty 3 my program uses 8-16mb of RAM while the Netty 4 version will utilize all that I can afford to give it and then crash.

There must be something else wrong. Coukd you share a reproducer?

> Am 25.07.2014 um 00:58 schrieb Blake Beaupain notifications@github.com:
> 
> I had to switch to Netty 3 to avoid this memory usage problem. I'm not sure what the difference is, but with Netty 3 my program uses 8-16mb of RAM while the Netty 4 version will utilize all that I can afford to give it and then crash.
> 
> —
> Reply to this email directly or view it on GitHub.

@trustin for a fix see https://github.com/netty/netty/pull/2704

I have meet this issue, I try to disable AUTO_READ, and manually trigger read() operation by my self, and this issue disappear, Java heap memory consumption is quite stable.

Pull request review comments:
This line should not be changed.

If this is not public, a user who extends `AbstractChannel` (and consequently `AbstractUnsafe`) will not be able to access the return value of `AbstractUnsafe.outboundBuffer()`.

Any reason this is not part of the interface?


Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/AbstractChannel.java b/transport/src/main/java/io/netty/channel/AbstractChannel.java
index 5642f226554..e270a98acb7 100644
--- a/transport/src/main/java/io/netty/channel/AbstractChannel.java
+++ b/transport/src/main/java/io/netty/channel/AbstractChannel.java
@@ -83,7 +83,7 @@ protected AbstractChannel(Channel parent) {
     @Override
     public boolean isWritable() {
         ChannelOutboundBuffer buf = unsafe.outboundBuffer();
-        return buf != null && buf.getWritable();
+        return buf != null && buf.isWritable();
     }
 
     @Override
@@ -649,7 +649,11 @@ public void write(Object msg, ChannelPromise promise) {
                 ReferenceCountUtil.release(msg);
                 return;
             }
-            outboundBuffer.addMessage(msg, promise);
+            int size = estimatorHandle().size(msg);
+            if (size < 0) {
+                size = 0;
+            }
+            outboundBuffer.addMessage(msg, size, promise);
         }
 
         @Override
diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
index 899ca3b289c..e0508bdb5d7 100644
--- a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
+++ b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
@@ -40,6 +40,8 @@
 /**
  * (Transport implementors only) an internal data structure used by {@link AbstractChannel} to store its pending
  * outbound write requests.
+ *
+ * All the methods should only be called by the {@link EventLoop} of the {@link Channel}.
  */
 public final class ChannelOutboundBuffer {
 
@@ -61,7 +63,7 @@ protected ByteBuffer[] initialValue() throws Exception {
         }
     };
 
-    private final AbstractChannel channel;
+    private final Channel channel;
 
     // Entry(flushedEntry) --> ... Entry(unflushedEntry) --> ... Entry(tailEntry)
     //
@@ -109,12 +111,11 @@ protected ByteBuffer[] initialValue() throws Exception {
         this.channel = channel;
     }
 
-    void addMessage(Object msg, ChannelPromise promise) {
-        int size = channel.estimatorHandle().size(msg);
-        if (size < 0) {
-            size = 0;
-        }
-
+    /**
+     * Add given message to this {@link ChannelOutboundBuffer}. The given {@link ChannelPromise} will be notified once
+     * the message was written.
+     */
+    public void addMessage(Object msg, int size, ChannelPromise promise) {
         Entry entry = Entry.newInstance(msg, size, total(msg), promise);
         if (tailEntry == null) {
             flushedEntry = null;
@@ -133,7 +134,11 @@ void addMessage(Object msg, ChannelPromise promise) {
         incrementPendingOutboundBytes(size);
     }
 
-    void addFlush() {
+    /**
+     * Add a flush to this {@link ChannelOutboundBuffer}. This means all previous added messages are marked as flushed
+     * and so you will be able to handle them.
+     */
+    public void addFlush() {
         // There is no need to process all entries if there was already a flush before and no new messages
         // where added in the meantime.
         //
@@ -206,10 +211,18 @@ private static long total(Object msg) {
         return -1;
     }
 
+    /**
+     * Return the current message to write or {@code null} if nothing was flushed before and so is ready to be written.
+     */
     public Object current() {
         return current(true);
     }
 
+    /**
+     * Return the current message to write or {@code null} if nothing was flushed before and so is ready to be written.
+     * If {@code true} is specified a direct {@link ByteBuf} or {@link ByteBufHolder} is prefered and
+     * so the current message may be copied into a direct buffer.
+     */
     public Object current(boolean preferDirect) {
         // TODO: Think of a smart way to handle ByteBufHolder messages
         Entry entry = flushedEntry;
@@ -250,7 +263,7 @@ public Object current(boolean preferDirect) {
 
     /**
      * Replace the current msg with the given one.
-     * The replaced msg will automatically be released
+     * {@link ReferenceCountUtil#release(Object)} will automatically be called on the replaced message.
      */
     public void current(Object msg) {
         Entry entry = flushedEntry;
@@ -259,6 +272,9 @@ public void current(Object msg) {
         entry.msg = msg;
     }
 
+    /**
+     * Notify the {@link ChannelPromise} of the current message about writing progress.
+     */
     public void progress(long amount) {
         Entry e = flushedEntry;
         assert e != null;
@@ -270,6 +286,11 @@ public void progress(long amount) {
         }
     }
 
+    /**
+     * Will remove the current message, mark its {@link ChannelPromise} as success and return {@code true}. If no
+     * flushed message exists at the time this method is called it will return {@code false} to signal that no more
+     * messages are ready to be handled.
+     */
     public boolean remove() {
         Entry e = flushedEntry;
         if (e == null) {
@@ -295,6 +316,11 @@ public boolean remove() {
         return true;
     }
 
+    /**
+     * Will remove the current message, mark its {@link ChannelPromise} as failure using the given {@link Throwable}
+     * and return {@code true}. If no   flushed message exists at the time this method is called it will return
+     * {@code false} to signal that no more messages are ready to be handled.
+     */
     public boolean remove(Throwable cause) {
         Entry e = flushedEntry;
         if (e == null) {
@@ -366,9 +392,8 @@ public void removeBytes(long writtenBytes) {
 
     /**
      * Returns an array of direct NIO buffers if the currently pending messages are made of {@link ByteBuf} only.
-     * {@code null} is returned otherwise.  If this method returns a non-null array, {@link #nioBufferCount()} and
-     * {@link #nioBufferSize()} will return the number of NIO buffers in the returned array and the total number
-     * of readable bytes of the NIO buffers respectively.
+     * {@link #nioBufferCount()} and {@link #nioBufferSize()} will return the number of NIO buffers in the returned
+     * array and the total number of readable bytes of the NIO buffers respectively.
      * <p>
      * Note that the returned array is reused and thus should not escape
      * {@link AbstractChannel#doWrite(ChannelOutboundBuffer)}.
@@ -479,22 +504,39 @@ private static ByteBuffer[] expandNioBufferArray(ByteBuffer[] array, int neededS
         return newArray;
     }
 
+    /**
+     * Returns the number of {@link ByteBuffer} that can be written out of the {@link ByteBuffer} array that was
+     * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
+     * was called.
+     */
     public int nioBufferCount() {
         return nioBufferCount;
     }
 
+    /**
+     * Returns the number of bytes that can be written out of the {@link ByteBuffer} array that was
+     * obtained via {@link #nioBuffers()}. This method <strong>MUST</strong> be called after {@link #nioBuffers()}
+     * was called.
+     */
     public long nioBufferSize() {
         return nioBufferSize;
     }
 
-    boolean getWritable() {
+    boolean isWritable() {
         return writable != 0;
     }
 
+    /**
+     * Returns the number of flushed messages in this {@link ChannelOutboundBuffer}.
+     */
     public int size() {
         return flushed;
     }
 
+    /**
+     * Returns {@code true} if there are flushed messages in this {@link ChannelOutboundBuffer} or {@code false}
+     * otherwise.
+     */
     public boolean isEmpty() {
         return flushed == 0;
     }


Test patch:
diff --git a/transport/src/test/java/io/netty/channel/ChannelOutboundBufferTest.java b/transport/src/test/java/io/netty/channel/ChannelOutboundBufferTest.java
index 11c0f6a1415..296a0e7d131 100644
--- a/transport/src/test/java/io/netty/channel/ChannelOutboundBufferTest.java
+++ b/transport/src/test/java/io/netty/channel/ChannelOutboundBufferTest.java
@@ -51,7 +51,7 @@ public void testNioBuffersSingleBacked() {
 
         ByteBuf buf = copiedBuffer("buf1", CharsetUtil.US_ASCII);
         ByteBuffer nioBuf = buf.internalNioBuffer(0, buf.readableBytes());
-        buffer.addMessage(buf, channel.voidPromise());
+        buffer.addMessage(buf, buf.readableBytes(), channel.voidPromise());
         assertEquals("Should still be 0 as not flushed yet", 0, buffer.nioBufferCount());
         buffer.addFlush();
         ByteBuffer[] buffers = buffer.nioBuffers();
@@ -75,7 +75,7 @@ public void testNioBuffersExpand() {
 
         ByteBuf buf = directBuffer().writeBytes("buf1".getBytes(CharsetUtil.US_ASCII));
         for (int i = 0; i < 64; i++) {
-            buffer.addMessage(buf.copy(), channel.voidPromise());
+            buffer.addMessage(buf.copy(), buf.readableBytes(), channel.voidPromise());
         }
         assertEquals("Should still be 0 as not flushed yet", 0, buffer.nioBufferCount());
         buffer.addFlush();
@@ -99,7 +99,7 @@ public void testNioBuffersExpand2() {
         for (int i = 0; i < 65; i++) {
             comp.addComponent(buf.copy()).writerIndex(comp.writerIndex() + buf.readableBytes());
         }
-        buffer.addMessage(comp, channel.voidPromise());
+        buffer.addMessage(comp, comp.readableBytes(), channel.voidPromise());
 
         assertEquals("Should still be 0 as not flushed yet", 0, buffer.nioBufferCount());
         buffer.addFlush();