You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by la...@apache.org on 2012/08/16 19:50:31 UTC
svn commit: r1373951 - in
/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase: HConstants.java
regionserver/MemStore.java regionserver/Store.java
Author: larsh
Date: Thu Aug 16 17:50:30 2012
New Revision: 1373951
URL: http://svn.apache.org/viewvc?rev=1373951&view=rev
Log:
HBASE-6561 Gets/Puts with many columns send the RegionServer into an 'endless' loop
Modified:
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/HConstants.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/HConstants.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/HConstants.java?rev=1373951&r1=1373950&r2=1373951&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/HConstants.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/HConstants.java Thu Aug 16 17:50:30 2012
@@ -196,6 +196,9 @@ public final class HConstants {
/** Parameter name for how often a region should should perform a major compaction */
public static final String MAJOR_COMPACTION_PERIOD = "hbase.hregion.majorcompaction";
+ /** Parameter name for the maximum batch of KVs to be used in flushes and compactions */
+ public static final String COMPACTION_KV_MAX = "hbase.hstore.compaction.kv.max";
+
/** Parameter name for HBase instance root directory */
public static final String HBASE_DIR = "hbase.rootdir";
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java?rev=1373951&r1=1373950&r2=1373951&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java Thu Aug 16 17:50:30 2012
@@ -672,6 +672,10 @@ public class MemStore implements HeapSiz
private KeyValue kvsetNextRow = null;
private KeyValue snapshotNextRow = null;
+ // last iterated KVs for kvset and snapshot (to restore iterator state after reseek)
+ private KeyValue kvsetItRow = null;
+ private KeyValue snapshotItRow = null;
+
// iterator based scanning.
private Iterator<KeyValue> kvsetIt;
private Iterator<KeyValue> snapshotIt;
@@ -680,10 +684,6 @@ public class MemStore implements HeapSiz
volatile KeyValueSkipListSet kvsetAtCreation;
volatile KeyValueSkipListSet snapshotAtCreation;
- // Sub lists on which we're iterating
- private SortedSet<KeyValue> kvTail;
- private SortedSet<KeyValue> snapshotTail;
-
// the pre-calculated KeyValue to be returned by peek() or next()
private KeyValue theNext;
@@ -715,17 +715,29 @@ public class MemStore implements HeapSiz
snapshotAtCreation = snapshot;
}
- protected KeyValue getNext(Iterator<KeyValue> it) {
+ private KeyValue getNext(Iterator<KeyValue> it) {
long readPoint = MultiVersionConsistencyControl.getThreadReadPoint();
- while (it.hasNext()) {
- KeyValue v = it.next();
- if (v.getMemstoreTS() <= readPoint) {
- return v;
+ KeyValue v = null;
+ try {
+ while (it.hasNext()) {
+ v = it.next();
+ if (v.getMemstoreTS() <= readPoint) {
+ return v;
+ }
}
- }
- return null;
+ return null;
+ } finally {
+ if (v != null) {
+ // in all cases, remember the last KV iterated to
+ if (it == snapshotIt) {
+ snapshotItRow = v;
+ } else {
+ kvsetItRow = v;
+ }
+ }
+ }
}
/**
@@ -744,8 +756,10 @@ public class MemStore implements HeapSiz
// kvset and snapshot will never be null.
// if tailSet can't find anything, SortedSet is empty (not null).
- kvTail = kvsetAtCreation.tailSet(key);
- snapshotTail = snapshotAtCreation.tailSet(key);
+ kvsetIt = kvsetAtCreation.tailSet(key).iterator();
+ snapshotIt = snapshotAtCreation.tailSet(key).iterator();
+ kvsetItRow = null;
+ snapshotItRow = null;
return seekInSubLists(key);
}
@@ -755,9 +769,6 @@ public class MemStore implements HeapSiz
* (Re)initialize the iterators after a seek or a reseek.
*/
private synchronized boolean seekInSubLists(KeyValue key){
- kvsetIt = kvTail.iterator();
- snapshotIt = snapshotTail.iterator();
-
kvsetNextRow = getNext(kvsetIt);
snapshotNextRow = getNext(snapshotIt);
@@ -777,25 +788,20 @@ public class MemStore implements HeapSiz
@Override
public synchronized boolean reseek(KeyValue key) {
/*
- See HBASE-4195 & HBASE-3855 for the background on this implementation.
+ See HBASE-4195 & HBASE-3855 & HBASE-6591 for the background on this implementation.
This code is executed concurrently with flush and puts, without locks.
Two points must be known when working on this code:
1) It's not possible to use the 'kvTail' and 'snapshot'
variables, as they are modified during a flush.
- 2) The ideal implementation for performances would use the sub skip list
+ 2) The ideal implementation for performance would use the sub skip list
implicitly pointed by the iterators 'kvsetIt' and
'snapshotIt'. Unfortunately the Java API does not offer a method to
- get it. So we're using the skip list that we kept when we created
- the iterators. As these iterators could have been moved forward after
- their creation, we're doing a kind of rewind here. It has a small
- performance impact (we're using a wider list than necessary), and we
- could see values that were not here when we read the list the first
- time. We expect that the new values will be skipped by the test on
- readpoint performed in the next() function.
+ get it. So we remember the last keys we iterated to and restore
+ the reseeked set to at least that point.
*/
- kvTail = kvTail.tailSet(key);
- snapshotTail = snapshotTail.tailSet(key);
+ kvsetIt = kvsetAtCreation.tailSet(getHighest(key, kvsetItRow)).iterator();
+ snapshotIt = snapshotAtCreation.tailSet(getHighest(key, snapshotItRow)).iterator();
return seekInSubLists(key);
}
@@ -836,7 +842,7 @@ public class MemStore implements HeapSiz
* This uses comparator.compare() to compare the KeyValue using the memstore
* comparator.
*/
- protected KeyValue getLowest(KeyValue first, KeyValue second) {
+ private KeyValue getLowest(KeyValue first, KeyValue second) {
if (first == null && second == null) {
return null;
}
@@ -847,12 +853,31 @@ public class MemStore implements HeapSiz
return (first != null ? first : second);
}
+ /*
+ * Returns the higher of the two key values, or null if they are both null.
+ * This uses comparator.compare() to compare the KeyValue using the memstore
+ * comparator.
+ */
+ private KeyValue getHighest(KeyValue first, KeyValue second) {
+ if (first == null && second == null) {
+ return null;
+ }
+ if (first != null && second != null) {
+ int compare = comparator.compare(first, second);
+ return (compare > 0 ? first : second);
+ }
+ return (first != null ? first : second);
+ }
+
public synchronized void close() {
this.kvsetNextRow = null;
this.snapshotNextRow = null;
this.kvsetIt = null;
this.snapshotIt = null;
+
+ this.kvsetItRow = null;
+ this.snapshotItRow = null;
}
/**
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1373951&r1=1373950&r2=1373951&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Thu Aug 16 17:50:30 2012
@@ -243,7 +243,7 @@ public class Store extends SchemaConfigu
this.region.memstoreFlushSize);
this.maxCompactSize
= conf.getLong("hbase.hstore.compaction.max.size", Long.MAX_VALUE);
- this.compactionKVMax = conf.getInt("hbase.hstore.compaction.kv.max", 10);
+ this.compactionKVMax = conf.getInt(HConstants.COMPACTION_KV_MAX, 10);
this.verifyBulkLoads = conf.getBoolean("hbase.hstore.bulkload.verify",
false);
@@ -706,9 +706,9 @@ public class Store extends SchemaConfigu
if (scanner == null) {
Scan scan = new Scan();
scan.setMaxVersions(scanInfo.getMaxVersions());
- scanner = new StoreScanner(this, scanInfo, scan, Collections.singletonList(new CollectionBackedScanner(
- set, this.comparator)), ScanType.MINOR_COMPACT, this.region.getSmallestReadPoint(),
- HConstants.OLDEST_TIMESTAMP);
+ scanner = new StoreScanner(this, scanInfo, scan,
+ Collections.singletonList(memstoreScanner), ScanType.MINOR_COMPACT,
+ this.region.getSmallestReadPoint(), HConstants.OLDEST_TIMESTAMP);
}
if (getHRegion().getCoprocessorHost() != null) {
InternalScanner cpScanner =
@@ -733,7 +733,7 @@ public class Store extends SchemaConfigu
List<KeyValue> kvs = new ArrayList<KeyValue>();
boolean hasMore;
do {
- hasMore = scanner.next(kvs);
+ hasMore = scanner.next(kvs, this.compactionKVMax);
if (!kvs.isEmpty()) {
for (KeyValue kv : kvs) {
// If we know that this KV is going to be included always, then let us