/*
 * 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;

import static java.lang.Boolean.TRUE;
import static java.lang.Long.MAX_VALUE;
import static java.lang.System.out;
import static org.apache.geode.distributed.DistributedLockService.getServiceNamed;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.ThreadUtils.dumpAllStacks;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;
import org.assertj.core.api.Assertions;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.SystemFailure;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.locks.DLockGrantor;
import org.apache.geode.distributed.internal.locks.DLockRemoteToken;
import org.apache.geode.distributed.internal.locks.DLockRequestProcessor;
import org.apache.geode.distributed.internal.locks.DLockRequestProcessor.DLockRequestMessage;
import org.apache.geode.distributed.internal.locks.DLockRequestProcessor.DLockResponseMessage;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.locks.DLockToken;
import org.apache.geode.distributed.internal.locks.RemoteThread;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.DLockTest;

/**
 * This class tests distributed ownership via the DistributedLockService api.
 */
@Category({DLockTest.class})
public final class DistributedLockServiceDUnitTest extends JUnit4DistributedTestCase {
  private static final Logger logger = LogService.getLogger();

  private static DistributedSystem dlstSystem;
  private static DistributedLockBlackboard blackboard;
  private static final Object monitor = new Object();

  private int hits = 0;
  private int completes = 0;
  private boolean done;
  private boolean got;

  /**
   * Returns a previously created (or new, if this is the first time this method is called in this
   * VM) distributed system which is somewhat configurable via hydra test parameters.
   */
  @Override
  public final void postSetUp() throws Exception {

    createBlackboard();
    Invoke.invokeInEveryVM(this::createBlackboard);

    // Create a DistributedSystem in every VM
    connectDistributedSystem();

    Invoke.invokeInEveryVM(DistributedLockServiceDUnitTest::connectDistributedSystem);
  }

  private void createBlackboard() throws Exception {
    if (blackboard == null) {
      blackboard = DistributedLockBlackboardImpl.getInstance();
    }
  }

  @Override
  public final void preTearDown() {
    Invoke.invokeInEveryVM(DistributedLockServiceDUnitTest::destroyAllDLockServices);
  }

  @Override
  public void postTearDown() {
    disconnectAllFromDS();
  }

  public static void destroyAllDLockServices() {
    DLockService.destroyAll();
    dlstSystem = null;
  }

  /**
   * Connects a DistributedSystem, saves it in static variable "system"
   */
  protected static void connectDistributedSystem() {
    dlstSystem = (new DistributedLockServiceDUnitTest()).getSystem();
  }

  private static volatile boolean stop_testFairness;

  @Test
  public void testFairness() throws InterruptedException, ExecutionException {
    final int[] vmThreads = new int[] {1, 4, 8, 16};
    final int numVms = vmThreads.length;
    final int numThreads = Arrays.stream(vmThreads).sum();
    final List<VM> vms = new ArrayList<>();
    final List<Future> futures = new ArrayList<>();
    final String serviceName = "testFairness_" + getUniqueName();
    final Object lock = "lock";

    // get the lock and hold it until all threads are ready to go
    DistributedLockService service = DistributedLockService.create(serviceName, dlstSystem);
    assertThat(service.lock(lock, -1, -1)).isTrue();

    // create the lock service in all vms
    vms.addAll(forNumVMsInvoke(numVms, () -> remoteCreateService(serviceName)));
    Thread.sleep(100);

    Invoke.invokeInEveryVM(() -> {
      stop_testFairness = false;
    });

    // line up threads for the fairness race...
    for (int vm = 0; vm < numVms; vm++) {
      logger.info("[testFairness] lining up " + vmThreads[vm] + " threads in vm " + vm);

      for (int j = 0; j < vmThreads[vm]; j++) {
        SerializableCallable<Integer> fairnessRunnable = new SerializableCallable<Integer>() {
          @Override
          public Integer call() {
            // lock, inc count, and unlock until stop_testFairness is set true
            final AtomicInteger lockCount = new AtomicInteger(0);
            try {
              DistributedLockService service =
                  DistributedLockService.getServiceNamed(serviceName);
              while (!stop_testFairness) {
                assertThat(service.lock(lock, -1, -1)).isTrue();
                lockCount.incrementAndGet();
                service.unlock(lock);
              }
            } catch (VirtualMachineError e) {
              SystemFailure.initiateFailure(e);
              throw e;
            } catch (Throwable t) {
              logger.warn(t);
              fail(t.getMessage());
            } finally {
              return lockCount.get();
            }
          }
        };

        futures.add(vms.get(vm).invokeAsync(fairnessRunnable::call));
      }
    }

    // wait for all threads to be ready to start the race
    for (int i = 0; i < numVms; i++) {
      final int vmId = i;

      vms.get(vmId).invoke(() -> {
        DLockService localService =
            (DLockService) DistributedLockService.getServiceNamed(serviceName);

        await()
            .untilAsserted(() -> assertThat(localService.getStats().getLockWaitsInProgress())
                .isEqualTo(vmThreads[vmId]));
      });
    }

    // start the race!
    service.unlock(lock);
    Thread.sleep(1000 * 5); // 5 seconds

    // stop the race...
    assertThat(service.lock(lock, -1, -1)).isTrue();
    for (VM vm : vms) {
      vm.invoke(new SerializableRunnable() {
        @Override
        public void run() {
          stop_testFairness = true;
        }
      });
    }

    // release the lock and destroy the lock service
    service.unlock(lock);

    // calc total locks granted...
    Integer totalLocks = 0;
    Integer maxLocks = 0;
    Integer minLocks = Integer.MAX_VALUE;

    for (Future future : futures) {
      Integer numLocks = (Integer) future.get();
      totalLocks += numLocks;
      if (minLocks > numLocks) {
        minLocks = numLocks;
      }
      if (maxLocks < numLocks) {
        maxLocks = numLocks;
      }
    }

    vms.forEach(vm -> vm.invoke(() -> DistributedLockService.destroy(serviceName)));

    logger.info("[testFairness] totalLocks=" + totalLocks + " minLocks="
        + minLocks + " maxLocks=" + maxLocks);

    int expectedLocks = (totalLocks / numThreads) + 1;

    int deviation = (int) (expectedLocks * 0.3);
    int lowThreshold = expectedLocks - deviation;
    int highThreshold = expectedLocks + deviation;

    logger.info("[testFairness] deviation=" + deviation + " expectedLocks="
        + expectedLocks + " lowThreshold=" + lowThreshold + " highThreshold=" + highThreshold);

    assertThat(minLocks >= lowThreshold).withFailMessage("minLocks is less than lowThreshold")
        .isTrue();
    assertThat(maxLocks <= highThreshold).withFailMessage("maxLocks is greater than highThreshold")
        .isTrue();
  }

  @Test
  public void testOneGetsAndOthersTimeOut() throws Exception {
    doOneGetsAndOthersTimeOut(1, 1);
    doOneGetsAndOthersTimeOut(4, 3);
  }

  private static InternalDistributedMember getLockGrantor(String serviceName) {
    DLockService service = (DLockService) DistributedLockService.getServiceNamed(serviceName);
    assertThat(service).isNotNull();
    InternalDistributedMember grantor = service.getLockGrantorId().getLockGrantorMember();
    assertThat(grantor).isNotNull();
    System.out.println("In identifyLockGrantor - grantor is " + grantor);
    return grantor;
  }

  private static Boolean isLockGrantor(String serviceName) {
    DLockService service = (DLockService) DistributedLockService.getServiceNamed(serviceName);
    assertThat(service).isNotNull();
    Boolean result = service.isLockGrantor();
    System.out.println("In isLockGrantor: " + result);
    return result;
  }

  protected static void becomeLockGrantor(String serviceName) {
    DLockService service = (DLockService) DistributedLockService.getServiceNamed(serviceName);
    assertThat(service).isNotNull();
    System.out.println("About to call becomeLockGrantor...");
    service.becomeLockGrantor();
  }

  @Test
  public void testGrantorSelection() {
    // TODO change distributedCreateService usage to be concurrent threads

    final String serviceName = "testGrantorSelection_" + getUniqueName();

    final List<VM> vmList = distributedCreateService(4, serviceName, true);


    getLockGrantor(serviceName);

    assertGrantorIsConsistent(serviceName, vmList);
  }

  @Test
  public void testBasicGrantorRecovery() {
    int numVMs = 4;
    final String serviceName = "testBasicGrantorRecovery_" + getUniqueName();
    final List<VM> vmList = distributedCreateService(numVMs, serviceName, false);

    // Apparently it's necessary to query the lock grantor to have a server be nominated grantor.
    vmList.get(0).invoke(() -> getLockGrantor(serviceName));

    VM originalGrantorVM = identifyLockGrantorWithSanityCheck(serviceName, vmList);
    VM lockHolder = null;
    for (VM vm : vmList) {
      if (vm != originalGrantorVM) {
        lockHolder = vm;
        vm.invoke(() -> DLockService.getServiceNamed(serviceName).lock("foo", 1200, -1));
        break;
      }
    }

    final InternalDistributedMember originalGrantorMember =
        originalGrantorVM.invoke(() -> getLockGrantor(serviceName));

    System.out.println("originalGrantorMember = " + originalGrantorMember);
    final InternalDistributedMember originalGrantorSelfReportedMember = originalGrantorVM
        .invoke(() -> InternalDistributedSystem.getAnyInstance().getDistributedMember());
    assertThat(originalGrantorMember).isEqualTo(originalGrantorSelfReportedMember);

    originalGrantorVM.invoke(JUnit4DistributedTestCase::disconnectFromDS);

    vmList.remove(originalGrantorVM);

    await("vm0 leave has been processed")
        .until(() -> vmList.parallelStream()
            .noneMatch(vm -> vm.invoke(() -> InternalDistributedSystem
                .getAnyInstance()
                .getAllOtherMembers()
                .contains(originalGrantorMember))));

    // It seems that we need to actually use the lock service to have a new grantor be picked.
    lockHolder.invoke(() -> DLockService.getServiceNamed(serviceName).unlock("foo"));

    final VM newGrantorMember = identifyLockGrantorWithSanityCheck(serviceName, vmList);
    assertThat(newGrantorMember).isNotEqualTo(originalGrantorMember);
  }

  /**
   * This is just a better-sounding wrapper for when we want to do only consistency checks.
   */
  private static void assertGrantorIsConsistent(String serviceName, Collection<VM> vmList) {
    identifyLockGrantorWithSanityCheck(serviceName, vmList);
  }

  private static VM identifyLockGrantorWithSanityCheck(String serviceName, Collection<VM> vmList) {
    VM grantorVM = null;

    for (VM vm : vmList) {
      if (vm.invoke(() -> isLockGrantor(serviceName))) {
        grantorVM = vm;
      }
    }

    assertThat(grantorVM).withFailMessage("Some VM should think it's the lock grantor").isNotNull();

    final DistributedMember grantorID =
        grantorVM.invoke(() -> InternalDistributedSystem.getAnyInstance().getDistributedMember());

    for (VM vm : vmList) {
      if (vm != grantorVM) {
        assertThat(vm.invoke(() -> isLockGrantor(serviceName))).isFalse();
      }
      assertThat(vm.invoke(() -> getLockGrantor(serviceName))).isEqualTo(grantorID);
    }

    return grantorVM;
  }

  @Test
  public void testLockFailover() {
    final int originalGrantorVM = 0;
    final int oneVM = 1;
    final int twoVM = 2;
    final String serviceName = "testLockFailover-" + getUniqueName();

    // create lock services...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockFailover] create services");
    }

    VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(serviceName));

    Boolean isGrantor = VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
    assertThat(isGrantor)
        .withFailMessage("First member calling getLockGrantor failed to become grantor")
        .isEqualTo(Boolean.TRUE);

    // get locks...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockFailover] get lock");
    }

    Boolean locked = VM.getVM(originalGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + originalGrantorVM));
    assertThat(locked).withFailMessage("Failed to get lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    locked = VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + twoVM));
    assertThat(locked).withFailMessage("Failed to get lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    locked = VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + oneVM));
    assertThat(locked).withFailMessage("Failed to get lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    // disconnect originalGrantorVM...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockFailover] disconnect originalGrantorVM");
    }

    VM.getVM(originalGrantorVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        disconnectFromDS();
      }
    });

    try {
      Thread.sleep(100);
    } catch (InterruptedException ignore) {
      fail("interrupted");
    }

    // verify locks by unlocking...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockFailover] release locks");
    }

    Boolean unlocked = VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + twoVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    unlocked = VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + oneVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    // switch locks...
    locked = VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + twoVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    locked = VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + oneVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    unlocked = VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + twoVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    unlocked = VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + oneVM));
    assertThat(unlocked).withFailMessage("Failed to release lock in testLockFailover")
        .isEqualTo(Boolean.TRUE);

    // verify grantor is unique...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockFailover] verify grantor identity");
    }

    InternalDistributedMember oneID = VM.getVM(oneVM)
        .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(serviceName));
    InternalDistributedMember twoID = VM.getVM(twoVM)
        .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(serviceName));
    assertThat(oneID != null && twoID != null)
        .withFailMessage("Failed to identifyLockGrantor in testLockFailover").isTrue();
    assertThat(twoID).withFailMessage("Failed grantor uniqueness in testLockFailover")
        .isEqualTo(oneID);
  }

  @Test
  public void testLockThenBecomeLockGrantor() {
    final int originalGrantorVM = 0;
    final int becomeGrantorVM = 1;
    final int thirdPartyVM = 2;
    final String serviceName = "testLockThenBecomeLockGrantor-" + getUniqueName();

    // create lock services...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockThenBecomeLockGrantor] create services");
    }

    VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    try {
      Thread.sleep(20);
    } catch (InterruptedException ignore) {
      fail("interrupted");
    }

    VM.getVM(becomeGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    VM.getVM(thirdPartyVM)
        .invoke(() -> DistributedLockServiceDUnitTest.remoteCreateService(serviceName));

    VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(serviceName));

    Boolean isGrantor = VM.getVM(originalGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
    assertThat(isGrantor).isEqualTo(Boolean.TRUE)
        .withFailMessage("First member calling getLockGrantor failed to become grantor");

    // control...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockThenBecomeLockGrantor] check control");
    }
    Boolean check = VM.getVM(becomeGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + becomeGrantorVM));
    assertThat(check).isEqualTo(Boolean.FALSE)
        .withFailMessage("Check of control failed... unlock succeeded but nothing locked");

    // get locks...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockThenBecomeLockGrantor] get lock");
    }

    Boolean locked = VM.getVM(originalGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + originalGrantorVM));
    assertThat(locked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to get lock in testLockThenBecomeLockGrantor");

    locked = VM.getVM(thirdPartyVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + thirdPartyVM));
    assertThat(locked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to get lock in testLockThenBecomeLockGrantor");

    locked = VM.getVM(becomeGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "KEY-" + becomeGrantorVM));
    assertThat(locked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to get lock in testLockThenBecomeLockGrantor");

    // become lock grantor...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockThenBecomeLockGrantor] become lock grantor");
    }

    VM.getVM(becomeGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.becomeLockGrantor(serviceName));

    try {
      Thread.sleep(20);
    } catch (InterruptedException ignore) {
      fail("interrupted");
    }

    isGrantor = VM.getVM(becomeGrantorVM)
        .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
    assertThat(isGrantor).isEqualTo(Boolean.TRUE).withFailMessage("Failed to become lock grantor");

    // verify locks by unlocking...
    if (logger.isDebugEnabled()) {
      logger.debug("[testLockThenBecomeLockGrantor] release locks");
    }

    Boolean unlocked = VM.getVM(originalGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + originalGrantorVM));
    assertThat(unlocked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to release lock in testLockThenBecomeLockGrantor");

    unlocked = VM.getVM(thirdPartyVM)
        .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + thirdPartyVM));
    assertThat(unlocked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to release lock in testLockThenBecomeLockGrantor");

    unlocked = VM.getVM(becomeGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + becomeGrantorVM));
    assertThat(unlocked).isEqualTo(Boolean.TRUE)
        .withFailMessage("Failed to release lock in testLockThenBecomeLockGrantor");

    // test for bug in which transferred token gets re-entered causing lock recursion
    unlocked = VM.getVM(becomeGrantorVM).invoke(
        () -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY-" + becomeGrantorVM));
    assertThat(unlocked).isEqualTo(Boolean.FALSE)
        .withFailMessage("Transfer of tokens caused lock recursion in held lock");
  }

  @Test
  public void testBecomeLockGrantor() {
    // create lock services...
    int numVMs = 4;
    final String serviceName = "testBecomeLockGrantor-" + getUniqueName();
    distributedCreateService(numVMs, serviceName, true);

    // each one gets a lock...
    for (int vm = 0; vm < numVMs; vm++) {
      final int finalvm = vm;
      Boolean locked = VM.getVM(finalvm)
          .invoke(() -> DistributedLockServiceDUnitTest.lock(serviceName, "obj-" + finalvm));
      assertThat(locked).isEqualTo(Boolean.TRUE)
          .withFailMessage("Failed to get lock in testBecomeLockGrantor");
    }

    // find the grantor...
    int originalVM = -1;
    InternalDistributedMember oldGrantor = null;
    for (int vm = 0; vm < numVMs; vm++) {
      Boolean isGrantor = VM.getVM(vm)
          .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
      if (isGrantor) {
        originalVM = vm;
        oldGrantor = VM.getVM(vm)
            .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(
                serviceName));
        break;
      }
    }

    if (logger.isDebugEnabled()) {
      logger.debug("[testBecomeLockGrantor] original grantor is " + oldGrantor);
    }

    // have one call becomeLockGrantor
    for (int vm = 0; vm < numVMs; vm++) {
      if (vm != originalVM) {
        VM.getVM(vm)
            .invoke(() -> DistributedLockServiceDUnitTest.becomeLockGrantor(serviceName));
        Boolean isGrantor = VM.getVM(vm)
            .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
        assertThat(isGrantor).isEqualTo(Boolean.TRUE)
            .withFailMessage("isLockGrantor is false after calling becomeLockGrantor");
        break;
      }
    }

    if (logger.isDebugEnabled()) {
      logger.debug("[testBecomeLockGrantor] one vm has called becomeLockGrantor...");
    }

    InternalDistributedMember newGrantor = null;
    for (int vm = 0; vm < numVMs; vm++) {
      Boolean isGrantor = VM.getVM(vm)
          .invoke(() -> DistributedLockServiceDUnitTest.isLockGrantor(serviceName));
      if (isGrantor) {
        newGrantor = VM.getVM(vm)
            .invoke(() -> DistributedLockServiceDUnitTest.getLockGrantor(
                serviceName));
        break;
      }
    }
    assertThat(newGrantor).isNotEqualTo(oldGrantor);
    // verify locks still held by unlocking
    // each one unlocks...
    for (int vm = 0; vm < numVMs; vm++) {
      final int finalvm = vm;
      Boolean unlocked = VM.getVM(finalvm)
          .invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "obj-" + finalvm));
      assertThat(unlocked).isEqualTo(Boolean.TRUE)
          .withFailMessage("Failed to unlock in testBecomeLockGrantor");
    }

    if (logger.isDebugEnabled()) {
      logger.debug("[testBecomeLockGrantor] finished");
    }

    // verify that pending requests are granted by unlocking them also
  }

  @Test
  public void testOnlyOneVmAcquiresWithTryLock() throws Exception {
    final Long waitMillis = 100L;

    // create lock services...
    if (logger.isDebugEnabled()) {
      logger.debug("[testTryLock] create lock services");
    }
    final String serviceName = "testTryLock-" + getUniqueName();
    final List<VM> vms = distributedCreateService(4, serviceName, false);

    final List<AsyncInvocation<Boolean>> invocations = vms.stream()
        .map(vm -> vm.invokeAsync(
            () -> DistributedLockServiceDUnitTest.tryLock(serviceName, "KEY", waitMillis)))
        .collect(Collectors
            .toList());

    int lockCount = 0;
    for (AsyncInvocation<Boolean> invocation : invocations) {
      if (invocation.get()) {
        lockCount++;
      }
    }

    assertThat(lockCount).isEqualTo(1)
        .withFailMessage("More than one vm acquired the tryLock");

    if (logger.isDebugEnabled()) {
      logger.debug("[testTryLock] unlock tryLock");
    }
    int unlockCount = 0;
    for (VM vm : vms) {
      Boolean unlocked =
          vm.invoke(() -> DistributedLockServiceDUnitTest.unlock(serviceName, "KEY"));
      if (unlocked) {
        unlockCount++;
      }
    }

    assertThat(unlockCount).withFailMessage("More than one vm unlocked the tryLock").isEqualTo(1);
  }

  @Test
  public void testOneGetsThenOtherGets() throws Exception { // (numVMs, numThreadsPerVM)
    doOneGetsThenOtherGets(1, 1);
    // doOneGetsThenOtherGets(2, 2);
    // doOneGetsThenOtherGets(3, 3);
    doOneGetsThenOtherGets(4, 3);
  }

  @Test
  public void testLockDifferentNames() {
    String serviceName = getUniqueName();

    // Same VM
    remoteCreateService(serviceName);
    DistributedLockService service = DistributedLockService.getServiceNamed(serviceName);
    assertThat(service.lock("obj1", -1, -1)).isTrue();
    assertThat(service.lock("obj2", -1, -1)).isTrue();
    service.unlock("obj1");
    service.unlock("obj2");

    // Different VMs
    VM vm = VM.getVM(0);
    vm.invoke(() -> remoteCreateService(serviceName));
    assertThat(service.lock("masterVMobj", -1, -1)).isTrue();

    assertThat(vm.invoke(() -> getLockAndIncrement(serviceName, "otherVMobj", -1, 0))).isTrue();

    service.unlock("masterVMobj");
  }

  @Test
  public void testLocalGetLockAndIncrement() throws Exception {
    String serviceName = getUniqueName();
    remoteCreateService(serviceName);
    DistributedLockService.getServiceNamed(serviceName);
    assertThat(getLockAndIncrement(serviceName, "localVMobj", -1, 0)).isTrue();
  }

  @Test
  public void testRemoteGetLockAndIncrement() {
    String serviceName = getUniqueName();
    VM vm = VM.getVM(0);
    vm.invoke(() -> remoteCreateService(serviceName));
    assertThat(vm.invoke(() -> getLockAndIncrement(serviceName, "remoteVMobj",
        -1, 0))).isTrue();
  }

  @Test
  public void testLockSameNameDifferentService() {
    String serviceName1 = getUniqueName() + "_1";
    String serviceName2 = getUniqueName() + "_2";
    String objName = "obj";

    // Same VM
    remoteCreateService(serviceName1);
    remoteCreateService(serviceName2);
    DistributedLockService service1 = DistributedLockService.getServiceNamed(serviceName1);
    DistributedLockService service2 = DistributedLockService.getServiceNamed(serviceName2);
    assertThat(service1.lock(objName, -1, -1)).isTrue();
    assertThat(service2.lock(objName, -1, -1)).isTrue();
    service1.unlock(objName);
    service2.unlock(objName);

    // Different VMs
    VM vm = VM.getVM(0);
    vm.invoke(() -> remoteCreateService(serviceName1));
    vm.invoke(() -> remoteCreateService(serviceName2));
    assertThat(service1.lock(objName, -1, -1)).isTrue();
    assertThat(vm.invoke(() -> getLockAndIncrement(serviceName2, objName, -1, 0))).isTrue();
    service1.unlock(objName);
  }

  @Test
  public void testLeaseDoesntExpire() {
    String serviceName = getUniqueName();
    final Object objName = 3;

    // Same VM
    remoteCreateService(serviceName);
    final DistributedLockService service = DistributedLockService.getServiceNamed(serviceName);
    // lock objName with a sufficiently long lease
    assertThat(service.lock(objName, -1, 60000)).isTrue();
    // try to lock in another thread, with a timeout shorter than above lease
    final boolean[] resultHolder = new boolean[] {false};
    Thread thread = new Thread(() -> resultHolder[0] = !service.lock(objName, 1000, -1));
    thread.start();
    ThreadUtils.join(thread, 30 * 1000);
    assertThat(resultHolder[0]).isTrue();
    // the unlock should succeed without throwing LeaseExpiredException
    service.unlock(objName);

    // Different VM
    VM vm = VM.getVM(0);
    vm.invoke(() -> remoteCreateService(serviceName));
    // lock objName in this VM with a sufficiently long lease
    assertThat(service.lock(objName, -1, 60000)).isTrue();
    // try to lock in another VM, with a timeout shorter than above lease
    assertThat(vm.invoke(() -> getLockAndIncrement(serviceName, objName, 1000L, 0L))).isFalse();
    // the unlock should succeed without throwing LeaseExpiredException
    service.unlock(objName);
  }

  @Test
  public void testLockUnlock() {
    String serviceName = getUniqueName();
    Object objName = 42;

    remoteCreateService(serviceName);
    DistributedLockService service = DistributedLockService.getServiceNamed(serviceName);

    assertThat(!service.isHeldByCurrentThread(objName)).isTrue();

    service.lock(objName, -1, -1);
    assertThat(service.isHeldByCurrentThread(objName)).isTrue();

    service.unlock(objName);
    assertThat(!service.isHeldByCurrentThread(objName)).isTrue();
  }

  @Test
  public void testLockExpireUnlock() throws Exception {
    long leaseMs = 200;
    long waitBeforeLockingMs = 210;

    String serviceName = getUniqueName();
    Object objName = 42;

    remoteCreateService(serviceName);
    DistributedLockService service = DistributedLockService.getServiceNamed(serviceName);

    assertThat(!service.isHeldByCurrentThread(objName)).isTrue();

    assertThat(service.lock(objName, -1, leaseMs)).isTrue();
    assertThat(service.isHeldByCurrentThread(objName)).isTrue();

    Thread.sleep(waitBeforeLockingMs); // should expire...
    assertThat(!service.isHeldByCurrentThread(objName)).isTrue();

    Assertions.assertThatThrownBy(() -> service.unlock(objName))
        .isInstanceOf(LeaseExpiredException.class);
  }

  @Test
