tipc: receive group membership events via member socket

Like with any other service, group members' availability can be
subscribed for by connecting to be topology server. However, because
the events arrive via a different socket than the member socket, there
is a real risk that membership events my arrive out of synch with the
actual JOIN/LEAVE action. I.e., it is possible to receive the first
messages from a new member before the corresponding JOIN event arrives,
just as it is possible to receive the last messages from a leaving
member after the LEAVE event has already been received.

Since each member socket is internally also subscribing for membership
events, we now fix this problem by passing those events on to the user
via the member socket. We leverage the already present member synch-
ronization protocol to guarantee correct message/event order. An event
is delivered to the user as an empty message where the two source
addresses identify the new/lost member. Furthermore, we set the MSG_OOB
bit in the message flags to mark it as an event. If the event is an
indication about a member loss we also set the MSG_EOR bit, so it can
be distinguished from a member addition event.

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

1 files changed, 31 insertions, 18 deletions

diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index 25ecf1201527..0a2eac309177 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -709,41 +709,43 @@ static unsigned int tipc_poll(struct file *file, struct socket *sock,
                               poll_table *wait)
 {
         struct sock *sk = sock->sk;
+        struct sk_buff *skb = skb_peek(&sk->sk_receive_queue);
         struct tipc_sock *tsk = tipc_sk(sk);
         struct tipc_group *grp = tsk->group;
-        u32 mask = 0;
+        u32 revents = 0;
 
         sock_poll_wait(file, sk_sleep(sk), wait);
 
         if (sk->sk_shutdown & RCV_SHUTDOWN)
-                mask |= POLLRDHUP | POLLIN | POLLRDNORM;
+                revents |= POLLRDHUP | POLLIN | POLLRDNORM;
         if (sk->sk_shutdown == SHUTDOWN_MASK)
-                mask |= POLLHUP;
+                revents |= POLLHUP;
 
         switch (sk->sk_state) {
         case TIPC_ESTABLISHED:
                 if (!tsk->cong_link_cnt && !tsk_conn_cong(tsk))
-                        mask |= POLLOUT;
+                        revents |= POLLOUT;
                 /* fall thru' */
         case TIPC_LISTEN:
         case TIPC_CONNECTING:
-                if (!skb_queue_empty(&sk->sk_receive_queue))
-                        mask |= (POLLIN | POLLRDNORM);
+                if (skb)
+                        revents |= POLLIN | POLLRDNORM;
                 break;
         case TIPC_OPEN:
                 if (!grp || tipc_group_size(grp))
                         if (!tsk->cong_link_cnt)
-                                mask |= POLLOUT;
-                if (tipc_sk_type_connectionless(sk) &&
-                    (!skb_queue_empty(&sk->sk_receive_queue)))
-                        mask |= (POLLIN | POLLRDNORM);
+                                revents |= POLLOUT;
+                if (!tipc_sk_type_connectionless(sk))
+                        break;
+                if (!skb)
+                        break;
+                revents |= POLLIN | POLLRDNORM;
                 break;
         case TIPC_DISCONNECTING:
-                mask = (POLLIN | POLLRDNORM | POLLHUP);
+                revents = POLLIN | POLLRDNORM | POLLHUP;
                 break;
         }
-
-        return mask;
+        return revents;
 }
 
 /**
@@ -1415,11 +1417,12 @@ static int tipc_recvmsg(struct socket *sock, struct msghdr *m,
                         size_t buflen,  int flags)
 {
         struct sock *sk = sock->sk;
-        struct tipc_sock *tsk = tipc_sk(sk);
-        struct sk_buff *skb;
-        struct tipc_msg *hdr;
         bool connected = !tipc_sk_type_connectionless(sk);
+        struct tipc_sock *tsk = tipc_sk(sk);
         int rc, err, hlen, dlen, copy;
+        struct tipc_msg *hdr;
+        struct sk_buff *skb;
+        bool grp_evt;
         long timeout;
 
         /* Catch invalid receive requests */
@@ -1443,6 +1446,7 @@ static int tipc_recvmsg(struct socket *sock, struct msghdr *m,
                 dlen = msg_data_sz(hdr);
                 hlen = msg_hdr_sz(hdr);
                 err = msg_errcode(hdr);
+                grp_evt = msg_is_grp_evt(hdr);
                 if (likely(dlen || err))
                         break;
                 tsk_advance_rx_queue(sk);
@@ -1469,11 +1473,20 @@ static int tipc_recvmsg(struct socket *sock, struct msghdr *m,
         if (unlikely(rc))
                 goto exit;
 
+        /* Mark message as group event if applicable */
+        if (unlikely(grp_evt)) {
+                if (msg_grp_evt(hdr) == TIPC_WITHDRAWN)
+                        m->msg_flags |= MSG_EOR;
+                m->msg_flags |= MSG_OOB;
+                copy = 0;
+        }
+
         /* Caption of data or error code/rejected data was successful */
         if (unlikely(flags & MSG_PEEK))
                 goto exit;
 
         tsk_advance_rx_queue(sk);
+
         if (likely(!connected))
                 goto exit;
 
@@ -1648,10 +1661,10 @@ static void tipc_sk_proto_rcv(struct sock *sk,
                 sk->sk_write_space(sk);
                 break;
         case GROUP_PROTOCOL:
-                tipc_group_proto_rcv(grp, hdr, xmitq);
+                tipc_group_proto_rcv(grp, hdr, inputq, xmitq);
                 break;
         case TOP_SRV:
-                tipc_group_member_evt(tsk->group, skb, xmitq);
+                tipc_group_member_evt(tsk->group, skb, inputq, xmitq);
                 skb = NULL;
                 break;
         default: