Skip to content

Commit

Permalink
ConditionVariable: Add uninterruptible block
Browse files Browse the repository at this point in the history
Sometimes it's useful to be able to block until something on some other thread
"really has finished". This will be needed for moving caching operations onto
the executor in Downloader implementations, since we need to guarantee that
Downloader.download doesn't return until it's no longer modifying the
underlying cache.

One solution to this is of course just to not interrupt the thread that's
blocking on the condition variable, but there are cases where you do need to do
this in case the thread is at some other point in its execution. This is true
for Downloader implementations, where the Download.download thread will also
be blocking on PriorityTaskManager.proceed. Arranging to conditionally
interrupt the thread based on where it's got to is probably possible, but seems
complicated and error prone.

Issue: #5978
PiperOrigin-RevId: 313152413
  • Loading branch information
ojw28 committed May 27, 2020
1 parent ee11d9d commit 03ea39b
Show file tree
Hide file tree
Showing 2 changed files with 56 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,26 @@ public synchronized boolean block(long timeoutMs) throws InterruptedException {
return isOpen;
}

/**
* Blocks until the condition is open. Unlike {@link #block}, this method will continue to block
* if the calling thread is interrupted. If the calling thread was interrupted then its {@link
* Thread#isInterrupted() interrupted status} will be set when the method returns.
*/
public synchronized void blockUninterruptible() {
boolean wasInterrupted = false;
while (!isOpen) {
try {
wait();
} catch (InterruptedException e) {
wasInterrupted = true;
}
}
if (wasInterrupted) {
// Restore the interrupted status.
Thread.currentThread().interrupt();
}
}

/** Returns whether the condition is opened. */
public synchronized boolean isOpen() {
return isOpen;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public void blockWithTimeout_blocksForAtLeastTimeout() throws InterruptedExcepti
}

@Test
public void blockWithoutTimeout_blocks() throws InterruptedException {
public void blockWithMaxTimeout_blocks_thenThrowsWhenInterrupted() throws InterruptedException {
ConditionVariable conditionVariable = buildTestConditionVariable();

AtomicBoolean blockReturned = new AtomicBoolean();
Expand All @@ -58,7 +58,7 @@ public void blockWithoutTimeout_blocks() throws InterruptedException {
new Thread(
() -> {
try {
conditionVariable.block();
conditionVariable.block(/* timeoutMs= */ Long.MAX_VALUE);
blockReturned.set(true);
} catch (InterruptedException e) {
blockWasInterrupted.set(true);
Expand All @@ -76,7 +76,7 @@ public void blockWithoutTimeout_blocks() throws InterruptedException {
}

@Test
public void blockWithMaxTimeout_blocks() throws InterruptedException {
public void block_blocks_thenThrowsWhenInterrupted() throws InterruptedException {
ConditionVariable conditionVariable = buildTestConditionVariable();

AtomicBoolean blockReturned = new AtomicBoolean();
Expand All @@ -85,7 +85,7 @@ public void blockWithMaxTimeout_blocks() throws InterruptedException {
new Thread(
() -> {
try {
conditionVariable.block(/* timeoutMs= */ Long.MAX_VALUE);
conditionVariable.block();
blockReturned.set(true);
} catch (InterruptedException e) {
blockWasInterrupted.set(true);
Expand All @@ -103,7 +103,7 @@ public void blockWithMaxTimeout_blocks() throws InterruptedException {
}

@Test
public void open_unblocksBlock() throws InterruptedException {
public void block_blocks_thenReturnsWhenOpened() throws InterruptedException {
ConditionVariable conditionVariable = buildTestConditionVariable();

AtomicBoolean blockReturned = new AtomicBoolean();
Expand All @@ -129,6 +129,37 @@ public void open_unblocksBlock() throws InterruptedException {
assertThat(conditionVariable.isOpen()).isTrue();
}

@Test
public void blockUnterruptible_blocksIfInterrupted_thenUnblocksWhenOpened()
throws InterruptedException {
ConditionVariable conditionVariable = buildTestConditionVariable();

AtomicBoolean blockReturned = new AtomicBoolean();
AtomicBoolean interruptedStatusSet = new AtomicBoolean();
Thread blockingThread =
new Thread(
() -> {
conditionVariable.blockUninterruptible();
blockReturned.set(true);
interruptedStatusSet.set(Thread.currentThread().isInterrupted());
});

blockingThread.start();
Thread.sleep(500);
assertThat(blockReturned.get()).isFalse();

blockingThread.interrupt();
Thread.sleep(500);
// blockUninterruptible should still be blocked.
assertThat(blockReturned.get()).isFalse();

conditionVariable.open();
blockingThread.join();
// blockUninterruptible should have set the thread's interrupted status on exit.
assertThat(interruptedStatusSet.get()).isTrue();
assertThat(conditionVariable.isOpen()).isTrue();
}

private static ConditionVariable buildTestConditionVariable() {
return new ConditionVariable(
new SystemClock() {
Expand Down

0 comments on commit 03ea39b

Please sign in to comment.