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 we...@apache.org on 2021/08/27 04:50:10 UTC
[hadoop] branch branch-3.3 updated: HDFS-16173.Improve
CopyCommands#Put#executor queue configurability. (#3302)
This is an automated email from the ASF dual-hosted git repository.
weichiu pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 7c66304 HDFS-16173.Improve CopyCommands#Put#executor queue configurability. (#3302)
7c66304 is described below
commit 7c663043b2c1e207dd8c05e09e17811d68badfac
Author: jianghuazhu <74...@qq.com>
AuthorDate: Fri Aug 27 11:41:44 2021 +0800
HDFS-16173.Improve CopyCommands#Put#executor queue configurability. (#3302)
Co-authored-by: zhujianghua <zh...@zhujianghuadeMacBook-Pro.local>
Reviewed-by: Hui Fei <fe...@apache.org>
Reviewed-by: Viraj Jasani <vj...@apache.org>
(cherry picked from commit 4c94831364e9258247029c22a222a665771ab4c0)
---
.../org/apache/hadoop/fs/shell/CopyCommands.java | 34 ++++++++++++++++++++--
.../src/site/markdown/FileSystemShell.md | 4 ++-
.../hadoop/fs/shell/TestCopyPreserveFlag.java | 16 ++++++++++
.../hadoop-common/src/test/resources/testConf.xml | 18 +++++++-----
4 files changed, 62 insertions(+), 10 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
index 77f6317..06809ec 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
@@ -38,6 +38,8 @@ import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathIsDirectoryException;
import org.apache.hadoop.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Various commands for copy files */
@InterfaceAudience.Private
@@ -239,7 +241,11 @@ class CopyCommands {
* Copy local files to a remote filesystem
*/
public static class Put extends CommandWithDestination {
+
+ public static final Logger LOG = LoggerFactory.getLogger(Put.class);
+
private ThreadPoolExecutor executor = null;
+ private int threadPoolQueueSize = 1024;
private int numThreads = 1;
private static final int MAX_THREADS =
@@ -247,7 +253,8 @@ class CopyCommands {
public static final String NAME = "put";
public static final String USAGE =
- "[-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>";
+ "[-f] [-p] [-l] [-d] [-t <thread count>] [-q <threadPool queue size>] " +
+ "<localsrc> ... <dst>";
public static final String DESCRIPTION =
"Copy files from the local file system " +
"into fs. Copying fails if the file already " +
@@ -256,6 +263,8 @@ class CopyCommands {
" -p : Preserves timestamps, ownership and the mode.\n" +
" -f : Overwrites the destination if it already exists.\n" +
" -t <thread count> : Number of threads to be used, default is 1.\n" +
+ " -q <threadPool size> : ThreadPool queue size to be used, " +
+ "default is 1024.\n" +
" -l : Allow DataNode to lazily persist the file to disk. Forces" +
" replication factor of 1. This flag will result in reduced" +
" durability. Use with care.\n" +
@@ -266,8 +275,10 @@ class CopyCommands {
CommandFormat cf =
new CommandFormat(1, Integer.MAX_VALUE, "f", "p", "l", "d");
cf.addOptionWithValue("t");
+ cf.addOptionWithValue("q");
cf.parse(args);
setNumberThreads(cf.getOptValue("t"));
+ setThreadPoolQueueSize(cf.getOptValue("q"));
setOverwrite(cf.getOpt("f"));
setPreserve(cf.getOpt("p"));
setLazyPersist(cf.getOpt("l"));
@@ -299,7 +310,7 @@ class CopyCommands {
}
executor = new ThreadPoolExecutor(numThreads, numThreads, 1,
- TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024),
+ TimeUnit.SECONDS, new ArrayBlockingQueue<>(threadPoolQueueSize),
new ThreadPoolExecutor.CallerRunsPolicy());
super.processArguments(args);
@@ -329,6 +340,25 @@ class CopyCommands {
}
}
+ private void setThreadPoolQueueSize(String numThreadPoolQueueSize) {
+ if (numThreadPoolQueueSize != null) {
+ int parsedValue = Integer.parseInt(numThreadPoolQueueSize);
+ if (parsedValue < 1) {
+ LOG.warn("The value of the thread pool queue size cannot be " +
+ "less than 1, and the default value is used here. " +
+ "The default size is 1024.");
+ threadPoolQueueSize = 1024;
+ } else {
+ threadPoolQueueSize = parsedValue;
+ }
+ }
+ }
+
+ @VisibleForTesting
+ protected int getThreadPoolQueueSize() {
+ return threadPoolQueueSize;
+ }
+
private void copyFile(PathData src, PathData target) throws IOException {
if (isPathRecursable(src)) {
throw new PathIsDirectoryException(src.toString());
diff --git a/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md b/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
index 33a6bf1..feed0a4 100644
--- a/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
+++ b/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
@@ -525,7 +525,7 @@ Returns 0 on success and -1 on error.
put
---
-Usage: `hadoop fs -put [-f] [-p] [-l] [-d] [-t <thread count>] [ - | <localsrc1> .. ]. <dst>`
+Usage: `hadoop fs -put [-f] [-p] [-l] [-d] [-t <thread count>] [-q <threadPool queue size>] [ - | <localsrc1> .. ]. <dst>`
Copy single src, or multiple srcs from local file system to the destination file system.
Also reads input from stdin and writes to destination file system if the source is set to "-"
@@ -542,6 +542,7 @@ Options:
* `-l` : Allow DataNode to lazily persist the file to disk, Forces a replication
factor of 1. This flag will result in reduced durability. Use with care.
* `-d` : Skip creation of temporary file with the suffix `._COPYING_`.
+* `-q <threadPool queue size>` : ThreadPool queue size to be used, default is 1024.
Examples:
@@ -550,6 +551,7 @@ Examples:
* `hadoop fs -put -f localfile1 localfile2 /user/hadoop/hadoopdir`
* `hadoop fs -put -d localfile hdfs://nn.example.com/hadoop/hadoopfile`
* `hadoop fs -put - hdfs://nn.example.com/hadoop/hadoopfile` Reads the input from stdin.
+* `hadoop fs -put -q 500 localfile3 hdfs://nn.example.com/hadoop/hadoopfile3`
Exit Code:
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestCopyPreserveFlag.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestCopyPreserveFlag.java
index 8d2e160..0f0ddcc 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestCopyPreserveFlag.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestCopyPreserveFlag.java
@@ -123,6 +123,22 @@ public class TestCopyPreserveFlag {
}
@Test(timeout = 10000)
+ public void testPutWithPQ() throws Exception {
+ Put put = new Put();
+ run(put, "-p", "-q", "100", FROM.toString(), TO.toString());
+ assertEquals(put.getThreadPoolQueueSize(), 100);
+ assertAttributesPreserved(TO);
+ }
+
+ @Test(timeout = 10000)
+ public void testPutWithQ() throws Exception {
+ Put put = new Put();
+ run(put, "-q", "100", FROM.toString(), TO.toString());
+ assertEquals(put.getThreadPoolQueueSize(), 100);
+ assertAttributesChanged(TO);
+ }
+
+ @Test(timeout = 10000)
public void testPutWithSplCharacter() throws Exception {
fs.mkdirs(DIR_FROM_SPL);
fs.createNewFile(FROM_SPL);
diff --git a/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml b/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
index edf598b..fd6ee11 100644
--- a/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
+++ b/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
@@ -498,7 +498,7 @@
<type>RegexpComparator</type>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^-put \[-f\] \[-p\] \[-l\] \[-d\] \[-t <thread count>\] <localsrc> \.\.\. <dst> :\s*</expected-output>
+ <expected-output>^-put \[-f\] \[-p\] \[-l\] \[-d\] \[-t <thread count>\] \[-q <threadPool queue size>\] <localsrc> \.\.\. <dst> :\s*</expected-output>
</comparator>
</comparator>
<comparator>
@@ -515,19 +515,23 @@
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-p Preserves timestamps, ownership and the mode.( )*</expected-output>
+ <expected-output>^\s*-p Preserves timestamps, ownership and the mode.( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-f Overwrites the destination if it already exists.( )*</expected-output>
+ <expected-output>^\s*-f Overwrites the destination if it already exists.( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-t <thread count> Number of threads to be used, default is 1.( )*</expected-output>
+ <expected-output>^\s*-t <thread count> Number of threads to be used, default is 1.( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-l Allow DataNode to lazily persist the file to disk. Forces( )*</expected-output>
+ <expected-output>^\s*-q <threadPool size> ThreadPool queue size to be used, default is 1024.( )*</expected-output>
+ </comparator>
+ <comparator>
+ <type>RegexpComparator</type>
+ <expected-output>^\s*-l Allow DataNode to lazily persist the file to disk. Forces( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
@@ -539,7 +543,7 @@
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-d Skip creation of temporary file\(<dst>\._COPYING_\).( )*</expected-output>
+ <expected-output>^\s*-d Skip creation of temporary file\(<dst>\._COPYING_\).( )*</expected-output>
</comparator>
</comparators>
</test>
@@ -554,7 +558,7 @@
<comparators>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^-copyFromLocal \[-f\] \[-p\] \[-l\] \[-d\] \[-t <thread count>\] <localsrc> \.\.\. <dst> :\s*</expected-output>
+ <expected-output>^-copyFromLocal \[-f\] \[-p\] \[-l\] \[-d\] \[-t <thread count>\] \[-q <threadPool queue size>\] <localsrc> \.\.\. <dst> :\s*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org