| /* |
| * 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" |
| |
| /** defines a default set of forwarding behaviors based on the semantics of an |
| * address. |
| */ |
| |
| |
| static void forward_to_direct_subscribers_LH(qd_address_t *addr, |
| qd_router_delivery_t *delivery, |
| qd_message_t *msg, |
| int *fanout) |
| { |
| qd_router_link_ref_t *dest_link_ref = DEQ_HEAD(addr->rlinks); |
| while (dest_link_ref) { |
| qd_routed_event_t *re = new_qd_routed_event_t(); |
| DEQ_ITEM_INIT(re); |
| re->delivery = 0; |
| re->message = qd_message_copy(msg); |
| re->settle = 0; |
| re->disposition = 0; |
| DEQ_INSERT_TAIL(dest_link_ref->link->msg_fifo, re); |
| |
| (*fanout)++; |
| if (*fanout == 1) { |
| re->delivery = delivery; |
| qd_router_delivery_fifo_enter_LH(delivery); |
| } |
| |
| addr->deliveries_egress++; |
| qd_link_activate(dest_link_ref->link->link); |
| |
| // |
| // If the fanout is single, exit the loop here. We only want to send one message copy. |
| // |
| if (QD_FANOUT(addr->semantics) == QD_FANOUT_SINGLE) |
| break; |
| |
| dest_link_ref = DEQ_NEXT(dest_link_ref); |
| } |
| |
| // |
| // If dest_link_ref is not null here, we exited after sending one message copy. |
| // If the number of local links is greater than one, rotate the head link to the |
| // tail so we balance the message deliveries. |
| // |
| if (dest_link_ref && DEQ_SIZE(addr->rlinks) > 1) { |
| assert(DEQ_HEAD(addr->rlinks) == dest_link_ref); |
| DEQ_REMOVE_HEAD(addr->rlinks); |
| DEQ_INSERT_TAIL(addr->rlinks, dest_link_ref); |
| } |
| } |
| |
| |
| static void forward_to_remote_subscribers_LH(qd_router_t *router, |
| qd_address_t *addr, |
| qd_router_delivery_t *delivery, |
| qd_message_t *msg, |
| int *fanout, |
| qd_field_iterator_t *ingress_iter) |
| { |
| // |
| // Get the mask bit associated with the ingress router for the message. |
| // This will be compared against the "valid_origin" masks for each |
| // candidate destination router. |
| // |
| int origin = -1; |
| if (ingress_iter && !(addr->semantics & QD_BYPASS_VALID_ORIGINS)) { |
| qd_address_iterator_reset_view(ingress_iter, ITER_VIEW_NODE_HASH); |
| qd_address_t *origin_addr; |
| qd_hash_retrieve(router->addr_hash, ingress_iter, (void*) &origin_addr); |
| if (origin_addr && DEQ_SIZE(origin_addr->rnodes) == 1) { |
| qd_router_ref_t *rref = DEQ_HEAD(origin_addr->rnodes); |
| origin = rref->router->mask_bit; |
| } |
| } else |
| origin = 0; |
| |
| // |
| // Forward to the next-hops for remote destinations. |
| // |
| if (origin >= 0) { |
| qd_router_ref_t *dest_node_ref = DEQ_HEAD(addr->rnodes); |
| qd_router_link_t *dest_link; |
| qd_bitmask_t *link_set = qd_bitmask(0); |
| |
| // |
| // Loop over the target nodes for this address. Build a set of outgoing links |
| // for which there are valid targets. We do this to avoid sending more than one |
| // message down a given link. It's possible that there are multiple destinations |
| // for this address that are all reachable over the same link. In this case, we |
| // will send only one copy of the message over the link and allow a downstream |
| // router to fan the message out. |
| // |
| while (dest_node_ref) { |
| if (dest_node_ref->router->next_hop) |
| dest_link = dest_node_ref->router->next_hop->peer_link; |
| else |
| dest_link = dest_node_ref->router->peer_link; |
| if (dest_link && qd_bitmask_value(dest_node_ref->router->valid_origins, origin)) |
| qd_bitmask_set_bit(link_set, dest_link->mask_bit); |
| dest_node_ref = DEQ_NEXT(dest_node_ref); |
| } |
| |
| // |
| // Send a copy of the message outbound on each identified link. |
| // |
| int link_bit; |
| while (qd_bitmask_first_set(link_set, &link_bit)) { |
| qd_bitmask_clear_bit(link_set, link_bit); |
| dest_link = router->out_links_by_mask_bit[link_bit]; |
| if (dest_link) { |
| qd_routed_event_t *re = new_qd_routed_event_t(); |
| DEQ_ITEM_INIT(re); |
| re->delivery = 0; |
| re->message = qd_message_copy(msg); |
| re->settle = 0; |
| re->disposition = 0; |
| DEQ_INSERT_TAIL(dest_link->msg_fifo, re); |
| |
| (*fanout)++; |
| if (*fanout == 1) { |
| re->delivery = delivery; |
| qd_router_delivery_fifo_enter_LH(delivery); |
| } |
| |
| addr->deliveries_transit++; |
| qd_link_activate(dest_link->link); |
| } |
| } |
| |
| qd_bitmask_free(link_set); |
| } |
| } |
| |
| |
| /** Multicast forwarder: |
| */ |
| static bool forwarder_multicast_LH(qd_router_forwarder_t *forwarder, |
| qd_router_t *router, |
| qd_message_t *msg, |
| qd_router_delivery_t *delivery, |
| qd_address_t *addr, |
| qd_field_iterator_t *ingress_iter, |
| bool is_direct) |
| { |
| int fanout = 0; |
| // |
| // Forward to all of the local links receiving this address. |
| // |
| forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout); |
| |
| // |
| // If the address form is direct to this router node, don't relay it on to |
| // any other part of the network. |
| // |
| if (!is_direct) |
| forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter); |
| |
| return fanout != 0; |
| } |
| |
| |
| /** Forward using the 'closest' bias: |
| */ |
| static bool forwarder_anycast_closest_LH(qd_router_forwarder_t *forwarder, |
| qd_router_t *router, |
| qd_message_t *msg, |
| qd_router_delivery_t *delivery, |
| qd_address_t *addr, |
| qd_field_iterator_t *ingress_iter, |
| bool is_direct) |
| { |
| int fanout = 0; |
| // |
| // First, try to find a directly connected consumer for the address. If |
| // there is none, then look for the closest remote consumer. |
| // |
| forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout); |
| if (fanout == 0 && !is_direct) |
| forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter); |
| |
| return fanout != 0; |
| } |
| |
| |
| /** Forwarding using a 'balanced' bias: |
| */ |
| static bool forwarder_anycast_balanced_LH(qd_router_forwarder_t *forwarder, |
| qd_router_t *router, |
| qd_message_t *msg, |
| qd_router_delivery_t *delivery, |
| qd_address_t *addr, |
| qd_field_iterator_t *ingress_iter, |
| bool is_direct) |
| { |
| int fanout = 0; |
| // |
| // Alternate between looking first for a local consumer and looking first |
| // for a remote consumer. |
| // |
| addr->toggle = !addr->toggle; |
| if (addr->toggle) { |
| forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout); |
| if (fanout == 0 && !is_direct) |
| forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter); |
| } else { |
| if (!is_direct) |
| forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter); |
| if (fanout == 0) |
| forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout); |
| } |
| |
| return fanout != 0; |
| } |
| |
| |
| /* release method for default forwarders: |
| */ |
| static void forwarder_release(qd_router_forwarder_t *forwarder) |
| { |
| // no-op - they're static singletons! |
| } |
| |
| |
| /* The default forwarders: |
| */ |
| static qd_router_forwarder_t multicast_forwarder = { |
| forwarder_multicast_LH, /* forward method */ |
| forwarder_release, |
| }; |
| static qd_router_forwarder_t anycast_closest_forwarder = { |
| forwarder_anycast_closest_LH, /* forward method */ |
| forwarder_release, |
| }; |
| static qd_router_forwarder_t anycast_balanced_forwarder = { |
| forwarder_anycast_balanced_LH, /* forward method */ |
| forwarder_release, |
| }; |
| |
| |
| /** Get the proper default forwarder for an address of the given semantics: |
| */ |
| qd_router_forwarder_t *qd_router_get_forwarder(qd_address_semantics_t semantics) |
| { |
| switch (QD_FANOUT(semantics)) { |
| case QD_FANOUT_MULTIPLE: |
| return &multicast_forwarder; |
| case QD_FANOUT_SINGLE: |
| switch (QD_BIAS(semantics)) { |
| case QD_BIAS_CLOSEST: |
| return &anycast_closest_forwarder; |
| case QD_BIAS_SPREAD: |
| return &anycast_balanced_forwarder; |
| } |
| } |
| assert(false); // invalid semantics? need new forwarder? |
| return 0; |
| } |
| |
| |
| |
