| /******************************************************************************* | |
| * Copyright (c) 1998, 2013 Oracle and/or its affiliates. All rights reserved. | |
| * This program and the accompanying materials are made available under the | |
| * terms of the Eclipse Public License v1.0 and Eclipse Distribution License v. 1.0 | |
| * which accompanies this distribution. | |
| * The Eclipse Public License is available at http://www.eclipse.org/legal/epl-v10.html | |
| * and the Eclipse Distribution License is available at | |
| * http://www.eclipse.org/org/documents/edl-v10.php. | |
| * | |
| * 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 | |
| ******************************************************************************/ | |
| package org.eclipse.persistence.internal.helper; | |
| import java.util.*; | |
| import org.eclipse.persistence.descriptors.ClassDescriptor; | |
| import org.eclipse.persistence.descriptors.FetchGroupManager; | |
| import org.eclipse.persistence.exceptions.ConcurrencyException; | |
| import org.eclipse.persistence.internal.queries.ContainerPolicy; | |
| import org.eclipse.persistence.mappings.DatabaseMapping; | |
| import org.eclipse.persistence.internal.sessions.*; | |
| import org.eclipse.persistence.internal.identitymaps.*; | |
| import org.eclipse.persistence.internal.localization.TraceLocalization; | |
| import org.eclipse.persistence.internal.helper.linkedlist.*; | |
| import org.eclipse.persistence.logging.SessionLog; | |
| /** | |
| * 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 { | |
| // this will allow us to prevent a readlock thread from looping forever. | |
| public static int MAXTRIES = 10000; | |
| public static 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) { | |
| boolean successful = false; | |
| IdentityHashMap lockedObjects = new IdentityHashMap(); | |
| IdentityHashMap refreshedObjects = new IdentityHashMap(); | |
| 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(); | |
| } | |
| synchronized (toWaitOn) { | |
| try { | |
| if (toWaitOn.isAcquired()) {//last minute check to insure it is still locked. | |
| toWaitOn.wait();// 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 | |
| } finally { | |
| 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 mappings = descriptor.getLockableMappings().iterator(); | |
| mappings.hasNext();) { | |
| DatabaseMapping mapping = (DatabaseMapping)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 | |
| */ | |
| public void acquireRequiredLocks(MergeManager mergeManager, UnitOfWorkChangeSet changeSet) { | |
| if (!MergeManager.LOCK_ON_MERGE) {//lockOnMerge is a backdoor and not public | |
| return; | |
| } | |
| boolean locksToAcquire = true; | |
| //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 = ((UnitOfWorkImpl)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 | |
| activeCacheKey = waitOnObjectLock(descriptor, objectChangeSet.getId(), targetSession, (int)Math.round((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()); | |
| } | |
| // failed to acquire, exit this loop to restart all over again. | |
| locksToAcquire = true; | |
| break; | |
| }else{ | |
| objectChangeSet.setActiveCacheKey(activeCacheKey); | |
| mergeManager.getAcquiredLocks().add(activeCacheKey); | |
| } | |
| } else { | |
| 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 (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); | |
| } | |
| } | |
| } | |
| /** | |
| * 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()){ | |
| lockedCacheKey.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 acquiredLocks = mergeManager.getAcquiredLocks(); | |
| Iterator locks = acquiredLocks.iterator(); | |
| RuntimeException exception = null; | |
| while (locks.hasNext()) { | |
| try { | |
| CacheKey cacheKeyToRemove = (CacheKey) 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); | |
| } | |
| } |