test/tck/SynchronousQueueTest.java

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

import junit.framework.*;
import java.util.*;
import java.util.concurrent.*;
import java.io.*;

public class SynchronousQueueTest extends JSR166TestCase {

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

    public static Test suite() {
        return new TestSuite(SynchronousQueueTest.class);
    }

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

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

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

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

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

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

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

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

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

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

    /**
     * put blocks interruptibly if no active taker
     */
    public void testBlockingPut() {
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        SynchronousQueue q = new SynchronousQueue();
                        q.put(zero);
                        threadShouldThrow();
                    } catch (InterruptedException ie) {
                    }
                }});
        t.start();
        try {
           Thread.sleep(SHORT_DELAY_MS);
           t.interrupt();
           t.join();
        }
        catch (InterruptedException ie) {
            unexpectedException();
        }
    }

    /**
     * put blocks waiting for take
     */
    public void testPutWithTake() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new Runnable() {
                public void run() {
                    int added = 0;
                    try {
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        threadShouldThrow();
                    } catch (InterruptedException e) {
                        assertTrue(added >= 1);
                    }
                }
            });
        try {
            t.start();
            Thread.sleep(SHORT_DELAY_MS);
            q.take();
            Thread.sleep(SHORT_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testTimedOffer() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {

                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        threadShouldThrow();
                    } catch (InterruptedException success) {}
                }
            });

        try {
            t.start();
            Thread.sleep(SMALL_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }


    /**
     * take blocks interruptibly when empty
     */
    public void testTakeFromEmpty() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        q.take();
                        threadShouldThrow();
                    } catch (InterruptedException success) { }
                }
            });
        try {
            t.start();
            Thread.sleep(SHORT_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }


    /**
     * put blocks interruptibly if no active taker
     */
    public void testFairBlockingPut() {
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        SynchronousQueue q = new SynchronousQueue(true);
                        q.put(zero);
                        threadShouldThrow();
                    } catch (InterruptedException ie) {
                    }
                }});
        t.start();
        try {
           Thread.sleep(SHORT_DELAY_MS);
           t.interrupt();
           t.join();
        }
        catch (InterruptedException ie) {
            unexpectedException();
        }
    }

    /**
     * put blocks waiting for take
     */
    public void testFairPutWithTake() {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new Runnable() {
                public void run() {
                    int added = 0;
                    try {
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        q.put(new Object());
                        ++added;
                        threadShouldThrow();
                    } catch (InterruptedException e) {
                        assertTrue(added >= 1);
                    }
                }
            });
        try {
            t.start();
            Thread.sleep(SHORT_DELAY_MS);
            q.take();
            Thread.sleep(SHORT_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }

    /**
     * timed offer times out if elements not taken
     */
    public void testFairTimedOffer() {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {

                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        threadShouldThrow();
                    } catch (InterruptedException success) {}
                }
            });

        try {
            t.start();
            Thread.sleep(SMALL_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }


    /**
     * take blocks interruptibly when empty
     */
    public void testFairTakeFromEmpty() {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        q.take();
                        threadShouldThrow();
                    } catch (InterruptedException success) { }
                }
            });
        try {
            t.start();
            Thread.sleep(SHORT_DELAY_MS);
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }

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

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

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

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testInterruptedTimedPoll() {
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        SynchronousQueue q = new SynchronousQueue();
                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                    } catch (InterruptedException success) {
                    }
                }});
        t.start();
        try {
           Thread.sleep(SHORT_DELAY_MS);
           t.interrupt();
           t.join();
        }
        catch (InterruptedException ie) {
            unexpectedException();
        }
    }

    /**
     *  timed poll before a delayed offer fails; after offer succeeds;
     *  on interruption throws
     */
    public void testTimedPollWithOffer() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        threadShouldThrow();
                    } catch (InterruptedException success) { }
                }
            });
        try {
            t.start();
            Thread.sleep(SMALL_DELAY_MS);
            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }

    /**
     * Interrupted timed poll throws InterruptedException instead of
     * returning timeout status
     */
    public void testFairInterruptedTimedPoll() {
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        SynchronousQueue q = new SynchronousQueue(true);
                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                    } catch (InterruptedException success) {
                    }
                }});
        t.start();
        try {
           Thread.sleep(SHORT_DELAY_MS);
           t.interrupt();
           t.join();
        }
        catch (InterruptedException ie) {
            unexpectedException();
        }
    }

    /**
     *  timed poll before a delayed offer fails; after offer succeeds;
     *  on interruption throws
     */
    public void testFairTimedPollWithOffer() {
        final SynchronousQueue q = new SynchronousQueue(true);
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
                        threadShouldThrow();
                    } catch (InterruptedException success) { }
                }
            });
        try {
            t.start();
            Thread.sleep(SMALL_DELAY_MS);
            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
            t.interrupt();
            t.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }


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

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

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

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

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

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

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

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

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


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

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

    /**
     * toArray(null) throws NPE
     */
    public void testToArray_BadArg() {
        try {
            SynchronousQueue q = new SynchronousQueue();
            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 Integer one = new Integer(1);

        executor.execute(new Runnable() {
            public void run() {
                threadAssertFalse(q.offer(one));
                try {
                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
                    threadAssertEquals(0, q.remainingCapacity());
                }
                catch (InterruptedException e) {
                    threadUnexpectedException();
                }
            }
        });

        executor.execute(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(SMALL_DELAY_MS);
                    threadAssertEquals(one, q.take());
                }
                catch (InterruptedException e) {
                    threadUnexpectedException();
                }
            }
        });

        joinPool(executor);

    }

    /**
     * poll retrieves elements across Executor threads
     */
    public void testPollInExecutor() {
        final SynchronousQueue q = new SynchronousQueue();
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(new Runnable() {
            public void run() {
                threadAssertNull(q.poll());
                try {
                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
                    threadAssertTrue(q.isEmpty());
                }
                catch (InterruptedException e) {
                    threadUnexpectedException();
                }
            }
        });

        executor.execute(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(SMALL_DELAY_MS);
                    q.put(new Integer(1));
                }
                catch (InterruptedException e) {
                    threadUnexpectedException();
                }
            }
        });

        joinPool(executor);
    }

    /**
     * a deserialized serialized queue is usable
     */
    public void testSerialization() {
        SynchronousQueue q = new SynchronousQueue();
        try {
            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());
        } catch (Exception e) {
            e.printStackTrace();
            unexpectedException();
        }
    }

    /**
     * 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() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t = new Thread(new Runnable() {
                public void run() {
                    try {
                        q.put(new Integer(1));
                    } catch (InterruptedException ie) {
                        threadUnexpectedException();
                    }
                }
            });
        try {
            t.start();
            ArrayList l = new ArrayList();
            Thread.sleep(SHORT_DELAY_MS);
            q.drainTo(l);
            assertTrue(l.size() <= 1);
            if (l.size() > 0)
                assertEquals(l.get(0), new Integer(1));
            t.join();
            assertTrue(l.size() <= 1);
        } catch (Exception e) {
            unexpectedException();
        }
    }

    /**
     * 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() {
        final SynchronousQueue q = new SynchronousQueue();
        Thread t1 = new Thread(new Runnable() {
                public void run() {
                    try {
                        q.put(one);
                    } catch (InterruptedException ie) {
                        threadUnexpectedException();
                    }
                }
            });
        Thread t2 = new Thread(new Runnable() {
                public void run() {
                    try {
                        q.put(two);
                    } catch (InterruptedException ie) {
                        threadUnexpectedException();
                    }
                }
            });

        try {
            t1.start();
            t2.start();
            ArrayList l = new ArrayList();
            Thread.sleep(SHORT_DELAY_MS);
            q.drainTo(l, 1);
            assertTrue(l.size() == 1);
            q.drainTo(l, 1);
            assertTrue(l.size() == 2);
            assertTrue(l.contains(one));
            assertTrue(l.contains(two));
            t1.join();
            t2.join();
        } catch (Exception e) {
            unexpectedException();
        }
    }


}
Revision:	1.12
Committed:	Mon Nov 16 05:30:08 2009 UTC (14 years, 6 months ago) by jsr166
Branch:	MAIN
Changes since 1.11:	+30 -30 lines
Log Message:	whitespace
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain, as explained at
4	* http://creativecommons.org/licenses/publicdomain
5	* Other contributors include Andrew Wright, Jeffrey Hayes,
6	* Pat Fisher, Mike Judd.
7	*/
8
9	import junit.framework.*;
10	import java.util.*;
11	import java.util.concurrent.*;
12	import java.io.*;
13
14	public class SynchronousQueueTest extends JSR166TestCase {
15
16	public static void main(String[] args) {
17	junit.textui.TestRunner.run (suite());
18	}
19
20	public static Test suite() {
21	return new TestSuite(SynchronousQueueTest.class);
22	}
23
24	/**
25	* A SynchronousQueue is both empty and full
26	*/
27	public void testEmptyFull() {
28	SynchronousQueue q = new SynchronousQueue();
29	assertTrue(q.isEmpty());
30	assertEquals(0, q.size());
31	assertEquals(0, q.remainingCapacity());
32	assertFalse(q.offer(zero));
33	}
34
35	/**
36	* A fair SynchronousQueue is both empty and full
37	*/
38	public void testFairEmptyFull() {
39	SynchronousQueue q = new SynchronousQueue(true);
40	assertTrue(q.isEmpty());
41	assertEquals(0, q.size());
42	assertEquals(0, q.remainingCapacity());
43	assertFalse(q.offer(zero));
44	}
45
46	/**
47	* offer(null) throws NPE
48	*/
49	public void testOfferNull() {
50	try {
51	SynchronousQueue q = new SynchronousQueue();
52	q.offer(null);
53	shouldThrow();
54	} catch (NullPointerException success) { }
55	}
56
57	/**
58	* add(null) throws NPE
59	*/
60	public void testAddNull() {
61	try {
62	SynchronousQueue q = new SynchronousQueue();
63	q.add(null);
64	shouldThrow();
65	} catch (NullPointerException success) { }
66	}
67
68	/**
69	* offer fails if no active taker
70	*/
71	public void testOffer() {
72	SynchronousQueue q = new SynchronousQueue();
73	assertFalse(q.offer(one));
74	}
75
76	/**
77	* add throws ISE if no active taker
78	*/
79	public void testAdd() {
80	try {
81	SynchronousQueue q = new SynchronousQueue();
82	assertEquals(0, q.remainingCapacity());
83	q.add(one);
84	shouldThrow();
85	} catch (IllegalStateException success) {
86	}
87	}
88
89	/**
90	* addAll(null) throws NPE
91	*/
92	public void testAddAll1() {
93	try {
94	SynchronousQueue q = new SynchronousQueue();
95	q.addAll(null);
96	shouldThrow();
97	}
98	catch (NullPointerException success) {}
99	}
100
101	/**
102	* addAll(this) throws IAE
103	*/
104	public void testAddAllSelf() {
105	try {
106	SynchronousQueue q = new SynchronousQueue();
107	q.addAll(q);
108	shouldThrow();
109	}
110	catch (IllegalArgumentException success) {}
111	}
112
113	/**
114	* addAll of a collection with null elements throws NPE
115	*/
116	public void testAddAll2() {
117	try {
118	SynchronousQueue q = new SynchronousQueue();
119	Integer[] ints = new Integer[1];
120	q.addAll(Arrays.asList(ints));
121	shouldThrow();
122	}
123	catch (NullPointerException success) {}
124	}
125	/**
126	* addAll throws ISE if no active taker
127	*/
128	public void testAddAll4() {
129	try {
130	SynchronousQueue q = new SynchronousQueue();
131	Integer[] ints = new Integer[1];
132	for (int i = 0; i < 1; ++i)
133	ints[i] = new Integer(i);
134	q.addAll(Arrays.asList(ints));
135	shouldThrow();
136	}
137	catch (IllegalStateException success) {}
138	}
139
140	/**
141	* put(null) throws NPE
142	*/
143	public void testPutNull() {
144	try {
145	SynchronousQueue q = new SynchronousQueue();
146	q.put(null);
147	shouldThrow();
148	}
149	catch (NullPointerException success) {
150	}
151	catch (InterruptedException ie) {
152	unexpectedException();
153	}
154	}
155
156	/**
157	* put blocks interruptibly if no active taker
158	*/
159	public void testBlockingPut() {
160	Thread t = new Thread(new Runnable() {
161	public void run() {
162	try {
163	SynchronousQueue q = new SynchronousQueue();
164	q.put(zero);
165	threadShouldThrow();
166	} catch (InterruptedException ie) {
167	}
168	}});
169	t.start();
170	try {
171	Thread.sleep(SHORT_DELAY_MS);
172	t.interrupt();
173	t.join();
174	}
175	catch (InterruptedException ie) {
176	unexpectedException();
177	}
178	}
179
180	/**
181	* put blocks waiting for take
182	*/
183	public void testPutWithTake() {
184	final SynchronousQueue q = new SynchronousQueue();
185	Thread t = new Thread(new Runnable() {
186	public void run() {
187	int added = 0;
188	try {
189	q.put(new Object());
190	++added;
191	q.put(new Object());
192	++added;
193	q.put(new Object());
194	++added;
195	q.put(new Object());
196	++added;
197	threadShouldThrow();
198	} catch (InterruptedException e) {
199	assertTrue(added >= 1);
200	}
201	}
202	});
203	try {
204	t.start();
205	Thread.sleep(SHORT_DELAY_MS);
206	q.take();
207	Thread.sleep(SHORT_DELAY_MS);
208	t.interrupt();
209	t.join();
210	} catch (Exception e) {
211	unexpectedException();
212	}
213	}
214
215	/**
216	* timed offer times out if elements not taken
217	*/
218	public void testTimedOffer() {
219	final SynchronousQueue q = new SynchronousQueue();
220	Thread t = new Thread(new Runnable() {
221	public void run() {
222	try {
223
224	threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
225	q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
226	threadShouldThrow();
227	} catch (InterruptedException success) {}
228	}
229	});
230
231	try {
232	t.start();
233	Thread.sleep(SMALL_DELAY_MS);
234	t.interrupt();
235	t.join();
236	} catch (Exception e) {
237	unexpectedException();
238	}
239	}
240
241
242	/**
243	* take blocks interruptibly when empty
244	*/
245	public void testTakeFromEmpty() {
246	final SynchronousQueue q = new SynchronousQueue();
247	Thread t = new Thread(new Runnable() {
248	public void run() {
249	try {
250	q.take();
251	threadShouldThrow();
252	} catch (InterruptedException success) { }
253	}
254	});
255	try {
256	t.start();
257	Thread.sleep(SHORT_DELAY_MS);
258	t.interrupt();
259	t.join();
260	} catch (Exception e) {
261	unexpectedException();
262	}
263	}
264
265
266	/**
267	* put blocks interruptibly if no active taker
268	*/
269	public void testFairBlockingPut() {
270	Thread t = new Thread(new Runnable() {
271	public void run() {
272	try {
273	SynchronousQueue q = new SynchronousQueue(true);
274	q.put(zero);
275	threadShouldThrow();
276	} catch (InterruptedException ie) {
277	}
278	}});
279	t.start();
280	try {
281	Thread.sleep(SHORT_DELAY_MS);
282	t.interrupt();
283	t.join();
284	}
285	catch (InterruptedException ie) {
286	unexpectedException();
287	}
288	}
289
290	/**
291	* put blocks waiting for take
292	*/
293	public void testFairPutWithTake() {
294	final SynchronousQueue q = new SynchronousQueue(true);
295	Thread t = new Thread(new Runnable() {
296	public void run() {
297	int added = 0;
298	try {
299	q.put(new Object());
300	++added;
301	q.put(new Object());
302	++added;
303	q.put(new Object());
304	++added;
305	q.put(new Object());
306	++added;
307	threadShouldThrow();
308	} catch (InterruptedException e) {
309	assertTrue(added >= 1);
310	}
311	}
312	});
313	try {
314	t.start();
315	Thread.sleep(SHORT_DELAY_MS);
316	q.take();
317	Thread.sleep(SHORT_DELAY_MS);
318	t.interrupt();
319	t.join();
320	} catch (Exception e) {
321	unexpectedException();
322	}
323	}
324
325	/**
326	* timed offer times out if elements not taken
327	*/
328	public void testFairTimedOffer() {
329	final SynchronousQueue q = new SynchronousQueue(true);
330	Thread t = new Thread(new Runnable() {
331	public void run() {
332	try {
333
334	threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
335	q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
336	threadShouldThrow();
337	} catch (InterruptedException success) {}
338	}
339	});
340
341	try {
342	t.start();
343	Thread.sleep(SMALL_DELAY_MS);
344	t.interrupt();
345	t.join();
346	} catch (Exception e) {
347	unexpectedException();
348	}
349	}
350
351
352	/**
353	* take blocks interruptibly when empty
354	*/
355	public void testFairTakeFromEmpty() {
356	final SynchronousQueue q = new SynchronousQueue(true);
357	Thread t = new Thread(new Runnable() {
358	public void run() {
359	try {
360	q.take();
361	threadShouldThrow();
362	} catch (InterruptedException success) { }
363	}
364	});
365	try {
366	t.start();
367	Thread.sleep(SHORT_DELAY_MS);
368	t.interrupt();
369	t.join();
370	} catch (Exception e) {
371	unexpectedException();
372	}
373	}
374
375	/**
376	* poll fails unless active taker
377	*/
378	public void testPoll() {
379	SynchronousQueue q = new SynchronousQueue();
380	assertNull(q.poll());
381	}
382
383	/**
384	* timed pool with zero timeout times out if no active taker
385	*/
386	public void testTimedPoll0() {
387	try {
388	SynchronousQueue q = new SynchronousQueue();
389	assertNull(q.poll(0, TimeUnit.MILLISECONDS));
390	} catch (InterruptedException e) {
391	unexpectedException();
392	}
393	}
394
395	/**
396	* timed pool with nonzero timeout times out if no active taker
397	*/
398	public void testTimedPoll() {
399	try {
400	SynchronousQueue q = new SynchronousQueue();
401	assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
402	} catch (InterruptedException e) {
403	unexpectedException();
404	}
405	}
406
407	/**
408	* Interrupted timed poll throws InterruptedException instead of
409	* returning timeout status
410	*/
411	public void testInterruptedTimedPoll() {
412	Thread t = new Thread(new Runnable() {
413	public void run() {
414	try {
415	SynchronousQueue q = new SynchronousQueue();
416	assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
417	} catch (InterruptedException success) {
418	}
419	}});
420	t.start();
421	try {
422	Thread.sleep(SHORT_DELAY_MS);
423	t.interrupt();
424	t.join();
425	}
426	catch (InterruptedException ie) {
427	unexpectedException();
428	}
429	}
430
431	/**
432	* timed poll before a delayed offer fails; after offer succeeds;
433	* on interruption throws
434	*/
435	public void testTimedPollWithOffer() {
436	final SynchronousQueue q = new SynchronousQueue();
437	Thread t = new Thread(new Runnable() {
438	public void run() {
439	try {
440	threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
441	q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
442	q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
443	threadShouldThrow();
444	} catch (InterruptedException success) { }
445	}
446	});
447	try {
448	t.start();
449	Thread.sleep(SMALL_DELAY_MS);
450	assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
451	t.interrupt();
452	t.join();
453	} catch (Exception e) {
454	unexpectedException();
455	}
456	}
457
458	/**
459	* Interrupted timed poll throws InterruptedException instead of
460	* returning timeout status
461	*/
462	public void testFairInterruptedTimedPoll() {
463	Thread t = new Thread(new Runnable() {
464	public void run() {
465	try {
466	SynchronousQueue q = new SynchronousQueue(true);
467	assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
468	} catch (InterruptedException success) {
469	}
470	}});
471	t.start();
472	try {
473	Thread.sleep(SHORT_DELAY_MS);
474	t.interrupt();
475	t.join();
476	}
477	catch (InterruptedException ie) {
478	unexpectedException();
479	}
480	}
481
482	/**
483	* timed poll before a delayed offer fails; after offer succeeds;
484	* on interruption throws
485	*/
486	public void testFairTimedPollWithOffer() {
487	final SynchronousQueue q = new SynchronousQueue(true);
488	Thread t = new Thread(new Runnable() {
489	public void run() {
490	try {
491	threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
492	q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
493	q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
494	threadShouldThrow();
495	} catch (InterruptedException success) { }
496	}
497	});
498	try {
499	t.start();
500	Thread.sleep(SMALL_DELAY_MS);
501	assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
502	t.interrupt();
503	t.join();
504	} catch (Exception e) {
505	unexpectedException();
506	}
507	}
508
509
510	/**
511	* peek returns null
512	*/
513	public void testPeek() {
514	SynchronousQueue q = new SynchronousQueue();
515	assertNull(q.peek());
516	}
517
518	/**
519	* element throws NSEE
520	*/
521	public void testElement() {
522	SynchronousQueue q = new SynchronousQueue();
523	try {
524	q.element();
525	shouldThrow();
526	}
527	catch (NoSuchElementException success) {}
528	}
529
530	/**
531	* remove throws NSEE if no active taker
532	*/
533	public void testRemove() {
534	SynchronousQueue q = new SynchronousQueue();
535	try {
536	q.remove();
537	shouldThrow();
538	} catch (NoSuchElementException success) {
539	}
540	}
541
542	/**
543	* remove(x) returns false
544	*/
545	public void testRemoveElement() {
546	SynchronousQueue q = new SynchronousQueue();
547	assertFalse(q.remove(zero));
548	assertTrue(q.isEmpty());
549	}
550
551	/**
552	* contains returns false
553	*/
554	public void testContains() {
555	SynchronousQueue q = new SynchronousQueue();
556	assertFalse(q.contains(zero));
557	}
558
559	/**
560	* clear ensures isEmpty
561	*/
562	public void testClear() {
563	SynchronousQueue q = new SynchronousQueue();
564	q.clear();
565	assertTrue(q.isEmpty());
566	}
567
568	/**
569	* containsAll returns false unless empty
570	*/
571	public void testContainsAll() {
572	SynchronousQueue q = new SynchronousQueue();
573	Integer[] empty = new Integer[0];
574	assertTrue(q.containsAll(Arrays.asList(empty)));
575	Integer[] ints = new Integer[1]; ints[0] = zero;
576	assertFalse(q.containsAll(Arrays.asList(ints)));
577	}
578
579	/**
580	* retainAll returns false
581	*/
582	public void testRetainAll() {
583	SynchronousQueue q = new SynchronousQueue();
584	Integer[] empty = new Integer[0];
585	assertFalse(q.retainAll(Arrays.asList(empty)));
586	Integer[] ints = new Integer[1]; ints[0] = zero;
587	assertFalse(q.retainAll(Arrays.asList(ints)));
588	}
589
590	/**
591	* removeAll returns false
592	*/
593	public void testRemoveAll() {
594	SynchronousQueue q = new SynchronousQueue();
595	Integer[] empty = new Integer[0];
596	assertFalse(q.removeAll(Arrays.asList(empty)));
597	Integer[] ints = new Integer[1]; ints[0] = zero;
598	assertFalse(q.containsAll(Arrays.asList(ints)));
599	}
600
601
602	/**
603	* toArray is empty
604	*/
605	public void testToArray() {
606	SynchronousQueue q = new SynchronousQueue();
607	Object[] o = q.toArray();
608	assertEquals(o.length, 0);
609	}
610
611	/**
612	* toArray(a) is nulled at position 0
613	*/
614	public void testToArray2() {
615	SynchronousQueue q = new SynchronousQueue();
616	Integer[] ints = new Integer[1];
617	assertNull(ints[0]);
618	}
619
620	/**
621	* toArray(null) throws NPE
622	*/
623	public void testToArray_BadArg() {
624	try {
625	SynchronousQueue q = new SynchronousQueue();
626	Object o[] = q.toArray(null);
627	shouldThrow();
628	} catch (NullPointerException success) {}
629	}
630
631
632	/**
633	* iterator does not traverse any elements
634	*/
635	public void testIterator() {
636	SynchronousQueue q = new SynchronousQueue();
637	Iterator it = q.iterator();
638	assertFalse(it.hasNext());
639	try {
640	Object x = it.next();
641	shouldThrow();
642	}
643	catch (NoSuchElementException success) {}
644	}
645
646	/**
647	* iterator remove throws ISE
648	*/
649	public void testIteratorRemove() {
650	SynchronousQueue q = new SynchronousQueue();
651	Iterator it = q.iterator();
652	try {
653	it.remove();
654	shouldThrow();
655	}
656	catch (IllegalStateException success) {}
657	}
658
659	/**
660	* toString returns a non-null string
661	*/
662	public void testToString() {
663	SynchronousQueue q = new SynchronousQueue();
664	String s = q.toString();
665	assertNotNull(s);
666	}
667
668
669	/**
670	* offer transfers elements across Executor tasks
671	*/
672	public void testOfferInExecutor() {
673	final SynchronousQueue q = new SynchronousQueue();
674	ExecutorService executor = Executors.newFixedThreadPool(2);
675	final Integer one = new Integer(1);
676
677	executor.execute(new Runnable() {
678	public void run() {
679	threadAssertFalse(q.offer(one));
680	try {
681	threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
682	threadAssertEquals(0, q.remainingCapacity());
683	}
684	catch (InterruptedException e) {
685	threadUnexpectedException();
686	}
687	}
688	});
689
690	executor.execute(new Runnable() {
691	public void run() {
692	try {
693	Thread.sleep(SMALL_DELAY_MS);
694	threadAssertEquals(one, q.take());
695	}
696	catch (InterruptedException e) {
697	threadUnexpectedException();
698	}
699	}
700	});
701
702	joinPool(executor);
703
704	}
705
706	/**
707	* poll retrieves elements across Executor threads
708	*/
709	public void testPollInExecutor() {
710	final SynchronousQueue q = new SynchronousQueue();
711	ExecutorService executor = Executors.newFixedThreadPool(2);
712	executor.execute(new Runnable() {
713	public void run() {
714	threadAssertNull(q.poll());
715	try {
716	threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
717	threadAssertTrue(q.isEmpty());
718	}
719	catch (InterruptedException e) {
720	threadUnexpectedException();
721	}
722	}
723	});
724
725	executor.execute(new Runnable() {
726	public void run() {
727	try {
728	Thread.sleep(SMALL_DELAY_MS);
729	q.put(new Integer(1));
730	}
731	catch (InterruptedException e) {
732	threadUnexpectedException();
733	}
734	}
735	});
736
737	joinPool(executor);
738	}
739
740	/**
741	* a deserialized serialized queue is usable
742	*/
743	public void testSerialization() {
744	SynchronousQueue q = new SynchronousQueue();
745	try {
746	ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
747	ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
748	out.writeObject(q);
749	out.close();
750
751	ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
752	ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
753	SynchronousQueue r = (SynchronousQueue)in.readObject();
754	assertEquals(q.size(), r.size());
755	while (!q.isEmpty())
756	assertEquals(q.remove(), r.remove());
757	} catch (Exception e) {
758	e.printStackTrace();
759	unexpectedException();
760	}
761	}
762
763	/**
764	* drainTo(null) throws NPE
765	*/
766	public void testDrainToNull() {
767	SynchronousQueue q = new SynchronousQueue();
768	try {
769	q.drainTo(null);
770	shouldThrow();
771	} catch (NullPointerException success) {
772	}
773	}
774
775	/**
776	* drainTo(this) throws IAE
777	*/
778	public void testDrainToSelf() {
779	SynchronousQueue q = new SynchronousQueue();
780	try {
781	q.drainTo(q);
782	shouldThrow();
783	} catch (IllegalArgumentException success) {
784	}
785	}
786
787	/**
788	* drainTo(c) of empty queue doesn't transfer elements
789	*/
790	public void testDrainTo() {
791	SynchronousQueue q = new SynchronousQueue();
792	ArrayList l = new ArrayList();
793	q.drainTo(l);
794	assertEquals(q.size(), 0);
795	assertEquals(l.size(), 0);
796	}
797
798	/**
799	* drainTo empties queue, unblocking a waiting put.
800	*/
801	public void testDrainToWithActivePut() {
802	final SynchronousQueue q = new SynchronousQueue();
803	Thread t = new Thread(new Runnable() {
804	public void run() {
805	try {
806	q.put(new Integer(1));
807	} catch (InterruptedException ie) {
808	threadUnexpectedException();
809	}
810	}
811	});
812	try {
813	t.start();
814	ArrayList l = new ArrayList();
815	Thread.sleep(SHORT_DELAY_MS);
816	q.drainTo(l);
817	assertTrue(l.size() <= 1);
818	if (l.size() > 0)
819	assertEquals(l.get(0), new Integer(1));
820	t.join();
821	assertTrue(l.size() <= 1);
822	} catch (Exception e) {
823	unexpectedException();
824	}
825	}
826
827	/**
828	* drainTo(null, n) throws NPE
829	*/
830	public void testDrainToNullN() {
831	SynchronousQueue q = new SynchronousQueue();
832	try {
833	q.drainTo(null, 0);
834	shouldThrow();
835	} catch (NullPointerException success) {
836	}
837	}
838
839	/**
840	* drainTo(this, n) throws IAE
841	*/
842	public void testDrainToSelfN() {
843	SynchronousQueue q = new SynchronousQueue();
844	try {
845	q.drainTo(q, 0);
846	shouldThrow();
847	} catch (IllegalArgumentException success) {
848	}
849	}
850
851	/**
852	* drainTo(c, n) empties up to n elements of queue into c
853	*/
854	public void testDrainToN() {
855	final SynchronousQueue q = new SynchronousQueue();
856	Thread t1 = new Thread(new Runnable() {
857	public void run() {
858	try {
859	q.put(one);
860	} catch (InterruptedException ie) {
861	threadUnexpectedException();
862	}
863	}
864	});
865	Thread t2 = new Thread(new Runnable() {
866	public void run() {
867	try {
868	q.put(two);
869	} catch (InterruptedException ie) {
870	threadUnexpectedException();
871	}
872	}
873	});
874
875	try {
876	t1.start();
877	t2.start();
878	ArrayList l = new ArrayList();
879	Thread.sleep(SHORT_DELAY_MS);
880	q.drainTo(l, 1);
881	assertTrue(l.size() == 1);
882	q.drainTo(l, 1);
883	assertTrue(l.size() == 2);
884	assertTrue(l.contains(one));
885	assertTrue(l.contains(two));
886	t1.join();
887	t2.join();
888	} catch (Exception e) {
889	unexpectedException();
890	}
891	}
892
893
894	}