Repository: netty/netty
Issue: DuplicatedByteBuf not correctly expand capacity
Labels: defect

The problem is that it delegate the call to the wrapped buffer which will use its own readerIndex to detect which data needs to get copied while expand it. Thus it leads to dataloss


Issue comments:
@trustin after thinking more about I think we should disallow to expand a duplicated buffer. WDYT ?

Please see issue #1800. Data loss can happen with others, as well, because currently only data from the reader index on is copied, even if a mark exists before that. I think that as long as you can arbitrarily set the reader index in the buffer, all buffer data should always be copied on expansion.

And the reasoning about your thought?

On 30/08/13 23:07, Norman Maurer wrote:

> @trustin https://github.com/trustin after thinking more about I 
> think we should disallow to expand a duplicated buffer. WDYT ?
> 
> —
> Reply to this email directly or view it on GitHub 
> https://github.com/netty/netty/issues/1802#issuecomment-23563314.

## 

https://twitter.com/trustin
https://twitter.com/trustin_ko
https://twitter.com/netty_project

#1800 is unrelated with this issue.

On 30/08/13 23:40, jhutton wrote:

> Please see issue #1800 https://github.com/netty/netty/issues/1800. 
> Data loss can happen with others, as well, because currently only data 
> from the reader index on is copied, even if a mark exists before that. 
> I think that as long as you can arbitrarily set the reader index in 
> the buffer, all buffer data should always be copied on expansion.
> 
> —
> Reply to this email directly or view it on GitHub 
> https://github.com/netty/netty/issues/1802#issuecomment-23565551.

## 

https://twitter.com/trustin
https://twitter.com/trustin_ko
https://twitter.com/netty_project

@normanmaurer Now I see what you mean.  Yeah, all derived buffers should be unallowed to perform expansion.

@trustin good :) can we do it in 4.0.9 as it is a change in behavior but I would argue it was a bug to allow it before and broken anyway

Sure.  We could probably override `ensureWritable()` and `capacity()` at `AbstractDerivedByteBuf`?  We still cannot prevent a user from expanding the original buffer, right?

Now we finally found a place to use `BufferOverflowException`. :-)

@trustin after thinking moe about it we need to copy the whole old buffer when adjust the capacity if a derived buffer was obtained as otherwise we may corrupt the data for a obtained data if the readerIndex of the original buffer is bigger then the one of the obtained one.

@trustin what you think about this ? See my last comment

Yeah, let's copy the whole region.

Always copy the whole region.


Pull request review comments:
Then why not use it ?
Was going to write the same...
Wouldn't it break subclasses that rely on the use of `allocateArray()` e.g. to [track buffer heap usage](https://github.com/netty/netty/blob/4.1/buffer/src/main/java/io/netty/buffer/UnpooledByteBufAllocator.java#L157) or impl [custom pooling](https://github.com/netty/netty/pull/8015)?

Though I noticed that the `copy()` method currently violates this, I can open another PR to fix that.
See #9440 for the other mentioned fix
I see.... Makes sense. Can you remove TODO: then ?
@normanmaurer done

Patch that resolved the issue:
diff --git a/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java b/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java
index 3a19f85c21a..254d40480ae 100644
--- a/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java
@@ -268,6 +268,13 @@ protected final void adjustMarkers(int decrement) {
         }
     }
 
+    // Called after a capacity reduction
+    protected final void trimIndicesToCapacity(int newCapacity) {
+        if (writerIndex() > newCapacity) {
+            setIndex0(Math.min(readerIndex(), newCapacity), newCapacity);
+        }
+    }
+
     @Override
     public ByteBuf ensureWritable(int minWritableBytes) {
         checkPositiveOrZero(minWritableBytes, "minWritableBytes");
diff --git a/buffer/src/main/java/io/netty/buffer/PoolArena.java b/buffer/src/main/java/io/netty/buffer/PoolArena.java
index 67955ce6d4d..2005e9a0ec3 100644
--- a/buffer/src/main/java/io/netty/buffer/PoolArena.java
+++ b/buffer/src/main/java/io/netty/buffer/PoolArena.java
@@ -379,9 +379,7 @@ int alignCapacity(int reqCapacity) {
     }
 
     void reallocate(PooledByteBuf<T> buf, int newCapacity, boolean freeOldMemory) {
-        if (newCapacity < 0 || newCapacity > buf.maxCapacity()) {
-            throw new IllegalArgumentException("newCapacity: " + newCapacity);
-        }
+        assert newCapacity >= 0 && newCapacity <= buf.maxCapacity();
 
         int oldCapacity = buf.length;
         if (oldCapacity == newCapacity) {
@@ -394,29 +392,17 @@ void reallocate(PooledByteBuf<T> buf, int newCapacity, boolean freeOldMemory) {
         T oldMemory = buf.memory;
         int oldOffset = buf.offset;
         int oldMaxLength = buf.maxLength;
-        int readerIndex = buf.readerIndex();
-        int writerIndex = buf.writerIndex();
 
+        // This does not touch buf's reader/writer indices
         allocate(parent.threadCache(), buf, newCapacity);
+        int bytesToCopy;
         if (newCapacity > oldCapacity) {
-            memoryCopy(
-                    oldMemory, oldOffset,
-                    buf.memory, buf.offset, oldCapacity);
-        } else if (newCapacity < oldCapacity) {
-            if (readerIndex < newCapacity) {
-                if (writerIndex > newCapacity) {
-                    writerIndex = newCapacity;
-                }
-                memoryCopy(
-                        oldMemory, oldOffset + readerIndex,
-                        buf.memory, buf.offset + readerIndex, writerIndex - readerIndex);
-            } else {
-                readerIndex = writerIndex = newCapacity;
-            }
+            bytesToCopy = oldCapacity;
+        } else {
+            buf.trimIndicesToCapacity(newCapacity);
+            bytesToCopy = newCapacity;
         }
-
-        buf.setIndex(readerIndex, writerIndex);
-
+        memoryCopy(oldMemory, oldOffset, buf.memory, buf.offset, bytesToCopy);
         if (freeOldMemory) {
             free(oldChunk, oldNioBuffer, oldHandle, oldMaxLength, buf.cache);
         }
diff --git a/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java b/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
index 8c2d54674db..5b984eb7e78 100644
--- a/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
@@ -110,7 +110,7 @@ public final ByteBuf capacity(int newCapacity) {
                     (maxLength > 512 || newCapacity > maxLength - 16)) {
                 // here newCapacity < length
                 length = newCapacity;
-                setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
+                trimIndicesToCapacity(newCapacity);
                 return this;
             }
         }
diff --git a/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java b/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
index af465c2cd9c..daad1408e7b 100644
--- a/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/UnpooledDirectByteBuf.java
@@ -146,35 +146,23 @@ public int capacity() {
     @Override
     public ByteBuf capacity(int newCapacity) {
         checkNewCapacity(newCapacity);
-
-        int readerIndex = readerIndex();
-        int writerIndex = writerIndex();
-
         int oldCapacity = capacity;
+        if (newCapacity == oldCapacity) {
+            return this;
+        }
+        int bytesToCopy;
         if (newCapacity > oldCapacity) {
-            ByteBuffer oldBuffer = buffer;
-            ByteBuffer newBuffer = allocateDirect(newCapacity);
-            oldBuffer.position(0).limit(oldBuffer.capacity());
-            newBuffer.position(0).limit(oldBuffer.capacity());
-            newBuffer.put(oldBuffer);
-            newBuffer.clear();
-            setByteBuffer(newBuffer, true);
-        } else if (newCapacity < oldCapacity) {
-            ByteBuffer oldBuffer = buffer;
-            ByteBuffer newBuffer = allocateDirect(newCapacity);
-            if (readerIndex < newCapacity) {
-                if (writerIndex > newCapacity) {
-                    writerIndex(writerIndex = newCapacity);
-                }
-                oldBuffer.position(readerIndex).limit(writerIndex);
-                newBuffer.position(readerIndex).limit(writerIndex);
-                newBuffer.put(oldBuffer);
-                newBuffer.clear();
-            } else {
-                setIndex(newCapacity, newCapacity);
-            }
-            setByteBuffer(newBuffer, true);
+            bytesToCopy = oldCapacity;
+        } else {
+            trimIndicesToCapacity(newCapacity);
+            bytesToCopy = newCapacity;
         }
+        ByteBuffer oldBuffer = buffer;
+        ByteBuffer newBuffer = allocateDirect(newCapacity);
+        oldBuffer.position(0).limit(bytesToCopy);
+        newBuffer.position(0).limit(bytesToCopy);
+        newBuffer.put(oldBuffer).clear();
+        setByteBuffer(newBuffer, true);
         return this;
     }
 
diff --git a/buffer/src/main/java/io/netty/buffer/UnpooledHeapByteBuf.java b/buffer/src/main/java/io/netty/buffer/UnpooledHeapByteBuf.java
index 0e1ff3488da..e2ec10ea7ba 100644
--- a/buffer/src/main/java/io/netty/buffer/UnpooledHeapByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/UnpooledHeapByteBuf.java
@@ -120,29 +120,23 @@ public int capacity() {
     @Override
     public ByteBuf capacity(int newCapacity) {
         checkNewCapacity(newCapacity);
-
-        int oldCapacity = array.length;
         byte[] oldArray = array;
+        int oldCapacity = oldArray.length;
+        if (newCapacity == oldCapacity) {
+            return this;
+        }
+
+        int bytesToCopy;
         if (newCapacity > oldCapacity) {
-            byte[] newArray = allocateArray(newCapacity);
-            System.arraycopy(oldArray, 0, newArray, 0, oldArray.length);
-            setArray(newArray);
-            freeArray(oldArray);
-        } else if (newCapacity < oldCapacity) {
-            byte[] newArray = allocateArray(newCapacity);
-            int readerIndex = readerIndex();
-            if (readerIndex < newCapacity) {
-                int writerIndex = writerIndex();
-                if (writerIndex > newCapacity) {
-                    writerIndex(writerIndex = newCapacity);
-                }
-                System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex);
-            } else {
-                setIndex(newCapacity, newCapacity);
-            }
-            setArray(newArray);
-            freeArray(oldArray);
+            bytesToCopy = oldCapacity;
+        } else {
+            trimIndicesToCapacity(newCapacity);
+            bytesToCopy = newCapacity;
         }
+        byte[] newArray = allocateArray(newCapacity);
+        System.arraycopy(oldArray, 0, newArray, 0, bytesToCopy);
+        setArray(newArray);
+        freeArray(oldArray);
         return this;
     }
 
diff --git a/buffer/src/main/java/io/netty/buffer/UnpooledUnsafeNoCleanerDirectByteBuf.java b/buffer/src/main/java/io/netty/buffer/UnpooledUnsafeNoCleanerDirectByteBuf.java
index 3b9c05b83b1..cc00e865adb 100644
--- a/buffer/src/main/java/io/netty/buffer/UnpooledUnsafeNoCleanerDirectByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/UnpooledUnsafeNoCleanerDirectByteBuf.java
@@ -48,18 +48,8 @@ public ByteBuf capacity(int newCapacity) {
             return this;
         }
 
-        ByteBuffer newBuffer = reallocateDirect(buffer, newCapacity);
-
-        if (newCapacity < oldCapacity) {
-            if (readerIndex() < newCapacity) {
-                if (writerIndex() > newCapacity) {
-                    writerIndex(newCapacity);
-                }
-            } else {
-                setIndex(newCapacity, newCapacity);
-            }
-        }
-        setByteBuffer(newBuffer, false);
+        trimIndicesToCapacity(newCapacity);
+        setByteBuffer(reallocateDirect(buffer, newCapacity), false);
         return this;
     }
 }


Test patch: