1 /*
   2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   3  *
   4  * This code is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This code is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  11  * version 2 for more details (a copy is included in the LICENSE file that
  12  * accompanied this code).
  13  *
  14  * You should have received a copy of the GNU General Public License version
  15  * 2 along with this work; if not, write to the Free Software Foundation,
  16  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  17  *
  18  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  19  * CA 95054 USA or visit www.sun.com if you need additional information or
  20  * have any 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             }
 139             catch (Exception ie) {
 140                 ie.printStackTrace();
 141                 return;
 142             }
 143         }
 144     }
 145 
 146     static class Consumer extends Stage {
 147         Consumer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
 148             super(q, b, iters);
 149         }
 150 
 151         public void run() {
 152             try {
 153                 barrier.await();
 154                 int l = 0;
 155                 int s = 0;
 156                 int last = -1;
 157                 for (int i = 0; i < iters; ++i) {
 158                     Integer item = queue.take();
 159                     int v = item.intValue();
 160                     if (v < last)
 161                         throw new Error("Out-of-Order transfer");
 162                     last = v;
 163                     l = LoopHelpers.compute1(v);
 164                     s += l;
 165                 }
 166                 barrier.await();
 167                 result = s;
 168             }
 169             catch (Exception ie) {
 170                 ie.printStackTrace();
 171                 return;
 172             }
 173         }
 174 
 175     }
 176 
 177     static void oneRun(BlockingQueue<Integer> q, int nconsumers, int iters) throws Exception {
 178         if (print)
 179             System.out.printf("%-18s", q.getClass().getSimpleName());
 180         LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
 181         CyclicBarrier barrier = new CyclicBarrier(nconsumers + 2, timer);
 182         pool.execute(new Producer(q, barrier, iters * nconsumers));
 183         for (int i = 0; i < nconsumers; ++i) {
 184             pool.execute(new Consumer(q, barrier, iters));
 185         }
 186         barrier.await();
 187         barrier.await();
 188         long time = timer.getTime();
 189         if (print)
 190             System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * nconsumers)) + " ns per transfer");
 191     }
 192 
 193 }