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 st...@apache.org on 2023/04/24 18:08:02 UTC
[hadoop] branch feature-HADOOP-18028-s3a-prefetch-branch-3.3 updated (f1fbac411d3 -> 5569b415935)
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a change to branch feature-HADOOP-18028-s3a-prefetch-branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
from f1fbac411d3 HADOOP-18380. fs.s3a.prefetch.block.size to be read through longBytesOption (#4762)
new 8e811394fe7 HADOOP-17450. Add Public IOStatistics API -missed backport
new f633c032610 HADOOP-18186. s3a prefetching to use SemaphoredDelegatingExecutor for submitting work (#4796)
new 5569b415935 HADOOP-18455. S3A prefetching executor should be closed (#4879)
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:
.../impl/prefetch/ExecutorServiceFuturePool.java | 20 ++++++++++-
.../hadoop/util/SemaphoredDelegatingExecutor.java | 41 ++++++++++++++++++----
.../hadoop/util/concurrent/HadoopExecutors.java | 5 ++-
.../hadoop-aws/dev-support/findbugs-exclude.xml | 4 +++
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 27 ++++++++------
.../fs/s3a/ITestS3APrefetchingInputStream.java | 9 +++++
6 files changed, 86 insertions(+), 20 deletions(-)
---------------------------------------------------------------------
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-17450. Add Public IOStatistics API -missed backport
Posted by st...@apache.org.
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 8e811394fe75e2bcb3aef30c9cb8947a752517f5
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Mon Apr 24 19:06:50 2023 +0100
HADOOP-17450. Add Public IOStatistics API -missed backport
This cherrypicks SemaphoredDelegatingExecutor from the trunk
commit; somehow it didn't get into the backport to branch-3.3
Change-Id: I8d4a91a92828755594a6c45e37095c372b47cc26
---
.../hadoop/util/SemaphoredDelegatingExecutor.java | 41 ++++++++++++++++++----
1 file changed, 35 insertions(+), 6 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java
index 45b9a98c68a..c4c11e57b37 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java
@@ -22,6 +22,8 @@ import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Forwarding
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Futures;
import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.fs.statistics.DurationTracker;
+import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
import java.util.Collection;
import java.util.List;
@@ -33,6 +35,10 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import static java.util.Objects.requireNonNull;
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.stubDurationTrackerFactory;
+import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_EXECUTOR_ACQUIRED;
+
/**
* This ExecutorService blocks the submission of new tasks when its queue is
* already full by using a semaphore. Task submissions require permits, task
@@ -53,20 +59,39 @@ public class SemaphoredDelegatingExecutor extends
private final Semaphore queueingPermits;
private final ExecutorService executorDelegatee;
private final int permitCount;
+ private final DurationTrackerFactory trackerFactory;
/**
* Instantiate.
* @param executorDelegatee Executor to delegate to
* @param permitCount number of permits into the queue permitted
* @param fair should the semaphore be "fair"
+ * @param trackerFactory duration tracker factory.
*/
public SemaphoredDelegatingExecutor(
ExecutorService executorDelegatee,
int permitCount,
- boolean fair) {
+ boolean fair,
+ DurationTrackerFactory trackerFactory) {
this.permitCount = permitCount;
queueingPermits = new Semaphore(permitCount, fair);
- this.executorDelegatee = executorDelegatee;
+ this.executorDelegatee = requireNonNull(executorDelegatee);
+ this.trackerFactory = trackerFactory != null
+ ? trackerFactory
+ : stubDurationTrackerFactory();
+ }
+
+ /**
+ * Instantiate without collecting executor aquisition duration information.
+ * @param executorDelegatee Executor to delegate to
+ * @param permitCount number of permits into the queue permitted
+ * @param fair should the semaphore be "fair"
+ */
+ public SemaphoredDelegatingExecutor(
+ ExecutorService executorDelegatee,
+ int permitCount,
+ boolean fair) {
+ this(executorDelegatee, permitCount, fair, null);
}
@Override
@@ -102,7 +127,8 @@ public class SemaphoredDelegatingExecutor extends
@Override
public <T> Future<T> submit(Callable<T> task) {
- try {
+ try (DurationTracker ignored =
+ trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
queueingPermits.acquire();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -113,7 +139,8 @@ public class SemaphoredDelegatingExecutor extends
@Override
public <T> Future<T> submit(Runnable task, T result) {
- try {
+ try (DurationTracker ignored =
+ trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
queueingPermits.acquire();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -124,7 +151,8 @@ public class SemaphoredDelegatingExecutor extends
@Override
public Future<?> submit(Runnable task) {
- try {
+ try (DurationTracker ignored =
+ trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
queueingPermits.acquire();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -135,7 +163,8 @@ public class SemaphoredDelegatingExecutor extends
@Override
public void execute(Runnable command) {
- try {
+ try (DurationTracker ignored =
+ trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
queueingPermits.acquire();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
---------------------------------------------------------------------
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-18455. S3A prefetching executor should be closed (#4879)
Posted by st...@apache.org.
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 5569b415935d376f64bc5161f202cc4d6b394665
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Wed Sep 21 11:52:41 2022 -0700
HADOOP-18455. S3A prefetching executor should be closed (#4879)
follow-on patch to HADOOP-18186.
Contributed by: Viraj Jasani
---
.../fs/impl/prefetch/ExecutorServiceFuturePool.java | 20 +++++++++++++++++++-
.../hadoop/util/concurrent/HadoopExecutors.java | 5 ++---
.../hadoop-aws/dev-support/findbugs-exclude.xml | 4 ++++
.../java/org/apache/hadoop/fs/s3a/S3AFileSystem.java | 10 ++++------
4 files changed, 29 insertions(+), 10 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/ExecutorServiceFuturePool.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/ExecutorServiceFuturePool.java
index 9ef50e50d7e..645de280394 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/ExecutorServiceFuturePool.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/ExecutorServiceFuturePool.java
@@ -22,8 +22,13 @@ package org.apache.hadoop.fs.impl.prefetch;
import java.util.Locale;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
+import org.slf4j.Logger;
+
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+
/**
* A FuturePool implementation backed by a java.util.concurrent.ExecutorService.
*
@@ -37,7 +42,8 @@ import java.util.function.Supplier;
*
*/
public class ExecutorServiceFuturePool {
- private ExecutorService executor;
+
+ private final ExecutorService executor;
public ExecutorServiceFuturePool(ExecutorService executor) {
this.executor = executor;
@@ -64,6 +70,18 @@ public class ExecutorServiceFuturePool {
return (Future<Void>) executor.submit(r::run);
}
+ /**
+ * Utility to shutdown the {@link ExecutorService} used by this class. Will wait up to a
+ * certain timeout for the ExecutorService to gracefully shutdown.
+ *
+ * @param logger Logger
+ * @param timeout the maximum time to wait
+ * @param unit the time unit of the timeout argument
+ */
+ public void shutdown(Logger logger, long timeout, TimeUnit unit) {
+ HadoopExecutors.shutdown(executor, logger, timeout, unit);
+ }
+
public String toString() {
return String.format(Locale.ROOT, "ExecutorServiceFuturePool(executor=%s)", executor);
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopExecutors.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopExecutors.java
index 0bbceb59c31..6e2838bfe9c 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopExecutors.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopExecutors.java
@@ -115,9 +115,8 @@ public final class HadoopExecutors {
try {
executorService.shutdown();
- logger.debug(
- "Gracefully shutting down executor service. Waiting max {} {}",
- timeout, unit);
+ logger.debug("Gracefully shutting down executor service {}. Waiting max {} {}",
+ executorService, timeout, unit);
if (!executorService.awaitTermination(timeout, unit)) {
logger.debug(
"Executor service has not shutdown yet. Forcing. "
diff --git a/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml b/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
index e823840fd71..b8e419f3c88 100644
--- a/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
+++ b/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
@@ -59,6 +59,10 @@
<Method name="openFileWithOptions"/>
<Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE"/>
</Match>
+ <Match>
+ <Class name="org.apache.hadoop.fs.s3a.S3AFileSystem"/>
+ <Bug pattern="IS2_INCONSISTENT_SYNC"/>
+ </Match>
<Match>
<Class name="org.apache.hadoop.fs.s3a.s3guard.S3GuardTool$BucketInfo"/>
<Method name="run"/>
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 29cd158641f..bdb444f8f0c 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
@@ -653,17 +653,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
// amazon client exception: stop all services then throw the translation
cleanupWithLogger(LOG, span);
stopAllServices();
- if (this.futurePool != null) {
- this.futurePool = null;
- }
throw translateException("initializing ", new Path(name), e);
} catch (IOException | RuntimeException e) {
// other exceptions: stop the services.
cleanupWithLogger(LOG, span);
stopAllServices();
- if (this.futurePool != null) {
- this.futurePool = null;
- }
throw e;
}
}
@@ -4056,6 +4050,10 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
HadoopExecutors.shutdown(unboundedThreadPool, LOG,
THREAD_POOL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS);
unboundedThreadPool = null;
+ if (futurePool != null) {
+ futurePool.shutdown(LOG, THREAD_POOL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS);
+ futurePool = null;
+ }
// other services are shutdown.
cleanupWithLogger(LOG,
instrumentation,
---------------------------------------------------------------------
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-18186. s3a prefetching to use SemaphoredDelegatingExecutor for submitting work (#4796)
Posted by st...@apache.org.
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 f633c032610501fe66df1e10c8e206ef8a4cf72b
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Fri Sep 9 03:32:20 2022 -0700
HADOOP-18186. s3a prefetching to use SemaphoredDelegatingExecutor for submitting work (#4796)
Contributed by Viraj Jasani
---
.../java/org/apache/hadoop/fs/s3a/S3AFileSystem.java | 17 +++++++++++++----
.../hadoop/fs/s3a/ITestS3APrefetchingInputStream.java | 9 +++++++++
2 files changed, 22 insertions(+), 4 deletions(-)
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 dd9f3368c59..29cd158641f 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
@@ -786,9 +786,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
DEFAULT_KEEPALIVE_TIME, 0);
int numPrefetchThreads = this.prefetchEnabled ? this.prefetchBlockCount : 0;
+ int activeTasksForBoundedThreadPool = maxThreads;
+ int waitingTasksForBoundedThreadPool = maxThreads + totalTasks + numPrefetchThreads;
boundedThreadPool = BlockingThreadPoolExecutorService.newInstance(
- maxThreads,
- maxThreads + totalTasks + numPrefetchThreads,
+ activeTasksForBoundedThreadPool,
+ waitingTasksForBoundedThreadPool,
keepAliveTime, TimeUnit.SECONDS,
name + "-bounded");
unboundedThreadPool = new ThreadPoolExecutor(
@@ -800,8 +802,15 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
unboundedThreadPool.allowCoreThreadTimeOut(true);
executorCapacity = intOption(conf,
EXECUTOR_CAPACITY, DEFAULT_EXECUTOR_CAPACITY, 1);
- if (this.prefetchEnabled) {
- this.futurePool = new ExecutorServiceFuturePool(boundedThreadPool);
+ if (prefetchEnabled) {
+ final S3AInputStreamStatistics s3AInputStreamStatistics =
+ statisticsContext.newInputStreamStatistics();
+ futurePool = new ExecutorServiceFuturePool(
+ new SemaphoredDelegatingExecutor(
+ boundedThreadPool,
+ activeTasksForBoundedThreadPool + waitingTasksForBoundedThreadPool,
+ true,
+ s3AInputStreamStatistics));
}
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3APrefetchingInputStream.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3APrefetchingInputStream.java
index 36d049cedf1..24f74b3a021 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3APrefetchingInputStream.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3APrefetchingInputStream.java
@@ -38,8 +38,10 @@ import org.apache.hadoop.fs.statistics.StreamStatisticNames;
import static org.apache.hadoop.fs.s3a.Constants.PREFETCH_BLOCK_DEFAULT_SIZE;
import static org.apache.hadoop.fs.s3a.Constants.PREFETCH_BLOCK_SIZE_KEY;
import static org.apache.hadoop.fs.s3a.Constants.PREFETCH_ENABLED_KEY;
+import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticMaximum;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticGaugeValue;
+import static org.apache.hadoop.fs.statistics.StoreStatisticNames.SUFFIX_MAX;
import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
/**
@@ -130,6 +132,8 @@ public class ITestS3APrefetchingInputStream extends AbstractS3ACostTest {
}
// Verify that once stream is closed, all memory is freed
verifyStatisticGaugeValue(ioStats, StreamStatisticNames.STREAM_READ_ACTIVE_MEMORY_IN_USE, 0);
+ assertThatStatisticMaximum(ioStats,
+ StoreStatisticNames.ACTION_EXECUTOR_ACQUIRED + SUFFIX_MAX).isGreaterThan(0);
}
@Test
@@ -159,6 +163,8 @@ public class ITestS3APrefetchingInputStream extends AbstractS3ACostTest {
}
verifyStatisticGaugeValue(ioStats, StreamStatisticNames.STREAM_READ_BLOCKS_IN_FILE_CACHE, 0);
verifyStatisticGaugeValue(ioStats, StreamStatisticNames.STREAM_READ_ACTIVE_MEMORY_IN_USE, 0);
+ assertThatStatisticMaximum(ioStats,
+ StoreStatisticNames.ACTION_EXECUTOR_ACQUIRED + SUFFIX_MAX).isGreaterThan(0);
}
@Test
@@ -183,6 +189,9 @@ public class ITestS3APrefetchingInputStream extends AbstractS3ACostTest {
verifyStatisticCounterValue(ioStats, StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS, 0);
// The buffer pool is not used
verifyStatisticGaugeValue(ioStats, StreamStatisticNames.STREAM_READ_ACTIVE_MEMORY_IN_USE, 0);
+ // no prefetch ops, so no action_executor_acquired
+ assertThatStatisticMaximum(ioStats,
+ StoreStatisticNames.ACTION_EXECUTOR_ACQUIRED + SUFFIX_MAX).isEqualTo(-1);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org