+ * 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 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. + *
+ * 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. + *
+ * 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
- * 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.
*
- * 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.
*
- * 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.
*
- * 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 AtomicReferencec. 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 size 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());
+ }
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
index 1230802f4a2..615b8604e30 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreChunkPool.java
@@ -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;
+ 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 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");
+ }
- 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;
}
-
- 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
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
index dc045801b1f..b8353052de8 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
@@ -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.
* c. 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 size 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();
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
index f00a8fea12f..2936f232ada 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
@@ -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());
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
index 0423b37cea9..4b0b5ad8761 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
@@ -29,7 +29,7 @@ import org.apache.hadoop.conf.Configuration;
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