src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.cassandra.utils.concurrent;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;

import org.apache.cassandra.utils.Intercept;
import org.apache.cassandra.utils.Shared;
import org.apache.cassandra.utils.concurrent.Awaitable.AbstractAwaitable;

import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES;
import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;

/**
 * <p>A relatively easy to use utility for general purpose thread signalling.</p>
 * <p>Usage on a thread awaiting a state change using a WaitQueue q is:</p>
 * <pre>
 * {@code
 *      while (!conditionMet())
 *          Signal s = q.register();
 *              if (!conditionMet())    // or, perhaps more correctly, !conditionChanged()
 *                  s.await();
 *              else
 *                  s.cancel();
 * }
 * </pre>
 * A signalling thread, AFTER changing the state, then calls q.signal() to wake up one, or q.signalAll()
 * to wake up all, waiting threads.
 * <p>To understand intuitively how this class works, the idea is simply that a thread, once it considers itself
 * incapable of making progress, registers to be awoken once that changes. Since this could have changed between
 * checking and registering (in which case the thread that made this change would have been unable to signal it),
 * it checks the condition again, sleeping only if it hasn't changed/still is not met.</p>
 * <p>This thread synchronisation scheme has some advantages over Condition objects and Object.wait/notify in that no monitor
 * acquisition is necessary and, in fact, besides the actual waiting on a signal, all operations are non-blocking.
 * As a result consumers can never block producers, nor each other, or vice versa, from making progress.
 * Threads that are signalled are also put into a RUNNABLE state almost simultaneously, so they can all immediately make
 * progress without having to serially acquire the monitor/lock, reducing scheduler delay incurred.</p>
 *
 * <p>A few notes on utilisation:</p>
 * <p>1. A thread will only exit await() when it has been signalled, but this does not guarantee the condition has not
 * been altered since it was signalled, and depending on your design it is likely the outer condition will need to be
 * checked in a loop, though this is not always the case.</p>
 * <p>2. Each signal is single use, so must be re-registered after each await(). This is true even if it times out.</p>
 * <p>3. If you choose not to wait on the signal (because the condition has been met before you waited on it)
 * you must cancel() the signal if the signalling thread uses signal() to awake waiters; otherwise signals will be
 * lost. If signalAll() is used but infrequent, and register() is frequent, cancel() should still be used to prevent the
 * queue growing unboundedly. Similarly, if you provide a TimerContext, cancel should be used to ensure it is not erroneously
 * counted towards wait time.</p>
 * <p>4. Care must be taken when selecting conditionMet() to ensure we are waiting on the condition that actually
 * indicates progress is possible. In some complex cases it may be tempting to wait on a condition that is only indicative
 * of local progress, not progress on the task we are aiming to complete, and a race may leave us waiting for a condition
 * to be met that we no longer need.
 * <p>5. This scheme is not fair</p>
 * <p>6. Only the thread that calls register() may call await()</p>
 *
 * TODO: this class should not be backed by CLQ (should use an intrusive linked-list with lower overhead)
 */
@Shared(scope = SIMULATION, inner = INTERFACES)
public interface WaitQueue
{
    /**
     * A Signal is a one-time-use mechanism for a thread to wait for notification that some condition
     * state has transitioned that it may be interested in (and hence should check if it is).
     * It is potentially transient, i.e. the state can change in the meantime, it only indicates
     * that it should be checked, not necessarily anything about what the expected state should be.
     *
     * Signal implementations should never wake up spuriously, they are always woken up by a
     * signal() or signalAll().
     *
     * This abstract definition of Signal does not need to be tied to a WaitQueue.
     * Whilst RegisteredSignal is the main building block of Signals, this abstract
     * definition allows us to compose Signals in useful ways. The Signal is 'owned' by the
     * thread that registered itself with WaitQueue(s) to obtain the underlying RegisteredSignal(s);
     * only the owning thread should use a Signal.
     */
    public static interface Signal extends Condition
    {
        /**
         * @return true if cancelled; once cancelled, must be discarded by the owning thread.
         */
        public boolean isCancelled();

        /**
         * @return isSignalled() || isCancelled(). Once true, the state is fixed and the Signal should be discarded
         * by the owning thread.
         */
        public boolean isSet();

        /**
         * atomically: cancels the Signal if !isSet(), or returns true if isSignalled()
         *
         * @return true if isSignalled()
         */
        public boolean checkAndClear();

        /**
         * Should only be called by the owning thread. Indicates the signal can be retired,
         * and if signalled propagates the signal to another waiting thread
         */
        public abstract void cancel();
    }

    /**
     * The calling thread MUST be the thread that uses the signal
     */
    public Signal register();

    /**
     * The calling thread MUST be the thread that uses the signal.
     * If the Signal is waited on, context.stop() will be called when the wait times out, the Signal is signalled,
     * or the waiting thread is interrupted.
     */
    public <V> Signal register(V supplyOnDone, Consumer<V> receiveOnDone);

    /**
     * Signal one waiting thread
     */
    public boolean signal();

    /**
     * Signal all waiting threads
     */
    public void signalAll();

    /** getWaiting() > 0 */
    public boolean hasWaiters();

    /** Return how many threads are waiting */
    public int getWaiting();

    /**
     * Factory method used to capture and redirect instantiations for simulation
     */
    @Intercept
    public static WaitQueue newWaitQueue()
    {
        return new Standard();
    }

    class Standard implements WaitQueue
    {
        private static final int CANCELLED = -1;
        private static final int SIGNALLED = 1;
        private static final int NOT_SET = 0;

        private static final AtomicIntegerFieldUpdater<RegisteredSignal> signalledUpdater = AtomicIntegerFieldUpdater.newUpdater(RegisteredSignal.class, "state");

        // the waiting signals
        private final ConcurrentLinkedQueue<RegisteredSignal> queue = new ConcurrentLinkedQueue<>();

        protected Standard() {}

        /**
         * The calling thread MUST be the thread that uses the signal
         * @return                                x
         */
        public Signal register()
        {
            RegisteredSignal signal = new RegisteredSignal();
            queue.add(signal);
            return signal;
        }

        /**
         * The calling thread MUST be the thread that uses the signal.
         * If the Signal is waited on, context.stop() will be called when the wait times out, the Signal is signalled,
         * or the waiting thread is interrupted.
         */
        public <V> Signal register(V supplyOnDone, Consumer<V> receiveOnDone)
        {
            RegisteredSignal signal = new SignalWithListener<>(supplyOnDone, receiveOnDone);
            queue.add(signal);
            return signal;
        }

        /**
         * Signal one waiting thread
         */
        public boolean signal()
        {
            while (true)
            {
                RegisteredSignal s = queue.poll();
                if (s == null || s.doSignal() != null)
                    return s != null;
            }
        }

        /**
         * Signal all waiting threads
         */
        public void signalAll()
        {
            if (!hasWaiters())
                return;

            // to avoid a race where the condition is not met and the woken thread managed to wait on the queue before
            // we finish signalling it all, we pick a random thread we have woken-up and hold onto it, so that if we encounter
            // it again we know we're looping. We reselect a random thread periodically, progressively less often.
            // the "correct" solution to this problem is to use a queue that permits snapshot iteration, but this solution is sufficient
            // TODO: this is only necessary because we use CLQ - which is only for historical any-NIH reasons
            int i = 0, s = 5;
            Thread randomThread = null;
            Iterator<RegisteredSignal> iter = queue.iterator();
            while (iter.hasNext())
            {
                RegisteredSignal signal = iter.next();
                Thread signalled = signal.doSignal();

                if (signalled != null)
                {
                    if (signalled == randomThread)
                        break;

                    if (++i == s)
                    {
                        randomThread = signalled;
                        s <<= 1;
                    }
                }

                iter.remove();
            }
        }

        private void cleanUpCancelled()
        {
            // TODO: attempt to remove the cancelled from the beginning only (need atomic cas of head)
            queue.removeIf(RegisteredSignal::isCancelled);
        }

        public boolean hasWaiters()
        {
            return !queue.isEmpty();
        }

        /**
         * @return how many threads are waiting
         */
        public int getWaiting()
        {
            if (!hasWaiters())
                return 0;
            Iterator<RegisteredSignal> iter = queue.iterator();
            int count = 0;
            while (iter.hasNext())
            {
                Signal next = iter.next();
                if (!next.isCancelled())
                    count++;
            }
            return count;
        }

        /**
         * An abstract signal implementation
         *
         * TODO: use intrusive linked list
         */
        public static abstract class AbstractSignal extends AbstractAwaitable implements Signal
        {
            public Signal await() throws InterruptedException
            {
                while (!isSignalled())
                {
                    checkInterrupted();
                    LockSupport.park();
                }
                checkAndClear();
                return this;
            }

            public boolean awaitUntil(long nanoTimeDeadline) throws InterruptedException
            {
                long now;
                while (nanoTimeDeadline > (now = nanoTime()) && !isSignalled())
                {
                    checkInterrupted();
                    long delta = nanoTimeDeadline - now;
                    LockSupport.parkNanos(delta);
                }
                return checkAndClear();
            }

            private void checkInterrupted() throws InterruptedException
            {
                if (Thread.interrupted())
                {
                    cancel();
                    throw new InterruptedException();
                }
            }
        }

        /**
         * A signal registered with this WaitQueue
         */
        private class RegisteredSignal extends AbstractSignal
        {
            private volatile Thread thread = Thread.currentThread();
            volatile int state;

            public boolean isSignalled()
            {
                return state == SIGNALLED;
            }

            public boolean isCancelled()
            {
                return state == CANCELLED;
            }

            public boolean isSet()
            {
                return state != NOT_SET;
            }

            private Thread doSignal()
            {
                if (!isSet() && signalledUpdater.compareAndSet(this, NOT_SET, SIGNALLED))
                {
                    Thread thread = this.thread;
                    LockSupport.unpark(thread);
                    this.thread = null;
                    return thread;
                }
                return null;
            }

            public void signal()
            {
                doSignal();
            }

            public boolean checkAndClear()
            {
                if (!isSet() && signalledUpdater.compareAndSet(this, NOT_SET, CANCELLED))
                {
                    thread = null;
                    cleanUpCancelled();
                    return false;
                }
                // must now be signalled assuming correct API usage
                return true;
            }

            /**
             * Should only be called by the registered thread. Indicates the signal can be retired,
             * and if signalled propagates the signal to another waiting thread
             */
            public void cancel()
            {
                if (isCancelled())
                    return;
                if (!signalledUpdater.compareAndSet(this, NOT_SET, CANCELLED))
                {
                    // must already be signalled - switch to cancelled and
                    state = CANCELLED;
                    // propagate the signal
                    WaitQueue.Standard.this.signal();
                }
                thread = null;
                cleanUpCancelled();
            }
        }

        /**
         * A RegisteredSignal that stores a TimerContext, and stops the timer when either cancelled or
         * finished waiting. i.e. if the timer is started when the signal is registered it tracks the
         * time in between registering and invalidating the signal.
         */
        private final class SignalWithListener<V> extends RegisteredSignal
        {
            private final V supplyOnDone;
            private final Consumer<V> receiveOnDone;

            private SignalWithListener(V supplyOnDone, Consumer<V> receiveOnDone)
            {
                this.receiveOnDone = receiveOnDone;
                this.supplyOnDone = supplyOnDone;
            }


            @Override
            public boolean checkAndClear()
            {
                receiveOnDone.accept(supplyOnDone);
                return super.checkAndClear();
            }

            @Override
            public void cancel()
            {
                if (!isCancelled())
                {
                    receiveOnDone.accept(supplyOnDone);
                    super.cancel();
                }
            }
        }
    }

    /**
     * Loops waiting on the supplied condition and WaitQueue and will not return until the condition is true
     */
    public static void waitOnCondition(BooleanSupplier condition, WaitQueue queue) throws InterruptedException
    {
        while (!condition.getAsBoolean())
        {
            Signal s = queue.register();
            if (!condition.getAsBoolean()) s.await();
            else s.cancel();
        }
    }
}