--- jsr166/src/test/loops/ProducerConsumerLoops.java 2007/02/19 00:46:06 1.2 +++ jsr166/src/test/loops/ProducerConsumerLoops.java 2009/10/23 19:57:06 1.3 @@ -4,11 +4,13 @@ * http://creativecommons.org/licenses/publicdomain */ +import java.util.*; import java.util.concurrent.*; +//import jsr166y.*; public class ProducerConsumerLoops { - static final int CAPACITY = 100; - + static final int NCPUS = Runtime.getRuntime().availableProcessors(); + static final Random rng = new Random(); static final ExecutorService pool = Executors.newCachedThreadPool(); static boolean print = false; static int producerSum; @@ -26,28 +28,36 @@ public class ProducerConsumerLoops { throw new Error("CheckSum mismatch"); } + // Number of elements passed around -- must be power of two + // Elements are reused from pool to minimize alloc impact + static final int POOL_SIZE = 1 << 7; + static final int POOL_MASK = POOL_SIZE-1; + static final Integer[] intPool = new Integer[POOL_SIZE]; + static { + for (int i = 0; i < POOL_SIZE; ++i) + intPool[i] = Integer.valueOf(i); + } + + // Number of puts by producers or takes by consumers + static final int ITERS = 1 << 20; + + // max lag between a producer and consumer to avoid + // this becoming a GC test rather than queue test. + // Used only per-pair to lessen impact on queue sync + static final int LAG_MASK = (1 << 12) - 1; + public static void main(String[] args) throws Exception { - int maxPairs = 100; - int iters = 100000; + int maxPairs = NCPUS * 3 / 2; if (args.length > 0) maxPairs = Integer.parseInt(args[0]); - print = false; - System.out.println("Warmup..."); - oneTest(1, 10000); - Thread.sleep(100); - oneTest(2, 10000); - Thread.sleep(100); - oneTest(2, 10000); - Thread.sleep(100); + warmup(); print = true; - int k = 1; for (int i = 1; i <= maxPairs;) { System.out.println("Pairs:" + i); - oneTest(i, iters); - Thread.sleep(100); + oneTest(i, ITERS); if (i == k) { k = i << 1; i = i + (i >>> 1); @@ -56,37 +66,86 @@ public class ProducerConsumerLoops { i = k; } pool.shutdown(); - } + } - static void oneTest(int pairs, int iters) throws Exception { - int fairIters = iters/20; + static void warmup() throws Exception { + print = false; + System.out.print("Warmup "); + int it = 2000; + for (int j = 5; j > 0; --j) { + oneTest(j, it); + System.out.print("."); + it += 1000; + } + System.gc(); + it = 20000; + for (int j = 5; j > 0; --j) { + oneTest(j, it); + System.out.print("."); + it += 10000; + } + System.gc(); + System.out.println(); + } + + static void oneTest(int n, int iters) throws Exception { + int fairIters = iters/16; + + Thread.sleep(100); // System.gc(); if (print) - System.out.print("ArrayBlockingQueue "); - oneRun(new ArrayBlockingQueue(CAPACITY), pairs, iters); + System.out.print("LinkedTransferQueue "); + oneRun(new LinkedTransferQueue(), n, iters); + Thread.sleep(100); // System.gc(); if (print) System.out.print("LinkedBlockingQueue "); - oneRun(new LinkedBlockingQueue(CAPACITY), pairs, iters); + oneRun(new LinkedBlockingQueue(), n, iters); + + Thread.sleep(100); // System.gc(); + if (print) + System.out.print("LinkedBlockingQueue(cap)"); + oneRun(new LinkedBlockingQueue(POOL_SIZE), n, iters); + Thread.sleep(100); // System.gc(); if (print) System.out.print("LinkedBlockingDeque "); - oneRun(new LinkedBlockingDeque(CAPACITY), pairs, iters); + oneRun(new LinkedBlockingDeque(), n, iters); + + Thread.sleep(100); // System.gc(); + if (print) + System.out.print("ArrayBlockingQueue "); + oneRun(new ArrayBlockingQueue(POOL_SIZE), n, iters); + Thread.sleep(100); // System.gc(); if (print) System.out.print("SynchronousQueue "); - oneRun(new SynchronousQueue(), pairs, iters); + oneRun(new SynchronousQueue(), n, iters); + + Thread.sleep(100); // System.gc(); if (print) System.out.print("SynchronousQueue(fair) "); - oneRun(new SynchronousQueue(true), pairs, fairIters); + oneRun(new SynchronousQueue(true), n, iters); + + Thread.sleep(100); // System.gc(); + if (print) + System.out.print("LinkedTransferQueue(xfer)"); + oneRun(new LTQasSQ(), n, iters); + Thread.sleep(100); // System.gc(); + if (print) + System.out.print("LinkedTransferQueue(half)"); + oneRun(new HalfSyncLTQ(), n, iters); + + Thread.sleep(100); // System.gc(); if (print) System.out.print("PriorityBlockingQueue "); - oneRun(new PriorityBlockingQueue(), pairs, fairIters); + oneRun(new PriorityBlockingQueue(), n, fairIters); + Thread.sleep(100); // System.gc(); if (print) System.out.print("ArrayBlockingQueue(fair)"); - oneRun(new ArrayBlockingQueue(CAPACITY, true), pairs, fairIters); + oneRun(new ArrayBlockingQueue(POOL_SIZE, true), n, fairIters); } @@ -94,29 +153,37 @@ public class ProducerConsumerLoops { final int iters; final BlockingQueue queue; final CyclicBarrier barrier; - Stage (BlockingQueue q, CyclicBarrier b, int iters) { + final Phaser lagPhaser; + Stage (BlockingQueue q, CyclicBarrier b, Phaser s, + int iters) { queue = q; barrier = b; + lagPhaser = s; this.iters = iters; } } static class Producer extends Stage { - Producer(BlockingQueue q, CyclicBarrier b, int iters) { - super(q, b, iters); + Producer(BlockingQueue q, CyclicBarrier b, Phaser s, + int iters) { + super(q, b, s, iters); } public void run() { try { barrier.await(); - int s = 0; - int l = hashCode(); + int ps = 0; + int r = hashCode(); for (int i = 0; i < iters; ++i) { - l = LoopHelpers.compute4(l); - queue.put(new Integer(l)); - s += LoopHelpers.compute4(l); + r = LoopHelpers.compute7(r); + Integer v = intPool[r & POOL_MASK]; + int k = v.intValue(); + queue.put(v); + ps += k; + if ((i & LAG_MASK) == LAG_MASK) + lagPhaser.arriveAndAwaitAdvance(); } - addProducerSum(s); + addProducerSum(ps); barrier.await(); } catch (Exception ie) { @@ -127,20 +194,23 @@ public class ProducerConsumerLoops { } static class Consumer extends Stage { - Consumer(BlockingQueue q, CyclicBarrier b, int iters) { - super(q, b, iters); + Consumer(BlockingQueue q, CyclicBarrier b, Phaser s, + int iters) { + super(q, b, s, iters); } public void run() { try { barrier.await(); - int l = 0; - int s = 0; + int cs = 0; for (int i = 0; i < iters; ++i) { - l = LoopHelpers.compute4(queue.take().intValue()); - s += l; + Integer v = queue.take(); + int k = v.intValue(); + cs += k; + if ((i & LAG_MASK) == LAG_MASK) + lagPhaser.arriveAndAwaitAdvance(); } - addConsumerSum(s); + addConsumerSum(cs); barrier.await(); } catch (Exception ie) { @@ -151,19 +221,42 @@ public class ProducerConsumerLoops { } - static void oneRun(BlockingQueue q, int npairs, int iters) throws Exception { + static void oneRun(BlockingQueue q, int n, int iters) throws Exception { LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer(); - CyclicBarrier barrier = new CyclicBarrier(npairs * 2 + 1, timer); - for (int i = 0; i < npairs; ++i) { - pool.execute(new Producer(q, barrier, iters)); - pool.execute(new Consumer(q, barrier, iters)); + CyclicBarrier barrier = new CyclicBarrier(n * 2 + 1, timer); + for (int i = 0; i < n; ++i) { + Phaser s = new Phaser(2); + pool.execute(new Producer(q, barrier, s, iters)); + pool.execute(new Consumer(q, barrier, s, iters)); } barrier.await(); barrier.await(); long time = timer.getTime(); checkSum(); if (print) - System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * npairs)) + " ns per transfer"); + System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * n)) + " ns per transfer"); + } + + static final class LTQasSQ extends LinkedTransferQueue { + LTQasSQ() { super(); } + public void put(T x) { + try { super.transfer(x); + } catch (InterruptedException ex) { throw new Error(); } + } } + static final class HalfSyncLTQ extends LinkedTransferQueue { + int calls; + HalfSyncLTQ() { super(); } + public void put(T x) { + if ((++calls & 1) == 0) + super.put(x); + else { + try { super.transfer(x); + } catch (InterruptedException ex) { + throw new Error(); + } + } + } + } }