17 *
18 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
19 * or visit www.oracle.com if you need additional information or have any
20 * 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 CancelledProducerConsumerLoops.java
38 * @run main/timeout=7000 CancelledProducerConsumerLoops
39 * @summary Checks for responsiveness of blocking queues to cancellation.
40 * Runs under the assumption that ITERS computations require more than
41 * TIMEOUT msecs to complete.
42 */
43
44 import java.util.concurrent.*;
45
46 public class CancelledProducerConsumerLoops {
47 static final int CAPACITY = 100;
48 static final long TIMEOUT = 100;
49
50 static final ExecutorService pool = Executors.newCachedThreadPool();
51 static boolean print = false;
52
53 public static void main(String[] args) throws Exception {
54 int maxPairs = 8;
55 int iters = 1000000;
56
57 if (args.length > 0)
102
103 if (!tooLate) {
104 for (int i = 1; i < npairs; ++i) {
105 if (!prods[i].isDone() || !prods[i].isCancelled())
106 throw new Error("Only one producer thread should complete");
107 if (!cons[i].isDone() || !cons[i].isCancelled())
108 throw new Error("Only one consumer thread should complete");
109 }
110 }
111 else
112 System.out.print("(cancelled too late) ");
113
114 long endTime = System.nanoTime();
115 long time = endTime - timer.startTime;
116 if (print) {
117 double secs = (double)(time) / 1000000000.0;
118 System.out.println("\t " + secs + "s run time");
119 }
120 }
121
122 static final class LTQasSQ<T> extends LinkedTransferQueue<T> {
123 LTQasSQ() { super(); }
124 public void put(T x) {
125 try { super.transfer(x); }
126 catch (InterruptedException ex) { throw new Error(); }
127 }
128 private final static long serialVersionUID = 42;
129 }
130
131 static final class HalfSyncLTQ<T> extends LinkedTransferQueue<T> {
132 HalfSyncLTQ() { super(); }
133 public void put(T x) {
134 if (ThreadLocalRandom.current().nextBoolean())
135 super.put(x);
136 else {
137 try { super.transfer(x); }
138 catch (InterruptedException ex) { throw new Error(); }
139 }
140 }
141 private final static long serialVersionUID = 42;
142 }
143
144 static void oneTest(int pairs, int iters) throws Exception {
145
146 oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
147 oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
148 oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
149 oneRun(new SynchronousQueue<Integer>(), pairs, iters / 8);
150
151 /* TODO: unbounded queue implementations are prone to OOME
152 oneRun(new PriorityBlockingQueue<Integer>(iters / 2 * pairs), pairs, iters / 4);
153 oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
154 oneRun(new LTQasSQ<Integer>(), pairs, iters);
155 oneRun(new HalfSyncLTQ<Integer>(), pairs, iters);
156 */
157 }
158
159 static abstract class Stage implements Callable<Integer> {
160 final BlockingQueue<Integer> queue;
161 final CyclicBarrier barrier;
162 final int iters;
163 Stage (BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
164 queue = q;
165 barrier = b;
166 this.iters = iters;
167 }
168 }
169
170 static class Producer extends Stage {
171 Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
172 super(q, b, iters);
173 }
174
175 public Integer call() throws Exception {
176 barrier.await();
177 int s = 0;
178 int l = 4321;
179 for (int i = 0; i < iters; ++i) {
180 l = LoopHelpers.compute1(l);
181 s += LoopHelpers.compute2(l);
182 if (!queue.offer(new Integer(l), 1, TimeUnit.SECONDS))
183 break;
|
17 *
18 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
19 * or visit www.oracle.com if you need additional information or have any
20 * 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 CancelledProducerConsumerLoops.java
38 * @run main/timeout=7000 CancelledProducerConsumerLoops
39 * @summary Checks for responsiveness of blocking queues to cancellation.
40 * Runs under the assumption that ITERS computations require more than
41 * TIMEOUT msecs to complete.
42 */
43
44 import java.util.concurrent.*;
45
46 public class CancelledProducerConsumerLoops {
47 static final int CAPACITY = 100;
48 static final long TIMEOUT = 100;
49
50 static final ExecutorService pool = Executors.newCachedThreadPool();
51 static boolean print = false;
52
53 public static void main(String[] args) throws Exception {
54 int maxPairs = 8;
55 int iters = 1000000;
56
57 if (args.length > 0)
102
103 if (!tooLate) {
104 for (int i = 1; i < npairs; ++i) {
105 if (!prods[i].isDone() || !prods[i].isCancelled())
106 throw new Error("Only one producer thread should complete");
107 if (!cons[i].isDone() || !cons[i].isCancelled())
108 throw new Error("Only one consumer thread should complete");
109 }
110 }
111 else
112 System.out.print("(cancelled too late) ");
113
114 long endTime = System.nanoTime();
115 long time = endTime - timer.startTime;
116 if (print) {
117 double secs = (double)(time) / 1000000000.0;
118 System.out.println("\t " + secs + "s run time");
119 }
120 }
121
122 static void oneTest(int pairs, int iters) throws Exception {
123
124 oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), pairs, iters);
125 oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), pairs, iters);
126 oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), pairs, iters);
127 oneRun(new LinkedTransferQueue<Integer>(), pairs, iters);
128 oneRun(new SynchronousQueue<Integer>(), pairs, iters / 8);
129
130 /* PriorityBlockingQueue is unbounded
131 oneRun(new PriorityBlockingQueue<Integer>(iters / 2 * pairs), pairs, iters / 4);
132 */
133 }
134
135 abstract static class Stage implements Callable<Integer> {
136 final BlockingQueue<Integer> queue;
137 final CyclicBarrier barrier;
138 final int iters;
139 Stage(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
140 queue = q;
141 barrier = b;
142 this.iters = iters;
143 }
144 }
145
146 static class Producer extends Stage {
147 Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
148 super(q, b, iters);
149 }
150
151 public Integer call() throws Exception {
152 barrier.await();
153 int s = 0;
154 int l = 4321;
155 for (int i = 0; i < iters; ++i) {
156 l = LoopHelpers.compute1(l);
157 s += LoopHelpers.compute2(l);
158 if (!queue.offer(new Integer(l), 1, TimeUnit.SECONDS))
159 break;
|