| /* |
| * 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.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 static org.junit.Assert.fail; |
| |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.stream.IntStream; |
| |
| import org.apache.commons.io.FileUtils; |
| import org.junit.Assert; |
| import org.junit.Ignore; |
| import org.junit.Test; |
| |
| import org.apache.geode.StatisticsFactory; |
| import org.apache.geode.cache.CacheWriterException; |
| import org.apache.geode.cache.DiskAccessException; |
| import org.apache.geode.cache.DiskStore; |
| import org.apache.geode.cache.DiskStoreFactory; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.RegionFactory; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.util.CacheWriterAdapter; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.Oplog.OPLOG_TYPE; |
| import org.apache.geode.internal.cache.entries.AbstractDiskLRURegionEntry; |
| import org.apache.geode.internal.cache.entries.DiskEntry; |
| import org.apache.geode.test.dunit.ThreadUtils; |
| |
| /** |
| * Testing Oplog API's |
| */ |
| public class OplogJUnitTest extends DiskRegionTestingBase { |
| |
| private boolean proceed = false; |
| |
| private final DiskRegionProperties diskProps = new DiskRegionProperties(); |
| |
| private long delta; |
| |
| private volatile boolean assertDone = false; |
| |
| private boolean failure = false; |
| |
| @Override |
| protected final void postSetUp() throws Exception { |
| diskProps.setDiskDirs(dirs); |
| DiskStoreImpl.SET_IGNORE_PREALLOCATE = true; |
| } |
| |
| @Override |
| protected final void postTearDown() throws Exception { |
| DiskStoreImpl.SET_IGNORE_PREALLOCATE = false; |
| } |
| |
| /** |
| * Test method for 'org.apache.geode.internal.cache.Oplog.isBackup()' |
| */ |
| @Test |
| public void testIsBackup() { |
| |
| InternalRegion overFlowAndPersistRegionRegion = |
| (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| assertTrue("Not correctly setup for overflow and persist", |
| overFlowAndPersistRegionRegion.getDiskRegion().isBackup()); |
| closeDown(overFlowAndPersistRegionRegion); |
| |
| InternalRegion overFlowOnlyRegion = |
| (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| assertFalse("Not correctly setup for overflow only mode", |
| overFlowOnlyRegion.getDiskRegion().isBackup()); |
| closeDown(overFlowOnlyRegion); |
| |
| InternalRegion persistOnlyRegion = (InternalRegion) DiskRegionHelperFactory |
| .getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertTrue("Not correctly setup for persist only mode", |
| persistOnlyRegion.getDiskRegion().isBackup()); |
| closeDown(persistOnlyRegion); |
| } |
| |
| /* |
| * Test method for 'org.apache.geode.internal.cache.Oplog.useSyncWrites()' |
| */ |
| @Test |
| public void testUseSyncWritesWhenSet() { |
| diskProps.setSynchronous(true); |
| InternalRegion syncOverFlowAndPersistRegion = |
| (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| assertTrue(syncOverFlowAndPersistRegion.getAttributes().isDiskSynchronous()); |
| closeDown(syncOverFlowAndPersistRegion); |
| |
| InternalRegion syncOverFlowOnlyRegion = |
| (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| assertTrue(syncOverFlowOnlyRegion.getAttributes().isDiskSynchronous()); |
| closeDown(syncOverFlowOnlyRegion); |
| |
| InternalRegion syncPersistOnlyRegion = (InternalRegion) DiskRegionHelperFactory |
| .getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertTrue(syncPersistOnlyRegion.getAttributes().isDiskSynchronous()); |
| closeDown(syncPersistOnlyRegion); |
| } |
| |
| @Test |
| public void testNotUseSyncWritesWhenNotSet() { |
| diskProps.setSynchronous(false); |
| InternalRegion asyncOverFlowAndPersistRegion = |
| (InternalRegion) DiskRegionHelperFactory.getAsyncOverFlowAndPersistRegion(cache, diskProps); |
| assertFalse(asyncOverFlowAndPersistRegion.getAttributes().isDiskSynchronous()); |
| closeDown(asyncOverFlowAndPersistRegion); |
| |
| InternalRegion asyncOverFlowOnlyRegion = |
| (InternalRegion) DiskRegionHelperFactory.getAsyncOverFlowOnlyRegion(cache, diskProps); |
| assertFalse(asyncOverFlowOnlyRegion.getAttributes().isDiskSynchronous()); |
| closeDown(asyncOverFlowOnlyRegion); |
| |
| InternalRegion asyncPersistOnlyRegion = |
| (InternalRegion) DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| assertFalse(asyncPersistOnlyRegion.getAttributes().isDiskSynchronous()); |
| closeDown(asyncPersistOnlyRegion); |
| } |
| |
| /** |
| * Test method for 'org.apache.geode.internal.cache.Oplog.clear(File)' |
| */ |
| @Test |
| public void testClear() { |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| putTillOverFlow(region); |
| region.clear(); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null); |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| putTillOverFlow(region); |
| region.clear(); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| assertTrue(" failed in get OverflowOnly ", region.get(0) == null); |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| put100Int(); |
| region.clear(); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertTrue(" failed in get PersistOnly ", region.get(0) == null); |
| closeDown(); |
| } |
| |
| /** |
| * Test method for 'org.apache.geode.internal.cache.Oplog.close()' |
| */ |
| @Test |
| public void testClose() { |
| { |
| deleteFiles(); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| Oplog oplog = dr.testHook_getChild(); |
| long id = oplog.getOplogId(); |
| oplog.close(); |
| |
| StatisticsFactory factory = cache.getDistributedSystem(); |
| Oplog newOplog = |
| new Oplog(id, dr.getOplogSet(), new DirectoryHolder(factory, dirs[0], 1000, 0)); |
| dr.getOplogSet().setChild(newOplog); |
| closeDown(); |
| } |
| { |
| deleteFiles(); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| dr.testHookCloseAllOverflowOplogs(); |
| checkIfContainsFile("OVERFLOW"); |
| closeDown(); |
| } |
| { |
| deleteFiles(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| Oplog oplog = dr.testHook_getChild(); |
| long id = oplog.getOplogId(); |
| oplog.close(); |
| StatisticsFactory factory = cache.getDistributedSystem(); |
| Oplog newOplog = |
| new Oplog(id, dr.getOplogSet(), new DirectoryHolder(factory, dirs[0], 1000, 2)); |
| dr.setChild(newOplog); |
| closeDown(); |
| } |
| |
| } |
| |
| private void closeDown(InternalRegion region) { |
| super.closeDown(region); |
| DiskRegion diskRegion = region != null ? region.getDiskRegion() : null; |
| if (diskRegion != null) { |
| diskRegion.getDiskStore().close(); |
| ((InternalCache) cache).removeDiskStore(diskRegion.getDiskStore()); |
| } |
| } |
| |
| @Override |
| protected void closeDown() { |
| DiskRegion dr = null; |
| if (region != null) { |
| dr = ((LocalRegion) region).getDiskRegion(); |
| } |
| super.closeDown(); |
| if (dr != null) { |
| dr.getDiskStore().close(); |
| ((LocalRegion) region).getGemFireCache().removeDiskStore(dr.getDiskStore()); |
| } |
| } |
| |
| private void checkIfContainsFile(String fileExtension) { |
| for (File dir : dirs) { |
| File[] files = dir.listFiles(); |
| for (File file : files) { |
| if (file.getAbsolutePath().endsWith(fileExtension)) { |
| fail("file " + file + " still exists after oplog.close()"); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Test method for 'org.apache.geode.internal.cache.Oplog.destroy()' |
| */ |
| @Test |
| public void testDestroy() { |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| put100Int(); |
| putTillOverFlow(region); |
| try { |
| region.destroy(0); |
| } catch (EntryNotFoundException e1) { |
| logWriter.error("Exception occurred", e1); |
| fail(" Entry not found when it was expected to be there"); |
| } |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null); |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| put100Int(); |
| putTillOverFlow(region); |
| try { |
| region.destroy(0); |
| } catch (EntryNotFoundException e1) { |
| logWriter.error("Exception occurred", e1); |
| fail(" Entry not found when it was expected to be there"); |
| } |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| assertTrue(" failed in get OverflowOnly ", region.get(0) == null); |
| |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| put100Int(); |
| try { |
| region.destroy(0); |
| } catch (EntryNotFoundException e1) { |
| logWriter.error("Exception occurred", e1); |
| fail(" Entry not found when it was expected to be there"); |
| } |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertTrue(" failed in get PersistOnly ", region.get(0) == null); |
| closeDown(); |
| |
| } |
| |
| /** |
| * Test method for 'org.apache.geode.internal.cache.Oplog.remove(long)' |
| */ |
| @Test |
| public void testRemove() { |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| putTillOverFlow(region); |
| region.remove(0); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null); |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| putTillOverFlow(region); |
| region.remove(0); |
| assertTrue(" failed in get OverflowOnly ", region.get(0) == null); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| closeDown(); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| put100Int(); |
| region.remove(0); |
| assertTrue(" failed in get PersistOnly ", region.get(0) == null); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| closeDown(); |
| |
| } |
| |
| /** |
| * Tests whether the data is written in the right format on the disk |
| * |
| */ |
| @Test |
| public void testFaultInOfValuesFromDisk() { |
| try { |
| // Asif First create a persist only disk region which is of aysnch |
| // & switch of OplOg type |
| diskProps.setMaxOplogSize(1000); |
| |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(true); |
| diskProps.setTimeInterval(-1); |
| diskProps.setOverflow(false); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| byte[] val = new byte[10]; |
| for (int i = 0; i < val.length; ++i) { |
| val[i] = (byte) i; |
| } |
| region.put(1, val); |
| |
| DiskEntry entry = ((DiskEntry) ((LocalRegion) region).basicGetEntry(1)); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| |
| val = (byte[]) dr.getNoBuffer(entry.getDiskId()); |
| for (int i = 0; i < val.length; ++i) { |
| if (val[i] != (byte) i) { |
| fail("Test for fault in from disk failed"); |
| } |
| } |
| val = (byte[]) DiskStoreImpl.convertBytesAndBitsIntoObject( |
| dr.getBytesAndBitsWithoutLock(entry.getDiskId(), true, false), (InternalCache) cache); |
| for (int i = 0; i < val.length; ++i) { |
| if (val[i] != (byte) i) { |
| fail("Test for fault in from disk failed"); |
| } |
| } |
| region.invalidate(1); |
| assertTrue(dr.getNoBuffer(entry.getDiskId()) == Token.INVALID); |
| |
| } catch (Exception e) { |
| logWriter.error("Exception occurred", e); |
| fail(e.toString()); |
| } |
| closeDown(); |
| } |
| |
| /** |
| * Tests the original ByteBufferPool gets transferred to the new Oplog for synch mode |
| * |
| */ |
| @Test |
| public void testByteBufferPoolTransferForSynchMode() { |
| diskProps.setMaxOplogSize(1024); |
| diskProps.setBytesThreshold(0); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(true); |
| diskProps.setTimeInterval(10000); |
| diskProps.setOverflow(false); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| // Populate data just below the switch over threshold |
| byte[] val = new byte[5]; |
| for (int i = 0; i < val.length; ++i) { |
| val[i] = (byte) i; |
| } |
| |
| region.put(1, val); |
| |
| ((LocalRegion) region).basicGetEntry(1); |
| Oplog old = dr.testHook_getChild(); |
| ByteBuffer oldWriteBuf = old.getWriteBuf(); |
| dr.forceRolling(); |
| region.put(2, val); |
| Oplog switched = dr.testHook_getChild(); |
| assertTrue(old != switched); |
| assertEquals(dr.getDiskStore().getPersistentOplogs().getChild(2), switched); |
| assertEquals(oldWriteBuf, switched.getWriteBuf()); |
| assertEquals(null, old.getWriteBuf()); |
| closeDown(); |
| |
| } |
| |
| /** |
| * Tests the bug which arises in case of asynch mode during oplog switching caused by conflation |
| * of create/destroy operation.The bug occurs if a create operation is followed by destroy but |
| * before destroy proceeds some other operation causes oplog switching |
| * |
| */ |
| @Test |
| public void testBug34615() { |
| final int MAX_OPLOG_SIZE = 100; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(false); |
| diskProps.setBytesThreshold(150); |
| |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| final byte[] val = new byte[50]; |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| final CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| @Override |
| public void afterConflation(ByteBuffer orig, ByteBuffer conflated) { |
| Thread th = new Thread(() -> region.put("2", new byte[75])); |
| assertNull(conflated); |
| th.start(); |
| ThreadUtils.join(th, 30 * 1000); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| |
| } |
| }); |
| |
| region.put("1", val); |
| region.remove("1"); |
| assertFalse(failureCause, testFailed); |
| CacheObserverHolder.setInstance(old); |
| closeDown(); |
| |
| } |
| |
| @Test |
| public void testConflation() throws Exception { |
| |
| final int MAX_OPLOG_SIZE = 1000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(false); |
| diskProps.setBytesThreshold(1500); |
| |
| final byte[] val = new byte[50]; |
| final byte[][] bb = new byte[2][]; |
| bb[0] = new byte[5]; |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| try { |
| region.put("1", val); |
| region.put("1", new byte[10]); |
| region.put("2", val); |
| region.put("2", new byte[100]); |
| region.create("3", null); |
| region.put("3", new byte[10]); |
| region.create("4", null); |
| region.put("4", new byte[0]); |
| |
| // tests for byte[][] |
| region.create("5", bb); |
| region.put("6", val); |
| region.put("6", bb); |
| region.create("7", null); |
| region.put("7", bb); |
| |
| region.create("8", new byte[9]); |
| region.invalidate("8"); |
| region.create("9", new byte[0]); |
| region.invalidate("9"); |
| |
| region.create("10", new byte[9]); |
| region.localInvalidate("10"); |
| region.create("11", new byte[0]); |
| region.localInvalidate("11"); |
| |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| dr.flushForTesting(); |
| byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDisk("1")); |
| assertEquals(val_1.length, 10); |
| byte[] val_2 = ((byte[]) ((LocalRegion) region).getValueOnDisk("2")); |
| assertEquals(val_2.length, 100); |
| byte[] val_3 = ((byte[]) ((LocalRegion) region).getValueOnDisk("3")); |
| assertEquals(val_3.length, 10); |
| byte[] val_4 = ((byte[]) ((LocalRegion) region).getValueOnDisk("4")); |
| assertEquals(val_4.length, 0); |
| byte[][] val_5 = (byte[][]) ((LocalRegion) region).getValueOnDisk("5"); |
| assertEquals(val_5.length, 2); |
| assertEquals(val_5[0].length, 5); |
| assertNull(val_5[1]); |
| byte[][] val_6 = (byte[][]) ((LocalRegion) region).getValueOnDisk("6"); |
| assertEquals(val_6.length, 2); |
| assertEquals(val_6[0].length, 5); |
| assertNull(val_6[1]); |
| byte[][] val_7 = (byte[][]) ((LocalRegion) region).getValueOnDisk("7"); |
| assertEquals(val_7.length, 2); |
| assertEquals(val_7[0].length, 5); |
| assertNull(val_7[1]); |
| Object val_8 = ((LocalRegion) region).getValueOnDisk("8"); |
| assertEquals(val_8, Token.INVALID); |
| Object val_9 = ((LocalRegion) region).getValueOnDisk("9"); |
| assertEquals(val_9, Token.INVALID); |
| Object val_10 = ((LocalRegion) region).getValueOnDisk("10"); |
| assertEquals(val_10, Token.LOCAL_INVALID); |
| Object val_11 = ((LocalRegion) region).getValueOnDisk("11"); |
| assertEquals(val_11, Token.LOCAL_INVALID); |
| |
| } finally { |
| closeDown(); |
| } |
| } |
| |
| /** |
| * This tests the retrieval of empty byte array when present in asynch buffers |
| * |
| */ |
| @Test |
| public void testGetEmptyByteArrayInAsynchBuffer() { |
| final int MAX_OPLOG_SIZE = 1000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(false); |
| diskProps.setBytesThreshold(1500); |
| |
| final byte[] val = new byte[50]; |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| try { |
| region.put("1", val); |
| region.put("1", new byte[0]); |
| byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDiskOrBuffer("1")); |
| assertEquals(val_1.length, 0); |
| } catch (Exception e) { |
| logWriter.error("Exception occurred", e); |
| fail("The test failed due to exception = " + e); |
| } |
| closeDown(); |
| } |
| |
| /** |
| * This tests the retrieval of empty byte array in synch mode |
| * |
| */ |
| @Test |
| public void testGetEmptyByteArrayInSynchMode() { |
| final int MAX_OPLOG_SIZE = 1000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setBytesThreshold(1500); |
| |
| final byte[] val = new byte[50]; |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| try { |
| region.put("1", val); |
| region.put("1", new byte[0]); |
| byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDiskOrBuffer("1")); |
| assertEquals(val_1.length, 0); |
| } catch (Exception e) { |
| logWriter.error("Exception occurred", e); |
| fail("The test failed due to exception = " + e); |
| } |
| closeDown(); |
| } |
| |
| /** |
| * This tests the bug which caused the oplogRoller to attempt to roll a removed entry whose value |
| * is Token.Removed This bug can occur if a remove operation causes oplog switching & hence roller |
| * thread gets notified, & the roller thread obtains the iterator of the concurrent region map |
| * before the remove |
| * |
| */ |
| @Test |
| public void testBug34702() { |
| final int MAX_OPLOG_SIZE = 500 * 2; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| final byte[] val = new byte[200]; |
| proceed = false; |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| region.put("key1", val); |
| region.put("key2", val); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| final CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| |
| @Override |
| public void afterSettingOplogOffSet(long offset) { |
| ((LocalRegion) region).getDiskRegion().forceRolling(); |
| // Let the operation thread yield to the Roller so that |
| // it is able to obtain the iterator of the concurrrent region map |
| // & thus get the reference to the entry which will contain |
| // value as Token.Removed as the entry though removed from |
| // concurrent |
| // map still will be available to the roller |
| Thread.yield(); |
| // Sleep for some time |
| try { |
| Thread.sleep(5000); |
| } catch (InterruptedException e) { |
| testFailed = true; |
| failureCause = "No guarantee that test is succesful"; |
| fail("No guarantee that test is succesful"); |
| } |
| } |
| |
| @Override |
| public void afterHavingCompacted() { |
| proceed = true; |
| synchronized (OplogJUnitTest.this) { |
| OplogJUnitTest.this.notify(); |
| } |
| } |
| }); |
| try { |
| region.destroy("key1"); |
| region.destroy("key2"); |
| } catch (Exception e1) { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(old); |
| fail("Test failed as entry deletion threw exception. Exception = " + e1); |
| } |
| // Wait for some time & check if the after having rolled callabck |
| // is issued sucessfully or not. |
| if (!proceed) { |
| synchronized (this) { |
| if (!proceed) { |
| try { |
| this.wait(20000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| // The test will automatically fail due to proceed flag |
| } |
| } |
| } |
| } |
| assertFalse(failureCause, testFailed); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(old); |
| |
| if (!proceed) { |
| fail("Test failed as afterHavingCompacted callabck not issued even after sufficient wait"); |
| } |
| closeDown(); |
| |
| } |
| |
| /** |
| * tests a potential deadlock situation if the operation causing a swithcing of Oplog is waiting |
| * for roller to free space. The problem can arise if the operation causing Oplog switching is |
| * going on an Entry , which already has its oplog ID referring to the Oplog being switched. In |
| * such case, when the roller will try to roll the entries referencing the current oplog , it will |
| * not be able to acquire the lock on the entry as the switching thread has already taken a lock |
| * on it. |
| * |
| */ |
| @Test |
| public void testRollingDeadlockSituation() { |
| final int MAX_OPLOG_SIZE = 2000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {1400}); |
| final byte[] val = new byte[500]; |
| proceed = false; |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| region.put("key1", val); |
| region.put("key1", val); |
| try { |
| region.put("key1", val); |
| } catch (DiskAccessException dae) { |
| logWriter.error("Exception occurred", dae); |
| fail( |
| "Test failed as DiskAccessException was encountered where as the operation should ideally have proceeded without issue . exception = " |
| + dae); |
| } |
| } |
| |
| /** |
| * This tests whether an empty byte array is correctly writtem to the disk as a zero value length |
| * operation & hence the 4 bytes field for recording the value length is absent & also since the |
| * value length is zero no byte for it should also get added. Similary during recover from HTree |
| * as well as Oplog , the empty byte array should be read correctly |
| * |
| */ |
| @Test |
| public void testEmptyByteArrayPutAndRecovery() { |
| CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| @Override |
| public void afterConflation(ByteBuffer origBB, ByteBuffer conflatedBB) { |
| if ((2 + 4 + 1 + EntryEventImpl.serialize("key1").length) != origBB.capacity()) { |
| failureCause = |
| "For a backup region, addition of an empty array should result in an offset of 6 bytes where as actual offset is =" |
| + origBB.capacity(); |
| testFailed = true; |
| } |
| Assert.assertTrue( |
| "For a backup region, addition of an empty array should result in an offset of 6 bytes where as actual offset is =" |
| + origBB.capacity(), |
| (2 + 4 + 1 + EntryEventImpl.serialize("key1").length) == origBB.capacity()); |
| |
| } |
| }); |
| try { |
| final int MAX_OPLOG_SIZE = 2000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {1400}); |
| final byte[] val = new byte[0]; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| region.put("key1", val); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| byte[] _val = (byte[]) region.get("key1"); |
| assertTrue( |
| "value of key1 after restarting the region is not an empty byte array. This may indicate problem in reading from Oplog", |
| _val.length == 0); |
| if (this.logWriter.infoEnabled()) { |
| this.logWriter.info( |
| "After first region close & opening again no problems encountered & hence Oplog has been read successfully."); |
| this.logWriter.info( |
| "Closing the region again without any operation done, would indicate that next time data will be loaded from HTree ."); |
| } |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| _val = (byte[]) region.get("key1"); |
| assertTrue( |
| "value of key1 after restarting the region is not an empty byte array. This may indicate problem in reading from HTRee", |
| _val.length == 0); |
| assertFalse(failureCause, testFailed); |
| // region.close(); |
| |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(old); |
| |
| } |
| } |
| |
| /** |
| * This is used to test bug 35012 where a remove operation on a key gets unrecorded due to |
| * switching of Oplog if it happens just after the remove operation has destroyed the in memory |
| * entry & is about to acquire the readlock in DiskRegion to record the same. If the Oplog has |
| * switched during that duration , the bug would appear |
| * |
| */ |
| |
| @Test |
| public void testBug35012() { |
| final int MAX_OPLOG_SIZE = 500; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| final byte[] val = new byte[200]; |
| try { |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| region.put("key1", val); |
| region.put("key2", val); |
| region.put("key3", val); |
| final Thread th = new Thread(() -> region.remove("key1")); |
| // main thread acquires the write lock |
| ((LocalRegion) region).getDiskRegion().acquireWriteLock(); |
| try { |
| th.start(); |
| Thread.yield(); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| dr.testHook_getChild().forceRolling(dr); |
| } finally { |
| ((LocalRegion) region).getDiskRegion().releaseWriteLock(); |
| } |
| ThreadUtils.join(th, 30 * 1000); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertEquals(region.size(), 2); |
| } catch (Exception e) { |
| this.logWriter.error("Exception occurred ", e); |
| fail("The test could not be completed because of exception .Exception=" + e); |
| } |
| closeDown(); |
| |
| } |
| |
| /** |
| * Tests if buffer size & time are not set , the asynch writer gets awakened on time basis of |
| * default 1 second |
| * |
| */ |
| @Test |
| public void testAsynchWriterAttribBehaviour1() throws Exception { |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| ((DiskStoreFactoryImpl) dsf).setMaxOplogSizeInBytes(10000); |
| File dir = new File("testingDirectoryDefault"); |
| dir.mkdir(); |
| dir.deleteOnExit(); |
| File[] dirs = {dir}; |
| dsf.setDiskDirs(dirs); |
| RegionFactory<Object, Object> factory = |
| cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT); |
| final long t1 = System.currentTimeMillis(); |
| DiskStore ds = dsf.create("test"); |
| factory.setDiskSynchronous(false); |
| factory.setDiskStoreName(ds.getName()); |
| factory.setScope(Scope.LOCAL); |
| region = factory.create("test"); |
| |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| CacheObserver old = CacheObserverHolder.setInstance( |
| |
| new CacheObserverAdapter() { |
| private long t2; |
| |
| @Override |
| public void goingToFlush() { |
| t2 = System.currentTimeMillis(); |
| delta = t2 - t1; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| synchronized (OplogJUnitTest.this) { |
| OplogJUnitTest.this.notify(); |
| } |
| } |
| }); |
| |
| region.put("key1", "111111111111"); |
| synchronized (this) { |
| if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) { |
| this.wait(10000); |
| assertFalse(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| } |
| } |
| CacheObserverHolder.setInstance(old); |
| // Windows clock has an accuracy of 15 ms. Accounting for the same. |
| assertTrue("delta is in miilliseconds=" + delta, delta >= 985); |
| closeDown(); |
| } |
| |
| /** |
| * Tests if buffer size is set but time is not set , the asynch writer gets awakened on buffer |
| * size basis |
| */ |
| @Ignore("TODO:DARREL_DISABLE: test is disabled") |
| @Test |
| public void testAsynchWriterAttribBehaviour2() throws Exception { |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| ((DiskStoreFactoryImpl) dsf).setMaxOplogSizeInBytes(10000); |
| dsf.setQueueSize(2); |
| File dir = new File("testingDirectoryDefault"); |
| dir.mkdir(); |
| dir.deleteOnExit(); |
| File[] dirs = {dir}; |
| dsf.setDiskDirs(dirs); |
| RegionFactory<Object, Object> factory = |
| cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT); |
| DiskStore ds = dsf.create("test"); |
| factory.setDiskSynchronous(false); |
| factory.setDiskStoreName(ds.getName()); |
| factory.setScope(Scope.LOCAL); |
| region = factory.create("test"); |
| |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| CacheObserver old = CacheObserverHolder.setInstance( |
| |
| new CacheObserverAdapter() { |
| |
| @Override |
| public void goingToFlush() { |
| synchronized (OplogJUnitTest.this) { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| OplogJUnitTest.this.notify(); |
| } |
| } |
| }); |
| |
| region.put("key1", new byte[25]); |
| Thread.sleep(1000); |
| assertTrue(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| region.put("key2", new byte[25]); |
| synchronized (this) { |
| if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) { |
| OplogJUnitTest.this.wait(10000); |
| assertFalse(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| } |
| } |
| CacheObserverHolder.setInstance(old); |
| closeDown(); |
| } |
| |
| /** |
| * Tests if buffer size & time interval are explicitly set to zero then the flush will occur due |
| * to asynchForceFlush or due to switching of Oplog |
| * |
| */ |
| @Test |
| public void testAsynchWriterAttribBehaviour3() throws Exception { |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| ((DiskStoreFactoryImpl) dsf).setMaxOplogSizeInBytes(500); |
| dsf.setQueueSize(0); |
| dsf.setTimeInterval(0); |
| File dir = new File("testingDirectoryDefault"); |
| dir.mkdir(); |
| dir.deleteOnExit(); |
| File[] dirs = {dir}; |
| dsf.setDiskDirs(dirs); |
| RegionFactory<Object, Object> factory = |
| cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT); |
| DiskStore ds = dsf.create("test"); |
| factory.setDiskSynchronous(false); |
| factory.setDiskStoreName(ds.getName()); |
| factory.setScope(Scope.LOCAL); |
| region = factory.create("test"); |
| |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| CacheObserver old = CacheObserverHolder.setInstance( |
| |
| new CacheObserverAdapter() { |
| |
| @Override |
| public void goingToFlush() { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| synchronized (OplogJUnitTest.this) { |
| OplogJUnitTest.this.notify(); |
| } |
| } |
| }); |
| region.put("key1", new byte[100]); |
| region.put("key2", new byte[100]); |
| region.put("key3", new byte[100]); |
| region.put("key4", new byte[100]); |
| region.put("key5", new byte[100]); |
| Thread.sleep(1000); |
| assertTrue(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| |
| |
| ((LocalRegion) region).getDiskRegion().forceRolling(); |
| synchronized (this) { |
| if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) { |
| OplogJUnitTest.this.wait(10000); |
| } |
| } |
| assertFalse(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| CacheObserverHolder.setInstance(old); |
| closeDown(); |
| } |
| |
| /** |
| * Tests if the byte buffer pool in asynch mode tries to contain the pool size |
| * |
| */ |
| @Test |
| public void testByteBufferPoolContainment() throws Exception { |
| |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(false); |
| diskProps.setBytesThreshold(10); // this is now item count |
| diskProps.setTimeInterval(0); |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| |
| CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| |
| @Override |
| public void goingToFlush() { // Delay flushing |
| |
| assertEquals(10, region.size()); |
| for (int i = 10; i < 20; ++i) { |
| region.put("" + i, val); |
| } |
| synchronized (OplogJUnitTest.this) { |
| OplogJUnitTest.this.notify(); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| |
| } |
| }); |
| for (int i = 0; i < 10; ++i) { |
| region.put("" + i, val); |
| } |
| synchronized (OplogJUnitTest.this) { |
| if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) { |
| OplogJUnitTest.this.wait(9000); |
| assertEquals(false, LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER); |
| } |
| } |
| } |
| |
| /** |
| * tests async stats are correctly updated |
| */ |
| @Test |
| public void testAsyncStats() throws InterruptedException { |
| diskProps.setBytesThreshold(101); |
| diskProps.setTimeInterval(1000000); |
| region = DiskRegionHelperFactory.getAsyncOverFlowAndPersistRegion(cache, diskProps); |
| final DiskStoreStats dss = ((LocalRegion) region).getDiskRegion().getDiskStore().getStats(); |
| |
| assertEquals(0, dss.getQueueSize()); |
| put100Int(); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(100, dss.getQueueSize())); |
| |
| assertEquals(0, dss.getFlushes()); |
| |
| DiskRegion diskRegion = ((LocalRegion) region).getDiskRegion(); |
| diskRegion.getDiskStore().flush(); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, dss.getQueueSize())); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(100, dss.getFlushes())); |
| put100Int(); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(100, dss.getQueueSize())); |
| diskRegion.getDiskStore().flush(); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, dss.getQueueSize())); |
| await() |
| .timeout(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(200, dss.getFlushes())); |
| closeDown(); |
| } |
| |
| /** |
| * Tests delayed creation of DiskID in overflow only mode |
| * |
| */ |
| @Test |
| public void testDelayedDiskIdCreationInOverflowOnlyMode() { |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(1); |
| region = DiskRegionHelperFactory.getAsyncOverFlowOnlyRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| region.put("1", val); |
| DiskEntry entry = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| assertTrue(entry instanceof AbstractDiskLRURegionEntry); |
| assertNull(entry.getDiskId()); |
| region.put("2", val); |
| assertNotNull(entry.getDiskId()); |
| entry = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| assertTrue(entry instanceof AbstractDiskLRURegionEntry); |
| assertNull(entry.getDiskId()); |
| } |
| |
| /** |
| * Tests immediate creation of DiskID in overflow With Persistence mode |
| * |
| */ |
| @Test |
| public void testImmediateDiskIdCreationInOverflowWithPersistMode() { |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(false); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(1); |
| region = DiskRegionHelperFactory.getAsyncOverFlowAndPersistRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| region.put("1", val); |
| DiskEntry entry = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| assertTrue(entry instanceof AbstractDiskLRURegionEntry); |
| assertNotNull(entry.getDiskId()); |
| region.put("2", val); |
| assertNotNull(entry.getDiskId()); |
| entry = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| assertTrue(entry instanceof AbstractDiskLRURegionEntry); |
| assertNotNull(entry.getDiskId()); |
| } |
| |
| /** |
| * An entry which is evicted to disk will have the flag already written to disk, appropriately set |
| * |
| */ |
| @Test |
| public void testEntryAlreadyWrittenIsCorrectlyUnmarkedForOverflowOnly() throws Exception { |
| diskProps.setPersistBackup(false); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(1); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| region.put("1", val); |
| region.put("2", val); |
| // "1" should now be on disk |
| region.get("1"); |
| // "2" should now be on disk |
| DiskEntry entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| DiskId did1 = entry1.getDiskId(); |
| DiskId.isInstanceofOverflowIntOplogOffsetDiskId(did1); |
| assertTrue(!did1.needsToBeWritten()); |
| region.put("1", "3"); |
| assertTrue(did1.needsToBeWritten()); |
| region.put("2", val); |
| DiskEntry entry2 = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| DiskId did2 = entry2.getDiskId(); |
| assertTrue(!did2.needsToBeWritten() || !did1.needsToBeWritten()); |
| tearDown(); |
| setUp(); |
| diskProps.setPersistBackup(false); |
| diskProps.setRolling(false); |
| long opsize = Integer.MAX_VALUE; |
| opsize += 100L; |
| diskProps.setMaxOplogSize(opsize); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(1); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| region.put("1", val); |
| region.put("2", val); |
| region.get("1"); |
| entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| did1 = entry1.getDiskId(); |
| DiskId.isInstanceofOverflowOnlyWithLongOffset(did1); |
| assertTrue(!did1.needsToBeWritten()); |
| region.put("1", "3"); |
| assertTrue(did1.needsToBeWritten()); |
| region.put("2", "3"); |
| did2 = entry2.getDiskId(); |
| assertTrue(!did2.needsToBeWritten() || !did1.needsToBeWritten()); |
| } |
| |
| |
| /** |
| * An persistent or overflow with persistence entry which is evicted to disk, will have the flag |
| * already written to disk, appropriately set |
| * |
| */ |
| @Test |
| public void testEntryAlreadyWrittenIsCorrectlyUnmarkedForOverflowWithPersistence() { |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(1); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| region.put("1", val); |
| DiskEntry entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| DiskId did1 = entry1.getDiskId(); |
| DiskId.isInstanceofPersistIntOplogOffsetDiskId(did1); |
| assertTrue(!did1.needsToBeWritten()); |
| region.put("2", val); |
| assertTrue(!did1.needsToBeWritten()); |
| } |
| |
| /** |
| * Tests the various DiskEntry.Helper APIs for correctness as there is now delayed creation of |
| * DiskId and accessing OplogkeyId will throw UnsupportedException |
| */ |
| @Test |
| public void testHelperAPIsForOverflowOnlyRegion() { |
| diskProps.setPersistBackup(false); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(true); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(2); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| region.put("1", val); |
| region.put("2", val); |
| region.put("3", val); |
| DiskEntry entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| DiskEntry entry2 = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| DiskEntry entry3 = (DiskEntry) ((LocalRegion) region).basicGetEntry("3"); |
| assertNull(entry2.getDiskId()); |
| assertNull(entry3.getDiskId()); |
| assertNotNull(entry1.getDiskId()); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry3, dr)); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry2, dr)); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry1, dr)); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry3, dr)); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry2, dr)); |
| |
| assertNull(entry2.getDiskId()); |
| assertNull(entry3.getDiskId()); |
| assertNotNull(entry1.getDiskId()); |
| assertNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry3, dr, (LocalRegion) region)); |
| assertNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry2, dr, (LocalRegion) region)); |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry1, dr, (LocalRegion) region)); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry3, dr)); |
| assertNull(DiskEntry.Helper.getValueOnDisk(entry2, dr)); |
| } |
| |
| /** |
| * Tests the various DiskEntry.Helper APIs for correctness as there is now delayed creation of |
| * DiskId and accessing OplogkeyId will throw UnsupportedException |
| */ |
| @Test |
| public void testHelperAPIsForOverflowWithPersistenceRegion() { |
| helperAPIsForPersistenceWithOrWithoutOverflowRegion(true /* should overflow */); |
| } |
| |
| /** |
| * Tests the various DiskEntry.Helper APIs for correctness as there is now delayed creation of |
| * DiskId and accessing OplogkeyId will throw UnsupportedException |
| */ |
| @Test |
| public void testHelperAPIsForPersistenceRegion() { |
| helperAPIsForPersistenceWithOrWithoutOverflowRegion(false /* should overflow */); |
| } |
| |
| /** |
| * Tests the various DiskEntry.Helper APIs for correctness as there is now delayed creation of |
| * DiskId and accessing OplogkeyId will throw UnsupportedException |
| */ |
| private void helperAPIsForPersistenceWithOrWithoutOverflowRegion(boolean overflow) { |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setMaxOplogSize(1024 * 1024); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(overflow); |
| diskProps.setBytesThreshold(10000); |
| diskProps.setTimeInterval(0); |
| diskProps.setOverFlowCapacity(2); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| final byte[] val = new byte[1000]; |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| region.put("1", val); |
| region.put("2", val); |
| region.put("3", val); |
| DiskEntry entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| DiskEntry entry2 = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| DiskEntry entry3 = (DiskEntry) ((LocalRegion) region).basicGetEntry("3"); |
| assertNotNull(entry2.getDiskId()); |
| assertNotNull(entry3.getDiskId()); |
| assertNotNull(entry1.getDiskId()); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry3, dr)); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry2, dr)); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry1, dr)); |
| |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry3, dr, (LocalRegion) region)); |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry2, dr, (LocalRegion) region)); |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry1, dr, (LocalRegion) region)); |
| |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps); |
| dr = ((LocalRegion) region).getDiskRegion(); |
| entry1 = (DiskEntry) ((LocalRegion) region).basicGetEntry("1"); |
| entry2 = (DiskEntry) ((LocalRegion) region).basicGetEntry("2"); |
| entry3 = (DiskEntry) ((LocalRegion) region).basicGetEntry("3"); |
| |
| assertNotNull(entry2.getDiskId()); |
| assertNotNull(entry3.getDiskId()); |
| assertNotNull(entry1.getDiskId()); |
| |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry3, dr)); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry2, dr)); |
| assertNotNull(DiskEntry.Helper.getValueOnDisk(entry1, dr)); |
| |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry3, dr, (LocalRegion) region)); |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry2, dr, (LocalRegion) region)); |
| assertNotNull(DiskEntry.Helper.getValueOnDiskOrBuffer(entry1, dr, (LocalRegion) region)); |
| } |
| |
| /** |
| * Tests the condition when a 'put' is in progress and concurrent 'clear' and 'put'(on the same |
| * key) occur. Thus if after Htree ref was set (in 'put'), the region got cleared (and same key |
| * re-'put'), the entry will get recorded in the new Oplog without a corresponding create ( |
| * because the Oplogs containing create have already been deleted due to the clear operation). |
| * This put should not proceed. Also, Region creation after closing should not give an exception. |
| */ |
| @Test |
| public void testPutClearPut() throws Exception { |
| try { |
| // Create a persist only region with rolling true |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setMaxOplogSize(1024); |
| diskProps.setSynchronous(true); |
| this.proceed = false; |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| final Thread clearOp = new Thread(() -> { |
| try { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| region.clear(); |
| region.put("key1", "value3"); |
| } catch (Exception e) { |
| testFailed = true; |
| failureCause = "Encountered Exception=" + e; |
| } |
| }); |
| region.getAttributesMutator().setCacheWriter(new CacheWriterAdapter() { |
| @Override |
| public void beforeUpdate(EntryEvent event) throws CacheWriterException { |
| clearOp.start(); |
| } |
| }); |
| try { |
| ThreadUtils.join(clearOp, 30 * 1000); |
| } catch (Exception e) { |
| testFailed = true; |
| failureCause = "Encountered Exception=" + e; |
| e.printStackTrace(); |
| } |
| region.create("key1", "value1"); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| region.put("key1", "value2"); |
| if (!testFailed) { |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| } else { |
| fail(failureCause); |
| } |
| } finally { |
| testFailed = false; |
| proceed = false; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| /** |
| * Tests the condition when a 'put' on an alreay created entry and concurrent 'clear' are |
| * happening. Thus if after HTree ref was set (in 'put'), the region got cleared (and same key |
| * re-'put'), the entry will actually become a create in the VM The new Oplog should record it as |
| * a create even though the Htree ref in ThreadLocal will not match with the current Htree Ref. |
| * But the operation is valid & should get recorded in Oplog |
| * |
| */ |
| @Test |
| public void testPutClearCreate() throws Exception { |
| failure = false; |
| try { |
| // Create a persist only region with rolling true |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setMaxOplogSize(1024); |
| diskProps.setSynchronous(true); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| region.create("key1", "value1"); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| @Override |
| public void afterSettingDiskRef() { |
| Thread clearTh = new Thread(() -> region.clear()); |
| clearTh.start(); |
| try { |
| ThreadUtils.join(clearTh, 120 * 1000); |
| failure = clearTh.isAlive(); |
| failureCause = "Clear Thread still running !"; |
| } catch (Exception e) { |
| failure = true; |
| failureCause = e.toString(); |
| } |
| } |
| }); |
| region.put("key1", "value2"); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| assertFalse(failureCause, failure); |
| assertEquals(1, region.size()); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| assertEquals(1, region.size()); |
| assertEquals("value2", region.get("key1")); |
| } finally { |
| testFailed = false; |
| proceed = false; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(new CacheObserverAdapter()); |
| failure = false; |
| } |
| } |
| |
| /** |
| * Tests if 'destroy' transaction is working correctly for sync-overflow-only disk region entry |
| */ |
| @Test |
| public void testOverFlowOnlySyncDestroyTx() { |
| diskProps.setMaxOplogSize(20480); |
| diskProps.setOverFlowCapacity(1); |
| diskProps.setDiskDirs(dirs); |
| region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps); |
| assertNotNull(region); |
| region.put("key", "createValue"); |
| region.put("key1", "createValue1"); |
| cache.getCacheTransactionManager().begin(); |
| region.destroy("key"); |
| cache.getCacheTransactionManager().commit(); |
| assertNull("The deleted entry should have been null", |
| ((LocalRegion) region).entries.getEntry("key")); |
| } |
| |
| /** |
| * Test to force a recovery to follow the path of switchOutFilesForRecovery and ensuring that |
| * IOExceptions do not come as a result. This is also a bug test for bug 37682 |
| */ |
| @Test |
| public void testSwitchFilesForRecovery() throws Exception { |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, null, Scope.LOCAL); |
| put100Int(); |
| ((LocalRegion) region).getDiskRegion().forceRolling(); |
| Thread.sleep(2000); |
| put100Int(); |
| int sizeOfRegion = region.size(); |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, null, Scope.LOCAL); |
| |
| if (sizeOfRegion != region.size()) { |
| fail(" Expected region size to be " + sizeOfRegion + " after recovery but it is " |
| + region.size()); |
| } |
| } |
| |
| /** |
| * tests directory stats are correctly updated in case of single directory (for bug 37531) |
| */ |
| @Test |
| public void testPersist1DirStats() { |
| final AtomicBoolean freezeRoller = new AtomicBoolean(); |
| CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() { |
| private volatile boolean didBeforeCall = false; |
| |
| @Override |
| public void beforeGoingToCompact() { |
| this.didBeforeCall = true; |
| synchronized (freezeRoller) { |
| if (!assertDone) { |
| try { |
| // Here, we are not allowing the Roller thread to roll the old oplog into htree |
| while (!freezeRoller.get()) { |
| freezeRoller.wait(); |
| } |
| freezeRoller.set(false); |
| } catch (InterruptedException e) { |
| fail("interrupted"); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void afterHavingCompacted() { |
| if (this.didBeforeCall) { |
| this.didBeforeCall = false; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| checkDiskStats(); |
| } |
| } |
| }); |
| try { |
| final int MAX_OPLOG_SIZE = 500; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {4000}); |
| final byte[] val = new byte[200]; |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| region.put("key1", val); |
| checkDiskStats(); |
| region.put("key2", val); |
| checkDiskStats(); |
| // This put will cause a switch as max-oplog size (500) will be exceeded (600) |
| region.put("key3", val); |
| synchronized (freezeRoller) { |
| checkDiskStats(); |
| assertDone = true; |
| freezeRoller.set(true); |
| freezeRoller.notifyAll(); |
| } |
| |
| region.close(); |
| closeDown(); |
| // Stop rolling to get accurate estimates: |
| diskProps.setRolling(false); |
| |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| // On recreating the region after closing, old Oplog file gets rolled into htree |
| // "Disk space usage zero when region recreated" |
| checkDiskStats(); |
| region.put("key4", val); |
| checkDiskStats(); |
| region.put("key5", val); |
| checkDiskStats(); |
| assertDone = false; |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| region.put("key6", val); |
| // again we expect a switch in oplog here |
| synchronized (freezeRoller) { |
| checkDiskStats(); |
| assertDone = true; |
| freezeRoller.set(true); |
| freezeRoller.notifyAll(); |
| } |
| region.close(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Test failed due to exception" + e); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(old); |
| synchronized (freezeRoller) { |
| assertDone = true; |
| freezeRoller.set(true); |
| freezeRoller.notifyAll(); |
| } |
| } |
| } |
| |
| /** |
| * Tests reduction in size of disk stats when the oplog is rolled. |
| */ |
| @Test |
| public void testStatsSizeReductionOnRolling() throws Exception { |
| final int MAX_OPLOG_SIZE = 500 * 2; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setCompactionThreshold(100); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {4000}); |
| final byte[] val = new byte[333]; |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| final DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| |
| // calculate sizes |
| final int extra_byte_num_per_entry = |
| InternalDataSerializer.calculateBytesForTSandDSID(getDSID((LocalRegion) region)); |
| final int key3_size = |
| CompactOfflineDiskStoreJUnitTest.getSize4Create(extra_byte_num_per_entry, "key3", val); |
| final int tombstone_key1 = |
| CompactOfflineDiskStoreJUnitTest.getSize4TombstoneWithKey(extra_byte_num_per_entry, "key1"); |
| final int tombstone_key2 = |
| CompactOfflineDiskStoreJUnitTest.getSize4TombstoneWithKey(extra_byte_num_per_entry, "key2"); |
| |
| CountDownLatch putsCompleted = new CountDownLatch(1); |
| // TODO: move static methods from CompactOfflineDiskStoreJUnitTest to shared util class |
| StatSizeTestCacheObserverAdapter testObserver = new StatSizeTestCacheObserverAdapter(dr, |
| key3_size, tombstone_key1, tombstone_key2, putsCompleted); |
| CacheObserver old = CacheObserverHolder.setInstance(testObserver); |
| |
| try { |
| |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| cache.getLogger().info("putting key1"); |
| region.put("key1", val); |
| checkDiskStats(); |
| cache.getLogger().info("putting key2"); |
| region.put("key2", val); |
| checkDiskStats(); |
| |
| cache.getLogger().info("removing key1"); |
| region.remove("key1"); |
| cache.getLogger().info("removing key2"); |
| region.remove("key2"); |
| |
| // This put will cause a switch as max-oplog size (900) will be exceeded (999) |
| testObserver.setSwitchExpected(); |
| cache.getLogger().info("putting key3"); |
| region.put("key3", val); |
| putsCompleted.countDown(); |
| cache.getLogger().info("waiting for compaction"); |
| await().until(() -> testObserver.hasCompacted()); |
| assertFalse(testObserver.exceptionOccured()); |
| |
| region.close(); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| CacheObserverHolder.setInstance(old); |
| } |
| } |
| |
| @Test |
| public void testUnPreblowOnRegionCreate() throws Exception { |
| final int MAX_OPLOG_SIZE = 20000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {40000}); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| try { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| for (int i = 0; i < 10; ++i) { |
| region.put("key-" + i, "value-"); |
| } |
| |
| assertEquals(18000, getOplogFileSizeSum(dirs[0], ".crf")); |
| assertEquals(2000, getOplogFileSizeSum(dirs[0], ".drf")); |
| |
| // make a copy of inflated crf. use this to replace compacted crf to |
| // simulate incomplete diskStore close |
| File[] files = dirs[0].listFiles(); |
| for (File file : files) { |
| if (file.getName().endsWith(".crf") || file.getName().endsWith(".drf")) { |
| File inflated = new File(file.getAbsolutePath() + "_inflated"); |
| FileUtils.copyFile(file, inflated); |
| } |
| } |
| |
| cache.close(); |
| assertTrue(500 > getOplogFileSizeSum(dirs[0], ".crf")); |
| assertTrue(100 > getOplogFileSizeSum(dirs[0], ".drf")); |
| |
| // replace compacted crf with inflated crf and remove krf |
| files = dirs[0].listFiles(); |
| for (File file : files) { |
| String name = file.getName(); |
| if (name.endsWith(".krf") || name.endsWith(".crf") || name.endsWith(".drf")) { |
| file.delete(); |
| } |
| } |
| for (File file : files) { |
| String name = file.getName(); |
| if (name.endsWith("_inflated")) { |
| assertTrue(file.renameTo(new File(file.getAbsolutePath().replace("_inflated", "")))); |
| } |
| } |
| assertEquals(18000, getOplogFileSizeSum(dirs[0], ".crf")); |
| assertEquals(2000, getOplogFileSizeSum(dirs[0], ".drf")); |
| |
| createCache(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| for (int i = 10; i < 20; ++i) { |
| region.put("key-" + i, "value-"); |
| } |
| |
| int sizeCrf = getOplogFileSizeSum(dirs[0], ".crf"); |
| assertTrue("crf too big:" + sizeCrf, sizeCrf < 18000 + 500); |
| assertTrue("crf too small:" + sizeCrf, sizeCrf > 18000); |
| int sizeDrf = getOplogFileSizeSum(dirs[0], ".drf"); |
| assertTrue("drf too big:" + sizeDrf, sizeDrf < 2000 + 100); |
| assertTrue("drf too small:" + sizeDrf, sizeDrf > 2000); |
| |
| // test that region recovery does not cause unpreblow |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| assertEquals(sizeCrf, getOplogFileSizeSum(dirs[0], ".crf")); |
| assertEquals(sizeDrf, getOplogFileSizeSum(dirs[0], ".drf")); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| private int getOplogFileSizeSum(File dir, String type) { |
| int sum = 0; |
| File[] files = dir.listFiles(); |
| for (File file : files) { |
| String name = file.getName(); |
| if (name.endsWith(type)) { |
| sum += file.length(); |
| } |
| } |
| return sum; |
| } |
| |
| @Test |
| public void testMagicSeqPresence() throws Exception { |
| final int MAX_OPLOG_SIZE = 200; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(true); |
| diskProps.setSynchronous(true); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {4000}); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| // 3 types of oplog files will be verified |
| verifyOplogHeader(dirs[0], ".if", ".crf", ".drf"); |
| |
| try { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| IntStream.range(0, 20).forEach(i -> region.put("key-" + i, "value-" + i)); |
| // krf is created, so 4 types of oplog files will be verified |
| verifyOplogHeader(dirs[0], ".if", ".crf", ".drf", ".krf"); |
| |
| region.close(); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| |
| verifyOplogHeader(dirs[0], ".if", ".crf", ".drf", ".krf"); |
| region.close(); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| private void verifyOplogHeader(File dir, String... oplogTypes) throws IOException { |
| |
| await().until(() -> { |
| List<String> types = new ArrayList<>(Arrays.asList(oplogTypes)); |
| Arrays.stream(dir.listFiles()).map(File::getName).map(f -> f.substring(f.indexOf("."))) |
| .forEach(types::remove); |
| return types.isEmpty(); |
| }); |
| |
| File[] files = dir.listFiles(); |
| HashSet<String> verified = new HashSet<>(); |
| for (File file : files) { |
| String name = file.getName(); |
| byte[] expect = new byte[Oplog.OPLOG_MAGIC_SEQ_REC_SIZE]; |
| if (name.endsWith(".crf")) { |
| expect[0] = Oplog.OPLOG_MAGIC_SEQ_ID; |
| System.arraycopy(OPLOG_TYPE.CRF.getBytes(), 0, expect, 1, OPLOG_TYPE.getLen()); |
| verified.add(".crf"); |
| } else if (name.endsWith(".drf")) { |
| expect[0] = Oplog.OPLOG_MAGIC_SEQ_ID; |
| System.arraycopy(OPLOG_TYPE.DRF.getBytes(), 0, expect, 1, OPLOG_TYPE.getLen()); |
| verified.add(".drf"); |
| } else if (name.endsWith(".krf")) { |
| expect[0] = Oplog.OPLOG_MAGIC_SEQ_ID; |
| System.arraycopy(OPLOG_TYPE.KRF.getBytes(), 0, expect, 1, OPLOG_TYPE.getLen()); |
| verified.add(".krf"); |
| } else if (name.endsWith(".if")) { |
| expect[0] = DiskInitFile.OPLOG_MAGIC_SEQ_ID; |
| System.arraycopy(OPLOG_TYPE.IF.getBytes(), 0, expect, 1, OPLOG_TYPE.getLen()); |
| verified.add(".if"); |
| } else { |
| System.out.println("Ignored: " + file); |
| continue; |
| } |
| expect[expect.length - 1] = 21; // EndOfRecord |
| |
| byte[] buf = new byte[Oplog.OPLOG_MAGIC_SEQ_REC_SIZE]; |
| |
| FileInputStream fis = new FileInputStream(file); |
| int count = fis.read(buf, 0, 8); |
| fis.close(); |
| |
| System.out.println("Verifying: " + file); |
| assertEquals("expected a read to return 8 but it returned " + count + " for file " + file, 8, |
| count); |
| assertTrue(Arrays.equals(expect, buf)); |
| } |
| |
| assertEquals(oplogTypes.length, verified.size()); |
| } |
| |
| /** |
| * Tests if without rolling the region size before close is same as after recreation |
| */ |
| @Test |
| public void testSizeStatsAfterRecreationInAsynchMode() throws Exception { |
| final int MAX_OPLOG_SIZE = 1000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(false); |
| diskProps.setBytesThreshold(800); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0], dirs[1]}, new int[] {4000, 4000}); |
| final byte[] val = new byte[25]; |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| |
| try { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| for (int i = 0; i < 42; ++i) { |
| region.put("key" + i, val); |
| } |
| // need to wait for writes to happen before getting size |
| dr.flushForTesting(); |
| long size1 = 0; |
| for (DirectoryHolder dh : dr.getDirectories()) { |
| size1 += dh.getDirStatsDiskSpaceUsage(); |
| } |
| System.out.println("Size before close = " + size1); |
| region.close(); |
| diskProps.setSynchronous(true); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| dr = ((LocalRegion) region).getDiskRegion(); |
| long size2 = 0; |
| for (DirectoryHolder dh : dr.getDirectories()) { |
| size2 += dh.getDirStatsDiskSpaceUsage(); |
| } |
| System.out.println("Size after recreation= " + size2); |
| assertEquals(size1, size2); |
| region.close(); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| @Test |
| public void testAsynchModeStatsBehaviour() throws Exception { |
| final int MAX_OPLOG_SIZE = 1000; |
| diskProps.setMaxOplogSize(MAX_OPLOG_SIZE); |
| diskProps.setPersistBackup(true); |
| diskProps.setRolling(false); |
| diskProps.setSynchronous(false); |
| diskProps.setBytesThreshold(800); |
| diskProps.setTimeInterval(Long.MAX_VALUE); |
| diskProps.setOverflow(false); |
| diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {4000}); |
| final byte[] val = new byte[25]; |
| region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps); |
| DiskRegion dr = ((LocalRegion) region).getDiskRegion(); |
| |
| try { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true; |
| for (int i = 0; i < 4; ++i) { |
| region.put("key" + i, val); |
| } |
| // This test now has a race condition in it since |
| // async puts no longer increment disk space. |
| // It is not until a everything is flushed that we will know the disk size. |
| dr.flushForTesting(); |
| |
| checkDiskStats(); |
| long size1 = 0; |
| for (DirectoryHolder dh : dr.getDirectories()) { |
| size1 += dh.getDirStatsDiskSpaceUsage(); |
| } |
| System.out.println("Size before close = " + size1); |
| region.close(); |
| diskProps.setSynchronous(true); |
| region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL); |
| dr = ((LocalRegion) region).getDiskRegion(); |
| long size2 = 0; |
| for (DirectoryHolder dh : dr.getDirectories()) { |
| size2 += dh.getDirStatsDiskSpaceUsage(); |
| } |
| System.out.println("Size after recreation= " + size2); |
| assertEquals(size1, size2); |
| region.close(); |
| } finally { |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| private long diskSpaceUsageStats() { |
| return ((LocalRegion) region).getDiskRegion().getInfoFileDir().getDirStatsDiskSpaceUsage(); |
| } |
| |
| private long calculatedDiskSpaceUsageStats() { |
| return oplogSize(); |
| } |
| |
| private void checkDiskStats() { |
| long actualDiskStats = diskSpaceUsageStats(); |
| long computedDiskStats = calculatedDiskSpaceUsageStats(); |
| int tries = 0; |
| while (actualDiskStats != computedDiskStats && tries++ <= 100) { |
| // race conditions exist in which the stats change |
| try { |
| Thread.sleep(100); |
| } catch (InterruptedException ignore) { |
| } |
| actualDiskStats = diskSpaceUsageStats(); |
| computedDiskStats = calculatedDiskSpaceUsageStats(); |
| } |
| assertEquals(computedDiskStats, actualDiskStats); |
| } |
| |
| private long oplogSize() { |
| long size = ((LocalRegion) region).getDiskRegion().getDiskStore().undeletedOplogSize.get(); |
| Oplog[] opArray = |
| ((LocalRegion) region).getDiskRegion().getDiskStore().getPersistentOplogs().getAllOplogs(); |
| if (opArray != null) { |
| for (Oplog log : opArray) { |
| size += log.getOplogSize(); |
| } |
| } |
| return size; |
| } |
| |
| private int getDSID(LocalRegion lr) { |
| return lr.getDistributionManager().getDistributedSystemId(); |
| } |
| |
| private class StatSizeTestCacheObserverAdapter extends CacheObserverAdapter { |
| private final AtomicBoolean switchExpected = new AtomicBoolean(false); |
| private final DiskRegion dr; |
| private final AtomicBoolean hasCompacted = new AtomicBoolean(false); |
| private final int key3Size; |
| private final int tombstoneKey1; |
| private final int tombstoneKey2; |
| private final AtomicBoolean exceptionOccurred = new AtomicBoolean(true); |
| private volatile long spaceUsageBefore = -1; |
| private DirectoryHolder dh; |
| private final AtomicLong oplogsSize = new AtomicLong(); |
| private final CountDownLatch putsCompleted; |
| |
| StatSizeTestCacheObserverAdapter(DiskRegion dr, int key3Size, int tombstoneKey1, |
| int tombstoneKey2, CountDownLatch putsCompleted) { |
| this.dr = dr; |
| this.key3Size = key3Size; |
| this.tombstoneKey1 = tombstoneKey1; |
| this.tombstoneKey2 = tombstoneKey2; |
| this.putsCompleted = putsCompleted; |
| } |
| |
| @Override |
| public void beforeSwitchingOplog() { |
| cache.getLogger().info("beforeSwitchingOplog"); |
| if (!switchExpected.get()) { |
| fail("unexpected oplog switch"); |
| } |
| if (spaceUsageBefore == -1) { |
| // only want to call this once; before the 1st oplog destroy |
| this.dh = dr.getNextDir(); |
| this.spaceUsageBefore = this.dh.getDirStatsDiskSpaceUsage(); |
| } |
| } |
| |
| @Override |
| public void beforeDeletingCompactedOplog(Oplog oplog) { |
| cache.getLogger().info("beforeDeletingCompactedOplog"); |
| oplogsSize.addAndGet(oplog.getOplogSize()); |
| } |
| |
| @Override |
| public void afterHavingCompacted() { |
| try { |
| putsCompleted.await(10, TimeUnit.SECONDS); |
| } catch (InterruptedException e) { |
| exceptionOccurred.set(true); |
| throw new RuntimeException(e); |
| } |
| cache.getLogger().info("afterHavingCompacted"); |
| if (spaceUsageBefore > -1) { |
| hasCompacted.set(true); |
| long after = this.dh.getDirStatsDiskSpaceUsage(); |
| // after compaction, in _2.crf, key3 is an create-entry, |
| // key1 and key2 are tombstones. |
| // _2.drf contained a rvvgc with drMap.size()==1 |
| int expectedDrfSize = Oplog.OPLOG_DISK_STORE_REC_SIZE + Oplog.OPLOG_MAGIC_SEQ_REC_SIZE |
| + Oplog.OPLOG_GEMFIRE_VERSION_REC_SIZE |
| + CompactOfflineDiskStoreJUnitTest.getRVVSize(1, new int[] {0}, true); |
| int expectedCrfSize = Oplog.OPLOG_DISK_STORE_REC_SIZE + Oplog.OPLOG_MAGIC_SEQ_REC_SIZE |
| + Oplog.OPLOG_GEMFIRE_VERSION_REC_SIZE |
| + CompactOfflineDiskStoreJUnitTest.getRVVSize(1, new int[] {1}, false) |
| + Oplog.OPLOG_NEW_ENTRY_BASE_REC_SIZE + key3Size + tombstoneKey1 + tombstoneKey2; |
| int oplog2Size = expectedDrfSize + expectedCrfSize; |
| if (after != oplog2Size) { |
| cache.getLogger().info("test failed before=" + spaceUsageBefore + " after=" + after |
| + " expected=" + oplog2Size); |
| exceptionOccurred.set(true); |
| } else { |
| exceptionOccurred.set(false); |
| } |
| LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false; |
| } |
| } |
| |
| boolean hasCompacted() { |
| return hasCompacted.get(); |
| } |
| |
| boolean exceptionOccured() { |
| return exceptionOccurred.get(); |
| } |
| |
| void setSwitchExpected() { |
| switchExpected.set(true); |
| } |
| } |
| } |
