You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2019/03/07 11:05:54 UTC
[ignite] branch master updated: IGNITE-10561: MVCC: Fix MVCC cache
rebalance. This closes #5799.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 9ce4a82 IGNITE-10561: MVCC: Fix MVCC cache rebalance. This closes #5799.
9ce4a82 is described below
commit 9ce4a82a3a1bdbd4d007d9a46c49fabd9a08e566
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Mar 7 14:05:46 2019 +0300
IGNITE-10561: MVCC: Fix MVCC cache rebalance. This closes #5799.
---
.../processors/cache/GridCacheEntryEx.java | 11 +
.../processors/cache/GridCacheMapEntry.java | 195 ++++++++++-----
.../processors/cache/GridCacheMvccEntryInfo.java | 9 +
.../cache/IgniteCacheOffheapManager.java | 55 ++---
.../cache/IgniteCacheOffheapManagerImpl.java | 262 +++++++++------------
.../processors/cache/IgniteRebalanceIterator.java | 6 +
.../dht/GridDhtTxAbstractEnlistFuture.java | 29 +--
.../dht/GridDhtTxQueryFirstEnlistRequest.java | 1 +
.../dht/preloader/GridDhtPartitionDemander.java | 262 ++++++++++++++++++---
.../dht/preloader/GridDhtPartitionSupplier.java | 85 ++++---
.../dht/preloader/IgniteRebalanceIteratorImpl.java | 31 +++
.../processors/cache/mvcc/MvccProcessorImpl.java | 12 +-
.../internal/processors/cache/mvcc/MvccUtils.java | 4 +-
.../processors/cache/mvcc/txlog/TxLog.java | 7 +-
.../cache/persistence/CacheDataRowAdapter.java | 18 +-
.../cache/persistence/GridCacheOffheapManager.java | 19 +-
.../cache/persistence/tree/io/DataPageIO.java | 42 +---
.../processors/cache/tree/CacheDataRowStore.java | 4 +-
.../processors/cache/tree/CacheDataTree.java | 2 +-
.../cache/tree/mvcc/data/MvccDataRow.java | 18 +-
.../tree/mvcc/search/MvccSnapshotSearchRow.java | 2 +-
.../ignite/cache/ResetLostPartitionTest.java | 2 -
.../processors/cache/GridCacheTestEntryEx.java | 16 ++
.../cache/IgnitePdsDataRegionMetricsTxTest.java | 4 -
.../GridCacheRebalancingSyncSelfTest.java | 30 ++-
.../IgnitePdsContinuousRestartTest.java | 9 -
26 files changed, 700 insertions(+), 435 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
index 7b6ab9e..9aec399 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
@@ -1203,6 +1203,17 @@ public interface GridCacheEntryEx {
throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
+ * Apply entry history if not exists.
+ *
+ * @param entries Entries.
+ * @return {@code True} if initial value was set.
+ * @throws IgniteCheckedException, If failed.
+ * @throws GridCacheEntryRemovedException, If entry has been removed.
+ */
+ public boolean mvccPreloadEntry(List<GridCacheMvccEntryInfo> entries)
+ throws IgniteCheckedException, GridCacheEntryRemovedException;
+
+ /**
* Touch this entry in its context's eviction manager.
*
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 64e436f..a3eda18 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -102,6 +102,7 @@ import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.thread.IgniteThread;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_EXPIRED;
@@ -3368,18 +3369,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
if (cctx.mvccEnabled()) {
- if (preload && mvccVer != null) {
- cctx.offheap().mvccInitialValueIfAbsent(this,
- val,
- ver,
- expTime,
- mvccVer,
- newMvccVer,
- mvccTxState,
- newMvccTxState);
- }
- else
- cctx.offheap().mvccInitialValue(this, val, ver, expTime, mvccVer, newMvccVer);
+ assert !preload;
+
+ cctx.offheap().mvccInitialValue(this, val, ver, expTime, mvccVer, newMvccVer);
}
else
storeValue(val, expTime, ver);
@@ -3406,18 +3398,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- if (preload && mvccVer != null) {
- cctx.offheap().mvccInitialValueIfAbsent(this,
- val,
- ver,
- expTime,
- mvccVer,
- newMvccVer,
- mvccTxState,
- newMvccTxState);
- }
- else
- cctx.offheap().mvccInitialValue(this, val, ver, expTime, mvccVer, newMvccVer);
+ assert !preload;
+
+ cctx.offheap().mvccInitialValue(this, val, ver, expTime, mvccVer, newMvccVer);
}
}
else
@@ -6718,6 +6701,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
IgniteUuid futId,
int batchNum)
throws IgniteCheckedException, GridCacheEntryRemovedException {
+ assert mvccVer != null &&
+ MvccUtils.mvccVersionIsValid(mvccVer.coordinatorVersion(), mvccVer.counter(), mvccVer.operationCounter());
+ assert !F.isEmpty(entries);
WALPointer logPtr = null;
@@ -6731,17 +6717,23 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
try {
checkObsolete();
+ boolean walEnabled = cctx.group().persistenceEnabled() && cctx.group().walEnabled();
+
+ List<DataEntry> walEntries = walEnabled ? new ArrayList<>(entries.size() + 1) : Collections.EMPTY_LIST;
+ // Apply updates in reverse order (oldest is last) until any of previous versions found.
+ // If we found prev version then it means history has been actualized either with previous update
+ // or via rebalance.
for (int i = 0; i < entries.size(); i++) {
GridCacheMvccEntryInfo info = (GridCacheMvccEntryInfo)entries.get(i);
- if (val == null && op != DELETE && MvccUtils.compare(info.mvccVersion(),
- mvccVer.coordinatorVersion(),
- mvccVer.counter(),
- mvccVer.operationCounter()) == 0)
- val = info.value();
+ assert info.mvccTxState() == TxState.COMMITTED ||
+ MvccUtils.compare(info, mvccVer.coordinatorVersion(), mvccVer.counter()) == 0;
+ assert info.newMvccTxState() == TxState.COMMITTED ||
+ MvccUtils.compareNewVersion(info, mvccVer.coordinatorVersion(), mvccVer.counter()) == 0 ||
+ info.newMvccCoordinatorVersion() == MvccUtils.MVCC_CRD_COUNTER_NA;
- cctx.offheap().mvccUpdateRowWithPreloadInfo(this,
+ boolean added = cctx.offheap().mvccUpdateRowWithPreloadInfo(this,
info.value(),
info.version(),
info.expireTime(),
@@ -6749,45 +6741,53 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
info.newMvccVersion(),
info.mvccTxState(),
info.newMvccTxState());
- }
-
- // Assume list contains only previous committed row and rows changed by the current tx.
- if (!entries.isEmpty()) {
- // Assume the oldest value is the last one.
- GridCacheMvccEntryInfo last = (GridCacheMvccEntryInfo)entries.get(entries.size() - 1);
+ if (walEnabled)
+ walEntries.add(toMvccDataEntry(info, tx));
- if (MvccUtils.compare(last.mvccVersion(),
+ if (oldVal == null
+ && MvccUtils.compare(info.mvccVersion(),
mvccVer.coordinatorVersion(),
- mvccVer.counter(),
- mvccVer.operationCounter()) != 0)
- oldVal = last.value();
+ mvccVer.counter()) != 0
+ && MvccUtils.compareNewVersion(info,
+ mvccVer.coordinatorVersion(),
+ mvccVer.counter()) == 0)
+ oldVal = info.value(); // Old means a value before current transaction.
+
+ if (!added)
+ break;
}
- if (cctx.deferredDelete() && deletedUnlocked() && !detached())
- deletedUnlocked(false);
+ GridCacheMvccEntryInfo last = (GridCacheMvccEntryInfo)entries.get(0);
- long expireTime = CU.EXPIRE_TIME_ETERNAL;
- long ttl = CU.TTL_ETERNAL;
+ if (walEnabled)
+ Collections.reverse(walEntries);
- GridCacheVersion ver = tx.writeVersion();
+ if (op == DELETE) {
+ assert MvccUtils.compareNewVersion(last, mvccVer) == 0;
- if (cctx.group().persistenceEnabled() && cctx.group().walEnabled())
- logPtr = cctx.shared().wal().log(new MvccDataRecord(new MvccDataEntry(
- cctx.cacheId(),
- key,
- val,
- op,
- tx.nearXidVersion(),
- ver,
- CU.EXPIRE_TIME_ETERNAL,
- key.partition(),
- 0L,
- mvccVer)));
+ if (walEnabled)
+ walEntries.add(new MvccDataEntry(
+ cctx.cacheId(),
+ key,
+ null,
+ DELETE,
+ tx.nearXidVersion(),
+ last.version(),
+ last.expireTime(),
+ key.partition(),
+ 0,
+ last.mvccVersion()));
+ }
+ else {
+ assert last.newMvccCoordinatorVersion() == MvccUtils.MVCC_CRD_COUNTER_NA;
+ assert MvccUtils.compare(last, mvccVer) == 0;
- update(val, expireTime, ttl, ver, true);
+ val = last.value();
+ }
- recordNodeId(affNodeId, topVer);
+ if (walEnabled)
+ logPtr = cctx.shared().wal().log(new MvccDataRecord(walEntries));
}
finally {
if (lockedByCurrentThread()) {
@@ -6805,6 +6805,83 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
return res;
}
+ /** {@inheritDoc} */
+ @Override public boolean mvccPreloadEntry(List<GridCacheMvccEntryInfo> entryHist)
+ throws IgniteCheckedException, GridCacheEntryRemovedException {
+ assert !entryHist.isEmpty();
+
+ WALPointer logPtr = null;
+
+ ensureFreeSpace();
+
+ boolean updated = false;
+
+ lockEntry();
+
+ try {
+ checkObsolete();
+
+ key.valueBytes(cctx.cacheObjectContext());
+
+ if (cctx.offheap().mvccApplyHistoryIfAbsent(this, entryHist)) {
+ updated = true;
+
+ if (!cctx.isNear() && cctx.group().persistenceEnabled() && cctx.group().walEnabled()) {
+ MvccDataRecord rec;
+
+ if (entryHist.size() == 1) {
+ GridCacheMvccEntryInfo info = entryHist.get(0);
+
+ rec = new MvccDataRecord(toMvccDataEntry(info, null));
+ }
+ else {
+ // Batched WAL update.
+ List<DataEntry> dataEntries = new ArrayList<>(entryHist.size());
+
+ for (GridCacheMvccEntryInfo info : entryHist)
+ dataEntries.add(toMvccDataEntry(info, null));
+
+ rec = new MvccDataRecord(dataEntries);
+ }
+
+ logPtr = cctx.shared().wal().log(rec);
+ }
+ }
+ }
+ finally {
+ if (lockedByCurrentThread()) {
+ unlockEntry();
+
+ cctx.evicts().touch(this);
+ }
+ }
+
+ if (logPtr != null)
+ cctx.shared().wal().flush(logPtr, false);
+
+ return updated;
+ }
+
+ /**
+ * Converts mvcc entry info to WAL record entry.
+ * @param info Mvcc entry info.
+ * @return Mvcc data entry.
+ */
+ private @NotNull MvccDataEntry toMvccDataEntry(@NotNull GridCacheMvccEntryInfo info, @Nullable IgniteInternalTx tx) {
+ return new MvccDataEntry(
+ cctx.cacheId(),
+ key,
+ info.value(),
+ CREATE,
+ tx == null ? null : tx.nearXidVersion(),
+ info.version(),
+ info.expireTime(),
+ key.partition(),
+ 0,
+ info.mvccVersion()
+ );
+ }
+
/**
* @return Common serialization error.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
index b9589de..4e844f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache;
import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersionAware;
import org.apache.ignite.internal.processors.cache.mvcc.MvccVersionAware;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -36,6 +37,9 @@ public class GridCacheMvccEntryInfo extends GridCacheEntryInfo implements MvccVe
private static final long serialVersionUID = 0L;
/** */
+ private static final int SIZE_OVERHEAD = 4 * 8 /* long */ + 2 * 4 /* int */;
+
+ /** */
private long mvccCrdVer;
/** */
@@ -122,6 +126,11 @@ public class GridCacheMvccEntryInfo extends GridCacheEntryInfo implements MvccVe
}
/** {@inheritDoc} */
+ @Override public int marshalledSize(CacheObjectContext ctx) throws IgniteCheckedException {
+ return SIZE_OVERHEAD + super.marshalledSize(ctx);
+ }
+
+ /** {@inheritDoc} */
@Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
writer.setBuffer(buf);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
index c495e12..b7e8ec7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
@@ -241,27 +241,14 @@ public interface IgniteCacheOffheapManager {
) throws IgniteCheckedException;
/**
- * @param entry Entry.
- * @param val Value.
- * @param ver Version.
- * @param expireTime Expire time.
- * @param mvccVer MVCC version.
- * @param newMvccVer New MVCC version.
- * @param txState Tx state hint for the mvcc version.
- * @param newTxState Tx state hint for the new mvcc version.
- * @return {@code True} if value was inserted.
- * @throws IgniteCheckedException If failed.
+ * Tries to apply entry history.
+ * Either applies full entry history or do nothing.
+ *
+ * @param entry Entry to update.
+ * @param hist Full entry history.
+ * @return {@code True} if history applied successfully, {@code False} otherwise.
*/
- public boolean mvccInitialValueIfAbsent(
- GridCacheMapEntry entry,
- @Nullable CacheObject val,
- GridCacheVersion ver,
- long expireTime,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState
- ) throws IgniteCheckedException;
+ boolean mvccApplyHistoryIfAbsent(GridCacheMapEntry entry, List<GridCacheMvccEntryInfo> hist) throws IgniteCheckedException;
/**
* @param entry Entry.
@@ -329,6 +316,9 @@ public interface IgniteCacheOffheapManager {
) throws IgniteCheckedException;
/**
+ * Apply update with full history.
+ * Note: History version may be skipped if it have already been actualized with previous update operation.
+ *
* @param entry Entry.
* @param val Value.
* @param ver Version.
@@ -748,29 +738,22 @@ public interface IgniteCacheOffheapManager {
MvccVersion newMvccVer) throws IgniteCheckedException;
/**
+ * Tries to apply entry history.
+ * Either applies full entry history or do nothing.
+ *
* @param cctx Cache context.
* @param key Key.
- * @param val Value.
- * @param ver Version.
- * @param mvccVer MVCC version.
- * @param newMvccVer New MVCC version.
- * @param txState Tx state hint for the mvcc version.
- * @param newTxState Tx state hint for the new mvcc version.
- * @return {@code True} if new value was inserted.
- * @throws IgniteCheckedException If failed.
+ * @param hist Full entry history.
+ * @return {@code True} if entry history applied successfully, {@code False} otherwise.
*/
- boolean mvccInitialValueIfAbsent(
+ boolean mvccApplyHistoryIfAbsent(
GridCacheContext cctx,
KeyCacheObject key,
- @Nullable CacheObject val,
- GridCacheVersion ver,
- long expireTime,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState) throws IgniteCheckedException;
+ List<GridCacheMvccEntryInfo> hist) throws IgniteCheckedException;
/**
+ * Apply update with full history.
+ * Note: History version may be skipped if it have already been actualized with previous update operation.
*
* @param cctx Grid cache context.
* @param key Key.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
index 209c9cf..30fcb7c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
@@ -481,25 +481,9 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public boolean mvccInitialValueIfAbsent(
- GridCacheMapEntry entry,
- CacheObject val,
- GridCacheVersion ver,
- long expireTime,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState) throws IgniteCheckedException {
- return dataStore(entry.localPartition()).mvccInitialValueIfAbsent(
- entry.context(),
- entry.key(),
- val,
- ver,
- expireTime,
- mvccVer,
- newMvccVer,
- txState,
- newTxState);
+ @Override public boolean mvccApplyHistoryIfAbsent(GridCacheMapEntry entry, List<GridCacheMvccEntryInfo> hist)
+ throws IgniteCheckedException {
+ return dataStore(entry.localPartition()).mvccApplyHistoryIfAbsent(entry.cctx, entry.key(), hist);
}
/** {@inheritDoc} */
@@ -1159,7 +1143,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
CacheDataStore data = partitionData(part);
- final GridCursor<? extends CacheDataRow> cur = data.cursor();
+ final GridCursor<? extends CacheDataRow> cur = data.cursor(CacheDataRowAdapter.RowData.FULL_WITH_HINTS);
return new GridCloseableIteratorAdapter<CacheDataRow>() {
/** */
@@ -1630,9 +1614,17 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** {@inheritDoc} */
@Override public void invoke(GridCacheContext cctx, KeyCacheObject key, OffheapInvokeClosure c)
throws IgniteCheckedException {
+ if (!busyLock.enterBusy())
+ throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
+
int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
- invoke0(cctx, new SearchRow(cacheId, key), c);
+ try {
+ invoke0(cctx, new SearchRow(cacheId, key), c);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
}
/**
@@ -1643,42 +1635,34 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
*/
private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClosure c)
throws IgniteCheckedException {
- if (!busyLock.enterBusy())
- throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
-
- try {
- assert cctx.shared().database().checkpointLockIsHeldByThread();
+ assert cctx.shared().database().checkpointLockIsHeldByThread();
- dataTree.invoke(row, CacheDataRowAdapter.RowData.NO_KEY, c);
+ dataTree.invoke(row, CacheDataRowAdapter.RowData.NO_KEY, c);
- switch (c.operationType()) {
- case PUT: {
- assert c.newRow() != null : c;
+ switch (c.operationType()) {
+ case PUT: {
+ assert c.newRow() != null : c;
- CacheDataRow oldRow = c.oldRow();
+ CacheDataRow oldRow = c.oldRow();
- finishUpdate(cctx, c.newRow(), oldRow);
+ finishUpdate(cctx, c.newRow(), oldRow);
- break;
- }
+ break;
+ }
- case REMOVE: {
- CacheDataRow oldRow = c.oldRow();
+ case REMOVE: {
+ CacheDataRow oldRow = c.oldRow();
- finishRemove(cctx, row.key(), oldRow);
+ finishRemove(cctx, row.key(), oldRow);
- break;
- }
+ break;
+ }
- case NOOP:
- break;
+ case NOOP:
+ break;
- default:
- assert false : c.operationType();
- }
- }
- finally {
- busyLock.leaveBusy();
+ default:
+ assert false : c.operationType();
}
}
@@ -1802,19 +1786,11 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public boolean mvccInitialValueIfAbsent(
+ @Override public boolean mvccApplyHistoryIfAbsent(
GridCacheContext cctx,
KeyCacheObject key,
- @Nullable CacheObject val,
- GridCacheVersion ver,
- long expireTime,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState)
- throws IgniteCheckedException
- {
- assert mvccVer != null;
+ List<GridCacheMvccEntryInfo> hist) throws IgniteCheckedException {
+ assert !F.isEmpty(hist);
if (!busyLock.enterBusy())
throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
@@ -1825,28 +1801,53 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
// Make sure value bytes initialized.
key.valueBytes(coCtx);
- if (val != null)
- val.valueBytes(coCtx);
-
assert cctx.shared().database().checkpointLockIsHeldByThread();
- MvccPutIfAbsentClosure clo = new MvccPutIfAbsentClosure(key,
- val,
- ver,
- partId,
- expireTime,
- cctx.cacheId(),
- mvccVer,
- newMvccVer,
- txState,
- newTxState);
+ int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
- dataTree.invoke(clo, CacheDataRowAdapter.RowData.LINK_ONLY, clo);
+ // No cursor needed here as we won't iterate over whole page items, but just check if page has smth or not.
+ CheckHistoryExistsClosure clo = new CheckHistoryExistsClosure();
- if (clo.operationType() == PUT)
- finishUpdate(cctx, clo, null);
+ dataTree.iterate(
+ new MvccMaxSearchRow(cacheId, key),
+ new MvccMinSearchRow(cacheId, key),
+ clo
+ );
- return clo.operationType() == PUT;
+ if (clo.found())
+ return false;
+
+ for (GridCacheMvccEntryInfo info : hist) {
+ MvccDataRow row = new MvccDataRow(key,
+ info.value(),
+ info.version(),
+ partId,
+ info.ttl(),
+ cacheId,
+ info.mvccVersion(),
+ info.newMvccVersion());
+
+ row.mvccTxState(info.mvccTxState());
+ row.newMvccTxState(info.newMvccTxState());
+
+ assert info.newMvccTxState() == TxState.NA || info.newMvccCoordinatorVersion() > MVCC_CRD_COUNTER_NA;
+ assert MvccUtils.mvccVersionIsValid(info.mvccCoordinatorVersion(), info.mvccCounter(), info.mvccOperationCounter());
+
+ if (!grp.storeCacheIdInDataPage() && cacheId != CU.UNDEFINED_CACHE_ID)
+ row.cacheId(CU.UNDEFINED_CACHE_ID);
+
+ rowStore.addRow(row, grp.statisticsHolderData());
+
+ row.cacheId(cacheId);
+
+ boolean hasOld = dataTree.putx(row);
+
+ assert !hasOld : row;
+
+ finishUpdate(cctx, row, null);
+ }
+
+ return true;
}
finally {
busyLock.leaveBusy();
@@ -1888,13 +1889,16 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
mvccTxState,
newMvccTxState);
+ assert newMvccTxState == TxState.NA || newMvccVer.coordinatorVersion() != 0;
+ assert MvccUtils.mvccVersionIsValid(mvccVer.coordinatorVersion(), mvccVer.counter(), mvccVer.operationCounter());
+
invoke0(cctx, clo, clo);
+
+ return clo.operationType() == PUT;
}
finally {
busyLock.leaveBusy();
}
-
- return true;
}
/** {@inheritDoc} */
@@ -2030,7 +2034,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
if (needHistory) {
assert updateRow.link() != 0;
- updateRow.history().add(new MvccLinkAwareSearchRow(cacheId,
+ updateRow.history().add(0, new MvccLinkAwareSearchRow(cacheId,
key,
updateRow.mvccCoordinatorVersion(),
updateRow.mvccCounter(),
@@ -3000,70 +3004,6 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/**
- * Put row if it doesn't exist yet.
- */
- private class MvccPutIfAbsentClosure extends MvccDataRow implements IgniteTree.InvokeClosure<CacheDataRow> {
- /** */
- private IgniteTree.OperationType op;
-
- /**
- * @param key Key.
- * @param val Value.
- * @param ver Version.
- * @param part Partition.
- * @param expireTime Expire time.
- * @param cacheId Cache ID.
- * @param mvccVer Mvcc version.
- * @param newMvccVer New mvcc version.
- * @param txState Tx state hint for mvcc version.
- * @param newTxState Tx state hint for new mvcc version.
- */
- MvccPutIfAbsentClosure(KeyCacheObject key,
- CacheObject val,
- GridCacheVersion ver,
- int part,
- long expireTime,
- int cacheId,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState) {
- super(key, val, ver, part, expireTime, cacheId, mvccVer, newMvccVer);
-
- mvccTxState(txState);
- newMvccTxState(newTxState);
- }
-
- /** {@inheritDoc} */
- @Override public void call(@Nullable CacheDataRow old) throws IgniteCheckedException {
- if (old == null) {
- op = PUT;
-
- int cacheId = cacheId();
-
- if (!grp.storeCacheIdInDataPage() && cacheId != CU.UNDEFINED_CACHE_ID)
- cacheId(CU.UNDEFINED_CACHE_ID);
-
- rowStore().addRow(this, grp.statisticsHolderData());
-
- cacheId(cacheId);
- }
- else
- op = NOOP;
- }
-
- /** {@inheritDoc} */
- @Override public MvccDataRow newRow() {
- return this;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteTree.OperationType operationType() {
- return op;
- }
- }
-
- /**
*
*/
private class MvccUpdateRowWithPreloadInfoClosure extends MvccDataRow implements OffheapInvokeClosure {
@@ -3095,9 +3035,9 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
super(key,
val,
ver,
- CacheDataStoreImpl.this.partId(),
+ partId,
expireTime,
- cctx.cacheId(),
+ grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID,
mvccVer,
newMvccVer);
@@ -3150,6 +3090,29 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
return op == null ? NOOP : op;
}
}
+
+ /**
+ * Check if any row version exists.
+ */
+ private final class CheckHistoryExistsClosure implements BPlusTree.TreeRowClosure<CacheSearchRow, CacheDataRow> {
+ /** */
+ private boolean found;
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(BPlusTree<CacheSearchRow, CacheDataRow> tree, BPlusIO<CacheSearchRow> io,
+ long pageAddr, int idx) {
+ found = true;
+
+ return false; // Stop search.
+ }
+
+ /**
+ * @return {@code True} if any row was found, {@code False} otherwise.
+ */
+ public boolean found() {
+ return found;
+ }
+ }
}
/**
@@ -3268,8 +3231,10 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
assert cntr == newRow.mvccCounter();
assert opCntr == newRow.mvccOperationCounter();
+ assert newRow.mvccTxState() != TxState.NA : newRow.mvccTxState();
+
if (txState != newRow.mvccTxState() && newRow.mvccTxState() != TxState.NA) {
- assert txState == TxState.NA;
+ assert txState == TxState.NA : txState;
iox.rawMvccOperationCounter(pageAddr, off, opCntr | (newRow.mvccTxState() << MVCC_HINTS_BIT_OFF));
@@ -3284,15 +3249,18 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
newRow.newMvccCounter(),
newRow.newMvccOperationCounter()) != 0) {
+ assert newRow.newMvccTxState() == TxState.NA || newRow.newMvccCoordinatorVersion() != MVCC_CRD_COUNTER_NA;
+
iox.updateNewVersion(pageAddr, off, newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(),
newRow.newMvccOperationCounter(), newRow.newMvccTxState());
if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
ctx.wal().log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
- newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(), newRow.newMvccOperationCounter()));
+ newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(),
+ newRow.newMvccOperationCounter() | (newRow.newMvccTxState() << MVCC_HINTS_BIT_OFF)));
}
else if (newTxState != newRow.newMvccTxState() && newRow.newMvccTxState() != TxState.NA) {
- assert newTxState == TxState.NA;
+ assert newTxState == TxState.NA : newTxState;
iox.rawNewMvccOperationCounter(pageAddr, off, newOpCntr | (newRow.newMvccTxState() << MVCC_HINTS_BIT_OFF));
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteRebalanceIterator.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteRebalanceIterator.java
index d2f29d5..0ccc9f7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteRebalanceIterator.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteRebalanceIterator.java
@@ -47,4 +47,10 @@ public interface IgniteRebalanceIterator extends GridCloseableIterator<CacheData
* @param partId Partition ID.
*/
public void setPartitionMissing(int partId);
+
+ /**
+ * Return next element without moving iterator cursor to the next one.
+ * @return Next element or {@code Null} if there is no more elements.
+ */
+ public CacheDataRow peek();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
index 6f3153c..6b443cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
@@ -725,8 +725,7 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd
* @return History entries.
* @throws IgniteCheckedException, if failed.
*/
- private CacheEntryInfoCollection fetchHistoryInfo(KeyCacheObject key, List<MvccLinkAwareSearchRow> hist)
- throws IgniteCheckedException {
+ private CacheEntryInfoCollection fetchHistoryInfo(KeyCacheObject key, List<MvccLinkAwareSearchRow> hist) {
List<GridCacheEntryInfo> res = new ArrayList<>();
for (int i = 0; i < hist.size(); i++) {
@@ -736,7 +735,7 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd
row0.hash(),
row0.link(),
key.partition(),
- CacheDataRowAdapter.RowData.NO_KEY,
+ CacheDataRowAdapter.RowData.NO_KEY_WITH_HINTS,
row0.mvccCoordinatorVersion(),
row0.mvccCounter(),
row0.mvccOperationCounter(),
@@ -745,22 +744,24 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd
GridCacheMvccEntryInfo entry = new GridCacheMvccEntryInfo();
+ entry.cacheId(cctx.cacheId());
entry.version(row.version());
- entry.mvccVersion(row);
- entry.newMvccVersion(row);
entry.value(row.value());
entry.expireTime(row.expireTime());
- if (MvccUtils.compare(mvccSnapshot, row.mvccCoordinatorVersion(), row.mvccCounter()) != 0) {
- entry.mvccTxState(row.mvccTxState() != TxState.NA ? row.mvccTxState() :
- MvccUtils.state(cctx, row.mvccCoordinatorVersion(), row.mvccCounter(), row.mvccOperationCounter()));
- }
+ // Row should be retrieved with actual hints.
+ entry.mvccVersion(row);
+ entry.newMvccVersion(row);
- if (MvccUtils.compare(mvccSnapshot, row.newMvccCoordinatorVersion(), row.newMvccCounter()) != 0) {
- entry.newMvccTxState(row.newMvccTxState() != TxState.NA ? row.newMvccTxState() :
- MvccUtils.state(cctx, row.newMvccCoordinatorVersion(), row.newMvccCounter(),
- row.newMvccOperationCounter()));
- }
+ if (MvccUtils.compare(mvccSnapshot, row.mvccCoordinatorVersion(), row.mvccCounter()) != 0)
+ entry.mvccTxState(row.mvccTxState());
+
+ if (row.newMvccCoordinatorVersion() != MvccUtils.MVCC_CRD_COUNTER_NA
+ && MvccUtils.compare(mvccSnapshot, row.newMvccCoordinatorVersion(), row.newMvccCounter()) != 0)
+ entry.newMvccTxState(row.newMvccTxState());
+
+
+ assert mvccSnapshot.coordinatorVersion() != MvccUtils.MVCC_CRD_COUNTER_NA;
res.add(entry);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
index 53ceb05..1f96e63 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
@@ -103,6 +103,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
List<KeyCacheObject> keys,
List<Message> vals) {
super(cacheId, dhtFutId, lockVer, op, batchId, snapshot.operationCounter(), keys, vals);
+
this.cacheId = cacheId;
this.subjId = subjId;
this.topVer = topVer;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index 0d0b2e8..61f1e06 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
+import org.apache.ignite.internal.processors.cache.GridCacheMvccEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCachePartitionExchangeManager;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
@@ -64,6 +65,7 @@ import org.apache.ignite.internal.util.lang.IgniteInClosureX;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI1;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -766,35 +768,10 @@ public class GridDhtPartitionDemander {
try {
Iterator<GridCacheEntryInfo> infos = e.getValue().infos().iterator();
- // Loop through all received entries and try to preload them.
- while (infos.hasNext()) {
- ctx.database().checkpointReadLock();
-
- try {
- for (int i = 0; i < 100; i++) {
- if (!infos.hasNext())
- break;
-
- GridCacheEntryInfo entry = infos.next();
-
- if (!preloadEntry(node, p, entry, topVer)) {
- if (log.isTraceEnabled())
- log.trace("Got entries for invalid partition during " +
- "preloading (will skip) [p=" + p + ", entry=" + entry + ']');
-
- break;
- }
-
- for (GridCacheContext cctx : grp.caches()) {
- if (cctx.statisticsEnabled())
- cctx.cache().metrics0().onRebalanceKeyReceived();
- }
- }
- }
- finally {
- ctx.database().checkpointReadUnlock();
- }
- }
+ if (grp.mvccEnabled())
+ mvccPreloadEntries(topVer, node, p, infos);
+ else
+ preloadEntries(topVer, node, p, infos);
// If message was last for this partition,
// then we take ownership.
@@ -875,12 +852,153 @@ public class GridDhtPartitionDemander {
}
/**
+ * Adds mvcc entries with theirs history to partition p.
+ *
+ * @param node Node which sent entry.
+ * @param p Partition id.
+ * @param infos Entries info for preload.
+ * @param topVer Topology version.
+ * @throws IgniteInterruptedCheckedException If interrupted.
+ */
+ private void mvccPreloadEntries(AffinityTopologyVersion topVer, ClusterNode node, int p,
+ Iterator<GridCacheEntryInfo> infos) throws IgniteCheckedException {
+ if (!infos.hasNext())
+ return;
+
+ List<GridCacheMvccEntryInfo> entryHist = new ArrayList<>();
+
+ GridCacheContext cctx = grp.sharedGroup() ? null : grp.singleCacheContext();
+
+ // Loop through all received entries and try to preload them.
+ while (infos.hasNext() || !entryHist.isEmpty()) {
+ ctx.database().checkpointReadLock();
+
+ try {
+ for (int i = 0; i < 100; i++) {
+ boolean hasMore = infos.hasNext();
+
+ assert hasMore || !entryHist.isEmpty();
+
+ GridCacheMvccEntryInfo entry = null;
+
+ boolean flushHistory;
+
+ if (hasMore) {
+ entry = (GridCacheMvccEntryInfo)infos.next();
+
+ GridCacheMvccEntryInfo prev = entryHist.isEmpty() ? null : entryHist.get(0);
+
+ flushHistory = prev != null && ((grp.sharedGroup() && prev.cacheId() != entry.cacheId())
+ || !prev.key().equals(entry.key()));
+ }
+ else
+ flushHistory = true;
+
+ if (flushHistory) {
+ assert !entryHist.isEmpty();
+
+ int cacheId = entryHist.get(0).cacheId();
+
+ if (grp.sharedGroup() && (cctx == null || cacheId != cctx.cacheId())) {
+ assert cacheId != CU.UNDEFINED_CACHE_ID;
+
+ cctx = grp.shared().cacheContext(cacheId);
+ }
+
+ if (cctx != null) {
+ if (!mvccPreloadEntry(cctx, node, entryHist, topVer, p)) {
+ if (log.isTraceEnabled())
+ log.trace("Got entries for invalid partition during " +
+ "preloading (will skip) [p=" + p +
+ ", entry=" + entryHist.get(entryHist.size() - 1) + ']');
+
+ return; // Skip current partition.
+ }
+
+ //TODO: IGNITE-11330: Update metrics for touched cache only.
+ for (GridCacheContext ctx : grp.caches()) {
+ if (ctx.statisticsEnabled())
+ ctx.cache().metrics0().onRebalanceKeyReceived();
+ }
+ }
+
+ if (!hasMore)
+ return;
+
+ entryHist.clear();
+ }
+
+ entryHist.add(entry);
+ }
+ }
+ finally {
+ ctx.database().checkpointReadUnlock();
+ }
+ }
+ }
+
+ /**
+ * Adds entries with theirs history to partition p.
+ *
+ * @param node Node which sent entry.
+ * @param p Partition id.
+ * @param infos Entries info for preload.
+ * @param topVer Topology version.
+ * @throws IgniteInterruptedCheckedException If interrupted.
+ */
+ private void preloadEntries(AffinityTopologyVersion topVer, ClusterNode node, int p,
+ Iterator<GridCacheEntryInfo> infos) throws IgniteCheckedException {
+ GridCacheContext cctx = null;
+
+ // Loop through all received entries and try to preload them.
+ while (infos.hasNext()) {
+ ctx.database().checkpointReadLock();
+
+ try {
+ for (int i = 0; i < 100; i++) {
+ if (!infos.hasNext())
+ break;
+
+ GridCacheEntryInfo entry = infos.next();
+
+ if (cctx == null || (grp.sharedGroup() && entry.cacheId() != cctx.cacheId())) {
+ cctx = grp.sharedGroup() ? grp.shared().cacheContext(entry.cacheId()) : grp.singleCacheContext();
+
+ if (cctx == null)
+ continue;
+ else if (cctx.isNear())
+ cctx = cctx.dhtCache().context();
+ }
+
+ if (!preloadEntry(node, p, entry, topVer, cctx)) {
+ if (log.isTraceEnabled())
+ log.trace("Got entries for invalid partition during " +
+ "preloading (will skip) [p=" + p + ", entry=" + entry + ']');
+
+ return;
+ }
+
+ //TODO: IGNITE-11330: Update metrics for touched cache only.
+ for (GridCacheContext ctx : grp.caches()) {
+ if (ctx.statisticsEnabled())
+ ctx.cache().metrics0().onRebalanceKeyReceived();
+ }
+ }
+ }
+ finally {
+ ctx.database().checkpointReadUnlock();
+ }
+ }
+ }
+
+ /**
* Adds {@code entry} to partition {@code p}.
*
* @param from Node which sent entry.
* @param p Partition id.
* @param entry Preloaded entry.
* @param topVer Topology version.
+ * @param cctx Cache context.
* @return {@code False} if partition has become invalid during preloading.
* @throws IgniteInterruptedCheckedException If interrupted.
*/
@@ -888,7 +1006,8 @@ public class GridDhtPartitionDemander {
ClusterNode from,
int p,
GridCacheEntryInfo entry,
- AffinityTopologyVersion topVer
+ AffinityTopologyVersion topVer,
+ GridCacheContext cctx
) throws IgniteCheckedException {
assert ctx.database().checkpointLockIsHeldByThread();
@@ -896,15 +1015,7 @@ public class GridDhtPartitionDemander {
GridCacheEntryEx cached = null;
try {
- GridCacheContext cctx = grp.sharedGroup() ? ctx.cacheContext(entry.cacheId()) : grp.singleCacheContext();
-
- if (cctx == null)
- return true;
-
- if (cctx.isNear())
- cctx = cctx.dhtCache().context();
-
- cached = cctx.cache().entryEx(entry.key());
+ cached = cctx.cache().entryEx(entry.key(), topVer);
if (log.isTraceEnabled())
log.trace("Rebalancing key [key=" + entry.key() + ", part=" + p + ", node=" + from.id() + ']');
@@ -966,6 +1077,79 @@ public class GridDhtPartitionDemander {
}
/**
+ * Adds mvcc {@code entry} with it's history to partition {@code p}.
+ *
+ * @param cctx Cache context.
+ * @param from Node which sent entry.
+ * @param history Mvcc entry history.
+ * @param topVer Topology version.
+ * @param p Partition id.
+ * @return {@code False} if partition has become invalid during preloading.
+ * @throws IgniteInterruptedCheckedException If interrupted.
+ */
+ private boolean mvccPreloadEntry(
+ GridCacheContext cctx,
+ ClusterNode from,
+ List<GridCacheMvccEntryInfo> history,
+ AffinityTopologyVersion topVer,
+ int p
+ ) throws IgniteCheckedException {
+ assert ctx.database().checkpointLockIsHeldByThread();
+ assert !history.isEmpty();
+
+ GridCacheMvccEntryInfo info = history.get(0);
+
+ assert info.key() != null;
+
+ try {
+ GridCacheEntryEx cached = null;
+
+ try {
+ cached = cctx.cache().entryEx(info.key(), topVer);
+
+ if (log.isTraceEnabled())
+ log.trace("Rebalancing key [key=" + info.key() + ", part=" + p + ", node=" + from.id() + ']');
+
+ if (cached.mvccPreloadEntry(history)) {
+ cached.touch(); // Start tracking.
+
+ if (cctx.events().isRecordable(EVT_CACHE_REBALANCE_OBJECT_LOADED) && !cached.isInternal())
+ cctx.events().addEvent(cached.partition(), cached.key(), cctx.localNodeId(), null,
+ null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, null, true, null,
+ false, null, null, null, true);
+ }
+ else {
+ cached.touch(); // Start tracking.
+
+ if (log.isTraceEnabled())
+ log.trace("Rebalancing entry is already in cache (will ignore) [key=" + cached.key() +
+ ", part=" + p + ']');
+ }
+ }
+ catch (GridCacheEntryRemovedException ignored) {
+ if (log.isTraceEnabled())
+ log.trace("Entry has been concurrently removed while rebalancing (will ignore) [key=" +
+ cached.key() + ", part=" + p + ']');
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ if (log.isDebugEnabled())
+ log.debug("Partition became invalid during rebalancing (will ignore): " + p);
+
+ return false;
+ }
+ }
+ catch (IgniteInterruptedCheckedException | ClusterTopologyCheckedException e) {
+ throw e;
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteCheckedException("Failed to cache rebalanced entry (will stop rebalancing) [local=" +
+ ctx.localNode() + ", node=" + from.id() + ", key=" + info.key() + ", part=" + p + ']', e);
+ }
+
+ return true;
+ }
+
+ /**
* String representation of demand routine.
*
* @param topicId Topic id.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
index c6bcc80..514f8fd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
@@ -316,8 +316,17 @@ class GridDhtPartitionSupplier {
long batchesCnt = 0;
+ CacheDataRow prevRow = null;
+
while (iter.hasNext()) {
- if (supplyMsg.messageSize() >= msgMaxSize) {
+ CacheDataRow row = iter.peek();
+
+ // Prevent mvcc entry history splitting into separate batches.
+ boolean canFlushHistory = !grp.mvccEnabled() ||
+ prevRow != null && ((grp.sharedGroup() && row.cacheId() != prevRow.cacheId()) ||
+ !row.key().equals(prevRow.key()));
+
+ if (canFlushHistory && supplyMsg.messageSize() >= msgMaxSize) {
if (++batchesCnt >= maxBatchesCnt) {
saveSupplyContext(contextId,
iter,
@@ -340,7 +349,9 @@ class GridDhtPartitionSupplier {
}
}
- CacheDataRow row = iter.next();
+ row = iter.next();
+
+ prevRow = row;
int part = row.partition();
@@ -364,38 +375,10 @@ class GridDhtPartitionSupplier {
if (!remainingParts.contains(part))
continue;
- GridCacheEntryInfo info = grp.mvccEnabled() ?
- new GridCacheMvccEntryInfo() : new GridCacheEntryInfo();
-
- info.key(row.key());
- info.cacheId(row.cacheId());
-
- if (grp.mvccEnabled()) {
- byte txState = row.mvccTxState() != TxState.NA ? row.mvccTxState() :
- MvccUtils.state(grp, row.mvccCoordinatorVersion(), row.mvccCounter(),
- row.mvccOperationCounter());
-
- if (txState != TxState.COMMITTED)
- continue;
-
- ((MvccVersionAware)info).mvccVersion(row);
- ((GridCacheMvccEntryInfo)info).mvccTxState(TxState.COMMITTED);
-
- byte newTxState = row.newMvccTxState() != TxState.NA ? row.newMvccTxState() :
- MvccUtils.state(grp, row.newMvccCoordinatorVersion(), row.newMvccCounter(),
- row.newMvccOperationCounter());
-
- if (newTxState != TxState.ABORTED) {
- ((MvccUpdateVersionAware)info).newMvccVersion(row);
-
- if (newTxState == TxState.COMMITTED)
- ((GridCacheMvccEntryInfo)info).newMvccTxState(TxState.COMMITTED);
- }
- }
+ GridCacheEntryInfo info = extractEntryInfo(row);
- info.value(row.value());
- info.version(row.version());
- info.expireTime(row.expireTime());
+ if (info == null)
+ continue;
if (preloadPred == null || preloadPred.apply(info))
supplyMsg.addEntry0(part, iter.historical(part), info, grp.shared(), grp.cacheObjectContext());
@@ -503,6 +486,42 @@ class GridDhtPartitionSupplier {
}
/**
+ * Extracts entry info from row.
+ * @param row Cache data row.
+ * @return Entry info.
+ */
+ private GridCacheEntryInfo extractEntryInfo(CacheDataRow row) {
+ GridCacheEntryInfo info = grp.mvccEnabled() ?
+ new GridCacheMvccEntryInfo() : new GridCacheEntryInfo();
+
+ info.key(row.key());
+ info.cacheId(row.cacheId());
+
+ if (grp.mvccEnabled()) {
+ assert row.mvccCoordinatorVersion() != MvccUtils.MVCC_CRD_COUNTER_NA;
+
+ // Rows from rebalance iterator have actual states already.
+ if (row.mvccTxState() != TxState.COMMITTED)
+ return null;
+
+ ((MvccVersionAware)info).mvccVersion(row);
+ ((GridCacheMvccEntryInfo)info).mvccTxState(TxState.COMMITTED);
+
+ if (row.newMvccCoordinatorVersion() != MvccUtils.MVCC_CRD_COUNTER_NA &&
+ row.newMvccTxState() == TxState.COMMITTED) {
+ ((MvccUpdateVersionAware)info).newMvccVersion(row);
+ ((GridCacheMvccEntryInfo)info).newMvccTxState(TxState.COMMITTED);
+ }
+ }
+
+ info.value(row.value());
+ info.version(row.version());
+ info.expireTime(row.expireTime());
+
+ return info;
+ }
+
+ /**
* Sends supply message to demand node.
*
* @param demander Recipient of supply message.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteRebalanceIteratorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteRebalanceIteratorImpl.java
index 87ba3a9..75ae89a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteRebalanceIteratorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteRebalanceIteratorImpl.java
@@ -50,6 +50,9 @@ public class IgniteRebalanceIteratorImpl implements IgniteRebalanceIterator {
/** Current full iterator. */
private Map.Entry<Integer, GridCloseableIterator<CacheDataRow>> current;
+ /** Next value. */
+ private CacheDataRow cached;
+
/** */
private boolean reachedEnd;
@@ -103,6 +106,9 @@ public class IgniteRebalanceIteratorImpl implements IgniteRebalanceIterator {
if (historical(partId))
return historicalIterator.isDone(partId);
+ if (cached != null)
+ return cached.partition() > partId;
+
return current == null || current.getKey() > partId;
}
@@ -142,12 +148,26 @@ public class IgniteRebalanceIteratorImpl implements IgniteRebalanceIterator {
}
/** {@inheritDoc} */
+ @Override public synchronized CacheDataRow peek() {
+ if (cached == null) {
+ if (!hasNext())
+ return null;
+
+ cached = next();
+ }
+
+ return cached;
+ }
+
+ /** {@inheritDoc} */
@Override public synchronized void removeX() throws IgniteCheckedException {
throw new UnsupportedOperationException("remove");
}
/** {@inheritDoc} */
@Override public synchronized void close() throws IgniteCheckedException {
+ cached = null;
+
if (historicalIterator != null)
historicalIterator.close();
@@ -172,6 +192,9 @@ public class IgniteRebalanceIteratorImpl implements IgniteRebalanceIterator {
/** {@inheritDoc} */
@Override public synchronized boolean hasNext() {
try {
+ if (cached != null)
+ return true;
+
return hasNextX();
}
catch (IgniteCheckedException e) {
@@ -182,6 +205,14 @@ public class IgniteRebalanceIteratorImpl implements IgniteRebalanceIterator {
/** {@inheritDoc} */
@Override public synchronized CacheDataRow next() {
try {
+ if (cached != null) {
+ CacheDataRow res = cached;
+
+ cached = null;
+
+ return res;
+ }
+
return nextX();
}
catch (IgniteCheckedException e) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
index dfe4f75..478d156 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
@@ -2173,8 +2173,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
int curCacheId = CU.UNDEFINED_CACHE_ID;
try {
- GridCursor<? extends CacheDataRow> cursor = part.dataStore().cursor(KEY_ONLY);
-
KeyCacheObject prevKey = null;
Object rest = null;
@@ -2190,6 +2188,8 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (!shared && (cctx = F.first(part.group().caches())) == null)
return metrics;
+ GridCursor<? extends CacheDataRow> cursor = part.dataStore().cursor(KEY_ONLY);
+
while (cursor.next()) {
if (isCancelled())
throw new IgniteInterruptedCheckedException("Operation has been cancelled.");
@@ -2199,14 +2199,14 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (prevKey == null)
prevKey = row.key();
- if (cctx == null) {
+ if (cctx == null) { // Shared group.
cctx = part.group().shared().cacheContext(curCacheId = row.cacheId());
if (cctx == null)
- return metrics;
+ continue;
}
- if (!prevKey.equals(row.key()) || (shared && curCacheId != row.cacheId())) {
+ if ((shared && curCacheId != row.cacheId()) || !prevKey.equals(row.key())) {
if (rest != null || !F.isEmpty(cleanupRows))
cleanup(part, prevKey, cleanupRows, rest, cctx, metrics);
@@ -2218,7 +2218,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
cctx = part.group().shared().cacheContext(curCacheId = row.cacheId());
if (cctx == null)
- return metrics;
+ continue;
}
prevKey = row.key();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
index cf7b62b..cbc7023 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
@@ -173,7 +173,7 @@ public class MvccUtils {
* @return TxState
* @see TxState
*/
- private static byte state(MvccProcessor proc, long mvccCrd, long mvccCntr, int mvccOpCntr) {
+ public static byte state(MvccProcessor proc, long mvccCrd, long mvccCntr, int mvccOpCntr) {
if (compare(INITIAL_VERSION, mvccCrd, mvccCntr, mvccOpCntr) == 0)
return TxState.COMMITTED; // Initial version is always committed;
@@ -550,7 +550,7 @@ public class MvccUtils {
* @return {@code True} if version is valid.
*/
public static boolean mvccVersionIsValid(long crdVer, long cntr) {
- return crdVer > MVCC_CRD_COUNTER_NA && cntr != MVCC_COUNTER_NA;
+ return crdVer > MVCC_CRD_COUNTER_NA && cntr > MVCC_COUNTER_NA;
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
index 528bbbc..2351d29 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.PageMemory;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageInitRecord;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.persistence.DbCheckpointListener;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
@@ -48,8 +49,6 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_IDX;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
-import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
-import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
/**
*
@@ -377,7 +376,6 @@ public class TxLog implements DbCheckpointListener {
/** {@inheritDoc} */
@Override public boolean apply(BPlusTree<TxKey, TxRow> tree, BPlusIO<TxKey> io, long pageAddr,
int idx) throws IgniteCheckedException {
-
if (rows == null)
rows = new ArrayList<>();
@@ -431,7 +429,8 @@ public class TxLog implements DbCheckpointListener {
* @param primary Flag if this is primary node.
*/
TxLogUpdateClosure(long major, long minor, byte newState, boolean primary) {
- assert major > MVCC_CRD_COUNTER_NA && minor > MVCC_COUNTER_NA && newState != TxState.NA;
+ assert MvccUtils.mvccVersionIsValid(major, minor) && newState != TxState.NA;
+
this.major = major;
this.minor = minor;
this.newState = newState;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
index c9d006f..3e3dc5c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
@@ -319,7 +319,7 @@ public class CacheDataRowAdapter implements CacheDataRow {
}
// Assume that row header is always located entirely on the very first page.
- hdrLen = readHeader(pageAddr, data.offset());
+ hdrLen = readHeader(sharedCtx, pageAddr, data.offset(), rowData);
if (rowData == LINK_WITH_HEADER)
return null;
@@ -346,11 +346,13 @@ public class CacheDataRowAdapter implements CacheDataRow {
/**
* Reads row header (i.e. MVCC info) which should be located on the very first page od data.
*
+ * @param sharedCtx Shared context.
* @param addr Address.
* @param off Offset
+ * @param rowData Required row data.
* @return Number of bytes read.
*/
- protected int readHeader(long addr, int off) {
+ protected int readHeader(GridCacheSharedContext<?, ?> sharedCtx, long addr, int off, RowData rowData) {
// No-op.
return 0;
}
@@ -461,7 +463,7 @@ public class CacheDataRowAdapter implements CacheDataRow {
) throws IgniteCheckedException {
int off = 0;
- off += readHeader(addr, off);
+ off += readHeader(sharedCtx, addr, off, rowData);
if (rowData == LINK_WITH_HEADER)
return;
@@ -478,7 +480,7 @@ public class CacheDataRowAdapter implements CacheDataRow {
int len = PageUtils.getInt(addr, off);
off += 4;
- if (rowData != RowData.NO_KEY) {
+ if (rowData != RowData.NO_KEY && rowData != RowData.NO_KEY_WITH_HINTS) {
byte type = PageUtils.getByte(addr, off);
off++;
@@ -856,7 +858,13 @@ public class CacheDataRowAdapter implements CacheDataRow {
LINK_ONLY,
/** */
- LINK_WITH_HEADER
+ LINK_WITH_HEADER,
+
+ /** Force instant hints actualization for rebalance (to avoid races with vacuum). */
+ FULL_WITH_HINTS,
+
+ /** Force instant hints actualization for update operation with history (to avoid races with vacuum). */
+ NO_KEY_WITH_HINTS
}
/** {@inheritDoc} */
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 16e9840..2978f69 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
@@ -55,6 +55,7 @@ import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMvccEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheTtlManager;
import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -1957,24 +1958,16 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
return delegate.mvccInitialValue(cctx, key, val, ver, expireTime, mvccVer, newMvccVer);
}
-
+
/** {@inheritDoc} */
- @Override public boolean mvccInitialValueIfAbsent(
+ @Override public boolean mvccApplyHistoryIfAbsent(
GridCacheContext cctx,
KeyCacheObject key,
- @Nullable CacheObject val,
- GridCacheVersion ver,
- long expireTime,
- MvccVersion mvccVer,
- MvccVersion newMvccVer,
- byte txState,
- byte newTxState)
- throws IgniteCheckedException
- {
+ List<GridCacheMvccEntryInfo> hist)
+ throws IgniteCheckedException {
CacheDataStore delegate = init0(false);
- return delegate.mvccInitialValueIfAbsent(cctx, key, val, ver, expireTime, mvccVer, newMvccVer,
- txState, newTxState);
+ return delegate.mvccApplyHistoryIfAbsent(cctx, key, hist);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
index 9ed7b72..f67c89e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
@@ -246,44 +246,6 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
/**
* @param pageAddr Page address.
* @param dataOff Data offset.
- * @param ver New version.
- */
- public void updateVersion(long pageAddr, int dataOff, MvccVersion ver) {
- long addr = pageAddr + dataOff;
-
- updateVersion(addr, ver.coordinatorVersion(), ver.counter(), ver.operationCounter());
- }
-
- /**
- * @param pageAddr Page address.
- * @param itemId Item ID.
- * @param pageSize Page size.
- * @param mvccCrd Mvcc coordinator.
- * @param mvccCntr Mvcc counter.
- * @param mvccOpCntr Operation counter.
- */
- public void updateVersion(long pageAddr, int itemId, int pageSize, long mvccCrd, long mvccCntr, int mvccOpCntr) {
- int dataOff = getDataOffset(pageAddr, itemId, pageSize);
-
- long addr = pageAddr + dataOff + (isFragmented(pageAddr, dataOff) ? 10 : 2);
-
- updateVersion(addr, mvccCrd, mvccCntr, mvccOpCntr);
- }
-
- /**
- * @param addr Address.
- * @param mvccCrd Mvcc coordinator.
- * @param mvccCntr Mvcc counter.
- */
- private void updateVersion(long addr, long mvccCrd, long mvccCntr, int mvccOpCntr) {
- PageUtils.putLong(addr, 0, mvccCrd);
- PageUtils.putLong(addr, 8, mvccCntr);
- PageUtils.putInt(addr, 16, mvccOpCntr);
- }
-
- /**
- * @param pageAddr Page address.
- * @param dataOff Data offset.
* @param mvccCrd Mvcc coordinator.
* @param mvccCntr Mvcc counter.
* @param mvccOpCntr Operation counter.
@@ -325,7 +287,7 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
int opCntr = rawMvccOperationCounter(addr, 0);
- rawMvccOperationCounter(addr, 0, ((int)txState << MVCC_HINTS_BIT_OFF) | (opCntr & ~MVCC_HINTS_MASK));
+ rawMvccOperationCounter(addr, 0, (opCntr & ~MVCC_HINTS_MASK) | ((int)txState << MVCC_HINTS_BIT_OFF));
}
/**
@@ -341,7 +303,7 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
int opCntr = rawNewMvccOperationCounter(addr, 0);
- rawNewMvccOperationCounter(addr, 0, ((int)txState << MVCC_HINTS_BIT_OFF) | (opCntr & ~MVCC_HINTS_MASK));
+ rawNewMvccOperationCounter(addr, 0, (opCntr & ~MVCC_HINTS_MASK) | ((int)txState << MVCC_HINTS_BIT_OFF));
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
index 0c460ee..197a685 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
@@ -102,7 +102,7 @@ public class CacheDataRowStore extends RowStore {
* @return Search row.
*/
MvccDataRow mvccRow(int cacheId, int hash, long link, CacheDataRowAdapter.RowData rowData, long crdVer, long mvccCntr, int opCntr) {
- MvccDataRow dataRow = new MvccDataRow(
+ MvccDataRow row = new MvccDataRow(
grp,
hash,
link,
@@ -114,7 +114,7 @@ public class CacheDataRowStore extends RowStore {
SKIP_VER.get()
);
- return initDataRow(dataRow, cacheId);
+ return initDataRow(row, cacheId);
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
index 71c8a86..b3c1c69 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
@@ -330,7 +330,7 @@ public class CacheDataTree extends BPlusTree<CacheSearchRow, CacheDataRow> {
/** {@inheritDoc} */
@Override protected int compare(BPlusIO<CacheSearchRow> iox, long pageAddr, int idx, CacheSearchRow row)
throws IgniteCheckedException {
- assert !grp.mvccEnabled() || row.mvccCoordinatorVersion() != 0
+ assert !grp.mvccEnabled() || row.mvccCoordinatorVersion() != MvccUtils.MVCC_CRD_COUNTER_NA
|| (row.getClass() == SearchRow.class && row.key() == null) : row;
RowLinkIO io = (RowLinkIO)iox;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
index 92f5aed..cd6866e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
@@ -21,6 +21,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.pagemem.PageUtils;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
@@ -37,6 +38,8 @@ import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HI
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_KEY_ABSENT_BEFORE_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.FULL_WITH_HINTS;
+import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.NO_KEY_WITH_HINTS;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
/**
@@ -168,7 +171,9 @@ public class MvccDataRow extends DataRow {
}
/** {@inheritDoc} */
- @Override protected int readHeader(long addr, int off) {
+ @Override protected int readHeader(GridCacheSharedContext<?, ?> sharedCtx, long addr, int off, RowData rowData) {
+ boolean addHints = rowData == FULL_WITH_HINTS || rowData == NO_KEY_WITH_HINTS;
+
// xid_min.
mvccCrd = PageUtils.getLong(addr, off);
mvccCntr = PageUtils.getLong(addr, off + 8);
@@ -178,6 +183,9 @@ public class MvccDataRow extends DataRow {
mvccOpCntr = withHint & MVCC_OP_COUNTER_MASK;
mvccTxState = (byte)(withHint >>> MVCC_HINTS_BIT_OFF);
+ if (addHints && mvccTxState == TxState.NA)
+ mvccTxState = MvccUtils.state(sharedCtx.coordinators(), mvccCrd, mvccCntr, mvccOpCntr);
+
assert MvccUtils.mvccVersionIsValid(mvccCrd, mvccCntr, mvccOpCntr);
keyAbsentFlag = (withHint & MVCC_KEY_ABSENT_BEFORE_MASK) != 0;
@@ -193,9 +201,13 @@ public class MvccDataRow extends DataRow {
assert newMvccCrd == MVCC_CRD_COUNTER_NA || MvccUtils.mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr);
- if (newMvccCrd != MVCC_CRD_COUNTER_NA)
+ if (newMvccCrd != MVCC_CRD_COUNTER_NA) {
keyAbsentFlag = (withHint & MVCC_KEY_ABSENT_BEFORE_MASK) != 0;
+ if (addHints && newMvccTxState == TxState.NA)
+ newMvccTxState = MvccUtils.state(sharedCtx.coordinators(), newMvccCrd, newMvccCntr, newMvccOpCntr);
+ }
+
return MVCC_INFO_SIZE;
}
@@ -233,6 +245,8 @@ public class MvccDataRow extends DataRow {
/** {@inheritDoc} */
@Override public int newMvccOperationCounter() {
+ assert (newMvccOpCntr & ~MVCC_OP_COUNTER_MASK) == 0;
+
return newMvccOpCntr;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
index 432593d..c0bc634 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
@@ -59,7 +59,7 @@ public class MvccSnapshotSearchRow extends MvccSearchRow implements MvccTreeClos
*/
public MvccSnapshotSearchRow(GridCacheContext cctx, KeyCacheObject key,
MvccSnapshot snapshot) {
- super(cctx.cacheId(), key, snapshot.coordinatorVersion(), snapshot.counter(), Integer.MAX_VALUE);
+ super(cctx.cacheId(), key, snapshot.coordinatorVersion(), snapshot.counter(), MvccUtils.MVCC_READ_OP_CNTR);
this.cctx = cctx;
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java b/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
index a32eb67..bef69a2 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
@@ -53,8 +53,6 @@ public class ResetLostPartitionTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10560", MvccFeatureChecker.forcedMvcc());
-
super.beforeTest();
stopAllGrids();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
index 9624bd8..d6e59af 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
@@ -84,6 +84,22 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
}
/** {@inheritDoc} */
+ @Override public boolean initialValue(CacheObject val, GridCacheVersion ver, long ttl, long expireTime,
+ boolean preload, AffinityTopologyVersion topVer, GridDrType drType, boolean fromStore) {
+ assert false;
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean mvccPreloadEntry(
+ List<GridCacheMvccEntryInfo> entries) {
+ assert false;
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean isInternal() {
return key instanceof GridCacheInternal;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgnitePdsDataRegionMetricsTxTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgnitePdsDataRegionMetricsTxTest.java
index 49b734b..f31a2a6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgnitePdsDataRegionMetricsTxTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgnitePdsDataRegionMetricsTxTest.java
@@ -21,8 +21,6 @@ import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsDataRegionMetricsTest;
-import org.apache.ignite.testframework.MvccFeatureChecker;
-import org.junit.Assume;
import org.junit.Ignore;
import org.junit.Test;
@@ -45,8 +43,6 @@ public class IgnitePdsDataRegionMetricsTxTest extends IgnitePdsDataRegionMetrics
@Ignore("https://issues.apache.org/jira/browse/IGNITE-10662")
@Test
@Override public void testMemoryUsageMultipleNodes() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10583", MvccFeatureChecker.forcedMvcc());
-
super.testMemoryUsageMultipleNodes();
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
index 3bd4a25..8d7ba7c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
@@ -30,9 +30,12 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheRebalanceMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteKernal;
@@ -55,12 +58,10 @@ import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.GridTestUtils.SF;
import org.apache.ignite.testframework.MvccFeatureChecker;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.Assume;
import org.junit.Test;
import static org.apache.ignite.cache.CacheMode.LOCAL;
@@ -107,18 +108,18 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration iCfg = super.getConfiguration(igniteInstanceName);
- ((TcpDiscoverySpi)iCfg.getDiscoverySpi()).setForceServerMode(true);
+ if (MvccFeatureChecker.forcedMvcc()) {
+ iCfg.setDataStorageConfiguration(new DataStorageConfiguration()
+ .setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration().setMaxSize(400L * 1024 * 1024)
+ ));
+ }
TcpCommunicationSpi commSpi = new CollectingCommunicationSpi();
-
- commSpi.setLocalPort(GridTestUtils.getNextCommPort(getClass()));
commSpi.setTcpNoDelay(true);
iCfg.setCommunicationSpi(commSpi);
- if (getTestIgniteInstanceName(10).equals(igniteInstanceName))
- iCfg.setClientMode(true);
-
CacheConfiguration<Integer, Integer> cachePCfg = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
cachePCfg.setName(CACHE_NAME_DHT_PARTITIONED);
@@ -129,6 +130,7 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
cachePCfg.setRebalanceBatchesPrefetchCount(1);
cachePCfg.setRebalanceOrder(2);
cachePCfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ cachePCfg.setAffinity(new RendezvousAffinityFunction().setPartitions(32));
CacheConfiguration<Integer, Integer> cachePCfg2 = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
@@ -139,7 +141,7 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
cachePCfg2.setRebalanceOrder(2);
cachePCfg2.setRebalanceDelay(SF.applyLB(5000, 500));
cachePCfg2.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
-
+ cachePCfg2.setAffinity(new RendezvousAffinityFunction().setPartitions(32));
CacheConfiguration<Integer, Integer> cacheRCfg = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
@@ -149,8 +151,7 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
cacheRCfg.setRebalanceBatchSize(1);
cacheRCfg.setRebalanceBatchesPrefetchCount(Integer.MAX_VALUE);
cacheRCfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
-
- ((TcpCommunicationSpi)iCfg.getCommunicationSpi()).setSharedMemoryPort(-1);//Shmem fail fix for Integer.MAX_VALUE.
+ cacheRCfg.setAffinity(new RendezvousAffinityFunction().setPartitions(32));
CacheConfiguration<Integer, Integer> cacheRCfg2 = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
@@ -159,6 +160,7 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
cacheRCfg2.setRebalanceMode(CacheRebalanceMode.SYNC);
cacheRCfg2.setRebalanceOrder(4);
cacheRCfg2.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ cacheRCfg2.setAffinity(new RendezvousAffinityFunction().setPartitions(32));
iCfg.setCacheConfiguration(cachePCfg, cachePCfg2, cacheRCfg, cacheRCfg2);
@@ -231,7 +233,7 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
TEST_SIZE + ", iteration=" + iter + ", cache=" + name + "]");
assertEquals("Value does not match [key=" + entry.getKey() + ", cache=" + name + ']',
- entry.getValue().intValue(), entry.getKey() + name.hashCode() + iter);
+ entry.getKey() + name.hashCode() + iter, entry.getValue().intValue());
});
assertEquals(TEST_SIZE, cnt.get());
@@ -260,8 +262,6 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
*/
@Test
public void testSimpleRebalancing() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10560", MvccFeatureChecker.forcedMvcc());
-
IgniteKernal ignite = (IgniteKernal)startGrid(0);
generateData(ignite, 0, 0);
@@ -502,8 +502,6 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
*/
@Test
public void testComplexRebalancing() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10561", MvccFeatureChecker.forcedMvcc());
-
final Ignite ignite = startGrid(0);
generateData(ignite, 0, 0);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
index 8b33339..b52d03d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsContinuousRestartTest.java
@@ -116,13 +116,6 @@ public class IgnitePdsContinuousRestartTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10561", MvccFeatureChecker.forcedMvcc());
-
- super.beforeTest();
- }
-
- /** {@inheritDoc} */
@Override protected void afterTest() throws Exception {
stopAllGrids();
@@ -230,8 +223,6 @@ public class IgnitePdsContinuousRestartTest extends GridCommonAbstractTest {
*/
@Test
public void testRebalancingDuringLoad_10_10_1_1() throws Exception {
- Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10583", MvccFeatureChecker.forcedMvcc());
-
checkRebalancingDuringLoad(10, 10, 1, 1);
}