diff options
| author | Ted Ross <tross@apache.org> | 2013-08-01 20:19:24 +0000 |
|---|---|---|
| committer | Ted Ross <tross@apache.org> | 2013-08-01 20:19:24 +0000 |
| commit | 437b93efc7f559a94f05e552527795c1df68516c (patch) | |
| tree | 549413e09e043b825f33991f23ec99538bcaf73e /qpid/extras/dispatch/src/py | |
| parent | 7919bb3eb92d85f9dcb3d6e3ada2370f3b419bb2 (diff) | |
| download | qpid-python-437b93efc7f559a94f05e552527795c1df68516c.tar.gz | |
QPID-4967 - Updates to the router
- The router module now tracks other router nodes (neighbors and non-neighbors)
- Tracked nodes are communicated to the router_node.c fast-path
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1509415 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/extras/dispatch/src/py')
| -rw-r--r-- | qpid/extras/dispatch/src/py/router/link.py | 3 | ||||
| -rw-r--r-- | qpid/extras/dispatch/src/py/router/neighbor.py | 2 | ||||
| -rw-r--r-- | qpid/extras/dispatch/src/py/router/node.py | 103 | ||||
| -rw-r--r-- | qpid/extras/dispatch/src/py/router/router_engine.py | 25 |
4 files changed, 133 insertions, 0 deletions
diff --git a/qpid/extras/dispatch/src/py/router/link.py b/qpid/extras/dispatch/src/py/router/link.py index 1e06d161f6..ce97794b8c 100644 --- a/qpid/extras/dispatch/src/py/router/link.py +++ b/qpid/extras/dispatch/src/py/router/link.py @@ -87,6 +87,7 @@ class LinkStateEngine(object): self.collection[msg.id] = ls self.collection_changed = True ls.last_seen = now + self.container.new_node(msg.id) self.container.log(LOG_INFO, "Learned link-state from new router: %s" % msg.id) # Schedule LSRs for any routers referenced in this LS that we don't know about for _id in msg.ls.peers: @@ -108,6 +109,7 @@ class LinkStateEngine(object): self.collection_changed = True self._send_ra() + def set_mobile_sequence(self, seq): self.mobile_seq = seq @@ -124,6 +126,7 @@ class LinkStateEngine(object): for key in to_delete: ls = self.collection.pop(key) self.collection_changed = True + self.container.lost_node(key) self.container.log(LOG_INFO, "Expired link-state from router: %s" % key) diff --git a/qpid/extras/dispatch/src/py/router/neighbor.py b/qpid/extras/dispatch/src/py/router/neighbor.py index 55c6bab62f..8d0dfceecf 100644 --- a/qpid/extras/dispatch/src/py/router/neighbor.py +++ b/qpid/extras/dispatch/src/py/router/neighbor.py @@ -64,6 +64,7 @@ class NeighborEngine(object): if msg.is_seen(self.id): if self.link_state.add_peer(msg.id): self.link_state_changed = True + self.container.new_neighbor(msg.id) self.container.log(LOG_INFO, "New neighbor established: %s" % msg.id) ## ## TODO - Use this function to detect area boundaries @@ -78,6 +79,7 @@ class NeighborEngine(object): self.hellos.pop(key) if self.link_state.del_peer(key): self.link_state_changed = True + self.container.lost_neighbor(key) self.container.log(LOG_INFO, "Neighbor lost: %s" % key) diff --git a/qpid/extras/dispatch/src/py/router/node.py b/qpid/extras/dispatch/src/py/router/node.py new file mode 100644 index 0000000000..482a83dfe8 --- /dev/null +++ b/qpid/extras/dispatch/src/py/router/node.py @@ -0,0 +1,103 @@ +# +# 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. +# + +try: + from dispatch import * +except ImportError: + from stubs import * + + +class NodeTracker(object): + """ + This module is responsible for tracking the set of router nodes that are known to this + router. It tracks whether they are neighbor or remote and whether they are reachable. + """ + def __init__(self, container): + self.container = container + self.id = self.container.id + self.area = self.container.area + self.nodes = {} # id => RemoteNode + + + def tick(self, now): + pass + + + def new_neighbor(self, node_id): + if node_id not in self.nodes: + self.nodes[node_id] = RemoteNode(node_id) + self.nodes[node_id].set_neighbor() + self._notify(self.nodes[node_id]) + + + def lost_neighbor(self, node_id): + node = self.nodes[node_id] + node.clear_neighbor() + self._notify(node) + if node.to_delete(): + self.nodes.pop(node_id) + + + def new_node(self, node_id): + if node_id not in self.nodes: + self.nodes[node_id] = RemoteNode(node_id) + self.nodes[node_id].set_remote() + self._notify(self.nodes[node_id]) + + + def lost_node(self, node_id): + node = self.nodes[node_id] + node.clear_remote() + self._notify(node) + if node.to_delete(): + self.nodes.pop(node_id) + + + def _notify(self, node): + if node.to_delete(): + self.container.adapter.node_updated("R%s" % node.id, 0, 0) + else: + is_neighbor = 0 + if node.neighbor: + is_neighbor = 1 + self.container.adapter.node_updated("R%s" % node.id, 1, is_neighbor) + + +class RemoteNode(object): + + def __init__(self, node_id): + self.id = node_id + self.neighbor = None + self.remote = None + + def set_neighbor(self): + self.neighbor = True + + def set_remote(self): + self.remote = True + + def clear_neighbor(self): + self.neighbor = None + + def clear_remote(self): + self.remote = None + + def to_delete(self): + return self.neighbor or self.remote + diff --git a/qpid/extras/dispatch/src/py/router/router_engine.py b/qpid/extras/dispatch/src/py/router/router_engine.py index 065204ad62..675d9c32e5 100644 --- a/qpid/extras/dispatch/src/py/router/router_engine.py +++ b/qpid/extras/dispatch/src/py/router/router_engine.py @@ -29,6 +29,7 @@ from mobile import MobileAddressEngine from routing import RoutingTableEngine from binding import BindingEngine from adapter import AdapterEngine +from node import NodeTracker ## ## Import the Dispatch adapters from the environment. If they are not found @@ -79,6 +80,7 @@ class RouterEngine: self.routing_table_engine = RoutingTableEngine(self) self.binding_engine = BindingEngine(self) self.adapter_engine = AdapterEngine(self) + self.node_tracker = NodeTracker(self) @@ -125,6 +127,7 @@ class RouterEngine: self.routing_table_engine.tick(now) self.binding_engine.tick(now) self.adapter_engine.tick(now) + self.node_tracker.tick(now) except Exception, e: self.log(LOG_ERROR, "Exception in timer processing: exception=%r" % e) @@ -222,6 +225,12 @@ class RouterEngine: self.log(LOG_TRACE, "SENT: %r dest=%s" % (msg, dest)) + def node_updated(self, addr, reachable, neighbor): + """ + """ + self.router_adapter(addr, reachable, neighbor) + + ##======================================================================================== ## Interconnect between the Sub-Modules ##======================================================================================== @@ -253,3 +262,19 @@ class RouterEngine: self.log(LOG_DEBUG, "Event: remote_routes_changed: class=%s routes=%r" % (key_class, routes)) self.adapter_engine.remote_routes_changed(key_class, routes) + def new_neighbor(self, rid): + self.log(LOG_DEBUG, "Event: new_neighbor: id=%s" % rid) + self.node_tracker.new_neighbor(rid) + + def lost_neighbor(self, rid): + self.log(LOG_DEBUG, "Event: lost_neighbor: id=%s" % rid) + self.node_tracker.lost_neighbor(rid) + + def new_node(self, rid): + self.log(LOG_DEBUG, "Event: new_node: id=%s" % rid) + self.node_tracker.new_node(rid) + + def lost_node(self, rid): + self.log(LOG_DEBUG, "Event: lost_node: id=%s" % rid) + self.node_tracker.lost_node(rid) + |
