tipc: improve groupcast scope handling

When a member joins a group, it also indicates a binding scope. This
makes it possible to create both node local groups, invisible to other
nodes, as well as cluster global groups, visible everywhere.

In order to avoid that different members end up having permanently
differing views of group size and memberhip, we must inhibit locally
and globally bound members from joining the same group.

We do this by using the binding scope as an additional separator between
groups. I.e., a member must ignore all membership events from sockets
using a different scope than itself, and all lookups for message
destinations must require an exact match between the message's lookup
scope and the potential target's binding scope.

Apart from making it possible to create local groups using the same
identity on different nodes, a side effect of this is that it now also
becomes possible to create a cluster global group with the same identity
across the same nodes, without interfering with the local groups.

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, 53 insertions, 35 deletions

diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index e3a02f1fcab5..b24dab3996c9 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -928,21 +928,22 @@ static int tipc_send_group_anycast(struct socket *sock, struct msghdr *m,
         struct list_head *cong_links = &tsk->cong_links;
         int blks = tsk_blocks(GROUP_H_SIZE + dlen);
         struct tipc_group *grp = tsk->group;
+        struct tipc_msg *hdr = &tsk->phdr;
         struct tipc_member *first = NULL;
         struct tipc_member *mbr = NULL;
         struct net *net = sock_net(sk);
         u32 node, port, exclude;
-        u32 type, inst, domain;
         struct list_head dsts;
+        u32 type, inst, scope;
         int lookups = 0;
         int dstcnt, rc;
         bool cong;
 
         INIT_LIST_HEAD(&dsts);
 
-        type = dest->addr.name.name.type;
+        type = msg_nametype(hdr);
         inst = dest->addr.name.name.instance;
-        domain = addr_domain(net, dest->scope);
+        scope = msg_lookup_scope(hdr);
         exclude = tipc_group_exclude(grp);
 
         while (++lookups < 4) {
@@ -950,7 +951,7 @@ static int tipc_send_group_anycast(struct socket *sock, struct msghdr *m,
 
                 /* Look for a non-congested destination member, if any */
                 while (1) {
-                        if (!tipc_nametbl_lookup(net, type, inst, domain, &dsts,
+                        if (!tipc_nametbl_lookup(net, type, inst, scope, &dsts,
                                                  &dstcnt, exclude, false))
                                 return -EHOSTUNREACH;
                         tipc_dest_pop(&dsts, &node, &port);
@@ -1079,22 +1080,23 @@ static int tipc_send_group_mcast(struct socket *sock, struct msghdr *m,
 {
         struct sock *sk = sock->sk;
         DECLARE_SOCKADDR(struct sockaddr_tipc *, dest, m->msg_name);
-        struct tipc_name_seq *seq = &dest->addr.nameseq;
         struct tipc_sock *tsk = tipc_sk(sk);
         struct tipc_group *grp = tsk->group;
+        struct tipc_msg *hdr = &tsk->phdr;
         struct net *net = sock_net(sk);
-        u32 domain, exclude, dstcnt;
+        u32 type, inst, scope, exclude;
         struct list_head dsts;
+        u32 dstcnt;
 
         INIT_LIST_HEAD(&dsts);
 
-        if (seq->lower != seq->upper)
-                return -ENOTSUPP;
-
-        domain = addr_domain(net, dest->scope);
+        type = msg_nametype(hdr);
+        inst = dest->addr.name.name.instance;
+        scope = msg_lookup_scope(hdr);
         exclude = tipc_group_exclude(grp);
-        if (!tipc_nametbl_lookup(net, seq->type, seq->lower, domain,
-                                 &dsts, &dstcnt, exclude, true))
+
+        if (!tipc_nametbl_lookup(net, type, inst, scope, &dsts,
+                                 &dstcnt, exclude, true))
                 return -EHOSTUNREACH;
 
         if (dstcnt == 1) {
@@ -1116,24 +1118,29 @@ static int tipc_send_group_mcast(struct socket *sock, struct msghdr *m,
 void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
                        struct sk_buff_head *inputq)
 {
-        u32 scope = TIPC_CLUSTER_SCOPE;
         u32 self = tipc_own_addr(net);
+        u32 type, lower, upper, scope;
         struct sk_buff *skb, *_skb;
-        u32 lower = 0, upper = ~0;
-        struct sk_buff_head tmpq;
         u32 portid, oport, onode;
+        struct sk_buff_head tmpq;
         struct list_head dports;
-        struct tipc_msg *msg;
-        int user, mtyp, hsz;
+        struct tipc_msg *hdr;
+        int user, mtyp, hlen;
+        bool exact;
 
         __skb_queue_head_init(&tmpq);
         INIT_LIST_HEAD(&dports);
 
         skb = tipc_skb_peek(arrvq, &inputq->lock);
         for (; skb; skb = tipc_skb_peek(arrvq, &inputq->lock)) {
-                msg = buf_msg(skb);
-                user = msg_user(msg);
-                mtyp = msg_type(msg);
+                hdr = buf_msg(skb);
+                user = msg_user(hdr);
+                mtyp = msg_type(hdr);
+                hlen = skb_headroom(skb) + msg_hdr_sz(hdr);
+                oport = msg_origport(hdr);
+                onode = msg_orignode(hdr);
+                type = msg_nametype(hdr);
+
                 if (mtyp == TIPC_GRP_UCAST_MSG || user == GROUP_PROTOCOL) {
                         spin_lock_bh(&inputq->lock);
                         if (skb_peek(arrvq) == skb) {
@@ -1144,21 +1151,31 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
                         spin_unlock_bh(&inputq->lock);
                         continue;
                 }
-                hsz = skb_headroom(skb) + msg_hdr_sz(msg);
-                oport = msg_origport(msg);
-                onode = msg_orignode(msg);
-                if (onode == self)
-                        scope = TIPC_NODE_SCOPE;
-
-                /* Create destination port list and message clones: */
-                if (!msg_in_group(msg)) {
-                        lower = msg_namelower(msg);
-                        upper = msg_nameupper(msg);
+
+                /* Group messages require exact scope match */
+                if (msg_in_group(hdr)) {
+                        lower = 0;
+                        upper = ~0;
+                        scope = msg_lookup_scope(hdr);
+                        exact = true;
+                } else {
+                        /* TIPC_NODE_SCOPE means "any scope" in this context */
+                        if (onode == self)
+                                scope = TIPC_NODE_SCOPE;
+                        else
+                                scope = TIPC_CLUSTER_SCOPE;
+                        exact = false;
+                        lower = msg_namelower(hdr);
+                        upper = msg_nameupper(hdr);
                 }
-                tipc_nametbl_mc_translate(net, msg_nametype(msg), lower, upper,
-                                          scope, &dports);
+
+                /* Create destination port list: */
+                tipc_nametbl_mc_lookup(net, type, lower, upper,
+                                       scope, exact, &dports);
+
+                /* Clone message per destination */
                 while (tipc_dest_pop(&dports, NULL, &portid)) {
-                        _skb = __pskb_copy(skb, hsz, GFP_ATOMIC);
+                        _skb = __pskb_copy(skb, hlen, GFP_ATOMIC);
                         if (_skb) {
                                 msg_set_destport(buf_msg(_skb), portid);
                                 __skb_queue_tail(&tmpq, _skb);
@@ -2731,7 +2748,6 @@ void tipc_sk_rht_destroy(struct net *net)
 static int tipc_sk_join(struct tipc_sock *tsk, struct tipc_group_req *mreq)
 {
         struct net *net = sock_net(&tsk->sk);
-        u32 domain = addr_domain(net, mreq->scope);
         struct tipc_group *grp = tsk->group;
         struct tipc_msg *hdr = &tsk->phdr;
         struct tipc_name_seq seq;
@@ -2739,6 +2755,8 @@ static int tipc_sk_join(struct tipc_sock *tsk, struct tipc_group_req *mreq)
 
         if (mreq->type < TIPC_RESERVED_TYPES)
                 return -EACCES;
+        if (mreq->scope > TIPC_NODE_SCOPE)
+                return -EINVAL;
         if (grp)
                 return -EACCES;
         grp = tipc_group_create(net, tsk->portid, mreq);
@@ -2751,7 +2769,7 @@ static int tipc_sk_join(struct tipc_sock *tsk, struct tipc_group_req *mreq)
         seq.type = mreq->type;
         seq.lower = mreq->instance;
         seq.upper = seq.lower;
-        tipc_nametbl_build_group(net, grp, mreq->type, domain);
+        tipc_nametbl_build_group(net, grp, mreq->type, mreq->scope);
         rc = tipc_sk_publish(tsk, mreq->scope, &seq);
         if (rc) {
                 tipc_group_delete(net, grp);