HBASE-20717 Implement CCSMap - a better concurrent map with compacted data structure

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

@@ -43,6 +43,9 @@ public final class UnsafeAccess {
 
   /** The offset to the first element in a byte array. */
   public static final int BYTE_ARRAY_BASE_OFFSET;
+  /** The offset to the first element in a object array. */
+  private static final int OBJECT_BASE_OFFSET;
+  private static final int OBJECT_SCALE;
 
   public static final boolean LITTLE_ENDIAN =
     ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN);
@@ -68,8 +71,12 @@ public Object run() {
 
     if(theUnsafe != null){
       BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
+      OBJECT_BASE_OFFSET = theUnsafe.arrayBaseOffset(Object[].class);
+      OBJECT_SCALE = theUnsafe.arrayIndexScale(Object[].class);
     } else{
       BYTE_ARRAY_BASE_OFFSET = -1;
+      OBJECT_BASE_OFFSET = -1;
+      OBJECT_SCALE = -1;
     }
   }
 
@@ -392,6 +399,21 @@ public static int putLong(ByteBuffer buf, int offset, long val) {
     return offset + Bytes.SIZEOF_LONG;
   }
 
+  /**
+   * Put a long value out to the specified BB position in big-endian format. (Volatile version)
+   * @param bytes the byte array
+   * @param offset position in the buffer
+   * @param val long to write out
+   * @return incremented offset
+   */
+  public static int putLongVolatile(byte[] bytes, int offset, long val) {
+    if (LITTLE_ENDIAN) {
+      val = Long.reverseBytes(val);
+    }
+    theUnsafe.putLongVolatile(bytes, BYTE_ARRAY_BASE_OFFSET + offset, val);
+    return offset + Bytes.SIZEOF_LONG;
+  }
+
   /**
    * Put a byte value out to the specified BB position in big-endian format.
    * @param buf the byte buffer
@@ -422,4 +444,61 @@ public static byte toByte(ByteBuffer buf, int offset) {
       return theUnsafe.getByte(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset);
     }
   }
+
+  /**
+   * A CAS operation for long.
+   * @param bytes the byte array
+   * @param offset position in the array
+   * @param expect an expect value (long)
+   * @param update an updating value (long)
+   * @return true if successful
+   */
+  public static boolean compareAndSetLong(byte[] bytes, int offset, long expect, long update) {
+    if (LITTLE_ENDIAN) {
+      expect = Long.reverseBytes(expect);
+      update = Long.reverseBytes(update);
+    }
+    return theUnsafe.compareAndSwapLong(bytes, BYTE_ARRAY_BASE_OFFSET + offset,
+                                        expect,
+                                        update);
+  }
+
+  /**
+   * A CAS operation for object.
+   * @param objects the objects array
+   * @param offset position in the array
+   * @param expect an expect value (Object)
+   * @param update an updating value (Object)
+   * @return true if successful
+   */
+  public static boolean compareAndSwapObject(Object[] objects, int offset,
+                                             Object expect, Object update) {
+    return theUnsafe.compareAndSwapObject(objects,
+                                          OBJECT_BASE_OFFSET + offset * OBJECT_SCALE,
+                                          expect,
+                                          update);
+  }
+
+  /**
+   * Read a long from bytes array starting from offset.
+   * @param bytes the bytes array
+   * @param offset the start offset
+   * @return a long value
+   */
+  public static long toLongVolatile(byte[] bytes, int offset) {
+    return LITTLE_ENDIAN ?
+      Long.reverseBytes(theUnsafe.getLongVolatile(bytes, BYTE_ARRAY_BASE_OFFSET + offset)) :
+      theUnsafe.getLongVolatile(bytes, BYTE_ARRAY_BASE_OFFSET + offset);
+  }
+
+  /**
+   * Read a object from objects array starting from offset.
+   * @param objects the object array
+   * @param offset the start offset
+   * @return an object value
+   */
+  public static Object getObjectVolatile(Object[] objects, int offset) {
+    return theUnsafe.getObjectVolatile(objects, OBJECT_BASE_OFFSET + offset * OBJECT_SCALE);
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/ccsmap/CCSMapUtils.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.hbase.ccsmap;
+
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Bytes;
+
+@InterfaceAudience.Private
+public final class CCSMapUtils {
+
+  public static final int DEFAULT_PAGE_SIZE = 2 * 1024 * 1024; // 2MB
+  public static final int DEFAULT_PAGES = 4096;
+  public static final int DEFAULT_HEAPKV_PAGE_SIZE = 1024;
+  public static final int DEFAULT_HEAPKV_PAGES = 1024;
+
+  public static int getShiftFromX(int x) {
+    if ((x & ~x) != 0) {
+      throw new IllegalArgumentException("Page size must be power of 2");
+    }
+    int ret = 0;
+    while ((x >>= 1) > 0) ret++;
+    return ret;
+  }
+
+  public static void clear(byte[] data, int localoffset, int len) {
+    Arrays.fill(data, localoffset, localoffset + len, (byte)0);
+  }
+
+  public static void writeSeqId(byte[] data, int offset, int length, long seqId) {
+    int pos = offset + lengthWithoutMemstore(length);
+    data[pos] = (byte) (seqId >> 56);
+    data[pos + 1] = (byte) (seqId >> 48);
+    data[pos + 2] = (byte) (seqId >> 40);
+    data[pos + 3] = (byte) (seqId >> 32);
+    data[pos + 4] = (byte) (seqId >> 24);
+    data[pos + 5] = (byte) (seqId >> 16);
+    data[pos + 6] = (byte) (seqId >> 8);
+    data[pos + 7] = (byte) (seqId);
+  }
+
+  public static long readSeqId(byte[] data, int offset, int len) {
+    int mo = offset + lengthWithoutMemstore(len);
+    return  (data[mo] & 0xFFL) << 56 | (data[mo + 1] & 0xFFL) << 48
+      | (data[mo + 2] & 0xFFL) << 40 | (data[mo + 3] & 0xFFL) << 32
+      | (data[mo + 4] & 0xFFL) << 24 | (data[mo + 5] & 0xFFL) << 16
+      | (data[mo + 6] & 0xFFL) << 8  | (data[mo + 7] & 0xFFL);
+  }
+
+  public static int lengthWithoutMemstore(int len) {
+    return len - Bytes.SIZEOF_LONG;
+  }
+
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/ccsmap/CompactedCellHelper.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.hbase.ccsmap;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.KVComparator;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Bytes;
+
+@InterfaceAudience.Private
+public class CompactedCellHelper implements CompactedTypeHelper<Cell, Cell> {
+
+  private final KVComparator comparator;
+
+  public CompactedCellHelper(KVComparator c) {
+    this.comparator = c;
+  }
+
+  @Override
+  public int getCompactedSize(Cell key, Cell value) {
+    long sz = KeyValue.getKeyValueDataStructureSize(key.getRowLength(),
+      key.getFamilyLength(), key.getQualifierLength(),
+      key.getValueLength(), key.getTagsLength()) + Bytes.SIZEOF_LONG; // mvcc
+    if (sz > Integer.MAX_VALUE) {
+      throw new RuntimeException("Too big cell");
+    }
+    return (int) sz;
+  }
+
+  @Override
+  public void compact(Cell key, Cell value, byte[] data, int offset, int len) {
+    KeyValue kv = new KeyValue(key);
+    System.arraycopy(kv.getRowArray(), kv.getOffset(), data, offset, kv.getLength());
+    CCSMapUtils.writeSeqId(data, offset, len, key.getSequenceId());
+  }
+
+  @Override
+  public KVPair<Cell, Cell> decomposite(byte[] data, int offset, int len) {
+    KeyValue kv = new OnPageKeyValue(data, offset, len);
+    return new KVPair<Cell, Cell>(kv, kv);
+  }
+
+  @Override
+  public int compare(byte[] ldata, int loffset, int llen, byte[] rdata,
+    int roffset, int rlen) {
+    KVPair<Cell, Cell> lc = decomposite(ldata, loffset, llen);
+    KVPair<Cell, Cell> rc = decomposite(rdata, roffset, rlen);
+    return comparator.compare(lc.key(), rc.key());
+  }
+
+  @Override
+  public int compare(Cell key, byte[] data, int offset, int len) {
+    KVPair<Cell, Cell> rc = decomposite(data, offset, len);
+    return comparator.compare(key, rc.key());
+  }
+
+  @Override
+  public int compare(Cell lkey, Cell rkey) {
+    return comparator.compare(lkey, rkey);
+  }
+
+}
+