You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/08/30 12:24:39 UTC
[33/45] ignite git commit: IGNITE-4191: MVCC and transactional SQL
support. Joint multi-man-years efforts of Semen Boikov, Igor Seliverstov,
Alexander Paschenko, Igor Sapego, Sergey Kalashnikov, Roman Kondakov,
Pavel Kuznetsov, Ivan Pavlukhin, Andrey Mas
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 516dfb9..357f3d7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -90,6 +90,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartit
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionsStateValidator;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFutureAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.Latch;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotDiscoveryMessage;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -97,6 +98,7 @@ import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
+import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
@@ -647,7 +649,17 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
boolean crdNode = crd != null && crd.isLocal();
- exchCtx = new ExchangeContext(crdNode, this);
+ MvccCoordinator mvccCrd = firstEvtDiscoCache.mvccCoordinator();
+
+ boolean mvccCrdChange = mvccCrd != null &&
+ (initialVersion().equals(mvccCrd.topologyVersion()) || activateCluster());
+
+ // Mvcc coordinator should has been initialized before exchange context is created.
+ cctx.kernalContext().coordinators().updateCoordinator(mvccCrd);
+
+ exchCtx = new ExchangeContext(crdNode, mvccCrdChange, this);
+
+ cctx.kernalContext().coordinators().onExchangeStart(mvccCrd, exchCtx, crd);
assert state == null : state;
@@ -658,6 +670,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
if (exchLog.isInfoEnabled()) {
exchLog.info("Started exchange init [topVer=" + topVer +
+ ", mvccCrd=" + mvccCrd +
+ ", mvccCrdChange=" + mvccCrdChange +
", crd=" + crdNode +
", evt=" + IgniteUtils.gridEventName(firstDiscoEvt.type()) +
", evtNode=" + firstDiscoEvt.eventNode().id() +
@@ -745,7 +759,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
}
}
- updateTopologies(crdNode);
+ updateTopologies(crd, crdNode, cctx.coordinators().currentCoordinator());
switch (exchange) {
case ALL: {
@@ -895,10 +909,12 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/**
* Updates topology versions and discovery caches on all topologies.
*
+ * @param exchCrd Exchange coordinator node.
* @param crd Coordinator flag.
+ * @param mvccCrd Mvcc coordinator.
* @throws IgniteCheckedException If failed.
*/
- private void updateTopologies(boolean crd) throws IgniteCheckedException {
+ private void updateTopologies(ClusterNode exchCrd, boolean crd, MvccCoordinator mvccCrd) throws IgniteCheckedException {
for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
if (grp.isLocal())
continue;
@@ -925,12 +941,18 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
top.updateTopologyVersion(
this,
events().discoveryCache(),
+ mvccCrd,
updSeq,
cacheGroupStopping(grp.groupId()));
}
- for (GridClientPartitionTopology top : cctx.exchange().clientTopologies())
- top.updateTopologyVersion(this, events().discoveryCache(), -1, cacheGroupStopping(top.groupId()));
+ for (GridClientPartitionTopology top : cctx.exchange().clientTopologies()) {
+ top.updateTopologyVersion(this,
+ events().discoveryCache(),
+ mvccCrd,
+ -1,
+ cacheGroupStopping(top.groupId()));
+ }
}
/**
@@ -1570,6 +1592,12 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
msg.partitionHistoryCounters(partHistReserved0);
}
+ if (exchCtx.newMvccCoordinator() && cctx.coordinators().currentCoordinatorId().equals(node.id())) {
+ Map<UUID, GridLongList> activeQueries = exchCtx.activeQueries();
+
+ msg.activeQueries(activeQueries != null ? activeQueries.get(cctx.localNodeId()) : null);
+ }
+
if ((stateChangeExchange() || dynamicCacheStartExchange()) && exchangeLocE != null)
msg.setError(exchangeLocE);
else if (localJoinExchange())
@@ -1813,6 +1841,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
if (!cctx.localNode().isClient())
tryToPerformLocalSnapshotOperation();
+ if (err == null)
+ cctx.coordinators().onExchangeDone(exchCtx.newMvccCoordinator(), exchCtx.events().discoveryCache(),
+ exchCtx.activeQueries());
+
cctx.cache().onExchangeDone(initialVersion(), exchActions, err);
cctx.exchange().onExchangeDone(res, initialVersion(), err);
@@ -2319,6 +2351,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
*/
private void processSingleMessage(UUID nodeId, GridDhtPartitionsSingleMessage msg) {
if (msg.client()) {
+ if (msg.activeQueries() != null)
+ cctx.coordinators().processClientActiveQueries(nodeId, msg.activeQueries());
+
waitAndReplyToNode(nodeId, msg);
return;
@@ -2856,6 +2891,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
for (Map.Entry<UUID, GridDhtPartitionsSingleMessage> e : msgs.entrySet()) {
GridDhtPartitionsSingleMessage msg = e.getValue();
+ if (exchCtx.newMvccCoordinator())
+ exchCtx.addActiveQueries(e.getKey(), msg.activeQueries());
+
// Apply update counters after all single messages are received.
for (Map.Entry<Integer, GridDhtPartitionMap> entry : msg.partitions().entrySet()) {
Integer grpId = entry.getKey();
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
index 804cc03..77d7eab 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
@@ -108,6 +109,9 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
*/
private GridDhtPartitionsFullMessage finishMsg;
+ /** */
+ private GridLongList activeQryTrackers;
+
/**
* Required by {@link Externalizable}.
*/
@@ -132,6 +136,20 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
}
/**
+ * @return Active queries started with previous coordinator.
+ */
+ GridLongList activeQueries() {
+ return activeQryTrackers;
+ }
+
+ /**
+ * @param activeQrys Active queries started with previous coordinator.
+ */
+ void activeQueries(GridLongList activeQrys) {
+ this.activeQryTrackers = activeQrys;
+ }
+
+ /**
* @param finishMsg Exchange finish message (used to restore exchange state on new coordinator).
*/
void finishMessage(GridDhtPartitionsFullMessage finishMsg) {
@@ -331,7 +349,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
byte[] partsBytes0 = null;
byte[] partCntrsBytes0 = null;
byte[] partHistCntrsBytes0 = null;
- byte[] partSizesBytes0 = null;
+ byte[] partsSizesBytes0 = null;
byte[] errBytes0 = null;
if (parts != null && partsBytes == null)
@@ -344,7 +362,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
partHistCntrsBytes0 = U.marshal(ctx, partHistCntrs);
if (partsSizes != null && partsSizesBytes == null)
- partSizesBytes0 = U.marshal(ctx, partsSizes);
+ partsSizesBytes0 = U.marshal(ctx, partsSizes);
if (err != null && errBytes == null)
errBytes0 = U.marshal(ctx, err);
@@ -356,13 +374,13 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
byte[] partsBytesZip = U.zip(partsBytes0);
byte[] partCntrsBytesZip = U.zip(partCntrsBytes0);
byte[] partHistCntrsBytesZip = U.zip(partHistCntrsBytes0);
- byte[] partSizesBytesZip = U.zip(partSizesBytes0);
+ byte[] partsSizesBytesZip = U.zip(partsSizesBytes0);
byte[] exBytesZip = U.zip(errBytes0);
partsBytes0 = partsBytesZip;
partCntrsBytes0 = partCntrsBytesZip;
partHistCntrsBytes0 = partHistCntrsBytesZip;
- partSizesBytes0 = partSizesBytesZip;
+ partsSizesBytes0 = partsSizesBytesZip;
errBytes0 = exBytesZip;
compressed(true);
@@ -375,7 +393,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
partsBytes = partsBytes0;
partCntrsBytes = partCntrsBytes0;
partHistCntrsBytes = partHistCntrsBytes0;
- partsSizesBytes = partSizesBytes0;
+ partsSizesBytes = partsSizesBytes0;
errBytes = errBytes0;
}
}
@@ -456,54 +474,60 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
switch (writer.state()) {
case 5:
- if (!writer.writeBoolean("client", client))
+ if (!writer.writeMessage("activeQryTrackers", activeQryTrackers))
return false;
writer.incrementState();
case 6:
- if (!writer.writeMap("dupPartsData", dupPartsData, MessageCollectionItemType.INT, MessageCollectionItemType.INT))
+ if (!writer.writeBoolean("client", client))
return false;
writer.incrementState();
case 7:
- if (!writer.writeByteArray("errBytes", errBytes))
+ if (!writer.writeMap("dupPartsData", dupPartsData, MessageCollectionItemType.INT, MessageCollectionItemType.INT))
return false;
writer.incrementState();
case 8:
- if (!writer.writeMessage("finishMsg", finishMsg))
+ if (!writer.writeByteArray("errBytes", errBytes))
return false;
writer.incrementState();
case 9:
- if (!writer.writeCollection("grpsAffRequest", grpsAffRequest, MessageCollectionItemType.INT))
+ if (!writer.writeMessage("finishMsg", finishMsg))
return false;
writer.incrementState();
case 10:
- if (!writer.writeByteArray("partCntrsBytes", partCntrsBytes))
+ if (!writer.writeCollection("grpsAffRequest", grpsAffRequest, MessageCollectionItemType.INT))
return false;
writer.incrementState();
case 11:
- if (!writer.writeByteArray("partHistCntrsBytes", partHistCntrsBytes))
+ if (!writer.writeByteArray("partCntrsBytes", partCntrsBytes))
return false;
writer.incrementState();
case 12:
- if (!writer.writeByteArray("partsBytes", partsBytes))
+ if (!writer.writeByteArray("partHistCntrsBytes", partHistCntrsBytes))
return false;
writer.incrementState();
case 13:
+ if (!writer.writeByteArray("partsBytes", partsBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 14:
if (!writer.writeByteArray("partsSizesBytes", partsSizesBytes))
return false;
@@ -525,7 +549,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
switch (reader.state()) {
case 5:
- client = reader.readBoolean("client");
+ activeQryTrackers = reader.readMessage("activeQryTrackers");
if (!reader.isLastRead())
return false;
@@ -533,7 +557,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 6:
- dupPartsData = reader.readMap("dupPartsData", MessageCollectionItemType.INT, MessageCollectionItemType.INT, false);
+ client = reader.readBoolean("client");
if (!reader.isLastRead())
return false;
@@ -541,7 +565,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 7:
- errBytes = reader.readByteArray("errBytes");
+ dupPartsData = reader.readMap("dupPartsData", MessageCollectionItemType.INT, MessageCollectionItemType.INT, false);
if (!reader.isLastRead())
return false;
@@ -549,7 +573,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 8:
- finishMsg = reader.readMessage("finishMsg");
+ errBytes = reader.readByteArray("errBytes");
if (!reader.isLastRead())
return false;
@@ -557,7 +581,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 9:
- grpsAffRequest = reader.readCollection("grpsAffRequest", MessageCollectionItemType.INT);
+ finishMsg = reader.readMessage("finishMsg");
if (!reader.isLastRead())
return false;
@@ -565,7 +589,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 10:
- partCntrsBytes = reader.readByteArray("partCntrsBytes");
+ grpsAffRequest = reader.readCollection("grpsAffRequest", MessageCollectionItemType.INT);
if (!reader.isLastRead())
return false;
@@ -573,7 +597,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 11:
- partHistCntrsBytes = reader.readByteArray("partHistCntrsBytes");
+ partCntrsBytes = reader.readByteArray("partCntrsBytes");
if (!reader.isLastRead())
return false;
@@ -581,7 +605,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 12:
- partsBytes = reader.readByteArray("partsBytes");
+ partHistCntrsBytes = reader.readByteArray("partHistCntrsBytes");
if (!reader.isLastRead())
return false;
@@ -589,6 +613,14 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
case 13:
+ partsBytes = reader.readByteArray("partsBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 14:
partsSizesBytes = reader.readByteArray("partsSizesBytes");
if (!reader.isLastRead())
@@ -607,7 +639,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 14;
+ return 15;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
new file mode 100644
index 0000000..1648da9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.near;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.util.typedef.CIX1;
+
+/** */
+public class AckCoordinatorOnRollback extends CIX1<IgniteInternalFuture<IgniteInternalTx>> {
+ /** */
+ private static final long serialVersionUID = 8172699207968328284L;
+
+ /** */
+ private final GridNearTxLocal tx;
+
+ /**
+ * @param tx Transaction.
+ */
+ public AckCoordinatorOnRollback(GridNearTxLocal tx) {
+ this.tx = tx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void applyx(IgniteInternalFuture<IgniteInternalTx> fut) throws IgniteCheckedException {
+ assert fut.isDone();
+
+ MvccQueryTracker tracker = tx.mvccQueryTracker();
+ MvccSnapshot mvccSnapshot = tx.mvccSnapshot();
+
+ if (tracker != null) // Optimistic tx.
+ tracker.onDone(tx, false);
+ else if (mvccSnapshot != null)// Pessimistic tx.
+ tx.context().coordinators().ackTxRollback(mvccSnapshot);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/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 ee06518..7cf5a6b 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
@@ -319,7 +319,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName == null ? 0 : taskName.hashCode(),
expiryPlc,
skipVals,
- recovery);
+ recovery,
+ null); // TODO IGNITE-7371
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -382,7 +383,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
true,
skipVals,
cctx.deploymentEnabled(),
- recovery);
+ recovery,
+ null); // TODO IGNITE-7371
add(fut); // Append new future.
@@ -455,7 +457,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- null);
+ null,
+ null); // TODO IGNITE-7371
if (res != null) {
v = res.value();
@@ -473,7 +476,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
null,
taskName,
expiryPlc,
- !deserializeBinary);
+ !deserializeBinary,
+ null); // TODO IGNITE-7371
}
}
@@ -599,7 +603,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- null);
+ null,
+ null); // TODO IGNITE-7371
if (res != null) {
v = res.value();
@@ -617,7 +622,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
null,
taskName,
expiryPlc,
- !deserializeBinary);
+ !deserializeBinary,
+ null); // TODO IGNITE-7371
}
// Entry was not in memory or in swap, so we remove it from cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/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 dcb167d..c891bfb 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,6 +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.MvccSnapshot;
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;
@@ -45,6 +46,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemTy
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Get request. Responsible for obtaining entry from primary node. 'Near' means 'Initiating node' here, not 'Near Cache'.
@@ -106,6 +108,9 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
/** TTL for read operation. */
private long accessTtl;
+ /** */
+ private MvccSnapshot mvccSnapshot;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -128,6 +133,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
* @param createTtl New TTL to set after entry is created, -1 to leave unchanged.
* @param accessTtl New TTL to set after entry is accessed, -1 to leave unchanged.
* @param addDepInfo Deployment info.
+ * @param mvccSnapshot Mvcc snapshot.
*/
public GridNearGetRequest(
int cacheId,
@@ -144,7 +150,8 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
boolean addReader,
boolean skipVals,
boolean addDepInfo,
- boolean recovery
+ boolean recovery,
+ @Nullable MvccSnapshot mvccSnapshot
) {
assert futId != null;
assert miniId != null;
@@ -173,6 +180,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.addDepInfo = addDepInfo;
+ this.mvccSnapshot = mvccSnapshot;
if (readThrough)
flags |= READ_THROUGH_FLAG_MASK;
@@ -188,6 +196,13 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
}
/**
+ * @return Mvcc version.
+ */
+ @Nullable public MvccSnapshot mvccSnapshot() {
+ return mvccSnapshot;
+ }
+
+ /**
* @return Future ID.
*/
public IgniteUuid futureId() {
@@ -382,30 +397,36 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
writer.incrementState();
case 9:
- if (!writer.writeCollection("readersFlags", readersFlags, MessageCollectionItemType.BOOLEAN))
+ if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
return false;
writer.incrementState();
case 10:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeCollection("readersFlags", readersFlags, MessageCollectionItemType.BOOLEAN))
return false;
writer.incrementState();
case 11:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 12:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 13:
+ if (!writer.writeMessage("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ case 14:
if (!writer.writeMessage("ver", ver))
return false;
@@ -476,7 +497,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 9:
- readersFlags = reader.readCollection("readersFlags", MessageCollectionItemType.BOOLEAN);
+ mvccSnapshot = reader.readMessage("mvccSnapshot");
if (!reader.isLastRead())
return false;
@@ -484,7 +505,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 10:
- subjId = reader.readUuid("subjId");
+ readersFlags = reader.readCollection("readersFlags", MessageCollectionItemType.BOOLEAN);
if (!reader.isLastRead())
return false;
@@ -492,7 +513,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 11:
- taskNameHash = reader.readInt("taskNameHash");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
@@ -500,7 +521,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 12:
- topVer = reader.readMessage("topVer");
+ taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
return false;
@@ -508,6 +529,14 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 13:
+ topVer = reader.readMessage("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 14:
ver = reader.readMessage("ver");
if (!reader.isLastRead())
@@ -527,7 +556,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 14;
+ return 15;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
index a997c45..140c1d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
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;
@@ -50,7 +51,6 @@ import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.P1;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -59,6 +59,7 @@ import org.apache.ignite.lang.IgniteReducer;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.noCoordinatorError;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;
@@ -184,10 +185,20 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
tx.removeMapping(m.primary().id());
}
+ prepareError(e);
+ }
+
+ /**
+ * @param e Error.
+ */
+ private void prepareError(Throwable e) {
ERR_UPD.compareAndSet(this, null, e);
if (keyLockFut != null)
keyLockFut.onDone(e);
+
+ if (mvccVerFut != null)
+ mvccVerFut.onDone();
}
/** {@inheritDoc} */
@@ -229,7 +240,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
// Avoid iterator creation.
for (int i = 0; i < size; i++) {
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = future(i);
+ IgniteInternalFuture fut = future(i);
if (!isMini(fut))
continue;
@@ -338,11 +349,25 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
boolean hasNearCache = false;
+ MvccCoordinator mvccCrd = null;
+
for (IgniteTxEntry write : writes) {
map(write, topVer, mappings, txMapping, remap, topLocked);
- if (write.context().isNear())
+ GridCacheContext cctx = write.context();
+
+ if (cctx.isNear())
hasNearCache = true;
+
+ if (cctx.mvccEnabled() && mvccCrd == null) {
+ mvccCrd = cctx.affinity().mvccCoordinator(topVer);
+
+ if (mvccCrd == null) {
+ onDone(noCoordinatorError(topVer));
+
+ return;
+ }
+ }
}
for (IgniteTxEntry read : reads)
@@ -358,6 +383,8 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
return;
}
+ assert !tx.txState().mvccEnabled(cctx) || mvccCrd != null || F.isEmpty(writes);
+
tx.addEntryMapping(mappings.values());
cctx.mvcc().recheckPendingLocks();
@@ -369,23 +396,32 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
MiniFuture locNearEntriesFut = null;
+ int lockCnt = keyLockFut != null ? 1 : 0;
+
// Create futures in advance to have all futures when process {@link GridNearTxPrepareResponse#clientRemapVersion}.
for (GridDistributedTxMapping m : mappings.values()) {
assert !m.empty();
MiniFuture fut = new MiniFuture(this, m, ++miniId);
- add(fut);
+ lockCnt++;
+
+ add((IgniteInternalFuture)fut);
if (m.primary().isLocal() && m.hasNearCacheEntries() && m.hasColocatedCacheEntries()) {
assert locNearEntriesFut == null;
locNearEntriesFut = fut;
- add(new MiniFuture(this, m, ++miniId));
+ add((IgniteInternalFuture)new MiniFuture(this, m, ++miniId));
+
+ lockCnt++;
}
}
+ if (mvccCrd != null)
+ initMvccVersionFuture(lockCnt, remap);
+
Collection<IgniteInternalFuture<?>> futs = (Collection)futures();
Iterator<IgniteInternalFuture<?>> it = futs.iterator();
@@ -639,7 +675,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
if (keyLockFut == null) {
keyLockFut = new KeyLockFuture();
- add(keyLockFut);
+ add((IgniteInternalFuture)keyLockFut);
}
keyLockFut.addLockKey(entry.txKey());
@@ -696,20 +732,20 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
Collection<String> futs = F.viewReadOnly(futures(),
new C1<IgniteInternalFuture<?>, String>() {
@Override public String apply(IgniteInternalFuture<?> f) {
- return "[node=" + ((MiniFuture)f).primary().id() +
- ", loc=" + ((MiniFuture)f).primary().isLocal() +
- ", done=" + f.isDone() + "]";
- }
- },
- new P1<IgniteInternalFuture<?>>() {
- @Override public boolean apply(IgniteInternalFuture<?> f) {
- return isMini(f);
+ if (isMini(f)) {
+ return "[node=" + ((MiniFuture)f).primary().id() +
+ ", loc=" + ((MiniFuture)f).primary().isLocal() +
+ ", done=" + f.isDone() +
+ ", err=" + f.error() + "]";
+ }
+ else
+ return f.toString();
}
});
return S.toString(GridNearOptimisticSerializableTxPrepareFuture.class, this,
"innerFuts", futs,
- "keyLockFut", keyLockFut,
+ "remap", remapFut != null,
"tx", tx,
"super", super.toString());
}
@@ -760,7 +796,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
*/
private static class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
/** Receive result flag updater. */
- private static AtomicIntegerFieldUpdater<MiniFuture> RCV_RES_UPD =
+ private static final AtomicIntegerFieldUpdater<MiniFuture> RCV_RES_UPD =
AtomicIntegerFieldUpdater.newUpdater(MiniFuture.class, "rcvRes");
/** */
@@ -924,7 +960,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
err0.retryReadyFuture(affFut);
- ERR_UPD.compareAndSet(parent, null, err0);
+ parent.prepareError(err0);
onDone(err0);
}
@@ -935,7 +971,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
parent);
}
- ERR_UPD.compareAndSet(parent, null, e);
+ parent.prepareError(e);
onDone(e);
}
@@ -950,6 +986,9 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
// Finish this mini future (need result only on client node).
onDone(parent.cctx.kernalContext().clientNode() ? res : null);
+
+ if (parent.mvccVerFut != null)
+ parent.mvccVerFut.onLockReceived();
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 566789c..06d7a8c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
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;
@@ -68,6 +69,7 @@ import org.apache.ignite.transactions.TransactionTimeoutException;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.noCoordinatorError;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;
@@ -220,7 +222,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
int size = futuresCountNoLock();
for (int i = 0; i < size; i++) {
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = future(i);
+ IgniteInternalFuture fut = future(i);
if (isMini(fut) && !fut.isDone()) {
MiniFuture miniFut = (MiniFuture)fut;
@@ -254,7 +256,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
// Avoid iterator creation.
for (int i = size - 1; i >= 0; i--) {
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = future(i);
+ IgniteInternalFuture fut = future(i);
if (!isMini(fut))
continue;
@@ -380,6 +382,18 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
tx.colocatedLocallyMapped(true);
}
+ if (write.context().mvccEnabled()) {
+ MvccCoordinator mvccCrd = write.context().affinity().mvccCoordinator(topVer);
+
+ if (mvccCrd == null) {
+ onDone(noCoordinatorError(topVer));
+
+ return;
+ }
+
+ initMvccVersionFuture(keyLockFut != null ? 2 : 1, remap);
+ }
+
if (keyLockFut != null)
keyLockFut.onAllKeysAdded();
@@ -424,6 +438,8 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
boolean hasNearCache = false;
+ MvccCoordinator mvccCrd = null;
+
for (IgniteTxEntry write : writes) {
write.clearEntryReadVersion();
@@ -433,6 +449,16 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
// an exception occurred while transaction mapping, stop further processing
break;
+ if (write.context().mvccEnabled() && mvccCrd == null) {
+ mvccCrd = write.context().affinity().mvccCoordinator(topVer);
+
+ if (mvccCrd == null) {
+ onDone(noCoordinatorError(topVer));
+
+ break;
+ }
+ }
+
if (write.context().isNear())
hasNearCache = true;
@@ -472,6 +498,11 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
return;
}
+ assert !tx.txState().mvccEnabled(cctx) || mvccCrd != null;
+
+ if (mvccCrd != null)
+ initMvccVersionFuture(keyLockFut != null ? 2 : 1, remap);
+
if (keyLockFut != null)
keyLockFut.onAllKeysAdded();
@@ -495,8 +526,12 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
private void proceedPrepare(final Queue<GridDistributedTxMapping> mappings) {
final GridDistributedTxMapping m = mappings.poll();
- if (m == null)
+ if (m == null) {
+ if (mvccVerFut != null)
+ mvccVerFut.onLockReceived();
+
return;
+ }
proceedPrepare(m, mappings);
}
@@ -562,7 +597,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
req.miniId(fut.futureId());
- add(fut); // Append new future.
+ add((IgniteInternalFuture)fut); // Append new future.
if (n.isLocal()) {
assert !(m.hasColocatedCacheEntries() && m.hasNearCacheEntries()) : m;
@@ -679,7 +714,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
if (keyLockFut == null) {
keyLockFut = new KeyLockFuture();
- add(keyLockFut);
+ add((IgniteInternalFuture)keyLockFut);
}
keyLockFut.addLockKey(entry.txKey());
@@ -738,7 +773,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
int size = futuresCountNoLock();
for (int i = 0; i < size; i++) {
- IgniteInternalFuture<GridNearTxPrepareResponse> fut = future(i);
+ IgniteInternalFuture fut = future(i);
if (isMini(fut) && !fut.isDone()) {
MiniFuture miniFut = (MiniFuture)fut;
@@ -756,7 +791,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
}
}
- add(new GridEmbeddedFuture<>(new IgniteBiClosure<TxDeadlock, Exception, GridNearTxPrepareResponse>() {
+ add(new GridEmbeddedFuture<>(new IgniteBiClosure<TxDeadlock, Exception, Object>() {
@Override public GridNearTxPrepareResponse apply(TxDeadlock deadlock, Exception e) {
if (e != null)
U.warn(log, "Failed to detect deadlock.", e);
@@ -787,7 +822,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
/** {@inheritDoc} */
@Override public void addDiagnosticRequest(IgniteDiagnosticPrepareContext ctx) {
if (!isDone()) {
- for (IgniteInternalFuture<GridNearTxPrepareResponse> fut : futures()) {
+ for (IgniteInternalFuture fut : futures()) {
if (!fut.isDone()) {
if (fut instanceof MiniFuture) {
MiniFuture miniFut = (MiniFuture)fut;
@@ -843,19 +878,22 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
@Override public String toString() {
Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
@Override public String apply(IgniteInternalFuture<?> f) {
- return "[node=" + ((MiniFuture)f).node().id() +
- ", loc=" + ((MiniFuture)f).node().isLocal() +
- ", done=" + f.isDone() + "]";
+ if (isMini(f)) {
+ return "[node=" + ((MiniFuture)f).node().id() +
+ ", loc=" + ((MiniFuture)f).node().isLocal() +
+ ", done=" + f.isDone() + "]";
+ }
+ else
+ return f.toString();
}
- }, new P1<IgniteInternalFuture<GridNearTxPrepareResponse>>() {
- @Override public boolean apply(IgniteInternalFuture<GridNearTxPrepareResponse> fut) {
+ }, new P1<IgniteInternalFuture<Object>>() {
+ @Override public boolean apply(IgniteInternalFuture<Object> fut) {
return isMini(fut);
}
});
return S.toString(GridNearOptimisticTxPrepareFuture.class, this,
"innerFuts", futs,
- "keyLockFut", keyLockFut,
"tx", tx,
"super", super.toString());
}
@@ -1001,6 +1039,8 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
// Proceed prepare before finishing mini future.
if (mappings != null)
parent.proceedPrepare(mappings);
+ else if (parent.mvccVerFut != null)
+ parent.mvccVerFut.onLockReceived();
// Finish this mini future.
onDone((GridNearTxPrepareResponse)null);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
index 45f8941..6f541d3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
@@ -18,19 +18,20 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
import java.util.Collection;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshotResponseListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
-import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.lang.GridAbsClosureX;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -42,9 +43,20 @@ import org.jetbrains.annotations.Nullable;
*/
public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearTxPrepareFutureAdapter {
/** */
+ private static final long serialVersionUID = 7460376140787916619L;
+
+ /** */
+ private static final AtomicIntegerFieldUpdater<MvccSnapshotFutureExt> LOCK_CNT_UPD =
+ AtomicIntegerFieldUpdater.newUpdater(MvccSnapshotFutureExt.class, "lockCnt");
+
+ /** */
@GridToStringExclude
protected KeyLockFuture keyLockFut;
+ /** */
+ @GridToStringExclude
+ protected MvccSnapshotFutureExt mvccVerFut;
+
/**
* @param cctx Context.
* @param tx Transaction.
@@ -73,7 +85,7 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
}
if (keyLockFut != null)
- add(keyLockFut);
+ add((IgniteInternalFuture)keyLockFut);
}
}
@@ -202,6 +214,29 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
protected abstract void prepare0(boolean remap, boolean topLocked);
/**
+ * @param lockCnt Expected number of lock responses.
+ * @param remap Remap flag.
+ */
+ @SuppressWarnings("unchecked")
+ final void initMvccVersionFuture(int lockCnt, boolean remap) {
+ if (!remap) {
+ mvccVerFut = new MvccSnapshotFutureExt();
+
+ mvccVerFut.init(lockCnt);
+
+ if (keyLockFut != null)
+ keyLockFut.listen(mvccVerFut);
+
+ add((IgniteInternalFuture)mvccVerFut);
+ }
+ else {
+ assert mvccVerFut != null;
+
+ mvccVerFut.init(lockCnt);
+ }
+ }
+
+ /**
* @param e Exception.
* @param timedOut {@code True} if timed out.
*/
@@ -223,7 +258,7 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
/**
* Keys lock future.
*/
- protected static class KeyLockFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
+ protected static class KeyLockFuture extends GridFutureAdapter<Void> {
/** */
@GridToStringInclude
protected Collection<IgniteTxKey> lockKeys = new GridConcurrentHashSet<>();
@@ -258,24 +293,20 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
checkLocks();
}
- /**
- * @return {@code True} if all locks are owned.
- */
- private boolean checkLocks() {
+ /** */
+ private void checkLocks() {
boolean locked = lockKeys.isEmpty();
if (locked && allKeysAdded) {
if (log.isDebugEnabled())
log.debug("All locks are acquired for near prepare future: " + this);
- onDone((GridNearTxPrepareResponse)null);
+ onDone((Void)null);
}
else {
if (log.isDebugEnabled())
log.debug("Still waiting for locks [fut=" + this + ", keys=" + lockKeys + ']');
}
-
- return locked;
}
/** {@inheritDoc} */
@@ -283,4 +314,82 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
return S.toString(KeyLockFuture.class, this, super.toString());
}
}
+
+ /**
+ *
+ */
+ class MvccSnapshotFutureExt extends GridFutureAdapter<Void> implements MvccSnapshotResponseListener, IgniteInClosure<IgniteInternalFuture<Void>> {
+ /** */
+ private static final long serialVersionUID = 5883078648683911226L;
+
+ /** */
+ volatile int lockCnt;
+
+ /** {@inheritDoc} */
+ @Override public void apply(IgniteInternalFuture<Void> keyLockFut) {
+ try {
+ keyLockFut.get();
+
+ onLockReceived();
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("MvccSnapshotFutureExt ignores key lock future failure: " + e);
+ }
+ }
+
+ /**
+ * @param lockCnt Expected number of lock responses.
+ */
+ void init(int lockCnt) {
+ assert lockCnt > 0;
+
+ this.lockCnt = lockCnt;
+
+ assert !isDone();
+ }
+
+ /** */
+ void onLockReceived() {
+ int remaining = LOCK_CNT_UPD.decrementAndGet(this);
+
+ assert remaining >= 0 : remaining;
+
+ if (remaining == 0) {
+ try {
+ MvccSnapshot snapshot = cctx.coordinators().tryRequestSnapshotLocal(tx);
+
+ if (snapshot != null)
+ onResponse(snapshot);
+ else
+ cctx.coordinators().requestSnapshotAsync(tx, this);
+ }
+ catch (ClusterTopologyCheckedException e) {
+ onError(e);
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onResponse(MvccSnapshot res) {
+ tx.mvccSnapshot(res);
+
+ onDone();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onError(IgniteCheckedException e) {
+ if (e instanceof ClusterTopologyCheckedException)
+ ((ClusterTopologyCheckedException)e).retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion()));
+
+ ERR_UPD.compareAndSet(GridNearOptimisticTxPrepareFutureAdapter.this, null, e);
+
+ onDone();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccSnapshotFutureExt.class, this, super.toString());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index ffde0f3..dbf54c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -36,6 +36,9 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshotResponseListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
@@ -43,11 +46,13 @@ import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.noCoordinatorError;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;
@@ -55,6 +60,9 @@ import static org.apache.ignite.transactions.TransactionState.PREPARING;
*
*/
public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureAdapter {
+ /** */
+ private static final long serialVersionUID = 4014479758215810181L;
+
/**
* @param cctx Context.
* @param tx Transaction.
@@ -80,17 +88,19 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
boolean found = false;
for (IgniteInternalFuture<?> fut : futures()) {
- MiniFuture f = (MiniFuture)fut;
+ if (fut instanceof MiniFuture) {
+ MiniFuture f = (MiniFuture)fut;
- if (f.primary().id().equals(nodeId)) {
- ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Remote node left grid: " +
- nodeId);
+ if (f.primary().id().equals(nodeId)) {
+ ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Remote node left grid: " +
+ nodeId);
- e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion()));
+ e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion()));
- f.onNodeLeft(e);
+ f.onNodeLeft(e);
- found = true;
+ found = true;
+ }
}
}
@@ -142,13 +152,17 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
// Avoid iterator creation.
for (int i = 0; i < size; i++) {
- MiniFuture mini = (MiniFuture)future(i);
+ IgniteInternalFuture fut = future(i);
- if (mini.futureId() == miniId) {
- if (!mini.isDone())
- return mini;
- else
- return null;
+ if (fut instanceof MiniFuture) {
+ MiniFuture mini = (MiniFuture)fut;
+
+ if (mini.futureId() == miniId) {
+ if (!mini.isDone())
+ return mini;
+ else
+ return null;
+ }
}
}
}
@@ -217,6 +231,8 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
true,
tx.activeCachesDeploymentEnabled());
+ req.queryUpdate(m.queryUpdate());
+
for (IgniteTxEntry txEntry : writes) {
if (txEntry.op() == TRANSFORM)
req.addDhtVersion(txEntry.txKey(), null);
@@ -231,6 +247,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
* @param miniId Mini future ID.
* @param nearEntries {@code True} if prepare near cache entries.
*/
+ @SuppressWarnings("unchecked")
private void prepareLocal(GridNearTxPrepareRequest req,
GridDistributedTxMapping m,
int miniId,
@@ -239,7 +256,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
req.miniId(fut.futureId());
- add(fut);
+ add((IgniteInternalFuture)fut);
IgniteInternalFuture<GridNearTxPrepareResponse> prepFut = nearEntries ?
cctx.tm().txHandler().prepareNearTxLocal(req) :
@@ -260,6 +277,7 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
/**
*
*/
+ @SuppressWarnings("unchecked")
private void preparePessimistic() {
Map<UUID, GridDistributedTxMapping> mappings = new HashMap<>();
@@ -267,48 +285,77 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
GridDhtTxMapping txMapping = new GridDhtTxMapping();
+ boolean queryMapped = false;
+
+ for (GridDistributedTxMapping m : F.view(tx.mappings().mappings(), CU.FILTER_QUERY_MAPPING)) {
+ GridDistributedTxMapping nodeMapping = mappings.get(m.primary().id());
+
+ if(nodeMapping == null)
+ mappings.put(m.primary().id(), m);
+
+ txMapping.addMapping(F.asList(m.primary()));
+
+ queryMapped = true;
+ }
+
+ MvccCoordinator mvccCrd = null;
+
boolean hasNearCache = false;
- for (IgniteTxEntry txEntry : tx.allEntries()) {
- txEntry.clearEntryReadVersion();
+ if (!queryMapped) {
+ for (IgniteTxEntry txEntry : tx.allEntries()) {
+ txEntry.clearEntryReadVersion();
- GridCacheContext cacheCtx = txEntry.context();
+ GridCacheContext cacheCtx = txEntry.context();
- if (cacheCtx.isNear())
- hasNearCache = true;
+ if (cacheCtx.isNear())
+ hasNearCache = true;
- List<ClusterNode> nodes;
+ List<ClusterNode> nodes;
- if (!cacheCtx.isLocal()) {
- GridDhtPartitionTopology top = cacheCtx.topology();
+ if (!cacheCtx.isLocal()) {
+ GridDhtPartitionTopology top = cacheCtx.topology();
- nodes = top.nodes(cacheCtx.affinity().partition(txEntry.key()), topVer);
- }
- else
- nodes = cacheCtx.affinity().nodesByKey(txEntry.key(), topVer);
+ nodes = top.nodes(cacheCtx.affinity().partition(txEntry.key()), topVer);
+ }
+ else
+ nodes = cacheCtx.affinity().nodesByKey(txEntry.key(), topVer);
- if (F.isEmpty(nodes)) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys to nodes (partition " +
- "is not mapped to any node) [key=" + txEntry.key() +
- ", partition=" + cacheCtx.affinity().partition(txEntry.key()) + ", topVer=" + topVer + ']'));
+ if (tx.mvccSnapshot() == null && mvccCrd == null && cacheCtx.mvccEnabled()) {
+ mvccCrd = cacheCtx.affinity().mvccCoordinator(topVer);
- return;
- }
+ if (mvccCrd == null) {
+ onDone(noCoordinatorError(topVer));
+
+ return;
+ }
+ }
+
+ if (F.isEmpty(nodes)) {
+ onDone(new ClusterTopologyServerNotFoundException("Failed to map keys to nodes (partition " +
+ "is not mapped to any node) [key=" + txEntry.key() +
+ ", partition=" + cacheCtx.affinity().partition(txEntry.key()) + ", topVer=" + topVer + ']'));
+
+ return;
+ }
- ClusterNode primary = nodes.get(0);
+ ClusterNode primary = nodes.get(0);
- GridDistributedTxMapping nodeMapping = mappings.get(primary.id());
+ GridDistributedTxMapping nodeMapping = mappings.get(primary.id());
- if (nodeMapping == null)
- mappings.put(primary.id(), nodeMapping = new GridDistributedTxMapping(primary));
+ if (nodeMapping == null)
+ mappings.put(primary.id(), nodeMapping = new GridDistributedTxMapping(primary));
- txEntry.nodeId(primary.id());
+ txEntry.nodeId(primary.id());
- nodeMapping.add(txEntry);
+ nodeMapping.add(txEntry);
- txMapping.addMapping(nodes);
+ txMapping.addMapping(nodes);
+ }
}
+ assert !tx.txState().mvccEnabled(cctx) || tx.mvccSnapshot() != null || mvccCrd != null;
+
tx.transactionNodes(txMapping.transactionNodes());
if (!hasNearCache)
@@ -329,6 +376,16 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
for (final GridDistributedTxMapping m : mappings.values()) {
final ClusterNode primary = m.primary();
+ boolean needCntr = false;
+
+ if (mvccCrd != null) {
+ if (tx.onePhaseCommit() || mvccCrd.nodeId().equals(primary.id())) {
+ needCntr = true;
+
+ mvccCrd = null;
+ }
+ }
+
if (primary.isLocal()) {
if (m.hasNearCacheEntries() && m.hasColocatedCacheEntries()) {
GridNearTxPrepareRequest nearReq = createRequest(txMapping.transactionNodes(),
@@ -337,6 +394,8 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
m.nearEntriesReads(),
m.nearEntriesWrites());
+ nearReq.requestMvccCounter(needCntr);
+
prepareLocal(nearReq, m, ++miniId, true);
GridNearTxPrepareRequest colocatedReq = createRequest(txNodes,
@@ -350,6 +409,8 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
else {
GridNearTxPrepareRequest req = createRequest(txNodes, m, timeout, m.reads(), m.writes());
+ req.requestMvccCounter(needCntr);
+
prepareLocal(req, m, ++miniId, m.hasNearCacheEntries());
}
}
@@ -360,11 +421,13 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
m.reads(),
m.writes());
+ req.requestMvccCounter(needCntr);
+
final MiniFuture fut = new MiniFuture(m, ++miniId);
req.miniId(fut.futureId());
- add(fut);
+ add((IgniteInternalFuture)fut);
try {
cctx.io().send(primary, req, tx.ioPolicy());
@@ -392,6 +455,16 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
}
}
+ if (mvccCrd != null) {
+ assert !tx.onePhaseCommit();
+
+ MvccSnapshotFutureExt fut = new MvccSnapshotFutureExt();
+
+ cctx.coordinators().requestSnapshotAsync(tx, fut);
+
+ add((IgniteInternalFuture)fut);
+ }
+
markInitialized();
}
@@ -423,20 +496,52 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
@Override public String toString() {
Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
@Override public String apply(IgniteInternalFuture<?> f) {
- return "[node=" + ((MiniFuture)f).primary().id() +
- ", loc=" + ((MiniFuture)f).primary().isLocal() +
- ", done=" + f.isDone() + "]";
+ if (f instanceof MiniFuture) {
+ return "[node=" + ((MiniFuture)f).primary().id() +
+ ", loc=" + ((MiniFuture)f).primary().isLocal() +
+ ", done=" + f.isDone() + "]";
+ }
+ else
+ return f.toString();
}
});
return S.toString(GridNearPessimisticTxPrepareFuture.class, this,
"innerFuts", futs,
+ "txId", tx.nearXidVersion(),
"super", super.toString());
}
/**
*
*/
+ private class MvccSnapshotFutureExt extends GridFutureAdapter<Void> implements MvccSnapshotResponseListener {
+ /** {@inheritDoc} */
+ @Override public void onResponse(MvccSnapshot res) {
+ tx.mvccSnapshot(res);
+
+ onDone();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onError(IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Error on tx prepare [fut=" + this + ", err=" + e + ", tx=" + tx + ']');
+
+ if (ERR_UPD.compareAndSet(GridNearPessimisticTxPrepareFuture.this, null, e))
+ tx.setRollbackOnly();
+
+ onDone(e);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccSnapshotFutureExt.class, this, super.toString());
+ }
+ }
+
+
+ /** */
private class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
/** */
private final int futId;
@@ -475,6 +580,9 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
if (res.error() != null)
onError(res.error());
else {
+ if (res.mvccSnapshot() != null)
+ tx.mvccSnapshot(res.mvccSnapshot());
+
onPrepareResponse(m, res, updateMapping);
onDone(res);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
index 00ff4bb..d1aa2e9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
@@ -26,11 +26,13 @@ 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.MvccSnapshot;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
*
@@ -81,6 +83,9 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
/** TTL for read operation. */
private long accessTtl;
+ /** */
+ private MvccSnapshot mvccSnapshot;
+
/**
* Empty constructor required for {@link Message}.
*/
@@ -103,6 +108,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
* @param addReader Add reader flag.
* @param needVer {@code True} if entry version is needed.
* @param addDepInfo Deployment info.
+ * @param mvccSnapshot MVCC snapshot.
*/
public GridNearSingleGetRequest(
int cacheId,
@@ -118,7 +124,8 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
boolean addReader,
boolean needVer,
boolean addDepInfo,
- boolean recovery
+ boolean recovery,
+ MvccSnapshot mvccSnapshot
) {
assert key != null;
@@ -131,6 +138,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.addDepInfo = addDepInfo;
+ this.mvccSnapshot = mvccSnapshot;
if (readThrough)
flags |= READ_THROUGH_FLAG_MASK;
@@ -149,6 +157,13 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
}
/**
+ * @return Mvcc version.
+ */
+ @Nullable public MvccSnapshot mvccSnapshot() {
+ return mvccSnapshot;
+ }
+
+ /**
* @return Key.
*/
public KeyCacheObject key() {
@@ -322,7 +337,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 8:
- subjId = reader.readUuid("subjId");
+ mvccSnapshot = reader.readMessage("mvccSnapshot");
if (!reader.isLastRead())
return false;
@@ -330,7 +345,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 9:
- taskNameHash = reader.readInt("taskNameHash");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
@@ -338,6 +353,14 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 10:
+ taskNameHash = reader.readInt("taskNameHash");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 11:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
@@ -396,18 +419,24 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
writer.incrementState();
case 8:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
return false;
writer.incrementState();
case 9:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 10:
+ if (!writer.writeInt("taskNameHash", taskNameHash))
+ return false;
+
+ writer.incrementState();
+
+ case 11:
if (!writer.writeMessage("topVer", topVer))
return false;
@@ -430,7 +459,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 11;
+ return 12;
}
/** {@inheritDoc} */