Skip to content

Commit

Permalink
tipc: use generic SKB list APIs to manage link transmission queue
Browse files Browse the repository at this point in the history 
Use standard SKB list APIs associated with struct sk_buff_head to
manage link transmission queue, having relevant code more clean.

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>
  • Loading branch information
Ying Xue authored and David S. Miller committed Nov 26, 2014
1 parent 58d78b3 commit 58dc55f
Show file tree
Hide file tree
Showing 5 changed files with 153 additions and 176 deletions.
67 changes: 29 additions & 38 deletions net/tipc/bcast.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -217,12 +217,13 @@ struct tipc_node *tipc_bclink_retransmit_to(void)
*/
static void bclink_retransmit_pkt(u32 after, u32 to)
{
struct sk_buff *buf;
struct sk_buff *skb;

buf = bcl->first_out;
while (buf && less_eq(buf_seqno(buf), after))
buf = buf->next;
tipc_link_retransmit(bcl, buf, mod(to - after));
skb_queue_walk(&bcl->outqueue, skb) {
if (more(buf_seqno(skb), after))
break;
}
tipc_link_retransmit(bcl, skb, mod(to - after));
}

/**
Expand All @@ -245,14 +246,14 @@ void tipc_bclink_wakeup_users(void)
*/
void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
{
struct sk_buff *crs;
struct sk_buff *skb, *tmp;
struct sk_buff *next;
unsigned int released = 0;

tipc_bclink_lock();
/* Bail out if tx queue is empty (no clean up is required) */
crs = bcl->first_out;
if (!crs)
skb = skb_peek(&bcl->outqueue);
if (!skb)
goto exit;

/* Determine which messages need to be acknowledged */
Expand All @@ -271,41 +272,41 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
* Bail out if specified sequence number does not correspond
* to a message that has been sent and not yet acknowledged
*/
if (less(acked, buf_seqno(crs)) ||
if (less(acked, buf_seqno(skb)) ||
less(bcl->fsm_msg_cnt, acked) ||
less_eq(acked, n_ptr->bclink.acked))
goto exit;
}

/* Skip over packets that node has previously acknowledged */
while (crs && less_eq(buf_seqno(crs), n_ptr->bclink.acked))
crs = crs->next;
skb_queue_walk(&bcl->outqueue, skb) {
if (more(buf_seqno(skb), n_ptr->bclink.acked))
break;
}

/* Update packets that node is now acknowledging */
skb_queue_walk_from_safe(&bcl->outqueue, skb, tmp) {
if (more(buf_seqno(skb), acked))
break;

while (crs && less_eq(buf_seqno(crs), acked)) {
next = crs->next;

if (crs != bcl->next_out)
bcbuf_decr_acks(crs);
else {
bcbuf_set_acks(crs, 0);
next = tipc_skb_queue_next(&bcl->outqueue, skb);
if (skb != bcl->next_out) {
bcbuf_decr_acks(skb);
} else {
bcbuf_set_acks(skb, 0);
bcl->next_out = next;
bclink_set_last_sent();
}

if (bcbuf_acks(crs) == 0) {
bcl->first_out = next;
bcl->out_queue_size--;
kfree_skb(crs);
if (bcbuf_acks(skb) == 0) {
__skb_unlink(skb, &bcl->outqueue);
kfree_skb(skb);
released = 1;
}
crs = next;
}
n_ptr->bclink.acked = acked;

/* Try resolving broadcast link congestion, if necessary */

if (unlikely(bcl->next_out)) {
tipc_link_push_packets(bcl);
bclink_set_last_sent();
Expand All @@ -327,32 +328,27 @@ void tipc_bclink_update_link_state(struct tipc_node *n_ptr, u32 last_sent)
struct sk_buff *buf;

/* Ignore "stale" link state info */

if (less_eq(last_sent, n_ptr->bclink.last_in))
return;

/* Update link synchronization state; quit if in sync */

bclink_update_last_sent(n_ptr, last_sent);

if (n_ptr->bclink.last_sent == n_ptr->bclink.last_in)
return;

/* Update out-of-sync state; quit if loss is still unconfirmed */

if ((++n_ptr->bclink.oos_state) == 1) {
if (n_ptr->bclink.deferred_size < (TIPC_MIN_LINK_WIN / 2))
return;
n_ptr->bclink.oos_state++;
}

/* Don't NACK if one has been recently sent (or seen) */

if (n_ptr->bclink.oos_state & 0x1)
return;

/* Send NACK */

buf = tipc_buf_acquire(INT_H_SIZE);
if (buf) {
struct tipc_msg *msg = buf_msg(buf);
Expand Down Expand Up @@ -425,9 +421,11 @@ int tipc_bclink_xmit(struct sk_buff *buf)
if (likely(bclink->bcast_nodes.count)) {
rc = __tipc_link_xmit(bcl, buf);
if (likely(!rc)) {
u32 len = skb_queue_len(&bcl->outqueue);

bclink_set_last_sent();
bcl->stats.queue_sz_counts++;
bcl->stats.accu_queue_sz += bcl->out_queue_size;
bcl->stats.accu_queue_sz += len;
}
bc = 1;
}
Expand Down Expand Up @@ -462,7 +460,6 @@ static void bclink_accept_pkt(struct tipc_node *node, u32 seqno)
* Unicast an ACK periodically, ensuring that
* all nodes in the cluster don't ACK at the same time
*/

if (((seqno - tipc_own_addr) % TIPC_MIN_LINK_WIN) == 0) {
tipc_link_proto_xmit(node->active_links[node->addr & 1],
STATE_MSG, 0, 0, 0, 0, 0);
Expand All @@ -484,7 +481,6 @@ void tipc_bclink_rcv(struct sk_buff *buf)
int deferred = 0;

/* Screen out unwanted broadcast messages */

if (msg_mc_netid(msg) != tipc_net_id)
goto exit;

Expand All @@ -497,7 +493,6 @@ void tipc_bclink_rcv(struct sk_buff *buf)
goto unlock;

/* Handle broadcast protocol message */

if (unlikely(msg_user(msg) == BCAST_PROTOCOL)) {
if (msg_type(msg) != STATE_MSG)
goto unlock;
Expand All @@ -518,14 +513,12 @@ void tipc_bclink_rcv(struct sk_buff *buf)
}

/* Handle in-sequence broadcast message */

seqno = msg_seqno(msg);
next_in = mod(node->bclink.last_in + 1);

if (likely(seqno == next_in)) {
receive:
/* Deliver message to destination */

if (likely(msg_isdata(msg))) {
tipc_bclink_lock();
bclink_accept_pkt(node, seqno);
Expand Down Expand Up @@ -574,7 +567,6 @@ void tipc_bclink_rcv(struct sk_buff *buf)
buf = NULL;

/* Determine new synchronization state */

tipc_node_lock(node);
if (unlikely(!tipc_node_is_up(node)))
goto unlock;
Expand All @@ -594,7 +586,6 @@ void tipc_bclink_rcv(struct sk_buff *buf)
goto unlock;

/* Take in-sequence message from deferred queue & deliver it */

buf = node->bclink.deferred_head;
node->bclink.deferred_head = buf->next;
buf->next = NULL;
Expand All @@ -603,7 +594,6 @@ void tipc_bclink_rcv(struct sk_buff *buf)
}

/* Handle out-of-sequence broadcast message */

if (less(next_in, seqno)) {
deferred = tipc_link_defer_pkt(&node->bclink.deferred_head,
&node->bclink.deferred_tail,
Expand Down Expand Up @@ -963,6 +953,7 @@ int tipc_bclink_init(void)
sprintf(bcbearer->media.name, "tipc-broadcast");

spin_lock_init(&bclink->lock);
__skb_queue_head_init(&bcl->outqueue);
__skb_queue_head_init(&bcl->waiting_sks);
bcl->next_out_no = 1;
spin_lock_init(&bclink->node.lock);
Expand Down
Loading

0 comments on commit 58dc55f

Please sign in to comment.