/*
    * 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.commons.transaction.locking;
  
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  
  import org.apache.commons.transaction.util.LoggerFacade;
  
  /**
   * A generic implementain of a simple multi level lock.
   * 
   * <p>
   * The idea is to have an ascending number of lock levels ranging from
   * <code>0</code> to <code>maxLockLevel</code> as specified in
   * {@link #GenericLock(Object, int, LoggerFacade)}: the higher the lock level
   * the stronger and more restrictive the lock. To determine which lock may
   * coexist with other locks you have to imagine matching pairs of lock levels.
   * For each pair both parts allow for all lock levels less than or equal to the
   * matching other part. Pairs are composed by the lowest and highest level not
   * yet part of a pair and successively applying this method until no lock level
   * is left. For an even amount of levels each level is part of exactly one pair.
   * For an odd amount the middle level is paired with itself. The highst lock
   * level may coexist with the lowest one (<code>0</code>) which by
   * definition means <code>NO LOCK</code>. This implies that you will have to
   * specify at least one other lock level and thus set <code>maxLockLevel</code>
   * to at least <code>1</code>.
   * </p>
   * 
   * <p>
   * Although this may sound complicated, in practice this is quite simple. Let us
   * imagine you have three lock levels:
   * <ul>
   * <li><code>0</code>:<code>NO LOCK</code> (always needed by the
   * implementation of this lock)
   * <li><code>1</code>:<code>SHARED</code>
   * <li><code>2</code>:<code>EXCLUSIVE</code>
   * </ul>
   * Accordingly, you will have to set <code>maxLockLevel</code> to
   * <code>2</code>. Now, there are two pairs of levels
   * <ul>
   * <li><code>NO LOCK</code> with <code>EXCLUSIVE</code>
   * <li><code>SHARED</code> with <code>SHARED</code>
   * </ul>
   * This means when the current highest lock level is <code>NO LOCK</code>
   * everything less or equal to <code>EXCLUSIVE</code> is allowed - which means
   * every other lock level. On the other side <code>EXCLUSIVE</code> allows
   * exacly for <code>NO LOCK</code>- which means nothing else. In conclusion,
   * <code>SHARED</code> allows for <code>SHARED</code> or <code>NO
   * LOCK</code>,
   * but not for <code>EXCLUSIVE</code>. To make this very clear have a look at
   * this table, where <code>o</code> means compatible or can coexist and
   * <code>x</code> means incompatible or can not coexist:
   * </p>
   * <table><tbody>
   * <tr>
   * <td align="center"></td>
   * <td align="center">NO LOCK</td>
   * <td align="center">SHARED</td>
   * <td align="center">EXCLUSIVE</td>
   * </tr>
   * <tr>
   * <td align="center">NO LOCK</td>
   * <td align="center">o</td>
   * <td align="center">o</td>
   * <td align="center">o</td>
   * </tr>
   * <tr>
   * <td align="center">SHARED</td>
   * <td align="center">o</td>
   * <td align="center">o</td>
   * <td align="center">x</td>
   * </tr>
   * <tr>
   * <td align="center">EXCLUSIVE</td>
   * <td align="center" align="center">o</td>
   * <td align="center">x</td>
   * <td align="center">x</td>
  * </tr>
  * </tbody> </table>
  * 
  * </p>
  * <p>
  * Additionally, there are preferences for specific locks you can pass to
  * {@link #acquire(Object, int, boolean, int, boolean, long)}. 
  * This means whenever more thanone party
  * waits for a lock you can specify which one is to be preferred. This gives you
  * every freedom you might need to specifcy e.g. 
  * <ul>
  * <li>priority to parties either applying for higher or lower lock levels
  * <li>priority not only to higher or lower locks, but to a specific level
  * <li>completely random preferences
  * </ul>
  * </p>
  * 
  * @version $Id: GenericLock.java 493628 2007-01-07 01:42:48Z joerg $
  */
 public class GenericLock implements MultiLevelLock2 {
 
     protected Object resourceId;
     // XXX needs to be synchronized to allow for unsynchronized access for deadlock detection
     // in getConflictingOwners to avoid deadlocks between lock to acquire and lock to check for
     // deadlocks
     protected Map owners = Collections.synchronizedMap(new HashMap());
     // XXX needs to be synchronized to allow for unsynchronized access for deadlock detection
     // in getConflictingWaiters to avoid deadlocks between lock to acquire and lock to check for
     // deadlocks
     // Note: having this as a list allows for fair mechanisms in sub classes
     protected List waitingOwners = Collections.synchronizedList(new ArrayList());
     private int maxLockLevel;
     protected LoggerFacade logger;
     protected int waiters = 0;
     
     /**
      * Creates a new lock.
      * 
      * @param resourceId identifier for the resource associated to this lock
      * @param maxLockLevel highest allowed lock level as described in class intro 
      * @param logger generic logger used for all kind of debug logging
      */
     public GenericLock(Object resourceId, int maxLockLevel, LoggerFacade logger) {
         if (maxLockLevel < 1)
             throw new IllegalArgumentException(
                 "The maximum lock level must be at least 1 (" + maxLockLevel + " was specified)");
         this.resourceId = resourceId;
         this.maxLockLevel = maxLockLevel;
         this.logger = logger;
     }
 
     public boolean equals(Object o) {
         if (o instanceof GenericLock) {
             return ((GenericLock)o).resourceId.equals(resourceId);
         }
         return false;
     }
 
     public int hashCode() {
         return resourceId.hashCode();
     }
 
     /**
      * @see MultiLevelLock2#test(Object, int, int)
      */
     public boolean test(Object ownerId, int targetLockLevel, int compatibility) {
         boolean success = tryLock(ownerId, targetLockLevel, compatibility, false, true);
         return success;
     }
     
     /**
      * @see MultiLevelLock2#has(Object, int)
      */
     public boolean has(Object ownerId, int lockLevel) {
         int level = getLockLevel(ownerId);
         return (lockLevel <= level);
     }
     
     /**
      * @see org.apache.commons.transaction.locking.MultiLevelLock#acquire(java.lang.Object,
      *      int, boolean, boolean, long)
      */
     public synchronized boolean acquire(Object ownerId, int targetLockLevel, boolean wait,
             boolean reentrant, long timeoutMSecs) throws InterruptedException {
         return acquire(ownerId, targetLockLevel, wait, reentrant ? COMPATIBILITY_REENTRANT
                 : COMPATIBILITY_NONE, timeoutMSecs);
     }
 
     /**
      * @see #acquire(Object, int, boolean, int, boolean, long)
      */
     public synchronized boolean acquire(Object ownerId, int targetLockLevel, boolean wait,
             int compatibility, long timeoutMSecs) throws InterruptedException {
         return acquire(ownerId, targetLockLevel, wait, compatibility, false, timeoutMSecs);
     }
     
     /**
      * Tries to blockingly acquire a lock which can be preferred.
      * 
      * @see #acquire(Object, int, boolean, int, boolean, long) 
      * @since 1.1 
      */
     public synchronized boolean acquire(Object ownerId, int targetLockLevel, boolean preferred,
             long timeoutMSecs) throws InterruptedException {
         return acquire(ownerId, targetLockLevel, true, COMPATIBILITY_REENTRANT, preferred,
                 timeoutMSecs);
     }
     
     /**
      * @see org.apache.commons.transaction.locking.MultiLevelLock2#acquire(Object,
      *      int, boolean, int, boolean, long)
      * @since 1.1 
      */
     public synchronized boolean acquire(
         Object ownerId,
         int targetLockLevel,
         boolean wait,
         int compatibility,
         boolean preferred,
         long timeoutMSecs)
         throws InterruptedException {
 
         if (logger.isFinerEnabled()) {
 	        logger.logFiner(
 	            ownerId.toString()
 	                + " trying to acquire lock for "
 	                + resourceId.toString()
 	                + " at level "
 	                + targetLockLevel
 	                + " at "
 	                + System.currentTimeMillis());
         }
 
         if (tryLock(ownerId, targetLockLevel, compatibility, preferred)) {
             
             if (logger.isFinerEnabled()) {
 	            logger.logFiner(
 	                ownerId.toString()
 	                    + " actually acquired lock for "
 	                    + resourceId.toString()
 	                    + " at "
 	                    + System.currentTimeMillis());
             }
 
             return true;
         } else {
             if (!wait) {
                 return false;
             } else {
                 long started = System.currentTimeMillis();
                 for (long remaining = timeoutMSecs;
                     remaining > 0;
                     remaining = timeoutMSecs - (System.currentTimeMillis() - started)) {
 
                     if (logger.isFinerEnabled()) {
 	                    logger.logFiner(
 	                        ownerId.toString()
 	                            + " waiting on "
 	                            + resourceId.toString()
 	                            + " for msecs "
 	                            + timeoutMSecs
 	                            + " at "
 	                            + System.currentTimeMillis());
                     }
 
                     LockOwner waitingOwner = new LockOwner(ownerId, targetLockLevel, compatibility,
                             preferred);
                     try {
                         registerWaiter(waitingOwner);
                         if (preferred) {
                             // while waiting we already make our claim we are next
                             LockOwner oldLock = null;
                             try {
                                 // we need to remember it to restore it after waiting
                                 oldLock = (LockOwner) owners.get(ownerId);
                                 // this creates a new owner, so we do not need to
                                 // copy the old one
                                 setLockLevel(ownerId, null, targetLockLevel, compatibility,
                                         preferred);
     
                                 // finally wait
                                 wait(remaining);
                                 
                             } finally {
                                 // we need to restore the old lock in order not to
                                 // interfere with the intention lock in the
                                 // following check
                                 // and not to have it in case of success either
                                 // as there will be an ordinary lock then
                                 if (oldLock != null) {
                                     owners.put(ownerId, oldLock);
                                 } else {
                                     owners.remove(ownerId);
                                 }
                             }
     
                         } else {
                             wait(remaining);
                         }
                     } finally {
                         unregisterWaiter(waitingOwner);
                     }
                     
                     if (tryLock(ownerId, targetLockLevel, compatibility, preferred)) {
 
                         if (logger.isFinerEnabled()) {
 	                        logger.logFiner(
 	                            ownerId.toString()
 	                                + " waiting on "
 	                                + resourceId.toString()
 	                                + " eventually got the lock at "
 	                                + System.currentTimeMillis());
                         }
 
                         return true;
                     }
                 }
                 return false;
             }
         }
     }
 
     protected void registerWaiter(LockOwner waitingOwner) {
         synchronized (waitingOwners) {
             unregisterWaiter(waitingOwner);
             waiters++;
             waitingOwners.add(waitingOwner);
         }
     }
 
     protected void unregisterWaiter(LockOwner waitingOwner) {
         synchronized (waitingOwners) {
             if (waitingOwners.remove(waitingOwner))
                 waiters--;
         }
     }
     
     /**
      * @see org.apache.commons.transaction.locking.MultiLevelLock#release(Object)
      */
     public synchronized boolean release(Object ownerId) {
         if (owners.remove(ownerId) != null) {
             if (logger.isFinerEnabled()) {
 	            logger.logFiner(
 	                ownerId.toString()
 	                    + " releasing lock for "
 	                    + resourceId.toString()
 	                    + " at "
 	                    + System.currentTimeMillis());
             }
             notifyAll();
             return true;
         }
         return false;
     }
 
     /**
      * @see org.apache.commons.transaction.locking.MultiLevelLock#getLockLevel(Object)
      */
     public int getLockLevel(Object ownerId) {
         LockOwner owner = (LockOwner) owners.get(ownerId);
         if (owner == null) {
             return 0;
         } else {
             return owner.lockLevel;
         }
     }
 
     /**
      * Gets the resource assotiated to this lock. 
      * 
      * @return identifier for the resource associated to this lock 
      */
     public Object getResourceId() {
         return resourceId;
     }
 
     /**
      * Gets the lowest lock level possible.
      * 
      * @return minimum lock level
      */
     public int getLevelMinLock() {
         return 0;
     }
 
     /**
      * Gets the highst lock level possible.
      * 
      * @return maximum lock level
      */
     public int getLevelMaxLock() {
         return maxLockLevel;
     }
 
     public Object getOwner() {
         LockOwner owner = getMaxLevelOwner();
         if (owner == null)
             return null;
         return owner.ownerId;
     }
 
     public synchronized String toString() {
         StringBuffer buf = new StringBuffer();
         buf.append(resourceId.toString()).append(":\n");
 
         for (Iterator it = owners.values().iterator(); it.hasNext();) {
             LockOwner owner = (LockOwner) it.next();
             buf.append("- ").append(owner.toString()).append("\n");
         }
 
         if (waiters != 0) {
             buf.append(waiters).append(" waiting:\n");
             for (Iterator it = waitingOwners.iterator(); it.hasNext();) {
                 LockOwner owner = (LockOwner) it.next();
                 buf.append("- ").append(owner.toString()).append("\n");
             }
         }
         
         return buf.toString();
     }
 
     protected synchronized LockOwner getMaxLevelOwner() {
         return getMaxLevelOwner(null, -1, false);
     }
 
     protected synchronized LockOwner getMaxLevelOwner(LockOwner reentrantOwner, boolean preferred) {
         return getMaxLevelOwner(reentrantOwner, -1, preferred);
     }
 
     protected synchronized LockOwner getMaxLevelOwner(int supportLockLevel, boolean preferred) {
         return getMaxLevelOwner(null, supportLockLevel, preferred);
     }
 
     protected synchronized LockOwner getMaxLevelOwner(LockOwner reentrantOwner,
             int supportLockLevel, boolean preferred) {
         LockOwner maxOwner = null;
         for (Iterator it = owners.values().iterator(); it.hasNext();) {
             LockOwner owner = (LockOwner) it.next();
             if (owner.lockLevel != supportLockLevel && !owner.equals(reentrantOwner)
                     && (maxOwner == null || maxOwner.lockLevel < owner.lockLevel)
                     // if we are a preferred lock we must not interfere with other intention
                     // locks as we otherwise might mututally lock without resolvation
                     && !(preferred && owner.intention)) {
                 maxOwner = owner;
             }
         }
         return maxOwner;
     }
     
     protected synchronized void setLockLevel(Object ownerId, LockOwner lock, int targetLockLevel,
             int compatibility, boolean intention) {
         // be sure there exists at most one lock per owner
         if (lock != null) {
             if (logger.isFinestEnabled()) {
 	            logger.logFinest(
 	                ownerId.toString()
 	                    + " upgrading lock for "
 	                    + resourceId.toString()
 	                    + " to level "
 	                    + targetLockLevel
 	                    + " at "
 	                    + System.currentTimeMillis());
             }
         } else {
             if (logger.isFinestEnabled()) {
 	            logger.logFinest(
 	                ownerId.toString()
 	                    + " getting new lock for "
 	                    + resourceId.toString()
 	                    + " at level "
 	                    + targetLockLevel
 	                    + " at "
 	                    + System.currentTimeMillis());
             }
         }
         owners.put(ownerId, new LockOwner(ownerId, targetLockLevel, compatibility, intention));
     }
 
     protected boolean tryLock(Object ownerId, int targetLockLevel, int compatibility,
             boolean preferred) {
         return tryLock(ownerId, targetLockLevel, compatibility, preferred, false);
     }
 
     protected synchronized boolean tryLock(Object ownerId, int targetLockLevel, int compatibility,
             boolean preferred, boolean tryOnly) {
 
         LockOwner myLock = (LockOwner) owners.get(ownerId);
 
         // determine highest owner        
         LockOwner highestOwner;
         if (compatibility == COMPATIBILITY_REENTRANT) {
             if (myLock != null && targetLockLevel <= myLock.lockLevel) {
                 // we already have it
                 return true;
             } else {
                 // our own lock will not be compromised by ourself
                 highestOwner = getMaxLevelOwner(myLock, preferred);
             }
         } else if (compatibility == COMPATIBILITY_SUPPORT) {
             // we are compatible with any other lock owner holding
             // the same lock level
             highestOwner = getMaxLevelOwner(targetLockLevel, preferred);
 
         } else if (compatibility == COMPATIBILITY_REENTRANT_AND_SUPPORT) {
             if (myLock != null && targetLockLevel <= myLock.lockLevel) {
                 // we already have it
                 return true;
             } else {
                 // our own lock will not be compromised by ourself and same lock level 
                 highestOwner = getMaxLevelOwner(myLock, targetLockLevel, preferred);
             }
         } else {
             highestOwner = getMaxLevelOwner();
         }
 
         int i;
         // what is our current lock level?
         int currentLockLevel;
         if (highestOwner != null) {
             currentLockLevel = highestOwner.lockLevel;
         } else {
             currentLockLevel = getLevelMinLock();
         }
 
         // we are only allowed to acquire our locks if we do not compromise locks of any other lock owner
         if (isCompatible(targetLockLevel, currentLockLevel)) {
             if (!tryOnly) {
                 // if we really have the lock, it no longer is an intention
                 setLockLevel(ownerId, myLock, targetLockLevel, compatibility, false);
             }
             return true;
         } else {
             return false;
         }
     }
 
     protected boolean isCompatible(int targetLockLevel, int currentLockLevel) {
         return (targetLockLevel <= getLevelMaxLock() - currentLockLevel);
     }
     
     protected Set getConflictingOwners(Object ownerId, int targetLockLevel, int compatibility) {
 
         LockOwner myLock = (LockOwner) owners.get(ownerId);
         if (myLock != null && targetLockLevel <= myLock.lockLevel) {
             // shortcut as we already have the lock
             return null;
         }
         
         LockOwner testLock = new LockOwner(ownerId, targetLockLevel, compatibility, false);
         List ownersCopy;
         synchronized (owners) {
             ownersCopy = new ArrayList(owners.values());
         }
         return getConflictingOwners(testLock, ownersCopy);
         
     }
 
     protected Collection getConflictingWaiters(Object ownerId) {
         LockOwner owner = (LockOwner) owners.get(ownerId);
         if (owner != null) {
             List waiterCopy;
             synchronized (waitingOwners) {
                 waiterCopy = new ArrayList(waitingOwners);
             }
             Collection conflicts = getConflictingOwners(owner, waiterCopy);
             return conflicts;
         }
         return null;
     }
     
     protected Set getConflictingOwners(LockOwner myOwner, Collection ownersToTest) {
 
         if (myOwner == null) return null;
         
         Set conflicts = new HashSet();
 
         // check if any locks conflict with ours
         for (Iterator it = ownersToTest.iterator(); it.hasNext();) {
             LockOwner owner = (LockOwner) it.next();
             
             // we do not interfere with ourselves, except when explicitely said so
             if ((myOwner.compatibility == COMPATIBILITY_REENTRANT || myOwner.compatibility == COMPATIBILITY_REENTRANT_AND_SUPPORT)
                     && owner.ownerId.equals(myOwner.ownerId))
                 continue;
             
             // otherwise find out the lock level of the owner and see if we conflict with it
             int onwerLockLevel = owner.lockLevel;
            
             if (myOwner.compatibility == COMPATIBILITY_SUPPORT
                     || myOwner.compatibility == COMPATIBILITY_REENTRANT_AND_SUPPORT
                     && myOwner.lockLevel == onwerLockLevel)
                 continue;
             
             if (!isCompatible(myOwner.lockLevel, onwerLockLevel)) {
                 conflicts.add(owner.ownerId);
             }
         }
         return (conflicts.isEmpty() ? null : conflicts);
     }
 
     protected static class LockOwner {
         public final Object ownerId;
         public final int lockLevel;
         public final boolean intention;
         public final int compatibility;
 
         public LockOwner(Object ownerId, int lockLevel, int compatibility, boolean intention) {
             this.ownerId = ownerId;
             this.lockLevel = lockLevel;
             this.intention = intention;
             this.compatibility = compatibility;
         }
 
         public String toString() {
             StringBuffer buf = new StringBuffer();
             buf.append(ownerId.toString()).append(": level ").append(lockLevel).append(", complevel ")
                     .append(compatibility).append(intention ? ", intention/preferred" : "");
             return buf.toString();
         }
         
         public boolean equals(Object o) {
             if (o instanceof LockOwner) {
                 return ((LockOwner)o).ownerId.equals(ownerId);
             }
             return false;
         }
         
         public int hashCode() {
             return ownerId.hashCode();
         }
     }
 
 }