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

Commit 56108c11 authored by David S. Miller's avatar David S. Miller
Browse files

Merge tag 'rxrpc-rewrite-20160615' of... 

Merge tag 'rxrpc-rewrite-20160615' of git://git.kernel.org/pub/scm/linux/kernel/git/dhowells/linux-fs



David Howells says:

====================
rxrpc: Rework endpoint record handling

Here's the next part of the AF_RXRPC rewrite.  In this set I rework
endpoint record handling.  There are two types of endpoint record, local
and peer.  The local endpoint record is used as an anchor for the transport
socket that AF_RXRPC uses (at the moment a UDP socket).  Local endpoints
can be shared between AF_RXRPC sockets under certain restricted
circumstances.

The peer endpoint is a record of the remote end.  It is (or will be) used
to keep track MTU and RTT values and, with these changes, is used to find
the call(s) to abort when a network error occurs.

The following significant changes are made:

 (1) The local endpoint event handling code is split out into its own file.

 (2) The local endpoint list bottom half-excluding spinlock is removed as
     things are arranged such that sk_user_data will not change whilst the
     transport socket callbacks are in progress.

 (3) Local endpoints can now only be shared if they have the same transport
     address (as before) and have a local service ID of 0 (ie. they're not
     listening for incoming calls).  This prevents callbacks from a server
     to one process being picked up by another process.

 (4) Local endpoint destruction is now accomplished by the same work item
     as processes events, meaning that the destructor doesn't need to wait
     for the event processor.

 (5) Peer endpoints are now held in a hash table rather than a flat list.

 (6) Peer endpoints are now destroyed by RCU rather than by work item.

 (7) Peer endpoints are now differentiated by local endpoint and remote
     transport port in addition to remote transport address and transport
     type and family.

     This means that a firewall that excludes access between a particular
     local port and remote port won't cause calls to be aborted that use a
     different port pair.

 (8) Error report handling now no longer assumes that the source is always
     an IPv4 ICMP message from a UDP port and has assumptions that an ICMP
     message comes from an IPv4 socket removed.  At some point IPv6 support
     will be added.

 (9) Peer endpoints rather than local endpoints are now the anchor point
     for distributing network error reports.

(10) Both types of endpoint records are now disposed of as soon as all
     references to them are gone.  There is less hanging around and once
     their usage counts hit zero, records can no longer be resurrected.
====================

Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parents 9c9ad412 4f95dd78

net/rxrpc/Makefile

+3 −1
Original line number Diff line number Diff line
@@ -12,6 +12,7 @@ af-rxrpc-y := \ 
	input.o \

	insecure.o \

	key.o \

	local_event.o \

	local_object.o \

	misc.o \

	output.o \
@@ -20,7 +21,8 @@ af-rxrpc-y := \ 
	recvmsg.o \

	security.o \

	skbuff.o \

	transport.o

	transport.o \

	utils.o



af-rxrpc-$(CONFIG_PROC_FS) += proc.o

af-rxrpc-$(CONFIG_RXKAD) += rxkad.o

net/rxrpc/af_rxrpc.c

+19 −3
Original line number Diff line number Diff line
@@ -102,6 +102,8 @@ static int rxrpc_validate_address(struct rxrpc_sock *rx, 


	switch (srx->transport.family) {

	case AF_INET:

		if (srx->transport_len < sizeof(struct sockaddr_in))

			return -EINVAL;

		_debug("INET: %x @ %pI4",

		       ntohs(srx->transport.sin.sin_port),

		       &srx->transport.sin.sin_addr);
@@ -244,7 +246,7 @@ struct rxrpc_transport *rxrpc_name_to_transport(struct rxrpc_sock *rx, 
		return ERR_PTR(-EAFNOSUPPORT);



	/* find a remote transport endpoint from the local one */

	peer = rxrpc_get_peer(srx, gfp);

	peer = rxrpc_lookup_peer(rx->local, srx, gfp);

	if (IS_ERR(peer))

		return ERR_CAST(peer);
@@ -835,13 +837,27 @@ static void __exit af_rxrpc_exit(void) 
	rxrpc_destroy_all_calls();

	rxrpc_destroy_all_connections();

	rxrpc_destroy_all_transports();

	rxrpc_destroy_all_peers();

	rxrpc_destroy_all_locals();



	ASSERTCMP(atomic_read(&rxrpc_n_skbs), ==, 0);



	/* We need to flush the scheduled work twice because the local endpoint

	 * records involve a work item in their destruction as they can only be

	 * destroyed from process context.  However, a connection may have a

	 * work item outstanding - and this will pin the local endpoint record

	 * until the connection goes away.

	 *

	 * Peers don't pin locals and calls pin sockets - which prevents the

	 * module from being unloaded - so we should only need two flushes.

	 */

	_debug("flush scheduled work");

	flush_workqueue(rxrpc_workqueue);

	_debug("flush scheduled work 2");

	flush_workqueue(rxrpc_workqueue);

	_debug("synchronise RCU");

	rcu_barrier();

	_debug("destroy locals");

	rxrpc_destroy_all_locals();



	remove_proc_entry("rxrpc_conns", init_net.proc_net);

	remove_proc_entry("rxrpc_calls", init_net.proc_net);

	destroy_workqueue(rxrpc_workqueue);

net/rxrpc/ar-internal.h

+82 −42
Original line number Diff line number Diff line
@@ -9,7 +9,9 @@ 
 * 2 of the License, or (at your option) any later version.

 */



#include <linux/atomic.h>

#include <net/sock.h>

#include <net/af_rxrpc.h>

#include <rxrpc/packet.h>



#if 0
@@ -168,46 +170,49 @@ struct rxrpc_security { 
};



/*

 * RxRPC local transport endpoint definition

 * - matched by local port, address and protocol type

 * RxRPC local transport endpoint description

 * - owned by a single AF_RXRPC socket

 * - pointed to by transport socket struct sk_user_data

 */

struct rxrpc_local {

	struct rcu_head		rcu;

	atomic_t		usage;

	struct list_head	link;

	struct socket		*socket;	/* my UDP socket */

	struct work_struct	destroyer;	/* endpoint destroyer */

	struct work_struct	acceptor;	/* incoming call processor */

	struct work_struct	rejecter;	/* packet reject writer */

	struct work_struct	event_processor; /* endpoint event processor */

	struct work_struct	processor;

	struct list_head	services;	/* services listening on this endpoint */

	struct list_head	link;		/* link in endpoint list */

	struct rw_semaphore	defrag_sem;	/* control re-enablement of IP DF bit */

	struct sk_buff_head	accept_queue;	/* incoming calls awaiting acceptance */

	struct sk_buff_head	reject_queue;	/* packets awaiting rejection */

	struct sk_buff_head	event_queue;	/* endpoint event packets awaiting processing */

	struct mutex		conn_lock;	/* Client connection creation lock */

	spinlock_t		lock;		/* access lock */

	rwlock_t		services_lock;	/* lock for services list */

	atomic_t		usage;

	int			debug_id;	/* debug ID for printks */

	volatile char		error_rcvd;	/* T if received ICMP error outstanding */

	bool			dead;

	struct sockaddr_rxrpc	srx;		/* local address */

};



/*

 * RxRPC remote transport endpoint definition

 * - matched by remote port, address and protocol type

 * - holds the connection ID counter for connections between the two endpoints

 * - matched by local endpoint, remote port, address and protocol type

 */

struct rxrpc_peer {

	struct work_struct	destroyer;	/* peer destroyer */

	struct list_head	link;		/* link in master peer list */

	struct list_head	error_targets;	/* targets for net error distribution */

	spinlock_t		lock;		/* access lock */

	struct rcu_head		rcu;		/* This must be first */

	atomic_t		usage;

	unsigned long		hash_key;

	struct hlist_node	hash_link;

	struct rxrpc_local	*local;

	struct hlist_head	error_targets;	/* targets for net error distribution */

	struct work_struct	error_distributor;

	spinlock_t		lock;		/* access lock */

	unsigned int		if_mtu;		/* interface MTU for this peer */

	unsigned int		mtu;		/* network MTU for this peer */

	unsigned int		maxdata;	/* data size (MTU - hdrsize) */

	unsigned short		hdrsize;	/* header size (IP + UDP + RxRPC) */

	int			debug_id;	/* debug ID for printks */

	int			net_error;	/* network error distributed */

	int			error_report;	/* Net (+0) or local (+1000000) to distribute */

#define RXRPC_LOCAL_ERROR_OFFSET 1000000

	struct sockaddr_rxrpc	srx;		/* remote address */



	/* calculated RTT cache */
@@ -226,12 +231,10 @@ struct rxrpc_peer { 
struct rxrpc_transport {

	struct rxrpc_local	*local;		/* local transport endpoint */

	struct rxrpc_peer	*peer;		/* remote transport endpoint */

	struct work_struct	error_handler;	/* network error distributor */

	struct rb_root		bundles;	/* client connection bundles on this transport */

	struct rb_root		client_conns;	/* client connections on this transport */

	struct rb_root		server_conns;	/* server connections on this transport */

	struct list_head	link;		/* link in master session list */

	struct sk_buff_head	error_queue;	/* error packets awaiting processing */

	unsigned long		put_time;	/* time at which to reap */

	spinlock_t		client_lock;	/* client connection allocation lock */

	rwlock_t		conn_lock;	/* lock for active/dead connections */
@@ -390,7 +393,7 @@ struct rxrpc_call { 
	struct work_struct	destroyer;	/* call destroyer */

	struct work_struct	processor;	/* packet processor and ACK generator */

	struct list_head	link;		/* link in master call list */

	struct list_head	error_link;	/* link in error distribution list */

	struct hlist_node	error_link;	/* link in error distribution list */

	struct list_head	accept_link;	/* calls awaiting acceptance */

	struct rb_node		sock_node;	/* node in socket call tree */

	struct rb_node		conn_node;	/* node in connection call tree */
@@ -408,7 +411,8 @@ struct rxrpc_call { 
	atomic_t		sequence;	/* Tx data packet sequence counter */

	u32			local_abort;	/* local abort code */

	u32			remote_abort;	/* remote abort code */

	int			error;		/* local error incurred */

	int			error_report;	/* Network error (ICMP/local transport) */

	int			error;		/* Local error incurred */

	enum rxrpc_call_state	state : 8;	/* current state of call */

	int			debug_id;	/* debug ID for printks */

	u8			channel;	/* connection channel occupied by this call */
@@ -484,7 +488,7 @@ extern struct rxrpc_transport *rxrpc_name_to_transport(struct rxrpc_sock *, 
/*

 * call_accept.c

 */

void rxrpc_accept_incoming_calls(struct work_struct *);

void rxrpc_accept_incoming_calls(struct rxrpc_local *);

struct rxrpc_call *rxrpc_accept_call(struct rxrpc_sock *, unsigned long);

int rxrpc_reject_call(struct rxrpc_sock *);
@@ -524,7 +528,7 @@ void __exit rxrpc_destroy_all_calls(void); 
 */

void rxrpc_process_connection(struct work_struct *);

void rxrpc_reject_packet(struct rxrpc_local *, struct sk_buff *);

void rxrpc_reject_packets(struct work_struct *);

void rxrpc_reject_packets(struct rxrpc_local *);



/*

 * conn_object.c
@@ -570,14 +574,34 @@ int rxrpc_get_server_data_key(struct rxrpc_connection *, const void *, time_t, 
			      u32);



/*

 * local_object.c

 * local_event.c

 */

extern rwlock_t rxrpc_local_lock;

extern void rxrpc_process_local_events(struct rxrpc_local *);



struct rxrpc_local *rxrpc_lookup_local(struct sockaddr_rxrpc *);

void rxrpc_put_local(struct rxrpc_local *);

/*

 * local_object.c

 */

struct rxrpc_local *rxrpc_lookup_local(const struct sockaddr_rxrpc *);

void __rxrpc_put_local(struct rxrpc_local *);

void __exit rxrpc_destroy_all_locals(void);



static inline void rxrpc_get_local(struct rxrpc_local *local)

{

	atomic_inc(&local->usage);

}



static inline

struct rxrpc_local *rxrpc_get_local_maybe(struct rxrpc_local *local)

{

	return atomic_inc_not_zero(&local->usage) ? local : NULL;

}



static inline void rxrpc_put_local(struct rxrpc_local *local)

{

	if (atomic_dec_and_test(&local->usage))

		__rxrpc_put_local(local);

}



/*

 * misc.c

 */
@@ -603,18 +627,37 @@ int rxrpc_send_packet(struct rxrpc_transport *, struct sk_buff *); 
int rxrpc_do_sendmsg(struct rxrpc_sock *, struct msghdr *, size_t);



/*

 * peer_error.c

 * peer_event.c

 */

void rxrpc_UDP_error_report(struct sock *);

void rxrpc_UDP_error_handler(struct work_struct *);

void rxrpc_error_report(struct sock *);

void rxrpc_peer_error_distributor(struct work_struct *);



/*

 * peer_object.c

 */

struct rxrpc_peer *rxrpc_get_peer(struct sockaddr_rxrpc *, gfp_t);

void rxrpc_put_peer(struct rxrpc_peer *);

struct rxrpc_peer *rxrpc_find_peer(struct rxrpc_local *, __be32, __be16);

void __exit rxrpc_destroy_all_peers(void);

struct rxrpc_peer *rxrpc_lookup_peer_rcu(struct rxrpc_local *,

					 const struct sockaddr_rxrpc *);

struct rxrpc_peer *rxrpc_lookup_peer(struct rxrpc_local *,

				     struct sockaddr_rxrpc *, gfp_t);

struct rxrpc_peer *rxrpc_alloc_peer(struct rxrpc_local *, gfp_t);



static inline void rxrpc_get_peer(struct rxrpc_peer *peer)

{

	atomic_inc(&peer->usage);

}



static inline

struct rxrpc_peer *rxrpc_get_peer_maybe(struct rxrpc_peer *peer)

{

	return atomic_inc_not_zero(&peer->usage) ? peer : NULL;

}



extern void __rxrpc_put_peer(struct rxrpc_peer *peer);

static inline void rxrpc_put_peer(struct rxrpc_peer *peer)

{

	if (atomic_dec_and_test(&peer->usage))

		__rxrpc_put_peer(peer);

}



/*

 * proc.c
@@ -672,6 +715,12 @@ void __exit rxrpc_destroy_all_transports(void); 
struct rxrpc_transport *rxrpc_find_transport(struct rxrpc_local *,

					     struct rxrpc_peer *);



/*

 * utils.c

 */

void rxrpc_get_addr_from_skb(struct rxrpc_local *, const struct sk_buff *,

			     struct sockaddr_rxrpc *);



/*

 * debug tracing

 */
@@ -841,15 +890,6 @@ static inline void rxrpc_purge_queue(struct sk_buff_head *list) 
		rxrpc_free_skb(skb);

}



static inline void __rxrpc_get_local(struct rxrpc_local *local, const char *f)

{

	CHECK_SLAB_OKAY(&local->usage);

	if (atomic_inc_return(&local->usage) == 1)

		printk("resurrected (%s)\n", f);

}



#define rxrpc_get_local(LOCAL) __rxrpc_get_local((LOCAL), __func__)



#define rxrpc_get_call(CALL)				\

do {							\

	CHECK_SLAB_OKAY(&(CALL)->usage);		\

net/rxrpc/call_accept.c

+6 −21
Original line number Diff line number Diff line
@@ -95,7 +95,7 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local, 
	rxrpc_new_skb(notification);

	notification->mark = RXRPC_SKB_MARK_NEW_CALL;



	peer = rxrpc_get_peer(srx, GFP_NOIO);

	peer = rxrpc_lookup_peer(local, srx, GFP_NOIO);

	if (IS_ERR(peer)) {

		_debug("no peer");

		ret = -EBUSY;
@@ -202,10 +202,8 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local, 
 * accept incoming calls that need peer, transport and/or connection setting up

 * - the packets we get are all incoming client DATA packets that have seq == 1

 */

void rxrpc_accept_incoming_calls(struct work_struct *work)

void rxrpc_accept_incoming_calls(struct rxrpc_local *local)

{

	struct rxrpc_local *local =

		container_of(work, struct rxrpc_local, acceptor);

	struct rxrpc_skb_priv *sp;

	struct sockaddr_rxrpc srx;

	struct rxrpc_sock *rx;
@@ -215,21 +213,8 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) 


	_enter("%d", local->debug_id);



	read_lock_bh(&rxrpc_local_lock);

	if (atomic_read(&local->usage) > 0)

		rxrpc_get_local(local);

	else

		local = NULL;

	read_unlock_bh(&rxrpc_local_lock);

	if (!local) {

		_leave(" [local dead]");

		return;

	}



process_next_packet:

	skb = skb_dequeue(&local->accept_queue);

	if (!skb) {

		rxrpc_put_local(local);

		_leave("\n");

		return;

	}
@@ -292,7 +277,7 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) 
	case -ECONNRESET: /* old calls are ignored */

	case -ECONNABORTED: /* aborted calls are reaborted or ignored */

	case 0:

		goto process_next_packet;

		return;

	case -ECONNREFUSED:

		goto invalid_service;

	case -EBUSY:
@@ -308,18 +293,18 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) 
busy:

	rxrpc_busy(local, &srx, &whdr);

	rxrpc_free_skb(skb);

	goto process_next_packet;

	return;



invalid_service:

	skb->priority = RX_INVALID_OPERATION;

	rxrpc_reject_packet(local, skb);

	goto process_next_packet;

	return;



	/* can't change connection security type mid-flow */

security_mismatch:

	skb->priority = RX_PROTOCOL_ERROR;

	rxrpc_reject_packet(local, skb);

	goto process_next_packet;

	return;

}



/*

net/rxrpc/call_event.c

+11 −4
Original line number Diff line number Diff line
@@ -864,17 +864,24 @@ void rxrpc_process_call(struct work_struct *work) 
	}



	if (test_bit(RXRPC_CALL_EV_RCVD_ERROR, &call->events)) {

		enum rxrpc_skb_mark mark;

		int error;



		clear_bit(RXRPC_CALL_EV_CONN_ABORT, &call->events);

		clear_bit(RXRPC_CALL_EV_REJECT_BUSY, &call->events);

		clear_bit(RXRPC_CALL_EV_ABORT, &call->events);



		error = call->conn->trans->peer->net_error;

		error = call->error_report;

		if (error < RXRPC_LOCAL_ERROR_OFFSET) {

			mark = RXRPC_SKB_MARK_NET_ERROR;

			_debug("post net error %d", error);

		} else {

			mark = RXRPC_SKB_MARK_LOCAL_ERROR;

			error -= RXRPC_LOCAL_ERROR_OFFSET;

			_debug("post net local error %d", error);

		}



		if (rxrpc_post_message(call, RXRPC_SKB_MARK_NET_ERROR,

				       error, true) < 0)

		if (rxrpc_post_message(call, mark, error, true) < 0)

			goto no_mem;

		clear_bit(RXRPC_CALL_EV_RCVD_ERROR, &call->events);

		goto kill_ACKs;