Author: todd
Date: Mon Feb 7 21:40:28 2011
New Revision: 1068150
URL: http://svn.apache.org/viewvc?rev=1068150&view=rev
Log:
HBASE-3455. Add memstore-local allocation buffers to combat heap fragmentation
in the region server.
Added:
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
Modified:
hbase/branches/0.90/CHANGES.txt
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
hbase/branches/0.90/src/main/resources/hbase-default.xml
hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStore.java
Modified: hbase/branches/0.90/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/CHANGES.txt?rev=1068150&r1=1068149&r2=1068150&view=diff
==============================================================================
--- hbase/branches/0.90/CHANGES.txt (original)
+++ hbase/branches/0.90/CHANGES.txt Mon Feb 7 21:40:28 2011
@@ -1,5 +1,11 @@
HBase Change Log
Release 0.90.1 - Unreleased
+
+ NEW FEATURES
+ HBASE-3455 Add memstore-local allocation buffers to combat heap
+ fragmentation in the region server. Experimental / disabled
+ by default in 0.90.1
+
BUG FIXES
HBASE-3387 Revert 'Pair does not deep check arrays for equality'
HBASE-3445 Master crashes on data that was moved from different host
Modified:
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java?rev=1068150&r1=1068149&r2=1068150&view=diff
==============================================================================
---
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
(original)
+++
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
Mon Feb 7 21:40:28 2011
@@ -34,10 +34,13 @@ import java.util.concurrent.locks.Reentr
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.HeapSize;
+import org.apache.hadoop.hbase.regionserver.MemStoreLAB.Allocation;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
@@ -55,6 +58,13 @@ import org.apache.hadoop.hbase.util.Clas
public class MemStore implements HeapSize {
private static final Log LOG = LogFactory.getLog(MemStore.class);
+ static final String USEMSLAB_KEY =
+ "hbase.hregion.memstore.mslab.enabled";
+ private static final boolean USEMSLAB_DEFAULT = false;
+
+
+ private Configuration conf;
+
// MemStore. Use a KeyValueSkipListSet rather than SkipListSet because of
the
// better semantics. The Map will overwrite if passed a key it already had
// whereas the Set will not add new KV if key is same though value might be
@@ -80,19 +90,23 @@ public class MemStore implements HeapSiz
TimeRangeTracker timeRangeTracker;
TimeRangeTracker snapshotTimeRangeTracker;
+
+ MemStoreLAB allocator;
/**
* Default constructor. Used for tests.
*/
public MemStore() {
- this(KeyValue.COMPARATOR);
+ this(HBaseConfiguration.create(), KeyValue.COMPARATOR);
}
/**
* Constructor.
* @param c Comparator
*/
- public MemStore(final KeyValue.KVComparator c) {
+ public MemStore(final Configuration conf,
+ final KeyValue.KVComparator c) {
+ this.conf = conf;
this.comparator = c;
this.comparatorIgnoreTimestamp =
this.comparator.getComparatorIgnoringTimestamps();
@@ -102,6 +116,11 @@ public class MemStore implements HeapSiz
timeRangeTracker = new TimeRangeTracker();
snapshotTimeRangeTracker = new TimeRangeTracker();
this.size = new AtomicLong(DEEP_OVERHEAD);
+ if (conf.getBoolean(USEMSLAB_KEY, USEMSLAB_DEFAULT)) {
+ this.allocator = new MemStoreLAB(conf);
+ } else {
+ this.allocator = null;
+ }
}
void dump() {
@@ -134,6 +153,10 @@ public class MemStore implements HeapSiz
this.timeRangeTracker = new TimeRangeTracker();
// Reset heap to not include any keys
this.size.set(DEEP_OVERHEAD);
+ // Reset allocator so we get a fresh buffer for the new memstore
+ if (allocator != null) {
+ this.allocator = new MemStoreLAB(conf);
+ }
}
}
} finally {
@@ -184,18 +207,47 @@ public class MemStore implements HeapSiz
* @return approximate size of the passed key and value.
*/
long add(final KeyValue kv) {
- long s = -1;
this.lock.readLock().lock();
try {
- s = heapSizeChange(kv, this.kvset.add(kv));
- timeRangeTracker.includeTimestamp(kv);
- this.size.addAndGet(s);
+ KeyValue toAdd = maybeCloneWithAllocator(kv);
+ return internalAdd(toAdd);
} finally {
this.lock.readLock().unlock();
}
+ }
+
+ /**
+ * Internal version of add() that doesn't clone KVs with the
+ * allocator, and doesn't take the lock.
+ *
+ * Callers should ensure they already have the read lock taken
+ */
+ private long internalAdd(final KeyValue toAdd) {
+ long s = heapSizeChange(toAdd, this.kvset.add(toAdd));
+ timeRangeTracker.includeTimestamp(toAdd);
+ this.size.addAndGet(s);
return s;
}
+ private KeyValue maybeCloneWithAllocator(KeyValue kv) {
+ if (allocator == null) {
+ return kv;
+ }
+
+ int len = kv.getLength();
+ Allocation alloc = allocator.allocateBytes(len);
+ if (alloc == null) {
+ // The allocation was too large, allocator decided
+ // not to do anything with it.
+ return kv;
+ }
+ assert alloc != null && alloc.getData() != null;
+ System.arraycopy(kv.getBuffer(), kv.getOffset(), alloc.getData(),
alloc.getOffset(), len);
+ KeyValue newKv = new KeyValue(alloc.getData(), alloc.getOffset(), len);
+ newKv.setMemstoreTS(kv.getMemstoreTS());
+ return newKv;
+ }
+
/**
* Write a delete
* @param delete
@@ -205,8 +257,9 @@ public class MemStore implements HeapSiz
long s = 0;
this.lock.readLock().lock();
try {
- s += heapSizeChange(delete, this.kvset.add(delete));
- timeRangeTracker.includeTimestamp(delete);
+ KeyValue toAdd = maybeCloneWithAllocator(delete);
+ s += heapSizeChange(toAdd, this.kvset.add(toAdd));
+ timeRangeTracker.includeTimestamp(toAdd);
} finally {
this.lock.readLock().unlock();
}
@@ -459,12 +512,20 @@ public class MemStore implements HeapSiz
* <p>
* If there are any existing KeyValues in this MemStore with the same row,
* family, and qualifier, they are removed.
+ * <p>
+ * Callers must hold the read lock.
+ *
* @param kv
* @return change in size of MemStore
*/
private long upsert(KeyValue kv) {
// Add the KeyValue to the MemStore
- long addedSize = add(kv);
+ // Use the internalAdd method here since we (a) already have a lock
+ // and (b) cannot safely use the MSLAB here without potentially
+ // hitting OOME - see TestMemStore.testUpsertMSLAB for a
+ // test that triggers the pathological case if we don't avoid MSLAB
+ // here.
+ long addedSize = internalAdd(kv);
// Get the KeyValues for the row/family/qualifier regardless of timestamp.
// For this case we want to clean up any other puts
@@ -732,7 +793,7 @@ public class MemStore implements HeapSiz
}
public final static long FIXED_OVERHEAD = ClassSize.align(
- ClassSize.OBJECT + (9 * ClassSize.REFERENCE));
+ ClassSize.OBJECT + (11 * ClassSize.REFERENCE));
public final static long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
ClassSize.REENTRANT_LOCK + ClassSize.ATOMIC_LONG +
Added:
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java?rev=1068150&view=auto
==============================================================================
---
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
(added)
+++
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
Mon Feb 7 21:40:28 2011
@@ -0,0 +1,262 @@
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * 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.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.conf.Configuration;
+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
+ */
+public class MemStoreLAB {
+ private AtomicReference<Chunk> curChunk = new AtomicReference<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;
+
+ public MemStoreLAB() {
+ this(new Configuration());
+ }
+
+ public MemStoreLAB(Configuration conf) {
+ chunkSize = conf.getInt(CHUNK_SIZE_KEY, CHUNK_SIZE_DEFAULT);
+ maxAlloc = conf.getInt(MAX_ALLOC_KEY, MAX_ALLOC_DEFAULT);
+
+ // 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.
+ */
+ 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);
+ }
+ }
+
+ /**
+ * Try to retire the current chunk if it is still
+ * <code>c</code>. Postcondition is that curChunk.get()
+ * != c
+ */
+ private void tryRetireChunk(Chunk c) {
+ @SuppressWarnings("unused")
+ boolean weRetiredIt = 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 = new Chunk(chunkSize);
+ if (curChunk.compareAndSet(null, c)) {
+ // we won race - now we need to actually do the expensive
+ // allocation step
+ c.init();
+ return 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.
+ */
+ private static class Chunk {
+ /** Actual underlying data */
+ private byte[] data;
+
+ private static final int UNINITIALIZED = -1;
+ /**
+ * 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
+ */
+ private 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;
+ data = new byte[size];
+ // 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");
+ }
+
+ /**
+ * 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 + 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(data=" + data +
+ " with capacity=" + data.length +
+ ", off=" + offset + ")";
+ }
+
+ byte[] getData() {
+ return data;
+ }
+
+ int getOffset() {
+ return offset;
+ }
+ }
+}
Modified:
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1068150&r1=1068149&r2=1068150&view=diff
==============================================================================
---
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
(original)
+++
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
Mon Feb 7 21:40:28 2011
@@ -173,7 +173,7 @@ public class Store implements HeapSize {
// second -> ms adjust for user data
this.ttl *= 1000;
}
- this.memstore = new MemStore(this.comparator);
+ this.memstore = new MemStore(conf, this.comparator);
this.storeNameStr = Bytes.toString(this.family.getName());
// By default, we compact if an HStore has more than
Modified: hbase/branches/0.90/src/main/resources/hbase-default.xml
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/resources/hbase-default.xml?rev=1068150&r1=1068149&r2=1068150&view=diff
==============================================================================
--- hbase/branches/0.90/src/main/resources/hbase-default.xml (original)
+++ hbase/branches/0.90/src/main/resources/hbase-default.xml Mon Feb 7
21:40:28 2011
@@ -360,6 +360,16 @@
</description>
</property>
<property>
+ <name>hbase.hregion.memstore.mslab.enabled</name>
+ <value>false</value>
+ <description>
+ Experimental: Enables the MemStore-Local Allocation Buffer,
+ a feature which works to prevent heap fragmentation under
+ heavy write loads. This can reduce the frequency of stop-the-world
+ GC pauses on large heaps.
+ </description>
+ </property>
+ <property>
<name>hbase.hregion.max.filesize</name>
<value>268435456</value>
<description>
Modified:
hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStore.java
URL:
http://svn.apache.org/viewvc/hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStore.java?rev=1068150&r1=1068149&r2=1068150&view=diff
==============================================================================
---
hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStore.java
(original)
+++
hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStore.java
Mon Feb 7 21:40:28 2011
@@ -20,6 +20,8 @@
package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
import java.rmi.UnexpectedException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -32,6 +34,8 @@ import junit.framework.TestCase;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueTestUtil;
@@ -470,7 +474,7 @@ public class TestMemStore extends TestCa
}
public void testMultipleVersionsSimple() throws Exception {
- MemStore m = new MemStore(KeyValue.COMPARATOR);
+ MemStore m = new MemStore(new Configuration(), KeyValue.COMPARATOR);
byte [] row = Bytes.toBytes("testRow");
byte [] family = Bytes.toBytes("testFamily");
byte [] qf = Bytes.toBytes("testQualifier");
@@ -490,7 +494,7 @@ public class TestMemStore extends TestCa
}
public void testBinary() throws IOException {
- MemStore mc = new MemStore(KeyValue.ROOT_COMPARATOR);
+ MemStore mc = new MemStore(new Configuration(), KeyValue.ROOT_COMPARATOR);
final int start = 43;
final int end = 46;
for (int k = start; k <= end; k++) {
@@ -757,7 +761,6 @@ public class TestMemStore extends TestCa
assertEquals(delete, memstore.kvset.first());
}
-
////////////////////////////////////
//Test for timestamps
////////////////////////////////////
@@ -790,7 +793,52 @@ public class TestMemStore extends TestCa
//assertTrue(!memstore.shouldSeek(scan));
}
-
+ ////////////////////////////////////
+ //Test for upsert with MSLAB
+ ////////////////////////////////////
+
+ /**
+ * Test a pathological pattern that shows why we can't currently
+ * use the MSLAB for upsert workloads. This test inserts data
+ * in the following pattern:
+ *
+ * - row0001 through row1000 (fills up one 2M Chunk)
+ * - row0002 through row1001 (fills up another 2M chunk, leaves one reference
+ * to the first chunk
+ * - row0003 through row1002 (another chunk, another dangling reference)
+ *
+ * This causes OOME pretty quickly if we use MSLAB for upsert
+ * since each 2M chunk is held onto by a single reference.
+ */
+ public void testUpsertMSLAB() throws Exception {
+ Configuration conf = HBaseConfiguration.create();
+ conf.setBoolean(MemStore.USEMSLAB_KEY, true);
+ memstore = new MemStore(conf, KeyValue.COMPARATOR);
+
+ int ROW_SIZE = 2048;
+ byte[] qualifier = new byte[ROW_SIZE - 4];
+
+ MemoryMXBean bean = ManagementFactory.getMemoryMXBean();
+ for (int i = 0; i < 3; i++) { System.gc(); }
+ long usageBefore = bean.getHeapMemoryUsage().getUsed();
+
+ long size = 0;
+ long ts=0;
+
+ for (int newValue = 0; newValue < 1000; newValue++) {
+ for (int row = newValue; row < newValue + 1000; row++) {
+ byte[] rowBytes = Bytes.toBytes(row);
+ size += memstore.updateColumnValue(rowBytes, FAMILY, qualifier,
newValue, ++ts);
+ }
+ }
+ System.out.println("Wrote " + ts + " vals");
+ for (int i = 0; i < 3; i++) { System.gc(); }
+ long usageAfter = bean.getHeapMemoryUsage().getUsed();
+ System.out.println("Memory used: " + (usageAfter - usageBefore)
+ + " (heapsize: " + memstore.heapSize() +
+ " size: " + size + ")");
+ }
+
//////////////////////////////////////////////////////////////////////////////
// Helpers
//////////////////////////////////////////////////////////////////////////////
