ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject ignite git commit: ignite-5932
Date Mon, 16 Oct 2017 11:37:45 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-5932 8745522b7 -> f5aa79e83


ignite-5932


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f5aa79e8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f5aa79e8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f5aa79e8

Branch: refs/heads/ignite-5932
Commit: f5aa79e834806a93b5a3317bd85edf2bec1cc3d1
Parents: 8745522
Author: sboikov <sboikov@gridgain.com>
Authored: Mon Oct 16 14:37:34 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Mon Oct 16 14:37:34 2017 +0300

----------------------------------------------------------------------
 .../communication/GridIoMessageFactory.java     |   6 +
 .../processors/cache/GridCacheEntryInfo.java    |  10 +-
 .../processors/cache/GridCacheMapEntry.java     |   5 +-
 .../cache/IgniteCacheOffheapManagerImpl.java    |   5 +-
 .../distributed/dht/GridDhtTxFinishFuture.java  |  10 +-
 .../distributed/dht/GridDhtTxPrepareFuture.java |  10 +-
 .../cache/mvcc/MvccCoordinatorVersion.java      |   5 +
 .../mvcc/MvccCoordinatorVersionResponse.java    |  12 +-
 .../mvcc/MvccCoordinatorVersionWithoutTxs.java  | 173 +++++++++++++++++++
 .../cache/mvcc/MvccEmptyLongList.java           |  53 ++++++
 .../processors/cache/mvcc/TxMvccInfo.java       |  12 ++
 .../datastreamer/DataStreamerImpl.java          |   4 +-
 .../cache/mvcc/CacheMvccTransactionsTest.java   |   2 +-
 13 files changed, 294 insertions(+), 13 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 a2440f6..35e9af3 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
@@ -113,6 +113,7 @@ import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorAckRequestTxA
 import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorTxCounterRequest;
 import org.apache.ignite.internal.processors.cache.mvcc.CoordinatorWaitTxsRequest;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionWithoutTxs;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCounter;
 import org.apache.ignite.internal.processors.cache.mvcc.NewCoordinatorQueryAckRequest;
 import org.apache.ignite.internal.processors.cache.mvcc.TxMvccInfo;
@@ -959,6 +960,11 @@ public class GridIoMessageFactory implements MessageFactory {
 
                 return msg;
 
+            case 145:
+                msg = new MvccCoordinatorVersionWithoutTxs();
+
+                return msg;
+
 
             // [-3..119] [124..128] [-23..-27] [-36..-55]- this
             // [120..123] - DR

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java
index e09d33c..86a2253 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.GridDirectTransient;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccEmptyLongList;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccLongList;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -82,8 +83,13 @@ public class GridCacheEntryInfo implements Message, MvccCoordinatorVersion
{
     }
 
     /** {@inheritDoc} */
-    @Override public MvccLongList activeTransactions() {
-        return null;
+    @Override public final MvccLongList activeTransactions() {
+        return MvccEmptyLongList.INSTANCE;
+    }
+
+    /** {@inheritDoc} */
+    @Override public final MvccCoordinatorVersion withoutActiveTransactions() {
+        return this;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 66fa626..391a56e 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
@@ -2292,7 +2292,10 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
         ver = newVer;
         flags &= ~IS_EVICT_DISABLED;
 
-        removeValue();
+        if (cctx.mvccEnabled())
+            cctx.offheap().mvccRemoveAll(this);
+        else
+            removeValue();
 
         onInvalidate();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 246827a..c07e2a8 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
@@ -42,6 +42,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalP
 import org.apache.ignite.internal.processors.cache.mvcc.CacheCoordinatorsProcessor;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersion;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionWithoutTxs;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCounter;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccLongList;
 import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
@@ -1418,7 +1419,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
 
                 // TODO IGNITE-3478: null is passed for loaded from store, need handle better.
                 if (mvccVer == null) {
-                    mvccVer = new MvccCoordinatorVersionResponse(1L, CacheCoordinatorsProcessor.START_VER,
0L);
+                    mvccVer = new MvccCoordinatorVersionWithoutTxs(1L, CacheCoordinatorsProcessor.START_VER,
0L);
 
                     newVal = true;
                 }
@@ -1495,6 +1496,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
             long expireTime,
             MvccCoordinatorVersion mvccVer) throws IgniteCheckedException {
             assert mvccVer != null;
+            assert primary || mvccVer.activeTransactions().size() == 0 : mvccVer;
 
             if (!busyLock.enterBusy())
                 throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
@@ -1556,6 +1558,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
             KeyCacheObject key,
             MvccCoordinatorVersion mvccVer) throws IgniteCheckedException {
             assert mvccVer != null;
+            assert primary || mvccVer.activeTransactions().size() == 0 : mvccVer;
 
             if (!busyLock.enterBusy())
                 throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 e0117a4..5e8428d 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
@@ -424,6 +424,12 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
 
         int miniId = 0;
 
+        // Do not need process active transactions on backups.
+        TxMvccInfo mvccInfo = tx.mvccInfo();
+
+        if (mvccInfo != null)
+            mvccInfo = mvccInfo.withoutActiveTransactions();
+
         // Create mini futures.
         for (GridDistributedTxMapping dhtMapping : dhtMap.values()) {
             ClusterNode n = dhtMapping.primary();
@@ -471,7 +477,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
                 updCntrs,
                 false,
                 false,
-                tx.mvccInfo());
+                mvccInfo);
 
             req.writeVersion(tx.writeVersion() != null ? tx.writeVersion() : tx.xidVersion());
 
@@ -541,7 +547,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
                     tx.activeCachesDeploymentEnabled(),
                     false,
                     false,
-                    tx.mvccInfo());
+                    mvccInfo);
 
                 req.writeVersion(tx.writeVersion());
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 3143c4f..623dea8 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
@@ -1344,6 +1344,12 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
 
         final long timeout = timeoutObj != null ? timeoutObj.timeout : 0;
 
+        // Do not need process active transactions on backups.
+        TxMvccInfo mvccInfo = tx.mvccInfo();
+
+        if (mvccInfo != null)
+            mvccInfo = mvccInfo.withoutActiveTransactions();
+
         // Create mini futures.
         for (GridDistributedTxMapping dhtMapping : tx.dhtMap().values()) {
             assert !dhtMapping.empty();
@@ -1387,7 +1393,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
                 tx.activeCachesDeploymentEnabled(),
                 tx.storeWriteThrough(),
                 retVal,
-                tx.mvccInfo());
+                mvccInfo);
 
             int idx = 0;
 
@@ -1501,7 +1507,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
                     tx.activeCachesDeploymentEnabled(),
                     tx.storeWriteThrough(),
                     retVal,
-                    tx.mvccInfo());
+                    mvccInfo);
 
                 for (IgniteTxEntry entry : nearMapping.entries()) {
                     if (CU.writes().apply(entry)) {

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersion.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersion.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersion.java
index d80e43c..5b2e69e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersion.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersion.java
@@ -42,4 +42,9 @@ public interface MvccCoordinatorVersion extends Message {
      * @return Counter.
      */
     public long counter();
+
+    /**
+     * @return Version without active transactions.
+     */
+    public MvccCoordinatorVersion withoutActiveTransactions();
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionResponse.java
index c037226..b6a4b1f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionResponse.java
@@ -46,7 +46,7 @@ public class MvccCoordinatorVersionResponse implements MvccCoordinatorMessage,
M
     private int txsCnt;
 
     /** */
-    private long[] txs; // TODO IGNITE-3478 (do not send on backups?)
+    private long[] txs;
 
     /** */
     private long cleanupVer;
@@ -63,7 +63,7 @@ public class MvccCoordinatorVersionResponse implements MvccCoordinatorMessage,
M
      * @param cntr Counter.
      * @param cleanupVer Cleanup version.
      */
-    public MvccCoordinatorVersionResponse(long crdVer, long cntr, long cleanupVer) {
+    MvccCoordinatorVersionResponse(long crdVer, long cntr, long cleanupVer) {
         this.crdVer = crdVer;
         this.cntr = cntr;
         this.cleanupVer = cleanupVer;
@@ -154,6 +154,14 @@ public class MvccCoordinatorVersionResponse implements MvccCoordinatorMessage,
M
     }
 
     /** {@inheritDoc} */
+    @Override public MvccCoordinatorVersion withoutActiveTransactions() {
+        if (txsCnt > 0)
+            return new MvccCoordinatorVersionWithoutTxs(crdVer, cntr, cleanupVer);
+
+        return this;
+    }
+
+    /** {@inheritDoc} */
     @Override public long coordinatorVersion() {
         return crdVer;
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionWithoutTxs.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionWithoutTxs.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionWithoutTxs.java
new file mode 100644
index 0000000..f4a7378
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccCoordinatorVersionWithoutTxs.java
@@ -0,0 +1,173 @@
+/*
+ * 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 java.nio.ByteBuffer;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+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 MvccCoordinatorVersionWithoutTxs implements MvccCoordinatorVersion {
+    /** */
+    private long crdVer;
+
+    /** */
+    private long cntr;
+
+    /** */
+    private long cleanupVer;
+
+    /**
+     * Required by {@link GridIoMessageFactory}.
+     */
+    public MvccCoordinatorVersionWithoutTxs() {
+        // No-op.
+    }
+
+    /**
+     * @param crdVer Coordinator version.
+     * @param cntr Counter.
+     * @param cleanupVer Cleanup version.
+     */
+    public MvccCoordinatorVersionWithoutTxs(long crdVer, long cntr, long cleanupVer) {
+        this.crdVer = crdVer;
+        this.cntr = cntr;
+        this.cleanupVer = cleanupVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public MvccLongList activeTransactions() {
+        return MvccEmptyLongList.INSTANCE;
+    }
+
+    /** {@inheritDoc} */
+    @Override public long coordinatorVersion() {
+        return crdVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public long cleanupVersion() {
+        return cleanupVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public long counter() {
+        return cntr;
+    }
+
+    /** {@inheritDoc} */
+    @Override public MvccCoordinatorVersion withoutActiveTransactions() {
+        return this;
+    }
+
+    /** {@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("cleanupVer", cleanupVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 1:
+                if (!writer.writeLong("cntr", cntr))
+                    return false;
+
+                writer.incrementState();
+
+            case 2:
+                if (!writer.writeLong("crdVer", crdVer))
+                    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:
+                cleanupVer = reader.readLong("cleanupVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 1:
+                cntr = reader.readLong("cntr");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 2:
+                crdVer = reader.readLong("crdVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return reader.afterMessageRead(MvccCoordinatorVersionWithoutTxs.class);
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 145;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 3;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void onAckReceived() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(MvccCoordinatorVersionWithoutTxs.class, this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccEmptyLongList.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccEmptyLongList.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccEmptyLongList.java
new file mode 100644
index 0000000..7963685
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccEmptyLongList.java
@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+/**
+ *
+ */
+public class MvccEmptyLongList implements MvccLongList {
+    /** */
+    public static MvccEmptyLongList INSTANCE = new MvccEmptyLongList();
+
+    /**
+     *
+     */
+    private MvccEmptyLongList() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public int size() {
+        return 0;
+    }
+
+    /** {@inheritDoc} */
+    @Override public long get(int i) {
+        throw new IndexOutOfBoundsException();
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean contains(long val) {
+        return false;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return "MvccEmptyLongList[]";
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/TxMvccInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/TxMvccInfo.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/TxMvccInfo.java
index 81e1d76..a8054a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/TxMvccInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/TxMvccInfo.java
@@ -57,6 +57,18 @@ public class TxMvccInfo implements Message {
     }
 
     /**
+     * @return Instance without active transactions.
+     */
+    public TxMvccInfo withoutActiveTransactions() {
+        MvccCoordinatorVersion mvccVer0 = mvccVer.withoutActiveTransactions();
+
+        if (mvccVer0 == mvccVer)
+            return this;
+
+        return new TxMvccInfo(crd, mvccVer0);
+    }
+
+    /**
      * @return Coordinator node ID.
      */
     public UUID coordinatorNodeId() {

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
index 30145ab..e6300a9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
@@ -88,7 +88,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartit
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
 import org.apache.ignite.internal.processors.cache.mvcc.CacheCoordinatorsProcessor;
 import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersion;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionResponse;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinatorVersionWithoutTxs;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
 import org.apache.ignite.internal.processors.dr.GridDrType;
@@ -134,7 +134,7 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K,
V>, Delayed
 
     /** Version which is less then any version generated on coordinator. */
     private static final MvccCoordinatorVersion ISOLATED_STREAMER_MVCC_VER =
-        new MvccCoordinatorVersionResponse(1L, CacheCoordinatorsProcessor.START_VER, 0L);
+        new MvccCoordinatorVersionWithoutTxs(1L, CacheCoordinatorsProcessor.START_VER, 0L);
 
     /** Cache receiver. */
     private StreamReceiver<K, V> rcvr = ISOLATED_UPDATER;

http://git-wip-us.apache.org/repos/asf/ignite/blob/f5aa79e8/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 7faa171..9da6876 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
@@ -4117,7 +4117,7 @@ public class CacheMvccTransactionsTest extends GridCommonAbstractTest
{
                 MvccCounter cntr = ver.get2();
 
                 MvccCoordinatorVersion readVer =
-                    new MvccCoordinatorVersionResponse(cntr.coordinatorVersion(), cntr.counter(),
0);
+                    new MvccCoordinatorVersionWithoutTxs(cntr.coordinatorVersion(), cntr.counter(),
0);
 
                 row = cctx.offheap().mvccRead(cctx, key0, readVer);
 


Mime
View raw message