summaryrefslogtreecommitdiff
path: root/libjava/classpath/external/jsr166/java/util/concurrent/Exchanger.java
blob: fb917f4321eabccea6e806145a90cb3ff8833919 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
/*
 * 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.atomic.*;
import java.util.concurrent.locks.LockSupport;

/**
 * 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.  An Exchanger may be
 * viewed as a bidirectional form of a {@link SynchronousQueue}.
 * Exchangers may be useful in applications such as genetic algorithms
 * and pipeline designs.
 *
 * <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> {
    /*
     * Algorithm Description:
     *
     * The basic idea is to maintain a "slot", which is a reference to
     * a Node containing both an Item to offer and a "hole" waiting to
     * get filled in.  If an incoming "occupying" thread sees that the
     * slot is null, it CAS'es (compareAndSets) a Node there and waits
     * for another to invoke exchange.  That second "fulfilling" thread
     * sees that the slot is non-null, and so CASes it back to null,
     * also exchanging items by CASing the hole, plus waking up the
     * occupying thread if it is blocked.  In each case CAS'es may
     * fail because a slot at first appears non-null but is null upon
     * CAS, or vice-versa.  So threads may need to retry these
     * actions.
     *
     * This simple approach works great when there are only a few
     * threads using an Exchanger, but performance rapidly
     * deteriorates due to CAS contention on the single slot when
     * there are lots of threads using an exchanger.  So instead we use
     * an "arena"; basically a kind of hash table with a dynamically
     * varying number of slots, any one of which can be used by
     * threads performing an exchange.  Incoming threads pick slots
     * based on a hash of their Thread ids.  If an incoming thread
     * fails to CAS in its chosen slot, it picks an alternative slot
     * instead.  And similarly from there.  If a thread successfully
     * CASes into a slot but no other thread arrives, it tries
     * another, heading toward the zero slot, which always exists even
     * if the table shrinks.  The particular mechanics controlling this
     * are as follows:
     *
     * Waiting: Slot zero is special in that it is the only slot that
     * exists when there is no contention.  A thread occupying slot
     * zero will block if no thread fulfills it after a short spin.
     * In other cases, occupying threads eventually give up and try
     * another slot.  Waiting threads spin for a while (a period that
     * should be a little less than a typical context-switch time)
     * before either blocking (if slot zero) or giving up (if other
     * slots) and restarting.  There is no reason for threads to block
     * unless there are unlikely to be any other threads present.
     * Occupants are mainly avoiding memory contention so sit there
     * quietly polling for a shorter period than it would take to
     * block and then unblock them.  Non-slot-zero waits that elapse
     * because of lack of other threads waste around one extra
     * context-switch time per try, which is still on average much
     * faster than alternative approaches.
     *
     * Sizing: Usually, using only a few slots suffices to reduce
     * contention.  Especially with small numbers of threads, using
     * too many slots can lead to just as poor performance as using
     * too few of them, and there's not much room for error.  The
     * variable "max" maintains the number of slots actually in
     * use.  It is increased when a thread sees too many CAS
     * failures.  (This is analogous to resizing a regular hash table
     * based on a target load factor, except here, growth steps are
     * just one-by-one rather than proportional.)  Growth requires
     * contention failures in each of three tried slots.  Requiring
     * multiple failures for expansion copes with the fact that some
     * failed CASes are not due to contention but instead to simple
     * races between two threads or thread pre-emptions occurring
     * between reading and CASing.  Also, very transient peak
     * contention can be much higher than the average sustainable
     * levels.  The max limit is decreased on average 50% of the times
     * that a non-slot-zero wait elapses without being fulfilled.
     * Threads experiencing elapsed waits move closer to zero, so
     * eventually find existing (or future) threads even if the table
     * has been shrunk due to inactivity.  The chosen mechanics and
     * thresholds for growing and shrinking are intrinsically
     * entangled with indexing and hashing inside the exchange code,
     * and can't be nicely abstracted out.
     *
     * Hashing: Each thread picks its initial slot to use in accord
     * with a simple hashcode.  The sequence is the same on each
     * encounter by any given thread, but effectively random across
     * threads.  Using arenas encounters the classic cost vs quality
     * tradeoffs of all hash tables.  Here, we use a one-step FNV-1a
     * hash code based on the current thread's Thread.getId(), along
     * with a cheap approximation to a mod operation to select an
     * index.  The downside of optimizing index selection in this way
     * is that the code is hardwired to use a maximum table size of
     * 32.  But this value more than suffices for known platforms and
     * applications.
     *
     * Probing: On sensed contention of a selected slot, we probe
     * sequentially through the table, analogously to linear probing
     * after collision in a hash table.  (We move circularly, in
     * reverse order, to mesh best with table growth and shrinkage
     * rules.)  Except that to minimize the effects of false-alarms
     * and cache thrashing, we try the first selected slot twice
     * before moving.
     *
     * Padding: Even with contention management, slots are heavily
     * contended, so use cache-padding to avoid poor memory
     * performance.  Because of this, slots are lazily constructed
     * only when used, to avoid wasting this space unnecessarily.
     * While isolation of locations is not much of an issue at first
     * in an application, as time goes on and garbage-collectors
     * perform compaction, slots are very likely to be moved adjacent
     * to each other, which can cause much thrashing of cache lines on
     * MPs unless padding is employed.
     *
     * This is an improvement of the algorithm described in 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 */
    private static final int NCPU = Runtime.getRuntime().availableProcessors();

    /**
     * The capacity of the arena.  Set to a value that provides more
     * than enough space to handle contention.  On small machines
     * most slots won't be used, but it is still not wasted because
     * the extra space provides some machine-level address padding
     * to minimize interference with heavily CAS'ed Slot locations.
     * And on very large machines, performance eventually becomes
     * bounded by memory bandwidth, not numbers of threads/CPUs.
     * This constant cannot be changed without also modifying
     * indexing and hashing algorithms.
     */
    private static final int CAPACITY = 32;

    /**
     * The value of "max" that will hold all threads without
     * contention.  When this value is less than CAPACITY, some
     * otherwise wasted expansion can be avoided.
     */
    private static final int FULL =
        Math.max(0, Math.min(CAPACITY, NCPU / 2) - 1);

    /**
     * The number of times to spin (doing nothing except polling a
     * memory location) before blocking or giving up while waiting to
     * be fulfilled.  Should be zero on uniprocessors.  On
     * multiprocessors, this value should be large enough so that two
     * threads exchanging items as fast as possible block only when
     * one of them is stalled (due to GC or preemption), but not much
     * longer, to avoid wasting CPU resources.  Seen differently, this
     * value is a little over half the number of cycles of an average
     * context switch time on most systems.  The value here is
     * approximately the average of those across a range of tested
     * systems.
     */
    private static final int SPINS = (NCPU == 1) ? 0 : 2000;

    /**
     * The number of times to spin before blocking in timed waits.
     * Timed waits spin more slowly because checking the time takes
     * time.  The best value relies mainly on the relative rate of
     * System.nanoTime vs memory accesses.  The value is empirically
     * derived to work well across a variety of systems.
     */
    private static final int TIMED_SPINS = SPINS / 20;

    /**
     * Sentinel item representing cancellation of a wait due to
     * interruption, timeout, or elapsed spin-waits.  This value is
     * placed in holes on cancellation, and used as a return value
     * from waiting methods to indicate failure to set or get hole.
     */
    private static final Object CANCEL = new Object();

    /**
     * Value representing null arguments/returns from public
     * methods.  This disambiguates from internal requirement that
     * holes start out as null to mean they are not yet set.
     */
    private static final Object NULL_ITEM = new Object();

    /**
     * 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.  This class cannot be parameterized as "V" because
     * of the use of non-V CANCEL sentinels.
     */
    private static final class Node extends AtomicReference<Object> {
        /** The element offered by the Thread creating this node. */
        public final Object item;

        /** The Thread waiting to be signalled; null until waiting. */
        public volatile Thread waiter;

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

    /**
     * A Slot is an AtomicReference with heuristic padding to lessen
     * cache effects of this heavily CAS'ed location.  While the
     * padding adds noticeable space, all slots are created only on
     * demand, and there will be more than one of them only when it
     * would improve throughput more than enough to outweigh using
     * extra space.
     */
    private static final class Slot extends AtomicReference<Object> {
        // Improve likelihood of isolation on <= 64 byte cache lines
        long q0, q1, q2, q3, q4, q5, q6, q7, q8, q9, qa, qb, qc, qd, qe;
    }

    /**
     * Slot array.  Elements are lazily initialized when needed.
     * Declared volatile to enable double-checked lazy construction.
     */
    private volatile Slot[] arena = new Slot[CAPACITY];

    /**
     * The maximum slot index being used.  The value sometimes
     * increases when a thread experiences too many CAS contentions,
     * and sometimes decreases when a spin-wait elapses.  Changes
     * are performed only via compareAndSet, to avoid stale values
     * when a thread happens to stall right before setting.
     */
    private final AtomicInteger max = new AtomicInteger();

    /**
     * Main exchange function, handling the different policy variants.
     * Uses Object, not "V" as argument and return value to simplify
     * handling of sentinel values.  Callers from public methods decode
     * and cast accordingly.
     *
     * @param item the (non-null) 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, or CANCEL if interrupted or timed out
     */
    private Object doExchange(Object item, boolean timed, long nanos) {
        Node me = new Node(item);                 // Create in case occupying
        int index = hashIndex();                  // Index of current slot
        int fails = 0;                            // Number of CAS failures

        for (;;) {
            Object y;                             // Contents of current slot
            Slot slot = arena[index];
            if (slot == null)                     // Lazily initialize slots
                createSlot(index);                // Continue loop to reread
            else if ((y = slot.get()) != null &&  // Try to fulfill
                     slot.compareAndSet(y, null)) {
                Node you = (Node)y;               // Transfer item
                if (you.compareAndSet(null, item)) {
                    LockSupport.unpark(you.waiter);
                    return you.item;
                }                                 // Else cancelled; continue
            }
            else if (y == null &&                 // Try to occupy
                     slot.compareAndSet(null, me)) {
                if (index == 0)                   // Blocking wait for slot 0
                    return timed? awaitNanos(me, slot, nanos): await(me, slot);
                Object v = spinWait(me, slot);    // Spin wait for non-0
                if (v != CANCEL)
                    return v;
                me = new Node(item);              // Throw away cancelled node
                int m = max.get();
                if (m > (index >>>= 1))           // Decrease index
                    max.compareAndSet(m, m - 1);  // Maybe shrink table
            }
            else if (++fails > 1) {               // Allow 2 fails on 1st slot
                int m = max.get();
                if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
                    index = m + 1;                // Grow on 3rd failed slot
                else if (--index < 0)
                    index = m;                    // Circularly traverse
            }
        }
    }

    /**
     * Returns a hash index for the current thread.  Uses a one-step
     * FNV-1a hash code (http://www.isthe.com/chongo/tech/comp/fnv/)
     * based on the current thread's Thread.getId().  These hash codes
     * have more uniform distribution properties with respect to small
     * moduli (here 1-31) than do other simple hashing functions.
     *
     * <p>To return an index between 0 and max, we use a cheap
     * approximation to a mod operation, that also corrects for bias
     * due to non-power-of-2 remaindering (see {@link
     * java.util.Random#nextInt}).  Bits of the hashcode are masked
     * with "nbits", the ceiling power of two of table size (looked up
     * in a table packed into three ints).  If too large, this is
     * retried after rotating the hash by nbits bits, while forcing new
     * top bit to 0, which guarantees eventual termination (although
     * with a non-random-bias).  This requires an average of less than
     * 2 tries for all table sizes, and has a maximum 2% difference
     * from perfectly uniform slot probabilities when applied to all
     * possible hash codes for sizes less than 32.
     *
     * @return a per-thread-random index, 0 <= index < max
     */
    private final int hashIndex() {
        long id = Thread.currentThread().getId();
        int hash = (((int)(id ^ (id >>> 32))) ^ 0x811c9dc5) * 0x01000193;

        int m = max.get();
        int nbits = (((0xfffffc00  >> m) & 4) | // Compute ceil(log2(m+1))
                     ((0x000001f8 >>> m) & 2) | // The constants hold
                     ((0xffff00f2 >>> m) & 1)); // a lookup table
        int index;
        while ((index = hash & ((1 << nbits) - 1)) > m)       // May retry on
            hash = (hash >>> nbits) | (hash << (33 - nbits)); // non-power-2 m
        return index;
    }

    /**
     * Creates a new slot at given index.  Called only when the slot
     * appears to be null.  Relies on double-check using builtin
     * locks, since they rarely contend.  This in turn relies on the
     * arena array being declared volatile.
     *
     * @param index the index to add slot at
     */
    private void createSlot(int index) {
        // Create slot outside of lock to narrow sync region
        Slot newSlot = new Slot();
        Slot[] a = arena;
        synchronized (a) {
            if (a[index] == null)
                a[index] = newSlot;
        }
    }

    /**
     * Tries to cancel a wait for the given node waiting in the given
     * slot, if so, helping clear the node from its slot to avoid
     * garbage retention.
     *
     * @param node the waiting node
     * @param the slot it is waiting in
     * @return true if successfully cancelled
     */
    private static boolean tryCancel(Node node, Slot slot) {
        if (!node.compareAndSet(null, CANCEL))
            return false;
        if (slot.get() == node) // pre-check to minimize contention
            slot.compareAndSet(node, null);
        return true;
    }

    // Three forms of waiting. Each just different enough not to merge
    // code with others.

    /**
     * Spin-waits for hole for a non-0 slot.  Fails if spin elapses
     * before hole filled.  Does not check interrupt, relying on check
     * in public exchange method to abort if interrupted on entry.
     *
     * @param node the waiting node
     * @return on success, the hole; on failure, CANCEL
     */
    private static Object spinWait(Node node, Slot slot) {
        int spins = SPINS;
        for (;;) {
            Object v = node.get();
            if (v != null)
                return v;
            else if (spins > 0)
                --spins;
            else
                tryCancel(node, slot);
        }
    }

    /**
     * Waits for (by spinning and/or blocking) and gets the hole
     * filled in by another thread.  Fails if interrupted before
     * hole filled.
     *
     * When a node/thread is about to block, it sets its waiter field
     * and then rechecks state at least one more time before actually
     * parking, thus covering race vs fulfiller noticing that waiter
     * is non-null so should be woken.
     *
     * Thread interruption status is checked only surrounding calls to
     * park.  The caller is assumed to have checked interrupt status
     * on entry.
     *
     * @param node the waiting node
     * @return on success, the hole; on failure, CANCEL
     */
    private static Object await(Node node, Slot slot) {
        Thread w = Thread.currentThread();
        int spins = SPINS;
        for (;;) {
            Object v = node.get();
            if (v != null)
                return v;
            else if (spins > 0)                 // Spin-wait phase
                --spins;
            else if (node.waiter == null)       // Set up to block next
                node.waiter = w;
            else if (w.isInterrupted())         // Abort on interrupt
                tryCancel(node, slot);
            else                                // Block
                LockSupport.park(node);
        }
    }

    /**
     * Waits for (at index 0) and gets the hole filled in by another
     * thread.  Fails if timed out or interrupted before hole filled.
     * Same basic logic as untimed version, but a bit messier.
     *
     * @param node the waiting node
     * @param nanos the wait time
     * @return on success, the hole; on failure, CANCEL
     */
    private Object awaitNanos(Node node, Slot slot, long nanos) {
        int spins = TIMED_SPINS;
        long lastTime = 0;
        Thread w = null;
        for (;;) {
            Object v = node.get();
            if (v != null)
                return v;
            long now = System.nanoTime();
            if (w == null)
                w = Thread.currentThread();
            else
                nanos -= now - lastTime;
            lastTime = now;
            if (nanos > 0) {
                if (spins > 0)
                    --spins;
                else if (node.waiter == null)
                    node.waiter = w;
                else if (w.isInterrupted())
                    tryCancel(node, slot);
                else
                    LockSupport.parkNanos(node, nanos);
            }
            else if (tryCancel(node, slot) && !w.isInterrupted())
                return scanOnTimeout(node);
        }
    }

    /**
     * Sweeps through arena checking for any waiting threads.  Called
     * only upon return from timeout while waiting in slot 0.  When a
     * thread gives up on a timed wait, it is possible that a
     * previously-entered thread is still waiting in some other
     * slot.  So we scan to check for any.  This is almost always
     * overkill, but decreases the likelihood of timeouts when there
     * are other threads present to far less than that in lock-based
     * exchangers in which earlier-arriving threads may still be
     * waiting on entry locks.
     *
     * @param node the waiting node
     * @return another thread's item, or CANCEL
     */
    private Object scanOnTimeout(Node node) {
        Object y;
        for (int j = arena.length - 1; j >= 0; --j) {
            Slot slot = arena[j];
            if (slot != null) {
                while ((y = slot.get()) != null) {
                    if (slot.compareAndSet(y, null)) {
                        Node you = (Node)y;
                        if (you.compareAndSet(null, node.item)) {
                            LockSupport.unpark(you.waiter);
                            return you.item;
                        }
                    }
                }
            }
        }
        return CANCEL;
    }

    /**
     * Creates a new Exchanger.
     */
    public Exchanger() {
    }

    /**
     * Waits for another thread to arrive at this exchange point (unless
     * the current thread is {@linkplain 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 {@linkplain 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 {@linkplain 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 {
        if (!Thread.interrupted()) {
            Object v = doExchange(x == null? NULL_ITEM : x, false, 0);
            if (v == NULL_ITEM)
                return null;
            if (v != CANCEL)
                return (V)v;
            Thread.interrupted(); // Clear interrupt status on IE throw
        }
        throw new InterruptedException();
    }

    /**
     * Waits for another thread to arrive at this exchange point (unless
     * the current thread is {@linkplain 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 {@linkplain 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 {@linkplain 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 {
        if (!Thread.interrupted()) {
            Object v = doExchange(x == null? NULL_ITEM : x,
                                  true, unit.toNanos(timeout));
            if (v == NULL_ITEM)
                return null;
            if (v != CANCEL)
                return (V)v;
            if (!Thread.interrupted())
                throw new TimeoutException();
        }
        throw new InterruptedException();
    }
}