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

Comparing jsr166/src/test/loops/ProducerConsumerLoops.java (file contents):
Revision 1.2 by dl, Mon Feb 19 00:46:06 2007 UTC vs.
Revision 1.3 by dl, Fri Oct 23 19:57:06 2009 UTC

# Line 4 | Line 4
4   * http://creativecommons.org/licenses/publicdomain
5   */
6  
7 + import java.util.*;
8   import java.util.concurrent.*;
9 + //import jsr166y.*;
10  
11   public class ProducerConsumerLoops {
12 <    static final int CAPACITY =      100;
13 <
12 >    static final int NCPUS = Runtime.getRuntime().availableProcessors();
13 >    static final Random rng = new Random();
14      static final ExecutorService pool = Executors.newCachedThreadPool();
15      static boolean print = false;
16      static int producerSum;
# Line 26 | Line 28 | public class ProducerConsumerLoops {
28              throw new Error("CheckSum mismatch");
29      }
30  
31 +    // Number of elements passed around -- must be power of two
32 +    // Elements are reused from pool to minimize alloc impact
33 +    static final int POOL_SIZE = 1 << 7;
34 +    static final int POOL_MASK = POOL_SIZE-1;
35 +    static final Integer[] intPool = new Integer[POOL_SIZE];
36 +    static {
37 +        for (int i = 0; i < POOL_SIZE; ++i)
38 +            intPool[i] = Integer.valueOf(i);
39 +    }
40 +
41 +    // Number of puts by producers or takes by consumers
42 +    static final int ITERS = 1 << 20;
43 +
44 +    // max lag between a producer and consumer to avoid
45 +    // this becoming a GC test rather than queue test.
46 +    // Used only per-pair to lessen impact on queue sync
47 +    static final int LAG_MASK = (1 << 12) - 1;
48 +
49      public static void main(String[] args) throws Exception {
50 <        int maxPairs = 100;
31 <        int iters = 100000;
50 >        int maxPairs = NCPUS * 3 / 2;
51  
52          if (args.length > 0)
53              maxPairs = Integer.parseInt(args[0]);
54  
55 <        print = false;
37 <        System.out.println("Warmup...");
38 <        oneTest(1, 10000);
39 <        Thread.sleep(100);
40 <        oneTest(2, 10000);
41 <        Thread.sleep(100);
42 <        oneTest(2, 10000);
43 <        Thread.sleep(100);
55 >        warmup();
56          print = true;
45 –        
57          int k = 1;
58          for (int i = 1; i <= maxPairs;) {
59              System.out.println("Pairs:" + i);
60 <            oneTest(i, iters);
50 <            Thread.sleep(100);
60 >            oneTest(i, ITERS);
61              if (i == k) {
62                  k = i << 1;
63                  i = i + (i >>> 1);
# Line 56 | Line 66 | public class ProducerConsumerLoops {
66                  i = k;
67          }
68          pool.shutdown();
69 <   }
69 >    }
70  
71 <    static void oneTest(int pairs, int iters) throws Exception {
72 <        int fairIters = iters/20;
71 >    static void warmup() throws Exception {
72 >        print = false;
73 >        System.out.print("Warmup ");
74 >        int it = 2000;
75 >        for (int j = 5; j > 0; --j) {
76 >            oneTest(j, it);
77 >            System.out.print(".");
78 >            it += 1000;
79 >        }
80 >        System.gc();
81 >        it = 20000;
82 >        for (int j = 5; j > 0; --j) {
83 >            oneTest(j, it);
84 >            System.out.print(".");
85 >            it += 10000;
86 >        }
87 >        System.gc();
88 >        System.out.println();
89 >    }
90 >
91 >    static void oneTest(int n, int iters) throws Exception {
92 >        int fairIters = iters/16;
93 >
94 >        Thread.sleep(100); // System.gc();
95          if (print)
96 <            System.out.print("ArrayBlockingQueue      ");
97 <        oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
96 >            System.out.print("LinkedTransferQueue     ");
97 >        oneRun(new LinkedTransferQueue<Integer>(), n, iters);
98  
99 +        Thread.sleep(100); // System.gc();
100          if (print)
101              System.out.print("LinkedBlockingQueue     ");
102 <        oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
102 >        oneRun(new LinkedBlockingQueue<Integer>(), n, iters);
103 >
104 >        Thread.sleep(100); // System.gc();
105 >        if (print)
106 >            System.out.print("LinkedBlockingQueue(cap)");
107 >        oneRun(new LinkedBlockingQueue<Integer>(POOL_SIZE), n, iters);
108  
109 +        Thread.sleep(100); // System.gc();
110          if (print)
111              System.out.print("LinkedBlockingDeque     ");
112 <        oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
112 >        oneRun(new LinkedBlockingDeque<Integer>(), n, iters);
113 >        
114 >        Thread.sleep(100); // System.gc();
115 >        if (print)
116 >            System.out.print("ArrayBlockingQueue      ");
117 >        oneRun(new ArrayBlockingQueue<Integer>(POOL_SIZE), n, iters);
118  
119 +        Thread.sleep(100); // System.gc();
120          if (print)
121              System.out.print("SynchronousQueue        ");
122 <        oneRun(new SynchronousQueue<Integer>(), pairs, iters);
122 >        oneRun(new SynchronousQueue<Integer>(), n, iters);
123  
124 +        
125 +        Thread.sleep(100); // System.gc();
126          if (print)
127              System.out.print("SynchronousQueue(fair)  ");
128 <        oneRun(new SynchronousQueue<Integer>(true), pairs, fairIters);
128 >        oneRun(new SynchronousQueue<Integer>(true), n, iters);
129 >
130 >        Thread.sleep(100); // System.gc();
131 >        if (print)
132 >            System.out.print("LinkedTransferQueue(xfer)");
133 >        oneRun(new LTQasSQ<Integer>(), n, iters);
134  
135 +        Thread.sleep(100); // System.gc();
136 +        if (print)
137 +            System.out.print("LinkedTransferQueue(half)");
138 +        oneRun(new HalfSyncLTQ<Integer>(), n, iters);
139 +        
140 +        Thread.sleep(100); // System.gc();
141          if (print)
142              System.out.print("PriorityBlockingQueue   ");
143 <        oneRun(new PriorityBlockingQueue<Integer>(), pairs, fairIters);
143 >        oneRun(new PriorityBlockingQueue<Integer>(), n, fairIters);
144  
145 +        Thread.sleep(100); // System.gc();
146          if (print)
147              System.out.print("ArrayBlockingQueue(fair)");
148 <        oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), pairs, fairIters);
148 >        oneRun(new ArrayBlockingQueue<Integer>(POOL_SIZE, true), n, fairIters);
149  
150      }
151      
# Line 94 | Line 153 | public class ProducerConsumerLoops {
153          final int iters;
154          final BlockingQueue<Integer> queue;
155          final CyclicBarrier barrier;
156 <        Stage (BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
156 >        final Phaser lagPhaser;
157 >        Stage (BlockingQueue<Integer> q, CyclicBarrier b, Phaser s,
158 >               int iters) {
159              queue = q;
160              barrier = b;
161 +            lagPhaser = s;
162              this.iters = iters;
163          }
164      }
165  
166      static class Producer extends Stage {
167 <        Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
168 <            super(q, b, iters);
167 >        Producer(BlockingQueue<Integer> q, CyclicBarrier b, Phaser s,
168 >                 int iters) {
169 >            super(q, b, s, iters);
170          }
171  
172          public void run() {
173              try {
174                  barrier.await();
175 <                int s = 0;
176 <                int l = hashCode();
175 >                int ps = 0;
176 >                int r = hashCode();
177                  for (int i = 0; i < iters; ++i) {
178 <                    l = LoopHelpers.compute4(l);
179 <                    queue.put(new Integer(l));
180 <                    s += LoopHelpers.compute4(l);
178 >                    r = LoopHelpers.compute7(r);
179 >                    Integer v = intPool[r & POOL_MASK];
180 >                    int k = v.intValue();
181 >                    queue.put(v);
182 >                    ps += k;
183 >                    if ((i & LAG_MASK) == LAG_MASK)
184 >                        lagPhaser.arriveAndAwaitAdvance();
185                  }
186 <                addProducerSum(s);
186 >                addProducerSum(ps);
187                  barrier.await();
188              }
189              catch (Exception ie) {
# Line 127 | Line 194 | public class ProducerConsumerLoops {
194      }
195  
196      static class Consumer extends Stage {
197 <        Consumer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
198 <            super(q, b, iters);
197 >        Consumer(BlockingQueue<Integer> q, CyclicBarrier b, Phaser s,
198 >                 int iters) {
199 >            super(q, b, s, iters);
200          }
201  
202          public void run() {
203              try {
204                  barrier.await();
205 <                int l = 0;
138 <                int s = 0;
205 >                int cs = 0;
206                  for (int i = 0; i < iters; ++i) {
207 <                    l = LoopHelpers.compute4(queue.take().intValue());
208 <                    s += l;
207 >                    Integer v = queue.take();
208 >                    int k = v.intValue();
209 >                    cs += k;
210 >                    if ((i & LAG_MASK) == LAG_MASK)
211 >                        lagPhaser.arriveAndAwaitAdvance();
212                  }
213 <                addConsumerSum(s);
213 >                addConsumerSum(cs);
214                  barrier.await();
215              }
216              catch (Exception ie) {
# Line 151 | Line 221 | public class ProducerConsumerLoops {
221  
222      }
223  
224 <    static void oneRun(BlockingQueue<Integer> q, int npairs, int iters) throws Exception {
224 >    static void oneRun(BlockingQueue<Integer> q, int n, int iters) throws Exception {
225          LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
226 <        CyclicBarrier barrier = new CyclicBarrier(npairs * 2 + 1, timer);
227 <        for (int i = 0; i < npairs; ++i) {
228 <            pool.execute(new Producer(q, barrier, iters));
229 <            pool.execute(new Consumer(q, barrier, iters));
226 >        CyclicBarrier barrier = new CyclicBarrier(n * 2 + 1, timer);
227 >        for (int i = 0; i < n; ++i) {
228 >            Phaser s = new Phaser(2);
229 >            pool.execute(new Producer(q, barrier, s, iters));
230 >            pool.execute(new Consumer(q, barrier, s, iters));
231          }
232          barrier.await();
233          barrier.await();
234          long time = timer.getTime();
235          checkSum();
236          if (print)
237 <            System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * npairs)) + " ns per transfer");
237 >            System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * n)) + " ns per transfer");
238 >    }
239 >
240 >    static final class LTQasSQ<T> extends LinkedTransferQueue<T> {
241 >        LTQasSQ() { super(); }
242 >        public void put(T x) {
243 >            try { super.transfer(x);
244 >            } catch (InterruptedException ex) { throw new Error(); }
245 >        }
246      }
247  
248 +    static final class HalfSyncLTQ<T> extends LinkedTransferQueue<T> {
249 +        int calls;
250 +        HalfSyncLTQ() { super(); }
251 +        public void put(T x) {
252 +            if ((++calls & 1) == 0)
253 +                super.put(x);
254 +            else {
255 +                try { super.transfer(x);
256 +                } catch (InterruptedException ex) {
257 +                    throw new Error();
258 +                }
259 +            }
260 +        }
261 +    }
262   }

Diff Legend

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