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 2016/04/22 23:55:22 UTC
[1/7] qpid-dispatch git commit: DISPATCH-10 - Added configuration of
inter-router cost up to the point of communicating cost in LSU messages. Cost
is not yet used in computing routes.
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 279c1b4c6 -> ba8e158d6
DISPATCH-10 - Added configuration of inter-router cost up to the point of communicating
cost in LSU messages. Cost is not yet used in computing routes.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/9f232923
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/9f232923
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/9f232923
Branch: refs/heads/master
Commit: 9f232923d120e2421214917268780b4c86cf959d
Parents: 279c1b4
Author: Ted Ross <tr...@redhat.com>
Authored: Wed Apr 20 15:08:23 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:49:43 2016 -0400
----------------------------------------------------------------------
include/qpid/dispatch/router_core.h | 4 +-
include/qpid/dispatch/server.h | 7 +
python/qpid_dispatch/management/qdrouter.json | 7 +
.../display_name/display_name.py | 5 +-
.../qpid_dispatch_internal/management/agent.py | 4 +-
python/qpid_dispatch_internal/router/data.py | 14 +-
python/qpid_dispatch_internal/router/engine.py | 8 +-
python/qpid_dispatch_internal/router/hello.py | 4 +-
python/qpid_dispatch_internal/router/node.py | 8 +-
src/connection_manager.c | 1 +
src/python_embedded.c | 4 +-
src/router_core/connections.c | 2 +
src/router_core/forwarder.c | 3 +-
src/router_core/management_agent.c | 2 +-
src/router_core/router_core_private.h | 4 +-
src/router_node.c | 7 +-
tests/router_engine_test.py | 128 +++++++++----------
17 files changed, 119 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/include/qpid/dispatch/router_core.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index 0291f64..81dd2f9 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -88,7 +88,7 @@ void qdr_core_route_table_handlers(qdr_core_t *core,
* In-process messaging functions
******************************************************************************
*/
-typedef void (*qdr_receive_t) (void *context, qd_message_t *msg, int link_maskbit);
+typedef void (*qdr_receive_t) (void *context, qd_message_t *msg, int link_maskbit, int inter_router_cost);
qdr_subscription_t *qdr_core_subscribe(qdr_core_t *core,
const char *address,
@@ -149,6 +149,7 @@ typedef enum {
* @param core Pointer to the core object
* @param incoming True iff this connection is associated with a listener, False if a connector
* @param role The configured role of this connection
+ * @param cost If the role is inter_router, this is the configured cost for the connection.
* @param label Optional label provided in the connection's configuration. This is used to
* correlate the connection with waypoints and link-route destinations that use the connection.
* @param strip_annotations_in True if configured to remove annotations on inbound messages.
@@ -159,6 +160,7 @@ typedef enum {
qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
bool incoming,
qdr_connection_role_t role,
+ int cost,
const char *label,
const char *remote_container_id,
bool strip_annotations_in,
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/include/qpid/dispatch/server.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/server.h b/include/qpid/dispatch/server.h
index 5a85e63..4683288 100644
--- a/include/qpid/dispatch/server.h
+++ b/include/qpid/dispatch/server.h
@@ -400,6 +400,13 @@ typedef struct qd_server_config_t {
char *role;
/**
+ * If the role is "inter-router", the cost can be set to a number greater than
+ * or equal to one. Inter-router cost is used to influence the routing algorithm
+ * such that it prefers lower-cost paths.
+ */
+ int inter_router_cost;
+
+ /**
* The maximum size of an AMQP frame in octets.
*/
uint32_t max_frame_size;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch/management/qdrouter.json
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index 69d5192..f5cce7a 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -43,6 +43,13 @@
"default": "normal",
"description": "The role of an established connection. In the normal role, the connection is assumed to be used for AMQP clients that are doing normal message delivery over the connection. In the inter-router role, the connection is assumed to be to another router in the network. Inter-router discovery and routing protocols can only be used over inter-router connections. route-container role can be used for router-container connections, for example, a router-broker connection. on-demand role has been deprecated.",
"create": true
+ },
+ "cost": {
+ "type": "integer",
+ "default": "1",
+ "required": false,
+ "create": true,
+ "description": "For the 'inter-router' role only. This value assigns a cost metric to the inter-router connection. The default (and minimum) value is one. Higher values represent higher costs. The cost is used to influence the routing algorithm as it attempts to use the path with the lowest total cost from ingress to egress."
}
}
},
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/display_name/display_name.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/display_name/display_name.py b/python/qpid_dispatch_internal/display_name/display_name.py
index 340e69d..5ecb867 100644
--- a/python/qpid_dispatch_internal/display_name/display_name.py
+++ b/python/qpid_dispatch_internal/display_name/display_name.py
@@ -95,14 +95,15 @@ class DisplayNameService(object):
return body
- def receive(self, message, link_id):
+ def receive(self, message, unused_link_id, unused_cost):
"""
This is the IOAdapter's callback function. Will be invoked when the IOAdapter receives a request.
Will only accept QUERY requests.
Matches the passed in profilename and userid to user name. If a matching user name is not found, returns the
passed in userid as the user name.
:param message:
- :param link_id:
+ :param unused_link_id:
+ :param unused_cost
"""
body = {}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/management/agent.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/management/agent.py b/python/qpid_dispatch_internal/management/agent.py
index 336ef86..b050b8a 100644
--- a/python/qpid_dispatch_internal/management/agent.py
+++ b/python/qpid_dispatch_internal/management/agent.py
@@ -751,7 +751,7 @@ class Agent(object):
except:
self.log(LOG_ERROR, "Can't respond to %s: %s"%(request, format_exc()))
- def receive(self, request, link_id):
+ def receive(self, request, unused_link_id, unused_cost):
"""Called when a management request is received."""
def error(e, trace):
"""Raise an error"""
@@ -766,7 +766,7 @@ class Agent(object):
with self.request_lock:
try:
self.entities.refresh_from_c()
- self.log(LOG_DEBUG, "Agent request %s on link %s"%(request, link_id))
+ self.log(LOG_DEBUG, "Agent request %s"% request)
status, body = self.handle(request)
self.respond(request, status=status, body=body)
except ManagementError, e:
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/router/data.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/data.py b/python/qpid_dispatch_internal/router/data.py
index 040f14b..5b739b8 100644
--- a/python/qpid_dispatch_internal/router/data.py
+++ b/python/qpid_dispatch_internal/router/data.py
@@ -52,7 +52,7 @@ class LinkState(object):
self.id = getMandatory(body, 'id', str)
self.area = '0'
self.ls_seq = getMandatory(body, 'ls_seq', long)
- self.peers = getMandatory(body, 'peers', list)
+ self.peers = getMandatory(body, 'peers', dict)
else:
self.id = _id
self.area = '0'
@@ -68,20 +68,20 @@ class LinkState(object):
'ls_seq' : self.ls_seq,
'peers' : self.peers}
- def add_peer(self, _id):
- if self.peers.count(_id) == 0:
- self.peers.append(_id)
+ def add_peer(self, _id, _cost):
+ if _id not in self.peers:
+ self.peers[_id] = _cost
return True
return False
def del_peer(self, _id):
- if self.peers.count(_id) > 0:
- self.peers.remove(_id)
+ if _id in self.peers:
+ self.peers.pop(_id)
return True
return False
def del_all_peers(self):
- self.peers = []
+ self.peers = {}
self.ls_seq = 0
def has_peers(self):
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/router/engine.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/engine.py b/python/qpid_dispatch_internal/router/engine.py
index 8905bac..5849d09 100644
--- a/python/qpid_dispatch_internal/router/engine.py
+++ b/python/qpid_dispatch_internal/router/engine.py
@@ -126,7 +126,7 @@ class RouterEngine:
except Exception:
self.log(LOG_ERROR, "Exception in timer processing\n%s" % format_exc(LOG_STACK_LIMIT))
- def handleControlMessage(self, opcode, body, link_id):
+ def handleControlMessage(self, opcode, body, link_id, cost):
"""
"""
try:
@@ -134,7 +134,7 @@ class RouterEngine:
if opcode == 'HELLO':
msg = MessageHELLO(body)
self.log_hello(LOG_TRACE, "RCVD: %r" % msg)
- self.hello_protocol.handle_hello(msg, now, link_id)
+ self.hello_protocol.handle_hello(msg, now, link_id, cost)
elif opcode == 'RA':
msg = MessageRA(body)
@@ -164,12 +164,12 @@ class RouterEngine:
except Exception:
self.log(LOG_ERROR, "Control message error: opcode=%s body=%r\n%s" % (opcode, body, format_exc(LOG_STACK_LIMIT)))
- def receive(self, message, link_id):
+ def receive(self, message, link_id, cost):
"""
This is the IoAdapter message-receive handler
"""
try:
- self.handleControlMessage(message.properties['opcode'], message.body, link_id)
+ self.handleControlMessage(message.properties['opcode'], message.body, link_id, cost)
except Exception:
self.log(LOG_ERROR, "Exception in raw message processing: properties=%r body=%r\n%s" %
(message.properties, message.body, format_exc(LOG_STACK_LIMIT)))
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/router/hello.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/hello.py b/python/qpid_dispatch_internal/router/hello.py
index 6dbed13..0b214a9 100644
--- a/python/qpid_dispatch_internal/router/hello.py
+++ b/python/qpid_dispatch_internal/router/hello.py
@@ -44,7 +44,7 @@ class HelloProtocol(object):
self.container.log_hello(LOG_TRACE, "SENT: %r" % msg)
- def handle_hello(self, msg, now, link_id):
+ def handle_hello(self, msg, now, link_id, cost):
if msg.id == self.id:
if not self.dup_reported and (msg.instance != self.container.instance):
self.dup_reported = True
@@ -52,7 +52,7 @@ class HelloProtocol(object):
return
self.hellos[msg.id] = now
if msg.is_seen(self.id):
- self.node_tracker.neighbor_refresh(msg.id, msg.instance, link_id, now)
+ self.node_tracker.neighbor_refresh(msg.id, msg.instance, link_id, cost, now)
def _expire_hellos(self, now):
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/python/qpid_dispatch_internal/router/node.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/node.py b/python/qpid_dispatch_internal/router/node.py
index 0719898..7ae08fd 100644
--- a/python/qpid_dispatch_internal/router/node.py
+++ b/python/qpid_dispatch_internal/router/node.py
@@ -33,7 +33,7 @@ class NodeTracker(object):
self.container = container
self.my_id = container.id
self.max_routers = max_routers
- self.link_state = LinkState(None, self.my_id, 0, [])
+ self.link_state = LinkState(None, self.my_id, 0, {})
self.link_state_changed = False
self.recompute_topology = False
self.last_topology_change = 0
@@ -184,7 +184,7 @@ class NodeTracker(object):
self.container.link_state_engine.send_ra(now)
- def neighbor_refresh(self, node_id, instance, link_id, now):
+ def neighbor_refresh(self, node_id, instance, link_id, cost, now):
"""
Invoked when the hello protocol has received positive confirmation
of continued bi-directional connectivity with a neighbor router.
@@ -204,7 +204,7 @@ class NodeTracker(object):
if node.set_link_id(link_id):
self.nodes_by_link_id[link_id] = node
node.request_link_state()
- if self.link_state.add_peer(node_id):
+ if self.link_state.add_peer(node_id, cost):
self.link_state_changed = True
##
@@ -364,7 +364,7 @@ class RouterNode(object):
self.maskbit = self.parent._allocate_maskbit()
self.neighbor_refresh_time = 0.0
self.peer_link_id = None
- self.link_state = LinkState(None, self.id, 0, [])
+ self.link_state = LinkState(None, self.id, 0, {})
self.next_hop_router = None
self.valid_origins = None
self.mobile_addresses = []
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/connection_manager.c
----------------------------------------------------------------------
diff --git a/src/connection_manager.c b/src/connection_manager.c
index 44a22a9..66466f1 100644
--- a/src/connection_manager.c
+++ b/src/connection_manager.c
@@ -139,6 +139,7 @@ static qd_error_t load_server_config(qd_dispatch_t *qd, qd_server_config_t *conf
config->port = qd_entity_get_string(entity, "port"); CHECK();
config->name = qd_entity_opt_string(entity, "name", 0); CHECK();
config->role = qd_entity_get_string(entity, "role"); CHECK();
+ config->inter_router_cost = qd_entity_opt_long(entity, "cost", 1); CHECK();
config->protocol_family = qd_entity_opt_string(entity, "protocolFamily", 0); CHECK();
config->max_frame_size = qd_entity_get_long(entity, "maxFrameSize"); CHECK();
config->idle_timeout_seconds = qd_entity_get_long(entity, "idleTimeoutSeconds"); CHECK();
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/python_embedded.c
----------------------------------------------------------------------
diff --git a/src/python_embedded.c b/src/python_embedded.c
index 4946856..2d5d50b 100644
--- a/src/python_embedded.c
+++ b/src/python_embedded.c
@@ -488,7 +488,7 @@ static qd_error_t iter_to_py_attr(qd_field_iterator_t *iter,
return qd_error_code();
}
-static void qd_io_rx_handler(void *context, qd_message_t *msg, int link_id)
+static void qd_io_rx_handler(void *context, qd_message_t *msg, int link_id, int inter_router_cost)
{
IoAdapter *self = (IoAdapter*) context;
@@ -513,7 +513,7 @@ static void qd_io_rx_handler(void *context, qd_message_t *msg, int link_id)
iter_to_py_attr(qd_message_field_iterator(msg, QD_FIELD_APPLICATION_PROPERTIES), py_iter_parse, py_msg, "properties");
iter_to_py_attr(qd_message_field_iterator(msg, QD_FIELD_BODY), py_iter_parse, py_msg, "body");
- PyObject *value = PyObject_CallFunction(self->handler, "Ol", py_msg, link_id);
+ PyObject *value = PyObject_CallFunction(self->handler, "Oll", py_msg, link_id, inter_router_cost);
Py_DECREF(py_msg);
Py_XDECREF(value);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/router_core/connections.c
----------------------------------------------------------------------
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 88de922..0a4f1ed 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -58,6 +58,7 @@ qdr_terminus_t *qdr_terminus_router_data(void)
qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
bool incoming,
qdr_connection_role_t role,
+ int cost,
const char *label,
const char *remote_container_id,
bool strip_annotations_in,
@@ -72,6 +73,7 @@ qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
conn->user_context = 0;
conn->incoming = incoming;
conn->role = role;
+ conn->inter_router_cost = cost;
conn->strip_annotations_in = strip_annotations_in;
conn->strip_annotations_out = strip_annotations_out;
conn->link_capacity = link_capacity;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/router_core/forwarder.c
----------------------------------------------------------------------
diff --git a/src/router_core/forwarder.c b/src/router_core/forwarder.c
index a012875..274beb0 100644
--- a/src/router_core/forwarder.c
+++ b/src/router_core/forwarder.c
@@ -118,7 +118,7 @@ void qdr_forward_deliver_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *
void qdr_forward_on_message(qdr_core_t *core, qdr_general_work_t *work)
{
- work->on_message(work->on_message_context, work->msg, work->maskbit);
+ work->on_message(work->on_message_context, work->msg, work->maskbit, work->inter_router_cost);
qd_message_free(work->msg);
}
@@ -130,6 +130,7 @@ void qdr_forward_on_message_CT(qdr_core_t *core, qdr_subscription_t *sub, qdr_li
work->on_message_context = sub->on_message_context;
work->msg = qd_message_copy(msg);
work->maskbit = link ? link->conn->mask_bit : 0;
+ work->inter_router_cost = link ? link->conn->inter_router_cost : 1;
qdr_post_general_work_CT(core, work);
}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/router_core/management_agent.c
----------------------------------------------------------------------
diff --git a/src/router_core/management_agent.c b/src/router_core/management_agent.c
index e048c85..0eef56a 100644
--- a/src/router_core/management_agent.c
+++ b/src/router_core/management_agent.c
@@ -442,7 +442,7 @@ static bool qd_can_handle_request(qd_parsed_field_t *properties_fld,
* Handler for the management agent.
*
*/
-void qdr_management_agent_on_message(void *context, qd_message_t *msg, int unused_link_id)
+void qdr_management_agent_on_message(void *context, qd_message_t *msg, int unused_link_id, int unused_cost)
{
qdr_core_t *core = (qdr_core_t*) context;
qd_field_iterator_t *app_properties_iter = qd_message_field_iterator(msg, QD_FIELD_APPLICATION_PROPERTIES);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/router_core/router_core_private.h
----------------------------------------------------------------------
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 62b8136..8abfba8 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -368,6 +368,7 @@ struct qdr_general_work_t {
qdr_general_work_handler_t handler;
qdr_field_t *field;
int maskbit;
+ int inter_router_cost;
qdr_receive_t on_message;
void *on_message_context;
qd_message_t *msg;
@@ -412,6 +413,7 @@ struct qdr_connection_t {
void *user_context;
bool incoming;
qdr_connection_role_t role;
+ int inter_router_cost;
qdr_conn_identifier_t *conn_id;
bool strip_annotations_in;
bool strip_annotations_out;
@@ -562,7 +564,7 @@ struct qdr_core_t {
void *router_core_thread(void *arg);
uint64_t qdr_identifier(qdr_core_t* core);
-void qdr_management_agent_on_message(void *context, qd_message_t *msg, int unused_link_id);
+void qdr_management_agent_on_message(void *context, qd_message_t *msg, int unused_link_id, int unused_cost);
void qdr_route_table_setup_CT(qdr_core_t *core);
void qdr_agent_setup_CT(qdr_core_t *core);
void qdr_forwarder_setup_CT(qdr_core_t *core);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/src/router_node.c
----------------------------------------------------------------------
diff --git a/src/router_node.c b/src/router_node.c
index af66278..8cba85e 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -43,6 +43,7 @@ static char *node_id;
*/
static void qd_router_connection_get_config(const qd_connection_t *conn,
qdr_connection_role_t *role,
+ int *cost,
const char **name,
bool *strip_annotations_in,
bool *strip_annotations_out,
@@ -59,6 +60,7 @@ static void qd_router_connection_get_config(const qd_connection_t *conn,
*strip_annotations_in = false;
*strip_annotations_out = false;
*role = QDR_ROLE_INTER_ROUTER;
+ *cost = cf->inter_router_cost;
} else if (cf && (strcmp(cf->role, container_role) == 0 ||
strcmp(cf->role, on_demand_role) == 0)) // backward compat
*role = QDR_ROLE_ROUTE_CONTAINER;
@@ -497,16 +499,17 @@ static int AMQP_link_detach_handler(void* context, qd_link_t *link, qd_detach_ty
static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool inbound)
{
qdr_connection_role_t role = 0;
+ int cost = 1;
bool strip_annotations_in = false;
bool strip_annotations_out = false;
int link_capacity = 1;
const char *name = 0;
pn_connection_t *pn_conn = qd_connection_pn(conn);
- qd_router_connection_get_config(conn, &role, &name,
+ qd_router_connection_get_config(conn, &role, &cost, &name,
&strip_annotations_in, &strip_annotations_out, &link_capacity);
- qdr_connection_t *qdrc = qdr_connection_opened(router->router_core, inbound, role, name,
+ qdr_connection_t *qdrc = qdr_connection_opened(router->router_core, inbound, role, cost, name,
pn_connection_remote_container(pn_conn),
strip_annotations_in, strip_annotations_out, link_capacity);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9f232923/tests/router_engine_test.py
----------------------------------------------------------------------
diff --git a/tests/router_engine_test.py b/tests/router_engine_test.py
index c78cfd9..76ae101 100644
--- a/tests/router_engine_test.py
+++ b/tests/router_engine_test.py
@@ -52,34 +52,34 @@ class Adapter(object):
class DataTest(unittest.TestCase):
def test_link_state(self):
- ls = LinkState(None, 'R1', 1, ['R2', 'R3'])
+ ls = LinkState(None, 'R1', 1, {'R2':1, 'R3':1})
self.assertEqual(ls.id, 'R1')
self.assertEqual(ls.ls_seq, 1)
- self.assertEqual(ls.peers, ['R2', 'R3'])
+ self.assertEqual(ls.peers, {'R2':1, 'R3':1})
ls.bump_sequence()
self.assertEqual(ls.id, 'R1')
self.assertEqual(ls.ls_seq, 2)
- self.assertEqual(ls.peers, ['R2', 'R3'])
+ self.assertEqual(ls.peers, {'R2':1, 'R3':1})
- result = ls.add_peer('R4')
+ result = ls.add_peer('R4', 5)
self.assertTrue(result)
- self.assertEqual(ls.peers, ['R2', 'R3', 'R4'])
- result = ls.add_peer('R2')
+ self.assertEqual(ls.peers, {'R2':1, 'R3':1, 'R4':5})
+ result = ls.add_peer('R2', 1)
self.assertFalse(result)
- self.assertEqual(ls.peers, ['R2', 'R3', 'R4'])
+ self.assertEqual(ls.peers, {'R2':1, 'R3':1, 'R4':5})
result = ls.del_peer('R3')
self.assertTrue(result)
- self.assertEqual(ls.peers, ['R2', 'R4'])
+ self.assertEqual(ls.peers, {'R2':1, 'R4':5})
result = ls.del_peer('R5')
self.assertFalse(result)
- self.assertEqual(ls.peers, ['R2', 'R4'])
+ self.assertEqual(ls.peers, {'R2':1, 'R4':5})
encoded = ls.to_dict()
new_ls = LinkState(encoded)
self.assertEqual(new_ls.id, 'R1')
self.assertEqual(new_ls.ls_seq, 2)
- self.assertEqual(new_ls.peers, ['R2', 'R4'])
+ self.assertEqual(new_ls.peers, {'R2':1, 'R4':5})
def test_hello_message(self):
@@ -141,8 +141,8 @@ class NeighborTest(unittest.TestCase):
def send(self, dest, msg):
self.sent.append((dest, msg))
- def neighbor_refresh(self, node_id, instance, link_id, now):
- self.neighbors[node_id] = (instance, link_id, now)
+ def neighbor_refresh(self, node_id, instance, link_id, cost, now):
+ self.neighbors[node_id] = (instance, link_id, cost, now)
def setUp(self):
self.sent = []
@@ -177,7 +177,7 @@ class NeighborTest(unittest.TestCase):
self.sent = []
self.neighbors = {}
self.engine = HelloProtocol(self, self)
- self.engine.handle_hello(MessageHELLO(None, 'R2', []), 2.0, 0)
+ self.engine.handle_hello(MessageHELLO(None, 'R2', []), 2.0, 0, 1)
self.engine.tick(5.0)
self.assertEqual(len(self.sent), 1)
dest, msg = self.sent.pop(0)
@@ -187,7 +187,7 @@ class NeighborTest(unittest.TestCase):
self.sent = []
self.neighbors = {}
self.engine = HelloProtocol(self, self)
- self.engine.handle_hello(MessageHELLO(None, 'R2', ['R1']), 0.5, 0)
+ self.engine.handle_hello(MessageHELLO(None, 'R2', ['R1']), 0.5, 0, 1)
self.engine.tick(1.0)
self.engine.tick(2.0)
self.engine.tick(3.0)
@@ -198,13 +198,13 @@ class NeighborTest(unittest.TestCase):
self.sent = []
self.neighbors = {}
self.engine = HelloProtocol(self, self)
- self.engine.handle_hello(MessageHELLO(None, 'R2', ['R1']), 0.5, 0)
+ self.engine.handle_hello(MessageHELLO(None, 'R2', ['R1']), 0.5, 0, 1)
self.engine.tick(1.0)
- self.engine.handle_hello(MessageHELLO(None, 'R3', ['R1', 'R2']), 1.5, 0)
+ self.engine.handle_hello(MessageHELLO(None, 'R3', ['R1', 'R2']), 1.5, 0, 1)
self.engine.tick(2.0)
- self.engine.handle_hello(MessageHELLO(None, 'R4', ['R1']), 2.5, 0)
- self.engine.handle_hello(MessageHELLO(None, 'R5', ['R2']), 2.5, 0)
- self.engine.handle_hello(MessageHELLO(None, 'R6', ['R1']), 2.5, 0)
+ self.engine.handle_hello(MessageHELLO(None, 'R4', ['R1']), 2.5, 0, 1)
+ self.engine.handle_hello(MessageHELLO(None, 'R5', ['R2']), 2.5, 0, 1)
+ self.engine.handle_hello(MessageHELLO(None, 'R6', ['R1']), 2.5, 0, 1)
self.engine.tick(3.0)
keys = self.neighbors.keys()
keys.sort()
@@ -227,9 +227,9 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R1': LinkState(None, 'R1', 1, ['R2']),
- 'R2': LinkState(None, 'R2', 1, ['R1', 'R3']),
- 'R3': LinkState(None, 'R3', 1, ['R2']) }
+ collection = { 'R1': LinkState(None, 'R1', 1, {'R2':1}),
+ 'R2': LinkState(None, 'R2', 1, {'R1':1, 'R3':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R2':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 2)
self.assertEqual(next_hops['R2'], 'R2')
@@ -252,12 +252,12 @@ class PathTest(unittest.TestCase):
+----+ +----+ +----+
"""
- collection = { 'R1': LinkState(None, 'R1', 1, ['R2']),
- 'R2': LinkState(None, 'R2', 1, ['R1', 'R3', 'R4']),
- 'R3': LinkState(None, 'R3', 1, ['R2', 'R5']),
- 'R4': LinkState(None, 'R4', 1, ['R2', 'R5']),
- 'R5': LinkState(None, 'R5', 1, ['R3', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5']) }
+ collection = { 'R1': LinkState(None, 'R1', 1, {'R2':1}),
+ 'R2': LinkState(None, 'R2', 1, {'R1':1, 'R3':1, 'R4':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R2':1, 'R5':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R2':1, 'R5':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R3':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 5)
self.assertEqual(next_hops['R2'], 'R2')
@@ -289,12 +289,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 5)
self.assertEqual(next_hops['R2'], 'R3')
@@ -326,12 +326,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5', 'R7']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R3')
@@ -366,12 +366,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3', 'R1']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5', 'R2']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5', 'R7']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1, 'R1':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1, 'R2':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R2')
@@ -406,12 +406,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5', 'R2']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5', 'R7']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1, 'R2':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R2')
@@ -446,12 +446,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3', 'R1']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4', 'R6']),
- 'R6': LinkState(None, 'R6', 1, ['R5', 'R7']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1, 'R1':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R3')
@@ -486,12 +486,12 @@ class PathTest(unittest.TestCase):
+====+ +----+ +----+
"""
- collection = { 'R2': LinkState(None, 'R2', 1, ['R3', 'R1']),
- 'R3': LinkState(None, 'R3', 1, ['R1', 'R2', 'R4']),
- 'R4': LinkState(None, 'R4', 1, ['R3', 'R5']),
- 'R1': LinkState(None, 'R1', 1, ['R3', 'R5']),
- 'R5': LinkState(None, 'R5', 1, ['R1', 'R4']),
- 'R6': LinkState(None, 'R6', 1, ['R5', 'R7']) }
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':1, 'R1':1}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':1, 'R2':1, 'R4':1}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':1, 'R5':1}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
next_hops, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 4)
self.assertEqual(next_hops['R2'], 'R3')
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/7] qpid-dispatch git commit: DISPATCH-10 - Use the inter_router
cost in the route computation.
Posted by tr...@apache.org.
DISPATCH-10 - Use the inter_router cost in the route computation.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/607e942b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/607e942b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/607e942b
Branch: refs/heads/master
Commit: 607e942b44500e051dafaa7fc505844469b0b9f1
Parents: 9f23292
Author: Ted Ross <tr...@redhat.com>
Authored: Wed Apr 20 16:51:00 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:50:20 2016 -0400
----------------------------------------------------------------------
python/qpid_dispatch_internal/router/node.py | 3 ++-
python/qpid_dispatch_internal/router/path.py | 21 +++++++++++----------
2 files changed, 13 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/607e942b/python/qpid_dispatch_internal/router/node.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/node.py b/python/qpid_dispatch_internal/router/node.py
index 7ae08fd..2dfcbcf 100644
--- a/python/qpid_dispatch_internal/router/node.py
+++ b/python/qpid_dispatch_internal/router/node.py
@@ -146,8 +146,9 @@ class NodeTracker(object):
collection = {self.my_id : self.link_state}
for node_id, node in self.nodes.items():
collection[node_id] = node.link_state
- next_hops, valid_origins = self.container.path_engine.calculate_routes(collection)
+ next_hops, cost, valid_origins = self.container.path_engine.calculate_routes(collection)
self.container.log_ls(LOG_TRACE, "Computed next hops: %r" % next_hops)
+ self.container.log_ls(LOG_TRACE, "Computed costs: %r" % cost)
self.container.log_ls(LOG_TRACE, "Computed valid origins: %r" % valid_origins)
##
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/607e942b/python/qpid_dispatch_internal/router/path.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/path.py b/python/qpid_dispatch_internal/router/path.py
index f47ae13..2e15257 100644
--- a/python/qpid_dispatch_internal/router/path.py
+++ b/python/qpid_dispatch_internal/router/path.py
@@ -40,7 +40,7 @@ class PathEngine(object):
link_states[_id] = ls.peers
for p in ls.peers:
if p not in link_states:
- link_states[p] = [_id]
+ link_states[p] = {_id:1L}
##
## Setup Dijkstra's Algorithm
@@ -61,27 +61,28 @@ class PathEngine(object):
if cost[u] == None:
# There are no more reachable nodes in unresolved
break
- for v in link_states[u]:
+ for v, v_cost in link_states[u].items():
if unresolved.contains(v):
- alt = cost[u] + 1 # TODO - Use link cost instead of 1
+ alt = cost[u] + v_cost
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
+ ## Remove unreachable nodes from the maps. 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)
+ cost.pop(u)
##
- ## Return previous-node map. This is a map of all reachable, remote nodes to
- ## their predecessor node.
+ ## Return previous-node and cost maps. Prev is a map of all reachable, remote nodes to
+ ## their predecessor node. Cost is a map of all reachable nodes and their costs.
##
- return prev
+ return prev, cost
def _calculate_valid_origins(self, nodeset, collection):
@@ -96,7 +97,7 @@ class PathEngine(object):
valid_origin[node] = []
for root in valid_origin.keys():
- prev = self._calculate_tree_from_root(root, collection)
+ prev, cost = self._calculate_tree_from_root(root, collection)
nodes = prev.keys()
while len(nodes) > 0:
u = nodes[0]
@@ -119,7 +120,7 @@ class PathEngine(object):
##
## Generate the shortest-path tree with the local node as root
##
- prev = self._calculate_tree_from_root(self.id, collection)
+ prev, cost = self._calculate_tree_from_root(self.id, collection)
nodes = prev.keys()
##
@@ -145,7 +146,7 @@ class PathEngine(object):
##
valid_origins = self._calculate_valid_origins(prev.keys(), collection)
- return (next_hops, valid_origins)
+ return (next_hops, cost, valid_origins)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[7/7] qpid-dispatch git commit: DISPATCH-10 - Exposed cost through
management for router nodes.
Posted by tr...@apache.org.
DISPATCH-10 - Exposed cost through management for router nodes.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/ba8e158d
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/ba8e158d
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/ba8e158d
Branch: refs/heads/master
Commit: ba8e158d64d6865ff44c1cfa28486f598e7f708a
Parents: 9b056fc
Author: Ted Ross <tr...@redhat.com>
Authored: Fri Apr 22 17:48:18 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:51:45 2016 -0400
----------------------------------------------------------------------
python/qpid_dispatch/management/qdrouter.json | 4 ++++
python/qpid_dispatch_internal/router/node.py | 3 ++-
tools/qdstat | 2 ++
3 files changed, 8 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ba8e158d/python/qpid_dispatch/management/qdrouter.json
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index f5cce7a..018d7d0 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -1168,6 +1168,10 @@
"routerLink": {
"description": "Local link to remote node",
"type": "entityId"
+ },
+ "cost": {
+ "description": "Reachability cost",
+ "type": "integer"
}
}
},
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ba8e158d/python/qpid_dispatch_internal/router/node.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/node.py b/python/qpid_dispatch_internal/router/node.py
index 5c53adf..f08cb5c 100644
--- a/python/qpid_dispatch_internal/router/node.py
+++ b/python/qpid_dispatch_internal/router/node.py
@@ -389,7 +389,8 @@ class RouterNode(object):
"nextHop": self.next_hop_router and self.next_hop_router.id,
"validOrigins": self.valid_origins,
"address": Address.topological(self.id, area=self.parent.container.area),
- "routerLink": self.peer_link_id
+ "routerLink": self.peer_link_id,
+ "cost": self.cost
})
def _logify(self, addr):
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ba8e158d/tools/qdstat
----------------------------------------------------------------------
diff --git a/tools/qdstat b/tools/qdstat
index 327eb1a..3dd1668 100755
--- a/tools/qdstat
+++ b/tools/qdstat
@@ -260,6 +260,7 @@ class BusManager(Node):
heads.append(Header("next-hop"))
heads.append(Header("link"))
if self.opts.verbose:
+ heads.append(Header("cost"))
heads.append(Header("neighbors"))
heads.append(Header("valid-origins"))
rows = []
@@ -277,6 +278,7 @@ class BusManager(Node):
row.append(node.routerLink)
if self.opts.verbose:
+ row.append(node.cost)
row.append('%r' % self._list_clean(node.linkState))
row.append('%r' % self._list_clean(node.validOrigins))
rows.append(row)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[6/7] qpid-dispatch git commit: DISPATCH-208 - Updated "closest"
implementation to use the node-cost data.
Posted by tr...@apache.org.
DISPATCH-208 - Updated "closest" implementation to use the node-cost data.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/9b056fca
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/9b056fca
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/9b056fca
Branch: refs/heads/master
Commit: 9b056fcad8fb7a5b23afa27301f55c6c509ffe4d
Parents: ff83fc2
Author: Ted Ross <tr...@redhat.com>
Authored: Fri Apr 22 17:42:59 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:51:38 2016 -0400
----------------------------------------------------------------------
src/router_core/connections.c | 1 +
src/router_core/forwarder.c | 46 ++++++++++++++++++++---
src/router_core/route_tables.c | 59 +++++++++++++++++++++++++-----
src/router_core/router_core_private.h | 9 ++++-
4 files changed, 98 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9b056fca/src/router_core/connections.c
----------------------------------------------------------------------
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 0a4f1ed..b53d157 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -694,6 +694,7 @@ void qdr_check_addr_CT(qdr_core_t *core, qdr_address_t *addr, bool was_local)
DEQ_REMOVE(core->addrs, addr);
qd_hash_handle_free(addr->hash_handle);
qd_bitmask_free(addr->rnodes);
+ qd_bitmask_free(addr->closest_remotes);
free_qdr_address_t(addr);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9b056fca/src/router_core/forwarder.c
----------------------------------------------------------------------
diff --git a/src/router_core/forwarder.c b/src/router_core/forwarder.c
index 274beb0..b82a6e5 100644
--- a/src/router_core/forwarder.c
+++ b/src/router_core/forwarder.c
@@ -71,6 +71,33 @@ static bool qdr_forward_attach_null_CT(qdr_core_t *core,
}
+static void qdr_forward_find_closest_remotes_CT(qdr_core_t *core, qdr_address_t *addr)
+{
+ qdr_node_t *rnode = DEQ_HEAD(core->routers);
+ int lowest_cost = 0;
+
+ if (!addr->closest_remotes)
+ addr->closest_remotes = qd_bitmask(0);
+ addr->cost_epoch = core->cost_epoch;
+ addr->next_remote = -1;
+
+ qd_bitmask_clear_all(addr->closest_remotes);
+ while (rnode) {
+ if (qd_bitmask_value(addr->rnodes, rnode->mask_bit)) {
+ if (lowest_cost == 0) {
+ lowest_cost = rnode->cost;
+ addr->next_remote = rnode->mask_bit;
+ }
+ if (lowest_cost == rnode->cost)
+ qd_bitmask_set_bit(addr->closest_remotes, rnode->mask_bit);
+ else
+ break;
+ }
+ rnode = DEQ_NEXT(rnode);
+ }
+}
+
+
qdr_delivery_t *qdr_forward_new_delivery_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, qdr_link_t *link, qd_message_t *msg)
{
qdr_delivery_t *dlv = new_qdr_delivery_t();
@@ -347,15 +374,22 @@ int qdr_forward_closest_CT(qdr_core_t *core,
// Forward to remote routers with subscribers using the appropriate
// link for the traffic class: control or data
//
- // TODO - presently, this picks one remote link to send to. This needs
- // to be enhanced so it properly chooses the route to the closest destination.
- //
- int router_bit;
qdr_node_t *next_node;
- if (qd_bitmask_first_set(addr->rnodes, &router_bit)) {
- qdr_node_t *rnode = core->routers_by_mask_bit[router_bit];
+ //
+ // If the cached list of closest remotes is stale (i.e. cost data has changed),
+ // recompute the closest remote routers.
+ //
+ if (addr->cost_epoch != core->cost_epoch)
+ qdr_forward_find_closest_remotes_CT(core, addr);
+
+ if (addr->next_remote >= 0) {
+ qdr_node_t *rnode = core->routers_by_mask_bit[addr->next_remote];
if (rnode) {
+ _qdbm_next(addr->closest_remotes, &addr->next_remote);
+ if (addr->next_remote == -1)
+ qd_bitmask_first_set(addr->closest_remotes, &addr->next_remote);
+
if (rnode->next_hop)
next_node = rnode->next_hop;
else
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9b056fca/src/router_core/route_tables.c
----------------------------------------------------------------------
diff --git a/src/router_core/route_tables.c b/src/router_core/route_tables.c
index 02c2c65..8f59d39 100644
--- a/src/router_core/route_tables.c
+++ b/src/router_core/route_tables.c
@@ -177,12 +177,49 @@ void qdr_core_unsubscribe(qdr_subscription_t *sub)
// In-Thread Functions
//==================================================================================
+//
+// React to the updated cost of a router node. The core->routers list is to be kept
+// sorted by cost, from least to most.
+//
+void qdr_route_table_update_cost_CT(qdr_core_t *core, qdr_node_t *rnode)
+{
+ qdr_node_t *ptr;
+ bool needs_reinsertion = false;
+
+ ptr = DEQ_PREV(rnode);
+ if (ptr && ptr->cost > rnode->cost)
+ needs_reinsertion = true;
+ else {
+ ptr = DEQ_NEXT(rnode);
+ if (ptr && ptr->cost < rnode->cost)
+ needs_reinsertion = true;
+ }
+
+ if (needs_reinsertion) {
+ core->cost_epoch++;
+ DEQ_REMOVE(core->routers, rnode);
+ ptr = DEQ_TAIL(core->routers);
+ while (ptr) {
+ if (rnode->cost >= ptr->cost) {
+ DEQ_INSERT_AFTER(core->routers, rnode, ptr);
+ break;
+ }
+ ptr = DEQ_PREV(ptr);
+ }
+
+ if (!ptr)
+ DEQ_INSERT_HEAD(core->routers, rnode);
+ }
+}
+
+
void qdr_route_table_setup_CT(qdr_core_t *core)
{
DEQ_INIT(core->addrs);
DEQ_INIT(core->routers);
core->addr_hash = qd_hash(12, 32, 0);
core->conn_id_hash = qd_hash(6, 4, 0);
+ core->cost_epoch = 1;
if (core->router_mode == QD_ROUTER_MODE_INTERIOR) {
core->hello_addr = qdr_add_local_address_CT(core, 'L', "qdhello", QD_TREATMENT_MULTICAST_FLOOD);
@@ -262,8 +299,15 @@ static void qdr_add_router_CT(qdr_core_t *core, qdr_action_t *action, bool disca
rnode->peer_data_link = 0;
rnode->ref_count = 0;
rnode->valid_origins = qd_bitmask(0);
+ rnode->cost = 0;
- DEQ_INSERT_TAIL(core->routers, rnode);
+ //
+ // Insert at the head of the list because we don't yet know the cost to this
+ // router node and we've set the cost to zero. This puts it in a properly-sorted
+ // position. Also, don't bump the cost_epoch here because this new router won't be
+ // used until it is assigned a cost.
+ //
+ DEQ_INSERT_HEAD(core->routers, rnode);
//
// Link the router record to the address record.
@@ -338,12 +382,15 @@ static void qdr_del_router_CT(qdr_core_t *core, qdr_action_t *action, bool disca
//
qd_bitmask_free(rnode->valid_origins);
DEQ_REMOVE(core->routers, rnode);
+ core->cost_epoch++;
free_qdr_node_t(rnode);
qd_hash_remove_by_handle(core->addr_hash, oaddr->hash_handle);
DEQ_REMOVE(core->addrs, oaddr);
qd_hash_handle_free(oaddr->hash_handle);
core->routers_by_mask_bit[router_maskbit] = 0;
+ qd_bitmask_free(oaddr->closest_remotes);
+ qd_bitmask_free(oaddr->rnodes);
free_qdr_address_t(oaddr);
}
@@ -465,6 +512,7 @@ static void qdr_set_cost_CT(qdr_core_t *core, qdr_action_t *action, bool discard
qdr_node_t *rnode = core->routers_by_mask_bit[router_maskbit];
rnode->cost = cost;
+ qdr_route_table_update_cost_CT(core, rnode);
}
@@ -503,10 +551,6 @@ static void qdr_set_valid_origins_CT(qdr_core_t *core, qdr_action_t *action, boo
static void qdr_map_destination_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
- //
- // TODO - handle the class-prefix and phase explicitly
- //
-
int router_maskbit = action->args.route_table.router_maskbit;
qdr_field_t *address = action->args.route_table.address;
@@ -540,11 +584,8 @@ static void qdr_map_destination_CT(qdr_core_t *core, qdr_action_t *action, bool
qdr_node_t *rnode = core->routers_by_mask_bit[router_maskbit];
qd_bitmask_set_bit(addr->rnodes, router_maskbit);
rnode->ref_count++;
+ addr->cost_epoch--;
qdr_addr_start_inlinks_CT(core, addr);
-
- //
- // TODO - If this affects a waypoint, create the proper side effects
- //
} while (false);
qdr_field_free(address);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/9b056fca/src/router_core/router_core_private.h
----------------------------------------------------------------------
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 49f7257..061e346 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -321,6 +321,10 @@ struct qdr_address_t {
bool block_deletion;
bool local;
+ uint64_t cost_epoch;
+ qd_bitmask_t *closest_remotes;
+ int next_remote;
+
/**@name Statistics */
///@{
uint64_t deliveries_ingress;
@@ -549,11 +553,12 @@ struct qdr_core_t {
qdr_address_t *router_addr_T;
qdr_address_t *routerma_addr_T;
- qdr_node_list_t routers;
+ qdr_node_list_t routers; ///< List of routers, in order of cost, from lowest to highest
qd_bitmask_t *neighbor_free_mask;
qdr_node_t **routers_by_mask_bit;
qdr_link_t **control_links_by_mask_bit;
qdr_link_t **data_links_by_mask_bit;
+ uint64_t cost_epoch;
uint64_t next_tag;
@@ -566,7 +571,7 @@ struct qdr_core_t {
void *router_core_thread(void *arg);
uint64_t qdr_identifier(qdr_core_t* core);
-void qdr_management_agent_on_message(void *context, qd_message_t *msg, int unused_link_id, int unused_cost);
+void qdr_management_agent_on_message(void *context, qd_message_t *msg, int link_id, int cost);
void qdr_route_table_setup_CT(qdr_core_t *core);
void qdr_agent_setup_CT(qdr_core_t *core);
void qdr_forwarder_setup_CT(qdr_core_t *core);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[5/7] qpid-dispatch git commit: DISPATCH-10 - Add negotiation of
connection-cost for inter-router connections.
Posted by tr...@apache.org.
DISPATCH-10 - Add negotiation of connection-cost for inter-router connections.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/ff83fc26
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/ff83fc26
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/ff83fc26
Branch: refs/heads/master
Commit: ff83fc265b45893ef1f13f4ab75a431c160aa00a
Parents: 2cb107b
Author: Ted Ross <tr...@redhat.com>
Authored: Thu Apr 21 16:17:55 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:51:18 2016 -0400
----------------------------------------------------------------------
include/qpid/dispatch/amqp.h | 8 +++++---
src/amqp.c | 6 ++++--
src/router_node.c | 33 +++++++++++++++++++++++++++++++++
src/server.c | 27 +++++++++++++++++----------
4 files changed, 59 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/include/qpid/dispatch/amqp.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/amqp.h b/include/qpid/dispatch/amqp.h
index 54773b5..99daf50 100644
--- a/include/qpid/dispatch/amqp.h
+++ b/include/qpid/dispatch/amqp.h
@@ -114,10 +114,12 @@ extern const char * const QD_CAPABILITY_ROUTER_DATA;
extern const char * const QD_DYNAMIC_NODE_PROPERTY_ADDRESS; ///< Address for routing dynamic sources
/// @}
-/** @name Miscellaneous Strings */
+/** @name Connection Properties */
/// @{
-extern const char * const QD_INTERNODE_LINK_NAME_1;
-extern const char * const QD_INTERNODE_LINK_NAME_2;
+extern const char * const QD_CONNECTION_PROPERTY_PRODUCT_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE;
+extern const char * const QD_CONNECTION_PROPERTY_VERSION_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_COST_KEY;
/// @}
/** @name AMQP error codes. */
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/amqp.c
----------------------------------------------------------------------
diff --git a/src/amqp.c b/src/amqp.c
index 656f7ef..d3b02f2 100644
--- a/src/amqp.c
+++ b/src/amqp.c
@@ -31,8 +31,10 @@ const char * const QD_CAPABILITY_ANONYMOUS_RELAY = "ANONYMOUS-RELAY";
const char * const QD_DYNAMIC_NODE_PROPERTY_ADDRESS = "x-opt-qd.address";
-const char * const QD_INTERNODE_LINK_NAME_1 = "qd.internode.1";
-const char * const QD_INTERNODE_LINK_NAME_2 = "qd.internode.2";
+const char * const QD_CONNECTION_PROPERTY_PRODUCT_KEY = "product";
+const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE = "qpid-dispatch-router";
+const char * const QD_CONNECTION_PROPERTY_VERSION_KEY = "version";
+const char * const QD_CONNECTION_PROPERTY_COST_KEY = "qd.inter-router-cost";
const qd_amqp_error_t QD_AMQP_OK = { 200, "OK" };
const qd_amqp_error_t QD_AMQP_CREATED = { 201, "Created" };
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/router_node.c
----------------------------------------------------------------------
diff --git a/src/router_node.c b/src/router_node.c
index 8cba85e..a6431fe 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -500,6 +500,7 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
{
qdr_connection_role_t role = 0;
int cost = 1;
+ int remote_cost = 1;
bool strip_annotations_in = false;
bool strip_annotations_out = false;
int link_capacity = 1;
@@ -509,6 +510,38 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
qd_router_connection_get_config(conn, &role, &cost, &name,
&strip_annotations_in, &strip_annotations_out, &link_capacity);
+ if (role == QDR_ROLE_INTER_ROUTER) {
+ //
+ // Check the remote properties for an inter-router cost value.
+ //
+ pn_data_t *props = pn_conn ? pn_connection_remote_properties(pn_conn) : 0;
+ if (props) {
+ pn_data_rewind(props);
+ pn_data_next(props);
+ if (props && pn_data_type(props) == PN_MAP) {
+ pn_data_enter(props);
+ while (pn_data_next(props)) {
+ if (pn_data_type(props) == PN_SYMBOL) {
+ pn_bytes_t sym = pn_data_get_symbol(props);
+ if (sym.size == strlen(QD_CONNECTION_PROPERTY_COST_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_COST_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_INT)
+ remote_cost = pn_data_get_int(props);
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ //
+ // Use the larger of the local and remote costs for this connection
+ //
+ if (remote_cost > cost)
+ cost = remote_cost;
+ }
+
qdr_connection_t *qdrc = qdr_connection_opened(router->router_core, inbound, role, cost, name,
pn_connection_remote_container(pn_conn),
strip_annotations_in, strip_annotations_out, link_capacity);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/server.c
----------------------------------------------------------------------
diff --git a/src/server.c b/src/server.c
index 785bb31..7750056 100644
--- a/src/server.c
+++ b/src/server.c
@@ -454,12 +454,9 @@ static const char *log_incoming(char *buf, size_t size, qdpn_connector_t *cxtr)
}
-static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn)
+static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn, const qd_server_config_t *config)
{
size_t clen = strlen(QD_CAPABILITY_ANONYMOUS_RELAY);
- static char *product_key = "product";
- static char *product_val = "qpid-dispatch-router";
- static char *version_key = "version";
//
// Set the container name
@@ -477,11 +474,21 @@ static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn)
pn_data_put_map(pn_connection_properties(conn));
pn_data_enter(pn_connection_properties(conn));
- pn_data_put_symbol(pn_connection_properties(conn), pn_bytes(strlen(product_key), product_key));
- pn_data_put_string(pn_connection_properties(conn), pn_bytes(strlen(product_val), product_val));
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_KEY), QD_CONNECTION_PROPERTY_PRODUCT_KEY));
+ pn_data_put_string(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_VALUE), QD_CONNECTION_PROPERTY_PRODUCT_VALUE));
- pn_data_put_symbol(pn_connection_properties(conn), pn_bytes(strlen(version_key), version_key));
- pn_data_put_string(pn_connection_properties(conn), pn_bytes(strlen(QPID_DISPATCH_VERSION), QPID_DISPATCH_VERSION));
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_VERSION_KEY), QD_CONNECTION_PROPERTY_VERSION_KEY));
+ pn_data_put_string(pn_connection_properties(conn),
+ pn_bytes(strlen(QPID_DISPATCH_VERSION), QPID_DISPATCH_VERSION));
+
+ if (config && config->inter_router_cost > 1) {
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_COST_KEY), QD_CONNECTION_PROPERTY_COST_KEY));
+ pn_data_put_int(pn_connection_properties(conn), config->inter_router_cost);
+ }
pn_data_exit(pn_connection_properties(conn));
}
@@ -536,7 +543,7 @@ static void thread_process_listeners_LH(qd_server_t *qd_server)
pn_connection_t *conn = pn_connection();
ctx->collector = pn_collector();
pn_connection_collect(conn, ctx->collector);
- decorate_connection(qd_server, conn);
+ decorate_connection(qd_server, conn, ctx->listener->config);
qdpn_connector_set_connection(cxtr, conn);
pn_connection_set_context(conn, ctx);
ctx->pn_conn = conn;
@@ -1086,7 +1093,7 @@ static void cxtr_try_open(void *context)
qd_log(ct->server->log_source, QD_LOG_TRACE, "Connecting to %s:%s", ct->config->host, ct->config->port);
pn_connection_collect(ctx->pn_conn, ctx->collector);
- decorate_connection(ctx->server, ctx->pn_conn);
+ decorate_connection(ctx->server, ctx->pn_conn, ct->config);
//
// qdpn_connector is not thread safe
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/7] qpid-dispatch git commit: DISPATCH-10 - Added tests for
topologies with costs.
Posted by tr...@apache.org.
DISPATCH-10 - Added tests for topologies with costs.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/faf0eefe
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/faf0eefe
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/faf0eefe
Branch: refs/heads/master
Commit: faf0eefee51e54cc8c5bd2d1fd46193619df46a1
Parents: 607e942
Author: Ted Ross <tr...@redhat.com>
Authored: Wed Apr 20 18:37:44 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:50:38 2016 -0400
----------------------------------------------------------------------
tests/router_engine_test.py | 121 ++++++++++++++++++++++++++++++++++++---
1 file changed, 113 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/faf0eefe/tests/router_engine_test.py
----------------------------------------------------------------------
diff --git a/tests/router_engine_test.py b/tests/router_engine_test.py
index 76ae101..003728c 100644
--- a/tests/router_engine_test.py
+++ b/tests/router_engine_test.py
@@ -230,7 +230,7 @@ class PathTest(unittest.TestCase):
collection = { 'R1': LinkState(None, 'R1', 1, {'R2':1}),
'R2': LinkState(None, 'R2', 1, {'R1':1, 'R3':1}),
'R3': LinkState(None, 'R3', 1, {'R2':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 2)
self.assertEqual(next_hops['R2'], 'R2')
self.assertEqual(next_hops['R3'], 'R2')
@@ -258,7 +258,7 @@ class PathTest(unittest.TestCase):
'R4': LinkState(None, 'R4', 1, {'R2':1, 'R5':1}),
'R5': LinkState(None, 'R5', 1, {'R3':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 5)
self.assertEqual(next_hops['R2'], 'R2')
self.assertEqual(next_hops['R3'], 'R2')
@@ -295,7 +295,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 5)
self.assertEqual(next_hops['R2'], 'R3')
self.assertEqual(next_hops['R3'], 'R3')
@@ -332,7 +332,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R3')
self.assertEqual(next_hops['R3'], 'R3')
@@ -372,7 +372,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1, 'R2':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R2')
self.assertEqual(next_hops['R3'], 'R3')
@@ -381,6 +381,13 @@ class PathTest(unittest.TestCase):
self.assertEqual(next_hops['R6'], 'R5')
self.assertEqual(next_hops['R7'], 'R5')
+ self.assertEqual(costs['R2'], 1)
+ self.assertEqual(costs['R3'], 1)
+ self.assertEqual(costs['R4'], 2)
+ self.assertEqual(costs['R5'], 1)
+ self.assertEqual(costs['R6'], 2)
+ self.assertEqual(costs['R7'], 3)
+
valid_origins['R2'].sort()
valid_origins['R3'].sort()
valid_origins['R4'].sort()
@@ -412,7 +419,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1, 'R2':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R2')
self.assertEqual(next_hops['R3'], 'R3')
@@ -452,7 +459,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1, 'R6':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 6)
self.assertEqual(next_hops['R2'], 'R3')
self.assertEqual(next_hops['R3'], 'R3')
@@ -492,7 +499,7 @@ class PathTest(unittest.TestCase):
'R1': LinkState(None, 'R1', 1, {'R3':1, 'R5':1}),
'R5': LinkState(None, 'R5', 1, {'R1':1, 'R4':1}),
'R6': LinkState(None, 'R6', 1, {'R5':1, 'R7':1}) }
- next_hops, valid_origins = self.engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
self.assertEqual(len(next_hops), 4)
self.assertEqual(next_hops['R2'], 'R3')
self.assertEqual(next_hops['R3'], 'R3')
@@ -508,6 +515,104 @@ class PathTest(unittest.TestCase):
self.assertEqual(valid_origins['R4'], [])
self.assertEqual(valid_origins['R5'], ['R2', 'R3'])
+ def test_topology5_with_costs1(self):
+ """
+
+ +----+ +----+ +----+
+ | R2 |--4---| R3 |---4--| R4 |
+ +----+ +----+ +----+
+ | | |
+ | 3 5
+ | | |
+ | +====+ +----+ +----+
+ +--20----| R1 |--10--| R5 |--2---| R6 |------ R7 (no ls from R7)
+ +====+ +----+ +----+
+
+ """
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':4, 'R1':20}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':3, 'R2':4, 'R4':4}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':4, 'R5':5}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':3, 'R5':10, 'R2':20}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':10, 'R4':5, 'R6':2}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':2, 'R7':1}) }
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
+ self.assertEqual(len(next_hops), 6)
+ self.assertEqual(next_hops['R2'], 'R3')
+ self.assertEqual(next_hops['R3'], 'R3')
+ self.assertEqual(next_hops['R4'], 'R3')
+ self.assertEqual(next_hops['R5'], 'R5')
+ self.assertEqual(next_hops['R6'], 'R5')
+ self.assertEqual(next_hops['R7'], 'R5')
+
+ self.assertEqual(costs['R2'], 7)
+ self.assertEqual(costs['R3'], 3)
+ self.assertEqual(costs['R4'], 7)
+ self.assertEqual(costs['R5'], 10)
+ self.assertEqual(costs['R6'], 12)
+ self.assertEqual(costs['R7'], 13)
+
+ valid_origins['R2'].sort()
+ valid_origins['R3'].sort()
+ valid_origins['R4'].sort()
+ valid_origins['R5'].sort()
+ valid_origins['R6'].sort()
+ valid_origins['R7'].sort()
+ self.assertEqual(valid_origins['R2'], [])
+ self.assertEqual(valid_origins['R3'], [])
+ self.assertEqual(valid_origins['R4'], [])
+ self.assertEqual(valid_origins['R5'], [])
+ self.assertEqual(valid_origins['R6'], [])
+ self.assertEqual(valid_origins['R7'], [])
+
+ def test_topology5_with_costs2(self):
+ """
+
+ +----+ +----+ +----+
+ | R2 |--4---| R3 |---4--| R4 |
+ +----+ +----+ +----+
+ | | |
+ | 100 100
+ | | |
+ | +====+ +----+ +----+
+ +---5----| R1 |--10--| R5 |--2---| R6 |------ R7 (no ls from R7)
+ +====+ +----+ +----+
+
+ """
+ collection = { 'R2': LinkState(None, 'R2', 1, {'R3':4, 'R1':5}),
+ 'R3': LinkState(None, 'R3', 1, {'R1':100, 'R2':4, 'R4':4}),
+ 'R4': LinkState(None, 'R4', 1, {'R3':4, 'R5':100}),
+ 'R1': LinkState(None, 'R1', 1, {'R3':100, 'R5':10, 'R2':5}),
+ 'R5': LinkState(None, 'R5', 1, {'R1':10, 'R4':100, 'R6':2}),
+ 'R6': LinkState(None, 'R6', 1, {'R5':2, 'R7':1}) }
+ next_hops, costs, valid_origins = self.engine.calculate_routes(collection)
+ self.assertEqual(len(next_hops), 6)
+ self.assertEqual(next_hops['R2'], 'R2')
+ self.assertEqual(next_hops['R3'], 'R2')
+ self.assertEqual(next_hops['R4'], 'R2')
+ self.assertEqual(next_hops['R5'], 'R5')
+ self.assertEqual(next_hops['R6'], 'R5')
+ self.assertEqual(next_hops['R7'], 'R5')
+
+ self.assertEqual(costs['R2'], 5)
+ self.assertEqual(costs['R3'], 9)
+ self.assertEqual(costs['R4'], 13)
+ self.assertEqual(costs['R5'], 10)
+ self.assertEqual(costs['R6'], 12)
+ self.assertEqual(costs['R7'], 13)
+
+ valid_origins['R2'].sort()
+ valid_origins['R3'].sort()
+ valid_origins['R4'].sort()
+ valid_origins['R5'].sort()
+ valid_origins['R6'].sort()
+ valid_origins['R7'].sort()
+ self.assertEqual(valid_origins['R2'], ['R5', 'R6', 'R7'])
+ self.assertEqual(valid_origins['R3'], ['R5', 'R6', 'R7'])
+ self.assertEqual(valid_origins['R4'], ['R5', 'R6', 'R7'])
+ self.assertEqual(valid_origins['R5'], ['R2', 'R3', 'R4'])
+ self.assertEqual(valid_origins['R6'], ['R2', 'R3', 'R4'])
+ self.assertEqual(valid_origins['R7'], ['R2', 'R3', 'R4'])
+
if __name__ == '__main__':
unittest.main(main_module())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[4/7] qpid-dispatch git commit: DISPATCH-10 Communicate the router
node costs to the router core.
Posted by tr...@apache.org.
DISPATCH-10 Communicate the router node costs to the router core.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/2cb107b9
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/2cb107b9
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/2cb107b9
Branch: refs/heads/master
Commit: 2cb107b93002a1e0d24309f915d45cf41ca2ee0a
Parents: faf0eef
Author: Ted Ross <tr...@redhat.com>
Authored: Thu Apr 21 13:42:38 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:50:56 2016 -0400
----------------------------------------------------------------------
include/qpid/dispatch/router_core.h | 1 +
python/qpid_dispatch_internal/router/node.py | 15 ++++++++++--
src/router_core/route_tables.c | 30 +++++++++++++++++++++++
src/router_core/router_core_private.h | 2 ++
src/router_pynode.c | 18 ++++++++++++++
5 files changed, 64 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2cb107b9/include/qpid/dispatch/router_core.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index 81dd2f9..951bb21 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -69,6 +69,7 @@ void qdr_core_set_link(qdr_core_t *core, int router_maskbit, int link_maskbit);
void qdr_core_remove_link(qdr_core_t *core, int router_maskbit);
void qdr_core_set_next_hop(qdr_core_t *core, int router_maskbit, int nh_router_maskbit);
void qdr_core_remove_next_hop(qdr_core_t *core, int router_maskbit);
+void qdr_core_set_cost(qdr_core_t *core, int router_maskbit, int cost);
void qdr_core_set_valid_origins(qdr_core_t *core, int router_maskbit, qd_bitmask_t *routers);
void qdr_core_map_destination(qdr_core_t *core, int router_maskbit, const char *address_hash);
void qdr_core_unmap_destination(qdr_core_t *core, int router_maskbit, const char *address_hash);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2cb107b9/python/qpid_dispatch_internal/router/node.py
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch_internal/router/node.py b/python/qpid_dispatch_internal/router/node.py
index 2dfcbcf..5c53adf 100644
--- a/python/qpid_dispatch_internal/router/node.py
+++ b/python/qpid_dispatch_internal/router/node.py
@@ -146,9 +146,9 @@ class NodeTracker(object):
collection = {self.my_id : self.link_state}
for node_id, node in self.nodes.items():
collection[node_id] = node.link_state
- next_hops, cost, valid_origins = self.container.path_engine.calculate_routes(collection)
+ next_hops, costs, valid_origins = self.container.path_engine.calculate_routes(collection)
self.container.log_ls(LOG_TRACE, "Computed next hops: %r" % next_hops)
- self.container.log_ls(LOG_TRACE, "Computed costs: %r" % cost)
+ self.container.log_ls(LOG_TRACE, "Computed costs: %r" % costs)
self.container.log_ls(LOG_TRACE, "Computed valid origins: %r" % valid_origins)
##
@@ -158,8 +158,10 @@ class NodeTracker(object):
node = self.nodes[node_id]
next_hop = self.nodes[next_hop_id]
vo = valid_origins[node_id]
+ cost = costs[node_id]
node.set_next_hop(next_hop)
node.set_valid_origins(vo)
+ node.set_cost(cost)
##
## Send link-state requests and mobile-address requests to the nodes
@@ -367,6 +369,7 @@ class RouterNode(object):
self.peer_link_id = None
self.link_state = LinkState(None, self.id, 0, {})
self.next_hop_router = None
+ self.cost = None
self.valid_origins = None
self.mobile_addresses = []
self.mobile_address_sequence = 0
@@ -443,6 +446,14 @@ class RouterNode(object):
self.log(LOG_TRACE, "Node %s valid origins: %r" % (self.id, valid_origins))
+ def set_cost(self, cost):
+ if self.cost == cost:
+ return
+ self.cost = cost
+ self.adapter.set_cost(self.maskbit, cost)
+ self.log(LOG_TRACE, "Node %s cost: %d" % (self.id, cost))
+
+
def remove_next_hop(self):
if self.next_hop_router:
self.next_hop_router = None
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2cb107b9/src/router_core/route_tables.c
----------------------------------------------------------------------
diff --git a/src/router_core/route_tables.c b/src/router_core/route_tables.c
index 0e27408..02c2c65 100644
--- a/src/router_core/route_tables.c
+++ b/src/router_core/route_tables.c
@@ -26,6 +26,7 @@ static void qdr_set_link_CT (qdr_core_t *core, qdr_action_t *action, bo
static void qdr_remove_link_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_set_next_hop_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_remove_next_hop_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
+static void qdr_set_cost_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_set_valid_origins_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_map_destination_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_unmap_destination_CT (qdr_core_t *core, qdr_action_t *action, bool discard);
@@ -88,6 +89,15 @@ void qdr_core_remove_next_hop(qdr_core_t *core, int router_maskbit)
}
+void qdr_core_set_cost(qdr_core_t *core, int router_maskbit, int cost)
+{
+ qdr_action_t *action = qdr_action(qdr_set_cost_CT, "set_cost");
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.cost = cost;
+ qdr_action_enqueue(core, action);
+}
+
+
void qdr_core_set_valid_origins(qdr_core_t *core, int router_maskbit, qd_bitmask_t *routers)
{
qdr_action_t *action = qdr_action(qdr_set_valid_origins_CT, "set_valid_origins");
@@ -438,6 +448,26 @@ static void qdr_remove_next_hop_CT(qdr_core_t *core, qdr_action_t *action, bool
}
+static void qdr_set_cost_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
+{
+ int router_maskbit = action->args.route_table.router_maskbit;
+ int cost = action->args.route_table.cost;
+
+ if (router_maskbit >= qd_bitmask_width() || router_maskbit < 0) {
+ qd_log(core->log, QD_LOG_CRITICAL, "set_cost: Router maskbit out of range: %d", router_maskbit);
+ return;
+ }
+
+ if (cost < 1) {
+ qd_log(core->log, QD_LOG_CRITICAL, "set_cost: Invalid cost %d for maskbit: %d", cost, router_maskbit);
+ return;
+ }
+
+ qdr_node_t *rnode = core->routers_by_mask_bit[router_maskbit];
+ rnode->cost = cost;
+}
+
+
static void qdr_set_valid_origins_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
int router_maskbit = action->args.route_table.router_maskbit;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2cb107b9/src/router_core/router_core_private.h
----------------------------------------------------------------------
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 8abfba8..49f7257 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -81,6 +81,7 @@ struct qdr_action_t {
int link_maskbit;
int router_maskbit;
int nh_router_maskbit;
+ int cost;
qd_bitmask_t *router_set;
qdr_field_t *address;
} route_table;
@@ -174,6 +175,7 @@ struct qdr_node_t {
qdr_link_t *peer_data_link; ///< Outgoing data link _if_ this is a neighbor node
uint32_t ref_count;
qd_bitmask_t *valid_origins;
+ int cost;
};
ALLOC_DECLARE(qdr_node_t);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2cb107b9/src/router_pynode.c
----------------------------------------------------------------------
diff --git a/src/router_pynode.c b/src/router_pynode.c
index 6b0cda8..3984f74 100644
--- a/src/router_pynode.c
+++ b/src/router_pynode.c
@@ -143,6 +143,23 @@ static PyObject* qd_remove_next_hop(PyObject *self, PyObject *args)
}
+static PyObject* qd_set_cost(PyObject *self, PyObject *args)
+{
+ RouterAdapter *adapter = (RouterAdapter*) self;
+ qd_router_t *router = adapter->router;
+ int router_maskbit;
+ int cost;
+
+ if (!PyArg_ParseTuple(args, "ii", &router_maskbit, &cost))
+ return 0;
+
+ qdr_core_set_cost(router->router_core, router_maskbit, cost);
+
+ Py_INCREF(Py_None);
+ return Py_None;
+}
+
+
static PyObject* qd_set_valid_origins(PyObject *self, PyObject *args)
{
RouterAdapter *adapter = (RouterAdapter*) self;
@@ -262,6 +279,7 @@ static PyMethodDef RouterAdapter_methods[] = {
{"remove_link", qd_remove_link, METH_VARARGS, "Remove the link for a neighbor router"},
{"set_next_hop", qd_set_next_hop, METH_VARARGS, "Set the next hop for a remote router"},
{"remove_next_hop", qd_remove_next_hop, METH_VARARGS, "Remove the next hop for a remote router"},
+ {"set_cost", qd_set_cost, METH_VARARGS, "Set the cost to reach a remote router"},
{"set_valid_origins", qd_set_valid_origins, METH_VARARGS, "Set the valid origins for a remote router"},
{"map_destination", qd_map_destination, METH_VARARGS, "Add a newly discovered destination mapping"},
{"unmap_destination", qd_unmap_destination, METH_VARARGS, "Delete a destination mapping"},
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org