Return-Path: X-Original-To: apmail-ignite-commits-archive@minotaur.apache.org Delivered-To: apmail-ignite-commits-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 3B47918127 for ; Tue, 19 Jan 2016 18:59:15 +0000 (UTC) Received: (qmail 55394 invoked by uid 500); 19 Jan 2016 18:59:15 -0000 Delivered-To: apmail-ignite-commits-archive@ignite.apache.org Received: (qmail 55357 invoked by uid 500); 19 Jan 2016 18:59:15 -0000 Mailing-List: contact commits-help@ignite.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@ignite.apache.org Delivered-To: mailing list commits@ignite.apache.org Received: (qmail 55347 invoked by uid 99); 19 Jan 2016 18:59:15 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 19 Jan 2016 18:59:15 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 05655DFCB5; Tue, 19 Jan 2016 18:59:15 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: ntikhonov@apache.org To: commits@ignite.apache.org Message-Id: X-Mailer: ASF-Git Admin Mailer Subject: ignite git commit: IGNITE-10837 WIP Date: Tue, 19 Jan 2016 18:59:15 +0000 (UTC) Repository: ignite Updated Branches: refs/heads/ignite-gg-10837 76685e376 -> 85e22a278 IGNITE-10837 WIP Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/85e22a27 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/85e22a27 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/85e22a27 Branch: refs/heads/ignite-gg-10837 Commit: 85e22a2785d62081348da3f5b07be2a0e4c110b4 Parents: 76685e3 Author: nikolay_tikhonov Authored: Tue Jan 19 21:58:49 2016 +0300 Committer: nikolay_tikhonov Committed: Tue Jan 19 21:58:49 2016 +0300 ---------------------------------------------------------------------- .../processors/cache/IgniteCacheProxy.java | 48 ------------- .../dht/atomic/GridDhtAtomicCache.java | 74 ++++++++------------ .../processors/cache/dr/GridCacheDrInfo.java | 8 +-- .../transactions/IgniteTxLocalAdapter.java | 12 ++-- 4 files changed, 40 insertions(+), 102 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/85e22a27/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java index 8cf7285..fb549eb 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java @@ -65,10 +65,8 @@ import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; -import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo; import org.apache.ignite.internal.processors.cache.query.CacheQuery; import org.apache.ignite.internal.processors.cache.query.CacheQueryFuture; -import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.query.GridQueryProcessor; import org.apache.ignite.internal.util.GridCloseableIteratorAdapter; import org.apache.ignite.internal.util.GridEmptyIterator; @@ -1082,52 +1080,6 @@ public class IgniteCacheProxy extends AsyncSupportAdapter conflictMap) { - try { - GridCacheGateway gate = this.gate; - - CacheOperationContext prev = onEnter(gate, opCtx); - - try { - delegate.putAllConflict(conflictMap); - } - finally { - onLeave(gate, prev); - } - } - catch (IgniteCheckedException e) { - throw cacheException(e); - } - } - - /** - * Remove entries from cache with conflict resolution. - * - * @param conflictMap Conflict map. - */ - public void removeAllConflict(Map conflictMap) { - try { - GridCacheGateway gate = this.gate; - - CacheOperationContext prev = onEnter(gate, opCtx); - - try { - delegate.removeAllConflict(conflictMap); - } - finally { - onLeave(gate, prev); - } - } - catch (IgniteCheckedException e) { - throw cacheException(e); - } - } - /** {@inheritDoc} */ @Override public V getAndPut(K key, V val) { try { http://git-wip-us.apache.org/repos/asf/ignite/blob/85e22a27/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java index 2c45121..4353232 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java @@ -37,7 +37,6 @@ import javax.cache.processor.EntryProcessor; import javax.cache.processor.EntryProcessorResult; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; -import org.apache.ignite.cache.CacheEntryProcessor; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; @@ -99,6 +98,7 @@ import org.apache.ignite.internal.util.typedef.internal.CU; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteBiTuple; +import org.apache.ignite.lang.IgniteClosure; import org.apache.ignite.lang.IgniteOutClosure; import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.plugin.security.SecurityPermission; @@ -620,21 +620,6 @@ public class GridDhtAtomicCache extends GridDhtCacheAdapter { } /** {@inheritDoc} */ - public void invokeAllConflict(Map conflictMap, Object... args) - throws IgniteCheckedException { - updateAllAsync0(null, - null, - args, - conflictMap, - null, - false, - false, - null, - true, - TRANSFORM).get(); - } - - /** {@inheritDoc} */ @Override public V getAndRemove(K key) throws IgniteCheckedException { return getAndRemoveAsync(key).get(); } @@ -943,41 +928,43 @@ public class GridDhtAtomicCache extends GridDhtCacheAdapter { ctx.checkSecurity(SecurityPermission.CACHE_PUT); - CacheOperationContext opCtx = ctx.operationContextPerCall(); + final CacheOperationContext opCtx = ctx.operationContextPerCall(); if (opCtx != null && opCtx.hasDataCenterId()) { assert conflictPutMap == null : conflictPutMap; assert conflictRmvMap == null : conflictRmvMap; if (op == GridCacheOperation.TRANSFORM) { - Map confMap = new HashMap<>(invokeMap.size()); - - for (Map.Entry e : invokeMap.entrySet()) - confMap.put(ctx.toCacheKeyObject(e.getKey()), new GridCacheDrInfo((CacheEntryProcessor)e.getValue(), - ctx.versions().next(opCtx.dataCenterId()))); + assert invokeMap != null : invokeMap; - conflictPutMap = confMap; + conflictPutMap = F.viewReadOnly((Map)invokeMap, + new IgniteClosure() { + @Override public GridCacheDrInfo apply(EntryProcessor o) { + return new GridCacheDrInfo(o, ctx.versions().next(opCtx.dataCenterId())); + } + }); invokeMap = null; } else if (op == GridCacheOperation.DELETE) { - Map confMap = new HashMap<>(map.size()); - - for (K key : map.keySet()) - confMap.put(ctx.toCacheKeyObject(key), ctx.versions().next(opCtx.dataCenterId())); + assert map != null : map; - conflictRmvMap = confMap; + conflictRmvMap = F.viewReadOnly((Map)map, new IgniteClosure() { + @Override public GridCacheVersion apply(V o) { + return ctx.versions().next(opCtx.dataCenterId()); + } + }); map = null; } else { - Map confMap = new HashMap<>(map.size()); + assert map != null : map; - for (Map.Entry e : map.entrySet()) - confMap.put(ctx.toCacheKeyObject(e.getKey()), new GridCacheDrInfo( - ctx.toCacheObject(e.getValue()), ctx.versions().next(opCtx.dataCenterId()))); - - conflictPutMap = confMap; + conflictPutMap = F.viewReadOnly((Map)map, new IgniteClosure() { + @Override public GridCacheDrInfo apply(V o) { + return new GridCacheDrInfo(ctx.toCacheObject(o), ctx.versions().next(opCtx.dataCenterId())); + } + }); map = null; } @@ -1048,23 +1035,22 @@ public class GridDhtAtomicCache extends GridDhtCacheAdapter { ctx.checkSecurity(SecurityPermission.CACHE_REMOVE); - CacheOperationContext opCtx = ctx.operationContextPerCall(); + final CacheOperationContext opCtx = ctx.operationContextPerCall(); UUID subjId = ctx.subjectIdPerCall(null, opCtx); int taskNameHash = ctx.kernalContext().job().currentTaskNameHash(); + Collection drVers = null; + if (opCtx != null && keys != null && opCtx.hasDataCenterId()) { assert conflictMap == null : conflictMap; - Map confMap = new HashMap<>(keys.size()); - - for (K key : keys) - confMap.put(ctx.toCacheKeyObject(key), ctx.versions().next(opCtx.dataCenterId())); - - conflictMap = confMap; - - keys = null; + drVers = F.transform(keys, new C1() { + @Override public GridCacheVersion apply(K k) { + return ctx.versions().next(opCtx.dataCenterId()); + } + }); } final GridNearAtomicUpdateFuture updateFut = new GridNearAtomicUpdateFuture( @@ -1076,7 +1062,7 @@ public class GridDhtAtomicCache extends GridDhtCacheAdapter { null, null, null, - keys != null ? null : conflictMap.values(), + drVers != null ? drVers : (keys != null ? null : conflictMap.values()), retval, rawRetval, (filter != null && opCtx != null) ? opCtx.expiry() : null, http://git-wip-us.apache.org/repos/asf/ignite/blob/85e22a27/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrInfo.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrInfo.java index bd3df02..02bc6b5 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrInfo.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrInfo.java @@ -21,7 +21,7 @@ import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; import java.io.ObjectOutput; -import org.apache.ignite.cache.CacheEntryProcessor; +import javax.cache.processor.EntryProcessor; import org.apache.ignite.internal.processors.cache.CacheObject; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.util.typedef.internal.CU; @@ -38,7 +38,7 @@ public class GridCacheDrInfo implements Externalizable { private CacheObject val; /** Entry processor. */ - private CacheEntryProcessor proc; + private EntryProcessor proc; /** DR version. */ private GridCacheVersion ver; @@ -79,7 +79,7 @@ public class GridCacheDrInfo implements Externalizable { * @param proc Entry processor. * @param ver Version. */ - public GridCacheDrInfo(CacheEntryProcessor proc, GridCacheVersion ver) { + public GridCacheDrInfo(EntryProcessor proc, GridCacheVersion ver) { assert proc != null; assert ver != null; @@ -97,7 +97,7 @@ public class GridCacheDrInfo implements Externalizable { /** * @return Entry processor. */ - public CacheEntryProcessor entryProcessor() { + public EntryProcessor entryProcessor() { return proc; } http://git-wip-us.apache.org/repos/asf/ignite/blob/85e22a27/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java index 0b19f30..32a9e27 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java @@ -22,13 +22,13 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.TreeMap; import java.util.UUID; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; @@ -3087,8 +3087,9 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig if (opCtx != null && opCtx.hasDataCenterId()) { assert drMap == null : drMap; + assert map != null || invokeMap != null; - drMap = (Map)F.viewReadOnly((Map) map, + drMap = (Map)F.viewReadOnly((Map)(map != null ? map : invokeMap), new IgniteClosure() { @Override public GridCacheDrInfo apply(V val) { return new GridCacheDrInfo(cctx.versions().next(opCtx.dataCenterId())); @@ -3348,10 +3349,9 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); - if (drMap == null && opCtx != null && opCtx.hasDataCenterId()) { - assert drMap == null : drMap; - - Map confMap = new TreeMap<>(); + // DrMap used as map. Keys order doesn't matter. + if (keys != null && drMap == null && opCtx != null && opCtx.hasDataCenterId()) { + Map confMap = new HashMap<>(keys.size()); for (K key : keys) confMap.put(key, cacheCtx.versions().next(opCtx.dataCenterId()));