/*
 * 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.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;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * 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(
          GeodeGlossary.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
   */
  final Map<Object, DLockBatch> 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) {
    dm = dlock.getDistributionManager();
    CancelCriterion stopper = dm.getCancelCriterion();
    whileInitializing = new StoppableCountDownLatch(stopper, 1);
    untilDestroyed = new StoppableCountDownLatch(stopper, 1);
    this.dlock = dlock;
    destroyLock = new StoppableReentrantReadWriteLock(stopper);
    versionId.set(vId);
    dm.addMembershipListener(membershipListener);
    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 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) {
    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 && dlock.isInterruptibleLockRequest()) {
        throw new InterruptedException();
      }
      while (true) {
        try {
          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 {
        untilDestroyed.await();
        break;
      } catch (InterruptedException e) {
        interrupted = true;
        throwIfInterruptible(e);
      } finally {
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
    }
  }

  @Override
  public String toString() {
    return '<' + "DLockGrantor" + "@"
        + Integer.toHexString(System.identityHashCode(this)) + " state="
        + stateToString(state) + " name=" + dlock.getName()
        + " version=" + getVersionId() + '>';
  }

  /**
   * 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 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 state == INITIALIZING;
  }

  /**
   * Returns true if this grantor has been destroyed.
   *
   * @return true if this grantor has been destroyed
   */
  public boolean isDestroyed() {
    return 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>
   * Acquires destroy read lock before synchronizing on {@link #batchLocks}.
   * If read lock not acquired, wait for the Grantor to be destroyed.
   *
   * @throws LockGrantorDestroyedException if grantor is destroyed
   */
  void handleLockBatch(DLockRequestMessage request) throws InterruptedException {
    DLockLessorDepartureHandler handler = 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();

    waitWhileInitializing();
    if (request.checkForTimeout()) {
      cleanupSuspendState(request);
      return;
    }
    if (acquireDestroyReadLock(0)) {
      try {
        synchronized (batchLocks) { // assures serial processing
          final boolean isTraceEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
          if (isTraceEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch]");
          }

          checkDestroyed();
          if (isTraceEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch] request: {}",
                request);
          }

          DLockBatch batch = (DLockBatch) request.getObjectName();
          checkIfHostDeparted(batch.getOwner());
          makeReservation(batch);
          if (isTraceEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.handleLockBatch] granting {}",
                batch.getBatchId());
          }
          batchLocks.put(batch.getBatchId(), batch);
          request.respondWithGrant(Long.MAX_VALUE);
        }
      } catch (CommitConflictException ex) {
        request.respondWithTryLockFailed(ex.getMessage());
      } finally {
        releaseDestroyReadLock();
      }
    } else {
      waitUntilDestroyed();
      checkDestroyed();
    }
  }

  void makeReservation(DLockBatch batch) {
    resMgr.makeReservation((IdentityArrayList) batch.getReqs());
  }

  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 (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 (DLockBatch batch : batchLocks.values()) {
        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<InternalDistributedMember, Long> 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 destroy read lock before synchronizing on {@link #batchLocks}.
   * If read lock not acquired, wait for the Grantor to be destroyed.
   * <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);
    }

    waitWhileInitializing();
    if (acquireDestroyReadLock(0)) {
      try {
        synchronized (batchLocks) {
          checkDestroyed();
          ret = batchLocks.get(batchId);
        }
      } finally {
        releaseDestroyReadLock();
      }
    } else {
      waitUntilDestroyed();
      checkDestroyed();
    }
    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 destroy read lock before synchronizing on {@link #batchLocks}.
   * If read lock not acquired, wait for the Grantor to be destroyed.
   * <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);
    }
    waitWhileInitializing();
    if (acquireDestroyReadLock(0)) {
      try {
        synchronized (batchLocks) {
          checkDestroyed();
          final DLockBatch oldBatch = batchLocks.get(batchId);
          if (oldBatch != null) {
            batchLocks.put(batchId, newBatch);
          }
        }
      } finally {
        releaseDestroyReadLock();
      }
    } else {
      waitUntilDestroyed();
      checkDestroyed();
    }
    if (isTraceEnabled_DLS) {
      logger.trace(LogMarker.DLS_VERBOSE, "[DLockGrantor.updateLockBatch] exit: {}", batchId);
    }
  }

  /**
   * Releases the transaction optimized lock batch.
   * <p>
   * Acquires destroy read lock before synchronizing on {@link #batchLocks}.
   * If read lock not acquired, wait for the Grantor to be destroyed.
   *
   * @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]");
    }
    waitWhileInitializing();
    if (acquireDestroyReadLock(0)) {
      try {
        synchronized (batchLocks) {
          checkDestroyed();
          DLockBatch batch = batchLocks.remove(batchId);
          if (batch != null) {
            releaseReservation(batch);
          }
        }
      } finally {
        releaseDestroyReadLock();
      }
    } else {
      waitUntilDestroyed();
      checkDestroyed();
    }
  }

  void releaseReservation(DLockBatch batch) {
    resMgr.releaseReservation((IdentityArrayList) batch.getReqs());
  }

  /**
   * 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(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) && 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 =
          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);
        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 (grantTokens) {
        Set members = dlock.getDistributionManager().getDistributionManagerIds();

        final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
        for (final Object o : tokens) {
          DLockRemoteToken token = (DLockRemoteToken) o;
          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(dm);
                thread.checkTimeToWait(grantToken.getLeaseExpireTime() - now, true);
              }

              rThread = grantToken.getRemoteThread();
