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

tipc: resolve race problem at unicast message reception 



TIPC handles message cardinality and sequencing at the link layer,
before passing messages upwards to the destination sockets. During the
upcall from link to socket no locks are held. It is therefore possible,
and we see it happen occasionally, that messages arriving in different
threads and delivered in sequence still bypass each other before they
reach the destination socket. This must not happen, since it violates
the sequentiality guarantee.

We solve this by adding a new input buffer queue to the link structure.
Arriving messages are added safely to the tail of that queue by the
link, while the head of the queue is consumed, also safely, by the
receiving socket. Sequentiality is secured per socket by only allowing
buffers to be dequeued inside the socket lock. Since there may be multiple
simultaneous readers of the queue, we use a 'filter' parameter to reduce
the risk that they peek the same buffer from the queue, hence also
reducing the risk of contention on the receiving socket locks.

This solves the sequentiality problem, and seems to cause no measurable
performance degradation.

A nice side effect of this change is that lock handling in the functions
tipc_rcv() and tipc_bcast_rcv() now becomes uniform, something that
will enable future simplifications of those functions.

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 94153e36

net/tipc/bcast.c

+11 −9
Original line number Diff line number Diff line
@@ -189,10 +189,8 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to) 
void tipc_bclink_wakeup_users(struct net *net)

{

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

	struct sk_buff *skb;



	while ((skb = skb_dequeue(&tn->bclink->link.waiting_sks)))

		tipc_sk_rcv(net, skb);

	tipc_sk_rcv(net, &tn->bclink->link.wakeupq);

}



/**
@@ -271,9 +269,8 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) 
		tipc_link_push_packets(tn->bcl);

		bclink_set_last_sent(net);

	}

	if (unlikely(released && !skb_queue_empty(&tn->bcl->waiting_sks)))

	if (unlikely(released && !skb_queue_empty(&tn->bcl->wakeupq)))

		n_ptr->action_flags |= TIPC_WAKEUP_BCAST_USERS;



exit:

	tipc_bclink_unlock(net);

}
@@ -450,6 +447,9 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf) 
	u32 next_in;

	u32 seqno;

	int deferred = 0;

	int pos = 0;

	struct sk_buff *iskb;

	struct sk_buff_head msgs;



	/* Screen out unwanted broadcast messages */

	if (msg_mc_netid(msg) != tn->net_id)
@@ -506,7 +506,8 @@ receive: 
			bcl->stats.recv_bundled += msg_msgcnt(msg);

			tipc_bclink_unlock(net);

			tipc_node_unlock(node);

			tipc_link_bundle_rcv(net, buf);

			while (tipc_msg_extract(buf, &iskb, &pos))

				tipc_sk_mcast_rcv(net, iskb);

		} else if (msg_user(msg) == MSG_FRAGMENTER) {

			tipc_buf_append(&node->bclink.reasm_buf, &buf);

			if (unlikely(!buf && !node->bclink.reasm_buf))
@@ -527,7 +528,9 @@ receive: 
			bclink_accept_pkt(node, seqno);

			tipc_bclink_unlock(net);

			tipc_node_unlock(node);

			tipc_named_rcv(net, buf);

			skb_queue_head_init(&msgs);

			skb_queue_tail(&msgs, buf);

			tipc_named_rcv(net, &msgs);

		} else {

			tipc_bclink_lock(net);

			bclink_accept_pkt(node, seqno);
@@ -944,10 +947,9 @@ int tipc_bclink_init(struct net *net) 
	spin_lock_init(&bclink->lock);

	__skb_queue_head_init(&bcl->outqueue);

	__skb_queue_head_init(&bcl->deferred_queue);

	skb_queue_head_init(&bcl->waiting_sks);

	skb_queue_head_init(&bcl->wakeupq);

	bcl->next_out_no = 1;

	spin_lock_init(&bclink->node.lock);

	__skb_queue_head_init(&bclink->node.waiting_sks);

	bcl->owner = &bclink->node;

	bcl->owner->net = net;

	bcl->max_pkt = MAX_PKT_DEFAULT_MCAST;

net/tipc/link.c

+104 −143
Original line number Diff line number Diff line
@@ -113,10 +113,8 @@ static void link_reset_statistics(struct tipc_link *l_ptr); 
static void link_print(struct tipc_link *l_ptr, const char *str);

static void tipc_link_sync_xmit(struct tipc_link *l);

static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);

static int tipc_link_input(struct net *net, struct tipc_link *l,

			   struct sk_buff *buf);

static int tipc_link_prepare_input(struct net *net, struct tipc_link *l,

				   struct sk_buff **buf);

static void tipc_link_input(struct tipc_link *l, struct sk_buff *skb);

static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);



/*

 *  Simple link routines
@@ -318,8 +316,9 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr, 
	l_ptr->next_out_no = 1;

	__skb_queue_head_init(&l_ptr->outqueue);

	__skb_queue_head_init(&l_ptr->deferred_queue);

	skb_queue_head_init(&l_ptr->waiting_sks);



	skb_queue_head_init(&l_ptr->wakeupq);

	skb_queue_head_init(&l_ptr->inputq);

	skb_queue_head_init(&l_ptr->namedq);

	link_reset_statistics(l_ptr);

	tipc_node_attach_link(n_ptr, l_ptr);

	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
@@ -387,7 +386,7 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport, 
		return false;

	TIPC_SKB_CB(buf)->chain_sz = chain_sz;

	TIPC_SKB_CB(buf)->chain_imp = imp;

	skb_queue_tail(&link->waiting_sks, buf);

	skb_queue_tail(&link->wakeupq, buf);

	link->stats.link_congs++;

	return true;

}
@@ -398,17 +397,19 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport, 
 * Move a number of waiting users, as permitted by available space in

 * the send queue, from link wait queue to node wait queue for wakeup

 */

static void link_prepare_wakeup(struct tipc_link *link)

void link_prepare_wakeup(struct tipc_link *link)

{

	uint pend_qsz = skb_queue_len(&link->outqueue);

	struct sk_buff *skb, *tmp;



	skb_queue_walk_safe(&link->waiting_sks, skb, tmp) {

	skb_queue_walk_safe(&link->wakeupq, skb, tmp) {

		if (pend_qsz >= link->queue_limit[TIPC_SKB_CB(skb)->chain_imp])

			break;

		pend_qsz += TIPC_SKB_CB(skb)->chain_sz;

		skb_unlink(skb, &link->waiting_sks);

		skb_queue_tail(&link->owner->waiting_sks, skb);

		skb_unlink(skb, &link->wakeupq);

		skb_queue_tail(&link->inputq, skb);

		link->owner->inputq = &link->inputq;

		link->owner->action_flags |= TIPC_MSG_EVT;

	}

}
@@ -461,13 +462,13 @@ void tipc_link_reset(struct tipc_link *l_ptr) 
		l_ptr->exp_msg_count = START_CHANGEOVER;

	}



	/* Clean up all queues: */

	/* Clean up all queues, except inputq: */

	__skb_queue_purge(&l_ptr->outqueue);

	__skb_queue_purge(&l_ptr->deferred_queue);

	if (!skb_queue_empty(&l_ptr->waiting_sks)) {

		skb_queue_splice_init(&l_ptr->waiting_sks, &owner->waiting_sks);

		owner->action_flags |= TIPC_WAKEUP_USERS;

	}

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

	if (!skb_queue_empty(&l_ptr->inputq))

		owner->action_flags |= TIPC_MSG_EVT;

	owner->inputq = &l_ptr->inputq;

	l_ptr->next_out = NULL;

	l_ptr->unacked_window = 0;

	l_ptr->checkpoint = 1;
@@ -795,7 +796,7 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link, 


static void skb2list(struct sk_buff *skb, struct sk_buff_head *list)

{

	__skb_queue_head_init(list);

	skb_queue_head_init(list);

	__skb_queue_tail(list, skb);

}
@@ -841,19 +842,13 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode, 
			rc = __tipc_link_xmit(net, link, list);

		tipc_node_unlock(node);

	}



	if (link)

		return rc;



	if (likely(in_own_node(net, 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(net, __skb_dequeue(list));

	}

	__skb_queue_purge(list);

	if (likely(in_own_node(net, dnode)))

		return tipc_sk_rcv(net, list);



	__skb_queue_purge(list);

	return rc;

}
@@ -1162,7 +1157,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) 
		/* Locate unicast link endpoint that should handle message */

		l_ptr = n_ptr->links[b_ptr->identity];

		if (unlikely(!l_ptr))

			goto unlock_discard;

			goto unlock;



		/* Verify that communication with node is currently allowed */

		if ((n_ptr->action_flags & TIPC_WAIT_PEER_LINKS_DOWN) &&
@@ -1173,7 +1168,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) 
			n_ptr->action_flags &= ~TIPC_WAIT_PEER_LINKS_DOWN;



		if (tipc_node_blocked(n_ptr))

			goto unlock_discard;

			goto unlock;



		/* Validate message sequence number info */

		seq_no = msg_seqno(msg);
@@ -1197,18 +1192,16 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) 
		if (unlikely(l_ptr->next_out))

			tipc_link_push_packets(l_ptr);



		if (released && !skb_queue_empty(&l_ptr->waiting_sks)) {

		if (released && !skb_queue_empty(&l_ptr->wakeupq))

			link_prepare_wakeup(l_ptr);

			l_ptr->owner->action_flags |= TIPC_WAKEUP_USERS;

		}



		/* Process the incoming packet */

		if (unlikely(!link_working_working(l_ptr))) {

			if (msg_user(msg) == LINK_PROTOCOL) {

				tipc_link_proto_rcv(l_ptr, skb);

				link_retrieve_defq(l_ptr, &head);

				tipc_node_unlock(n_ptr);

				continue;

				skb = NULL;

				goto unlock;

			}



			/* Traffic message. Conditionally activate link */
@@ -1217,18 +1210,18 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) 
			if (link_working_working(l_ptr)) {

				/* Re-insert buffer in front of queue */

				__skb_queue_head(&head, skb);

				tipc_node_unlock(n_ptr);

				continue;

				skb = NULL;

				goto unlock;

			}

			goto unlock_discard;

			goto unlock;

		}



		/* Link is now in state WORKING_WORKING */

		if (unlikely(seq_no != mod(l_ptr->next_in_no))) {

			link_handle_out_of_seq_msg(l_ptr, skb);

			link_retrieve_defq(l_ptr, &head);

			tipc_node_unlock(n_ptr);

			continue;

			skb = NULL;

			goto unlock;

		}

		l_ptr->next_in_no++;

		if (unlikely(!skb_queue_empty(&l_ptr->deferred_queue)))
@@ -1238,97 +1231,102 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) 
			l_ptr->stats.sent_acks++;

			tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);

		}



		if (tipc_link_prepare_input(net, l_ptr, &skb)) {

			tipc_node_unlock(n_ptr);

			continue;

		}

		tipc_node_unlock(n_ptr);



		if (tipc_link_input(net, l_ptr, skb) != 0)

			goto discard;

		continue;

unlock_discard:

		tipc_link_input(l_ptr, skb);

		skb = NULL;

unlock:

		tipc_node_unlock(n_ptr);

discard:

		if (unlikely(skb))

			kfree_skb(skb);

	}

}



/**

 * tipc_link_prepare_input - process TIPC link messages

 *

 * returns nonzero if the message was consumed

/* tipc_data_input - deliver data and name distr msgs to upper layer

 *

 * Consumes buffer if message is of right type

 * Node lock must be held

 */

static int tipc_link_prepare_input(struct net *net, struct tipc_link *l,

				   struct sk_buff **buf)

static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)

{

	struct tipc_node *n;

	struct tipc_msg *msg;

	int res = -EINVAL;

	struct tipc_node *node = link->owner;

	struct tipc_msg *msg = buf_msg(skb);

	u32 dport = msg_destport(msg);



	n = l->owner;

	msg = buf_msg(*buf);

	switch (msg_user(msg)) {

	case CHANGEOVER_PROTOCOL:

		if (tipc_link_tunnel_rcv(n, buf))

			res = 0;

		break;

	case MSG_FRAGMENTER:

		l->stats.recv_fragments++;

		if (tipc_buf_append(&l->reasm_buf, buf)) {

			l->stats.recv_fragmented++;

			res = 0;

		} else if (!l->reasm_buf) {

			tipc_link_reset(l);

	case TIPC_LOW_IMPORTANCE:

	case TIPC_MEDIUM_IMPORTANCE:

	case TIPC_HIGH_IMPORTANCE:

	case TIPC_CRITICAL_IMPORTANCE:

	case CONN_MANAGER:

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

			node->inputq = &link->inputq;

			node->action_flags |= TIPC_MSG_EVT;

		}

		break;

	case MSG_BUNDLER:

		l->stats.recv_bundles++;

		l->stats.recv_bundled += msg_msgcnt(msg);

		res = 0;

		break;

		return true;

	case NAME_DISTRIBUTOR:

		n->bclink.recv_permitted = true;

		res = 0;

		break;

		node->bclink.recv_permitted = true;

		node->namedq = &link->namedq;

		skb_queue_tail(&link->namedq, skb);

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

			node->action_flags |= TIPC_NAMED_MSG_EVT;

		return true;

	case MSG_BUNDLER:

	case CHANGEOVER_PROTOCOL:

	case MSG_FRAGMENTER:

	case BCAST_PROTOCOL:

		tipc_link_sync_rcv(n, *buf);

		break;

		return false;

	default:

		res = 0;

	}

	return res;

		pr_warn("Dropping received illegal msg type\n");

		kfree_skb(skb);

		return false;

	};

}

/**

 * tipc_link_input - Deliver message too higher layers



/* tipc_link_input - process packet that has passed link protocol check

 *

 * Consumes buffer

 * Node lock must be held

 */

static int tipc_link_input(struct net *net, struct tipc_link *l,

			   struct sk_buff *buf)

static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)

{

	struct tipc_msg *msg = buf_msg(buf);

	int res = 0;

	struct tipc_node *node = link->owner;

	struct tipc_msg *msg = buf_msg(skb);

	struct sk_buff *iskb;

	int pos = 0;



	if (likely(tipc_data_input(link, skb)))

		return;



	switch (msg_user(msg)) {

	case TIPC_LOW_IMPORTANCE:

	case TIPC_MEDIUM_IMPORTANCE:

	case TIPC_HIGH_IMPORTANCE:

	case TIPC_CRITICAL_IMPORTANCE:

	case CONN_MANAGER:

		tipc_sk_rcv(net, buf);

	case CHANGEOVER_PROTOCOL:

		if (!tipc_link_tunnel_rcv(node, &skb))

			break;

	case NAME_DISTRIBUTOR:

		tipc_named_rcv(net, buf);

		if (msg_user(buf_msg(skb)) != MSG_BUNDLER) {

			tipc_data_input(link, skb);

			break;

		}

	case MSG_BUNDLER:

		tipc_link_bundle_rcv(net, buf);

		link->stats.recv_bundles++;

		link->stats.recv_bundled += msg_msgcnt(msg);



		while (tipc_msg_extract(skb, &iskb, &pos))

			tipc_data_input(link, iskb);

		break;

	default:

		res = -EINVAL;

	case MSG_FRAGMENTER:

		link->stats.recv_fragments++;

		if (tipc_buf_append(&link->reasm_buf, &skb)) {

			link->stats.recv_fragmented++;

			tipc_data_input(link, skb);

		} else if (!link->reasm_buf) {

			tipc_link_reset(link);

		}

	return res;

		break;

	case BCAST_PROTOCOL:

		tipc_link_sync_rcv(node, skb);

		break;

	default:

		break;

	};

}



/**
@@ -1779,7 +1777,7 @@ void tipc_link_dup_queue_xmit(struct tipc_link *l_ptr, 
 * @from_pos: offset to extract from

 *

 * Returns a new message buffer containing an embedded message.  The

 * encapsulating message itself is left unchanged.

 * encapsulating buffer is left unchanged.

 */

static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos)

{
@@ -1793,8 +1791,6 @@ static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos) 
	return eb;

}







/* tipc_link_dup_rcv(): Receive a tunnelled DUPLICATE_MSG packet.

 * Owner node is locked.

 */
@@ -1893,41 +1889,6 @@ exit: 
	return *buf != NULL;

}



/*

 *  Bundler functionality:

 */

void tipc_link_bundle_rcv(struct net *net, struct sk_buff *buf)

{

	u32 msgcount = msg_msgcnt(buf_msg(buf));

	u32 pos = INT_H_SIZE;

	struct sk_buff *obuf;

	struct tipc_msg *omsg;



	while (msgcount--) {

		obuf = buf_extract(buf, pos);

		if (obuf == NULL) {

			pr_warn("Link unable to unbundle message(s)\n");

			break;

		}

		omsg = buf_msg(obuf);

		pos += align(msg_size(omsg));

		if (msg_isdata(omsg)) {

			if (unlikely(msg_type(omsg) == TIPC_MCAST_MSG))

				tipc_sk_mcast_rcv(net, obuf);

			else

				tipc_sk_rcv(net, obuf);

		} else if (msg_user(omsg) == CONN_MANAGER) {

			tipc_sk_rcv(net, obuf);

		} else if (msg_user(omsg) == NAME_DISTRIBUTOR) {

			tipc_named_rcv(net, obuf);

		} else {

			pr_warn("Illegal bundled msg: %u\n", msg_user(omsg));

			kfree_skb(obuf);

		}

	}

	kfree_skb(buf);

}



static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)

{

	unsigned long intv = ((tol / 4) > 500) ? 500 : tol / 4;

net/tipc/link.h

+7 −3
Original line number Diff line number Diff line
@@ -131,8 +131,10 @@ struct tipc_stats { 
 * @next_in_no: next sequence number to expect for inbound messages

 * @deferred_queue: deferred queue saved OOS b'cast message received from node

 * @unacked_window: # of inbound messages rx'd without ack'ing back to peer

 * @inputq: buffer queue for messages to be delivered upwards

 * @namedq: buffer queue for name table messages to be delivered upwards

 * @next_out: ptr to first unsent outbound message in queue

 * @waiting_sks: linked list of sockets waiting for link congestion to abate

 * @wakeupq: linked list of wakeup msgs waiting for link congestion to abate

 * @long_msg_seq_no: next identifier to use for outbound fragmented messages

 * @reasm_buf: head of partially reassembled inbound message fragments

 * @stats: collects statistics regarding link activity
@@ -184,10 +186,12 @@ struct tipc_link { 
	u32 next_in_no;

	struct sk_buff_head deferred_queue;

	u32 unacked_window;

	struct sk_buff_head inputq;

	struct sk_buff_head namedq;



	/* Congestion handling */

	struct sk_buff *next_out;

	struct sk_buff_head waiting_sks;

	struct sk_buff_head wakeupq;



	/* Fragmentation/reassembly */

	u32 long_msg_seq_no;
@@ -228,7 +232,6 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dest, 
		   u32 selector);

int __tipc_link_xmit(struct net *net, struct tipc_link *link,

		     struct sk_buff_head *list);

void tipc_link_bundle_rcv(struct net *net, struct sk_buff *buf);

void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int prob,

			  u32 gap, u32 tolerance, u32 priority, u32 acked_mtu);

void tipc_link_push_packets(struct tipc_link *l_ptr);
@@ -244,6 +247,7 @@ int tipc_nl_link_get(struct sk_buff *skb, struct genl_info *info); 
int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info);

int tipc_nl_link_reset_stats(struct sk_buff *skb, struct genl_info *info);

int tipc_nl_parse_link_prop(struct nlattr *prop, struct nlattr *props[]);

void link_prepare_wakeup(struct tipc_link *l);



/*

 * Link sequence number manipulation routines (uses modulo 2**16 arithmetic)

net/tipc/msg.c

+34 −0
Original line number Diff line number Diff line
@@ -326,6 +326,40 @@ bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu) 
	return true;

}



/**

 *  tipc_msg_extract(): extract bundled inner packet from buffer

 *  @skb: linear outer buffer, to be extracted from.

 *  @iskb: extracted inner buffer, to be returned

 *  @pos: position of msg to be extracted. Returns with pointer of next msg

 *  Consumes outer buffer when last packet extracted

 *  Returns true when when there is an extracted buffer, otherwise false

 */

bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos)

{

	struct tipc_msg *msg = buf_msg(skb);

	int imsz;

	struct tipc_msg *imsg = (struct tipc_msg *)(msg_data(msg) + *pos);



	/* Is there space left for shortest possible message? */

	if (*pos > (msg_data_sz(msg) - SHORT_H_SIZE))

		goto none;

	imsz = msg_size(imsg);



	/* Is there space left for current message ? */

	if ((*pos + imsz) > msg_data_sz(msg))

		goto none;

	*iskb = tipc_buf_acquire(imsz);

	if (!*iskb)

		goto none;

	skb_copy_to_linear_data(*iskb, imsg, imsz);

	*pos += align(imsz);

	return true;

none:

	kfree_skb(skb);

	*iskb = NULL;

	return false;

}



/**

 * tipc_msg_make_bundle(): Create bundle buf and append message to its tail

 * @list: the buffer chain

net/tipc/msg.h

+73 −0
Original line number Diff line number Diff line
@@ -45,6 +45,7 @@ 
 * Note: Some items are also used with TIPC internal message headers

 */

#define TIPC_VERSION              2

struct plist;



/*

 * Payload message users are defined in TIPC's public API:
@@ -759,10 +760,82 @@ 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_make_bundle(struct sk_buff_head *list,

			  struct sk_buff *skb, u32 mtu, u32 dnode);

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 offset, int dsz, int mtu, struct sk_buff_head *list);

bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, u32 *dnode,

			  int *err);

struct sk_buff *tipc_msg_reassemble(struct sk_buff_head *list);



/* tipc_skb_peek_port(): find a destination port, ignoring all destinations

 *                       up to and including 'filter'.

 * Note: ignoring previously tried destinations minimizes the risk of

 *       contention on the socket lock

 * @list: list to be peeked in

 * @filter: last destination to be ignored from search

 * Returns a destination port number, of applicable.

 */

static inline u32 tipc_skb_peek_port(struct sk_buff_head *list, u32 filter)

{

	struct sk_buff *skb;

	u32 dport = 0;

	bool ignore = true;



	spin_lock_bh(&list->lock);

	skb_queue_walk(list, skb) {

		dport = msg_destport(buf_msg(skb));

		if (!filter || skb_queue_is_last(list, skb))

			break;

		if (dport == filter)

			ignore = false;

		else if (!ignore)

			break;

	}

	spin_unlock_bh(&list->lock);

	return dport;

}



/* tipc_skb_dequeue(): unlink first buffer with dest 'dport' from list

 * @list: list to be unlinked from

 * @dport: selection criteria for buffer to unlink

 */

static inline struct sk_buff *tipc_skb_dequeue(struct sk_buff_head *list,

					       u32 dport)

{

	struct sk_buff *_skb, *tmp, *skb = NULL;



	spin_lock_bh(&list->lock);

	skb_queue_walk_safe(list, _skb, tmp) {

		if (msg_destport(buf_msg(_skb)) == dport) {

			__skb_unlink(_skb, list);

			skb = _skb;

			break;

		}

	}

	spin_unlock_bh(&list->lock);

	return skb;

}



/* tipc_skb_queue_tail(): add buffer to tail of list;

 * @list: list to be appended to

 * @skb: buffer to append. Always appended

 * @dport: the destination port of the buffer

 * returns true if dport differs from previous destination

 */

static inline bool tipc_skb_queue_tail(struct sk_buff_head *list,

				       struct sk_buff *skb, u32 dport)

{

	struct sk_buff *_skb = NULL;

	bool rv = false;



	spin_lock_bh(&list->lock);

	_skb = skb_peek_tail(list);

	if (!_skb || (msg_destport(buf_msg(_skb)) != dport) ||

	    (skb_queue_len(list) > 32))

		rv = true;

	__skb_queue_tail(list, skb);

	spin_unlock_bh(&list->lock);

	return rv;

}



#endif

CRA Git | Maintained and supported by SUSTech CRA and CCSE