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 2018/01/17 10:35:25 UTC
[13/17] ignite git commit: IEP-4 Baseline topology for persistent
caches (Phase 1) Contributed by: Dmitriy Govorukhin
Dmitry Pavlov Eduard
Shangareev Ily
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index 6e4b726..3599594 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -77,6 +77,8 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.mem.DirectMemoryProvider;
+import org.apache.ignite.internal.mem.file.MappedFileMemoryProvider;
+import org.apache.ignite.internal.mem.unsafe.UnsafeMemoryProvider;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.PageMemory;
@@ -91,6 +93,7 @@ import org.apache.ignite.internal.pagemem.wal.record.CheckpointRecord;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.pagemem.wal.record.MemoryRecoveryRecord;
+import org.apache.ignite.internal.pagemem.wal.record.MetastoreDataRecord;
import org.apache.ignite.internal.pagemem.wal.record.PageSnapshot;
import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.PageDeltaRecord;
@@ -108,6 +111,8 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
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.metastorage.MetaStorage;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.CheckpointMetricsTracker;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
@@ -140,6 +145,7 @@ import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteOutClosure;
+import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.mxbean.DataStorageMetricsMXBean;
import org.apache.ignite.thread.IgniteThread;
import org.apache.ignite.thread.IgniteThreadPoolExecutor;
@@ -147,8 +153,10 @@ import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static java.nio.file.StandardOpenOption.READ;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_MAX_CHECKPOINT_MEMORY_HISTORY_SIZE;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_SKIP_CRC;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage.METASTORAGE_CACHE_ID;
/**
*
@@ -158,6 +166,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/** */
public static final String IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC = "IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC";
+ /** MemoryPolicyConfiguration name reserved for meta store. */
+ private static final String METASTORE_DATA_REGION_NAME = "metastoreMemPlc";
+
/** */
private static final long GB = 1024L * 1024 * 1024;
@@ -190,6 +201,12 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/** Checkpoint file name pattern. */
private static final Pattern CP_FILE_NAME_PATTERN = Pattern.compile("(\\d+)-(.*)-(START|END)\\.bin");
+ /** Node started file patter. */
+ private static final Pattern NODE_STARTED_FILE_NAME_PATTERN = Pattern.compile("(\\d+)-node-started\\.bin");
+
+ /** Node started file suffix. */
+ private static final String NODE_STARTED_FILE_NAME_SUFFIX = "-node-started.bin";
+
/** */
private static final FileFilter CP_FILE_FILTER = new FileFilter() {
@Override public boolean accept(File f) {
@@ -198,6 +215,13 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
};
/** */
+ private static final FileFilter NODE_STARTED_FILE_FILTER = new FileFilter() {
+ @Override public boolean accept(File f) {
+ return f.getName().endsWith(NODE_STARTED_FILE_NAME_SUFFIX);
+ }
+ };
+
+ /** */
private static final Comparator<GridDhtLocalPartition> ASC_PART_COMPARATOR = new Comparator<GridDhtLocalPartition>() {
@Override public int compare(GridDhtLocalPartition a, GridDhtLocalPartition b) {
return Integer.compare(a.id(), b.id());
@@ -296,6 +320,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
private final long lockWaitTime;
/** */
+ private final int maxCpHistMemSize;
+
+ /** */
private Map<Integer, Map<Integer, T2<Long, WALPointer>>> reservedForExchange;
/** */
@@ -316,6 +343,12 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/** Number of pages in current checkpoint. */
private volatile int currCheckpointPagesCnt;
+ /** */
+ private MetaStorage metaStorage;
+
+ /** */
+ private List<MetastorageLifecycleListener> metastorageLifecycleLsnrs;
+
/**
* @param ctx Kernal context.
*/
@@ -335,6 +368,23 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
persistenceCfg.getMetricsRateTimeInterval(),
persistenceCfg.getMetricsSubIntervalCount()
);
+
+ metastorageLifecycleLsnrs = ctx.internalSubscriptionProcessor().getMetastorageSubscribers();
+
+ maxCpHistMemSize = Math.min(persistenceCfg.getWalHistorySize(),
+ IgniteSystemProperties.getInteger(IGNITE_PDS_MAX_CHECKPOINT_MEMORY_HISTORY_SIZE, 100));
+ }
+
+ /** */
+ private void notifyMetastorageReadyForRead() throws IgniteCheckedException {
+ for (MetastorageLifecycleListener lsnr : metastorageLifecycleLsnrs)
+ lsnr.onReadyForRead(metaStorage);
+ }
+
+ /** */
+ private void notifyMetastorageReadyForReadWrite() throws IgniteCheckedException {
+ for (MetastorageLifecycleListener lsnr : metastorageLifecycleLsnrs)
+ lsnr.onReadyForReadWrite(metaStorage);
}
/**
@@ -360,6 +410,31 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/** {@inheritDoc} */
+ @Override protected void initDataRegions(DataStorageConfiguration memCfg) throws IgniteCheckedException {
+ super.initDataRegions(memCfg);
+
+ addDataRegion(
+ memCfg,
+ createDataRegionConfiguration(memCfg),
+ false
+ );
+ }
+
+ /**
+ * @param storageCfg Data storage configuration.
+ * @return Data region configuration.
+ */
+ private DataRegionConfiguration createDataRegionConfiguration(DataStorageConfiguration storageCfg) {
+ DataRegionConfiguration cfg = new DataRegionConfiguration();
+
+ cfg.setName(METASTORE_DATA_REGION_NAME);
+ cfg.setInitialSize(storageCfg.getSystemRegionInitialSize());
+ cfg.setMaxSize(storageCfg.getSystemRegionMaxSize());
+ cfg.setPersistenceEnabled(true);
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
@Override protected void start0() throws IgniteCheckedException {
super.start0();
@@ -397,6 +472,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
fileLockHolder = new FileLockHolder(storeMgr.workDir().getPath(), kernalCtx, log);
persStoreMetrics.wal(cctx.wal());
+
+ // Here we can get data from metastorage
+ readMetastore();
}
}
@@ -415,6 +493,59 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
);
}
+ /** */
+ private void readMetastore() throws IgniteCheckedException {
+ try {
+ DataStorageConfiguration memCfg = cctx.kernalContext().config().getDataStorageConfiguration();
+
+ DataRegionConfiguration plcCfg = createDataRegionConfiguration(memCfg);
+
+ File allocPath = buildAllocPath(plcCfg);
+
+ DirectMemoryProvider memProvider = allocPath == null ?
+ new UnsafeMemoryProvider(log) :
+ new MappedFileMemoryProvider(
+ log,
+ allocPath);
+
+ DataRegionMetricsImpl memMetrics = new DataRegionMetricsImpl(plcCfg);
+
+ PageMemoryEx storePageMem = (PageMemoryEx)createPageMemory(memProvider, memCfg, plcCfg, memMetrics, false);
+
+ DataRegion regCfg = new DataRegion(storePageMem, plcCfg, memMetrics, createPageEvictionTracker(plcCfg, storePageMem));
+
+ CheckpointStatus status = readCheckpointStatus();
+
+ cctx.pageStore().initializeForMetastorage();
+
+ storePageMem.start();
+
+ checkpointReadLock();
+
+ try {
+ restoreMemory(status, true, storePageMem);
+
+ metaStorage = new MetaStorage(cctx.wal(), regCfg, memMetrics, true);
+
+ metaStorage.init(this);
+
+ applyLastUpdates(status, true);
+
+ notifyMetastorageReadyForRead();
+ }
+ finally {
+ checkpointReadUnlock();
+ }
+
+ metaStorage = null;
+
+ storePageMem.stop();
+ }
+ catch (StorageException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
/**
* Get checkpoint buffer size for the given configuration.
*
@@ -552,7 +683,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/** {@inheritDoc} */
@Override public void readCheckpointAndRestoreMemory(
- List<DynamicCacheDescriptor> cachesToStart) throws IgniteCheckedException {
+ List<DynamicCacheDescriptor> cachesToStart
+ ) throws IgniteCheckedException {
+ assert !cctx.localNode().isClient();
+
checkpointReadLock();
try {
@@ -565,13 +699,29 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
CheckpointStatus status = readCheckpointStatus();
+ cctx.pageStore().initializeForMetastorage();
+
+ metaStorage = new MetaStorage(cctx.wal(), dataRegionMap.get(METASTORE_DATA_REGION_NAME),
+ (DataRegionMetricsImpl)memMetricsMap.get(METASTORE_DATA_REGION_NAME));
+
+ WALPointer restore = restoreMemory(status);
+
// First, bring memory to the last consistent checkpoint state if needed.
// This method should return a pointer to the last valid record in the WAL.
- WALPointer restore = restoreMemory(status);
cctx.wal().resumeLogging(restore);
- cctx.wal().log(new MemoryRecoveryRecord(U.currentTimeMillis()));
+ WALPointer ptr = cctx.wal().log(new MemoryRecoveryRecord(U.currentTimeMillis()));
+
+ if (ptr != null) {
+ cctx.wal().fsync(ptr);
+
+ nodeStart(ptr);
+ }
+
+ metaStorage.init(this);
+
+ notifyMetastorageReadyForReadWrite();
}
catch (StorageException e) {
throw new IgniteCheckedException(e);
@@ -581,6 +731,93 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
}
+ /**
+ * @param ptr Memory recovery wal pointer.
+ */
+ private void nodeStart(WALPointer ptr) throws IgniteCheckedException {
+ FileWALPointer p = (FileWALPointer)ptr;
+
+ String fileName = U.currentTimeMillis() + "-node-started.bin";
+
+ ByteBuffer buf = ByteBuffer.allocate(20);
+ buf.order(ByteOrder.nativeOrder());
+
+ try (FileChannel ch = FileChannel.open(
+ Paths.get(cpDir.getAbsolutePath(), fileName),
+ StandardOpenOption.CREATE_NEW, StandardOpenOption.APPEND)
+ ) {
+ buf.putLong(p.index());
+
+ buf.putInt(p.fileOffset());
+
+ buf.putInt(p.length());
+
+ buf.flip();
+
+ ch.write(buf);
+
+ buf.clear();
+
+ ch.force(true);
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /**
+ *
+ */
+ public List<T2<Long, WALPointer>> nodeStartedPointers() throws IgniteCheckedException {
+ List<T2<Long, WALPointer>> res = new ArrayList<>();
+
+ File[] files = cpDir.listFiles(NODE_STARTED_FILE_FILTER);
+
+ Arrays.sort(files, new Comparator<File>() {
+ @Override public int compare(File o1, File o2) {
+ String n1 = o1.getName();
+ String n2 = o2.getName();
+
+ Long ts1 = Long.valueOf(n1.substring(0, n1.length() - NODE_STARTED_FILE_NAME_SUFFIX.length()));
+ Long ts2 = Long.valueOf(n2.substring(0, n2.length() - NODE_STARTED_FILE_NAME_SUFFIX.length()));
+
+ if (ts1 == ts2)
+ return 0;
+ else if (ts1 < ts2)
+ return -1;
+ else
+ return 1;
+ }
+ });
+
+ ByteBuffer buf = ByteBuffer.allocate(20);
+ buf.order(ByteOrder.nativeOrder());
+
+ for (File f : files){
+ String name = f.getName();
+
+ Long ts = Long.valueOf(name.substring(0, name.length() - NODE_STARTED_FILE_NAME_SUFFIX.length()));
+
+ try (FileChannel ch = FileChannel.open(f.toPath(), READ)) {
+ ch.read(buf);
+
+ buf.flip();
+
+ FileWALPointer ptr = new FileWALPointer(
+ buf.getLong(), buf.getInt(), buf.getInt());
+
+ res.add(new T2<Long, WALPointer>(ts, ptr));
+
+ buf.clear();
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException("Failed to read node started marker file: " + f.getAbsolutePath(), e);
+ }
+ }
+
+ return res;
+ }
+
/** {@inheritDoc} */
@Override public void lock() throws IgniteCheckedException {
if (fileLockHolder != null) {
@@ -655,10 +892,11 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
DirectMemoryProvider memProvider,
DataStorageConfiguration memCfg,
DataRegionConfiguration plcCfg,
- DataRegionMetricsImpl memMetrics
+ DataRegionMetricsImpl memMetrics,
+ final boolean trackable
) {
if (!plcCfg.isPersistenceEnabled())
- return super.createPageMemory(memProvider, memCfg, plcCfg, memMetrics);
+ return super.createPageMemory(memProvider, memCfg, plcCfg, memMetrics, trackable);
memMetrics.persistenceEnabled(true);
@@ -680,6 +918,22 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
if (IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_OVERRIDE_WRITE_THROTTLING_ENABLED, false))
writeThrottlingEnabled = true;
+ GridInClosure3X<Long, FullPageId, PageMemoryEx> changeTracker;
+
+ if (trackable)
+ changeTracker = new GridInClosure3X<Long, FullPageId, PageMemoryEx>() {
+ @Override public void applyx(
+ Long page,
+ FullPageId fullId,
+ PageMemoryEx pageMem
+ ) throws IgniteCheckedException {
+ if (trackable)
+ snapshotMgr.onChangeTrackerPage(page, fullId, pageMem);
+ }
+ };
+ else
+ changeTracker = null;
+
PageMemoryImpl pageMem = new PageMemoryImpl(
memProvider,
calculateFragmentSizes(
@@ -702,15 +956,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
snapshotMgr.flushDirtyPageHandler(fullId, pageBuf, tag);
}
},
- new GridInClosure3X<Long, FullPageId, PageMemoryEx>() {
- @Override public void applyx(
- Long page,
- FullPageId fullId,
- PageMemoryEx pageMem
- ) throws IgniteCheckedException {
- snapshotMgr.onChangeTrackerPage(page, fullId, pageMem);
- }
- },
+ changeTracker,
this,
memMetrics,
writeThrottlingEnabled
@@ -1071,7 +1317,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
checkpointReadLock();
try {
- applyLastUpdates(status);
+ applyLastUpdates(status, false);
}
finally {
checkpointReadUnlock();
@@ -1082,6 +1328,11 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
checkpointer = new Checkpointer(cctx.igniteInstanceName(), "db-checkpoint-thread", log);
new IgniteThread(cctx.igniteInstanceName(), "db-checkpoint-thread", checkpointer).start();
+
+ CheckpointProgressSnapshot chp = checkpointer.wakeupForCheckpoint(0, "node started");
+
+ if (chp != null)
+ chp.cpBeginFut.get();
}
catch (StorageException e) {
throw new IgniteCheckedException(e);
@@ -1440,6 +1691,17 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
* @param status Checkpoint status.
*/
private WALPointer restoreMemory(CheckpointStatus status) throws IgniteCheckedException {
+ return restoreMemory(status, false, (PageMemoryEx)metaStorage.pageMemory());
+ }
+
+ /**
+ * @param status Checkpoint status.
+ * @param storeOnly If {@code True} restores Metastorage only.
+ */
+ private WALPointer restoreMemory(CheckpointStatus status, boolean storeOnly,
+ PageMemoryEx storePageMem) throws IgniteCheckedException {
+ assert !storeOnly || storePageMem != null;
+
if (log.isInfoEnabled())
log.info("Checking memory state [lastValidPos=" + status.endPtr + ", lastMarked="
+ status.startPtr + ", lastCheckpointId=" + status.cpStartId + ']');
@@ -1491,9 +1753,13 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
// Here we do not require tag check because we may be applying memory changes after
// several repetitive restarts and the same pages may have changed several times.
int grpId = pageRec.fullPageId().groupId();
+
+ if (storeOnly && grpId != METASTORAGE_CACHE_ID)
+ continue;
+
long pageId = pageRec.fullPageId().pageId();
- PageMemoryEx pageMem = getPageMemoryForCacheGroup(grpId);
+ PageMemoryEx pageMem = grpId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(grpId);
long page = pageMem.acquirePage(grpId, pageId, true);
@@ -1521,9 +1787,13 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
PartitionDestroyRecord destroyRec = (PartitionDestroyRecord)rec;
final int gId = destroyRec.groupId();
+
+ if (storeOnly && gId != METASTORAGE_CACHE_ID)
+ continue;
+
final int pId = destroyRec.partitionId();
- PageMemoryEx pageMem = getPageMemoryForCacheGroup(gId);
+ PageMemoryEx pageMem = gId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(gId);
pageMem.clearAsync(new P3<Integer, Long, Integer>() {
@Override public boolean apply(Integer cacheId, Long pageId, Integer tag) {
@@ -1539,9 +1809,13 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
PageDeltaRecord r = (PageDeltaRecord)rec;
int grpId = r.groupId();
+
+ if (storeOnly && grpId != METASTORAGE_CACHE_ID)
+ continue;
+
long pageId = r.pageId();
- PageMemoryEx pageMem = getPageMemoryForCacheGroup(grpId);
+ PageMemoryEx pageMem = grpId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(grpId);
// Here we do not require tag check because we may be applying memory changes after
// several repetitive restarts and the same pages may have changed several times.
@@ -1567,6 +1841,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
}
+ if (storeOnly)
+ return null;
+
if (status.needRestoreMemory()) {
if (apply)
throw new IgniteCheckedException("Failed to restore memory state (checkpoint marker is present " +
@@ -1607,16 +1884,87 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/**
+ * Apply update from some iterator and with specific filters.
+ *
+ * @param it WalIterator.
+ * @param recPredicate Wal record filter.
+ * @param entryPredicate Entry filter.
+ * @param partStates Partition to restore state.
+ */
+ public void applyUpdatesOnRecovery(
+ WALIterator it,
+ IgnitePredicate<IgniteBiTuple<WALPointer, WALRecord>> recPredicate,
+ IgnitePredicate<DataEntry> entryPredicate,
+ Map<T2<Integer, Integer>, T2<Integer, Long>> partStates
+ ) throws IgniteCheckedException {
+ while (it.hasNextX()) {
+ IgniteBiTuple<WALPointer, WALRecord> next = it.nextX();
+
+ WALRecord rec = next.get2();
+
+ if (!recPredicate.apply(next))
+ break;
+
+ switch (rec.type()) {
+ case DATA_RECORD:
+ checkpointReadLock();
+
+ try {
+ DataRecord dataRec = (DataRecord) rec;
+
+ for (DataEntry dataEntry : dataRec.writeEntries()) {
+ if (entryPredicate.apply(dataEntry)) {
+ checkpointReadLock();
+
+ try {
+ int cacheId = dataEntry.cacheId();
+
+ GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+
+ if (cacheCtx != null)
+ applyUpdate(cacheCtx, dataEntry);
+ else if (log != null)
+ log.warning("Cache (cacheId=" + cacheId + ") is not started, can't apply updates.");
+ }
+ finally {
+ checkpointReadUnlock();
+ }
+ }
+ }
+ }
+ finally {
+ checkpointReadUnlock();
+ }
+
+ break;
+
+ default:
+ // Skip other records.
+ }
+ }
+
+ checkpointReadLock();
+
+ try {
+ restorePartitionState(partStates);
+ }
+ finally {
+ checkpointReadUnlock();
+ }
+ }
+
+ /**
* @param status Last registered checkpoint status.
* @throws IgniteCheckedException If failed to apply updates.
* @throws StorageException If IO exception occurred while reading write-ahead log.
*/
- private void applyLastUpdates(CheckpointStatus status) throws IgniteCheckedException {
+ private void applyLastUpdates(CheckpointStatus status, boolean metastoreOnly) throws IgniteCheckedException {
if (log.isInfoEnabled())
log.info("Applying lost cache updates since last checkpoint record [lastMarked="
+ status.startPtr + ", lastCheckpointId=" + status.cpStartId + ']');
- cctx.kernalContext().query().skipFieldLookup(true);
+ if (!metastoreOnly)
+ cctx.kernalContext().query().skipFieldLookup(true);
long start = U.currentTimeMillis();
int applied = 0;
@@ -1631,6 +1979,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
switch (rec.type()) {
case DATA_RECORD:
+ if (metastoreOnly)
+ continue;
+
DataRecord dataRec = (DataRecord)rec;
for (DataEntry dataEntry : dataRec.writeEntries()) {
@@ -1646,6 +1997,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
break;
case PART_META_UPDATE_STATE:
+ if (metastoreOnly)
+ continue;
+
PartitionMetaStateRecord metaStateRecord = (PartitionMetaStateRecord)rec;
partStates.put(new T2<>(metaStateRecord.groupId(), metaStateRecord.partitionId()),
@@ -1653,15 +2007,52 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
break;
+ case METASTORE_DATA_RECORD:
+ MetastoreDataRecord metastoreDataRecord = (MetastoreDataRecord)rec;
+
+ metaStorage.applyUpdate(metastoreDataRecord.key(), metastoreDataRecord.value());
+
+ break;
+
+ case META_PAGE_UPDATE_NEXT_SNAPSHOT_ID:
+ case META_PAGE_UPDATE_LAST_SUCCESSFUL_SNAPSHOT_ID:
+ case META_PAGE_UPDATE_LAST_SUCCESSFUL_FULL_SNAPSHOT_ID:
+ if (metastoreOnly)
+ continue;
+
+ PageDeltaRecord rec0 = (PageDeltaRecord) rec;
+
+ PageMemoryEx pageMem = getPageMemoryForCacheGroup(rec0.groupId());
+
+ long page = pageMem.acquirePage(rec0.groupId(), rec0.pageId(), true);
+
+ try {
+ long addr = pageMem.writeLock(rec0.groupId(), rec0.pageId(), page, true);
+
+ try {
+ rec0.applyDelta(pageMem, addr);
+ }
+ finally {
+ pageMem.writeUnlock(rec0.groupId(), rec0.pageId(), page, null, true, true);
+ }
+ }
+ finally {
+ pageMem.releasePage(rec0.groupId(), rec0.pageId(), page);
+ }
+
+ break;
+
default:
// Skip other records.
}
}
- restorePartitionState(partStates);
+ if (!metastoreOnly)
+ restorePartitionState(partStates);
}
finally {
- cctx.kernalContext().query().skipFieldLookup(false);
+ if (!metastoreOnly)
+ cctx.kernalContext().query().skipFieldLookup(false);
}
if (log.isInfoEnabled())
@@ -1690,6 +2081,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
for (int i = 0; i < grp.affinity().partitions(); i++) {
+ T2<Integer, Long> restore = partStates.get(new T2<>(grpId, i));
+
if (storeMgr.exists(grpId, i)) {
storeMgr.ensure(grpId, i);
@@ -1703,41 +2096,99 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
// TODO: https://issues.apache.org/jira/browse/IGNITE-6097
grp.offheap().onPartitionInitialCounterUpdated(i, 0);
- long partMetaId = pageMem.partitionMetaPageId(grpId, i);
- long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
+ checkpointReadLock();
try {
- long pageAddr = pageMem.writeLock(grpId, partMetaId, partMetaPage);
-
- boolean changed = false;
+ long partMetaId = pageMem.partitionMetaPageId(grpId, i);
+ long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
try {
- PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.forPage(pageAddr);
+ long pageAddr = pageMem.writeLock(grpId, partMetaId, partMetaPage);
- T2<Integer, Long> fromWal = partStates.get(new T2<>(grpId, i));
+ boolean changed = false;
- if (fromWal != null) {
- int stateId = fromWal.get1();
+ try {
+ PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.forPage(pageAddr);
+
+ if (restore != null) {
+ int stateId = restore.get1();
- io.setPartitionState(pageAddr, (byte)stateId);
+ io.setPartitionState(pageAddr, (byte)stateId);
- changed = updateState(part, stateId);
+ changed = updateState(part, stateId);
- if (stateId == GridDhtPartitionState.OWNING.ordinal()) {
- grp.offheap().onPartitionInitialCounterUpdated(i, fromWal.get2());
+ if (stateId == GridDhtPartitionState.OWNING.ordinal()
+ || (stateId == GridDhtPartitionState.MOVING.ordinal()
- if (part.initialUpdateCounter() < fromWal.get2()) {
- part.initialUpdateCounter(fromWal.get2());
+ && part.initialUpdateCounter() < restore.get2())) {
+ part.initialUpdateCounter(restore.get2());
changed = true;
}
}
+ else
+ updateState(part, (int)io.getPartitionState(pageAddr));
+ }
+ finally {
+ pageMem.writeUnlock(grpId, partMetaId, partMetaPage, null, changed);
+ }
+ }
+ finally {
+ pageMem.releasePage(grpId, partMetaId, partMetaPage);
+ }
+ }
+ finally {
+ checkpointReadUnlock();
+ }
+ }
+ else if (restore != null) {
+ GridDhtLocalPartition part = grp.topology().forceCreatePartition(i);
+
+ assert part != null;
+
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-6097
+ grp.offheap().onPartitionInitialCounterUpdated(i, 0);
+
+ updateState(part, restore.get1());
+ }
+ }
+ }
+ }
+
+ /**
+ * @param grpCtx Group context.
+ * @param partId Partition ID.
+ * @return Partition state.
+ */
+ public GridDhtPartitionState readPartitionState(CacheGroupContext grpCtx, int partId) {
+ int grpId = grpCtx.groupId();
+ PageMemoryEx pageMem = (PageMemoryEx)grpCtx.dataRegion().pageMemory();
+
+ try {
+ if (storeMgr.exists(grpId, partId)) {
+ storeMgr.ensure(grpId, partId);
+
+ if (storeMgr.pages(grpId, partId) > 1) {
+ long partMetaId = pageMem.partitionMetaPageId(grpId, partId);
+ long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
+
+ try {
+ long pageAddr = pageMem.readLock(grpId, partMetaId, partMetaPage);
+
+ try {
+ if (PageIO.getType(pageAddr) == PageIO.T_PART_META) {
+ PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.forPage(pageAddr);
+
+ GridDhtPartitionState state = GridDhtPartitionState.fromOrdinal((int)io.getPartitionState(pageAddr));
+
+ if (state == null)
+ state = GridDhtPartitionState.MOVING;
+
+ return state;
}
- else
- changed = updateState(part, (int)io.getPartitionState(pageAddr));
}
finally {
- pageMem.writeUnlock(grpId, partMetaId, partMetaPage, null, changed);
+ pageMem.readUnlock(grpId, partMetaId, partMetaPage);
}
}
finally {
@@ -1746,6 +2197,21 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
}
}
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to read partition state (will default to MOVING) [grp=" + grpCtx +
+ ", partId=" + partId + "]", e);
+ }
+
+ return GridDhtPartitionState.MOVING;
+ }
+
+ /**
+ * Wal truncate callBack.
+ *
+ * @param highBound WALPointer.
+ */
+ public void onWalTruncated(WALPointer highBound) {
+ checkpointHist.onWalTruncated(highBound);
}
/**
@@ -1805,6 +2271,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
break;
+ case READ:
+ // do nothing
+ break;
+
default:
throw new IgniteCheckedException("Invalid operation for WAL entry update: " + dataEntry.op());
}
@@ -2330,8 +2800,14 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
CacheState state = new CacheState(locParts.size());
- for (GridDhtLocalPartition part : grp.topology().currentLocalPartitions())
- state.addPartitionState(part.id(), part.dataStore().fullSize(), part.updateCounter());
+ for (GridDhtLocalPartition part : grp.topology().currentLocalPartitions()) {
+ state.addPartitionState(
+ part.id(),
+ part.dataStore().fullSize(),
+ part.updateCounter(),
+ (byte)part.state().ordinal()
+ );
+ }
cpRec.addCacheGroupState(grp.groupId(), state);
}
@@ -2340,7 +2816,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
hasPages = hasPageForWrite(cpPagesTuple.get1());
- if (hasPages) {
+ if (hasPages || curr.nextSnapshot) {
// No page updates for this checkpoint are allowed from now on.
cpPtr = cctx.wal().log(cpRec);
@@ -2631,15 +3107,22 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
int grpId = fullId.groupId();
- CacheGroupContext grp = context().cache().cacheGroup(grpId);
+ PageMemoryEx pageMem;
- if (grp == null)
- continue;
+ if (grpId != MetaStorage.METASTORAGE_CACHE_ID) {
+ CacheGroupContext grp = context().cache().cacheGroup(grpId);
- if (!grp.dataRegion().config().isPersistenceEnabled())
- continue;
+ if (grp == null)
+ continue;
+
+ if (!grp.dataRegion().config().isPersistenceEnabled())
+ continue;
+
+ pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
+ }
+ else
+ pageMem = (PageMemoryEx)metaStorage.pageMemory();
- PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
Integer tag = pageMem.getForCheckpoint(
fullId, tmpWriteBuf, persStoreMetrics.metricsEnabled() ? tracker : null);
@@ -2935,15 +3418,18 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/**
- * Clears checkpoint history.
+ * Callback on truncate wal.
*/
- private void onCheckpointFinished(Checkpoint chp) {
- int deleted = 0;
+ private void onWalTruncated(WALPointer ptr) {
+ FileWALPointer highBound = (FileWALPointer)ptr;
- while (histMap.size() > persistenceCfg.getWalHistorySize()) {
- Map.Entry<Long, CheckpointEntry> entry = histMap.firstEntry();
+ List<CheckpointEntry> cpToRemove = new ArrayList<>();
- CheckpointEntry cpEntry = entry.getValue();
+ for (CheckpointEntry cpEntry : histMap.values()) {
+ FileWALPointer cpPnt = (FileWALPointer)cpEntry.checkpointMark();
+
+ if (highBound.compareTo(cpPnt) <= 0)
+ break;
if (cctx.wal().reserved(cpEntry.checkpointMark())) {
U.warn(log, "Could not clear historyMap due to WAL reservation on cpEntry " + cpEntry.cpId +
@@ -2952,27 +3438,39 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
break;
}
- File startFile = new File(cpDir.getAbsolutePath(), cpEntry.startFile());
- File endFile = new File(cpDir.getAbsolutePath(), cpEntry.endFile());
+ if (!removeCheckpointFiles(cpEntry))
+ cpToRemove.add(cpEntry);
+ }
+
+ for (CheckpointEntry cpEntry : cpToRemove)
+ histMap.remove(cpEntry.cpTs);
+ }
- boolean rmvdStart = !startFile.exists() || startFile.delete();
- boolean rmvdEnd = !endFile.exists() || endFile.delete();
+ /**
+ * Clears checkpoint history.
+ */
+ private void onCheckpointFinished(Checkpoint chp) {
+ int deleted = 0;
- boolean fail = !rmvdStart || !rmvdEnd;
+ boolean dropWal = persistenceCfg.getWalHistorySize() != Integer.MAX_VALUE;
- if (fail) {
- U.warn(log, "Failed to remove stale checkpoint files [startFile=" + startFile.getAbsolutePath() +
- ", endFile=" + endFile.getAbsolutePath() + ']');
+ while (histMap.size() > maxCpHistMemSize) {
+ Map.Entry<Long, CheckpointEntry> entry = histMap.firstEntry();
- if (histMap.size() > 2 * persistenceCfg.getWalHistorySize()) {
- U.error(log, "Too many stale checkpoint entries in the map, will truncate WAL archive anyway.");
+ CheckpointEntry cpEntry = entry.getValue();
- fail = false;
- }
+ if (cctx.wal().reserved(cpEntry.checkpointMark())) {
+ U.warn(log, "Could not clear historyMap due to WAL reservation on cpEntry " + cpEntry.cpId +
+ ", history map size is " + histMap.size());
+
+ break;
}
+ boolean fail = removeCheckpointFiles(cpEntry);
+
if (!fail) {
- deleted += cctx.wal().truncate(cpEntry.checkpointMark());
+ if (dropWal)
+ deleted += cctx.wal().truncate(null, cpEntry.checkpointMark());
histMap.remove(entry.getKey());
}
@@ -2987,6 +3485,33 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/**
+ * @param cpEntry Checkpoint entry.
+ * @return {True} if delete fail.
+ */
+ private boolean removeCheckpointFiles(CheckpointEntry cpEntry) {
+ File startFile = new File(cpDir.getAbsolutePath(), cpEntry.startFile());
+ File endFile = new File(cpDir.getAbsolutePath(), cpEntry.endFile());
+
+ boolean rmvdStart = !startFile.exists() || startFile.delete();
+ boolean rmvdEnd = !endFile.exists() || endFile.delete();
+
+ boolean fail = !rmvdStart || !rmvdEnd;
+
+ if (fail) {
+ U.warn(log, "Failed to remove stale checkpoint files [startFile=" + startFile.getAbsolutePath() +
+ ", endFile=" + endFile.getAbsolutePath() + ']');
+
+ if (histMap.size() > 2 * maxCpHistMemSize) {
+ U.error(log, "Too many stale checkpoint entries in the map, will truncate WAL archive anyway.");
+
+ fail = false;
+ }
+ }
+
+ return fail;
+ }
+
+ /**
* @param cacheId Cache ID.
* @param partId Partition ID.
* @return Reserved counter or null if couldn't reserve.
@@ -3360,4 +3885,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
public DataStorageMetricsImpl persistentStoreMetricsImpl() {
return persStoreMetrics;
}
-}
\ No newline at end of file
+
+ /** {@inheritDoc} */
+ @Override public MetaStorage metaStorage() {
+ return metaStorage;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index 5ccbfb9..f8fd86c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -30,6 +30,7 @@ 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.PageSupport;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.WALPointer;
@@ -49,8 +50,7 @@ import org.apache.ignite.internal.processors.cache.IgniteRebalanceIterator;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.cache.persistence.partstate.PagesAllocationRange;
@@ -67,7 +67,6 @@ import org.apache.ignite.internal.processors.cache.tree.CacheDataTree;
import org.apache.ignite.internal.processors.cache.tree.PendingEntriesTree;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
-import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.T2;
@@ -76,12 +75,16 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.MOVING;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.OWNING;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.RENTING;
+
/**
* Used when persistence enabled.
*/
public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl implements DbCheckpointListener {
/** */
- private MetaStore metaStore;
+ private IndexStorage indexStorage;
/** */
private ReuseListImpl reuseList;
@@ -101,7 +104,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
RootPage metastoreRoot = metas.treeRoot;
- metaStore = new MetadataStorage(grp.dataRegion().pageMemory(),
+ indexStorage = new IndexStorageImpl(grp.dataRegion().pageMemory(),
ctx.wal(),
globalRemoveId(),
grp.groupId(),
@@ -122,7 +125,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
try {
final String name = "PendingEntries";
- RootPage pendingRootPage = metaStore.getOrAllocateForTree(name);
+ RootPage pendingRootPage = indexStorage.getOrAllocateForTree(name);
pendingEntries = new PendingEntriesTree(
grp,
@@ -169,8 +172,12 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
* @param store Store to save metadata.
* @throws IgniteCheckedException If failed.
*/
- private boolean saveStoreMetadata(CacheDataStore store, Context ctx, boolean saveMeta,
- boolean beforeDestroy) throws IgniteCheckedException {
+ private boolean saveStoreMetadata(
+ CacheDataStore store,
+ Context ctx,
+ boolean saveMeta,
+ boolean beforeDestroy
+ ) throws IgniteCheckedException {
RowStore rowStore0 = store.rowStore();
boolean needSnapshot = ctx != null && ctx.nextSnapshot() && ctx.needToSnapshot(grp.cacheOrGroupName());
@@ -178,7 +185,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
boolean wasSaveToMeta = false;
if (rowStore0 != null) {
- FreeListImpl freeList = (FreeListImpl)rowStore0.freeList();
+ CacheFreeListImpl freeList = (CacheFreeListImpl)rowStore0.freeList();
freeList.saveMetadata();
@@ -192,13 +199,15 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (size > 0 || updCntr > 0) {
GridDhtPartitionState state = null;
+ // localPartition will not acquire writeLock here because create=false.
+ GridDhtLocalPartition part = null;
+
if (!grp.isLocal()) {
if (beforeDestroy)
state = GridDhtPartitionState.EVICTED;
else {
- // localPartition will not acquire writeLock here because create=false.
- GridDhtLocalPartition part = grp.topology().localPartition(store.partId(),
- AffinityTopologyVersion.NONE, false, true);
+ part = grp.topology().localPartition(store.partId(),
+ AffinityTopologyVersion.NONE, false, true);
if (part != null && part.state() != GridDhtPartitionState.EVICTED)
state = part.state();
@@ -241,63 +250,21 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
long cntrsPageId;
if (grp.sharedGroup()) {
- cntrsPageId = io.getCountersPageId(partMetaPageAddr);
-
- byte[] data = serializeCacheSizes(store.cacheSizes());
-
- int items = data.length / 12;
- int written = 0;
- int pageSize = pageMem.pageSize();
-
- boolean init = cntrsPageId == 0;
-
- if (init && items > 0) {
- cntrsPageId = pageMem.allocatePage(grpId, store.partId(), PageIdAllocator.FLAG_DATA);
-
- io.setCountersPageId(partMetaPageAddr, cntrsPageId);
-
- changed = true;
- }
-
- long nextId = cntrsPageId;
-
- while (written != items) {
- final long curId = nextId;
- final long curPage = pageMem.acquirePage(grpId, curId);
-
- try {
- final long curAddr = pageMem.writeLock(grpId, curId, curPage);
-
- assert curAddr != 0;
-
- try {
- PagePartitionCountersIO partMetaIo;
+ long initCntrPageId = io.getCountersPageId(partMetaPageAddr);
- if (init) {
- partMetaIo = PagePartitionCountersIO.VERSIONS.latest();
+ Map<Integer, Long> newSizes = store.cacheSizes();
+ Map<Integer, Long> prevSizes = readSharedGroupCacheSizes(pageMem, grpId, initCntrPageId);
- partMetaIo.initNewPage(curAddr, curId, pageSize);
- }
- else
- partMetaIo = PageIO.getPageIO(curAddr);
+ if (prevSizes != null && prevSizes.equals(newSizes))
+ cntrsPageId = initCntrPageId; // Preventing modification of sizes pages for store
+ else {
+ cntrsPageId = writeSharedGroupCacheSizes(pageMem, grpId, initCntrPageId,
+ store.partId(), newSizes);
- written += partMetaIo.writeCacheSizes(pageSize, curAddr, data, written);
+ if (initCntrPageId == 0 && cntrsPageId != 0) {
+ io.setCountersPageId(partMetaPageAddr, cntrsPageId);
- nextId = partMetaIo.getNextCountersPageId(curAddr);
-
- if(written != items && (init = nextId == 0)) {
- //allocate new counters page
- nextId = pageMem.allocatePage(grpId, store.partId(), PageIdAllocator.FLAG_DATA);
- partMetaIo.setNextCountersPageId(curAddr, nextId);
- }
- }
- finally {
- // Write full page
- pageMem.writeUnlock(grpId, curId, curPage, Boolean.TRUE, true);
- }
- }
- finally {
- pageMem.releasePage(grpId, curId, curPage);
+ changed = true;
}
}
}
@@ -311,49 +278,35 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
io.setCandidatePageCount(partMetaPageAddr, pageCnt);
if (saveMeta) {
- long metaPageId = pageMem.metaPageId(grpId);
- long metaPage = pageMem.acquirePage(grpId, metaPageId);
-
- try {
- long metaPageAddr = pageMem.writeLock(grpId, metaPageId, metaPage);
-
- try {
- PageMetaIO metaIo = PageMetaIO.getPageIO(metaPageAddr);
-
- long nextSnapshotTag = metaIo.getNextSnapshotTag(metaPageAddr);
-
- metaIo.setNextSnapshotTag(metaPageAddr, nextSnapshotTag + 1);
-
- if (log != null && log.isDebugEnabled())
- log.debug("Save next snapshot before checkpoint start for grId = " + grpId
- + ", nextSnapshotTag = " + nextSnapshotTag);
-
- if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, metaPageId,
- metaPage, wal, null))
- wal.log(new MetaPageUpdateNextSnapshotId(grpId, metaPageId,
- nextSnapshotTag + 1));
-
- if (state == GridDhtPartitionState.OWNING)
- addPartition(ctx.partitionStatMap(), metaPageAddr, metaIo, grpId, PageIdAllocator.INDEX_PARTITION,
- this.ctx.kernalContext().cache().context().pageStore().pages(grpId, PageIdAllocator.INDEX_PARTITION));
- }
- finally {
- pageMem.writeUnlock(grpId, metaPageId, metaPage, null, true);
- }
- }
- finally {
- pageMem.releasePage(grpId, metaPageId, metaPage);
- }
+ saveMeta(ctx);
wasSaveToMeta = true;
}
- GridDhtPartitionMap partMap = grp.topology().localPartitionMap();
-
- if (partMap.containsKey(store.partId()) &&
- partMap.get(store.partId()) == GridDhtPartitionState.OWNING)
- addPartition(ctx.partitionStatMap(), partMetaPageAddr, io, grpId, store.partId(),
- this.ctx.pageStore().pages(grpId, store.partId()));
+ if (state == OWNING) {
+ assert part != null;
+
+ if(!addPartition(
+ part,
+ ctx.partitionStatMap(),
+ partMetaPageAddr,
+ io,
+ grpId,
+ store.partId(),
+ this.ctx.pageStore().pages(grpId, store.partId()),
+ store.fullSize()
+ ))
+ U.warn(log,"Partition was concurrently evicted grpId=" + grpId +
+ ", partitionId=" + part.id());
+ }
+ else if (state == MOVING || state == RENTING) {
+ if (ctx.partitionStatMap().forceSkipIndexPartition(grpId)) {
+ if (log.isInfoEnabled())
+ log.info("Will not include SQL indexes to snapshot because there is " +
+ "a partition not in " + OWNING + " state [grp=" + grp.cacheOrGroupName() +
+ ", partId=" + store.partId() + ", state=" + state + ']');
+ }
+ }
changed = true;
}
@@ -386,47 +339,211 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/**
- * @param cacheSizes Cache sizes.
- * @return Serialized cache sizes
+ * Loads cache sizes for all caches in shared group.
+ *
+ * @param pageMem page memory to perform operations on pages.
+ * @param grpId Cache group ID.
+ * @param cntrsPageId Counters page ID, if zero is provided that means no counters page exist.
+ * @return Cache sizes if store belongs to group containing multiple caches and sizes are available in memory. May
+ * return null if counter page does not exist.
+ * @throws IgniteCheckedException If page memory operation failed.
+ */
+ @Nullable private static Map<Integer, Long> readSharedGroupCacheSizes(PageSupport pageMem, int grpId,
+ long cntrsPageId) throws IgniteCheckedException {
+
+ if (cntrsPageId == 0L)
+ return null;
+
+ Map<Integer, Long> cacheSizes = new HashMap<>();
+
+ long nextId = cntrsPageId;
+
+ while (true){
+ final long curId = nextId;
+ final long curPage = pageMem.acquirePage(grpId, curId);
+
+ try {
+ final long curAddr = pageMem.readLock(grpId, curId, curPage);
+
+ assert curAddr != 0;
+
+ try {
+ PagePartitionCountersIO cntrsIO = PageIO.getPageIO(curAddr);
+
+ if (cntrsIO.readCacheSizes(curAddr, cacheSizes))
+ break;
+
+ nextId = cntrsIO.getNextCountersPageId(curAddr);
+
+ assert nextId != 0;
+ }
+ finally {
+ pageMem.readUnlock(grpId, curId, curPage);
+ }
+ }
+ finally {
+ pageMem.releasePage(grpId, curId, curPage);
+ }
+ }
+ return cacheSizes;
+ }
+
+ /**
+ * Saves cache sizes for all caches in shared group. Unconditionally marks pages as dirty.
+ *
+ * @param pageMem page memory to perform operations on pages.
+ * @param grpId Cache group ID.
+ * @param cntrsPageId Counters page ID, if zero is provided that means no counters page exist.
+ * @param partId Partition ID.
+ * @param sizes Cache sizes of all caches in group. Not null.
+ * @return new counter page Id. Same as {@code cntrsPageId} or new value if cache size pages were initialized.
+ * @throws IgniteCheckedException if page memory operation failed.
*/
- private byte[] serializeCacheSizes(Map<Integer, Long> cacheSizes) {
- // Item size = 4 bytes (cache ID) + 8 bytes (cache size) = 12 bytes
- byte[] data = new byte[cacheSizes.size() * 12];
- long off = GridUnsafe.BYTE_ARR_OFF;
+ private static long writeSharedGroupCacheSizes(PageMemory pageMem, int grpId,
+ long cntrsPageId, int partId, Map<Integer, Long> sizes) throws IgniteCheckedException {
+ byte[] data = PagePartitionCountersIO.VERSIONS.latest().serializeCacheSizes(sizes);
+
+ int items = data.length / PagePartitionCountersIO.ITEM_SIZE;
+ boolean init = cntrsPageId == 0;
+
+ if (init && !sizes.isEmpty())
+ cntrsPageId = pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
+
+ long nextId = cntrsPageId;
+ int written = 0;
+
+ while (written != items) {
+ final long curId = nextId;
+ final long curPage = pageMem.acquirePage(grpId, curId);
+
+ try {
+ final long curAddr = pageMem.writeLock(grpId, curId, curPage);
+
+ int pageSize = pageMem.pageSize();
+
+ assert curAddr != 0;
+
+ try {
+ PagePartitionCountersIO partCntrIo;
+
+ if (init) {
+ partCntrIo = PagePartitionCountersIO.VERSIONS.latest();
+
+ partCntrIo.initNewPage(curAddr, curId, pageSize);
+ }
+ else
+ partCntrIo = PageIO.getPageIO(curAddr);
- for (Map.Entry<Integer, Long> entry : cacheSizes.entrySet()) {
- GridUnsafe.putInt(data, off, entry.getKey()); off += 4;
- GridUnsafe.putLong(data, off, entry.getValue()); off += 8;
+ written += partCntrIo.writeCacheSizes(pageSize, curAddr, data, written);
+
+ nextId = partCntrIo.getNextCountersPageId(curAddr);
+
+ if (written != items && (init = nextId == 0)) {
+ //allocate new counters page
+ nextId = pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
+ partCntrIo.setNextCountersPageId(curAddr, nextId);
+ }
+ }
+ finally {
+ // Write full page
+ pageMem.writeUnlock(grpId, curId, curPage, Boolean.TRUE, true);
+ }
+ }
+ finally {
+ pageMem.releasePage(grpId, curId, curPage);
+ }
}
- return data;
+ return cntrsPageId;
}
/**
+ * @param ctx Context.
+ */
+ private void saveMeta(Context ctx) throws IgniteCheckedException {
+ int grpId = grp.groupId();
+ PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
+ IgniteWriteAheadLogManager wal = this.ctx.wal();
+
+ long metaPageId = pageMem.metaPageId(grpId);
+ long metaPage = pageMem.acquirePage(grpId, metaPageId);
+
+ try {
+ long metaPageAddr = pageMem.writeLock(grpId, metaPageId, metaPage);
+
+ try {
+ PageMetaIO metaIo = PageMetaIO.getPageIO(metaPageAddr);
+
+ long nextSnapshotTag = metaIo.getNextSnapshotTag(metaPageAddr);
+
+ metaIo.setNextSnapshotTag(metaPageAddr, nextSnapshotTag + 1);
+
+ if (log != null && log.isDebugEnabled())
+ log.debug("Save next snapshot before checkpoint start for grId = " + grpId
+ + ", nextSnapshotTag = " + nextSnapshotTag);
+
+ if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, metaPageId,
+ metaPage, wal, null))
+ wal.log(new MetaPageUpdateNextSnapshotId(grpId, metaPageId,
+ nextSnapshotTag + 1));
+
+ addPartition(
+ null,
+ ctx.partitionStatMap(),
+ metaPageAddr,
+ metaIo,
+ grpId,
+ PageIdAllocator.INDEX_PARTITION,
+ this.ctx.pageStore().pages(grpId, PageIdAllocator.INDEX_PARTITION),
+ -1);
+ }
+ finally {
+ pageMem.writeUnlock(grpId, metaPageId, metaPage, null, true);
+ }
+ }
+ finally {
+ pageMem.releasePage(grpId, metaPageId, metaPage);
+ }
+ }
+
+ /**
+ * @param part
* @param map Map to add values to.
* @param metaPageAddr Meta page address
* @param io Page Meta IO
* @param cacheId Cache ID.
- * @param partId Partition ID. Or {@link PageIdAllocator#INDEX_PARTITION} for index partition
* @param currAllocatedPageCnt total number of pages allocated for partition <code>[partition, cacheId]</code>
*/
- private static void addPartition(
- final PartitionAllocationMap map,
- final long metaPageAddr,
- final PageMetaIO io,
- final int cacheId,
- final int partId,
- final int currAllocatedPageCnt
+ private static boolean addPartition(
+ GridDhtLocalPartition part,
+ final PartitionAllocationMap map,
+ final long metaPageAddr,
+ final PageMetaIO io,
+ final int cacheId,
+ final int partId,
+ final int currAllocatedPageCnt,
+ final int partSize
) {
- if (currAllocatedPageCnt <= 1)
- return;
+ if (part != null) {
+ boolean reserved = part.reserve();
+
+ if(!reserved)
+ return false;
+ }
+ else
+ assert partId == PageIdAllocator.INDEX_PARTITION : partId;
assert PageIO.getPageId(metaPageAddr) != 0;
int lastAllocatedPageCnt = io.getLastAllocatedPageCount(metaPageAddr);
+
+ int curPageCnt = lastAllocatedPageCnt == 0 && partSize == 0 ? 0 : currAllocatedPageCnt;
+
map.put(
new GroupPartitionId(cacheId, partId),
- new PagesAllocationRange(lastAllocatedPageCnt, currAllocatedPageCnt));
+ new PagesAllocationRange(lastAllocatedPageCnt, curPageCnt));
+
+ return true;
}
/** {@inheritDoc} */
@@ -485,7 +602,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (grp.sharedGroup())
idxName = Integer.toString(cacheId) + "_" + idxName;
- return metaStore.getOrAllocateForTree(idxName);
+ return indexStorage.getOrAllocateForTree(idxName);
}
/** {@inheritDoc} */
@@ -493,7 +610,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (grp.sharedGroup())
idxName = Integer.toString(cacheId) + "_" + idxName;
- metaStore.dropRootPage(idxName);
+ indexStorage.dropRootPage(idxName);
}
/** {@inheritDoc} */
@@ -613,7 +730,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
for (CacheDataStore store : partDataStores.values()) {
assert store instanceof GridCacheDataStore;
- FreeListImpl freeList = ((GridCacheDataStore)store).freeList;
+ CacheFreeListImpl freeList = ((GridCacheDataStore)store).freeList;
if (freeList == null)
continue;
@@ -863,7 +980,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
private String name;
/** */
- private volatile FreeListImpl freeList;
+ private volatile CacheFreeListImpl freeList;
/** */
private volatile CacheDataStore delegate;
@@ -913,7 +1030,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
RootPage reuseRoot = metas.reuseListRoot;
- freeList = new FreeListImpl(
+ freeList = new CacheFreeListImpl(
grp.groupId(),
grp.cacheOrGroupName() + "-" + partId,
grp.dataRegion().memoryMetrics(),
@@ -964,43 +1081,8 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
Map<Integer, Long> cacheSizes = null;
- if (grp.sharedGroup()) {
- long cntrsPageId = io.getCountersPageId(pageAddr);
-
- if (cntrsPageId != 0L) {
- cacheSizes = new HashMap<>();
-
- long nextId = cntrsPageId;
-
- while (true){
- final long curId = nextId;
- final long curPage = pageMem.acquirePage(grpId, curId);
-
- try {
- final long curAddr = pageMem.readLock(grpId, curId, curPage);
-
- assert curAddr != 0;
-
- try {
- PagePartitionCountersIO cntrsIO = PageIO.getPageIO(curAddr);
-
- if (cntrsIO.readCacheSizes(curAddr, cacheSizes))
- break;
-
- nextId = cntrsIO.getNextCountersPageId(curAddr);
-
- assert nextId != 0;
- }
- finally {
- pageMem.readUnlock(grpId, curId, curPage);
- }
- }
- finally {
- pageMem.releasePage(grpId, curId, curPage);
- }
- }
- }
- }
+ if (grp.sharedGroup())
+ cacheSizes = readSharedGroupCacheSizes(pageMem, grpId, io.getCountersPageId(pageAddr));
delegate0.init(io.getSize(pageAddr), io.getUpdateCounter(pageAddr), cacheSizes);
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
index 8771f6a..1260147 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
@@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import javax.management.InstanceNotFoundException;
import org.apache.ignite.DataRegionMetrics;
import org.apache.ignite.DataStorageMetrics;
import org.apache.ignite.IgniteCheckedException;
@@ -51,8 +52,9 @@ import org.apache.ignite.internal.processors.cache.persistence.evict.PageEvictio
import org.apache.ignite.internal.processors.cache.persistence.evict.Random2LruPageEvictionTracker;
import org.apache.ignite.internal.processors.cache.persistence.evict.RandomLruPageEvictionTracker;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport;
import org.apache.ignite.internal.util.typedef.F;
@@ -83,7 +85,10 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
private static final long MAX_PAGE_MEMORY_INIT_SIZE_32_BIT = 2L * 1024 * 1024 * 1024;
/** */
- protected Map<String, DataRegion> dataRegionMap;
+ protected volatile Map<String, DataRegion> dataRegionMap;
+
+ /** */
+ private volatile boolean dataRegionsInitialized;
/** */
protected Map<String, DataRegionMetrics> memMetricsMap;
@@ -92,10 +97,10 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
protected DataRegion dfltDataRegion;
/** */
- protected Map<String, FreeListImpl> freeListMap;
+ protected Map<String, CacheFreeListImpl> freeListMap;
/** */
- private FreeListImpl dfltFreeList;
+ private CacheFreeListImpl dfltFreeList;
/** Page size from memory configuration, may be set only for fake(standalone) IgniteCacheDataBaseSharedManager */
private int pageSize;
@@ -112,6 +117,8 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
validateConfiguration(memCfg);
pageSize = memCfg.getPageSize();
+
+ initDataRegions(memCfg);
}
/**
@@ -172,7 +179,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
boolean persistenceEnabled = memPlcCfg.isPersistenceEnabled();
- FreeListImpl freeList = new FreeListImpl(0,
+ CacheFreeListImpl freeList = new CacheFreeListImpl(0,
cctx.igniteInstanceName(),
memMetrics,
memPlc,
@@ -210,21 +217,25 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
* @throws IgniteCheckedException If failed to initialize swap path.
*/
protected void initDataRegions(DataStorageConfiguration memCfg) throws IgniteCheckedException {
+ if (dataRegionsInitialized)
+ return;
+
DataRegionConfiguration[] dataRegionCfgs = memCfg.getDataRegionConfigurations();
int dataRegions = dataRegionCfgs == null ? 0 : dataRegionCfgs.length;
- dataRegionMap = U.newHashMap(2 + dataRegions);
- memMetricsMap = U.newHashMap(2 + dataRegions);
+ dataRegionMap = U.newHashMap(3 + dataRegions);
+ memMetricsMap = U.newHashMap(3 + dataRegions);
if (dataRegionCfgs != null) {
for (DataRegionConfiguration dataRegionCfg : dataRegionCfgs)
- addDataRegion(memCfg, dataRegionCfg);
+ addDataRegion(memCfg, dataRegionCfg, dataRegionCfg.isPersistenceEnabled());
}
addDataRegion(
memCfg,
- memCfg.getDefaultDataRegionConfiguration()
+ memCfg.getDefaultDataRegionConfiguration(),
+ memCfg.getDefaultDataRegionConfiguration().isPersistenceEnabled()
);
addDataRegion(
@@ -233,8 +244,11 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
memCfg.getSystemRegionInitialSize(),
memCfg.getSystemRegionMaxSize(),
CU.isPersistenceEnabled(memCfg)
- )
+ ),
+ CU.isPersistenceEnabled(memCfg)
);
+
+ dataRegionsInitialized = true;
}
/**
@@ -242,9 +256,10 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
* @param dataRegionCfg Data region config.
* @throws IgniteCheckedException If failed to initialize swap path.
*/
- private void addDataRegion(
+ protected void addDataRegion(
DataStorageConfiguration dataStorageCfg,
- DataRegionConfiguration dataRegionCfg
+ DataRegionConfiguration dataRegionCfg,
+ boolean trackable
) throws IgniteCheckedException {
String dataRegionName = dataRegionCfg.getName();
@@ -255,7 +270,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
DataRegionMetricsImpl memMetrics = new DataRegionMetricsImpl(dataRegionCfg, fillFactorProvider(dataRegionCfg));
- DataRegion memPlc = initMemory(dataStorageCfg, dataRegionCfg, memMetrics);
+ DataRegion memPlc = initMemory(dataStorageCfg, dataRegionCfg, memMetrics, trackable);
dataRegionMap.put(dataRegionName, memPlc);
@@ -278,11 +293,11 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
final String dataRegName = dataRegCfg.getName();
return new IgniteOutClosure<Float>() {
- private FreeListImpl freeList;
+ private CacheFreeListImpl freeList;
@Override public Float apply() {
if (freeList == null) {
- FreeListImpl freeList0 = freeListMap.get(dataRegName);
+ CacheFreeListImpl freeList0 = freeListMap.get(dataRegName);
if (freeList0 == null)
return (float) 0;
@@ -526,7 +541,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
*/
public void dumpStatistics(IgniteLogger log) {
if (freeListMap != null) {
- for (FreeListImpl freeList : freeListMap.values())
+ for (CacheFreeListImpl freeList : freeListMap.values())
freeList.dumpStatistics(log);
}
}
@@ -641,6 +656,8 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
dataRegionMap.clear();
dataRegionMap = null;
+
+ dataRegionsInitialized = false;
}
}
@@ -661,6 +678,9 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
"DataRegionMetrics", name
));
}
+ catch (InstanceNotFoundException ignored) {
+ // We tried to unregister a non-existing MBean, not a big deal.
+ }
catch (Throwable e) {
U.error(log, "Failed to unregister MBean for memory metrics: " +
name, e);
@@ -798,7 +818,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
int sysPageSize = pageMem.systemPageSize();
- FreeListImpl freeListImpl = freeListMap.get(plcCfg.getName());
+ CacheFreeListImpl freeListImpl = freeListMap.get(plcCfg.getName());
for (;;) {
long allocatedPagesCnt = pageMem.loadedPages();
@@ -826,7 +846,8 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
private DataRegion initMemory(
DataStorageConfiguration memCfg,
DataRegionConfiguration plcCfg,
- DataRegionMetricsImpl memMetrics
+ DataRegionMetricsImpl memMetrics,
+ boolean trackable
) throws IgniteCheckedException {
File allocPath = buildAllocPath(plcCfg);
@@ -836,7 +857,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
log,
allocPath);
- PageMemory pageMem = createPageMemory(memProvider, memCfg, plcCfg, memMetrics);
+ PageMemory pageMem = createPageMemory(memProvider, memCfg, plcCfg, memMetrics, trackable);
return new DataRegion(pageMem, plcCfg, memMetrics, createPageEvictionTracker(plcCfg, pageMem));
}
@@ -845,7 +866,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
* @param plc data region Configuration.
* @param pageMem Page memory.
*/
- private PageEvictionTracker createPageEvictionTracker(DataRegionConfiguration plc, PageMemory pageMem) {
+ protected PageEvictionTracker createPageEvictionTracker(DataRegionConfiguration plc, PageMemory pageMem) {
if (plc.getPageEvictionMode() == DataPageEvictionMode.DISABLED || plc.isPersistenceEnabled())
return new NoOpPageEvictionTracker();
@@ -873,7 +894,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
*
* @throws IgniteCheckedException If resolving swap directory fails.
*/
- @Nullable private File buildAllocPath(DataRegionConfiguration plc) throws IgniteCheckedException {
+ @Nullable protected File buildAllocPath(DataRegionConfiguration plc) throws IgniteCheckedException {
String path = plc.getSwapPath();
if (path == null)
@@ -901,7 +922,8 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
DirectMemoryProvider memProvider,
DataStorageConfiguration memCfg,
DataRegionConfiguration memPlcCfg,
- DataRegionMetricsImpl memMetrics
+ DataRegionMetricsImpl memMetrics,
+ boolean trackable
) {
memMetrics.persistenceEnabled(false);
@@ -969,4 +991,11 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
protected void setPageSize(int pageSize) {
this.pageSize = pageSize;
}
+
+ /**
+ * @return MetaStorage
+ */
+ public MetaStorage metaStorage() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
new file mode 100644
index 0000000..5141b04
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
@@ -0,0 +1,51 @@
+/*
+ * 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.IgniteCheckedException;
+
+/**
+ * Meta store.
+ */
+public interface IndexStorage {
+ /**
+ * Get or allocate initial page for an index.
+ *
+ * @param idxName Index name.
+ * @return {@link RootPage} that keeps pageId, allocated flag that shows whether the page
+ * was newly allocated, and rootId that is counter which increments each time new page allocated.
+ * @throws IgniteCheckedException If failed.
+ */
+ public RootPage getOrAllocateForTree(String idxName) throws IgniteCheckedException;
+
+ /**
+ * Deallocate index page and remove from tree.
+ *
+ * @param idxName Index name.
+ * @return Root ID or -1 if no page was removed.
+ * @throws IgniteCheckedException If failed.
+ */
+ public RootPage dropRootPage(String idxName) throws IgniteCheckedException;
+
+ /**
+ * Destroy this meta store.
+ *
+ * @throws IgniteCheckedException If failed.
+ */
+ public void destroy() throws IgniteCheckedException;
+}