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.57 by jsr166, Sun May 14 03:48:35 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 <    }
179 <
180 <    /**
181 <     * put blocks waiting for take
182 <     */
183 <    public void testPutWithTake() {
184 <        final SynchronousQueue q = new SynchronousQueue();
185 <        Thread t = new Thread(new Runnable() {
186 <                public void run() {
187 <                    int added = 0;
188 <                    try {
189 <                        q.put(new Object());
190 <                        ++added;
191 <                        q.put(new Object());
192 <                        ++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 <        }
213 <    }
214 <
215 <    /**
216 <     * timed offer times out if elements not taken
217 <     */
218 <    public void testTimedOffer() {
219 <        final SynchronousQueue q = new SynchronousQueue();
220 <        Thread t = new Thread(new Runnable() {
221 <                public void run() {
222 <                    try {
223 <
224 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
225 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
226 <                        threadShouldThrow();
227 <                    } catch (InterruptedException success){}
228 <                }
229 <            });
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 <        try {
134 <            t.start();
135 <            Thread.sleep(SMALL_DELAY_MS);
136 <            t.interrupt();
137 <            t.join();
138 <        } catch (Exception e){
139 <            unexpectedException();
140 <        }
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  
241
148      /**
149 <     * take blocks interruptibly when empty
149 >     * put blocks interruptibly waiting for take
150       */
151 <    public void testTakeFromEmpty() {
152 <        final SynchronousQueue q = new SynchronousQueue();
153 <        Thread t = new Thread(new Runnable() {
154 <                public void run() {
155 <                    try {
156 <                        q.take();
157 <                        threadShouldThrow();
158 <                    } catch (InterruptedException success){ }
159 <                }
160 <            });
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 <    }
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 +                pleaseInterrupt.countDown();
163 +                try {
164 +                    q.put(99);
165 +                    shouldThrow();
166 +                } catch (InterruptedException success) {}
167 +                assertFalse(Thread.interrupted());
168 +            }});
169  
170 <    /**
171 <     * put blocks interruptibly if no active taker
172 <     */
173 <    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 <    }
170 >        await(pleaseTake);
171 >        assertEquals(0, q.remainingCapacity());
172 >        try { assertSame(one, q.take()); }
173 >        catch (InterruptedException e) { threadUnexpectedException(e); }
174  
175 <    /**
176 <     * put blocks waiting for take
177 <     */
178 <    public void testFairPutWithTake() {
179 <        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 <        }
175 >        await(pleaseInterrupt);
176 >        assertThreadBlocks(t, Thread.State.WAITING);
177 >        t.interrupt();
178 >        awaitTermination(t);
179 >        assertEquals(0, q.remainingCapacity());
180      }
181  
182      /**
183       * timed offer times out if elements not taken
184       */
185 <    public void testFairTimedOffer() {
186 <        final SynchronousQueue q = new SynchronousQueue(true);
187 <        Thread t = new Thread(new Runnable() {
188 <                public void run() {
189 <                    try {
190 <
191 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
192 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
193 <                        threadShouldThrow();
194 <                    } catch (InterruptedException success){}
195 <                }
196 <            });
185 >    public void testTimedOffer() {
186 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
187 >        final SynchronousQueue q = new SynchronousQueue(fair);
188 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
189 >        Thread t = newStartedThread(new CheckedRunnable() {
190 >            public void realRun() throws InterruptedException {
191 >                long startTime = System.nanoTime();
192 >                assertFalse(q.offer(new Object(), timeoutMillis(), MILLISECONDS));
193 >                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
194 >                pleaseInterrupt.countDown();
195 >                try {
196 >                    q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
197 >                    shouldThrow();
198 >                } catch (InterruptedException success) {}
199 >                assertFalse(Thread.interrupted());
200 >            }});
201  
202 <        try {
203 <            t.start();
204 <            Thread.sleep(SMALL_DELAY_MS);
205 <            t.interrupt();
345 <            t.join();
346 <        } catch (Exception e){
347 <            unexpectedException();
348 <        }
202 >        await(pleaseInterrupt);
203 >        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
204 >        t.interrupt();
205 >        awaitTermination(t);
206      }
207  
351
208      /**
209 <     * take blocks interruptibly when empty
209 >     * poll return null if no active putter
210       */
211 <    public void testFairTakeFromEmpty() {
212 <        final SynchronousQueue q = new SynchronousQueue(true);
213 <        Thread t = new Thread(new Runnable() {
214 <                public void run() {
215 <                    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 <        }
211 >    public void testPoll()      { testPoll(false); }
212 >    public void testPoll_fair() { testPoll(true); }
213 >    public void testPoll(boolean fair) {
214 >        final SynchronousQueue q = new SynchronousQueue(fair);
215 >        assertNull(q.poll());
216      }
217  
218      /**
219 <     * poll fails unless active taker
219 >     * timed poll with zero timeout times out if no active putter
220       */
221 <    public void testPoll() {
222 <        SynchronousQueue q = new SynchronousQueue();
223 <        assertNull(q.poll());
221 >    public void testTimedPoll0()      { testTimedPoll0(false); }
222 >    public void testTimedPoll0_fair() { testTimedPoll0(true); }
223 >    public void testTimedPoll0(boolean fair) {
224 >        final SynchronousQueue q = new SynchronousQueue(fair);
225 >        try { assertNull(q.poll(0, MILLISECONDS)); }
226 >        catch (InterruptedException e) { threadUnexpectedException(e); }
227      }
228  
229      /**
230 <     * 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
230 >     * timed poll with nonzero timeout times out if no active putter
231       */
232      public void testTimedPoll() {
233 <        try {
234 <            SynchronousQueue q = new SynchronousQueue();
235 <            assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
236 <        } catch (InterruptedException e){
237 <            unexpectedException();
238 <        }
233 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
234 >        final SynchronousQueue q = new SynchronousQueue(fair);
235 >        final long startTime = System.nanoTime();
236 >        try { assertNull(q.poll(timeoutMillis(), MILLISECONDS)); }
237 >        catch (InterruptedException e) { threadUnexpectedException(e); }
238 >        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
239      }
240  
241      /**
242 <     * Interrupted timed poll throws InterruptedException instead of
243 <     * 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
242 >     * timed poll before a delayed offer times out, returning null;
243 >     * after offer succeeds; on interruption throws
244       */
245      public void testTimedPollWithOffer() {
246 <        final SynchronousQueue q = new SynchronousQueue();
247 <        Thread t = new Thread(new Runnable() {
248 <                public void run() {
249 <                    try {
250 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
251 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
252 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
253 <                        threadShouldThrow();
254 <                    } catch (InterruptedException success) { }
255 <                }
256 <            });
257 <        try {
258 <            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 <    }
246 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
247 >        final SynchronousQueue q = new SynchronousQueue(fair);
248 >        final CountDownLatch pleaseOffer = new CountDownLatch(1);
249 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
250 >        Thread t = newStartedThread(new CheckedRunnable() {
251 >            public void realRun() throws InterruptedException {
252 >                long startTime = System.nanoTime();
253 >                assertNull(q.poll(timeoutMillis(), MILLISECONDS));
254 >                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
255 >
256 >                pleaseOffer.countDown();
257 >                startTime = System.nanoTime();
258 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
259  
260 <    /**
261 <     * Interrupted timed poll throws InterruptedException instead of
262 <     * returning timeout status
263 <     */
264 <    public void testFairInterruptedTimedPoll() {
265 <        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 <    }
260 >                Thread.currentThread().interrupt();
261 >                try {
262 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
263 >                    shouldThrow();
264 >                } catch (InterruptedException success) {}
265 >                assertFalse(Thread.interrupted());
266  
267 <    /**
268 <     *  timed poll before a delayed offer fails; after offer succeeds;
269 <     *  on interruption throws
270 <     */
271 <    public void testFairTimedPollWithOffer() {
272 <        final SynchronousQueue q = new SynchronousQueue(true);
273 <        Thread t = new Thread(new Runnable() {
274 <                public void run() {
275 <                    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 <    }
267 >                pleaseInterrupt.countDown();
268 >                try {
269 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
270 >                    shouldThrow();
271 >                } catch (InterruptedException success) {}
272 >                assertFalse(Thread.interrupted());
273 >
274 >                assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
275 >            }});
276  
277 +        await(pleaseOffer);
278 +        long startTime = System.nanoTime();
279 +        try { assertTrue(q.offer(zero, LONG_DELAY_MS, MILLISECONDS)); }
280 +        catch (InterruptedException e) { threadUnexpectedException(e); }
281 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
282 +
283 +        await(pleaseInterrupt);
284 +        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
285 +        t.interrupt();
286 +        awaitTermination(t);
287 +    }
288  
289      /**
290 <     * peek returns null
290 >     * peek() returns null if no active putter
291       */
292 <    public void testPeek() {
293 <        SynchronousQueue q = new SynchronousQueue();
294 <        assertNull(q.peek());
292 >    public void testPeek()      { testPeek(false); }
293 >    public void testPeek_fair() { testPeek(true); }
294 >    public void testPeek(boolean fair) {
295 >        final SynchronousQueue q = new SynchronousQueue(fair);
296 >        assertNull(q.peek());
297      }
298  
299      /**
300 <     * element throws NSEE
300 >     * element() throws NoSuchElementException if no active putter
301       */
302 <    public void testElement() {
303 <        SynchronousQueue q = new SynchronousQueue();
302 >    public void testElement()      { testElement(false); }
303 >    public void testElement_fair() { testElement(true); }
304 >    public void testElement(boolean fair) {
305 >        final SynchronousQueue q = new SynchronousQueue(fair);
306          try {
307              q.element();
308              shouldThrow();
309 <        }
527 <        catch (NoSuchElementException success) {}
309 >        } catch (NoSuchElementException success) {}
310      }
311  
312      /**
313 <     * remove throws NSEE if no active taker
313 >     * remove() throws NoSuchElementException if no active putter
314       */
315 <    public void testRemove() {
316 <        SynchronousQueue q = new SynchronousQueue();
315 >    public void testRemove()      { testRemove(false); }
316 >    public void testRemove_fair() { testRemove(true); }
317 >    public void testRemove(boolean fair) {
318 >        final SynchronousQueue q = new SynchronousQueue(fair);
319          try {
320              q.remove();
321              shouldThrow();
322 <        } 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());
322 >        } catch (NoSuchElementException success) {}
323      }
324  
325      /**
326       * contains returns false
327       */
328 <    public void testContains() {
329 <        SynchronousQueue q = new SynchronousQueue();
328 >    public void testContains()      { testContains(false); }
329 >    public void testContains_fair() { testContains(true); }
330 >    public void testContains(boolean fair) {
331 >        final SynchronousQueue q = new SynchronousQueue(fair);
332          assertFalse(q.contains(zero));
333      }
334  
335      /**
336       * clear ensures isEmpty
337       */
338 <    public void testClear() {
339 <        SynchronousQueue q = new SynchronousQueue();
338 >    public void testClear()      { testClear(false); }
339 >    public void testClear_fair() { testClear(true); }
340 >    public void testClear(boolean fair) {
341 >        final SynchronousQueue q = new SynchronousQueue(fair);
342          q.clear();
343          assertTrue(q.isEmpty());
344      }
# Line 568 | Line 346 | public class SynchronousQueueTest extend
346      /**
347       * containsAll returns false unless empty
348       */
349 <    public void testContainsAll() {
350 <        SynchronousQueue q = new SynchronousQueue();
349 >    public void testContainsAll()      { testContainsAll(false); }
350 >    public void testContainsAll_fair() { testContainsAll(true); }
351 >    public void testContainsAll(boolean fair) {
352 >        final SynchronousQueue q = new SynchronousQueue(fair);
353          Integer[] empty = new Integer[0];
354          assertTrue(q.containsAll(Arrays.asList(empty)));
355          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 579 | Line 359 | public class SynchronousQueueTest extend
359      /**
360       * retainAll returns false
361       */
362 <    public void testRetainAll() {
363 <        SynchronousQueue q = new SynchronousQueue();
362 >    public void testRetainAll()      { testRetainAll(false); }
363 >    public void testRetainAll_fair() { testRetainAll(true); }
364 >    public void testRetainAll(boolean fair) {
365 >        final SynchronousQueue q = new SynchronousQueue(fair);
366          Integer[] empty = new Integer[0];
367          assertFalse(q.retainAll(Arrays.asList(empty)));
368          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 590 | Line 372 | public class SynchronousQueueTest extend
372      /**
373       * removeAll returns false
374       */
375 <    public void testRemoveAll() {
376 <        SynchronousQueue q = new SynchronousQueue();
375 >    public void testRemoveAll()      { testRemoveAll(false); }
376 >    public void testRemoveAll_fair() { testRemoveAll(true); }
377 >    public void testRemoveAll(boolean fair) {
378 >        final SynchronousQueue q = new SynchronousQueue(fair);
379          Integer[] empty = new Integer[0];
380          assertFalse(q.removeAll(Arrays.asList(empty)));
381          Integer[] ints = new Integer[1]; ints[0] = zero;
382          assertFalse(q.containsAll(Arrays.asList(ints)));
383      }
384  
601
385      /**
386       * toArray is empty
387       */
388 <    public void testToArray() {
389 <        SynchronousQueue q = new SynchronousQueue();
390 <        Object[] o = q.toArray();
391 <        assertEquals(o.length, 0);
388 >    public void testToArray()      { testToArray(false); }
389 >    public void testToArray_fair() { testToArray(true); }
390 >    public void testToArray(boolean fair) {
391 >        final SynchronousQueue q = new SynchronousQueue(fair);
392 >        Object[] o = q.toArray();
393 >        assertEquals(0, o.length);
394      }
395  
396      /**
397 <     * toArray(a) is nulled at position 0
398 <     */
399 <    public void testToArray2() {
400 <        SynchronousQueue q = new SynchronousQueue();
401 <        Integer[] ints = new Integer[1];
402 <        assertNull(ints[0]);
397 >     * toArray(Integer array) returns its argument with the first
398 >     * element (if present) nulled out
399 >     */
400 >    public void testToArray2()      { testToArray2(false); }
401 >    public void testToArray2_fair() { testToArray2(true); }
402 >    public void testToArray2(boolean fair) {
403 >        final SynchronousQueue<Integer> q = new SynchronousQueue<>(fair);
404 >        Integer[] a;
405 >
406 >        a = new Integer[0];
407 >        assertSame(a, q.toArray(a));
408 >
409 >        a = new Integer[3];
410 >        Arrays.fill(a, 42);
411 >        assertSame(a, q.toArray(a));
412 >        assertNull(a[0]);
413 >        for (int i = 1; i < a.length; i++)
414 >            assertEquals(42, (int) a[i]);
415      }
416  
417      /**
418       * toArray(null) throws NPE
419       */
420 <    public void testToArray_BadArg() {
421 <        try {
422 <            SynchronousQueue q = new SynchronousQueue();
423 <            Object o[] = q.toArray(null);
424 <            shouldThrow();
425 <        } catch (NullPointerException success){}
420 >    public void testToArray_null()      { testToArray_null(false); }
421 >    public void testToArray_null_fair() { testToArray_null(true); }
422 >    public void testToArray_null(boolean fair) {
423 >        final SynchronousQueue q = new SynchronousQueue(fair);
424 >        try {
425 >            Object[] o = q.toArray(null);
426 >            shouldThrow();
427 >        } catch (NullPointerException success) {}
428      }
429  
631
430      /**
431       * iterator does not traverse any elements
432       */
433 <    public void testIterator() {
434 <        SynchronousQueue q = new SynchronousQueue();
435 <        Iterator it = q.iterator();
436 <        assertFalse(it.hasNext());
639 <        try {
640 <            Object x = it.next();
641 <            shouldThrow();
642 <        }
643 <        catch (NoSuchElementException success) {}
433 >    public void testIterator()      { testIterator(false); }
434 >    public void testIterator_fair() { testIterator(true); }
435 >    public void testIterator(boolean fair) {
436 >        assertIteratorExhausted(new SynchronousQueue(fair).iterator());
437      }
438  
439      /**
440       * iterator remove throws ISE
441       */
442 <    public void testIteratorRemove() {
443 <        SynchronousQueue q = new SynchronousQueue();
444 <        Iterator it = q.iterator();
442 >    public void testIteratorRemove()      { testIteratorRemove(false); }
443 >    public void testIteratorRemove_fair() { testIteratorRemove(true); }
444 >    public void testIteratorRemove(boolean fair) {
445 >        final SynchronousQueue q = new SynchronousQueue(fair);
446 >        Iterator it = q.iterator();
447          try {
448              it.remove();
449              shouldThrow();
450 <        }
656 <        catch (IllegalStateException success) {}
450 >        } catch (IllegalStateException success) {}
451      }
452  
453      /**
454       * toString returns a non-null string
455       */
456 <    public void testToString() {
457 <        SynchronousQueue q = new SynchronousQueue();
456 >    public void testToString()      { testToString(false); }
457 >    public void testToString_fair() { testToString(true); }
458 >    public void testToString(boolean fair) {
459 >        final SynchronousQueue q = new SynchronousQueue(fair);
460          String s = q.toString();
461          assertNotNull(s);
462      }
463  
668
464      /**
465       * offer transfers elements across Executor tasks
466       */
467 <    public void testOfferInExecutor() {
468 <        final SynchronousQueue q = new SynchronousQueue();
469 <        ExecutorService executor = Executors.newFixedThreadPool(2);
470 <        final Integer one = new Integer(1);
471 <
472 <        executor.execute(new Runnable() {
473 <            public void run() {
474 <                threadAssertFalse(q.offer(one));
475 <                try {
476 <                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
477 <                    threadAssertEquals(0, q.remainingCapacity());
478 <                }
479 <                catch (InterruptedException e) {
480 <                    threadUnexpectedException();
481 <                }
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);
467 >    public void testOfferInExecutor()      { testOfferInExecutor(false); }
468 >    public void testOfferInExecutor_fair() { testOfferInExecutor(true); }
469 >    public void testOfferInExecutor(boolean fair) {
470 >        final SynchronousQueue q = new SynchronousQueue(fair);
471 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
472 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
473 >        try (PoolCleaner cleaner = cleaner(executor)) {
474 >
475 >            executor.execute(new CheckedRunnable() {
476 >                public void realRun() throws InterruptedException {
477 >                    assertFalse(q.offer(one));
478 >                    threadsStarted.await();
479 >                    assertTrue(q.offer(one, LONG_DELAY_MS, MILLISECONDS));
480 >                    assertEquals(0, q.remainingCapacity());
481 >                }});
482  
483 +            executor.execute(new CheckedRunnable() {
484 +                public void realRun() throws InterruptedException {
485 +                    threadsStarted.await();
486 +                    assertSame(one, q.take());
487 +                }});
488 +        }
489      }
490  
491      /**
492 <     * poll retrieves elements across Executor threads
492 >     * timed poll retrieves elements across Executor threads
493       */
494 <    public void testPollInExecutor() {
495 <        final SynchronousQueue q = new SynchronousQueue();
496 <        ExecutorService executor = Executors.newFixedThreadPool(2);
497 <        executor.execute(new Runnable() {
498 <            public void run() {
499 <                threadAssertNull(q.poll());
500 <                try {
501 <                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
502 <                    threadAssertTrue(q.isEmpty());
503 <                }
504 <                catch (InterruptedException e) {
505 <                    threadUnexpectedException();
506 <                }
507 <            }
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 <        });
494 >    public void testPollInExecutor()      { testPollInExecutor(false); }
495 >    public void testPollInExecutor_fair() { testPollInExecutor(true); }
496 >    public void testPollInExecutor(boolean fair) {
497 >        final SynchronousQueue q = new SynchronousQueue(fair);
498 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
499 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
500 >        try (PoolCleaner cleaner = cleaner(executor)) {
501 >            executor.execute(new CheckedRunnable() {
502 >                public void realRun() throws InterruptedException {
503 >                    assertNull(q.poll());
504 >                    threadsStarted.await();
505 >                    assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
506 >                    assertTrue(q.isEmpty());
507 >                }});
508  
509 <        joinPool(executor);
509 >            executor.execute(new CheckedRunnable() {
510 >                public void realRun() throws InterruptedException {
511 >                    threadsStarted.await();
512 >                    q.put(one);
513 >                }});
514 >        }
515      }
516  
517      /**
518       * a deserialized serialized queue is usable
519       */
520      public void testSerialization() {
521 <        SynchronousQueue q = new SynchronousQueue();
522 <        try {
523 <            ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
524 <            ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
525 <            out.writeObject(q);
526 <            out.close();
527 <
528 <            ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
529 <            ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
530 <            SynchronousQueue r = (SynchronousQueue)in.readObject();
531 <            assertEquals(q.size(), r.size());
532 <            while (!q.isEmpty())
533 <                assertEquals(q.remove(), r.remove());
534 <        } 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) {
521 >        final SynchronousQueue x = new SynchronousQueue();
522 >        final SynchronousQueue y = new SynchronousQueue(false);
523 >        final SynchronousQueue z = new SynchronousQueue(true);
524 >        assertSerialEquals(x, y);
525 >        assertNotSerialEquals(x, z);
526 >        SynchronousQueue[] qs = { x, y, z };
527 >        for (SynchronousQueue q : qs) {
528 >            SynchronousQueue clone = serialClone(q);
529 >            assertNotSame(q, clone);
530 >            assertSerialEquals(q, clone);
531 >            assertTrue(clone.isEmpty());
532 >            assertEquals(0, clone.size());
533 >            assertEquals(0, clone.remainingCapacity());
534 >            assertFalse(clone.offer(zero));
535          }
536      }
537  
538      /**
539       * drainTo(c) of empty queue doesn't transfer elements
540       */
541 <    public void testDrainTo() {
542 <        SynchronousQueue q = new SynchronousQueue();
541 >    public void testDrainTo()      { testDrainTo(false); }
542 >    public void testDrainTo_fair() { testDrainTo(true); }
543 >    public void testDrainTo(boolean fair) {
544 >        final SynchronousQueue q = new SynchronousQueue(fair);
545          ArrayList l = new ArrayList();
546          q.drainTo(l);
547 <        assertEquals(q.size(), 0);
548 <        assertEquals(l.size(), 0);
547 >        assertEquals(0, q.size());
548 >        assertEquals(0, l.size());
549      }
550  
551      /**
552       * drainTo empties queue, unblocking a waiting put.
553       */
554 <    public void testDrainToWithActivePut() {
555 <        final SynchronousQueue q = new SynchronousQueue();
556 <        Thread t = new Thread(new Runnable() {
557 <                public void run() {
558 <                    try {
559 <                        q.put(new Integer(1));
560 <                    } catch (InterruptedException ie){
561 <                        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 <    }
554 >    public void testDrainToWithActivePut()      { testDrainToWithActivePut(false); }
555 >    public void testDrainToWithActivePut_fair() { testDrainToWithActivePut(true); }
556 >    public void testDrainToWithActivePut(boolean fair) {
557 >        final SynchronousQueue q = new SynchronousQueue(fair);
558 >        Thread t = newStartedThread(new CheckedRunnable() {
559 >            public void realRun() throws InterruptedException {
560 >                q.put(one);
561 >            }});
562  
563 <    /**
564 <     * drainTo(this, n) throws IAE
565 <     */
566 <    public void testDrainToSelfN() {
567 <        SynchronousQueue q = new SynchronousQueue();
568 <        try {
569 <            q.drainTo(q, 0);
570 <            shouldThrow();
571 <        } catch (IllegalArgumentException success) {
572 <        }
563 >        ArrayList l = new ArrayList();
564 >        long startTime = System.nanoTime();
565 >        while (l.isEmpty()) {
566 >            q.drainTo(l);
567 >            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
568 >                fail("timed out");
569 >            Thread.yield();
570 >        }
571 >        assertEquals(1, l.size());
572 >        assertSame(one, l.get(0));
573 >        awaitTermination(t);
574      }
575  
576      /**
577       * drainTo(c, n) empties up to n elements of queue into c
578       */
579 <    public void testDrainToN() {
579 >    public void testDrainToN() throws InterruptedException {
580          final SynchronousQueue q = new SynchronousQueue();
581 <        Thread t1 = new Thread(new Runnable() {
582 <                public void run() {
583 <                    try {
584 <                        q.put(one);
585 <                    } catch (InterruptedException ie){
586 <                        threadUnexpectedException();
587 <                    }
588 <                }
589 <            });
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 <            });
581 >        Thread t1 = newStartedThread(new CheckedRunnable() {
582 >            public void realRun() throws InterruptedException {
583 >                q.put(one);
584 >            }});
585 >
586 >        Thread t2 = newStartedThread(new CheckedRunnable() {
587 >            public void realRun() throws InterruptedException {
588 >                q.put(two);
589 >            }});
590  
591 <        try {
592 <            t1.start();
593 <            t2.start();
594 <            ArrayList l = new ArrayList();
595 <            Thread.sleep(SHORT_DELAY_MS);
596 <            q.drainTo(l, 1);
597 <            assertTrue(l.size() == 1);
598 <            q.drainTo(l, 1);
599 <            assertTrue(l.size() == 2);
600 <            assertTrue(l.contains(one));
601 <            assertTrue(l.contains(two));
602 <            t1.join();
887 <            t2.join();
888 <        } catch (Exception e){
889 <            unexpectedException();
890 <        }
591 >        ArrayList l = new ArrayList();
592 >        int drained;
593 >        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
594 >        assertEquals(1, drained);
595 >        assertEquals(1, l.size());
596 >        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
597 >        assertEquals(1, drained);
598 >        assertEquals(2, l.size());
599 >        assertTrue(l.contains(one));
600 >        assertTrue(l.contains(two));
601 >        awaitTermination(t1);
602 >        awaitTermination(t2);
603      }
604  
605 +    /**
606 +     * remove(null), contains(null) always return false
607 +     */
608 +    public void testNeverContainsNull() {
609 +        Collection<?> q = new SynchronousQueue();
610 +        assertFalse(q.contains(null));
611 +        assertFalse(q.remove(null));
612 +    }
613  
614   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines