ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/test/loops/ConcurrentQueueLoops.java
(Generate patch)

Comparing jsr166/src/test/loops/ConcurrentQueueLoops.java (file contents):
Revision 1.1 by dl, Mon May 2 19:19:38 2005 UTC vs.
Revision 1.5 by dl, Mon Feb 19 00:46:06 2007 UTC

# Line 1 | Line 1
1   /*
2 – * @test %I% %E%
3 – * @bug 4486658
4 – * @compile -source 1.5 ConcurrentQueueLoops.java
5 – * @run main/timeout=230 ConcurrentQueueLoops
6 – * @summary Checks that a set of threads can repeatedly get and modify items
7 – */
8 – /*
2   * Written by Doug Lea with assistance from members of JCP JSR-166
3 < * Expert Group and released to the public domain. Use, modify, and
4 < * redistribute this code in any way without acknowledgement.
3 > * Expert Group and released to the public domain, as explained at
4 > * http://creativecommons.org/licenses/publicdomain
5   */
6  
7   import java.util.*;
8   import java.util.concurrent.*;
9 + import java.util.concurrent.locks.*;
10   import java.util.concurrent.atomic.*;
11  
12   public class ConcurrentQueueLoops {
13      static final ExecutorService pool = Executors.newCachedThreadPool();
20 –    static AtomicInteger totalItems;
14      static boolean print = false;
15 +    static final Integer zero = new Integer(0);
16 +    static final Integer one = new Integer(1);
17 +    static int workMask;
18 +    static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;
19 +    static final int BATCH_SIZE = 8;
20  
21      public static void main(String[] args) throws Exception {
22 <        int maxStages = 8;
23 <        int items = 100000;
26 <
22 >        int maxStages = 100;
23 >        int work = 1024;
24          Class klass = null;
25          if (args.length > 0) {
26              try {
# Line 32 | Line 29 | public class ConcurrentQueueLoops {
29                  throw new RuntimeException("Class " + args[0] + " not found.");
30              }
31          }
35 –        else
36 –            klass = java.util.concurrent.ConcurrentLinkedQueue.class;
32  
33          if (args.length > 1)
34              maxStages = Integer.parseInt(args[1]);
35  
36 +        if (args.length > 2)
37 +            work = Integer.parseInt(args[2]);
38 +
39 +        workMask = work - 1;
40          System.out.print("Class: " + klass.getName());
41 <        System.out.println(" stages: " + maxStages);
41 >        System.out.print(" stages: " + maxStages);
42 >        System.out.println(" work: " + work);
43  
44          print = false;
45          System.out.println("Warmup...");
46 <        oneRun(klass, 1, items);
46 >        //        oneRun(klass, 4);
47 >        //
48          Thread.sleep(100);
49 <        oneRun(klass, 1, items);
49 >        oneRun(klass, 1);
50          Thread.sleep(100);
51          print = true;
52  
53 <        for (int i = 1; i <= maxStages; i += (i+1) >>> 1) {
54 <            oneRun(klass, i, items);
53 >        int k = 1;
54 >        for (int i = 1; i <= maxStages;) {
55 >            oneRun(klass, i);
56 >            if (i == k) {
57 >                k = i << 1;
58 >                i = i + (i >>> 1);
59 >            }
60 >            else
61 >                i = k;
62          }
63          pool.shutdown();
64     }
65  
66 <    static class Stage implements Callable<Integer> {
66 >    static final class Stage implements Callable<Integer> {
67          final Queue<Integer> queue;
68          final CyclicBarrier barrier;
69 <        int items;
70 <        Stage (Queue<Integer> q, CyclicBarrier b, int items) {
69 >        final int nthreads;
70 >        Stage (Queue<Integer> q, CyclicBarrier b, int nthreads) {
71              queue = q;
72              barrier = b;
73 <            this.items = items;
73 >            this.nthreads = nthreads;
74 >        }
75 >
76 >        static int compute(int l) {
77 >            if (l == 0)
78 >                return (int)System.nanoTime();
79 >            int nn =  (l >>> 7) & workMask;
80 >            while (nn-- > 0)
81 >                l = LoopHelpers.compute6(l);
82 >            return l;
83          }
84  
85          public Integer call() {
69 –            // Repeatedly take something from queue if possible,
70 –            // transform it, and put back in.
86              try {
87                  barrier.await();
88 <                int l = (int)System.nanoTime();
88 >                long now = System.nanoTime();
89 >                long stopTime = now + RUN_TIME_NANOS;
90 >                int l = (int)now;
91                  int takes = 0;
92 <                int seq = l;
92 >                int misses = 0;
93 >                int lmask = 1;
94                  for (;;) {
95 +                    l = compute(l);
96                      Integer item = queue.poll();
97                      if (item != null) {
98                          ++takes;
99 <                        l = LoopHelpers.compute2(item.intValue());
99 >                        if (item == one)
100 >                            l = LoopHelpers.compute6(l);
101 >                    } else if ((misses++ & 255) == 0 &&
102 >                               System.nanoTime() >= stopTime) {
103 >                        return new Integer(takes);
104 >                    } else {
105 >                        for (int i = 0; i < BATCH_SIZE; ++i) {
106 >                            queue.offer(((l & lmask)== 0)? zero : one);
107 >                            if ((lmask <<= 1) == 0) lmask = 1;
108 >                            if (i != 0) l = compute(l);
109 >                        }
110                      }
82 –                    else if (takes != 0) {
83 –                        totalItems.getAndAdd(-takes);
84 –                        takes = 0;
85 –                    }
86 –                    else if (totalItems.get() <= 0)
87 –                        break;
88 –                    l = LoopHelpers.compute1(l);
89 –                    if (items > 0) {
90 –                        --items;
91 –                        while (!queue.offer(new Integer(l^seq++))) ;
92 –                    }
93 –                    else if ( (l & (3 << 5)) == 0) // spinwait
94 –                        Thread.sleep(1);
111                  }
96 –                return new Integer(l);
112              }
113              catch (Exception ie) {
114                  ie.printStackTrace();
# Line 102 | Line 117 | public class ConcurrentQueueLoops {
117          }
118      }
119  
120 <    static void oneRun(Class klass, int n, int items) throws Exception {
120 >    static void oneRun(Class klass, int n) throws Exception {
121          Queue<Integer> q = (Queue<Integer>)klass.newInstance();
122          LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
123          CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
109 –        totalItems = new AtomicInteger(n * items);
124          ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);
125          for (int i = 0; i < n; ++i)
126 <            results.add(pool.submit(new Stage(q, barrier, items)));
126 >            results.add(pool.submit(new Stage(q, barrier, n)));
127  
128          if (print)
129              System.out.print("Threads: " + n + "\t:");
# Line 122 | Line 136 | public class ConcurrentQueueLoops {
136          }
137          long endTime = System.nanoTime();
138          long time = endTime - timer.startTime;
139 <        if (print)
126 <            System.out.println(LoopHelpers.rightJustify(time / (items * n)) + " ns per item");
127 <        if (total == 0) // avoid overoptimization
128 <            System.out.println("useless result: " + total);
139 >        long ips = 1000000000L * total / time;
140          
141 +        if (print)
142 +            System.out.print(LoopHelpers.rightJustify(ips) + " items per sec");
143 +        if (print)
144 +            System.out.println();
145      }
146 +
147   }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines