You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by gv...@apache.org on 2018/11/22 11:35:44 UTC
[1/2] ignite git commit: IGNITE-10047: MVCC: Fixed MVCC coordinator
assignment on several nodes failure. This closes #5353.
Repository: ignite
Updated Branches:
refs/heads/master d81acdc29 -> 547704808
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccPreviousCoordinatorQueries.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccPreviousCoordinatorQueries.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccPreviousCoordinatorQueries.java
index cd7560f..6218bc0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccPreviousCoordinatorQueries.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccPreviousCoordinatorQueries.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache.mvcc;
import java.util.Collection;
import java.util.HashSet;
-import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -58,7 +57,7 @@ class MvccPreviousCoordinatorQueries {
* @param nodes Cluster nodes.
* @param mgr Discovery manager.
*/
- void init(Map<UUID, GridLongList> nodeQueries, Collection<ClusterNode> nodes, GridDiscoveryManager mgr) {
+ void init(GridLongList nodeQueries, Collection<ClusterNode> nodes, GridDiscoveryManager mgr) {
synchronized (this) {
assert !initDone;
assert waitNodes == null;
@@ -66,18 +65,14 @@ class MvccPreviousCoordinatorQueries {
waitNodes = new HashSet<>();
for (ClusterNode node : nodes) {
- if ((nodeQueries == null || !nodeQueries.containsKey(node.id())) &&
- mgr.alive(node) &&
- !F.contains(rcvd, node.id()))
+ if (!node.isLocal() && mgr.alive(node) && !F.contains(rcvd, node.id()))
waitNodes.add(node.id());
}
initDone = waitNodes.isEmpty();
- if (nodeQueries != null) {
- for (Map.Entry<UUID, GridLongList> e : nodeQueries.entrySet())
- mergeToActiveQueries(e.getKey(), e.getValue());
- }
+ if (nodeQueries != null)
+ mergeToActiveQueries(mgr.localNode().id(), nodeQueries);
if (initDone && !prevQueriesDone)
prevQueriesDone = activeQueries.isEmpty() && rcvdAcks.isEmpty();
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessor.java
index a926acf..fd45c7a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessor.java
@@ -17,21 +17,16 @@
package org.apache.ignite.internal.processors.cache.mvcc;
-import java.util.Collection;
-import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.internal.IgniteDiagnosticPrepareContext;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.GridProcessor;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.ExchangeContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.util.GridLongList;
@@ -42,31 +37,18 @@ import org.jetbrains.annotations.Nullable;
*/
public interface MvccProcessor extends GridProcessor {
/**
- * @param evtType Event type.
- * @param nodes Current nodes.
- * @param topVer Topology version.
- * @param customMsg Message
- */
- void onDiscoveryEvent(int evtType, Collection<ClusterNode> nodes, long topVer,
- @Nullable DiscoveryCustomMessage customMsg);
-
- /**
- * Exchange start callback.
+ * Local join callback.
*
- * @param mvccCrd Mvcc coordinator.
- * @param exchCtx Exchange context.
- * @param exchCrd Exchange coordinator.
+ * @param evt Discovery event.
*/
- void onExchangeStart(MvccCoordinator mvccCrd, ExchangeContext exchCtx, ClusterNode exchCrd);
+ void onLocalJoin(DiscoveryEvent evt);
/**
* Exchange done callback.
*
- * @param newCoord New coordinator flag.
* @param discoCache Disco cache.
- * @param activeQueries Active queries.
*/
- void onExchangeDone(boolean newCoord, DiscoCache discoCache, Map<UUID, GridLongList> activeQueries);
+ void onExchangeDone(DiscoCache discoCache);
/**
* @param nodeId Node ID
@@ -75,33 +57,16 @@ public interface MvccProcessor extends GridProcessor {
void processClientActiveQueries(UUID nodeId, @Nullable GridLongList activeQueries);
/**
- * @return Mvcc coordinator received from discovery event.
- */
- @Nullable MvccCoordinator assignedCoordinator();
-
- /**
* @return Coordinator.
*/
@Nullable MvccCoordinator currentCoordinator();
/**
- * Check that the given topology is greater or equals to coordinator's one and returns current coordinator.
- * @param topVer Topology version.
- * @return Mvcc coordinator.
- */
- @Nullable MvccCoordinator currentCoordinator(AffinityTopologyVersion topVer);
-
- /**
* @return Current coordinator node ID.
*/
UUID currentCoordinatorId();
/**
- * @param curCrd Coordinator.
- */
- void updateCoordinator(MvccCoordinator curCrd);
-
- /**
* @param crdVer Mvcc coordinator version.
* @param cntr Mvcc counter.
* @return State for given mvcc version.
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
index 851a5ea..c6515c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
@@ -48,15 +48,14 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.discovery.CustomEventListener;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest;
-import org.apache.ignite.internal.processors.cache.ExchangeContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -108,13 +107,10 @@ import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
-import static org.apache.ignite.events.EventType.EVT_CLIENT_NODE_DISCONNECTED;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
+import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
-import static org.apache.ignite.events.EventType.EVT_NODE_METRICS_UPDATED;
-import static org.apache.ignite.events.EventType.EVT_NODE_SEGMENTED;
import static org.apache.ignite.internal.GridTopic.TOPIC_CACHE_COORDINATOR;
-import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker.MVCC_TRACKER_ID_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
@@ -159,15 +155,12 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** Topology version when local node was assigned as coordinator. */
- private long crdVer;
+ private volatile long crdVer;
/** */
private volatile MvccCoordinator curCrd;
/** */
- private volatile MvccCoordinator assignedCrd;
-
- /** */
private TxLog txLog;
/** */
@@ -226,6 +219,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** Flag whether all nodes in cluster support MVCC. */
private volatile boolean mvccSupported = true;
+ /** Flag whether coordinator was changed by the last discovery event. */
+ private volatile boolean crdChanged;
+
/**
* Maps failed node id to votes accumulator for that node.
*/
@@ -242,10 +238,21 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** {@inheritDoc} */
@Override public void start() throws IgniteCheckedException {
- ctx.event().addLocalEventListener(new CacheCoordinatorNodeFailListener(),
- EVT_NODE_FAILED, EVT_NODE_LEFT);
+ ctx.event().addLocalEventListener(new GridLocalEventListener() {
+ @Override public void onEvent(Event evt) {
+ onDiscovery((DiscoveryEvent)evt);
+ }
+ },
+ EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_NODE_JOINED);
ctx.io().addMessageListener(TOPIC_CACHE_COORDINATOR, new CoordinatorMessageListener());
+
+ ctx.discovery().setCustomEventListener(DynamicCacheChangeBatch.class,
+ new CustomEventListener<DynamicCacheChangeBatch>() {
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, DynamicCacheChangeBatch msg) {
+ checkMvccCacheStarted(msg);
+ }
+ });
}
/** {@inheritDoc} */
@@ -352,72 +359,171 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** {@inheritDoc} */
- @Override public void onDiscoveryEvent(int evtType, Collection<ClusterNode> nodes, long topVer,
- @Nullable DiscoveryCustomMessage customMsg) {
- if (evtType == EVT_NODE_METRICS_UPDATED)
- return;
+ @Override public void onExchangeDone(DiscoCache discoCache) {
+ MvccCoordinator curCrd0 = curCrd;
- if (evtType == EVT_DISCOVERY_CUSTOM_EVT)
- checkMvccCacheStarted(customMsg);
- else
- assignMvccCoordinator(evtType, nodes, topVer);
+ if (crdChanged) {
+ // Rollback all transactions with old snapshots.
+ ctx.cache().context().tm().rollbackMvccTxOnCoordinatorChange();
+
+ // Complete init future if local node is a new coordinator. All previous txs are already completed here.
+ if (crdVer != 0 && !initFut.isDone()) {
+ assert curCrd0 != null && curCrd0.nodeId().equals(ctx.localNodeId());
+
+ initFut.onDone();
+ }
+
+ crdChanged = false;
+ }
+ else {
+ if (curCrd0 != null && ctx.localNodeId().equals(curCrd0.nodeId()) && discoCache != null)
+ cleanupOrphanedServerTransactions(discoCache.serverNodes());
+ }
}
/** {@inheritDoc} */
- @Override public void onExchangeStart(MvccCoordinator mvccCrd, ExchangeContext exchCtx, ClusterNode exchCrd) {
- if (!exchCtx.newMvccCoordinator())
+ @Override public void onLocalJoin(DiscoveryEvent evt) {
+ assert evt.type() == EVT_NODE_JOINED && ctx.localNodeId().equals(evt.eventNode().id());
+
+ onCoordinatorChanged(evt.topologyNodes(), evt.topologyVersion(), false);
+ }
+
+ /**
+ * Discovery listener. Note: initial join event is handled by {@link MvccProcessorImpl#onLocalJoin(DiscoveryEvent)}
+ * method.
+ *
+ * @param evt Discovery event.
+ */
+ private void onDiscovery(DiscoveryEvent evt) {
+ assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_JOINED;
+
+ UUID nodeId = evt.eventNode().id();
+
+ MvccCoordinator curCrd0 = curCrd;
+
+ if (evt.type() == EVT_NODE_JOINED) {
+ if (curCrd0 == null) // Handle join event only if coordinator has not been elected yet.
+ onCoordinatorChanged(evt.topologyNodes(), evt.topologyVersion(), false);
+
return;
+ }
- GridLongList activeQryTrackers = collectActiveQueryTrackers();
+ // Process mvcc coordinator left event on the rest nodes.
+ if (nodeId.equals(curCrd0.nodeId())) {
+ // 1. Notify all listeners waiting for a snapshot.
+ Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.remove(nodeId);
- exchCtx.addActiveQueries(ctx.localNodeId(), activeQryTrackers);
+ if (map != null) {
+ ClusterTopologyCheckedException ex = new ClusterTopologyCheckedException("Failed to request mvcc " +
+ "version, coordinator failed: " + nodeId);
- if (exchCrd == null || !mvccCrd.nodeId().equals(exchCrd.id())) {
- try {
- sendMessage(mvccCrd.nodeId(), new MvccActiveQueriesMessage(activeQryTrackers));
+ MvccSnapshotResponseListener lsnr;
+
+ for (Long id : map.keySet()) {
+ if ((lsnr = map.remove(id)) != null)
+ lsnr.onError(ex);
+ }
}
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to send active queries to mvcc coordinator: " + e);
+
+ // 2. Notify acknowledge futures.
+ for (WaitAckFuture fut : ackFuts.values())
+ fut.onNodeLeft(nodeId);
+
+ // 3. Process coordinator change.
+ onCoordinatorChanged(evt.topologyNodes(), evt.topologyVersion(), true);
+ }
+ // Process node left event on the current mvcc coordinator.
+ else if (curCrd0.nodeId().equals(ctx.localNodeId())) {
+ // 1. Notify active queries.
+ activeQueries.onNodeFailed(nodeId);
+
+ // 2. Notify previous queries.
+ prevCrdQueries.onNodeFailed(nodeId);
+
+ // 3. Recover transactions started by the failed node.
+ recoveryBallotBoxes.forEach((nearNodeId, ballotBox) -> {
+ // Put synthetic vote from another failed node
+ ballotBox.vote(nodeId);
+
+ tryFinishRecoveryVoting(nearNodeId, ballotBox);
+ });
+
+ if (evt.eventNode().isClient()) {
+ RecoveryBallotBox ballotBox = recoveryBallotBoxes
+ .computeIfAbsent(nodeId, uuid -> new RecoveryBallotBox());
+
+ ballotBox
+ .voters(evt.topologyNodes().stream().map(ClusterNode::id).collect(Collectors.toList()));
+
+ tryFinishRecoveryVoting(nodeId, ballotBox);
}
}
}
- /** {@inheritDoc} */
- @Override public void onExchangeDone(boolean newCrd, DiscoCache discoCache, Map<UUID, GridLongList> activeQueries) {
- if (!newCrd) {
- if (curCrd != null && ctx.localNodeId().equals(curCrd.nodeId()) && discoCache != null)
- cleanupOrphanedServerTransactions(discoCache.serverNodes());
+ /**
+ * Coordinator change callback. Performs all needed actions for handling new coordinator assignment.
+ *
+ * @param nodes Cluster topology snapshot.
+ * @param topVer Topology version.
+ * @param sndQrys {@code True} if it is need to send an active queries list to the new coordinator.
+ */
+ private void onCoordinatorChanged(Collection<ClusterNode> nodes, long topVer, boolean sndQrys) {
+ MvccCoordinator newCrd = pickMvccCoordinator(nodes, topVer);
+ if (newCrd == null)
return;
- }
- ctx.cache().context().tm().rollbackMvccTxOnCoordinatorChange();
+ // Update current coordinator, collect active queries and send it to the new coordinator if needed.
+ GridLongList activeQryTrackers = null;
- if (ctx.localNodeId().equals(curCrd.nodeId())) {
- assert ctx.localNodeId().equals(curCrd.nodeId());
+ synchronized (activeTrackers) {
+ assert curCrd == null || newCrd.topologyVersion().compareTo(curCrd.topologyVersion()) > 0;
- MvccCoordinator crd = discoCache.mvccCoordinator();
+ if (sndQrys) {
+ activeQryTrackers = new GridLongList();
- assert crd != null;
+ for (MvccQueryTracker tracker : activeTrackers.values()) {
+ long trackerId = tracker.onMvccCoordinatorChange(newCrd);
- // No need to re-initialize if coordinator version hasn't changed (e.g. it was cluster activation).
- if (crdVer == crd.coordinatorVersion())
- return;
+ if (trackerId != MVCC_TRACKER_ID_NA)
+ activeQryTrackers.add(trackerId);
+ }
+ }
+
+ curCrd = newCrd;
+ }
+
+ // Send local active queries to remote coordinator, if needed.
+ if (!newCrd.nodeId().equals(ctx.localNodeId())) {
+ try {
+ if (sndQrys)
+ sendMessage(newCrd.nodeId(), new MvccActiveQueriesMessage(activeQryTrackers));
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send active queries to mvcc coordinator: " + e);
+ }
+ }
+ // If a current node was elected as a new mvcc coordinator, we need to pre-initialize it.
+ else {
+ assert crdVer == 0 : crdVer;
- crdVer = crd.coordinatorVersion();
+ crdVer = newCrd.coordinatorVersion();
if (log.isInfoEnabled())
log.info("Initialize local node as mvcc coordinator [node=" + ctx.localNodeId() +
", crdVer=" + crdVer + ']');
- prevCrdQueries.init(activeQueries, F.view(discoCache.allNodes(), this::supportsMvcc), ctx.discovery());
+ prevCrdQueries.init(activeQryTrackers, F.view(nodes, this::supportsMvcc), ctx.discovery());
- initFut.onDone();
+ // Do not complete init future here, because we should wait until all old transactions become terminated.
}
+
+ crdChanged = true;
}
/**
- * Cleans up active transacitons lost near node which is server. Executed on coordinator.
+ * Cleans up active transactions lost near node which is server. Executed on coordinator.
+ *
* @param liveSrvs Live server nodes at the moment of cleanup.
*/
private void cleanupOrphanedServerTransactions(Collection<ClusterNode> liveSrvs) {
@@ -450,24 +556,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** {@inheritDoc} */
@Override @Nullable public MvccCoordinator currentCoordinator() {
- return currentCoordinator(AffinityTopologyVersion.NONE);
- }
-
- /** {@inheritDoc} */
- @Override @Nullable public MvccCoordinator currentCoordinator(AffinityTopologyVersion topVer) {
- MvccCoordinator crd = curCrd;
-
- // Assert coordinator did not already change.
- assert crd == null
- || topVer == AffinityTopologyVersion.NONE
- || crd.topologyVersion().compareTo(topVer) <= 0 : "Invalid coordinator [crd=" + crd + ", topVer=" + topVer + ']';
-
- return crd;
- }
-
- /** {@inheritDoc} */
- @Override @Nullable public MvccCoordinator assignedCoordinator() {
- return assignedCrd;
+ return curCrd;
}
/** {@inheritDoc} */
@@ -478,11 +567,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** {@inheritDoc} */
- @Override public void updateCoordinator(MvccCoordinator curCrd) {
- this.curCrd = curCrd;
- }
-
- /** {@inheritDoc} */
@Override public byte state(MvccVersion ver) throws IgniteCheckedException {
return state(ver.coordinatorVersion(), ver.counter());
}
@@ -837,51 +921,47 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
return ctx.config().getDataStorageConfiguration();
}
- /** */
- private void assignMvccCoordinator(int evtType, Collection<ClusterNode> nodes, long topVer) {
+ /**
+ * Picks mvcc coordinator from the given list of nodes.
+ *
+ * @param nodes List of nodes.
+ * @param topVer Topology version.
+ * @return Chosen mvcc coordinator.
+ */
+ private MvccCoordinator pickMvccCoordinator(Collection<ClusterNode> nodes, long topVer) {
checkMvccSupported(nodes);
- MvccCoordinator crd;
-
- if (evtType == EVT_NODE_SEGMENTED || evtType == EVT_CLIENT_NODE_DISCONNECTED)
- crd = null;
- else {
- crd = assignedCrd;
-
- if (crd == null ||
- ((evtType == EVT_NODE_FAILED || evtType == EVT_NODE_LEFT) && !F.nodeIds(nodes).contains(crd.nodeId()))) {
- ClusterNode crdNode = null;
+ ClusterNode crdNode = null;
- if (crdC != null) {
- crdNode = crdC.apply(nodes);
+ if (crdC != null) {
+ crdNode = crdC.apply(nodes);
- if (log.isInfoEnabled())
- log.info("Assigned coordinator using test closure: " + crd);
- }
- else {
- // Expect nodes are sorted by order.
- for (ClusterNode node : nodes) {
- if (!node.isClient() && supportsMvcc(node)) {
- crdNode = node;
+ if (crdNode != null && log.isInfoEnabled())
+ log.info("Assigned coordinator using test closure: " + crdNode.id());
+ }
+ else {
+ // Expect nodes are sorted by order.
+ for (ClusterNode node : nodes) {
+ if (!node.isClient() && supportsMvcc(node)) {
+ crdNode = node;
- break;
- }
- }
+ break;
}
-
- crd = crdNode != null ? new MvccCoordinator(crdNode.id(), coordinatorVersion(crdNode),
- new AffinityTopologyVersion(topVer, 0)) : null;
-
- if (log.isInfoEnabled() && crd != null)
- log.info("Assigned mvcc coordinator [crd=" + crd + ", crdNode=" + crdNode + ']');
- else if (crd == null)
- U.warn(log, "New mvcc coordinator was not assigned [topVer=" + topVer + ']');
}
}
- assignedCrd = crd;
+ MvccCoordinator crd = crdNode != null ? new MvccCoordinator(crdNode.id(), coordinatorVersion(crdNode),
+ new AffinityTopologyVersion(topVer, 0)) : null;
+
+ if (log.isInfoEnabled() && crd != null)
+ log.info("Assigned mvcc coordinator [crd=" + crd + ", crdNode=" + crdNode + ']');
+ else if (crd == null)
+ U.warn(log, "New mvcc coordinator was not assigned [topVer=" + topVer + ']');
+
+ return crd;
}
+
/**
* @param crdNode Assigned coordinator node.
* @return Coordinator version.
@@ -918,11 +998,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** */
- private void checkMvccCacheStarted(@Nullable DiscoveryCustomMessage customMsg) {
- assert customMsg != null;
-
- if (!mvccEnabled && customMsg instanceof DynamicCacheChangeBatch) {
- for (DynamicCacheChangeRequest req : ((DynamicCacheChangeBatch)customMsg).requests()) {
+ private void checkMvccCacheStarted(DynamicCacheChangeBatch cacheMsg) {
+ if (!mvccEnabled) {
+ for (DynamicCacheChangeRequest req : cacheMsg.requests()) {
CacheConfiguration ccfg = req.startCacheConfiguration();
if (ccfg == null)
@@ -937,24 +1015,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
}
- /**
- * @return Active queries list.
- */
- private GridLongList collectActiveQueryTrackers() {
- assert curCrd != null;
-
- GridLongList activeQryTrackers = new GridLongList();
-
- for (MvccQueryTracker tracker : activeTrackers.values()) {
- long trackerId = tracker.onMvccCoordinatorChange(curCrd);
-
- if (trackerId != MVCC_TRACKER_ID_NA)
- activeQryTrackers.add(trackerId);
- }
-
- return activeQryTrackers;
- }
-
/** */
private MvccSnapshotResponse assignTxSnapshot(long futId, UUID nearId, boolean client) {
assert initFut.isDone();
@@ -1342,7 +1402,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
fut.onDone(); // No need to ack, finish without error.
}
else
-
fut.onDone(e);
}
}
@@ -1743,63 +1802,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/**
*
*/
- private class CacheCoordinatorNodeFailListener implements GridLocalEventListener {
- /** {@inheritDoc} */
- @Override public void onEvent(Event evt) {
- assert evt instanceof DiscoveryEvent : evt;
-
- DiscoveryEvent discoEvt = (DiscoveryEvent)evt;
-
- UUID nodeId = discoEvt.eventNode().id();
-
- Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.remove(nodeId);
-
- if (map != null) {
- ClusterTopologyCheckedException ex = new ClusterTopologyCheckedException("Failed to request mvcc " +
- "version, coordinator failed: " + nodeId);
-
- MvccSnapshotResponseListener lsnr;
-
- for (Long id : map.keySet()) {
- if ((lsnr = map.remove(id)) != null)
- lsnr.onError(ex);
- }
- }
-
- for (WaitAckFuture fut : ackFuts.values())
- fut.onNodeLeft(nodeId);
-
- activeQueries.onNodeFailed(nodeId);
-
- prevCrdQueries.onNodeFailed(nodeId);
-
- recoveryBallotBoxes.forEach((nearNodeId, ballotBox) -> {
- // Put synthetic vote from another failed node
- ballotBox.vote(nodeId);
-
- tryFinishRecoveryVoting(nearNodeId, ballotBox);
- });
-
- if (discoEvt.eventNode().isClient()) {
- RecoveryBallotBox ballotBox = recoveryBallotBoxes
- .computeIfAbsent(nodeId, uuid -> new RecoveryBallotBox());
-
- ballotBox
- .voters(discoEvt.topologyNodes().stream().map(ClusterNode::id).collect(Collectors.toList()));
-
- tryFinishRecoveryVoting(nodeId, ballotBox);
- }
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return "CacheCoordinatorDiscoveryListener[]";
- }
- }
-
- /**
- *
- */
private class CoordinatorMessageListener implements GridMessageListener {
/** {@inheritDoc} */
@Override public void onMessage(UUID nodeId, Object msg, byte plc) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTrackerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTrackerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTrackerImpl.java
index 7296d66..d86f5ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTrackerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTrackerImpl.java
@@ -227,7 +227,7 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
* @return {@code True} if topology is valid.
*/
private boolean checkTopology(AffinityTopologyVersion topVer, MvccSnapshotResponseListener lsnr) {
- MvccCoordinator crd = cctx.affinity().mvccCoordinator(topVer);
+ MvccCoordinator crd = cctx.shared().coordinators().currentCoordinator();
if (crd == null) {
lsnr.onError(noCoordinatorError(topVer));
@@ -241,16 +241,6 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
crdVer = crd.coordinatorVersion();
}
- MvccCoordinator curCrd = cctx.topology().mvccCoordinator();
-
- if (!crd.equals(curCrd)) {
- assert cctx.topology().topologyVersionFuture().initialVersion().compareTo(topVer) > 0;
-
- tryRemap(lsnr);
-
- return false;
- }
-
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 791bf76..a8b594b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -267,7 +267,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
// Wait some time in case there are some unprocessed messages from failed node.
cctx.time().addTimeoutObject(
- new NodeFailureTimeoutObject(evt.eventNode(), discoCache.mvccCoordinator()));
+ new NodeFailureTimeoutObject(evt.eventNode(), cctx.coordinators().currentCoordinator()));
if (txFinishSync != null)
txFinishSync.onNodeLeft(nodeId);
@@ -323,7 +323,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*/
public void rollbackMvccTxOnCoordinatorChange() {
for (IgniteInternalTx tx : activeTransactions()) {
- if (tx.mvccSnapshot() != null)
+ if (tx.mvccSnapshot() != null && tx instanceof GridNearTxLocal)
((GridNearTxLocal)tx).rollbackNearTxLocalAsync(false, false);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractBasicCoordinatorFailoverTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractBasicCoordinatorFailoverTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractBasicCoordinatorFailoverTest.java
index c1718b5..62ea12c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractBasicCoordinatorFailoverTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractBasicCoordinatorFailoverTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache.mvcc;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -27,19 +28,27 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.cache.Cache;
import javax.cache.CacheException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.GridCachePartitionExchangeManager;
import org.apache.ignite.internal.processors.cache.distributed.TestCacheNodeExcludingFilter;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccSnapshotResponse;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteBiPredicate;
@@ -675,4 +684,179 @@ public abstract class CacheMvccAbstractBasicCoordinatorFailoverTest extends Cach
for (Ignite node : G.allGrids())
checkActiveQueriesCleanup(node);
}
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testMultipleCoordinatorsLeft2Persistence() throws Exception {
+ persistence = true;
+
+ checkCoordinatorsLeft(2, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testMultipleCoordinatorsLeft3Persistence() throws Exception {
+ persistence = true;
+
+ checkCoordinatorsLeft(3, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testMultipleCoordinatorsLeft4() throws Exception {
+ checkCoordinatorsLeft(4, true);
+ }
+
+ /**
+ * @param num Number of coordinators to stop.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void checkCoordinatorsLeft(int num, boolean stopCrdFirst) throws Exception {
+ disableScheduledVacuum = true;
+
+ final int DATA_NODES = 3;
+
+ final int NODES = num + DATA_NODES;
+
+ nodeAttr = CRD_ATTR;
+
+ // Do not use startMultithreaded here.
+ startGrids(num);
+
+ nodeAttr = null;
+
+ startGridsMultiThreaded(num, DATA_NODES);
+
+ List<Ignite> victims = new ArrayList<>(num);
+ List<Ignite> survivors = new ArrayList<>(DATA_NODES);
+
+ for (int i = 0; i < NODES; i++) {
+ if (i < num)
+ victims.add(grid(i));
+ else
+ survivors.add(grid(i));
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info("Nodes to be stopped [" +
+ victims.stream().
+ map(n -> n.cluster().localNode().id().toString())
+ .collect(Collectors.joining(", ")) + ']');
+
+ log.info("Nodes not to be stopped [" +
+ survivors.stream().
+ map(n -> n.cluster().localNode().id().toString())
+ .collect(Collectors.joining(", ")) + ']');
+ }
+
+ Ignite nearNode = survivors.get(0);
+
+ if (persistence)
+ nearNode.cluster().active(true);
+
+ CacheConfiguration ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, DATA_NODES - 1, DFLT_PARTITION_COUNT)
+ .setNodeFilter(new CoordinatorNodeFilter());
+
+ IgniteCache cache = nearNode.createCache(ccfg);
+
+ try (Transaction tx = nearNode.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (int key = 0; key < 10; key++)
+ cache.put(key, 1);
+
+ tx.commit();
+ }
+
+ List<Thread> stopThreads = victims.stream()
+ .map(v -> new Thread(() -> stopGrid(v.name())))
+ .collect(Collectors.toList());
+
+ ScanQuery<Object, Object> scan = new ScanQuery<>();
+
+ QueryCursor<Cache.Entry<Object, Object>> cur = survivors.get(0).cache(DEFAULT_CACHE_NAME).query(scan);
+
+ Iterator<Cache.Entry<Object, Object>> it = cur.iterator();
+
+ assertTrue(it.hasNext());
+ assertEquals(1, it.next().getValue());
+
+ if (log.isInfoEnabled())
+ log.info("Start stopping nodes.");
+
+ // Stop nodes and join threads.
+ if (stopCrdFirst) {
+ for (Thread t : stopThreads)
+ t.start();
+ }
+ else {
+ // We should stop the oldest node last.
+ GridCachePartitionExchangeManager exch = ((IgniteEx)survivors.get(1)).context().cache().context().exchange();
+
+ GridDhtTopologyFuture lastFinished = exch.lastFinishedFuture();
+
+ for (int i = 1; i < stopThreads.size(); i++)
+ stopThreads.get(i).start();
+
+ while (lastFinished == exch.lastTopologyFuture())
+ doSleep(1);
+
+ stopThreads.get(0).start();
+ }
+
+ for (Thread t : stopThreads)
+ t.join();
+
+ if (log.isInfoEnabled())
+ log.info("All nodes stopped.");
+
+ assertTrue(it.hasNext());
+ assertEquals(1, it.next().getValue());
+
+ for (Ignite node : survivors) {
+ for (int key = 0; key < 10; key++)
+ assertEquals(1, node.cache(DEFAULT_CACHE_NAME).get(key));
+ }
+
+ try (Transaction tx = nearNode.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (int key = 0; key < 10; key++)
+ cache.put(key, 2);
+
+ tx.commit();
+ }
+ catch (Exception e) {
+ stopAllGrids(true);
+
+ fail(X.getFullStackTrace(e));
+ }
+
+ for (Ignite node : survivors) {
+ for (int key = 0; key < 10; key++)
+ assertEquals(2, node.cache(DEFAULT_CACHE_NAME).get(key));
+ }
+
+ try (Transaction tx = nearNode.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (int key = 0; key < 10; key++)
+ cache.put(key, 3);
+
+ tx.commit();
+ }
+ catch (Exception e) {
+ stopAllGrids(true);
+
+ fail(X.getFullStackTrace(e));
+ }
+
+ for (Ignite node : survivors) {
+ for (int key = 0; key < 10; key++)
+ assertEquals(3, node.cache(DEFAULT_CACHE_NAME).get(key));
+ }
+
+ while (it.hasNext())
+ assertEquals(1, (int)it.next().getValue());
+
+ cur.close();
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedCoordinatorFailoverTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedCoordinatorFailoverTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedCoordinatorFailoverTest.java
index 3ea1c5b..3425382 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedCoordinatorFailoverTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedCoordinatorFailoverTest.java
@@ -17,8 +17,12 @@
package org.apache.ignite.internal.processors.cache.mvcc;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.internal.processors.cache.mvcc.CacheMvccAbstractTest.ReadMode.GET;
import static org.apache.ignite.internal.processors.cache.mvcc.CacheMvccAbstractTest.ReadMode.SCAN;
import static org.apache.ignite.internal.processors.cache.mvcc.CacheMvccAbstractTest.WriteMode.PUT;
@@ -141,4 +145,35 @@ public class CacheMvccPartitionedCoordinatorFailoverTest extends CacheMvccAbstra
public void testReadInProgressCoordinatorFailsSimple_FromServerPutGet() throws Exception {
readInProgressCoordinatorFailsSimple(false, null, GET, PUT);
}
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testActivateDeactivateCLuster() throws Exception {
+ disableScheduledVacuum = true;
+ persistence = true;
+
+ final int DATA_NODES = 3;
+
+ // Do not use startMultithreaded here.
+ startGrids(DATA_NODES);
+
+ Ignite near = grid(DATA_NODES - 1);
+
+ CacheConfiguration ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, DATA_NODES - 1, DFLT_PARTITION_COUNT);
+
+ near.cluster().active(true);
+
+ IgniteCache cache = near.createCache(ccfg);
+
+ cache.put(1, 1);
+
+ near.cluster().active(false);
+
+ stopGrid(0);
+
+ near.cluster().active(true);
+
+ assertEquals(1, cache.get(1));
+ }
}
[2/2] ignite git commit: IGNITE-10047: MVCC: Fixed MVCC coordinator
assignment on several nodes failure. This closes #5353.
Posted by gv...@apache.org.
IGNITE-10047: MVCC: Fixed MVCC coordinator assignment on several nodes failure. This closes #5353.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/54770480
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/54770480
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/54770480
Branch: refs/heads/master
Commit: 5477048082711bba0501c73637b4ebaf5270cda4
Parents: d81acdc
Author: rkondakov <ko...@mail.ru>
Authored: Thu Nov 22 14:35:35 2018 +0300
Committer: Igor Seliverstov <gv...@gmail.com>
Committed: Thu Nov 22 14:35:35 2018 +0300
----------------------------------------------------------------------
.../internal/managers/discovery/DiscoCache.java | 15 -
.../discovery/GridDiscoveryManager.java | 13 +-
.../processors/affinity/AffinityAssignment.java | 6 -
.../affinity/GridAffinityAssignment.java | 15 +-
.../affinity/GridAffinityAssignmentCache.java | 22 +-
.../affinity/GridAffinityProcessor.java | 2 +-
.../processors/affinity/GridAffinityUtils.java | 2 +-
.../affinity/HistoryAffinityAssignment.java | 19 +-
.../cache/CacheAffinitySharedManager.java | 13 +-
.../processors/cache/ExchangeContext.java | 43 +--
.../cache/GridCacheAffinityManager.java | 5 -
.../GridCachePartitionExchangeManager.java | 8 -
.../distributed/dht/GridDhtTxPrepareFuture.java | 23 --
.../GridDhtPartitionsExchangeFuture.java | 43 +--
.../GridDhtPartitionsSingleMessage.java | 68 +---
.../topology/GridClientPartitionTopology.java | 7 -
.../dht/topology/GridDhtPartitionTopology.java | 4 -
.../topology/GridDhtPartitionTopologyImpl.java | 11 -
...arOptimisticSerializableTxPrepareFuture.java | 35 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 39 +-
...ridNearOptimisticTxPrepareFutureAdapter.java | 114 ------
.../GridNearPessimisticTxPrepareFuture.java | 78 +---
.../near/GridNearTxPrepareRequest.java | 17 -
.../near/GridNearTxPrepareResponse.java | 52 +--
.../mvcc/MvccPreviousCoordinatorQueries.java | 13 +-
.../processors/cache/mvcc/MvccProcessor.java | 45 +--
.../cache/mvcc/MvccProcessorImpl.java | 372 ++++++++++---------
.../cache/mvcc/MvccQueryTrackerImpl.java | 12 +-
.../cache/transactions/IgniteTxManager.java | 4 +-
...vccAbstractBasicCoordinatorFailoverTest.java | 184 +++++++++
...eMvccPartitionedCoordinatorFailoverTest.java | 35 ++
31 files changed, 467 insertions(+), 852 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
index 70e2013..e99e478 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
@@ -26,7 +26,6 @@ import java.util.UUID;
import org.apache.ignite.cluster.BaselineNode;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -109,14 +108,10 @@ public class DiscoCache {
/** */
private final P1<ClusterNode> aliveNodePred;
- /** */
- private final MvccCoordinator mvccCrd;
-
/**
* @param topVer Topology version.
* @param state Current cluster state.
* @param loc Local node.
- * @param mvccCrd MVCC coordinator node.
* @param rmtNodes Remote nodes.
* @param allNodes All nodes.
* @param srvNodes Server nodes.
@@ -135,7 +130,6 @@ public class DiscoCache {
AffinityTopologyVersion topVer,
DiscoveryDataClusterState state,
ClusterNode loc,
- MvccCoordinator mvccCrd,
List<ClusterNode> rmtNodes,
List<ClusterNode> allNodes,
List<ClusterNode> srvNodes,
@@ -154,7 +148,6 @@ public class DiscoCache {
this.topVer = topVer;
this.state = state;
this.loc = loc;
- this.mvccCrd = mvccCrd;
this.rmtNodes = rmtNodes;
this.allNodes = allNodes;
this.srvNodes = srvNodes;
@@ -184,13 +177,6 @@ public class DiscoCache {
}
/**
- * @return Mvcc coordinator node.
- */
- @Nullable public MvccCoordinator mvccCoordinator() {
- return mvccCrd;
- }
-
- /**
* @return Topology version.
*/
public AffinityTopologyVersion version() {
@@ -475,7 +461,6 @@ public class DiscoCache {
ver,
state == null ? this.state : state,
loc,
- mvccCrd,
rmtNodes,
allNodes,
srvNodes,
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index fbfeb12..476fe7e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -85,7 +85,6 @@ import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest;
import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
@@ -657,8 +656,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
updateClientNodes(node.id());
}
- ctx.coordinators().onDiscoveryEvent(type, topSnapshot, topVer, customMsg);
-
boolean locJoinEvt = type == EVT_NODE_JOINED && node.id().equals(locNode.id());
ChangeGlobalStateFinishMessage stateFinishMsg = null;
@@ -791,6 +788,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
discoWrk.discoCache = discoCache;
if (!isLocDaemon && !ctx.clientDisconnected()) {
+ ctx.cache().context().coordinators().onLocalJoin(discoEvt);
+
ctx.cache().context().exchange().onLocalJoin(discoEvt, discoCache);
ctx.authentication().onLocalJoin();
@@ -2338,8 +2337,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
Collection<ClusterNode> topSnapshot) {
assert topSnapshot.contains(loc);
- MvccCoordinator mvccCrd = ctx.coordinators().assignedCoordinator();
-
HashSet<UUID> alives = U.newHashSet(topSnapshot.size());
HashMap<UUID, ClusterNode> nodeMap = U.newHashMap(topSnapshot.size());
@@ -2441,7 +2438,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
topVer,
state,
loc,
- mvccCrd,
Collections.unmodifiableList(rmtNodes),
Collections.unmodifiableList(allNodes),
Collections.unmodifiableList(srvNodes),
@@ -2613,8 +2609,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
AffinityTopologyVersion.NONE,
ctx.state().clusterState(),
node,
- locNodeOnlyTop
- ), locNodeOnlyTop,
+ locNodeOnlyTop),
+ locNodeOnlyTop,
null);
lastSegChkRes.set(false);
@@ -3447,7 +3443,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
topVer,
discoCache.state(),
discoCache.localNode(),
- discoCache.mvccCoordinator(),
discoCache.remoteNodes(),
allNodes,
discoCache.serverNodes(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityAssignment.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityAssignment.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityAssignment.java
index eb06d3e..b603c32 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityAssignment.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityAssignment.java
@@ -22,7 +22,6 @@ import java.util.List;
import java.util.Set;
import java.util.UUID;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
/**
* Cached affinity calculations.
@@ -84,9 +83,4 @@ public interface AffinityAssignment {
* @return Backup partitions for specified node ID.
*/
public Set<Integer> backupPartitions(UUID nodeId);
-
- /**
- * @return Mvcc coordinator.
- */
- public MvccCoordinator mvccCoordinator();
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
index fe51036..95cf76f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
@@ -27,7 +27,6 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.typedef.internal.S;
/**
@@ -41,9 +40,6 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
/** Topology version. */
private final AffinityTopologyVersion topVer;
- /** */
- private final MvccCoordinator mvccCrd;
-
/** Collection of calculated affinity nodes. */
private List<List<ClusterNode>> assignment;
@@ -74,7 +70,6 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
this.topVer = topVer;
primary = new HashMap<>();
backup = new HashMap<>();
- mvccCrd = null;
}
/**
@@ -84,8 +79,7 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
*/
GridAffinityAssignment(AffinityTopologyVersion topVer,
List<List<ClusterNode>> assignment,
- List<List<ClusterNode>> idealAssignment,
- MvccCoordinator mvccCrd) {
+ List<List<ClusterNode>> idealAssignment) {
assert topVer != null;
assert assignment != null;
assert idealAssignment != null;
@@ -93,7 +87,6 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
this.topVer = topVer;
this.assignment = assignment;
this.idealAssignment = idealAssignment.equals(assignment) ? assignment : idealAssignment;
- this.mvccCrd = mvccCrd;
primary = new HashMap<>();
backup = new HashMap<>();
@@ -112,7 +105,6 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
idealAssignment = aff.idealAssignment;
primary = aff.primary;
backup = aff.backup;
- mvccCrd = aff.mvccCrd;
}
/**
@@ -276,11 +268,6 @@ public class GridAffinityAssignment implements AffinityAssignment, Serializable
}
/** {@inheritDoc} */
- @Override public MvccCoordinator mvccCoordinator() {
- return mvccCrd;
- }
-
- /** {@inheritDoc} */
@Override public int hashCode() {
return topVer.hashCode();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index 4d5cf27..0e8a298 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -43,7 +43,6 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.NodeOrderComparator;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents;
import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -202,28 +201,11 @@ public class GridAffinityAssignmentCache {
* @param affAssignment Affinity assignment for topology version.
*/
public void initialize(AffinityTopologyVersion topVer, List<List<ClusterNode>> affAssignment) {
- MvccCoordinator mvccCrd = null;
-
- if (!locCache)
- mvccCrd = ctx.cache().context().coordinators().currentCoordinator(topVer);
-
- initialize(topVer, affAssignment, mvccCrd);
- }
-
- /**
- * Initializes affinity with given topology version and assignment.
- *
- * @param topVer Topology version.
- * @param affAssignment Affinity assignment for topology version.
- * @param mvccCrd Mvcc coordinator.
- */
- public void initialize(AffinityTopologyVersion topVer, List<List<ClusterNode>> affAssignment, MvccCoordinator mvccCrd) {
assert topVer.compareTo(lastVersion()) >= 0 : "[topVer = " + topVer + ", last=" + lastVersion() + ']';
assert idealAssignment != null;
- assert mvccCrd == null || topVer.compareTo(mvccCrd.topologyVersion()) >= 0 : "[mvccCrd=" + mvccCrd + ", topVer=" + topVer + ']';
- GridAffinityAssignment assignment = new GridAffinityAssignment(topVer, affAssignment, idealAssignment, mvccCrd);
+ GridAffinityAssignment assignment = new GridAffinityAssignment(topVer, affAssignment, idealAssignment);
HistoryAffinityAssignment hAff = affCache.put(topVer, new HistoryAffinityAssignment(assignment));
@@ -786,7 +768,7 @@ public class GridAffinityAssignmentCache {
AffinityAssignment assign = aff.cachedAffinity(aff.lastVersion());
- initialize(aff.lastVersion(), assign.assignment(), assign.mvccCoordinator());
+ initialize(aff.lastVersion(), assign.assignment());
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
index 41c1aee..07beb14 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
@@ -422,7 +422,7 @@ public class GridAffinityProcessor extends GridProcessorAdapter {
try {
GridAffinityAssignment assign = assign0 instanceof GridAffinityAssignment ?
(GridAffinityAssignment)assign0 :
- new GridAffinityAssignment(topVer, assign0.assignment(), assign0.idealAssignment(), assign0.mvccCoordinator());
+ new GridAffinityAssignment(topVer, assign0.assignment(), assign0.idealAssignment());
AffinityInfo info = new AffinityInfo(
cctx.config().getAffinity(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java
index 15d7e4e..abd5292 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java
@@ -184,7 +184,7 @@ class GridAffinityUtils {
GridAffinityAssignment assign = assign0 instanceof GridAffinityAssignment ?
(GridAffinityAssignment)assign0 :
- new GridAffinityAssignment(topVer, assign0.assignment(), assign0.idealAssignment(), assign0.mvccCoordinator());
+ new GridAffinityAssignment(topVer, assign0.assignment(), assign0.idealAssignment());
return F.t(
affinityMessage(ctx, cctx.config().getAffinity()),
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/HistoryAffinityAssignment.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/HistoryAffinityAssignment.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/HistoryAffinityAssignment.java
index a949648..6bdf7f0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/HistoryAffinityAssignment.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/HistoryAffinityAssignment.java
@@ -17,16 +17,14 @@
package org.apache.ignite.internal.processors.affinity;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
-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 java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
+import org.apache.ignite.cluster.ClusterNode;
+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;
/**
*
@@ -42,9 +40,6 @@ public class HistoryAffinityAssignment implements AffinityAssignment {
/** */
private final List<List<ClusterNode>> idealAssignment;
- /** */
- private final MvccCoordinator mvccCrd;
-
/**
* @param assign Assignment.
*/
@@ -52,12 +47,6 @@ public class HistoryAffinityAssignment implements AffinityAssignment {
this.topVer = assign.topologyVersion();
this.assignment = assign.assignment();
this.idealAssignment = assign.idealAssignment();
- this.mvccCrd = assign.mvccCoordinator();
- }
-
- /** {@inheritDoc} */
- @Override public MvccCoordinator mvccCoordinator() {
- return mvccCrd;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index 936dc3e..881f8db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@ -61,7 +61,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridClientPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.GridPartitionStateMap;
@@ -480,7 +479,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
grp.topology().updateTopologyVersion(topFut,
discoCache,
- cctx.coordinators().currentCoordinator(),
-1,
false);
@@ -530,7 +528,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert grp != null;
GridDhtAffinityAssignmentResponse res = fetchAffinity(topVer,
- cctx.coordinators().currentCoordinator(),
null,
discoCache,
grp.affinity(),
@@ -555,7 +552,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
grp.topology().updateTopologyVersion(topFut,
discoCache,
- cctx.coordinators().currentCoordinator(),
-1,
false);
@@ -1409,7 +1405,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
fetchFut.init(false);
fetchAffinity(evts.topologyVersion(),
- cctx.coordinators().currentCoordinator(),
evts,
evts.discoveryCache(),
aff,
@@ -1852,7 +1847,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
int grpId = fetchFut.groupId();
fetchAffinity(topVer,
- cctx.coordinators().currentCoordinator(),
fut.events(),
fut.events().discoveryCache(),
cctx.cache().cacheGroup(grpId).affinity(),
@@ -1864,7 +1858,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
/**
* @param topVer Topology version.
- * @param mvccCrd Mvcc coordinator to set in affinity.
* @param events Discovery events.
* @param discoCache Discovery data cache.
* @param affCache Affinity.
@@ -1874,7 +1867,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
*/
private GridDhtAffinityAssignmentResponse fetchAffinity(
AffinityTopologyVersion topVer,
- MvccCoordinator mvccCrd,
@Nullable ExchangeDiscoveryEvents events,
DiscoCache discoCache,
GridAffinityAssignmentCache affCache,
@@ -1887,7 +1879,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
if (res == null) {
List<List<ClusterNode>> aff = affCache.calculate(topVer, events, discoCache);
- affCache.initialize(topVer, aff, mvccCrd);
+ affCache.initialize(topVer, aff);
}
else {
List<List<ClusterNode>> idealAff = res.idealAffinityAssignment(discoCache);
@@ -1904,7 +1896,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert aff != null : res;
- affCache.initialize(topVer, aff, mvccCrd);
+ affCache.initialize(topVer, aff);
}
return res;
@@ -2027,7 +2019,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
@Override public void applyx(IgniteInternalFuture<GridDhtAffinityAssignmentResponse> fetchFut)
throws IgniteCheckedException {
fetchAffinity(prev.topologyVersion(),
- null, // Pass null mvcc coordinator, this affinity version should be used for queries.
prev.events(),
prev.events().discoveryCache(),
aff,
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
index 34ed048..4046c98 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
@@ -17,15 +17,11 @@
package org.apache.ignite.internal.processors.cache;
-import java.util.HashMap;
import java.util.HashSet;
-import java.util.Map;
import java.util.Set;
-import java.util.UUID;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.jetbrains.annotations.Nullable;
@@ -55,20 +51,11 @@ public class ExchangeContext {
/** */
private final boolean compatibilityNode = getBoolean(IGNITE_EXCHANGE_COMPATIBILITY_VER_1, false);
- /** */
- private final boolean newMvccCrd;
-
- /** Currently running mvcc queries, initialized when mvcc coordinator is changed. */
- private Map<UUID, GridLongList> activeQueries;
-
/**
* @param crd Coordinator flag.
- * @param newMvccCrd {@code True} if new coordinator assigned during this exchange.
* @param fut Exchange future.
*/
- public ExchangeContext(boolean crd, boolean newMvccCrd, GridDhtPartitionsExchangeFuture fut) {
- this.newMvccCrd = newMvccCrd;
-
+ public ExchangeContext(boolean crd, GridDhtPartitionsExchangeFuture fut) {
int protocolVer = exchangeProtocolVersion(fut.firstEventCache().minimumNodeVersion());
if (compatibilityNode || (crd && fut.localJoinExchange())) {
@@ -137,34 +124,6 @@ public class ExchangeContext {
return merge;
}
- /**
- * @return {@code True} if new node assigned as mvcc coordinator node during this exchange.
- */
- public boolean newMvccCoordinator() {
- return newMvccCrd;
- }
-
- /**
- * @return Active queries.
- */
- public Map<UUID, GridLongList> activeQueries() {
- return activeQueries;
- }
-
- /**
- * @param nodeId Node ID.
- * @param nodeQueries Node queries.
- */
- public void addActiveQueries(UUID nodeId, @Nullable GridLongList nodeQueries) {
- if (nodeQueries == null)
- return;
-
- if (activeQueries == null)
- activeQueries = new HashMap<>();
-
- activeQueries.put(nodeId, nodeQueries);
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(ExchangeContext.class, this);
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index f811f6d..e20855f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -32,7 +32,6 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
@@ -254,10 +253,6 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
return aff0.cachedAffinity(topVer, lastAffChangedTopVer);
}
- public MvccCoordinator mvccCoordinator(AffinityTopologyVersion topVer) {
- return assignment(topVer).mvccCoordinator();
- }
-
/**
* @param key Key to check.
* @param topVer Topology version.
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index fa63b7e..fed5686 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -2299,14 +2299,6 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
ClusterNode node = evt.eventNode();
- if ((evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT) &&
- node.equals(cctx.coordinators().currentCoordinator())) {
- if (log.isInfoEnabled())
- log.info("Stop merge, need exchange for mvcc coordinator failure: " + node);
-
- break;
- }
-
if (!curFut.context().supportsMergeExchanges(node)) {
if (log.isInfoEnabled())
log.info("Stop merge, node does not support merge: " + node);
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/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 946b2a2..cc07d77 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
@@ -898,8 +898,6 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
tx.onePhaseCommit(),
tx.activeCachesDeploymentEnabled());
- res.mvccSnapshot(tx.mvccSnapshot());
-
if (prepErr == null) {
if (tx.needReturnValue() || tx.nearOnOriginatingNode() || tx.hasInterceptor())
addDhtValues(res);
@@ -1258,27 +1256,6 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
}
}
- if (req.requestMvccCounter()) {
- assert last;
-
- assert tx.txState().mvccEnabled();
-
- try {
- // Request snapshot locally only because
- // Mvcc Coordinator is expected to be local.
- MvccSnapshot snapshot = cctx.coordinators().tryRequestSnapshotLocal(tx);
-
- assert snapshot != null : tx.topologyVersion();
-
- tx.mvccSnapshot(snapshot);
- }
- catch (ClusterTopologyCheckedException e) {
- onDone(e);
-
- return;
- }
- }
-
onEntriesLocked();
// We are holding transaction-level locks for entries here, so we can get next write version.
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/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 e9d9442..9d81405 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
@@ -91,7 +91,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.topology.Grid
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionsStateValidator;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.persistence.DatabaseLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotDiscoveryMessage;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
@@ -100,7 +99,6 @@ 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;
@@ -698,25 +696,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
boolean crdNode = crd != null && crd.isLocal();
- 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);
+ exchCtx = new ExchangeContext(crdNode, this);
cctx.exchange().exchangerBlockingSectionBegin();
- try {
- cctx.kernalContext().coordinators().onExchangeStart(mvccCrd, exchCtx, crd);
- }
- finally {
- cctx.exchange().exchangerBlockingSectionEnd();
- }
-
assert state == null : state;
if (crdNode)
@@ -726,8 +709,6 @@ 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() +
@@ -817,7 +798,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
cctx.cache().registrateProxyRestart(resolveCacheRequests(exchActions), afterLsnrCompleteFut);
- updateTopologies(crdNode, cctx.coordinators().currentCoordinator());
+ updateTopologies(crdNode);
switch (exchange) {
case ALL: {
@@ -977,10 +958,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
* Updates topology versions and discovery caches on all topologies.
*
* @param crd Coordinator flag.
- * @param mvccCrd Mvcc coordinator.
* @throws IgniteCheckedException If failed.
*/
- private void updateTopologies(boolean crd, MvccCoordinator mvccCrd) throws IgniteCheckedException {
+ private void updateTopologies(boolean crd) throws IgniteCheckedException {
for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
if (grp.isLocal())
continue;
@@ -1017,7 +997,6 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
top.updateTopologyVersion(
this,
events().discoveryCache(),
- mvccCrd,
updSeq,
cacheGroupStopping(grp.groupId()));
}
@@ -1032,7 +1011,6 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
for (GridClientPartitionTopology top : cctx.exchange().clientTopologies()) {
top.updateTopologyVersion(this,
events().discoveryCache(),
- mvccCrd,
-1,
cacheGroupStopping(top.groupId()));
}
@@ -1838,12 +1816,6 @@ 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())
@@ -2097,8 +2069,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
tryToPerformLocalSnapshotOperation();
if (err == null)
- cctx.coordinators().onExchangeDone(exchCtx.newMvccCoordinator(), exchCtx.events().discoveryCache(),
- exchCtx.activeQueries());
+ cctx.coordinators().onExchangeDone(exchCtx.events().discoveryCache());
// Create and destory caches and cache proxies.
cctx.cache().onExchangeDone(initialVersion(), exchActions, err);
@@ -2673,9 +2644,6 @@ 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;
@@ -3235,9 +3203,6 @@ 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/54770480/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 b5de155..8db7176 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,7 +30,6 @@ import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.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;
@@ -109,9 +108,6 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
*/
private GridDhtPartitionsFullMessage finishMsg;
- /** */
- private GridLongList activeQryTrackers;
-
/**
* Required by {@link Externalizable}.
*/
@@ -136,20 +132,6 @@ 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) {
@@ -474,65 +456,58 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
switch (writer.state()) {
case 6:
- if (!writer.writeMessage("activeQryTrackers", activeQryTrackers))
- return false;
-
- writer.incrementState();
-
- case 7:
if (!writer.writeBoolean("client", client))
return false;
writer.incrementState();
- case 8:
+ case 7:
if (!writer.writeMap("dupPartsData", dupPartsData, MessageCollectionItemType.INT, MessageCollectionItemType.INT))
return false;
writer.incrementState();
- case 9:
+ case 8:
if (!writer.writeByteArray("errBytes", errBytes))
return false;
writer.incrementState();
- case 10:
+ case 9:
if (!writer.writeMessage("finishMsg", finishMsg))
return false;
writer.incrementState();
- case 11:
+ case 10:
if (!writer.writeCollection("grpsAffRequest", grpsAffRequest, MessageCollectionItemType.INT))
return false;
writer.incrementState();
- case 12:
+ case 11:
if (!writer.writeByteArray("partCntrsBytes", partCntrsBytes))
return false;
writer.incrementState();
- case 13:
+ case 12:
if (!writer.writeByteArray("partHistCntrsBytes", partHistCntrsBytes))
return false;
writer.incrementState();
- case 14:
+ case 13:
if (!writer.writeByteArray("partsBytes", partsBytes))
return false;
writer.incrementState();
- case 15:
+ case 14:
if (!writer.writeByteArray("partsSizesBytes", partsSizesBytes))
return false;
writer.incrementState();
-
}
return true;
@@ -550,14 +525,6 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
switch (reader.state()) {
case 6:
- activeQryTrackers = reader.readMessage("activeQryTrackers");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 7:
client = reader.readBoolean("client");
if (!reader.isLastRead())
@@ -565,7 +532,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 8:
+ case 7:
dupPartsData = reader.readMap("dupPartsData", MessageCollectionItemType.INT, MessageCollectionItemType.INT, false);
if (!reader.isLastRead())
@@ -573,7 +540,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 9:
+ case 8:
errBytes = reader.readByteArray("errBytes");
if (!reader.isLastRead())
@@ -581,7 +548,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 10:
+ case 9:
finishMsg = reader.readMessage("finishMsg");
if (!reader.isLastRead())
@@ -589,7 +556,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 11:
+ case 10:
grpsAffRequest = reader.readCollection("grpsAffRequest", MessageCollectionItemType.INT);
if (!reader.isLastRead())
@@ -597,7 +564,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 12:
+ case 11:
partCntrsBytes = reader.readByteArray("partCntrsBytes");
if (!reader.isLastRead())
@@ -605,7 +572,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 13:
+ case 12:
partHistCntrsBytes = reader.readByteArray("partHistCntrsBytes");
if (!reader.isLastRead())
@@ -613,7 +580,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 14:
+ case 13:
partsBytes = reader.readByteArray("partsBytes");
if (!reader.isLastRead())
@@ -621,14 +588,13 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
reader.incrementState();
- case 15:
+ case 14:
partsSizesBytes = reader.readByteArray("partsSizesBytes");
if (!reader.isLastRead())
return false;
reader.incrementState();
-
}
return reader.afterMessageRead(GridDhtPartitionsSingleMessage.class);
@@ -641,7 +607,7 @@ public class GridDhtPartitionsSingleMessage extends GridDhtPartitionsAbstractMes
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 16;
+ return 15;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
index b1d1048..fc8975b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
@@ -48,7 +48,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridPartitionStateMap;
@@ -202,11 +201,6 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public MvccCoordinator mvccCoordinator() {
- throw new UnsupportedOperationException();
- }
-
- /** {@inheritDoc} */
@Override public boolean holdsLock() {
return lock.isWriteLockedByCurrentThread() || lock.getReadHoldCount() > 0;
}
@@ -215,7 +209,6 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
@Override public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
DiscoCache discoCache,
- MvccCoordinator mvccCrd,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java
index 2f5e57b..81eab84 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java
@@ -38,7 +38,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.jetbrains.annotations.Nullable;
@@ -78,7 +77,6 @@ public interface GridDhtPartitionTopology {
public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
DiscoCache discoCache,
- MvccCoordinator mvccCrd,
long updateSeq,
boolean stopping
) throws IgniteInterruptedCheckedException;
@@ -432,6 +430,4 @@ public interface GridDhtPartitionTopology {
* @param updateRebalanceVer {@code True} if need check rebalance state.
*/
public void onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, boolean updateRebalanceVer);
-
- public MvccCoordinator mvccCoordinator();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
index fb30ecd..887f3d0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
@@ -56,7 +56,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridLongList;
@@ -152,9 +151,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** */
private volatile AffinityTopologyVersion rebalancedTopVer = AffinityTopologyVersion.NONE;
- /** */
- private volatile MvccCoordinator mvccCrd;
-
/**
* @param ctx Cache shared context.
* @param grp Cache group.
@@ -247,11 +243,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public MvccCoordinator mvccCoordinator() {
- return mvccCrd;
- }
-
- /** {@inheritDoc} */
@Override public boolean holdsLock() {
return lock.isWriteLockedByCurrentThread() || lock.getReadHoldCount() > 0;
}
@@ -260,7 +251,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
@Override public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
@NotNull DiscoCache discoCache,
- MvccCoordinator mvccCrd,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
@@ -287,7 +277,6 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
lastTopChangeVer = exchTopVer;
this.discoCache = discoCache;
- this.mvccCrd = mvccCrd;
}
finally {
lock.writeLock().unlock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/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 a7142f9..ef88a6b 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,7 +38,6 @@ 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;
@@ -59,7 +58,6 @@ 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;
@@ -196,9 +194,6 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
if (keyLockFut != null)
keyLockFut.onDone(e);
-
- if (mvccVerFut != null)
- mvccVerFut.onDone();
}
/** {@inheritDoc} */
@@ -348,25 +343,11 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
boolean hasNearCache = false;
- MvccCoordinator mvccCrd = null;
-
for (IgniteTxEntry write : writes) {
map(write, topVer, mappings, txMapping, remap, topLocked);
- GridCacheContext cctx = write.context();
-
- if (cctx.isNear())
+ if (write.context().isNear())
hasNearCache = true;
-
- if (cctx.mvccEnabled() && mvccCrd == null) {
- mvccCrd = cctx.affinity().mvccCoordinator(topVer);
-
- if (mvccCrd == null) {
- onDone(noCoordinatorError(topVer));
-
- return;
- }
- }
}
for (IgniteTxEntry read : reads)
@@ -382,8 +363,6 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
return;
}
- assert !tx.txState().mvccEnabled() || mvccCrd != null || F.isEmpty(writes);
-
tx.addEntryMapping(mappings.values());
cctx.mvcc().recheckPendingLocks();
@@ -395,16 +374,12 @@ 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);
- lockCnt++;
-
add((IgniteInternalFuture)fut);
if (m.primary().isLocal() && m.hasNearCacheEntries() && m.hasColocatedCacheEntries()) {
@@ -413,14 +388,9 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
locNearEntriesFut = fut;
add((IgniteInternalFuture)new MiniFuture(this, m, ++miniId));
-
- lockCnt++;
}
}
- if (mvccCrd != null)
- initMvccVersionFuture(lockCnt, remap);
-
Collection<IgniteInternalFuture<?>> futs = (Collection)futures();
Iterator<IgniteInternalFuture<?>> it = futs.iterator();
@@ -984,9 +954,6 @@ 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/54770480/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 809d149..d47420f 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,7 +43,6 @@ 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;
@@ -69,7 +68,6 @@ 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;
@@ -380,18 +378,6 @@ 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();
@@ -436,8 +422,6 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
boolean hasNearCache = false;
- MvccCoordinator mvccCrd = null;
-
for (IgniteTxEntry write : writes) {
write.clearEntryReadVersion();
@@ -447,16 +431,6 @@ 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;
@@ -496,11 +470,6 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
return;
}
- assert !tx.txState().mvccEnabled() || mvccCrd != null;
-
- if (mvccCrd != null)
- initMvccVersionFuture(keyLockFut != null ? 2 : 1, remap);
-
if (keyLockFut != null)
keyLockFut.onAllKeysAdded();
@@ -524,12 +493,8 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
private void proceedPrepare(final Queue<GridDistributedTxMapping> mappings) {
final GridDistributedTxMapping m = mappings.poll();
- if (m == null) {
- if (mvccVerFut != null)
- mvccVerFut.onLockReceived();
-
+ if (m == null)
return;
- }
proceedPrepare(m, mappings);
}
@@ -1034,8 +999,6 @@ 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/54770480/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 6f541d3..913e9bc 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,15 +18,11 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
import java.util.Collection;
-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.transactions.IgniteTxTimeoutCheckedException;
@@ -35,7 +31,6 @@ import org.apache.ignite.internal.util.future.GridFutureAdapter;
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;
-import org.apache.ignite.lang.IgniteInClosure;
import org.jetbrains.annotations.Nullable;
/**
@@ -46,17 +41,9 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
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.
@@ -214,29 +201,6 @@ 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.
*/
@@ -314,82 +278,4 @@ 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/54770480/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 ef41f5e..5b6d440 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
@@ -34,11 +34,8 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
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.topology.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.distributed.dht.topology.GridDhtPartitionTopology;
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;
@@ -51,7 +48,6 @@ 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;
@@ -283,8 +279,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
AffinityTopologyVersion topVer = tx.topologyVersion();
- MvccCoordinator mvccCrd = null;
-
boolean hasNearCache = false;
Map<UUID, Collection<UUID>> txNodes;
@@ -322,16 +316,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
else
nodes = cacheCtx.affinity().nodesByKey(txEntry.key(), topVer);
- if (tx.mvccSnapshot() == null && mvccCrd == null && cacheCtx.mvccEnabled()) {
- mvccCrd = cacheCtx.affinity().mvccCoordinator(topVer);
-
- 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() +
@@ -357,8 +341,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
txNodes = txMapping.transactionNodes();
}
- assert !tx.txState().mvccEnabled() || tx.mvccSnapshot() != null || mvccCrd != null;
-
tx.transactionNodes(txNodes);
if (!hasNearCache)
@@ -377,16 +359,6 @@ 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(txNodes,
@@ -395,8 +367,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
m.nearEntriesReads(),
m.nearEntriesWrites());
- nearReq.requestMvccCounter(needCntr);
-
prepareLocal(nearReq, m, ++miniId, true);
GridNearTxPrepareRequest colocatedReq = createRequest(txNodes,
@@ -410,8 +380,6 @@ 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());
}
}
@@ -422,8 +390,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
m.reads(),
m.writes());
- req.requestMvccCounter(needCntr);
-
final MiniFuture fut = new MiniFuture(m, ++miniId);
req.miniId(fut.futureId());
@@ -456,16 +422,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
}
}
- if (mvccCrd != null) {
- assert !tx.onePhaseCommit();
-
- MvccSnapshotFutureExt fut = new MvccSnapshotFutureExt();
-
- cctx.coordinators().requestSnapshotAsync(tx, fut);
-
- add((IgniteInternalFuture)fut);
- }
-
markInitialized();
}
@@ -513,35 +469,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
"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> {
/** */
@@ -581,9 +508,6 @@ 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/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
index c3eae2c..87546aa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
@@ -59,9 +59,6 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** */
private static final int ALLOW_WAIT_TOP_FUT_FLAG_MASK = 0x10;
- /** */
- private static final int REQUEST_MVCC_CNTR_FLAG_MASK = 0x20;
-
/** Future ID. */
private IgniteUuid futId;
@@ -160,20 +157,6 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
/**
- * @return {@code True} if need request MVCC counter on primary node on prepare step.
- */
- public boolean requestMvccCounter() {
- return isFlag(REQUEST_MVCC_CNTR_FLAG_MASK);
- }
-
- /**
- * @param val {@code True} if need request MVCC counter on primary node on prepare step.
- */
- public void requestMvccCounter(boolean val) {
- setFlag(val, REQUEST_MVCC_CNTR_FLAG_MASK);
- }
-
- /**
* @return {@code True} if it is safe for first client request to wait for topology future
* completion.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/54770480/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
index 67f7116..ef547cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
@@ -33,7 +33,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareResponse;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
@@ -98,9 +97,6 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
/** Not {@code null} if client node should remap transaction. */
private AffinityTopologyVersion clientRemapVer;
- /** */
- private MvccSnapshot mvccSnapshot;
-
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -150,20 +146,6 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
}
/**
- * @param mvccSnapshot Mvcc info.
- */
- public void mvccSnapshot(MvccSnapshot mvccSnapshot) {
- this.mvccSnapshot = mvccSnapshot;
- }
-
- /**
- * @return Mvcc info.
- */
- @Nullable public MvccSnapshot mvccSnapshot() {
- return mvccSnapshot;
- }
-
- /**
* @return One-phase commit state on primary node.
*/
public boolean onePhaseCommit() {
@@ -407,41 +389,34 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
writer.incrementState();
case 16:
- if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
- return false;
-
- writer.incrementState();
-
- case 17:
if (!writer.writeCollection("ownedValKeys", ownedValKeys, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
- case 18:
+ case 17:
if (!writer.writeCollection("ownedValVals", ownedValVals, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
- case 19:
+ case 18:
if (!writer.writeCollection("pending", pending, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
- case 20:
+ case 19:
if (!writer.writeMessage("retVal", retVal))
return false;
writer.incrementState();
- case 21:
+ case 20:
if (!writer.writeMessage("writeVer", writeVer))
return false;
writer.incrementState();
-
}
return true;
@@ -499,14 +474,6 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
reader.incrementState();
case 16:
- mvccSnapshot = reader.readMessage("mvccSnapshot");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 17:
ownedValKeys = reader.readCollection("ownedValKeys", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -514,7 +481,7 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
reader.incrementState();
- case 18:
+ case 17:
ownedValVals = reader.readCollection("ownedValVals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -522,7 +489,7 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
reader.incrementState();
- case 19:
+ case 18:
pending = reader.readCollection("pending", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -530,7 +497,7 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
reader.incrementState();
- case 20:
+ case 19:
retVal = reader.readMessage("retVal");
if (!reader.isLastRead())
@@ -538,14 +505,13 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
reader.incrementState();
- case 21:
+ case 20:
writeVer = reader.readMessage("writeVer");
if (!reader.isLastRead())
return false;
reader.incrementState();
-
}
return reader.afterMessageRead(GridNearTxPrepareResponse.class);
@@ -558,7 +524,7 @@ public class GridNearTxPrepareResponse extends GridDistributedTxPrepareResponse
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 22;
+ return 21;
}
/** {@inheritDoc} */