This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch feature-HADOOP-18028-s3a-prefetch-branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit caf58937a0a7691c24199ef81be23d48e158a320
Author: Steve Loughran <ste...@cloudera.com>
AuthorDate: Wed Aug 31 11:16:52 2022 +0100
HADOOP-18410. S3AInputStream.unbuffer() does not release http connections
-prefetch changes(#4766)
Changes in HADOOP-18410 which are needed for the S3A prefetching stream;
needed
as part of the HADOOP-18703 backport
Change-Id: Ib403ca793e29a4416e5d892f9081de5832da3b68
---
.../hadoop/fs/s3a/prefetch/S3ARemoteObject.java | 108 +++------------------
.../fs/s3a/performance/ITestUnbufferDraining.java | 2 +
2 files changed, 17 insertions(+), 93 deletions(-)
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/prefetch/S3ARemoteObject.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/prefetch/S3ARemoteObject.java
index d749e9df02f..3ab0022bb08 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/prefetch/S3ARemoteObject.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/prefetch/S3ARemoteObject.java
@@ -27,6 +27,7 @@ import java.util.Map;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
+import com.amazonaws.services.s3.model.S3ObjectInputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,12 +37,10 @@ import org.apache.hadoop.fs.s3a.S3AInputStream;
import org.apache.hadoop.fs.s3a.S3AReadOpContext;
import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
+import org.apache.hadoop.fs.s3a.impl.SDKStreamDrainer;
import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
import org.apache.hadoop.fs.statistics.DurationTracker;
-import static
org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration;
-import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
-
/**
* Encapsulates low level interactions with S3 object on AWS.
*/
@@ -79,7 +78,7 @@ public class S3ARemoteObject {
* Maps a stream returned by openForRead() to the associated S3 object.
* That allows us to close the object when closing the stream.
*/
- private Map<InputStream, S3Object> s3Objects;
+ private final Map<InputStream, S3Object> s3Objects;
/**
* uri of the object being read.
@@ -225,104 +224,27 @@ public class S3ARemoteObject {
void close(InputStream inputStream, int numRemainingBytes) {
S3Object obj;
synchronized (s3Objects) {
- obj = s3Objects.get(inputStream);
+ obj = s3Objects.remove(inputStream);
if (obj == null) {
throw new IllegalArgumentException("inputStream not found");
}
- s3Objects.remove(inputStream);
}
-
+ SDKStreamDrainer drainer = new SDKStreamDrainer(
+ uri,
+ obj,
+ (S3ObjectInputStream)inputStream,
+ false,
+ numRemainingBytes,
+ streamStatistics,
+ "close() operation");
if (numRemainingBytes <= context.getAsyncDrainThreshold()) {
// don't bother with async io.
- drain(false, "close() operation", numRemainingBytes, obj, inputStream);
+ drainer.apply();
} else {
LOG.debug("initiating asynchronous drain of {} bytes",
numRemainingBytes);
- // schedule an async drain/abort with references to the fields so they
- // can be reused
- client.submit(
- () -> drain(false, "close() operation", numRemainingBytes, obj,
- inputStream));
+ // schedule an async drain/abort
+ client.submit(drainer);
}
}
- /**
- * drain the stream. This method is intended to be
- * used directly or asynchronously, and measures the
- * duration of the operation in the stream statistics.
- *
- * @param shouldAbort force an abort; used if explicitly requested.
- * @param reason reason for stream being closed; used in messages
- * @param remaining remaining bytes
- * @param requestObject http request object;
- * @param inputStream stream to close.
- * @return was the stream aborted?
- */
- private boolean drain(
- final boolean shouldAbort,
- final String reason,
- final long remaining,
- final S3Object requestObject,
- final InputStream inputStream) {
-
- try {
- return invokeTrackingDuration(
- streamStatistics.initiateInnerStreamClose(shouldAbort),
- () -> drainOrAbortHttpStream(shouldAbort, reason, remaining,
- requestObject, inputStream));
- } catch (IOException e) {
- // this is only here because invokeTrackingDuration() has it in its
- // signature
- return shouldAbort;
- }
- }
-
- /**
- * Drain or abort the inner stream.
- * Exceptions are swallowed.
- * If a close() is attempted and fails, the operation escalates to
- * an abort.
- *
- * @param shouldAbort force an abort; used if explicitly requested.
- * @param reason reason for stream being closed; used in messages
- * @param remaining remaining bytes
- * @param requestObject http request object
- * @param inputStream stream to close.
- * @return was the stream aborted?
- */
- private boolean drainOrAbortHttpStream(
- boolean shouldAbort,
- final String reason,
- final long remaining,
- final S3Object requestObject,
- final InputStream inputStream) {
-
- if (!shouldAbort && remaining > 0) {
- try {
- long drained = 0;
- byte[] buffer = new byte[DRAIN_BUFFER_SIZE];
- while (true) {
- final int count = inputStream.read(buffer);
- if (count < 0) {
- // no more data is left
- break;
- }
- drained += count;
- }
- LOG.debug("Drained stream of {} bytes", drained);
- } catch (Exception e) {
- // exception escalates to an abort
- LOG.debug("When closing {} stream for {}, will abort the stream", uri,
- reason, e);
- shouldAbort = true;
- }
- }
- cleanupWithLogger(LOG, inputStream);
- cleanupWithLogger(LOG, requestObject);
- streamStatistics.streamClose(shouldAbort, remaining);
-
- LOG.debug("Stream {} {}: {}; remaining={}", uri,
- (shouldAbort ? "aborted" : "closed"), reason,
- remaining);
- return shouldAbort;
- }
}
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
index 0295c07e567..a03f181cb38 100644
---
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
@@ -45,6 +45,7 @@ import static
org.apache.hadoop.fs.s3a.Constants.ESTABLISH_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADVISE;
import static org.apache.hadoop.fs.s3a.Constants.MAXIMUM_CONNECTIONS;
import static org.apache.hadoop.fs.s3a.Constants.MAX_ERROR_RETRIES;
+import static org.apache.hadoop.fs.s3a.Constants.PREFETCH_ENABLED_KEY;
import static org.apache.hadoop.fs.s3a.Constants.READAHEAD_RANGE;
import static org.apache.hadoop.fs.s3a.Constants.REQUEST_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.RETRY_LIMIT;
@@ -102,6 +103,7 @@ public class ITestUnbufferDraining extends
AbstractS3ACostTest {
INPUT_FADVISE,
MAX_ERROR_RETRIES,
MAXIMUM_CONNECTIONS,
+ PREFETCH_ENABLED_KEY,
READAHEAD_RANGE,
REQUEST_TIMEOUT,
RETRY_LIMIT,
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-commits-h...@hadoop.apache.org
