You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/08/30 12:24:42 UTC
[36/45] ignite git commit: IGNITE-4191: MVCC and transactional SQL
support. Joint multi-man-years efforts of Semen Boikov, Igor Seliverstov,
Alexander Paschenko, Igor Sapego, Sergey Kalashnikov, Roman Kondakov,
Pavel Kuznetsov, Ivan Pavlukhin, Andrey Mas
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/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 c1293fc..dd6ea48 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
@@ -50,6 +50,8 @@ import org.apache.ignite.internal.processors.cache.GridCacheReturnCompletableWra
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
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.GridDhtPartitionsUpdateCountersMap;
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;
@@ -205,6 +207,16 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
return false;
}
+ /** {@inheritDoc} */
+ @Override public void activeCachesDeploymentEnabled(boolean depEnabled) {
+ throw new UnsupportedOperationException("Remote tx doesn't support deployment.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void addActiveCache(GridCacheContext cacheCtx, boolean recovery) throws IgniteCheckedException {
+ txState.addActiveCache(cacheCtx, recovery, this);
+ }
+
/**
* @return Checks if transaction has no entries.
*/
@@ -460,7 +472,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
GridCacheReturnCompletableWrapper wrapper = null;
- if (!F.isEmpty(writeMap)) {
+ if (!F.isEmpty(writeMap) || mvccSnapshot != null) {
GridCacheReturn ret = null;
if (!near() && !local() && onePhaseCommit()) {
@@ -488,6 +500,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
cctx.database().checkpointReadLock();
try {
+ assert !txState.mvccEnabled(cctx) || mvccSnapshot != null : "Mvcc is not initialized: " + this;
+
Collection<IgniteTxEntry> entries = near() || cctx.snapshot().needTxReadLogging() ? allEntries() : writeEntries();
// Data entry to write to WAL and associated with it TxEntry.
@@ -613,7 +627,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter());
+ txEntry.updateCounter(),
+ mvccSnapshot());
else {
assert val != null : txEntry;
@@ -637,9 +652,10 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter());
+ txEntry.updateCounter(),
+ mvccSnapshot());
- txEntry.updateCounter(updRes.updatePartitionCounter());
+ txEntry.updateCounter(updRes.updateCounter());
if (updRes.loggedPointer() != null)
ptr = updRes.loggedPointer();
@@ -674,9 +690,10 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
CU.subjectId(this, cctx),
resolveTaskName(),
dhtVer,
- txEntry.updateCounter());
+ txEntry.updateCounter(),
+ mvccSnapshot());
- txEntry.updateCounter(updRes.updatePartitionCounter());
+ txEntry.updateCounter(updRes.updateCounter());
if (updRes.loggedPointer() != null)
ptr = updRes.loggedPointer();
@@ -790,6 +807,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
}
+ updateLocalCounters();
+
if (!near() && !F.isEmpty(dataEntries) && cctx.wal() != null) {
// Set new update counters for data entries received from persisted tx entries.
List<DataEntry> entriesWithCounters = dataEntries.stream()
@@ -803,6 +822,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
cctx.wal().flush(ptr, false);
}
catch (StorageException e) {
+ err = e;
+
throw new IgniteCheckedException("Failed to log transaction record " +
"(transaction will be rolled back): " + this, e);
}
@@ -810,6 +831,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
finally {
cctx.database().checkpointReadUnlock();
+ notifyDrManager(state() == COMMITTING && err == null);
+
if (wrapper != null)
wrapper.initialize(ret);
}
@@ -822,6 +845,33 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
}
+ /**
+ * Applies update counters to the local partitions.
+ */
+ private void updateLocalCounters() {
+ Map<Integer, GridDhtPartitionsUpdateCountersMap> updCntrsMap = updateCountersMap();
+
+ if (F.isEmpty(updCntrsMap))
+ return;
+
+ for (Map.Entry<Integer, GridDhtPartitionsUpdateCountersMap> entry : updCntrsMap.entrySet()) {
+ GridCacheContext cacheCtx = cctx.cacheContext(entry.getKey());
+
+ GridDhtPartitionsUpdateCountersMap cacheUpdCntrs = entry.getValue();
+
+ assert cacheUpdCntrs != null && !F.isEmpty(cacheUpdCntrs.updateCounters());
+
+ for (Map.Entry<Integer, Long> e : cacheUpdCntrs.updateCounters().entrySet()) {
+ Long updCntr = e.getValue();
+ GridDhtLocalPartition part = cacheCtx.topology().localPartition(e.getKey());
+
+ assert part != null && updCntr != null && updCntr > 0;
+
+ part.updateCounter(updCntr);
+ }
+ }
+ }
+
/** {@inheritDoc} */
@Override public final void commitRemoteTx() throws IgniteCheckedException {
if (optimistic())
@@ -901,6 +951,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
/** {@inheritDoc} */
@Override public final void rollbackRemoteTx() {
try {
+ notifyDrManager(false);
+
// Note that we don't evict near entries here -
// they will be deleted by their corresponding transactions.
if (state(ROLLING_BACK) || state() == UNKNOWN) {
@@ -971,8 +1023,17 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/** {@inheritDoc} */
+ @Override public void updateCountersMap(Map<Integer, GridDhtPartitionsUpdateCountersMap> updCntrsMap) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<Integer, GridDhtPartitionsUpdateCountersMap> updateCountersMap() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return GridToStringBuilder.toString(GridDistributedTxRemoteAdapter.class, this, "super", super.toString());
}
-
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java
new file mode 100644
index 0000000..63eff40
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java
@@ -0,0 +1,112 @@
+/*
+ * 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.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxAbstractEnlistFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.lang.IgniteInClosure;
+
+/**
+ *
+ */
+public class CompoundLockFuture extends GridFutureAdapter<Void> implements DhtLockFuture<Void>, IgniteInClosure<IgniteInternalFuture<?>> {
+ /** */
+ private static final long serialVersionUID = 4644646033267042131L;
+ /** */
+ private static final AtomicIntegerFieldUpdater<CompoundLockFuture> CNT_UPD =
+ AtomicIntegerFieldUpdater.newUpdater(CompoundLockFuture.class, "cnt");
+ /** */
+ private volatile int cnt;
+ /** */
+ private final GridDhtTxLocalAdapter tx;
+
+ /**
+ * @param cnt ResultSet futures count.
+ * @param tx Transaction.
+ */
+ public CompoundLockFuture(int cnt, GridDhtTxLocalAdapter tx) {
+ this.cnt = cnt;
+ this.tx = tx;
+ }
+
+ /**
+ * @param fut ResultSet future.
+ */
+ public void register(IgniteInternalFuture<?> fut) {
+ fut.listen(this);
+ }
+
+ /**
+ * Init method.
+ */
+ public void init() {
+ while(true) {
+ IgniteInternalFuture<?> fut = tx.lockFuture();
+
+ if (fut == GridDhtTxLocalAdapter.ROLLBACK_FUT) {
+ onDone(tx.timedOut() ? tx.timeoutException() : tx.rollbackException());
+
+ break;
+ }
+ else if (fut != null) {
+ // Wait for previous future.
+ assert fut instanceof GridNearTxAbstractEnlistFuture
+ || fut instanceof GridDhtTxAbstractEnlistFuture
+ || fut instanceof CompoundLockFuture
+ || fut instanceof GridNearTxSelectForUpdateFuture : fut;
+
+ // Terminate this future if parent future is terminated by rollback.
+ if (!fut.isDone()) {
+ fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
+ @Override public void apply(IgniteInternalFuture fut) {
+ if (fut.error() != null)
+ onDone(fut.error());
+ }
+ });
+ }
+ else if (fut.error() != null)
+ onDone(fut.error());
+
+ break;
+ }
+ else if (tx.updateLockFuture(null, this))
+ break;
+ }
+ }
+
+ @Override public void apply(IgniteInternalFuture<?> future) {
+ if (!isDone() && (future.error() != null || CNT_UPD.decrementAndGet(this) == 0)) {
+ Throwable err = future.error();
+
+ if (err == null)
+ tx.clearLockFuture(this);
+
+ onDone(err);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onError(Throwable error) {
+ assert error != null;
+
+ onDone(error);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/DhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/DhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/DhtLockFuture.java
new file mode 100644
index 0000000..b729dcd
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/DhtLockFuture.java
@@ -0,0 +1,30 @@
+/*
+ * 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 org.apache.ignite.internal.IgniteInternalFuture;
+
+/**
+ * Marker interface.
+ */
+public interface DhtLockFuture<T> extends IgniteInternalFuture<T> {
+ /**
+ * @param error Error.
+ */
+ public void onError(Throwable error);
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ExceptionAware.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ExceptionAware.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ExceptionAware.java
new file mode 100644
index 0000000..036492f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ExceptionAware.java
@@ -0,0 +1,30 @@
+/*
+ * 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 org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public interface ExceptionAware {
+ /**
+ * @return Exception.
+ */
+ @Nullable Throwable error();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 54a850c..70ce082 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -46,6 +46,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridPartitionStateMap;
@@ -199,6 +200,11 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
+ @Override public MvccCoordinator mvccCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
@Override public boolean holdsLock() {
return lock.isWriteLockedByCurrentThread() || lock.getReadHoldCount() > 0;
}
@@ -207,6 +213,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
@Override public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
DiscoCache discoCache,
+ MvccCoordinator mvccCrd,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index 5adbda7..cd37650 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -68,6 +68,8 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetR
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
+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.version.GridCacheVersion;
import org.apache.ignite.internal.processors.platform.cache.PlatformCacheEntryFilter;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -230,16 +232,30 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
entry.unswap();
- GridCacheEntryInfo info = entry.info();
+ if (ctx.mvccEnabled()) {
+ List<GridCacheEntryInfo> infos = entry.allVersionsInfo();
- if (info == null) {
- assert entry.obsolete() : entry;
+ if (infos == null) {
+ assert entry.obsolete() : entry;
- continue;
+ continue;
+ }
+
+ for (int i = 0; i < infos.size(); i++)
+ res.addInfo(infos.get(i));
}
+ else {
+ GridCacheEntryInfo info = entry.info();
+
+ if (info == null) {
+ assert entry.obsolete() : entry;
- if (!info.isNew())
- res.addInfo(info);
+ continue;
+ }
+
+ if (!info.isNew())
+ res.addInfo(info);
+ }
entry.touch(msg.topologyVersion());
@@ -588,6 +604,9 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
return;
}
+ //TODO IGNITE-7954
+ MvccUtils.verifyMvccOperationSupport(ctx, "Load");
+
final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion();
// Version for all loaded entries.
@@ -766,6 +785,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param taskName Task name.
* @param expiry Expiry policy.
* @param skipVals Skip values flag.
+ * @param mvccSnapshot MVCC snapshot.
* @return Get future.
*/
IgniteInternalFuture<Map<KeyCacheObject, EntryGetResult>> getDhtAllAsync(
@@ -776,7 +796,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
String taskName,
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
- boolean recovery
+ boolean recovery,
+ MvccSnapshot mvccSnapshot
) {
return getAllAsync0(keys,
readerArgs,
@@ -789,7 +810,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
skipVals,
/*keep cache objects*/true,
recovery,
- /*need version*/true);
+ /*need version*/true,
+ mvccSnapshot);
}
/**
@@ -803,6 +825,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param taskNameHash Task name hash code.
* @param expiry Expiry policy.
* @param skipVals Skip values flag.
+ * @param mvccSnapshot MVCC snapshot.
* @return DHT future.
*/
public GridDhtFuture<Collection<GridCacheEntryInfo>> getDhtAsync(UUID reader,
@@ -815,7 +838,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
int taskNameHash,
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
- boolean recovery
+ boolean recovery,
+ MvccSnapshot mvccSnapshot
) {
GridDhtGetFuture<K, V> fut = new GridDhtGetFuture<>(ctx,
msgId,
@@ -828,7 +852,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
expiry,
skipVals,
recovery,
- addReaders);
+ addReaders,
+ mvccSnapshot);
fut.init();
@@ -846,9 +871,10 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param taskNameHash Task name hash.
* @param expiry Expiry.
* @param skipVals Skip vals flag.
+ * @param mvccSnapshot Mvcc snapshot.
* @return Future for the operation.
*/
- public GridDhtGetSingleFuture getDhtSingleAsync(
+ GridDhtGetSingleFuture getDhtSingleAsync(
UUID nodeId,
long msgId,
KeyCacheObject key,
@@ -859,7 +885,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
int taskNameHash,
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
- boolean recovery
+ boolean recovery,
+ MvccSnapshot mvccSnapshot
) {
GridDhtGetSingleFuture fut = new GridDhtGetSingleFuture<>(
ctx,
@@ -873,7 +900,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
taskNameHash,
expiry,
skipVals,
- recovery);
+ recovery,
+ mvccSnapshot);
fut.init();
@@ -901,7 +929,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
req.taskNameHash(),
expiryPlc,
req.skipValues(),
- req.recovery());
+ req.recovery(),
+ req.mvccSnapshot());
fut.listen(new CI1<IgniteInternalFuture<GridCacheEntryInfo>>() {
@Override public void apply(IgniteInternalFuture<GridCacheEntryInfo> f) {
@@ -1004,7 +1033,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
req.taskNameHash(),
expiryPlc,
req.skipValues(),
- req.recovery());
+ req.recovery(),
+ req.mvccSnapshot());
fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() {
@Override public void apply(IgniteInternalFuture<Collection<GridCacheEntryInfo>> f) {
@@ -1230,6 +1260,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
if (expVer.equals(curVer))
return false;
+ // TODO IGNITE-7164 check mvcc crd for mvcc enabled txs.
+
Collection<ClusterNode> cacheNodes0 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), expVer);
Collection<ClusterNode> cacheNodes1 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), curVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/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 fe02090..343e418 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,6 +102,11 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
}
/** {@inheritDoc} */
+ @Override protected long nextMvccPartitionCounter() {
+ return locPart.nextMvccUpdateCounter();
+ }
+
+ /** {@inheritDoc} */
@Override public int memorySize() throws IgniteCheckedException {
int rdrsOverhead;
@@ -648,7 +653,10 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
']');
}
- removeValue();
+ if (cctx.mvccEnabled())
+ cctx.offheap().mvccRemoveAll(this);
+ else
+ removeValue();
// Give to GC.
update(null, 0L, 0L, ver, true);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
index a969dd3..023c058 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.ReaderArguments;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridCompoundIdentityFuture;
@@ -114,6 +115,9 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
/** */
private final boolean addReaders;
+ /** */
+ private final MvccSnapshot mvccSnapshot;
+
/**
* @param cctx Context.
* @param msgId Message ID.
@@ -125,6 +129,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
* @param taskNameHash Task name hash code.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
+ * @param mvccSnapshot MVCC snapshot.
*/
public GridDhtGetFuture(
GridCacheContext<K, V> cctx,
@@ -138,7 +143,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean skipVals,
boolean recovery,
- boolean addReaders
+ boolean addReaders,
+ MvccSnapshot mvccSnapshot
) {
super(CU.<GridCacheEntryInfo>collectionsReducer(keys.size()));
@@ -157,6 +163,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
this.skipVals = skipVals;
this.recovery = recovery;
this.addReaders = addReaders;
+ this.mvccSnapshot = mvccSnapshot;
futId = IgniteUuid.randomUuid();
@@ -422,7 +429,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
taskName,
expiryPlc,
skipVals,
- recovery);
+ recovery,
+ mvccSnapshot);
}
else {
final ReaderArguments args = readerArgs;
@@ -445,7 +453,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
taskName,
expiryPlc,
skipVals,
- recovery);
+ recovery,
+ mvccSnapshot);
}
}
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
index 99a8135..de76eb1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.ReaderArguments;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
@@ -103,6 +104,9 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
/** Recovery context flag. */
private final boolean recovery;
+ /** */
+ private final MvccSnapshot mvccSnapshot;
+
/**
* @param cctx Context.
* @param msgId Message ID.
@@ -115,6 +119,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
* @param taskNameHash Task name hash code.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
+ * @param mvccSnapshot Mvcc snapshot.
*/
public GridDhtGetSingleFuture(
GridCacheContext<K, V> cctx,
@@ -128,7 +133,8 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
int taskNameHash,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean skipVals,
- boolean recovery
+ boolean recovery,
+ @Nullable MvccSnapshot mvccSnapshot
) {
assert reader != null;
assert key != null;
@@ -145,6 +151,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
this.expiryPlc = expiryPlc;
this.skipVals = skipVals;
this.recovery = recovery;
+ this.mvccSnapshot = mvccSnapshot;
futId = IgniteUuid.randomUuid();
@@ -365,7 +372,8 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
taskName,
expiryPlc,
skipVals,
- recovery);
+ recovery,
+ mvccSnapshot);
}
else {
final ReaderArguments args = readerArgs;
@@ -390,7 +398,8 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
taskName,
expiryPlc,
skipVals,
- recovery);
+ recovery,
+ mvccSnapshot);
fut0.listen(createGetFutureListener());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/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 5f976c9..f56df00 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
@@ -976,6 +976,20 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
}
/**
+ * @return Current mvcc update counter value.
+ */
+ public long mvccUpdateCounter() {
+ return store.mvccUpdateCounter();
+ }
+
+ /**
+ * @return Next mvcc update counter.
+ */
+ public long nextMvccUpdateCounter() {
+ return store.nextMvccUpdateCounter();
+ }
+
+ /**
* @return Initial update counter.
*/
public long initialUpdateCounter() {
@@ -1259,6 +1273,15 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
}
/**
+ * Returns group context.
+ *
+ * @return Group context.
+ */
+ public CacheGroupContext group() {
+ return grp;
+ }
+
+ /**
* @param cacheId Cache ID.
*/
void onCacheStopped(int cacheId) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index 529d965..c2cd781 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -54,6 +54,9 @@ import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCa
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedLockFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUpdateVersionAware;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersionAware;
+import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
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.version.GridCacheVersion;
@@ -82,7 +85,7 @@ import static org.apache.ignite.internal.processors.dr.GridDrType.DR_PRELOAD;
* Cache lock future.
*/
public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boolean>
- implements GridCacheVersionedFuture<Boolean>, GridDhtFuture<Boolean>, GridCacheMappedVersion {
+ implements GridCacheVersionedFuture<Boolean>, GridDhtFuture<Boolean>, GridCacheMappedVersion, DhtLockFuture<Boolean> {
/** */
private static final long serialVersionUID = 0L;
@@ -264,7 +267,7 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
if (tx != null) {
while(true) {
- IgniteInternalFuture<Boolean> fut = tx.lockFut;
+ IgniteInternalFuture fut = tx.lockFut;
if (fut != null) {
if (fut == GridDhtTxLocalAdapter.ROLLBACK_FUT)
@@ -274,8 +277,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
assert fut instanceof GridDhtColocatedLockFuture : fut;
// Terminate this future if parent(collocated) future is terminated by rollback.
- fut.listen(new IgniteInClosure<IgniteInternalFuture<Boolean>>() {
- @Override public void apply(IgniteInternalFuture<Boolean> fut) {
+ fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
+ @Override public void apply(IgniteInternalFuture fut) {
try {
fut.get();
}
@@ -1307,6 +1310,10 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
try {
if (entry.initialValue(info.value(),
info.version(),
+ cctx.mvccEnabled() ? ((MvccVersionAware)info).mvccVersion() : null,
+ cctx.mvccEnabled() ? ((MvccUpdateVersionAware)info).newMvccVersion() : null,
+ cctx.mvccEnabled() ? ((MvccVersionAware)entry).mvccTxState() : TxState.NA,
+ cctx.mvccEnabled() ? ((MvccUpdateVersionAware)entry).newMvccTxState() : TxState.NA,
info.ttl(),
info.expireTime(),
true,
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index 42ef309..3ae3c8f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.jetbrains.annotations.Nullable;
@@ -74,6 +75,7 @@ public interface GridDhtPartitionTopology {
public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
DiscoCache discoCache,
+ MvccCoordinator mvccCrd,
long updateSeq,
boolean stopping
) throws IgniteInterruptedCheckedException;
@@ -421,4 +423,6 @@ public interface GridDhtPartitionTopology {
* @param updateRebalanceVer {@code True} if need check rebalance state.
*/
public void onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, boolean updateRebalanceVer);
+
+ public MvccCoordinator mvccCoordinator();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index c3ff15a..1256a8b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Gri
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridPartitionStateMap;
@@ -146,6 +147,9 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** */
private volatile AffinityTopologyVersion rebalancedTopVer = AffinityTopologyVersion.NONE;
+ /** */
+ private volatile MvccCoordinator mvccCrd;
+
/**
* @param ctx Cache shared context.
* @param grp Cache group.
@@ -238,6 +242,11 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
+ @Override public MvccCoordinator mvccCoordinator() {
+ return mvccCrd;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean holdsLock() {
return lock.isWriteLockedByCurrentThread() || lock.getReadHoldCount() > 0;
}
@@ -246,6 +255,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
@Override public void updateTopologyVersion(
GridDhtTopologyFuture exchFut,
@NotNull DiscoCache discoCache,
+ MvccCoordinator mvccCrd,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
@@ -272,6 +282,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
lastTopChangeVer = exchTopVer;
this.discoCache = discoCache;
+ this.mvccCrd = mvccCrd;
}
finally {
lock.writeLock().unlock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsStateValidator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsStateValidator.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsStateValidator.java
index 866c513..7faf0ff 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsStateValidator.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsStateValidator.java
@@ -32,6 +32,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.lang.IgniteProductVersion;
@@ -93,11 +94,13 @@ public class GridDhtPartitionsStateValidator {
ignoringNodes.add(id);
}
- // Validate cache sizes.
- result = validatePartitionsSizes(top, messages, ignoringNodes);
+ if (!MvccUtils.mvccEnabled(cctx.kernalContext())) { // TODO: Remove "if" clause in IGNITE-8149.
+ // Validate cache sizes.
+ result = validatePartitionsSizes(top, messages, ignoringNodes);
- if (!result.isEmpty())
- throw new IgniteCheckedException("Partitions cache sizes are inconsistent for " + fold(topVer, result));
+ if (!result.isEmpty())
+ throw new IgniteCheckedException("Partitions cache sizes are inconsistent for " + fold(topVer, result));
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsUpdateCountersMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsUpdateCountersMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsUpdateCountersMap.java
new file mode 100644
index 0000000..7a345d1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsUpdateCountersMap.java
@@ -0,0 +1,119 @@
+/*
+ * 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.HashMap;
+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;
+
+/**
+ * Partitions update counters message.
+ */
+public class GridDhtPartitionsUpdateCountersMap implements Message {
+ /** */
+ private static final long serialVersionUID = -4599730112233297219L;
+
+ /** Map of update counters made by this tx. Mapping: partId -> updCntr. */
+ @GridDirectMap(keyType = Integer.class, valueType = Long.class)
+ private Map<Integer, Long> updCntrs;
+
+ /**
+ *
+ */
+ public GridDhtPartitionsUpdateCountersMap() {
+ updCntrs = new HashMap<>();
+ }
+
+ /**
+ * @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(GridDhtPartitionsUpdateCountersMap.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/00bb4d4d/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 f93e6a3..be1c7e2 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
@@ -28,7 +28,9 @@ import java.util.ListIterator;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
@@ -55,8 +57,16 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLock
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryEnlistFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryEnlistRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryEnlistResponse;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxRemote;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearUnlockRequest;
+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.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
@@ -76,6 +86,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.thread.IgniteThread;
import org.apache.ignite.transactions.TransactionIsolation;
import org.jetbrains.annotations.Nullable;
@@ -83,7 +94,9 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOOP;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
import static org.apache.ignite.transactions.TransactionState.COMMITTING;
/**
@@ -164,6 +177,18 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
}
});
+ ctx.io().addCacheHandler(ctx.cacheId(), GridNearTxQueryEnlistRequest.class, new CI2<UUID, GridNearTxQueryEnlistRequest>() {
+ @Override public void apply(UUID nodeId, GridNearTxQueryEnlistRequest req) {
+ processNearTxQueryEnlistRequest(nodeId, req);
+ }
+ });
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridNearTxQueryEnlistResponse.class, new CI2<UUID, GridNearTxQueryEnlistResponse>() {
+ @Override public void apply(UUID nodeId, GridNearTxQueryEnlistResponse req) {
+ processNearEnlistResponse(nodeId, req);
+ }
+ });
+
ctx.io().addCacheHandler(ctx.cacheId(), GridDhtForceKeysRequest.class,
new MessageHandler<GridDhtForceKeysRequest>() {
@Override public void onMessage(ClusterNode node, GridDhtForceKeysRequest msg) {
@@ -177,6 +202,41 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
processForceKeyResponse(node, msg);
}
});
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridNearTxQueryResultsEnlistRequest.class,
+ new CI2<UUID, GridNearTxQueryResultsEnlistRequest>() {
+ @Override public void apply(UUID nodeId, GridNearTxQueryResultsEnlistRequest req) {
+ processNearTxQueryResultsEnlistRequest(nodeId, req);
+ }
+ });
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridNearTxQueryResultsEnlistResponse.class,
+ new CI2<UUID, GridNearTxQueryResultsEnlistResponse>() {
+ @Override public void apply(UUID nodeId, GridNearTxQueryResultsEnlistResponse req) {
+ processNearTxEnlistResponse(nodeId, req);
+ }
+ });
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridDhtTxQueryEnlistRequest.class,
+ new CI2<UUID, GridDhtTxQueryEnlistRequest>() {
+ @Override public void apply(UUID nodeId, GridDhtTxQueryEnlistRequest msg) {
+ processDhtTxQueryEnlistRequest(nodeId, msg, false);
+ }
+ });
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridDhtTxQueryFirstEnlistRequest.class,
+ new CI2<UUID, GridDhtTxQueryEnlistRequest>() {
+ @Override public void apply(UUID nodeId, GridDhtTxQueryEnlistRequest msg) {
+ processDhtTxQueryEnlistRequest(nodeId, msg, true);
+ }
+ });
+
+ ctx.io().addCacheHandler(ctx.cacheId(), GridDhtTxQueryEnlistResponse.class,
+ new CI2<UUID, GridDhtTxQueryEnlistResponse>() {
+ @Override public void apply(UUID nodeId, GridDhtTxQueryEnlistResponse msg) {
+ processDhtTxQueryEnlistResponse(nodeId, msg);
+ }
+ });
}
/** {@inheritDoc} */
@@ -627,6 +687,85 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
* @param nodeId Node ID.
* @param req Request.
*/
+ private void processNearTxQueryEnlistRequest(UUID nodeId, final GridNearTxQueryEnlistRequest req) {
+ assert nodeId != null;
+ assert req != null;
+
+ ClusterNode nearNode = ctx.discovery().node(nodeId);
+
+ GridDhtTxLocal tx;
+
+ try {
+ tx = initTxTopologyVersion(nodeId,
+ nearNode,
+ req.version(),
+ req.futureId(),
+ req.miniId(),
+ req.firstClientRequest(),
+ req.topologyVersion(),
+ req.threadId(),
+ req.txTimeout(),
+ req.subjectId(),
+ req.taskNameHash());
+ }
+ catch (IgniteCheckedException | IgniteException ex) {
+ GridNearTxQueryEnlistResponse res = new GridNearTxQueryEnlistResponse(req.cacheId(),
+ req.futureId(),
+ req.miniId(),
+ req.version(),
+ ex);
+
+ try {
+ ctx.io().send(nearNode, res, ctx.ioPolicy());
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send near enlist response [" +
+ "txId=" + req.version() +
+ ", node=" + nodeId +
+ ", res=" + res + ']', e);
+ }
+
+ return;
+ }
+
+ GridDhtTxQueryEnlistFuture fut = new GridDhtTxQueryEnlistFuture(
+ nodeId,
+ req.version(),
+ req.mvccSnapshot(),
+ req.threadId(),
+ req.futureId(),
+ req.miniId(),
+ tx,
+ req.cacheIds(),
+ req.partitions(),
+ req.schemaName(),
+ req.query(),
+ req.parameters(),
+ req.flags(),
+ req.pageSize(),
+ req.timeout(),
+ ctx);
+
+ fut.listen(NearTxQueryEnlistResultHandler.instance());
+
+ fut.init();
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param res Response.
+ */
+ private void processNearEnlistResponse(UUID nodeId, final GridNearTxQueryEnlistResponse res) {
+ GridNearTxQueryEnlistFuture fut = (GridNearTxQueryEnlistFuture)ctx.mvcc().versionedFuture(res.version(), res.futureId());
+
+ if (fut != null)
+ fut.onResult(nodeId, res);
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param req Request.
+ */
private void processNearLockRequest(UUID nodeId, GridNearLockRequest req) {
assert ctx.affinityNode();
assert nodeId != null;
@@ -1276,7 +1415,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
CacheObject val = null;
- if (ret)
+ if (ret) {
val = e.innerGet(
null,
tx,
@@ -1287,7 +1426,9 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
null,
tx != null ? tx.resolveTaskName() : null,
null,
- req.keepBinary());
+ req.keepBinary(),
+ null); // TODO IGNITE-7371
+ }
assert e.lockedBy(mappedVer) || ctx.mvcc().isRemoved(e.context(), mappedVer) :
"Entry does not own lock for tx [locNodeId=" + ctx.localNodeId() +
@@ -1380,6 +1521,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
U.error(log, "Failed to acquire lock for request: " + req, err);
try {
+ // TODO Async rollback
// Don't send reply message to this node or if lock was cancelled or tx was rolled back asynchronously.
if (!nearNode.id().equals(ctx.nodeId()) && !X.hasCause(err, GridDistributedLockCancelledException.class) &&
!X.hasCause(err, IgniteTxRollbackCheckedException.class)) {
@@ -1576,7 +1718,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
* @param nodes Nodes.
* @param map Map.
*/
- @SuppressWarnings( {"MismatchedQueryAndUpdateOfCollection"})
+ @SuppressWarnings({"MismatchedQueryAndUpdateOfCollection"})
private void map(GridCacheEntryEx entry,
@Nullable Iterable<? extends ClusterNode> nodes,
Map<ClusterNode, List<KeyCacheObject>> map) {
@@ -1785,4 +1927,321 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
if (nearEntry != null)
nearEntry.markObsolete(ctx.versions().next());
}
+
+ /**
+ * @param nodeId Node ID.
+ * @param req Request.
+ */
+ private void processNearTxQueryResultsEnlistRequest(UUID nodeId, final GridNearTxQueryResultsEnlistRequest req) {
+ assert nodeId != null;
+ assert req != null;
+
+ ClusterNode nearNode = ctx.discovery().node(nodeId);
+
+ GridDhtTxLocal tx;
+
+ try {
+ tx = initTxTopologyVersion(nodeId,
+ nearNode,
+ req.version(),
+ req.futureId(),
+ req.miniId(),
+ req.firstClientRequest(),
+ req.topologyVersion(),
+ req.threadId(),
+ req.txTimeout(),
+ req.subjectId(),
+ req.taskNameHash());
+ }
+ catch (IgniteCheckedException | IgniteException ex) {
+ GridNearTxQueryResultsEnlistResponse res = new GridNearTxQueryResultsEnlistResponse(req.cacheId(),
+ req.futureId(),
+ req.miniId(),
+ req.version(),
+ ex);
+
+ try {
+ ctx.io().send(nearNode, res, ctx.ioPolicy());
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send near enlist response [" +
+ "txId=" + req.version() +
+ ", node=" + nodeId +
+ ", res=" + res + ']', e);
+ }
+
+ return;
+ }
+
+ GridDhtTxQueryResultsEnlistFuture fut = new GridDhtTxQueryResultsEnlistFuture(
+ nodeId,
+ req.version(),
+ req.mvccSnapshot(),
+ req.threadId(),
+ req.futureId(),
+ req.miniId(),
+ tx,
+ req.timeout(),
+ ctx,
+ req.rows(),
+ req.operation());
+
+ fut.listen(NearTxQueryEnlistResultHandler.instance());
+
+ fut.init();
+ }
+
+ /**
+ * @param nodeId Near node id.
+ * @param nearNode Near node.
+ * @param nearLockVer Near lock version.
+ * @param nearFutId Near future id.
+ * @param nearMiniId Near mini-future id.
+ * @param firstClientReq First client request flag.
+ * @param topVer Topology version.
+ * @param nearThreadId Near node thread id.
+ * @param timeout Timeout.
+ * @param txSubjectId Transaction subject id.
+ * @param txTaskNameHash Transaction task name hash.
+ * @return Transaction.
+ */
+ public GridDhtTxLocal initTxTopologyVersion(UUID nodeId,
+ ClusterNode nearNode,
+ GridCacheVersion nearLockVer,
+ IgniteUuid nearFutId,
+ int nearMiniId,
+ boolean firstClientReq,
+ AffinityTopologyVersion topVer,
+ long nearThreadId,
+ long timeout,
+ UUID txSubjectId,
+ int txTaskNameHash) throws IgniteException, IgniteCheckedException {
+
+ assert ctx.affinityNode();
+
+ if (txLockMsgLog.isDebugEnabled()) {
+ txLockMsgLog.debug("Received near enlist request [txId=" + nearLockVer +
+ ", node=" + nodeId + ']');
+ }
+
+ if (nearNode == null) {
+ U.warn(txLockMsgLog, "Received near enlist request from unknown node (will ignore) [txId=" + nearLockVer +
+ ", node=" + nodeId + ']');
+
+ return null;
+ }
+
+ GridDhtTxLocal tx = null;
+
+ GridCacheVersion dhtVer = ctx.tm().mappedVersion(nearLockVer);
+
+ if (dhtVer != null)
+ tx = ctx.tm().tx(dhtVer);
+
+ GridDhtPartitionTopology top = null;
+
+ if (tx == null) {
+ if (firstClientReq) {
+ assert nearNode.isClient();
+
+ top = topology();
+
+ top.readLock();
+
+ GridDhtTopologyFuture topFut = top.topologyVersionFuture();
+
+ if (!topFut.isDone() || !topFut.topologyVersion().equals(topVer)) {
+ // TODO IGNITE-7164 Wait for topology change, remap client TX in case affinity was changed.
+ top.readUnlock();
+
+ throw new ClusterTopologyException("Topology was changed. Please retry on stable topology.");
+ }
+ }
+
+ try {
+ tx = new GridDhtTxLocal(
+ ctx.shared(),
+ topVer,
+ nearNode.id(),
+ nearLockVer,
+ nearFutId,
+ nearMiniId,
+ nearThreadId,
+ false,
+ false,
+ ctx.systemTx(),
+ false,
+ ctx.ioPolicy(),
+ PESSIMISTIC,
+ REPEATABLE_READ,
+ timeout,
+ false,
+ false,
+ false,
+ -1,
+ null,
+ txSubjectId,
+ txTaskNameHash);
+
+ // if (req.syncCommit())
+ tx.syncMode(FULL_SYNC);
+
+ tx = ctx.tm().onCreated(null, tx);
+
+ if (tx == null || !tx.init()) {
+ String msg = "Failed to acquire lock (transaction has been completed): " +
+ nearLockVer;
+
+ U.warn(log, msg);
+
+ try {
+ if (tx != null)
+ tx.rollbackDhtLocal();
+ }
+ catch (IgniteCheckedException ex) {
+ U.error(log, "Failed to rollback the transaction: " + tx, ex);
+ }
+
+ throw new IgniteCheckedException(msg);
+ }
+
+ tx.topologyVersion(topVer);
+ }
+ finally {
+ if (top != null)
+ top.readUnlock();
+ }
+ }
+
+ ctx.tm().txContext(tx);
+
+ return tx;
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param res Response.
+ */
+ private void processNearTxEnlistResponse(UUID nodeId, final GridNearTxQueryResultsEnlistResponse res) {
+ GridNearTxQueryResultsEnlistFuture fut = (GridNearTxQueryResultsEnlistFuture)
+ ctx.mvcc().versionedFuture(res.version(), res.futureId());
+
+ if (fut != null)
+ fut.onResult(nodeId, res);
+ }
+
+ /**
+ * @param primary Primary node.
+ * @param req Request.
+ * @param e Error.
+ */
+ private void onError(UUID primary, GridDhtTxQueryEnlistRequest req, Throwable e) {
+ GridDhtTxQueryEnlistResponse res = new GridDhtTxQueryEnlistResponse(ctx.cacheId(),
+ req.dhtFutureId(),
+ req.batchId(),
+ e);
+
+ try {
+ ctx.io().send(primary, res, ctx.ioPolicy());
+ }
+ catch (IgniteCheckedException ioEx) {
+ U.error(log, "Failed to send DHT enlist reply to primary node [node: " + primary + ", req=" + req +
+ ']', ioEx);
+ }
+ }
+
+ /**
+ * @param primary Primary node.
+ * @param req Message.
+ * @param first Flag if this is a first request in current operation.
+ */
+ private void processDhtTxQueryEnlistRequest(UUID primary, GridDhtTxQueryEnlistRequest req, boolean first) {
+ try {
+ assert req.version() != null && req.op() != null;
+
+ GridDhtTxRemote tx = ctx.tm().tx(req.version());
+
+ if (tx == null) {
+ if (!first)
+ throw new IgniteCheckedException("Can not find a transaction for version [version="
+ + req.version() + ']');
+
+ GridDhtTxQueryFirstEnlistRequest req0 = (GridDhtTxQueryFirstEnlistRequest)req;
+
+ tx = new GridDhtTxRemote(ctx.shared(),
+ req0.nearNodeId(),
+ req0.dhtFutureId(),
+ primary,
+ req0.nearXidVersion(),
+ req0.topologyVersion(),
+ req0.version(),
+ null,
+ ctx.systemTx(),
+ ctx.ioPolicy(),
+ PESSIMISTIC,
+ REPEATABLE_READ,
+ false,
+ req0.timeout(),
+ -1,
+ req0.subjectId(),
+ req0.taskNameHash(),
+ false);
+
+ tx.mvccSnapshot(new MvccSnapshotWithoutTxs(req0.coordinatorVersion(), req0.counter(),
+ MVCC_OP_COUNTER_NA, req0.cleanupVersion()));
+
+ tx = ctx.tm().onCreated(null, tx);
+
+ if (tx == null || !ctx.tm().onStarted(tx)) {
+ throw new IgniteTxRollbackCheckedException("Failed to update backup " +
+ "(transaction has been completed): " + req0.version());
+ }
+ }
+
+ assert tx != null;
+
+ MvccSnapshot s0 = tx.mvccSnapshot();
+
+ MvccSnapshot snapshot = new MvccSnapshotWithoutTxs(s0.coordinatorVersion(), s0.counter(),
+ req.operationCounter(), s0.cleanupVersion());
+
+ tx.mvccEnlistBatch(ctx, req.op(), req.keys(), req.values(), snapshot, req.updateCounters());
+
+ GridDhtTxQueryEnlistResponse res = new GridDhtTxQueryEnlistResponse(req.cacheId(),
+ req.dhtFutureId(),
+ req.batchId(),
+ null);
+
+ try {
+ ctx.io().send(primary, res, ctx.ioPolicy());
+ }
+ catch (IgniteCheckedException ioEx) {
+ U.error(log, "Failed to send DHT enlist reply to primary node [node: " + primary + ", req=" +
+ req + ']', ioEx);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ onError(primary, req, e);
+ }
+ }
+
+ /**
+ * @param backup Backup node.
+ * @param res Response message.
+ */
+ private void processDhtTxQueryEnlistResponse(UUID backup, GridDhtTxQueryEnlistResponse res) {
+ GridDhtTxAbstractEnlistFuture fut = (GridDhtTxAbstractEnlistFuture)
+ ctx.mvcc().future(res.futureId());
+
+ if (fut == null) {
+ U.warn(log, "Received dht enlist response for unknown future [futId=" + res.futureId() +
+ ", batchId=" + res.batchId() +
+ ", node=" + backup + ']');
+
+ return;
+ }
+
+ fut.onResult(backup, res);
+ }
+
}