This is an automated email from the ASF dual-hosted git repository. hxd pushed a commit to branch cluster- in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c2f0cb613e762ddd39817092dcc306067aa862f6 Author: xiangdong huang <[email protected]> AuthorDate: Wed Aug 11 12:12:41 2021 +0800 continue for threadpool --- .../iotdb/cluster/server/member/RaftMember.java | 11 ++++------- .../db/concurrent/IoTDBThreadPoolFactory.java | 22 +++++++++++++++++++++- .../threadpool/WrappedThreadPoolExecutor.java | 19 ++++++------------- 3 files changed, 31 insertions(+), 21 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java index 87238c4..d714540 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java @@ -19,6 +19,7 @@ package org.apache.iotdb.cluster.server.member; +import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.iotdb.cluster.ClusterIoTDB; import org.apache.iotdb.cluster.client.async.AsyncClientPool; import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor; @@ -62,6 +63,7 @@ import org.apache.iotdb.cluster.utils.ClientUtils; import org.apache.iotdb.cluster.utils.IOUtils; import org.apache.iotdb.cluster.utils.PlanSerializer; import org.apache.iotdb.cluster.utils.StatusUtils; +import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.exception.BatchProcessException; import org.apache.iotdb.db.exception.IoTDBException; @@ -79,8 +81,6 @@ import org.apache.iotdb.db.utils.TestOnly; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSStatus; - -import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -103,7 +103,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -302,7 +301,7 @@ public abstract class RaftMember implements RaftMemberMBean { new ThreadFactoryBuilder().setNameFormat(getName() + "-AppendLog%d").build()); if (!config.isUseAsyncServer()) { serialToParallelPool = - new ThreadPoolExecutor( + IoTDBThreadPoolFactory.newThreadPool( allNodes.size(), Math.max(allNodes.size(), Runtime.getRuntime().availableProcessors()), 1000L, @@ -310,9 +309,7 @@ public abstract class RaftMember implements RaftMemberMBean { new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setNameFormat(getName() + "-SerialToParallel%d").build()); } - commitLogPool = - Executors.newSingleThreadExecutor( - new ThreadFactoryBuilder().setNameFormat(getName() + "-CommitLog%d").build()); + commitLogPool = IoTDBThreadPoolFactory.newSingleThreadExecutor("RaftCommitLog"); } public String getName() { diff --git a/server/src/main/java/org/apache/iotdb/db/concurrent/IoTDBThreadPoolFactory.java b/server/src/main/java/org/apache/iotdb/db/concurrent/IoTDBThreadPoolFactory.java index 77799eb..72dacc9 100644 --- a/server/src/main/java/org/apache/iotdb/db/concurrent/IoTDBThreadPoolFactory.java +++ b/server/src/main/java/org/apache/iotdb/db/concurrent/IoTDBThreadPoolFactory.java @@ -22,11 +22,11 @@ import org.apache.iotdb.db.concurrent.threadpool.WrappedScheduledExecutorService import org.apache.iotdb.db.concurrent.threadpool.WrappedSingleThreadExecutorService; import org.apache.iotdb.db.concurrent.threadpool.WrappedSingleThreadScheduledExecutor; import org.apache.iotdb.db.concurrent.threadpool.WrappedThreadPoolExecutor; - import org.apache.thrift.server.TThreadPoolServer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; @@ -146,12 +146,14 @@ public class IoTDBThreadPoolFactory { * @return scheduled thread pool. */ public static ScheduledExecutorService newSingleThreadScheduledExecutor(String poolName) { + logger.info("new single scheduled thread pool: {}", poolName); return new WrappedSingleThreadScheduledExecutor( Executors.newSingleThreadScheduledExecutor(new IoTThreadFactory(poolName)), poolName); } public static ScheduledExecutorService newSingleThreadScheduledExecutor( String poolName, Thread.UncaughtExceptionHandler handler) { + logger.info("new single scheduled thread pool: {}", poolName); return new WrappedSingleThreadScheduledExecutor( Executors.newSingleThreadScheduledExecutor(new IoTThreadFactory(poolName, handler)), poolName); @@ -165,20 +167,36 @@ public class IoTDBThreadPoolFactory { * @return thread pool. */ public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, String poolName) { + logger.info("new scheduled thread pool: {}", poolName); return new WrappedScheduledExecutorService( Executors.newScheduledThreadPool(corePoolSize, new IoTThreadFactory(poolName)), poolName); } public static ScheduledExecutorService newScheduledThreadPool( int corePoolSize, String poolName, Thread.UncaughtExceptionHandler handler) { + logger.info("new scheduled thread pool: {}", poolName); return new WrappedScheduledExecutorService( Executors.newScheduledThreadPool(corePoolSize, new IoTThreadFactory(poolName, handler)), poolName); } + public static ExecutorService newThreadPool( + int corePoolSize, + int maximumPoolSize, + long keepAliveTime, + TimeUnit unit, + BlockingQueue<Runnable> workQueue, + IoTThreadFactory ioTThreadFactory, + String poolName) { + logger.info("new thread pool: {}", poolName); + return new WrappedThreadPoolExecutor( + corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, ioTThreadFactory, poolName); + } + /** function for creating thrift rpc client thread pool. */ public static ExecutorService createThriftRpcClientThreadPool( TThreadPoolServer.Args args, String poolName) { + logger.info("new SynchronousQueue thread pool: {}", poolName); SynchronousQueue<Runnable> executorQueue = new SynchronousQueue<>(); return new WrappedThreadPoolExecutor( args.minWorkerThreads, @@ -197,6 +215,7 @@ public class IoTDBThreadPoolFactory { int stopTimeoutVal, TimeUnit stopTimeoutUnit, String poolName) { + logger.info("new SynchronousQueue thread pool: {}", poolName); SynchronousQueue<Runnable> executorQueue = new SynchronousQueue<>(); return new WrappedThreadPoolExecutor( minWorkerThreads, @@ -211,6 +230,7 @@ public class IoTDBThreadPoolFactory { /** function for creating thrift rpc client thread pool. */ public static ExecutorService createThriftRpcClientThreadPool( TThreadPoolServer.Args args, String poolName, Thread.UncaughtExceptionHandler handler) { + logger.info("new SynchronousQueue thread pool: {}", poolName); SynchronousQueue<Runnable> executorQueue = new SynchronousQueue<>(); return new WrappedThreadPoolExecutor( args.minWorkerThreads, diff --git a/server/src/main/java/org/apache/iotdb/db/concurrent/threadpool/WrappedThreadPoolExecutor.java b/server/src/main/java/org/apache/iotdb/db/concurrent/threadpool/WrappedThreadPoolExecutor.java index 2b71ab3..3ed4040 100644 --- a/server/src/main/java/org/apache/iotdb/db/concurrent/threadpool/WrappedThreadPoolExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/concurrent/threadpool/WrappedThreadPoolExecutor.java @@ -24,7 +24,6 @@ import org.apache.iotdb.db.service.JMXService; import java.util.List; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -49,20 +48,14 @@ public class WrappedThreadPoolExecutor extends ThreadPoolExecutor } public WrappedThreadPoolExecutor( - int minWorkerThreads, - int maxWorkerThreads, - int stopTimeoutVal, - TimeUnit stopTimeoutUnit, - SynchronousQueue<Runnable> executorQueue, + int corePoolSize, + int maximumPoolSize, + long keepAliveTime, + TimeUnit unit, + BlockingQueue<Runnable> workQueue, IoTThreadFactory ioTThreadFactory, String mbeanName) { - super( - minWorkerThreads, - maxWorkerThreads, - stopTimeoutVal, - stopTimeoutUnit, - executorQueue, - ioTThreadFactory); + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, ioTThreadFactory); this.mbeanName = String.format( "%s:%s=%s", IoTDBConstant.IOTDB_THREADPOOL_PACKAGE, IoTDBConstant.JMX_TYPE, mbeanName);
