util/concurrent/Exchanger.java

/*
 * Written by Doug Lea, Bill Scherer, and Michael Scott with
 * assistance from members of JCP JSR-166 Expert Group and released to
 * the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 */

package java.util.concurrent;
import java.util.concurrent.*; // for javadoc (till 6280605 is fixed)
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.*;
import java.util.Random;

/**
 * A synchronization point at which threads can pair and swap elements
 * within pairs.  Each thread presents some object on entry to the
 * {@link #exchange exchange} method, matches with a partner thread,
 * and receives its partner's object on return.
 *
 * <p><b>Sample Usage:</b>
 * Here are the highlights of a class that uses an {@code Exchanger}
 * to swap buffers between threads so that the thread filling the
 * buffer gets a freshly emptied one when it needs it, handing off the
 * filled one to the thread emptying the buffer.
 * <pre>{@code
 * class FillAndEmpty {
 *   Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
 *   DataBuffer initialEmptyBuffer = ... a made-up type
 *   DataBuffer initialFullBuffer = ...
 *
 *   class FillingLoop implements Runnable {
 *     public void run() {
 *       DataBuffer currentBuffer = initialEmptyBuffer;
 *       try {
 *         while (currentBuffer != null) {
 *           addToBuffer(currentBuffer);
 *           if (currentBuffer.isFull())
 *             currentBuffer = exchanger.exchange(currentBuffer);
 *         }
 *       } catch (InterruptedException ex) { ... handle ... }
 *     }
 *   }
 *
 *   class EmptyingLoop implements Runnable {
 *     public void run() {
 *       DataBuffer currentBuffer = initialFullBuffer;
 *       try {
 *         while (currentBuffer != null) {
 *           takeFromBuffer(currentBuffer);
 *           if (currentBuffer.isEmpty())
 *             currentBuffer = exchanger.exchange(currentBuffer);
 *         }
 *       } catch (InterruptedException ex) { ... handle ...}
 *     }
 *   }
 *
 *   void start() {
 *     new Thread(new FillingLoop()).start();
 *     new Thread(new EmptyingLoop()).start();
 *   }
 * }
 * }</pre>
 *
 * <p>Memory consistency effects: For each pair of threads that
 * successfully exchange objects via an {@code Exchanger}, actions
 * prior to the {@code exchange()} in each thread
 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * those subsequent to a return from the corresponding {@code exchange()}
 * in the other thread.
 *
 * @since 1.5
 * @author Doug Lea and Bill Scherer and Michael Scott
 * @param <V> The type of objects that may be exchanged
 */
public class Exchanger<V> {
    /*
     * The underlying idea is to use a stack to hold nodes containing
     * pairs of items to be exchanged. Except that:
     *
     *  *  Only one element of the pair is known on creation by a
     *     first-arriving thread; the other is a "hole" waiting to be
     *     filled in. This is a degenerate form of the dual stacks
     *     described in "Nonblocking Concurrent Objects with Condition
     *     Synchronization",  by W. N. Scherer III and M. L. Scott.
     *     18th Annual Conf. on  Distributed Computing, Oct. 2004.
     *     It is "degenerate" in that both the items and the holes
     *     are shared in the same nodes.
     *
     *  *  There isn't really a stack here! There can't be -- if two
     *     nodes were both in the stack, they should cancel themselves
     *     out by combining. So that's what we do. The 0th element of
     *     the "arena" array serves only as the top of stack.  The
     *     remainder of the array is a form of the elimination backoff
     *     collision array described in "A Scalable Lock-free Stack
     *     Algorithm", by D. Hendler, N. Shavit, and L. Yerushalmi.
     *     16th ACM Symposium on Parallelism in Algorithms and
     *     Architectures, June 2004. Here, threads spin (using short
     *     timed waits with exponential backoff) looking for each
     *     other.  If they fail to find others waiting, they try the
     *     top spot again.  As shown in that paper, this always
     *     converges.
     *
     * The backoff elimination mechanics never come into play in
     * common usages where only two threads ever meet to exchange
     * items, but they prevent contention bottlenecks when an
     * exchanger is used by a large number of threads.
     *
     * For more details, see the paper "A Scalable Elimination-based
     * Exchange Channel" by William Scherer, Doug Lea, and Michael
     * Scott in Proceedings of SCOOL05 workshop. Available at:
     * http://hdl.handle.net/1802/2104
     */

    /** The number of CPUs, for sizing and spin control */
    static final int NCPUS = Runtime.getRuntime().availableProcessors();

    /**
     * Size of collision space. Using a size of half the number of
     * CPUs provides enough space for threads to find each other but
     * not so much that it would always require one or more to time
     * out to become unstuck. Note that the arena array holds SIZE+1
     * elements, to include the top-of-stack slot.
     */
    private static final int SIZE = (NCPUS + 1) / 2;

    /**
     * The number of times to spin before blocking in timed waits.
     * The value is empirically derived -- it works well across a
     * variety of processors and OSes. Empirically, the best value
     * seems not to vary with number of CPUs (beyond 2) so is just
     * a constant.
     */
    static final int maxTimedSpins = (NCPUS < 2)? 0 : 16;

    /**
     * The number of times to spin before blocking in untimed waits.
     * This is greater than timed value because untimed waits spin
     * faster since they don't need to check times on each spin.
     */
    static final int maxUntimedSpins = maxTimedSpins * 32;

    /**
     * The number of nanoseconds for which it is faster to spin
     * rather than to use timed park. A rough estimate suffices.
     */
    static final long spinForTimeoutThreshold = 1000L;

    /**
     * Base unit in nanoseconds for backoffs. Must be a power of two.
     * Should be small because backoffs exponentially increase from
     * base.
     */
    private static final long BACKOFF_BASE = 128L;

    /**
     * Sentinel item representing cancellation.  This value is placed
     * in holes on cancellation, and used as a return value from Node
     * methods to indicate failure to set or get hole.
     */
    static final Object FAIL = new Object();

    /**
     * The collision arena. arena[0] is used as the top of the stack.
     * The remainder is used as the collision elimination space.
     */
    private final AtomicReference<Node>[] arena;

    /** Generator for random backoffs and delays. */
    private final Random random = new Random();

    /**
     * Creates a new Exchanger.
     */
    public Exchanger() {
        arena = (AtomicReference<Node>[]) new AtomicReference[SIZE + 1];
        for (int i = 0; i < arena.length; ++i)
            arena[i] = new AtomicReference<Node>();
    }

    /**
     * Main exchange function, handling the different policy variants.
     * Uses Object, not "V" as argument and return value to simplify
     * handling of internal sentinel values. Callers from public
     * methods cast accordingly.
     *
     * @param item the item to exchange
     * @param timed true if the wait is timed
     * @param nanos if timed, the maximum wait time
     * @return the other thread's item
     */
    private Object doExchange(Object item, boolean timed, long nanos)
        throws InterruptedException, TimeoutException {
        Node me = new Node(item);
        long lastTime = timed ? System.nanoTime() : 0;
        int idx = 0;     // start out at slot representing top
        int backoff = 0; // increases on failure to occupy a slot

        for (;;) {
            AtomicReference<Node> slot = arena[idx];

            // If this slot is already occupied, there is a waiting item...
            Node you = slot.get();
            if (you != null) {
                Object v = you.fillHole(item);
                slot.compareAndSet(you, null);
                if (v != FAIL)       // ... unless it was cancelled
                    return v;
            }

            // Try to occupy this slot
            if (slot.compareAndSet(null, me)) {
                // If this is top slot, use regular wait, else backoff-wait
                Object v = ((idx == 0)?
                            me.waitForHole(timed, nanos) :
                            me.waitForHole(true, randomDelay(backoff)));
                slot.compareAndSet(me, null);
                if (v != FAIL)
                    return v;
                if (Thread.interrupted())
                    throw new InterruptedException();
                if (timed) {
                    long now = System.nanoTime();
                    nanos -= now - lastTime;
                    lastTime = now;
                    if (nanos <= 0)
                        throw new TimeoutException();
                }

                me = new Node(item);      // Throw away nodes on failure
                if (backoff < SIZE - 1)   // Increase or stay saturated
                    ++backoff;
                idx = 0;                  // Restart at top
            }

            else // Retry with a random non-top slot <= backoff
                idx = 1 + random.nextInt(backoff + 1);

        }
    }

    /**
     * Returns a random delay less than (base times (2 raised to backoff)).
     */
    private long randomDelay(int backoff) {
        return ((BACKOFF_BASE << backoff) - 1) & random.nextInt();
    }

    /**
     * Nodes hold partially exchanged data. This class
     * opportunistically subclasses AtomicReference to represent the
     * hole. So get() returns hole, and compareAndSet CAS'es value
     * into hole.  Note that this class cannot be parameterized as V
     * because the sentinel value FAIL is only of type Object.
     */
    static final class Node extends AtomicReference<Object> {
        private static final long serialVersionUID = -3221313401284163686L;

        /** The element offered by the Thread creating this node. */
        final Object item;

        /** The Thread creating this node. */
        final Thread waiter;

        /**
         * Creates node with given item and empty hole.
         *
         * @param item the item
         */
        Node(Object item) {
            this.item = item;
            waiter = Thread.currentThread();
        }

        /**
         * Tries to fill in hole. On success, wakes up the waiter.
         *
         * @param val the value to place in hole
         * @return on success, the item; on failure, FAIL
         */
        Object fillHole(Object val) {
            if (compareAndSet(null, val)) {
                LockSupport.unpark(waiter);
                return item;
            }
            return FAIL;
        }

        /**
         * Waits for and gets the hole filled in by another thread.
         * Fails if timed out or interrupted before hole filled.
         *
         * @param timed true if the wait is timed
         * @param nanos if timed, the maximum wait time
         * @return on success, the hole; on failure, FAIL
         */
        Object waitForHole(boolean timed, long nanos) {
            long lastTime = timed ? System.nanoTime() : 0;
            int spins = timed? maxTimedSpins : maxUntimedSpins;
            Object h;
            while ((h = get()) == null) {
                // If interrupted or timed out, try to cancel by
                // CASing FAIL as hole value.
                if (Thread.currentThread().isInterrupted() ||
                    (timed && nanos <= 0)) {
                    if (compareAndSet(null, FAIL))
                        return FAIL;
                } else {
                    if (timed) {
                        long now = System.nanoTime();
                        nanos -= now - lastTime;
                        lastTime = now;
                    }
                    if (spins > 0)
                        --spins;
                    else if (!timed) 
                        LockSupport.park();
                    else if (nanos > spinForTimeoutThreshold)
                        LockSupport.parkNanos(nanos);
                }
            }
            return h;
        }
    }

    /**
     * Waits for another thread to arrive at this exchange point (unless
     * the current thread is {@link Thread#interrupt interrupted}),
     * and then transfers the given object to it, receiving its object
     * in return.
     *
     * <p>If another thread is already waiting at the exchange point then
     * it is resumed for thread scheduling purposes and receives the object
     * passed in by the current thread. The current thread returns immediately,
     * receiving the object passed to the exchange by that other thread.
     *
     * <p>If no other thread is already waiting at the exchange then the
     * current thread is disabled for thread scheduling purposes and lies
     * dormant until one of two things happens:
     * <ul>
     * <li>Some other thread enters the exchange; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread.
     * </ul>
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting
     * for the exchange,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     *
     * @param x the object to exchange
     * @return the object provided by the other thread
     * @throws InterruptedException if the current thread was
     *         interrupted while waiting
     */
    public V exchange(V x) throws InterruptedException {
        try {
            return (V)doExchange(x, false, 0);
        } catch (TimeoutException cannotHappen) {
            throw new Error(cannotHappen);
        }
    }

    /**
     * Waits for another thread to arrive at this exchange point (unless
     * the current thread is {@link Thread#interrupt interrupted} or
     * the specified waiting time elapses), and then transfers the given
     * object to it, receiving its object in return.
     *
     * <p>If another thread is already waiting at the exchange point then
     * it is resumed for thread scheduling purposes and receives the object
     * passed in by the current thread. The current thread returns immediately,
     * receiving the object passed to the exchange by that other thread.
     *
     * <p>If no other thread is already waiting at the exchange then the
     * current thread is disabled for thread scheduling purposes and lies
     * dormant until one of three things happens:
     * <ul>
     * <li>Some other thread enters the exchange; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread; or
     * <li>The specified waiting time elapses.
     * </ul>
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting
     * for the exchange,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     *
     * <p>If the specified waiting time elapses then {@link TimeoutException}
     * is thrown.
     * If the time is
     * less than or equal to zero, the method will not wait at all.
     *
     * @param x the object to exchange
     * @param timeout the maximum time to wait
     * @param unit the time unit of the <tt>timeout</tt> argument
     * @return the object provided by the other thread
     * @throws InterruptedException if the current thread was
     *         interrupted while waiting
     * @throws TimeoutException if the specified waiting time elapses
     *         before another thread enters the exchange
     */
    public V exchange(V x, long timeout, TimeUnit unit)
        throws InterruptedException, TimeoutException {
        return (V)doExchange(x, true, unit.toNanos(timeout));
    }
}
Revision:	1.32
Committed:	Sat Dec 10 20:09:28 2005 UTC (18 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.31:	+39 -10 lines
Log Message:	Use same spin control as SynchronousQueue
#	User	Rev	Content
1	dl	1.2	/*
2	dl	1.16	* Written by Doug Lea, Bill Scherer, and Michael Scott with
3			* assistance from members of JCP JSR-166 Expert Group and released to
4			* the public domain, as explained at
5	dl	1.14	* http://creativecommons.org/licenses/publicdomain
6	dl	1.2	*/
7
8	tim	1.1	package java.util.concurrent;
9	jsr166	1.21	import java.util.concurrent.*; // for javadoc (till 6280605 is fixed)
10	dl	1.4	import java.util.concurrent.locks.*;
11	dl	1.16	import java.util.concurrent.atomic.*;
12			import java.util.Random;
13	tim	1.1
14			/**
15	dl	1.28	* A synchronization point at which threads can pair and swap elements
16			* within pairs. Each thread presents some object on entry to the
17			* {@link #exchange exchange} method, matches with a partner thread,
18			* and receives its partner's object on return.
19	tim	1.1	*
20			* <p><b>Sample Usage:</b>
21	jsr166	1.29	* Here are the highlights of a class that uses an {@code Exchanger}
22			* to swap buffers between threads so that the thread filling the
23			* buffer gets a freshly emptied one when it needs it, handing off the
24			* filled one to the thread emptying the buffer.
25			* <pre>{@code
26	tim	1.1	* class FillAndEmpty {
27	jsr166	1.29	* Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
28	dl	1.9	* DataBuffer initialEmptyBuffer = ... a made-up type
29			* DataBuffer initialFullBuffer = ...
30	tim	1.1	*
31			* class FillingLoop implements Runnable {
32			* public void run() {
33	dl	1.9	* DataBuffer currentBuffer = initialEmptyBuffer;
34	tim	1.1	* try {
35			* while (currentBuffer != null) {
36			* addToBuffer(currentBuffer);
37	dl	1.30	* if (currentBuffer.isFull())
38	tim	1.1	* currentBuffer = exchanger.exchange(currentBuffer);
39			* }
40	tim	1.7	* } catch (InterruptedException ex) { ... handle ... }
41	tim	1.1	* }
42			* }
43			*
44			* class EmptyingLoop implements Runnable {
45			* public void run() {
46	dl	1.9	* DataBuffer currentBuffer = initialFullBuffer;
47	tim	1.1	* try {
48			* while (currentBuffer != null) {
49			* takeFromBuffer(currentBuffer);
50	dl	1.30	* if (currentBuffer.isEmpty())
51	tim	1.1	* currentBuffer = exchanger.exchange(currentBuffer);
52			* }
53	tim	1.7	* } catch (InterruptedException ex) { ... handle ...}
54	tim	1.1	* }
55			* }
56			*
57			* void start() {
58			* new Thread(new FillingLoop()).start();
59			* new Thread(new EmptyingLoop()).start();
60			* }
61			* }
62	jsr166	1.29	* }</pre>
63	tim	1.1	*
64	jsr166	1.27	* <p>Memory consistency effects: For each pair of threads that
65			* successfully exchange objects via an {@code Exchanger}, actions
66			* prior to the {@code exchange()} in each thread
67			* <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
68			* those subsequent to a return from the corresponding {@code exchange()}
69			* in the other thread.
70	brian	1.22	*
71	tim	1.1	* @since 1.5
72	dl	1.16	* @author Doug Lea and Bill Scherer and Michael Scott
73	dl	1.11	* @param <V> The type of objects that may be exchanged
74	tim	1.1	*/
75			public class Exchanger<V> {
76	dl	1.16	/*
77			* The underlying idea is to use a stack to hold nodes containing
78			* pairs of items to be exchanged. Except that:
79			*
80			* * Only one element of the pair is known on creation by a
81			* first-arriving thread; the other is a "hole" waiting to be
82			* filled in. This is a degenerate form of the dual stacks
83			* described in "Nonblocking Concurrent Objects with Condition
84			* Synchronization", by W. N. Scherer III and M. L. Scott.
85			* 18th Annual Conf. on Distributed Computing, Oct. 2004.
86			* It is "degenerate" in that both the items and the holes
87			* are shared in the same nodes.
88			*
89			* * There isn't really a stack here! There can't be -- if two
90			* nodes were both in the stack, they should cancel themselves
91			* out by combining. So that's what we do. The 0th element of
92			* the "arena" array serves only as the top of stack. The
93			* remainder of the array is a form of the elimination backoff
94			* collision array described in "A Scalable Lock-free Stack
95			* Algorithm", by D. Hendler, N. Shavit, and L. Yerushalmi.
96			* 16th ACM Symposium on Parallelism in Algorithms and
97			* Architectures, June 2004. Here, threads spin (using short
98			* timed waits with exponential backoff) looking for each
99			* other. If they fail to find others waiting, they try the
100			* top spot again. As shown in that paper, this always
101			* converges.
102			*
103			* The backoff elimination mechanics never come into play in
104			* common usages where only two threads ever meet to exchange
105			* items, but they prevent contention bottlenecks when an
106			* exchanger is used by a large number of threads.
107	dl	1.30	*
108			* For more details, see the paper "A Scalable Elimination-based
109			* Exchange Channel" by William Scherer, Doug Lea, and Michael
110			* Scott in Proceedings of SCOOL05 workshop. Available at:
111			* http://hdl.handle.net/1802/2104
112	dl	1.16	*/
113	dl	1.2
114	dl	1.32	/** The number of CPUs, for sizing and spin control */
115			static final int NCPUS = Runtime.getRuntime().availableProcessors();
116
117	jsr166	1.17	/**
118	dl	1.16	* Size of collision space. Using a size of half the number of
119			* CPUs provides enough space for threads to find each other but
120			* not so much that it would always require one or more to time
121			* out to become unstuck. Note that the arena array holds SIZE+1
122			* elements, to include the top-of-stack slot.
123			*/
124	dl	1.32	private static final int SIZE = (NCPUS + 1) / 2;
125
126			/**
127			* The number of times to spin before blocking in timed waits.
128			* The value is empirically derived -- it works well across a
129			* variety of processors and OSes. Empirically, the best value
130			* seems not to vary with number of CPUs (beyond 2) so is just
131			* a constant.
132			*/
133			static final int maxTimedSpins = (NCPUS < 2)? 0 : 16;
134
135			/**
136			* The number of times to spin before blocking in untimed waits.
137			* This is greater than timed value because untimed waits spin
138			* faster since they don't need to check times on each spin.
139			*/
140			static final int maxUntimedSpins = maxTimedSpins * 32;
141
142			/**
143			* The number of nanoseconds for which it is faster to spin
144			* rather than to use timed park. A rough estimate suffices.
145			*/
146			static final long spinForTimeoutThreshold = 1000L;
147	jsr166	1.15
148	dl	1.2	/**
149	dl	1.16	* Base unit in nanoseconds for backoffs. Must be a power of two.
150			* Should be small because backoffs exponentially increase from
151			* base.
152	dl	1.2	*/
153	dl	1.16	private static final long BACKOFF_BASE = 128L;
154	dl	1.2
155	jsr166	1.17	/**
156	dl	1.16	* Sentinel item representing cancellation. This value is placed
157			* in holes on cancellation, and used as a return value from Node
158			* methods to indicate failure to set or get hole.
159	jsr166	1.17	*/
160	dl	1.16	static final Object FAIL = new Object();
161
162	jsr166	1.17	/**
163	dl	1.16	* The collision arena. arena[0] is used as the top of the stack.
164			* The remainder is used as the collision elimination space.
165	dl	1.3	*/
166	dl	1.30	private final AtomicReference<Node>[] arena;
167	dl	1.5
168	dl	1.16	/** Generator for random backoffs and delays. */
169			private final Random random = new Random();
170	dl	1.5
171	dl	1.16	/**
172			* Creates a new Exchanger.
173			*/
174			public Exchanger() {
175	dl	1.30	arena = (AtomicReference<Node>[]) new AtomicReference[SIZE + 1];
176	dl	1.16	for (int i = 0; i < arena.length; ++i)
177	dl	1.30	arena[i] = new AtomicReference<Node>();
178	dl	1.16	}
179	dl	1.2
180	dl	1.16	/**
181			* Main exchange function, handling the different policy variants.
182			* Uses Object, not "V" as argument and return value to simplify
183			* handling of internal sentinel values. Callers from public
184			* methods cast accordingly.
185	dl	1.30	*
186			* @param item the item to exchange
187			* @param timed true if the wait is timed
188			* @param nanos if timed, the maximum wait time
189			* @return the other thread's item
190	dl	1.16	*/
191			private Object doExchange(Object item, boolean timed, long nanos)
192			throws InterruptedException, TimeoutException {
193			Node me = new Node(item);
194	dl	1.30	long lastTime = timed ? System.nanoTime() : 0;
195	dl	1.16	int idx = 0; // start out at slot representing top
196			int backoff = 0; // increases on failure to occupy a slot
197
198			for (;;) {
199	dl	1.30	AtomicReference<Node> slot = arena[idx];
200	dl	1.16
201			// If this slot is already occupied, there is a waiting item...
202			Node you = slot.get();
203			if (you != null) {
204			Object v = you.fillHole(item);
205			slot.compareAndSet(you, null);
206			if (v != FAIL) // ... unless it was cancelled
207			return v;
208	dl	1.2	}
209
210	dl	1.16	// Try to occupy this slot
211			if (slot.compareAndSet(null, me)) {
212			// If this is top slot, use regular wait, else backoff-wait
213			Object v = ((idx == 0)?
214			me.waitForHole(timed, nanos) :
215			me.waitForHole(true, randomDelay(backoff)));
216			slot.compareAndSet(me, null);
217			if (v != FAIL)
218			return v;
219			if (Thread.interrupted())
220			throw new InterruptedException();
221			if (timed) {
222			long now = System.nanoTime();
223			nanos -= now - lastTime;
224			lastTime = now;
225			if (nanos <= 0)
226			throw new TimeoutException();
227			}
228	dl	1.2
229	dl	1.16	me = new Node(item); // Throw away nodes on failure
230			if (backoff < SIZE - 1) // Increase or stay saturated
231			++backoff;
232			idx = 0; // Restart at top
233	dl	1.2	}
234
235	dl	1.16	else // Retry with a random non-top slot <= backoff
236			idx = 1 + random.nextInt(backoff + 1);
237	dl	1.2
238			}
239			}
240	tim	1.1
241			/**
242	jsr166	1.31	* Returns a random delay less than (base times (2 raised to backoff)).
243	dl	1.16	*/
244			private long randomDelay(int backoff) {
245			return ((BACKOFF_BASE << backoff) - 1) & random.nextInt();
246			}
247
248			/**
249			* Nodes hold partially exchanged data. This class
250			* opportunistically subclasses AtomicReference to represent the
251			* hole. So get() returns hole, and compareAndSet CAS'es value
252			* into hole. Note that this class cannot be parameterized as V
253			* because the sentinel value FAIL is only of type Object.
254	jsr166	1.15	*/
255	dl	1.16	static final class Node extends AtomicReference<Object> {
256	dl	1.20	private static final long serialVersionUID = -3221313401284163686L;
257	jsr166	1.21
258	dl	1.16	/** The element offered by the Thread creating this node. */
259			final Object item;
260	jsr166	1.31
261	dl	1.16	/** The Thread creating this node. */
262			final Thread waiter;
263
264			/**
265			* Creates node with given item and empty hole.
266	jsr166	1.31	*
267			* @param item the item
268	dl	1.16	*/
269			Node(Object item) {
270			this.item = item;
271			waiter = Thread.currentThread();
272			}
273
274			/**
275			* Tries to fill in hole. On success, wakes up the waiter.
276	jsr166	1.31	*
277			* @param val the value to place in hole
278			* @return on success, the item; on failure, FAIL
279	dl	1.16	*/
280			Object fillHole(Object val) {
281			if (compareAndSet(null, val)) {
282			LockSupport.unpark(waiter);
283			return item;
284			}
285			return FAIL;
286			}
287
288			/**
289	jsr166	1.17	* Waits for and gets the hole filled in by another thread.
290			* Fails if timed out or interrupted before hole filled.
291	dl	1.30	*
292			* @param timed true if the wait is timed
293			* @param nanos if timed, the maximum wait time
294			* @return on success, the hole; on failure, FAIL
295	dl	1.16	*/
296			Object waitForHole(boolean timed, long nanos) {
297	dl	1.30	long lastTime = timed ? System.nanoTime() : 0;
298	dl	1.32	int spins = timed? maxTimedSpins : maxUntimedSpins;
299	dl	1.18	Object h;
300			while ((h = get()) == null) {
301			// If interrupted or timed out, try to cancel by
302			// CASing FAIL as hole value.
303			if (Thread.currentThread().isInterrupted() \|\|
304	dl	1.30	(timed && nanos <= 0)) {
305			if (compareAndSet(null, FAIL))
306			return FAIL;
307	dl	1.32	} else {
308			if (timed) {
309			long now = System.nanoTime();
310			nanos -= now - lastTime;
311			lastTime = now;
312			}
313			if (spins > 0)
314			--spins;
315			else if (!timed)
316			LockSupport.park();
317			else if (nanos > spinForTimeoutThreshold)
318			LockSupport.parkNanos(nanos);
319	dl	1.16	}
320			}
321			return h;
322			}
323	tim	1.1	}
324
325			/**
326			* Waits for another thread to arrive at this exchange point (unless
327	jsr166	1.31	* the current thread is {@link Thread#interrupt interrupted}),
328	tim	1.1	* and then transfers the given object to it, receiving its object
329			* in return.
330	jsr166	1.17	*
331	tim	1.1	* <p>If another thread is already waiting at the exchange point then
332			* it is resumed for thread scheduling purposes and receives the object
333			* passed in by the current thread. The current thread returns immediately,
334			* receiving the object passed to the exchange by that other thread.
335	jsr166	1.17	*
336	jsr166	1.15	* <p>If no other thread is already waiting at the exchange then the
337	tim	1.1	* current thread is disabled for thread scheduling purposes and lies
338			* dormant until one of two things happens:
339			* <ul>
340			* <li>Some other thread enters the exchange; or
341			* <li>Some other thread {@link Thread#interrupt interrupts} the current
342			* thread.
343			* </ul>
344			* <p>If the current thread:
345			* <ul>
346	jsr166	1.15	* <li>has its interrupted status set on entry to this method; or
347	tim	1.1	* <li>is {@link Thread#interrupt interrupted} while waiting
348	jsr166	1.15	* for the exchange,
349	tim	1.1	* </ul>
350	jsr166	1.15	* then {@link InterruptedException} is thrown and the current thread's
351			* interrupted status is cleared.
352	tim	1.1	*
353			* @param x the object to exchange
354	dl	1.30	* @return the object provided by the other thread
355			* @throws InterruptedException if the current thread was
356			* interrupted while waiting
357	jsr166	1.15	*/
358	tim	1.1	public V exchange(V x) throws InterruptedException {
359	dl	1.2	try {
360	dl	1.16	return (V)doExchange(x, false, 0);
361	jsr166	1.15	} catch (TimeoutException cannotHappen) {
362	dl	1.2	throw new Error(cannotHappen);
363			}
364	tim	1.1	}
365
366			/**
367			* Waits for another thread to arrive at this exchange point (unless
368	jsr166	1.31	* the current thread is {@link Thread#interrupt interrupted} or
369			* the specified waiting time elapses), and then transfers the given
370			* object to it, receiving its object in return.
371	tim	1.1	*
372			* <p>If another thread is already waiting at the exchange point then
373			* it is resumed for thread scheduling purposes and receives the object
374			* passed in by the current thread. The current thread returns immediately,
375			* receiving the object passed to the exchange by that other thread.
376			*
377	jsr166	1.15	* <p>If no other thread is already waiting at the exchange then the
378	tim	1.1	* current thread is disabled for thread scheduling purposes and lies
379			* dormant until one of three things happens:
380			* <ul>
381			* <li>Some other thread enters the exchange; or
382			* <li>Some other thread {@link Thread#interrupt interrupts} the current
383			* thread; or
384			* <li>The specified waiting time elapses.
385			* </ul>
386			* <p>If the current thread:
387			* <ul>
388	jsr166	1.15	* <li>has its interrupted status set on entry to this method; or
389	tim	1.1	* <li>is {@link Thread#interrupt interrupted} while waiting
390	jsr166	1.15	* for the exchange,
391	tim	1.1	* </ul>
392	jsr166	1.15	* then {@link InterruptedException} is thrown and the current thread's
393			* interrupted status is cleared.
394	tim	1.1	*
395			* <p>If the specified waiting time elapses then {@link TimeoutException}
396			* is thrown.
397	jsr166	1.15	* If the time is
398	tim	1.1	* less than or equal to zero, the method will not wait at all.
399			*
400			* @param x the object to exchange
401			* @param timeout the maximum time to wait
402	dl	1.30	* @param unit the time unit of the <tt>timeout</tt> argument
403			* @return the object provided by the other thread
404			* @throws InterruptedException if the current thread was
405			* interrupted while waiting
406			* @throws TimeoutException if the specified waiting time elapses
407			* before another thread enters the exchange
408	jsr166	1.15	*/
409			public V exchange(V x, long timeout, TimeUnit unit)
410	tim	1.1	throws InterruptedException, TimeoutException {
411	dl	1.16	return (V)doExchange(x, true, unit.toNanos(timeout));
412	tim	1.1	}
413			}