You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2020/03/10 17:17:17 UTC
[qpid-dispatch] branch master updated (2b7c0fc -> 074efc5)
This is an automated email from the ASF dual-hosted git repository.
kgiusti pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git.
from 2b7c0fc DISPATCH-1590: ensure msg content buffer list is valid on fanout
new 0264301 DISPATCH-1579: open dedicated inter-router sessions for data and control links
new 074efc5 DISPATCH-1579: add logic to activate all links on session Q3 clear
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
include/qpid/dispatch/container.h | 38 ++++-
include/qpid/dispatch/message.h | 20 ++-
include/qpid/dispatch/router_core.h | 5 +-
src/container.c | 195 ++++++++++++++++++++---
src/message.c | 7 +-
src/router_core/connections.c | 28 ++--
src/router_core/core_link_endpoint.c | 3 +-
src/router_core/forwarder.c | 1 +
src/router_core/modules/edge_router/addr_proxy.c | 10 +-
src/router_core/route_control.c | 3 +-
src/router_core/router_core_private.h | 14 +-
src/router_node.c | 51 ++++--
src/server.c | 12 +-
src/server_private.h | 2 +-
tests/CMakeLists.txt | 2 +-
tests/system_tests_link_routes.py | 139 ++++++++++++++++
tests/system_tests_router_mesh.py | 28 ++++
tests/test-sender.c | 43 +++--
tests/test_broker.py | 5 +
19 files changed, 521 insertions(+), 85 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 02/02: DISPATCH-1579: add logic to activate all
links on session Q3 clear
Posted by kg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kgiusti pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
commit 074efc57c678d8e06a8fefb0a3d6a2f88c0102c6
Author: Kenneth Giusti <kg...@apache.org>
AuthorDate: Thu Feb 27 13:52:32 2020 -0500
DISPATCH-1579: add logic to activate all links on session Q3 clear
This closes #693
---
include/qpid/dispatch/container.h | 10 +++
include/qpid/dispatch/message.h | 20 +++--
src/container.c | 179 ++++++++++++++++++++++++++++++++++----
src/message.c | 7 +-
src/router_node.c | 46 ++++++++--
src/server.c | 12 ++-
tests/CMakeLists.txt | 2 +-
tests/system_tests_link_routes.py | 139 +++++++++++++++++++++++++++++
tests/system_tests_router_mesh.py | 7 +-
tests/test-sender.c | 43 +++++----
tests/test_broker.py | 5 ++
11 files changed, 413 insertions(+), 57 deletions(-)
diff --git a/include/qpid/dispatch/container.h b/include/qpid/dispatch/container.h
index 905c385..86ea2e2 100644
--- a/include/qpid/dispatch/container.h
+++ b/include/qpid/dispatch/container.h
@@ -95,9 +95,11 @@ typedef enum {
typedef struct qd_node_t qd_node_t;
+typedef struct qd_session_t qd_session_t;
typedef struct qd_link_t qd_link_t;
ALLOC_DECLARE(qd_link_t);
+DEQ_DECLARE(qd_link_t, qd_link_list_t);
typedef bool (*qd_container_delivery_handler_t) (void *node_context, qd_link_t *link);
typedef void (*qd_container_disposition_handler_t) (void *node_context, qd_link_t *link, pn_delivery_t *pnd);
@@ -226,6 +228,14 @@ void qd_link_detach(qd_link_t *link);
void qd_link_free(qd_link_t *link);
void *qd_link_get_node_context(const qd_link_t *link);
void qd_link_restart_rx(qd_link_t *link);
+void qd_link_q3_block(qd_link_t *link);
+void qd_link_q3_unblock(qd_link_t *link);
+
+qd_session_t *qd_session(pn_session_t *pn_ssn);
+void qd_session_cleanup(qd_connection_t *qd_conn);
+void qd_session_free(qd_session_t *qd_ssn);
+bool qd_session_is_q3_blocked(const qd_session_t *qd_ssn);
+qd_link_list_t *qd_session_q3_blocked_links(qd_session_t *qd_ssn);
///@}
#endif
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 0d669c4..9ed2bc2 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -39,12 +39,22 @@
// DISPATCH-807 Queue depth limits
// upper and lower limits for bang bang hysteresis control
//
-// Q2 defines the number of buffers allowed in a message's buffer chain
-#define QD_QLIMIT_Q2_UPPER 256
-#define QD_QLIMIT_Q2_LOWER 128
+// Q2 defines the maximum number of buffers allowed in a message's buffer
+// chain. This limits the number of bytes that will be read from an incoming
+// link (pn_link_recv) for the current message. Once Q2 is enabled no further
+// pn_link_recv calls will be done on the link. Q2 remains in effect until enough
+// bytes have been consumed by the outgoing link(s) to drop the number of
+// buffered bytes below the lower threshold.
+#define QD_QLIMIT_Q2_UPPER 256 // disable pn_link_recv (qd_buffer_t's)
+#define QD_QLIMIT_Q2_LOWER 128 // re-enable pn_link_recv
//
-// Q3 defines the number of bytes allowed in a session's outgoing_bytes
-#define QD_QLIMIT_Q3_UPPER (256 * 1000)
+// Q3 limits the number of bytes allowed to be buffered in a session's outgoing
+// buffer. Once the Q3 upper limit is hit (read via pn_session_outgoing_bytes),
+// pn_link_send will no longer be called for ALL outgoing links sharing the
+// session. When enough outgoing bytes have been drained below the lower limit
+// pn_link_sends will resume.
+#define QD_QLIMIT_Q3_UPPER (QD_QLIMIT_Q3_LOWER * 2) // in pn_buffer_t's
+#define QD_QLIMIT_Q3_LOWER (QD_QLIMIT_Q2_UPPER * 2) // 2 == a guess
// Callback for status change (confirmed persistent, loaded-in-memory, etc.)
diff --git a/src/container.c b/src/container.c
index 986ee9c..50a05a7 100644
--- a/src/container.c
+++ b/src/container.c
@@ -61,13 +61,23 @@ struct qd_link_t {
pn_snd_settle_mode_t remote_snd_settle_mode;
qd_link_ref_list_t ref_list;
bool q2_limit_unbounded;
+ bool q3_blocked;
+ DEQ_LINKS_N(Q3, qd_link_t); ///< Q3 blocked links
};
-DEQ_DECLARE(qd_link_t, qd_link_list_t);
-
ALLOC_DEFINE(qd_link_t);
ALLOC_DEFINE(qd_link_ref_t);
+/** Encapsulates a proton session */
+struct qd_session_t {
+ DEQ_LINKS(qd_session_t);
+ pn_session_t *pn_sess;
+ qd_link_list_t q3_blocked_links; ///< Q3 blocked if !empty
+};
+
+ALLOC_DECLARE(qd_session_t);
+ALLOC_DEFINE(qd_session_t);
+
typedef struct qdc_node_type_t {
DEQ_LINKS(struct qdc_node_type_t);
const qd_node_type_t *ntype;
@@ -298,6 +308,22 @@ static void close_links(qd_container_t *container, pn_connection_t *conn, bool p
}
+/** cleanup_link
+ * Prior to freeing qd_link_t - safe to call during shutdown
+ */
+static void cleanup_link(qd_link_t *link)
+{
+ if (link) {
+ if (link->q3_blocked)
+ qd_link_q3_unblock(link);
+ if (link->pn_link) {
+ pn_link_set_context(link->pn_link, NULL);
+ link->pn_link = 0;
+ }
+ link->pn_sess = 0;
+ }
+}
+
static int close_handler(qd_container_t *container, pn_connection_t *conn, qd_connection_t* qd_conn)
{
//
@@ -307,6 +333,7 @@ static int close_handler(qd_container_t *container, pn_connection_t *conn, qd_co
if (qd_conn)
qd_conn->closed = true;
close_links(container, conn, true);
+ qd_session_cleanup(qd_conn);
if (qd_conn)
notify_closed(container, qd_conn, qd_connection_get_context(qd_conn));
return 0;
@@ -457,10 +484,12 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
qd_conn->closed = true;
if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
close_links(container, conn, false);
+ qd_session_cleanup(qd_conn);
pn_connection_close(conn);
qd_conn_event_batch_complete(container, qd_conn, true);
} else if (pn_connection_state(conn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) {
close_links(container, conn, false);
+ qd_session_cleanup(qd_conn);
notify_closed(container, qd_conn, qd_connection_get_context(qd_conn));
qd_conn_event_batch_complete(container, qd_conn, true);
}
@@ -470,12 +499,24 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) {
ssn = pn_event_session(event);
if (pn_session_state(ssn) & PN_LOCAL_UNINIT) {
+ // remote created new session
+ assert(pn_session_get_context(ssn) == 0);
+ qd_session_t *qd_ssn = qd_session(ssn);
+ if (!qd_ssn) {
+ pn_condition_t *cond = pn_session_condition(ssn);
+ pn_condition_set_name(cond, QD_AMQP_COND_INTERNAL_ERROR);
+ pn_condition_set_description(cond, "Insufficient memory");
+ pn_session_close(ssn);
+ break;
+ }
if (qd_conn->policy_settings) {
if (!qd_policy_approve_amqp_session(ssn, qd_conn)) {
+ qd_session_free(qd_ssn);
break;
}
qd_conn->n_sessions++;
}
+ pn_session_set_context(ssn, qd_ssn);
qd_policy_apply_session_settings(ssn, qd_conn);
pn_session_open(ssn);
}
@@ -501,6 +542,8 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
}
if (pn_session_state(ssn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) {
+ qd_session_t *qd_ssn = pn_session_get_context(ssn);
+ qd_session_free(qd_ssn);
add_session_to_free_list(&qd_conn->free_link_session_list, ssn);
}
break;
@@ -561,6 +604,8 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
pn_session_close(ssn);
}
else if (pn_session_state(ssn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) {
+ qd_session_t *qd_ssn = pn_session_get_context(ssn);
+ qd_session_free(qd_ssn);
add_session_to_free_list(&qd_conn->free_link_session_list, ssn);
}
}
@@ -704,6 +749,7 @@ void qd_container_free(qd_container_t *container)
qd_link_t *link = DEQ_HEAD(container->links);
while (link) {
DEQ_REMOVE_HEAD(container->links);
+ cleanup_link(link);
free_qd_link_t(link);
link = DEQ_HEAD(container->links);
}
@@ -856,26 +902,36 @@ qd_lifetime_policy_t qd_container_node_get_life_policy(const qd_node_t *node)
qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char* name, qd_session_class_t ssn_class)
{
+ const qd_server_config_t * cf = qd_connection_config(conn);
+
+ pn_session_t *pn_ssn = conn->pn_sessions[ssn_class];
+ if (!pn_ssn) {
+ pn_ssn = pn_session(qd_connection_pn(conn));
+ if (!pn_ssn) {
+ return NULL;
+ }
+ qd_session_t *qd_ssn = qd_session(pn_ssn);
+ if (!qd_ssn) {
+ pn_session_free(pn_ssn);
+ return NULL; // PN_LOCAL_CLOSE event will free pn_ssn
+ }
+
+ conn->pn_sessions[ssn_class] = pn_ssn;
+ pn_session_set_incoming_capacity(pn_ssn, cf->incoming_capacity);
+ pn_session_open(pn_ssn);
+ }
+
qd_link_t *link = new_qd_link_t();
if (!link) {
- return NULL;
+ return NULL; // ok to keep qd_session around - will reuse & free on conn close
}
- const qd_server_config_t * cf = qd_connection_config(conn);
-
ZERO(link);
+
sys_mutex_lock(node->container->lock);
DEQ_INSERT_TAIL(node->container->links, link);
sys_mutex_unlock(node->container->lock);
- pn_session_t *ssn = conn->pn_sessions[ssn_class];
- if (!ssn) {
- ssn = pn_session(qd_connection_pn(conn));
- conn->pn_sessions[ssn_class] = ssn;
- pn_session_set_incoming_capacity(ssn, cf->incoming_capacity);
- pn_session_open(ssn);
- }
-
- link->pn_sess = ssn;
+ link->pn_sess = pn_ssn;
if (dir == QD_OUTGOING)
link->pn_link = pn_sender(link->pn_sess, name);
@@ -896,11 +952,6 @@ qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, c
void qd_link_free(qd_link_t *link)
{
if (!link) return;
- if (link->pn_link) {
- pn_link_set_context(link->pn_link, NULL);
- link->pn_link = 0;
- }
- link->pn_sess = 0;
qd_container_t *container = link->node->container;
sys_mutex_lock(container->lock);
DEQ_REMOVE(container->links, link);
@@ -909,6 +960,7 @@ void qd_link_free(qd_link_t *link)
qd_node_t *node = link->node;
node->ntype->link_abandoned_deliveries_handler(node->context, link);
+ cleanup_link(link);
free_qd_link_t(link);
}
@@ -1050,3 +1102,92 @@ void *qd_link_get_node_context(const qd_link_t *link)
{
return (link && link->node) ? link->node->context : 0;
}
+
+
+/** sending link has entered Q3 flow control */
+void qd_link_q3_block(qd_link_t *link)
+{
+ assert(link);
+ if (!link->q3_blocked && link->pn_sess) {
+ qd_session_t *qd_ssn = pn_session_get_context(link->pn_sess);
+ assert(qd_ssn);
+ link->q3_blocked = true;
+ DEQ_INSERT_TAIL_N(Q3, qd_ssn->q3_blocked_links, link);
+ }
+}
+
+
+void qd_link_q3_unblock(qd_link_t *link)
+{
+ assert(link);
+ if (link->q3_blocked) {
+ qd_session_t *qd_ssn = pn_session_get_context(link->pn_sess);
+ assert(qd_ssn);
+ DEQ_REMOVE_N(Q3, qd_ssn->q3_blocked_links, link);
+ link->q3_blocked = false;
+ }
+}
+
+
+qd_session_t *qd_session(pn_session_t *pn_ssn)
+{
+ assert(pn_ssn);
+ qd_session_t *qd_ssn = pn_session_get_context(pn_ssn);
+ if (!qd_ssn) {
+ qd_ssn = new_qd_session_t();
+ if (qd_ssn) {
+ ZERO(qd_ssn);
+ qd_ssn->pn_sess = pn_ssn;
+ DEQ_INIT(qd_ssn->q3_blocked_links);
+ pn_session_set_context(pn_ssn, qd_ssn);
+ }
+ }
+ return qd_ssn;
+}
+
+
+void qd_session_free(qd_session_t *qd_ssn)
+{
+ if (qd_ssn) {
+ qd_link_t *link = DEQ_HEAD(qd_ssn->q3_blocked_links);
+ while (link) {
+ qd_link_q3_unblock(link); // removes link from list
+ link = DEQ_HEAD(qd_ssn->q3_blocked_links);
+ }
+ if (qd_ssn->pn_sess) {
+ pn_session_set_context(qd_ssn->pn_sess, 0);
+ }
+ free_qd_session_t(qd_ssn);
+ }
+}
+
+
+qd_link_list_t *qd_session_q3_blocked_links(qd_session_t *qd_ssn)
+{
+ return qd_ssn ? &qd_ssn->q3_blocked_links : 0;
+}
+
+
+bool qd_session_is_q3_blocked(const qd_session_t *qd_ssn)
+{
+ return qd_ssn && !DEQ_IS_EMPTY(qd_ssn->q3_blocked_links);
+}
+
+
+
+/** release all qd_session_t instances for the connection
+ * called prior to releasing the qd_connection_t
+ */
+void qd_session_cleanup(qd_connection_t *qd_conn)
+{
+ pn_connection_t *pn_conn = qd_conn->pn_conn;
+ if (!pn_conn)
+ return;
+
+ pn_session_t *pn_ssn = pn_session_head(pn_conn, 0);
+ while (pn_ssn) {
+ qd_session_t *qd_ssn = pn_session_get_context(pn_ssn);
+ qd_session_free(qd_ssn);
+ pn_ssn = pn_session_next(pn_ssn, 0);
+ }
+}
diff --git a/src/message.c b/src/message.c
index 77ac056..a3420a1 100644
--- a/src/message.c
+++ b/src/message.c
@@ -1684,11 +1684,12 @@ void qd_message_send(qd_message_t *in_msg,
buf = msg->cursor.buffer;
- pn_session_t *pns = pn_link_session(pnl);
+ pn_session_t *pns = pn_link_session(pnl);
+ const size_t q3_upper = BUFFER_SIZE * QD_QLIMIT_Q3_UPPER;
while (!content->aborted
&& buf
- && pn_session_outgoing_bytes(pns) <= QD_QLIMIT_Q3_UPPER) {
+ && pn_session_outgoing_bytes(pns) < q3_upper) {
// This will send the remaining data in the buffer if any. There may be
// zero bytes left to send if we stopped here last time and there was
@@ -1790,7 +1791,7 @@ void qd_message_send(qd_message_t *in_msg,
}
}
- *q3_stalled = (pn_session_outgoing_bytes(pns) > QD_QLIMIT_Q3_UPPER);
+ *q3_stalled = (pn_session_outgoing_bytes(pns) >= q3_upper);
}
diff --git a/src/router_node.c b/src/router_node.c
index ea35944..73d0ff8 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -786,19 +786,45 @@ static int AMQP_link_attach_handler(void* context, qd_link_t *link)
/**
- * Handler for flow events on links
+ * Handler for flow events on links. Flow updates include session window
+ * state, which needs to be checked for unblocking Q3.
*/
static int AMQP_link_flow_handler(void* context, qd_link_t *link)
{
- qd_router_t *router = (qd_router_t*) context;
- qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link);
- pn_link_t *pnlink = qd_link_pn(link);
-
- if (!rlink)
- return 0;
+ qd_router_t *router = (qd_router_t*) context;
+ pn_link_t *pnlink = qd_link_pn(link);
+ qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link);
- qdr_link_flow(router->router_core, rlink, pn_link_remote_credit(pnlink), pn_link_get_drain(pnlink));
+ if (rlink) {
+ qdr_link_flow(router->router_core, rlink, pn_link_remote_credit(pnlink), pn_link_get_drain(pnlink));
+ }
+ // check if Q3 can be unblocked
+ pn_session_t *pn_ssn = pn_link_session(pnlink);
+ if (pn_ssn) {
+ qd_session_t *qd_ssn = pn_session_get_context(pn_ssn);
+ if (qd_ssn && qd_session_is_q3_blocked(qd_ssn)) {
+ // Q3 blocked - have we drained enough outgoing bytes?
+ const size_t q3_lower = BUFFER_SIZE * QD_QLIMIT_Q3_LOWER;
+ if (pn_session_outgoing_bytes(pn_ssn) < q3_lower) {
+ // yes. We must now unblock all links that have been blocked by Q3
+ qd_link_list_t *blinks = qd_session_q3_blocked_links(qd_ssn);
+ qd_link_t *blink = DEQ_HEAD(*blinks);
+ while (blink) {
+ qd_link_q3_unblock(blink); // removes from blinks list!
+ if (blink != link) { // already flowed this link
+ rlink = (qdr_link_t *) qd_link_get_context(blink);
+ if (rlink) {
+ pnlink = qd_link_pn(blink);
+ // signalling flow to the core causes the link to be re-activated
+ qdr_link_flow(router->router_core, rlink, pn_link_remote_credit(pnlink), pn_link_get_drain(pnlink));
+ }
+ }
+ blink = DEQ_HEAD(*blinks);
+ }
+ }
+ }
+ }
return 0;
}
@@ -1665,8 +1691,10 @@ static uint64_t CORE_link_deliver(void *context, qdr_link_t *link, qdr_delivery_
qd_message_send(msg_out, qlink, qdr_link_strip_annotations_out(link), &restart_rx, &q3_stalled);
- if (q3_stalled)
+ if (q3_stalled) {
+ qd_link_q3_block(qlink);
qdr_link_stalled_outbound(link);
+ }
if (restart_rx) {
qd_link_restart_rx(qd_message_get_receiving_link(msg_out));
diff --git a/src/server.c b/src/server.c
index 4bc78fd..ae1d886 100644
--- a/src/server.c
+++ b/src/server.c
@@ -1273,10 +1273,19 @@ qd_server_t *qd_server(qd_dispatch_t *qd, int thread_count, const char *containe
void qd_server_free(qd_server_t *qd_server)
{
if (!qd_server) return;
- pn_proactor_free(qd_server->proactor);
qd_connection_t *ctx = DEQ_HEAD(qd_server->conn_list);
while (ctx) {
+ qd_log(qd_server->log_source, QD_LOG_INFO,
+ "[C%"PRIu64"] Closing connection on shutdown",
+ ctx->connection_id);
DEQ_REMOVE_HEAD(qd_server->conn_list);
+ if (ctx->pn_conn) {
+ pn_transport_t *tport = pn_connection_transport(ctx->pn_conn);
+ if (tport)
+ pn_transport_set_context(tport, 0); /* for transport_tracer */
+ qd_session_cleanup(ctx);
+ pn_connection_set_context(ctx->pn_conn, 0);
+ }
if (ctx->free_user_id) free((char*)ctx->user_id);
sys_mutex_free(ctx->deferred_call_lock);
free(ctx->name);
@@ -1284,6 +1293,7 @@ void qd_server_free(qd_server_t *qd_server)
free_qd_connection_t(ctx);
ctx = DEQ_HEAD(qd_server->conn_list);
}
+ pn_proactor_free(qd_server->proactor);
qd_timer_finalize();
sys_mutex_free(qd_server->lock);
sys_mutex_free(qd_server->conn_activation_lock);
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 81257b8..e34db23 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -53,7 +53,7 @@ add_executable(unit_tests_size ${unit_test_size_SOURCES})
target_link_libraries(unit_tests_size qpid-dispatch)
add_executable(test-sender test-sender.c)
-target_link_libraries(test-sender ${Proton_LIBRARIES})
+target_link_libraries(test-sender ${Proton_LIBRARIES} qpid-dispatch)
add_executable(test-receiver test-receiver.c)
target_link_libraries(test-receiver ${Proton_LIBRARIES})
diff --git a/tests/system_tests_link_routes.py b/tests/system_tests_link_routes.py
index fea6c3a..81513d2 100644
--- a/tests/system_tests_link_routes.py
+++ b/tests/system_tests_link_routes.py
@@ -33,6 +33,7 @@ from system_test import QdManager
from system_test import MgmtMsgProxy
from system_test import unittest, QdManager
from test_broker import FakeBroker
+from test_broker import FakeService
from proton import Delivery
from proton import Message
@@ -2370,5 +2371,143 @@ class SendReceive(MessagingHandler):
def run(self):
Container(self).run()
+
+class LinkRoute3Hop(TestCase):
+ """
+ Sets up a linear 3 hop router network for testing multi-hop link routes.
+
+ +---------+ +---------+ +---------+ +------------------+
+ | | <------ | | <----- | |<----| blocking_senders |
+ | QDR.A | | QDR.B | | QDR.C | +------------------+
+ | | ------> | | ------> | | +--------------------+
+ +---------+ +---------+ +---------+---->| blocking_receivers |
+ ^ +--------------------+
+ |
+ V
+ +-------------+
+ | FakeService |
+ +-------------+
+
+ """
+
+ @classmethod
+ def setUpClass(cls):
+ super(LinkRoute3Hop, cls).setUpClass()
+
+ b_port = cls.tester.get_port()
+ configs = [
+ # QDR.A:
+ [('router', {'mode': 'interior', 'id': 'QDR.A'}),
+ # for client access
+ ('listener', {'role': 'normal',
+ 'host': '0.0.0.0',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+ # for fake service:
+ ('listener', {'role': 'route-container',
+ 'host': '0.0.0.0',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+ # to connect to the QDR.B
+ ('connector', {'role': 'inter-router',
+ 'host': '127.0.0.1',
+ 'port': b_port,
+ 'saslMechanisms': 'ANONYMOUS'}),
+ # the routes
+ ('linkRoute', {'prefix': 'closest/test-client', 'containerId': 'FakeService', 'direction': 'in'}),
+ ('linkRoute', {'prefix': 'closest/test-client', 'containerId': 'FakeService', 'direction': 'out'})
+ ],
+ # QDR.B:
+ [('router', {'mode': 'interior', 'id': 'QDR.B'}),
+ # for client connections
+ ('listener', {'role': 'normal',
+ 'host': '0.0.0.0',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+ # for inter-router connections from QDR.A and QDR.C
+ ('listener', {'role': 'inter-router',
+ 'host': '0.0.0.0',
+ 'port': b_port,
+ 'saslMechanisms': 'ANONYMOUS'}),
+ ('linkRoute', {'prefix': 'closest/test-client', 'direction': 'in'}),
+ ('linkRoute', {'prefix': 'closest/test-client', 'direction': 'out'})
+ ],
+ # QDR.C
+ [('router', {'mode': 'interior', 'id': 'QDR.C'}),
+ # for client connections
+ ('listener', {'role': 'normal',
+ 'host': '0.0.0.0',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+ # to connect to the QDR.B
+ ('connector', {'role': 'inter-router',
+ 'host': '127.0.0.1',
+ 'port': b_port,
+ 'saslMechanisms': 'ANONYMOUS'}),
+ ('linkRoute', {'prefix': 'closest/test-client', 'direction': 'in'}),
+ ('linkRoute', {'prefix': 'closest/test-client', 'direction': 'out'})
+ ]
+ ]
+
+ cls.routers=[]
+ for c in configs:
+ config = Qdrouterd.Config(c)
+ cls.routers.append(cls.tester.qdrouterd(config=config, wait=False))
+ cls.QDR_A = cls.routers[0]
+ cls.QDR_B = cls.routers[1]
+ cls.QDR_C = cls.routers[2]
+
+ cls.QDR_A.wait_router_connected('QDR.B')
+ cls.QDR_B.wait_router_connected('QDR.A')
+ cls.QDR_B.wait_router_connected('QDR.C')
+ cls.QDR_C.wait_router_connected('QDR.B')
+
+ cls.fake_service = FakeService(cls.QDR_A.addresses[1],
+ container_id="FakeService")
+ cls.QDR_C.wait_address("closest/test-client",
+ remotes=1)
+
+ def test_01_parallel_link_routes(self):
+ """
+ Verify Q2/Q3 recovery in the case of multiple link-routes sharing the
+ same session.
+ """
+ send_clients = 10
+ send_batch = 25
+ total = send_clients * send_batch
+
+ start_in = self.fake_service.in_count
+ start_out = self.fake_service.out_count
+
+ rx = self.popen(["test-receiver",
+ "-a", self.QDR_C.addresses[0],
+ "-c", str(total),
+ "-s", "closest/test-client"],
+ expect=Process.EXIT_OK)
+
+ def _spawn_sender(x):
+ return self.popen( ["test-sender",
+ "-a", self.QDR_C.addresses[0],
+ "-c", str(send_batch),
+ "-i", "TestSender-%s" % x,
+ "-sx", # huge message size to trigger Q2/Q3
+ "-t", "closest/test-client"],
+ expect=Process.EXIT_OK)
+
+ senders = [_spawn_sender(s) for s in range(send_clients)]
+
+ if rx.wait(timeout=TIMEOUT):
+ raise Exception("Receiver failed to consume all messages in=%s out=%s",
+ self.fake_service.in_count,
+ self.fake_service.out_count)
+ for tx in senders:
+ out_text, out_err = tx.communicate(timeout=TIMEOUT)
+ if tx.returncode:
+ raise Exception("Sender failed: %s %s" % (out_text, out_err))
+
+ self.assertEqual(start_in + total, self.fake_service.in_count)
+ self.assertEqual(start_out + total, self.fake_service.out_count)
+
+
if __name__ == '__main__':
unittest.main(main_module())
diff --git a/tests/system_tests_router_mesh.py b/tests/system_tests_router_mesh.py
index e8fab01..c235d98 100644
--- a/tests/system_tests_router_mesh.py
+++ b/tests/system_tests_router_mesh.py
@@ -212,12 +212,11 @@ class ThreeRouterTest(TestCase):
def test_06_parallel_priority(self):
"""
Create 10 senders each with a different priority. Send large messages
- - large enough to trigger Qx flow control. Ensure all messages arrive
- as expected
+ - large enough to trigger Qx flow control (sender argument "-sx").
+ Ensure all messages arrive as expected.
"""
-
priorities = 10
- send_batch = 150
+ send_batch = 25
total = priorities * send_batch
rx = self.spawn_receiver(self.RouterC,
diff --git a/tests/test-sender.c b/tests/test-sender.c
index 1386619..38b6823 100644
--- a/tests/test-sender.c
+++ b/tests/test-sender.c
@@ -40,22 +40,20 @@
#include "proton/event.h"
#include "proton/handlers.h"
+#include <qpid/dispatch/buffer.h>
+#include <qpid/dispatch/message.h>
+
#define BOOL2STR(b) ((b)?"true":"false")
-#define BODY_SIZE_SMALL 100
-#define BODY_SIZE_MEDIUM 2000
-#define BODY_SIZE_LARGE 65 * 1024
-#define BODY_SIZE_HUGE 1024 * 1024 // will trigger Q2/Q3
+#define BODY_SIZE_SMALL 100L
+#define BODY_SIZE_MEDIUM ((long int)((4 * 1024) + 1))
+#define BODY_SIZE_LARGE ((long int)((65 * 1024) + 1))
+#define BODY_SIZE_HUGE ((long int)((BUFFER_SIZE * QD_QLIMIT_Q3_UPPER * 3) + 1))
#define DEFAULT_PRIORITY 4
// body data - block of 0's
//
-char _payload[BODY_SIZE_HUGE] = {0};
-pn_bytes_t body_data = {
- .size = 0,
- .start = _payload,
-};
bool stop = false;
@@ -69,15 +67,16 @@ uint64_t rejected = 0;
uint64_t modified = 0;
uint64_t released = 0;
-
-bool use_anonymous = false; // use anonymous link if true
-bool presettle = false; // true = send presettled
+char body_data_pattern = 'B'; // fill body data
+bool use_anonymous = false; // use anonymous link if true
+bool presettle = false; // true = send presettled
bool add_annotations = false;
-int body_size = BODY_SIZE_SMALL;
+long int body_size = BODY_SIZE_SMALL;
bool drop_connection = false;
unsigned int priority = DEFAULT_PRIORITY;
// buffer for encoded message
+pn_bytes_t body_data = {0};
char *encode_buffer = NULL;
size_t encode_buffer_size = 0; // size of malloced memory
size_t encoded_data_size = 0; // length of encoded content
@@ -173,6 +172,17 @@ void generate_message(void)
pn_data_t *body = pn_message_body(out_message);
pn_data_clear(body);
+
+ if (!body_data.start) {
+ char *ptr = malloc(BODY_SIZE_HUGE);
+ if (!ptr) {
+ perror("Out of memory!");
+ exit(-1);
+ }
+ memset(ptr, (int)body_data_pattern, BODY_SIZE_HUGE);
+ body_data.start = ptr;
+ }
+
body_data.size = body_size;
pn_data_put_binary(body, body_data);
@@ -229,6 +239,7 @@ static void delete_handler(pn_handler_t *handler)
{
free(encode_buffer);
pn_message_free(out_message);
+ free((void *)body_data.start);
}
@@ -339,13 +350,14 @@ static void usage(void)
printf("-c \t# of messages to send, 0 == nonstop [%"PRIu64"]\n", limit);
printf("-i \tContainer name [%s]\n", container_name);
printf("-n \tUse an anonymous link [%s]\n", BOOL2STR(use_anonymous));
- printf("-s \tBody size in bytes ('s'=%d 'm'=%d 'l'=%d 'x'=%d) [%d]\n",
+ printf("-s \tBody size in bytes ('s'=%ld 'm'=%ld 'l'=%ld 'x'=%ld) [%ld]\n",
BODY_SIZE_SMALL, BODY_SIZE_MEDIUM, BODY_SIZE_LARGE, BODY_SIZE_HUGE, body_size);
printf("-t \tTarget address [%s]\n", target_address);
printf("-u \tSend all messages presettled [%s]\n", BOOL2STR(presettle));
printf("-M \tAdd dummy Message Annotations section [off]\n");
printf("-E \tExit without cleanly closing the connection [off]\n");
printf("-p \tMessage priority [%d]\n", priority);
+ printf("-X \tMessage body data pattern [%c]\n", (char)body_data_pattern);
exit(1);
}
@@ -354,7 +366,7 @@ int main(int argc, char** argv)
/* command line options */
opterr = 0;
int c;
- while ((c = getopt(argc, argv, "ha:c:i:ns:t:uMEp:")) != -1) {
+ while ((c = getopt(argc, argv, "ha:c:i:ns:t:uMEp:X:")) != -1) {
switch(c) {
case 'h': usage(); break;
case 'a': host_address = optarg; break;
@@ -378,6 +390,7 @@ int main(int argc, char** argv)
case 'u': presettle = true; break;
case 'M': add_annotations = true; break;
case 'E': drop_connection = true; break;
+ case 'X': body_data_pattern = optarg[0]; break;
case 'p':
if (sscanf(optarg, "%u", &priority) != 1)
usage();
diff --git a/tests/test_broker.py b/tests/test_broker.py
index 2f101a7..95d59a2 100644
--- a/tests/test_broker.py
+++ b/tests/test_broker.py
@@ -92,6 +92,7 @@ class FakeBroker(MessagingHandler):
self.acceptor = None
self.in_count = 0
self.out_count = 0
+ self.link_errors = 0
self._connections = []
self._error = None
self._container = Container(self)
@@ -149,6 +150,10 @@ class FakeBroker(MessagingHandler):
if link.source.address in self.queues and self.queues[link.source.address].unsubscribe(link):
del self.queues[link.source.address]
+ def on_link_error(self, event):
+ self.link_errors += 1
+ self.on_link_closing(event)
+
def on_link_closing(self, event):
if event.link.is_sender:
self._unsubscribe(event.link)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 01/02: DISPATCH-1579: open dedicated inter-router
sessions for data and control links
Posted by kg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kgiusti pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
commit 0264301cf9a1fb6e767f75d4e1aeabecf7355e2a
Author: Kenneth Giusti <kg...@apache.org>
AuthorDate: Wed Feb 19 13:45:58 2020 -0500
DISPATCH-1579: open dedicated inter-router sessions for data and control links
---
include/qpid/dispatch/container.h | 28 +++++++++++++++++++++-
include/qpid/dispatch/router_core.h | 5 ++--
src/container.c | 30 ++++++++++++++++--------
src/router_core/connections.c | 28 +++++++++++++---------
src/router_core/core_link_endpoint.c | 3 ++-
src/router_core/forwarder.c | 1 +
src/router_core/modules/edge_router/addr_proxy.c | 10 ++++----
src/router_core/route_control.c | 3 ++-
src/router_core/router_core_private.h | 14 ++++++-----
src/router_node.c | 5 ++--
src/server_private.h | 2 +-
tests/system_tests_router_mesh.py | 29 +++++++++++++++++++++++
tests/test-sender.c | 4 ++--
13 files changed, 121 insertions(+), 41 deletions(-)
diff --git a/include/qpid/dispatch/container.h b/include/qpid/dispatch/container.h
index cf1b0d4..905c385 100644
--- a/include/qpid/dispatch/container.h
+++ b/include/qpid/dispatch/container.h
@@ -68,6 +68,32 @@ typedef enum {
} qd_detach_type_t;
+/**
+ * Session Class
+ *
+ * Used when creating new links from the router. A connection maintains a set
+ * of sessions over which links can be created. The session class indicates
+ * which session to use when creating a link.
+ */
+typedef enum {
+ QD_SSN_ENDPOINT, ///< client data links
+ QD_SSN_ROUTER_CONTROL, ///< router protocol
+ QD_SSN_ROUTER_DATA_PRI_0, ///< inter-router data links (by priority)
+ QD_SSN_ROUTER_DATA_PRI_1,
+ QD_SSN_ROUTER_DATA_PRI_2,
+ QD_SSN_ROUTER_DATA_PRI_3,
+ QD_SSN_ROUTER_DATA_PRI_4,
+ QD_SSN_ROUTER_DATA_PRI_5,
+ QD_SSN_ROUTER_DATA_PRI_6,
+ QD_SSN_ROUTER_DATA_PRI_7,
+ QD_SSN_ROUTER_DATA_PRI_8,
+ QD_SSN_ROUTER_DATA_PRI_9,
+ QD_SSN_CORE_ENDPOINT, ///< core subscriptions
+ QD_SSN_LINK_ROUTE, ///< link routes
+ QD_SSN_CLASS_COUNT
+} qd_session_class_t;
+
+
typedef struct qd_node_t qd_node_t;
typedef struct qd_link_t qd_link_t;
@@ -160,7 +186,7 @@ void qd_container_node_set_context(qd_node_t *node, void *node_context);
qd_dist_mode_t qd_container_node_get_dist_modes(const qd_node_t *node);
qd_lifetime_policy_t qd_container_node_get_life_policy(const qd_node_t *node);
-qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char *name);
+qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char *name, qd_session_class_t);
void qd_link_free(qd_link_t *link);
/**
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index 18bef79..7339714 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -700,8 +700,9 @@ void qdr_link_flow(qdr_core_t *core, qdr_link_t *link, int credit, bool drain_mo
*/
void qdr_link_set_drained(qdr_core_t *core, qdr_link_t *link);
-typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target);
+typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target,
+ qd_session_class_t);
typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
qdr_terminus_t *source, qdr_terminus_t *target);
typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
diff --git a/src/container.c b/src/container.c
index 8b19074..986ee9c 100644
--- a/src/container.c
+++ b/src/container.c
@@ -484,8 +484,12 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
case PN_SESSION_LOCAL_CLOSE :
ssn = pn_event_session(event);
- if (ssn == qd_conn->pn_sess)
- qd_conn->pn_sess = 0;
+ for (int i = 0; i < QD_SSN_CLASS_COUNT; ++i) {
+ if (ssn == qd_conn->pn_sessions[i]) {
+ qd_conn->pn_sessions[i] = 0;
+ break;
+ }
+ }
pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
while (pn_link) {
if (pn_link_session(pn_link) == ssn) {
@@ -503,8 +507,12 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event,
case PN_SESSION_REMOTE_CLOSE :
ssn = pn_event_session(event);
- if (ssn == qd_conn->pn_sess)
- qd_conn->pn_sess = 0;
+ for (int i = 0; i < QD_SSN_CLASS_COUNT; ++i) {
+ if (ssn == qd_conn->pn_sessions[i]) {
+ qd_conn->pn_sessions[i] = 0;
+ break;
+ }
+ }
if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) {
if (pn_session_state(ssn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
@@ -846,7 +854,7 @@ qd_lifetime_policy_t qd_container_node_get_life_policy(const qd_node_t *node)
}
-qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char* name)
+qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char* name, qd_session_class_t ssn_class)
{
qd_link_t *link = new_qd_link_t();
if (!link) {
@@ -859,13 +867,15 @@ qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, c
DEQ_INSERT_TAIL(node->container->links, link);
sys_mutex_unlock(node->container->lock);
- if (!conn->pn_sess) {
- conn->pn_sess = pn_session(qd_connection_pn(conn));
- pn_session_set_incoming_capacity(conn->pn_sess, cf->incoming_capacity);
- pn_session_open(conn->pn_sess);
+ pn_session_t *ssn = conn->pn_sessions[ssn_class];
+ if (!ssn) {
+ ssn = pn_session(qd_connection_pn(conn));
+ conn->pn_sessions[ssn_class] = ssn;
+ pn_session_set_incoming_capacity(ssn, cf->incoming_capacity);
+ pn_session_open(ssn);
}
- link->pn_sess = conn->pn_sess;
+ link->pn_sess = ssn;
if (dir == QD_OUTGOING)
link->pn_link = pn_sender(link->pn_sess, name);
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 31e523d..a6672c9 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -19,6 +19,7 @@
#include <qpid/dispatch/router_core.h>
#include <qpid/dispatch/discriminator.h>
+#include <qpid/dispatch/static_assert.h>
#include "route_control.h"
#include <qpid/dispatch/amqp.h>
#include <stdio.h>
@@ -295,7 +296,7 @@ int qdr_connection_process(qdr_connection_t *conn)
switch (work->work_type) {
case QDR_CONNECTION_WORK_FIRST_ATTACH :
- core->first_attach_handler(core->user_context, conn, work->link, work->source, work->target);
+ core->first_attach_handler(core->user_context, conn, work->link, work->source, work->target, work->ssn_class);
break;
case QDR_CONNECTION_WORK_SECOND_ATTACH :
@@ -1046,12 +1047,13 @@ static void qdr_link_cleanup_protected_CT(qdr_core_t *core, qdr_connection_t *co
}
-qdr_link_t *qdr_create_link_CT(qdr_core_t *core,
- qdr_connection_t *conn,
- qd_link_type_t link_type,
- qd_direction_t dir,
- qdr_terminus_t *source,
- qdr_terminus_t *target)
+qdr_link_t *qdr_create_link_CT(qdr_core_t *core,
+ qdr_connection_t *conn,
+ qd_link_type_t link_type,
+ qd_direction_t dir,
+ qdr_terminus_t *source,
+ qdr_terminus_t *target,
+ qd_session_class_t ssn_class)
{
//
// Create a new link, initiated by the router core. This will involve issuing a first-attach outbound.
@@ -1093,6 +1095,7 @@ qdr_link_t *qdr_create_link_CT(qdr_core_t *core,
work->link = link;
work->source = source;
work->target = target;
+ work->ssn_class = ssn_class;
char source_str[1000];
char target_str[1000];
@@ -1314,12 +1317,15 @@ static void qdr_connection_opened_CT(qdr_core_t *core, qdr_action_t *action, boo
// inter-router links: Two (in and out) for control, 2 * QDR_N_PRIORITIES for
// routed-message transfer.
//
- (void) qdr_create_link_CT(core, conn, QD_LINK_CONTROL, QD_INCOMING, qdr_terminus_router_control(), qdr_terminus_router_control());
- (void) qdr_create_link_CT(core, conn, QD_LINK_CONTROL, QD_OUTGOING, qdr_terminus_router_control(), qdr_terminus_router_control());
+ (void) qdr_create_link_CT(core, conn, QD_LINK_CONTROL, QD_INCOMING, qdr_terminus_router_control(), qdr_terminus_router_control(), QD_SSN_ROUTER_CONTROL);
+ (void) qdr_create_link_CT(core, conn, QD_LINK_CONTROL, QD_OUTGOING, qdr_terminus_router_control(), qdr_terminus_router_control(), QD_SSN_ROUTER_CONTROL);
+ STATIC_ASSERT((QD_SSN_ROUTER_DATA_PRI_9 - QD_SSN_ROUTER_DATA_PRI_0 + 1) == QDR_N_PRIORITIES, PRIORITY_SESSION_NOT_SAME);
for (int priority = 0; priority < QDR_N_PRIORITIES; ++ priority) {
- (void) qdr_create_link_CT(core, conn, QD_LINK_ROUTER, QD_INCOMING, qdr_terminus_router_data(), qdr_terminus_router_data());
- (void) qdr_create_link_CT(core, conn, QD_LINK_ROUTER, QD_OUTGOING, qdr_terminus_router_data(), qdr_terminus_router_data());
+ // a session is reserved for each priority link
+ qd_session_class_t sc = (qd_session_class_t)(QD_SSN_ROUTER_DATA_PRI_0 + priority);
+ (void) qdr_create_link_CT(core, conn, QD_LINK_ROUTER, QD_INCOMING, qdr_terminus_router_data(), qdr_terminus_router_data(), sc);
+ (void) qdr_create_link_CT(core, conn, QD_LINK_ROUTER, QD_OUTGOING, qdr_terminus_router_data(), qdr_terminus_router_data(), sc);
}
}
}
diff --git a/src/router_core/core_link_endpoint.c b/src/router_core/core_link_endpoint.c
index e48b80e..bbfc84f 100644
--- a/src/router_core/core_link_endpoint.c
+++ b/src/router_core/core_link_endpoint.c
@@ -72,7 +72,8 @@ qdrc_endpoint_t *qdrc_endpoint_create_link_CT(qdr_core_t *core,
ep->desc = desc;
ep->link_context = link_context;
- ep->link = qdr_create_link_CT(core, conn, QD_LINK_ENDPOINT, dir, source, target);
+ ep->link = qdr_create_link_CT(core, conn, QD_LINK_ENDPOINT, dir, source, target,
+ QD_SSN_CORE_ENDPOINT);
ep->link->core_endpoint = ep;
return ep;
diff --git a/src/router_core/forwarder.c b/src/router_core/forwarder.c
index 52e479a..ad3fb02 100644
--- a/src/router_core/forwarder.c
+++ b/src/router_core/forwarder.c
@@ -930,6 +930,7 @@ void qdr_forward_link_direct_CT(qdr_core_t *core,
work->link = out_link;
work->source = source;
work->target = target;
+ work->ssn_class = QD_SSN_LINK_ROUTE;
qdr_connection_enqueue_work_CT(core, conn, work);
diff --git a/src/router_core/modules/edge_router/addr_proxy.c b/src/router_core/modules/edge_router/addr_proxy.c
index af6e5c2..7ed6d08 100644
--- a/src/router_core/modules/edge_router/addr_proxy.c
+++ b/src/router_core/modules/edge_router/addr_proxy.c
@@ -156,7 +156,7 @@ static void add_inlink(qcm_edge_addr_proxy_t *ap, const char *key, qdr_address_t
}
qdr_link_t *link = qdr_create_link_CT(ap->core, ap->edge_conn, QD_LINK_ENDPOINT, QD_INCOMING,
- term, qdr_terminus_normal(0));
+ term, qdr_terminus_normal(0), QD_SSN_ENDPOINT);
qdr_core_bind_address_link_CT(ap->core, addr, link);
addr->edge_inlink = link;
}
@@ -199,7 +199,7 @@ static void add_outlink(qcm_edge_addr_proxy_t *ap, const char *key, qdr_address_
}
qdr_link_t *link = qdr_create_link_CT(ap->core, ap->edge_conn, QD_LINK_ENDPOINT, QD_OUTGOING,
- qdr_terminus_normal(0), term);
+ qdr_terminus_normal(0), term, QD_SSN_ENDPOINT);
addr->edge_outlink = link;
}
}
@@ -233,7 +233,8 @@ static void on_conn_event(void *context, qdrc_event_t event, qdr_connection_t *c
//
qdr_link_t *out_link = qdr_create_link_CT(ap->core, conn,
QD_LINK_ENDPOINT, QD_OUTGOING,
- qdr_terminus(0), qdr_terminus(0));
+ qdr_terminus(0), qdr_terminus(0),
+ QD_SSN_ENDPOINT);
//
// Associate the anonymous sender with the edge connection address. This will cause
@@ -248,7 +249,8 @@ static void on_conn_event(void *context, qdrc_event_t event, qdr_connection_t *c
(void) qdr_create_link_CT(ap->core, conn,
QD_LINK_ENDPOINT, QD_INCOMING,
qdr_terminus_edge_downlink(ap->core->router_id),
- qdr_terminus_edge_downlink(0));
+ qdr_terminus_edge_downlink(0),
+ QD_SSN_ENDPOINT);
//
// Attach a receiving link for edge address tracking updates.
diff --git a/src/router_core/route_control.c b/src/router_core/route_control.c
index 9ec0023..68de4a0 100644
--- a/src/router_core/route_control.c
+++ b/src/router_core/route_control.c
@@ -253,7 +253,8 @@ static void qdr_auto_link_activate_CT(qdr_core_t *core, qdr_auto_link_t *al, qdr
al->internal_addr = &key[2];
} else
qdr_terminus_set_address(term, &key[2]); // truncate the "Mp" annotation (where p = phase)
- al->link = qdr_create_link_CT(core, conn, QD_LINK_ENDPOINT, al->dir, source, target);
+ al->link = qdr_create_link_CT(core, conn, QD_LINK_ENDPOINT, al->dir, source, target,
+ QD_SSN_ENDPOINT);
al->link->auto_link = al;
al->link->phase = al->phase;
al->link->fallback = al->fallback;
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 611ecd8..7939763 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -271,6 +271,7 @@ typedef struct qdr_connection_work_t {
qdr_link_t *link;
qdr_terminus_t *source;
qdr_terminus_t *target;
+ qd_session_class_t ssn_class;
} qdr_connection_work_t;
ALLOC_DECLARE(qdr_connection_work_t);
@@ -964,12 +965,13 @@ void qdr_link_enqueue_work_CT(qdr_core_t *core,
qdr_link_t *conn,
qdr_link_work_t *work);
-qdr_link_t *qdr_create_link_CT(qdr_core_t *core,
- qdr_connection_t *conn,
- qd_link_type_t link_type,
- qd_direction_t dir,
- qdr_terminus_t *source,
- qdr_terminus_t *target);
+qdr_link_t *qdr_create_link_CT(qdr_core_t *core,
+ qdr_connection_t *conn,
+ qd_link_type_t link_type,
+ qd_direction_t dir,
+ qdr_terminus_t *source,
+ qdr_terminus_t *target,
+ qd_session_class_t ssn_class);
void qdr_link_outbound_detach_CT(qdr_core_t *core, qdr_link_t *link, qdr_error_t *error, qdr_condition_t condition, bool close);
void qdr_link_outbound_second_attach_CT(qdr_core_t *core, qdr_link_t *link, qdr_terminus_t *source, qdr_terminus_t *target);
diff --git a/src/router_node.c b/src/router_node.c
index 365a88f..ea35944 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1353,7 +1353,8 @@ static void CORE_link_first_attach(void *context,
qdr_connection_t *conn,
qdr_link_t *link,
qdr_terminus_t *source,
- qdr_terminus_t *target)
+ qdr_terminus_t *target,
+ qd_session_class_t ssn_class)
{
qd_router_t *router = (qd_router_t*) context;
qd_connection_t *qconn = (qd_connection_t*) qdr_connection_get_context(conn);
@@ -1362,7 +1363,7 @@ static void CORE_link_first_attach(void *context,
//
// Create a new link to be attached
//
- qd_link_t *qlink = qd_link(router->node, qconn, qdr_link_direction(link), qdr_link_name(link));
+ qd_link_t *qlink = qd_link(router->node, qconn, qdr_link_direction(link), qdr_link_name(link), ssn_class);
//
// Copy the source and target termini to the link
diff --git a/src/server_private.h b/src/server_private.h
index 167a5ff..4176429 100644
--- a/src/server_private.h
+++ b/src/server_private.h
@@ -147,7 +147,7 @@ struct qd_connection_t {
int enqueued;
qd_timer_t *timer; // Timer for initial-setup
pn_connection_t *pn_conn;
- pn_session_t *pn_sess;
+ pn_session_t *pn_sessions[QD_SSN_CLASS_COUNT];
pn_ssl_t *ssl;
qd_listener_t *listener;
qd_connector_t *connector;
diff --git a/tests/system_tests_router_mesh.py b/tests/system_tests_router_mesh.py
index 99b9829..e8fab01 100644
--- a/tests/system_tests_router_mesh.py
+++ b/tests/system_tests_router_mesh.py
@@ -209,6 +209,35 @@ class ThreeRouterTest(TestCase):
self.assertEqual(1, ats.released)
self.assertEqual(0, ats.rejected)
+ def test_06_parallel_priority(self):
+ """
+ Create 10 senders each with a different priority. Send large messages
+ - large enough to trigger Qx flow control. Ensure all messages arrive
+ as expected
+ """
+
+ priorities = 10
+ send_batch = 150
+
+ total = priorities * send_batch
+ rx = self.spawn_receiver(self.RouterC,
+ count=total,
+ address="closest/test_06_address")
+ self.RouterA.wait_address("closest/test_06_address")
+
+ senders = [self.spawn_sender(self.RouterA,
+ send_batch,
+ "closest/test_06_address",
+ "-sx", "-p%s" % p)
+ for p in range(priorities)]
+
+ if rx.wait(timeout=TIMEOUT):
+ raise Exception("Receiver failed to consume all messages")
+ for tx in senders:
+ out_text, out_err = tx.communicate(timeout=TIMEOUT)
+ if tx.returncode:
+ raise Exception("Sender failed: %s %s" % (out_text, out_err))
+
if __name__ == '__main__':
unittest.main(main_module())
diff --git a/tests/test-sender.c b/tests/test-sender.c
index 0fecea0..1386619 100644
--- a/tests/test-sender.c
+++ b/tests/test-sender.c
@@ -44,8 +44,8 @@
#define BODY_SIZE_SMALL 100
#define BODY_SIZE_MEDIUM 2000
-#define BODY_SIZE_LARGE 60000
-#define BODY_SIZE_HUGE 257 * 1024 // will trigger Q2/Q3
+#define BODY_SIZE_LARGE 65 * 1024
+#define BODY_SIZE_HUGE 1024 * 1024 // will trigger Q2/Q3
#define DEFAULT_PRIORITY 4
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org