Skip to content

Commit

Permalink
Apply feedback
Browse files Browse the repository at this point in the history
  • Loading branch information
@tlrx
tlrx committed Nov 7, 2018
1 parent 4d792af commit b15647b
Show file tree
Hide file tree
Showing 3 changed files with 112 additions and 125 deletions.
33 changes: 21 additions & 12 deletions server/src/main/java/org/elasticsearch/index/shard/IndexShard.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2286,23 +2286,32 @@ private <E extends Exception> void bumpPrimaryTerm(long newPrimaryTerm, final Ch
assert newPrimaryTerm > pendingPrimaryTerm;
assert operationPrimaryTerm <= pendingPrimaryTerm;
final CountDownLatch termUpdated = new CountDownLatch(1);
indexShardOperationPermits.asyncBlockOperations(30, TimeUnit.MINUTES, () -> {
assert operationPrimaryTerm <= pendingPrimaryTerm;
termUpdated.await();
// indexShardOperationPermits doesn't guarantee that async submissions are executed
// in the order submitted. We need to guard against another term bump
if (operationPrimaryTerm < newPrimaryTerm) {
operationPrimaryTerm = newPrimaryTerm;
onBlocked.run();
}
},
e -> {
indexShardOperationPermits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onFailure(final Exception e) {
try {
failShard("exception during primary term transition", e);
} catch (AlreadyClosedException ace) {
// ignore, shard is already closed
}
});
}

@Override
public void onResponse(final Releasable releasable) {
try (Releasable ignored = releasable) {
assert operationPrimaryTerm <= pendingPrimaryTerm;
termUpdated.await();
// indexShardOperationPermits doesn't guarantee that async submissions are executed
// in the order submitted. We need to guard against another term bump
if (operationPrimaryTerm < newPrimaryTerm) {
operationPrimaryTerm = newPrimaryTerm;
onBlocked.run();
}
} catch (final Exception e) {
onFailure(e);
}
}
}, 30, TimeUnit.MINUTES);
pendingPrimaryTerm = newPrimaryTerm;
termUpdated.countDown();
}
Expand Down
32 changes: 0 additions & 32 deletions server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -111,37 +110,6 @@ <E extends Exception> void blockOperations(
}
}

/**
* Immediately delays operations and on another thread waits for in-flight operations to finish and then executes {@code onBlocked}
* under the guarantee that no new operations are started. Delayed operations are run after {@code onBlocked} has executed. After
* operations are delayed and the blocking is forked to another thread, returns to the caller. If a failure occurs while blocking
* operations or executing {@code onBlocked} then the {@code onFailure} handler will be invoked.
*
* @param timeout the maximum time to wait for the in-flight operations block
* @param timeUnit the time unit of the {@code timeout} argument
* @param onBlocked the action to run once the block has been acquired
* @param onFailure the action to run if a failure occurs while blocking operations
* @param <E> the type of checked exception thrown by {@code onBlocked} (not thrown on the calling thread)
*/
<E extends Exception> void asyncBlockOperations(final long timeout, final TimeUnit timeUnit,
final CheckedRunnable<E> onBlocked, final Consumer<Exception> onFailure) {
asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onFailure(final Exception e) {
onFailure.accept(e);
}

@Override
public void onResponse(final Releasable releasable) {
try (Releasable ignored = releasable) {
onBlocked.run();
} catch (final Exception e) {
onFailure.accept(e);
}
}
}, timeout, timeUnit);
}

/**
* Immediately delays operations and on another thread waits for in-flight operations to finish and then acquires all permits. When all
* permits are acquired, the provided {@link ActionListener} is called under the guarantee that no new operations are started. Delayed
Expand Down
172 changes: 91 additions & 81 deletions server/src/test/java/org/elasticsearch/index/shard/IndexShardOperationPermitsTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.common.CheckedRunnable;
import org.elasticsearch.common.lease.Releasable;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
Expand Down Expand Up @@ -49,7 +48,6 @@
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -201,8 +199,19 @@ public void testBlockIfClosed() {
permits.close();
expectThrows(IndexShardClosedException.class, () -> permits.blockOperations(randomInt(10), TimeUnit.MINUTES,
() -> { throw new IllegalArgumentException("fake error"); }));
expectThrows(IndexShardClosedException.class, () -> permits.asyncBlockOperations(randomInt(10), TimeUnit.MINUTES,
() -> { throw new IllegalArgumentException("fake error"); }, e -> { throw new AssertionError(e); }));
expectThrows(IndexShardClosedException.class,
() -> permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(final Releasable releasable) {
releasable.close();
throw new IllegalArgumentException("fake error");
}

@Override
public void onFailure(final Exception e) {
throw new AssertionError(e);
}
}, randomInt(10), TimeUnit.MINUTES));
}

public void testOperationsDelayedIfBlock() throws ExecutionException, InterruptedException, TimeoutException {
Expand All @@ -222,17 +231,23 @@ public void testGetBlockWhenBlocked() throws ExecutionException, InterruptedExce
try (Releasable ignored = blockAndWait()) {
permits.acquire(future, ThreadPool.Names.GENERIC, true, "");

randomAsyncBlockOperations(permits,
30,
TimeUnit.MINUTES,
() -> {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
},
e -> {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(final Releasable releasable) {
try (Releasable ignored = releasable) {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
} catch (final Exception e) {
throw new RuntimeException(e);
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
});
}
}, 30, TimeUnit.MINUTES);
assertFalse(blocked.get());
assertFalse(future.isDone());
}
Expand Down Expand Up @@ -294,7 +309,7 @@ public void onResponse(Releasable releasable) {
future2.get(1, TimeUnit.HOURS).close();
}

protected Releasable blockAndWait() throws InterruptedException {
private Releasable blockAndWait() throws InterruptedException {
CountDownLatch blockAcquired = new CountDownLatch(1);
CountDownLatch releaseBlock = new CountDownLatch(1);
CountDownLatch blockReleased = new CountDownLatch(1);
Expand Down Expand Up @@ -336,17 +351,23 @@ public void testAsyncBlockOperationsOperationWhileBlocked() throws InterruptedEx
final CountDownLatch blockAcquired = new CountDownLatch(1);
final CountDownLatch releaseBlock = new CountDownLatch(1);
final AtomicBoolean blocked = new AtomicBoolean();
randomAsyncBlockOperations(permits,
30,
TimeUnit.MINUTES,
() -> {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
},
e -> {
} catch (final InterruptedException e) {
throw new RuntimeException(e);
});
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
blockAcquired.await();
assertTrue(blocked.get());

Expand Down Expand Up @@ -394,16 +415,20 @@ public void testAsyncBlockOperationsOperationBeforeBlocked() throws InterruptedE
// now we will delay operations while the first operation is still executing (because it is latched)
final CountDownLatch blockedLatch = new CountDownLatch(1);
final AtomicBoolean onBlocked = new AtomicBoolean();
randomAsyncBlockOperations(permits,
30,
TimeUnit.MINUTES,
() -> {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
onBlocked.set(true);
blockedLatch.countDown();
}, e -> {
throw new RuntimeException(e);
});
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
assertFalse(onBlocked.get());

// if we submit another operation, it should be delayed
Expand Down Expand Up @@ -488,15 +513,20 @@ public void onFailure(Exception e) {
} catch (final BrokenBarrierException | InterruptedException e) {
throw new RuntimeException(e);
}
randomAsyncBlockOperations(permits,
30,
TimeUnit.MINUTES,
() -> {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
values.add(operations);
operationLatch.countDown();
}, e -> {
throw new RuntimeException(e);
});
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
});
blockingThread.start();

Expand Down Expand Up @@ -561,16 +591,20 @@ public void testActiveOperationsCount() throws ExecutionException, InterruptedEx
public void testAsyncBlockOperationsOnFailure() throws InterruptedException {
final AtomicReference<Exception> reference = new AtomicReference<>();
final CountDownLatch onFailureLatch = new CountDownLatch(1);
randomAsyncBlockOperations(permits,
10,
TimeUnit.MINUTES,
() -> {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
throw new RuntimeException("simulated");
},
e -> {
reference.set(e);
onFailureLatch.countDown();
});
}
}

@Override
public void onFailure(final Exception e) {
reference.set(e);
onFailureLatch.countDown();
}
}, 10, TimeUnit.MINUTES);
onFailureLatch.await();
assertThat(reference.get(), instanceOf(RuntimeException.class));
assertThat(reference.get(), hasToString(containsString("simulated")));
Expand Down Expand Up @@ -598,14 +632,18 @@ public void testTimeout() throws BrokenBarrierException, InterruptedException {
{
final AtomicReference<Exception> reference = new AtomicReference<>();
final CountDownLatch onFailureLatch = new CountDownLatch(1);
randomAsyncBlockOperations(permits,
1,
TimeUnit.MILLISECONDS,
() -> {},
e -> {
reference.set(e);
onFailureLatch.countDown();
});
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
releasable.close();
}

@Override
public void onFailure(final Exception e) {
reference.set(e);
onFailureLatch.countDown();
}
}, 1, TimeUnit.MILLISECONDS);
onFailureLatch.await();
assertThat(reference.get(), hasToString(containsString("timeout while blocking operations")));
}
Expand Down Expand Up @@ -718,32 +756,4 @@ public void testPermitTraceCapturing() throws ExecutionException, InterruptedExc
assertThat(permits.getActiveOperationsCount(), equalTo(0));
assertThat(permits.getActiveOperations(), emptyIterable());
}

/**
* Randomizes the usage of {@link IndexShardOperationPermits#asyncBlockOperations(ActionListener, long, TimeUnit)} and
* {@link IndexShardOperationPermits#asyncBlockOperations(long, TimeUnit, CheckedRunnable, Consumer)}
*/
private <E extends Exception> void randomAsyncBlockOperations(final IndexShardOperationPermits permits,
final long timeout, final TimeUnit timeUnit,
final CheckedRunnable<E> onBlocked, final Consumer<Exception> onFailure) {
if (randomBoolean()) {
permits.asyncBlockOperations(timeout, timeUnit, onBlocked, onFailure);
} else {
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(final Releasable releasable) {
try (Releasable ignored = releasable) {
onBlocked.run();
} catch (final Exception e) {
onFailure.accept(e);
}
}

@Override
public void onFailure(final Exception e) {
onFailure.accept(e);
}
}, timeout, timeUnit);
}
}
}

0 comments on commit b15647b

Please sign in to comment.