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.3 by dl, Fri Oct 7 14:58:09 2005 UTC

# Line 13 | Line 13
13  
14   import java.util.*;
15   import java.util.concurrent.*;
16 + import java.util.concurrent.locks.*;
17   import java.util.concurrent.atomic.*;
18  
19   public class ConcurrentQueueLoops {
20      static final ExecutorService pool = Executors.newCachedThreadPool();
20 –    static AtomicInteger totalItems;
21      static boolean print = false;
22 +    static final Integer zero = new Integer(0);
23 +    static final Integer one = new Integer(1);
24 +    static int workMask;
25 +    static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;
26 +    static final int BATCH_SIZE = 8;
27  
28      public static void main(String[] args) throws Exception {
29 <        int maxStages = 8;
30 <        int items = 100000;
26 <
29 >        int maxStages = 100;
30 >        int work = 1024;
31          Class klass = null;
32          if (args.length > 0) {
33              try {
# Line 32 | Line 36 | public class ConcurrentQueueLoops {
36                  throw new RuntimeException("Class " + args[0] + " not found.");
37              }
38          }
35 –        else
36 –            klass = java.util.concurrent.ConcurrentLinkedQueue.class;
39  
40          if (args.length > 1)
41              maxStages = Integer.parseInt(args[1]);
42  
43 +        if (args.length > 2)
44 +            work = Integer.parseInt(args[2]);
45 +
46 +        workMask = work - 1;
47          System.out.print("Class: " + klass.getName());
48 <        System.out.println(" stages: " + maxStages);
48 >        System.out.print(" stages: " + maxStages);
49 >        System.out.println(" work: " + work);
50  
51          print = false;
52          System.out.println("Warmup...");
53 <        oneRun(klass, 1, items);
53 >        oneRun(klass, 4);
54          Thread.sleep(100);
55 <        oneRun(klass, 1, items);
55 >        oneRun(klass, 1);
56          Thread.sleep(100);
57          print = true;
58  
59 <        for (int i = 1; i <= maxStages; i += (i+1) >>> 1) {
60 <            oneRun(klass, i, items);
59 >        int k = 1;
60 >        for (int i = 1; i <= maxStages;) {
61 >            oneRun(klass, i);
62 >            if (i == k) {
63 >                k = i << 1;
64 >                i = i + (i >>> 1);
65 >            }
66 >            else
67 >                i = k;
68          }
69          pool.shutdown();
70     }
71  
72 <    static class Stage implements Callable<Integer> {
72 >    static final class Stage implements Callable<Integer> {
73          final Queue<Integer> queue;
74          final CyclicBarrier barrier;
75 <        int items;
76 <        Stage (Queue<Integer> q, CyclicBarrier b, int items) {
75 >        final int nthreads;
76 >        Stage (Queue<Integer> q, CyclicBarrier b, int nthreads) {
77              queue = q;
78              barrier = b;
79 <            this.items = items;
79 >            this.nthreads = nthreads;
80 >        }
81 >
82 >        static int compute(int l) {
83 >            if (l == 0)
84 >                return (int)System.nanoTime();
85 >            int nn =  (l >>> 7) & workMask;
86 >            while (nn-- > 0)
87 >                l = LoopHelpers.compute6(l);
88 >            return l;
89          }
90  
91          public Integer call() {
69 –            // Repeatedly take something from queue if possible,
70 –            // transform it, and put back in.
92              try {
93                  barrier.await();
94 <                int l = (int)System.nanoTime();
94 >                long now = System.nanoTime();
95 >                long stopTime = now + RUN_TIME_NANOS;
96 >                int l = (int)now;
97                  int takes = 0;
98 <                int seq = l;
98 >                int misses = 0;
99 >                int lmask = 1;
100                  for (;;) {
101 +                    l = compute(l);
102                      Integer item = queue.poll();
103                      if (item != null) {
104                          ++takes;
105 <                        l = LoopHelpers.compute2(item.intValue());
106 <                    }
107 <                    else if (takes != 0) {
108 <                        totalItems.getAndAdd(-takes);
109 <                        takes = 0;
105 >                        if (item == one)
106 >                            l = LoopHelpers.compute6(l);
107 >                    } else if ((misses++ & 255) == 0 &&
108 >                               System.nanoTime() >= stopTime) {
109 >                        return new Integer(takes);
110 >                    } else {
111 >                        for (int i = 0; i < BATCH_SIZE; ++i) {
112 >                            queue.offer(((l & lmask)== 0)? zero : one);
113 >                            if ((lmask <<= 1) == 0) lmask = 1;
114 >                            if (i != 0) l = compute(l);
115 >                        }
116                      }
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);
117                  }
96 –                return new Integer(l);
118              }
119              catch (Exception ie) {
120                  ie.printStackTrace();
# Line 102 | Line 123 | public class ConcurrentQueueLoops {
123          }
124      }
125  
126 <    static void oneRun(Class klass, int n, int items) throws Exception {
126 >    static void oneRun(Class klass, int n) throws Exception {
127          Queue<Integer> q = (Queue<Integer>)klass.newInstance();
128          LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
129          CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
109 –        totalItems = new AtomicInteger(n * items);
130          ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);
131          for (int i = 0; i < n; ++i)
132 <            results.add(pool.submit(new Stage(q, barrier, items)));
132 >            results.add(pool.submit(new Stage(q, barrier, n)));
133  
134          if (print)
135              System.out.print("Threads: " + n + "\t:");
# Line 122 | Line 142 | public class ConcurrentQueueLoops {
142          }
143          long endTime = System.nanoTime();
144          long time = endTime - timer.startTime;
145 <        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);
145 >        long ips = 1000000000L * total / time;
146          
147 +        if (print)
148 +            System.out.print(LoopHelpers.rightJustify(ips) + " items per sec");
149 +        if (print)
150 +            System.out.println();
151      }
152 +
153   }

Diff Legend

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