Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-15784 Misuse core/maxPoolSize of LinkedBlockingQueue in 

ThreadPoolExecutor (Jingcheng Du)
This commit is contained in:
Ramkrishna 2016-05-18 12:40:43 +05:30
parent b2b3b1fa4d
commit 7b5d5394c0
12 changed files with 20 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java
View File

@ -352,8 +352,8 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
if (batchPool == null) {
synchronized (this) {
if (batchPool == null) {
this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256),
conf.getInt("hbase.hconnection.threads.core", 256), "-shared", null);
int threads = conf.getInt("hbase.hconnection.threads.max", 256);
this.batchPool = getThreadPool(threads, threads, "-shared", null);
this.cleanupPool = true;
}
}
@ -377,6 +377,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
new LinkedBlockingQueue<Runnable>(maxThreads *
conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
coreThreads = maxThreads;
}
ThreadPoolExecutor tpe = new ThreadPoolExecutor(
coreThreads,
@ -397,9 +398,10 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
//To start with, threads.max.core threads can hit the meta (including replicas).
//After that, requests will get queued up in the passed queue, and only after
//the queue is full, a new thread will be started
int threads = conf.getInt("hbase.hconnection.meta.lookup.threads.max", 128);
this.metaLookupPool = getThreadPool(
conf.getInt("hbase.hconnection.meta.lookup.threads.max", 128),
conf.getInt("hbase.hconnection.meta.lookup.threads.core", 10),
threads,
threads,
"-metaLookup-shared-", new LinkedBlockingQueue<Runnable>());
}
}

4
hbase-client/src/test/resources/hbase-site.xml
View File

@ -25,10 +25,6 @@
<name>hbase.defaults.for.version.skip</name>
<value>true</value>
</property>
<property>
<name>hbase.hconnection.meta.lookup.threads.core</name>
<value>4</value>
</property>
<property>
<name>hbase.hconnection.threads.keepalivetime</name>
<value>3</value>

2
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java
View File

@ -107,7 +107,7 @@ public class HFileReplicator {
ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
builder.setNameFormat("HFileReplicationCallable-%1$d");
this.exec =
new ThreadPoolExecutor(1, maxCopyThreads, 60, TimeUnit.SECONDS,
new ThreadPoolExecutor(maxCopyThreads, maxCopyThreads, 60, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(), builder.build());
this.exec.allowCoreThreadTimeOut(true);
this.copiesPerThread =

6
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java
View File

@ -235,20 +235,16 @@ public class RegionReplicaReplicationEndpoint extends HBaseReplicationEndpoint {
*/
private ExecutorService getDefaultThreadPool(Configuration conf) {
int maxThreads = conf.getInt("hbase.region.replica.replication.threads.max", 256);
int coreThreads = conf.getInt("hbase.region.replica.replication.threads.core", 16);
if (maxThreads == 0) {
maxThreads = Runtime.getRuntime().availableProcessors() * 8;
}
if (coreThreads == 0) {
coreThreads = Runtime.getRuntime().availableProcessors() * 8;
}
long keepAliveTime = conf.getLong("hbase.region.replica.replication.threads.keepalivetime", 60);
LinkedBlockingQueue<Runnable> workQueue =
new LinkedBlockingQueue<Runnable>(maxThreads *
conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
ThreadPoolExecutor tpe = new ThreadPoolExecutor(
coreThreads,
maxThreads,
maxThreads,
keepAliveTime,
TimeUnit.SECONDS,

6
hbase-server/src/main/java/org/apache/hadoop/hbase/util/MultiHConnection.java
View File

@ -135,20 +135,16 @@ public class MultiHConnection {
private void createBatchPool(Configuration conf) {
// Use the same config for keep alive as in ConnectionImplementation.getBatchPool();
int maxThreads = conf.getInt("hbase.multihconnection.threads.max", 256);
int coreThreads = conf.getInt("hbase.multihconnection.threads.core", 256);
if (maxThreads == 0) {
maxThreads = Runtime.getRuntime().availableProcessors() * 8;
}
if (coreThreads == 0) {
coreThreads = Runtime.getRuntime().availableProcessors() * 8;
}
long keepAliveTime = conf.getLong("hbase.multihconnection.threads.keepalivetime", 60);
LinkedBlockingQueue<Runnable> workQueue =
new LinkedBlockingQueue<Runnable>(maxThreads
* conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
ThreadPoolExecutor tpe =
new ThreadPoolExecutor(coreThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
new ThreadPoolExecutor(maxThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
Threads.newDaemonThreadFactory("MultiHConnection" + "-shared-"));
tpe.allowCoreThreadTimeOut(true);
this.batchPool = tpe;

1
hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckMOB.java
View File

@ -57,7 +57,6 @@ public class TestHBaseFsckMOB extends BaseTestHBaseFsck {
conf.setInt("hbase.htable.threads.max", POOL_SIZE);
conf.setInt("hbase.hconnection.threads.max", 2 * POOL_SIZE);
conf.setInt("hbase.hconnection.threads.core", POOL_SIZE);
conf.setInt("hbase.hbck.close.timeout", 2 * REGION_ONLINE_TIMEOUT);
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 8 * REGION_ONLINE_TIMEOUT);
TEST_UTIL.startMiniCluster(1);

1
hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckOneRS.java
View File

@ -109,7 +109,6 @@ public class TestHBaseFsckOneRS extends BaseTestHBaseFsck {
conf.setInt("hbase.htable.threads.max", POOL_SIZE);
conf.setInt("hbase.hconnection.threads.max", 2 * POOL_SIZE);
conf.setInt("hbase.hconnection.threads.core", POOL_SIZE);
conf.setInt("hbase.hbck.close.timeout", 2 * REGION_ONLINE_TIMEOUT);
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 8 * REGION_ONLINE_TIMEOUT);
TEST_UTIL.startMiniCluster(1);

1
hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckReplicas.java
View File

@ -64,7 +64,6 @@ public class TestHBaseFsckReplicas extends BaseTestHBaseFsck {
conf.setInt("hbase.htable.threads.max", POOL_SIZE);
conf.setInt("hbase.hconnection.threads.max", 2 * POOL_SIZE);
conf.setInt("hbase.hconnection.threads.core", POOL_SIZE);
conf.setInt("hbase.hbck.close.timeout", 2 * REGION_ONLINE_TIMEOUT);
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 8 * REGION_ONLINE_TIMEOUT);
TEST_UTIL.startMiniCluster(3);

1
hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckTwoRS.java
View File

@ -73,7 +73,6 @@ public class TestHBaseFsckTwoRS extends BaseTestHBaseFsck {
conf.setInt("hbase.htable.threads.max", POOL_SIZE);
conf.setInt("hbase.hconnection.threads.max", 2 * POOL_SIZE);
conf.setInt("hbase.hconnection.threads.core", POOL_SIZE);
conf.setInt("hbase.hbck.close.timeout", 2 * REGION_ONLINE_TIMEOUT);
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 8 * REGION_ONLINE_TIMEOUT);
TEST_UTIL.startMiniCluster(2);

4
hbase-server/src/test/resources/hbase-site.xml
View File

@ -162,10 +162,6 @@
Enable replay sanity checks on procedure tests.
</description>
</property>
<property>
<name>hbase.hconnection.meta.lookup.threads.core</name>
<value>4</value>
</property>
<property>
<name>hbase.hconnection.threads.keepalivetime</name>
<value>3</value>

10
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/TBoundedThreadPoolServer.java
View File

@ -132,7 +132,7 @@ public class TBoundedThreadPoolServer extends TServer {
}
/** Executor service for handling client connections */
private ExecutorService executorService;
private ThreadPoolExecutor executorService;
/** Flag for stopping the server */
private volatile boolean stopped;
@ -142,9 +142,12 @@ public class TBoundedThreadPoolServer extends TServer {
public TBoundedThreadPoolServer(Args options, ThriftMetrics metrics) {
super(options);
int minWorkerThreads = options.minWorkerThreads;
int maxWorkerThreads = options.maxWorkerThreads;
if (options.maxQueuedRequests > 0) {
this.callQueue = new CallQueue(
new LinkedBlockingQueue<Call>(options.maxQueuedRequests), metrics);
minWorkerThreads = maxWorkerThreads;
} else {
this.callQueue = new CallQueue(new SynchronousQueue<Call>(), metrics);
}
@ -153,9 +156,10 @@ public class TBoundedThreadPoolServer extends TServer {
tfb.setDaemon(true);
tfb.setNameFormat("thrift-worker-%d");
executorService =
new ThreadPoolExecutor(options.minWorkerThreads,
options.maxWorkerThreads, options.threadKeepAliveTimeSec,
new ThreadPoolExecutor(minWorkerThreads,
maxWorkerThreads, options.threadKeepAliveTimeSec,
TimeUnit.SECONDS, this.callQueue, tfb.build());
executorService.allowCoreThreadTimeOut(true);
serverOptions = options;
}

6
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServerRunner.java
View File

@ -555,7 +555,7 @@ public class ThriftServerRunner implements Runnable {
CallQueue callQueue =
new CallQueue(new LinkedBlockingQueue<Call>(), metrics);
ExecutorService executorService = createExecutor(
callQueue, serverArgs.getMinWorkerThreads(), serverArgs.getMaxWorkerThreads());
callQueue, serverArgs.getMaxWorkerThreads(), serverArgs.getMaxWorkerThreads());
serverArgs.executorService(executorService)
.processor(processor)
.transportFactory(transportFactory)
@ -620,8 +620,10 @@ public class ThriftServerRunner implements Runnable {
ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
tfb.setDaemon(true);
tfb.setNameFormat("thrift-worker-%d");
return new ThreadPoolExecutor(minWorkers, maxWorkers,
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(minWorkers, maxWorkers,
Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build());
threadPool.allowCoreThreadTimeOut(true);
return threadPool;
}
private InetAddress getBindAddress(Configuration conf)