/*
    * 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.Map;
  import java.util.Set;
  
  import org.apache.commons.transaction.util.LoggerFacade;
  
  /**
   * Manager for {@link GenericLock}s on resources. This implementation includes 
   * <ul>
   * <li>deadlock detection, which is configurable to come into effect after an initial short waiting
   * lock request; this is useful as it is somewhat expensive
   * <li>global transaction timeouts that actively revoke granted rights from transactions
   * </ul>
   * 
   * @version $Id: GenericLockManager.java 545634 2007-06-08 21:36:59Z ozeigermann $
   */
  public class GenericLockManager implements LockManager, LockManager2 {
  
      public static final long DEFAULT_TIMEOUT = 30000;
      public static final long DEFAULT_CHECK_THRESHHOLD = 500;
      
      /** Maps onwerId to locks it (partially) owns. */
      protected Map globalOwners = Collections.synchronizedMap(new HashMap());
  
      /** Maps resourceId to lock. */
      protected Map globalLocks = new HashMap();
      
      /** Maps onwerId to global effective time outs (i.e. the time the lock will time out). */
      protected Map effectiveGlobalTimeouts = Collections.synchronizedMap(new HashMap());
  
      protected Set timedOutOwners = Collections.synchronizedSet(new HashSet());
      
      protected int maxLockLevel = -1;
      protected LoggerFacade logger;
      protected long globalTimeoutMSecs;
      protected long checkThreshhold;
      
      /**
       * Creates a new generic lock manager.
       * 
       * @param maxLockLevel
       *            highest allowed lock level as described in {@link GenericLock}
       *            's class intro
       * @param logger
       *            generic logger used for all kind of debug logging
       * @param timeoutMSecs
       *            specifies the maximum time to wait for a lock in milliseconds
       * @param checkThreshholdMSecs
       *            specifies a special wait threshhold before deadlock and
       *            timeout detection come into play or <code>-1</code> switch
       *            it off and check for directly
       * @throws IllegalArgumentException
       *             if maxLockLevel is less than 1
       * 
       * @since 1.1
       */
      public GenericLockManager(int maxLockLevel, LoggerFacade logger, long timeoutMSecs,
              long checkThreshholdMSecs) throws IllegalArgumentException {
          if (maxLockLevel < 1)
              throw new IllegalArgumentException("The maximum lock level must be at least 1 ("
                      + maxLockLevel + " was specified)");
          this.maxLockLevel = maxLockLevel;
          this.logger = logger.createLogger("Locking");
          this.globalTimeoutMSecs = timeoutMSecs;
          this.checkThreshhold = checkThreshholdMSecs;
      }
  
      public GenericLockManager(int maxLockLevel, LoggerFacade logger, long timeoutMSecs)
              throws IllegalArgumentException {
          this(maxLockLevel, logger, timeoutMSecs, DEFAULT_CHECK_THRESHHOLD);
      }
  
      public GenericLockManager(int maxLockLevel, LoggerFacade logger)
              throws IllegalArgumentException {
          this(maxLockLevel, logger, DEFAULT_TIMEOUT);
      }
 
     /**
      * @see LockManager2#startGlobalTimeout(Object, long)
      * @since 1.1
      */
     public void startGlobalTimeout(Object ownerId, long timeoutMSecs) {
         long now = System.currentTimeMillis();
         long timeout = now + timeoutMSecs;
         effectiveGlobalTimeouts.put(ownerId, new Long(timeout));
     }
     
     /**
      * @see LockManager2#tryLock(Object, Object, int, boolean)
      * @since 1.1
      */
     public boolean tryLock(Object ownerId, Object resourceId, int targetLockLevel, boolean reentrant) {
         timeoutCheck(ownerId);
 
         GenericLock lock = (GenericLock) atomicGetOrCreateLock(resourceId);
         boolean acquired = lock.tryLock(ownerId, targetLockLevel,
                 reentrant ? GenericLock.COMPATIBILITY_REENTRANT : GenericLock.COMPATIBILITY_NONE,
                 false);
         
         if (acquired) {
             addOwner(ownerId, lock);
         }
         return acquired;
     }
 
     /**
      * @see LockManager2#checkLock(Object, Object, int, boolean)
      * @since 1.1
      */
     public boolean checkLock(Object ownerId, Object resourceId, int targetLockLevel, boolean reentrant) {
         timeoutCheck(ownerId);
         boolean possible = true;
 
         GenericLock lock = (GenericLock) getLock(resourceId);
         if (lock != null) {
             possible = lock.test(ownerId, targetLockLevel,
                     reentrant ? GenericLock.COMPATIBILITY_REENTRANT
                             : GenericLock.COMPATIBILITY_NONE);
         }
         return possible;
     }
 
     /**
      * @see LockManager2#hasLock(Object, Object, int)
      * @since 1.1
      */
     public boolean hasLock(Object ownerId, Object resourceId, int lockLevel) {
         timeoutCheck(ownerId);
         boolean owned = false;
 
         GenericLock lock = (GenericLock) getLock(resourceId);
         if (lock != null) {
             owned = lock.has(ownerId, lockLevel);
         }
         return owned;
     }
 
     /**
      * @see LockManager2#lock(Object, Object, int, boolean)
      * @since 1.1
      */
     public void lock(Object ownerId, Object resourceId, int targetLockLevel, boolean reentrant)
             throws LockException {
         lock(ownerId, resourceId, targetLockLevel, reentrant, globalTimeoutMSecs);
     }
 
     /**
      * @see LockManager2#lock(Object, Object, int, boolean, long)
      * @since 1.1
      */
     public void lock(Object ownerId, Object resourceId, int targetLockLevel, boolean reentrant,
             long timeoutMSecs) throws LockException {
         lock(ownerId, resourceId, targetLockLevel, reentrant ? GenericLock.COMPATIBILITY_REENTRANT
                 : GenericLock.COMPATIBILITY_NONE, false, timeoutMSecs);
     }
 
     /**
      * @see LockManager2#lock(Object, Object, int, int, boolean, long)
      * @since 1.1
      */
     public void lock(Object ownerId, Object resourceId, int targetLockLevel, int compatibility,
             boolean preferred, long timeoutMSecs) throws LockException {
         timeoutCheck(ownerId);
         GenericLock lock = (GenericLock) atomicGetOrCreateLock(resourceId);
         doLock(lock, ownerId, resourceId, targetLockLevel, compatibility, preferred, timeoutMSecs);
     }
 
     protected void doLock(GenericLock lock, Object ownerId, Object resourceId, int targetLockLevel,
                           int compatibility, boolean preferred, long timeoutMSecs)
     {
         long now = System.currentTimeMillis();
         long waitEnd = now + timeoutMSecs;
 
         timeoutCheck(ownerId);
         
         GenericLock.LockOwner lockWaiter = new GenericLock.LockOwner(ownerId, targetLockLevel,
                 compatibility, preferred);
         
         boolean acquired = false;
         try {
             
             // detection for deadlocks and time outs is rather expensive, 
             // so we wait for the lock for a  
             // short time (<5 seconds) to see if we get it without checking;
             // if not we still can check what the reason for this is
             if (checkThreshhold != -1 && timeoutMSecs > checkThreshhold) {
                 acquired = lock
                         .acquire(ownerId, targetLockLevel, true, compatibility,
                                 preferred, checkThreshhold);
                 timeoutMSecs -= checkThreshhold;
             } else {
                 acquired = lock
                 .acquire(ownerId, targetLockLevel, false, compatibility,
                         preferred, checkThreshhold);
                 
             }
             if (acquired) {
                 addOwner(ownerId, lock);
                 return;
             }
         } catch (InterruptedException e) {
             throw new LockException("Interrupted", LockException.CODE_INTERRUPTED, resourceId);
         }
         try {
             lock.registerWaiter(lockWaiter);
             
             boolean deadlock = wouldDeadlock(ownerId, new HashSet());
             if (deadlock) {
                 throw new LockException("Lock would cause deadlock",
                         LockException.CODE_DEADLOCK_VICTIM, resourceId);
             }
 
             now = System.currentTimeMillis();
             while (!acquired && waitEnd > now) {
             
                 // first be sure all locks are stolen from owners that have already timed out
                 releaseTimedOutOwners();
 
                 // if there are owners we conflict with lets see if one of them globally times
                 // out earlier than this lock, if so we will wake up then to check again
                 Set conflicts = lock.getConflictingOwners(ownerId, targetLockLevel, compatibility);
                 long nextConflictTimeout = getNextGlobalConflictTimeout(conflicts);
                 if (nextConflictTimeout != -1 && nextConflictTimeout < waitEnd) {
                     timeoutMSecs = nextConflictTimeout - now;
                     // XXX add 10% to ensure the lock really is timed out
                     timeoutMSecs += timeoutMSecs / 10;
                 } else {
                     timeoutMSecs = waitEnd - now;
                 }
 
                 // XXX acquire will remove us as a waiter, but it is important to remain us such
                 // to constantly indicate it to other owners, otherwise there might be undetected
                 // deadlocks
                 synchronized (lock) {
                     acquired = lock.acquire(ownerId, targetLockLevel, true, compatibility,
                             preferred, timeoutMSecs);
                     lock.registerWaiter(lockWaiter);
                 }
                 now = System.currentTimeMillis();
             }
             if (!acquired) {
                 throw new LockException("Lock wait timed out", LockException.CODE_TIMED_OUT,
                         resourceId);
             } else {
                 addOwner(ownerId, lock);
             }
         } catch (InterruptedException e) {
             throw new LockException("Interrupted", LockException.CODE_INTERRUPTED, resourceId);
         } finally {
             lock.unregisterWaiter(lockWaiter);
         }
     }
 
     /**
      * @see LockManager2#getLevel(Object, Object)
      * @since 1.1
      */
     public int getLevel(Object ownerId, Object resourceId) {
         timeoutCheck(ownerId);
         GenericLock lock = (GenericLock) getLock(resourceId);
         if (lock != null) {
             return lock.getLockLevel(ownerId);
         } else {
             return 0;
         }
     }
 
     /**
      * @see LockManager2#release(Object, Object)
      * @since 1.1
      */
     public boolean release(Object ownerId, Object resourceId) {
         timeoutCheck(ownerId);
         boolean released = false;
 
         GenericLock lock = (GenericLock) getLock(resourceId);
         if (lock != null) {
             released = lock.release(ownerId);
             removeOwner(ownerId, lock);
         }
         return released;
     }
 
     /**
      * @see LockManager2#releaseAll(Object)
      * @since 1.1
      */
     public void releaseAll(Object ownerId) {
         releaseAllNoTimeOutReset(ownerId);
         // reset time out status for this owner
         timedOutOwners.remove(ownerId);
         effectiveGlobalTimeouts.remove(ownerId);
     }
 
     protected void releaseAllNoTimeOutReset(Object ownerId) {
         Set locks = (Set) globalOwners.get(ownerId);
         if (locks != null) {
             Collection locksCopy;
             // need to copy in order not to interfere with wouldDeadlock
             // possibly called by
             // other threads
             synchronized (locks) {
                 locksCopy = new ArrayList(locks);
             }
             for (Iterator it = locksCopy.iterator(); it.hasNext();) {
                 GenericLock lock = (GenericLock) it.next();
                 lock.release(ownerId);
                 locks.remove(lock);
             }
         }
         removeOwnerWithoutLocks(ownerId);
     }
     
     /**
      * @see LockManager2#getAll(Object)
      * @since 1.1
      */
     public Set getAll(Object ownerId) {
         Set locks = (Set) globalOwners.get(ownerId);
         if (locks == null) {
             return new HashSet();
         } else {
             return locks;
         }
     }
 
     protected void addOwner(Object ownerId, GenericLock lock) {
         synchronized (globalOwners) {
             Set locks = (Set) globalOwners.get(ownerId);
             if (locks == null) {
                 locks = Collections.synchronizedSet(new HashSet());
                 globalOwners.put(ownerId, locks);
             }
             locks.add(lock);
         }
     }
 
     protected void removeOwner(Object ownerId, GenericLock lock) {
         Set locks = (Set) globalOwners.get(ownerId);
         if (locks != null) {
             locks.remove(lock);
         }
         removeOwnerWithoutLocks(ownerId);
     }
 
     /**
      * Checks if an owner is deadlocked. <br>
      * <br>
      * We traverse the tree recursively formed by owners, locks held by them and
      * then again owners waiting for the locks. If there is a cycle in one of
      * the paths from the root to a leaf we have a deadlock. <br>
      * <br>
      * A more detailed discussion on deadlocks and definitions and how to detect
      * them can be found in <a
      * href="http://www.onjava.com/pub/a/onjava/2004/10/20/threads2.html?page=1">this
      * nice article </a>. <br>
      * <em>Caution:</em> This computation can be very expensive with many
      * owners and locks. Worst (unlikely) case is exponential.
      * 
      * @param ownerId
      *            the owner to check for being deadlocked
      * @param path
      *            initially should be called with an empty set
      * @return <code>true</code> if the owner is deadlocked,
      *         <code>false</code> otherwise
      */
     protected boolean wouldDeadlock(Object ownerId, Set path) {
         path.add(ownerId);
         // these are our locks
         Set locks = (Set) globalOwners.get(ownerId);
         if (locks != null) {
             Collection locksCopy;
             // need to copy in order not to interfere with releaseAll possibly called by
             // other threads
             synchronized (locks) {
                 locksCopy = new ArrayList(locks);
             }
             for (Iterator i = locksCopy.iterator(); i.hasNext();) {
                 GenericLock mylock = (GenericLock) i.next();
                 // these are the ones waiting for one of our locks
                 Collection conflicts = mylock.getConflictingWaiters(ownerId);
                 if (conflicts != null) {
                     for (Iterator j = conflicts.iterator(); j.hasNext();) {
                         Object waitingOwnerId = j.next();
                         if (path.contains(waitingOwnerId)) {
                             return true;
                         } else if (wouldDeadlock(waitingOwnerId, path)) {
                             return true;
                         }
                     }
                 }
             }
         }
         path.remove(ownerId);
         return false;
     }
 
     protected boolean releaseTimedOutOwners() {
         boolean released = false;
         synchronized (effectiveGlobalTimeouts) {
             for (Iterator it = effectiveGlobalTimeouts.entrySet().iterator(); it.hasNext();) {
                 Map.Entry entry = (Map.Entry) it.next();
                 Object ownerId = entry.getKey();
                 long timeout = ((Long) entry.getValue()).longValue();
                 long now = System.currentTimeMillis();
                 if (timeout < now) {
                     releaseAllNoTimeOutReset(ownerId);
                     timedOutOwners.add(ownerId);
                     released = true;
                 }
             }
         }
         return released;
     }
     
     protected boolean timeOut(Object ownerId) {
         Long timeout = (Long)effectiveGlobalTimeouts.get(ownerId);
         long now = System.currentTimeMillis();
         if (timeout != null && timeout.longValue() < now) {
             releaseAll(ownerId);
             timedOutOwners.add(ownerId);
             return true;
         } else {
             return false;
         }
     }
     
     protected long getNextGlobalConflictTimeout(Set conflicts) {
         long minTimeout = -1;
         long now = System.currentTimeMillis();
         if (conflicts != null) {
             synchronized (effectiveGlobalTimeouts) {
                 for (Iterator it = effectiveGlobalTimeouts.entrySet().iterator(); it.hasNext();) {
                     Map.Entry entry = (Map.Entry) it.next();
                     Object ownerId = entry.getKey();
                     if (conflicts.contains(ownerId)) {
                         long timeout = ((Long) entry.getValue()).longValue();
                         if (minTimeout == -1 || timeout < minTimeout) {
                             minTimeout = timeout;
                         }
                     }
                 }
             }
         }
         return minTimeout;
     }
     
     public MultiLevelLock getLock(Object resourceId) {
         synchronized (globalLocks) {
             return (MultiLevelLock) globalLocks.get(resourceId);
         }
     }
 
     public MultiLevelLock atomicGetOrCreateLock(Object resourceId) {
         synchronized (globalLocks) {
             MultiLevelLock lock = getLock(resourceId);
             if (lock == null) {
                 lock = createLock(resourceId);
             }
             return lock;
         }
     }
 
     public void removeLock(MultiLevelLock lock) {
         synchronized (globalLocks) {
             globalLocks.remove(lock);
         }
     }
     
     /**
      * Gets all locks as orignials, <em>no copies</em>.
      * 
      * @return collection holding all locks.
      */
     public Collection getLocks() {
         synchronized (globalLocks) {
            return globalLocks.values();
         }
     }
 
     public synchronized String toString() {
         StringBuffer buf = new StringBuffer(1000);
         for (Iterator it = globalLocks.values().iterator(); it.hasNext();) {
             GenericLock lock = (GenericLock) it.next();
             buf.append(lock.toString()).append('\n');
         }
         return buf.toString();
     }
 
     protected GenericLock createLock(Object resourceId) {
         synchronized (globalLocks) {
             GenericLock lock = new GenericLock(resourceId, maxLockLevel, logger);
             globalLocks.put(resourceId, lock);
             return lock;
         }
     }
     
     protected void timeoutCheck(Object ownerId) throws LockException {
         timeOut(ownerId);
         if (timedOutOwners.contains(ownerId)) {
             throw new LockException(
                     "All locks of owner "
                             + ownerId
                             + " have globally timed out."
                             + " You will not be able to to continue with this owner until you call releaseAll.",
                     LockException.CODE_TIMED_OUT, null);
         }
     }
 
     protected void removeOwnerWithoutLocks(Object ownerId) {
         synchronized (globalOwners) {
             Set locks = (Set) globalOwners.get(ownerId);
             if (locks == null || locks.isEmpty()) {
                 globalOwners.remove(ownerId);
             }
         }
     }
 }