You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/09/12 11:27:18 UTC
[9/9] ignite git commit: ignite-6149
ignite-6149
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6d170264
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6d170264
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6d170264
Branch: refs/heads/ignite-6149
Commit: 6d1702646fc0abbba690b1b4f7b7fde718d9f4d2
Parents: 774af44
Author: sboikov <sb...@gridgain.com>
Authored: Tue Sep 12 11:48:16 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Sep 12 14:14:28 2017 +0300
----------------------------------------------------------------------
.../cache/IgniteCacheOffheapManagerImpl.java | 10 +-
.../distributed/dht/GridDhtTxFinishFuture.java | 2 +
.../near/GridNearTxFinishFuture.java | 23 +-
.../mvcc/CacheCoordinatorsSharedManager.java | 110 +++++--
.../cache/mvcc/CoordinatorTxCounterRequest.java | 26 +-
.../cache/persistence/CacheDataRow.java | 4 -
.../cache/persistence/CacheDataRowAdapter.java | 2 +-
.../cache/persistence/CacheSearchRow.java | 2 +-
.../persistence/GridCacheOffheapManager.java | 2 +-
.../transactions/IgniteTxLocalAdapter.java | 2 +
.../cache/tree/AbstractDataInnerIO.java | 4 +-
.../cache/tree/AbstractDataLeafIO.java | 2 +-
.../processors/cache/tree/CacheDataTree.java | 4 +-
.../processors/cache/tree/MvccDataRow.java | 2 +-
.../processors/cache/tree/MvccSearchRow.java | 16 +-
.../processors/cache/tree/SearchRow.java | 8 +-
.../cache/mvcc/CacheMvccTransactionsTest.java | 327 ++++++++++++++-----
.../database/FreeListImplSelfTest.java | 2 +-
.../processors/query/h2/opt/GridH2Row.java | 2 +-
19 files changed, 391 insertions(+), 159 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
index 5549c78..5f5445c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
@@ -1308,7 +1308,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
if (cmp != 0)
return cmp;
- return Long.compare(row.mvccUpdateCounter(), mvccCntr);
+ return Long.compare(row.mvccCounter(), mvccCntr);
}
/** {@inheritDoc} */
@@ -1364,11 +1364,13 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
CacheDataRow oldVal = cur.get();
if (activeTxs != null && oldVal.mvccCoordinatorVersion() == mvccVer.coordinatorVersion() &&
- activeTxs.contains(oldVal.mvccUpdateCounter())) {
+ activeTxs.contains(oldVal.mvccCounter())) {
if (waitTxs == null)
waitTxs = new GridLongList();
- waitTxs.add(oldVal.mvccUpdateCounter());
+ assert oldVal.mvccCounter() != mvccVer.counter();
+
+ waitTxs.add(oldVal.mvccCounter());
}
else if (!first) {
int cmp = compare(oldVal, mvccVer.coordinatorVersion(), mvccVer.cleanupVersion());
@@ -1641,7 +1643,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
if (txs != null) {
visible = row0.mvccCoordinatorVersion() != ver.coordinatorVersion()
- || !txs.contains(row0.mvccUpdateCounter());
+ || !txs.contains(row0.mvccCounter());
}
else
visible = true;
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/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 6858c82..c72b90c 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
@@ -302,6 +302,8 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
IgniteInternalFuture fut = cctx.coordinators().waitTxsFuture(crd, waitTxs);
add(fut);
+
+ sync = true;
}
markInitialized();
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/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 e57976b..b1748b8 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
@@ -393,14 +393,6 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
@SuppressWarnings("ForLoopReplaceableByForEach")
/** {@inheritDoc} */
public void finish(boolean commit) {
- if (!commit && tx.mvccCoordinatorVersion() != null) {
- ClusterNode crd = cctx.coordinators().coordinator(tx.topologyVersion());
-
- assert crd != null;
-
- cctx.coordinators().ackTxRollback(crd, tx.nearXidVersion());
- }
-
if (tx.onNeedCheckBackup()) {
assert tx.onePhaseCommit();
@@ -413,6 +405,14 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
return;
}
+ if (!commit && tx.mvccCoordinatorVersion() != null) {
+ ClusterNode crd = cctx.coordinators().coordinator(tx.topologyVersion());
+
+ assert crd != null;
+
+ cctx.coordinators().ackTxRollback(crd, tx.nearXidVersion());
+ }
+
try {
if (tx.localFinish(commit) || (!commit && tx.state() == UNKNOWN)) {
GridLongList waitTxs = tx.mvccWaitTransactions();
@@ -437,8 +437,11 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
finish(1, mapping, commit);
}
}
- else
+ else {
+ assert !hasFutures() || waitTxs != null : futures();
+
finish(mappings.mappings(), commit);
+ }
}
markInitialized();
@@ -692,8 +695,6 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
* @param commit Commit flag.
*/
private void finish(Iterable<GridDistributedTxMapping> mappings, boolean commit) {
- assert !hasFutures() : futures();
-
int miniId = 0;
// Create mini futures.
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CacheCoordinatorsSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CacheCoordinatorsSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CacheCoordinatorsSharedManager.java
index d19af59..c82633f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CacheCoordinatorsSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CacheCoordinatorsSharedManager.java
@@ -41,9 +41,11 @@ import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
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.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -104,11 +106,7 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
public MvccCoordinatorVersion requestTxCounterOnCoordinator(IgniteInternalTx tx) {
assert cctx.localNode().equals(assignHist.currentCoordinator());
- AffinityTopologyVersion txTopVer = tx.topologyVersionSnapshot();
-
- assert txTopVer != null && txTopVer.initialized() : txTopVer;
-
- return assignTxCounter(tx.nearXidVersion(), 0L, txTopVer.topologyVersion());
+ return assignTxCounter(tx.nearXidVersion(), 0L);
}
/**
@@ -119,10 +117,6 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
public IgniteInternalFuture<MvccCoordinatorVersion> requestTxCounter(ClusterNode crd, GridDhtTxLocalAdapter tx) {
assert !crd.isLocal() : crd;
- AffinityTopologyVersion txTopVer = tx.topologyVersionSnapshot();
-
- assert txTopVer != null && txTopVer.initialized() : txTopVer;
-
MvccVersionFuture fut = new MvccVersionFuture(futIdCntr.incrementAndGet(),
crd,
tx);
@@ -132,7 +126,7 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
try {
cctx.gridIO().sendToGridTopic(crd,
TOPIC_CACHE_COORDINATOR,
- new CoordinatorTxCounterRequest(fut.id, tx.nearXidVersion(), txTopVer.topologyVersion()),
+ new CoordinatorTxCounterRequest(fut.id, tx.nearXidVersion()),
SYSTEM_POOL);
}
catch (IgniteCheckedException e) {
@@ -195,6 +189,11 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
* @return Future.
*/
public IgniteInternalFuture<Void> waitTxsFuture(ClusterNode crd, GridLongList txs) {
+ assert crd != null;
+ assert txs != null && txs.size() > 0;
+
+ // TODO IGNITE-3478: special case for local?
+
WaitAckFuture fut = new WaitAckFuture(futIdCntr.incrementAndGet(), crd);
ackFuts.put(fut.id, fut);
@@ -223,6 +222,9 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
* @return Acknowledge future.
*/
public IgniteInternalFuture<Void> ackTxCommit(ClusterNode crd, GridCacheVersion txId) {
+ assert crd != null;
+ assert txId != null;
+
WaitAckFuture fut = new WaitAckFuture(futIdCntr.incrementAndGet(), crd);
ackFuts.put(fut.id, fut);
@@ -283,7 +285,7 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
return;
}
- MvccCoordinatorVersionResponse res = assignTxCounter(msg.txId(), msg.futureId(), msg.topologyVersion());
+ MvccCoordinatorVersionResponse res = assignTxCounter(msg.txId(), msg.futureId());
try {
cctx.gridIO().sendToGridTopic(node,
@@ -403,10 +405,9 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
/**
* @param txId Transaction ID.
- * @param topVer Topology version.
* @return Counter.
*/
- private synchronized MvccCoordinatorVersionResponse assignTxCounter(GridCacheVersion txId, long futId, long topVer) {
+ private synchronized MvccCoordinatorVersionResponse assignTxCounter(GridCacheVersion txId, long futId) {
assert crdVer != 0;
long nextCtr = mvccCntr.incrementAndGet();
@@ -438,14 +439,21 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
/**
* @param txId Transaction ID.
*/
- private synchronized void onTxDone(GridCacheVersion txId) {
- Long cntr = activeTxs.remove(txId);
+ private void onTxDone(GridCacheVersion txId) {
+ GridFutureAdapter fut; // TODO IGNITE-3478.
+
+ synchronized (this) {
+ Long cntr = activeTxs.remove(txId);
+
+ assert cntr != null;
- assert cntr != null;
+ committedCntr.setIfGreater(cntr);
- committedCntr.setIfGreater(cntr);
+ fut = waitTxFuts.remove(cntr);
+ }
- notifyAll(); // TODO IGNITE-3478.
+ if (fut != null)
+ fut.onDone();
}
/**
@@ -492,27 +500,58 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
activeQueries.remove(mvccCntr);
}
+ /** */
+ private Map<Long, GridFutureAdapter> waitTxFuts = new HashMap<>(); // TODO IGNITE-3478.
+
/**
* @param msg Message.
*/
- private void processCoordinatorWaitTxsRequest(UUID nodeId, CoordinatorWaitTxsRequest msg) {
+ private void processCoordinatorWaitTxsRequest(final UUID nodeId, final CoordinatorWaitTxsRequest msg) {
GridLongList txs = msg.transactions();
// TODO IGNITE-3478.
+ GridCompoundFuture fut = null;
+
synchronized (this) {
for (int i = 0; i < txs.size(); i++) {
long txId = txs.get(i);
- while (hasActiveTx(txId)) {
- try {
- wait();
- }
- catch (InterruptedException e) {
- e.printStackTrace();
+ if (hasActiveTx(txId)) {
+ GridFutureAdapter fut0 = waitTxFuts.get(txId);
+
+ if (fut0 == null) {
+ fut0 = new GridFutureAdapter();
+
+ waitTxFuts.put(txId, fut0);
}
+
+ if (fut == null)
+ fut = new GridCompoundFuture();
+
+ fut.add(fut0);
}
}
}
+
+ if (fut != null)
+ fut.markInitialized();
+
+ if (fut == null || fut.isDone())
+ sendFutureResponse(nodeId, msg);
+ else {
+ fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
+ @Override public void apply(IgniteInternalFuture fut) {
+ sendFutureResponse(nodeId, msg);
+ }
+ });
+ }
+ }
+
+ /**
+ * @param nodeId
+ * @param msg
+ */
+ private void sendFutureResponse(UUID nodeId, CoordinatorWaitTxsRequest msg) {
try {
cctx.gridIO().sendToGridTopic(nodeId,
TOPIC_CACHE_COORDINATOR,
@@ -526,7 +565,6 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
catch (IgniteCheckedException e) {
U.error(log, "Failed to send tx ack response [msg=" + msg + ", node=" + nodeId + ']', e);
}
-
}
private boolean hasActiveTx(long txId) {
@@ -621,6 +659,11 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
"coordinator failed: " + nodeId));
}
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "MvccVersionFuture [crd=" + crd + ", id=" + id + ']';
+ }
}
/**
@@ -656,6 +699,11 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
if (crd.id().equals(nodeId) && verFuts.remove(id) != null)
onDone();
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "WaitAckFuture [crd=" + crd + ", id=" + id + ']';
+ }
}
/**
@@ -676,6 +724,11 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
for (WaitAckFuture fut : ackFuts.values())
fut.onNodeLeft(nodeId);
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "CacheCoordinatorDiscoveryListener[]";
+ }
}
/**
*
@@ -718,5 +771,10 @@ public class CacheCoordinatorsSharedManager<K, V> extends GridCacheSharedManager
else
U.warn(log, "Unexpected message received [node=" + nodeId + ", msg=" + msg + ']');
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "CoordinatorMessageListener[]";
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CoordinatorTxCounterRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CoordinatorTxCounterRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CoordinatorTxCounterRequest.java
index fe3c547..6073042 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CoordinatorTxCounterRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/CoordinatorTxCounterRequest.java
@@ -37,9 +37,6 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
/** */
private GridCacheVersion txId;
- /** */
- private long topVer;
-
/**
* Required by {@link GridIoMessageFactory}.
*/
@@ -51,12 +48,11 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
* @param futId Future ID.
* @param txId Transaction ID.
*/
- CoordinatorTxCounterRequest(long futId, GridCacheVersion txId, long topVer) {
+ CoordinatorTxCounterRequest(long futId, GridCacheVersion txId) {
assert txId != null;
this.futId = futId;
this.txId = txId;
- this.topVer = topVer;
}
/** {@inheritDoc} */
@@ -64,10 +60,6 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
return true;
}
- public long topologyVersion() {
- return topVer;
- }
-
/**
* @return Future ID.
*/
@@ -101,12 +93,6 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
writer.incrementState();
case 1:
- if (!writer.writeLong("topVer", topVer))
- return false;
-
- writer.incrementState();
-
- case 2:
if (!writer.writeMessage("txId", txId))
return false;
@@ -134,14 +120,6 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
reader.incrementState();
case 1:
- topVer = reader.readLong("topVer");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 2:
txId = reader.readMessage("txId");
if (!reader.isLastRead())
@@ -161,7 +139,7 @@ public class CoordinatorTxCounterRequest implements MvccCoordinatorMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 3;
+ return 2;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
index 7c52c7d..57aeaef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
@@ -54,8 +54,4 @@ public interface CacheDataRow extends CacheSearchRow {
* @param key Key.
*/
public void key(KeyCacheObject key);
-
- public long mvccCoordinatorVersion();
-
- public long mvccUpdateCounter();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
index 4aef9f0..925431f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
@@ -578,7 +578,7 @@ public class CacheDataRowAdapter implements CacheDataRow {
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return 0;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheSearchRow.java
index 533d8f5..5bf53d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheSearchRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheSearchRow.java
@@ -45,5 +45,5 @@ public interface CacheSearchRow {
public long mvccCoordinatorVersion();
- public long mvccUpdateCounter();
+ public long mvccCounter();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index ffcfd8e..bbb2c83 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -826,7 +826,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return 0; // TODO IGNITE-3478.
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/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 1b386d8..949b6e7 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
@@ -506,6 +506,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
cctx.tm().addCommittedTx(this);
if (!empty) {
+ assert mvccWaitTxs == null;
+
batchStoreCommit(writeEntries());
WALPointer ptr = null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataInnerIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataInnerIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataInnerIO.java
index d87b5ca..a076e5c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataInnerIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataInnerIO.java
@@ -60,12 +60,12 @@ public abstract class AbstractDataInnerIO extends BPlusInnerIO<CacheSearchRow> i
if (storeMvccVersion()) {
assert row.mvccCoordinatorVersion() > 0 : row;
- assert row.mvccUpdateCounter() != CacheCoordinatorsSharedManager.COUNTER_NA : row;
+ assert row.mvccCounter() != CacheCoordinatorsSharedManager.COUNTER_NA : row;
PageUtils.putLong(pageAddr, off, row.mvccCoordinatorVersion());
off += 8;
- PageUtils.putLong(pageAddr, off, row.mvccUpdateCounter());
+ PageUtils.putLong(pageAddr, off, row.mvccCounter());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataLeafIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataLeafIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataLeafIO.java
index 0be84c6..a3a8416 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataLeafIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/AbstractDataLeafIO.java
@@ -59,7 +59,7 @@ public abstract class AbstractDataLeafIO extends BPlusLeafIO<CacheSearchRow> imp
if (storeMvccVersion()) {
long mvccCrdVer = row.mvccCoordinatorVersion();
- long mvccUpdateCntr = row.mvccUpdateCounter();
+ long mvccUpdateCntr = row.mvccCounter();
assert mvccCrdVer > 0 : mvccCrdVer;
assert mvccUpdateCntr != CacheCoordinatorsSharedManager.COUNTER_NA;
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
index 3bd0b02..7345106 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataTree.java
@@ -167,9 +167,9 @@ public class CacheDataTree extends BPlusTree<CacheSearchRow, CacheDataRow> {
long mvccCntr = io.getMvccUpdateCounter(pageAddr, idx);
- assert row.mvccUpdateCounter() != CacheCoordinatorsSharedManager.COUNTER_NA;
+ assert row.mvccCounter() != CacheCoordinatorsSharedManager.COUNTER_NA;
- cmp = Long.compare(row.mvccUpdateCounter(), mvccCntr);
+ cmp = Long.compare(row.mvccCounter(), mvccCntr);
return cmp;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccDataRow.java
index a3d2ec4..09dc739 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccDataRow.java
@@ -78,7 +78,7 @@ public class MvccDataRow extends DataRow {
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return mvccCntr;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccSearchRow.java
index e6c5268..a2adc4f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccSearchRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/MvccSearchRow.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache.tree;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.util.typedef.internal.S;
/**
*
@@ -30,10 +31,10 @@ public class MvccSearchRow extends SearchRow {
private long mvccCntr;
/**
- * @param cacheId
- * @param key
- * @param crdVer
- * @param mvccCntr
+ * @param cacheId Cache ID.
+ * @param key Key.
+ * @param crdVer Coordinator version.
+ * @param mvccCntr Mvcc counter.
*/
public MvccSearchRow(int cacheId, KeyCacheObject key, long crdVer, long mvccCntr) {
super(cacheId, key);
@@ -48,7 +49,12 @@ public class MvccSearchRow extends SearchRow {
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return mvccCntr;
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccSearchRow.class, this);
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRow.java
index 6ab80d0..8eb667c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRow.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache.tree;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.mvcc.CacheCoordinatorsSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+import org.apache.ignite.internal.util.typedef.internal.S;
/**
*
@@ -81,7 +82,12 @@ public class SearchRow implements CacheSearchRow {
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return CacheCoordinatorsSharedManager.COUNTER_NA;
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(SearchRow.class, this);
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTransactionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTransactionsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTransactionsTest.java
index e3b751e..9891f2f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTransactionsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTransactionsTest.java
@@ -72,11 +72,15 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
/**
* TODO IGNITE-3478: extend tests to use single/mutiple nodes, all tx types.
*/
+@SuppressWarnings("unchecked")
public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
/** */
private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
/** */
+ private static final long DEFAULT_TEST_TIME = 30_000;
+
+ /** */
private static final int SRVS = 4;
/** */
@@ -102,6 +106,11 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return DEFAULT_TEST_TIME + 60_000;
+ }
+
+ /** {@inheritDoc} */
@Override protected void afterTest() throws Exception {
verifyCoordinatorInternalState();
@@ -305,7 +314,7 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
/**
* @throws Exception If failed.
*/
- public void testMyUpdatesAreVisible() throws Exception {
+ public void testThreadUpdatesAreVisibleForThisThread() throws Exception {
final Ignite ignite = startGrid(0);
final IgniteCache<Object, Object> cache = ignite.createCache(cacheConfiguration(PARTITIONED, FULL_SYNC, 0, 1));
@@ -547,20 +556,61 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
/**
* @throws Exception If failed.
*/
- public void testPutAllGetAll() throws Exception {
- final int RANGE = 20;
+ public void testPutAllGetAll_SingleNode() throws Exception {
+ putAllGetAll(1, 0, 0, 64);
+ }
- final long time = 10_000;
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutAllGetAll_SingleNode_SinglePartition() throws Exception {
+ putAllGetAll(1, 0, 0, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutAllGetAll_ClientServer_Backups0() throws Exception {
+ putAllGetAll(4, 2, 0, 64);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutAllGetAll_ClientServer_Backups1() throws Exception {
+ putAllGetAll(4, 2, 1, 64);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutAllGetAll_ClientServer_Backups2() throws Exception {
+ putAllGetAll(4, 2, 2, 64);
+ }
+
+ /**
+ * @param srvs Number of server nodes.
+ * @param clients Number of client nodes.
+ * @param cacheBackups Number of cache backups.
+ * @param cacheParts Number of cache partitions.
+ * @throws Exception If failed.
+ */
+ private void putAllGetAll(
+ final int srvs,
+ final int clients,
+ int cacheBackups,
+ int cacheParts
+ ) throws Exception
+ {
+ final int RANGE = 20;
final int writers = 4;
final int readers = 4;
- GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> writer =
- new GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean>() {
- @Override public void apply(Integer idx, IgniteCache<Object, Object> cache, AtomicBoolean stop) {
- final IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
-
+ GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> writer =
+ new GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<IgniteCache> caches, AtomicBoolean stop) {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
int min = idx * RANGE;
@@ -576,6 +626,10 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
while (map.size() < RANGE)
map.put(rnd.nextInt(min, max), v);
+ IgniteCache<Integer, Integer> cache = randomCache(caches, rnd);
+
+ IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
+
try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
cache.putAll(map);
@@ -591,17 +645,14 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
}
};
- GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> reader =
- new GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean>() {
- @Override public void apply(Integer idx, IgniteCache<Object, Object> cache, AtomicBoolean stop) {
+ GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> reader =
+ new GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<IgniteCache> caches, AtomicBoolean stop) {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
Set<Integer> keys = new LinkedHashSet<>();
- Map<Integer, Set<Integer>> uniqueReads = new HashMap<>();
-
- for (int i = 0; i < writers; i++)
- uniqueReads.put(i, new HashSet<Integer>());
+ Map<Integer, Integer> readVals = new HashMap<>();
while (!stop.get()) {
int range = rnd.nextInt(0, writers);
@@ -612,22 +663,29 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
while (keys.size() < RANGE)
keys.add(rnd.nextInt(min, max));
- Map<Object, Object> map = cache.getAll(keys);
+ IgniteCache<Integer, Integer> cache = randomCache(caches, rnd);
+
+ Map<Integer, Integer> map = cache.getAll(keys);
assertTrue("Invalid map size: " + map.size(),
map.isEmpty() || map.size() == RANGE);
Integer val0 = null;
- for (Map.Entry<Object, Object> e: map.entrySet()) {
- Object val = e.getValue();
+ for (Map.Entry<Integer, Integer> e: map.entrySet()) {
+ Integer val = e.getValue();
assertNotNull(val);
if (val0 == null) {
- uniqueReads.get(range).add((Integer)val);
+ Integer readVal = readVals.get(range);
+
+ if (readVal != null)
+ assertTrue(readVal <= val);
+
+ readVals.put(range, val);
- val0 = (Integer)val;
+ val0 = val;
}
else {
if (!F.eq(val0, val)) {
@@ -640,27 +698,75 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
keys.clear();
}
-
- info("Reader done, unique reads: ");
-
- for (Map.Entry<Integer, Set<Integer>> e : uniqueReads.entrySet())
- info("Range [idx=" + e.getKey() + ", uniqueReads=" + e.getValue().size() + ']');
}
};
- readWriteTest(time, writers, readers, null, writer, reader);
+ readWriteTest(srvs,
+ clients,
+ cacheBackups,
+ cacheParts,
+ writers,
+ readers,
+ DEFAULT_TEST_TIME,
+ null,
+ writer,
+ reader);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAccountsTxGetAll_SingleNode() throws Exception {
+ accountsTxGetAll(1, 0, 0, 64);
}
/**
* @throws Exception If failed.
*/
- public void testAccountsTxGetAll() throws Exception {
+ public void testAccountsTxGetAll_SingleNode_SinglePartition() throws Exception {
+ accountsTxGetAll(1, 0, 0, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAccountsTxGetAll_ClientServer_Backups0() throws Exception {
+ accountsTxGetAll(4, 2, 0, 64);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAccountsTxGetAll_ClientServer_Backups1() throws Exception {
+ accountsTxGetAll(4, 2, 1, 64);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAccountsTxGetAll_ClientServer_Backups2() throws Exception {
+ accountsTxGetAll(4, 2, 2, 64);
+ }
+
+ /**
+ * @param srvs Number of server nodes.
+ * @param clients Number of client nodes.
+ * @param cacheBackups Number of cache backups.
+ * @param cacheParts Number of cache partitions.
+ * @throws Exception If failed.
+ */
+ private void accountsTxGetAll(
+ final int srvs,
+ final int clients,
+ int cacheBackups,
+ int cacheParts
+ )
+ throws Exception
+ {
final int ACCOUNTS = 20;
final int ACCOUNT_START_VAL = 1000;
- final long time = 10_000;
-
final int writers = 4;
final int readers = 4;
@@ -669,10 +775,10 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
@Override public void apply(IgniteCache<Object, Object> cache) {
final IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
- Map<Integer, Account> accounts = new HashMap<>();
+ Map<Integer, MvccTestAccount> accounts = new HashMap<>();
for (int i = 0; i < ACCOUNTS; i++)
- accounts.put(i, new Account(ACCOUNT_START_VAL));
+ accounts.put(i, new MvccTestAccount(ACCOUNT_START_VAL, 1));
try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
cache.putAll(accounts);
@@ -682,16 +788,17 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
}
};
- GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> writer =
- new GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean>() {
- @Override public void apply(Integer idx, IgniteCache<Object, Object> cache, AtomicBoolean stop) {
- final IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
-
+ GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> writer =
+ new GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<IgniteCache> caches, AtomicBoolean stop) {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
int cnt = 0;
while (!stop.get()) {
+ IgniteCache<Integer, MvccTestAccount> cache = randomCache(caches, rnd);
+ IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
+
cnt++;
Integer id1 = rnd.nextInt(ACCOUNTS);
@@ -700,69 +807,99 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
while (id1.equals(id2))
id2 = rnd.nextInt(ACCOUNTS);
- try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
- Account a1;
- Account a2;
+ TreeSet<Integer> keys = new TreeSet<>();
+
+ keys.add(id1);
+ keys.add(id2);
- TreeSet<Integer> keys = new TreeSet<>();
+ Integer cntr1;
+ Integer cntr2;
- keys.add(id1);
- keys.add(id2);
+ try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ MvccTestAccount a1;
+ MvccTestAccount a2;
- Map<Object, Object> accounts = cache.getAll(keys);
+ Map<Integer, MvccTestAccount> accounts = cache.getAll(keys);
- a1 = (Account)accounts.get(id1);
- a2 = (Account)accounts.get(id2);
+ a1 = accounts.get(id1);
+ a2 = accounts.get(id2);
assertNotNull(a1);
assertNotNull(a2);
- cache.put(id1, new Account(a1.val + 1));
- cache.put(id2, new Account(a2.val - 1));
+ cntr1 = a1.updateCnt + 1;
+ cntr2 = a2.updateCnt + 1;
+
+ cache.put(id1, new MvccTestAccount(a1.val + 1, cntr1));
+ cache.put(id2, new MvccTestAccount(a2.val - 1, cntr2));
tx.commit();
}
+
+ Map<Integer, MvccTestAccount> accounts = cache.getAll(keys);
+
+ MvccTestAccount a1 = accounts.get(id1);
+ MvccTestAccount a2 = accounts.get(id2);
+
+ assertNotNull(a1);
+ assertNotNull(a2);
+
+ assertTrue(a1.updateCnt >= cntr1);
+ assertTrue(a2.updateCnt >= cntr2);
}
info("Writer finished, updates: " + cnt);
}
};
- GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> reader =
- new GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean>() {
- @Override public void apply(Integer idx, IgniteCache<Object, Object> cache, AtomicBoolean stop) {
+ GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> reader =
+ new GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<IgniteCache> caches, AtomicBoolean stop) {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
Set<Integer> keys = new LinkedHashSet<>();
+ Map<Integer, Integer> lastUpdateCntrs = new HashMap<>();
+
while (!stop.get()) {
while (keys.size() < ACCOUNTS)
keys.add(rnd.nextInt(ACCOUNTS));
- Map<Object, Object> accounts = cache.getAll(keys);
+ IgniteCache<Integer, MvccTestAccount> cache = randomCache(caches, rnd);
+
+ Map<Integer, MvccTestAccount> accounts = cache.getAll(keys);
assertEquals(ACCOUNTS, accounts.size());
int sum = 0;
for (int i = 0; i < ACCOUNTS; i++) {
- Account account = (Account)accounts.get(i);
+ MvccTestAccount account = accounts.get(i);
assertNotNull(account);
sum += account.val;
+
+ Integer cntr = lastUpdateCntrs.get(i);
+
+ if (cntr != null)
+ assertTrue(cntr <= account.updateCnt);
+
+ lastUpdateCntrs.put(i, cntr);
}
assertEquals(ACCOUNTS * ACCOUNT_START_VAL, sum);
}
if (idx == 0) {
- Map<Object, Object> accounts = cache.getAll(keys);
+ IgniteCache<Integer, MvccTestAccount> cache = randomCache(caches, rnd);
+
+ Map<Integer, MvccTestAccount> accounts = cache.getAll(keys);
int sum = 0;
for (int i = 0; i < ACCOUNTS; i++) {
- Account account = (Account)accounts.get(i);
+ MvccTestAccount account = (MvccTestAccount)accounts.get(i);
info("Account [id=" + i + ", val=" + account.val + ']');
@@ -774,10 +911,23 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
}
};
- readWriteTest(time, writers, readers, init, writer, reader);
+ readWriteTest(srvs,
+ clients,
+ cacheBackups,
+ cacheParts,
+ writers,
+ readers,
+ DEFAULT_TEST_TIME,
+ init,
+ writer,
+ reader);
}
/**
+ * @param srvs Number of server nodes.
+ * @param clients Number of client nodes.
+ * @param cacheBackups Number of cache backups.
+ * @param cacheParts Number of cache partitions.
* @param time Test time.
* @param writers Number of writers.
* @param readers Number of readers.
@@ -786,19 +936,41 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
* @param reader Readers threads closure.
* @throws Exception If failed.
*/
- private void readWriteTest(final long time,
+ private void readWriteTest(
+ final int srvs,
+ final int clients,
+ int cacheBackups,
+ int cacheParts,
final int writers,
final int readers,
+ final long time,
IgniteInClosure<IgniteCache<Object, Object>> init,
- final GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> writer,
- final GridInClosure3<Integer, IgniteCache<Object, Object>, AtomicBoolean> reader) throws Exception {
- final Ignite ignite = startGrid(0);
+ final GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> writer,
+ final GridInClosure3<Integer, List<IgniteCache>, AtomicBoolean> reader) throws Exception {
+ Ignite srv0 = startGridsMultiThreaded(srvs);
- final IgniteCache<Object, Object> cache = ignite.createCache(cacheConfiguration(PARTITIONED, FULL_SYNC, 0, 1));
+ if (clients > 0) {
+ client = true;
+
+ startGridsMultiThreaded(srvs, clients);
+ }
+
+ IgniteCache<Object, Object> cache = srv0.createCache(cacheConfiguration(PARTITIONED,
+ FULL_SYNC,
+ cacheBackups,
+ cacheParts));
if (init != null)
init.apply(cache);
+ final List<IgniteCache> caches = new ArrayList<>(srvs + clients);
+
+ for (int i = 0; i < srvs + clients; i++) {
+ Ignite node = grid(i);
+
+ caches.add(node.cache(cache.getName()));
+ }
+
final long stopTime = U.currentTimeMillis() + time;
final AtomicBoolean stop = new AtomicBoolean();
@@ -811,7 +983,7 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
try {
int idx = writerIdx.getAndIncrement();
- writer.apply(idx, cache, stop);
+ writer.apply(idx, caches, stop);
}
catch (Throwable e) {
error("Unexpected error: " + e, e);
@@ -832,7 +1004,7 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
try {
int idx = readerIdx.getAndIncrement();
- reader.apply(idx, cache, stop);
+ reader.apply(idx, caches, stop);
}
catch (Throwable e) {
error("Unexpected error: " + e, e);
@@ -953,29 +1125,38 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest {
}
/**
+ * @param caches Caches.
+ * @param rnd Random.
+ * @return Random cache.
+ */
+ private static <K, V> IgniteCache<K, V> randomCache(List<IgniteCache> caches, ThreadLocalRandom rnd) {
+ return caches.size() > 1 ? caches.get(rnd.nextInt(caches.size())): caches.get(0);
+ }
+
+ /**
*
*/
- static class Account {
+ static class MvccTestAccount {
/** */
private final int val;
+ /** */
+ private final int updateCnt;
+
/**
* @param val Value.
+ * @param updateCnt Updates counter.
*/
- public Account(int val) {
- this.val = val;
- }
+ MvccTestAccount(int val, int updateCnt) {
+ assert updateCnt > 0;
- /**
- * @return Value.
- */
- public int value() {
- return val;
+ this.val = val;
+ this.updateCnt = updateCnt;
}
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(Account.class, this);
+ return S.toString(MvccTestAccount.class, this);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
index 6ae5d6b..d0897e1 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
@@ -432,7 +432,7 @@ public class FreeListImplSelfTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
return 0;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6d170264/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Row.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Row.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Row.java
index 02e4df8..5899b7b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Row.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Row.java
@@ -186,7 +186,7 @@ public abstract class GridH2Row implements GridSearchRowPointer, CacheDataRow, R
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public long mvccCounter() {
throw new UnsupportedOperationException();
}
}
\ No newline at end of file