Commit d39bbd44 authored by Jon Paul Maloy's avatar Jon Paul Maloy Committed by David S. Miller
Browse files

tipc: move link input queue to tipc_node 



At present, the link input queue and the name distributor receive
queues are fields aggregated in struct tipc_link. This is a hazard,
because a link might be deleted while a receiving socket still keeps
reference to one of the queues.

This commit fixes this bug. However, rather than adding yet another
reference counter to the critical data path, we move the two queues
to safe ground inside struct tipc_node, which is already protected, and
let the link code only handle references to the queues. This is also
in line with planned later changes in this area.

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 d3a43b90

net/tipc/link.c

+15 −12
Original line number Original line Diff line number Diff line
@@ -227,7 +227,9 @@ static void link_set_timer(struct tipc_link *link, unsigned long time) 
 */

 */

struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,

struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,

				   struct tipc_bearer *b_ptr,

				   struct tipc_bearer *b_ptr,

				   const struct tipc_media_addr *media_addr)

				   const struct tipc_media_addr *media_addr,

				   struct sk_buff_head *inputq,

				   struct sk_buff_head *namedq)

{

{

	struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);

	struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);

	struct tipc_link *l_ptr;

	struct tipc_link *l_ptr;
@@ -289,8 +291,9 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr, 
	__skb_queue_head_init(&l_ptr->backlogq);

	__skb_queue_head_init(&l_ptr->backlogq);

	__skb_queue_head_init(&l_ptr->deferdq);

	__skb_queue_head_init(&l_ptr->deferdq);

	skb_queue_head_init(&l_ptr->wakeupq);

	skb_queue_head_init(&l_ptr->wakeupq);

	skb_queue_head_init(&l_ptr->inputq);

	l_ptr->inputq = inputq;

	skb_queue_head_init(&l_ptr->namedq);

	l_ptr->namedq = namedq;

	skb_queue_head_init(l_ptr->inputq);

	link_reset_statistics(l_ptr);

	link_reset_statistics(l_ptr);

	tipc_node_attach_link(n_ptr, l_ptr);

	tipc_node_attach_link(n_ptr, l_ptr);

	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);

	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
@@ -391,8 +394,8 @@ void link_prepare_wakeup(struct tipc_link *l) 
		if ((pnd[imp] + l->backlog[imp].len) >= lim)

		if ((pnd[imp] + l->backlog[imp].len) >= lim)

			break;

			break;

		skb_unlink(skb, &l->wakeupq);

		skb_unlink(skb, &l->wakeupq);

		skb_queue_tail(&l->inputq, skb);

		skb_queue_tail(l->inputq, skb);

		l->owner->inputq = &l->inputq;

		l->owner->inputq = l->inputq;

		l->owner->action_flags |= TIPC_MSG_EVT;

		l->owner->action_flags |= TIPC_MSG_EVT;

	}

	}

}

}
@@ -465,7 +468,7 @@ void tipc_link_reset(struct tipc_link *l_ptr) 
	__skb_queue_purge(&l_ptr->transmq);

	__skb_queue_purge(&l_ptr->transmq);

	__skb_queue_purge(&l_ptr->deferdq);

	__skb_queue_purge(&l_ptr->deferdq);

	if (!owner->inputq)

	if (!owner->inputq)

		owner->inputq = &l_ptr->inputq;

		owner->inputq = l_ptr->inputq;

	skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);

	skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);

	if (!skb_queue_empty(owner->inputq))

	if (!skb_queue_empty(owner->inputq))

		owner->action_flags |= TIPC_MSG_EVT;

		owner->action_flags |= TIPC_MSG_EVT;
@@ -962,7 +965,7 @@ static bool link_synch(struct tipc_link *l) 




	/* Is it still in the input queue ? */

	/* Is it still in the input queue ? */

	post_synch = mod(pl->rcv_nxt - l->synch_point) - 1;

	post_synch = mod(pl->rcv_nxt - l->synch_point) - 1;

	if (skb_queue_len(&pl->inputq) > post_synch)

	if (skb_queue_len(pl->inputq) > post_synch)

		return false;

		return false;

synched:

synched:

	l->flags &= ~LINK_SYNCHING;

	l->flags &= ~LINK_SYNCHING;
@@ -1141,16 +1144,16 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb) 
	case TIPC_HIGH_IMPORTANCE:

	case TIPC_HIGH_IMPORTANCE:

	case TIPC_CRITICAL_IMPORTANCE:

	case TIPC_CRITICAL_IMPORTANCE:

	case CONN_MANAGER:

	case CONN_MANAGER:

		if (tipc_skb_queue_tail(&link->inputq, skb, dport)) {

		if (tipc_skb_queue_tail(link->inputq, skb, dport)) {

			node->inputq = &link->inputq;

			node->inputq = link->inputq;

			node->action_flags |= TIPC_MSG_EVT;

			node->action_flags |= TIPC_MSG_EVT;

		}

		}

		return true;

		return true;

	case NAME_DISTRIBUTOR:

	case NAME_DISTRIBUTOR:

		node->bclink.recv_permitted = true;

		node->bclink.recv_permitted = true;

		node->namedq = &link->namedq;

		node->namedq = link->namedq;

		skb_queue_tail(&link->namedq, skb);

		skb_queue_tail(link->namedq, skb);

		if (skb_queue_len(&link->namedq) == 1)

		if (skb_queue_len(link->namedq) == 1)

			node->action_flags |= TIPC_NAMED_MSG_EVT;

			node->action_flags |= TIPC_NAMED_MSG_EVT;

		return true;

		return true;

	case MSG_BUNDLER:

	case MSG_BUNDLER:

net/tipc/link.h

+7 −5
Original line number Original line Diff line number Diff line
@@ -192,8 +192,8 @@ struct tipc_link { 
	u16 rcv_nxt;

	u16 rcv_nxt;

	u32 rcv_unacked;

	u32 rcv_unacked;

	struct sk_buff_head deferdq;

	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_head wakeupq;

	struct sk_buff_head wakeupq;
@@ -207,9 +207,11 @@ struct tipc_link { 




struct tipc_port;

struct tipc_port;





struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,

struct tipc_link *tipc_link_create(struct tipc_node *n,

			      struct tipc_bearer *b_ptr,

				   struct tipc_bearer *b,

			      const struct tipc_media_addr *media_addr);

				   const struct tipc_media_addr *maddr,

				   struct sk_buff_head *inputq,

				   struct sk_buff_head *namedq);

void tipc_link_delete(struct tipc_link *link);

void tipc_link_delete(struct tipc_link *link);

void tipc_link_delete_list(struct net *net, unsigned int bearer_id);

void tipc_link_delete_list(struct net *net, unsigned int bearer_id);

void tipc_link_failover_send_queue(struct tipc_link *l_ptr);

void tipc_link_failover_send_queue(struct tipc_link *l_ptr);

net/tipc/node.c

+3 −1
Original line number Original line Diff line number Diff line
@@ -132,6 +132,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.namedq);

	__skb_queue_head_init(&n_ptr->bclink.deferdq);

	__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) {
@@ -350,9 +351,10 @@ bool tipc_node_update_dest(struct tipc_node *n, struct tipc_bearer *b, 
{

{

	struct tipc_link *l = n->links[b->identity].link;

	struct tipc_link *l = n->links[b->identity].link;

	struct tipc_media_addr *curr = &n->links[b->identity].maddr;

	struct tipc_media_addr *curr = &n->links[b->identity].maddr;

	struct sk_buff_head *inputq = &n->links[b->identity].inputq;





	if (!l)

	if (!l)

		l = tipc_link_create(n, b, maddr);

		l = tipc_link_create(n, b, maddr, inputq, &n->bclink.namedq);

	if (!l)

	if (!l)

		return false;

		return false;

	memcpy(&l->media_addr, maddr, sizeof(*maddr));

	memcpy(&l->media_addr, maddr, sizeof(*maddr));

net/tipc/node.h

+2 −1
Original line number Original line Diff line number Diff line
@@ -85,13 +85,14 @@ struct tipc_node_bclink { 
	u32 deferred_size;

	u32 deferred_size;

	struct sk_buff_head deferdq;

	struct sk_buff_head deferdq;

	struct sk_buff *reasm_buf;

	struct sk_buff *reasm_buf;

	int inputq_map;

	struct sk_buff_head namedq;

	bool recv_permitted;

	bool recv_permitted;

};

};





struct tipc_link_entry {

struct tipc_link_entry {

	struct tipc_link *link;

	struct tipc_link *link;

	u32 mtu;

	u32 mtu;

	struct sk_buff_head inputq;

	struct tipc_media_addr maddr;

	struct tipc_media_addr maddr;

};

};

CRA Git | Maintained and supported by SUSTech CRA and CCSE