You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/05/23 13:07:40 UTC
[17/18] ignite git commit: Merge remote-tracking branch
'remotes/community/ignite-gg-12163' into ignite-5075-pds
Merge remote-tracking branch 'remotes/community/ignite-gg-12163' into ignite-5075-pds
# Conflicts:
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
# modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
# modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/pagemem/PageMemoryImpl.java
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/926686ec
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/926686ec
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/926686ec
Branch: refs/heads/ignite-5075-pds
Commit: 926686ec4be325837c46a020ba7f992e8ee0be9e
Parents: 5e83866 7058c71
Author: sboikov <sb...@gridgain.com>
Authored: Tue May 23 16:07:00 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue May 23 16:07:00 2017 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 2 +-
.../StartSnapshotOperationDiscoveryMessage.java | 4 +-
.../processors/cache/GridCacheProcessor.java | 8 +-
.../distributed/dht/GridDhtLocalPartition.java | 5 +-
.../dht/atomic/GridDhtAtomicCache.java | 4 +-
.../IgniteDhtPartitionHistorySuppliersMap.java | 3 +
.../GridCacheDatabaseSharedManager.java | 83 +--
.../database/file/FilePageStoreManager.java | 14 +-
.../cache/database/pagemem/PageMemoryImpl.java | 509 +++++++++----------
.../database/wal/FileWriteAheadLogManager.java | 11 +-
...istentStoreCacheRebalancingAbstractTest.java | 71 ++-
...entStoreRecoveryAfterFileCorruptionTest.java | 59 ++-
.../db/IgniteDbPageEvictionSelfTest.java | 14 +-
.../db/IgniteDbWholeClusterRestartSelfTest.java | 28 +-
.../IgniteWalHistoryReservationsSelfTest.java | 23 +-
.../PageStoreCheckpointSimulationSelfTest.java | 2 +-
.../db/file/PageStoreEvictionSelfTest.java | 2 +-
.../file/WalRecoveryTxLogicalRecordsTest.java | 34 +-
18 files changed, 461 insertions(+), 415 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index e1fc404,76ea847..fcddf6b
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@@ -1220,12 -1166,11 +1220,8 @@@ public class GridCacheProcessor extend
cacheCtx.onStarted();
- if (log.isInfoEnabled()) {
- log.info("Started cache [name=" + cfg.getName() +
- (cfg.getGroupName() != null ? ", group=" + cfg.getGroupName() : "") +
- if (log.isInfoEnabled())
- log.info("Started cache [name=" + U.maskName(cfg.getName()) +
-- ", memoryPolicyName=" + cfg.getMemoryPolicyName() +
- ", mode=" + cfg.getCacheMode() + ']');
- }
- ", mode=" + cfg.getCacheMode() + ']'
- );
++ if (log.isInfoEnabled()){
++ log.info("Started cache [name=" + cfg.getName() +(cfg.getGroupName() != null ? ", group=" + cfg.getGroupName() : "") + ", memoryPolicyName=" + cfg.getMemoryPolicyName() + ", mode=" + cfg.getCacheMode() + ']');}
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index 07c6136,a35c168..b5659b4
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@@ -171,27 -141,17 +171,26 @@@ public class GridDhtLocalPartition exte
* @param id Partition ID.
* @param entryFactory Entry factory.
*/
- @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
- GridDhtLocalPartition(GridCacheSharedContext ctx,
- CacheGroupInfrastructure grp, int id, GridCacheMapEntryFactory entryFactory) {
- @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor") GridDhtLocalPartition(
- GridCacheContext cctx,
- int id,
- GridCacheMapEntryFactory entryFactory
- ) {
- super(cctx, entryFactory, Math.max(10, GridCacheAdapter.DFLT_START_CACHE_SIZE / cctx.affinity().partitions()));
++ @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor") GridDhtLocalPartition(GridCacheSharedContext ctx,
++ CacheGroupInfrastructure grp,int id, GridCacheMapEntryFactory entryFactory) {
+ super( entryFactory);
this.id = id;
- this.cctx = cctx;
+ this.ctx = ctx;
+ this.grp = grp;
- log = U.logger(cctx.kernalContext(), logRef, this);
+ log = U.logger(ctx.kernalContext(), logRef, this);
+
+ if (grp.sharedGroup()) {
+ singleCacheEntryMap = null;
+ cachesEntryMaps = new ConcurrentHashMap<>();
+ cacheSizes = new ConcurrentHashMap<>();
+ }
+ else {
+ singleCacheEntryMap = createEntriesMap();
+ cachesEntryMaps = null;
+ cacheSizes = null;
+ }
rent = new GridFutureAdapter<Object>() {
@Override public String toString() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --cc modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/GridCacheDatabaseSharedManager.java
index 6635e2d,c1fed9f..1aada23
--- 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
@@@ -879,40 -875,31 +879,31 @@@ public class GridCacheDatabaseSharedMan
reservedForExchange = new HashMap<>();
- for (GridCacheContext cacheCtx : (Collection<GridCacheContext>)cctx.cacheContexts()) {
- if (cacheCtx.isLocal())
+ for (CacheGroupInfrastructure grp : cctx.cache().cacheGroups()) {
+ if (grp.isLocal())
continue;
- for (GridDhtLocalPartition part : cacheCtx.topology().currentLocalPartitions()) {
- if (part.state() != GridDhtPartitionState.OWNING || part.dataStore().size() <= ggWalRebalanceThreshold)
+ for (GridDhtLocalPartition part : grp.topology().currentLocalPartitions()) {
+ if (part.state() != GridDhtPartitionState.OWNING || part.dataStore().fullSize() <= ggWalRebalanceThreshold)
continue;
- for (Long cpTs : checkpointHist.checkpoints()) {
- try {
- CheckpointEntry entry = checkpointHist.entry(cpTs);
-
- if (!entry.cacheGrpStates.containsKey(grp.groupId()) ||
- !entry.cacheGrpStates.get(grp.groupId()).partitions().containsKey(part.id()))
- continue;
-
- WALPointer ptr = searchPartitionCounter(grp.groupId(), part.id(), entry.checkpointTimestamp());
-
- if (ptr != null && cctx.wal().reserve(ptr)) {
- Map<Integer, T2<Long, WALPointer>> cacheMap = reservedForExchange.get(grp.groupId());
+ CheckpointEntry cpEntry = searchCheckpointEntry(cacheCtx, part.id(), null);
- if (cacheMap == null) {
- cacheMap = new HashMap<>();
+ try {
+ if (cpEntry != null && cctx.wal().reserve(cpEntry.cpMark)) {
+ Map<Integer, T2<Long, WALPointer>> cacheMap = reservedForExchange.get(cacheCtx.cacheId());
- reservedForExchange.put(grp.groupId(), cacheMap);
- }
+ if (cacheMap == null) {
+ cacheMap = new HashMap<>();
- cacheMap.put(part.id(), new T2<>(entry.partitionCounter(grp.groupId(), part.id()), ptr));
- reservedForExchange.put(cacheCtx.cacheId(), cacheMap);
++ reservedForExchange.put(grp.groupId(), cacheMap);
}
- }
- catch (IgniteCheckedException ex) {
- U.error(log, "Error while trying to reserve history", ex);
- }
- cacheMap.put(part.id(), new T2<>(cpEntry.partitionCounter(cacheCtx.cacheId(), part.id()), cpEntry.cpMark));
++ cacheMap.put(part.id(), new T2<>(cpEntry.partitionCounter(grp.groupId(), part.id()), cpEntry.cpMark));
+ }
+ }
+ catch (IgniteCheckedException ex) {
+ U.error(log, "Error while trying to reserve history", ex);
}
}
}
@@@ -951,8 -938,13 +942,13 @@@
}
/** {@inheritDoc} */
- @Override public boolean reserveHistoryForPreloading(int cacheId, int partId, long cntr) {
- CheckpointEntry cpEntry = searchCheckpointEntry(cctx.cacheContext(cacheId), partId, cntr);
+ @Override public boolean reserveHistoryForPreloading(int grpId, int partId, long cntr) {
- WALPointer ptr = searchPartitionCounter(grpId, partId, cntr);
++ CheckpointEntry cpEntry = searchCheckpointEntry(grpId, partId, cntr);
+
+ if (cpEntry == null)
+ return false;
+
+ WALPointer ptr = cpEntry.cpMark;
if (ptr == null)
return false;
@@@ -1052,24 -1057,51 +1048,51 @@@
/**
* Tries to search for a WAL pointer for the given partition counter start.
*
- * @param cacheCtx Cache context.
+ * @param grpId Cache group ID.
* @param part Partition ID.
- * @param partCntrSince Partition counter.
- * @return WAL pointer or {@code null} if failed to search.
+ * @param partCntrSince Partition counter or {@code null} to search for minimal counter.
+ * @return Checkpoint entry or {@code null} if failed to search.
*/
- WALPointer searchPartitionCounter(int grpId, int part, Long partCntrSince) {
+ @Nullable public WALPointer searchPartitionCounter(GridCacheContext cacheCtx, int part, @Nullable Long partCntrSince) {
+ CheckpointEntry entry = searchCheckpointEntry(cacheCtx, part, partCntrSince);
+
+ if (entry == null)
+ return null;
+
+ return entry.cpMark;
+ }
+
+ /**
+ * Tries to search for a WAL pointer for the given partition counter start.
+ *
+ * @param cacheCtx Cache context.
+ * @param part Partition ID.
+ * @param partCntrSince Partition counter or {@code null} to search for minimal counter.
+ * @return Checkpoint entry or {@code null} if failed to search.
+ */
+ @Nullable private CheckpointEntry searchCheckpointEntry(GridCacheContext cacheCtx, int part, @Nullable Long partCntrSince) {
boolean hasGap = false;
- WALPointer first = null;
+ CheckpointEntry first = null;
for (Long cpTs : checkpointHist.checkpoints()) {
try {
CheckpointEntry entry = checkpointHist.entry(cpTs);
- Long foundCntr = entry.partitionCounter(cacheCtx.cacheId(), part);
+ Long foundCntr = entry.partitionCounter(grpId, part);
if (foundCntr != null) {
- if (foundCntr <= partCntrSince) {
- first = entry.cpMark;
+ if (partCntrSince == null) {
+ if (hasGap) {
+ first = entry;
+
+ hasGap = false;
+ }
+
+ if (first == null)
+ first = entry;
+ }
+ else if (foundCntr <= partCntrSince) {
+ first = entry;
hasGap = false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/file/FilePageStoreManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/926686ec/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/pagemem/PageMemoryImpl.java
----------------------------------------------------------------------
diff --cc modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/pagemem/PageMemoryImpl.java
index a791ec9,c05af57..3df931a
--- a/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/pagemem/PageMemoryImpl.java
+++ b/modules/pds/src/main/java/org/apache/ignite/internal/processors/cache/database/pagemem/PageMemoryImpl.java
@@@ -1935,10 -1891,13 +1891,13 @@@ public class PageMemoryImpl implements
}
}
+ /**
+ * @param cacheId Cache id.
+ */
- private void resetPartTags(int cacheId) {
+ private void resetPartTags(int grpId) {
assert getWriteHoldCount() > 0;
- Iterator<T2<Integer, Integer>> iter = partitionTagMap.keySet().iterator();
+ Iterator<T2<Integer, Integer>> iter = partTagMap.keySet().iterator();
while (iter.hasNext()) {
T2<Integer, Integer> t = iter.next();