HBASE-14940 Make our unsafe based ops more safe.

hbase-client/src/main/java/org/apache/hadoop/hbase/filter/FuzzyRowFilter.java

@@ -94,7 +94,7 @@ public class FuzzyRowFilter extends FilterBase {
   }
 
   private void preprocessSearchKey(Pair<byte[], byte[]> p) {
-    if (UnsafeAccess.isAvailable() == false) {
+    if (UnsafeAccess.unaligned() == false) {
       // do nothing
       return;
     }
@@ -113,7 +113,7 @@ public class FuzzyRowFilter extends FilterBase {
    * @return mask array
    */
   private byte[] preprocessMask(byte[] mask) {
-    if (UnsafeAccess.isAvailable() == false) {
+    if (UnsafeAccess.unaligned() == false) {
       // do nothing
       return mask;
     }
@@ -322,7 +322,7 @@ public class FuzzyRowFilter extends FilterBase {
   static SatisfiesCode satisfies(boolean reverse, byte[] row, int offset, int length,
       byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
 
-    if (UnsafeAccess.isAvailable() == false) {
+    if (UnsafeAccess.unaligned() == false) {
       return satisfiesNoUnsafe(reverse, row, offset, length, fuzzyKeyBytes, fuzzyKeyMeta);
     }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/nio/SingleByteBuff.java

@@ -34,6 +34,7 @@ import sun.nio.ch.DirectBuffer;
 public class SingleByteBuff extends ByteBuff {
 
   private static final boolean UNSAFE_AVAIL = UnsafeAccess.isAvailable();
+  private static final boolean UNSAFE_UNALIGNED = UnsafeAccess.unaligned();
 
   // Underlying BB
   private final ByteBuffer buf;
@@ -237,7 +238,7 @@ public class SingleByteBuff extends ByteBuff {
 
   @Override
   public short getShort(int index) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toShort(unsafeRef, unsafeOffset + index);
     }
     return this.buf.getShort(index);
@@ -261,7 +262,7 @@ public class SingleByteBuff extends ByteBuff {
 
   @Override
   public int getInt(int index) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toInt(unsafeRef, unsafeOffset + index);
     }
     return this.buf.getInt(index);
@@ -285,7 +286,7 @@ public class SingleByteBuff extends ByteBuff {
 
   @Override
   public long getLong(int index) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toLong(unsafeRef, unsafeOffset + index);
     }
     return this.buf.getLong(index);

hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferUtils.java

@@ -47,6 +47,7 @@ public final class ByteBufferUtils {
   public final static int NEXT_BIT_SHIFT = 7;
   public final static int NEXT_BIT_MASK = 1 << 7;
   private static final boolean UNSAFE_AVAIL = UnsafeAccess.isAvailable();
+  private static final boolean UNSAFE_UNALIGNED = UnsafeAccess.unaligned();
 
   private ByteBufferUtils() {
   }
@@ -392,9 +393,12 @@ public final class ByteBufferUtils {
     } else if (UNSAFE_AVAIL) {
       UnsafeAccess.copy(in, sourceOffset, out, destinationOffset, length);
     } else {
-      for (int i = 0; i < length; ++i) {
+      int outOldPos = out.position();
-        putByte(out, destinationOffset + i, toByte(in, sourceOffset + i));
+      out.position(destinationOffset);
-      }
+      ByteBuffer inDup = in.duplicate();
+      inDup.position(sourceOffset).limit(sourceOffset + length);
+      out.put(inDup);
+      out.position(outOldPos);
     }
     return destinationOffset + length;
   }
@@ -414,15 +418,15 @@ public final class ByteBufferUtils {
     if (in.hasArray() && out.hasArray()) {
       System.arraycopy(in.array(), sourceOffset + in.arrayOffset(), out.array(), out.position()
           + out.arrayOffset(), length);
+      skip(out, length);
     } else if (UNSAFE_AVAIL) {
       UnsafeAccess.copy(in, sourceOffset, out, out.position(), length);
-    } else {
-      int destOffset = out.position();
-      for (int i = 0; i < length; ++i) {
-        putByte(out, destOffset + i, toByte(in, sourceOffset + i));
-      }
-    }
       skip(out, length);
+    } else {
+      ByteBuffer inDup = in.duplicate();
+      inDup.position(sourceOffset).limit(sourceOffset + length);
+      out.put(inDup);
+    }
   }
 
   /**
@@ -574,7 +578,7 @@ public final class ByteBufferUtils {
   }
 
   public static int compareTo(ByteBuffer buf1, int o1, int l1, ByteBuffer buf2, int o2, int l2) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       long offset1Adj, offset2Adj;
       Object refObj1 = null, refObj2 = null;
       if (buf1.isDirect()) {
@@ -612,7 +616,7 @@ public final class ByteBufferUtils {
   }
 
   public static int compareTo(ByteBuffer buf1, int o1, int l1, byte[] buf2, int o2, int l2) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       long offset1Adj;
       Object refObj1 = null;
       if (buf1.isDirect()) {
@@ -725,7 +729,7 @@ public final class ByteBufferUtils {
    * @return short value at offset
    */
   public static short toShort(ByteBuffer buffer, int offset) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toShort(buffer, offset);
     } else {
       return buffer.getShort(offset);
@@ -739,7 +743,7 @@ public final class ByteBufferUtils {
    * @return int value at offset
    */
   public static int toInt(ByteBuffer buffer, int offset) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toInt(buffer, offset);
     } else {
       return buffer.getInt(offset);
@@ -753,7 +757,7 @@ public final class ByteBufferUtils {
    * @return long value at offset
    */
   public static long toLong(ByteBuffer buffer, int offset) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       return UnsafeAccess.toLong(buffer, offset);
     } else {
       return buffer.getLong(offset);
@@ -767,7 +771,7 @@ public final class ByteBufferUtils {
    * @param val int to write out
    */
   public static void putInt(ByteBuffer buffer, int val) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       int newPos = UnsafeAccess.putInt(buffer, buffer.position(), val);
       buffer.position(newPos);
     } else {
@@ -810,7 +814,7 @@ public final class ByteBufferUtils {
    * @param val short to write out
    */
   public static void putShort(ByteBuffer buffer, short val) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       int newPos = UnsafeAccess.putShort(buffer, buffer.position(), val);
       buffer.position(newPos);
     } else {
@@ -825,7 +829,7 @@ public final class ByteBufferUtils {
    * @param val long to write out
    */
   public static void putLong(ByteBuffer buffer, long val) {
-    if (UNSAFE_AVAIL) {
+    if (UNSAFE_UNALIGNED) {
       int newPos = UnsafeAccess.putLong(buffer, buffer.position(), val);
       buffer.position(newPos);
     } else {
@@ -870,9 +874,10 @@ public final class ByteBufferUtils {
     } else if (UNSAFE_AVAIL) {
       UnsafeAccess.copy(in, sourceOffset, out, destinationOffset, length);
     } else {
-      for (int i = 0; i < length; i++) {
+      int oldPos = in.position();
-        out[destinationOffset + i] = in.get(sourceOffset + i);
+      in.position(sourceOffset);
-      }
+      in.get(out, destinationOffset, length);
+      in.position(oldPos);
     }
   }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java

@@ -789,7 +789,7 @@ public class Bytes implements Comparable<Bytes> {
     if (length != SIZEOF_LONG || offset + length > bytes.length) {
       throw explainWrongLengthOrOffset(bytes, offset, length, SIZEOF_LONG);
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.toLong(bytes, offset);
     } else {
       long l = 0;
@@ -830,7 +830,7 @@ public class Bytes implements Comparable<Bytes> {
       throw new IllegalArgumentException("Not enough room to put a long at"
           + " offset " + offset + " in a " + bytes.length + " byte array");
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.putLong(bytes, offset, val);
     } else {
       for(int i = offset + 7; i > offset; i--) {
@@ -981,7 +981,7 @@ public class Bytes implements Comparable<Bytes> {
     if (length != SIZEOF_INT || offset + length > bytes.length) {
       throw explainWrongLengthOrOffset(bytes, offset, length, SIZEOF_INT);
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.toInt(bytes, offset);
     } else {
       int n = 0;
@@ -1065,7 +1065,7 @@ public class Bytes implements Comparable<Bytes> {
       throw new IllegalArgumentException("Not enough room to put an int at"
           + " offset " + offset + " in a " + bytes.length + " byte array");
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.putInt(bytes, offset, val);
     } else {
       for(int i= offset + 3; i > offset; i--) {
@@ -1135,7 +1135,7 @@ public class Bytes implements Comparable<Bytes> {
     if (length != SIZEOF_SHORT || offset + length > bytes.length) {
       throw explainWrongLengthOrOffset(bytes, offset, length, SIZEOF_SHORT);
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.toShort(bytes, offset);
     } else {
       short n = 0;
@@ -1173,7 +1173,7 @@ public class Bytes implements Comparable<Bytes> {
       throw new IllegalArgumentException("Not enough room to put a short at"
           + " offset " + offset + " in a " + bytes.length + " byte array");
     }
-    if (UnsafeAccess.isAvailable()) {
+    if (UnsafeAccess.unaligned()) {
       return UnsafeAccess.putShort(bytes, offset, val);
     } else {
       bytes[offset+1] = (byte) val;
@@ -1477,7 +1477,7 @@ public class Bytes implements Comparable<Bytes> {
 
       static final Unsafe theUnsafe;
       static {
-        if (UnsafeAccess.isAvailable()) {
+        if (UnsafeAccess.unaligned()) {
           theUnsafe = UnsafeAccess.theUnsafe;
         } else {
           // It doesn't matter what we throw;

hbase-common/src/main/java/org/apache/hadoop/hbase/util/UnsafeAccess.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.util;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.security.AccessController;
@@ -38,6 +39,7 @@ public final class UnsafeAccess {
   private static final Log LOG = LogFactory.getLog(UnsafeAccess.class);
 
   static final Unsafe theUnsafe;
+  private static boolean unaligned;
 
   /** The offset to the first element in a byte array. */
   public static final long BYTE_ARRAY_BASE_OFFSET;
@@ -45,6 +47,11 @@ public final class UnsafeAccess {
   static final boolean littleEndian = ByteOrder.nativeOrder()
       .equals(ByteOrder.LITTLE_ENDIAN);
 
+  // This number limits the number of bytes to copy per call to Unsafe's
+  // copyMemory method. A limit is imposed to allow for safepoint polling
+  // during a large copy
+  static final long UNSAFE_COPY_THRESHOLD = 1024L * 1024L;
+
   static {
     theUnsafe = (Unsafe) AccessController.doPrivileged(new PrivilegedAction<Object>() {
       @Override
@@ -62,20 +69,38 @@ public final class UnsafeAccess {
 
     if (theUnsafe != null) {
       BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
+      try {
+        // Using java.nio.Bits#unaligned() to check for unaligned-access capability
+        Class<?> clazz = Class.forName("java.nio.Bits");
+        Method m = clazz.getDeclaredMethod("unaligned");
+        m.setAccessible(true);
+        unaligned = (boolean) m.invoke(null);
+      } catch (Exception e) {
+        unaligned = false;
+      }
     } else{
       BYTE_ARRAY_BASE_OFFSET = -1;
+      unaligned = false;
     }
   }
 
   private UnsafeAccess(){}
 
   /**
-   * @return true when the running JVM is having sun's Unsafe package available in it.
+   * @return true when running JVM is having sun's Unsafe package available in it.
    */
   public static boolean isAvailable() {
     return theUnsafe != null;
   }
 
+  /**
+   * @return true when running JVM is having sun's Unsafe package available in it and underlying
+   *         system having unaligned-access capability.
+   */
+  public static boolean unaligned() {
+    return unaligned;
+  }
+
   // APIs to read primitive data from a byte[] using Unsafe way
   /**
    * Converts a byte array to a short value considering it was written in big-endian format.
@@ -330,7 +355,17 @@ public final class UnsafeAccess {
       destBase = dest.array();
     }
     long srcAddress = srcOffset + BYTE_ARRAY_BASE_OFFSET;
-    theUnsafe.copyMemory(src, srcAddress, destBase, destAddress, length);
+    unsafeCopy(src, srcAddress, destBase, destAddress, length);
+  }
+
+  private static void unsafeCopy(Object src, long srcAddr, Object dst, long destAddr, long len) {
+    while (len > 0) {
+      long size = (len > UNSAFE_COPY_THRESHOLD) ? UNSAFE_COPY_THRESHOLD : len;
+      theUnsafe.copyMemory(src, srcAddr, dst, destAddr, len);
+      len -= size;
+      srcAddr += size;
+      destAddr += size;
+    }
   }
 
   /**
@@ -354,7 +389,7 @@ public final class UnsafeAccess {
       srcBase = src.array();
     }
     long destAddress = destOffset + BYTE_ARRAY_BASE_OFFSET;
-    theUnsafe.copyMemory(srcBase, srcAddress, dest, destAddress, length);
+    unsafeCopy(srcBase, srcAddress, dest, destAddress, length);
   }
 
   /**
@@ -383,8 +418,9 @@ public final class UnsafeAccess {
       destAddress = destOffset + BYTE_ARRAY_BASE_OFFSET + dest.arrayOffset();
       destBase = dest.array();
     }
-    theUnsafe.copyMemory(srcBase, srcAddress, destBase, destAddress, length);
+    unsafeCopy(srcBase, srcAddress, destBase, destAddress, length);
   }
+
   // APIs to add primitives to BBs
   /**
    * Put a short value out to the specified BB position in big-endian format.