Commit 05dcc5aa authored by Jon Paul Maloy's avatar Jon Paul Maloy Committed by David S. Miller

tipc: split link outqueue

struct tipc_link contains one single queue for outgoing packets,
where both transmitted and waiting packets are queued.

This infrastructure is hard to maintain, because we need
to keep a number of fields to keep track of which packets are
sent or unsent, and the number of packets in each category.

A lot of code becomes simpler if we split this queue into a transmission
queue, where sent/unacknowledged packets are kept, and a backlog queue,
where we keep the not yet sent packets.

In this commit we do this separation.
Reviewed-by: default avatarErik Hugne <erik.hugne@ericsson.com>
Reviewed-by: default avatarYing Xue <ying.xue@windriver.com>
Signed-off-by: default avatarJon Maloy <jon.maloy@ericsson.com>
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parent 2cdf3918
...@@ -135,9 +135,10 @@ static void bclink_set_last_sent(struct net *net) ...@@ -135,9 +135,10 @@ static void bclink_set_last_sent(struct net *net)
{ {
struct tipc_net *tn = net_generic(net, tipc_net_id); struct tipc_net *tn = net_generic(net, tipc_net_id);
struct tipc_link *bcl = tn->bcl; struct tipc_link *bcl = tn->bcl;
struct sk_buff *skb = skb_peek(&bcl->backlogq);
if (bcl->next_out) if (skb)
bcl->fsm_msg_cnt = mod(buf_seqno(bcl->next_out) - 1); bcl->fsm_msg_cnt = mod(buf_seqno(skb) - 1);
else else
bcl->fsm_msg_cnt = mod(bcl->next_out_no - 1); bcl->fsm_msg_cnt = mod(bcl->next_out_no - 1);
} }
...@@ -180,7 +181,7 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to) ...@@ -180,7 +181,7 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to)
struct sk_buff *skb; struct sk_buff *skb;
struct tipc_link *bcl = tn->bcl; struct tipc_link *bcl = tn->bcl;
skb_queue_walk(&bcl->outqueue, skb) { skb_queue_walk(&bcl->transmq, skb) {
if (more(buf_seqno(skb), after)) { if (more(buf_seqno(skb), after)) {
tipc_link_retransmit(bcl, skb, mod(to - after)); tipc_link_retransmit(bcl, skb, mod(to - after));
break; break;
...@@ -210,7 +211,6 @@ void tipc_bclink_wakeup_users(struct net *net) ...@@ -210,7 +211,6 @@ void tipc_bclink_wakeup_users(struct net *net)
void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
{ {
struct sk_buff *skb, *tmp; struct sk_buff *skb, *tmp;
struct sk_buff *next;
unsigned int released = 0; unsigned int released = 0;
struct net *net = n_ptr->net; struct net *net = n_ptr->net;
struct tipc_net *tn = net_generic(net, tipc_net_id); struct tipc_net *tn = net_generic(net, tipc_net_id);
...@@ -221,7 +221,7 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) ...@@ -221,7 +221,7 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
tipc_bclink_lock(net); tipc_bclink_lock(net);
/* Bail out if tx queue is empty (no clean up is required) */ /* Bail out if tx queue is empty (no clean up is required) */
skb = skb_peek(&tn->bcl->outqueue); skb = skb_peek(&tn->bcl->transmq);
if (!skb) if (!skb)
goto exit; goto exit;
...@@ -248,27 +248,19 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) ...@@ -248,27 +248,19 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
} }
/* Skip over packets that node has previously acknowledged */ /* Skip over packets that node has previously acknowledged */
skb_queue_walk(&tn->bcl->outqueue, skb) { skb_queue_walk(&tn->bcl->transmq, skb) {
if (more(buf_seqno(skb), n_ptr->bclink.acked)) if (more(buf_seqno(skb), n_ptr->bclink.acked))
break; break;
} }
/* Update packets that node is now acknowledging */ /* Update packets that node is now acknowledging */
skb_queue_walk_from_safe(&tn->bcl->outqueue, skb, tmp) { skb_queue_walk_from_safe(&tn->bcl->transmq, skb, tmp) {
if (more(buf_seqno(skb), acked)) if (more(buf_seqno(skb), acked))
break; break;
bcbuf_decr_acks(skb);
next = tipc_skb_queue_next(&tn->bcl->outqueue, skb); bclink_set_last_sent(net);
if (skb != tn->bcl->next_out) {
bcbuf_decr_acks(skb);
} else {
bcbuf_set_acks(skb, 0);
tn->bcl->next_out = next;
bclink_set_last_sent(net);
}
if (bcbuf_acks(skb) == 0) { if (bcbuf_acks(skb) == 0) {
__skb_unlink(skb, &tn->bcl->outqueue); __skb_unlink(skb, &tn->bcl->transmq);
kfree_skb(skb); kfree_skb(skb);
released = 1; released = 1;
} }
...@@ -276,7 +268,7 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) ...@@ -276,7 +268,7 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
n_ptr->bclink.acked = acked; n_ptr->bclink.acked = acked;
/* Try resolving broadcast link congestion, if necessary */ /* Try resolving broadcast link congestion, if necessary */
if (unlikely(tn->bcl->next_out)) { if (unlikely(skb_peek(&tn->bcl->backlogq))) {
tipc_link_push_packets(tn->bcl); tipc_link_push_packets(tn->bcl);
bclink_set_last_sent(net); bclink_set_last_sent(net);
} }
...@@ -323,7 +315,7 @@ void tipc_bclink_update_link_state(struct tipc_node *n_ptr, ...@@ -323,7 +315,7 @@ void tipc_bclink_update_link_state(struct tipc_node *n_ptr,
buf = tipc_buf_acquire(INT_H_SIZE); buf = tipc_buf_acquire(INT_H_SIZE);
if (buf) { if (buf) {
struct tipc_msg *msg = buf_msg(buf); struct tipc_msg *msg = buf_msg(buf);
struct sk_buff *skb = skb_peek(&n_ptr->bclink.deferred_queue); struct sk_buff *skb = skb_peek(&n_ptr->bclink.deferdq);
u32 to = skb ? buf_seqno(skb) - 1 : n_ptr->bclink.last_sent; u32 to = skb ? buf_seqno(skb) - 1 : n_ptr->bclink.last_sent;
tipc_msg_init(tn->own_addr, msg, BCAST_PROTOCOL, STATE_MSG, tipc_msg_init(tn->own_addr, msg, BCAST_PROTOCOL, STATE_MSG,
...@@ -398,7 +390,7 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list) ...@@ -398,7 +390,7 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list)
if (likely(bclink->bcast_nodes.count)) { if (likely(bclink->bcast_nodes.count)) {
rc = __tipc_link_xmit(net, bcl, list); rc = __tipc_link_xmit(net, bcl, list);
if (likely(!rc)) { if (likely(!rc)) {
u32 len = skb_queue_len(&bcl->outqueue); u32 len = skb_queue_len(&bcl->transmq);
bclink_set_last_sent(net); bclink_set_last_sent(net);
bcl->stats.queue_sz_counts++; bcl->stats.queue_sz_counts++;
...@@ -563,25 +555,25 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf) ...@@ -563,25 +555,25 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf)
if (node->bclink.last_in == node->bclink.last_sent) if (node->bclink.last_in == node->bclink.last_sent)
goto unlock; goto unlock;
if (skb_queue_empty(&node->bclink.deferred_queue)) { if (skb_queue_empty(&node->bclink.deferdq)) {
node->bclink.oos_state = 1; node->bclink.oos_state = 1;
goto unlock; goto unlock;
} }
msg = buf_msg(skb_peek(&node->bclink.deferred_queue)); msg = buf_msg(skb_peek(&node->bclink.deferdq));
seqno = msg_seqno(msg); seqno = msg_seqno(msg);
next_in = mod(next_in + 1); next_in = mod(next_in + 1);
if (seqno != next_in) if (seqno != next_in)
goto unlock; goto unlock;
/* Take in-sequence message from deferred queue & deliver it */ /* Take in-sequence message from deferred queue & deliver it */
buf = __skb_dequeue(&node->bclink.deferred_queue); buf = __skb_dequeue(&node->bclink.deferdq);
goto receive; goto receive;
} }
/* Handle out-of-sequence broadcast message */ /* Handle out-of-sequence broadcast message */
if (less(next_in, seqno)) { if (less(next_in, seqno)) {
deferred = tipc_link_defer_pkt(&node->bclink.deferred_queue, deferred = tipc_link_defer_pkt(&node->bclink.deferdq,
buf); buf);
bclink_update_last_sent(node, seqno); bclink_update_last_sent(node, seqno);
buf = NULL; buf = NULL;
...@@ -638,7 +630,6 @@ static int tipc_bcbearer_send(struct net *net, struct sk_buff *buf, ...@@ -638,7 +630,6 @@ static int tipc_bcbearer_send(struct net *net, struct sk_buff *buf,
msg_set_non_seq(msg, 1); msg_set_non_seq(msg, 1);
msg_set_mc_netid(msg, tn->net_id); msg_set_mc_netid(msg, tn->net_id);
tn->bcl->stats.sent_info++; tn->bcl->stats.sent_info++;
if (WARN_ON(!bclink->bcast_nodes.count)) { if (WARN_ON(!bclink->bcast_nodes.count)) {
dump_stack(); dump_stack();
return 0; return 0;
...@@ -917,8 +908,9 @@ int tipc_bclink_init(struct net *net) ...@@ -917,8 +908,9 @@ int tipc_bclink_init(struct net *net)
sprintf(bcbearer->media.name, "tipc-broadcast"); sprintf(bcbearer->media.name, "tipc-broadcast");
spin_lock_init(&bclink->lock); spin_lock_init(&bclink->lock);
__skb_queue_head_init(&bcl->outqueue); __skb_queue_head_init(&bcl->transmq);
__skb_queue_head_init(&bcl->deferred_queue); __skb_queue_head_init(&bcl->backlogq);
__skb_queue_head_init(&bcl->deferdq);
skb_queue_head_init(&bcl->wakeupq); skb_queue_head_init(&bcl->wakeupq);
bcl->next_out_no = 1; bcl->next_out_no = 1;
spin_lock_init(&bclink->node.lock); spin_lock_init(&bclink->node.lock);
......
This diff is collapsed.
...@@ -124,7 +124,8 @@ struct tipc_stats { ...@@ -124,7 +124,8 @@ struct tipc_stats {
* @max_pkt: current maximum packet size for this link * @max_pkt: current maximum packet size for this link
* @max_pkt_target: desired maximum packet size for this link * @max_pkt_target: desired maximum packet size for this link
* @max_pkt_probes: # of probes based on current (max_pkt, max_pkt_target) * @max_pkt_probes: # of probes based on current (max_pkt, max_pkt_target)
* @outqueue: outbound message queue * @transmitq: queue for sent, non-acked messages
* @backlogq: queue for messages waiting to be sent
* @next_out_no: next sequence number to use for outbound messages * @next_out_no: next sequence number to use for outbound messages
* @last_retransmitted: sequence number of most recently retransmitted message * @last_retransmitted: sequence number of most recently retransmitted message
* @stale_count: # of identical retransmit requests made by peer * @stale_count: # of identical retransmit requests made by peer
...@@ -177,20 +178,21 @@ struct tipc_link { ...@@ -177,20 +178,21 @@ struct tipc_link {
u32 max_pkt_probes; u32 max_pkt_probes;
/* Sending */ /* Sending */
struct sk_buff_head outqueue; struct sk_buff_head transmq;
struct sk_buff_head backlogq;
u32 next_out_no; u32 next_out_no;
u32 window;
u32 last_retransmitted; u32 last_retransmitted;
u32 stale_count; u32 stale_count;
/* Reception */ /* Reception */
u32 next_in_no; u32 next_in_no;
struct sk_buff_head deferred_queue; u32 rcv_unacked;
u32 unacked_window; struct sk_buff_head deferdq;
struct sk_buff_head inputq; struct sk_buff_head inputq;
struct sk_buff_head namedq; struct sk_buff_head namedq;
/* Congestion handling */ /* Congestion handling */
struct sk_buff *next_out;
struct sk_buff_head wakeupq; struct sk_buff_head wakeupq;
/* Fragmentation/reassembly */ /* Fragmentation/reassembly */
...@@ -302,9 +304,4 @@ static inline int link_reset_reset(struct tipc_link *l_ptr) ...@@ -302,9 +304,4 @@ static inline int link_reset_reset(struct tipc_link *l_ptr)
return l_ptr->state == RESET_RESET; return l_ptr->state == RESET_RESET;
} }
static inline int link_congested(struct tipc_link *l_ptr)
{
return skb_queue_len(&l_ptr->outqueue) >= l_ptr->queue_limit[0];
}
#endif #endif
...@@ -330,33 +330,36 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m, ...@@ -330,33 +330,36 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m,
/** /**
* tipc_msg_bundle(): Append contents of a buffer to tail of an existing one * tipc_msg_bundle(): Append contents of a buffer to tail of an existing one
* @list: the buffer chain of the existing buffer ("bundle") * @bskb: the buffer to append to ("bundle")
* @skb: buffer to be appended * @skb: buffer to be appended
* @mtu: max allowable size for the bundle buffer * @mtu: max allowable size for the bundle buffer
* Consumes buffer if successful * Consumes buffer if successful
* Returns true if bundling could be performed, otherwise false * Returns true if bundling could be performed, otherwise false
*/ */
bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu) bool tipc_msg_bundle(struct sk_buff *bskb, struct sk_buff *skb, u32 mtu)
{ {
struct sk_buff *bskb = skb_peek_tail(list); struct tipc_msg *bmsg;
struct tipc_msg *bmsg = buf_msg(bskb);
struct tipc_msg *msg = buf_msg(skb); struct tipc_msg *msg = buf_msg(skb);
unsigned int bsz = msg_size(bmsg); unsigned int bsz;
unsigned int msz = msg_size(msg); unsigned int msz = msg_size(msg);
u32 start = align(bsz); u32 start, pad;
u32 max = mtu - INT_H_SIZE; u32 max = mtu - INT_H_SIZE;
u32 pad = start - bsz;
if (likely(msg_user(msg) == MSG_FRAGMENTER)) if (likely(msg_user(msg) == MSG_FRAGMENTER))
return false; return false;
if (!bskb)
return false;
bmsg = buf_msg(bskb);
bsz = msg_size(bmsg);
start = align(bsz);
pad = start - bsz;
if (unlikely(msg_user(msg) == CHANGEOVER_PROTOCOL)) if (unlikely(msg_user(msg) == CHANGEOVER_PROTOCOL))
return false; return false;
if (unlikely(msg_user(msg) == BCAST_PROTOCOL)) if (unlikely(msg_user(msg) == BCAST_PROTOCOL))
return false; return false;
if (likely(msg_user(bmsg) != MSG_BUNDLER)) if (likely(msg_user(bmsg) != MSG_BUNDLER))
return false; return false;
if (likely(!TIPC_SKB_CB(bskb)->bundling))
return false;
if (unlikely(skb_tailroom(bskb) < (pad + msz))) if (unlikely(skb_tailroom(bskb) < (pad + msz)))
return false; return false;
if (unlikely(max < (start + msz))) if (unlikely(max < (start + msz)))
...@@ -419,12 +422,11 @@ bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos) ...@@ -419,12 +422,11 @@ bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos)
* Replaces buffer if successful * Replaces buffer if successful
* Returns true if success, otherwise false * Returns true if success, otherwise false
*/ */
bool tipc_msg_make_bundle(struct sk_buff_head *list, bool tipc_msg_make_bundle(struct sk_buff **skb, u32 mtu, u32 dnode)
struct sk_buff *skb, u32 mtu, u32 dnode)
{ {
struct sk_buff *bskb; struct sk_buff *bskb;
struct tipc_msg *bmsg; struct tipc_msg *bmsg;
struct tipc_msg *msg = buf_msg(skb); struct tipc_msg *msg = buf_msg(*skb);
u32 msz = msg_size(msg); u32 msz = msg_size(msg);
u32 max = mtu - INT_H_SIZE; u32 max = mtu - INT_H_SIZE;
...@@ -448,9 +450,9 @@ bool tipc_msg_make_bundle(struct sk_buff_head *list, ...@@ -448,9 +450,9 @@ bool tipc_msg_make_bundle(struct sk_buff_head *list,
msg_set_seqno(bmsg, msg_seqno(msg)); msg_set_seqno(bmsg, msg_seqno(msg));
msg_set_ack(bmsg, msg_ack(msg)); msg_set_ack(bmsg, msg_ack(msg));
msg_set_bcast_ack(bmsg, msg_bcast_ack(msg)); msg_set_bcast_ack(bmsg, msg_bcast_ack(msg));
TIPC_SKB_CB(bskb)->bundling = true; tipc_msg_bundle(bskb, *skb, mtu);
__skb_queue_tail(list, bskb); *skb = bskb;
return tipc_msg_bundle(list, skb, mtu); return true;
} }
/** /**
......
...@@ -767,9 +767,9 @@ struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz, ...@@ -767,9 +767,9 @@ struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz,
uint data_sz, u32 dnode, u32 onode, uint data_sz, u32 dnode, u32 onode,
u32 dport, u32 oport, int errcode); u32 dport, u32 oport, int errcode);
int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf); int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf);
bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu); bool tipc_msg_bundle(struct sk_buff *bskb, struct sk_buff *skb, u32 mtu);
bool tipc_msg_make_bundle(struct sk_buff_head *list,
struct sk_buff *skb, u32 mtu, u32 dnode); bool tipc_msg_make_bundle(struct sk_buff **skb, u32 mtu, u32 dnode);
bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos); bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos);
int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m, int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m,
int offset, int dsz, int mtu, struct sk_buff_head *list); int offset, int dsz, int mtu, struct sk_buff_head *list);
......
...@@ -111,7 +111,7 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr) ...@@ -111,7 +111,7 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr)
INIT_LIST_HEAD(&n_ptr->list); INIT_LIST_HEAD(&n_ptr->list);
INIT_LIST_HEAD(&n_ptr->publ_list); INIT_LIST_HEAD(&n_ptr->publ_list);
INIT_LIST_HEAD(&n_ptr->conn_sks); INIT_LIST_HEAD(&n_ptr->conn_sks);
__skb_queue_head_init(&n_ptr->bclink.deferred_queue); __skb_queue_head_init(&n_ptr->bclink.deferdq);
hlist_add_head_rcu(&n_ptr->hash, &tn->node_htable[tipc_hashfn(addr)]); hlist_add_head_rcu(&n_ptr->hash, &tn->node_htable[tipc_hashfn(addr)]);
list_for_each_entry_rcu(temp_node, &tn->node_list, list) { list_for_each_entry_rcu(temp_node, &tn->node_list, list) {
if (n_ptr->addr < temp_node->addr) if (n_ptr->addr < temp_node->addr)
...@@ -354,7 +354,7 @@ static void node_lost_contact(struct tipc_node *n_ptr) ...@@ -354,7 +354,7 @@ static void node_lost_contact(struct tipc_node *n_ptr)
/* Flush broadcast link info associated with lost node */ /* Flush broadcast link info associated with lost node */
if (n_ptr->bclink.recv_permitted) { if (n_ptr->bclink.recv_permitted) {
__skb_queue_purge(&n_ptr->bclink.deferred_queue); __skb_queue_purge(&n_ptr->bclink.deferdq);
if (n_ptr->bclink.reasm_buf) { if (n_ptr->bclink.reasm_buf) {
kfree_skb(n_ptr->bclink.reasm_buf); kfree_skb(n_ptr->bclink.reasm_buf);
......
...@@ -84,7 +84,7 @@ struct tipc_node_bclink { ...@@ -84,7 +84,7 @@ struct tipc_node_bclink {
u32 last_sent; u32 last_sent;
u32 oos_state; u32 oos_state;
u32 deferred_size; u32 deferred_size;
struct sk_buff_head deferred_queue; struct sk_buff_head deferdq;
struct sk_buff *reasm_buf; struct sk_buff *reasm_buf;
int inputq_map; int inputq_map;
bool recv_permitted; bool recv_permitted;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment