You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by nt...@apache.org on 2016/02/17 09:36:24 UTC
[2/4] ignite git commit: ignite-2587 Fixed continuous query
notifications in offheap mode and BinaryObjectOffheapImpl usage.
ignite-2587 Fixed continuous query notifications in offheap mode and BinaryObjectOffheapImpl usage.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6291c64e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6291c64e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6291c64e
Branch: refs/heads/ignite-1.5.7
Commit: 6291c64efab4a2e3d5d9986479bc7a4372fdb553
Parents: 4de124c
Author: sboikov <sb...@gridgain.com>
Authored: Wed Feb 10 14:07:40 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Wed Feb 17 11:09:18 2016 +0300
----------------------------------------------------------------------
.../processors/cache/CacheLazyEntry.java | 3 +
.../processors/cache/GridCacheContext.java | 4 +-
.../processors/cache/GridCacheMapEntry.java | 118 ++++-
.../binary/CacheObjectBinaryProcessorImpl.java | 6 +-
.../dht/atomic/GridDhtAtomicCache.java | 79 +++-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 85 +++-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 38 +-
.../cache/query/GridCacheQueryManager.java | 30 +-
.../continuous/CacheContinuousQueryHandler.java | 99 ++--
.../CacheContinuousQueryListener.java | 2 +-
.../continuous/CacheContinuousQueryManager.java | 143 +++---
.../continuous/GridContinuousProcessor.java | 60 +--
.../IgniteCacheEntryListenerAbstractTest.java | 454 +++++++++++++------
...cheEntryListenerAtomicOffheapTieredTest.java | 32 ++
...cheEntryListenerAtomicOffheapValuesTest.java | 32 ++
...teCacheEntryListenerTxOffheapTieredTest.java | 32 ++
...teCacheEntryListenerTxOffheapValuesTest.java | 32 ++
.../cache/IgniteCacheEntryListenerTxTest.java | 1 +
...ContinuousQueryFailoverAbstractSelfTest.java | 10 +
...tomicPrimaryWriteOrderOffheapTieredTest.java | 33 ++
...tinuousQueryFailoverTxOffheapTieredTest.java | 32 ++
...ridCacheContinuousQueryAbstractSelfTest.java | 20 +-
...eContinuousQueryAtomicOffheapTieredTest.java | 32 ++
...eContinuousQueryAtomicOffheapValuesTest.java | 32 ++
...CacheContinuousQueryTxOffheapTieredTest.java | 32 ++
...CacheContinuousQueryTxOffheapValuesTest.java | 32 ++
.../junits/common/GridCommonAbstractTest.java | 2 +-
.../ignite/testsuites/IgniteCacheTestSuite.java | 8 +
.../IgniteCacheQuerySelfTestSuite.java | 28 +-
29 files changed, 1140 insertions(+), 371 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLazyEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLazyEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLazyEntry.java
index 05a6fef..30933e5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLazyEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLazyEntry.java
@@ -50,6 +50,7 @@ public class CacheLazyEntry<K, V> implements Cache.Entry<K, V> {
* @param cctx Cache context.
* @param keyObj Key cache object.
* @param valObj Cache object value.
+ * @param keepBinary Keep binary flag.
*/
public CacheLazyEntry(GridCacheContext cctx, KeyCacheObject keyObj, CacheObject valObj, boolean keepBinary) {
this.cctx = cctx;
@@ -61,6 +62,7 @@ public class CacheLazyEntry<K, V> implements Cache.Entry<K, V> {
/**
* @param keyObj Key cache object.
* @param val Value.
+ * @param keepBinary Keep binary flag.
* @param cctx Cache context.
*/
public CacheLazyEntry(GridCacheContext cctx, KeyCacheObject keyObj, V val, boolean keepBinary) {
@@ -75,6 +77,7 @@ public class CacheLazyEntry<K, V> implements Cache.Entry<K, V> {
* @param keyObj Key cache object.
* @param key Key value.
* @param valObj Cache object
+ * @param keepBinary Keep binary flag.
* @param val Cache value.
*/
public CacheLazyEntry(GridCacheContext<K, V> ctx,
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index e875df0..5729959 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -1729,10 +1729,10 @@ public class GridCacheContext<K, V> implements Externalizable {
* @return Heap-based object.
*/
@Nullable public <T> T unwrapTemporary(@Nullable Object obj) {
- if (!offheapTiered())
+ if (!useOffheapEntry())
return (T)obj;
- return (T) cacheObjects().unwrapTemporary(this, obj);
+ return (T)cacheObjects().unwrapTemporary(this, obj);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
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 64cfd01..5900963 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
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Collections;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import javax.cache.Cache;
@@ -44,6 +45,7 @@ import org.apache.ignite.internal.processors.cache.extras.GridCacheMvccEntryExtr
import org.apache.ignite.internal.processors.cache.extras.GridCacheObsoleteEntryExtras;
import org.apache.ignite.internal.processors.cache.extras.GridCacheTtlEntryExtras;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
@@ -1122,7 +1124,13 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert newVer != null : "Failed to get write version for tx: " + tx;
- old = (retval || intercept) ? rawGetOrUnmarshalUnlocked(!retval) : this.val;
+ boolean internal = isInternal() || !context().userCache();
+
+ Map<UUID, CacheContinuousQueryListener> lsnrCol =
+ notifyContinuousQueries(tx) ? cctx.continuousQueries().updateListeners(internal, false) : null;
+
+ old = (retval || intercept || lsnrCol != null) ?
+ rawGetOrUnmarshalUnlocked(!retval && !isOffHeapValuesOnly()) : this.val;
if (intercept) {
val0 = CU.value(val, cctx, false);
@@ -1206,10 +1214,19 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
keepBinary);
}
- if (cctx.isLocal() || cctx.isReplicated() ||
- (!isNear() && !(tx != null && tx.onePhaseCommit() && !tx.local())))
- cctx.continuousQueries().onEntryUpdated(key, val, old, isInternal() || !context().userCache(),
- partition(), tx.local(), false, updateCntr0, topVer);
+ if (lsnrCol != null) {
+ cctx.continuousQueries().onEntryUpdated(
+ lsnrCol,
+ key,
+ val,
+ old,
+ internal,
+ partition(),
+ tx.local(),
+ false,
+ updateCntr0,
+ topVer);
+ }
cctx.dataStructures().onEntryUpdated(key, false, keepBinary);
}
@@ -1304,7 +1321,13 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
newVer = explicitVer != null ? explicitVer : tx == null ? nextVersion() : tx.writeVersion();
- old = (retval || intercept) ? rawGetOrUnmarshalUnlocked(!retval) : val;
+ boolean internal = isInternal() || !context().userCache();
+
+ Map<UUID, CacheContinuousQueryListener> lsnrCol =
+ notifyContinuousQueries(tx) ? cctx.continuousQueries().updateListeners(internal, false) : null;
+
+ old = (retval || intercept || lsnrCol != null) ?
+ rawGetOrUnmarshalUnlocked(!retval && !isOffHeapValuesOnly()) : val;
if (intercept) {
entry0 = new CacheLazyEntry(cctx, key, old, keepBinary);
@@ -1388,10 +1411,19 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
keepBinary);
}
- if (cctx.isLocal() || cctx.isReplicated() ||
- (!isNear() && !(tx != null && tx.onePhaseCommit() && !tx.local())))
- cctx.continuousQueries().onEntryUpdated(key, null, old, isInternal()
- || !context().userCache(),partition(), tx.local(), false, updateCntr0, topVer);
+ if (lsnrCol != null) {
+ cctx.continuousQueries().onEntryUpdated(
+ lsnrCol,
+ key,
+ null,
+ old,
+ internal,
+ partition(),
+ tx.local(),
+ false,
+ updateCntr0,
+ topVer);
+ }
cctx.dataStructures().onEntryUpdated(key, true, keepBinary);
@@ -1440,6 +1472,16 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
return new GridCacheUpdateTxResult(false, null);
}
+ /**
+ * @param tx Transaction.
+ * @return {@code True} if should notify continuous query manager.
+ */
+ private boolean notifyContinuousQueries(@Nullable IgniteInternalTx tx) {
+ return cctx.isLocal() ||
+ cctx.isReplicated() ||
+ (!isNear() && !(tx != null && tx.onePhaseCommit() && !tx.local()));
+ }
+
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public GridTuple3<Boolean, Object, EntryProcessorResult<Object>> innerUpdateLocal(
@@ -1470,7 +1512,16 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
EntryProcessorResult<Object> invokeRes = null;
synchronized (this) {
- boolean needVal = retval || intercept || op == GridCacheOperation.TRANSFORM || !F.isEmpty(filter);
+ boolean internal = isInternal() || !context().userCache();
+
+ Map<UUID, CacheContinuousQueryListener> lsnrCol =
+ cctx.continuousQueries().updateListeners(internal, false);
+
+ boolean needVal = retval ||
+ intercept ||
+ op == GridCacheOperation.TRANSFORM ||
+ !F.isEmpty(filter) ||
+ lsnrCol != null;
checkObsolete();
@@ -1479,7 +1530,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
unswap(retval);
// Possibly get old value form store.
- old = needVal ? rawGetOrUnmarshalUnlocked(!retval) : val;
+ old = needVal ? rawGetOrUnmarshalUnlocked(!retval && !isOffHeapValuesOnly()) : val;
boolean readFromStore = false;
@@ -1731,11 +1782,20 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (res)
updateMetrics(op, metrics);
- if (!isNear()) {
+ if (lsnrCol != null) {
long updateCntr = nextPartCounter(AffinityTopologyVersion.NONE);
- cctx.continuousQueries().onEntryUpdated(key, val, old, isInternal() || !context().userCache(),
- partition(), true, false, updateCntr, AffinityTopologyVersion.NONE);
+ cctx.continuousQueries().onEntryUpdated(
+ lsnrCol,
+ key,
+ val,
+ old,
+ internal,
+ partition(),
+ true,
+ false,
+ updateCntr,
+ AffinityTopologyVersion.NONE);
}
cctx.dataStructures().onEntryUpdated(key, op == GridCacheOperation.DELETE, keepBinary);
@@ -1997,8 +2057,16 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (updateCntr != null)
updateCntr0 = updateCntr;
- cctx.continuousQueries().onEntryUpdated(key, evtVal, prevVal, isInternal()
- || !context().userCache(), partition(), primary, false, updateCntr0, topVer);
+ cctx.continuousQueries().onEntryUpdated(
+ key,
+ evtVal,
+ prevVal,
+ isInternal() || !context().userCache(),
+ partition(),
+ primary,
+ false,
+ updateCntr0,
+ topVer);
}
return new GridCacheUpdateAtomicResult(false,
@@ -2019,7 +2087,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
// Prepare old value and value bytes.
- oldVal = needVal ? rawGetOrUnmarshalUnlocked(!retval) : val;
+ oldVal = needVal ? rawGetOrUnmarshalUnlocked(!retval && !isOffHeapValuesOnly()) : val;
// Possibly read value from store.
boolean readFromStore = false;
@@ -2937,7 +3005,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/**
* @return {@code True} if values should be stored off-heap.
*/
- protected boolean isOffHeapValuesOnly() {
+ protected final boolean isOffHeapValuesOnly() {
return cctx.config().getMemoryMode() == CacheMemoryMode.OFFHEAP_VALUES;
}
@@ -3236,8 +3304,16 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
drReplicate(drType, val, ver);
if (!skipQryNtf) {
- cctx.continuousQueries().onEntryUpdated(key, val, null, this.isInternal()
- || !this.context().userCache(), this.partition(), true, preload, updateCntr, topVer);
+ cctx.continuousQueries().onEntryUpdated(
+ key,
+ val,
+ null,
+ this.isInternal() || !this.context().userCache(),
+ this.partition(),
+ true,
+ preload,
+ updateCntr,
+ topVer);
cctx.dataStructures().onEntryUpdated(key, false, true);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index ac18eb5..2769e13 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -826,10 +826,10 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
Object val = unmarshal(valPtr, !tmp);
- if (val instanceof BinaryObjectOffheapImpl)
- return (BinaryObjectOffheapImpl)val;
+ if (val instanceof CacheObject)
+ return (CacheObject)val;
- return new CacheObjectImpl(val, null);
+ return toCacheObject(ctx.cacheObjectContext(), val, false);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/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 d413c18..0fac19c 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
@@ -76,6 +76,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSing
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
@@ -1995,6 +1996,10 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean intercept = ctx.config().getInterceptor() != null;
+ boolean initLsnrs = false;
+ Map<UUID, CacheContinuousQueryListener> lsnrs = null;
+ boolean internal = false;
+
// Avoid iterator creation.
for (int i = 0; i < keys.size(); i++) {
KeyCacheObject k = keys.get(i);
@@ -2009,6 +2014,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (entry == null)
continue;
+ if (!initLsnrs) {
+ internal = entry.isInternal() || !context().userCache();
+
+ lsnrs = ctx.continuousQueries().updateListeners(internal, false);
+
+ initLsnrs = true;
+ }
+
GridCacheVersion newConflictVer = req.conflictVersion(i);
long newConflictTtl = req.conflictTtl(i);
long newConflictExpireTime = req.conflictExpireTime(i);
@@ -2037,7 +2050,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.invokeArguments(),
primary && writeThrough() && !req.skipStore(),
!req.skipStore(),
- sndPrevVal || req.returnValue(),
+ lsnrs != null || sndPrevVal || req.returnValue(),
req.keepBinary(),
expiry,
true,
@@ -2064,6 +2077,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
if (dhtFut != null) {
+ dhtFut.listeners(lsnrs);
+
if (updRes.sendToDht()) { // Send to backups even in case of remove-remove scenarios.
GridCacheVersionConflictContext<?, ?> conflictCtx = updRes.conflictResolveResult();
@@ -2100,10 +2115,18 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
"[entry=" + entry + ", filter=" + Arrays.toString(req.filter()) + ']');
}
}
- else if (!entry.isNear() && updRes.success()) {
- ctx.continuousQueries().onEntryUpdated(entry.key(), updRes.newValue(), updRes.oldValue(),
- entry.isInternal() || !context().userCache(), entry.partition(), primary, false,
- updRes.updateCounter(), topVer);
+ else if (lsnrs != null && updRes.success()) {
+ ctx.continuousQueries().onEntryUpdated(
+ lsnrs,
+ entry.key(),
+ updRes.newValue(),
+ updRes.oldValue(),
+ internal,
+ entry.partition(),
+ primary,
+ false,
+ updRes.updateCounter(),
+ topVer);
}
if (hasNear) {
@@ -2278,6 +2301,9 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean intercept = ctx.config().getInterceptor() != null;
+ boolean initLsnrs = false;
+ Map<UUID, CacheContinuousQueryListener> lsnrs = null;
+
// Avoid iterator creation.
for (int i = 0; i < entries.size(); i++) {
GridDhtCacheEntry entry = entries.get(i);
@@ -2311,6 +2337,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
filteredReaders = F.view(entry.readers(), F.notEqualTo(node.id()));
}
+ if (!initLsnrs) {
+ lsnrs = ctx.continuousQueries().updateListeners(
+ entry.isInternal() || !context().userCache(),
+ false);
+
+ initLsnrs = true;
+ }
+
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
ver,
node.id(),
@@ -2320,7 +2354,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
/*write-through*/false,
/*read-through*/false,
- /*retval*/sndPrevVal,
+ /*retval*/sndPrevVal || lsnrs != null,
req.keepBinary(),
expiry,
/*event*/true,
@@ -2369,6 +2403,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
if (dhtFut != null) {
+ dhtFut.listeners(lsnrs);
+
EntryProcessor<Object, Object, Object> entryProcessor =
entryProcessorMap == null ? null : entryProcessorMap.get(entry.key());
@@ -2766,6 +2802,10 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
+ boolean initLsnrs = false;
+ Map<UUID, CacheContinuousQueryListener> lsnrs = null;
+ boolean internal = false;
+
for (int i = 0; i < req.size(); i++) {
KeyCacheObject key = req.key(i);
@@ -2788,6 +2828,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
long ttl = req.ttl(i);
long expireTime = req.conflictExpireTime(i);
+ if (!initLsnrs) {
+ internal = entry.isInternal() || !context().userCache();
+
+ lsnrs = ctx.continuousQueries().updateListeners(internal, false);
+
+ initLsnrs = true;
+ }
+
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
ver,
nodeId,
@@ -2797,7 +2845,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op == TRANSFORM ? req.invokeArguments() : null,
/*write-through*/false,
/*read-through*/false,
- /*retval*/false,
+ /*retval*/lsnrs != null,
req.keepBinary(),
/*expiry policy*/null,
/*event*/true,
@@ -2820,10 +2868,19 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (updRes.removeVersion() != null)
ctx.onDeferredDelete(entry, updRes.removeVersion());
- if (updRes.success() && !entry.isNear())
- ctx.continuousQueries().onEntryUpdated(entry.key(), updRes.newValue(),
- updRes.oldValue(), entry.isInternal() || !context().userCache(), entry.partition(),
- false, false, updRes.updateCounter(), req.topologyVersion());
+ if (lsnrs != null && updRes.success()) {
+ ctx.continuousQueries().onEntryUpdated(
+ lsnrs,
+ entry.key(),
+ updRes.newValue(),
+ updRes.oldValue(),
+ internal,
+ entry.partition(),
+ false,
+ false,
+ updRes.updateCounter(),
+ req.topologyVersion());
+ }
entry.onUnlock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index e31af19..07f0e14 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
@@ -102,6 +103,9 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
/** Response count. */
private volatile int resCnt;
+ /** */
+ private Map<UUID, CacheContinuousQueryListener> lsnrs;
+
/**
* @param cctx Cache context.
* @param completionCb Callback to invoke when future is completed.
@@ -136,6 +140,13 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
waitForExchange = !topLocked;
}
+ /**
+ * @param lsnrs Continuous query listeners.
+ */
+ void listeners(@Nullable Map<UUID, CacheContinuousQueryListener> lsnrs) {
+ this.lsnrs = lsnrs;
+ }
+
/** {@inheritDoc} */
@Override public IgniteUuid futureId() {
return futVer.asGridUuid();
@@ -215,6 +226,8 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
* @param ttl TTL (optional).
* @param conflictExpireTime Conflict expire time (optional).
* @param conflictVer Conflict version (optional).
+ * @param addPrevVal If {@code true} sends previous value to backups.
+ * @param prevVal Previous value.
* @param updateCntr Partition update counter.
*/
public void addWriteEntry(GridDhtCacheEntry entry,
@@ -270,13 +283,22 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
addPrevVal,
entry.partition(),
prevVal,
- updateCntr);
+ updateCntr,
+ lsnrs != null);
}
- else if (dhtNodes.size() == 1) {
+ else if (lsnrs != null && dhtNodes.size() == 1) {
try {
- cctx.continuousQueries().onEntryUpdated(entry.key(), val, prevVal,
- entry.key().internal() || !cctx.userCache(), entry.partition(), true, false,
- updateCntr, updateReq.topologyVersion());
+ cctx.continuousQueries().onEntryUpdated(
+ lsnrs,
+ entry.key(),
+ val,
+ prevVal,
+ entry.key().internal() || !cctx.userCache(),
+ entry.partition(),
+ true,
+ false,
+ updateCntr,
+ updateReq.topologyVersion());
}
catch (IgniteCheckedException e) {
U.warn(log, "Failed to send continuous query message. [key=" + entry.key() + ", newVal="
@@ -352,7 +374,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
cctx.mvcc().removeAtomicFuture(version());
if (err != null) {
- if (!mappings.isEmpty()) {
+ if (!mappings.isEmpty() && lsnrs != null) {
Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
@@ -362,7 +384,11 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
if (!hndKeys.contains(key)) {
updateRes.addFailedKey(key, err);
- cctx.continuousQueries().skipUpdateEvent(key, req.partitionId(i), req.updateCounter(i),
+ cctx.continuousQueries().skipUpdateEvent(
+ lsnrs,
+ key,
+ req.partitionId(i),
+ req.updateCounter(i),
updateReq.topologyVersion());
hndKeys.add(key);
@@ -378,27 +404,38 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
updateRes.addFailedKey(key, err);
}
else {
- Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
+ if (lsnrs != null) {
+ Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
- exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
- for (int i = 0; i < req.size(); i++) {
- KeyCacheObject key = req.key(i);
+ exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
+ for (int i = 0; i < req.size(); i++) {
+ KeyCacheObject key = req.key(i);
- if (!hndKeys.contains(key)) {
- try {
- cctx.continuousQueries().onEntryUpdated(key, req.value(i), req.localPreviousValue(i),
- key.internal() || !cctx.userCache(), req.partitionId(i), true, false,
- req.updateCounter(i), updateReq.topologyVersion());
- }
- catch (IgniteCheckedException e) {
- U.warn(log, "Failed to send continuous query message. [key=" + key + ", newVal="
- + req.value(i) + ", err=" + e + "]");
- }
+ if (!hndKeys.contains(key)) {
+ try {
+ cctx.continuousQueries().onEntryUpdated(
+ lsnrs,
+ key,
+ req.value(i),
+ req.localPreviousValue(i),
+ key.internal() || !cctx.userCache(),
+ req.partitionId(i),
+ true,
+ false,
+ req.updateCounter(i),
+ updateReq.topologyVersion());
+ }
+ catch (IgniteCheckedException e) {
+ U.warn(log, "Failed to send continuous query message. [key=" + key +
+ ", newVal=" + req.value(i) +
+ ", err=" + e + "]");
+ }
- hndKeys.add(key);
+ hndKeys.add(key);
- if (hndKeys.size() == keys.size())
- break exit;
+ if (hndKeys.size() == keys.size())
+ break exit;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 7bee5a3..1b26fee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.IgniteCodeGeneratingFail;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
@@ -49,6 +50,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Lite dht cache backup update request.
*/
+@IgniteCodeGeneratingFail // Need add 'cleanup' call in 'writeTo' method.
public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements GridCacheDeployable {
/** */
private static final long serialVersionUID = 0L;
@@ -215,7 +217,6 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
keys = new ArrayList<>();
partIds = new ArrayList<>();
- locPrevVals = new ArrayList<>();
if (forceTransformBackups) {
entryProcessors = new ArrayList<>();
@@ -240,7 +241,10 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
* @param conflictExpireTime Conflict expire time (optional).
* @param conflictVer Conflict version (optional).
* @param addPrevVal If {@code true} adds previous value.
+ * @param partId Partition.
* @param prevVal Previous value.
+ * @param updateCntr Update counter.
+ * @param storeLocPrevVal If {@code true} stores previous value.
*/
public void addWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
@@ -251,12 +255,18 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
boolean addPrevVal,
int partId,
@Nullable CacheObject prevVal,
- @Nullable Long updateIdx) {
+ @Nullable Long updateCntr,
+ boolean storeLocPrevVal) {
keys.add(key);
partIds.add(partId);
- locPrevVals.add(prevVal);
+ if (storeLocPrevVal) {
+ if (locPrevVals == null)
+ locPrevVals = new ArrayList<>();
+
+ locPrevVals.add(prevVal);
+ }
if (forceTransformBackups) {
assert entryProcessor != null;
@@ -273,11 +283,11 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
prevVals.add(prevVal);
}
- if (updateIdx != null) {
+ if (updateCntr != null) {
if (updateCntrs == null)
updateCntrs = new GridLongList();
- updateCntrs.add(updateIdx);
+ updateCntrs.add(updateCntr);
}
// In case there is no conflict, do not create the list.
@@ -521,6 +531,8 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
* @return Value.
*/
@Nullable public CacheObject localPreviousValue(int idx) {
+ assert locPrevVals != null;
+
return locPrevVals.get(idx);
}
@@ -844,6 +856,8 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
}
+ cleanup();
+
return true;
}
@@ -1043,6 +1057,20 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
return reader.afterMessageRead(GridDhtAtomicUpdateRequest.class);
}
+ /**
+ * Cleanup values not needed after message was sent.
+ */
+ private void cleanup() {
+ nearVals = null;
+ prevVals = null;
+
+ // Do not keep values if they are not needed for continuous query notification.
+ if (locPrevVals == null) {
+ vals = null;
+ locPrevVals = null;
+ }
+ }
+
/** {@inheritDoc} */
@Override public byte directType() {
return 38;
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
index 8f0cab7..0d8f795 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
@@ -1107,7 +1107,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
next = null;
while (it.hasNext()) {
- final LazySwapEntry e = new LazySwapEntry(it.next(), keepBinary);
+ final LazySwapEntry e = new LazySwapEntry(it.next());
if (filter != null) {
K key = (K)cctx.unwrapBinaryIfNeeded(e.key(), keepBinary);
@@ -2524,15 +2524,11 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
/** */
private final Map.Entry<byte[], byte[]> e;
- /** */
- private boolean keepBinary;
-
/**
* @param e Entry with
*/
- LazySwapEntry(Map.Entry<byte[], byte[]> e, boolean keepBinary) {
+ LazySwapEntry(Map.Entry<byte[], byte[]> e) {
this.e = e;
- this.keepBinary = keepBinary;
}
/** {@inheritDoc} */
@@ -2545,9 +2541,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
@Override protected V unmarshalValue() throws IgniteCheckedException {
IgniteBiTuple<byte[], Byte> t = GridCacheSwapEntryImpl.getValue(e.getValue());
- CacheObject obj = cctx.cacheObjects().toCacheObject(cctx.cacheObjectContext(), t.get2(), t.get1());
-
- return (V)cctx.cacheObjectContext().unwrapBinaryIfNeeded(obj, keepBinary);
+ return (V)cctx.cacheObjects().toCacheObject(cctx.cacheObjectContext(), t.get2(), t.get1());
}
/** {@inheritDoc} */
@@ -2597,13 +2591,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
@Override protected V unmarshalValue() throws IgniteCheckedException {
long ptr = GridCacheOffheapSwapEntry.valueAddress(valPtr.get1(), valPtr.get2());
- CacheObject obj = cctx.fromOffheap(ptr, false);
-
- V val = CU.value(obj, cctx, false);
-
- assert val != null;
-
- return val;
+ return (V)cctx.fromOffheap(ptr, false);
}
/** {@inheritDoc} */
@@ -2661,7 +2649,15 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
if (!filter.apply(key, val))
return null;
- return new IgniteBiTuple<>(e.key(), (V)cctx.unwrapTemporary(e.value()));
+ if (key instanceof CacheObject)
+ ((CacheObject)key).prepareMarshal(cctx.cacheObjectContext());
+
+ val = (V)cctx.unwrapTemporary(e.value());
+
+ if (val instanceof CacheObject)
+ ((CacheObject)val).prepareMarshal(cctx.cacheObjectContext());
+
+ return new IgniteBiTuple<>(e.key(), val);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 08fe62a..cf9b439 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -325,9 +325,9 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
// skipPrimaryCheck is set only when listen locally for replicated cache events.
assert !skipPrimaryCheck || (cctx.isReplicated() && ctx.localNodeId().equals(nodeId));
- boolean notify = !evt.entry().isFiltered();
+ boolean notify = true;
- if (notify && rmtFilter != null) {
+ if (rmtFilter != null) {
try {
notify = rmtFilter.evaluate(evt);
}
@@ -472,15 +472,38 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
sendBackupAcknowledge(ackBuf.acknowledgeOnTimeout(), routineId, ctx);
}
- @Override public void skipUpdateEvent(CacheContinuousQueryEvent<K, V> evt, AffinityTopologyVersion topVer,
- boolean primary) {
- assert evt != null;
+ @Override public void skipUpdateEvent(CacheContinuousQueryEvent<K, V> evt, AffinityTopologyVersion topVer) {
+ try {
+ assert evt != null;
- CacheContinuousQueryEntry e = evt.entry();
+ CacheContinuousQueryEntry e = evt.entry();
- e.markFiltered();
+ EntryBuffer buf = entryBufs.get(e.partition());
+
+ if (buf == null) {
+ buf = new EntryBuffer();
+
+ EntryBuffer oldRec = entryBufs.putIfAbsent(e.partition(), buf);
+
+ if (oldRec != null)
+ buf = oldRec;
+ }
+
+ e = buf.skipEntry(e);
+
+ if (e != null && !ctx.localNodeId().equals(nodeId))
+ ctx.continuous().addNotification(nodeId, routineId, e, topic, sync, true);
+ }
+ catch (ClusterTopologyCheckedException ex) {
+ IgniteLogger log = ctx.log(getClass());
- onEntryUpdated(evt, primary, false);
+ if (log.isDebugEnabled())
+ log.debug("Failed to send event notification to node, node left cluster " +
+ "[node=" + nodeId + ", err=" + ex + ']');
+ }
+ catch (IgniteCheckedException ex) {
+ U.error(ctx.log(getClass()), "Failed to send event notification to node: " + nodeId, ex);
+ }
}
@Override public void onPartitionEvicted(int part) {
@@ -595,22 +618,20 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
for (CacheContinuousQueryEntry e : entries)
entries0.addAll(handleEvent(ctx, e));
- if (!entries0.isEmpty()) {
- Iterable<CacheEntryEvent<? extends K, ? extends V>> evts = F.viewReadOnly(entries0,
- new C1<CacheContinuousQueryEntry, CacheEntryEvent<? extends K, ? extends V>>() {
- @Override public CacheEntryEvent<? extends K, ? extends V> apply(CacheContinuousQueryEntry e) {
- return new CacheContinuousQueryEvent<>(cache, cctx, e);
- }
- },
- new IgnitePredicate<CacheContinuousQueryEntry>() {
- @Override public boolean apply(CacheContinuousQueryEntry entry) {
- return !entry.isFiltered();
- }
+ Iterable<CacheEntryEvent<? extends K, ? extends V>> evts = F.viewReadOnly(entries0,
+ new C1<CacheContinuousQueryEntry, CacheEntryEvent<? extends K, ? extends V>>() {
+ @Override public CacheEntryEvent<? extends K, ? extends V> apply(CacheContinuousQueryEntry e) {
+ return new CacheContinuousQueryEvent<>(cache, cctx, e);
}
- );
+ },
+ new IgnitePredicate<CacheContinuousQueryEntry>() {
+ @Override public boolean apply(CacheContinuousQueryEntry entry) {
+ return !entry.isFiltered();
+ }
+ }
+ );
- locLsnr.onUpdated(evts);
- }
+ locLsnr.onUpdated(evts);
}
/**
@@ -710,11 +731,11 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
* @param initCntr Update counters.
*/
public PartitionRecovery(IgniteLogger log, AffinityTopologyVersion topVer, @Nullable Long initCntr) {
+ assert topVer.topologyVersion() > 0 : topVer;
+
this.log = log;
if (initCntr != null) {
- assert topVer.topologyVersion() > 0 : topVer;
-
this.lastFiredEvt = initCntr;
curTop = topVer;
@@ -857,6 +878,32 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
/**
+ * @param e Entry.
+ * @return Continuous query entry.
+ */
+ private CacheContinuousQueryEntry skipEntry(CacheContinuousQueryEntry e) {
+ if (lastFiredCntr.get() > e.updateCounter() || e.updateCounter() == 1L) {
+ e.markFiltered();
+
+ return e;
+ }
+ else {
+ buf.add(e.updateCounter());
+
+ // Double check. If another thread sent a event with counter higher than this event.
+ if (lastFiredCntr.get() > e.updateCounter() && buf.contains(e.updateCounter())) {
+ buf.remove(e.updateCounter());
+
+ e.markFiltered();
+
+ return e;
+ }
+ else
+ return null;
+ }
+ }
+
+ /**
* Add continuous entry.
*
* @param e Cache continuous query entry.
@@ -987,10 +1034,10 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
Collection<ClusterNode> nodes = new HashSet<>();
for (AffinityTopologyVersion topVer : t.get2())
- nodes.addAll(ctx.discovery().cacheAffinityNodes(cctx.name(), topVer));
+ nodes.addAll(ctx.discovery().cacheNodes(topVer));
for (ClusterNode node : nodes) {
- if (!node.isLocal()) {
+ if (!node.id().equals(ctx.localNodeId())) {
try {
cctx.io().send(node, msg, GridIoPolicy.SYSTEM_POOL);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
index 5246fac..83ff32c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
@@ -24,7 +24,7 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
/**
* Continuous query listener.
*/
-interface CacheContinuousQueryListener<K, V> {
+public interface CacheContinuousQueryListener<K, V> {
/**
* Query execution callback.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
index bb3c479..cc59989 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
@@ -60,6 +60,7 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.resources.LoggerResource;
+import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
import static javax.cache.event.EventType.CREATED;
@@ -115,19 +116,17 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
// Append cache name to the topic.
topicPrefix = "CONTINUOUS_QUERY" + (cctx.name() == null ? "" : "_" + cctx.name());
- if (cctx.affinityNode()) {
- cctx.io().addHandler(cctx.cacheId(), CacheContinuousQueryBatchAck.class,
- new CI2<UUID, CacheContinuousQueryBatchAck>() {
- @Override public void apply(UUID uuid, CacheContinuousQueryBatchAck msg) {
- CacheContinuousQueryListener lsnr = lsnrs.get(msg.routineId());
+ cctx.io().addHandler(cctx.cacheId(), CacheContinuousQueryBatchAck.class,
+ new CI2<UUID, CacheContinuousQueryBatchAck>() {
+ @Override public void apply(UUID uuid, CacheContinuousQueryBatchAck msg) {
+ CacheContinuousQueryListener lsnr = lsnrs.get(msg.routineId());
- if (lsnr != null)
- lsnr.cleanupBackupQueue(msg.updateCntrs());
- }
- });
+ if (lsnr != null)
+ lsnr.cleanupBackupQueue(msg.updateCntrs());
+ }
+ });
- cctx.time().schedule(new BackupCleaner(lsnrs, cctx.kernalContext()), BACKUP_ACK_FREQ, BACKUP_ACK_FREQ);
- }
+ cctx.time().schedule(new BackupCleaner(lsnrs, cctx.kernalContext()), BACKUP_ACK_FREQ, BACKUP_ACK_FREQ);
}
/** {@inheritDoc} */
@@ -157,55 +156,102 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
}
/**
+ * @param lsnrs Listeners to notify.
+ * @param key Entry key.
* @param partId Partition id.
* @param updCntr Updated counter.
* @param topVer Topology version.
*/
- public void skipUpdateEvent(KeyCacheObject key,
- int partId,
- long updCntr,
- AffinityTopologyVersion topVer) {
- skipUpdateEvent(key, partId, updCntr, true, topVer);
+ public void skipUpdateEvent(Map<UUID, CacheContinuousQueryListener> lsnrs,
+ KeyCacheObject key, int partId, long updCntr, AffinityTopologyVersion topVer) {
+ assert lsnrs != null;
+
+ for (CacheContinuousQueryListener lsnr : lsnrs.values()) {
+ CacheContinuousQueryEntry e0 = new CacheContinuousQueryEntry(
+ cctx.cacheId(),
+ UPDATED,
+ key,
+ null,
+ null,
+ lsnr.keepBinary(),
+ partId,
+ updCntr,
+ topVer);
+
+ CacheContinuousQueryEvent evt = new CacheContinuousQueryEvent<>(
+ cctx.kernalContext().cache().jcache(cctx.name()), cctx, e0);
+
+ lsnr.skipUpdateEvent(evt, topVer);
+ }
}
/**
- * @param key Entry key.
- * @param partId Partition id.
- * @param updCntr Updated counter.
+ * @param internal Internal entry flag (internal key or not user cache).
+ * @param preload Whether update happened during preloading.
+ * @return Registered listeners.
+ */
+ @Nullable public Map<UUID, CacheContinuousQueryListener> updateListeners(
+ boolean internal,
+ boolean preload) {
+ if (preload && !internal)
+ return null;
+
+ ConcurrentMap<UUID, CacheContinuousQueryListener> lsnrCol;
+
+ if (internal)
+ lsnrCol = intLsnrCnt.get() > 0 ? intLsnrs : null;
+ else
+ lsnrCol = lsnrCnt.get() > 0 ? lsnrs : null;
+
+ return F.isEmpty(lsnrCol) ? null : lsnrCol;
+ }
+
+ /**
+ * @param key Key.
+ * @param newVal New value.
+ * @param oldVal Old value.
+ * @param internal Internal entry (internal key or not user cache).
+ * @param partId Partition.
+ * @param primary {@code True} if called on primary node.
+ * @param preload Whether update happened during preloading.
+ * @param updateCntr Update counter.
* @param topVer Topology version.
- * @param primary Primary.
+ * @throws IgniteCheckedException In case of error.
*/
- public void skipUpdateEvent(KeyCacheObject key,
+ public void onEntryUpdated(
+ KeyCacheObject key,
+ CacheObject newVal,
+ CacheObject oldVal,
+ boolean internal,
int partId,
- long updCntr,
boolean primary,
- AffinityTopologyVersion topVer) {
- if (lsnrCnt.get() > 0) {
- for (CacheContinuousQueryListener lsnr : lsnrs.values()) {
- CacheContinuousQueryEntry e0 = new CacheContinuousQueryEntry(
- cctx.cacheId(),
- UPDATED,
- key,
- null,
- null,
- lsnr.keepBinary(),
- partId,
- updCntr,
- topVer);
-
- CacheContinuousQueryEvent evt = new CacheContinuousQueryEvent<>(
- cctx.kernalContext().cache().jcache(cctx.name()), cctx, e0);
+ boolean preload,
+ long updateCntr,
+ AffinityTopologyVersion topVer) throws IgniteCheckedException {
+ Map<UUID, CacheContinuousQueryListener> lsnrCol = updateListeners(internal, preload);
- lsnr.skipUpdateEvent(evt, topVer, primary);
- }
+ if (lsnrCol != null) {
+ onEntryUpdated(
+ lsnrCol,
+ key,
+ newVal,
+ oldVal,
+ internal,
+ partId,
+ primary,
+ preload,
+ updateCntr,
+ topVer);
}
}
/**
+ * @param lsnrCol Listeners to notify.
* @param key Key.
* @param newVal New value.
* @param oldVal Old value.
* @param internal Internal entry (internal key or not user cache),
+ * @param partId Partition.
* @param primary {@code True} if called on primary node.
* @param preload Whether update happened during preloading.
* @param updateCntr Update counter.
@@ -213,6 +259,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
* @throws IgniteCheckedException In case of error.
*/
public void onEntryUpdated(
+ Map<UUID, CacheContinuousQueryListener> lsnrCol,
KeyCacheObject key,
CacheObject newVal,
CacheObject oldVal,
@@ -225,25 +272,13 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
throws IgniteCheckedException
{
assert key != null;
-
- if (preload && !internal)
- return;
-
- ConcurrentMap<UUID, CacheContinuousQueryListener> lsnrCol;
-
- if (internal)
- lsnrCol = intLsnrCnt.get() > 0 ? intLsnrs : null;
- else
- lsnrCol = lsnrCnt.get() > 0 ? lsnrs : null;
-
- if (F.isEmpty(lsnrCol))
- return;
+ assert lsnrCol != null;
boolean hasNewVal = newVal != null;
boolean hasOldVal = oldVal != null;
if (!hasNewVal && !hasOldVal) {
- skipUpdateEvent(key, partId, updateCntr, primary, topVer);
+ skipUpdateEvent(lsnrCol, key, partId, updateCntr, topVer);
return;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6291c64e/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index 4b02f57..0218897 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -40,7 +40,6 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.GridMessageListenHandler;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteDeploymentCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -56,6 +55,7 @@ import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -428,17 +428,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.resource().injectGeneric(item.prjPred);
// Register handler only if local node passes projection predicate.
- if ((item.prjPred == null || item.prjPred.apply(ctx.discovery().localNode())) &&
- !locInfos.containsKey(item.routineId)) {
+ if (item.prjPred == null || item.prjPred.apply(ctx.discovery().localNode())) {
if (registerHandler(data.nodeId, item.routineId, item.hnd, item.bufSize, item.interval,
item.autoUnsubscribe, false))
item.hnd.onListenerRegistered(item.routineId, ctx);
}
-
- if (!item.autoUnsubscribe)
- // Register routine locally.
- locInfos.putIfAbsent(item.routineId, new LocalRoutineInfo(
- item.prjPred, item.hnd, item.bufSize, item.interval, item.autoUnsubscribe));
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to register continuous handler.", e);
@@ -860,10 +854,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
U.error(log, "Failed to register handler [nodeId=" + node.id() + ", routineId=" + routineId + ']', e);
}
- GridContinuousHandler hnd0 = hnd instanceof GridMessageListenHandler ?
- new GridMessageListenHandler((GridMessageListenHandler)hnd) :
- hnd;
-
if (node.isClient()) {
Map<UUID, LocalRoutineInfo> clientRoutineMap = clientInfos.get(node.id());
@@ -876,7 +866,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
clientRoutineMap.put(routineId, new LocalRoutineInfo(data.projectionPredicate(),
- hnd0,
+ hnd,
data.bufferSize(),
data.interval(),
data.autoUnsubscribe()));
@@ -891,16 +881,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
if (prjPred != null)
ctx.resource().injectGeneric(prjPred);
- if ((prjPred == null || prjPred.apply(ctx.discovery().node(ctx.localNodeId()))) &&
- !locInfos.containsKey(routineId)) {
- registered = registerHandler(node.id(), routineId, hnd0, data.bufferSize(), data.interval(),
+ if (prjPred == null || prjPred.apply(ctx.discovery().node(ctx.localNodeId()))) {
+ registered = registerHandler(node.id(), routineId, hnd, data.bufferSize(), data.interval(),
data.autoUnsubscribe(), false);
}
-
- if (!data.autoUnsubscribe())
- // Register routine locally.
- locInfos.putIfAbsent(routineId, new LocalRoutineInfo(
- prjPred, hnd0, data.bufferSize(), data.interval(), data.autoUnsubscribe()));
}
catch (IgniteCheckedException e) {
err = e;
@@ -910,18 +894,25 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
// Load partition counters.
- if (hnd0.isQuery() && ctx.cache() != null && ctx.cache().internalCache(hnd0.cacheName()) != null) {
- Map<Integer, Long> cntrs = ctx.cache().internalCache(hnd0.cacheName())
- .context().topology().updateCounters();
+ if (hnd.isQuery()) {
+ GridCacheProcessor proc = ctx.cache();
+
+ if (proc != null) {
+ GridCacheAdapter cache = ctx.cache().internalCache(hnd.cacheName());
+
+ if (cache != null && !cache.isLocal()) {
+ Map<Integer, Long> cntrs = cache.context().topology().updateCounters();
- req.addUpdateCounters(cntrs);
+ req.addUpdateCounters(cntrs);
+ }
+ }
}
if (err != null)
req.addError(ctx.localNodeId(), err);
if (registered)
- hnd0.onListenerRegistered(routineId, ctx);
+ hnd.onListenerRegistered(routineId, ctx);
}
/**
@@ -1104,29 +1095,22 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
*/
@SuppressWarnings("TooBroadScope")
private void unregisterRemote(UUID routineId) {
- RemoteRoutineInfo remote;
- LocalRoutineInfo loc;
+ RemoteRoutineInfo info;
stopLock.lock();
try {
- remote = rmtInfos.remove(routineId);
+ info = rmtInfos.remove(routineId);
- loc = locInfos.remove(routineId);
-
- if (remote == null)
+ if (info == null)
stopped.add(routineId);
}
finally {
stopLock.unlock();
}
- if (remote != null)
- unregisterHandler(routineId, remote.hnd, false);
- else if (loc != null) {
- // Removes routine at node started it when stopRoutine called from another node.
- unregisterHandler(routineId, loc.hnd, false);
- }
+ if (info != null)
+ unregisterHandler(routineId, info.hnd, false);
}
/**