--- jsr166/src/test/loops/TimeoutProducerConsumerLoops.java 2007/02/19 00:46:06 1.2 +++ jsr166/src/test/loops/TimeoutProducerConsumerLoops.java 2015/01/15 18:34:19 1.12 @@ -1,14 +1,30 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ import java.util.concurrent.*; public class TimeoutProducerConsumerLoops { - static final int CAPACITY = 100; + static final int NCPUS = Runtime.getRuntime().availableProcessors(); static final ExecutorService pool = Executors.newCachedThreadPool(); + + // 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 << 8; + 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); + } + + // 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; + static boolean print = false; static int producerSum; static int consumerSum; @@ -27,83 +43,86 @@ public class TimeoutProducerConsumerLoop } public static void main(String[] args) throws Exception { - int maxPairs = 100; - int iters = 100000; + int maxPairs = NCPUS * 3 / 2; + int iters = 1000000; - if (args.length > 0) + 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); print = true; - - int k = 1; - for (int i = 1; i <= maxPairs;) { + for (int k = 1, i = 1; i <= maxPairs;) { System.out.println("Pairs:" + i); oneTest(i, iters); Thread.sleep(100); if (i == k) { k = i << 1; i = i + (i >>> 1); - } - else + } + else i = k; } pool.shutdown(); - } + } + + static void oneTest(int n, int iters) throws Exception { + if (print) + System.out.print("LinkedTransferQueue "); + oneRun(new LinkedTransferQueue(), n, iters); + + if (print) + System.out.print("LinkedTransferQueue(xfer)"); + oneRun(new LTQasSQ(), n, iters); - static void oneTest(int pairs, int iters) throws Exception { - int fairIters = iters/20; if (print) - System.out.print("ArrayBlockingQueue "); - oneRun(new ArrayBlockingQueue(CAPACITY), pairs, iters); + System.out.print("LinkedBlockingQueue "); + oneRun(new LinkedBlockingQueue(), n, iters); if (print) - System.out.print("LinkedBlockingQueue "); - oneRun(new LinkedBlockingQueue(CAPACITY), pairs, iters); + System.out.print("LinkedBlockingQueue(cap) "); + oneRun(new LinkedBlockingQueue(POOL_SIZE), n, iters); if (print) - System.out.print("LinkedBlockingDeque "); - oneRun(new LinkedBlockingDeque(CAPACITY), pairs, iters); + System.out.print("ArrayBlockingQueue(cap) "); + oneRun(new ArrayBlockingQueue(POOL_SIZE), n, iters); if (print) - System.out.print("SynchronousQueue "); - oneRun(new SynchronousQueue(), pairs, iters); + System.out.print("LinkedBlockingDeque "); + oneRun(new LinkedBlockingDeque(), n, iters); if (print) - System.out.print("SynchronousQueue(fair) "); - oneRun(new SynchronousQueue(true), pairs, fairIters); + System.out.print("SynchronousQueue "); + oneRun(new SynchronousQueue(), n, iters); if (print) - System.out.print("PriorityBlockingQueue "); - oneRun(new PriorityBlockingQueue(), pairs, fairIters); + System.out.print("SynchronousQueue(fair) "); + oneRun(new SynchronousQueue(true), n, iters); if (print) - System.out.print("ArrayBlockingQueue(fair)"); - oneRun(new ArrayBlockingQueue(CAPACITY, true), pairs, fairIters); + System.out.print("PriorityBlockingQueue "); + oneRun(new PriorityBlockingQueue(), n, iters / 16); + + if (print) + System.out.print("ArrayBlockingQueue(fair) "); + oneRun(new ArrayBlockingQueue(POOL_SIZE, true), n, iters/16); } - - static abstract class Stage implements Runnable { + + abstract static class Stage implements Runnable { final int iters; final BlockingQueue queue; final CyclicBarrier barrier; - Stage (BlockingQueue q, CyclicBarrier b, int iters) { - queue = q; + 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() { @@ -112,15 +131,17 @@ public class TimeoutProducerConsumerLoop int s = 0; int l = hashCode(); int i = 0; - long timeout = 1; + long timeout = 1000; while (i < iters) { l = LoopHelpers.compute4(l); - if (queue.offer(new Integer(l), - timeout, TimeUnit.NANOSECONDS)) { - s += LoopHelpers.compute4(l); + Integer v = intPool[l & POOL_MASK]; + if (queue.offer(v, timeout, TimeUnit.NANOSECONDS)) { + s += LoopHelpers.compute4(v.intValue()); ++i; if (timeout > 1) timeout--; + if ((i & LAG_MASK) == LAG_MASK) + lagPhaser.arriveAndAwaitAdvance(); } else timeout++; @@ -128,16 +149,16 @@ public class TimeoutProducerConsumerLoop addProducerSum(s); barrier.await(); } - catch (Exception ie) { - ie.printStackTrace(); - return; + catch (Exception ie) { + ie.printStackTrace(); + return; } } } 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() { @@ -146,9 +167,9 @@ public class TimeoutProducerConsumerLoop int l = 0; int s = 0; int i = 0; - long timeout = 1; + long timeout = 1000; while (i < iters) { - Integer e = queue.poll(timeout, + Integer e = queue.poll(timeout, TimeUnit.NANOSECONDS); if (e != null) { l = LoopHelpers.compute4(e.intValue()); @@ -156,6 +177,8 @@ public class TimeoutProducerConsumerLoop ++i; if (timeout > 1) --timeout; + if ((i & LAG_MASK) == LAG_MASK) + lagPhaser.arriveAndAwaitAdvance(); } else ++timeout; @@ -163,9 +186,9 @@ public class TimeoutProducerConsumerLoop addConsumerSum(s); barrier.await(); } - catch (Exception ie) { - ie.printStackTrace(); - return; + catch (Exception ie) { + ie.printStackTrace(); + return; } } @@ -175,15 +198,30 @@ public class TimeoutProducerConsumerLoop 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)); + 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(); + q.clear(); if (print) System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * npairs)) + " 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(); } + } + + public boolean offer(T x, long timeout, TimeUnit unit) { + return super.offer(x, timeout, unit); + } + + } + }