[people/ms/linux.git] / net / rds / connection.c

/*
 * Copyright (c) 2006 Oracle.  All rights reserved.
 *
 * This software is available to you under a choice of one of two
 * licenses.  You may choose to be licensed under the terms of the GNU
 * General Public License (GPL) Version 2, available from the file
 * COPYING in the main directory of this source tree, or the
 * OpenIB.org BSD license below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 */
#include <linux/kernel.h>
#include <linux/list.h>
#include <linux/slab.h>
#include <linux/export.h>
#include <net/inet_hashtables.h>

#include "rds.h"
#include "loop.h"

#define RDS_CONNECTION_HASH_BITS 12
#define RDS_CONNECTION_HASH_ENTRIES (1 << RDS_CONNECTION_HASH_BITS)
#define RDS_CONNECTION_HASH_MASK (RDS_CONNECTION_HASH_ENTRIES - 1)

/* converting this to RCU is a chore for another day.. */
static DEFINE_SPINLOCK(rds_conn_lock);
static unsigned long rds_conn_count;
static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];
static struct kmem_cache *rds_conn_slab;

static struct hlist_head *rds_conn_bucket(__be32 laddr, __be32 faddr)
{
	static u32 rds_hash_secret __read_mostly;

	unsigned long hash;

	net_get_random_once(&rds_hash_secret, sizeof(rds_hash_secret));

	/* Pass NULL, don't need struct net for hash */
	hash = __inet_ehashfn(be32_to_cpu(laddr), 0,
			      be32_to_cpu(faddr), 0,
			      rds_hash_secret);
	return &rds_conn_hash[hash & RDS_CONNECTION_HASH_MASK];
}

#define rds_conn_info_set(var, test, suffix) do {		\
	if (test)						\
		var |= RDS_INFO_CONNECTION_FLAG_##suffix;	\
} while (0)

/* rcu read lock must be held or the connection spinlock */
static struct rds_connection *rds_conn_lookup(struct net *net,
					      struct hlist_head *head,
					      __be32 laddr, __be32 faddr,
					      struct rds_transport *trans)
{
	struct rds_connection *conn, *ret = NULL;

	hlist_for_each_entry_rcu(conn, head, c_hash_node) {
		if (conn->c_faddr == faddr && conn->c_laddr == laddr &&
		    conn->c_trans == trans && net == rds_conn_net(conn)) {
			ret = conn;
			break;
		}
	}
	rdsdebug("returning conn %p for %pI4 -> %pI4\n", ret,
		 &laddr, &faddr);
	return ret;
}

/*
 * This is called by transports as they're bringing down a connection.
 * It clears partial message state so that the transport can start sending
 * and receiving over this connection again in the future.  It is up to
 * the transport to have serialized this call with its send and recv.
 */
static void rds_conn_reset(struct rds_connection *conn)
{
	rdsdebug("connection %pI4 to %pI4 reset\n",
	  &conn->c_laddr, &conn->c_faddr);

	rds_stats_inc(s_conn_reset);
	rds_send_reset(conn);
	conn->c_flags = 0;

	/* Do not clear next_rx_seq here, else we cannot distinguish
	 * retransmitted packets from new packets, and will hand all
	 * of them to the application. That is not consistent with the
	 * reliability guarantees of RDS. */
}

/*
 * There is only every one 'conn' for a given pair of addresses in the
 * system at a time.  They contain messages to be retransmitted and so
 * span the lifetime of the actual underlying transport connections.
 *
 * For now they are not garbage collected once they're created.  They
 * are torn down as the module is removed, if ever.
 */
static struct rds_connection *__rds_conn_create(struct net *net,
						__be32 laddr, __be32 faddr,
				       struct rds_transport *trans, gfp_t gfp,
				       int is_outgoing)
{
	struct rds_connection *conn, *parent = NULL;
	struct hlist_head *head = rds_conn_bucket(laddr, faddr);
	struct rds_transport *loop_trans;
	unsigned long flags;
	int ret;

	rcu_read_lock();
	conn = rds_conn_lookup(net, head, laddr, faddr, trans);
	if (conn && conn->c_loopback && conn->c_trans != &rds_loop_transport &&
	    laddr == faddr && !is_outgoing) {
		/* This is a looped back IB connection, and we're
		 * called by the code handling the incoming connect.
		 * We need a second connection object into which we
		 * can stick the other QP. */
		parent = conn;
		conn = parent->c_passive;
	}
	rcu_read_unlock();
	if (conn)
		goto out;

	conn = kmem_cache_zalloc(rds_conn_slab, gfp);
	if (!conn) {
		conn = ERR_PTR(-ENOMEM);
		goto out;
	}

	INIT_HLIST_NODE(&conn->c_hash_node);
	conn->c_laddr = laddr;
	conn->c_faddr = faddr;
	spin_lock_init(&conn->c_lock);
	conn->c_next_tx_seq = 1;
	rds_conn_net_set(conn, net);

	init_waitqueue_head(&conn->c_waitq);
	INIT_LIST_HEAD(&conn->c_send_queue);
	INIT_LIST_HEAD(&conn->c_retrans);

	ret = rds_cong_get_maps(conn);
	if (ret) {
		kmem_cache_free(rds_conn_slab, conn);
		conn = ERR_PTR(ret);
		goto out;
	}

	/*
	 * This is where a connection becomes loopback.  If *any* RDS sockets
	 * can bind to the destination address then we'd rather the messages
	 * flow through loopback rather than either transport.
	 */
	loop_trans = rds_trans_get_preferred(net, faddr);
	if (loop_trans) {
		rds_trans_put(loop_trans);
		conn->c_loopback = 1;
		if (is_outgoing && trans->t_prefer_loopback) {
			/* "outgoing" connection - and the transport
			 * says it wants the connection handled by the
			 * loopback transport. This is what TCP does.
			 */
			trans = &rds_loop_transport;
		}
	}

	conn->c_trans = trans;

	ret = trans->conn_alloc(conn, gfp);
	if (ret) {
		kmem_cache_free(rds_conn_slab, conn);
		conn = ERR_PTR(ret);
		goto out;
	}

	atomic_set(&conn->c_state, RDS_CONN_DOWN);
	conn->c_send_gen = 0;
	conn->c_outgoing = (is_outgoing ? 1 : 0);
	conn->c_reconnect_jiffies = 0;
	INIT_DELAYED_WORK(&conn->c_send_w, rds_send_worker);
	INIT_DELAYED_WORK(&conn->c_recv_w, rds_recv_worker);
	INIT_DELAYED_WORK(&conn->c_conn_w, rds_connect_worker);
	INIT_WORK(&conn->c_down_w, rds_shutdown_worker);
	mutex_init(&conn->c_cm_lock);
	conn->c_flags = 0;

	rdsdebug("allocated conn %p for %pI4 -> %pI4 over %s %s\n",
	  conn, &laddr, &faddr,
	  trans->t_name ? trans->t_name : "[unknown]",
	  is_outgoing ? "(outgoing)" : "");

	/*
	 * Since we ran without holding the conn lock, someone could
	 * have created the same conn (either normal or passive) in the
	 * interim. We check while holding the lock. If we won, we complete
	 * init and return our conn. If we lost, we rollback and return the
	 * other one.
	 */
	spin_lock_irqsave(&rds_conn_lock, flags);
	if (parent) {
		/* Creating passive conn */
		if (parent->c_passive) {
			trans->conn_free(conn->c_transport_data);
			kmem_cache_free(rds_conn_slab, conn);
			conn = parent->c_passive;
		} else {
			parent->c_passive = conn;
			rds_cong_add_conn(conn);
			rds_conn_count++;
		}
	} else {
		/* Creating normal conn */
		struct rds_connection *found;

		found = rds_conn_lookup(net, head, laddr, faddr, trans);
		if (found) {
			trans->conn_free(conn->c_transport_data);
			kmem_cache_free(rds_conn_slab, conn);
			conn = found;
		} else {
			hlist_add_head_rcu(&conn->c_hash_node, head);
			rds_cong_add_conn(conn);
			rds_conn_count++;
		}
	}
	spin_unlock_irqrestore(&rds_conn_lock, flags);

out:
	return conn;
}

struct rds_connection *rds_conn_create(struct net *net,
				       __be32 laddr, __be32 faddr,
				       struct rds_transport *trans, gfp_t gfp)
{
	return __rds_conn_create(net, laddr, faddr, trans, gfp, 0);
}
EXPORT_SYMBOL_GPL(rds_conn_create);

struct rds_connection *rds_conn_create_outgoing(struct net *net,
						__be32 laddr, __be32 faddr,
				       struct rds_transport *trans, gfp_t gfp)
{
	return __rds_conn_create(net, laddr, faddr, trans, gfp, 1);
}
EXPORT_SYMBOL_GPL(rds_conn_create_outgoing);

void rds_conn_shutdown(struct rds_connection *conn)
{
	/* shut it down unless it's down already */
	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_DOWN)) {
		/*
		 * Quiesce the connection mgmt handlers before we start tearing
		 * things down. We don't hold the mutex for the entire
		 * duration of the shutdown operation, else we may be
		 * deadlocking with the CM handler. Instead, the CM event
		 * handler is supposed to check for state DISCONNECTING
		 */
		mutex_lock(&conn->c_cm_lock);
		if (!rds_conn_transition(conn, RDS_CONN_UP, RDS_CONN_DISCONNECTING)
		 && !rds_conn_transition(conn, RDS_CONN_ERROR, RDS_CONN_DISCONNECTING)) {
			rds_conn_error(conn, "shutdown called in state %d\n",
					atomic_read(&conn->c_state));
			mutex_unlock(&conn->c_cm_lock);
			return;
		}
		mutex_unlock(&conn->c_cm_lock);

		wait_event(conn->c_waitq,
			   !test_bit(RDS_IN_XMIT, &conn->c_flags));
		wait_event(conn->c_waitq,
			   !test_bit(RDS_RECV_REFILL, &conn->c_flags));

		conn->c_trans->conn_shutdown(conn);
		rds_conn_reset(conn);

		if (!rds_conn_transition(conn, RDS_CONN_DISCONNECTING, RDS_CONN_DOWN)) {
			/* This can happen - eg when we're in the middle of tearing
			 * down the connection, and someone unloads the rds module.
			 * Quite reproduceable with loopback connections.
			 * Mostly harmless.
			 */
			rds_conn_error(conn,
				"%s: failed to transition to state DOWN, "
				"current state is %d\n",
				__func__,
				atomic_read(&conn->c_state));
			return;
		}
	}

	/* Then reconnect if it's still live.
	 * The passive side of an IB loopback connection is never added
	 * to the conn hash, so we never trigger a reconnect on this
	 * conn - the reconnect is always triggered by the active peer. */
	cancel_delayed_work_sync(&conn->c_conn_w);
	rcu_read_lock();
	if (!hlist_unhashed(&conn->c_hash_node)) {
		rcu_read_unlock();
		if (conn->c_trans->t_type != RDS_TRANS_TCP ||
		    conn->c_outgoing == 1)
			rds_queue_reconnect(conn);
	} else {
		rcu_read_unlock();
	}
}

/*
 * Stop and free a connection.
 *
 * This can only be used in very limited circumstances.  It assumes that once
 * the conn has been shutdown that no one else is referencing the connection.
 * We can only ensure this in the rmmod path in the current code.
 */
void rds_conn_destroy(struct rds_connection *conn)
{
	struct rds_message *rm, *rtmp;
	unsigned long flags;

	rdsdebug("freeing conn %p for %pI4 -> "
		 "%pI4\n", conn, &conn->c_laddr,
		 &conn->c_faddr);

	/* Ensure conn will not be scheduled for reconnect */
	spin_lock_irq(&rds_conn_lock);
	hlist_del_init_rcu(&conn->c_hash_node);
	spin_unlock_irq(&rds_conn_lock);
	synchronize_rcu();

	/* shut the connection down */
	rds_conn_drop(conn);
	flush_work(&conn->c_down_w);

	/* make sure lingering queued work won't try to ref the conn */
	cancel_delayed_work_sync(&conn->c_send_w);
	cancel_delayed_work_sync(&conn->c_recv_w);

	/* tear down queued messages */
	list_for_each_entry_safe(rm, rtmp,
				 &conn->c_send_queue,
				 m_conn_item) {
		list_del_init(&rm->m_conn_item);
		BUG_ON(!list_empty(&rm->m_sock_item));
		rds_message_put(rm);
	}
	if (conn->c_xmit_rm)
		rds_message_put(conn->c_xmit_rm);

	conn->c_trans->conn_free(conn->c_transport_data);

	/*
	 * The congestion maps aren't freed up here.  They're
	 * freed by rds_cong_exit() after all the connections
	 * have been freed.
	 */
	rds_cong_remove_conn(conn);

	BUG_ON(!list_empty(&conn->c_retrans));
	kmem_cache_free(rds_conn_slab, conn);

	spin_lock_irqsave(&rds_conn_lock, flags);
	rds_conn_count--;
	spin_unlock_irqrestore(&rds_conn_lock, flags);
}
EXPORT_SYMBOL_GPL(rds_conn_destroy);

static void rds_conn_message_info(struct socket *sock, unsigned int len,
				  struct rds_info_iterator *iter,
				  struct rds_info_lengths *lens,
				  int want_send)
{
	struct hlist_head *head;
	struct list_head *list;
	struct rds_connection *conn;
	struct rds_message *rm;
	unsigned int total = 0;
	unsigned long flags;
	size_t i;

	len /= sizeof(struct rds_info_message);

	rcu_read_lock();

	for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
	     i++, head++) {
		hlist_for_each_entry_rcu(conn, head, c_hash_node) {
			if (want_send)
				list = &conn->c_send_queue;
			else
				list = &conn->c_retrans;

			spin_lock_irqsave(&conn->c_lock, flags);

			/* XXX too lazy to maintain counts.. */
			list_for_each_entry(rm, list, m_conn_item) {
				total++;
				if (total <= len)
					rds_inc_info_copy(&rm->m_inc, iter,
							  conn->c_laddr,
							  conn->c_faddr, 0);
			}

			spin_unlock_irqrestore(&conn->c_lock, flags);
		}
	}
	rcu_read_unlock();

	lens->nr = total;
	lens->each = sizeof(struct rds_info_message);
}

static void rds_conn_message_info_send(struct socket *sock, unsigned int len,
				       struct rds_info_iterator *iter,
				       struct rds_info_lengths *lens)
{
	rds_conn_message_info(sock, len, iter, lens, 1);
}

static void rds_conn_message_info_retrans(struct socket *sock,
					  unsigned int len,
					  struct rds_info_iterator *iter,
					  struct rds_info_lengths *lens)
{
	rds_conn_message_info(sock, len, iter, lens, 0);
}

void rds_for_each_conn_info(struct socket *sock, unsigned int len,
			  struct rds_info_iterator *iter,
			  struct rds_info_lengths *lens,
			  int (*visitor)(struct rds_connection *, void *),
			  size_t item_len)
{
	uint64_t buffer[(item_len + 7) / 8];
	struct hlist_head *head;
	struct rds_connection *conn;
	size_t i;

	rcu_read_lock();

	lens->nr = 0;
	lens->each = item_len;

	for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
	     i++, head++) {
		hlist_for_each_entry_rcu(conn, head, c_hash_node) {

			/* XXX no c_lock usage.. */
			if (!visitor(conn, buffer))
				continue;

			/* We copy as much as we can fit in the buffer,
			 * but we count all items so that the caller
			 * can resize the buffer. */
			if (len >= item_len) {
				rds_info_copy(iter, buffer, item_len);
				len -= item_len;
			}
			lens->nr++;
		}
	}
	rcu_read_unlock();
}
EXPORT_SYMBOL_GPL(rds_for_each_conn_info);

static int rds_conn_info_visitor(struct rds_connection *conn,
				  void *buffer)
{
	struct rds_info_connection *cinfo = buffer;

	cinfo->next_tx_seq = conn->c_next_tx_seq;
	cinfo->next_rx_seq = conn->c_next_rx_seq;
	cinfo->laddr = conn->c_laddr;
	cinfo->faddr = conn->c_faddr;
	strncpy(cinfo->transport, conn->c_trans->t_name,
		sizeof(cinfo->transport));
	cinfo->flags = 0;

	rds_conn_info_set(cinfo->flags, test_bit(RDS_IN_XMIT, &conn->c_flags),
			  SENDING);
	/* XXX Future: return the state rather than these funky bits */
	rds_conn_info_set(cinfo->flags,
			  atomic_read(&conn->c_state) == RDS_CONN_CONNECTING,
			  CONNECTING);
	rds_conn_info_set(cinfo->flags,
			  atomic_read(&conn->c_state) == RDS_CONN_UP,
			  CONNECTED);
	return 1;
}

static void rds_conn_info(struct socket *sock, unsigned int len,
			  struct rds_info_iterator *iter,
			  struct rds_info_lengths *lens)
{
	rds_for_each_conn_info(sock, len, iter, lens,
				rds_conn_info_visitor,
				sizeof(struct rds_info_connection));
}

int rds_conn_init(void)
{
	rds_conn_slab = kmem_cache_create("rds_connection",
					  sizeof(struct rds_connection),
					  0, 0, NULL);
	if (!rds_conn_slab)
		return -ENOMEM;

	rds_info_register_func(RDS_INFO_CONNECTIONS, rds_conn_info);
	rds_info_register_func(RDS_INFO_SEND_MESSAGES,
			       rds_conn_message_info_send);
	rds_info_register_func(RDS_INFO_RETRANS_MESSAGES,
			       rds_conn_message_info_retrans);

	return 0;
}

void rds_conn_exit(void)
{
	rds_loop_exit();

	WARN_ON(!hlist_empty(rds_conn_hash));

	kmem_cache_destroy(rds_conn_slab);

	rds_info_deregister_func(RDS_INFO_CONNECTIONS, rds_conn_info);
	rds_info_deregister_func(RDS_INFO_SEND_MESSAGES,
				 rds_conn_message_info_send);
	rds_info_deregister_func(RDS_INFO_RETRANS_MESSAGES,
				 rds_conn_message_info_retrans);
}

/*
 * Force a disconnect
 */
void rds_conn_drop(struct rds_connection *conn)
{
	atomic_set(&conn->c_state, RDS_CONN_ERROR);
	queue_work(rds_wq, &conn->c_down_w);
}
EXPORT_SYMBOL_GPL(rds_conn_drop);

/*
 * If the connection is down, trigger a connect. We may have scheduled a
 * delayed reconnect however - in this case we should not interfere.
 */
void rds_conn_connect_if_down(struct rds_connection *conn)
{
	if (rds_conn_state(conn) == RDS_CONN_DOWN &&
	    !test_and_set_bit(RDS_RECONNECT_PENDING, &conn->c_flags))
		queue_delayed_work(rds_wq, &conn->c_conn_w, 0);
}
EXPORT_SYMBOL_GPL(rds_conn_connect_if_down);

/*
 * An error occurred on the connection
 */
void
__rds_conn_error(struct rds_connection *conn, const char *fmt, ...)
{
	va_list ap;

	va_start(ap, fmt);
	vprintk(fmt, ap);
	va_end(ap);

	rds_conn_drop(conn);
}
Commit	Line	Data
00e0f34c AG	1	/*
	2	* Copyright (c) 2006 Oracle. All rights reserved.
	3	*
	4	* This software is available to you under a choice of one of two
	5	* licenses. You may choose to be licensed under the terms of the GNU
	6	* General Public License (GPL) Version 2, available from the file
	7	* COPYING in the main directory of this source tree, or the
	8	* OpenIB.org BSD license below:
	9	*
	10	* Redistribution and use in source and binary forms, with or
	11	* without modification, are permitted provided that the following
	12	* conditions are met:
	13	*
	14	* - Redistributions of source code must retain the above
	15	* copyright notice, this list of conditions and the following
	16	* disclaimer.
	17	*
	18	* - Redistributions in binary form must reproduce the above
	19	* copyright notice, this list of conditions and the following
	20	* disclaimer in the documentation and/or other materials
	21	* provided with the distribution.
	22	*
	23	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
	24	* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
	25	* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
	26	* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
	27	* BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
	28	* ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
	29	* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
	30	* SOFTWARE.
	31	*
	32	*/
	33	#include <linux/kernel.h>
	34	#include <linux/list.h>
5a0e3ad6	35	#include <linux/slab.h>
bc3b2d7f	36	#include <linux/export.h>
00e0f34c AG	37	#include <net/inet_hashtables.h>
	38
	39	#include "rds.h"
	40	#include "loop.h"
00e0f34c AG	41
	42	#define RDS_CONNECTION_HASH_BITS 12
	43	#define RDS_CONNECTION_HASH_ENTRIES (1 << RDS_CONNECTION_HASH_BITS)
	44	#define RDS_CONNECTION_HASH_MASK (RDS_CONNECTION_HASH_ENTRIES - 1)
	45
	46	/* converting this to RCU is a chore for another day.. */
	47	static DEFINE_SPINLOCK(rds_conn_lock);
	48	static unsigned long rds_conn_count;
	49	static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];
	50	static struct kmem_cache *rds_conn_slab;
	51
	52	static struct hlist_head *rds_conn_bucket(__be32 laddr, __be32 faddr)
	53	{
1bbdceef HFS	54	static u32 rds_hash_secret __read_mostly;
	55
	56	unsigned long hash;
	57
	58	net_get_random_once(&rds_hash_secret, sizeof(rds_hash_secret));
	59
00e0f34c	60	/* Pass NULL, don't need struct net for hash */
1bbdceef HFS	61	hash = __inet_ehashfn(be32_to_cpu(laddr), 0,
	62	be32_to_cpu(faddr), 0,
	63	rds_hash_secret);
00e0f34c AG	64	return &rds_conn_hash[hash & RDS_CONNECTION_HASH_MASK];
	65	}
	66
	67	#define rds_conn_info_set(var, test, suffix) do { \
	68	if (test) \
	69	var \|= RDS_INFO_CONNECTION_FLAG_##suffix; \
	70	} while (0)
	71
bcf50ef2	72	/* rcu read lock must be held or the connection spinlock */
8f384c01 SV	73	static struct rds_connection rds_conn_lookup(struct net net,
8f384c01 SV	74	struct hlist_head *head,
00e0f34c AG	75	__be32 laddr, __be32 faddr,
	76	struct rds_transport *trans)
	77	{
	78	struct rds_connection conn, ret = NULL;
00e0f34c	79
b67bfe0d	80	hlist_for_each_entry_rcu(conn, head, c_hash_node) {
00e0f34c	81	if (conn->c_faddr == faddr && conn->c_laddr == laddr &&
8f384c01	82	conn->c_trans == trans && net == rds_conn_net(conn)) {
00e0f34c AG	83	ret = conn;
	84	break;
	85	}
	86	}
	87	rdsdebug("returning conn %p for %pI4 -> %pI4\n", ret,
	88	&laddr, &faddr);
	89	return ret;
	90	}
	91
	92	/*
	93	* This is called by transports as they're bringing down a connection.
	94	* It clears partial message state so that the transport can start sending
	95	* and receiving over this connection again in the future. It is up to
	96	* the transport to have serialized this call with its send and recv.
	97	*/
ff51bf84	98	static void rds_conn_reset(struct rds_connection *conn)
00e0f34c AG	99	{
	100	rdsdebug("connection %pI4 to %pI4 reset\n",
	101	&conn->c_laddr, &conn->c_faddr);
	102
	103	rds_stats_inc(s_conn_reset);
	104	rds_send_reset(conn);
	105	conn->c_flags = 0;
	106
	107	/* Do not clear next_rx_seq here, else we cannot distinguish
	108	* retransmitted packets from new packets, and will hand all
	109	* of them to the application. That is not consistent with the
	110	* reliability guarantees of RDS. */
	111	}
	112
	113	/*
	114	* There is only every one 'conn' for a given pair of addresses in the
	115	* system at a time. They contain messages to be retransmitted and so
	116	* span the lifetime of the actual underlying transport connections.
	117	*
	118	* For now they are not garbage collected once they're created. They
	119	* are torn down as the module is removed, if ever.
	120	*/
d5a8ac28 SV	121	static struct rds_connection __rds_conn_create(struct net net,
d5a8ac28 SV	122	__be32 laddr, __be32 faddr,
00e0f34c AG	123	struct rds_transport *trans, gfp_t gfp,
	124	int is_outgoing)
	125	{
cb24405e	126	struct rds_connection conn, parent = NULL;
00e0f34c	127	struct hlist_head *head = rds_conn_bucket(laddr, faddr);
5adb5bc6	128	struct rds_transport *loop_trans;
00e0f34c AG	129	unsigned long flags;
	130	int ret;
	131
bcf50ef2	132	rcu_read_lock();
8f384c01	133	conn = rds_conn_lookup(net, head, laddr, faddr, trans);
f64f9e71	134	if (conn && conn->c_loopback && conn->c_trans != &rds_loop_transport &&
1789b2c0	135	laddr == faddr && !is_outgoing) {
00e0f34c AG	136	/* This is a looped back IB connection, and we're
	137	* called by the code handling the incoming connect.
	138	* We need a second connection object into which we
	139	* can stick the other QP. */
	140	parent = conn;
	141	conn = parent->c_passive;
	142	}
bcf50ef2	143	rcu_read_unlock();
00e0f34c AG	144	if (conn)
	145	goto out;
	146
05a178ec	147	conn = kmem_cache_zalloc(rds_conn_slab, gfp);
8690bfa1	148	if (!conn) {
00e0f34c AG	149	conn = ERR_PTR(-ENOMEM);
	150	goto out;
	151	}
	152
00e0f34c	153	INIT_HLIST_NODE(&conn->c_hash_node);
00e0f34c AG	154	conn->c_laddr = laddr;
	155	conn->c_faddr = faddr;
	156	spin_lock_init(&conn->c_lock);
	157	conn->c_next_tx_seq = 1;
d5a8ac28	158	rds_conn_net_set(conn, net);
00e0f34c	159
0f4b1c7e	160	init_waitqueue_head(&conn->c_waitq);
00e0f34c AG	161	INIT_LIST_HEAD(&conn->c_send_queue);
	162	INIT_LIST_HEAD(&conn->c_retrans);
	163
	164	ret = rds_cong_get_maps(conn);
	165	if (ret) {
	166	kmem_cache_free(rds_conn_slab, conn);
	167	conn = ERR_PTR(ret);
	168	goto out;
	169	}
	170
	171	/*
	172	* This is where a connection becomes loopback. If any RDS sockets
	173	* can bind to the destination address then we'd rather the messages
	174	* flow through loopback rather than either transport.
	175	*/
d5a8ac28	176	loop_trans = rds_trans_get_preferred(net, faddr);
5adb5bc6 ZB	177	if (loop_trans) {
5adb5bc6 ZB	178	rds_trans_put(loop_trans);
00e0f34c AG	179	conn->c_loopback = 1;
	180	if (is_outgoing && trans->t_prefer_loopback) {
	181	/* "outgoing" connection - and the transport
	182	* says it wants the connection handled by the
	183	* loopback transport. This is what TCP does.
	184	*/
	185	trans = &rds_loop_transport;
	186	}
	187	}
	188
	189	conn->c_trans = trans;
	190
	191	ret = trans->conn_alloc(conn, gfp);
	192	if (ret) {
	193	kmem_cache_free(rds_conn_slab, conn);
	194	conn = ERR_PTR(ret);
	195	goto out;
	196	}
	197
	198	atomic_set(&conn->c_state, RDS_CONN_DOWN);
443be0e5	199	conn->c_send_gen = 0;
3b20fc38	200	conn->c_outgoing = (is_outgoing ? 1 : 0);
00e0f34c AG	201	conn->c_reconnect_jiffies = 0;
	202	INIT_DELAYED_WORK(&conn->c_send_w, rds_send_worker);
	203	INIT_DELAYED_WORK(&conn->c_recv_w, rds_recv_worker);
	204	INIT_DELAYED_WORK(&conn->c_conn_w, rds_connect_worker);
	205	INIT_WORK(&conn->c_down_w, rds_shutdown_worker);
	206	mutex_init(&conn->c_cm_lock);
	207	conn->c_flags = 0;
	208
	209	rdsdebug("allocated conn %p for %pI4 -> %pI4 over %s %s\n",
	210	conn, &laddr, &faddr,
	211	trans->t_name ? trans->t_name : "[unknown]",
	212	is_outgoing ? "(outgoing)" : "");
	213
cb24405e AG	214	/*
	215	* Since we ran without holding the conn lock, someone could
	216	* have created the same conn (either normal or passive) in the
	217	* interim. We check while holding the lock. If we won, we complete
	218	* init and return our conn. If we lost, we rollback and return the
	219	* other one.
	220	*/
00e0f34c	221	spin_lock_irqsave(&rds_conn_lock, flags);
cb24405e AG	222	if (parent) {
	223	/* Creating passive conn */
	224	if (parent->c_passive) {
	225	trans->conn_free(conn->c_transport_data);
	226	kmem_cache_free(rds_conn_slab, conn);
	227	conn = parent->c_passive;
	228	} else {
00e0f34c	229	parent->c_passive = conn;
cb24405e AG	230	rds_cong_add_conn(conn);
	231	rds_conn_count++;
	232	}
00e0f34c	233	} else {
cb24405e AG	234	/* Creating normal conn */
	235	struct rds_connection *found;
	236
3b20fc38	237	found = rds_conn_lookup(net, head, laddr, faddr, trans);
cb24405e AG	238	if (found) {
	239	trans->conn_free(conn->c_transport_data);
	240	kmem_cache_free(rds_conn_slab, conn);
	241	conn = found;
	242	} else {
3b20fc38	243	hlist_add_head_rcu(&conn->c_hash_node, head);
cb24405e AG	244	rds_cong_add_conn(conn);
	245	rds_conn_count++;
	246	}
00e0f34c	247	}
00e0f34c AG	248	spin_unlock_irqrestore(&rds_conn_lock, flags);
	249
	250	out:
	251	return conn;
	252	}
	253
d5a8ac28 SV	254	struct rds_connection rds_conn_create(struct net net,
d5a8ac28 SV	255	__be32 laddr, __be32 faddr,
00e0f34c AG	256	struct rds_transport *trans, gfp_t gfp)
00e0f34c AG	257	{
d5a8ac28	258	return __rds_conn_create(net, laddr, faddr, trans, gfp, 0);
00e0f34c	259	}
616b757a	260	EXPORT_SYMBOL_GPL(rds_conn_create);
00e0f34c	261
d5a8ac28 SV	262	struct rds_connection rds_conn_create_outgoing(struct net net,
d5a8ac28 SV	263	__be32 laddr, __be32 faddr,
00e0f34c AG	264	struct rds_transport *trans, gfp_t gfp)
00e0f34c AG	265	{
d5a8ac28	266	return __rds_conn_create(net, laddr, faddr, trans, gfp, 1);
00e0f34c	267	}
616b757a	268	EXPORT_SYMBOL_GPL(rds_conn_create_outgoing);
00e0f34c	269
2dc39357 AG	270	void rds_conn_shutdown(struct rds_connection *conn)
	271	{
	272	/* shut it down unless it's down already */
	273	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_DOWN)) {
	274	/*
	275	* Quiesce the connection mgmt handlers before we start tearing
	276	* things down. We don't hold the mutex for the entire
	277	* duration of the shutdown operation, else we may be
	278	* deadlocking with the CM handler. Instead, the CM event
	279	* handler is supposed to check for state DISCONNECTING
	280	*/
	281	mutex_lock(&conn->c_cm_lock);
	282	if (!rds_conn_transition(conn, RDS_CONN_UP, RDS_CONN_DISCONNECTING)
	283	&& !rds_conn_transition(conn, RDS_CONN_ERROR, RDS_CONN_DISCONNECTING)) {
	284	rds_conn_error(conn, "shutdown called in state %d\n",
	285	atomic_read(&conn->c_state));
	286	mutex_unlock(&conn->c_cm_lock);
	287	return;
	288	}
	289	mutex_unlock(&conn->c_cm_lock);
	290
0f4b1c7e ZB	291	wait_event(conn->c_waitq,
0f4b1c7e ZB	292	!test_bit(RDS_IN_XMIT, &conn->c_flags));
73ce4317	293	wait_event(conn->c_waitq,
73ce4317	294	!test_bit(RDS_RECV_REFILL, &conn->c_flags));
7e3f2952	295
2dc39357 AG	296	conn->c_trans->conn_shutdown(conn);
2dc39357 AG	297	rds_conn_reset(conn);
2dc39357 AG	298
	299	if (!rds_conn_transition(conn, RDS_CONN_DISCONNECTING, RDS_CONN_DOWN)) {
	300	/* This can happen - eg when we're in the middle of tearing
	301	* down the connection, and someone unloads the rds module.
	302	* Quite reproduceable with loopback connections.
	303	* Mostly harmless.
	304	*/
	305	rds_conn_error(conn,
	306	"%s: failed to transition to state DOWN, "
	307	"current state is %d\n",
	308	__func__,
	309	atomic_read(&conn->c_state));
	310	return;
	311	}
	312	}
	313
	314	/* Then reconnect if it's still live.
	315	* The passive side of an IB loopback connection is never added
	316	* to the conn hash, so we never trigger a reconnect on this
	317	* conn - the reconnect is always triggered by the active peer. */
	318	cancel_delayed_work_sync(&conn->c_conn_w);
bcf50ef2 CM	319	rcu_read_lock();
	320	if (!hlist_unhashed(&conn->c_hash_node)) {
	321	rcu_read_unlock();
3b20fc38 SV	322	if (conn->c_trans->t_type != RDS_TRANS_TCP \|\|
	323	conn->c_outgoing == 1)
	324	rds_queue_reconnect(conn);
bcf50ef2 CM	325	} else {
	326	rcu_read_unlock();
	327	}
2dc39357 AG	328	}
	329
	330	/*
	331	* Stop and free a connection.
ffcec0e1 ZB	332	*
	333	* This can only be used in very limited circumstances. It assumes that once
	334	* the conn has been shutdown that no one else is referencing the connection.
	335	* We can only ensure this in the rmmod path in the current code.
2dc39357	336	*/
00e0f34c AG	337	void rds_conn_destroy(struct rds_connection *conn)
	338	{
	339	struct rds_message rm, rtmp;
fe8ff6b5	340	unsigned long flags;
00e0f34c AG	341
	342	rdsdebug("freeing conn %p for %pI4 -> "
	343	"%pI4\n", conn, &conn->c_laddr,
	344	&conn->c_faddr);
	345
abf45439 CM	346	/* Ensure conn will not be scheduled for reconnect */
abf45439 CM	347	spin_lock_irq(&rds_conn_lock);
bcf50ef2	348	hlist_del_init_rcu(&conn->c_hash_node);
abf45439	349	spin_unlock_irq(&rds_conn_lock);
bcf50ef2 CM	350	synchronize_rcu();
bcf50ef2 CM	351
ffcec0e1 ZB	352	/* shut the connection down */
	353	rds_conn_drop(conn);
	354	flush_work(&conn->c_down_w);
00e0f34c	355
4518071a ZB	356	/* make sure lingering queued work won't try to ref the conn */
	357	cancel_delayed_work_sync(&conn->c_send_w);
	358	cancel_delayed_work_sync(&conn->c_recv_w);
	359
00e0f34c AG	360	/* tear down queued messages */
	361	list_for_each_entry_safe(rm, rtmp,
	362	&conn->c_send_queue,
	363	m_conn_item) {
	364	list_del_init(&rm->m_conn_item);
	365	BUG_ON(!list_empty(&rm->m_sock_item));
	366	rds_message_put(rm);
	367	}
	368	if (conn->c_xmit_rm)
	369	rds_message_put(conn->c_xmit_rm);
	370
	371	conn->c_trans->conn_free(conn->c_transport_data);
	372
	373	/*
	374	* The congestion maps aren't freed up here. They're
	375	* freed by rds_cong_exit() after all the connections
	376	* have been freed.
	377	*/
	378	rds_cong_remove_conn(conn);
	379
	380	BUG_ON(!list_empty(&conn->c_retrans));
	381	kmem_cache_free(rds_conn_slab, conn);
	382
fe8ff6b5	383	spin_lock_irqsave(&rds_conn_lock, flags);
00e0f34c	384	rds_conn_count--;
fe8ff6b5	385	spin_unlock_irqrestore(&rds_conn_lock, flags);
00e0f34c	386	}
616b757a	387	EXPORT_SYMBOL_GPL(rds_conn_destroy);
00e0f34c AG	388
	389	static void rds_conn_message_info(struct socket *sock, unsigned int len,
	390	struct rds_info_iterator *iter,
	391	struct rds_info_lengths *lens,
	392	int want_send)
	393	{
	394	struct hlist_head *head;
00e0f34c AG	395	struct list_head *list;
	396	struct rds_connection *conn;
	397	struct rds_message *rm;
00e0f34c	398	unsigned int total = 0;
501dcccd	399	unsigned long flags;
00e0f34c AG	400	size_t i;
	401
	402	len /= sizeof(struct rds_info_message);
	403
bcf50ef2	404	rcu_read_lock();
00e0f34c AG	405
	406	for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
	407	i++, head++) {
b67bfe0d	408	hlist_for_each_entry_rcu(conn, head, c_hash_node) {
00e0f34c AG	409	if (want_send)
	410	list = &conn->c_send_queue;
	411	else
	412	list = &conn->c_retrans;
	413
501dcccd	414	spin_lock_irqsave(&conn->c_lock, flags);
00e0f34c AG	415
	416	/* XXX too lazy to maintain counts.. */
	417	list_for_each_entry(rm, list, m_conn_item) {
	418	total++;
	419	if (total <= len)
	420	rds_inc_info_copy(&rm->m_inc, iter,
	421	conn->c_laddr,
	422	conn->c_faddr, 0);
	423	}
	424
501dcccd	425	spin_unlock_irqrestore(&conn->c_lock, flags);
00e0f34c AG	426	}
00e0f34c AG	427	}
bcf50ef2	428	rcu_read_unlock();
00e0f34c AG	429
	430	lens->nr = total;
	431	lens->each = sizeof(struct rds_info_message);
	432	}
	433
	434	static void rds_conn_message_info_send(struct socket *sock, unsigned int len,
	435	struct rds_info_iterator *iter,
	436	struct rds_info_lengths *lens)
	437	{
	438	rds_conn_message_info(sock, len, iter, lens, 1);
	439	}
	440
	441	static void rds_conn_message_info_retrans(struct socket *sock,
	442	unsigned int len,
	443	struct rds_info_iterator *iter,
	444	struct rds_info_lengths *lens)
	445	{
	446	rds_conn_message_info(sock, len, iter, lens, 0);
	447	}
	448
	449	void rds_for_each_conn_info(struct socket *sock, unsigned int len,
	450	struct rds_info_iterator *iter,
	451	struct rds_info_lengths *lens,
	452	int (visitor)(struct rds_connection , void *),
	453	size_t item_len)
	454	{
	455	uint64_t buffer[(item_len + 7) / 8];
	456	struct hlist_head *head;
00e0f34c	457	struct rds_connection *conn;
00e0f34c AG	458	size_t i;
00e0f34c AG	459
bcf50ef2	460	rcu_read_lock();
00e0f34c AG	461
	462	lens->nr = 0;
	463	lens->each = item_len;
	464
	465	for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
	466	i++, head++) {
b67bfe0d	467	hlist_for_each_entry_rcu(conn, head, c_hash_node) {
00e0f34c AG	468
	469	/* XXX no c_lock usage.. */
	470	if (!visitor(conn, buffer))
	471	continue;
	472
	473	/* We copy as much as we can fit in the buffer,
	474	* but we count all items so that the caller
	475	* can resize the buffer. */
	476	if (len >= item_len) {
	477	rds_info_copy(iter, buffer, item_len);
	478	len -= item_len;
	479	}
	480	lens->nr++;
	481	}
	482	}
bcf50ef2	483	rcu_read_unlock();
00e0f34c	484	}
616b757a	485	EXPORT_SYMBOL_GPL(rds_for_each_conn_info);
00e0f34c AG	486
	487	static int rds_conn_info_visitor(struct rds_connection *conn,
	488	void *buffer)
	489	{
	490	struct rds_info_connection *cinfo = buffer;
	491
	492	cinfo->next_tx_seq = conn->c_next_tx_seq;
	493	cinfo->next_rx_seq = conn->c_next_rx_seq;
	494	cinfo->laddr = conn->c_laddr;
	495	cinfo->faddr = conn->c_faddr;
	496	strncpy(cinfo->transport, conn->c_trans->t_name,
	497	sizeof(cinfo->transport));
	498	cinfo->flags = 0;
	499
0f4b1c7e ZB	500	rds_conn_info_set(cinfo->flags, test_bit(RDS_IN_XMIT, &conn->c_flags),
0f4b1c7e ZB	501	SENDING);
00e0f34c AG	502	/* XXX Future: return the state rather than these funky bits */
	503	rds_conn_info_set(cinfo->flags,
	504	atomic_read(&conn->c_state) == RDS_CONN_CONNECTING,
	505	CONNECTING);
	506	rds_conn_info_set(cinfo->flags,
	507	atomic_read(&conn->c_state) == RDS_CONN_UP,
	508	CONNECTED);
	509	return 1;
	510	}
	511
	512	static void rds_conn_info(struct socket *sock, unsigned int len,
	513	struct rds_info_iterator *iter,
	514	struct rds_info_lengths *lens)
	515	{
	516	rds_for_each_conn_info(sock, len, iter, lens,
	517	rds_conn_info_visitor,
	518	sizeof(struct rds_info_connection));
	519	}
	520
ef87b7ea	521	int rds_conn_init(void)
00e0f34c AG	522	{
	523	rds_conn_slab = kmem_cache_create("rds_connection",
	524	sizeof(struct rds_connection),
	525	0, 0, NULL);
8690bfa1	526	if (!rds_conn_slab)
00e0f34c AG	527	return -ENOMEM;
	528
	529	rds_info_register_func(RDS_INFO_CONNECTIONS, rds_conn_info);
	530	rds_info_register_func(RDS_INFO_SEND_MESSAGES,
	531	rds_conn_message_info_send);
	532	rds_info_register_func(RDS_INFO_RETRANS_MESSAGES,
	533	rds_conn_message_info_retrans);
	534
	535	return 0;
	536	}
	537
	538	void rds_conn_exit(void)
	539	{
	540	rds_loop_exit();
	541
	542	WARN_ON(!hlist_empty(rds_conn_hash));
	543
	544	kmem_cache_destroy(rds_conn_slab);
	545
	546	rds_info_deregister_func(RDS_INFO_CONNECTIONS, rds_conn_info);
	547	rds_info_deregister_func(RDS_INFO_SEND_MESSAGES,
	548	rds_conn_message_info_send);
	549	rds_info_deregister_func(RDS_INFO_RETRANS_MESSAGES,
	550	rds_conn_message_info_retrans);
	551	}
	552
	553	/*
	554	* Force a disconnect
	555	*/
	556	void rds_conn_drop(struct rds_connection *conn)
	557	{
	558	atomic_set(&conn->c_state, RDS_CONN_ERROR);
	559	queue_work(rds_wq, &conn->c_down_w);
	560	}
616b757a	561	EXPORT_SYMBOL_GPL(rds_conn_drop);
00e0f34c	562
f3c6808d ZB	563	/*
	564	* If the connection is down, trigger a connect. We may have scheduled a
	565	* delayed reconnect however - in this case we should not interfere.
	566	*/
	567	void rds_conn_connect_if_down(struct rds_connection *conn)
	568	{
	569	if (rds_conn_state(conn) == RDS_CONN_DOWN &&
	570	!test_and_set_bit(RDS_RECONNECT_PENDING, &conn->c_flags))
	571	queue_delayed_work(rds_wq, &conn->c_conn_w, 0);
	572	}
	573	EXPORT_SYMBOL_GPL(rds_conn_connect_if_down);
	574
00e0f34c AG	575	/*
	576	* An error occurred on the connection
	577	*/
	578	void
	579	__rds_conn_error(struct rds_connection conn, const char fmt, ...)
	580	{
	581	va_list ap;
	582
	583	va_start(ap, fmt);
	584	vprintk(fmt, ap);
	585	va_end(ap);
	586
	587	rds_conn_drop(conn);
	588	}