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 b15647b commit ef1a506
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 77 deletions.
6 changes: 3 additions & 3 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 @@ -114,7 +114,7 @@ <E extends Exception> void blockOperations(
* 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
* operations are run once the {@link Releasable} is released or if a failure occurs while acquiring all permits; in this case the
* {@code onFailure} handler will be invoked before running delayed operations.
* {@code onFailure} handler will be invoked after delayed operations are released.
*
* @param onAcquired {@link ActionListener} that is invoked once acquisition is successful or failed
* @param timeout the maximum time to wait for the in-flight operations block
Expand All @@ -129,9 +129,9 @@ public void asyncBlockOperations(final ActionListener<Releasable> onAcquired, fi
@Override
public void onFailure(final Exception e) {
try {
onAcquired.onFailure(e);
releaseDelayedOperationsIfNeeded(); // resume delayed operations as soon as possible
} finally {
releaseDelayedOperationsIfNeeded();
onAcquired.onFailure(e);
}
}

Expand Down
113 changes: 39 additions & 74 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,6 +20,7 @@

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 @@ -200,18 +201,8 @@ public void testBlockIfClosed() {
expectThrows(IndexShardClosedException.class, () -> permits.blockOperations(randomInt(10), TimeUnit.MINUTES,
() -> { throw new IllegalArgumentException("fake error"); }));
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));
() -> permits.asyncBlockOperations(wrap(() -> { throw new IllegalArgumentException("fake error");}),
randomInt(10), TimeUnit.MINUTES));
}

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

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);
permits.asyncBlockOperations(wrap(() -> {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
}), 30, TimeUnit.MINUTES);
assertFalse(blocked.get());
assertFalse(future.isDone());
}
Expand Down Expand Up @@ -351,23 +330,11 @@ public void testAsyncBlockOperationsOperationWhileBlocked() throws InterruptedEx
final CountDownLatch blockAcquired = new CountDownLatch(1);
final CountDownLatch releaseBlock = new CountDownLatch(1);
final AtomicBoolean blocked = new AtomicBoolean();
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
} catch (final InterruptedException e) {
throw new RuntimeException(e);
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
permits.asyncBlockOperations(wrap(() -> {
blocked.set(true);
blockAcquired.countDown();
releaseBlock.await();
}), 30, TimeUnit.MINUTES);
blockAcquired.await();
assertTrue(blocked.get());

Expand Down Expand Up @@ -415,20 +382,10 @@ 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();
permits.asyncBlockOperations(new ActionListener<Releasable>() {
@Override
public void onResponse(Releasable releasable) {
try (Releasable ignored = releasable) {
onBlocked.set(true);
blockedLatch.countDown();
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
permits.asyncBlockOperations(wrap(() -> {
onBlocked.set(true);
blockedLatch.countDown();
}), 30, TimeUnit.MINUTES);
assertFalse(onBlocked.get());

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

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
}, 30, TimeUnit.MINUTES);
permits.asyncBlockOperations(wrap(() -> {
values.add(operations);
operationLatch.countDown();
}), 30, TimeUnit.MINUTES);
});
blockingThread.start();

Expand Down Expand Up @@ -756,4 +703,22 @@ public void testPermitTraceCapturing() throws ExecutionException, InterruptedExc
assertThat(permits.getActiveOperationsCount(), equalTo(0));
assertThat(permits.getActiveOperations(), emptyIterable());
}

private static ActionListener<Releasable> wrap(final CheckedRunnable<Exception> onResponse) {
return new ActionListener<Releasable>() {
@Override
public void onResponse(final Releasable releasable) {
try (Releasable ignored = releasable) {
onResponse.run();
} catch (final Exception e) {
onFailure(e);
}
}

@Override
public void onFailure(final Exception e) {
throw new RuntimeException(e);
}
};
}
}

0 comments on commit ef1a506

Please sign in to comment.