Commit 245500d8 authored by David Howells's avatar David Howells
Browse files

rxrpc: Rewrite the client connection manager 



Rewrite the rxrpc client connection manager so that it can support multiple
connections for a given security key to a peer.  The following changes are
made:

 (1) For each open socket, the code currently maintains an rbtree with the
     connections placed into it, keyed by communications parameters.  This
     is tricky to maintain as connections can be culled from the tree or
     replaced within it.  Connections can require replacement for a number
     of reasons, e.g. their IDs span too great a range for the IDR data
     type to represent efficiently, the call ID numbers on that conn would
     overflow or the conn got aborted.

     This is changed so that there's now a connection bundle object placed
     in the tree, keyed on the same parameters.  The bundle, however, does
     not need to be replaced.

 (2) An rxrpc_bundle object can now manage the available channels for a set
     of parallel connections.  The lock that manages this is moved there
     from the rxrpc_connection struct (channel_lock).

 (3) There'a a dummy bundle for all incoming connections to share so that
     they have a channel_lock too.  It might be better to give each
     incoming connection its own bundle.  This bundle is not needed to
     manage which channels incoming calls are made on because that's the
     solely at whim of the client.

 (4) The restrictions on how many client connections are around are
     removed.  Instead, a previous patch limits the number of client calls
     that can be allocated.  Ordinarily, client connections are reaped
     after 2 minutes on the idle queue, but when more than a certain number
     of connections are in existence, the reaper starts reaping them after
     2s of idleness instead to get the numbers back down.

     It could also be made such that new call allocations are forced to
     wait until the number of outstanding connections subsides.

Signed-off-by: default avatarDavid Howells <dhowells@redhat.com>
parent b7a7d674

include/trace/events/rxrpc.h

+4 −29
Original line number Diff line number Diff line
@@ -68,21 +68,14 @@ enum rxrpc_client_trace { 
	rxrpc_client_chan_activate,

	rxrpc_client_chan_disconnect,

	rxrpc_client_chan_pass,

	rxrpc_client_chan_unstarted,

	rxrpc_client_chan_wait_failed,

	rxrpc_client_cleanup,

	rxrpc_client_count,

	rxrpc_client_discard,

	rxrpc_client_duplicate,

	rxrpc_client_exposed,

	rxrpc_client_replace,

	rxrpc_client_to_active,

	rxrpc_client_to_culled,

	rxrpc_client_to_idle,

	rxrpc_client_to_inactive,

	rxrpc_client_to_upgrade,

	rxrpc_client_to_waiting,

	rxrpc_client_uncount,

};



enum rxrpc_call_trace {
@@ -271,29 +264,14 @@ enum rxrpc_tx_point { 
	EM(rxrpc_client_chan_activate,		"ChActv") \

	EM(rxrpc_client_chan_disconnect,	"ChDisc") \

	EM(rxrpc_client_chan_pass,		"ChPass") \

	EM(rxrpc_client_chan_unstarted,		"ChUnst") \

	EM(rxrpc_client_chan_wait_failed,	"ChWtFl") \

	EM(rxrpc_client_cleanup,		"Clean ") \

	EM(rxrpc_client_count,			"Count ") \

	EM(rxrpc_client_discard,		"Discar") \

	EM(rxrpc_client_duplicate,		"Duplic") \

	EM(rxrpc_client_exposed,		"Expose") \

	EM(rxrpc_client_replace,		"Replac") \

	EM(rxrpc_client_to_active,		"->Actv") \

	EM(rxrpc_client_to_culled,		"->Cull") \

	EM(rxrpc_client_to_idle,		"->Idle") \

	EM(rxrpc_client_to_inactive,		"->Inac") \

	EM(rxrpc_client_to_upgrade,		"->Upgd") \

	EM(rxrpc_client_to_waiting,		"->Wait") \

	E_(rxrpc_client_uncount,		"Uncoun")



#define rxrpc_conn_cache_states \

	EM(RXRPC_CONN_CLIENT_INACTIVE,		"Inac") \

	EM(RXRPC_CONN_CLIENT_WAITING,		"Wait") \

	EM(RXRPC_CONN_CLIENT_ACTIVE,		"Actv") \

	EM(RXRPC_CONN_CLIENT_UPGRADE,		"Upgd") \

	EM(RXRPC_CONN_CLIENT_CULLED,		"Cull") \

	E_(RXRPC_CONN_CLIENT_IDLE,		"Idle") \

	E_(rxrpc_client_to_idle,		"->Idle")



#define rxrpc_call_traces \

	EM(rxrpc_call_connected,		"CON") \
@@ -594,23 +572,20 @@ TRACE_EVENT(rxrpc_client, 
		    __field(int,			channel		)

		    __field(int,			usage		)

		    __field(enum rxrpc_client_trace,	op		)

		    __field(enum rxrpc_conn_cache_state, cs		)

			     ),



	    TP_fast_assign(

		    __entry->conn = conn->debug_id;

		    __entry->conn = conn ? conn->debug_id : 0;

		    __entry->channel = channel;

		    __entry->usage = atomic_read(&conn->usage);

		    __entry->usage = conn ? atomic_read(&conn->usage) : -2;

		    __entry->op = op;

		    __entry->cid = conn->proto.cid;

		    __entry->cs = conn->cache_state;

			   ),



	    TP_printk("C=%08x h=%2d %s %s i=%08x u=%d",

	    TP_printk("C=%08x h=%2d %s i=%08x u=%d",

		      __entry->conn,

		      __entry->channel,

		      __print_symbolic(__entry->op, rxrpc_client_traces),

		      __print_symbolic(__entry->cs, rxrpc_conn_cache_states),

		      __entry->cid,

		      __entry->usage)

	    );

net/rxrpc/ar-internal.h

+32 −36
Original line number Diff line number Diff line
@@ -76,14 +76,12 @@ struct rxrpc_net { 
	struct work_struct	service_conn_reaper;

	struct timer_list	service_conn_reap_timer;



	unsigned int		nr_client_conns;

	unsigned int		nr_active_client_conns;

	bool			kill_all_client_conns;

	bool			live;



	bool			kill_all_client_conns;

	atomic_t		nr_client_conns;

	spinlock_t		client_conn_cache_lock; /* Lock for ->*_client_conns */

	spinlock_t		client_conn_discard_lock; /* Prevent multiple discarders */

	struct list_head	waiting_client_conns;

	struct list_head	active_client_conns;

	struct list_head	idle_client_conns;

	struct work_struct	client_conn_reaper;

	struct timer_list	client_conn_reap_timer;
@@ -275,8 +273,8 @@ struct rxrpc_local { 
	struct rw_semaphore	defrag_sem;	/* control re-enablement of IP DF bit */

	struct sk_buff_head	reject_queue;	/* packets awaiting rejection */

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

	struct rb_root		client_conns;	/* Client connections by socket params */

	spinlock_t		client_conns_lock; /* Lock for client_conns */

	struct rb_root		client_bundles;	/* Client connection bundles by socket params */

	spinlock_t		client_bundles_lock; /* Lock for client_bundles */

	spinlock_t		lock;		/* access lock */

	rwlock_t		services_lock;	/* lock for services list */

	int			debug_id;	/* debug ID for printks */
@@ -353,10 +351,7 @@ struct rxrpc_conn_parameters { 
enum rxrpc_conn_flag {

	RXRPC_CONN_HAS_IDR,		/* Has a client conn ID assigned */

	RXRPC_CONN_IN_SERVICE_CONNS,	/* Conn is in peer->service_conns */

	RXRPC_CONN_IN_CLIENT_CONNS,	/* Conn is in local->client_conns */

	RXRPC_CONN_EXPOSED,		/* Conn has extra ref for exposure */

	RXRPC_CONN_DONT_REUSE,		/* Don't reuse this connection */

	RXRPC_CONN_COUNTED,		/* Counted by rxrpc_nr_client_conns */

	RXRPC_CONN_PROBING_FOR_UPGRADE,	/* Probing for service upgrade */

	RXRPC_CONN_FINAL_ACK_0,		/* Need final ACK for channel 0 */

	RXRPC_CONN_FINAL_ACK_1,		/* Need final ACK for channel 1 */
@@ -376,19 +371,6 @@ enum rxrpc_conn_event { 
	RXRPC_CONN_EV_CHALLENGE,	/* Send challenge packet */

};



/*

 * The connection cache state.

 */

enum rxrpc_conn_cache_state {

	RXRPC_CONN_CLIENT_INACTIVE,	/* Conn is not yet listed */

	RXRPC_CONN_CLIENT_WAITING,	/* Conn is on wait list, waiting for capacity */

	RXRPC_CONN_CLIENT_ACTIVE,	/* Conn is on active list, doing calls */

	RXRPC_CONN_CLIENT_UPGRADE,	/* Conn is on active list, probing for upgrade */

	RXRPC_CONN_CLIENT_CULLED,	/* Conn is culled and delisted, doing calls */

	RXRPC_CONN_CLIENT_IDLE,		/* Conn is on idle list, doing mostly nothing */

	RXRPC_CONN__NR_CACHE_STATES

};



/*

 * The connection protocol state.

 */
@@ -404,6 +386,23 @@ enum rxrpc_conn_proto_state { 
	RXRPC_CONN__NR_STATES

};



/*

 * RxRPC client connection bundle.

 */

struct rxrpc_bundle {

	struct rxrpc_conn_parameters params;

	atomic_t		usage;

	unsigned int		debug_id;

	bool			try_upgrade;	/* True if the bundle is attempting upgrade */

	bool			alloc_conn;	/* True if someone's getting a conn */

	unsigned short		alloc_error;	/* Error from last conn allocation */

	spinlock_t		channel_lock;

	struct rb_node		local_node;	/* Node in local->client_conns */

	struct list_head	waiting_calls;	/* Calls waiting for channels */

	unsigned long		avail_chans;	/* Mask of available channels */

	struct rxrpc_connection	*conns[4];	/* The connections in the bundle (max 4) */

};



/*

 * RxRPC connection definition

 * - matched by { local, peer, epoch, conn_id, direction }
@@ -417,10 +416,7 @@ struct rxrpc_connection { 
	struct rcu_head		rcu;

	struct list_head	cache_link;



	spinlock_t		channel_lock;

	unsigned char		active_chans;	/* Mask of active channels */

#define RXRPC_ACTIVE_CHANS_MASK	((1 << RXRPC_MAXCALLS) - 1)

	struct list_head	waiting_calls;	/* Calls waiting for channels */

	unsigned char		act_chans;	/* Mask of active channels */

	struct rxrpc_channel {

		unsigned long		final_ack_at;	/* Time at which to issue final ACK */

		struct rxrpc_call __rcu	*call;		/* Active call */
@@ -437,10 +433,8 @@ struct rxrpc_connection { 


	struct timer_list	timer;		/* Conn event timer */

	struct work_struct	processor;	/* connection event processor */

	union {

		struct rb_node	client_node;	/* Node in local->client_conns */

	struct rxrpc_bundle	*bundle;	/* Client connection bundle */

	struct rb_node		service_node;	/* Node in peer->service_conns */

	};

	struct list_head	proc_link;	/* link in procfs list */

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

	struct sk_buff_head	rx_queue;	/* received conn-level packets */
@@ -452,7 +446,6 @@ struct rxrpc_connection { 
	unsigned long		events;

	unsigned long		idle_timestamp;	/* Time at which last became idle */

	spinlock_t		state_lock;	/* state-change lock */

	enum rxrpc_conn_cache_state cache_state;

	enum rxrpc_conn_proto_state state;	/* current state of connection */

	u32			abort_code;	/* Abort code of connection abort */

	int			debug_id;	/* debug ID for printks */
@@ -464,6 +457,7 @@ struct rxrpc_connection { 
	u8			security_size;	/* security header size */

	u8			security_ix;	/* security type */

	u8			out_clientflag;	/* RXRPC_CLIENT_INITIATED if we are client */

	u8			bundle_shift;	/* Index into bundle->avail_chans */

	short			error;		/* Local error code */

};
@@ -494,6 +488,7 @@ enum rxrpc_call_flag { 
	RXRPC_CALL_RX_UNDERRUN,		/* Got data underrun */

	RXRPC_CALL_DISCONNECTED,	/* The call has been disconnected */

	RXRPC_CALL_KERNEL,		/* The call was made by the kernel */

	RXRPC_CALL_UPGRADE,		/* Service upgrade was requested for the call */

};



/*
@@ -578,7 +573,7 @@ struct rxrpc_call { 
	struct work_struct	processor;	/* Event processor */

	rxrpc_notify_rx_t	notify_rx;	/* kernel service Rx notification function */

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

	struct list_head	chan_wait_link;	/* Link in conn->waiting_calls */

	struct list_head	chan_wait_link;	/* Link in conn->bundle->waiting_calls */

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

	struct list_head	accept_link;	/* Link in rx->acceptq */

	struct list_head	recvmsg_link;	/* Link in rx->recvmsg_q */
@@ -817,18 +812,19 @@ static inline bool rxrpc_is_client_call(const struct rxrpc_call *call) 
/*

 * conn_client.c

 */

extern unsigned int rxrpc_max_client_connections;

extern unsigned int rxrpc_reap_client_connections;

extern unsigned long rxrpc_conn_idle_client_expiry;

extern unsigned long rxrpc_conn_idle_client_fast_expiry;

extern struct idr rxrpc_client_conn_ids;



void rxrpc_destroy_client_conn_ids(void);

struct rxrpc_bundle *rxrpc_get_bundle(struct rxrpc_bundle *);

void rxrpc_put_bundle(struct rxrpc_bundle *);

int rxrpc_connect_call(struct rxrpc_sock *, struct rxrpc_call *,

		       struct rxrpc_conn_parameters *, struct sockaddr_rxrpc *,

		       gfp_t);

void rxrpc_expose_client_call(struct rxrpc_call *);

void rxrpc_disconnect_client_call(struct rxrpc_call *);

void rxrpc_disconnect_client_call(struct rxrpc_bundle *, struct rxrpc_call *);

void rxrpc_put_client_conn(struct rxrpc_connection *);

void rxrpc_discard_expired_client_conns(struct work_struct *);

void rxrpc_destroy_all_client_connections(struct rxrpc_net *);
@@ -854,7 +850,7 @@ void rxrpc_disconnect_call(struct rxrpc_call *); 
void rxrpc_kill_connection(struct rxrpc_connection *);

bool rxrpc_queue_conn(struct rxrpc_connection *);

void rxrpc_see_connection(struct rxrpc_connection *);

void rxrpc_get_connection(struct rxrpc_connection *);

struct rxrpc_connection *rxrpc_get_connection(struct rxrpc_connection *);

struct rxrpc_connection *rxrpc_get_connection_maybe(struct rxrpc_connection *);

void rxrpc_put_service_conn(struct rxrpc_connection *);

void rxrpc_service_connection_reaper(struct work_struct *);

net/rxrpc/conn_client.c

+487 −594

File changed.

Preview size limit exceeded, changes collapsed.

net/rxrpc/conn_event.c

+8 −6
Original line number Diff line number Diff line
@@ -157,12 +157,12 @@ static void rxrpc_abort_calls(struct rxrpc_connection *conn, 


	_enter("{%d},%x", conn->debug_id, conn->abort_code);



	spin_lock(&conn->channel_lock);

	spin_lock(&conn->bundle->channel_lock);



	for (i = 0; i < RXRPC_MAXCALLS; i++) {

		call = rcu_dereference_protected(

			conn->channels[i].call,

			lockdep_is_held(&conn->channel_lock));

			lockdep_is_held(&conn->bundle->channel_lock));

		if (call) {

			if (compl == RXRPC_CALL_LOCALLY_ABORTED)

				trace_rxrpc_abort(call->debug_id,
@@ -179,7 +179,7 @@ static void rxrpc_abort_calls(struct rxrpc_connection *conn, 
		}

	}



	spin_unlock(&conn->channel_lock);

	spin_unlock(&conn->bundle->channel_lock);

	_leave("");

}
@@ -210,6 +210,7 @@ static int rxrpc_abort_connection(struct rxrpc_connection *conn, 
	conn->error = error;

	conn->abort_code = abort_code;

	conn->state = RXRPC_CONN_LOCALLY_ABORTED;

	set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);

	spin_unlock_bh(&conn->state_lock);



	msg.msg_name	= &conn->params.peer->srx.transport;
@@ -319,6 +320,7 @@ static int rxrpc_process_event(struct rxrpc_connection *conn, 
		conn->error = -ECONNABORTED;

		conn->abort_code = abort_code;

		conn->state = RXRPC_CONN_REMOTELY_ABORTED;

		set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);

		rxrpc_abort_calls(conn, RXRPC_CALL_REMOTELY_ABORTED, sp->hdr.serial);

		return -ECONNABORTED;
@@ -339,7 +341,7 @@ static int rxrpc_process_event(struct rxrpc_connection *conn, 
		if (ret < 0)

			return ret;



		spin_lock(&conn->channel_lock);

		spin_lock(&conn->bundle->channel_lock);

		spin_lock(&conn->state_lock);



		if (conn->state == RXRPC_CONN_SERVICE_CHALLENGING) {
@@ -349,12 +351,12 @@ static int rxrpc_process_event(struct rxrpc_connection *conn, 
				rxrpc_call_is_secure(

					rcu_dereference_protected(

						conn->channels[loop].call,

						lockdep_is_held(&conn->channel_lock)));

						lockdep_is_held(&conn->bundle->channel_lock)));

		} else {

			spin_unlock(&conn->state_lock);

		}



		spin_unlock(&conn->channel_lock);

		spin_unlock(&conn->bundle->channel_lock);

		return 0;



	default:

net/rxrpc/conn_object.c

+6 −6
Original line number Diff line number Diff line
@@ -41,8 +41,6 @@ struct rxrpc_connection *rxrpc_alloc_connection(gfp_t gfp) 
	conn = kzalloc(sizeof(struct rxrpc_connection), gfp);

	if (conn) {

		INIT_LIST_HEAD(&conn->cache_link);

		spin_lock_init(&conn->channel_lock);

		INIT_LIST_HEAD(&conn->waiting_calls);

		timer_setup(&conn->timer, &rxrpc_connection_timer, 0);

		INIT_WORK(&conn->processor, &rxrpc_process_connection);

		INIT_LIST_HEAD(&conn->proc_link);
@@ -219,11 +217,11 @@ void rxrpc_disconnect_call(struct rxrpc_call *call) 
	}



	if (rxrpc_is_client_call(call))

		return rxrpc_disconnect_client_call(call);

		return rxrpc_disconnect_client_call(conn->bundle, call);



	spin_lock(&conn->channel_lock);

	spin_lock(&conn->bundle->channel_lock);

	__rxrpc_disconnect_call(conn, call);

	spin_unlock(&conn->channel_lock);

	spin_unlock(&conn->bundle->channel_lock);



	set_bit(RXRPC_CALL_DISCONNECTED, &call->flags);

	conn->idle_timestamp = jiffies;
@@ -292,12 +290,13 @@ void rxrpc_see_connection(struct rxrpc_connection *conn) 
/*

 * Get a ref on a connection.

 */

void rxrpc_get_connection(struct rxrpc_connection *conn)

struct rxrpc_connection *rxrpc_get_connection(struct rxrpc_connection *conn)

{

	const void *here = __builtin_return_address(0);

	int n = atomic_inc_return(&conn->usage);



	trace_rxrpc_conn(conn->debug_id, rxrpc_conn_got, n, here);

	return conn;

}



/*
@@ -365,6 +364,7 @@ static void rxrpc_destroy_connection(struct rcu_head *rcu) 
	conn->security->clear(conn);

	key_put(conn->params.key);

	key_put(conn->server_key);

	rxrpc_put_bundle(conn->bundle);

	rxrpc_put_peer(conn->params.peer);



	if (atomic_dec_and_test(&conn->params.local->rxnet->nr_conns))

CRA Git | Maintained and supported by SUSTech CRA and CCSE