geode-core/src/distributedTest/java/org/apache/geode/internal/cache/NetSearchMessagingDUnitTest.java

/*
 * 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.internal.cache;

import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.Test;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.SubscriptionAttributes;
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.internal.cache.SearchLoadAndWriteProcessor.NetSearchRequestMessage;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;


public class NetSearchMessagingDUnitTest extends JUnit4CacheTestCase {
  private static final AtomicBoolean listenerHasFinished = new AtomicBoolean();

  @Test
  public void testOneMessageWithReplicates() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    VM vm3 = host.getVM(3);

    createReplicate(vm0);
    createReplicate(vm1);
    createNormal(vm2);
    createEmpty(vm3);

    // Test with a null value
    long vm0Count = getReceivedMessages(vm0);
    long vm1Count = getReceivedMessages(vm1);
    long vm2Count = getReceivedMessages(vm2);
    long vm3Count = getReceivedMessages(vm3);

    assertEquals(null, get(vm3, "a"));

    // Make sure we only processed one message
    assertEquals(vm3Count + 1, getReceivedMessages(vm3));

    // Make sure the replicates only saw one message between them
    assertEquals(vm0Count + vm1Count + 1, getReceivedMessages(vm0) + getReceivedMessages(vm1));

    // Make sure the normal vm didn't see any messages
    assertEquals(vm2Count, getReceivedMessages(vm2));

    // Test with a real value value
    put(vm3, "a", "b");

    vm0Count = getReceivedMessages(vm0);
    vm1Count = getReceivedMessages(vm1);
    vm2Count = getReceivedMessages(vm2);
    vm3Count = getReceivedMessages(vm3);

    assertEquals("b", get(vm3, "a"));

    // Make sure we only processed one message
    assertEquals(vm3Count + 1, getReceivedMessages(vm3));

    // Make sure the replicates only saw one message between them

    assertEquals(vm0Count + vm1Count + 1, getReceivedMessages(vm0) + getReceivedMessages(vm1));

    // Make sure the normal vm didn't see any messages
    assertEquals(vm2Count, getReceivedMessages(vm2));
  }

  @Test
  public void testNetSearchNormals() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);

    createNormal(vm0);
    createNormal(vm1);
    createNormal(vm2);
    createEmpty(vm3);

    // Test with a null value
    long vm0Count = getReceivedMessages(vm0);
    long vm1Count = getReceivedMessages(vm1);
    long vm2Count = getReceivedMessages(vm2);
    long vm3Count = getReceivedMessages(vm3);

    assertEquals(null, get(vm3, "a"));

    // Make sure we only processed one message
    waitForReceivedMessages(vm3, vm3Count + 3);

    // Make sure the normal vms each saw 1 query message.
    assertEquals(vm0Count + vm1Count + vm2Count + 3,
        getReceivedMessages(vm0) + getReceivedMessages(vm1) + getReceivedMessages(vm2));

    // Test with a real value value
    put(vm3, "a", "b");

    vm0Count = getReceivedMessages(vm0);
    vm1Count = getReceivedMessages(vm1);
    vm2Count = getReceivedMessages(vm2);
    vm3Count = getReceivedMessages(vm3);

    assertEquals("b", get(vm3, "a"));

    // Make sure we only processed one message
    waitForReceivedMessages(vm2, vm2Count + 1);

    waitForReceivedMessages(vm3, vm3Count + 3);

    // Make sure the normal vms each saw 1 query message.
    assertEquals(vm0Count + vm1Count + vm2Count + 3,
        getReceivedMessages(vm0) + getReceivedMessages(vm1) + getReceivedMessages(vm2));
  }

  /**
   * In bug #48186 a deadlock occurs when a netsearch pulls in a value from the disk and causes a
   * LRU eviction of another entry. Here we merely demonstrate that a netsearch that gets the value
   * of an overflow entry does not update the LRU status of that entry.
   */
  @Test
  public void testNetSearchNoLRU() {
    Host host = Host.getHost(0);
    VM vm2 = host.getVM(2);
    VM vm1 = host.getVM(1);

    createOverflow(vm2, 5);
    createEmpty(vm1);

    // Test with a null value
    put(vm2, "a", "1");
    put(vm2, "b", "2");
    put(vm2, "c", "3");
    put(vm2, "d", "4");
    put(vm2, "e", "5");
    // the cache in vm0 is now full and LRU will occur on this next put()
    put(vm2, "f", "6");

    SerializableCallable verifyEvicted = new SerializableCallable("verify eviction of 'a'") {
      @Override
      public Object call() {
        Cache cache = getCache();
        LocalRegion region = (LocalRegion) cache.getRegion("region");
        RegionEntry re = region.getRegionEntry("a");
        Object o = re.getValueInVM(region);
        LogWriterUtils.getLogWriter().info("key a=" + o);;
        return o == null || o == Token.NOT_AVAILABLE;
      }
    };

    boolean evicted = (Boolean) vm2.invoke(verifyEvicted);
    assertTrue("expected 'a' to be evicted", evicted);

    // now netsearch for 'a' from the other VM and verify again
    Object value = get(vm1, "a");
    assertEquals("expected to find '1' result from netSearch", "1", value);

    evicted = (Boolean) vm2.invoke(verifyEvicted);
    assertTrue("expected 'a' to still be evicted", evicted);
    vm2.invoke(new SerializableRunnable("verify other entries are not evicted") {
      @Override
      public void run() {
        Cache cache = getCache();
        LocalRegion region = (LocalRegion) cache.getRegion("region");
        String[] keys = new String[] {"b", "c", "d", "e", "f"};
        for (String key : keys) {
          RegionEntry re = region.getRegionEntry(key);
          Object o = re.getValueInVM(region);
          LogWriterUtils.getLogWriter().info("key " + key + "=" + o);
          assertTrue("expected key " + key + " to not be evicted",
              (o != null) && (o != Token.NOT_AVAILABLE));
        }
      }
    });
  }

  /**
   * The system prefers net searching replicates. If a replicate fails after it responds to a query
   * message and before it returns a value, the system should fall back to net searching normal
   * members.
   */
  @Test
  public void testNetSearchFailoverFromReplicate() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    VM vm3 = host.getVM(3);

    installListenerToDisconnectOnNetSearchRequest(vm0);

    createReplicate(vm0);
    createNormal(vm1);
    createNormal(vm2);
    createEmpty(vm3);

    put(vm3, "a", "b");

    assertEquals("b", get(vm3, "a"));

    vm0.invoke(() -> await("system to shut down")
        .untilAsserted(
            () -> assertThat(InternalDistributedSystem.getConnectedInstance()).isNull()));

    waitForListenerToFinish(vm0);
  }

  /**
   * When a replicate fails after responding to a query message, the net search should fail over to
   * the next replicate that responded.
   */
  @Test
  public void testNetSearchFailoverFromOneReplicateToAnother() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);

    installListenerToDisconnectOnNetSearchRequest(vm0);

    createReplicate(vm0);
    createReplicate(vm1);
    createEmpty(vm2);

    put(vm2, "a", "b");

    boolean disconnected = false;
    while (!disconnected) {
      // get() causes vm2 to send a query message to all replicated members. All members with that
      // key reply. vm2 then sends a net search request to the first one that replies. If that
      // fails, it sends a net search request to the next one that replied. And so on.
      //
      // Because we can't be sure which member will respond first to each query, we repeat the loop
      // until vm0 is the first responder. vm0 will then disconnect when it receives the net search
      // request, forcing failover to another vm (in this case vm1). If we got the correct answer
      // AND the system is disconnected, then failover occurred.
      assertEquals("b", get(vm2, "a"));

      // Make sure we were disconnected in vm0
      disconnected = (Boolean) vm0.invoke(new SerializableCallable("check disconnected") {
        @Override
        public Object call() {
          return InternalDistributedSystem.getConnectedInstance() == null;
        }
      });
    }

    waitForListenerToFinish(vm0);
  }

  private Object put(VM vm, final String key, final String value) {
    return vm.invoke(new SerializableCallable() {

      @Override
      public Object call() {
        Cache cache = getCache();
        Region region = cache.getRegion("region");
        LogWriterUtils.getLogWriter().info("putting key=" + key + "=" + value);
        Object result = region.put(key, value);
        LogWriterUtils.getLogWriter().info("done putting key=" + key);
        return result;
      }
    });
  }

  private Object get(VM vm, final Object key) {
    return vm.invoke(new SerializableCallable("get " + key) {

      @Override
      public Object call() {
        Cache cache = getCache();
        Region region = cache.getRegion("region");
        return region.get(key);
      }
    });
  }

  private void waitForReceivedMessages(final VM vm, final long expected) {
    await().untilAsserted(new WaitCriterion() {

      @Override
      public boolean done() {
        return getReceivedMessages(vm) == expected;
      }

      @Override
      public String description() {
        return "Expected " + expected + " but got " + getReceivedMessages(vm);
      }
    });
  }

  private long getReceivedMessages(VM vm) {
    return ((Long) vm.invoke(new SerializableCallable() {

      @Override
      public Object call() {
        GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
        return cache.getInternalDistributedSystem().getDMStats().getReceivedMessages();
      }
    })).intValue();
  }

  private void createEmpty(VM vm) {
    vm.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Cache cache = getCache();
        RegionFactory rf = new RegionFactory();
        rf.setScope(Scope.DISTRIBUTED_ACK);
        rf.setConcurrencyChecksEnabled(false);
        rf.setDataPolicy(DataPolicy.EMPTY);
        rf.create("region");
      }
    });

  }

  private void createNormal(VM vm) {
    vm.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Cache cache = getCache();
        RegionFactory rf = new RegionFactory();
        rf.setScope(Scope.DISTRIBUTED_ACK);
        rf.setConcurrencyChecksEnabled(false);
        rf.setDataPolicy(DataPolicy.NORMAL);
        rf.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
        rf.create("region");
      }
    });

  }

  private void createOverflow(VM vm, final int count) {
    vm.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Cache cache = getCache();
        RegionFactory rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
        rf.setEvictionAttributes(
            EvictionAttributes.createLRUEntryAttributes(count, EvictionAction.OVERFLOW_TO_DISK));
        rf.create("region");
      }
    });

  }

  private void createReplicate(VM vm) {
    System.out.println("START SETUP createReplicate in vm " + vm.getId() + " for test "
        + getTestClass().getSimpleName() + "." + getTestMethodName());

    vm.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Cache cache = getCache();
        RegionFactory rf = new RegionFactory();
        rf.setScope(Scope.DISTRIBUTED_ACK);
        rf.setConcurrencyChecksEnabled(false);
        rf.setDataPolicy(DataPolicy.REPLICATE);
        rf.create("region");
      }
    });

    System.out.println("END SETUP createReplicate in vm " + vm.getId() + " for test "
        + getTestClass().getSimpleName() + "." + getTestMethodName());

  }

  private void installListenerToDisconnectOnNetSearchRequest(VM vm) {
    vm.invoke(new SerializableRunnable("Install listener") {
      @Override
      public void run() {
        listenerHasFinished.set(false);
        DistributionMessageObserver ob = new DistributionMessageObserver() {
          @Override
          public void beforeProcessMessage(ClusterDistributionManager dm,
              DistributionMessage message) {
            if (message instanceof NetSearchRequestMessage) {
              DistributionMessageObserver.setInstance(null);
              disconnectFromDS();
              listenerHasFinished.set(true);
            }
          }
        };
        DistributionMessageObserver.setInstance(ob);
      }
    });
  }

  private void waitForListenerToFinish(VM vm) {
    vm.invoke("wait for listener to finish", () -> {
      assertThat(DistributionMessageObserver.getInstance())
          .withFailMessage("listener was not invoked")
          .isNull();
      await("listener to finish")
          .untilAsserted(() -> assertThat(listenerHasFinished).isTrue());
    });
  }
}