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 mt...@apache.org on 2022/09/28 21:56:05 UTC
[hadoop] branch branch-3.3.5 updated (01761acbaa6 -> bae34c42361)
This is an automated email from the ASF dual-hosted git repository.
mthakur pushed a change to branch branch-3.3.5
in repository https://gitbox.apache.org/repos/asf/hadoop.git
from 01761acbaa6 HADOOP-18470. Release hadoop 3.3.5
new 1bc63365a57 HADOOP-18347. S3A Vectored IO to use bounded thread pool. (#4918)
new 052d7bd2275 HADOOP-18463. Add an integration test to process data asynchronously during vectored read. (#4921)
new bae34c42361 HADOOP-18416. fix ITestS3AIOStatisticsContext test failure (#4931)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/hadoop/fs/StreamCapabilities.java | 2 +-
.../contract/AbstractContractVectoredReadTest.java | 68 ++++++++++++++++++++++
.../java/org/apache/hadoop/fs/s3a/Constants.java | 19 ++++++
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 13 ++++-
.../org/apache/hadoop/fs/s3a/S3AInputStream.java | 14 ++---
.../site/markdown/tools/hadoop-aws/performance.md | 23 +++++---
.../hadoop/fs/s3a/ITestS3AIOStatisticsContext.java | 8 +++
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 9 +++
8 files changed, 140 insertions(+), 16 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 02/03: HADOOP-18463. Add an integration test to process data asynchronously during vectored read. (#4921)
Posted by mt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mthakur pushed a commit to branch branch-3.3.5
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 052d7bd2275c2ae8570f4362756802c97ab8e69d
Author: Mukund Thakur <mt...@cloudera.com>
AuthorDate: Wed Sep 28 23:16:47 2022 +0530
HADOOP-18463. Add an integration test to process data asynchronously during vectored read. (#4921)
part of HADOOP-18103.
Contributed by: Mukund Thakur
---
.../contract/AbstractContractVectoredReadTest.java | 68 ++++++++++++++++++++++
1 file changed, 68 insertions(+)
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
index 86b645b9ec9..a39201df249 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
@@ -19,12 +19,17 @@
package org.apache.hadoop.fs.contract;
import java.io.EOFException;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.IntFunction;
import org.assertj.core.api.Assertions;
@@ -42,7 +47,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.impl.FutureIOSupport;
import org.apache.hadoop.io.WeakReferencedElasticByteBufferPool;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+import org.apache.hadoop.util.functional.FutureIO;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS;
import static org.apache.hadoop.fs.contract.ContractTestUtils.assertCapabilities;
import static org.apache.hadoop.fs.contract.ContractTestUtils.assertDatasetEquals;
import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
@@ -364,6 +372,66 @@ public abstract class AbstractContractVectoredReadTest extends AbstractFSContrac
}
}
+ /**
+ * This test creates list of ranges and then submit a readVectored
+ * operation and then uses a separate thread pool to process the
+ * results asynchronously.
+ */
+ @Test
+ public void testVectoredIOEndToEnd() throws Exception {
+ FileSystem fs = getFileSystem();
+ List<FileRange> fileRanges = new ArrayList<>();
+ fileRanges.add(FileRange.createFileRange(8 * 1024, 100));
+ fileRanges.add(FileRange.createFileRange(14 * 1024, 100));
+ fileRanges.add(FileRange.createFileRange(10 * 1024, 100));
+ fileRanges.add(FileRange.createFileRange(2 * 1024 - 101, 100));
+ fileRanges.add(FileRange.createFileRange(40 * 1024, 1024));
+
+ ExecutorService dataProcessor = Executors.newFixedThreadPool(5);
+ CountDownLatch countDown = new CountDownLatch(fileRanges.size());
+
+ try (FSDataInputStream in = fs.open(path(VECTORED_READ_FILE_NAME))) {
+ in.readVectored(fileRanges, value -> pool.getBuffer(true, value));
+ for (FileRange res : fileRanges) {
+ dataProcessor.submit(() -> {
+ try {
+ readBufferValidateDataAndReturnToPool(res, countDown);
+ } catch (Exception e) {
+ String error = String.format("Error while processing result for %s", res);
+ LOG.error(error, e);
+ ContractTestUtils.fail(error, e);
+ }
+ });
+ }
+ // user can perform other computations while waiting for IO.
+ if (!countDown.await(VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ ContractTestUtils.fail("Timeout/Error while processing vectored io results");
+ }
+ } finally {
+ HadoopExecutors.shutdown(dataProcessor, LOG,
+ VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+ }
+
+ private void readBufferValidateDataAndReturnToPool(FileRange res,
+ CountDownLatch countDownLatch)
+ throws IOException, TimeoutException {
+ CompletableFuture<ByteBuffer> data = res.getData();
+ // Read the data and perform custom operation. Here we are just
+ // validating it with original data.
+ FutureIO.awaitFuture(data.thenAccept(buffer -> {
+ assertDatasetEquals((int) res.getOffset(),
+ "vecRead", buffer, res.getLength(), DATASET);
+ // return buffer to the pool once read.
+ pool.putBuffer(buffer);
+ }),
+ VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+ // countdown to notify main thread that processing has been done.
+ countDownLatch.countDown();
+ }
+
+
protected List<FileRange> createSampleNonOverlappingRanges() {
List<FileRange> fileRanges = new ArrayList<>();
fileRanges.add(FileRange.createFileRange(0, 100));
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 01/03: HADOOP-18347. S3A Vectored IO to use bounded thread pool. (#4918)
Posted by mt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mthakur pushed a commit to branch branch-3.3.5
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 1bc63365a57292068a655be03307864419c53304
Author: Mukund Thakur <mt...@cloudera.com>
AuthorDate: Tue Sep 27 21:13:07 2022 +0530
HADOOP-18347. S3A Vectored IO to use bounded thread pool. (#4918)
part of HADOOP-18103.
Also introducing a config fs.s3a.vectored.active.ranged.reads
to configure the maximum number of number of range reads a
single input stream can have active (downloading, or queued)
to the central FileSystem instance's pool of queued operations.
This stops a single stream overloading the shared thread pool.
Contributed by: Mukund Thakur
Conflicts:
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
---
.../org/apache/hadoop/fs/StreamCapabilities.java | 2 +-
.../java/org/apache/hadoop/fs/s3a/Constants.java | 19 ++++++++++++++++++
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 13 +++++++++++-
.../org/apache/hadoop/fs/s3a/S3AInputStream.java | 14 ++++++-------
.../site/markdown/tools/hadoop-aws/performance.md | 23 +++++++++++++++-------
5 files changed, 55 insertions(+), 16 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
index c925e50889d..93ed57ef830 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
@@ -84,7 +84,7 @@ public interface StreamCapabilities {
* Support for vectored IO api.
* See {@code PositionedReadable#readVectored(List, IntFunction)}.
*/
- String VECTOREDIO = "readvectored";
+ String VECTOREDIO = "in:readvectored";
/**
* Stream abort() capability implemented by {@link Abortable#abort()}.
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 764a6adaca2..306fd8e1c67 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -1203,4 +1203,23 @@ public final class Constants {
* Default maximum read size in bytes during vectored reads : {@value}.
*/
public static final int DEFAULT_AWS_S3_VECTOR_READS_MAX_MERGED_READ_SIZE = 1253376; //1M
+
+ /**
+ * Maximum number of range reads a single input stream can have
+ * active (downloading, or queued) to the central FileSystem
+ * instance's pool of queued operations.
+ * This stops a single stream overloading the shared thread pool.
+ * {@value}
+ * <p>
+ * Default is {@link #DEFAULT_AWS_S3_VECTOR_ACTIVE_RANGE_READS}
+ */
+ public static final String AWS_S3_VECTOR_ACTIVE_RANGE_READS =
+ "fs.s3a.vectored.active.ranged.reads";
+
+ /**
+ * Limit of queued range data download operations during vectored
+ * read. Value: {@value}
+ */
+ public static final int DEFAULT_AWS_S3_VECTOR_ACTIVE_RANGE_READS = 4;
+
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index e36214dc1fe..aa6f69c2a2a 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -319,6 +319,12 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
/** Vectored IO context. */
private VectoredIOContext vectoredIOContext;
+ /**
+ * Maximum number of active range read operation a single
+ * input stream can have.
+ */
+ private int vectoredActiveRangeReads;
+
private long readAhead;
private ChangeDetectionPolicy changeDetectionPolicy;
private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -597,6 +603,8 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
longBytesOption(conf, ASYNC_DRAIN_THRESHOLD,
DEFAULT_ASYNC_DRAIN_THRESHOLD, 0),
inputPolicy);
+ vectoredActiveRangeReads = intOption(conf,
+ AWS_S3_VECTOR_ACTIVE_RANGE_READS, DEFAULT_AWS_S3_VECTOR_ACTIVE_RANGE_READS, 1);
vectoredIOContext = populateVectoredIOContext(conf);
} catch (AmazonClientException e) {
// amazon client exception: stop all services then throw the translation
@@ -1504,7 +1512,10 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
createObjectAttributes(path, fileStatus),
createInputStreamCallbacks(auditSpan),
inputStreamStats,
- unboundedThreadPool));
+ new SemaphoredDelegatingExecutor(
+ boundedThreadPool,
+ vectoredActiveRangeReads,
+ true)));
}
/**
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java
index b6ac8669a67..39d41f5ffd2 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java
@@ -27,7 +27,7 @@ import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.IntFunction;
@@ -139,7 +139,7 @@ public class S3AInputStream extends FSInputStream implements CanSetReadahead,
/**
* Thread pool used for vectored IO operation.
*/
- private final ThreadPoolExecutor unboundedThreadPool;
+ private final ExecutorService boundedThreadPool;
private final String bucket;
private final String key;
private final String pathStr;
@@ -196,13 +196,13 @@ public class S3AInputStream extends FSInputStream implements CanSetReadahead,
* @param s3Attributes object attributes
* @param client S3 client to use
* @param streamStatistics stream io stats.
- * @param unboundedThreadPool thread pool to use.
+ * @param boundedThreadPool thread pool to use.
*/
public S3AInputStream(S3AReadOpContext ctx,
S3ObjectAttributes s3Attributes,
InputStreamCallbacks client,
S3AInputStreamStatistics streamStatistics,
- ThreadPoolExecutor unboundedThreadPool) {
+ ExecutorService boundedThreadPool) {
Preconditions.checkArgument(isNotEmpty(s3Attributes.getBucket()),
"No Bucket");
Preconditions.checkArgument(isNotEmpty(s3Attributes.getKey()), "No Key");
@@ -224,7 +224,7 @@ public class S3AInputStream extends FSInputStream implements CanSetReadahead,
setInputPolicy(ctx.getInputPolicy());
setReadahead(ctx.getReadahead());
this.asyncDrainThreshold = ctx.getAsyncDrainThreshold();
- this.unboundedThreadPool = unboundedThreadPool;
+ this.boundedThreadPool = boundedThreadPool;
this.vectoredIOContext = context.getVectoredIOContext();
this.threadIOStatistics = requireNonNull(ctx.getIOStatisticsAggregator());
}
@@ -882,7 +882,7 @@ public class S3AInputStream extends FSInputStream implements CanSetReadahead,
streamStatistics.readVectoredOperationStarted(sortedRanges.size(), sortedRanges.size());
for (FileRange range: sortedRanges) {
ByteBuffer buffer = allocate.apply(range.getLength());
- unboundedThreadPool.submit(() -> readSingleRange(range, buffer));
+ boundedThreadPool.submit(() -> readSingleRange(range, buffer));
}
} else {
LOG.debug("Trying to merge the ranges as they are not disjoint");
@@ -893,7 +893,7 @@ public class S3AInputStream extends FSInputStream implements CanSetReadahead,
LOG.debug("Number of original ranges size {} , Number of combined ranges {} ",
ranges.size(), combinedFileRanges.size());
for (CombinedFileRange combinedFileRange: combinedFileRanges) {
- unboundedThreadPool.submit(
+ boundedThreadPool.submit(
() -> readCombinedRangeAndUpdateChildren(combinedFileRange, allocate));
}
}
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
index 06eb137cd9b..88e6e8a0b21 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
@@ -75,13 +75,22 @@ on the client requirements.
</description>
</property>
<property>
-<name>fs.s3a.vectored.read.max.merged.size</name>
-<value>1M</value>
-<description>
- What is the largest merged read size in bytes such
- that we group ranges together during vectored read.
- Setting this value to 0 will disable merging of ranges.
-</description>
+ <name>fs.s3a.vectored.read.max.merged.size</name>
+ <value>1M</value>
+ <description>
+ What is the largest merged read size in bytes such
+ that we group ranges together during vectored read.
+ Setting this value to 0 will disable merging of ranges.
+ </description>
+<property>
+ <name>fs.s3a.vectored.active.ranged.reads</name>
+ <value>4</value>
+ <description>
+ Maximum number of range reads a single input stream can have
+ active (downloading, or queued) to the central FileSystem
+ instance's pool of queued operations.
+ This stops a single stream overloading the shared thread pool.
+ </description>
</property>
```
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 03/03: HADOOP-18416. fix ITestS3AIOStatisticsContext test failure (#4931)
Posted by mt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mthakur pushed a commit to branch branch-3.3.5
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit bae34c42361769610a56883ee471f8368f9153b0
Author: Mehakmeet Singh <me...@gmail.com>
AuthorDate: Wed Sep 28 14:17:56 2022 +0530
HADOOP-18416. fix ITestS3AIOStatisticsContext test failure (#4931)
Follow on to HADOOP-17461.
Contributed by: Mehakmeet Singh
---
.../org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java | 8 ++++++++
.../src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java | 9 +++++++++
2 files changed, 17 insertions(+)
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java
index d2ae9d7dc3c..70dc5ee476c 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.util.functional.TaskPool;
import static org.apache.hadoop.fs.contract.ContractTestUtils.assertCapabilities;
import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.disablePrefetching;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
@@ -67,6 +68,7 @@ public class ITestS3AIOStatisticsContext extends AbstractS3ATestBase {
@Override
protected Configuration createConfiguration() {
Configuration configuration = super.createConfiguration();
+ disablePrefetching(configuration);
enableIOStatisticsContext();
return configuration;
}
@@ -253,6 +255,7 @@ public class ITestS3AIOStatisticsContext extends AbstractS3ATestBase {
// Worker thread work and wait for it to finish.
TestWorkerThread workerThread = new TestWorkerThread(path, null);
long workerThreadID = workerThread.getId();
+ LOG.info("Worker thread ID: {} ", workerThreadID);
workerThread.start();
workerThread.join();
@@ -463,6 +466,8 @@ public class ITestS3AIOStatisticsContext extends AbstractS3ATestBase {
@Override
public void run() {
+ // Setting the worker thread's name.
+ Thread.currentThread().setName("worker thread");
S3AFileSystem fs = getFileSystem();
byte[] data = new byte[BYTES_SMALL];
@@ -470,6 +475,9 @@ public class ITestS3AIOStatisticsContext extends AbstractS3ATestBase {
if (ioStatisticsContext != null) {
IOStatisticsContext.setThreadIOStatisticsContext(ioStatisticsContext);
}
+ // Storing context in a field to not lose the reference in a GC.
+ IOStatisticsContext ioStatisticsContextWorkerThread =
+ getCurrentIOStatisticsContext();
// Write in the worker thread.
try (FSDataOutputStream out = fs.create(workerThreadPath)) {
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
index 6162ed13123..260c5fc5041 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
@@ -1488,4 +1488,13 @@ public final class S3ATestUtils {
throw new AssertionError("Not an S3AInputStream: " + inner);
}
}
+
+ /**
+ * Disable Prefetching streams from S3AFileSystem in tests.
+ * @param conf Configuration to remove the prefetch property from.
+ */
+ public static void disablePrefetching(Configuration conf) {
+ // TODO: Reinstate once HADOOP-18028 is merged.
+ // removeBaseAndBucketOverrides(conf, PREFETCH_ENABLED_KEY);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org