ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sevdoki...@apache.org
Subject [09/31] incubator-ignite git commit: # IGNITE-831 Support listeners from clients.
Date Fri, 15 May 2015 10:06:30 GMT
# IGNITE-831 Support listeners from clients.


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

Branch: refs/heads/ignite-709_2
Commit: 37c60075d234480b9217a6456cbfc94e380c3b3a
Parents: e028de8
Author: sevdokimov <sevdokimov@gridgain.com>
Authored: Tue May 5 19:38:29 2015 +0300
Committer: sevdokimov <sevdokimov@gridgain.com>
Committed: Tue May 5 19:38:29 2015 +0300

----------------------------------------------------------------------
 .../continuous/GridContinuousProcessor.java     | 109 ++++++++++++-------
 1 file changed, 67 insertions(+), 42 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37c60075/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 d71609b..d1923d9 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
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.util.worker.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.marshaller.*;
 import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.spi.discovery.tcp.internal.*;
 import org.apache.ignite.thread.*;
 import org.jetbrains.annotations.*;
 import org.jsr166.*;
@@ -59,6 +60,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /** Local infos. */
     private final ConcurrentMap<UUID, LocalRoutineInfo> locInfos = new ConcurrentHashMap8<>();
 
+    /** Local infos. */
+    private final ConcurrentMap<UUID, Map<UUID, LocalRoutineInfo>> clientInfos
= new ConcurrentHashMap8<>();
+
     /** Remote infos. */
     private final ConcurrentMap<UUID, RemoteRoutineInfo> rmtInfos = new ConcurrentHashMap8<>();
 
@@ -77,9 +81,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /** Stopped IDs. */
     private final Collection<UUID> stopped = new HashSet<>();
 
-    /** Lock for pending requests. */
-    private final Lock pendingLock = new ReentrantLock();
-
     /** Lock for stop process. */
     private final Lock stopLock = new ReentrantLock();
 
@@ -113,10 +114,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             @SuppressWarnings({"fallthrough", "TooBroadScope"})
             @Override public void onEvent(Event evt) {
                 assert evt instanceof DiscoveryEvent;
+                assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED;
 
                 UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
 
-                assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED;
+                clientInfos.remove(nodeId);
 
                 // Unregister handlers created by left node.
                 for (Map.Entry<UUID, RemoteRoutineInfo> e : rmtInfos.entrySet()) {
@@ -148,7 +150,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             new CustomEventListener<StartRoutineDiscoveryMessage>() {
                 @Override public void onCustomEvent(ClusterNode snd, StartRoutineDiscoveryMessage
msg) {
                     if (!snd.id().equals(ctx.localNodeId()))
-                        processStartRequest(snd.id(), msg);
+                        processStartRequest(snd, msg);
                 }
             });
 
@@ -179,6 +181,13 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                         UUID routineId = msg.routineId();
 
                         unregisterRemote(routineId);
+
+                        if (((TcpDiscoveryNode)snd).clientRouterNodeId() != null) {
+                            Map<UUID, LocalRoutineInfo> infoMap = clientInfos.get(snd.id());
+
+                            if (infoMap != null)
+                                infoMap.remove(msg.routineId());
+                        }
                     }
                 }
             });
@@ -251,26 +260,19 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /** {@inheritDoc} */
     @Override @Nullable public Serializable collectDiscoveryData(UUID nodeId) {
         if (!nodeId.equals(ctx.localNodeId())) {
-            pendingLock.lock();
-
-            try {
-                DiscoveryData data = new DiscoveryData(ctx.localNodeId());
+            DiscoveryData data = new DiscoveryData(ctx.localNodeId(), clientInfos);
 
-                // Collect listeners information (will be sent to
-                // joining node during discovery process).
-                for (Map.Entry<UUID, LocalRoutineInfo> e : locInfos.entrySet()) {
-                    UUID routineId = e.getKey();
-                    LocalRoutineInfo info = e.getValue();
-
-                    data.addItem(new DiscoveryDataItem(routineId, info.prjPred,
-                        info.hnd, info.bufSize, info.interval));
-                }
+            // Collect listeners information (will be sent to
+            // joining node during discovery process).
+            for (Map.Entry<UUID, LocalRoutineInfo> e : locInfos.entrySet()) {
+                UUID routineId = e.getKey();
+                LocalRoutineInfo info = e.getValue();
 
-                return data;
-            }
-            finally {
-                pendingLock.unlock();
+                data.addItem(new DiscoveryDataItem(routineId, info.prjPred,
+                    info.hnd, info.bufSize, info.interval));
             }
+
+            return data;
         }
         else
             return null;
@@ -294,6 +296,18 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                     }
                 }
             }
+
+            for (Map.Entry<UUID, Map<UUID, LocalRoutineInfo>> entry : data.clientInfos.entrySet())
{
+                Map<UUID, LocalRoutineInfo> map = clientInfos.get(entry.getKey());
+
+                if (map == null) {
+                    map = new HashMap<>();
+
+                    clientInfos.put(entry.getKey(), map);
+                }
+
+                map.putAll(entry.getValue());
+            }
         }
     }
 
@@ -414,15 +428,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             });
         }
 
-        pendingLock.lock();
-
-        try {
-            // Register routine locally.
-            locInfos.put(routineId, new LocalRoutineInfo(prjPred, hnd, bufSize, interval));
-        }
-        finally {
-            pendingLock.unlock();
-        }
+        // Register routine locally.
+        locInfos.put(routineId, new LocalRoutineInfo(prjPred, hnd, bufSize, interval));
 
         StartFuture fut = new StartFuture(ctx, routineId);
 
@@ -581,13 +588,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     }
 
     /**
-     * @param nodeId Sender ID.
+     * @param node Sender.
      * @param req Start request.
      */
-    private void processStartRequest(UUID nodeId, StartRoutineDiscoveryMessage req) {
-        assert nodeId != null;
-        assert req != null;
-
+    private void processStartRequest(ClusterNode node, StartRoutineDiscoveryMessage req)
{
         UUID routineId = req.routineId();
         StartRequestData data = req.startRequestData();
 
@@ -603,7 +607,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                     GridDeploymentInfo depInfo = data.deploymentInfo();
 
                     GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(),
clsName, clsName,
-                        depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(),
null);
+                        depInfo.userVersion(), node.id(), depInfo.classLoaderId(), depInfo.participants(),
null);
 
                     if (dep == null)
                         throw new IgniteDeploymentCheckedException("Failed to obtain deployment
for class: " + clsName);
@@ -611,13 +615,28 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                     data.p2pUnmarshal(marsh, dep.classLoader());
                 }
 
-                hnd.p2pUnmarshal(nodeId, ctx);
+                hnd.p2pUnmarshal(node.id(), ctx);
             }
         }
         catch (IgniteCheckedException e) {
             err = e;
 
-            U.error(log, "Failed to register handler [nodeId=" + nodeId + ", routineId="
+ routineId + ']', e);
+            U.error(log, "Failed to register handler [nodeId=" + node.id() + ", routineId="
+ routineId + ']', e);
+        }
+
+        if (((TcpDiscoveryNode)node).clientRouterNodeId() != null) {
+            Map<UUID, LocalRoutineInfo> clientRouteMap = clientInfos.get(node.id());
+
+            if (clientRouteMap == null) {
+                clientRouteMap = new HashMap<>();
+
+                Map<UUID, LocalRoutineInfo> old = clientInfos.put(node.id(), clientRouteMap);
+
+                assert old == null;
+            }
+
+            clientRouteMap.put(routineId, new LocalRoutineInfo(data.projectionPredicate(),
hnd, data.bufferSize(),
+                data.interval()));
         }
 
         boolean registered = false;
@@ -627,14 +646,14 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                 IgnitePredicate<ClusterNode> prjPred = data.projectionPredicate();
 
                 if (prjPred == null || prjPred.apply(ctx.discovery().node(ctx.localNodeId())))
{
-                    registered = registerHandler(nodeId, routineId, hnd, data.bufferSize(),
data.interval(),
+                    registered = registerHandler(node.id(), routineId, hnd, data.bufferSize(),
data.interval(),
                         data.autoUnsubscribe(), false);
                 }
             }
             catch (IgniteCheckedException e) {
                 err = e;
 
-                U.error(log, "Failed to register handler [nodeId=" + nodeId + ", routineId="
+ routineId + ']', e);
+                U.error(log, "Failed to register handler [nodeId=" + node.id() + ", routineId="
+ routineId + ']', e);
             }
         }
 
@@ -1129,6 +1148,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
         @GridToStringInclude
         private Collection<DiscoveryDataItem> items;
 
+        private Map<UUID, Map<UUID, LocalRoutineInfo>> clientInfos;
+
         /**
          * Required by {@link Externalizable}.
          */
@@ -1139,11 +1160,13 @@ public class GridContinuousProcessor extends GridProcessorAdapter
{
         /**
          * @param nodeId Node ID.
          */
-        DiscoveryData(UUID nodeId) {
+        DiscoveryData(UUID nodeId, Map<UUID, Map<UUID, LocalRoutineInfo>> clientInfos)
{
             assert nodeId != null;
 
             this.nodeId = nodeId;
 
+            this.clientInfos = clientInfos;
+
             items = new ArrayList<>();
         }
 
@@ -1158,12 +1181,14 @@ public class GridContinuousProcessor extends GridProcessorAdapter
{
         @Override public void writeExternal(ObjectOutput out) throws IOException {
             U.writeUuid(out, nodeId);
             U.writeCollection(out, items);
+            U.writeMap(out, clientInfos);
         }
 
         /** {@inheritDoc} */
         @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
             nodeId = U.readUuid(in);
             items = U.readCollection(in);
+            clientInfos = U.readMap(in);
         }
 
         /** {@inheritDoc} */


Mime
View raw message