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/05/25 15:37:04 UTC
[01/18] ignite git commit: cc
Repository: ignite
Updated Branches:
refs/heads/ignite-5075-cc-debug 8adfe7d7f -> 58919c5b5
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/79e34c2c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/79e34c2c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/79e34c2c
Branch: refs/heads/ignite-5075-cc-debug
Commit: 79e34c2cfdd28e0df479bb96ba8b667b30566463
Parents: c2c9277
Author: sboikov <sb...@gridgain.com>
Authored: Wed May 24 20:12:00 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed May 24 20:12:00 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/79e34c2c/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 c046095..f7547f5 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
@@ -1156,7 +1156,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
boolean fire = e.getKey() == lastFiredEvt + 1;;
if (!fire && filtered > 0)
- fire = e.getKey() - filtered == lastFiredEvt + 1;
+ fire = e.getKey() - filtered <= lastFiredEvt + 1;
if (fire) {
lastFiredEvt = e.getKey();
[13/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e6ffd828
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e6ffd828
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e6ffd828
Branch: refs/heads/ignite-5075-cc-debug
Commit: e6ffd828828ddc2aba00140742cb0cf30038b520
Parents: 0305930
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:32:03 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:32:03 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e6ffd828/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 98eaecb..b4f2dbd 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
@@ -456,6 +456,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (backupQueue != null && node != null) {
for (CacheContinuousQueryEntry e : backupQueue) {
+ e.markBackup();
+
if (!e.isFiltered())
prepareEntry(cctx, nodeId, e);
}
@@ -891,8 +893,6 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx, e.partition());
- e.markBackup();
-
buf.processEntry(e.forBackupQueue(), true);
}
[14/18] ignite git commit: Merge branch 'ignite-5075-cc' into
ignite-5075-cc-debug
Posted by sb...@apache.org.
Merge branch 'ignite-5075-cc' into ignite-5075-cc-debug
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8fafde28
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8fafde28
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8fafde28
Branch: refs/heads/ignite-5075-cc-debug
Commit: 8fafde2889f44e870635d97fb67b84e5c15c311e
Parents: f5408cb e6ffd82
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:34:51 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:34:51 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8fafde28/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
[12/18] ignite git commit: Merge branch 'ignite-5075-cc' into
ignite-5075-cc-debug
Posted by sb...@apache.org.
Merge branch 'ignite-5075-cc' into ignite-5075-cc-debug
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f5408cbb
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f5408cbb
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f5408cbb
Branch: refs/heads/ignite-5075-cc-debug
Commit: f5408cbb42564c6a6129a914f8726244ea77fe9b
Parents: 529dec1 0305930
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:25:45 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:25:45 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5408cbb/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
[04/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/31f32998
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/31f32998
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/31f32998
Branch: refs/heads/ignite-5075-cc-debug
Commit: 31f32998c8cbcd51797d34069711e27db65e48ab
Parents: 7bf63c0
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 11:47:09 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 11:47:09 2017 +0300
----------------------------------------------------------------------
.../continuous/CacheContinuousQueryEventBuffer.java | 14 +++++++-------
.../query/continuous/CacheContinuousQueryHandler.java | 6 ++++++
.../IgniteCacheContinuousQueryBackupQueueTest.java | 13 +++++++++----
3 files changed, 22 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/31f32998/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 949ea67..f0640b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -41,13 +41,6 @@ public class CacheContinuousQueryEventBuffer {
/** */
protected final int part;
- /**
- * @param part Partition number.
- */
- CacheContinuousQueryEventBuffer(int part) {
- this.part = part;
- }
-
/** */
private AtomicReference<Batch> curBatch = new AtomicReference<>();
@@ -58,6 +51,13 @@ public class CacheContinuousQueryEventBuffer {
private ConcurrentSkipListMap<Long, CacheContinuousQueryEntry> pending = new ConcurrentSkipListMap<>();
/**
+ * @param part Partition number.
+ */
+ CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
+ /**
* @param updateCntr Acknowledged counter.
*/
void cleanupBackupQueue(Long updateCntr) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/31f32998/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 540f871..9866e7b 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
@@ -954,6 +954,12 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** {@inheritDoc} */
@Override public void onNodeLeft() {
nodeLeft = true;
+
+ for (Map.Entry<Integer, CacheContinuousQueryEventBuffer> bufE : entryBufs.entrySet()) {
+ CacheContinuousQueryEventBuffer buf = bufE.getValue();
+
+ buf.resetBackupQueue();
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/31f32998/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
index 26c7d41..85d68d3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
@@ -262,11 +262,16 @@ public class IgniteCacheContinuousQueryBackupQueueTest extends GridCommonAbstrac
GridContinuousHandler hnd = GridTestUtils.getFieldValue(info, "hnd");
if (hnd.isQuery() && hnd.cacheName().equals(CACHE_NAME)) {
- Collection<Object> q = GridTestUtils.getFieldValue(hnd,
- CacheContinuousQueryHandler.class, "backupQueue");
+ Map<Integer, CacheContinuousQueryEventBuffer> map = GridTestUtils.getFieldValue(hnd,
+ CacheContinuousQueryHandler.class, "entryBufs");
- if (q != null)
- backupQueues.add(q);
+ for (CacheContinuousQueryEventBuffer buf : map.values()) {
+ Collection<Object> q = GridTestUtils.getFieldValue(buf,
+ CacheContinuousQueryEventBuffer.class, "backupQ");
+
+ if (q != null)
+ backupQueues.add(q);
+ }
}
}
[03/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7bf63c0e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7bf63c0e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7bf63c0e
Branch: refs/heads/ignite-5075-cc-debug
Commit: 7bf63c0ee7d41317dc558d40267e40d9ed65fda3
Parents: 79e34c2
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 11:30:17 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 11:30:17 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryAcknowledgeBuffer.java | 120 ++++
.../CacheContinuousQueryDeployableObject.java | 110 ++++
.../CacheContinuousQueryEventBuffer.java | 91 ++-
.../continuous/CacheContinuousQueryHandler.java | 602 +++----------------
.../CacheContinuousQueryHandlerV2.java | 6 +-
.../CacheContinuousQueryPartitionRecovery.java | 252 ++++++++
.../continuous/GridContinuousProcessor.java | 7 +-
.../CacheContinuousQueryEventBufferTest.java | 65 +-
8 files changed, 682 insertions(+), 571 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAcknowledgeBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAcknowledgeBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAcknowledgeBuffer.java
new file mode 100644
index 0000000..c95dc42
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAcknowledgeBuffer.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.continuous.GridContinuousBatch;
+import org.apache.ignite.internal.processors.continuous.GridContinuousQueryBatch;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+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.jetbrains.annotations.Nullable;
+
+/** */
+class CacheContinuousQueryAcknowledgeBuffer {
+ /** */
+ private int size;
+
+ /** */
+ @GridToStringInclude
+ private Map<Integer, Long> updateCntrs = new HashMap<>();
+
+ /** */
+ @GridToStringInclude
+ private Set<AffinityTopologyVersion> topVers = U.newHashSet(1);
+
+ /**
+ * @param batch Batch.
+ * @return Non-null tuple if acknowledge should be sent to backups.
+ */
+ @SuppressWarnings("unchecked")
+ @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>onAcknowledged(
+ GridContinuousBatch batch) {
+ assert batch instanceof GridContinuousQueryBatch;
+
+ size += ((GridContinuousQueryBatch)batch).entriesCount();
+
+ Collection<CacheContinuousQueryEntry> entries = (Collection)batch.collect();
+
+ for (CacheContinuousQueryEntry e : entries)
+ addEntry(e);
+
+ return size >= CacheContinuousQueryHandler.BACKUP_ACK_THRESHOLD ? acknowledgeData() : null;
+ }
+
+ /**
+ * @param e Entry.
+ * @return Non-null tuple if acknowledge should be sent to backups.
+ */
+ @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>
+ onAcknowledged(CacheContinuousQueryEntry e) {
+ size++;
+
+ addEntry(e);
+
+ return size >= CacheContinuousQueryHandler.BACKUP_ACK_THRESHOLD ? acknowledgeData() : null;
+ }
+
+ /**
+ * @param e Entry.
+ */
+ private void addEntry(CacheContinuousQueryEntry e) {
+ topVers.add(e.topologyVersion());
+
+ Long cntr0 = updateCntrs.get(e.partition());
+
+ if (cntr0 == null || e.updateCounter() > cntr0)
+ updateCntrs.put(e.partition(), e.updateCounter());
+ }
+
+ /**
+ * @return Non-null tuple if acknowledge should be sent to backups.
+ */
+ @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>
+ acknowledgeOnTimeout() {
+ return size > 0 ? acknowledgeData() : null;
+ }
+
+ /**
+ * @return Tuple with acknowledge information.
+ */
+ private IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>> acknowledgeData() {
+ assert size > 0;
+
+ Map<Integer, Long> cntrs = new HashMap<>(updateCntrs);
+
+ IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>> res =
+ new IgniteBiTuple<>(cntrs, topVers);
+
+ topVers = U.newHashSet(1);
+
+ size = 0;
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(CacheContinuousQueryAcknowledgeBuffer.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java
new file mode 100644
index 0000000..f888467
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteDeploymentCheckedException;
+import org.apache.ignite.internal.managers.deployment.GridDeployment;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Deployable object.
+ */
+class CacheContinuousQueryDeployableObject implements Externalizable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Serialized object. */
+ private byte[] bytes;
+
+ /** Deployment class name. */
+ private String clsName;
+
+ /** Deployment info. */
+ private GridDeploymentInfo depInfo;
+
+ /**
+ * Required by {@link Externalizable}.
+ */
+ public CacheContinuousQueryDeployableObject() {
+ // No-op.
+ }
+
+ /**
+ * @param obj Object.
+ * @param ctx Kernal context.
+ * @throws IgniteCheckedException In case of error.
+ */
+ protected CacheContinuousQueryDeployableObject(Object obj, GridKernalContext ctx) throws IgniteCheckedException {
+ assert obj != null;
+ assert ctx != null;
+
+ Class cls = U.detectClass(obj);
+
+ clsName = cls.getName();
+
+ GridDeployment dep = ctx.deploy().deploy(cls, U.detectClassLoader(cls));
+
+ if (dep == null)
+ throw new IgniteDeploymentCheckedException("Failed to deploy object: " + obj);
+
+ depInfo = new GridDeploymentInfoBean(dep);
+
+ bytes = U.marshal(ctx, obj);
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param ctx Kernal context.
+ * @return Deserialized object.
+ * @throws IgniteCheckedException In case of error.
+ */
+ <T> T unmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException {
+ assert ctx != null;
+
+ GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
+ depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(), null);
+
+ if (dep == null)
+ throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + clsName);
+
+ return U.unmarshal(ctx, bytes, U.resolveClassLoader(dep.classLoader(), ctx.config()));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ U.writeByteArray(out, bytes);
+ U.writeString(out, clsName);
+ out.writeObject(depInfo);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ bytes = U.readByteArray(in);
+ clsName = U.readString(in);
+ depInfo = (GridDeploymentInfo)in.readObject();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index a308e39..949ea67 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -19,8 +19,11 @@ package org.apache.ignite.internal.processors.cache.query.continuous;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteSystemProperties;
@@ -36,10 +39,52 @@ public class CacheContinuousQueryEventBuffer {
IgniteSystemProperties.getInteger("IGNITE_CONTINUOUS_QUERY_SERVER_BUFFER_SIZE", 5);
/** */
+ protected final int part;
+
+ /**
+ * @param part Partition number.
+ */
+ CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
+ /** */
private AtomicReference<Batch> curBatch = new AtomicReference<>();
/** */
- private ConcurrentSkipListMap<Long, Object> pending = new ConcurrentSkipListMap<>();
+ private ConcurrentLinkedDeque<CacheContinuousQueryEntry> backupQ = new ConcurrentLinkedDeque<>();
+
+ /** */
+ private ConcurrentSkipListMap<Long, CacheContinuousQueryEntry> pending = new ConcurrentSkipListMap<>();
+
+ /**
+ * @param updateCntr Acknowledged counter.
+ */
+ void cleanupBackupQueue(Long updateCntr) {
+ Iterator<CacheContinuousQueryEntry> it = backupQ.iterator();
+
+ while (it.hasNext()) {
+ CacheContinuousQueryEntry backupEntry = it.next();
+
+ if (backupEntry.updateCounter() <= updateCntr)
+ it.remove();
+ }
+ }
+
+ /**
+ * @return Backup entries.
+ */
+ @Nullable Collection<CacheContinuousQueryEntry> resetBackupQueue() {
+ if (!backupQ.isEmpty()) {
+ ConcurrentLinkedDeque<CacheContinuousQueryEntry> ret = this.backupQ;
+
+ backupQ = new ConcurrentLinkedDeque<>();
+
+ return ret;
+ }
+
+ return null;
+ }
/**
* @return Initial partition counter.
@@ -61,26 +106,20 @@ public class CacheContinuousQueryEventBuffer {
/**
* @param e Entry to process.
+ * @param backup Backup entry flag.
* @return Collected entries to pass to listener (single entry or entries list).
*/
- @Nullable Object processEntry(CacheContinuousQueryEntry e) {
- return process0(e.updateCounter(), e);
- }
-
- /**
- * @param cntr Filtered counter.
- * @return Collected entries to pass to listener (single entry or entries list).
- */
- @Nullable Object processFiltered(long cntr) {
- return process0(cntr, cntr);
+ @Nullable Object processEntry(CacheContinuousQueryEntry e, boolean backup) {
+ return process0(e.updateCounter(), e, backup);
}
/**
+ * @param backup Backup entry flag.
* @param cntr Entry counter.
* @param entry Entry.
* @return Collected entries.
*/
- private Object process0(long cntr, Object entry) {
+ private Object process0(long cntr, CacheContinuousQueryEntry entry, boolean backup) {
assert cntr >= 0 : cntr;
Batch batch = initBatch();
@@ -88,13 +127,16 @@ public class CacheContinuousQueryEventBuffer {
if (batch == null || cntr < batch.startCntr) {
assert entry != null : cntr;
+ if (backup)
+ backupQ.add(entry);
+
return entry;
}
Object res = null;
if (cntr <= batch.endCntr)
- res = batch.processEvent0(null, cntr, entry);
+ res = batch.processEvent0(null, cntr, entry, backup);
else
pending.put(cntr, entry);
@@ -104,7 +146,7 @@ public class CacheContinuousQueryEventBuffer {
do {
batch = batch0;
- res = processPending(res, batch);
+ res = processPending(res, batch, backup);
batch0 = curBatch.get();
}
@@ -139,17 +181,18 @@ public class CacheContinuousQueryEventBuffer {
/**
* @param res Current result.
* @param batch Current batch.
+ * @param backup Backup entry flag.
* @return New result.
*/
- @Nullable private Object processPending(@Nullable Object res, Batch batch) {
+ @Nullable private Object processPending(@Nullable Object res, Batch batch, boolean backup) {
if (pending.floorKey(batch.endCntr) != null) {
- for (Map.Entry<Long, Object> p : pending.headMap(batch.endCntr, true).entrySet()) {
+ for (Map.Entry<Long, CacheContinuousQueryEntry> p : pending.headMap(batch.endCntr, true).entrySet()) {
long cntr = p.getKey();
assert cntr >= batch.startCntr && cntr <= batch.endCntr : cntr;
if (pending.remove(p.getKey()) != null)
- res = batch.processEvent0(res, p.getKey(), p.getValue());
+ res = batch.processEvent0(res, p.getKey(), p.getValue(), backup);
}
}
@@ -195,13 +238,15 @@ public class CacheContinuousQueryEventBuffer {
* @param res Current result.
* @param cntr Event counter.
* @param evt Event.
+ * @param backup Backup entry flag.
* @return New result.
*/
@SuppressWarnings("unchecked")
@Nullable private Object processEvent0(
@Nullable Object res,
long cntr,
- Object evt) {
+ CacheContinuousQueryEntry evt,
+ boolean backup) {
int pos = (int)(cntr - startCntr);
synchronized (this) {
@@ -224,9 +269,15 @@ public class CacheContinuousQueryEventBuffer {
filtered = 0;
- if (res == null)
- res = evt0;
+ if (res == null) {
+ if (backup)
+ backupQ.add(evt0);
+ else
+ res = evt0;
+ }
else {
+ assert !backup;
+
List<CacheContinuousQueryEntry> resList;
if (res instanceof CacheContinuousQueryEntry) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/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 f7547f5..540f871 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
@@ -24,12 +24,9 @@ import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -46,13 +43,10 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheQueryExecutedEvent;
import org.apache.ignite.events.CacheQueryReadEvent;
import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.IgniteDeploymentCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
-import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
-import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheAffinityManager;
@@ -67,12 +61,10 @@ import org.apache.ignite.internal.processors.continuous.GridContinuousBatch;
import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
import org.apache.ignite.internal.processors.continuous.GridContinuousQueryBatch;
import org.apache.ignite.internal.processors.platform.cache.query.PlatformContinuousQueryFilter;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteAsyncCallback;
@@ -80,7 +72,6 @@ import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentLinkedDeque8;
import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_EXECUTED;
import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_OBJECT_READ;
@@ -93,11 +84,11 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private static final long serialVersionUID = 0L;
/** */
- private static final int BACKUP_ACK_THRESHOLD =
+ static final int BACKUP_ACK_THRESHOLD =
IgniteSystemProperties.getInteger("IGNITE_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD", 100);
/** */
- private static final int LSNR_MAX_BUF_SIZE =
+ static final int LSNR_MAX_BUF_SIZE =
IgniteSystemProperties.getInteger("IGNITE_CONTINUOUS_QUERY_LISTENER_MAX_BUFFER_SIZE", 10_000);
/** Cache name. */
@@ -113,7 +104,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private CacheEntryEventSerializableFilter<K, V> rmtFilter;
/** Deployable object for filter. */
- private DeployableObject rmtFilterDep;
+ private CacheContinuousQueryDeployableObject rmtFilterDep;
/** Internal flag. */
private boolean internal;
@@ -136,9 +127,6 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** Whether to skip primary check for REPLICATED cache. */
private transient boolean skipPrimaryCheck;
- /** Backup queue. */
- private transient volatile Collection<CacheContinuousQueryEntry> backupQueue;
-
/** */
private boolean locCache;
@@ -146,13 +134,13 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private boolean keepBinary;
/** */
- private transient ConcurrentMap<Integer, PartitionRecovery> rcvs;
+ private transient ConcurrentMap<Integer, CacheContinuousQueryPartitionRecovery> rcvs;
/** */
private transient ConcurrentMap<Integer, CacheContinuousQueryEventBuffer> entryBufs;
/** */
- private transient AcknowledgeBuffer ackBuf;
+ private transient CacheContinuousQueryAcknowledgeBuffer ackBuf;
/** */
private transient int cacheId;
@@ -167,6 +155,9 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private transient volatile AffinityTopologyVersion initTopVer;
/** */
+ private transient volatile boolean nodeLeft;
+
+ /** */
private transient boolean ignoreClsNotFound;
/** */
@@ -341,9 +332,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
entryBufs = new ConcurrentHashMap<>();
- backupQueue = new ConcurrentLinkedDeque8<>();
-
- ackBuf = new AcknowledgeBuffer();
+ ackBuf = new CacheContinuousQueryAcknowledgeBuffer();
rcvs = new ConcurrentHashMap<>();
@@ -413,7 +402,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
ctx.asyncCallbackPool().execute(clsr, evt.partitionId());
}
else {
- final boolean notify = filter(evt, primary);
+ final boolean notify = filter(evt);
if (log.isDebugEnabled())
log.debug("Filter invoked for event [evt=" + evt + ", primary=" + primary
@@ -433,6 +422,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}, sync);
}
}
+ else
+ handleBackupEntry(cctx, evt.entry());
}
}
@@ -442,50 +433,38 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
@Override public void cleanupBackupQueue(Map<Integer, Long> updateCntrs) {
- Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
-
- if (backupQueue0 != null) {
- Iterator<CacheContinuousQueryEntry> it = backupQueue0.iterator();
-
- while (it.hasNext()) {
- CacheContinuousQueryEntry backupEntry = it.next();
-
- Long updateCntr = updateCntrs.get(backupEntry.partition());
+ for (Map.Entry<Integer, Long> e : updateCntrs.entrySet()) {
+ CacheContinuousQueryEventBuffer buf = entryBufs.get(e.getKey());
- if (updateCntr != null && backupEntry.updateCounter() <= updateCntr)
- it.remove();
- }
+ if (buf != null)
+ buf.cleanupBackupQueue(e.getValue());
}
}
@Override public void flushBackupQueue(GridKernalContext ctx, AffinityTopologyVersion topVer) {
assert topVer != null;
- Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
+ try {
+ GridCacheContext<K, V> cctx = cacheContext(ctx);
- if (backupQueue0 == null)
- return;
+ ClusterNode node = ctx.discovery().node(nodeId);
- try {
- ClusterNode nodeId0 = ctx.discovery().node(nodeId);
+ for (Map.Entry<Integer, CacheContinuousQueryEventBuffer> bufE : entryBufs.entrySet()) {
+ CacheContinuousQueryEventBuffer buf = bufE.getValue();
- if (nodeId0 != null) {
- GridCacheContext<K, V> cctx = cacheContext(ctx);
+ Collection<CacheContinuousQueryEntry> backupQueue = buf.resetBackupQueue();
- for (CacheContinuousQueryEntry e : backupQueue0) {
- if (!e.isFiltered())
- prepareEntry(cctx, nodeId, e);
+ if (backupQueue != null && node != null) {
+ for (CacheContinuousQueryEntry e : backupQueue) {
+ if (!e.isFiltered())
+ prepareEntry(cctx, nodeId, e);
- e.topologyVersion(topVer);
- }
+ e.topologyVersion(topVer);
+ }
- ctx.continuous().addBackupNotification(nodeId, routineId, backupQueue0, topic);
+ ctx.continuous().addBackupNotification(nodeId, routineId, backupQueue, topic);
+ }
}
- else
- // Node which start CQ leave topology. Not needed to put data to backup queue.
- backupQueue = null;
-
- backupQueue0.clear();
}
catch (IgniteCheckedException e) {
U.error(ctx.log(CU.CONTINUOUS_QRY_LOG_CATEGORY),
@@ -509,14 +488,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
@Override public void onPartitionEvicted(int part) {
- Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
-
- if (backupQueue0 != null) {
- for (Iterator<CacheContinuousQueryEntry> it = backupQueue0.iterator(); it.hasNext(); ) {
- if (it.next().partition() == part)
- it.remove();
- }
- }
+ entryBufs.remove(part);
}
@Override public boolean oldValueRequired() {
@@ -743,17 +715,16 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
Collections.<CacheEntryEvent<? extends K, ? extends V>>emptyList();
}
- PartitionRecovery rec = getOrCreatePartitionRecovery(ctx, e.partition(), e.topologyVersion());
+ CacheContinuousQueryPartitionRecovery rec = getOrCreatePartitionRecovery(ctx, e.partition(), e.topologyVersion());
return rec.collectEntries(e, cctx, cache);
}
/**
- * @param primary Primary.
* @param evt Query event.
* @return {@code True} if event passed filter otherwise {@code true}.
*/
- public boolean filter(CacheContinuousQueryEvent evt, boolean primary) {
+ public boolean filter(CacheContinuousQueryEvent evt) {
CacheContinuousQueryEntry entry = evt.entry();
boolean notify = !entry.isFiltered();
@@ -769,15 +740,6 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (!notify)
entry.markFiltered();
- if (!primary && !internal && entry.updateCounter() != -1L /* Skip init query and expire entries */) {
- entry.markBackup();
-
- Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
-
- if (backupQueue0 != null)
- backupQueue0.add(entry.forBackupQueue());
- }
-
return notify;
}
@@ -869,7 +831,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (internal)
return;
- for (PartitionRecovery rec : rcvs.values())
+ for (CacheContinuousQueryPartitionRecovery rec : rcvs.values())
rec.resetTopologyCache();
}
@@ -879,12 +841,12 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
* @param topVer Topology version for current operation.
* @return Partition recovery.
*/
- @NotNull private PartitionRecovery getOrCreatePartitionRecovery(GridKernalContext ctx,
+ @NotNull private CacheContinuousQueryPartitionRecovery getOrCreatePartitionRecovery(GridKernalContext ctx,
int partId,
AffinityTopologyVersion topVer) {
assert topVer != null && topVer.topologyVersion() > 0 : topVer;
- PartitionRecovery rec = rcvs.get(partId);
+ CacheContinuousQueryPartitionRecovery rec = rcvs.get(partId);
if (rec == null) {
T2<Long, Long> partCntrs = null;
@@ -909,10 +871,10 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
else if (initUpdCntrs != null)
partCntrs = initUpdCntrs.get(partId);
- rec = new PartitionRecovery(ctx.log(CU.CONTINUOUS_QRY_LOG_CATEGORY), topVer,
+ rec = new CacheContinuousQueryPartitionRecovery(ctx.log(CU.CONTINUOUS_QRY_LOG_CATEGORY), topVer,
partCntrs != null ? partCntrs.get2() : null);
- PartitionRecovery oldRec = rcvs.putIfAbsent(partId, rec);
+ CacheContinuousQueryPartitionRecovery oldRec = rcvs.putIfAbsent(partId, rec);
if (oldRec != null)
rec = oldRec;
@@ -924,6 +886,19 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/**
* @param cctx Cache context.
* @param e Entry.
+ */
+ private void handleBackupEntry(final GridCacheContext cctx, CacheContinuousQueryEntry e) {
+ if (internal || e.updateCounter() == -1L || nodeLeft) // Skip internal query and expire entries.
+ return;
+
+ CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx, e.partition());
+
+ buf.processEntry(e.forBackupQueue(), true);
+ }
+
+ /**
+ * @param cctx Cache context.
+ * @param e Entry.
* @return Entry.
*/
private Object handleEntry(final GridCacheContext cctx, CacheContinuousQueryEntry e) {
@@ -942,12 +917,21 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (e.updateCounter() == -1L)
return e;
- CacheContinuousQueryEventBuffer buf = entryBufs.get(e.partition());
+ CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx, e.partition());
- if (buf == null) {
- final int part = e.partition();
+ return buf.processEntry(e, false);
+ }
- buf = new CacheContinuousQueryEventBuffer() {
+ /**
+ * @param cctx Cache context.
+ * @param part Partition.
+ * @return Event buffer.
+ */
+ private CacheContinuousQueryEventBuffer partitionBuffer(final GridCacheContext cctx, int part) {
+ CacheContinuousQueryEventBuffer buf = entryBufs.get(part);
+
+ if (buf == null) {
+ buf = new CacheContinuousQueryEventBuffer(part) {
@Override protected long currentPartitionCounter() {
GridDhtLocalPartition locPart = cctx.topology().localPartition(part, null, false);
@@ -958,239 +942,18 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
};
- CacheContinuousQueryEventBuffer oldBuf = entryBufs.putIfAbsent(e.partition(), buf);
+ CacheContinuousQueryEventBuffer oldBuf = entryBufs.putIfAbsent(part, buf);
if (oldBuf != null)
buf = oldBuf;
}
- return buf.processEntry(e);
- }
-
- /**
- *
- */
- private static class PartitionRecovery {
- /** Event which means hole in sequence. */
- private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
-
- /** */
- private final static int MAX_BUFF_SIZE = LSNR_MAX_BUF_SIZE;
-
- /** */
- private IgniteLogger log;
-
- /** */
- private long lastFiredEvt;
-
- /** */
- private AffinityTopologyVersion curTop = AffinityTopologyVersion.NONE;
-
- /** */
- private final Map<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
-
- /**
- * @param log Logger.
- * @param topVer Topology version.
- * @param initCntr Update counters.
- */
- PartitionRecovery(IgniteLogger log, AffinityTopologyVersion topVer, @Nullable Long initCntr) {
- this.log = log;
-
- if (initCntr != null) {
- assert topVer.topologyVersion() > 0 : topVer;
-
- this.lastFiredEvt = initCntr;
-
- curTop = topVer;
- }
- }
-
- /**
- * Resets cached topology.
- */
- void resetTopologyCache() {
- curTop = AffinityTopologyVersion.NONE;
- }
-
- /**
- * Add continuous entry.
- *
- * @param cctx Cache context.
- * @param cache Cache.
- * @param entry Cache continuous query entry.
- * @return Collection entries which will be fired. This collection should contains only non-filtered events.
- */
- <K, V> Collection<CacheEntryEvent<? extends K, ? extends V>> collectEntries(
- CacheContinuousQueryEntry entry,
- GridCacheContext cctx,
- IgniteCache cache
- ) {
- assert entry != null;
-
- if (entry.topologyVersion() == null) { // Possible if entry is sent from old node.
- assert entry.updateCounter() == 0L : entry;
-
- return F.<CacheEntryEvent<? extends K, ? extends V>>
- asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
- }
-
- List<CacheEntryEvent<? extends K, ? extends V>> entries;
-
- synchronized (pendingEvts) {
- if (log.isDebugEnabled()) {
- log.debug("Handling event [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- // Received first event.
- if (curTop == AffinityTopologyVersion.NONE) {
- lastFiredEvt = entry.updateCounter();
-
- curTop = entry.topologyVersion();
-
- if (log.isDebugEnabled()) {
- log.debug("First event [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() + ']');
- }
-
- return !entry.isFiltered() ?
- F.<CacheEntryEvent<? extends K, ? extends V>>
- asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry)) :
- Collections.<CacheEntryEvent<? extends K, ? extends V>>emptyList();
- }
-
- if (curTop.compareTo(entry.topologyVersion()) < 0) {
- if (entry.updateCounter() == 1L && !entry.isBackup()) {
- entries = new ArrayList<>(pendingEvts.size());
-
- for (CacheContinuousQueryEntry evt : pendingEvts.values()) {
- if (evt != HOLE && !evt.isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, evt));
- }
-
- pendingEvts.clear();
-
- curTop = entry.topologyVersion();
-
- lastFiredEvt = entry.updateCounter();
-
- if (!entry.isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
-
- if (log.isDebugEnabled())
- log.debug("Partition was lost [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
-
- return entries;
- }
-
- curTop = entry.topologyVersion();
- }
-
- // Check duplicate.
- if (entry.updateCounter() > lastFiredEvt)
- pendingEvts.put(entry.updateCounter(), entry);
- else {
- if (log.isDebugEnabled())
- log.debug("Skip duplicate continuous query message: " + entry);
-
- return Collections.emptyList();
- }
-
- if (pendingEvts.isEmpty()) {
- if (log.isDebugEnabled()) {
- log.debug("Nothing sent to listener [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() + ']');
- }
-
- return Collections.emptyList();
- }
-
- Iterator<Map.Entry<Long, CacheContinuousQueryEntry>> iter = pendingEvts.entrySet().iterator();
-
- entries = new ArrayList<>();
-
- if (pendingEvts.size() >= MAX_BUFF_SIZE) {
- if (log.isDebugEnabled()) {
- log.debug("Pending events reached max of buffer size [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- LT.warn(log, "Pending events reached max of buffer size [cache=" + cctx.name() +
- ", bufSize=" + MAX_BUFF_SIZE +
- ", partId=" + entry.partition() + ']');
-
- for (int i = 0; i < MAX_BUFF_SIZE - (MAX_BUFF_SIZE / 10); i++) {
- Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
-
- if (e.getValue() != HOLE && !e.getValue().isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
-
- lastFiredEvt = e.getKey();
-
- iter.remove();
- }
- }
- else {
- while (iter.hasNext()) {
- Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
-
- CacheContinuousQueryEntry pending = e.getValue();
-
- long filtered = pending.filteredCount();
-
- boolean fire = e.getKey() == lastFiredEvt + 1;;
-
- if (!fire && filtered > 0)
- fire = e.getKey() - filtered <= lastFiredEvt + 1;
-
- if (fire) {
- lastFiredEvt = e.getKey();
-
- if (e.getValue() != HOLE && !e.getValue().isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
-
- iter.remove();
- }
- else
- break;
- }
- }
- }
-
- if (log.isDebugEnabled()) {
- log.debug("Will send to listener the following events [entries=" + entries +
- ", lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- return entries;
- }
+ return buf;
}
/** {@inheritDoc} */
@Override public void onNodeLeft() {
- Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
-
- if (backupQueue0 != null)
- backupQueue = null;
+ nodeLeft = true;
}
/** {@inheritDoc} */
@@ -1199,7 +962,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
assert ctx.config().isPeerClassLoadingEnabled();
if (rmtFilter != null && !U.isGrid(rmtFilter.getClass()))
- rmtFilterDep = new DeployableObject(rmtFilter, ctx);
+ rmtFilterDep = new CacheContinuousQueryDeployableObject(rmtFilter, ctx);
}
/** {@inheritDoc} */
@@ -1320,7 +1083,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
boolean b = in.readBoolean();
if (b)
- rmtFilterDep = (DeployableObject)in.readObject();
+ rmtFilterDep = (CacheContinuousQueryDeployableObject)in.readObject();
else
rmtFilter = (CacheEntryEventSerializableFilter<K, V>)in.readObject();
@@ -1345,95 +1108,6 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
return ctx.cache().<K, V>context().cacheContext(cacheId);
}
- /** */
- private static class AcknowledgeBuffer {
- /** */
- private int size;
-
- /** */
- @GridToStringInclude
- private Map<Integer, Long> updateCntrs = new HashMap<>();
-
- /** */
- @GridToStringInclude
- private Set<AffinityTopologyVersion> topVers = U.newHashSet(1);
-
- /**
- * @param batch Batch.
- * @return Non-null tuple if acknowledge should be sent to backups.
- */
- @SuppressWarnings("unchecked")
- @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>
- onAcknowledged(GridContinuousBatch batch) {
- assert batch instanceof GridContinuousQueryBatch;
-
- size += ((GridContinuousQueryBatch)batch).entriesCount();
-
- Collection<CacheContinuousQueryEntry> entries = (Collection)batch.collect();
-
- for (CacheContinuousQueryEntry e : entries)
- addEntry(e);
-
- return size >= BACKUP_ACK_THRESHOLD ? acknowledgeData() : null;
- }
-
- /**
- * @param e Entry.
- * @return Non-null tuple if acknowledge should be sent to backups.
- */
- @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>
- onAcknowledged(CacheContinuousQueryEntry e) {
- size++;
-
- addEntry(e);
-
- return size >= BACKUP_ACK_THRESHOLD ? acknowledgeData() : null;
- }
-
- /**
- * @param e Entry.
- */
- private void addEntry(CacheContinuousQueryEntry e) {
- topVers.add(e.topologyVersion());
-
- Long cntr0 = updateCntrs.get(e.partition());
-
- if (cntr0 == null || e.updateCounter() > cntr0)
- updateCntrs.put(e.partition(), e.updateCounter());
- }
-
- /**
- * @return Non-null tuple if acknowledge should be sent to backups.
- */
- @Nullable synchronized IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>>
- acknowledgeOnTimeout() {
- return size > 0 ? acknowledgeData() : null;
- }
-
- /**
- * @return Tuple with acknowledge information.
- */
- private IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>> acknowledgeData() {
- assert size > 0;
-
- Map<Integer, Long> cntrs = new HashMap<>(updateCntrs);
-
- IgniteBiTuple<Map<Integer, Long>, Set<AffinityTopologyVersion>> res =
- new IgniteBiTuple<>(cntrs, topVers);
-
- topVers = U.newHashSet(1);
-
- size = 0;
-
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(AcknowledgeBuffer.class, this);
- }
- }
-
/**
*
*/
@@ -1469,44 +1143,38 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** {@inheritDoc} */
@Override public void run() {
- final boolean notify = filter(evt, primary);
-
- if (!primary())
- return;
+ final boolean notify = filter(evt);
- if (fut == null) {
- onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+ if (primary || skipPrimaryCheck) {
+ if (fut == null) {
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
- return;
- }
+ return;
+ }
- if (fut.isDone()) {
- if (fut.error() != null)
- evt.entry().markFiltered();
+ if (fut.isDone()) {
+ if (fut.error() != null)
+ evt.entry().markFiltered();
- onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
- }
- else {
- fut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> f) {
- if (f.error() != null)
- evt.entry().markFiltered();
-
- ctx.asyncCallbackPool().execute(new Runnable() {
- @Override public void run() {
- onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
- }
- }, evt.entry().partition());
- }
- });
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+ }
+ else {
+ fut.listen(new CI1<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> f) {
+ if (f.error() != null)
+ evt.entry().markFiltered();
+
+ ctx.asyncCallbackPool().execute(new Runnable() {
+ @Override public void run() {
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+ }
+ }, evt.entry().partition());
+ }
+ });
+ }
}
- }
-
- /**
- * @return {@code True} if event fired on this node.
- */
- private boolean primary() {
- return primary || skipPrimaryCheck;
+ else
+ handleBackupEntry(cacheContext(ctx), evt.entry());
}
/** {@inheritDoc} */
@@ -1515,82 +1183,4 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
}
- /**
- * Deployable object.
- */
- protected static class DeployableObject implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Serialized object. */
- private byte[] bytes;
-
- /** Deployment class name. */
- private String clsName;
-
- /** Deployment info. */
- private GridDeploymentInfo depInfo;
-
- /**
- * Required by {@link Externalizable}.
- */
- public DeployableObject() {
- // No-op.
- }
-
- /**
- * @param obj Object.
- * @param ctx Kernal context.
- * @throws IgniteCheckedException In case of error.
- */
- protected DeployableObject(Object obj, GridKernalContext ctx) throws IgniteCheckedException {
- assert obj != null;
- assert ctx != null;
-
- Class cls = U.detectClass(obj);
-
- clsName = cls.getName();
-
- GridDeployment dep = ctx.deploy().deploy(cls, U.detectClassLoader(cls));
-
- if (dep == null)
- throw new IgniteDeploymentCheckedException("Failed to deploy object: " + obj);
-
- depInfo = new GridDeploymentInfoBean(dep);
-
- bytes = U.marshal(ctx, obj);
- }
-
- /**
- * @param nodeId Node ID.
- * @param ctx Kernal context.
- * @return Deserialized object.
- * @throws IgniteCheckedException In case of error.
- */
- <T> T unmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException {
- assert ctx != null;
-
- GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
- depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(), null);
-
- if (dep == null)
- throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + clsName);
-
- return U.unmarshal(ctx, bytes, U.resolveClassLoader(dep.classLoader(), ctx.config()));
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- U.writeByteArray(out, bytes);
- U.writeString(out, clsName);
- out.writeObject(depInfo);
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- bytes = U.readByteArray(in);
- clsName = U.readString(in);
- depInfo = (GridDeploymentInfo)in.readObject();
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
index 7aef4dd..e48d22e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
@@ -44,7 +44,7 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
private Factory<? extends CacheEntryEventFilter> rmtFilterFactory;
/** Deployable object for filter factory. */
- private DeployableObject rmtFilterFactoryDep;
+ private CacheContinuousQueryDeployableObject rmtFilterFactoryDep;
/** Event types for JCache API. */
private byte types;
@@ -122,7 +122,7 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
super.p2pMarshal(ctx);
if (rmtFilterFactory != null && !U.isGrid(rmtFilterFactory.getClass()))
- rmtFilterFactoryDep = new DeployableObject(rmtFilterFactory, ctx);
+ rmtFilterFactoryDep = new CacheContinuousQueryDeployableObject(rmtFilterFactory, ctx);
}
/** {@inheritDoc} */
@@ -167,7 +167,7 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
boolean b = in.readBoolean();
if (b)
- rmtFilterFactoryDep = (DeployableObject)in.readObject();
+ rmtFilterFactoryDep = (CacheContinuousQueryDeployableObject)in.readObject();
else
rmtFilterFactory = (Factory)in.readObject();
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
new file mode 100644
index 0000000..534ce9c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.cache.event.CacheEntryEvent;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.LT;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+class CacheContinuousQueryPartitionRecovery {
+ /** Event which means hole in sequence. */
+ private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
+
+ /** */
+ private final static int MAX_BUFF_SIZE = CacheContinuousQueryHandler.LSNR_MAX_BUF_SIZE;
+
+ /** */
+ private IgniteLogger log;
+
+ /** */
+ private long lastFiredEvt;
+
+ /** */
+ private AffinityTopologyVersion curTop = AffinityTopologyVersion.NONE;
+
+ /** */
+ private final Map<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
+
+ /**
+ * @param log Logger.
+ * @param topVer Topology version.
+ * @param initCntr Update counters.
+ */
+ CacheContinuousQueryPartitionRecovery(IgniteLogger log, AffinityTopologyVersion topVer, @Nullable Long initCntr) {
+ this.log = log;
+
+ if (initCntr != null) {
+ assert topVer.topologyVersion() > 0 : topVer;
+
+ this.lastFiredEvt = initCntr;
+
+ curTop = topVer;
+ }
+ }
+
+ /**
+ * Resets cached topology.
+ */
+ void resetTopologyCache() {
+ curTop = AffinityTopologyVersion.NONE;
+ }
+
+ /**
+ * Add continuous entry.
+ *
+ * @param cctx Cache context.
+ * @param cache Cache.
+ * @param entry Cache continuous query entry.
+ * @return Collection entries which will be fired. This collection should contains only non-filtered events.
+ */
+ <K, V> Collection<CacheEntryEvent<? extends K, ? extends V>> collectEntries(
+ CacheContinuousQueryEntry entry,
+ GridCacheContext cctx,
+ IgniteCache cache
+ ) {
+ assert entry != null;
+
+ if (entry.topologyVersion() == null) { // Possible if entry is sent from old node.
+ assert entry.updateCounter() == 0L : entry;
+
+ return F.<CacheEntryEvent<? extends K, ? extends V>>
+ asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
+ }
+
+ List<CacheEntryEvent<? extends K, ? extends V>> entries;
+
+ synchronized (pendingEvts) {
+ if (log.isDebugEnabled()) {
+ log.debug("Handling event [lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() +
+ ", pendingEvts=" + pendingEvts + ']');
+ }
+
+ // Received first event.
+ if (curTop == AffinityTopologyVersion.NONE) {
+ lastFiredEvt = entry.updateCounter();
+
+ curTop = entry.topologyVersion();
+
+ if (log.isDebugEnabled()) {
+ log.debug("First event [lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() + ']');
+ }
+
+ return !entry.isFiltered() ?
+ F.<CacheEntryEvent<? extends K, ? extends V>>
+ asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry)) :
+ Collections.<CacheEntryEvent<? extends K, ? extends V>>emptyList();
+ }
+
+ if (curTop.compareTo(entry.topologyVersion()) < 0) {
+ if (entry.updateCounter() == 1L && !entry.isBackup()) {
+ entries = new ArrayList<>(pendingEvts.size());
+
+ for (CacheContinuousQueryEntry evt : pendingEvts.values()) {
+ if (evt != HOLE && !evt.isFiltered())
+ entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, evt));
+ }
+
+ pendingEvts.clear();
+
+ curTop = entry.topologyVersion();
+
+ lastFiredEvt = entry.updateCounter();
+
+ if (!entry.isFiltered())
+ entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
+
+ if (log.isDebugEnabled())
+ log.debug("Partition was lost [lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() +
+ ", pendingEvts=" + pendingEvts + ']');
+
+ return entries;
+ }
+
+ curTop = entry.topologyVersion();
+ }
+
+ // Check duplicate.
+ if (entry.updateCounter() > lastFiredEvt)
+ pendingEvts.put(entry.updateCounter(), entry);
+ else {
+ if (log.isDebugEnabled())
+ log.debug("Skip duplicate continuous query message: " + entry);
+
+ return Collections.emptyList();
+ }
+
+ if (pendingEvts.isEmpty()) {
+ if (log.isDebugEnabled()) {
+ log.debug("Nothing sent to listener [lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() + ']');
+ }
+
+ return Collections.emptyList();
+ }
+
+ Iterator<Map.Entry<Long, CacheContinuousQueryEntry>> iter = pendingEvts.entrySet().iterator();
+
+ entries = new ArrayList<>();
+
+ if (pendingEvts.size() >= MAX_BUFF_SIZE) {
+ if (log.isDebugEnabled()) {
+ log.debug("Pending events reached max of buffer size [lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() +
+ ", pendingEvts=" + pendingEvts + ']');
+ }
+
+ LT.warn(log, "Pending events reached max of buffer size [cache=" + cctx.name() +
+ ", bufSize=" + MAX_BUFF_SIZE +
+ ", partId=" + entry.partition() + ']');
+
+ for (int i = 0; i < MAX_BUFF_SIZE - (MAX_BUFF_SIZE / 10); i++) {
+ Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
+
+ if (e.getValue() != HOLE && !e.getValue().isFiltered())
+ entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
+
+ lastFiredEvt = e.getKey();
+
+ iter.remove();
+ }
+ }
+ else {
+ while (iter.hasNext()) {
+ Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
+
+ CacheContinuousQueryEntry pending = e.getValue();
+
+ long filtered = pending.filteredCount();
+
+ boolean fire = e.getKey() == lastFiredEvt + 1;;
+
+ if (!fire && filtered > 0)
+ fire = e.getKey() - filtered <= lastFiredEvt + 1;
+
+ if (fire) {
+ lastFiredEvt = e.getKey();
+
+ if (e.getValue() != HOLE && !e.getValue().isFiltered())
+ entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
+
+ iter.remove();
+ }
+ else
+ break;
+ }
+ }
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("Will send to listener the following events [entries=" + entries +
+ ", lastFiredEvt=" + lastFiredEvt +
+ ", curTop=" + curTop +
+ ", entUpdCnt=" + entry.updateCounter() +
+ ", partId=" + entry.partition() +
+ ", pendingEvts=" + pendingEvts + ']');
+ }
+
+ return entries;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/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 da951f2..a72dcd6 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
@@ -75,7 +75,6 @@ import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -872,10 +871,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
sendNotification(nodeId, routineId, null, toSnd, orderedTopic, true, null);
}
else {
- LocalRoutineInfo localRoutineInfo = locInfos.get(routineId);
+ LocalRoutineInfo locRoutineInfo = locInfos.get(routineId);
- if (localRoutineInfo != null)
- localRoutineInfo.handler().notifyCallback(nodeId, routineId, objs, ctx);
+ if (locRoutineInfo != null)
+ locRoutineInfo.handler().notifyCallback(nodeId, routineId, objs, ctx);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/7bf63c0e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBufferTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBufferTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBufferTest.java
index bc32e00..4710593 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBufferTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBufferTest.java
@@ -66,14 +66,14 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
for (int i = 0; i < 10; i++) {
int cnt = rnd.nextInt(10_000) + 1;
- testBuffer(rnd, new CacheContinuousQueryEventBuffer(), cnt, 1, 0.5f, threads);
- testBuffer(rnd, new CacheContinuousQueryEventBuffer(), cnt, 1, 0.9f, threads);
- testBuffer(rnd, new CacheContinuousQueryEventBuffer(), cnt, 1, 0.99f, threads);
- testBuffer(rnd, new CacheContinuousQueryEventBuffer(), cnt, 1, 0.01f, threads);
- testBuffer(rnd, new CacheContinuousQueryEventBuffer(), cnt, 1, 0.f, threads);
+ testBuffer(rnd, new CacheContinuousQueryEventBuffer(0), cnt, 1, 0.5f, threads);
+ testBuffer(rnd, new CacheContinuousQueryEventBuffer(0), cnt, 1, 0.9f, threads);
+ testBuffer(rnd, new CacheContinuousQueryEventBuffer(0), cnt, 1, 0.99f, threads);
+ testBuffer(rnd, new CacheContinuousQueryEventBuffer(0), cnt, 1, 0.01f, threads);
+ testBuffer(rnd, new CacheContinuousQueryEventBuffer(0), cnt, 1, 0.f, threads);
}
- CacheContinuousQueryEventBuffer b = new CacheContinuousQueryEventBuffer();
+ CacheContinuousQueryEventBuffer b = new CacheContinuousQueryEventBuffer(0);
long cntr = 1;
@@ -106,32 +106,31 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
{
List<CacheContinuousQueryEntry> expEntries = new ArrayList<>();
- List<Object> entries = new ArrayList<>();
+ List<CacheContinuousQueryEntry> entries = new ArrayList<>();
long filtered = b.currentFiltered();
for (int i = 0; i < cnt; i++) {
- if (rnd.nextFloat() < filterRatio) {
- entries.add(cntr);
+ CacheContinuousQueryEntry entry = new CacheContinuousQueryEntry(
+ 0,
+ EventType.CREATED,
+ null,
+ null,
+ null,
+ false,
+ 0,
+ cntr,
+ null);
+
- cntr++;
+ entries.add(entry);
+
+ if (rnd.nextFloat() < filterRatio) {
+ entry.markFiltered();
filtered++;
}
else {
- CacheContinuousQueryEntry entry = new CacheContinuousQueryEntry(
- 0,
- EventType.CREATED,
- null,
- null,
- null,
- false,
- 0,
- cntr,
- null);
-
- entries.add(entry);
-
CacheContinuousQueryEntry expEntry = new CacheContinuousQueryEntry(
0,
EventType.CREATED,
@@ -145,12 +144,12 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
expEntry.filteredCount(filtered);
- cntr++;
-
expEntries.add(expEntry);
filtered = 0;
}
+
+ cntr++;
}
Collections.shuffle(entries, rnd);
@@ -161,12 +160,7 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
for (int i = 0; i < entries.size(); i++) {
Object o = entries.get(i);
- Object res;
-
- if (o instanceof Long)
- res = b.processFiltered((Long)o);
- else
- res = b.processEntry((CacheContinuousQueryEntry)o);
+ Object res = b.processEntry((CacheContinuousQueryEntry)o, false);
if (res != null) {
if (res instanceof CacheContinuousQueryEntry)
@@ -179,7 +173,7 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
else {
final CyclicBarrier barrier = new CyclicBarrier(threads);
- final ConcurrentLinkedQueue<Object> q = new ConcurrentLinkedQueue<>(entries);
+ final ConcurrentLinkedQueue<CacheContinuousQueryEntry> q = new ConcurrentLinkedQueue<>(entries);
final ConcurrentSkipListMap<Long, CacheContinuousQueryEntry> act0 = new ConcurrentSkipListMap<>();
@@ -190,12 +184,7 @@ public class CacheContinuousQueryEventBufferTest extends GridCommonAbstractTest
Object o;
while ((o = q.poll()) != null) {
- Object res;
-
- if (o instanceof Long)
- res = b.processFiltered((Long)o);
- else
- res = b.processEntry((CacheContinuousQueryEntry)o);
+ Object res = b.processEntry((CacheContinuousQueryEntry)o, false);
if (res != null) {
if (res instanceof CacheContinuousQueryEntry)
[06/18] ignite git commit: Merge remote-tracking branch
'origin/ignite-5075-cc' into ignite-5075-cc
Posted by sb...@apache.org.
Merge remote-tracking branch 'origin/ignite-5075-cc' into ignite-5075-cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b65a63d8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b65a63d8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b65a63d8
Branch: refs/heads/ignite-5075-cc-debug
Commit: b65a63d82d2ebf29d84b4bf75123e0ef14590322
Parents: 7872545 31f3299
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 11:51:10 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 11:51:10 2017 +0300
----------------------------------------------------------------------
----------------------------------------------------------------------
[02/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/460ba11a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/460ba11a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/460ba11a
Branch: refs/heads/ignite-5075-cc-debug
Commit: 460ba11adfdc49f98ba8c87c7e1390f17f720d7d
Parents: 8adfe7d
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 09:27:42 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 09:50:49 2017 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheMapEntry.java | 7 ++--
.../CacheContinuousQueryEventBuffer.java | 36 +++++++++++++++++++-
.../continuous/CacheContinuousQueryHandler.java | 12 +++++--
...ContinuousQueryFailoverAbstractSelfTest.java | 14 +++++---
4 files changed, 59 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/460ba11a/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 569d638..8a7cb95 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
@@ -1845,9 +1845,12 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
CacheObject evtOldVal = cctx.unwrapTemporary(oldVal);
if (primary)
- TestDebugLog.addEntryMessage(partition(), evtVal.value(cctx.cacheObjectContext(), false), "primary notify cntr=" + c.updateRes.updateCounter() + " k=" + key.value(null, false));
+ TestDebugLog.addEntryMessage(partition(), topVer, // evtVal.value(cctx.cacheObjectContext(), false)
+ "primary notify cntr=" + c.updateRes.updateCounter() +
+ " k=" + key.value(null, false));
else
- TestDebugLog.addEntryMessage(key.value(null, false), evtVal.value(cctx.cacheObjectContext(), false), "backup notify cntr=" + c.updateRes.updateCounter() + " k=" + key.value(null, false));
+ TestDebugLog.addEntryMessage(key.value(null, false), topVer,
+ "backup notify cntr=" + c.updateRes.updateCounter() + " k=" + key.value(null, false));
cctx.continuousQueries().onEntryUpdated(lsnrs,
key,
http://git-wip-us.apache.org/repos/asf/ignite/blob/460ba11a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index a308e39..b1bc7b0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.spi.communication.tcp.TestDebugLog;
import org.jetbrains.annotations.Nullable;
/**
@@ -41,6 +42,17 @@ public class CacheContinuousQueryEventBuffer {
/** */
private ConcurrentSkipListMap<Long, Object> pending = new ConcurrentSkipListMap<>();
+ /** */
+ private final int part;
+
+ public CacheContinuousQueryEventBuffer() {
+ part = 0;
+ }
+
+ public CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
/**
* @return Initial partition counter.
*/
@@ -88,6 +100,12 @@ public class CacheContinuousQueryEventBuffer {
if (batch == null || cntr < batch.startCntr) {
assert entry != null : cntr;
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer rcd small start=" + batch.startCntr +
+ " cntr=" + cntr +
+ " topVer=" + ((CacheContinuousQueryEntry)entry).topologyVersion());
+
return entry;
}
@@ -95,8 +113,15 @@ public class CacheContinuousQueryEventBuffer {
if (cntr <= batch.endCntr)
res = batch.processEvent0(null, cntr, entry);
- else
+ else {
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer add pending start=" + batch.startCntr +
+ " cntr=" + cntr +
+ " topVer=" + ((CacheContinuousQueryEntry)entry).topologyVersion());
+
pending.put(cntr, entry);
+ }
Batch batch0 = curBatch.get();
@@ -128,6 +153,8 @@ public class CacheContinuousQueryEventBuffer {
if (curCntr == -1)
return null;
+ TestDebugLog.addEntryMessage(part, curCntr, "created batch");
+
batch = new Batch(curCntr + 1, 0L, new Object[BUF_SIZE]);
if (curBatch.compareAndSet(null, batch))
@@ -205,6 +232,13 @@ public class CacheContinuousQueryEventBuffer {
int pos = (int)(cntr - startCntr);
synchronized (this) {
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer process start=" + startCntr +
+ ", lastProc=" + lastProc +
+ " pos=" + pos +
+ " topVer=" + ((CacheContinuousQueryEntry)evt).topologyVersion());
+
evts[pos] = evt;
int next = lastProc + 1;
http://git-wip-us.apache.org/repos/asf/ignite/blob/460ba11a/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 01fe5e7..3982815 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
@@ -775,8 +775,14 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
Collection<CacheContinuousQueryEntry> backupQueue0 = backupQueue;
- if (backupQueue0 != null)
+ if (backupQueue0 != null) {
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "add backup " +
+ " topVer=" + entry.topologyVersion());
+
backupQueue0.add(entry.forBackupQueue());
+ }
}
return notify;
@@ -948,7 +954,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (buf == null) {
final int part = e.partition();
- buf = new CacheContinuousQueryEventBuffer() {
+ buf = new CacheContinuousQueryEventBuffer(part) {
@Override protected long currentPartitionCounter() {
GridDhtLocalPartition locPart = cctx.topology().localPartition(part, null, false);
@@ -1182,7 +1188,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
boolean fire = e.getKey() == lastFiredEvt + 1;;
if (!fire && filtered > 0)
- fire = e.getKey() - filtered <= lastFiredEvt;
+ fire = e.getKey() - filtered <= lastFiredEvt + 1;
if (fire) {
TestDebugLog.addEntryMessage(entry.partition(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/460ba11a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index 050af5d..447a203 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -1220,15 +1220,15 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
}
}
- if (!lostAllow && lostEvts.size() > 100) {
+ if (!lostAllow && lostEvts.size() > 50) {
log.error("Lost event cnt: " + lostEvts.size());
for (T3<Object, Object, Object> e : lostEvts) {
- log.error("Lost event: " + ignite(0).affinity(DEFAULT_CACHE_NAME).partition(e.get1()) + " " + e);
+ log.error("Lost event: " + ignite(4).affinity(DEFAULT_CACHE_NAME).partition(e.get1()) + " " + e);
- TestDebugLog.addEntryMessage(ignite(0).affinity(DEFAULT_CACHE_NAME).partition(e.get1()), e.get2(), "lost event " + e.get1() + " " + e.get2());
+ TestDebugLog.addEntryMessage(ignite(4).affinity(DEFAULT_CACHE_NAME).partition(e.get1()), e.get2(), "lost event " + e.get1() + " " + e.get2());
- TestDebugLog.printKeyMessages(true, ignite(0).affinity(DEFAULT_CACHE_NAME).partition(e.get1()));
+ TestDebugLog.printKeyMessages(true, ignite(4).affinity(DEFAULT_CACHE_NAME).partition(e.get1()));
System.exit(1);
}
@@ -1598,6 +1598,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
for (int i = 0; i < 10; i++) {
final int idx = i % (SRV_NODES - 1);
+ TestDebugLog.addMessage("Stop node: " + idx);
+
log.info("Stop node: " + idx);
stopGrid(idx);
@@ -1609,6 +1611,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
for (int j = 0; j < aff.partitions(); j++) {
Integer oldVal = (Integer)qryClnCache.get(j);
+ TestDebugLog.addEntryMessage(ignite(4).affinity(DEFAULT_CACHE_NAME).partition(j), i, "do put " + j + " " + i);
+
qryClnCache.put(j, i);
afterRestEvts.add(new T3<>((Object)j, (Object)i, (Object)oldVal));
@@ -1616,6 +1620,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
checkEvents(new ArrayList<>(afterRestEvts), lsnr, false);
+ TestDebugLog.addMessage("Start node: " + idx);
+
log.info("Start node: " + idx);
startGrid(idx);
[05/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/78725456
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/78725456
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/78725456
Branch: refs/heads/ignite-5075-cc-debug
Commit: 78725456c3f19fddab3a8cfce4c700c1f2a77b2f
Parents: 7bf63c0
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 11:47:09 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 11:49:33 2017 +0300
----------------------------------------------------------------------
.../continuous/CacheContinuousQueryEventBuffer.java | 14 +++++++-------
.../query/continuous/CacheContinuousQueryHandler.java | 6 ++++++
...cheContinuousQueryAsyncFailoverAtomicSelfTest.java | 1 -
.../IgniteCacheContinuousQueryBackupQueueTest.java | 13 +++++++++----
4 files changed, 22 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/78725456/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 949ea67..f0640b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -41,13 +41,6 @@ public class CacheContinuousQueryEventBuffer {
/** */
protected final int part;
- /**
- * @param part Partition number.
- */
- CacheContinuousQueryEventBuffer(int part) {
- this.part = part;
- }
-
/** */
private AtomicReference<Batch> curBatch = new AtomicReference<>();
@@ -58,6 +51,13 @@ public class CacheContinuousQueryEventBuffer {
private ConcurrentSkipListMap<Long, CacheContinuousQueryEntry> pending = new ConcurrentSkipListMap<>();
/**
+ * @param part Partition number.
+ */
+ CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
+ /**
* @param updateCntr Acknowledged counter.
*/
void cleanupBackupQueue(Long updateCntr) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/78725456/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 540f871..9866e7b 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
@@ -954,6 +954,12 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** {@inheritDoc} */
@Override public void onNodeLeft() {
nodeLeft = true;
+
+ for (Map.Entry<Integer, CacheContinuousQueryEventBuffer> bufE : entryBufs.entrySet()) {
+ CacheContinuousQueryEventBuffer buf = bufE.getValue();
+
+ buf.resetBackupQueue();
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/78725456/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicSelfTest.java
index 3cab9e0..d505d19 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicSelfTest.java
@@ -25,7 +25,6 @@ import org.apache.ignite.cache.CacheMode;
*/
public class CacheContinuousQueryAsyncFailoverAtomicSelfTest
extends CacheContinuousQueryFailoverAbstractSelfTest {
-
/** {@inheritDoc} */
@Override protected CacheMode cacheMode() {
return CacheMode.PARTITIONED;
http://git-wip-us.apache.org/repos/asf/ignite/blob/78725456/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
index 26c7d41..85d68d3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryBackupQueueTest.java
@@ -262,11 +262,16 @@ public class IgniteCacheContinuousQueryBackupQueueTest extends GridCommonAbstrac
GridContinuousHandler hnd = GridTestUtils.getFieldValue(info, "hnd");
if (hnd.isQuery() && hnd.cacheName().equals(CACHE_NAME)) {
- Collection<Object> q = GridTestUtils.getFieldValue(hnd,
- CacheContinuousQueryHandler.class, "backupQueue");
+ Map<Integer, CacheContinuousQueryEventBuffer> map = GridTestUtils.getFieldValue(hnd,
+ CacheContinuousQueryHandler.class, "entryBufs");
- if (q != null)
- backupQueues.add(q);
+ for (CacheContinuousQueryEventBuffer buf : map.values()) {
+ Collection<Object> q = GridTestUtils.getFieldValue(buf,
+ CacheContinuousQueryEventBuffer.class, "backupQ");
+
+ if (q != null)
+ backupQueues.add(q);
+ }
}
}
[17/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9cc1daec
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9cc1daec
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9cc1daec
Branch: refs/heads/ignite-5075-cc-debug
Commit: 9cc1daeccd99223eaed66e64ddabe860c9eb5075
Parents: d2942a5 ff0a2dd
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 16:56:14 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 16:56:14 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryEventBuffer.java | 29 ++++++++++++--------
.../CacheContinuousQueryPartitionRecovery.java | 22 ++++++++-------
2 files changed, 29 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/9cc1daec/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 264a6f1,f496c8c..d65731b
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@@ -28,7 -28,6 +28,8 @@@ import java.util.concurrent.ConcurrentS
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.spi.communication.tcp.TestDebugLog;
++import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.jetbrains.annotations.Nullable;
/**
@@@ -99,17 -96,7 +98,15 @@@ public class CacheContinuousQueryEventB
else
ret = entries;
+ if (ret != null) {
+ for (CacheContinuousQueryEntry e : ret)
+ TestDebugLog.addEntryMessage(part,
+ e.updateCounter(),
+ "filtered1 " + e.filteredCount() +
+ " reset backup");
+ }
- else
- TestDebugLog.addEntryMessage(part, part, "no backup");
+
- return entries;
+ return ret;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/9cc1daec/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
index e031428,59252d2..e72f7b6
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
@@@ -277,14 -240,11 +277,16 @@@ class CacheContinuousQueryPartitionReco
iter.remove();
}
- else if (!pending.isFiltered()) {
- skip = true;
-
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "stop process last=" + lastFiredEvt + " cntr=" + e.getKey() + " topVer=" + e.getValue().topologyVersion() + " f=" + pending.filteredCount());
-
- break;
+ else {
+ if (pending.isFiltered())
+ skippedFiltered = true;
- else
++ else {
++ TestDebugLog.addEntryMessage(entry.partition(),
++ entry.updateCounter(),
++ "stop process last=" + lastFiredEvt + " cntr=" + e.getKey() + " topVer=" + e.getValue().topologyVersion() + " f=" + pending.filteredCount());
++
+ break;
++ }
}
}
[09/18] ignite git commit: Merge branch 'ignite-5075-cc' into
ignite-5075-cc-debug
Posted by sb...@apache.org.
Merge branch 'ignite-5075-cc' into ignite-5075-cc-debug
# Conflicts:
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8ee752d8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8ee752d8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8ee752d8
Branch: refs/heads/ignite-5075-cc-debug
Commit: 8ee752d87daa3b9337f01dce3008f04c0bcc2238
Parents: 460ba11 2c68adb
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:21:02 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:21:02 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryAcknowledgeBuffer.java | 120 ++++++
.../CacheContinuousQueryDeployableObject.java | 110 ++++++
.../CacheContinuousQueryEventBuffer.java | 93 ++++-
.../continuous/CacheContinuousQueryHandler.java | 388 +++++--------------
.../CacheContinuousQueryHandlerV2.java | 6 +-
.../CacheContinuousQueryPartitionRecovery.java | 252 ++++++++++++
.../continuous/GridContinuousProcessor.java | 7 +-
...tinuousQueryAsyncFailoverAtomicSelfTest.java | 1 -
.../CacheContinuousQueryEventBufferTest.java | 65 ++--
...ContinuousQueryFailoverAbstractSelfTest.java | 79 ++--
...niteCacheContinuousQueryBackupQueueTest.java | 13 +-
11 files changed, 732 insertions(+), 402 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8ee752d8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index b1bc7b0,f0640b1..6295b0b
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@@ -40,19 -45,47 +46,58 @@@ public class CacheContinuousQueryEventB
private AtomicReference<Batch> curBatch = new AtomicReference<>();
/** */
- private ConcurrentSkipListMap<Long, Object> pending = new ConcurrentSkipListMap<>();
+ private ConcurrentLinkedDeque<CacheContinuousQueryEntry> backupQ = new ConcurrentLinkedDeque<>();
+
+ /** */
+ private ConcurrentSkipListMap<Long, CacheContinuousQueryEntry> pending = new ConcurrentSkipListMap<>();
+
+ /**
+ * @param part Partition number.
+ */
+ CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
+ /**
+ * @param updateCntr Acknowledged counter.
+ */
+ void cleanupBackupQueue(Long updateCntr) {
+ Iterator<CacheContinuousQueryEntry> it = backupQ.iterator();
+
+ while (it.hasNext()) {
+ CacheContinuousQueryEntry backupEntry = it.next();
+
+ if (backupEntry.updateCounter() <= updateCntr)
+ it.remove();
+ }
+ }
+
+ /**
+ * @return Backup entries.
+ */
+ @Nullable Collection<CacheContinuousQueryEntry> resetBackupQueue() {
+ if (!backupQ.isEmpty()) {
+ ConcurrentLinkedDeque<CacheContinuousQueryEntry> ret = this.backupQ;
+
+ backupQ = new ConcurrentLinkedDeque<>();
+
+ return ret;
+ }
+
+ return null;
+ }
+ /** */
+ private final int part;
+
+ public CacheContinuousQueryEventBuffer() {
+ part = 0;
+ }
+
+ public CacheContinuousQueryEventBuffer(int part) {
+ this.part = part;
+ }
+
/**
* @return Initial partition counter.
*/
@@@ -100,28 -127,18 +139,31 @@@
if (batch == null || cntr < batch.startCntr) {
assert entry != null : cntr;
+ if (backup)
+ backupQ.add(entry);
+
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer rcd small start=" + batch.startCntr +
+ " cntr=" + cntr +
++ ", backup=" + backup +
+ " topVer=" + ((CacheContinuousQueryEntry)entry).topologyVersion());
+
return entry;
}
Object res = null;
if (cntr <= batch.endCntr)
- res = batch.processEvent0(null, cntr, entry);
+ res = batch.processEvent0(null, cntr, entry, backup);
- else
+ else {
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer add pending start=" + batch.startCntr +
+ " cntr=" + cntr +
+ " topVer=" + ((CacheContinuousQueryEntry)entry).topologyVersion());
-
pending.put(cntr, entry);
+ }
Batch batch0 = curBatch.get();
http://git-wip-us.apache.org/repos/asf/ignite/blob/8ee752d8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8ee752d8/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
[08/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2c68adb8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2c68adb8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2c68adb8
Branch: refs/heads/ignite-5075-cc-debug
Commit: 2c68adb8dfb206640c86c24cebc899cf2c18f690
Parents: 52b716b
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:11:51 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:11:51 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 2 --
.../CacheContinuousQueryFailoverAbstractSelfTest.java | 6 +++---
2 files changed, 3 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2c68adb8/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 9866e7b..11ed922 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
@@ -458,8 +458,6 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
for (CacheContinuousQueryEntry e : backupQueue) {
if (!e.isFiltered())
prepareEntry(cctx, nodeId, e);
-
- e.topologyVersion(topVer);
}
ctx.continuous().addBackupNotification(nodeId, routineId, backupQueue, topic);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2c68adb8/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index 3108edc..2fb7fcb 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -30,6 +30,7 @@ import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ThreadLocalRandom;
@@ -2421,7 +2422,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
private ConcurrentHashMap<Object, CacheEntryEvent<?, ?>> evts = new ConcurrentHashMap<>();
/** */
- private List<CacheEntryEvent<?, ?>> allEvts;
+ private final CopyOnWriteArrayList<CacheEntryEvent<?, ?>> allEvts;
/** */
@LoggerResource
@@ -2431,8 +2432,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
* @param saveAll Save all events flag.
*/
CacheEventListener1(boolean saveAll) {
- if (saveAll)
- allEvts = new ArrayList<>();
+ allEvts = saveAll ? new CopyOnWriteArrayList<CacheEntryEvent<?, ?>>() : null;
}
/** {@inheritDoc} */
[11/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/0305930e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/0305930e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/0305930e
Branch: refs/heads/ignite-5075-cc-debug
Commit: 0305930e806d9394b5fe08af8a3be82048d46ca3
Parents: 2c68adb
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:23:03 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:23:03 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryHandler.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/0305930e/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 11ed922..98eaecb 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
@@ -891,6 +891,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx, e.partition());
+ e.markBackup();
+
buf.processEntry(e.forBackupQueue(), true);
}
[18/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/58919c5b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/58919c5b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/58919c5b
Branch: refs/heads/ignite-5075-cc-debug
Commit: 58919c5b5e8f89b25d8fde9d68eaa5184ab6c946
Parents: 9cc1dae
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 16:58:03 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 16:58:03 2017 +0300
----------------------------------------------------------------------
.../cache/query/continuous/CacheContinuousQueryEventBuffer.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/58919c5b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index d65731b..5346144 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -102,7 +102,7 @@ public class CacheContinuousQueryEventBuffer {
for (CacheContinuousQueryEntry e : ret)
TestDebugLog.addEntryMessage(part,
e.updateCounter(),
- "filtered1 " + e.filteredCount() +
+ "filtered " + e.filteredCount() +
" reset backup");
}
[10/18] ignite git commit: Merge branch 'ignite-5075-cc' into
ignite-5075-cc-debug
Posted by sb...@apache.org.
Merge branch 'ignite-5075-cc' into ignite-5075-cc-debug
# Conflicts:
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
# modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/529dec10
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/529dec10
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/529dec10
Branch: refs/heads/ignite-5075-cc-debug
Commit: 529dec1018290fb9d05890150f7dd7ca410902e5
Parents: 8ee752d
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 13:21:12 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 13:21:12 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryEventBuffer.java | 11 -
.../continuous/CacheContinuousQueryHandler.java | 259 +------------------
.../CacheContinuousQueryPartitionRecovery.java | 48 +++-
3 files changed, 46 insertions(+), 272 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/529dec10/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 6295b0b..59b92eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -87,17 +87,6 @@ public class CacheContinuousQueryEventBuffer {
return null;
}
- /** */
- private final int part;
-
- public CacheContinuousQueryEventBuffer() {
- part = 0;
- }
-
- public CacheContinuousQueryEventBuffer(int part) {
- this.part = part;
- }
-
/**
* @return Initial partition counter.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/529dec10/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 2f88827..72fdd83 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
@@ -947,264 +947,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
buf = oldBuf;
}
- return buf.processEntry(e);
- }
-
- /**
- *
- */
- private static class PartitionRecovery {
- /** Event which means hole in sequence. */
- private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
-
- /** */
- private final static int MAX_BUFF_SIZE = LSNR_MAX_BUF_SIZE;
-
- /** */
- private IgniteLogger log;
-
- /** */
- private long lastFiredEvt;
-
- /** */
- private AffinityTopologyVersion curTop = AffinityTopologyVersion.NONE;
-
- /** */
- private final Map<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
-
- /**
- * @param log Logger.
- * @param topVer Topology version.
- * @param initCntr Update counters.
- */
- PartitionRecovery(IgniteLogger log, AffinityTopologyVersion topVer, @Nullable Long initCntr) {
- this.log = log;
-
- if (initCntr != null) {
- assert topVer.topologyVersion() > 0 : topVer;
-
- this.lastFiredEvt = initCntr;
-
- curTop = topVer;
- }
- }
-
- /**
- * Resets cached topology.
- */
- void resetTopologyCache() {
- curTop = AffinityTopologyVersion.NONE;
- }
-
- /**
- * Add continuous entry.
- *
- * @param cctx Cache context.
- * @param cache Cache.
- * @param entry Cache continuous query entry.
- * @return Collection entries which will be fired. This collection should contains only non-filtered events.
- */
- <K, V> Collection<CacheEntryEvent<? extends K, ? extends V>> collectEntries(
- CacheContinuousQueryEntry entry,
- GridCacheContext cctx,
- IgniteCache cache
- ) {
- assert entry != null;
-
- if (entry.topologyVersion() == null) { // Possible if entry is sent from old node.
- assert entry.updateCounter() == 0L : entry;
-
- return F.<CacheEntryEvent<? extends K, ? extends V>>
- asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
- }
-
- List<CacheEntryEvent<? extends K, ? extends V>> entries;
-
- synchronized (pendingEvts) {
- if (log.isDebugEnabled()) {
- log.debug("Handling event [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- // Received first event.
- if (curTop == AffinityTopologyVersion.NONE) {
- lastFiredEvt = entry.updateCounter();
-
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "collect first cntr=" + entry.updateCounter() + " topVer=" + entry.topologyVersion());
-
- curTop = entry.topologyVersion();
-
- if (log.isDebugEnabled()) {
- log.debug("First event [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() + ']');
- }
-
- return !entry.isFiltered() ?
- F.<CacheEntryEvent<? extends K, ? extends V>>
- asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry)) :
- Collections.<CacheEntryEvent<? extends K, ? extends V>>emptyList();
- }
-
- if (curTop.compareTo(entry.topologyVersion()) < 0) {
- if (entry.updateCounter() == 1L && !entry.isBackup()) {
- entries = new ArrayList<>(pendingEvts.size());
-
- for (CacheContinuousQueryEntry evt : pendingEvts.values()) {
- if (evt != HOLE && !evt.isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, evt));
- }
-
- pendingEvts.clear();
-
- curTop = entry.topologyVersion();
-
- lastFiredEvt = entry.updateCounter();
-
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "collect for lost topVer cntr=" + entry.updateCounter() + " topVer=" + entry.topologyVersion());
-
- if (!entry.isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
-
- if (log.isDebugEnabled())
- log.debug("Partition was lost [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
-
- return entries;
- }
-
- curTop = entry.topologyVersion();
- }
-
- // Check duplicate.
- if (entry.updateCounter() > lastFiredEvt) {
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "add event last=" + lastFiredEvt +
- " cntr=" + entry.updateCounter() +
- " key=" + (entry.isFiltered() ? "filtered" : entry.key().value(cctx.cacheObjectContext(), false)) +
- " val=" + (entry.isFiltered() ? "filtered" : entry.value().value(cctx.cacheObjectContext(), false)) +
- " topVer=" + entry.topologyVersion());
-
- pendingEvts.put(entry.updateCounter(), entry);
- }
- else {
- if (log.isDebugEnabled())
- log.debug("Skip duplicate continuous query message: " + entry);
-
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "skip duplicate last=" + lastFiredEvt +
- " cntr=" + entry.updateCounter() +
- " key=" + (entry.isFiltered() ? "filtered" : entry.key().value(cctx.cacheObjectContext(), false)) +
- " val=" + (entry.isFiltered() ? "filtered" : entry.value().value(cctx.cacheObjectContext(), false)) +
- " topVer=" + entry.topologyVersion());
-
- return Collections.emptyList();
- }
-
- if (pendingEvts.isEmpty()) {
- if (log.isDebugEnabled()) {
- log.debug("Nothing sent to listener [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() + ']');
- }
-
- return Collections.emptyList();
- }
-
- Iterator<Map.Entry<Long, CacheContinuousQueryEntry>> iter = pendingEvts.entrySet().iterator();
-
- entries = new ArrayList<>();
-
- if (pendingEvts.size() >= MAX_BUFF_SIZE) {
- if (log.isDebugEnabled()) {
- log.debug("Pending events reached max of buffer size [lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- LT.warn(log, "Pending events reached max of buffer size [cache=" + cctx.name() +
- ", bufSize=" + MAX_BUFF_SIZE +
- ", partId=" + entry.partition() + ']');
-
- for (int i = 0; i < MAX_BUFF_SIZE - (MAX_BUFF_SIZE / 10); i++) {
- Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
-
- if (e.getValue() != HOLE && !e.getValue().isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
-
- lastFiredEvt = e.getKey();
-
- iter.remove();
- }
- }
- else {
- while (iter.hasNext()) {
- Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
-
- CacheContinuousQueryEntry pending = e.getValue();
-
- long filtered = pending.filteredCount();
-
- boolean fire = e.getKey() == lastFiredEvt + 1;;
-
- if (!fire && filtered > 0)
- fire = e.getKey() - filtered <= lastFiredEvt + 1;
-
- if (fire) {
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "process last=" + lastFiredEvt +
- " cntr=" + e.getKey() +
- " key=" + (pending.isFiltered() ? "filtered" : pending.key().value(cctx.cacheObjectContext(), false)) +
- " val=" + (pending.isFiltered() ? "filtered" : pending.value().value(cctx.cacheObjectContext(), false)) +
- " topVer=" + e.getValue().topologyVersion() +
- " f=" + pending.filteredCount());
-
- lastFiredEvt = e.getKey();
-
- if (e.getValue() != HOLE && !e.getValue().isFiltered())
- entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, e.getValue()));
-
- iter.remove();
- }
- else {
- TestDebugLog.addEntryMessage(entry.partition(),
- entry.updateCounter(),
- "stop process last=" + lastFiredEvt + " cntr=" + e.getKey() + " topVer=" + e.getValue().topologyVersion() + " f=" + pending.filteredCount());
-
- break;
- }
- }
- }
- }
-
- if (log.isDebugEnabled()) {
- log.debug("Will send to listener the following events [entries=" + entries +
- ", lastFiredEvt=" + lastFiredEvt +
- ", curTop=" + curTop +
- ", entUpdCnt=" + entry.updateCounter() +
- ", partId=" + entry.partition() +
- ", pendingEvts=" + pendingEvts + ']');
- }
-
- return entries;
- }
+ return buf;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/529dec10/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
index 534ce9c..12eaa20 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
@@ -31,8 +31,11 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.LT;
+import org.apache.ignite.spi.communication.tcp.TestDebugLog;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.LSNR_MAX_BUF_SIZE;
+
/**
*
*/
@@ -41,7 +44,7 @@ class CacheContinuousQueryPartitionRecovery {
private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
/** */
- private final static int MAX_BUFF_SIZE = CacheContinuousQueryHandler.LSNR_MAX_BUF_SIZE;
+ private final static int MAX_BUFF_SIZE = LSNR_MAX_BUF_SIZE;
/** */
private IgniteLogger log;
@@ -116,6 +119,10 @@ class CacheContinuousQueryPartitionRecovery {
if (curTop == AffinityTopologyVersion.NONE) {
lastFiredEvt = entry.updateCounter();
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "collect first cntr=" + entry.updateCounter() + " topVer=" + entry.topologyVersion());
+
curTop = entry.topologyVersion();
if (log.isDebugEnabled()) {
@@ -146,6 +153,10 @@ class CacheContinuousQueryPartitionRecovery {
lastFiredEvt = entry.updateCounter();
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "collect for lost topVer cntr=" + entry.updateCounter() + " topVer=" + entry.topologyVersion());
+
if (!entry.isFiltered())
entries.add(new CacheContinuousQueryEvent<K, V>(cache, cctx, entry));
@@ -163,12 +174,29 @@ class CacheContinuousQueryPartitionRecovery {
}
// Check duplicate.
- if (entry.updateCounter() > lastFiredEvt)
+ if (entry.updateCounter() > lastFiredEvt) {
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "add event last=" + lastFiredEvt +
+ " cntr=" + entry.updateCounter() +
+ " key=" + (entry.isFiltered() ? "filtered" : entry.key().value(cctx.cacheObjectContext(), false)) +
+ " val=" + (entry.isFiltered() ? "filtered" : entry.value().value(cctx.cacheObjectContext(), false)) +
+ " topVer=" + entry.topologyVersion());
+
pendingEvts.put(entry.updateCounter(), entry);
+ }
else {
if (log.isDebugEnabled())
log.debug("Skip duplicate continuous query message: " + entry);
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "skip duplicate last=" + lastFiredEvt +
+ " cntr=" + entry.updateCounter() +
+ " key=" + (entry.isFiltered() ? "filtered" : entry.key().value(cctx.cacheObjectContext(), false)) +
+ " val=" + (entry.isFiltered() ? "filtered" : entry.value().value(cctx.cacheObjectContext(), false)) +
+ " topVer=" + entry.topologyVersion());
+
return Collections.emptyList();
}
@@ -225,6 +253,15 @@ class CacheContinuousQueryPartitionRecovery {
fire = e.getKey() - filtered <= lastFiredEvt + 1;
if (fire) {
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "process last=" + lastFiredEvt +
+ " cntr=" + e.getKey() +
+ " key=" + (pending.isFiltered() ? "filtered" : pending.key().value(cctx.cacheObjectContext(), false)) +
+ " val=" + (pending.isFiltered() ? "filtered" : pending.value().value(cctx.cacheObjectContext(), false)) +
+ " topVer=" + e.getValue().topologyVersion() +
+ " f=" + pending.filteredCount());
+
lastFiredEvt = e.getKey();
if (e.getValue() != HOLE && !e.getValue().isFiltered())
@@ -232,8 +269,13 @@ class CacheContinuousQueryPartitionRecovery {
iter.remove();
}
- else
+ else {
+ TestDebugLog.addEntryMessage(entry.partition(),
+ entry.updateCounter(),
+ "stop process last=" + lastFiredEvt + " cntr=" + e.getKey() + " topVer=" + e.getValue().topologyVersion() + " f=" + pending.filteredCount());
+
break;
+ }
}
}
}
[15/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d2942a58
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d2942a58
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d2942a58
Branch: refs/heads/ignite-5075-cc-debug
Commit: d2942a58971a6d3797781151283c42549f0cc8f6
Parents: 8fafde2
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 15:25:32 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 15:25:32 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryEventBuffer.java | 220 ++++++++++++++-----
.../CacheContinuousQueryPartitionRecovery.java | 19 +-
2 files changed, 184 insertions(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d2942a58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 59b92eb..264a6f1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.spi.communication.tcp.TestDebugLog;
import org.jetbrains.annotations.Nullable;
@@ -76,15 +77,39 @@ public class CacheContinuousQueryEventBuffer {
* @return Backup entries.
*/
@Nullable Collection<CacheContinuousQueryEntry> resetBackupQueue() {
+ Collection<CacheContinuousQueryEntry> ret = null;
+
+ List<CacheContinuousQueryEntry> entries = null;
+
+ Batch batch = curBatch.get();
+
+ if (batch != null)
+ entries = batch.backupFlushEntries();
+
if (!backupQ.isEmpty()) {
- ConcurrentLinkedDeque<CacheContinuousQueryEntry> ret = this.backupQ;
+ ret = this.backupQ;
+
+ if (entries != null) {
+ for (CacheContinuousQueryEntry e : entries)
+ ((ConcurrentLinkedDeque)ret).addFirst(e);
+ }
backupQ = new ConcurrentLinkedDeque<>();
+ }
+ else
+ ret = entries;
- return ret;
+ if (ret != null) {
+ for (CacheContinuousQueryEntry e : ret)
+ TestDebugLog.addEntryMessage(part,
+ e.updateCounter(),
+ "filtered1 " + e.filteredCount() +
+ " reset backup");
}
+ else
+ TestDebugLog.addEntryMessage(part, part, "no backup");
- return null;
+ return entries;
}
/**
@@ -123,7 +148,7 @@ public class CacheContinuousQueryEventBuffer {
private Object process0(long cntr, CacheContinuousQueryEntry entry, boolean backup) {
assert cntr >= 0 : cntr;
- Batch batch = initBatch();
+ Batch batch = initBatch(entry.topologyVersion());
if (batch == null || cntr < batch.startCntr) {
assert entry != null : cntr;
@@ -173,7 +198,7 @@ public class CacheContinuousQueryEventBuffer {
/**
* @return Current batch.
*/
- @Nullable private Batch initBatch() {
+ @Nullable private Batch initBatch(AffinityTopologyVersion topVer) {
Batch batch = curBatch.get();
if (batch != null)
@@ -186,7 +211,7 @@ public class CacheContinuousQueryEventBuffer {
TestDebugLog.addEntryMessage(part, curCntr, "created batch");
- batch = new Batch(curCntr + 1, 0L, new Object[BUF_SIZE]);
+ batch = new Batch(curCntr + 1, 0L, new CacheContinuousQueryEntry[BUF_SIZE], topVer);
if (curBatch.compareAndSet(null, batch))
return batch;
@@ -232,28 +257,108 @@ public class CacheContinuousQueryEventBuffer {
private int lastProc = -1;
/** */
- private final Object[] evts;
+ private final CacheContinuousQueryEntry[] entries;
+
+ /** */
+ private final AffinityTopologyVersion topVer;
/**
* @param filtered Number of filtered events before this batch.
- * @param evts Events array.
+ * @param entries Entries array.
* @param startCntr Start counter.
*/
- Batch(long startCntr, long filtered, Object[] evts) {
+ Batch(long startCntr, long filtered, CacheContinuousQueryEntry[] entries, AffinityTopologyVersion topVer) {
assert startCntr >= 0;
assert filtered >= 0;
this.startCntr = startCntr;
this.filtered = filtered;
- this.evts = evts;
+ this.entries = entries;
+ this.topVer = topVer;
endCntr = startCntr + BUF_SIZE - 1;
}
+ synchronized List<CacheContinuousQueryEntry> backupFlushEntries() {
+ List<CacheContinuousQueryEntry> res = null;
+
+ long filtered = this.filtered;
+ long cntr = startCntr;
+
+ for (int i = 0; i < entries.length; i++) {
+ CacheContinuousQueryEntry e = entries[i];
+
+ CacheContinuousQueryEntry flushEntry = null;
+
+ if (e == null) {
+ if (filtered != 0) {
+ flushEntry = filteredEntry(cntr - 1, filtered - 1);
+
+ filtered = 0;
+ }
+ }
+ else {
+ if (e.isFiltered())
+ filtered++;
+ else {
+ flushEntry = new CacheContinuousQueryEntry(e.cacheId(),
+ e.eventType(),
+ e.key(),
+ e.value(),
+ e.oldValue(),
+ e.isKeepBinary(),
+ e.partition(),
+ e.updateCounter(),
+ e.topologyVersion());
+
+ flushEntry.filteredCount(filtered);
+
+ filtered = 0;
+ }
+ }
+
+ if (flushEntry != null) {
+ if (res == null)
+ res = new ArrayList<>();
+
+ res.add(flushEntry);
+ }
+
+ cntr++;
+ }
+
+ if (filtered != 0L) {
+ if (res == null)
+ res = new ArrayList<>();
+
+ res.add(filteredEntry(cntr - 1, filtered - 1));
+ }
+
+ return res;
+ }
+
+ private CacheContinuousQueryEntry filteredEntry(long cntr, long filtered) {
+ CacheContinuousQueryEntry e = new CacheContinuousQueryEntry(0,
+ null,
+ null,
+ null,
+ null,
+ false,
+ part,
+ cntr,
+ topVer);
+
+ e.markFiltered();
+
+ e.filteredCount(filtered);
+
+ return e;
+ }
+
/**
* @param res Current result.
- * @param cntr Event counter.
- * @param evt Event.
+ * @param cntr Entry counter.
+ * @param entry Entry.
* @param backup Backup entry flag.
* @return New result.
*/
@@ -261,7 +366,7 @@ public class CacheContinuousQueryEventBuffer {
@Nullable private Object processEvent0(
@Nullable Object res,
long cntr,
- CacheContinuousQueryEntry evt,
+ CacheContinuousQueryEntry entry,
boolean backup) {
int pos = (int)(cntr - startCntr);
@@ -271,56 +376,67 @@ public class CacheContinuousQueryEventBuffer {
"buffer process start=" + startCntr +
", lastProc=" + lastProc +
" pos=" + pos +
- " topVer=" + ((CacheContinuousQueryEntry)evt).topologyVersion());
+ " topVer=" + entry.topologyVersion());
- evts[pos] = evt;
+ entries[pos] = entry;
int next = lastProc + 1;
if (next == pos) {
- for (int i = next; i < evts.length; i++) {
- Object e = evts[i];
-
- if (e != null) {
- if (e.getClass() == Long.class)
- filtered++;
- else {
- CacheContinuousQueryEntry evt0 = (CacheContinuousQueryEntry)e;
+ for (int i = next; i < entries.length; i++) {
+ CacheContinuousQueryEntry entry0 = entries[i];
+
+ if (entry0 != null) {
+ if (!entry0.isFiltered()) {
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer process res start=" + startCntr +
+ ", lastProc=" + lastProc +
+ " pos=" + pos +
+ ", filtered=" + filtered +
+ " topVer=" + entry0.topologyVersion());
+
+ entry0.filteredCount(filtered);
+
+ filtered = 0;
+
+ if (res == null) {
+ if (backup)
+ backupQ.add(entry0);
+ else
+ res = entry0;
+ }
+ else {
+ assert !backup;
- if (!evt0.isFiltered()) {
- evt0.filteredCount(filtered);
+ List<CacheContinuousQueryEntry> resList;
- filtered = 0;
+ if (res instanceof CacheContinuousQueryEntry) {
+ resList = new ArrayList<>();
- if (res == null) {
- if (backup)
- backupQ.add(evt0);
- else
- res = evt0;
+ resList.add((CacheContinuousQueryEntry)res);
}
else {
- assert !backup;
-
- List<CacheContinuousQueryEntry> resList;
-
- if (res instanceof CacheContinuousQueryEntry) {
- resList = new ArrayList<>();
-
- resList.add((CacheContinuousQueryEntry)res);
- }
- else {
- assert res instanceof List : res;
+ assert res instanceof List : res;
- resList = (List<CacheContinuousQueryEntry>)res;
- }
+ resList = (List<CacheContinuousQueryEntry>)res;
+ }
- resList.add(evt0);
+ resList.add(entry0);
- res = resList;
- }
+ res = resList;
}
- else
- filtered++;
+ }
+ else {
+ filtered++;
+
+ TestDebugLog.addEntryMessage(part,
+ cntr,
+ "buffer process inc filtered start=" + startCntr +
+ ", lastProc=" + lastProc +
+ " pos=" + pos +
+ ", filtered=" + filtered +
+ " topVer=" + entry0.topologyVersion());
}
pos = i;
@@ -335,10 +451,10 @@ public class CacheContinuousQueryEventBuffer {
return res;
}
- if (pos == evts.length -1) {
- Arrays.fill(evts, null);
+ if (pos == entries.length -1) {
+ Arrays.fill(entries, null);
- Batch nextBatch = new Batch(this.startCntr + BUF_SIZE, filtered, evts);
+ Batch nextBatch = new Batch(this.startCntr + BUF_SIZE, filtered, entries, entry.topologyVersion());
curBatch.set(nextBatch);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/d2942a58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
index 12eaa20..e031428 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
@@ -41,7 +41,13 @@ import static org.apache.ignite.internal.processors.cache.query.continuous.Cache
*/
class CacheContinuousQueryPartitionRecovery {
/** Event which means hole in sequence. */
- private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
+ private static final CacheContinuousQueryEntry HOLE;
+
+ static {
+ HOLE = new CacheContinuousQueryEntry();
+
+ HOLE.markFiltered();
+ }
/** */
private final static int MAX_BUFF_SIZE = LSNR_MAX_BUF_SIZE;
@@ -56,7 +62,7 @@ class CacheContinuousQueryPartitionRecovery {
private AffinityTopologyVersion curTop = AffinityTopologyVersion.NONE;
/** */
- private final Map<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
+ private final TreeMap<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
/**
* @param log Logger.
@@ -240,6 +246,8 @@ class CacheContinuousQueryPartitionRecovery {
}
}
else {
+ boolean skip = false;
+
while (iter.hasNext()) {
Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
@@ -269,7 +277,9 @@ class CacheContinuousQueryPartitionRecovery {
iter.remove();
}
- else {
+ else if (!pending.isFiltered()) {
+ skip = true;
+
TestDebugLog.addEntryMessage(entry.partition(),
entry.updateCounter(),
"stop process last=" + lastFiredEvt + " cntr=" + e.getKey() + " topVer=" + e.getValue().topologyVersion() + " f=" + pending.filteredCount());
@@ -277,6 +287,9 @@ class CacheContinuousQueryPartitionRecovery {
break;
}
}
+
+ if (skip)
+ pendingEvts.headMap(lastFiredEvt).clear();
}
}
[07/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/52b716b8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/52b716b8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/52b716b8
Branch: refs/heads/ignite-5075-cc-debug
Commit: 52b716b8dfd73651d461eebe4098efd71688f05c
Parents: b65a63d
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 12:01:52 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 12:18:27 2017 +0300
----------------------------------------------------------------------
...ContinuousQueryFailoverAbstractSelfTest.java | 73 ++++++++++----------
1 file changed, 36 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/52b716b8/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index befd1d7..3108edc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -400,7 +400,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
* @param ignite Ignite.
- * @param topVer Topology version.
+ * @param topVer Major topology version.
+ * @param minorVer Minor topology version.
* @throws Exception If failed.
*/
private void waitRebalanceFinished(Ignite ignite, long topVer, int minorVer) throws Exception {
@@ -511,9 +512,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
* @param nodes Count nodes.
* @param killedNodeIdx Killed node index.
* @param updCntrs Update counters.
- * @return {@code True} if counters matches.
*/
- private boolean checkPartCounter(int nodes, int killedNodeIdx, Map<Integer, Long> updCntrs) {
+ private void checkPartCounter(int nodes, int killedNodeIdx, Map<Integer, Long> updCntrs) {
for (int i = 0; i < nodes; i++) {
if (i == killedNodeIdx)
continue;
@@ -527,8 +527,6 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
assertEquals(e.getValue(), act.get(e.getKey()).get2());
}
}
-
- return true;
}
/**
@@ -753,8 +751,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
assert GridTestUtils.waitForCondition(new PA() {
@Override public boolean apply() {
// (SRV_NODES + 1 client node) - 1 primary - backup nodes.
- return qryClient.cluster().nodes().size() == (SRV_NODES + 1 /** client node */)
- - 1 /** Primary node */ - backups;
+ return qryClient.cluster().nodes().size() == (SRV_NODES + 1 /* client node */)
+ - 1 /* Primary node */ - backups;
}
}, 5000L);
@@ -1253,6 +1251,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
* @param expEvts Expected events.
* @param lsnr Listener.
+ * @throws Exception If failed.
*/
private void checkEvents(final List<T3<Object, Object, Object>> expEvts, final CacheEventListener3 lsnr,
boolean allowLoseEvt) throws Exception {
@@ -1347,9 +1346,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
QueryCursor<?> cur = qryClient.cache(DEFAULT_CACHE_NAME).query(qry);
- final Collection<Object> backupQueue = backupQueue(ignite(1));
-
- assertEquals(0, backupQueue.size());
+ assertEquals(0, backupQueue(ignite(1)).size());
IgniteCache<Object, Object> cache0 = ignite(0).cache(DEFAULT_CACHE_NAME);
@@ -1367,11 +1364,12 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
GridTestUtils.waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
- return backupQueue.isEmpty();
+ return backupQueue(ignite(1)).isEmpty();
}
}, 2000);
- assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+ assertTrue("Backup queue is not cleared: " + backupQueue(ignite(1)),
+ backupQueue(ignite(1)).size() < BACKUP_ACK_THRESHOLD);
if (!latch.await(5, SECONDS))
fail("Failed to wait for notifications [exp=" + keys.size() + ", left=" + lsnr.latch.getCount() + ']');
@@ -1389,11 +1387,11 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
GridTestUtils.waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
- return backupQueue.isEmpty();
+ return backupQueue(ignite(1)).isEmpty();
}
}, ACK_FREQ + 2000);
- assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.isEmpty());
+ assertTrue("Backup queue is not cleared: " + backupQueue(ignite(1)), backupQueue(ignite(1)).isEmpty());
if (!latch.await(5, SECONDS))
fail("Failed to wait for notifications [exp=" + keys.size() + ", left=" + lsnr.latch.getCount() + ']');
@@ -1421,9 +1419,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
QueryCursor<?> cur = qryClient.cache(DEFAULT_CACHE_NAME).query(qry);
- final Collection<Object> backupQueue = backupQueue(ignite(0));
-
- assertEquals(0, backupQueue.size());
+ assertEquals(0, backupQueue(ignite(0)).size());
long ttl = 100;
@@ -1433,9 +1429,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
final List<Integer> keys = primaryKeys(ignite(1).cache(DEFAULT_CACHE_NAME), BACKUP_ACK_THRESHOLD);
- CountDownLatch latch = new CountDownLatch(keys.size());
-
- lsnr.latch = latch;
+ lsnr.latch = new CountDownLatch(keys.size());
for (Integer key : keys) {
log.info("Put: " + key);
@@ -1445,11 +1439,12 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
GridTestUtils.waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
- return backupQueue.isEmpty();
+ return backupQueue(ignite(0)).isEmpty();
}
}, 2000);
- assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+ assertTrue("Backup queue is not cleared: " + backupQueue(ignite(0)),
+ backupQueue(ignite(0)).size() < BACKUP_ACK_THRESHOLD);
boolean wait = waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
@@ -1461,14 +1456,14 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
GridTestUtils.waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
- return backupQueue.isEmpty();
+ return backupQueue(ignite(0)).isEmpty();
}
}, 2000);
- assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+ assertTrue("Backup queue is not cleared: " + backupQueue(ignite(0)), backupQueue(ignite(0)).size() < BACKUP_ACK_THRESHOLD);
- if (backupQueue.size() != 0) {
- for (Object o : backupQueue) {
+ if (backupQueue(ignite(0)).size() != 0) {
+ for (Object o : backupQueue(ignite(0))) {
CacheContinuousQueryEntry e = (CacheContinuousQueryEntry)o;
assertNotSame("Evicted entry added to backup queue.", -1L, e.updateCounter());
@@ -1494,9 +1489,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
QueryCursor<?> cur = cache.query(qry);
- final Collection<Object> backupQueue = backupQueue(ignite(1));
-
- assertEquals(0, backupQueue.size());
+ assertEquals(0, backupQueue(ignite(1)).size());
List<Integer> keys = primaryKeys(cache, BACKUP_ACK_THRESHOLD);
@@ -1512,11 +1505,12 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
GridTestUtils.waitForCondition(new GridAbsPredicate() {
@Override public boolean apply() {
- return backupQueue.isEmpty();
+ return backupQueue(ignite(1)).isEmpty();
}
}, 3000);
- assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+ assertTrue("Backup queue is not cleared: " + backupQueue(ignite(1)),
+ backupQueue(ignite(1)).size() < BACKUP_ACK_THRESHOLD);
if (!latch.await(5, SECONDS))
fail("Failed to wait for notifications [exp=" + keys.size() + ", left=" + lsnr.latch.getCount() + ']');
@@ -1533,20 +1527,25 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
ConcurrentMap<Object, Object> infos = GridTestUtils.getFieldValue(proc, "rmtInfos");
- Collection<Object> backupQueue = null;
+ Collection<Object> backupQueue = new ArrayList<>();
for (Object info : infos.values()) {
GridContinuousHandler hnd = GridTestUtils.getFieldValue(info, "hnd");
- if (hnd.isQuery() && DEFAULT_CACHE_NAME.equals(hnd.cacheName())) {
- backupQueue = GridTestUtils.getFieldValue(hnd, CacheContinuousQueryHandler.class, "backupQueue");
+ if (hnd.isQuery() && hnd.cacheName().equals(DEFAULT_CACHE_NAME)) {
+ Map<Integer, CacheContinuousQueryEventBuffer> map = GridTestUtils.getFieldValue(hnd,
+ CacheContinuousQueryHandler.class, "entryBufs");
+
+ for (CacheContinuousQueryEventBuffer buf : map.values()) {
+ Collection<Object> q = GridTestUtils.getFieldValue(buf,
+ CacheContinuousQueryEventBuffer.class, "backupQ");
- break;
+ if (q != null)
+ backupQueue.addAll(q);
+ }
}
}
- assertNotNull(backupQueue);
-
return backupQueue;
}
[16/18] ignite git commit: cc
Posted by sb...@apache.org.
cc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ff0a2dd8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ff0a2dd8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ff0a2dd8
Branch: refs/heads/ignite-5075-cc-debug
Commit: ff0a2dd8aeaf621b79c504934f6830537323fcd1
Parents: e6ffd82
Author: sboikov <sb...@gridgain.com>
Authored: Thu May 25 15:37:47 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu May 25 15:37:47 2017 +0300
----------------------------------------------------------------------
.../CacheContinuousQueryEventBuffer.java | 201 ++++++++++++++-----
.../CacheContinuousQueryPartitionRecovery.java | 23 ++-
2 files changed, 167 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff0a2dd8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index f0640b1..f496c8c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.jetbrains.annotations.Nullable;
/**
@@ -75,15 +76,27 @@ public class CacheContinuousQueryEventBuffer {
* @return Backup entries.
*/
@Nullable Collection<CacheContinuousQueryEntry> resetBackupQueue() {
+ Collection<CacheContinuousQueryEntry> ret;
+
+ List<CacheContinuousQueryEntry> entries = null;
+
+ Batch batch = curBatch.get();
+
+ if (batch != null)
+ entries = batch.backupFlushEntries();
+
if (!backupQ.isEmpty()) {
- ConcurrentLinkedDeque<CacheContinuousQueryEntry> ret = this.backupQ;
+ if (entries != null)
+ backupQ.addAll(entries);
- backupQ = new ConcurrentLinkedDeque<>();
+ ret = this.backupQ;
- return ret;
+ backupQ = new ConcurrentLinkedDeque<>();
}
+ else
+ ret = entries;
- return null;
+ return ret;
}
/**
@@ -122,11 +135,9 @@ public class CacheContinuousQueryEventBuffer {
private Object process0(long cntr, CacheContinuousQueryEntry entry, boolean backup) {
assert cntr >= 0 : cntr;
- Batch batch = initBatch();
+ Batch batch = initBatch(entry.topologyVersion());
if (batch == null || cntr < batch.startCntr) {
- assert entry != null : cntr;
-
if (backup)
backupQ.add(entry);
@@ -157,9 +168,10 @@ public class CacheContinuousQueryEventBuffer {
}
/**
+ * @param topVer Current event topology version.
* @return Current batch.
*/
- @Nullable private Batch initBatch() {
+ @Nullable private Batch initBatch(AffinityTopologyVersion topVer) {
Batch batch = curBatch.get();
if (batch != null)
@@ -170,7 +182,7 @@ public class CacheContinuousQueryEventBuffer {
if (curCntr == -1)
return null;
- batch = new Batch(curCntr + 1, 0L, new Object[BUF_SIZE]);
+ batch = new Batch(curCntr + 1, 0L, new CacheContinuousQueryEntry[BUF_SIZE], topVer);
if (curBatch.compareAndSet(null, batch))
return batch;
@@ -216,28 +228,117 @@ public class CacheContinuousQueryEventBuffer {
private int lastProc = -1;
/** */
- private final Object[] evts;
+ private final CacheContinuousQueryEntry[] entries;
+
+ /** */
+ private final AffinityTopologyVersion topVer;
/**
* @param filtered Number of filtered events before this batch.
- * @param evts Events array.
+ * @param entries Entries array.
+ * @param topVer Current event topology version.
* @param startCntr Start counter.
*/
- Batch(long startCntr, long filtered, Object[] evts) {
+ Batch(long startCntr, long filtered, CacheContinuousQueryEntry[] entries, AffinityTopologyVersion topVer) {
assert startCntr >= 0;
assert filtered >= 0;
this.startCntr = startCntr;
this.filtered = filtered;
- this.evts = evts;
+ this.entries = entries;
+ this.topVer = topVer;
endCntr = startCntr + BUF_SIZE - 1;
}
/**
+ * @return Entries to send as part of backup queue.
+ */
+ @Nullable synchronized List<CacheContinuousQueryEntry> backupFlushEntries() {
+ List<CacheContinuousQueryEntry> res = null;
+
+ long filtered = this.filtered;
+ long cntr = startCntr;
+
+ for (int i = 0; i < entries.length; i++) {
+ CacheContinuousQueryEntry e = entries[i];
+
+ CacheContinuousQueryEntry flushEntry = null;
+
+ if (e == null) {
+ if (filtered != 0) {
+ flushEntry = filteredEntry(cntr - 1, filtered - 1);
+
+ filtered = 0;
+ }
+ }
+ else {
+ if (e.isFiltered())
+ filtered++;
+ else {
+ flushEntry = new CacheContinuousQueryEntry(e.cacheId(),
+ e.eventType(),
+ e.key(),
+ e.value(),
+ e.oldValue(),
+ e.isKeepBinary(),
+ e.partition(),
+ e.updateCounter(),
+ e.topologyVersion());
+
+ flushEntry.filteredCount(filtered);
+
+ filtered = 0;
+ }
+ }
+
+ if (flushEntry != null) {
+ if (res == null)
+ res = new ArrayList<>();
+
+ res.add(flushEntry);
+ }
+
+ cntr++;
+ }
+
+ if (filtered != 0L) {
+ if (res == null)
+ res = new ArrayList<>();
+
+ res.add(filteredEntry(cntr - 1, filtered - 1));
+ }
+
+ return res;
+ }
+
+ /**
+ * @param cntr Entry counter.
+ * @param filtered Number of entries filtered before this entry.
+ * @return Entry.
+ */
+ private CacheContinuousQueryEntry filteredEntry(long cntr, long filtered) {
+ CacheContinuousQueryEntry e = new CacheContinuousQueryEntry(0,
+ null,
+ null,
+ null,
+ null,
+ false,
+ part,
+ cntr,
+ topVer);
+
+ e.markFiltered();
+
+ e.filteredCount(filtered);
+
+ return e;
+ }
+
+ /**
* @param res Current result.
- * @param cntr Event counter.
- * @param evt Event.
+ * @param cntr Entry counter.
+ * @param entry Entry.
* @param backup Backup entry flag.
* @return New result.
*/
@@ -245,60 +346,54 @@ public class CacheContinuousQueryEventBuffer {
@Nullable private Object processEvent0(
@Nullable Object res,
long cntr,
- CacheContinuousQueryEntry evt,
+ CacheContinuousQueryEntry entry,
boolean backup) {
int pos = (int)(cntr - startCntr);
synchronized (this) {
- evts[pos] = evt;
+ entries[pos] = entry;
int next = lastProc + 1;
if (next == pos) {
- for (int i = next; i < evts.length; i++) {
- Object e = evts[i];
-
- if (e != null) {
- if (e.getClass() == Long.class)
- filtered++;
- else {
- CacheContinuousQueryEntry evt0 = (CacheContinuousQueryEntry)e;
+ for (int i = next; i < entries.length; i++) {
+ CacheContinuousQueryEntry entry0 = entries[i];
- if (!evt0.isFiltered()) {
- evt0.filteredCount(filtered);
+ if (entry0 != null) {
+ if (!entry0.isFiltered()) {
+ entry0.filteredCount(filtered);
- filtered = 0;
+ filtered = 0;
- if (res == null) {
- if (backup)
- backupQ.add(evt0);
- else
- res = evt0;
- }
- else {
- assert !backup;
+ if (res == null) {
+ if (backup)
+ backupQ.add(entry0);
+ else
+ res = entry0;
+ }
+ else {
+ assert !backup;
- List<CacheContinuousQueryEntry> resList;
+ List<CacheContinuousQueryEntry> resList;
- if (res instanceof CacheContinuousQueryEntry) {
- resList = new ArrayList<>();
+ if (res instanceof CacheContinuousQueryEntry) {
+ resList = new ArrayList<>();
- resList.add((CacheContinuousQueryEntry)res);
- }
- else {
- assert res instanceof List : res;
+ resList.add((CacheContinuousQueryEntry)res);
+ }
+ else {
+ assert res instanceof List : res;
- resList = (List<CacheContinuousQueryEntry>)res;
- }
+ resList = (List<CacheContinuousQueryEntry>)res;
+ }
- resList.add(evt0);
+ resList.add(entry0);
- res = resList;
- }
+ res = resList;
}
- else
- filtered++;
}
+ else
+ filtered++;
pos = i;
}
@@ -312,10 +407,10 @@ public class CacheContinuousQueryEventBuffer {
return res;
}
- if (pos == evts.length -1) {
- Arrays.fill(evts, null);
+ if (pos == entries.length -1) {
+ Arrays.fill(entries, null);
- Batch nextBatch = new Batch(this.startCntr + BUF_SIZE, filtered, evts);
+ Batch nextBatch = new Batch(this.startCntr + BUF_SIZE, filtered, entries, entry.topologyVersion());
curBatch.set(nextBatch);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff0a2dd8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
index 534ce9c..59252d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryPartitionRecovery.java
@@ -38,7 +38,13 @@ import org.jetbrains.annotations.Nullable;
*/
class CacheContinuousQueryPartitionRecovery {
/** Event which means hole in sequence. */
- private static final CacheContinuousQueryEntry HOLE = new CacheContinuousQueryEntry();
+ private static final CacheContinuousQueryEntry HOLE;
+
+ static {
+ HOLE = new CacheContinuousQueryEntry();
+
+ HOLE.markFiltered();
+ }
/** */
private final static int MAX_BUFF_SIZE = CacheContinuousQueryHandler.LSNR_MAX_BUF_SIZE;
@@ -53,7 +59,7 @@ class CacheContinuousQueryPartitionRecovery {
private AffinityTopologyVersion curTop = AffinityTopologyVersion.NONE;
/** */
- private final Map<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
+ private final TreeMap<Long, CacheContinuousQueryEntry> pendingEvts = new TreeMap<>();
/**
* @param log Logger.
@@ -212,6 +218,8 @@ class CacheContinuousQueryPartitionRecovery {
}
}
else {
+ boolean skippedFiltered = false;
+
while (iter.hasNext()) {
Map.Entry<Long, CacheContinuousQueryEntry> e = iter.next();
@@ -232,9 +240,16 @@ class CacheContinuousQueryPartitionRecovery {
iter.remove();
}
- else
- break;
+ else {
+ if (pending.isFiltered())
+ skippedFiltered = true;
+ else
+ break;
+ }
}
+
+ if (skippedFiltered)
+ pendingEvts.headMap(lastFiredEvt).clear();
}
}