test/tck/SynchronousQueueTest.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 * Other contributors include Andrew Wright, Jeffrey Hayes,
 * Pat Fisher, Mike Judd.
 */

import junit.framework.*;
import java.util.*;
import java.util.concurrent.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.io.*;

public class SynchronousQueueTest extends JSR166TestCase {

    public static class Fair extends BlockingQueueTest {
        protected BlockingQueue emptyCollection() {
            return new SynchronousQueue(true);
        }
    }

    public static class NonFair extends BlockingQueueTest {
        protected BlockingQueue emptyCollection() {
            return new SynchronousQueue(false);
        }
    }

    public static void main(String[] args) {
        junit.textui.TestRunner.run(suite());
    }

    public static Test suite() {
        return newTestSuite(SynchronousQueueTest.class,
                            new Fair().testSuite(),
                            new NonFair().testSuite());
    }

    /**
     * A SynchronousQueue is both empty and full
     */
    public void testEmptyFull() {
        SynchronousQueue q = new SynchronousQueue();
        assertTrue(q.isEmpty());
        assertEquals(0, q.size());
        assertEquals(0, q.remainingCapacity());
        assertFalse(q.offer(zero));
    }

    /**
     * A fair SynchronousQueue is both empty and full
     */
    public void testFairEmptyFull() {
        SynchronousQueue q = new SynchronousQueue(true);
        assertTrue(q.isEmpty());
        assertEquals(0, q.size());
        assertEquals(0, q.remainingCapacity());
        assertFalse(q.offer(zero));
    }

    /**
     * offer(null) throws NPE
     */
    public void testOfferNull() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            q.offer(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * add(null) throws NPE
     */
    public void testAddNull() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            q.add(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * offer fails if no active taker
     */
    public void testOffer() {
        SynchronousQueue q = new SynchronousQueue();
        assertFalse(q.offer(one));
    }

    /**
     * add throws ISE if no active taker
     */
    public void testAdd() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            assertEquals(0, q.remainingCapacity());
            q.add(one);
            shouldThrow();
        } catch (IllegalStateException success) {}
    }

    /**
     * addAll(null) throws NPE
     */
    public void testAddAll1() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            q.addAll(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * addAll(this) throws IAE
     */
    public void testAddAllSelf() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            q.addAll(q);
            shouldThrow();
        } catch (IllegalArgumentException success) {}
    }

    /**
     * addAll of a collection with null elements throws NPE
     */
    public void testAddAll2() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            Integer[] ints = new Integer[1];
            q.addAll(Arrays.asList(ints));
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * addAll throws ISE if no active taker
     */
    public void testAddAll4() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            Integer[] ints = new Integer[1];
            for (int i = 0; i < 1; ++i)
                ints[i] = new Integer(i);
            q.addAll(Arrays.asList(ints));
            shouldThrow();
        } catch (IllegalStateException success) {}
    }

    /**
     * put(null) throws NPE
     */
    public void testPutNull() throws InterruptedException {
        try {
            SynchronousQueue q = new SynchronousQueue();
            q.put(null);
            shouldThrow();
        } catch (NullPointerException success) {}
     }

    /**
     * put blocks interruptibly if no active taker
     */
    public void testBlockingPut() throws InterruptedException {
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                SynchronousQueue q = new SynchronousQueue();
                q.put(zero);
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * put blocks waiting for take
     */
    public void testPutWithTake() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                int added = 0;
                try {
                    while (true) {
                        q.put(added);
                        ++added;
                    }
                } catch (InterruptedException success) {
                    assertEquals(1, added);
                }
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        assertEquals(0, q.take());
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testTimedOffer() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
            }});

        t.start();
        Thread.sleep(SMALL_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * put blocks interruptibly if no active taker
     */
    public void testFairBlockingPut() throws InterruptedException {
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                SynchronousQueue q = new SynchronousQueue(true);
                q.put(zero);
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * put blocks waiting for take
     */
    public void testFairPutWithTake() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                int added = 0;
                try {
                    while (true) {
                        q.put(added);
                        ++added;
                    }
                } catch (InterruptedException success) {
                    assertEquals(1, added);
                }
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        assertEquals(0, q.take());
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testFairTimedOffer() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
            }});

        t.start();
        Thread.sleep(SMALL_DELAY_MS);
        t.interrupt();
        t.join();
    }


    /**
     * take blocks interruptibly when empty
     */
    public void testFairTakeFromEmpty() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                q.take();
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * poll fails unless active taker
     */
    public void testPoll() {
        SynchronousQueue q = new SynchronousQueue();
        assertNull(q.poll());
    }

    /**
     * timed poll with zero timeout times out if no active taker
     */
    public void testTimedPoll0() throws InterruptedException {
        SynchronousQueue q = new SynchronousQueue();
        assertNull(q.poll(0, MILLISECONDS));
    }

    /**
     * timed poll with nonzero timeout times out if no active taker
     */
    public void testTimedPoll() throws InterruptedException {
        SynchronousQueue q = new SynchronousQueue();
        assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testInterruptedTimedPoll() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                q.poll(SMALL_DELAY_MS, MILLISECONDS);
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testFairInterruptedTimedPoll() throws InterruptedException {
        Thread t = new Thread(new CheckedInterruptedRunnable() {
            public void realRun() throws InterruptedException {
                SynchronousQueue q = new SynchronousQueue(true);
                q.poll(SMALL_DELAY_MS, MILLISECONDS);
            }});

        t.start();
        Thread.sleep(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * timed poll before a delayed offer fails; after offer succeeds;
     * on interruption throws
     */
    public void testFairTimedPollWithOffer() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                try {
                    assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
                    assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
                    q.poll(LONG_DELAY_MS, MILLISECONDS);
                    threadShouldThrow();
                } catch (InterruptedException success) {}
            }});

        t.start();
        Thread.sleep(SMALL_DELAY_MS);
        assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
        t.interrupt();
        t.join();
    }


    /**
     * peek returns null
     */
    public void testPeek() {
        SynchronousQueue q = new SynchronousQueue();
        assertNull(q.peek());
    }

    /**
     * element throws NSEE
     */
    public void testElement() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.element();
            shouldThrow();
        } catch (NoSuchElementException success) {}
    }

    /**
     * remove throws NSEE if no active taker
     */
    public void testRemove() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.remove();
            shouldThrow();
        } catch (NoSuchElementException success) {}
    }

    /**
     * remove(x) returns false
     */
    public void testRemoveElement() {
        SynchronousQueue q = new SynchronousQueue();
        assertFalse(q.remove(zero));
        assertTrue(q.isEmpty());
    }

    /**
     * contains returns false
     */
    public void testContains() {
        SynchronousQueue q = new SynchronousQueue();
        assertFalse(q.contains(zero));
    }

    /**
     * clear ensures isEmpty
     */
    public void testClear() {
        SynchronousQueue q = new SynchronousQueue();
        q.clear();
        assertTrue(q.isEmpty());
    }

    /**
     * containsAll returns false unless empty
     */
    public void testContainsAll() {
        SynchronousQueue q = new SynchronousQueue();
        Integer[] empty = new Integer[0];
        assertTrue(q.containsAll(Arrays.asList(empty)));
        Integer[] ints = new Integer[1]; ints[0] = zero;
        assertFalse(q.containsAll(Arrays.asList(ints)));
    }

    /**
     * retainAll returns false
     */
    public void testRetainAll() {
        SynchronousQueue q = new SynchronousQueue();
        Integer[] empty = new Integer[0];
        assertFalse(q.retainAll(Arrays.asList(empty)));
        Integer[] ints = new Integer[1]; ints[0] = zero;
        assertFalse(q.retainAll(Arrays.asList(ints)));
    }

    /**
     * removeAll returns false
     */
    public void testRemoveAll() {
        SynchronousQueue q = new SynchronousQueue();
        Integer[] empty = new Integer[0];
        assertFalse(q.removeAll(Arrays.asList(empty)));
        Integer[] ints = new Integer[1]; ints[0] = zero;
        assertFalse(q.containsAll(Arrays.asList(ints)));
    }


    /**
     * toArray is empty
     */
    public void testToArray() {
        SynchronousQueue q = new SynchronousQueue();
        Object[] o = q.toArray();
        assertEquals(o.length, 0);
    }

    /**
     * toArray(a) is nulled at position 0
     */
    public void testToArray2() {
        SynchronousQueue q = new SynchronousQueue();
        Integer[] ints = new Integer[1];
        assertNull(ints[0]);
    }

    /**
     * toArray(null) throws NPE
     */
    public void testToArray_BadArg() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            Object o[] = q.toArray(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }


    /**
     * iterator does not traverse any elements
     */
    public void testIterator() {
        SynchronousQueue q = new SynchronousQueue();
        Iterator it = q.iterator();
        assertFalse(it.hasNext());
        try {
            Object x = it.next();
            shouldThrow();
        } catch (NoSuchElementException success) {}
    }

    /**
     * iterator remove throws ISE
     */
    public void testIteratorRemove() {
        SynchronousQueue q = new SynchronousQueue();
        Iterator it = q.iterator();
        try {
            it.remove();
            shouldThrow();
        } catch (IllegalStateException success) {}
    }

    /**
     * toString returns a non-null string
     */
    public void testToString() {
        SynchronousQueue q = new SynchronousQueue();
        String s = q.toString();
        assertNotNull(s);
    }


    /**
     * offer transfers elements across Executor tasks
     */
    public void testOfferInExecutor() {
        final SynchronousQueue q = new SynchronousQueue();
        ExecutorService executor = Executors.newFixedThreadPool(2);

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertFalse(q.offer(one));
                assertTrue(q.offer(one, MEDIUM_DELAY_MS, MILLISECONDS));
                assertEquals(0, q.remainingCapacity());
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                Thread.sleep(SMALL_DELAY_MS);
                assertSame(one, q.take());
            }});

        joinPool(executor);
    }

    /**
     * poll retrieves elements across Executor threads
     */
    public void testPollInExecutor() {
        final SynchronousQueue q = new SynchronousQueue();
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertNull(q.poll());
                assertSame(one, q.poll(MEDIUM_DELAY_MS, MILLISECONDS));
                assertTrue(q.isEmpty());
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                Thread.sleep(SHORT_DELAY_MS);
                q.put(one);
            }});

        joinPool(executor);
    }

    /**
     * a deserialized serialized queue is usable
     */
    public void testSerialization() throws Exception {
        SynchronousQueue q = new SynchronousQueue();
        ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
        ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
        out.writeObject(q);
        out.close();

        ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
        ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
        SynchronousQueue r = (SynchronousQueue)in.readObject();
        assertEquals(q.size(), r.size());
        while (!q.isEmpty())
            assertEquals(q.remove(), r.remove());
    }

    /**
     * drainTo(null) throws NPE
     */
    public void testDrainToNull() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.drainTo(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * drainTo(this) throws IAE
     */
    public void testDrainToSelf() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.drainTo(q);
            shouldThrow();
        } catch (IllegalArgumentException success) {}
    }

    /**
     * drainTo(c) of empty queue doesn't transfer elements
     */
    public void testDrainTo() {
        SynchronousQueue q = new SynchronousQueue();
        ArrayList l = new ArrayList();
        q.drainTo(l);
        assertEquals(q.size(), 0);
        assertEquals(l.size(), 0);
    }

    /**
     * drainTo empties queue, unblocking a waiting put.
     */
    public void testDrainToWithActivePut() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                q.put(new Integer(1));
            }});

        t.start();
        ArrayList l = new ArrayList();
        Thread.sleep(SHORT_DELAY_MS);
        q.drainTo(l);
        assertTrue(l.size() <= 1);
        if (l.size() > 0)
            assertEquals(l.get(0), new Integer(1));
        t.join();
        assertTrue(l.size() <= 1);
    }

    /**
     * drainTo(null, n) throws NPE
     */
    public void testDrainToNullN() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.drainTo(null, 0);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * drainTo(this, n) throws IAE
     */
    public void testDrainToSelfN() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            q.drainTo(q, 0);
            shouldThrow();
        } catch (IllegalArgumentException success) {}
    }

    /**
     * drainTo(c, n) empties up to n elements of queue into c
     */
    public void testDrainToN() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t1 = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                q.put(one);
            }});

        Thread t2 = new Thread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                q.put(two);
            }});

        t1.start();
        t2.start();
        ArrayList l = new ArrayList();
        Thread.sleep(SHORT_DELAY_MS);
        q.drainTo(l, 1);
        assertEquals(1, l.size());
        q.drainTo(l, 1);
        assertEquals(2, l.size());
        assertTrue(l.contains(one));
        assertTrue(l.contains(two));
        t1.join();
        t2.join();
    }

}
Revision:	1.28
Committed:	Thu Oct 28 22:20:47 2010 UTC (13 years, 6 months ago) by jsr166
Branch:	MAIN
Changes since 1.27:	+2 -2 lines
Log Message:	typo
#	Content
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
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 static java.util.concurrent.TimeUnit.MILLISECONDS;
13	import java.io.*;
14
15	public class SynchronousQueueTest extends JSR166TestCase {
16
17	public static class Fair extends BlockingQueueTest {
18	protected BlockingQueue emptyCollection() {
19	return new SynchronousQueue(true);
20	}
21	}
22
23	public static class NonFair extends BlockingQueueTest {
24	protected BlockingQueue emptyCollection() {
25	return new SynchronousQueue(false);
26	}
27	}
28
29	public static void main(String[] args) {
30	junit.textui.TestRunner.run(suite());
31	}
32
33	public static Test suite() {
34	return newTestSuite(SynchronousQueueTest.class,
35	new Fair().testSuite(),
36	new NonFair().testSuite());
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());
47	assertFalse(q.offer(zero));
48	}
49
50	/**
51	* A fair SynchronousQueue is both empty and full
52	*/
53	public void testFairEmptyFull() {
54	SynchronousQueue q = new SynchronousQueue(true);
55	assertTrue(q.isEmpty());
56	assertEquals(0, q.size());
57	assertEquals(0, q.remainingCapacity());
58	assertFalse(q.offer(zero));
59	}
60
61	/**
62	* offer(null) throws NPE
63	*/
64	public void testOfferNull() {
65	try {
66	SynchronousQueue q = new SynchronousQueue();
67	q.offer(null);
68	shouldThrow();
69	} catch (NullPointerException success) {}
70	}
71
72	/**
73	* add(null) throws NPE
74	*/
75	public void testAddNull() {
76	try {
77	SynchronousQueue q = new SynchronousQueue();
78	q.add(null);
79	shouldThrow();
80	} catch (NullPointerException success) {}
81	}
82
83	/**
84	* offer fails if no active taker
85	*/
86	public void testOffer() {
87	SynchronousQueue q = new SynchronousQueue();
88	assertFalse(q.offer(one));
89	}
90
91	/**
92	* add throws ISE if no active taker
93	*/
94	public void testAdd() {
95	try {
96	SynchronousQueue q = new SynchronousQueue();
97	assertEquals(0, q.remainingCapacity());
98	q.add(one);
99	shouldThrow();
100	} catch (IllegalStateException success) {}
101	}
102
103	/**
104	* addAll(null) throws NPE
105	*/
106	public void testAddAll1() {
107	try {
108	SynchronousQueue q = new SynchronousQueue();
109	q.addAll(null);
110	shouldThrow();
111	} catch (NullPointerException success) {}
112	}
113
114	/**
115	* addAll(this) throws IAE
116	*/
117	public void testAddAllSelf() {
118	try {
119	SynchronousQueue q = new SynchronousQueue();
120	q.addAll(q);
121	shouldThrow();
122	} catch (IllegalArgumentException success) {}
123	}
124
125	/**
126	* addAll of a collection with null elements throws NPE
127	*/
128	public void testAddAll2() {
129	try {
130	SynchronousQueue q = new SynchronousQueue();
131	Integer[] ints = new Integer[1];
132	q.addAll(Arrays.asList(ints));
133	shouldThrow();
134	} catch (NullPointerException success) {}
135	}
136
137	/**
138	* addAll throws ISE if no active taker
139	*/
140	public void testAddAll4() {
141	try {
142	SynchronousQueue q = new SynchronousQueue();
143	Integer[] ints = new Integer[1];
144	for (int i = 0; i < 1; ++i)
145	ints[i] = new Integer(i);
146	q.addAll(Arrays.asList(ints));
147	shouldThrow();
148	} catch (IllegalStateException success) {}
149	}
150
151	/**
152	* put(null) throws NPE
153	*/
154	public void testPutNull() throws InterruptedException {
155	try {
156	SynchronousQueue q = new SynchronousQueue();
157	q.put(null);
158	shouldThrow();
159	} catch (NullPointerException success) {}
160	}
161
162	/**
163	* put blocks interruptibly if no active taker
164	*/
165	public void testBlockingPut() throws InterruptedException {
166	Thread t = new Thread(new CheckedInterruptedRunnable() {
167	public void realRun() throws InterruptedException {
168	SynchronousQueue q = new SynchronousQueue();
169	q.put(zero);
170	}});
171
172	t.start();
173	Thread.sleep(SHORT_DELAY_MS);
174	t.interrupt();
175	t.join();
176	}
177
178	/**
179	* put blocks waiting for take
180	*/
181	public void testPutWithTake() throws InterruptedException {
182	final SynchronousQueue q = new SynchronousQueue();
183	Thread t = new Thread(new CheckedRunnable() {
184	public void realRun() throws InterruptedException {
185	int added = 0;
186	try {
187	while (true) {
188	q.put(added);
189	++added;
190	}
191	} catch (InterruptedException success) {
192	assertEquals(1, added);
193	}
194	}});
195
196	t.start();
197	Thread.sleep(SHORT_DELAY_MS);
198	assertEquals(0, q.take());
199	Thread.sleep(SHORT_DELAY_MS);
200	t.interrupt();
201	t.join();
202	}
203
204	/**
205	* timed offer times out if elements not taken
206	*/
207	public void testTimedOffer() throws InterruptedException {
208	final SynchronousQueue q = new SynchronousQueue();
209	Thread t = new Thread(new CheckedInterruptedRunnable() {
210	public void realRun() throws InterruptedException {
211	assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
212	q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
213	}});
214
215	t.start();
216	Thread.sleep(SMALL_DELAY_MS);
217	t.interrupt();
218	t.join();
219	}
220
221	/**
222	* put blocks interruptibly if no active taker
223	*/
224	public void testFairBlockingPut() throws InterruptedException {
225	Thread t = new Thread(new CheckedInterruptedRunnable() {
226	public void realRun() throws InterruptedException {
227	SynchronousQueue q = new SynchronousQueue(true);
228	q.put(zero);
229	}});
230
231	t.start();
232	Thread.sleep(SHORT_DELAY_MS);
233	t.interrupt();
234	t.join();
235	}
236
237	/**
238	* put blocks waiting for take
239	*/
240	public void testFairPutWithTake() throws InterruptedException {
241	final SynchronousQueue q = new SynchronousQueue(true);
242	Thread t = new Thread(new CheckedRunnable() {
243	public void realRun() throws InterruptedException {
244	int added = 0;
245	try {
246	while (true) {
247	q.put(added);
248	++added;
249	}
250	} catch (InterruptedException success) {
251	assertEquals(1, added);
252	}
253	}});
254
255	t.start();
256	Thread.sleep(SHORT_DELAY_MS);
257	assertEquals(0, q.take());
258	Thread.sleep(SHORT_DELAY_MS);
259	t.interrupt();
260	t.join();
261	}
262
263	/**
264	* timed offer times out if elements not taken
265	*/
266	public void testFairTimedOffer() throws InterruptedException {
267	final SynchronousQueue q = new SynchronousQueue(true);
268	Thread t = new Thread(new CheckedInterruptedRunnable() {
269	public void realRun() throws InterruptedException {
270	assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
271	q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
272	}});
273
274	t.start();
275	Thread.sleep(SMALL_DELAY_MS);
276	t.interrupt();
277	t.join();
278	}
279
280
281	/**
282	* take blocks interruptibly when empty
283	*/
284	public void testFairTakeFromEmpty() throws InterruptedException {
285	final SynchronousQueue q = new SynchronousQueue(true);
286	Thread t = new Thread(new CheckedInterruptedRunnable() {
287	public void realRun() throws InterruptedException {
288	q.take();
289	}});
290
291	t.start();
292	Thread.sleep(SHORT_DELAY_MS);
293	t.interrupt();
294	t.join();
295	}
296
297	/**
298	* poll fails unless active taker
299	*/
300	public void testPoll() {
301	SynchronousQueue q = new SynchronousQueue();
302	assertNull(q.poll());
303	}
304
305	/**
306	* timed poll with zero timeout times out if no active taker
307	*/
308	public void testTimedPoll0() throws InterruptedException {
309	SynchronousQueue q = new SynchronousQueue();
310	assertNull(q.poll(0, MILLISECONDS));
311	}
312
313	/**
314	* timed poll with nonzero timeout times out if no active taker
315	*/
316	public void testTimedPoll() throws InterruptedException {
317	SynchronousQueue q = new SynchronousQueue();
318	assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
319	}
320
321	/**
322	* Interrupted timed poll throws InterruptedException instead of
323	* returning timeout status
324	*/
325	public void testInterruptedTimedPoll() throws InterruptedException {
326	final SynchronousQueue q = new SynchronousQueue();
327	Thread t = new Thread(new CheckedInterruptedRunnable() {
328	public void realRun() throws InterruptedException {
329	q.poll(SMALL_DELAY_MS, MILLISECONDS);
330	}});
331
332	t.start();
333	Thread.sleep(SHORT_DELAY_MS);
334	t.interrupt();
335	t.join();
336	}
337
338	/**
339	* Interrupted timed poll throws InterruptedException instead of
340	* returning timeout status
341	*/
342	public void testFairInterruptedTimedPoll() throws InterruptedException {
343	Thread t = new Thread(new CheckedInterruptedRunnable() {
344	public void realRun() throws InterruptedException {
345	SynchronousQueue q = new SynchronousQueue(true);
346	q.poll(SMALL_DELAY_MS, MILLISECONDS);
347	}});
348
349	t.start();
350	Thread.sleep(SHORT_DELAY_MS);
351	t.interrupt();
352	t.join();
353	}
354
355	/**
356	* timed poll before a delayed offer fails; after offer succeeds;
357	* on interruption throws
358	*/
359	public void testFairTimedPollWithOffer() throws InterruptedException {
360	final SynchronousQueue q = new SynchronousQueue(true);
361	Thread t = new Thread(new CheckedRunnable() {
362	public void realRun() throws InterruptedException {
363	try {
364	assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
365	assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
366	q.poll(LONG_DELAY_MS, MILLISECONDS);
367	threadShouldThrow();
368	} catch (InterruptedException success) {}
369	}});
370
371	t.start();
372	Thread.sleep(SMALL_DELAY_MS);
373	assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
374	t.interrupt();
375	t.join();
376	}
377
378
379	/**
380	* peek returns null
381	*/
382	public void testPeek() {
383	SynchronousQueue q = new SynchronousQueue();
384	assertNull(q.peek());
385	}
386
387	/**
388	* element throws NSEE
389	*/
390	public void testElement() {
391	SynchronousQueue q = new SynchronousQueue();
392	try {
393	q.element();
394	shouldThrow();
395	} catch (NoSuchElementException success) {}
396	}
397
398	/**
399	* remove throws NSEE if no active taker
400	*/
401	public void testRemove() {
402	SynchronousQueue q = new SynchronousQueue();
403	try {
404	q.remove();
405	shouldThrow();
406	} catch (NoSuchElementException success) {}
407	}
408
409	/**
410	* remove(x) returns false
411	*/
412	public void testRemoveElement() {
413	SynchronousQueue q = new SynchronousQueue();
414	assertFalse(q.remove(zero));
415	assertTrue(q.isEmpty());
416	}
417
418	/**
419	* contains returns false
420	*/
421	public void testContains() {
422	SynchronousQueue q = new SynchronousQueue();
423	assertFalse(q.contains(zero));
424	}
425
426	/**
427	* clear ensures isEmpty
428	*/
429	public void testClear() {
430	SynchronousQueue q = new SynchronousQueue();
431	q.clear();
432	assertTrue(q.isEmpty());
433	}
434
435	/**
436	* containsAll returns false unless empty
437	*/
438	public void testContainsAll() {
439	SynchronousQueue q = new SynchronousQueue();
440	Integer[] empty = new Integer[0];
441	assertTrue(q.containsAll(Arrays.asList(empty)));
442	Integer[] ints = new Integer[1]; ints[0] = zero;
443	assertFalse(q.containsAll(Arrays.asList(ints)));
444	}
445
446	/**
447	* retainAll returns false
448	*/
449	public void testRetainAll() {
450	SynchronousQueue q = new SynchronousQueue();
451	Integer[] empty = new Integer[0];
452	assertFalse(q.retainAll(Arrays.asList(empty)));
453	Integer[] ints = new Integer[1]; ints[0] = zero;
454	assertFalse(q.retainAll(Arrays.asList(ints)));
455	}
456
457	/**
458	* removeAll returns false
459	*/
460	public void testRemoveAll() {
461	SynchronousQueue q = new SynchronousQueue();
462	Integer[] empty = new Integer[0];
463	assertFalse(q.removeAll(Arrays.asList(empty)));
464	Integer[] ints = new Integer[1]; ints[0] = zero;
465	assertFalse(q.containsAll(Arrays.asList(ints)));
466	}
467
468
469	/**
470	* toArray is empty
471	*/
472	public void testToArray() {
473	SynchronousQueue q = new SynchronousQueue();
474	Object[] o = q.toArray();
475	assertEquals(o.length, 0);
476	}
477
478	/**
479	* toArray(a) is nulled at position 0
480	*/
481	public void testToArray2() {
482	SynchronousQueue q = new SynchronousQueue();
483	Integer[] ints = new Integer[1];
484	assertNull(ints[0]);
485	}
486
487	/**
488	* toArray(null) throws NPE
489	*/
490	public void testToArray_BadArg() {
491	SynchronousQueue q = new SynchronousQueue();
492	try {
493	Object o[] = q.toArray(null);
494	shouldThrow();
495	} catch (NullPointerException success) {}
496	}
497
498
499	/**
500	* iterator does not traverse any elements
501	*/
502	public void testIterator() {
503	SynchronousQueue q = new SynchronousQueue();
504	Iterator it = q.iterator();
505	assertFalse(it.hasNext());
506	try {
507	Object x = it.next();
508	shouldThrow();
509	} catch (NoSuchElementException success) {}
510	}
511
512	/**
513	* iterator remove throws ISE
514	*/
515	public void testIteratorRemove() {
516	SynchronousQueue q = new SynchronousQueue();
517	Iterator it = q.iterator();
518	try {
519	it.remove();
520	shouldThrow();
521	} catch (IllegalStateException success) {}
522	}
523
524	/**
525	* toString returns a non-null string
526	*/
527	public void testToString() {
528	SynchronousQueue q = new SynchronousQueue();
529	String s = q.toString();
530	assertNotNull(s);
531	}
532
533
534	/**
535	* offer transfers elements across Executor tasks
536	*/
537	public void testOfferInExecutor() {
538	final SynchronousQueue q = new SynchronousQueue();
539	ExecutorService executor = Executors.newFixedThreadPool(2);
540
541	executor.execute(new CheckedRunnable() {
542	public void realRun() throws InterruptedException {
543	assertFalse(q.offer(one));
544	assertTrue(q.offer(one, MEDIUM_DELAY_MS, MILLISECONDS));
545	assertEquals(0, q.remainingCapacity());
546	}});
547
548	executor.execute(new CheckedRunnable() {
549	public void realRun() throws InterruptedException {
550	Thread.sleep(SMALL_DELAY_MS);
551	assertSame(one, q.take());
552	}});
553
554	joinPool(executor);
555	}
556
557	/**
558	* poll retrieves elements across Executor threads
559	*/
560	public void testPollInExecutor() {
561	final SynchronousQueue q = new SynchronousQueue();
562	ExecutorService executor = Executors.newFixedThreadPool(2);
563	executor.execute(new CheckedRunnable() {
564	public void realRun() throws InterruptedException {
565	assertNull(q.poll());
566	assertSame(one, q.poll(MEDIUM_DELAY_MS, MILLISECONDS));
567	assertTrue(q.isEmpty());
568	}});
569
570	executor.execute(new CheckedRunnable() {
571	public void realRun() throws InterruptedException {
572	Thread.sleep(SHORT_DELAY_MS);
573	q.put(one);
574	}});
575
576	joinPool(executor);
577	}
578
579	/**
580	* a deserialized serialized queue is usable
581	*/
582	public void testSerialization() throws Exception {
583	SynchronousQueue q = new SynchronousQueue();
584	ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
585	ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
586	out.writeObject(q);
587	out.close();
588
589	ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
590	ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
591	SynchronousQueue r = (SynchronousQueue)in.readObject();
592	assertEquals(q.size(), r.size());
593	while (!q.isEmpty())
594	assertEquals(q.remove(), r.remove());
595	}
596
597	/**
598	* drainTo(null) throws NPE
599	*/
600	public void testDrainToNull() {
601	SynchronousQueue q = new SynchronousQueue();
602	try {
603	q.drainTo(null);
604	shouldThrow();
605	} catch (NullPointerException success) {}
606	}
607
608	/**
609	* drainTo(this) throws IAE
610	*/
611	public void testDrainToSelf() {
612	SynchronousQueue q = new SynchronousQueue();
613	try {
614	q.drainTo(q);
615	shouldThrow();
616	} catch (IllegalArgumentException success) {}
617	}
618
619	/**
620	* drainTo(c) of empty queue doesn't transfer elements
621	*/
622	public void testDrainTo() {
623	SynchronousQueue q = new SynchronousQueue();
624	ArrayList l = new ArrayList();
625	q.drainTo(l);
626	assertEquals(q.size(), 0);
627	assertEquals(l.size(), 0);
628	}
629
630	/**
631	* drainTo empties queue, unblocking a waiting put.
632	*/
633	public void testDrainToWithActivePut() throws InterruptedException {
634	final SynchronousQueue q = new SynchronousQueue();
635	Thread t = new Thread(new CheckedRunnable() {
636	public void realRun() throws InterruptedException {
637	q.put(new Integer(1));
638	}});
639
640	t.start();
641	ArrayList l = new ArrayList();
642	Thread.sleep(SHORT_DELAY_MS);
643	q.drainTo(l);
644	assertTrue(l.size() <= 1);
645	if (l.size() > 0)
646	assertEquals(l.get(0), new Integer(1));
647	t.join();
648	assertTrue(l.size() <= 1);
649	}
650
651	/**
652	* drainTo(null, n) throws NPE
653	*/
654	public void testDrainToNullN() {
655	SynchronousQueue q = new SynchronousQueue();
656	try {
657	q.drainTo(null, 0);
658	shouldThrow();
659	} catch (NullPointerException success) {}
660	}
661
662	/**
663	* drainTo(this, n) throws IAE
664	*/
665	public void testDrainToSelfN() {
666	SynchronousQueue q = new SynchronousQueue();
667	try {
668	q.drainTo(q, 0);
669	shouldThrow();
670	} catch (IllegalArgumentException success) {}
671	}
672
673	/**
674	* drainTo(c, n) empties up to n elements of queue into c
675	*/
676	public void testDrainToN() throws InterruptedException {
677	final SynchronousQueue q = new SynchronousQueue();
678	Thread t1 = new Thread(new CheckedRunnable() {
679	public void realRun() throws InterruptedException {
680	q.put(one);
681	}});
682
683	Thread t2 = new Thread(new CheckedRunnable() {
684	public void realRun() throws InterruptedException {
685	q.put(two);
686	}});
687
688	t1.start();
689	t2.start();
690	ArrayList l = new ArrayList();
691	Thread.sleep(SHORT_DELAY_MS);
692	q.drainTo(l, 1);
693	assertEquals(1, l.size());
694	q.drainTo(l, 1);
695	assertEquals(2, l.size());
696	assertTrue(l.contains(one));
697	assertTrue(l.contains(two));
698	t1.join();
699	t2.join();
700	}
701
702	}