From f07b9540dc11c667686cfea24849659decc5a0d8 Mon Sep 17 00:00:00 2001 From: cpovirk Date: Wed, 22 Jul 2020 08:40:53 -0700 Subject: [PATCH] Add awaitTerminationUninterruptibly. Fixes #3908 Fixes #1315 Somewhat relevant to https://github.com/google/guava/issues/3655 Vaguely relevant to https://github.com/google/error-prone/issues/1490, since it creates a `@CheckReturnValue` variant of a `java.util.concurrent` method that returns `false` to indicate timeout. RELNOTES=`util.concurrent`: Added `awaitTerminationUninterruptibly`. ------------- Created by MOE: https://github.com/google/moe MOE_MIGRATED_REVID=322581454 --- .../util/concurrent/UninterruptiblesTest.java | 43 +++++++++++++ .../util/concurrent/Uninterruptibles.java | 47 ++++++++++++++ .../util/concurrent/UninterruptiblesTest.java | 64 +++++++++++++++++++ .../util/concurrent/Uninterruptibles.java | 59 +++++++++++++++++ 4 files changed, 213 insertions(+) diff --git a/android/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java b/android/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java index 976e2ee9f4f4..e58cf6a60ef1 100644 --- a/android/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java +++ b/android/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java @@ -17,12 +17,14 @@ package com.google.common.util.concurrent; import static com.google.common.util.concurrent.InterruptionUtil.repeatedlyInterruptTestThread; +import static com.google.common.util.concurrent.Uninterruptibles.awaitTerminationUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.joinUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.putUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.takeUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.tryAcquireUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.tryLockUninterruptibly; +import static java.util.concurrent.Executors.newFixedThreadPool; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; @@ -36,6 +38,7 @@ import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; @@ -465,6 +468,37 @@ public void testTryAcquireTimeoutMultiInterruptExpiredMultiPermit() { assertInterrupted(); } + // executor.awaitTermination Testcases + public void testTryAwaitTerminationUninterruptiblyLongTimeUnit_success() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(1000)); + executor.shutdown(); + assertTrue(awaitTerminationUninterruptibly(executor, LONG_DELAY_MS, MILLISECONDS)); + assertTrue(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationUninterruptiblyLongTimeUnit_failure() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(10000)); + executor.shutdown(); + assertFalse(awaitTerminationUninterruptibly(executor, 1000, MILLISECONDS)); + assertFalse(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationInfiniteTimeout() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(1000)); + executor.shutdown(); + awaitTerminationUninterruptibly(executor); + assertTrue(executor.isTerminated()); + assertInterrupted(); + } + /** * Wrapper around {@link Stopwatch} which also contains an "expected completion time." Creating a * {@code Completion} starts the underlying stopwatch. @@ -754,6 +788,15 @@ protected void doAction() { } } + private static final class SleepTask extends DelayedActionRunnable { + SleepTask(long tMinus) { + super(tMinus); + } + + @Override + protected void doAction() {} + } + private static void sleepSuccessfully(long sleepMillis) { Completion completed = new Completion(sleepMillis - SLEEP_SLACK); Uninterruptibles.sleepUninterruptibly(sleepMillis, MILLISECONDS); diff --git a/android/guava/src/com/google/common/util/concurrent/Uninterruptibles.java b/android/guava/src/com/google/common/util/concurrent/Uninterruptibles.java index c7a8e32dafa2..5860a9d8f95f 100644 --- a/android/guava/src/com/google/common/util/concurrent/Uninterruptibles.java +++ b/android/guava/src/com/google/common/util/concurrent/Uninterruptibles.java @@ -14,8 +14,10 @@ package com.google.common.util.concurrent; +import static com.google.common.base.Verify.verify; import static java.util.concurrent.TimeUnit.NANOSECONDS; +import com.google.common.annotations.Beta; import com.google.common.annotations.GwtCompatible; import com.google.common.annotations.GwtIncompatible; import com.google.common.base.Preconditions; @@ -24,6 +26,7 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -396,6 +399,50 @@ public static boolean tryLockUninterruptibly(Lock lock, long timeout, TimeUnit u } } + /** + * Invokes {@code executor.}{@link ExecutorService#awaitTermination(long, TimeUnit) + * awaitTermination(long, TimeUnit)} uninterruptibly with no timeout. + * + * @since NEXT + */ + @Beta + @GwtIncompatible // concurrency + public static void awaitTerminationUninterruptibly(ExecutorService executor) { + // TODO(cpovirk): We could optimize this to avoid calling nanoTime() at all. + verify(awaitTerminationUninterruptibly(executor, Long.MAX_VALUE, NANOSECONDS)); + } + + /** + * Invokes {@code executor.}{@link ExecutorService#awaitTermination(long, TimeUnit) + * awaitTermination(long, TimeUnit)} uninterruptibly. + * + * @since NEXT + */ + @Beta + @GwtIncompatible // concurrency + @SuppressWarnings("GoodTime") + public static boolean awaitTerminationUninterruptibly( + ExecutorService executor, long timeout, TimeUnit unit) { + boolean interrupted = false; + try { + long remainingNanos = unit.toNanos(timeout); + long end = System.nanoTime() + remainingNanos; + + while (true) { + try { + return executor.awaitTermination(remainingNanos, NANOSECONDS); + } catch (InterruptedException e) { + interrupted = true; + remainingNanos = end - System.nanoTime(); + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + // TODO(user): Add support for waitUninterruptibly. private Uninterruptibles() {} diff --git a/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java b/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java index 976e2ee9f4f4..4343d894dac6 100644 --- a/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java +++ b/guava-tests/test/com/google/common/util/concurrent/UninterruptiblesTest.java @@ -17,12 +17,14 @@ package com.google.common.util.concurrent; import static com.google.common.util.concurrent.InterruptionUtil.repeatedlyInterruptTestThread; +import static com.google.common.util.concurrent.Uninterruptibles.awaitTerminationUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.joinUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.putUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.takeUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.tryAcquireUninterruptibly; import static com.google.common.util.concurrent.Uninterruptibles.tryLockUninterruptibly; +import static java.util.concurrent.Executors.newFixedThreadPool; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; @@ -32,10 +34,12 @@ import com.google.common.testing.TearDown; import com.google.common.testing.TearDownStack; import com.google.errorprone.annotations.CanIgnoreReturnValue; +import java.time.Duration; import java.util.Date; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; @@ -465,6 +469,57 @@ public void testTryAcquireTimeoutMultiInterruptExpiredMultiPermit() { assertInterrupted(); } + // executor.awaitTermination Testcases + public void testTryAwaitTerminationUninterruptiblyDuration_success() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(1000)); + executor.shutdown(); + assertTrue(awaitTerminationUninterruptibly(executor, Duration.ofMillis(LONG_DELAY_MS))); + assertTrue(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationUninterruptiblyDuration_failure() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(10000)); + executor.shutdown(); + assertFalse(awaitTerminationUninterruptibly(executor, Duration.ofMillis(1000))); + assertFalse(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationUninterruptiblyLongTimeUnit_success() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(1000)); + executor.shutdown(); + assertTrue(awaitTerminationUninterruptibly(executor, LONG_DELAY_MS, MILLISECONDS)); + assertTrue(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationUninterruptiblyLongTimeUnit_failure() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(10000)); + executor.shutdown(); + assertFalse(awaitTerminationUninterruptibly(executor, 1000, MILLISECONDS)); + assertFalse(executor.isTerminated()); + assertInterrupted(); + } + + public void testTryAwaitTerminationInfiniteTimeout() { + ExecutorService executor = newFixedThreadPool(1); + requestInterruptIn(500); + executor.execute(new SleepTask(1000)); + executor.shutdown(); + awaitTerminationUninterruptibly(executor); + assertTrue(executor.isTerminated()); + assertInterrupted(); + } + /** * Wrapper around {@link Stopwatch} which also contains an "expected completion time." Creating a * {@code Completion} starts the underlying stopwatch. @@ -754,6 +809,15 @@ protected void doAction() { } } + private static final class SleepTask extends DelayedActionRunnable { + SleepTask(long tMinus) { + super(tMinus); + } + + @Override + protected void doAction() {} + } + private static void sleepSuccessfully(long sleepMillis) { Completion completed = new Completion(sleepMillis - SLEEP_SLACK); Uninterruptibles.sleepUninterruptibly(sleepMillis, MILLISECONDS); diff --git a/guava/src/com/google/common/util/concurrent/Uninterruptibles.java b/guava/src/com/google/common/util/concurrent/Uninterruptibles.java index 784efa82da2c..a38b351c079f 100644 --- a/guava/src/com/google/common/util/concurrent/Uninterruptibles.java +++ b/guava/src/com/google/common/util/concurrent/Uninterruptibles.java @@ -14,6 +14,7 @@ package com.google.common.util.concurrent; +import static com.google.common.base.Verify.verify; import static com.google.common.util.concurrent.Internal.toNanosSaturated; import static java.util.concurrent.TimeUnit.NANOSECONDS; @@ -27,6 +28,7 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -513,6 +515,63 @@ public static boolean tryLockUninterruptibly(Lock lock, long timeout, TimeUnit u } } + /** + * Invokes {@code executor.}{@link ExecutorService#awaitTermination(long, TimeUnit) + * awaitTermination(long, TimeUnit)} uninterruptibly with no timeout. + * + * @since NEXT + */ + @Beta + @GwtIncompatible // concurrency + public static void awaitTerminationUninterruptibly(ExecutorService executor) { + // TODO(cpovirk): We could optimize this to avoid calling nanoTime() at all. + verify(awaitTerminationUninterruptibly(executor, Long.MAX_VALUE, NANOSECONDS)); + } + + /** + * Invokes {@code executor.}{@link ExecutorService#awaitTermination(long, TimeUnit) + * awaitTermination(long, TimeUnit)} uninterruptibly. + * + * @since NEXT + */ + @Beta + @GwtIncompatible // concurrency + public static boolean awaitTerminationUninterruptibly( + ExecutorService executor, Duration timeout) { + return awaitTerminationUninterruptibly(executor, toNanosSaturated(timeout), NANOSECONDS); + } + + /** + * Invokes {@code executor.}{@link ExecutorService#awaitTermination(long, TimeUnit) + * awaitTermination(long, TimeUnit)} uninterruptibly. + * + * @since NEXT + */ + @Beta + @GwtIncompatible // concurrency + @SuppressWarnings("GoodTime") + public static boolean awaitTerminationUninterruptibly( + ExecutorService executor, long timeout, TimeUnit unit) { + boolean interrupted = false; + try { + long remainingNanos = unit.toNanos(timeout); + long end = System.nanoTime() + remainingNanos; + + while (true) { + try { + return executor.awaitTermination(remainingNanos, NANOSECONDS); + } catch (InterruptedException e) { + interrupted = true; + remainingNanos = end - System.nanoTime(); + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + // TODO(user): Add support for waitUninterruptibly. private Uninterruptibles() {}