content: /*
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   *
   * This code is free software; you can redistribute it and/or modify it
   * under the terms of the GNU General Public License version 2 only, as
   * published by the Free Software Foundation.  Oracle designates this
   * particular file as subject to the "Classpath" exception as provided
   * by Oracle in the LICENSE file that accompanied this code.
   *
   * This code is distributed in the hope that it will be useful, but WITHOUT
   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
   * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
   * version 2 for more details (a copy is included in the LICENSE file that
   * accompanied this code).
   *
   * You should have received a copy of the GNU General Public License version
   * 2 along with this work; if not, write to the Free Software Foundation,
   * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
   *
   * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
   * or visit www.oracle.com if you need additional information or have any
   * questions.
   */
  
  /*
   * This file is available under and governed by the GNU General Public
   * License version 2 only, as published by the Free Software Foundation.
   * However, the following notice accompanied the original version of this
   * file:
   *
   * 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/publicdomain/zero/1.0/
   */
  
  package java.util.concurrent;
  
  /**
   * 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 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> {
  
      /*
       * Overview: The core algorithm is, for an exchange "slot",
       * and a participant (caller) with an item:
       *
       * for (;;) {
       *   if (slot is empty) {                       // offer
       *     place item in a Node;
       *     if (can CAS slot from empty to node) {
       *       wait for release;
       *       return matching item in node;
       *     }
       *   }
       *   else if (can CAS slot from node to empty) { // release
       *     get the item in node;
       *     set matching item in node;
       *     release waiting thread;
       *   }
       *   // else retry on CAS failure
       * }
       *
       * This is among the simplest forms of a "dual data structure" --
       * see Scott and Scherer's DISC 04 paper and
       * http://www.cs.rochester.edu/research/synchronization/pseudocode/duals.html
       *
       * This works great in principle. But in practice, like many
       * algorithms centered on atomic updates to a single location, it
       * scales horribly when there are more than a few participants
       * using the same Exchanger. So the implementation instead uses a
       * form of elimination arena, that spreads out this contention by
       * arranging that some threads typically use different slots,
       * while still ensuring that eventually, any two parties will be
       * able to exchange items. That is, we cannot completely partition
       * across threads, but instead give threads arena indices that
       * will on average grow under contention and shrink under lack of
       * contention. We approach this by defining the Nodes that we need
       * anyway as ThreadLocals, and include in them per-thread index
       * and related bookkeeping state. (We can safely reuse per-thread
       * nodes rather than creating them fresh each time because slots
       * alternate between pointing to a node vs null, so cannot
       * encounter ABA problems. However, we do need some care in
       * resetting them between uses.)
       *
       * Implementing an effective arena requires allocating a bunch of
       * space, so we only do so upon detecting contention (except on
       * uniprocessors, where they wouldn't help, so aren't used).
       * Otherwise, exchanges use the single-slot slotExchange method.
       * On contention, not only must the slots be in different
       * locations, but the locations must not encounter memory
       * contention due to being on the same cache line (or more
       * generally, the same coherence unit).  Because, as of this
       * writing, there is no way to determine cacheline size, we define
       * a value that is enough for common platforms.  Additionally,
       * extra care elsewhere is taken to avoid other false/unintended
       * sharing and to enhance locality, including adding padding (via
       * @Contended) to Nodes, embedding "bound" as an Exchanger field,
       * and reworking some park/unpark mechanics compared to
       * LockSupport versions.
       *
       * The arena starts out with only one used slot. We expand the
       * effective arena size by tracking collisions; i.e., failed CASes
       * while trying to exchange. By nature of the above algorithm, the
       * only kinds of collision that reliably indicate contention are
       * when two attempted releases collide -- one of two attempted
       * offers can legitimately fail to CAS without indicating
       * contention by more than one other thread. (Note: it is possible
       * but not worthwhile to more precisely detect contention by
       * reading slot values after CAS failures.)  When a thread has
       * collided at each slot within the current arena bound, it tries
       * to expand the arena size by one. We track collisions within
       * bounds by using a version (sequence) number on the "bound"
       * field, and conservatively reset collision counts when a
       * participant notices that bound has been updated (in either
       * direction).
       *
       * The effective arena size is reduced (when there is more than
       * one slot) by giving up on waiting after a while and trying to
       * decrement the arena size on expiration. The value of "a while"
       * is an empirical matter.  We implement by piggybacking on the
       * use of spin->yield->block that is essential for reasonable
       * waiting performance anyway -- in a busy exchanger, offers are
       * usually almost immediately released, in which case context
       * switching on multiprocessors is extremely slow/wasteful.  Arena
       * waits just omit the blocking part, and instead cancel. The spin
       * count is empirically chosen to be a value that avoids blocking
       * 99% of the time under maximum sustained exchange rates on a
       * range of test machines. Spins and yields entail some limited
       * randomness (using a cheap xorshift) to avoid regular patterns
       * that can induce unproductive grow/shrink cycles. (Using a
       * pseudorandom also helps regularize spin cycle duration by
       * making branches unpredictable.)  Also, during an offer, a
       * waiter can "know" that it will be released when its slot has
       * changed, but cannot yet proceed until match is set.  In the
       * mean time it cannot cancel the offer, so instead spins/yields.
       * Note: It is possible to avoid this secondary check by changing
       * the linearization point to be a CAS of the match field (as done
       * in one case in the Scott & Scherer DISC paper), which also
       * increases asynchrony a bit, at the expense of poorer collision
       * detection and inability to always reuse per-thread nodes. So
       * the current scheme is typically a better tradeoff.
       *
       * On collisions, indices traverse the arena cyclically in reverse
       * order, restarting at the maximum index (which will tend to be
       * sparsest) when bounds change. (On expirations, indices instead
       * are halved until reaching 0.) It is possible (and has been
       * tried) to use randomized, prime-value-stepped, or double-hash
       * style traversal instead of simple cyclic traversal to reduce
       * bunching.  But empirically, whatever benefits these may have
       * don't overcome their added overhead: We are managing operations
       * that occur very quickly unless there is sustained contention,
       * so simpler/faster control policies work better than more
       * accurate but slower ones.
       *
       * Because we use expiration for arena size control, we cannot
       * throw TimeoutExceptions in the timed version of the public
       * exchange method until the arena size has shrunken to zero (or
       * the arena isn't enabled). This may delay response to timeout
       * but is still within spec.
       *
       * Essentially all of the implementation is in methods
       * slotExchange and arenaExchange. These have similar overall
       * structure, but differ in too many details to combine. The
       * slotExchange method uses the single Exchanger field "slot"
       * rather than arena array elements. However, it still needs
       * minimal collision detection to trigger arena construction.
       * (The messiest part is making sure interrupt status and
       * InterruptedExceptions come out right during transitions when
       * both methods may be called. This is done by using null return
       * as a sentinel to recheck interrupt status.)
       *
       * As is too common in this sort of code, methods are monolithic
       * because most of the logic relies on reads of fields that are
       * maintained as local variables so can't be nicely factored --
       * mainly, here, bulky spin->yield->block/cancel code), and
       * heavily dependent on intrinsics (Unsafe) to use inlined
       * embedded CAS and related memory access operations (that tend
       * not to be as readily inlined by dynamic compilers when they are
       * hidden behind other methods that would more nicely name and
       * encapsulate the intended effects). This includes the use of
       * putOrderedX to clear fields of the per-thread Nodes between
       * uses. Note that field Node.item is not declared as volatile
       * even though it is read by releasing threads, because they only
       * do so after CAS operations that must precede access, and all
       * uses by the owning thread are otherwise acceptably ordered by
       * other operations. (Because the actual points of atomicity are
       * slot CASes, it would also be legal for the write to Node.match
       * in a release to be weaker than a full volatile write. However,
       * this is not done because it could allow further postponement of
       * the write, delaying progress.)
       */
  
      /**
       * The byte distance (as a shift value) between any two used slots
       * in the arena.  1 << ASHIFT should be at least cacheline size.
       */
      private static final int ASHIFT = 7;
  
      /**
       * The maximum supported arena index. The maximum allocatable
       * arena size is MMASK + 1. Must be a power of two minus one, less
       * than (1<<(31-ASHIFT)). The cap of 255 (0xff) more than suffices
       * for the expected scaling limits of the main algorithms.
       */
      private static final int MMASK = 0xff;
  
      /**
       * Unit for sequence/version bits of bound field. Each successful
       * change to the bound also adds SEQ.
       */
      private static final int SEQ = MMASK + 1;
  
      /** The number of CPUs, for sizing and spin control */
      private static final int NCPU = Runtime.getRuntime().availableProcessors();
  
      /**
       * The maximum slot index of the arena: The number of slots that
       * can in principle hold all threads without contention, or at
       * most the maximum indexable value.
       */
      static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;
  
      /**
       * The bound for spins while waiting for a match. The actual
       * number of iterations will on average be about twice this value
       * due to randomization. Note: Spinning is disabled when NCPU==1.
       */
      private static final int SPINS = 1 << 10;
  
      /**
       * Value representing null arguments/returns from public
       * methods. Needed because the API originally didn't disallow null
       * arguments, which it should have.
       */
      private static final Object NULL_ITEM = new Object();
  
      /**
       * Sentinel value returned by internal exchange methods upon
       * timeout, to avoid need for separate timed versions of these
       * methods.
       */
      private static final Object TIMED_OUT = new Object();
  
      /**
       * Nodes hold partially exchanged data, plus other per-thread
       * bookkeeping. Padded via @Contended to reduce memory contention.
       */
      // Android-removed: @Contended, this hint is not used by the Android runtime.
      //@jdk.internal.vm.annotation.Contended
      static final class Node {
          int index;              // Arena index
          int bound;              // Last recorded value of Exchanger.bound
          int collides;           // Number of CAS failures at current bound
          int hash;               // Pseudo-random for spins
          Object item;            // This thread's current item
          volatile Object match;  // Item provided by releasing thread
          volatile Thread parked; // Set to this thread when parked, else null
      }
  
      /** The corresponding thread local class */
      static final class Participant extends ThreadLocal<Node> {
          public Node initialValue() { return new Node(); }
      }
  
      /**
       * Per-thread state.
       */
      private final Participant participant;
  
      /**
       * Elimination array; null until enabled (within slotExchange).
       * Element accesses use emulation of volatile gets and CAS.
       */
      private volatile Node[] arena;
  
      /**
       * Slot used until contention detected.
       */
      private volatile Node slot;
  
      /**
       * The index of the largest valid arena position, OR'ed with SEQ
       * number in high bits, incremented on each update.  The initial
       * update from 0 to SEQ is used to ensure that the arena array is
       * constructed only once.
       */
      private volatile int bound;
  
      /**
       * Exchange function when arenas enabled. See above for explanation.
       *
       * @param item the (non-null) item to exchange
       * @param timed true if the wait is timed
       * @param ns if timed, the maximum wait time, else 0L
       * @return the other thread's item; or null if interrupted; or
       * TIMED_OUT if timed and timed out
       */
      private final Object arenaExchange(Object item, boolean timed, long ns) {
          Node[] a = arena;
          Node p = participant.get();
          for (int i = p.index;;) {                      // access slot at i
              int b, m, c; long j;                       // j is raw array offset
              Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
              if (q != null && U.compareAndSwapObject(a, j, q, null)) {
                  Object v = q.item;                     // release
                  q.match = item;
                  Thread w = q.parked;
                  if (w != null)
                      U.unpark(w);
                  return v;
              }
              else if (i <= (m = (b = bound) & MMASK) && q == null) {
                  p.item = item;                         // offer
                  if (U.compareAndSwapObject(a, j, null, p)) {
                      long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;
                      Thread t = Thread.currentThread(); // wait
                      for (int h = p.hash, spins = SPINS;;) {
                          Object v = p.match;
                          if (v != null) {
                              U.putOrderedObject(p, MATCH, null);
                              p.item = null;             // clear for next use
                              p.hash = h;
                              return v;
                          }
                          else if (spins > 0) {
                              h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
                              if (h == 0)                // initialize hash
                                  h = SPINS | (int)t.getId();
                              else if (h < 0 &&          // approx 50% true
                                       (--spins & ((SPINS >>> 1) - 1)) == 0)
                                  Thread.yield();        // two yields per wait
                          }
                          else if (U.getObjectVolatile(a, j) != p)
                              spins = SPINS;       // releaser hasn't set match yet
                          else if (!t.isInterrupted() && m == 0 &&
                                   (!timed ||
                                    (ns = end - System.nanoTime()) > 0L)) {
                              U.putObject(t, BLOCKER, this); // emulate LockSupport
                              p.parked = t;              // minimize window
                              if (U.getObjectVolatile(a, j) == p)
                                  U.park(false, ns);
                              p.parked = null;
                              U.putObject(t, BLOCKER, null);
                          }
                          else if (U.getObjectVolatile(a, j) == p &&
                                   U.compareAndSwapObject(a, j, p, null)) {
                              if (m != 0)                // try to shrink
                                  U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);
                              p.item = null;
                              p.hash = h;
                              i = p.index >>>= 1;        // descend
                              if (Thread.interrupted())
                                  return null;
                              if (timed && m == 0 && ns <= 0L)
                                  return TIMED_OUT;
                              break;                     // expired; restart
                          }
                      }
                  }
                  else
                      p.item = null;                     // clear offer
              }
              else {
                  if (p.bound != b) {                    // stale; reset
                      p.bound = b;
                      p.collides = 0;
                      i = (i != m || m == 0) ? m : m - 1;
                  }
                  else if ((c = p.collides) < m || m == FULL ||
                           !U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {
                      p.collides = c + 1;
                      i = (i == 0) ? m : i - 1;          // cyclically traverse
                  }
                  else
                      i = m + 1;                         // grow
                  p.index = i;
              }
          }
      }
  
      /**
       * Exchange function used until arenas enabled. See above for explanation.
       *
       * @param item the item to exchange
       * @param timed true if the wait is timed
       * @param ns if timed, the maximum wait time, else 0L
       * @return the other thread's item; or null if either the arena
       * was enabled or the thread was interrupted before completion; or
       * TIMED_OUT if timed and timed out
       */
      private final Object slotExchange(Object item, boolean timed, long ns) {
          Node p = participant.get();
          Thread t = Thread.currentThread();
          if (t.isInterrupted()) // preserve interrupt status so caller can recheck
              return null;
  
          for (Node q;;) {
              if ((q = slot) != null) {
                  if (U.compareAndSwapObject(this, SLOT, q, null)) {
                      Object v = q.item;
                      q.match = item;
                      Thread w = q.parked;
                      if (w != null)
                          U.unpark(w);
                      return v;
                  }
                  // create arena on contention, but continue until slot null
                  if (NCPU > 1 && bound == 0 &&
                      U.compareAndSwapInt(this, BOUND, 0, SEQ))
                      arena = new Node[(FULL + 2) << ASHIFT];
              }
              else if (arena != null)
                  return null; // caller must reroute to arenaExchange
              else {
                  p.item = item;
                  if (U.compareAndSwapObject(this, SLOT, null, p))
                      break;
                  p.item = null;
              }
          }
  
          // await release
          int h = p.hash;
          long end = timed ? System.nanoTime() + ns : 0L;
          int spins = (NCPU > 1) ? SPINS : 1;
          Object v;
          while ((v = p.match) == null) {
              if (spins > 0) {
                  h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
                  if (h == 0)
                      h = SPINS | (int)t.getId();
                  else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
                      Thread.yield();
              }
              else if (slot != p)
                  spins = SPINS;
              else if (!t.isInterrupted() && arena == null &&
                       (!timed || (ns = end - System.nanoTime()) > 0L)) {
                  U.putObject(t, BLOCKER, this);
                  p.parked = t;
                  if (slot == p)
                      U.park(false, ns);
                  p.parked = null;
                  U.putObject(t, BLOCKER, null);
              }
              else if (U.compareAndSwapObject(this, SLOT, p, null)) {
                  v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
                  break;
              }
          }
          U.putOrderedObject(p, MATCH, null);
          p.item = null;
          p.hash = h;
          return v;
      }
  
      /**
       * Creates a new Exchanger.
       */
      public Exchanger() {
          participant = new Participant();
      }
  
      /**
       * 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
       */
      @SuppressWarnings("unchecked")
      public V exchange(V x) throws InterruptedException {
          Object v;
          Object item = (x == null) ? NULL_ITEM : x; // translate null args
          if ((arena != null ||
               (v = slotExchange(item, false, 0L)) == null) &&
              ((Thread.interrupted() || // disambiguates null return
                (v = arenaExchange(item, false, 0L)) == null)))
              throw new InterruptedException();
          return (v == NULL_ITEM) ? null : (V)v;
      }
  
      /**
       * 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 {@code timeout} 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
       */
      @SuppressWarnings("unchecked")
      public V exchange(V x, long timeout, TimeUnit unit)
          throws InterruptedException, TimeoutException {
          Object v;
          Object item = (x == null) ? NULL_ITEM : x;
          long ns = unit.toNanos(timeout);
          if ((arena != null ||
               (v = slotExchange(item, true, ns)) == null) &&
              ((Thread.interrupted() ||
                (v = arenaExchange(item, true, ns)) == null)))
              throw new InterruptedException();
          if (v == TIMED_OUT)
              throw new TimeoutException();
          return (v == NULL_ITEM) ? null : (V)v;
      }
  
      // Unsafe mechanics
      private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
      private static final long BOUND;
      private static final long SLOT;
      private static final long MATCH;
      private static final long BLOCKER;
      private static final int ABASE;
      static {
          try {
              BOUND = U.objectFieldOffset
                  (Exchanger.class.getDeclaredField("bound"));
              SLOT = U.objectFieldOffset
                  (Exchanger.class.getDeclaredField("slot"));
  
              MATCH = U.objectFieldOffset
                  (Node.class.getDeclaredField("match"));
  
              BLOCKER = U.objectFieldOffset
                  (Thread.class.getDeclaredField("parkBlocker"));
  
              int scale = U.arrayIndexScale(Node[].class);
              if ((scale & (scale - 1)) != 0 || scale > (1 << ASHIFT))
                  throw new Error("Unsupported array scale");
              // ABASE absorbs padding in front of element 0
              ABASE = U.arrayBaseOffset(Node[].class) + (1 << ASHIFT);
          } catch (ReflectiveOperationException e) {
              throw new Error(e);
          }
      }
  
  }
  