ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/Exchanger.java
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

File Contents

# Content
1 /*
2 * 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 * http://creativecommons.org/licenses/publicdomain
6 */
7
8 package java.util.concurrent;
9 import java.util.concurrent.*; // for javadoc (till 6280605 is fixed)
10 import java.util.concurrent.locks.*;
11 import java.util.concurrent.atomic.*;
12 import java.util.Random;
13
14 /**
15 * 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 *
20 * <p><b>Sample Usage:</b>
21 * 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 * class FillAndEmpty {
27 * Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
28 * DataBuffer initialEmptyBuffer = ... a made-up type
29 * DataBuffer initialFullBuffer = ...
30 *
31 * class FillingLoop implements Runnable {
32 * public void run() {
33 * DataBuffer currentBuffer = initialEmptyBuffer;
34 * try {
35 * while (currentBuffer != null) {
36 * addToBuffer(currentBuffer);
37 * if (currentBuffer.isFull())
38 * currentBuffer = exchanger.exchange(currentBuffer);
39 * }
40 * } catch (InterruptedException ex) { ... handle ... }
41 * }
42 * }
43 *
44 * class EmptyingLoop implements Runnable {
45 * public void run() {
46 * DataBuffer currentBuffer = initialFullBuffer;
47 * try {
48 * while (currentBuffer != null) {
49 * takeFromBuffer(currentBuffer);
50 * if (currentBuffer.isEmpty())
51 * currentBuffer = exchanger.exchange(currentBuffer);
52 * }
53 * } catch (InterruptedException ex) { ... handle ...}
54 * }
55 * }
56 *
57 * void start() {
58 * new Thread(new FillingLoop()).start();
59 * new Thread(new EmptyingLoop()).start();
60 * }
61 * }
62 * }</pre>
63 *
64 * <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 *
71 * @since 1.5
72 * @author Doug Lea and Bill Scherer and Michael Scott
73 * @param <V> The type of objects that may be exchanged
74 */
75 public class Exchanger<V> {
76 /*
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 *
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 */
113
114 /** The number of CPUs, for sizing and spin control */
115 static final int NCPUS = Runtime.getRuntime().availableProcessors();
116
117 /**
118 * 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 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
148 /**
149 * Base unit in nanoseconds for backoffs. Must be a power of two.
150 * Should be small because backoffs exponentially increase from
151 * base.
152 */
153 private static final long BACKOFF_BASE = 128L;
154
155 /**
156 * 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 */
160 static final Object FAIL = new Object();
161
162 /**
163 * The collision arena. arena[0] is used as the top of the stack.
164 * The remainder is used as the collision elimination space.
165 */
166 private final AtomicReference<Node>[] arena;
167
168 /** Generator for random backoffs and delays. */
169 private final Random random = new Random();
170
171 /**
172 * Creates a new Exchanger.
173 */
174 public Exchanger() {
175 arena = (AtomicReference<Node>[]) new AtomicReference[SIZE + 1];
176 for (int i = 0; i < arena.length; ++i)
177 arena[i] = new AtomicReference<Node>();
178 }
179
180 /**
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 *
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 */
191 private Object doExchange(Object item, boolean timed, long nanos)
192 throws InterruptedException, TimeoutException {
193 Node me = new Node(item);
194 long lastTime = timed ? System.nanoTime() : 0;
195 int idx = 0; // start out at slot representing top
196 int backoff = 0; // increases on failure to occupy a slot
197
198 for (;;) {
199 AtomicReference<Node> slot = arena[idx];
200
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 }
209
210 // 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
229 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 }
234
235 else // Retry with a random non-top slot <= backoff
236 idx = 1 + random.nextInt(backoff + 1);
237
238 }
239 }
240
241 /**
242 * Returns a random delay less than (base times (2 raised to backoff)).
243 */
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 */
255 static final class Node extends AtomicReference<Object> {
256 private static final long serialVersionUID = -3221313401284163686L;
257
258 /** The element offered by the Thread creating this node. */
259 final Object item;
260
261 /** The Thread creating this node. */
262 final Thread waiter;
263
264 /**
265 * Creates node with given item and empty hole.
266 *
267 * @param item the item
268 */
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 *
277 * @param val the value to place in hole
278 * @return on success, the item; on failure, FAIL
279 */
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 * Waits for and gets the hole filled in by another thread.
290 * Fails if timed out or interrupted before hole filled.
291 *
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 */
296 Object waitForHole(boolean timed, long nanos) {
297 long lastTime = timed ? System.nanoTime() : 0;
298 int spins = timed? maxTimedSpins : maxUntimedSpins;
299 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 (timed && nanos <= 0)) {
305 if (compareAndSet(null, FAIL))
306 return FAIL;
307 } 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 }
320 }
321 return h;
322 }
323 }
324
325 /**
326 * Waits for another thread to arrive at this exchange point (unless
327 * the current thread is {@link Thread#interrupt interrupted}),
328 * and then transfers the given object to it, receiving its object
329 * in return.
330 *
331 * <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 *
336 * <p>If no other thread is already waiting at the exchange then the
337 * 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 * <li>has its interrupted status set on entry to this method; or
347 * <li>is {@link Thread#interrupt interrupted} while waiting
348 * for the exchange,
349 * </ul>
350 * then {@link InterruptedException} is thrown and the current thread's
351 * interrupted status is cleared.
352 *
353 * @param x the object to exchange
354 * @return the object provided by the other thread
355 * @throws InterruptedException if the current thread was
356 * interrupted while waiting
357 */
358 public V exchange(V x) throws InterruptedException {
359 try {
360 return (V)doExchange(x, false, 0);
361 } catch (TimeoutException cannotHappen) {
362 throw new Error(cannotHappen);
363 }
364 }
365
366 /**
367 * Waits for another thread to arrive at this exchange point (unless
368 * 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 *
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 * <p>If no other thread is already waiting at the exchange then the
378 * 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 * <li>has its interrupted status set on entry to this method; or
389 * <li>is {@link Thread#interrupt interrupted} while waiting
390 * for the exchange,
391 * </ul>
392 * then {@link InterruptedException} is thrown and the current thread's
393 * interrupted status is cleared.
394 *
395 * <p>If the specified waiting time elapses then {@link TimeoutException}
396 * is thrown.
397 * If the time is
398 * 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 * @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 */
409 public V exchange(V x, long timeout, TimeUnit unit)
410 throws InterruptedException, TimeoutException {
411 return (V)doExchange(x, true, unit.toNanos(timeout));
412 }
413 }