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.5 by dl, Thu Sep 25 11:02:41 2003 UTC vs.
Revision 1.62 by jsr166, Fri Jun 22 00:04:58 2018 UTC

# Line 1 | Line 1
1   /*
2 < * Written by members of JCP JSR-166 Expert Group and released to the
3 < * public domain. Use, modify, and redistribute this code in any way
4 < * without acknowledgement. Other contributors include Andrew Wright,
5 < * Jeffrey Hayes, Pat Fischer, Mike Judd.
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/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 class Fair extends BlockingQueueTest {
28 +        protected BlockingQueue emptyCollection() {
29 +            return new SynchronousQueue(true);
30 +        }
31 +    }
32 +
33 +    public static class NonFair extends BlockingQueueTest {
34 +        protected BlockingQueue emptyCollection() {
35 +            return new SynchronousQueue(false);
36 +        }
37 +    }
38 +
39      public static void main(String[] args) {
40 <        junit.textui.TestRunner.run (suite());  
40 >        main(suite(), args);
41      }
42  
43      public static Test suite() {
44 <        return new TestSuite(SynchronousQueueTest.class);
44 >        return newTestSuite(SynchronousQueueTest.class,
45 >                            new Fair().testSuite(),
46 >                            new NonFair().testSuite());
47      }
48  
49      /**
50 <     * A SynchronousQueue is both empty and full
50 >     * Any SynchronousQueue is both empty and full
51       */
52 <    public void testEmptyFull() {
53 <        SynchronousQueue q = new SynchronousQueue();
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      /**
35     * offer(null) throws NPE
36     */
37    public void testOfferNull() {
38        try {
39            SynchronousQueue q = new SynchronousQueue();
40            q.offer(null);
41            shouldThrow();
42        } catch (NullPointerException success) { }  
43    }
44
45    /**
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());
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              q.add(one);
82              shouldThrow();
83 <        } catch (IllegalStateException success){
63 <        }  
83 >        } catch (IllegalStateException success) {}
84      }
85  
86      /**
87 <     * addAll(null) throws NPE
68 <     */
69 <    public void testAddAll1() {
70 <        try {
71 <            SynchronousQueue q = new SynchronousQueue();
72 <            q.addAll(null);
73 <            shouldThrow();
74 <        }
75 <        catch (NullPointerException success) {}
76 <    }
77 <    /**
78 <     * addAll of a collection with null elements throws NPE
87 >     * addAll(this) throws IllegalArgumentException
88       */
89 <    public void testAddAll2() {
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 {
94 <            SynchronousQueue q = new SynchronousQueue();
83 <            Integer[] ints = new Integer[1];
84 <            q.addAll(Arrays.asList(ints));
94 >            q.addAll(q);
95              shouldThrow();
96 <        }
87 <        catch (NullPointerException success) {}
96 >        } catch (IllegalArgumentException success) {}
97      }
98 +
99      /**
100 <     * addAll throws ISE if no active taker
100 >     * addAll throws IllegalStateException 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();
95 <            Integer[] ints = new Integer[1];
96 <            for (int i = 0; i < 1; ++i)
97 <                ints[i] = new Integer(i);
98 <            q.addAll(Arrays.asList(ints));
111 >            q.addAll(coll);
112              shouldThrow();
113 <        }
101 <        catch (IllegalStateException success) {}
113 >        } catch (IllegalStateException success) {}
114      }
115  
116      /**
105     * put(null) throws NPE
106     */
107    public void testPutNull() {
108        try {
109            SynchronousQueue q = new SynchronousQueue();
110            q.put(null);
111            shouldThrow();
112        }
113        catch (NullPointerException success){
114        }  
115        catch (InterruptedException ie) {
116            unexpectedException();
117        }
118     }
119
120    /**
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);
132 <           t.interrupt();
133 <           t.join();
134 <        }
135 <        catch (InterruptedException ie) {
136 <            unexpectedException();
137 <        }
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 >                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 <     * put blocks waiting for take
149 >     * put blocks interruptibly waiting for take
150       */
151 <    public void testPutWithTake() {
152 <        final SynchronousQueue q = new SynchronousQueue();
153 <        Thread t = new Thread(new Runnable() {
154 <                public void run() {
155 <                    int added = 0;
156 <                    try {
157 <                        q.put(new Object());
158 <                        ++added;
159 <                        q.put(new Object());
160 <                        ++added;
161 <                        q.put(new Object());
162 <                        ++added;
163 <                        q.put(new Object());
164 <                        ++added;
165 <                        threadShouldThrow();
166 <                    } catch (InterruptedException e){
167 <                        assertTrue(added >= 1);
168 <                    }
169 <                }
170 <            });
171 <        try {
172 <            t.start();
173 <            Thread.sleep(SHORT_DELAY_MS);
174 <            q.take();
175 <            Thread.sleep(SHORT_DELAY_MS);
176 <            t.interrupt();
177 <            t.join();
178 <        } catch (Exception e){
179 <            unexpectedException();
180 <        }
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 >                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 >        await(pleaseTake);
178 >        assertEquals(0, q.remainingCapacity());
179 >        try { assertSame(one, q.take()); }
180 >        catch (InterruptedException e) { threadUnexpectedException(e); }
181 >
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 testTimedOffer() {
193 <        final SynchronousQueue q = new SynchronousQueue();
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();
191 <                    } catch (InterruptedException success){}
192 <                }
193 <            });
194 <        
195 <        try {
196 <            t.start();
197 <            Thread.sleep(SMALL_DELAY_MS);
198 <            t.interrupt();
199 <            t.join();
200 <        } catch (Exception e){
201 <            unexpectedException();
202 <        }
203 <    }
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  
202 +                Thread.currentThread().interrupt();
203 +                try {
204 +                    q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
205 +                    shouldThrow();
206 +                } catch (InterruptedException success) {}
207 +                assertFalse(Thread.interrupted());
208  
209 <    /**
210 <     * take blocks interruptibly when empty
211 <     */
212 <    public void testTakeFromEmpty() {
213 <        final SynchronousQueue q = new SynchronousQueue();
214 <        Thread t = new Thread(new Runnable() {
215 <                public void run() {
216 <                    try {
217 <                        q.take();
218 <                        threadShouldThrow();
219 <                    } catch (InterruptedException success){ }                
220 <                }
218 <            });
219 <        try {
220 <            t.start();
221 <            Thread.sleep(SHORT_DELAY_MS);
222 <            t.interrupt();
223 <            t.join();
224 <        } catch (Exception e){
225 <            unexpectedException();
226 <        }
209 >                pleaseInterrupt.countDown();
210 >                try {
211 >                    q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
212 >                    shouldThrow();
213 >                } catch (InterruptedException success) {}
214 >                assertFalse(Thread.interrupted());
215 >            }});
216 >
217 >        await(pleaseInterrupt);
218 >        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
219 >        t.interrupt();
220 >        awaitTermination(t);
221      }
222  
223      /**
224 <     * poll fails unless active taker
224 >     * poll return null if no active putter
225       */
226 <    public void testPoll() {
227 <        SynchronousQueue q = new SynchronousQueue();
228 <        assertNull(q.poll());
226 >    public void testPoll()      { testPoll(false); }
227 >    public void testPoll_fair() { testPoll(true); }
228 >    public void testPoll(boolean fair) {
229 >        final SynchronousQueue q = new SynchronousQueue(fair);
230 >        assertNull(q.poll());
231      }
232  
233      /**
234 <     * timed pool with zero timeout times out if no active taker
234 >     * timed poll with zero timeout times out if no active putter
235       */
236 <    public void testTimedPoll0() {
237 <        try {
238 <            SynchronousQueue q = new SynchronousQueue();
239 <            assertNull(q.poll(0, TimeUnit.MILLISECONDS));
240 <        } catch (InterruptedException e){
241 <            unexpectedException();
246 <        }  
236 >    public void testTimedPoll0()      { testTimedPoll0(false); }
237 >    public void testTimedPoll0_fair() { testTimedPoll0(true); }
238 >    public void testTimedPoll0(boolean fair) {
239 >        final SynchronousQueue q = new SynchronousQueue(fair);
240 >        try { assertNull(q.poll(0, MILLISECONDS)); }
241 >        catch (InterruptedException e) { threadUnexpectedException(e); }
242      }
243  
244      /**
245 <     * timed pool with nonzero timeout times out if no active taker
245 >     * timed poll with nonzero timeout times out if no active putter
246       */
247      public void testTimedPoll() {
248 <        try {
249 <            SynchronousQueue q = new SynchronousQueue();
250 <            assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
251 <        } catch (InterruptedException e){
252 <            unexpectedException();
253 <        }  
248 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
249 >        final SynchronousQueue q = new SynchronousQueue(fair);
250 >        final long startTime = System.nanoTime();
251 >        try { assertNull(q.poll(timeoutMillis(), MILLISECONDS)); }
252 >        catch (InterruptedException e) { threadUnexpectedException(e); }
253 >        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
254      }
255  
256      /**
257 <     * Interrupted timed poll throws InterruptedException instead of
258 <     * returning timeout status
264 <     */
265 <    public void testInterruptedTimedPoll() {
266 <        Thread t = new Thread(new Runnable() {
267 <                public void run() {
268 <                    try {
269 <                        SynchronousQueue q = new SynchronousQueue();
270 <                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
271 <                    } catch (InterruptedException success){
272 <                    }  
273 <                }});
274 <        t.start();
275 <        try {
276 <           Thread.sleep(SHORT_DELAY_MS);
277 <           t.interrupt();
278 <           t.join();
279 <        }
280 <        catch (InterruptedException ie) {
281 <            unexpectedException();
282 <        }
283 <    }
284 <
285 <    /**
286 <     *  timed poll before a delayed offer fails; after offer succeeds;
287 <     *  on interruption throws
257 >     * timed poll before a delayed offer times out, returning null;
258 >     * after offer succeeds; on interruption throws
259       */
260      public void testTimedPollWithOffer() {
261 <        final SynchronousQueue q = new SynchronousQueue();
262 <        Thread t = new Thread(new Runnable() {
263 <                public void run() {
264 <                    try {
265 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
266 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
267 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
268 <                        threadShouldThrow();
269 <                    } catch (InterruptedException success) { }                
270 <                }
271 <            });
272 <        try {
273 <            t.start();
274 <            Thread.sleep(SMALL_DELAY_MS);
275 <            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
276 <            t.interrupt();
277 <            t.join();
278 <        } catch (Exception e){
279 <            unexpectedException();
280 <        }
310 <    }  
261 >        final boolean fair = ThreadLocalRandom.current().nextBoolean();
262 >        final SynchronousQueue q = new SynchronousQueue(fair);
263 >        final CountDownLatch pleaseOffer = new CountDownLatch(1);
264 >        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
265 >        Thread t = newStartedThread(new CheckedRunnable() {
266 >            public void realRun() throws InterruptedException {
267 >                long startTime = System.nanoTime();
268 >                assertNull(q.poll(timeoutMillis(), MILLISECONDS));
269 >                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
270 >
271 >                pleaseOffer.countDown();
272 >                startTime = System.nanoTime();
273 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
274 >
275 >                Thread.currentThread().interrupt();
276 >                try {
277 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
278 >                    shouldThrow();
279 >                } catch (InterruptedException success) {}
280 >                assertFalse(Thread.interrupted());
281  
282 +                pleaseInterrupt.countDown();
283 +                try {
284 +                    q.poll(LONG_DELAY_MS, MILLISECONDS);
285 +                    shouldThrow();
286 +                } catch (InterruptedException success) {}
287 +                assertFalse(Thread.interrupted());
288 +
289 +                assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
290 +            }});
291 +
292 +        await(pleaseOffer);
293 +        long startTime = System.nanoTime();
294 +        try { assertTrue(q.offer(zero, LONG_DELAY_MS, MILLISECONDS)); }
295 +        catch (InterruptedException e) { threadUnexpectedException(e); }
296 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
297 +
298 +        await(pleaseInterrupt);
299 +        assertThreadBlocks(t, Thread.State.TIMED_WAITING);
300 +        t.interrupt();
301 +        awaitTermination(t);
302 +    }
303  
304      /**
305 <     * peek returns null
305 >     * peek() returns null if no active putter
306       */
307 <    public void testPeek() {
308 <        SynchronousQueue q = new SynchronousQueue();
309 <        assertNull(q.peek());
307 >    public void testPeek()      { testPeek(false); }
308 >    public void testPeek_fair() { testPeek(true); }
309 >    public void testPeek(boolean fair) {
310 >        final SynchronousQueue q = new SynchronousQueue(fair);
311 >        assertNull(q.peek());
312      }
313  
314      /**
315 <     * element throws NSEE
315 >     * element() throws NoSuchElementException if no active putter
316       */
317 <    public void testElement() {
318 <        SynchronousQueue q = new SynchronousQueue();
317 >    public void testElement()      { testElement(false); }
318 >    public void testElement_fair() { testElement(true); }
319 >    public void testElement(boolean fair) {
320 >        final SynchronousQueue q = new SynchronousQueue(fair);
321          try {
322              q.element();
323              shouldThrow();
324 <        }
330 <        catch (NoSuchElementException success) {}
324 >        } catch (NoSuchElementException success) {}
325      }
326  
327      /**
328 <     * remove throws NSEE if no active taker
328 >     * remove() throws NoSuchElementException if no active putter
329       */
330 <    public void testRemove() {
331 <        SynchronousQueue q = new SynchronousQueue();
330 >    public void testRemove()      { testRemove(false); }
331 >    public void testRemove_fair() { testRemove(true); }
332 >    public void testRemove(boolean fair) {
333 >        final SynchronousQueue q = new SynchronousQueue(fair);
334          try {
335              q.remove();
336              shouldThrow();
337 <        } catch (NoSuchElementException success){
342 <        }  
337 >        } catch (NoSuchElementException success) {}
338      }
339  
340      /**
346     * remove(x) returns false
347     */
348    public void testRemoveElement() {
349        SynchronousQueue q = new SynchronousQueue();
350        assertFalse(q.remove(zero));
351        assertTrue(q.isEmpty());
352    }
353        
354    /**
341       * contains returns false
342       */
343 <    public void testContains() {
344 <        SynchronousQueue q = new SynchronousQueue();
343 >    public void testContains()      { testContains(false); }
344 >    public void testContains_fair() { testContains(true); }
345 >    public void testContains(boolean fair) {
346 >        final SynchronousQueue q = new SynchronousQueue(fair);
347          assertFalse(q.contains(zero));
348      }
349  
350      /**
351       * clear ensures isEmpty
352       */
353 <    public void testClear() {
354 <        SynchronousQueue q = new SynchronousQueue();
353 >    public void testClear()      { testClear(false); }
354 >    public void testClear_fair() { testClear(true); }
355 >    public void testClear(boolean fair) {
356 >        final SynchronousQueue q = new SynchronousQueue(fair);
357          q.clear();
358          assertTrue(q.isEmpty());
359      }
# Line 371 | Line 361 | public class SynchronousQueueTest extend
361      /**
362       * containsAll returns false unless empty
363       */
364 <    public void testContainsAll() {
365 <        SynchronousQueue q = new SynchronousQueue();
364 >    public void testContainsAll()      { testContainsAll(false); }
365 >    public void testContainsAll_fair() { testContainsAll(true); }
366 >    public void testContainsAll(boolean fair) {
367 >        final SynchronousQueue q = new SynchronousQueue(fair);
368          Integer[] empty = new Integer[0];
369          assertTrue(q.containsAll(Arrays.asList(empty)));
370          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 382 | Line 374 | public class SynchronousQueueTest extend
374      /**
375       * retainAll returns false
376       */
377 <    public void testRetainAll() {
378 <        SynchronousQueue q = new SynchronousQueue();
377 >    public void testRetainAll()      { testRetainAll(false); }
378 >    public void testRetainAll_fair() { testRetainAll(true); }
379 >    public void testRetainAll(boolean fair) {
380 >        final SynchronousQueue q = new SynchronousQueue(fair);
381          Integer[] empty = new Integer[0];
382          assertFalse(q.retainAll(Arrays.asList(empty)));
383          Integer[] ints = new Integer[1]; ints[0] = zero;
# Line 393 | Line 387 | public class SynchronousQueueTest extend
387      /**
388       * removeAll returns false
389       */
390 <    public void testRemoveAll() {
391 <        SynchronousQueue q = new SynchronousQueue();
390 >    public void testRemoveAll()      { testRemoveAll(false); }
391 >    public void testRemoveAll_fair() { testRemoveAll(true); }
392 >    public void testRemoveAll(boolean fair) {
393 >        final SynchronousQueue q = new SynchronousQueue(fair);
394          Integer[] empty = new Integer[0];
395          assertFalse(q.removeAll(Arrays.asList(empty)));
396          Integer[] ints = new Integer[1]; ints[0] = zero;
397          assertFalse(q.containsAll(Arrays.asList(ints)));
398      }
399  
404
400      /**
401       * toArray is empty
402       */
403 <    public void testToArray() {
404 <        SynchronousQueue q = new SynchronousQueue();
405 <        Object[] o = q.toArray();
406 <        assertEquals(o.length, 0);
403 >    public void testToArray()      { testToArray(false); }
404 >    public void testToArray_fair() { testToArray(true); }
405 >    public void testToArray(boolean fair) {
406 >        final SynchronousQueue q = new SynchronousQueue(fair);
407 >        Object[] o = q.toArray();
408 >        assertEquals(0, o.length);
409      }
410  
411      /**
412 <     * toArray(a) is nulled at position 0
412 >     * toArray(Integer array) returns its argument with the first
413 >     * element (if present) nulled out
414       */
415 <    public void testToArray2() {
416 <        SynchronousQueue q = new SynchronousQueue();
417 <        Integer[] ints = new Integer[1];
418 <        assertNull(ints[0]);
415 >    public void testToArray2()      { testToArray2(false); }
416 >    public void testToArray2_fair() { testToArray2(true); }
417 >    public void testToArray2(boolean fair) {
418 >        final SynchronousQueue<Integer> q = new SynchronousQueue<>(fair);
419 >        Integer[] a;
420 >
421 >        a = new Integer[0];
422 >        assertSame(a, q.toArray(a));
423 >
424 >        a = new Integer[3];
425 >        Arrays.fill(a, 42);
426 >        assertSame(a, q.toArray(a));
427 >        assertNull(a[0]);
428 >        for (int i = 1; i < a.length; i++)
429 >            assertEquals(42, (int) a[i]);
430      }
431 <    
431 >
432      /**
433 <     * iterator does not traverse any elements
433 >     * toArray(null) throws NPE
434       */
435 <    public void testIterator() {
436 <        SynchronousQueue q = new SynchronousQueue();
437 <        Iterator it = q.iterator();
438 <        assertFalse(it.hasNext());
435 >    public void testToArray_null()      { testToArray_null(false); }
436 >    public void testToArray_null_fair() { testToArray_null(true); }
437 >    public void testToArray_null(boolean fair) {
438 >        final SynchronousQueue q = new SynchronousQueue(fair);
439          try {
440 <            Object x = it.next();
440 >            Object[] o = q.toArray((Object[])null);
441              shouldThrow();
442 <        }
434 <        catch (NoSuchElementException success) {}
442 >        } catch (NullPointerException success) {}
443      }
444  
445      /**
446 <     * iterator remove throws ISE
446 >     * iterator does not traverse any elements
447 >     */
448 >    public void testIterator()      { testIterator(false); }
449 >    public void testIterator_fair() { testIterator(true); }
450 >    public void testIterator(boolean fair) {
451 >        assertIteratorExhausted(new SynchronousQueue(fair).iterator());
452 >    }
453 >
454 >    /**
455 >     * iterator remove throws IllegalStateException
456       */
457 <    public void testIteratorRemove() {
458 <        SynchronousQueue q = new SynchronousQueue();
459 <        Iterator it = q.iterator();
457 >    public void testIteratorRemove()      { testIteratorRemove(false); }
458 >    public void testIteratorRemove_fair() { testIteratorRemove(true); }
459 >    public void testIteratorRemove(boolean fair) {
460 >        final SynchronousQueue q = new SynchronousQueue(fair);
461 >        Iterator it = q.iterator();
462          try {
463              it.remove();
464              shouldThrow();
465 <        }
447 <        catch (IllegalStateException success) {}
465 >        } catch (IllegalStateException success) {}
466      }
467  
468      /**
469       * toString returns a non-null string
470       */
471 <    public void testToString() {
472 <        SynchronousQueue q = new SynchronousQueue();
471 >    public void testToString()      { testToString(false); }
472 >    public void testToString_fair() { testToString(true); }
473 >    public void testToString(boolean fair) {
474 >        final SynchronousQueue q = new SynchronousQueue(fair);
475          String s = q.toString();
476          assertNotNull(s);
477 <    }        
458 <
477 >    }
478  
479      /**
480       * offer transfers elements across Executor tasks
481       */
482 <    public void testOfferInExecutor() {
483 <        final SynchronousQueue q = new SynchronousQueue();
484 <        ExecutorService executor = Executors.newFixedThreadPool(2);
485 <        final Integer one = new Integer(1);
482 >    public void testOfferInExecutor()      { testOfferInExecutor(false); }
483 >    public void testOfferInExecutor_fair() { testOfferInExecutor(true); }
484 >    public void testOfferInExecutor(boolean fair) {
485 >        final SynchronousQueue q = new SynchronousQueue(fair);
486 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
487 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
488 >        try (PoolCleaner cleaner = cleaner(executor)) {
489 >
490 >            executor.execute(new CheckedRunnable() {
491 >                public void realRun() throws InterruptedException {
492 >                    assertFalse(q.offer(one));
493 >                    threadsStarted.await();
494 >                    assertTrue(q.offer(one, LONG_DELAY_MS, MILLISECONDS));
495 >                    assertEquals(0, q.remainingCapacity());
496 >                }});
497  
498 <        executor.execute(new Runnable() {
499 <            public void run() {
500 <                threadAssertFalse(q.offer(one));
501 <                try {
502 <                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
503 <                    threadAssertEquals(0, q.remainingCapacity());
504 <                }
475 <                catch (InterruptedException e) {
476 <                    threadUnexpectedException();
477 <                }
478 <            }
479 <        });
498 >            executor.execute(new CheckedRunnable() {
499 >                public void realRun() throws InterruptedException {
500 >                    threadsStarted.await();
501 >                    assertSame(one, q.take());
502 >                }});
503 >        }
504 >    }
505  
506 <        executor.execute(new Runnable() {
507 <            public void run() {
508 <                try {
509 <                    Thread.sleep(SMALL_DELAY_MS);
510 <                    threadAssertEquals(one, q.take());
511 <                }
512 <                catch (InterruptedException e) {
513 <                    threadUnexpectedException();
514 <                }
515 <            }
516 <        });
517 <        
518 <        joinPool(executor);
506 >    /**
507 >     * timed poll retrieves elements across Executor threads
508 >     */
509 >    public void testPollInExecutor()      { testPollInExecutor(false); }
510 >    public void testPollInExecutor_fair() { testPollInExecutor(true); }
511 >    public void testPollInExecutor(boolean fair) {
512 >        final SynchronousQueue q = new SynchronousQueue(fair);
513 >        final CheckedBarrier threadsStarted = new CheckedBarrier(2);
514 >        final ExecutorService executor = Executors.newFixedThreadPool(2);
515 >        try (PoolCleaner cleaner = cleaner(executor)) {
516 >            executor.execute(new CheckedRunnable() {
517 >                public void realRun() throws InterruptedException {
518 >                    assertNull(q.poll());
519 >                    threadsStarted.await();
520 >                    assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
521 >                    assertTrue(q.isEmpty());
522 >                }});
523  
524 +            executor.execute(new CheckedRunnable() {
525 +                public void realRun() throws InterruptedException {
526 +                    threadsStarted.await();
527 +                    q.put(one);
528 +                }});
529 +        }
530      }
531  
532      /**
533 <     * poll retrieves elements across Executor threads
533 >     * a deserialized/reserialized queue is usable
534       */
535 <    public void testPollInExecutor() {
536 <        final SynchronousQueue q = new SynchronousQueue();
537 <        ExecutorService executor = Executors.newFixedThreadPool(2);
538 <        executor.execute(new Runnable() {
539 <            public void run() {
540 <                threadAssertNull(q.poll());
541 <                try {
542 <                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
543 <                    threadAssertTrue(q.isEmpty());
544 <                }
545 <                catch (InterruptedException e) {
546 <                    threadUnexpectedException();
547 <                }
548 <            }
549 <        });
535 >    public void testSerialization() {
536 >        final SynchronousQueue x = new SynchronousQueue();
537 >        final SynchronousQueue y = new SynchronousQueue(false);
538 >        final SynchronousQueue z = new SynchronousQueue(true);
539 >        assertSerialEquals(x, y);
540 >        assertNotSerialEquals(x, z);
541 >        SynchronousQueue[] qs = { x, y, z };
542 >        for (SynchronousQueue q : qs) {
543 >            SynchronousQueue clone = serialClone(q);
544 >            assertNotSame(q, clone);
545 >            assertSerialEquals(q, clone);
546 >            assertTrue(clone.isEmpty());
547 >            assertEquals(0, clone.size());
548 >            assertEquals(0, clone.remainingCapacity());
549 >            assertFalse(clone.offer(zero));
550 >        }
551 >    }
552  
553 <        executor.execute(new Runnable() {
554 <            public void run() {
555 <                try {
556 <                    Thread.sleep(SMALL_DELAY_MS);
557 <                    q.put(new Integer(1));
558 <                }
559 <                catch (InterruptedException e) {
560 <                    threadUnexpectedException();
561 <                }
562 <            }
563 <        });
527 <        
528 <        joinPool(executor);
553 >    /**
554 >     * drainTo(c) of empty queue doesn't transfer elements
555 >     */
556 >    public void testDrainTo()      { testDrainTo(false); }
557 >    public void testDrainTo_fair() { testDrainTo(true); }
558 >    public void testDrainTo(boolean fair) {
559 >        final SynchronousQueue q = new SynchronousQueue(fair);
560 >        ArrayList l = new ArrayList();
561 >        q.drainTo(l);
562 >        assertEquals(0, q.size());
563 >        assertEquals(0, l.size());
564      }
565  
566      /**
567 <     * a deserialized serialized queue is usable
567 >     * drainTo empties queue, unblocking a waiting put.
568       */
569 <    public void testSerialization() {
570 <        SynchronousQueue q = new SynchronousQueue();
571 <        try {
572 <            ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
573 <            ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
574 <            out.writeObject(q);
575 <            out.close();
576 <
577 <            ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
578 <            ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
579 <            SynchronousQueue r = (SynchronousQueue)in.readObject();
580 <            assertEquals(q.size(), r.size());
581 <            while (!q.isEmpty())
582 <                assertEquals(q.remove(), r.remove());
583 <        } catch(Exception e){
584 <            unexpectedException();
569 >    public void testDrainToWithActivePut()      { testDrainToWithActivePut(false); }
570 >    public void testDrainToWithActivePut_fair() { testDrainToWithActivePut(true); }
571 >    public void testDrainToWithActivePut(boolean fair) {
572 >        final SynchronousQueue q = new SynchronousQueue(fair);
573 >        Thread t = newStartedThread(new CheckedRunnable() {
574 >            public void realRun() throws InterruptedException {
575 >                q.put(one);
576 >            }});
577 >
578 >        ArrayList l = new ArrayList();
579 >        long startTime = System.nanoTime();
580 >        while (l.isEmpty()) {
581 >            q.drainTo(l);
582 >            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
583 >                fail("timed out");
584 >            Thread.yield();
585          }
586 +        assertEquals(1, l.size());
587 +        assertSame(one, l.get(0));
588 +        awaitTermination(t);
589 +    }
590 +
591 +    /**
592 +     * drainTo(c, n) empties up to n elements of queue into c
593 +     */
594 +    public void testDrainToN() throws InterruptedException {
595 +        final SynchronousQueue q = new SynchronousQueue();
596 +        Thread t1 = newStartedThread(new CheckedRunnable() {
597 +            public void realRun() throws InterruptedException {
598 +                q.put(one);
599 +            }});
600 +
601 +        Thread t2 = newStartedThread(new CheckedRunnable() {
602 +            public void realRun() throws InterruptedException {
603 +                q.put(two);
604 +            }});
605 +
606 +        ArrayList l = new ArrayList();
607 +        int drained;
608 +        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
609 +        assertEquals(1, drained);
610 +        assertEquals(1, l.size());
611 +        while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
612 +        assertEquals(1, drained);
613 +        assertEquals(2, l.size());
614 +        assertTrue(l.contains(one));
615 +        assertTrue(l.contains(two));
616 +        awaitTermination(t1);
617 +        awaitTermination(t2);
618 +    }
619 +
620 +    /**
621 +     * remove(null), contains(null) always return false
622 +     */
623 +    public void testNeverContainsNull() {
624 +        Collection<?> q = new SynchronousQueue();
625 +        assertFalse(q.contains(null));
626 +        assertFalse(q.remove(null));
627      }
628  
629   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines