/*
 * Copyright (c) 1998, 2021 Oracle and/or its affiliates. All rights reserved.
 * Copyright (c) 1998, 2021 IBM Corporation. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0,
 * or the Eclipse Distribution License v. 1.0 which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
 */

// Contributors:
//     Oracle - initial API and implementation from Oracle TopLink
//     02/11/2009-1.1 Michael O'Brien
//        - 259993: As part 2) During mergeClonesAfterCompletion()
//           If the the acquire and release threads are different
//           switch back to the stored acquire thread stored on the mergeManager.
//      tware, David Mulligan - fix performance issue with releasing locks
//     11/07/2017 - Dalia Abo Sheasha
//       - 526957 : Split the logging and trace messages
package org.eclipse.persistence.internal.helper;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

import org.eclipse.persistence.descriptors.ClassDescriptor;
import org.eclipse.persistence.descriptors.FetchGroupManager;
import org.eclipse.persistence.exceptions.ConcurrencyException;
import org.eclipse.persistence.internal.helper.linkedlist.ExposedNodeLinkedList;
import org.eclipse.persistence.internal.identitymaps.CacheKey;
import org.eclipse.persistence.internal.localization.TraceLocalization;
import org.eclipse.persistence.internal.queries.ContainerPolicy;
import org.eclipse.persistence.internal.sessions.AbstractSession;
import org.eclipse.persistence.internal.sessions.MergeManager;
import org.eclipse.persistence.internal.sessions.ObjectChangeSet;
import org.eclipse.persistence.internal.sessions.UnitOfWorkChangeSet;
import org.eclipse.persistence.logging.SessionLog;
import org.eclipse.persistence.mappings.DatabaseMapping;

import static java.util.Collections.unmodifiableMap;

/**
 * INTERNAL:
 * <p>
 * <b>Purpose</b>: Acquires all required locks for a particular merge process.
 * Implements a deadlock avoidance algorithm to prevent concurrent merge conflicts.
 *
 * <p>
 * <b>Responsibilities</b>:
 * <ul>
 * <li> Acquires locks for writing threads.
 * <li> Provides deadlock avoidance behavior.
 * <li> Releases locks for writing threads.
 * </ul>
 * @author Gordon Yorke
 * @since 10.0.3
 */
public class WriteLockManager {

    /**
     * The code spots where we use this constant are code spots where we afraid the thread might be trying to run a
     * commit. Blowing up the thread with an interrupted exception might be too dangerous. We are not certain the
     * eclipselink code is able to cope with it and release all resources appropriately.
     *
     */
    private static final Boolean ALLOW_INTERRUPTED_EXCEPTION_TO_BE_FIRED_UP_FALSE = false;

    /**
     * This flag we use if the write lock manager is stuck building clones of objects. Because we are not in the code
     * area of commit to the db anything.
     */
    private static final Boolean ALLOW_INTERRUPTED_EXCEPTION_TO_BE_FIRED_UP_TRUE = true;

    /**
     * This a map from a thread to cache keys the thread is finding itself not being able to acquire. This map is
     * important to explain why a thread might be stuck in a stack trace of the form: {@code
     * at java.lang.Class.getEnclosingMethod0(Native Method)
    at java.lang.Class.getEnclosingMethodInfo(Class.java:1072)
    at java.lang.Class.getEnclosingClass(Class.java:1272)
    at java.lang.Class.getSimpleBinaryName(Class.java:1443)
    at java.lang.Class.getSimpleName(Class.java:1309)
    at org.eclipse.persistence.internal.identitymaps.IdentityMapManager.acquireLockNoWait(IdentityMapManager.java:205)
    at org.eclipse.persistence.internal.sessions.IdentityMapAccessor.acquireLockNoWait(IdentityMapAccessor.java:108)


    at org.eclipse.persistence.internal.helper.WriteLockManager.attemptToAcquireLock(WriteLockManager.java:431)
    at org.eclipse.persistence.internal.helper.WriteLockManager.acquireRequiredLocks(WriteLockManager.java:280)
     * }
     *
     * We want to be able to trace these dead lock situations. To put them our on the massive log dump as to do the dead
     * lock detection.
     *
     */
    private static final Map<Thread, Set<ConcurrencyManager>> THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS = new ConcurrentHashMap<>();

    /**
     * We want to have traceability of what objects where changed by thread that is in the middle of a commit. This
     * information can be useful when a massive dump is performed to explain the situation of any thread that might
     * eventually be stuck inside of the write lock manager to tells us what exactly are the objects it has changed and
     * wants to commit or merge into the shared cache. Relates to the
     * {@link #THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS} but this map does not tells us about any
     * specific problem such as a cache key that could not be acquired just tells us what objects were modified.
     *
     */
    private static final Map<Thread, Set<Object>> MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET = new ConcurrentHashMap<>();

    /** Semaphore related properties */
    private static final transient ThreadLocal<Boolean> SEMAPHORE_THREAD_LOCAL_VAR = new ThreadLocal<>();
    private static final transient int SEMAPHORE_MAX_NUMBER_THREADS = ConcurrencyUtil.SINGLETON.getNoOfThreadsAllowedToDoWriteLockManagerAcquireRequiredLocksInParallel();
    private static final transient Semaphore SEMAPHORE_LIMIT_MAX_NUMBER_OF_THREADS_WRITE_LOCK_MANAGER = new Semaphore(SEMAPHORE_MAX_NUMBER_THREADS);
    private transient ConcurrencySemaphore writeLockManagerSemaphore = new ConcurrencySemaphore(SEMAPHORE_THREAD_LOCAL_VAR, SEMAPHORE_MAX_NUMBER_THREADS, SEMAPHORE_LIMIT_MAX_NUMBER_OF_THREADS_WRITE_LOCK_MANAGER, this,"write_lock_manager_semaphore_acquired_01");

    // this will allow us to prevent a readlock thread from looping forever.
    public static final int MAXTRIES = 10000;

    public static final int MAX_WAIT = 600000; //10 mins

    /* This attribute stores the list of threads that have had a problem acquiring locks */
    /*  the first element in this list will be the prevailing thread */
    protected ExposedNodeLinkedList prevailingQueue;

    public WriteLockManager() {
        this.prevailingQueue = new ExposedNodeLinkedList();
    }

    /**
     * INTERNAL:
     * This method will return once the object is locked and all non-indirect
     * related objects are also locked.
     */
    public Map acquireLocksForClone(Object objectForClone, ClassDescriptor descriptor, CacheKey cacheKey, AbstractSession cloningSession) {
        // determineIfReleaseDeferredLockAppearsToBeDeadLocked
        final long whileStartTimeMillis = System.currentTimeMillis();
        final Thread currentThread = Thread.currentThread();
        DeferredLockManager lockManager = ConcurrencyManager.getDeferredLockManager(currentThread);
        ReadLockManager readLockManager = ConcurrencyManager.getReadLockManager(currentThread);

        boolean successful = false;
        IdentityHashMap lockedObjects = new IdentityHashMap();
        IdentityHashMap refreshedObjects = new IdentityHashMap();

        CacheKey lastCacheKeyWeNeededToWaitToAcquire = null;

        try {
            // if the descriptor has indirection for all mappings then wait as there will be no deadlock risks
            CacheKey toWaitOn = acquireLockAndRelatedLocks(objectForClone, lockedObjects, refreshedObjects, cacheKey, descriptor, cloningSession);
            int tries = 0;
            while (toWaitOn != null) {// loop until we've tried too many times.
                for (Iterator lockedList = lockedObjects.values().iterator(); lockedList.hasNext();) {
                    ((CacheKey)lockedList.next()).releaseReadLock();
                    lockedList.remove();
                }

                // of the concurrency manager that we use for creating the massive log dump
                // to indicate that the current thread is now stuck trying to acquire some arbitrary
                // cache key for writing
                StackTraceElement stackTraceElement = Thread.currentThread().getStackTrace()[1];
                lastCacheKeyWeNeededToWaitToAcquire = toWaitOn;
                lastCacheKeyWeNeededToWaitToAcquire.putThreadAsWaitingToAcquireLockForWriting(currentThread, stackTraceElement.getClassName() + "." + stackTraceElement.getMethodName() + "(...)");

                // Since we know this one of those methods that can appear in the dead locks
                // we threads frozen here forever inside of the wait that used to have no timeout
                // we will now always check for how long the current thread is stuck in this while loop going nowhere
                // using the exact same approach we have been adding to the concurrency manager
                ConcurrencyUtil.SINGLETON.determineIfReleaseDeferredLockAppearsToBeDeadLocked(toWaitOn, whileStartTimeMillis, lockManager, readLockManager, ALLOW_INTERRUPTED_EXCEPTION_TO_BE_FIRED_UP_TRUE);

                synchronized (toWaitOn) {
                    try {
                        if (toWaitOn.isAcquired()) {//last minute check to insure it is still locked.
                            toWaitOn.wait(ConcurrencyUtil.SINGLETON.getAcquireWaitTime());// wait for lock on object to be released
                        }
                    } catch (InterruptedException ex) {
                        // Ignore exception thread should continue.
                    }
                }
                Object waitObject = toWaitOn.getObject();
                // Object may be null for loss of identity.
                if (waitObject != null) {
                    cloningSession.checkAndRefreshInvalidObject(waitObject, toWaitOn, cloningSession.getDescriptor(waitObject));
                    refreshedObjects.put(waitObject, waitObject);
                }
                toWaitOn = acquireLockAndRelatedLocks(objectForClone, lockedObjects, refreshedObjects, cacheKey, descriptor, cloningSession);
                if ((toWaitOn != null) && ((++tries) > MAXTRIES)) {
                    // If we've tried too many times abort.
                    throw ConcurrencyException.maxTriesLockOnCloneExceded(objectForClone);
                }
            }
            successful = true;//successfully acquired all locks
        } catch (InterruptedException exception) {
            throw ConcurrencyException.maxTriesLockOnCloneExceded(objectForClone);
        } finally {
            if (lastCacheKeyWeNeededToWaitToAcquire != null) {
                lastCacheKeyWeNeededToWaitToAcquire.removeThreadNoLongerWaitingToAcquireLockForWriting(currentThread);
            }
            if (!successful) {//did not acquire locks but we are exiting
                for (Iterator lockedList = lockedObjects.values().iterator(); lockedList.hasNext();) {
                    ((CacheKey)lockedList.next()).releaseReadLock();
                    lockedList.remove();
                }
            }
        }
        return lockedObjects;
    }

    /**
     * INTERNAL:
     * This is a recursive method used to acquire read locks on all objects that
     * will be cloned.  These include all related objects for which there is no
     * indirection.
     * The returned object is the first object that the lock could not be acquired for.
     * The caller must try for exceptions and release locked objects in the case
     * of an exception.
     */
    public CacheKey acquireLockAndRelatedLocks(Object objectForClone, Map lockedObjects, Map refreshedObjects, CacheKey cacheKey, ClassDescriptor descriptor, AbstractSession cloningSession) {
        if (!refreshedObjects.containsKey(objectForClone) && cloningSession.isConsideredInvalid(objectForClone, cacheKey, descriptor)) {
            return cacheKey;
        }
        // Attempt to get a read-lock, null is returned if cannot be read-locked.
        if (cacheKey.acquireReadLockNoWait()) {
            if (cacheKey.getObject() == null) {
                // This will be the case for deleted objects, NoIdentityMap, and aggregates.
                lockedObjects.put(objectForClone, cacheKey);
            } else {
                objectForClone = cacheKey.getObject();
                if (lockedObjects.containsKey(objectForClone)) {
                    // This is a check for loss of identity, the original check in
                    // checkAndLockObject() will shortcircuit in the usual case.
                    cacheKey.releaseReadLock();
                    return null;
                }
                // Store locked cachekey for release later.
                lockedObjects.put(objectForClone, cacheKey);
            }
            return traverseRelatedLocks(objectForClone, lockedObjects, refreshedObjects, descriptor, cloningSession);
        } else {
            // Return the cache key that could not be locked.
            return cacheKey;
        }
    }

    /**
     * INTERNAL:
     * This method will transition the previously acquired active
     * locks to deferred locks in the case a readlock could not be acquired for
     * a related object. Deferred locks must be employed to prevent deadlock
     * when waiting for the readlock while still protecting readers from
     * incomplete data.
     */
    public void transitionToDeferredLocks(MergeManager mergeManager){
        try{
            if (mergeManager.isTransitionedToDeferredLocks()) {
                return;
            }
            for (CacheKey cacheKey : mergeManager.getAcquiredLocks()){
                cacheKey.transitionToDeferredLock();
            }
            mergeManager.transitionToDeferredLocks();
        }catch (RuntimeException ex){
            for (CacheKey cacheKey : mergeManager.getAcquiredLocks()){
                cacheKey.release();
            }
            ConcurrencyManager.getDeferredLockManager(Thread.currentThread()).setIsThreadComplete(true);
            ConcurrencyManager.removeDeferredLockManager(Thread.currentThread());
            mergeManager.getAcquiredLocks().clear();
            throw ex;
        }
    }

    /**
     * INTERNAL:
     * Traverse the object and acquire locks on all related objects.
     */
    public CacheKey traverseRelatedLocks(Object objectForClone, Map lockedObjects, Map refreshedObjects, ClassDescriptor descriptor, AbstractSession cloningSession) {
        // If all mappings have indirection short-circuit.
        if (descriptor.shouldAcquireCascadedLocks()) {
            FetchGroupManager fetchGroupManager = descriptor.getFetchGroupManager();
            boolean isPartialObject = (fetchGroupManager != null) && fetchGroupManager.isPartialObject(objectForClone);
            for (Iterator<DatabaseMapping> mappings = descriptor.getLockableMappings().iterator();
                 mappings.hasNext();) {
                DatabaseMapping mapping = mappings.next();
                // Only cascade fetched mappings.
                if (!isPartialObject || (fetchGroupManager.isAttributeFetched(objectForClone, mapping.getAttributeName()))) {
                    // any mapping in this list must not have indirection.
                    Object objectToLock = mapping.getAttributeValueFromObject(objectForClone);
                    if (mapping.isCollectionMapping()) {
                        // Ignore null, means empty.
                        if (objectToLock != null) {
                            ContainerPolicy cp = mapping.getContainerPolicy();
                            Object iterator = cp.iteratorFor(objectToLock);
                            while (cp.hasNext(iterator)) {
                                Object object = cp.next(iterator, cloningSession);
                                if (mapping.getReferenceDescriptor().hasWrapperPolicy()) {
                                    object = mapping.getReferenceDescriptor().getWrapperPolicy().unwrapObject(object, cloningSession);
                                }
                            CacheKey toWaitOn = checkAndLockObject(object, lockedObjects, refreshedObjects, mapping, cloningSession);
                                if (toWaitOn != null) {
                                    return toWaitOn;
                                }
                            }
                        }
                    } else {
                        if (mapping.getReferenceDescriptor().hasWrapperPolicy()) {
                            objectToLock = mapping.getReferenceDescriptor().getWrapperPolicy().unwrapObject(objectToLock, cloningSession);
                        }
                    CacheKey toWaitOn = checkAndLockObject(objectToLock, lockedObjects, refreshedObjects, mapping, cloningSession);
                        if (toWaitOn != null) {
                            return toWaitOn;
                        }
                    }
                }
            }
        }
        return null;
    }

    /**
     * INTERNAL:
     * This method will be the entry point for threads attempting to acquire locks for all objects that have
     * a changeset.  This method will hand off the processing of the deadlock algorithm to other member
     * methods.  The mergeManager must be the active mergemanager for the calling thread.
     * Returns true if all required locks were acquired
     * This is wrapper method with semaphore logic.
     */
    public void acquireRequiredLocks(MergeManager mergeManager, UnitOfWorkChangeSet changeSet) {
        boolean semaphoreWasAcquired = false;
        boolean useSemaphore = ConcurrencyUtil.SINGLETON.isUseSemaphoreToLimitConcurrencyOnWriteLockManagerAcquireRequiredLocks();
        try {
            semaphoreWasAcquired = writeLockManagerSemaphore.acquireSemaphoreIfAppropriate(useSemaphore);
            acquireRequiredLocksInternal(mergeManager, changeSet);
        } finally {
            writeLockManagerSemaphore.releaseSemaphoreAllowOtherThreadsToStartDoingObjectBuilding(semaphoreWasAcquired);
        }
    }

    /**
     * INTERNAL:
     * This method will be the entry point for threads attempting to acquire locks for all objects that have
     * a changeset.  This method will hand off the processing of the deadlock algorithm to other member
     * methods.  The mergeManager must be the active mergemanager for the calling thread.
     * Returns true if all required locks were acquired
     */
    private void acquireRequiredLocksInternal(MergeManager mergeManager, UnitOfWorkChangeSet changeSet) {
        if (!MergeManager.LOCK_ON_MERGE) {//lockOnMerge is a backdoor and not public
            return;
        }
        boolean locksToAcquire = true;

        final Thread currentThread = Thread.currentThread();
        final long timeWhenLocksToAcquireLoopStarted = System.currentTimeMillis();
        populateMapThreadToObjectIdsWithChagenSet(currentThread, changeSet.getAllChangeSets().values());
        clearMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(currentThread);

        //while that thread has locks to acquire continue to loop.
        try {
            // initialize the MergeManager during this commit or merge for insert/updates only
            // this call is not required in acquireLocksForClone() or acquireLockAndRelatedLocks()
            mergeManager.setLockThread(Thread.currentThread());

            AbstractSession session = mergeManager.getSession();
            // If the session in the mergemanager is not a unit of work then the
            // merge is of a changeSet into a distributed session.
            if (session.isUnitOfWork()) {
                session = session.getParent();
            }
            while (locksToAcquire) {
                //lets assume all locks will be acquired
                locksToAcquire = false;
                //first access the changeSet and begin to acquire locks
                ClassDescriptor descriptor = null;
                for (ObjectChangeSet objectChangeSet : changeSet.getAllChangeSets().values()) {
                    // No Need to acquire locks for invalidated objects.
                    if ((mergeManager.shouldMergeChangesIntoDistributedCache() && (objectChangeSet.getSynchronizationType() == ClassDescriptor.INVALIDATE_CHANGED_OBJECTS))
                            || objectChangeSet.getId() == null) {
                        //skip this process as we will be unable to acquire the correct cachekey anyway
                        //this is a new object with identity after write sequencing, ? huh, all objects must have an id by merge?
                        continue;
                    }
                    descriptor = objectChangeSet.getDescriptor();
                    // Maybe null for distributed merge, initialize it.
                    if (descriptor == null) {
                        descriptor = session.getDescriptor(objectChangeSet.getClassType(session));
                        objectChangeSet.setDescriptor(descriptor);
                    }
                    // PERF: Do not merge nor lock into the session cache if descriptor set to unit of work isolated.
                    if (descriptor.getCachePolicy().shouldIsolateObjectsInUnitOfWork()) {
                        continue;
                    }
                    AbstractSession targetSession = session.getParentIdentityMapSession(descriptor, true, true);
                    CacheKey activeCacheKey = attemptToAcquireLock(descriptor, objectChangeSet.getId(), targetSession);
                    if (activeCacheKey == null) {
                        // if cacheKey is null then the lock was not available no need to synchronize this block,because if the
                        // check fails then this thread will just return to the queue until it gets woken up.
                        if (this.prevailingQueue.getFirst() == mergeManager) {
                            // wait on this object until it is free,  or until wait time expires because
                            // this thread is the prevailing thread
                            // see bug 483478
                            activeCacheKey = waitOnObjectLock(descriptor, objectChangeSet.getId(),
                                    targetSession, (int) Math.round(((0.001d + Math.random()) * 500)));
                        }
                        if (activeCacheKey == null) {
                            // failed to acquire lock, release all acquired
                            // locks and place thread on waiting list
                            releaseAllAcquiredLocks(mergeManager);
                            // get cacheKey
                            activeCacheKey = targetSession.getIdentityMapAccessorInstance().getCacheKeyForObjectForLock(objectChangeSet.getId(), descriptor.getJavaClass(), descriptor);
                            if (session.shouldLog(SessionLog.FINER, SessionLog.CACHE)) {
                                Object[] params = new Object[3];
                                params[0] = descriptor.getJavaClass();
                                params[1] = objectChangeSet.getId();
                                params[2] = Thread.currentThread().getName();
                                session.log(SessionLog.FINER, SessionLog.CACHE, "dead_lock_encountered_on_write_no_cachekey", params, null);
                            }
                            if (mergeManager.getWriteLockQueued() == null) {
                                // thread is entering the wait queue for the
                                // first time
                                // set the QueueNode to be the node from the
                                // linked list for quick removal upon
                                // acquiring all locks
                                synchronized (this.prevailingQueue) {
                                    mergeManager.setQueueNode(this.prevailingQueue.addLast(mergeManager));
                                }
                            }

                            // set the cache key on the merge manager for
                            // the object that could not be acquired
                            mergeManager.setWriteLockQueued(objectChangeSet.getId());
                            try {
                                if (activeCacheKey != null){
                                    //wait on the lock of the object that we couldn't get.
                                    synchronized (activeCacheKey) {
                                        // verify that the cache key is still locked before we wait on it, as
                                        //it may have been released since we tried to acquire it.
                                        if (activeCacheKey.isAcquired() && (activeCacheKey.getActiveThread() != Thread.currentThread())) {
                                                Thread thread = activeCacheKey.getActiveThread();
                                                if (thread.isAlive()){
                                                    long time = System.currentTimeMillis();
                                                    activeCacheKey.wait(MAX_WAIT);
                                                    if (System.currentTimeMillis() - time >= MAX_WAIT){
                                                        Object[] params = new Object[]{MAX_WAIT /1000, descriptor.getJavaClassName(), activeCacheKey.getKey(), thread.getName()};
                                                        StringBuilder buffer = new StringBuilder(TraceLocalization.buildMessage("max_time_exceeded_for_acquirerequiredlocks_wait", params));
                                                        StackTraceElement[] trace = thread.getStackTrace();
                                                        for (StackTraceElement element : trace){
                                                            buffer.append("\t\tat");
                                                            buffer.append(element.toString());
                                                            buffer.append("\n");
                                                        }
                                                        session.log(SessionLog.SEVERE, SessionLog.CACHE, buffer.toString());
                                                        session.getIdentityMapAccessor().printIdentityMapLocks();
                                                    }
                                                }else{
                                                    session.log(SessionLog.SEVERE, SessionLog.CACHE, "releasing_invalid_lock", new Object[] { thread.getName(),descriptor.getJavaClass(), objectChangeSet.getId()});
                                                    //thread that held lock is no longer alive.  Something bad has happened like
                                                    while (activeCacheKey.isAcquired()){
                                                        // could have a depth greater than one.
                                                        activeCacheKey.release();
                                                    }
                                                }
                                            }
                                        }
                                    }
                            } catch (InterruptedException exception) {
                                throw org.eclipse.persistence.exceptions.ConcurrencyException.waitWasInterrupted(exception.getMessage());
                            }
                            // we want to record this information so that we have traceability over this sort of problems
                            addCacheKeyToMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(currentThread, activeCacheKey, timeWhenLocksToAcquireLoopStarted);
                            // failed to acquire, exit this loop to restart all over again.
                            locksToAcquire = true;
                            break;
                        }else{
                            removeCacheKeyFromMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(currentThread, activeCacheKey);
                            objectChangeSet.setActiveCacheKey(activeCacheKey);
                            mergeManager.getAcquiredLocks().add(activeCacheKey);
                        }
                    } else {
                        removeCacheKeyFromMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(currentThread, activeCacheKey);
                        objectChangeSet.setActiveCacheKey(activeCacheKey);
                        mergeManager.getAcquiredLocks().add(activeCacheKey);
                    }
                }
            }
        } catch (RuntimeException exception) {
            // if there was an exception then release.
            //must not release in a finally block as release only occurs in this method
            // if there is a problem or all of the locks can not be acquired.
            releaseAllAcquiredLocks(mergeManager);
            throw exception;
        } catch (InterruptedException exception) {
            releaseAllAcquiredLocks(mergeManager);
            throw ConcurrencyException.waitFailureOnClientSession(exception);
        } catch (Error error){
            releaseAllAcquiredLocks(mergeManager);
            mergeManager.getSession().logThrowable(SessionLog.SEVERE, SessionLog.TRANSACTION, error);
            throw error;
        }finally {
            if (mergeManager.getWriteLockQueued() != null) {
                //the merge manager entered the wait queue and must be cleaned up
                synchronized(this.prevailingQueue) {
                    this.prevailingQueue.remove(mergeManager.getQueueNode());
                }
                mergeManager.setWriteLockQueued(null);
            }
            clearMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(currentThread);
            clearMapThreadToObjectIdsWithChagenSet(currentThread);
        }
    }

    /**
     * INTERNAL:
     * This method will be called by a merging thread that is attempting to lock
     * a new object that was not locked previously.  Unlike the other methods
     * within this class this method will lock only this object.
     */
    public CacheKey appendLock(Object primaryKey, Object objectToLock, ClassDescriptor descriptor, MergeManager mergeManager, AbstractSession session) {
        CacheKey lockedCacheKey = session.getIdentityMapAccessorInstance().acquireLockNoWait(primaryKey, descriptor.getJavaClass(), false, descriptor);
        if (lockedCacheKey == null) {
            session.getIdentityMapAccessorInstance().getWriteLockManager().transitionToDeferredLocks(mergeManager);
            lockedCacheKey = session.getIdentityMapAccessorInstance().acquireDeferredLock(primaryKey, descriptor.getJavaClass(), descriptor, true);
            Object cachedObject = lockedCacheKey.getObject();
            if (cachedObject == null) {
                if (lockedCacheKey.getActiveThread() == Thread.currentThread()) {
                    lockedCacheKey.setObject(objectToLock);
                } else {
                    cachedObject = lockedCacheKey.waitForObject();
                }
            }
            lockedCacheKey.releaseDeferredLock();
            return lockedCacheKey;
        } else {
            if (lockedCacheKey.getObject() == null) {
                lockedCacheKey.setObject(objectToLock); // set the object in the
                // cachekey
                // for others to find an prevent cycles
            }
            if (mergeManager.isTransitionedToDeferredLocks()){
                ConcurrencyManager.getDeferredLockManager(Thread.currentThread()).getActiveLocks().add(lockedCacheKey);
            }else{
                 mergeManager.getAcquiredLocks().add(lockedCacheKey);
            }
            return lockedCacheKey;
        }
    }

    /**
     * INTERNAL:
     * This method performs the operations of finding the cacheKey and locking it if possible.
     * Returns True if the lock was acquired, false otherwise
     */
    protected CacheKey attemptToAcquireLock(ClassDescriptor descriptor, Object primaryKey, AbstractSession session) {
        return session.getIdentityMapAccessorInstance().acquireLockNoWait(primaryKey, descriptor.getJavaClass(), true, descriptor);
    }

    /**
     * INTERNAL:
     * Simply check that the object is not already locked then pass it on to the locking method
     */
    protected CacheKey checkAndLockObject(Object objectToLock, Map lockedObjects, Map refreshedObjects, DatabaseMapping mapping, AbstractSession cloningSession) {
        //the cachekey should always reference an object otherwise what would we be cloning.
        if ((objectToLock != null) && !lockedObjects.containsKey(objectToLock)) {
            Object primaryKeyToLock = null;
            ClassDescriptor referenceDescriptor = null;
            if (mapping.getReferenceDescriptor().hasInheritance() || mapping.getReferenceDescriptor().isDescriptorForInterface()) {
                referenceDescriptor = cloningSession.getDescriptor(objectToLock);
            } else {
                referenceDescriptor = mapping.getReferenceDescriptor();
            }
            // Need to traverse aggregates, but not lock aggregates directly.
            if (referenceDescriptor.isDescriptorTypeAggregate()) {
                traverseRelatedLocks(objectToLock, lockedObjects, refreshedObjects, referenceDescriptor, cloningSession);
            } else {
                primaryKeyToLock = referenceDescriptor.getObjectBuilder().extractPrimaryKeyFromObject(objectToLock, cloningSession);
                CacheKey cacheKey = cloningSession.getIdentityMapAccessorInstance().getCacheKeyForObjectForLock(primaryKeyToLock, objectToLock.getClass(), referenceDescriptor);
                if (cacheKey == null) {
                    // Cache key may be null for no-identity map, missing or deleted object, just create a new one to be locked.
                    cacheKey = new CacheKey(primaryKeyToLock);
                    cacheKey.setReadTime(System.currentTimeMillis());
                }
                CacheKey toWaitOn = acquireLockAndRelatedLocks(objectToLock, lockedObjects, refreshedObjects, cacheKey, referenceDescriptor, cloningSession);
                if (toWaitOn != null) {
                    return toWaitOn;
                }
            }
        }
        return null;
    }

    /**
     * INTERNAL:
     * This method will release all acquired locks
     */
    public void releaseAllAcquiredLocks(MergeManager mergeManager) {
        if (!MergeManager.LOCK_ON_MERGE) {//lockOnMerge is a backdoor and not public
            return;
        }
        List<CacheKey> acquiredLocks = mergeManager.getAcquiredLocks();
        Iterator<CacheKey> locks = acquiredLocks.iterator();
        RuntimeException exception = null;
        while (locks.hasNext()) {
            try {
                CacheKey cacheKeyToRemove = locks.next();
                if (cacheKeyToRemove.getObject() == null) {
                    cacheKeyToRemove.removeFromOwningMap();
                }
                if (mergeManager.isTransitionedToDeferredLocks()) {
                    cacheKeyToRemove.releaseDeferredLock();
                } else {
                    cacheKeyToRemove.release();
                }
            } catch (RuntimeException e){
                if (exception == null){
                    exception = e;
                }
            }
        }
        acquiredLocks.clear();
        if (exception != null){
            throw exception;
        }
    }

    /**
     * INTERNAL:
     * This method performs the operations of finding the cacheKey and locking it if possible.
     * Waits until the lock can be acquired
     */
    protected CacheKey waitOnObjectLock(ClassDescriptor descriptor, Object primaryKey, AbstractSession session, int waitTime) {
        return session.getIdentityMapAccessorInstance().acquireLockWithWait(primaryKey, descriptor.getJavaClass(), true, descriptor, waitTime);
    }

    // Helper data structures to have tracebility about object ids with change sets and cache keys we are sturggling to acquire

    /** Getter for {@link #THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS} */
    public static Map<Thread, Set<ConcurrencyManager>> getThreadToFailToAcquireCacheKeys() {
        return unmodifiableMap(THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS);
    }

    /** Getter for {@link #MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET} */
    public static Map<Thread, Set<Object>> getMapWriteLockManagerThreadToObjectIdsWithChangeSet() {
        return unmodifiableMap(MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET);
    }

    /**
     * Remove the current thread from the map of object ids with change sets that are about to bec ommited
     *
     * @param thread
     *            the thread that is clearing itself out of the map of change sets it needs to merge into the shared
     *            cache
     */
    public static void clearMapThreadToObjectIdsWithChagenSet(Thread thread) {
        MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET.remove(thread);
    }
    /**
     * Before a thread starts long wait loop to acquire write locks during a commit transaction the thread will record
     * in this map the object ids it holds with chance sets. It will be useful information if a dead lock is taking
     * place.
     *
     * @param thread
     *            the thread that is in the middle of merge to the shared cache trying to acquire write locks to do this
     *            merge
     * @param objectChangeSets
     *            the object change sets it has in its hands and that it would like to merge into the cache
     */
    public static void populateMapThreadToObjectIdsWithChagenSet(Thread thread,
                                                                 Collection<ObjectChangeSet> objectChangeSets) {
        // (a) make sure the map has an entry for the the thread
        boolean hasKey = MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET.containsKey(thread);
        if (!hasKey) {
            MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET.putIfAbsent(thread,
                    ConcurrentHashMap.newKeySet());
        }

        // (b) The ids of the objects with change sets
        Set<Object> primarykeys = MAP_WRITE_LOCK_MANAGER_THREAD_TO_OBJECT_IDS_WITH_CHANGE_SET.get(thread);
        primarykeys.clear();
        for (ObjectChangeSet objectChangeSet : objectChangeSets) {
            Object primaryKey = objectChangeSet.getId();
            primarykeys.add(primaryKey);
        }
    }

    /**
     * Before the problematic while loop starts we should always clear for this thread the set of cache keys it could
     * not acquire.
     *
     * @param thread
     *            the thread that what clear his set of cache keys it is struggling to acquire.
     */
    public static void clearMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(Thread thread) {
        THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS.remove(thread);
    }

    /**
     * The thread was doing its while loop to acquire all required locks to proceed with the commmit and it realized
     * there was one cache key it is unable to acquire
     *
     * @param thread
     *            thread the thread working on updating the shared cache
     * @param cacheKeyThatCouldNotBeAcquired
     *            the cache key it is not managing to acquire
     * @throws InterruptedException
     *             Should be fired because we are passing a flag into the
     *             determineIfReleaseDeferredLockAppearsToBeDeadLocked to say we do not want the thread to be blown up
     *             (e.g. we are afraid of breaking threads in the middle of a commit process could be quite dangerous).
     *             See
     *             {@link #ALLOW_INTERRUPTED_EXCEPTION_TO_BE_FIRED_UP_FALSE}
     */
    public static void addCacheKeyToMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(Thread thread, ConcurrencyManager cacheKeyThatCouldNotBeAcquired, long whileStartDate) throws InterruptedException {

        // sanity check, make sure the cacheKeyThatCouldNotBeAcquired is not null
        // should never happen because when the write lock manager fails to acquire the cache key both with acquire no
        // wait and acquire with wait
        // then the code will just grab the cache key fro loggging puprposes using the
        // see the code getCacheKeyForObjectForLock
        // this is why we believe this is never null. But the sanity check does not hurt us.
        if (cacheKeyThatCouldNotBeAcquired == null) {
            return;
        }

        // (b) add the cache key to the set if absent
        Set<ConcurrencyManager> cacheKeysWeAreHavingDifficultyAcquiring = getCacheKeysThatCouldNotBeAcquiredByThread(
                thread);
        if(!cacheKeysWeAreHavingDifficultyAcquiring.contains(cacheKeyThatCouldNotBeAcquired)) {
            cacheKeysWeAreHavingDifficultyAcquiring.add(cacheKeyThatCouldNotBeAcquired);
        }

        // (c) If a write lock fails to be acquired and goes into the basked of cache keys that could not be acquired
        // it could be an indication this thread is stuck for a long while
        // NOTE:
        // it might be best to not even give the possibility for an exception to be fired
        // for code that is in the lock manager
        final Thread currentThread = Thread.currentThread();
        DeferredLockManager lockManager = ConcurrencyManager.getDeferredLockManager(currentThread);
        ReadLockManager readLockManager = ConcurrencyManager.getReadLockManager(currentThread);
        ConcurrencyUtil.SINGLETON.determineIfReleaseDeferredLockAppearsToBeDeadLocked(
                cacheKeyThatCouldNotBeAcquired, whileStartDate, lockManager, readLockManager,
                ALLOW_INTERRUPTED_EXCEPTION_TO_BE_FIRED_UP_FALSE);
    }

    /**
     * A cache keys was successfully acquired we want to make sure it is not recorded in the map of cache keys that
     * could not be acquired. The situation theoretically can change. Failing to acquire a write lock can be a temporary
     * situation. The lock might become available eventually. Otherwise there would be no point for the while loop that
     * is trying to acquire these locks.
     *
     * @param thread
     *            the thread that just managed to grab a write lock
     * @param cacheKeyThatCouldNotBeAcquired
     *            the cache key it managed to acquire for writing.
     */
    public static void removeCacheKeyFromMapWriteLockManagerToCacheKeysThatCouldNotBeAcquired(Thread thread,
                                                                                              ConcurrencyManager cacheKeyThatCouldNotBeAcquired) {
        Set<ConcurrencyManager> cacheKeysWeAreHavingDifficultyAcquiring = getCacheKeysThatCouldNotBeAcquiredByThread(
                thread);
        cacheKeysWeAreHavingDifficultyAcquiring.remove(cacheKeyThatCouldNotBeAcquired);
    }

    /**
     * If the thread is not yet registered in the map it will get registered with an empty map.
     *
     * @param thread
     *            the thread that wants to get its set of cache keys it is not managing to acquire.
     * @return the set of cache keys the thrad is struggling to acquire
     */
    private static Set<ConcurrencyManager> getCacheKeysThatCouldNotBeAcquiredByThread(Thread thread) {
        // (a) make sure the map has an entry for the the thread
        boolean hasKey = THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS.containsKey(thread);
        if (!hasKey) {
            Set<ConcurrencyManager> cacheKeySet = ConcurrentHashMap.newKeySet();
            THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS.putIfAbsent(thread, cacheKeySet);
        }

        // (b) We are certain the map is not empty anymore return the set
        return THREAD_TO_FAIL_TO_ACQUIRE_CACHE_KEYS.get(thread);
    }

}