ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [33/46] incubator-ignite git commit: GG-9141 - Renaming.
Date Sun, 21 Dec 2014 23:04:26 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cd46f567/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
new file mode 100644
index 0000000..7c8584c
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -0,0 +1,167 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.kernal.processors.cache.transactions;
+
+import org.apache.ignite.*;
+import org.apache.ignite.lang.*;
+import org.gridgain.grid.cache.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.gridgain.grid.kernal.processors.cache.dr.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+
+/**
+ * Local transaction API.
+ */
+public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K, V> {
+    /**
+     * @return Minimum version involved in transaction.
+     */
+    public GridCacheVersion minVersion();
+
+    /**
+     * @return Future for this transaction.
+     */
+    public IgniteFuture<IgniteTxEx<K, V>> future();
+
+    /**
+     * @return Commit error.
+     */
+    @Nullable public Throwable commitError();
+
+    /**
+     * @param e Commit error.
+     */
+    public void commitError(Throwable e);
+
+    /**
+     * @throws IgniteCheckedException If commit failed.
+     */
+    public void userCommit() throws IgniteCheckedException;
+
+    /**
+     * @throws IgniteCheckedException If rollback failed.
+     */
+    public void userRollback() throws IgniteCheckedException;
+
+    /**
+     * @return Group lock entry if this is a group-lock transaction.
+     */
+    @Nullable public IgniteTxEntry<K, V> groupLockEntry();
+
+    /**
+     * @param keys Keys to get.
+     * @param cached Cached entry if this method is called from entry wrapper.
+     *      Cached entry is passed if and only if there is only one key in collection of keys.
+     * @param deserializePortable Deserialize portable flag.
+     * @param filter Entry filter.
+     * @return Future for this get.
+     */
+    public IgniteFuture<Map<K, V>> getAllAsync(
+        GridCacheContext<K, V> cacheCtx,
+        Collection<? extends K> keys,
+        @Nullable GridCacheEntryEx<K, V> cached,
+        boolean deserializePortable,
+        IgnitePredicate<GridCacheEntry<K, V>>[] filter);
+
+    /**
+     * @param map Map to put.
+     * @param retval Flag indicating whether a value should be returned.
+     * @param cached Cached entry, if any. Will be provided only if map has size 1.
+     * @param filter Filter.
+     * @param ttl Time to live for entry. If negative, leave unchanged.
+     * @return Future for put operation.
+     */
+    public IgniteFuture<GridCacheReturn<V>> putAllAsync(
+        GridCacheContext<K, V> cacheCtx,
+        Map<? extends K, ? extends V> map,
+        boolean retval,
+        @Nullable GridCacheEntryEx<K, V> cached,
+        long ttl,
+        IgnitePredicate<GridCacheEntry<K, V>>[] filter);
+
+    /**
+     * @param map Map to put.
+     * @return Transform operation future.
+     */
+    public IgniteFuture<GridCacheReturn<V>> transformAllAsync(
+        GridCacheContext<K, V> cacheCtx,
+        @Nullable Map<? extends K, ? extends IgniteClosure<V, V>> map,
+        boolean retval,
+        @Nullable GridCacheEntryEx<K, V> cached,
+        long ttl);
+
+    /**
+     * @param keys Keys to remove.
+     * @param retval Flag indicating whether a value should be returned.
+     * @param cached Cached entry, if any. Will be provided only if size of keys collection is 1.
+     * @param filter Filter.
+     * @return Future for asynchronous remove.
+     */
+    public IgniteFuture<GridCacheReturn<V>> removeAllAsync(
+        GridCacheContext<K, V> cacheCtx,
+        Collection<? extends K> keys,
+        @Nullable GridCacheEntryEx<K, V> cached,
+        boolean retval,
+        IgnitePredicate<GridCacheEntry<K, V>>[] filter);
+
+    /**
+     * @param drMap DR map to put.
+     * @return Future for DR put operation.
+     */
+    public IgniteFuture<?> putAllDrAsync(
+        GridCacheContext<K, V> cacheCtx,
+        Map<? extends K, GridCacheDrInfo<V>> drMap);
+
+    /**
+     * @param drMap DR map.
+     * @return Future for asynchronous remove.
+     */
+    public IgniteFuture<?> removeAllDrAsync(
+        GridCacheContext<K, V> cacheCtx,
+        Map<? extends K, GridCacheVersion> drMap);
+
+    /**
+     * Performs keys locking for affinity-based group lock transactions.
+     *
+     * @param keys Keys to lock.
+     * @return Lock future.
+     */
+    public IgniteFuture<?> groupLockAsync(GridCacheContext<K, V> cacheCtx, Collection<K> keys);
+
+    /**
+     * @return {@code True} if keys from the same partition are allowed to be enlisted in group-lock transaction.
+     */
+    public boolean partitionLock();
+
+    /**
+     * Finishes transaction (either commit or rollback).
+     *
+     * @param commit {@code True} if commit, {@code false} if rollback.
+     * @return {@code True} if state has been changed.
+     * @throws IgniteCheckedException If finish failed.
+     */
+    public boolean finish(boolean commit) throws IgniteCheckedException;
+
+    /**
+     * @param async if {@code True}, then loading will happen in a separate thread.
+     * @param keys Keys.
+     * @param c Closure.
+     * @param deserializePortable Deserialize portable flag.
+     * @return Future with {@code True} value if loading took place.
+     */
+    public IgniteFuture<Boolean> loadMissing(
+        GridCacheContext<K, V> cacheCtx,
+        boolean async,
+        Collection<? extends K> keys,
+        boolean deserializePortable,
+        IgniteBiInClosure<K, V> c);
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cd46f567/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxManager.java
new file mode 100644
index 0000000..a176ff1
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxManager.java
@@ -0,0 +1,2213 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.kernal.processors.cache.transactions;
+
+import org.apache.ignite.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.transactions.*;
+import org.gridgain.grid.kernal.managers.eventstorage.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.gridgain.grid.kernal.processors.cache.distributed.*;
+import org.gridgain.grid.kernal.processors.cache.distributed.dht.*;
+import org.gridgain.grid.kernal.processors.cache.distributed.near.*;
+import org.gridgain.grid.kernal.processors.timeout.*;
+import org.gridgain.grid.util.*;
+import org.gridgain.grid.util.future.*;
+import org.gridgain.grid.util.lang.*;
+import org.gridgain.grid.util.typedef.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.jdk8.backport.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import static org.apache.ignite.IgniteSystemProperties.*;
+import static org.apache.ignite.events.IgniteEventType.*;
+import static org.apache.ignite.transactions.IgniteTxConcurrency.*;
+import static org.apache.ignite.transactions.IgniteTxState.*;
+import static org.gridgain.grid.kernal.processors.cache.transactions.IgniteTxEx.FinalizationStatus.*;
+import static org.gridgain.grid.kernal.processors.cache.GridCacheUtils.*;
+import static org.gridgain.grid.util.GridConcurrentFactory.*;
+
+/**
+ * Cache transaction manager.
+ */
+public class IgniteTxManager<K, V> extends GridCacheSharedManagerAdapter<K, V> {
+    /** Default maximum number of transactions that have completed. */
+    private static final int DFLT_MAX_COMPLETED_TX_CNT = 262144; // 2^18
+
+    /** Slow tx warn timeout (initialized to 0). */
+    private static final int SLOW_TX_WARN_TIMEOUT = Integer.getInteger(GG_SLOW_TX_WARN_TIMEOUT, 0);
+
+    /** Tx salvage timeout (default 3s). */
+    private static final int TX_SALVAGE_TIMEOUT = Integer.getInteger(GG_TX_SALVAGE_TIMEOUT, 100);
+
+    /** Committing transactions. */
+    private final ThreadLocal<IgniteTxEx> threadCtx = new GridThreadLocalEx<>();
+
+    /** Per-thread transaction map. */
+    private final ConcurrentMap<Long, IgniteTxEx<K, V>> threadMap = newMap();
+
+    /** Per-ID map. */
+    private final ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> idMap = newMap();
+
+    /** Per-ID map for near transactions. */
+    private final ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> nearIdMap = newMap();
+
+    /** TX handler. */
+    private IgniteTxHandler<K, V> txHandler;
+
+    /** All transactions. */
+    private final Queue<IgniteTxEx<K, V>> committedQ = new ConcurrentLinkedDeque8<>();
+
+    /** Preparing transactions. */
+    private final Queue<IgniteTxEx<K, V>> prepareQ = new ConcurrentLinkedDeque8<>();
+
+    /** Minimum start version. */
+    private final ConcurrentNavigableMap<GridCacheVersion, AtomicInt> startVerCnts =
+        new ConcurrentSkipListMap<>();
+
+    /** Committed local transactions. */
+    private final GridBoundedConcurrentOrderedSet<GridCacheVersion> committedVers =
+        new GridBoundedConcurrentOrderedSet<>(Integer.getInteger(GG_MAX_COMPLETED_TX_COUNT, DFLT_MAX_COMPLETED_TX_CNT));
+
+    /** Rolled back local transactions. */
+    private final NavigableSet<GridCacheVersion> rolledbackVers =
+        new GridBoundedConcurrentOrderedSet<>(Integer.getInteger(GG_MAX_COMPLETED_TX_COUNT, DFLT_MAX_COMPLETED_TX_CNT));
+
+    /** Pessimistic commit buffer. */
+    private GridCacheTxCommitBuffer<K, V> pessimisticRecoveryBuf;
+
+    /** Transaction synchronizations. */
+    private final Collection<IgniteTxSynchronization> syncs =
+        new GridConcurrentHashSet<>();
+
+    /** Transaction finish synchronizer. */
+    private GridCacheTxFinishSync<K, V> txFinishSync;
+
+    /** For test purposes only. */
+    private boolean finishSyncDisabled;
+
+    /** Slow tx warn timeout. */
+    private int slowTxWarnTimeout = SLOW_TX_WARN_TIMEOUT;
+
+    /**
+     * Near version to DHT version map. Note that we initialize to 5K size from get go,
+     * to avoid future map resizings.
+     */
+    private final ConcurrentMap<GridCacheVersion, GridCacheVersion> mappedVers =
+        new ConcurrentHashMap8<>(5120);
+
+    /** {@inheritDoc} */
+    @Override protected void onKernalStart0() {
+        cctx.gridEvents().addLocalEventListener(
+            new GridLocalEventListener() {
+                @Override public void onEvent(IgniteEvent evt) {
+                    assert evt instanceof IgniteDiscoveryEvent;
+                    assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT;
+
+                    IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+
+                    cctx.time().addTimeoutObject(new NodeFailureTimeoutObject(discoEvt.eventNode().id()));
+
+                    if (txFinishSync != null)
+                        txFinishSync.onNodeLeft(discoEvt.eventNode().id());
+                }
+            },
+            EVT_NODE_FAILED, EVT_NODE_LEFT);
+
+        for (IgniteTxEx<K, V> tx : idMap.values()) {
+            if ((!tx.local() || tx.dht()) && !cctx.discovery().aliveAll(tx.masterNodeIds())) {
+                if (log.isDebugEnabled())
+                    log.debug("Remaining transaction from left node: " + tx);
+
+                salvageTx(tx, true, USER_FINISH);
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void start0() throws IgniteCheckedException {
+        pessimisticRecoveryBuf = new GridCachePerThreadTxCommitBuffer<>(cctx);
+
+        txFinishSync = new GridCacheTxFinishSync<>(cctx);
+
+        txHandler = new IgniteTxHandler<>(cctx);
+    }
+
+    /**
+     * @return TX handler.
+     */
+    public IgniteTxHandler<K, V> txHandler() {
+        return txHandler;
+    }
+
+    /**
+     * Invalidates transaction.
+     *
+     * @param tx Transaction.
+     * @return {@code True} if transaction was salvaged by this call.
+     */
+    public boolean salvageTx(IgniteTxEx<K, V> tx) {
+        return salvageTx(tx, false, USER_FINISH);
+    }
+
+    /**
+     * Invalidates transaction.
+     *
+     * @param tx Transaction.
+     * @param warn {@code True} if warning should be logged.
+     * @param status Finalization status.
+     * @return {@code True} if transaction was salvaged by this call.
+     */
+    private boolean salvageTx(IgniteTxEx<K, V> tx, boolean warn, IgniteTxEx.FinalizationStatus status) {
+        assert tx != null;
+
+        IgniteTxState state = tx.state();
+
+        if (state == ACTIVE || state == PREPARING || state == PREPARED) {
+            try {
+                if (!tx.markFinalizing(status)) {
+                    if (log.isDebugEnabled())
+                        log.debug("Will not try to commit invalidate transaction (could not mark finalized): " + tx);
+
+                    return false;
+                }
+
+                tx.systemInvalidate(true);
+
+                tx.prepare();
+
+                if (tx.state() == PREPARING) {
+                    if (log.isDebugEnabled())
+                        log.debug("Ignoring transaction in PREPARING state as it is currently handled " +
+                            "by another thread: " + tx);
+
+                    return false;
+                }
+
+                if (tx instanceof IgniteTxRemoteEx) {
+                    IgniteTxRemoteEx<K, V> rmtTx = (IgniteTxRemoteEx<K, V>)tx;
+
+                    rmtTx.doneRemote(tx.xidVersion(), Collections.<GridCacheVersion>emptyList(),
+                        Collections.<GridCacheVersion>emptyList(), Collections.<GridCacheVersion>emptyList());
+                }
+
+                tx.commit();
+
+                if (warn) {
+                    // This print out cannot print any peer-deployed entity either
+                    // directly or indirectly.
+                    U.warn(log, "Invalidated transaction because originating node either " +
+                        "crashed or left grid: " + CU.txString(tx));
+                }
+            }
+            catch (IgniteTxOptimisticException ignore) {
+                if (log.isDebugEnabled())
+                    log.debug("Optimistic failure while invalidating transaction (will rollback): " +
+                        tx.xidVersion());
+
+                try {
+                    tx.rollback();
+                }
+                catch (IgniteCheckedException e) {
+                    U.error(log, "Failed to rollback transaction: " + tx.xidVersion(), e);
+                }
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to invalidate transaction: " + tx, e);
+            }
+        }
+        else if (state == MARKED_ROLLBACK) {
+            try {
+                tx.rollback();
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to rollback transaction: " + tx.xidVersion(), e);
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * Prints out memory stats to standard out.
+     * <p>
+     * USE ONLY FOR MEMORY PROFILING DURING TESTS.
+     */
+    @Override public void printMemoryStats() {
+        IgniteTxEx<K, V> firstTx = committedQ.peek();
+
+        int committedSize = committedQ.size();
+
+        Map.Entry<GridCacheVersion, AtomicInt> startVerEntry = startVerCnts.firstEntry();
+
+        GridCacheVersion minStartVer = null;
+        long dur = 0;
+
+        if (committedSize > 3000) {
+            minStartVer = new GridCacheVersion(Integer.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE, Integer.MAX_VALUE, 0);
+
+            IgniteTxEx<K, V> stuck = null;
+
+            for (IgniteTxEx<K, V> tx : txs())
+                if (tx.startVersion().isLess(minStartVer)) {
+                    minStartVer = tx.startVersion();
+                    dur = U.currentTimeMillis() - tx.startTime();
+
+                    stuck = tx;
+                }
+
+            X.println("Stuck transaction: " + stuck);
+        }
+
+        X.println(">>> ");
+        X.println(">>> Transaction manager memory stats [grid=" + cctx.gridName() + ']');
+        X.println(">>>   threadMapSize: " + threadMap.size());
+        X.println(">>>   idMap [size=" + idMap.size() + ", minStartVer=" + minStartVer + ", dur=" + dur + "ms]");
+        X.println(">>>   committedQueue [size=" + committedSize +
+            ", firstStartVersion=" + (firstTx == null ? "null" : firstTx.startVersion()) +
+            ", firstEndVersion=" + (firstTx == null ? "null" : firstTx.endVersion()) + ']');
+        X.println(">>>   prepareQueueSize: " + prepareQ.size());
+        X.println(">>>   startVerCntsSize [size=" + startVerCnts.size() +
+            ", firstVer=" + startVerEntry + ']');
+        X.println(">>>   committedVersSize: " + committedVers.size());
+        X.println(">>>   rolledbackVersSize: " + rolledbackVers.size());
+
+        if (pessimisticRecoveryBuf != null)
+            X.println(">>>   pessimsticCommitBufSize: " + pessimisticRecoveryBuf.size());
+    }
+
+    /**
+     * @return Thread map size.
+     */
+    public int threadMapSize() {
+        return threadMap.size();
+    }
+
+    /**
+     * @return ID map size.
+     */
+    public int idMapSize() {
+        return idMap.size();
+    }
+
+    /**
+     * @return Committed queue size.
+     */
+    public int commitQueueSize() {
+        return committedQ.size();
+    }
+
+    /**
+     * @return Prepare queue size.
+     */
+    public int prepareQueueSize() {
+        return prepareQ.size();
+    }
+
+    /**
+     * @return Start version counts.
+     */
+    public int startVersionCountsSize() {
+        return startVerCnts.size();
+    }
+
+    /**
+     * @return Committed versions size.
+     */
+    public int committedVersionsSize() {
+        return committedVers.size();
+    }
+
+    /**
+     * @return Rolled back versions size.
+     */
+    public int rolledbackVersionsSize() {
+        return rolledbackVers.size();
+    }
+
+    /**
+     *
+     * @param tx Transaction to check.
+     * @return {@code True} if transaction has been committed or rolled back,
+     *      {@code false} otherwise.
+     */
+    public boolean isCompleted(IgniteTxEx<K, V> tx) {
+        return committedVers.contains(tx.xidVersion()) || rolledbackVers.contains(tx.xidVersion());
+    }
+
+    /**
+     * @param implicit {@code True} if transaction is implicit.
+     * @param implicitSingle Implicit-with-single-key flag.
+     * @param concurrency Concurrency.
+     * @param isolation Isolation.
+     * @param timeout transaction timeout.
+     * @param txSize Expected transaction size.
+     * @param grpLockKey Group lock key if this is a group-lock transaction.
+     * @param partLock {@code True} if partition is locked.
+     * @return New transaction.
+     */
+    public IgniteTxLocalAdapter<K, V> newTx(
+        boolean implicit,
+        boolean implicitSingle,
+        boolean sys,
+        IgniteTxConcurrency concurrency,
+        IgniteTxIsolation isolation,
+        long timeout,
+        boolean invalidate,
+        boolean storeEnabled,
+        int txSize,
+        @Nullable IgniteTxKey grpLockKey,
+        boolean partLock) {
+        UUID subjId = null; // TODO GG-9141 how to get subj ID?
+
+        int taskNameHash = cctx.kernalContext().job().currentTaskNameHash();
+
+        GridNearTxLocal<K, V> tx = new GridNearTxLocal<>(
+            cctx,
+            implicit,
+            implicitSingle,
+            sys,
+            concurrency,
+            isolation,
+            timeout,
+            invalidate,
+            storeEnabled,
+            txSize,
+            grpLockKey,
+            partLock,
+            subjId,
+            taskNameHash);
+
+        return onCreated(tx);
+    }
+
+    /**
+     * @param tx Created transaction.
+     * @return Started transaction.
+     */
+    @Nullable public <T extends IgniteTxEx<K, V>> T onCreated(T tx) {
+        ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> txIdMap = transactionMap(tx);
+
+        // Start clean.
+        txContextReset();
+
+        if (isCompleted(tx)) {
+            if (log.isDebugEnabled())
+                log.debug("Attempt to create a completed transaction (will ignore): " + tx);
+
+            return null;
+        }
+
+        IgniteTxEx<K, V> t;
+
+        if ((t = txIdMap.putIfAbsent(tx.xidVersion(), tx)) == null) {
+            // Add both, explicit and implicit transactions.
+            // Do not add remote and dht local transactions as remote node may have the same thread ID
+            // and overwrite local transaction.
+            if (tx.local() && !tx.dht())
+                threadMap.put(tx.threadId(), tx);
+
+            // Handle mapped versions.
+            if (tx instanceof GridCacheMappedVersion) {
+                GridCacheMappedVersion mapped = (GridCacheMappedVersion)tx;
+
+                GridCacheVersion from = mapped.mappedVersion();
+
+                if (from != null)
+                    mappedVers.put(from, tx.xidVersion());
+
+                if (log.isDebugEnabled())
+                    log.debug("Added transaction version mapping [from=" + from + ", to=" + tx.xidVersion() +
+                        ", tx=" + tx + ']');
+            }
+        }
+        else {
+            if (log.isDebugEnabled())
+                log.debug("Attempt to create an existing transaction (will ignore) [newTx=" + tx + ", existingTx=" +
+                    t + ']');
+
+            return null;
+        }
+
+        if (cctx.txConfig().isTxSerializableEnabled()) {
+            AtomicInt next = new AtomicInt(1);
+
+            boolean loop = true;
+
+            while (loop) {
+                AtomicInt prev = startVerCnts.putIfAbsent(tx.startVersion(), next);
+
+                if (prev == null)
+                    break; // Put succeeded - exit.
+
+                // Previous value was 0, which means that it will be deleted
+                // by another thread in "decrementStartVersionCount(..)" method.
+                // In that case, we delete here too, so we can safely try again.
+                for (;;) {
+                    int p = prev.get();
+
+                    assert p >= 0 : p;
+
+                    if (p == 0) {
+                        if (startVerCnts.remove(tx.startVersion(), prev))
+                            if (log.isDebugEnabled())
+                                log.debug("Removed count from onCreated callback: " + tx);
+
+                        break; // Retry outer loop.
+                    }
+
+                    if (prev.compareAndSet(p, p + 1)) {
+                        loop = false; // Increment succeeded - exit outer loop.
+
+                        break;
+                    }
+                }
+            }
+        }
+
+        if (tx.timeout() > 0) {
+            cctx.time().addTimeoutObject(tx);
+
+            if (log.isDebugEnabled())
+                log.debug("Registered transaction with timeout processor: " + tx);
+        }
+
+        if (log.isDebugEnabled())
+            log.debug("Transaction created: " + tx);
+
+        return tx;
+    }
+
+    /**
+     * Creates a future that will wait for all ongoing transactions that maybe affected by topology update
+     * to be finished. This set of transactions include
+     * <ul>
+     *     <li/> All {@link IgniteTxConcurrency#PESSIMISTIC} transactions with topology version
+     *     less or equal to {@code topVer}.
+     *     <li/> {@link IgniteTxConcurrency#OPTIMISTIC} transactions in PREPARING state with topology
+     *     version less or equal to {@code topVer} and having transaction key with entry that belongs to
+     *     one of partitions in {@code parts}.
+     * </ul>
+     *
+     * @param topVer Topology version.
+     * @return Future that will be completed when all ongoing transactions are finished.
+     */
+    public IgniteFuture<Boolean> finishTxs(long topVer) {
+        GridCompoundFuture<IgniteTx, Boolean> res =
+            new GridCompoundFuture<>(context().kernalContext(),
+                new IgniteReducer<IgniteTx, Boolean>() {
+                    @Override public boolean collect(IgniteTx e) {
+                        return true;
+                    }
+
+                    @Override public Boolean reduce() {
+                        return true;
+                    }
+                });
+
+        for (IgniteTxEx<K, V> tx : txs()) {
+            // Must wait for all transactions, even for DHT local and DHT remote since preloading may acquire
+            // values pending to be overwritten by prepared transaction.
+
+            if (tx.concurrency() == PESSIMISTIC) {
+                if (tx.topologyVersion() > 0 && tx.topologyVersion() < topVer)
+                    // For PESSIMISTIC mode we must wait for all uncompleted txs
+                    // as we do not know in advance which keys will participate in tx.
+                    res.add(tx.finishFuture());
+            }
+            else if (tx.concurrency() == OPTIMISTIC) {
+                // For OPTIMISTIC mode we wait only for txs in PREPARING state that
+                // have keys for given partitions.
+                IgniteTxState state = tx.state();
+                long txTopVer = tx.topologyVersion();
+
+                if ((state == PREPARING || state == PREPARED || state == COMMITTING)
+                    && txTopVer > 0 && txTopVer < topVer) {
+                    res.add(tx.finishFuture());
+                }
+            }
+        }
+
+        res.markInitialized();
+
+        return res;
+    }
+
+    /**
+     * Transaction start callback (has to do with when any operation was
+     * performed on this transaction).
+     *
+     * @param tx Started transaction.
+     * @return {@code True} if transaction is not in completed set.
+     */
+    public boolean onStarted(IgniteTxEx<K, V> tx) {
+        assert tx.state() == ACTIVE || tx.isRollbackOnly() : "Invalid transaction state [locId=" + cctx.localNodeId() +
+            ", tx=" + tx + ']';
+
+        if (isCompleted(tx)) {
+            if (log.isDebugEnabled())
+                log.debug("Attempt to start a completed transaction (will ignore): " + tx);
+
+            return false;
+        }
+
+        onTxStateChange(null, ACTIVE, tx);
+
+        if (log.isDebugEnabled())
+            log.debug("Transaction started: " + tx);
+
+        return true;
+    }
+
+    /**
+     * Reverse mapped version look up.
+     *
+     * @param dhtVer Dht version.
+     * @return Near version.
+     */
+    @Nullable public GridCacheVersion nearVersion(GridCacheVersion dhtVer) {
+        IgniteTxEx<K, V> tx = idMap.get(dhtVer);
+
+        if (tx != null)
+            return tx.nearXidVersion();
+
+        return null;
+    }
+
+    /**
+     * @param from Near version.
+     * @return DHT version for a near version.
+     */
+    public GridCacheVersion mappedVersion(GridCacheVersion from) {
+        GridCacheVersion to = mappedVers.get(from);
+
+        if (log.isDebugEnabled())
+            log.debug("Found mapped version [from=" + from + ", to=" + to);
+
+        return to;
+    }
+
+    /**
+     *
+     * @param ver Alternate version.
+     * @param tx Transaction.
+     */
+    public void addAlternateVersion(GridCacheVersion ver, IgniteTxEx<K, V> tx) {
+        if (idMap.putIfAbsent(ver, tx) == null)
+            if (log.isDebugEnabled())
+                log.debug("Registered alternate transaction version [ver=" + ver + ", tx=" + tx + ']');
+    }
+
+    /**
+     * @return Local transaction.
+     */
+    @SuppressWarnings({"unchecked"})
+    @Nullable public <T> T localTx() {
+        IgniteTxEx<K, V> tx = tx();
+
+        return tx != null && tx.local() ? (T)tx : null;
+    }
+
+    /**
+     * @return Transaction for current thread.
+     */
+    @SuppressWarnings({"unchecked"})
+    public <T> T threadLocalTx() {
+        IgniteTxEx<K, V> tx = tx(Thread.currentThread().getId());
+
+        return tx != null && tx.local() && (!tx.dht() || tx.colocated()) && !tx.implicit() ? (T)tx : null;
+    }
+
+    /**
+     * @return Transaction for current thread.
+     */
+    @SuppressWarnings({"unchecked", "RedundantCast"})
+    public <T> T tx() {
+        IgniteTxEx<K, V> tx = txContext();
+
+        return tx != null ? (T)tx : (T)tx(Thread.currentThread().getId());
+    }
+
+    /**
+     * @return Local transaction.
+     */
+    @Nullable public IgniteTxEx<K, V> localTxx() {
+        IgniteTxEx<K, V> tx = txx();
+
+        return tx != null && tx.local() ? tx : null;
+    }
+
+    /**
+     * @return Transaction for current thread.
+     */
+    @SuppressWarnings({"unchecked"})
+    public IgniteTxEx<K, V> txx() {
+        return tx();
+    }
+
+    /**
+     * @return User transaction for current thread.
+     */
+    @Nullable public IgniteTx userTx() {
+        IgniteTxEx<K, V> tx = txContext();
+
+        if (tx != null && tx.user() && tx.state() == ACTIVE)
+            return tx;
+
+        tx = tx(Thread.currentThread().getId());
+
+        return tx != null && tx.user() && tx.state() == ACTIVE ? tx : null;
+    }
+
+    /**
+     * @return User transaction.
+     */
+    @SuppressWarnings({"unchecked"})
+    @Nullable public <T extends IgniteTxLocalEx<K, V>> T userTxx() {
+        return (T)userTx();
+    }
+
+    /**
+     * @param threadId Id of thread for transaction.
+     * @return Transaction for thread with given ID.
+     */
+    @SuppressWarnings({"unchecked"})
+    public <T> T tx(long threadId) {
+        return (T)threadMap.get(threadId);
+    }
+
+    /**
+     * @return {@code True} if current thread is currently within transaction.
+     */
+    public boolean inUserTx() {
+        return userTx() != null;
+    }
+
+    /**
+     * @param txId Transaction ID.
+     * @return Transaction with given ID.
+     */
+    @SuppressWarnings({"unchecked"})
+    @Nullable public <T extends IgniteTxEx<K, V>> T tx(GridCacheVersion txId) {
+        return (T)idMap.get(txId);
+    }
+
+    /**
+     * @param txId Transaction ID.
+     * @return Transaction with given ID.
+     */
+    @SuppressWarnings({"unchecked"})
+    @Nullable public <T extends IgniteTxEx<K, V>> T nearTx(GridCacheVersion txId) {
+        return (T)nearIdMap.get(txId);
+    }
+
+    /**
+     * @param txId Transaction ID.
+     * @return Transaction with given ID.
+     */
+    @Nullable public IgniteTxEx<K, V> txx(GridCacheVersion txId) {
+        return idMap.get(txId);
+    }
+
+    /**
+     * Handles prepare stage of 2PC.
+     *
+     * @param tx Transaction to prepare.
+     * @throws IgniteCheckedException If preparation failed.
+     */
+    public void prepareTx(IgniteTxEx<K, V> tx) throws IgniteCheckedException {
+        if (tx.state() == MARKED_ROLLBACK) {
+            if (tx.timedOut())
+                throw new IgniteTxTimeoutException("Transaction timed out: " + this);
+
+            throw new IgniteCheckedException("Transaction is marked for rollback: " + tx);
+        }
+
+        if (tx.remainingTime() == 0) {
+            tx.setRollbackOnly();
+
+            throw new IgniteTxTimeoutException("Transaction timed out: " + this);
+        }
+
+        boolean txSerializableEnabled = cctx.txConfig().isTxSerializableEnabled();
+
+        // Clean up committed transactions queue.
+        if (tx.pessimistic()) {
+            if (tx.enforceSerializable() && txSerializableEnabled) {
+                for (Iterator<IgniteTxEx<K, V>> it = committedQ.iterator(); it.hasNext();) {
+                    IgniteTxEx<K, V> committedTx = it.next();
+
+                    assert committedTx != tx;
+
+                    // Clean up.
+                    if (isSafeToForget(committedTx))
+                        it.remove();
+                }
+            }
+
+            // Nothing else to do in pessimistic mode.
+            return;
+        }
+
+        if (txSerializableEnabled && tx.optimistic() && tx.enforceSerializable()) {
+            Set<IgniteTxKey<K>> readSet = tx.readSet();
+            Set<IgniteTxKey<K>> writeSet = tx.writeSet();
+
+            GridCacheVersion startTn = tx.startVersion();
+
+            GridCacheVersion finishTn = cctx.versions().last();
+
+            // Add future to prepare queue only on first prepare call.
+            if (tx.markPreparing())
+                prepareQ.offer(tx);
+
+            // Check that our read set does not intersect with write set
+            // of all transactions that completed their write phase
+            // while our transaction was in read phase.
+            for (Iterator<IgniteTxEx<K, V>> it = committedQ.iterator(); it.hasNext();) {
+                IgniteTxEx<K, V> committedTx = it.next();
+
+                assert committedTx != tx;
+
+                // Clean up.
+                if (isSafeToForget(committedTx)) {
+                    it.remove();
+
+                    continue;
+                }
+
+                GridCacheVersion tn = committedTx.endVersion();
+
+                // We only care about transactions
+                // with tn > startTn and tn <= finishTn
+                if (tn.compareTo(startTn) <= 0 || tn.compareTo(finishTn) > 0)
+                    continue;
+
+                if (tx.serializable()) {
+                    if (GridFunc.intersects(committedTx.writeSet(), readSet)) {
+                        tx.setRollbackOnly();
+
+                        throw new IgniteTxOptimisticException("Failed to prepare transaction " +
+                            "(committed vs. read-set conflict): " + tx);
+                    }
+                }
+            }
+
+            // Check that our read and write sets do not intersect with write
+            // sets of all active transactions.
+            for (Iterator<IgniteTxEx<K, V>> iter = prepareQ.iterator(); iter.hasNext();) {
+                IgniteTxEx<K, V> prepareTx = iter.next();
+
+                if (prepareTx == tx)
+                    // Skip yourself.
+                    continue;
+
+                // Optimistically remove completed transactions.
+                if (prepareTx.done()) {
+                    iter.remove();
+
+                    if (log.isDebugEnabled())
+                        log.debug("Removed finished transaction from active queue: " + prepareTx);
+
+                    continue;
+                }
+
+                // Check if originating node left.
+                if (cctx.discovery().node(prepareTx.nodeId()) == null) {
+                    iter.remove();
+
+                    rollbackTx(prepareTx);
+
+                    if (log.isDebugEnabled())
+                        log.debug("Removed and rolled back transaction because sender node left grid: " +
+                            CU.txString(prepareTx));
+
+                    continue;
+                }
+
+                if (tx.serializable() && !prepareTx.isRollbackOnly()) {
+                    Set<IgniteTxKey<K>> prepareWriteSet = prepareTx.writeSet();
+
+                    if (GridFunc.intersects(prepareWriteSet, readSet, writeSet)) {
+                        // Remove from active set.
+                        iter.remove();
+
+                        tx.setRollbackOnly();
+
+                        throw new IgniteTxOptimisticException(
+                            "Failed to prepare transaction (read-set/write-set conflict): " + tx);
+                    }
+                }
+            }
+        }
+
+        // Optimistic.
+        assert tx.optimistic();
+
+        if (!lockMultiple(tx, tx.optimisticLockEntries())) {
+            tx.setRollbackOnly();
+
+            throw new IgniteTxOptimisticException("Failed to prepare transaction (lock conflict): " + tx);
+        }
+    }
+
+    /**
+     * @param tx Transaction to check.
+     * @return {@code True} if transaction can be discarded.
+     */
+    private boolean isSafeToForget(IgniteTxEx<K, V> tx) {
+        Map.Entry<GridCacheVersion, AtomicInt> e = startVerCnts.firstEntry();
+
+        if (e == null)
+            return true;
+
+        assert e.getValue().get() >= 0;
+
+        return tx.endVersion().compareTo(e.getKey()) <= 0;
+    }
+
+    /**
+     * Decrement start version count.
+     *
+     * @param tx Cache transaction.
+     */
+    private void decrementStartVersionCount(IgniteTxEx<K, V> tx) {
+        AtomicInt cnt = startVerCnts.get(tx.startVersion());
+
+        assert cnt != null : "Failed to find start version count for transaction [startVerCnts=" + startVerCnts +
+            ", tx=" + tx + ']';
+
+        assert cnt.get() > 0;
+
+        if (cnt.decrementAndGet() == 0)
+            if (startVerCnts.remove(tx.startVersion(), cnt))
+                if (log.isDebugEnabled())
+                    log.debug("Removed start version for transaction: " + tx);
+    }
+
+    /**
+     * @param tx Transaction.
+     */
+    private void removeObsolete(IgniteTxEx<K, V> tx) {
+        Collection<IgniteTxEntry<K, V>> entries = (tx.local() && !tx.dht()) ? tx.allEntries() : tx.writeEntries();
+
+        for (IgniteTxEntry<K, V> entry : entries) {
+            GridCacheEntryEx<K, V> cached = entry.cached();
+
+            GridCacheContext<K, V> cacheCtx = entry.context();
+
+            if (cached == null)
+                cached = cacheCtx.cache().peekEx(entry.key());
+
+            if (cached.detached())
+                continue;
+
+            try {
+                if (cached.obsolete() || cached.markObsoleteIfEmpty(tx.xidVersion()))
+                    cacheCtx.cache().removeEntry(cached);
+
+                if (!tx.near() && isNearEnabled(cacheCtx)) {
+                    GridNearCacheAdapter<K, V> near = cacheCtx.isNear() ? cacheCtx.near() : cacheCtx.dht().near();
+
+                    GridNearCacheEntry<K, V> e = near.peekExx(entry.key());
+
+                    if (e != null && e.markObsoleteIfEmpty(tx.xidVersion()))
+                        near.removeEntry(e);
+                }
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to remove obsolete entry from cache: " + cached, e);
+            }
+        }
+    }
+
+    /**
+     * @param c Collection to copy.
+     * @return Copy of the collection.
+     */
+    private Collection<GridCacheVersion> copyOf(Iterable<GridCacheVersion> c) {
+        Collection<GridCacheVersion> l = new LinkedList<>();
+
+        for (GridCacheVersion v : c)
+            l.add(v);
+
+        return l;
+    }
+
+    /**
+     * Gets committed transactions starting from the given version (inclusive). // TODO: GG-4011: why inclusive?
+     *
+     * @param min Start (or minimum) version.
+     * @return Committed transactions starting from the given version (non-inclusive).
+     */
+    public Collection<GridCacheVersion> committedVersions(GridCacheVersion min) {
+        Set<GridCacheVersion> set = committedVers.tailSet(min, true);
+
+        return set == null || set.isEmpty() ? Collections.<GridCacheVersion>emptyList() : copyOf(set);
+    }
+
+    /**
+     * Gets rolledback transactions starting from the given version (inclusive). // TODO: GG-4011: why inclusive?
+     *
+     * @param min Start (or minimum) version.
+     * @return Committed transactions starting from the given version (non-inclusive).
+     */
+    public Collection<GridCacheVersion> rolledbackVersions(GridCacheVersion min) {
+        Set<GridCacheVersion> set = rolledbackVers.tailSet(min, true);
+
+        return set == null || set.isEmpty() ? Collections.<GridCacheVersion>emptyList() : copyOf(set);
+    }
+
+    /**
+     * @param tx Tx to remove.
+     */
+    public void removeCommittedTx(IgniteTxEx<K, V> tx) {
+        committedVers.remove(tx.xidVersion());
+    }
+
+    /**
+     * @param tx Committed transaction.
+     * @return If transaction was not already present in committed set.
+     */
+    public boolean addCommittedTx(IgniteTxEx<K, V> tx) {
+        return addCommittedTx(tx.xidVersion(), tx.nearXidVersion());
+    }
+
+    /**
+     * @param tx Committed transaction.
+     * @return If transaction was not already present in committed set.
+     */
+    public boolean addRolledbackTx(IgniteTxEx<K, V> tx) {
+        return addRolledbackTx(tx.xidVersion());
+    }
+
+    /**
+     * @param xidVer Completed transaction version.
+     * @param nearXidVer Optional near transaction ID.
+     * @return If transaction was not already present in completed set.
+     */
+    public boolean addCommittedTx(GridCacheVersion xidVer, @Nullable GridCacheVersion nearXidVer) {
+        assert !rolledbackVers.contains(xidVer) : "Version was rolled back: " + xidVer;
+
+        if (nearXidVer != null)
+            xidVer = new CommittedVersion(xidVer, nearXidVer);
+
+        if (committedVers.add(xidVer)) {
+            if (log.isDebugEnabled())
+                log.debug("Added transaction to committed version set: " + xidVer);
+
+            return true;
+        }
+        else {
+            if (log.isDebugEnabled())
+                log.debug("Transaction is already present in committed version set: " + xidVer);
+
+            return false;
+        }
+    }
+
+    /**
+     * @param xidVer Completed transaction version.
+     * @return If transaction was not already present in completed set.
+     */
+    public boolean addRolledbackTx(GridCacheVersion xidVer) {
+        assert !committedVers.contains(xidVer);
+
+        if (rolledbackVers.add(xidVer)) {
+            if (log.isDebugEnabled())
+                log.debug("Added transaction to rolled back version set: " + xidVer);
+
+            return true;
+        }
+        else {
+            if (log.isDebugEnabled())
+                log.debug("Transaction is already present in rolled back version set: " + xidVer);
+
+            return false;
+        }
+    }
+
+    /**
+     * @param tx Transaction.
+     */
+    private void processCompletedEntries(IgniteTxEx<K, V> tx) {
+        if (tx.needsCompletedVersions()) {
+            GridCacheVersion min = minVersion(tx.readEntries(), tx.xidVersion(), tx);
+
+            min = minVersion(tx.writeEntries(), min, tx);
+
+            assert min != null;
+
+            tx.completedVersions(min, committedVersions(min), rolledbackVersions(min));
+        }
+    }
+
+    /**
+     * Collects versions for all pending locks for all entries within transaction
+     *
+     * @param dhtTxLoc Transaction being committed.
+     */
+    private void collectPendingVersions(GridDhtTxLocal<K, V> dhtTxLoc) {
+        if (dhtTxLoc.needsCompletedVersions()) {
+            if (log.isDebugEnabled())
+                log.debug("Checking for pending locks with version less then tx version: " + dhtTxLoc);
+
+            Set<GridCacheVersion> vers = new LinkedHashSet<>();
+
+            collectPendingVersions(dhtTxLoc.readEntries(), dhtTxLoc.xidVersion(), vers);
+            collectPendingVersions(dhtTxLoc.writeEntries(), dhtTxLoc.xidVersion(), vers);
+
+            if (!vers.isEmpty())
+                dhtTxLoc.pendingVersions(vers);
+        }
+    }
+
+    /**
+     * Gets versions of all not acquired locks for collection of tx entries that are less then base version.
+     *
+     * @param entries Tx entries to process.
+     * @param baseVer Base version to compare with.
+     * @param vers Collection of versions that will be populated.
+     */
+    @SuppressWarnings("TypeMayBeWeakened")
+    private void collectPendingVersions(Iterable<IgniteTxEntry<K, V>> entries,
+        GridCacheVersion baseVer, Set<GridCacheVersion> vers) {
+
+        // The locks are not released yet, so we can safely list pending candidates versions.
+        for (IgniteTxEntry<K, V> txEntry : entries) {
+            GridCacheEntryEx<K, V> cached = txEntry.cached();
+
+            try {
+                // If check should be faster then exception handling.
+                if (!cached.obsolete()) {
+                    for (GridCacheMvccCandidate<K> cand : cached.localCandidates()) {
+                        if (!cand.owner() && cand.version().compareTo(baseVer) < 0) {
+                            if (log.isDebugEnabled())
+                                log.debug("Adding candidate version to pending set: " + cand);
+
+                            vers.add(cand.version());
+                        }
+                    }
+                }
+            }
+            catch (GridCacheEntryRemovedException ignored) {
+                if (log.isDebugEnabled())
+                    log.debug("There are no pending locks for entry (entry was deleted in transaction): " + txEntry);
+            }
+        }
+    }
+
+    /**
+     * Go through all candidates for entries involved in transaction and find their min
+     * version. We know that these candidates will commit after this transaction, and
+     * therefore we can grab the min version so we can send all committed and rolled
+     * back versions from min to current to remote nodes for re-ordering.
+     *
+     * @param entries Entries.
+     * @param min Min version so far.
+     * @param tx Transaction.
+     * @return Minimal available version.
+     */
+    private GridCacheVersion minVersion(Iterable<IgniteTxEntry<K, V>> entries, GridCacheVersion min,
+        IgniteTxEx<K, V> tx) {
+        for (IgniteTxEntry<K, V> txEntry : entries) {
+            GridCacheEntryEx<K, V> cached = txEntry.cached();
+
+            // We are assuming that this method is only called on commit. In that
+            // case, if lock is held, entry can never be removed.
+            assert txEntry.isRead() || !cached.obsolete(tx.xidVersion()) :
+                "Invalid obsolete version for transaction [entry=" + cached + ", tx=" + tx + ']';
+
+            for (GridCacheMvccCandidate<K> cand : cached.remoteMvccSnapshot())
+                if (min == null || cand.version().isLess(min))
+                    min = cand.version();
+        }
+
+        return min;
+    }
+
+    /**
+     * Commits a transaction.
+     *
+     * @param tx Transaction to commit.
+     */
+    public void commitTx(IgniteTxEx<K, V> tx) {
+        assert tx != null;
+        assert tx.state() == COMMITTING : "Invalid transaction state for commit from tm [state=" + tx.state() +
+            ", expected=COMMITTING, tx=" + tx + ']';
+
+        if (log.isDebugEnabled())
+            log.debug("Committing from TM [locNodeId=" + cctx.localNodeId() + ", tx=" + tx + ']');
+
+        if (tx.timeout() > 0) {
+            cctx.time().removeTimeoutObject(tx);
+
+            if (log.isDebugEnabled())
+                log.debug("Unregistered transaction with timeout processor: " + tx);
+        }
+
+        /*
+         * Note that write phase is handled by transaction adapter itself,
+         * so we don't do it here.
+         */
+
+        // 1. Make sure that committed version has been recorded.
+        if (!(committedVers.contains(tx.xidVersion()) || tx.writeSet().isEmpty() || tx.isSystemInvalidate())) {
+            uncommitTx(tx);
+
+            throw new IgniteException("Missing commit version (consider increasing " +
+                GG_MAX_COMPLETED_TX_COUNT + " system property) [ver=" + tx.xidVersion() + ", firstVer=" +
+                committedVers.firstx() + ", lastVer=" + committedVers.lastx() + ", tx=" + tx.xid() + ']');
+        }
+
+        ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> txIdMap = transactionMap(tx);
+
+        if (txIdMap.remove(tx.xidVersion(), tx)) {
+            // 2. Must process completed entries before unlocking!
+            processCompletedEntries(tx);
+
+            if (tx instanceof GridDhtTxLocal) {
+                GridDhtTxLocal<K, V> dhtTxLoc = (GridDhtTxLocal<K, V>)tx;
+
+                collectPendingVersions(dhtTxLoc);
+            }
+
+            // 3.1 Call dataStructures manager.
+            for (GridCacheContext<K, V> cacheCtx : cctx.cacheContexts())
+                cacheCtx.dataStructures().onTxCommitted(tx);
+
+            // 3.2 Add to pessimistic commit buffer if needed.
+            addPessimisticRecovery(tx);
+
+            // 4. Unlock write resources.
+            if (tx.groupLock())
+                unlockGroupLocks(tx);
+            else
+                unlockMultiple(tx, tx.writeEntries());
+
+            // 5. For pessimistic transaction, unlock read resources if required.
+            if (tx.pessimistic() && !tx.readCommitted() && !tx.groupLock())
+                unlockMultiple(tx, tx.readEntries());
+
+            // 6. Notify evictions.
+            notifyEvitions(tx);
+
+            // 7. Remove obsolete entries from cache.
+            removeObsolete(tx);
+
+            // 8. Assign transaction number at the end of transaction.
+            tx.endVersion(cctx.versions().next(tx.topologyVersion()));
+
+            // 9. Clean start transaction number for this transaction.
+            if (cctx.txConfig().isTxSerializableEnabled())
+                decrementStartVersionCount(tx);
+
+            // 10. Add to committed queue only if it is possible
+            //    that this transaction can affect other ones.
+            if (cctx.txConfig().isTxSerializableEnabled() && tx.enforceSerializable() && !isSafeToForget(tx))
+                committedQ.add(tx);
+
+            // 11. Remove from per-thread storage.
+            if (tx.local() && !tx.dht())
+                threadMap.remove(tx.threadId(), tx);
+
+            // 12. Unregister explicit locks.
+            if (!tx.alternateVersions().isEmpty()) {
+                for (GridCacheVersion ver : tx.alternateVersions())
+                    idMap.remove(ver);
+            }
+
+            // 13. Remove Near-2-DHT mappings.
+            if (tx instanceof GridCacheMappedVersion) {
+                GridCacheVersion mapped = ((GridCacheMappedVersion)tx).mappedVersion();
+
+                if (mapped != null)
+                    mappedVers.remove(mapped);
+            }
+
+            // 14. Clear context.
+            txContextReset();
+
+            // 15. Update metrics.
+            if (!tx.dht() && tx.local()) {
+                cctx.txMetrics().onTxCommit();
+
+                for (int cacheId : tx.activeCacheIds()) {
+                    GridCacheContext<K, V> cacheCtx = cctx.cacheContext(cacheId);
+
+                    cacheCtx.cache().metrics0().onTxCommit();
+                }
+            }
+
+            if (slowTxWarnTimeout > 0 && tx.local() &&
+                U.currentTimeMillis() - tx.startTime() > slowTxWarnTimeout)
+                U.warn(log, "Slow transaction detected [tx=" + tx +
+                    ", slowTxWarnTimeout=" + slowTxWarnTimeout + ']') ;
+
+            if (log.isDebugEnabled())
+                log.debug("Committed from TM [locNodeId=" + cctx.localNodeId() + ", tx=" + tx + ']');
+        }
+        else if (log.isDebugEnabled())
+            log.debug("Did not commit from TM (was already committed): " + tx);
+    }
+
+    /**
+     * Rolls back a transaction.
+     *
+     * @param tx Transaction to rollback.
+     */
+    public void rollbackTx(IgniteTxEx<K, V> tx) {
+        assert tx != null;
+
+        if (log.isDebugEnabled())
+            log.debug("Rolling back from TM [locNodeId=" + cctx.localNodeId() + ", tx=" + tx + ']');
+
+        // 1. Record transaction version to avoid duplicates.
+        addRolledbackTx(tx);
+
+        ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> txIdMap = transactionMap(tx);
+
+        if (txIdMap.remove(tx.xidVersion(), tx)) {
+            // 2. Unlock write resources.
+            unlockMultiple(tx, tx.writeEntries());
+
+            // 3. For pessimistic transaction, unlock read resources if required.
+            if (tx.pessimistic() && !tx.readCommitted())
+                unlockMultiple(tx, tx.readEntries());
+
+            // 4. Notify evictions.
+            notifyEvitions(tx);
+
+            // 5. Remove obsolete entries.
+            removeObsolete(tx);
+
+            // 6. Clean start transaction number for this transaction.
+            if (cctx.txConfig().isTxSerializableEnabled())
+                decrementStartVersionCount(tx);
+
+            // 7. Remove from per-thread storage.
+            if (tx.local() && !tx.dht())
+                threadMap.remove(tx.threadId(), tx);
+
+            // 8. Unregister explicit locks.
+            if (!tx.alternateVersions().isEmpty())
+                for (GridCacheVersion ver : tx.alternateVersions())
+                    idMap.remove(ver);
+
+            // 9. Remove Near-2-DHT mappings.
+            if (tx instanceof GridCacheMappedVersion)
+                mappedVers.remove(((GridCacheMappedVersion)tx).mappedVersion());
+
+            // 10. Clear context.
+            txContextReset();
+
+            // 11. Update metrics.
+            if (!tx.dht() && tx.local()) {
+                cctx.txMetrics().onTxRollback();
+
+                for (int cacheId : tx.activeCacheIds()) {
+                    GridCacheContext<K, V> cacheCtx = cctx.cacheContext(cacheId);
+
+                    cacheCtx.cache().metrics0().onTxRollback();
+                }
+            }
+
+            if (log.isDebugEnabled())
+                log.debug("Rolled back from TM: " + tx);
+        }
+        else if (log.isDebugEnabled())
+            log.debug("Did not rollback from TM (was already rolled back): " + tx);
+    }
+
+    /**
+     * Tries to minimize damage from partially-committed transaction.
+     *
+     * @param tx Tx to uncommit.
+     */
+    public void uncommitTx(IgniteTxEx<K, V> tx) {
+        assert tx != null;
+
+        if (log.isDebugEnabled())
+            log.debug("Uncommiting from TM: " + tx);
+
+        ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> txIdMap = transactionMap(tx);
+
+        if (txIdMap.remove(tx.xidVersion(), tx)) {
+            // 1. Unlock write resources.
+            unlockMultiple(tx, tx.writeEntries());
+
+            // 2. For pessimistic transaction, unlock read resources if required.
+            if (tx.pessimistic() && !tx.readCommitted())
+                unlockMultiple(tx, tx.readEntries());
+
+            // 3. Notify evictions.
+            notifyEvitions(tx);
+
+            // 4. Clean start transaction number for this transaction.
+            if (cctx.txConfig().isTxSerializableEnabled())
+                decrementStartVersionCount(tx);
+
+            // 5. Remove from per-thread storage.
+            if (tx.local() && !tx.dht())
+                threadMap.remove(tx.threadId(), tx);
+
+            // 6. Unregister explicit locks.
+            if (!tx.alternateVersions().isEmpty())
+                for (GridCacheVersion ver : tx.alternateVersions())
+                    idMap.remove(ver);
+
+            // 7. Remove Near-2-DHT mappings.
+            if (tx instanceof GridCacheMappedVersion)
+                mappedVers.remove(((GridCacheMappedVersion)tx).mappedVersion());
+
+            // 8. Clear context.
+            txContextReset();
+
+            if (log.isDebugEnabled())
+                log.debug("Uncommitted from TM: " + tx);
+        }
+        else if (log.isDebugEnabled())
+            log.debug("Did not uncommit from TM (was already committed or rolled back): " + tx);
+    }
+
+    /**
+     * Gets transaction ID map depending on transaction type.
+     *
+     * @param tx Transaction.
+     * @return Transaction map.
+     */
+    private ConcurrentMap<GridCacheVersion, IgniteTxEx<K, V>> transactionMap(IgniteTxEx<K, V> tx) {
+        return (tx.near() && !tx.local()) ? nearIdMap : idMap;
+    }
+
+    /**
+     * @param tx Transaction to notify evictions for.
+     */
+    private void notifyEvitions(IgniteTxEx<K, V> tx) {
+        if (tx.internal() && !tx.groupLock())
+            return;
+
+        for (IgniteTxEntry<K, V> txEntry : tx.allEntries())
+            txEntry.cached().context().evicts().touch(txEntry, tx.local());
+    }
+
+    /**
+     * Callback invoked whenever a member of a transaction acquires
+     * lock ownership.
+     *
+     * @param entry Cache entry.
+     * @param owner Candidate that won ownership.
+     * @return {@code True} if transaction was notified, {@code false} otherwise.
+     */
+    public boolean onOwnerChanged(GridCacheEntryEx<K, V> entry, GridCacheMvccCandidate<K> owner) {
+        // We only care about acquired locks.
+        if (owner != null) {
+            IgniteTxAdapter<K, V> tx = tx(owner.version());
+
+            if (tx == null)
+                tx = nearTx(owner.version());
+
+            if (tx != null) {
+                if (!tx.local()) {
+                    if (log.isDebugEnabled())
+                        log.debug("Found transaction for owner changed event [owner=" + owner + ", entry=" + entry +
+                            ", tx=" + tx + ']');
+
+                    tx.onOwnerChanged(entry, owner);
+
+                    return true;
+                }
+                else if (log.isDebugEnabled())
+                    log.debug("Ignoring local transaction for owner change event: " + tx);
+            }
+            else if (log.isDebugEnabled())
+                log.debug("Transaction not found for owner changed event [owner=" + owner + ", entry=" + entry + ']');
+        }
+
+        return false;
+    }
+
+    /**
+     * Callback called by near finish future before sending near finish request to remote node. Will increment
+     * per-thread counter so that further awaitAck call will wait for finish response.
+     *
+     * @param rmtNodeId Remote node ID for which finish request is being sent.
+     * @param threadId Near tx thread ID.
+     */
+    public void beforeFinishRemote(UUID rmtNodeId, long threadId) {
+        if (finishSyncDisabled)
+            return;
+
+        assert txFinishSync != null;
+
+        txFinishSync.onFinishSend(rmtNodeId, threadId);
+    }
+
+    /**
+     * Callback invoked when near finish response is received from remote node.
+     *
+     * @param rmtNodeId Remote node ID from which response is received.
+     * @param threadId Near tx thread ID.
+     */
+    public void onFinishedRemote(UUID rmtNodeId, long threadId) {
+        if (finishSyncDisabled)
+            return;
+
+        assert txFinishSync != null;
+
+        txFinishSync.onAckReceived(rmtNodeId, threadId);
+    }
+
+    /**
+     * Asynchronously waits for last finish request ack.
+     *
+     * @param rmtNodeId Remote node ID.
+     * @param threadId Near tx thread ID.
+     * @return {@code null} if ack was received or future that will be completed when ack is received.
+     */
+    @Nullable public IgniteFuture<?> awaitFinishAckAsync(UUID rmtNodeId, long threadId) {
+        if (finishSyncDisabled)
+            return null;
+
+        assert txFinishSync != null;
+
+        return txFinishSync.awaitAckAsync(rmtNodeId, threadId);
+    }
+
+    /**
+     * For test purposes only.
+     *
+     * @param finishSyncDisabled {@code True} if finish sync should be disabled.
+     */
+    public void finishSyncDisabled(boolean finishSyncDisabled) {
+        this.finishSyncDisabled = finishSyncDisabled;
+    }
+
+    /**
+     * @param tx Transaction.
+     * @param entries Entries to lock.
+     * @return {@code True} if all keys were locked.
+     * @throws IgniteCheckedException If lock has been cancelled.
+     */
+    private boolean lockMultiple(IgniteTxEx<K, V> tx, Iterable<IgniteTxEntry<K, V>> entries)
+        throws IgniteCheckedException {
+        assert tx.optimistic();
+
+        long remainingTime = U.currentTimeMillis() - (tx.startTime() + tx.timeout());
+
+        // For serializable transactions, failure to acquire lock means
+        // that there is a serializable conflict. For all other isolation levels,
+        // we wait for the lock.
+        long timeout = tx.timeout() == 0 ? 0 : remainingTime;
+
+        for (IgniteTxEntry<K, V> txEntry1 : entries) {
+            // Check if this entry was prepared before.
+            if (!txEntry1.markPrepared())
+                continue;
+
+            GridCacheContext<K, V> cacheCtx = txEntry1.context();
+
+            while (true) {
+                try {
+                    GridCacheEntryEx<K, V> entry1 = txEntry1.cached();
+
+                    assert !entry1.detached() : "Expected non-detached entry for near transaction " +
+                        "[locNodeId=" + cctx.localNodeId() + ", entry=" + entry1 + ']';
+
+                    if (!entry1.tmLock(tx, timeout)) {
+                        // Unlock locks locked so far.
+                        for (IgniteTxEntry<K, V> txEntry2 : entries) {
+                            if (txEntry2 == txEntry1)
+                                break;
+
+                            txEntry2.cached().txUnlock(tx);
+                        }
+
+                        return false;
+                    }
+
+                    entry1.unswap();
+
+                    break;
+                }
+                catch (GridCacheEntryRemovedException ignored) {
+                    if (log.isDebugEnabled())
+                        log.debug("Got removed entry in TM lockMultiple(..) method (will retry): " + txEntry1);
+
+                    try {
+                        // Renew cache entry.
+                        txEntry1.cached(cacheCtx.cache().entryEx(txEntry1.key()), txEntry1.keyBytes());
+                    }
+                    catch (GridDhtInvalidPartitionException e) {
+                        assert tx.dht() : "Received invalid partition for non DHT transaction [tx=" +
+                            tx + ", invalidPart=" + e.partition() + ']';
+
+                        // If partition is invalid, we ignore this entry.
+                        tx.addInvalidPartition(cacheCtx, e.partition());
+
+                        break;
+                    }
+                }
+                catch (GridDistributedLockCancelledException ignore) {
+                    tx.setRollbackOnly();
+
+                    throw new IgniteCheckedException("Entry lock has been cancelled for transaction: " + tx);
+                }
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * Unlocks entries locked by group transaction.
+     *
+     * @param txx Transaction.
+     */
+    @SuppressWarnings("unchecked")
+    private void unlockGroupLocks(IgniteTxEx txx) {
+        IgniteTxKey grpLockKey = txx.groupLockKey();
+
+        assert grpLockKey != null;
+
+        if (grpLockKey == null)
+            return;
+
+        IgniteTxEntry txEntry = txx.entry(grpLockKey);
+
+        assert txEntry != null || (txx.near() && !txx.local());
+
+        if (txEntry != null) {
+            GridCacheContext cacheCtx = txEntry.context();
+
+            // Group-locked entries must be locked.
+            while (true) {
+                try {
+                    GridCacheEntryEx<K, V> entry = txEntry.cached();
+
+                    assert entry != null;
+
+                    entry.txUnlock(txx);
+
+                    break;
+                }
+                catch (GridCacheEntryRemovedException ignored) {
+                    if (log.isDebugEnabled())
+                        log.debug("Got removed entry in TM unlockGroupLocks(..) method (will retry): " + txEntry);
+
+                    GridCacheAdapter cache = cacheCtx.cache();
+
+                    // Renew cache entry.
+                    txEntry.cached(cache.entryEx(txEntry.key()), txEntry.keyBytes());
+                }
+            }
+        }
+    }
+
+    /**
+     * @param tx Owning transaction.
+     * @param entries Entries to unlock.
+     */
+    private void unlockMultiple(IgniteTxEx<K, V> tx, Iterable<IgniteTxEntry<K, V>> entries) {
+        for (IgniteTxEntry<K, V> txEntry : entries) {
+            GridCacheContext<K, V> cacheCtx = txEntry.context();
+
+            while (true) {
+                try {
+                    GridCacheEntryEx<K, V> entry = txEntry.cached();
+
+                    if (entry.detached())
+                        break;
+
+                    assert entry != null;
+
+                    entry.txUnlock(tx);
+
+                    break;
+                }
+                catch (GridCacheEntryRemovedException ignored) {
+                    if (log.isDebugEnabled())
+                        log.debug("Got removed entry in TM unlockMultiple(..) method (will retry): " + txEntry);
+
+                    // Renew cache entry.
+                    txEntry.cached(cacheCtx.cache().entryEx(txEntry.key()), txEntry.keyBytes());
+                }
+            }
+        }
+    }
+
+    /**
+     * @param sync Transaction synchronizations to add.
+     */
+    public void addSynchronizations(IgniteTxSynchronization... sync) {
+        if (F.isEmpty(sync))
+            return;
+
+        F.copy(syncs, sync);
+    }
+
+    /**
+     * @param sync Transaction synchronizations to remove.
+     */
+    public void removeSynchronizations(IgniteTxSynchronization... sync) {
+        if (F.isEmpty(sync))
+            return;
+
+        F.lose(syncs, false, Arrays.asList(sync));
+    }
+
+    /**
+     * @return Registered transaction synchronizations
+     */
+    public Collection<IgniteTxSynchronization> synchronizations() {
+        return Collections.unmodifiableList(new LinkedList<>(syncs));
+    }
+
+    /**
+     * @param prevState Previous state.
+     * @param newState New state.
+     * @param tx Cache transaction.
+     */
+    public void onTxStateChange(@Nullable IgniteTxState prevState, IgniteTxState newState, IgniteTx tx) {
+        // Notify synchronizations.
+        for (IgniteTxSynchronization s : syncs)
+            s.onStateChanged(prevState, newState, tx);
+    }
+
+    /**
+     * @param tx Committing transaction.
+     */
+    public void txContext(IgniteTxEx tx) {
+        threadCtx.set(tx);
+    }
+
+    /**
+     * @return Currently committing transaction.
+     */
+    @SuppressWarnings({"unchecked"})
+    private IgniteTxEx<K, V> txContext() {
+        return threadCtx.get();
+    }
+
+    /**
+     * Gets version of transaction in tx context or {@code null}
+     * if tx context is empty.
+     * <p>
+     * This is a convenience method provided mostly for debugging.
+     *
+     * @return Transaction version from transaction context.
+     */
+    @Nullable public GridCacheVersion txContextVersion() {
+        IgniteTxEx<K, V> tx = txContext();
+
+        return tx == null ? null : tx.xidVersion();
+    }
+
+    /**
+     * Commit ended.
+     */
+    public void txContextReset() {
+        threadCtx.set(null);
+    }
+
+    /**
+     * @return All transactions.
+     */
+    public Collection<IgniteTxEx<K, V>> txs() {
+        return F.concat(false, idMap.values(), nearIdMap.values());
+    }
+
+    /**
+     * @return Slow tx warn timeout.
+     */
+    public int slowTxWarnTimeout() {
+        return slowTxWarnTimeout;
+    }
+
+    /**
+     * @param slowTxWarnTimeout Slow tx warn timeout.
+     */
+    public void slowTxWarnTimeout(int slowTxWarnTimeout) {
+        this.slowTxWarnTimeout = slowTxWarnTimeout;
+    }
+
+    /**
+     * Checks if transactions with given near version ID was prepared or committed.
+     *
+     * @param nearVer Near version ID.
+     * @param txNum Number of transactions.
+     * @return {@code True} if transactions were prepared or committed.
+     */
+    public boolean txsPreparedOrCommitted(GridCacheVersion nearVer, int txNum) {
+        Collection<GridCacheVersion> processedVers = null;
+
+        for (IgniteTxEx<K, V> tx : txs()) {
+            if (nearVer.equals(tx.nearXidVersion())) {
+                IgniteTxState state = tx.state();
+
+                if (state == PREPARED || state == COMMITTING || state == COMMITTED) {
+                    if (--txNum == 0)
+                        return true;
+                }
+                else {
+                    if (tx.state(MARKED_ROLLBACK) || tx.state() == UNKNOWN) {
+                        tx.rollbackAsync();
+
+                        if (log.isDebugEnabled())
+                            log.debug("Transaction was not prepared (rolled back): " + tx);
+
+                        return false;
+                    }
+                    else {
+                        if (tx.state() == COMMITTED) {
+                            if (--txNum == 0)
+                                return true;
+                        }
+                        else {
+                            if (log.isDebugEnabled())
+                                log.debug("Transaction is not prepared: " + tx);
+
+                            return false;
+                        }
+                    }
+                }
+
+                if (processedVers == null)
+                    processedVers = new HashSet<>(txNum, 1.0f);
+
+                processedVers.add(tx.xidVersion());
+            }
+        }
+
+        // Not all transactions were found. Need to scan committed versions to check
+        // if transaction was already committed.
+        for (GridCacheVersion ver : committedVers) {
+            if (processedVers != null && processedVers.contains(ver))
+                continue;
+
+            if (ver instanceof CommittedVersion) {
+                CommittedVersion commitVer = (CommittedVersion)ver;
+
+                if (commitVer.nearVer.equals(nearVer)) {
+                    if (--txNum == 0)
+                        return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Adds transaction to pessimistic recovery buffer if needed.
+     *
+     * @param tx Committed transaction to add.
+     */
+    private void addPessimisticRecovery(IgniteTxEx<K, V> tx) {
+        if (pessimisticRecoveryBuf == null)
+            return;
+
+        // Do not store recovery information for optimistic or replicated local transactions.
+        if (tx.optimistic() || (tx.local() && tx.replicated()))
+            return;
+
+        pessimisticRecoveryBuf.addCommittedTx(tx);
+    }
+
+    /**
+     * Checks whether transaction with given near version was committed on this node and returns commit info.
+     *
+     * @param nearTxVer Near tx version.
+     * @param originatingNodeId Originating node ID.
+     * @param originatingThreadId Originating thread ID.
+     * @return Commit info, if present.
+     */
+    @Nullable public GridCacheCommittedTxInfo<K, V> txCommitted(GridCacheVersion nearTxVer,
+        UUID originatingNodeId, long originatingThreadId) {
+        assert pessimisticRecoveryBuf != null : "Should not be called for LOCAL cache.";
+
+        return pessimisticRecoveryBuf.committedTx(nearTxVer, originatingNodeId, originatingThreadId);
+    }
+
+    /**
+     * Gets local transaction for pessimistic tx recovery.
+     *
+     * @param nearXidVer Near tx ID.
+     * @return Near local or colocated local transaction.
+     */
+    @Nullable public IgniteTxEx<K, V> localTxForRecovery(GridCacheVersion nearXidVer, boolean markFinalizing) {
+        // First check if we have near transaction with this ID.
+        IgniteTxEx<K, V> tx = idMap.get(nearXidVer);
+
+        if (tx == null) {
+            // Check all local transactions and mark them as waiting for recovery to prevent finish race.
+            for (IgniteTxEx<K, V> txEx : idMap.values()) {
+                if (nearXidVer.equals(txEx.nearXidVersion())) {
+                    if (!markFinalizing || !txEx.markFinalizing(RECOVERY_WAIT))
+                        tx = txEx;
+                }
+            }
+        }
+
+        // Either we found near transaction or one of transactions is being committed by user.
+        // Wait for it and send reply.
+        if (tx != null && tx.local())
+            return tx;
+
+        return null;
+    }
+
+    /**
+     * Commits or rolls back prepared transaction.
+     *
+     * @param tx Transaction.
+     * @param commit Whether transaction should be committed or rolled back.
+     */
+    public void finishOptimisticTxOnRecovery(final IgniteTxEx<K, V> tx, boolean commit) {
+        if (log.isDebugEnabled())
+            log.debug("Finishing prepared transaction [tx=" + tx + ", commit=" + commit + ']');
+
+        if (!tx.markFinalizing(RECOVERY_FINISH)) {
+            if (log.isDebugEnabled())
+                log.debug("Will not try to commit prepared transaction (could not mark finalized): " + tx);
+
+            return;
+        }
+
+        if (tx instanceof GridDistributedTxRemoteAdapter) {
+            IgniteTxRemoteEx<K,V> rmtTx = (IgniteTxRemoteEx<K, V>)tx;
+
+            rmtTx.doneRemote(tx.xidVersion(), Collections.<GridCacheVersion>emptyList(), Collections.<GridCacheVersion>emptyList(),
+                Collections.<GridCacheVersion>emptyList());
+        }
+
+        if (commit)
+            tx.commitAsync().listenAsync(new CommitListener(tx));
+        else
+            tx.rollbackAsync();
+    }
+
+    /**
+     * Commits or rolls back pessimistic transaction.
+     *
+     * @param tx Transaction to finish.
+     * @param commitInfo Commit information.
+     */
+    public void finishPessimisticTxOnRecovery(final IgniteTxEx<K, V> tx, GridCacheCommittedTxInfo<K, V> commitInfo) {
+        if (!tx.markFinalizing(RECOVERY_FINISH)) {
+            if (log.isDebugEnabled())
+                log.debug("Will not try to finish pessimistic transaction (could not mark as finalizing): " + tx);
+
+            return;
+        }
+
+        if (tx instanceof GridDistributedTxRemoteAdapter) {
+            IgniteTxRemoteEx<K,V> rmtTx = (IgniteTxRemoteEx<K, V>)tx;
+
+            rmtTx.doneRemote(tx.xidVersion(), Collections.<GridCacheVersion>emptyList(), Collections.<GridCacheVersion>emptyList(),
+                Collections.<GridCacheVersion>emptyList());
+        }
+
+        try {
+            tx.prepare();
+
+            if (commitInfo != null) {
+                for (IgniteTxEntry<K, V> entry : commitInfo.recoveryWrites()) {
+                    IgniteTxEntry<K, V> write = tx.writeMap().get(entry.txKey());
+
+                    if (write != null) {
+                        GridCacheEntryEx<K, V> cached = write.cached();
+
+                        IgniteTxEntry<K, V> recovered = entry.cleanCopy(write.context());
+
+                        if (cached == null || cached.detached())
+                            cached = write.context().cache().entryEx(entry.key(), tx.topologyVersion());
+
+                        recovered.cached(cached, cached.keyBytes());
+
+                        tx.writeMap().put(entry.txKey(), recovered);
+
+                        continue;
+                    }
+
+                    ((IgniteTxAdapter<K, V>)tx).recoveryWrites(commitInfo.recoveryWrites());
+
+                    // If write was not found, check read.
+                    IgniteTxEntry<K, V> read = tx.readMap().remove(entry.txKey());
+
+                    if (read != null)
+                        tx.writeMap().put(entry.txKey(), entry);
+                }
+
+                tx.commitAsync().listenAsync(new CommitListener(tx));
+            }
+            else
+                tx.rollbackAsync();
+        }
+        catch (IgniteCheckedException e) {
+            U.error(log, "Failed to prepare pessimistic transaction (will invalidate): " + tx, e);
+
+            salvageTx(tx);
+        }
+    }
+
+    /**
+     * @param req Check committed request.
+     * @return Check committed future.
+     */
+    public IgniteFuture<GridCacheCommittedTxInfo<K, V>> checkPessimisticTxCommitted(GridCachePessimisticCheckCommittedTxRequest req) {
+        // First check if we have near transaction with this ID.
+        IgniteTxEx<K, V> tx = localTxForRecovery(req.nearXidVersion(), !req.nearOnlyCheck());
+
+        // Either we found near transaction or one of transactions is being committed by user.
+        // Wait for it and send reply.
+        if (tx != null) {
+            assert tx.local();
+
+            if (log.isDebugEnabled())
+                log.debug("Found active near transaction, will wait for completion [req=" + req + ", tx=" + tx + ']');
+
+            final IgniteTxEx<K, V> tx0 = tx;
+
+            return tx.finishFuture().chain(new C1<IgniteFuture<IgniteTx>, GridCacheCommittedTxInfo<K, V>>() {
+                @Override public GridCacheCommittedTxInfo<K, V> apply(IgniteFuture<IgniteTx> txFut) {
+                    GridCacheCommittedTxInfo<K, V> info = null;
+
+                    if (tx0.state() == COMMITTED)
+                        info = new GridCacheCommittedTxInfo<>(tx0);
+
+                    return info;
+                }
+            });
+        }
+
+        GridCacheCommittedTxInfo<K, V> info = txCommitted(req.nearXidVersion(), req.originatingNodeId(),
+            req.originatingThreadId());
+
+        if (info == null)
+            info = txCommitted(req.nearXidVersion(), req.originatingNodeId(), req.originatingThreadId());
+
+        return new GridFinishedFutureEx<>(info);
+    }
+
+    /**
+     * Timeout object for node failure handler.
+     */
+    private final class NodeFailureTimeoutObject extends GridTimeoutObjectAdapter {
+        /** Left or failed node. */
+        private final UUID evtNodeId;
+
+        /**
+         * @param evtNodeId Event node ID.
+         */
+        private NodeFailureTimeoutObject(UUID evtNodeId) {
+            super(IgniteUuid.fromUuid(cctx.localNodeId()), TX_SALVAGE_TIMEOUT);
+
+            this.evtNodeId = evtNodeId;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onTimeout() {
+            try {
+                cctx.kernalContext().gateway().readLock();
+            }
+            catch (IllegalStateException ignore) {
+                if (log.isDebugEnabled())
+                    log.debug("Failed to acquire kernal gateway (grid is stopping).");
+
+                return;
+            }
+
+            try {
+                if (log.isDebugEnabled())
+                    log.debug("Processing node failed event [locNodeId=" + cctx.localNodeId() +
+                        ", failedNodeId=" + evtNodeId + ']');
+
+                for (IgniteTxEx<K, V> tx : txs()) {
+                    if ((tx.near() && !tx.local()) || (tx.storeUsed() && tx.masterNodeIds().contains(evtNodeId))) {
+                        // Invalidate transactions.
+                        salvageTx(tx, false, RECOVERY_FINISH);
+                    }
+                    else if (tx.optimistic()) {
+                        // Check prepare only if originating node ID failed. Otherwise parent node will finish this tx.
+                        if (tx.originatingNodeId().equals(evtNodeId)) {
+                            if (tx.state() == PREPARED)
+                                commitIfPrepared(tx);
+                            else {
+                                if (tx.setRollbackOnly())
+                                    tx.rollbackAsync();
+                                // If we could not mark tx as rollback, it means that transaction is being committed.
+                            }
+                        }
+                    }
+                    else {
+                        // Pessimistic.
+                        if (tx.originatingNodeId().equals(evtNodeId)) {
+                            if (tx.state() != COMMITTING && tx.state() != COMMITTED)
+                                commitIfRemotelyCommitted(tx);
+                            else {
+                                if (log.isDebugEnabled())
+                                    log.debug("Skipping pessimistic transaction check (transaction is being committed) " +
+                                        "[tx=" + tx + ", locNodeId=" + cctx.localNodeId() + ']');
+                            }
+                        }
+                        else {
+                            if (log.isDebugEnabled())
+                                log.debug("Skipping pessimistic transaction check [tx=" + tx +
+                                    ", evtNodeId=" + evtNodeId + ", locNodeId=" + cctx.localNodeId() + ']');
+                        }
+                    }
+                }
+            }
+            finally {
+                cctx.kernalContext().gateway().readUnlock();
+            }
+        }
+
+        /**
+         * Commits optimistic transaction in case when node started transaction failed, but all related
+         * transactions were prepared (invalidates transaction if it is not fully prepared).
+         *
+         * @param tx Transaction.
+         */
+        private void commitIfPrepared(IgniteTxEx<K, V> tx) {
+            assert tx instanceof GridDhtTxLocal || tx instanceof GridDhtTxRemote  : tx;
+            assert !F.isEmpty(tx.transactionNodes());
+            assert tx.nearXidVersion() != null;
+
+
+            GridCacheOptimisticCheckPreparedTxFuture<K, V> fut = new GridCacheOptimisticCheckPreparedTxFuture<>(
+                cctx, tx, evtNodeId, tx.transactionNodes());
+
+            cctx.mvcc().addFuture(fut);
+
+            if (log.isDebugEnabled())
+                log.debug("Checking optimistic transaction state on remote nodes [tx=" + tx + ", fut=" + fut + ']');
+
+            fut.prepare();
+        }
+
+        /**
+         * Commits pessimistic transaction if at least one of remote nodes has committed this transaction.
+         *
+         * @param tx Transaction.
+         */
+        private void commitIfRemotelyCommitted(IgniteTxEx<K, V> tx) {
+            assert tx instanceof GridDhtTxLocal || tx instanceof GridDhtTxRemote : tx;
+
+            GridCachePessimisticCheckCommittedTxFuture<K, V> fut = new GridCachePessimisticCheckCommittedTxFuture<>(
+                cctx, tx, evtNodeId);
+
+            cctx.mvcc().addFuture(fut);
+
+            if (log.isDebugEnabled())
+                log.debug("Checking pessimistic transaction state on remote nodes [tx=" + tx + ", fut=" + fut + ']');
+
+            fut.prepare();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class CommittedVersion extends GridCacheVersion {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** Corresponding near version. Transient. */
+        private GridCacheVersion nearVer;
+
+        /**
+         * Empty constructor required by {@link Externalizable}.
+         */
+        public CommittedVersion() {
+            // No-op.
+        }
+
+        /**
+         * @param ver Committed version.
+         * @param nearVer Near transaction version.
+         */
+        private CommittedVersion(GridCacheVersion ver, GridCacheVersion nearVer) {
+            super(ver.topologyVersion(), ver.globalTime(), ver.order(), ver.nodeOrder(), ver.dataCenterId());
+
+            assert nearVer != null;
+
+            this.nearVer = nearVer;
+        }
+    }
+
+    /**
+     * Atomic integer that compares only using references, not values.
+     */
+    private static final class AtomicInt extends AtomicInteger {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /**
+         * @param initVal Initial value.
+         */
+        private AtomicInt(int initVal) {
+            super(initVal);
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object obj) {
+            // Reference only.
+            return obj == this;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            return super.hashCode();
+        }
+    }
+
+    /**
+     * Commit listener. Checks if commit succeeded and rollbacks if case of error.
+     */
+    private class CommitListener implements CI1<IgniteFuture<IgniteTx>> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** Transaction. */
+        private final IgniteTxEx<K, V> tx;
+
+        /**
+         * @param tx Transaction.
+         */
+        private CommitListener(IgniteTxEx<K, V> tx) {
+            this.tx = tx;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void apply(IgniteFuture<IgniteTx> t) {
+            try {
+                t.get();
+            }
+            catch (IgniteTxOptimisticException ignore) {
+                if (log.isDebugEnabled())
+                    log.debug("Optimistic failure while committing prepared transaction (will rollback): " +
+                        tx);
+
+                tx.rollbackAsync();
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to commit transaction during failover: " + tx, e);
+            }
+        }
+    }
+}


Mime
View raw message