Google Git
Sign in
apache / geode / b60806c49bdbd1b32dc77ab5ffbb4d2d64cee54a / . / geode-core / src / main / java / org / apache / geode / distributed / internal / locks / DLockService.java
blob: f453f61b1401c7f5efb5891eef3a0d82709d1b36 [file] [log] [blame]
/*
* 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.geode.distributed.internal.locks;
import java.io.DataInput;
import java.io.DataOutput;
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 java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LeaseExpiredException;
import org.apache.geode.distributed.LockNotHeldException;
import org.apache.geode.distributed.LockServiceDestroyedException;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.deadlock.UnsafeThreadLocal;
import org.apache.geode.distributed.internal.locks.DLockQueryProcessor.DLockQueryReplyMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.DataSerializableFixedID;
import org.apache.geode.internal.OSProcess;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.internal.util.concurrent.FutureResult;
/**
* Implements the distributed locking service with distributed lock grantors.
*
*/
public class DLockService extends DistributedLockService {
private static final Logger logger = LogService.getLogger();
private static final long NOT_GRANTOR_SLEEP = Long
.getLong(DistributionConfig.GEMFIRE_PREFIX + "DLockService.notGrantorSleep", 100).longValue();
private static final boolean DEBUG_NONGRANTOR_DESTROY_LOOP = Boolean
.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "DLockService.debug.nonGrantorDestroyLoop");
private static final int DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT = Integer
.getInteger(
DistributionConfig.GEMFIRE_PREFIX + "DLockService.debug.nonGrantorDestroyLoopCount", 20)
.intValue();
private static final boolean AUTOMATE_FREE_RESOURCES =
Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "DLockService.automateFreeResources");
static final int INVALID_LEASE_ID = -1;
/** Unique name for this instance of the named locking service */
protected final String serviceName;
/** DistributionManager for this member */
private final DistributionManager dm;
/**
* DistributedSystem connection for this member (used for DisconnectListener, logging, etc)
*/
protected final InternalDistributedSystem ds;
/** Known lock tokens for this service. Key:Object(name), Value:DLockToken */
private final Map<Object, DLockToken> tokens = new HashMap<Object, DLockToken>();
/**
* True if this member has destroyed this named locking service. Field is volatile only because
* it's referenced in {@link #toString()} (never synchronize in <code>toString</code>).
*/
private volatile boolean destroyed = false;
/**
* True if this is a distributed lock service; false if local to this vm only. TX has a "local"
* lock service which sets this to false.
*/
private final boolean isDistributed;
/** Optional handler for departure of lease holders; used by grantor */
private DLockLessorDepartureHandler lessorDepartureHandler;
/**
* Hook for transactions which allows custom behavior in processing DLockRecoverGrantorMessage
*/
private DLockRecoverGrantorProcessor.MessageProcessor recoverGrantorProcessor;
/** Thread-safe reference to DistributedLockStats */
private final DistributedLockStats dlockStats;
/**
* Protects {@link #lockGrantorId}, {@link #grantor} and {@link #lockGrantorFutureResult}. Final
* granting of a lock occurs under this synchronization and only if <code>lockGrantorId</code>
* matches the grantor that granted the lock.
*/
private final Object lockGrantorIdLock = new Object();
/** Identifies the current grantor for this lock service. */
private LockGrantorId lockGrantorId;
/**
* Local instance of the lock grantor if this process is the grantor. This field is volatile for
* one use: 1) {@link #toString()} which should not use synchronization due to potential for wrong
* lock ordering. Can we make this non-volatile??
*/
private volatile DLockGrantor grantor;
/**
* Count of currently active locks and lock requests. Used to determine if destroy must tell the
* grantor to release all held locks.
*/
private int activeLocks = 0;
/** True if this service should be destroyed in system DisconnectListener */
private final boolean destroyOnDisconnect;
/** True if this service should automatically freeResources */
private final boolean automateFreeResources;
/** Identifies the thread that is destroying this lock service. */
private final ThreadLocal<Boolean> destroyingThread = new ThreadLocal<Boolean>();
/// ** Held during destory and creation of this lock service. */
// private final Object serviceLock = new Object();
/** Protects access to {@link #destroyed} and {@link #activeLocks}. */
private final Object destroyLock = new Object();
/**
* Created by the thread communicating directly with the elder. Other threads will wait on this
* and then use the resulting lockGrantorId. This ensures that only one message is sent to the
* elder and that only one thread does so at a time. Protected by {@link #lockGrantorIdLock} and
* holds a reference to a {@link LockGrantorId}.
* <p>
* Only outbound threads and operations should ever wait on this. Do NOT allow inbound threads to
* use the <code>lockGrantorFutureResult</code>.
*/
private FutureResult lockGrantorFutureResult;
private final DLockStopper stopper;
// -------------------------------------------------------------------------
// State and concurrency construct methods
// -------------------------------------------------------------------------
public boolean isDestroyed() {
synchronized (this.destroyLock) {
if (this.destroyed) {
if (!isCurrentThreadDoingDestroy()) {
return true;
}
}
return false;
}
}
public void checkDestroyed() {
if (isDestroyed()) {
throw generateLockServiceDestroyedException(generateLockServiceDestroyedMessage());
}
}
/**
* Create a new LockServiceDestroyedException for this lock service.
*
* @param message the detail message that explains the exception
* @return new LockServiceDestroyedException
*/
protected LockServiceDestroyedException generateLockServiceDestroyedException(String message) {
return new LockServiceDestroyedException(message);
}
/**
* Returns the string message to use in a LockServiceDestroyedException for this lock service.
*
* @return the detail message that explains LockServiceDestroyedException
*/
protected String generateLockServiceDestroyedMessage() {
return String.format("%s has been destroyed", this);
}
/**
* Returns true if {@link #lockGrantorId} is the same as the specified LockGrantorId. Caller must
* synchronize on {@link #lockGrantorIdLock}.
*
* @param someLockGrantorId the LockGrantorId to check
*/
private boolean checkLockGrantorId(LockGrantorId someLockGrantorId) {
Assert.assertHoldsLock(this.lockGrantorIdLock, true);
if (this.lockGrantorId == null) {
return false;
}
return this.lockGrantorId.sameAs(someLockGrantorId);
}
/**
* Returns true if lockGrantorId is the same as the specified LockGrantorId. Caller must
* synchronize on lockGrantorIdLock.
*
* @param someLockGrantorId the LockGrantorId to check
*/
public boolean isLockGrantorId(LockGrantorId someLockGrantorId) {
synchronized (this.lockGrantorIdLock) {
return checkLockGrantorId(someLockGrantorId);
}
}
private boolean isCurrentThreadDoingDestroy() {
return Boolean.TRUE.equals(this.destroyingThread.get());
}
private void setDestroyingThread() {
this.destroyingThread.set(Boolean.TRUE);
}
private void clearDestroyingThread() {
this.destroyingThread.remove();
}
private InternalDistributedMember getElderId() {
InternalDistributedMember elder = this.dm.getElderId();
if (elder == null) {
this.dm.getSystem().getCancelCriterion().checkCancelInProgress(null);
}
Assert.assertTrue(elder != null);
return elder;
}
/**
* Returns id of the current lock grantor for this service. If necessary, a request will be sent
* to the elder to fetch this information.
*/
public LockGrantorId getLockGrantorId() {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
boolean ownLockGrantorFutureResult = false;
FutureResult lockGrantorFutureResultRef = null;
long statStart = -1;
LockGrantorId theLockGrantorId = null;
while (theLockGrantorId == null) {
ownLockGrantorFutureResult = false;
try {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
if (this.lockGrantorFutureResult != null) {
lockGrantorFutureResultRef = this.lockGrantorFutureResult;
} else if (this.lockGrantorId != null) {
return this.lockGrantorId;
} else {
ownLockGrantorFutureResult = true;
lockGrantorFutureResultRef = new FutureResult(this.dm.getCancelCriterion());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[getLockGrantorId] creating lockGrantorFutureResult");
}
this.lockGrantorFutureResult = lockGrantorFutureResultRef;
}
}
statStart = getStats().startGrantorWait();
if (!ownLockGrantorFutureResult) {
LockGrantorId lockGrantorIdRef =
waitForLockGrantorFutureResult(lockGrantorFutureResultRef, 0, TimeUnit.MILLISECONDS);
if (lockGrantorIdRef != null) {
return lockGrantorIdRef;
} else {
continue;
}
}
InternalDistributedMember elder = getElderId();
Assert.assertTrue(elder != null);
GrantorInfo gi = getGrantorRequest();
theLockGrantorId =
new LockGrantorId(this.dm, gi.getId(), gi.getVersionId(), gi.getSerialNumber());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[getLockGrantorId] elder says grantor is {}",
theLockGrantorId);
}
// elder tells us to be the grantor...
if (theLockGrantorId.isLocal(getSerialNumber())) {
boolean needsRecovery = gi.needsRecovery();
if (!needsRecovery) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[getLockGrantorId] needsRecovery is false");
}
synchronized (this.lockGrantorIdLock) {
// either no previous grantor or grantor is newer
Assert.assertTrue(
this.lockGrantorId == null || this.lockGrantorId.isNewerThan(theLockGrantorId)
|| this.lockGrantorId.sameAs(theLockGrantorId),
this.lockGrantorId + " should be null or newer than or same as "
+ theLockGrantorId);
}
}
if (!createLocalGrantor(elder, needsRecovery, theLockGrantorId)) {
theLockGrantorId = this.lockGrantorId;
}
}
// elder says another member is the grantor
else {
synchronized (this.lockGrantorIdLock) {
if (!setLockGrantorId(theLockGrantorId)) {
theLockGrantorId = this.lockGrantorId;
}
}
}
} finally {
synchronized (this.lockGrantorIdLock) {
boolean getLockGrantorIdFailed = theLockGrantorId == null;
if (statStart > -1) {
getStats().endGrantorWait(statStart, getLockGrantorIdFailed);
}
if (ownLockGrantorFutureResult) {
// this thread is doing the real work and must finish the future
Assert.assertTrue(this.lockGrantorFutureResult == lockGrantorFutureResultRef);
if (getLockGrantorIdFailed) {
// failed so cancel lockGrantorFutureResult
lockGrantorFutureResultRef.cancel(false);
} else {
// succeeded so set lockGrantorFutureResult
lockGrantorFutureResultRef.set(theLockGrantorId);
}
// null out the reference so it is free for next usage
this.lockGrantorFutureResult = null;
}
}
} // finally block for lockGrantorFutureResult
} // while theLockGrantorId == null
return theLockGrantorId;
}
/**
* Creates a local {@link DLockGrantor}.
*
* if (!createLocalGrantor(xxx)) { theLockGrantorId = this.lockGrantorId; }
*
* @param elder the elder that told us to be the grantor
* @param needsRecovery true if recovery is required
* @param myLockGrantorId lockGrantorId to use
* @return true if successfully created local grantor; false if aborted
*/
private boolean createLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
LockGrantorId myLockGrantorId) {
DLockGrantor myGrantor =
DLockGrantor.createGrantor(this, myLockGrantorId.getLockGrantorVersion());
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[createLocalGrantor] Calling makeLocalGrantor");
}
return makeLocalGrantor(elder, needsRecovery, myLockGrantorId, myGrantor);
}
private boolean makeLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
LockGrantorId myLockGrantorId, DLockGrantor myGrantor) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
boolean success = false;
try {
synchronized (this.lockGrantorIdLock) {
if (isDestroyed()) {
checkDestroyed(); // exit
}
InternalDistributedMember currentElder = getElderId();
if (!currentElder.equals(elder)) {
// abort because elder changed
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"Failed to create {} because elder changed from {} to {}", myLockGrantorId, elder,
currentElder);
}
return false; // exit
}
if (this.deposingLockGrantorId != null) {
if (this.deposingLockGrantorId.isNewerThan(myLockGrantorId)) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Failed to create {} because I was deposed by {}",
myLockGrantorId, this.deposingLockGrantorId);
}
this.deposingLockGrantorId = null;
return false; // exit
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{} failed to depose {}",
this.deposingLockGrantorId, myLockGrantorId);
}
// older grantor couldn't depose us, so null it out...
this.deposingLockGrantorId = null;
}
if (!setLockGrantorId(myLockGrantorId, myGrantor)) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[getLockGrantorId] failed to create {} because current grantor is {}",
myLockGrantorId, this.lockGrantorId);
}
return false; // exit
}
} // release sync on this.lockGrantorIdLock
// do NOT sync while doing recovery (because it waits for replies)
if (needsRecovery) {
boolean recovered =
DLockRecoverGrantorProcessor.recoverLockGrantor(this.dm.getDistributionManagerIds(), // include
// this
// vm
this, // this lock service
myGrantor, this.dm, elder); // the elder that told us to be the grantor
if (!recovered) {
checkDestroyed();
return false; // exit
}
}
// after recovery, resynchronize on lockGrantorIdLock again
// check to see if myLockGrantorId has been deposed
synchronized (this.lockGrantorIdLock) {
if (isDestroyed()) {
checkDestroyed(); // exit
}
if (this.deposingLockGrantorId != null) {
if (this.deposingLockGrantorId.isNewerThan(myLockGrantorId)) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Failed to create {} because I was deposed by {}",
myLockGrantorId, this.deposingLockGrantorId);
}
this.deposingLockGrantorId = null;
return false; // exit
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{} failed to depose {}",
this.deposingLockGrantorId, myLockGrantorId);
}
this.deposingLockGrantorId = null;
}
if (checkLockGrantorId(myLockGrantorId)) {
success = myGrantor.makeReady(true); // do not enforce initializing
}
}
return success; // exit
} catch (VirtualMachineError err) {
SystemFailure.initiateFailure(err);
// If this ever returns, rethrow the error. We're poisoned
// now, so don't let this thread continue.
throw err;
} catch (Error e) {
// Whenever you catch Error or Throwable, you must also
// catch VirtualMachineError (see above). However, there is
// _still_ a possibility that you are dealing with a cascading
// error condition, so you also need to check to see if the JVM
// is still usable:
SystemFailure.checkFailure();
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] throwing Error", e);
}
throw e;
} catch (RuntimeException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] throwing RuntimeException", e);
}
throw e;
} finally {
try {
// abort if unsuccessful or if lock service was destroyed
if (!success || isDestroyed()) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] aborting {} and {}",
myLockGrantorId, myGrantor);
}
nullLockGrantorId(myLockGrantorId);
if (!myGrantor.isDestroyed()) {
myGrantor.destroy();
}
}
} finally {
// assertion: grantor should now be either ready or destroyed!
if (myGrantor.isInitializing() && !dm.getCancelCriterion().isCancelInProgress()) {
logger.error(LogMarker.DLS_MARKER,
"Grantor is still initializing");
}
if (!success && !myGrantor.isDestroyed() && !dm.getCancelCriterion().isCancelInProgress()) {
logger.error(LogMarker.DLS_MARKER,
"Grantor creation was aborted but grantor was not destroyed");
}
}
}
}
/**
* Set {@link #lockGrantorId} to the given new value if the current value is null or is an older
* grantor version. Caller must hold {@link #lockGrantorIdLock}.
*
* @param newLockGrantorId the new value for lockGrantorId
*/
private boolean setLockGrantorId(LockGrantorId newLockGrantorId) {
Assert.assertHoldsLock(this.lockGrantorIdLock, true);
if (equalsLockGrantorId(newLockGrantorId)) {
return true;
} else if (!newLockGrantorId.hasLockGrantorVersion()) {
// proceed with temporary placeholder used by become grantor
this.lockGrantorId = newLockGrantorId;
return true;
} else if (newLockGrantorId.isRemote() && this.lockGrantorId != null
&& this.lockGrantorId.hasLockGrantorVersion()) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[setLockGrantorId] tried to replace {} with {}",
this.lockGrantorId, newLockGrantorId);
}
return false;
} else if (newLockGrantorId.isNewerThan(this.lockGrantorId)) {
this.lockGrantorId = newLockGrantorId;
return true;
} else {
return false;
}
}
/**
* Set {@link #lockGrantorId} to the <code>localLockGrantorId</code> if current value is null or
* is an older grantor version. This also atomically sets {@link #grantor} to ensure that the two
* fields are kept in sync. Caller must hold {@link #lockGrantorIdLock}.
*
* @param localLockGrantorId the new value for lockGrantorId
* @param localGrantor the new local intance of DLockGrantor
*/
private boolean setLockGrantorId(LockGrantorId localLockGrantorId, DLockGrantor localGrantor) {
Assert.assertHoldsLock(this.lockGrantorIdLock, true);
Assert.assertTrue(localLockGrantorId.isLocal(getSerialNumber()));
if (setLockGrantorId(localLockGrantorId)) {
this.grantor = localGrantor;
return true;
}
return false;
}
private LockGrantorId deposingLockGrantorId;
/**
* Deposes {@link #lockGrantorId} if <code>newLockGrantorId</code> is newer.
*
* @param newLockGrantorId the new lock grantor
*/
void deposeOlderLockGrantorId(LockGrantorId newLockGrantorId) {
LockGrantorId deposedLockGrantorId = null;
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
synchronized (this.lockGrantorIdLock) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[deposeOlderLockGrantorId] pre-deposing {} for new {}",
deposedLockGrantorId, newLockGrantorId);
}
this.deposingLockGrantorId = newLockGrantorId;
deposedLockGrantorId = this.lockGrantorId;
}
if (deposedLockGrantorId != null && deposedLockGrantorId.hasLockGrantorVersion()
&& newLockGrantorId.isNewerThan(deposedLockGrantorId)) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[deposeOlderLockGrantorId] post-deposing {} for new {}", deposedLockGrantorId,
newLockGrantorId);
}
nullLockGrantorId(deposedLockGrantorId);
}
}
/**
* Sets {@link #lockGrantorId} to null if the current value equals the expected old value. Caller
* must hold {@link #lockGrantorIdLock}.
*
* @param oldLockGrantorId the expected old value
* @return true if lockGrantorId was set to null
*/
private boolean nullLockGrantorId(LockGrantorId oldLockGrantorId) {
Assert.assertHoldsLock(this.destroyLock, false);
Assert.assertHoldsLock(this.lockGrantorIdLock, false);
if (oldLockGrantorId == null) {
return false;
}
DLockGrantor grantorToDestroy = null;
try {
synchronized (this.lockGrantorIdLock) {
if (equalsLockGrantorId(oldLockGrantorId)
|| (oldLockGrantorId.isLocal(getSerialNumber()) && isMakingLockGrantor())) {
// this.lockGrantorId != null && this.lockGrantorId.isLocal())) {
if (oldLockGrantorId.isLocal(getSerialNumber())
&& isLockGrantorVersion(this.grantor, oldLockGrantorId.getLockGrantorVersion())) {
// need to destroy and remove grantor
grantorToDestroy = this.grantor;
this.grantor = null;
}
this.lockGrantorId = null;
return true;
} else {
return false;
}
}
} finally {
if (grantorToDestroy != null) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[nullLockGrantorId] destroying {}",
grantorToDestroy);
}
grantorToDestroy.destroy();
}
}
}
/**
* Returns true if the grantor version of <code>dlockGrantor</code> equals the
* <code>grantorVersion</code>.
*
* @param dlockGrantor the grantor instance to compare to grantorVersion
* @param grantorVersion the grantor version number
* @return true if dlockGrantor is the same grantor version
*/
private boolean isLockGrantorVersion(DLockGrantor dlockGrantor, long grantorVersion) {
if (dlockGrantor == null) {
return false;
}
return dlockGrantor.getVersionId() == grantorVersion;
}
/**
* Returns true if <code>someLockGrantor</code> equals the current {@link #lockGrantorId}.
*
* @return true if someLockGrantor equals the current lockGrantorId
*/
private boolean equalsLockGrantorId(LockGrantorId someLockGrantor) {
Assert.assertHoldsLock(this.lockGrantorIdLock, true);
if (someLockGrantor == null) {
return this.lockGrantorId == null;
}
return someLockGrantor.equals(this.lockGrantorId);
}
/**
* Returns id of the current lock grantor for this service. If necessary, a request will be sent
* to the elder to fetch this information. Unlike getLockGrantorId this call will not become the
* lock grantor.
*/
public LockGrantorId peekLockGrantorId() {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
LockGrantorId currentLockGrantorId = this.lockGrantorId;
if (currentLockGrantorId != null) {
return currentLockGrantorId;
}
}
long statStart = getStats().startGrantorWait();
LockGrantorId theLockGrantorId = null;
try {
// 1st thread wins the right to request grantor info from elder
GrantorInfo gi = peekGrantor();
InternalDistributedMember lockGrantorMember = gi.getId();
if (lockGrantorMember == null) {
return null;
}
theLockGrantorId =
new LockGrantorId(this.dm, lockGrantorMember, gi.getVersionId(), gi.getSerialNumber());
return theLockGrantorId;
} finally {
boolean getLockGrantorIdFailed = theLockGrantorId == null;
getStats().endGrantorWait(statStart, getLockGrantorIdFailed);
}
}
/**
* Increments {@link #activeLocks} while synchronized on {@link #destroyLock} after calling
* {@link #checkDestroyed()}.
*/
private void incActiveLocks() {
synchronized (this.destroyLock) {
checkDestroyed();
this.activeLocks++;
}
}
/**
* Decrements {@link #activeLocks} while synchronized on {@link #destroyLock}.
*/
private void decActiveLocks() {
synchronized (this.destroyLock) {
this.activeLocks--;
}
}
/**
* Returns lockGrantorId when lockGrantorFutureResultRef has been set by another thread.
*
* @param lockGrantorFutureResultRef FutureResult to wait for
* @param timeToWait how many ms to wait, 0 = forever
* @param timeUnit the unit of measure for timeToWait
* @return the LockGrantorId or null if FutureResult was cancelled
*/
private LockGrantorId waitForLockGrantorFutureResult(FutureResult lockGrantorFutureResultRef,
long timeToWait, final TimeUnit timeUnit) {
LockGrantorId lockGrantorIdRef = null;
while (lockGrantorIdRef == null) {
boolean interrupted = Thread.interrupted();
try {
checkDestroyed();
if (timeToWait == 0) {
lockGrantorIdRef = (LockGrantorId) lockGrantorFutureResultRef.get();
} else {
lockGrantorIdRef = (LockGrantorId) lockGrantorFutureResultRef.get(timeToWait, timeUnit);
}
} catch (TimeoutException e) {
break;
} catch (InterruptedException e) {
interrupted = true;
this.dm.getCancelCriterion().checkCancelInProgress(e);
if (lockGrantorFutureResultRef.isCancelled()) {
// cancelled Future might throw InterruptedException...?
checkDestroyed();
break; // return null
}
continue;
} catch (CancellationException e) { // Future was cancelled
checkDestroyed();
break; // return null
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
return lockGrantorIdRef;
}
/**
* nulls out grantor to force call to elder
*
* @param timeToWait how long to wait for a new grantor. -1 don't wait, 0 no time limit
* @param timeUnit the unit of measure of timeToWait
*/
private void notLockGrantorId(LockGrantorId notLockGrantorId, long timeToWait,
final TimeUnit timeUnit) {
if (notLockGrantorId.isLocal(getSerialNumber())) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"notLockGrantorId {} returning early because notGrantor {} was equal to the local dm {}",
this.serviceName, notLockGrantorId, this.dm.getId());
}
// Let the local destroy or processing of transfer do the clear
return;
}
boolean ownLockGrantorFutureResult = false;
FutureResult lockGrantorFutureResultRef = null;
long statStart = -1;
LockGrantorId currentLockGrantorId = null;
try {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
currentLockGrantorId = this.lockGrantorId;
if (this.lockGrantorFutureResult != null) {
// some other thread is talking to elder
lockGrantorFutureResultRef = this.lockGrantorFutureResult;
} else if (!notLockGrantorId.sameAs(currentLockGrantorId)) {
return;
} else {
// this thread needs to talk to elder
ownLockGrantorFutureResult = true;
lockGrantorFutureResultRef = new FutureResult(this.dm.getCancelCriterion());
this.lockGrantorFutureResult = lockGrantorFutureResultRef;
}
}
statStart = getStats().startGrantorWait();
if (!ownLockGrantorFutureResult) {
if (timeToWait >= 0) {
waitForLockGrantorFutureResult(lockGrantorFutureResultRef, timeToWait, timeUnit);
}
return;
}
InternalDistributedMember elder = getElderId();
Assert.assertTrue(elder != null);
LockGrantorId elderLockGrantorId = null;
GrantorInfo gi = peekGrantor();
if (gi.getId() != null) {
elderLockGrantorId =
new LockGrantorId(this.dm, gi.getId(), gi.getVersionId(), gi.getSerialNumber());
}
if (notLockGrantorId.sameAs(elderLockGrantorId)) {
// elder says that notLockGrantorId is still the grantor...
sleep(NOT_GRANTOR_SLEEP);
return;
} else {
// elder says another member is the grantor
nullLockGrantorId(notLockGrantorId);
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"notLockGrantorId cleared lockGrantorId for service {}", this.serviceName);
}
}
} finally {
synchronized (this.lockGrantorIdLock) {
if (statStart > -1) {
getStats().endGrantorWait(statStart, false);
}
if (ownLockGrantorFutureResult) {
// this thread is doing the real work and must finish the future
Assert.assertTrue(this.lockGrantorFutureResult == lockGrantorFutureResultRef);
// cancel lockGrantorFutureResult
lockGrantorFutureResultRef.cancel(false);
// null out the reference so it is free for next usage
this.lockGrantorFutureResult = null;
}
}
} // finally block for lockGrantorFutureResult
}
/**
* All calls to GrantorRequestProcessor.clearGrantor must come through this synchronization point.
* <p>
* This fixes a deadlock between this.becomeGrantorMonitor and DistributionManager.elderLock
* <p>
* All calls to the elder may result in elder recovery which may call back into dlock and acquire
* synchronization on this.becomeGrantorMonitor.
*/
void clearGrantor(long grantorVersion, boolean withLocks) {
GrantorRequestProcessor.clearGrantor(grantorVersion, this, getSerialNumber(), this.ds,
withLocks);
}
/**
* All calls to GrantorRequestProcessor.getGrantor must come through this synchronization point.
* <p>
* This fixes a deadlock between this.becomeGrantorMonitor and DistributionManager.elderLock
* <p>
* All calls to the elder may result in elder recovery which may call back into dlock and acquire
* synchronization on this.becomeGrantorMonitor.
*/
private GrantorInfo getGrantorRequest() {
return GrantorRequestProcessor.getGrantor(this, getSerialNumber(), this.ds);
}
/**
* All calls to GrantorRequestProcessor.peekGrantor must come through this synchronization point.
* <p>
* This fixes a deadlock between this.becomeGrantorMonitor and DistributionManager.elderLock
* <p>
* All calls to the elder may result in elder recovery which may call back into dlock and acquire
* synchronization on this.becomeGrantorMonitor.
*/
private GrantorInfo peekGrantor() {
return GrantorRequestProcessor.peekGrantor(this, this.ds);
}
/**
* All calls to GrantorRequestProcessor.becomeGrantor must come through this synchronization
* point.
* <p>
* This fixes a deadlock between this.becomeGrantorMonitor and DistributionManager.elderLock
* <p>
* All calls to the elder may result in elder recovery which may call back into dlock and acquire
* synchronization on this.becomeGrantorMonitor.
*/
private GrantorInfo becomeGrantor(InternalDistributedMember predecessor) {
return GrantorRequestProcessor.becomeGrantor(this, getSerialNumber(), predecessor, this.ds);
}
// -------------------------------------------------------------------------
// New external API methods
// -------------------------------------------------------------------------
@Override
public void becomeLockGrantor() {
becomeLockGrantor((InternalDistributedMember) null);
}
public DLockGrantor getGrantor() {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
return this.grantor;
}
}
public DLockGrantor getGrantorWithNoSync() {
return this.grantor;
}
/**
* @param predecessor non-null if a predecessor asked us to take over for it
*/
private void becomeLockGrantor(InternalDistributedMember predecessor) {
Assert.assertTrue(predecessor == null);
boolean ownLockGrantorFutureResult = false;
FutureResult lockGrantorFutureResultRef = null;
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
LockGrantorId myLockGrantorId = null;
try { // finally handles lockGrantorFutureResult
// loop while other threads control the lockGrantorFutureResult
// terminate loop if other thread has already made us lock grantor
// terminate loop if this thread gets control of lockGrantorFutureResult
while (!ownLockGrantorFutureResult) {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
if (isCurrentlyOrIsMakingLockGrantor()) {
return;
} else if (this.lockGrantorFutureResult != null) {
// need to wait for other thread controlling lockGrantorFutureResult
lockGrantorFutureResultRef = this.lockGrantorFutureResult;
} else {
// this thread is in control and will procede to become grantor
// create new lockGrantorFutureResult for other threads to block on
ownLockGrantorFutureResult = true;
lockGrantorFutureResultRef = new FutureResult(this.dm.getCancelCriterion());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[becomeLockGrantor] creating lockGrantorFutureResult");
}
this.lockGrantorFutureResult = lockGrantorFutureResultRef;
}
}
if (!ownLockGrantorFutureResult) {
waitForLockGrantorFutureResult(lockGrantorFutureResultRef, 0, TimeUnit.MILLISECONDS);
}
}
// this thread is now in charge of the lockGrantorFutureResult future
getStats().incBecomeGrantorRequests();
// create the new grantor instance in non-ready state...
long tempGrantorVersion = -1;
LockGrantorId tempLockGrantorId =
new LockGrantorId(this.dm, this.dm.getId(), tempGrantorVersion, getSerialNumber());
DLockGrantor myGrantor = DLockGrantor.createGrantor(this, tempGrantorVersion);
try { // finally handles myGrantor
synchronized (this.lockGrantorIdLock) {
Assert.assertTrue(setLockGrantorId(tempLockGrantorId, myGrantor));
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "become set lockGrantorId to {} for service {}",
this.lockGrantorId, this.serviceName);
}
InternalDistributedMember elder = getElderId();
Assert.assertTrue(elder != null);
// NOTE: elder currently returns GrantorInfo for the previous grantor
// CONSIDER: add elderCommunicatedWith to GrantorInfo
GrantorInfo gi = becomeGrantor(predecessor);
boolean needsRecovery = gi.needsRecovery();
long myGrantorVersion = gi.getVersionId() + 1;
myGrantor.setVersionId(myGrantorVersion);
myLockGrantorId =
new LockGrantorId(this.dm, this.dm.getId(), myGrantorVersion, getSerialNumber());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[becomeLockGrantor] Calling makeLocalGrantor");
}
if (!makeLocalGrantor(elder, needsRecovery, myLockGrantorId, myGrantor)) {
return;
}
} finally {
Assert.assertTrue(
myGrantor == null || !myGrantor.isInitializing()
|| this.dm.getCancelCriterion().isCancelInProgress() || isDestroyed(),
"BecomeLockGrantor failed and left grantor non-ready");
}
} finally {
synchronized (this.lockGrantorIdLock) {
if (ownLockGrantorFutureResult) {
// this thread is doing the real work and must finish the future
Assert.assertTrue(this.lockGrantorFutureResult == lockGrantorFutureResultRef);
boolean getLockGrantorIdFailed = myLockGrantorId == null;
if (getLockGrantorIdFailed) {
// failed so cancel lockGrantorFutureResult
lockGrantorFutureResultRef.cancel(true); // interrupt waiting threads
} else {
this.dm.getCancelCriterion().checkCancelInProgress(null); // don't succeed if shutting
// down
// succeeded so set lockGrantorFutureResult
lockGrantorFutureResultRef.set(myLockGrantorId);
}
// null out the reference so it is free for next usage
this.lockGrantorFutureResult = null;
}
}
}
}
@Override
public boolean isLockGrantor() {
if (isDestroyed()) {
return false;
} else {
return isCurrentlyLockGrantor();
}
}
boolean isMakingLockGrantor() {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
return this.lockGrantorId != null && this.lockGrantorId.isLocal(getSerialNumber())
&& this.grantor != null && this.grantor.isInitializing();
}
}
boolean isCurrentlyOrIsMakingLockGrantor() {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
return this.lockGrantorId != null && this.lockGrantorId.isLocal(getSerialNumber());
}
}
boolean isCurrentlyLockGrantor() {
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
return this.lockGrantorId != null && this.lockGrantorId.isLocal(getSerialNumber())
&& this.grantor != null && this.grantor.isReady();
}
}
// -------------------------------------------------------------------------
// External API methods
// -------------------------------------------------------------------------
@Override
public void freeResources(Object name) {
checkDestroyed();
if (name == null) {
removeAllUnusedTokens();
} else {
removeTokenIfUnused(name);
}
}
/**
* Attempt to destroy and remove lock token. Synchronizes on tokens map and the lock token.
*
* @param name the name of the lock token
* @return true if token has been destroyed and removed
*/
private boolean removeTokenIfUnused(Object name) {
synchronized (this.tokens) {
if (this.destroyed) {
getStats().incFreeResourcesFailed();
return false;
}
DLockToken token = this.tokens.get(name);
if (token != null) {
synchronized (token) {
if (!token.isBeingUsed()) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Freeing {} in {}", token, this);
}
removeTokenFromMap(name);
token.destroy();
getStats().incTokens(-1);
getStats().incFreeResourcesCompleted();
return true;
}
}
}
}
getStats().incFreeResourcesFailed();
return false;
}
protected Object removeTokenFromMap(Object name) {
return this.tokens.remove(name);
}
/**
* Attempt to destroy and remove all unused lock tokens. Synchronizes on tokens map and each lock
* token.
*/
private void removeAllUnusedTokens() {
synchronized (this.tokens) {
if (this.destroyed) {
getStats().incFreeResourcesFailed();
return;
}
Set<DLockToken> unusedTokens = Collections.emptySet();
for (DLockToken token : this.tokens.values()) {
synchronized (token) {
if (!token.isBeingUsed()) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Freeing {} in {}", token, this);
}
if (unusedTokens == Collections.EMPTY_SET) {
unusedTokens = new HashSet<>();
}
unusedTokens.add(token);
} else {
getStats().incFreeResourcesFailed();
}
}
}
for (DLockToken token : unusedTokens) {
synchronized (token) {
int tokensSizeBefore = this.tokens.size();
Object obj = removeTokenFromMap(token.getName());
Assert.assertTrue(obj != null);
int tokensSizeAfter = this.tokens.size();
Assert.assertTrue(tokensSizeBefore - tokensSizeAfter == 1);
token.destroy();
getStats().incTokens(-1);
getStats().incFreeResourcesCompleted();
}
}
}
}
/**
* Destroys and removes all lock tokens. Caller must synchronize on destroyLock. Synchronizes on
* tokens map and each token.
*/
private void removeAllTokens() {
synchronized (this.tokens) {
Assert.assertTrue(this.destroyed);
for (DLockToken token : this.tokens.values()) {
synchronized (token) {
token.destroy();
}
}
getStats().incTokens(-this.tokens.size());
this.tokens.clear();
}
}
@Override
public boolean isHeldByCurrentThread(Object name) {
checkDestroyed();
synchronized (this.tokens) {
DLockToken token = basicGetToken(name);
if (token == null)
return false;
synchronized (token) {
token.checkForExpiration();
return token.isLeaseHeldByCurrentThread();
}
}
}
public boolean isHeldByThreadId(Object name, int threadId) {
checkDestroyed();
synchronized (this.tokens) {
DLockToken token = basicGetToken(name);
if (token == null)
return false;
synchronized (token) {
token.checkForExpiration();
if (token.getLesseeThread() == null) {
return false;
}
return token.getLesseeThread().getThreadId() == threadId;
}
}
}
@Override
public boolean isLockingSuspendedByCurrentThread() {
checkDestroyed();
return isHeldByCurrentThread(SUSPEND_LOCKING_TOKEN);
}
@Override
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis) {
boolean tryLock = false;
return lock(name, waitTimeMillis, leaseTimeMillis, tryLock);
}
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis, boolean tryLock) {
return lock(name, waitTimeMillis, leaseTimeMillis, tryLock, false);
}
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis, boolean tryLock,
boolean disallowReentrant) {
return lock(name, waitTimeMillis, leaseTimeMillis, tryLock, false, false);
}
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis, boolean tryLock,
boolean disallowReentrant, boolean diableAlerts) {
checkDestroyed();
try {
boolean interruptible = false;
return lockInterruptibly(name, waitTimeMillis, leaseTimeMillis, tryLock, interruptible,
disallowReentrant, diableAlerts);
} catch (InterruptedException ex) { // LOST INTERRUPT
Thread.currentThread().interrupt();
// fail assertion
logger.error(LogMarker.DLS_MARKER, "lock() was interrupted", ex);
Assert.assertTrue(false, "lock() was interrupted: " + ex.getMessage());
}
return false;
}
@Override
public boolean lockInterruptibly(Object name, long waitTimeMillis, long leaseTimeMillis)
throws InterruptedException {
checkDestroyed();
boolean tryLock = false;
boolean interruptible = true;
return lockInterruptibly(name, waitTimeMillis, leaseTimeMillis, tryLock, interruptible, false);
}
/** Causes the current thread to sleep for millis and may or may not be interruptible */
private void sleep(long millis, boolean interruptible) throws InterruptedException {
if (interruptible) {
if (Thread.interrupted())
throw new InterruptedException();
Thread.sleep(millis);
return;
} else {
sleep(millis);
}
}
/** Causes the current thread to sleep for millis uninterruptibly */
private void sleep(long millis) {
// Non-interruptible case
StopWatch timer = new StopWatch(true);
while (true) {
boolean interrupted = Thread.interrupted();
try {
long timeLeft = millis - timer.elapsedTimeMillis();
if (timeLeft <= 0) {
break;
}
Thread.sleep(timeLeft);
break;
} catch (InterruptedException e) {
interrupted = true;
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
}
protected DLockRequestProcessor createRequestProcessor(LockGrantorId grantorId, Object name,
int threadId, long startTime, long requestLeaseTime, long requestWaitTime, boolean reentrant,
boolean tryLock) {
return createRequestProcessor(grantorId, name, threadId, startTime, requestLeaseTime,
requestWaitTime, reentrant, tryLock, false);
}
protected DLockRequestProcessor createRequestProcessor(LockGrantorId grantorId, Object name,
int threadId, long startTime, long requestLeaseTime, long requestWaitTime, boolean reentrant,
boolean tryLock, boolean disableAlerts) {
return new DLockRequestProcessor(grantorId, this, name, threadId, startTime, requestLeaseTime,
requestWaitTime, reentrant, tryLock, disableAlerts, this.dm);
}
protected boolean callReleaseProcessor(InternalDistributedMember grantor, Object name,
boolean lockBatch, int lockId) {
return DLockService.callReleaseProcessor(this.dm, this.serviceName, grantor, name, lockBatch,
lockId);
}
protected static boolean callReleaseProcessor(DistributionManager dm, String serviceName,
InternalDistributedMember grantor, Object name, boolean lockBatch, int lockId) {
DLockReleaseProcessor processor = new DLockReleaseProcessor(dm, grantor, serviceName, name);
return processor.release(grantor, serviceName, lockBatch, lockId);
}
public boolean lockInterruptibly(final Object name, final long waitTimeMillis,
final long leaseTimeMillis, final boolean tryLock, final boolean interruptible,
final boolean disallowReentrant) throws InterruptedException {
return lockInterruptibly(name, waitTimeMillis, leaseTimeMillis, tryLock, interruptible,
disallowReentrant, false);
}
/**
* @param name the name of the lock to acquire in this service. This object must conform to the
* general contract of <code>equals(Object)</code> and <code>hashCode()</code> as described
* in {@link java.lang.Object#hashCode()}.
*
* @param waitTimeMillis the number of milliseconds to try to acquire the lock before giving up
* and returning false. A value of -1 causes this method to block until the lock is
* acquired.
*
* @param leaseTimeMillis the number of milliseconds to hold the lock after granting it, before
* automatically releasing it if it hasn't already been released by invoking
* {@link #unlock(Object)}. If <code>leaseTimeMillis</code> is -1, hold the lock until
* explicitly unlocked.
*
* @param tryLock true if the lock should be acquired or fail if currently held. waitTimeMillis
* will be ignored if the lock is currently held by another client.
*
* @param interruptible true if this lock request is interruptible
*
* @param disableAlerts true to disable logging alerts if the dlock is taking a long time to
* acquired.
*
* @return true if the lock was acquired, false if the timeout <code>waitTimeMillis</code> passed
* without acquiring the lock.
*
* @throws InterruptedException if the thread is interrupted before or during this method.
*
* @throws UnsupportedOperationException if attempt to lock batch involves non-tryLocks
*/
public boolean lockInterruptibly(final Object name, final long waitTimeMillis,
final long leaseTimeMillis, final boolean tryLock, final boolean interruptible,
final boolean disallowReentrant, final boolean disableAlerts) throws InterruptedException {
checkDestroyed();
boolean interrupted = Thread.interrupted();
if (interrupted && interruptible) {
throw new InterruptedException();
}
try {
long statStart = getStats().startLockWait();
long startTime = getLockTimeStamp(dm);
long requestWaitTime = waitTimeMillis;
long requestLeaseTime = leaseTimeMillis;
// -1 means "lease forever". Long.MAX_VALUE is pretty close.
if (requestLeaseTime == -1)
requestLeaseTime = Long.MAX_VALUE;
// -1 means "wait forever". Long.MAX_VALUE is pretty close.
if (requestWaitTime == -1)
requestWaitTime = Long.MAX_VALUE;
long waitLimit = startTime + requestWaitTime;
if (waitLimit < 0)
waitLimit = Long.MAX_VALUE;
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - entering lock()", this, name);
}
DLockToken token = getOrCreateToken(name);
boolean gotLock = false;
blockedOn.set(name);
try { // try-block for end stats, token cleanup, and interrupt check
ThreadRequestState requestState = (ThreadRequestState) this.threadRequestState.get();
if (requestState == null) {
requestState = new ThreadRequestState(incThreadSequence(), interruptible);
this.threadRequestState.set(requestState);
} else {
requestState.interruptible = interruptible;
}
final int threadId = requestState.threadId;
// if reentry and no change to expiration then grantor is not bothered
long leaseExpireTime = 0;
boolean keepTrying = true;
int lockId = -1;
incActiveLocks();
while (keepTrying) {
checkDestroyed();
interrupted = Thread.interrupted() || interrupted; // clear
if (interrupted && interruptible) {
throw new InterruptedException();
}
// Check for recursive lock
boolean reentrant = false;
int recursionBefore = -1;
synchronized (token) {
token.checkForExpiration();
if (token.isLeaseHeldByCurrentThread()) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{} , name: {} - lock() is reentrant: {}", this,
name, token);
}
reentrant = true;
if (reentrant && disallowReentrant) {
throw new IllegalStateException(
String.format("%s attempted to reenter non-reentrant lock %s",
new Object[] {Thread.currentThread(), token}));
}
recursionBefore = token.getRecursion();
lockId = token.getLeaseId(); // keep lockId
if (lockId < 0) {
// loop back around due to expiration
continue;
}
} // isLeaseHeldByCurrentThread
} // token sync
LockGrantorId theLockGrantorId = getLockGrantorId();
if (reentrant) {
Assert.assertTrue(lockId > -1, "Reentrant lock must have lockId > -1");
// lockId = token.getLockId(); // keep lockId
} else {
// this thread is not current owner...
lockId = -1; // reset lockId back to -1
}
DLockRequestProcessor processor = createRequestProcessor(theLockGrantorId, name, threadId,
startTime, requestLeaseTime, requestWaitTime, reentrant, tryLock, disableAlerts);
if (reentrant) {
// check for race condition... reentrant expired already...
// related to bug 32765, but client-side... see bug 33402
synchronized (token) {
if (!token.isLeaseHeldByCurrentThread()) {
reentrant = false;
recursionBefore = -1;
token.checkForExpiration();
}
}
} else {
// set lockId since this is the first granting (non-reentrant)
lockId = processor.getProcessorId();
}
gotLock = processor.requestLock(interruptible, lockId); // can throw
// InterruptedException
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Grantor {} replied {}", theLockGrantorId,
processor.getResponseCodeString());
}
if (gotLock) {
leaseExpireTime = processor.getLeaseExpireTime();
int recursion = recursionBefore + 1;
if (!grantLocalDLockAfterObtainingRemoteLock(name, token, threadId, leaseExpireTime,
lockId, theLockGrantorId, processor, recursion)) {
continue;
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - granted lock: {}", this, name,
token);
}
keepTrying = false;
} else if (processor.repliedDestroyed()) {
checkDestroyed(); // throws LockServiceDestroyedException
Assert.assertTrue(isDestroyed(),
"Grantor reports service " + this + " is destroyed: " + name);
} else if (processor.repliedNotGrantor() || processor.hadNoResponse()) {
long waitForGrantorTime = waitLimit - token.getCurrentTime();
if (waitForGrantorTime <= 0) {
waitForGrantorTime = 100;
}
notLockGrantorId(theLockGrantorId, waitForGrantorTime, TimeUnit.MILLISECONDS);
// keepTrying is still true... loop back around
} else if (processor.repliedNotHolder()) {
// fix part of bug 32765 - reentrant/expiration problem
// probably expired... try to get non-reentrant lock
reentrant = false;
recursionBefore = -1;
synchronized (token) {
token.checkForExpiration();
if (token.isLeaseHeldByCurrentThread()) {
// THIS SHOULDN'T HAPPEN -- some sort of weird consistency
// problem. Do what the grantor says and release the lock...
logger.warn(LogMarker.DLS_MARKER, "Grantor reports reentrant lock not held: {}",
token);
// Attempt at fault tolerance: We thought we owned it, but we
// don't; let's release it. Removes hot loop in bug 37276,
// but does not address underlying consistency failure.
RemoteThread rThread = new RemoteThread(getDistributionManager().getId(), threadId);
token.releaseLock(lockId, rThread, false);
}
} // token sync
} // grantor replied NOT_HOLDER for reentrant lock
else {
// either dlock service is suspended or tryLock failed
// fixed the math here... bug 32765
if (waitLimit > token.getCurrentTime() + 20) {
sleep(20, interruptible);
}
keepTrying = waitLimit > token.getCurrentTime();
}
} // while (keepTrying)
// try-block for end stats, token cleanup, and interrupt check
} finally {
getStats().endLockWait(statStart, gotLock);
// cleanup token if failed to get lock
if (!gotLock) {
synchronized (token) {
token.decUsage();
}
freeResources(token.getName());
}
// reset the interrupt state
if (interrupted) {
Thread.currentThread().interrupt();
}
// throw InterruptedException only if failed to get lock and interrupted
if (!gotLock && interruptible && Thread.interrupted()) {
throw new InterruptedException();
}
blockedOn.set(null);
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - exiting lock() returning {}", this,
name, gotLock);
}
return gotLock;
} finally {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - exiting lock() without returning value",
this, name);
}
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
private boolean grantLocalDLockAfterObtainingRemoteLock(Object name, DLockToken token,
int threadId, long leaseExpireTime, int lockId, LockGrantorId theLockGrantorId,
DLockRequestProcessor processor, int recursion) {
boolean needToReleaseOrphanedGrant = false;
Assert.assertHoldsLock(this.destroyLock, false);
synchronized (this.lockGrantorIdLock) {
if (!checkLockGrantorId(theLockGrantorId)) {
// race: grantor changed
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"Cannot honor grant from {} because {} is now a grantor.", theLockGrantorId,
this.lockGrantorId);
}
} else if (isDestroyed()) {
// race: dls was destroyed
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"Cannot honor grant from {} because this lock service has been destroyed.",
theLockGrantorId);
}
needToReleaseOrphanedGrant = true;
} else {
synchronized (this.tokens) {
checkDestroyed();
Assert.assertTrue(token == basicGetToken(name));
RemoteThread rThread = new RemoteThread(getDistributionManager().getId(), threadId);
token.grantLock(leaseExpireTime, lockId, recursion, rThread);
return true;
} // tokens sync
}
}
if (needToReleaseOrphanedGrant) {
processor.getResponse().releaseOrphanedGrant(this.dm);
}
return false;
}
/**
* Allow locking to resume.
*/
@Override
public void resumeLocking() {
checkDestroyed();
try {
// need to resumeLocking before unlocking to avoid deadlock with
// other thread attempting to suspendLocking
unlock(SUSPEND_LOCKING_TOKEN);
} catch (IllegalStateException e) {
checkDestroyed();
throw e;
}
}
/**
* Suspends granting of locks for this instance of DLockService. If distribute is true, sends
* suspendLocking to all other members that have created this service. Blocks until all
* outstanding locks have been released (excluding those held by the initial calling thread).
*
* @param waitTimeMillis -1 means "wait forever", >=0 = milliseconds to wait
*
* @return true if locking is suspended and all locks have been released. Otherwise, resumeLocking
* is invoked and false is returned.
*/
@Override
public boolean suspendLocking(final long waitTimeMillis) {
long startTime = System.currentTimeMillis();
long requestWaitTime = waitTimeMillis;
boolean interrupted = false;
try {
do {
checkDestroyed();
try {
return suspendLockingInterruptibly(requestWaitTime, false);
} catch (InterruptedException ex) {
interrupted = true;
long millisPassed = System.currentTimeMillis() - startTime;
if (requestWaitTime >= 0) {
requestWaitTime = Math.max(0, requestWaitTime - millisPassed);
}
}
} while (requestWaitTime != 0);
} finally {
if (interrupted)
Thread.currentThread().interrupt();
}
return false;
}
@Override
public boolean suspendLockingInterruptibly(long waitTimeMillis) throws InterruptedException {
return suspendLockingInterruptibly(waitTimeMillis, true);
}
public boolean suspendLockingInterruptibly(long waitTimeMillis, boolean interruptible)
throws InterruptedException {
checkDestroyed();
boolean wasInterrupted = false;
if (Thread.interrupted()) {
if (interruptible) {
throw new InterruptedException();
} else {
wasInterrupted = true;
}
}
try {
if (isLockingSuspendedByCurrentThread()) {
throw new IllegalStateException(
"Current thread has already locked entire service");
}
// have to use tryLock to avoid deadlock with other members that are
// simultaneously attempting to suspend locking
boolean tryLock = false; // go with false to queue up suspend lock requests
// when tryLock is false, we get deadlock:
// thread 1 is this thread
// thread 2 is processing a SuspendMessage... goes thru
// suspendLocking with distribute=false and gets stuck in
// waitForGrantorCallsInProgress
SuspendLockingToken suspendToken = SUSPEND_LOCKING_TOKEN;
boolean gotToken = false;
boolean keepTrying = true;
long startTime = System.currentTimeMillis();
long waitLimit = startTime + waitTimeMillis;
if (waitLimit < 0)
waitLimit = Long.MAX_VALUE;
while (!gotToken && keepTrying) {
gotToken =
lockInterruptibly(suspendToken, waitTimeMillis, -1, tryLock, interruptible, false);
keepTrying = !gotToken && waitLimit > System.currentTimeMillis();
}
return gotToken;
} finally {
if (wasInterrupted) {
Thread.currentThread().interrupt();
}
}
}
@Override
public void unlock(Object name) throws LockNotHeldException, LeaseExpiredException {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - entering unlock()", this, name);
}
long statStart = getStats().startLockRelease();
boolean hadRecursion = false;
boolean unlocked = false;
int lockId = -1;
DLockToken token = null;
RemoteThread rThread = null;
try {
synchronized (this.tokens) {
checkDestroyed();
token = basicGetToken(name);
if (token == null) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, [unlock] no token found for: {}", this, name);
}
throw new LockNotHeldException(
String.format(
"Attempting to unlock %s : %s , but this thread does not own the lock.",
new Object[] {this, name}));
}
synchronized (token) {
token.checkForExpiration();
rThread = token.getLesseeThread();
if (!token.isLeaseHeldByCurrentOrRemoteThread(rThread)) {
token.throwIfCurrentThreadHadExpiredLease();
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, [unlock] {} not leased by this thread.",
this, token);
}
throw new LockNotHeldException(
String.format(
"Attempting to unlock %s : %s , but this thread does not own the lock. %s",
new Object[] {this, name, token}));
}
// if recursion > 0 then token will still be locked after calling release
hadRecursion = token.getRecursion() > 0;
lockId = token.getLeaseId();
Assert.assertTrue(lockId > -1);
if (hadRecursion) {
unlocked = token.releaseLock(lockId, rThread);
} else {
token.setIgnoreForRecovery(true);
}
} // token sync
} // tokens map sync
if (!hadRecursion) {
boolean lockBatch = false;
boolean released = false;
while (!released) {
checkDestroyed();
LockGrantorId theLockGrantorId = getLockGrantorId();
try {
synchronized (this.lockGrantorIdLock) {
unlocked = token.releaseLock(lockId, rThread);
}
released = callReleaseProcessor(theLockGrantorId.getLockGrantorMember(), name,
lockBatch, lockId);
} catch (LockGrantorDestroyedException e) { // part of fix for bug 35239
// loop back around to get next lock grantor
} catch (LockServiceDestroyedException e) { // part of fix for bug 35239
// done... NonGrantorDestroyedMessage will release locks for us
released = true;
} finally {
if (!released) {
notLockGrantorId(theLockGrantorId, 0, TimeUnit.MILLISECONDS);
}
}
} // while !released
} // !hadRecursion
} // try
finally {
try {
if (!hadRecursion && lockId > -1 && token != null) {
decActiveLocks();
if (!unlocked) {
// // token is still held if grantor was remote, so now we unlock...
// checkDestroyed(); // part of fix for bug 35239
// // this release is ok even if we have become the lock grantor
// // because the grantor will have no state for this lock
token.releaseLock(lockId, rThread);
}
}
} finally {
getStats().endLockRelease(statStart);
if (this.automateFreeResources) {
freeResources(name);
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - exiting unlock()", this, name);
}
}
}
}
/**
* Query the grantor for current leasing information of a lock. Returns the current lease info.
*
* @param name the named lock to get lease information for
* @return snapshot of the remote lock information
* @throws LockServiceDestroyedException if local instance of lock service has been destroyed
*/
public DLockRemoteToken queryLock(final Object name) {
DLockQueryReplyMessage queryReply = null;
while (queryReply == null || queryReply.repliedNotGrantor()) {
checkDestroyed();
LockGrantorId theLockGrantorId = getLockGrantorId();
try {
queryReply = DLockQueryProcessor.query(theLockGrantorId.getLockGrantorMember(),
this.serviceName, name, false /* lockBatch */, this.dm);
} catch (LockGrantorDestroyedException e) {
// loop back around to get next lock grantor
} finally {
if (queryReply != null && queryReply.repliedNotGrantor()) {
notLockGrantorId(theLockGrantorId, 0, TimeUnit.MILLISECONDS);
}
}
} // while querying
return DLockRemoteToken.create(name, queryReply.getLesseeThread(), queryReply.getLeaseId(),
queryReply.getLeaseExpireTime());
}
// -------------------------------------------------------------------------
// Creation methods
// -------------------------------------------------------------------------
/**
* Factory method for creating a new instance of <code>DLockService</code>. This ensures that
* adding the {@link #disconnectListener} is done while synchronized on the fully constructed
* instance.
* <p>
* Caller must be synchronized on {@link DLockService#services}.
*
* @see org.apache.geode.distributed.DistributedLockService#create(String, DistributedSystem)
*/
static DLockService basicCreate(String serviceName, InternalDistributedSystem ds,
boolean isDistributed, boolean destroyOnDisconnect, boolean automateFreeResources)
throws IllegalArgumentException {
Assert.assertHoldsLock(services, true);
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "About to create DistributedLockService <{}>",
serviceName);
}
DLockService svc = new DLockService(serviceName, ds, isDistributed, destroyOnDisconnect,
automateFreeResources);
svc.init();
return svc;
}
/** initialize this DLockService object */
protected boolean init() {
boolean success = false;
try {
services.put(this.serviceName, this);
getStats().incServices(1);
this.ds.addDisconnectListener(disconnectListener);
success = true;
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Created DistributedLockService <{}>",
this.serviceName);
}
} finally {
if (!success) {
services.remove(this.serviceName);
getStats().incServices(-1);
}
}
ds.handleResourceEvent(ResourceEvent.LOCKSERVICE_CREATE, this);
return success;
}
// -------------------------------------------------------------------------
// Constructors
// -------------------------------------------------------------------------
/**
* To create an instance, use DistributedLockService.create(Object, DistributedSystem) or
* DLockService.create(Object, DistributedSystem, DistributionAdvisor)
*/
protected DLockService(String serviceName, DistributedSystem ds, boolean isDistributed,
boolean destroyOnDisconnect, boolean automateFreeResources) {
super();
this.dlockStats = getOrCreateStats(ds);
this.serialNumber = createSerialNumber();
this.serviceName = serviceName;
this.ds = (InternalDistributedSystem) ds;
this.dm = this.ds.getDistributionManager();
this.stopper = new DLockStopper(this.dm, this);
this.isDistributed = isDistributed;
this.destroyOnDisconnect = destroyOnDisconnect;
this.automateFreeResources = automateFreeResources || AUTOMATE_FREE_RESOURCES;
}
// -------------------------------------------------------------------------
// java.lang.Object methods
// -------------------------------------------------------------------------
@Override
public String toString() {
StringBuffer buffer = new StringBuffer(128);
buffer.append('<').append("DLockService").append("@")
.append(Integer.toHexString(System.identityHashCode(this))).append(" named ")
.append(this.serviceName).append(" destroyed=").append(this.destroyed).append(" grantorId=")
.append(this.lockGrantorId).append(" grantor=").append(this.grantor).append('>');
return buffer.toString();
}
// -------------------------------------------------------------------------
// Public instance methods
// -------------------------------------------------------------------------
public DistributedLockStats getStats() {
return this.dlockStats;
}
public void releaseTryLocks(DLockBatchId batchId, Callable<Boolean> untilCondition) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockService.releaseTryLocks] enter: {}", batchId);
}
long statStart = getStats().startLockRelease();
try {
boolean lockBatch = true;
boolean released = false;
while (!released) {
try {
boolean quit = untilCondition.call();
if (quit) {
return;
}
} catch (Exception e) {
throw new InternalGemFireException("unexpected exception", e);
}
checkDestroyed();
LockGrantorId theLockGrantorId = null;
theLockGrantorId = batchId.getLockGrantorId();
synchronized (this.lockGrantorIdLock) {
if (!checkLockGrantorId(theLockGrantorId)) {
// the grantor is different so break and skip DLockReleaseProcessor
break;
}
}
released =
callReleaseProcessor(theLockGrantorId.getLockGrantorMember(), batchId, lockBatch, -1);
if (!released) {
notLockGrantorId(theLockGrantorId, 100, TimeUnit.MILLISECONDS);
}
}
} finally {
decActiveLocks();
getStats().endLockRelease(statStart);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockService.releaseTryLocks] exit: {}", batchId);
}
}
}
public boolean acquireTryLocks(final DLockBatch dlockBatch, final long waitTimeMillis,
final long leaseTimeMillis, final Object[] keyIfFailed) throws InterruptedException {
checkDestroyed();
if (Thread.interrupted())
throw new InterruptedException();
if (keyIfFailed.length < 1) {
throw new IllegalArgumentException(
"keyIfFailed must have a length of one or greater");
}
long startTime = getLockTimeStamp(dm);
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[acquireTryLocks] acquiring {}", dlockBatch);
}
long requestWaitTime = waitTimeMillis;
long requestLeaseTime = leaseTimeMillis;
// -1 means "lease forever". Long.MAX_VALUE is pretty close.
if (requestLeaseTime == -1)
requestLeaseTime = Long.MAX_VALUE;
// -1 means "wait forever". Long.MAX_VALUE is pretty close.
if (requestWaitTime == -1)
requestWaitTime = Long.MAX_VALUE;
long waitLimit = startTime + requestWaitTime;
if (waitLimit < 0)
waitLimit = Long.MAX_VALUE;
long statStart = getStats().startLockWait();
boolean gotLocks = false;
try {
ThreadRequestState requestState = (ThreadRequestState) this.threadRequestState.get();
if (requestState == null) {
requestState = new ThreadRequestState(incThreadSequence(), false);
this.threadRequestState.set(requestState);
} else {
requestState.interruptible = false;
}
final int threadId = requestState.threadId;
boolean keepTrying = true;
incActiveLocks();
while (keepTrying) {
checkDestroyed();
LockGrantorId theLockGrantorId = getLockGrantorId();
boolean tryLock = true;
boolean reentrant = false;
DLockRequestProcessor processor = createRequestProcessor(theLockGrantorId, dlockBatch,
threadId, startTime, requestLeaseTime, requestWaitTime, reentrant, tryLock, false);
boolean interruptible = true;
int lockId = processor.getProcessorId();
gotLocks = processor.requestLock(interruptible, lockId);
if (gotLocks) {
dlockBatch.grantedBy(theLockGrantorId);
} else if (processor.repliedDestroyed()) {
checkDestroyed();
// should have thrown LockServiceDestroyedException
Assert.assertTrue(isDestroyed(), "Grantor reports service " + this + " is destroyed");
} else if (processor.repliedNotGrantor() || processor.hadNoResponse()) {
notLockGrantorId(theLockGrantorId, 0, TimeUnit.MILLISECONDS);
} else {
keyIfFailed[0] = processor.getKeyIfFailed();
if (keyIfFailed[0] == null) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[acquireTryLocks] lock request failed but provided no conflict key; responseCode=",
processor.getResponseCodeString());
}
} else {
break;
}
}
long timeLeft = requestWaitTime;
if (requestWaitTime < Long.MAX_VALUE) {
// prevent txLock from performing next line...
timeLeft = waitLimit - getLockTimeStamp(this.dm);
}
keepTrying = !gotLocks && timeLeft > 0;
if (keepTrying && timeLeft > 10) {
// didn't receive msg or processor timed out... sleep briefly
try {
Thread.sleep(10);
} catch (InterruptedException ignore) {
Thread.currentThread().interrupt();
}
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[acquireTryLocks] {} locks for {}",
(gotLocks ? "acquired" : "failed to acquire"), dlockBatch);
}
} finally {
getStats().endLockWait(statStart, gotLocks);
}
return gotLocks;
}
/**
* Returns copy of the tokens map. Synchronizes on token map.
* <p>
* Called by {@link org.apache.geode.internal.admin.remote.FetchDistLockInfoResponse}.
*
* @return copy of the tokens map
*/
public Map<Object, DLockToken> snapshotService() {
synchronized (this.tokens) {
return new HashMap(this.tokens);
}
}
/**
* Used for instrumenting blocked threads
*/
public UnsafeThreadLocal<Object> getBlockedOn() {
return blockedOn;
}
/** Returns true if the lock service is distributed; false if local only */
public boolean isDistributed() {
return this.isDistributed;
}
public void setDLockLessorDepartureHandler(DLockLessorDepartureHandler handler) {
this.lessorDepartureHandler = handler;
}
public DLockLessorDepartureHandler getDLockLessorDepartureHandler() {
return this.lessorDepartureHandler;
}
/** The name of this service */
public String getName() {
return this.serviceName;
}
public DistributionManager getDistributionManager() {
return this.dm;
}
public void setDLockRecoverGrantorMessageProcessor(
DLockRecoverGrantorProcessor.MessageProcessor recoverGrantorProcessor) {
this.recoverGrantorProcessor = recoverGrantorProcessor;
}
public DLockRecoverGrantorProcessor.MessageProcessor getDLockRecoverGrantorMessageProcessor() {
return this.recoverGrantorProcessor;
}
/**
* Returns true if any tokens in this service are currently held. Synchronizes on tokens map and
* on each token to check the lease.
*
* @return true if any tokens in this service are currently held
*/
boolean hasHeldLocks() {
synchronized (this.tokens) {
for (DLockToken token : this.tokens.values()) {
if (token.isLeaseHeld()) {
return true;
}
}
}
return false;
}
/**
* @see org.apache.geode.distributed.DistributedLockService#destroy(String)
*/
public static void destroyServiceNamed(String serviceName) throws IllegalArgumentException {
DLockService svc = null;
synchronized (services) {
svc = services.get(serviceName);
}
if (svc == null) {
throw new IllegalArgumentException(
String.format("Service named %s not created", serviceName));
} else {
svc.destroyAndRemove();
}
}
/** Destroys all lock services in this VM. Used in test tearDown code. */
public static void destroyAll() {
Collection<DLockService> svcs;
synchronized (services) {
svcs = new HashSet<>(services.values());
}
for (DLockService svc : svcs) {
try {
svc.destroyAndRemove();
} catch (CancelException e) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"destroyAndRemove of {} terminated due to cancellation: ", svc, e);
}
}
}
}
/**
* Destroys an existing service and removes it from the map
*
* @since GemFire 3.5
*/
public void destroyAndRemove() {
// isLockGrantor determines if we need to tell elder of destroy
boolean isCurrentlyLockGrantor = false;
boolean isMakingLockGrantor = false;
// maybeHasActiveLocks determines if we need to tell grantor of destroy
boolean maybeHasActiveLocks = false;
synchronized (creationLock) {
try {
synchronized (services) {
try {
if (isDestroyed())
return;
setDestroyingThread();
synchronized (this.lockGrantorIdLock) { // force ordering in lock request
synchronized (this.destroyLock) {
this.destroyed = true;
maybeHasActiveLocks = this.activeLocks > 0;
}
isCurrentlyLockGrantor = this.isCurrentlyLockGrantor();
isMakingLockGrantor = this.isMakingLockGrantor();
}
} finally {
if (isCurrentThreadDoingDestroy()) {
removeLockService(this);
}
}
} // services sync
} catch (CancelException e) {
// don't report to caller
} finally {
if (isCurrentThreadDoingDestroy()) {
try {
this.basicDestroy(isCurrentlyLockGrantor, isMakingLockGrantor, maybeHasActiveLocks);
} catch (CancelException e) {
// don't propagate
} finally {
clearDestroyingThread();
}
}
postDestroyAction();
}
} // creationLock sync
}
/**
* Unlock all locks currently held by this process, and mark it as destroyed. Returns true if
* caller performed actual destroy. Returns false if this lock service has already been destroyed.
* <p>
* {@link #services} is held for the entire operation. {@link #destroyLock} is held only while
* setting {@link #destroyed} to true and just prior to performing any real work.
* <p>
* Caller must be synchronized on {@link DLockService#services};
*/
private void basicDestroy(boolean isCurrentlyLockGrantor, boolean isMakingLockGrantor,
boolean maybeHasActiveLocks) {
Assert.assertHoldsLock(services, false);
// synchronized (this.serviceLock) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockService.basicDestroy] Destroying {}, isCurrentlyLockGrantor={}, isMakingLockGrantor={}",
this, isCurrentlyLockGrantor, isMakingLockGrantor);
}
// if hasActiveLocks, tell grantor we're destroying...
if (!isCurrentlyLockGrantor && maybeHasActiveLocks && !this.ds.isDisconnectThread()) {
boolean retry;
int nonGrantorDestroyLoopCount = 0;
do {
retry = false;
LockGrantorId theLockGrantorId = peekLockGrantorId();
if (theLockGrantorId != null && !theLockGrantorId.isLocal(getSerialNumber())) {
if (!NonGrantorDestroyedProcessor.send(this.serviceName, theLockGrantorId, dm)) {
// grantor responded NOT_GRANTOR
notLockGrantorId(theLockGrantorId, 0, TimeUnit.MILLISECONDS); // nulls out grantor to
// force call to elder
retry = true;
}
}
if (DEBUG_NONGRANTOR_DESTROY_LOOP) {
nonGrantorDestroyLoopCount++;
if (nonGrantorDestroyLoopCount >= DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT) {
logger.fatal(LogMarker.DLS_MARKER,
"Failed to notify grantor of destruction within {} attempts.",
Integer.valueOf(DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT));
Assert.assertTrue(false,
String.format("Failed to notify grantor of destruction within %s attempts.",
new Object[] {Integer.valueOf(DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT)}));
}
}
} while (retry);
}
if (isCurrentlyLockGrantor || isMakingLockGrantor) {
// If forcedDisconnect is in progress, the membership view will not
// change and no-one else can contact this member, so don't wait for a grantor
if (this.ds.getCancelCriterion().isCancelInProgress()) {
try {
DLockGrantor.waitForGrantor(this);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (DistributedSystemDisconnectedException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"No longer waiting for grantor because of disconnect.", e);
}
}
}
nullLockGrantorId(this.lockGrantorId);
}
// }
}
protected void postDestroyAction() {
ds.handleResourceEvent(ResourceEvent.LOCKSERVICE_REMOVE, this);
}
// -------------------------------------------------------------------------
// Package instance methods
// -------------------------------------------------------------------------
boolean destroyOnDisconnect() {
return this.destroyOnDisconnect;
}
/**
* Called by grantor recovery to return set of locks held by this process. Synchronizes on
* lockGrantorIdLock, tokens map, and each lock token.
*
* @param newlockGrantorId the newly recovering grantor
*/
Set<DLockRemoteToken> getLockTokensForRecovery(LockGrantorId newlockGrantorId) {
Set<DLockRemoteToken> heldLockSet = Collections.EMPTY_SET;
LockGrantorId currentLockGrantorId = null;
synchronized (this.lockGrantorIdLock) {
if (isDestroyed()) {
return heldLockSet;
}
currentLockGrantorId = this.lockGrantorId;
}
// destroy local grantor if currentLockGrantorId is local
// and grantorVersion is greater than currentGrantorVersion
if (currentLockGrantorId != null && currentLockGrantorId.hasLockGrantorVersion()
&& newlockGrantorId.isNewerThan(currentLockGrantorId)) {
nullLockGrantorId(currentLockGrantorId);
}
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
synchronized (this.lockGrantorIdLock) {
synchronized (this.tokens) {
// build up set of currently held locks
for (DLockToken token : this.tokens.values()) {
synchronized (token) {
if (token.isLeaseHeld()) {
// skip over token if ignoreForRecovery is true
if (token.ignoreForRecovery()) {
// unlock of token must be in progress... ignore for recovery
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "getLockTokensForRecovery is skipping {}",
token);
}
}
// add token to heldLockSet
else {
if (heldLockSet == Collections.EMPTY_SET) {
heldLockSet = new HashSet<>();
}
heldLockSet.add(DLockRemoteToken.createFromDLockToken(token));
}
} // isLeaseHeld
} // token sync
} // tokens iter
} // tokens sync
return heldLockSet;
}
}
/**
* Returns the named lock token or null if it doesn't exist. Synchronizes on tokens map.
*
* @return the named lock token or null if it doesn't exist
*/
public DLockToken getToken(Object name) {
synchronized (this.tokens) {
return this.tokens.get(name);
}
}
/**
* Returns the named lock token or null if it doesn't exist. Caller must synchronize on tokens
* map.
*
* @return the named lock token or null if it doesn't exist
*/
private DLockToken basicGetToken(Object name) {
return this.tokens.get(name);
}
/**
* Returns an unmodifiable collection backed by the values of the DLockToken map for testing
* purposes only. Synchronizes on tokens map.
*
* @return an unmodifiable collection of the tokens map values
*/
public Collection getTokens() {
synchronized (this.tokens) {
return Collections.unmodifiableCollection(this.tokens.values());
}
}
/**
* Returns an existing or creates a new DLockToken. Synchronizes on tokens map and the lock token.
*
* @param name the name of the lock
* @return an existing or new instantiated lock token
*/
DLockToken getOrCreateToken(Object name) {
synchronized (this.tokens) {
checkDestroyed(); // check destroy after acquiring tokens map sync
DLockToken token = this.tokens.get(name);
boolean createNewToken = token == null;
if (createNewToken) {
token = new DLockToken(this.dm, name);
}
synchronized (token) {
if (createNewToken) {
this.tokens.put(name, token);
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Creating {} in {}", token, this);
}
getStats().incTokens(1);
}
token.incUsage();
}
return token;
}
}
// -------------------------------------------------------------------------
// Private instance methods
// -------------------------------------------------------------------------
/**
* Returns number of lock tokens currently leased by this member. Synchronizes on tokens map and
* each lock token.
*
* @return number of lock tokens currently leased by this member
*/
private int numLocksHeldInThisVM() {
int numLocksHeld = 0;
synchronized (this.tokens) {
for (DLockToken token : this.tokens.values()) {
synchronized (token) {
if (token.isLeaseHeld()) {
numLocksHeld++;
}
}
}
}
return numLocksHeld;
}
/**
* TEST HOOK: Logs all lock tokens for this service at INFO level. Synchronizes on tokens map and
* each lock token.
*/
protected void dumpService() {
synchronized (this.tokens) {
StringBuffer buffer = new StringBuffer();
buffer.append(" ").append(this.tokens.size()).append(" tokens, ");
buffer.append(numLocksHeldInThisVM()).append(" locks held\n");
for (Map.Entry<Object, DLockToken> entry : this.tokens.entrySet()) {
buffer.append(" ").append(entry.getKey()).append(": ");
DLockToken token = entry.getValue();
buffer.append(token.toString()).append("\n");
}
logger.info(LogMarker.DLS_MARKER, buffer);
}
}
// ----------- new thread state for interruptible and threadId ------------
private final AtomicInteger threadSequence = new AtomicInteger();
protected static class ThreadRequestState {
protected final int threadId;
protected boolean interruptible;
ThreadRequestState(int threadId, boolean interruptible) {
this.threadId = threadId;
this.interruptible = interruptible;
}
}
private final ThreadLocal threadRequestState = new ThreadLocal();
private final UnsafeThreadLocal<Object> blockedOn = new UnsafeThreadLocal<Object>();
/**
* Returns true if the calling thread has an active lock request that is interruptible
*/
boolean isInterruptibleLockRequest() {
// this method is called by grantor for local lock requests in grantor vm
final ThreadRequestState requestState = (ThreadRequestState) threadRequestState.get();
if (requestState == null) {
return false;
}
return requestState.interruptible;
}
ThreadLocal getThreadRequestState() {
return threadRequestState;
}
protected int incThreadSequence() {
return this.threadSequence.incrementAndGet();
}
// -------------------------------------------------------------------------
// System disconnect listener
// -------------------------------------------------------------------------
/** Destroys all named locking services on disconnect from system */
@MakeNotStatic("This object is actually immutable ... but it uses singletons")
protected static final InternalDistributedSystem.DisconnectListener disconnectListener =
new InternalDistributedSystem.DisconnectListener() {
@Override
public String toString() {
return "Disconnect listener for DistributedLockService";
}
@Override
public void onDisconnect(final InternalDistributedSystem sys) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Shutting down Distributed Lock Services");
}
long start = System.currentTimeMillis();
try {
destroyAll();
} finally {
closeStats();
long delta = System.currentTimeMillis() - start;
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Distributed Lock Services stopped (took {} ms)",
delta);
}
}
}
};
// ----------------------------------------------------------------
@Immutable
private static final DummyDLockStats DUMMY_STATS = new DummyDLockStats();
@Immutable
static final SuspendLockingToken SUSPEND_LOCKING_TOKEN = new SuspendLockingToken();
// -------------------------------------------------------------------------
// Static fields
// -------------------------------------------------------------------------
/** Map of all locking services. */
@MakeNotStatic
protected static final Map<String, DLockService> services = new HashMap<>();
private static final Object creationLock = new Object();
/** All DLock threads belong to this group */
@MakeNotStatic
private static ThreadGroup threadGroup;
/** DLock statistics; static because multiple dlock instances can exist */
@MakeNotStatic
private static DistributedLockStats stats = DUMMY_STATS;
// -------------------------------------------------------------------------
// Reserved lock service names
// -------------------------------------------------------------------------
public static final String LTLS = "LTLS";
public static final String DTLS = "DTLS";
private static final String[] reservedNames = new String[] {LTLS, DTLS};
// -------------------------------------------------------------------------
// DLS serial number (uniquely identifies local instance of DLS)
// -------------------------------------------------------------------------
/**
* Specifies the starting serial number for the serialNumberSequencer
*/
public static final int START_SERIAL_NUMBER = Integer
.getInteger(DistributionConfig.GEMFIRE_PREFIX + "DistributedLockService.startSerialNumber", 1)
.intValue();
/**
* Incrementing serial number used to identify order of DLS creation
*
* @see DLockService#getSerialNumber()
*/
@MakeNotStatic
private static final AtomicInteger serialNumberSequencer = new AtomicInteger(START_SERIAL_NUMBER);
/**
* Identifies the static order in which this DLS was created in relation to other DLS or other
* instances of this DLS during the life of this JVM. Rollover to negative is allowed.
*/
private final int serialNumber;
/**
* Generates a serial number for identifying a DLS. Later instances of the same named DLS will
* have a greater serial number than earlier instances. This number increments statically
* throughout the life of this JVM. Rollover to negative is allowed.
*
* @return the new serial number
*/
protected static int createSerialNumber() {
// NOTE: AtomicInteger should rollover if value is Integer.MAX_VALUE
return serialNumberSequencer.incrementAndGet();
}
/**
* Returns the serial number which identifies the static order in which this DLS was created in
* relation to other DLS'es or other instances of this named DLS during the life of this JVM.
*/
public int getSerialNumber() {
return this.serialNumber;
}
// -------------------------------------------------------------------------
// External API methods
// -------------------------------------------------------------------------
/**
* @see org.apache.geode.distributed.DistributedLockService#getServiceNamed(String)
*/
public static DistributedLockService getServiceNamed(String serviceName) {
synchronized (services) {
return services.get(serviceName);
}
}
public static DistributedLockService create(String serviceName, InternalDistributedSystem ds,
boolean distributed, boolean destroyOnDisconnect) {
return create(serviceName, ds, distributed, destroyOnDisconnect, false);
}
/**
* Creates named <code>DistributedLockService</code>.
*
* @param serviceName name of the service
* @param ds InternalDistributedSystem
* @param distributed true if lock service will be distributed; false will be local only
* @param destroyOnDisconnect true if lock service should destroy itself using system disconnect
* listener
* @param automateFreeResources true if freeResources should be automatically called during unlock
*
* @throws IllegalArgumentException if serviceName is invalid or this process has already created
* named service
*
* @throws IllegalStateException if system is in process of disconnecting
*
* @see org.apache.geode.distributed.DistributedLockService#create(String, DistributedSystem)
*/
public static DistributedLockService create(String serviceName, InternalDistributedSystem ds,
boolean distributed, boolean destroyOnDisconnect, boolean automateFreeResources)
throws IllegalArgumentException, IllegalStateException {
// basicCreate will construct DLockService and it calls getOrCreateStats...
synchronized (creationLock) {
synchronized (services) { // disconnectListener syncs on this
ds.getCancelCriterion().checkCancelInProgress(null);
if (services.get(serviceName) != null) {
throw new IllegalArgumentException(
String.format("Service named %s already created",
serviceName));
}
return DLockService.basicCreate(serviceName, ds, distributed, destroyOnDisconnect,
automateFreeResources);
}
}
}
public static void becomeLockGrantor(String serviceName) throws IllegalArgumentException {
becomeLockGrantor(serviceName, null);
}
public static void becomeLockGrantor(String serviceName, InternalDistributedMember oldTurk)
throws IllegalArgumentException {
if (serviceName == null || serviceName.length() == 0) {
throw new IllegalArgumentException(String.format("Service named '%s' is not valid",
serviceName));
}
DLockService svc = getInternalServiceNamed(serviceName);
if (svc == null) {
throw new IllegalArgumentException(
String.format("Service named %s not created", serviceName));
} else {
svc.becomeLockGrantor(oldTurk);
}
}
public static boolean isLockGrantor(String serviceName) throws IllegalArgumentException {
if (serviceName == null || serviceName.length() == 0) {
throw new IllegalArgumentException(String.format("Service named '%s' is not valid",
serviceName));
}
DLockService svc = null;
synchronized (services) {
svc = services.get(serviceName);
}
if (svc == null) {
throw new IllegalArgumentException(
String.format("Service named %s not created", serviceName));
} else {
return svc.isLockGrantor();
}
}
/**
* Fills lists of service names.
*
* @param grantors filled with service names of all services we are currently the grantor of.
* @param grantorVersions elder assigned grantor version ids
* @param grantorSerialNumbers member specific DLS serial number hosting grantor
* @param nonGrantors filled with service names of all services we have that we are not the
* grantor of.
*/
public static void recoverRmtElder(ArrayList grantors, ArrayList grantorVersions,
ArrayList grantorSerialNumbers, ArrayList nonGrantors) {
synchronized (services) {
Iterator entries = services.entrySet().iterator();
while (entries.hasNext()) {
Map.Entry entry = (Map.Entry) entries.next();
String serviceName = (String) entry.getKey();
DLockService service = (DLockService) entry.getValue();
boolean foundGrantor = false;
DLockGrantor grantor = service.getGrantor();
if (grantor != null && grantor.getVersionId() != -1 && !grantor.isDestroyed()) {
foundGrantor = true;
grantors.add(serviceName);
grantorVersions.add(Long.valueOf(grantor.getVersionId()));
grantorSerialNumbers.add(Integer.valueOf(service.getSerialNumber()));
}
if (!foundGrantor) {
nonGrantors.add(serviceName);
}
}
}
}
/**
* Called when an elder is doing recovery. For every service that we are the grantor for add it to
* the grantorMap For every service we have that is not in the grantorMap add its name to
* needsRecovery set.
*
* @param dm our local DM
*/
public static void recoverLocalElder(DistributionManager dm, Map<String, GrantorInfo> grantors,
Set<String> needsRecovery) {
synchronized (services) {
Iterator entries = services.entrySet().iterator();
while (entries.hasNext()) {
Map.Entry entry = (Map.Entry) entries.next();
String serviceName = (String) entry.getKey();
DLockService service = (DLockService) entry.getValue();
boolean foundGrantor = false;
DLockGrantor grantor = service.getGrantor();
if (grantor != null && grantor.getVersionId() != -1 && !grantor.isDestroyed()) {
foundGrantor = true;
GrantorInfo oldgi = grantors.get(serviceName);
if (oldgi == null || oldgi.getVersionId() < grantor.getVersionId()) {
grantors.put(serviceName, new GrantorInfo(dm.getId(), grantor.getVersionId(),
service.getSerialNumber(), false));
needsRecovery.remove(serviceName);
}
}
if (!foundGrantor && !dm.isLoner()) {
if (!grantors.containsKey(serviceName)) {
needsRecovery.add(serviceName);
}
}
}
}
}
// -------------------------------------------------------------------------
// Public static methods
// -------------------------------------------------------------------------
/** Convenience method to get named DLockService */
public static DLockService getInternalServiceNamed(String serviceName) {
synchronized (services) {
return services.get(serviceName);
}
}
/** Validates service name for external creation */
public static void validateServiceName(String serviceName) {
if (serviceName == null || serviceName.length() == 0) {
throw new IllegalArgumentException(
"Lock service name must not be null or empty");
}
for (String reservedName : reservedNames) {
if (serviceName.startsWith(reservedName)) {
throw new IllegalArgumentException(
String.format("Service named %s is reserved for internal use only",
serviceName));
}
}
}
/** Return a snapshot of all services */
public static Map<String, DLockService> snapshotAllServices() { // used by: internal/admin/remote
synchronized (services) {
return new HashMap<>(services);
}
}
/**
* TEST HOOK: Logs all lock tokens for every service at INFO level. Synchronizes on services map,
* service tokens maps and each lock token.
*/
public static void dumpAllServices() { // used by: distributed/DistributedLockServiceTest
StringBuffer buffer = new StringBuffer();
synchronized (services) {
logger.info(LogMarker.DLS_MARKER,
"DLockService.dumpAllServices() - " + services.size() + " services:\n");
for (Map.Entry entry : services.entrySet()) {
buffer.append(" " + entry.getKey() + ":\n");
DLockService svc = (DLockService) entry.getValue();
svc.dumpService();
if (svc.isCurrentlyLockGrantor()) {
svc.grantor.dumpService();
}
}
}
}
// -------------------------------------------------------------------------
// Package static methods
// -------------------------------------------------------------------------
static ThreadGroup getThreadGroup() {
return threadGroup;
}
/**
* Return a timestamp that represents the current time for locking purposes.
*
* @since GemFire 3.5
*/
static long getLockTimeStamp(DistributionManager dm) {
return dm.cacheTimeMillis();
}
/** Get or create static dlock stats */
protected static synchronized DistributedLockStats getOrCreateStats(DistributedSystem ds) {
if (stats == DUMMY_STATS) {
Assert.assertTrue(ds != null, "Need an instance of InternalDistributedSystem");
StatisticsFactory statFactory = ds;
long statId = OSProcess.getId();
stats = new DLockStats(statFactory, statId);
}
return stats;
}
protected static synchronized DistributedLockStats getDistributedLockStats() {
return stats;
}
public static void addLockServiceForTests(String name, DLockService service) {
synchronized (services) {
services.put(name, service);
}
}
public static void removeLockServiceForTests(String name) {
synchronized (services) {
services.remove(name);
}
}
protected static void removeLockService(DLockService service) {
service.removeAllTokens();
InternalDistributedSystem system = null;
synchronized (services) {
DLockService removedService = services.remove(service.getName());
if (removedService == null) {
// another thread beat us to the removal... return
return;
}
if (removedService != service) {
services.put(service.getName(), removedService);
} else {
service.getStats().incServices(-1);
}
system = removedService.getDistributionManager().getSystem();
}
// if disconnecting and this was the last service, cleanup!
if (services.isEmpty() && system.isDisconnecting()) {
// get rid of stats and thread group...
synchronized (DLockService.class) {
closeStats();
}
}
}
static void closeStats() {
if (stats != DUMMY_STATS) {
((DLockStats) stats).close();
stats = DUMMY_STATS;
}
}
/** Provide way to peek at current lock grantor id when dls does not exist */
static GrantorInfo checkLockGrantorInfo(String serviceName, InternalDistributedSystem system) {
GrantorInfo gi = GrantorRequestProcessor.peekGrantor(serviceName, system);
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[checkLockGrantorId] returning {}", gi);
}
return gi;
}
// -------------------------------------------------------------------------
// SuspendLockingToken inner class
// -------------------------------------------------------------------------
/** Used as the name (key) for the suspend locking entry in the tokens map */
@Immutable
public static class SuspendLockingToken implements DataSerializableFixedID {
public SuspendLockingToken() {}
@Override
public boolean equals(Object o) {
if (o == null)
return false;
return o instanceof SuspendLockingToken;
}
@Override
public int hashCode() {
// Since instances always equal each other, they return the same hashCode
return 15325;
}
@Override
public int getDSFID() {
return SUSPEND_LOCKING_TOKEN;
}
@Override
public void fromData(DataInput in) {}
@Override
public void toData(DataOutput out) {}
@Override
public Version[] getSerializationVersions() {
return null;
}
}
/**
* CancelCriterion which provides cancellation for DLS request and release messages. Cancellation
* may occur if shutdown or if DLS is destroyed.
*/
private static class DLockStopper extends CancelCriterion {
/**
* The DLockService this stopper will check for cancellation.
*/
private final DLockService dls;
/**
* Creates a new DLockStopper for the specified DLockService and DM.
*
* @param dm the DM to check for shutdown
* @param dls the DLockService to check for DLS destroy
*/
DLockStopper(DistributionManager dm, DLockService dls) {
Assert.assertTrue(dls != null);
this.dls = dls;
Assert.assertTrue(dls.getDistributionManager() != null);
}
@Override
public String cancelInProgress() {
String cancelInProgressString =
this.dls.getDistributionManager().getCancelCriterion().cancelInProgress();
if (cancelInProgressString != null) {
// delegate to underlying DM's CancelCriterion...
return cancelInProgressString;
} else if (this.dls.isDestroyed()) {
// this.stoppedByDLS = true;
return this.dls.generateLockServiceDestroyedMessage();
} else {
// return null since neither DM nor DLS are shutting down
// cannot call super.cancelInProgress because it's abstract
return null;
}
}
@Override
public RuntimeException generateCancelledException(Throwable e) {
String reason = cancelInProgress();
if (reason == null) {
return null;
}
return this.dls.generateLockServiceDestroyedException(reason);
}
}
public CancelCriterion getCancelCriterion() {
return stopper;
}
}