test/java/util/concurrent/BlockingQueue/CancelledProducerConsumerLoops.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 CancelledProducerConsumerLoops.java
  38  * @run main/timeout=7000 CancelledProducerConsumerLoops
  39  * @summary Checks for responsiveness of blocking queues to cancellation.
  40  * Runs under the assumption that ITERS computations require more than
  41  * TIMEOUT msecs to complete.
  42  */
  43 
  44 import java.util.concurrent.*;
  45 
  46 public class CancelledProducerConsumerLoops {
  47     static final int CAPACITY =      100;
  48     static final long TIMEOUT = 100;
  49 
  50     static final ExecutorService pool = Executors.newCachedThreadPool();
  51     static boolean print = false;
  52 
  53     public static void main(String[] args) throws Exception {
  54         int maxPairs = 8;
  55         int iters = 1000000;
  56 
  57         if (args.length > 0)


 102 
 103         if (!tooLate) {
 104             for (int i = 1; i < npairs; ++i) {
 105                 if (!prods[i].isDone() || !prods[i].isCancelled())
 106                     throw new Error("Only one producer thread should complete");
 107                 if (!cons[i].isDone() || !cons[i].isCancelled())
 108                     throw new Error("Only one consumer thread should complete");
 109             }
 110         }
 111         else
 112             System.out.print("(cancelled too late) ");
 113 
 114         long endTime = System.nanoTime();
 115         long time = endTime - timer.startTime;
 116         if (print) {
 117             double secs = (double)(time) / 1000000000.0;
 118             System.out.println("\t " + secs + "s run time");
 119         }
 120     }
 121 
 122     static final class LTQasSQ<T> extends LinkedTransferQueue<T> {
 123         LTQasSQ() { super(); }
 124         public void put(T x) {
 125             try { super.transfer(x); }
 126             catch (InterruptedException ex) { throw new Error(); }
 127         }
 128         private final static long serialVersionUID = 42;
 129     }
 130 
 131     static final class HalfSyncLTQ<T> extends LinkedTransferQueue<T> {
 132         HalfSyncLTQ() { super(); }
 133         public void put(T x) {
 134             if (ThreadLocalRandom.current().nextBoolean())
 135                 super.put(x);
 136             else {
 137                 try { super.transfer(x); }
 138                 catch (InterruptedException ex) { throw new Error(); }
 139             }
 140         }
 141         private final static long serialVersionUID = 42;
 142     }
 143 
 144     static void oneTest(int pairs, int iters) throws Exception {
 145 
 146         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
 147         oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
 148         oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);

 149         oneRun(new SynchronousQueue<Integer>(), pairs, iters / 8);
 150 
 151         /* TODO: unbounded queue implementations are prone to OOME
 152         oneRun(new PriorityBlockingQueue<Integer>(iters / 2 * pairs), pairs, iters / 4);
 153         oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
 154         oneRun(new LTQasSQ<Integer>(), pairs, iters);
 155         oneRun(new HalfSyncLTQ<Integer>(), pairs, iters);
 156         */
 157     }
 158 
 159     static abstract class Stage implements Callable<Integer> {
 160         final BlockingQueue<Integer> queue;
 161         final CyclicBarrier barrier;
 162         final int iters;
 163         Stage (BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 164             queue = q;
 165             barrier = b;
 166             this.iters = iters;
 167         }
 168     }
 169 
 170     static class Producer extends Stage {
 171         Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 172             super(q, b, iters);
 173         }
 174 
 175         public Integer call() throws Exception {
 176             barrier.await();
 177             int s = 0;
 178             int l = 4321;
 179             for (int i = 0; i < iters; ++i) {
 180                 l = LoopHelpers.compute1(l);
 181                 s += LoopHelpers.compute2(l);
 182                 if (!queue.offer(new Integer(l), 1, TimeUnit.SECONDS))
 183                     break;




  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 CancelledProducerConsumerLoops.java
  38  * @run main/timeout=7000 CancelledProducerConsumerLoops
  39  * @summary Checks for responsiveness of blocking queues to cancellation.
  40  * Runs under the assumption that ITERS computations require more than
  41  * TIMEOUT msecs to complete.
  42  */
  43 
  44 import java.util.concurrent.*;
  45 
  46 public class CancelledProducerConsumerLoops {
  47     static final int CAPACITY =      100;
  48     static final long TIMEOUT = 100;
  49 
  50     static final ExecutorService pool = Executors.newCachedThreadPool();
  51     static boolean print = false;
  52 
  53     public static void main(String[] args) throws Exception {
  54         int maxPairs = 8;
  55         int iters = 1000000;
  56 
  57         if (args.length > 0)


 102 
 103         if (!tooLate) {
 104             for (int i = 1; i < npairs; ++i) {
 105                 if (!prods[i].isDone() || !prods[i].isCancelled())
 106                     throw new Error("Only one producer thread should complete");
 107                 if (!cons[i].isDone() || !cons[i].isCancelled())
 108                     throw new Error("Only one consumer thread should complete");
 109             }
 110         }
 111         else
 112             System.out.print("(cancelled too late) ");
 113 
 114         long endTime = System.nanoTime();
 115         long time = endTime - timer.startTime;
 116         if (print) {
 117             double secs = (double)(time) / 1000000000.0;
 118             System.out.println("\t " + secs + "s run time");
 119         }
 120     }
 121 






















 122     static void oneTest(int pairs, int iters) throws Exception {
 123 
 124         oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
 125         oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
 126         oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
 127         oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
 128         oneRun(new SynchronousQueue<Integer>(), pairs, iters / 8);
 129 
 130         /* PriorityBlockingQueue is unbounded
 131         oneRun(new PriorityBlockingQueue<Integer>(iters / 2 * pairs), pairs, iters / 4);



 132         */
 133     }
 134 
 135     abstract static class Stage implements Callable<Integer> {
 136         final BlockingQueue<Integer> queue;
 137         final CyclicBarrier barrier;
 138         final int iters;
 139         Stage(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 140             queue = q;
 141             barrier = b;
 142             this.iters = iters;
 143         }
 144     }
 145 
 146     static class Producer extends Stage {
 147         Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 148             super(q, b, iters);
 149         }
 150 
 151         public Integer call() throws Exception {
 152             barrier.await();
 153             int s = 0;
 154             int l = 4321;
 155             for (int i = 0; i < iters; ++i) {
 156                 l = LoopHelpers.compute1(l);
 157                 s += LoopHelpers.compute2(l);
 158                 if (!queue.offer(new Integer(l), 1, TimeUnit.SECONDS))
 159                     break;