/*
 * 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 java.util.*;

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

/**
 * Implements a reentrant lock on top of a possibly non-reentrant remote lock
 * so that local threads are scheduled within the JVM instead of competing
 * for a remote resource. Depends on a correctly implemented semantics of
 * equals for the remote lock (see {@link RemoteLock}).
 *
 * @author Dirk Gorissen <dgorissen@gmx.net>
 * @author Dawid Kurzyniec
 */
public class ReentrantDistributedLock implements RemoteLock {

    // potentially non-reentrant remote lock to which we want to optimize
    // local access
    private final RemoteLock remoteLock;

    private final ReentrantLock localLock;

    /**
     * Create a new ReentrantRemoteLock instance.
     */
    public ReentrantDistributedLock(RemoteLock remoteLock) {
        this.remoteLock = remoteLock;
        this.localLock = getLocalLockFor(remoteLock);
    }

    /**
    * {@inheritDoc}
    */
    public void lock() throws RemoteException {
        localLock.lock();
        if (localLock.getHoldCount() > 1) return;

        pin(localLock);
        try {
            remoteLock.lock();
        }
        catch (Throwable e) {
            unpin(localLock);
            localLock.unlock();
            rethrow(e);
        }
    }

    /**
    * {@inheritDoc}
    */
    public boolean tryLock() throws RemoteException {
        if (!localLock.tryLock()) return false;
        if (localLock.getHoldCount() > 1) return true;

        pin(localLock);
        try {
            boolean success = remoteLock.tryLock();
            if (success) return true;
        }
        catch (Throwable e) {
            unpin(localLock);
            localLock.unlock();
            rethrow(e);
        }

        // remote tryLock failed
        unpin(localLock);
        localLock.unlock();
        return false;
    }

    /**
     * {@inheritDoc}
     */
    public void lockInterruptibly() throws InterruptedException, RemoteException {
        localLock.lockInterruptibly();
        if (localLock.getHoldCount() > 1) return;

        pin(localLock);
        try {
            remoteLock.lockInterruptibly();
        }
        catch (InterruptedException e) {
            unpin(localLock);
            localLock.unlock();
            throw e;
        }
        catch (Throwable e) {
            unpin(localLock);
            localLock.unlock();
            rethrow(e);
        }
    }

    /**
    * {@inheritDoc}
    */
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException, RemoteException {
        long nanos = unit.toNanos(timeout);
        if (!localLock.tryLock()) {
            long deadline = Utils.nanoTime() + nanos;
            if (!localLock.tryLock(timeout, unit)) return false;
            nanos = deadline - Utils.nanoTime();
        }

        if (localLock.getHoldCount() > 1) return true;

        pin(localLock);
        try {
            boolean success = remoteLock.tryLock(nanos, TimeUnit.NANOSECONDS);
            if (success) return true;
        }
        catch (Throwable e) {
            unpin(localLock);
            localLock.unlock();
            rethrow(e);
        }

        // remote tryLock failed
        unpin(localLock);
        localLock.unlock();
        return false;
    }

    /**
     * {@inheritDoc}
     */
    public void unlock() throws RemoteException {
        if (!localLock.isHeldByCurrentThread()) {
            throw new IllegalMonitorStateException("Not owner");
        }
        if (localLock.getHoldCount() > 1) {
            localLock.unlock();
            return;
        }

        try {
            remoteLock.unlock();
        }
        catch (IllegalMonitorStateException e) {
            // we're not owner of the remote lock, or it has been unlocked.
            // propagate the exception (as remote, since we've been playing
            // nice at the local side; don't want unchecked exceptions),
            // but mark as unlocked anyway - denial will not do us any good
            unpin(localLock);
            localLock.unlock();
            throw new RemoteException("Remote inconsistency - the lock "
                                      + "has been already taken away", e);
        }
        catch (Throwable e) {
            // on any other exception, we keep it locked
            rethrow(e);
        }

        // if everything went smooth
        unpin(localLock);
        localLock.unlock();
    }

    /**
     * @return A new Condition instance for this <tt>RemoteLock</tt>
     * instance.
     */
    public RemoteCondition newCondition() throws RemoteException {
        return remoteLock.newCondition();
    }

    private static WeakValueHashMap lockResolver = new WeakValueHashMap();

    // ensures that locked ReentrantLocks do not dissappear (from the volatile
    // resolver map), which would be terrible and lead to deadlocks and other
    // very bad things. (It is OK for unlocked ones to disappear though).
    private static Set lockedLocks = new HashSet();

    /**
     * Creates a new ReentrantLock object (called by the constructor of DistributedLock).
     * To make it thread safe each DistributedLock object contains a ReentrantLock object.
     * To ensure that only threads locking the same resource mutually exclude, each ReentrantLock
     * is associated with a resource name and a ReentrantLock object is shared among threads with
     * the same resource names.  A static instance of VolatileHashMap is used to keep track of the
     * (name,reentrant lock) pairs.
     *
     * @param name Name of the resource with which the created lock should be associated.
     * @return a reentrant lock associated with name
     */
    synchronized static ReentrantLock getLocalLockFor(RemoteLock remote) {
        ReentrantLock local = (ReentrantLock)lockResolver.get(remote);
        if (local == null) {
            local = new ReentrantLock();
            lockResolver.put(remote, local);
        }
        return local;
    }

    /**
     * Adds a reentrant lock to the set of locked reentrant locks, removing
     * the possibility that it becomes subject to garbage collection.
     *
     * @param lock reentrant lock to add
     */
    synchronized static void pin(ReentrantLock lock) {
        lockedLocks.add(lock);
    }

    /**
     * Removes a reentrant lock from the set of locked reentrant locks, opening
     * up the possibility that it becomes subject to garbage collection.
     *
     * @param lock reentrant lock to remove
     */
    synchronized static void unpin(ReentrantLock lock) {
        lockedLocks.remove(lock);
    }

    private static void rethrow(Throwable e) throws RemoteException {
        if (e instanceof RemoteException) throw (RemoteException)e;
        if (e instanceof RuntimeException) throw (RuntimeException)e;
        if (e instanceof Error) throw (Error)e;
        throw new RuntimeException(e);
    }

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

    public boolean equals(Object other) {
        if (other == this) {
            return true;
        }
        if (! (other instanceof ReentrantDistributedLock)) {
            return false;
        }
        ReentrantDistributedLock that = (ReentrantDistributedLock) other;
        return this.remoteLock.equals(that.remoteLock);
    }
}
