tipc: use generic SKB list APIs to manage TIPC outgoing packet chains
Use standard SKB list APIs associated with struct sk_buff_head to
manage socket outgoing packet chain and name table outgoing packet
chain, having relevant code simpler and more readable.
Signed-off-by: Ying Xue <ying.xue@windriver.com>
Reviewed-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
diff --git a/net/tipc/link.c b/net/tipc/link.c
index 0e04508..34bf15c 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -664,9 +664,10 @@
* - For all other messages we discard the buffer and return -EHOSTUNREACH
* - For TIPC internal messages we also reset the link
*/
-static int tipc_link_cong(struct tipc_link *link, struct sk_buff *buf)
+static int tipc_link_cong(struct tipc_link *link, struct sk_buff_head *list)
{
- struct tipc_msg *msg = buf_msg(buf);
+ struct sk_buff *skb = skb_peek(list);
+ struct tipc_msg *msg = buf_msg(skb);
uint imp = tipc_msg_tot_importance(msg);
u32 oport = msg_tot_origport(msg);
@@ -679,28 +680,29 @@
goto drop;
if (unlikely(msg_reroute_cnt(msg)))
goto drop;
- if (TIPC_SKB_CB(buf)->wakeup_pending)
+ if (TIPC_SKB_CB(skb)->wakeup_pending)
return -ELINKCONG;
- if (link_schedule_user(link, oport, TIPC_SKB_CB(buf)->chain_sz, imp))
+ if (link_schedule_user(link, oport, skb_queue_len(list), imp))
return -ELINKCONG;
drop:
- kfree_skb_list(buf);
+ __skb_queue_purge(list);
return -EHOSTUNREACH;
}
/**
* __tipc_link_xmit(): same as tipc_link_xmit, but destlink is known & locked
* @link: link to use
- * @skb: chain of buffers containing message
+ * @list: chain of buffers containing message
+ *
* Consumes the buffer chain, except when returning -ELINKCONG
* Returns 0 if success, otherwise errno: -ELINKCONG, -EMSGSIZE (plain socket
* user data messages) or -EHOSTUNREACH (all other messages/senders)
* Only the socket functions tipc_send_stream() and tipc_send_packet() need
* to act on the return value, since they may need to do more send attempts.
*/
-int __tipc_link_xmit(struct tipc_link *link, struct sk_buff *skb)
+int __tipc_link_xmit(struct tipc_link *link, struct sk_buff_head *list)
{
- struct tipc_msg *msg = buf_msg(skb);
+ struct tipc_msg *msg = buf_msg(skb_peek(list));
uint psz = msg_size(msg);
uint sndlim = link->queue_limit[0];
uint imp = tipc_msg_tot_importance(msg);
@@ -710,21 +712,21 @@
uint bc_last_in = link->owner->bclink.last_in;
struct tipc_media_addr *addr = &link->media_addr;
struct sk_buff_head *outqueue = &link->outqueue;
- struct sk_buff *next;
+ struct sk_buff *skb, *tmp;
/* Match queue limits against msg importance: */
if (unlikely(skb_queue_len(outqueue) >= link->queue_limit[imp]))
- return tipc_link_cong(link, skb);
+ return tipc_link_cong(link, list);
/* Has valid packet limit been used ? */
if (unlikely(psz > mtu)) {
- kfree_skb_list(skb);
+ __skb_queue_purge(list);
return -EMSGSIZE;
}
/* Prepare each packet for sending, and add to outqueue: */
- while (skb) {
- next = skb->next;
+ skb_queue_walk_safe(list, skb, tmp) {
+ __skb_unlink(skb, list);
msg = buf_msg(skb);
msg_set_word(msg, 2, ((ack << 16) | mod(seqno)));
msg_set_bcast_ack(msg, bc_last_in);
@@ -736,7 +738,6 @@
link->unacked_window = 0;
} else if (tipc_msg_bundle(outqueue, skb, mtu)) {
link->stats.sent_bundled++;
- skb = next;
continue;
} else if (tipc_msg_make_bundle(outqueue, skb, mtu,
link->addr)) {
@@ -750,22 +751,43 @@
link->next_out = skb;
}
seqno++;
- skb = next;
}
link->next_out_no = seqno;
return 0;
}
+static void skb2list(struct sk_buff *skb, struct sk_buff_head *list)
+{
+ __skb_queue_head_init(list);
+ __skb_queue_tail(list, skb);
+}
+
+static int __tipc_link_xmit_skb(struct tipc_link *link, struct sk_buff *skb)
+{
+ struct sk_buff_head head;
+
+ skb2list(skb, &head);
+ return __tipc_link_xmit(link, &head);
+}
+
+int tipc_link_xmit_skb(struct sk_buff *skb, u32 dnode, u32 selector)
+{
+ struct sk_buff_head head;
+
+ skb2list(skb, &head);
+ return tipc_link_xmit(&head, dnode, selector);
+}
+
/**
* tipc_link_xmit() is the general link level function for message sending
- * @buf: chain of buffers containing message
+ * @list: chain of buffers containing message
* @dsz: amount of user data to be sent
* @dnode: address of destination node
* @selector: a number used for deterministic link selection
* Consumes the buffer chain, except when returning -ELINKCONG
* Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
*/
-int tipc_link_xmit(struct sk_buff *buf, u32 dnode, u32 selector)
+int tipc_link_xmit(struct sk_buff_head *list, u32 dnode, u32 selector)
{
struct tipc_link *link = NULL;
struct tipc_node *node;
@@ -776,17 +798,22 @@
tipc_node_lock(node);
link = node->active_links[selector & 1];
if (link)
- rc = __tipc_link_xmit(link, buf);
+ rc = __tipc_link_xmit(link, list);
tipc_node_unlock(node);
}
if (link)
return rc;
- if (likely(in_own_node(dnode)))
- return tipc_sk_rcv(buf);
+ if (likely(in_own_node(dnode))) {
+ /* As a node local message chain never contains more than one
+ * buffer, we just need to dequeue one SKB buffer from the
+ * head list.
+ */
+ return tipc_sk_rcv(__skb_dequeue(list));
+ }
+ __skb_queue_purge(list);
- kfree_skb_list(buf);
return rc;
}
@@ -800,17 +827,17 @@
*/
static void tipc_link_sync_xmit(struct tipc_link *link)
{
- struct sk_buff *buf;
+ struct sk_buff *skb;
struct tipc_msg *msg;
- buf = tipc_buf_acquire(INT_H_SIZE);
- if (!buf)
+ skb = tipc_buf_acquire(INT_H_SIZE);
+ if (!skb)
return;
- msg = buf_msg(buf);
+ msg = buf_msg(skb);
tipc_msg_init(msg, BCAST_PROTOCOL, STATE_MSG, INT_H_SIZE, link->addr);
msg_set_last_bcast(msg, link->owner->bclink.acked);
- __tipc_link_xmit(link, buf);
+ __tipc_link_xmit_skb(link, skb);
}
/*
@@ -1053,8 +1080,7 @@
u32 ackd;
u32 released;
- __skb_queue_head_init(&head);
- __skb_queue_tail(&head, skb);
+ skb2list(skb, &head);
while ((skb = __skb_dequeue(&head))) {
/* Ensure message is well-formed */
@@ -1573,7 +1599,7 @@
u32 selector)
{
struct tipc_link *tunnel;
- struct sk_buff *buf;
+ struct sk_buff *skb;
u32 length = msg_size(msg);
tunnel = l_ptr->owner->active_links[selector & 1];
@@ -1582,14 +1608,14 @@
return;
}
msg_set_size(tunnel_hdr, length + INT_H_SIZE);
- buf = tipc_buf_acquire(length + INT_H_SIZE);
- if (!buf) {
+ skb = tipc_buf_acquire(length + INT_H_SIZE);
+ if (!skb) {
pr_warn("%sunable to send tunnel msg\n", link_co_err);
return;
}
- skb_copy_to_linear_data(buf, tunnel_hdr, INT_H_SIZE);
- skb_copy_to_linear_data_offset(buf, INT_H_SIZE, msg, length);
- __tipc_link_xmit(tunnel, buf);
+ skb_copy_to_linear_data(skb, tunnel_hdr, INT_H_SIZE);
+ skb_copy_to_linear_data_offset(skb, INT_H_SIZE, msg, length);
+ __tipc_link_xmit_skb(tunnel, skb);
}
@@ -1620,7 +1646,7 @@
if (skb) {
skb_copy_to_linear_data(skb, &tunnel_hdr, INT_H_SIZE);
msg_set_size(&tunnel_hdr, INT_H_SIZE);
- __tipc_link_xmit(tunnel, skb);
+ __tipc_link_xmit_skb(tunnel, skb);
} else {
pr_warn("%sunable to send changeover msg\n",
link_co_err);
@@ -1691,7 +1717,7 @@
skb_copy_to_linear_data(outskb, &tunnel_hdr, INT_H_SIZE);
skb_copy_to_linear_data_offset(outskb, INT_H_SIZE, skb->data,
length);
- __tipc_link_xmit(tunnel, outskb);
+ __tipc_link_xmit_skb(tunnel, outskb);
if (!tipc_link_is_up(l_ptr))
return;
}