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/10/25 18:44:37 UTC
svn commit: r1402215 - in
/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver:
CompactSplitThread.java Store.java
Author: larsh
Date: Thu Oct 25 16:44:37 2012
New Revision: 1402215
URL: http://svn.apache.org/viewvc?rev=1402215&view=rev
Log:
HBASE-7040 Port HBASE-5867 Improve Compaction Throttle Default to 0.94 (Sergey Shelukhin)
Modified:
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.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/regionserver/CompactSplitThread.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java?rev=1402215&r1=1402214&r2=1402215&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java Thu Oct 25 16:44:37 2012
@@ -50,7 +50,6 @@ public class CompactSplitThread implemen
private final ThreadPoolExecutor largeCompactions;
private final ThreadPoolExecutor smallCompactions;
private final ThreadPoolExecutor splits;
- private final long throttleSize;
/**
* Splitting should not take place if the total number of regions exceed this.
@@ -71,22 +70,11 @@ public class CompactSplitThread implemen
"hbase.regionserver.thread.compaction.large", 1));
int smallThreads = conf.getInt(
"hbase.regionserver.thread.compaction.small", 1);
- if (conf.get("hbase.regionserver.thread.compaction.throttle") != null) {
- throttleSize = conf.getLong(
- "hbase.regionserver.thread.compaction.throttle", 0);
- } else {
- // we have a complicated default. see HBASE-3877
- long flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
- HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);
- long splitSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE,
- HConstants.DEFAULT_MAX_FILE_SIZE);
- throttleSize = Math.min(flushSize * 2, splitSize / 2);
- }
int splitThreads = conf.getInt("hbase.regionserver.thread.split", 1);
// if we have throttle threads, make sure the user also specified size
- Preconditions.checkArgument(smallThreads == 0 || throttleSize > 0);
+ Preconditions.checkArgument(largeThreads > 0 && smallThreads > 0);
final String n = Thread.currentThread().getName();
@@ -102,22 +90,18 @@ public class CompactSplitThread implemen
});
this.largeCompactions
.setRejectedExecutionHandler(new CompactionRequest.Rejection());
- if (smallThreads <= 0) {
- this.smallCompactions = null;
- } else {
- this.smallCompactions = new ThreadPoolExecutor(smallThreads, smallThreads,
- 60, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(),
- new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(r);
- t.setName(n + "-smallCompactions-" + System.currentTimeMillis());
- return t;
- }
- });
- this.smallCompactions
- .setRejectedExecutionHandler(new CompactionRequest.Rejection());
- }
+ this.smallCompactions = new ThreadPoolExecutor(smallThreads, smallThreads,
+ 60, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(),
+ new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r);
+ t.setName(n + "-smallCompactions-" + System.currentTimeMillis());
+ return t;
+ }
+ });
+ this.smallCompactions
+ .setRejectedExecutionHandler(new CompactionRequest.Rejection());
this.splits = (ThreadPoolExecutor)
Executors.newFixedThreadPool(splitThreads,
new ThreadFactory() {
@@ -132,14 +116,12 @@ public class CompactSplitThread implemen
@Override
public String toString() {
- return "compaction_queue="
- + (smallCompactions != null ? "("
- + largeCompactions.getQueue().size() + ":"
- + smallCompactions.getQueue().size() + ")"
- : largeCompactions.getQueue().size())
+ return "compaction_queue=("
+ + largeCompactions.getQueue().size() + ":"
+ + smallCompactions.getQueue().size() + ")"
+ ", split_queue=" + splits.getQueue().size();
}
-
+
public String dumpQueue() {
StringBuffer queueLists = new StringBuffer();
queueLists.append("Compaction/Split Queue dump:\n");
@@ -238,17 +220,11 @@ public class CompactSplitThread implemen
if (priority != Store.NO_PRIORITY) {
cr.setPriority(priority);
}
- ThreadPoolExecutor pool = largeCompactions;
- if (smallCompactions != null && throttleSize > cr.getSize()) {
- // smallCompactions is like the 10 items or less line at Walmart
- pool = smallCompactions;
- }
+ ThreadPoolExecutor pool = s.throttleCompaction(cr.getSize())
+ ? largeCompactions : smallCompactions;
pool.execute(cr);
if (LOG.isDebugEnabled()) {
- String type = "";
- if (smallCompactions != null) {
- type = (pool == smallCompactions) ? "Small " : "Large ";
- }
+ String type = (pool == smallCompactions) ? "Small " : "Large ";
LOG.debug(type + "Compaction requested: " + cr
+ (why != null && !why.isEmpty() ? "; Because: " + why : "")
+ "; " + this);
@@ -267,8 +243,7 @@ public class CompactSplitThread implemen
void interruptIfNecessary() {
splits.shutdown();
largeCompactions.shutdown();
- if (smallCompactions != null)
- smallCompactions.shutdown();
+ smallCompactions.shutdown();
}
private void waitFor(ThreadPoolExecutor t, String name) {
@@ -286,9 +261,7 @@ public class CompactSplitThread implemen
void join() {
waitFor(splits, "Split Thread");
waitFor(largeCompactions, "Large Compaction Thread");
- if (smallCompactions != null) {
- waitFor(smallCompactions, "Small Compaction Thread");
- }
+ waitFor(smallCompactions, "Small Compaction Thread");
}
/**
@@ -298,10 +271,7 @@ public class CompactSplitThread implemen
* @return The current size of the regions queue.
*/
public int getCompactionQueueSize() {
- int size = largeCompactions.getQueue().size();
- if (smallCompactions != null)
- size += smallCompactions.getQueue().size();
- return size;
+ return largeCompactions.getQueue().size() + smallCompactions.getQueue().size();
}
private boolean shouldSplitRegion() {
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=1402215&r1=1402214&r2=1402215&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 Oct 25 16:44:37 2012
@@ -2181,6 +2181,13 @@ public class Store extends SchemaConfigu
}
}
+ boolean throttleCompaction(long compactionSize) {
+ long throttlePoint = conf.getLong(
+ "hbase.regionserver.thread.compaction.throttle",
+ 2 * this.minFilesToCompact * this.region.memstoreFlushSize);
+ return compactionSize > throttlePoint;
+ }
+
public HRegion getHRegion() {
return this.region;
}