/*
 * Written by Dirk Gorissen and Dawid Kurzyniec and released to the public
 * domain, as explained at http://creativecommons.org/licenses/publicdomain
 */

package edu.emory.mathcs.util.remote.locks;

import java.rmi.*;
import edu.emory.mathcs.backport.java.util.concurrent.*;
import edu.emory.mathcs.backport.java.util.concurrent.helpers.*;


/**
 * This class provides a generic implementation of a distributed, non-reentrant
 * mutual exclusion lock, based on Eisenberg and MgGuire's algorithm.
 * <p>
 * There are N processes (N may vary) who share some form of memory.  This
 * memory may be phisical memory, a remote database,... just some resource
 * that enables the N processes to maintain some global state.
 * Each of the N processes has in its code a Critical Section (CS) and at any
 * point in time only one process can be in its CS.
 * To enforce this we need a mutual exclusion mechanism which enforces
 * mutual exclusion for N processes with shared memory.
 * Many algorithms that do this exist, here the algorithm used is
 * Eisenberg and McGuire's Algorithm.
 * Source: <a href="http://www.cs.wvu.edu/~jdm/classes/cs356/notes/mutex/Eisenberg.html">http://www.cs.wvu.edu/~jdm/classes/cs356/notes/mutex/Eisenberg.html</a>
 * <p>
 * To maintain global state the algorithm uses a list of flags where each flag[i] holds
 * the state of of process i.  The state is one of IDLE, WAITING or ACTIVE.
 * In addition a global turn variable is kept which is set to i if process i is in
 * its CS.
 * <p>
 * Since this class is a generic implementation of the algorithm the exact communication
 * mechanism is left up to the user.  Processes may coordinate through a Jini LUS,
 * a JavaSpace, a SQL Database, an in memory hashmap, etc.  All that is required is that
 * the user implements the {@link Backend} interface and passes an instance of the implementation
 * to this class.
 * <p>
 *
 * @author Dirk Gorissen <dgorissen@gmx.net>
 */
public class RemoteEisMcGLock implements RemoteLock {

    public static interface Backend {

        /**
         * Returns the ID of self.
         * @return Comparable the ID of self
         */
        Comparable getMyID();

        /**
         * Gets the participant ID of the turn holder, or null if turn is
         * unclaimed.
         */
        Comparable getTurnID() throws RemoteException;

        /**
         * Sets turn holder to a given participant ID
         */
        void setTurnID(Comparable id) throws RemoteException;

        void setMyState(Participant.State state) throws RemoteException;

        void resetCursor(Comparable id);

        /**
         * Fetch the info about the next possibly non-idle participant.
         *
         * @throws RemoteException if a communication error occurs
         */
        Participant fetchNext() throws RemoteException;

        void waitForEvent(long nanos) throws InterruptedException;
    }

    public static class Participant implements Comparable {
        final Comparable id;
        final State state;
        public Participant(Comparable id, State state) {
            this.id = id;
            this.state = state;
        }
        public Comparable getID() { return id; }
        public State getState() { return state; }

        public int hashCode() { return id.hashCode(); }
        public boolean equals(Object other) {
            if (other == this) return true;
            if (! (other instanceof Participant)) return false;
            Participant that = (Participant) other;
            return this.id.equals(that.id);
        }

        public int compareTo(Object other) {
            if (other == this) return 0;
            Participant that = (Participant)other;
            return this.id.compareTo(that.id);
        }

        public final static class State {

            public final static State IDLE    = new State(0, "IDLE");
            public final static State WAITING = new State(1, "WAITING");
            public final static State ACTIVE  = new State(2, "ACTIVE");

            public final int id;
            public final String name;
            private State(int id, String name) { this.id = id; this.name = name; }

            public final static State forID(int id) {
                switch (id) {
                    case 0: return IDLE;
                    case 1: return WAITING;
                    case 2: return ACTIVE;
                    default: throw new IllegalArgumentException();
                }
            }

            public final static State forName(String name) {
                if ("IDLE".equalsIgnoreCase(name)) return IDLE;
                if ("WAITING".equalsIgnoreCase(name)) return WAITING;
                if ("ACTIVE".equalsIgnoreCase(name)) return ACTIVE;
                throw new IllegalArgumentException();
            }
        }
    }

    final Backend backend;

    /**
     * Creates a new non-reentrant remote lock, synchronizing on a specified
     * backend.
     * @param backend Backend to synchronize on
     */
    public RemoteEisMcGLock(Backend backend) {
        if (backend == null) throw new NullPointerException();
        this.backend = backend;
    }

    /**
    * {@inheritDoc}
    */
    public void lock() throws RemoteException {
        boolean wasInterrupted = false;
        while (true) {
            try {
                lockInterruptibly();
                if (wasInterrupted) {
                    Thread.currentThread().interrupt();
                }
                return;
            }
            catch (InterruptedException e) {
                wasInterrupted = true;
            }
        }
    }

    /**
    * {@inheritDoc}
    */
    public void lockInterruptibly() throws InterruptedException, RemoteException {
        tryLockImpl(false, 0);
    }

    /**
    * {@inheritDoc}
    */
    public boolean tryLock() throws RemoteException {
        try {
            return tryLockImpl(true, 0);
        }
        catch (InterruptedException e) {
            // cannot happen
            throw new RuntimeException(e);
        }
    }

    /**
    * {@inheritDoc}
    */
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException, RemoteException {
        return tryLockImpl(true, unit.toNanos(timeout));
    }

    public RemoteCondition newCondition() throws RemoteException {
        throw new UnsupportedOperationException("Not implemented");
    }

    boolean tryLockImpl(boolean timed, long nanos) throws InterruptedException, RemoteException {
        Comparable myID = backend.getMyID();
        Comparable turn;
        long deadline = timed ? Utils.nanoTime() + nanos : 0;

main:   while (true) {

            // Announce that we need the resource
            backend.setMyState(Participant.State.WAITING);

            // Retrieve all flags
            turn = backend.getTurnID();

            Participant p;

waitActive: while (true) {
                backend.resetCursor(turn);
                while (true) {
                    p = backend.fetchNext();
                    if (p == null) {
                        throw new SharingProtocolException(
                            "The WAITING flag that we've set has disappeared");
                    }
                    if (p.getID().equals(myID)) {
                        break waitActive;
                    }
                    if (p.getState() != Participant.State.IDLE) {
                        if (nanos <= 0) {
                            backend.setMyState(Participant.State.IDLE);
                            return false;
                        }
                        backend.waitForEvent(nanos);
                        nanos = timed ? deadline - Utils.nanoTime() : 0;
                        continue waitActive;
                    }
                }
            }

            // Ok, now hopefully everybody between me and the process who has
            // turn is Idle; Now tentatively claim the resource
            backend.setMyState(Participant.State.ACTIVE);

            //Now find the first active process besides ourselves, if any

            turn = backend.getTurnID();
            backend.resetCursor(null);

            while (true) {
                p = backend.fetchNext();
                if (p == null) { // iterated through the whole list
                    break main;
                }
                if (p.getID().equals(myID)) {
                    continue;
                }
                if (p.getState() == Participant.State.ACTIVE ||
                        (p.getID().equals(turn) && p.getState() != Participant.State.IDLE)) {
                    if (nanos <= 0) {
                        backend.setMyState(Participant.State.IDLE);
                        return false;
                    }
                    backend.waitForEvent(nanos);
                    nanos = timed ? deadline - Utils.nanoTime() : 0;
                    continue main;
                }
            }
        }

        //Claim the turn and proceed
        backend.setTurnID(myID);
        turn = myID;
        return true;
    }

    /**
     * {@inheritDoc}
     */
    public void unlock() throws RemoteException {
        Comparable myID = backend.getMyID();
        Comparable turn = backend.getTurnID();

        if (!myID.equals(turn)) {
            throw new SharingProtocolException("We are no longer holding a turn");
        }

        backend.resetCursor(myID);
        Participant p;
        //Find a process behind us which is not idle
        do {
            p = backend.fetchNext();
            if (p == null) {
                backend.setTurnID(null);
                return;
            }
        }
        while (p.getState() == Participant.State.IDLE || p.equals(myID));

        backend.setTurnID(p.getID());
        backend.setMyState(Participant.State.IDLE);
    }

    public int hashCode() {
        return backend.hashCode();
    }

    public boolean equals(Object other) {
        if (!(other instanceof RemoteEisMcGLock)) return false;
        RemoteEisMcGLock that = (RemoteEisMcGLock)other;
        return this.backend.equals(that.backend);
    }
}
