ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [2/2] ignite git commit: zk
Date Thu, 28 Dec 2017 09:03:55 GMT
zk


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/0b78f318
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/0b78f318
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/0b78f318

Branch: refs/heads/ignite-zk
Commit: 0b78f318b12bb03f4ddce32e7f9c5d4eba8305ca
Parents: 6961ddc
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Dec 28 12:03:34 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Dec 28 12:03:34 2017 +0300

----------------------------------------------------------------------
 .../DiscoveryMessageResultsCollector.java       | 222 +++++++++++++++++++
 .../continuous/GridContinuousProcessor.java     | 191 +++++++---------
 2 files changed, 302 insertions(+), 111 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/0b78f318/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageResultsCollector.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageResultsCollector.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageResultsCollector.java
new file mode 100644
index 0000000..72a4636
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageResultsCollector.java
@@ -0,0 +1,222 @@
+/*
+ * 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.managers.discovery;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public abstract class DiscoveryMessageResultsCollector<M, R>  {
+    /** */
+    private final Map<UUID, NodeMessage<M>> rcvd = new HashMap<>();
+
+    /** */
+    private int leftMsgs;
+
+    /** */
+    protected DiscoCache discoCache;
+
+    /** */
+    protected final GridKernalContext ctx;
+
+    /**
+     * @param ctx Context.
+     */
+    protected DiscoveryMessageResultsCollector(GridKernalContext ctx) {
+        this.ctx = ctx;
+    }
+
+    /**
+     * @param rcvd Received messages.
+     * @return Result.
+     */
+    protected abstract R createResult(Map<UUID, NodeMessage<M>> rcvd);
+
+    /**
+     * @param r Result.
+     */
+    protected abstract void onResultsCollected(R r);
+
+    /**
+     * @param discoCache Discovery state when discovery message was received.
+     * @param node Node.
+     * @return {@code True} if need wait for result from given node.
+     */
+    protected abstract boolean waitForNode(DiscoCache discoCache, ClusterNode node);
+
+    /**
+     * @param discoCache Discovery state.
+     */
+    public final void init(DiscoCache discoCache) {
+        assert discoCache != null;
+
+        R res = null;
+
+        synchronized (this) {
+            assert this.discoCache == null;
+            assert leftMsgs == 0 : leftMsgs;
+
+            this.discoCache = discoCache;
+
+            for (ClusterNode node : discoCache.allNodes()) {
+                if (ctx.discovery().alive(node) && waitForNode(discoCache, node)
&& !rcvd.containsKey(node.id())) {
+                    rcvd.put(node.id(), new NodeMessage<>((M)null));
+
+                    leftMsgs++;
+                }
+            }
+
+            if (leftMsgs == 0)
+                res = createResult(rcvd);
+        }
+
+        if (res != null)
+            onResultsCollected(res);
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @param msg Message.
+     */
+    public final void onMessage(UUID nodeId, M msg) {
+        R res = null;
+
+        synchronized (this) {
+            if (allReceived())
+                return;
+
+            NodeMessage<M> expMsg = rcvd.get(nodeId);
+
+            if (expMsg == null)
+                rcvd.put(nodeId, new NodeMessage<>(msg));
+            else if (expMsg.set(msg)) {
+                assert leftMsgs > 0;
+
+                leftMsgs--;
+
+                if (allReceived())
+                    res = createResult(rcvd);
+            }
+        }
+
+        if (res != null)
+            onResultsCollected(res);
+    }
+
+    /**
+     * @param nodeId Failed node ID.
+     */
+    public final void onNodeFail(UUID nodeId) {
+        R res = null;
+
+        synchronized (this) {
+            if (allReceived())
+                return;
+
+            NodeMessage expMsg = rcvd.get(nodeId);
+
+            if (expMsg != null && expMsg.onNodeFailed()) {
+                assert leftMsgs > 0 : leftMsgs;
+
+                leftMsgs--;
+
+                if (allReceived())
+                    res = createResult(rcvd);
+            }
+        }
+
+        if (res != null)
+            onResultsCollected(res);
+    }
+
+    /**
+     * @return {@code True} if expected messages are initialized and all message are received.
+     */
+    private boolean allReceived() {
+        return discoCache != null && leftMsgs == 0;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(DiscoveryMessageResultsCollector.class, this);
+    }
+
+    /**
+     *
+     */
+    protected static class NodeMessage<M> {
+        /** */
+        boolean nodeFailed;
+
+        /** */
+        M msg;
+
+        /**
+         * @param msg Message.
+         */
+        NodeMessage(M msg) {
+            this.msg = msg;
+        }
+
+        /**
+         * @return Message or {@code null} if node failed.
+         */
+        @Nullable public M message() {
+            return msg;
+        }
+
+        /**
+         * @return {@code True} if node result was not set before.
+         */
+        boolean onNodeFailed() {
+            if (nodeFailed || msg != null)
+                return false;
+
+            nodeFailed = true;
+
+            return false;
+        }
+
+        /**
+         * @param msg Received message.
+         * @return {@code True} if node result was not set before.
+         */
+        boolean set(M msg) {
+            assert msg != null;
+
+            if (this.msg != null)
+                return false;
+
+            this.msg = msg;
+
+            return !nodeFailed;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(NodeMessage.class, this);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/0b78f318/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 ee12de3..950ac7e 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
@@ -49,12 +49,13 @@ import org.apache.ignite.internal.IgniteDeploymentCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
-import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
 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.managers.discovery.CustomEventListener;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageResultsCollector;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
 import org.apache.ignite.internal.managers.eventstorage.HighPriorityListener;
 import org.apache.ignite.internal.processors.GridProcessorAdapter;
@@ -1403,7 +1404,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
 
         routinesInfo.addRoutineInfo(routineInfo);
 
-        final Collection<ClusterNode> nodes = ctx.discovery().nodes(topVer);
+        final DiscoCache discoCache = ctx.discovery().discoCache(topVer);
 
         // Should not use marshaller and send messages from discovery thread.
         ctx.getSystemExecutorService().execute(new Runnable() {
@@ -1412,7 +1413,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                     StartFuture fut = startFuts.get(msg.routineId());
 
                     if (fut != null)
-                        fut.initRemoteNodes(topVer, nodes);
+                        fut.initRemoteNodes(discoCache);
 
                     return;
                 }
@@ -2340,19 +2341,83 @@ public class GridContinuousProcessor extends GridProcessorAdapter
{
         private volatile boolean rmt;
 
         /** */
-        private AffinityTopologyVersion topVer;
-
-        /** */
-        private int expRes;
-
-        /** */
-        private final Map<UUID, ContinuousRoutineStartResultMessage> res = new HashMap<>();
+        private final DiscoveryMessageResultsCollector<ContinuousRoutineStartResultMessage,
RoutineRegisterResults>
+            resCollect;
 
         /**
          * @param routineId Consume ID.
          */
         StartFuture(UUID routineId) {
             this.routineId = routineId;
+
+            resCollect = new DiscoveryMessageResultsCollector<ContinuousRoutineStartResultMessage,
RoutineRegisterResults>(ctx) {
+                @Override protected RoutineRegisterResults createResult(Map<UUID, NodeMessage<ContinuousRoutineStartResultMessage>>
rcvd) {
+                    Map<UUID, Exception> errs = null;
+                    Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode
= null;
+
+                    for (Map.Entry<UUID, NodeMessage<ContinuousRoutineStartResultMessage>>
entry : rcvd.entrySet()) {
+                        ContinuousRoutineStartResultMessage msg = entry.getValue().message();
+
+                        if (msg == null)
+                            continue;
+
+                        if (msg.error()) {
+                            byte[] errBytes = msg.errorBytes();
+
+                            Exception err = null;
+
+                            if (errBytes != null) {
+                                try {
+                                    err = U.unmarshal(marsh, errBytes, U.resolveClassLoader(ctx.config()));
+                                }
+                                catch (Exception e) {
+                                    U.warn(log, "Failed to unmarhal continuous routine start
error: " + e);
+                                }
+                            }
+
+                            if (err == null) {
+                                err = new IgniteCheckedException("Failed to start continuous
" +
+                                    "routine on node: " + entry.getKey());
+                            }
+
+                            if (errs == null)
+                                errs = new HashMap<>();
+
+                            errs.put(entry.getKey(), err);
+                        }
+                        else {
+                            byte[] cntrsMapBytes = msg.countersMapBytes();
+
+                            if (cntrsMapBytes != null) {
+                                try {
+                                    CachePartitionPartialCountersMap cntrsMap = U.unmarshal(
+                                        marsh,
+                                        cntrsMapBytes,
+                                        U.resolveClassLoader(ctx.config()));
+
+                                    if (cntrsPerNode == null)
+                                        cntrsPerNode = new HashMap<>();
+
+                                    cntrsPerNode.put(entry.getKey(), CachePartitionPartialCountersMap.toCountersMap(cntrsMap));
+                                }
+                                catch (Exception e) {
+                                    U.warn(log, "Failed to unmarhal continuous query update
counters: " + e);
+                                }
+                            }
+                        }
+                    }
+
+                    return new RoutineRegisterResults(discoCache.version(), errs, cntrsPerNode);
+                }
+
+                @Override protected void onResultsCollected(RoutineRegisterResults res0)
{
+                    onAllRemoteRegistered(res0.topVer, res0.errs, res0.cntrsPerNode, null);
+                }
+
+                @Override protected boolean waitForNode(DiscoCache discoCache, ClusterNode
node) {
+                    return !ctx.localNodeId().equals(node.id());
+                }
+            };
         }
 
         /**
@@ -2400,28 +2465,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter
{
         }
 
         /**
-         * @param topVer Topology version.
-         * @param nodes Nodes.
+         * @param discoCache Discovery state.
          */
-        void initRemoteNodes(AffinityTopologyVersion topVer, Collection<ClusterNode>
nodes) {
-            RoutineRegisterResults res0 = null;
-
-            synchronized (res) {
-                assert this.topVer == null && expRes == 0;
-
-                this.topVer = topVer;
-
-                for (ClusterNode node : nodes) {
-                    if (!ctx.localNodeId().equals(node.id()) && ctx.discovery().alive(node.id()))
-                        expRes++;
-                }
-
-                if (res.size() >= expRes)
-                    res0 = createRegisterResults();
-            }
-
-            if (res0 != null)
-                onAllRemoteRegistered(res0.topVer, res0.errs, res0.cntrsPerNode, null);
+        void initRemoteNodes(DiscoCache discoCache) {
+            resCollect.init(discoCache);
         }
 
         /**
@@ -2429,27 +2476,14 @@ public class GridContinuousProcessor extends GridProcessorAdapter
{
          * @param msg Message.
          */
         void onResult(UUID nodeId, ContinuousRoutineStartResultMessage msg) {
-            RoutineRegisterResults res0 = null;
-
-            synchronized (res) {
-                if (res.containsKey(nodeId) || (topVer != null && res.size() >=
expRes))
-                    return;
-
-                res.put(nodeId, msg);
-
-                if (topVer != null && res.size() >= expRes)
-                    res0 = createRegisterResults();
-            }
-
-            if (res0 != null)
-                onAllRemoteRegistered(res0.topVer, res0.errs, res0.cntrsPerNode, null);
+            resCollect.onMessage(nodeId, msg);
         }
 
         /**
          * @param nodeId Failed node ID.
          */
         void onNodeFail(UUID nodeId) {
-            onResult(nodeId, new ContinuousRoutineStartResultMessage(routineId, null, null,
false));
+            resCollect.onNodeFail(nodeId);
         }
 
         /**
@@ -2472,65 +2506,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                 onDone(routineId);
         }
 
-        /**
-         * @return Results.
-         */
-        private RoutineRegisterResults createRegisterResults() {
-            Map<UUID, Exception> errs = null;
-            Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode = null;
-
-            for (Map.Entry<UUID, ContinuousRoutineStartResultMessage> entry : res.entrySet())
{
-                ContinuousRoutineStartResultMessage msg = entry.getValue();
-
-                if (msg.error()) {
-                    byte[] errBytes = msg.errorBytes();
-
-                    Exception err = null;
-
-                    if (errBytes != null) {
-                        try {
-                            err = U.unmarshal(marsh, errBytes, U.resolveClassLoader(ctx.config()));
-                        }
-                        catch (Exception e) {
-                            U.warn(log, "Failed to unmarhal continuous routine start error:
" + e);
-                        }
-                    }
-
-                    if (err == null) {
-                        err = new IgniteCheckedException("Failed to start continuous " +
-                            "routine on node: " + entry.getKey());
-                    }
-
-                    if (errs == null)
-                        errs = new HashMap<>();
-
-                    errs.put(entry.getKey(), err);
-                }
-                else {
-                    byte[] cntrsMapBytes = msg.countersMapBytes();
-
-                    if (cntrsMapBytes != null) {
-                        try {
-                            CachePartitionPartialCountersMap cntrsMap = U.unmarshal(
-                                marsh,
-                                cntrsMapBytes,
-                                U.resolveClassLoader(ctx.config()));
-
-                            if (cntrsPerNode == null)
-                                cntrsPerNode = new HashMap<>();
-
-                            cntrsPerNode.put(entry.getKey(), CachePartitionPartialCountersMap.toCountersMap(cntrsMap));
-                        }
-                        catch (Exception e) {
-                            U.warn(log, "Failed to unmarhal continuous query update counters:
" + e);
-                        }
-                    }
-                }
-            }
-
-            return new RoutineRegisterResults(topVer, errs, cntrsPerNode);
-        }
-
         /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(StartFuture.class, this);
@@ -2568,9 +2543,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      * Future for stop routine.
      */
     private static class StopFuture extends GridFutureAdapter<Object> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
         /** Timeout object. */
         private volatile GridTimeoutObject timeoutObj;
 
@@ -2614,9 +2586,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      */
     private static class SyncMessageAckFuture extends GridFutureAdapter<Object> {
         /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
         private UUID nodeId;
 
         /**


Mime
View raw message