honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Correctly release threads from starting gate in o.e.c.c.CacheTests

This commit is contained in:
Jason Tedor 2015-12-28 08:07:37 -05:00
parent 0a816cd343
commit 35cc749c9a
1 changed files with 114 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

194
core/src/test/java/org/elasticsearch/common/cache/CacheTests.java vendored
View File

@ -494,33 +494,41 @@ public class CacheTests extends ESTestCase {
public void testComputeIfAbsentCallsOnce() throws InterruptedException { public void testComputeIfAbsentCallsOnce() throws InterruptedException {
int numberOfThreads = randomIntBetween(2, 32); int numberOfThreads = randomIntBetween(2, 32);
final Cache<Integer, String> cache = CacheBuilder.<Integer, String>builder().build(); final Cache<Integer, String> cache = CacheBuilder.<Integer, String>builder().build();
List<Thread> threads = new ArrayList<>();
AtomicReferenceArray flags = new AtomicReferenceArray(numberOfEntries); AtomicReferenceArray flags = new AtomicReferenceArray(numberOfEntries);
for (int j = 0; j < numberOfEntries; j++) { for (int j = 0; j < numberOfEntries; j++) {
flags.set(j, false); flags.set(j, false);
} }
CountDownLatch latch = new CountDownLatch(1 + numberOfThreads); CountDownLatch startGate = new CountDownLatch(1);
CountDownLatch endGate = new CountDownLatch(numberOfThreads);
AtomicBoolean interrupted = new AtomicBoolean();
for (int i = 0; i < numberOfThreads; i++) { for (int i = 0; i < numberOfThreads; i++) {
Thread thread = new Thread(() -> { Thread thread = new Thread(() -> {
latch.countDown(); try {
for (int j = 0; j < numberOfEntries; j++) {
try { try {
cache.computeIfAbsent(j, key -> { startGate.await();
assertTrue(flags.compareAndSet(key, false, true)); } catch (InterruptedException e) {
return Integer.toString(key); interrupted.set(true);
}); return;
} catch (ExecutionException e) {
throw new RuntimeException(e);
} }
for (int j = 0; j < numberOfEntries; j++) {
try {
cache.computeIfAbsent(j, key -> {
assertTrue(flags.compareAndSet(key, false, true));
return Integer.toString(key);
});
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
} finally {
endGate.countDown();
} }
}); });
threads.add(thread);
thread.start(); thread.start();
} }
latch.countDown(); startGate.countDown();
for (Thread thread : threads) { endGate.await();
thread.join(); assertFalse(interrupted.get());
}
} }
public void testComputeIfAbsentThrowsExceptionIfLoaderReturnsANullValue() { public void testComputeIfAbsentThrowsExceptionIfLoaderReturnsANullValue() {
@ -560,30 +568,39 @@ public class CacheTests extends ESTestCase {
int numberOfThreads = randomIntBetween(2, 32); int numberOfThreads = randomIntBetween(2, 32);
final Cache<Key, Integer> cache = CacheBuilder.<Key, Integer>builder().build(); final Cache<Key, Integer> cache = CacheBuilder.<Key, Integer>builder().build();
CountDownLatch latch = new CountDownLatch(1 + numberOfThreads); CountDownLatch startGate = new CountDownLatch(1);
CountDownLatch deadlockLatch = new CountDownLatch(numberOfThreads); CountDownLatch deadlockLatch = new CountDownLatch(numberOfThreads);
AtomicBoolean interrupted = new AtomicBoolean();
List<Thread> threads = new ArrayList<>(); List<Thread> threads = new ArrayList<>();
for (int i = 0; i < numberOfThreads; i++) { for (int i = 0; i < numberOfThreads; i++) {
Thread thread = new Thread(() -> { Thread thread = new Thread(() -> {
Random random = new Random(random().nextLong()); try {
latch.countDown();
for (int j = 0; j < numberOfEntries; j++) {
Key key = new Key(random.nextInt(numberOfEntries));
try { try {
cache.computeIfAbsent(key, k -> { startGate.await();
if (k.key == 0) { } catch (InterruptedException e) {
return 0; interrupted.set(true);
} else { return;
Integer value = cache.get(new Key(k.key / 2));
return value != null ? value : 0;
}
});
} catch (ExecutionException e) {
fail(e.getMessage());
} }
Random random = new Random(random().nextLong());
for (int j = 0; j < numberOfEntries; j++) {
Key key = new Key(random.nextInt(numberOfEntries));
try {
cache.computeIfAbsent(key, k -> {
if (k.key == 0) {
return 0;
} else {
Integer value = cache.get(new Key(k.key / 2));
return value != null ? value : 0;
}
});
} catch (ExecutionException e) {
fail(e.getMessage());
}
}
} finally {
// successfully avoided deadlock, release the main thread
deadlockLatch.countDown();
} }
// successfully avoided deadlock, release the main thread
deadlockLatch.countDown();
}); });
threads.add(thread); threads.add(thread);
thread.start(); thread.start();
@ -614,7 +631,7 @@ public class CacheTests extends ESTestCase {
}, 1, 1, TimeUnit.SECONDS); }, 1, 1, TimeUnit.SECONDS);
// everything is setup, release the hounds // everything is setup, release the hounds
latch.countDown(); startGate.countDown();
// wait for either deadlock to be detected or the threads to terminate // wait for either deadlock to be detected or the threads to terminate
deadlockLatch.await(); deadlockLatch.await();
@ -628,49 +645,57 @@ public class CacheTests extends ESTestCase {
public void testCachePollution() throws InterruptedException { public void testCachePollution() throws InterruptedException {
int numberOfThreads = randomIntBetween(2, 32); int numberOfThreads = randomIntBetween(2, 32);
final Cache<Integer, String> cache = CacheBuilder.<Integer, String>builder().build(); final Cache<Integer, String> cache = CacheBuilder.<Integer, String>builder().build();
CountDownLatch latch = new CountDownLatch(1 + numberOfThreads); CountDownLatch startGate = new CountDownLatch(1);
List<Thread> threads = new ArrayList<>(); CountDownLatch endGate = new CountDownLatch(numberOfThreads);
AtomicBoolean interrupted = new AtomicBoolean();
for (int i = 0; i < numberOfThreads; i++) { for (int i = 0; i < numberOfThreads; i++) {
Thread thread = new Thread(() -> { Thread thread = new Thread(() -> {
latch.countDown(); try {
Random random = new Random(random().nextLong()); try {
for (int j = 0; j < numberOfEntries; j++) { startGate.await();
Integer key = random.nextInt(numberOfEntries); } catch (InterruptedException e) {
boolean first; interrupted.set(true);
boolean second; return;
do {
first = random.nextBoolean();
second = random.nextBoolean();
} while (first && second);
if (first) {
try {
cache.computeIfAbsent(key, k -> {
if (random.nextBoolean()) {
return Integer.toString(k);
} else {
throw new Exception("testCachePollution");
}
});
} catch (ExecutionException e) {
assertNotNull(e.getCause());
assertThat(e.getCause(), instanceOf(Exception.class));
assertEquals(e.getCause().getMessage(), "testCachePollution");
}
} else if (second) {
cache.invalidate(key);
} else {
cache.get(key);
} }
Random random = new Random(random().nextLong());
for (int j = 0; j < numberOfEntries; j++) {
Integer key = random.nextInt(numberOfEntries);
boolean first;
boolean second;
do {
first = random.nextBoolean();
second = random.nextBoolean();
} while (first && second);
if (first) {
try {
cache.computeIfAbsent(key, k -> {
if (random.nextBoolean()) {
return Integer.toString(k);
} else {
throw new Exception("testCachePollution");
}
});
} catch (ExecutionException e) {
assertNotNull(e.getCause());
assertThat(e.getCause(), instanceOf(Exception.class));
assertEquals(e.getCause().getMessage(), "testCachePollution");
}
} else if (second) {
cache.invalidate(key);
} else {
cache.get(key);
}
}
} finally {
endGate.countDown();
} }
}); });
threads.add(thread);
thread.start(); thread.start();
} }
latch.countDown(); startGate.countDown();
for (Thread thread : threads) { endGate.await();
thread.join(); assertFalse(interrupted.get());
}
} }
// test that the cache is not corrupted under lots of concurrent modifications, even hitting the same key // test that the cache is not corrupted under lots of concurrent modifications, even hitting the same key
@ -683,24 +708,33 @@ public class CacheTests extends ESTestCase {
.weigher((k, v) -> 2) .weigher((k, v) -> 2)
.build(); .build();
CountDownLatch latch = new CountDownLatch(1 + numberOfThreads); CountDownLatch startGate = new CountDownLatch(1);
List<Thread> threads = new ArrayList<>(); CountDownLatch endGate = new CountDownLatch(numberOfThreads);
AtomicBoolean interrupted = new AtomicBoolean();
for (int i = 0; i < numberOfThreads; i++) { for (int i = 0; i < numberOfThreads; i++) {
Thread thread = new Thread(() -> { Thread thread = new Thread(() -> {
Random random = new Random(random().nextLong()); try {
latch.countDown(); try {
for (int j = 0; j < numberOfEntries; j++) { startGate.await();
Integer key = random.nextInt(numberOfEntries); } catch (InterruptedException e) {
cache.put(key, Integer.toString(j)); interrupted.set(true);
return;
}
Random random = new Random(random().nextLong());
for (int j = 0; j < numberOfEntries; j++) {
Integer key = random.nextInt(numberOfEntries);
cache.put(key, Integer.toString(j));
}
} finally {
endGate.countDown();
} }
}); });
threads.add(thread);
thread.start(); thread.start();
} }
latch.countDown(); startGate.countDown();
for (Thread thread : threads) { endGate.await();
thread.join(); assertFalse(interrupted.get());
}
cache.refresh(); cache.refresh();
assertEquals(500, cache.count()); assertEquals(500, cache.count());
} }