Merge branch 'tipc-next' (6e9feb33) · Commits · 戴 / test

net/tipc/msg.c

+48 −30

Original line number	Diff line number	Diff line
		@@ -499,54 +499,56 @@ bool tipc_msg_make_bundle(struct sk_buff *skb, struct tipc_msg msg,
		/**
		* tipc_msg_reverse(): swap source and destination addresses and add error code
		* @own_node: originating node id for reversed message
		* @skb: buffer containing message to be reversed; may be replaced.
		* @skb: buffer containing message to be reversed; will be consumed
		* @err: error code to be set in message, if any
		* Consumes buffer at failure
		* Replaces consumed buffer with new one when successful
		* Returns true if success, otherwise false
		*/
		bool tipc_msg_reverse(u32 own_node, struct sk_buff **skb, int err)
		{
		struct sk_buff _skb = skb;
		struct tipc_msg *hdr;
		struct tipc_msg ohdr;
		int dlen;
		struct tipc_msg _hdr, hdr;
		int hlen, dlen;

		if (skb_linearize(_skb))
		goto exit;
		hdr = buf_msg(_skb);
		dlen = min_t(uint, msg_data_sz(hdr), MAX_FORWARD_SIZE);
		if (msg_dest_droppable(hdr))
		_hdr = buf_msg(_skb);
		dlen = min_t(uint, msg_data_sz(_hdr), MAX_FORWARD_SIZE);
		hlen = msg_hdr_sz(_hdr);

		if (msg_dest_droppable(_hdr))
		goto exit;
		if (msg_errcode(hdr))
		if (msg_errcode(_hdr))
		goto exit;

		/* Take a copy of original header before altering message */
		memcpy(&ohdr, hdr, msg_hdr_sz(hdr));
		/* Never return SHORT header */
		if (hlen == SHORT_H_SIZE)
		hlen = BASIC_H_SIZE;

		/* Don't return data along with SYN+, - sender has a clone */
		if (msg_is_syn(_hdr) && err == TIPC_ERR_OVERLOAD)
		dlen = 0;

		/* Never return SHORT header; expand by replacing buffer if necessary */
		if (msg_short(hdr)) {
		*skb = tipc_buf_acquire(BASIC_H_SIZE + dlen, GFP_ATOMIC);
		/* Allocate new buffer to return */
		*skb = tipc_buf_acquire(hlen + dlen, GFP_ATOMIC);
		if (!*skb)
		goto exit;
		memcpy((*skb)->data + BASIC_H_SIZE, msg_data(hdr), dlen);
		kfree_skb(_skb);
		_skb = *skb;
		hdr = buf_msg(_skb);
		memcpy(hdr, &ohdr, BASIC_H_SIZE);
		msg_set_hdr_sz(hdr, BASIC_H_SIZE);
		}
		memcpy((*skb)->data, _skb->data, msg_hdr_sz(_hdr));
		memcpy((*skb)->data + hlen, msg_data(_hdr), dlen);

		/* Now reverse the concerned fields */
		/* Build reverse header in new buffer */
		hdr = buf_msg(*skb);
		msg_set_hdr_sz(hdr, hlen);
		msg_set_errcode(hdr, err);
		msg_set_non_seq(hdr, 0);
		msg_set_origport(hdr, msg_destport(&ohdr));
		msg_set_destport(hdr, msg_origport(&ohdr));
		msg_set_destnode(hdr, msg_prevnode(&ohdr));
		msg_set_origport(hdr, msg_destport(_hdr));
		msg_set_destport(hdr, msg_origport(_hdr));
		msg_set_destnode(hdr, msg_prevnode(_hdr));
		msg_set_prevnode(hdr, own_node);
		msg_set_orignode(hdr, own_node);
		msg_set_size(hdr, msg_hdr_sz(hdr) + dlen);
		skb_trim(_skb, msg_size(hdr));
		msg_set_size(hdr, hlen + dlen);
		skb_orphan(_skb);
		kfree_skb(_skb);
		return true;
		exit:
		kfree_skb(_skb);
		@@ -554,6 +556,22 @@ exit:
		return false;
		}

		bool tipc_msg_skb_clone(struct sk_buff_head msg, struct sk_buff_head cpy)
		{
		struct sk_buff skb, _skb;

		skb_queue_walk(msg, skb) {
		_skb = skb_clone(skb, GFP_ATOMIC);
		if (!_skb) {
		__skb_queue_purge(cpy);
		pr_err_ratelimited("Failed to clone buffer chain\n");
		return false;
		}
		__skb_queue_tail(cpy, _skb);
		}
		return true;
		}

		/**
		* tipc_msg_lookup_dest(): try to find new destination for named message
		* @skb: the buffer containing the message.

net/tipc/msg.h

+11 −0

Original line number	Diff line number	Diff line
		@@ -216,6 +216,16 @@ static inline void msg_set_non_seq(struct tipc_msg *m, u32 n)
		msg_set_bits(m, 0, 20, 1, n);
		}

		static inline int msg_is_syn(struct tipc_msg *m)
		{
		return msg_bits(m, 0, 17, 1);
		}

		static inline void msg_set_syn(struct tipc_msg *m, u32 d)
		{
		msg_set_bits(m, 0, 17, 1, d);
		}

		static inline int msg_dest_droppable(struct tipc_msg *m)
		{
		return msg_bits(m, 0, 19, 1);
		@@ -970,6 +980,7 @@ bool tipc_msg_pskb_copy(u32 dst, struct sk_buff_head *msg,
		struct sk_buff_head *cpy);
		void __tipc_skb_queue_sorted(struct sk_buff_head *list, u16 seqno,
		struct sk_buff *skb);
		bool tipc_msg_skb_clone(struct sk_buff_head msg, struct sk_buff_head cpy);

		static inline u16 buf_seqno(struct sk_buff *skb)
		{

net/tipc/node.h

+7 −5

Original line number	Diff line number	Diff line
		@@ -45,6 +45,7 @@
		/* Optional capabilities supported by this code version
		*/
		enum {
		TIPC_SYN_BIT = (1),
		TIPC_BCAST_SYNCH = (1 << 1),
		TIPC_BCAST_STATE_NACK = (1 << 2),
		TIPC_BLOCK_FLOWCTL = (1 << 3),
		@@ -53,7 +54,8 @@ enum {
		TIPC_LINK_PROTO_SEQNO = (1 << 6)
		};

		#define TIPC_NODE_CAPABILITIES (TIPC_BCAST_SYNCH \| \
		#define TIPC_NODE_CAPABILITIES (TIPC_SYN_BIT \| \
		TIPC_BCAST_SYNCH \| \
		TIPC_BCAST_STATE_NACK \| \
		TIPC_BCAST_RCAST \| \
		TIPC_BLOCK_FLOWCTL \| \

net/tipc/socket.c

+124 −83

Original line number	Diff line number	Diff line
		@@ -80,7 +80,6 @@ struct sockaddr_pair {
		* @publications: list of publications for port
		* @blocking_link: address of the congested link we are currently sleeping on
		* @pub_count: total # of publications port has made during its lifetime
		* @probing_state:
		* @conn_timeout: the time we can wait for an unresponded setup request
		* @dupl_rcvcnt: number of bytes counted twice, in both backlog and rcv queue
		* @cong_link_cnt: number of congested links
		@@ -102,8 +101,8 @@ struct tipc_sock {
		struct list_head cong_links;
		struct list_head publications;
		u32 pub_count;
		uint conn_timeout;
		atomic_t dupl_rcvcnt;
		u16 conn_timeout;
		bool probe_unacked;
		u16 cong_link_cnt;
		u16 snt_unacked;
		@@ -507,6 +506,9 @@ static void __tipc_shutdown(struct socket *sock, int error)
		tipc_wait_for_cond(sock, &timeout, (!tsk->cong_link_cnt &&
		!tsk_conn_cong(tsk)));

		/* Remove any pending SYN message */
		__skb_queue_purge(&sk->sk_write_queue);

		/* Reject all unreceived messages, except on an active connection
		* (which disconnects locally & sends a 'FIN+' to peer).
		*/
		@@ -1319,6 +1321,7 @@ static int __tipc_sendmsg(struct socket sock, struct msghdr m, size_t dlen)
		tsk->conn_type = dest->addr.name.name.type;
		tsk->conn_instance = dest->addr.name.name.instance;
		}
		msg_set_syn(hdr, 1);
		}

		seq = &dest->addr.nameseq;
		@@ -1361,6 +1364,8 @@ static int __tipc_sendmsg(struct socket sock, struct msghdr m, size_t dlen)
		rc = tipc_msg_build(hdr, m, 0, dlen, mtu, &pkts);
		if (unlikely(rc != dlen))
		return rc;
		if (unlikely(syn && !tipc_msg_skb_clone(&pkts, &sk->sk_write_queue)))
		return -ENOMEM;

		rc = tipc_node_xmit(net, &pkts, dnode, tsk->portid);
		if (unlikely(rc == -ELINKCONG)) {
		@@ -1478,6 +1483,7 @@ static void tipc_sk_finish_conn(struct tipc_sock *tsk, u32 peer_port,
		struct net *net = sock_net(sk);
		struct tipc_msg *msg = &tsk->phdr;

		msg_set_syn(msg, 0);
		msg_set_destnode(msg, peer_node);
		msg_set_destport(msg, peer_port);
		msg_set_type(msg, TIPC_CONN_MSG);
		@@ -1489,6 +1495,7 @@ static void tipc_sk_finish_conn(struct tipc_sock *tsk, u32 peer_port,
		tipc_node_add_conn(net, peer_node, tsk->portid, peer_port);
		tsk->max_pkt = tipc_node_get_mtu(net, peer_node, tsk->portid);
		tsk->peer_caps = tipc_node_get_capabilities(net, peer_node);
		__skb_queue_purge(&sk->sk_write_queue);
		if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL)
		return;

		@@ -1959,91 +1966,90 @@ static void tipc_sk_proto_rcv(struct sock *sk,
		}

		/**
		* tipc_filter_connect - Handle incoming message for a connection-based socket
		* tipc_sk_filter_connect - check incoming message for a connection-based socket
		* @tsk: TIPC socket
		* @skb: pointer to message buffer. Set to NULL if buffer is consumed
		*
		* Returns true if everything ok, false otherwise
		* @skb: pointer to message buffer.
		* Returns true if message should be added to receive queue, false otherwise
		*/
		static bool tipc_sk_filter_connect(struct tipc_sock tsk, struct sk_buff skb)
		{
		struct sock *sk = &tsk->sk;
		struct net *net = sock_net(sk);
		struct tipc_msg *hdr = buf_msg(skb);
		u32 pport = msg_origport(hdr);
		u32 pnode = msg_orignode(hdr);
		bool con_msg = msg_connected(hdr);
		u32 pport = tsk_peer_port(tsk);
		u32 pnode = tsk_peer_node(tsk);
		u32 oport = msg_origport(hdr);
		u32 onode = msg_orignode(hdr);
		int err = msg_errcode(hdr);
		unsigned long delay;

		if (unlikely(msg_mcast(hdr)))
		return false;

		switch (sk->sk_state) {
		case TIPC_CONNECTING:
		/* Accept only ACK or NACK message */
		if (unlikely(!msg_connected(hdr))) {
		if (pport != tsk_peer_port(tsk) \|\|
		pnode != tsk_peer_node(tsk))
		return false;

		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		sk->sk_err = ECONNREFUSED;
		sk->sk_state_change(sk);
		return true;
		}

		if (unlikely(msg_errcode(hdr))) {
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		sk->sk_err = ECONNREFUSED;
		sk->sk_state_change(sk);
		return true;
		}

		if (unlikely(!msg_isdata(hdr))) {
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		sk->sk_err = EINVAL;
		sk->sk_state_change(sk);
		return true;
		}

		tipc_sk_finish_conn(tsk, msg_origport(hdr), msg_orignode(hdr));
		/* Setup ACK */
		if (likely(con_msg)) {
		if (err)
		break;
		tipc_sk_finish_conn(tsk, oport, onode);
		msg_set_importance(&tsk->phdr, msg_importance(hdr));

		/* If 'ACK+' message, add to socket receive queue */
		/* ACK+ message with data is added to receive queue */
		if (msg_data_sz(hdr))
		return true;

		/* If empty 'ACK-' message, wake up sleeping connect() */
		/* Empty ACK-, - wake up sleeping connect() and drop */
		sk->sk_data_ready(sk);

		/* 'ACK-' message is neither accepted nor rejected: */
		msg_set_dest_droppable(hdr, 1);
		return false;
		}
		/* Ignore connectionless message if not from listening socket */
		if (oport != pport \|\| onode != pnode)
		return false;

		/* Rejected SYN */
		if (err != TIPC_ERR_OVERLOAD)
		break;

		/* Prepare for new setup attempt if we have a SYN clone */
		if (skb_queue_empty(&sk->sk_write_queue))
		break;
		get_random_bytes(&delay, 2);
		delay %= (tsk->conn_timeout / 4);
		delay = msecs_to_jiffies(delay + 100);
		sk_reset_timer(sk, &sk->sk_timer, jiffies + delay);
		return false;
		case TIPC_OPEN:
		case TIPC_DISCONNECTING:
		break;
		return false;
		case TIPC_LISTEN:
		/* Accept only SYN message */
		if (!msg_connected(hdr) && !(msg_errcode(hdr)))
		if (!msg_is_syn(hdr) &&
		tipc_node_get_capabilities(net, onode) & TIPC_SYN_BIT)
		return false;
		if (!con_msg && !err)
		return true;
		break;
		return false;
		case TIPC_ESTABLISHED:
		/* Accept only connection-based messages sent by peer */
		if (unlikely(!tsk_peer_msg(tsk, hdr)))
		if (likely(con_msg && !err && pport == oport && pnode == onode))
		return true;
		if (!tsk_peer_msg(tsk, hdr))
		return false;

		if (unlikely(msg_errcode(hdr))) {
		if (!err)
		return true;
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		/* Let timer expire on it's own */
		tipc_node_remove_conn(net, tsk_peer_node(tsk),
		tsk->portid);
		tipc_node_remove_conn(net, pnode, tsk->portid);
		sk->sk_state_change(sk);
		}
		return true;
		default:
		pr_err("Unknown sk_state %u\n", sk->sk_state);
		}

		return false;
		/* Abort connection setup attempt */
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		sk->sk_err = ECONNREFUSED;
		sk->sk_state_change(sk);
		return true;
		}

		/**
		@@ -2545,43 +2551,78 @@ static int tipc_shutdown(struct socket *sock, int how)
		return res;
		}

		static void tipc_sk_check_probing_state(struct sock *sk,
		struct sk_buff_head *list)
		{
		struct tipc_sock *tsk = tipc_sk(sk);
		u32 pnode = tsk_peer_node(tsk);
		u32 pport = tsk_peer_port(tsk);
		u32 self = tsk_own_node(tsk);
		u32 oport = tsk->portid;
		struct sk_buff *skb;

		if (tsk->probe_unacked) {
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		sk->sk_err = ECONNABORTED;
		tipc_node_remove_conn(sock_net(sk), pnode, pport);
		sk->sk_state_change(sk);
		return;
		}
		/* Prepare new probe */
		skb = tipc_msg_create(CONN_MANAGER, CONN_PROBE, INT_H_SIZE, 0,
		pnode, self, pport, oport, TIPC_OK);
		if (skb)
		__skb_queue_tail(list, skb);
		tsk->probe_unacked = true;
		sk_reset_timer(sk, &sk->sk_timer, jiffies + CONN_PROBING_INTV);
		}

		static void tipc_sk_retry_connect(struct sock sk, struct sk_buff_head list)
		{
		struct tipc_sock *tsk = tipc_sk(sk);

		/* Try again later if dest link is congested */
		if (tsk->cong_link_cnt) {
		sk_reset_timer(sk, &sk->sk_timer, msecs_to_jiffies(100));
		return;
		}
		/* Prepare SYN for retransmit */
		tipc_msg_skb_clone(&sk->sk_write_queue, list);
		}

		static void tipc_sk_timeout(struct timer_list *t)
		{
		struct sock *sk = from_timer(sk, t, sk_timer);
		struct tipc_sock *tsk = tipc_sk(sk);
		u32 peer_port = tsk_peer_port(tsk);
		u32 peer_node = tsk_peer_node(tsk);
		u32 own_node = tsk_own_node(tsk);
		u32 own_port = tsk->portid;
		struct net *net = sock_net(sk);
		struct sk_buff *skb = NULL;
		u32 pnode = tsk_peer_node(tsk);
		struct sk_buff_head list;
		int rc = 0;

		skb_queue_head_init(&list);
		bh_lock_sock(sk);
		if (!tipc_sk_connected(sk))
		goto exit;

		/* Try again later if socket is busy */
		if (sock_owned_by_user(sk)) {
		sk_reset_timer(sk, &sk->sk_timer, jiffies + HZ / 20);
		goto exit;
		bh_unlock_sock(sk);
		return;
		}

		if (tsk->probe_unacked) {
		tipc_set_sk_state(sk, TIPC_DISCONNECTING);
		tipc_node_remove_conn(net, peer_node, peer_port);
		sk->sk_state_change(sk);
		goto exit;
		}
		/* Send new probe */
		skb = tipc_msg_create(CONN_MANAGER, CONN_PROBE, INT_H_SIZE, 0,
		peer_node, own_node, peer_port, own_port,
		TIPC_OK);
		tsk->probe_unacked = true;
		sk_reset_timer(sk, &sk->sk_timer, jiffies + CONN_PROBING_INTV);
		exit:
		if (sk->sk_state == TIPC_ESTABLISHED)
		tipc_sk_check_probing_state(sk, &list);
		else if (sk->sk_state == TIPC_CONNECTING)
		tipc_sk_retry_connect(sk, &list);

		bh_unlock_sock(sk);
		if (skb)
		tipc_node_xmit_skb(net, skb, peer_node, own_port);

		if (!skb_queue_empty(&list))
		rc = tipc_node_xmit(sock_net(sk), &list, pnode, tsk->portid);

		/* SYN messages may cause link congestion */
		if (rc == -ELINKCONG) {
		tipc_dest_push(&tsk->cong_links, pnode, 0);
		tsk->cong_link_cnt = 1;
		}
		sock_put(sk);
		}

Admin message