geode-core/src/integrationTest/java/org/apache/geode/internal/offheap/OffHeapValidationJUnitTest.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.offheap;

 import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
 import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;

 import java.io.IOException;
 import java.io.Serializable;
 import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Hashtable;
 import java.util.IdentityHashMap;
 import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Stack;
 import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.UUID;
 import java.util.Vector;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;

 import org.apache.geode.DataSerializer;
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.compression.SnappyCompressor;
 import org.apache.geode.distributed.ConfigurationProperties;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.Version;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.test.junit.categories.OffHeapTest;

 /**
  * Basic integration tests for validating the off-heap implementation.
  *
  */
 @Category({OffHeapTest.class})
 public class OffHeapValidationJUnitTest {

   private GemFireCacheImpl cache;

   @Before
   public void setUp() throws Exception {
     this.cache = createCache();
   }

   @After
   public void tearDown() throws Exception {
     closeCache(this.cache);
   }

   protected GemFireCacheImpl createCache() {
     Properties props = new Properties();
     props.setProperty(LOCATORS, "");
     props.setProperty(MCAST_PORT, "0");
     props.setProperty(ConfigurationProperties.OFF_HEAP_MEMORY_SIZE, getOffHeapMemorySize());
     GemFireCacheImpl result = (GemFireCacheImpl) new CacheFactory(props).create();
     return result;
   }

   protected void closeCache(GemFireCacheImpl gfc) {
     gfc.close();
   }

   protected String getOffHeapMemorySize() {
     return "2m";
   }

   protected RegionShortcut getRegionShortcut() {
     return RegionShortcut.REPLICATE;
   }

   protected String getRegionName() {
     return "region1";
   }

   @Test
   public void testMemoryInspection() throws IOException {
     // validate initial state
     MemoryAllocator allocator = this.cache.getOffHeapStore();
     assertNotNull(allocator);
     MemoryInspector inspector = allocator.getMemoryInspector();
     assertNotNull(inspector);
     inspector.createSnapshot();
     try {
       MemoryBlock firstBlock = inspector.getFirstBlock();
       assertNotNull(firstBlock);
       assertEquals(1024 * 1024 * 2, firstBlock.getBlockSize());
       assertEquals("N/A", firstBlock.getDataType());
       assertEquals(-1, firstBlock.getFreeListId());
       assertTrue(firstBlock.getAddress() > 0);
       assertNull(firstBlock.getNextBlock());
       assertEquals(0, firstBlock.getRefCount());
       assertEquals(0, firstBlock.getSlabId());
       assertEquals(MemoryBlock.State.UNUSED, firstBlock.getState());
       assertFalse(firstBlock.isCompressed());
       assertFalse(firstBlock.isSerialized());
     } finally {
       inspector.clearSnapshot();
     }

     // create off-heap region
     Region<Object, Object> region = this.cache.createRegionFactory(getRegionShortcut())
         .setOffHeap(true).create(getRegionName());
     Region<Object, Object> compressedRegion = this.cache.createRegionFactory(getRegionShortcut())
         .setOffHeap(true).setCompressor(SnappyCompressor.getDefaultInstance())
         .create(getRegionName() + "Compressed");

     // perform some ops
     List<ExpectedValues> expected = new ArrayList<ExpectedValues>();

     // Chunk.OFF_HEAP_HEADER_SIZE + 4 ?

     putString(region, expected);
     putCompressedString(compressedRegion, expected);
     putDate(region, expected);
     putByteArray(region, expected);
     putCompressedByteArray(compressedRegion, expected);
     putByteArrayArray(region, expected);
     putShortArray(region, expected);
     putStringArray(region, expected);
     putObjectArray(region, expected);
     putArrayList(region, expected);
     putLinkedList(region, expected);
     putHashSet(region, expected);
     putLinkedHashSet(region, expected);
     putHashMap(region, expected);
     putIdentityHashMap(region, expected);
     putHashtable(region, expected);
     putProperties(region, expected);
     putVector(region, expected);
     putStack(region, expected);
     putTreeMap(region, expected);
     putTreeSet(region, expected);
     putClass(region, expected);
     putUUID(region, expected);
     putTimestamp(region, expected);
     putSerializableClass(region, expected);

     // TODO: USER_DATA_SERIALIZABLE

     // TODO: PDX

     // TODO: PDX_ENUM

     // TODO: GEMFIRE_ENUM

     // TODO: PDX_INLINE_ENUM

     // validate inspection
     inspector.createSnapshot();
     try {
       MemoryBlock firstBlock = inspector.getFirstBlock();
       assertEquals(MemoryBlock.State.UNUSED, firstBlock.getState());

       // System.out.println(((MemoryAllocatorImpl)inspector).getSnapshot());

       // sort the ExpectedValues into the same order as the MemberBlocks from inspector
       Collections.sort(expected, new Comparator<ExpectedValues>() {
         @Override
         public int compare(ExpectedValues o1, ExpectedValues o2) {
           return Long.valueOf(o1.memoryAddress).compareTo(o2.memoryAddress);
         }
       });

       int i = 0;
       MemoryBlock block = firstBlock.getNextBlock();
       while (block != null) {
         ExpectedValues values = expected.get(i);
         assertEquals(i + ":" + values.dataType, values.blockSize, block.getBlockSize());
         assertEquals(i + ":" + values.dataType, values.dataType, block.getDataType());
         assertEquals(i + ":" + values.dataType, values.freeListId, block.getFreeListId());
         assertEquals(i + ":" + values.dataType, values.memoryAddress, block.getAddress());
         assertEquals(i + ":" + values.dataType, values.refCount, block.getRefCount());
         assertEquals(i + ":" + values.dataType, values.slabId, block.getSlabId());
         assertEquals(i + ":" + values.dataType, values.isCompressed, block.isCompressed());
         assertEquals(i + ":" + values.dataType, values.isSerialized, block.isSerialized());
         // compare block.getDataValue() but only for String types
         if (values.dataType.equals("java.lang.String")) {
           Object obj = block.getDataValue();
           assertNotNull(block.toString(), obj);
           assertTrue(obj instanceof String);
           assertEquals("this is a string", (String) obj);
         }
         if ((values.dataType.contains("byte [")
             && values.dataType.lastIndexOf('[') == values.dataType.indexOf('['))
             || values.dataType.startsWith("compressed")) {
           assertTrue(
               "for dataType=" + values.dataType + " expected "
                   + Arrays.toString((byte[]) values.dataValue) + " but was "
                   + Arrays.toString((byte[]) block.getDataValue()),
               Arrays.equals((byte[]) values.dataValue, (byte[]) block.getDataValue()));
         } else if (values.dataType.contains("[")) {
           // TODO: multiple dimension arrays or non-byte arrays
         } else if (values.dataValue instanceof Collection) {
           int diff = joint((Collection<?>) values.dataValue, (Collection<?>) block.getDataValue());
           assertEquals(i + ":" + values.dataType, 0, diff);
         } else if (values.dataValue instanceof IdentityHashMap) {
           // TODO
         } else if (values.dataValue instanceof Map) {
           int diff = joint((Map<?, ?>) values.dataValue, (Map<?, ?>) block.getDataValue());
           assertEquals(i + ":" + values.dataType, 0, diff);
         } else {
           assertEquals(i + ":" + values.dataType, values.dataValue, block.getDataValue());
         }
         block = block.getNextBlock();
         i++;
       }
       assertEquals("All blocks: " + inspector.getAllBlocks(), expected.size(), i);
     } finally {
       inspector.clearSnapshot();
     }

     // perform more ops

     // validate more inspection

   }

   /**
    * Returns -1 if c1 is missing an element in c2, 1 if c2 is missing an element in c1, or 0 is they
    * contain the exact same elements.
    *
    * @throws NullPointerException if either c1 or c2 is null
    */
   private static int joint(Collection<?> c1, Collection<?> c2) {
     if (c1.size() < c2.size()) {
       return -1;
     } else if (c2.size() < c1.size()) {
       return 1;
     }
     Collection<Object> c3 = new ArrayList<Object>();
     c3.addAll(c1);
     c3.removeAll(c2);
     if (c3.size() > 0) {
       return -1;
     }
     c3.addAll(c2);
     c3.removeAll(c1);
     if (c3.size() > 0) {
       return 1;
     }
     return 0;
   }

   /**
    * Returns -1 if m1 is missing a key in m2, 1 if m2 is missing a key in m1, or 0 is they contain
    * the exact same keys.
    *
    * @throws NullPointerException if either c1 or c2 is null
    */
   private static int joint(Map<?, ?> m1, Map<?, ?> m2) {
     if (m1.size() < m2.size()) {
       return -1;
     } else if (m2.size() < m1.size()) {
       return 1;
     }
     Collection<Object> c3 = new ArrayList<Object>();
     c3.addAll(m1.keySet());
     c3.removeAll(m2.keySet());
     if (c3.size() > 0) {
       return -1;
     }
     c3.addAll(m2.keySet());
     c3.removeAll(m1.keySet());
     if (c3.size() > 0) {
       return 1;
     }
     return 0;
   }

   private long getMemoryAddress(Region region, String key) {
     Object entry = ((LocalRegion) region).getRegionEntry(key).getValue();
     assertTrue(entry instanceof OffHeapStoredObject);
     long memoryAddress = ((OffHeapStoredObject) entry).getAddress();
     assertTrue(memoryAddress > 0);
     return memoryAddress;
   }

   private void putString(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyString";
     String value = "this is a string";
     region.put(key, value);
     expected.add(new ExpectedValues(value, value.length() * 2, "java.lang.String", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putCompressedString(Region<Object, Object> region, List<ExpectedValues> expected)
       throws IOException {
     String key = "keyString";
     String value = "this is a string";
     region.put(key, value);
     HeapDataOutputStream hdos = new HeapDataOutputStream(Version.CURRENT);
     DataSerializer.writeObject(value, hdos);
     byte[] uncompressedBytes = hdos.toByteArray();
     byte[] expectedValue = SnappyCompressor.getDefaultInstance().compress(uncompressedBytes);
     expected.add(
         new ExpectedValues(expectedValue, 32, "compressed object of size " + expectedValue.length,
             -1, getMemoryAddress(region, key), 1, 0, true, true));
   }

   private void putDate(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyDate";
     Date value = new Date();
     region.put(key, value);
     expected.add(new ExpectedValues(value, 24, "java.util.Date", -1, getMemoryAddress(region, key),
         1, 0, false, true));
   }

   private void putByteArray(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyByteArray";
     byte[] value = new byte[10];
     region.put(key, value);
     expected.add(new ExpectedValues(value, 24, "byte[10]", -1, getMemoryAddress(region, key), 1, 0,
         false, false));
   }

   private void putCompressedByteArray(Region<Object, Object> region, List<ExpectedValues> expected)
       throws IOException {
     String key = "keyByteArray";
     byte[] value = new byte[10];
     region.put(key, value);
     byte[] expectedValue = SnappyCompressor.getDefaultInstance().compress(value);
     expected
         .add(new ExpectedValues(expectedValue, 24, "compressed byte[" + expectedValue.length + "]",
             -1, getMemoryAddress(region, key), 1, 0, true, false));
   }

   private void putByteArrayArray(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyByteArrayArray";
     byte[][] value = new byte[10][10];
     region.put(key, value);
     expected.add(new ExpectedValues(value, 120, "byte[][]", -1, getMemoryAddress(region, key), 1, 0,
         false, true));
   }

   private void putShortArray(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyShortArray(";
     short[] value = new short[10];
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "short[]", -1, getMemoryAddress(region, key), 1, 0,
         false, true));
   }

   private void putStringArray(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyStringArray";
     String[] value = new String[10];
     region.put(key, value);
     expected.add(new ExpectedValues(value, 24, "java.lang.String[]", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putObjectArray(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyObjectArray";
     Object[] value = new Object[10];
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.lang.Object[]", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putArrayList(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyArrayList";
     ArrayList<Object> value = new ArrayList<Object>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.ArrayList", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putLinkedList(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyLinkedList";
     LinkedList<Object> value = new LinkedList<Object>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.LinkedList", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putHashSet(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyHashSet";
     HashSet<Object> value = new HashSet<Object>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.HashSet", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putLinkedHashSet(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyLinkedHashSet";
     LinkedHashSet<Object> value = new LinkedHashSet<Object>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.LinkedHashSet", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putHashMap(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyHashMap";
     HashMap<Object, Object> value = new HashMap<Object, Object>();
     value.put("1", "string 1");
     value.put("2", "string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.util.HashMap", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putIdentityHashMap(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyIdentityHashMap";
     IdentityHashMap<Object, Object> value = new IdentityHashMap<Object, Object>();
     value.put("1", "string 1");
     value.put("2", "string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.util.IdentityHashMap", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putHashtable(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyHashtable";
     Hashtable<Object, Object> value = new Hashtable<Object, Object>();
     value.put("1", "string 1");
     value.put("2", "string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.util.Hashtable", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putProperties(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyProperties";
     Properties value = new Properties();
     value.put("1", "string 1");
     value.put("2", "string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.util.Properties", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putVector(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyVector";
     Vector<String> value = new Vector<String>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.Vector", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putStack(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyStack";
     Stack<String> value = new Stack<String>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.Stack", -1, getMemoryAddress(region, key),
         1, 0, false, true));
   }

   private void putTreeMap(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyTreeMap";
     TreeMap<String, String> value = new TreeMap<String, String>();
     value.put("1", "string 1");
     value.put("2", "string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 48, "java.util.TreeMap", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putTreeSet(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyTreeSet";
     TreeSet<String> value = new TreeSet<String>();
     value.add("string 1");
     value.add("string 2");
     region.put(key, value);
     expected.add(new ExpectedValues(value, 40, "java.util.TreeSet", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putClass(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyClass";
     Class<String> value = String.class;
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.lang.Class", -1, getMemoryAddress(region, key),
         1, 0, false, true));
   }

   private void putUUID(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyUUID";
     UUID value = UUID.randomUUID();
     region.put(key, value);
     expected.add(new ExpectedValues(value, 32, "java.util.UUID", -1, getMemoryAddress(region, key),
         1, 0, false, true));
   }

   private void putTimestamp(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keyTimestamp";
     Timestamp value = new Timestamp(System.currentTimeMillis());
     region.put(key, value);
     expected.add(new ExpectedValues(value, 24, "java.sql.Timestamp", -1,
         getMemoryAddress(region, key), 1, 0, false, true));
   }

   private void putSerializableClass(Region<Object, Object> region, List<ExpectedValues> expected) {
     String key = "keySerializableClass";
     SerializableClass value = new SerializableClass();
     region.put(key, value);
     expected.add(
         new ExpectedValues(value, 112, "java.io.Serializable:" + SerializableClass.class.getName(),
             -1, getMemoryAddress(region, key), 1, 0, false, true));
   }

   static class ExpectedValues {
     final Object dataValue;
     final int blockSize;
     final String dataType;
     final int freeListId;
     final long memoryAddress;
     final int refCount;
     final int slabId;
     final boolean isCompressed;
     final boolean isSerialized;

     ExpectedValues(Object dataValue, int blockSize, String dataType, int freeListId,
         long memoryAddress, int refCount, int slabId, boolean isCompressed, boolean isSerialized) {
       this.dataValue = dataValue;
       this.blockSize = blockSize;
       this.dataType = dataType;
       this.freeListId = freeListId;
       this.memoryAddress = memoryAddress;
       this.refCount = refCount;
       this.slabId = slabId;
       this.isCompressed = isCompressed;
       this.isSerialized = isSerialized;
     }
   }

   @SuppressWarnings("serial")
   public static class SerializableClass implements Serializable {
     public boolean equals(Object obj) {
       return obj instanceof SerializableClass;
     }

     public int hashCode() {
       return 42;
     }
   }
 }
	/*
	* 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.offheap;

	import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
	import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertNotNull;
	import static org.junit.Assert.assertNull;
	import static org.junit.Assert.assertTrue;

	import java.io.IOException;
	import java.io.Serializable;
	import java.sql.Timestamp;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.Comparator;
	import java.util.Date;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.Hashtable;
	import java.util.IdentityHashMap;
	import java.util.LinkedHashSet;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Map;
	import java.util.Properties;
	import java.util.Stack;
	import java.util.TreeMap;
	import java.util.TreeSet;
	import java.util.UUID;
	import java.util.Vector;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;
	import org.junit.experimental.categories.Category;

	import org.apache.geode.DataSerializer;
	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.compression.SnappyCompressor;
	import org.apache.geode.distributed.ConfigurationProperties;
	import org.apache.geode.internal.HeapDataOutputStream;
	import org.apache.geode.internal.Version;
	import org.apache.geode.internal.cache.GemFireCacheImpl;
	import org.apache.geode.internal.cache.LocalRegion;
	import org.apache.geode.test.junit.categories.OffHeapTest;

	/**
	* Basic integration tests for validating the off-heap implementation.
	*
	*/
	@Category({OffHeapTest.class})
	public class OffHeapValidationJUnitTest {

	private GemFireCacheImpl cache;

	@Before
	public void setUp() throws Exception {
	this.cache = createCache();
	}

	@After
	public void tearDown() throws Exception {
	closeCache(this.cache);
	}

	protected GemFireCacheImpl createCache() {
	Properties props = new Properties();
	props.setProperty(LOCATORS, "");
	props.setProperty(MCAST_PORT, "0");
	props.setProperty(ConfigurationProperties.OFF_HEAP_MEMORY_SIZE, getOffHeapMemorySize());
	GemFireCacheImpl result = (GemFireCacheImpl) new CacheFactory(props).create();
	return result;
	}

	protected void closeCache(GemFireCacheImpl gfc) {
	gfc.close();
	}

	protected String getOffHeapMemorySize() {
	return "2m";
	}

	protected RegionShortcut getRegionShortcut() {
	return RegionShortcut.REPLICATE;
	}

	protected String getRegionName() {
	return "region1";
	}

	@Test
	public void testMemoryInspection() throws IOException {
	// validate initial state
	MemoryAllocator allocator = this.cache.getOffHeapStore();
	assertNotNull(allocator);
	MemoryInspector inspector = allocator.getMemoryInspector();
	assertNotNull(inspector);
	inspector.createSnapshot();
	try {
	MemoryBlock firstBlock = inspector.getFirstBlock();
	assertNotNull(firstBlock);
	assertEquals(1024 * 1024 * 2, firstBlock.getBlockSize());
	assertEquals("N/A", firstBlock.getDataType());
	assertEquals(-1, firstBlock.getFreeListId());
	assertTrue(firstBlock.getAddress() > 0);
	assertNull(firstBlock.getNextBlock());
	assertEquals(0, firstBlock.getRefCount());
	assertEquals(0, firstBlock.getSlabId());
	assertEquals(MemoryBlock.State.UNUSED, firstBlock.getState());
	assertFalse(firstBlock.isCompressed());
	assertFalse(firstBlock.isSerialized());
	} finally {
	inspector.clearSnapshot();
	}

	// create off-heap region
	Region<Object, Object> region = this.cache.createRegionFactory(getRegionShortcut())
	.setOffHeap(true).create(getRegionName());
	Region<Object, Object> compressedRegion = this.cache.createRegionFactory(getRegionShortcut())
	.setOffHeap(true).setCompressor(SnappyCompressor.getDefaultInstance())
	.create(getRegionName() + "Compressed");

	// perform some ops
	List<ExpectedValues> expected = new ArrayList<ExpectedValues>();

	// Chunk.OFF_HEAP_HEADER_SIZE + 4 ?

	putString(region, expected);
	putCompressedString(compressedRegion, expected);
	putDate(region, expected);
	putByteArray(region, expected);
	putCompressedByteArray(compressedRegion, expected);
	putByteArrayArray(region, expected);
	putShortArray(region, expected);
	putStringArray(region, expected);
	putObjectArray(region, expected);
	putArrayList(region, expected);
	putLinkedList(region, expected);
	putHashSet(region, expected);
	putLinkedHashSet(region, expected);
	putHashMap(region, expected);
	putIdentityHashMap(region, expected);
	putHashtable(region, expected);
	putProperties(region, expected);
	putVector(region, expected);
	putStack(region, expected);
	putTreeMap(region, expected);
	putTreeSet(region, expected);
	putClass(region, expected);
	putUUID(region, expected);
	putTimestamp(region, expected);
	putSerializableClass(region, expected);

	// TODO: USER_DATA_SERIALIZABLE

	// TODO: PDX

	// TODO: PDX_ENUM

	// TODO: GEMFIRE_ENUM

	// TODO: PDX_INLINE_ENUM

	// validate inspection
	inspector.createSnapshot();
	try {
	MemoryBlock firstBlock = inspector.getFirstBlock();
	assertEquals(MemoryBlock.State.UNUSED, firstBlock.getState());

	// System.out.println(((MemoryAllocatorImpl)inspector).getSnapshot());

	// sort the ExpectedValues into the same order as the MemberBlocks from inspector
	Collections.sort(expected, new Comparator<ExpectedValues>() {
	@Override
	public int compare(ExpectedValues o1, ExpectedValues o2) {
	return Long.valueOf(o1.memoryAddress).compareTo(o2.memoryAddress);
	}
	});

	int i = 0;
	MemoryBlock block = firstBlock.getNextBlock();
	while (block != null) {
	ExpectedValues values = expected.get(i);
	assertEquals(i + ":" + values.dataType, values.blockSize, block.getBlockSize());
	assertEquals(i + ":" + values.dataType, values.dataType, block.getDataType());
	assertEquals(i + ":" + values.dataType, values.freeListId, block.getFreeListId());
	assertEquals(i + ":" + values.dataType, values.memoryAddress, block.getAddress());
	assertEquals(i + ":" + values.dataType, values.refCount, block.getRefCount());
	assertEquals(i + ":" + values.dataType, values.slabId, block.getSlabId());
	assertEquals(i + ":" + values.dataType, values.isCompressed, block.isCompressed());
	assertEquals(i + ":" + values.dataType, values.isSerialized, block.isSerialized());
	// compare block.getDataValue() but only for String types
	if (values.dataType.equals("java.lang.String")) {
	Object obj = block.getDataValue();
	assertNotNull(block.toString(), obj);
	assertTrue(obj instanceof String);
	assertEquals("this is a string", (String) obj);
	}
	if ((values.dataType.contains("byte [")
	&& values.dataType.lastIndexOf('[') == values.dataType.indexOf('['))
	\|\| values.dataType.startsWith("compressed")) {
	assertTrue(
	"for dataType=" + values.dataType + " expected "
	+ Arrays.toString((byte[]) values.dataValue) + " but was "
	+ Arrays.toString((byte[]) block.getDataValue()),
	Arrays.equals((byte[]) values.dataValue, (byte[]) block.getDataValue()));
	} else if (values.dataType.contains("[")) {
	// TODO: multiple dimension arrays or non-byte arrays
	} else if (values.dataValue instanceof Collection) {
	int diff = joint((Collection<?>) values.dataValue, (Collection<?>) block.getDataValue());
	assertEquals(i + ":" + values.dataType, 0, diff);
	} else if (values.dataValue instanceof IdentityHashMap) {
	// TODO
	} else if (values.dataValue instanceof Map) {
	int diff = joint((Map<?, ?>) values.dataValue, (Map<?, ?>) block.getDataValue());
	assertEquals(i + ":" + values.dataType, 0, diff);
	} else {
	assertEquals(i + ":" + values.dataType, values.dataValue, block.getDataValue());
	}
	block = block.getNextBlock();
	i++;
	}
	assertEquals("All blocks: " + inspector.getAllBlocks(), expected.size(), i);
	} finally {
	inspector.clearSnapshot();
	}

	// perform more ops

	// validate more inspection

	}

	/**
	* Returns -1 if c1 is missing an element in c2, 1 if c2 is missing an element in c1, or 0 is they
	* contain the exact same elements.
	*
	* @throws NullPointerException if either c1 or c2 is null
	*/
	private static int joint(Collection<?> c1, Collection<?> c2) {
	if (c1.size() < c2.size()) {
	return -1;
	} else if (c2.size() < c1.size()) {
	return 1;
	}
	Collection<Object> c3 = new ArrayList<Object>();
	c3.addAll(c1);
	c3.removeAll(c2);
	if (c3.size() > 0) {
	return -1;
	}
	c3.addAll(c2);
	c3.removeAll(c1);
	if (c3.size() > 0) {
	return 1;
	}
	return 0;
	}

	/**
	* Returns -1 if m1 is missing a key in m2, 1 if m2 is missing a key in m1, or 0 is they contain
	* the exact same keys.
	*
	* @throws NullPointerException if either c1 or c2 is null
	*/
	private static int joint(Map<?, ?> m1, Map<?, ?> m2) {
	if (m1.size() < m2.size()) {
	return -1;
	} else if (m2.size() < m1.size()) {
	return 1;
	}
	Collection<Object> c3 = new ArrayList<Object>();
	c3.addAll(m1.keySet());
	c3.removeAll(m2.keySet());
	if (c3.size() > 0) {
	return -1;
	}
	c3.addAll(m2.keySet());
	c3.removeAll(m1.keySet());
	if (c3.size() > 0) {
	return 1;
	}
	return 0;
	}

	private long getMemoryAddress(Region region, String key) {
	Object entry = ((LocalRegion) region).getRegionEntry(key).getValue();
	assertTrue(entry instanceof OffHeapStoredObject);
	long memoryAddress = ((OffHeapStoredObject) entry).getAddress();
	assertTrue(memoryAddress > 0);
	return memoryAddress;
	}

	private void putString(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyString";
	String value = "this is a string";
	region.put(key, value);
	expected.add(new ExpectedValues(value, value.length() * 2, "java.lang.String", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putCompressedString(Region<Object, Object> region, List<ExpectedValues> expected)
	throws IOException {
	String key = "keyString";
	String value = "this is a string";
	region.put(key, value);
	HeapDataOutputStream hdos = new HeapDataOutputStream(Version.CURRENT);
	DataSerializer.writeObject(value, hdos);
	byte[] uncompressedBytes = hdos.toByteArray();
	byte[] expectedValue = SnappyCompressor.getDefaultInstance().compress(uncompressedBytes);
	expected.add(
	new ExpectedValues(expectedValue, 32, "compressed object of size " + expectedValue.length,
	-1, getMemoryAddress(region, key), 1, 0, true, true));
	}

	private void putDate(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyDate";
	Date value = new Date();
	region.put(key, value);
	expected.add(new ExpectedValues(value, 24, "java.util.Date", -1, getMemoryAddress(region, key),
	1, 0, false, true));
	}

	private void putByteArray(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyByteArray";
	byte[] value = new byte[10];
	region.put(key, value);
	expected.add(new ExpectedValues(value, 24, "byte[10]", -1, getMemoryAddress(region, key), 1, 0,
	false, false));
	}

	private void putCompressedByteArray(Region<Object, Object> region, List<ExpectedValues> expected)
	throws IOException {
	String key = "keyByteArray";
	byte[] value = new byte[10];
	region.put(key, value);
	byte[] expectedValue = SnappyCompressor.getDefaultInstance().compress(value);
	expected
	.add(new ExpectedValues(expectedValue, 24, "compressed byte[" + expectedValue.length + "]",
	-1, getMemoryAddress(region, key), 1, 0, true, false));
	}

	private void putByteArrayArray(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyByteArrayArray";
	byte[][] value = new byte[10][10];
	region.put(key, value);
	expected.add(new ExpectedValues(value, 120, "byte[][]", -1, getMemoryAddress(region, key), 1, 0,
	false, true));
	}

	private void putShortArray(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyShortArray(";
	short[] value = new short[10];
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "short[]", -1, getMemoryAddress(region, key), 1, 0,
	false, true));
	}

	private void putStringArray(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyStringArray";
	String[] value = new String[10];
	region.put(key, value);
	expected.add(new ExpectedValues(value, 24, "java.lang.String[]", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putObjectArray(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyObjectArray";
	Object[] value = new Object[10];
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.lang.Object[]", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putArrayList(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyArrayList";
	ArrayList<Object> value = new ArrayList<Object>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.ArrayList", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putLinkedList(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyLinkedList";
	LinkedList<Object> value = new LinkedList<Object>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.LinkedList", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putHashSet(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyHashSet";
	HashSet<Object> value = new HashSet<Object>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.HashSet", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putLinkedHashSet(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyLinkedHashSet";
	LinkedHashSet<Object> value = new LinkedHashSet<Object>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.LinkedHashSet", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putHashMap(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyHashMap";
	HashMap<Object, Object> value = new HashMap<Object, Object>();
	value.put("1", "string 1");
	value.put("2", "string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.util.HashMap", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putIdentityHashMap(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyIdentityHashMap";
	IdentityHashMap<Object, Object> value = new IdentityHashMap<Object, Object>();
	value.put("1", "string 1");
	value.put("2", "string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.util.IdentityHashMap", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putHashtable(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyHashtable";
	Hashtable<Object, Object> value = new Hashtable<Object, Object>();
	value.put("1", "string 1");
	value.put("2", "string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.util.Hashtable", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putProperties(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyProperties";
	Properties value = new Properties();
	value.put("1", "string 1");
	value.put("2", "string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.util.Properties", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putVector(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyVector";
	Vector<String> value = new Vector<String>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.Vector", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putStack(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyStack";
	Stack<String> value = new Stack<String>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.Stack", -1, getMemoryAddress(region, key),
	1, 0, false, true));
	}

	private void putTreeMap(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyTreeMap";
	TreeMap<String, String> value = new TreeMap<String, String>();
	value.put("1", "string 1");
	value.put("2", "string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 48, "java.util.TreeMap", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putTreeSet(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyTreeSet";
	TreeSet<String> value = new TreeSet<String>();
	value.add("string 1");
	value.add("string 2");
	region.put(key, value);
	expected.add(new ExpectedValues(value, 40, "java.util.TreeSet", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putClass(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyClass";
	Class<String> value = String.class;
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.lang.Class", -1, getMemoryAddress(region, key),
	1, 0, false, true));
	}

	private void putUUID(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyUUID";
	UUID value = UUID.randomUUID();
	region.put(key, value);
	expected.add(new ExpectedValues(value, 32, "java.util.UUID", -1, getMemoryAddress(region, key),
	1, 0, false, true));
	}

	private void putTimestamp(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keyTimestamp";
	Timestamp value = new Timestamp(System.currentTimeMillis());
	region.put(key, value);
	expected.add(new ExpectedValues(value, 24, "java.sql.Timestamp", -1,
	getMemoryAddress(region, key), 1, 0, false, true));
	}

	private void putSerializableClass(Region<Object, Object> region, List<ExpectedValues> expected) {
	String key = "keySerializableClass";
	SerializableClass value = new SerializableClass();
	region.put(key, value);
	expected.add(
	new ExpectedValues(value, 112, "java.io.Serializable:" + SerializableClass.class.getName(),
	-1, getMemoryAddress(region, key), 1, 0, false, true));
	}

	static class ExpectedValues {
	final Object dataValue;
	final int blockSize;
	final String dataType;
	final int freeListId;
	final long memoryAddress;
	final int refCount;
	final int slabId;
	final boolean isCompressed;
	final boolean isSerialized;

	ExpectedValues(Object dataValue, int blockSize, String dataType, int freeListId,
	long memoryAddress, int refCount, int slabId, boolean isCompressed, boolean isSerialized) {
	this.dataValue = dataValue;
	this.blockSize = blockSize;
	this.dataType = dataType;
	this.freeListId = freeListId;
	this.memoryAddress = memoryAddress;
	this.refCount = refCount;
	this.slabId = slabId;
	this.isCompressed = isCompressed;
	this.isSerialized = isSerialized;
	}
	}

	@SuppressWarnings("serial")
	public static class SerializableClass implements Serializable {
	public boolean equals(Object obj) {
	return obj instanceof SerializableClass;
	}

	public int hashCode() {
	return 42;
	}
	}
	}