You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2016/03/19 00:07:08 UTC
[50/50] [abbrv] qpid-dispatch git commit: DISPATCH-179 - Flush out
the undelivered-list on incoming links when the address goes from zero to one
reachable destinations.
DISPATCH-179 - Flush out the undelivered-list on incoming links when the address goes
from zero to one reachable destinations.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/326d09a8
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/326d09a8
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/326d09a8
Branch: refs/heads/master
Commit: 326d09a89cc2663686326ab32554117ba0ee8aa1
Parents: 2291e5c
Author: Ted Ross <tr...@redhat.com>
Authored: Fri Mar 18 18:25:44 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Mar 18 18:25:44 2016 -0400
----------------------------------------------------------------------
src/router_core/forwarder.c | 31 ++--
src/router_core/router_core_private.h | 4 +-
src/router_core/transfer.c | 251 +++++++++++++++++------------
3 files changed, 156 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/326d09a8/src/router_core/forwarder.c
----------------------------------------------------------------------
diff --git a/src/router_core/forwarder.c b/src/router_core/forwarder.c
index 5f24a4b..02c0bb5 100644
--- a/src/router_core/forwarder.c
+++ b/src/router_core/forwarder.c
@@ -30,8 +30,7 @@ typedef int (*qdr_forward_message_t) (qdr_core_t *core,
qd_message_t *msg,
qdr_delivery_t *in_delivery,
bool exclude_inprocess,
- bool control,
- qd_bitmask_t *link_exclusion);
+ bool control);
typedef bool (*qdr_forward_attach_t) (qdr_core_t *core,
qdr_address_t *addr,
@@ -54,8 +53,7 @@ static int qdr_forward_message_null_CT(qdr_core_t *core,
qd_message_t *msg,
qdr_delivery_t *in_delivery,
bool exclude_inprocess,
- bool control,
- qd_bitmask_t *link_exclusion)
+ bool control)
{
qd_log(core->log, QD_LOG_CRITICAL, "NULL Message Forwarder Invoked");
return 0;
@@ -137,8 +135,7 @@ int qdr_forward_multicast_CT(qdr_core_t *core,
qd_message_t *msg,
qdr_delivery_t *in_delivery,
bool exclude_inprocess,
- bool control,
- qd_bitmask_t *link_exclusion)
+ bool control)
{
bool bypass_valid_origins = addr->forwarder->bypass_valid_origins;
int fanout = 0;
@@ -222,7 +219,7 @@ int qdr_forward_multicast_CT(qdr_core_t *core,
dest_link = control ?
core->control_links_by_mask_bit[link_bit] :
core->data_links_by_mask_bit[link_bit];
- if (dest_link && (!link_exclusion || qd_bitmask_value(link_exclusion, link_bit) == 0)) {
+ if (dest_link && (!in_delivery->link_exclusion || qd_bitmask_value(in_delivery->link_exclusion, link_bit) == 0)) {
qdr_delivery_t *out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, dest_link, msg);
qdr_forward_deliver_CT(core, dest_link, out_delivery);
fanout++;
@@ -246,8 +243,6 @@ int qdr_forward_multicast_CT(qdr_core_t *core,
}
}
- if (link_exclusion)
- qd_bitmask_free(link_exclusion);
return fanout;
}
@@ -257,19 +252,12 @@ int qdr_forward_closest_CT(qdr_core_t *core,
qd_message_t *msg,
qdr_delivery_t *in_delivery,
bool exclude_inprocess,
- bool control,
- qd_bitmask_t *link_exclusion)
+ bool control)
{
qdr_link_t *out_link;
qdr_delivery_t *out_delivery;
//
- // The Anycast forwarders don't respect link exclusions.
- //
- if (link_exclusion)
- qd_bitmask_free(link_exclusion);
-
- //
// Forward to an in-process subscriber if there is one.
//
if (!exclude_inprocess) {
@@ -360,10 +348,9 @@ int qdr_forward_balanced_CT(qdr_core_t *core,
qd_message_t *msg,
qdr_delivery_t *in_delivery,
bool exclude_inprocess,
- bool control,
- qd_bitmask_t *link_exclusion)
+ bool control)
{
- return qdr_forward_closest_CT(core, addr, msg, in_delivery, exclude_inprocess, control, link_exclusion);
+ return qdr_forward_closest_CT(core, addr, msg, in_delivery, exclude_inprocess, control);
}
@@ -485,10 +472,10 @@ qdr_forwarder_t *qdr_forwarder_CT(qdr_core_t *core, qd_address_treatment_t treat
int qdr_forward_message_CT(qdr_core_t *core, qdr_address_t *addr, qd_message_t *msg, qdr_delivery_t *in_delivery,
- bool exclude_inprocess, bool control, qd_bitmask_t *link_exclusion)
+ bool exclude_inprocess, bool control)
{
if (addr->forwarder)
- return addr->forwarder->forward_message(core, addr, msg, in_delivery, exclude_inprocess, control, link_exclusion);
+ return addr->forwarder->forward_message(core, addr, msg, in_delivery, exclude_inprocess, control);
// TODO - Deal with this delivery's disposition
return 0;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/326d09a8/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 74eb8c1..458c93f 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -38,7 +38,7 @@ typedef struct qdr_connection_ref_t qdr_connection_ref_t;
qdr_forwarder_t *qdr_forwarder_CT(qdr_core_t *core, qd_address_treatment_t treatment);
int qdr_forward_message_CT(qdr_core_t *core, qdr_address_t *addr, qd_message_t *msg, qdr_delivery_t *in_delivery,
- bool exclude_inprocess, bool control, qd_bitmask_t *link_exclusion);
+ bool exclude_inprocess, bool control);
bool qdr_forward_attach_CT(qdr_core_t *core, qdr_address_t *addr, qdr_link_t *in_link, qdr_terminus_t *source,
qdr_terminus_t *target);
@@ -93,7 +93,6 @@ struct qdr_action_t {
qdr_terminus_t *target;
qdr_error_t *error;
qd_detach_type_t dt;
- qd_bitmask_t *link_exclusion;
int credit;
bool drain;
} connection;
@@ -193,6 +192,7 @@ struct qdr_delivery_t {
uint64_t disposition;
bool settled;
uint64_t tag;
+ qd_bitmask_t *link_exclusion;
};
ALLOC_DECLARE(qdr_delivery_t);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/326d09a8/src/router_core/transfer.c
----------------------------------------------------------------------
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index 6aa2d04..5ab27cf 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -43,14 +43,14 @@ qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_field_i
qdr_delivery_t *dlv = new_qdr_delivery_t();
ZERO(dlv);
- dlv->link = link;
- dlv->msg = msg;
- dlv->to_addr = 0;
- dlv->origin = ingress;
- dlv->settled = settled;
+ dlv->link = link;
+ dlv->msg = msg;
+ dlv->to_addr = 0;
+ dlv->origin = ingress;
+ dlv->settled = settled;
+ dlv->link_exclusion = link_exclusion;
action->args.connection.delivery = dlv;
- action->args.connection.link_exclusion = link_exclusion;
qdr_action_enqueue(link->core, action);
return dlv;
}
@@ -64,14 +64,14 @@ qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
qdr_delivery_t *dlv = new_qdr_delivery_t();
ZERO(dlv);
- dlv->link = link;
- dlv->msg = msg;
- dlv->to_addr = addr;
- dlv->origin = ingress;
- dlv->settled = settled;
+ dlv->link = link;
+ dlv->msg = msg;
+ dlv->to_addr = addr;
+ dlv->origin = ingress;
+ dlv->settled = settled;
+ dlv->link_exclusion = link_exclusion;
action->args.connection.delivery = dlv;
- action->args.connection.link_exclusion = link_exclusion;
qdr_action_enqueue(link->core, action);
return dlv;
}
@@ -198,6 +198,7 @@ void qdr_delivery_free(qdr_delivery_t *delivery)
qd_message_free(delivery->msg);
if (delivery->to_addr)
qd_field_iterator_free(delivery->to_addr);
+ qd_bitmask_free(delivery->link_exclusion);
free_qdr_delivery_t(delivery);
}
@@ -242,60 +243,6 @@ qd_message_t *qdr_delivery_message(const qdr_delivery_t *delivery)
// In-Thread Functions
//==================================================================================
-/**
- * Check the link's accumulated credit. If the credit given to the connection thread
- * has been issued to Proton, provide the next batch of credit to the connection thread.
- */
-void qdr_link_issue_credit_CT(qdr_core_t *core, qdr_link_t *link, int credit)
-{
- link->incremental_credit_CT += credit;
- link->flow_started = true;
-
- if (link->incremental_credit_CT && link->incremental_credit == 0) {
- //
- // Move the credit from the core-thread value to the connection-thread value.
- //
- link->incremental_credit = link->incremental_credit_CT;
- link->incremental_credit_CT = 0;
-
- //
- // Put this link on the connection's has-credit list.
- //
- sys_mutex_lock(link->conn->work_lock);
- qdr_add_link_ref(&link->conn->links_with_credit, link, QDR_LINK_LIST_CLASS_FLOW);
- sys_mutex_unlock(link->conn->work_lock);
-
- //
- // Activate the connection
- //
- qdr_connection_activate_CT(core, link->conn);
- }
-}
-
-
-/**
- * This function should be called after adding a new destination (subscription, local link,
- * or remote node) to an address. If this address now has exactly one destination (i.e. it
- * transitioned from unreachable to reachable), make sure any unstarted in-links are issued
- * initial credit.
- */
-void qdr_addr_start_inlinks_CT(qdr_core_t *core, qdr_address_t *addr)
-{
- if (DEQ_SIZE(addr->inlinks) == 0)
- return;
-
- if (DEQ_SIZE(addr->subscriptions) + DEQ_SIZE(addr->rlinks) + qd_bitmask_cardinality(addr->rnodes) == 1) {
- qdr_link_ref_t *ref = DEQ_HEAD(addr->inlinks);
- while (ref) {
- qdr_link_t *link = ref->link;
- if (!link->flow_started)
- qdr_link_issue_credit_CT(core, link, link->capacity);
- ref = DEQ_NEXT(ref);
- }
- }
-}
-
-
static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
if (discard)
@@ -335,49 +282,19 @@ static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discar
}
-static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
+static int qdr_link_forward_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *dlv, qdr_address_t *addr)
{
- if (discard)
- return;
-
- qdr_delivery_t *dlv = action->args.connection.delivery;
- qd_bitmask_t *link_exclude = action->args.connection.link_exclusion;
- qdr_link_t *link = dlv->link;
- int fanout = 0;
- bool presettled = dlv->settled;
-
- //
- // If this is an attach-routed link, put the delivery directly onto the peer link
- //
- if (link->connected_link) {
- qdr_delivery_t *peer = qdr_forward_new_delivery_CT(core, dlv, link->connected_link, dlv->msg);
- qdr_forward_deliver_CT(core, link->connected_link, peer);
- qd_message_free(dlv->msg);
- dlv->msg = 0;
- return;
- }
-
- //
- // NOTE: The link->undelivered list does not need to be protected by the
- // connection's work lock for incoming links. This protection is only
- // needed for outgoing links.
- //
-
- if (DEQ_IS_EMPTY(link->undelivered)) {
- qdr_address_t *addr = link->owning_addr;
- if (!addr && dlv->to_addr)
- qd_hash_retrieve(core->addr_hash, dlv->to_addr, (void**) &addr);
- if (addr) {
- fanout = qdr_forward_message_CT(core, addr, dlv->msg, dlv, false,
- link->link_type == QD_LINK_CONTROL, link_exclude);
- if (link->link_type != QD_LINK_CONTROL && link->link_type != QD_LINK_ROUTER)
- addr->deliveries_ingress++;
- link->total_deliveries++;
- }
+ int fanout = 0;
+ bool presettled = dlv->settled;
+
+ if (addr) {
+ fanout = qdr_forward_message_CT(core, addr, dlv->msg, dlv, false, link->link_type == QD_LINK_CONTROL);
+ if (link->link_type != QD_LINK_CONTROL && link->link_type != QD_LINK_ROUTER)
+ addr->deliveries_ingress++;
+ link->total_deliveries++;
}
if (fanout == 0) {
- printf("TODO fanout == 0\n");
if (link->owning_addr) {
//
// Message was not delivered and the link is not anonymous.
@@ -391,6 +308,7 @@ static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool dis
//
}
} else if (fanout == 1) {
+ qd_bitmask_free(dlv->link_exclusion);
if (dlv->settled) {
//
// The delivery is settled. Keep it off the unsettled list and issue
@@ -411,6 +329,7 @@ static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool dis
//
// The fanout is greater than one. Do something! TODO
//
+ qd_bitmask_free(dlv->link_exclusion);
if (presettled) {
qdr_link_issue_credit_CT(core, link, 1);
@@ -418,6 +337,43 @@ static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool dis
qdr_delivery_free(dlv);
}
}
+
+ return fanout;
+}
+
+
+static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
+{
+ if (discard)
+ return;
+
+ qdr_delivery_t *dlv = action->args.connection.delivery;
+ qdr_link_t *link = dlv->link;
+
+ //
+ // If this is an attach-routed link, put the delivery directly onto the peer link
+ //
+ if (link->connected_link) {
+ qdr_delivery_t *peer = qdr_forward_new_delivery_CT(core, dlv, link->connected_link, dlv->msg);
+ qdr_forward_deliver_CT(core, link->connected_link, peer);
+ qd_message_free(dlv->msg);
+ dlv->msg = 0;
+ return;
+ }
+
+ //
+ // NOTE: The link->undelivered list does not need to be protected by the
+ // connection's work lock for incoming links. This protection is only
+ // needed for outgoing links.
+ //
+
+ if (DEQ_IS_EMPTY(link->undelivered)) {
+ qdr_address_t *addr = link->owning_addr;
+ if (!addr && dlv->to_addr)
+ qd_hash_retrieve(core->addr_hash, dlv->to_addr, (void**) &addr);
+ qdr_link_forward_CT(core, link, dlv, addr);
+ } else
+ DEQ_INSERT_TAIL(link->undelivered, dlv);
}
@@ -436,7 +392,7 @@ static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
// Forward the message. We don't care what the fanout count is.
//
(void) qdr_forward_message_CT(core, addr, msg, 0, action->args.io.exclude_inprocess,
- action->args.io.control, 0);
+ action->args.io.control);
addr->deliveries_from_container++;
} else
qd_log(core->log, QD_LOG_DEBUG, "In-process send to an unknown address");
@@ -508,6 +464,89 @@ static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool
}
+/**
+ * Check the link's accumulated credit. If the credit given to the connection thread
+ * has been issued to Proton, provide the next batch of credit to the connection thread.
+ */
+void qdr_link_issue_credit_CT(qdr_core_t *core, qdr_link_t *link, int credit)
+{
+ link->incremental_credit_CT += credit;
+ link->flow_started = true;
+
+ if (link->incremental_credit_CT && link->incremental_credit == 0) {
+ //
+ // Move the credit from the core-thread value to the connection-thread value.
+ //
+ link->incremental_credit = link->incremental_credit_CT;
+ link->incremental_credit_CT = 0;
+
+ //
+ // Put this link on the connection's has-credit list.
+ //
+ sys_mutex_lock(link->conn->work_lock);
+ qdr_add_link_ref(&link->conn->links_with_credit, link, QDR_LINK_LIST_CLASS_FLOW);
+ sys_mutex_unlock(link->conn->work_lock);
+
+ //
+ // Activate the connection
+ //
+ qdr_connection_activate_CT(core, link->conn);
+ }
+}
+
+
+/**
+ * This function should be called after adding a new destination (subscription, local link,
+ * or remote node) to an address. If this address now has exactly one destination (i.e. it
+ * transitioned from unreachable to reachable), make sure any unstarted in-links are issued
+ * initial credit.
+ *
+ * Also, check the inlinks to see if there are undelivered messages. If so, drain them to
+ * the forwarder.
+ */
+void qdr_addr_start_inlinks_CT(qdr_core_t *core, qdr_address_t *addr)
+{
+ if (DEQ_SIZE(addr->inlinks) == 0)
+ return;
+
+ if (DEQ_SIZE(addr->subscriptions) + DEQ_SIZE(addr->rlinks) + qd_bitmask_cardinality(addr->rnodes) == 1) {
+ qdr_link_ref_t *ref = DEQ_HEAD(addr->inlinks);
+ while (ref) {
+ qdr_link_t *link = ref->link;
+
+ //
+ // Issue credit to stalled links
+ //
+ if (!link->flow_started)
+ qdr_link_issue_credit_CT(core, link, link->capacity);
+
+ //
+ // Drain undelivered deliveries via the forwarder
+ //
+ if (DEQ_SIZE(link->undelivered) > 0) {
+
+ //
+ // Move all the undelivered to a local list in case not all can be delivered.
+ // We don't want to loop here forever putting the same messages on the undelivered
+ // list.
+ //
+ qdr_delivery_list_t deliveries;
+ DEQ_MOVE(link->undelivered, deliveries);
+
+ qdr_delivery_t *dlv = DEQ_HEAD(deliveries);
+ while (dlv) {
+ DEQ_REMOVE_HEAD(deliveries);
+ qdr_link_forward_CT(core, link, dlv, addr);
+ dlv = DEQ_HEAD(deliveries);
+ }
+ }
+
+ ref = DEQ_NEXT(ref);
+ }
+ }
+}
+
+
void qdr_delivery_push_CT(qdr_core_t *core, qdr_delivery_t *dlv)
{
if (!dlv || !dlv->link)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org