You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2018/07/11 15:45:40 UTC
ignite git commit: IGNITE-8946 AssertionError can occur during
release of WAL history that was reserved for historical rebalance
Repository: ignite
Updated Branches:
refs/heads/master ee909a36d -> 54055ec0c
IGNITE-8946 AssertionError can occur during release of WAL history that was reserved for historical rebalance
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/54055ec0
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/54055ec0
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/54055ec0
Branch: refs/heads/master
Commit: 54055ec0c09caadf939e00a3a86098f610b52d8c
Parents: ee909a3
Author: Ivan Rakov <ir...@apache.org>
Authored: Wed Jul 11 18:45:31 2018 +0300
Committer: Ivan Rakov <ir...@apache.org>
Committed: Wed Jul 11 18:45:31 2018 +0300
----------------------------------------------------------------------
.../checkpoint/CheckpointHistory.java | 35 +--
...SlowHistoricalRebalanceSmallHistoryTest.java | 236 +++++++++++++++++++
.../ignite/testsuites/IgnitePdsTestSuite2.java | 3 +
3 files changed, 257 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/54055ec0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistory.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistory.java
index cef2093..95f150b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistory.java
@@ -326,26 +326,13 @@ public class CheckpointHistory {
if (!reserved)
break;
- for (Integer grpId : groupsAndPartitions.keySet())
+ for (Integer grpId : new HashSet<>(groupsAndPartitions.keySet()))
if (!isCheckpointApplicableForGroup(grpId, chpEntry))
groupsAndPartitions.remove(grpId);
- // All groups are no more applicable, release history and stop searching.
- if (groupsAndPartitions.isEmpty()) {
- cctx.wal().release(chpEntry.checkpointMark());
-
- break;
- }
-
- // Release previous checkpoint marker.
- if (prevReserved != null)
- cctx.wal().release(prevReserved.checkpointMark());
-
- prevReserved = chpEntry;
-
for (Map.Entry<Integer, CheckpointEntry.GroupState> state : chpEntry.groupState(cctx).entrySet()) {
int grpId = state.getKey();
- CheckpointEntry.GroupState cpGroupState = state.getValue();
+ CheckpointEntry.GroupState cpGrpState = state.getValue();
Set<Integer> applicablePartitions = groupsAndPartitions.get(grpId);
@@ -355,7 +342,7 @@ public class CheckpointHistory {
Set<Integer> inapplicablePartitions = null;
for (Integer partId : applicablePartitions) {
- int pIdx = cpGroupState.indexByPartition(partId);
+ int pIdx = cpGrpState.indexByPartition(partId);
if (pIdx >= 0)
res.computeIfAbsent(grpId, k -> new HashMap<>()).put(partId, chpEntry);
@@ -374,9 +361,23 @@ public class CheckpointHistory {
}
// Remove groups from search with empty set of applicable partitions.
- for (Map.Entry<Integer, Set<Integer>> e : groupsAndPartitions.entrySet())
+ for (Map.Entry<Integer, Set<Integer>> e : new HashSet<>(groupsAndPartitions.entrySet()))
if (e.getValue().isEmpty())
groupsAndPartitions.remove(e.getKey());
+
+ // All groups are no more applicable, release history and stop searching.
+ if (groupsAndPartitions.isEmpty()) {
+ cctx.wal().release(chpEntry.checkpointMark());
+
+ break;
+ }
+ else {
+ // Release previous checkpoint marker.
+ if (prevReserved != null)
+ cctx.wal().release(prevReserved.checkpointMark());
+
+ prevReserved = chpEntry;
+ }
}
catch (IgniteCheckedException ex) {
U.error(log, "Failed to process checkpoint: " + (chpEntry != null ? chpEntry : "none"), ex);
http://git-wip-us.apache.org/repos/asf/ignite/blob/54055ec0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/SlowHistoricalRebalanceSmallHistoryTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/SlowHistoricalRebalanceSmallHistoryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/SlowHistoricalRebalanceSmallHistoryTest.java
new file mode 100644
index 0000000..8f2e738
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/SlowHistoricalRebalanceSmallHistoryTest.java
@@ -0,0 +1,236 @@
+/*
+* 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.db;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheGroupIdMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+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 SlowHistoricalRebalanceSmallHistoryTest extends GridCommonAbstractTest {
+ /** Ip finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** Slow rebalance cache name. */
+ private static final String SLOW_REBALANCE_CACHE = "b13813ce";
+
+ /** Regular cache name. */
+ private static final String REGULAR_CACHE = "another-cache";
+
+ /** Supply message latch. */
+ private static final AtomicReference<CountDownLatch> SUPPLY_MESSAGE_LATCH = new AtomicReference<>();
+
+ /** Wal history size. */
+ private static final int WAL_HISTORY_SIZE = 5;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String name) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(name);
+
+ cfg.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(IP_FINDER));
+
+ cfg.setDataStorageConfiguration(
+ new DataStorageConfiguration()
+ .setWalHistorySize(WAL_HISTORY_SIZE)
+ .setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration()
+ .setPersistenceEnabled(true)
+ )
+ .setWalSegmentSize(512 * 1024)
+ );
+
+ cfg.setFailureHandler(new StopNodeFailureHandler());
+
+ cfg.setCommunicationSpi(new RebalanceBlockingSPI());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ SUPPLY_MESSAGE_LATCH.set(new CountDownLatch(1));
+
+ System.setProperty(IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD, "1000");
+
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ SUPPLY_MESSAGE_LATCH.get().countDown();
+
+ SUPPLY_MESSAGE_LATCH.set(null);
+
+ System.clearProperty(IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD);
+
+ cleanPersistenceDir();
+ }
+
+ /**
+ * Checks that we reserve and release the same WAL index on exchange.
+ */
+ public void testReservation() throws Exception {
+ IgniteEx ig = startGrid(0);
+
+ ig.cluster().active(true);
+
+ ig.getOrCreateCache(new CacheConfiguration<>()
+ .setName(SLOW_REBALANCE_CACHE)
+ .setAffinity(new RendezvousAffinityFunction(false, 1))
+ .setBackups(1)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setRebalanceBatchSize(100));
+
+ try (IgniteDataStreamer<Object, Object> streamer = ig.dataStreamer(SLOW_REBALANCE_CACHE)) {
+ for (int i = 0; i < 3_000; i++)
+ streamer.addData(i, new byte[5 * 1000]);
+
+ streamer.flush();
+ }
+
+ startGrid(1);
+
+ resetBaselineTopology();
+
+ IgniteCache<Object, Object> anotherCache = ig.getOrCreateCache(new CacheConfiguration<>()
+ .setName(REGULAR_CACHE)
+ .setAffinity(new RendezvousAffinityFunction(false, 1))
+ .setBackups(1)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setRebalanceBatchSize(100));
+
+ Thread.sleep(7_000); // To let distributedExchange() finish.
+
+ for (int i = 0; i < WAL_HISTORY_SIZE; i++) {
+ for (int j = 0; i < 500; i++)
+ anotherCache.put(j, new byte[5 * 1000]);
+
+ forceCheckpoint(); // Checkpoints where partition is OWNING on grid(0), MOVING on grid(1)
+
+ for (int j = 0; i < 500; i++)
+ anotherCache.put(j, new byte[5 * 1000]);
+ }
+
+ SUPPLY_MESSAGE_LATCH.get().countDown();
+
+ waitForRebalancing(); // Partition is OWNING on grid(0) and grid(1)
+
+ for (int i = 0; i < 2; i++) {
+ for (int j = 0; i < 500; i++)
+ anotherCache.put(j, new byte[5 * 1000]);
+
+ forceCheckpoint(); // A few more checkpoints when partition is OWNING everywhere
+
+ for (int j = 0; i < 500; i++)
+ anotherCache.put(j, new byte[5 * 1000]);
+ }
+
+ stopGrid(0);
+
+ IgniteCache<Object, Object> anotherCacheGrid1 = grid(1).cache(REGULAR_CACHE);
+
+ for (int i = 0; i < 500; i++)
+ anotherCacheGrid1.put(i, new byte[5 * 1000]);
+
+ startGrid(0);
+
+ waitForRebalancing();
+
+ assertEquals(2, grid(1).context().discovery().aliveServerNodes().size());
+ }
+
+ /**
+ *
+ */
+ private static class RebalanceBlockingSPI extends TcpCommunicationSpi {
+ /** */
+ public static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException {
+ if (msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof GridDhtPartitionSupplyMessage) {
+ int grpId = ((GridCacheGroupIdMessage)((GridIoMessage)msg).message()).groupId();
+
+ if (grpId == CU.cacheId(SLOW_REBALANCE_CACHE)) {
+ CountDownLatch latch0 = SUPPLY_MESSAGE_LATCH.get();
+
+ if (latch0 != null)
+ try {
+ latch0.await();
+ }
+ catch (InterruptedException ex) {
+ throw new IgniteException(ex);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg,
+ IgniteInClosure<IgniteException> ackC) throws IgniteSpiException {
+ if (msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof GridDhtPartitionSupplyMessage) {
+ int grpId = ((GridCacheGroupIdMessage)((GridIoMessage)msg).message()).groupId();
+
+ if (grpId == CU.cacheId(SLOW_REBALANCE_CACHE)) {
+ CountDownLatch latch0 = SUPPLY_MESSAGE_LATCH.get();
+
+ if (latch0 != null)
+ try {
+ latch0.await();
+ }
+ catch (InterruptedException ex) {
+ throw new IgniteException(ex);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54055ec0/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
index 1b25b0f..529fb60 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsPageE
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsRebalancingOnNotStableTopologyTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsTransactionsHangTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsWholeClusterRestartTest;
+import org.apache.ignite.internal.processors.cache.persistence.db.SlowHistoricalRebalanceSmallHistoryTest;
import org.apache.ignite.internal.processors.cache.persistence.db.checkpoint.IgniteCheckpointDirtyPagesForLowLoadTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsUnusedWalSegmentsTest;
import org.apache.ignite.internal.processors.cache.persistence.db.filename.IgniteUidAsConsistentIdMigrationTest;
@@ -142,6 +143,8 @@ public class IgnitePdsTestSuite2 extends TestSuite {
// Rebalancing test
suite.addTestSuite(IgniteWalHistoryReservationsTest.class);
+ suite.addTestSuite(SlowHistoricalRebalanceSmallHistoryTest.class);
+
suite.addTestSuite(IgnitePersistentStoreDataStructuresTest.class);
// Failover test