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

/*
 * Copyright (c) 2006, 2010, Oracle and/or its affiliates. All rights reserved.
 */

#include <sys/types.h>
#include <sys/errno.h>
#include <sys/sysmacros.h>
#include <sys/param.h>
#include <sys/machsystm.h>
#include <sys/stream.h>
#include <sys/strsubr.h>
#include <sys/kmem.h>
#include <sys/strsun.h>
#include <sys/callb.h>
#include <sys/sdt.h>
#include <sys/mach_descrip.h>
#include <sys/mdeg.h>
#include <net/if.h>
#include <sys/vsw.h>
#include <sys/vio_mailbox.h>
#include <sys/vio_common.h>
#include <sys/vnet_common.h>
#include <sys/vnet_mailbox.h>
#include <sys/vio_util.h>

/*
 * This file contains the implementation of RxDringData transfer mode of VIO
 * Protocol in vsw. The functions in this file are invoked from vsw_ldc.c
 * after RxDringData mode is negotiated with the peer during attribute phase of
 * handshake. This file contains functions that setup the transmit and receive
 * descriptor rings, and associated resources in RxDringData mode. It also
 * contains the transmit and receive data processing functions that are invoked
 * in RxDringData mode. The data processing routines in this file have the
 * suffix '_shm' to indicate the shared memory mechanism used in RxDringData
 * mode.
 */

/* Functions exported to vsw_ldc.c */
vio_dring_reg_msg_t *vsw_create_rx_dring_info(vsw_ldc_t *);
void vsw_destroy_rx_dring(vsw_ldc_t *ldcp);
dring_info_t *vsw_map_tx_dring(vsw_ldc_t *ldcp, void *pkt);
void vsw_unmap_tx_dring(vsw_ldc_t *ldcp);
int vsw_dringsend_shm(vsw_ldc_t *, mblk_t *);
void vsw_ldc_rcv_worker(void *arg);
void vsw_stop_rcv_thread(vsw_ldc_t *ldcp);
void vsw_process_dringdata_shm(void *, void *);

/* Internal functions */
static dring_info_t *vsw_create_rx_dring(vsw_ldc_t *);
static int vsw_setup_rx_dring(vsw_ldc_t *ldcp, dring_info_t *dp);
static void vsw_process_dringdata_info_shm(vsw_ldc_t *ldcp,
	vio_dring_msg_t *msg);
static void vsw_process_dringdata_ack_shm(vsw_ldc_t *ldcp,
	vio_dring_msg_t *msg);
static void vsw_ldc_rcv_shm(vsw_ldc_t *ldcp);
static int vsw_receive_packet(vsw_ldc_t *ldcp, mblk_t **bp);
static int vsw_send_msg_shm(vsw_ldc_t *ldcp, void *msgp, int size,
    boolean_t handle_reset);

/* Functions imported from vsw_ldc.c */
extern void vsw_process_pkt(void *);
extern void vsw_destroy_rxpools(void *);
extern dring_info_t *vsw_map_dring_cmn(vsw_ldc_t *ldcp,
    vio_dring_reg_msg_t *dring_pkt);
extern void vsw_process_conn_evt(vsw_ldc_t *, uint16_t);
extern mblk_t *vsw_vlan_frame_pretag(void *arg, int type, mblk_t *mp);

/* Tunables */
extern int vsw_wretries;
extern int vsw_recv_delay;
extern int vsw_recv_retries;
extern uint32_t vsw_chain_len;
extern uint32_t vsw_num_descriptors;
extern uint32_t vsw_nrbufs_factor;

#define	VSW_SWITCH_FRAMES(vswp, ldcp, bp, bpt, count, total_count)	\
{									\
	DTRACE_PROBE2(vsw_rx_pkts, vsw_ldc_t *, (ldcp), int, (count));	\
	(vswp)->vsw_switch_frame((vswp), (bp), VSW_VNETPORT,		\
	    (ldcp)->ldc_port, NULL);					\
	(bp) = (bpt) = NULL;						\
	(count) = 0;							\
}

vio_dring_reg_msg_t *
vsw_create_rx_dring_info(vsw_ldc_t *ldcp)
{
	vio_dring_reg_msg_t	*mp;
	vio_dring_reg_ext_msg_t	*emsg;
	dring_info_t		*dp;
	uint8_t			*buf;
	vsw_t			*vswp = ldcp->ldc_vswp;

	D1(vswp, "%s enter\n", __func__);

	/*
	 * If we can't create a dring, obviously no point sending
	 * a message.
	 */
	if ((dp = vsw_create_rx_dring(ldcp)) == NULL)
		return (NULL);

	mp = kmem_zalloc(VNET_DRING_REG_EXT_MSG_SIZE(dp->data_ncookies),
	    KM_SLEEP);

	mp->tag.vio_msgtype = VIO_TYPE_CTRL;
	mp->tag.vio_subtype = VIO_SUBTYPE_INFO;
	mp->tag.vio_subtype_env = VIO_DRING_REG;
	mp->tag.vio_sid = ldcp->local_session;

	/* payload */
	mp->num_descriptors = dp->num_descriptors;
	mp->descriptor_size = dp->descriptor_size;
	mp->options = dp->options;
	mp->ncookies = dp->dring_ncookies;
	bcopy(&dp->dring_cookie[0], &mp->cookie[0],
	    sizeof (ldc_mem_cookie_t));

	mp->dring_ident = 0;

	buf = (uint8_t *)mp->cookie;

	/* skip over dring cookies */
	ASSERT(mp->ncookies == 1);
	buf += (mp->ncookies * sizeof (ldc_mem_cookie_t));

	emsg = (vio_dring_reg_ext_msg_t *)buf;

	/* copy data_ncookies in the msg */
	emsg->data_ncookies = dp->data_ncookies;

	/* copy data area size in the msg */
	emsg->data_area_size = dp->data_sz;

	/* copy data area cookies in the msg */
	bcopy(dp->data_cookie, (ldc_mem_cookie_t *)emsg->data_cookie,
	    sizeof (ldc_mem_cookie_t) * dp->data_ncookies);

	D1(vswp, "%s exit\n", __func__);

	return (mp);
}

/*
 * Allocate receive resources for the channel. The resources consist of a
 * receive descriptor ring and an associated receive buffer area.
 */
static dring_info_t *
vsw_create_rx_dring(vsw_ldc_t *ldcp)
{
	vsw_t			*vswp = ldcp->ldc_vswp;
	ldc_mem_info_t		minfo;
	dring_info_t		*dp;

	dp = (dring_info_t *)kmem_zalloc(sizeof (dring_info_t), KM_SLEEP);
	mutex_init(&dp->dlock, NULL, MUTEX_DRIVER, NULL);
	ldcp->lane_out.dringp = dp;

	/* Create the receive descriptor ring */
	if ((ldc_mem_dring_create(vsw_num_descriptors,
	    sizeof (vnet_rx_dringdata_desc_t), &dp->dring_handle)) != 0) {
		DERR(vswp, "vsw_create_rx_dring(%lld): ldc dring create "
		    "failed", ldcp->ldc_id);
		goto fail;
	}

	ASSERT(dp->dring_handle != NULL);

	/* Get the addr of descriptor ring */
	if ((ldc_mem_dring_info(dp->dring_handle, &minfo)) != 0) {
		DERR(vswp, "vsw_create_rx_dring(%lld): dring info failed\n",
		    ldcp->ldc_id);
		goto fail;
	} else {
		ASSERT(minfo.vaddr != 0);
		dp->pub_addr = minfo.vaddr;
	}

	dp->num_descriptors = vsw_num_descriptors;
	dp->descriptor_size = sizeof (vnet_rx_dringdata_desc_t);
	dp->options = VIO_RX_DRING_DATA;
	dp->dring_ncookies = 1;	/* guaranteed by ldc */
	dp->num_bufs = VSW_RXDRING_NRBUFS;

	/*
	 * Allocate a table that maps descriptor to its associated buffer;
	 * used while receiving to validate that the peer has not changed the
	 * buffer offset provided in the descriptor.
	 */
	dp->rxdp_to_vmp = kmem_zalloc(dp->num_descriptors * sizeof (uintptr_t),
	    KM_SLEEP);

	/* Setup the descriptor ring */
	if (vsw_setup_rx_dring(ldcp, dp)) {
		DERR(vswp, "%s: unable to setup ring", __func__);
		goto fail;
	}

	/*
	 * The descriptors and the associated buffers are all ready;
	 * now bind descriptor ring to the channel.
	 */
	if ((ldc_mem_dring_bind(ldcp->ldc_handle, dp->dring_handle,
	    LDC_DIRECT_MAP | LDC_SHADOW_MAP, LDC_MEM_RW,
	    &dp->dring_cookie[0], &dp->dring_ncookies)) != 0) {
		DERR(vswp, "vsw_create_rx_dring: unable to bind to channel "
		    "%lld", ldcp->ldc_id);
		goto fail;
	}

	/* haven't used any descriptors yet */
	dp->end_idx = 0;
	dp->last_ack_recv = -1;
	dp->next_rxi = 0;
	return (dp);

fail:
	vsw_destroy_rx_dring(ldcp);
	return (NULL);
}

/*
 * Setup the descriptors in the rx dring.
 * Returns 0 on success, 1 on failure.
 */
static int
vsw_setup_rx_dring(vsw_ldc_t *ldcp, dring_info_t *dp)
{
	int				i, j;
	int				rv;
	size_t				data_sz;
	vio_mblk_t			*vmp;
	vio_mblk_t			**rxdp_to_vmp;
	vnet_rx_dringdata_desc_t	*rxdp;
	vnet_rx_dringdata_desc_t	*pub_addr;
	vsw_t				*vswp = ldcp->ldc_vswp;
	uint32_t			ncookies = 0;
	static char			*name = "vsw_setup_rx_dring";
	void				*data_addr = NULL;

	/*
	 * Allocate a single large buffer that serves as the rx buffer area.
	 * We allocate a ldc memory handle and export the buffer area as shared
	 * memory. We send the ldc memcookie for this buffer space to the peer,
	 * as part of dring registration phase during handshake. We manage this
	 * buffer area as individual buffers of max_frame_size and provide
	 * specific buffer offsets in each descriptor to the peer. Note that
	 * the factor used to compute the # of buffers (above) must be > 1 to
	 * ensure that there are more buffers than the # of descriptors. This
	 * is needed because, while the shared memory buffers are sent up our
	 * stack during receive, the sender needs additional buffers that can
	 * be used for further transmits. This also means there is no one to
	 * one correspondence between the descriptor index and buffer offset.
	 * The sender has to read the buffer offset in the descriptor and use
	 * the specified offset to copy the tx data into the shared buffer. We
	 * (receiver) manage the individual buffers and their state (see
	 * VIO_MBLK_STATEs in vio_util.h).
	 */
	data_sz = RXDRING_DBLK_SZ(vswp->max_frame_size);

	dp->desc_data_sz = data_sz;
	dp->data_sz = (dp->num_bufs * data_sz);
	data_addr = kmem_zalloc(dp->data_sz, KM_SLEEP);
	dp->data_addr = data_addr;

	D2(vswp, "%s: allocated %lld bytes at 0x%llx\n", name,
	    dp->data_sz, dp->data_addr);

	/* Allocate a ldc memhandle for the entire rx data area */
	rv = ldc_mem_alloc_handle(ldcp->ldc_handle, &dp->data_handle);
	if (rv != 0) {
		DERR(vswp, "%s: alloc mem handle failed", name);
		goto fail;
	}

	/* Allocate memory for the data cookies */
	dp->data_cookie = kmem_zalloc(VNET_DATA_AREA_COOKIES *
	    sizeof (ldc_mem_cookie_t), KM_SLEEP);

	/*
	 * Bind ldc memhandle to the corresponding rx data area.
	 */
	rv = ldc_mem_bind_handle(dp->data_handle, (caddr_t)data_addr,
	    dp->data_sz, LDC_DIRECT_MAP, LDC_MEM_W,
	    dp->data_cookie, &ncookies);
	if (rv != 0) {
		DERR(vswp, "%s(%lld): ldc_mem_bind_handle failed "
		    "(rv %d)", name, ldcp->ldc_id, rv);
		goto fail;
	}
	if ((ncookies == 0) || (ncookies > VNET_DATA_AREA_COOKIES)) {
		goto fail;
	}
	dp->data_ncookies = ncookies;

	for (j = 1; j < ncookies; j++) {
		rv = ldc_mem_nextcookie(dp->data_handle,
		    &(dp->data_cookie[j]));
		if (rv != 0) {
			DERR(vswp, "%s: ldc_mem_nextcookie "
			    "failed rv (%d)", name, rv);
			goto fail;
		}
	}

	/*
	 * Successful in binding the handle to rx data area. Now setup mblks
	 * around each data buffer and setup the descriptors to point to these
	 * rx data buffers. We associate each descriptor with a buffer
	 * by specifying the buffer offset in the descriptor. When the peer
	 * needs to transmit data, this offset is read by the peer to determine
	 * the buffer in the mapped buffer area where the data to be
	 * transmitted should be copied, for a specific descriptor.
	 */
	rv = vio_create_mblks(dp->num_bufs, data_sz, (uint8_t *)data_addr,
	    &dp->rx_vmp);
	if (rv != 0) {
		goto fail;
	}

	pub_addr = dp->pub_addr;
	rxdp_to_vmp = dp->rxdp_to_vmp;
	for (i = 0; i < dp->num_descriptors; i++) {
		rxdp = &pub_addr[i];
		/* allocate an mblk around this data buffer */
		vmp = vio_allocb(dp->rx_vmp);
		ASSERT(vmp != NULL);
		rxdp->data_buf_offset = VIO_MBLK_DATA_OFF(vmp) + VNET_IPALIGN;
		rxdp->dstate = VIO_DESC_FREE;
		rxdp_to_vmp[i] = vmp;
	}

	return (0);

fail:
	/* return failure; caller will cleanup */
	return (1);
}

/*
 * Free receive resources for the channel.
 */
void
vsw_destroy_rx_dring(vsw_ldc_t *ldcp)
{
	vsw_t		*vswp = ldcp->ldc_vswp;
	lane_t		*lp = &ldcp->lane_out;
	dring_info_t	*dp;

	dp = lp->dringp;
	if (dp == NULL) {
		return;
	}

	mutex_enter(&dp->dlock);

	if (dp->rx_vmp != NULL) {
		vio_clobber_pool(dp->rx_vmp);
		/*
		 * If we can't destroy the rx pool for this channel, dispatch a
		 * task to retry and clean up those rx pools. Note that we
		 * don't need to wait for the task to complete. If the vsw
		 * device itself gets detached (vsw_detach()), it will wait for
		 * the task to complete implicitly in ddi_taskq_destroy().
		 */
		if (vio_destroy_mblks(dp->rx_vmp) != 0)  {
			(void) ddi_taskq_dispatch(vswp->rxp_taskq,
			    vsw_destroy_rxpools, dp->rx_vmp, DDI_SLEEP);
		}
	}

	/* Free rx data area cookies */
	if (dp->data_cookie != NULL) {
		kmem_free(dp->data_cookie, VNET_DATA_AREA_COOKIES *
		    sizeof (ldc_mem_cookie_t));
		dp->data_cookie = NULL;
	}

	/* Unbind rx data area memhandle */
	if (dp->data_ncookies != 0) {
		(void) ldc_mem_unbind_handle(dp->data_handle);
		dp->data_ncookies = 0;
	}

	/* Free rx data area memhandle */
	if (dp->data_handle) {
		(void) ldc_mem_free_handle(dp->data_handle);
		dp->data_handle = 0;
	}

	/* Now free the rx data area itself */
	if (dp->data_addr != NULL) {
		kmem_free(dp->data_addr, dp->data_sz);
	}

	/* Finally, free the receive descriptor ring */
	if (dp->dring_handle != NULL) {
		(void) ldc_mem_dring_unbind(dp->dring_handle);
		(void) ldc_mem_dring_destroy(dp->dring_handle);
	}

	if (dp->rxdp_to_vmp != NULL) {
		kmem_free(dp->rxdp_to_vmp,
		    dp->num_descriptors * sizeof (uintptr_t));
		dp->rxdp_to_vmp = NULL;
	}

	mutex_exit(&dp->dlock);
	mutex_destroy(&dp->dlock);
	mutex_destroy(&dp->restart_lock);
	kmem_free(dp, sizeof (dring_info_t));
	lp->dringp = NULL;
}

/*
 * Map the receive descriptor ring exported by the peer, as our transmit
 * descriptor ring.
 */
dring_info_t *
vsw_map_tx_dring(vsw_ldc_t *ldcp, void *pkt)
{
	int				i;
	int				rv;
	dring_info_t			*dp;
	vnet_rx_dringdata_desc_t	*txdp;
	on_trap_data_t			otd;
	vio_dring_reg_msg_t		*dring_pkt = pkt;

	dp = vsw_map_dring_cmn(ldcp, dring_pkt);
	if (dp == NULL) {
		return (NULL);
	}

	/* RxDringData mode specific initializations */
	mutex_init(&dp->txlock, NULL, MUTEX_DRIVER, NULL);
	mutex_init(&dp->restart_lock, NULL, MUTEX_DRIVER, NULL);
	dp->next_txi = dp->restart_peer_txi = 0;
	dp->restart_reqd = B_TRUE;
	ldcp->dringdata_msgid = 0;
	ldcp->lane_in.dringp = dp;

	/*
	 * Mark the descriptor state as 'done'. This is implementation specific
	 * and not required by the protocol. In our implementation, we only
	 * need the descripor to be in 'done' state to be used by the transmit
	 * function and the peer is not aware of it. As the protocol requires
	 * that during initial registration the exporting end point mark the
	 * dstate as 'free', we change it 'done' here. After this, the dstate
	 * in our implementation will keep moving between 'ready', set by our
	 * transmit function; and and 'done', set by the peer (per protocol)
	 * after receiving data.
	 * Setup on_trap() protection before accessing dring shared memory area.
	 */
	rv = LDC_ON_TRAP(&otd);
	if (rv != 0) {
		/*
		 * Data access fault occured down the code path below while
		 * accessing the descriptors. Return failure.
		 */
		goto fail;
	}

	txdp = (vnet_rx_dringdata_desc_t *)dp->pub_addr;
	for (i = 0; i < dp->num_descriptors; i++) {
		txdp[i].dstate = VIO_DESC_DONE;
	}

	(void) LDC_NO_TRAP();

	return (dp);

fail:
	if (dp->dring_handle != NULL) {
		(void) ldc_mem_dring_unmap(dp->dring_handle);
	}
	kmem_free(dp, sizeof (*dp));
	return (NULL);
}

/*
 * Unmap the transmit descriptor ring.
 */
void
vsw_unmap_tx_dring(vsw_ldc_t *ldcp)
{
	lane_t		*lp = &ldcp->lane_in;
	dring_info_t	*dp;

	if ((dp = lp->dringp) == NULL) {
		return;
	}

	/* Unmap tx data area and free data handle */
	if (dp->data_handle != NULL) {
		(void) ldc_mem_unmap(dp->data_handle);
		(void) ldc_mem_free_handle(dp->data_handle);
		dp->data_handle = NULL;
	}

	/* Free tx data area cookies */
	if (dp->data_cookie != NULL) {
		kmem_free(dp->data_cookie, dp->data_ncookies *
		    sizeof (ldc_mem_cookie_t));
		dp->data_cookie = NULL;
		dp->data_ncookies = 0;
	}

	/* Unmap peer's dring */
	if (dp->dring_handle != NULL) {
		(void) ldc_mem_dring_unmap(dp->dring_handle);
		dp->dring_handle = NULL;
	}

	mutex_destroy(&dp->txlock);
	kmem_free(dp, sizeof (dring_info_t));
	lp->dringp = NULL;
}

/*
 * A per LDC worker thread to process the rx dring and receive packets. This
 * thread is woken up by the LDC interrupt handler when a dring data info
 * message is received.
 */
void
vsw_ldc_rcv_worker(void *arg)
{
	callb_cpr_t	cprinfo;
	vsw_ldc_t	*ldcp = (vsw_ldc_t *)arg;
	vsw_t		*vswp = ldcp->ldc_vswp;

	D1(vswp, "%s(%lld):enter\n", __func__, ldcp->ldc_id);
	CALLB_CPR_INIT(&cprinfo, &ldcp->rcv_thr_lock, callb_generic_cpr,
	    "vsw_rcv_thread");
	mutex_enter(&ldcp->rcv_thr_lock);
	while (!(ldcp->rcv_thr_flags & VSW_WTHR_STOP)) {

		CALLB_CPR_SAFE_BEGIN(&cprinfo);
		/*
		 * Wait until the data is received or a stop
		 * request is received.
		 */
		while (!(ldcp->rcv_thr_flags &
		    (VSW_WTHR_DATARCVD | VSW_WTHR_STOP))) {
			cv_wait(&ldcp->rcv_thr_cv, &ldcp->rcv_thr_lock);
		}
		CALLB_CPR_SAFE_END(&cprinfo, &ldcp->rcv_thr_lock)

		/*
		 * First process the stop request.
		 */
		if (ldcp->rcv_thr_flags & VSW_WTHR_STOP) {
			D2(vswp, "%s(%lld):Rx thread stopped\n",
			    __func__, ldcp->ldc_id);
			break;
		}
		ldcp->rcv_thr_flags &= ~VSW_WTHR_DATARCVD;
		mutex_exit(&ldcp->rcv_thr_lock);
		D1(vswp, "%s(%lld):calling vsw_process_pkt\n",
		    __func__, ldcp->ldc_id);
		vsw_ldc_rcv_shm(ldcp);
		mutex_enter(&ldcp->rcv_thr_lock);
	}

	/*
	 * Update the run status and wakeup the thread that
	 * has sent the stop request.
	 */
	ldcp->rcv_thr_flags &= ~VSW_WTHR_STOP;
	ldcp->rcv_thread = NULL;
	CALLB_CPR_EXIT(&cprinfo);
	D1(vswp, "%s(%lld):exit\n", __func__, ldcp->ldc_id);
	thread_exit();
}

/*
 * Process the rx descriptor ring in the context of receive worker
 * thread and switch the received packets to their destinations.
 */
static void
vsw_ldc_rcv_shm(vsw_ldc_t *ldcp)
{
	int		rv;
	uint32_t	end_ix;
	vio_dring_msg_t msg;
	vio_dring_msg_t	*msgp = &msg;
	int		count = 0;
	int		total_count = 0;
	uint32_t	retries = 0;
	mblk_t		*bp = NULL;
	mblk_t		*bpt = NULL;
	mblk_t		*mp = NULL;
	vsw_t		*vswp = ldcp->ldc_vswp;
	lane_t		*lp = &ldcp->lane_out;
	dring_info_t	*dp = lp->dringp;

	do {
again:
		rv = vsw_receive_packet(ldcp, &mp);
		if (rv != 0) {
			if (rv == EINVAL) {
				/* Invalid descriptor error; get next */
				continue;
			}
			if (rv != EAGAIN) {
				break;
			}

			/* Descriptor not ready for processsing */
			if (retries == vsw_recv_retries) {
				DTRACE_PROBE1(vsw_noready_rxds,
				    vsw_ldc_t *, ldcp);
				break;
			}

			/* Switch packets received so far before retrying */
			if (bp != NULL) {
				VSW_SWITCH_FRAMES(vswp, ldcp, bp, bpt, count,
				    total_count);
			}
			retries++;
			drv_usecwait(vsw_recv_delay);
			goto again;
		}
		retries = 0;

		/* Build a chain of received packets */
		if (bp == NULL) {
			/* first pkt */
			bp = mp;
			bpt = bp;
			bpt->b_next = NULL;
		} else {
			mp->b_next = NULL;
			bpt->b_next = mp;
			bpt = mp;
		}

		total_count++;
		count++;

		/*
		 * If we have gathered vsw_chain_len (tunable)
		 * # of packets in the chain, switch them.
		 */
		if (count == vsw_chain_len) {
			VSW_SWITCH_FRAMES(vswp, ldcp, bp, bpt, count,
			    total_count);
		}

		/*
		 * Stop further processing if we processed the entire dring
		 * once; otherwise continue.
		 */
	} while (total_count < dp->num_bufs);

	DTRACE_PROBE2(vsw_rx_total_count, vsw_ldc_t *, ldcp,
	    int, (total_count));
	if (bp != NULL) {
		VSW_SWITCH_FRAMES(vswp, ldcp, bp, bpt, count,
		    total_count);
	}

	/* Send stopped signal to peer (sender) */
	end_ix = lp->dringp->next_rxi;
	DECR_RXI(dp, end_ix);
	msgp->tag.vio_msgtype = VIO_TYPE_DATA;
	msgp->tag.vio_subtype = VIO_SUBTYPE_ACK;
	msgp->tag.vio_subtype_env = VIO_DRING_DATA;
	msgp->dring_ident = ldcp->lane_in.dringp->ident;
	msgp->tag.vio_sid = ldcp->local_session;
	msgp->dring_process_state = VIO_DP_STOPPED;
	msgp->start_idx = VNET_START_IDX_UNSPEC;
	msgp->end_idx = end_ix;

	(void) vsw_send_msg_shm(ldcp, (void *)msgp,
	    sizeof (vio_dring_msg_t), B_TRUE);

	ldcp->ldc_stats.dring_data_acks_sent++;
	ldcp->ldc_stats.dring_stopped_acks_sent++;
}

/*
 * Process the next index in the rx dring and receive the associated packet.
 *
 * Returns:
 *	bp:	Success: The received packet.
 *		Failure: NULL
 *      retval:
 *		Success: 0
 *		Failure: EAGAIN: Descriptor not ready
 *			 EIO:    Descriptor contents invalid.
 */
static int
vsw_receive_packet(vsw_ldc_t *ldcp, mblk_t **bp)
{
	uint32_t			rxi;
	vio_mblk_t			*vmp;
	vio_mblk_t			*new_vmp;
	struct ether_header		*ehp;
	vnet_rx_dringdata_desc_t	*rxdp;
	int				err = 0;
	uint_t				nbytes = 0;
	mblk_t				*mp = NULL;
	mblk_t				*dmp = NULL;
	vgen_stats_t			*statsp = &ldcp->ldc_stats;
	dring_info_t			*dp = ldcp->lane_out.dringp;
	vnet_rx_dringdata_desc_t	*pub_addr = dp->pub_addr;

	rxi = dp->next_rxi;
	rxdp = &(pub_addr[rxi]);
	vmp = dp->rxdp_to_vmp[rxi];

	if (rxdp->dstate != VIO_DESC_READY) {
		/*
		 * Descriptor is not ready.
		 */
		return (EAGAIN);
	}

	/*
	 * Ensure load ordering of dstate and nbytes.
	 */
	MEMBAR_CONSUMER();

	if ((rxdp->nbytes < ETHERMIN) ||
	    (rxdp->nbytes > ldcp->lane_in.mtu) ||
	    (rxdp->data_buf_offset !=
	    (VIO_MBLK_DATA_OFF(vmp) + VNET_IPALIGN))) {
		/*
		 * Descriptor contents invalid.
		 */
		statsp->ierrors++;
		rxdp->dstate = VIO_DESC_DONE;
		err = EIO;
		goto done;
	}

	/*
	 * Now allocate a new buffer for this descriptor before sending up the
	 * buffer being processed. If that fails, stop processing; as we are
	 * out of receive buffers.
	 */
	new_vmp = vio_allocb(dp->rx_vmp);

	/*
	 * Process the current buffer being received.
	 */
	nbytes = rxdp->nbytes;
	mp = vmp->mp;

	if (new_vmp == NULL) {
		/*
		 * We failed to get a new mapped buffer that is needed to
		 * refill the descriptor. In that case, leave the current
		 * buffer bound to the descriptor; allocate an mblk dynamically
		 * and copy the contents of the buffer to the mblk. Then send
		 * up this mblk. This way the sender has the same buffer as
		 * before that can be used to send new data.
		 */
		statsp->norcvbuf++;
		dmp = allocb(nbytes + VNET_IPALIGN, BPRI_MED);
		bcopy(mp->b_rptr + VNET_IPALIGN,
		    dmp->b_rptr + VNET_IPALIGN, nbytes);
		mp = dmp;
	} else {
		/* Mark the status of the current rbuf */
		vmp->state = VIO_MBLK_HAS_DATA;

		/* Set the offset of the new buffer in the descriptor */
		rxdp->data_buf_offset =
		    VIO_MBLK_DATA_OFF(new_vmp) + VNET_IPALIGN;
		dp->rxdp_to_vmp[rxi] = new_vmp;
	}
	mp->b_rptr += VNET_IPALIGN;
	mp->b_wptr = mp->b_rptr + nbytes;

	/*
	 * Ensure store ordering of data_buf_offset and dstate; so that the
	 * peer sees the right data_buf_offset after it checks that the dstate
	 * is DONE.
	 */
	MEMBAR_PRODUCER();

	/* Now mark the descriptor 'done' */
	rxdp->dstate = VIO_DESC_DONE;

	/* Update stats */
	statsp->ipackets++;
	statsp->rbytes += rxdp->nbytes;
	ehp = (struct ether_header *)mp->b_rptr;
	if (IS_BROADCAST(ehp))
		statsp->brdcstrcv++;
	else if (IS_MULTICAST(ehp))
		statsp->multircv++;
done:
	/* Update the next index to be processed */
	INCR_RXI(dp, rxi);

	/* Save the new recv index */
	dp->next_rxi = rxi;

	/* Return the packet received */
	*bp = mp;
	return (err);
}

void
vsw_stop_rcv_thread(vsw_ldc_t *ldcp)
{
	kt_did_t	tid = 0;
	vsw_t		*vswp = ldcp->ldc_vswp;

	D1(vswp, "%s(%lld):enter\n", __func__, ldcp->ldc_id);
	/*
	 * Send a stop request by setting the stop flag and
	 * wait until the rcv process thread stops.
	 */
	mutex_enter(&ldcp->rcv_thr_lock);
	if (ldcp->rcv_thread != NULL) {
		tid = ldcp->rcv_thread->t_did;
		ldcp->rcv_thr_flags |= VSW_WTHR_STOP;
		cv_signal(&ldcp->rcv_thr_cv);
	}
	mutex_exit(&ldcp->rcv_thr_lock);

	if (tid != 0) {
		thread_join(tid);
	}
	D1(vswp, "%s(%lld):exit\n", __func__, ldcp->ldc_id);
}

int
vsw_dringsend_shm(vsw_ldc_t *ldcp, mblk_t *mp)
{
	uint32_t			next_txi;
	uint32_t			txi;
	vnet_rx_dringdata_desc_t	*txdp;
	struct ether_header		*ehp;
	size_t				mblksz;
	caddr_t				dst;
	mblk_t				*bp;
	size_t				size;
	on_trap_data_t			otd;
	uint32_t			buf_offset;
	vnet_rx_dringdata_desc_t	*pub_addr;
	vio_dring_msg_t			msg;
	vio_dring_msg_t			*msgp = &msg;
	int				rv = 0;
	boolean_t			resched_peer = B_FALSE;
	boolean_t			is_bcast = B_FALSE;
	boolean_t			is_mcast = B_FALSE;
	vgen_stats_t			*statsp = &ldcp->ldc_stats;
	lane_t				*lane_in = &ldcp->lane_in;
	lane_t				*lane_out = &ldcp->lane_out;
	dring_info_t			*dp = lane_in->dringp;
	vsw_t				*vswp = ldcp->ldc_vswp;

	if ((!(lane_in->lstate & VSW_LANE_ACTIVE)) ||
	    (ldcp->ldc_status != LDC_UP) || (ldcp->ldc_handle == NULL)) {
		DWARN(vswp, "%s(%lld) status(%d) lstate(0x%llx), dropping "
		    "packet\n", __func__, ldcp->ldc_id, ldcp->ldc_status,
		    lane_in->lstate);
		statsp->oerrors++;
		return (LDC_TX_FAILURE);
	}

	if (dp == NULL) {
		DERR(vswp, "%s(%lld): no dring for outbound lane on"
		    " channel %d", __func__, ldcp->ldc_id, ldcp->ldc_id);
		statsp->oerrors++;
		return (LDC_TX_FAILURE);
	}
	pub_addr = dp->pub_addr;

	size = msgsize(mp);

	/*
	 * Note: In RxDringData mode, lane_in is associated with transmit and
	 * lane_out is associated with receive. However, we still keep the
	 * negotiated mtu in lane_out (our exported attributes).
	 */
	if (size > (size_t)lane_out->mtu) {
		DERR(vswp, "%s(%lld) invalid size (%ld)\n", __func__,
		    ldcp->ldc_id, size);
		statsp->oerrors++;
		return (LDC_TX_FAILURE);
	}

	if (size < ETHERMIN)
		size = ETHERMIN;

	ehp = (struct ether_header *)mp->b_rptr;
	is_bcast = IS_BROADCAST(ehp);
	is_mcast = IS_MULTICAST(ehp);

	/*
	 * Setup on_trap() protection before accessing shared memory areas
	 * (descriptor and data buffer). Note that we enable this protection a
	 * little early and turn it off slightly later, than keeping it enabled
	 * strictly at the points in code below where the descriptor and data
	 * buffer are accessed. This is done for performance reasons:
	 * (a) to avoid calling the trap protection code while holding mutex.
	 * (b) to avoid multiple on/off steps for descriptor and data accesses.
	 */
	rv = LDC_ON_TRAP(&otd);
	if (rv != 0) {
		/*
		 * Data access fault occured down the code path below while
		 * accessing either the descriptor or the data buffer. Release
		 * any locks that we might have acquired in the code below and
		 * return failure.
		 */
		DERR(vswp, "%s(%lld) data access fault occured\n",
		    __func__, ldcp->ldc_id);
		statsp->oerrors++;
		if (mutex_owned(&dp->txlock)) {
			mutex_exit(&dp->txlock);
		}
		if (mutex_owned(&dp->restart_lock)) {
			mutex_exit(&dp->restart_lock);
		}
		goto dringsend_shm_exit;
	}

	/*
	 * Allocate a descriptor
	 */
	mutex_enter(&dp->txlock);
	txi = next_txi = dp->next_txi;
	INCR_TXI(dp, next_txi);
	txdp = &(pub_addr[txi]);
	if (txdp->dstate != VIO_DESC_DONE) { /* out of descriptors */
		statsp->tx_no_desc++;
		mutex_exit(&dp->txlock);
		(void) LDC_NO_TRAP();
		return (LDC_TX_NORESOURCES);
	} else {
		txdp->dstate = VIO_DESC_INITIALIZING;
	}

	/* Update descriptor ring index */
	dp->next_txi = next_txi;
	mutex_exit(&dp->txlock);

	/* Ensure load ordering of dstate (above) and data_buf_offset. */
	MEMBAR_CONSUMER();

	/* Get the offset of the buffer to be used */
	buf_offset = txdp->data_buf_offset;

	/* Access the buffer using the offset */
	dst = (caddr_t)dp->data_addr + buf_offset;

	/* Copy data into mapped transmit buffer */
	for (bp = mp; bp != NULL; bp = bp->b_cont) {
		mblksz = MBLKL(bp);
		bcopy(bp->b_rptr, dst, mblksz);
		dst += mblksz;
	}

	/* Set the size of data in the descriptor */
	txdp->nbytes = size;

	/*
	 * Ensure store ordering of nbytes and dstate (below); so that the peer
	 * sees the right nbytes value after it checks that the dstate is READY.
	 */
	MEMBAR_PRODUCER();

	mutex_enter(&dp->restart_lock);

	ASSERT(txdp->dstate == VIO_DESC_INITIALIZING);

	/* Mark the descriptor ready */
	txdp->dstate = VIO_DESC_READY;

	/* Check if peer needs wake up (handled below) */
	if (dp->restart_reqd == B_TRUE && dp->restart_peer_txi == txi) {
		dp->restart_reqd = B_FALSE;
		resched_peer = B_TRUE;
	}

	/* Update tx stats */
	statsp->opackets++;
	statsp->obytes += size;
	if (is_bcast)
		statsp->brdcstxmt++;
	else if (is_mcast)
		statsp->multixmt++;

	mutex_exit(&dp->restart_lock);

	/*
	 * We are done accessing shared memory; clear trap protection.
	 */
	(void) LDC_NO_TRAP();

	/*
	 * Need to wake up the peer ?
	 */
	if (resched_peer == B_TRUE) {
		msgp->tag.vio_msgtype = VIO_TYPE_DATA;
		msgp->tag.vio_subtype = VIO_SUBTYPE_INFO;
		msgp->tag.vio_subtype_env = VIO_DRING_DATA;
		msgp->tag.vio_sid = ldcp->local_session;
		msgp->dring_ident = lane_out->dringp->ident;
		msgp->start_idx = txi;
		msgp->end_idx = -1;

		rv = vsw_send_msg_shm(ldcp, (void *)msgp, sizeof (*msgp),
		    B_FALSE);
		if (rv != 0) {
			/* error: drop the packet */
			DERR(vswp, "%s(%lld) failed sending dringdata msg\n",
			    __func__, ldcp->ldc_id);
			mutex_enter(&dp->restart_lock);
			statsp->oerrors++;
			dp->restart_reqd = B_TRUE;
			mutex_exit(&dp->restart_lock);
		}
		statsp->dring_data_msgs_sent++;
	}

dringsend_shm_exit:
	if (rv == ECONNRESET || rv == EACCES) {
		vsw_process_conn_evt(ldcp, VSW_CONN_RESET);
	}
	return (LDC_TX_SUCCESS);
}

void
vsw_process_dringdata_shm(void *arg, void *dpkt)
{
	vsw_ldc_t		*ldcp = arg;
	vsw_t			*vswp = ldcp->ldc_vswp;
	vio_dring_msg_t		*dring_pkt = dpkt;

	switch (dring_pkt->tag.vio_subtype) {
	case VIO_SUBTYPE_INFO:
		D2(vswp, "%s(%lld): VIO_SUBTYPE_INFO", __func__, ldcp->ldc_id);
		vsw_process_dringdata_info_shm(ldcp, dring_pkt);
		break;

	case VIO_SUBTYPE_ACK:
		D2(vswp, "%s(%lld): VIO_SUBTYPE_ACK", __func__, ldcp->ldc_id);
		vsw_process_dringdata_ack_shm(ldcp, dring_pkt);
		break;

	case VIO_SUBTYPE_NACK:
		DWARN(vswp, "%s(%lld): VIO_SUBTYPE_NACK",
		    __func__, ldcp->ldc_id);
		/*
		 * Something is badly wrong if we are getting NACK's
		 * for our data pkts. So reset the channel.
		 */
		vsw_process_conn_evt(ldcp, VSW_CONN_RESTART);
		break;

	default:
		DERR(vswp, "%s(%lld): Unknown vio_subtype %x\n", __func__,
		    ldcp->ldc_id, dring_pkt->tag.vio_subtype);
	}
}

static void
vsw_process_dringdata_info_shm(vsw_ldc_t *ldcp, vio_dring_msg_t *msg)
{
	dring_info_t	*dp = ldcp->lane_in.dringp;
	vsw_t		*vswp = ldcp->ldc_vswp;
	vgen_stats_t	*statsp = &ldcp->ldc_stats;

	if (dp->ident != msg->dring_ident) {
		/* drop the message */
		DERR(vswp, "%s(%lld): Invalid dring ident 0x%llx",
		    __func__, ldcp->ldc_id, msg->dring_ident);
		return;
	}

	statsp->dring_data_msgs_rcvd++;

	/*
	 * Wake up the rcv worker thread to process the rx dring.
	 */
	ASSERT(MUTEX_HELD(&ldcp->ldc_cblock));
	mutex_exit(&ldcp->ldc_cblock);
	mutex_enter(&ldcp->rcv_thr_lock);
	if (!(ldcp->rcv_thr_flags & VSW_WTHR_DATARCVD)) {
		ldcp->rcv_thr_flags |= VSW_WTHR_DATARCVD;
		cv_signal(&ldcp->rcv_thr_cv);
	}
	mutex_exit(&ldcp->rcv_thr_lock);
	mutex_enter(&ldcp->ldc_cblock);
}

static void
vsw_process_dringdata_ack_shm(vsw_ldc_t *ldcp, vio_dring_msg_t *msg)
{
	dring_info_t			*dp;
	uint32_t			start;
	int32_t				end;
	int				rv;
	on_trap_data_t			otd;
	uint32_t			txi;
	vnet_rx_dringdata_desc_t	*txdp;
	vnet_rx_dringdata_desc_t	*pub_addr;
	boolean_t			ready_txd = B_FALSE;
	vsw_t				*vswp = ldcp->ldc_vswp;
	vgen_stats_t			*statsp = &ldcp->ldc_stats;

	dp = ldcp->lane_in.dringp;
	start = msg->start_idx;
	end = msg->end_idx;
	pub_addr = dp->pub_addr;

	/*
	 * In RxDringData mode (v1.6), start index of -1 can be used by the
	 * peer to indicate that it is unspecified. However, the end index
	 * must be set correctly indicating the last descriptor index processed.
	 */
	if (((start != VNET_START_IDX_UNSPEC) && !(CHECK_TXI(dp, start))) ||
	    !(CHECK_TXI(dp, end))) {
		/* drop the message if invalid index */
		DWARN(vswp, "%s(%lld): Invalid Tx ack start(%d) or end(%d)\n",
		    __func__, ldcp->ldc_id, start, end);
		return;
	}

	/* Validate dring_ident */
	if (msg->dring_ident != ldcp->lane_out.dringp->ident) {
		/* invalid dring_ident, drop the msg */
		DWARN(vswp, "%s(%lld): Invalid dring ident 0x%x\n",
		    __func__, ldcp->ldc_id, msg->dring_ident);
		return;
	}
	statsp->dring_data_acks_rcvd++;

	if (msg->dring_process_state != VIO_DP_STOPPED) {
		/*
		 * Receiver continued processing
		 * dring after sending us the ack.
		 */
		return;
	}

	statsp->dring_stopped_acks_rcvd++;

	/*
	 * Setup on_trap() protection before accessing dring shared memory area.
	 */
	rv = LDC_ON_TRAP(&otd);
	if (rv != 0) {
		/*
		 * Data access fault occured down the code path below while
		 * accessing the descriptors. Release any locks that we might
		 * have acquired in the code below and return failure.
		 */
		if (mutex_owned(&dp->restart_lock)) {
			mutex_exit(&dp->restart_lock);
		}
		return;
	}

	/*
	 * Determine if there are any pending tx descriptors ready to be
	 * processed by the receiver(peer) and if so, send a message to the
	 * peer to restart receiving.
	 */
	mutex_enter(&dp->restart_lock);

	ready_txd = B_FALSE;
	txi = end;
	INCR_TXI(dp, txi);
	txdp = &pub_addr[txi];
	if (txdp->dstate == VIO_DESC_READY) {
		ready_txd = B_TRUE;
	}

	/*
	 * We are done accessing shared memory; clear trap protection.
	 */
	(void) LDC_NO_TRAP();

	if (ready_txd == B_FALSE) {
		/*
		 * No ready tx descriptors. Set the flag to send a message to
		 * the peer when tx descriptors are ready in transmit routine.
		 */
		dp->restart_reqd = B_TRUE;
		dp->restart_peer_txi = txi;
		mutex_exit(&dp->restart_lock);
		return;
	}

	/*
	 * We have some tx descriptors ready to be processed by the receiver.
	 * Send a dring data message to the peer to restart processing.
	 */
	dp->restart_reqd = B_FALSE;
	mutex_exit(&dp->restart_lock);

	msg->tag.vio_msgtype = VIO_TYPE_DATA;
	msg->tag.vio_subtype = VIO_SUBTYPE_INFO;
	msg->tag.vio_subtype_env = VIO_DRING_DATA;
	msg->tag.vio_sid = ldcp->local_session;
	msg->dring_ident = ldcp->lane_out.dringp->ident;
	msg->start_idx = txi;
	msg->end_idx = -1;
	rv = vsw_send_msg_shm(ldcp, (void *)msg,
	    sizeof (vio_dring_msg_t), B_FALSE);
	statsp->dring_data_msgs_sent++;
	if (rv != 0) {
		mutex_enter(&dp->restart_lock);
		dp->restart_reqd = B_TRUE;
		mutex_exit(&dp->restart_lock);
	}

	if (rv == ECONNRESET) {
		vsw_process_conn_evt(ldcp, VSW_CONN_RESET);
	}
}

/*
 * Send dring data msgs (info/ack/nack) over LDC.
 */
int
vsw_send_msg_shm(vsw_ldc_t *ldcp, void *msgp, int size, boolean_t handle_reset)
{
	int			rv;
	int			retries = vsw_wretries;
	size_t			msglen = size;
	vsw_t			*vswp = ldcp->ldc_vswp;
	vio_dring_msg_t		*dmsg = (vio_dring_msg_t *)msgp;

	D1(vswp, "vsw_send_msg (%lld) enter : sending %d bytes",
	    ldcp->ldc_id, size);

	dmsg->seq_num = atomic_inc_32_nv(&ldcp->dringdata_msgid);

	do {
		msglen = size;
		rv = ldc_write(ldcp->ldc_handle, (caddr_t)msgp, &msglen);
	} while (rv == EWOULDBLOCK && --retries > 0);

	if ((rv != 0) || (msglen != size)) {
		DERR(vswp, "vsw_send_msg_shm:ldc_write failed: "
		    "chan(%lld) rv(%d) size (%d) msglen(%d)\n",
		    ldcp->ldc_id, rv, size, msglen);
		ldcp->ldc_stats.oerrors++;
	}

	/*
	 * If channel has been reset we either handle it here or
	 * simply report back that it has been reset and let caller
	 * decide what to do.
	 */
	if (rv == ECONNRESET) {
		DWARN(vswp, "%s (%lld) channel reset", __func__, ldcp->ldc_id);

		if (handle_reset) {
			vsw_process_conn_evt(ldcp, VSW_CONN_RESET);
		}
	}

	return (rv);
}