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 2015/10/20 23:35:43 UTC
[1/2] qpid-dispatch git commit: DISPATCH-179 - Rearranged the source
files to make the core more modular.
Repository: qpid-dispatch
Updated Branches:
refs/heads/tross-DISPATCH-179-1 ea2d90b6f -> f03cca6bb
DISPATCH-179 - Rearranged the source files to make the core more modular.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/3333d774
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/3333d774
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/3333d774
Branch: refs/heads/tross-DISPATCH-179-1
Commit: 3333d7744077761c7a85a9c8d40ed09bd3c840cb
Parents: ea2d90b
Author: Ted Ross <tr...@redhat.com>
Authored: Tue Oct 20 14:35:25 2015 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Tue Oct 20 14:35:25 2015 -0400
----------------------------------------------------------------------
include/qpid/dispatch/router_core.h | 65 +++++++------
src/CMakeLists.txt | 3 +-
src/router_core.c | 151 -----------------------------
src/router_core/router_core.c | 65 +++++++++++++
src/router_core/router_core_private.h | 90 +++++++++++++++++
src/router_core/router_core_thread.c | 81 ++++++++++++++++
6 files changed, 275 insertions(+), 180 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/include/qpid/dispatch/router_core.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index b2e4aac..a78da02 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -22,6 +22,10 @@
#include <qpid/dispatch.h>
#include <qpid/dispatch/bitmask.h>
+//
+// All callbacks in this module shall be invoked on a connection thread from the server thread pool.
+//
+
typedef struct qdr_core_t qdr_core_t;
typedef struct qdr_connection_t qdr_connection_t;
typedef struct qdr_link_t qdr_link_t;
@@ -42,14 +46,6 @@ void qdr_core_free(qdr_core_t *core);
* Route table maintenance functions
******************************************************************************
*/
-typedef enum {
- QD_WAYPOINT_SOURCE,
- QD_WAYPOINT_SINK,
- QD_WAYPOINT_THROUGH,
- QD_WAYPOINT_BYPASS,
- QD_WAYPOINT_TAP
-} qd_waypoint_style_t;
-
void qdr_core_add_router(qdr_core_t *core, const char *address, int router_maskbit);
void qdr_core_del_router(qdr_core_t *core, int router_maskbit);
void qdr_core_set_link(qdr_core_t *core, int router_maskbit, int link_maskbit);
@@ -60,25 +56,21 @@ void qdr_core_set_valid_origins(qdr_core_t *core, const qd_bitmask_t *routers);
void qdr_core_map_destination(qdr_core_t *core, int router_maskbit, const char *address, char phase);
void qdr_core_unmap_destination(qdr_core_t *core, int router_maskbit, const char *address, char phase);
-void qdr_core_add_link_route(qdr_core_t *core, qd_field_iterator_t *conn_label, qd_field_iterator_t *prefix);
-void qdr_core_del_link_route(qdr_core_t *core, qd_field_iterator_t *conn_label, qd_field_iterator_t *prefix);
+typedef void (*qdr_mobile_added_t) (void *context, const char *address);
+typedef void (*qdr_mobile_removed_t) (void *context, const char *address);
+typedef void (*qdr_link_lost_t) (void *context, int link_maskbit);
-void qdr_core_add_waypoint(qdr_core_t *core, qd_field_iterator_t *conn_label, qd_field_iterator_t *address, qd_waypoint_style_t style, char in_phase, char out_phase);
-
-//
-// The following callbacks shall be invoked on a connection thread from the server thread pool.
-//
-typedef void (*qdr_mobile_added_t) (void *context, qd_field_iterator_t *address);
-typedef void (*qdr_mobile_removed_t) (void *context, qd_field_iterator_t *address);
-typedef void (*qdr_link_lost_t) (void *context, int link_maskbit);
-typedef void (*qdr_connection_activate_t) (void *context, const qdr_connection_t *connection);
-typedef void (*qdr_receive_t) (void *context, qd_message_t *msg, int link_maskbit);
+void qdr_core_route_table_handlers(void *context,
+ qdr_mobile_added_t mobile_added,
+ qdr_mobile_removed_t mobile_removed,
+ qdr_link_lost_t link_lost);
-void qdr_core_route_table_handlers(void *context,
- qdr_mobile_added_t mobile_added,
- qdr_mobile_removed_t mobile_removed,
- qdr_link_lost_t link_lost,
- qdr_connection_activate_t connection_activate);
+/**
+ ******************************************************************************
+ * In-process message-receiver functions
+ ******************************************************************************
+ */
+typedef void (*qdr_receive_t) (void *context, qd_message_t *msg, int link_maskbit);
void qdr_core_subscribe(qdr_core_t *core, const char *address, bool local, bool mobile, qdr_receive_t on_message, void *context);
@@ -109,6 +101,9 @@ void qdr_connection_set_context(qdr_connection_t *conn, void *context);
void *qdr_connection_get_context(qdr_connection_t *conn);
qdr_work_t *qdr_connection_work(qdr_connection_t *conn);
+typedef void (*qdr_connection_activate_t) (void *context, const qdr_connection_t *connection);
+void qdr_connection_activate_handler(qdr_core_t *core, qdr_connection_activate_t handler, void *context);
+
/**
******************************************************************************
* Link functions
@@ -133,15 +128,29 @@ void qdr_delivery_process(qdr_delivery_t *delivery);
/**
******************************************************************************
- * Management instrumentation functions
+ * Management functions
******************************************************************************
*/
typedef enum {
QD_ROUTER_CONNECTION,
QD_ROUTER_LINK,
- QD_ROUTER_ADDRESS
+ QD_ROUTER_ADDRESS,
+ QD_ROUTER_WAYPOINT,
+ QD_ROUTER_EXCHANGE,
+ QD_ROUTER_BINDING
} qd_router_entity_type_t;
-void qdr_core_query(qdr_core_t *core, qd_router_entity_type_t type, const char *filter, void *context);
+typedef struct qdr_query_t qdr_query_t;
+
+void qdr_manage_create(qdr_core_t *core, void *context, qd_router_entity_type_t type, qd_parsed_field_t *attributes);
+void qdr_manage_delete(qdr_core_t *core, void *context, qd_router_entity_type_t type, qd_parsed_field_t *attributes);
+void qdr_manage_read(qdr_core_t *core, void *context, qd_router_entity_type_t type, qd_parsed_field_t *attributes);
+
+qdr_query_t *qdr_manage_get_first(qdr_core_t *core, void *context, qd_router_entity_type_t type, int offset);
+void qdr_manage_get_next(qdr_query_t *query);
+void qdr_query_cancel(qdr_query_t *query);
+
+typedef void (*qdr_manage_response_t) (void *context, int status_code, qd_composed_field_t *body);
+void qdr_manage_handler(qdr_core_t *core, qdr_manage_response_t response_handler);
#endif
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 4b3bd07..b20a1a5 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -65,7 +65,8 @@ set(qpid_dispatch_SOURCES
python_embedded.c
router_agent.c
router_config.c
- router_core.c
+ router_core/router_core.c
+ router_core/router_core_thread.c
router_delivery.c
router_node.c
router_forwarders.c
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/src/router_core.c
----------------------------------------------------------------------
diff --git a/src/router_core.c b/src/router_core.c
deleted file mode 100644
index fd5e22d..0000000
--- a/src/router_core.c
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * 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.
- */
-
-#include "dispatch_private.h"
-#include <qpid/dispatch/router_core.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/log.h>
-#include <memory.h>
-
-/**
- * Creates a thread that is dedicated to managing and using the routing table.
- * The purpose of moving this function into one thread is to remove the widespread
- * lock contention that happens with synchrounous multi-threaded routing.
- *
- * This module owns, manages, and uses the router-link list and the address hash table
- */
-
-
-/**
- * The following data structures are private to the core module. They are defined
- * here to ensure their invisibiliy outside the core.
- */
-
-typedef struct qdr_core_work_t {
- DEQ_LINKS(struct qdr_core_work_t);
-} qdr_core_work_t;
-
-ALLOC_DECLARE(qdr_core_work_t);
-ALLOC_DEFINE(qdr_core_work_t);
-DEQ_DECLARE(qdr_core_work_t, qdr_core_work_list_t);
-
-struct qdr_core_t {
- qd_log_source_t *log;
- sys_cond_t *cond;
- sys_mutex_t *lock;
- sys_thread_t *thread;
- bool running;
- qdr_core_work_list_t work_list;
-};
-
-
-static void router_core_do_work(qdr_core_t *core, qdr_core_work_t *work)
-{
-}
-
-
-static void *router_core_thread(void *arg)
-{
- qdr_core_t *core = (qdr_core_t*) arg;
- qdr_core_work_list_t work_list;
- qdr_core_work_t *work;
-
- qd_log(core->log, QD_LOG_INFO, "Router Core thread running");
- while (core->running) {
- //
- // Use the lock only to protect the condition variable and the work list
- //
- sys_mutex_lock(core->lock);
-
- //
- // Block on the condition variable when there is no work to do
- //
- while (core->running && DEQ_IS_EMPTY(core->work_list))
- sys_cond_wait(core->cond, core->lock);
-
- //
- // Move the entire work list to a private list so we can process it without
- // holding the lock
- //
- DEQ_MOVE(core->work_list, work_list);
- sys_mutex_unlock(core->lock);
-
- //
- // Process and free all of the work items in the list
- //
- work = DEQ_HEAD(work_list);
- while (work) {
- DEQ_REMOVE_HEAD(work_list);
-
- if (core->running)
- router_core_do_work(core, work);
-
- free_qdr_core_work_t(work);
- work = DEQ_HEAD(work_list);
- }
- }
-
- qd_log(core->log, QD_LOG_INFO, "Router Core thread exited");
- return 0;
-}
-
-
-qdr_core_t *qdr_core(void)
-{
- qdr_core_t *core = NEW(qdr_core_t);
- ZERO(core);
-
- //
- // Set up the logging source for the router core
- //
- core->log = qd_log_source("ROUTER_CORE");
-
- //
- // Set up the threading support
- //
- core->cond = sys_cond();
- core->lock = sys_mutex();
- core->running = true;
- DEQ_INIT(core->work_list);
- core->thread = sys_thread(router_core_thread, core);
-
- return core;
-}
-
-
-void qdr_core_free(qdr_core_t *core)
-{
- //
- // Stop and join the thread
- //
- core->running = false;
- sys_cond_signal(core->cond);
- sys_thread_join(core->thread);
-
- //
- // Free the core resources
- //
- sys_thread_free(core->thread);
- sys_cond_free(core->cond);
- sys_mutex_free(core->lock);
- free(core);
-}
-
-
-
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/src/router_core/router_core.c
----------------------------------------------------------------------
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
new file mode 100644
index 0000000..eba6150
--- /dev/null
+++ b/src/router_core/router_core.c
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+#include "router_core_private.h"
+
+
+qdr_core_t *qdr_core(void)
+{
+ qdr_core_t *core = NEW(qdr_core_t);
+ ZERO(core);
+
+ //
+ // Set up the logging source for the router core
+ //
+ core->log = qd_log_source("ROUTER_CORE");
+
+ //
+ // Set up the threading support
+ //
+ core->cond = sys_cond();
+ core->lock = sys_mutex();
+ core->running = true;
+ DEQ_INIT(core->action_list);
+ core->thread = sys_thread(router_core_thread, core);
+
+ return core;
+}
+
+
+void qdr_core_free(qdr_core_t *core)
+{
+ //
+ // Stop and join the thread
+ //
+ core->running = false;
+ sys_cond_signal(core->cond);
+ sys_thread_join(core->thread);
+
+ //
+ // Free the core resources
+ //
+ sys_thread_free(core->thread);
+ sys_cond_free(core->cond);
+ sys_mutex_free(core->lock);
+ free(core);
+}
+
+
+
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/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
new file mode 100644
index 0000000..673ae68
--- /dev/null
+++ b/src/router_core/router_core_private.h
@@ -0,0 +1,90 @@
+#ifndef qd_router_core_private
+#define qd_router_core_private 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.
+ */
+
+#include "dispatch_private.h"
+#include <qpid/dispatch/router_core.h>
+#include <qpid/dispatch/threading.h>
+#include <qpid/dispatch/log.h>
+#include <memory.h>
+
+typedef struct {
+ qd_buffer_t *buffer;
+ qd_field_iterator_t *iterator;
+} qdr_field_t;
+
+typedef enum {
+ QDR_ACTION_ADD_ROUTER,
+ QDR_ACTION_DEL_ROUTER,
+ QDR_ACTION_SET_LINK,
+ QDR_ACTION_REMOVE_LINK,
+ QDR_ACTION_SET_NEXT_HOP,
+ QDR_ACTION_REMOVE_NEXT_HOP,
+ QDR_ACTION_SET_VALID_ORIGINS,
+ QDR_ACTION_MAP_DESTINATION,
+ QDR_ACTION_UNMAP_DESTINATION,
+ QDR_ACTION_SUBSCRIBE,
+ QDR_ACTION_CONNECTION_OPENED,
+ QDR_ACTION_CONNECTION_CLOSED,
+ QDR_ACTION_LINK_FIRST_ATTACH,
+ QDR_ACTION_LINK_SECOND_ATTACH,
+ QDR_ACTION_LINK_DETACH,
+ QDR_ACTION_DELIVER,
+ QDR_ACTION_DELIVER_TO,
+ QDR_ACTION_DISPOSITION_CHANGE,
+ QDR_ACTION_FLOW_CHANGE,
+ QDR_ACTION_MANAGE_CREATE,
+ QDR_ACTION_MANAGE_DELETE,
+ QDR_ACTION_MANAGE_READ,
+ QDR_ACTION_MANAGE_GET_FIRST,
+ QDR_ACTION_MANAGE_GET_NEXT
+} qdr_action_type_t;
+
+typedef struct qdr_action_t {
+ DEQ_LINKS(struct qdr_action_t);
+ qdr_action_type_t action_type;
+ union {
+ struct {
+ int link_maskbit;
+ int router_maskbit;
+ int nh_router_maskbit;
+ qd_bitmask_t *router_set;
+ qdr_field_t *address;
+ char address_class;
+ char address_phase;
+ } route_table;
+ } args;
+} qdr_action_t;
+
+ALLOC_DECLARE(qdr_action_t);
+DEQ_DECLARE(qdr_action_t, qdr_action_list_t);
+
+struct qdr_core_t {
+ qd_log_source_t *log;
+ sys_cond_t *cond;
+ sys_mutex_t *lock;
+ sys_thread_t *thread;
+ bool running;
+ qdr_action_list_t action_list;
+};
+
+void *router_core_thread(void *arg);
+
+#endif
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/3333d774/src/router_core/router_core_thread.c
----------------------------------------------------------------------
diff --git a/src/router_core/router_core_thread.c b/src/router_core/router_core_thread.c
new file mode 100644
index 0000000..1d08bd3
--- /dev/null
+++ b/src/router_core/router_core_thread.c
@@ -0,0 +1,81 @@
+/*
+ * 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.
+ */
+
+#include "router_core_private.h"
+
+/**
+ * Creates a thread that is dedicated to managing and using the routing table.
+ * The purpose of moving this function into one thread is to remove the widespread
+ * lock contention that happens with synchrounous multi-threaded routing.
+ *
+ * This module owns, manages, and uses the router-link list and the address hash table
+ */
+
+ALLOC_DEFINE(qdr_action_t);
+
+
+static void router_core_do_action(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+void *router_core_thread(void *arg)
+{
+ qdr_core_t *core = (qdr_core_t*) arg;
+ qdr_action_list_t action_list;
+ qdr_action_t *action;
+
+ qd_log(core->log, QD_LOG_INFO, "Router Core thread running");
+ while (core->running) {
+ //
+ // Use the lock only to protect the condition variable and the action list
+ //
+ sys_mutex_lock(core->lock);
+
+ //
+ // Block on the condition variable when there is no action to do
+ //
+ while (core->running && DEQ_IS_EMPTY(core->action_list))
+ sys_cond_wait(core->cond, core->lock);
+
+ //
+ // Move the entire action list to a private list so we can process it without
+ // holding the lock
+ //
+ DEQ_MOVE(core->action_list, action_list);
+ sys_mutex_unlock(core->lock);
+
+ //
+ // Process and free all of the action items in the list
+ //
+ action = DEQ_HEAD(action_list);
+ while (action) {
+ DEQ_REMOVE_HEAD(action_list);
+
+ if (core->running)
+ router_core_do_action(core, action);
+
+ free_qdr_action_t(action);
+ action = DEQ_HEAD(action_list);
+ }
+ }
+
+ qd_log(core->log, QD_LOG_INFO, "Router Core thread exited");
+ return 0;
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/2] qpid-dispatch git commit: DISPATCH-179 - Added framework for
route table management.
Posted by tr...@apache.org.
DISPATCH-179 - Added framework for route table management.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/f03cca6b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/f03cca6b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/f03cca6b
Branch: refs/heads/tross-DISPATCH-179-1
Commit: f03cca6bbb64aa142326ab35a51716d16d696181
Parents: 3333d77
Author: Ted Ross <tr...@redhat.com>
Authored: Tue Oct 20 17:34:38 2015 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Tue Oct 20 17:34:38 2015 -0400
----------------------------------------------------------------------
include/qpid/dispatch/router_core.h | 9 +-
src/CMakeLists.txt | 1 +
src/router_core/route_tables.c | 203 +++++++++++++++++++++++++++++
src/router_core/route_tables.h | 22 ++++
src/router_core/router_core.c | 34 +++++
src/router_core/router_core_private.h | 46 +++----
src/router_core/router_core_thread.c | 7 +-
7 files changed, 281 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/include/qpid/dispatch/router_core.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index a78da02..87bcd5c 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -52,15 +52,16 @@ 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_valid_origins(qdr_core_t *core, const qd_bitmask_t *routers);
-void qdr_core_map_destination(qdr_core_t *core, int router_maskbit, const char *address, char phase);
-void qdr_core_unmap_destination(qdr_core_t *core, int router_maskbit, const char *address, char phase);
+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, char aclass, char phase);
+void qdr_core_unmap_destination(qdr_core_t *core, int router_maskbit, const char *address, char aclass, char phase);
typedef void (*qdr_mobile_added_t) (void *context, const char *address);
typedef void (*qdr_mobile_removed_t) (void *context, const char *address);
typedef void (*qdr_link_lost_t) (void *context, int link_maskbit);
-void qdr_core_route_table_handlers(void *context,
+void qdr_core_route_table_handlers(qdr_core_t *core,
+ void *context,
qdr_mobile_added_t mobile_added,
qdr_mobile_removed_t mobile_removed,
qdr_link_lost_t link_lost);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index b20a1a5..a9136b1 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -67,6 +67,7 @@ set(qpid_dispatch_SOURCES
router_config.c
router_core/router_core.c
router_core/router_core_thread.c
+ router_core/route_tables.c
router_delivery.c
router_node.c
router_forwarders.c
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/src/router_core/route_tables.c
----------------------------------------------------------------------
diff --git a/src/router_core/route_tables.c b/src/router_core/route_tables.c
new file mode 100644
index 0000000..bd102f0
--- /dev/null
+++ b/src/router_core/route_tables.c
@@ -0,0 +1,203 @@
+/*
+ * 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.
+ */
+
+#include "router_core_private.h"
+//#include "route_tables.h"
+
+static qdr_action_t *qdr_action(qdr_action_handler_t action_handler);
+static void qdr_action_enqueue(qdr_core_t *core, qdr_action_t *action);
+
+static void qdrh_add_router(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_del_router(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_set_link(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_remove_link(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_set_next_hop(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_remove_next_hop(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_set_valid_origins(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_map_destination(qdr_core_t *core, qdr_action_t *action);
+static void qdrh_unmap_destination(qdr_core_t *core, qdr_action_t *action);
+
+
+//==================================================================================
+// Interface Functions
+//==================================================================================
+
+void qdr_core_add_router(qdr_core_t *core, const char *address, int router_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_add_router);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.address = qdr_field(address);
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_del_router(qdr_core_t *core, int router_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_del_router);
+ action->args.route_table.router_maskbit = router_maskbit;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_set_link(qdr_core_t *core, int router_maskbit, int link_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_set_link);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.link_maskbit = link_maskbit;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_remove_link(qdr_core_t *core, int router_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_remove_link);
+ action->args.route_table.router_maskbit = router_maskbit;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_set_next_hop(qdr_core_t *core, int router_maskbit, int nh_router_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_set_next_hop);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.nh_router_maskbit = nh_router_maskbit;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_remove_next_hop(qdr_core_t *core, int router_maskbit)
+{
+ qdr_action_t *action = qdr_action(qdrh_remove_next_hop);
+ action->args.route_table.router_maskbit = router_maskbit;
+ 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(qdrh_set_valid_origins);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.router_set = routers;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_map_destination(qdr_core_t *core, int router_maskbit, const char *address, char aclass, char phase)
+{
+ qdr_action_t *action = qdr_action(qdrh_map_destination);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.address = qdr_field(address);
+ action->args.route_table.address_phase = phase;
+ action->args.route_table.address_class = aclass;
+ qdr_action_enqueue(core, action);
+}
+
+
+void qdr_core_unmap_destination(qdr_core_t *core, int router_maskbit, const char *address, char aclass, char phase)
+{
+ qdr_action_t *action = qdr_action(qdrh_unmap_destination);
+ action->args.route_table.router_maskbit = router_maskbit;
+ action->args.route_table.address = qdr_field(address);
+ action->args.route_table.address_phase = phase;
+ action->args.route_table.address_class = aclass;
+ qdr_action_enqueue(core, action);
+}
+
+void qdr_core_route_table_handlers(qdr_core_t *core,
+ void *context,
+ qdr_mobile_added_t mobile_added,
+ qdr_mobile_removed_t mobile_removed,
+ qdr_link_lost_t link_lost)
+{
+ core->rt_context = context;
+ core->rt_mobile_added = mobile_added;
+ core->rt_mobile_removed = mobile_removed;
+ core->rt_link_lost = link_lost;
+}
+
+
+//==================================================================================
+// Internal Functions
+//==================================================================================
+
+static qdr_action_t *qdr_action(qdr_action_handler_t action_handler)
+{
+ qdr_action_t *action = new_qdr_action_t();
+ ZERO(action);
+ action->action_handler = action_handler;
+ return action;
+}
+
+static void qdr_action_enqueue(qdr_core_t *core, qdr_action_t *action)
+{
+ sys_mutex_lock(core->lock);
+ DEQ_INSERT_TAIL(core->action_list, action);
+ sys_mutex_unlock(core->lock);
+}
+
+
+//==================================================================================
+// In-Thread Functions
+//==================================================================================
+
+static void qdrh_add_router(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_del_router(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_set_link(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_remove_link(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_set_next_hop(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_remove_next_hop(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_set_valid_origins(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_map_destination(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
+static void qdrh_unmap_destination(qdr_core_t *core, qdr_action_t *action)
+{
+}
+
+
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/src/router_core/route_tables.h
----------------------------------------------------------------------
diff --git a/src/router_core/route_tables.h b/src/router_core/route_tables.h
new file mode 100644
index 0000000..13980c4
--- /dev/null
+++ b/src/router_core/route_tables.h
@@ -0,0 +1,22 @@
+#ifndef __core_route_tables_h__
+#define __core_route_tables_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.
+ */
+
+#endif
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/src/router_core/router_core.c
----------------------------------------------------------------------
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index eba6150..7293caf 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -62,4 +62,38 @@ void qdr_core_free(qdr_core_t *core)
}
+ALLOC_DECLARE(qdr_field_t);
+ALLOC_DEFINE(qdr_field_t);
+
+qdr_field_t *qdr_field(const char *text)
+{
+ size_t length = strlen(text);
+ size_t ilength = length;
+ qdr_field_t *field = new_qdr_field_t();
+ qd_buffer_t *buf;
+ ZERO(field);
+
+ while (length > 0) {
+ buf = qd_buffer();
+ size_t cap = qd_buffer_capacity(buf);
+ size_t copy = length > cap ? cap : length;
+ memcpy(qd_buffer_cursor(buf), text, copy);
+ qd_buffer_insert(buf, copy);
+ length -= copy;
+ text += copy;
+ DEQ_INSERT_TAIL(field->buffers, buf);
+ }
+
+ field->iterator = qd_field_iterator_buffer(DEQ_HEAD(field->buffers), 0, ilength);
+
+ return field;
+}
+
+
+void qdr_field_free(qdr_field_t *field)
+{
+ qd_field_iterator_free(field->iterator);
+ qd_buffer_list_free_buffers(&field->buffers);
+ free_qdr_field_t(field);
+}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/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 673ae68..94ec4a1 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -26,40 +26,19 @@
#include <memory.h>
typedef struct {
- qd_buffer_t *buffer;
+ qd_buffer_list_t buffers;
qd_field_iterator_t *iterator;
} qdr_field_t;
-typedef enum {
- QDR_ACTION_ADD_ROUTER,
- QDR_ACTION_DEL_ROUTER,
- QDR_ACTION_SET_LINK,
- QDR_ACTION_REMOVE_LINK,
- QDR_ACTION_SET_NEXT_HOP,
- QDR_ACTION_REMOVE_NEXT_HOP,
- QDR_ACTION_SET_VALID_ORIGINS,
- QDR_ACTION_MAP_DESTINATION,
- QDR_ACTION_UNMAP_DESTINATION,
- QDR_ACTION_SUBSCRIBE,
- QDR_ACTION_CONNECTION_OPENED,
- QDR_ACTION_CONNECTION_CLOSED,
- QDR_ACTION_LINK_FIRST_ATTACH,
- QDR_ACTION_LINK_SECOND_ATTACH,
- QDR_ACTION_LINK_DETACH,
- QDR_ACTION_DELIVER,
- QDR_ACTION_DELIVER_TO,
- QDR_ACTION_DISPOSITION_CHANGE,
- QDR_ACTION_FLOW_CHANGE,
- QDR_ACTION_MANAGE_CREATE,
- QDR_ACTION_MANAGE_DELETE,
- QDR_ACTION_MANAGE_READ,
- QDR_ACTION_MANAGE_GET_FIRST,
- QDR_ACTION_MANAGE_GET_NEXT
-} qdr_action_type_t;
+qdr_field_t *qdr_field(const char *string);
+void qdr_field_free(qdr_field_t *field);
-typedef struct qdr_action_t {
- DEQ_LINKS(struct qdr_action_t);
- qdr_action_type_t action_type;
+typedef struct qdr_action_t qdr_action_t;
+typedef void (*qdr_action_handler_t) (qdr_core_t *core, qdr_action_t *action);
+
+struct qdr_action_t {
+ DEQ_LINKS(qdr_action_t);
+ qdr_action_handler_t action_handler;
union {
struct {
int link_maskbit;
@@ -71,7 +50,7 @@ typedef struct qdr_action_t {
char address_phase;
} route_table;
} args;
-} qdr_action_t;
+};
ALLOC_DECLARE(qdr_action_t);
DEQ_DECLARE(qdr_action_t, qdr_action_list_t);
@@ -83,6 +62,11 @@ struct qdr_core_t {
sys_thread_t *thread;
bool running;
qdr_action_list_t action_list;
+
+ void *rt_context;
+ qdr_mobile_added_t rt_mobile_added;
+ qdr_mobile_removed_t rt_mobile_removed;
+ qdr_link_lost_t rt_link_lost;
};
void *router_core_thread(void *arg);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/f03cca6b/src/router_core/router_core_thread.c
----------------------------------------------------------------------
diff --git a/src/router_core/router_core_thread.c b/src/router_core/router_core_thread.c
index 1d08bd3..87f8c02 100644
--- a/src/router_core/router_core_thread.c
+++ b/src/router_core/router_core_thread.c
@@ -30,11 +30,6 @@
ALLOC_DEFINE(qdr_action_t);
-static void router_core_do_action(qdr_core_t *core, qdr_action_t *action)
-{
-}
-
-
void *router_core_thread(void *arg)
{
qdr_core_t *core = (qdr_core_t*) arg;
@@ -69,7 +64,7 @@ void *router_core_thread(void *arg)
DEQ_REMOVE_HEAD(action_list);
if (core->running)
- router_core_do_action(core, action);
+ action->action_handler(core, action);
free_qdr_action_t(action);
action = DEQ_HEAD(action_list);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org