geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSizeDUnitTest.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.apache.geode.cache.EvictionAction.OVERFLOW_TO_DISK;
 import static org.apache.geode.cache.EvictionAttributes.createLRUEntryAttributes;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.io.File;

 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;

 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.DiskStore;
 import org.apache.geode.cache.DiskStoreFactory;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.cache.CacheTestCase;
 import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

 /**
  * This test verifies the size API for 100 thousand put operations (done synch/asynch) on
  * PartitionedRegions with different combinations of Scope and Redundancy (Scope DIST_ACK,
  * Redundancy 1 AND Scope DIST_NO_ACK, Redundancy 0).
  */

 public class PartitionedRegionSizeDUnitTest extends CacheTestCase {

   private static final String DISK_STORE_NAME = "DISKSTORE";
   private static final String REGION_NAME = "PR";
   private static final int CNT = 100;
   private static final int TOTAL_NUMBER_OF_BUCKETS = 5;

   private File overflowDirectory;

   private VM vm0;
   private VM vm1;
   private VM vm2;
   private VM vm3;

   @Rule
   public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

   @Before
   public void setUp() throws Exception {
     vm0 = VM.getVM(0);
     vm1 = VM.getVM(1);
     vm2 = VM.getVM(2);
     vm3 = VM.getVM(3);

     overflowDirectory = temporaryFolder.newFolder("overflowDir");
   }

   /**
    * This test method invokes methods doing size validation on PRs.
    */
   @Test
   public void testSize() throws Exception {
     // Create PRs with dataStore on 3 VMs
     vm0.invoke(() -> createPartitionedRegion(200));
     vm1.invoke(() -> createPartitionedRegion(200));
     vm2.invoke(() -> createPartitionedRegion(200));

     // Create only accessor on 4th VM
     vm3.invoke(() -> createPartitionedRegion(0));

     // Do put operations on PR synchronously.
     vm3.invoke(() -> {
       Region<Integer, Integer> region =
           getCache().getRegion(PartitionedRegionSizeDUnitTest.REGION_NAME);
       for (int k = 0; k < CNT; k++) {
         region.put(k, k);
       }
     });

     // Validate the size against the total put operations
     vm3.invoke(() -> {
       Region region = getCache().getRegion(PartitionedRegionSizeDUnitTest.REGION_NAME);
       await().until(() -> region.size() == CNT);
     });
   }

   /**
    * Regression test for TRAC #39868
    *
    * <p>
    * TRAC #39868: PartitionMemberDetails.getSize() reports negative PR sizes when redundancy is 0
    */
   @Test
   public void testBug39868() throws Exception {
     vm0.invoke(() -> createPartitionedRegion(200));

     vm0.invoke(() -> {
       Region<Integer, byte[]> region = getRegion(REGION_NAME);
       for (int i = 0; i < 100; i++) {
         region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
       }
     });

     vm1.invoke(() -> createPartitionedRegion(200));

     vm0.invoke(() -> {
       Region<Integer, byte[]> region = getRegion(REGION_NAME);
       for (int i = 0; i < 100; i++) {
         region.destroy(i * TOTAL_NUMBER_OF_BUCKETS);
       }
     });

     vm1.invoke(() -> {
       PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
       long bytes = partitionedRegion.getDataStore().currentAllocatedMemory();
       assertThat(bytes).isEqualTo(0);
     });
   }

   @Test
   public void testByteSize() throws Exception {
     vm0.invoke(() -> createPartitionedRegion(200));
     vm1.invoke(() -> createPartitionedRegion(200));

     long bucketSizeWithOneEntry = vm0.invoke(() -> {
       Region<Integer, byte[]> region = getRegion(REGION_NAME);
       region.put(0, new byte[100]);

       PartitionedRegion partitionedRegion = (PartitionedRegion) region;
       PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
       long size = dataStore.getBucketSize(0);

       for (int i = 1; i < 100; i++) {
         region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
       }
       await().until(() -> dataStore.getBucketsManaged() == (short) 1);

       // make sure the size is proportional to the amount of data
       await().until(() -> dataStore.getBucketSize(0) == 100 * size);

       // destroy and invalidate entries and make sure the size goes down
       for (int i = 0; i < 25; i++) {
         region.destroy(i * TOTAL_NUMBER_OF_BUCKETS);
       }

       for (int i = 25; i < 50; i++) {
         region.invalidate(i * TOTAL_NUMBER_OF_BUCKETS);
       }

       await().until(() -> dataStore.getBucketSize(0) == 50 * size);

       // put some larger values in and make sure the size goes up
       for (int i = 50; i < 75; i++) {
         region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[150]);
       }

       // Now put in some smaller values and see if the size balances out
       for (int i = 75; i < 100; i++) {
         region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[50]);
       }

       await().until(() -> dataStore.getBucketSize(0) == 50 * size);

       return size;
     });

     vm1.invoke(() -> {
       PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
       await().until(
           () -> partitionedRegion.getDataStore().getBucketSize(0) == 50 * bucketSizeWithOneEntry);
     });

     vm1.invoke(() -> {
       PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
       PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
       await().until(() -> dataStore.currentAllocatedMemory() == 50 * bucketSizeWithOneEntry);
     });
   }

   @Test
   public void testByteSizeWithEviction() throws Exception {
     vm0.invoke(this::createPartitionedRegionWithOverflow);

     long bucketSizeWithOneEntry = vm0.invoke(() -> {
       Region<Integer, byte[]> region = getRegion(REGION_NAME);
       region.put(0, new byte[100]);

       PartitionedRegion partitionedRegion = (PartitionedRegion) region;
       PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
       long size = dataStore.getBucketSize(0);

       for (int i = 1; i < 100; i++) {
         region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
       }
       await().until(() -> dataStore.getBucketsManaged() == (short) 1);
       return size;
     });

     vm0.invoke(() -> {
       Region<Integer, byte[]> region = getRegion(REGION_NAME);
       PartitionedRegion partitionedRegion = (PartitionedRegion) region;
       PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();

       // there should only be 2 items in memory
       await().until(() -> dataStore.currentAllocatedMemory() == 2 * bucketSizeWithOneEntry);

       // fault something else into memory and check again.
       region.get(82 * TOTAL_NUMBER_OF_BUCKETS);
       await().until(() -> dataStore.currentAllocatedMemory() == 2 * bucketSizeWithOneEntry);
     });
   }

   private void createPartitionedRegionWithOverflow() {
     Cache cache = getCache();

     File[] diskDirs = new File[] {overflowDirectory};

     DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
     diskStoreFactory.setDiskDirs(diskDirs);
     DiskStore diskStore = diskStoreFactory.create(DISK_STORE_NAME);

     PartitionAttributesFactory paf = new PartitionAttributesFactory();
     paf.setRedundantCopies(1);
     paf.setLocalMaxMemory(200);
     paf.setTotalNumBuckets(TOTAL_NUMBER_OF_BUCKETS);

     RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setDiskStoreName(diskStore.getName());
     regionFactory.setDiskSynchronous(true);
     regionFactory.setEvictionAttributes(createLRUEntryAttributes(2, OVERFLOW_TO_DISK));
     regionFactory.setPartitionAttributes(paf.create());

     regionFactory.create(REGION_NAME);
   }

   private void createPartitionedRegion(final int localMaxMemory) {
     PartitionAttributesFactory paf = new PartitionAttributesFactory();
     paf.setRedundantCopies(1);
     paf.setLocalMaxMemory(localMaxMemory);
     paf.setTotalNumBuckets(TOTAL_NUMBER_OF_BUCKETS);

     RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setPartitionAttributes(paf.create());

     regionFactory.create(REGION_NAME);
   }

   private <K, V> Region<K, V> getRegion(String regionName) {
     return getCache().getRegion(regionName);
   }

   private PartitionedRegion getPartitionedRegion(String regionName) {
     return (PartitionedRegion) getCache().getRegion(regionName);
   }
 }
	/*
	* 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.cache.EvictionAction.OVERFLOW_TO_DISK;
	import static org.apache.geode.cache.EvictionAttributes.createLRUEntryAttributes;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.io.File;

	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;

	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.DiskStore;
	import org.apache.geode.cache.DiskStoreFactory;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.cache.CacheTestCase;
	import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

	/**
	* This test verifies the size API for 100 thousand put operations (done synch/asynch) on
	* PartitionedRegions with different combinations of Scope and Redundancy (Scope DIST_ACK,
	* Redundancy 1 AND Scope DIST_NO_ACK, Redundancy 0).
	*/

	public class PartitionedRegionSizeDUnitTest extends CacheTestCase {

	private static final String DISK_STORE_NAME = "DISKSTORE";
	private static final String REGION_NAME = "PR";
	private static final int CNT = 100;
	private static final int TOTAL_NUMBER_OF_BUCKETS = 5;

	private File overflowDirectory;

	private VM vm0;
	private VM vm1;
	private VM vm2;
	private VM vm3;

	@Rule
	public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

	@Before
	public void setUp() throws Exception {
	vm0 = VM.getVM(0);
	vm1 = VM.getVM(1);
	vm2 = VM.getVM(2);
	vm3 = VM.getVM(3);

	overflowDirectory = temporaryFolder.newFolder("overflowDir");
	}

	/**
	* This test method invokes methods doing size validation on PRs.
	*/
	@Test
	public void testSize() throws Exception {
	// Create PRs with dataStore on 3 VMs
	vm0.invoke(() -> createPartitionedRegion(200));
	vm1.invoke(() -> createPartitionedRegion(200));
	vm2.invoke(() -> createPartitionedRegion(200));

	// Create only accessor on 4th VM
	vm3.invoke(() -> createPartitionedRegion(0));

	// Do put operations on PR synchronously.
	vm3.invoke(() -> {
	Region<Integer, Integer> region =
	getCache().getRegion(PartitionedRegionSizeDUnitTest.REGION_NAME);
	for (int k = 0; k < CNT; k++) {
	region.put(k, k);
	}
	});

	// Validate the size against the total put operations
	vm3.invoke(() -> {
	Region region = getCache().getRegion(PartitionedRegionSizeDUnitTest.REGION_NAME);
	await().until(() -> region.size() == CNT);
	});
	}

	/**
	* Regression test for TRAC #39868
	*
	* <p>
	* TRAC #39868: PartitionMemberDetails.getSize() reports negative PR sizes when redundancy is 0
	*/
	@Test
	public void testBug39868() throws Exception {
	vm0.invoke(() -> createPartitionedRegion(200));

	vm0.invoke(() -> {
	Region<Integer, byte[]> region = getRegion(REGION_NAME);
	for (int i = 0; i < 100; i++) {
	region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
	}
	});

	vm1.invoke(() -> createPartitionedRegion(200));

	vm0.invoke(() -> {
	Region<Integer, byte[]> region = getRegion(REGION_NAME);
	for (int i = 0; i < 100; i++) {
	region.destroy(i * TOTAL_NUMBER_OF_BUCKETS);
	}
	});

	vm1.invoke(() -> {
	PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
	long bytes = partitionedRegion.getDataStore().currentAllocatedMemory();
	assertThat(bytes).isEqualTo(0);
	});
	}

	@Test
	public void testByteSize() throws Exception {
	vm0.invoke(() -> createPartitionedRegion(200));
	vm1.invoke(() -> createPartitionedRegion(200));

	long bucketSizeWithOneEntry = vm0.invoke(() -> {
	Region<Integer, byte[]> region = getRegion(REGION_NAME);
	region.put(0, new byte[100]);

	PartitionedRegion partitionedRegion = (PartitionedRegion) region;
	PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
	long size = dataStore.getBucketSize(0);

	for (int i = 1; i < 100; i++) {
	region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
	}
	await().until(() -> dataStore.getBucketsManaged() == (short) 1);

	// make sure the size is proportional to the amount of data
	await().until(() -> dataStore.getBucketSize(0) == 100 * size);

	// destroy and invalidate entries and make sure the size goes down
	for (int i = 0; i < 25; i++) {
	region.destroy(i * TOTAL_NUMBER_OF_BUCKETS);
	}

	for (int i = 25; i < 50; i++) {
	region.invalidate(i * TOTAL_NUMBER_OF_BUCKETS);
	}

	await().until(() -> dataStore.getBucketSize(0) == 50 * size);

	// put some larger values in and make sure the size goes up
	for (int i = 50; i < 75; i++) {
	region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[150]);
	}

	// Now put in some smaller values and see if the size balances out
	for (int i = 75; i < 100; i++) {
	region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[50]);
	}

	await().until(() -> dataStore.getBucketSize(0) == 50 * size);

	return size;
	});

	vm1.invoke(() -> {
	PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
	await().until(
	() -> partitionedRegion.getDataStore().getBucketSize(0) == 50 * bucketSizeWithOneEntry);
	});

	vm1.invoke(() -> {
	PartitionedRegion partitionedRegion = getPartitionedRegion(REGION_NAME);
	PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
	await().until(() -> dataStore.currentAllocatedMemory() == 50 * bucketSizeWithOneEntry);
	});
	}

	@Test
	public void testByteSizeWithEviction() throws Exception {
	vm0.invoke(this::createPartitionedRegionWithOverflow);

	long bucketSizeWithOneEntry = vm0.invoke(() -> {
	Region<Integer, byte[]> region = getRegion(REGION_NAME);
	region.put(0, new byte[100]);

	PartitionedRegion partitionedRegion = (PartitionedRegion) region;
	PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
	long size = dataStore.getBucketSize(0);

	for (int i = 1; i < 100; i++) {
	region.put(i * TOTAL_NUMBER_OF_BUCKETS, new byte[100]);
	}
	await().until(() -> dataStore.getBucketsManaged() == (short) 1);
	return size;
	});

	vm0.invoke(() -> {
	Region<Integer, byte[]> region = getRegion(REGION_NAME);
	PartitionedRegion partitionedRegion = (PartitionedRegion) region;
	PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();

	// there should only be 2 items in memory
	await().until(() -> dataStore.currentAllocatedMemory() == 2 * bucketSizeWithOneEntry);

	// fault something else into memory and check again.
	region.get(82 * TOTAL_NUMBER_OF_BUCKETS);
	await().until(() -> dataStore.currentAllocatedMemory() == 2 * bucketSizeWithOneEntry);
	});
	}

	private void createPartitionedRegionWithOverflow() {
	Cache cache = getCache();

	File[] diskDirs = new File[] {overflowDirectory};

	DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
	diskStoreFactory.setDiskDirs(diskDirs);
	DiskStore diskStore = diskStoreFactory.create(DISK_STORE_NAME);

	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setRedundantCopies(1);
	paf.setLocalMaxMemory(200);
	paf.setTotalNumBuckets(TOTAL_NUMBER_OF_BUCKETS);

	RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setDiskStoreName(diskStore.getName());
	regionFactory.setDiskSynchronous(true);
	regionFactory.setEvictionAttributes(createLRUEntryAttributes(2, OVERFLOW_TO_DISK));
	regionFactory.setPartitionAttributes(paf.create());

	regionFactory.create(REGION_NAME);
	}

	private void createPartitionedRegion(final int localMaxMemory) {
	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setRedundantCopies(1);
	paf.setLocalMaxMemory(localMaxMemory);
	paf.setTotalNumBuckets(TOTAL_NUMBER_OF_BUCKETS);

	RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setPartitionAttributes(paf.create());

	regionFactory.create(REGION_NAME);
	}

	private <K, V> Region<K, V> getRegion(String regionName) {
	return getCache().getRegion(regionName);
	}

	private PartitionedRegion getPartitionedRegion(String regionName) {
	return (PartitionedRegion) getCache().getRegion(regionName);
	}
	}