geode-core/src/distributedTest/java/org/apache/geode/internal/cache/DeltaForceSizingFlagDUnitTest.java - geode - Git at Google

 /*
  * 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.assertj.core.api.Assertions.assertThat;

 import java.io.File;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.logging.log4j.Logger;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;

 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.DiskStoreFactory;
 import org.apache.geode.cache.EvictionAction;
 import org.apache.geode.cache.EvictionAttributes;
 import org.apache.geode.cache.InterestPolicy;
 import org.apache.geode.cache.PartitionAttributes;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.SubscriptionAttributes;
 import org.apache.geode.cache.util.ObjectSizer;
 import org.apache.geode.logging.internal.log4j.api.LogService;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.rules.ClusterStartupRule;
 import org.apache.geode.test.dunit.rules.MemberVM;

 /**
  * Tests the use of the per-delta "forceRecalculateSize" flag.
  */

 public class DeltaForceSizingFlagDUnitTest {
   private static final String TEST_REGION_NAME = "forceResizeTestRegionName";
   public static final String SMALLER_DELTA_DATA = "12345";
   public static final String LARGER_DELTA_DATA = "1234567890";
   public static final String DELTA_KEY = "a_key";
   public static final String RR_DISK_STORE_NAME = "_forceRecalculateSize_replicate_store";
   private static final Logger logger = LogService.getLogger();

   @Rule
   public ClusterStartupRule cluster = new ClusterStartupRule();

   protected MemberVM locator;
   protected MemberVM server1;
   protected MemberVM server2;

   @Before
   public void setup() {
     int locatorPort;
     locator = cluster.startLocatorVM(0);
     locatorPort = locator.getPort();

     server1 = cluster.startServerVM(1, locatorPort);
     server2 = cluster.startServerVM(2, locatorPort);
   }

   @Test
   public void testRRMemLRUDelta() {
     doRRMemLRUDeltaTest(false);
   }

   @Test
   public void testRRMemLRUDeltaAndFlag() {
     doRRMemLRUDeltaTest(true);
   }

   @Test
   public void testPRNoLRUDelta() {
     doPRNoLRUDeltaTest(false);
   }

   @Test
   public void testPRNoLRUAndFlagDelta() {
     doPRNoLRUDeltaTest(true);
   }

   private void doRRMemLRUDeltaTest(boolean shouldSizeChange) {
     VM vm1 = server1.getVM();
     VM vm2 = server2.getVM();

     createRR(server1);
     createRR(server2);
     TestDelta delta1 = new TestDelta(false, SMALLER_DELTA_DATA, shouldSizeChange);
     put(vm1, delta1);

     assertValueType(vm1, ValueType.RAW_VALUE);
     assertValueType(vm2, ValueType.CD_SERIALIZED);
     assertThat(getObjectSizerInvocations(vm1)).isEqualTo(1);
     assertThat(getObjectSizerInvocations(vm2)).isEqualTo(0);

     long origEvictionSize0 = getSizeFromEvictionStats(vm1);
     long origEvictionSize1 = getSizeFromEvictionStats(vm2);
     delta1.info = LARGER_DELTA_DATA;
     delta1.hasDelta = true;
     // Update the delta
     put(vm1, delta1);

     assertValueType(vm1, ValueType.RAW_VALUE);
     assertValueType(vm2, ValueType.CD_DESERIALIZED);

     assertThat(getObjectSizerInvocations(vm1)).isEqualTo(2);

     long finalEvictionSize0 = getSizeFromEvictionStats(vm1);
     long finalEvictionSize1 = getSizeFromEvictionStats(vm2);
     assertThat(finalEvictionSize0 - origEvictionSize0).isEqualTo(5);
     if (shouldSizeChange) {
       assertThat(getObjectSizerInvocations(vm2)).isEqualTo(1);
       // I'm not sure what the change in size should be, because we went
       // from serialized to deserialized
       assertThat(finalEvictionSize1 - origEvictionSize1).isNotEqualTo(0);
     } else {
       // we invoke the sizer once when we deserialize the original to apply the delta to it
       assertThat(getObjectSizerInvocations(vm2)).isEqualTo(0);
       assertThat(finalEvictionSize1 - origEvictionSize1).isEqualTo(0);
     }
   }

   private void doPRNoLRUDeltaTest(boolean shouldSizeChange) {
     VM vm1 = server1.getVM();
     VM vm2 = server2.getVM();

     createPR(server1);
     createPR(server2);

     TestDelta delta1 = new TestDelta(false, SMALLER_DELTA_DATA, shouldSizeChange);
     put(vm1, delta1);
     long origPRSize0 = getSizeFromPRStats(vm1);
     long origPRSize1 = getSizeFromPRStats(vm2);

     // Update the delta
     delta1.info = LARGER_DELTA_DATA;
     delta1.hasDelta = true;
     put(vm1, delta1);
     long finalPRSize0 = getSizeFromPRStats(vm1);
     long finalPRSize1 = getSizeFromPRStats(vm2);

     if (shouldSizeChange) {
       // I'm not sure what the change in size should be, because we went
       // from serialized to deserialized
       assertThat(finalPRSize0 - origPRSize0).isNotEqualTo(0);
       assertThat(finalPRSize1 - origPRSize1).isNotEqualTo(0);
     } else {
       assertThat(finalPRSize0 - origPRSize0).isEqualTo(0);
       assertThat(finalPRSize1 - origPRSize1).isEqualTo(0);
     }
   }

   private long getSizeFromPRStats(VM vm0) {
     return vm0.invoke("getSizeFromPRStats", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();
       InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
       if (region instanceof PartitionedRegion) {
         long total = 0;
         PartitionedRegion pr = (PartitionedRegion) region;
         int totalNumBuckets = pr.getPartitionAttributes().getTotalNumBuckets();
         for (int i = 0; i < totalNumBuckets; i++) {
           total += pr.getDataStore().getBucketSize(i);
         }
         return total;
       } else {
         return 0L;
       }
     });
   }

   private long getSizeFromEvictionStats(VM vm0) {
     return vm0.invoke("getSizeFromEvictionStats", () -> {

       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();
       InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
       return region.getEvictionCounter();
     });
   }

   private int getObjectSizerInvocations(VM vm0) {
     return vm0.invoke("getObjectSizerInvocations", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();
       InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
       return getObjectSizerInvocations(region);
     });
   }

   private void put(VM vm0, final Object value) {
     vm0.invoke("Put data", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();
       InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
       region.put(DeltaForceSizingFlagDUnitTest.DELTA_KEY, value);
     });
   }

   protected static int getObjectSizerInvocations(InternalRegion region) {
     TestObjectSizer sizer = (TestObjectSizer) region.getEvictionAttributes().getObjectSizer();
     int result = sizer.invocations.get();
     logger.info("objectSizerInvocations=" + result);
     return result;
   }

   private void createRR(MemberVM memberVM) {
     memberVM.invoke("Create replicateRegion", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();

       DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
       diskStoreFactory.setDiskDirs(getMyDiskDirs());
       diskStoreFactory.create(RR_DISK_STORE_NAME);

       RegionFactory<Integer, TestDelta> regionFactory = cache.createRegionFactory();
       regionFactory.setDataPolicy(DataPolicy.REPLICATE);
       regionFactory.setDiskStoreName(RR_DISK_STORE_NAME);
       regionFactory.setDiskSynchronous(true);
       regionFactory.setEvictionAttributes(EvictionAttributes.createLRUMemoryAttributes(1,
           new TestObjectSizer(), EvictionAction.OVERFLOW_TO_DISK));
       regionFactory.setScope(Scope.DISTRIBUTED_ACK);

       regionFactory.create(TEST_REGION_NAME);
     });
   }

   private void assertValueType(VM vm, final ValueType expectedType) {
     vm.invoke("assertValueType", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();
       InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
       Object value = region.getValueInVM(DeltaForceSizingFlagDUnitTest.DELTA_KEY);
       switch (expectedType) {
         case RAW_VALUE:
           assertThat(value).isNotInstanceOf(CachedDeserializable.class);
           break;
         case CD_SERIALIZED:
           assertThat(value).isInstanceOf(CachedDeserializable.class);

           Object serializedValue = ((CachedDeserializable) value).getValue();
           assertThat(serializedValue).isInstanceOf(byte[].class);
           break;
         case CD_DESERIALIZED:
           assertThat(value).isInstanceOf(CachedDeserializable.class);

           Object deserializedValue = ((CachedDeserializable) value).getValue();
           assertThat(deserializedValue).isNotInstanceOf(byte[].class);
           break;
         case EVICTED:
           assertThat(value).isNull();
           break;
       }
     });
   }

   private static File[] getMyDiskDirs() {
     long random = new Random().nextLong();
     File file = new File(Long.toString(random));
     assertThat(file.mkdirs()).isTrue();
     return new File[] {file};
   }

   private void createPR(MemberVM memberVM) {
     memberVM.invoke("Create partitioned region", () -> {
       Cache cache = ClusterStartupRule.getCache();
       assertThat(cache).isNotNull();

       PartitionAttributesFactory<Integer, TestDelta> paf =
           new PartitionAttributesFactory<>();
       paf.setRedundantCopies(1);
       PartitionAttributes<Integer, TestDelta> prAttr = paf.create();

       RegionFactory<Integer, TestDelta> regionFactory = cache.createRegionFactory();
       regionFactory.setDataPolicy(DataPolicy.PARTITION);
       regionFactory.setDiskSynchronous(true);
       regionFactory.setPartitionAttributes(prAttr);
       regionFactory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
       regionFactory.create(TEST_REGION_NAME);
     });
   }

   private static class TestObjectSizer implements ObjectSizer {
     private final AtomicInteger invocations = new AtomicInteger();

     @Override
     public int sizeof(Object o) {
       logger.info("TestObjectSizer invoked");
       if (o instanceof TestDelta) {
         invocations.incrementAndGet();
         return ((TestDelta) o).info.length();
       }
       if (o instanceof Integer) {
         return 0;
       }
       throw new RuntimeException("Unexpected type to be sized " + o.getClass() + ", object=" + o);
     }
   }

   enum ValueType {
     RAW_VALUE, CD_SERIALIZED, CD_DESERIALIZED, EVICTED
   }
 }
	/*
	* 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.assertj.core.api.Assertions.assertThat;

	import java.io.File;
	import java.util.Random;
	import java.util.concurrent.atomic.AtomicInteger;

	import org.apache.logging.log4j.Logger;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;

	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.DiskStoreFactory;
	import org.apache.geode.cache.EvictionAction;
	import org.apache.geode.cache.EvictionAttributes;
	import org.apache.geode.cache.InterestPolicy;
	import org.apache.geode.cache.PartitionAttributes;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.Scope;
	import org.apache.geode.cache.SubscriptionAttributes;
	import org.apache.geode.cache.util.ObjectSizer;
	import org.apache.geode.logging.internal.log4j.api.LogService;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.rules.ClusterStartupRule;
	import org.apache.geode.test.dunit.rules.MemberVM;

	/**
	* Tests the use of the per-delta "forceRecalculateSize" flag.
	*/

	public class DeltaForceSizingFlagDUnitTest {
	private static final String TEST_REGION_NAME = "forceResizeTestRegionName";
	public static final String SMALLER_DELTA_DATA = "12345";
	public static final String LARGER_DELTA_DATA = "1234567890";
	public static final String DELTA_KEY = "a_key";
	public static final String RR_DISK_STORE_NAME = "_forceRecalculateSize_replicate_store";
	private static final Logger logger = LogService.getLogger();

	@Rule
	public ClusterStartupRule cluster = new ClusterStartupRule();

	protected MemberVM locator;
	protected MemberVM server1;
	protected MemberVM server2;

	@Before
	public void setup() {
	int locatorPort;
	locator = cluster.startLocatorVM(0);
	locatorPort = locator.getPort();

	server1 = cluster.startServerVM(1, locatorPort);
	server2 = cluster.startServerVM(2, locatorPort);
	}

	@Test
	public void testRRMemLRUDelta() {
	doRRMemLRUDeltaTest(false);
	}

	@Test
	public void testRRMemLRUDeltaAndFlag() {
	doRRMemLRUDeltaTest(true);
	}

	@Test
	public void testPRNoLRUDelta() {
	doPRNoLRUDeltaTest(false);
	}

	@Test
	public void testPRNoLRUAndFlagDelta() {
	doPRNoLRUDeltaTest(true);
	}

	private void doRRMemLRUDeltaTest(boolean shouldSizeChange) {
	VM vm1 = server1.getVM();
	VM vm2 = server2.getVM();

	createRR(server1);
	createRR(server2);
	TestDelta delta1 = new TestDelta(false, SMALLER_DELTA_DATA, shouldSizeChange);
	put(vm1, delta1);

	assertValueType(vm1, ValueType.RAW_VALUE);
	assertValueType(vm2, ValueType.CD_SERIALIZED);
	assertThat(getObjectSizerInvocations(vm1)).isEqualTo(1);
	assertThat(getObjectSizerInvocations(vm2)).isEqualTo(0);

	long origEvictionSize0 = getSizeFromEvictionStats(vm1);
	long origEvictionSize1 = getSizeFromEvictionStats(vm2);
	delta1.info = LARGER_DELTA_DATA;
	delta1.hasDelta = true;
	// Update the delta
	put(vm1, delta1);

	assertValueType(vm1, ValueType.RAW_VALUE);
	assertValueType(vm2, ValueType.CD_DESERIALIZED);

	assertThat(getObjectSizerInvocations(vm1)).isEqualTo(2);

	long finalEvictionSize0 = getSizeFromEvictionStats(vm1);
	long finalEvictionSize1 = getSizeFromEvictionStats(vm2);
	assertThat(finalEvictionSize0 - origEvictionSize0).isEqualTo(5);
	if (shouldSizeChange) {
	assertThat(getObjectSizerInvocations(vm2)).isEqualTo(1);
	// I'm not sure what the change in size should be, because we went
	// from serialized to deserialized
	assertThat(finalEvictionSize1 - origEvictionSize1).isNotEqualTo(0);
	} else {
	// we invoke the sizer once when we deserialize the original to apply the delta to it
	assertThat(getObjectSizerInvocations(vm2)).isEqualTo(0);
	assertThat(finalEvictionSize1 - origEvictionSize1).isEqualTo(0);
	}
	}

	private void doPRNoLRUDeltaTest(boolean shouldSizeChange) {
	VM vm1 = server1.getVM();
	VM vm2 = server2.getVM();

	createPR(server1);
	createPR(server2);

	TestDelta delta1 = new TestDelta(false, SMALLER_DELTA_DATA, shouldSizeChange);
	put(vm1, delta1);
	long origPRSize0 = getSizeFromPRStats(vm1);
	long origPRSize1 = getSizeFromPRStats(vm2);

	// Update the delta
	delta1.info = LARGER_DELTA_DATA;
	delta1.hasDelta = true;
	put(vm1, delta1);
	long finalPRSize0 = getSizeFromPRStats(vm1);
	long finalPRSize1 = getSizeFromPRStats(vm2);

	if (shouldSizeChange) {
	// I'm not sure what the change in size should be, because we went
	// from serialized to deserialized
	assertThat(finalPRSize0 - origPRSize0).isNotEqualTo(0);
	assertThat(finalPRSize1 - origPRSize1).isNotEqualTo(0);
	} else {
	assertThat(finalPRSize0 - origPRSize0).isEqualTo(0);
	assertThat(finalPRSize1 - origPRSize1).isEqualTo(0);
	}
	}

	private long getSizeFromPRStats(VM vm0) {
	return vm0.invoke("getSizeFromPRStats", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();
	InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
	if (region instanceof PartitionedRegion) {
	long total = 0;
	PartitionedRegion pr = (PartitionedRegion) region;
	int totalNumBuckets = pr.getPartitionAttributes().getTotalNumBuckets();
	for (int i = 0; i < totalNumBuckets; i++) {
	total += pr.getDataStore().getBucketSize(i);
	}
	return total;
	} else {
	return 0L;
	}
	});
	}

	private long getSizeFromEvictionStats(VM vm0) {
	return vm0.invoke("getSizeFromEvictionStats", () -> {

	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();
	InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
	return region.getEvictionCounter();
	});
	}

	private int getObjectSizerInvocations(VM vm0) {
	return vm0.invoke("getObjectSizerInvocations", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();
	InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
	return getObjectSizerInvocations(region);
	});
	}

	private void put(VM vm0, final Object value) {
	vm0.invoke("Put data", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();
	InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
	region.put(DeltaForceSizingFlagDUnitTest.DELTA_KEY, value);
	});
	}

	protected static int getObjectSizerInvocations(InternalRegion region) {
	TestObjectSizer sizer = (TestObjectSizer) region.getEvictionAttributes().getObjectSizer();
	int result = sizer.invocations.get();
	logger.info("objectSizerInvocations=" + result);
	return result;
	}

	private void createRR(MemberVM memberVM) {
	memberVM.invoke("Create replicateRegion", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();

	DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
	diskStoreFactory.setDiskDirs(getMyDiskDirs());
	diskStoreFactory.create(RR_DISK_STORE_NAME);

	RegionFactory<Integer, TestDelta> regionFactory = cache.createRegionFactory();
	regionFactory.setDataPolicy(DataPolicy.REPLICATE);
	regionFactory.setDiskStoreName(RR_DISK_STORE_NAME);
	regionFactory.setDiskSynchronous(true);
	regionFactory.setEvictionAttributes(EvictionAttributes.createLRUMemoryAttributes(1,
	new TestObjectSizer(), EvictionAction.OVERFLOW_TO_DISK));
	regionFactory.setScope(Scope.DISTRIBUTED_ACK);

	regionFactory.create(TEST_REGION_NAME);
	});
	}

	private void assertValueType(VM vm, final ValueType expectedType) {
	vm.invoke("assertValueType", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();
	InternalRegion region = (InternalRegion) cache.getRegion(TEST_REGION_NAME);
	Object value = region.getValueInVM(DeltaForceSizingFlagDUnitTest.DELTA_KEY);
	switch (expectedType) {
	case RAW_VALUE:
	assertThat(value).isNotInstanceOf(CachedDeserializable.class);
	break;
	case CD_SERIALIZED:
	assertThat(value).isInstanceOf(CachedDeserializable.class);

	Object serializedValue = ((CachedDeserializable) value).getValue();
	assertThat(serializedValue).isInstanceOf(byte[].class);
	break;
	case CD_DESERIALIZED:
	assertThat(value).isInstanceOf(CachedDeserializable.class);

	Object deserializedValue = ((CachedDeserializable) value).getValue();
	assertThat(deserializedValue).isNotInstanceOf(byte[].class);
	break;
	case EVICTED:
	assertThat(value).isNull();
	break;
	}
	});
	}

	private static File[] getMyDiskDirs() {
	long random = new Random().nextLong();
	File file = new File(Long.toString(random));
	assertThat(file.mkdirs()).isTrue();
	return new File[] {file};
	}

	private void createPR(MemberVM memberVM) {
	memberVM.invoke("Create partitioned region", () -> {
	Cache cache = ClusterStartupRule.getCache();
	assertThat(cache).isNotNull();

	PartitionAttributesFactory<Integer, TestDelta> paf =
	new PartitionAttributesFactory<>();
	paf.setRedundantCopies(1);
	PartitionAttributes<Integer, TestDelta> prAttr = paf.create();

	RegionFactory<Integer, TestDelta> regionFactory = cache.createRegionFactory();
	regionFactory.setDataPolicy(DataPolicy.PARTITION);
	regionFactory.setDiskSynchronous(true);
	regionFactory.setPartitionAttributes(prAttr);
	regionFactory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
	regionFactory.create(TEST_REGION_NAME);
	});
	}

	private static class TestObjectSizer implements ObjectSizer {
	private final AtomicInteger invocations = new AtomicInteger();

	@Override
	public int sizeof(Object o) {
	logger.info("TestObjectSizer invoked");
	if (o instanceof TestDelta) {
	invocations.incrementAndGet();
	return ((TestDelta) o).info.length();
	}
	if (o instanceof Integer) {
	return 0;
	}
	throw new RuntimeException("Unexpected type to be sized " + o.getClass() + ", object=" + o);
	}
	}

	enum ValueType {
	RAW_VALUE, CD_SERIALIZED, CD_DESERIALIZED, EVICTED
	}
	}