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/09/07 14:23:01 UTC
[3/3] ignite git commit: ignite-3484
ignite-3484
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e40b4d9d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e40b4d9d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e40b4d9d
Branch: refs/heads/ignite-3484
Commit: e40b4d9dcd6fe6c1cd2640bdd7116ca5a08ed781
Parents: 91b9911
Author: sboikov <sb...@gridgain.com>
Authored: Thu Sep 7 12:12:32 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Sep 7 17:22:52 2017 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/MvccTestApp2.java | 311 ++++++++++++-------
.../communication/GridIoMessageFactory.java | 20 +-
.../processors/cache/GridCacheAdapter.java | 15 +-
.../processors/cache/GridCacheEntryEx.java | 8 +-
.../processors/cache/GridCacheMapEntry.java | 23 +-
.../cache/IgniteCacheOffheapManager.java | 6 +-
.../cache/IgniteCacheOffheapManagerImpl.java | 71 ++++-
.../GridDistributedTxRemoteAdapter.java | 1 -
.../distributed/dht/GridDhtCacheAdapter.java | 11 +-
.../cache/distributed/dht/GridDhtGetFuture.java | 11 +-
.../distributed/dht/GridDhtGetSingleFuture.java | 5 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 3 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 6 +-
.../distributed/dht/GridDhtTxFinishRequest.java | 4 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 10 +-
.../dht/GridDhtTxPrepareRequest.java | 4 +-
.../dht/GridPartitionedGetFuture.java | 27 +-
.../dht/GridPartitionedSingleGetFuture.java | 5 +-
.../dht/atomic/GridDhtAtomicCache.java | 14 +-
.../dht/colocated/GridDhtColocatedCache.java | 5 +-
.../distributed/near/GridNearGetFuture.java | 13 +-
.../distributed/near/GridNearGetRequest.java | 16 +-
.../GridNearPessimisticTxPrepareFuture.java | 12 +-
.../near/GridNearTxFinishAndAckFuture.java | 4 +-
.../near/GridNearTxFinishFuture.java | 6 +-
.../near/GridNearTxFinishRequest.java | 4 +-
.../cache/distributed/near/GridNearTxLocal.java | 19 +-
.../near/GridNearTxPrepareResponse.java | 4 +-
.../local/atomic/GridLocalAtomicCache.java | 13 +-
.../mvcc/CacheCoordinatorsSharedManager.java | 176 ++++++++---
.../mvcc/CoordinatorMvccCounterResponse.java | 147 ---------
.../mvcc/CoordinatorQueryCounterRequest.java | 121 --------
.../mvcc/CoordinatorQueryVersionRequest.java | 121 ++++++++
.../mvcc/CoordinatorQueryVersionResponse.java | 201 ++++++++++++
.../cache/mvcc/CoordinatorTxCounterRequest.java | 26 +-
.../mvcc/CoordinatorTxCounterResponse.java | 147 +++++++++
.../processors/cache/mvcc/MvccQueryVersion.java | 45 +++
.../cache/mvcc/MvccUpdateVersion.java | 182 +++++++++++
.../processors/cache/mvcc/TxMvccVersion.java | 90 ------
.../persistence/GridCacheOffheapManager.java | 5 +-
.../cache/transactions/IgniteTxAdapter.java | 8 +-
.../cache/transactions/IgniteTxHandler.java | 3 +-
.../transactions/IgniteTxLocalAdapter.java | 3 +-
.../cache/tree/AbstractDataInnerIO.java | 13 +-
.../cache/tree/AbstractDataLeafIO.java | 9 +-
.../cache/tree/CacheDataRowStore.java | 4 +-
.../processors/cache/tree/CacheDataTree.java | 17 +-
.../cache/tree/CacheIdAwareDataInnerIO.java | 4 +-
.../cache/tree/CacheIdAwareDataLeafIO.java | 4 +-
.../processors/cache/tree/DataInnerIO.java | 4 +-
.../processors/cache/tree/DataLeafIO.java | 4 +-
.../processors/cache/tree/MvccDataRow.java | 4 +-
.../processors/cache/tree/SearchRow.java | 4 +-
.../processors/cache/GridCacheTestEntryEx.java | 7 +-
.../cache/mvcc/CacheMvccTransactionsTest.java | 210 +++++++++++--
.../processors/query/h2/opt/GridH2Row.java | 10 +
56 files changed, 1502 insertions(+), 718 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/MvccTestApp2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/MvccTestApp2.java b/modules/core/src/main/java/org/apache/ignite/internal/MvccTestApp2.java
index 36c88bc..1f8c9ac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/MvccTestApp2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/MvccTestApp2.java
@@ -46,6 +46,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.NotNull;
@@ -59,7 +60,7 @@ public class MvccTestApp2 {
private static final boolean DEBUG_LOG = false;
/** */
- private static final boolean SQL = true;
+ private static final boolean SQL = false;
public static void main1(String[] args) throws Exception {
final TestCluster cluster = new TestCluster(1);
@@ -124,7 +125,7 @@ public class MvccTestApp2 {
cluster.txPutAll(data);
- cluster.txRemoveTransfer(0, 1);
+ //cluster.txRemoveTransfer(0, 1);
Map<Object, Object> getData = cluster.sqlAll();;//cluster.getAll(data.keySet());
@@ -283,52 +284,60 @@ public class MvccTestApp2 {
if (REMOVES) {
thread = new Thread(new Runnable() {
@Override public void run() {
- Thread.currentThread().setName("update" + id);
+ try {
+ Thread.currentThread().setName("update" + id);
- ThreadLocalRandom rnd = ThreadLocalRandom.current();
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
- while (!stop.get()) {
- int id1 = rnd.nextInt(ACCOUNTS);
+ while (!stop.get()) {
+ int id1 = rnd.nextInt(ACCOUNTS);
- int id2 = rnd.nextInt(ACCOUNTS);
+ int id2 = rnd.nextInt(ACCOUNTS);
- while (id2 == id1)
- id2 = rnd.nextInt(ACCOUNTS);
+ while (id2 == id1)
+ id2 = rnd.nextInt(ACCOUNTS);
- if (rnd.nextBoolean()) {
- cluster.txRemoveTransfer(id1, id2);
+ if (rnd.nextBoolean()) {
+ //cluster.txRemoveTransfer(id1, id2);
+ }
+ else
+ cluster.txTransfer(id1, id2, rnd.nextBoolean());
}
- else
- cluster.txTransfer(id1, id2, rnd.nextBoolean());
}
-
+ catch (Exception e) {
+ e.printStackTrace();
+ }
}
});
}
else {
thread = new Thread(new Runnable() {
@Override public void run() {
- Thread.currentThread().setName("update" + id);
+ try {
+ Thread.currentThread().setName("update" + id);
- ThreadLocalRandom rnd = ThreadLocalRandom.current();
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
- while (!stop.get()) {
- int id1 = rnd.nextInt(ACCOUNTS);
+ while (!stop.get()) {
+ int id1 = rnd.nextInt(ACCOUNTS);
- int id2 = rnd.nextInt(ACCOUNTS);
+ int id2 = rnd.nextInt(ACCOUNTS);
- while (id2 == id1)
- id2 = rnd.nextInt(ACCOUNTS);
+ while (id2 == id1)
+ id2 = rnd.nextInt(ACCOUNTS);
- if (id1 > id2) {
- int tmp = id1;
- id1 = id2;
- id2 = tmp;
- }
+ if (id1 > id2) {
+ int tmp = id1;
+ id1 = id2;
+ id2 = tmp;
+ }
- cluster.txTransfer(id1, id2, rnd.nextBoolean());
+ cluster.txTransfer(id1, id2, rnd.nextBoolean());
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
}
-
}
});
}
@@ -429,9 +438,9 @@ public class MvccTestApp2 {
mappedEntries.put(key, node);
}
- CoordinatorCounter cntr = crd.nextTxCounter(txId);
+ TxVersion ver = crd.nextTxCounter(txId);
- MvccUpdateVersion mvccVer = new MvccUpdateVersion(cntr, txId);
+ MvccUpdateVersion mvccVer = new MvccUpdateVersion(ver.cntr, txId);
for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
Node node = e.getValue();
@@ -445,10 +454,10 @@ public class MvccTestApp2 {
node.dataStore.unlockEntry(e.getKey());
}
- crd.txDone(txId, cntr.cntr);
+ crd.txDone(txId, ver.cntr.cntr);
}
- void txTransfer(Integer id1, Integer id2, boolean fromFirst) {
+ void txTransfer(Integer id1, Integer id2, boolean fromFirst) throws Exception {
TreeSet<Integer> keys = new TreeSet<>();
keys.add(id1);
@@ -472,7 +481,29 @@ public class MvccTestApp2 {
mappedEntries.put(key, node);
}
- CoordinatorCounter cntr = crd.nextTxCounter(txId);
+ TxVersion ver = crd.nextTxCounter(txId);
+
+ Collection<TxId> waitTxs = null;
+
+ for (Object key : keys) {
+ int nodeIdx = nodeForKey(key);
+
+ Node node = nodes.get(nodeIdx);
+
+ Collection<TxId> txs = node.dataStore.waitTxsAck(key, ver.activeTxs);
+
+ if (txs != null) {
+ if (waitTxs == null)
+ waitTxs = txs;
+ else
+ waitTxs.addAll(txs);
+ }
+ }
+
+ if (waitTxs != null) {
+ crd.waitTxs(waitTxs);
+ }
+
Integer curVal1 = (Integer)vals.get(id1);
Integer curVal2 = (Integer)vals.get(id2);
@@ -507,10 +538,10 @@ public class MvccTestApp2 {
newVals.put(id1, newVal1);
newVals.put(id2, newVal2);
- MvccUpdateVersion mvccVer = new MvccUpdateVersion(cntr, txId);
+ MvccUpdateVersion mvccVer = new MvccUpdateVersion(ver.cntr, txId);
if (DEBUG_LOG) {
- TestDebugLog.msgs.add(new TestDebugLog.Msg6("update", txId, id1, newVal1, id2, newVal2, cntr));
+ TestDebugLog.msgs.add(new TestDebugLog.Msg6("update", txId, id1, newVal1, id2, newVal2, ver.cntr));
}
for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
@@ -524,6 +555,8 @@ public class MvccTestApp2 {
node.dataStore.unlockEntry(e.getKey());
}
+
+ crd.txDone(txId, ver.cntr.cntr);
}
else {
for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
@@ -531,82 +564,82 @@ public class MvccTestApp2 {
node.dataStore.unlockEntry(e.getKey());
}
- }
- crd.txDone(txId, cntr.cntr);
+ crd.txDone(txId, ver.cntr.cntr);
+ }
// if (DEBUG_LOG)
// TestDebugLog.msgs.add(new TestDebugLog.Msg2("tx done", txId, cntr.cntr));
}
- void txRemoveTransfer(Integer from, Integer to) {
- TreeSet<Integer> keys = new TreeSet<>();
-
- keys.add(from);
- keys.add(to);
-
- TxId txId = new TxId(txIdGen.incrementAndGet());
-
- Map<Object, Node> mappedEntries = new LinkedHashMap<>();
-
- Map<Object, Object> vals = new HashMap<>();
-
- for (Object key : keys) {
- int nodeIdx = nodeForKey(key);
-
- Node node = nodes.get(nodeIdx);
-
- node.dataStore.lockEntry(key);
-
- vals.put(key, node.dataStore.lastValue(key));
-
- mappedEntries.put(key, node);
- }
-
- CoordinatorCounter cntr = crd.nextTxCounter(txId);
-
- Integer fromVal = (Integer)vals.get(from);
- Integer toVal = (Integer)vals.get(to);
-
- boolean update = fromVal != null && toVal != null;
-
- if (update) {
- Map<Object, Object> newVals = new HashMap<>();
-
- newVals.put(from, null);
- newVals.put(to, fromVal + toVal);
-
- MvccUpdateVersion mvccVer = new MvccUpdateVersion(cntr, txId);
-
- if (DEBUG_LOG) {
- TestDebugLog.msgs.add(new TestDebugLog.Msg6("remove", txId, from, fromVal, to, toVal, cntr));
- }
-
- for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
- Node node = e.getValue();
-
- node.dataStore.updateEntry(e.getKey(), newVals.get(e.getKey()), mvccVer);
- }
-
- for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
- Node node = e.getValue();
-
- node.dataStore.unlockEntry(e.getKey());
- }
- }
- else {
- for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
- Node node = e.getValue();
-
- node.dataStore.unlockEntry(e.getKey());
- }
- }
-
- crd.txDone(txId, cntr.cntr);
-
- if (DEBUG_LOG)
- TestDebugLog.msgs.add(new TestDebugLog.Msg2("tx done", txId, cntr.cntr));
- }
+// void txRemoveTransfer(Integer from, Integer to) {
+// TreeSet<Integer> keys = new TreeSet<>();
+//
+// keys.add(from);
+// keys.add(to);
+//
+// TxId txId = new TxId(txIdGen.incrementAndGet());
+//
+// Map<Object, Node> mappedEntries = new LinkedHashMap<>();
+//
+// Map<Object, Object> vals = new HashMap<>();
+//
+// for (Object key : keys) {
+// int nodeIdx = nodeForKey(key);
+//
+// Node node = nodes.get(nodeIdx);
+//
+// node.dataStore.lockEntry(key);
+//
+// vals.put(key, node.dataStore.lastValue(key));
+//
+// mappedEntries.put(key, node);
+// }
+//
+// CoordinatorCounter cntr = crd.nextTxCounter(txId);
+//
+// Integer fromVal = (Integer)vals.get(from);
+// Integer toVal = (Integer)vals.get(to);
+//
+// boolean update = fromVal != null && toVal != null;
+//
+// if (update) {
+// Map<Object, Object> newVals = new HashMap<>();
+//
+// newVals.put(from, null);
+// newVals.put(to, fromVal + toVal);
+//
+// MvccUpdateVersion mvccVer = new MvccUpdateVersion(cntr, txId);
+//
+// if (DEBUG_LOG) {
+// TestDebugLog.msgs.add(new TestDebugLog.Msg6("remove", txId, from, fromVal, to, toVal, cntr));
+// }
+//
+// for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
+// Node node = e.getValue();
+//
+// node.dataStore.updateEntry(e.getKey(), newVals.get(e.getKey()), mvccVer);
+// }
+//
+// for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
+// Node node = e.getValue();
+//
+// node.dataStore.unlockEntry(e.getKey());
+// }
+// }
+// else {
+// for (Map.Entry<Object, Node> e : mappedEntries.entrySet()) {
+// Node node = e.getValue();
+//
+// node.dataStore.unlockEntry(e.getKey());
+// }
+// }
+//
+// crd.txDone(txId, cntr.cntr);
+//
+// if (DEBUG_LOG)
+// TestDebugLog.msgs.add(new TestDebugLog.Msg2("tx done", txId, cntr.cntr));
+// }
public void dumpMvccInfo() {
for (Node node : nodes) {
@@ -701,6 +734,19 @@ public class MvccTestApp2 {
}
}
+ static class TxVersion {
+ final CoordinatorCounter cntr;
+
+ /** */
+ @GridToStringInclude
+ final Collection<TxId> activeTxs;
+
+ public TxVersion(CoordinatorCounter cntr, Collection<TxId> activeTxs) {
+ this.cntr = cntr;
+ this.activeTxs = activeTxs;
+ }
+ }
+
/**
*
*/
@@ -718,14 +764,28 @@ public class MvccTestApp2 {
@GridToStringInclude
private final ConcurrentHashMap8<TxId, Long> activeTxs = new ConcurrentHashMap8<>();
- synchronized CoordinatorCounter nextTxCounter(TxId txId) {
+ synchronized void waitTxs(Collection<TxId> waitTxs) throws InterruptedException {
+ for (TxId txId : waitTxs) {
+ while (activeTxs.containsKey(txId))
+ wait();
+ }
+ }
+
+ synchronized TxVersion nextTxCounter(TxId txId) {
long cur = cntr.get();
activeTxs.put(txId, cur + 1);
CoordinatorCounter newCtr = new CoordinatorCounter(cntr.incrementAndGet());
- return newCtr;
+ Set<TxId> txs = new HashSet<>();
+
+ for (Map.Entry<TxId, Long> e : activeTxs.entrySet())
+ txs.add(e.getKey());
+
+ TxVersion ver = new TxVersion(newCtr, txs);
+
+ return ver;
}
synchronized void txDone(TxId txId, long cntr) {
@@ -734,6 +794,8 @@ public class MvccTestApp2 {
assert rmvd != null;
commitCntr.setIfGreater(cntr);
+
+ notifyAll();
}
private Long minActive(Set<TxId> txs) {
@@ -762,12 +824,15 @@ public class MvccTestApp2 {
synchronized MvccQueryVersion queryVersion() {
long useCntr = commitCntr.get();
+// Long minActive = minActive(txs);
+//
+// if (minActive != null && minActive < useCntr)
+// useCntr = minActive - 1;
+
Set<TxId> txs = new HashSet<>();
- Long minActive = minActive(txs);
-
- if (minActive != null && minActive < useCntr)
- useCntr = minActive - 1;
+ for (Map.Entry<TxId, Long> e : activeTxs.entrySet())
+ txs.add(e.getKey());
MvccQueryVersion qryVer = new MvccQueryVersion(new CoordinatorCounter(useCntr), txs);
@@ -1058,6 +1123,28 @@ public class MvccTestApp2 {
}
}
+ Collection<TxId> waitTxsAck(Object key, Collection<TxId> activeTxs) {
+ if (!F.isEmpty(activeTxs))
+ return null;
+
+ List<MvccValue> list = mvccIdx.get(key);
+
+ List<TxId> waitTxs = null;
+
+ if (list != null) {
+ for (MvccValue val : list) {
+ if (activeTxs.contains(val.ver.txId)) {
+ if (waitTxs == null)
+ waitTxs = new ArrayList<>();
+
+ waitTxs.add(val.ver.txId);
+ }
+ }
+ }
+
+ return waitTxs;
+ }
+
void lockEntry(Object key) {
ReentrantLock e = lock(key);
@@ -1165,7 +1252,7 @@ public class MvccTestApp2 {
private boolean versionVisible(MvccUpdateVersion ver, MvccQueryVersion qryVer) {
int cmp = ver.cntr.compareTo(qryVer.cntr);
- return cmp <= 0;// && !qryVer.activeTxs.contains(ver.txId);
+ return cmp <= 0 && !qryVer.activeTxs.contains(ver.txId);
}
Object get(Object key, MvccQueryVersion ver) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index eae435e..9bd7a9a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -102,12 +102,14 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFi
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearUnlockRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorMvccCounterResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorQueryVersionResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorTxCounterResponse;
import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorQueryAckRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorQueryCounterRequest;
+import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorQueryVersionRequest;
import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorTxAckRequest;
import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorTxAckResponse;
import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorTxCounterRequest;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
@@ -887,7 +889,7 @@ public class GridIoMessageFactory implements MessageFactory {
break;
case 130:
- msg = new CoordinatorMvccCounterResponse();
+ msg = new CoordinatorTxCounterResponse();
break;
@@ -902,7 +904,7 @@ public class GridIoMessageFactory implements MessageFactory {
break;
case 133:
- msg = new CoordinatorQueryCounterRequest();
+ msg = new CoordinatorQueryVersionRequest();
break;
@@ -911,6 +913,16 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case 135:
+ msg = new MvccUpdateVersion();
+
+ return msg;
+
+ case 136:
+ msg = new CoordinatorQueryVersionResponse();
+
+ return msg;
+
// [-3..119] [124..128] [-23..-27] [-36..-55]- this
// [120..123] - DR
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 83b085a..693f399 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -89,7 +89,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvali
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalAdapter;
@@ -1813,7 +1813,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
recovery,
canRemap,
needVer,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478.
+ null); // TODO IGNITE-3478.
}
/**
@@ -1845,7 +1845,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean recovery,
boolean canRemap,
final boolean needVer,
- long mvccCrdCntr
+ MvccQueryVersion mvccVer
) {
if (F.isEmpty(keys))
return new GridFinishedFuture<>(Collections.<K1, V1>emptyMap());
@@ -1902,7 +1902,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
boolean skipEntry = readNoEntry;
if (readNoEntry) {
- CacheDataRow row = ctx.offheap().read(ctx, key);
+ CacheDataRow row = mvccVer != null ? ctx.offheap().mvccRead(ctx, key, mvccVer) :
+ ctx.offheap().read(ctx, key);
if (row != null) {
long expireTime = row.expireTime();
@@ -1956,7 +1957,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
expiry,
!deserializeBinary,
- mvccCrdCntr,
+ mvccVer,
readerArgs);
assert res != null;
@@ -1981,7 +1982,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
expiry,
!deserializeBinary,
- mvccCrdCntr,
+ mvccVer,
readerArgs);
if (res == null)
@@ -4752,7 +4753,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*taskName*/null,
/*expiryPlc*/null,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
if (val == null)
return null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
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 8c17d4e..24c066a 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
@@ -28,7 +28,7 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
@@ -270,7 +270,7 @@ public interface GridCacheEntryEx {
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- long mvccCntr)
+ @Nullable MvccQueryVersion mvccVer)
throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
@@ -298,7 +298,7 @@ public interface GridCacheEntryEx {
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- long mvccCntr,
+ @Nullable MvccQueryVersion mvccVer,
@Nullable ReaderArguments readerArgs)
throws IgniteCheckedException, GridCacheEntryRemovedException;
@@ -320,7 +320,7 @@ public interface GridCacheEntryEx {
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- long mvccCntr,
+ @Nullable MvccQueryVersion mvccVer,
@Nullable ReaderArguments readerArgs) throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
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 3e39b6b..40837f9 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
@@ -46,7 +46,8 @@ import org.apache.ignite.internal.processors.cache.extras.GridCacheEntryExtras;
import org.apache.ignite.internal.processors.cache.extras.GridCacheMvccEntryExtras;
import org.apache.ignite.internal.processors.cache.extras.GridCacheObsoleteEntryExtras;
import org.apache.ignite.internal.processors.cache.extras.GridCacheTtlEntryExtras;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
@@ -467,7 +468,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
String taskName,
@Nullable IgniteCacheExpiryPolicy expirePlc,
boolean keepBinary,
- long mvccCntr)
+ MvccQueryVersion mvccVer)
throws IgniteCheckedException, GridCacheEntryRemovedException {
return (CacheObject)innerGet0(
ver,
@@ -482,7 +483,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
false,
keepBinary,
false,
- mvccCntr,
+ mvccVer,
null);
}
@@ -493,7 +494,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- long mvccCntr,
+ MvccQueryVersion mvccVer,
@Nullable ReaderArguments readerArgs) throws IgniteCheckedException, GridCacheEntryRemovedException {
return (EntryGetResult)innerGet0(
/*ver*/null,
@@ -508,7 +509,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
keepBinary,
/*reserve*/true,
- mvccCntr,
+ mvccVer,
readerArgs);
}
@@ -523,7 +524,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- long mvccCntr,
+ MvccQueryVersion mvccVer,
@Nullable ReaderArguments readerArgs)
throws IgniteCheckedException, GridCacheEntryRemovedException {
return (EntryGetResult)innerGet0(
@@ -539,7 +540,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
keepBinary,
false,
- mvccCntr,
+ mvccVer,
readerArgs);
}
@@ -558,7 +559,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
boolean retVer,
boolean keepBinary,
boolean reserveForLoad,
- long mvccCntr,
+ MvccQueryVersion mvccVer,
@Nullable ReaderArguments readerArgs
) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert !(retVer && readThrough);
@@ -582,8 +583,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
CacheObject val;
- if (mvccCntr != TxMvccVersion.COUNTER_NA) {
- CacheDataRow row = cctx.offheap().readMvcc(cctx, key, 0, mvccCntr); // TODO IGNITE-3484.
+ if (mvccVer != null) {
+ CacheDataRow row = cctx.offheap().mvccRead(cctx, key, mvccVer);
if (row != null) {
val = row.value();
@@ -1007,7 +1008,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert val != null;
if (cctx.mvccEnabled()) {
- assert mvccCntr != TxMvccVersion.COUNTER_NA;
+ assert mvccCntr != MvccUpdateVersion.COUNTER_NA;
cctx.offheap().mvccUpdate(this, val, newVer, topVer.topologyVersion(), mvccCntr);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
----------------------------------------------------------------------
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 cbf5455..67a02b8 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
@@ -21,6 +21,7 @@ import java.util.Map;
import javax.cache.Cache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.RootPage;
import org.apache.ignite.internal.processors.cache.persistence.RowStore;
@@ -114,7 +115,7 @@ public interface IgniteCacheOffheapManager {
* @return Cached row, if available, null otherwise.
* @throws IgniteCheckedException If failed.
*/
- @Nullable public CacheDataRow readMvcc(GridCacheContext cctx, KeyCacheObject key, long topVer, long mvccCntr)
+ @Nullable public CacheDataRow mvccRead(GridCacheContext cctx, KeyCacheObject key, MvccQueryVersion ver)
throws IgniteCheckedException;
/**
@@ -511,7 +512,8 @@ public interface IgniteCacheOffheapManager {
* @return Data row.
* @throws IgniteCheckedException If failed.
*/
- public CacheDataRow findMvcc(GridCacheContext cctx, KeyCacheObject key, long topVer, long mvccCntr) throws IgniteCheckedException;
+ public CacheDataRow mvccFind(GridCacheContext cctx, KeyCacheObject key, MvccQueryVersion ver)
+ throws IgniteCheckedException;
/**
* @return Data cursor.
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
----------------------------------------------------------------------
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 9a4b17b..50b6e3a 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
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
@@ -36,6 +37,8 @@ import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
@@ -414,11 +417,13 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Nullable @Override public CacheDataRow readMvcc(GridCacheContext cctx, KeyCacheObject key, long topVer, long mvccCntr)
+ @Nullable @Override public CacheDataRow mvccRead(GridCacheContext cctx, KeyCacheObject key, MvccQueryVersion ver)
throws IgniteCheckedException {
+ assert ver != null;
+
CacheDataStore dataStore = dataStore(cctx, key);
- CacheDataRow row = dataStore != null ? dataStore.findMvcc(cctx, key, topVer, mvccCntr) : null;
+ CacheDataRow row = dataStore != null ? dataStore.mvccFind(cctx, key, ver) : null;
assert row == null || row.value() != null : row;
@@ -1313,6 +1318,10 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
try {
int cacheId = grp.storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
+// log.info("mvccUpdate [k=" + key.value(cctx.cacheObjectContext(), false) +
+// ", topVer=" + topVer +
+// ", cntr=" + mvccCntr + ']');
+
MvccDataRow dataRow = new MvccDataRow(key, val, ver, partId, cacheId, topVer, mvccCntr);
CacheObjectContext coCtx = cctx.cacheObjectContext();
@@ -1536,7 +1545,20 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
- CacheDataRow row = dataTree.findOne(new SearchRow(cacheId, key), CacheDataRowAdapter.RowData.NO_KEY);
+ CacheDataRow row;
+
+ if (grp.mvccEnabled()) {
+ // TODO IGNITE-3484: need special method.
+ GridCursor<CacheDataRow> cur = dataTree.find(new MvccSearchRow(cacheId, key, Long.MAX_VALUE, Long.MAX_VALUE),
+ new MvccSearchRow(cacheId, key, 1, 1));
+
+ if (cur.next())
+ row = cur.get();
+ else
+ row = null;
+ }
+ else
+ row = dataTree.findOne(new SearchRow(cacheId, key), CacheDataRowAdapter.RowData.NO_KEY);
afterRowFound(row, key);
@@ -1544,24 +1566,49 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public CacheDataRow findMvcc(GridCacheContext cctx,
+ @Override public CacheDataRow mvccFind(GridCacheContext cctx,
KeyCacheObject key,
- long topVer,
- long mvccCntr) throws IgniteCheckedException {
+ MvccQueryVersion ver) throws IgniteCheckedException {
+// log.info("mvccFind [k=" + key.value(cctx.cacheObjectContext(), false) +
+// ", topVer=" + ver.topologyVersion() +
+// ", cntr=" + ver.counter() + ']');
+
key.valueBytes(cctx.cacheObjectContext());
int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
// TODO IGNITE-3484: need special method.
-
- GridCursor<CacheDataRow> cur = dataTree.find(new MvccSearchRow(cacheId, key, topVer, mvccCntr),
- new MvccSearchRow(cacheId, key, topVer + 1, mvccCntr)/*,
- CacheDataRowAdapter.RowData.NO_KEY*/);
+ GridCursor<CacheDataRow> cur = dataTree.find(
+ new MvccSearchRow(cacheId, key, ver.topologyVersion(), ver.counter()),
+ new MvccSearchRow(cacheId, key, 1, 1));
CacheDataRow row = null;
- if (cur.next())
- row = cur.get();
+ List<MvccUpdateVersion> txs = ver.activeTransactions();
+
+ while (cur.next()) {
+ CacheDataRow row0 = cur.get();
+
+ assert row0.mvccUpdateTopologyVersion() > 0 : row0;
+
+ boolean visible;
+
+ if (txs != null) {
+ MvccUpdateVersion rowTx = new MvccUpdateVersion(
+ row0.mvccUpdateTopologyVersion(),
+ row0.mvccUpdateCounter());
+
+ visible = !txs.contains(rowTx);
+ }
+ else
+ visible = true;
+
+ if (visible) {
+ row = row0;
+
+ break;
+ }
+ }
assert row == null || key.equals(row.key());
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index fef3e7e..543f137 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -47,7 +47,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheReturnCompletableWra
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index 818a859..760ac85 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetR
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.platform.cache.PlatformCacheEntryFilter;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -776,7 +777,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
boolean skipVals,
boolean canRemap,
boolean recovery,
- long mvccCrdCntr
+ MvccQueryVersion mvccVer
) {
return getAllAsync0(keys,
readerArgs,
@@ -791,7 +792,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
recovery,
canRemap,
/*need version*/true,
- mvccCrdCntr);
+ mvccVer);
}
/**
@@ -818,7 +819,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
- long mvccCrdCntr
+ MvccQueryVersion mvccVer
) {
GridDhtGetFuture<K, V> fut = new GridDhtGetFuture<>(ctx,
msgId,
@@ -832,7 +833,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
skipVals,
recovery,
addReaders,
- mvccCrdCntr);
+ mvccVer);
fut.init();
@@ -1005,7 +1006,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
expiryPlc,
req.skipValues(),
req.recovery(),
- req.mvccCoordinatorCounter());
+ req.mvccVersion());
fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() {
@Override public void apply(IgniteInternalFuture<Collection<GridCacheEntryInfo>> f) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
index 737c276..3b87bb3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.ReaderArguments;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridCompoundIdentityFuture;
@@ -115,7 +116,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
private final boolean addReaders;
/** */
- private final long mvccCrdCntr;
+ private final MvccQueryVersion mvccVer;
/**
* @param cctx Context.
@@ -142,7 +143,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
boolean skipVals,
boolean recovery,
boolean addReaders,
- long mvccCrdCntr
+ MvccQueryVersion mvccVer
) {
super(CU.<GridCacheEntryInfo>collectionsReducer(keys.size()));
@@ -161,7 +162,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
this.skipVals = skipVals;
this.recovery = recovery;
this.addReaders = addReaders;
- this.mvccCrdCntr = mvccCrdCntr;
+ this.mvccVer = mvccVer;
futId = IgniteUuid.randomUuid();
@@ -429,7 +430,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
skipVals,
/*can remap*/true,
recovery,
- mvccCrdCntr);
+ mvccVer);
}
else {
final ReaderArguments args = readerArgs;
@@ -454,7 +455,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
skipVals,
/*can remap*/true,
recovery,
- mvccCrdCntr);
+ mvccVer);
}
}
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
index a5eedec..67e726e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
@@ -35,7 +35,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.ReaderArguments;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
@@ -368,7 +367,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
skipVals,
/*can remap*/true,
recovery,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
else {
final ReaderArguments args = readerArgs;
@@ -395,7 +394,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
skipVals,
/*can remap*/true,
recovery,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
fut0.listen(createGetFutureListener());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index efc76bf..47719b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -56,7 +56,6 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSing
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxRemote;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearUnlockRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
@@ -1296,7 +1295,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
tx != null ? tx.resolveTaskName() : null,
null,
req.keepBinary(),
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
assert e.lockedBy(mappedVer) ||
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index 9ca1412..2e64485 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -35,7 +35,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheCompoundIdentityFutu
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -349,7 +349,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
tx.activeCachesDeploymentEnabled(),
false,
false,
- TxMvccVersion.COUNTER_NA);
+ MvccUpdateVersion.COUNTER_NA);
try {
cctx.io().send(n, req, tx.ioPolicy());
@@ -397,7 +397,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
if (tx.onePhaseCommit())
return false;
- assert !commit || !tx.txState().mvccEnabled(cctx) || tx.mvccCoordinatorCounter() != TxMvccVersion.COUNTER_NA;
+ assert !commit || !tx.txState().mvccEnabled(cctx) || tx.mvccCoordinatorCounter() != MvccUpdateVersion.COUNTER_NA;
boolean sync = tx.syncMode() == FULL_SYNC;
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
index 976a534..807b7c9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
@@ -25,7 +25,7 @@ import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxFinishRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -68,7 +68,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
private GridCacheVersion writeVer;
/** */
- private long mvccCrdCntr = TxMvccVersion.COUNTER_NA;
+ private long mvccCrdCntr = MvccUpdateVersion.COUNTER_NA;
/**
* Empty constructor required for {@link Externalizable}.
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 3bb4ea8..c0a2c4c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -61,7 +61,7 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTx
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
@@ -394,7 +394,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
tx.resolveTaskName(),
null,
keepBinary,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
if (retVal || txEntry.op() == TRANSFORM) {
if (!F.isEmpty(txEntry.entryProcessors())) {
@@ -495,7 +495,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
/*taskName*/null,
/*expiryPlc*/null,
/*keepBinary*/true,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
if (oldVal != null)
@@ -1235,7 +1235,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
assert crd != null : tx.topologyVersion();
if (crd.isLocal())
- tx.mvccCoordinatorCounter(cctx.coordinators().requestTxCounterOnCoordinator(tx.nearXidVersion()));
+ tx.mvccCoordinatorCounter(cctx.coordinators().requestTxCounterOnCoordinator(tx));
else {
IgniteInternalFuture<Long> crdCntrFut = cctx.coordinators().requestTxCounter(crd, tx);
@@ -1311,7 +1311,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
}
}
- assert !tx.txState().mvccEnabled(cctx) || !tx.onePhaseCommit() || tx.mvccCoordinatorCounter() != TxMvccVersion.COUNTER_NA;
+ assert !tx.txState().mvccEnabled(cctx) || !tx.onePhaseCommit() || tx.mvccCoordinatorCounter() != MvccUpdateVersion.COUNTER_NA;
int miniId = 0;
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
index 805c34d..b0148ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
@@ -34,7 +34,7 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -105,7 +105,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
private List<IgniteTxKey> nearWritesCacheMissed;
/** */
- private long mvccCrdCntr = TxMvccVersion.COUNTER_NA;
+ private long mvccCrdCntr = MvccUpdateVersion.COUNTER_NA;
/**
* Empty constructor required for {@link Externalizable}.
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 8359043..82facc1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -41,7 +41,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
@@ -82,7 +82,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
private ClusterNode mvccCrd;
/** */
- private long mvccCntr = TxMvccVersion.COUNTER_NA;
+ private MvccQueryVersion mvccVer;
/**
* @param cctx Context.
@@ -162,12 +162,13 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
return;
}
- IgniteInternalFuture<Long> cntrFut = cctx.shared().coordinators().requestQueryCounter(mvccCrd);
+ IgniteInternalFuture<MvccQueryVersion> cntrFut = cctx.shared().coordinators().requestQueryCounter(mvccCrd,
+ topVer.topologyVersion());
- cntrFut.listen(new IgniteInClosure<IgniteInternalFuture<Long>>() {
- @Override public void apply(IgniteInternalFuture<Long> fut) {
+ cntrFut.listen(new IgniteInClosure<IgniteInternalFuture<MvccQueryVersion>>() {
+ @Override public void apply(IgniteInternalFuture<MvccQueryVersion> fut) {
try {
- mvccCntr = fut.get();
+ mvccVer = fut.get();
map(keys,
Collections.<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>>emptyMap(),
@@ -246,10 +247,10 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (trackable)
cctx.mvcc().removeFuture(futId);
- if (mvccCntr != TxMvccVersion.COUNTER_NA) {
+ if (mvccVer != null) {
assert mvccCrd != null;
- cctx.shared().coordinators().ackQueryDone(mvccCrd, mvccCntr);
+ cctx.shared().coordinators().ackQueryDone(mvccCrd, mvccVer.counter());
}
cache().sendTtlUpdateRequest(expiryPlc);
@@ -345,7 +346,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
expiryPlc,
skipVals,
recovery,
- mvccCntr);
+ mvccVer);
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -402,7 +403,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
skipVals,
cctx.deploymentEnabled(),
recovery,
- mvccCntr);
+ mvccVer);
add(fut); // Append new future.
@@ -508,7 +509,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (readNoEntry) {
CacheDataRow row = cctx.mvccEnabled() ?
- cctx.offheap().readMvcc(cctx, key, topVer.topologyVersion(), mvccCntr) :
+ cctx.offheap().mvccRead(cctx, key, mvccVer) :
cctx.offheap().read(cctx, key);
if (row != null) {
@@ -552,7 +553,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
taskName,
expiryPlc,
!deserializeBinary,
- mvccCntr,
+ mvccVer,
null);
if (getRes != null) {
@@ -572,7 +573,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
taskName,
expiryPlc,
!deserializeBinary,
- mvccCntr);
+ mvccVer);
}
cache.context().evicts().touch(entry, topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 950295e..649798d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -41,7 +41,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
@@ -400,7 +399,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
taskName,
expiryPlc,
true,
- TxMvccVersion.COUNTER_NA,
+ null,
null); // TODO IGNITE-3478
if (res != null) {
@@ -420,7 +419,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
taskName,
expiryPlc,
true,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
colocated.context().evicts().touch(entry, topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 5de6d19..da27a85 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -61,9 +61,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtFuture;
@@ -81,6 +78,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSing
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
@@ -1555,7 +1553,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
expiry,
true,
- TxMvccVersion.COUNTER_NA,
+ null,
null);
if (getRes != null) {
@@ -1575,7 +1573,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
expiry,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA);
+ null);
}
// Entry was not in memory or in swap, so we remove it from cache.
@@ -2101,7 +2099,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
null,
req.keepBinary(),
- TxMvccVersion.COUNTER_NA);
+ null);
Object oldVal = null;
Object updatedVal = null;
@@ -2256,7 +2254,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
null,
req.keepBinary(),
- TxMvccVersion.COUNTER_NA);
+ null);
Object val = ctx.config().getInterceptor().onBeforePut(
new CacheLazyEntry(
@@ -2300,7 +2298,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
null,
req.keepBinary(),
- TxMvccVersion.COUNTER_NA);
+ null);
IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor()
.onBeforeRemove(new CacheLazyEntry(ctx, entry.key(), old, req.keepBinary()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index 5e87a05..5465ce6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -42,7 +42,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
@@ -516,7 +515,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA,
+ null,
null);
if (getRes != null) {
@@ -536,7 +535,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA);
+ null);
}
// Entry was not in memory or in swap, so we remove it from cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 848cf19..4d3b0dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -45,7 +45,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.CacheDistribu
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridLeanMap;
@@ -323,7 +322,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
expiryPlc,
skipVals,
recovery,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -386,7 +385,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
skipVals,
cctx.deploymentEnabled(),
recovery,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478.
+ null); // TODO IGNITE-3478.
add(fut); // Append new future.
@@ -457,7 +456,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA,
+ null,
null); // TODO IGNITE-3478
if (res != null) {
@@ -477,7 +476,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
}
@@ -596,7 +595,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA,
+ null,
null); // TODO IGNITE-3478
if (res != null) {
@@ -616,7 +615,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- TxMvccVersion.COUNTER_NA); // TODO IGNITE-3478
+ null); // TODO IGNITE-3478
}
// Entry was not in memory or in swap, so we remove it from cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/e40b4d9d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
index 4f343a8..44de8c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
@@ -34,7 +34,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.mvcc.TxMvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionable;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -108,7 +108,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
private long accessTtl;
/** */
- private long mvccCrdCntr = TxMvccVersion.COUNTER_NA;
+ private MvccQueryVersion mvccVer;
/**
* Empty constructor required for {@link Externalizable}.
@@ -149,7 +149,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
boolean skipVals,
boolean addDepInfo,
boolean recovery,
- long mvccCrdCntr
+ MvccQueryVersion mvccVer
) {
assert futId != null;
assert miniId != null;
@@ -178,7 +178,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.addDepInfo = addDepInfo;
- this.mvccCrdCntr = mvccCrdCntr;
+ this.mvccVer = mvccVer;
if (readThrough)
flags |= READ_THROUGH_FLAG_MASK;
@@ -196,8 +196,8 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
/**
* @return Counter.
*/
- public long mvccCoordinatorCounter() {
- return mvccCrdCntr;
+ public MvccQueryVersion mvccVersion() {
+ return mvccVer;
}
/**
@@ -395,7 +395,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
writer.incrementState();
case 9:
- if (!writer.writeLong("mvccCrdCntr", mvccCrdCntr))
+ if (!writer.writeMessage("mvccVer", mvccVer))
return false;
writer.incrementState();
@@ -495,7 +495,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 9:
- mvccCrdCntr = reader.readLong("mvccCrdCntr");
+ mvccVer = reader.readMessage("mvccVer");
if (!reader.isLastRead())
return false;