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(SynchronousQueue q) {
        assertTrue(q.isEmpty());
        assertEquals(0, q.size());
        assertEquals(0, q.remainingCapacity());
        assertFalse(q.offer(zero));
    }

    /**
     * A non-fair SynchronousQueue is both empty and full
     */
    public void testEmptyFull() {
        testEmptyFull(new SynchronousQueue());
    }

    /**
     * A fair SynchronousQueue is both empty and full
     */
    public void testFairEmptyFull() {
        testEmptyFull(new SynchronousQueue(true));
    }

    /**
     * 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();
        delay(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();
        delay(SHORT_DELAY_MS);
        assertEquals(0, q.take());
        delay(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testTimedOffer(final SynchronousQueue q)
            throws InterruptedException {
        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);
        t.interrupt();
        awaitTermination(t);
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testTimedOffer() throws InterruptedException {
        testTimedOffer(new SynchronousQueue());
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testFairTimedOffer() throws InterruptedException {
        testTimedOffer(new SynchronousQueue(true));
    }

    /**
     * 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();
        delay(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();
        delay(SHORT_DELAY_MS);
        assertEquals(0, q.take());
        delay(SHORT_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();
        delay(SHORT_DELAY_MS);
        t.interrupt();
        t.join();
    }

    /**
     * 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 t0 = System.nanoTime();
        assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
        assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS);
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testInterruptedTimedPoll(final SynchronousQueue q)
            throws InterruptedException {
        final CountDownLatch threadStarted = new CountDownLatch(1);
        Thread t = newStartedThread(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                long t0 = System.nanoTime();
                threadStarted.countDown();
                try {
                    q.poll(LONG_DELAY_MS, MILLISECONDS);
                    shouldThrow();
                } catch (InterruptedException success) {}
                assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS);
                assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS);
            }});

        threadStarted.await();
        delay(SHORT_DELAY_MS);
        t.interrupt();
        awaitTermination(t, MEDIUM_DELAY_MS);
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testInterruptedTimedPoll() throws InterruptedException {
        testInterruptedTimedPoll(new SynchronousQueue());
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testFairInterruptedTimedPoll() throws InterruptedException {
        testInterruptedTimedPoll(new SynchronousQueue(true));
    }

    /**
     * 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);

        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 {
                delay(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 {
                delay(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();
        delay(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();
        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));
        t1.join();
        t2.join();
    }

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