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:19 UTC
[qpid-dispatch] 02/02: DISPATCH-1579: add logic to activate all
links on session Q3 clear
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