提交 dae3a856 编写于 作者: D dl

6868712: Improve concurrent queue tests

Summary: Fix all known flaky tests, plus minor maintenance
Reviewed-by: martin, chegar
上级 dd70c1b8
......@@ -178,6 +178,11 @@ public class BiggernYours {
new ConcurrentLinkedQueue() {
public int size() {return randomize(super.size());}});
// testCollections(
// new LinkedTransferQueue(),
// new LinkedTransferQueue() {
// public int size() {return randomize(super.size());}});
testCollections(
new LinkedBlockingQueue(),
new LinkedBlockingQueue() {
......
......@@ -49,6 +49,7 @@ public class IteratorAtEnd {
testCollection(new LinkedBlockingQueue());
testCollection(new ArrayBlockingQueue(100));
testCollection(new ConcurrentLinkedQueue());
// testCollection(new LinkedTransferQueue());
testMap(new HashMap());
testMap(new Hashtable());
......
......@@ -76,6 +76,7 @@ public class MOAT {
testCollection(new LinkedBlockingQueue<Integer>(20));
testCollection(new LinkedBlockingDeque<Integer>(20));
testCollection(new ConcurrentLinkedQueue<Integer>());
// testCollection(new LinkedTransferQueue<Integer>());
testCollection(new ConcurrentSkipListSet<Integer>());
testCollection(Arrays.asList(new Integer(42)));
testCollection(Arrays.asList(1,2,3));
......@@ -161,6 +162,7 @@ public class MOAT {
equal(c.toString(),"[]");
equal(c.toArray().length, 0);
equal(c.toArray(new Object[0]).length, 0);
check(c.toArray(new Object[]{42})[0] == null);
Object[] a = new Object[1]; a[0] = Boolean.TRUE;
equal(c.toArray(a), a);
......
......@@ -234,6 +234,7 @@ public class RacingCollections {
List<Queue<Integer>> list =
new ArrayList<Queue<Integer>>(newConcurrentDeques());
list.add(new LinkedBlockingQueue<Integer>(10));
// list.add(new LinkedTransferQueue<Integer>());
return list;
}
......
......@@ -69,6 +69,7 @@ public class RemoveContains {
test(new ArrayBlockingQueue<String>(10));
test(new LinkedBlockingQueue<String>(10));
test(new LinkedBlockingDeque<String>(10));
// test(new LinkedTransferQueue<String>());
test(new ArrayDeque<String>(10));
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
......
......@@ -75,10 +75,12 @@ public class CancelledProducerConsumerLoops {
}
static void oneRun(BlockingQueue<Integer> q, int npairs, int iters) throws Exception {
if (print)
System.out.printf("%-18s", q.getClass().getSimpleName());
LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
CyclicBarrier barrier = new CyclicBarrier(npairs * 2 + 1, timer);
Future[] prods = new Future[npairs];
Future[] cons = new Future[npairs];
Future<?>[] prods = new Future<?>[npairs];
Future<?>[] cons = new Future<?>[npairs];
for (int i = 0; i < npairs; ++i) {
prods[i] = pool.submit(new Producer(q, barrier, iters));
......@@ -119,21 +121,13 @@ public class CancelledProducerConsumerLoops {
static void oneTest(int pairs, int iters) throws Exception {
if (print)
System.out.print("ArrayBlockingQueue ");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
if (print)
System.out.print("LinkedBlockingQueue ");
oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
if (print)
System.out.print("SynchronousQueue ");
oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
// oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
oneRun(new SynchronousQueue<Integer>(), pairs, iters / 8);
/* PriorityBlockingQueue is unbounded
if (print)
System.out.print("PriorityBlockingQueue ");
oneRun(new PriorityBlockingQueue<Integer>(iters / 2 * pairs), pairs, iters / 4);
*/
}
......
......@@ -66,7 +66,8 @@ public class Interrupt {
static void testQueue(final BlockingQueue<Object> q) {
try {
final BlockingDeque<Object> deq =
q instanceof BlockingDeque ? (BlockingDeque<Object>) q : null;
(q instanceof BlockingDeque<?>) ?
(BlockingDeque<Object>) q : null;
q.clear();
List<Fun> fs = new ArrayList<Fun>();
fs.add(new Fun() { void f() throws Throwable
......@@ -107,7 +108,10 @@ public class Interrupt {
{ deq.offerLast(1, 7, SECONDS); }});
}
checkInterrupted(fs);
} catch (Throwable t) { unexpected(t); }
} catch (Throwable t) {
System.out.printf("Failed: %s%n", q.getClass().getSimpleName());
unexpected(t);
}
}
private static void realMain(final String[] args) throws Throwable {
......
......@@ -32,44 +32,18 @@ import java.util.*;
import java.util.concurrent.*;
public class LastElement {
static volatile int passed = 0, failed = 0;
static void fail(String msg) {
failed++;
new Exception(msg).printStackTrace();
}
static void pass() {
passed++;
}
static void unexpected(Throwable t) {
failed++;
t.printStackTrace();
}
static void check(boolean condition, String msg) {
if (condition)
passed++;
else
fail(msg);
}
static void check(boolean condition) {
check(condition, "Assertion failure");
}
public static void main(String[] args) throws Throwable {
void test(String[] args) throws Throwable {
testQueue(new LinkedBlockingQueue<Integer>());
// Uncomment when LinkedBlockingDeque is integrated
//testQueue(new LinkedBlockingDeque<Integer>());
testQueue(new ArrayBlockingQueue<Integer>(10));
testQueue(new LinkedBlockingDeque<Integer>());
testQueue(new ArrayBlockingQueue<Integer>(10, true));
testQueue(new ArrayBlockingQueue<Integer>(10, false));
// testQueue(new LinkedTransferQueue<Integer>());
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
if (failed > 0) throw new Exception("Some tests failed");
}
private static void testQueue(BlockingQueue<Integer> q) throws Throwable {
void testQueue(BlockingQueue<Integer> q) throws Throwable {
Integer one = 1;
Integer two = 2;
Integer three = 3;
......@@ -102,4 +76,21 @@ public class LastElement {
catch (Throwable t) {unexpected(t);}
check(q.isEmpty() && q.size() == 0);
}
//--------------------- Infrastructure ---------------------------
volatile int passed = 0, failed = 0;
void pass() {passed++;}
void fail() {failed++; Thread.dumpStack();}
void fail(String msg) {System.err.println(msg); fail();}
void unexpected(Throwable t) {failed++; t.printStackTrace();}
void check(boolean cond) {if (cond) pass(); else fail();}
void equal(Object x, Object y) {
if (x == null ? y == null : x.equals(y)) pass();
else fail(x + " not equal to " + y);}
public static void main(String[] args) throws Throwable {
new LastElement().instanceMain(args);}
public void instanceMain(String[] args) throws Throwable {
try {test(args);} catch (Throwable t) {unexpected(t);}
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
if (failed > 0) throw new AssertionError("Some tests failed");}
}
......@@ -77,6 +77,7 @@ public class MultipleProducersSingleConsumerLoops {
print = true;
for (int i = 1; i <= maxProducers; i += (i+1) >>> 1) {
System.out.println("----------------------------------------");
System.out.println("Producers:" + i);
oneTest(i, iters);
Thread.sleep(100);
......@@ -87,29 +88,20 @@ public class MultipleProducersSingleConsumerLoops {
}
static void oneTest(int producers, int iters) throws Exception {
if (print)
System.out.print("ArrayBlockingQueue ");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), producers, iters);
if (print)
System.out.print("LinkedBlockingQueue ");
oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), producers, iters);
oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), producers, iters);
// oneRun(new LinkedTransferQueue<Integer>(), producers, iters);
// Don't run PBQ since can legitimately run out of memory
// if (print)
// System.out.print("PriorityBlockingQueue ");
// oneRun(new PriorityBlockingQueue<Integer>(), producers, iters);
if (print)
System.out.print("SynchronousQueue ");
oneRun(new SynchronousQueue<Integer>(), producers, iters);
if (print)
System.out.print("SynchronousQueue(fair) ");
System.out.println("fair implementations:");
oneRun(new SynchronousQueue<Integer>(true), producers, iters);
if (print)
System.out.print("ArrayBlockingQueue(fair)");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), producers, iters);
}
......@@ -174,6 +166,8 @@ public class MultipleProducersSingleConsumerLoops {
}
static void oneRun(BlockingQueue<Integer> q, int nproducers, int iters) throws Exception {
if (print)
System.out.printf("%-18s", q.getClass().getSimpleName());
LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
CyclicBarrier barrier = new CyclicBarrier(nproducers + 2, timer);
for (int i = 0; i < nproducers; ++i) {
......
......@@ -34,81 +34,135 @@
/*
* @test
* @bug 6805775 6815766
* @run main OfferDrainToLoops 300
* @summary Test concurrent offer vs. drainTo
*/
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@SuppressWarnings({"unchecked", "rawtypes"})
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
public class OfferDrainToLoops {
final long testDurationMillisDefault = 10L * 1000L;
final long testDurationMillis;
OfferDrainToLoops(String[] args) {
testDurationMillis = (args.length > 0) ?
Long.valueOf(args[0]) : testDurationMillisDefault;
}
void checkNotContainsNull(Iterable it) {
for (Object x : it)
check(x != null);
}
abstract class CheckedThread extends Thread {
abstract protected void realRun();
public void run() {
try { realRun(); } catch (Throwable t) { unexpected(t); }
}
{
setDaemon(true);
start();
}
}
void test(String[] args) throws Throwable {
test(new LinkedBlockingQueue());
test(new LinkedBlockingQueue(2000));
test(new LinkedBlockingDeque());
test(new LinkedBlockingDeque(2000));
test(new ArrayBlockingQueue(2000));
// test(new LinkedTransferQueue());
}
Random getRandom() {
return new Random();
// return ThreadLocalRandom.current();
}
void test(final BlockingQueue q) throws Throwable {
System.out.println(q.getClass().getSimpleName());
final long testDurationSeconds = 1L;
final long testDurationMillis = testDurationSeconds * 1000L;
final long quittingTimeNanos
= System.nanoTime() + testDurationSeconds * 1000L * 1000L * 1000L;
final long testDurationNanos = testDurationMillis * 1000L * 1000L;
final long quittingTimeNanos = System.nanoTime() + testDurationNanos;
final long timeoutMillis = 10L * 1000L;
Thread offerer = new CheckedThread() {
/** Poor man's bounded buffer. */
final AtomicLong approximateCount = new AtomicLong(0L);
abstract class CheckedThread extends Thread {
CheckedThread(String name) {
super(name);
setDaemon(true);
start();
}
/** Polls for quitting time. */
protected boolean quittingTime() {
return System.nanoTime() - quittingTimeNanos > 0;
}
/** Polls occasionally for quitting time. */
protected boolean quittingTime(long i) {
return (i % 1024) == 0 && quittingTime();
}
abstract protected void realRun();
public void run() {
try { realRun(); } catch (Throwable t) { unexpected(t); }
}
}
Thread offerer = new CheckedThread("offerer") {
protected void realRun() {
for (long i = 0; ; i++) {
if ((i % 1024) == 0 &&
System.nanoTime() - quittingTimeNanos > 0)
break;
while (! q.offer(i))
long c = 0;
for (long i = 0; ! quittingTime(i); i++) {
if (q.offer(c)) {
if ((++c % 1024) == 0) {
approximateCount.getAndAdd(1024);
while (approximateCount.get() > 10000)
Thread.yield();
}
} else {
Thread.yield();
}}};
}}}};
Thread drainer = new CheckedThread() {
Thread drainer = new CheckedThread("drainer") {
protected void realRun() {
for (long i = 0; ; i++) {
if (System.nanoTime() - quittingTimeNanos > 0)
break;
final Random rnd = getRandom();
while (! quittingTime()) {
List list = new ArrayList();
int n = q.drainTo(list);
int n = rnd.nextBoolean() ?
q.drainTo(list) :
q.drainTo(list, 100);
approximateCount.getAndAdd(-n);
equal(list.size(), n);
for (int j = 0; j < n - 1; j++)
equal((Long) list.get(j) + 1L, list.get(j + 1));
Thread.yield();
}}};
}
q.clear();
approximateCount.set(0); // Releases waiting offerer thread
}};
Thread scanner = new CheckedThread() {
Thread scanner = new CheckedThread("scanner") {
protected void realRun() {
for (long i = 0; ; i++) {
if (System.nanoTime() - quittingTimeNanos > 0)
final Random rnd = getRandom();
while (! quittingTime()) {
switch (rnd.nextInt(3)) {
case 0: checkNotContainsNull(q); break;
case 1: q.size(); break;
case 2:
Long[] a = (Long[]) q.toArray(new Long[0]);
int n = a.length;
for (int j = 0; j < n - 1; j++) {
check(a[j] < a[j+1]);
check(a[j] != null);
}
break;
checkNotContainsNull(q);
}
Thread.yield();
}}};
offerer.join(10 * testDurationMillis);
drainer.join(10 * testDurationMillis);
check(! offerer.isAlive());
check(! drainer.isAlive());
for (Thread thread : new Thread[] { offerer, drainer, scanner }) {
thread.join(timeoutMillis + testDurationMillis);
if (thread.isAlive()) {
System.err.printf("Hung thread: %s%n", thread.getName());
failed++;
for (StackTraceElement e : thread.getStackTrace())
System.err.println(e);
// Kludge alert
thread.stop();
thread.join(timeoutMillis);
}
}
}
//--------------------- Infrastructure ---------------------------
......@@ -122,7 +176,7 @@ public class OfferDrainToLoops {
if (x == null ? y == null : x.equals(y)) pass();
else fail(x + " not equal to " + y);}
public static void main(String[] args) throws Throwable {
new OfferDrainToLoops().instanceMain(args);}
new OfferDrainToLoops(args).instanceMain(args);}
public void instanceMain(String[] args) throws Throwable {
try {test(args);} catch (Throwable t) {unexpected(t);}
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
......
......@@ -46,6 +46,7 @@ public class PollMemoryLeak {
public static void main(String[] args) throws InterruptedException {
final BlockingQueue[] qs = {
new LinkedBlockingQueue(10),
// new LinkedTransferQueue(),
new ArrayBlockingQueue(10),
new SynchronousQueue(),
new SynchronousQueue(true),
......
......@@ -77,7 +77,8 @@ public class ProducerConsumerLoops {
print = true;
for (int i = 1; i <= maxPairs; i += (i+1) >>> 1) {
System.out.println("Pairs:" + i);
System.out.println("----------------------------------------");
System.out.println("Pairs: " + i);
oneTest(i, iters);
Thread.sleep(100);
}
......@@ -87,28 +88,17 @@ public class ProducerConsumerLoops {
}
static void oneTest(int pairs, int iters) throws Exception {
if (print)
System.out.print("ArrayBlockingQueue ");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
if (print)
System.out.print("LinkedBlockingQueue ");
oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
if (print)
System.out.print("PriorityBlockingQueue ");
oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
// oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
oneRun(new PriorityBlockingQueue<Integer>(), pairs, iters);
if (print)
System.out.print("SynchronousQueue ");
oneRun(new SynchronousQueue<Integer>(), pairs, iters);
if (print)
System.out.print("SynchronousQueue(fair) ");
oneRun(new SynchronousQueue<Integer>(true), pairs, iters);
System.out.println("fair implementations:");
if (print)
System.out.print("ArrayBlockingQueue(fair)");
oneRun(new SynchronousQueue<Integer>(true), pairs, iters);
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), pairs, iters);
}
......@@ -174,6 +164,8 @@ public class ProducerConsumerLoops {
}
static void oneRun(BlockingQueue<Integer> q, int npairs, int iters) throws Exception {
if (print)
System.out.printf("%-18s", q.getClass().getSimpleName());
LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
CyclicBarrier barrier = new CyclicBarrier(npairs * 2 + 1, timer);
for (int i = 0; i < npairs; ++i) {
......
......@@ -63,7 +63,8 @@ public class SingleProducerMultipleConsumerLoops {
print = true;
for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
System.out.println("Consumers:" + i);
System.out.println("----------------------------------------");
System.out.println("Consumers: " + i);
oneTest(i, iters);
Thread.sleep(100);
}
......@@ -73,28 +74,15 @@ public class SingleProducerMultipleConsumerLoops {
}
static void oneTest(int consumers, int iters) throws Exception {
if (print)
System.out.print("ArrayBlockingQueue ");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), consumers, iters);
if (print)
System.out.print("LinkedBlockingQueue ");
oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), consumers, iters);
if (print)
System.out.print("PriorityBlockingQueue ");
oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), consumers, iters);
// oneRun(new LinkedTransferQueue<Integer>(), consumers, iters);
oneRun(new PriorityBlockingQueue<Integer>(), consumers, iters);
if (print)
System.out.print("SynchronousQueue ");
oneRun(new SynchronousQueue<Integer>(), consumers, iters);
if (print)
System.out.print("SynchronousQueue(fair) ");
System.out.println("fair implementations:");
oneRun(new SynchronousQueue<Integer>(true), consumers, iters);
if (print)
System.out.print("ArrayBlockingQueue(fair)");
oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), consumers, iters);
}
......@@ -163,6 +151,8 @@ public class SingleProducerMultipleConsumerLoops {
}
static void oneRun(BlockingQueue<Integer> q, int nconsumers, int iters) throws Exception {
if (print)
System.out.printf("%-18s", q.getClass().getSimpleName());
LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
CyclicBarrier barrier = new CyclicBarrier(nconsumers + 2, timer);
pool.execute(new Producer(q, barrier, iters * nconsumers));
......
......@@ -60,20 +60,10 @@ public class ConcurrentQueueLoops {
//queues.add(new ArrayBlockingQueue<Integer>(count, true));
queues.add(new LinkedBlockingQueue<Integer>());
queues.add(new LinkedBlockingDeque<Integer>());
try {
queues.add((Queue<Integer>)
Class.forName("java.util.concurrent.LinkedTransferQueue")
.newInstance());
} catch (IllegalAccessException e) {
} catch (InstantiationException e) {
} catch (ClassNotFoundException e) {
// OK; not yet added to JDK
}
// queues.add(new LinkedTransferQueue<Integer>());
// Following additional implementations are available from:
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
// queues.add(new LinkedTransferQueue<Integer>());
// queues.add(new SynchronizedLinkedListQueue<Integer>());
// Avoid "first fast, second slow" benchmark effect.
......
......@@ -41,8 +41,9 @@
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
// import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.LinkedList;
import java.util.PriorityQueue;
......@@ -69,20 +70,10 @@ public class GCRetention {
queues.add(new PriorityBlockingQueue<Boolean>());
queues.add(new PriorityQueue<Boolean>());
queues.add(new LinkedList<Boolean>());
try {
queues.add((Queue<Boolean>)
Class.forName("java.util.concurrent.LinkedTransferQueue")
.newInstance());
} catch (IllegalAccessException e) {
} catch (InstantiationException e) {
} catch (ClassNotFoundException e) {
// OK; not yet added to JDK
}
// queues.add(new LinkedTransferQueue<Boolean>());
// Following additional implementations are available from:
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
// queues.add(new LinkedTransferQueue<Boolean>());
// queues.add(new SynchronizedLinkedListQueue<Boolean>());
// Avoid "first fast, second slow" benchmark effect.
......
......@@ -49,6 +49,7 @@ public class IteratorWeakConsistency {
test(new LinkedBlockingDeque());
test(new LinkedBlockingDeque(20));
test(new ConcurrentLinkedQueue());
// test(new LinkedTransferQueue());
// Other concurrent queues (e.g. ArrayBlockingQueue) do not
// currently have weakly consistent iterators.
// test(new ArrayBlockingQueue(20));
......
......@@ -23,63 +23,136 @@
/*
* @test
* @bug 6316155 6595669
* @bug 6316155 6595669 6871697 6868712
* @summary Test concurrent offer vs. remove
* @run main OfferRemoveLoops 300
* @author Martin Buchholz
*/
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
public class OfferRemoveLoops {
final long testDurationMillisDefault = 10L * 1000L;
final long testDurationMillis;
OfferRemoveLoops(String[] args) {
testDurationMillis = (args.length > 0) ?
Long.valueOf(args[0]) : testDurationMillisDefault;
}
void checkNotContainsNull(Iterable it) {
for (Object x : it)
check(x != null);
}
void test(String[] args) throws Throwable {
testQueue(new LinkedBlockingQueue<String>(10));
testQueue(new LinkedBlockingQueue<String>());
testQueue(new LinkedBlockingDeque<String>(10));
testQueue(new LinkedBlockingDeque<String>());
testQueue(new ArrayBlockingQueue<String>(10));
testQueue(new PriorityBlockingQueue<String>(10));
testQueue(new ConcurrentLinkedQueue<String>());
testQueue(new LinkedBlockingQueue(10));
testQueue(new LinkedBlockingQueue());
testQueue(new LinkedBlockingDeque(10));
testQueue(new LinkedBlockingDeque());
testQueue(new ArrayBlockingQueue(10));
testQueue(new PriorityBlockingQueue(10));
testQueue(new ConcurrentLinkedQueue());
// testQueue(new LinkedTransferQueue());
}
abstract class CheckedThread extends Thread {
abstract protected void realRun();
public void run() {
try { realRun(); } catch (Throwable t) { unexpected(t); }
}
Random getRandom() {
return new Random();
// return ThreadLocalRandom.current();
}
void testQueue(final Queue<String> q) throws Throwable {
void testQueue(final Queue q) throws Throwable {
System.out.println(q.getClass().getSimpleName());
final int count = 1000 * 1000;
final long testDurationSeconds = 1L;
final long testDurationMillis = testDurationSeconds * 1000L;
final long quittingTimeNanos
= System.nanoTime() + testDurationSeconds * 1000L * 1000L * 1000L;
Thread t1 = new CheckedThread() {
final long testDurationNanos = testDurationMillis * 1000L * 1000L;
final long quittingTimeNanos = System.nanoTime() + testDurationNanos;
final long timeoutMillis = 10L * 1000L;
final int maxChunkSize = 1042;
final int maxQueueSize = 10 * maxChunkSize;
/** Poor man's bounded buffer. */
final AtomicLong approximateCount = new AtomicLong(0L);
abstract class CheckedThread extends Thread {
CheckedThread(String name) {
super(name);
setDaemon(true);
start();
}
/** Polls for quitting time. */
protected boolean quittingTime() {
return System.nanoTime() - quittingTimeNanos > 0;
}
/** Polls occasionally for quitting time. */
protected boolean quittingTime(long i) {
return (i % 1024) == 0 && quittingTime();
}
abstract protected void realRun();
public void run() {
try { realRun(); } catch (Throwable t) { unexpected(t); }
}
}
Thread offerer = new CheckedThread("offerer") {
protected void realRun() {
for (int i = 0; i < count; i++) {
if ((i % 1024) == 0 &&
System.nanoTime() - quittingTimeNanos > 0)
return;
while (! q.remove(String.valueOf(i)))
final long chunkSize = getRandom().nextInt(maxChunkSize) + 2;
long c = 0;
for (long i = 0; ! quittingTime(i); i++) {
if (q.offer(Long.valueOf(c))) {
if ((++c % chunkSize) == 0) {
approximateCount.getAndAdd(chunkSize);
while (approximateCount.get() > maxQueueSize)
Thread.yield();
}
} else {
Thread.yield();
}}};
Thread t2 = new CheckedThread() {
}}}};
Thread remover = new CheckedThread("remover") {
protected void realRun() {
for (int i = 0; i < count; i++) {
if ((i % 1024) == 0 &&
System.nanoTime() - quittingTimeNanos > 0)
return;
while (! q.offer(String.valueOf(i)))
final long chunkSize = getRandom().nextInt(maxChunkSize) + 2;
long c = 0;
for (long i = 0; ! quittingTime(i); i++) {
if (q.remove(Long.valueOf(c))) {
if ((++c % chunkSize) == 0) {
approximateCount.getAndAdd(-chunkSize);
}
} else {
Thread.yield();
}}};
t1.setDaemon(true); t2.setDaemon(true);
t1.start(); t2.start();
t1.join(10 * testDurationMillis);
t2.join(10 * testDurationMillis);
check(! t1.isAlive());
check(! t2.isAlive());
}
}
q.clear();
approximateCount.set(0); // Releases waiting offerer thread
}};
Thread scanner = new CheckedThread("scanner") {
protected void realRun() {
final Random rnd = getRandom();
while (! quittingTime()) {
switch (rnd.nextInt(3)) {
case 0: checkNotContainsNull(q); break;
case 1: q.size(); break;
case 2: checkNotContainsNull
(Arrays.asList(q.toArray(new Long[0])));
break;
}
Thread.yield();
}}};
for (Thread thread : new Thread[] { offerer, remover, scanner }) {
thread.join(timeoutMillis + testDurationMillis);
if (thread.isAlive()) {
System.err.printf("Hung thread: %s%n", thread.getName());
failed++;
for (StackTraceElement e : thread.getStackTrace())
System.err.println(e);
// Kludge alert
thread.stop();
thread.join(timeoutMillis);
}
}
}
//--------------------- Infrastructure ---------------------------
......@@ -93,7 +166,7 @@ public class OfferRemoveLoops {
if (x == null ? y == null : x.equals(y)) pass();
else fail(x + " not equal to " + y);}
public static void main(String[] args) throws Throwable {
new OfferRemoveLoops().instanceMain(args);}
new OfferRemoveLoops(args).instanceMain(args);}
public void instanceMain(String[] args) throws Throwable {
try {test(args);} catch (Throwable t) {unexpected(t);}
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
......
......@@ -45,6 +45,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
// import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.ArrayList;
import java.util.Collection;
......@@ -66,20 +67,10 @@ public class RemovePollRace {
queues.add(new ArrayBlockingQueue<Boolean>(count, true));
queues.add(new LinkedBlockingQueue<Boolean>());
queues.add(new LinkedBlockingDeque<Boolean>());
try {
queues.add((Queue<Boolean>)
Class.forName("java.util.concurrent.LinkedTransferQueue")
.newInstance());
} catch (IllegalAccessException e) {
} catch (InstantiationException e) {
} catch (ClassNotFoundException e) {
// OK; not yet added to JDK
}
// queues.add(new LinkedTransferQueue<Boolean>());
// Following additional implementations are available from:
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
// queues.add(new LinkedTransferQueue<Boolean>());
// queues.add(new SynchronizedLinkedListQueue<Boolean>());
// Avoid "first fast, second slow" benchmark effect.
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册