Harden OrderedExecutorTest to use concurrent latches/barriers for testing parallelism instead of making assumpions about how milliseconds something should take in another thread

This commit is contained in:
Chris Hostetter 2019-02-06 14:32:12 -07:00
parent 0add3bb0ac
commit ea2956fda3
1 changed files with 137 additions and 30 deletions

View File

@ -17,15 +17,29 @@
package org.apache.solr.util;
import java.lang.invoke.MethodHandles;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.solr.common.util.ExecutorUtil;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class OrderedExecutorTest extends LuceneTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@Test
public void testExecutionInOrder() {
@ -40,44 +54,137 @@ public class OrderedExecutorTest extends LuceneTestCase {
@Test
public void testLockWhenQueueIsFull() {
OrderedExecutor orderedExecutor = new OrderedExecutor(10, ExecutorUtil.newMDCAwareCachedThreadPool("testLockWhenQueueIsFull"));
IntBox intBox = new IntBox();
long t = System.nanoTime();
orderedExecutor.execute(1, () -> {
final ExecutorService controlExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("testLockWhenQueueIsFull_control");
final OrderedExecutor orderedExecutor = new OrderedExecutor
(10, ExecutorUtil.newMDCAwareCachedThreadPool("testLockWhenQueueIsFull_test"));
try {
Thread.sleep(500L);
// AAA and BBB events will both depend on the use of the same lockId
final BlockingQueue<String> events = new ArrayBlockingQueue<>(2);
final Integer lockId = 1;
// AAA enters executor first so it should execute first (even though it's waiting on latch)
final CountDownLatch latchAAA = new CountDownLatch(1);
orderedExecutor.execute(lockId, () -> {
try {
if (latchAAA.await(120, TimeUnit.SECONDS)) {
events.add("AAA");
} else {
events.add("AAA Timed Out");
}
} catch (InterruptedException e) {
log.error("Interrupt in AAA worker", e);
Thread.currentThread().interrupt();
}
intBox.value++;
});
assertTrue(System.nanoTime() - t < 100 * 1000000);
// BBB doesn't care about the latch, but because it uses the same lockId, it's blocked on AAA
// so we execute it in a background thread...
controlExecutor.execute(() -> {
orderedExecutor.execute(lockId, () -> {
events.add("BBB");
});
});
t = System.nanoTime();
orderedExecutor.execute(1, () -> {
intBox.value++;
});
assertTrue(System.nanoTime() - t > 300 * 1000000);
// now if we release the latchAAA, AAA should be garunteed to fire first, then BBB
latchAAA.countDown();
try {
assertEquals("AAA", events.poll(120, TimeUnit.SECONDS));
assertEquals("BBB", events.poll(120, TimeUnit.SECONDS));
} catch (InterruptedException e) {
log.error("Interrupt polling event queue", e);
Thread.currentThread().interrupt();
fail("interupt while trying to poll event queue");
}
} finally {
ExecutorUtil.shutdownAndAwaitTermination(controlExecutor);
orderedExecutor.shutdownAndAwaitTermination();
assertEquals(intBox.value, 2);
}
}
@Test
public void testRunInParallel() {
OrderedExecutor orderedExecutor = new OrderedExecutor(10, ExecutorUtil.newMDCAwareCachedThreadPool("testLockWhenQueueIsFull"));
AtomicInteger atomicInteger = new AtomicInteger(0);
orderedExecutor.execute(1, () -> {
final int parallelism = atLeast(3);
final ExecutorService controlExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("testRunInParallel_control");
final OrderedExecutor orderedExecutor = new OrderedExecutor
(parallelism, ExecutorUtil.newMDCAwareCachedThreadPool("testRunInParallel_test"));
try {
Thread.sleep(500L);
// distinct lockIds should be able to be used in parallel, up to the size of the executor,
// w/o any execute calls blocking... until the test Runables being executed are all
// waiting on the same cyclic barrier...
final CyclicBarrier barrier = new CyclicBarrier(parallelism + 1);
final CountDownLatch preBarrierLatch = new CountDownLatch(parallelism);
final CountDownLatch postBarrierLatch = new CountDownLatch(parallelism);
for (int i = 0; i < parallelism; i++) {
final int lockId = i;
controlExecutor.execute(() -> {
orderedExecutor.execute(lockId, () -> {
try {
log.info("Worker #{} starting", lockId);
preBarrierLatch.countDown();
barrier.await(120, TimeUnit.SECONDS);
postBarrierLatch.countDown();
} catch (TimeoutException t) {
log.error("Timeout in worker#" + lockId + "awaiting barrier", t);
} catch (BrokenBarrierException b) {
log.error("Broken Barrier in worker#" + lockId, b);
} catch (InterruptedException e) {
log.error("Interrupt in worker#" + lockId + "awaiting barrier", e);
Thread.currentThread().interrupt();
}
if (atomicInteger.get() == 1) atomicInteger.incrementAndGet();
});
});
}
orderedExecutor.execute(2, atomicInteger::incrementAndGet);
log.info("main thread: about to wait on pre-barrier latch, barrier={}, post-barrier latch={}",
barrier.getNumberWaiting(), postBarrierLatch.getCount());
try {
// this latch should have fully counted down by now
// (or with a small await for thread scheduling but no other external action)
assertTrue("Timeout awaiting pre barrier latch",
preBarrierLatch.await(120, TimeUnit.SECONDS));
} catch (InterruptedException e) {
log.error("Interrupt awwaiting pre barrier latch", e);
Thread.currentThread().interrupt();
fail("interupt while trying to await the preBarrierLatch");
}
log.info("main thread: pre-barrier latch done, barrier={}, post-barrier latch={}",
barrier.getNumberWaiting(), postBarrierLatch.getCount());
// nothing should have counted down yet on the postBarrierLatch
assertEquals(parallelism, postBarrierLatch.getCount());
try {
// if we now await on the the barrier, it should release
// (once all other threads get to the barrier as well, but no external action needed)
barrier.await(120, TimeUnit.SECONDS);
log.info("main thread: barrier has released, post-barrier latch={}",
postBarrierLatch.getCount());
// and now the post-barrier latch should release immediately
// (or with a small await for thread scheduling but no other external action)
assertTrue("Timeout awaiting post barrier latch",
postBarrierLatch.await(120, TimeUnit.SECONDS));
} catch (TimeoutException t) {
log.error("Timeout awaiting barrier", t);
fail("barrier timed out");
} catch (BrokenBarrierException b) {
log.error("Broken Barrier in main test thread", b);
fail("broken barrier while trying to release the barrier");
} catch (InterruptedException e) {
log.error("Interrupt awwaiting barrier / post barrier latch", e);
Thread.currentThread().interrupt();
fail("interupt while trying to release the barrier and await the postBarrierLatch");
}
} finally {
ExecutorUtil.shutdownAndAwaitTermination(controlExecutor);
orderedExecutor.shutdownAndAwaitTermination();
assertEquals(atomicInteger.get(), 2);
}
}
@Test