tipc: introduce replicast as transport option for multicast

TIPC multicast messages are currently carried over a reliable
'broadcast link', making use of the underlying media's ability to
transport packets as L2 broadcast or IP multicast to all nodes in
the cluster.

When the used bearer is lacking that ability, we can instead emulate
the broadcast service by replicating and sending the packets over as
many unicast links as needed to reach all identified destinations.
We now introduce a new TIPC link-level 'replicast' service that does
this.

Reviewed-by: Parthasarathy Bhuvaragan <parthasarathy.bhuvaragan@ericsson.com>
Acked-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

1 files changed, 18 insertions, 9 deletions

diff --git a/net/tipc/node.c b/net/tipc/node.c
index 2883f6a0ed98..f96dacf173ab 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -1257,6 +1257,19 @@ void tipc_node_broadcast(struct net *net, struct sk_buff *skb)
         kfree_skb(skb);
 }
 
+static void tipc_node_mcast_rcv(struct tipc_node *n)
+{
+        struct tipc_bclink_entry *be = &n->bc_entry;
+
+        /* 'arrvq' is under inputq2's lock protection */
+        spin_lock_bh(&be->inputq2.lock);
+        spin_lock_bh(&be->inputq1.lock);
+        skb_queue_splice_tail_init(&be->inputq1, &be->arrvq);
+        spin_unlock_bh(&be->inputq1.lock);
+        spin_unlock_bh(&be->inputq2.lock);
+        tipc_sk_mcast_rcv(n->net, &be->arrvq, &be->inputq2);
+}
+
 static void tipc_node_bc_sync_rcv(struct tipc_node *n, struct tipc_msg *hdr,
                                   int bearer_id, struct sk_buff_head *xmitq)
 {
@@ -1330,15 +1343,8 @@ static void tipc_node_bc_rcv(struct net *net, struct sk_buff *skb, int bearer_id
         if (!skb_queue_empty(&xmitq))
                 tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr);
 
-        /* Deliver. 'arrvq' is under inputq2's lock protection */
-        if (!skb_queue_empty(&be->inputq1)) {
-                spin_lock_bh(&be->inputq2.lock);
-                spin_lock_bh(&be->inputq1.lock);
-                skb_queue_splice_tail_init(&be->inputq1, &be->arrvq);
-                spin_unlock_bh(&be->inputq1.lock);
-                spin_unlock_bh(&be->inputq2.lock);
-                tipc_sk_mcast_rcv(net, &be->arrvq, &be->inputq2);
-        }
+        if (!skb_queue_empty(&be->inputq1))
+                tipc_node_mcast_rcv(n);
 
         if (rc & TIPC_LINK_DOWN_EVT) {
                 /* Reception reassembly failure => reset all links to peer */
@@ -1565,6 +1571,9 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b)
         if (unlikely(!skb_queue_empty(&n->bc_entry.namedq)))
                 tipc_named_rcv(net, &n->bc_entry.namedq);
 
+        if (unlikely(!skb_queue_empty(&n->bc_entry.inputq1)))
+                tipc_node_mcast_rcv(n);
+
         if (!skb_queue_empty(&le->inputq))
                 tipc_sk_rcv(net, &le->inputq);