Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit aba79f33 authored by Ying Xue's avatar Ying Xue Committed by Paul Gortmaker
Browse files

tipc: byte-based overload control on socket receive queue 



Change overload control to be purely byte-based, using
sk->sk_rmem_alloc as byte counter, and compare it to a calculated
upper limit for the socket receive queue.

For all connection messages, irrespective of message importance,
the overload limit is set to a constant value (i.e, 67MB). This
limit should normally never be reached because of the lower
limit used by the flow control algorithm, and is there only
as a last resort in case a faulty peer doesn't respect the send
window limit.

For datagram messages, message importance is taken into account
when calculating the overload limit. The calculation is based
on sk->sk_rcvbuf, and is hence configurable via the socket option
SO_RCVBUF.

Cc: Neil Horman <nhorman@tuxdriver.com>
Signed-off-by: default avatarYing Xue <ying.xue@windriver.com>
Signed-off-by: default avatarJon Maloy <jon.maloy@ericsson.com>
Signed-off-by: default avatarPaul Gortmaker <paul.gortmaker@windriver.com>
parent 57467e56

net/tipc/socket.c

+39 −38
Original line number Diff line number Diff line
@@ -43,7 +43,8 @@ 
#define SS_LISTENING	-1	/* socket is listening */

#define SS_READY	-2	/* socket is connectionless */



#define OVERLOAD_LIMIT_BASE	10000

#define CONN_OVERLOAD_LIMIT	((TIPC_FLOW_CONTROL_WIN * 2 + 1) * \

				SKB_TRUESIZE(TIPC_MAX_USER_MSG_SIZE))

#define CONN_TIMEOUT_DEFAULT	8000	/* default connect timeout = 8s */



struct tipc_sock {
@@ -202,7 +203,6 @@ static int tipc_create(struct net *net, struct socket *sock, int protocol, 


	sock_init_data(sock, sk);

	sk->sk_backlog_rcv = backlog_rcv;

	sk->sk_rcvbuf = TIPC_FLOW_CONTROL_WIN * 2 * TIPC_MAX_USER_MSG_SIZE * 2;

	sk->sk_data_ready = tipc_data_ready;

	sk->sk_write_space = tipc_write_space;

	tipc_sk(sk)->p = tp_ptr;
@@ -1141,34 +1141,6 @@ static void tipc_data_ready(struct sock *sk, int len) 
	rcu_read_unlock();

}



/**

 * rx_queue_full - determine if receive queue can accept another message

 * @msg: message to be added to queue

 * @queue_size: current size of queue

 * @base: nominal maximum size of queue

 *

 * Returns 1 if queue is unable to accept message, 0 otherwise

 */

static int rx_queue_full(struct tipc_msg *msg, u32 queue_size, u32 base)

{

	u32 threshold;

	u32 imp = msg_importance(msg);



	if (imp == TIPC_LOW_IMPORTANCE)

		threshold = base;

	else if (imp == TIPC_MEDIUM_IMPORTANCE)

		threshold = base * 2;

	else if (imp == TIPC_HIGH_IMPORTANCE)

		threshold = base * 100;

	else

		return 0;



	if (msg_connected(msg))

		threshold *= 4;



	return queue_size >= threshold;

}



/**

 * filter_connect - Handle all incoming messages for a connection-based socket

 * @tsock: TIPC socket
@@ -1246,6 +1218,36 @@ static u32 filter_connect(struct tipc_sock *tsock, struct sk_buff **buf) 
	return retval;

}



/**

 * rcvbuf_limit - get proper overload limit of socket receive queue

 * @sk: socket

 * @buf: message

 *

 * For all connection oriented messages, irrespective of importance,

 * the default overload value (i.e. 67MB) is set as limit.

 *

 * For all connectionless messages, by default new queue limits are

 * as belows:

 *

 * TIPC_LOW_IMPORTANCE       (5MB)

 * TIPC_MEDIUM_IMPORTANCE    (10MB)

 * TIPC_HIGH_IMPORTANCE      (20MB)

 * TIPC_CRITICAL_IMPORTANCE  (40MB)

 *

 * Returns overload limit according to corresponding message importance

 */

static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *buf)

{

	struct tipc_msg *msg = buf_msg(buf);

	unsigned int limit;



	if (msg_connected(msg))

		limit = CONN_OVERLOAD_LIMIT;

	else

		limit = sk->sk_rcvbuf << (msg_importance(msg) + 5);

	return limit;

}



/**

 * filter_rcv - validate incoming message

 * @sk: socket
@@ -1262,7 +1264,7 @@ static u32 filter_rcv(struct sock *sk, struct sk_buff *buf) 
{

	struct socket *sock = sk->sk_socket;

	struct tipc_msg *msg = buf_msg(buf);

	u32 recv_q_len;

	unsigned int limit = rcvbuf_limit(sk, buf);

	u32 res = TIPC_OK;



	/* Reject message if it is wrong sort of message for socket */
@@ -1279,15 +1281,13 @@ static u32 filter_rcv(struct sock *sk, struct sk_buff *buf) 
	}



	/* Reject message if there isn't room to queue it */

	recv_q_len = skb_queue_len(&sk->sk_receive_queue);

	if (unlikely(recv_q_len >= (OVERLOAD_LIMIT_BASE / 2))) {

		if (rx_queue_full(msg, recv_q_len, OVERLOAD_LIMIT_BASE / 2))

	if (sk_rmem_alloc_get(sk) + buf->truesize >= limit)

		return TIPC_ERR_OVERLOAD;

	}



	/* Enqueue message (finally!) */

	/* Enqueue message */

	TIPC_SKB_CB(buf)->handle = 0;

	__skb_queue_tail(&sk->sk_receive_queue, buf);

	skb_set_owner_r(buf, sk);



	sk->sk_data_ready(sk, 0);

	return TIPC_OK;
@@ -1336,7 +1336,7 @@ static u32 dispatch(struct tipc_port *tport, struct sk_buff *buf) 
	if (!sock_owned_by_user(sk)) {

		res = filter_rcv(sk, buf);

	} else {

		if (sk_add_backlog(sk, buf, sk->sk_rcvbuf))

		if (sk_add_backlog(sk, buf, rcvbuf_limit(sk, buf)))

			res = TIPC_ERR_OVERLOAD;

		else

			res = TIPC_OK;
@@ -1570,6 +1570,7 @@ static int accept(struct socket *sock, struct socket *new_sock, int flags) 
	} else {

		__skb_dequeue(&sk->sk_receive_queue);

		__skb_queue_head(&new_sk->sk_receive_queue, buf);

		skb_set_owner_r(buf, new_sk);

	}

	release_sock(new_sk);