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/08 16:11:29 UTC
[11/11] ignite git commit: IGNITE-5267 - Moved WAL system properties
to WAL configuration
IGNITE-5267 - Moved WAL system properties to WAL configuration
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/cd4d0400
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/cd4d0400
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/cd4d0400
Branch: refs/heads/ignite-5267-merge-ea
Commit: cd4d04007ca7ad1bfbd7f358f9182f970d38ec3e
Parents: eff3e09
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Thu Jun 8 19:10:54 2017 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Thu Jun 8 19:10:55 2017 +0300
----------------------------------------------------------------------
.../PersistentStoreConfiguration.java | 158 ++-
.../apache/ignite/configuration/WALMode.java | 43 +
.../GridCacheDatabaseSharedManager.java | 2 +-
.../database/wal/FileWriteAheadLogManager.java | 245 ++--
.../IgnitePdsAtomicCacheRebalancingTest.java | 40 +
.../IgnitePdsCacheRebalancingAbstractTest.java | 523 +++++++
.../IgnitePdsClientNearCachePutGetTest.java | 55 +
.../IgnitePdsContinuousRestartTest.java | 259 ++++
.../database/IgnitePdsDynamicCacheTest.java | 150 ++
.../IgnitePdsMultiNodePutGetRestartTest.java | 252 ++++
.../cache/database/IgnitePdsPageSizesTest.java | 152 +++
...gnitePdsRecoveryAfterFileCorruptionTest.java | 361 +++++
.../IgnitePdsRemoveDuringRebalancingTest.java | 143 ++
...gnitePdsSingleNodePutGetPersistenceTest.java | 55 +
...dexingAndGroupPutGetPersistenceSelfTest.java | 42 +
...leNodeWithIndexingPutGetPersistenceTest.java | 55 +
.../IgnitePdsTxCacheRebalancingTest.java | 61 +
.../IgnitePersistenceMetricsSelfTest.java | 14 +-
...rsistentStoreAtomicCacheRebalancingTest.java | 40 -
.../IgnitePersistentStoreCacheGroupsTest.java | 14 +-
...istentStoreCacheRebalancingAbstractTest.java | 522 -------
...tNearCachePutGetWithPersistenceSelfTest.java | 56 -
...ersistentStoreContinuousRestartSelfTest.java | 265 ----
...IgnitePersistentStoreDataStructuresTest.java | 13 +-
.../IgnitePersistentStoreDynamicCacheTest.java | 150 --
...tentStoreMultiNodePutGetRestartSelfTest.java | 252 ----
.../IgnitePersistentStorePageSizesTest.java | 156 ---
...entStoreRecoveryAfterFileCorruptionTest.java | 365 -----
...entStoreRemoveDuringRebalancingSelfTest.java | 142 --
...toreSingleNodePutGetPersistenceSelfTest.java | 56 -
...dexingAndGroupPutGetPersistenceSelfTest.java | 42 -
...deWithIndexingPutGetPersistenceSelfTest.java | 56 -
...tePersistentStoreTxCacheRebalancingTest.java | 61 -
.../IgnitePersistentStoreWalTlbSelfTest.java | 144 --
...ageEvictionDuringPartitionClearSelfTest.java | 173 ---
.../IgniteDbMultiNodePutGetRestartSelfTest.java | 250 ----
.../db/IgniteDbPageEvictionSelfTest.java | 241 ----
.../db/IgniteDbWholeClusterRestartSelfTest.java | 171 ---
.../db/IgnitePdsMultiNodePutGetRestartTest.java | 242 ++++
...PdsPageEvictionDuringPartitionClearTest.java | 174 +++
.../database/db/IgnitePdsPageEvictionTest.java | 239 ++++
...tePdsRebalancingOnNotStableTopologyTest.java | 215 +++
.../db/IgnitePdsTransactionsHangTest.java | 304 +++++
.../db/IgnitePdsWholeClusterRestartTest.java | 165 +++
.../db/RebalancingOnNotStableTopologyTest.java | 216 ---
.../cache/database/db/TransactionsHangTest.java | 305 -----
...IgniteCachePageStoreIntegrationSelfTest.java | 290 ----
.../file/IgniteNoActualWalHistorySelfTest.java | 219 ---
.../db/file/IgnitePdsCacheIntegrationTest.java | 285 ++++
.../database/db/file/IgnitePdsEvictionTest.java | 297 ++++
.../file/IgnitePdsNoActualWalHistoryTest.java | 215 +++
.../IgniteWalDirectoriesConfigurationTest.java | 54 -
.../IgniteWalHistoryReservationsSelfTest.java | 395 ------
.../db/file/IgniteWalRecoverySelfTest.java | 1286 -----------------
.../IgniteWalRecoverySeveralRestartsTest.java | 350 -----
.../PageStoreCheckpointSimulationSelfTest.java | 1002 --------------
.../db/file/PageStoreEvictionSelfTest.java | 297 ----
.../file/WalRecoveryTxLogicalRecordsTest.java | 951 -------------
.../file/ignitePdsCheckpointSimulationTest.java | 1000 ++++++++++++++
.../database/db/wal/IgnitePdsWalTlbTest.java | 134 ++
.../IgniteWalDirectoriesConfigurationTest.java | 54 +
.../wal/IgniteWalHistoryReservationsTest.java | 395 ++++++
.../IgniteWalRecoverySeveralRestartsTest.java | 347 +++++
.../database/db/wal/IgniteWalRecoveryTest.java | 1288 ++++++++++++++++++
.../db/wal/WalRecoveryTxLogicalRecordsTest.java | 951 +++++++++++++
.../pagemem/BPlusTreePageMemoryImplTest.java | 98 ++
...PlusTreeReuseListPageMemoryImplSelfTest.java | 114 --
.../BPlusTreeReuseListPageMemoryImplTest.java | 99 ++
...BPlusTreeSelfTestPageMemoryImplSelfTest.java | 113 --
.../MetadataStoragePageMemoryImplSelfTest.java | 104 --
.../MetadataStoragePageMemoryImplTest.java | 104 ++
.../pagemem/PageMemoryImplNoLoadSelfTest.java | 100 --
.../pagemem/PageMemoryImplNoLoadTest.java | 100 ++
.../GridChangeGlobalStateAbstractTest.java | 366 -----
.../GridChangeGlobalStateCacheTest.java | 168 ---
.../GridChangeGlobalStateDataStreamerTest.java | 112 --
.../GridChangeGlobalStateDataStructureTest.java | 278 ----
.../GridChangeGlobalStateFailOverTest.java | 353 -----
.../GridChangeGlobalStateServiceTest.java | 102 --
.../GridChangeGlobalStateTest.java | 711 ----------
.../IgniteChangeGlobalStateAbstractTest.java | 366 +++++
.../IgniteChangeGlobalStateCacheTest.java | 168 +++
...IgniteChangeGlobalStateDataStreamerTest.java | 112 ++
...gniteChangeGlobalStateDataStructureTest.java | 278 ++++
.../IgniteChangeGlobalStateFailOverTest.java | 353 +++++
.../IgniteChangeGlobalStateServiceTest.java | 102 ++
.../IgniteChangeGlobalStateTest.java | 711 ++++++++++
.../IgnitePdsOutOfMemoryTestSuite.java | 4 +-
.../ignite/testsuites/IgnitePdsTestSuite.java | 58 +-
.../ignite/testsuites/IgnitePdsTestSuite2.java | 53 +-
.../testsuites/IgniteStandByClusterSuite.java | 22 +-
91 files changed, 11338 insertions(+), 11260 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd4d0400/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
index 362dcdf..d3cee9f 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
@@ -48,13 +48,37 @@ public class PersistentStoreConfiguration implements Serializable {
public static final int DFLT_CHECKPOINTING_THREADS = 1;
/** */
- private static final int DFLT_WAL_HISTORY_SIZE = 20;
+ public static final int DFLT_WAL_HISTORY_SIZE = 20;
/** */
public static final int DFLT_WAL_SEGMENTS = 10;
/** */
- private static final int DFLT_WAL_SEGMENT_SIZE = 64 * 1024 * 1024;
+ public static final int DFLT_WAL_SEGMENT_SIZE = 64 * 1024 * 1024;
+
+ /** Default wal mode. */
+ public static final WALMode DFLT_WAL_MODE = WALMode.DEFAULT;
+
+ /** Default thread local buffer size. */
+ public static final int DFLT_TLB_SIZE = 128 * 1024;
+
+ /** Default Wal flush frequency. */
+ public static final int DFLT_WAL_FLUSH_FREQ = 2000;
+
+ /** Default wal fsync delay. */
+ public static final int DFLT_WAL_FSYNC_DELAY = 1;
+
+ /** Default wal record iterator buffer size. */
+ public static final int DFLT_WAL_RECORD_ITERATOR_BUFFER_SIZE = 64 * 1024 * 1024;
+
+ /** Default wal always write full pages. */
+ public static final boolean DFLT_WAL_ALWAYS_WRITE_FULL_PAGES = false;
+
+ /** Default wal directory. */
+ public static final String DFLT_WAL_STORE_PATH = "db/wal";
+
+ /** Default wal archive directory. */
+ public static final String DFLT_WAL_ARCHIVE_PATH = "db/wal/archive";
/** */
private String persistenceStorePath;
@@ -81,14 +105,32 @@ public class PersistentStoreConfiguration implements Serializable {
private int walSegmentSize = DFLT_WAL_SEGMENT_SIZE;
/** WAL persistence path. */
- private String walStorePath;
+ private String walStorePath = DFLT_WAL_STORE_PATH;
/** WAL archive path. */
- private String walArchivePath;
+ private String walArchivePath = DFLT_WAL_ARCHIVE_PATH;
/** Metrics enabled flag. */
private boolean metricsEnabled = DFLT_METRICS_ENABLED;
+ /** Wal mode. */
+ private WALMode walMode = DFLT_WAL_MODE;
+
+ /** WAl thread local buffer size. */
+ private int tlbSize = DFLT_TLB_SIZE;
+
+ /** Wal flush frequency. */
+ private int walFlushFreq = DFLT_WAL_FLUSH_FREQ;
+
+ /** Wal fsync delay. */
+ private int walFsyncDelay = DFLT_WAL_FSYNC_DELAY;
+
+ /** Wal record iterator buffer size. */
+ private int walRecordIterBuffSize = DFLT_WAL_RECORD_ITERATOR_BUFFER_SIZE;
+
+ /** Always write full pages. */
+ private boolean alwaysWriteFullPages = DFLT_WAL_ALWAYS_WRITE_FULL_PAGES;
+
/**
* Number of sub-intervals the whole {@link #setRateTimeInterval(long)} will be split into to calculate
* rate-based metrics.
@@ -380,4 +422,112 @@ public class PersistentStoreConfiguration implements Serializable {
return this;
}
+
+ /**
+ * Type define behavior wal fsync.
+ * Different type provide different guarantees for consistency.
+ *
+ * @return WAL mode.
+ */
+ public WALMode getWalMode() {
+ return walMode == null ? DFLT_WAL_MODE : walMode;
+ }
+
+ /**
+ * @param walMode Wal mode.
+ */
+ public PersistentStoreConfiguration setWalMode(WALMode walMode) {
+ this.walMode = walMode;
+
+ return this;
+ }
+
+ /**
+ * Property define size thread local buffer.
+ * Each thread which write to wal have thread local buffer for serialize recode before write in wal.
+ *
+ * @return Thread local buffer size.
+ */
+ public int getTlbSize() {
+ return tlbSize <= 0 ? DFLT_TLB_SIZE : tlbSize;
+ }
+
+ /**
+ * @param tlbSize Tlb size.
+ */
+ public PersistentStoreConfiguration setTlbSize(int tlbSize) {
+ this.tlbSize = tlbSize;
+
+ return this;
+ }
+
+ /**
+ * Property define how often will be fsync.
+ * In background mode, exist thread which do fsync by timeout.
+ *
+ * @return Flush frequency.
+ */
+ public int getWalFlushFrequency() {
+ return walFlushFreq;
+ }
+
+ /**
+ * @param walFlushFreq Wal flush frequency.
+ */
+ public PersistentStoreConfiguration setWalFlushFrequency(int walFlushFreq) {
+ this.walFlushFreq = walFlushFreq;
+
+ return this;
+ }
+
+ /**
+ *
+ */
+ public int getWalFsyncDelay() {
+ return walFsyncDelay <= 0 ? DFLT_WAL_FSYNC_DELAY : walFsyncDelay;
+ }
+
+ /**
+ * @param walFsyncDelay Wal fsync delay.
+ */
+ public PersistentStoreConfiguration setWalFsyncDelay(int walFsyncDelay) {
+ this.walFsyncDelay = walFsyncDelay;
+
+ return this;
+ }
+
+ /**
+ * Property define how many bytes iterator read from
+ * disk (for one reading), during go ahead wal.
+ *
+ * @return Record iterator buffer size.
+ */
+ public int getWalRecordIteratorBufferSize() {
+ return walRecordIterBuffSize <= 0 ? DFLT_WAL_RECORD_ITERATOR_BUFFER_SIZE : walRecordIterBuffSize;
+ }
+
+ /**
+ * @param walRecordIterBuffSize Wal record iterator buffer size.
+ */
+ public PersistentStoreConfiguration setWalRecordIteratorBufferSize(int walRecordIterBuffSize) {
+ this.walRecordIterBuffSize = walRecordIterBuffSize;
+
+ return this;
+ }
+
+ /**
+ *
+ */
+ public boolean isAlwaysWriteFullPages() {
+ return alwaysWriteFullPages;
+ }
+
+ /**
+ * @param alwaysWriteFullPages Always write full pages.
+ */
+ public PersistentStoreConfiguration setAlwaysWriteFullPages(boolean alwaysWriteFullPages) {
+ this.alwaysWriteFullPages = alwaysWriteFullPages;
+
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd4d0400/modules/core/src/main/java/org/apache/ignite/configuration/WALMode.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/WALMode.java b/modules/core/src/main/java/org/apache/ignite/configuration/WALMode.java
new file mode 100644
index 0000000..f30ab8d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/WALMode.java
@@ -0,0 +1,43 @@
+/*
+ * 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.configuration;
+
+/**
+ * WAL Mode. This enum defines crash recovery guarantees when Ignite persistence is enabled.
+ */
+public enum WALMode {
+ /**
+ *
+ */
+ DEFAULT,
+
+ /**
+ *
+ */
+ LOG_ONLY,
+
+ /**
+ *
+ */
+ BACKGROUND,
+
+ /**
+ *
+ */
+ NONE
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd4d0400/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java b/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
index 1faffde..555fa60 100755
--- a/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
+++ b/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
@@ -603,7 +603,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
snapshotMgr.stop(cancel);
}
- /** {@inheritDoc} */
+ /** */
private long[] calculateFragmentSizes(int concLvl, long cacheSize) {
if (concLvl < 2)
concLvl = Runtime.getRuntime().availableProcessors();
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd4d0400/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/wal/FileWriteAheadLogManager.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/wal/FileWriteAheadLogManager.java b/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/wal/FileWriteAheadLogManager.java
index 292118c..51a2437 100644
--- a/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/wal/FileWriteAheadLogManager.java
+++ b/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/wal/FileWriteAheadLogManager.java
@@ -43,9 +43,9 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
@@ -76,6 +76,9 @@ import org.jetbrains.annotations.Nullable;
*/
public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter implements IgniteWriteAheadLogManager {
/** */
+ public static final FileDescriptor[] EMPTY_DESCRIPTORS = new FileDescriptor[0];
+
+ /** */
public static final String WAL_SEGMENT_FILE_EXT = ".wal";
/** */
@@ -101,42 +104,26 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
}
};
- /** System property (env variable) for configuring DB WAL mode, see values in {@link Mode} */
- public static final String IGNITE_PDS_WAL_MODE = "IGNITE_PDS_WAL_MODE";
-
- /** Thread local byte buffer size */
- public static final String IGNITE_PDS_WAL_TLB_SIZE = "IGNITE_PDS_WAL_TLB_SIZE";
-
- /** WAL flush frequency for {@link Mode#BACKGROUND} log mode */
- public static final String IGNITE_PDS_WAL_FLUSH_FREQ = "IGNITE_PDS_WAL_FLUSH_FREQUENCY";
-
/** */
- public static final String IGNITE_PDS_WAL_FSYNC_DELAY = "IGNITE_PDS_WAL_FSYNC_DELAY"; // TODO may be move to config
-
- /** Ignite pds wal record iterator buffer size. */
- public static final String IGNITE_PDS_WAL_RECORD_ITERATOR_BUFFER_SIZE = "IGNITE_PDS_WAL_RECORD_ITERATOR_BUFFER_SIZE";
+ private final boolean alwaysWriteFullPages;
- /** */
- public static final String IGNITE_PDS_WAL_ALWAYS_WRITE_FULL_PAGES = "IGNITE_PDS_WAL_ALWAYS_WRITE_FULL_PAGES";
+ /** WAL segment size in bytes */
+ private final long maxWalSegmentSize;
/** */
- private static long fsyncDelayNanos = IgniteSystemProperties.getLong(IGNITE_PDS_WAL_FSYNC_DELAY, 1);
+ private final WALMode mode;
/** Thread local byte buffer size, see {@link #tlb} */
- public final int tlbSize = IgniteSystemProperties.getInteger(IGNITE_PDS_WAL_TLB_SIZE, 128 * 1024);
-
- /** WAL flush frequency. Makes sense only for {@link Mode#BACKGROUND} log mode. */
- public static final int FLUSH_FREQ = IgniteSystemProperties.getInteger(IGNITE_PDS_WAL_FLUSH_FREQ, 2_000);
+ private final int tlbSize;
- /** */
- private final boolean alwaysWriteFullPages =
- IgniteSystemProperties.getBoolean(IGNITE_PDS_WAL_ALWAYS_WRITE_FULL_PAGES, false);
+ /** WAL flush frequency. Makes sense only for {@link WALMode#BACKGROUND} log WALMode. */
+ public final int flushFreq;
- /** WAL segment size in bytes */
- private long maxWalSegmentSize;
+ /** Fsync delay. */
+ private final long fsyncDelay;
/** */
- private final PersistentStoreConfiguration dbCfg;
+ private final PersistentStoreConfiguration psCfg;
/** */
private IgniteConfiguration igCfg;
@@ -150,8 +137,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** */
private File walArchiveDir;
- /** Current log segment handle */
- private volatile FileWriteHandle currentHnd;
+ /** */
+ private RecordSerializer serializer;
/** */
private volatile long oldestArchiveSegmentIdx;
@@ -176,19 +163,16 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
};
/** */
- private RecordSerializer serializer;
-
- /** WAL file archiver thread, started on server nodes at cluster init */
private volatile FileArchiver archiver;
/** */
- private final Mode mode;
-
- /** */
private QueueFlusher flusher;
/** */
- private ThreadLocal<WALPointer> lastWALPtr = new ThreadLocal<>();
+ private final ThreadLocal<WALPointer> lastWALPtr = new ThreadLocal<>();
+
+ /** Current log segment handle */
+ private volatile FileWriteHandle currentHnd;
/**
* @param ctx Kernal context.
@@ -196,16 +180,18 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
public FileWriteAheadLogManager(GridKernalContext ctx) {
igCfg = ctx.config();
- PersistentStoreConfiguration dbCfg = igCfg.getPersistentStoreConfiguration();
-
- assert dbCfg != null : "WAL should not be created if persistence is disabled.";
+ PersistentStoreConfiguration psCfg = igCfg.getPersistentStoreConfiguration();
- this.dbCfg = dbCfg;
+ assert psCfg != null : "WAL should not be created if persistence is disabled.";
- maxWalSegmentSize = dbCfg.getWalSegmentSize();
+ this.psCfg = psCfg;
- String modeStr = IgniteSystemProperties.getString(IGNITE_PDS_WAL_MODE);
- mode = modeStr == null ? Mode.DEFAULT : Mode.valueOf(modeStr.trim().toUpperCase());
+ maxWalSegmentSize = psCfg.getWalSegmentSize();
+ mode = psCfg.getWalMode();
+ tlbSize = psCfg.getTlbSize();
+ flushFreq = psCfg.getWalFlushFrequency();
+ fsyncDelay = psCfg.getWalFsyncDelay();
+ alwaysWriteFullPages = psCfg.isAlwaysWriteFullPages();
}
/** {@inheritDoc} */
@@ -218,9 +204,20 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
consId = U.maskForFileName(consId);
checkWalConfiguration();
- walWorkDir = initDirectory(dbCfg.getWalStorePath(), "db/wal", consId, "write ahead log work directory");
- walArchiveDir = initDirectory(dbCfg.getWalArchivePath(), "db/wal/archive", consId,
- "write ahead log archive directory");
+
+ walWorkDir = initDirectory(
+ psCfg.getWalStorePath(),
+ PersistentStoreConfiguration.DFLT_WAL_STORE_PATH,
+ consId,
+ "write ahead log work directory"
+ );
+
+ walArchiveDir = initDirectory(
+ psCfg.getWalArchivePath(),
+ PersistentStoreConfiguration.DFLT_WAL_ARCHIVE_PATH,
+ consId,
+ "write ahead log archive directory"
+ );
serializer = new RecordV1Serializer(cctx);
@@ -236,7 +233,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
archiver = new FileArchiver(tup == null ? -1 : tup.get2());
- if (mode != Mode.DEFAULT) {
+ if (mode != WALMode.DEFAULT) {
if (log.isInfoEnabled())
log.info("Started write-ahead log manager [mode=" + mode + ']');
}
@@ -247,9 +244,12 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
* @throws IgniteCheckedException if WAL store path is configured and archive path isn't (or vice versa)
*/
private void checkWalConfiguration() throws IgniteCheckedException {
- if (dbCfg.getWalStorePath() == null ^ dbCfg.getWalArchivePath() == null) {
- throw new IgniteCheckedException("Properties should be either both specified or both null " +
- "[walStorePath = " + dbCfg.getWalStorePath() + ", walArchivePath = " + dbCfg.getWalArchivePath() + "]");
+ if (psCfg.getWalStorePath() == null ^ psCfg.getWalArchivePath() == null) {
+ throw new IgniteCheckedException(
+ "Properties should be either both specified or both null " +
+ "[walStorePath = " + psCfg.getWalStorePath() +
+ ", walArchivePath = " + psCfg.getWalArchivePath() + "]"
+ );
}
}
@@ -326,7 +326,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** {@inheritDoc} */
@Override public boolean isFullSync() {
- return mode == Mode.DEFAULT;
+ return mode == WALMode.DEFAULT;
}
/** {@inheritDoc} */
@@ -348,7 +348,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
rollOver(currentHnd);
}
- if (mode == Mode.BACKGROUND) {
+ if (mode == WALMode.BACKGROUND) {
flusher = new QueueFlusher(cctx.igniteInstanceName());
flusher.start();
@@ -362,7 +362,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** {@inheritDoc} */
@SuppressWarnings("TooBroadScope")
@Override public WALPointer log(WALRecord record) throws IgniteCheckedException, StorageException {
- if (serializer == null || mode == Mode.NONE)
+ if (serializer == null || mode == WALMode.NONE)
return null;
FileWriteHandle current = currentHandle();
@@ -392,7 +392,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** {@inheritDoc} */
@Override public void fsync(WALPointer ptr) throws IgniteCheckedException, StorageException {
- if (serializer == null || mode == Mode.NONE)
+ if (serializer == null || mode == WALMode.NONE)
return;
FileWriteHandle cur = currentHandle();
@@ -405,10 +405,10 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
boolean forceFlush = filePtr != null && filePtr.forceFlush();
- if (mode == Mode.BACKGROUND && !forceFlush)
+ if (mode == WALMode.BACKGROUND && !forceFlush)
return;
- if (mode == Mode.LOG_ONLY || forceFlush) {
+ if (mode == WALMode.LOG_ONLY || forceFlush) {
cur.flushOrWait(filePtr);
return;
@@ -433,15 +433,25 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
if (hnd != null)
end = hnd.position();
- return new RecordsIterator(cctx, walWorkDir, walArchiveDir, (FileWALPointer)start, end, dbCfg, serializer,
- archiver, log, tlbSize);
+ return new RecordsIterator(
+ cctx,
+ walWorkDir,
+ walArchiveDir,
+ (FileWALPointer)start,
+ end,
+ psCfg,
+ serializer,
+ archiver,
+ log,
+ tlbSize
+ );
}
/** {@inheritDoc} */
@Override public boolean reserve(WALPointer start) throws IgniteCheckedException {
assert start != null && start instanceof FileWALPointer : "Invalid start pointer: " + start;
- if (mode == Mode.NONE)
+ if (mode == WALMode.NONE)
return false;
FileArchiver archiver0 = archiver;
@@ -464,7 +474,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
@Override public void release(WALPointer start) throws IgniteCheckedException {
assert start != null && start instanceof FileWALPointer : "Invalid start pointer: " + start;
- if (mode == Mode.NONE)
+ if (mode == WALMode.NONE)
return;
FileArchiver archiver0 = archiver;
@@ -617,7 +627,6 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
if (cfg != null) {
File workDir0 = new File(cfg);
- //TODO check path
dir = workDir0.isAbsolute() ?
new File(workDir0, consId) :
new File(U.resolveWorkDirectory(igCfg.getWorkDirectory(), cfg, false), consId);
@@ -654,7 +663,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
assert swapped : "Concurrent updates on rollover are not allowed";
- hnd.signalNextAvailable(); // let other threads to proceed with new segment
+ // Let other threads to proceed with new segment.
+ hnd.signalNextAvailable();
}
else
hnd.awaitNext();
@@ -670,7 +680,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
private FileWriteHandle restoreWriteHandle(FileWALPointer lastReadPtr) throws IgniteCheckedException {
long absIdx = lastReadPtr == null ? 0 : lastReadPtr.index();
- long segNo = absIdx % dbCfg.getWalSegments();
+ long segNo = absIdx % psCfg.getWalSegments();
File curFile = new File(walWorkDir, FileDescriptor.fileName(segNo));
@@ -784,9 +794,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
File[] allFiles = walWorkDir.listFiles(WAL_SEGMENT_FILE_FILTER);
- if (allFiles.length != 0 && allFiles.length > dbCfg.getWalSegments())
+ if (allFiles.length != 0 && allFiles.length > psCfg.getWalSegments())
throw new IgniteCheckedException("Failed to initialize wal (work directory contains " +
- "incorrect number of segments) [cur=" + allFiles.length + ", expected=" + dbCfg.getWalSegments() + ']');
+ "incorrect number of segments) [cur=" + allFiles.length + ", expected=" + psCfg.getWalSegments() + ']');
// Allocate the first segment synchronously. All other segments will be allocated by archiver in background.
if (allFiles.length == 0) {
@@ -808,9 +818,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
log.debug("Formatting file [exists=" + file.exists() + ", file=" + file.getAbsolutePath() + ']');
try (RandomAccessFile rnd = new RandomAccessFile(file, "rw")) {
- int left = dbCfg.getWalSegmentSize();
+ int left = psCfg.getWalSegmentSize();
- if (mode == Mode.DEFAULT) {
+ if (mode == WALMode.DEFAULT) {
while (left > 0) {
int toWrite = Math.min(FILL_BUF.length, left);
@@ -821,9 +831,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
rnd.getChannel().force(false);
}
- else {
+ else
rnd.setLength(0);
- }
}
catch (IOException e) {
throw new IgniteCheckedException("Failed to format WAL segment file: " + file.getAbsolutePath(), e);
@@ -868,7 +877,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
// Signal to archiver that we are done with the segment and it can be archived.
long absNextIdx = archiver.nextAbsoluteSegmentIndex(curIdx);
- long segmentIdx = absNextIdx % dbCfg.getWalSegments();
+ long segmentIdx = absNextIdx % psCfg.getWalSegments();
return new File(walWorkDir, FileDescriptor.fileName(segmentIdx));
}
@@ -896,7 +905,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
*/
private static FileDescriptor[] scan(File[] allFiles) {
if (allFiles == null)
- return new FileDescriptor[0];
+ return EMPTY_DESCRIPTORS;
FileDescriptor[] descs = new FileDescriptor[allFiles.length];
@@ -913,7 +922,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/**
* File archiver operates on absolute segment indexes. For any given absolute segment index N we can calculate
- * the work WAL segment: S(N) = N % dbCfg.walSegments.
+ * the work WAL segment: S(N) = N % psCfg.walSegments.
* When a work segment is finished, it is given to the archiver. If the absolute index of last archived segment
* is denoted by A and the absolute index of next segment we want to write is denoted by W, then we can allow
* write to S(W) if W - A <= walSegments. <br>
@@ -1119,7 +1128,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
// Notify archiver thread.
notifyAll();
- while (curAbsWalIdx - lastAbsArchivedIdx > dbCfg.getWalSegments() && cleanException == null)
+ while (curAbsWalIdx - lastAbsArchivedIdx > psCfg.getWalSegments() && cleanException == null)
wait();
return curAbsWalIdx;
@@ -1186,7 +1195,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
* @param absIdx Absolute index to archive.
*/
private File archiveSegment(long absIdx) throws IgniteCheckedException {
- long segIdx = absIdx % dbCfg.getWalSegments();
+ long segIdx = absIdx % psCfg.getWalSegments();
File origFile = new File(walWorkDir, FileDescriptor.fileName(segIdx));
@@ -1207,7 +1216,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
Files.move(dstTmpFile.toPath(), dstFile.toPath());
- if (mode == Mode.DEFAULT) {
+ if (mode == WALMode.DEFAULT) {
try (RandomAccessFile f0 = new RandomAccessFile(dstFile, "rw")) {
f0.getChannel().force(false);
}
@@ -1256,14 +1265,14 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
* @throws IgniteCheckedException if validation or create file fail.
*/
private void checkFiles(int startWith, boolean create, IgnitePredicate<Integer> p) throws IgniteCheckedException {
- for (int i = startWith; i < dbCfg.getWalSegments() && (p == null || (p != null && p.apply(i))); i++) {
+ for (int i = startWith; i < psCfg.getWalSegments() && (p == null || (p != null && p.apply(i))); i++) {
File checkFile = new File(walWorkDir, FileDescriptor.fileName(i));
if (checkFile.exists()) {
if (checkFile.isDirectory())
throw new IgniteCheckedException("Failed to initialize WAL log segment (a directory with " +
"the same name already exists): " + checkFile.getAbsolutePath());
- else if (checkFile.length() != dbCfg.getWalSegmentSize() && mode == Mode.DEFAULT)
+ else if (checkFile.length() != psCfg.getWalSegmentSize() && mode == WALMode.DEFAULT)
throw new IgniteCheckedException("Failed to initialize WAL log segment " +
"(WAL segment size change is not supported):" + checkFile.getAbsolutePath());
}
@@ -1331,10 +1340,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
int end = fileName.length() - WAL_SEGMENT_FILE_EXT.length();
- if (idx == null)
- this.idx = Long.parseLong(fileName.substring(0, end));
- else
- this.idx = idx;
+ this.idx = idx == null ? Long.parseLong(fileName.substring(0, end)) : idx;
}
/**
@@ -1426,6 +1432,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
}
}
+ /**
+ *
+ */
private static class ReadFileHandle extends FileHandle {
/** Entry serializer. */
private RecordSerializer ser;
@@ -1505,7 +1514,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** Condition activated each time writeBuffer() completes. Used to wait previously flushed write to complete */
private final Condition writeComplete = lock.newCondition();
- /** Condition for timed wait of several threads, see {@link #fsyncDelayNanos} */
+ /** Condition for timed wait of several threads, see {@link PersistentStoreConfiguration#getWalFsyncDelay()} */
private final Condition fsync = lock.newCondition();
/**
@@ -1585,8 +1594,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
idx,
(int)nextPos,
rec.size(),
- // We need to force checkpoint records into file in BACKGROUND mode.
- mode == Mode.BACKGROUND && rec instanceof CheckpointRecord);
+ // We need to force checkpoint records into file in BACKGROUND WALMode.
+ mode == WALMode.BACKGROUND && rec instanceof CheckpointRecord);
rec.position(ptr);
@@ -1819,9 +1828,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
if (!needFsync(ptr))
return;
- if (fsyncDelayNanos > 0 && !stop.get()) {
+ if (fsyncDelay > 0 && !stop.get()) {
// Delay fsync to collect as many updates as possible: trade latency for throughput.
- U.await(fsync, fsyncDelayNanos, TimeUnit.NANOSECONDS);
+ U.await(fsync, fsyncDelay, TimeUnit.NANOSECONDS);
if (!needFsync(ptr))
return;
@@ -1846,7 +1855,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
lastFsyncPos = written;
- if (fsyncDelayNanos > 0)
+ if (fsyncDelay > 0)
fsync.signalAll();
long end = metricsEnabled ? System.nanoTime() : 0;
@@ -1868,7 +1877,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
private boolean close(boolean rollOver) throws IgniteCheckedException, StorageException {
if (stop.compareAndSet(false, true)) {
// Here we can be sure that no other records will be added and this fsync will be the last.
- if (mode == Mode.DEFAULT)
+ if (mode == WALMode.DEFAULT)
fsync(null);
else
flushOrWait(null);
@@ -1880,7 +1889,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
ch.write(allocate, written);
- if (mode == Mode.DEFAULT)
+ if (mode == WALMode.DEFAULT)
ch.force(false);
}
@@ -1899,14 +1908,15 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
return false;
}
-
- /** Signals next segment available to wake up other worker threads waiting for WAL to write */
+ /**
+ * Signals next segment available to wake up other worker threads waiting for WAL to write
+ */
private void signalNextAvailable() {
lock.lock();
try {
assert head.get() instanceof FakeRecord: "head";
- assert written == lastFsyncPos || mode != Mode.DEFAULT :
+ assert written == lastFsyncPos || mode != WALMode.DEFAULT :
"fsync [written=" + written + ", lastFsync=" + lastFsyncPos + ']';
ch = null;
@@ -1959,8 +1969,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
while (written != pos) {
assert written < pos : "written = " + written + ", pos = " + pos; // No one can write further than we are now.
- // Permutation occurred between blocks write operations
- // order of acquiring lock is not the same as order of write
+ // Permutation occurred between blocks write operations.
+ // Order of acquiring lock is not the same as order of write.
long now = U.currentTimeMillis();
if (now - lastLogged >= logBackoff) {
@@ -2116,15 +2126,10 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/**
* Iterator over WAL-log.
*/
- public static class RecordsIterator extends GridCloseableIteratorAdapter<IgniteBiTuple<WALPointer, WALRecord>>
+ private static class RecordsIterator extends GridCloseableIteratorAdapter<IgniteBiTuple<WALPointer, WALRecord>>
implements WALIterator {
/** */
private static final long serialVersionUID = 0L;
-
- /** Buffer size. */
- private final int buffSize = IgniteSystemProperties.getInteger(
- IGNITE_PDS_WAL_RECORD_ITERATOR_BUFFER_SIZE, 64 * 1024 * 1024);
-
/** */
private final File walWorkDir;
@@ -2135,7 +2140,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
private final FileArchiver archiver;
/** */
- private final PersistentStoreConfiguration dbCfg;
+ private final PersistentStoreConfiguration psCfg;
/** */
private final RecordSerializer serializer;
@@ -2170,18 +2175,18 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
* @param walArchiveDir WAL archive dir.
* @param start Optional start pointer.
* @param end Optional end pointer.
- * @param dbCfg Database configuration.
+ * @param psCfg Database configuration.
* @param serializer Serializer.
* @param archiver Archiver.
* @throws IgniteCheckedException If failed to initialize WAL segment.
*/
- public RecordsIterator(
+ private RecordsIterator(
GridCacheSharedContext cctx,
File walWorkDir,
File walArchiveDir,
FileWALPointer start,
FileWALPointer end,
- PersistentStoreConfiguration dbCfg,
+ PersistentStoreConfiguration psCfg,
RecordSerializer serializer,
FileArchiver archiver,
IgniteLogger log,
@@ -2190,17 +2195,17 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
this.cctx = cctx;
this.walWorkDir = walWorkDir;
this.walArchiveDir = walArchiveDir;
- this.dbCfg = dbCfg;
+ this.psCfg = psCfg;
this.serializer = serializer;
this.archiver = archiver;
this.start = start;
this.end = end;
this.log = log;
- int buffSize0 = Math.min(16 * tlbSize, buffSize);
+ int buffSize = Math.min(16 * tlbSize, psCfg.getWalRecordIteratorBufferSize());
// Do not allocate direct buffer for iterator.
- buf = ByteBuffer.allocate(buffSize0);
+ buf = ByteBuffer.allocate(buffSize);
buf.order(ByteOrder.nativeOrder());
init();
@@ -2363,7 +2368,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
FileDescriptor.fileName(curIdx)));
}
else {
- long workIdx = curIdx % dbCfg.getWalSegments();
+ long workIdx = curIdx % psCfg.getWalSegments();
fd = new FileDescriptor(
new File(walWorkDir, FileDescriptor.fileName(workIdx)),
@@ -2477,7 +2482,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
}
}
- /** Periodically flushes current file handle for {@link Mode#BACKGROUND} mode */
+ /**
+ * Periodically flushes current file handle for {@link WALMode#BACKGROUND} WALMode.
+ */
private class QueueFlusher extends Thread {
/** */
private volatile boolean stopped;
@@ -2492,7 +2499,7 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/** {@inheritDoc} */
@Override public void run() {
while (!stopped) {
- long wakeup = U.currentTimeMillis() + FLUSH_FREQ;
+ long wakeup = U.currentTimeMillis() + flushFreq;
LockSupport.parkUntil(wakeup);
@@ -2507,7 +2514,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
}
}
- /** Signals stop, wakes up thread and waiting until completion */
+ /**
+ * Signals stop, wakes up thread and waiting until completion.
+ */
private void shutdown() {
stopped = true;
@@ -2521,18 +2530,4 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
}
}
}
-
- /**
- * WAL Mode.
- */
- private enum Mode {
- NONE, LOG_ONLY,
-
- /**
- * Write is performed periodically, initiated by background thread,
- * calls to {@link IgniteWriteAheadLogManager#fsync(org.apache.ignite.internal.pagemem.wal.WALPointer)} have no effect.
- * Using this mode will decrease persistence reliability for performance
- */
- BACKGROUND, DEFAULT
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd4d0400/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsAtomicCacheRebalancingTest.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsAtomicCacheRebalancingTest.java b/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsAtomicCacheRebalancingTest.java
new file mode 100644
index 0000000..0c4812d
--- /dev/null
+++ b/modules/pds/src/test/java/org/apache/ignite/cache/database/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.cache.database;
+
+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/cd4d0400/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsCacheRebalancingAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsCacheRebalancingAbstractTest.java b/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsCacheRebalancingAbstractTest.java
new file mode 100644
index 0000000..6c0ec19
--- /dev/null
+++ b/modules/pds/src/test/java/org/apache/ignite/cache/database/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.cache.database;
+
+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/cd4d0400/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsClientNearCachePutGetTest.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsClientNearCachePutGetTest.java b/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsClientNearCachePutGetTest.java
new file mode 100644
index 0000000..240171e
--- /dev/null
+++ b/modules/pds/src/test/java/org/apache/ignite/cache/database/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.cache.database;
+
+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/cd4d0400/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsContinuousRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsContinuousRestartTest.java b/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsContinuousRestartTest.java
new file mode 100644
index 0000000..a149836
--- /dev/null
+++ b/modules/pds/src/test/java/org/apache/ignite/cache/database/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.cache.database;
+
+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/cd4d0400/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsDynamicCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsDynamicCacheTest.java b/modules/pds/src/test/java/org/apache/ignite/cache/database/IgnitePdsDynamicCacheTest.java
new file mode 100644
index 0000000..63bf557
--- /dev/null
+++ b/modules/pds/src/test/java/org/apache/ignite/cache/database/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.cache.database;
+
+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.database.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));
+ }
+}