Google Git
Sign in
apache / geode / 7038f08ea88ec42f6c58764525460a1c7fe1abcf / . / geode-core / src / main / java / org / apache / geode / distributed / internal / locks / DLockGrantor.java
blob: 1d81005c29c0e020e1002dddc904ae9753780701 [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 static java.util.concurrent.TimeUnit.DAYS;
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.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.TransactionDataNodeHasDepartedException;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
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.MembershipListener;
import org.apache.geode.distributed.internal.locks.DLockQueryProcessor.DLockQueryMessage;
import org.apache.geode.distributed.internal.locks.DLockRequestProcessor.DLockRequestMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.IdentityArrayList;
import org.apache.geode.internal.cache.TXReservationMgr;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
/**
* Provides lock grantor authority to a distributed lock service. This is responsible for granting,
* releasing, and timing out locks as well as exposing hooks for recovery or transfer of lock
* grantor.
* <p>
* ReadWriteLocks are not currently handled by grantor recovery or transfer.
*
*/
@SuppressWarnings("unchecked")
public class DLockGrantor {
private static final Logger logger = LogService.getLogger();
public static final boolean DEBUG_SUSPEND_LOCK = // TODO:LOG:CONVERT: REMOVE THIS
Boolean.getBoolean(
DistributionConfig.GEMFIRE_PREFIX + "DLockService.DLockGrantor.debugSuspendLock");
/**
* Default wait before grantor thread will reawaken to check for expirations and timeouts.
*/
public static final long GRANTOR_THREAD_MAX_WAIT = DLockGrantorThread.MAX_WAIT;
/**
* Newly constructed grantor is INITIALIZING. All lock requests and related messages will block.
*/
private static final int INITIALIZING = 0; // msgs will block until READY
/**
* Grantor is READY and handling lock requests.
*/
private static final int READY = 1; // msgs will be processed
/**
* Grantor is DESTROYED and will respond as NOT_GRANTOR to any requests.
*/
private static final int DESTROYED = 5; // NOT_GRANTOR
/**
* DistributedLockService that this grantor is granting locks for.
*/
protected final DLockService dlock;
/**
* Map of grant tokens for tracking grantor-side state of distributed locks. Key: Object name,
* Value: DLockGrantToken grant
*
* guarded.By grantTokens
*/
private final Map grantTokens = new HashMap();
/**
* Dedicated thread responsible for handling expirations and timeouts.
*/
protected final DLockGrantorThread thread;
/**
* The current state of this grantor. Volatile for read access. Mutation must occur while
* synchronized on this grantor.
* <p>
* State starts out as INITIALIZING and moves to READY at which point the grantor will start
* servicing requests.
* <p>
* If a newcomer requests transfer of grantorship, then the state will change to HALTED and then
* finally become DESTROYED when transfer is complete. During HALTED, the elder recognizes the
* newcomer as the current grantor.
* <p>
* If this service is shutting down, it will seek out a successor to become the new grantor.
* During this phase, the state will be PENDING_SHUTDOWN which means no requests or releases will
* be replied to until a successor is found, registers transfer intentions with the elder and
* sends this process a transfer grantorship request. At that point the state will be HALTED as
* described above.
*
* guarded.By this
*/
private volatile int state = INITIALIZING;
/**
* ReadWriteLock to protect in-progress operations from destroying service.
*/
private final StoppableReentrantReadWriteLock destroyLock;
/**
* Specialized lock information for Transaction lock batches.
* <p>
* Key: Object batchId, Value: DLockBatch batch
* <p>
* Handling of batch locks synchronizes on this to assure serial processing.
*
* guarded.By batchLocks
*/
private final Map batchLocks = new HashMap();
/**
* Handles special lock-reservation type for transactions.
*/
private final TXReservationMgr resMgr = new TXReservationMgr(false);
private final Map<InternalDistributedMember, Long> membersDepartedTime = new LinkedHashMap();
private final long departedMemberKeptInMapMilliSeconds = DAYS.toMillis(1);
/**
* Enforces waiting until this grantor is initialized. Used to block all lock requests until
* INITIALIZED.
*/
private final StoppableCountDownLatch whileInitializing;
/**
* Enforces waiting until this grantor is destroyed. Used to block all lock requests while
* destroying. Latch opens after state becomes DESTROYED and grantor begins replying with
* NOT_GRANTOR.
*/
private final StoppableCountDownLatch untilDestroyed;
/**
* If -1 then it has not yet been fetched from elder. Otherwise it is the versionId that the elder
* gave us. During explicit becomeGrantor, the value is -1, and then the elder provides a real
* versionId.
*/
private final AtomicLong versionId = new AtomicLong(-1);
/**
* Used to verify that requestor member is still in view when granting.
*/
protected final DistributionManager dm;
// -------------------------------------------------------------------------
// SuspendLocking state (BEGIN)
/**
* Synchronization for protecting all SuspendLocking state. Guards all suspend locking state.
*/
protected final Object suspendLock = new Object();
/**
* Identifies remote thread that has currently suspended locking or null.
*
* Concurrency: protected by synchronization of {@link #suspendLock}
*/
protected RemoteThread lockingSuspendedBy = null;
/**
* Value indicates nonexistent lock
*/
protected static final int INVALID_LOCK_ID = -1;
/**
* Identifies the lockId used by the remote thread to suspend locking.
*
* Concurrency: protected by synchronization of {@link #suspendLock}
*/
protected int suspendedLockId = INVALID_LOCK_ID;
/**
* FIFO queue for suspend read and write lock waiters.
*
* guarded.By {@link #suspendLock}
*/
private final LinkedList suspendQueue = new LinkedList();
/**
* Map of read lock counts for each RemoteThread currently holding locks.
* <p>
* Key=RemoteThread, Value=ReadLockCount
*
* guarded.By {@link #suspendLock}
*/
private final HashMap readLockCountMap = new HashMap();
/**
* Number of suspend waiters waiting for write lock.
*
* guarded.By {@link #suspendLock}
*/
private int writeLockWaiters = 0;
/**
* Total number of read locks held against suspend write lock.
*
* guarded.By {@link #suspendLock}
*/
private int totalReadLockCount = 0;
/**
* List of next requests to process after handling an unlock or resume.
*
* guarded.By {@link #suspendLock}
*/
private ArrayList permittedRequests = new ArrayList();
/**
* List of active drains of permittedRequests. TODO: does this need to be a synchronizedList?
*
* guarded.By {@link #suspendLock}
*/
private final List permittedRequestsDrain = Collections.synchronizedList(new LinkedList());
// SuspendLocking state (END)
// -------------------------------------------------------------------------
// -------------------------------------------------------------------------
// Static methods
// -------------------------------------------------------------------------
/**
* Creates new instance of grantor for the lock service.
*
* @param dlock the lock service the grantor is authority for
* @param versionId the version, from the elder, of this grantor
* @return new instance of grantor for the lock service
*/
static DLockGrantor createGrantor(DLockService dlock, long versionId) {
return new DLockGrantor(dlock, versionId);
}
/**
* Gets the local instance of grantor for the lock service if one exists.
*
* @param dlock the lock service the grantor is authority for
* @return instance of grantor for the lock service or null if no local grantor has been created
*/
static DLockGrantor getGrantorForService(DLockService dlock) {
if (dlock == null)
return null;
return dlock.getGrantor();
}
/**
* Returns instance of DLockGrantor that will handle distributed lock granting for specified
* service.
* <p>
* Returns null if unable to get ready grantor or if this process is not the grantor for the
* service.
*
* @param svc the lock service to return the grantor instance for
*/
public static DLockGrantor waitForGrantor(DLockService svc) throws InterruptedException {
if (svc == null)
return null;
// now we need to get the grantor and make sure it's ready...
DLockGrantor oldGrantor = null;
DLockGrantor grantor = getGrantorForService(svc);
do {
if (grantor == null || grantor.isDestroyed())
return null;
grantor.waitWhileInitializing();
if (svc.isDestroyed())
return null;
// make sure we are lock grantor
if (!svc.isCurrentlyOrIsMakingLockGrantor())
return null;
if (!grantor.isReady())
return null;
// Now make sure the service still has this member as its grantor
oldGrantor = grantor;
grantor = getGrantorForService(svc);
} while (oldGrantor != grantor);
return grantor;
}
// -------------------------------------------------------------------------
// Constructor
// -------------------------------------------------------------------------
/**
* Creates instance of grantor for the lock service.
*
* @param dlock the lock service the grantor is authority for
* @param vId unique id that the elder increments for each new grantor
*/
private DLockGrantor(DLockService dlock, long vId) {
this.dm = dlock.getDistributionManager();
CancelCriterion stopper = this.dm.getCancelCriterion();
this.whileInitializing = new StoppableCountDownLatch(stopper, 1);
this.untilDestroyed = new StoppableCountDownLatch(stopper, 1);
this.dlock = dlock;
this.destroyLock = new StoppableReentrantReadWriteLock(stopper);
this.versionId.set(vId);
this.dm.addMembershipListener(this.membershipListener);
this.thread = new DLockGrantorThread(this, stopper);
this.dlock.getStats().incGrantors(1);
}
// -------------------------------------------------------------------------
// Public and package methods
// -------------------------------------------------------------------------
/**
* Returns the grantor's version id which was assigned by an elder. Required to help uniquely
* identify this grantor instance.
*
* @return the grantor's version id which was assigned by an elder
*/
public long getVersionId() {
return this.versionId.get();
}
/**
* Sets the version id after the elder tells us what it is. This is called during the explicit
* become grantor process.
*
* @param v the elder assigned version id for this grantor
*/
public void setVersionId(long v) {
this.versionId.set(v);
}
/**
* Waits uninterruptibly while this grantor is initializing. Returns when grantor is ready to
* handle lock requests.
*
* @throws DistributedSystemDisconnectedException if system shuts down before grantor is ready
*/
public void waitWhileInitializing() throws InterruptedException {
boolean interrupted = Thread.interrupted();
try {
if (interrupted && this.dlock.isInterruptibleLockRequest()) {
throw new InterruptedException();
}
while (true) {
try {
this.whileInitializing.await();
break;
} catch (InterruptedException e) {
interrupted = true;
throwIfInterruptible(e);
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
/**
* Waits uninterruptibly until this service is destroyed. Returns when grantor has been completely
* destroyed.
*
* @throws DistributedSystemDisconnectedException if system shuts down before grantor is destroyed
*/
public void waitUntilDestroyed() throws InterruptedException {
while (true) {
boolean interrupted = Thread.interrupted();
try {
this.untilDestroyed.await();
break;
} catch (InterruptedException e) {
interrupted = true;
throwIfInterruptible(e);
} finally {
if (interrupted)
Thread.currentThread().interrupt();
}
}
}
@Override
public String toString() {
StringBuffer buffer = new StringBuffer(128);
buffer.append('<').append("DLockGrantor").append("@")
.append(Integer.toHexString(System.identityHashCode(this))).append(" state=")
.append(stateToString(this.state)).append(" name=").append(this.dlock.getName())
.append(" version=").append(this.getVersionId()).append('>');
return buffer.toString();
}
/**
* Returns true if this grantor is ready to handle lock requests.
*
* @return true if this grantor is ready to handle lock requests
*/
boolean isReady() {
return this.state == READY;
}
/**
* Returns true if this grantor is still initializing and not yet ready for lock requests.
*
* @return true if this grantor is still initializing
*/
boolean isInitializing() {
return this.state == INITIALIZING;
}
/**
* Returns true if this grantor has been destroyed.
*
* @return true if this grantor has been destroyed
*/
public boolean isDestroyed() {
return this.state == DESTROYED;
}
/**
* Throws LockGrantorDestroyedException if this grantor has been destroyed.
*
* @throws LockGrantorDestroyedException if grantor is destroyed
*/
void checkDestroyed() {
throwIfDestroyed(isDestroyed());
}
/**
* Throws LockGrantorDestroyedException if destroyed is true.
*
* @param destroyed if true then throw LockGrantorDestroyedException
* @throws LockGrantorDestroyedException if destroyed is true
*/
private void throwIfDestroyed(boolean destroyed) {
if (destroyed) {
throw new LockGrantorDestroyedException(
"Grantor is destroyed");
}
}
/**
* Handles request for a batch of locks using optimization for transactions.
* <p>
* Synchronizes on {@link #batchLocks}.
*
* @throws LockGrantorDestroyedException if grantor is destroyed
*/
void handleLockBatch(DLockRequestMessage request) throws InterruptedException {
DLockLessorDepartureHandler handler = this.dlock.getDLockLessorDepartureHandler();
// make sure the tx locks of departed members have been cleared so we don't have
// conflicts with non-existent members. This is done in a waiting-pool thread launched
// when the member-departure is announced.
handler.waitForInProcessDepartures();
synchronized (this.batchLocks) { // assures serial processing
waitWhileInitializing();
if (request.checkForTimeout()) {
cleanupSuspendState(request);
return;
}
final boolean isTraceEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch]");
}
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch] request: {}",
request);
}
DLockBatch batch = (DLockBatch) request.getObjectName();
checkIfHostDeparted(batch.getOwner());
resMgr.makeReservation((IdentityArrayList) batch.getReqs());
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch] granting {}",
batch.getBatchId());
}
this.batchLocks.put(batch.getBatchId(), batch);
request.respondWithGrant(Long.MAX_VALUE);
} catch (CommitConflictException ex) {
request.respondWithTryLockFailed(ex.getMessage());
} finally {
releaseDestroyReadLock();
}
}
}
private void checkIfHostDeparted(InternalDistributedMember owner) {
// Already held batchLocks; hold membersDepartedTime lock just for clarity
synchronized (membersDepartedTime) {
// the transaction host/txLock requester has departed.
if (membersDepartedTime.containsKey(owner)) {
throw new TransactionDataNodeHasDepartedException(
"The transaction host " + owner + " is no longer a member of the cluster.");
}
}
}
/**
* Returns transaction optimized lock batches that were created by the specified owner.
* <p>
* Synchronizes on batchLocks.
*
* @param owner member that owned the lock batches to return
* @return lock batches that were created by owner
*/
public DLockBatch[] getLockBatches(InternalDistributedMember owner) {
// Key: Object batchId, Value: DLockBatch batch
synchronized (this.batchLocks) {
// put owner into the map first so that no new threads will handle in-flight requests
// from the departed member to lock keys
recordMemberDepartedTime(owner);
List batchList = new ArrayList();
for (Iterator iter = this.batchLocks.values().iterator(); iter.hasNext();) {
DLockBatch batch = (DLockBatch) iter.next();
if (batch.getOwner().equals(owner)) {
batchList.add(batch);
}
}
return (DLockBatch[]) batchList.toArray(new DLockBatch[0]);
}
}
void recordMemberDepartedTime(InternalDistributedMember owner) {
// Already held batchLocks; hold membersDepartedTime lock just for clarity
synchronized (membersDepartedTime) {
long currentTime = getCurrentTime();
for (Iterator iterator = membersDepartedTime.values().iterator(); iterator.hasNext();) {
if ((long) iterator.next() < currentTime - departedMemberKeptInMapMilliSeconds) {
iterator.remove();
} else {
break;
}
}
membersDepartedTime.put(owner, currentTime);
}
}
long getCurrentTime() {
return System.currentTimeMillis();
}
@VisibleForTesting
Map getMembersDepartedTimeRecords() {
return membersDepartedTime;
}
/**
* Get the batch for the given batchId (for example use a txLockId from TXLockBatch in order to
* update its participants). This operation was added as part of the solution to bug 32999.
* <p>
* Acquires acquireDestroyReadLock. Synchronizes on batchLocks.
* <p>
* see org.apache.geode.internal.cache.TXCommitMessage#updateLockMembers()
*
* @param batchId the identifier for the batch to retrieve
* @return the transaction lock batch identified by the given batchId
* @see org.apache.geode.internal.cache.locks.TXLockUpdateParticipantsMessage
* @see org.apache.geode.internal.cache.locks.TXLockBatch#getBatchId()
*/
public DLockBatch getLockBatch(Object batchId) throws InterruptedException {
DLockBatch ret = null;
final boolean isTraceEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.getLockBatch] enter: {}", batchId);
}
synchronized (this.batchLocks) {
waitWhileInitializing();
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
ret = (DLockBatch) this.batchLocks.get(batchId);
} finally {
releaseDestroyReadLock();
}
}
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.getLockBatch] exit: {}", batchId);
}
return ret;
}
/**
* Update the batch for the given batch. This operation was added as part of the solution to bug
* 32999.
* <p>
* Acquires acquireDestroyReadLock. Synchronizes on batchLocks.
* <p>
* see org.apache.geode.internal.cache.locks.TXCommitMessage#updateLockMembers()
*
* @param batchId the identify of the transaction lock batch
* @param newBatch the new lock batch to be used
* @see org.apache.geode.internal.cache.locks.TXLockUpdateParticipantsMessage
* @see org.apache.geode.internal.cache.locks.TXLockBatch#getBatchId()
*/
public void updateLockBatch(Object batchId, DLockBatch newBatch) throws InterruptedException {
final boolean isTraceEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.updateLockBatch] enter: {}", batchId);
}
synchronized (this.batchLocks) {
waitWhileInitializing();
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
final DLockBatch oldBatch = (DLockBatch) this.batchLocks.get(batchId);
if (oldBatch != null) {
this.batchLocks.put(batchId, newBatch);
}
} finally {
releaseDestroyReadLock();
}
}
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.updateLockBatch] exit: {}", batchId);
}
}
/**
* Releases the transaction optimized lock batch.
* <p>
* Acquires acquireDestroyReadLock. Synchronizes on batchLocks.
*
* @param batchId the identify of the transaction lock batch to release
* @param owner the member that has created and locked the lock batch
* @throws LockGrantorDestroyedException if grantor is destroyed or interrupted
*/
public void releaseLockBatch(Object batchId, InternalDistributedMember owner)
throws InterruptedException {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.releaseLockBatch]");
}
synchronized (this.batchLocks) {
waitWhileInitializing();
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
DLockBatch batch = (DLockBatch) this.batchLocks.remove(batchId);
if (batch != null) {
this.resMgr.releaseReservation((IdentityArrayList) batch.getReqs());
}
} finally {
releaseDestroyReadLock();
}
}
}
/**
* Returns true if the request comes from the local member.
*
* @param request the lock request to check
* @return true if the request comes from the local member
*/
private boolean isLocalRequest(DLockRequestMessage request) {
return request.getSender().equals(this.dlock.getDistributionManager().getId());
}
/**
* TEST HOOK: Allows testing to determine if there are waiting requests for a lock.
* <p>
* Synchronizes on grantTokens and the grant token if one exists.
*
* @param name the lock to check for waiting requests for
* @return true if the named lock has requests waiting to acquire it
*/
boolean hasWaitingRequests(Object name) {
DLockGrantToken grant = getGrantToken(name);
if (grant == null)
return false;
synchronized (grant) {
return grant.hasWaitingRequests();
}
}
/**
* Handles a DLockQueryMessage. Returns DLockGrantToken for the lock or null.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens.
*
* @param query the dlock query message to handle
* @return DLockGrantToken for the lock or null
*/
DLockGrantToken handleLockQuery(DLockQueryMessage query) throws InterruptedException {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockQuery] {}", query);
}
if (acquireDestroyReadLock(0)) {
try {
checkDestroyed();
return getGrantToken(query.objectName);
} finally {
releaseDestroyReadLock();
}
}
return null;
}
/**
* Handles the provided lock request. The lock will either be granted, refused if try-lock, or
* scheduled at end of waiting queue to eventually be granted or timed out.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens, suspendLock and the grant token.
*
* @param request the lock request to be processed by this grantor
* @throws LockGrantorDestroyedException if grantor is destroyed
*/
void handleLockRequest(DLockRequestMessage request) throws InterruptedException {
Assert.assertTrue(request.getRemoteThread() != null);
if (request.getObjectName() instanceof DLockBatch) {
handleLockBatch(request);
return;
}
waitWhileInitializing(); // calcWaitMillisFromNow
final boolean isTraceEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockRequest] {}", request);
}
if (!acquireDestroyReadLock(0)) {
if (isLocalRequest(request) && this.dlock.isDestroyed()) {
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.handleLockRequest] about to throwIfDestroyed");
}
// this special case is one fix for deadlock between waitUntilDestroyed
// and dlock waitForGrantorCallsInProgress (when request is local)
throwIfDestroyed(true);
} else {
if (isTraceEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.handleLockRequest] about to waitUntilDestroyed");
}
// is there still a deadlock when an explicit become is destroying
// this grantor instead of destroying the dlock service?
waitUntilDestroyed();
checkDestroyed();
}
}
try {
// make sure we don't grant a dlock held by a departed member until that member's
// transactions are resolved
DLockLessorDepartureHandler dLockLessorDepartureHandler =
this.dlock.getDLockLessorDepartureHandler();
if (dLockLessorDepartureHandler != null) {
dLockLessorDepartureHandler.waitForInProcessDepartures();
}
checkDestroyed();
if (acquireLockPermission(request)) {
handlePermittedLockRequest(request);
} else {
// request has been added to suspendQueue for deferred handling
}
} finally {
releaseDestroyReadLock();
}
}
/**
* Internally handles a lock request which has permission to proceed.
* <p>
* Calling thread must hold destroyReadLock. Synchronizes on grantTokens, suspendLock and the
* grant token.
*
* @param request the lock request to be processed by this grantor guarded.By
* {@link #acquireDestroyReadLock(long)}
*/
private void handlePermittedLockRequest(final DLockRequestMessage request) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handlePermittedLockRequest] {}", request);
}
Assert.assertTrue(request.getRemoteThread() != null);
DLockGrantToken grant = getOrCreateGrant(request.getObjectName());
try {
// try to grant immediately if not currently granted...
if (grant.grantLockToRequest(request)) {
// do nothing
}
// if request was local and then interrupted/released...
else if (request.responded()) {
// do nothing
}
// if request was a failed try-lock...
else if (request.isTryLock()) {
cleanupSuspendState(request);
request.respondWithTryLockFailed(request.getObjectName());
}
// if request has timed out...
else if (request.checkForTimeout()) {
cleanupSuspendState(request);
}
// schedule into waiting queue for eventual granting...
else {
grant.schedule(request);
this.thread.checkTimeToWait(calcWaitMillisFromNow(request), false);
}
} finally {
grant.decAccess();
}
}
/**
* Initializes this new grantor with previously held locks as provided during grantor recovery.
* <p>
* Acquires destroyReadLock. Synchronizes on this grantor, grantTokens, suspendLock, the grant
* token.
*
* @param owner the member that owns the tokens to be scheduled
* @param tokens set of DLockRemoteTokens to be scheduled for owner
*/
void initializeHeldLocks(InternalDistributedMember owner, Set tokens)
throws InterruptedException {
synchronized (this) {
if (isDestroyed())
return;
if (!acquireDestroyReadLock(0)) {
return;
}
}
try {
synchronized (this.grantTokens) {
Set members = this.dlock.getDistributionManager().getDistributionManagerIds();
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
for (Iterator iter = tokens.iterator(); iter.hasNext();) {
DLockRemoteToken token = (DLockRemoteToken) iter.next();
DLockGrantToken grantToken = getOrCreateGrant(token.getName());
try {
// make sure the token's owner is still in the system
if (!members.contains(owner)) {
// skipping because member is no longer in view
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"Initialization of held locks is skipping {} because owner {} is not in view: ",
token, owner, members);
}
continue;
}
RemoteThread rThread = null;
boolean isSuspendLock = false;
int lockId = -1;
synchronized (grantToken) {
if (grantToken.isLeaseHeld()) {
logger.error(LogMarker.DLS_MARKER,
"Initialization of held locks is skipping {} because lock is already held: {}",
token, grantToken);
continue;
}
grantToken.grantLock(owner, token.getLeaseExpireTime(), token.getLeaseId(),
token.getLesseeThread());
// grantToken may have already expired or is about to expire
// complete initialization but make sure grantor thread will wake
// up and expire it as soon as it's running
if (grantToken.getLeaseExpireTime() > -1
&& grantToken.getLeaseExpireTime() < Long.MAX_VALUE) {
long now = DLockService.getLockTimeStamp(this.dm);
this.thread.checkTimeToWait(grantToken.getLeaseExpireTime() - now, true);
}
rThread = grantToken.getRemoteThread();
isSuspendLock = grantToken.isSuspendLockingToken();
lockId = grantToken.getLockId();
}
// update the readLock and suspendLocking states...
synchronized (suspendLock) {
if (isSuspendLock) {
suspendLocking(rThread, lockId);
} else {
Assert.assertTrue(!isLockingSuspended() || isLockingSuspendedBy(rThread),
"Locking is suspended by a different thread: " + token);
Integer integer = (Integer) readLockCountMap.get(rThread);
int readLockCount = integer == null ? 0 : integer.intValue();
readLockCount++;
readLockCountMap.put(rThread, Integer.valueOf(readLockCount));
totalReadLockCount++;
checkTotalReadLockCount();
}
} // suspendLock sync
} finally {
grantToken.decAccess();
}
} // tokens iter
} // grantTokens sync
return;
} finally {
releaseDestroyReadLock();
}
}
/**
* Handles a request for extending the lease time of an already held lock.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens and the grant token.
*
* @param request the lock request to be reentered for lease extension
* @return new extended leaseExpireTime or 0 if requestor no longer holds lock
*/
long reenterLock(DLockRequestMessage request) throws InterruptedException {
waitWhileInitializing(); // calcWaitMillisFromNow
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
// to fix GEODE-678 no longer call request.checkForTimeout
DLockGrantToken grant = getGrantToken(request.getObjectName());
if (grant == null) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.reenterLock] no grantToken found for {}", request.getObjectName());
}
return 0;
}
synchronized (grant) { // synchronize against grant.expireAndGrantLock
if (!this.dm.isCurrentMember(request.getSender()) || grant.isDestroyed()) {
return 0;
}
if (!grant.isLockedBy(request.getSender(), request.getLockId())) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.reenterLock] grant is not locked by sender={} lockId={} grant={}",
request.getSender(), request.getLockId(), grant);
}
return 0;
}
long leaseExpireTime =
Math.max(grant.getLeaseExpireTime(), grant.calcLeaseExpireTime(request.getLeaseTime()));
grant.grantLock(request.getSender(), leaseExpireTime, request.getLockId(),
grant.getRemoteThread());
return grant.getLeaseExpireTime();
}
} finally {
releaseDestroyReadLock();
}
}
/**
* Release named lock if held by owner using lockId. Called from DLockReleaseMessage.basicProcess
* for remote unlock.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens and the grant token.
*
* @param name the name of the lock to release
* @param owner the member releasing the lock
* @param lockId the identity of the lease used by the owner
* @throws LockGrantorDestroyedException if grantor is destroyed
*/
void releaseIfLocked(Object name, InternalDistributedMember owner, int lockId)
throws InterruptedException {
waitWhileInitializing();
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
getAndReleaseGrantIfLockedBy(name, owner, lockId);
} finally {
releaseDestroyReadLock();
}
}
/**
* Fetches the actual grant token and releases it if leased by owner using lockId.
* DLockReleaseMessage.basicProcess -> releaseIfLocked -> getAndReleaseGrantIfLockedBy
* <p>
* Caller must hold destroyReadLock. Synchronizes on grantTokens and the grant token.
*
* @param name the name of the lock to release
* @param owner the member attempting to release the granted lock
* @param lockId the id of the lease used by the owner guarded.By
* {@link #acquireDestroyReadLock(long)}
*/
private void getAndReleaseGrantIfLockedBy(Object name, InternalDistributedMember owner,
int lockId) {
synchronized (this.grantTokens) {
DLockGrantToken grantToken = basicGetGrantToken(name);
if (grantToken != null) { // checking isTokenDestroyed here will deadlock
synchronized (grantToken) {
// if (!grantToken.isTokenDestroyed())
try {
grantToken.releaseIfLockedBy(owner, lockId);
removeGrantIfUnused(grantToken);
} catch (IllegalStateException e) {
this.dlock.checkDestroyed();
checkDestroyed();
// must have hit race... grantor doesn't have the token
return;
}
}
}
}
}
/**
* Fetches the grant token for named lock and attempts to grant it to the next waiting requestor
* if one exists. Called from DLockReleaseProcessor when another process releases a lock and after
* the reply has been sent.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens and the grant token.
*
* @param name the name of the lock to grant
* @throws LockGrantorDestroyedException if grantor is destroyed
*/
void grantLock(Object name) throws InterruptedException {
waitWhileInitializing();
if (!acquireDestroyReadLock(0)) {
waitUntilDestroyed();
checkDestroyed();
}
try {
checkDestroyed();
DLockGrantToken grant = getGrantToken(name);
if (grant != null) {
removeGrantIfUnused(grant);
}
} finally {
releaseDestroyReadLock();
}
}
/**
* Handles the departure of a member by releasing every lock it owned.
* <p>
* Acquires destroyReadLock. Synchronizes on grantTokens, suspendLock, and the grant token.
*
* @param owner the member that departed
*/
void handleDepartureOf(InternalDistributedMember owner) throws InterruptedException {
// bug 32657 has another cause in this method... interrupted thread from
// connection/channel layer caused acquireDestroyReadLock to fail...
// fixed by Darrel in org.apache.geode.internal.tcp.Connection
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (acquireDestroyReadLock(0)) {
try {
if (isDestroyed()) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.handleDepartureOf] grantor is destroyed; ignoring {}", owner);
}
return;
}
try {
DLockLessorDepartureHandler handler = this.dlock.getDLockLessorDepartureHandler();
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleDepartureOf] handler = {}",
handler);
}
if (handler != null) {
handler.handleDepartureOf(owner, this);
}
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DlockGrantor.handleDepartureOf] ignored cancellation (1)");
}
} finally {
synchronized (this.suspendLock) {
HashSet removals = new HashSet();
for (Iterator it = readLockCountMap.entrySet().iterator(); it.hasNext();) {
Map.Entry entry = (Map.Entry) it.next();
RemoteThread rThread = (RemoteThread) entry.getKey();
if (rThread.getDistributedMember().equals(owner)) {
removals.add(rThread);
}
}
for (Iterator it = removals.iterator(); it.hasNext();) {
try {
postReleaseLock((RemoteThread) it.next(), null);
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DlockGrantor.handleDepartureOf] ignored cancellation (2)");
}
}
}
} // synchronized
synchronized (this.grantTokens) {
// do not call handleDepartureOf while iterating grantTokens
// changes fix bug 39172 (ConcurrentModificationException)
// 1) built up list of grants that reference departed member
List grantsReferencingMember = new ArrayList();
Collection grants = this.grantTokens.values();
for (Iterator iter = grants.iterator(); iter.hasNext();) {
DLockGrantToken grant = (DLockGrantToken) iter.next();
try {
grant.checkDepartureOf(owner, grantsReferencingMember);
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DlockGrantor.handleDepartureOf] ignored cancellation (3)");
}
}
} // for
// 2) call handleDepartureOf on list of grantsReferencingMember
ArrayList grantsToRemoveIfUnused = new ArrayList();
for (Iterator iter = grantsReferencingMember.iterator(); iter.hasNext();) {
DLockGrantToken grant = (DLockGrantToken) iter.next();
try {
grant.handleDepartureOf(owner, grantsToRemoveIfUnused);
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DlockGrantor.handleDepartureOf] ignored cancellation (4)");
}
}
} // for
// 3) remove grants in grantsToRemoveIfUnused list
// TODO: if grantsReferencingMember is always empty remove this
for (Iterator iter = grantsToRemoveIfUnused.iterator(); iter.hasNext();) {
DLockGrantToken grant = (DLockGrantToken) iter.next();
try {
removeGrantIfUnused(grant);
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DlockGrantor.handleDepartureOf] ignored cancellation (5)");
}
}
} // for
} // synchronized this.grantTokens
} // finally
} finally {
releaseDestroyReadLock();
}
}
}
/**
* Destroys this grantor without attempting to transfer grant tokens to a successor.
* <p>
* Acquires destroyWriteLock. Synchronizes on this grantor, grantTokens, and each grant token.
*/
void destroy() {
synchronized (this) {
if (isDestroyed())
return;
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[simpleDestroy]");
}
// wait for the destroy write lock ignoring interrupts...
boolean acquired = false;
try {
boolean locksHeld = false;
try {
acquireDestroyWriteLock(Long.MAX_VALUE);
acquired = true;
// check for any held locks...
if (isInitializing()) {
// assume the worst case and tell the elder that recovery will be required
locksHeld = true;
} else {
synchronized (this.grantTokens) {
InternalDistributedMember me = this.dlock.getDistributionManager().getId();
for (Iterator iter = this.grantTokens.values().iterator(); iter.hasNext();) {
DLockGrantToken grant = (DLockGrantToken) iter.next();
InternalDistributedMember owner = grant.getOwner();
if (owner != null && !owner.equals(me)) {
locksHeld = true;
break;
}
}
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[simpleDestroy] {} locks held",
(locksHeld ? "with" : "without"));
}
} finally {
// make sure the following occurs even if checking locks above failed
try {
// release latches and change internal state
destroyGrantor();
} finally {
// tell the elder we are not the grantor anymore
this.dlock.clearGrantor(this.getVersionId(), locksHeld);
}
}
} finally {
if (acquired) {
releaseDestroyWriteLock();
}
}
}
}
/**
* Send replies to all waiting requestors to notify them that this is no longer the grantor.
* <p>
* Caller must acquire destroyWriteLock. Synchronizes on suspendLock, grantTokens, and each grant
* token.
*
* guarded.By {@link #acquireDestroyWriteLock(long)}
*/
private void destroyGrantor() {
Assert.assertHoldsLock(this, true);
makeDestroyed();
// reply to all pending requests w/ NOT_GRANTOR
synchronized (this.grantTokens) {
Collection grants = this.grantTokens.values();
for (Iterator iter = grants.iterator(); iter.hasNext();) {
DLockGrantToken grant = (DLockGrantToken) iter.next();
grant.handleGrantorDestruction();
}
}
synchronized (suspendLock) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.destroyAndRemove] responding to {} permitted requests.",
permittedRequests.size());
}
respondWithNotGrantor(permittedRequests.iterator());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.destroyAndRemove] responding to {} requests awaiting permission.",
suspendQueue.size());
}
respondWithNotGrantor(suspendQueue.iterator());
for (Iterator iter = permittedRequestsDrain.iterator(); iter.hasNext();) {
final List drain = (List) iter.next();
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.destroyAndRemove] responding to {} drained permitted requests.",
drain.size());
}
respondWithNotGrantor(drain.iterator());
}
}
}
/**
* Send responses to specified requests informing the senders that this is no longer the grantor.
* <p>
* Caller must acquire destroyWriteLock.
*
* @param requests the requests to respond to guarded.By {@link #acquireDestroyWriteLock(long)}
*/
private void respondWithNotGrantor(Iterator requests) {
while (requests.hasNext()) {
final DLockRequestMessage request = (DLockRequestMessage) requests.next();
request.respondWithNotGrantor();
}
}
/**
* Make this grantor ready for handling lock requests.
* <p>
* Synchronizes on this grantor.
*
* @param enforceInitializing true if this should assert isInitializing
* @return true if grantor was successfully made ready
*/
synchronized boolean makeReady(boolean enforceInitializing) {
if (isDestroyed()) {
this.dlock.checkDestroyed();
}
if (!enforceInitializing) {
if (!isInitializing()) {
return false;
}
}
assertInitializing();
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
StringBuffer sb =
new StringBuffer("DLockGrantor " + this.dlock.getName() + " initialized with:");
for (Iterator tokens = grantTokens.values().iterator(); tokens.hasNext();) {
sb.append("\n\t" + tokens.next());
}
logger.trace(LogMarker.DLS_VERBOSE, sb.toString());
}
this.state = READY;
this.whileInitializing.countDown();
this.thread.start();
return true;
}
// -------------------------------------------------------------------------
// Private methods
// -------------------------------------------------------------------------
/**
* Drain currently permitted requests and grant lock to next requestor.
* <p>
* Acquires destroyReadLock. Synchronizes on suspendLock, grantTokens, and the grant token.
*
* @param objectName the lock to perform post release tasks for
*/
void postRemoteReleaseLock(Object objectName) throws InterruptedException {
if (!acquireDestroyReadLock(0)) {
return;
}
try {
checkDestroyed();
drainPermittedRequests();
grantLock(objectName);
} catch (LockServiceDestroyedException | LockGrantorDestroyedException e) {
// ignore... service was destroyed and that's ok
} finally {
releaseDestroyReadLock();
}
}
/**
* Acquires a read lock on the destroy ReadWrite lock uninterruptibly using millis for try-lock
* attempt.
*
* @param millis the milliseconds to try to acquire lock within
* @return true if destroy read lock was acquired
* @throws DistributedSystemDisconnectedException if system has been disconnected
*/
private boolean acquireDestroyReadLock(long millis) throws InterruptedException {
boolean interrupted = Thread.interrupted();
try {
if (interrupted && this.dlock.isInterruptibleLockRequest()) {
throw new InterruptedException();
}
while (true) {
try {
this.dm.getCancelCriterion().checkCancelInProgress(null); // is this needed?
boolean acquired = this.destroyLock.readLock().tryLock(millis);
return acquired;
} catch (InterruptedException e) {
interrupted = true;
throwIfInterruptible(e);
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
/**
* Releases a read lock on the destroy ReadWrite lock.
*/
private void releaseDestroyReadLock() {
this.destroyLock.readLock().unlock();
}
/**
* Acquires the write lock on the destroy ReadWrite lock within specified millis.
*
* @param millis the milliseconds to attempt to acquire the lock within
* @throws DistributedSystemDisconnectedException if system has been disconnected
*/
private void acquireDestroyWriteLock(long millis) {
for (;;) {
boolean interrupted = Thread.interrupted();
try {
this.dm.getCancelCriterion().checkCancelInProgress(null);
boolean acquired = this.destroyLock.writeLock().tryLock(millis);
if (acquired) {
return;
}
} catch (InterruptedException e) {
interrupted = true;
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
}
/**
* Releases the write lock on the destroy ReadWrite lock.
*/
private void releaseDestroyWriteLock() {
this.destroyLock.writeLock().unlock();
}
/**
* Returns time to wait in millis from now based on start and wait in request.
*
* @return the current wait time for the request before it times out
*/
private long calcWaitMillisFromNow(DLockRequestMessage request) {
long result = request.getTimeoutTS();
if (result != Long.MAX_VALUE) {
long now = DLockService.getLockTimeStamp(this.dlock.getDistributionManager());
result = result - now;
}
return result;
}
/**
* Shuts down the grantor thread and changes internal state to destroyed.
*/
private void makeDestroyed() {
try {
this.thread.shutdown();
this.state = DESTROYED;
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "DLockGrantor {} state is DESTROYED",
this.dlock.getName());
}
if (this.untilDestroyed.getCount() > 0) {
this.untilDestroyed.countDown();
}
if (this.whileInitializing.getCount() > 0) {
this.whileInitializing.countDown();
}
this.dlock.getDistributionManager().removeMembershipListener(this.membershipListener);
} finally {
this.dlock.getStats().incGrantors(-1);
}
}
/**
* Returns a snapshot of the current grant tokens.
* <p>
* Synchronizes on grantTokens.
*
* @return a snapshot of the current grant tokens
*/
protected Collection snapshotGrantTokens() {
Collection snapshot = null;
synchronized (this.grantTokens) {
snapshot = new ArrayList(this.grantTokens.values());
}
return snapshot;
}
/**
* Fetches or creates a new grant token for the named lock.
* <p>
* Synchronizes on grantTokens and the grant token.
*
* @param name the name of the lock
* @return the grant token for the named lock
*/
private DLockGrantToken getOrCreateGrant(Object name) {
DLockGrantToken grantToken = null;
synchronized (this.grantTokens) {
grantToken = basicGetGrantToken(name);
if (grantToken == null) { // checking isTokenDestroyed here will deadlock
grantToken = new DLockGrantToken(this.dlock, this, name);
grantToken.incAccess();
basicPutGrantToken(grantToken);
} else {
synchronized (grantToken) {
if (grantToken.isDestroyed()) {
grantToken = new DLockGrantToken(this.dlock, this, name);
grantToken.incAccess();
basicPutGrantToken(grantToken);
} else {
grantToken.incAccess();
}
}
}
}
return grantToken;
}
/**
* TEST HOOK: Returns an unmodifible collection backed by the values of the DLockGrantToken map
* for testing purposes only.
* <p>
* Synchronizes on grantTokens.
*
* @return unmodifible collection of the grant tokens
*/
public Collection getGrantTokens() {
synchronized (this.grantTokens) {
return Collections.unmodifiableCollection(this.grantTokens.values());
}
}
/**
* Remove the grant token if it is unused.
* <p>
* Synchronizes on grantTokens and the grant token.
*
* @param grant the grant token to remove
*/
protected void removeGrantIfUnused(DLockGrantToken grant) {
synchronized (this.grantTokens) {
synchronized (grant) {
if (isDestroyed() || grant.isDestroyed()) {
return;
} else if (grant.grantLockToNextRequest()) {
return;
} else if (!grant.isBeingAccessed() && !grant.isGranted(false)
&& !grant.hasWaitingRequests()) {
basicRemoveGrantToken(grant);
}
}
}
}
/**
* Iterates over grants and attempts to remove any that are no longer in use.
* <p>
* Synchronizes on grantTokens and the grant token.
*
* @param grants the grants to be checked for removal
*/
protected void removeUnusedGrants(Iterator grants) {
while (grants.hasNext()) {
DLockGrantToken grant = (DLockGrantToken) grants.next();
removeGrantIfUnused(grant);
}
}
/**
* Returns the DLockGrantToken from grant tokens map stored under the key name.
* <p>
* Synchronizes on grantTokens.
*/
public DLockGrantToken getGrantToken(Object name) {
synchronized (this.grantTokens) {
return basicGetGrantToken(name);
}
}
/**
* Fetches the grant token value stored in the map under key name.
* <p>
* Caller must synchronize on grantTokens
*
* @param name the key to fetch the grant token value for
* @return the grant token stored under key name guarded.By {@link #grantTokens}
*/
private DLockGrantToken basicGetGrantToken(Object name) {
return (DLockGrantToken) this.grantTokens.get(name);
}
/**
* Stores the grant token as a value in the map under the key of its name.
* <p>
* Caller must synchronize on grantTokens
*
* @param grantToken the grant token to store in the map guarded.By {@link #grantTokens}
*/
private void basicPutGrantToken(DLockGrantToken grantToken) {
this.grantTokens.put(grantToken.getName(), grantToken);
dlock.getStats().incGrantTokens(1);
}
/**
* Removes the grant token from the map.
* <p>
* Caller must synchronize on grantTokens and then the grantToken.
*
* @param grantToken the grant token to remove from the map. guarded.By {@link #grantTokens} and
* grantToken
*/
private void basicRemoveGrantToken(DLockGrantToken grantToken) {
Object removed = this.grantTokens.remove(grantToken.getName()); // changed to ref token
if (removed != null) {
Assert.assertTrue(removed == grantToken);
grantToken.destroy();
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.basicRemoveGrantToken] removed {}; removed={}", grantToken, removed);
}
}
}
/**
* Iterates over grants and handles any that have expired.
* <p>
* Synchronizes on each grant token.
*
* @param grants the grants to iterate over
* @return the next smallest expiration time
*/
protected long expireAndGrantLocks(Iterator grants) {
long smallestExpire = Long.MAX_VALUE;
while (grants.hasNext()) {
DLockGrantToken grant = (DLockGrantToken) grants.next();
if (grant.isDestroyed()) {
continue;
}
long expire = grant.expireAndGrantLock();
if (expire < smallestExpire) {
smallestExpire = expire;
}
}
return smallestExpire;
}
/**
* Iterates over grants and handles any that have timed out.
* <p>
* Synchronizes on each grant token.
*
* @param grants the grants to iterate over
* @return the next smallest timeout
*/
protected long handleRequestTimeouts(Iterator grants) {
long smallestTimeout = Long.MAX_VALUE;
while (grants.hasNext()) {
DLockGrantToken grant = (DLockGrantToken) grants.next();
if (grant.isDestroyed()) {
continue;
}
long timeout = grant.handleRequestTimeouts();
if (timeout < smallestTimeout) {
smallestTimeout = timeout;
}
}
return smallestTimeout;
}
/**
* TEST HOOK: Specifies time to sleep while handling suspend in order to cause a timeout.
* <p>
* Synchronizes on suspendLock.
*
*/
public void setDebugHandleSuspendTimeouts(int value) {
synchronized (suspendLock) {
debugHandleSuspendTimeouts = value;
}
}
/**
* True to enable test hook to sleep while handling suspend to cause timeout.
*
* guarded.By {@link #suspendLock}
*/
private int debugHandleSuspendTimeouts = 0;
/**
* Iterates through a copy of suspendQueue and handles any requests that have timed out.
* <p>
* Synchronizes on suspendLock.
*
* @return the next smallest timeout in the suspendQueue
*/
protected long handleSuspendTimeouts() {
long smallestTimeout = Long.MAX_VALUE;
synchronized (suspendLock) {
if (suspendQueue.isEmpty())
return smallestTimeout;
if (isDestroyed())
return smallestTimeout;
}
List timeouts = new ArrayList();
List copySuspendQueue = null;
synchronized (suspendLock) {
copySuspendQueue = new ArrayList(suspendQueue);
}
for (Iterator iter = copySuspendQueue.iterator(); iter.hasNext();) {
DLockRequestMessage req = (DLockRequestMessage) iter.next();
if (req.checkForTimeout()) { // sends DLockResponseMessage if timeout
cleanupSuspendState(req);
timeouts.add(req);
} else {
long timeout = req.getTimeoutTS();
if (timeout < smallestTimeout) {
smallestTimeout = timeout;
}
}
}
int localDebugHandleSuspendTimeouts = 0;
synchronized (suspendLock) {
localDebugHandleSuspendTimeouts = debugHandleSuspendTimeouts;
}
if (localDebugHandleSuspendTimeouts > 0) {
try {
logger.info(LogMarker.DLS_MARKER,
"debugHandleSuspendTimeouts sleeping for {}",
localDebugHandleSuspendTimeouts);
Thread.sleep(localDebugHandleSuspendTimeouts);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
if (!timeouts.isEmpty()) {
synchronized (suspendLock) {
if (writeLockWaiters > 0) {
// suspenders exist... must iterate through for safe removal
for (Iterator iter = timeouts.iterator(); iter.hasNext();) {
DLockRequestMessage req = (DLockRequestMessage) iter.next();
// attempt to remove timed out req from suspendQueue
if (suspendQueue.remove(req)) {
// request was still in suspendQueue, so check if suspender
if (req.isSuspendLockingRequest()) {
writeLockWaiters--;
}
}
} // for
} else {
// no suspenders so safe to removeAll
Assert.assertTrue(writeLockWaiters == 0,
"Grantor state writeLockWaiters changed while holding suspendLock");
suspendQueue.removeAll(timeouts);
}
checkWriteLockWaiters();
} // synchronized
}
return smallestTimeout;
}
/**
* Returns string representation for the enumerated grantor state.
*
* @param stateInt the number of the state to return a string for
* @return string representation for the enumerated grantor state
*/
private String stateToString(int stateInt) {
String stateDesc = null;
switch (stateInt) {
case INITIALIZING:
stateDesc = "INITIALIZING";
break;
case READY:
stateDesc = "READY";
break;
case DESTROYED:
stateDesc = "DESTROYED";
break;
default:
stateDesc = null;
break;
}
if (stateDesc == null) {
throw new IllegalArgumentException(String.format("Unknown state for grantor: %s",
Integer.valueOf(state)));
}
return stateDesc;
}
/**
* Throws IllegalStateException if this grantor is not still initializing.
*
* @throws IllegalStateException if this grantor is not still initializing
*/
private void assertInitializing() {
if (this.state != INITIALIZING) {
String stateDesc = stateToString(this.state);
throw new IllegalStateException(
String.format("DLockGrantor operation only allowed when initializing, not %s",
stateDesc));
}
}
/**
* Suspends locking by the remote thread and lease id.
* <p>
* Caller must synchronize on suspendLock.
*
* Concurrency: protected by synchronization of {@link #suspendLock}
*
* @param myRThread the remote thread that is about to suspend locking
* @param lockId the id of the lock request used to suspend locking
*/
protected void suspendLocking(final RemoteThread myRThread, final int lockId) {
if (DEBUG_SUSPEND_LOCK) {
Assert.assertHoldsLock(this.suspendLock, true);
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Suspend locking of {} by {} with lockId of {}",
this.dlock, myRThread, lockId);
}
Assert.assertTrue(myRThread != null, "Attempted to suspend locking for null RemoteThread");
Assert.assertTrue(this.lockingSuspendedBy == null || this.lockingSuspendedBy.equals(myRThread),
"Attempted to suspend locking for " + myRThread + " but locking is already suspended by "
+ this.lockingSuspendedBy);
this.suspendedLockId = lockId;
this.lockingSuspendedBy = myRThread;
}
/**
* Resume locking after it has been suspended.
* <p>
* Caller must synchronize on suspendLock.
*/
private void resumeLocking() {
if (DEBUG_SUSPEND_LOCK) {
Assert.assertHoldsLock(this.suspendLock, true);
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Resume locking of {}", this.dlock);
}
this.lockingSuspendedBy = null;
this.suspendedLockId = INVALID_LOCK_ID;
}
/**
* Returns true if locking has been suspended.
* <p>
* Caller must synchronize on suspendLock.
*
* Concurrency: protected by synchronization of {@link #suspendLock}
*
* @return true if locking has been suspended
*/
protected boolean isLockingSuspended() {
if (DEBUG_SUSPEND_LOCK) {
Assert.assertHoldsLock(this.suspendLock, true);
}
return this.lockingSuspendedBy != null;
}
/**
* Returns true if locking has been suspended.
* <p>
* Synchronizes on suspendLock.
*
* @return true if locking has been suspended
*/
protected boolean isLockingSuspendedWithSync() {
synchronized (this.suspendLock) {
return this.lockingSuspendedBy != null;
}
}
/**
* Returns true if locking has been suspended by the remote thread.
* <p>
* Caller must synchronize on suspendLock.
*
* Concurrency: protected by synchronization of {@link #suspendLock}
*
* @return true if locking has been suspended by the remote thread
*/
protected boolean isLockingSuspendedBy(final RemoteThread rThread) {
if (DEBUG_SUSPEND_LOCK) {
Assert.assertHoldsLock(this.suspendLock, true);
}
if (rThread == null)
return false;
return rThread.equals(this.lockingSuspendedBy);
}
String displayStatus(RemoteThread rThread, Object name) {
StringBuffer sb = new StringBuffer();
synchronized (this.suspendLock) {
sb.append(' ');
sb.append(this.toString());
sb.append(" id=" + this.hashCode());
sb.append(" rThread=" + rThread);
if (name != null) {
sb.append(" name=" + name);
}
sb.append(" permittedRequests (" + permittedRequests.size() + ")="
+ permittedRequests.toString() + "");
sb.append(" suspendedLockId = " + suspendedLockId);
sb.append(" lockingSuspendedBy = " + lockingSuspendedBy);
sb.append(" writeLockWaiters = " + writeLockWaiters);
sb.append(" totalReadLockCount = " + totalReadLockCount);
sb.append("\nsuspendQueue (" + suspendQueue.size() + ")=" + suspendQueue.toString());
// Kirk said it was ok to not log the list of readLockers to cut
// down on how much logging is done at fine level.
sb.append("\nreadLockers (" + readLockCountMap.size()
+ ")" /* + "=" + readLockCountMap.toString() */);
}
return sb.toString();
}
/**
* guarded.By {@link #suspendLock}
*/
private void postReleaseSuspendLock(RemoteThread rThread, Object lock) {
if (!isLockingSuspendedBy(rThread)) {
// hit bug related to 35749
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[postReleaseSuspendLock] locking is no longer suspended by {}", rThread);
}
return;
}
boolean resume = true;
Integer integer = (Integer) readLockCountMap.get(rThread);
int readLockCount = integer == null ? 0 : integer.intValue();
if (readLockCount == 0 && !suspendQueue.isEmpty()) {
final DLockRequestMessage nextRequest = (DLockRequestMessage) suspendQueue.getFirst();
if (nextRequest.isSuspendLockingRequest()) {
resume = false;
// final RemoteThread myRemoteThread = nextRequest.getRemoteThread();
// hand-off suspendLocking while under sync...
resumeLocking();
suspendLocking(nextRequest.getRemoteThread(), nextRequest.getLockId());
permittedRequests.add(suspendQueue.removeFirst());
writeLockWaiters--;
checkWriteLockWaiters();
}
}
if (resume) {
resumeLocking();
// drain readLocks from suspendQueue into permittedRequests queue
while (!suspendQueue.isEmpty()) {
final DLockRequestMessage nextRequest = (DLockRequestMessage) suspendQueue.getFirst();
if (nextRequest.isSuspendLockingRequest()) {
Assert.assertTrue(writeLockWaiters > 0,
"SuspendLocking request is waiting but writeLockWaiters is 0");
break;
}
RemoteThread nextRThread = nextRequest.getRemoteThread();
integer = (Integer) readLockCountMap.get(nextRThread);
readLockCount = integer == null ? 0 : integer.intValue();
readLockCount++;
readLockCountMap.put(nextRThread, Integer.valueOf(readLockCount));
totalReadLockCount++;
checkTotalReadLockCount();
permittedRequests.add(suspendQueue.removeFirst());
}
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[postReleaseSuspendLock] new status {}",
displayStatus(rThread, null));
}
}
/**
* guarded.By {@link #suspendLock}
*/
private void postReleaseReadLock(RemoteThread rThread, Object lock) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
// handle release of regular lock
// boolean permitSuspend = false;
Integer integer = (Integer) readLockCountMap.get(rThread);
int readLockCount = integer == null ? 0 : integer.intValue();
if (readLockCount < 1) {
// hit bug 35749
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[postReleaseReadLock] no locks are currently held by {}", rThread);
}
return;
}
readLockCount--;
if (readLockCount == 0) {
readLockCountMap.remove(rThread);
} else {
readLockCountMap.put(rThread, Integer.valueOf(readLockCount));
}
totalReadLockCount--;
if (totalReadLockCount < 0) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Total readlock count has dropped to {} for {}",
totalReadLockCount, this);
}
}
if (totalReadLockCount == 0 && !suspendQueue.isEmpty()) {
final DLockRequestMessage nextRequest = (DLockRequestMessage) suspendQueue.getFirst();
if (nextRequest.isSuspendLockingRequest()) {
suspendLocking(nextRequest.getRemoteThread(), nextRequest.getLockId());
writeLockWaiters--;
permittedRequests.add(suspendQueue.removeFirst());
checkWriteLockWaiters();
} else {
String s = new StringBuilder("\n (readLockCount=").append(readLockCount)
.append(", totalReadLockCount=").append(totalReadLockCount)
.append(", writeLockWaiters=").append(writeLockWaiters).append(",\nsuspendQueue=")
.append(suspendQueue).append(",\npermittedRequests=").append(permittedRequests)
.toString();
logger.warn("Released regular lock with waiting read lock: {}", s);
Assert.assertTrue(false,
String.format("Released regular lock with waiting read lock: %s",
s));
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[postReleaseReadLock] new status {}",
displayStatus(rThread, null));
}
checkTotalReadLockCount();
}
/**
* Handles post release lock tasks including tracking the current suspend locking states.
* <p>
* Synchronizes on suspendLock.
*
* @param rThread the remote thread that released the lock
* @param lock the named lock that was released
*/
protected void postReleaseLock(RemoteThread rThread, Object lock) {
Assert.assertTrue(rThread != null);
synchronized (suspendLock) {
checkDestroyed();
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[postReleaseLock] rThread={} lock={} permittedRequests={} suspendQueue={}", rThread,
lock, permittedRequests, suspendQueue);
}
if (DLockService.SUSPEND_LOCKING_TOKEN.equals(lock)) {
postReleaseSuspendLock(rThread, lock);
} else {
postReleaseReadLock(rThread, lock);
}
} // suspendLock sync
}
/**
* Departure or other codepath NOT specific to unlock requires that we cleanup suspend state that
* was already permitted to request. This needs to be invoked for both regular and suspend locks.
* <p>
* Synchronizes on suspendLock.
*
* @param request the request to cleanup after due to departure of sender
*/
protected void cleanupSuspendState(DLockRequestMessage request) {
postReleaseLock(request.getRemoteThread(), request.getObjectName());
}
/**
* Drains newly permitted requests that have been removed from suspendQueue. All requests in the
* permittedRequests queue already have permission to proceed with granting or scheduling.
* <p>
* Caller must acquire destroyReadLock. Synchronizes on suspendLock, grantTokens and each grant
* token.
*
* Concurrency: protected by {@link #destroyLock} via invoking
* {@link #acquireDestroyReadLock(long)}
*/
protected void drainPermittedRequests() {
ArrayList drain = null;
synchronized (suspendLock) {
checkDestroyed();
if (this.permittedRequests.isEmpty()) {
return;
}
drain = this.permittedRequests;
this.permittedRequestsDrain.add(drain);
this.permittedRequests = new ArrayList();
} // suspendLock sync
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[drainPermittedRequests] draining {}", drain);
}
// iterate and attempt to grantOrSchedule each request
for (Iterator iter = drain.iterator(); iter.hasNext();) {
DLockRequestMessage request = (DLockRequestMessage) iter.next();
checkDestroyed(); // destroyAndRemove should respond to all of these
try {
handlePermittedLockRequest(request); // synchronizes on grant instance
} catch (LockGrantorDestroyedException e) {
try {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"LockGrantorDestroyedException respondWithNotGrantor to {}", request);
}
request.respondWithNotGrantor();
} finally {
}
} catch (LockServiceDestroyedException e) {
try {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"LockServiceDestroyedException respondWithNotGrantor to {}", request);
}
request.respondWithNotGrantor();
} finally {
}
} catch (RuntimeException e) {
logger.error(
"Processing of postRemoteReleaseLock threw unexpected RuntimeException",
e);
request.respondWithException(e);
} finally {
}
}
synchronized (suspendLock) {
checkDestroyed();
this.permittedRequestsDrain.remove(drain);
}
}
/**
* Synchronizes on suspendLock.
*/
private boolean acquireSuspendLockPermission(DLockRequestMessage request) {
boolean permitLockRequest = false;
final RemoteThread rThread = request.getRemoteThread();
Assert.assertTrue(rThread != null);
synchronized (suspendLock) {
checkDestroyed();
if (!dm.isCurrentMember(request.getSender())) {
logger.info(LogMarker.DLS_MARKER, "Ignoring lock request from non-member: {}", request);
return false;
}
Integer integer = (Integer) readLockCountMap.get(rThread);
int readLockCount = integer == null ? 0 : integer.intValue();
boolean othersHaveReadLocks = totalReadLockCount > readLockCount;
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isLockingSuspended() || writeLockWaiters > 0 || othersHaveReadLocks) {
writeLockWaiters++;
suspendQueue.addLast(request);
this.thread.checkTimeToWait(calcWaitMillisFromNow(request), false);
checkWriteLockWaiters();
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireSuspend] added '{}' to end of suspendQueue.", request);
}
} else {
permitLockRequest = true;
suspendLocking(rThread, request.getLockId());
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireSuspendLockPermission] permitted and suspended for {}",
request);
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireSuspendLockPermission] new status permitLockRequest = {}{}",
permitLockRequest, displayStatus(rThread, null));
}
} // suspendLock sync
return permitLockRequest;
}
/**
* Synchronizes on suspendLock.
*/
private boolean acquireReadLockPermission(DLockRequestMessage request) {
boolean permitLockRequest = false;
final RemoteThread rThread = request.getRemoteThread();
Assert.assertTrue(rThread != null);
synchronized (suspendLock) {
checkDestroyed();
if (!dm.isCurrentMember(request.getSender())) {
logger.info(LogMarker.DLS_MARKER, "Ignoring lock request from non-member: %s", request);
return false;
}
Integer integer = (Integer) readLockCountMap.get(rThread);
int readLockCount = integer == null ? 0 : integer.intValue();
boolean threadHoldsLock = readLockCount > 0 || isLockingSuspendedBy(rThread);
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (!threadHoldsLock && (isLockingSuspended() || writeLockWaiters > 0)) {
suspendQueue.addLast(request);
this.thread.checkTimeToWait(calcWaitMillisFromNow(request), false);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireReadLockPermission] added {} to end of suspendQueue.", request);
}
} else {
readLockCount++;
readLockCountMap.put(rThread, Integer.valueOf(readLockCount));
totalReadLockCount++;
permitLockRequest = true;
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireReadLockPermission] permitted {}", request);
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.acquireReadLockPermission] new status threadHoldsLock = {} permitLockRequest = {}{}",
threadHoldsLock, permitLockRequest, displayStatus(rThread, null));
}
checkTotalReadLockCount();
} // suspendLock sync
return permitLockRequest;
}
/**
* Returns true if lock request has permission to proceed; else adds the request to the end of
* suspendQueue and returns false.
* <p>
* Synchronizes on suspendLock.
*
* @param request the lock request to acquire permission for
*/
private boolean acquireLockPermission(final DLockRequestMessage request) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.acquireLockPermission] {}", request);
}
boolean permitLockRequest = false;
if (request.getObjectName().equals(DLockService.SUSPEND_LOCKING_TOKEN)) {
permitLockRequest = acquireSuspendLockPermission(request);
} else {
permitLockRequest = acquireReadLockPermission(request);
}
return permitLockRequest;
}
/**
* Throws InterruptedException if local lock request exists and is interruptible or
* CancelException if DistributionManager is forcing us to cancel for shutdown.
*
* @param e the throwable that caused this check
*/
private void throwIfInterruptible(InterruptedException e) throws InterruptedException {
// This needs to be first, otherwise user gets a complaint that
// the TV is off when the problem is the house is burning down...
this.dm.getCancelCriterion().checkCancelInProgress(e);
if (this.dlock.isInterruptibleLockRequest()) {
throw e;
}
}
/**
* TEST HOOK: Logs all grant tokens and other lock information for this service at INFO level.
* <p>
* Synchronizes on grantTokens.
*/
protected void dumpService() {
synchronized (this.grantTokens) {
StringBuffer buffer = new StringBuffer();
buffer.append("DLockGrantor.dumpService() for ").append(this);
buffer.append("\n").append(this.grantTokens.size()).append(" grantTokens\n");
for (Iterator iter = this.grantTokens.entrySet().iterator(); iter.hasNext();) {
Map.Entry entry = (Map.Entry) iter.next();
buffer.append(" ").append(entry.getKey()).append(": ");
DLockGrantToken token = (DLockGrantToken) entry.getValue();
buffer.append(token.toString()).append("\n");
}
logger.info(LogMarker.DLS_MARKER, "{}", buffer);
logger.info(LogMarker.DLS_MARKER, "{}",
"\nreadLockCountMap:\n" + readLockCountMap);
}
}
/**
* Verify the waiters (for debugging)
*
* guarded.By {@link #suspendLock}
*/
private void checkWriteLockWaiters() {
if (!DEBUG_SUSPEND_LOCK) {
return;
}
Assert.assertHoldsLock(this.suspendLock, true);
int result = 0;
Iterator it = this.suspendQueue.iterator();
while (it.hasNext()) {
DLockRequestMessage r = (DLockRequestMessage) it.next();
if (r.isSuspendLockingRequest()) {
result++;
}
} // while
Assert.assertTrue(result == this.writeLockWaiters);
}
/**
* Debugging method
*
* guarded.By {@link #suspendLock}
*/
private void checkTotalReadLockCount() {
if (!DEBUG_SUSPEND_LOCK) {
return;
}
Assert.assertHoldsLock(this.suspendLock, true);
int result = 0;
Iterator it = readLockCountMap.values().iterator();
while (it.hasNext()) {
result += ((Integer) it.next()).intValue();
}
Assert.assertTrue(result == totalReadLockCount);
}
// -------------------------------------------------------------------------
// DLockGrantToken (static inner class)
// -------------------------------------------------------------------------
/**
* Handles leasing and queued scheduling for an individual distributed lock.
*/
public static class DLockGrantToken {
/**
* DLS which contains this lock. Reference is used for stats and lifecycle.
*/
private final DLockService dlock;
/**
* Grantor instance that handles leasing of this lock.
*/
private final DLockGrantor grantor;
/**
* The uniquely identifying object name for this lock
*/
private final Object lockName;
/**
* Pending requests queued up for the lock
*
* guarded.By this
*/
private LinkedList pendingRequests;
/**
* The reply processor id is used to identify the specific lock operation used by the lessee to
* lease this lock
*
* guarded.By this
*/
private int leaseId = -1;
/**
* Distributed member that currently has a lease on this lock
*
* guarded.By this
*/
private InternalDistributedMember lessee;
/**
* Absolute time in milliseconds when the current lease will expire. When this lock is not
* leased out, the value is -1. When the lock is leased out, the value is > 0. A value of
* Long.MAX_VALUE indicates a non-expiring (infinite) lease.
*
* guarded.By this
*/
private long leaseExpireTime = -1;
/**
* Current count of threads attempting to access this grant token.
*
* guarded.By this
*/
private int accessCount = 0;
/**
* True if this token has been destroyed and removed from usage.
*
* guarded.By this
*/
private boolean destroyed = false;
/**
* RemoteThread identity of thread currently holding lease on this lock
*
* guarded.By this
*/
private RemoteThread lesseeThread = null;
/**
* Instatiates a new instance of DLockGrantToken.
*
* @param dlock the lock service scope for this lock
* @param grantor the grantor handling locks for the lock service
* @param name the name of this lock
*/
protected DLockGrantToken(DLockService dlock, DLockGrantor grantor, Object name) {
this.lockName = name;
this.dlock = dlock;
this.grantor = grantor;
}
/**
* Schedules the lock request for immediate or later granting of lock. This will grant the lock
* if it is available, otherwise it will add the request at the end of the pending requests
* queue.
* <p>
* Synchronizes on this grant token.
*
* @param request the request to grant or schedule
* @return true if the lock request was immediately granted
*/
protected synchronized boolean schedule(DLockRequestMessage request) {
if (!this.grantor.dm.isCurrentMember(request.getSender())) {
this.grantor.cleanupSuspendState(request);
return false;
}
if (!isGranted(false) && !hasWaitingRequests()) {
// don't need to schedule... just grant it
if (grantLockToRequest(request)) {
return true;
}
}
// add the request to the sorted set...
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantToken.schedule] {} scheduling: {}", this,
request);
}
if (this.pendingRequests == null) {
this.pendingRequests = new LinkedList();
this.dlock.getStats().incRequestQueues(1);
}
this.pendingRequests.add(request);
this.dlock.getStats().incPendingRequests(1);
return true;
}
/**
* Sends NOT_GRANTOR replies to every request waiting for this grant token and then destroys the
* grant token.
* <p>
* Synchronizes on this grant token.
*/
protected synchronized void handleGrantorDestruction() {
try {
if (this.pendingRequests != null) {
for (Iterator iter = this.pendingRequests.iterator(); iter.hasNext();) {
DLockRequestMessage request = (DLockRequestMessage) iter.next();
request.respondWithNotGrantor();
}
}
} finally {
destroy();
}
}
/**
* Checks current lock for expiration and attempts to grant the lock if it is available.
* <p>
* Synchronizes on this grant token.
* <p>
* NOTE: expiration is only as accurate as clock synchronization on the hardware that the
* members are running on probably should have Requestors handle expirations and send Release
* msg - need an Evictor thread in each Requestor
*
* @return the lease expiration time in millis for the currently held lock or Long.MAX_VALUE if
* lock has no owner
*/
protected synchronized long expireAndGrantLock() {
// isGranted calls checkForExpiration...
if (this.grantor.isDestroyed())
return Long.MAX_VALUE;
if (!isGranted(true) && !this.grantor.isLockingSuspendedWithSync()) {
grantLockToNextRequest();
}
long result = getLeaseExpireTime();
if (result <= 0) {
result = Long.MAX_VALUE;
}
return result;
}
/**
* Returns true if there are pending requests waiting to lock this grant token.
* <p>
* Caller must synchronize on this grant token.
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @return true if there are pending requests waiting to lock this
*/
protected synchronized boolean hasWaitingRequests() {
if (this.pendingRequests == null)
return false;
return !this.pendingRequests.isEmpty();
}
/**
* Grant this lock to the request if possible. Returns true if lock was granted to the request.
* <p>
* Synchronizes on this grant token.
*
* @param request the lock request asking for this lock
* @return true if lock was granted to the request
*/
protected synchronized boolean grantLockToRequest(DLockRequestMessage request) {
Assert.assertTrue(request.getRemoteThread() != null);
if (isGranted(true) || hasWaitingRequests()) {
return false;
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantToken.grantLockToRequest] granting: {}",
request);
}
long newLeaseExpireTime = grantAndRespondToRequest(request);
if (newLeaseExpireTime == -1)
return false;
if (newLeaseExpireTime < Long.MAX_VALUE) {
long now = DLockService.getLockTimeStamp(this.grantor.dm);
this.grantor.thread.checkTimeToWait(newLeaseExpireTime - now, true);
}
return true;
}
/**
* Called to release a remote lock when processing a DLockReleaseMessage.
* <p>
* Caller must synchronize on this grant token.
* <p>
* Call stack: DLockReleaseMessage -> releaseIfLocked -> getAndReleaseGrantIfLockedBy ->
* grant.releaseIfLockedBy
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @param owner the member to release the lock for
* @param lockId the lock id that the member used to acquire the lock
*/
protected void releaseIfLockedBy(InternalDistributedMember owner, int lockId) {
final RemoteThread rThread = getRemoteThread();
boolean released = false;
try {
released = releaseLock(owner, lockId);
} catch (IllegalStateException e) {
this.dlock.checkDestroyed();
this.grantor.checkDestroyed();
// must have hit race... grantor doesn't have the token
return;
}
if (released) {
// don't bother synchronizing requests for this log statement...
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
synchronized (this) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.releaseIfLockedBy] pending requests: {}",
(this.pendingRequests == null ? "none" : "" + this.pendingRequests.size()));
}
}
Assert.assertTrue(rThread != null);
// releaseIfLockedBy (remote unlock)
this.grantor.postReleaseLock(rThread, getName());
// note: DLockReleaseMessage calls drainPermittedRequests next...
}
}
/**
* Returns true if lock is currently leased by the owner with the specified lock id.
* <p>
* Caller must synchronize on this grant token.
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @param owner the member to check for lock ownership
* @param lockId the lock id that the member used for locking
* @return true if lock is currently leased by the owner with the specified lock id
*/
protected boolean isLockedBy(InternalDistributedMember owner, int lockId) {
return isLeaseHeldBy(owner, lockId);
}
/**
* Handle timeouts for requests waiting on this lock. Any requests that have timed out will be
* removed. Calculates and returns the next smallest timeout of the requests still waiting on
* this lock.
* <p>
* Synchronizes on this grant token.
*
* @return next smallest timeout of the requests still waiting on this lock
*/
protected long handleRequestTimeouts() {
long smallestTimeout = Long.MAX_VALUE;
synchronized (this) {
if (this.pendingRequests == null)
return smallestTimeout;
if (this.grantor.isDestroyed())
return smallestTimeout;
}
List timeouts = new ArrayList();
// narrow timeouts to just contain requests that have timed out...
DLockRequestMessage req = null;
// ... copyRequests is synchronized on this ...
synchronized (this) {
for (Iterator iter = this.pendingRequests.iterator(); iter.hasNext();) {
req = (DLockRequestMessage) iter.next();
if (req.checkForTimeout()) { // sends DLockResponseMessage if timeout
this.grantor.cleanupSuspendState(req);
timeouts.add(req);
} else {
long timeout = req.getTimeoutTS();
if (timeout < smallestTimeout) {
smallestTimeout = timeout;
}
}
}
removeRequests(timeouts);
}
return smallestTimeout;
}
/**
* Cleans up any state for the departed member. If the lock is held by this member, it will be
* released. Any pending lock requests for this member will be removed.
* <p>
* Synchronizes on this grant token, suspendLock, and grantTokens.
*
* @param member the departed member
*/
protected void handleDepartureOf(final InternalDistributedMember member,
final ArrayList grantsToRemoveIfUnused) {
boolean released = false;
RemoteThread rThread = null;
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
try {
synchronized (this) {
try {
if (isDestroyed())
return;
if (this.pendingRequests == null)
return;
// remove member from pendingRequests...
DLockRequestMessage req = null;
for (Iterator iter = this.pendingRequests.iterator(); iter.hasNext();) {
req = (DLockRequestMessage) iter.next();
if (member.equals(req.getSender())) {
// found departed member, respondWithNotHolder to end dlock stats
try {
req.handleDepartureOfSender();
// cleanup suspend state for this request
this.grantor.cleanupSuspendState(req);
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.handleDepartureOf] ignored cancellation (1)");
}
}
// remove the request
iter.remove();
this.dlock.getStats().incPendingRequests(-1);
}
}
} finally {
synchronized (this) {
// bugfix 32657 release lock AFTER removing member from queued requests
// because release will grant to first request in queued requests
rThread = getRemoteThread();
boolean releasedToken = false;
try {
releasedToken = releaseLock(member, getLockId());
} catch (IllegalStateException e) {
this.dlock.checkDestroyed();
this.grantor.checkDestroyed();
// must have hit race... grantor doesn't have the token
return;
}
if (releasedToken) {
released = true;
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.handleDepartureOf] pending requests: {}",
(this.pendingRequests == null ? "none" : "" + this.pendingRequests.size()));
}
Assert.assertTrue(rThread != null);
}
}
}
}
} finally {
if (released) {
try {
this.grantor.postReleaseLock(rThread, getName());
} catch (CancelException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.handleDepartureOf] ignored cancellation (2)");
}
}
this.grantor.drainPermittedRequests(); // destroyReadLock{grant{}, suspendLock{}}
grantsToRemoveIfUnused.add(this);
}
}
}
/**
* Adds this grant to the list if it references the departed member.
* <p>
* Synchronizes on this grant token.
*
* @param member the departed member
* @param grantsReferencingMember list to add grant to if it references departed member
*/
protected synchronized void checkDepartureOf(final InternalDistributedMember member,
final List grantsReferencingMember) {
if (this.destroyed) {
return;
}
if (member.equals(this.lessee)) {
grantsReferencingMember.add(this);
return;
}
if (this.pendingRequests != null) {
DLockRequestMessage req = null;
for (Iterator iter = this.pendingRequests.iterator(); iter.hasNext();) {
req = (DLockRequestMessage) iter.next();
if (member.equals(req.getSender())) {
grantsReferencingMember.add(this);
return;
}
}
}
}
/**
* Remove all the specified pending requests.
* <p>
* Caller must synchronize on this grant token.
*
* @param requestsToRemove the pending requests to remove guarded.By this
*/
private void removeRequests(Collection requestsToRemove) {
if (!requestsToRemove.isEmpty()) {
synchronized (this) {
this.pendingRequests.removeAll(requestsToRemove);
}
this.dlock.getStats().incPendingRequests(-requestsToRemove.size());
}
}
/**
* Grants this lock to the next waiting request if one exists.
* <p>
* Caller must synchronize on this grant token.
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @return true if the lock was granted to next request
*/
protected boolean grantLockToNextRequest() {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.grantLock] {} isGranted={} hasWaitingRequests={}", getName(),
isLeaseHeld(), hasWaitingRequests());
}
while (!isGranted(true) && hasWaitingRequests()) {
try {
// get request at front of queue...
DLockRequestMessage request = null;
synchronized (this) {
request = (DLockRequestMessage) this.pendingRequests.remove(0);
}
this.dlock.getStats().incPendingRequests(-1);
// grant lock to the request unless it is timed out...
if (request.checkForTimeout()) {
this.grantor.cleanupSuspendState(request);
continue;
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantToken.grantLock] granting {} to {}",
getName(), request.getSender());
}
long newLeaseExpireTime = grantAndRespondToRequest(request);
if (newLeaseExpireTime == -1)
continue;
if (newLeaseExpireTime < Long.MAX_VALUE) {
long now = DLockService.getLockTimeStamp(this.grantor.dm);
this.grantor.thread.checkTimeToWait(newLeaseExpireTime - now, true);
}
} catch (IndexOutOfBoundsException e) {
// ignore... entry may have timed out between empty check and remove
}
}
return isGranted(false);
}
/**
* Grants the lock to the specified request and sends a reply to the member that initiated the
* request.
* <p>
* Caller must synchronize on this grant token.
*
* @param request the request to grant the lock to
* @return leaseExpireTime or -1 if failed to grant. guarded.By this
*/
private long grantAndRespondToRequest(DLockRequestMessage request) {
synchronized (request) {
if (request.respondedNoSync()) {
return -1;
}
Assert.assertTrue(request.getRemoteThread() != null);
if (!this.grantor.dm.isCurrentMember(request.getSender())) {
this.grantor.cleanupSuspendState(request);
return -1;
}
if (isSuspendLockingToken()) {
synchronized (this.grantor.suspendLock) {
Assert.assertTrue(
this.grantor.lockingSuspendedBy == null
|| this.grantor.isLockingSuspendedBy(request.getRemoteThread()),
"Locking is suspended by " + this.grantor.lockingSuspendedBy + " with lockId of "
+ this.grantor.suspendedLockId + " instead of " + request.getRemoteThread()
+ " with lockId of " + request.getLockId());
} // suspendLock sync
}
long newLeaseExpireTime = calcLeaseExpireTime(request.getLeaseTime());
grantLock(request.getSender(), newLeaseExpireTime, request.getLockId(),
request.getRemoteThread());
if (isSuspendLockingToken()) {
synchronized (this.grantor.suspendLock) {
// no-op if already suspend by this RemoteThread...
this.grantor.suspendLocking(request.getRemoteThread(), request.getLockId());
Assert.assertTrue(this.grantor.isLockingSuspendedBy(request.getRemoteThread()),
"Locking should now be suspended by " + request.getRemoteThread()
+ " with lockId of " + request.getLockId() + " instead of "
+ this.grantor.lockingSuspendedBy + " with lockId of "
+ this.grantor.suspendedLockId);
} // suspendLock sync
}
// NOTE: if grantor is local and client interrupts the request, the
// following will release the lock because the reply processor is gone
request.respondWithGrant(newLeaseExpireTime);
if (!isLeaseHeldBy(request.getSender(), request.getLockId())) {
// lock request was local and interrupted then released
return -1;
}
return newLeaseExpireTime;
}
}
/**
* Returns the absolute time at which the specified lease time will expire from now. This call
* does not change or check any state other than current time.
*
* @param leaseTime the desired length of lease time
* @return the absolute time at which the lease will expire
*/
protected long calcLeaseExpireTime(long leaseTime) {
if (leaseTime == Long.MAX_VALUE || leaseTime == -1) {
return Long.MAX_VALUE;
}
long currentTime = getCurrentTime();
long newLeaseExpireTime = currentTime + leaseTime;
if (newLeaseExpireTime < leaseTime) { // rolled over MAX_VALUE...
newLeaseExpireTime = Long.MAX_VALUE;
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.calcLeaseExpireTime] currentTime={} newLeaseExpireTime={}",
currentTime, newLeaseExpireTime);
}
return newLeaseExpireTime;
}
/**
* Returns true if this grant token is currently granted.
* <p>
* Caller must synchronize on this grant token.
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @param checkForExpiration true if expiration should be attempted before checking if this
* grant token is currently granted
* @return true if this grant token is currently granted
*/
protected boolean isGranted(boolean checkForExpiration) {
if (checkForExpiration) {
checkForExpiration();
}
return isLeaseHeld();
}
/**
* Creates a string of the pending requests for logging or debugging.
* <p>
* Caller must synchronize on this grant token.
*
* @return a string of the pending requests for logging or debugging guarded.By this
*/
private String pendingRequestsToString() {
if (this.pendingRequests == null) {
return "(null)";
}
StringBuffer sb = new StringBuffer();
Iterator it = this.pendingRequests.iterator();
while (it.hasNext()) {
Object req = it.next();
sb.append("[");
sb.append(req.toString());
sb.append("]");
}
return sb.toString();
}
/**
* Returns string representation of this object.
* <p>
* Synchronizes on this grant token.
*/
@Override
public String toString() {
return toString(true);
}
/**
* Returns string representation of this object.
* <p>
* Synchronizes on this grant token.
* <p>
*
* @param displayPendingRequests true if string should include pendingRequests
*/
public String toString(boolean displayPendingRequests) {
StringBuffer sb = new StringBuffer("DLockGrantToken");
sb.append("@").append(Integer.toHexString(hashCode()));
synchronized (this) {
sb.append(" {name: ").append(getName());
sb.append(", isGranted: ").append(isLeaseHeld());
sb.append(", isDestroyed: ").append(this.destroyed);
sb.append(", accessCount: ").append(this.accessCount);
sb.append(", lessee: ").append(this.lessee);
sb.append(", leaseExpireTime: ").append(this.leaseExpireTime);
sb.append(", leaseId: ").append(this.leaseId);
sb.append(", lesseeThread: ").append(this.lesseeThread);
if (displayPendingRequests) {
sb.append(", pendingRequests: ").append(pendingRequestsToString());
}
sb.append("}");
}
return sb.toString();
}
/**
* Returns the name of this lock.
*
* @return the name of this lock
*/
Object getName() {
return this.lockName;
}
/**
* Returns true if this lock represents suspend locking.
*
* @return true if this lock represents suspend locking
* @see org.apache.geode.distributed.DistributedLockService#suspendLocking(long)
*/
boolean isSuspendLockingToken() {
return DLockService.SUSPEND_LOCKING_TOKEN.equals(this.lockName);
}
/**
* Returns the lock id used to lease this lock.
* <p>
* Caller must synchronize on this grant token.
*
* @return the lock id used to lease this lock guarded.By this
*/
int getLockId() {
return this.leaseId;
}
/**
* Returns the identity of the thread that has this lock leased.
* <p>
* Caller must synchronize on this grant token.
*
* @return the identity of the thread that has this lock leased guarded.By this
*/
RemoteThread getRemoteThread() {
return this.lesseeThread;
}
/**
* Increments or decrements access count by the specified amount.
* <p>
* Synchronizes on this grant token.
*
* @param amount the amount to inc or dec access count by
*/
private synchronized void incAccess(int amount) {
if (amount < 0) {
Assert.assertTrue(this.accessCount - amount >= 0,
amount + " cannot be subtracted from accessCount " + this.accessCount);
}
this.accessCount += amount;
}
/**
* Increments the access count by one.
* <p>
* Synchronizes on this grant token.
*/
void incAccess() {
incAccess(1);
}
/**
* Decrements the access count by one.
* <p>
* Synchronizes on this grant token.
*/
void decAccess() {
incAccess(-1);
}
/**
* Returns true if the access count is greater than zero.
* <p>
* Synchronizes on this grant token.
*
* @return true if the access count is greater than zero
*/
boolean isBeingAccessed() {
synchronized (this) {
return this.accessCount > 0;
}
}
/**
* Returns the member that currently holds a lease on this lock.
* <p>
* Synchronizes on this grant token.
*
* @return the member that currently holds a lease on this lock
*/
public synchronized InternalDistributedMember getOwner() {
return this.lessee;
}
/**
* Returns the lease expiration time. This the absolute time in milliseconds when the current
* lease will expire.
* <p>
* Caller must synchronize on this grant token.
*
* Concurrency: protected by synchronization of *this* DLockGrantToken
*
* @return the lease expiration time
*/
public long getLeaseExpireTime() {
return this.leaseExpireTime;
}
/**
* Returns true if this grant token has been destroyed.
* <p>
* Synchronizes on this grant token.
*
* @return true if this grant token has been destroyed
*/
public synchronized boolean isDestroyed() {
return this.destroyed;
}
/**
* Returns the current time in milliseconds.
*
* @return the current time in milliseconds
*/
long getCurrentTime() {
return DLockService.getLockTimeStamp(this.grantor.dm);
}
/**
* Handle expiration if the lease expire time has been reached for the current lease on this
* grant token.
* <p>
* Synchronizes on this grant token.
*
* @return true if the lease is expired
*/
synchronized boolean checkForExpiration() {
if (this.lessee != null && this.leaseId > -1) {
if (this.leaseExpireTime == Long.MAX_VALUE)
return false;
long currentTime = getCurrentTime();
if (currentTime > this.leaseExpireTime) {
// expired!
final RemoteThread rThread = this.lesseeThread;
this.lessee = null;
this.leaseId = -1;
this.lesseeThread = null;
this.leaseExpireTime = -1;
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[checkForExpiration] Expired token at {}: {}",
currentTime, toString(true));
}
this.grantor.postReleaseLock(rThread, this.lockName);
return true;
}
}
return false;
}
/**
* Grants this lock.
* <p>
* Caller must synchronize on this grant token.
*
* @param owner the member that is being granted the lock
* @param newLeaseExpireTime the absolute expiration time
* @param lockId the lock id used to request the lock
* @param remoteThread identity of the locking thread guarded.By this
*/
void grantLock(InternalDistributedMember owner, long newLeaseExpireTime, int lockId,
RemoteThread remoteThread) {
Assert.assertTrue(remoteThread != null);
checkDestroyed();
basicGrantLock(owner, newLeaseExpireTime, lockId, remoteThread);
}
/**
* Modify grant token state to mark the lock as granted.
* <p>
* Caller must synchronize on this grant token.
*
* @param owner the member that has been granted the lock
* @param newLeaseExpireTime the absolute expiration time
* @param lockId the lock id used to request the lock
* @param remoteThread identity of the locking thread guarded.By this
*/
private void basicGrantLock(InternalDistributedMember owner, long newLeaseExpireTime,
int lockId, RemoteThread remoteThread) {
Assert.assertTrue(remoteThread != null);
Assert.assertTrue(lockId > -1, "Invalid attempt to grant lock with lockId " + lockId);
this.lessee = owner;
this.leaseExpireTime = newLeaseExpireTime;
this.leaseId = lockId;
this.lesseeThread = remoteThread;
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantToken.grantLock.grantor] Granting {}",
toString(false));
}
}
/**
* Returns true if this lock is currently leased out.
* <p>
* Caller must synchronize on this grant token.
*
* @return true if this lock is currently leased out guarded.By this
*/
boolean isLeaseHeld() {
return this.lessee != null && this.leaseId > -1;
}
/**
* Mark this grant token as destroyed. This should only happen to a token that is no longer in
* use.
* <p>
* Caller must synchronize on this grant token. guarded.By this
*/
void destroy() {
if (!this.destroyed) {
this.destroyed = true;
this.dlock.getStats().incGrantTokens(-1);
if (this.pendingRequests != null) {
this.dlock.getStats().incPendingRequests(-this.pendingRequests.size());
this.dlock.getStats().incRequestQueues(-1);
}
}
}
/**
* Throws IllegalStateException if this grant token has been destroyed.
* <p>
* Caller must synchronize on this grant token.
*
* @throws IllegalStateException if this grant token has been destroyed guarded.By this
*/
private void checkDestroyed() {
if (this.destroyed) {
String s = "Attempting to use destroyed grant token: " + this;
IllegalStateException e = new IllegalStateException(s);
throw e;
}
}
/**
* Called by the grantor. Releases lock on this token if it is currently locked by the specified
* member and lockId.
* <p>
* Caller must synchronize on this grant token.
*
* @param member the member to release the lock from
* @param lockId the lock id that the member used when locking
* @return true if lock was released guarded.By this
*/
private boolean releaseLock(InternalDistributedMember member, int lockId) {
if (lockId == -1)
return false;
checkDestroyed();
if (isLeaseHeldBy(member, lockId)) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.releaseLock] releasing ownership: {}", this);
}
this.lessee = null;
this.leaseId = -1;
this.lesseeThread = null;
this.leaseExpireTime = -1;
return true;
}
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantToken.releaseLock] {} attempted to release: {}", member, this);
}
return false;
}
/**
* Returns true if the sender holds a lease on this lock using lockId.
* <p>
* Caller must synchronize on this grant token.
*
* @param sender the member that potentially holds a lease
* @param lockId the lock id provided by the member
* @return true if the sender holds a lease on this lock guarded.By this
*/
private boolean isLeaseHeldBy(InternalDistributedMember sender, int lockId) {
Assert.assertTrue(sender != null, "sender is null: " + this);
Assert.assertTrue(lockId > -1, "lockId is < 0: " + this);
return sender.equals(this.lessee) && lockId == this.leaseId;
}
}
// -------------------------------------------------------------------------
// DLockGrantorThread (static inner class)
// -------------------------------------------------------------------------
/**
* Thread dedicated to handling background tasks for this grantor.
*/
private static class DLockGrantorThread extends LoggingThread {
private static final long MAX_WAIT = 60 * 1000; // 60 seconds...
private volatile boolean shutdown = false;
private boolean waiting = false;
private boolean requireTimeToWait = false;
private boolean goIntoWait = false;
private long timeToWait = MAX_WAIT;
private long expectedWakeupTimeStamp = 0;
private final Object lock = new Object();
private final DLockGrantor grantor;
private final CancelCriterion stopper;
/** Time in millis that next pending request will timeout */
private long nextTimeout = DLockGrantorThread.MAX_WAIT;
/** Time in millis that next lock is due to expire */
private long nextExpire = DLockGrantorThread.MAX_WAIT;
DLockGrantorThread(DLockGrantor grantor, CancelCriterion stopper) {
super("Lock Grantor for " + grantor.dlock.getName());
this.grantor = grantor;
this.stopper = stopper;
}
private long now() {
DistributionManager dm = this.grantor.dlock.getDistributionManager();
return DLockService.getLockTimeStamp(dm);
}
protected void shutdown() {
this.shutdown = true;
this.interrupt();
}
protected void checkTimeToWait(long newTimeToWaitArg, boolean expire) {
long newTimeToWait = newTimeToWaitArg;
if (newTimeToWait == Long.MAX_VALUE) {
// never expire
return;
} else if (newTimeToWait < 0) {
// negative means already expired or timed out so we wakeup immediately
newTimeToWait = 0;
}
synchronized (this.lock) {
if (expire && newTimeToWait < this.nextExpire) {
this.nextExpire = newTimeToWait;
}
if (!expire && newTimeToWait < this.nextTimeout) {
this.nextTimeout = newTimeToWait;
}
if (newTimeToWait < this.timeToWait) {
if (this.waiting) {
long newWakeupTimeStamp = now() + newTimeToWait;
if (newWakeupTimeStamp > -1 // accounts for overflow
&& newWakeupTimeStamp < this.expectedWakeupTimeStamp) {
this.timeToWait = newTimeToWait;
this.requireTimeToWait = true;
this.goIntoWait = true;
this.lock.notify();
}
} else {
this.timeToWait = newTimeToWait;
this.requireTimeToWait = true;
}
} // end if newTimeToWait
} // end sync this.lock
}
@Override
public void run() {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
DistributedLockStats stats = this.grantor.dlock.getStats();
boolean recalcTimeToWait = false;
while (!this.shutdown) {
if (stopper.isCancelInProgress()) {
break; // done
}
try {
// go into wait if we know we have no timeouts or expires for a while
synchronized (this.lock) { // synchronized
if (recalcTimeToWait || this.requireTimeToWait) {
recalcTimeToWait = false;
long nextTS = Math.min(this.nextExpire, this.nextTimeout);
this.nextExpire = Long.MAX_VALUE;
this.nextTimeout = Long.MAX_VALUE;
if (nextTS != Long.MAX_VALUE || this.requireTimeToWait) {
this.requireTimeToWait = false;
long now = now();
// fix bug 39355 by using current timeToWait if smaller
long newTimeToWait = nextTS - now;
if (this.requireTimeToWait) {
this.timeToWait = Math.min(this.timeToWait, newTimeToWait);
} else {
this.timeToWait = newTimeToWait;
}
if (this.timeToWait < 0)
this.timeToWait = 0;
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread will wait for {} ms. nextExpire={} nextTimeout={} now={}",
this.timeToWait, this.nextExpire, this.nextTimeout, now);
}
} else {
this.timeToWait = Long.MAX_VALUE;
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread will wait until rescheduled.");
}
}
}
if (this.timeToWait > 0) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread is about to wait for {} ms.", this.timeToWait);
}
if (this.timeToWait != Long.MAX_VALUE) {
this.expectedWakeupTimeStamp = now() + this.timeToWait;
if (this.expectedWakeupTimeStamp < 0) {
// overflow
this.expectedWakeupTimeStamp = Long.MAX_VALUE;
}
} else {
this.expectedWakeupTimeStamp = Long.MAX_VALUE;
}
if (this.expectedWakeupTimeStamp == Long.MAX_VALUE) {
while (!this.goIntoWait) {
this.waiting = true;
this.lock.wait(); // spurious wakeup ok
this.waiting = false;
}
} else {
long timeToWaitThisTime = this.timeToWait;
for (;;) {
this.waiting = true;
this.lock.wait(timeToWaitThisTime); // spurious wakeup ok
this.waiting = false;
if (this.goIntoWait)
break; // out of for loop
timeToWaitThisTime = this.expectedWakeupTimeStamp - now();
if (timeToWaitThisTime <= 0)
break; // out of for loop
}
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "DLockGrantorThread has woken up...");
}
if (this.shutdown)
break;
// if goIntoWait, continue back around and enter wait again
if (this.goIntoWait) {
this.goIntoWait = false;
continue;
}
}
} // synchronized
long statStart = stats.startGrantorThread();
try {
Collection grants = this.grantor.snapshotGrantTokens();
// TASK: expire and grant locks
if (this.shutdown) {
return;
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread about to expireAndGrantLocks...");
}
{
long smallestExpire = this.grantor.expireAndGrantLocks(grants.iterator());
synchronized (this.lock) {
if (smallestExpire < this.nextExpire) {
this.nextExpire = smallestExpire;
}
}
}
long timing = stats.endGrantorThreadExpireAndGrantLocks(statStart);
// TASK: timeout waiting requests
if (this.shutdown) {
return;
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread about to handleRequestTimeouts...");
}
{
long smallestRequestTimeout = this.grantor.handleRequestTimeouts(grants.iterator());
long smallestSuspendTimeout = this.grantor.handleSuspendTimeouts();
synchronized (this.lock) {
if (smallestRequestTimeout < this.nextTimeout) {
this.nextTimeout = smallestRequestTimeout;
}
if (smallestSuspendTimeout < this.nextTimeout) {
this.nextTimeout = smallestSuspendTimeout;
}
}
}
timing = stats.endGrantorThreadHandleRequestTimeouts(timing);
// TASK: remove unused tokens
if (this.shutdown) {
return;
}
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"DLockGrantorThread about to removeUnusedGrants...");
}
this.grantor.removeUnusedGrants(grants.iterator());
stats.endGrantorThreadRemoveUnusedTokens(timing);
} catch (CancelException e) {
// so, exit then.
} finally {
recalcTimeToWait = true;
stats.endGrantorThread(statStart);
}
} catch (LockGrantorDestroyedException ex) {
this.shutdown = true;
return;
} catch (InterruptedException e) {
// shutdown probably interrupted us
// Not necessary to reset the interrupt bit, we're going to go
// away of our own accord.
if (this.shutdown) {
// ok to ignore since this thread will now shutdown
} else {
logger.warn("DLockGrantorThread was unexpectedly interrupted",
e);
// do not set interrupt flag since this thread needs to resume
stopper.checkCancelInProgress(e);
}
}
}
}
}
// -------------------------------------------------------------------------
// MembershipListener inner classes
// -------------------------------------------------------------------------
/** Detects loss of the lock grantor and initiates grantor recovery. */
private MembershipListener membershipListener = new MembershipListener() {
@Override
public void memberJoined(DistributionManager distributionManager,
InternalDistributedMember id) {}
@Override
public void quorumLost(DistributionManager distributionManager,
Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {}
@Override
public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
InternalDistributedMember whoSuspected, String reason) {}
@Override
public void memberDeparted(DistributionManager distMgr, final InternalDistributedMember id,
final boolean crashed) {
final DLockGrantor me = DLockGrantor.this;
// if the VM is being forcibly disconnected, we shouldn't release locks as it
// will take longer than the time allowed by the InternalDistributedSystem
// shutdown mechanism.
if (distMgr.getCancelCriterion().isCancelInProgress()) {
return;
}
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
try {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.memberDeparted] waiting thread pool will process id={}", id);
}
distMgr.getExecutors().getWaitingThreadPool().execute(new Runnable() {
@Override
public void run() {
try {
processMemberDeparted(id, crashed, me);
} catch (InterruptedException e) {
// ignore
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE, "Ignored interrupt processing departed member");
}
}
}
});
} catch (RejectedExecutionException e) {
if (isDebugEnabled_DLS) {
logger.trace(LogMarker.DLS_VERBOSE,
"[DLockGrantor.memberDeparted] rejected handling of id={}", id);
}
}
}
protected void processMemberDeparted(InternalDistributedMember id, boolean crashed,
DLockGrantor me) throws InterruptedException {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.processMemberDeparted] id={}", id);
}
try {
me.waitWhileInitializing();
// one cause of bug 32657 is "if (crashed) {" around handleDepartureOf
// ... we cannot rely on the value of crashed to determine if grantor
// has or will receive a NonGrantorDestroy message
me.handleDepartureOf(id);
} catch (LockGrantorDestroyedException e) {
// ignore... grantor was destroyed
} // outer try-catch
}
};
}