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/12/09 19:31:05 UTC
[1/3] ignite git commit: pending
Repository: ignite
Updated Branches:
refs/heads/ignite-10537 [created] 3a8b66f31
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/StaticMvccQueryTracker.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/StaticMvccQueryTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/StaticMvccQueryTracker.java
index 52fb1db..123ddf1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/StaticMvccQueryTracker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/StaticMvccQueryTracker.java
@@ -80,11 +80,6 @@ public class StaticMvccQueryTracker implements MvccQueryTracker {
}
/** {@inheritDoc} */
- @Override public IgniteInternalFuture<Void> onDone(@NotNull GridNearTxLocal tx, boolean commit) {
- throw new UnsupportedOperationException("Operation is not supported.");
- }
-
- /** {@inheritDoc} */
@Override public long onMvccCoordinatorChange(MvccCoordinator newCrd) {
return MVCC_TRACKER_ID_NA;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/MvccWaitTxsRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/MvccWaitTxsRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/MvccWaitTxsRequest.java
deleted file mode 100644
index 5c6d4aa..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/MvccWaitTxsRequest.java
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache.mvcc.msg;
-
-import java.nio.ByteBuffer;
-
-import org.apache.ignite.internal.util.GridLongList;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
-
-/**
- *
- */
-public class MvccWaitTxsRequest implements MvccMessage {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private long futId;
-
- /** */
- private GridLongList txs;
-
- /**
- *
- */
- public MvccWaitTxsRequest() {
- // No-op.
- }
-
- /**
- * @param futId Future ID.
- * @param txs Transactions to wait for.
- */
- public MvccWaitTxsRequest(long futId, GridLongList txs) {
- assert txs != null && !txs.isEmpty() : txs;
-
- this.futId = futId;
- this.txs = txs;
- }
-
- /**
- * @return Future ID.
- */
- public long futureId() {
- return futId;
- }
-
- /**
- * @return Transactions to wait for.
- */
- public GridLongList transactions() {
- return txs;
- }
-
- /** {@inheritDoc} */
- @Override public boolean waitForCoordinatorInit() {
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public boolean processedFromNioThread() {
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
- writer.setBuffer(buf);
-
- if (!writer.isHeaderWritten()) {
- if (!writer.writeHeader(directType(), fieldsCount()))
- return false;
-
- writer.onHeaderWritten();
- }
-
- switch (writer.state()) {
- case 0:
- if (!writer.writeLong("futId", futId))
- return false;
-
- writer.incrementState();
-
- case 1:
- if (!writer.writeMessage("txs", txs))
- return false;
-
- writer.incrementState();
-
- }
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
- reader.setBuffer(buf);
-
- if (!reader.beforeMessageRead())
- return false;
-
- switch (reader.state()) {
- case 0:
- futId = reader.readLong("futId");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 1:
- txs = reader.readMessage("txs");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- }
-
- return reader.afterMessageRead(MvccWaitTxsRequest.class);
- }
-
- /** {@inheritDoc} */
- @Override public short directType() {
- return 142;
- }
-
- /** {@inheritDoc} */
- @Override public byte fieldsCount() {
- return 2;
- }
-
- /** {@inheritDoc} */
- @Override public void onAckReceived() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MvccWaitTxsRequest.class, this);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 47a0276..5f44606 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -17,8 +17,6 @@
package org.apache.ignite.internal.processors.cache.transactions;
-import javax.cache.expiry.ExpiryPolicy;
-import javax.cache.processor.EntryProcessor;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
@@ -39,6 +37,8 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
@@ -1659,8 +1659,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
/*closure name */recordEvt ? F.first(txEntry.entryProcessors()).get1() : null,
resolveTaskName(),
null,
- keepBinary,
- null); // TODO IGNITE-7371
+ keepBinary);
}
boolean modified = false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 319073b..ab7fdd3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -1789,8 +1789,7 @@ public class IgniteTxHandler {
/*transformClo*/null,
tx.resolveTaskName(),
/*expiryPlc*/null,
- /*keepBinary*/true,
- null); // TODO IGNITE-7371
+ /*keepBinary*/true);
if (val == null)
val = cacheCtx.toCacheObject(cacheCtx.store().load(null, entry.key()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index df43615..b1e2731 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -68,7 +68,6 @@ import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.lang.GridClosureException;
import org.apache.ignite.internal.util.lang.GridTuple;
@@ -157,9 +156,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
protected CacheWriteSynchronizationMode syncMode;
/** */
- private GridLongList mvccWaitTxs;
-
- /** */
private volatile boolean qryEnlisted;
/** Whether to skip update of completed versions map during rollback caused by empty update set in MVCC TX. */
@@ -225,10 +221,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
txState = implicitSingle ? new IgniteTxImplicitSingleStateImpl() : new IgniteTxStateImpl();
}
- public GridLongList mvccWaitTransactions() {
- return mvccWaitTxs;
- }
-
/**
* @return Transaction write synchronization mode.
*/
@@ -576,8 +568,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
cctx.tm().addCommittedTx(this);
if (!empty) {
- assert mvccWaitTxs == null;
-
batchStoreCommit(writeEntries());
WALPointer ptr = null;
@@ -750,17 +740,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- null,
- mvccSnapshot());
+ null);
- if (updRes.success()) {
+ if (updRes.success())
txEntry.updateCounter(updRes.updateCounter());
- GridLongList waitTxs = updRes.mvccWaitTransactions();
-
- updateWaitTxs(waitTxs);
- }
-
if (updRes.loggedPointer() != null)
ptr = updRes.loggedPointer();
@@ -790,8 +774,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer0,
- null,
- mvccSnapshot())
+ null)
);
}
}
@@ -813,17 +796,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- null,
- mvccSnapshot());
+ null);
- if (updRes.success()) {
+ if (updRes.success())
txEntry.updateCounter(updRes.updateCounter());
- GridLongList waitTxs = updRes.mvccWaitTransactions();
-
- updateWaitTxs(waitTxs);
- }
-
if (updRes.loggedPointer() != null)
ptr = updRes.loggedPointer();
@@ -848,8 +825,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer0,
- null,
- mvccSnapshot())
+ null)
);
}
}
@@ -982,18 +958,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/**
- * @param waitTxs Tx ids to wait for.
- */
- private void updateWaitTxs(@Nullable GridLongList waitTxs) {
- if (waitTxs != null) {
- if (this.mvccWaitTxs == null)
- this.mvccWaitTxs = waitTxs;
- else
- this.mvccWaitTxs.addAll(waitTxs);
- }
- }
-
- /**
* Safely performs {@code updateClojure} operation on near cache entry with given {@code entryKey}.
* In case of {@link GridCacheEntryRemovedException} operation will be retried.
*
@@ -1237,8 +1201,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
null,
resolveTaskName(),
null,
- txEntry.keepBinary(),
- null); // TODO IGNITE-7371
+ txEntry.keepBinary());
}
}
else {
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
index 6fb200d..a4cdfda 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
@@ -408,8 +408,7 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
Object transformClo,
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
- boolean keepBinary,
- MvccSnapshot mvccVer) {
+ boolean keepBinary) {
return val;
}
@@ -426,7 +425,6 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- MvccSnapshot mvccVer,
@Nullable ReaderArguments args) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert false;
@@ -444,7 +442,6 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs) {
assert false;
@@ -477,8 +474,7 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- MvccSnapshot mvccVer
+ @Nullable Long updateCntr
)
throws IgniteCheckedException, GridCacheEntryRemovedException
{
@@ -592,9 +588,8 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- MvccSnapshot mvccVer
- ) throws IgniteCheckedException, GridCacheEntryRemovedException {
+ @Nullable Long updateCntr
+ ) throws IgniteCheckedException, GridCacheEntryRemovedException {
obsoleteVer = ver;
val = null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractTest.java
index a6100d8..4751d86 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccAbstractTest.java
@@ -115,7 +115,8 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
*/
public abstract class CacheMvccAbstractTest extends GridCommonAbstractTest {
/** */
- private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true)
+ .setAddresses(Collections.singleton("127.0.0.1:47500"));
/** */
protected static final ObjectCodec<Integer> INTEGER_CODEC = new IntegerCodec();
[3/3] ignite git commit: pending
Posted by gv...@apache.org.
pending
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3a8b66f3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3a8b66f3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3a8b66f3
Branch: refs/heads/ignite-10537
Commit: 3a8b66f31a0e7efa66c522fab25d55097c61a87b
Parents: 7301ada
Author: Igor Seliverstov <gv...@gmail.com>
Authored: Sun Dec 9 22:30:45 2018 +0300
Committer: Igor Seliverstov <gv...@gmail.com>
Committed: Sun Dec 9 22:30:45 2018 +0300
----------------------------------------------------------------------
.../discovery/GridDiscoveryManager.java | 4 +-
.../processors/cache/GridCacheAdapter.java | 6 +
.../GridDhtPartitionsExchangeFuture.java | 2 +-
.../processors/cache/mvcc/MvccCoordinator.java | 54 ++--
.../cache/mvcc/MvccDiscoveryData.java | 52 ----
.../processors/cache/mvcc/MvccProcessor.java | 24 +-
.../cache/mvcc/MvccProcessorImpl.java | 282 +++++++++----------
.../cache/mvcc/MvccQueryTrackerImpl.java | 6 +-
.../processors/cache/mvcc/MvccUtils.java | 2 +-
.../ignite/internal/util/GridLongList.java | 1 +
10 files changed, 191 insertions(+), 242 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/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 5abe63c..04a4c5e 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
@@ -785,7 +785,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
discoWrk.discoCache = discoCache;
if (!isLocDaemon && !ctx.clientDisconnected()) {
- ctx.cache().context().coordinators().onLocalJoin(discoEvt);
+ ctx.cache().context().coordinators().onLocalJoin(discoEvt, discoCache);
ctx.cache().context().exchange().onLocalJoin(discoEvt, discoCache);
@@ -845,6 +845,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
((IgniteKernal)ctx.grid()).onReconnected(clusterRestarted);
+ ctx.cache().context().coordinators().onLocalJoin(localJoinEvent(), discoCache);
+
ctx.cache().context().exchange().onLocalJoin(localJoinEvent(), discoCache);
ctx.cluster().clientReconnectFuture().listen(new CI1<IgniteFuture<?>>() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 1cd94d2..4ac9774 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -4316,6 +4316,12 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
AffinityTopologyVersion awaitVer = new AffinityTopologyVersion(
topVer.topologyVersion() + 1, 0);
+ U.warn(log, X.getFullStackTrace(e));
+
+ U.warn(log,"txTopVer=" + topVer.topologyVersion() + ", waitVer=" + awaitVer);
+
+ U.warn(log, "curCrd=" + ctx.shared().coordinators().currentCoordinator());
+
ctx.shared().exchange().affinityReadyFuture(awaitVer).get();
continue;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/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 ffc55a9..7849f1e 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
@@ -2070,7 +2070,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
tryToPerformLocalSnapshotOperation();
if (err == null)
- cctx.coordinators().onExchangeDone(exchCtx.events().discoveryCache());
+ cctx.coordinators().onExchangeDone(initialVersion(), res, exchCtx.events().discoveryCache());
// Create and destory caches and cache proxies.
cctx.cache().onExchangeDone(initialVersion(), exchActions, err);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinator.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinator.java
index 045177a..a9dd18e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinator.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinator.java
@@ -31,38 +31,46 @@ public class MvccCoordinator implements Serializable {
private static final long serialVersionUID = 0L;
/** */
+ @GridToStringInclude
+ private final AffinityTopologyVersion topVer;
+
+ /** */
private final UUID nodeId;
/**
* Unique coordinator version, increases when new coordinator is assigned,
* can differ from topVer if we decide to assign coordinator manually.
*/
- private final long crdVer;
+ private final long ver;
/** */
- @GridToStringInclude
- private final AffinityTopologyVersion topVer;
+ private final boolean local;
- /**
- * @param nodeId Coordinator node ID.
- * @param crdVer Coordinator version.
+ /**
* @param topVer Topology version when coordinator was assigned.
*/
- public MvccCoordinator(UUID nodeId, long crdVer, AffinityTopologyVersion topVer) {
- assert nodeId != null;
- assert crdVer > 0 : crdVer;
- assert topVer != null;
+ public MvccCoordinator(AffinityTopologyVersion topVer) {
+ this(topVer, null, 0, false);
+ }
+ /**
+ * @param topVer Topology version when coordinator was assigned.
+ * @param nodeId Coordinator node ID.
+ * @param ver Coordinator version.
+ */
+ public MvccCoordinator(AffinityTopologyVersion topVer, UUID nodeId, long ver,
+ boolean local) {
this.nodeId = nodeId;
- this.crdVer = crdVer;
+ this.ver = ver;
this.topVer = topVer;
+ this.local = local;
}
/**
- * @return Unique coordinator version.
+ * @return Topology version when coordinator was assigned.
*/
- public long coordinatorVersion() {
- return crdVer;
+ public AffinityTopologyVersion topologyVersion() {
+ return topVer;
}
/**
@@ -73,10 +81,18 @@ public class MvccCoordinator implements Serializable {
}
/**
- * @return Topology version when coordinator was assigned.
+ * @return Unique coordinator version.
*/
- public AffinityTopologyVersion topologyVersion() {
- return topVer;
+ public long version() {
+ return ver;
+ }
+
+ /**
+ *
+ * @return {@code True} if the coordinator is local.
+ */
+ public boolean local() {
+ return local;
}
/** {@inheritDoc} */
@@ -89,12 +105,12 @@ public class MvccCoordinator implements Serializable {
MvccCoordinator that = (MvccCoordinator)o;
- return crdVer == that.crdVer;
+ return ver == that.ver;
}
/** {@inheritDoc} */
@Override public int hashCode() {
- return (int)(crdVer ^ (crdVer >>> 32));
+ return (int)(ver ^ (ver >>> 32));
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccDiscoveryData.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccDiscoveryData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccDiscoveryData.java
deleted file mode 100644
index d2e936f..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccDiscoveryData.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache.mvcc;
-
-import org.apache.ignite.internal.util.typedef.internal.S;
-
-import java.io.Serializable;
-
-/**
- * MVCC discovery data to be shared between nodes on join.
- */
-public class MvccDiscoveryData implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Current coordinator. */
- private MvccCoordinator crd;
-
- /**
- * @param crd Coordinator.
- */
- public MvccDiscoveryData(MvccCoordinator crd) {
- this.crd = crd;
- }
-
- /**
- * @return Current coordinator.
- */
- public MvccCoordinator coordinator() {
- return crd;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MvccDiscoveryData.class, this);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/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 161342f..5a648b9 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,7 +17,6 @@
package org.apache.ignite.internal.processors.cache.mvcc;
-import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -27,9 +26,10 @@ 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.processors.GridProcessor;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
-import org.apache.ignite.internal.util.GridLongList;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -41,30 +41,22 @@ public interface MvccProcessor extends GridProcessor {
*
* @param evt Discovery event.
*/
- void onLocalJoin(DiscoveryEvent evt);
+ void onLocalJoin(DiscoveryEvent evt, DiscoCache discoCache);
/**
* Exchange done callback.
*
+ * @param initialVersion Initial exchange version.
+ * @param resultVersion Result exchange version, defers from initial if several exchange were merged.
* @param discoCache Disco cache.
*/
- void onExchangeDone(DiscoCache discoCache);
-
- /**
- * @param nodeId Node ID
- * @param activeQueries Active queries.
- */
- void processClientActiveQueries(UUID nodeId, @Nullable GridLongList activeQueries);
+ void onExchangeDone(AffinityTopologyVersion initialVersion,
+ AffinityTopologyVersion resultVersion, DiscoCache discoCache);
/**
* @return Coordinator.
*/
- @Nullable MvccCoordinator currentCoordinator();
-
- /**
- * @return Current coordinator node ID.
- */
- UUID currentCoordinatorId();
+ @NotNull MvccCoordinator currentCoordinator();
/**
* @param crdVer Mvcc coordinator version.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/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 d026607..b29c782 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
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.mvcc;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -40,7 +41,6 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
-import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteDiagnosticPrepareContext;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
@@ -51,7 +51,6 @@ 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.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;
@@ -106,6 +105,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
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;
@@ -156,11 +156,8 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
MvccProcessorImpl.crdC = crdC;
}
- /** Topology version when local node was assigned as coordinator. */
- private volatile long crdVer;
-
/** */
- private volatile MvccCoordinator curCrd;
+ private volatile MvccCoordinator curCrd = new MvccCoordinator(AffinityTopologyVersion.NONE);
/** */
private TxLog txLog;
@@ -218,9 +215,6 @@ 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.
*/
@@ -237,12 +231,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** {@inheritDoc} */
@Override public void start() throws IgniteCheckedException {
- ctx.event().addLocalEventListener(new GridLocalEventListener() {
- @Override public void onEvent(Event evt) {
- onDiscovery((DiscoveryEvent)evt);
- }
- },
- EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_NODE_JOINED);
+ ctx.event().addDiscoveryEventListener(this::onDiscovery, EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_CLIENT_NODE_DISCONNECTED, EVT_NODE_JOINED);
ctx.io().addMessageListener(TOPIC_CACHE_COORDINATOR, new CoordinatorMessageListener());
@@ -370,81 +359,78 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** {@inheritDoc} */
- @Override public void onExchangeDone(DiscoCache discoCache) {
+ @Override public void onExchangeDone(AffinityTopologyVersion initialVersion,
+ AffinityTopologyVersion resultVersion, DiscoCache discoCache) {
MvccCoordinator curCrd0 = curCrd;
- if (crdChanged) {
+ if (coordinatorChanged(curCrd0, initialVersion, resultVersion)) {
// 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());
-
+ if (curCrd0.local())
initFut.onDone();
- }
-
- crdChanged = false;
}
else {
- if (curCrd0 != null && ctx.localNodeId().equals(curCrd0.nodeId()) && discoCache != null)
+ if (curCrd0.local() && discoCache != null)
cleanupOrphanedServerTransactions(discoCache.serverNodes());
}
}
/** {@inheritDoc} */
- @Override public void onLocalJoin(DiscoveryEvent evt) {
- assert evt.type() == EVT_NODE_JOINED && ctx.localNodeId().equals(evt.eventNode().id());
+ @Override public void onLocalJoin(DiscoveryEvent evt, DiscoCache discoCache) {
+ assert evt.type() == EVT_NODE_JOINED && evt.eventNode().isLocal();
+
+ AffinityTopologyVersion topVer = discoCache.version();
+ List<ClusterNode> nodes = discoCache.allNodes();
- onCoordinatorChanged(evt.topologyNodes(), evt.topologyVersion(), false);
+ checkMvccSupported(nodes);
+
+ onCoordinatorChanged(topVer, nodes, false);
}
/**
- * Discovery listener. Note: initial join event is handled by {@link MvccProcessorImpl#onLocalJoin(DiscoveryEvent)}
+ * Discovery listener. Note: initial join event is handled by {@link MvccProcessorImpl#onLocalJoin}
* 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;
+ private void onDiscovery(DiscoveryEvent evt, DiscoCache discoCache) {
+ assert evt.type() == EVT_NODE_FAILED
+ || evt.type() == EVT_NODE_LEFT
+ || evt.type() == EVT_NODE_JOINED
+ || evt.type() == EVT_CLIENT_NODE_DISCONNECTED;
UUID nodeId = evt.eventNode().id();
+ AffinityTopologyVersion topVer = discoCache.version();
+ List<ClusterNode> nodes = discoCache.allNodes();
+
+ checkMvccSupported(nodes);
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;
+ if (curCrd0.nodeId() == null) // Handle join event only if coordinator has not been elected yet.
+ onCoordinatorChanged(topVer, nodes, false);
}
+ else if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ if (curCrd0.nodeId() != null) {
+ // 1. Notify all listeners waiting for a snapshot.
+ onCoordinatorFailed(curCrd0.nodeId());
- // 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);
-
- 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);
- }
+ // 2. Process coordinator change.
+ onCoordinatorChanged(topVer, Collections.emptyList(), false);
}
+ }
+ else if (nodeId.equals(curCrd0.nodeId())) {
+ // 1. Notify all listeners waiting for a snapshot.
+ onCoordinatorFailed(nodeId);
- // 2. Notify acknowledge futures.
- for (WaitAckFuture fut : ackFuts.values())
- fut.onNodeLeft(nodeId);
-
- // 3. Process coordinator change.
- onCoordinatorChanged(evt.topologyNodes(), evt.topologyVersion(), true);
+ // 2. Process coordinator change.
+ onCoordinatorChanged(topVer, discoCache.allNodes(), true);
}
// Process node left event on the current mvcc coordinator.
- else if (curCrd0.nodeId().equals(ctx.localNodeId())) {
+ else if (curCrd0.local()) {
// 1. Notify active queries.
activeQueries.onNodeFailed(nodeId);
@@ -471,65 +457,83 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
}
+ /** */
+ private void onCoordinatorFailed(UUID nodeId) {
+ // 1. Notify all listeners waiting for a snapshot.
+ Map<Long, MvccSnapshotResponseListener> map = snapLsnrs.remove(nodeId);
+
+ if (map != null) {
+ ClusterTopologyCheckedException ex = new ClusterTopologyCheckedException("Failed to request mvcc " +
+ "version, coordinator left: " + nodeId);
+
+ MvccSnapshotResponseListener lsnr;
+
+ for (Long id : map.keySet()) {
+ if ((lsnr = map.remove(id)) != null)
+ lsnr.onError(ex);
+ }
+ }
+
+ // 2. Notify acknowledge futures.
+ for (WaitAckFuture fut : ackFuts.values())
+ fut.onNodeLeft(nodeId);
+ }
+
/**
* 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);
+ private void onCoordinatorChanged(AffinityTopologyVersion topVer, Collection<ClusterNode> nodes, boolean sndQrys) {
+ MvccCoordinator crd0 = pickMvccCoordinator(nodes, topVer);
- if (newCrd == null)
- return;
+ if (sndQrys && crd0.nodeId() == null)
+ sndQrys = false; // Nowhere to send
- // Update current coordinator, collect active queries and send it to the new coordinator if needed.
- GridLongList activeQryTrackers = null;
+ assert crd0.topologyVersion().compareTo(curCrd.topologyVersion()) >= 0;
- synchronized (activeTrackers) {
- assert curCrd == null || newCrd.topologyVersion().compareTo(curCrd.topologyVersion()) > 0;
+ curCrd = crd0;
- if (sndQrys) {
- activeQryTrackers = new GridLongList();
+ GridLongList qrys = null;
- for (MvccQueryTracker tracker : activeTrackers.values()) {
- long trackerId = tracker.onMvccCoordinatorChange(newCrd);
+ if (sndQrys) {
+ qrys = new GridLongList();
- if (trackerId != MVCC_TRACKER_ID_NA)
- activeQryTrackers.add(trackerId);
- }
- }
+ for (MvccQueryTracker tracker : activeTrackers.values()) {
+ long trackerId = tracker.onMvccCoordinatorChange(crd0);
- curCrd = newCrd;
+ if (trackerId != MVCC_TRACKER_ID_NA)
+ qrys.add(trackerId);
+ }
}
- // Send local active queries to remote coordinator, if needed.
- if (!newCrd.nodeId().equals(ctx.localNodeId())) {
+ if (crd0.local())
+ prevCrdQueries.init(qrys, F.view(nodes, this::supportsMvcc), ctx.discovery());
+ // Send local active queries to remote coordinator, if needed.
+ else if (qrys != null && !qrys.isEmpty()) {
try {
- if (sndQrys)
- sendMessage(newCrd.nodeId(), new MvccActiveQueriesMessage(activeQryTrackers));
+ sendMessage(crd0.nodeId(), new MvccActiveQueriesMessage(qrys));
}
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 = newCrd.coordinatorVersion();
+ }
- if (log.isInfoEnabled())
- log.info("Initialize local node as mvcc coordinator [node=" + ctx.localNodeId() +
- ", crdVer=" + crdVer + ']');
+ /**
+ * @param currCrd Current Mvcc coordinator.
+ * @param from Start topology version.
+ * @param to End topology version
+ * @return {@code True} if coordinator was changed between two passed topology versions.
+ */
+ private boolean coordinatorChanged(MvccCoordinator currCrd, AffinityTopologyVersion from,
+ AffinityTopologyVersion to) {
+ if (currCrd == null)
+ return false;
- prevCrdQueries.init(activeQryTrackers, F.view(nodes, this::supportsMvcc), ctx.discovery());
+ AffinityTopologyVersion crdVersion = currCrd.topologyVersion();
- // Do not complete init future here, because we should wait until all old transactions become terminated.
- }
-
- crdChanged = true;
+ return from.compareTo(crdVersion) <= 0 && to .compareTo(crdVersion) >= 0;
}
/**
@@ -561,23 +565,11 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** {@inheritDoc} */
- @Override public void processClientActiveQueries(UUID nodeId, @Nullable GridLongList activeQueries) {
- prevCrdQueries.addNodeActiveQueries(nodeId, activeQueries);
- }
-
- /** {@inheritDoc} */
- @Override @Nullable public MvccCoordinator currentCoordinator() {
+ @Override @NotNull public MvccCoordinator currentCoordinator() {
return curCrd;
}
/** {@inheritDoc} */
- @Override public UUID currentCoordinatorId() {
- MvccCoordinator curCrd = this.curCrd;
-
- return curCrd != null ? curCrd.nodeId() : null;
- }
-
- /** {@inheritDoc} */
@Override public byte state(MvccVersion ver) throws IgniteCheckedException {
return state(ver.coordinatorVersion(), ver.counter());
}
@@ -661,7 +653,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
@Override public MvccSnapshot tryRequestSnapshotLocal(@Nullable IgniteInternalTx tx) throws ClusterTopologyCheckedException {
MvccCoordinator crd = currentCoordinator();
- if (crd == null)
+ if (crd.nodeId() == null)
throw noCoordinatorError();
if (tx != null) {
@@ -672,7 +664,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
"for the locked topology version. [crd=" + crd + ", tx=" + tx + ']');
}
- if (!ctx.localNodeId().equals(crd.nodeId()) || !initFut.isDone())
+ if (!crd.local() || !initFut.isDone())
return null;
else if (tx != null)
return assignTxSnapshot(0L, ctx.localNodeId(), false);
@@ -698,7 +690,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
@Override public void requestSnapshotAsync(IgniteInternalTx tx, MvccSnapshotResponseListener lsnr) {
MvccCoordinator crd = currentCoordinator();
- if (crd == null) {
+ if (crd.nodeId() == null) {
lsnr.onError(noCoordinatorError());
return;
@@ -759,10 +751,10 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
MvccCoordinator crd = curCrd;
- if (updateVer.coordinatorVersion() == crd.coordinatorVersion())
- return sendTxCommit(crd, new MvccAckRequestTx(futIdCntr.incrementAndGet(), updateVer.counter()));
+ if (crd.version() != updateVer.coordinatorVersion())
+ return new GridFinishedFuture<>();
- return new GridFinishedFuture<>();
+ return sendTxCommit(crd, new MvccAckRequestTx(futIdCntr.incrementAndGet(), updateVer.counter()));
}
/** {@inheritDoc} */
@@ -771,7 +763,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
MvccCoordinator crd = curCrd;
- if (crd.coordinatorVersion() != updateVer.coordinatorVersion())
+ if (crd.version() != updateVer.coordinatorVersion())
return;
MvccAckRequestTx msg = new MvccAckRequestTx((long)-1, updateVer.counter());
@@ -794,17 +786,18 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
@Override public void ackQueryDone(MvccSnapshot snapshot, long qryId) {
MvccCoordinator crd = currentCoordinator();
- if (crd == null || snapshot != null
- && crd.coordinatorVersion() == snapshot.coordinatorVersion()
- && sendQueryDone(crd, new MvccAckRequestQueryCntr(queryTrackCounter(snapshot))))
+ if (crd.nodeId() == null || snapshot == null)
return;
- Message msg = new MvccAckRequestQueryId(qryId);
+ if (crd.version() != snapshot.coordinatorVersion()
+ || !sendQueryDone(crd, new MvccAckRequestQueryCntr(queryTrackCounter(snapshot)))) {
+ Message msg = new MvccAckRequestQueryId(qryId);
- do {
- crd = currentCoordinator();
+ do {
+ crd = currentCoordinator();
+ }
+ while (!sendQueryDone(crd, msg));
}
- while (!sendQueryDone(crd, msg));
}
/** {@inheritDoc} */
@@ -871,13 +864,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/**
* 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);
-
+ private @NotNull MvccCoordinator pickMvccCoordinator(Collection<ClusterNode> nodes, AffinityTopologyVersion topVer) {
ClusterNode crdNode = null;
if (crdC != null) {
@@ -897,15 +886,16 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
}
- MvccCoordinator crd = crdNode != null ? new MvccCoordinator(crdNode.id(), coordinatorVersion(crdNode),
- new AffinityTopologyVersion(topVer, 0)) : null;
+ if (crdNode != null)
+ return new MvccCoordinator(topVer, crdNode.id(), coordinatorVersion(crdNode), crdNode.isLocal());
- 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 + ']');
+// TODO
+// 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;
+ return new MvccCoordinator(topVer);
}
@@ -964,9 +954,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** */
private MvccSnapshotResponse assignTxSnapshot(long futId, UUID nearId, boolean client) {
- assert initFut.isDone();
- assert crdVer != 0;
- assert ctx.localNodeId().equals(currentCoordinatorId());
+ assert initFut.isDone() && curCrd != null && curCrd.local();
MvccSnapshotResponse res = new MvccSnapshotResponse();
@@ -998,7 +986,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
cleanup = prevCrdQueries.previousQueriesDone() ? cleanup - 1 : MVCC_COUNTER_NA;
- res.init(futId, crdVer, ver, MVCC_START_OP_CNTR, cleanup, tracking);
+ res.init(futId, curCrd.version(), ver, MVCC_START_OP_CNTR, cleanup, tracking);
return res;
}
@@ -1121,11 +1109,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
IgniteInternalFuture<VacuumMetrics> runVacuum() {
assert !ctx.clientNode();
- MvccCoordinator crd0 = currentCoordinator();
-
- if (Thread.currentThread().isInterrupted() ||
- crd0 == null ||
- crdVer == 0 && ctx.localNodeId().equals(crd0.nodeId()))
+ if (Thread.currentThread().isInterrupted())
return new GridFinishedFuture<>(new VacuumMetrics());
final GridFutureAdapter<VacuumMetrics> res = new GridFutureAdapter<>();
@@ -1236,8 +1220,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (U.assertionsEnabled()) {
MvccCoordinator crd = currentCoordinator();
- assert crd != null
- && crd.coordinatorVersion() >= snapshot.coordinatorVersion();
+ assert crd.version() >= snapshot.coordinatorVersion();
for (TxKey key : waitMap.keySet()) {
if (!( key.major() == snapshot.coordinatorVersion()
@@ -1318,10 +1301,11 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/**
* @param crd Mvcc coordinator.
* @param msg Message.
- * @return {@code True} if no need to resend the message to a new coordinator.
+ * @return {@code True} if the message was sent successfully.
*/
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
private boolean sendQueryDone(MvccCoordinator crd, Message msg) {
- if (crd == null)
+ if (crd.nodeId() == null)
return true; // no need to send ack;
try {
@@ -1336,7 +1320,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
MvccCoordinator crd0 = currentCoordinator();
// Coordinator is unassigned or still the same.
- return crd0 == null || crd.coordinatorVersion() == crd0.coordinatorVersion();
+ return crd0.nodeId() == null || crd.version() == crd0.version();
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to send query ack [crd=" + crd + ", msg=" + msg + ']', e);
@@ -1502,7 +1486,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
* @param nodeId Node ID.
* @param msg Message.
*/
- private void processCoordinatorActiveQueriesMessage(UUID nodeId, MvccActiveQueriesMessage msg) {
+ private void processActiveQueriesMessage(UUID nodeId, MvccActiveQueriesMessage msg) {
prevCrdQueries.addNodeActiveQueries(nodeId, msg.activeQueries());
}
@@ -1558,7 +1542,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (minQry == null)
minQry = tracking;
- res.init(futId, crdVer, ver, MVCC_READ_OP_CNTR, MVCC_COUNTER_NA, tracking);
+ res.init(futId, curCrd.version(), ver, MVCC_READ_OP_CNTR, MVCC_COUNTER_NA, tracking);
return res;
}
@@ -1663,7 +1647,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (msg0.waitForCoordinatorInit() && !initFut.isDone()) {
initFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() {
@Override public void apply(IgniteInternalFuture<Void> future) {
- assert crdVer != 0L;
+ assert curCrd.local();
processMessage(nodeId, msg);
}
@@ -1695,7 +1679,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
else if (msg instanceof MvccAckRequestQueryId)
processNewCoordinatorQueryAckRequest(nodeId, (MvccAckRequestQueryId)msg);
else if (msg instanceof MvccActiveQueriesMessage)
- processCoordinatorActiveQueriesMessage(nodeId, (MvccActiveQueriesMessage)msg);
+ processActiveQueriesMessage(nodeId, (MvccActiveQueriesMessage)msg);
else if (msg instanceof MvccRecoveryFinishedMessage)
processRecoveryFinishedMessage(nodeId, ((MvccRecoveryFinishedMessage)msg));
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/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 d93a2e9..4db1c2b 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
@@ -147,8 +147,8 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
if (snapshot != null) {
assert crdVer != 0 : this;
- if (crdVer != newCrd.coordinatorVersion()) {
- crdVer = newCrd.coordinatorVersion();
+ if (crdVer != newCrd.version()) {
+ crdVer = newCrd.version();
return id;
}
@@ -223,7 +223,7 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
if (done)
return false;
- crdVer = crd.coordinatorVersion();
+ crdVer = crd.version();
}
return true;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
index 43f87e3..b717895 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
@@ -188,7 +188,7 @@ public class MvccUtils {
if ((state == TxState.NA || state == TxState.PREPARED)
&& (proc.currentCoordinator() == null // Recovery from WAL.
- || mvccCrd < proc.currentCoordinator().coordinatorVersion()))
+ || mvccCrd < proc.currentCoordinator().version()))
state = TxState.ABORTED;
return state;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a8b66f3/modules/core/src/main/java/org/apache/ignite/internal/util/GridLongList.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridLongList.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridLongList.java
index 1c022b0..38ca146 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridLongList.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridLongList.java
@@ -26,6 +26,7 @@ import java.io.ObjectOutput;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.NoSuchElementException;
+import java.util.stream.Collector;
import org.apache.ignite.internal.IgniteCodeGeneratingFail;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
[2/3] ignite git commit: pending
Posted by gv...@apache.org.
pending
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7301ada3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7301ada3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7301ada3
Branch: refs/heads/ignite-10537
Commit: 7301ada36bb64ad6c2050687ed14d1e740fbb677
Parents: 24a50e1
Author: Igor Seliverstov <gv...@gmail.com>
Authored: Fri Dec 7 12:21:01 2018 +0300
Committer: Igor Seliverstov <gv...@gmail.com>
Committed: Fri Dec 7 12:21:01 2018 +0300
----------------------------------------------------------------------
.../communication/GridIoMessageFactory.java | 10 +-
.../processors/cache/GridCacheAdapter.java | 10 +-
.../processors/cache/GridCacheEntryEx.java | 13 +-
.../processors/cache/GridCacheMapEntry.java | 101 +++--------
.../cache/GridCacheUpdateTxResult.java | 26 ---
.../GridDistributedTxRemoteAdapter.java | 9 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 3 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 18 --
.../distributed/dht/GridDhtTxPrepareFuture.java | 6 +-
.../dht/GridPartitionedGetFuture.java | 6 +-
.../dht/GridPartitionedSingleGetFuture.java | 6 +-
.../dht/atomic/GridDhtAtomicCache.java | 13 +-
.../dht/colocated/GridDhtColocatedCache.java | 8 +-
.../near/AckCoordinatorOnRollback.java | 6 +-
.../distributed/near/GridNearGetFuture.java | 12 +-
.../near/GridNearTxFinishAndAckFuture.java | 7 +-
.../near/GridNearTxFinishFuture.java | 24 +--
.../cache/distributed/near/GridNearTxLocal.java | 50 +-----
.../local/atomic/GridLocalAtomicCache.java | 13 +-
.../processors/cache/mvcc/MvccProcessor.java | 22 ---
.../cache/mvcc/MvccProcessorImpl.java | 180 +------------------
.../processors/cache/mvcc/MvccQueryTracker.java | 9 -
.../cache/mvcc/MvccQueryTrackerImpl.java | 78 ++++----
.../processors/cache/mvcc/MvccUtils.java | 2 +-
.../cache/mvcc/StaticMvccQueryTracker.java | 5 -
.../cache/mvcc/msg/MvccWaitTxsRequest.java | 159 ----------------
.../cache/transactions/IgniteTxAdapter.java | 7 +-
.../cache/transactions/IgniteTxHandler.java | 3 +-
.../transactions/IgniteTxLocalAdapter.java | 51 +-----
.../processors/cache/GridCacheTestEntryEx.java | 13 +-
.../cache/mvcc/CacheMvccAbstractTest.java | 3 +-
31 files changed, 135 insertions(+), 738 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 3f4eb18..7d8f314 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -34,6 +34,8 @@ import org.apache.ignite.internal.managers.checkpoint.GridCheckpointRequest;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.managers.deployment.GridDeploymentRequest;
import org.apache.ignite.internal.managers.deployment.GridDeploymentResponse;
+import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
+import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyResponse;
import org.apache.ignite.internal.managers.eventstorage.GridEventStorageMessage;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.authentication.UserAuthenticateRequestMessage;
@@ -46,8 +48,6 @@ import org.apache.ignite.internal.processors.cache.CacheEvictionEntry;
import org.apache.ignite.internal.processors.cache.CacheInvokeDirectResult;
import org.apache.ignite.internal.processors.cache.CacheObjectByteArrayImpl;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
-import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
-import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyResponse;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheMvccEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
@@ -137,7 +137,6 @@ import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccQuerySnapshotReq
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccRecoveryFinishedMessage;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccSnapshotResponse;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccTxSnapshotRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccWaitTxsRequest;
import org.apache.ignite.internal.processors.cache.mvcc.msg.PartitionCountersNeighborcastRequest;
import org.apache.ignite.internal.processors.cache.mvcc.msg.PartitionCountersNeighborcastResponse;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
@@ -989,11 +988,6 @@ public class GridIoMessageFactory implements MessageFactory {
break;
- case 142:
- msg = new MvccWaitTxsRequest();
-
- break;
-
case 143:
msg = new GridCacheMvccEntryInfo();
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 465f1f1..1cd94d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -87,9 +87,9 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.affinity.GridCacheAffinityImpl;
import org.apache.ignite.internal.processors.cache.distributed.IgniteExternalizableExpiryPolicy;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
@@ -2005,7 +2005,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean storeEnabled = !skipVals && readThrough && ctx.readThrough();
- boolean readNoEntry = ctx.readNoEntry(expiry, readerArgs != null);
+ boolean readNoEntry = ctx.mvccEnabled() || ctx.readNoEntry(expiry, readerArgs != null);
for (KeyCacheObject key : keys) {
while (true) {
@@ -2086,7 +2086,6 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
expiry,
!deserializeBinary,
- mvccSnapshot,
readerArgs);
assert res != null;
@@ -2111,7 +2110,6 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
expiry,
!deserializeBinary,
- mvccSnapshot,
readerArgs);
if (res == null)
@@ -4996,8 +4994,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*transformClo*/null,
/*taskName*/null,
/*expiryPlc*/null,
- !deserializeBinary,
- null); // TODO IGNITE-7371
+ !deserializeBinary
+ );
if (val == null)
return null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
index 319c134..62360d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
@@ -272,7 +272,7 @@ public interface GridCacheEntryEx {
* @throws IgniteCheckedException If loading value failed.
* @throws GridCacheEntryRemovedException If entry was removed.
*/
- @Nullable public CacheObject innerGet(@Nullable GridCacheVersion ver,
+ public CacheObject innerGet(@Nullable GridCacheVersion ver,
@Nullable IgniteInternalTx tx,
boolean readThrough,
boolean updateMetrics,
@@ -281,8 +281,7 @@ public interface GridCacheEntryEx {
Object transformClo,
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
- boolean keepBinary,
- @Nullable MvccSnapshot mvccVer)
+ boolean keepBinary)
throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
@@ -310,7 +309,6 @@ public interface GridCacheEntryEx {
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- @Nullable MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs)
throws IgniteCheckedException, GridCacheEntryRemovedException;
@@ -332,7 +330,6 @@ public interface GridCacheEntryEx {
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- @Nullable MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs) throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
@@ -467,8 +464,7 @@ public interface GridCacheEntryEx {
@Nullable UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- @Nullable MvccSnapshot mvccVer
+ @Nullable Long updateCntr
) throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
@@ -510,8 +506,7 @@ public interface GridCacheEntryEx {
@Nullable UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- MvccSnapshot mvccVer
+ @Nullable Long updateCntr
) throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index bbdff35..da3f61a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -80,7 +80,6 @@ import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheFilter;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.IgniteTree;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridClosureException;
@@ -622,7 +621,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
/** {@inheritDoc} */
- @Nullable @Override public final CacheObject innerGet(
+ @Override public final CacheObject innerGet(
@Nullable GridCacheVersion ver,
@Nullable IgniteInternalTx tx,
boolean readThrough,
@@ -632,8 +631,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
Object transformClo,
String taskName,
@Nullable IgniteCacheExpiryPolicy expirePlc,
- boolean keepBinary,
- MvccSnapshot mvccVer)
+ boolean keepBinary)
throws IgniteCheckedException, GridCacheEntryRemovedException {
return (CacheObject)innerGet0(
ver,
@@ -648,7 +646,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
false,
keepBinary,
false,
- mvccVer,
null);
}
@@ -659,7 +656,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs) throws IgniteCheckedException, GridCacheEntryRemovedException {
return (EntryGetResult)innerGet0(
/*ver*/null,
@@ -674,7 +670,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
keepBinary,
/*reserve*/true,
- mvccVer,
readerArgs);
}
@@ -689,7 +684,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
String taskName,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean keepBinary,
- MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs)
throws IgniteCheckedException, GridCacheEntryRemovedException {
return (EntryGetResult)innerGet0(
@@ -705,7 +699,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
keepBinary,
false,
- mvccVer,
readerArgs);
}
@@ -724,7 +717,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
boolean retVer,
boolean keepBinary,
boolean reserveForLoad,
- @Nullable MvccSnapshot mvccVer,
@Nullable ReaderArguments readerArgs
) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert !(retVer && readThrough);
@@ -750,51 +742,39 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
CacheObject val;
- if (mvccVer != null) {
- CacheDataRow row = cctx.offheap().mvccRead(cctx, key, mvccVer);
-
- if (row != null) {
- val = row.value();
- resVer = row.version();
- }
- else
- val = null;
- }
- else {
- boolean valid = valid(tx != null ? tx.topologyVersion() : cctx.affinity().affinityTopologyVersion());
+ boolean valid = valid(tx != null ? tx.topologyVersion() : cctx.affinity().affinityTopologyVersion());
- if (valid) {
- val = this.val;
+ if (valid) {
+ val = this.val;
- if (val == null) {
- if (isStartVersion()) {
- unswap(null, false);
+ if (val == null) {
+ if (isStartVersion()) {
+ unswap(null, false);
- val = this.val;
- }
+ val = this.val;
}
+ }
- if (val != null) {
- long expireTime = expireTimeExtras();
+ if (val != null) {
+ long expireTime = expireTimeExtras();
- if (expireTime > 0 && (expireTime < U.currentTimeMillis())) {
- if (onExpired((CacheObject)cctx.unwrapTemporary(val), null)) {
- val = null;
- evt = false;
+ if (expireTime > 0 && (expireTime < U.currentTimeMillis())) {
+ if (onExpired((CacheObject)cctx.unwrapTemporary(val), null)) {
+ val = null;
+ evt = false;
- if (cctx.deferredDelete()) {
- deferred = true;
- ver0 = ver;
- }
- else
- obsolete = true;
+ if (cctx.deferredDelete()) {
+ deferred = true;
+ ver0 = ver;
}
+ else
+ obsolete = true;
}
}
}
- else
- val = null;
}
+ else
+ val = null;
CacheObject ret = val;
@@ -1460,8 +1440,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- @Nullable MvccSnapshot mvccVer
+ @Nullable Long updateCntr
) throws IgniteCheckedException, GridCacheEntryRemovedException {
CacheObject old;
@@ -1483,8 +1462,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
ensureFreeSpace();
- GridLongList mvccWaitTxs = null;
-
lockListenerReadLock();
lockEntry();
@@ -1565,18 +1542,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert val != null;
- if (cctx.mvccEnabled()) {
- assert mvccVer != null;
-
- mvccWaitTxs = cctx.offheap().mvccUpdateNative(tx.local(),
- this,
- val,
- newVer,
- expireTime,
- mvccVer);
- }
- else
- storeValue(val, expireTime, newVer);
+ storeValue(val, expireTime, newVer);
if (cctx.deferredDelete() && deletedUnlocked() && !isInternal() && !detached())
deletedUnlocked(false);
@@ -1655,7 +1621,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (intercept)
cctx.config().getInterceptor().onAfterPut(new CacheLazyEntry(cctx, key, key0, val, val0, keepBinary, updateCntr0));
- return valid ? new GridCacheUpdateTxResult(true, updateCntr0, logPtr, mvccWaitTxs) :
+ return valid ? new GridCacheUpdateTxResult(true, updateCntr0, logPtr) :
new GridCacheUpdateTxResult(false, logPtr);
}
@@ -1685,8 +1651,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable UUID subjId,
String taskName,
@Nullable GridCacheVersion dhtVer,
- @Nullable Long updateCntr,
- @Nullable MvccSnapshot mvccVer
+ @Nullable Long updateCntr
) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert cctx.transactional();
@@ -1716,8 +1681,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
boolean marked = false;
- GridLongList mvccWaitTxs = null;
-
lockListenerReadLock();
lockEntry();
@@ -1762,13 +1725,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- if (cctx.mvccEnabled()) {
- assert mvccVer != null;
-
- mvccWaitTxs = cctx.offheap().mvccRemoveNative(tx.local(), this, mvccVer);
- }
- else
- removeValue();
+ removeValue();
update(null, 0, 0, newVer, true);
@@ -1889,7 +1846,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
cctx.config().getInterceptor().onAfterRemove(entry0);
if (valid)
- return new GridCacheUpdateTxResult(true, updateCntr0, logPtr, mvccWaitTxs);
+ return new GridCacheUpdateTxResult(true, updateCntr0, logPtr);
else
return new GridCacheUpdateTxResult(false, logPtr);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateTxResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateTxResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateTxResult.java
index 8a68100..7df27d6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateTxResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateTxResult.java
@@ -21,7 +21,6 @@ import java.util.List;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.jetbrains.annotations.Nullable;
@@ -37,9 +36,6 @@ public class GridCacheUpdateTxResult {
private long updateCntr;
/** */
- private GridLongList mvccWaitTxs;
-
- /** */
private GridFutureAdapter<GridCacheUpdateTxResult> fut;
/** */
@@ -108,21 +104,6 @@ public class GridCacheUpdateTxResult {
}
/**
- * Constructor.
- *
- * @param success Success flag.
- * @param updateCntr Update counter.
- * @param logPtr Logger WAL pointer for the update.
- * @param mvccWaitTxs List of transactions to wait for completion.
- */
- GridCacheUpdateTxResult(boolean success, long updateCntr, WALPointer logPtr, GridLongList mvccWaitTxs) {
- this.success = success;
- this.updateCntr = updateCntr;
- this.logPtr = logPtr;
- this.mvccWaitTxs = mvccWaitTxs;
- }
-
- /**
* @return Partition update counter.
*/
public long updateCounter() {
@@ -151,13 +132,6 @@ public class GridCacheUpdateTxResult {
}
/**
- * @return List of transactions to wait for completion.
- */
- @Nullable public GridLongList mvccWaitTransactions() {
- return mvccWaitTxs;
- }
-
- /**
* @return Mvcc history rows.
*/
@Nullable public List<MvccLinkAwareSearchRow> mvccHistory() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index a915478..94ab814 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -628,8 +628,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter(),
- mvccSnapshot());
+ txEntry.updateCounter());
else {
assert val != null : txEntry;
@@ -653,8 +652,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter(),
- mvccSnapshot());
+ txEntry.updateCounter());
txEntry.updateCounter(updRes.updateCounter());
@@ -691,8 +689,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter(),
- mvccSnapshot());
+ txEntry.updateCounter());
txEntry.updateCounter(updRes.updateCounter());
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index dcd7be5..cb66e74 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -1437,8 +1437,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
null,
tx != null ? tx.resolveTaskName() : null,
null,
- req.keepBinary(),
- null); // TODO IGNITE-7371
+ req.keepBinary());
}
assert e.lockedBy(mappedVer) || ctx.mvcc().isRemoved(e.context(), mappedVer) :
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index a9c133a..a5cbbdf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -35,13 +35,11 @@ import org.apache.ignite.internal.processors.cache.GridCacheCompoundIdentityFutu
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.mvcc.MvccFuture;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
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;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -299,22 +297,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
// No backup or near nodes to send commit message to (just complete then).
sync = false;
- GridLongList waitTxs = tx.mvccWaitTransactions();
-
- if (waitTxs != null) {
- MvccSnapshot snapshot = tx.mvccSnapshot();
-
- assert snapshot != null;
-
- MvccCoordinator crd = cctx.coordinators().currentCoordinator();
-
- if (crd != null && crd.coordinatorVersion() == snapshot.coordinatorVersion()) {
- add((IgniteInternalFuture)cctx.coordinators().waitTxsFuture(crd.nodeId(), waitTxs));
-
- sync = true;
- }
- }
-
markInitialized();
if (!sync)
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/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 58edc9d..1933efa 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
@@ -410,8 +410,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
entryProc,
tx.resolveTaskName(),
null,
- keepBinary,
- null); // TODO IGNITE-7371
+ keepBinary);
if (retVal || txEntry.op() == TRANSFORM) {
if (!F.isEmpty(txEntry.entryProcessors())) {
@@ -524,8 +523,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
/*transformClo*/null,
/*taskName*/null,
/*expiryPlc*/null,
- /*keepBinary*/true,
- null); // TODO IGNITE-7371
+ /*keepBinary*/true);
}
if (oldVal != null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 0629754..d38903d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -470,7 +470,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
GridDhtCacheAdapter<K, V> cache = cache();
- boolean readNoEntry = cctx.readNoEntry(expiryPlc, false);
+ boolean readNoEntry = cctx.mvccEnabled() || cctx.readNoEntry(expiryPlc, false);
boolean evt = !skipVals;
while (true) {
@@ -528,7 +528,6 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
taskName,
expiryPlc,
!deserializeBinary,
- mvccSnapshot(),
null);
if (getRes != null) {
@@ -547,8 +546,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
null,
taskName,
expiryPlc,
- !deserializeBinary,
- mvccSnapshot());
+ !deserializeBinary);
}
entry.touch(topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 0d69485..ab7e911 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -429,7 +429,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
GridDhtCacheAdapter colocated = cctx.dht();
- boolean readNoEntry = cctx.readNoEntry(expiryPlc, false);
+ boolean readNoEntry = cctx.mvccEnabled() || cctx.readNoEntry(expiryPlc, false);
boolean evt = !skipVals;
while (true) {
@@ -486,7 +486,6 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
taskName,
expiryPlc,
true,
- mvccSnapshot,
null);
if (res != null) {
@@ -505,8 +504,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
null,
taskName,
expiryPlc,
- true,
- mvccSnapshot);
+ true);
}
entry.touch(topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 6118fbb..516b9cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1534,7 +1534,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
expiry,
true,
- null,
null);
if (getRes != null) {
@@ -1553,8 +1552,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
taskName,
expiry,
- !deserializeBinary,
- null);
+ !deserializeBinary);
}
// Entry was not in memory or in swap, so we remove it from cache.
@@ -2125,8 +2123,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
entryProcessor,
taskName,
null,
- req.keepBinary(),
- null);
+ req.keepBinary());
Object oldVal = null;
Object updatedVal = null;
@@ -2306,8 +2303,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
taskName,
null,
- req.keepBinary(),
- null);
+ req.keepBinary());
Object val = ctx.config().getInterceptor().onBeforePut(
new CacheLazyEntry(
@@ -2352,8 +2348,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
taskName,
null,
- req.keepBinary(),
- null);
+ req.keepBinary());
IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor()
.onBeforeRemove(new CacheLazyEntry(ctx, entry.key(), old, req.keepBinary()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index 8fc353c..7546144 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -48,11 +48,11 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedUn
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtEmbeddedFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtFinishedFuture;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTransactionalCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
@@ -524,7 +524,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
Map<K, V> locVals = null;
boolean success = true;
- boolean readNoEntry = ctx.readNoEntry(expiryPlc, false);
+ boolean readNoEntry = ctx.mvccEnabled() || ctx.readNoEntry(expiryPlc, false);
boolean evt = !skipVals;
for (KeyCacheObject key : keys) {
@@ -595,7 +595,6 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
taskName,
expiryPlc,
!deserializeBinary,
- mvccSnapshot,
null);
if (getRes != null) {
@@ -614,8 +613,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
null,
taskName,
expiryPlc,
- !deserializeBinary,
- mvccSnapshot);
+ !deserializeBinary);
}
// Entry was not in memory or in swap, so we remove it from cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
index 1648da9..07ee5d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/AckCoordinatorOnRollback.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache.distributed.near;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.util.typedef.CIX1;
@@ -43,12 +42,9 @@ public class AckCoordinatorOnRollback extends CIX1<IgniteInternalFuture<IgniteIn
@Override public void applyx(IgniteInternalFuture<IgniteInternalTx> fut) throws IgniteCheckedException {
assert fut.isDone();
- MvccQueryTracker tracker = tx.mvccQueryTracker();
MvccSnapshot mvccSnapshot = tx.mvccSnapshot();
- if (tracker != null) // Optimistic tx.
- tracker.onDone(tx, false);
- else if (mvccSnapshot != null)// Pessimistic tx.
+ if (mvccSnapshot != null)// Pessimistic tx.
tx.context().coordinators().ackTxRollback(mvccSnapshot);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 09445f7..adacc81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -361,8 +361,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- null,
- null); // TODO IGNITE-7371
+ null);
if (res != null) {
v = res.value();
@@ -380,8 +379,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
null,
taskName,
expiryPlc,
- !deserializeBinary,
- null); // TODO IGNITE-7371
+ !deserializeBinary);
}
}
@@ -502,8 +500,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
taskName,
expiryPlc,
!deserializeBinary,
- null,
- null); // TODO IGNITE-7371
+ null);
if (res != null) {
v = res.value();
@@ -521,8 +518,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
null,
taskName,
expiryPlc,
- !deserializeBinary,
- null); // TODO IGNITE-7371
+ !deserializeBinary);
}
// Entry was not in memory or in swap, so we remove it from cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishAndAckFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishAndAckFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishAndAckFuture.java
index a3a5cdb..22e9aee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishAndAckFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishAndAckFuture.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache.distributed.near;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -62,13 +61,9 @@ public class GridNearTxFinishAndAckFuture extends GridFutureAdapter<IgniteIntern
IgniteInternalFuture<Void> ackFut = null;
- MvccQueryTracker tracker = tx.mvccQueryTracker();
-
MvccSnapshot mvccSnapshot = tx.mvccSnapshot();
- if (tracker != null)
- ackFut = tracker.onDone(tx, commit);
- else if (mvccSnapshot != null) {
+ if (mvccSnapshot != null) {
if (commit)
ackFut = tx.context().coordinators().ackTxCommit(mvccSnapshot);
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index 20d9e50..8484503 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -40,13 +40,10 @@ import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishResponse;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.mvcc.MvccFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.C1;
@@ -442,23 +439,6 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
private void doFinish(boolean commit, boolean clearThreadMap) {
try {
if (tx.localFinish(commit, clearThreadMap) || (!commit && tx.state() == UNKNOWN)) {
- GridLongList waitTxs = tx.mvccWaitTransactions();
-
- if (waitTxs != null) {
- MvccSnapshot snapshot = tx.mvccSnapshot();
-
- MvccCoordinator crd = cctx.coordinators().currentCoordinator();
-
- assert snapshot != null;
-
- if (snapshot.coordinatorVersion() == crd.coordinatorVersion()) {
- IgniteInternalFuture fut = cctx.coordinators()
- .waitTxsFuture(cctx.coordinators().currentCoordinatorId(), waitTxs);
-
- add(fut);
- }
- }
-
// Cleanup transaction if heuristic failure.
if (tx.state() == UNKNOWN)
cctx.tm().rollbackTx(tx, clearThreadMap, false);
@@ -468,13 +448,13 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
GridDistributedTxMapping mapping = mappings.singleMapping();
if (mapping != null) {
- assert !hasFutures() || isDone() || waitTxs != null : futures();
+ assert !hasFutures() || isDone() : futures();
finish(1, mapping, commit, !clearThreadMap);
}
}
else {
- assert !hasFutures() || isDone() || waitTxs != null : futures();
+ assert !hasFutures() || isDone() : futures();
finish(mappings.mappings(), commit, !clearThreadMap);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 9f1f86d..01c4970 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -62,8 +62,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrep
import org.apache.ignite.internal.processors.cache.distributed.dht.GridInvokeValue;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtDetachedCacheEntry;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
@@ -190,9 +188,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
/** Tx label. */
@Nullable private String lb;
- /** */
- private MvccQueryTracker mvccTracker;
-
/** Whether this is Mvcc transaction or not.<p>
* {@code null} means there haven't been any calls made on this transaction, and first operation will give this
* field actual value.
@@ -266,13 +261,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
trackTimeout = timeout() > 0 && !implicit() && cctx.time().addTimeoutObject(this);
}
- /**
- * @return Mvcc query version tracker.
- */
- public MvccQueryTracker mvccQueryTracker() {
- return mvccTracker;
- }
-
/** {@inheritDoc} */
@Override public boolean near() {
return true;
@@ -1396,7 +1384,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
resolveTaskName(),
null,
keepBinary,
- null, // TODO IGNITE-7371
null) : null;
if (res != null) {
@@ -1415,8 +1402,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
entryProcessor,
resolveTaskName(),
null,
- keepBinary,
- null); // TODO IGNITE-7371
+ keepBinary);
}
}
catch (ClusterTopologyCheckedException e) {
@@ -1969,17 +1955,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
}
/**
- * @param cctx Cache context.
- * @return Mvcc snapshot for read inside tx (initialized once for OPTIMISTIC SERIALIZABLE and REPEATABLE_READ txs).
- */
- private MvccSnapshot mvccReadSnapshot(GridCacheContext cctx) {
- if (!cctx.mvccEnabled() || mvccTracker == null)
- return null;
-
- return mvccTracker.snapshot();
- }
-
- /**
* @param cacheCtx Cache context.
* @param cacheIds Involved cache ids.
* @param parts Partitions.
@@ -2292,7 +2267,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
resolveTaskName(),
null,
txEntry.keepBinary(),
- null, // TODO IGNITE-7371
null);
if (getRes != null) {
@@ -2311,8 +2285,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
transformClo,
resolveTaskName(),
null,
- txEntry.keepBinary(),
- null); // TODO IGNITE-7371
+ txEntry.keepBinary());
}
// If value is in cache and passed the filter.
@@ -2592,8 +2565,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
resolveTaskName(),
null,
txEntry.keepBinary(),
- null,
- null); // TODO IGNITE-7371
+ null);
if (getRes != null) {
val = getRes.value();
@@ -2611,8 +2583,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
transformClo,
resolveTaskName(),
null,
- txEntry.keepBinary(),
- null); // TODO IGNITE-7371
+ txEntry.keepBinary());
}
if (val != null) {
@@ -2680,7 +2651,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
resolveTaskName(),
accessPlc,
!deserializeBinary,
- mvccReadSnapshot(cacheCtx), // TODO IGNITE-7371
null) : null;
if (getRes != null) {
@@ -2699,8 +2669,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
null,
resolveTaskName(),
accessPlc,
- !deserializeBinary,
- mvccReadSnapshot(cacheCtx)); // TODO IGNITE-7371
+ !deserializeBinary);
}
if (val != null) {
@@ -3039,7 +3008,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
needVer,
/*keepCacheObject*/true,
recovery,
- mvccReadSnapshot(cacheCtx),
+ null,
label()
).chain(new C1<IgniteInternalFuture<Object>, Void>() {
@Override public Void apply(IgniteInternalFuture<Object> f) {
@@ -3073,7 +3042,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
needVer,
/*keepCacheObject*/true,
label(),
- mvccReadSnapshot(cacheCtx)
+ null
).chain(new C1<IgniteInternalFuture<Map<Object, Object>>, Void>() {
@Override public Void apply(IgniteInternalFuture<Map<Object, Object>> f) {
try {
@@ -3170,8 +3139,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
resolveTaskName(),
expiryPlc0,
txEntry == null ? keepBinary : txEntry.keepBinary(),
- null,
- null); // TODO IGNITE-7371
+ null);
if (res == null) {
if (misses == null)
@@ -3883,7 +3851,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
NearTxFinishFuture fut = fast ? new GridNearTxFastFinishFuture(this, commit) :
new GridNearTxFinishFuture<>(cctx, this, commit);
- if (mvccQueryTracker() != null || mvccSnapshot != null || txState.mvccEnabled()) {
+ if (mvccSnapshot != null || txState.mvccEnabled()) {
if (commit)
fut = new GridNearTxFinishAndAckFuture(fut);
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 919f6d2..617a24d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -462,7 +462,6 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
taskName,
expiry,
!deserializeBinary,
- null,
null);
if (res != null) {
@@ -490,8 +489,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
null,
taskName,
expiry,
- !deserializeBinary,
- null);
+ !deserializeBinary);
if (v != null) {
ctx.addResult(vals,
@@ -1099,8 +1097,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
entryProcessor,
taskName,
null,
- keepBinary,
- null);
+ keepBinary);
Object oldVal = null;
@@ -1242,8 +1239,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
null,
taskName,
null,
- keepBinary,
- null);
+ keepBinary);
Object interceptorVal = ctx.config().getInterceptor().onBeforePut(new CacheLazyEntry(
ctx, entry.key(), old, keepBinary), val);
@@ -1278,8 +1274,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
null,
taskName,
null,
- keepBinary,
- null);
+ keepBinary);
IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor()
.onBeforeRemove(new CacheLazyEntry(ctx, entry.key(), old, keepBinary));
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/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 fd45c7a..161342f 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
@@ -173,38 +173,16 @@ public interface MvccProcessor extends GridProcessor {
/**
* @param updateVer Transaction update version.
- * @param readSnapshot Transaction read version.
- * @param qryId Query tracker id.
- * @return Acknowledge future.
- */
- IgniteInternalFuture<Void> ackTxCommit(MvccVersion updateVer, MvccSnapshot readSnapshot, long qryId);
-
- /**
- * @param updateVer Transaction update version.
*/
void ackTxRollback(MvccVersion updateVer);
/**
- * @param updateVer Transaction update version.
- * @param readSnapshot Transaction read version.
- * @param qryTrackerId Query tracker id.
- */
- void ackTxRollback(MvccVersion updateVer, MvccSnapshot readSnapshot, long qryTrackerId);
-
- /**
* @param snapshot Query version.
* @param qryId Query tracker ID.
*/
void ackQueryDone(MvccSnapshot snapshot, long qryId);
/**
- * @param crdId Coordinator ID.
- * @param txs Transaction IDs.
- * @return Future.
- */
- IgniteInternalFuture<Void> waitTxsFuture(UUID crdId, GridLongList txs);
-
- /**
* @param log Logger.
* @param diagCtx Diagnostic request.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/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 7f386d8..d026607 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
@@ -65,8 +65,6 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLo
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestQueryCntr;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestQueryId;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTx;
-import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTxAndQueryCntr;
-import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccAckRequestTxAndQueryId;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccActiveQueriesMessage;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccFutureResponse;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccMessage;
@@ -74,7 +72,6 @@ import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccQuerySnapshotReq
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccRecoveryFinishedMessage;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccSnapshotResponse;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccTxSnapshotRequest;
-import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccWaitTxsRequest;
import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxKey;
import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxLog;
import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
@@ -87,7 +84,6 @@ import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccDataRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridLongList;
-import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridCompoundIdentityFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -134,7 +130,6 @@ import static org.apache.ignite.internal.processors.cache.persistence.CacheDataR
/**
* MVCC processor.
*/
-@SuppressWarnings("serial")
public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProcessor, DatabaseLifecycleListener {
/** */
private static final boolean FORCE_MVCC =
@@ -206,9 +201,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
private final Map<Long, WaitAckFuture> ackFuts = new ConcurrentHashMap<>();
/** */
- private final Map<Long, GridFutureAdapter> waitTxFuts = new ConcurrentHashMap<>();
-
- /** */
private final Map<TxKey, Waiter> waitMap = new ConcurrentHashMap<>();
/** */
@@ -271,6 +263,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
@Override public void preProcessCacheConfiguration(CacheConfiguration ccfg) {
if (FORCE_MVCC && ccfg.getAtomicityMode() == TRANSACTIONAL && !CU.isSystemCache(ccfg.getName())) {
ccfg.setAtomicityMode(TRANSACTIONAL_SNAPSHOT);
+ //noinspection unchecked
ccfg.setNearConfiguration(null);
}
@@ -371,6 +364,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
private void txLogPageStoreInit(IgniteCacheDatabaseSharedManager mgr) throws IgniteCheckedException {
assert CU.isPersistenceEnabled(ctx.config());
+ //noinspection ConstantConditions
ctx.cache().context().pageStore().initialize(TX_LOG_CACHE_ID, 1,
TX_LOG_CACHE_NAME, mgr.dataRegion(TX_LOG_CACHE_NAME).memoryMetrics());
}
@@ -650,9 +644,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
@Override public void addQueryTracker(MvccQueryTracker tracker) {
assert tracker.id() != MVCC_TRACKER_ID_NA;
- MvccQueryTracker tr = activeTrackers.put(tracker.id(), tracker);
-
- assert tr == null;
+ activeTrackers.putIfAbsent(tracker.id(), tracker);
}
/** {@inheritDoc} */
@@ -763,20 +755,12 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** {@inheritDoc} */
@Override public IgniteInternalFuture<Void> ackTxCommit(MvccSnapshot updateVer) {
- return ackTxCommit(updateVer, null, 0L);
- }
-
- /** {@inheritDoc} */
- @Override public IgniteInternalFuture<Void> ackTxCommit(MvccVersion updateVer, MvccSnapshot readSnapshot,
- long qryId) {
assert updateVer != null;
MvccCoordinator crd = curCrd;
if (updateVer.coordinatorVersion() == crd.coordinatorVersion())
- return sendTxCommit(crd, createTxAckMessage(futIdCntr.incrementAndGet(), updateVer, readSnapshot, qryId));
- else if (readSnapshot != null)
- ackQueryDone(readSnapshot, qryId);
+ return sendTxCommit(crd, new MvccAckRequestTx(futIdCntr.incrementAndGet(), updateVer.counter()));
return new GridFinishedFuture<>();
}
@@ -790,36 +774,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
if (crd.coordinatorVersion() != updateVer.coordinatorVersion())
return;
- MvccAckRequestTx msg = createTxAckMessage(-1, updateVer, null, 0L);
-
- msg.skipResponse(true);
-
- try {
- sendMessage(crd.nodeId(), msg);
- }
- catch (ClusterTopologyCheckedException e) {
- if (log.isDebugEnabled())
- log.debug("Failed to send tx rollback ack, node left [msg=" + msg + ", node=" + crd.nodeId() + ']');
- }
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to send tx rollback ack [msg=" + msg + ", node=" + crd.nodeId() + ']', e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void ackTxRollback(MvccVersion updateVer, MvccSnapshot readSnapshot, long qryTrackerId) {
- assert updateVer != null;
-
- MvccCoordinator crd = curCrd;
-
- if (crd.coordinatorVersion() != updateVer.coordinatorVersion()) {
- if (readSnapshot != null)
- ackQueryDone(readSnapshot, qryTrackerId);
-
- return;
- }
-
- MvccAckRequestTx msg = createTxAckMessage(-1, updateVer, readSnapshot, qryTrackerId);
+ MvccAckRequestTx msg = new MvccAckRequestTx((long)-1, updateVer.counter());
msg.skipResponse(true);
@@ -837,11 +792,10 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** {@inheritDoc} */
@Override public void ackQueryDone(MvccSnapshot snapshot, long qryId) {
- assert snapshot != null;
-
MvccCoordinator crd = currentCoordinator();
- if (crd == null || crd.coordinatorVersion() == snapshot.coordinatorVersion()
+ if (crd == null || snapshot != null
+ && crd.coordinatorVersion() == snapshot.coordinatorVersion()
&& sendQueryDone(crd, new MvccAckRequestQueryCntr(queryTrackCounter(snapshot))))
return;
@@ -854,30 +808,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/** {@inheritDoc} */
- @Override public IgniteInternalFuture<Void> waitTxsFuture(UUID crdId, GridLongList txs) {
- assert crdId != null;
- assert txs != null && !txs.isEmpty();
-
- WaitAckFuture fut = new WaitAckFuture(futIdCntr.incrementAndGet(), crdId, false);
-
- ackFuts.put(fut.id, fut);
-
- try {
- sendMessage(crdId, new MvccWaitTxsRequest(fut.id, txs));
- }
- catch (IgniteCheckedException e) {
- if (ackFuts.remove(fut.id) != null) {
- if (e instanceof ClusterTopologyCheckedException)
- fut.onDone(); // No need to wait, new coordinator will be assigned, finish without error.
- else
- fut.onDone(e);
- }
- }
-
- return fut;
- }
-
- /** {@inheritDoc} */
// TODO: Proper use of diagnostic context.
@Override public void dumpDebugInfo(IgniteLogger log, @Nullable IgniteDiagnosticPrepareContext diagCtx) {
boolean first = true;
@@ -1077,19 +1007,12 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
private void onTxDone(Long txCntr, boolean increaseCommittedCntr) {
assert initFut.isDone();
- GridFutureAdapter fut;
-
synchronized (this) {
activeTxs.remove(txCntr);
if (increaseCommittedCntr)
committedCntr.setIfGreater(txCntr);
}
-
- fut = waitTxFuts.remove(txCntr);
-
- if (fut != null)
- fut.onDone();
}
/**
@@ -1100,23 +1023,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
/**
- * @param futId Future ID.
- * @param updateVer Update version.
- * @param readSnapshot Optional read version.
- * @param qryTrackerId Query tracker id.
- * @return Message.
- */
- private MvccAckRequestTx createTxAckMessage(long futId, MvccVersion updateVer, MvccSnapshot readSnapshot,
- long qryTrackerId) {
- if (readSnapshot == null)
- return new MvccAckRequestTx(futId, updateVer.counter());
- else if (readSnapshot.coordinatorVersion() == updateVer.coordinatorVersion())
- return new MvccAckRequestTxAndQueryCntr(futId, updateVer.counter(), queryTrackCounter(readSnapshot));
- else
- return new MvccAckRequestTxAndQueryId(futId, updateVer.counter(), qryTrackerId);
- }
-
- /**
* @param mvccVer Read version.
* @return Tracker counter.
*/
@@ -1384,23 +1290,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
return new NodeStoppingException("Operation has been cancelled (node is stopping).");
}
- /**
- * @param nodeId Node ID.
- * @param msg Message.
- */
- private void sendFutureResponse(UUID nodeId, MvccWaitTxsRequest msg) {
- try {
- sendMessage(nodeId, new MvccFutureResponse(msg.futureId()));
- }
- catch (ClusterTopologyCheckedException e) {
- if (log.isDebugEnabled())
- log.debug("Failed to send tx ack response, node left [msg=" + msg + ", node=" + nodeId + ']');
- }
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to send tx ack response [msg=" + msg + ", node=" + nodeId + ']', e);
- }
- }
-
/** */
@NotNull private IgniteInternalFuture<Void> sendTxCommit(MvccCoordinator crd, MvccAckRequestTx msg) {
WaitAckFuture fut = new WaitAckFuture(msg.futureId(), crd.nodeId(), true);
@@ -1613,59 +1502,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
* @param nodeId Node ID.
* @param msg Message.
*/
- @SuppressWarnings("unchecked")
- private void processCoordinatorWaitTxsRequest(final UUID nodeId, final MvccWaitTxsRequest msg) {
- GridLongList txs = msg.transactions();
-
- GridCompoundFuture resFut = null;
-
- for (int i = 0; i < txs.size(); i++) {
- Long txId = txs.get(i);
-
- GridFutureAdapter fut = waitTxFuts.get(txId);
-
- if (fut == null) {
- GridFutureAdapter old = waitTxFuts.putIfAbsent(txId, fut = new GridFutureAdapter());
-
- if (old != null)
- fut = old;
- }
-
- boolean isDone;
-
- synchronized (this) {
- isDone = !activeTxs.containsKey(txId);
- }
-
- if (isDone)
- fut.onDone();
-
- if (!fut.isDone()) {
- if (resFut == null)
- resFut = new GridCompoundFuture();
-
- resFut.add(fut);
- }
- }
-
- if (resFut != null)
- resFut.markInitialized();
-
- if (resFut == null || resFut.isDone())
- sendFutureResponse(nodeId, msg);
- else {
- resFut.listen(new IgniteInClosure<IgniteInternalFuture>() {
- @Override public void apply(IgniteInternalFuture fut) {
- sendFutureResponse(nodeId, msg);
- }
- });
- }
- }
-
- /**
- * @param nodeId Node ID.
- * @param msg Message.
- */
private void processCoordinatorActiveQueriesMessage(UUID nodeId, MvccActiveQueriesMessage msg) {
prevCrdQueries.addNodeActiveQueries(nodeId, msg.activeQueries());
}
@@ -1856,8 +1692,6 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
processCoordinatorQuerySnapshotRequest(nodeId, (MvccQuerySnapshotRequest)msg);
else if (msg instanceof MvccSnapshotResponse)
processCoordinatorSnapshotResponse(nodeId, (MvccSnapshotResponse)msg);
- else if (msg instanceof MvccWaitTxsRequest)
- processCoordinatorWaitTxsRequest(nodeId, (MvccWaitTxsRequest)msg);
else if (msg instanceof MvccAckRequestQueryId)
processNewCoordinatorQueryAckRequest(nodeId, (MvccAckRequestQueryId)msg);
else if (msg instanceof MvccActiveQueriesMessage)
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTracker.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTracker.java
index f143a43..d2e3b3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTracker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccQueryTracker.java
@@ -84,15 +84,6 @@ public interface MvccQueryTracker {
public void onDone();
/**
- * Marks tracker as done.
- *
- * @param tx Transaction.
- * @param commit Commit flag.
- * @return Acknowledge future.
- */
- @Nullable public IgniteInternalFuture<Void> onDone(@NotNull GridNearTxLocal tx, boolean commit);
-
- /**
* Mvcc coordinator change callback.
*
* @param newCrd New mvcc coordinator.
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/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 d86f5ec..d93a2e9 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
@@ -24,7 +24,6 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -139,40 +138,8 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
/** {@inheritDoc} */
@Override public void onDone() {
- if (!checkDone())
- return;
-
- MvccProcessor prc = cctx.shared().coordinators();
-
- MvccSnapshot snapshot = snapshot();
-
- if (snapshot != null) {
- prc.removeQueryTracker(id);
-
- prc.ackQueryDone(snapshot, id);
- }
- }
-
- /** {@inheritDoc} */
- @Override public IgniteInternalFuture<Void> onDone(@NotNull GridNearTxLocal tx, boolean commit) {
- MvccSnapshot snapshot = snapshot(), txSnapshot = tx.mvccSnapshot();
-
- if (!checkDone() || snapshot == null && txSnapshot == null)
- return commit ? new GridFinishedFuture<>() : null;
-
- MvccProcessor prc = cctx.shared().coordinators();
-
- if (snapshot != null)
- prc.removeQueryTracker(id);
-
- if (txSnapshot == null)
- prc.ackQueryDone(snapshot, id);
- else if (commit)
- return prc.ackTxCommit(txSnapshot, snapshot, id);
- else
- prc.ackTxRollback(txSnapshot, snapshot, id);
-
- return null;
+ if (checkDone())
+ ackQueryDone(snapshot());
}
/** {@inheritDoc} */
@@ -195,8 +162,21 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
}
/** */
+ private void ackQueryDone(MvccSnapshot snapshot) {
+ MvccProcessor prc = cctx.shared().coordinators();
+
+ if (snapshot != null) {
+ prc.removeQueryTracker(id);
+
+ prc.ackQueryDone(snapshot, id);
+ }
+ }
+
+ /** */
private void requestSnapshot0(AffinityTopologyVersion topVer, MvccSnapshotResponseListener lsnr) {
if (checkTopology(topVer, lsnr = decorate(lsnr))) {
+ cctx.shared().coordinators().addQueryTracker(this);
+
try {
MvccSnapshot snapshot = cctx.shared().coordinators().tryRequestSnapshotLocal();
@@ -209,6 +189,8 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
lsnr.onError(e);
}
}
+ else
+ cctx.shared().coordinators().removeQueryTracker(id);
}
/** */
@@ -238,6 +220,9 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
this.topVer = topVer;
synchronized (this) {
+ if (done)
+ return false;
+
crdVer = crd.coordinatorVersion();
}
@@ -280,27 +265,30 @@ public class MvccQueryTrackerImpl implements MvccQueryTracker {
* @return {@code false} if need to remap.
*/
private boolean onResponse0(@NotNull MvccSnapshot res, MvccSnapshotResponseListener lsnr) {
- boolean needRemap = false;
+ boolean ackQueryDone = false, needRemap = false;
synchronized (this) {
assert snapshot() == null : "[this=" + this + ", rcvdVer=" + res + "]";
- if (crdVer != 0) {
+ if (!done && crdVer != 0) {
this.snapshot = res;
+
+ return true;
}
- else
- needRemap = true;
- }
- if (needRemap) { // Coordinator failed or reassigned, need remap.
- tryRemap(lsnr);
- return false;
+ if (crdVer != 0)
+ ackQueryDone = true;
+ else if (!done)
+ needRemap = true;
}
- cctx.shared().coordinators().addQueryTracker(this);
+ if (needRemap)
+ tryRemap(lsnr); // Coordinator is failed or reassigned, need remap.
+ else if (ackQueryDone)
+ ackQueryDone(res); // Coordinator is not failed, but the tracker is already closed.
- return true;
+ return false;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/7301ada3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
index 0ceed09..43f87e3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUtils.java
@@ -809,7 +809,7 @@ public class MvccUtils {
if (tx == null)
tracker = new MvccQueryTrackerImpl(cctx);
- else if ((tracker = tx.mvccQueryTracker()) == null)
+ else
tracker = new StaticMvccQueryTracker(cctx, requestSnapshot(cctx, tx));
if (tracker.snapshot() == null)