HADOOP-17450. Add Public IOStatistics API -missed backport (#5590)
This cherrypicks SemaphoredDelegatingExecutor HADOOP-17450 changes from trunk somehow they didn't get into the main IOStatistics backport to branch-3.3
This commit is contained in:
parent
1b59e3123b
commit
21cf507db3
@ -22,6 +22,8 @@
|
|||||||
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Futures;
|
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Futures;
|
||||||
|
|
||||||
import org.apache.hadoop.classification.InterfaceAudience;
|
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.Collection;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@ -33,6 +35,10 @@
|
|||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.TimeoutException;
|
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
|
* This ExecutorService blocks the submission of new tasks when its queue is
|
||||||
* already full by using a semaphore. Task submissions require permits, task
|
* 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 Semaphore queueingPermits;
|
||||||
private final ExecutorService executorDelegatee;
|
private final ExecutorService executorDelegatee;
|
||||||
private final int permitCount;
|
private final int permitCount;
|
||||||
|
private final DurationTrackerFactory trackerFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Instantiate.
|
* Instantiate.
|
||||||
* @param executorDelegatee Executor to delegate to
|
* @param executorDelegatee Executor to delegate to
|
||||||
* @param permitCount number of permits into the queue permitted
|
* @param permitCount number of permits into the queue permitted
|
||||||
* @param fair should the semaphore be "fair"
|
* @param fair should the semaphore be "fair"
|
||||||
|
* @param trackerFactory duration tracker factory.
|
||||||
|
*/
|
||||||
|
public SemaphoredDelegatingExecutor(
|
||||||
|
ExecutorService executorDelegatee,
|
||||||
|
int permitCount,
|
||||||
|
boolean fair,
|
||||||
|
DurationTrackerFactory trackerFactory) {
|
||||||
|
this.permitCount = permitCount;
|
||||||
|
queueingPermits = new Semaphore(permitCount, fair);
|
||||||
|
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(
|
public SemaphoredDelegatingExecutor(
|
||||||
ExecutorService executorDelegatee,
|
ExecutorService executorDelegatee,
|
||||||
int permitCount,
|
int permitCount,
|
||||||
boolean fair) {
|
boolean fair) {
|
||||||
this.permitCount = permitCount;
|
this(executorDelegatee, permitCount, fair, null);
|
||||||
queueingPermits = new Semaphore(permitCount, fair);
|
|
||||||
this.executorDelegatee = executorDelegatee;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -102,7 +127,8 @@ public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout,
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public <T> Future<T> submit(Callable<T> task) {
|
public <T> Future<T> submit(Callable<T> task) {
|
||||||
try {
|
try (DurationTracker ignored =
|
||||||
|
trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
|
||||||
queueingPermits.acquire();
|
queueingPermits.acquire();
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
@ -113,7 +139,8 @@ public <T> Future<T> submit(Callable<T> task) {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public <T> Future<T> submit(Runnable task, T result) {
|
public <T> Future<T> submit(Runnable task, T result) {
|
||||||
try {
|
try (DurationTracker ignored =
|
||||||
|
trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
|
||||||
queueingPermits.acquire();
|
queueingPermits.acquire();
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
@ -124,7 +151,8 @@ public <T> Future<T> submit(Runnable task, T result) {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Future<?> submit(Runnable task) {
|
public Future<?> submit(Runnable task) {
|
||||||
try {
|
try (DurationTracker ignored =
|
||||||
|
trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
|
||||||
queueingPermits.acquire();
|
queueingPermits.acquire();
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
@ -135,7 +163,8 @@ public Future<?> submit(Runnable task) {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void execute(Runnable command) {
|
public void execute(Runnable command) {
|
||||||
try {
|
try (DurationTracker ignored =
|
||||||
|
trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
|
||||||
queueingPermits.acquire();
|
queueingPermits.acquire();
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
|
Loading…
Reference in New Issue
Block a user