/*
 * 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.Collections.emptySet;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.DataInput;
import java.io.DataOutput;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LeaseExpiredException;
import org.apache.geode.distributed.LockNotHeldException;
import org.apache.geode.distributed.LockServiceDestroyedException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.deadlock.UnsafeThreadLocal;
import org.apache.geode.distributed.internal.locks.DLockQueryProcessor.DLockQueryReplyMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.internal.util.concurrent.FutureResult;
import org.apache.geode.logging.internal.OSProcess;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implements the distributed locking service with distributed lock grantors.
 *
 */
public class DLockService extends DistributedLockService {

  private static final Logger logger = LogService.getLogger();

  private static final long NOT_GRANTOR_SLEEP = Long
      .getLong(GeodeGlossary.GEMFIRE_PREFIX + "DLockService.notGrantorSleep", 100);

  private static final boolean DEBUG_NONGRANTOR_DESTROY_LOOP = Boolean
      .getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "DLockService.debug.nonGrantorDestroyLoop");

  private static final int DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT = Integer
      .getInteger(
          GeodeGlossary.GEMFIRE_PREFIX + "DLockService.debug.nonGrantorDestroyLoopCount", 20);

  private static final boolean AUTOMATE_FREE_RESOURCES =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "DLockService.automateFreeResources");

  static final int INVALID_LEASE_ID = -1;

  /** Unique name for this instance of the named locking service */
  protected final String serviceName;

  /** DistributionManager for this member */
  private final DistributionManager dm;

  /**
   * DistributedSystem connection for this member (used for DisconnectListener, logging, etc)
   */
  protected final InternalDistributedSystem ds;

  /** Known lock tokens for this service. Key:Object(name), Value:DLockToken */
  private final Map<Object, DLockToken> tokens = new HashMap<>();

  /**
   * True if this member has destroyed this named locking service. Field is volatile only because
   * it's referenced in {@link #toString()} (never synchronize in <code>toString</code>).
   */
  private volatile boolean destroyed = false;

  /**
   * True if this is a distributed lock service; false if local to this vm only. TX has a "local"
   * lock service which sets this to false.
   */
  private final boolean isDistributed;

  /** Optional handler for departure of lease holders; used by grantor */
  private DLockLessorDepartureHandler lessorDepartureHandler;

  /**
   * Hook for transactions which allows custom behavior in processing DLockRecoverGrantorMessage
   */
  private DLockRecoverGrantorProcessor.MessageProcessor recoverGrantorProcessor;

  /** Thread-safe reference to DistributedLockStats */
  private final DistributedLockStats dlockStats;

  /**
   * Protects {@link #lockGrantorId}, {@link #grantor} and {@link #lockGrantorFutureResult}. Final
   * granting of a lock occurs under this synchronization and only if <code>lockGrantorId</code>
   * matches the grantor that granted the lock.
   */
  private final Object lockGrantorIdLock = new Object();

  /** Identifies the current grantor for this lock service. */
  private LockGrantorId lockGrantorId;

  /**
   * Local instance of the lock grantor if this process is the grantor. This field is volatile for
   * one use: 1) {@link #toString()} which should not use synchronization due to potential for wrong
   * lock ordering. Can we make this non-volatile??
   */
  private volatile DLockGrantor grantor;

  /**
   * Count of currently active locks and lock requests. Used to determine if destroy must tell the
   * grantor to release all held locks.
   */
  private int activeLocks = 0;

  /** True if this service should automatically freeResources */
  private final boolean automateFreeResources;

  /** Identifies the thread that is destroying this lock service. */
  private final ThreadLocal<Boolean> destroyingThread = new ThreadLocal<>();

  /// ** Held during destroy and creation of this lock service. */
  // private final Object serviceLock = new Object();

  /** Protects access to {@link #destroyed} and {@link #activeLocks}. */
  private final Object destroyLock = new Object();

  /**
   * Created by the thread communicating directly with the elder. Other threads will wait on this
   * and then use the resulting lockGrantorId. This ensures that only one message is sent to the
   * elder and that only one thread does so at a time. Protected by {@link #lockGrantorIdLock} and
   * holds a reference to a {@link LockGrantorId}.
   * <p>
   * Only outbound threads and operations should ever wait on this. Do NOT allow inbound threads to
   * use the <code>lockGrantorFutureResult</code>.
   */
  private FutureResult<Object> lockGrantorFutureResult;

  private final DLockStopper stopper;

  // -------------------------------------------------------------------------
  // State and concurrency construct methods
  // -------------------------------------------------------------------------

  public boolean isDestroyed() {
    synchronized (destroyLock) {
      if (destroyed) {
        return !isCurrentThreadDoingDestroy();
      }
      return false;
    }
  }

  public void checkDestroyed() {
    if (isDestroyed()) {
      throw generateLockServiceDestroyedException(generateLockServiceDestroyedMessage());
    }
  }

  /**
   * Create a new LockServiceDestroyedException for this lock service.
   *
   * @param message the detail message that explains the exception
   * @return new LockServiceDestroyedException
   */
  protected LockServiceDestroyedException generateLockServiceDestroyedException(String message) {
    return new LockServiceDestroyedException(message);
  }

  /**
   * Returns the string message to use in a LockServiceDestroyedException for this lock service.
   *
   * @return the detail message that explains LockServiceDestroyedException
   */
  protected String generateLockServiceDestroyedMessage() {
    return String.format("%s has been destroyed", this);
  }

  /**
   * Returns true if {@link #lockGrantorId} is the same as the specified LockGrantorId. Caller must
   * synchronize on {@link #lockGrantorIdLock}.
   *
   * @param someLockGrantorId the LockGrantorId to check
   */
  private boolean checkLockGrantorId(LockGrantorId someLockGrantorId) {
    Assert.assertHoldsLock(lockGrantorIdLock, true);
    if (lockGrantorId == null) {
      return false;
    }
    return lockGrantorId.sameAs(someLockGrantorId);
  }

  /**
   * Returns true if lockGrantorId is the same as the specified LockGrantorId. Caller must
   * synchronize on lockGrantorIdLock.
   *
   * @param someLockGrantorId the LockGrantorId to check
   */
  public boolean isLockGrantorId(LockGrantorId someLockGrantorId) {
    synchronized (lockGrantorIdLock) {
      return checkLockGrantorId(someLockGrantorId);
    }
  }

  private boolean isCurrentThreadDoingDestroy() {
    return Boolean.TRUE.equals(destroyingThread.get());
  }

  private void setDestroyingThread() {
    destroyingThread.set(Boolean.TRUE);
  }

  private void clearDestroyingThread() {
    destroyingThread.remove();
  }

  private InternalDistributedMember getElderId() {
    InternalDistributedMember elder = dm.getElderId();
    if (elder == null) {
      dm.getSystem().getCancelCriterion().checkCancelInProgress(null);
    }
    Assert.assertTrue(elder != null);
    return elder;
  }

  /**
   * Returns id of the current lock grantor for this service. If necessary, a request will be sent
   * to the elder to fetch this information.
   */
  public LockGrantorId getLockGrantorId() {
    final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
    boolean ownLockGrantorFutureResult;
    FutureResult<Object> lockGrantorFutureResultRef = null;

    long statStart = -1;
    LockGrantorId theLockGrantorId = null;
    while (theLockGrantorId == null) {

      ownLockGrantorFutureResult = false;
      try {
        Assert.assertHoldsLock(destroyLock, false);
        synchronized (lockGrantorIdLock) {
          if (lockGrantorFutureResult != null) {
            lockGrantorFutureResultRef = lockGrantorFutureResult;
          } else if (lockGrantorId != null) {
            return lockGrantorId;
          } else {
            ownLockGrantorFutureResult = true;
            lockGrantorFutureResultRef = new FutureResult<>(dm.getCancelCriterion());
            if (isDebugEnabled_DLS) {
              logger.trace(LogMarker.DLS_VERBOSE,
                  "[getLockGrantorId] creating lockGrantorFutureResult");
            }
            lockGrantorFutureResult = lockGrantorFutureResultRef;
          }
        }

        statStart = getStats().startGrantorWait();
        if (!ownLockGrantorFutureResult) {
          LockGrantorId lockGrantorIdRef =
              waitForLockGrantorFutureResult(uncheckedCast(lockGrantorFutureResultRef), 0,
                  TimeUnit.MILLISECONDS);
          if (lockGrantorIdRef != null) {
            return lockGrantorIdRef;
          } else {
            continue;
          }
        }

        InternalDistributedMember elder = getElderId();
        Assert.assertTrue(elder != null);

        GrantorInfo gi = getGrantorRequest();
        theLockGrantorId =
            new LockGrantorId(dm, gi.getId(), gi.getVersionId(), gi.getSerialNumber());

        if (isDebugEnabled_DLS) {
          logger.trace(LogMarker.DLS_VERBOSE, "[getLockGrantorId] elder says grantor is {}",
              theLockGrantorId);
        }

        // elder tells us to be the grantor...
        if (theLockGrantorId.isLocal(getSerialNumber())) {
          boolean needsRecovery = gi.needsRecovery();
          if (!needsRecovery) {
            if (isDebugEnabled_DLS) {
              logger.trace(LogMarker.DLS_VERBOSE, "[getLockGrantorId] needsRecovery is false");
            }
            synchronized (lockGrantorIdLock) {
              // either no previous grantor or grantor is newer
              Assert.assertTrue(
                  lockGrantorId == null || lockGrantorId.isNewerThan(theLockGrantorId)
                      || lockGrantorId.sameAs(theLockGrantorId),
                  lockGrantorId + " should be null or newer than or same as "
                      + theLockGrantorId);
            }
          }
          if (!createLocalGrantor(elder, needsRecovery, theLockGrantorId)) {
            theLockGrantorId = lockGrantorId;
          }
        }

        // elder says another member is the grantor
        else {
          synchronized (lockGrantorIdLock) {
            if (!setLockGrantorId(theLockGrantorId)) {
              theLockGrantorId = lockGrantorId;
            }
          }
        }
      } finally {
        synchronized (lockGrantorIdLock) {
          boolean getLockGrantorIdFailed = theLockGrantorId == null;
          if (statStart > -1) {
            getStats().endGrantorWait(statStart, getLockGrantorIdFailed);
          }
          if (ownLockGrantorFutureResult) {
            // this thread is doing the real work and must finish the future
            Assert.assertTrue(lockGrantorFutureResult == lockGrantorFutureResultRef);
            if (getLockGrantorIdFailed) {
              // failed so cancel lockGrantorFutureResult
              lockGrantorFutureResultRef.cancel(false);
            } else {
              // succeeded so set lockGrantorFutureResult
              lockGrantorFutureResultRef.set(theLockGrantorId);
            }
            // null out the reference so it is free for next usage
            lockGrantorFutureResult = null;
          }
        }
      } // finally block for lockGrantorFutureResult
    } // while theLockGrantorId == null
    return theLockGrantorId;
  }

  /**
   * Creates a local {@link DLockGrantor}.
   *
   * if (!createLocalGrantor(xxx)) { theLockGrantorId = this.lockGrantorId; }
   *
   * @param elder the elder that told us to be the grantor
   * @param needsRecovery true if recovery is required
   * @param myLockGrantorId lockGrantorId to use
   * @return true if successfully created local grantor; false if aborted
   */
  private boolean createLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
      LockGrantorId myLockGrantorId) {
    DLockGrantor myGrantor =
        DLockGrantor.createGrantor(this, myLockGrantorId.getLockGrantorVersion());
    if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
      logger.trace(LogMarker.DLS_VERBOSE, "[createLocalGrantor] Calling makeLocalGrantor");
    }
    return makeLocalGrantor(elder, needsRecovery, myLockGrantorId, myGrantor);
  }

  private boolean makeLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
      LockGrantorId myLockGrantorId, DLockGrantor myGrantor) {
    final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
    boolean success = false;
    try {
      synchronized (lockGrantorIdLock) {
        if (isDestroyed()) {
          checkDestroyed(); // exit
        }

        InternalDistributedMember currentElder = getElderId();
        if (!currentElder.equals(elder)) {
          // abort because elder changed
          if (isDebugEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE,
                "Failed to create {} because elder changed from {} to {}", myLockGrantorId, elder,
                currentElder);
          }
          return false; // exit
        }

        if (deposingLockGrantorId != null) {
          if (deposingLockGrantorId.isNewerThan(myLockGrantorId)) {
            if (isDebugEnabled_DLS) {
              logger.trace(LogMarker.DLS_VERBOSE, "Failed to create {} because I was deposed by {}",
                  myLockGrantorId, deposingLockGrantorId);
            }
            deposingLockGrantorId = null;
            return false; // exit
          }

          if (isDebugEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "{} failed to depose {}",
                deposingLockGrantorId, myLockGrantorId);
          }
          // older grantor couldn't depose us, so null it out...
          deposingLockGrantorId = null;
        }

        if (!setLockGrantorId(myLockGrantorId, myGrantor)) {
          if (isDebugEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE,
                "[getLockGrantorId] failed to create {} because current grantor is {}",
                myLockGrantorId, lockGrantorId);
          }
          return false; // exit
        }
      } // release sync on this.lockGrantorIdLock

      // do NOT sync while doing recovery (because it waits for replies)
      if (needsRecovery) {
        boolean recovered =
            DLockRecoverGrantorProcessor.recoverLockGrantor(dm.getDistributionManagerIds(),
                this, // this lock service
                myGrantor, dm, elder); // the elder that told us to be the grantor
        if (!recovered) {
          checkDestroyed();
          return false; // exit
        }
      }

      // after recovery, resynchronize on lockGrantorIdLock again
      // check to see if myLockGrantorId has been deposed
      synchronized (lockGrantorIdLock) {
        if (isDestroyed()) {
          checkDestroyed(); // exit
        }

        if (deposingLockGrantorId != null) {
          if (deposingLockGrantorId.isNewerThan(myLockGrantorId)) {
            if (isDebugEnabled_DLS) {
              logger.trace(LogMarker.DLS_VERBOSE, "Failed to create {} because I was deposed by {}",
                  myLockGrantorId, deposingLockGrantorId);
            }
            deposingLockGrantorId = null;
            return false; // exit
          }

          if (isDebugEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "{} failed to depose {}",
                deposingLockGrantorId, myLockGrantorId);
          }
          deposingLockGrantorId = null;
        }

        if (checkLockGrantorId(myLockGrantorId)) {
          success = myGrantor.makeReady(true); // do not enforce initializing
        }
      }

      return success; // exit
    } catch (VirtualMachineError err) {
      SystemFailure.initiateFailure(err);
      // If this ever returns, rethrow the error. We're poisoned
      // now, so don't let this thread continue.
      throw err;
    } catch (Error e) {
      // Whenever you catch Error or Throwable, you must also
      // catch VirtualMachineError (see above). However, there is
      // _still_ a possibility that you are dealing with a cascading
      // error condition, so you also need to check to see if the JVM
      // is still usable:
      SystemFailure.checkFailure();
      if (isDebugEnabled_DLS) {
        logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] throwing Error", e);
      }
      throw e;
    } catch (RuntimeException e) {
      if (isDebugEnabled_DLS) {
        logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] throwing RuntimeException", e);
      }
      throw e;
    } finally {

      try {
        // abort if unsuccessful or if lock service was destroyed
        if (!success || isDestroyed()) {
          if (isDebugEnabled_DLS) {
            logger.trace(LogMarker.DLS_VERBOSE, "[makeLocalGrantor] aborting {} and {}",
                myLockGrantorId, myGrantor);
          }
          nullLockGrantorId(myLockGrantorId);
          if (!myGrantor.isDestroyed()) {
            myGrantor.destroy();
          }
        }
      } finally {
        // assertion: grantor should now be either ready or destroyed!

        if (myGrantor.isInitializing() && !dm.getCancelCriterion().isCancelInProgress()) {
          logger.error(LogMarker.DLS_MARKER,
              "Grantor is still initializing");
        }
        if (!success && !myGrantor.isDestroyed() && !dm.getCancelCriterion().isCancelInProgress()) {
          logger.error(LogMarker.DLS_MARKER,
              "Grantor creation was aborted but grantor was not destroyed");
        }
      }
    }
  }

  /**
   * Set {@link #lockGrantorId} to the given new value if the current value is null or is an older
   * grantor version. Caller must hold {@link #lockGrantorIdLock}.
   *
   * @param newLockGrantorId the new value for lockGrantorId
   */
  private boolean setLockGrantorId(LockGrantorId newLockGrantorId) {
    Assert.assertHoldsLock(lockGrantorIdLock, true);
    if (equalsLockGrantorId(newLockGrantorId)) {
      return true;
    } else if (!newLockGrantorId.hasLockGrantorVersion()) {
      // proceed with temporary placeholder used by become grantor
      lockGrantorId = newLockGrantorId;
      return true;
    } else if (newLockGrantorId.isRemote() && lockGrantorId != null
        && lockGrantorId.hasLockGrantorVersion()) {
      if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
        logger.trace(LogMarker.DLS_VERBOSE, "[setLockGrantorId] tried to replace {} with {}",
            lockGrantorId, newLockGrantorId);
      }
      return false;
    } else if (newLockGrantorId.isNewerThan(lockGrantorId)) {
      lockGrantorId = newLockGrantorId;
      return true;
    } else {
      return false;
    }
  }

  /**
   * Set {@link #lockGrantorId} to the <code>localLockGrantorId</code> if current value is null or
   * is an older grantor version. This also atomically sets {@link #grantor} to ensure that the two
   * fields are kept in sync. Caller must hold {@link #lockGrantorIdLock}.
   *
   * @param localLockGrantorId the new value for lockGrantorId
   * @param localGrantor the new local intance of DLockGrantor
   */
  private boolean setLockGrantorId(LockGrantorId localLockGrantorId, DLockGrantor localGrantor) {
    Assert.assertHoldsLock(lockGrantorIdLock, true);
    Assert.assertTrue(localLockGrantorId.isLocal(getSerialNumber()));
    if (setLockGrantorId(localLockGrantorId)) {
      grantor = localGrantor;
      return true;
    }
    return false;
  }

  private LockGrantorId deposingLockGrantorId;

  /**
   * Deposes {@link #lockGrantorId} if <code>newLockGrantorId</code> is newer.
   *
   * @param newLockGrantorId the new lock grantor
   */
  void deposeOlderLockGrantorId(LockGrantorId newLockGrantorId) {
    LockGrantorId deposedLockGrantorId = null;
    final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
    synchronized (lockGrantorIdLock) {
      if (isDebugEnabled_DLS) {
        logger.trace(LogMarker.DLS_VERBOSE, "[deposeOlderLockGrantorId] pre-deposing {} for new {}",
            deposedLockGrantorId, newLockGrantorId);
      }
      deposingLockGrantorId = newLockGrantorId;
      deposedLockGrantorId = lockGrantorId;
    }
    if (deposedLockGrantorId != null && deposedLockGrantorId.hasLockGrantorVersion()
        && newLockGrantorId.isNewerThan(deposedLockGrantorId)) {
      if (isDebugEnabled_DLS) {
        logger.trace(LogMarker.DLS_VERBOSE,
            "[deposeOlderLockGrantorId] post-deposing {} for new {}", deposedLockGrantorId,
            newLockGrantorId);
      }
      nullLockGrantorId(deposedLockGrantorId);
    }
  }

  /**
   * Sets {@link #lockGrantorId} to null if the current value equals the expected old value. Caller
   * must hold {@link #lockGrantorIdLock}.
   *
   * @param oldLockGrantorId the expected old value
   * @return true if lockGrantorId was set to null
   */
  private boolean nullLockGrantorId(LockGrantorId oldLockGrantorId) {
    Assert.assertHoldsLock(destroyLock, false);
    Assert.assertHoldsLock(lockGrantorIdLock, false);
    if (oldLockGrantorId == null) {
      return false;
    }
    DLockGrantor grantorToDestroy = null;
    try {
      synchronized (lockGrantorIdLock) {
        if (equalsLockGrantorId(oldLockGrantorId)
            || (oldLockGrantorId.isLocal(getSerialNumber()) && isMakingLockGrantor())) {
          // this.lockGrantorId != null && this.lockGrantorId.isLocal())) {
          if (oldLockGrantorId.isLocal(getSerialNumber())
              && isLockGrantorVersion(grantor, oldLockGrantorId.getLockGrantorVersion())) {
            // need to destroy and remove grantor
            grantorToDestroy = grantor;
            grantor = null;
          }
          lockGrantorId = null;
          return true;
        } else {
          return false;
        }
      }
    } finally {
      if (grantorToDestroy != null) {
        if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
          logger.trace(LogMarker.DLS_VERBOSE, "[nullLockGrantorId] destroying {}",
              grantorToDestroy);
        }
        grantorToDestroy.destroy();
      }
    }
  }

  /**
   * Returns true if the grantor version of <code>dlockGrantor</code> equals the
   * <code>grantorVersion</code>.
   *
   * @param dlockGrantor the grantor instance to compare to grantorVersion
   * @param grantorVersion the grantor version number
   * @return true if dlockGrantor is the same grantor version
   */
  private boolean isLockGrantorVersion(DLockGrantor dlockGrantor, long grantorVersion) {
    if (dlockGrantor == null) {
      return false;
    }
    return dlockGrantor.getVersionId() == grantorVersion;
  }

  /**
   * Returns true if <code>someLockGrantor</code> equals the current {@link #lockGrantorId}.
   *
   * @return true if someLockGrantor equals the current lockGrantorId
   */
  private boolean equalsLockGrantorId(LockGrantorId someLockGrantor) {
    Assert.assertHoldsLock(lockGrantorIdLock, true);
    if (someLockGrantor == null) {
      return lockGrantorId == null;
    }
    return someLockGrantor.equals(lockGrantorId);
  }

  /**
   * Returns id of the current lock grantor for this service. If necessary, a request will be sent
   * to the elder to fetch this information. Unlike getLockGrantorId this call will not become the
   * lock grantor.
   */
  public LockGrantorId peekLockGrantorId() {
    Assert.assertHoldsLock(destroyLock, false);
    synchronized (lockGrantorIdLock) {
      LockGrantorId currentLockGrantorId = lockGrantorId;
      if (currentLockGrantorId != null) {
        return currentLockGrantorId;
      }
    }

    long statStart = getStats().startGrantorWait();
    LockGrantorId theLockGrantorId = null;
    try {
      // 1st thread wins the right to request grantor info from elder
      GrantorInfo gi = peekGrantor();
      InternalDistributedMember lockGrantorMember = gi.getId();
      if (lockGrantorMember == null) {
        return null;
      }
      theLockGrantorId =
          new LockGrantorId(dm, lockGrantorMember, gi.getVersionId(), gi.getSerialNumber());
      return theLockGrantorId;
    } finally {
      boolean getLockGrantorIdFailed = theLockGrantorId == null;
      getStats().endGrantorWait(statStart, getLockGrantorIdFailed);
    }
  }

  /**
   * Increments {@link #activeLocks} while synchronized on {@link #destroyLock} after calling
   * {@link #checkDestroyed()}.
   */
  private void incActiveLocks() {
    synchronized (destroyLock) {
      checkDestroyed();
      activeLocks++;
    }
  }

  /**
   * Decrements {@link #activeLocks} while synchronized on {@link #destroyLock}.
   */
  private void decActiveLocks() {
    synchronized (destroyLock) {
      activeLocks--;
    }
  }

  /**
   * Returns lockGrantorId when lockGrantorFutureResultRef has been set by another thread.
   *
   * @param lockGrantorFutureResultRef FutureResult to wait for
   * @param timeToWait how many ms to wait, 0 = forever
   * @param timeUnit the unit of measure for timeToWait
   * @return the LockGrantorId or null if FutureResult was cancelled
   */
  private LockGrantorId waitForLockGrantorFutureResult(
      FutureResult<LockGrantorId> lockGrantorFutureResultRef,
      long timeToWait, final TimeUnit timeUnit) {
    LockGrantorId lockGrantorIdRef = null;
    while (lockGrantorIdRef == null) {
      boolean interrupted = Thread.interrupted();
      try {
        checkDestroyed();
        if (timeToWait == 0) {
          lockGrantorIdRef = lockGrantorFutureResultRef.get();
        } else {
          lockGrantorIdRef = lockGrantorFutureResultRef.get(timeToWait, timeUnit);
        }
      } catch (TimeoutException e) {
        break;
      } catch (InterruptedException e) {
        interrupted = true;
        dm.getCancelCriterion().checkCancelInProgress(e);
        if (lockGrantorFutureResultRef.isCancelled()) {
          // cancelled Future might throw InterruptedException...?
          checkDestroyed();
          break; // return null
        }
      } catch (CancellationException e) { // Future was cancelled
        checkDestroyed();
        break; // return null
      } finally {
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
    }
    return lockGrantorIdRef;
  }

  /**
   * nulls out grantor to force call to elder
   *
   * @param timeToWait how long to wait for a new grantor. -1 don't wait, 0 no time limit
   * @param timeUnit the unit of measure of timeToWait
   */
  private void notLockGrantorId(LockGrantorId notLockGrantorId, long timeToWait,
      final TimeUnit timeUnit) {
    if (notLockGrantorId.isLocal(getSerialNumber())) {
      if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
        logger.trace(LogMarker.DLS_VERBOSE,
            "notLockGrantorId {} returning early because notGrantor {} was equal to the local dm {}",
            serviceName, notLockGrantorId, dm.getId());
      }
      // Let the local destroy or processing of transfer do the clear
      return;
    }

    boolean ownLockGrantorFutureResult = false;
    FutureResult<Object> lockGrantorFutureResultRef = null;

    long statStart = -1;
    final LockGrantorId currentLockGrantorId;

    try {
      Assert.assertHoldsLock(destroyLock, false);
      synchronized (lockGrantorIdLock) {
        currentLockGrantorId = lockGrantorId;
        if (lockGrantorFutureResult != null) {
          // some other thread is talking to elder
          lockGrantorFutureResultRef = lockGrantorFutureResult;
        } else if (!notLockGrantorId.sameAs(currentLockGrantorId)) {
          return;
        } else {
          // this thread needs to talk to elder
          ownLockGrantorFutureResult = true;
          lockGrantorFutureResultRef = new FutureResult<>(dm.getCancelCriterion());
          lockGrantorFutureResult = lockGrantorFutureResultRef;
        }
      }

      statStart = getStats().startGrantorWait();
      if (!ownLockGrantorFutureResult) {
        if (timeToWait >= 0) {
          waitForLockGrantorFutureResult(uncheckedCast(lockGrantorFutureResultRef), timeToWait,
              timeUnit);
        }
        return;
      }

      InternalDistributedMember elder = getElderId();
      Assert.assertTrue(elder != null);

      LockGrantorId elderLockGrantorId = null;
      GrantorInfo gi = peekGrantor();
      if (gi.getId() != null) {
        elderLockGrantorId =
            new LockGrantorId(dm, gi.getId(), gi.getVersionId(), gi.getSerialNumber());
      }

      if (notLockGrantorId.sameAs(elderLockGrantorId)) {
        // elder says that notLockGrantorId is still the grantor...
        sleep(NOT_GRANTOR_SLEEP);
      } else {
        // elder says another member is the grantor
        nullLockGrantorId(notLockGrantorId);
        if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
          logger.trace(LogMarker.DLS_VERBOSE,
              "notLockGrantorId cleared lockGrantorId for service {}", serviceName);
        }
      }
    } finally {
      synchronized (lockGrantorIdLock) {
        if (statStart > -1) {
          getStats().endGrantorWait(statStart, false);
        }
        if (ownLockGrantorFutureResult) {
          // this thread is doing the real work and must finish the future
          Assert.assertTrue(lockGrantorFutureResult == lockGrantorFutureResultRef);
          // cancel lockGrantorFutureResult
          lockGrantorFutureResultRef.cancel(false);
          // null out the reference so it is free for next usage
          lockGrantorFutureResult = null;
        }
      }
    } // finally block for lockGrantorFutureResult
  }

  /**
   * All calls to GrantorRequestProcessor.clearGrantor must come through this synchronization point.
   * <p>
