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
#	User	Rev	Content
1	dl	1.1	/*
2	dl	1.7	* 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	jsr166	1.31	* http://creativecommons.org/publicdomain/zero/1.0/
5	jsr166	1.10	* Other contributors include Andrew Wright, Jeffrey Hayes,
6			* Pat Fisher, Mike Judd.
7	dl	1.1	*/
8
9			import junit.framework.*;
10			import java.util.*;
11			import java.util.concurrent.*;
12	jsr166	1.13	import static java.util.concurrent.TimeUnit.MILLISECONDS;
13	dl	1.2	import java.io.*;
14	dl	1.1
15	dl	1.3	public class SynchronousQueueTest extends JSR166TestCase {
16	dl	1.1
17	jsr166	1.25	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	dl	1.1	public static void main(String[] args) {
30	jsr166	1.21	junit.textui.TestRunner.run(suite());
31	dl	1.1	}
32
33			public static Test suite() {
34	jsr166	1.25	return newTestSuite(SynchronousQueueTest.class,
35			new Fair().testSuite(),
36			new NonFair().testSuite());
37	dl	1.1	}
38
39	dl	1.4	/**
40	jsr166	1.29	* Any SynchronousQueue is both empty and full
41	dl	1.4	*/
42	jsr166	1.35	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	dl	1.1	assertTrue(q.isEmpty());
47	jsr166	1.14	assertEquals(0, q.size());
48	dl	1.1	assertEquals(0, q.remainingCapacity());
49	dl	1.5	assertFalse(q.offer(zero));
50	dl	1.1	}
51
52	dl	1.4	/**
53	dl	1.5	* offer(null) throws NPE
54	dl	1.4	*/
55			public void testOfferNull() {
56	jsr166	1.14	try {
57	dl	1.1	SynchronousQueue q = new SynchronousQueue();
58			q.offer(null);
59	dl	1.4	shouldThrow();
60	jsr166	1.13	} catch (NullPointerException success) {}
61	dl	1.1	}
62
63	dl	1.4	/**
64	dl	1.6	* add(null) throws NPE
65			*/
66			public void testAddNull() {
67	jsr166	1.14	try {
68	dl	1.6	SynchronousQueue q = new SynchronousQueue();
69			q.add(null);
70			shouldThrow();
71	jsr166	1.13	} catch (NullPointerException success) {}
72	dl	1.6	}
73
74			/**
75	dl	1.5	* offer fails if no active taker
76	dl	1.4	*/
77			public void testOffer() {
78	dl	1.1	SynchronousQueue q = new SynchronousQueue();
79	dl	1.5	assertFalse(q.offer(one));
80	dl	1.1	}
81
82	dl	1.4	/**
83	dl	1.5	* add throws ISE if no active taker
84	dl	1.4	*/
85			public void testAdd() {
86	jsr166	1.14	try {
87	dl	1.1	SynchronousQueue q = new SynchronousQueue();
88			assertEquals(0, q.remainingCapacity());
89	dl	1.5	q.add(one);
90			shouldThrow();
91	jsr166	1.13	} catch (IllegalStateException success) {}
92	dl	1.1	}
93
94	dl	1.4	/**
95	dl	1.5	* addAll(null) throws NPE
96	dl	1.4	*/
97			public void testAddAll1() {
98	dl	1.1	try {
99			SynchronousQueue q = new SynchronousQueue();
100			q.addAll(null);
101	dl	1.4	shouldThrow();
102	jsr166	1.13	} catch (NullPointerException success) {}
103	dl	1.1	}
104	dl	1.6
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	jsr166	1.13	} catch (IllegalArgumentException success) {}
114	dl	1.6	}
115
116	dl	1.4	/**
117	dl	1.5	* addAll of a collection with null elements throws NPE
118	dl	1.4	*/
119			public void testAddAll2() {
120	dl	1.1	try {
121			SynchronousQueue q = new SynchronousQueue();
122	dl	1.3	Integer[] ints = new Integer[1];
123	dl	1.1	q.addAll(Arrays.asList(ints));
124	dl	1.4	shouldThrow();
125	jsr166	1.13	} catch (NullPointerException success) {}
126	dl	1.1	}
127	jsr166	1.22
128	dl	1.4	/**
129	dl	1.5	* addAll throws ISE if no active taker
130	dl	1.4	*/
131			public void testAddAll4() {
132	dl	1.1	try {
133			SynchronousQueue q = new SynchronousQueue();
134	dl	1.3	Integer[] ints = new Integer[1];
135			for (int i = 0; i < 1; ++i)
136	dl	1.1	ints[i] = new Integer(i);
137			q.addAll(Arrays.asList(ints));
138	dl	1.4	shouldThrow();
139	jsr166	1.13	} catch (IllegalStateException success) {}
140	dl	1.1	}
141
142	dl	1.4	/**
143	dl	1.5	* put(null) throws NPE
144	dl	1.4	*/
145	jsr166	1.13	public void testPutNull() throws InterruptedException {
146	jsr166	1.14	try {
147	dl	1.1	SynchronousQueue q = new SynchronousQueue();
148			q.put(null);
149	dl	1.4	shouldThrow();
150	jsr166	1.13	} catch (NullPointerException success) {}
151	jsr166	1.32	}
152	dl	1.1
153	dl	1.4	/**
154	dl	1.5	* put blocks interruptibly if no active taker
155	dl	1.4	*/
156	jsr166	1.35	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	jsr166	1.14	public void realRun() throws InterruptedException {
163	jsr166	1.35	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	jsr166	1.13	}});
177
178	jsr166	1.35	await(pleaseInterrupt);
179			assertThreadStaysAlive(t);
180	jsr166	1.13	t.interrupt();
181	jsr166	1.35	awaitTermination(t);
182			assertEquals(0, q.remainingCapacity());
183	dl	1.1	}
184
185	dl	1.4	/**
186	jsr166	1.35	* put blocks interruptibly waiting for take
187	dl	1.4	*/
188	jsr166	1.35	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	jsr166	1.14	public void realRun() throws InterruptedException {
196	jsr166	1.35	pleaseTake.countDown();
197			q.put(one);
198
199			pleaseInterrupt.countDown();
200	jsr166	1.13	try {
201	jsr166	1.35	q.put(99);
202			shouldThrow();
203			} catch (InterruptedException success) {}
204			assertFalse(Thread.interrupted());
205	jsr166	1.13	}});
206
207	jsr166	1.35	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	jsr166	1.13	t.interrupt();
215	jsr166	1.35	awaitTermination(t);
216			assertEquals(q.remainingCapacity(), 0);
217	dl	1.1	}
218
219	dl	1.4	/**
220	dl	1.5	* timed offer times out if elements not taken
221	dl	1.4	*/
222	jsr166	1.35	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	jsr166	1.29	final CountDownLatch pleaseInterrupt = new CountDownLatch(1);
227			Thread t = newStartedThread(new CheckedRunnable() {
228	jsr166	1.14	public void realRun() throws InterruptedException {
229	jsr166	1.34	long startTime = System.nanoTime();
230			assertFalse(q.offer(new Object(), timeoutMillis(), MILLISECONDS));
231			assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
232	jsr166	1.29	pleaseInterrupt.countDown();
233			try {
234	jsr166	1.34	q.offer(new Object(), 2 * LONG_DELAY_MS, MILLISECONDS);
235	jsr166	1.29	shouldThrow();
236			} catch (InterruptedException success) {}
237	jsr166	1.13	}});
238	jsr166	1.10
239	jsr166	1.34	await(pleaseInterrupt);
240	jsr166	1.35	assertThreadStaysAlive(t);
241	jsr166	1.13	t.interrupt();
242	jsr166	1.34	awaitTermination(t);
243	jsr166	1.29	}
244
245			/**
246			* poll return null if no active putter
247	dl	1.4	*/
248			public void testPoll() {
249	dl	1.1	SynchronousQueue q = new SynchronousQueue();
250	jsr166	1.14	assertNull(q.poll());
251	dl	1.1	}
252
253	dl	1.4	/**
254	jsr166	1.29	* timed poll with zero timeout times out if no active putter
255	dl	1.4	*/
256	jsr166	1.13	public void testTimedPoll0() throws InterruptedException {
257			SynchronousQueue q = new SynchronousQueue();
258			assertNull(q.poll(0, MILLISECONDS));
259	dl	1.1	}
260
261	dl	1.4	/**
262	jsr166	1.29	* timed poll with nonzero timeout times out if no active putter
263	dl	1.4	*/
264	jsr166	1.13	public void testTimedPoll() throws InterruptedException {
265			SynchronousQueue q = new SynchronousQueue();
266	jsr166	1.35	long startTime = System.nanoTime();
267			assertNull(q.poll(timeoutMillis(), MILLISECONDS));
268			assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
269	dl	1.8	}
270
271			/**
272	jsr166	1.29	* timed poll before a delayed offer times out, returning null;
273			* after offer succeeds; on interruption throws
274	dl	1.8	*/
275	jsr166	1.13	public void testFairTimedPollWithOffer() throws InterruptedException {
276	dl	1.8	final SynchronousQueue q = new SynchronousQueue(true);
277	jsr166	1.29	final CountDownLatch pleaseOffer = new CountDownLatch(1);
278			Thread t = newStartedThread(new CheckedRunnable() {
279	jsr166	1.14	public void realRun() throws InterruptedException {
280	jsr166	1.29	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	jsr166	1.13	try {
291			q.poll(LONG_DELAY_MS, MILLISECONDS);
292	jsr166	1.29	shouldThrow();
293	jsr166	1.13	} catch (InterruptedException success) {}
294	jsr166	1.29	assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS);
295	jsr166	1.13	}});
296
297	jsr166	1.29	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	jsr166	1.30
302	jsr166	1.13	t.interrupt();
303	jsr166	1.29	awaitTermination(t, MEDIUM_DELAY_MS);
304	jsr166	1.10	}
305	dl	1.1
306	dl	1.4	/**
307	jsr166	1.29	* peek() returns null if no active putter
308	dl	1.4	*/
309			public void testPeek() {
310	dl	1.1	SynchronousQueue q = new SynchronousQueue();
311	jsr166	1.14	assertNull(q.peek());
312	dl	1.1	}
313
314	dl	1.4	/**
315	jsr166	1.29	* element() throws NSEE if no active putter
316	dl	1.4	*/
317			public void testElement() {
318	dl	1.1	SynchronousQueue q = new SynchronousQueue();
319			try {
320			q.element();
321	dl	1.4	shouldThrow();
322	jsr166	1.13	} catch (NoSuchElementException success) {}
323	dl	1.1	}
324
325	dl	1.4	/**
326	jsr166	1.29	* remove() throws NSEE if no active putter
327	dl	1.4	*/
328			public void testRemove() {
329	dl	1.1	SynchronousQueue q = new SynchronousQueue();
330			try {
331			q.remove();
332	dl	1.4	shouldThrow();
333	jsr166	1.16	} catch (NoSuchElementException success) {}
334	dl	1.1	}
335
336	dl	1.4	/**
337	dl	1.5	* remove(x) returns false
338	dl	1.4	*/
339			public void testRemoveElement() {
340	dl	1.1	SynchronousQueue q = new SynchronousQueue();
341	dl	1.5	assertFalse(q.remove(zero));
342	dl	1.2	assertTrue(q.isEmpty());
343	dl	1.1	}
344	jsr166	1.10
345	dl	1.4	/**
346	dl	1.5	* contains returns false
347	dl	1.4	*/
348			public void testContains() {
349	dl	1.1	SynchronousQueue q = new SynchronousQueue();
350	dl	1.5	assertFalse(q.contains(zero));
351	dl	1.1	}
352
353	dl	1.4	/**
354	dl	1.5	* clear ensures isEmpty
355	dl	1.4	*/
356			public void testClear() {
357	dl	1.1	SynchronousQueue q = new SynchronousQueue();
358			q.clear();
359			assertTrue(q.isEmpty());
360			}
361
362	dl	1.4	/**
363	dl	1.5	* containsAll returns false unless empty
364	dl	1.4	*/
365			public void testContainsAll() {
366	dl	1.1	SynchronousQueue q = new SynchronousQueue();
367			Integer[] empty = new Integer[0];
368	dl	1.5	assertTrue(q.containsAll(Arrays.asList(empty)));
369			Integer[] ints = new Integer[1]; ints[0] = zero;
370	dl	1.1	assertFalse(q.containsAll(Arrays.asList(ints)));
371			}
372
373	dl	1.4	/**
374	dl	1.5	* retainAll returns false
375	dl	1.4	*/
376			public void testRetainAll() {
377	dl	1.1	SynchronousQueue q = new SynchronousQueue();
378			Integer[] empty = new Integer[0];
379	dl	1.5	assertFalse(q.retainAll(Arrays.asList(empty)));
380			Integer[] ints = new Integer[1]; ints[0] = zero;
381			assertFalse(q.retainAll(Arrays.asList(ints)));
382	dl	1.1	}
383
384	dl	1.4	/**
385	dl	1.5	* removeAll returns false
386	dl	1.4	*/
387			public void testRemoveAll() {
388	dl	1.1	SynchronousQueue q = new SynchronousQueue();
389			Integer[] empty = new Integer[0];
390	dl	1.5	assertFalse(q.removeAll(Arrays.asList(empty)));
391			Integer[] ints = new Integer[1]; ints[0] = zero;
392	dl	1.1	assertFalse(q.containsAll(Arrays.asList(ints)));
393			}
394
395	dl	1.4	/**
396	dl	1.5	* toArray is empty
397	dl	1.4	*/
398			public void testToArray() {
399	dl	1.1	SynchronousQueue q = new SynchronousQueue();
400	jsr166	1.14	Object[] o = q.toArray();
401	dl	1.1	assertEquals(o.length, 0);
402			}
403
404	dl	1.4	/**
405	dl	1.5	* toArray(a) is nulled at position 0
406	dl	1.4	*/
407			public void testToArray2() {
408	dl	1.1	SynchronousQueue q = new SynchronousQueue();
409	jsr166	1.14	Integer[] ints = new Integer[1];
410	dl	1.1	assertNull(ints[0]);
411			}
412	jsr166	1.10
413	dl	1.4	/**
414	dl	1.6	* toArray(null) throws NPE
415			*/
416			public void testToArray_BadArg() {
417	jsr166	1.18	SynchronousQueue q = new SynchronousQueue();
418	jsr166	1.14	try {
419			Object o[] = q.toArray(null);
420			shouldThrow();
421			} catch (NullPointerException success) {}
422	dl	1.6	}
423
424			/**
425	dl	1.5	* iterator does not traverse any elements
426	dl	1.4	*/
427			public void testIterator() {
428	dl	1.1	SynchronousQueue q = new SynchronousQueue();
429	jsr166	1.14	Iterator it = q.iterator();
430	dl	1.1	assertFalse(it.hasNext());
431			try {
432			Object x = it.next();
433	dl	1.4	shouldThrow();
434	jsr166	1.13	} catch (NoSuchElementException success) {}
435	dl	1.1	}
436
437	dl	1.4	/**
438	dl	1.5	* iterator remove throws ISE
439	dl	1.4	*/
440			public void testIteratorRemove() {
441	dl	1.1	SynchronousQueue q = new SynchronousQueue();
442	jsr166	1.14	Iterator it = q.iterator();
443	dl	1.1	try {
444			it.remove();
445	dl	1.4	shouldThrow();
446	jsr166	1.13	} catch (IllegalStateException success) {}
447	dl	1.1	}
448
449	dl	1.4	/**
450	dl	1.5	* toString returns a non-null string
451	dl	1.4	*/
452			public void testToString() {
453	dl	1.1	SynchronousQueue q = new SynchronousQueue();
454			String s = q.toString();
455	dl	1.5	assertNotNull(s);
456	jsr166	1.10	}
457	dl	1.1
458	dl	1.4	/**
459	dl	1.5	* offer transfers elements across Executor tasks
460	dl	1.4	*/
461	dl	1.1	public void testOfferInExecutor() {
462			final SynchronousQueue q = new SynchronousQueue();
463			ExecutorService executor = Executors.newFixedThreadPool(2);
464	jsr166	1.35	final CheckedBarrier threadsStarted = new CheckedBarrier(2);
465	dl	1.1
466	jsr166	1.13	executor.execute(new CheckedRunnable() {
467	jsr166	1.14	public void realRun() throws InterruptedException {
468	jsr166	1.19	assertFalse(q.offer(one));
469	jsr166	1.35	threadsStarted.await();
470			assertTrue(q.offer(one, LONG_DELAY_MS, MILLISECONDS));
471	jsr166	1.19	assertEquals(0, q.remainingCapacity());
472	jsr166	1.13	}});
473
474			executor.execute(new CheckedRunnable() {
475	jsr166	1.14	public void realRun() throws InterruptedException {
476	jsr166	1.35	threadsStarted.await();
477	jsr166	1.19	assertSame(one, q.take());
478	jsr166	1.13	}});
479	jsr166	1.10
480	dl	1.3	joinPool(executor);
481	dl	1.1	}
482
483	dl	1.4	/**
484	jsr166	1.35	* timed poll retrieves elements across Executor threads
485	dl	1.4	*/
486	dl	1.1	public void testPollInExecutor() {
487			final SynchronousQueue q = new SynchronousQueue();
488	jsr166	1.35	final CheckedBarrier threadsStarted = new CheckedBarrier(2);
489	dl	1.1	ExecutorService executor = Executors.newFixedThreadPool(2);
490	jsr166	1.13	executor.execute(new CheckedRunnable() {
491	jsr166	1.14	public void realRun() throws InterruptedException {
492	jsr166	1.19	assertNull(q.poll());
493	jsr166	1.35	threadsStarted.await();
494			assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
495	jsr166	1.19	assertTrue(q.isEmpty());
496	jsr166	1.13	}});
497
498			executor.execute(new CheckedRunnable() {
499	jsr166	1.14	public void realRun() throws InterruptedException {
500	jsr166	1.35	threadsStarted.await();
501	jsr166	1.19	q.put(one);
502	jsr166	1.13	}});
503	jsr166	1.10
504	dl	1.3	joinPool(executor);
505	dl	1.2	}
506
507	dl	1.4	/**
508	dl	1.5	* a deserialized serialized queue is usable
509	dl	1.4	*/
510	jsr166	1.13	public void testSerialization() throws Exception {
511	dl	1.2	SynchronousQueue q = new SynchronousQueue();
512	jsr166	1.13	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	dl	1.1	}
524	dl	1.6
525			/**
526			* drainTo(null) throws NPE
527	jsr166	1.10	*/
528	dl	1.6	public void testDrainToNull() {
529			SynchronousQueue q = new SynchronousQueue();
530			try {
531			q.drainTo(null);
532			shouldThrow();
533	jsr166	1.13	} catch (NullPointerException success) {}
534	dl	1.6	}
535
536			/**
537			* drainTo(this) throws IAE
538	jsr166	1.10	*/
539	dl	1.6	public void testDrainToSelf() {
540			SynchronousQueue q = new SynchronousQueue();
541			try {
542			q.drainTo(q);
543			shouldThrow();
544	jsr166	1.13	} catch (IllegalArgumentException success) {}
545	dl	1.6	}
546
547			/**
548			* drainTo(c) of empty queue doesn't transfer elements
549	jsr166	1.10	*/
550	dl	1.6	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	jsr166	1.10	*/
561	jsr166	1.13	public void testDrainToWithActivePut() throws InterruptedException {
562	dl	1.6	final SynchronousQueue q = new SynchronousQueue();
563	jsr166	1.35	Thread t = newStartedThread(new CheckedRunnable() {
564	jsr166	1.14	public void realRun() throws InterruptedException {
565	jsr166	1.35	q.put(one);
566	jsr166	1.13	}});
567
568			ArrayList l = new ArrayList();
569	jsr166	1.35	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	dl	1.6	}
580
581			/**
582			* drainTo(null, n) throws NPE
583	jsr166	1.10	*/
584	dl	1.6	public void testDrainToNullN() {
585			SynchronousQueue q = new SynchronousQueue();
586			try {
587			q.drainTo(null, 0);
588			shouldThrow();
589	jsr166	1.13	} catch (NullPointerException success) {}
590	dl	1.6	}
591
592			/**
593			* drainTo(this, n) throws IAE
594	jsr166	1.10	*/
595	dl	1.6	public void testDrainToSelfN() {
596			SynchronousQueue q = new SynchronousQueue();
597			try {
598			q.drainTo(q, 0);
599			shouldThrow();
600	jsr166	1.16	} catch (IllegalArgumentException success) {}
601	dl	1.6	}
602
603			/**
604			* drainTo(c, n) empties up to n elements of queue into c
605	jsr166	1.10	*/
606	jsr166	1.13	public void testDrainToN() throws InterruptedException {
607	dl	1.6	final SynchronousQueue q = new SynchronousQueue();
608	jsr166	1.35	Thread t1 = newStartedThread(new CheckedRunnable() {
609	jsr166	1.14	public void realRun() throws InterruptedException {
610	jsr166	1.13	q.put(one);
611			}});
612
613	jsr166	1.35	Thread t2 = newStartedThread(new CheckedRunnable() {
614	jsr166	1.14	public void realRun() throws InterruptedException {
615	jsr166	1.13	q.put(two);
616			}});
617	dl	1.6
618	jsr166	1.13	ArrayList l = new ArrayList();
619	dl	1.33	delay(SHORT_DELAY_MS);
620	jsr166	1.13	q.drainTo(l, 1);
621	jsr166	1.23	assertEquals(1, l.size());
622	jsr166	1.13	q.drainTo(l, 1);
623	jsr166	1.23	assertEquals(2, l.size());
624	jsr166	1.13	assertTrue(l.contains(one));
625			assertTrue(l.contains(two));
626	jsr166	1.35	awaitTermination(t1);
627			awaitTermination(t2);
628	dl	1.6	}
629
630	dl	1.1	}