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 <[email protected]>
AuthorDate: Fri Aug 27 11:41:44 2021 +0800
HDFS-16173.Improve CopyCommands#Put#executor queue configurability. (#3302)
Co-authored-by: zhujianghua <[email protected]>
Reviewed-by: Hui Fei <[email protected]>
Reviewed-by: Viraj Jasani <[email protected]>
(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: [email protected]
For additional commands, e-mail: [email protected]
