util/concurrent/SynchronousQueue.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;
import java.util.*;

/**
 * A Queue in which each put must wait for a take, and vice versa.
 * SynchronousQueues are similar to rendezvous channels used in CSP
 * and Ada. They are well suited for handoff designs, in which an
 * object running in one thread must synch up with an object running
 * in another thread in order to hand it some information, event, or
 * task.
 **/
public class SynchronousQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /*
      This implementation divides actions into two cases for puts:

      * An arriving putter that does not already have a waiting taker 
      creates a node holding item, and then waits for a taker to take it.
      * An arriving putter that does already have a waiting taker fills
      the slot node created by the taker, and notifies it to continue.

      And symmetrically, two for takes:

      * An arriving taker that does not already have a waiting putter
      creates an empty slot node, and then waits for a putter to fill it.
      * An arriving taker that does already have a waiting putter takes
      item from the node created by the putter, and notifies it to continue.

      This requires keeping two simple queues: waitingPuts and waitingTakes.
   
      When a put or take waiting for the actions of its counterpart
      aborts due to interruption or timeout, it marks the node
      it created as "CANCELLED", which causes its counterpart to retry
      the entire put or take sequence.
    */

    /** 
     * Special marker used in queue nodes to indicate that
     * the thread waiting for a change in the node has timed out
     * or been interrupted.
     **/
    private static final Object CANCELLED = new Object();

    /*
     * Note that all fields are transient final, so there is
     * no explicit serialization code.
     */

    private transient final WaitQueue waitingPuts = new WaitQueue();
    private transient final WaitQueue waitingTakes = new WaitQueue();
    private transient final ReentrantLock qlock = new ReentrantLock();

    /**
     * Nodes each maintain an item and handle waits and signals for
     * getting and setting it. The class opportunistically extends
     * ReentrantLock to save an extra object allocation per
     * rendezvous.
     */
    private static class Node extends ReentrantLock {
        Condition done;
        Object item;
        Node next;
        Node(Object x) { item = x; }

        /**
         * Fill in the slot created by the taker and signal taker to
         * continue.
         */
        boolean set(Object x) {
            this.lock();
            try {
                if (item != CANCELLED) {
                    item = x;
                    if (done != null)
                        done.signal();
                    return true;
                }
                else // taker has cancelled
                    return false;
            }
            finally {
                this.unlock();
            }
        }

        /**
         * Remove item from slot created by putter and signal putter
         * to continue.
         */
        Object get() {
            this.lock();
            try {
                Object x = item;
                if (x != CANCELLED) {
                    item = null;
                    next = null;
                    if (done != null)
                        done.signal();
                    return x;
                }
                else
                    return null;
            }
            finally {
                this.unlock();
            }
        }


        /**
         * Wait for a taker to take item placed by putter, or time out.
         */
        boolean waitForTake(boolean timed, long nanos) throws InterruptedException {
            this.lock();
            try {
                for (;;) {
                    if (item == null)
                        return true;
                    if (done == null)
                        done = this.newCondition();
                    if (timed) {
                        if (nanos <= 0) {
                            item = CANCELLED;
                            return false;
                        }
                        nanos = done.awaitNanos(nanos);
                    }
                    else
                        done.await();
                }
            }
            catch (InterruptedException ie) {
                // If taken, return normally but set interrupt status
                if (item == null) {
                    Thread.currentThread().interrupt();
                    return true;
                }
                else {
                    item = CANCELLED;
                    done.signal(); // propagate signal
                    throw ie;
                }
            }
            finally {
                this.unlock();
            }
        }


        /**
         * Wait for a putter to put item placed by taker, or time out.
         */
        Object waitForPut(boolean timed, long nanos) throws InterruptedException {
            this.lock();
            try {
                for (;;) {
                    Object x = item;
                    if (x != null) {
                        item = null;
                        next = null;
                        return x;
                    }
                    if (done == null)
                        done = this.newCondition();
                    if (timed) {
                        if (nanos <= 0) {
                            item = CANCELLED;
                            return null;
                        }
                        nanos = done.awaitNanos(nanos);
                    }
                    else
                        done.await();
                }
            }
            catch(InterruptedException ie) {
                Object x = item;
                if (x != null) {
                    item = null;
                    next = null;
                    Thread.currentThread().interrupt();
                    return x;
                }
                else {
                    item = CANCELLED;
                    done.signal(); // propagate signal
                    throw ie;
                }
            }
            finally {
                this.unlock();
            }
        }
    }

    /**
     * Simple FIFO queue class to hold waiting puts/takes.
     **/
    private static class WaitQueue<E> {
        Node head;
        Node last;

        Node enq(Object x) { 
            Node p = new Node(x);
            if (last == null) 
                last = head = p;
            else 
                last = last.next = p;
            return p;
        }

        Node deq() {
            Node p = head;
            if (p != null && (head = p.next) == null) 
                last = null;
            return p;
        }
    }

    /**
     * Main put algorithm, used by put, timed offer
     */ 
    private boolean doPut(E x, boolean timed, long nanos) throws InterruptedException {
        if (x == null) throw new IllegalArgumentException();
        for (;;) { 
            Node node;
            boolean mustWait;
            
            qlock.lockInterruptibly();
            try {
                node = waitingTakes.deq();
                if ( (mustWait = (node == null)) )
                    node = waitingPuts.enq(x);
            }
            finally {
                qlock.unlock();
            }

            if (mustWait)
                return node.waitForTake(timed, nanos);

            else if (node.set(x))
                return true;

            // else taker cancelled, so retry
        }
    }

    /**
     * Main take algorithm, used by take, timed poll
     */ 
    private E doTake(boolean timed, long nanos) throws InterruptedException {
        for (;;) {
            Node node;
            boolean mustWait;

            qlock.lockInterruptibly();
            try {
                node = waitingPuts.deq();
                if ( (mustWait = (node == null)) )
                    node = waitingTakes.enq(null);
            }
            finally {
                qlock.unlock();
            }

            if (mustWait)
                return (E)node.waitForPut(timed, nanos);

            else {
                E x = (E)node.get();
                if (x != null)
                    return x;
                // else cancelled, so retry
            }
        }
    }

    public SynchronousQueue() {}

    public boolean isEmpty() {
        return true;
    }

    public int size() {
        return 0;
    }

    /**
     * Always returns zero. SynchronousQueues have no internal capacity.
     * @return zero.
     */
    public int remainingCapacity() {
        return 0;
    }

    public E peek() {
        return null;
    }


    public void put(E x) throws InterruptedException {
        doPut(x, false, 0);
    }

    public boolean offer(E x, long timeout, TimeUnit unit) throws InterruptedException {
        return doPut(x, true, unit.toNanos(timeout));
    }


    public E take() throws InterruptedException {
        return doTake(false, 0);
    }

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        return doTake(true, unit.toNanos(timeout));
    }

    // Untimed nonblocking versions

    public boolean offer(E x) {
        if (x == null) throw new IllegalArgumentException();
        
        for (;;) { 
            qlock.lock();
            Node node;
            try {
                node = waitingTakes.deq();
            }
            finally {
                qlock.unlock();
            }
            if (node == null)
                return false;
            
            else if (node.set(x))
                return true;
            // else retry
        }
    }

    public E poll() {
        for (;;) {
            Node node;
            qlock.lock();
            try {
                node = waitingPuts.deq();
            }
            finally {
                qlock.unlock();
            }
            if (node == null)
                return null;

            else {
                Object x = node.get();
                if (x != null)
                    return (E)x;
                // else retry
            }
        }
    }

    public boolean remove(Object x) {
        return false;
    }

    static class EmptyIterator<E> implements Iterator {
        public boolean hasNext() {
            return false;
        }
        public E next() {
            throw new NoSuchElementException();
        }
        public void remove() {
            throw new UnsupportedOperationException();
        }
    }

    public Iterator<E> iterator() {
        return new EmptyIterator();
    }


    public Object[] toArray() {
        return new E[0];
    }

    public <T> T[] toArray(T[] a) {
        if (a.length > 0)
            a[0] = null;
        return a;
    }
}
Revision:	1.4
Committed:	Sat Jun 7 11:54:20 2003 UTC (21 years ago) by dl
Branch:	MAIN
Changes since 1.3:	+4 -0 lines
Log Message:	More javadoc fixes
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain. Use, modify, and
4	* redistribute this code in any way without acknowledgement.
5	*/
6
7	package java.util.concurrent;
8	import java.util.*;
9
10	/**
11	* A Queue in which each put must wait for a take, and vice versa.
12	* SynchronousQueues are similar to rendezvous channels used in CSP
13	* and Ada. They are well suited for handoff designs, in which an
14	* object running in one thread must synch up with an object running
15	* in another thread in order to hand it some information, event, or
16	* task.
17	**/
18	public class SynchronousQueue<E> extends AbstractQueue<E>
19	implements BlockingQueue<E>, java.io.Serializable {
20
21	/*
22	This implementation divides actions into two cases for puts:
23
24	* An arriving putter that does not already have a waiting taker
25	creates a node holding item, and then waits for a taker to take it.
26	* An arriving putter that does already have a waiting taker fills
27	the slot node created by the taker, and notifies it to continue.
28
29	And symmetrically, two for takes:
30
31	* An arriving taker that does not already have a waiting putter
32	creates an empty slot node, and then waits for a putter to fill it.
33	* An arriving taker that does already have a waiting putter takes
34	item from the node created by the putter, and notifies it to continue.
35
36	This requires keeping two simple queues: waitingPuts and waitingTakes.
37
38	When a put or take waiting for the actions of its counterpart
39	aborts due to interruption or timeout, it marks the node
40	it created as "CANCELLED", which causes its counterpart to retry
41	the entire put or take sequence.
42	*/
43
44	/**
45	* Special marker used in queue nodes to indicate that
46	* the thread waiting for a change in the node has timed out
47	* or been interrupted.
48	**/
49	private static final Object CANCELLED = new Object();
50
51	/*
52	* Note that all fields are transient final, so there is
53	* no explicit serialization code.
54	*/
55
56	private transient final WaitQueue waitingPuts = new WaitQueue();
57	private transient final WaitQueue waitingTakes = new WaitQueue();
58	private transient final ReentrantLock qlock = new ReentrantLock();
59
60	/**
61	* Nodes each maintain an item and handle waits and signals for
62	* getting and setting it. The class opportunistically extends
63	* ReentrantLock to save an extra object allocation per
64	* rendezvous.
65	*/
66	private static class Node extends ReentrantLock {
67	Condition done;
68	Object item;
69	Node next;
70	Node(Object x) { item = x; }
71
72	/**
73	* Fill in the slot created by the taker and signal taker to
74	* continue.
75	*/
76	boolean set(Object x) {
77	this.lock();
78	try {
79	if (item != CANCELLED) {
80	item = x;
81	if (done != null)
82	done.signal();
83	return true;
84	}
85	else // taker has cancelled
86	return false;
87	}
88	finally {
89	this.unlock();
90	}
91	}
92
93	/**
94	* Remove item from slot created by putter and signal putter
95	* to continue.
96	*/
97	Object get() {
98	this.lock();
99	try {
100	Object x = item;
101	if (x != CANCELLED) {
102	item = null;
103	next = null;
104	if (done != null)
105	done.signal();
106	return x;
107	}
108	else
109	return null;
110	}
111	finally {
112	this.unlock();
113	}
114	}
115
116
117	/**
118	* Wait for a taker to take item placed by putter, or time out.
119	*/
120	boolean waitForTake(boolean timed, long nanos) throws InterruptedException {
121	this.lock();
122	try {
123	for (;;) {
124	if (item == null)
125	return true;
126	if (done == null)
127	done = this.newCondition();
128	if (timed) {
129	if (nanos <= 0) {
130	item = CANCELLED;
131	return false;
132	}
133	nanos = done.awaitNanos(nanos);
134	}
135	else
136	done.await();
137	}
138	}
139	catch (InterruptedException ie) {
140	// If taken, return normally but set interrupt status
141	if (item == null) {
142	Thread.currentThread().interrupt();
143	return true;
144	}
145	else {
146	item = CANCELLED;
147	done.signal(); // propagate signal
148	throw ie;
149	}
150	}
151	finally {
152	this.unlock();
153	}
154	}
155
156
157	/**
158	* Wait for a putter to put item placed by taker, or time out.
159	*/
160	Object waitForPut(boolean timed, long nanos) throws InterruptedException {
161	this.lock();
162	try {
163	for (;;) {
164	Object x = item;
165	if (x != null) {
166	item = null;
167	next = null;
168	return x;
169	}
170	if (done == null)
171	done = this.newCondition();
172	if (timed) {
173	if (nanos <= 0) {
174	item = CANCELLED;
175	return null;
176	}
177	nanos = done.awaitNanos(nanos);
178	}
179	else
180	done.await();
181	}
182	}
183	catch(InterruptedException ie) {
184	Object x = item;
185	if (x != null) {
186	item = null;
187	next = null;
188	Thread.currentThread().interrupt();
189	return x;
190	}
191	else {
192	item = CANCELLED;
193	done.signal(); // propagate signal
194	throw ie;
195	}
196	}
197	finally {
198	this.unlock();
199	}
200	}
201	}
202
203	/**
204	* Simple FIFO queue class to hold waiting puts/takes.
205	**/
206	private static class WaitQueue<E> {
207	Node head;
208	Node last;
209
210	Node enq(Object x) {
211	Node p = new Node(x);
212	if (last == null)
213	last = head = p;
214	else
215	last = last.next = p;
216	return p;
217	}
218
219	Node deq() {
220	Node p = head;
221	if (p != null && (head = p.next) == null)
222	last = null;
223	return p;
224	}
225	}
226
227	/**
228	* Main put algorithm, used by put, timed offer
229	*/
230	private boolean doPut(E x, boolean timed, long nanos) throws InterruptedException {
231	if (x == null) throw new IllegalArgumentException();
232	for (;;) {
233	Node node;
234	boolean mustWait;
235
236	qlock.lockInterruptibly();
237	try {
238	node = waitingTakes.deq();
239	if ( (mustWait = (node == null)) )
240	node = waitingPuts.enq(x);
241	}
242	finally {
243	qlock.unlock();
244	}
245
246	if (mustWait)
247	return node.waitForTake(timed, nanos);
248
249	else if (node.set(x))
250	return true;
251
252	// else taker cancelled, so retry
253	}
254	}
255
256	/**
257	* Main take algorithm, used by take, timed poll
258	*/
259	private E doTake(boolean timed, long nanos) throws InterruptedException {
260	for (;;) {
261	Node node;
262	boolean mustWait;
263
264	qlock.lockInterruptibly();
265	try {
266	node = waitingPuts.deq();
267	if ( (mustWait = (node == null)) )
268	node = waitingTakes.enq(null);
269	}
270	finally {
271	qlock.unlock();
272	}
273
274	if (mustWait)
275	return (E)node.waitForPut(timed, nanos);
276
277	else {
278	E x = (E)node.get();
279	if (x != null)
280	return x;
281	// else cancelled, so retry
282	}
283	}
284	}
285
286	public SynchronousQueue() {}
287
288	public boolean isEmpty() {
289	return true;
290	}
291
292	public int size() {
293	return 0;
294	}
295
296	/**
297	* Always returns zero. SynchronousQueues have no internal capacity.
298	* @return zero.
299	*/
300	public int remainingCapacity() {
301	return 0;
302	}
303
304	public E peek() {
305	return null;
306	}
307
308
309	public void put(E x) throws InterruptedException {
310	doPut(x, false, 0);
311	}
312
313	public boolean offer(E x, long timeout, TimeUnit unit) throws InterruptedException {
314	return doPut(x, true, unit.toNanos(timeout));
315	}
316
317
318
319	public E take() throws InterruptedException {
320	return doTake(false, 0);
321	}
322
323	public E poll(long timeout, TimeUnit unit) throws InterruptedException {
324	return doTake(true, unit.toNanos(timeout));
325	}
326
327	// Untimed nonblocking versions
328
329	public boolean offer(E x) {
330	if (x == null) throw new IllegalArgumentException();
331
332	for (;;) {
333	qlock.lock();
334	Node node;
335	try {
336	node = waitingTakes.deq();
337	}
338	finally {
339	qlock.unlock();
340	}
341	if (node == null)
342	return false;
343
344	else if (node.set(x))
345	return true;
346	// else retry
347	}
348	}
349
350	public E poll() {
351	for (;;) {
352	Node node;
353	qlock.lock();
354	try {
355	node = waitingPuts.deq();
356	}
357	finally {
358	qlock.unlock();
359	}
360	if (node == null)
361	return null;
362
363	else {
364	Object x = node.get();
365	if (x != null)
366	return (E)x;
367	// else retry
368	}
369	}
370	}
371
372	public boolean remove(Object x) {
373	return false;
374	}
375
376	static class EmptyIterator<E> implements Iterator {
377	public boolean hasNext() {
378	return false;
379	}
380	public E next() {
381	throw new NoSuchElementException();
382	}
383	public void remove() {
384	throw new UnsupportedOperationException();
385	}
386	}
387
388	public Iterator<E> iterator() {
389	return new EmptyIterator();
390	}
391
392
393	public Object[] toArray() {
394	return new E[0];
395	}
396
397	public <T> T[] toArray(T[] a) {
398	if (a.length > 0)
399	a[0] = null;
400	return a;
401	}
402	}