ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/Exchanger.java
Revision: 1.29
Committed: Fri Sep 16 03:59:07 2005 UTC (18 years, 8 months ago) by jsr166
Branch: MAIN
Changes since 1.28: +7 -8 lines
Log Message:
doc fixes

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.full())
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.empty())
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
109 /**
110 * Size of collision space. Using a size of half the number of
111 * CPUs provides enough space for threads to find each other but
112 * not so much that it would always require one or more to time
113 * out to become unstuck. Note that the arena array holds SIZE+1
114 * elements, to include the top-of-stack slot.
115 */
116 private static final int SIZE =
117 (Runtime.getRuntime().availableProcessors() + 1) / 2;
118
119 /**
120 * Base unit in nanoseconds for backoffs. Must be a power of two.
121 * Should be small because backoffs exponentially increase from
122 * base.
123 */
124 private static final long BACKOFF_BASE = 128L;
125
126 /**
127 * Sentinel item representing cancellation. This value is placed
128 * in holes on cancellation, and used as a return value from Node
129 * methods to indicate failure to set or get hole.
130 */
131 static final Object FAIL = new Object();
132
133 /**
134 * The collision arena. arena[0] is used as the top of the stack.
135 * The remainder is used as the collision elimination space.
136 * Each slot holds an AtomicReference<Node>, but this cannot be
137 * expressed for arrays, so elements are casted on each use.
138 */
139 private final AtomicReference[] arena;
140
141 /** Generator for random backoffs and delays. */
142 private final Random random = new Random();
143
144 /**
145 * Creates a new Exchanger.
146 */
147 public Exchanger() {
148 arena = new AtomicReference[SIZE + 1];
149 for (int i = 0; i < arena.length; ++i)
150 arena[i] = new AtomicReference();
151 }
152
153 /**
154 * Main exchange function, handling the different policy variants.
155 * Uses Object, not "V" as argument and return value to simplify
156 * handling of internal sentinel values. Callers from public
157 * methods cast accordingly.
158 * @param item the item to exchange.
159 * @param timed true if the wait is timed.
160 * @param nanos if timed, the maximum wait time.
161 * @return the other thread's item.
162 */
163 private Object doExchange(Object item, boolean timed, long nanos)
164 throws InterruptedException, TimeoutException {
165 Node me = new Node(item);
166 long lastTime = (timed)? System.nanoTime() : 0;
167 int idx = 0; // start out at slot representing top
168 int backoff = 0; // increases on failure to occupy a slot
169
170 for (;;) {
171 AtomicReference<Node> slot = (AtomicReference<Node>)arena[idx];
172
173 // If this slot is already occupied, there is a waiting item...
174 Node you = slot.get();
175 if (you != null) {
176 Object v = you.fillHole(item);
177 slot.compareAndSet(you, null);
178 if (v != FAIL) // ... unless it was cancelled
179 return v;
180 }
181
182 // Try to occupy this slot
183 if (slot.compareAndSet(null, me)) {
184 // If this is top slot, use regular wait, else backoff-wait
185 Object v = ((idx == 0)?
186 me.waitForHole(timed, nanos) :
187 me.waitForHole(true, randomDelay(backoff)));
188 slot.compareAndSet(me, null);
189 if (v != FAIL)
190 return v;
191 if (Thread.interrupted())
192 throw new InterruptedException();
193 if (timed) {
194 long now = System.nanoTime();
195 nanos -= now - lastTime;
196 lastTime = now;
197 if (nanos <= 0)
198 throw new TimeoutException();
199 }
200
201 me = new Node(item); // Throw away nodes on failure
202 if (backoff < SIZE - 1) // Increase or stay saturated
203 ++backoff;
204 idx = 0; // Restart at top
205 }
206
207 else // Retry with a random non-top slot <= backoff
208 idx = 1 + random.nextInt(backoff + 1);
209
210 }
211 }
212
213 /**
214 * Returns a random delay less than (base times (2 raised to backoff))
215 */
216 private long randomDelay(int backoff) {
217 return ((BACKOFF_BASE << backoff) - 1) & random.nextInt();
218 }
219
220 /**
221 * Nodes hold partially exchanged data. This class
222 * opportunistically subclasses AtomicReference to represent the
223 * hole. So get() returns hole, and compareAndSet CAS'es value
224 * into hole. Note that this class cannot be parameterized as V
225 * because the sentinel value FAIL is only of type Object.
226 */
227 static final class Node extends AtomicReference<Object> {
228 private static final long serialVersionUID = -3221313401284163686L;
229
230 /** The element offered by the Thread creating this node. */
231 final Object item;
232 /** The Thread creating this node. */
233 final Thread waiter;
234
235 /**
236 * Creates node with given item and empty hole.
237 * @param item the item.
238 */
239 Node(Object item) {
240 this.item = item;
241 waiter = Thread.currentThread();
242 }
243
244 /**
245 * Tries to fill in hole. On success, wakes up the waiter.
246 * @param val the value to place in hole.
247 * @return on success, the item; on failure, FAIL.
248 */
249 Object fillHole(Object val) {
250 if (compareAndSet(null, val)) {
251 LockSupport.unpark(waiter);
252 return item;
253 }
254 return FAIL;
255 }
256
257 /**
258 * Waits for and gets the hole filled in by another thread.
259 * Fails if timed out or interrupted before hole filled.
260 * @param timed true if the wait is timed.
261 * @param nanos if timed, the maximum wait time.
262 * @return on success, the hole; on failure, FAIL.
263 */
264 Object waitForHole(boolean timed, long nanos) {
265 long lastTime = (timed)? System.nanoTime() : 0;
266 Object h;
267 while ((h = get()) == null) {
268 // If interrupted or timed out, try to cancel by
269 // CASing FAIL as hole value.
270 if (Thread.currentThread().isInterrupted() ||
271 (timed && nanos <= 0))
272 compareAndSet(null, FAIL);
273 else if (!timed)
274 LockSupport.park();
275 else {
276 LockSupport.parkNanos(nanos);
277 long now = System.nanoTime();
278 nanos -= now - lastTime;
279 lastTime = now;
280 }
281 }
282 return h;
283 }
284 }
285
286 /**
287 * Waits for another thread to arrive at this exchange point (unless
288 * it is {@link Thread#interrupt interrupted}),
289 * and then transfers the given object to it, receiving its object
290 * in return.
291 *
292 * <p>If another thread is already waiting at the exchange point then
293 * it is resumed for thread scheduling purposes and receives the object
294 * passed in by the current thread. The current thread returns immediately,
295 * receiving the object passed to the exchange by that other thread.
296 *
297 * <p>If no other thread is already waiting at the exchange then the
298 * current thread is disabled for thread scheduling purposes and lies
299 * dormant until one of two things happens:
300 * <ul>
301 * <li>Some other thread enters the exchange; or
302 * <li>Some other thread {@link Thread#interrupt interrupts} the current
303 * thread.
304 * </ul>
305 * <p>If the current thread:
306 * <ul>
307 * <li>has its interrupted status set on entry to this method; or
308 * <li>is {@link Thread#interrupt interrupted} while waiting
309 * for the exchange,
310 * </ul>
311 * then {@link InterruptedException} is thrown and the current thread's
312 * interrupted status is cleared.
313 *
314 * @param x the object to exchange
315 * @return the object provided by the other thread.
316 * @throws InterruptedException if current thread was interrupted
317 * while waiting
318 */
319 public V exchange(V x) throws InterruptedException {
320 try {
321 return (V)doExchange(x, false, 0);
322 } catch (TimeoutException cannotHappen) {
323 throw new Error(cannotHappen);
324 }
325 }
326
327 /**
328 * Waits for another thread to arrive at this exchange point (unless
329 * it is {@link Thread#interrupt interrupted}, or the specified waiting
330 * time elapses),
331 * and then transfers the given object to it, receiving its object
332 * in return.
333 *
334 * <p>If another thread is already waiting at the exchange point then
335 * it is resumed for thread scheduling purposes and receives the object
336 * passed in by the current thread. The current thread returns immediately,
337 * receiving the object passed to the exchange by that other thread.
338 *
339 * <p>If no other thread is already waiting at the exchange then the
340 * current thread is disabled for thread scheduling purposes and lies
341 * dormant until one of three things happens:
342 * <ul>
343 * <li>Some other thread enters the exchange; or
344 * <li>Some other thread {@link Thread#interrupt interrupts} the current
345 * thread; or
346 * <li>The specified waiting time elapses.
347 * </ul>
348 * <p>If the current thread:
349 * <ul>
350 * <li>has its interrupted status set on entry to this method; or
351 * <li>is {@link Thread#interrupt interrupted} while waiting
352 * for the exchange,
353 * </ul>
354 * then {@link InterruptedException} is thrown and the current thread's
355 * interrupted status is cleared.
356 *
357 * <p>If the specified waiting time elapses then {@link TimeoutException}
358 * is thrown.
359 * If the time is
360 * less than or equal to zero, the method will not wait at all.
361 *
362 * @param x the object to exchange
363 * @param timeout the maximum time to wait
364 * @param unit the time unit of the <tt>timeout</tt> argument.
365 * @return the object provided by the other thread.
366 * @throws InterruptedException if current thread was interrupted
367 * while waiting
368 * @throws TimeoutException if the specified waiting time elapses before
369 * another thread enters the exchange.
370 */
371 public V exchange(V x, long timeout, TimeUnit unit)
372 throws InterruptedException, TimeoutException {
373 return (V)doExchange(x, true, unit.toNanos(timeout));
374 }
375 }