Calculate correct count for tiny/small/normal allocation

Motivation:

Disable ThreadLocal Cache, then allocate Pooled ByteBuf and release all these buffers, PoolArena's tiny/small/normal allocation count is incorrect.

Modifications:

- Calculate PoolArena's tiny/small/normal allocation one time
- Add testAllocationCounter TestCase

Result:

Fixes #6282 .

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

@@ -201,16 +201,15 @@ abstract class PoolArena<T> implements PoolArenaMetric {
                     long handle = s.allocate();
                     assert handle >= 0;
                     s.chunk.initBufWithSubpage(buf, handle, reqCapacity);
-
-                    if (tiny) {
-                        allocationsTiny.increment();
-                    } else {
-                        allocationsSmall.increment();
-                    }
+                    incTinySmallAllocation(tiny);
                     return;
                 }
             }
-            allocateNormal(buf, reqCapacity, normCapacity);
+            synchronized (this) {
+                allocateNormal(buf, reqCapacity, normCapacity);
+            }
+
+            incTinySmallAllocation(tiny);
             return;
         }
         if (normCapacity <= chunkSize) {
@@ -218,30 +217,40 @@ abstract class PoolArena<T> implements PoolArenaMetric {
                 // was able to allocate out of the cache so move on
                 return;
             }
-            allocateNormal(buf, reqCapacity, normCapacity);
+            synchronized (this) {
+                allocateNormal(buf, reqCapacity, normCapacity);
+                ++allocationsNormal;
+            }
         } else {
             // Huge allocations are never served via the cache so just call allocateHuge
             allocateHuge(buf, reqCapacity);
         }
     }
 
-    private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
+    // Method must be called insided synchronized(this) { ... } block
+    private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
         if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||
             q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||
             q075.allocate(buf, reqCapacity, normCapacity)) {
-            ++allocationsNormal;
             return;
         }
 
         // Add a new chunk.
         PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
         long handle = c.allocate(normCapacity);
-        ++allocationsNormal;
         assert handle > 0;
         c.initBuf(buf, handle, reqCapacity);
         qInit.add(c);
     }
 
+    private void incTinySmallAllocation(boolean tiny) {
+        if (tiny) {
+            allocationsTiny.increment();
+        } else {
+            allocationsSmall.increment();
+        }
+    }
+
     private void allocateHuge(PooledByteBuf<T> buf, int reqCapacity) {
         PoolChunk<T> chunk = newUnpooledChunk(reqCapacity);
         activeBytesHuge.add(chunk.chunkSize());

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

@@ -32,4 +32,49 @@ public class PoolArenaTest {
             Assert.assertEquals(expectedResult[i], arena.normalizeCapacity(reqCapacities[i]));
         }
     }
+
+    @Test
+    public final void testAllocationCounter() {
+        final PooledByteBufAllocator allocator = new PooledByteBufAllocator(
+                true,   // preferDirect
+                0,      // nHeapArena
+                1,      // nDirectArena
+                8192,   // pageSize
+                11,     // maxOrder
+                0,      // tinyCacheSize
+                0,      // smallCacheSize
+                0,      // normalCacheSize
+                true    // useCacheForAllThreads
+                );
+
+        // create tiny buffer
+        final ByteBuf b1 = allocator.directBuffer(24);
+        // create small buffer
+        final ByteBuf b2 = allocator.directBuffer(800);
+        // create normal buffer
+        final ByteBuf b3 = allocator.directBuffer(8192 * 2);
+
+        Assert.assertNotNull(b1);
+        Assert.assertNotNull(b2);
+        Assert.assertNotNull(b3);
+
+        // then release buffer to deallocated memory while threadlocal cache has been disabled
+        // allocations counter value must equals deallocations counter value
+        Assert.assertTrue(b1.release());
+        Assert.assertTrue(b2.release());
+        Assert.assertTrue(b3.release());
+
+        Assert.assertTrue(allocator.directArenas().size() >= 1);
+        final PoolArenaMetric metric = allocator.directArenas().get(0);
+
+        Assert.assertEquals(3, metric.numDeallocations());
+        Assert.assertEquals(3, metric.numAllocations());
+
+        Assert.assertEquals(1, metric.numTinyDeallocations());
+        Assert.assertEquals(1, metric.numTinyAllocations());
+        Assert.assertEquals(1, metric.numSmallDeallocations());
+        Assert.assertEquals(1, metric.numSmallAllocations());
+        Assert.assertEquals(1, metric.numNormalDeallocations());
+        Assert.assertEquals(1, metric.numNormalAllocations());
+    }
 }