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

Comparing jsr166/src/test/loops/TimeoutProducerConsumerLoops.java (file contents):
Revision 1.4 by jsr166, Thu Oct 29 23:09:08 2009 UTC vs.
Revision 1.15 by jsr166, Sat Dec 31 21:34:47 2016 UTC

# Line 1 | Line 1
1   /*
2   * Written by Doug Lea with assistance from members of JCP JSR-166
3   * Expert Group and released to the public domain, as explained at
4 < * http://creativecommons.org/licenses/publicdomain
4 > * http://creativecommons.org/publicdomain/zero/1.0/
5   */
6  
7 < import java.util.concurrent.*;
8 < //import jsr166y.*;
7 > import java.util.concurrent.ArrayBlockingQueue;
8 > import java.util.concurrent.CyclicBarrier;
9 > import java.util.concurrent.BlockingQueue;
10 > import java.util.concurrent.ExecutorService;
11 > import java.util.concurrent.Executors;
12 > import java.util.concurrent.LinkedBlockingDeque;
13 > import java.util.concurrent.LinkedBlockingQueue;
14 > import java.util.concurrent.LinkedTransferQueue;
15 > import java.util.concurrent.Phaser;
16 > import java.util.concurrent.PriorityBlockingQueue;
17 > import java.util.concurrent.SynchronousQueue;
18 > import java.util.concurrent.TimeUnit;
19  
20   public class TimeoutProducerConsumerLoops {
21      static final int NCPUS = Runtime.getRuntime().availableProcessors();
# Line 21 | Line 31 | public class TimeoutProducerConsumerLoop
31              intPool[i] = Integer.valueOf(i);
32      }
33  
34 +    // max lag between a producer and consumer to avoid
35 +    // this becoming a GC test rather than queue test.
36 +    // Used only per-pair to lessen impact on queue sync
37 +    static final int LAG_MASK = (1 << 12) - 1;
38  
39      static boolean print = false;
40      static int producerSum;
# Line 47 | Line 61 | public class TimeoutProducerConsumerLoop
61              maxPairs = Integer.parseInt(args[0]);
62  
63          print = true;
64 <        int k = 1;
51 <        for (int i = 1; i <= maxPairs;) {
64 >        for (int k = 1, i = 1; i <= maxPairs;) {
65              System.out.println("Pairs:" + i);
66              oneTest(i, iters);
67              Thread.sleep(100);
# Line 60 | Line 73 | public class TimeoutProducerConsumerLoop
73                  i = k;
74          }
75          pool.shutdown();
76 <   }
76 >    }
77  
78      static void oneTest(int n, int iters) throws Exception {
79          if (print)
# Line 102 | Line 115 | public class TimeoutProducerConsumerLoop
115          if (print)
116              System.out.print("ArrayBlockingQueue(fair) ");
117          oneRun(new ArrayBlockingQueue<Integer>(POOL_SIZE, true), n, iters/16);
105 –
118      }
119  
120 <    static abstract class Stage implements Runnable {
120 >    abstract static class Stage implements Runnable {
121          final int iters;
122          final BlockingQueue<Integer> queue;
123          final CyclicBarrier barrier;
124 <        Stage (BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
124 >        final Phaser lagPhaser;
125 >        Stage(BlockingQueue<Integer> q, CyclicBarrier b, Phaser s, int iters) {
126              queue = q;
127              barrier = b;
128 +            lagPhaser = s;
129              this.iters = iters;
130          }
131      }
132  
133      static class Producer extends Stage {
134 <        Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
135 <            super(q, b, iters);
134 >        Producer(BlockingQueue<Integer> q, CyclicBarrier b, Phaser s, int iters) {
135 >            super(q, b, s, iters);
136          }
137  
138          public void run() {
# Line 136 | Line 150 | public class TimeoutProducerConsumerLoop
150                          ++i;
151                          if (timeout > 1)
152                              timeout--;
153 +                        if ((i & LAG_MASK) == LAG_MASK)
154 +                            lagPhaser.arriveAndAwaitAdvance();
155                      }
156                      else
157                          timeout++;
# Line 151 | Line 167 | public class TimeoutProducerConsumerLoop
167      }
168  
169      static class Consumer extends Stage {
170 <        Consumer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
171 <            super(q, b, iters);
170 >        Consumer(BlockingQueue<Integer> q, CyclicBarrier b, Phaser s, int iters) {
171 >            super(q, b, s, iters);
172          }
173  
174          public void run() {
# Line 171 | Line 187 | public class TimeoutProducerConsumerLoop
187                          ++i;
188                          if (timeout > 1)
189                              --timeout;
190 +                        if ((i & LAG_MASK) == LAG_MASK)
191 +                            lagPhaser.arriveAndAwaitAdvance();
192                      }
193                      else
194                          ++timeout;
# Line 190 | Line 208 | public class TimeoutProducerConsumerLoop
208          LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
209          CyclicBarrier barrier = new CyclicBarrier(npairs * 2 + 1, timer);
210          for (int i = 0; i < npairs; ++i) {
211 <            pool.execute(new Producer(q, barrier, iters));
212 <            pool.execute(new Consumer(q, barrier, iters));
211 >            Phaser s = new Phaser(2);
212 >            pool.execute(new Producer(q, barrier, s, iters));
213 >            pool.execute(new Consumer(q, barrier, s, iters));
214          }
215          barrier.await();
216          barrier.await();
# Line 205 | Line 224 | public class TimeoutProducerConsumerLoop
224      static final class LTQasSQ<T> extends LinkedTransferQueue<T> {
225          LTQasSQ() { super(); }
226          public void put(T x) {
227 <            try { super.transfer(x);
228 <            } catch (InterruptedException ex) { throw new Error(); }
227 >            try { super.transfer(x); }
228 >            catch (InterruptedException ex) { throw new Error(ex); }
229          }
230  
231          public boolean offer(T x, long timeout, TimeUnit unit) {

Diff Legend

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