PoolChunkList not correctly move PoolChunks when these are moved.

Motivation:

When a PoolChunk needs to get moved to the previous PoolChunkList because of the minUsage / maxUsage constraints we always just moved it one level which is incorrect and so could lead to have PoolChunks in the wrong PoolChunkList (in respect to their minUsage / maxUsage settings). This then could have the effect that PoolChunks are not released / freed in a timely fashion and so.

Modifications:

- Correctly move PoolChunks between PoolChunkLists, which includes moving it multiple "levels".
- Add unit test

Result:

Correctlty move the PoolChunk to PoolChunkList when it is freed, even if its multiple layers.

buffer/src/main/java/io/netty/buffer/PoolChunkList.java

@@ -75,15 +75,37 @@ final class PoolChunkList<T> implements PoolChunkListMetric {
         chunk.free(handle);
         if (chunk.usage() < minUsage) {
             remove(chunk);
+            // Move the PoolChunk down the PoolChunkList linked-list.
+            return move0(chunk);
+        }
+        return true;
+    }
+
+    private boolean move(PoolChunk<T> chunk) {
+        assert chunk.usage() < maxUsage;
+
+        if (chunk.usage() < minUsage) {
+            // Move the PoolChunk down the PoolChunkList linked-list.
+            return move0(chunk);
+        }
+
+        // PoolChunk fits into this PoolChunkList, adding it here.
+        add0(chunk);
+        return true;
+    }
+
+    /**
+     * Moves the {@link PoolChunk} down the {@link PoolChunkList} linked-list so it will end up in the right
+     * {@link PoolChunkList} that has the correct minUsage / maxUsage in respect to {@link PoolChunk#usage()}.
+     */
+    private boolean move0(PoolChunk<T> chunk) {
         if (prevList == null) {
+            // There is no previous PoolChunkList so return false which result in having the PoolChunk destroyed and
+            // all memory associated with the PoolChunk will be released.
             assert chunk.usage() == 0;
             return false;
-            } else {
-                prevList.add(chunk);
-                return true;
         }
-        }
-        return true;
+        return prevList.move(chunk);
     }
 
     void add(PoolChunk<T> chunk) {
@@ -91,7 +113,13 @@ final class PoolChunkList<T> implements PoolChunkListMetric {
             nextList.add(chunk);
             return;
         }
+        add0(chunk);
+    }
 
+    /**
+     * Adds the {@link PoolChunk} to this {@link PoolChunkList}.
+     */
+    void add0(PoolChunk<T> chunk) {
         chunk.parent = this;
         if (head == null) {
             head = chunk;

buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java

@@ -31,6 +31,7 @@ import java.util.concurrent.locks.LockSupport;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class PooledByteBufAllocatorTest {
@@ -57,6 +58,35 @@ public class PooledByteBufAllocatorTest {
         assertEquals(max, m.maxUsage());
     }
 
+    @Test
+    public void testFreePoolChunk() {
+        int chunkSize = 16 * 1024 * 1024;
+        PooledByteBufAllocator allocator = new PooledByteBufAllocator(true, 1, 0, 8192, 11, 0, 0, 0);
+        ByteBuf buffer = allocator.heapBuffer(chunkSize);
+        List<PoolArenaMetric> arenas = allocator.heapArenas();
+        assertEquals(1, arenas.size());
+        List<PoolChunkListMetric> lists = arenas.get(0).chunkLists();
+        assertEquals(6, lists.size());
+
+        assertFalse(lists.get(0).iterator().hasNext());
+        assertFalse(lists.get(1).iterator().hasNext());
+        assertFalse(lists.get(2).iterator().hasNext());
+        assertFalse(lists.get(3).iterator().hasNext());
+        assertFalse(lists.get(4).iterator().hasNext());
+
+        // Must end up in the 6th PoolChunkList
+        assertTrue(lists.get(5).iterator().hasNext());
+        assertTrue(buffer.release());
+
+        // Should be completely removed and so all PoolChunkLists must be empty
+        assertFalse(lists.get(0).iterator().hasNext());
+        assertFalse(lists.get(1).iterator().hasNext());
+        assertFalse(lists.get(2).iterator().hasNext());
+        assertFalse(lists.get(3).iterator().hasNext());
+        assertFalse(lists.get(4).iterator().hasNext());
+        assertFalse(lists.get(5).iterator().hasNext());
+    }
+
     // The ThreadDeathWatcher sleeps 1s, give it double that time.
     @Test (timeout = 2000)
     public void testThreadCacheDestroyedByThreadDeathWatcher() {