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/19 08:54:32 UTC
ignite git commit: IGNITE-7471 Use soft reference for checkpoint
entry contents to avoid excessive memory usage
Repository: ignite
Updated Branches:
refs/heads/master 5efba7d6a -> 90d8bbf5a
IGNITE-7471 Use soft reference for checkpoint entry contents to avoid excessive memory usage
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/90d8bbf5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/90d8bbf5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/90d8bbf5
Branch: refs/heads/master
Commit: 90d8bbf5a636b81877051b69367c76cce1b43402
Parents: 5efba7d
Author: Dmitriy Govorukhin <dm...@gmail.com>
Authored: Thu Jan 18 17:14:26 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Jan 19 11:54:15 2018 +0300
----------------------------------------------------------------------
.../GridCachePartitionExchangeManager.java | 5 +-
.../GridCacheDatabaseSharedManager.java | 544 ++++++++++++++-----
.../wal/IgniteWalHistoryReservationsTest.java | 13 +-
3 files changed, 422 insertions(+), 140 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/90d8bbf5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 7107125..ef97238 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -959,8 +959,11 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
*/
private void refreshPartitions() {
// TODO https://issues.apache.org/jira/browse/IGNITE-6857
- if (cctx.snapshot().snapshotOperationInProgress())
+ if (cctx.snapshot().snapshotOperationInProgress()) {
+ scheduleResendPartitions();
+
return;
+ }
ClusterNode oldest = cctx.discovery().oldestAliveServerNode(AffinityTopologyVersion.NONE);
http://git-wip-us.apache.org/repos/asf/ignite/blob/90d8bbf5/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 1b0eb6d..e56b352 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
@@ -21,6 +21,7 @@ import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.RandomAccessFile;
+import java.lang.ref.SoftReference;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.FileChannel;
@@ -41,6 +42,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -323,10 +325,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
private final int maxCpHistMemSize;
/** */
- private Map<Integer, Map<Integer, T2<Long, WALPointer>>> reservedForExchange;
+ private Map</*grpId*/Integer, Map</*partId*/Integer, T2</*updCntr*/Long, WALPointer>>> reservedForExchange;
/** */
- private final ConcurrentMap<T2<Integer, Integer>, T2<Long, WALPointer>> reservedForPreloading = new ConcurrentHashMap<>();
+ private final ConcurrentMap<T2</*grpId*/Integer, /*partId*/Integer>, T2</*updCntr*/Long, WALPointer>> reservedForPreloading = new ConcurrentHashMap<>();
/** Snapshot manager. */
private IgniteCacheSnapshotManager snapshotMgr;
@@ -1345,47 +1347,78 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
reservedForExchange = new HashMap<>();
- for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
- if (grp.isLocal())
- continue;
+ Map</*grpId*/Integer, Set</*partId*/Integer>> parts4CheckpointHistSearch = partsForCheckpointHistorySearch();
- for (GridDhtLocalPartition part : grp.topology().currentLocalPartitions()) {
- if (part.state() != GridDhtPartitionState.OWNING || part.dataStore().fullSize() <= walRebalanceThreshold)
- continue;
+ Map</*grpId*/Integer, Map</*partId*/Integer, CheckpointEntry>> lastCheckpointEntry4Grp =
+ searchLastCheckpointEntryPerPartition(parts4CheckpointHistSearch);
- CheckpointEntry cpEntry = searchCheckpointEntry(grp.groupId(), part.id(), null);
+ Map</*grpId*/Integer, Map</*partId*/Integer, /*updCntr*/Long>> grpPartsWithCnts = new HashMap<>();
- try {
- if (cpEntry != null && cctx.wal().reserve(cpEntry.cpMark)) {
- Map<Integer, T2<Long, WALPointer>> cacheMap = reservedForExchange.get(grp.groupId());
+ try {
+ for (Map.Entry<Integer, Map<Integer, CheckpointEntry>> e : lastCheckpointEntry4Grp.entrySet()) {
+ Integer grpId = e.getKey();
+
+ for (Map.Entry<Integer, CheckpointEntry> e0 : e.getValue().entrySet()) {
+ CheckpointEntry cpEntry = e0.getValue();
+
+ Integer partId = e0.getKey();
+
+ if (cctx.wal().reserve(cpEntry.cpMark)) {
+ Map</*partId*/Integer, T2</*updCntr*/Long, WALPointer>> grpChpState = reservedForExchange.get(grpId);
- if (cacheMap == null) {
- cacheMap = new HashMap<>();
+ Map<Integer, Long> grpCnts = grpPartsWithCnts.get(grpId);
- reservedForExchange.put(grp.groupId(), cacheMap);
+ if (grpChpState == null) {
+ reservedForExchange.put(grpId, grpChpState = new HashMap<>());
+
+ grpPartsWithCnts.put(grpId, grpCnts = new HashMap<>());
}
- cacheMap.put(part.id(), new T2<>(cpEntry.partitionCounter(grp.groupId(), part.id()), cpEntry.cpMark));
+ Long partCnt = cpEntry.partitionCounter(cctx, grpId, partId);
+
+ if (partCnt != null) {
+ grpChpState.put(partId, new T2<>(partCnt, cpEntry.cpMark));
+
+ grpCnts.put(partId, partCnt);
+ }
+ else
+ cctx.wal().release(cpEntry.cpMark);
}
}
- catch (IgniteCheckedException ex) {
- U.error(log, "Error while trying to reserve history", ex);
- }
}
}
+ catch (IgniteCheckedException ex) {
+ U.error(log, "Error while trying to reserve history", ex);
+ }
- Map<Integer, Map<Integer, Long>> resMap = new HashMap<>();
+ return grpPartsWithCnts;
+ }
- for (Map.Entry<Integer, Map<Integer, T2<Long, WALPointer>>> e : reservedForExchange.entrySet()) {
- Map<Integer, Long> cacheMap = new HashMap<>();
+ /**
+ *
+ * @return Map of group id -> Set parts.
+ */
+ private Map<Integer, Set<Integer>> partsForCheckpointHistorySearch() {
+ Map<Integer, Set<Integer>> part4CheckpointHistSearch = new HashMap<>();
- for (Map.Entry<Integer, T2<Long, WALPointer>> e0 : e.getValue().entrySet())
- cacheMap.put(e0.getKey(), e0.getValue().get1());
+ for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
+ if (grp.isLocal())
+ continue;
- resMap.put(e.getKey(), cacheMap);
+ for (GridDhtLocalPartition part : grp.topology().currentLocalPartitions()) {
+ if (part.state() != GridDhtPartitionState.OWNING || part.dataStore().fullSize() <= walRebalanceThreshold)
+ continue;
+
+ Set<Integer> parts = part4CheckpointHistSearch.get(grp.groupId());
+
+ if (parts == null)
+ part4CheckpointHistSearch.put(grp.groupId(), parts = new HashSet<>());
+
+ parts.add(part.id());
+ }
}
- return resMap;
+ return part4CheckpointHistSearch;
}
/** {@inheritDoc} */
@@ -1497,6 +1530,67 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/**
* Tries to search for a WAL pointer for the given partition counter start.
*
+ * @return Checkpoint entry or {@code null} if failed to search.
+ */
+ private Map<Integer, Map<Integer, CheckpointEntry>> searchLastCheckpointEntryPerPartition(
+ final Map<Integer, Set<Integer>> part4reserve
+ ) {
+ final Map<Integer, Map<Integer, CheckpointEntry>> res = new HashMap<>();
+
+ for (Long cpTs : checkpointHist.checkpoints()) {
+ try {
+ final CheckpointEntry chpEntry = checkpointHist.entry(cpTs);
+
+ Map<Integer, CheckpointEntry.GroupState> grpsState = chpEntry.groupState(cctx);
+
+ if (grpsState.isEmpty()){
+ res.clear();
+
+ continue;
+ }
+
+ for (Map.Entry<Integer, Set<Integer>> grps : part4reserve.entrySet()) {
+ Integer grpId = grps.getKey();
+
+ Map<Integer, CheckpointEntry> partToCheckPntEntry = res.get(grpId);
+
+ CheckpointEntry.GroupState grpState = grpsState.get(grpId);
+
+ if (grpState == null) {
+ res.remove(grpId);
+
+ continue;
+ }
+
+ if (partToCheckPntEntry == null)
+ res.put(grpId, partToCheckPntEntry = new HashMap<>());
+
+ for (Integer partId : grps.getValue()) {
+ int idx = grpState.indexByPartition(partId);
+
+ if (idx < 0)
+ partToCheckPntEntry.remove(partId);
+ else {
+ if (partToCheckPntEntry.containsKey(partId))
+ continue;
+
+ partToCheckPntEntry.put(partId, chpEntry);
+ }
+ }
+ }
+ }
+ catch (IgniteCheckedException ignore) {
+ // Treat exception the same way as a gap.
+ res.clear();
+ }
+ }
+
+ return res;
+ }
+
+ /**
+ * Tries to search for a WAL pointer for the given partition counter start.
+ *
* @param grpId Cache group ID.
* @param part Partition ID.
* @param partCntrSince Partition counter or {@code null} to search for minimal counter.
@@ -1527,7 +1621,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
try {
CheckpointEntry entry = checkpointHist.entry(cpTs);
- Long foundCntr = entry.partitionCounter(grpId, part);
+ Long foundCntr = entry.partitionCounter(cctx, grpId, part);
if (foundCntr != null) {
if (partCntrSince == null) {
@@ -1682,8 +1776,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
return new FileWALPointer(buf.getLong(), buf.getInt(), buf.getInt());
}
catch (IOException e) {
- throw new IgniteCheckedException("Failed to read checkpoint pointer from marker file: " +
- cpMarkerFile.getAbsolutePath(), e);
+ throw new IgniteCheckedException(
+ "Failed to read checkpoint pointer from marker file: " + cpMarkerFile.getAbsolutePath(), e);
}
}
@@ -2403,8 +2497,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
if (!skipSync)
ch.force(true);
- return type == CheckpointEntryType.START ?
- new CheckpointEntry(cpTs, ptr, cpId, rec.cacheGroupStates()) : null;
+ return createCheckPointEntry(cpTs, ptr, cpId, rec, type);
}
catch (IOException e) {
throw new IgniteCheckedException(e);
@@ -2445,6 +2538,41 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/**
+ * @param cpTs Checkpoint timestamp.
+ * @param ptr Wal pointer of checkpoint.
+ * @param cpId Checkpoint ID.
+ * @param rec Checkpoint record.
+ * @param type Checkpoint type.
+ *
+ * @return Checkpoint entry.
+ */
+ private CheckpointEntry createCheckPointEntry(
+ long cpTs,
+ WALPointer ptr,
+ UUID cpId,
+ @Nullable CheckpointRecord rec,
+ CheckpointEntryType type
+ ) {
+ assert cpTs > 0;
+ assert ptr != null;
+ assert cpId != null;
+ assert type != null;
+
+ if (type != CheckpointEntryType.START)
+ return null;
+
+ CheckpointEntry entry;
+
+ Map<Integer, CacheState> cacheGrpStates = null;
+
+ // Create lazy checkpoint entry.
+ if ((checkpointHist.histMap.size() + 1 < maxCpHistMemSize) && rec != null)
+ cacheGrpStates = rec.cacheGroupStates();
+
+ return new CheckpointEntry(cpTs, ptr, cpId, cacheGrpStates);
+ }
+
+ /**
*
*/
@SuppressWarnings("NakedNotify")
@@ -3215,8 +3343,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
@NotNull GridMultiCollectionWrapper<FullPageId> cpPages,
CheckpointProgress progress
) {
- assert cpEntry == null || cpEntry.initGuard != 0;
-
this.cpEntry = cpEntry;
this.cpPages = cpPages;
this.progress = progress;
@@ -3378,13 +3504,14 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
if (type == CheckpointEntryType.START) {
long cpTs = Long.parseLong(matcher.group(1));
+ UUID cpId = UUID.fromString(matcher.group(2));
+
WALPointer ptr = readPointer(file, buf);
if (ptr == null)
continue;
- // Create lazy checkpoint entry.
- CheckpointEntry entry = new CheckpointEntry(cpTs, ptr);
+ CheckpointEntry entry = createCheckPointEntry(cpTs, ptr, cpId, null, type);
histMap.put(cpTs, entry);
}
@@ -3404,8 +3531,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
if (entry == null)
throw new IgniteCheckedException("Checkpoint entry was removed: " + cpTs);
- entry.initIfNeeded(cctx);
-
return entry;
}
@@ -3519,102 +3644,44 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
return fail;
}
-
- /**
- * @param cacheId Cache ID.
- * @param partId Partition ID.
- * @return Reserved counter or null if couldn't reserve.
- */
- @Nullable private Long reserve(int cacheId, int partId) {
- for (CheckpointEntry entry : histMap.values()) {
- try {
- entry.initIfNeeded(cctx);
-
- if (entry.cacheGrpStates == null)
- continue;
-
- CacheState grpState = entry.cacheGrpStates.get(cacheId);
-
- if (grpState == null)
- continue;
-
- long partCntr = grpState.counterByPartition(partId);
-
- if (partCntr >= 0) {
- if (cctx.wal().reserve(entry.checkpointMark()))
- return partCntr;
- }
- }
- catch (Exception e) {
- U.error(log, "Error while trying to reserve history", e);
- }
- }
-
- return null;
- }
}
/**
- *
+ * Checkpoint entry.
*/
private static class CheckpointEntry {
- /** */
- private static final AtomicIntegerFieldUpdater<CheckpointEntry> initGuardUpdater =
- AtomicIntegerFieldUpdater.newUpdater(CheckpointEntry.class, "initGuard");
-
/** Checkpoint timestamp. */
private long cpTs;
/** Checkpoint end mark. */
private WALPointer cpMark;
- /** Initialization latch. */
- private CountDownLatch initLatch;
-
- /** */
- @SuppressWarnings("unused")
- private volatile int initGuard;
-
- /** Checkpoint ID. Initialized lazily. */
+ /** Checkpoint ID. */
private UUID cpId;
- /** Cache states. Initialized lazily. */
- private Map<Integer, CacheState> cacheGrpStates;
-
- /** Initialization exception. */
- private IgniteCheckedException initEx;
+ /** */
+ private volatile SoftReference<GroupStateLazyStore> grpStateLazyStore;
/**
- * Lazy entry constructor.
+ * Checkpoint entry constructor.
*
- * @param cpTs Checkpoint timestamp.
- * @param cpMark Checkpoint end mark (WAL pointer).
- */
- private CheckpointEntry(long cpTs, WALPointer cpMark) {
- assert cpMark != null;
-
- this.cpTs = cpTs;
- this.cpMark = cpMark;
-
- initLatch = new CountDownLatch(1);
- }
-
- /**
- * Creates complete entry.
+ * If {@code grpStates} is null then it will be inited lazy from wal pointer.
*
* @param cpTs Checkpoint timestamp.
* @param cpMark Checkpoint mark pointer.
* @param cpId Checkpoint ID.
* @param cacheGrpStates Cache groups states.
*/
- private CheckpointEntry(long cpTs, WALPointer cpMark, UUID cpId, Map<Integer, CacheState> cacheGrpStates) {
+ private CheckpointEntry(
+ long cpTs,
+ WALPointer cpMark,
+ UUID cpId,
+ @Nullable Map<Integer, CacheState> cacheGrpStates
+ ) {
this.cpTs = cpTs;
this.cpMark = cpMark;
this.cpId = cpId;
- this.cacheGrpStates = cacheGrpStates;
-
- initGuard = 1;
- initLatch = new CountDownLatch(0);
+ this.grpStateLazyStore = new SoftReference<>(new GroupStateLazyStore(cacheGrpStates));
}
/**
@@ -3653,57 +3720,260 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
/**
+ * @param cctx Cache shred context.
+ */
+ public Map<Integer, GroupState> groupState(
+ GridCacheSharedContext cctx
+ ) throws IgniteCheckedException {
+ GroupStateLazyStore store = initIfNeeded(cctx);
+
+ return store.grpStates;
+ }
+
+ /**
+ * @param cctx Cache shred context.
+ * @return Group lazy store.
+ */
+ private GroupStateLazyStore initIfNeeded(GridCacheSharedContext cctx) throws IgniteCheckedException {
+ GroupStateLazyStore store = grpStateLazyStore.get();
+
+ if (store == null) {
+ store = new GroupStateLazyStore();
+
+ grpStateLazyStore = new SoftReference<>(store);
+ }
+
+ store.initIfNeeded(cctx, cpMark);
+
+ return store;
+ }
+
+ /**
+ * @param cctx Cache shared context.
* @param grpId Cache group ID.
* @param part Partition ID.
* @return Partition counter or {@code null} if not found.
*/
- private Long partitionCounter(int grpId, int part) {
- assert initGuard != 0;
+ private Long partitionCounter(GridCacheSharedContext cctx, int grpId, int part) {
+ GroupStateLazyStore store;
- if (initEx != null || cacheGrpStates == null)
+ try {
+ store = initIfNeeded(cctx);
+ }
+ catch (IgniteCheckedException e) {
return null;
+ }
- CacheState state = cacheGrpStates.get(grpId);
+ return store.partitionCounter(grpId, part);
+ }
- if (state != null) {
- long cntr = state.counterByPartition(part);
+ /**
+ *
+ */
+ private static class GroupState {
+ /** */
+ private int[] parts;
+
+ /** */
+ private long[] cnts;
+
+ /** */
+ private int idx;
+
+ /**
+ * @param partsCnt Partitions count.
+ */
+ private GroupState(int partsCnt) {
+ parts = new int[partsCnt];
+ cnts = new long[partsCnt];
+ }
+
+ /**
+ * @param partId Partition ID to add.
+ * @param cntr Partition counter.
+ */
+ public void addPartitionCounter(int partId, long cntr) {
+ if (idx == parts.length)
+ throw new IllegalStateException("Failed to add new partition to the partitions state " +
+ "(no enough space reserved) [partId=" + partId + ", reserved=" + parts.length + ']');
+
+ if (idx > 0) {
+ if (parts[idx - 1] >= partId)
+ throw new IllegalStateException("Adding partition in a wrong order [prev=" + parts[idx - 1] +
+ ", cur=" + partId + ']');
+ }
- return cntr < 0 ? null : cntr;
+ parts[idx] = partId;
+
+ cnts[idx] = cntr;
+
+ idx++;
}
- return null;
+ /**
+ * Gets partition counter by partition ID.
+ *
+ * @param partId Partition ID.
+ * @return Partition update counter (will return {@code -1} if partition is not present in the record).
+ */
+ public long counterByPartition(int partId) {
+ int idx = indexByPartition(partId);
+
+ return idx >= 0 ? cnts[idx] : 0;
+ }
+
+ public long size(){
+ return idx;
+ }
+
+ /**
+ * @param partId Partition ID to search.
+ * @return Non-negative index of partition if found or negative value if not found.
+ */
+ private int indexByPartition(int partId) {
+ return Arrays.binarySearch(parts, 0, idx, partId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "GroupState [cap=" + parts.length + ", size=" + idx + ']';
+ }
}
/**
- * @throws IgniteCheckedException If failed to read WAL entry.
+ * Group state lazy store.
*/
- private void initIfNeeded(GridCacheSharedContext cctx) throws IgniteCheckedException {
- if (initGuardUpdater.compareAndSet(this, 0, 1)) {
- try (WALIterator it = cctx.wal().replay(cpMark)) {
- if (it.hasNextX()) {
- IgniteBiTuple<WALPointer, WALRecord> tup = it.nextX();
+ private static class GroupStateLazyStore {
+ /** */
+ private static final AtomicIntegerFieldUpdater<GroupStateLazyStore> initGuardUpdater =
+ AtomicIntegerFieldUpdater.newUpdater(GroupStateLazyStore.class, "initGuard");
- CheckpointRecord rec = (CheckpointRecord)tup.get2();
+ /** Cache states. Initialized lazily. */
+ private Map<Integer, GroupState> grpStates;
- cpId = rec.checkpointId();
- cacheGrpStates = rec.cacheGroupStates();
- }
- else
- initEx = new IgniteCheckedException("Failed to find checkpoint record at " +
- "the given WAL pointer: " + cpMark);
- }
- catch (IgniteCheckedException e) {
- initEx = e;
+ /** */
+ private volatile CountDownLatch latch;
+
+ /** */
+ @SuppressWarnings("unused")
+ private volatile int initGuard;
+
+ /** Initialization exception. */
+ private IgniteCheckedException initEx;
+
+ /**
+ * Default constructor.
+ */
+ private GroupStateLazyStore() {
+ this(null);
+ }
+
+ /**
+ * @param cacheGrpStates Cache group state.
+ */
+ private GroupStateLazyStore(Map<Integer, CacheState> cacheGrpStates) {
+ CountDownLatch latch;
+
+ if (cacheGrpStates != null) {
+ initGuard = 1;
+
+ this.latch = new CountDownLatch(0);
}
- finally {
- initLatch.countDown();
+ else
+ this.latch = new CountDownLatch(1);
+
+ this.grpStates = remap(cacheGrpStates);
+ }
+
+ /**
+ * @param stateRec Cache group state.
+ */
+ private Map<Integer, GroupState> remap(Map<Integer, CacheState> stateRec) {
+ if (stateRec == null)
+ return null;
+
+ Map<Integer, GroupState> grpStates = new HashMap<>(stateRec.size());
+
+ for (Integer grpId : stateRec.keySet()) {
+ CacheState recState = stateRec.get(grpId);
+
+ GroupState groupState = new GroupState(recState.size());
+
+ for (int i = 0; i < recState.size(); i++) {
+ groupState.addPartitionCounter(
+ recState.partitionByIndex(i),
+ recState.partitionCounterByIndex(i)
+ );
+ }
+
+ grpStates.put(grpId, groupState);
}
+
+ return grpStates;
}
- else {
- U.await(initLatch);
+
+ /**
+ * @param grpId Group id.
+ * @param part Partition id.
+ * @return Partition counter.
+ */
+ private Long partitionCounter(int grpId, int part) {
+ assert initGuard != 0 : initGuard;
if (initEx != null)
- throw initEx;
+ return null;
+
+ GroupState state = grpStates.get(grpId);
+
+ if (state != null) {
+ long cntr = state.counterByPartition(part);
+
+ return cntr < 0 ? null : cntr;
+ }
+
+ return null;
+ }
+
+ /**
+ * @param cctx Cache shared context.
+ * @param ptr Checkpoint wal pointer.
+ * @throws IgniteCheckedException If failed to read WAL entry.
+ */
+ private void initIfNeeded(
+ GridCacheSharedContext cctx,
+ WALPointer ptr
+ ) throws IgniteCheckedException {
+ if (initGuardUpdater.compareAndSet(this, 0, 1)) {
+ try (WALIterator it = cctx.wal().replay(ptr)) {
+ if (it.hasNextX()) {
+ IgniteBiTuple<WALPointer, WALRecord> tup = it.nextX();
+
+ CheckpointRecord rec = (CheckpointRecord)tup.get2();
+
+ Map<Integer, CacheState> stateRec = rec.cacheGroupStates();
+
+ if (stateRec != null)
+ this.grpStates = remap(stateRec);
+ else
+ grpStates = Collections.emptyMap();
+ }
+ else
+ initEx = new IgniteCheckedException(
+ "Failed to find checkpoint record at the given WAL pointer: " + ptr);
+ }
+ catch (IgniteCheckedException e) {
+ initEx = e;
+ }
+ finally {
+ latch.countDown();
+ }
+ }
+ else {
+ U.await(latch);
+
+ if (initEx != null)
+ throw initEx;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/90d8bbf5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalHistoryReservationsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalHistoryReservationsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalHistoryReservationsTest.java
index c6d58e5..66a8aa9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalHistoryReservationsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalHistoryReservationsTest.java
@@ -40,6 +40,7 @@ import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Assert;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
@@ -57,9 +58,13 @@ public class IgniteWalHistoryReservationsTest extends GridCommonAbstractTest {
cfg.setClientMode(client);
+ cfg.setConsistentId("NODE$" + gridName.charAt(gridName.length() - 1));
+
DataStorageConfiguration memCfg = new DataStorageConfiguration()
.setDefaultDataRegionConfiguration(
- new DataRegionConfiguration().setMaxSize(200 * 1024 * 1024).setPersistenceEnabled(true))
+ new DataRegionConfiguration()
+ .setMaxSize(200 * 1024 * 1024)
+ .setPersistenceEnabled(true))
.setWalMode(WALMode.LOG_ONLY);
cfg.setDataStorageConfiguration(memCfg);
@@ -104,10 +109,14 @@ public class IgniteWalHistoryReservationsTest extends GridCommonAbstractTest {
final int entryCnt = 10_000;
final int initGridCnt = 4;
- final IgniteEx ig0 = (IgniteEx)startGrids(initGridCnt);
+ final IgniteEx ig0 = (IgniteEx)startGrids(initGridCnt + 1);
ig0.active(true);
+ stopGrid(initGridCnt);
+
+ Assert.assertEquals(5, ig0.context().state().clusterState().baselineTopology().consistentIds().size());
+
long start = U.currentTimeMillis();
log.warning("Start loading");