tipc: use pseudo message to wake up sockets after link congestion

The current link implementation keeps a linked list of blocked ports/
sockets that is populated when there is link congestion. The purpose
of this is to let the link know which users to wake up when the
congestion abates.

This adds unnecessary complexity to the data structure and the code,
since it forces us to involve the link each time we want to delete
a socket. It also forces us to grab the spinlock port_lock within
the scope of node_lock. We want to get rid of this direct dependence,
as well as the deadlock hazard resulting from the usage of port_lock.

In this commit, we instead let the link keep list of a "wakeup" pseudo
messages for use in such situations. Those messages are sent to the
pending sockets via the ordinary message reception path, and wake up
the socket's owner when they are received.

This enables us to get rid of the 'waiting_ports' linked lists in struct
tipc_port that manifest this direct reference. As a consequence, we can
eliminate another BH entry into the socket, and hence the need to grab
port_lock. This is a further step in our effort to remove port_lock
altogether.

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

1 files changed, 5 insertions, 2 deletions

diff --git a/net/tipc/msg.c b/net/tipc/msg.c
index fdb92e247050..74745a47d72a 100644
--- a/net/tipc/msg.c
+++ b/net/tipc/msg.c
@@ -182,7 +182,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct iovec const *iov,
         struct sk_buff *buf, *prev;
         char *pktpos;
         int rc;
-
+        uint chain_sz = 0;
         msg_set_size(mhdr, msz);
 
         /* No fragmentation needed? */
@@ -193,6 +193,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct iovec const *iov,
                         return -ENOMEM;
                 skb_copy_to_linear_data(buf, mhdr, mhsz);
                 pktpos = buf->data + mhsz;
+                TIPC_SKB_CB(buf)->chain_sz = 1;
                 if (!dsz || !memcpy_fromiovecend(pktpos, iov, offset, dsz))
                         return dsz;
                 rc = -EFAULT;
@@ -209,6 +210,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct iovec const *iov,
         *chain = buf = tipc_buf_acquire(pktmax);
         if (!buf)
                 return -ENOMEM;
+        chain_sz = 1;
         pktpos = buf->data;
         skb_copy_to_linear_data(buf, &pkthdr, INT_H_SIZE);
         pktpos += INT_H_SIZE;
@@ -242,6 +244,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct iovec const *iov,
                         rc = -ENOMEM;
                         goto error;
                 }
+                chain_sz++;
                 prev->next = buf;
                 msg_set_type(&pkthdr, FRAGMENT);
                 msg_set_size(&pkthdr, pktsz);
@@ -251,7 +254,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct iovec const *iov,
                 pktrem = pktsz - INT_H_SIZE;
 
         } while (1);
-
+        TIPC_SKB_CB(*chain)->chain_sz = chain_sz;
         msg_set_type(buf_msg(buf), LAST_FRAGMENT);
         return dsz;
 error: