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/publicdomain/zero/1.0/
 * 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());
    }

    /**
     * Any SynchronousQueue is both empty and full
     */
    public void testEmptyFull()      { testEmptyFull(false); }
    public void testEmptyFull_fair() { testEmptyFull(true); }
    public void testEmptyFull(boolean fair) {
        final SynchronousQueue q = new SynchronousQueue(fair);
        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()      { testBlockingPut(false); }
    public void testBlockingPut_fair() { testBlockingPut(true); }
    public void testBlockingPut(boolean fair) {
        final SynchronousQueue q = new SynchronousQueue(fair);
        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
        Thread t = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                Thread.currentThread().interrupt();
                try {
                    q.put(99);
                    shouldThrow();
                } catch (InterruptedException success) {}
                assertFalse(Thread.interrupted());

                pleaseInterrupt.countDown();
                try {
                    q.put(99);
                    shouldThrow();
                } catch (InterruptedException success) {}
                assertFalse(Thread.interrupted());
            }});

        await(pleaseInterrupt);
        assertThreadStaysAlive(t);
        t.interrupt();
        awaitTermination(t);
        assertEquals(0, q.remainingCapacity());
    }

    /**
     * put blocks interruptibly waiting for take
     */
    public void testPutWithTake()      { testPutWithTake(false); }
    public void testPutWithTake_fair() { testPutWithTake(true); }
    public void testPutWithTake(boolean fair) {
        final SynchronousQueue q = new SynchronousQueue(fair);
        final CountDownLatch pleaseTake = new CountDownLatch(1);
        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
        Thread t = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                pleaseTake.countDown();
                q.put(one);

                pleaseInterrupt.countDown();
                try {
                    q.put(99);
                    shouldThrow();
                } catch (InterruptedException success) {}
                assertFalse(Thread.interrupted());
            }});

        await(pleaseTake);
        assertEquals(q.remainingCapacity(), 0);
        try { assertSame(one, q.take()); }
        catch (InterruptedException e) { threadUnexpectedException(e); }

        await(pleaseInterrupt);
        assertThreadStaysAlive(t);
        t.interrupt();
        awaitTermination(t);
        assertEquals(q.remainingCapacity(), 0);
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testTimedOffer()      { testTimedOffer(false); }
    public void testTimedOffer_fair() { testTimedOffer(true); }
    public void testTimedOffer(boolean fair) {
        final SynchronousQueue q = new SynchronousQueue(fair);
        final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
        Thread t = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                long startTime = System.nanoTime();
                assertFalse(q.offer(new Object(), timeoutMillis(), MILLISECONDS));
                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
                pleaseInterrupt.countDown();
                try {
                    q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
                    shouldThrow();
                } catch (InterruptedException success) {}
            }});

        await(pleaseInterrupt);
        assertThreadStaysAlive(t);
        t.interrupt();
        awaitTermination(t);
    }

    /**
     * poll return null if no active putter
     */
    public void testPoll() {
        SynchronousQueue q = new SynchronousQueue();
        assertNull(q.poll());
    }

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

    /**
     * timed poll with nonzero timeout times out if no active putter
     */
    public void testTimedPoll() throws InterruptedException {
        SynchronousQueue q = new SynchronousQueue();
        long startTime = System.nanoTime();
        assertNull(q.poll(timeoutMillis(), MILLISECONDS));
        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
    }

    /**
     * timed poll before a delayed offer times out, returning null;
     * after offer succeeds; on interruption throws
     */
    public void testFairTimedPollWithOffer() throws InterruptedException {
        final SynchronousQueue q = new SynchronousQueue(true);
        final CountDownLatch pleaseOffer = new CountDownLatch(1);
        Thread t = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                long t0 = System.nanoTime();
                assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
                assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS);

                pleaseOffer.countDown();
                t0 = System.nanoTime();
                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
                assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS);

                t0 = System.nanoTime();
                try {
                    q.poll(LONG_DELAY_MS, MILLISECONDS);
                    shouldThrow();
                } catch (InterruptedException success) {}
                assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS);
            }});

        assertTrue(pleaseOffer.await(MEDIUM_DELAY_MS, MILLISECONDS));
        long t0 = System.nanoTime();
        assertTrue(q.offer(zero, LONG_DELAY_MS, MILLISECONDS));
        assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS);

        t.interrupt();
        awaitTermination(t, MEDIUM_DELAY_MS);
    }

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

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

    /**
     * remove() throws NSEE if no active putter
     */
    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);
        final CheckedBarrier threadsStarted = new CheckedBarrier(2);

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

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

        joinPool(executor);
    }

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

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadsStarted.await();
                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 = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                q.put(one);
            }});

        ArrayList l = new ArrayList();
        long startTime = System.nanoTime();
        while (l.isEmpty()) {
            q.drainTo(l);
            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
                fail("timed out");
            Thread.yield();
        }
        assertTrue(l.size() == 1);
        assertSame(one, l.get(0));
        awaitTermination(t);
    }

    /**
     * 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 = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                q.put(one);
            }});

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

        ArrayList l = new ArrayList();
        delay(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));
        awaitTermination(t1);
        awaitTermination(t2);
    }

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