You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/03/14 15:16:07 UTC
[5/5] ignite git commit: ignite-4768 txs
ignite-4768 txs
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5523eac7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5523eac7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5523eac7
Branch: refs/heads/ignite-4768-1
Commit: 5523eac7d099da356732d12c3353e75c5dde044f
Parents: 901be4f
Author: sboikov <sb...@gridgain.com>
Authored: Tue Mar 14 16:39:11 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Mar 14 18:15:38 2017 +0300
----------------------------------------------------------------------
.../ignite/internal/IgniteTransactionsEx.java | 7 +-
.../processors/cache/GridCacheAdapter.java | 85 +-
.../processors/cache/GridCacheProxyImpl.java | 3 +-
.../cache/GridCacheSharedContext.java | 3 +-
.../processors/cache/IgniteInternalCache.java | 3 +-
.../GridDistributedCacheAdapter.java | 2 +-
.../GridDistributedTxRemoteAdapter.java | 22 +-
.../distributed/dht/GridDhtTxLocalAdapter.java | 10 +-
.../dht/colocated/GridDhtColocatedCache.java | 8 +-
.../near/GridNearTransactionalCache.java | 4 +-
.../cache/distributed/near/GridNearTxLocal.java | 2698 ++++++++++++++++-
.../store/GridCacheStoreManagerAdapter.java | 142 +-
.../cache/transactions/IgniteInternalTx.java | 8 -
.../transactions/IgniteTransactionsImpl.java | 12 +-
.../cache/transactions/IgniteTxAdapter.java | 29 +-
.../cache/transactions/IgniteTxEntry.java | 4 +-
.../cache/transactions/IgniteTxHandler.java | 4 +-
.../transactions/IgniteTxLocalAdapter.java | 2777 +-----------------
.../cache/transactions/IgniteTxLocalEx.java | 136 -
.../cache/transactions/IgniteTxManager.java | 142 +-
.../transactions/TransactionProxyImpl.java | 13 +-
.../HibernateReadWriteAccessStrategy.java | 11 +-
.../processors/cache/jta/CacheJtaManager.java | 3 +-
.../processors/cache/jta/CacheJtaResource.java | 7 +-
24 files changed, 3087 insertions(+), 3046 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/IgniteTransactionsEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteTransactionsEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteTransactionsEx.java
index 9772dcc..14791ed 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteTransactionsEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteTransactionsEx.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal;
import org.apache.ignite.IgniteTransactions;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
@@ -35,7 +36,7 @@ public interface IgniteTransactionsEx extends IgniteTransactions {
* @param txSize Number of entries participating in transaction (may be approximate).
* @return New transaction.
*/
- public IgniteInternalTx txStartEx(GridCacheContext ctx,
+ public GridNearTxLocal txStartEx(GridCacheContext ctx,
TransactionConcurrency concurrency,
TransactionIsolation isolation,
long timeout,
@@ -47,5 +48,7 @@ public interface IgniteTransactionsEx extends IgniteTransactions {
* @param isolation Isolation.
* @return New transaction.
*/
- public IgniteInternalTx txStartEx(GridCacheContext ctx, TransactionConcurrency concurrency, TransactionIsolation isolation);
+ public GridNearTxLocal txStartEx(GridCacheContext ctx,
+ TransactionConcurrency concurrency,
+ TransactionIsolation isolation);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 71be718..c98cd24 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -86,6 +86,7 @@ import org.apache.ignite.internal.processors.cache.distributed.IgniteExternaliza
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
@@ -1876,7 +1877,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
if (F.isEmpty(keys))
return new GridFinishedFuture<>(Collections.<K1, V1>emptyMap());
- IgniteTxLocalAdapter tx = null;
+ GridNearTxLocal tx = null;
if (checkTx) {
try {
@@ -2132,7 +2133,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
else {
return asyncOp(tx, new AsyncOp<Map<K1, V1>>(keys) {
- @Override public IgniteInternalFuture<Map<K1, V1>> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<Map<K1, V1>> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.getAllAsync(ctx,
readyTopVer,
keys,
@@ -2187,7 +2188,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
protected V getAndPut0(final K key, final V val, @Nullable final CacheEntryPredicate filter)
throws IgniteCheckedException {
return syncOp(new SyncOp<V>(true) {
- @Override public V op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public V op(GridNearTxLocal tx) throws IgniteCheckedException {
return (V)tx.putAsync(ctx, null, key, val, true, filter).get().value();
}
@@ -2237,7 +2238,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
@Nullable final CacheEntryPredicate filter)
{
return asyncOp(new AsyncOp<V>() {
- @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<V> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.putAsync(ctx, readyTopVer, key, val, true, filter)
.chain((IgniteClosure<IgniteInternalFuture<GridCacheReturn>, V>)RET2VAL);
}
@@ -2293,7 +2294,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
protected boolean put0(final K key, final V val, final CacheEntryPredicate filter)
throws IgniteCheckedException {
Boolean res = syncOp(new SyncOp<Boolean>(true) {
- @Override public Boolean op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public Boolean op(GridNearTxLocal tx) throws IgniteCheckedException {
return tx.putAsync(ctx, null, key, val, false, filter).get().success();
}
@@ -2316,7 +2317,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
syncOp(new SyncInOp(drMap.size() == 1) {
- @Override public void inOp(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public void inOp(GridNearTxLocal tx) throws IgniteCheckedException {
tx.putAllDrAsync(ctx, drMap).get();
}
@@ -2335,7 +2336,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
return asyncOp(new AsyncOp(drMap.keySet()) {
- @Override public IgniteInternalFuture op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.putAllDrAsync(ctx, drMap);
}
@@ -2380,7 +2381,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKey(key);
return syncOp(new SyncOp<EntryProcessorResult<T>>(true) {
- @Nullable @Override public EntryProcessorResult<T> op(IgniteTxLocalAdapter tx)
+ @Nullable @Override public EntryProcessorResult<T> op(GridNearTxLocal tx)
throws IgniteCheckedException {
assert topVer == null || tx.implicit();
@@ -2418,7 +2419,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKeys(keys);
return syncOp(new SyncOp<Map<K, EntryProcessorResult<T>>>(keys.size() == 1) {
- @Nullable @Override public Map<K, EntryProcessorResult<T>> op(IgniteTxLocalAdapter tx)
+ @Nullable @Override public Map<K, EntryProcessorResult<T>> op(GridNearTxLocal tx)
throws IgniteCheckedException {
Map<? extends K, EntryProcessor<K, V, Object>> invokeMap = F.viewAsMap(keys,
new C1<K, EntryProcessor<K, V, Object>>() {
@@ -2448,7 +2449,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKey(key);
IgniteInternalFuture<?> fut = asyncOp(new AsyncOp() {
- @Override public IgniteInternalFuture op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
Map<? extends K, EntryProcessor<K, V, Object>> invokeMap =
Collections.singletonMap(key, (EntryProcessor<K, V, Object>)entryProcessor);
@@ -2491,7 +2492,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKeys(keys);
IgniteInternalFuture<?> fut = asyncOp(new AsyncOp(keys) {
- @Override public IgniteInternalFuture<GridCacheReturn> op(IgniteTxLocalAdapter tx,
+ @Override public IgniteInternalFuture<GridCacheReturn> op(GridNearTxLocal tx,
AffinityTopologyVersion readyTopVer) {
Map<? extends K, EntryProcessor<K, V, Object>> invokeMap = F.viewAsMap(keys, new C1<K, EntryProcessor<K, V, Object>>() {
@Override public EntryProcessor apply(K k) {
@@ -2532,7 +2533,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKeys(map.keySet());
IgniteInternalFuture<?> fut = asyncOp(new AsyncOp(map.keySet()) {
- @Override public IgniteInternalFuture<GridCacheReturn> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<GridCacheReturn> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.invokeAsync(ctx,
readyTopVer,
(Map<? extends K, ? extends EntryProcessor<K, V, Object>>)map,
@@ -2568,7 +2569,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
validateCacheKeys(map.keySet());
return syncOp(new SyncOp<Map<K, EntryProcessorResult<T>>>(map.size() == 1) {
- @Nullable @Override public Map<K, EntryProcessorResult<T>> op(IgniteTxLocalAdapter tx)
+ @Nullable @Override public Map<K, EntryProcessorResult<T>> op(GridNearTxLocal tx)
throws IgniteCheckedException {
IgniteInternalFuture<GridCacheReturn> fut =
tx.invokeAsync(ctx, null, (Map<? extends K, ? extends EntryProcessor<K, V, Object>>)map, args);
@@ -2616,7 +2617,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
public IgniteInternalFuture<Boolean> putAsync0(final K key, final V val,
@Nullable final CacheEntryPredicate filter) {
return asyncOp(new AsyncOp<Boolean>() {
- @Override public IgniteInternalFuture<Boolean> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<Boolean> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.putAsync(ctx,
readyTopVer,
key,
@@ -2721,7 +2722,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected void putAll0(final Map<? extends K, ? extends V> m) throws IgniteCheckedException {
syncOp(new SyncInOp(m.size() == 1) {
- @Override public void inOp(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public void inOp(GridNearTxLocal tx) throws IgniteCheckedException {
tx.putAllAsync(ctx, null, m, false).get();
}
@@ -2748,7 +2749,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected IgniteInternalFuture<?> putAllAsync0(final Map<? extends K, ? extends V> m) {
return asyncOp(new AsyncOp(m.keySet()) {
- @Override public IgniteInternalFuture<?> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<?> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.putAllAsync(ctx,
readyTopVer,
m,
@@ -2789,7 +2790,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean keepBinary = ctx.keepBinary();
return syncOp(new SyncOp<V>(true) {
- @Override public V op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public V op(GridNearTxLocal tx) throws IgniteCheckedException {
K key0 = keepBinary ? (K) ctx.toCacheKeyObject(key) : key;
V ret = tx.removeAllAsync(ctx,
@@ -2839,7 +2840,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected IgniteInternalFuture<V> getAndRemoveAsync0(final K key) {
return asyncOp(new AsyncOp<V>() {
- @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<V> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
// TODO should we invoke interceptor here?
return tx.removeAllAsync(ctx,
readyTopVer,
@@ -2897,7 +2898,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected void removeAll0(final Collection<? extends K> keys) throws IgniteCheckedException {
syncOp(new SyncInOp(keys.size() == 1) {
- @Override public void inOp(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public void inOp(GridNearTxLocal tx) throws IgniteCheckedException {
tx.removeAllAsync(ctx,
null,
keys,
@@ -2938,7 +2939,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected IgniteInternalFuture<Object> removeAllAsync0(final Collection<? extends K> keys) {
return asyncOp(new AsyncOp(keys) {
- @Override public IgniteInternalFuture<?> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<?> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.removeAllAsync(ctx,
readyTopVer,
keys,
@@ -2990,7 +2991,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected boolean remove0(final K key, final CacheEntryPredicate filter) throws IgniteCheckedException {
Boolean res = syncOp(new SyncOp<Boolean>(true) {
- @Override public Boolean op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Override public Boolean op(GridNearTxLocal tx) throws IgniteCheckedException {
return tx.removeAllAsync(ctx,
null,
Collections.singletonList(key),
@@ -3046,7 +3047,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
protected IgniteInternalFuture<Boolean> removeAsync0(final K key, @Nullable final CacheEntryPredicate filter) {
return asyncOp(new AsyncOp<Boolean>() {
- @Override public IgniteInternalFuture<Boolean> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<Boolean> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.removeAllAsync(ctx,
readyTopVer,
Collections.singletonList(key),
@@ -3071,8 +3072,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
syncOp(new SyncInOp(false) {
- @Override public void inOp(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
- tx.removeAllDrAsync(ctx, (Map)drMap).get();
+ @Override public void inOp(GridNearTxLocal tx) throws IgniteCheckedException {
+ tx.removeAllDrAsync(ctx, drMap).get();
}
@Override public String toString() {
@@ -3090,8 +3091,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
return asyncOp(new AsyncOp(drMap.keySet()) {
- @Override public IgniteInternalFuture<?> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
- return tx.removeAllDrAsync(ctx, (Map)drMap);
+ @Override public IgniteInternalFuture<?> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
+ return tx.removeAllDrAsync(ctx, drMap);
}
@Override public String toString() {
@@ -3160,10 +3161,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
- @Nullable @Override public Transaction tx() {
- IgniteTxAdapter tx = ctx.tm().threadLocalTx(ctx);
-
- return tx == null ? null : new TransactionProxyImpl<>(tx, ctx.shared(), false);
+ @Nullable @Override public GridNearTxLocal tx() {
+ return ctx.tm().threadLocalTx(ctx);
}
/** {@inheritDoc} */
@@ -4142,7 +4141,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @return Transaction commit future.
*/
@SuppressWarnings("unchecked")
- public IgniteInternalFuture<IgniteInternalTx> commitTxAsync(final IgniteInternalTx tx) {
+ IgniteInternalFuture<IgniteInternalTx> commitTxAsync(final GridNearTxLocal tx) {
FutureHolder holder = lastFut.get();
holder.lock();
@@ -4208,7 +4207,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
awaitLastFut();
- IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+ GridNearTxLocal tx = ctx.tm().threadLocalTx(ctx);
if (tx == null || tx.implicit()) {
TransactionConfiguration tCfg = CU.transactionConfiguration(ctx, ctx.kernalContext().config());
@@ -4304,7 +4303,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
if (log.isDebugEnabled())
log.debug("Performing async op: " + op);
- IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+ GridNearTxLocal tx = ctx.tm().threadLocalTx(ctx);
CacheOperationContext opCtx = ctx.operationContextPerCall();
@@ -4348,7 +4347,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
@SuppressWarnings("unchecked")
protected <T> IgniteInternalFuture<T> asyncOp(
- IgniteTxLocalAdapter tx,
+ GridNearTxLocal tx,
final AsyncOp<T> op,
final CacheOperationContext opCtx
) {
@@ -4364,7 +4363,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
try {
IgniteInternalFuture fut = holder.future();
- final IgniteTxLocalAdapter tx0 = tx;
+ final GridNearTxLocal tx0 = tx;
if (fut != null && !fut.isDone()) {
IgniteInternalFuture<T> f = new GridEmbeddedFuture(fut,
@@ -4925,7 +4924,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
private int retries;
/** */
- private IgniteTxLocalAdapter tx;
+ private GridNearTxLocal tx;
/** */
private CacheOperationContext opCtx;
@@ -5173,7 +5172,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @return Operation return value.
* @throws IgniteCheckedException If failed.
*/
- @Nullable public abstract T op(IgniteTxLocalAdapter tx) throws IgniteCheckedException;
+ @Nullable public abstract T op(GridNearTxLocal tx) throws IgniteCheckedException;
}
/**
@@ -5188,7 +5187,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
- @Nullable @Override public final Object op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
+ @Nullable @Override public final Object op(GridNearTxLocal tx) throws IgniteCheckedException {
inOp(tx);
return null;
@@ -5198,7 +5197,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param tx Transaction.
* @throws IgniteCheckedException If failed.
*/
- public abstract void inOp(IgniteTxLocalAdapter tx) throws IgniteCheckedException;
+ public abstract void inOp(GridNearTxLocal tx) throws IgniteCheckedException;
}
/**
@@ -5234,14 +5233,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param readyTopVer Ready topology version.
* @return Operation future.
*/
- public abstract IgniteInternalFuture<T> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer);
+ public abstract IgniteInternalFuture<T> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer);
/**
* @param tx Transaction.
* @param opCtx Operation context.
* @return Operation future.
*/
- public IgniteInternalFuture<T> op(final IgniteTxLocalAdapter tx, CacheOperationContext opCtx) {
+ public IgniteInternalFuture<T> op(final GridNearTxLocal tx, CacheOperationContext opCtx) {
AffinityTopologyVersion txTopVer = tx.topologyVersionSnapshot();
if (txTopVer != null)
@@ -5267,7 +5266,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
private IgniteInternalFuture<T> waitTopologyFuture(IgniteInternalFuture<?> topFut,
final AffinityTopologyVersion topVer,
- final IgniteTxLocalAdapter tx,
+ final GridNearTxLocal tx,
final CacheOperationContext opCtx) {
final GridFutureAdapter fut0 = new GridFutureAdapter();
@@ -5304,7 +5303,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param opCtx Operation context.
* @return Future.
*/
- private IgniteInternalFuture<T> runOp(IgniteTxLocalAdapter tx,
+ private IgniteInternalFuture<T> runOp(GridNearTxLocal tx,
AffinityTopologyVersion topVer,
CacheOperationContext opCtx) {
ctx.operationContextPerCall(opCtx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 00898ec..731d23b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -40,6 +40,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.affinity.GridCacheAffinityProxy;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -977,7 +978,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
- @Override public Transaction tx() {
+ @Override public GridNearTxLocal tx() {
CacheOperationContext prev = gate.enter(opCtx);
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index 0f79100..e4f8fec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.managers.deployment.GridDeploymentManager;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.jta.CacheJtaManagerAdapter;
import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
@@ -742,7 +743,7 @@ public class GridCacheSharedContext<K, V> {
* @return Commit future.
*/
@SuppressWarnings("unchecked")
- public IgniteInternalFuture<IgniteInternalTx> commitTxAsync(IgniteInternalTx tx) {
+ public IgniteInternalFuture<IgniteInternalTx> commitTxAsync(GridNearTxLocal tx) {
GridCacheContext ctx = tx.txState().singleCacheContext(this);
if (ctx == null) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
index 0ac98fb..02b6461 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
@@ -43,6 +43,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -976,7 +977,7 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
* @return Transaction started by this thread or {@code null} if this thread
* does not have a transaction.
*/
- @Nullable public Transaction tx();
+ @Nullable public GridNearTxLocal tx();
/**
* Evicts entry associated with given key from cache. Note, that entry will be evicted
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
index 7e4deff..00bc6d7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
@@ -111,7 +111,7 @@ public abstract class GridDistributedCacheAdapter<K, V> extends GridCacheAdapter
/** {@inheritDoc} */
@Override public IgniteInternalFuture<Boolean> lockAllAsync(Collection<? extends K> keys, long timeout) {
- IgniteTxLocalEx tx = ctx.tm().userTxx();
+ IgniteTxLocalEx tx = ctx.tm().userTx();
// Return value flag is true because we choose to bring values for explicit locks.
return lockAllAsync(ctx.cacheKeysView(keys),
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 68c0e57..e60d3dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -83,7 +83,7 @@ import static org.apache.ignite.transactions.TransactionState.UNKNOWN;
/**
* Transaction created by system implicitly on remote nodes.
*/
-public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
+public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
implements IgniteTxRemoteEx {
/** */
private static final long serialVersionUID = 0L;
@@ -180,11 +180,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/** {@inheritDoc} */
- @Override public Collection<UUID> masterNodeIds() {
- return Collections.singleton(nodeId);
- }
-
- /** {@inheritDoc} */
@Override public UUID originatingNodeId() {
return nodeId;
}
@@ -378,11 +373,9 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/**
- * Prepare phase.
- *
- * @throws IgniteCheckedException If prepare failed.
+ * @throws IgniteCheckedException If failed.
*/
- @Override public void prepare() throws IgniteCheckedException {
+ public final void prepareRemoteTx() throws IgniteCheckedException {
// If another thread is doing prepare or rollback.
if (!state(PREPARING)) {
// In optimistic mode prepare may be called multiple times.
@@ -408,6 +401,15 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/**
+ * Prepare phase.
+ *
+ * @throws IgniteCheckedException If prepare failed.
+ */
+ @Override public void prepare() throws IgniteCheckedException {
+ prepareRemoteTx();
+ }
+
+ /**
* @throws IgniteCheckedException If commit failed.
*/
@SuppressWarnings({"CatchGenericClass"})
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 67e1993..c7967d3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -161,7 +161,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
/**
* @param node Node.
*/
- public void addLockTransactionNode(ClusterNode node) {
+ void addLockTransactionNode(ClusterNode node) {
assert node != null;
assert !node.isLocal();
@@ -185,7 +185,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
*
* @return Has near cache flag.
*/
- public boolean nearOnOriginatingNode() {
+ boolean nearOnOriginatingNode() {
return nearOnOriginatingNode;
}
@@ -206,7 +206,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
/**
* @return Nodes where transactions were started on lock step.
*/
- @Nullable public Set<ClusterNode> lockTransactionNodes() {
+ @Nullable Set<ClusterNode> lockTransactionNodes() {
return lockTxNodes;
}
@@ -349,14 +349,14 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
/**
* @param mappings Mappings to add.
*/
- void addDhtNodeEntryMapping(Map<ClusterNode, List<GridDhtCacheEntry>> mappings) {
+ private void addDhtNodeEntryMapping(Map<ClusterNode, List<GridDhtCacheEntry>> mappings) {
addMapping(mappings, dhtMap);
}
/**
* @param mappings Mappings to add.
*/
- void addNearNodeEntryMapping(Map<ClusterNode, List<GridDhtCacheEntry>> mappings) {
+ private void addNearNodeEntryMapping(Map<ClusterNode, List<GridDhtCacheEntry>> mappings) {
addMapping(mappings, nearMap);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index e1e0ec2..03bbfe0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -207,13 +207,13 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
if (keyCheck)
validateCacheKey(key);
- IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+ GridNearTxLocal tx = ctx.tm().threadLocalTx(ctx);
final CacheOperationContext opCtx = ctx.operationContextPerCall();
if (tx != null && !tx.implicit() && !skipTx) {
return asyncOp(tx, new AsyncOp<V>() {
- @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<V> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
IgniteInternalFuture<Map<Object, Object>> fut = tx.getAllAsync(ctx,
readyTopVer,
Collections.singleton(ctx.toCacheKeyObject(key)),
@@ -289,13 +289,13 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
if (keyCheck)
validateCacheKeys(keys);
- IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+ GridNearTxLocal tx = ctx.tm().threadLocalTx(ctx);
final CacheOperationContext opCtx = ctx.operationContextPerCall();
if (tx != null && !tx.implicit() && !skipTx) {
return asyncOp(tx, new AsyncOp<Map<K, V>>(keys) {
- @Override public IgniteInternalFuture<Map<K, V>> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<Map<K, V>> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.getAllAsync(ctx,
readyTopVer,
ctx.cacheKeysView(keys),
http://git-wip-us.apache.org/repos/asf/ignite/blob/5523eac7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index 940dd80..c39af34 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -133,7 +133,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
if (keyCheck)
validateCacheKeys(keys);
- IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
+ GridNearTxLocal tx = ctx.tm().threadLocalTx(ctx);
CacheOperationContext opCtx = ctx.operationContextPerCall();
@@ -141,7 +141,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
if (tx != null && !tx.implicit() && !skipTx) {
return asyncOp(tx, new AsyncOp<Map<K, V>>(keys) {
- @Override public IgniteInternalFuture<Map<K, V>> op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
+ @Override public IgniteInternalFuture<Map<K, V>> op(GridNearTxLocal tx, AffinityTopologyVersion readyTopVer) {
return tx.getAllAsync(ctx,
readyTopVer,
ctx.cacheKeysView(keys),