/*

 * CDDL HEADER START

 *

 * The contents of this file are subject to the terms of the

 * Common Development and Distribution License (the "License").

 * You may not use this file except in compliance with the License.

 *

 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

 * or http://www.opensolaris.org/os/licensing.

 * See the License for the specific language governing permissions

 * and limitations under the License.

 *

 * When distributing Covered Code, include this CDDL HEADER in each

 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.

 * If applicable, add the following below this CDDL HEADER, with the

 * fields enclosed by brackets "[]" replaced with your own identifying

 * information: Portions Copyright [yyyy] [name of copyright owner]

 *

 * CDDL HEADER END

 */

/*

 * Use is subject to license terms.

 */

/*

 * Copyright (c) 2007, The Ohio State University. All rights reserved.

 *

 * Portions of this source code is developed by the team members of

 * The Ohio State University's Network-Based Computing Laboratory (NBCL),

 * headed by Professor Dhabaleswar K. (DK) Panda.

 *

 * Acknowledgements to contributions from developors:

 *   Ranjit Noronha: [email protected]

 *   Lei Chai      : [email protected]

 *   Weikuan Yu    : [email protected]

 *

 */

/*

 * xdr_rdma.c, XDR implementation using RDMA to move large chunks

 */

#include <sys/param.h>

#include <sys/types.h>

#include <sys/systm.h>

#include <sys/kmem.h>

#include <sys/sdt.h>

#include <sys/debug.h>

#include <rpc/types.h>

#include <rpc/xdr.h>

#include <sys/cmn_err.h>

#include <rpc/rpc_sztypes.h>

#include <rpc/rpc_rdma.h>

#include <sys/sysmacros.h>

static bool_t   xdrrdma_getint32(XDR *, int32_t *);

static bool_t   xdrrdma_putint32(XDR *, int32_t *);

static bool_t   xdrrdma_getbytes(XDR *, caddr_t, int);

static bool_t   xdrrdma_putbytes(XDR *, caddr_t, int);

uint_t		xdrrdma_getpos(XDR *);

bool_t		xdrrdma_setpos(XDR *, uint_t);

static rpc_inline_t *xdrrdma_inline(XDR *, int);

void		xdrrdma_destroy(XDR *);

static bool_t   xdrrdma_control(XDR *, int, void *);

struct xdr_ops  xdrrdmablk_ops = {

	xdrrdma_getbytes,

	xdrrdma_putbytes,

	xdrrdma_getpos,

	xdrrdma_setpos,

	xdrrdma_inline,

	xdrrdma_destroy,

	xdrrdma_control,

	xdrrdma_getint32,

	xdrrdma_putint32

};

struct xdr_ops  xdrrdma_ops = {

	xdrrdma_getbytes,

	xdrrdma_putbytes,

	xdrrdma_getpos,

	xdrrdma_setpos,

	xdrrdma_inline,

	xdrrdma_destroy,

	xdrrdma_control,

	xdrrdma_getint32,

	xdrrdma_putint32

};

/*

 * A chunk list entry identifies a chunk of opaque data to be moved

 * separately from the rest of the RPC message. xp_min_chunk = 0, is a

 * special case for ENCODING, which means do not chunk the incoming stream of

 * data.

 */

typedef struct {

	caddr_t		xp_offp;

	int		xp_min_chunk;

	uint_t		xp_flags;	/* Controls setting for rdma xdr */

	int		xp_buf_size;	/* size of xdr buffer */

	struct clist	**xp_rcl_next;	/* location to place/find next chunk */

	struct clist	*xp_wcl;	/* head of write chunk list */

	CONN		*xp_conn;	/* connection for chunk data xfer */

	uint_t		xp_reply_chunk_len;

	/* used to track length for security modes: integrity/privacy */

	uint_t		xp_reply_chunk_len_alt;

} xrdma_private_t;

extern kmem_cache_t *clist_cache;

bool_t

xdrrdma_getrdmablk(XDR *xdrs, struct clist **rlist, uint_t *sizep,

    CONN **conn, const uint_t maxsize)

{

	xrdma_private_t	*xdrp = (xrdma_private_t *)(xdrs->x_private);

	struct clist	*cle = *(xdrp->xp_rcl_next);

	struct clist	*rdclist = NULL, *prev = NULL;

	bool_t		retval = TRUE;

	uint32_t	cur_offset = 0;

	uint32_t	total_segments = 0;

	uint32_t	actual_segments = 0;

	uint32_t	alen;

	uint_t		total_len;

	ASSERT(xdrs->x_op != XDR_FREE);

	/*

	 * first deal with the length since xdr bytes are counted

	 */

	if (!xdr_u_int(xdrs, sizep)) {

		DTRACE_PROBE(xdr__e__getrdmablk_sizep_fail);

		return (FALSE);

	}

	total_len = *sizep;

	if (total_len > maxsize) {

		DTRACE_PROBE2(xdr__e__getrdmablk_bad_size,

		    int, total_len, int, maxsize);

		return (FALSE);

	}

	(*conn) = xdrp->xp_conn;

	/*

	 * if no data we are done

	 */

	if (total_len == 0)

		return (TRUE);

	while (cle) {

		total_segments++;

		cle = cle->c_next;

	}

	cle = *(xdrp->xp_rcl_next);

	/*

	 * If there was a chunk at the current offset, then setup a read

	 * chunk list which records the destination address and length

	 * and will RDMA READ the data in later.

	 */

	if (cle == NULL)

		return (FALSE);

	if (cle->c_xdroff != (xdrp->xp_offp - xdrs->x_base))

		return (FALSE);

	/*

	 * Setup the chunk list with appropriate

	 * address (offset) and length

	 */

	for (actual_segments = 0;

	    actual_segments < total_segments; actual_segments++) {

		if (total_len <= 0)

			break;

		cle->u.c_daddr = (uint64) cur_offset;

		alen = 0;

		if (cle->c_len > total_len) {

			alen = cle->c_len;

			cle->c_len = total_len;

		}

		if (!alen)

			xdrp->xp_rcl_next = &cle->c_next;

		cur_offset += cle->c_len;

		total_len -= cle->c_len;

		if ((total_segments - actual_segments - 1) == 0 &&

		    total_len > 0) {

			DTRACE_PROBE(krpc__e__xdrrdma_getblk_chunktooshort);

			retval = FALSE;

		}

		if ((total_segments - actual_segments - 1) > 0 &&

		    total_len == 0) {

			DTRACE_PROBE2(krpc__e__xdrrdma_getblk_toobig,

			    int, total_segments, int, actual_segments);

		}

		rdclist = clist_alloc();

		(*rdclist) = (*cle);

		if ((*rlist) == NULL)

			(*rlist) = rdclist;

		if (prev == NULL)

			prev = rdclist;

		else {

			prev->c_next = rdclist;

			prev = rdclist;

		}

	}

out:

	if (prev != NULL)

		prev->c_next = NULL;

	if (alen) {

		cle->w.c_saddr =

		    (uint64)(uintptr_t)cle->w.c_saddr + cle->c_len;

		cle->c_len = alen - cle->c_len;

	}

	return (retval);

}

/*

 * The procedure xdrrdma_create initializes a stream descriptor for a memory

 * buffer.

 */

void

xdrrdma_create(XDR *xdrs, caddr_t addr, uint_t size,

    int min_chunk, struct clist *cl, enum xdr_op op, CONN *conn)

{

	xrdma_private_t *xdrp;

	struct clist   *cle;

	xdrs->x_op = op;

	xdrs->x_ops = &xdrrdma_ops;

	xdrs->x_base = addr;

	xdrs->x_handy = size;

	xdrs->x_public = NULL;

	xdrp = (xrdma_private_t *)kmem_zalloc(sizeof (xrdma_private_t),

	    KM_SLEEP);

	xdrs->x_private = (caddr_t)xdrp;

	xdrp->xp_offp = addr;

	xdrp->xp_min_chunk = min_chunk;

	xdrp->xp_flags = 0;

	xdrp->xp_buf_size = size;

	xdrp->xp_rcl = cl;

	xdrp->xp_reply_chunk_len = 0;

	xdrp->xp_reply_chunk_len_alt = 0;

	if (op == XDR_ENCODE && cl != NULL) {

		/* Find last element in chunk list and set xp_rcl_next */

		for (cle = cl; cle->c_next != NULL; cle = cle->c_next)

			continue;

		xdrp->xp_rcl_next = &(cle->c_next);

	} else {

		xdrp->xp_rcl_next = &(xdrp->xp_rcl);

	}

	xdrp->xp_wcl = NULL;

	xdrp->xp_conn = conn;

	if (xdrp->xp_min_chunk != 0)

		xdrp->xp_flags |= XDR_RDMA_CHUNK;

}

/* ARGSUSED */

void

xdrrdma_destroy(XDR * xdrs)

{

	xrdma_private_t	*xdrp = (xrdma_private_t *)(xdrs->x_private);

	if (xdrp == NULL)

		return;

	if (xdrp->xp_wcl) {

		if (xdrp->xp_flags & XDR_RDMA_WLIST_REG) {

			rdma_buf_free(xdrp->xp_conn,

			    &xdrp->xp_wcl->rb_longbuf);

		}

		clist_free(xdrp->xp_wcl);

	}

	if (xdrp->xp_rcl) {

		if (xdrp->xp_flags & XDR_RDMA_RLIST_REG) {

			rdma_buf_free(xdrp->xp_conn,

			    &xdrp->xp_rcl->rb_longbuf);

		}

		clist_free(xdrp->xp_rcl);

	}

	(void) kmem_free(xdrs->x_private, sizeof (xrdma_private_t));

	xdrs->x_private = NULL;

}

static	bool_t

xdrrdma_getint32(XDR *xdrs, int32_t *int32p)

{

	xrdma_private_t	*xdrp = (xrdma_private_t *)(xdrs->x_private);

	/* LINTED pointer alignment */

	*int32p = (int32_t)ntohl((uint32_t)(*((int32_t *)(xdrp->xp_offp))));

	xdrp->xp_offp += sizeof (int32_t);

	return (TRUE);

}

static	bool_t

xdrrdma_putint32(XDR *xdrs, int32_t *int32p)

{

	xrdma_private_t	*xdrp = (xrdma_private_t *)(xdrs->x_private);

	if ((xdrs->x_handy -= (int)sizeof (int32_t)) < 0)

		return (FALSE);

	/* LINTED pointer alignment */

	*(int32_t *)xdrp->xp_offp = (int32_t)htonl((uint32_t)(*int32p));

	xdrp->xp_offp += sizeof (int32_t);

	return (TRUE);

}

/*

 * DECODE bytes from XDR stream for rdma.

 * If the XDR stream contains a read chunk list,

 * it will go through xdrrdma_getrdmablk instead.

 */

static	bool_t

xdrrdma_getbytes(XDR *xdrs, caddr_t addr, int len)

{

	xrdma_private_t	*xdrp = (xrdma_private_t *)(xdrs->x_private);

	struct clist	*cle = *(xdrp->xp_rcl_next);

	struct clist	*cls = *(xdrp->xp_rcl_next);

	struct clist	cl;

	bool_t		retval = TRUE;

	uint32_t	total_len = len;

	uint32_t	cur_offset = 0;

	uint32_t	total_segments = 0;

	uint32_t	actual_segments = 0;

	uint32_t	status;

	uint32_t	alen;

	while (cle) {

		total_segments++;

		cle = cle->c_next;

	}

	cle = *(xdrp->xp_rcl_next);

	/*

	 * If there was a chunk at the current offset, then setup a read

	 * chunk list which records the destination address and length

	 * and will RDMA READ the data in later.

	 */

		for (actual_segments = 0;

		    actual_segments < total_segments; actual_segments++) {

			if (total_len <= 0)

				break;

			cle->u.c_daddr = (uint64)(uintptr_t)addr + cur_offset;

			alen = 0;

			if (cle->c_len > total_len) {

				alen = cle->c_len;

				cle->c_len = total_len;

			}

			if (!alen)

				xdrp->xp_rcl_next = &cle->c_next;

			cur_offset += cle->c_len;

			total_len -= cle->c_len;

			if ((total_segments - actual_segments - 1) == 0 &&

			    total_len > 0) {

				DTRACE_PROBE(

				    krpc__e__xdrrdma_getbytes_chunktooshort);

				retval = FALSE;

			}

			if ((total_segments - actual_segments - 1) > 0 &&

			    total_len == 0) {

				DTRACE_PROBE2(krpc__e__xdrrdma_getbytes_toobig,

				    int, total_segments, int, actual_segments);

			}

			/*

			 * RDMA READ the chunk data from the remote end.

			 * First prep the destination buffer by registering

			 * it, then RDMA READ the chunk data. Since we are

			 * doing streaming memory, sync the destination

			 * buffer to CPU and deregister the buffer.

			 */

			if (xdrp->xp_conn == NULL) {

				return (FALSE);

			}

			cl = *cle;