You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2013/09/26 23:14:59 UTC
svn commit: r1526694 [1/3] - in /qpid/trunk/qpid/extras/dispatch: ./
include/qpid/ include/qpid/dispatch/ python/qpid/dispatch/router/
router/src/ src/ tests/
Author: tross
Date: Thu Sep 26 21:14:59 2013
New Revision: 1526694
URL: http://svn.apache.org/r1526694
Log:
QPID-5173
QPID-5045
QPID-5181
- Major refactor of the routing data structure in preparation for multi-router operation.
- Fixed the CMake bug in QPID-5173
- Added Dynamic assignment of routable addresses for outbound links (QPID-5181)
- Changed the indentation of the Python code from 2 spaces to 4 spaces.
- Reduced the default log level to make the console less chatty.
Added:
qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h (with props)
qpid/trunk/qpid/extras/dispatch/src/amqp.c (with props)
qpid/trunk/qpid/extras/dispatch/src/bitmask.c (with props)
qpid/trunk/qpid/extras/dispatch/src/router_private.h (with props)
Modified:
qpid/trunk/qpid/extras/dispatch/CMakeLists.txt
qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch.h
qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/amqp.h
qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/router.h
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/adapter.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/binding.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/configuration.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/data.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/link.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/mobile.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/neighbor.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/node.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/path.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/router_engine.py
qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/routing.py
qpid/trunk/qpid/extras/dispatch/router/src/main.c
qpid/trunk/qpid/extras/dispatch/src/agent.c
qpid/trunk/qpid/extras/dispatch/src/message_private.h
qpid/trunk/qpid/extras/dispatch/src/python_embedded.c
qpid/trunk/qpid/extras/dispatch/src/router_node.c
qpid/trunk/qpid/extras/dispatch/tests/router_engine_test.py
Modified: qpid/trunk/qpid/extras/dispatch/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/CMakeLists.txt?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/CMakeLists.txt (original)
+++ qpid/trunk/qpid/extras/dispatch/CMakeLists.txt Thu Sep 26 21:14:59 2013
@@ -61,13 +61,6 @@ if (PYTHONLIBS_FOUND)
OUTPUT_STRIP_TRAILING_WHITESPACE)
endif (PYTHONLIBS_FOUND)
-include_directories(
- ${CMAKE_CURRENT_SOURCE_DIR}/include
- ${CMAKE_CURRENT_SOURCE_DIR}/src
- ${proton_include}
- ${PYTHON_INCLUDE_PATH}
- )
-
##
## Find dependencies
##
@@ -76,6 +69,13 @@ find_library(pthread_lib pthread)
find_library(rt_lib rt)
find_path(proton_include proton/driver.h)
+include_directories(
+ ${CMAKE_CURRENT_SOURCE_DIR}/include
+ ${CMAKE_CURRENT_SOURCE_DIR}/src
+ ${proton_include}
+ ${PYTHON_INCLUDE_PATH}
+ )
+
set(CMAKE_C_FLAGS "-pthread -Wall -Werror -std=gnu99")
set(CATCH_UNDEFINED "-Wl,--no-undefined")
@@ -85,6 +85,8 @@ set(CATCH_UNDEFINED "-Wl,--no-undefined"
set(server_SOURCES
src/agent.c
src/alloc.c
+ src/amqp.c
+ src/bitmask.c
src/buffer.c
src/compose.c
src/config.c
Modified: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch.h?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch.h (original)
+++ qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch.h Thu Sep 26 21:14:59 2013
@@ -20,6 +20,7 @@
*/
#include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/bitmask.h>
#include <qpid/dispatch/buffer.h>
#include <qpid/dispatch/ctools.h>
#include <qpid/dispatch/hash.h>
Modified: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/amqp.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/amqp.h?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/amqp.h (original)
+++ qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/amqp.h Thu Sep 26 21:14:59 2013
@@ -76,5 +76,23 @@
#define DX_AMQP_ARRAY8 0xe0
#define DX_AMQP_ARRAY32 0xf0
+/**
+ * Delivery Annotation Headers
+ */
+const char * const DX_DA_INGRESS; // Ingress Router
+const char * const DX_DA_TRACE; // Trace
+const char * const DX_DA_TO; // To-Override
+
+/**
+ * Link Terminus Capabilities
+ */
+const char * const DX_CAPABILITY_ROUTER;
+
+/**
+ * Miscellaneous Strings
+ */
+const char * const DX_INTERNODE_LINK_NAME_1;
+const char * const DX_INTERNODE_LINK_NAME_2;
+
#endif
Added: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h?rev=1526694&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h (added)
+++ qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h Thu Sep 26 21:14:59 2013
@@ -0,0 +1,37 @@
+#ifndef __dispatch_bitmask_h__
+#define __dispatch_bitmask_h__ 1
+/*
+ * 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.
+ */
+
+typedef struct dx_bitmask_t dx_bitmask_t;
+
+int dx_bitmask_width();
+dx_bitmask_t *dx_bitmask(int initial);
+void dx_bitmask_free(dx_bitmask_t *b);
+void dx_bitmask_set_all(dx_bitmask_t *b);
+void dx_bitmask_clear_all(dx_bitmask_t *b);
+void dx_bitmask_set_bit(dx_bitmask_t *b, int bitnum);
+void dx_bitmask_clear_bit(dx_bitmask_t *b, int bitnum);
+int dx_bitmask_value(dx_bitmask_t *b, int bitnum);
+int dx_bitmask_first_set(dx_bitmask_t *b, int *bitnum);
+
+
+
+#endif
+
Propchange: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/bitmask.h
------------------------------------------------------------------------------
svn:keywords = Author Date Id Rev URL
Modified: qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/router.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/router.h?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/router.h (original)
+++ qpid/trunk/qpid/extras/dispatch/include/qpid/dispatch/router.h Thu Sep 26 21:14:59 2013
@@ -27,7 +27,7 @@
typedef struct dx_address_t dx_address_t;
-typedef void (*dx_router_message_cb)(void *context, dx_message_t *msg);
+typedef void (*dx_router_message_cb)(void *context, dx_message_t *msg, int link_id);
const char *dx_router_id(const dx_dispatch_t *dx);
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/adapter.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/adapter.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/adapter.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/adapter.py Thu Sep 26 21:14:59 2013
@@ -18,82 +18,82 @@
#
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
ENTRY_OLD = 1
ENTRY_CURRENT = 2
ENTRY_NEW = 3
class AdapterEngine(object):
- """
- This module is responsible for managing the Adapter's key bindings (list of address-subject:next-hop).
- Key binding lists are kept in disjoint key-classes that can come from different parts of the router
- (i.e. topological keys for inter-router communication and mobile keys for end users).
-
- For each key-class, a mirror copy of what the adapter has is kept internally. This allows changes to the
- routing tables to be efficiently communicated to the adapter in the form of table deltas.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.key_classes = {} # map [key_class] => (addr-key, next-hop)
-
-
- def tick(self, now):
- """
- There is no periodic processing needed for this module.
"""
- pass
+ This module is responsible for managing the Adapter's key bindings (list of address-subject:next-hop).
+ Key binding lists are kept in disjoint key-classes that can come from different parts of the router
+ (i.e. topological keys for inter-router communication and mobile keys for end users).
-
- def remote_routes_changed(self, key_class, new_table):
- old_table = []
- if key_class in self.key_classes:
- old_table = self.key_classes[key_class]
-
- # flag all of the old entries
- old_flags = {}
- for a,b in old_table:
- old_flags[(a,b)] = ENTRY_OLD
-
- # flag the new entries
- new_flags = {}
- for a,b in new_table:
- new_flags[(a,b)] = ENTRY_NEW
-
- # calculate the differences from old to new
- for a,b in new_table:
- if old_table.count((a,b)) > 0:
- old_flags[(a,b)] = ENTRY_CURRENT
- new_flags[(a,b)] = ENTRY_CURRENT
-
- # make to_add and to_delete lists
- to_add = []
- to_delete = []
- for (a,b),f in old_flags.items():
- if f == ENTRY_OLD:
- to_delete.append((a,b))
- for (a,b),f in new_flags.items():
- if f == ENTRY_NEW:
- to_add.append((a,b))
-
- # set the routing table to the new contents
- self.key_classes[key_class] = new_table
-
- # update the adapter's routing tables
- # Note: Do deletions before adds to avoid overlapping routes that may cause
- # messages to be duplicated. It's better to have gaps in the routing
- # tables momentarily because unroutable messages are stored for retry.
- for a,b in to_delete:
- self.container.router_adapter.remote_unbind(a, b)
- for a,b in to_add:
- self.container.router_adapter.remote_bind(a, b)
-
- self.container.log(LOG_INFO, "New Routing Table (class=%s):" % key_class)
- for a,b in new_table:
- self.container.log(LOG_INFO, " %s => %s" % (a, b))
+ For each key-class, a mirror copy of what the adapter has is kept internally. This allows changes to the
+ routing tables to be efficiently communicated to the adapter in the form of table deltas.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.key_classes = {} # map [key_class] => (addr-key, next-hop)
+
+
+ def tick(self, now):
+ """
+ There is no periodic processing needed for this module.
+ """
+ pass
+
+
+ def remote_routes_changed(self, key_class, new_table):
+ old_table = []
+ if key_class in self.key_classes:
+ old_table = self.key_classes[key_class]
+
+ # flag all of the old entries
+ old_flags = {}
+ for a,b in old_table:
+ old_flags[(a,b)] = ENTRY_OLD
+
+ # flag the new entries
+ new_flags = {}
+ for a,b in new_table:
+ new_flags[(a,b)] = ENTRY_NEW
+
+ # calculate the differences from old to new
+ for a,b in new_table:
+ if old_table.count((a,b)) > 0:
+ old_flags[(a,b)] = ENTRY_CURRENT
+ new_flags[(a,b)] = ENTRY_CURRENT
+
+ # make to_add and to_delete lists
+ to_add = []
+ to_delete = []
+ for (a,b),f in old_flags.items():
+ if f == ENTRY_OLD:
+ to_delete.append((a,b))
+ for (a,b),f in new_flags.items():
+ if f == ENTRY_NEW:
+ to_add.append((a,b))
+
+ # set the routing table to the new contents
+ self.key_classes[key_class] = new_table
+
+ # update the adapter's routing tables
+ # Note: Do deletions before adds to avoid overlapping routes that may cause
+ # messages to be duplicated. It's better to have gaps in the routing
+ # tables momentarily because unroutable messages are stored for retry.
+ for a,b in to_delete:
+ self.container.router_adapter.remote_unbind(a, b)
+ for a,b in to_add:
+ self.container.router_adapter.remote_bind(a, b)
+
+ self.container.log(LOG_INFO, "New Routing Table (class=%s):" % key_class)
+ for a,b in new_table:
+ self.container.log(LOG_INFO, " %s => %s" % (a, b))
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/binding.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/binding.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/binding.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/binding.py Thu Sep 26 21:14:59 2013
@@ -18,116 +18,117 @@
#
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
class BindingEngine(object):
- """
- This module is responsible for responding to two different events:
+ """
+ This module is responsible for responding to two different events:
1) The learning of new remote mobile addresses
2) The change of topology (i.e. different next-hops for remote routers)
- When these occur, this module converts the mobile routing table (address => router)
- to a next-hop routing table (address => next-hop), compresses the keys in case there
- are wild-card overlaps, and notifies outbound of changes in the "mobile-key" address class.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.current_keys = {}
-
-
- def tick(self, now):
- pass
-
-
- def mobile_keys_changed(self, keys):
- self.current_keys = keys
- next_hop_keys = self._convert_ids_to_next_hops(keys)
- routing_table = self._compress_keys(next_hop_keys)
- self.container.remote_routes_changed('mobile-key', routing_table)
-
-
- def next_hops_changed(self):
- next_hop_keys = self._convert_ids_to_next_hops(self.current_keys)
- routing_table = self._compress_keys(next_hop_keys)
- self.container.remote_routes_changed('mobile-key', routing_table)
-
-
- def _convert_ids_to_next_hops(self, keys):
- next_hops = self.container.get_next_hops()
- new_keys = {}
- for _id, value in keys.items():
- if _id in next_hops:
- next_hop = next_hops[_id]
- if next_hop not in new_keys:
- new_keys[next_hop] = []
- new_keys[next_hop].extend(value)
- return new_keys
-
-
- def _compress_keys(self, keys):
- trees = {}
- for _id, key_list in keys.items():
- trees[_id] = TopicElementList()
- for key in key_list:
- trees[_id].add_key(key)
- routing_table = []
- for _id, tree in trees.items():
- tree_keys = tree.get_list()
- for tk in tree_keys:
- routing_table.append((tk, _id))
- return routing_table
+ When these occur, this module converts the mobile routing table (address => router)
+ to a next-hop routing table (address => next-hop), compresses the keys in case there
+ are wild-card overlaps, and notifies outbound of changes in the "mobile-key" address class.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.current_keys = {}
+
+
+ def tick(self, now):
+ pass
+
+
+ def mobile_keys_changed(self, keys):
+ self.current_keys = keys
+ next_hop_keys = self._convert_ids_to_next_hops(keys)
+ routing_table = self._compress_keys(next_hop_keys)
+ self.container.remote_routes_changed('mobile-key', routing_table)
+
+
+ def next_hops_changed(self):
+ next_hop_keys = self._convert_ids_to_next_hops(self.current_keys)
+ routing_table = self._compress_keys(next_hop_keys)
+ self.container.remote_routes_changed('mobile-key', routing_table)
+
+
+ def _convert_ids_to_next_hops(self, keys):
+ next_hops = self.container.get_next_hops()
+ new_keys = {}
+ for _id, value in keys.items():
+ if _id in next_hops:
+ next_hop = next_hops[_id]
+ if next_hop not in new_keys:
+ new_keys[next_hop] = []
+ new_keys[next_hop].extend(value)
+ return new_keys
+
+
+ def _compress_keys(self, keys):
+ trees = {}
+ for _id, key_list in keys.items():
+ trees[_id] = TopicElementList()
+ for key in key_list:
+ trees[_id].add_key(key)
+ routing_table = []
+ for _id, tree in trees.items():
+ tree_keys = tree.get_list()
+ for tk in tree_keys:
+ routing_table.append((tk, _id))
+ return routing_table
class TopicElementList(object):
- """
- """
- def __init__(self):
- self.elements = {} # map text => (terminal, sub-list)
-
- def __repr__(self):
- return "%r" % self.elements
-
- def add_key(self, key):
- self.add_tokens(key.split('.'))
-
- def add_tokens(self, tokens):
- first = tokens.pop(0)
- terminal = len(tokens) == 0
-
- if terminal and first == '#':
- ## Optimization #1A (A.B.C.D followed by A.B.#)
- self.elements = {'#':(True, TopicElementList())}
- return
-
- if '#' in self.elements:
- _t,_el = self.elements['#']
- if _t:
- ## Optimization #1B (A.B.# followed by A.B.C.D)
- return
-
- if first not in self.elements:
- self.elements[first] = (terminal, TopicElementList())
- else:
- _t,_el = self.elements[first]
- if terminal and not _t:
- self.elements[first] = (terminal, _el)
-
- if not terminal:
- _t,_el = self.elements[first]
- _el.add_tokens(tokens)
-
- def get_list(self):
- keys = []
- for token, (_t,_el) in self.elements.items():
- if _t: keys.append(token)
- _el.build_list(token, keys)
- return keys
-
- def build_list(self, prefix, keys):
- for token, (_t,_el) in self.elements.items():
- if _t: keys.append("%s.%s" % (prefix, token))
- _el.build_list("%s.%s" % (prefix, token), keys)
+ """
+ """
+ def __init__(self):
+ self.elements = {} # map text => (terminal, sub-list)
+
+ def __repr__(self):
+ return "%r" % self.elements
+
+ def add_key(self, key):
+ self.add_tokens(key.split('.'))
+
+ def add_tokens(self, tokens):
+ first = tokens.pop(0)
+ terminal = len(tokens) == 0
+
+ if terminal and first == '#':
+ ## Optimization #1A (A.B.C.D followed by A.B.#)
+ self.elements = {'#':(True, TopicElementList())}
+ return
+
+ if '#' in self.elements:
+ _t,_el = self.elements['#']
+ if _t:
+ ## Optimization #1B (A.B.# followed by A.B.C.D)
+ return
+
+ if first not in self.elements:
+ self.elements[first] = (terminal, TopicElementList())
+ else:
+ _t,_el = self.elements[first]
+ if terminal and not _t:
+ self.elements[first] = (terminal, _el)
+
+ if not terminal:
+ _t,_el = self.elements[first]
+ _el.add_tokens(tokens)
+
+ def get_list(self):
+ keys = []
+ for token, (_t,_el) in self.elements.items():
+ if _t: keys.append(token)
+ _el.build_list(token, keys)
+ return keys
+
+ def build_list(self, prefix, keys):
+ for token, (_t,_el) in self.elements.items():
+ if _t: keys.append("%s.%s" % (prefix, token))
+ _el.build_list("%s.%s" % (prefix, token), keys)
+
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/configuration.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/configuration.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/configuration.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/configuration.py Thu Sep 26 21:14:59 2013
@@ -18,30 +18,30 @@
#
class Configuration(object):
- """
- This module manages and holds the configuration and tuning parameters for a router.
- """
- def __init__(self, overrides={}):
- ##
- ## Load default values
- ##
- self.values = { 'hello_interval' : 1.0,
- 'hello_max_age' : 3.0,
- 'ra_interval' : 30.0,
- 'remote_ls_max_age' : 60.0,
- 'mobile_addr_max_age' : 60.0 }
+ """
+ This module manages and holds the configuration and tuning parameters for a router.
+ """
+ def __init__(self, overrides={}):
+ ##
+ ## Load default values
+ ##
+ self.values = { 'hello_interval' : 1.0,
+ 'hello_max_age' : 3.0,
+ 'ra_interval' : 30.0,
+ 'remote_ls_max_age' : 60.0,
+ 'mobile_addr_max_age' : 60.0 }
- ##
- ## Apply supplied overrides
- ##
- for k, v in overrides.items():
- self.values[k] = v
+ ##
+ ## Apply supplied overrides
+ ##
+ for k, v in overrides.items():
+ self.values[k] = v
- def __getattr__(self, key):
- if key in self.values:
- return self.values[key]
- raise KeyError
+ def __getattr__(self, key):
+ if key in self.values:
+ return self.values[key]
+ raise KeyError
- def __repr__(self):
- return "%r" % self.values
+ def __repr__(self):
+ return "%r" % self.values
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/data.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/data.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/data.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/data.py Thu Sep 26 21:14:59 2013
@@ -19,257 +19,257 @@
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
def getMandatory(data, key, cls=None):
- """
- Get the value mapped to the requested key. If it's not present, raise an exception.
- """
- if key in data:
- value = data[key]
- if cls and value.__class__ != cls:
- raise Exception("Protocol field has wrong data type: '%s' type=%r expected=%r" % (key, value.__class__, cls))
- return value
- raise Exception("Mandatory protocol field missing: '%s'" % key)
+ """
+ Get the value mapped to the requested key. If it's not present, raise an exception.
+ """
+ if key in data:
+ value = data[key]
+ if cls and value.__class__ != cls:
+ raise Exception("Protocol field has wrong data type: '%s' type=%r expected=%r" % (key, value.__class__, cls))
+ return value
+ raise Exception("Mandatory protocol field missing: '%s'" % key)
def getOptional(data, key, default=None, cls=None):
- """
- Get the value mapped to the requested key. If it's not present, return the default value.
- """
- if key in data:
- value = data[key]
- if cls and value.__class__ != cls:
- raise Exception("Protocol field has wrong data type: '%s' type=%r expected=%r" % (key, value.__class__, cls))
- return value
- return default
+ """
+ Get the value mapped to the requested key. If it's not present, return the default value.
+ """
+ if key in data:
+ value = data[key]
+ if cls and value.__class__ != cls:
+ raise Exception("Protocol field has wrong data type: '%s' type=%r expected=%r" % (key, value.__class__, cls))
+ return value
+ return default
class LinkState(object):
- """
- The link-state of a single router. The link state consists of a list of neighbor routers reachable from
- the reporting router. The link-state-sequence number is incremented each time the link state changes.
- """
- def __init__(self, body, _id=None, _area=None, _ls_seq=None, _peers=None):
- self.last_seen = 0
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.ls_seq = getMandatory(body, 'ls_seq', long)
- self.peers = getMandatory(body, 'peers', list)
- else:
- self.id = _id
- self.area = _area
- self.ls_seq = long(_ls_seq)
- self.peers = _peers
-
- def __repr__(self):
- return "LS(id=%s area=%s ls_seq=%d peers=%r)" % (self.id, self.area, self.ls_seq, self.peers)
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area,
- 'ls_seq' : self.ls_seq,
- 'peers' : self.peers}
-
- def add_peer(self, _id):
- if self.peers.count(_id) == 0:
- self.peers.append(_id)
- return True
- return False
-
- def del_peer(self, _id):
- if self.peers.count(_id) > 0:
- self.peers.remove(_id)
- return True
- return False
+ """
+ The link-state of a single router. The link state consists of a list of neighbor routers reachable from
+ the reporting router. The link-state-sequence number is incremented each time the link state changes.
+ """
+ def __init__(self, body, _id=None, _area=None, _ls_seq=None, _peers=None):
+ self.last_seen = 0
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.ls_seq = getMandatory(body, 'ls_seq', long)
+ self.peers = getMandatory(body, 'peers', list)
+ else:
+ self.id = _id
+ self.area = _area
+ self.ls_seq = long(_ls_seq)
+ self.peers = _peers
+
+ def __repr__(self):
+ return "LS(id=%s area=%s ls_seq=%d peers=%r)" % (self.id, self.area, self.ls_seq, self.peers)
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area,
+ 'ls_seq' : self.ls_seq,
+ 'peers' : self.peers}
+
+ def add_peer(self, _id):
+ if self.peers.count(_id) == 0:
+ self.peers.append(_id)
+ return True
+ return False
+
+ def del_peer(self, _id):
+ if self.peers.count(_id) > 0:
+ self.peers.remove(_id)
+ return True
+ return False
- def bump_sequence(self):
- self.ls_seq += 1
+ def bump_sequence(self):
+ self.ls_seq += 1
class MessageHELLO(object):
- """
- HELLO Message
- scope: neighbors only - HELLO messages travel at most one hop
- This message is used by directly connected routers to determine with whom they have
- bidirectional connectivity.
- """
- def __init__(self, body, _id=None, _area=None, _seen_peers=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.seen_peers = getMandatory(body, 'seen', list)
- else:
- self.id = _id
- self.area = _area
- self.seen_peers = _seen_peers
-
- def __repr__(self):
- return "HELLO(id=%s area=%s seen=%r)" % (self.id, self.area, self.seen_peers)
-
- def get_opcode(self):
- return 'HELLO'
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area,
- 'seen' : self.seen_peers}
+ """
+ HELLO Message
+ scope: neighbors only - HELLO messages travel at most one hop
+ This message is used by directly connected routers to determine with whom they have
+ bidirectional connectivity.
+ """
+ def __init__(self, body, _id=None, _area=None, _seen_peers=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.seen_peers = getMandatory(body, 'seen', list)
+ else:
+ self.id = _id
+ self.area = _area
+ self.seen_peers = _seen_peers
+
+ def __repr__(self):
+ return "HELLO(id=%s area=%s seen=%r)" % (self.id, self.area, self.seen_peers)
+
+ def get_opcode(self):
+ return 'HELLO'
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area,
+ 'seen' : self.seen_peers}
- def is_seen(self, _id):
- return self.seen_peers.count(_id) > 0
+ def is_seen(self, _id):
+ return self.seen_peers.count(_id) > 0
class MessageRA(object):
- """
- Router Advertisement (RA) Message
- scope: all routers in the area and all designated routers
- This message is sent periodically to indicate the originating router's sequence numbers
- for link-state and mobile-address-state.
- """
- def __init__(self, body, _id=None, _area=None, _ls_seq=None, _mobile_seq=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.ls_seq = getMandatory(body, 'ls_seq', long)
- self.mobile_seq = getMandatory(body, 'mobile_seq', long)
- else:
- self.id = _id
- self.area = _area
- self.ls_seq = long(_ls_seq)
- self.mobile_seq = long(_mobile_seq)
-
- def get_opcode(self):
- return 'RA'
-
- def __repr__(self):
- return "RA(id=%s area=%s ls_seq=%d mobile_seq=%d)" % \
- (self.id, self.area, self.ls_seq, self.mobile_seq)
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area,
- 'ls_seq' : self.ls_seq,
- 'mobile_seq' : self.mobile_seq}
+ """
+ Router Advertisement (RA) Message
+ scope: all routers in the area and all designated routers
+ This message is sent periodically to indicate the originating router's sequence numbers
+ for link-state and mobile-address-state.
+ """
+ def __init__(self, body, _id=None, _area=None, _ls_seq=None, _mobile_seq=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.ls_seq = getMandatory(body, 'ls_seq', long)
+ self.mobile_seq = getMandatory(body, 'mobile_seq', long)
+ else:
+ self.id = _id
+ self.area = _area
+ self.ls_seq = long(_ls_seq)
+ self.mobile_seq = long(_mobile_seq)
+
+ def get_opcode(self):
+ return 'RA'
+
+ def __repr__(self):
+ return "RA(id=%s area=%s ls_seq=%d mobile_seq=%d)" % \
+ (self.id, self.area, self.ls_seq, self.mobile_seq)
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area,
+ 'ls_seq' : self.ls_seq,
+ 'mobile_seq' : self.mobile_seq}
class MessageLSU(object):
- """
- """
- def __init__(self, body, _id=None, _area=None, _ls_seq=None, _ls=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.ls_seq = getMandatory(body, 'ls_seq', long)
- self.ls = LinkState(getMandatory(body, 'ls', dict))
- else:
- self.id = _id
- self.area = _area
- self.ls_seq = long(_ls_seq)
- self.ls = _ls
-
- def get_opcode(self):
- return 'LSU'
-
- def __repr__(self):
- return "LSU(id=%s area=%s ls_seq=%d ls=%r)" % \
- (self.id, self.area, self.ls_seq, self.ls)
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area,
- 'ls_seq' : self.ls_seq,
- 'ls' : self.ls.to_dict()}
+ """
+ """
+ def __init__(self, body, _id=None, _area=None, _ls_seq=None, _ls=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.ls_seq = getMandatory(body, 'ls_seq', long)
+ self.ls = LinkState(getMandatory(body, 'ls', dict))
+ else:
+ self.id = _id
+ self.area = _area
+ self.ls_seq = long(_ls_seq)
+ self.ls = _ls
+
+ def get_opcode(self):
+ return 'LSU'
+
+ def __repr__(self):
+ return "LSU(id=%s area=%s ls_seq=%d ls=%r)" % \
+ (self.id, self.area, self.ls_seq, self.ls)
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area,
+ 'ls_seq' : self.ls_seq,
+ 'ls' : self.ls.to_dict()}
class MessageLSR(object):
- """
- """
- def __init__(self, body, _id=None, _area=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- else:
- self.id = _id
- self.area = _area
-
- def get_opcode(self):
- return 'LSR'
-
- def __repr__(self):
- return "LSR(id=%s area=%s)" % (self.id, self.area)
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area}
+ """
+ """
+ def __init__(self, body, _id=None, _area=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ else:
+ self.id = _id
+ self.area = _area
+
+ def get_opcode(self):
+ return 'LSR'
+
+ def __repr__(self):
+ return "LSR(id=%s area=%s)" % (self.id, self.area)
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area}
class MessageMAU(object):
- """
- """
- def __init__(self, body, _id=None, _area=None, _seq=None, _add_list=None, _del_list=None, _exist_list=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.mobile_seq = getMandatory(body, 'mobile_seq', long)
- self.add_list = getOptional(body, 'add', None, list)
- self.del_list = getOptional(body, 'del', None, list)
- self.exist_list = getOptional(body, 'exist', None, list)
- else:
- self.id = _id
- self.area = _area
- self.mobile_seq = long(_seq)
- self.add_list = _add_list
- self.del_list = _del_list
- self.exist_list = _exist_list
-
- def get_opcode(self):
- return 'MAU'
-
- def __repr__(self):
- _add = ''
- _del = ''
- _exist = ''
- if self.add_list: _add = ' add=%r' % self.add_list
- if self.del_list: _del = ' del=%r' % self.del_list
- if self.exist_list: _exist = ' exist=%r' % self.exist_list
- return "MAU(id=%s area=%s mobile_seq=%d%s%s%s)" % \
- (self.id, self.area, self.mobile_seq, _add, _del, _exist)
-
- def to_dict(self):
- body = { 'id' : self.id,
- 'area' : self.area,
- 'mobile_seq' : self.mobile_seq }
- if self.add_list: body['add'] = self.add_list
- if self.del_list: body['del'] = self.del_list
- if self.exist_list: body['exist'] = self.exist_list
- return body
+ """
+ """
+ def __init__(self, body, _id=None, _area=None, _seq=None, _add_list=None, _del_list=None, _exist_list=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.mobile_seq = getMandatory(body, 'mobile_seq', long)
+ self.add_list = getOptional(body, 'add', None, list)
+ self.del_list = getOptional(body, 'del', None, list)
+ self.exist_list = getOptional(body, 'exist', None, list)
+ else:
+ self.id = _id
+ self.area = _area
+ self.mobile_seq = long(_seq)
+ self.add_list = _add_list
+ self.del_list = _del_list
+ self.exist_list = _exist_list
+
+ def get_opcode(self):
+ return 'MAU'
+
+ def __repr__(self):
+ _add = ''
+ _del = ''
+ _exist = ''
+ if self.add_list: _add = ' add=%r' % self.add_list
+ if self.del_list: _del = ' del=%r' % self.del_list
+ if self.exist_list: _exist = ' exist=%r' % self.exist_list
+ return "MAU(id=%s area=%s mobile_seq=%d%s%s%s)" % \
+ (self.id, self.area, self.mobile_seq, _add, _del, _exist)
+
+ def to_dict(self):
+ body = { 'id' : self.id,
+ 'area' : self.area,
+ 'mobile_seq' : self.mobile_seq }
+ if self.add_list: body['add'] = self.add_list
+ if self.del_list: body['del'] = self.del_list
+ if self.exist_list: body['exist'] = self.exist_list
+ return body
class MessageMAR(object):
- """
- """
- def __init__(self, body, _id=None, _area=None, _have_seq=None):
- if body:
- self.id = getMandatory(body, 'id', str)
- self.area = getMandatory(body, 'area', str)
- self.have_seq = getMandatory(body, 'have_seq', long)
- else:
- self.id = _id
- self.area = _area
- self.have_seq = long(_have_seq)
-
- def get_opcode(self):
- return 'MAR'
-
- def __repr__(self):
- return "MAR(id=%s area=%s have_seq=%d)" % (self.id, self.area, self.have_seq)
-
- def to_dict(self):
- return {'id' : self.id,
- 'area' : self.area,
- 'have_seq' : self.have_seq}
+ """
+ """
+ def __init__(self, body, _id=None, _area=None, _have_seq=None):
+ if body:
+ self.id = getMandatory(body, 'id', str)
+ self.area = getMandatory(body, 'area', str)
+ self.have_seq = getMandatory(body, 'have_seq', long)
+ else:
+ self.id = _id
+ self.area = _area
+ self.have_seq = long(_have_seq)
+
+ def get_opcode(self):
+ return 'MAR'
+
+ def __repr__(self):
+ return "MAR(id=%s area=%s have_seq=%d)" % (self.id, self.area, self.have_seq)
+
+ def to_dict(self):
+ return {'id' : self.id,
+ 'area' : self.area,
+ 'have_seq' : self.have_seq}
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/link.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/link.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/link.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/link.py Thu Sep 26 21:14:59 2013
@@ -21,123 +21,123 @@ from data import MessageRA, MessageLSU,
from time import time
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
class LinkStateEngine(object):
- """
- This module is responsible for running the Link State protocol and maintaining the set
- of link states that are gathered from the domain. It notifies outbound when changes to
- the link-state-collection are detected.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.ra_interval = self.container.config.ra_interval
- self.remote_ls_max_age = self.container.config.remote_ls_max_age
- self.last_ra_time = 0
- self.collection = {}
- self.collection_changed = False
- self.mobile_seq = 0
- self.needed_lsrs = {}
-
-
- def tick(self, now):
- self._expire_ls(now)
- self._send_lsrs()
-
- if now - self.last_ra_time >= self.ra_interval:
- self.last_ra_time = now
- self._send_ra()
-
- if self.collection_changed:
- self.collection_changed = False
- self.container.log(LOG_INFO, "New Link-State Collection:")
- for a,b in self.collection.items():
- self.container.log(LOG_INFO, " %s => %r" % (a, b.peers))
- self.container.ls_collection_changed(self.collection)
-
-
- def handle_ra(self, msg, now):
- if msg.id == self.id:
- return
- if msg.id in self.collection:
- ls = self.collection[msg.id]
- ls.last_seen = now
- if ls.ls_seq < msg.ls_seq:
- self.needed_lsrs[(msg.area, msg.id)] = None
- else:
- self.needed_lsrs[(msg.area, msg.id)] = None
-
-
- def handle_lsu(self, msg, now):
- if msg.id == self.id:
- return
- if msg.id in self.collection:
- ls = self.collection[msg.id]
- if ls.ls_seq < msg.ls_seq:
- ls = msg.ls
- self.collection[msg.id] = ls
+ """
+ This module is responsible for running the Link State protocol and maintaining the set
+ of link states that are gathered from the domain. It notifies outbound when changes to
+ the link-state-collection are detected.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.ra_interval = self.container.config.ra_interval
+ self.remote_ls_max_age = self.container.config.remote_ls_max_age
+ self.last_ra_time = 0
+ self.collection = {}
+ self.collection_changed = False
+ self.mobile_seq = 0
+ self.needed_lsrs = {}
+
+
+ def tick(self, now):
+ self._expire_ls(now)
+ self._send_lsrs()
+
+ if now - self.last_ra_time >= self.ra_interval:
+ self.last_ra_time = now
+ self._send_ra()
+
+ if self.collection_changed:
+ self.collection_changed = False
+ self.container.log(LOG_INFO, "New Link-State Collection:")
+ for a,b in self.collection.items():
+ self.container.log(LOG_INFO, " %s => %r" % (a, b.peers))
+ self.container.ls_collection_changed(self.collection)
+
+
+ def handle_ra(self, msg, now):
+ if msg.id == self.id:
+ return
+ if msg.id in self.collection:
+ ls = self.collection[msg.id]
+ ls.last_seen = now
+ if ls.ls_seq < msg.ls_seq:
+ self.needed_lsrs[(msg.area, msg.id)] = None
+ else:
+ self.needed_lsrs[(msg.area, msg.id)] = None
+
+
+ def handle_lsu(self, msg, now):
+ if msg.id == self.id:
+ return
+ if msg.id in self.collection:
+ ls = self.collection[msg.id]
+ if ls.ls_seq < msg.ls_seq:
+ ls = msg.ls
+ self.collection[msg.id] = ls
+ self.collection_changed = True
+ ls.last_seen = now
+ else:
+ ls = msg.ls
+ 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:
+ if _id not in self.collection:
+ self.needed_lsrs[(msg.area, _id)] = None
+
+
+ def handle_lsr(self, msg, now):
+ if msg.id == self.id:
+ return
+ if self.id not in self.collection:
+ return
+ my_ls = self.collection[self.id]
+ self.container.send('_topo/%s/%s' % (msg.area, msg.id), MessageLSU(None, self.id, self.area, my_ls.ls_seq, my_ls))
+
+
+ def new_local_link_state(self, link_state):
+ self.collection[self.id] = link_state
self.collection_changed = True
- ls.last_seen = now
- else:
- ls = msg.ls
- 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:
- if _id not in self.collection:
- self.needed_lsrs[(msg.area, _id)] = None
-
-
- def handle_lsr(self, msg, now):
- if msg.id == self.id:
- return
- if self.id not in self.collection:
- return
- my_ls = self.collection[self.id]
- self.container.send('_topo/%s/%s' % (msg.area, msg.id), MessageLSU(None, self.id, self.area, my_ls.ls_seq, my_ls))
-
-
- def new_local_link_state(self, link_state):
- self.collection[self.id] = link_state
- self.collection_changed = True
- self._send_ra()
-
-
- def set_mobile_sequence(self, seq):
- self.mobile_seq = seq
-
-
- def get_collection(self):
- return self.collection
-
-
- def _expire_ls(self, now):
- to_delete = []
- for key, ls in self.collection.items():
- if key != self.id and now - ls.last_seen > self.remote_ls_max_age:
- to_delete.append(key)
- 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)
-
-
- def _send_lsrs(self):
- for (_area, _id) in self.needed_lsrs.keys():
- self.container.send('_topo/%s/%s' % (_area, _id), MessageLSR(None, self.id, self.area))
- self.needed_lsrs = {}
-
-
- def _send_ra(self):
- ls_seq = 0
- if self.id in self.collection:
- ls_seq = self.collection[self.id].ls_seq
- self.container.send('_topo/%s/all' % self.area, MessageRA(None, self.id, self.area, ls_seq, self.mobile_seq))
+ self._send_ra()
+
+
+ def set_mobile_sequence(self, seq):
+ self.mobile_seq = seq
+
+
+ def get_collection(self):
+ return self.collection
+
+
+ def _expire_ls(self, now):
+ to_delete = []
+ for key, ls in self.collection.items():
+ if key != self.id and now - ls.last_seen > self.remote_ls_max_age:
+ to_delete.append(key)
+ 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)
+
+
+ def _send_lsrs(self):
+ for (_area, _id) in self.needed_lsrs.keys():
+ self.container.send('_topo/%s/%s' % (_area, _id), MessageLSR(None, self.id, self.area))
+ self.needed_lsrs = {}
+
+
+ def _send_ra(self):
+ ls_seq = 0
+ if self.id in self.collection:
+ ls_seq = self.collection[self.id].ls_seq
+ self.container.send('_topo/%s/all' % self.area, MessageRA(None, self.id, self.area, ls_seq, self.mobile_seq))
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/mobile.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/mobile.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/mobile.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/mobile.py Thu Sep 26 21:14:59 2013
@@ -20,169 +20,169 @@
from data import MessageRA, MessageMAR, MessageMAU
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
class MobileAddressEngine(object):
- """
- This module is responsible for maintaining an up-to-date list of mobile addresses in the domain.
- It runs the Mobile-Address protocol and generates an un-optimized routing table for mobile addresses.
- Note that this routing table maps from the mobile address to the remote router where that address
- is directly bound.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.mobile_addr_max_age = self.container.config.mobile_addr_max_age
- self.mobile_seq = 0
- self.local_keys = []
- self.added_keys = []
- self.deleted_keys = []
- self.remote_lists = {} # map router_id => (sequence, list of keys)
- self.remote_last_seen = {} # map router_id => time of last seen advertizement/update
- self.remote_changed = False
- self.needed_mars = {}
-
-
- def tick(self, now):
- self._expire_remotes(now)
- self._send_mars()
-
- ##
- ## If local keys have changed, collect the changes and send a MAU with the diffs
- ## Note: it is important that the differential-MAU be sent before a RA is sent
- ##
- if len(self.added_keys) > 0 or len(self.deleted_keys) > 0:
- self.mobile_seq += 1
- self.container.send('_topo.%s.all' % self.area,
- MessageMAU(None, self.id, self.area, self.mobile_seq, self.added_keys, self.deleted_keys))
- self.local_keys.extend(self.added_keys)
- for key in self.deleted_keys:
- self.local_keys.remove(key)
- self.added_keys = []
- self.deleted_keys = []
- self.container.mobile_sequence_changed(self.mobile_seq)
-
- ##
- ## If remotes have changed, start the process of updating local bindings
- ##
- if self.remote_changed:
- self.remote_changed = False
- self._update_remote_keys()
-
-
- def add_local_address(self, key):
"""
+ This module is responsible for maintaining an up-to-date list of mobile addresses in the domain.
+ It runs the Mobile-Address protocol and generates an un-optimized routing table for mobile addresses.
+ Note that this routing table maps from the mobile address to the remote router where that address
+ is directly bound.
"""
- if self.local_keys.count(key) == 0:
- if self.added_keys.count(key) == 0:
- self.added_keys.append(key)
- else:
- if self.deleted_keys.count(key) > 0:
- self.deleted_keys.remove(key)
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.mobile_addr_max_age = self.container.config.mobile_addr_max_age
+ self.mobile_seq = 0
+ self.local_keys = []
+ self.added_keys = []
+ self.deleted_keys = []
+ self.remote_lists = {} # map router_id => (sequence, list of keys)
+ self.remote_last_seen = {} # map router_id => time of last seen advertizement/update
+ self.remote_changed = False
+ self.needed_mars = {}
+
+
+ def tick(self, now):
+ self._expire_remotes(now)
+ self._send_mars()
+
+ ##
+ ## If local keys have changed, collect the changes and send a MAU with the diffs
+ ## Note: it is important that the differential-MAU be sent before a RA is sent
+ ##
+ if len(self.added_keys) > 0 or len(self.deleted_keys) > 0:
+ self.mobile_seq += 1
+ self.container.send('_topo.%s.all' % self.area,
+ MessageMAU(None, self.id, self.area, self.mobile_seq, self.added_keys, self.deleted_keys))
+ self.local_keys.extend(self.added_keys)
+ for key in self.deleted_keys:
+ self.local_keys.remove(key)
+ self.added_keys = []
+ self.deleted_keys = []
+ self.container.mobile_sequence_changed(self.mobile_seq)
+
+ ##
+ ## If remotes have changed, start the process of updating local bindings
+ ##
+ if self.remote_changed:
+ self.remote_changed = False
+ self._update_remote_keys()
+
+
+ def add_local_address(self, key):
+ """
+ """
+ if self.local_keys.count(key) == 0:
+ if self.added_keys.count(key) == 0:
+ self.added_keys.append(key)
+ else:
+ if self.deleted_keys.count(key) > 0:
+ self.deleted_keys.remove(key)
- def del_local_address(self, key):
- """
- """
- if self.local_keys.count(key) > 0:
- if self.deleted_keys.count(key) == 0:
- self.deleted_keys.append(key)
- else:
- if self.added_keys.count(key) > 0:
- self.added_keys.remove(key)
-
-
- def handle_ra(self, msg, now):
- if msg.id == self.id:
- return
-
- if msg.mobile_seq == 0:
- return
-
- if msg.id in self.remote_lists:
- _seq, _list = self.remote_lists[msg.id]
- self.remote_last_seen[msg.id] = now
- if _seq < msg.mobile_seq:
- self.needed_mars[(msg.id, msg.area, _seq)] = None
- else:
- self.needed_mars[(msg.id, msg.area, 0)] = None
-
-
- def handle_mau(self, msg, now):
- ##
- ## If the MAU is differential, we can only use it if its sequence is exactly one greater
- ## than our stored sequence. If not, we will ignore the content and schedule a MAR.
- ##
- ## If the MAU is absolute, we can use it in all cases.
- ##
- if msg.id == self.id:
- return
-
- if msg.exist_list:
- ##
- ## Absolute MAU
- ##
- if msg.id in self.remote_lists:
- _seq, _list = self.remote_lists[msg.id]
- if _seq >= msg.mobile_seq: # ignore duplicates
- return
- self.remote_lists[msg.id] = (msg.mobile_seq, msg.exist_list)
- self.remote_last_seen[msg.id] = now
- self.remote_changed = True
- else:
- ##
- ## Differential MAU
- ##
- if msg.id in self.remote_lists:
- _seq, _list = self.remote_lists[msg.id]
- if _seq == msg.mobile_seq: # ignore duplicates
- return
- self.remote_last_seen[msg.id] = now
- if _seq + 1 == msg.mobile_seq:
- ##
- ## This is one greater than our stored value, incorporate the deltas
- ##
- if msg.add_list and msg.add_list.__class__ == list:
- _list.extend(msg.add_list)
- if msg.del_list and msg.del_list.__class__ == list:
- for key in msg.del_list:
- _list.remove(key)
- self.remote_lists[msg.id] = (msg.mobile_seq, _list)
- self.remote_changed = True
+ def del_local_address(self, key):
+ """
+ """
+ if self.local_keys.count(key) > 0:
+ if self.deleted_keys.count(key) == 0:
+ self.deleted_keys.append(key)
+ else:
+ if self.added_keys.count(key) > 0:
+ self.added_keys.remove(key)
+
+
+ def handle_ra(self, msg, now):
+ if msg.id == self.id:
+ return
+
+ if msg.mobile_seq == 0:
+ return
+
+ if msg.id in self.remote_lists:
+ _seq, _list = self.remote_lists[msg.id]
+ self.remote_last_seen[msg.id] = now
+ if _seq < msg.mobile_seq:
+ self.needed_mars[(msg.id, msg.area, _seq)] = None
+ else:
+ self.needed_mars[(msg.id, msg.area, 0)] = None
+
+
+ def handle_mau(self, msg, now):
+ ##
+ ## If the MAU is differential, we can only use it if its sequence is exactly one greater
+ ## than our stored sequence. If not, we will ignore the content and schedule a MAR.
+ ##
+ ## If the MAU is absolute, we can use it in all cases.
+ ##
+ if msg.id == self.id:
+ return
+
+ if msg.exist_list:
+ ##
+ ## Absolute MAU
+ ##
+ if msg.id in self.remote_lists:
+ _seq, _list = self.remote_lists[msg.id]
+ if _seq >= msg.mobile_seq: # ignore duplicates
+ return
+ self.remote_lists[msg.id] = (msg.mobile_seq, msg.exist_list)
+ self.remote_last_seen[msg.id] = now
+ self.remote_changed = True
else:
- self.needed_mars[(msg.id, msg.area, _seq)] = None
- else:
- self.needed_mars[(msg.id, msg.area, 0)] = None
-
-
- def handle_mar(self, msg, now):
- if msg.id == self.id:
- return
- if msg.have_seq < self.mobile_seq:
- self.container.send('_topo.%s.%s' % (msg.area, msg.id),
- MessageMAU(None, self.id, self.area, self.mobile_seq, None, None, self.local_keys))
-
-
- def _update_remote_keys(self):
- keys = {}
- for _id,(seq,key_list) in self.remote_lists.items():
- keys[_id] = key_list
- self.container.mobile_keys_changed(keys)
-
-
- def _expire_remotes(self, now):
- for _id, t in self.remote_last_seen.items():
- if now - t > self.mobile_addr_max_age:
- self.remote_lists.pop(_id)
- self.remote_last_seen.pop(_id)
- self.remote_changed = True
-
-
- def _send_mars(self):
- for _id, _area, _seq in self.needed_mars.keys():
- self.container.send('_topo.%s.%s' % (_area, _id), MessageMAR(None, self.id, self.area, _seq))
- self.needed_mars = {}
+ ##
+ ## Differential MAU
+ ##
+ if msg.id in self.remote_lists:
+ _seq, _list = self.remote_lists[msg.id]
+ if _seq == msg.mobile_seq: # ignore duplicates
+ return
+ self.remote_last_seen[msg.id] = now
+ if _seq + 1 == msg.mobile_seq:
+ ##
+ ## This is one greater than our stored value, incorporate the deltas
+ ##
+ if msg.add_list and msg.add_list.__class__ == list:
+ _list.extend(msg.add_list)
+ if msg.del_list and msg.del_list.__class__ == list:
+ for key in msg.del_list:
+ _list.remove(key)
+ self.remote_lists[msg.id] = (msg.mobile_seq, _list)
+ self.remote_changed = True
+ else:
+ self.needed_mars[(msg.id, msg.area, _seq)] = None
+ else:
+ self.needed_mars[(msg.id, msg.area, 0)] = None
+
+
+ def handle_mar(self, msg, now):
+ if msg.id == self.id:
+ return
+ if msg.have_seq < self.mobile_seq:
+ self.container.send('_topo.%s.%s' % (msg.area, msg.id),
+ MessageMAU(None, self.id, self.area, self.mobile_seq, None, None, self.local_keys))
+
+
+ def _update_remote_keys(self):
+ keys = {}
+ for _id,(seq,key_list) in self.remote_lists.items():
+ keys[_id] = key_list
+ self.container.mobile_keys_changed(keys)
+
+
+ def _expire_remotes(self, now):
+ for _id, t in self.remote_last_seen.items():
+ if now - t > self.mobile_addr_max_age:
+ self.remote_lists.pop(_id)
+ self.remote_last_seen.pop(_id)
+ self.remote_changed = True
+
+
+ def _send_mars(self):
+ for _id, _area, _seq in self.needed_mars.keys():
+ self.container.send('_topo.%s.%s' % (_area, _id), MessageMAR(None, self.id, self.area, _seq))
+ self.needed_mars = {}
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/neighbor.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/neighbor.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/neighbor.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/neighbor.py Thu Sep 26 21:14:59 2013
@@ -21,63 +21,63 @@ from data import LinkState, MessageHELLO
from time import time
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
class NeighborEngine(object):
- """
- This module is responsible for maintaining this router's link-state. It runs the HELLO protocol
- with the router's neighbors and notifies outbound when the list of neighbors-in-good-standing (the
- link-state) changes.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.last_hello_time = 0.0
- self.hello_interval = container.config.hello_interval
- self.hello_max_age = container.config.hello_max_age
- self.hellos = {}
- self.link_state_changed = False
- self.link_state = LinkState(None, self.id, self.area, 0, [])
-
-
- def tick(self, now):
- self._expire_hellos(now)
-
- if now - self.last_hello_time >= self.hello_interval:
- self.last_hello_time = now
- self.container.send('_local/qdxrouter', MessageHELLO(None, self.id, self.area, self.hellos.keys()))
-
- if self.link_state_changed:
- self.link_state_changed = False
- self.link_state.bump_sequence()
- self.container.local_link_state_changed(self.link_state)
-
-
- def handle_hello(self, msg, now):
- if msg.id == self.id:
- return
- self.hellos[msg.id] = now
- 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
- ##
-
- def _expire_hellos(self, now):
- to_delete = []
- for key, last_seen in self.hellos.items():
- if now - last_seen > self.hello_max_age:
- to_delete.append(key)
- for key in to_delete:
- 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)
+ """
+ This module is responsible for maintaining this router's link-state. It runs the HELLO protocol
+ with the router's neighbors and notifies outbound when the list of neighbors-in-good-standing (the
+ link-state) changes.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.last_hello_time = 0.0
+ self.hello_interval = container.config.hello_interval
+ self.hello_max_age = container.config.hello_max_age
+ self.hellos = {}
+ self.link_state_changed = False
+ self.link_state = LinkState(None, self.id, self.area, 0, [])
+
+
+ def tick(self, now):
+ self._expire_hellos(now)
+
+ if now - self.last_hello_time >= self.hello_interval:
+ self.last_hello_time = now
+ self.container.send('_local/qdxrouter', MessageHELLO(None, self.id, self.area, self.hellos.keys()))
+
+ if self.link_state_changed:
+ self.link_state_changed = False
+ self.link_state.bump_sequence()
+ self.container.local_link_state_changed(self.link_state)
+
+
+ def handle_hello(self, msg, now, link_id):
+ if msg.id == self.id:
+ return
+ self.hellos[msg.id] = now
+ 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, link_id)
+ self.container.log(LOG_INFO, "New neighbor established: %s on link: %d" % (msg.id, link_id))
+ ##
+ ## TODO - Use this function to detect area boundaries
+ ##
+
+ def _expire_hellos(self, now):
+ to_delete = []
+ for key, last_seen in self.hellos.items():
+ if now - last_seen > self.hello_max_age:
+ to_delete.append(key)
+ for key in to_delete:
+ 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)
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/node.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/node.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/node.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/node.py Thu Sep 26 21:14:59 2013
@@ -18,86 +18,131 @@
#
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ 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)
+ """
+ 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.
+
+ This module is also responsible for assigning a unique mask bit value to each router.
+ The mask bit is used in the main router to represent sets of valid destinations for addresses.
+ """
+ def __init__(self, container, max_routers):
+ self.container = container
+ self.max_routers = max_routers
+ self.nodes = {} # id => RemoteNode
+ self.maskbits = []
+ self.next_maskbit = 0
+ for i in range(max_routers):
+ self.maskbits.append(None)
+
+
+ def tick(self, now):
+ pass
+
+
+ def new_neighbor(self, node_id, link_maskbit):
+ """
+ A node, designated by node_id, has been discovered as a neighbor over a link with
+ a maskbit of link_maskbit.
+ """
+ if node_id not in self.nodes:
+ self.nodes[node_id] = RemoteNode(node_id, self._allocate_maskbit())
+ self.nodes[node_id].set_neighbor(link_maskbit)
+ self._notify(self.nodes[node_id])
+
+
+ def lost_neighbor(self, node_id):
+ """
+ We have lost contact with a neighboring node node_id.
+ """
+ node = self.nodes[node_id]
+ node.clear_neighbor()
+ self._notify(node)
+ if node.to_delete():
+ self._free_maskbit(node.maskbit)
+ self.nodes.pop(node_id)
+
+
+ def new_node(self, node_id):
+ """
+ A node, designated by node_id, has been discovered through the an advertisement from a
+ remote peer.
+ """
+ if node_id not in self.nodes:
+ self.nodes[node_id] = RemoteNode(node_id, self._allocate_maskbit())
+ self.nodes[node_id].set_remote()
+ self._notify(self.nodes[node_id])
+
+
+ def lost_node(self, node_id):
+ """
+ A remote node, node_id, has not been heard from for too long and is being deemed lost.
+ """
+ node = self.nodes[node_id]
+ node.clear_remote()
+ self._notify(node)
+ if node.to_delete():
+ self._free_maskbit(node.maskbit)
+ self.nodes.pop(node_id)
+
+
+ def _allocate_maskbit(self):
+ if self.next_maskbit == None:
+ raise Exception("Exceeded Maximum Router Count")
+ result = self.next_maskbit
+ self.next_maskbit = None
+ self.maskbits[result] = True
+ for n in range(result + 1, self.max_routers):
+ if self.maskbits[n] == None:
+ self.next_maskbit = n
+ break
+ return result
+
+
+ def _free_maskbit(self, i):
+ self.maskbits[i] = None
+ if self.next_maskbit == None or i < self.next_maskbit:
+ self.next_maskbit = i
+
+
+ def _notify(self, node):
+ if node.to_delete():
+ self.container.node_updated("R%s" % node.id, 0, 0, 0, 0)
+ else:
+ is_neighbor = 0
+ if node.neighbor:
+ is_neighbor = 1
+ self.container.node_updated("R%s" % node.id, 1, is_neighbor, node.link_maskbit, node.maskbit)
class RemoteNode(object):
- def __init__(self, node_id):
- self.id = node_id
- self.neighbor = None
- self.remote = None
+ def __init__(self, node_id, maskbit):
+ self.id = node_id
+ self.neighbor = None
+ self.link_maskbit = None
+ self.maskbit = maskbit
+ self.remote = None
+
+ def set_neighbor(self, link_maskbit):
+ self.neighbor = True
+ self.link_maskbit = link_maskbit
+
+ def set_remote(self):
+ self.remote = True
+
+ def clear_neighbor(self):
+ self.neighbor = None
+ self.link_maskbit = None
- def set_neighbor(self):
- self.neighbor = True
+ def clear_remote(self):
+ self.remote = None
- 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
+ def to_delete(self):
+ return self.neighbor == None and self.remote == None
Modified: qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/path.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/path.py?rev=1526694&r1=1526693&r2=1526694&view=diff
==============================================================================
--- qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/path.py (original)
+++ qpid/trunk/qpid/extras/dispatch/python/qpid/dispatch/router/path.py Thu Sep 26 21:14:59 2013
@@ -18,185 +18,185 @@
#
try:
- from dispatch import *
+ from dispatch import *
except ImportError:
- from ..stubs import *
+ from ..stubs import *
class PathEngine(object):
- """
- This module is responsible for computing the next-hop for every router/area in the domain
- based on the collection of link states that have been gathered.
- """
- def __init__(self, container):
- self.container = container
- self.id = self.container.id
- self.area = self.container.area
- self.recalculate = False
- self.collection = None
-
-
- def tick(self, now_unused):
- if self.recalculate:
- self.recalculate = False
- self._calculate_routes()
-
-
- def ls_collection_changed(self, collection):
- self.recalculate = True
- self.collection = collection
-
-
- def _calculate_tree_from_root(self, root):
- ##
- ## Make a copy of the current collection of link-states that contains
- ## an empty link-state for nodes that are known-peers but are not in the
- ## collection currently. This is needed to establish routes to those nodes
- ## so we can trade link-state information with them.
- ##
- link_states = {}
- for _id, ls in self.collection.items():
- link_states[_id] = ls.peers
- for p in ls.peers:
- if p not in link_states:
- link_states[p] = []
-
- ##
- ## Setup Dijkstra's Algorithm
- ##
- cost = {}
- prev = {}
- for _id in link_states:
- cost[_id] = None # infinite
- prev[_id] = None # undefined
- cost[root] = 0 # no cost to the root node
- unresolved = NodeSet(cost)
-
- ##
- ## Process unresolved nodes until lowest cost paths to all reachable nodes have been found.
- ##
- while not unresolved.empty():
- u = unresolved.lowest_cost()
- if cost[u] == None:
- # There are no more reachable nodes in unresolved
- break
- for v in link_states[u]:
- if unresolved.contains(v):
- alt = cost[u] + 1 # TODO - Use link cost instead of 1
- if cost[v] == None or alt < cost[v]:
- cost[v] = alt
- prev[v] = u
- unresolved.set_cost(v, alt)
-
- ##
- ## Remove unreachable nodes from the map. Note that this will also remove the
- ## root node (has no previous node) from the map.
- ##
- for u, val in prev.items():
- if not val:
- prev.pop(u)
-
- ##
- ## Return previous-node map. This is a map of all reachable, remote nodes to
- ## their predecessor node.
- ##
- return prev
-
-
- def _calculate_routes(self):
- ##
- ## Generate the shortest-path tree with the local node as root
- ##
- prev = self._calculate_tree_from_root(self.id)
- nodes = prev.keys()
-
- ##
- ## Distill the path tree into a map of next hops for each node
- ##
- next_hops = {}
- while len(nodes) > 0:
- u = nodes[0] # pick any destination
- path = [u]
- nodes.remove(u)
- v = prev[u]
- while v != self.id: # build a list of nodes in the path back to the root
- if v in nodes:
- path.append(v)
- nodes.remove(v)
- u = v
- v = prev[u]
- for w in path: # mark each node in the path as reachable via the next hop
- next_hops[w] = u
-
- ##
- ## TODO - Calculate the tree from each origin, determine the set of origins-per-dest
- ## for which the path from origin to dest passes through us. This is the set
- ## of valid origins for forwarding to the destination.
- ##
+ """
+ This module is responsible for computing the next-hop for every router/area in the domain
+ based on the collection of link states that have been gathered.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.recalculate = False
+ self.collection = None
- self.container.next_hops_changed(next_hops)
+ def tick(self, now_unused):
+ if self.recalculate:
+ self.recalculate = False
+ self._calculate_routes()
-class NodeSet(object):
- """
- This data structure is an ordered list of node IDs, sorted in increasing order by their cost.
- Equal cost nodes are secondarily sorted by their ID in order to provide deterministic and
- repeatable ordering.
- """
- def __init__(self, cost_map):
- self.nodes = []
- for _id, cost in cost_map.items():
- ##
- ## Assume that nodes are either unreachable (cost = None) or local (cost = 0)
- ## during this initialization.
- ##
- if cost == 0:
- self.nodes.insert(0, (_id, cost))
- else:
+
+ def ls_collection_changed(self, collection):
+ self.recalculate = True
+ self.collection = collection
+
+
+ def _calculate_tree_from_root(self, root):
##
- ## There is no need to sort unreachable nodes by ID
+ ## Make a copy of the current collection of link-states that contains
+ ## an empty link-state for nodes that are known-peers but are not in the
+ ## collection currently. This is needed to establish routes to those nodes
+ ## so we can trade link-state information with them.
##
- self.nodes.append((_id, cost))
+ link_states = {}
+ for _id, ls in self.collection.items():
+ link_states[_id] = ls.peers
+ for p in ls.peers:
+ if p not in link_states:
+ link_states[p] = []
+ ##
+ ## Setup Dijkstra's Algorithm
+ ##
+ cost = {}
+ prev = {}
+ for _id in link_states:
+ cost[_id] = None # infinite
+ prev[_id] = None # undefined
+ cost[root] = 0 # no cost to the root node
+ unresolved = NodeSet(cost)
- def __repr__(self):
- return self.nodes.__repr__()
+ ##
+ ## Process unresolved nodes until lowest cost paths to all reachable nodes have been found.
+ ##
+ while not unresolved.empty():
+ u = unresolved.lowest_cost()
+ if cost[u] == None:
+ # There are no more reachable nodes in unresolved
+ break
+ for v in link_states[u]:
+ if unresolved.contains(v):
+ alt = cost[u] + 1 # TODO - Use link cost instead of 1
+ if cost[v] == None or alt < cost[v]:
+ cost[v] = alt
+ prev[v] = u
+ unresolved.set_cost(v, alt)
+ ##
+ ## Remove unreachable nodes from the map. Note that this will also remove the
+ ## root node (has no previous node) from the map.
+ ##
+ for u, val in prev.items():
+ if not val:
+ prev.pop(u)
- def empty(self):
- return len(self.nodes) == 0
+ ##
+ ## Return previous-node map. This is a map of all reachable, remote nodes to
+ ## their predecessor node.
+ ##
+ return prev
- def contains(self, _id):
- for a, b in self.nodes:
- if a == _id:
- return True
- return False
+ def _calculate_routes(self):
+ ##
+ ## Generate the shortest-path tree with the local node as root
+ ##
+ prev = self._calculate_tree_from_root(self.id)
+ nodes = prev.keys()
+ ##
+ ## Distill the path tree into a map of next hops for each node
+ ##
+ next_hops = {}
+ while len(nodes) > 0:
+ u = nodes[0] # pick any destination
+ path = [u]
+ nodes.remove(u)
+ v = prev[u]
+ while v != self.id: # build a list of nodes in the path back to the root
+ if v in nodes:
+ path.append(v)
+ nodes.remove(v)
+ u = v
+ v = prev[u]
+ for w in path: # mark each node in the path as reachable via the next hop
+ next_hops[w] = u
- def lowest_cost(self):
- """
- Remove and return the lowest cost node ID.
- """
- _id, cost = self.nodes.pop(0)
- return _id
+ ##
+ ## TODO - Calculate the tree from each origin, determine the set of origins-per-dest
+ ## for which the path from origin to dest passes through us. This is the set
+ ## of valid origins for forwarding to the destination.
+ ##
+ self.container.next_hops_changed(next_hops)
- def set_cost(self, _id, new_cost):
+class NodeSet(object):
"""
- Set the cost for an ID in the NodeSet and re-insert the ID so that the list
- remains sorted in increasing cost order.
+ This data structure is an ordered list of node IDs, sorted in increasing order by their cost.
+ Equal cost nodes are secondarily sorted by their ID in order to provide deterministic and
+ repeatable ordering.
"""
- index = 0
- for i, c in self.nodes:
- if i == _id:
- break
- index += 1
- self.nodes.pop(index)
-
- index = 0
- for i, c in self.nodes:
- if c == None or new_cost < c or (new_cost == c and _id < i):
- break
- index += 1
+ def __init__(self, cost_map):
+ self.nodes = []
+ for _id, cost in cost_map.items():
+ ##
+ ## Assume that nodes are either unreachable (cost = None) or local (cost = 0)
+ ## during this initialization.
+ ##
+ if cost == 0:
+ self.nodes.insert(0, (_id, cost))
+ else:
+ ##
+ ## There is no need to sort unreachable nodes by ID
+ ##
+ self.nodes.append((_id, cost))
+
+
+ def __repr__(self):
+ return self.nodes.__repr__()
+
+
+ def empty(self):
+ return len(self.nodes) == 0
+
+
+ def contains(self, _id):
+ for a, b in self.nodes:
+ if a == _id:
+ return True
+ return False
+
+
+ def lowest_cost(self):
+ """
+ Remove and return the lowest cost node ID.
+ """
+ _id, cost = self.nodes.pop(0)
+ return _id
+
+
+ def set_cost(self, _id, new_cost):
+ """
+ Set the cost for an ID in the NodeSet and re-insert the ID so that the list
+ remains sorted in increasing cost order.
+ """
+ index = 0
+ for i, c in self.nodes:
+ if i == _id:
+ break
+ index += 1
+ self.nodes.pop(index)
+
+ index = 0
+ for i, c in self.nodes:
+ if c == None or new_cost < c or (new_cost == c and _id < i):
+ break
+ index += 1
+
+ self.nodes.insert(index, (_id, new_cost))
- self.nodes.insert(index, (_id, new_cost))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org