You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/09/25 13:55:58 UTC
[1/3] ignite git commit: IGNITE-9538: MVCC: update partition counters
on prepare phase. This closes #4795.
Repository: ignite
Updated Branches:
refs/heads/master eff5751e4 -> 3a82d4aa2
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
index 17aa9b0..7422815 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
@@ -480,6 +480,11 @@ public class CacheFreeListImplSelfTest extends GridCommonAbstractTest {
@Override public byte newMvccTxState() {
return 0;
}
+
+ /** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return false;
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 2133e1a..6246aa5 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
@@ -126,6 +126,10 @@ public abstract class GridH2Row extends GridH2SearchRowAdapter implements CacheD
return row.newMvccTxState();
}
+ /** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return row.isKeyAbsentBefore();
+ }
/** {@inheritDoc} */
@Override public boolean indexSearchRow() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSqlUpdateCountersTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSqlUpdateCountersTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSqlUpdateCountersTest.java
new file mode 100644
index 0000000..407248d
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSqlUpdateCountersTest.java
@@ -0,0 +1,549 @@
+/*
+ * 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.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteTransactions;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.query.FieldsQueryCursor;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.util.lang.GridInClosure3;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.transactions.Transaction;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.processors.cache.mvcc.CacheMvccAbstractTest.ReadMode.SQL;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Test for MVCC caches update counters behaviour.
+ */
+@SuppressWarnings("unchecked")
+public class CacheMvccSqlUpdateCountersTest extends CacheMvccAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return CacheMode.PARTITIONED;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testUpdateCountersInsertSimple() throws Exception {
+ ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, 2, DFLT_PARTITION_COUNT)
+ .setIndexedTypes(Integer.class, Integer.class);
+
+ Ignite node = startGridsMultiThreaded(3);
+
+ IgniteCache cache = node.cache(DEFAULT_CACHE_NAME);
+
+ Affinity aff = affinity(cache);
+
+ Integer key1 = 1;
+
+ int part1 = aff.partition(key1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("INSERT INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("UPDATE Integer SET _val=2 WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testUpdateCountersDoubleUpdate() throws Exception {
+ ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, 2, DFLT_PARTITION_COUNT)
+ .setIndexedTypes(Integer.class, Integer.class);
+
+ Ignite node = startGridsMultiThreaded(3);
+
+ IgniteCache cache = node.cache(DEFAULT_CACHE_NAME);
+
+ Affinity aff = affinity(cache);
+
+ int key1 = 1;
+
+ int part1 = aff.partition(key1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("INSERT INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("UPDATE Integer SET _val=2 WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("UPDATE Integer SET _val=2 WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testUpdateCountersRollback() throws Exception {
+ ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, 2, DFLT_PARTITION_COUNT)
+ .setIndexedTypes(Integer.class, Integer.class);
+
+ Ignite node = startGridsMultiThreaded(3);
+
+ IgniteCache cache = node.cache(DEFAULT_CACHE_NAME);
+
+ Affinity aff = affinity(cache);
+
+ int key1 = 1;
+
+ int part1 = aff.partition(key1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("INSERT INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("UPDATE Integer SET _val=2 WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.rollback();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 0);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("UPDATE Integer SET _val=2 WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.rollback();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 0);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDeleteOwnKey() throws Exception {
+ ccfg = cacheConfiguration(cacheMode(), FULL_SYNC, 2, 1)
+ .setCacheMode(CacheMode.REPLICATED)
+ .setIndexedTypes(Integer.class, Integer.class);
+
+ Ignite node = startGridsMultiThreaded(1);
+
+ IgniteCache cache = node.cache(DEFAULT_CACHE_NAME);
+
+ Affinity aff = affinity(cache);
+
+ int key1 = 1;
+
+ int part1 = aff.partition(key1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("INSERT INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",2)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",3)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",4)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",5)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 0);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",1)");
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 1);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 2);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 2);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",5)");
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 3);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",5)");
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 4);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",6)");
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 5);
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM Integer WHERE _key=" + key1);
+
+ cache.query(qry).getAll();
+
+ qry = new SqlFieldsQuery("MERGE INTO Integer (_key, _val) values (" + key1 + ",5)");
+
+ cache.query(qry).getAll();
+
+ tx.commit();
+ }
+
+ checkUpdateCounters(DEFAULT_CACHE_NAME, part1, 6);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testUpdateCountersMultithreaded() throws Exception {
+ final int writers = 4;
+
+ final int readers = 0;
+
+ int parts = 8;
+
+ int keys = 20;
+
+ final Map<Integer, AtomicLong> tracker = new ConcurrentHashMap<>();
+
+ for (int i = 0; i< keys; i++)
+ tracker.put(i, new AtomicLong(1));
+
+ final IgniteInClosure<IgniteCache<Object, Object>> init = new IgniteInClosure<IgniteCache<Object, Object>>() {
+ @Override public void apply(IgniteCache<Object, Object> cache) {
+ final IgniteTransactions txs = cache.unwrap(Ignite.class).transactions();
+ try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ SqlFieldsQuery qry = new SqlFieldsQuery("INSERT INTO MvccTestAccount(_key, val, updateCnt) VALUES " +
+ "(?, 0, 1)");
+
+ for (int i = 0; i < keys; i++) {
+ try (FieldsQueryCursor<List<?>> cur = cache.query(qry.setArgs(i))) {
+ assertEquals(1L, cur.iterator().next().get(0));
+ }
+
+ tx.commit();
+ }
+ }
+ }
+ };
+
+ GridInClosure3<Integer, List<TestCache>, AtomicBoolean> writer =
+ new GridInClosure3<Integer, List<TestCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<TestCache> caches, AtomicBoolean stop) {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ Map<Integer, AtomicLong> acc = new HashMap<>();
+
+ int v = 0;
+
+ while (!stop.get()) {
+ int cnt = rnd.nextInt(keys / 3);
+
+ if (cnt == 0)
+ cnt = 2;
+
+ // Generate key set to be changed in tx.
+ while (acc.size() < cnt)
+ acc.put(rnd.nextInt(cnt), new AtomicLong());
+
+ TestCache<Integer, Integer> cache = randomCache(caches, rnd);
+
+ boolean success = true;
+
+ try {
+ IgniteTransactions txs = cache.cache.unwrap(Ignite.class).transactions();
+
+ try (Transaction tx = txs.txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ Map<Integer, MvccTestAccount> allVals = readAllByMode(cache.cache, tracker.keySet(), SQL, ACCOUNT_CODEC);
+
+ boolean rmv = allVals.size() > keys * 2 / 3;
+
+ for (Map.Entry<Integer, AtomicLong> e : acc.entrySet()) {
+ int key = e.getKey();
+
+ AtomicLong accCntr = e.getValue();
+
+ boolean exists = allVals.containsKey(key);
+
+ int delta = 0;
+
+ boolean createdInTx = false;
+
+ if (rmv && rnd.nextBoolean()) {
+ if (exists)
+ delta = 1;
+
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM MvccTestAccount WHERE _key=" + key);
+
+ cache.cache.query(qry).getAll();
+ }
+ else {
+ delta = 1;
+
+ if (!exists)
+ createdInTx = true;
+
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO MvccTestAccount " +
+ "(_key, val, updateCnt) VALUES (" + key + ", " + rnd.nextInt(100) + ", 1)");
+
+ cache.cache.query(qry).getAll();
+ }
+
+ if (rnd.nextBoolean()) {
+ if (createdInTx)
+ delta = 0; // Do not count cases when key created and removed in the same tx.
+
+ SqlFieldsQuery qry = new SqlFieldsQuery("DELETE FROM MvccTestAccount WHERE _key=" + key);
+
+ cache.cache.query(qry).getAll();
+ }
+ else {
+ delta = 1;
+
+ SqlFieldsQuery qry = new SqlFieldsQuery("MERGE INTO MvccTestAccount " +
+ "(_key, val, updateCnt) VALUES (" + key + ", " + rnd.nextInt(100) + ", 1)");
+
+ cache.cache.query(qry).getAll();
+ }
+
+ accCntr.addAndGet(delta);
+ }
+
+ tx.commit();
+ }
+ }
+ catch (Exception e) {
+ handleTxException(e);
+
+ success = false;
+
+ int r= 0;
+
+ for (Map.Entry<Integer, AtomicLong> en : acc.entrySet()) {
+ if (((IgniteCacheProxy)cache.cache).context().affinity().partition(en.getKey()) == 0)
+ r += en.getValue().intValue();
+ }
+ }
+ finally {
+ cache.readUnlock();
+
+ if (success) {
+ v++;
+
+ for (Map.Entry<Integer, AtomicLong> e : acc.entrySet()) {
+ int k = e.getKey();
+ long updCntr = e.getValue().get();
+
+ tracker.get(k).addAndGet(updCntr);
+ }
+
+ int r= 0;
+
+ for (Map.Entry<Integer, AtomicLong> en : acc.entrySet()) {
+ if (((IgniteCacheProxy)cache.cache).context().affinity().partition(en.getKey()) == 0)
+ r += en.getValue().intValue();
+ }
+ }
+
+ acc.clear();
+ }
+ }
+
+ info("Writer done, updates: " + v);
+ }
+ };
+
+ GridInClosure3<Integer, List<TestCache>, AtomicBoolean> reader =
+ new GridInClosure3<Integer, List<TestCache>, AtomicBoolean>() {
+ @Override public void apply(Integer idx, List<TestCache> caches, AtomicBoolean stop) {
+ // No-op.
+ }
+ };
+
+ readWriteTest(
+ null,
+ 4,
+ 1,
+ 2,
+ parts,
+ writers,
+ readers,
+ 30_000,
+ new InitIndexing(Integer.class, MvccTestAccount.class),
+ init,
+ writer,
+ reader);
+
+ Map<Integer, AtomicLong> updPerParts = new HashMap<>(parts);
+
+ Affinity aff = grid(1).cachex(DEFAULT_CACHE_NAME).affinity();
+
+ for (Map.Entry<Integer, AtomicLong> e : tracker.entrySet()) {
+ int k = e.getKey();
+ long updCntr = e.getValue().get();
+
+ int p = aff.partition(k);
+
+ AtomicLong cntr = updPerParts.get(p);
+
+ if (cntr == null) {
+ cntr = new AtomicLong();
+
+ updPerParts.putIfAbsent(p, cntr);
+ }
+
+ cntr.addAndGet(updCntr);
+ }
+
+ for (Map.Entry<Integer, AtomicLong> e : updPerParts.entrySet())
+ checkUpdateCounters(DEFAULT_CACHE_NAME, e.getKey(), e.getValue().get());
+ }
+
+ /**
+ * Checks update counter value on all nodes.
+ *
+ * @param cacheName Cache name.
+ * @param p Part number.
+ * @param val Expected partition counter value.
+ */
+ private void checkUpdateCounters(String cacheName, int p, long val) {
+ for (Ignite node : G.allGrids()) {
+ if (!node.configuration().isClientMode()) {
+ IgniteCacheProxy cache = (IgniteCacheProxy)node.cache(cacheName);
+
+
+ GridDhtLocalPartition part = cache.context().topology().localPartition(p);
+
+ if (!cache.context().mvccEnabled() || part == null)
+ continue;
+
+ assertEquals("part=" + p, val, part.updateCounter());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
index 888b1ba..726c4e9 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccScanQueryWithCo
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSizeTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSizeWithConcurrentJdbcTransactionTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSqlConfigurationValidationTest;
+import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSqlUpdateCountersTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSqlLockTimeoutTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccStreamingInsertTest;
import org.apache.ignite.internal.processors.query.h2.GridIndexRebuildWithMvccEnabledSelfTest;
@@ -58,6 +59,7 @@ public class IgniteCacheMvccSqlTestSuite extends TestSuite {
suite.addTestSuite(CacheMvccDmlSimpleTest.class);
suite.addTestSuite(SqlTransactionsCommandsWithMvccEnabledSelfTest.class);
suite.addTestSuite(CacheMvccSizeTest.class);
+ suite.addTestSuite(CacheMvccSqlUpdateCountersTest.class);
suite.addTestSuite(CacheMvccSqlLockTimeoutTest.class);
suite.addTestSuite(GridIndexRebuildWithMvccEnabledSelfTest.class);
[2/3] ignite git commit: IGNITE-9538: MVCC: update partition counters
on prepare phase. This closes #4795.
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
index 5df1d58..5c1bf6c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryFirstEnlistRequest.java
@@ -26,7 +26,6 @@ import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshotWithoutTxs;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.EnlistOperation;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -88,7 +87,6 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
* @param batchId Batch id.
* @param keys Keys.
* @param vals Values.
- * @param updCntrs Update counters.
*/
GridDhtTxQueryFirstEnlistRequest(int cacheId,
IgniteUuid dhtFutId,
@@ -103,9 +101,8 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
EnlistOperation op,
int batchId,
List<KeyCacheObject> keys,
- List<Message> vals,
- GridLongList updCntrs) {
- super(cacheId, dhtFutId, lockVer, op, batchId, snapshot.operationCounter(), keys, vals, updCntrs);
+ List<Message> vals) {
+ super(cacheId, dhtFutId, lockVer, op, batchId, snapshot.operationCounter(), keys, vals);
this.cacheId = cacheId;
this.subjId = subjId;
this.topVer = topVer;
@@ -211,55 +208,55 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
}
switch (writer.state()) {
- case 11:
+ case 10:
if (!writer.writeLong("cleanupVer", cleanupVer))
return false;
writer.incrementState();
- case 12:
+ case 11:
if (!writer.writeLong("cntr", cntr))
return false;
writer.incrementState();
- case 13:
+ case 12:
if (!writer.writeLong("crdVer", crdVer))
return false;
writer.incrementState();
- case 14:
+ case 13:
if (!writer.writeUuid("nearNodeId", nearNodeId))
return false;
writer.incrementState();
- case 15:
+ case 14:
if (!writer.writeMessage("nearXidVer", nearXidVer))
return false;
writer.incrementState();
- case 16:
+ case 15:
if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
- case 17:
+ case 16:
if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
- case 18:
+ case 17:
if (!writer.writeLong("timeout", timeout))
return false;
writer.incrementState();
- case 19:
+ case 18:
if (!writer.writeMessage("topVer", topVer))
return false;
@@ -281,7 +278,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
return false;
switch (reader.state()) {
- case 11:
+ case 10:
cleanupVer = reader.readLong("cleanupVer");
if (!reader.isLastRead())
@@ -289,7 +286,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 12:
+ case 11:
cntr = reader.readLong("cntr");
if (!reader.isLastRead())
@@ -297,7 +294,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 13:
+ case 12:
crdVer = reader.readLong("crdVer");
if (!reader.isLastRead())
@@ -305,7 +302,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 14:
+ case 13:
nearNodeId = reader.readUuid("nearNodeId");
if (!reader.isLastRead())
@@ -313,7 +310,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 15:
+ case 14:
nearXidVer = reader.readMessage("nearXidVer");
if (!reader.isLastRead())
@@ -321,7 +318,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 16:
+ case 15:
subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
@@ -329,7 +326,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 17:
+ case 16:
taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
@@ -337,7 +334,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 18:
+ case 17:
timeout = reader.readLong("timeout");
if (!reader.isLastRead())
@@ -345,7 +342,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
reader.incrementState();
- case 19:
+ case 18:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
@@ -360,7 +357,7 @@ public class GridDhtTxQueryFirstEnlistRequest extends GridDhtTxQueryEnlistReques
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 20;
+ return 19;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
index 2fdb6ac..609bc4a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
@@ -387,12 +387,11 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
* @param keys Keys.
* @param vals Values.
* @param snapshot Mvcc snapshot.
- * @param updCntrs Update counters.
* @throws IgniteCheckedException If failed.
*/
public void mvccEnlistBatch(GridCacheContext ctx, EnlistOperation op, List<KeyCacheObject> keys,
- List<Message> vals, MvccSnapshot snapshot, GridLongList updCntrs) throws IgniteCheckedException {
- assert keys != null && updCntrs != null && keys.size() == updCntrs.size();
+ List<Message> vals, MvccSnapshot snapshot) throws IgniteCheckedException {
+ assert keys != null && (vals == null || vals.size() == keys.size());
WALPointer ptr = null;
@@ -438,7 +437,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
this,
ctx.localNodeId(),
topologyVersion(),
- updCntrs.get(i),
snapshot,
false);
@@ -453,7 +451,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
val,
0,
topologyVersion(),
- updCntrs.get(i),
snapshot,
op.cacheOperation(),
false,
@@ -471,7 +468,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
updRes = entry.mvccUpdateRowsWithPreloadInfo(this,
ctx.localNodeId(),
topologyVersion(),
- updCntrs.get(i),
entries.infos(),
op.cacheOperation(),
snapshot);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java
index bb863fb..8043def 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java
@@ -82,7 +82,7 @@ public final class NearTxQueryEnlistResultHandler implements CI1<IgniteInternalF
assert future.tx.queryEnlisted() || future.cnt == 0;
return new GridNearTxQueryEnlistResponse(future.cctx.cacheId(), future.nearFutId, future.nearMiniId,
- future.nearLockVer, future.cnt, future.tx.empty() && !future.tx.queryEnlisted());
+ future.nearLockVer, future.cnt, future.tx.empty() && !future.tx.queryEnlisted(), future.newDhtNodes);
}
catch (IgniteCheckedException e) {
return new GridNearTxQueryEnlistResponse(future.cctx.cacheId(), future.nearFutId, future.nearMiniId, future.nearLockVer, e);
@@ -90,29 +90,28 @@ public final class NearTxQueryEnlistResultHandler implements CI1<IgniteInternalF
}
/**
- * @param future Enlist future.
+ * @param fut Enlist future.
* @return Enlist response.
*/
- @NotNull private static GridNearTxQueryResultsEnlistResponse createResponse(GridDhtTxQueryResultsEnlistFuture future) {
+ @NotNull private static GridNearTxQueryResultsEnlistResponse createResponse(GridDhtTxQueryResultsEnlistFuture fut) {
try {
- future.get();
+ fut.get();
GridCacheVersion ver = null;
IgniteUuid id = null;
- GridLongList updCntrs = null;
- if (future.hasNearNodeUpdates) {
- ver = future.cctx.tm().mappedVersion(future.nearLockVer);
- id = future.futId;
- updCntrs = future.nearUpdCntrs;
+ if (fut.hasNearNodeUpdates) {
+ ver = fut.cctx.tm().mappedVersion(fut.nearLockVer);
+
+ id = fut.futId;
}
- return new GridNearTxQueryResultsEnlistResponse(future.cctx.cacheId(), future.nearFutId, future.nearMiniId,
- future.nearLockVer, future.cnt, ver, id, updCntrs);
+ return new GridNearTxQueryResultsEnlistResponse(fut.cctx.cacheId(), fut.nearFutId, fut.nearMiniId,
+ fut.nearLockVer, fut.cnt, ver, id, fut.newDhtNodes);
}
catch (IgniteCheckedException e) {
- return new GridNearTxQueryResultsEnlistResponse(future.cctx.cacheId(), future.nearFutId, future.nearMiniId,
- future.nearLockVer, e);
+ return new GridNearTxQueryResultsEnlistResponse(fut.cctx.cacheId(), fut.nearFutId, fut.nearMiniId,
+ fut.nearLockVer, e);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCounters.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCounters.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCounters.java
deleted file mode 100644
index 5b1eccd..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCounters.java
+++ /dev/null
@@ -1,123 +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.distributed.dht;
-
-import java.nio.ByteBuffer;
-import java.util.Map;
-import org.apache.ignite.internal.GridDirectMap;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
-
-/**
- * Partition update counters message.
- */
-public class PartitionUpdateCounters implements Message {
- /** */
- private static final long serialVersionUID = 193442457510062844L;
-
- /** Map of update counters made by this tx. Mapping: partId -> updCntr. */
- @GridDirectMap(keyType = Integer.class, valueType = Long.class)
- private Map<Integer, Long> updCntrs;
-
- /** */
- public PartitionUpdateCounters() {
- // No-op.
- }
-
- /**
- * @param updCntrs Update counters map.
- */
- public PartitionUpdateCounters(Map<Integer, Long> updCntrs) {
- this.updCntrs = updCntrs;
- }
-
- /**
- * @return Update counters.
- */
- public Map<Integer, Long> updateCounters() {
- return updCntrs;
- }
-
- /**
- * @param updCntrs Update counters.
- */
- public void updateCounters(Map<Integer, Long> updCntrs) {
- this.updCntrs = updCntrs;
- }
-
- /** {@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.writeMap("updCntrs", updCntrs, MessageCollectionItemType.INT, MessageCollectionItemType.LONG))
- 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:
- updCntrs = reader.readMap("updCntrs", MessageCollectionItemType.INT, MessageCollectionItemType.LONG, false);
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- }
-
- return reader.afterMessageRead(PartitionUpdateCounters.class);
- }
-
- /** {@inheritDoc} */
- @Override public short directType() {
- return 157;
- }
-
- /** {@inheritDoc} */
- @Override public byte fieldsCount() {
- return 1;
- }
-
- /** {@inheritDoc} */
- @Override public void onAckReceived() {
- // No-op.
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
new file mode 100644
index 0000000..6ee0294
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionUpdateCountersMessage.java
@@ -0,0 +1,259 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * Partition update counters message.
+ */
+public class PartitionUpdateCountersMessage implements Message {
+ /** */
+ private static final int ITEM_SIZE = 4 /* partition */ + 8 /* initial counter */ + 8 /* updates count */;
+
+ /** */
+ private static final long serialVersionUID = 193442457510062844L;
+
+ /** */
+ private byte data[];
+
+ /** */
+ private int cacheId;
+
+ /** */
+ @GridDirectTransient
+ private int size;
+
+ /** */
+ public PartitionUpdateCountersMessage() {
+ // No-op.
+ }
+
+ /**
+ * @param cacheId Cache id.
+ * @param initSize Initial size.
+ */
+ public PartitionUpdateCountersMessage(int cacheId, int initSize) {
+ assert initSize >= 1;
+
+ this.cacheId = cacheId;
+ data = new byte[initSize * ITEM_SIZE];
+ }
+
+ /**
+ * @return Cache id.
+ */
+ public int cacheId() {
+ return cacheId;
+ }
+
+ /**
+ * @return Size.
+ */
+ public int size() {
+ return size;
+ }
+
+ /**
+ * @param idx Item number.
+ * @return Partition number.
+ */
+ public int partition(int idx) {
+ if (idx >= size)
+ throw new ArrayIndexOutOfBoundsException();
+
+ long off = GridUnsafe.BYTE_ARR_OFF + idx * ITEM_SIZE;
+
+ return GridUnsafe.getInt(data, off);
+ }
+
+ /**
+ * @param idx Item number.
+ * @return Partition number.
+ */
+ public long initialCounter(int idx){
+ if (idx >= size)
+ throw new ArrayIndexOutOfBoundsException();
+
+ long off = GridUnsafe.BYTE_ARR_OFF + idx * ITEM_SIZE + 4;
+
+ return GridUnsafe.getLong(data, off);
+ }
+
+ /**
+ * @param idx Item number.
+ * @param value Initial partition counter.
+ */
+ public void initialCounter(int idx, long value){
+ if (idx >= size)
+ throw new ArrayIndexOutOfBoundsException();
+
+ long off = GridUnsafe.BYTE_ARR_OFF + idx * ITEM_SIZE + 4;
+
+ GridUnsafe.putLong(data, off, value);
+ }
+
+ /**
+ * @param idx Item number.
+ * @return Update counter delta.
+ */
+ public long updatesCount(int idx){
+ if (idx >= size)
+ throw new ArrayIndexOutOfBoundsException();
+
+ long off = GridUnsafe.BYTE_ARR_OFF + idx * ITEM_SIZE + 12;
+
+ return GridUnsafe.getLong(data, off);
+ }
+
+ /**
+ * @param part Partition number.
+ * @param init Init partition counter.
+ * @param updatesCnt Update counter delta.
+ */
+ public void add(int part, long init, long updatesCnt) {
+ ensureSpace(size + 1);
+
+ long off = GridUnsafe.BYTE_ARR_OFF + size++ * ITEM_SIZE;
+
+ GridUnsafe.putInt(data, off, part); off += 4;
+ GridUnsafe.putLong(data, off, init); off += 8;
+ GridUnsafe.putLong(data, off, updatesCnt);
+ }
+
+ /**
+ * Clears message.
+ */
+ public void clear() {
+ size = 0;
+ }
+
+ /**
+ * Check if there is enough space is allocated.
+ *
+ * @param newSize Size to ensure.
+ */
+ private void ensureSpace(int newSize) {
+ int req = newSize * ITEM_SIZE;
+
+ if (data.length < req)
+ data = Arrays.copyOf(data, data.length << 1);
+ }
+
+ /** {@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.writeInt("cacheId", cacheId))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (!writer.writeByteArray("data", data, 0, size * ITEM_SIZE))
+ 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:
+ cacheId = reader.readInt("cacheId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ data = reader.readByteArray("data");
+
+ if (!reader.isLastRead())
+ return false;
+
+ size = data.length / ITEM_SIZE;
+
+ reader.incrementState();
+
+ }
+
+ return reader.afterMessageRead(PartitionUpdateCountersMessage.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 157;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ StringBuilder sb = new StringBuilder();
+
+ for (int i = 0; i < size; i++) {
+ sb.append("[part=")
+ .append(partition(i))
+ .append(", initCntr=")
+ .append(initialCounter(i))
+ .append(", cntr=")
+ .append(updatesCount(i))
+ .append(']');
+ }
+
+ return "PartitionUpdateCountersMessage{" +
+ "cacheId=" + cacheId +
+ ", size=" + size +
+ ", cntrs=" + sb +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 cea3030..4a4d8e3 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
@@ -947,6 +947,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
tx.activeCachesDeploymentEnabled(),
!waitRemoteTxs && (tx.needReturnValue() && tx.implicit()),
waitRemoteTxs,
+ null,
null);
finishReq.checkCommitted(true);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 d03d0d3..b8c78bd 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
@@ -64,6 +64,7 @@ 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.MvccQueryTrackerImpl;
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;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
@@ -1862,8 +1863,14 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
assert mvccSnapshot != null;
assert res != null;
- if (res > 0)
+ if (res > 0) {
+ if (mvccSnapshot.operationCounter() == ~MvccUtils.MVCC_OP_COUNTER_MASK) {
+ throw new IgniteCheckedException("The maximum limit of the number of statements allowed in" +
+ " one transaction is reached. [max=" + mvccSnapshot.operationCounter() + ']');
+ }
+
mvccSnapshot.incrementOperationCounter();
+ }
return res;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistResponse.java
index 39e33c4..dae1e81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistResponse.java
@@ -18,7 +18,12 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Set;
+import java.util.UUID;
+
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
@@ -27,6 +32,7 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.Nullable;
@@ -60,6 +66,10 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
/** */
private GridCacheVersion lockVer;
+ /** New DHT nodes involved into transaction. */
+ @GridDirectCollection(UUID.class)
+ private Collection<UUID> newDhtNodes;
+
/**
* Default constructor.
*/
@@ -90,13 +100,15 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
* @param res Result.
* @param removeMapping Remove mapping flag.
*/
- public GridNearTxQueryEnlistResponse(int cacheId, IgniteUuid futId, int miniId, GridCacheVersion lockVer, long res, boolean removeMapping) {
+ public GridNearTxQueryEnlistResponse(int cacheId, IgniteUuid futId, int miniId, GridCacheVersion lockVer, long res,
+ boolean removeMapping, Set<UUID> newDhtNodes) {
this.cacheId = cacheId;
this.futId = futId;
this.miniId = miniId;
this.lockVer = lockVer;
this.res = res;
this.removeMapping = removeMapping;
+ this.newDhtNodes = newDhtNodes;
}
/**
@@ -128,6 +140,13 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
}
/**
+ * @return New DHT nodes involved into transaction.
+ */
+ public Collection<UUID> newDhtNodes() {
+ return newDhtNodes;
+ }
+
+ /**
* @return Remove mapping flag.
*/
public boolean removeMapping() {
@@ -146,7 +165,7 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 9;
+ return 10;
}
/** {@inheritDoc} */
@@ -200,6 +219,11 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
writer.incrementState();
+ case 9:
+ if (!writer.writeCollection("newDhtNodes", newDhtNodes, MessageCollectionItemType.UUID))
+ return false;
+
+ writer.incrementState();
}
return true;
@@ -264,6 +288,13 @@ public class GridNearTxQueryEnlistResponse extends GridCacheIdMessage implements
reader.incrementState();
+ case 9:
+ newDhtNodes = reader.readCollection("newDhtNodes", MessageCollectionItemType.UUID);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
}
return reader.afterMessageRead(GridNearTxQueryEnlistResponse.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
index 910deb9..2452b92 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
@@ -42,7 +42,6 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.EnlistOperation;
import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
@@ -302,15 +301,13 @@ public class GridNearTxQueryResultsEnlistFuture extends GridNearTxAbstractEnlist
}
/**
- *
* @param primaryId Primary node id.
* @param rows Rows.
* @param dhtVer Dht version assigned at primary node.
* @param dhtFutId Dht future id assigned at primary node.
- * @param updCntrs Update counters.
*/
private void processBatchLocalBackupKeys(UUID primaryId, List<Object> rows, GridCacheVersion dhtVer,
- IgniteUuid dhtFutId, GridLongList updCntrs) {
+ IgniteUuid dhtFutId) {
assert dhtVer != null;
assert dhtFutId != null;
@@ -366,7 +363,7 @@ public class GridNearTxQueryResultsEnlistFuture extends GridNearTxAbstractEnlist
}
}
- dhtTx.mvccEnlistBatch(cctx, it.operation(), keys, vals, mvccSnapshot.withoutActiveTransactions(), updCntrs);
+ dhtTx.mvccEnlistBatch(cctx, it.operation(), keys, vals, mvccSnapshot.withoutActiveTransactions());
}
catch (IgniteCheckedException e) {
onDone(e);
@@ -507,8 +504,7 @@ public class GridNearTxQueryResultsEnlistFuture extends GridNearTxAbstractEnlist
Batch batch = batches.get(nodeId);
if (batch != null && !F.isEmpty(batch.localBackupRows()) && res.dhtFutureId() != null)
- processBatchLocalBackupKeys(nodeId, batch.localBackupRows(), res.dhtVersion(), res.dhtFutureId(),
- res.updateCounters());
+ processBatchLocalBackupKeys(nodeId, batch.localBackupRows(), res.dhtVersion(), res.dhtFutureId());
else
sendNextBatches(nodeId);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistResponse.java
index 749afd3..94cacfa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistResponse.java
@@ -18,8 +18,9 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
import java.nio.ByteBuffer;
+import java.util.Set;
+import java.util.UUID;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
@@ -38,9 +39,6 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
/** */
private IgniteUuid dhtFutId;
- /** */
- private GridLongList updCntrs;
-
/**
* Default-constructor.
*/
@@ -56,7 +54,7 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
* @param res Result.
* @param dhtFutId Dht future id.
* @param dhtVer Dht version.
- * @param updCntrs Update counters.
+ * @param newDhtNodes New
*/
public GridNearTxQueryResultsEnlistResponse(int cacheId,
IgniteUuid futId,
@@ -65,12 +63,11 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
long res,
GridCacheVersion dhtVer,
IgniteUuid dhtFutId,
- GridLongList updCntrs) {
- super(cacheId, futId, miniId, lockVer, res, false);
+ Set<UUID> newDhtNodes) {
+ super(cacheId, futId, miniId, lockVer, res, false, newDhtNodes);
this.dhtVer = dhtVer;
this.dhtFutId = dhtFutId;
- this.updCntrs = updCntrs;
}
/**
@@ -102,12 +99,6 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
return dhtFutId;
}
- /**
- * @return Update counters.
- */
- public GridLongList updateCounters() {
- return updCntrs;
- }
/** {@inheritDoc} */
@Override public byte fieldsCount() {
@@ -129,20 +120,14 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
}
switch (writer.state()) {
- case 9:
- if (!writer.writeIgniteUuid("dhtFutId", dhtFutId))
- return false;
-
- writer.incrementState();
-
case 10:
- if (!writer.writeMessage("dhtVer", dhtVer))
+ if (!writer.writeIgniteUuid("dhtFutId", dhtFutId))
return false;
writer.incrementState();
case 11:
- if (!writer.writeMessage("updCntrs", updCntrs))
+ if (!writer.writeMessage("dhtVer", dhtVer))
return false;
writer.incrementState();
@@ -162,16 +147,8 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
return false;
switch (reader.state()) {
- case 9:
- dhtFutId = reader.readIgniteUuid("dhtFutId");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
case 10:
- dhtVer = reader.readMessage("dhtVer");
+ dhtFutId = reader.readIgniteUuid("dhtFutId");
if (!reader.isLastRead())
return false;
@@ -179,7 +156,7 @@ public class GridNearTxQueryResultsEnlistResponse extends GridNearTxQueryEnlistR
reader.incrementState();
case 11:
- updCntrs = reader.readMessage("updCntrs");
+ dhtVer = reader.readMessage("dhtVer");
if (!reader.isLastRead())
return false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 5df9bb1..f59220b 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
@@ -118,6 +118,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDh
import static org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker.MVCC_TRACKER_ID_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_INITIAL_CNTR;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_READ_OP_CNTR;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_START_CNTR;
@@ -1084,7 +1085,8 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
}
if (workers == null) {
- U.warn(log, "Attempting to stop inactive vacuum.");
+ if (log.isInfoEnabled())
+ log.info("Attempting to stop inactive vacuum.");
return;
}
@@ -2223,9 +2225,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
return compare(row, snapshot.coordinatorVersion(), snapshot.cleanupVersion()) <= 0
&& hasNewVersion(row) && MvccUtils.compareNewVersion(row, snapshot.coordinatorVersion(), snapshot.cleanupVersion()) <= 0
&& MvccUtils.state(cctx, row.newMvccCoordinatorVersion(), row.newMvccCounter(),
- row.newMvccOperationCounter() | (row.newMvccTxState() << PageIO.MVCC_HINTS_BIT_OFF)) == TxState.COMMITTED
+ row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF)) == TxState.COMMITTED
|| MvccUtils.state(cctx, row.mvccCoordinatorVersion(), row.mvccCounter(),
- row.mvccOperationCounter() | (row.mvccTxState() << PageIO.MVCC_HINTS_BIT_OFF)) == TxState.ABORTED;
+ row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF)) == TxState.ABORTED;
}
/** */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUpdateVersionAware.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUpdateVersionAware.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUpdateVersionAware.java
index 17804c4..fee928e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUpdateVersionAware.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccUpdateVersionAware.java
@@ -73,4 +73,9 @@ public interface MvccUpdateVersionAware {
public default MvccVersion newMvccVersion() {
return new MvccVersionImpl(newMvccCoordinatorVersion(), newMvccCounter(), newMvccOperationCounter());
}
+
+ /**
+ * @return {@code True} if this key was inserted in the cache with this row in the same transaction.
+ */
+ public boolean isKeyAbsentBefore();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 c3f9a6d..896c9aa 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
@@ -44,8 +44,6 @@ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.internal.pagemem.PageIdUtils.itemId;
import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.TRANSACTION_COMPLETED;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
@@ -55,26 +53,50 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
*/
public class MvccUtils {
/** */
+ public static final int MVCC_KEY_ABSENT_BEFORE_OFF = 29;
+
+ /** */
+ public static final int MVCC_KEY_ABSENT_BEFORE_MASK = 0b0001 << MVCC_KEY_ABSENT_BEFORE_OFF;
+
+ /** */
+ public static final int MVCC_HINTS_BIT_OFF = MVCC_KEY_ABSENT_BEFORE_OFF + 1;
+
+ /** */
+ public static final int MVCC_HINTS_MASK = 0b0011 << MVCC_HINTS_BIT_OFF;
+
+ /** Mask for all masked high bits in operation counter field. */
+ public static final int MVCC_OP_COUNTER_MASK = 0b0111 << MVCC_KEY_ABSENT_BEFORE_OFF;
+
+ /** */
public static final long MVCC_CRD_COUNTER_NA = 0L;
+
/** */
public static final long MVCC_CRD_START_CNTR = 1L;
+
/** */
public static final long MVCC_COUNTER_NA = 0L;
+
/** */
public static final long MVCC_INITIAL_CNTR = 1L;
+
/** */
public static final long MVCC_START_CNTR = 3L;
+
/** */
public static final int MVCC_OP_COUNTER_NA = 0;
+
/** */
public static final int MVCC_START_OP_CNTR = 1;
+
/** */
public static final int MVCC_READ_OP_CNTR = ~MVCC_HINTS_MASK;
/** */
public static final int MVCC_INVISIBLE = 0;
+
/** */
public static final int MVCC_VISIBLE_REMOVED = 1;
+
/** */
public static final int MVCC_VISIBLE = 2;
@@ -559,13 +581,13 @@ public class MvccUtils {
long mvccCrd = dataIo.mvccCoordinator(pageAddr, offset);
long mvccCntr = dataIo.mvccCounter(pageAddr, offset);
- int mvccOpCntr = dataIo.mvccOperationCounter(pageAddr, offset);
+ int mvccOpCntr = dataIo.mvccOperationCounter(pageAddr, offset) & ~MVCC_KEY_ABSENT_BEFORE_MASK;
assert mvccVersionIsValid(mvccCrd, mvccCntr, mvccOpCntr) : mvccVersion(mvccCrd, mvccCntr, mvccOpCntr);
long newMvccCrd = dataIo.newMvccCoordinator(pageAddr, offset);
long newMvccCntr = dataIo.newMvccCounter(pageAddr, offset);
- int newMvccOpCntr = dataIo.newMvccOperationCounter(pageAddr, offset);
+ int newMvccOpCntr = dataIo.newMvccOperationCounter(pageAddr, offset) & ~MVCC_KEY_ABSENT_BEFORE_MASK;
assert newMvccCrd == MVCC_CRD_COUNTER_NA || mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr)
: mvccVersion(newMvccCrd, newMvccCntr, newMvccOpCntr);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 b8245df..574e6d5 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
@@ -661,6 +661,11 @@ public class CacheDataRowAdapter implements CacheDataRow {
return TxState.NA;
}
+ /** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return false;
+ }
+
/**
*
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 04476ad..90099cf 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
@@ -1181,6 +1181,11 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
@Override public byte newMvccTxState() {
return 0; // TODO IGNITE-7384
}
+
+ /** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return false;
+ }
}
/**
@@ -1588,11 +1593,11 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
- @Override public long nextMvccUpdateCounter() {
+ @Override public long getAndIncrementUpdateCounter(long delta) {
try {
CacheDataStore delegate0 = init0(true);
- return delegate0 == null ? 0 : delegate0.nextMvccUpdateCounter();
+ return delegate0 == null ? 0 : delegate0.getAndIncrementUpdateCounter(delta);
}
catch (IgniteCheckedException e) {
throw new IgniteException(e);
@@ -1600,11 +1605,17 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
+ @Override public void init(long size, long updCntr, @Nullable Map<Integer, Long> cacheSizes) {
+ throw new IllegalStateException("Should be never called.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void updateCounter(long val) {
try {
- CacheDataStore delegate0 = init0(true);
+ CacheDataStore delegate0 = init0(false);
- return delegate0 == null ? 0 : delegate0.mvccUpdateCounter();
+ if (delegate0 != null)
+ delegate0.updateCounter(val);
}
catch (IgniteCheckedException e) {
throw new IgniteException(e);
@@ -1612,17 +1623,12 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
- @Override public void init(long size, long updCntr, @Nullable Map<Integer, Long> cacheSizes) {
- throw new IllegalStateException("Should be never called.");
- }
-
- /** {@inheritDoc} */
- @Override public void updateCounter(long val) {
+ @Override public void updateCounter(long start, long delta) {
try {
CacheDataStore delegate0 = init0(false);
if (delegate0 != null)
- delegate0.updateCounter(val);
+ delegate0.updateCounter(start, delta);
}
catch (IgniteCheckedException e) {
throw new IgniteException(e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
index 87e37f6..95a46eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
@@ -27,6 +27,9 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridStringBuilder;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_MASK;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_KEY_ABSENT_BEFORE_OFF;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.EntryPart.CACHE_ID;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.EntryPart.EXPIRE_TIME;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.EntryPart.KEY;
@@ -71,7 +74,8 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
// xid_min.
PageUtils.putLong(addr, 0, row.mvccCoordinatorVersion());
PageUtils.putLong(addr, 8, row.mvccCounter());
- PageUtils.putInt(addr, 16, row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF));
+ PageUtils.putInt(addr, 16, row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF) |
+ ((row.isKeyAbsentBefore() ? 1 : 0) << MVCC_KEY_ABSENT_BEFORE_OFF));
assert row.newMvccCoordinatorVersion() == 0
|| MvccUtils.mvccVersionIsValid(row.newMvccCoordinatorVersion(), row.newMvccCounter(), row.newMvccOperationCounter());
@@ -79,7 +83,8 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
// xid_max.
PageUtils.putLong(addr, 20, row.newMvccCoordinatorVersion());
PageUtils.putLong(addr, 28, row.newMvccCounter());
- PageUtils.putInt(addr, 36, row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF));
+ PageUtils.putInt(addr, 36, row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF) |
+ ((row.isKeyAbsentBefore() ? 1 : 0) << MVCC_KEY_ABSENT_BEFORE_OFF));
addr += mvccInfoSize;
}
@@ -211,10 +216,12 @@ public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
writeMvccInfoFragment(buf,
row.mvccCoordinatorVersion(),
row.mvccCounter(),
- row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF),
+ row.mvccOperationCounter() | (row.mvccTxState() << MVCC_HINTS_BIT_OFF) |
+ ((row.isKeyAbsentBefore() ? 1 : 0) << MVCC_KEY_ABSENT_BEFORE_OFF),
row.newMvccCoordinatorVersion(),
row.newMvccCounter(),
- row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF),
+ row.newMvccOperationCounter() | (row.newMvccTxState() << MVCC_HINTS_BIT_OFF) |
+ ((row.isKeyAbsentBefore() ? 1 : 0) << MVCC_KEY_ABSENT_BEFORE_OFF),
len);
else if (type != VERSION) {
// Write key or value.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
index f167d0c..22d2420 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
@@ -147,12 +147,6 @@ public abstract class PageIO {
/** */
public static final int COMMON_HEADER_END = RESERVED_3_OFF + 8; // 40=type(2)+ver(2)+crc(4)+pageId(8)+rotatedIdPart(1)+reserved(1+2+4+2*8)
- /** */
- public static final int MVCC_HINTS_MASK = 0xC0000000;
-
- /** */
- public static final int MVCC_HINTS_BIT_OFF = 30;
-
/* All the page types. */
/** */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 c6d1991..357b992 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
@@ -2033,9 +2033,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
}
/**
- * Make counters accumulated during transaction visible outside of transaciton.
+ * Makes cache sizes changes accumulated during transaction visible outside of transaction.
*/
- protected void applyTxCounters() {
+ protected void applyTxSizes() {
TxCounters txCntrs = txCounters(false);
if (txCntrs == null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 32f4dd4..8655caa 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
@@ -1129,6 +1129,9 @@ public class IgniteTxHandler {
nearTx = !F.isEmpty(req.nearWrites()) ? startNearRemoteTx(ctx.deploy().globalLoader(), nodeId, req) : null;
dhtTx = startRemoteTx(nodeId, req, res);
+ if (dhtTx != null && req.updateCounters() != null) // Remember update counters on prepare state.
+ dhtTx.txCounters(true).updateCounters(req.updateCounters());
+
// Set evicted keys from near transaction.
if (nearTx != null)
res.nearEvicted(nearTx.evicted());
@@ -1372,9 +1375,14 @@ public class IgniteTxHandler {
return;
}
- else if (log.isDebugEnabled())
- log.debug("Received finish request for transaction [senderNodeId=" + nodeId + ", req=" + req +
- ", tx=" + tx + ']');
+ else {
+ if (req.updateCounters() != null)
+ tx.txCounters(true).updateCounters(req.updateCounters());
+
+ if (log.isDebugEnabled())
+ log.debug("Received finish request for transaction [senderNodeId=" + nodeId + ", req=" + req +
+ ", tx=" + tx + ']');
+ }
req.txState(tx.txState());
@@ -1391,8 +1399,6 @@ public class IgniteTxHandler {
tx.setPartitionUpdateCounters(
req.partUpdateCounters() != null ? req.partUpdateCounters().array() : null);
- tx.txCounters(true).updateCounters(req.updateCounters());
-
tx.commitRemoteTx();
}
else {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 bfe67ee..8354c8b 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
@@ -18,13 +18,15 @@
package org.apache.ignite.internal.processors.cache.transactions;
import java.io.Externalizable;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import javax.cache.expiry.Duration;
import javax.cache.expiry.ExpiryPolicy;
@@ -57,7 +59,9 @@ import org.apache.ignite.internal.processors.cache.GridCacheUpdateTxResult;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCounters;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
@@ -442,6 +446,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
try {
cctx.tm().prepareTx(this, entries);
+
+ calculatePartitionUpdateCounters();
}
catch (IgniteCheckedException e) {
throw e;
@@ -457,6 +463,64 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/**
+ * Calculates partition update counters for current transaction. Each partition will be supplied with
+ * pair (init, delta) values, where init - initial update counter, and delta - updates count made
+ * by current transaction for a given partition.
+ */
+ private void calculatePartitionUpdateCounters() {
+ TxCounters counters = txCounters(false);
+
+ if (counters != null && F.isEmpty(counters.updateCounters())) {
+ List<PartitionUpdateCountersMessage> cntrMsgs = new ArrayList<>();
+
+ for (Map.Entry<Integer, Map<Integer, AtomicLong>> record : counters.accumulatedUpdateCounters().entrySet()) {
+ int cacheId = record.getKey();
+
+ Map<Integer, AtomicLong> partToCntrs = record.getValue();
+
+ assert partToCntrs != null;
+
+ if (F.isEmpty(partToCntrs))
+ continue;
+
+ PartitionUpdateCountersMessage msg = new PartitionUpdateCountersMessage(cacheId, partToCntrs.size());
+ GridCacheContext ctx0 = cctx.cacheContext(cacheId);
+
+ assert ctx0 != null && ctx0.mvccEnabled();
+
+ GridDhtPartitionTopology top = ctx0.topology();
+
+ assert top != null;
+
+ for (Map.Entry<Integer, AtomicLong> e : partToCntrs.entrySet()) {
+ AtomicLong acc = e.getValue();
+
+ assert acc != null;
+
+ long cntr = acc.get();
+
+ assert cntr >= 0;
+
+ if (cntr != 0) {
+ int p = e.getKey();
+
+ GridDhtLocalPartition part = top.localPartition(p);
+
+ assert part != null && part.state() == GridDhtPartitionState.OWNING;
+
+ msg.add(p, part.getAndIncrementUpdateCounter(cntr), cntr);
+ }
+ }
+
+ if (msg.size() > 0)
+ cntrMsgs.add(msg);
+ }
+
+ counters.updateCounters(cntrMsgs);
+ }
+ }
+
+ /**
* Checks that locks are in proper state for commit.
*
* @param entry Cache entry to check.
@@ -913,7 +977,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
}
- applyTxCounters();
+ // Apply cache sizes only for primary nodes. Update counters were applied on prepare state.
+ applyTxSizes();
if (ptr != null && !cctx.tm().logTxRecords())
cctx.wal().flush(ptr, false);
@@ -1644,45 +1709,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/**
- * Merges mvcc update counters to the partition update counters. For mvcc transactions we update partitions
- * counters only on commit phase.
- */
- private Map<Integer, PartitionUpdateCounters> applyAndCollectLocalUpdateCounters() {
- if (F.isEmpty(txState.touchedPartitions()))
- return null;
-
- HashMap<Integer, PartitionUpdateCounters> updCntrs = new HashMap<>();
-
- for (Map.Entry<Integer, Set<Integer>> entry : txState.touchedPartitions().entrySet()) {
- Integer cacheId = entry.getKey();
-
- Set<Integer> parts = entry.getValue();
-
- assert !F.isEmpty(parts);
-
- GridCacheContext ctx0 = cctx.cacheContext(cacheId);
-
- Map<Integer, Long> partCntrs = new HashMap<>(parts.size());
-
- for (Integer p : parts) {
- GridDhtLocalPartition dhtPart = ctx0.topology().localPartition(p);
-
- assert dhtPart != null;
-
- long cntr = dhtPart.mvccUpdateCounter();
-
- dhtPart.updateCounter(cntr);
-
- partCntrs.put(p, cntr);
- }
-
- updCntrs.put(cacheId, new PartitionUpdateCounters(partCntrs));
- }
-
- return updCntrs;
- }
-
- /**
* @return {@code True} if there are entries, enlisted by query.
*/
public boolean queryEnlisted() {
@@ -1703,16 +1729,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
}
- /** {@inheritDoc} */
- @Override protected void applyTxCounters() {
- super.applyTxCounters();
-
- Map<Integer, PartitionUpdateCounters> updCntrs = applyAndCollectLocalUpdateCounters();
-
- // remember counters for subsequent sending to backups
- txCounters(true).updateCounters(updCntrs);
- }
-
/**
* Post-lock closure alias.
*
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxCounters.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxCounters.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxCounters.java
index 2ad4f94..e1a0bd6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxCounters.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxCounters.java
@@ -17,21 +17,25 @@
package org.apache.ignite.internal.processors.cache.transactions;
-import java.util.Collections;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCounters;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
/**
* Values which should be tracked during transaction execution and applied on commit.
*/
public class TxCounters {
/** Size changes for cache partitions made by transaction */
- private final ConcurrentMap<Integer, ConcurrentMap<Integer, AtomicLong>> sizeDeltas = new ConcurrentHashMap<>();
- /** Update counters for cache partitions in the end of transaction */
- private Map<Integer, PartitionUpdateCounters> updCntrs;
+ private final Map<Integer, Map<Integer, AtomicLong>> sizeDeltas = new ConcurrentHashMap<>();
+
+ /** Per-partition update counter accumulator. */
+ private final Map<Integer, Map<Integer, AtomicLong>> updCntrsAcc = new HashMap<>();
+
+ /** Final update counters for cache partitions in the end of transaction */
+ private Collection<PartitionUpdateCountersMessage> updCntrs;
/**
* Accumulates size change for cache partition.
@@ -41,42 +45,85 @@ public class TxCounters {
* @param delta Size delta.
*/
public void accumulateSizeDelta(int cacheId, int part, long delta) {
- ConcurrentMap<Integer, AtomicLong> partDeltas = sizeDeltas.get(cacheId);
-
- if (partDeltas == null) {
- ConcurrentMap<Integer, AtomicLong> partDeltas0 =
- sizeDeltas.putIfAbsent(cacheId, partDeltas = new ConcurrentHashMap<>());
-
- if (partDeltas0 != null)
- partDeltas = partDeltas0;
- }
-
- AtomicLong accDelta = partDeltas.get(part);
-
- if (accDelta == null) {
- AtomicLong accDelta0 = partDeltas.putIfAbsent(part, accDelta = new AtomicLong());
-
- if (accDelta0 != null)
- accDelta = accDelta0;
- }
+ AtomicLong accDelta = accumulator(sizeDeltas, cacheId, part);
// here AtomicLong is used more as a container,
// every instance is assumed to be accessed in thread-confined manner
accDelta.set(accDelta.get() + delta);
}
- /** */
- public void updateCounters(Map<Integer, PartitionUpdateCounters> updCntrs) {
+ /**
+ * @return Map of size changes for cache partitions made by transaction.
+ */
+ public Map<Integer, Map<Integer, AtomicLong>> sizeDeltas() {
+ return sizeDeltas;
+ }
+
+ /**
+ * @param updCntrs Final update counters.
+ */
+ public void updateCounters(Collection<PartitionUpdateCountersMessage> updCntrs) {
this.updCntrs = updCntrs;
}
- /** */
- public Map<Integer, PartitionUpdateCounters> updateCounters() {
- return updCntrs != null ? Collections.unmodifiableMap(updCntrs) : Collections.emptyMap();
+ /**
+ * @return Final update counters.
+ */
+ public Collection<PartitionUpdateCountersMessage> updateCounters() {
+ return updCntrs;
}
- /** */
- public Map<Integer, ? extends Map<Integer, AtomicLong>> sizeDeltas() {
- return Collections.unmodifiableMap(sizeDeltas);
+ /**
+ * @return Accumulated update counters.
+ */
+ public Map<Integer, Map<Integer, AtomicLong>> accumulatedUpdateCounters() {
+ return updCntrsAcc;
+ }
+
+ /**
+ * @param cacheId Cache id.
+ * @param part Partition number.
+ */
+ public void incrementUpdateCounter(int cacheId, int part) {
+ accumulator(updCntrsAcc, cacheId, part).incrementAndGet();
+ }
+
+ /**
+ * @param cacheId Cache id.
+ * @param part Partition number.
+ */
+ public void decrementUpdateCounter(int cacheId, int part) {
+ long acc = accumulator(updCntrsAcc, cacheId, part).decrementAndGet();
+
+ assert acc >= 0;
+ }
+
+ /**
+ * @param accMap Map to obtain accumulator from.
+ * @param cacheId Cache id.
+ * @param part Partition number.
+ * @return Accumulator.
+ */
+ private AtomicLong accumulator(Map<Integer, Map<Integer, AtomicLong>> accMap, int cacheId, int part) {
+ Map<Integer, AtomicLong> cacheAccs = accMap.get(cacheId);
+
+ if (cacheAccs == null) {
+ Map<Integer, AtomicLong> cacheAccs0 =
+ accMap.putIfAbsent(cacheId, cacheAccs = new ConcurrentHashMap<>());
+
+ if (cacheAccs0 != null)
+ cacheAccs = cacheAccs0;
+ }
+
+ AtomicLong acc = cacheAccs.get(part);
+
+ if (acc == null) {
+ AtomicLong accDelta0 = cacheAccs.putIfAbsent(part, acc = new AtomicLong());
+
+ if (accDelta0 != null)
+ acc = accDelta0;
+ }
+
+ return acc;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
index fc98452..9ded71f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
@@ -33,10 +33,12 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_KEY_ABSENT_BEFORE_MASK;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_KEY_ABSENT_BEFORE_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
/**
*
@@ -74,6 +76,9 @@ public class MvccDataRow extends DataRow {
@GridToStringInclude
protected byte newMvccTxState;
+ /** Flag, whether this key was absent in cache before this transaction. */
+ protected boolean keyAbsentBefore;
+
/**
* @param link Link.
*/
@@ -153,8 +158,9 @@ public class MvccDataRow extends DataRow {
int withHint = PageUtils.getInt(addr, off + 16);
- mvccOpCntr = withHint & ~MVCC_HINTS_MASK;
+ mvccOpCntr = withHint & ~MVCC_OP_COUNTER_MASK;
mvccTxState = (byte)(withHint >>> MVCC_HINTS_BIT_OFF);
+ keyAbsentBefore = ((withHint & MVCC_KEY_ABSENT_BEFORE_MASK) >>> MVCC_KEY_ABSENT_BEFORE_OFF) == 1;
assert MvccUtils.mvccVersionIsValid(mvccCrd, mvccCntr, mvccOpCntr);
@@ -164,9 +170,12 @@ public class MvccDataRow extends DataRow {
withHint = PageUtils.getInt(addr, off + 36);
- newMvccOpCntr = withHint & ~MVCC_HINTS_MASK;
+ newMvccOpCntr = withHint & ~MVCC_OP_COUNTER_MASK;
newMvccTxState = (byte)(withHint >>> MVCC_HINTS_BIT_OFF);
+ if (newMvccCrd != MVCC_CRD_COUNTER_NA)
+ keyAbsentBefore = ((withHint & MVCC_KEY_ABSENT_BEFORE_MASK) >>> MVCC_KEY_ABSENT_BEFORE_OFF) == 1;
+
assert newMvccCrd == MVCC_CRD_COUNTER_NA || MvccUtils.mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr);
return MVCC_INFO_SIZE;
@@ -247,6 +256,11 @@ public class MvccDataRow extends DataRow {
}
/** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return keyAbsentBefore;
+ }
+
+ /** {@inheritDoc} */
@Override public int size() throws IgniteCheckedException {
return super.size() + MVCC_INFO_SIZE;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
index 81b1b58..716094e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
@@ -41,14 +41,14 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compare;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isActive;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isVisible;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccVersionIsValid;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.unexpectedStateException;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
/**
*
@@ -56,16 +56,22 @@ import static org.apache.ignite.internal.processors.cache.persistence.tree.io.Pa
public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult, BPlusTree.TreeVisitorClosure<CacheSearchRow, CacheDataRow> {
/** */
private static final int FIRST = DIRTY << 1;
+
/** */
private static final int CHECK_VERSION = FIRST << 1;
+
/** */
private static final int LAST_COMMITTED_FOUND = CHECK_VERSION << 1;
+
/** */
private static final int CAN_CLEANUP = LAST_COMMITTED_FOUND << 1;
+
/** */
private static final int PRIMARY = CAN_CLEANUP << 1;
+
/** */
private static final int REMOVE_OR_LOCK = PRIMARY << 1;
+
/** */
private static final int NEED_HISTORY = REMOVE_OR_LOCK << 1;
/**
@@ -78,11 +84,16 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
* but more versions should be checked.
*/
private static final int FAST_UPDATE = NEED_HISTORY << 1;
+
/** */
private static final int FAST_MISMATCH = FAST_UPDATE << 1;
+
/** */
private static final int DELETED = FAST_MISMATCH << 1;
+ /** Whether tx has overridden it's own update. */
+ private static final int OWN_VALUE_OVERRIDDEN = DELETED << 1;
+
/** */
@GridToStringExclude
private final GridCacheContext cctx;
@@ -152,6 +163,7 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
this.mvccSnapshot = mvccSnapshot;
this.cctx = cctx;
+ this.keyAbsentBefore = primary; // True for primary and false for backup (backups do not use this flag).
assert !lockOnly || val == null;
@@ -228,7 +240,11 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
else
oldRow = row;
- setFlags(LAST_COMMITTED_FOUND);
+ setFlags(LAST_COMMITTED_FOUND | OWN_VALUE_OVERRIDDEN);
+
+ // Copy new key flag from the previous row version if it was created by the current tx.
+ if (isFlagsSet(PRIMARY))
+ keyAbsentBefore = row.isKeyAbsentBefore();
}
}
@@ -238,13 +254,13 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
long rowCntr = row.mvccCounter();
// with hint bits
- int rowOpCntr = (row.mvccTxState() << MVCC_HINTS_BIT_OFF) | (row.mvccOperationCounter() & ~MVCC_HINTS_MASK);
+ int rowOpCntr = (row.mvccTxState() << MVCC_HINTS_BIT_OFF) | (row.mvccOperationCounter() & ~MVCC_OP_COUNTER_MASK);
long rowNewCrd = row.newMvccCoordinatorVersion();
long rowNewCntr = row.newMvccCounter();
// with hint bits
- int rowNewOpCntr = (row.newMvccTxState() << MVCC_HINTS_BIT_OFF) | (row.newMvccOperationCounter() & ~MVCC_HINTS_MASK);
+ int rowNewOpCntr = (row.newMvccTxState() << MVCC_HINTS_BIT_OFF) | (row.newMvccOperationCounter() & ~MVCC_OP_COUNTER_MASK);
// Search for youngest committed by another transaction row.
if (!isFlagsSet(LAST_COMMITTED_FOUND)) {
@@ -278,6 +294,8 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
res = ResultType.PREV_NOT_NULL;
oldRow = row;
+
+ keyAbsentBefore = false;
}
if (isFlagsSet(CHECK_VERSION)) {
@@ -380,7 +398,7 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
if (cleanupRows == null)
cleanupRows = new ArrayList<>();
- cleanupRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_HINTS_MASK, rowLink));
+ cleanupRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_OP_COUNTER_MASK, rowLink));
}
else {
// Row obsoleted by current operation, all rows created or updated with current tx.
@@ -391,7 +409,7 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
if (historyRows == null)
historyRows = new ArrayList<>();
- historyRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_HINTS_MASK, rowLink));
+ historyRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_OP_COUNTER_MASK, rowLink));
}
if (cleanupVer > MVCC_OP_COUNTER_NA // Do not clean if cleanup version is not assigned.
@@ -463,6 +481,11 @@ public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult,
return historyRows;
}
+ /** {@inheritDoc} */
+ @Override public boolean isOwnValueOverridden() {
+ return isFlagsSet(OWN_VALUE_OVERRIDDEN);
+ }
+
/** */
private boolean isFlagsSet(int flags) {
return (state & flags) == flags;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
index 086a30d..d76a6e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
@@ -40,4 +40,14 @@ public interface MvccUpdateResult {
* @return Collection of row created or affected by the current tx.
*/
public List<MvccLinkAwareSearchRow> history();
+
+ /**
+ * @return {@code True} if this key was inserted in the cache with this row in the same transaction.
+ */
+ public boolean isKeyAbsentBefore();
+
+ /**
+ * @return Flag whether tx has overridden it's own update.
+ */
+ public boolean isOwnValueOverridden();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 8bc620f..9bbf03d 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
@@ -482,7 +482,7 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
/** {@inheritDoc} */
@Override public GridCacheUpdateTxResult mvccSet(@Nullable IgniteInternalTx tx, UUID affNodeId, CacheObject val,
- long ttl0, AffinityTopologyVersion topVer, @Nullable Long updateCntr, MvccSnapshot mvccVer,
+ long ttl0, AffinityTopologyVersion topVer, MvccSnapshot mvccVer,
GridCacheOperation op, boolean needHistory,
boolean noCreate) throws IgniteCheckedException, GridCacheEntryRemovedException {
rawPut(val, ttl);
@@ -492,7 +492,7 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
/** {@inheritDoc} */
@Override public GridCacheUpdateTxResult mvccRemove(@Nullable IgniteInternalTx tx, UUID affNodeId,
- AffinityTopologyVersion topVer, @Nullable Long updateCntr, MvccSnapshot mvccVer, boolean needHistory)
+ AffinityTopologyVersion topVer, MvccSnapshot mvccVer, boolean needHistory)
throws IgniteCheckedException, GridCacheEntryRemovedException {
obsoleteVer = ver;
@@ -943,7 +943,6 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
@Override public GridCacheUpdateTxResult mvccUpdateRowsWithPreloadInfo(IgniteInternalTx tx,
UUID affNodeId,
AffinityTopologyVersion topVer,
- @Nullable Long updateCntr,
List<GridCacheEntryInfo> entries,
GridCacheOperation op,
MvccSnapshot mvccVer) throws IgniteCheckedException, GridCacheEntryRemovedException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccConfigurationValidationTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccConfigurationValidationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccConfigurationValidationTest.java
index a10c6c7..8b70468 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccConfigurationValidationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccConfigurationValidationTest.java
@@ -286,7 +286,7 @@ public class CacheMvccConfigurationValidationTest extends GridCommonAbstractTest
return null;
}
- }, IgniteCheckedException.class, "Failed to start processor");
+ }, IgniteCheckedException.class, "Cannot start cache. Statically configured atomicity mode differs from");
}
/**
@@ -332,7 +332,7 @@ public class CacheMvccConfigurationValidationTest extends GridCommonAbstractTest
return null;
}
- }, IgniteCheckedException.class, "Failed to start processor");
+ }, IgniteCheckedException.class, "Cannot start cache. Statically configured atomicity mode");
}
/**
[3/3] ignite git commit: IGNITE-9538: MVCC: update partition counters
on prepare phase. This closes #4795.
Posted by vo...@apache.org.
IGNITE-9538: MVCC: update partition counters on prepare phase. This closes #4795.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3a82d4aa
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3a82d4aa
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3a82d4aa
Branch: refs/heads/master
Commit: 3a82d4aa2cae8aa88feeb8bf9505fec41b122417
Parents: eff5751
Author: rkondakov <ko...@mail.ru>
Authored: Tue Sep 25 16:55:43 2018 +0300
Committer: devozerov <vo...@gridgain.com>
Committed: Tue Sep 25 16:55:43 2018 +0300
----------------------------------------------------------------------
.../communication/GridIoMessageFactory.java | 5 +-
.../processors/cache/GridCacheEntryEx.java | 6 -
.../processors/cache/GridCacheMapEntry.java | 132 ++---
.../cache/GridCacheMvccEntryInfo.java | 14 +-
.../cache/GridCacheSharedContext.java | 1 -
.../cache/IgniteCacheOffheapManager.java | 18 +-
.../cache/IgniteCacheOffheapManagerImpl.java | 70 +--
.../cache/PartitionUpdateCounter.java | 209 +++++++
.../GridDistributedTxRemoteAdapter.java | 81 +--
.../distributed/dht/GridDhtCacheEntry.java | 5 -
.../distributed/dht/GridDhtLocalPartition.java | 38 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../dht/GridDhtTxAbstractEnlistFuture.java | 62 +--
.../distributed/dht/GridDhtTxFinishFuture.java | 13 +-
.../distributed/dht/GridDhtTxFinishRequest.java | 31 +-
.../distributed/dht/GridDhtTxLocalAdapter.java | 39 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 8 +-
.../dht/GridDhtTxPrepareRequest.java | 37 +-
.../dht/GridDhtTxQueryEnlistRequest.java | 32 +-
.../dht/GridDhtTxQueryFirstEnlistRequest.java | 45 +-
.../cache/distributed/dht/GridDhtTxRemote.java | 8 +-
.../dht/NearTxQueryEnlistResultHandler.java | 25 +-
.../dht/PartitionUpdateCounters.java | 123 -----
.../dht/PartitionUpdateCountersMessage.java | 259 +++++++++
.../near/GridNearTxFinishFuture.java | 1 +
.../cache/distributed/near/GridNearTxLocal.java | 9 +-
.../near/GridNearTxQueryEnlistResponse.java | 35 +-
.../GridNearTxQueryResultsEnlistFuture.java | 10 +-
.../GridNearTxQueryResultsEnlistResponse.java | 41 +-
.../cache/mvcc/MvccProcessorImpl.java | 8 +-
.../cache/mvcc/MvccUpdateVersionAware.java | 5 +
.../processors/cache/mvcc/MvccUtils.java | 30 +-
.../cache/persistence/CacheDataRowAdapter.java | 5 +
.../persistence/GridCacheOffheapManager.java | 30 +-
.../cache/persistence/tree/io/DataPageIO.java | 15 +-
.../cache/persistence/tree/io/PageIO.java | 6 -
.../cache/transactions/IgniteTxAdapter.java | 4 +-
.../cache/transactions/IgniteTxHandler.java | 16 +-
.../transactions/IgniteTxLocalAdapter.java | 120 ++--
.../cache/transactions/TxCounters.java | 111 ++--
.../cache/tree/mvcc/data/MvccDataRow.java | 22 +-
.../cache/tree/mvcc/data/MvccUpdateDataRow.java | 37 +-
.../cache/tree/mvcc/data/MvccUpdateResult.java | 10 +
.../processors/cache/GridCacheTestEntryEx.java | 5 +-
.../CacheMvccConfigurationValidationTest.java | 4 +-
.../database/CacheFreeListImplSelfTest.java | 5 +
.../processors/query/h2/opt/GridH2Row.java | 4 +
.../mvcc/CacheMvccSqlUpdateCountersTest.java | 549 +++++++++++++++++++
.../testsuites/IgniteCacheMvccSqlTestSuite.java | 2 +
49 files changed, 1698 insertions(+), 649 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 04f6b5f..41c75be 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
@@ -79,7 +79,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxQuer
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxQueryEnlistResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxQueryFirstEnlistRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnlockRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCounters;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicDeferredUpdateResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicNearResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateRequest;
@@ -103,7 +103,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.LatchAckMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
@@ -1058,7 +1057,7 @@ public class GridIoMessageFactory implements MessageFactory {
break;
case 157:
- msg = new PartitionUpdateCounters();
+ msg = new PartitionUpdateCountersMessage();
break;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 3a2af5d..1f38eb7 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
@@ -348,7 +348,6 @@ public interface GridCacheEntryEx {
* @param val Value to set.
* @param ttl0 TTL.
* @param topVer Topology version.
- * @param updateCntr Update counter.
* @param mvccVer Mvcc version.
* @param op Cache operation.
* @param needHistory Whether to collect rows created or affected by the current tx.
@@ -364,7 +363,6 @@ public interface GridCacheEntryEx {
CacheObject val,
long ttl0,
AffinityTopologyVersion topVer,
- @Nullable Long updateCntr,
MvccSnapshot mvccVer,
GridCacheOperation op,
boolean needHistory,
@@ -374,7 +372,6 @@ public interface GridCacheEntryEx {
* @param tx Cache transaction.
* @param affNodeId Partitioned node iD.
* @param topVer Topology version.
- * @param updateCntr Update counter.
* @param mvccVer Mvcc version.
* @param needHistory Whether to collect rows created or affected by the current tx.
* @return Tuple containing success flag and old value. If success is {@code false},
@@ -386,7 +383,6 @@ public interface GridCacheEntryEx {
@Nullable IgniteInternalTx tx,
UUID affNodeId,
AffinityTopologyVersion topVer,
- @Nullable Long updateCntr,
MvccSnapshot mvccVer,
boolean needHistory) throws IgniteCheckedException, GridCacheEntryRemovedException;
@@ -1157,7 +1153,6 @@ public interface GridCacheEntryEx {
* @param tx Transaction.
* @param affNodeId Affinity node id.
* @param topVer Topology version.
- * @param updateCntr Update counter.
* @param op Cache operation.
* @param mvccVer Mvcc version. @return Update result.
* @throws IgniteCheckedException, If failed.
@@ -1167,7 +1162,6 @@ public interface GridCacheEntryEx {
IgniteInternalTx tx,
UUID affNodeId,
AffinityTopologyVersion topVer,
- Long updateCntr,
List<GridCacheEntryInfo> entries,
GridCacheOperation op,
MvccSnapshot mvccVer)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 fd8b2cd..cdb5656 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
@@ -39,7 +39,6 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.UnregisteredBinaryTypeException;
import org.apache.ignite.internal.UnregisteredClassException;
-import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
@@ -61,11 +60,13 @@ import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
+import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalAdapter;
+import org.apache.ignite.internal.processors.cache.transactions.TxCounters;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateResult;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.ResultType;
import org.apache.ignite.internal.processors.cache.version.GridCacheLazyPlainVersionedEntry;
@@ -111,11 +112,11 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.DEL
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE;
+import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.INVOKE_NO_OP;
+import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.REMOVE_NO_VAL;
import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.NO_KEY;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.CONCURRENT_UPDATE;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.DUPLICATE_KEY;
-import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.INVOKE_NO_OP;
-import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.REMOVE_NO_VAL;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_BACKUP;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_PRIMARY;
@@ -1042,7 +1043,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
CacheObject val,
long ttl0,
AffinityTopologyVersion topVer,
- @Nullable Long updateCntr,
MvccSnapshot mvccVer,
GridCacheOperation op,
boolean needHistory,
@@ -1111,7 +1111,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
IgniteInternalFuture<?> lockFut = cctx.kernalContext().coordinators().waitFor(cctx, lockVer);
- lockFut.listen(new MvccUpdateLockListener(tx, this, affNodeId, topVer, val, ttl0, updateCntr, mvccVer,
+ lockFut.listen(new MvccUpdateLockListener(tx, this, affNodeId, topVer, val, ttl0, mvccVer,
op, needHistory, noCreate, resFut));
return new GridCacheUpdateTxResult(false, resFut);
@@ -1123,13 +1123,23 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (cctx.deferredDelete() && deletedUnlocked() && !detached())
deletedUnlocked(false);
- assert tx.local() && updateCntr == null || !tx.local() && updateCntr != null && updateCntr > 0;
+ if (res.resultType() == ResultType.PREV_NULL) {
+ TxCounters counters = tx.txCounters(true);
- if (tx.local())
- updateCntr = nextMvccPartitionCounter();
+ if (res.isOwnValueOverridden()) {
+ if (res.isKeyAbsentBefore())
+ counters.incrementUpdateCounter(cctx.cacheId(), partition());
+ }
+ else
+ counters.incrementUpdateCounter(cctx.cacheId(), partition());
+
+ counters.accumulateSizeDelta(cctx.cacheId(), partition(), 1);
+ }
+ else if (res.resultType() == ResultType.PREV_NOT_NULL && !res.isOwnValueOverridden()) {
+ TxCounters counters = tx.txCounters(true);
- if (res.resultType() == ResultType.PREV_NULL)
- tx.txCounters(true).accumulateSizeDelta(cctx.cacheId(), partition(), 1);
+ counters.incrementUpdateCounter(cctx.cacheId(), partition());
+ }
if (cctx.group().persistenceEnabled() && cctx.group().walEnabled())
logPtr = cctx.shared().wal().log(new DataRecord(new DataEntry(
@@ -1141,7 +1151,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
newVer,
expireTime,
key.partition(),
- updateCntr)));
+ 0L)));
update(val, expireTime, ttl, newVer, true);
@@ -1157,9 +1167,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- onUpdateFinished(updateCntr);
+ onUpdateFinished(0L);
- GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, updateCntr, logPtr) :
+ GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, 0L, logPtr) :
new GridCacheUpdateTxResult(false, logPtr);
updRes.mvccHistory(res.history());
@@ -1172,7 +1182,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
IgniteInternalTx tx,
UUID affNodeId,
AffinityTopologyVersion topVer,
- @Nullable Long updateCntr,
MvccSnapshot mvccVer,
boolean needHistory) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert tx != null;
@@ -1212,7 +1221,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
IgniteInternalFuture<?> lockFut = cctx.kernalContext().coordinators().waitFor(cctx, lockVer);
- lockFut.listen(new MvccRemoveLockListener(tx, this, affNodeId, topVer, updateCntr, mvccVer, needHistory,
+ lockFut.listen(new MvccRemoveLockListener(tx, this, affNodeId, topVer, mvccVer, needHistory,
resFut));
return new GridCacheUpdateTxResult(false, resFut);
@@ -1221,16 +1230,21 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (cctx.deferredDelete() && deletedUnlocked() && !detached())
deletedUnlocked(false);
- assert tx.local() && updateCntr == null || !tx.local() && updateCntr != null && updateCntr > 0;
+ if (res.resultType() == ResultType.PREV_NOT_NULL) {
+ TxCounters counters = tx.txCounters(true);
- if (tx.local())
- updateCntr = nextMvccPartitionCounter();
+ if (res.isOwnValueOverridden()) {
+ if (res.isKeyAbsentBefore()) // Do not count own update removal.
+ counters.decrementUpdateCounter(cctx.cacheId(), partition());
+ }
+ else
+ counters.incrementUpdateCounter(cctx.cacheId(), partition());
- if (res.resultType() == ResultType.PREV_NOT_NULL)
- tx.txCounters(true).accumulateSizeDelta(cctx.cacheId(), partition(), -1);
+ counters.accumulateSizeDelta(cctx.cacheId(), partition(), -1);
+ }
if (cctx.group().persistenceEnabled() && cctx.group().walEnabled())
- logPtr = logTxUpdate(tx, null, 0, updateCntr);
+ logPtr = logTxUpdate(tx, null, 0, 0L);
update(null, 0, 0, newVer, true);
@@ -1246,9 +1260,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- onUpdateFinished(updateCntr);
+ onUpdateFinished(0L);
- GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, updateCntr, logPtr) :
+ GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, 0L, logPtr) :
new GridCacheUpdateTxResult(false, logPtr);
updRes.mvccHistory(res.history());
@@ -3420,14 +3434,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
return 0;
}
- /**
- * @return Next mvcc update counter.
- */
- protected long nextMvccPartitionCounter() {
- return 0;
- }
-
-
/** {@inheritDoc} */
@Override public GridCacheVersionedEntryEx versionedEntry(final boolean keepBinary)
throws IgniteCheckedException, GridCacheEntryRemovedException {
@@ -4929,9 +4935,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
private final MvccSnapshot mvccVer;
/** */
- private final Long updateCntr;
-
- /** */
private final boolean needHistory;
/** */
@@ -4945,7 +4948,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
GridCacheMapEntry entry,
UUID affNodeId,
AffinityTopologyVersion topVer,
- Long updateCntr,
MvccSnapshot mvccVer,
boolean needHistory,
GridFutureAdapter<GridCacheUpdateTxResult> resFut) {
@@ -4954,7 +4956,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
this.topVer = topVer;
this.affNodeId = affNodeId;
this.mvccVer = mvccVer;
- this.updateCntr = updateCntr;
this.needHistory = needHistory;
this.resFut = resFut;
}
@@ -4962,7 +4963,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** {@inheritDoc} */
@Override public void apply(IgniteInternalFuture lockFut) {
WALPointer logPtr = null;
- long updateCntr0;
boolean valid;
GridCacheContext cctx = entry.context();
@@ -5019,15 +5019,18 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (cctx.deferredDelete() && entry.deletedUnlocked() && !entry.detached())
entry.deletedUnlocked(false);
- assert tx.local() && updateCntr == null || !tx.local() && updateCntr != null && updateCntr > 0;
+ if (res.resultType() == ResultType.PREV_NOT_NULL) {
+ TxCounters counters = tx.txCounters(true);
- updateCntr0 = tx.local() ? entry.nextMvccPartitionCounter() : updateCntr;
-
- if (updateCntr != null && updateCntr != 0)
- updateCntr0 = updateCntr;
+ if (res.isOwnValueOverridden()) {
+ if (res.isKeyAbsentBefore()) // Do not count own update removal.
+ counters.decrementUpdateCounter(cctx.cacheId(), entry.partition());
+ }
+ else
+ counters.incrementUpdateCounter(cctx.cacheId(), entry.partition());
- if (res.resultType() == ResultType.PREV_NOT_NULL)
- tx.txCounters(true).accumulateSizeDelta(cctx.cacheId(), entry.partition(), -1);
+ counters.accumulateSizeDelta(cctx.cacheId(), entry.partition(), -1);
+ }
if (cctx.group().persistenceEnabled() && cctx.group().walEnabled())
logPtr = cctx.shared().wal().log(new DataRecord(new DataEntry(
@@ -5039,7 +5042,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
tx.writeVersion(),
0,
entry.key().partition(),
- updateCntr0)));
+ 0)));
entry.update(null, 0, 0, newVer, true);
@@ -5060,9 +5063,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- entry.onUpdateFinished(updateCntr0);
+ entry.onUpdateFinished(0L);
- GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, updateCntr0, logPtr)
+ GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, 0L, logPtr)
: new GridCacheUpdateTxResult(false, logPtr);
updRes.mvccHistory(res.history());
@@ -5189,9 +5192,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
private final long ttl;
/** */
- private final Long updateCntr;
-
- /** */
private final MvccSnapshot mvccVer;
/** */
@@ -5216,7 +5216,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
AffinityTopologyVersion topVer,
CacheObject val,
long ttl,
- Long updateCntr,
MvccSnapshot mvccVer,
GridCacheOperation op,
boolean needHistory,
@@ -5228,7 +5227,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
this.topVer = topVer;
this.val = val;
this.ttl = ttl;
- this.updateCntr = updateCntr;
this.mvccVer = mvccVer;
this.op = op;
this.needHistory = needHistory;
@@ -5239,7 +5237,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** {@inheritDoc} */
@Override public void apply(IgniteInternalFuture lockFut) {
WALPointer logPtr = null;
- long updateCntr0;
boolean valid;
GridCacheContext cctx = entry.context();
@@ -5312,12 +5309,23 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (cctx.deferredDelete() && entry.deletedUnlocked() && !entry.detached())
entry.deletedUnlocked(false);
- assert tx.local() && updateCntr == null || !tx.local() && updateCntr != null && updateCntr > 0;
+ if (res.resultType() == ResultType.PREV_NULL) {
+ TxCounters counters = tx.txCounters(true);
+
+ if (res.isOwnValueOverridden()) {
+ if (res.isKeyAbsentBefore())
+ counters.incrementUpdateCounter(cctx.cacheId(), entry.partition());
+ }
+ else
+ counters.incrementUpdateCounter(cctx.cacheId(), entry.partition());
- updateCntr0 = tx.local() ? entry.nextMvccPartitionCounter() : updateCntr;
+ counters.accumulateSizeDelta(cctx.cacheId(), entry.partition(), 1);
+ }
+ else if (res.resultType() == ResultType.PREV_NOT_NULL && !res.isOwnValueOverridden()) {
+ TxCounters counters = tx.txCounters(true);
- if (res.resultType() == ResultType.PREV_NULL)
- tx.txCounters(true).accumulateSizeDelta(cctx.cacheId(), entry.partition(), 1);
+ counters.incrementUpdateCounter(cctx.cacheId(), entry.partition());
+ }
if (cctx.group().persistenceEnabled() && cctx.group().walEnabled())
logPtr = cctx.shared().wal().log(new DataRecord(new DataEntry(
@@ -5329,7 +5337,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
newVer,
expireTime,
entry.key().partition(),
- updateCntr0)));
+ 0L)));
entry.update(val, expireTime, ttl, newVer, true);
@@ -5350,9 +5358,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- entry.onUpdateFinished(updateCntr0);
+ entry.onUpdateFinished(0L);
- GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, updateCntr0, logPtr)
+ GridCacheUpdateTxResult updRes = valid ? new GridCacheUpdateTxResult(true, 0L, logPtr)
: new GridCacheUpdateTxResult(false, logPtr);
updRes.mvccHistory(res.history());
@@ -6451,12 +6459,10 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
IgniteInternalTx tx,
UUID affNodeId,
AffinityTopologyVersion topVer,
- Long updateCntr,
List<GridCacheEntryInfo> entries,
GridCacheOperation op,
MvccSnapshot mvccVer)
throws IgniteCheckedException, GridCacheEntryRemovedException {
- assert updateCntr != null && updateCntr > 0 && !tx.local();
WALPointer logPtr = null;
@@ -6506,7 +6512,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
ver,
CU.EXPIRE_TIME_ETERNAL,
key.partition(),
- updateCntr)));
+ 0L)));
update(val, expireTime, ttl, ver, true);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
index a80ddc0..bd4146e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccEntryInfo.java
@@ -24,8 +24,9 @@ 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;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_MASK;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
/**
*
@@ -64,7 +65,7 @@ public class GridCacheMvccEntryInfo extends GridCacheEntryInfo implements MvccVe
/** {@inheritDoc} */
@Override public int newMvccOperationCounter() {
- return newMvccOpCntr & ~MVCC_HINTS_MASK;
+ return newMvccOpCntr & ~MVCC_OP_COUNTER_MASK;
}
/** {@inheritDoc} */
@@ -84,7 +85,7 @@ public class GridCacheMvccEntryInfo extends GridCacheEntryInfo implements MvccVe
/** {@inheritDoc} */
@Override public int mvccOperationCounter() {
- return mvccOpCntr & ~MVCC_HINTS_MASK;
+ return mvccOpCntr & ~MVCC_OP_COUNTER_MASK;
}
/** {@inheritDoc} */
@@ -121,6 +122,11 @@ public class GridCacheMvccEntryInfo extends GridCacheEntryInfo implements MvccVe
}
/** {@inheritDoc} */
+ @Override public boolean isKeyAbsentBefore() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
writer.setBuffer(buf);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index bfe0001..a4dddc1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -43,7 +43,6 @@ import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.jta.CacheJtaManagerAdapter;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
index fdf42fe..c30f9e6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
@@ -638,19 +638,25 @@ public interface IgniteCacheOffheapManager {
void updateCounter(long val);
/**
- * @return Next update counter.
+ * Updates counters from start value by delta value.
+ *
+ * @param start Start.
+ * @param delta Delta
*/
- public long nextUpdateCounter();
+ void updateCounter(long start, long delta);
/**
- * @return Next mvcc update counter.
+ * @return Next update counter.
*/
- long nextMvccUpdateCounter();
+ public long nextUpdateCounter();
/**
- * @return Current mvcc update counter value.
+ * Returns current value and updates counter by delta.
+ *
+ * @param delta Delta.
+ * @return Current value.
*/
- long mvccUpdateCounter();
+ public long getAndIncrementUpdateCounter(long delta);
/**
* @return Initial update counter.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 9919240..dd4b452 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
@@ -110,6 +110,8 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDh
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.INITIAL_VERSION;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compare;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compareNewVersion;
@@ -119,8 +121,6 @@ import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.state;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.unexpectedStateException;
import static org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager.EMPTY_CURSOR;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
-import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
import static org.apache.ignite.internal.util.IgniteTree.OperationType.NOOP;
import static org.apache.ignite.internal.util.IgniteTree.OperationType.PUT;
@@ -1401,14 +1401,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
private final CacheDataTree dataTree;
/** Update counter. */
- protected final AtomicLong cntr = new AtomicLong();
-
- /**
- * Mvcc update counter. This counter is used for an mvcc-style entries updates where this counter is
- * incremented on each entry write (which happens before commit), but main update counter is updated
- * on commit phase only.
- */
- protected final AtomicLong mvccUpdCntr = new AtomicLong();
+ protected final PartitionUpdateCounter pCntr = new PartitionUpdateCounter(log);
/** Partition size. */
private final AtomicLong storageSize = new AtomicLong();
@@ -1416,9 +1409,6 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** */
private final ConcurrentMap<Integer, AtomicLong> cacheSizes = new ConcurrentHashMap<>();
- /** Initial update counter. */
- protected long initCntr;
-
/** Mvcc remove handler. */
private final PageHandler<MvccVersion, Boolean> mvccUpdateMarker = new MvccMarkUpdatedHandler();
@@ -1514,35 +1504,37 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** {@inheritDoc} */
@Override public long nextUpdateCounter() {
- return cntr.incrementAndGet();
+ return pCntr.next();
}
/** {@inheritDoc} */
- @Override public long updateCounter() {
- return cntr.get();
+ @Override public long initialUpdateCounter() {
+ return pCntr.initial();
}
/** {@inheritDoc} */
- @Override public void updateCounter(long val) {
- while (true) {
- long val0 = cntr.get();
+ @Override public void updateInitialCounter(long cntr) {
+ pCntr.updateInitial(cntr);
+ }
- if (val0 >= val)
- break;
+ /** {@inheritDoc} */
+ @Override public long getAndIncrementUpdateCounter(long delta) {
+ return pCntr.getAndAdd(delta);
+ }
- if (cntr.compareAndSet(val0, val))
- break;
- }
+ /** {@inheritDoc} */
+ @Override public long updateCounter() {
+ return pCntr.get();
}
/** {@inheritDoc} */
- @Override public long nextMvccUpdateCounter() {
- return mvccUpdCntr.incrementAndGet();
+ @Override public void updateCounter(long val) {
+ pCntr.update(val);
}
/** {@inheritDoc} */
- @Override public long mvccUpdateCounter() {
- return mvccUpdCntr.get();
+ @Override public void updateCounter(long start, long delta) {
+ pCntr.update(start, delta);
}
/** {@inheritDoc} */
@@ -2771,29 +2763,15 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public long initialUpdateCounter() {
- return initCntr;
- }
-
- /** {@inheritDoc} */
- @Override public void updateInitialCounter(long cntr) {
- if (updateCounter() < cntr)
- updateCounter(cntr);
-
- initCntr = cntr;
- }
-
- /** {@inheritDoc} */
@Override public void setRowCacheCleaner(GridQueryRowCacheCleaner rowCacheCleaner) {
rowStore().setRowCacheCleaner(rowCacheCleaner);
}
/** {@inheritDoc} */
@Override public void init(long size, long updCntr, @Nullable Map<Integer, Long> cacheSizes) {
- initCntr = updCntr;
- storageSize.set(size);
+ pCntr.init(updCntr);
- cntr.set(updCntr);
+ storageSize.set(size);
if (cacheSizes != null) {
for (Map.Entry<Integer, Long> e : cacheSizes.entrySet())
@@ -3143,13 +3121,13 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
long crd = iox.mvccCoordinator(pageAddr, offset);
long cntr = iox.mvccCounter(pageAddr, offset);
int opCntrAndHint = iox.mvccOperationCounter(pageAddr, offset);
- int opCntr = opCntrAndHint & ~MVCC_HINTS_MASK;
+ int opCntr = opCntrAndHint & ~MVCC_OP_COUNTER_MASK;
byte txState = (byte)(opCntrAndHint >>> MVCC_HINTS_BIT_OFF);
long newCrd = iox.newMvccCoordinator(pageAddr, offset);
long newCntr = iox.newMvccCounter(pageAddr, offset);
int newOpCntrAndHint = iox.newMvccOperationCounter(pageAddr, offset);
- int newOpCntr = newOpCntrAndHint & ~MVCC_HINTS_MASK;
+ int newOpCntr = newOpCntrAndHint & ~MVCC_OP_COUNTER_MASK;
byte newTxState = (byte)(newOpCntrAndHint >>> MVCC_HINTS_BIT_OFF);
assert crd == newRow.mvccCoordinatorVersion();
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounter.java
new file mode 100644
index 0000000..b5960ab
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounter.java
@@ -0,0 +1,209 @@
+/*
+ * 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;
+
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.IgniteLogger;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Partition update counter with MVCC delta updates capabilities.
+ */
+public class PartitionUpdateCounter {
+ /** */
+ private IgniteLogger log;
+
+ /** Queue of counter update tasks*/
+ private final Queue<Item> queue = new PriorityQueue<>();
+
+ /** Counter. */
+ private final AtomicLong cntr = new AtomicLong();
+
+ /** Initial counter. */
+ private long initCntr;
+
+ /**
+ * @param log Logger.
+ */
+ PartitionUpdateCounter(IgniteLogger log) {
+ this.log = log;
+ }
+
+ /**
+ * Sets init counter.
+ *
+ * @param updateCntr Init counter valus.
+ */
+ public void init(long updateCntr) {
+ initCntr = updateCntr;
+
+ cntr.set(updateCntr);
+ }
+
+ /**
+ * @return Initial counter value.
+ */
+ public long initial() {
+ return initCntr;
+ }
+
+ /**
+ * @return Current update counter value.
+ */
+ public long get() {
+ return cntr.get();
+ }
+
+ /**
+ * Adds delta to current counter value.
+ *
+ * @param delta Delta.
+ * @return Value before add.
+ */
+ public long getAndAdd(long delta) {
+ return cntr.getAndAdd(delta);
+ }
+
+ /**
+ * @return Next update counter.
+ */
+ public long next() {
+ return cntr.incrementAndGet();
+ }
+
+ /**
+ * Sets value to update counter,
+ *
+ * @param val Values.
+ */
+ public void update(long val) {
+ while (true) {
+ long val0 = cntr.get();
+
+ if (val0 >= val)
+ break;
+
+ if (cntr.compareAndSet(val0, val))
+ break;
+ }
+ }
+
+ /**
+ * Updates counter by delta from start position.
+ *
+ * @param start Start.
+ * @param delta Delta.
+ */
+ public synchronized void update(long start, long delta) {
+ long cur = cntr.get(), next;
+
+ if (cur > start) {
+ log.warning("Stale update counter task [cur=" + cur + ", start=" + start + ", delta=" + delta + ']');
+
+ return;
+ }
+
+ if (cur < start) {
+ // backup node with gaps
+ offer(new Item(start, delta));
+
+ return;
+ }
+
+ while (true) {
+ boolean res = cntr.compareAndSet(cur, next = start + delta);
+
+ assert res;
+
+ Item peek = peek();
+
+ if (peek == null || peek.start != next)
+ return;
+
+ Item item = poll();
+
+ assert peek == item;
+
+ start = item.start;
+ delta = item.delta;
+ cur = next;
+ }
+ }
+
+ /**
+ * @param cntr Sets initial counter.
+ */
+ public void updateInitial(long cntr) {
+ if (get() < cntr)
+ update(cntr);
+
+ initCntr = cntr;
+ }
+
+ /**
+ * @return Retrieves the minimum update counter task from queue.
+ */
+ private Item poll() {
+ return queue.poll();
+ }
+
+ /**
+ * @return Checks the minimum update counter task from queue.
+ */
+ private Item peek() {
+ return queue.peek();
+ }
+
+ /**
+ * @param item Adds update task to priority queue.
+ */
+ private void offer(Item item) {
+ queue.offer(item);
+ }
+
+ /**
+ * Update counter task. Update from start value by delta value.
+ */
+ private static class Item implements Comparable<Item> {
+ /** */
+ private final long start;
+
+ /** */
+ private final long delta;
+
+ /**
+ * @param start Start value.
+ * @param delta Delta value.
+ */
+ private Item(long start, long delta) {
+ this.start = start;
+ this.delta = delta;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int compareTo(@NotNull Item o) {
+ int cmp = Long.compare(this.start, o.start);
+
+ assert cmp != 0;
+
+ return cmp;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 8e96ae2..1eccd6f 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
@@ -52,7 +52,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheUpdateTxResult;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
-import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCounters;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
@@ -809,7 +809,14 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
}
- applyTxCounters();
+ // Apply cache size deltas.
+ applyTxSizes();
+
+ TxCounters txCntrs = txCounters(false);
+
+ // Apply update counters.
+ if (txCntrs != null)
+ applyPartitionsUpdatesCounters(txCntrs.updateCounters());
if (!near() && !F.isEmpty(dataEntries) && cctx.wal() != null) {
// Set new update counters for data entries received from persisted tx entries.
@@ -848,38 +855,6 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/** {@inheritDoc} */
- @Override protected void applyTxCounters() {
- super.applyTxCounters();
-
- TxCounters txCntrs = txCounters(false);
-
- if (txCntrs == null)
- return;
-
- Map<Integer, PartitionUpdateCounters> updCntrs = txCntrs.updateCounters();
-
- for (Map.Entry<Integer, PartitionUpdateCounters> entry : updCntrs.entrySet()) {
- int cacheId = entry.getKey();
-
- GridDhtPartitionTopology top = cctx.cacheContext(cacheId).topology();
-
- Map<Integer, Long> cacheUpdCntrs = entry.getValue().updateCounters();
-
- assert cacheUpdCntrs != null;
-
- for (Map.Entry<Integer, Long> e : cacheUpdCntrs.entrySet()) {
- long updCntr = e.getValue();
-
- GridDhtLocalPartition dhtPart = top.localPartition(e.getKey());
-
- assert dhtPart != null && updCntr > 0;
-
- dhtPart.updateCounter(updCntr);
- }
- }
- }
-
- /** {@inheritDoc} */
@Override public final void commitRemoteTx() throws IgniteCheckedException {
if (optimistic())
state(PREPARED);
@@ -965,6 +940,11 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
if (state(ROLLING_BACK) || state() == UNKNOWN) {
cctx.tm().rollbackTx(this, false, skipCompletedVers);
+ TxCounters counters = txCounters(false);
+
+ if (counters != null)
+ applyPartitionsUpdatesCounters(counters.updateCounters());
+
state(ROLLED_BACK);
}
}
@@ -1029,6 +1009,39 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
}
+ /**
+ * Applies partition counters updates for mvcc transactions.
+ *
+ * @param counters Counters values to be updated.
+ */
+ private void applyPartitionsUpdatesCounters(Iterable<PartitionUpdateCountersMessage> counters) {
+ if (counters == null)
+ return;
+
+ int cacheId = CU.UNDEFINED_CACHE_ID;
+ GridDhtPartitionTopology top = null;
+
+ for (PartitionUpdateCountersMessage counter : counters) {
+ if (counter.cacheId() != cacheId) {
+ GridCacheContext ctx0 = cctx.cacheContext(cacheId = counter.cacheId());
+
+ assert ctx0.mvccEnabled();
+
+ top = ctx0.topology();
+ }
+
+ assert top != null;
+
+ for (int i = 0; i < counter.size(); i++) {
+ GridDhtLocalPartition part = top.localPartition(counter.partition(i));
+
+ assert part != null;
+
+ part.updateCounter(counter.initialCounter(i), counter.updatesCount(i));
+ }
+ }
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return GridToStringBuilder.toString(GridDistributedTxRemoteAdapter.class, this, "super", super.toString());
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
index 343e418..96f5c9b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
@@ -102,11 +102,6 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
}
/** {@inheritDoc} */
- @Override protected long nextMvccPartitionCounter() {
- return locPart.nextMvccUpdateCounter();
- }
-
- /** {@inheritDoc} */
@Override public int memorySize() throws IgniteCheckedException {
int rdrsOverhead;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index f56df00..a5a12a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@ -969,45 +969,51 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
}
/**
- * @return Current update index.
+ * @return Current update counter.
*/
public long updateCounter() {
return store.updateCounter();
}
/**
- * @return Current mvcc update counter value.
+ * @param val Update counter value.
*/
- public long mvccUpdateCounter() {
- return store.mvccUpdateCounter();
+ public void updateCounter(long val) {
+ store.updateCounter(val);
}
/**
- * @return Next mvcc update counter.
+ * @return Initial update counter.
*/
- public long nextMvccUpdateCounter() {
- return store.nextMvccUpdateCounter();
+ public long initialUpdateCounter() {
+ return store.initialUpdateCounter();
}
/**
- * @return Initial update counter.
+ * @param val Initial update counter value.
*/
- public long initialUpdateCounter() {
- return store.initialUpdateCounter();
+ public void initialUpdateCounter(long val) {
+ store.updateInitialCounter(val);
}
/**
- * @param val Update index value.
+ * Updates MVCC cache update counter on primary node.
+ *
+ * @param delta Value to be added to update counter.
+ * @return Update counter value before update.
*/
- public void updateCounter(long val) {
- store.updateCounter(val);
+ public long getAndIncrementUpdateCounter(long delta) {
+ return store.getAndIncrementUpdateCounter(delta);
}
/**
- * @param val Initial update index value.
+ * Updates MVCC cache update counter on backup node.
+ *
+ * @param start Start position
+ * @param delta Delta.
*/
- public void initialUpdateCounter(long val) {
- store.updateInitialCounter(val);
+ public void updateCounter(long start, long delta) {
+ store.updateCounter(start, delta);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 be1c7e2..f6df80b 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
@@ -2205,7 +2205,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
MvccSnapshot snapshot = new MvccSnapshotWithoutTxs(s0.coordinatorVersion(), s0.counter(),
req.operationCounter(), s0.cleanupVersion());
- tx.mvccEnlistBatch(ctx, req.op(), req.keys(), req.values(), snapshot, req.updateCounters());
+ tx.mvccEnlistBatch(ctx, req.op(), req.keys(), req.values(), snapshot);
GridDhtTxQueryEnlistResponse res = new GridDhtTxQueryEnlistResponse(req.cacheId(),
req.dhtFutureId(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
index a3471c7..2b34a41 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
@@ -61,7 +61,6 @@ import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
@@ -135,6 +134,9 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
/** Processed entries count. */
protected long cnt;
+ /** New DHT nodes. */
+ protected Set<UUID> newDhtNodes = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
/** Near node ID. */
protected final UUID nearNodeId;
@@ -186,9 +188,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
/** Moving partitions. */
private Map<Integer, Boolean> movingParts;
- /** Update counters to be sent to the near node in case it is a backup node also. */
- protected GridLongList nearUpdCntrs;
-
/**
* @param nearNodeId Near node ID.
* @param nearLockVer Near lock version.
@@ -388,7 +387,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
tx,
cctx.localNodeId(),
topVer,
- null,
mvccSnapshot,
isMoving(key.partition()));
@@ -403,7 +401,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
val,
0,
topVer,
- null,
mvccSnapshot,
op.cacheOperation(),
isMoving(key.partition()),
@@ -575,7 +572,7 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
cnt++;
if (op != EnlistOperation.LOCK)
- addToBatch(entry.key(), val, updRes.mvccHistory(), updRes.updateCounter(), entry.context().cacheId());
+ addToBatch(entry.key(), val, updRes.mvccHistory(), entry.context().cacheId());
}
/**
@@ -585,9 +582,8 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
* @param key Key.
* @param val Value.
* @param hist History rows.
- * @param updCntr Update counter.
*/
- private void addToBatch(KeyCacheObject key, CacheObject val, List<MvccLinkAwareSearchRow> hist, long updCntr,
+ private void addToBatch(KeyCacheObject key, CacheObject val, List<MvccLinkAwareSearchRow> hist,
int cacheId) throws IgniteCheckedException {
List<ClusterNode> backups = backupNodes(key);
@@ -609,15 +605,10 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
updateMappings(node);
if (newRemoteTx(node))
- tx.addLockTransactionNode(node);
+ addNewRemoteTxNode(node);
hasNearNodeUpdates = true;
- if (nearUpdCntrs == null)
- nearUpdCntrs = new GridLongList();
-
- nearUpdCntrs.add(updCntr);
-
continue;
}
@@ -637,7 +628,7 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
hist0 = fetchHistoryInfo(key, hist);
}
- batch.add(key, moving ? hist0 : val, updCntr);
+ batch.add(key, moving ? hist0 : val);
if (batch.size() == BATCH_SIZE) {
assert batches != null;
@@ -705,6 +696,17 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
}
/**
+ * Add new involved DHT node.
+ *
+ * @param node Node.
+ */
+ private void addNewRemoteTxNode(ClusterNode node) {
+ tx.addLockTransactionNode(node);
+
+ newDhtNodes.add(node.id());
+ }
+
+ /**
* Checks if there free space in batches or free slot in in-flight batches is available for the given key.
*
* @param key Key.
@@ -752,7 +754,7 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
GridDhtTxQueryEnlistRequest req;
if (newRemoteTx(node)) {
- tx.addLockTransactionNode(node);
+ addNewRemoteTxNode(node);
// If this is a first request to this node, send full info.
req = new GridDhtTxQueryFirstEnlistRequest(cctx.cacheId(),
@@ -768,8 +770,8 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
it.operation(),
FIRST_BATCH_ID,
batch.keys(),
- batch.values(),
- batch.updateCounters());
+ batch.values()
+ );
}
else {
// Send only keys, values, LockVersion and batchId if this is not a first request to this backup.
@@ -780,8 +782,8 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
++batchIdCntr,
mvccSnapshot.operationCounter(),
batch.keys(),
- batch.values(),
- batch.updateCounters());
+ batch.values()
+ );
}
ConcurrentMap<Integer, Batch> pending0 = null;
@@ -1036,9 +1038,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
*/
private List<Message> vals;
- /** Update counters. */
- private GridLongList updCntrs;
-
/**
* @param node Cluster node.
*/
@@ -1059,9 +1058,8 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
* @param key Key.
* @param val Value or preload entries collection.
*/
- public void add(KeyCacheObject key, Message val, long updCntr) {
+ public void add(KeyCacheObject key, Message val) {
assert val == null || val instanceof CacheObject || val instanceof CacheEntryInfoCollection;
- assert updCntr > 0;
if (keys == null)
keys = new ArrayList<>();
@@ -1075,11 +1073,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
vals.add(val);
}
- if (updCntrs == null)
- updCntrs = new GridLongList();
-
- updCntrs.add(updCntr);
-
assert (vals == null) || keys.size() == vals.size();
}
@@ -1103,13 +1096,6 @@ public abstract class GridDhtTxAbstractEnlistFuture extends GridCacheFutureAdapt
public List<Message> values() {
return vals;
}
-
- /**
- * @return Update counters.
- */
- public GridLongList updateCounters() {
- return updCntrs;
- }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 4c72e6f..9283939 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
@@ -371,7 +371,8 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
tx.activeCachesDeploymentEnabled(),
false,
false,
- tx.mvccSnapshot());
+ tx.mvccSnapshot(),
+ tx.filterUpdateCountersForBackupNode(n));
try {
cctx.io().send(n, req, tx.ioPolicy());
@@ -457,11 +458,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
for (IgniteTxEntry e : dhtMapping.entries())
updCntrs.add(e.updateCounter());
- Map<Integer, PartitionUpdateCounters> updCntrsForNode = null;
-
- if (dhtMapping.queryUpdate() && commit)
- updCntrsForNode = tx.filterUpdateCountersForBackupNode(n);
-
GridDhtTxFinishRequest req = new GridDhtTxFinishRequest(
tx.nearNodeId(),
futId,
@@ -489,7 +485,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
false,
false,
mvccSnapshot,
- updCntrsForNode);
+ commit ? null : tx.filterUpdateCountersForBackupNode(n));
req.writeVersion(tx.writeVersion() != null ? tx.writeVersion() : tx.xidVersion());
@@ -559,7 +555,8 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity
tx.activeCachesDeploymentEnabled(),
false,
false,
- mvccSnapshot);
+ mvccSnapshot,
+ null);
req.writeVersion(tx.writeVersion());
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
index 8e9ece6..61896b5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
@@ -20,11 +20,9 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.io.Externalizable;
import java.nio.ByteBuffer;
import java.util.Collection;
-import java.util.Map;
import java.util.UUID;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.GridDirectCollection;
-import org.apache.ignite.internal.GridDirectMap;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxFinishRequest;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
@@ -74,8 +72,8 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
private MvccSnapshot mvccSnapshot;
/** */
- @GridDirectMap(keyType = Integer.class, valueType = PartitionUpdateCounters.class)
- private Map<Integer, PartitionUpdateCounters> updCntrs;
+ @GridDirectCollection(PartitionUpdateCountersMessage.class)
+ private Collection<PartitionUpdateCountersMessage> updCntrs;
/**
* Empty constructor required for {@link Externalizable}.
@@ -107,6 +105,10 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
* @param subjId Subject ID.
* @param taskNameHash Task name hash.
* @param addDepInfo Deployment info flag.
+ * @param retVal Need return value
+ * @param waitRemoteTxs Wait remote transactions flag
+ * @param mvccSnapshot Mvcc snapshot.
+ * @param updCntrs Update counters for mvcc Tx.
*/
public GridDhtTxFinishRequest(
UUID nearNodeId,
@@ -133,7 +135,8 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
boolean addDepInfo,
boolean retVal,
boolean waitRemoteTxs,
- MvccSnapshot mvccSnapshot
+ MvccSnapshot mvccSnapshot,
+ Collection<PartitionUpdateCountersMessage> updCntrs
) {
super(
xidVer,
@@ -163,6 +166,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
this.isolation = isolation;
this.miniId = miniId;
this.mvccSnapshot = mvccSnapshot;
+ this.updCntrs = updCntrs;
needReturnValue(retVal);
waitRemoteTransactions(waitRemoteTxs);
@@ -193,6 +197,10 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
* @param taskNameHash Task name hash.
* @param updateIdxs Partition update idxs.
* @param addDepInfo Deployment info flag.
+ * @param retVal Need return value
+ * @param waitRemoteTxs Wait remote transactions flag
+ * @param mvccSnapshot Mvcc snapshot.
+ * @param updCntrs Update counters for mvcc Tx.
*/
public GridDhtTxFinishRequest(
UUID nearNodeId,
@@ -221,7 +229,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
boolean retVal,
boolean waitRemoteTxs,
MvccSnapshot mvccSnapshot,
- Map<Integer, PartitionUpdateCounters> updCntrs
+ Collection<PartitionUpdateCountersMessage> updCntrs
) {
this(nearNodeId,
futId,
@@ -247,7 +255,8 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
addDepInfo,
retVal,
waitRemoteTxs,
- mvccSnapshot);
+ mvccSnapshot,
+ updCntrs);
if (updateIdxs != null && !updateIdxs.isEmpty()) {
partUpdateCnt = new GridLongList(updateIdxs.size());
@@ -255,8 +264,6 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
for (Long idx : updateIdxs)
partUpdateCnt.add(idx);
}
-
- this.updCntrs = updCntrs;
}
/**
@@ -367,7 +374,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
/**
* @return Partition counters update deferred until transaction commit.
*/
- public Map<Integer, PartitionUpdateCounters> updateCounters() {
+ public Collection<PartitionUpdateCountersMessage> updateCounters() {
return updCntrs;
}
@@ -429,7 +436,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
writer.incrementState();
case 28:
- if (!writer.writeMap("updCntrs", updCntrs, MessageCollectionItemType.INT, MessageCollectionItemType.MSG))
+ if (!writer.writeCollection("updCntrs", updCntrs, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
@@ -511,7 +518,7 @@ public class GridDhtTxFinishRequest extends GridDistributedTxFinishRequest {
reader.incrementState();
case 28:
- updCntrs = reader.readMap("updCntrs", MessageCollectionItemType.INT, MessageCollectionItemType.MSG, false);
+ updCntrs = reader.readCollection("updCntrs", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 613f160..a0c9d15 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -18,9 +18,9 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.io.Externalizable;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -944,43 +944,32 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
* @param node Backup node.
* @return Partition counters map for the given backup node.
*/
- public Map<Integer, PartitionUpdateCounters> filterUpdateCountersForBackupNode(ClusterNode node) {
+ public List<PartitionUpdateCountersMessage> filterUpdateCountersForBackupNode(ClusterNode node) {
TxCounters txCntrs = txCounters(false);
- if (txCntrs == null)
+ if (txCntrs == null || F.isEmpty(txCntrs.updateCounters()))
return null;
- Map<Integer, PartitionUpdateCounters> updCntrs = txCntrs.updateCounters();
+ Collection<PartitionUpdateCountersMessage> updCntrs = txCntrs.updateCounters();
- Map<Integer, PartitionUpdateCounters> res = new HashMap<>();
+ List<PartitionUpdateCountersMessage> res = new ArrayList<>(updCntrs.size());
AffinityTopologyVersion top = topologyVersionSnapshot();
- for (Map.Entry<Integer, PartitionUpdateCounters> entry : updCntrs.entrySet()) {
- Integer cacheId = entry.getKey();
+ for (PartitionUpdateCountersMessage partCntrs : updCntrs) {
+ GridCacheAffinityManager affinity = cctx.cacheContext(partCntrs.cacheId()).affinity();
- Map<Integer, Long> partsCntrs = entry.getValue().updateCounters();
+ PartitionUpdateCountersMessage resCntrs = new PartitionUpdateCountersMessage(partCntrs.cacheId(), partCntrs.size());
- assert !F.isEmpty(partsCntrs);
+ for (int i = 0; i < partCntrs.size(); i++) {
+ int part = partCntrs.partition(i);
- GridCacheAffinityManager affinity = cctx.cacheContext(cacheId).affinity();
-
- Map<Integer, Long> resCntrs = new HashMap<>(partsCntrs.size());
-
- for (Map.Entry<Integer, Long> e : partsCntrs.entrySet()) {
- Integer p = e.getKey();
-
- Long cntr = e.getValue();
-
- if (affinity.backupByPartition(node, p, top)) {
- assert cntr != null && cntr > 0 : cntr;
-
- resCntrs.put(p, cntr);
- }
+ if (affinity.backupByPartition(node, part, top))
+ resCntrs.add(part, partCntrs.initialCounter(i), partCntrs.updatesCount(i));
}
- if (!resCntrs.isEmpty())
- res.put(cacheId, new PartitionUpdateCounters(resCntrs));
+ if (resCntrs.size() > 0)
+ res.add(resCntrs);
}
return res;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/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 1974038..0edf63f 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
@@ -61,7 +61,6 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTx
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersionAware;
import org.apache.ignite.internal.processors.cache.mvcc.MvccVersionAware;
@@ -69,6 +68,7 @@ import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
+import org.apache.ignite.internal.processors.cache.transactions.TxCounters;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
@@ -1397,7 +1397,8 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
tx.activeCachesDeploymentEnabled(),
tx.storeWriteThrough(),
retVal,
- mvccSnapshot);
+ mvccSnapshot,
+ tx.filterUpdateCountersForBackupNode(n));
req.queryUpdate(dhtMapping.queryUpdate());
@@ -1503,7 +1504,8 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
tx.activeCachesDeploymentEnabled(),
tx.storeWriteThrough(),
retVal,
- mvccSnapshot);
+ mvccSnapshot,
+ null);
for (IgniteTxEntry entry : nearMapping.entries()) {
if (CU.writes().apply(entry)) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
index 68c1f39..30e8ceb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
@@ -88,6 +88,10 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
@GridDirectCollection(GridCacheVersion.class)
private Collection<GridCacheVersion> ownedVals;
+ /** */
+ @GridDirectCollection(PartitionUpdateCountersMessage.class)
+ private Collection<PartitionUpdateCountersMessage> counters;
+
/** Near transaction ID. */
private GridCacheVersion nearXidVer;
@@ -130,7 +134,9 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
* @param last {@code True} if this is last prepare request for node.
* @param addDepInfo Deployment info flag.
* @param storeWriteThrough Cache store write through flag.
- * @param retVal Need return value flag.
+ * @param retVal Need return value flag
+ * @param mvccSnapshot Mvcc snapshot.
+ * @param counters Update counters for mvcc Tx.
*/
public GridDhtTxPrepareRequest(
IgniteUuid futId,
@@ -149,7 +155,8 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
boolean addDepInfo,
boolean storeWriteThrough,
boolean retVal,
- MvccSnapshot mvccInfo) {
+ MvccSnapshot mvccSnapshot,
+ Collection<PartitionUpdateCountersMessage> counters) {
super(tx,
timeout,
null,
@@ -170,7 +177,8 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
this.nearXidVer = nearXidVer;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
- this.mvccSnapshot = mvccInfo;
+ this.mvccSnapshot = mvccSnapshot;
+ this.counters = counters;
storeWriteThrough(storeWriteThrough);
needReturnValue(retVal);
@@ -190,6 +198,13 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
/**
+ * @return Update counters list.
+ */
+ public Collection<PartitionUpdateCountersMessage> updateCounters() {
+ return counters;
+ }
+
+ /**
* @return Near cache writes for which cache was not found (possible if client near cache was closed).
*/
@Nullable public List<IgniteTxKey> nearWritesCacheMissed() {
@@ -492,6 +507,12 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
writer.incrementState();
+ case 34:
+ if (!writer.writeCollection("counters", counters, MessageCollectionItemType.MSG))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -620,6 +641,14 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
+ case 34:
+ counters = reader.readCollection("counters", MessageCollectionItemType.MSG);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtTxPrepareRequest.class);
@@ -632,7 +661,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 34;
+ return 35;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a82d4aa/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistRequest.java
index 650e1dc..a1bc26b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistRequest.java
@@ -30,7 +30,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.EnlistOperation;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -68,9 +67,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
@GridDirectCollection(Message.class)
private List<Message> vals;
- /** */
- private GridLongList updCntrs;
-
/**
*
*/
@@ -86,7 +82,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
* @param mvccOpCnt Mvcc operation counter.
* @param keys Keys.
* @param vals Values.
- * @param updCntrs Update counters.
*/
GridDhtTxQueryEnlistRequest(int cacheId,
IgniteUuid dhtFutId,
@@ -95,8 +90,7 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
int batchId,
int mvccOpCnt,
List<KeyCacheObject> keys,
- List<Message> vals,
- GridLongList updCntrs) {
+ List<Message> vals) {
this.cacheId = cacheId;
this.dhtFutId = dhtFutId;
this.lockVer = lockVer;
@@ -105,7 +99,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
this.mvccOpCnt = mvccOpCnt;
this.keys = keys;
this.vals = vals;
- this.updCntrs = updCntrs;
}
/**
@@ -160,13 +153,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
}
/**
- * @return Update counters.
- */
- public GridLongList updateCounters() {
- return updCntrs;
- }
-
- /**
* @return Batch id.
*/
public int batchId() {
@@ -292,12 +278,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
writer.incrementState();
case 9:
- if (!writer.writeMessage("updCntrs", updCntrs))
- return false;
-
- writer.incrementState();
-
- case 10:
if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
return false;
@@ -372,14 +352,6 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
reader.incrementState();
case 9:
- updCntrs = reader.readMessage("updCntrs");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 10:
vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -394,7 +366,7 @@ public class GridDhtTxQueryEnlistRequest extends GridCacheIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 11;
+ return 10;
}
/** {@inheritDoc} */