Ensure Unsafe buffer implementations are used when sun.misc.Unsafe is present

Motivation:

When sun.misc.Unsafe is present we want to use *Unsafe*ByteBuf implementations. We missed to do so in PooledByteBufAllocator when the heapArena is null.

Modifications:

- Correctly use UnpooledUnsafeHeapByteBuf
- Add unit tests

Result:

Use most optimal ByteBuf implementation.

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

@@ -296,11 +296,13 @@ public class PooledByteBufAllocator extends AbstractByteBufAllocator {
         PoolThreadCache cache = threadCache.get();
         PoolArena<byte[]> heapArena = cache.heapArena;
 
-        ByteBuf buf;
+        final ByteBuf buf;
         if (heapArena != null) {
             buf = heapArena.allocate(cache, initialCapacity, maxCapacity);
         } else {
-            buf = new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
+            buf = PlatformDependent.hasUnsafe() ?
+                    new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
+                    new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
         }
 
         return toLeakAwareBuffer(buf);
@@ -311,15 +313,13 @@ public class PooledByteBufAllocator extends AbstractByteBufAllocator {
         PoolThreadCache cache = threadCache.get();
         PoolArena<ByteBuffer> directArena = cache.directArena;
 
-        ByteBuf buf;
+        final ByteBuf buf;
         if (directArena != null) {
             buf = directArena.allocate(cache, initialCapacity, maxCapacity);
         } else {
-            if (PlatformDependent.hasUnsafe()) {
-                buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
-            } else {
-                buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
-            }
+            buf = PlatformDependent.hasUnsafe() ?
+                    UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
+                    new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
         }
 
         return toLeakAwareBuffer(buf);

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

@@ -19,6 +19,7 @@ import io.netty.util.internal.PlatformDependent;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public abstract class AbstractByteBufAllocatorTest extends ByteBufAllocatorTest {
@@ -26,6 +27,8 @@ public abstract class AbstractByteBufAllocatorTest extends ByteBufAllocatorTest
     @Override
     protected abstract AbstractByteBufAllocator newAllocator(boolean preferDirect);
 
+    protected abstract AbstractByteBufAllocator newUnpooledAllocator();
+
     @Override
     protected boolean isDirectExpected(boolean preferDirect) {
         return preferDirect && PlatformDependent.hasUnsafe();
@@ -72,4 +75,23 @@ public abstract class AbstractByteBufAllocatorTest extends ByteBufAllocatorTest
             // expected
         }
     }
+
+    @Test
+    public void testUnsafeHeapBufferAndUnsafeDirectBuffer() {
+        AbstractByteBufAllocator allocator = newUnpooledAllocator();
+        ByteBuf directBuffer = allocator.directBuffer();
+        assertInstanceOf(directBuffer,
+                PlatformDependent.hasUnsafe() ? UnpooledUnsafeDirectByteBuf.class : UnpooledDirectByteBuf.class);
+        directBuffer.release();
+
+        ByteBuf heapBuffer = allocator.heapBuffer();
+        assertInstanceOf(heapBuffer,
+                PlatformDependent.hasUnsafe() ? UnpooledUnsafeHeapByteBuf.class : UnpooledHeapByteBuf.class);
+        heapBuffer.release();
+    }
+
+    protected static void assertInstanceOf(ByteBuf buffer, Class<? extends ByteBuf> clazz) {
+        // Unwrap if needed
+        assertTrue(clazz.isInstance(buffer instanceof SimpleLeakAwareByteBuf ? buffer.unwrap() : buffer));
+    }
 }

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

@@ -18,6 +18,7 @@ package io.netty.buffer;
 
 import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.concurrent.FastThreadLocalThread;
+import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.SystemPropertyUtil;
 import org.junit.Assume;
 import org.junit.Test;
@@ -43,6 +44,25 @@ public class PooledByteBufAllocatorTest extends AbstractByteBufAllocatorTest {
         return new PooledByteBufAllocator(preferDirect);
     }
 
+    @Override
+    protected AbstractByteBufAllocator newUnpooledAllocator() {
+        return new PooledByteBufAllocator(0, 0, 8192, 1);
+    }
+
+    @Test
+    public void testPooledUnsafeHeapBufferAndUnsafeDirectBuffer() {
+        AbstractByteBufAllocator allocator = newAllocator(true);
+        ByteBuf directBuffer = allocator.directBuffer();
+        assertInstanceOf(directBuffer,
+                PlatformDependent.hasUnsafe() ? PooledUnsafeDirectByteBuf.class : PooledDirectByteBuf.class);
+        directBuffer.release();
+
+        ByteBuf heapBuffer = allocator.heapBuffer();
+        assertInstanceOf(heapBuffer,
+                PlatformDependent.hasUnsafe() ? PooledUnsafeHeapByteBuf.class : PooledHeapByteBuf.class);
+        heapBuffer.release();
+    }
+
     @Test
     public void testArenaMetricsNoCache() {
         testArenaMetrics0(new PooledByteBufAllocator(true, 2, 2, 8192, 11, 0, 0, 0), 100, 0, 100, 100);

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

@@ -21,4 +21,9 @@ public class UnpooledByteBufAllocatorTest extends AbstractByteBufAllocatorTest {
     protected AbstractByteBufAllocator newAllocator(boolean preferDirect) {
         return new UnpooledByteBufAllocator(preferDirect);
     }
+
+    @Override
+    protected AbstractByteBufAllocator newUnpooledAllocator() {
+        return new UnpooledByteBufAllocator(false);
+    }
 }