You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2017/06/11 20:03:51 UTC
[25/47] ignite git commit: IGNITE-5267 - Moved ignite-ps module to
ignite-core
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/MemoryPolicyInitializationTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/MemoryPolicyInitializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/MemoryPolicyInitializationTest.java
deleted file mode 100644
index ec798ee..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/MemoryPolicyInitializationTest.java
+++ /dev/null
@@ -1,313 +0,0 @@
-/*
- * 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.ignite.internal.processors.cache.database;
-
-import java.util.Collection;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.configuration.MemoryConfiguration;
-import org.apache.ignite.configuration.MemoryPolicyConfiguration;
-import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-
-import static org.apache.ignite.configuration.MemoryConfiguration.DFLT_MEM_PLC_DEFAULT_NAME;
-
-/**
- *
- */
-public class MemoryPolicyInitializationTest extends GridCommonAbstractTest {
- /** */
- private static final String CUSTOM_NON_DEFAULT_MEM_PLC_NAME = "custom_mem_plc";
-
- /** */
- private static final long USER_CUSTOM_MEM_PLC_SIZE = 10 * 1024 * 1024;
-
- /** */
- private static final long USER_DEFAULT_MEM_PLC_SIZE = 99 * 1024 * 1024;
-
- /** */
- private MemoryConfiguration memCfg;
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
- cfg.setMemoryConfiguration(memCfg);
-
- return cfg;
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- memCfg = null;
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids();
- }
-
- /**
- * Verifies that expected memory policies are allocated when used doesn't provide any MemoryPolicyConfiguration.
- */
- public void testNoConfigProvided() throws Exception {
- memCfg = null;
-
- IgniteEx ignite = startGrid(0);
-
- Collection<MemoryPolicy> allMemPlcs = ignite.context().cache().context().database().memoryPolicies();
-
- assertTrue(allMemPlcs.size() == 2);
-
- verifyDefaultAndSystemMemoryPolicies(allMemPlcs);
- }
-
- /**
- * Verifies that expected memory policies are allocated when used provides MemoryPolicyConfiguration
- * with non-default custom MemoryPolicy.
- */
- public void testCustomConfigNoDefault() throws Exception {
- prepareCustomNoDefaultConfig();
-
- IgniteEx ignite = startGrid(0);
-
- Collection<MemoryPolicy> allMemPlcs = ignite.context().cache().context().database().memoryPolicies();
-
- assertTrue(allMemPlcs.size() == 3);
-
- verifyDefaultAndSystemMemoryPolicies(allMemPlcs);
-
- assertTrue("Custom non-default memory policy is not presented",
- isMemoryPolicyPresented(allMemPlcs, CUSTOM_NON_DEFAULT_MEM_PLC_NAME));
- }
-
- /**
- * User is allowed to configure memory policy with 'default' name,
- * in that case Ignite instance will use this user-defined memory policy as a default one.
- */
- public void testCustomConfigOverridesDefault() throws Exception {
- prepareCustomConfigWithOverridingDefault();
-
- IgniteEx ignite = startGrid(0);
-
- IgniteCacheDatabaseSharedManager dbMgr = ignite.context().cache().context().database();
-
- Collection<MemoryPolicy> allMemPlcs = dbMgr.memoryPolicies();
-
- assertTrue(allMemPlcs.size() == 2);
-
- verifyDefaultAndSystemMemoryPolicies(allMemPlcs);
-
- MemoryPolicy dfltMemPlc = U.field(dbMgr, "dfltMemPlc");
-
- assertTrue(dfltMemPlc.config().getMaxSize() == USER_DEFAULT_MEM_PLC_SIZE);
- }
-
- /**
- * User is allowed to define fully custom memory policy and make it default by setting its name to memory config.
- *
- * At the same time user still can create a memory policy with name 'default'
- * which although won't be used as default.
- */
- public void testCustomConfigOverridesDefaultNameAndDeclaresDefault() throws Exception {
- prepareCustomConfigWithOverriddenDefaultName();
-
- IgniteEx ignite = startGrid(0);
-
- IgniteCacheDatabaseSharedManager dbMgr = ignite.context().cache().context().database();
-
- Collection<MemoryPolicy> allMemPlcs = dbMgr.memoryPolicies();
-
- assertTrue(allMemPlcs.size() == 3);
-
- verifyDefaultAndSystemMemoryPolicies(allMemPlcs);
-
- MemoryPolicy dfltMemPlc = U.field(dbMgr, "dfltMemPlc");
-
- assertTrue(dfltMemPlc.config().getMaxSize() == USER_CUSTOM_MEM_PLC_SIZE);
- }
-
- /**
- * Test for verification that caches with not specified memory policy name,
- * with specified default memory policy name and specified custom memory policy name
- * all started with correct memory policy.
- */
- public void testCachesOnOverriddenMemoryPolicy() throws Exception {
- prepareCustomConfigWithOverridingDefaultAndCustom();
-
- IgniteEx ignite = startGrid(0);
-
- CacheConfiguration cache1Cfg = new CacheConfiguration()
- .setName("cache1");
-
- IgniteCache cache1 = ignite.createCache(cache1Cfg);
-
- verifyCacheMemoryPolicy(cache1, DFLT_MEM_PLC_DEFAULT_NAME);
-
- CacheConfiguration cache2Cfg = new CacheConfiguration()
- .setName("cache2")
- .setMemoryPolicyName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- IgniteCache cache2 = ignite.createCache(cache2Cfg);
-
- verifyCacheMemoryPolicy(cache2, CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- CacheConfiguration cache3Cfg = new CacheConfiguration()
- .setName("cache3")
- .setMemoryPolicyName(DFLT_MEM_PLC_DEFAULT_NAME);
-
- IgniteCache cache3 = ignite.createCache(cache3Cfg);
-
- verifyCacheMemoryPolicy(cache3, DFLT_MEM_PLC_DEFAULT_NAME);
- }
-
- /**
- * Test for verification that caches with not specified memory policy name,
- * with specified default memory policy name and specified custom memory policy name
- * all started with correct memory policy.
- */
- public void testCachesOnUserDefinedDefaultMemoryPolicy() throws Exception {
- prepareCustomConfigWithOverriddenDefaultName();
-
- IgniteEx ignite = startGrid(0);
-
- CacheConfiguration cache1Cfg = new CacheConfiguration()
- .setName("cache1");
-
- IgniteCache cache1 = ignite.createCache(cache1Cfg);
-
- verifyCacheMemoryPolicy(cache1, CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- CacheConfiguration cache2Cfg = new CacheConfiguration()
- .setName("cache2")
- .setMemoryPolicyName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- IgniteCache cache2 = ignite.createCache(cache2Cfg);
-
- verifyCacheMemoryPolicy(cache2, CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- CacheConfiguration cache3Cfg = new CacheConfiguration()
- .setName("cache3")
- .setMemoryPolicyName(DFLT_MEM_PLC_DEFAULT_NAME);
-
- IgniteCache cache3 = ignite.createCache(cache3Cfg);
-
- verifyCacheMemoryPolicy(cache3, DFLT_MEM_PLC_DEFAULT_NAME);
- }
-
- /**
- * @param cache Cache.
- * @param plcName Policy name.
- */
- private void verifyCacheMemoryPolicy(IgniteCache cache, String plcName) {
- GridCacheContext ctx = U.field(cache, "ctx");
-
- assertEquals(plcName, ctx.memoryPolicy().config().getName());
- }
-
- /**
- *
- */
- private void prepareCustomConfigWithOverriddenDefaultName() {
- memCfg = new MemoryConfiguration();
-
- memCfg.setDefaultMemoryPolicyName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME);
-
- memCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
- .setName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_CUSTOM_MEM_PLC_SIZE),
-
- new MemoryPolicyConfiguration()
- .setName(DFLT_MEM_PLC_DEFAULT_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_DEFAULT_MEM_PLC_SIZE)
- );
- }
-
-
- /**
- *
- */
- private void prepareCustomConfigWithOverridingDefault() {
- memCfg = new MemoryConfiguration();
-
- memCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
- .setName(DFLT_MEM_PLC_DEFAULT_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_DEFAULT_MEM_PLC_SIZE)
- );
- }
-
- /**
- *
- */
- private void prepareCustomConfigWithOverridingDefaultAndCustom() {
- memCfg = new MemoryConfiguration();
-
- memCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
- .setName(DFLT_MEM_PLC_DEFAULT_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_DEFAULT_MEM_PLC_SIZE),
-
- new MemoryPolicyConfiguration()
- .setName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_CUSTOM_MEM_PLC_SIZE)
- );
- }
-
- /**
- * @param allMemPlcs Collection of all memory policies.
- */
- private void verifyDefaultAndSystemMemoryPolicies(Collection<MemoryPolicy> allMemPlcs) {
- assertTrue("Default memory policy is not presented",
- isMemoryPolicyPresented(allMemPlcs, DFLT_MEM_PLC_DEFAULT_NAME));
-
- assertTrue("System memory policy is not presented",
- isMemoryPolicyPresented(allMemPlcs, IgniteCacheDatabaseSharedManager.SYSTEM_MEMORY_POLICY_NAME));
- }
-
- /**
- *
- */
- private void prepareCustomNoDefaultConfig() {
- memCfg = new MemoryConfiguration();
-
- memCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
- .setName(CUSTOM_NON_DEFAULT_MEM_PLC_NAME)
- .setInitialSize(USER_CUSTOM_MEM_PLC_SIZE)
- .setMaxSize(USER_CUSTOM_MEM_PLC_SIZE)
- );
- }
-
- /**
- * @param memPlcs Collection of memory policies.
- * @param nameToVerify Excepted name of memory policy.
- */
- private boolean isMemoryPolicyPresented(Collection<MemoryPolicy> memPlcs, String nameToVerify) {
- for (MemoryPolicy memPlc : memPlcs) {
- if (nameToVerify.equals(memPlc.config().getName()))
- return true;
- }
-
- return false;
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/tree/io/TrackingPageIOTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/tree/io/TrackingPageIOTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/tree/io/TrackingPageIOTest.java
deleted file mode 100644
index e2767bb..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/database/tree/io/TrackingPageIOTest.java
+++ /dev/null
@@ -1,283 +0,0 @@
-/*
- * 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.ignite.internal.processors.cache.database.tree.io;
-
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.NavigableSet;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ThreadLocalRandom;
-import junit.framework.TestCase;
-import org.apache.ignite.internal.util.GridUnsafe;
-
-/**
- *
- */
-public class TrackingPageIOTest extends TestCase {
- /** Page size. */
- public static final int PAGE_SIZE = 2048;
-
- /** */
- private final TrackingPageIO io = TrackingPageIO.VERSIONS.latest();
-
- /**
- *
- */
- public void testBasics() {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- io.markChanged(buf, 2, 0, -1, PAGE_SIZE);
-
- assertTrue(io.wasChanged(buf, 2, 0, -1, PAGE_SIZE));
-
- assertFalse(io.wasChanged(buf, 1, 0, -1, PAGE_SIZE));
- assertFalse(io.wasChanged(buf, 3, 0, -1, PAGE_SIZE));
- assertFalse(io.wasChanged(buf, 2, 1, 0, PAGE_SIZE));
- }
-
- /**
- *
- */
- public void testMarkingRandomly() {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- int cntOfPageToTrack = io.countOfPageToTrack(PAGE_SIZE);
-
- for (int i = 0; i < 1001; i++)
- checkMarkingRandomly(buf, i, false);
- }
-
- /**
- *
- */
- public void testZeroingRandomly() {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- for (int i = 0; i < 1001; i++)
- checkMarkingRandomly(buf, i, true);
- }
-
- /**
- * @param buf Buffer.
- * @param backupId Backup id.
- */
- private void checkMarkingRandomly(ByteBuffer buf, int backupId, boolean testZeroing) {
- ThreadLocalRandom rand = ThreadLocalRandom.current();
-
- int track = io.countOfPageToTrack(PAGE_SIZE);
-
- long basePageId = io.trackingPageFor(Math.max(rand.nextLong(Integer.MAX_VALUE - track), 0), PAGE_SIZE);
-
- long maxId = testZeroing ? basePageId + rand.nextInt(1, track) : basePageId + track;
-
- assert basePageId >= 0;
-
- PageIO.setPageId(GridUnsafe.bufferAddress(buf), basePageId);
-
- Map<Long, Boolean> map = new HashMap<>();
-
- int cntOfChanged = 0;
-
- try {
- for (long i = basePageId; i < basePageId + track; i++) {
- boolean changed = (i == basePageId || rand.nextDouble() < 0.5) && i < maxId;
-
- map.put(i, changed);
-
- if (changed) {
- io.markChanged(buf, i, backupId, backupId - 1, PAGE_SIZE);
-
- cntOfChanged++;
- }
-
- assertEquals(basePageId, PageIO.getPageId(buf));
- assertEquals(cntOfChanged, io.countOfChangedPage(buf, backupId, PAGE_SIZE));
- }
-
- assertEquals(cntOfChanged, io.countOfChangedPage(buf, backupId, PAGE_SIZE));
-
- for (Map.Entry<Long, Boolean> e : map.entrySet())
- assertEquals(
- e.getValue().booleanValue(),
- io.wasChanged(buf, e.getKey(), backupId, backupId -1, PAGE_SIZE));
- }
- catch (Throwable e) {
- System.out.println("snapshotId = " + backupId + ", basePageId = " + basePageId);
- throw e;
- }
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testFindNextChangedPage() throws Exception {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- for (int i = 0; i < 101; i++)
- checkFindingRandomly(buf, i);
- }
-
- /**
- * @param buf Buffer.
- * @param backupId Backup id.
- */
- private void checkFindingRandomly(ByteBuffer buf, int backupId) {
- ThreadLocalRandom rand = ThreadLocalRandom.current();
-
- int track = io.countOfPageToTrack(PAGE_SIZE);
-
- long basePageId = io.trackingPageFor(Math.max(rand.nextLong(Integer.MAX_VALUE - track), 0), PAGE_SIZE);
-
- long maxId = basePageId + rand.nextInt(1, track);
-
- assert basePageId >= 0;
-
- PageIO.setPageId(GridUnsafe.bufferAddress(buf), basePageId);
-
- try {
- TreeSet<Long> setIdx = new TreeSet<>();
-
- generateMarking(buf, track, basePageId, maxId, setIdx, backupId, backupId -1);
-
- for (long pageId = basePageId; pageId < basePageId + track; pageId++) {
- Long foundNextChangedPage = io.findNextChangedPage(buf, pageId, backupId, backupId - 1, PAGE_SIZE);
-
- if (io.trackingPageFor(pageId, PAGE_SIZE) == pageId)
- assertEquals((Long) pageId, foundNextChangedPage);
-
- else if (setIdx.contains(pageId))
- assertEquals((Long) pageId, foundNextChangedPage);
-
- else {
- NavigableSet<Long> tailSet = setIdx.tailSet(pageId, false);
- Long next = tailSet.isEmpty() ? null : tailSet.first();
-
- assertEquals(next, foundNextChangedPage);
- }
- }
- }
- catch (Throwable e) {
- System.out.println("snapshotId = " + backupId + ", basePageId = " + basePageId);
- throw e;
- }
- }
-
- /**
- *
- */
- public void testMerging() {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- ThreadLocalRandom rand = ThreadLocalRandom.current();
-
- int track = io.countOfPageToTrack(PAGE_SIZE);
-
- long basePageId = io.trackingPageFor(Math.max(rand.nextLong(Integer.MAX_VALUE - track), 0), PAGE_SIZE);
-
- assert basePageId >= 0;
-
- PageIO.setPageId(GridUnsafe.bufferAddress(buf), basePageId);
-
- TreeSet<Long> setIdx = new TreeSet<>();
-
- for (int i = 0; i < 4; i++)
- generateMarking(buf, track, basePageId, basePageId + rand.nextInt(1, track), setIdx, i, -1);
-
- TreeSet<Long> setIdx2 = new TreeSet<>();
-
- generateMarking(buf, track, basePageId, basePageId + rand.nextInt(1, track), setIdx2, 4, -1);
-
- assertEquals(setIdx2.size(), io.countOfChangedPage(buf, 4, PAGE_SIZE));
- assertEquals(setIdx.size(), io.countOfChangedPage(buf, 3, PAGE_SIZE));
-
- for (long i = basePageId; i < basePageId + track; i++)
- assertEquals("pageId = " + i, setIdx.contains(i), io.wasChanged(buf, i, 3, -1, PAGE_SIZE));
-
- for (long i = basePageId; i < basePageId + track; i++)
- assertEquals("pageId = " + i, setIdx2.contains(i), io.wasChanged(buf, i, 4, 3, PAGE_SIZE));
-
- for (long i = basePageId; i < basePageId + track; i++)
- assertFalse(io.wasChanged(buf, i, 5, 4, PAGE_SIZE));
- }
-
- /**
- *
- */
- public void testMerging_MarksShouldBeDropForSuccessfulBackup() {
- ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
- buf.order(ByteOrder.nativeOrder());
-
- ThreadLocalRandom rand = ThreadLocalRandom.current();
-
- int track = io.countOfPageToTrack(PAGE_SIZE);
-
- long basePageId = io.trackingPageFor(Math.max(rand.nextLong(Integer.MAX_VALUE - track), 0), PAGE_SIZE);
-
- assert basePageId >= 0;
-
- PageIO.setPageId(GridUnsafe.bufferAddress(buf), basePageId);
-
- TreeSet<Long> setIdx = new TreeSet<>();
-
- for (int i = 0; i < 4; i++)
- generateMarking(buf, track, basePageId, basePageId + rand.nextInt(1, track), setIdx, i, -1);
-
- setIdx.clear();
-
- generateMarking(buf, track, basePageId, basePageId + rand.nextInt(1, track), setIdx, 4, -1);
-
- TreeSet<Long> setIdx2 = new TreeSet<>();
-
- generateMarking(buf, track, basePageId, basePageId + rand.nextInt(1, track), setIdx2, 5, 3);
-
- assertEquals(setIdx.size(), io.countOfChangedPage(buf, 4, PAGE_SIZE));
- assertEquals(setIdx2.size(), io.countOfChangedPage(buf, 5, PAGE_SIZE));
-
- for (long i = basePageId; i < basePageId + track; i++)
- assertEquals("pageId = " + i, setIdx2.contains(i), io.wasChanged(buf, i, 5, 4, PAGE_SIZE));
- }
-
- private void generateMarking(
- ByteBuffer buf,
- int track,
- long basePageId,
- long maxPageId,
- Set<Long> setIdx,
- int backupId, int successfulBackupId
- ) {
- ThreadLocalRandom rand = ThreadLocalRandom.current();
-
- for (long i = basePageId; i < basePageId + track; i++) {
- boolean changed = (i == basePageId || rand.nextDouble() < 0.1) && i < maxPageId;
-
- if (changed) {
- io.markChanged(buf, i, backupId, successfulBackupId, PAGE_SIZE);
-
- setIdx.add(i);
- }
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsAtomicCacheRebalancingTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsAtomicCacheRebalancingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsAtomicCacheRebalancingTest.java
new file mode 100644
index 0000000..21551b6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsAtomicCacheRebalancingTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+
+/**
+ *
+ */
+public class IgnitePdsAtomicCacheRebalancingTest extends IgnitePdsCacheRebalancingAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheConfiguration cacheConfiguration(String cacheName) {
+ CacheConfiguration ccfg = new CacheConfiguration(cacheName);
+
+ ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
+ ccfg.setCacheMode(CacheMode.PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+
+ return ccfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsCacheRebalancingAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsCacheRebalancingAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsCacheRebalancingAbstractTest.java
new file mode 100644
index 0000000..a9428f8
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsCacheRebalancingAbstractTest.java
@@ -0,0 +1,523 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.PartitionLossPolicy;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+
+import static org.apache.ignite.testframework.GridTestUtils.runMultiThreadedAsync;
+
+/**
+ * Test for rebalancing and persistence integration.
+ */
+public abstract class IgnitePdsCacheRebalancingAbstractTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** Cache name. */
+ private final String cacheName = "cache";
+
+ /** */
+ protected boolean explicitTx;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration ccfg1 = cacheConfiguration(cacheName);
+ ccfg1.setPartitionLossPolicy(PartitionLossPolicy.READ_ONLY_SAFE);
+ ccfg1.setBackups(1);
+ ccfg1.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+
+ CacheConfiguration ccfg2 = cacheConfiguration("indexed");
+ ccfg2.setBackups(1);
+ ccfg2.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+
+ QueryEntity qryEntity = new QueryEntity(Integer.class.getName(), TestValue.class.getName());
+
+ LinkedHashMap<String, String> fields = new LinkedHashMap<>();
+
+ fields.put("v1", Integer.class.getName());
+ fields.put("v2", Integer.class.getName());
+
+ qryEntity.setFields(fields);
+
+ QueryIndex qryIdx = new QueryIndex("v1", true);
+
+ qryEntity.setIndexes(Collections.singleton(qryIdx));
+
+ ccfg2.setQueryEntities(Collections.singleton(qryEntity));
+
+ cfg.setCacheConfiguration(ccfg1, ccfg2);
+
+ MemoryConfiguration memCfg = new MemoryConfiguration();
+
+ memCfg.setConcurrencyLevel(Runtime.getRuntime().availableProcessors() * 4);
+ memCfg.setPageSize(1024);
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName("dfltMemPlc");
+ memPlcCfg.setMaxSize(100 * 1024 * 1024);
+ memPlcCfg.setInitialSize(100 * 1024 * 1024);
+ memPlcCfg.setSwapFilePath("work/swap");
+
+ memCfg.setMemoryPolicies(memPlcCfg);
+ memCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+ cfg.setMemoryConfiguration(memCfg);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ );
+
+ cfg.setDiscoverySpi(
+ new TcpDiscoverySpi()
+ .setIpFinder(IP_FINDER)
+ );
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 20 * 60 * 1000;
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @return Cache configuration.
+ */
+ protected abstract CacheConfiguration cacheConfiguration(String cacheName);
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ stopAllGrids();
+
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /**
+ * Test that outdated partitions on restarted nodes are correctly replaced with newer versions.
+ *
+ * @throws Exception If fails.
+ */
+ public void testRebalancingOnRestart() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ startGrid(1);
+
+ IgniteEx ignite2 = startGrid(2);
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<Integer, Integer> cache1 = ignite0.cache(cacheName);
+
+ for (int i = 0; i < 5000; i++)
+ cache1.put(i, i);
+
+ ignite2.close();
+
+ awaitPartitionMapExchange();
+
+ ignite0.resetLostPartitions(Collections.singletonList(cache1.getName()));
+
+ assert cache1.lostPartitions().isEmpty();
+
+ for (int i = 0; i < 5000; i++)
+ cache1.put(i, i * 2);
+
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+
+ info(">>> Done puts...");
+
+ ignite2 = startGrid(2);
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<Integer, Integer> cache3 = ignite2.cache(cacheName);
+
+ for (int i = 0; i < 100; i++)
+ assertEquals(String.valueOf(i), (Integer)(i * 2), cache3.get(i));
+ }
+
+ /**
+ * Test that outdated partitions on restarted nodes are correctly replaced with newer versions.
+ *
+ * @throws Exception If fails.
+ */
+ public void testRebalancingOnRestartAfterCheckpoint() throws Exception {
+ IgniteEx ignite0 = startGrid(0);
+
+ IgniteEx ignite1 = startGrid(1);
+
+ IgniteEx ignite2 = startGrid(2);
+ IgniteEx ignite3 = startGrid(3);
+
+ ignite0.cache(cacheName).rebalance().get();
+ ignite1.cache(cacheName).rebalance().get();
+ ignite2.cache(cacheName).rebalance().get();
+ ignite3.cache(cacheName).rebalance().get();
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<Integer, Integer> cache1 = ignite0.cache(cacheName);
+
+ for (int i = 0; i < 1000; i++)
+ cache1.put(i, i);
+
+ ignite0.context().cache().context().database().waitForCheckpoint("test");
+ ignite1.context().cache().context().database().waitForCheckpoint("test");
+
+ info("++++++++++ After checkpoint");
+
+ ignite2.close();
+ ignite3.close();
+
+ awaitPartitionMapExchange();
+
+ ignite0.resetLostPartitions(Collections.singletonList(cache1.getName()));
+
+ assert cache1.lostPartitions().isEmpty();
+
+ for (int i = 0; i < 1000; i++)
+ cache1.put(i, i * 2);
+
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>");
+
+ info(">>> Done puts...");
+
+ ignite2 = startGrid(2);
+ ignite3 = startGrid(3);
+
+ ignite2.cache(cacheName).rebalance().get();
+ ignite3.cache(cacheName).rebalance().get();
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<Integer, Integer> cache2 = ignite2.cache(cacheName);
+ IgniteCache<Integer, Integer> cache3 = ignite3.cache(cacheName);
+
+ for (int i = 0; i < 100; i++) {
+ assertEquals(String.valueOf(i), (Integer)(i * 2), cache2.get(i));
+ assertEquals(String.valueOf(i), (Integer)(i * 2), cache3.get(i));
+ }
+ }
+
+ /**
+ * Test that all data is correctly restored after non-graceful restart.
+ *
+ * @throws Exception If fails.
+ */
+ public void testDataCorrectnessAfterRestart() throws Exception {
+ IgniteEx ignite1 = (IgniteEx)G.start(getConfiguration("test1"));
+ IgniteEx ignite2 = (IgniteEx)G.start(getConfiguration("test2"));
+ IgniteEx ignite3 = (IgniteEx)G.start(getConfiguration("test3"));
+ IgniteEx ignite4 = (IgniteEx)G.start(getConfiguration("test4"));
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<Integer, Integer> cache1 = ignite1.cache(cacheName);
+
+ for (int i = 0; i < 100; i++)
+ cache1.put(i, i);
+
+ ignite1.close();
+ ignite2.close();
+ ignite3.close();
+ ignite4.close();
+
+ ignite1 = (IgniteEx)G.start(getConfiguration("test1"));
+ ignite2 = (IgniteEx)G.start(getConfiguration("test2"));
+ ignite3 = (IgniteEx)G.start(getConfiguration("test3"));
+ ignite4 = (IgniteEx)G.start(getConfiguration("test4"));
+
+ awaitPartitionMapExchange();
+
+ cache1 = ignite1.cache(cacheName);
+ IgniteCache<Integer, Integer> cache2 = ignite2.cache(cacheName);
+ IgniteCache<Integer, Integer> cache3 = ignite3.cache(cacheName);
+ IgniteCache<Integer, Integer> cache4 = ignite4.cache(cacheName);
+
+ for (int i = 0; i < 100; i++) {
+ assert cache1.get(i).equals(i);
+ assert cache2.get(i).equals(i);
+ assert cache3.get(i).equals(i);
+ assert cache4.get(i).equals(i);
+ }
+ }
+
+ /**
+ * Test that partitions are marked as lost when all owners leave cluster, but recover after nodes rejoin.
+ *
+ * @throws Exception If fails.
+ */
+ public void testPartitionLossAndRecover() throws Exception {
+ fail("IGNITE-5302");
+
+ Ignite ignite1 = startGrid(0);
+ Ignite ignite2 = startGrid(1);
+ Ignite ignite3 = startGrid(2);
+ Ignite ignite4 = startGrid(3);
+
+ awaitPartitionMapExchange();
+
+ IgniteCache<String, String> cache1 = ignite1.cache(cacheName);
+
+ final int offset = 10;
+
+ for (int i = 0; i < 100; i++)
+ cache1.put(String.valueOf(i), String.valueOf(i + offset));
+
+ ignite3.close();
+ ignite4.close();
+
+ awaitPartitionMapExchange();
+
+ assert !ignite1.cache(cacheName).lostPartitions().isEmpty();
+
+ ignite3 = startGrid(2);
+ ignite4 = startGrid(3);
+
+ ignite1.resetLostPartitions(Collections.singletonList(cacheName));
+
+ IgniteCache<String, String> cache2 = ignite2.cache(cacheName);
+ IgniteCache<String, String> cache3 = ignite3.cache(cacheName);
+ IgniteCache<String, String> cache4 = ignite4.cache(cacheName);
+
+ //Thread.sleep(5_000);
+
+ for (int i = 0; i < 100; i++) {
+ String key = String.valueOf(i);
+ String expected = String.valueOf(i + offset);
+
+ assertEquals(expected, cache1.get(key));
+ assertEquals(expected, cache2.get(key));
+ assertEquals(expected, cache3.get(key));
+ assertEquals(expected, cache4.get(key));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTopologyChangesWithConstantLoad() throws Exception {
+ final int entriesCnt = 10_000;
+ int maxNodesCount = 4;
+ int topChanges = 20;
+ final String cacheName = "indexed";
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ final ConcurrentMap<Integer, TestValue> map = new ConcurrentHashMap<>();
+
+ Ignite ignite = startGrid(0);
+
+ IgniteCache<Integer, TestValue> cache = ignite.cache(cacheName);
+
+ for (int i = 0; i < entriesCnt; i++) {
+ cache.put(i, new TestValue(i, i));
+ map.put(i, new TestValue(i, i));
+ }
+
+ final AtomicInteger nodesCnt = new AtomicInteger();
+
+ IgniteInternalFuture fut = runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ while (true) {
+ if (stop.get())
+ return null;
+
+ int k = ThreadLocalRandom.current().nextInt(entriesCnt);
+ int v1 = ThreadLocalRandom.current().nextInt();
+ int v2 = ThreadLocalRandom.current().nextInt();
+
+ int n = nodesCnt.get();
+
+ if (n <= 0)
+ continue;
+
+ Ignite ignite;
+
+ try {
+ ignite = grid(ThreadLocalRandom.current().nextInt(n));
+ }
+ catch (Exception ignored) {
+ continue;
+ }
+
+ if (ignite == null)
+ continue;
+
+ Transaction tx = null;
+ boolean success = true;
+
+ if (explicitTx)
+ tx = ignite.transactions().txStart();
+
+ try {
+ ignite.cache(cacheName).put(k, new TestValue(v1, v2));
+ }
+ catch (Exception ignored) {
+ success = false;
+ }
+ finally {
+ if (tx != null) {
+ try {
+ tx.commit();
+ }
+ catch (Exception ignored) {
+ success = false;
+ }
+ }
+ }
+
+ if (success)
+ map.put(k, new TestValue(v1, v2));
+ }
+ }
+ }, 1, "load-runner");
+
+ for (int i = 0; i < topChanges; i++) {
+ U.sleep(3_000);
+
+ boolean add;
+
+ if (nodesCnt.get() <= maxNodesCount / 2)
+ add = true;
+ else if (nodesCnt.get() > maxNodesCount)
+ add = false;
+ else
+ add = ThreadLocalRandom.current().nextBoolean();
+
+ if (add)
+ startGrid(nodesCnt.incrementAndGet());
+ else
+ stopGrid(nodesCnt.getAndDecrement());
+
+ awaitPartitionMapExchange();
+
+ cache.rebalance().get();
+ }
+
+ stop.set(true);
+
+ fut.get();
+
+ awaitPartitionMapExchange();
+
+ for (Map.Entry<Integer, TestValue> entry : map.entrySet())
+ assertEquals(Integer.toString(entry.getKey()), entry.getValue(), cache.get(entry.getKey()));
+ }
+
+ /**
+ *
+ */
+ private static class TestValue implements Serializable {
+ /** V 1. */
+ private final int v1;
+ /** V 2. */
+ private final int v2;
+
+ /**
+ * @param v1 V 1.
+ * @param v2 V 2.
+ */
+ private TestValue(int v1, int v2) {
+ this.v1 = v1;
+ this.v2 = v2;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ TestValue val = (TestValue)o;
+
+ return v1 == val.v1 && v2 == val.v2;
+
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = v1;
+
+ res = 31 * res + v2;
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "TestValue{" +
+ "v1=" + v1 +
+ ", v2=" + v2 +
+ '}';
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsClientNearCachePutGetTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsClientNearCachePutGetTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsClientNearCachePutGetTest.java
new file mode 100644
index 0000000..17dad10
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsClientNearCachePutGetTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.processors.database.IgniteDbClientNearCachePutGetTest;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ *
+ */
+public class IgnitePdsClientNearCachePutGetTest extends IgniteDbClientNearCachePutGetTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ );
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+
+ super.beforeTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
new file mode 100644
index 0000000..3721031
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
@@ -0,0 +1,259 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import java.util.Map;
+import java.util.Random;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgnitePdsContinuousRestartTest extends GridCommonAbstractTest {
+ /** */
+ private static final int GRID_CNT = 4;
+
+ /** */
+ private static final int ENTRIES_COUNT = 10_000;
+
+ /** */
+ public static final String CACHE_NAME = "cache1";
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ MemoryConfiguration memCfg = new MemoryConfiguration();
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName("dfltMemPlc");
+ memPlcCfg.setMaxSize(400 * 1024 * 1024);
+ memPlcCfg.setInitialSize(400 * 1024 * 1024);
+
+ memCfg.setMemoryPolicies(memPlcCfg);
+ memCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+ cfg.setMemoryConfiguration(memCfg);
+
+ CacheConfiguration ccfg1 = new CacheConfiguration();
+
+ ccfg1.setName(CACHE_NAME);
+ ccfg1.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ ccfg1.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ ccfg1.setAffinity(new RendezvousAffinityFunction(false, 128));
+ ccfg1.setBackups(2);
+
+ cfg.setCacheConfiguration(ccfg1);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ );
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ stopAllGrids();
+
+ deleteWorkFiles();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ deleteWorkFiles();
+ }
+
+ /**
+ * @throws IgniteCheckedException If failed.
+ */
+ private void deleteWorkFiles() throws IgniteCheckedException {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_500_1_1() throws Exception {
+ checkRebalancingDuringLoad(1000, 500, 1, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_500_1_1() throws Exception {
+ checkRebalancingDuringLoad(8000, 500, 1, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_20000_1_1() throws Exception {
+ checkRebalancingDuringLoad(1000, 20000, 1, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_8000_1_1() throws Exception {
+ checkRebalancingDuringLoad(8000, 8000, 1, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_500_8_1() throws Exception {
+ checkRebalancingDuringLoad(1000, 500, 8, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_500_8_1() throws Exception {
+ checkRebalancingDuringLoad(8000, 500, 8, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_20000_8_1() throws Exception {
+ checkRebalancingDuringLoad(1000, 20000, 8, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_8000_8_1() throws Exception {
+ checkRebalancingDuringLoad(8000, 8000, 8, 1);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_500_8_16() throws Exception {
+ checkRebalancingDuringLoad(1000, 500, 8, 16);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_500_8_16() throws Exception {
+ checkRebalancingDuringLoad(8000, 500, 8, 16);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_1000_20000_8_16() throws Exception {
+ checkRebalancingDuringLoad(1000, 20000, 8, 16);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testRebalancingDuringLoad_8000_8000_8_16() throws Exception {
+ checkRebalancingDuringLoad(8000, 8000, 8, 16);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ private void checkRebalancingDuringLoad(
+ int restartDelay,
+ int checkpointDelay,
+ int threads,
+ final int batch
+ ) throws Exception {
+
+ startGrids(GRID_CNT);
+
+ final Ignite load = ignite(0);
+
+ try (IgniteDataStreamer<Object, Object> s = load.dataStreamer(CACHE_NAME)) {
+ s.allowOverwrite(true);
+
+ for (int i = 0; i < ENTRIES_COUNT; i++)
+ s.addData(i, i);
+ }
+
+ final AtomicBoolean done = new AtomicBoolean(false);
+
+ IgniteInternalFuture<?> busyFut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
+ /** {@inheritDoc} */
+ @Override public Object call() throws Exception {
+ IgniteCache<Object, Object> cache = load.cache(CACHE_NAME);
+ Random rnd = ThreadLocalRandom.current();
+
+ while (!done.get()) {
+ Map<Integer, Integer> map = new TreeMap<>();
+
+ for (int i = 0; i < batch; i++)
+ map.put(rnd.nextInt(ENTRIES_COUNT), rnd.nextInt());
+
+ cache.putAll(map);
+ }
+
+ return null;
+ }
+ }, threads, "updater");
+
+ long end = System.currentTimeMillis() + 90_000;
+
+ Random rnd = ThreadLocalRandom.current();
+
+ while (System.currentTimeMillis() < end) {
+ int idx = rnd.nextInt(GRID_CNT - 1) + 1;
+
+ stopGrid(idx);
+
+ U.sleep(restartDelay);
+
+ startGrid(idx);
+
+ U.sleep(restartDelay);
+ }
+
+ done.set(true);
+
+ busyFut.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsDynamicCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsDynamicCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsDynamicCacheTest.java
new file mode 100644
index 0000000..ba55c09
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsDynamicCacheTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheRebalanceMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.database.IgniteDbDynamicCacheSelfTest;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ *
+ */
+public class IgnitePdsDynamicCacheTest extends IgniteDbDynamicCacheSelfTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ MemoryConfiguration dbCfg = new MemoryConfiguration();
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName("dfltMemPlc");
+ memPlcCfg.setInitialSize(200 * 1024 * 1024);
+ memPlcCfg.setMaxSize(200 * 1024 * 1024);
+
+ dbCfg.setMemoryPolicies(memPlcCfg);
+ dbCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+ cfg.setMemoryConfiguration(dbCfg);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ );
+
+ if ("client".equals(gridName))
+ cfg.setClientMode(true);
+
+ return cfg;
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ System.setProperty(GridCacheDatabaseSharedManager.IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC, "true");
+
+ super.beforeTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ deleteWorkFiles();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ System.clearProperty(GridCacheDatabaseSharedManager.IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC);
+
+ deleteWorkFiles();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testRestartAndCreate() throws Exception {
+ startGrids(3);
+
+ Ignite ignite = ignite(0);
+
+ CacheConfiguration ccfg1 = new CacheConfiguration();
+
+ ccfg1.setName("cache1");
+ ccfg1.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ ccfg1.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ ccfg1.setRebalanceMode(CacheRebalanceMode.NONE);
+ ccfg1.setAffinity(new RendezvousAffinityFunction(false, 32));
+
+ CacheConfiguration ccfg2 = new CacheConfiguration();
+
+ ccfg2.setName("cache2");
+ ccfg2.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ ccfg2.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ ccfg2.setRebalanceMode(CacheRebalanceMode.NONE);
+ ccfg2.setAffinity(new RendezvousAffinityFunction(false, 32));
+
+ ignite.createCache(ccfg1);
+ ignite.createCache(ccfg2);
+
+ int iterations = 20;
+
+ long stopTime = U.currentTimeMillis() + 20_000;
+
+ for (int k = 0; k < iterations && U.currentTimeMillis() < stopTime; k++) {
+ log.info("Iteration: " + k);
+
+ stopAllGrids();
+
+ startGrids(3);
+
+ ignite = ignite(0);
+
+ ignite.getOrCreateCache(ccfg1);
+
+ ignite.getOrCreateCache(ccfg2);
+
+ ignite.destroyCache(ccfg2.getName());
+
+ ignite.getOrCreateCache(ccfg2);
+
+ ignite.destroyCache(ccfg1.getName());
+ }
+ }
+
+ /**
+ * @throws IgniteCheckedException If failed.
+ */
+ private void deleteWorkFiles() throws IgniteCheckedException {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsMultiNodePutGetRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsMultiNodePutGetRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsMultiNodePutGetRestartTest.java
new file mode 100644
index 0000000..21ea626
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsMultiNodePutGetRestartTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.CacheRebalanceMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.configuration.BinaryConfiguration;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgnitePdsMultiNodePutGetRestartTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int GRID_CNT = 3;
+
+ /** */
+ private final File allocPath;
+
+ /**
+ * Default constructor.
+ */
+ public IgnitePdsMultiNodePutGetRestartTest() {
+ String home = U.getIgniteHome();
+
+ allocPath = new File(home, "work/db/" + UUID.randomUUID());
+
+ allocPath.mkdirs();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ MemoryConfiguration memCfg = new MemoryConfiguration();
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName("dfltMemPlc");
+ memPlcCfg.setInitialSize(100 * 1024 * 1024);
+ memPlcCfg.setMaxSize(100 * 1024 * 1024);
+
+ memCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+ memCfg.setMemoryPolicies(memPlcCfg);
+
+ cfg.setMemoryConfiguration(memCfg);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setIndexedTypes(Integer.class, DbValue.class);
+
+ ccfg.setRebalanceMode(CacheRebalanceMode.NONE);
+
+ ccfg.setAffinity(new RendezvousAffinityFunction(false, 32));
+
+ ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ cfg.setPersistentStoreConfiguration(new PersistentStoreConfiguration());
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ cfg.setMarshaller(null);
+
+ BinaryConfiguration bCfg = new BinaryConfiguration();
+
+ bCfg.setCompactFooter(false);
+
+ cfg.setBinaryConfiguration(bCfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+
+ super.beforeTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPutGetSimple() throws Exception {
+ info(">>> Will use path: " + allocPath);
+
+ startGrids(GRID_CNT);
+
+ try {
+ IgniteEx ig = grid(0);
+
+ checkPutGetSql(ig, true);
+ }
+ finally {
+ stopAllGrids();
+ }
+
+ info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
+ info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
+
+ startGrids(GRID_CNT);
+
+ try {
+ IgniteEx ig = grid(0);
+
+ checkPutGetSql(ig, false);
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ * @param ig Ig.
+ * @param write Write.
+ */
+ private void checkPutGetSql(IgniteEx ig, boolean write) {
+ IgniteCache<Integer, DbValue> cache = ig.cache(null);
+
+ if (write) {
+ try (IgniteDataStreamer<Object, Object> streamer = ig.dataStreamer(null)) {
+ for (int i = 0; i < 10_000; i++)
+ streamer.addData(i, new DbValue(i, "value-" + i, i));
+ }
+ }
+
+ List<List<?>> res = cache.query(new SqlFieldsQuery("select ival from dbvalue where ival < ? order by ival asc")
+ .setArgs(10_000)).getAll();
+
+ assertEquals(10_000, res.size());
+
+ for (int i = 0; i < 10_000; i++) {
+ assertEquals(1, res.get(i).size());
+ assertEquals(i, res.get(i).get(0));
+ }
+
+ assertEquals(1, cache.query(new SqlFieldsQuery("select lval from dbvalue where ival = 7899")).getAll().size());
+ assertEquals(5000, cache.query(new SqlFieldsQuery("select lval from dbvalue where ival >= 5000 and ival < 10000"))
+ .getAll().size());
+
+ for (int i = 0; i < 10_000; i++)
+ assertEquals(new DbValue(i, "value-" + i, i), cache.get(i));
+ }
+
+ /**
+ *
+ */
+ private static class DbValue implements Serializable {
+ /** */
+ @QuerySqlField(index = true)
+ private int iVal;
+
+ /** */
+ @QuerySqlField(index = true)
+ private String sVal;
+
+ /** */
+ @QuerySqlField
+ private long lVal;
+
+ /**
+ * @param iVal Integer value.
+ * @param sVal String value.
+ * @param lVal Long value.
+ */
+ public DbValue(int iVal, String sVal, long lVal) {
+ this.iVal = iVal;
+ this.sVal = sVal;
+ this.lVal = lVal;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ DbValue dbVal = (DbValue)o;
+
+ return iVal == dbVal.iVal && lVal == dbVal.lVal &&
+ !(sVal != null ? !sVal.equals(dbVal.sVal) : dbVal.sVal != null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = iVal;
+
+ res = 31 * res + (sVal != null ? sVal.hashCode() : 0);
+ res = 31 * res + (int)(lVal ^ (lVal >>> 32));
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(DbValue.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsPageSizesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsPageSizesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsPageSizesTest.java
new file mode 100644
index 0000000..5014399
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsPageSizesTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgnitePdsPageSizesTest extends GridCommonAbstractTest {
+ /** Cache name. */
+ private final String cacheName = "cache";
+
+ /** */
+ private int pageSize;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ MemoryConfiguration memCfg = new MemoryConfiguration();
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName("dfltMemPlc");
+ memPlcCfg.setInitialSize(100 * 1024 * 1024);
+ memPlcCfg.setMaxSize(100 * 1024 * 1024);
+
+ memCfg.setMemoryPolicies(memPlcCfg);
+ memCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+ memCfg.setPageSize(pageSize);
+
+ cfg.setMemoryConfiguration(memCfg);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ );
+
+ cfg.setCacheConfiguration(
+ new CacheConfiguration(cacheName)
+ .setAffinity(new RendezvousAffinityFunction(false, 32))
+ );
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageSize_1k() throws Exception {
+ checkPageSize(1024);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageSize_2k() throws Exception {
+ checkPageSize(2 * 1024);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageSize_4k() throws Exception {
+ checkPageSize(4 * 1024);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageSize_8k() throws Exception {
+ checkPageSize(8 * 1024);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageSize_16k() throws Exception {
+ checkPageSize(16 * 1024);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ private void checkPageSize(int pageSize) throws Exception {
+ this.pageSize = pageSize;
+
+ IgniteEx ignite = startGrid(0);
+
+ try {
+ final IgniteCache<Object, Object> cache = ignite.cache(cacheName);
+ final long endTime = System.currentTimeMillis() + 60_000;
+
+ GridTestUtils.runMultiThreaded(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Random rnd = ThreadLocalRandom.current();
+
+ while (System.currentTimeMillis() < endTime) {
+ for (int i = 0; i < 500; i++)
+ cache.put(rnd.nextInt(100_000), rnd.nextInt());
+ }
+
+ return null;
+ }
+ }, 16, "runner");
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6bf5ce46/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
new file mode 100644
index 0000000..4981095
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
@@ -0,0 +1,361 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
+import java.util.Collection;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheRebalanceMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.pagemem.FullPageId;
+import org.apache.ignite.internal.pagemem.PageIdAllocator;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
+import org.apache.ignite.internal.pagemem.store.PageStore;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
+import org.apache.ignite.internal.pagemem.wal.WALPointer;
+import org.apache.ignite.internal.pagemem.wal.record.CheckpointRecord;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgnitePdsRecoveryAfterFileCorruptionTest extends GridCommonAbstractTest {
+ /** Ip finder. */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** Total pages. */
+ private static final int totalPages = 1024;
+
+ /** Cache name. */
+ private final String cacheName = "cache";
+
+ /** Policy name. */
+ private final String policyName = "dfltMemPlc";
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration ccfg = new CacheConfiguration(cacheName);
+ ccfg.setAffinity(new RendezvousAffinityFunction(true, 1));
+
+ ccfg.setRebalanceMode(CacheRebalanceMode.NONE);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ MemoryConfiguration dbCfg = new MemoryConfiguration();
+
+ MemoryPolicyConfiguration memPlcCfg = new MemoryPolicyConfiguration();
+
+ memPlcCfg.setName(policyName);
+ memPlcCfg.setInitialSize(1024 * 1024 * 1024);
+ memPlcCfg.setMaxSize(1024 * 1024 * 1024);
+
+ dbCfg.setMemoryPolicies(memPlcCfg);
+ dbCfg.setDefaultMemoryPolicyName(policyName);
+
+ cfg.setMemoryConfiguration(dbCfg);
+
+ cfg.setPersistentStoreConfiguration(
+ new PersistentStoreConfiguration()
+ .setCheckpointingFrequency(500)
+ .setAlwaysWriteFullPages(true)
+ );
+
+ cfg.setDiscoverySpi(
+ new TcpDiscoverySpi()
+ .setIpFinder(ipFinder)
+ );
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ stopAllGrids();
+
+ deleteWorkFiles();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ deleteWorkFiles();
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testPageRecoveryAfterFileCorruption() throws Exception {
+ IgniteEx ig = startGrid(0);
+
+ IgniteCache<Integer, Integer> cache = ig.cache(cacheName);
+
+ // Put for create data store and init meta page.
+ cache.put(1, 1);
+
+ GridCacheSharedContext sharedCtx = ig.context().cache().context();
+
+ GridCacheDatabaseSharedManager psMgr = (GridCacheDatabaseSharedManager)sharedCtx.database();
+
+ FilePageStoreManager pageStore = (FilePageStoreManager)sharedCtx.pageStore();
+
+ U.sleep(1_000);
+
+ // Disable integrated checkpoint thread.
+ psMgr.enableCheckpoints(false).get();
+
+ PageMemory mem = sharedCtx.database().memoryPolicy(policyName).pageMemory();
+
+ int cacheId = sharedCtx.cache().cache(cacheName).context().cacheId();
+
+ FullPageId[] pages = new FullPageId[totalPages];
+
+ for (int i = 0; i < totalPages; i++)
+ pages[i] = new FullPageId(mem.allocatePage(cacheId, 0, PageIdAllocator.FLAG_DATA), cacheId);
+
+ generateWal(
+ (PageMemoryImpl)mem,
+ sharedCtx.pageStore(),
+ sharedCtx.wal(),
+ cacheId,
+ pages
+ );
+
+ eraseDataFromDisk(pageStore, cacheId, pages[0]);
+
+ stopAllGrids();
+
+ ig = startGrid(0);
+
+ checkRestore(ig, pages);
+ }
+
+ /**
+ * @param pageStore Page store.
+ * @param cacheId Cache id.
+ * @param page Page.
+ */
+ private void eraseDataFromDisk(
+ FilePageStoreManager pageStore,
+ int cacheId,
+ FullPageId page
+ ) throws IgniteCheckedException, IOException {
+ PageStore store = pageStore.getStore(
+ cacheId,
+ PageIdUtils.partId(page.pageId())
+ );
+
+ FilePageStore filePageStore = (FilePageStore)store;
+
+ FileChannel ch = U.field(filePageStore, "ch");
+
+ long size = ch.size();
+
+ ch.write(ByteBuffer.allocate((int)size - FilePageStore.HEADER_SIZE), FilePageStore.HEADER_SIZE);
+
+ ch.force(false);
+ }
+
+ /**
+ * @param ig Ig.
+ * @param pages Pages.
+ */
+ private void checkRestore(IgniteEx ig, FullPageId[] pages) throws IgniteCheckedException {
+ GridCacheSharedContext<Object, Object> shared = ig.context().cache().context();
+
+ GridCacheDatabaseSharedManager dbMgr = (GridCacheDatabaseSharedManager)shared.database();
+
+ dbMgr.enableCheckpoints(false).get();
+
+ PageMemory mem = shared.database().memoryPolicy(null).pageMemory();
+
+ for (FullPageId fullId : pages) {
+ long page = mem.acquirePage(fullId.cacheId(), fullId.pageId());
+
+ try {
+ long pageAddr = mem.readLock(fullId.cacheId(), fullId.pageId(), page);
+
+ for (int j = PageIO.COMMON_HEADER_END; j < mem.pageSize(); j += 4)
+ assertEquals(j + (int)fullId.pageId(), PageUtils.getInt(pageAddr, j));
+
+ mem.readUnlock(fullId.cacheId(), fullId.pageId(), page);
+ }
+ finally {
+ mem.releasePage(fullId.cacheId(), fullId.pageId(), page);
+ }
+ }
+ }
+
+ /**
+ * @param mem Mem.
+ * @param storeMgr Store manager.
+ * @param wal Wal.
+ * @param cacheId Cache id.
+ * @param pages Pages.
+ */
+ private void generateWal(
+ final PageMemoryImpl mem,
+ final IgnitePageStoreManager storeMgr,
+ final IgniteWriteAheadLogManager wal,
+ final int cacheId, FullPageId[] pages
+ ) throws Exception {
+ // Mark the start position.
+ CheckpointRecord cpRec = new CheckpointRecord(null, false);
+
+ WALPointer start = wal.log(cpRec);
+
+ wal.fsync(start);
+
+ for (int i = 0; i < totalPages; i++) {
+ FullPageId fullId = pages[i];
+
+ long page = mem.acquirePage(fullId.cacheId(), fullId.pageId());
+
+ try {
+ long pageAddr = mem.writeLock(fullId.cacheId(), fullId.pageId(), page);
+
+ PageIO.setPageId(pageAddr, fullId.pageId());
+
+ try {
+ for (int j = PageIO.COMMON_HEADER_END; j < mem.pageSize(); j += 4)
+ PageUtils.putInt(pageAddr, j, j + (int)fullId.pageId());
+ }
+ finally {
+ mem.writeUnlock(fullId.cacheId(), fullId.pageId(), page, null, true);
+ }
+ }
+ finally {
+ mem.releasePage(fullId.cacheId(), fullId.pageId(), page);
+ }
+ }
+
+ Collection<FullPageId> pageIds = mem.beginCheckpoint();
+
+ info("Acquired pages for checkpoint: " + pageIds.size());
+
+ try {
+ ByteBuffer tmpBuf = ByteBuffer.allocate(mem.pageSize());
+
+ tmpBuf.order(ByteOrder.nativeOrder());
+
+ long begin = System.currentTimeMillis();
+
+ long cp = 0;
+
+ long write = 0;
+
+ for (int i = 0; i < totalPages; i++) {
+ FullPageId fullId = pages[i];
+
+ if (pageIds.contains(fullId)) {
+ long cpStart = System.nanoTime();
+
+ Integer tag = mem.getForCheckpoint(fullId, tmpBuf, null);
+
+ if (tag == null)
+ continue;
+
+ long cpEnd = System.nanoTime();
+
+ cp += cpEnd - cpStart;
+ tmpBuf.rewind();
+
+ for (int j = PageIO.COMMON_HEADER_END; j < mem.pageSize(); j += 4)
+ assertEquals(j + (int)fullId.pageId(), tmpBuf.getInt(j));
+
+ tmpBuf.rewind();
+
+ long writeStart = System.nanoTime();
+
+ storeMgr.write(cacheId, fullId.pageId(), tmpBuf, tag);
+
+ long writeEnd = System.nanoTime();
+
+ write += writeEnd - writeStart;
+
+ tmpBuf.rewind();
+ }
+ }
+
+ long syncStart = System.currentTimeMillis();
+
+ storeMgr.sync(cacheId, 0);
+
+ long end = System.currentTimeMillis();
+
+ info("Written pages in " + (end - begin) + "ms, copy took " + (cp / 1_000_000) + "ms, " +
+ "write took " + (write / 1_000_000) + "ms, sync took " + (end - syncStart) + "ms");
+ }
+ finally {
+ info("Finishing checkpoint...");
+
+ mem.finishCheckpoint();
+
+ info("Finished checkpoint");
+ }
+
+ wal.fsync(wal.log(new CheckpointRecord(null, false)));
+
+ for (FullPageId fullId : pages) {
+ long page = mem.acquirePage(fullId.cacheId(), fullId.pageId());
+
+ try {
+ assertFalse("Page has a temp heap copy after the last checkpoint: [cacheId=" +
+ fullId.cacheId() + ", pageId=" + fullId.pageId() + "]", mem.hasTempCopy(page));
+
+ assertFalse("Page is dirty after the last checkpoint: [cacheId=" +
+ fullId.cacheId() + ", pageId=" + fullId.pageId() + "]", mem.isDirty(fullId.cacheId(), fullId.pageId(), page));
+ }
+ finally {
+ mem.releasePage(fullId.cacheId(), fullId.pageId(), page);
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private void deleteWorkFiles() throws IgniteCheckedException {
+ deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+ }
+}