RDS: Add iWARP support

/*

 * 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 "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;

DEFINE_SPINLOCK(iw_nodev_conns_lock);

LIST_HEAD(iw_nodev_conns);

void rds_iw_add_one(struct ib_device *device)

{

	struct rds_iw_device *rds_iwdev;

	struct ib_device_attr *dev_attr;

	/* Only handle iwarp devices */

	if (device->node_type != RDMA_NODE_RNIC)

		return;

	dev_attr = kmalloc(sizeof *dev_attr, GFP_KERNEL);

	if (!dev_attr)

		return;

	if (ib_query_device(device, dev_attr)) {

		rdsdebug("Query device failed for %s\n", device->name);

		goto free_attr;

	}

	rds_iwdev = kmalloc(sizeof *rds_iwdev, GFP_KERNEL);

	if (!rds_iwdev)

		goto free_attr;

	spin_lock_init(&rds_iwdev->spinlock);

	rds_iwdev->dma_local_lkey = !!(dev_attr->device_cap_flags & IB_DEVICE_LOCAL_DMA_LKEY);

	rds_iwdev->max_wrs = dev_attr->max_qp_wr;

	rds_iwdev->max_sge = min(dev_attr->max_sge, RDS_IW_MAX_SGE);

	rds_iwdev->page_shift = max(PAGE_SHIFT, ffs(dev_attr->page_size_cap) - 1);

	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) {

		if (device->node_type != RDMA_NODE_RNIC) {

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

						IB_ACCESS_LOCAL_WRITE);

		} else {

			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);

	goto free_attr;

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);

free_attr:

	kfree(dev_attr);

}

void rds_iw_remove_one(struct ib_device *device)

{

	struct rds_iw_device *rds_iwdev;

	struct rds_iw_cm_id *i_cm_id, *next;

	rds_iwdev = ib_get_client_data(device, &rds_iw_client);

	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_remove_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);

	while (ib_dealloc_pd(rds_iwdev->pd)) {

		rdsdebug("Failed to dealloc pd %p\n", rds_iwdev->pd);

		msleep(1);

	}

	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;

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

		ib_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(__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(NULL, NULL, RDMA_PS_TCP);

	if (!cm_id)

		return -EADDRNOTAVAIL;

	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->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_remove_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_cong_map		= NULL,

	.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_purge		= rds_iw_inc_purge,

	.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_prefer_loopback	= 1,

};

int __init 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 <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 rds_rdma_op	*s_op;

	struct rds_iw_mapping	*s_mapping;

	struct ib_mr		*s_mr;

	struct ib_fast_reg_page_list *s_page_list;

	unsigned char		s_remap_count;

	struct ib_send_wr	s_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 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;

	u64			i_ack_next;	/* next ACK to send */

	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			page_shift;

	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_FAST_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 void rds_iw_add_one(struct ib_device *device);

extern void rds_iw_remove_one(struct ib_device *device);

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 __init 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);

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

void rds_iw_remove_nodev_conns(void);

void rds_iw_remove_conns(struct rds_iw_device *rds_iwdev);

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);

void rds_iw_remove_cm_id(struct rds_iw_device *rds_iwdev, struct rdma_cm_id *cm_id);

/* ib_recv.c */

int __init 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_purge(struct rds_incoming *inc);

void rds_iw_inc_free(struct rds_incoming *inc);

int rds_iw_inc_copy_to_user(struct rds_incoming *inc, struct iovec *iov,

			     size_t size);

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

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 rds_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);

/* 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)

unsigned int rds_iw_stats_info_copy(struct rds_info_iterator *iter,

				    unsigned int avail);

/* ib_sysctl.c */

int __init rds_iw_sysctl_init(void);

void rds_iw_sysctl_exit(void);

extern unsigned long rds_iw_sysctl_max_send_wr;

extern unsigned long rds_iw_sysctl_max_recv_wr;

extern unsigned long rds_iw_sysctl_max_unsig_wrs;

extern unsigned long rds_iw_sysctl_max_unsig_bytes;

extern unsigned long rds_iw_sysctl_max_recv_allocation;

extern unsigned int rds_iw_sysctl_flow_control;

extern ctl_table rds_iw_sysctl_table[];

/*

 * Helper functions for getting/setting the header and data SGEs in

 * RDS packets (not RDMA)

 */

static inline struct ib_sge *

rds_iw_header_sge(struct rds_iw_connection *ic, struct ib_sge *sge)

{

	return &sge[0];

}

static inline struct ib_sge *

rds_iw_data_sge(struct rds_iw_connection *ic, struct ib_sge *sge)

{

	return &sge[1];

}

static inline void rds_iw_set_64bit(u64 *ptr, u64 val)

{

#if BITS_PER_LONG == 64

	*ptr = val;

#else