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 -source 1.5 ConcurrentQueueLoops.java

    38  * @run main/timeout=230 ConcurrentQueueLoops

    39  * @summary Checks that a set of threads can repeatedly get and modify items

    40  */

    41 

    42 import java.util.*;

    43 import java.util.concurrent.*;

    44 import java.util.concurrent.atomic.*;

    45 

    46 public class ConcurrentQueueLoops {

    47     static final ExecutorService pool = Executors.newCachedThreadPool();

    48     static AtomicInteger totalItems;

    49     static boolean print = false;

    50 

    51     public static void main(String[] args) throws Exception {

    52         int maxStages = 8;

    53         int items = 100000;

    54 

    55         if (args.length > 0)

    56             maxStages = Integer.parseInt(args[0]);

    57 

    58         print = false;

    59         System.out.println("Warmup...");

    60         oneRun(1, items);

    61         Thread.sleep(100);

    62         oneRun(1, items);

    63         Thread.sleep(100);

    64         print = true;

    65 

    66         for (int i = 1; i <= maxStages; i += (i+1) >>> 1) {

    67             oneRun(i, items);

    68         }

    69         pool.shutdown();

    70         if (! pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS))

    71             throw new Error();

    72    }

    73 

    74     static class Stage implements Callable<Integer> {

    75         final Queue<Integer> queue;

    76         final CyclicBarrier barrier;

    77         int items;

    78         Stage (Queue<Integer> q, CyclicBarrier b, int items) {

    79             queue = q;

    80             barrier = b;

    81             this.items = items;

    82         }

    83 

    84         public Integer call() {

    85             // Repeatedly take something from queue if possible,

    86             // transform it, and put back in.

    87             try {

    88                 barrier.await();

    89                 int l = 4321;

    90                 int takes = 0;

    91                 for (;;) {

    92                     Integer item = queue.poll();

    93                     if (item != null) {

    94                         ++takes;

    95                         l = LoopHelpers.compute2(item.intValue());

    96                     }

    97                     else if (takes != 0) {

    98                         totalItems.getAndAdd(-takes);

    99                         takes = 0;

   100                     }

   101                     else if (totalItems.get() <= 0)

   102                         break;

   103                     l = LoopHelpers.compute1(l);

   104                     if (items > 0) {

   105                         --items;

   106                         queue.offer(new Integer(l));

   107                     }

   108                     else if ( (l & (3 << 5)) == 0) // spinwait

   109                         Thread.sleep(1);

   110                 }

   111                 return new Integer(l);

   112             }

   113             catch (Exception ie) {

   114                 ie.printStackTrace();

   115                 throw new Error("Call loop failed");

   116             }

   117         }

   118     }

   119 

   120     static void oneRun(int n, int items) throws Exception {

   121         Queue<Integer> q = new ConcurrentLinkedQueue<Integer>();

   122         LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();

   123         CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);

   124         totalItems = new AtomicInteger(n * items);

   125         ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);

   126         for (int i = 0; i < n; ++i)

   127             results.add(pool.submit(new Stage(q, barrier, items)));

   128 

   129         if (print)

   130             System.out.print("Threads: " + n + "\t:");

   131         barrier.await();

   132         int total = 0;

   133         for (int i = 0; i < n; ++i) {

   134             Future<Integer> f = results.get(i);

   135             Integer r = f.get();

   136             total += r.intValue();

   137         }

   138         long endTime = System.nanoTime();

   139         long time = endTime - timer.startTime;

   140         if (print)

   141             System.out.println(LoopHelpers.rightJustify(time / (items * n)) + " ns per item");

   142         if (total == 0) // avoid overoptimization

   143             System.out.println("useless result: " + total);

   144 

   145     }

   146 }