Allow to use native transports when sun.misc.Unsafe is not present on… (

netty#8231)

* Allow to use native transports when sun.misc.Unsafe is not present on the system

Motivation:

We should be able to use the native transports (epoll / kqueue) even when sun.misc.Unsafe is not present on the system. This is especially important as Java11 will be released soon and does not allow access to it by default.

Modifications:

- Correctly disable usage of sun.misc.Unsafe when -PnoUnsafe is used while running the build
- Correctly increment metric when UnpooledDirectByteBuf is allocated. This was uncovered once -PnoUnsafe usage was fixed.
- Implement fallbacks in all our native transport code for when sun.misc.Unsafe is not present.

Result:

Fixes netty#8229.

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

@@ -64,7 +64,7 @@ public UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int ma
         }
 
         this.alloc = alloc;
-        setByteBuffer(ByteBuffer.allocateDirect(initialCapacity));
+        setByteBuffer(allocateDirect(initialCapacity));
     }
 
     /**

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

@@ -401,7 +401,7 @@ public void testHasMemoryAddressWhenEmpty() {
         buf.release();
     }
 
-    @Test(expected = UnsupportedOperationException.class)
+    @Test
     public void testHasNoMemoryAddressWhenMultipleBuffers() {
         ByteBuf buf1 = directBuffer(10);
         if (!buf1.hasMemoryAddress()) {
@@ -415,6 +415,8 @@ public void testHasNoMemoryAddressWhenMultipleBuffers() {
         try {
             buf.memoryAddress();
             fail();
+        } catch (UnsupportedOperationException expected) {
+            // expected
         } finally {
             buf.release();
         }

pom.xml

@@ -136,7 +136,7 @@
     <profile>
       <id>noUnsafe</id>
       <properties>
-        <argLine.noUnsafe>-Dio.netty.noUnsafe</argLine.noUnsafe>
+        <argLine.noUnsafe>-Dio.netty.noUnsafe=true</argLine.noUnsafe>
       </properties>
     </profile>
     <profile>

transport-native-epoll/src/main/c/netty_epoll_native.c

@@ -38,6 +38,7 @@
 #include <time.h>
 
 #include "netty_epoll_linuxsocket.h"
+#include "netty_unix_buffer.h"
 #include "netty_unix_errors.h"
 #include "netty_unix_filedescriptor.h"
 #include "netty_unix_jni.h"
@@ -452,6 +453,9 @@ static jint netty_epoll_native_JNI_OnLoad(JNIEnv* env, const char* packagePrefix
     if (netty_unix_socket_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
         return JNI_ERR;
     }
+    if (netty_unix_buffer_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
+        return JNI_ERR;
+    }
     if (netty_epoll_linuxsocket_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
         return JNI_ERR;
     }
@@ -501,6 +505,7 @@ static void netty_epoll_native_JNI_OnUnLoad(JNIEnv* env) {
     netty_unix_errors_JNI_OnUnLoad(env);
     netty_unix_filedescriptor_JNI_OnUnLoad(env);
     netty_unix_socket_JNI_OnUnLoad(env);
+    netty_unix_buffer_JNI_OnUnLoad(env);
     netty_epoll_linuxsocket_JNI_OnUnLoad(env);
 }
 

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java

@@ -513,22 +513,13 @@ protected int doWriteSingle(ChannelOutboundBuffer in) throws Exception {
      */
     private int doWriteMultiple(ChannelOutboundBuffer in) throws Exception {
         final long maxBytesPerGatheringWrite = config().getMaxBytesPerGatheringWrite();
-        if (PlatformDependent.hasUnsafe()) {
-            IovArray array = ((EpollEventLoop) eventLoop()).cleanIovArray();
-            array.maxBytes(maxBytesPerGatheringWrite);
-            in.forEachFlushedMessage(array);
-
-            if (array.count() >= 1) {
-                // TODO: Handle the case where cnt == 1 specially.
-                return writeBytesMultiple(in, array);
-            }
-        } else {
-            ByteBuffer[] buffers = in.nioBuffers();
-            int cnt = in.nioBufferCount();
-            if (cnt >= 1) {
-                // TODO: Handle the case where cnt == 1 specially.
-                return writeBytesMultiple(in, buffers, cnt, in.nioBufferSize(), maxBytesPerGatheringWrite);
-            }
+        IovArray array = ((EpollEventLoop) eventLoop()).cleanIovArray();
+        array.maxBytes(maxBytesPerGatheringWrite);
+        in.forEachFlushedMessage(array);
+
+        if (array.count() >= 1) {
+            // TODO: Handle the case where cnt == 1 specially.
+            return writeBytesMultiple(in, array);
         }
         // cnt == 0, which means the outbound buffer contained empty buffers only.
         in.removeBytes(0);

transport-native-epoll/src/main/java/io/netty/channel/epoll/Epoll.java

@@ -16,7 +16,6 @@
 package io.netty.channel.epoll;
 
 import io.netty.channel.unix.FileDescriptor;
-import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.SystemPropertyUtil;
 
 /**
@@ -58,15 +57,7 @@ public final class Epoll {
             }
         }
 
-        if (cause != null) {
-            UNAVAILABILITY_CAUSE = cause;
-        } else {
-            UNAVAILABILITY_CAUSE = PlatformDependent.hasUnsafe()
-                    ? null
-                    : new IllegalStateException(
-                    "sun.misc.Unsafe not available",
-                    PlatformDependent.getUnsafeUnavailabilityCause());
-        }
+        UNAVAILABILITY_CAUSE = cause;
     }
 
     /**

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventArray.java

@@ -15,8 +15,11 @@
  */
 package io.netty.channel.epoll;
 
+import io.netty.channel.unix.Buffer;
 import io.netty.util.internal.PlatformDependent;
 
+import java.nio.ByteBuffer;
+
 /**
  * This is an internal datastructure which can be directly passed to epoll_wait to reduce the overhead.
  *
@@ -41,6 +44,7 @@ final class EpollEventArray {
     // The offsiet of the data union in the epoll_event struct
     private static final int EPOLL_DATA_OFFSET = Native.offsetofEpollData();
 
+    private ByteBuffer memory;
     private long memoryAddress;
     private int length;
 
@@ -49,11 +53,8 @@ final class EpollEventArray {
             throw new IllegalArgumentException("length must be >= 1 but was " + length);
         }
         this.length = length;
-        memoryAddress = allocate(length);
-    }
-
-    private static long allocate(int length) {
-        return PlatformDependent.allocateMemory(length * EPOLL_EVENT_SIZE);
+        memory = Buffer.allocateDirectWithNativeOrder(calculateBufferCapacity(length));
+        memoryAddress = Buffer.memoryAddress(memory);
     }
 
     /**
@@ -77,28 +78,43 @@ int length() {
     void increase() {
         // double the size
         length <<= 1;
-        free();
-        memoryAddress = allocate(length);
+        // There is no need to preserve what was in the memory before.
+        ByteBuffer buffer = Buffer.allocateDirectWithNativeOrder(calculateBufferCapacity(length));
+        Buffer.free(memory);
+        memory = buffer;
+        memoryAddress = Buffer.memoryAddress(buffer);
     }
 
     /**
      * Free this {@link EpollEventArray}. Any usage after calling this method may segfault the JVM!
      */
     void free() {
-        PlatformDependent.freeMemory(memoryAddress);
+        Buffer.free(memory);
+        memoryAddress = 0;
     }
 
     /**
      * Return the events for the {@code epoll_event} on this index.
      */
     int events(int index) {
-        return PlatformDependent.getInt(memoryAddress + index * EPOLL_EVENT_SIZE);
+        return getInt(index, 0);
     }
 
     /**
      * Return the file descriptor for the {@code epoll_event} on this index.
      */
     int fd(int index) {
-        return PlatformDependent.getInt(memoryAddress + index * EPOLL_EVENT_SIZE + EPOLL_DATA_OFFSET);
+        return getInt(index, EPOLL_DATA_OFFSET);
+    }
+
+    private int getInt(int index, int offset) {
+        if (PlatformDependent.hasUnsafe()) {
+            return PlatformDependent.getInt(memoryAddress + index * EPOLL_EVENT_SIZE + offset);
+        }
+        return memory.getInt(index * EPOLL_EVENT_SIZE + offset);
+    }
+
+    private static int calculateBufferCapacity(int capacity) {
+        return capacity * EPOLL_EVENT_SIZE;
     }
 }

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollRecvByteAllocatorHandle.java

@@ -19,10 +19,13 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.unix.PreferredDirectByteBufAllocator;
 import io.netty.util.UncheckedBooleanSupplier;
 import io.netty.util.internal.ObjectUtil;
 
 class EpollRecvByteAllocatorHandle implements RecvByteBufAllocator.ExtendedHandle {
+    private final PreferredDirectByteBufAllocator preferredDirectByteBufAllocator =
+            new PreferredDirectByteBufAllocator();
     private final RecvByteBufAllocator.ExtendedHandle delegate;
     private final UncheckedBooleanSupplier defaultMaybeMoreDataSupplier = new UncheckedBooleanSupplier() {
         @Override
@@ -34,7 +37,7 @@ public boolean get() {
     private boolean receivedRdHup;
 
     EpollRecvByteAllocatorHandle(RecvByteBufAllocator.ExtendedHandle handle) {
-        this.delegate = ObjectUtil.checkNotNull(handle, "handle");
+        delegate = ObjectUtil.checkNotNull(handle, "handle");
     }
 
     final void receivedRdHup() {
@@ -69,7 +72,9 @@ final boolean isEdgeTriggered() {
 
     @Override
     public final ByteBuf allocate(ByteBufAllocator alloc) {
-        return delegate.allocate(alloc);
+        // We need to ensure we always allocate a direct ByteBuf as we can only use a direct buffer to read via JNI.
+        preferredDirectByteBufAllocator.updateAllocator(alloc);
+        return delegate.allocate(preferredDirectByteBufAllocator);
     }
 
     @Override

transport-native-kqueue/src/main/c/netty_kqueue_native.c

@@ -28,6 +28,7 @@
 
 #include "netty_kqueue_bsdsocket.h"
 #include "netty_kqueue_eventarray.h"
+#include "netty_unix_buffer.h"
 #include "netty_unix_errors.h"
 #include "netty_unix_filedescriptor.h"
 #include "netty_unix_jni.h"
@@ -293,6 +294,9 @@ static jint netty_kqueue_native_JNI_OnLoad(JNIEnv* env, const char* packagePrefi
     if (netty_unix_socket_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
         return JNI_ERR;
     }
+    if (netty_unix_buffer_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
+        return JNI_ERR;
+    }
     if (netty_kqueue_bsdsocket_JNI_OnLoad(env, packagePrefix) == JNI_ERR) {
         return JNI_ERR;
     }
@@ -314,6 +318,7 @@ static void netty_kqueue_native_JNI_OnUnLoad(JNIEnv* env) {
     netty_unix_errors_JNI_OnUnLoad(env);
     netty_unix_filedescriptor_JNI_OnUnLoad(env);
     netty_unix_socket_JNI_OnUnLoad(env);
+    netty_unix_buffer_JNI_OnUnLoad(env);
     netty_kqueue_bsdsocket_JNI_OnUnLoad(env);
     netty_kqueue_eventarray_JNI_OnUnLoad(env);
 }

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java

@@ -33,7 +33,6 @@
 import io.netty.channel.unix.IovArray;
 import io.netty.channel.unix.SocketWritableByteChannel;
 import io.netty.channel.unix.UnixChannelUtil;
-import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.UnstableApi;
 import io.netty.util.internal.logging.InternalLogger;
@@ -345,22 +344,13 @@ protected int doWriteSingle(ChannelOutboundBuffer in) throws Exception {
      */
     private int doWriteMultiple(ChannelOutboundBuffer in) throws Exception {
         final long maxBytesPerGatheringWrite = config().getMaxBytesPerGatheringWrite();
-        if (PlatformDependent.hasUnsafe()) {
-            IovArray array = ((KQueueEventLoop) eventLoop()).cleanArray();
-            array.maxBytes(maxBytesPerGatheringWrite);
-            in.forEachFlushedMessage(array);
-
-            if (array.count() >= 1) {
-                // TODO: Handle the case where cnt == 1 specially.
-                return writeBytesMultiple(in, array);
-            }
-        } else {
-            ByteBuffer[] buffers = in.nioBuffers();
-            int cnt = in.nioBufferCount();
-            if (cnt >= 1) {
-                // TODO: Handle the case where cnt == 1 specially.
-                return writeBytesMultiple(in, buffers, cnt, in.nioBufferSize(), maxBytesPerGatheringWrite);
-            }
+        IovArray array = ((KQueueEventLoop) eventLoop()).cleanArray();
+        array.maxBytes(maxBytesPerGatheringWrite);
+        in.forEachFlushedMessage(array);
+
+        if (array.count() >= 1) {
+            // TODO: Handle the case where cnt == 1 specially.
+            return writeBytesMultiple(in, array);
         }
         // cnt == 0, which means the outbound buffer contained empty buffers only.
         in.removeBytes(0);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueue.java

@@ -16,7 +16,6 @@
 package io.netty.channel.kqueue;
 
 import io.netty.channel.unix.FileDescriptor;
-import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.SystemPropertyUtil;
 import io.netty.util.internal.UnstableApi;
 
@@ -48,15 +47,7 @@ public final class KQueue {
             }
         }
 
-        if (cause != null) {
-            UNAVAILABILITY_CAUSE = cause;
-        } else {
-            UNAVAILABILITY_CAUSE = PlatformDependent.hasUnsafe()
-                    ? null
-                    : new IllegalStateException(
-                            "sun.misc.Unsafe not available",
-                            PlatformDependent.getUnsafeUnavailabilityCause());
-        }
+        UNAVAILABILITY_CAUSE = cause;
     }
 
     /**