ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [40/46] incubator-ignite git commit: GG-9141 - Renaming.
Date Sun, 21 Dec 2014 23:04:33 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cd46f567/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalEx.java
deleted file mode 100644
index 524243c..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalEx.java
+++ /dev/null
@@ -1,166 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.kernal.processors.cache;
-
-import org.apache.ignite.*;
-import org.apache.ignite.lang.*;
-import org.gridgain.grid.cache.*;
-import org.gridgain.grid.kernal.processors.cache.dr.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-
-/**
- * Local transaction API.
- */
-public interface GridCacheTxLocalEx<K, V> extends GridCacheTxEx<K, V> {
-    /**
-     * @return Minimum version involved in transaction.
-     */
-    public GridCacheVersion minVersion();
-
-    /**
-     * @return Future for this transaction.
-     */
-    public IgniteFuture<GridCacheTxEx<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 GridCacheTxEntry<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/GridCacheTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxManager.java
deleted file mode 100644
index 7645f75..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxManager.java
+++ /dev/null
@@ -1,2212 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.kernal.processors.cache;
-
-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.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.GridCacheTxEx.FinalizationStatus.*;
-import static org.gridgain.grid.kernal.processors.cache.GridCacheUtils.*;
-import static org.gridgain.grid.util.GridConcurrentFactory.*;
-
-/**
- * Cache transaction manager.
- */
-public class GridCacheTxManager<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<GridCacheTxEx> threadCtx = new GridThreadLocalEx<>();
-
-    /** Per-thread transaction map. */
-    private final ConcurrentMap<Long, GridCacheTxEx<K, V>> threadMap = newMap();
-
-    /** Per-ID map. */
-    private final ConcurrentMap<GridCacheVersion, GridCacheTxEx<K, V>> idMap = newMap();
-
-    /** Per-ID map for near transactions. */
-    private final ConcurrentMap<GridCacheVersion, GridCacheTxEx<K, V>> nearIdMap = newMap();
-
-    /** TX handler. */
-    private GridCacheTxHandler<K, V> txHandler;
-
-    /** All transactions. */
-    private final Queue<GridCacheTxEx<K, V>> committedQ = new ConcurrentLinkedDeque8<>();
-
-    /** Preparing transactions. */
-    private final Queue<GridCacheTxEx<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 (GridCacheTxEx<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 GridCacheTxHandler<>(cctx);
-    }
-
-    /**
-     * @return TX handler.
-     */
-    public GridCacheTxHandler<K, V> txHandler() {
-        return txHandler;
-    }
-
-    /**
-     * Invalidates transaction.
-     *
-     * @param tx Transaction.
-     * @return {@code True} if transaction was salvaged by this call.
-     */
-    public boolean salvageTx(GridCacheTxEx<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(GridCacheTxEx<K, V> tx, boolean warn, GridCacheTxEx.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 GridCacheTxRemoteEx) {
-                    GridCacheTxRemoteEx<K, V> rmtTx = (GridCacheTxRemoteEx<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() {
-        GridCacheTxEx<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);
-
-            GridCacheTxEx<K, V> stuck = null;
-
-            for (GridCacheTxEx<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(GridCacheTxEx<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 GridCacheTxLocalAdapter<K, V> newTx(
-        boolean implicit,
-        boolean implicitSingle,
-        boolean sys,
-        IgniteTxConcurrency concurrency,
-        IgniteTxIsolation isolation,
-        long timeout,
-        boolean invalidate,
-        boolean storeEnabled,
-        int txSize,
-        @Nullable GridCacheTxKey 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 GridCacheTxEx<K, V>> T onCreated(T tx) {
-        ConcurrentMap<GridCacheVersion, GridCacheTxEx<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;
-        }
-
-        GridCacheTxEx<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 (GridCacheTxEx<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(GridCacheTxEx<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) {
-        GridCacheTxEx<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, GridCacheTxEx<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() {
-        GridCacheTxEx<K, V> tx = tx();
-
-        return tx != null && tx.local() ? (T)tx : null;
-    }
-
-    /**
-     * @return Transaction for current thread.
-     */
-    @SuppressWarnings({"unchecked"})
-    public <T> T threadLocalTx() {
-        GridCacheTxEx<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() {
-        GridCacheTxEx<K, V> tx = txContext();
-
-        return tx != null ? (T)tx : (T)tx(Thread.currentThread().getId());
-    }
-
-    /**
-     * @return Local transaction.
-     */
-    @Nullable public GridCacheTxEx<K, V> localTxx() {
-        GridCacheTxEx<K, V> tx = txx();
-
-        return tx != null && tx.local() ? tx : null;
-    }
-
-    /**
-     * @return Transaction for current thread.
-     */
-    @SuppressWarnings({"unchecked"})
-    public GridCacheTxEx<K, V> txx() {
-        return tx();
-    }
-
-    /**
-     * @return User transaction for current thread.
-     */
-    @Nullable public IgniteTx userTx() {
-        GridCacheTxEx<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 GridCacheTxLocalEx<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 GridCacheTxEx<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 GridCacheTxEx<K, V>> T nearTx(GridCacheVersion txId) {
-        return (T)nearIdMap.get(txId);
-    }
-
-    /**
-     * @param txId Transaction ID.
-     * @return Transaction with given ID.
-     */
-    @Nullable public GridCacheTxEx<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(GridCacheTxEx<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<GridCacheTxEx<K, V>> it = committedQ.iterator(); it.hasNext();) {
-                    GridCacheTxEx<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<GridCacheTxKey<K>> readSet = tx.readSet();
-            Set<GridCacheTxKey<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<GridCacheTxEx<K, V>> it = committedQ.iterator(); it.hasNext();) {
-                GridCacheTxEx<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<GridCacheTxEx<K, V>> iter = prepareQ.iterator(); iter.hasNext();) {
-                GridCacheTxEx<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<GridCacheTxKey<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(GridCacheTxEx<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(GridCacheTxEx<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(GridCacheTxEx<K, V> tx) {
-        Collection<GridCacheTxEntry<K, V>> entries = (tx.local() && !tx.dht()) ? tx.allEntries() : tx.writeEntries();
-
-        for (GridCacheTxEntry<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(GridCacheTxEx<K, V> tx) {
-        committedVers.remove(tx.xidVersion());
-    }
-
-    /**
-     * @param tx Committed transaction.
-     * @return If transaction was not already present in committed set.
-     */
-    public boolean addCommittedTx(GridCacheTxEx<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(GridCacheTxEx<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(GridCacheTxEx<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<GridCacheTxEntry<K, V>> entries,
-        GridCacheVersion baseVer, Set<GridCacheVersion> vers) {
-
-        // The locks are not released yet, so we can safely list pending candidates versions.
-        for (GridCacheTxEntry<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<GridCacheTxEntry<K, V>> entries, GridCacheVersion min,
-        GridCacheTxEx<K, V> tx) {
-        for (GridCacheTxEntry<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(GridCacheTxEx<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, GridCacheTxEx<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(GridCacheTxEx<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, GridCacheTxEx<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(GridCacheTxEx<K, V> tx) {
-        assert tx != null;
-
-        if (log.isDebugEnabled())
-            log.debug("Uncommiting from TM: " + tx);
-
-        ConcurrentMap<GridCacheVersion, GridCacheTxEx<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, GridCacheTxEx<K, V>> transactionMap(GridCacheTxEx<K, V> tx) {
-        return (tx.near() && !tx.local()) ? nearIdMap : idMap;
-    }
-
-    /**
-     * @param tx Transaction to notify evictions for.
-     */
-    private void notifyEvitions(GridCacheTxEx<K, V> tx) {
-        if (tx.internal() && !tx.groupLock())
-            return;
-
-        for (GridCacheTxEntry<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) {
-            GridCacheTxAdapter<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(GridCacheTxEx<K, V> tx, Iterable<GridCacheTxEntry<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 (GridCacheTxEntry<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 (GridCacheTxEntry<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(GridCacheTxEx txx) {
-        GridCacheTxKey grpLockKey = txx.groupLockKey();
-
-        assert grpLockKey != null;
-
-        if (grpLockKey == null)
-            return;
-
-        GridCacheTxEntry 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(GridCacheTxEx<K, V> tx, Iterable<GridCacheTxEntry<K, V>> entries) {
-        for (GridCacheTxEntry<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(GridCacheTxEx tx) {
-        threadCtx.set(tx);
-    }
-
-    /**
-     * @return Currently committing transaction.
-     */
-    @SuppressWarnings({"unchecked"})
-    private GridCacheTxEx<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() {
-        GridCacheTxEx<K, V> tx = txContext();
-
-        return tx == null ? null : tx.xidVersion();
-    }
-
-    /**
-     * Commit ended.
-     */
-    public void txContextReset() {
-        threadCtx.set(null);
-    }
-
-    /**
-     * @return All transactions.
-     */
-    public Collection<GridCacheTxEx<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 (GridCacheTxEx<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(GridCacheTxEx<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 GridCacheTxEx<K, V> localTxForRecovery(GridCacheVersion nearXidVer, boolean markFinalizing) {
-        // First check if we have near transaction with this ID.
-        GridCacheTxEx<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 (GridCacheTxEx<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 GridCacheTxEx<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) {
-            GridCacheTxRemoteEx<K,V> rmtTx = (GridCacheTxRemoteEx<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 GridCacheTxEx<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) {
-            GridCacheTxRemoteEx<K,V> rmtTx = (GridCacheTxRemoteEx<K, V>)tx;
-
-            rmtTx.doneRemote(tx.xidVersion(), Collections.<GridCacheVersion>emptyList(), Collections.<GridCacheVersion>emptyList(),
-                Collections.<GridCacheVersion>emptyList());
-        }
-
-        try {
-            tx.prepare();
-
-            if (commitInfo != null) {
-                for (GridCacheTxEntry<K, V> entry : commitInfo.recoveryWrites()) {
-                    GridCacheTxEntry<K, V> write = tx.writeMap().get(entry.txKey());
-
-                    if (write != null) {
-                        GridCacheEntryEx<K, V> cached = write.cached();
-
-                        GridCacheTxEntry<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;
-                    }
-
-                    ((GridCacheTxAdapter<K, V>)tx).recoveryWrites(commitInfo.recoveryWrites());
-
-                    // If write was not found, check read.
-                    GridCacheTxEntry<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.
-        GridCacheTxEx<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 GridCacheTxEx<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 (GridCacheTxEx<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(GridCacheTxEx<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(GridCacheTxEx<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 GridCacheTxEx<K, V> tx;
-
-        /**
-         * @param tx Transaction.
-         */
-        private CommitListener(GridCacheTxEx<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