hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoopsamj...@apache.org
Subject svn commit: r1581287 - in /hbase/trunk/hbase-server/src: main/java/org/apache/hadoop/hbase/regionserver/ test/java/org/apache/hadoop/hbase/regionserver/
Date Tue, 25 Mar 2014 09:47:29 GMT
Author: anoopsamjohn
Date: Tue Mar 25 09:47:28 2014
New Revision: 1581287

URL: http://svn.apache.org/r1581287
Log:
HBASE-10750 Pluggable MemStoreLAB.(Anoop)

Added:
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
Modified:
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java?rev=1581287&r1=1581286&r2=1581287&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
Tue Mar 25 09:47:28 2014
@@ -40,11 +40,12 @@ import org.apache.hadoop.hbase.HConstant
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.KeyValueUtil;
 import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.regionserver.MemStoreLAB.Allocation;
+import org.apache.hadoop.hbase.util.ByteRange;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.ClassSize;
 import org.apache.hadoop.hbase.util.CollectionBackedScanner;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.ReflectionUtils;
 
 /**
  * The MemStore holds in-memory modifications to the Store.  Modifications
@@ -65,10 +66,9 @@ import org.apache.hadoop.hbase.util.Envi
 @InterfaceAudience.Private
 public class DefaultMemStore implements MemStore {
   private static final Log LOG = LogFactory.getLog(DefaultMemStore.class);
-
-  static final String USEMSLAB_KEY =
-    "hbase.hregion.memstore.mslab.enabled";
+  static final String USEMSLAB_KEY = "hbase.hregion.memstore.mslab.enabled";
   private static final boolean USEMSLAB_DEFAULT = true;
+  static final String MSLAB_CLASS_NAME = "hbase.regionserver.mslab.class";
 
   private Configuration conf;
 
@@ -94,7 +94,6 @@ public class DefaultMemStore implements 
   TimeRangeTracker timeRangeTracker;
   TimeRangeTracker snapshotTimeRangeTracker;
 
-  MemStoreChunkPool chunkPool;
   volatile MemStoreLAB allocator;
   volatile MemStoreLAB snapshotAllocator;
   volatile long snapshotId;
@@ -121,11 +120,11 @@ public class DefaultMemStore implements 
     this.size = new AtomicLong(DEEP_OVERHEAD);
     this.snapshotSize = 0;
     if (conf.getBoolean(USEMSLAB_KEY, USEMSLAB_DEFAULT)) {
-      this.chunkPool = MemStoreChunkPool.getPool(conf);
-      this.allocator = new MemStoreLAB(conf, chunkPool);
+      String className = conf.get(MSLAB_CLASS_NAME, HeapMemStoreLAB.class.getName());
+      this.allocator = ReflectionUtils.instantiateWithCustomCtor(className,
+          new Class[] { Configuration.class }, new Object[] { conf });
     } else {
       this.allocator = null;
-      this.chunkPool = null;
     }
   }
 
@@ -162,7 +161,9 @@ public class DefaultMemStore implements 
         this.snapshotAllocator = this.allocator;
         // Reset allocator so we get a fresh buffer for the new memstore
         if (allocator != null) {
-          this.allocator = new MemStoreLAB(conf, chunkPool);
+          String className = conf.get(MSLAB_CLASS_NAME, HeapMemStoreLAB.class.getName());
+          this.allocator = ReflectionUtils.instantiateWithCustomCtor(className,
+              new Class[] { Configuration.class }, new Object[] { conf });
         }
         timeOfOldestEdit = Long.MAX_VALUE;
       }
@@ -258,15 +259,15 @@ public class DefaultMemStore implements 
     }
 
     int len = kv.getLength();
-    Allocation alloc = allocator.allocateBytes(len);
+    ByteRange alloc = allocator.allocateBytes(len);
     if (alloc == null) {
       // The allocation was too large, allocator decided
       // not to do anything with it.
       return kv;
     }
-    assert alloc.getData() != null;
-    System.arraycopy(kv.getBuffer(), kv.getOffset(), alloc.getData(), alloc.getOffset(),
len);
-    KeyValue newKv = new KeyValue(alloc.getData(), alloc.getOffset(), len);
+    assert alloc.getBytes() != null;
+    alloc.put(0, kv.getBuffer(), kv.getOffset(), len);
+    KeyValue newKv = new KeyValue(alloc.getBytes(), alloc.getOffset(), len);
     newKv.setMvccVersion(kv.getMvccVersion());
     return newKv;
   }
@@ -987,7 +988,7 @@ public class DefaultMemStore implements 
   }
 
   public final static long FIXED_OVERHEAD = ClassSize.align(
-      ClassSize.OBJECT + (10 * ClassSize.REFERENCE) + (3 * Bytes.SIZEOF_LONG));
+      ClassSize.OBJECT + (9 * ClassSize.REFERENCE) + (3 * Bytes.SIZEOF_LONG));
 
   public final static long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
       ClassSize.ATOMIC_LONG + (2 * ClassSize.TIMERANGE_TRACKER) +

Added: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java?rev=1581287&view=auto
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
(added)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
Tue Mar 25 09:47:28 2014
@@ -0,0 +1,315 @@
+/**
+ *
+ * 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.regionserver;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.util.ByteRange;
+import org.apache.hadoop.hbase.util.SimpleByteRange;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A memstore-local allocation buffer.
+ * <p>
+ * The MemStoreLAB is basically a bump-the-pointer allocator that allocates
+ * big (2MB) byte[] chunks from and then doles it out to threads that request
+ * slices into the array.
+ * <p>
+ * The purpose of this class is to combat heap fragmentation in the
+ * regionserver. By ensuring that all KeyValues in a given memstore refer
+ * only to large chunks of contiguous memory, we ensure that large blocks
+ * get freed up when the memstore is flushed.
+ * <p>
+ * Without the MSLAB, the byte array allocated during insertion end up
+ * interleaved throughout the heap, and the old generation gets progressively
+ * more fragmented until a stop-the-world compacting collection occurs.
+ * <p>
+ * TODO: we should probably benchmark whether word-aligning the allocations
+ * would provide a performance improvement - probably would speed up the
+ * Bytes.toLong/Bytes.toInt calls in KeyValue, but some of those are cached
+ * anyway
+ */
+@InterfaceAudience.Private
+public class HeapMemStoreLAB implements MemStoreLAB {
+
+  static final String CHUNK_SIZE_KEY = "hbase.hregion.memstore.mslab.chunksize";
+  static final int CHUNK_SIZE_DEFAULT = 2048 * 1024;
+  static final String MAX_ALLOC_KEY = "hbase.hregion.memstore.mslab.max.allocation";
+  static final int MAX_ALLOC_DEFAULT = 256 * 1024; // allocs bigger than this don't go through
+                                                   // allocator
+
+  private AtomicReference<Chunk> curChunk = new AtomicReference<Chunk>();
+  // A queue of chunks contained by this memstore
+  private BlockingQueue<Chunk> chunkQueue = new LinkedBlockingQueue<Chunk>();
+  final int chunkSize;
+  final int maxAlloc;
+  private final MemStoreChunkPool chunkPool;
+
+  // This flag is for closing this instance, its set when clearing snapshot of
+  // memstore
+  private volatile boolean closed = false;
+  // This flag is for reclaiming chunks. Its set when putting chunks back to
+  // pool
+  private AtomicBoolean reclaimed = new AtomicBoolean(false);
+  // Current count of open scanners which reading data from this MemStoreLAB
+  private final AtomicInteger openScannerCount = new AtomicInteger();
+
+  // Used in testing
+  public HeapMemStoreLAB() {
+    this(new Configuration());
+  }
+
+  public HeapMemStoreLAB(Configuration conf) {
+    chunkSize = conf.getInt(CHUNK_SIZE_KEY, CHUNK_SIZE_DEFAULT);
+    maxAlloc = conf.getInt(MAX_ALLOC_KEY, MAX_ALLOC_DEFAULT);
+    this.chunkPool = MemStoreChunkPool.getPool(conf);
+
+    // if we don't exclude allocations >CHUNK_SIZE, we'd infiniteloop on one!
+    Preconditions.checkArgument(
+      maxAlloc <= chunkSize,
+      MAX_ALLOC_KEY + " must be less than " + CHUNK_SIZE_KEY);
+  }
+
+  /**
+   * Allocate a slice of the given length.
+   *
+   * If the size is larger than the maximum size specified for this
+   * allocator, returns null.
+   */
+  @Override
+  public ByteRange allocateBytes(int size) {
+    Preconditions.checkArgument(size >= 0, "negative size");
+
+    // Callers should satisfy large allocations directly from JVM since they
+    // don't cause fragmentation as badly.
+    if (size > maxAlloc) {
+      return null;
+    }
+
+    while (true) {
+      Chunk c = getOrMakeChunk();
+
+      // Try to allocate from this chunk
+      int allocOffset = c.alloc(size);
+      if (allocOffset != -1) {
+        // We succeeded - this is the common case - small alloc
+        // from a big buffer
+        return new SimpleByteRange(c.data, allocOffset, size);
+      }
+
+      // not enough space!
+      // try to retire this chunk
+      tryRetireChunk(c);
+    }
+  }
+
+  /**
+   * Close this instance since it won't be used any more, try to put the chunks
+   * back to pool
+   */
+  @Override
+  public void close() {
+    this.closed = true;
+    // We could put back the chunks to pool for reusing only when there is no
+    // opening scanner which will read their data
+    if (chunkPool != null && openScannerCount.get() == 0
+        && reclaimed.compareAndSet(false, true)) {
+      chunkPool.putbackChunks(this.chunkQueue);
+    }
+  }
+
+  /**
+   * Called when opening a scanner on the data of this MemStoreLAB
+   */
+  @Override
+  public void incScannerCount() {
+    this.openScannerCount.incrementAndGet();
+  }
+
+  /**
+   * Called when closing a scanner on the data of this MemStoreLAB
+   */
+  @Override
+  public void decScannerCount() {
+    int count = this.openScannerCount.decrementAndGet();
+    if (chunkPool != null && count == 0 && this.closed
+        && reclaimed.compareAndSet(false, true)) {
+      chunkPool.putbackChunks(this.chunkQueue);
+    }
+  }
+
+  /**
+   * Try to retire the current chunk if it is still
+   * <code>c</code>. Postcondition is that curChunk.get()
+   * != c
+   */
+  private void tryRetireChunk(Chunk c) {
+    curChunk.compareAndSet(c, null);
+    // If the CAS succeeds, that means that we won the race
+    // to retire the chunk. We could use this opportunity to
+    // update metrics on external fragmentation.
+    //
+    // If the CAS fails, that means that someone else already
+    // retired the chunk for us.
+  }
+
+  /**
+   * Get the current chunk, or, if there is no current chunk,
+   * allocate a new one from the JVM.
+   */
+  private Chunk getOrMakeChunk() {
+    while (true) {
+      // Try to get the chunk
+      Chunk c = curChunk.get();
+      if (c != null) {
+        return c;
+      }
+
+      // No current chunk, so we want to allocate one. We race
+      // against other allocators to CAS in an uninitialized chunk
+      // (which is cheap to allocate)
+      c = (chunkPool != null) ? chunkPool.getChunk() : new Chunk(chunkSize);
+      if (curChunk.compareAndSet(null, c)) {
+        // we won race - now we need to actually do the expensive
+        // allocation step
+        c.init();
+        this.chunkQueue.add(c);
+        return c;
+      } else if (chunkPool != null) {
+        chunkPool.putbackChunk(c);
+      }
+      // someone else won race - that's fine, we'll try to grab theirs
+      // in the next iteration of the loop.
+    }
+  }
+
+  /**
+   * A chunk of memory out of which allocations are sliced.
+   */
+  static class Chunk {
+    /** Actual underlying data */
+    private byte[] data;
+
+    private static final int UNINITIALIZED = -1;
+    private static final int OOM = -2;
+    /**
+     * Offset for the next allocation, or the sentinel value -1
+     * which implies that the chunk is still uninitialized.
+     * */
+    private AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED);
+
+    /** Total number of allocations satisfied from this buffer */
+    private AtomicInteger allocCount = new AtomicInteger();
+
+    /** Size of chunk in bytes */
+    private final int size;
+
+    /**
+     * Create an uninitialized chunk. Note that memory is not allocated yet, so
+     * this is cheap.
+     * @param size in bytes
+     */
+    Chunk(int size) {
+      this.size = size;
+    }
+
+    /**
+     * Actually claim the memory for this chunk. This should only be called from
+     * the thread that constructed the chunk. It is thread-safe against other
+     * threads calling alloc(), who will block until the allocation is complete.
+     */
+    public void init() {
+      assert nextFreeOffset.get() == UNINITIALIZED;
+      try {
+        if (data == null) {
+          data = new byte[size];
+        }
+      } catch (OutOfMemoryError e) {
+        boolean failInit = nextFreeOffset.compareAndSet(UNINITIALIZED, OOM);
+        assert failInit; // should be true.
+        throw e;
+      }
+      // Mark that it's ready for use
+      boolean initted = nextFreeOffset.compareAndSet(
+          UNINITIALIZED, 0);
+      // We should always succeed the above CAS since only one thread
+      // calls init()!
+      Preconditions.checkState(initted,
+          "Multiple threads tried to init same chunk");
+    }
+
+    /**
+     * Reset the offset to UNINITIALIZED before before reusing an old chunk
+     */
+    void reset() {
+      if (nextFreeOffset.get() != UNINITIALIZED) {
+        nextFreeOffset.set(UNINITIALIZED);
+        allocCount.set(0);
+      }
+    }
+
+    /**
+     * Try to allocate <code>size</code> bytes from the chunk.
+     * @return the offset of the successful allocation, or -1 to indicate not-enough-space
+     */
+    public int alloc(int size) {
+      while (true) {
+        int oldOffset = nextFreeOffset.get();
+        if (oldOffset == UNINITIALIZED) {
+          // The chunk doesn't have its data allocated yet.
+          // Since we found this in curChunk, we know that whoever
+          // CAS-ed it there is allocating it right now. So spin-loop
+          // shouldn't spin long!
+          Thread.yield();
+          continue;
+        }
+        if (oldOffset == OOM) {
+          // doh we ran out of ram. return -1 to chuck this away.
+          return -1;
+        }
+
+        if (oldOffset + size > data.length) {
+          return -1; // alloc doesn't fit
+        }
+
+        // Try to atomically claim this chunk
+        if (nextFreeOffset.compareAndSet(oldOffset, oldOffset + size)) {
+          // we got the alloc
+          allocCount.incrementAndGet();
+          return oldOffset;
+        }
+        // we raced and lost alloc, try again
+      }
+    }
+
+    @Override
+    public String toString() {
+      return "Chunk@" + System.identityHashCode(this) +
+        " allocs=" + allocCount.get() + "waste=" +
+        (data.length - nextFreeOffset.get());
+    }
+  }
+}

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java?rev=1581287&r1=1581286&r2=1581287&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
Tue Mar 25 09:47:28 2014
@@ -30,13 +30,13 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.regionserver.MemStoreLAB.Chunk;
+import org.apache.hadoop.hbase.regionserver.HeapMemStoreLAB.Chunk;
 import org.apache.hadoop.util.StringUtils;
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 /**
- * A pool of {@link MemStoreLAB$Chunk} instances.
+ * A pool of {@link HeapMemStoreLAB$Chunk} instances.
  * 
  * MemStoreChunkPool caches a number of retired chunks for reusing, it could
  * decrease allocating bytes when writing, thereby optimizing the garbage
@@ -177,42 +177,39 @@ public class MemStoreChunkPool {
    * @param conf
    * @return the global MemStoreChunkPool instance
    */
-  static synchronized MemStoreChunkPool getPool(Configuration conf) {
+  static MemStoreChunkPool getPool(Configuration conf) {
     if (globalInstance != null) return globalInstance;
     if (chunkPoolDisabled) return null;
 
-
-    float poolSizePercentage = conf.getFloat(CHUNK_POOL_MAXSIZE_KEY,
-        POOL_MAX_SIZE_DEFAULT);
-    if (poolSizePercentage <= 0) {
-      chunkPoolDisabled = true;
-      return null;
-    }
-    if (poolSizePercentage > 1.0) {
-      throw new IllegalArgumentException(CHUNK_POOL_MAXSIZE_KEY
-          + " must be between 0.0 and 1.0");
-    }
-    long heapMax = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage()
-        .getMax();
-    long globalMemStoreLimit = (long) (heapMax * MemStoreFlusher.getGlobalMemStorePercent(conf));
-    int chunkSize = conf.getInt(MemStoreLAB.CHUNK_SIZE_KEY,
-        MemStoreLAB.CHUNK_SIZE_DEFAULT);
-    int maxCount = (int) (globalMemStoreLimit * poolSizePercentage / chunkSize);
-
-    float initialCountPercentage = conf.getFloat(CHUNK_POOL_INITIALSIZE_KEY,
-        POOL_INITIAL_SIZE_DEFAULT);
-    if (initialCountPercentage > 1.0 || initialCountPercentage < 0) {
-      throw new IllegalArgumentException(CHUNK_POOL_INITIALSIZE_KEY
-          + " must be between 0.0 and 1.0");
-    }
-
-    int initialCount = (int) (initialCountPercentage * maxCount);
-    LOG.info("Allocating MemStoreChunkPool with chunk size "
-        + StringUtils.byteDesc(chunkSize) + ", max count " + maxCount
-        + ", initial count " + initialCount);
-    globalInstance = new MemStoreChunkPool(conf, chunkSize, maxCount,
-        initialCount);
-    return globalInstance;
+    synchronized (MemStoreChunkPool.class) {
+      if (globalInstance != null) return globalInstance;
+      float poolSizePercentage = conf.getFloat(CHUNK_POOL_MAXSIZE_KEY, POOL_MAX_SIZE_DEFAULT);
+      if (poolSizePercentage <= 0) {
+        chunkPoolDisabled = true;
+        return null;
+      }
+      if (poolSizePercentage > 1.0) {
+        throw new IllegalArgumentException(CHUNK_POOL_MAXSIZE_KEY + " must be between 0.0
and 1.0");
+      }
+      long heapMax = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax();
+      long globalMemStoreLimit = (long) (heapMax * MemStoreFlusher.getGlobalMemStorePercent(conf));
+      int chunkSize = conf.getInt(HeapMemStoreLAB.CHUNK_SIZE_KEY,
+          HeapMemStoreLAB.CHUNK_SIZE_DEFAULT);
+      int maxCount = (int) (globalMemStoreLimit * poolSizePercentage / chunkSize);
+
+      float initialCountPercentage = conf.getFloat(CHUNK_POOL_INITIALSIZE_KEY,
+          POOL_INITIAL_SIZE_DEFAULT);
+      if (initialCountPercentage > 1.0 || initialCountPercentage < 0) {
+        throw new IllegalArgumentException(CHUNK_POOL_INITIALSIZE_KEY
+            + " must be between 0.0 and 1.0");
+      }
+
+      int initialCount = (int) (initialCountPercentage * maxCount);
+      LOG.info("Allocating MemStoreChunkPool with chunk size " + StringUtils.byteDesc(chunkSize)
+          + ", max count " + maxCount + ", initial count " + initialCount);
+      globalInstance = new MemStoreChunkPool(conf, chunkSize, maxCount, initialCount);
+      return globalInstance;
+    }
   }
 
 }
\ No newline at end of file

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java?rev=1581287&r1=1581286&r2=1581287&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
Tue Mar 25 09:47:28 2014
@@ -1,5 +1,4 @@
 /**
- *
  * 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
@@ -18,325 +17,47 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
 import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.conf.Configuration;
-
-import com.google.common.base.Preconditions;
+import org.apache.hadoop.hbase.util.ByteRange;
 
 /**
  * A memstore-local allocation buffer.
  * <p>
- * The MemStoreLAB is basically a bump-the-pointer allocator that allocates
- * big (2MB) byte[] chunks from and then doles it out to threads that request
- * slices into the array.
+ * The MemStoreLAB is basically a bump-the-pointer allocator that allocates big (2MB) chunks
from
+ * and then doles it out to threads that request slices into the array.
  * <p>
- * The purpose of this class is to combat heap fragmentation in the
- * regionserver. By ensuring that all KeyValues in a given memstore refer
- * only to large chunks of contiguous memory, we ensure that large blocks
- * get freed up when the memstore is flushed.
+ * The purpose of this is to combat heap fragmentation in the regionserver. By ensuring that
all
+ * KeyValues in a given memstore refer only to large chunks of contiguous memory, we ensure
that
+ * large blocks get freed up when the memstore is flushed.
  * <p>
- * Without the MSLAB, the byte array allocated during insertion end up
- * interleaved throughout the heap, and the old generation gets progressively
- * more fragmented until a stop-the-world compacting collection occurs.
+ * Without the MSLAB, the byte array allocated during insertion end up interleaved throughout
the
+ * heap, and the old generation gets progressively more fragmented until a stop-the-world
compacting
+ * collection occurs.
  * <p>
- * TODO: we should probably benchmark whether word-aligning the allocations
- * would provide a performance improvement - probably would speed up the
- * Bytes.toLong/Bytes.toInt calls in KeyValue, but some of those are cached
- * anyway
  */
 @InterfaceAudience.Private
-public class MemStoreLAB {
-  private AtomicReference<Chunk> curChunk = new AtomicReference<Chunk>();
-  // A queue of chunks contained by this memstore
-  private BlockingQueue<Chunk> chunkQueue = new LinkedBlockingQueue<Chunk>();
-
-  final static String CHUNK_SIZE_KEY = "hbase.hregion.memstore.mslab.chunksize";
-  final static int CHUNK_SIZE_DEFAULT = 2048 * 1024;
-  final int chunkSize;
-
-  final static String MAX_ALLOC_KEY = "hbase.hregion.memstore.mslab.max.allocation";
-  final static int MAX_ALLOC_DEFAULT = 256  * 1024; // allocs bigger than this don't go through
allocator
-  final int maxAlloc;
-
-  private final MemStoreChunkPool chunkPool;
-
-  // This flag is for closing this instance, its set when clearing snapshot of
-  // memstore
-  private volatile boolean closed = false;
-  // This flag is for reclaiming chunks. Its set when putting chunks back to
-  // pool
-  private AtomicBoolean reclaimed = new AtomicBoolean(false);
-  // Current count of open scanners which reading data from this MemStoreLAB
-  private final AtomicInteger openScannerCount = new AtomicInteger();
-
-  // Used in testing
-  public MemStoreLAB() {
-    this(new Configuration());
-  }
-
-  private MemStoreLAB(Configuration conf) {
-    this(conf, MemStoreChunkPool.getPool(conf));
-  }
-
-  public MemStoreLAB(Configuration conf, MemStoreChunkPool pool) {
-    chunkSize = conf.getInt(CHUNK_SIZE_KEY, CHUNK_SIZE_DEFAULT);
-    maxAlloc = conf.getInt(MAX_ALLOC_KEY, MAX_ALLOC_DEFAULT);
-    this.chunkPool = pool;
-
-    // if we don't exclude allocations >CHUNK_SIZE, we'd infiniteloop on one!
-    Preconditions.checkArgument(
-      maxAlloc <= chunkSize,
-      MAX_ALLOC_KEY + " must be less than " + CHUNK_SIZE_KEY);
-  }
+public interface MemStoreLAB {
 
   /**
-   * Allocate a slice of the given length.
-   *
-   * If the size is larger than the maximum size specified for this
-   * allocator, returns null.
+   * Allocate a slice of the given length. If the size is larger than the maximum size specified
for
+   * this allocator, returns null.
+   * @param size
+   * @return {@link ByteRange}
    */
-  public Allocation allocateBytes(int size) {
-    Preconditions.checkArgument(size >= 0, "negative size");
-
-    // Callers should satisfy large allocations directly from JVM since they
-    // don't cause fragmentation as badly.
-    if (size > maxAlloc) {
-      return null;
-    }
-
-    while (true) {
-      Chunk c = getOrMakeChunk();
-
-      // Try to allocate from this chunk
-      int allocOffset = c.alloc(size);
-      if (allocOffset != -1) {
-        // We succeeded - this is the common case - small alloc
-        // from a big buffer
-        return new Allocation(c.data, allocOffset);
-      }
-
-      // not enough space!
-      // try to retire this chunk
-      tryRetireChunk(c);
-    }
-  }
+  ByteRange allocateBytes(int size);
 
   /**
-   * Close this instance since it won't be used any more, try to put the chunks
-   * back to pool
+   * Close instance since it won't be used any more, try to put the chunks back to pool
    */
-  void close() {
-    this.closed = true;
-    // We could put back the chunks to pool for reusing only when there is no
-    // opening scanner which will read their data
-    if (chunkPool != null && openScannerCount.get() == 0
-        && reclaimed.compareAndSet(false, true)) {
-      chunkPool.putbackChunks(this.chunkQueue);
-    }
-  }
+  void close();
 
   /**
    * Called when opening a scanner on the data of this MemStoreLAB
    */
-  void incScannerCount() {
-    this.openScannerCount.incrementAndGet();
-  }
+  void incScannerCount();
 
   /**
    * Called when closing a scanner on the data of this MemStoreLAB
    */
-  void decScannerCount() {
-    int count = this.openScannerCount.decrementAndGet();
-    if (chunkPool != null && count == 0 && this.closed
-        && reclaimed.compareAndSet(false, true)) {
-      chunkPool.putbackChunks(this.chunkQueue);
-    }
-  }
-
-  /**
-   * Try to retire the current chunk if it is still
-   * <code>c</code>. Postcondition is that curChunk.get()
-   * != c
-   */
-  private void tryRetireChunk(Chunk c) {
-    curChunk.compareAndSet(c, null);
-    // If the CAS succeeds, that means that we won the race
-    // to retire the chunk. We could use this opportunity to
-    // update metrics on external fragmentation.
-    //
-    // If the CAS fails, that means that someone else already
-    // retired the chunk for us.
-  }
-
-  /**
-   * Get the current chunk, or, if there is no current chunk,
-   * allocate a new one from the JVM.
-   */
-  private Chunk getOrMakeChunk() {
-    while (true) {
-      // Try to get the chunk
-      Chunk c = curChunk.get();
-      if (c != null) {
-        return c;
-      }
-
-      // No current chunk, so we want to allocate one. We race
-      // against other allocators to CAS in an uninitialized chunk
-      // (which is cheap to allocate)
-      c = (chunkPool != null) ? chunkPool.getChunk() : new Chunk(chunkSize);
-      if (curChunk.compareAndSet(null, c)) {
-        // we won race - now we need to actually do the expensive
-        // allocation step
-        c.init();
-        this.chunkQueue.add(c);
-        return c;
-      } else if (chunkPool != null) {
-        chunkPool.putbackChunk(c);
-      }
-      // someone else won race - that's fine, we'll try to grab theirs
-      // in the next iteration of the loop.
-    }
-  }
-
-  /**
-   * A chunk of memory out of which allocations are sliced.
-   */
-  static class Chunk {
-    /** Actual underlying data */
-    private byte[] data;
-
-    private static final int UNINITIALIZED = -1;
-    private static final int OOM = -2;
-    /**
-     * Offset for the next allocation, or the sentinel value -1
-     * which implies that the chunk is still uninitialized.
-     * */
-    private AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED);
-
-    /** Total number of allocations satisfied from this buffer */
-    private AtomicInteger allocCount = new AtomicInteger();
-
-    /** Size of chunk in bytes */
-    private final int size;
-
-    /**
-     * Create an uninitialized chunk. Note that memory is not allocated yet, so
-     * this is cheap.
-     * @param size in bytes
-     */
-    Chunk(int size) {
-      this.size = size;
-    }
-
-    /**
-     * Actually claim the memory for this chunk. This should only be called from
-     * the thread that constructed the chunk. It is thread-safe against other
-     * threads calling alloc(), who will block until the allocation is complete.
-     */
-    public void init() {
-      assert nextFreeOffset.get() == UNINITIALIZED;
-      try {
-        if (data == null) {
-          data = new byte[size];
-        }
-      } catch (OutOfMemoryError e) {
-        boolean failInit = nextFreeOffset.compareAndSet(UNINITIALIZED, OOM);
-        assert failInit; // should be true.
-        throw e;
-      }
-      // Mark that it's ready for use
-      boolean initted = nextFreeOffset.compareAndSet(
-          UNINITIALIZED, 0);
-      // We should always succeed the above CAS since only one thread
-      // calls init()!
-      Preconditions.checkState(initted,
-          "Multiple threads tried to init same chunk");
-    }
-
-    /**
-     * Reset the offset to UNINITIALIZED before before reusing an old chunk
-     */
-    void reset() {
-      if (nextFreeOffset.get() != UNINITIALIZED) {
-        nextFreeOffset.set(UNINITIALIZED);
-        allocCount.set(0);
-      }
-    }
-
-    /**
-     * Try to allocate <code>size</code> bytes from the chunk.
-     * @return the offset of the successful allocation, or -1 to indicate not-enough-space
-     */
-    public int alloc(int size) {
-      while (true) {
-        int oldOffset = nextFreeOffset.get();
-        if (oldOffset == UNINITIALIZED) {
-          // The chunk doesn't have its data allocated yet.
-          // Since we found this in curChunk, we know that whoever
-          // CAS-ed it there is allocating it right now. So spin-loop
-          // shouldn't spin long!
-          Thread.yield();
-          continue;
-        }
-        if (oldOffset == OOM) {
-          // doh we ran out of ram. return -1 to chuck this away.
-          return -1;
-        }
-
-        if (oldOffset + size > data.length) {
-          return -1; // alloc doesn't fit
-        }
-
-        // Try to atomically claim this chunk
-        if (nextFreeOffset.compareAndSet(oldOffset, oldOffset + size)) {
-          // we got the alloc
-          allocCount.incrementAndGet();
-          return oldOffset;
-        }
-        // we raced and lost alloc, try again
-      }
-    }
-
-    @Override
-    public String toString() {
-      return "Chunk@" + System.identityHashCode(this) +
-        " allocs=" + allocCount.get() + "waste=" +
-        (data.length - nextFreeOffset.get());
-    }
-  }
-
-  /**
-   * The result of a single allocation. Contains the chunk that the
-   * allocation points into, and the offset in this array where the
-   * slice begins.
-   */
-  public static class Allocation {
-    private final byte[] data;
-    private final int offset;
-
-    private Allocation(byte[] data, int off) {
-      this.data = data;
-      this.offset = off;
-    }
-
-    @Override
-    public String toString() {
-      return "Allocation(" + "capacity=" + data.length + ", off=" + offset
-          + ")";
-    }
-
-    byte[] getData() {
-      return data;
-    }
-
-    int getOffset() {
-      return offset;
-    }
-  }
+  void decScannerCount();
 }

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java?rev=1581287&r1=1581286&r2=1581287&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
Tue Mar 25 09:47:28 2014
@@ -28,7 +28,7 @@ import java.util.Random;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.SmallTests;
-import org.apache.hadoop.hbase.regionserver.MemStoreLAB.Allocation;
+import org.apache.hadoop.hbase.util.ByteRange;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -68,21 +68,21 @@ public class TestMemStoreChunkPool {
   @Test
   public void testReusingChunks() {
     Random rand = new Random();
-    MemStoreLAB mslab = new MemStoreLAB(conf, chunkPool);
+    MemStoreLAB mslab = new HeapMemStoreLAB(conf);
     int expectedOff = 0;
     byte[] lastBuffer = null;
     // Randomly allocate some bytes
     for (int i = 0; i < 100; i++) {
       int size = rand.nextInt(1000);
-      Allocation alloc = mslab.allocateBytes(size);
+      ByteRange alloc = mslab.allocateBytes(size);
 
-      if (alloc.getData() != lastBuffer) {
+      if (alloc.getBytes() != lastBuffer) {
         expectedOff = 0;
-        lastBuffer = alloc.getData();
+        lastBuffer = alloc.getBytes();
       }
       assertEquals(expectedOff, alloc.getOffset());
-      assertTrue("Allocation " + alloc + " overruns buffer", alloc.getOffset()
-          + size <= alloc.getData().length);
+      assertTrue("Allocation overruns buffer", alloc.getOffset()
+          + size <= alloc.getBytes().length);
       expectedOff += size;
     }
     // chunks will be put back to pool after close
@@ -90,7 +90,7 @@ public class TestMemStoreChunkPool {
     int chunkCount = chunkPool.getPoolSize();
     assertTrue(chunkCount > 0);
     // reconstruct mslab
-    mslab = new MemStoreLAB(conf, chunkPool);
+    mslab = new HeapMemStoreLAB(conf);
     // chunk should be got from the pool, so we can reuse it.
     mslab.allocateBytes(1000);
     assertEquals(chunkCount - 1, chunkPool.getPoolSize());

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java?rev=1581287&r1=1581286&r2=1581287&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
Tue Mar 25 09:47:28 2014
@@ -29,7 +29,7 @@ import org.apache.hadoop.conf.Configurat
 import org.apache.hadoop.hbase.MultithreadedTestUtil;
 import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
 import org.apache.hadoop.hbase.SmallTests;
-import org.apache.hadoop.hbase.regionserver.MemStoreLAB.Allocation;
+import org.apache.hadoop.hbase.util.ByteRange;
 import org.junit.Test;
 
 import com.google.common.collect.Iterables;
@@ -47,7 +47,7 @@ public class TestMemStoreLAB {
   @Test
   public void testLABRandomAllocation() {
     Random rand = new Random();
-    MemStoreLAB mslab = new MemStoreLAB();
+    MemStoreLAB mslab = new HeapMemStoreLAB();
     int expectedOff = 0;
     byte[] lastBuffer = null;
     // 100K iterations by 0-1K alloc -> 50MB expected
@@ -55,23 +55,23 @@ public class TestMemStoreLAB {
     // behavior
     for (int i = 0; i < 100000; i++) {
       int size = rand.nextInt(1000);
-      Allocation alloc = mslab.allocateBytes(size);
+      ByteRange alloc = mslab.allocateBytes(size);
       
-      if (alloc.getData() != lastBuffer) {
+      if (alloc.getBytes() != lastBuffer) {
         expectedOff = 0;
-        lastBuffer = alloc.getData();
+        lastBuffer = alloc.getBytes();
       }
       assertEquals(expectedOff, alloc.getOffset());
-      assertTrue("Allocation " + alloc + " overruns buffer",
-          alloc.getOffset() + size <= alloc.getData().length);
+      assertTrue("Allocation overruns buffer",
+          alloc.getOffset() + size <= alloc.getBytes().length);
       expectedOff += size;
     }
   }
 
   @Test
   public void testLABLargeAllocation() {
-    MemStoreLAB mslab = new MemStoreLAB();
-    Allocation alloc = mslab.allocateBytes(2*1024*1024);
+    MemStoreLAB mslab = new HeapMemStoreLAB();
+    ByteRange alloc = mslab.allocateBytes(2*1024*1024);
     assertNull("2MB allocation shouldn't be satisfied by LAB.",
       alloc);
   } 
@@ -88,7 +88,7 @@ public class TestMemStoreLAB {
     
     final AtomicInteger totalAllocated = new AtomicInteger();
     
-    final MemStoreLAB mslab = new MemStoreLAB();
+    final MemStoreLAB mslab = new HeapMemStoreLAB();
     List<List<AllocRecord>> allocations = Lists.newArrayList();
     
     for (int i = 0; i < 10; i++) {
@@ -100,7 +100,7 @@ public class TestMemStoreLAB {
         @Override
         public void doAnAction() throws Exception {
           int size = r.nextInt(1000);
-          Allocation alloc = mslab.allocateBytes(size);
+          ByteRange alloc = mslab.allocateBytes(size);
           totalAllocated.addAndGet(size);
           allocsByThisThread.add(new AllocRecord(alloc, size));
         }
@@ -125,10 +125,10 @@ public class TestMemStoreLAB {
       if (rec.size == 0) continue;
       
       Map<Integer, AllocRecord> mapForThisByteArray =
-        mapsByChunk.get(rec.alloc.getData());
+        mapsByChunk.get(rec.alloc.getBytes());
       if (mapForThisByteArray == null) {
         mapForThisByteArray = Maps.newTreeMap();
-        mapsByChunk.put(rec.alloc.getData(), mapForThisByteArray);
+        mapsByChunk.put(rec.alloc.getBytes(), mapForThisByteArray);
       }
       AllocRecord oldVal = mapForThisByteArray.put(rec.alloc.getOffset(), rec);
       assertNull("Already had an entry " + oldVal + " for allocation " + rec,
@@ -141,8 +141,8 @@ public class TestMemStoreLAB {
       int expectedOff = 0;
       for (AllocRecord alloc : allocsInChunk.values()) {
         assertEquals(expectedOff, alloc.alloc.getOffset());
-        assertTrue("Allocation " + alloc + " overruns buffer",
-            alloc.alloc.getOffset() + alloc.size <= alloc.alloc.getData().length);
+        assertTrue("Allocation overruns buffer",
+            alloc.alloc.getOffset() + alloc.size <= alloc.alloc.getBytes().length);
         expectedOff += alloc.size;
       }
     }
@@ -150,9 +150,9 @@ public class TestMemStoreLAB {
   }
   
   private static class AllocRecord implements Comparable<AllocRecord>{
-    private final Allocation alloc;
+    private final ByteRange alloc;
     private final int size;
-    public AllocRecord(Allocation alloc, int size) {
+    public AllocRecord(ByteRange alloc, int size) {
       super();
       this.alloc = alloc;
       this.size = size;
@@ -160,7 +160,7 @@ public class TestMemStoreLAB {
 
     @Override
     public int compareTo(AllocRecord e) {
-      if (alloc.getData() != e.alloc.getData()) {
+      if (alloc.getBytes() != e.alloc.getBytes()) {
         throw new RuntimeException("Can only compare within a particular array");
       }
       return Ints.compare(alloc.getOffset(), e.alloc.getOffset());
@@ -168,7 +168,7 @@ public class TestMemStoreLAB {
     
     @Override
     public String toString() {
-      return "AllocRecord(alloc=" + alloc + ", size=" + size + ")";
+      return "AllocRecord(offset=" + alloc.getOffset() + ", size=" + size + ")";
     }
     
   }



Mime
View raw message