package locklib;
   
   import java.util.HashMap;
   import java.util.HashSet;
   import java.util.LinkedList;
   import java.util.List;
   import java.util.Map;
   import java.util.Set;
   
  import ensure.Ensure;
  
  /**
   * A target is an object that can be locked. Targets can be arranged
   * hierarchically. Locking a target may impose locks on its parent targets,
   * depending on the {@link LockPolicy} present in the {@link LockManager}.
   * @param <LockType> the type of locks used by this target
   */
  /*
   * Implementation notes:
   * 
   * This class is prone to deadlocks. In order to avoid
   * deadlocks while navigating the hierarchy, we must always acquire child
   * synchronization locks before parent synchronization locks. We also must
   * acquire target synchronization locks before acquiring synchronization
   * locks on grants or futures.
   * 
   * Locking is done top-down in the hierarchy.
   * 
   * Cancellation and unlock are done bottom-up in the hierarchy. Cancellation
   * starts at the bottom and keeps canceling futures until it either reaches
   * the top of the hierarchy or until it finds a locked node. If it finds a
   * locked node, it switches to unlocking. Unlocking unlocks all nodes from
   * the bottom of a chain (or, at least, the bottom of the chain that has been
   * acquired) upwards.
   * 
   * The most complex part of the algorithm is locking that may happen during
   * unlock. When unlock of a target is done, it may be possible to grant
   * pending locks. However, granting locks may need to proceed down the
   * hierarchy. Because synchronization locks on children must be acquired
   * before synchronization locks on parents, it is not possible to proceed
   * downwards with the synchronization.
   * 
   * To solve this, we will collect the targets that had locks released during
   * unlock (because after releasing the locks the hierarchy may change) and
   * will try to grant locks pending on those targets after all synchronization
   * locks have been released.
   * 
   * The way locklib works, unlock and cancellation are "immediate" actions and
   * lock may take time.
   */
  public class Target<LockType> {
  	/**
  	 * Children of this target. Maps children to all the children's
  	 * children. 
  	 */
  	private Map<Target<LockType>, Set<Target<LockType>>> m_children;
  	
  	/**
  	 * Parent of this target, <code>null</code> if none.
  	 */
  	private Target<LockType> m_parent;
  	
  	/**
  	 * The lock manager.
  	 */
  	private LockManager<LockType> m_manager;
  	
  	/**
  	 * Lock requests currently held.
  	 */
  	private Set<LockGrant<LockType>> m_locks;
  	
  	/**
  	 * The waiting queue, if any.
  	 */
  	private WaitQueue<LockType> m_queue;
  	
  	/**
  	 * Creates a new target.
  	 * @param lm the target's lock manager
  	 */
  	public Target(LockManager<LockType> lm) {
  		Ensure.not_null(lm, "lm == null");
  		m_manager = lm;
  		m_parent = null;
  		m_children = new HashMap<>();
  		m_locks = new HashSet<>();
  		m_queue = null;
  	}
  	
  	/**
  	 * Creates a new target, child of an existing parent.
  	 * @param parent the target's parent
  	 */
  	public Target(Target<LockType> parent) {
  		this(Ensure.not_null(parent, "parent == null").m_manager);
  		parent(parent);
  	}
  
 	/**
 	 * Sets the parent target of this target. Invoking this method does not
 	 * add or remove any locks from either this target, its current parent (if
 	 * any) or its new parent (if any).
 	 * @param p the parent target or <code>null</code> if this target should
 	 * not have a parent
 	 * @throws CyclicTargetHierarchyException setting the parent as
 	 * <code>p</code> would yield a cycle
 	 */
 	public synchronized void parent(Target<LockType> p) {
 		if (p == m_parent) {
 			return;
 		}
 		
 		if (p == this) {
 			throw new CyclicTargetHierarchyException("Cannot add a target as "
 					+ "a parent of itself.");
 		}
 		
 		if (p != null && p.m_manager != m_manager) {
 			throw new DifferentLockManagersException("Target {" + p + "} has "
 					+ "lock manager {" + p.m_manager + "} and this {" + this
 					+ "} has lock manager { " + m_manager + "}.");
 		}
 		
 		if (m_locks.size() > 0) {
 			throw new HierarchyChangeBreaksChainException("Cannot set parent "
 					+ "of target {" + this + "} to {" + p + "} because it "
 					+ "would break existing lock chains.");
 		}
 		
 		if (m_parent != null) {
 			m_parent.remove_child(this);
 			m_parent = null;
 		}
 		
 		if (p != null) {
 			/*
 			 * Lock the new parent and check that we are not a parent of p.
 			 */
 			synchronized (p) {
 				if (m_children.containsKey(p)) {
 					throw new CyclicTargetHierarchyException("Cannot add {"
 							+ p + "} as a parent of {" + this + "} because "
 							+ "it is a child.");
 				}
 				
 				for (Set<Target<LockType>> ch : m_children.values()) {
 					if (ch.contains(p)) {
 						throw new CyclicTargetHierarchyException("Cannot add {"
 								+ p + "} as a parent of {" + this + "} "
 								+ "because it is a (non-immediate) "
 								+ "child.");
 					}
 				}
 				
 				p.m_children.put(this, new HashSet<Target<LockType>>());
 				m_parent = p;
 				update_parent_children();
 			}
 		}
 	}
 	
 	/**
 	 * Removes a child from the parent. This method is invoked with the
 	 * child already lock by the current thread.
 	 * @param c the child
 	 */
 	private synchronized void remove_child(Target<LockType> c) {
 		Ensure.not_null(c, "c == null");
 		Ensure.is_true(m_children.containsKey(c), "!m_children.containsKey(c)");
 		m_children.remove(c);
 		
 		update_parent_children();
 	}
 	
 	/**
 	 * Informs the parent, if any, that the target's children have changed.
 	 * This method is invoked with this object already locked.
 	 */
 	private void update_parent_children() {
 		if (m_parent != null) {
 			Set<Target<LockType>> ch = new HashSet<>(m_children.keySet());
 			
 			for (Set<Target<LockType>> cch : m_children.values()) {
 				ch.addAll(cch);
 			}
 			
 			m_parent.update_children(this, ch);
 		}
 	}
 	
 	/**
 	 * Updates the list of children of one of this object's child. The child
 	 * is already locked.
 	 * @param c the child
 	 * @param cch the child's children
 	 */
 	private synchronized void update_children(Target<LockType> c,
 			Set<Target<LockType>> cch) {
 		Ensure.not_null(c, "c == null");
 		Ensure.is_true(m_children.containsKey(c), "!m_children.containsKey(c)");
 		Ensure.not_null(cch, "cch == null");
 		
 		m_children.put(c,  cch);
 		update_parent_children();
 	}
 
 	/**
 	 * Obtains the parent of this target.
 	 * @return the parent or <code>null</code> if none
 	 */
 	public synchronized Target<LockType> parent() {
 		return m_parent;
 	}
 	
 	/**
 	 * Obtains the set of all children (non-recursive) of this target.
 	 * @return all children
 	 */
 	public synchronized Set<Target<LockType>> children() {
 		return new HashSet<>(m_children.keySet());
 	}
 
 	/**
 	 * <p>Requests an asynchronous lock of this target. This method will return
 	 * immediately. The {@link LockGrantFuture#get()} method (or its timed
 	 * variant) may be used to wait for the lock to be acquired.</p>
 	 * 
 	 * <p>This method will create a chain starting from the top of the
 	 * hierarchy of this target up to this target.</p>
 	 * @param request the lock request which may not be already acquired
 	 * @return the lock grant information
 	 */
 	public synchronized LockGrantFuture<LockType> lock(
 			LockRequest<LockType> request) {
 		Ensure.not_null(request, "request == null");
 		
 		LockGrantFuture<LockType> future = null;
 		
 		/*
 		 * If there is a parent, we must get its future before this one.
 		 * Locking must be done upwards in the hierarchy.
 		 */
 		LockGrantFuture<LockType> plock = null;
 		if (m_parent != null) {
 			LockType plocktype = m_manager.policy().parent_lock(
 					request.type());
 			plock = m_parent.lock(new LockRequest<>(plocktype,
 					request.holder()));
 		}
 		
 		/*
 		 * Create the future with the lock grant.
 		 */
 		future = new LockGrantFuture<>(request, this, plock);
 		
 		/*
 		 * Try to acquire the lock. If we fail to acquire the lock, put it in
 		 * the queue.
 		 */
 		TryLockResult<LockType> r = try_lock(future);
 		Ensure.is_null(r.successor(), "r.successor() != null");
 		if (!r.granted()) {
 			m_manager.pre_checker().pre_check_wait(future.request(), this);
 			if (m_queue == null) {
 				m_queue = m_manager.wait_queue_factory().make();
 			}
 			
 			m_queue.enqueue(future);
 		}
 		
 		return future;
 	}
 	
 	/**
 	 * Tries to lock a future of a lock in this target. If the future's
 	 * predecessor has not been acquired yet, it does nothing. Otherwise tries
 	 * to acquire the lock. If the future is acquired, it is removed from the
 	 * wait queue, if it is there, and the future for the successor in the
 	 * chain is returned.
 	 * @param future the future
 	 * @return the first element in the pair is whether the lock was
 	 * granted or not; the second element in the pair is
 	 * <code>null</code> if the lock was not acquired or if it was
 	 * but there is no successor in the chain; the second element in the pair
 	 * is the successor of the successfully granted lock if the lock was
 	 * granted
 	 */
 	private synchronized TryLockResult<LockType> try_lock(
 			LockGrantFuture<LockType> future) {
 		Ensure.not_null(future, "future == null");
 		Ensure.same(this, future.target(), "future.target() != this");
 		
 		/*
 		 * Set to not null if we should process the successor.
 		 */
 		LockGrantFuture<LockType> successor;
 		boolean succeeded;
 		
 		synchronized (future) {
 			Ensure.is_false(future.isDone(), "future.isDone()");
 			
 			successor = future.successor_future();
 		
 			/*
 			 * If the future has a predecessor which is not done yet,
 			 * we can't do anything.
 			 */
 			LockGrantFuture<LockType> pf = future.predecessor_future();
 			LockGrant<LockType> plg = null;
 			if (pf != null) {
 				synchronized (pf) {
 					if (!pf.isDone()) {
 						return new TryLockResult<>(false, null);
 					}
 					
 					plg = pf.get_now();
 				}
 			}
 			
 			/*
 			 * The future either doesn't have a predecessor or, if it has,
 			 * been granted. Note that the predecessor cannot be cancelled
 			 * because that would require canceling this future.
 			 */
 			if (pf != null) {
 				Ensure.is_false(pf.isCancelled(), "pf.isCancelled()");
 			}
 			
 			if (can_grant(future.request())) {
 				LockGrant<LockType> lg;
 				lg = grant(future.request(), plg);
 				future.acquired(lg);
 				if (m_queue != null) {
 					m_queue.dequeue(future);
 				}
 				
 				succeeded = true;
 			} else {
 				succeeded = false;
 			}
 		}
 		
 		return new TryLockResult<>(succeeded, successor);
 	}
 	
 	/**
 	 * Unlocks a target, which must have been previously locked. This method
 	 * releases all locks a holder has in this target.
 	 * @param g the lock grant
 	 * @throws NotLockedException the lock is not locked or was unlocked
 	 * concurrently during execution of this operation
 	 */
 	public void unlock(LockGrant<LockType> g) {
 		Ensure.not_null(g, "g == null");
 		Ensure.same(this, g.target(), "g.target() != this");
 		Ensure.is_null(g.successor(), "g.successor() != null");
 		
 		List<Target<LockType>> unlocked_targets = new LinkedList<>();
 		unlock_chain(g, unlocked_targets);
 		
 		/*
 		 * See if there are any pending locks in unlocked targets that can be
 		 * granted. Note that we are currently not holding any synchronization
 		 * locks.
 		 */
 		for (Target<LockType> t : unlocked_targets) {
 			t.try_lock_any();
 		}
 	}
 	
 	/**
 	 * Tries to lock any pending locks. This method may propagate to children
 	 * targets if a lock is granted and it has successor futures.
 	 */
 	private void try_lock_any() {
 		/*
 		 * Place in next all futures that we want to try to lock. These are
 		 * initially, 
 		 */
 		LinkedList<LockGrantFuture<LockType>> next = new LinkedList<>();
 		
 		/*
 		 * Process pending lock requests by queue order. Collect all successors
 		 * in next.
 		 */
 		synchronized (this) {
 			boolean done;
 			do {
 				done = false;
 				if (m_queue != null) {
 					/*
 					 * Try all pending locks.
 					 */
 					for (LockGrantFuture<LockType> pending : m_queue) {
 						TryLockResult<LockType> r = try_lock(pending);
 						if (r.granted()) {
 							done = true;
 							if (r.successor() != null) {
 								next.add(r.successor());
 							}
 							
 							break;
 						}
 					}
 				}
 			} while (done);
 		}
 		
 		/*
 		 * Keep processing next and adding successors while we manage to
 		 * succeed in granting locks.
 		 */
 		while (next.size() > 0) {
 			LockGrantFuture<LockType> f = next.removeFirst();
 			Target<LockType> t = f.target();
 			synchronized (t) {
 				synchronized (f) {
 					if (!f.isDone()) {
 						TryLockResult<LockType> r = t.try_lock(f);
 						if (r.successor() != null) {
 							next.add(r.successor());
 						}
 					}
 				}
 			}
 		}
 	}
 	
 	/**
 	 * Unlocks a grant in the current target. Also releases any locks on
 	 * predecessors in the lock chain.
 	 * @param g the lock grant
 	 * @param unlocked receives unlocked targets
 	 * @throws NotLockedException the lock is not locked (or was unlocked
 	 * concurrently during execution of this operation)
 	 */
 	private synchronized void unlock_chain(LockGrant<LockType> g,
 			List<Target<LockType>> unlocked) {
 		Ensure.not_null(g, "g == null");
 		
 		LockGrant<LockType> pred = g.predecessor();
 		synchronized (g) {
 			/*
 			 * Check that the lock is still valid.
 			 */
 			if (!g.valid()) {
 				throw new NotLockedException("Lock grant is no longer "
 						+ "valid. It has already been released.");
 			}
 			
 			/*
 			 * If there is a successor in the chain, it must have already
 			 * been released as locklib users cannot unlock chains from
 			 * the middle.
 			 */
 			LockGrant<LockType> sl = g.successor();
 			if (sl != null) {
 				Ensure.is_false(sl.valid(), "sl.valid()");
 			}
 			
 			/*
 			 * Release the lock.
 			 */
 			Ensure.is_true(m_locks.contains(g), "!m_locks.contains(g)");
 			m_locks.remove(g);
 			m_manager.api().unlocked(g);
 			g.invalidate();
 			unlocked.add(this);
 		}
 		
 		
 		/*
 		 * Unlock upwards in the hierarchy. Note that it is not possible
 		 * that the predecessor has been released concurrently because we
 		 * hold the synchronization lock for this target.
 		 */
 		if (pred != null) {
 			Ensure.is_true(pred.valid(), "!pred.valid()");
 			pred.target().unlock_chain(pred, unlocked);
 		}
 	}
 	
 	/**
 	 * Checks if a given lock request can be granted immediately, ignoring
 	 * the need to acquire a parent lock.
 	 * @param request the request
 	 * @return can the request be granted?
 	 */
 	synchronized boolean can_grant(LockRequest<LockType> request) {
 		/*
 		 * Promotion locks are treated differently as they ignore
 		 * starvation rules.
 		 */
 		boolean is_promotion = false;
 		
 		Ensure.not_null(request, "request == null");
 		for (LockGrant<LockType> l : m_locks) {
 			if (l.request().holder() == request.holder()) {
 				is_promotion = true;
 			} else if (!m_manager.policy().is_compatible(l.request().type(),
 					request.type())) {
 				return false;
 			}
 		}
 		
 		if (!is_promotion && !m_manager.policy().starvation_allowed()) {
 			/*
 			 * Starvation control: we will deny a grant even if there are no
 			 * incompatible locks if we're not the first lock in the queue. If
 			 * there is no queue, it is granted, of course :)
 			 */
 			if (m_queue != null) {
 				for (LockGrantFuture<LockType> f : m_queue) {
 					if (f.request() == request) {
 						/*
 						 * The first queued request is the same as this one so
 						 * we will grant the lock.
 						 */
 						break;
 					}
 					
 					if (f.predecessor_future() == null ||
 							f.predecessor_future().isDone()) {
 						/*
 						 * The first queued request does not have a predecessor
 						 * or has one but it has already been granted. We won't
 						 * allow the request 'request' to move ahead of this
 						 * one to prevent starvation. 
 						 */
 						return false;
 					}
 					
 					/*
 					 * The queued request has a predecessor that has not been
 					 * granted. It will be ignored as it won't be used to
 					 * prevent starvation.
 					 */
 				}
 			}
 			
 		}
 		
 		return true;
 	}
 	
 	/**
 	 * Creates a grant of a request. This method can only be invoked when the
 	 * grant can actually be granted.
 	 * @param request the request
 	 * @param plock an optional parent lock
 	 * @return the grant
 	 */
 	private synchronized LockGrant<LockType> grant(
 			LockRequest<LockType> request, LockGrant<LockType> plock) {
 		Ensure.not_null(request, "request == null");
 		LockGrant<LockType> g = new LockGrant<>(request, this, plock);
 		m_locks.add(g);
 		m_manager.api().locked(g);
 		return g;
 	}
 	
 	/**
 	 * Cancels a future. If the future has already been locked, it does
 	 * nothing. If the future has not yet been locked, it cancels the future
 	 * and cancels or unlocks all its predecessors in the chain. The future
 	 * must be the last element in the chain.
 	 * @param future the future
 	 * @return has the future been cancelled? If the future was already
 	 * cancelled, returns <code>true</code>
 	 */
 	boolean cancel(LockGrantFuture<LockType> future) {
 		Ensure.not_null(future, "future == null");
 		
 		LockGrant<LockType> to_release;
 		synchronized (this) {
 			synchronized (future) {
 				Ensure.is_null(future.successor_future(),
 						"future.successor_future() == null");
 				
 				if (future.isDone()) {
 					return future.isCancelled();
 				}
 				
 				to_release = cancel_chain(future);
 			}
 		}
 		
 		if (to_release != null) {
 			unlock(to_release);
 		}
 		
 		return true;
 	}
 	
 	/**
 	 * Cancels a future. This method will operate recursively to all
 	 * predecessors of a future until all predecessors are cancelled or a
 	 * locked future is found. If a locked future is found, it is returned.
 	 * @param future the future
 	 * @return the locked future found or <code>null</code> if all locks
 	 * were cancelled
 	 */
 	synchronized LockGrant<LockType> cancel_chain(
 			LockGrantFuture<LockType> future) {
 		Ensure.not_null(future, "future == null");
 		
 		synchronized (future) {
 			Ensure.same(future.target(), this, "future.target() == this");
 			if (future.isDone()) {
 				LockGrant<LockType> grant = future.get_now();
 				Ensure.is_true(grant.valid(), "!grant.valid()");
 				return grant;
 			}
 			
 			future.cancelled();
 			
 			/*
 			 * A future cancelled must exist in the wait queue, so remove it.
 			 */
 			Ensure.not_null(m_queue, "m_queue == null");
 			m_queue.dequeue(future);
 			
 			LockGrantFuture<LockType> p = future.predecessor_future();
 			if (p != null) {
 				return p.target().cancel_chain(p);
 			} else {
 				return null;
 			}
 		}
 	}
 	
 	/**
 	 * Result of {@link Target#try_lock(LockGrantFuture)}.
 	 * @param <LockType> the type of locks used by this target
 	 */
 	private static class TryLockResult<LockType> {
 		/**
 		 * Was the lock was granted?
 		 */
 		private boolean m_granted;
 		
 		/**
 		 * Successor of the successfully granted lock if the lock was granted.
 		 * <code>null</code> if the lock was not acquired or if it was but there
 		 * is no successor in the chain.
 		 */
 		private LockGrantFuture<LockType> m_successor;
 		
 		/**
 		 * Creates a new try lock result
 		 * @param granted was the lock granted?
 		 * @param successor what is the lock successor?
 		 */
 		TryLockResult(boolean granted, LockGrantFuture<LockType> successor) {
 			m_granted = granted;
 			m_successor = successor;
 		}
 		
 		/**
 		 * Checks if the lock was granted.
 		 * @return was the lock granted?
 		 */
 		boolean granted() {
 			return m_granted;
 		}
 		
 		/**
 		 * Obtains the lock successor.
 		 * @return the lock successor, <code>null</code> if none
 		 */
 		LockGrantFuture<LockType> successor() {
 			return m_successor;
 		}
 	}
 }