From 5df2b65e50104347aaf090795cf6d29a88477741 Mon Sep 17 00:00:00 2001 From: Yakov Zhdanov Date: Fri, 6 Feb 2015 17:47:18 +0300 Subject: [PATCH] Merge branch 'sprint-1' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-96 Conflicts: modules/core/src/main/java/org/apache/ignite/cache/eviction/EvictableEntry.java modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java --- .../processors/cache/CacheEntryImpl.java | 19 +++++ .../processors/cache/GridCacheAdapter.java | 32 +++++++-- .../cache/GridCacheEvictionManager.java | 2 +- .../processors/cache/GridCacheIterator.java | 14 ++-- .../processors/cache/GridCacheKeySet.java | 14 ++-- .../processors/cache/GridCacheMapAdapter.java | 5 +- .../processors/cache/GridCacheMapEntry.java | 6 +- .../cache/GridCacheValueCollection.java | 13 ++-- .../distributed/dht/GridDhtCacheAdapter.java | 6 +- .../near/GridNearCacheAdapter.java | 11 +-- .../GridCacheDistributedQueryManager.java | 1 + .../GridCacheContinuousQueryHandler.java | 4 +- .../ignite/internal/util/lang/GridFunc.java | 69 ++----------------- .../cache/GridCacheAbstractSelfTest.java | 7 +- 14 files changed, 98 insertions(+), 105 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryImpl.java index d6feefc6ade59..e5a084e18db89 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryImpl.java @@ -17,6 +17,8 @@ package org.apache.ignite.internal.processors.cache; +import org.jetbrains.annotations.*; + import javax.cache.*; /** @@ -29,6 +31,9 @@ public class CacheEntryImpl implements Cache.Entry { /** */ private final V val; + /** Version. */ + private Object ver; + /** * @param key Key. * @param val Value. @@ -48,6 +53,20 @@ public CacheEntryImpl(K key, V val) { return val; } + /** + * @return Version. + */ + @Nullable public Object version() { + return ver; + } + + /** + * @param ver Version. + */ + public void version(Object ver) { + this.ver = ver; + } + /** {@inheritDoc} */ @SuppressWarnings("unchecked") @Override public T unwrap(Class cls) { 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 378bd418fd81b..668091f758b5b 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 @@ -1304,7 +1304,8 @@ public Set> allEntries() { /** {@inheritDoc} */ @Override public Set> primaryEntrySetx(IgnitePredicate>... filter) { - return map.entriesx(F.and(filter, F.cachePrimary())); + //return map.entriesx(F.and(filter, F.cachePrimary())); + return null; // TODO ignite-96 } /** {@inheritDoc} */ @@ -3715,9 +3716,24 @@ public CacheMetricsImpl metrics0() { if (keyCheck) validateCacheKey(key); - GridCacheEntryEx entry = peekEx(key); + try { + GridCacheEntryEx e = entry0(key, ctx.discovery().topologyVersion(), false, false); + + if (e == null) + return false; + + // Delegate to near if dht. + if (e.isDht() && CU.isNearEnabled(ctx)) { + GridCache near = ctx.isDht() ? ctx.dht().near() : ctx.near(); + + return near.isLockedByThread(key) || e.lockedByThread(); + } - return entry != null && entry.wrap(false).isLockedByThread(); + return e.lockedByThread(); + } + catch (GridCacheEntryRemovedException ignore) { + return false; + } } /** {@inheritDoc} */ @@ -4923,7 +4939,8 @@ public Set> entrySet(@Nullable Collection keys, */ public Set> primaryEntrySet( @Nullable IgnitePredicate>... filter) { - return map.entries(F.and(filter, F.cachePrimary())); +// return map.entries(F.and(filter, F.cachePrimary())); + return null; // TODO ignite-96 } /** @@ -4939,7 +4956,9 @@ public Set keySet(@Nullable IgnitePredicate>... filter) { * @return Primary key set. */ public Set primaryKeySet(@Nullable IgnitePredicate>... filter) { - return map.keySet(F.and(filter, F.cachePrimary())); +// return map.keySet(F.and(filter, F.cachePrimary())); + return null; + // TODO ignite-96 } /** @@ -4947,7 +4966,8 @@ public Set primaryKeySet(@Nullable IgnitePredicate>... filter) { * @return Primary values. */ public Collection primaryValues(@Nullable IgnitePredicate>... filter) { - return map.values(F.and(filter, F.cachePrimary())); +// return map.values(F.and(filter, F.cachePrimary())); + return null; } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java index 0b52d60a8bbf1..88d6e401df2ba 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java @@ -1237,7 +1237,7 @@ private IgnitePredicate>[] versionFilter(final EvictionInfo info) { // then we should not evict entry. return cctx.vararg(new P1>() { @Override public boolean apply(Entry e) { - GridCacheVersion ver = (GridCacheVersion)e.version(); + GridCacheVersion ver = (GridCacheVersion)((CacheEntryImpl)e).version(); return info.version().equals(ver) && F.isAll(info.filter()); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIterator.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIterator.java index 4d177fc50e404..fc936e68e3e54 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIterator.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIterator.java @@ -75,12 +75,14 @@ public GridCacheIterator(Iterable> c, @Override public void remove() { it.remove(); - try { +// try { // Back remove operation by actual cache. - cur.removex(); - } - catch (IgniteCheckedException e) { - throw new GridClosureException(e); - } + // cur.removex(); + // TODO ignite-96 + +// } +// catch (IgniteCheckedException e) { +// throw new GridClosureException(e); +// } } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java index fc7f9790e143e..68ca70f41e3b3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java @@ -85,12 +85,14 @@ public GridCacheKeySet(GridCacheContext ctx, Collection prj) { /** {@inheritDoc} */ @SuppressWarnings({"unchecked", "RedundantCast"}) - @Override public Set> entrySet() { - return (Set>)(Set>)prj.entrySet(); + @Override public Set> entrySet() { +// return (Set>)(Set>)prj.entrySet(); + return null; // TODO ignite-96 } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java index 7d033ce52404f..d83025d0969e3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java @@ -3695,7 +3695,11 @@ protected V saveValueForIndexUnlocked() throws IgniteCheckedException { @Override public Entry wrap(boolean prjAware) { try { - return new CacheEntryImpl<>(key, rawGetOrUnmarshal(true)); + CacheEntryImpl entry = new CacheEntryImpl<>(key, rawGetOrUnmarshal(true)); + + entry.version(ver); + + return entry; } catch (IgniteCheckedException e) { throw new RuntimeException("Fixme"); //TODO ignite-96 diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheValueCollection.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheValueCollection.java index 66dd627832dde..f3d7702784e9d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheValueCollection.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheValueCollection.java @@ -135,12 +135,13 @@ private void advance() { if (F.isAll(e, filter) && F.eq(o, e.getValue())) { it.remove(); - try { - e.removex(); - } - catch (IgniteCheckedException ex) { - throw new IgniteException(ex); - } + // TODO ignite-96 +// try { +// e.removex(); +// } +// catch (IgniteCheckedException ex) { +// throw new IgniteException(ex); +// } rmv = true; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java index 3c1a761bb8de8..b43cbb0e2a1d3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java @@ -868,7 +868,8 @@ private PartitionEntrySet(int partId) { Entry entry = (Entry)o; - return partId == entry.partition() && F.eq(entry.peek(), peek(entry.getKey())); + return partId == ctx.affinity().partition(entry.getKey()) && + F.eq(entry.getValue(), peek(entry.getKey())); } /** {@inheritDoc} */ @@ -958,7 +959,8 @@ private PartitionEntryIterator(@Nullable Iterator> partI if (last == null) throw new IllegalStateException(); - ctx.cache().remove(last.getKey(), last.getValue()); + // TODO ignite-96 + // ctx.cache().remove(last.getKey(), last.getValue()); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java index 542db3361e676..f8c5c24eb879f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java @@ -434,7 +434,9 @@ public Set nearKeySet(@Nullable IgnitePredicate> filter) { /** {@inheritDoc} */ @Override public Collection primaryValues(@Nullable IgnitePredicate>... filter) { - return new GridCacheValueCollection<>(ctx, entrySet(filter), ctx.vararg(F.cachePrimary())); + return null; + // return new GridCacheValueCollection<>(ctx, entrySet(filter), ctx.vararg(F.cachePrimary())); + // TODO ignite-96 } /** {@inheritDoc} */ @@ -471,13 +473,6 @@ public boolean evictNearOnly(K key, @Nullable IgnitePredicate>[] fil return super.compact(key, filter) | dht().compact(key, filter); } - /** {@inheritDoc} */ - @Override public Entry entry(K key) { - // We don't try wrap entry from near or dht cache. - // Created object will be wrapped once some method is called. - return new GridPartitionedCacheEntryImpl<>(ctx.projectionPerCall(), ctx, key, null); - } - /** * Peeks only near cache without looking into DHT cache. * diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java index d65f7744fb8c8..f0219d7e3a881 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java @@ -32,6 +32,7 @@ import org.jdk8.backport.*; import org.jetbrains.annotations.*; +import javax.cache.Cache.*; import java.util.*; import java.util.concurrent.*; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java index e0e7bee526bff..797bff86afc7d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java @@ -192,7 +192,9 @@ public GridCacheContinuousQueryHandler() { @Override public void onEntryUpdate(GridCacheContinuousQueryEntry e, boolean recordEvt) { GridCacheContext cctx = cacheContext(ctx); - if (cctx.isReplicated() && !skipPrimaryCheck && !e.primary()) + if (cctx.isReplicated() && + !skipPrimaryCheck && + !cctx.affinity().primary(cctx.localNode(), e.getKey(), cctx.topology().topologyVersion())) return; boolean notify; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java index 5bee688e35928..609ac70cd69e9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java @@ -394,12 +394,7 @@ public class GridFunc { private static final IgnitePredicate CACHE_ENTRY_NO_GET_VAL = new IgnitePredicate() { @SuppressWarnings({"unchecked"}) @Override public boolean apply(Object o) { - try { - return ((Entry)o).get() == null; - } - catch (IgniteCheckedException e) { - throw new GridClosureException(e); - } + return ((Entry)o).getValue() == null; } @Override public String toString() { @@ -411,7 +406,7 @@ public class GridFunc { private static final IgnitePredicate CACHE_ENTRY_HAS_PEEK_VAL = new IgnitePredicate() { @SuppressWarnings({"unchecked"}) @Override public boolean apply(Object o) { - return ((Entry)o).peek() != null; + return ((Entry)o).getValue() != null; } @Override public String toString() { @@ -423,7 +418,7 @@ public class GridFunc { private static final IgnitePredicate CACHE_ENTRY_NO_PEEK_VAL = new IgnitePredicate() { @SuppressWarnings({"unchecked"}) @Override public boolean apply(Object o) { - return ((Entry)o).peek() == null; + return ((Entry)o).getValue() == null; } @Override public String toString() { @@ -431,30 +426,6 @@ public class GridFunc { } }; - /** */ - private static final IgnitePredicate CACHE_ENTRY_PRIMARY = new IgnitePredicate() { - @SuppressWarnings({"unchecked"}) - @Override public boolean apply(Object o) { - return ((Entry)o).primary(); - } - - @Override public String toString() { - return "Cache primary entry predicate."; - } - }; - - /** */ - private static final IgnitePredicate CACHE_ENTRY_BACKUP = new IgnitePredicate() { - @SuppressWarnings({"unchecked"}) - @Override public boolean apply(Object o) { - return ((Entry)o).backup(); - } - - @Override public String toString() { - return "Cache backup entry predicate."; - } - }; - /** */ private static final IgniteClosure NODE2ID = new IgniteClosure() { @Override public UUID apply(ClusterNode n) { @@ -7210,7 +7181,7 @@ public static boolean contains(Integer[] arr, Integer val) { * @return Predicate that accepts {@link Entry} value and compares its value * to the given value. */ - public static IgnitePredicate> mapValue(@Nullable final V val) { + public static IgnitePredicate> mapValue(@Nullable final V val) { return new P1>() { @Override public boolean apply(Map.Entry e) { return e.getValue().equals(val); @@ -7228,7 +7199,7 @@ public static IgnitePredicate> mapValue(@Nullable final V val * @return Predicate that accepts {@code Map.Entry} value and compares its key * to the given value. */ - public static IgnitePredicate> mapKey(@Nullable final K key) { + public static IgnitePredicate> mapKey(@Nullable final K key) { return new P1>() { @Override public boolean apply(Map.Entry e) { return e.getKey().equals(key); @@ -7835,34 +7806,6 @@ public static IgnitePredicate> cacheNoPeekValue() { return (IgnitePredicate>)CACHE_ENTRY_NO_PEEK_VAL; } - /** - * Gets predicate which returns {@code true} if {@link org.apache.ignite.cache.Entry#primary()} - * method returns {@code true}. - * - * @param Cache key type. - * @param Cache value type. - * @return Predicate which returns {@code true} if {@link org.apache.ignite.cache.Entry#primary()} - * method returns {@code true}. - */ - @SuppressWarnings({"unchecked"}) - public static IgnitePredicate> cachePrimary() { - return (IgnitePredicate>)CACHE_ENTRY_PRIMARY; - } - - /** - * Gets predicate which returns {@code true} if {@link org.apache.ignite.cache.Entry#primary()} - * method returns {@code false}. - * - * @param Cache key type. - * @param Cache value type. - * @return Predicate which returns {@code true} if {@link org.apache.ignite.cache.Entry#primary()} - * method returns {@code false}. - */ - @SuppressWarnings({"unchecked"}) - public static IgnitePredicate> cacheBackup() { - return (IgnitePredicate>)CACHE_ENTRY_BACKUP; - } - /** * Gets predicate which returns true if {@link org.apache.ignite.cache.Entry#get()} * method returns value that is contained in given collection. Note that if collection @@ -7925,7 +7868,7 @@ public static IgnitePredicate> cacheContainsPeek( return isEmpty(vals) ? F.>alwaysFalse() : new IgnitePredicate>() { @Override public boolean apply(Entry e) { - V v = e.peek(); + V v = e.getValue(); assert vals != null; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java index 928322236c13f..06b19e6f5b491 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java @@ -37,6 +37,7 @@ import org.jdk8.backport.*; import org.jetbrains.annotations.*; +import javax.cache.Cache.*; import javax.cache.configuration.*; import java.util.*; import java.util.concurrent.atomic.*; @@ -513,7 +514,7 @@ protected boolean containsOffheapValue(GridCache cache, Object val) throws Excep protected static final IgnitePredicate> lt50 = new P1>() { @Override public boolean apply(Entry entry) { - Integer i = entry.peek(); + Integer i = entry.getValue(); return i != null && i < 50; } @@ -525,7 +526,7 @@ protected boolean containsOffheapValue(GridCache cache, Object val) throws Excep protected static final IgnitePredicate> gte100 = new P1>() { @Override public boolean apply(Entry entry) { - Integer i = entry.peek(); + Integer i = entry.getValue(); return i != null && i >= 100; } @@ -541,7 +542,7 @@ protected boolean containsOffheapValue(GridCache cache, Object val) throws Excep protected static final IgnitePredicate> gte200 = new P1>() { @Override public boolean apply(Entry entry) { - Integer i = entry.peek(); + Integer i = entry.getValue(); return i != null && i >= 200; }