RDS: Drop stale iWARP RDMA transport

RDS is not Infiniband-specific; it was designed to support different

transports.  The current implementation used to support RDS over TCP as well

as IB. Work is in progress to support RDS over iWARP, and using DCE to

guarantee no dropped packets on Ethernet, it may be possible to use RDS over

UDP in the future.

as IB.

The high-level semantics of RDS from the application's point of view are

	depends on INET

	---help---

	  The RDS (Reliable Datagram Sockets) protocol provides reliable,

	  sequenced delivery of datagrams over Infiniband, iWARP,

	  or TCP.

	  sequenced delivery of datagrams over Infiniband or TCP.

config RDS_RDMA

	tristate "RDS over Infiniband and iWARP"

	tristate "RDS over Infiniband"

	depends on RDS && INFINIBAND && INFINIBAND_ADDR_TRANS

	---help---

	  Allow RDS to use Infiniband and iWARP as a transport.

	  Allow RDS to use Infiniband as a transport.

	  This transport supports RDMA operations.

config RDS_TCP

obj-$(CONFIG_RDS_RDMA) += rds_rdma.o

rds_rdma-y :=	rdma_transport.o \

			ib.o ib_cm.o ib_recv.o ib_ring.o ib_send.o ib_stats.o \

			ib_sysctl.o ib_rdma.o \

			iw.o iw_cm.o iw_recv.o iw_ring.o iw_send.o iw_stats.o \

			iw_sysctl.o iw_rdma.o

			ib_sysctl.o ib_rdma.o

obj-$(CONFIG_RDS_TCP) += rds_tcp.o

/*

 * 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/in.h>

#include <linux/if.h>

#include <linux/netdevice.h>

#include <linux/inetdevice.h>

#include <linux/if_arp.h>

#include <linux/delay.h>

#include <linux/slab.h>

#include <linux/module.h>

#include "rds.h"

#include "iw.h"

unsigned int fastreg_pool_size = RDS_FASTREG_POOL_SIZE;

unsigned int fastreg_message_size = RDS_FASTREG_SIZE + 1; /* +1 allows for unaligned MRs */

module_param(fastreg_pool_size, int, 0444);

MODULE_PARM_DESC(fastreg_pool_size, " Max number of fastreg MRs per device");

module_param(fastreg_message_size, int, 0444);

MODULE_PARM_DESC(fastreg_message_size, " Max size of a RDMA transfer (fastreg MRs)");

struct list_head rds_iw_devices;

/* NOTE: if also grabbing iwdev lock, grab this first */

DEFINE_SPINLOCK(iw_nodev_conns_lock);

LIST_HEAD(iw_nodev_conns);

static void rds_iw_add_one(struct ib_device *device)

{

	struct rds_iw_device *rds_iwdev;

	/* Only handle iwarp devices */

	if (device->node_type != RDMA_NODE_RNIC)

		return;

	rds_iwdev = kmalloc(sizeof *rds_iwdev, GFP_KERNEL);

	if (!rds_iwdev)

		return;

	spin_lock_init(&rds_iwdev->spinlock);

	rds_iwdev->dma_local_lkey = !!(device->attrs.device_cap_flags & IB_DEVICE_LOCAL_DMA_LKEY);

	rds_iwdev->max_wrs = device->attrs.max_qp_wr;

	rds_iwdev->max_sge = min(device->attrs.max_sge, RDS_IW_MAX_SGE);

	rds_iwdev->dev = device;

	rds_iwdev->pd = ib_alloc_pd(device);

	if (IS_ERR(rds_iwdev->pd))

		goto free_dev;

	if (!rds_iwdev->dma_local_lkey) {

		rds_iwdev->mr = ib_get_dma_mr(rds_iwdev->pd,

					IB_ACCESS_REMOTE_READ |

					IB_ACCESS_REMOTE_WRITE |

					IB_ACCESS_LOCAL_WRITE);

		if (IS_ERR(rds_iwdev->mr))

			goto err_pd;

	} else

		rds_iwdev->mr = NULL;

	rds_iwdev->mr_pool = rds_iw_create_mr_pool(rds_iwdev);

	if (IS_ERR(rds_iwdev->mr_pool)) {

		rds_iwdev->mr_pool = NULL;

		goto err_mr;

	}

	INIT_LIST_HEAD(&rds_iwdev->cm_id_list);

	INIT_LIST_HEAD(&rds_iwdev->conn_list);

	list_add_tail(&rds_iwdev->list, &rds_iw_devices);

	ib_set_client_data(device, &rds_iw_client, rds_iwdev);

	return;

err_mr:

	if (rds_iwdev->mr)

		ib_dereg_mr(rds_iwdev->mr);

err_pd:

	ib_dealloc_pd(rds_iwdev->pd);

free_dev:

	kfree(rds_iwdev);

}

static void rds_iw_remove_one(struct ib_device *device, void *client_data)

{

	struct rds_iw_device *rds_iwdev = client_data;

	struct rds_iw_cm_id *i_cm_id, *next;

	if (!rds_iwdev)

		return;

	spin_lock_irq(&rds_iwdev->spinlock);

	list_for_each_entry_safe(i_cm_id, next, &rds_iwdev->cm_id_list, list) {

		list_del(&i_cm_id->list);

		kfree(i_cm_id);

	}

	spin_unlock_irq(&rds_iwdev->spinlock);

	rds_iw_destroy_conns(rds_iwdev);

	if (rds_iwdev->mr_pool)

		rds_iw_destroy_mr_pool(rds_iwdev->mr_pool);

	if (rds_iwdev->mr)

		ib_dereg_mr(rds_iwdev->mr);

	ib_dealloc_pd(rds_iwdev->pd);

	list_del(&rds_iwdev->list);

	kfree(rds_iwdev);

}

struct ib_client rds_iw_client = {

	.name   = "rds_iw",

	.add    = rds_iw_add_one,

	.remove = rds_iw_remove_one

};

static int rds_iw_conn_info_visitor(struct rds_connection *conn,

				    void *buffer)

{

	struct rds_info_rdma_connection *iinfo = buffer;

	struct rds_iw_connection *ic;

	/* We will only ever look at IB transports */

	if (conn->c_trans != &rds_iw_transport)

		return 0;

	iinfo->src_addr = conn->c_laddr;

	iinfo->dst_addr = conn->c_faddr;

	memset(&iinfo->src_gid, 0, sizeof(iinfo->src_gid));

	memset(&iinfo->dst_gid, 0, sizeof(iinfo->dst_gid));

	if (rds_conn_state(conn) == RDS_CONN_UP) {

		struct rds_iw_device *rds_iwdev;

		struct rdma_dev_addr *dev_addr;

		ic = conn->c_transport_data;

		dev_addr = &ic->i_cm_id->route.addr.dev_addr;

		rdma_addr_get_sgid(dev_addr, (union ib_gid *) &iinfo->src_gid);

		rdma_addr_get_dgid(dev_addr, (union ib_gid *) &iinfo->dst_gid);

		rds_iwdev = ib_get_client_data(ic->i_cm_id->device, &rds_iw_client);

		iinfo->max_send_wr = ic->i_send_ring.w_nr;

		iinfo->max_recv_wr = ic->i_recv_ring.w_nr;

		iinfo->max_send_sge = rds_iwdev->max_sge;

		rds_iw_get_mr_info(rds_iwdev, iinfo);

	}

	return 1;

}

static void rds_iw_ic_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_iw_conn_info_visitor,

				sizeof(struct rds_info_rdma_connection));

}

/*

 * Early RDS/IB was built to only bind to an address if there is an IPoIB

 * device with that address set.

 *

 * If it were me, I'd advocate for something more flexible.  Sending and

 * receiving should be device-agnostic.  Transports would try and maintain

 * connections between peers who have messages queued.  Userspace would be

 * allowed to influence which paths have priority.  We could call userspace

 * asserting this policy "routing".

 */

static int rds_iw_laddr_check(struct net *net, __be32 addr)

{

	int ret;

	struct rdma_cm_id *cm_id;

	struct sockaddr_in sin;

	/* Create a CMA ID and try to bind it. This catches both

	 * IB and iWARP capable NICs.

	 */

	cm_id = rdma_create_id(&init_net, NULL, NULL, RDMA_PS_TCP, IB_QPT_RC);

	if (IS_ERR(cm_id))

		return PTR_ERR(cm_id);

	memset(&sin, 0, sizeof(sin));

	sin.sin_family = AF_INET;

	sin.sin_addr.s_addr = addr;

	/* rdma_bind_addr will only succeed for IB & iWARP devices */

	ret = rdma_bind_addr(cm_id, (struct sockaddr *)&sin);

	/* due to this, we will claim to support IB devices unless we

	   check node_type. */

	if (ret || !cm_id->device ||

	    cm_id->device->node_type != RDMA_NODE_RNIC)

		ret = -EADDRNOTAVAIL;

	rdsdebug("addr %pI4 ret %d node type %d\n",

		&addr, ret,

		cm_id->device ? cm_id->device->node_type : -1);

	rdma_destroy_id(cm_id);

	return ret;

}

void rds_iw_exit(void)

{

	rds_info_deregister_func(RDS_INFO_IWARP_CONNECTIONS, rds_iw_ic_info);

	rds_iw_destroy_nodev_conns();

	ib_unregister_client(&rds_iw_client);

	rds_iw_sysctl_exit();

	rds_iw_recv_exit();

	rds_trans_unregister(&rds_iw_transport);

}

struct rds_transport rds_iw_transport = {

	.laddr_check		= rds_iw_laddr_check,

	.xmit_complete		= rds_iw_xmit_complete,

	.xmit			= rds_iw_xmit,

	.xmit_rdma		= rds_iw_xmit_rdma,

	.recv			= rds_iw_recv,

	.conn_alloc		= rds_iw_conn_alloc,

	.conn_free		= rds_iw_conn_free,

	.conn_connect		= rds_iw_conn_connect,

	.conn_shutdown		= rds_iw_conn_shutdown,

	.inc_copy_to_user	= rds_iw_inc_copy_to_user,

	.inc_free		= rds_iw_inc_free,

	.cm_initiate_connect	= rds_iw_cm_initiate_connect,

	.cm_handle_connect	= rds_iw_cm_handle_connect,

	.cm_connect_complete	= rds_iw_cm_connect_complete,

	.stats_info_copy	= rds_iw_stats_info_copy,

	.exit			= rds_iw_exit,

	.get_mr			= rds_iw_get_mr,

	.sync_mr		= rds_iw_sync_mr,

	.free_mr		= rds_iw_free_mr,

	.flush_mrs		= rds_iw_flush_mrs,

	.t_owner		= THIS_MODULE,

	.t_name			= "iwarp",

	.t_type			= RDS_TRANS_IWARP,

	.t_prefer_loopback	= 1,

};

int rds_iw_init(void)

{

	int ret;

	INIT_LIST_HEAD(&rds_iw_devices);

	ret = ib_register_client(&rds_iw_client);

	if (ret)

		goto out;

	ret = rds_iw_sysctl_init();

	if (ret)

		goto out_ibreg;

	ret = rds_iw_recv_init();

	if (ret)

		goto out_sysctl;

	ret = rds_trans_register(&rds_iw_transport);

	if (ret)

		goto out_recv;

	rds_info_register_func(RDS_INFO_IWARP_CONNECTIONS, rds_iw_ic_info);

	goto out;

out_recv:

	rds_iw_recv_exit();

out_sysctl:

	rds_iw_sysctl_exit();

out_ibreg:

	ib_unregister_client(&rds_iw_client);

out:

	return ret;

}

MODULE_LICENSE("GPL");

#ifndef _RDS_IW_H

#define _RDS_IW_H

#include <linux/interrupt.h>

#include <rdma/ib_verbs.h>

#include <rdma/rdma_cm.h>

#include "rds.h"

#include "rdma_transport.h"

#define RDS_FASTREG_SIZE		20

#define RDS_FASTREG_POOL_SIZE		2048

#define RDS_IW_MAX_SGE			8

#define RDS_IW_RECV_SGE 		2

#define RDS_IW_DEFAULT_RECV_WR		1024

#define RDS_IW_DEFAULT_SEND_WR		256

#define RDS_IW_SUPPORTED_PROTOCOLS	0x00000003	/* minor versions supported */

extern struct list_head rds_iw_devices;

/*

 * IB posts RDS_FRAG_SIZE fragments of pages to the receive queues to

 * try and minimize the amount of memory tied up both the device and

 * socket receive queues.

 */

/* page offset of the final full frag that fits in the page */

#define RDS_PAGE_LAST_OFF (((PAGE_SIZE  / RDS_FRAG_SIZE) - 1) * RDS_FRAG_SIZE)

struct rds_page_frag {

	struct list_head	f_item;

	struct page		*f_page;

	unsigned long		f_offset;

	dma_addr_t 		f_mapped;

};

struct rds_iw_incoming {

	struct list_head	ii_frags;

	struct rds_incoming	ii_inc;

};

struct rds_iw_connect_private {

	/* Add new fields at the end, and don't permute existing fields. */

	__be32			dp_saddr;

	__be32			dp_daddr;

	u8			dp_protocol_major;

	u8			dp_protocol_minor;

	__be16			dp_protocol_minor_mask; /* bitmask */

	__be32			dp_reserved1;

	__be64			dp_ack_seq;

	__be32			dp_credit;		/* non-zero enables flow ctl */

};

struct rds_iw_scatterlist {

	struct scatterlist	*list;

	unsigned int		len;

	int			dma_len;

	unsigned int		dma_npages;

	unsigned int		bytes;

};

struct rds_iw_mapping {

	spinlock_t		m_lock;	/* protect the mapping struct */

	struct list_head	m_list;

	struct rds_iw_mr	*m_mr;

	uint32_t		m_rkey;

	struct rds_iw_scatterlist m_sg;

};

struct rds_iw_send_work {

	struct rds_message	*s_rm;

	/* We should really put these into a union: */

	struct rm_rdma_op	*s_op;

	struct rds_iw_mapping	*s_mapping;

	struct ib_mr		*s_mr;

	unsigned char		s_remap_count;

	union {

		struct ib_send_wr	s_send_wr;

		struct ib_rdma_wr	s_rdma_wr;

		struct ib_reg_wr	s_reg_wr;

	};

	struct ib_sge		s_sge[RDS_IW_MAX_SGE];

	unsigned long		s_queued;

};

struct rds_iw_recv_work {

	struct rds_iw_incoming 	*r_iwinc;

	struct rds_page_frag	*r_frag;

	struct ib_recv_wr	r_wr;

	struct ib_sge		r_sge[2];

};

struct rds_iw_work_ring {

	u32		w_nr;

	u32		w_alloc_ptr;

	u32		w_alloc_ctr;

	u32		w_free_ptr;

	atomic_t	w_free_ctr;

};

struct rds_iw_device;

struct rds_iw_connection {

	struct list_head	iw_node;

	struct rds_iw_device 	*rds_iwdev;

	struct rds_connection	*conn;

	/* alphabet soup, IBTA style */

	struct rdma_cm_id	*i_cm_id;

	struct ib_pd		*i_pd;

	struct ib_mr		*i_mr;

	struct ib_cq		*i_send_cq;

	struct ib_cq		*i_recv_cq;

	/* tx */

	struct rds_iw_work_ring	i_send_ring;

	struct rds_message	*i_rm;

	struct rds_header	*i_send_hdrs;

	u64			i_send_hdrs_dma;

	struct rds_iw_send_work *i_sends;

	/* rx */

	struct tasklet_struct	i_recv_tasklet;

	struct mutex		i_recv_mutex;

	struct rds_iw_work_ring	i_recv_ring;

	struct rds_iw_incoming	*i_iwinc;

	u32			i_recv_data_rem;

	struct rds_header	*i_recv_hdrs;

	u64			i_recv_hdrs_dma;

	struct rds_iw_recv_work *i_recvs;

	struct rds_page_frag	i_frag;

	u64			i_ack_recv;	/* last ACK received */

	/* sending acks */

	unsigned long		i_ack_flags;

#ifdef KERNEL_HAS_ATOMIC64

	atomic64_t		i_ack_next;	/* next ACK to send */

#else

	spinlock_t		i_ack_lock;	/* protect i_ack_next */

	u64			i_ack_next;	/* next ACK to send */

#endif

	struct rds_header	*i_ack;

	struct ib_send_wr	i_ack_wr;

	struct ib_sge		i_ack_sge;

	u64			i_ack_dma;

	unsigned long		i_ack_queued;

	/* Flow control related information

	 *

	 * Our algorithm uses a pair variables that we need to access

	 * atomically - one for the send credits, and one posted

	 * recv credits we need to transfer to remote.

	 * Rather than protect them using a slow spinlock, we put both into

	 * a single atomic_t and update it using cmpxchg

	 */

	atomic_t		i_credits;

	/* Protocol version specific information */

	unsigned int		i_flowctl:1;	/* enable/disable flow ctl */

	unsigned int		i_dma_local_lkey:1;

	unsigned int		i_fastreg_posted:1; /* fastreg posted on this connection */

	/* Batched completions */

	unsigned int		i_unsignaled_wrs;

	long			i_unsignaled_bytes;

};

/* This assumes that atomic_t is at least 32 bits */

#define IB_GET_SEND_CREDITS(v)	((v) & 0xffff)

#define IB_GET_POST_CREDITS(v)	((v) >> 16)

#define IB_SET_SEND_CREDITS(v)	((v) & 0xffff)

#define IB_SET_POST_CREDITS(v)	((v) << 16)

struct rds_iw_cm_id {

	struct list_head	list;

	struct rdma_cm_id	*cm_id;

};

struct rds_iw_device {

	struct list_head	list;

	struct list_head	cm_id_list;

	struct list_head	conn_list;

	struct ib_device	*dev;

	struct ib_pd		*pd;

	struct ib_mr		*mr;

	struct rds_iw_mr_pool	*mr_pool;

	int			max_sge;

	unsigned int		max_wrs;

	unsigned int		dma_local_lkey:1;

	spinlock_t		spinlock;	/* protect the above */

};

/* bits for i_ack_flags */

#define IB_ACK_IN_FLIGHT	0

#define IB_ACK_REQUESTED	1

/* Magic WR_ID for ACKs */

#define RDS_IW_ACK_WR_ID	((u64)0xffffffffffffffffULL)

#define RDS_IW_REG_WR_ID	((u64)0xefefefefefefefefULL)

#define RDS_IW_LOCAL_INV_WR_ID	((u64)0xdfdfdfdfdfdfdfdfULL)

struct rds_iw_statistics {

	uint64_t	s_iw_connect_raced;

	uint64_t	s_iw_listen_closed_stale;

	uint64_t	s_iw_tx_cq_call;

	uint64_t	s_iw_tx_cq_event;

	uint64_t	s_iw_tx_ring_full;

	uint64_t	s_iw_tx_throttle;

	uint64_t	s_iw_tx_sg_mapping_failure;

	uint64_t	s_iw_tx_stalled;

	uint64_t	s_iw_tx_credit_updates;

	uint64_t	s_iw_rx_cq_call;

	uint64_t	s_iw_rx_cq_event;

	uint64_t	s_iw_rx_ring_empty;

	uint64_t	s_iw_rx_refill_from_cq;

	uint64_t	s_iw_rx_refill_from_thread;

	uint64_t	s_iw_rx_alloc_limit;

	uint64_t	s_iw_rx_credit_updates;

	uint64_t	s_iw_ack_sent;

	uint64_t	s_iw_ack_send_failure;

	uint64_t	s_iw_ack_send_delayed;

	uint64_t	s_iw_ack_send_piggybacked;

	uint64_t	s_iw_ack_received;

	uint64_t	s_iw_rdma_mr_alloc;

	uint64_t	s_iw_rdma_mr_free;

	uint64_t	s_iw_rdma_mr_used;

	uint64_t	s_iw_rdma_mr_pool_flush;

	uint64_t	s_iw_rdma_mr_pool_wait;

	uint64_t	s_iw_rdma_mr_pool_depleted;

};

extern struct workqueue_struct *rds_iw_wq;

/*

 * Fake ib_dma_sync_sg_for_{cpu,device} as long as ib_verbs.h

 * doesn't define it.

 */

static inline void rds_iw_dma_sync_sg_for_cpu(struct ib_device *dev,

		struct scatterlist *sg, unsigned int sg_dma_len, int direction)

{

	unsigned int i;

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

		ib_dma_sync_single_for_cpu(dev,

				ib_sg_dma_address(dev, &sg[i]),

				ib_sg_dma_len(dev, &sg[i]),

				direction);

	}

}

#define ib_dma_sync_sg_for_cpu	rds_iw_dma_sync_sg_for_cpu

static inline void rds_iw_dma_sync_sg_for_device(struct ib_device *dev,

		struct scatterlist *sg, unsigned int sg_dma_len, int direction)

{

	unsigned int i;

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

		ib_dma_sync_single_for_device(dev,

				ib_sg_dma_address(dev, &sg[i]),

				ib_sg_dma_len(dev, &sg[i]),

				direction);

	}

}

#define ib_dma_sync_sg_for_device	rds_iw_dma_sync_sg_for_device

static inline u32 rds_iw_local_dma_lkey(struct rds_iw_connection *ic)

{

	return ic->i_dma_local_lkey ? ic->i_cm_id->device->local_dma_lkey : ic->i_mr->lkey;

}

/* ib.c */

extern struct rds_transport rds_iw_transport;

extern struct ib_client rds_iw_client;

extern unsigned int fastreg_pool_size;

extern unsigned int fastreg_message_size;

extern spinlock_t iw_nodev_conns_lock;

extern struct list_head iw_nodev_conns;

/* ib_cm.c */

int rds_iw_conn_alloc(struct rds_connection *conn, gfp_t gfp);

void rds_iw_conn_free(void *arg);

int rds_iw_conn_connect(struct rds_connection *conn);

void rds_iw_conn_shutdown(struct rds_connection *conn);

void rds_iw_state_change(struct sock *sk);

int rds_iw_listen_init(void);

void rds_iw_listen_stop(void);

void __rds_iw_conn_error(struct rds_connection *conn, const char *, ...);

int rds_iw_cm_handle_connect(struct rdma_cm_id *cm_id,

			     struct rdma_cm_event *event);

int rds_iw_cm_initiate_connect(struct rdma_cm_id *cm_id);

void rds_iw_cm_connect_complete(struct rds_connection *conn,

				struct rdma_cm_event *event);

#define rds_iw_conn_error(conn, fmt...) \

	__rds_iw_conn_error(conn, KERN_WARNING "RDS/IW: " fmt)

/* ib_rdma.c */

int rds_iw_update_cm_id(struct rds_iw_device *rds_iwdev, struct rdma_cm_id *cm_id);

void rds_iw_add_conn(struct rds_iw_device *rds_iwdev, struct rds_connection *conn);

void rds_iw_remove_conn(struct rds_iw_device *rds_iwdev, struct rds_connection *conn);

void __rds_iw_destroy_conns(struct list_head *list, spinlock_t *list_lock);

static inline void rds_iw_destroy_nodev_conns(void)

{

	__rds_iw_destroy_conns(&iw_nodev_conns, &iw_nodev_conns_lock);

}

static inline void rds_iw_destroy_conns(struct rds_iw_device *rds_iwdev)

{

	__rds_iw_destroy_conns(&rds_iwdev->conn_list, &rds_iwdev->spinlock);

}

struct rds_iw_mr_pool *rds_iw_create_mr_pool(struct rds_iw_device *);

void rds_iw_get_mr_info(struct rds_iw_device *rds_iwdev, struct rds_info_rdma_connection *iinfo);

void rds_iw_destroy_mr_pool(struct rds_iw_mr_pool *);

void *rds_iw_get_mr(struct scatterlist *sg, unsigned long nents,

		    struct rds_sock *rs, u32 *key_ret);

void rds_iw_sync_mr(void *trans_private, int dir);

void rds_iw_free_mr(void *trans_private, int invalidate);

void rds_iw_flush_mrs(void);

/* ib_recv.c */

int rds_iw_recv_init(void);

void rds_iw_recv_exit(void);

int rds_iw_recv(struct rds_connection *conn);

int rds_iw_recv_refill(struct rds_connection *conn, gfp_t kptr_gfp,

		       gfp_t page_gfp, int prefill);

void rds_iw_inc_free(struct rds_incoming *inc);

int rds_iw_inc_copy_to_user(struct rds_incoming *inc, struct iov_iter *to);

void rds_iw_recv_cq_comp_handler(struct ib_cq *cq, void *context);

void rds_iw_recv_tasklet_fn(unsigned long data);

void rds_iw_recv_init_ring(struct rds_iw_connection *ic);

void rds_iw_recv_clear_ring(struct rds_iw_connection *ic);

void rds_iw_recv_init_ack(struct rds_iw_connection *ic);

void rds_iw_attempt_ack(struct rds_iw_connection *ic);

void rds_iw_ack_send_complete(struct rds_iw_connection *ic);

u64 rds_iw_piggyb_ack(struct rds_iw_connection *ic);

/* ib_ring.c */

void rds_iw_ring_init(struct rds_iw_work_ring *ring, u32 nr);

void rds_iw_ring_resize(struct rds_iw_work_ring *ring, u32 nr);

u32 rds_iw_ring_alloc(struct rds_iw_work_ring *ring, u32 val, u32 *pos);

void rds_iw_ring_free(struct rds_iw_work_ring *ring, u32 val);

void rds_iw_ring_unalloc(struct rds_iw_work_ring *ring, u32 val);

int rds_iw_ring_empty(struct rds_iw_work_ring *ring);

int rds_iw_ring_low(struct rds_iw_work_ring *ring);

u32 rds_iw_ring_oldest(struct rds_iw_work_ring *ring);

u32 rds_iw_ring_completed(struct rds_iw_work_ring *ring, u32 wr_id, u32 oldest);

extern wait_queue_head_t rds_iw_ring_empty_wait;

/* ib_send.c */

void rds_iw_xmit_complete(struct rds_connection *conn);

int rds_iw_xmit(struct rds_connection *conn, struct rds_message *rm,

		unsigned int hdr_off, unsigned int sg, unsigned int off);

void rds_iw_send_cq_comp_handler(struct ib_cq *cq, void *context);

void rds_iw_send_init_ring(struct rds_iw_connection *ic);

void rds_iw_send_clear_ring(struct rds_iw_connection *ic);

int rds_iw_xmit_rdma(struct rds_connection *conn, struct rm_rdma_op *op);

void rds_iw_send_add_credits(struct rds_connection *conn, unsigned int credits);

void rds_iw_advertise_credits(struct rds_connection *conn, unsigned int posted);

int rds_iw_send_grab_credits(struct rds_iw_connection *ic, u32 wanted,

			     u32 *adv_credits, int need_posted, int max_posted);

/* ib_stats.c */

DECLARE_PER_CPU(struct rds_iw_statistics, rds_iw_stats);

#define rds_iw_stats_inc(member) rds_stats_inc_which(rds_iw_stats, member)