tipc: introduce new tipc_sk_respond() function

Currently, we use the code sequence

if (msg_reverse())
   tipc_link_xmit_skb()

at numerous locations in socket.c. The preparation of arguments
for these calls, as well as the sequence itself, makes the code
unecessarily complex.

In this commit, we introduce a new function, tipc_sk_respond(),
that performs this call combination. We also replace some, but not
yet all, of these explicit call sequences with calls to the new
function. Notably, we let the function tipc_sk_proto_rcv() use
the new function to directly send out PROBE_REPLY messages,
instead of deferring this to the calling tipc_sk_rcv() function,
as we do now.

Reviewed-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>

3 files changed, 47 insertions, 39 deletions

diff --git a/net/tipc/msg.c b/net/tipc/msg.c
index 4339aab93034..b6cc58ec7346 100644
--- a/net/tipc/msg.c
+++ b/net/tipc/msg.c
@@ -469,7 +469,7 @@ bool tipc_msg_make_bundle(struct sk_buff **skb,  struct tipc_msg *msg,
  * Consumes buffer at failure
  * Returns true if success, otherwise false
  */
-bool tipc_msg_reverse(u32 own_node,  struct sk_buff **skb, u32 *dnode, int err)
+bool tipc_msg_reverse(u32 own_node,  struct sk_buff **skb, int err)
 {
         struct sk_buff *_skb = *skb;
         struct tipc_msg *hdr = buf_msg(_skb);
@@ -508,7 +508,6 @@ bool tipc_msg_reverse(u32 own_node,  struct sk_buff **skb, u32 *dnode, int err)
         msg_set_prevnode(hdr, own_node);
         msg_set_orignode(hdr, own_node);
         msg_set_size(hdr, msg_hdr_sz(hdr) + dlen);
-        *dnode = msg_destnode(hdr);
         skb_trim(_skb, msg_size(hdr));
         skb_orphan(_skb);
         return true;
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index 0e96f59e3315..d0834bc519aa 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -785,7 +785,7 @@ static inline bool msg_peer_is_up(struct tipc_msg *m)
 
 struct sk_buff *tipc_buf_acquire(u32 size);
 bool tipc_msg_validate(struct sk_buff *skb);
-bool tipc_msg_reverse(u32 own_addr, struct sk_buff **skb, u32 *dnode, int err);
+bool tipc_msg_reverse(u32 own_addr, struct sk_buff **skb, int err);
 void tipc_msg_init(u32 own_addr, struct tipc_msg *m, u32 user, u32 type,
                    u32 hsize, u32 destnode);
 struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz,
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index e2d5b9831485..71d88adadb18 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -248,6 +248,22 @@ static void tsk_advance_rx_queue(struct sock *sk)
         kfree_skb(__skb_dequeue(&sk->sk_receive_queue));
 }
 
+/* tipc_sk_respond() : send response message back to sender
+ */
+static void tipc_sk_respond(struct sock *sk, struct sk_buff *skb, int err)
+{
+        u32 selector;
+        u32 dnode;
+        u32 onode = tipc_own_addr(sock_net(sk));
+
+        if (!tipc_msg_reverse(onode, &skb, err))
+                return;
+
+        dnode = msg_destnode(buf_msg(skb));
+        selector = msg_origport(buf_msg(skb));
+        tipc_node_xmit_skb(sock_net(sk), skb, dnode, selector);
+}
+
 /**
  * tsk_rej_rx_queue - reject all buffers in socket receive queue
  *
@@ -256,13 +272,9 @@ static void tsk_advance_rx_queue(struct sock *sk)
 static void tsk_rej_rx_queue(struct sock *sk)
 {
         struct sk_buff *skb;
-        u32 dnode;
-        u32 own_node = tsk_own_node(tipc_sk(sk));
 
-        while ((skb = __skb_dequeue(&sk->sk_receive_queue))) {
-                if (tipc_msg_reverse(own_node, &skb, &dnode, TIPC_ERR_NO_PORT))
-                        tipc_node_xmit_skb(sock_net(sk), skb, dnode, 0);
-        }
+        while ((skb = __skb_dequeue(&sk->sk_receive_queue)))
+                tipc_sk_respond(sk, skb, TIPC_ERR_NO_PORT);
 }
 
 /* tsk_peer_msg - verify if message was sent by connected port's peer
@@ -441,9 +453,7 @@ static int tipc_release(struct socket *sock)
                                 tsk->connected = 0;
                                 tipc_node_remove_conn(net, dnode, tsk->portid);
                         }
-                        if (tipc_msg_reverse(tsk_own_node(tsk), &skb, &dnode,
-                                             TIPC_ERR_NO_PORT))
-                                tipc_node_xmit_skb(net, skb, dnode, 0);
+                        tipc_sk_respond(sk, skb, TIPC_ERR_NO_PORT);
                 }
         }
 
@@ -764,35 +774,35 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
 /**
  * tipc_sk_proto_rcv - receive a connection mng protocol message
  * @tsk: receiving socket
- * @skb: pointer to message buffer. Set to NULL if buffer is consumed.
+ * @skb: pointer to message buffer.
  */
-static void tipc_sk_proto_rcv(struct tipc_sock *tsk, struct sk_buff **skb)
+static void tipc_sk_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb)
 {
-        struct tipc_msg *msg = buf_msg(*skb);
+        struct sock *sk = &tsk->sk;
+        struct tipc_msg *hdr = buf_msg(skb);
+        int mtyp = msg_type(hdr);
         int conn_cong;
-        u32 dnode;
-        u32 own_node = tsk_own_node(tsk);
+
         /* Ignore if connection cannot be validated: */
-        if (!tsk_peer_msg(tsk, msg))
+        if (!tsk_peer_msg(tsk, hdr))
                 goto exit;
 
         tsk->probing_state = TIPC_CONN_OK;
 
-        if (msg_type(msg) == CONN_ACK) {
+        if (mtyp == CONN_PROBE) {
+                msg_set_type(hdr, CONN_PROBE_REPLY);
+                tipc_sk_respond(sk, skb, TIPC_OK);
+                return;
+        } else if (mtyp == CONN_ACK) {
                 conn_cong = tsk_conn_cong(tsk);
-                tsk->sent_unacked -= msg_msgcnt(msg);
+                tsk->sent_unacked -= msg_msgcnt(hdr);
                 if (conn_cong)
-                        tsk->sk.sk_write_space(&tsk->sk);
-        } else if (msg_type(msg) == CONN_PROBE) {
-                if (tipc_msg_reverse(own_node, skb, &dnode, TIPC_OK)) {
-                        msg_set_type(msg, CONN_PROBE_REPLY);
-                        return;
-                }
+                        sk->sk_write_space(sk);
+        } else if (mtyp != CONN_PROBE_REPLY) {
+                pr_warn("Received unknown CONN_PROTO msg\n");
         }
-        /* Do nothing if msg_type() == CONN_PROBE_REPLY */
 exit:
-        kfree_skb(*skb);
-        *skb = NULL;
+        kfree_skb(skb);
 }
 
 static int tipc_wait_for_sndmsg(struct socket *sock, long *timeo_p)
@@ -1638,7 +1648,7 @@ static int filter_rcv(struct sock *sk, struct sk_buff **skb)
         int rc = TIPC_OK;
 
         if (unlikely(msg_user(msg) == CONN_MANAGER)) {
-                tipc_sk_proto_rcv(tsk, skb);
+                tipc_sk_proto_rcv(tsk, *skb);
                 return TIPC_OK;
         }
 
@@ -1690,7 +1700,7 @@ static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb)
 {
         int err;
         atomic_t *dcnt;
-        u32 dnode;
+        u32 dnode = msg_prevnode(buf_msg(skb));
         struct tipc_sock *tsk = tipc_sk(sk);
         struct net *net = sock_net(sk);
         uint truesize = skb->truesize;
@@ -1702,7 +1712,7 @@ static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb)
                         atomic_add(truesize, dcnt);
                 return 0;
         }
-        if (!err || tipc_msg_reverse(tsk_own_node(tsk), &skb, &dnode, -err))
+        if (!err || tipc_msg_reverse(tsk_own_node(tsk), &skb, -err))
                 tipc_node_xmit_skb(net, skb, dnode, tsk->portid);
         return 0;
 }
@@ -1794,9 +1804,11 @@ int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq)
                 if (!err) {
                         dnode = msg_destnode(buf_msg(skb));
                         goto xmit;
+                } else {
+                        dnode = msg_prevnode(buf_msg(skb));
                 }
                 tn = net_generic(net, tipc_net_id);
-                if (!tipc_msg_reverse(tn->own_addr, &skb, &dnode, -err))
+                if (!tipc_msg_reverse(tn->own_addr, &skb, -err))
                         continue;
 xmit:
                 tipc_node_xmit_skb(net, skb, dnode, dport);
@@ -2083,6 +2095,8 @@ static int tipc_shutdown(struct socket *sock, int how)
         case SS_CONNECTED:
 
 restart:
+                dnode = tsk_peer_node(tsk);
+
                 /* Disconnect and send a 'FIN+' or 'FIN-' message to peer */
                 skb = __skb_dequeue(&sk->sk_receive_queue);
                 if (skb) {
@@ -2090,13 +2104,8 @@ restart:
                                 kfree_skb(skb);
                                 goto restart;
                         }
-                        if (tipc_msg_reverse(tsk_own_node(tsk), &skb, &dnode,
-                                             TIPC_CONN_SHUTDOWN))
-                                tipc_node_xmit_skb(net, skb, dnode,
-                                                   tsk->portid);
+                        tipc_sk_respond(sk, skb, TIPC_CONN_SHUTDOWN);
                 } else {
-                        dnode = tsk_peer_node(tsk);
-
                         skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE,
                                               TIPC_CONN_MSG, SHORT_H_SIZE,
                                               0, dnode, tsk_own_node(tsk),