You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by va...@apache.org on 2016/11/06 09:58:07 UTC
[21/50] [abbrv] hadoop git commit: HDFS-10909. De-duplicate code in
ErasureCodingWorker#initializeStripedReadThreadPool and
DFSClient#initThreadsNumForStripedReads. (Manoj Govindassamy via lei)
HDFS-10909. De-duplicate code in ErasureCodingWorker#initializeStripedReadThreadPool and DFSClient#initThreadsNumForStripedReads. (Manoj Govindassamy via lei)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b5920619
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b5920619
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b5920619
Branch: refs/heads/YARN-5355
Commit: b59206190e6f773fc223bcb81774a09715551367
Parents: 9b0c17f
Author: Lei Xu <le...@apache.org>
Authored: Wed Nov 2 16:45:25 2016 -0700
Committer: Lei Xu <le...@apache.org>
Committed: Wed Nov 2 16:45:25 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/hadoop/hdfs/DFSClient.java | 30 ++----------
.../org/apache/hadoop/hdfs/DFSUtilClient.java | 50 ++++++++++++++++++++
.../erasurecode/ErasureCodingWorker.java | 23 +++------
3 files changed, 61 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b5920619/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index 93c0ff0..0128b07 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -2800,37 +2800,17 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
/**
* Create thread pool for parallel reading in striped layout,
* STRIPED_READ_THREAD_POOL, if it does not already exist.
- * @param num Number of threads for striped reads thread pool.
+ * @param numThreads Number of threads for striped reads thread pool.
*/
- private void initThreadsNumForStripedReads(int num) {
- assert num > 0;
+ private void initThreadsNumForStripedReads(int numThreads) {
+ assert numThreads > 0;
if (STRIPED_READ_THREAD_POOL != null) {
return;
}
synchronized (DFSClient.class) {
if (STRIPED_READ_THREAD_POOL == null) {
- STRIPED_READ_THREAD_POOL = new ThreadPoolExecutor(1, num, 60,
- TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
- new Daemon.DaemonFactory() {
- private final AtomicInteger threadIndex = new AtomicInteger(0);
-
- @Override
- public Thread newThread(Runnable r) {
- Thread t = super.newThread(r);
- t.setName("stripedRead-" + threadIndex.getAndIncrement());
- return t;
- }
- },
- new ThreadPoolExecutor.CallerRunsPolicy() {
- @Override
- public void rejectedExecution(Runnable runnable,
- ThreadPoolExecutor e) {
- LOG.info("Execution for striped reading rejected, "
- + "Executing in current thread");
- // will run in the current thread
- super.rejectedExecution(runnable, e);
- }
- });
+ STRIPED_READ_THREAD_POOL = DFSUtilClient.getThreadPoolExecutor(1,
+ numThreads, 60, "StripedRead-", true);
STRIPED_READ_THREAD_POOL.allowCoreThreadTimeOut(true);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b5920619/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
index b93632a..d267530 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs;
import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.google.common.primitives.SignedBytes;
import org.apache.hadoop.conf.Configuration;
@@ -51,6 +52,7 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.net.NodeBase;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.Daemon;
import org.apache.hadoop.util.KMSUtil;
import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
@@ -81,6 +83,10 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_DATA_TRANSFER_CLIENT_TCPNODELAY_DEFAULT;
import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_DATA_TRANSFER_CLIENT_TCPNODELAY_KEY;
@@ -776,4 +782,48 @@ public class DFSUtilClient {
DFS_DATA_TRANSFER_CLIENT_TCPNODELAY_KEY,
DFS_DATA_TRANSFER_CLIENT_TCPNODELAY_DEFAULT);
}
+
+ /**
+ * Utility to create a {@link ThreadPoolExecutor}.
+ *
+ * @param corePoolSize - min threads in the pool, even if idle
+ * @param maxPoolSize - max threads in the pool
+ * @param keepAliveTimeSecs - max seconds beyond which excess idle threads
+ * will be terminated
+ * @param threadNamePrefix - name prefix for the pool threads
+ * @param runRejectedExec - when true, rejected tasks from
+ * ThreadPoolExecutor are run in the context of calling thread
+ * @return ThreadPoolExecutor
+ */
+ public static ThreadPoolExecutor getThreadPoolExecutor(int corePoolSize,
+ int maxPoolSize, long keepAliveTimeSecs, String threadNamePrefix,
+ boolean runRejectedExec) {
+ Preconditions.checkArgument(corePoolSize > 0);
+ ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize,
+ maxPoolSize, keepAliveTimeSecs, TimeUnit.SECONDS,
+ new SynchronousQueue<Runnable>(), new Daemon.DaemonFactory() {
+ private final AtomicInteger threadIndex = new AtomicInteger(0);
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = super.newThread(r);
+ t.setName(threadNamePrefix + threadIndex.getAndIncrement());
+ return t;
+ }
+ });
+ if (runRejectedExec) {
+ threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor
+ .CallerRunsPolicy() {
+ @Override
+ public void rejectedExecution(Runnable runnable,
+ ThreadPoolExecutor e) {
+ LOG.info(threadNamePrefix + " task is rejected by " +
+ "ThreadPoolExecutor. Executing it in current thread.");
+ // will run in the current thread
+ super.rejectedExecution(runnable, e);
+ }
+ });
+ }
+ return threadPoolExecutor;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b5920619/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java
index aacbb2d..1492e5d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java
@@ -20,13 +20,13 @@ package org.apache.hadoop.hdfs.server.datanode.erasurecode;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.DFSUtilClient;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.protocol.BlockECReconstructionCommand.BlockECReconstructionInfo;
import org.apache.hadoop.util.Daemon;
import org.slf4j.Logger;
import java.util.Collection;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -89,22 +89,11 @@ public final class ErasureCodingWorker {
stripedReadPool.allowCoreThreadTimeOut(true);
}
- private void initializeStripedBlkReconstructionThreadPool(int num) {
- LOG.debug("Using striped block reconstruction; pool threads={}", num);
- stripedReconstructionPool = new ThreadPoolExecutor(2, num, 60,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<Runnable>(),
- new Daemon.DaemonFactory() {
- private final AtomicInteger threadIdx = new AtomicInteger(0);
-
- @Override
- public Thread newThread(Runnable r) {
- Thread t = super.newThread(r);
- t.setName("stripedBlockReconstruction-"
- + threadIdx.getAndIncrement());
- return t;
- }
- });
+ private void initializeStripedBlkReconstructionThreadPool(int numThreads) {
+ LOG.debug("Using striped block reconstruction; pool threads={}",
+ numThreads);
+ stripedReconstructionPool = DFSUtilClient.getThreadPoolExecutor(2,
+ numThreads, 60, "StripedBlockReconstruction-", false);
stripedReconstructionPool.allowCoreThreadTimeOut(true);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org