Patched in Tegra support.

1 files changed, 357 insertions, 335 deletions

diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c
index 54f89f90ac3..759b318b5ff 100644
--- a/net/tipc/bcast.c
+++ b/net/tipc/bcast.c
@@ -39,14 +39,13 @@
 #include "link.h"
 #include "port.h"
 #include "bcast.h"
-#include "name_distr.h"
 
 #define MAX_PKT_DEFAULT_MCAST 1500      /* bcast link max packet size (fixed) */
 
 #define BCLINK_WIN_DEFAULT 20           /* bcast link window size (default) */
 
 /**
- * struct tipc_bcbearer_pair - a pair of bearers used by broadcast link
+ * struct bcbearer_pair - a pair of bearers used by broadcast link
  * @primary: pointer to primary bearer
  * @secondary: pointer to secondary bearer
  *
@@ -54,13 +53,13 @@
  * to be paired.
  */
 
-struct tipc_bcbearer_pair {
+struct bcbearer_pair {
         struct tipc_bearer *primary;
         struct tipc_bearer *secondary;
 };
 
 /**
- * struct tipc_bcbearer - bearer used by broadcast link
+ * struct bcbearer - bearer used by broadcast link
  * @bearer: (non-standard) broadcast bearer structure
  * @media: (non-standard) broadcast media structure
  * @bpairs: array of bearer pairs
@@ -73,46 +72,51 @@ struct tipc_bcbearer_pair {
  * large local variables within multicast routines.  Concurrent access is
  * prevented through use of the spinlock "bc_lock".
  */
-struct tipc_bcbearer {
+
+struct bcbearer {
         struct tipc_bearer bearer;
-        struct tipc_media media;
-        struct tipc_bcbearer_pair bpairs[MAX_BEARERS];
-        struct tipc_bcbearer_pair bpairs_temp[TIPC_MAX_LINK_PRI + 1];
+        struct media media;
+        struct bcbearer_pair bpairs[MAX_BEARERS];
+        struct bcbearer_pair bpairs_temp[TIPC_MAX_LINK_PRI + 1];
         struct tipc_node_map remains;
         struct tipc_node_map remains_new;
 };
 
 /**
- * struct tipc_bclink - link used for broadcast messages
+ * struct bclink - link used for broadcast messages
  * @link: (non-standard) broadcast link structure
  * @node: (non-standard) node structure representing b'cast link's peer node
- * @bcast_nodes: map of broadcast-capable nodes
  * @retransmit_to: node that most recently requested a retransmit
  *
  * Handles sequence numbering, fragmentation, bundling, etc.
  */
-struct tipc_bclink {
-        struct tipc_link link;
+
+struct bclink {
+        struct link link;
         struct tipc_node node;
-        struct tipc_node_map bcast_nodes;
         struct tipc_node *retransmit_to;
 };
 
-static struct tipc_bcbearer bcast_bearer;
-static struct tipc_bclink bcast_link;
-
-static struct tipc_bcbearer *bcbearer = &bcast_bearer;
-static struct tipc_bclink *bclink = &bcast_link;
-static struct tipc_link *bcl = &bcast_link.link;
 
+static struct bcbearer *bcbearer;
+static struct bclink *bclink;
+static struct link *bcl;
 static DEFINE_SPINLOCK(bc_lock);
 
+/* broadcast-capable node map */
+struct tipc_node_map tipc_bcast_nmap;
+
 const char tipc_bclink_name[] = "broadcast-link";
 
 static void tipc_nmap_diff(struct tipc_node_map *nm_a,
                            struct tipc_node_map *nm_b,
                            struct tipc_node_map *nm_diff);
 
+static u32 buf_seqno(struct sk_buff *buf)
+{
+        return msg_seqno(buf_msg(buf));
+}
+
 static u32 bcbuf_acks(struct sk_buff *buf)
 {
         return (u32)(unsigned long)TIPC_SKB_CB(buf)->handle;
@@ -128,19 +132,6 @@ static void bcbuf_decr_acks(struct sk_buff *buf)
         bcbuf_set_acks(buf, bcbuf_acks(buf) - 1);
 }
 
-void tipc_bclink_add_node(u32 addr)
-{
-        spin_lock_bh(&bc_lock);
-        tipc_nmap_add(&bclink->bcast_nodes, addr);
-        spin_unlock_bh(&bc_lock);
-}
-
-void tipc_bclink_remove_node(u32 addr)
-{
-        spin_lock_bh(&bc_lock);
-        tipc_nmap_remove(&bclink->bcast_nodes, addr);
-        spin_unlock_bh(&bc_lock);
-}
 
 static void bclink_set_last_sent(void)
 {
@@ -155,10 +146,35 @@ u32 tipc_bclink_get_last_sent(void)
         return bcl->fsm_msg_cnt;
 }
 
-static void bclink_update_last_sent(struct tipc_node *node, u32 seqno)
+/**
+ * bclink_set_gap - set gap according to contents of current deferred pkt queue
+ *
+ * Called with 'node' locked, bc_lock unlocked
+ */
+
+static void bclink_set_gap(struct tipc_node *n_ptr)
 {
-        node->bclink.last_sent = less_eq(node->bclink.last_sent, seqno) ?
-                                                seqno : node->bclink.last_sent;
+        struct sk_buff *buf = n_ptr->bclink.deferred_head;
+
+        n_ptr->bclink.gap_after = n_ptr->bclink.gap_to =
+                mod(n_ptr->bclink.last_in);
+        if (unlikely(buf != NULL))
+                n_ptr->bclink.gap_to = mod(buf_seqno(buf) - 1);
+}
+
+/**
+ * bclink_ack_allowed - test if ACK or NACK message can be sent at this moment
+ *
+ * This mechanism endeavours to prevent all nodes in network from trying
+ * to ACK or NACK at the same time.
+ *
+ * Note: TIPC uses a different trigger to distribute ACKs than it does to
+ *       distribute NACKs, but tries to use the same spacing (divide by 16).
+ */
+
+static int bclink_ack_allowed(u32 n)
+{
+        return (n % TIPC_MIN_LINK_WIN) == tipc_own_tag;
 }
 
 
@@ -167,6 +183,7 @@ static void bclink_update_last_sent(struct tipc_node *node, u32 seqno)
  *
  * Called with bc_lock locked
  */
+
 struct tipc_node *tipc_bclink_retransmit_to(void)
 {
         return bclink->retransmit_to;
@@ -179,6 +196,7 @@ struct tipc_node *tipc_bclink_retransmit_to(void)
  *
  * Called with bc_lock locked
  */
+
 static void bclink_retransmit_pkt(u32 after, u32 to)
 {
         struct sk_buff *buf;
@@ -196,42 +214,21 @@ static void bclink_retransmit_pkt(u32 after, u32 to)
  *
  * Node is locked, bc_lock unlocked.
  */
+
 void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
 {
         struct sk_buff *crs;
         struct sk_buff *next;
         unsigned int released = 0;
 
-        spin_lock_bh(&bc_lock);
-
-        /* Bail out if tx queue is empty (no clean up is required) */
-        crs = bcl->first_out;
-        if (!crs)
-                goto exit;
+        if (less_eq(acked, n_ptr->bclink.acked))
+                return;
 
-        /* Determine which messages need to be acknowledged */
-        if (acked == INVALID_LINK_SEQ) {
-                /*
-                 * Contact with specified node has been lost, so need to
-                 * acknowledge sent messages only (if other nodes still exist)
-                 * or both sent and unsent messages (otherwise)
-                 */
-                if (bclink->bcast_nodes.count)
-                        acked = bcl->fsm_msg_cnt;
-                else
-                        acked = bcl->next_out_no;
-        } else {
-                /*
-                 * Bail out if specified sequence number does not correspond
-                 * to a message that has been sent and not yet acknowledged
-                 */
-                if (less(acked, buf_seqno(crs)) ||
-                    less(bcl->fsm_msg_cnt, acked) ||
-                    less_eq(acked, n_ptr->bclink.acked))
-                        goto exit;
-        }
+        spin_lock_bh(&bc_lock);
 
         /* Skip over packets that node has previously acknowledged */
+
+        crs = bcl->first_out;
         while (crs && less_eq(buf_seqno(crs), n_ptr->bclink.acked))
                 crs = crs->next;
 
@@ -239,19 +236,11 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
 
         while (crs && less_eq(buf_seqno(crs), acked)) {
                 next = crs->next;
-
-                if (crs != bcl->next_out)
-                        bcbuf_decr_acks(crs);
-                else {
-                        bcbuf_set_acks(crs, 0);
-                        bcl->next_out = next;
-                        bclink_set_last_sent();
-                }
-
+                bcbuf_decr_acks(crs);
                 if (bcbuf_acks(crs) == 0) {
                         bcl->first_out = next;
                         bcl->out_queue_size--;
-                        kfree_skb(crs);
+                        buf_discard(crs);
                         released = 1;
                 }
                 crs = next;
@@ -266,144 +255,166 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
         }
         if (unlikely(released && !list_empty(&bcl->waiting_ports)))
                 tipc_link_wakeup_ports(bcl, 0);
-exit:
         spin_unlock_bh(&bc_lock);
 }
 
 /**
- * tipc_bclink_update_link_state - update broadcast link state
+ * bclink_send_ack - unicast an ACK msg
  *
  * tipc_net_lock and node lock set
  */
-void tipc_bclink_update_link_state(struct tipc_node *n_ptr, u32 last_sent)
-{
-        struct sk_buff *buf;
-
-        /* Ignore "stale" link state info */
-
-        if (less_eq(last_sent, n_ptr->bclink.last_in))
-                return;
-
-        /* Update link synchronization state; quit if in sync */
 
-        bclink_update_last_sent(n_ptr, last_sent);
-
-        if (n_ptr->bclink.last_sent == n_ptr->bclink.last_in)
-                return;
+static void bclink_send_ack(struct tipc_node *n_ptr)
+{
+        struct link *l_ptr = n_ptr->active_links[n_ptr->addr & 1];
 
-        /* Update out-of-sync state; quit if loss is still unconfirmed */
+        if (l_ptr != NULL)
+                tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
+}
 
-        if ((++n_ptr->bclink.oos_state) == 1) {
-                if (n_ptr->bclink.deferred_size < (TIPC_MIN_LINK_WIN / 2))
-                        return;
-                n_ptr->bclink.oos_state++;
-        }
+/**
+ * bclink_send_nack- broadcast a NACK msg
+ *
+ * tipc_net_lock and node lock set
+ */
 
-        /* Don't NACK if one has been recently sent (or seen) */
+static void bclink_send_nack(struct tipc_node *n_ptr)
+{
+        struct sk_buff *buf;
+        struct tipc_msg *msg;
 
-        if (n_ptr->bclink.oos_state & 0x1)
+        if (!less(n_ptr->bclink.gap_after, n_ptr->bclink.gap_to))
                 return;
 
-        /* Send NACK */
-
         buf = tipc_buf_acquire(INT_H_SIZE);
         if (buf) {
-                struct tipc_msg *msg = buf_msg(buf);
-
+                msg = buf_msg(buf);
                 tipc_msg_init(msg, BCAST_PROTOCOL, STATE_MSG,
-                              INT_H_SIZE, n_ptr->addr);
+                         INT_H_SIZE, n_ptr->addr);
                 msg_set_non_seq(msg, 1);
                 msg_set_mc_netid(msg, tipc_net_id);
-                msg_set_bcast_ack(msg, n_ptr->bclink.last_in);
-                msg_set_bcgap_after(msg, n_ptr->bclink.last_in);
-                msg_set_bcgap_to(msg, n_ptr->bclink.deferred_head
-                                 ? buf_seqno(n_ptr->bclink.deferred_head) - 1
-                                 : n_ptr->bclink.last_sent);
-
-                spin_lock_bh(&bc_lock);
-                tipc_bearer_send(&bcbearer->bearer, buf, NULL);
-                bcl->stats.sent_nacks++;
-                spin_unlock_bh(&bc_lock);
-                kfree_skb(buf);
-
-                n_ptr->bclink.oos_state++;
+                msg_set_bcast_ack(msg, mod(n_ptr->bclink.last_in));
+                msg_set_bcgap_after(msg, n_ptr->bclink.gap_after);
+                msg_set_bcgap_to(msg, n_ptr->bclink.gap_to);
+                msg_set_bcast_tag(msg, tipc_own_tag);
+
+                if (tipc_bearer_send(&bcbearer->bearer, buf, NULL)) {
+                        bcl->stats.sent_nacks++;
+                        buf_discard(buf);
+                } else {
+                        tipc_bearer_schedule(bcl->b_ptr, bcl);
+                        bcl->proto_msg_queue = buf;
+                        bcl->stats.bearer_congs++;
+                }
+
+                /*
+                 * Ensure we doesn't send another NACK msg to the node
+                 * until 16 more deferred messages arrive from it
+                 * (i.e. helps prevent all nodes from NACK'ing at same time)
+                 */
+
+                n_ptr->bclink.nack_sync = tipc_own_tag;
         }
 }
 
 /**
- * bclink_peek_nack - monitor retransmission requests sent by other nodes
+ * tipc_bclink_check_gap - send a NACK if a sequence gap exists
  *
- * Delay any upcoming NACK by this node if another node has already
- * requested the first message this node is going to ask for.
+ * tipc_net_lock and node lock set
+ */
+
+void tipc_bclink_check_gap(struct tipc_node *n_ptr, u32 last_sent)
+{
+        if (!n_ptr->bclink.supported ||
+            less_eq(last_sent, mod(n_ptr->bclink.last_in)))
+                return;
+
+        bclink_set_gap(n_ptr);
+        if (n_ptr->bclink.gap_after == n_ptr->bclink.gap_to)
+                n_ptr->bclink.gap_to = last_sent;
+        bclink_send_nack(n_ptr);
+}
+
+/**
+ * tipc_bclink_peek_nack - process a NACK msg meant for another node
  *
  * Only tipc_net_lock set.
  */
-static void bclink_peek_nack(struct tipc_msg *msg)
+
+static void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to)
 {
-        struct tipc_node *n_ptr = tipc_node_find(msg_destnode(msg));
+        struct tipc_node *n_ptr = tipc_node_find(dest);
+        u32 my_after, my_to;
 
-        if (unlikely(!n_ptr))
+        if (unlikely(!n_ptr || !tipc_node_is_up(n_ptr)))
                 return;
-
         tipc_node_lock(n_ptr);
+        /*
+         * Modify gap to suppress unnecessary NACKs from this node
+         */
+        my_after = n_ptr->bclink.gap_after;
+        my_to = n_ptr->bclink.gap_to;
+
+        if (less_eq(gap_after, my_after)) {
+                if (less(my_after, gap_to) && less(gap_to, my_to))
+                        n_ptr->bclink.gap_after = gap_to;
+                else if (less_eq(my_to, gap_to))
+                        n_ptr->bclink.gap_to = n_ptr->bclink.gap_after;
+        } else if (less_eq(gap_after, my_to)) {
+                if (less_eq(my_to, gap_to))
+                        n_ptr->bclink.gap_to = gap_after;
+        } else {
+                /*
+                 * Expand gap if missing bufs not in deferred queue:
+                 */
+                struct sk_buff *buf = n_ptr->bclink.deferred_head;
+                u32 prev = n_ptr->bclink.gap_to;
 
-        if (n_ptr->bclink.recv_permitted &&
-            (n_ptr->bclink.last_in != n_ptr->bclink.last_sent) &&
-            (n_ptr->bclink.last_in == msg_bcgap_after(msg)))
-                n_ptr->bclink.oos_state = 2;
+                for (; buf; buf = buf->next) {
+                        u32 seqno = buf_seqno(buf);
 
+                        if (mod(seqno - prev) != 1) {
+                                buf = NULL;
+                                break;
+                        }
+                        if (seqno == gap_after)
+                                break;
+                        prev = seqno;
+                }
+                if (buf == NULL)
+                        n_ptr->bclink.gap_to = gap_after;
+        }
+        /*
+         * Some nodes may send a complementary NACK now:
+         */
+        if (bclink_ack_allowed(sender_tag + 1)) {
+                if (n_ptr->bclink.gap_to != n_ptr->bclink.gap_after) {
+                        bclink_send_nack(n_ptr);
+                        bclink_set_gap(n_ptr);
+                }
+        }
         tipc_node_unlock(n_ptr);
 }
 
-/*
+/**
  * tipc_bclink_send_msg - broadcast a packet to all nodes in cluster
  */
+
 int tipc_bclink_send_msg(struct sk_buff *buf)
 {
         int res;
 
         spin_lock_bh(&bc_lock);
 
-        if (!bclink->bcast_nodes.count) {
-                res = msg_data_sz(buf_msg(buf));
-                kfree_skb(buf);
-                goto exit;
-        }
-
         res = tipc_link_send_buf(bcl, buf);
-        if (likely(res >= 0)) {
+        if (likely(res > 0))
                 bclink_set_last_sent();
-                bcl->stats.queue_sz_counts++;
-                bcl->stats.accu_queue_sz += bcl->out_queue_size;
-        }
-exit:
-        spin_unlock_bh(&bc_lock);
-        return res;
-}
-
-/**
- * bclink_accept_pkt - accept an incoming, in-sequence broadcast packet
- *
- * Called with both sending node's lock and bc_lock taken.
- */
-static void bclink_accept_pkt(struct tipc_node *node, u32 seqno)
-{
-        bclink_update_last_sent(node, seqno);
-        node->bclink.last_in = seqno;
-        node->bclink.oos_state = 0;
-        bcl->stats.recv_info++;
 
-        /*
-         * Unicast an ACK periodically, ensuring that
-         * all nodes in the cluster don't ACK at the same time
-         */
+        bcl->stats.queue_sz_counts++;
+        bcl->stats.accu_queue_sz += bcl->out_queue_size;
 
-        if (((seqno - tipc_own_addr) % TIPC_MIN_LINK_WIN) == 0) {
-                tipc_link_send_proto_msg(
-                        node->active_links[node->addr & 1],
-                        STATE_MSG, 0, 0, 0, 0, 0);
-                bcl->stats.sent_acks++;
-        }
+        spin_unlock_bh(&bc_lock);
+        return res;
 }
 
 /**
@@ -411,33 +422,24 @@ static void bclink_accept_pkt(struct tipc_node *node, u32 seqno)
  *
  * tipc_net_lock is read_locked, no other locks set
  */
+
 void tipc_bclink_recv_pkt(struct sk_buff *buf)
 {
         struct tipc_msg *msg = buf_msg(buf);
-        struct tipc_node *node;
+        struct tipc_node *node = tipc_node_find(msg_prevnode(msg));
         u32 next_in;
         u32 seqno;
-        int deferred;
-
-        /* Screen out unwanted broadcast messages */
+        struct sk_buff *deferred;
 
-        if (msg_mc_netid(msg) != tipc_net_id)
-                goto exit;
-
-        node = tipc_node_find(msg_prevnode(msg));
-        if (unlikely(!node))
-                goto exit;
-
-        tipc_node_lock(node);
-        if (unlikely(!node->bclink.recv_permitted))
-                goto unlock;
-
-        /* Handle broadcast protocol message */
+        if (unlikely(!node || !tipc_node_is_up(node) || !node->bclink.supported ||
+                     (msg_mc_netid(msg) != tipc_net_id))) {
+                buf_discard(buf);
+                return;
+        }
 
         if (unlikely(msg_user(msg) == BCAST_PROTOCOL)) {
-                if (msg_type(msg) != STATE_MSG)
-                        goto unlock;
                 if (msg_destnode(msg) == tipc_own_addr) {
+                        tipc_node_lock(node);
                         tipc_bclink_acknowledge(node, msg_bcast_ack(msg));
                         tipc_node_unlock(node);
                         spin_lock_bh(&bc_lock);
@@ -447,124 +449,85 @@ void tipc_bclink_recv_pkt(struct sk_buff *buf)
                                               msg_bcgap_to(msg));
                         spin_unlock_bh(&bc_lock);
                 } else {
-                        tipc_node_unlock(node);
-                        bclink_peek_nack(msg);
+                        tipc_bclink_peek_nack(msg_destnode(msg),
+                                              msg_bcast_tag(msg),
+                                              msg_bcgap_after(msg),
+                                              msg_bcgap_to(msg));
                 }
-                goto exit;
+                buf_discard(buf);
+                return;
         }
 
-        /* Handle in-sequence broadcast message */
-
-        seqno = msg_seqno(msg);
+        tipc_node_lock(node);
+receive:
+        deferred = node->bclink.deferred_head;
         next_in = mod(node->bclink.last_in + 1);
+        seqno = msg_seqno(msg);
 
         if (likely(seqno == next_in)) {
-receive:
-                /* Deliver message to destination */
-
+                bcl->stats.recv_info++;
+                node->bclink.last_in++;
+                bclink_set_gap(node);
+                if (unlikely(bclink_ack_allowed(seqno))) {
+                        bclink_send_ack(node);
+                        bcl->stats.sent_acks++;
+                }
                 if (likely(msg_isdata(msg))) {
-                        spin_lock_bh(&bc_lock);
-                        bclink_accept_pkt(node, seqno);
-                        spin_unlock_bh(&bc_lock);
                         tipc_node_unlock(node);
-                        if (likely(msg_mcast(msg)))
-                                tipc_port_recv_mcast(buf, NULL);
-                        else
-                                kfree_skb(buf);
+                        tipc_port_recv_mcast(buf, NULL);
                 } else if (msg_user(msg) == MSG_BUNDLER) {
-                        spin_lock_bh(&bc_lock);
-                        bclink_accept_pkt(node, seqno);
                         bcl->stats.recv_bundles++;
                         bcl->stats.recv_bundled += msg_msgcnt(msg);
-                        spin_unlock_bh(&bc_lock);
                         tipc_node_unlock(node);
                         tipc_link_recv_bundle(buf);
                 } else if (msg_user(msg) == MSG_FRAGMENTER) {
-                        int ret = tipc_link_recv_fragment(&node->bclink.defragm,
-                                                      &buf, &msg);
-                        if (ret < 0)
-                                goto unlock;
-                        spin_lock_bh(&bc_lock);
-                        bclink_accept_pkt(node, seqno);
                         bcl->stats.recv_fragments++;
-                        if (ret > 0)
+                        if (tipc_link_recv_fragment(&node->bclink.defragm,
+                                                    &buf, &msg))
                                 bcl->stats.recv_fragmented++;
-                        spin_unlock_bh(&bc_lock);
                         tipc_node_unlock(node);
                         tipc_net_route_msg(buf);
-                } else if (msg_user(msg) == NAME_DISTRIBUTOR) {
-                        spin_lock_bh(&bc_lock);
-                        bclink_accept_pkt(node, seqno);
-                        spin_unlock_bh(&bc_lock);
-                        tipc_node_unlock(node);
-                        tipc_named_recv(buf);
                 } else {
-                        spin_lock_bh(&bc_lock);
-                        bclink_accept_pkt(node, seqno);
-                        spin_unlock_bh(&bc_lock);
                         tipc_node_unlock(node);
-                        kfree_skb(buf);
+                        tipc_net_route_msg(buf);
                 }
-                buf = NULL;
-
-                /* Determine new synchronization state */
-
-                tipc_node_lock(node);
-                if (unlikely(!tipc_node_is_up(node)))
-                        goto unlock;
-
-                if (node->bclink.last_in == node->bclink.last_sent)
-                        goto unlock;
-
-                if (!node->bclink.deferred_head) {
-                        node->bclink.oos_state = 1;
-                        goto unlock;
+                if (deferred && (buf_seqno(deferred) == mod(next_in + 1))) {
+                        tipc_node_lock(node);
+                        buf = deferred;
+                        msg = buf_msg(buf);
+                        node->bclink.deferred_head = deferred->next;
+                        goto receive;
                 }
-
-                msg = buf_msg(node->bclink.deferred_head);
-                seqno = msg_seqno(msg);
-                next_in = mod(next_in + 1);
-                if (seqno != next_in)
-                        goto unlock;
-
-                /* Take in-sequence message from deferred queue & deliver it */
-
-                buf = node->bclink.deferred_head;
-                node->bclink.deferred_head = buf->next;
-                node->bclink.deferred_size--;
-                goto receive;
-        }
-
-        /* Handle out-of-sequence broadcast message */
-
-        if (less(next_in, seqno)) {
-                deferred = tipc_link_defer_pkt(&node->bclink.deferred_head,
-                                               &node->bclink.deferred_tail,
-                                               buf);
-                node->bclink.deferred_size += deferred;
-                bclink_update_last_sent(node, seqno);
-                buf = NULL;
-        } else
-                deferred = 0;
-
-        spin_lock_bh(&bc_lock);
-
-        if (deferred)
-                bcl->stats.deferred_recv++;
-        else
+                return;
+        } else if (less(next_in, seqno)) {
+                u32 gap_after = node->bclink.gap_after;
+                u32 gap_to = node->bclink.gap_to;
+
+                if (tipc_link_defer_pkt(&node->bclink.deferred_head,
+                                        &node->bclink.deferred_tail,
+                                        buf)) {
+                        node->bclink.nack_sync++;
+                        bcl->stats.deferred_recv++;
+                        if (seqno == mod(gap_after + 1))
+                                node->bclink.gap_after = seqno;
+                        else if (less(gap_after, seqno) && less(seqno, gap_to))
+                                node->bclink.gap_to = seqno;
+                }
+                if (bclink_ack_allowed(node->bclink.nack_sync)) {
+                        if (gap_to != gap_after)
+                                bclink_send_nack(node);
+                        bclink_set_gap(node);
+                }
+        } else {
                 bcl->stats.duplicates++;
-
-        spin_unlock_bh(&bc_lock);
-
-unlock:
+                buf_discard(buf);
+        }
         tipc_node_unlock(node);
-exit:
-        kfree_skb(buf);
 }
 
 u32 tipc_bclink_acks_missing(struct tipc_node *n_ptr)
 {
-        return (n_ptr->bclink.recv_permitted &&
+        return (n_ptr->bclink.supported &&
                 (tipc_bclink_get_last_sent() != n_ptr->bclink.acked));
 }
 
@@ -572,41 +535,38 @@ u32 tipc_bclink_acks_missing(struct tipc_node *n_ptr)
 /**
  * tipc_bcbearer_send - send a packet through the broadcast pseudo-bearer
  *
- * Send packet over as many bearers as necessary to reach all nodes
- * that have joined the broadcast link.
+ * Send through as many bearers as necessary to reach all nodes
+ * that support TIPC multicasting.
  *
- * Returns 0 (packet sent successfully) under all circumstances,
- * since the broadcast link's pseudo-bearer never blocks
+ * Returns 0 if packet sent successfully, non-zero if not
  */
+
 static int tipc_bcbearer_send(struct sk_buff *buf,
                               struct tipc_bearer *unused1,
                               struct tipc_media_addr *unused2)
 {
         int bp_index;
 
-        /*
-         * Prepare broadcast link message for reliable transmission,
-         * if first time trying to send it;
-         * preparation is skipped for broadcast link protocol messages
-         * since they are sent in an unreliable manner and don't need it
-         */
+        /* Prepare buffer for broadcasting (if first time trying to send it) */
+
         if (likely(!msg_non_seq(buf_msg(buf)))) {
                 struct tipc_msg *msg;
 
-                bcbuf_set_acks(buf, bclink->bcast_nodes.count);
+                bcbuf_set_acks(buf, tipc_bcast_nmap.count);
                 msg = buf_msg(buf);
                 msg_set_non_seq(msg, 1);
                 msg_set_mc_netid(msg, tipc_net_id);
                 bcl->stats.sent_info++;
 
-                if (WARN_ON(!bclink->bcast_nodes.count)) {
+                if (WARN_ON(!tipc_bcast_nmap.count)) {
                         dump_stack();
                         return 0;
                 }
         }
 
         /* Send buffer over bearers until all targets reached */
-        bcbearer->remains = bclink->bcast_nodes;
+
+        bcbearer->remains = tipc_bcast_nmap;
 
         for (bp_index = 0; bp_index < MAX_BEARERS; bp_index++) {
                 struct tipc_bearer *p = bcbearer->bpairs[bp_index].primary;
@@ -619,14 +579,16 @@ static int tipc_bcbearer_send(struct sk_buff *buf,
                 if (bcbearer->remains_new.count == bcbearer->remains.count)
                         continue;       /* bearer pair doesn't add anything */
 
-                if (!tipc_bearer_blocked(p))
-                        tipc_bearer_send(p, buf, &p->media->bcast_addr);
-                else if (s && !tipc_bearer_blocked(s))
+                if (p->blocked ||
+                    p->media->send_msg(buf, p, &p->media->bcast_addr)) {
                         /* unable to send on primary bearer */
-                        tipc_bearer_send(s, buf, &s->media->bcast_addr);
-                else
-                        /* unable to send on either bearer */
-                        continue;
+                        if (!s || s->blocked ||
+                            s->media->send_msg(buf, s,
+                                               &s->media->bcast_addr)) {
+                                /* unable to send on either bearer */
+                                continue;
+                        }
+                }
 
                 if (s) {
                         bcbearer->bpairs[bp_index].primary = s;
@@ -634,27 +596,35 @@ static int tipc_bcbearer_send(struct sk_buff *buf,
                 }
 
                 if (bcbearer->remains_new.count == 0)
-                        break;  /* all targets reached */
+                        return 0;
 
                 bcbearer->remains = bcbearer->remains_new;
         }
 
-        return 0;
+        /*
+         * Unable to reach all targets (indicate success, since currently
+         * there isn't code in place to properly block & unblock the
+         * pseudo-bearer used by the broadcast link)
+         */
+
+        return TIPC_OK;
 }
 
 /**
  * tipc_bcbearer_sort - create sets of bearer pairs used by broadcast bearer
  */
+
 void tipc_bcbearer_sort(void)
 {
-        struct tipc_bcbearer_pair *bp_temp = bcbearer->bpairs_temp;
-        struct tipc_bcbearer_pair *bp_curr;
+        struct bcbearer_pair *bp_temp = bcbearer->bpairs_temp;
+        struct bcbearer_pair *bp_curr;
         int b_index;
         int pri;
 
         spin_lock_bh(&bc_lock);
 
         /* Group bearers by priority (can assume max of two per priority) */
+
         memset(bp_temp, 0, sizeof(bcbearer->bpairs_temp));
 
         for (b_index = 0; b_index < MAX_BEARERS; b_index++) {
@@ -670,6 +640,7 @@ void tipc_bcbearer_sort(void)
         }
 
         /* Create array of bearer pairs for broadcasting */
+
         bp_curr = bcbearer->bpairs;
         memset(bcbearer->bpairs, 0, sizeof(bcbearer->bpairs));
 
@@ -696,46 +667,72 @@ void tipc_bcbearer_sort(void)
         spin_unlock_bh(&bc_lock);
 }
 
+/**
+ * tipc_bcbearer_push - resolve bearer congestion
+ *
+ * Forces bclink to push out any unsent packets, until all packets are gone
+ * or congestion reoccurs.
+ * No locks set when function called
+ */
+
+void tipc_bcbearer_push(void)
+{
+        struct tipc_bearer *b_ptr;
+
+        spin_lock_bh(&bc_lock);
+        b_ptr = &bcbearer->bearer;
+        if (b_ptr->blocked) {
+                b_ptr->blocked = 0;
+                tipc_bearer_lock_push(b_ptr);
+        }
+        spin_unlock_bh(&bc_lock);
+}
+
 
 int tipc_bclink_stats(char *buf, const u32 buf_size)
 {
-        int ret;
-        struct tipc_stats *s;
+        struct print_buf pb;
 
         if (!bcl)
                 return 0;
 
+        tipc_printbuf_init(&pb, buf, buf_size);
+
         spin_lock_bh(&bc_lock);
 
-        s = &bcl->stats;
-
-        ret = tipc_snprintf(buf, buf_size, "Link <%s>\n"
-                            "  Window:%u packets\n",
-                            bcl->name, bcl->queue_limit[0]);
-        ret += tipc_snprintf(buf + ret, buf_size - ret,
-                             "  RX packets:%u fragments:%u/%u bundles:%u/%u\n",
-                             s->recv_info, s->recv_fragments,
-                             s->recv_fragmented, s->recv_bundles,
-                             s->recv_bundled);
-        ret += tipc_snprintf(buf + ret, buf_size - ret,
-                             "  TX packets:%u fragments:%u/%u bundles:%u/%u\n",
-                             s->sent_info, s->sent_fragments,
-                             s->sent_fragmented, s->sent_bundles,
-                             s->sent_bundled);
-        ret += tipc_snprintf(buf + ret, buf_size - ret,
-                             "  RX naks:%u defs:%u dups:%u\n",
-                             s->recv_nacks, s->deferred_recv, s->duplicates);
-        ret += tipc_snprintf(buf + ret, buf_size - ret,
-                             "  TX naks:%u acks:%u dups:%u\n",
-                             s->sent_nacks, s->sent_acks, s->retransmitted);
-        ret += tipc_snprintf(buf + ret, buf_size - ret,
-                             "  Congestion link:%u  Send queue max:%u avg:%u\n",
-                             s->link_congs, s->max_queue_sz,
-                             s->queue_sz_counts ?
-                             (s->accu_queue_sz / s->queue_sz_counts) : 0);
+        tipc_printf(&pb, "Link <%s>\n"
+                         "  Window:%u packets\n",
+                    bcl->name, bcl->queue_limit[0]);
+        tipc_printf(&pb, "  RX packets:%u fragments:%u/%u bundles:%u/%u\n",
+                    bcl->stats.recv_info,
+                    bcl->stats.recv_fragments,
+                    bcl->stats.recv_fragmented,
+                    bcl->stats.recv_bundles,
+                    bcl->stats.recv_bundled);
+        tipc_printf(&pb, "  TX packets:%u fragments:%u/%u bundles:%u/%u\n",
+                    bcl->stats.sent_info,
+                    bcl->stats.sent_fragments,
+                    bcl->stats.sent_fragmented,
+                    bcl->stats.sent_bundles,
+                    bcl->stats.sent_bundled);
+        tipc_printf(&pb, "  RX naks:%u defs:%u dups:%u\n",
+                    bcl->stats.recv_nacks,
+                    bcl->stats.deferred_recv,
+                    bcl->stats.duplicates);
+        tipc_printf(&pb, "  TX naks:%u acks:%u dups:%u\n",
+                    bcl->stats.sent_nacks,
+                    bcl->stats.sent_acks,
+                    bcl->stats.retransmitted);
+        tipc_printf(&pb, "  Congestion bearer:%u link:%u  Send queue max:%u avg:%u\n",
+                    bcl->stats.bearer_congs,
+                    bcl->stats.link_congs,
+                    bcl->stats.max_queue_sz,
+                    bcl->stats.queue_sz_counts
+                    ? (bcl->stats.accu_queue_sz / bcl->stats.queue_sz_counts)
+                    : 0);
 
         spin_unlock_bh(&bc_lock);
-        return ret;
+        return tipc_printbuf_validate(&pb);
 }
 
 int tipc_bclink_reset_stats(void)
@@ -762,12 +759,25 @@ int tipc_bclink_set_queue_limits(u32 limit)
         return 0;
 }
 
-void tipc_bclink_init(void)
+int tipc_bclink_init(void)
 {
+        bcbearer = kzalloc(sizeof(*bcbearer), GFP_ATOMIC);
+        bclink = kzalloc(sizeof(*bclink), GFP_ATOMIC);
+        if (!bcbearer || !bclink) {
+                warn("Multicast link creation failed, no memory\n");
+                kfree(bcbearer);
+                bcbearer = NULL;
+                kfree(bclink);
+                bclink = NULL;
+                return -ENOMEM;
+        }
+
+        INIT_LIST_HEAD(&bcbearer->bearer.cong_links);
         bcbearer->bearer.media = &bcbearer->media;
         bcbearer->media.send_msg = tipc_bcbearer_send;
-        sprintf(bcbearer->media.name, "tipc-broadcast");
+        sprintf(bcbearer->media.name, "tipc-multicast");
 
+        bcl = &bclink->link;
         INIT_LIST_HEAD(&bcl->waiting_ports);
         bcl->next_out_no = 1;
         spin_lock_init(&bclink->node.lock);
@@ -777,22 +787,29 @@ void tipc_bclink_init(void)
         bcl->b_ptr = &bcbearer->bearer;
         bcl->state = WORKING_WORKING;
         strlcpy(bcl->name, tipc_bclink_name, TIPC_MAX_LINK_NAME);
+
+        return 0;
 }
 
 void tipc_bclink_stop(void)
 {
         spin_lock_bh(&bc_lock);
-        tipc_link_stop(bcl);
+        if (bcbearer) {
+                tipc_link_stop(bcl);
+                bcl = NULL;
+                kfree(bclink);
+                bclink = NULL;
+                kfree(bcbearer);
+                bcbearer = NULL;
+        }
         spin_unlock_bh(&bc_lock);
-
-        memset(bclink, 0, sizeof(*bclink));
-        memset(bcbearer, 0, sizeof(*bcbearer));
 }
 
 
 /**
  * tipc_nmap_add - add a node to a node map
  */
+
 void tipc_nmap_add(struct tipc_node_map *nm_ptr, u32 node)
 {
         int n = tipc_node(node);
@@ -808,6 +825,7 @@ void tipc_nmap_add(struct tipc_node_map *nm_ptr, u32 node)
 /**
  * tipc_nmap_remove - remove a node from a node map
  */
+
 void tipc_nmap_remove(struct tipc_node_map *nm_ptr, u32 node)
 {
         int n = tipc_node(node);
@@ -826,6 +844,7 @@ void tipc_nmap_remove(struct tipc_node_map *nm_ptr, u32 node)
  * @nm_b: input node map B
  * @nm_diff: output node map A-B (i.e. nodes of A that are not in B)
  */
+
 static void tipc_nmap_diff(struct tipc_node_map *nm_a,
                            struct tipc_node_map *nm_b,
                            struct tipc_node_map *nm_diff)
@@ -851,9 +870,10 @@ static void tipc_nmap_diff(struct tipc_node_map *nm_a,
 /**
  * tipc_port_list_add - add a port to a port list, ensuring no duplicates
  */
-void tipc_port_list_add(struct tipc_port_list *pl_ptr, u32 port)
+
+void tipc_port_list_add(struct port_list *pl_ptr, u32 port)
 {
-        struct tipc_port_list *item = pl_ptr;
+        struct port_list *item = pl_ptr;
         int i;
         int item_sz = PLSIZE;
         int cnt = pl_ptr->count;
@@ -872,7 +892,7 @@ void tipc_port_list_add(struct tipc_port_list *pl_ptr, u32 port)
                 if (!item->next) {
                         item->next = kmalloc(sizeof(*item), GFP_ATOMIC);
                         if (!item->next) {
-                                pr_warn("Incomplete multicast delivery, no memory\n");
+                                warn("Incomplete multicast delivery, no memory\n");
                                 return;
                         }
                         item->next->next = NULL;
@@ -884,13 +904,15 @@ void tipc_port_list_add(struct tipc_port_list *pl_ptr, u32 port)
  * tipc_port_list_free - free dynamically created entries in port_list chain
  *
  */
-void tipc_port_list_free(struct tipc_port_list *pl_ptr)
+
+void tipc_port_list_free(struct port_list *pl_ptr)
 {
-        struct tipc_port_list *item;
-        struct tipc_port_list *next;
+        struct port_list *item;
+        struct port_list *next;
 
         for (item = pl_ptr->next; item; item = next) {
                 next = item->next;
                 kfree(item);
         }
 }
+