Field updates happen only in blocks that are synchronized on the {@link * AbstractQueueVisitor} object. * *
If {@link AbstractQueueVisitor} clients don't like the semantics of storing and propagating * the most severe error, then they should be provide an {@link ErrorClassifier} that does the * right thing (e.g. to cause the _first_ error to be propagated, you'd want to provide an {@link * ErrorClassifier} that gives all errors the exact same {@link ErrorClassification}). * *
Note that this is not a performance-critical path.
*/
private volatile Throwable unhandled = null;
/**
* An uncaught exception when submitting a job to the {@link ExecutorService} is catastrophic, and
* usually indicates a lack of stack space on which to allocate a native thread. The {@link
* ExecutorService} may reach an inconsistent state in such circumstances, so we avoid blocking on
* its termination when this field is non-{@code null}.
*/
private volatile Throwable catastrophe;
/**
* An object used in the manner of a {@link java.util.concurrent.locks.Condition} object, for the
* condition {@code remainingTasks.get() == 0 || jobsMustBeStopped}. TODO(bazel-team): Replace
* with an actual {@link java.util.concurrent.locks.Condition} object.
*/
private final Object zeroRemainingTasks = new Object();
/** The number of {@link Runnable}s {@link #execute}-d that have not finished evaluation. */
private final AtomicLong remainingTasks = new AtomicLong(0);
/**
* A thread that wants to add or remove a future from {@code outstandingFutures} must first obtain
* the read lock and then check the {@code threadInterrupted} flag. The thread that
* cancels all futures must first set the {@code threadInterrupted} flag, and then obtain the
* write lock. Only once both have happened is the main thread allowed to iterate over
* the {@code outstandingFutures}. This allows concurrent future registration, but ensures that
* canceling only happens when there are no concurrent modifications to the set since that
* requires iterating over all elements.
*/
private final ReadWriteLock outstandingFuturesLock = new ReentrantReadWriteLock();
private final Set Except for {@link #mustJobsBeStopped}, may only be accessed in a block that is synchronized
* on {@link #zeroRemainingTasks}.
*/
private volatile boolean jobsMustBeStopped = false;
/** Map from thread to number of jobs executing in the thread. Used for interrupt handling. */
private final Map When this is {@code true}, adding tasks to the {@link ExecutorService} will fail quietly as
* a part of the process of shutting down the worker threads.
*/
private volatile boolean threadInterrupted;
/**
* Latches used to signal when the visitor has been interrupted or seen an exception. Used only
* for testing.
*/
private final CountDownLatch interruptedLatch = new CountDownLatch(1);
private final CountDownLatch exceptionLatch = new CountDownLatch(1);
/** If {@code true}, don't run new actions after an uncaught exception. */
private final ExceptionHandlingMode exceptionHandlingMode;
/** If {@code true}, shut down the {@link ExecutorService} on completion. */
private final ExecutorOwnership executorOwnership;
private final ErrorClassifier errorClassifier;
private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
/**
* Default function for constructing {@link ThreadPoolExecutor}s. The {@link ThreadPoolExecutor}s
* this creates have the same value for {@code corePoolSize} and {@code maximumPoolSize} because
* that results in a fixed-size thread pool, and the current use cases for {@link
* AbstractQueueVisitor} don't require any more sophisticated thread pool size management.
*
* If client use cases change, they may invoke one of the {@link
* AbstractQueueVisitor#AbstractQueueVisitor} constructors that accepts a pre-constructed {@link
* ThreadPoolExecutor}.
*/
private static ExecutorService createExecutorService(
int parallelism,
long keepAliveTime,
TimeUnit units,
BlockingQueue Callers must not shut down the {@link ExecutorService} while queue visitors use it.
*/
PRIVATE,
/** Keep the executor running after the visitation is done. */
SHARED
}
/** What to do if a task throws an uncaught exception. */
public enum ExceptionHandlingMode {
/** Don't run new tasks after one throws an uncaught exception. */
FAIL_FAST,
/** Keep running new tasks when one throws an uncaught exception. */
KEEP_GOING,
}
/**
* Create the AbstractQueueVisitor.
*
* @param executorService the {@link ExecutorService} to use.
* @param executorOwnership whether the {@link AbstractQueueVisitor} being created owns the {@link
* ExecutorService} it uses.
* @param exceptionHandlingMode what to do when a task throws an uncaught exception.
* @param errorClassifier an error classifier used to determine whether to log and/or stop jobs.
*/
protected AbstractQueueVisitor(
ExecutorService executorService,
ExecutorOwnership executorOwnership,
ExceptionHandlingMode exceptionHandlingMode,
ErrorClassifier errorClassifier) {
this.exceptionHandlingMode = exceptionHandlingMode;
this.executorOwnership = executorOwnership;
this.executorService = Preconditions.checkNotNull(executorService);
this.errorClassifier = Preconditions.checkNotNull(errorClassifier);
}
@Override
public final void awaitQuiescence(boolean interruptWorkers) throws InterruptedException {
Throwables.propagateIfPossible(catastrophe);
try {
synchronized (zeroRemainingTasks) {
while (remainingTasks.get() != 0 && !jobsMustBeStopped) {
zeroRemainingTasks.wait();
}
}
} catch (InterruptedException e) {
// Mark the visitor, so that it's known to be interrupted, and
// then break out of here, stop the worker threads and return ASAP,
// sending the interruption to the parent thread.
setInterrupted();
}
awaitTermination(interruptWorkers);
}
/** Schedules a call. Called in a worker thread. */
@Override
public final void execute(Runnable runnable) {
executeWithExecutorService(runnable, executorService);
}
protected void executeWithExecutorService(Runnable runnable, ExecutorService executorService) {
WrappedRunnable wrappedRunnable = new WrappedRunnable(runnable);
try {
// It's impossible for this increment to result in remainingTasks.get <= 0 because
// remainingTasks is never negative. Therefore it isn't necessary to check its value for
// the purpose of updating zeroRemainingTasks.
long tasks = remainingTasks.incrementAndGet();
Preconditions.checkState(
tasks > 0,
"Incrementing remaining tasks counter resulted in impossible non-positive number.");
executeWrappedRunnable(wrappedRunnable, executorService);
} catch (Throwable e) {
if (!wrappedRunnable.ran) {
// Note that keeping track of ranTask is necessary to disambiguate the case where
// execute() itself failed, vs. a caller-runs policy on pool exhaustion, where the
// runnable threw. To be extra cautious, we decrement the task count in a finally
// block, even though the CountDownLatch is unlikely to throw.
recordError(e);
}
}
}
protected void executeWrappedRunnable(WrappedRunnable runnable, ExecutorService executorService) {
executorService.execute(runnable);
}
private synchronized void maybeSaveUnhandledThrowable(Throwable e, boolean markToStopJobs) {
boolean critical = false;
ErrorClassification errorClassification = errorClassifier.classify(e);
switch (errorClassification) {
case AS_CRITICAL_AS_POSSIBLE:
case CRITICAL_AND_LOG:
critical = true;
logger.atWarning().withCause(e).log("Found critical error in queue visitor");
break;
case CRITICAL:
critical = true;
break;
default:
break;
}
if (unhandled == null
|| errorClassification.compareTo(errorClassifier.classify(unhandled)) > 0) {
// Save the most severe error.
unhandled = e;
exceptionLatch.countDown();
}
if (markToStopJobs) {
synchronized (zeroRemainingTasks) {
if (critical && !jobsMustBeStopped) {
jobsMustBeStopped = true;
// This introduces a benign race, but it's the best we can do. When we have multiple
// errors of the same severity that is at least CRITICAL, we'll end up saving (above) and
// propagating (in 'awaitQuiescence') the most severe one we see, but the set of errors we
// see is non-deterministic and is at the mercy of how quickly the calling thread of
// 'awaitQuiescence' can do its thing after this 'notify' call.
zeroRemainingTasks.notify();
}
}
}
}
private void recordError(Throwable e) {
try {
// If threadInterrupted is true, then RejectedExecutionExceptions are expected. There's no
// need to remember them, but there is a need to call decrementRemainingTasks, which is
// satisfied by the finally block below.
if (e instanceof RejectedExecutionException && threadInterrupted) {
return;
}
catastrophe = e;
maybeSaveUnhandledThrowable(e, /*markToStopJobs=*/ false);
} finally {
decrementRemainingTasks();
}
}
/**
* A wrapped {@link Runnable} that:
*
* This function returns the CURRENT state of whether jobs should be stopped. If the value is
* false right now, it may be changed to true by another thread later.
*/
protected final boolean mustJobsBeStopped() {
return jobsMustBeStopped;
}
/**
* Waits for the task queue to drain. Then if {@code ownExecutorService} is true, shuts down the
* {@link ExecutorService} and waits for it to terminate. Throws (the same) unchecked exception if
* any worker thread failed unexpectedly.
*/
protected final void awaitTermination(boolean interruptWorkers) throws InterruptedException {
reallyAwaitTermination(interruptWorkers);
if (isInterrupted()) {
// Set interrupted bit on current thread so that callers can see that it was interrupted. Note
// that if the thread was interrupted while awaiting termination, we might not hit this
// code path, but then the current thread's interrupt bit is already set, so we are fine.
Thread.currentThread().interrupt();
}
// Throw the first unhandled (worker thread) exception in the main thread. We throw an unchecked
// exception instead of InterruptedException if both are present because an unchecked exception
// may indicate a catastrophic failure that should shut down the program. The caller can
// check the interrupted bit if they will handle the unchecked exception without crashing.
Throwables.propagateIfPossible(unhandled);
if (Thread.interrupted()) {
throw new InterruptedException();
}
}
private void reallyAwaitTermination(boolean interruptWorkers) {
// TODO(bazel-team): verify that interrupt() is safe for every use of
// AbstractQueueVisitor and remove the interruptWorkers flag.
if (interruptWorkers && !jobs.isEmpty()) {
interruptInFlightTasks();
}
if (interruptWorkers) {
// If the computation is done, this does nothing because there are no outstanding futures. Do
// not predicate on outstandingFutures.isEmpty() here: in the case of an interrupt, there may
// still be threads concurrently adding futures to the set, and we need to make sure that
// those are canceled correctly.
cancelAllFutures();
}
if (isInterrupted()) {
interruptedLatch.countDown();
}
Throwables.propagateIfPossible(catastrophe);
synchronized (zeroRemainingTasks) {
while (remainingTasks.get() != 0) {
try {
zeroRemainingTasks.wait();
} catch (InterruptedException e) {
setInterrupted();
}
}
}
if (executorOwnership == ExecutorOwnership.PRIVATE) {
shutdownExecutorService(catastrophe);
}
}
protected void shutdownExecutorService(Throwable catastrophe) {
executorService.shutdown();
for (; ; ) {
try {
Throwables.propagateIfPossible(catastrophe);
executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
break;
} catch (InterruptedException e) {
setInterrupted();
}
}
}
private void interruptInFlightTasks() {
Thread thisThread = Thread.currentThread();
for (Thread thread : jobs.keySet()) {
if (thisThread != thread) {
thread.interrupt();
}
}
}
private void cancelAllFutures() {
// Nobody else can modify outstandingFutures while this thread is holding the write lock.
outstandingFuturesLock.writeLock().lock();
try {
for (ListenableFuture future : outstandingFutures) {
future.cancel(/*mayInterruptIfRunning=*/ true);
}
outstandingFutures.clear();
} finally {
outstandingFuturesLock.writeLock().unlock();
}
}
}
*
*/
protected final class WrappedRunnable implements Runnable, Comparable