test/java/util/concurrent/BlockingQueue/SingleProducerMultipleConsumerLoops.java

Print this page




  17  *
  18  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  19  * or visit www.oracle.com if you need additional information or have any
  20  * questions.
  21  */
  22 
  23 /*
  24  * This file is available under and governed by the GNU General Public
  25  * License version 2 only, as published by the Free Software Foundation.
  26  * However, the following notice accompanied the original version of this
  27  * file:
  28  *
  29  * Written by Doug Lea with assistance from members of JCP JSR-166
  30  * Expert Group and released to the public domain, as explained at
  31  * http://creativecommons.org/licenses/publicdomain
  32  */
  33 
  34 /*
  35  * @test
  36  * @bug 4486658
  37  * @compile SingleProducerMultipleConsumerLoops.java
  38  * @run main/timeout=600 SingleProducerMultipleConsumerLoops
  39  * @summary  check ordering for blocking queues with 1 producer and multiple consumers
  40  */
  41 
  42 import java.util.concurrent.*;
  43 
  44 public class SingleProducerMultipleConsumerLoops {
  45     static final int CAPACITY =      100;
  46 
  47     static final ExecutorService pool = Executors.newCachedThreadPool();
  48     static boolean print = false;
  49 
  50     public static void main(String[] args) throws Exception {
  51         int maxConsumers = 5;
  52         int iters = 10000;
  53 
  54         if (args.length > 0)
  55             maxConsumers = Integer.parseInt(args[0]);
  56 
  57         print = false;
  58         System.out.println("Warmup...");
  59         oneTest(1, 10000);
  60         Thread.sleep(100);
  61         oneTest(2, 10000);
  62         Thread.sleep(100);
  63         print = true;
  64 
  65         for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
  66             System.out.println("----------------------------------------");
  67             System.out.println("Consumers: " + i);
  68             oneTest(i, iters);
  69             Thread.sleep(100);
  70         }
  71         pool.shutdown();
  72         if (! pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS))
  73             throw new Error();
  74    }
  75 
  76     static final class LTQasSQ<T> extends LinkedTransferQueue<T> {
  77         LTQasSQ() { super(); }
  78         public void put(T x) {
  79             try { super.transfer(x); }
  80             catch (InterruptedException ex) { throw new Error(); }
  81         }
  82         private final static long serialVersionUID = 42;
  83     }
  84 
  85     static final class HalfSyncLTQ<T> extends LinkedTransferQueue<T> {
  86         HalfSyncLTQ() { super(); }
  87         public void put(T x) {
  88             if (ThreadLocalRandom.current().nextBoolean())
  89                 super.put(x);
  90             else {
  91                 try { super.transfer(x); }
  92                 catch (InterruptedException ex) { throw new Error(); }
  93             }
  94         }
  95         private final static long serialVersionUID = 42;
  96     }
  97 
  98     static void oneTest(int consumers, int iters) throws Exception {
  99         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), consumers, iters);
 100         oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), consumers, iters);
 101         oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), consumers, iters);
 102         oneRun(new LinkedTransferQueue<Integer>(), consumers, iters);
 103         oneRun(new LTQasSQ<Integer>(), consumers, iters);
 104         oneRun(new HalfSyncLTQ<Integer>(), consumers, iters);
 105         oneRun(new PriorityBlockingQueue<Integer>(), consumers, iters);
 106         oneRun(new SynchronousQueue<Integer>(), consumers, iters);
 107         if (print)
 108             System.out.println("fair implementations:");
 109         oneRun(new SynchronousQueue<Integer>(true), consumers, iters);
 110         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), consumers, iters);
 111     }
 112 
 113     static abstract class Stage implements Runnable {
 114         final int iters;
 115         final BlockingQueue<Integer> queue;
 116         final CyclicBarrier barrier;
 117         volatile int result;
 118         Stage (BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 119             queue = q;
 120             barrier = b;
 121             this.iters = iters;
 122         }
 123     }
 124 
 125     static class Producer extends Stage {
 126         Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 127             super(q, b, iters);
 128         }
 129 
 130         public void run() {
 131             try {
 132                 barrier.await();
 133                 for (int i = 0; i < iters; ++i) {
 134                     queue.put(new Integer(i));
 135                 }
 136                 barrier.await();
 137                 result = 432;
 138             }




  17  *
  18  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  19  * or visit www.oracle.com if you need additional information or have any
  20  * questions.
  21  */
  22 
  23 /*
  24  * This file is available under and governed by the GNU General Public
  25  * License version 2 only, as published by the Free Software Foundation.
  26  * However, the following notice accompanied the original version of this
  27  * file:
  28  *
  29  * Written by Doug Lea with assistance from members of JCP JSR-166
  30  * Expert Group and released to the public domain, as explained at
  31  * http://creativecommons.org/licenses/publicdomain
  32  */
  33 
  34 /*
  35  * @test
  36  * @bug 4486658
  37  * @compile -source 1.5 SingleProducerMultipleConsumerLoops.java
  38  * @run main/timeout=600 SingleProducerMultipleConsumerLoops
  39  * @summary  check ordering for blocking queues with 1 producer and multiple consumers
  40  */
  41 
  42 import java.util.concurrent.*;
  43 
  44 public class SingleProducerMultipleConsumerLoops {
  45     static final int CAPACITY =      100;
  46 
  47     static final ExecutorService pool = Executors.newCachedThreadPool();
  48     static boolean print = false;
  49 
  50     public static void main(String[] args) throws Exception {
  51         int maxConsumers = 5;
  52         int iters = 10000;
  53 
  54         if (args.length > 0)
  55             maxConsumers = Integer.parseInt(args[0]);
  56 
  57         print = false;
  58         System.out.println("Warmup...");
  59         oneTest(1, 10000);
  60         Thread.sleep(100);
  61         oneTest(2, 10000);
  62         Thread.sleep(100);
  63         print = true;
  64 
  65         for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
  66             System.out.println("----------------------------------------");
  67             System.out.println("Consumers: " + i);
  68             oneTest(i, iters);
  69             Thread.sleep(100);
  70         }
  71         pool.shutdown();
  72         if (! pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS))
  73             throw new Error();
  74    }
  75 






















  76     static void oneTest(int consumers, int iters) throws Exception {
  77         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), consumers, iters);
  78         oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), consumers, iters);
  79         oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), consumers, iters);
  80         oneRun(new LinkedTransferQueue<Integer>(), consumers, iters);


  81         oneRun(new PriorityBlockingQueue<Integer>(), consumers, iters);
  82         oneRun(new SynchronousQueue<Integer>(), consumers, iters);
  83         if (print)
  84             System.out.println("fair implementations:");
  85         oneRun(new SynchronousQueue<Integer>(true), consumers, iters);
  86         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), consumers, iters);
  87     }
  88 
  89     abstract static class Stage implements Runnable {
  90         final int iters;
  91         final BlockingQueue<Integer> queue;
  92         final CyclicBarrier barrier;
  93         volatile int result;
  94         Stage(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
  95             queue = q;
  96             barrier = b;
  97             this.iters = iters;
  98         }
  99     }
 100 
 101     static class Producer extends Stage {
 102         Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 103             super(q, b, iters);
 104         }
 105 
 106         public void run() {
 107             try {
 108                 barrier.await();
 109                 for (int i = 0; i < iters; ++i) {
 110                     queue.put(new Integer(i));
 111                 }
 112                 barrier.await();
 113                 result = 432;
 114             }