/*
 * 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;

import jdk.internal.invoke.MhUtil;

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
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 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 holding references to
     * transfered items 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 must ensure
     * that object transfer fields are reset between uses. Given this,
     * Participant nodes can be defined as static ThreadLocals. As
     * seen for example in class Striped64, using indices established
     * in one instance across others usually improves overall
     * performance.  Nodes also include a participant-local random
     * number generator.
     *
     * Spreading out contention requires that the memory locations
     * used by the arena slots don't share a cache line -- otherwise,
     * the arena would have almost no benefit. We arrange this by
     * adding another level of indirection: The arena elements point
     * to "Slots", each of which is padded using @Contended. We only
     * create a single Slot on intialization, adding more when
     * needed. The per-thread Participant Nodes may also be subject to
     * false-sharing contention, but tend to be more scattered in
     * memory, so are unpadded, with some occasional performance impact.
     *
     * 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. And shrink it via "spinouts" in which
     * threads give up waiting at a slot.  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.
     *
     * Arena size (the value of field "bound") is controlled by random
     * sampling. On each miss (collision or spinout), a thread chooses
     * a new random index within the arena.  Upon the third collision
     * with the same current bound, it tries to grow the arena. And
     * upon the second spinout, it tries to shrink. The asymmetry in
     * part reflects relative costs, and reduces flailing. Because
     * they cannot be changed without also changing the sampling
     * strategy, these rules are directly incorporated into uses of
     * the xchg "misses" variable.  The bound field is tagged with
     * sequence numbers to reduce stale decisions. Uniform random
     * indices are generated using XorShift with enough bits so that
     * bias (See Knuth TAoCP vol 2) is negligible for moduli used here
     * (at most 256) without requiring rejection tests. Using
     * nonuniform randoms with greater weight to higher indices is
     * also possible but does not seem worthwhile in practice.
     *
     * These mechanics rely on a reasonable choice of constant SPINS.
     * The time cost of SPINS * Thread.onSpinWait() should be at least
     * the expected cost of a park/unpark context switch, and larger
     * than that of two failed CASes, but still small enough to avoid
     * excessive delays during arena shrinkage.  We also deal with the
     * possibility that when an offering thread waits for a release,
     * spin-waiting would be useless because the releasing thread is
     * descheduled. On multiprocessors, we cannot know this in
     * general. But when Virtual Threads are used, method
     * ForkJoinWorkerThread.hasKnownQueuedWork serves as a guide to
     * whether to spin or immediately block, allowing a context switch
     * that may enable a releaser.  Note also that when many threads
     * are being run on few cores, enountering enough collisions to
     * trigger arena growth is rare, and soon followed by shrinkage,
     * so this doesn't require special handling.
     *
     * The basic exchange mechanics rely on checks that Node item
     * fields are not null, which doesn't work when offered items are
     * null. We trap this case by translating nulls to the
     * (un-Exchangeable) value of the static Participant
     * reference.
     *
     * Essentially all of the implementation is in method xchg.  As is
     * too common in this sort of code, most of the logic relies on
     * reads of fields that are maintained as local variables so can't
     * be nicely factored. It is structured as a main loop with a
     * leading volatile read (of field bound), that causes others to
     * be freshly read even though declared in plain mode.  We don't
     * use compareAndExchange that would otherwise save some re-reads
     * because of the need to recheck indices and bounds on failures.
     *
     * Support for optional timeouts in a single method adds further
     * complexity. Note that for the sake of arena bounds control,
     * time bounds must be ignored during spinouts, which may delay
     * TimeoutExceptions (but no more so than would excessive context
     * switching that could occur otherwise).  Responses to
     * interruption are handled similarly, postponing commitment to
     * throw InterruptedException until successfully cancelled.
     *
     * Design differences from previous releases include:
     * * Accommodation of VirtualThreads.
     * * Use of Slots vs spaced indices for the arena and static
     *   ThreadLocals, avoiding separate arena vs non-arena modes.
     * * Use of random sampling for grow/shrink decisions, with typically
     *   faster and more stable adaptation (as was mentioned as a
     *   possible improvement in previous version).
     */

    /**
     * The maximum supported arena index. The maximum allocatable
     * arena size is MMASK + 1. Must be a power of two minus one. 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 bound for spins while waiting for a match before either
     * blocking or possibly shrinking arena.
     */
    private static final int SPINS = 1 << 10;

    /**
     * Padded arena cells to avoid false-sharing memory contention
     */
    @jdk.internal.vm.annotation.Contended
    static final class Slot {
        Node entry;
    }

    /**
     * Nodes hold partially exchanged data, plus other per-thread
     * bookkeeping.
     */
    static final class Node {
        long seed;              // Random seed
        int index;              // Arena index
        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
        Node() {
            index = -1;         // initialize on first use
            seed = Thread.currentThread().threadId();
        }
    }

    /** The corresponding thread local class */
    static final class Participant extends ThreadLocal<Node> {
        public Node initialValue() { return new Node(); }
    }

    /**
     * The participant thread-locals. Because it is impossible to
     * exchange, we also use this reference for dealing with null user
     * arguments that are translated in and out of this value
     * surrounding use.
     */
    private static final Participant participant = new Participant();

    /**
     * Elimination array; element accesses use emulation of volatile
     * gets and CAS.
     */
    private final Slot[] arena;

    /**
     * Number of cores, for sizing and spin control. Computed only
     * upon construction.
     */
    private final int ncpu;

    /**
     * The index of the largest valid arena position.
     */
    private volatile int bound;

    /**
     * Exchange function. See above for explanation.
     *
     * @param x the item to exchange
     * @param deadline if zero, untimed, else timeout deadline
     * @return the other thread's item
     * @throws InterruptedException if interrupted while waiting
     * @throws TimeoutException if deadline nonzero and timed out
     */
    private final V xchg(V x, long deadline)
        throws InterruptedException, TimeoutException {
        Slot[] a = arena;
        int alen = a.length;
        Participant ps = participant;
        Object item = (x == null) ? ps : x;      // translate nulls
        Node p = ps.get();
        int i = p.index;                         // if < 0, move
        int misses = 0;                          // ++ on collide, -- on spinout
        Object offered = null;                   // for cleanup
        Object v = null;
        outer: for (;;) {
            int b, m; Slot s; Node q;
            if ((m = (b = bound) & MMASK) == 0)  // volatile read
                i = 0;
            if (i < 0 || i > m || i >= alen || (s = a[i]) == null) {
                long r = p.seed;                 // randomly move
                r ^= r << 13; r ^= r >>> 7; r ^= r << 17; // xorShift
                i = p.index = (int)((p.seed = r) % (m + 1));
            }
            else if ((q = s.entry) != null) {    // try release
                if (ENTRY.compareAndSet(s, q, null)) {
                    Thread w;
                    v = q.item;
                    q.match = item;
                    if (i == 0 && (w = q.parked) != null)
                        LockSupport.unpark(w);
                    break;
                }
                else {                           // collision
                    int nb;
                    i = -1;                      // move index
                    if (b != bound)              // stale
                        misses = 0;
                    else if (misses <= 2)        // continue sampling
                        ++misses;
                    else if ((nb = (b + 1) & MMASK) < alen) {
                        misses = 0;              // try to grow
                        if (BOUND.compareAndSet(this, b, b + 1 + SEQ) &&
                            a[i = p.index = nb] == null)
                            AA.compareAndSet(a, nb, null, new Slot());
                    }
                }
            }
            else {                               // try offer
                if (offered == null)
                    offered = p.item = item;
                if (ENTRY.compareAndSet(s, null, p)) {
                    boolean tryCancel;           // true if interrupted
                    Thread t = Thread.currentThread();
                    if (!(tryCancel = t.isInterrupted()) && ncpu > 1 &&
                        (i != 0 ||               // check for busy VTs
                         (!ForkJoinWorkerThread.hasKnownQueuedWork()))) {
                        for (int j = SPINS; j > 0; --j) {
                            if ((v = p.match) != null) {
                                MATCH.set(p, null);
                                break outer;     // spin wait
                            }
                            Thread.onSpinWait();
                        }
                    }
                    for (long ns = 1L;;) {       // block or cancel offer
                        if ((v = p.match) != null) {
                            MATCH.set(p, null);
                            break outer;
                        }
                        if (i == 0 && !tryCancel &&
                            (deadline == 0L ||
                             ((ns = deadline - System.nanoTime()) > 0L))) {
                            p.parked = t;        // emable unpark and recheck
                            if (p.match == null) {
                                if (deadline == 0L)
                                    LockSupport.park(this);
                                else
                                    LockSupport.parkNanos(this, ns);
                                tryCancel = t.isInterrupted();
                            }
                            p.parked = null;
                        }
                        else if (ENTRY.compareAndSet(s, p, null)) { // cancel
                            offered = p.item = null;
                            if (Thread.interrupted())
                                throw new InterruptedException();
                            if (deadline != 0L && ns <= 0L)
                                throw new TimeoutException();
                            i = -1;              // move and restart
                            if (bound != b)
                                misses = 0;      // stale
                            else if (misses >= 0)
                                --misses;        // continue sampling
                            else if ((b & MMASK) != 0) {
                                misses = 0;      // try to shrink
                                BOUND.compareAndSet(this, b, b - 1 + SEQ);
                            }
                            continue outer;
                        }
                    }
                }
            }
        }
        if (offered != null)                     // cleanup
            p.item = null;
        @SuppressWarnings("unchecked") V ret = (v == participant) ? null : (V)v;
        return ret;
    }

    /**
     * Creates a new Exchanger.
     */
    public Exchanger() {
        int h = (ncpu = Runtime.getRuntime().availableProcessors()) >>> 1;
        int size = (h == 0) ? 1 : (h > MMASK) ? MMASK + 1 : h;
        (arena = new Slot[size])[0] = new Slot();
    }

    /**
     * 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 {
        try {
            return xchg(x, 0L);
        } catch (TimeoutException cannotHappen) {
            return null; // not reached
        }
    }

    /**
     * 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
     */
    public V exchange(V x, long timeout, TimeUnit unit)
        throws InterruptedException, TimeoutException {
        long d = unit.toNanos(timeout) + System.nanoTime();
        return xchg(x, (d == 0L) ? 1L : d); // avoid zero deadline
    }

    // VarHandle mechanics
    private static final VarHandle BOUND;
    private static final VarHandle MATCH;
    private static final VarHandle ENTRY;
    private static final VarHandle AA;
    static {
        MethodHandles.Lookup l = MethodHandles.lookup();
        BOUND = MhUtil.findVarHandle(l, "bound", int.class);
        MATCH = MhUtil.findVarHandle(l, Node.class, "match", Object.class);
        ENTRY = MhUtil.findVarHandle(l, Slot.class, "entry", Node.class);
        AA = MethodHandles.arrayElementVarHandle(Slot[].class);
    }

}