ar-connection.c

linux 内核源代码

/* RxRPC virtual connection handler
 *
 * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells ([e-mail])
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version
 * 2 of the License, or (at your option) any later version.
 */

#include <linux/module.h>
#include <linux/net.h>
#include <linux/skbuff.h>
#include <linux/crypto.h>
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include "ar-internal.h"

static void rxrpc_connection_reaper(struct work_struct *work);

LIST_HEAD(rxrpc_connections);
DEFINE_RWLOCK(rxrpc_connection_lock);
static unsigned long rxrpc_connection_timeout = 10 * 60;
static DECLARE_DELAYED_WORK(rxrpc_connection_reap, rxrpc_connection_reaper);

/*
 * allocate a new client connection bundle
 */
static struct rxrpc_conn_bundle *rxrpc_alloc_bundle(gfp_t gfp)
{
	struct rxrpc_conn_bundle *bundle;

	_enter("");

	bundle = kzalloc(sizeof(struct rxrpc_conn_bundle), gfp);
	if (bundle) {
		INIT_LIST_HEAD(&bundle->unused_conns);
		INIT_LIST_HEAD(&bundle->avail_conns);
		INIT_LIST_HEAD(&bundle->busy_conns);
		init_waitqueue_head(&bundle->chanwait);
		atomic_set(&bundle->usage, 1);
	}

	_leave(" = %p", bundle);
	return bundle;
}

/*
 * compare bundle parameters with what we're looking for
 * - return -ve, 0 or +ve
 */
static inline
int rxrpc_cmp_bundle(const struct rxrpc_conn_bundle *bundle,
		     struct key *key, __be16 service_id)
{
	return (bundle->service_id - service_id) ?:
		((unsigned long) bundle->key - (unsigned long) key);
}

/*
 * get bundle of client connections that a client socket can make use of
 */
struct rxrpc_conn_bundle *rxrpc_get_bundle(struct rxrpc_sock *rx,
					   struct rxrpc_transport *trans,
					   struct key *key,
					   __be16 service_id,
					   gfp_t gfp)
{
	struct rxrpc_conn_bundle *bundle, *candidate;
	struct rb_node *p, *parent, **pp;

	_enter("%p{%x},%x,%hx,",
	       rx, key_serial(key), trans->debug_id, ntohs(service_id));

	if (rx->trans == trans && rx->bundle) {
		atomic_inc(&rx->bundle->usage);
		return rx->bundle;
	}

	/* search the extant bundles first for one that matches the specified
	 * user ID */
	spin_lock(&trans->client_lock);

	p = trans->bundles.rb_node;
	while (p) {
		bundle = rb_entry(p, struct rxrpc_conn_bundle, node);

		if (rxrpc_cmp_bundle(bundle, key, service_id) < 0)
			p = p->rb_left;
		else if (rxrpc_cmp_bundle(bundle, key, service_id) > 0)
			p = p->rb_right;
		else
			goto found_extant_bundle;
	}

	spin_unlock(&trans->client_lock);

	/* not yet present - create a candidate for a new record and then
	 * redo the search */
	candidate = rxrpc_alloc_bundle(gfp);
	if (!candidate) {
		_leave(" = -ENOMEM");
		return ERR_PTR(-ENOMEM);
	}

	candidate->key = key_get(key);
	candidate->service_id = service_id;

	spin_lock(&trans->client_lock);

	pp = &trans->bundles.rb_node;
	parent = NULL;
	while (*pp) {
		parent = *pp;
		bundle = rb_entry(parent, struct rxrpc_conn_bundle, node);

		if (rxrpc_cmp_bundle(bundle, key, service_id) < 0)
			pp = &(*pp)->rb_left;
		else if (rxrpc_cmp_bundle(bundle, key, service_id) > 0)
			pp = &(*pp)->rb_right;
		else
			goto found_extant_second;
	}

	/* second search also failed; add the new bundle */
	bundle = candidate;
	candidate = NULL;

	rb_link_node(&bundle->node, parent, pp);
	rb_insert_color(&bundle->node, &trans->bundles);
	spin_unlock(&trans->client_lock);
	_net("BUNDLE new on trans %d", trans->debug_id);
	if (!rx->bundle && rx->sk.sk_state == RXRPC_CLIENT_CONNECTED) {
		atomic_inc(&bundle->usage);
		rx->bundle = bundle;
	}
	_leave(" = %p [new]", bundle);
	return bundle;

	/* we found the bundle in the list immediately */
found_extant_bundle:
	atomic_inc(&bundle->usage);
	spin_unlock(&trans->client_lock);
	_net("BUNDLE old on trans %d", trans->debug_id);
	if (!rx->bundle && rx->sk.sk_state == RXRPC_CLIENT_CONNECTED) {
		atomic_inc(&bundle->usage);
		rx->bundle = bundle;
	}
	_leave(" = %p [extant %d]", bundle, atomic_read(&bundle->usage));
	return bundle;

	/* we found the bundle on the second time through the list */
found_extant_second:
	atomic_inc(&bundle->usage);
	spin_unlock(&trans->client_lock);
	kfree(candidate);
	_net("BUNDLE old2 on trans %d", trans->debug_id);
	if (!rx->bundle && rx->sk.sk_state == RXRPC_CLIENT_CONNECTED) {
		atomic_inc(&bundle->usage);
		rx->bundle = bundle;
	}
	_leave(" = %p [second %d]", bundle, atomic_read(&bundle->usage));
	return bundle;
}

/*
 * release a bundle
 */
void rxrpc_put_bundle(struct rxrpc_transport *trans,
		      struct rxrpc_conn_bundle *bundle)
{
	_enter("%p,%p{%d}",trans, bundle, atomic_read(&bundle->usage));

	if (atomic_dec_and_lock(&bundle->usage, &trans->client_lock)) {
		_debug("Destroy bundle");
		rb_erase(&bundle->node, &trans->bundles);
		spin_unlock(&trans->client_lock);
		ASSERT(list_empty(&bundle->unused_conns));
		ASSERT(list_empty(&bundle->avail_conns));
		ASSERT(list_empty(&bundle->busy_conns));
		ASSERTCMP(bundle->num_conns, ==, 0);
		key_put(bundle->key);
		kfree(bundle);
	}

	_leave("");
}

/*
 * allocate a new connection
 */
static struct rxrpc_connection *rxrpc_alloc_connection(gfp_t gfp)
{
	struct rxrpc_connection *conn;

	_enter("");

	conn = kzalloc(sizeof(struct rxrpc_connection), gfp);
	if (conn) {
		INIT_WORK(&conn->processor, &rxrpc_process_connection);
		INIT_LIST_HEAD(&conn->bundle_link);
		conn->calls = RB_ROOT;
		skb_queue_head_init(&conn->rx_queue);
		rwlock_init(&conn->lock);
		spin_lock_init(&conn->state_lock);
		atomic_set(&conn->usage, 1);
		conn->debug_id = atomic_inc_return(&rxrpc_debug_id);
		conn->avail_calls = RXRPC_MAXCALLS;
		conn->size_align = 4;
		conn->header_size = sizeof(struct rxrpc_header);
	}

	_leave(" = %p{%d}", conn, conn ? conn->debug_id : 0);
	return conn;
}

/*
 * assign a connection ID to a connection and add it to the transport's
 * connection lookup tree
 * - called with transport client lock held
 */
static void rxrpc_assign_connection_id(struct rxrpc_connection *conn)
{
	struct rxrpc_connection *xconn;
	struct rb_node *parent, **p;
	__be32 epoch;
	u32 real_conn_id;

	_enter("");

	epoch = conn->epoch;

	write_lock_bh(&conn->trans->conn_lock);

	conn->trans->conn_idcounter += RXRPC_CID_INC;
	if (conn->trans->conn_idcounter < RXRPC_CID_INC)
		conn->trans->conn_idcounter = RXRPC_CID_INC;
	real_conn_id = conn->trans->conn_idcounter;

attempt_insertion:
	parent = NULL;
	p = &conn->trans->client_conns.rb_node;

	while (*p) {
		parent = *p;
		xconn = rb_entry(parent, struct rxrpc_connection, node);

		if (epoch < xconn->epoch)
			p = &(*p)->rb_left;
		else if (epoch > xconn->epoch)
			p = &(*p)->rb_right;
		else if (real_conn_id < xconn->real_conn_id)
			p = &(*p)->rb_left;
		else if (real_conn_id > xconn->real_conn_id)
			p = &(*p)->rb_right;
		else
			goto id_exists;
	}

	/* we've found a suitable hole - arrange for this connection to occupy
	 * it */
	rb_link_node(&conn->node, parent, p);
	rb_insert_color(&conn->node, &conn->trans->client_conns);

	conn->real_conn_id = real_conn_id;
	conn->cid = htonl(real_conn_id);
	write_unlock_bh(&conn->trans->conn_lock);
	_leave(" [CONNID %x CID %x]", real_conn_id, ntohl(conn->cid));
	return;

	/* we found a connection with the proposed ID - walk the tree from that
	 * point looking for the next unused ID */
id_exists:
	for (;;) {
		real_conn_id += RXRPC_CID_INC;
		if (real_conn_id < RXRPC_CID_INC) {
			real_conn_id = RXRPC_CID_INC;
			conn->trans->conn_idcounter = real_conn_id;
			goto attempt_insertion;
		}

		parent = rb_next(parent);
		if (!parent)
			goto attempt_insertion;

		xconn = rb_entry(parent, struct rxrpc_connection, node);
		if (epoch < xconn->epoch ||
		    real_conn_id < xconn->real_conn_id)
			goto attempt_insertion;
	}
}

/*
 * add a call to a connection's call-by-ID tree
 */
static void rxrpc_add_call_ID_to_conn(struct rxrpc_connection *conn,
				      struct rxrpc_call *call)
{
	struct rxrpc_call *xcall;
	struct rb_node *parent, **p;
	__be32 call_id;

	write_lock_bh(&conn->lock);

	call_id = call->call_id;
	p = &conn->calls.rb_node;
	parent = NULL;
	while (*p) {
		parent = *p;
		xcall = rb_entry(parent, struct rxrpc_call, conn_node);

		if (call_id < xcall->call_id)
			p = &(*p)->rb_left;
		else if (call_id > xcall->call_id)
			p = &(*p)->rb_right;
		else
			BUG();
	}

	rb_link_node(&call->conn_node, parent, p);
	rb_insert_color(&call->conn_node, &conn->calls);

	write_unlock_bh(&conn->lock);
}

/*
 * connect a call on an exclusive connection
 */
static int rxrpc_connect_exclusive(struct rxrpc_sock *rx,
				   struct rxrpc_transport *trans,
				   __be16 service_id,
				   struct rxrpc_call *call,
				   gfp_t gfp)
{
	struct rxrpc_connection *conn;
	int chan, ret;

	_enter("");

	conn = rx->conn;
	if (!conn) {
		/* not yet present - create a candidate for a new connection
		 * and then redo the check */
		conn = rxrpc_alloc_connection(gfp);
		if (IS_ERR(conn)) {
			_leave(" = %ld", PTR_ERR(conn));
			return PTR_ERR(conn);
		}

		conn->trans = trans;
		conn->bundle = NULL;
		conn->service_id = service_id;
		conn->epoch = rxrpc_epoch;
		conn->in_clientflag = 0;
		conn->out_clientflag = RXRPC_CLIENT_INITIATED;
		conn->cid = 0;
		conn->state = RXRPC_CONN_CLIENT;
		conn->avail_calls = RXRPC_MAXCALLS - 1;
		conn->security_level = rx->min_sec_level;
		conn->key = key_get(rx->key);

		ret = rxrpc_init_client_conn_security(conn);
		if (ret < 0) {
			key_put(conn->key);
			kfree(conn);
			_leave(" = %d [key]", ret);
			return ret;
		}

		write_lock_bh(&rxrpc_connection_lock);
		list_add_tail(&conn->link, &rxrpc_connections);
		write_unlock_bh(&rxrpc_connection_lock);

		spin_lock(&trans->client_lock);
		atomic_inc(&trans->usage);

		_net("CONNECT EXCL new %d on TRANS %d",
		     conn->debug_id, conn->trans->debug_id);

		rxrpc_assign_connection_id(conn);
		rx->conn = conn;
	}

	/* we've got a connection with a free channel and we can now attach the
	 * call to it
	 * - we're holding the transport's client lock
	 * - we're holding a reference on the connection
	 */
	for (chan = 0; chan < RXRPC_MAXCALLS; chan++)
		if (!conn->channels[chan])
			goto found_channel;
	goto no_free_channels;

found_channel:
	atomic_inc(&conn->usage);
	conn->channels[chan] = call;
	call->conn = conn;
	call->channel = chan;
	call->cid = conn->cid | htonl(chan);
	call->call_id = htonl(++conn->call_counter);

	_net("CONNECT client on conn %d chan %d as call %x",
	     conn->debug_id, chan, ntohl(call->call_id));

	spin_unlock(&trans->client_lock);

	rxrpc_add_call_ID_to_conn(conn, call);
	_leave(" = 0");
	return 0;

no_free_channels:
	spin_unlock(&trans->client_lock);
	_leave(" = -ENOSR");
	return -ENOSR;
}

/*
 * find a connection for a call
 * - called in process context with IRQs enabled
 */
int rxrpc_connect_call(struct rxrpc_sock *rx,
		       struct rxrpc_transport *trans,
		       struct rxrpc_conn_bundle *bundle,
		       struct rxrpc_call *call,
		       gfp_t gfp)
{
	struct rxrpc_connection *conn, *candidate;
	int chan, ret;

	DECLARE_WAITQUEUE(myself, current);

	_enter("%p,%lx,", rx, call->user_call_ID);

	if (test_bit(RXRPC_SOCK_EXCLUSIVE_CONN, &rx->flags))
		return rxrpc_connect_exclusive(rx, trans, bundle->service_id,
					       call, gfp);

	spin_lock(&trans->client_lock);
	for (;;) {
		/* see if the bundle has a call slot available */
		if (!list_empty(&bundle->avail_conns)) {
			_debug("avail");
			conn = list_entry(bundle->avail_conns.next,
					  struct rxrpc_connection,
					  bundle_link);
			if (--conn->avail_calls == 0)
				list_move(&conn->bundle_link,
					  &bundle->busy_conns);
			ASSERTCMP(conn->avail_calls, <, RXRPC_MAXCALLS);
			ASSERT(conn->channels[0] == NULL ||
			       conn->channels[1] == NULL ||
			       conn->channels[2] == NULL ||
			       conn->channels[3] == NULL);
			atomic_inc(&conn->usage);
			break;