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

Comparing jsr166/src/test/tck/SynchronousQueueTest.java (file contents):
Revision 1.11 by jsr166, Mon Nov 16 04:57:10 2009 UTC vs.
Revision 1.58 by jsr166, Sun May 14 04:02:06 2017 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   * Other contributors include Andrew Wright, Jeffrey Hayes,
6   * Pat Fisher, Mike Judd.
7   */
8  
9 < import junit.framework.*;
10 < import java.util.*;
11 < import java.util.concurrent.*;
12 < import java.io.*;
9 > import static java.util.concurrent.TimeUnit.MILLISECONDS;
10 >
11 > import java.util.ArrayList;
12 > import java.util.Arrays;
13 > import java.util.Collection;
14 > import java.util.Iterator;
15 > import java.util.NoSuchElementException;
16 > import java.util.concurrent.BlockingQueue;
17 > import java.util.concurrent.CountDownLatch;
18 > import java.util.concurrent.Executors;
19 > import java.util.concurrent.ExecutorService;
20 > import java.util.concurrent.SynchronousQueue;
21 > import java.util.concurrent.ThreadLocalRandom;
22 >
23 > import junit.framework.Test;
24  
25   public class SynchronousQueueTest extends JSR166TestCase {
26  
27 <    public static void main(String[] args) {
28 <        junit.textui.TestRunner.run (suite());
27 >    public static class Fair extends BlockingQueueTest {
28 >        protected BlockingQueue emptyCollection() {
29 >            return new SynchronousQueue(true);
30 >        }
31      }
32  
33 <    public static Test suite() {
34 <        return new TestSuite(SynchronousQueueTest.class);
33 >    public static class NonFair extends BlockingQueueTest {
34 >        protected BlockingQueue emptyCollection() {
35 >            return new SynchronousQueue(false);
36 >        }
37      }
38  
39 <    /**
40 <     * A SynchronousQueue is both empty and full
41 <     */
42 <    public void testEmptyFull() {
43 <        SynchronousQueue q = new SynchronousQueue();
44 <        assertTrue(q.isEmpty());
45 <        assertEquals(0, q.size());
46 <        assertEquals(0, q.remainingCapacity());
32 <        assertFalse(q.offer(zero));
39 >    public static void main(String[] args) {
40 >        main(suite(), args);
41 >    }
42 >
43 >    public static Test suite() {
44 >        return newTestSuite(SynchronousQueueTest.class,
45 >                            new Fair().testSuite(),
46 >                            new NonFair().testSuite());
47      }
48  
49      /**
50 <     * A fair SynchronousQueue is both empty and full
50 >     * Any SynchronousQueue is both empty and full
51       */
52 <    public void testFairEmptyFull() {
53 <        SynchronousQueue q = new SynchronousQueue(true);
52 >    public void testEmptyFull()      { testEmptyFull(false); }
53 >    public void testEmptyFull_fair() { testEmptyFull(true); }
54 >    public void testEmptyFull(boolean fair) {
55 >        final SynchronousQueue q = new SynchronousQueue(fair);
56          assertTrue(q.isEmpty());
57 <        assertEquals(0, q.size());
57 >        assertEquals(0, q.size());
58          assertEquals(0, q.remainingCapacity());
59          assertFalse(q.offer(zero));
60      }
61  
62      /**
47     * offer(null) throws NPE
48     */
49    public void testOfferNull() {
50        try {
51            SynchronousQueue q = new SynchronousQueue();
52            q.offer(null);
53            shouldThrow();
54        } catch (NullPointerException success) { }
55    }
56
57    /**
58     * add(null) throws NPE
59     */
60    public void testAddNull() {
61        try {
62            SynchronousQueue q = new SynchronousQueue();
63            q.add(null);
64            shouldThrow();
65        } catch (NullPointerException success) { }
66    }
67
68    /**
63       * offer fails if no active taker
64       */
65 <    public void testOffer() {
66 <        SynchronousQueue q = new SynchronousQueue();
65 >    public void testOffer()      { testOffer(false); }
66 >    public void testOffer_fair() { testOffer(true); }
67 >    public void testOffer(boolean fair) {
68 >        SynchronousQueue q = new SynchronousQueue(fair);
69          assertFalse(q.offer(one));
70      }
71  
72      /**
73 <     * add throws ISE if no active taker
73 >     * add throws IllegalStateException if no active taker
74       */
75 <    public void testAdd() {
76 <        try {
77 <            SynchronousQueue q = new SynchronousQueue();
78 <            assertEquals(0, q.remainingCapacity());
79 <            q.add(one);
84 <            shouldThrow();
85 <        } catch (IllegalStateException success){
86 <        }
87 <    }
88 <
89 <    /**
90 <     * addAll(null) throws NPE
91 <     */
92 <    public void testAddAll1() {
75 >    public void testAdd()      { testAdd(false); }
76 >    public void testAdd_fair() { testAdd(true); }
77 >    public void testAdd(boolean fair) {
78 >        SynchronousQueue q = new SynchronousQueue(fair);
79 >        assertEquals(0, q.remainingCapacity());
80          try {
81 <            SynchronousQueue q = new SynchronousQueue();
95 <            q.addAll(null);
81 >            q.add(one);
82              shouldThrow();
83 <        }
98 <        catch (NullPointerException success) {}
83 >        } catch (IllegalStateException success) {}
84      }
85  
86      /**
87 <     * addAll(this) throws IAE
87 >     * addAll(this) throws IllegalArgumentException
88       */
89 <    public void testAddAllSelf() {
89 >    public void testAddAll_self()      { testAddAll_self(false); }
90 >    public void testAddAll_self_fair() { testAddAll_self(true); }
91 >    public void testAddAll_self(boolean fair) {
92 >        SynchronousQueue q = new SynchronousQueue(fair);
93          try {
106            SynchronousQueue q = new SynchronousQueue();
94              q.addAll(q);
95              shouldThrow();
96 <        }
110 <        catch (IllegalArgumentException success) {}
96 >        } catch (IllegalArgumentException success) {}
97      }
98  
99      /**
114     * addAll of a collection with null elements throws NPE
115     */
116    public void testAddAll2() {
117        try {
118            SynchronousQueue q = new SynchronousQueue();
119            Integer[] ints = new Integer[1];
120            q.addAll(Arrays.asList(ints));
121            shouldThrow();
122        }
123        catch (NullPointerException success) {}
124    }
125    /**
100       * addAll throws ISE if no active taker
101       */
102 <    public void testAddAll4() {
102 >    public void testAddAll_ISE()      { testAddAll_ISE(false); }
103 >    public void testAddAll_ISE_fair() { testAddAll_ISE(true); }
104 >    public void testAddAll_ISE(boolean fair) {
105 >        SynchronousQueue q = new SynchronousQueue(fair);
106 >        Integer[] ints = new Integer[1];
107 >        for (int i = 0; i < ints.length; i++)
108 >            ints[i] = i;
109 >        Collection<Integer> coll = Arrays.asList(ints);
110          try {
111 <            SynchronousQueue q = new SynchronousQueue();
131 <            Integer[] ints = new Integer[1];
132 <            for (int i = 0; i < 1; ++i)
133 <                ints[i] = new Integer(i);
134 <            q.addAll(Arrays.asList(ints));
111 >            q.addAll(coll);
112              shouldThrow();
113 <        }
137 <        catch (IllegalStateException success) {}
113 >        } catch (IllegalStateException success) {}
114      }
115  
116      /**
141     * put(null) throws NPE
142     */
143    public void testPutNull() {
144        try {
145            SynchronousQueue q = new SynchronousQueue();
146            q.put(null);
147            shouldThrow();
148        }
149        catch (NullPointerException success){
150        }
151        catch (InterruptedException ie) {
152            unexpectedException();
153        }
154     }
155
156    /**
117       * put blocks interruptibly if no active taker
118       */
119 <    public void testBlockingPut() {
120 <        Thread t = new Thread(new Runnable() {
121 <                public void run() {
122 <                    try {
123 <                        SynchronousQueue q = new SynchronousQueue();
124 <                        q.put(zero);
125 <                        threadShouldThrow();
126 <                    } catch (InterruptedException ie){
127 <                    }
128 <                }});
129 <        t.start();
130 <        try {
131 <           Thread.sleep(SHORT_DELAY_MS);
172 <           t.interrupt();
173 <           t.join();
174 <        }
175 <        catch (InterruptedException ie) {
176 <            unexpectedException();
177 <        }
178 <    }
119 >    public void testBlockingPut()      { testBlockingPut(false); }
120 >    public void testBlockingPut_fair() { testBlockingPut(true); }
121 >    public void testBlockingPut(boolean fair) {
122 >        final SynchronousQueue q = new SynchronousQueue(fair);
123 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
124 >        Thread t = newStartedThread(new CheckedRunnable() {
125 >            public void realRun() throws InterruptedException {
126 >                Thread.currentThread().interrupt();
127 >                try {
128 >                    q.put(99);
129 >                    shouldThrow();
130 >                } catch (InterruptedException success) {}
131 >                assertFalse(Thread.interrupted());
132  
133 <    /**
134 <     * put blocks waiting for take
135 <     */
136 <    public void testPutWithTake() {
137 <        final SynchronousQueue q = new SynchronousQueue();
138 <        Thread t = new Thread(new Runnable() {
139 <                public void run() {
140 <                    int added = 0;
141 <                    try {
142 <                        q.put(new Object());
143 <                        ++added;
144 <                        q.put(new Object());
145 <                        ++added;
193 <                        q.put(new Object());
194 <                        ++added;
195 <                        q.put(new Object());
196 <                        ++added;
197 <                        threadShouldThrow();
198 <                    } catch (InterruptedException e){
199 <                        assertTrue(added >= 1);
200 <                    }
201 <                }
202 <            });
203 <        try {
204 <            t.start();
205 <            Thread.sleep(SHORT_DELAY_MS);
206 <            q.take();
207 <            Thread.sleep(SHORT_DELAY_MS);
208 <            t.interrupt();
209 <            t.join();
210 <        } catch (Exception e){
211 <            unexpectedException();
212 <        }
133 >                pleaseInterrupt.countDown();
134 >                try {
135 >                    q.put(99);
136 >                    shouldThrow();
137 >                } catch (InterruptedException success) {}
138 >                assertFalse(Thread.interrupted());
139 >            }});
140 >
141 >        await(pleaseInterrupt);
142 >        assertThreadBlocks(t, Thread.State.WAITING);
143 >        t.interrupt();
144 >        awaitTermination(t);
145 >        assertEquals(0, q.remainingCapacity());
146      }
147  
148      /**
149 <     * timed offer times out if elements not taken
149 >     * put blocks interruptibly waiting for take
150       */
151 <    public void testTimedOffer() {
152 <        final SynchronousQueue q = new SynchronousQueue();
153 <        Thread t = new Thread(new Runnable() {
154 <                public void run() {
155 <                    try {
156 <
157 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
158 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
159 <                        threadShouldThrow();
160 <                    } catch (InterruptedException success){}
228 <                }
229 <            });
151 >    public void testPutWithTake()      { testPutWithTake(false); }
152 >    public void testPutWithTake_fair() { testPutWithTake(true); }
153 >    public void testPutWithTake(boolean fair) {
154 >        final SynchronousQueue q = new SynchronousQueue(fair);
155 >        final CountDownLatch pleaseTake = new CountDownLatch(1);
156 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
157 >        Thread t = newStartedThread(new CheckedRunnable() {
158 >            public void realRun() throws InterruptedException {
159 >                pleaseTake.countDown();
160 >                q.put(one);
161  
162 <        try {
163 <            t.start();
164 <            Thread.sleep(SMALL_DELAY_MS);
165 <            t.interrupt();
166 <            t.join();
167 <        } catch (Exception e){
237 <            unexpectedException();
238 <        }
239 <    }
240 <
241 <
242 <    /**
243 <     * take blocks interruptibly when empty
244 <     */
245 <    public void testTakeFromEmpty() {
246 <        final SynchronousQueue q = new SynchronousQueue();
247 <        Thread t = new Thread(new Runnable() {
248 <                public void run() {
249 <                    try {
250 <                        q.take();
251 <                        threadShouldThrow();
252 <                    } catch (InterruptedException success){ }
253 <                }
254 <            });
255 <        try {
256 <            t.start();
257 <            Thread.sleep(SHORT_DELAY_MS);
258 <            t.interrupt();
259 <            t.join();
260 <        } catch (Exception e){
261 <            unexpectedException();
262 <        }
263 <    }
162 >                Thread.currentThread().interrupt();
163 >                try {
164 >                    q.put(99);
165 >                    shouldThrow();
166 >                } catch (InterruptedException success) {}
167 >                assertFalse(Thread.interrupted());
168  
169 +                pleaseInterrupt.countDown();
170 +                try {
171 +                    q.put(99);
172 +                    shouldThrow();
173 +                } catch (InterruptedException success) {}
174 +                assertFalse(Thread.interrupted());
175 +            }});
176  
177 <    /**
178 <     * put blocks interruptibly if no active taker
179 <     */
180 <    public void testFairBlockingPut() {
270 <        Thread t = new Thread(new Runnable() {
271 <                public void run() {
272 <                    try {
273 <                        SynchronousQueue q = new SynchronousQueue(true);
274 <                        q.put(zero);
275 <                        threadShouldThrow();
276 <                    } catch (InterruptedException ie){
277 <                    }
278 <                }});
279 <        t.start();
280 <        try {
281 <           Thread.sleep(SHORT_DELAY_MS);
282 <           t.interrupt();
283 <           t.join();
284 <        }
285 <        catch (InterruptedException ie) {
286 <            unexpectedException();
287 <        }
288 <    }
177 >        await(pleaseTake);
178 >        assertEquals(0, q.remainingCapacity());
179 >        try { assertSame(one, q.take()); }
180 >        catch (InterruptedException e) { threadUnexpectedException(e); }
181  
182 <    /**
183 <     * put blocks waiting for take
184 <     */
185 <    public void testFairPutWithTake() {
186 <        final SynchronousQueue q = new SynchronousQueue(true);
295 <        Thread t = new Thread(new Runnable() {
296 <                public void run() {
297 <                    int added = 0;
298 <                    try {
299 <                        q.put(new Object());
300 <                        ++added;
301 <                        q.put(new Object());
302 <                        ++added;
303 <                        q.put(new Object());
304 <                        ++added;
305 <                        q.put(new Object());
306 <                        ++added;
307 <                        threadShouldThrow();
308 <                    } catch (InterruptedException e){
309 <                        assertTrue(added >= 1);
310 <                    }
311 <                }
312 <            });
313 <        try {
314 <            t.start();
315 <            Thread.sleep(SHORT_DELAY_MS);
316 <            q.take();
317 <            Thread.sleep(SHORT_DELAY_MS);
318 <            t.interrupt();
319 <            t.join();
320 <        } catch (Exception e){
321 <            unexpectedException();
322 <        }
182 >        await(pleaseInterrupt);
183 >        assertThreadBlocks(t, Thread.State.WAITING);
184 >        t.interrupt();
185 >        awaitTermination(t);
186 >        assertEquals(0, q.remainingCapacity());
187      }
188  
189      /**
190       * timed offer times out if elements not taken
191       */
192 <    public void testFairTimedOffer() {
193 <        final SynchronousQueue q = new SynchronousQueue(true);
194 <        Thread t = new Thread(new Runnable() {
195 <                public void run() {
196 <                    try {
197 <
198 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
199 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
200 <                        threadShouldThrow();
201 <                    } catch (InterruptedException success){}
202 <                }
203 <            });
192 >    public void testTimedOffer() {
193 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
194 >        final SynchronousQueue q = new SynchronousQueue(fair);
195 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
196 >        Thread t = newStartedThread(new CheckedRunnable() {
197 >            public void realRun() throws InterruptedException {
198 >                long startTime = System.nanoTime();
199 >                assertFalse(q.offer(new Object(), timeoutMillis(), MILLISECONDS));
200 >                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
201 >                pleaseInterrupt.countDown();
202 >                try {
203 >                    q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
204 >                    shouldThrow();
205 >                } catch (InterruptedException success) {}
206 >                assertFalse(Thread.interrupted());
207 >            }});
208  
209 <        try {
210 <            t.start();
211 <            Thread.sleep(SMALL_DELAY_MS);
212 <            t.interrupt();
345 <            t.join();
346 <        } catch (Exception e){
347 <            unexpectedException();
348 <        }
209 >        await(pleaseInterrupt);
210 >        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
211 >        t.interrupt();
212 >        awaitTermination(t);
213      }
214  
351
215      /**
216 <     * take blocks interruptibly when empty
216 >     * poll return null if no active putter
217       */
218 <    public void testFairTakeFromEmpty() {
219 <        final SynchronousQueue q = new SynchronousQueue(true);
220 <        Thread t = new Thread(new Runnable() {
221 <                public void run() {
222 <                    try {
360 <                        q.take();
361 <                        threadShouldThrow();
362 <                    } catch (InterruptedException success){ }
363 <                }
364 <            });
365 <        try {
366 <            t.start();
367 <            Thread.sleep(SHORT_DELAY_MS);
368 <            t.interrupt();
369 <            t.join();
370 <        } catch (Exception e){
371 <            unexpectedException();
372 <        }
218 >    public void testPoll()      { testPoll(false); }
219 >    public void testPoll_fair() { testPoll(true); }
220 >    public void testPoll(boolean fair) {
221 >        final SynchronousQueue q = new SynchronousQueue(fair);
222 >        assertNull(q.poll());
223      }
224  
225      /**
226 <     * poll fails unless active taker
226 >     * timed poll with zero timeout times out if no active putter
227       */
228 <    public void testPoll() {
229 <        SynchronousQueue q = new SynchronousQueue();
230 <        assertNull(q.poll());
228 >    public void testTimedPoll0()      { testTimedPoll0(false); }
229 >    public void testTimedPoll0_fair() { testTimedPoll0(true); }
230 >    public void testTimedPoll0(boolean fair) {
231 >        final SynchronousQueue q = new SynchronousQueue(fair);
232 >        try { assertNull(q.poll(0, MILLISECONDS)); }
233 >        catch (InterruptedException e) { threadUnexpectedException(e); }
234      }
235  
236      /**
237 <     * timed pool with zero timeout times out if no active taker
385 <     */
386 <    public void testTimedPoll0() {
387 <        try {
388 <            SynchronousQueue q = new SynchronousQueue();
389 <            assertNull(q.poll(0, TimeUnit.MILLISECONDS));
390 <        } catch (InterruptedException e){
391 <            unexpectedException();
392 <        }
393 <    }
394 <
395 <    /**
396 <     * timed pool with nonzero timeout times out if no active taker
237 >     * timed poll with nonzero timeout times out if no active putter
238       */
239      public void testTimedPoll() {
240 <        try {
241 <            SynchronousQueue q = new SynchronousQueue();
242 <            assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
243 <        } catch (InterruptedException e){
244 <            unexpectedException();
245 <        }
240 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
241 >        final SynchronousQueue q = new SynchronousQueue(fair);
242 >        final long startTime = System.nanoTime();
243 >        try { assertNull(q.poll(timeoutMillis(), MILLISECONDS)); }
244 >        catch (InterruptedException e) { threadUnexpectedException(e); }
245 >        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
246      }
247  
248      /**
249 <     * Interrupted timed poll throws InterruptedException instead of
250 <     * returning timeout status
410 <     */
411 <    public void testInterruptedTimedPoll() {
412 <        Thread t = new Thread(new Runnable() {
413 <                public void run() {
414 <                    try {
415 <                        SynchronousQueue q = new SynchronousQueue();
416 <                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
417 <                    } catch (InterruptedException success){
418 <                    }
419 <                }});
420 <        t.start();
421 <        try {
422 <           Thread.sleep(SHORT_DELAY_MS);
423 <           t.interrupt();
424 <           t.join();
425 <        }
426 <        catch (InterruptedException ie) {
427 <            unexpectedException();
428 <        }
429 <    }
430 <
431 <    /**
432 <     *  timed poll before a delayed offer fails; after offer succeeds;
433 <     *  on interruption throws
249 >     * timed poll before a delayed offer times out, returning null;
250 >     * after offer succeeds; on interruption throws
251       */
252      public void testTimedPollWithOffer() {
253 <        final SynchronousQueue q = new SynchronousQueue();
254 <        Thread t = new Thread(new Runnable() {
255 <                public void run() {
256 <                    try {
257 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
258 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
259 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
260 <                        threadShouldThrow();
261 <                    } catch (InterruptedException success) { }
262 <                }
263 <            });
264 <        try {
265 <            t.start();
449 <            Thread.sleep(SMALL_DELAY_MS);
450 <            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
451 <            t.interrupt();
452 <            t.join();
453 <        } catch (Exception e){
454 <            unexpectedException();
455 <        }
456 <    }
253 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
254 >        final SynchronousQueue q = new SynchronousQueue(fair);
255 >        final CountDownLatch pleaseOffer = new CountDownLatch(1);
256 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
257 >        Thread t = newStartedThread(new CheckedRunnable() {
258 >            public void realRun() throws InterruptedException {
259 >                long startTime = System.nanoTime();
260 >                assertNull(q.poll(timeoutMillis(), MILLISECONDS));
261 >                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
262 >
263 >                pleaseOffer.countDown();
264 >                startTime = System.nanoTime();
265 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
266  
267 <    /**
268 <     * Interrupted timed poll throws InterruptedException instead of
269 <     * returning timeout status
270 <     */
271 <    public void testFairInterruptedTimedPoll() {
272 <        Thread t = new Thread(new Runnable() {
464 <                public void run() {
465 <                    try {
466 <                        SynchronousQueue q = new SynchronousQueue(true);
467 <                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
468 <                    } catch (InterruptedException success){
469 <                    }
470 <                }});
471 <        t.start();
472 <        try {
473 <           Thread.sleep(SHORT_DELAY_MS);
474 <           t.interrupt();
475 <           t.join();
476 <        }
477 <        catch (InterruptedException ie) {
478 <            unexpectedException();
479 <        }
480 <    }
267 >                Thread.currentThread().interrupt();
268 >                try {
269 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
270 >                    shouldThrow();
271 >                } catch (InterruptedException success) {}
272 >                assertFalse(Thread.interrupted());
273  
274 <    /**
275 <     *  timed poll before a delayed offer fails; after offer succeeds;
276 <     *  on interruption throws
277 <     */
278 <    public void testFairTimedPollWithOffer() {
279 <        final SynchronousQueue q = new SynchronousQueue(true);
280 <        Thread t = new Thread(new Runnable() {
281 <                public void run() {
282 <                    try {
491 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
492 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
493 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
494 <                        threadShouldThrow();
495 <                    } catch (InterruptedException success) { }
496 <                }
497 <            });
498 <        try {
499 <            t.start();
500 <            Thread.sleep(SMALL_DELAY_MS);
501 <            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
502 <            t.interrupt();
503 <            t.join();
504 <        } catch (Exception e){
505 <            unexpectedException();
506 <        }
507 <    }
274 >                pleaseInterrupt.countDown();
275 >                try {
276 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
277 >                    shouldThrow();
278 >                } catch (InterruptedException success) {}
279 >                assertFalse(Thread.interrupted());
280 >
281 >                assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
282 >            }});
283  
284 +        await(pleaseOffer);
285 +        long startTime = System.nanoTime();
286 +        try { assertTrue(q.offer(zero, LONG_DELAY_MS, MILLISECONDS)); }
287 +        catch (InterruptedException e) { threadUnexpectedException(e); }
288 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
289 +
290 +        await(pleaseInterrupt);
291 +        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
292 +        t.interrupt();
293 +        awaitTermination(t);
294 +    }
295  
296      /**
297 <     * peek returns null
297 >     * peek() returns null if no active putter
298       */
299 <    public void testPeek() {
300 <        SynchronousQueue q = new SynchronousQueue();
301 <        assertNull(q.peek());
299 >    public void testPeek()      { testPeek(false); }
300 >    public void testPeek_fair() { testPeek(true); }
301 >    public void testPeek(boolean fair) {
302 >        final SynchronousQueue q = new SynchronousQueue(fair);
303 >        assertNull(q.peek());
304      }
305  
306      /**
307 <     * element throws NSEE
307 >     * element() throws NoSuchElementException if no active putter
308       */
309 <    public void testElement() {
310 <        SynchronousQueue q = new SynchronousQueue();
309 >    public void testElement()      { testElement(false); }
310 >    public void testElement_fair() { testElement(true); }
311 >    public void testElement(boolean fair) {
312 >        final SynchronousQueue q = new SynchronousQueue(fair);
313          try {
314              q.element();
315              shouldThrow();
316 <        }
527 <        catch (NoSuchElementException success) {}
316 >        } catch (NoSuchElementException success) {}
317      }
318  
319      /**
320 <     * remove throws NSEE if no active taker
320 >     * remove() throws NoSuchElementException if no active putter
321       */
322 <    public void testRemove() {
323 <        SynchronousQueue q = new SynchronousQueue();
322 >    public void testRemove()      { testRemove(false); }
323 >    public void testRemove_fair() { testRemove(true); }
324 >    public void testRemove(boolean fair) {
325 >        final SynchronousQueue q = new SynchronousQueue(fair);
326          try {
327              q.remove();
328              shouldThrow();
329 <        } catch (NoSuchElementException success){
539 <        }
540 <    }
541 <
542 <    /**
543 <     * remove(x) returns false
544 <     */
545 <    public void testRemoveElement() {
546 <        SynchronousQueue q = new SynchronousQueue();
547 <        assertFalse(q.remove(zero));
548 <        assertTrue(q.isEmpty());
329 >        } catch (NoSuchElementException success) {}
330      }
331  
332      /**
333       * contains returns false
334       */
335 <    public void testContains() {
336 <        SynchronousQueue q = new SynchronousQueue();
335 >    public void testContains()      { testContains(false); }
336 >    public void testContains_fair() { testContains(true); }
337 >    public void testContains(boolean fair) {
338 >        final SynchronousQueue q = new SynchronousQueue(fair);
339          assertFalse(q.contains(zero));
340      }
341  
342      /**
343       * clear ensures isEmpty
344       */
345 <    public void testClear() {
346 <        SynchronousQueue q = new SynchronousQueue();
345 >    public void testClear()      { testClear(false); }
346 >    public void testClear_fair() { testClear(true); }
347 >    public void testClear(boolean fair) {
348 >        final SynchronousQueue q = new SynchronousQueue(fair);
349          q.clear();
350          assertTrue(q.isEmpty());
351      }
# Line 568 | Line 353 | public class SynchronousQueueTest extend
353      /**
354       * containsAll returns false unless empty
355       */
356 <    public void testContainsAll() {
357 <        SynchronousQueue q = new SynchronousQueue();
356 >    public void testContainsAll()      { testContainsAll(false); }
357 >    public void testContainsAll_fair() { testContainsAll(true); }
358 >    public void testContainsAll(boolean fair) {
359 >        final SynchronousQueue q = new SynchronousQueue(fair);
360          Integer[] empty = new Integer[0];
361          assertTrue(q.containsAll(Arrays.asList(empty)));
362          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 579 | Line 366 | public class SynchronousQueueTest extend
366      /**
367       * retainAll returns false
368       */
369 <    public void testRetainAll() {
370 <        SynchronousQueue q = new SynchronousQueue();
369 >    public void testRetainAll()      { testRetainAll(false); }
370 >    public void testRetainAll_fair() { testRetainAll(true); }
371 >    public void testRetainAll(boolean fair) {
372 >        final SynchronousQueue q = new SynchronousQueue(fair);
373          Integer[] empty = new Integer[0];
374          assertFalse(q.retainAll(Arrays.asList(empty)));
375          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 590 | Line 379 | public class SynchronousQueueTest extend
379      /**
380       * removeAll returns false
381       */
382 <    public void testRemoveAll() {
383 <        SynchronousQueue q = new SynchronousQueue();
382 >    public void testRemoveAll()      { testRemoveAll(false); }
383 >    public void testRemoveAll_fair() { testRemoveAll(true); }
384 >    public void testRemoveAll(boolean fair) {
385 >        final SynchronousQueue q = new SynchronousQueue(fair);
386          Integer[] empty = new Integer[0];
387          assertFalse(q.removeAll(Arrays.asList(empty)));
388          Integer[] ints = new Integer[1]; ints[0] = zero;
389          assertFalse(q.containsAll(Arrays.asList(ints)));
390      }
391  
601
392      /**
393       * toArray is empty
394       */
395 <    public void testToArray() {
396 <        SynchronousQueue q = new SynchronousQueue();
397 <        Object[] o = q.toArray();
398 <        assertEquals(o.length, 0);
395 >    public void testToArray()      { testToArray(false); }
396 >    public void testToArray_fair() { testToArray(true); }
397 >    public void testToArray(boolean fair) {
398 >        final SynchronousQueue q = new SynchronousQueue(fair);
399 >        Object[] o = q.toArray();
400 >        assertEquals(0, o.length);
401      }
402  
403      /**
404 <     * toArray(a) is nulled at position 0
405 <     */
406 <    public void testToArray2() {
407 <        SynchronousQueue q = new SynchronousQueue();
408 <        Integer[] ints = new Integer[1];
409 <        assertNull(ints[0]);
404 >     * toArray(Integer array) returns its argument with the first
405 >     * element (if present) nulled out
406 >     */
407 >    public void testToArray2()      { testToArray2(false); }
408 >    public void testToArray2_fair() { testToArray2(true); }
409 >    public void testToArray2(boolean fair) {
410 >        final SynchronousQueue<Integer> q = new SynchronousQueue<>(fair);
411 >        Integer[] a;
412 >
413 >        a = new Integer[0];
414 >        assertSame(a, q.toArray(a));
415 >
416 >        a = new Integer[3];
417 >        Arrays.fill(a, 42);
418 >        assertSame(a, q.toArray(a));
419 >        assertNull(a[0]);
420 >        for (int i = 1; i < a.length; i++)
421 >            assertEquals(42, (int) a[i]);
422      }
423  
424      /**
425       * toArray(null) throws NPE
426       */
427 <    public void testToArray_BadArg() {
428 <        try {
429 <            SynchronousQueue q = new SynchronousQueue();
430 <            Object o[] = q.toArray(null);
431 <            shouldThrow();
432 <        } catch (NullPointerException success){}
427 >    public void testToArray_null()      { testToArray_null(false); }
428 >    public void testToArray_null_fair() { testToArray_null(true); }
429 >    public void testToArray_null(boolean fair) {
430 >        final SynchronousQueue q = new SynchronousQueue(fair);
431 >        try {
432 >            Object[] o = q.toArray(null);
433 >            shouldThrow();
434 >        } catch (NullPointerException success) {}
435      }
436  
631
437      /**
438       * iterator does not traverse any elements
439       */
440 <    public void testIterator() {
441 <        SynchronousQueue q = new SynchronousQueue();
442 <        Iterator it = q.iterator();
443 <        assertFalse(it.hasNext());
639 <        try {
640 <            Object x = it.next();
641 <            shouldThrow();
642 <        }
643 <        catch (NoSuchElementException success) {}
440 >    public void testIterator()      { testIterator(false); }
441 >    public void testIterator_fair() { testIterator(true); }
442 >    public void testIterator(boolean fair) {
443 >        assertIteratorExhausted(new SynchronousQueue(fair).iterator());
444      }
445  
446      /**
447       * iterator remove throws ISE
448       */
449 <    public void testIteratorRemove() {
450 <        SynchronousQueue q = new SynchronousQueue();
451 <        Iterator it = q.iterator();
449 >    public void testIteratorRemove()      { testIteratorRemove(false); }
450 >    public void testIteratorRemove_fair() { testIteratorRemove(true); }
451 >    public void testIteratorRemove(boolean fair) {
452 >        final SynchronousQueue q = new SynchronousQueue(fair);
453 >        Iterator it = q.iterator();
454          try {
455              it.remove();
456              shouldThrow();
457 <        }
656 <        catch (IllegalStateException success) {}
457 >        } catch (IllegalStateException success) {}
458      }
459  
460      /**
461       * toString returns a non-null string
462       */
463 <    public void testToString() {
464 <        SynchronousQueue q = new SynchronousQueue();
463 >    public void testToString()      { testToString(false); }
464 >    public void testToString_fair() { testToString(true); }
465 >    public void testToString(boolean fair) {
466 >        final SynchronousQueue q = new SynchronousQueue(fair);
467          String s = q.toString();
468          assertNotNull(s);
469      }
470  
668
471      /**
472       * offer transfers elements across Executor tasks
473       */
474 <    public void testOfferInExecutor() {
475 <        final SynchronousQueue q = new SynchronousQueue();
476 <        ExecutorService executor = Executors.newFixedThreadPool(2);
477 <        final Integer one = new Integer(1);
478 <
479 <        executor.execute(new Runnable() {
480 <            public void run() {
481 <                threadAssertFalse(q.offer(one));
482 <                try {
483 <                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
484 <                    threadAssertEquals(0, q.remainingCapacity());
485 <                }
486 <                catch (InterruptedException e) {
487 <                    threadUnexpectedException();
488 <                }
687 <            }
688 <        });
689 <
690 <        executor.execute(new Runnable() {
691 <            public void run() {
692 <                try {
693 <                    Thread.sleep(SMALL_DELAY_MS);
694 <                    threadAssertEquals(one, q.take());
695 <                }
696 <                catch (InterruptedException e) {
697 <                    threadUnexpectedException();
698 <                }
699 <            }
700 <        });
701 <
702 <        joinPool(executor);
474 >    public void testOfferInExecutor()      { testOfferInExecutor(false); }
475 >    public void testOfferInExecutor_fair() { testOfferInExecutor(true); }
476 >    public void testOfferInExecutor(boolean fair) {
477 >        final SynchronousQueue q = new SynchronousQueue(fair);
478 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
479 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
480 >        try (PoolCleaner cleaner = cleaner(executor)) {
481 >
482 >            executor.execute(new CheckedRunnable() {
483 >                public void realRun() throws InterruptedException {
484 >                    assertFalse(q.offer(one));
485 >                    threadsStarted.await();
486 >                    assertTrue(q.offer(one, LONG_DELAY_MS, MILLISECONDS));
487 >                    assertEquals(0, q.remainingCapacity());
488 >                }});
489  
490 +            executor.execute(new CheckedRunnable() {
491 +                public void realRun() throws InterruptedException {
492 +                    threadsStarted.await();
493 +                    assertSame(one, q.take());
494 +                }});
495 +        }
496      }
497  
498      /**
499 <     * poll retrieves elements across Executor threads
499 >     * timed poll retrieves elements across Executor threads
500       */
501 <    public void testPollInExecutor() {
502 <        final SynchronousQueue q = new SynchronousQueue();
503 <        ExecutorService executor = Executors.newFixedThreadPool(2);
504 <        executor.execute(new Runnable() {
505 <            public void run() {
506 <                threadAssertNull(q.poll());
507 <                try {
508 <                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
509 <                    threadAssertTrue(q.isEmpty());
510 <                }
511 <                catch (InterruptedException e) {
512 <                    threadUnexpectedException();
513 <                }
514 <            }
723 <        });
724 <
725 <        executor.execute(new Runnable() {
726 <            public void run() {
727 <                try {
728 <                    Thread.sleep(SMALL_DELAY_MS);
729 <                    q.put(new Integer(1));
730 <                }
731 <                catch (InterruptedException e) {
732 <                    threadUnexpectedException();
733 <                }
734 <            }
735 <        });
501 >    public void testPollInExecutor()      { testPollInExecutor(false); }
502 >    public void testPollInExecutor_fair() { testPollInExecutor(true); }
503 >    public void testPollInExecutor(boolean fair) {
504 >        final SynchronousQueue q = new SynchronousQueue(fair);
505 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
506 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
507 >        try (PoolCleaner cleaner = cleaner(executor)) {
508 >            executor.execute(new CheckedRunnable() {
509 >                public void realRun() throws InterruptedException {
510 >                    assertNull(q.poll());
511 >                    threadsStarted.await();
512 >                    assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
513 >                    assertTrue(q.isEmpty());
514 >                }});
515  
516 <        joinPool(executor);
516 >            executor.execute(new CheckedRunnable() {
517 >                public void realRun() throws InterruptedException {
518 >                    threadsStarted.await();
519 >                    q.put(one);
520 >                }});
521 >        }
522      }
523  
524      /**
525       * a deserialized serialized queue is usable
526       */
527      public void testSerialization() {
528 <        SynchronousQueue q = new SynchronousQueue();
529 <        try {
530 <            ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
531 <            ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
532 <            out.writeObject(q);
533 <            out.close();
534 <
535 <            ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
536 <            ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
537 <            SynchronousQueue r = (SynchronousQueue)in.readObject();
538 <            assertEquals(q.size(), r.size());
539 <            while (!q.isEmpty())
540 <                assertEquals(q.remove(), r.remove());
541 <        } catch (Exception e){
758 <            e.printStackTrace();
759 <            unexpectedException();
760 <        }
761 <    }
762 <
763 <    /**
764 <     * drainTo(null) throws NPE
765 <     */
766 <    public void testDrainToNull() {
767 <        SynchronousQueue q = new SynchronousQueue();
768 <        try {
769 <            q.drainTo(null);
770 <            shouldThrow();
771 <        } catch (NullPointerException success) {
772 <        }
773 <    }
774 <
775 <    /**
776 <     * drainTo(this) throws IAE
777 <     */
778 <    public void testDrainToSelf() {
779 <        SynchronousQueue q = new SynchronousQueue();
780 <        try {
781 <            q.drainTo(q);
782 <            shouldThrow();
783 <        } catch (IllegalArgumentException success) {
528 >        final SynchronousQueue x = new SynchronousQueue();
529 >        final SynchronousQueue y = new SynchronousQueue(false);
530 >        final SynchronousQueue z = new SynchronousQueue(true);
531 >        assertSerialEquals(x, y);
532 >        assertNotSerialEquals(x, z);
533 >        SynchronousQueue[] qs = { x, y, z };
534 >        for (SynchronousQueue q : qs) {
535 >            SynchronousQueue clone = serialClone(q);
536 >            assertNotSame(q, clone);
537 >            assertSerialEquals(q, clone);
538 >            assertTrue(clone.isEmpty());
539 >            assertEquals(0, clone.size());
540 >            assertEquals(0, clone.remainingCapacity());
541 >            assertFalse(clone.offer(zero));
542          }
543      }
544  
545      /**
546       * drainTo(c) of empty queue doesn't transfer elements
547       */
548 <    public void testDrainTo() {
549 <        SynchronousQueue q = new SynchronousQueue();
548 >    public void testDrainTo()      { testDrainTo(false); }
549 >    public void testDrainTo_fair() { testDrainTo(true); }
550 >    public void testDrainTo(boolean fair) {
551 >        final SynchronousQueue q = new SynchronousQueue(fair);
552          ArrayList l = new ArrayList();
553          q.drainTo(l);
554 <        assertEquals(q.size(), 0);
555 <        assertEquals(l.size(), 0);
554 >        assertEquals(0, q.size());
555 >        assertEquals(0, l.size());
556      }
557  
558      /**
559       * drainTo empties queue, unblocking a waiting put.
560       */
561 <    public void testDrainToWithActivePut() {
562 <        final SynchronousQueue q = new SynchronousQueue();
563 <        Thread t = new Thread(new Runnable() {
564 <                public void run() {
565 <                    try {
566 <                        q.put(new Integer(1));
567 <                    } catch (InterruptedException ie){
568 <                        threadUnexpectedException();
809 <                    }
810 <                }
811 <            });
812 <        try {
813 <            t.start();
814 <            ArrayList l = new ArrayList();
815 <            Thread.sleep(SHORT_DELAY_MS);
816 <            q.drainTo(l);
817 <            assertTrue(l.size() <= 1);
818 <            if (l.size() > 0)
819 <                assertEquals(l.get(0), new Integer(1));
820 <            t.join();
821 <            assertTrue(l.size() <= 1);
822 <        } catch (Exception e){
823 <            unexpectedException();
824 <        }
825 <    }
826 <
827 <    /**
828 <     * drainTo(null, n) throws NPE
829 <     */
830 <    public void testDrainToNullN() {
831 <        SynchronousQueue q = new SynchronousQueue();
832 <        try {
833 <            q.drainTo(null, 0);
834 <            shouldThrow();
835 <        } catch (NullPointerException success) {
836 <        }
837 <    }
561 >    public void testDrainToWithActivePut()      { testDrainToWithActivePut(false); }
562 >    public void testDrainToWithActivePut_fair() { testDrainToWithActivePut(true); }
563 >    public void testDrainToWithActivePut(boolean fair) {
564 >        final SynchronousQueue q = new SynchronousQueue(fair);
565 >        Thread t = newStartedThread(new CheckedRunnable() {
566 >            public void realRun() throws InterruptedException {
567 >                q.put(one);
568 >            }});
569  
570 <    /**
571 <     * drainTo(this, n) throws IAE
572 <     */
573 <    public void testDrainToSelfN() {
574 <        SynchronousQueue q = new SynchronousQueue();
575 <        try {
576 <            q.drainTo(q, 0);
577 <            shouldThrow();
578 <        } catch (IllegalArgumentException success) {
579 <        }
570 >        ArrayList l = new ArrayList();
571 >        long startTime = System.nanoTime();
572 >        while (l.isEmpty()) {
573 >            q.drainTo(l);
574 >            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
575 >                fail("timed out");
576 >            Thread.yield();
577 >        }
578 >        assertEquals(1, l.size());
579 >        assertSame(one, l.get(0));
580 >        awaitTermination(t);
581      }
582  
583      /**
584       * drainTo(c, n) empties up to n elements of queue into c
585       */
586 <    public void testDrainToN() {
586 >    public void testDrainToN() throws InterruptedException {
587          final SynchronousQueue q = new SynchronousQueue();
588 <        Thread t1 = new Thread(new Runnable() {
589 <                public void run() {
590 <                    try {
591 <                        q.put(one);
592 <                    } catch (InterruptedException ie){
593 <                        threadUnexpectedException();
594 <                    }
595 <                }
596 <            });
865 <        Thread t2 = new Thread(new Runnable() {
866 <                public void run() {
867 <                    try {
868 <                        q.put(two);
869 <                    } catch (InterruptedException ie){
870 <                        threadUnexpectedException();
871 <                    }
872 <                }
873 <            });
588 >        Thread t1 = newStartedThread(new CheckedRunnable() {
589 >            public void realRun() throws InterruptedException {
590 >                q.put(one);
591 >            }});
592 >
593 >        Thread t2 = newStartedThread(new CheckedRunnable() {
594 >            public void realRun() throws InterruptedException {
595 >                q.put(two);
596 >            }});
597  
598 <        try {
599 <            t1.start();
600 <            t2.start();
601 <            ArrayList l = new ArrayList();
602 <            Thread.sleep(SHORT_DELAY_MS);
603 <            q.drainTo(l, 1);
604 <            assertTrue(l.size() == 1);
605 <            q.drainTo(l, 1);
606 <            assertTrue(l.size() == 2);
607 <            assertTrue(l.contains(one));
608 <            assertTrue(l.contains(two));
609 <            t1.join();
887 <            t2.join();
888 <        } catch (Exception e){
889 <            unexpectedException();
890 <        }
598 >        ArrayList l = new ArrayList();
599 >        int drained;
600 >        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
601 >        assertEquals(1, drained);
602 >        assertEquals(1, l.size());
603 >        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
604 >        assertEquals(1, drained);
605 >        assertEquals(2, l.size());
606 >        assertTrue(l.contains(one));
607 >        assertTrue(l.contains(two));
608 >        awaitTermination(t1);
609 >        awaitTermination(t2);
610      }
611  
612 +    /**
613 +     * remove(null), contains(null) always return false
614 +     */
615 +    public void testNeverContainsNull() {
616 +        Collection<?> q = new SynchronousQueue();
617 +        assertFalse(q.contains(null));
618 +        assertFalse(q.remove(null));
619 +    }
620  
621   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines