tipc: smooth change between replicast and broadcast

Currently, a multicast stream may start out using replicast, because
there are few destinations, and then it should ideally switch to
L2/broadcast IGMP/multicast when the number of destinations grows beyond
a certain limit. The opposite should happen when the number decreases
below the limit.

To eliminate the risk of message reordering caused by method change,
a sending socket must stick to a previously selected method until it
enters an idle period of 5 seconds. Means there is a 5 seconds pause
in the traffic from the sender socket.

If the sender never makes such a pause, the method will never change,
and transmission may become very inefficient as the cluster grows.

With this commit, we allow such a switch between replicast and
broadcast without any need for a traffic pause.

Solution is to send a dummy message with only the header, also with
the SYN bit set, via broadcast or replicast. For the data message,
the SYN bit is set and sending via replicast or broadcast (inverse
method with dummy).

Then, at receiving side any messages follow first SYN bit message
(data or dummy message), they will be held in deferred queue until
another pair (dummy or data message) arrived in other link.

v2: reverse christmas tree declaration

Acked-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: Hoang Le <hoang.h.le@dektech.com.au>
Signed-off-by: David S. Miller <davem@davemloft.net>

1 files changed, 5 insertions, 0 deletions

diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index b542f14ed444..922b75ff56d3 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -485,6 +485,7 @@ static int tipc_sk_create(struct net *net, struct socket *sock,
                 tsk_set_unreturnable(tsk, true);
                 if (sock->type == SOCK_DGRAM)
                         tsk_set_unreliable(tsk, true);
+                __skb_queue_head_init(&tsk->mc_method.deferredq);
         }
 
         trace_tipc_sk_create(sk, NULL, TIPC_DUMP_NONE, " ");
@@ -582,6 +583,7 @@ static int tipc_release(struct socket *sock)
         sk->sk_shutdown = SHUTDOWN_MASK;
         tipc_sk_leave(tsk);
         tipc_sk_withdraw(tsk, 0, NULL);
+        __skb_queue_purge(&tsk->mc_method.deferredq);
         sk_stop_timer(sk, &sk->sk_timer);
         tipc_sk_remove(tsk);
 
@@ -2162,6 +2164,9 @@ static void tipc_sk_filter_rcv(struct sock *sk, struct sk_buff *skb,
         if (unlikely(grp))
                 tipc_group_filter_msg(grp, &inputq, xmitq);
 
+        if (msg_type(hdr) == TIPC_MCAST_MSG)
+                tipc_mcast_filter_msg(&tsk->mc_method.deferredq, &inputq);
+
         /* Validate and add to receive buffer if there is space */
         while ((skb = __skb_dequeue(&inputq))) {
                 hdr = buf_msg(skb);