/*
 * 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 2010 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 *
 *  	Copyright (c) 1983,1984,1985,1986,1987,1988,1989  AT&T.
 *	All rights reserved.
 */

#include <sys/param.h>
#include <sys/types.h>
#include <sys/systm.h>
#include <sys/thread.h>
#include <sys/t_lock.h>
#include <sys/time.h>
#include <sys/vnode.h>
#include <sys/vfs.h>
#include <sys/errno.h>
#include <sys/buf.h>
#include <sys/stat.h>
#include <sys/cred.h>
#include <sys/kmem.h>
#include <sys/debug.h>
#include <sys/dnlc.h>
#include <sys/vmsystm.h>
#include <sys/flock.h>
#include <sys/share.h>
#include <sys/cmn_err.h>
#include <sys/tiuser.h>
#include <sys/sysmacros.h>
#include <sys/callb.h>
#include <sys/acl.h>
#include <sys/kstat.h>
#include <sys/signal.h>
#include <sys/list.h>
#include <sys/zone.h>

#include <rpc/types.h>
#include <rpc/xdr.h>
#include <rpc/auth.h>
#include <rpc/clnt.h>

#include <nfs/nfs.h>
#include <nfs/nfs_clnt.h>

#include <nfs/rnode.h>
#include <nfs/nfs_acl.h>
#include <nfs/lm.h>

#include <vm/hat.h>
#include <vm/as.h>
#include <vm/page.h>
#include <vm/pvn.h>
#include <vm/seg.h>
#include <vm/seg_map.h>
#include <vm/seg_vn.h>

static void	nfs3_attr_cache(vnode_t *, vattr_t *, vattr_t *, hrtime_t,
			cred_t *);
static int	nfs_getattr_cache(vnode_t *, struct vattr *);
static int	nfs_remove_locking_id(vnode_t *, int, char *, char *, int *);

struct mi_globals {
	kmutex_t	mig_lock;  /* lock protecting mig_list */
	list_t		mig_list;  /* list of NFS v2 or v3 mounts in zone */
	boolean_t	mig_destructor_called;
};

static zone_key_t mi_list_key;

/* Debugging flag for PC file shares. */
extern int	share_debug;

/*
 * Attributes caching:
 *
 * Attributes are cached in the rnode in struct vattr form.
 * There is a time associated with the cached attributes (r_attrtime)
 * which tells whether the attributes are valid. The time is initialized
 * to the difference between current time and the modify time of the vnode
 * when new attributes are cached. This allows the attributes for
 * files that have changed recently to be timed out sooner than for files
 * that have not changed for a long time. There are minimum and maximum
 * timeout values that can be set per mount point.
 */

int
nfs_waitfor_purge_complete(vnode_t *vp)
{
	rnode_t *rp;
	k_sigset_t smask;

	rp = VTOR(vp);
	if (rp->r_serial != NULL && rp->r_serial != curthread) {
		mutex_enter(&rp->r_statelock);
		sigintr(&smask, VTOMI(vp)->mi_flags & MI_INT);
		while (rp->r_serial != NULL) {
			if (!cv_wait_sig(&rp->r_cv, &rp->r_statelock)) {
				sigunintr(&smask);
				mutex_exit(&rp->r_statelock);
				return (EINTR);
			}
		}
		sigunintr(&smask);
		mutex_exit(&rp->r_statelock);
	}
	return (0);
}

/*
 * Validate caches by checking cached attributes. If the cached
 * attributes have timed out, then get new attributes from the server.
 * As a side affect, this will do cache invalidation if the attributes
 * have changed.
 *
 * If the attributes have not timed out and if there is a cache
 * invalidation being done by some other thread, then wait until that
 * thread has completed the cache invalidation.
 */
int
nfs_validate_caches(vnode_t *vp, cred_t *cr)
{
	int error;
	struct vattr va;

	if (ATTRCACHE_VALID(vp)) {
		error = nfs_waitfor_purge_complete(vp);
		if (error)
			return (error);
		return (0);
	}

	va.va_mask = AT_ALL;
	return (nfs_getattr_otw(vp, &va, cr));
}

/*
 * Validate caches by checking cached attributes. If the cached
 * attributes have timed out, then get new attributes from the server.
 * As a side affect, this will do cache invalidation if the attributes
 * have changed.
 *
 * If the attributes have not timed out and if there is a cache
 * invalidation being done by some other thread, then wait until that
 * thread has completed the cache invalidation.
 */
int
nfs3_validate_caches(vnode_t *vp, cred_t *cr)
{
	int error;
	struct vattr va;

	if (ATTRCACHE_VALID(vp)) {
		error = nfs_waitfor_purge_complete(vp);
		if (error)
			return (error);
		return (0);
	}

	va.va_mask = AT_ALL;
	return (nfs3_getattr_otw(vp, &va, cr));
}

/*
 * Purge all of the various NFS `data' caches.
 */
void
nfs_purge_caches(vnode_t *vp, int purge_dnlc, cred_t *cr)
{
	rnode_t *rp;
	char *contents;
	int size;
	int error;

	/*
	 * Purge the DNLC for any entries which refer to this file.
	 * Avoid recursive entry into dnlc_purge_vp() in case of a directory.
	 */
	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);
	if (vp->v_count > 1 &&
	    (vp->v_type == VDIR || purge_dnlc == NFS_PURGE_DNLC) &&
	    !(rp->r_flags & RINDNLCPURGE)) {
		/*
		 * Set the RINDNLCPURGE flag to prevent recursive entry
		 * into dnlc_purge_vp()
		 */
		if (vp->v_type == VDIR)
			rp->r_flags |= RINDNLCPURGE;
		mutex_exit(&rp->r_statelock);
		dnlc_purge_vp(vp);
		mutex_enter(&rp->r_statelock);
		if (rp->r_flags & RINDNLCPURGE)
			rp->r_flags &= ~RINDNLCPURGE;
	}

	/*
	 * Clear any readdir state bits and purge the readlink response cache.
	 */
	contents = rp->r_symlink.contents;
	size = rp->r_symlink.size;
	rp->r_symlink.contents = NULL;
	mutex_exit(&rp->r_statelock);

	if (contents != NULL) {

		kmem_free((void *)contents, size);
	}

	/*
	 * Flush the page cache.
	 */
	if (vn_has_cached_data(vp)) {
		error = VOP_PUTPAGE(vp, (u_offset_t)0, 0, B_INVAL, cr, NULL);
		if (error && (error == ENOSPC || error == EDQUOT)) {
			mutex_enter(&rp->r_statelock);
			if (!rp->r_error)
				rp->r_error = error;
			mutex_exit(&rp->r_statelock);
		}
	}

	/*
	 * Flush the readdir response cache.
	 */
	if (HAVE_RDDIR_CACHE(rp))
		nfs_purge_rddir_cache(vp);
}

/*
 * Purge the readdir cache of all entries
 */
void
nfs_purge_rddir_cache(vnode_t *vp)
{
	rnode_t *rp;
	rddir_cache *rdc;
	rddir_cache *nrdc;

	rp = VTOR(vp);
top:
	mutex_enter(&rp->r_statelock);
	rp->r_direof = NULL;
	rp->r_flags &= ~RLOOKUP;
	rp->r_flags |= RREADDIRPLUS;
	rdc = avl_first(&rp->r_dir);
	while (rdc != NULL) {
		nrdc = AVL_NEXT(&rp->r_dir, rdc);
		avl_remove(&rp->r_dir, rdc);
		rddir_cache_rele(rdc);
		rdc = nrdc;
	}
	mutex_exit(&rp->r_statelock);
}

/*
 * Do a cache check based on the post-operation attributes.
 * Then make them the new cached attributes.  If no attributes
 * were returned, then mark the attributes as timed out.
 */
void
nfs3_cache_post_op_attr(vnode_t *vp, post_op_attr *poap, hrtime_t t, cred_t *cr)
{
	vattr_t attr;

	if (!poap->attributes) {
		PURGE_ATTRCACHE(vp);
		return;
	}
	(void) nfs3_cache_fattr3(vp, &poap->attr, &attr, t, cr);
}

/*
 * Same as above, but using a vattr
 */
void
nfs3_cache_post_op_vattr(vnode_t *vp, post_op_vattr *poap, hrtime_t t,
    cred_t *cr)
{
	if (!poap->attributes) {
		PURGE_ATTRCACHE(vp);
		return;
	}
	nfs_attr_cache(vp, poap->fres.vap, t, cr);
}

/*
 * Do a cache check based on the weak cache consistency attributes.
 * These consist of a small set of pre-operation attributes and the
 * full set of post-operation attributes.
 *
 * If we are given the pre-operation attributes, then use them to
 * check the validity of the various caches.  Then, if we got the
 * post-operation attributes, make them the new cached attributes.
 * If we didn't get the post-operation attributes, then mark the
 * attribute cache as timed out so that the next reference will
 * cause a GETATTR to the server to refresh with the current
 * attributes.
 *
 * Otherwise, if we didn't get the pre-operation attributes, but
 * we did get the post-operation attributes, then use these
 * attributes to check the validity of the various caches.  This
 * will probably cause a flush of the caches because if the
 * operation succeeded, the attributes of the object were changed
 * in some way from the old post-operation attributes.  This
 * should be okay because it is the safe thing to do.  After
 * checking the data caches, then we make these the new cached
 * attributes.
 *
 * Otherwise, we didn't get either the pre- or post-operation
 * attributes.  Simply mark the attribute cache as timed out so
 * the next reference will cause a GETATTR to the server to
 * refresh with the current attributes.
 *
 * If an error occurred trying to convert the over the wire
 * attributes to a vattr, then simply mark the attribute cache as
 * timed out.
 */
void
nfs3_cache_wcc_data(vnode_t *vp, wcc_data *wccp, hrtime_t t, cred_t *cr)
{
	vattr_t bva;
	vattr_t ava;

	if (wccp->after.attributes) {
		if (fattr3_to_vattr(vp, &wccp->after.attr, &ava)) {
			PURGE_ATTRCACHE(vp);
			return;
		}
		if (wccp->before.attributes) {
			bva.va_ctime.tv_sec = wccp->before.attr.ctime.seconds;
			bva.va_ctime.tv_nsec = wccp->before.attr.ctime.nseconds;
			bva.va_mtime.tv_sec = wccp->before.attr.mtime.seconds;
			bva.va_mtime.tv_nsec = wccp->before.attr.mtime.nseconds;
			bva.va_size = wccp->before.attr.size;
			nfs3_attr_cache(vp, &bva, &ava, t, cr);
		} else
			nfs_attr_cache(vp, &ava, t, cr);
	} else {
		PURGE_ATTRCACHE(vp);
	}
}

/*
 * Set attributes cache for given vnode using nfsattr.
 *
 * This routine does not do cache validation with the attributes.
 *
 * If an error occurred trying to convert the over the wire
 * attributes to a vattr, then simply mark the attribute cache as
 * timed out.
 */
void
nfs_attrcache(vnode_t *vp, struct nfsfattr *na, hrtime_t t)
{
	rnode_t *rp;
	struct vattr va;

	if (!nattr_to_vattr(vp, na, &va)) {
		rp = VTOR(vp);
		mutex_enter(&rp->r_statelock);
		if (rp->r_mtime <= t)
			nfs_attrcache_va(vp, &va);
		mutex_exit(&rp->r_statelock);
	} else {
		PURGE_ATTRCACHE(vp);
	}
}

/*
 * Set attributes cache for given vnode using fattr3.
 *
 * This routine does not do cache validation with the attributes.
 *
 * If an error occurred trying to convert the over the wire
 * attributes to a vattr, then simply mark the attribute cache as
 * timed out.
 */
void
nfs3_attrcache(vnode_t *vp, fattr3 *na, hrtime_t t)
{
	rnode_t *rp;
	struct vattr va;

	if (!fattr3_to_vattr(vp, na, &va)) {
		rp = VTOR(vp);
		mutex_enter(&rp->r_statelock);
		if (rp->r_mtime <= t)
			nfs_attrcache_va(vp, &va);
		mutex_exit(&rp->r_statelock);
	} else {
		PURGE_ATTRCACHE(vp);
	}
}

/*
 * Do a cache check based on attributes returned over the wire.  The
 * new attributes are cached.
 *
 * If an error occurred trying to convert the over the wire attributes
 * to a vattr, then just return that error.
 *
 * As a side affect, the vattr argument is filled in with the converted
 * attributes.
 */
int
nfs_cache_fattr(vnode_t *vp, struct nfsfattr *na, vattr_t *vap, hrtime_t t,
    cred_t *cr)
{
	int error;

	error = nattr_to_vattr(vp, na, vap);
	if (error)
		return (error);
	nfs_attr_cache(vp, vap, t, cr);
	return (0);
}

/*
 * Do a cache check based on attributes returned over the wire.  The
 * new attributes are cached.
 *
 * If an error occurred trying to convert the over the wire attributes
 * to a vattr, then just return that error.
 *
 * As a side affect, the vattr argument is filled in with the converted
 * attributes.
 */
int
nfs3_cache_fattr3(vnode_t *vp, fattr3 *na, vattr_t *vap, hrtime_t t, cred_t *cr)
{
	int error;

	error = fattr3_to_vattr(vp, na, vap);
	if (error)
		return (error);
	nfs_attr_cache(vp, vap, t, cr);
	return (0);
}

/*
 * Use the passed in virtual attributes to check to see whether the
 * data and metadata caches are valid, cache the new attributes, and
 * then do the cache invalidation if required.
 *
 * The cache validation and caching of the new attributes is done
 * atomically via the use of the mutex, r_statelock.  If required,
 * the cache invalidation is done atomically w.r.t. the cache
 * validation and caching of the attributes via the pseudo lock,
 * r_serial.
 *
 * This routine is used to do cache validation and attributes caching
 * for operations with a single set of post operation attributes.
 */
void
nfs_attr_cache(vnode_t *vp, vattr_t *vap, hrtime_t t, cred_t *cr)
{
	rnode_t *rp;
	int mtime_changed = 0;
	int ctime_changed = 0;
	vsecattr_t *vsp;
	int was_serial;
	len_t preattr_rsize;
	boolean_t writeattr_set = B_FALSE;
	boolean_t cachepurge_set = B_FALSE;

	rp = VTOR(vp);

	mutex_enter(&rp->r_statelock);

	if (rp->r_serial != curthread) {
		klwp_t *lwp = ttolwp(curthread);

		was_serial = 0;
		if (lwp != NULL)
			lwp->lwp_nostop++;
		while (rp->r_serial != NULL) {
			if (!cv_wait_sig(&rp->r_cv, &rp->r_statelock)) {
				mutex_exit(&rp->r_statelock);
				if (lwp != NULL)
					lwp->lwp_nostop--;
				return;
			}
		}
		if (lwp != NULL)
			lwp->lwp_nostop--;
	} else
		was_serial = 1;

	if (rp->r_mtime > t) {
		if (!CACHE_VALID(rp, vap->va_mtime, vap->va_size))
			PURGE_ATTRCACHE_LOCKED(rp);
		mutex_exit(&rp->r_statelock);
		return;
	}

	/*
	 * Write thread after writing data to file on remote server,
	 * will always set RWRITEATTR to indicate that file on remote
	 * server was modified with a WRITE operation and would have
	 * marked attribute cache as timed out. If RWRITEATTR
	 * is set, then do not check for mtime and ctime change.
	 */
	if (!(rp->r_flags & RWRITEATTR)) {
		if (!CACHE_VALID(rp, vap->va_mtime, vap->va_size))
			mtime_changed = 1;

		if (rp->r_attr.va_ctime.tv_sec != vap->va_ctime.tv_sec ||
		    rp->r_attr.va_ctime.tv_nsec != vap->va_ctime.tv_nsec)
			ctime_changed = 1;
	} else {
		writeattr_set = B_TRUE;
	}

	preattr_rsize = rp->r_size;

	nfs_attrcache_va(vp, vap);

	/*
	 * If we have updated filesize in nfs_attrcache_va, as soon as we
	 * drop statelock we will be in transition of purging all
	 * our caches and updating them. It is possible for another
	 * thread to pick this new file size and read in zeroed data.
	 * stall other threads till cache purge is complete.
	 */
	if ((vp->v_type == VREG) && (rp->r_size != preattr_rsize)) {
		/*
		 * If RWRITEATTR was set and we have updated the file
		 * size, Server's returned file size need not necessarily
		 * be because of this Client's WRITE. We need to purge
		 * all caches.
		 */
		if (writeattr_set)
			mtime_changed = 1;

		if (mtime_changed && !(rp->r_flags & RINCACHEPURGE)) {
			rp->r_flags |= RINCACHEPURGE;
			cachepurge_set = B_TRUE;
		}
	}

	if (!mtime_changed && !ctime_changed) {
		mutex_exit(&rp->r_statelock);
		return;
	}

	rp->r_serial = curthread;

	mutex_exit(&rp->r_statelock);

	if (mtime_changed)
		nfs_purge_caches(vp, NFS_NOPURGE_DNLC, cr);

	if ((rp->r_flags & RINCACHEPURGE) && cachepurge_set) {
		mutex_enter(&rp->r_statelock);
		rp->r_flags &= ~RINCACHEPURGE;
		cv_broadcast(&rp->r_cv);
		mutex_exit(&rp->r_statelock);
		cachepurge_set = B_FALSE;
	}

	if (ctime_changed) {
		(void) nfs_access_purge_rp(rp);
		if (rp->r_secattr != NULL) {
			mutex_enter(&rp->r_statelock);
			vsp = rp->r_secattr;
			rp->r_secattr = NULL;
			mutex_exit(&rp->r_statelock);
			if (vsp != NULL)
				nfs_acl_free(vsp);
		}
	}

	if (!was_serial) {
		mutex_enter(&rp->r_statelock);
		rp->r_serial = NULL;
		cv_broadcast(&rp->r_cv);
		mutex_exit(&rp->r_statelock);
	}
}

/*
 * Use the passed in "before" virtual attributes to check to see
 * whether the data and metadata caches are valid, cache the "after"
 * new attributes, and then do the cache invalidation if required.
 *
 * The cache validation and caching of the new attributes is done
 * atomically via the use of the mutex, r_statelock.  If required,
 * the cache invalidation is done atomically w.r.t. the cache
 * validation and caching of the attributes via the pseudo lock,
 * r_serial.
 *
 * This routine is used to do cache validation and attributes caching
 * for operations with both pre operation attributes and post operation
 * attributes.
 */
static void
nfs3_attr_cache(vnode_t *vp, vattr_t *bvap, vattr_t *avap, hrtime_t t,
    cred_t *cr)
{
	rnode_t *rp;
	int mtime_changed = 0;
	int ctime_changed = 0;
	vsecattr_t *vsp;
	int was_serial;
	len_t preattr_rsize;
	boolean_t writeattr_set = B_FALSE;
	boolean_t cachepurge_set = B_FALSE;

	rp = VTOR(vp);

	mutex_enter(&rp->r_statelock);

	if (rp->r_serial != curthread) {
		klwp_t *lwp = ttolwp(curthread);

		was_serial = 0;
		if (lwp != NULL)
			lwp->lwp_nostop++;
		while (rp->r_serial != NULL) {
			if (!cv_wait_sig(&rp->r_cv, &rp->r_statelock)) {
				mutex_exit(&rp->r_statelock);
				if (lwp != NULL)
					lwp->lwp_nostop--;
				return;
			}
		}
		if (lwp != NULL)
			lwp->lwp_nostop--;
	} else
		was_serial = 1;

	if (rp->r_mtime > t) {
		if (!CACHE_VALID(rp, avap->va_mtime, avap->va_size))
			PURGE_ATTRCACHE_LOCKED(rp);
		mutex_exit(&rp->r_statelock);
		return;
	}

	/*
	 * Write thread after writing data to file on remote server,
	 * will always set RWRITEATTR to indicate that file on remote
	 * server was modified with a WRITE operation and would have
	 * marked attribute cache as timed out. If RWRITEATTR
	 * is set, then do not check for mtime and ctime change.
	 */
	if (!(rp->r_flags & RWRITEATTR)) {
		if (!CACHE_VALID(rp, bvap->va_mtime, bvap->va_size))
			mtime_changed = 1;

		if (rp->r_attr.va_ctime.tv_sec != bvap->va_ctime.tv_sec ||
		    rp->r_attr.va_ctime.tv_nsec != bvap->va_ctime.tv_nsec)
			ctime_changed = 1;
	} else {
		writeattr_set = B_TRUE;
	}

	preattr_rsize = rp->r_size;

	nfs_attrcache_va(vp, avap);

	/*
	 * If we have updated filesize in nfs_attrcache_va, as soon as we
	 * drop statelock we will be in transition of purging all
	 * our caches and updating them. It is possible for another
	 * thread to pick this new file size and read in zeroed data.
	 * stall other threads till cache purge is complete.
	 */
	if ((vp->v_type == VREG) && (rp->r_size != preattr_rsize)) {
		/*
		 * If RWRITEATTR was set and we have updated the file
		 * size, Server's returned file size need not necessarily
		 * be because of this Client's WRITE. We need to purge
		 * all caches.
		 */
		if (writeattr_set)
			mtime_changed = 1;

		if (mtime_changed && !(rp->r_flags & RINCACHEPURGE)) {
			rp->r_flags |= RINCACHEPURGE;
			cachepurge_set = B_TRUE;
		}
	}

	if (!mtime_changed && !ctime_changed) {
		mutex_exit(&rp->r_statelock);
		return;
	}

	rp->r_serial = curthread;

	mutex_exit(&rp->r_statelock);

	if (mtime_changed)
		nfs_purge_caches(vp, NFS_NOPURGE_DNLC, cr);

	if ((rp->r_flags & RINCACHEPURGE) && cachepurge_set) {
		mutex_enter(&rp->r_statelock);
		rp->r_flags &= ~RINCACHEPURGE;
		cv_broadcast(&rp->r_cv);
		mutex_exit(&rp->r_statelock);
		cachepurge_set = B_FALSE;
	}

	if (ctime_changed) {
		(void) nfs_access_purge_rp(rp);
		if (rp->r_secattr != NULL) {
			mutex_enter(&rp->r_statelock);
			vsp = rp->r_secattr;
			rp->r_secattr = NULL;
			mutex_exit(&rp->r_statelock);
			if (vsp != NULL)
				nfs_acl_free(vsp);
		}
	}

	if (!was_serial) {
		mutex_enter(&rp->r_statelock);
		rp->r_serial = NULL;
		cv_broadcast(&rp->r_cv);
		mutex_exit(&rp->r_statelock);
	}
}

/*
 * Set attributes cache for given vnode using virtual attributes.
 *
 * Set the timeout value on the attribute cache and fill it
 * with the passed in attributes.
 *
 * The caller must be holding r_statelock.
 */
void
nfs_attrcache_va(vnode_t *vp, struct vattr *va)
{
	rnode_t *rp;
	mntinfo_t *mi;
	hrtime_t delta;
	hrtime_t now;

	rp = VTOR(vp);

	ASSERT(MUTEX_HELD(&rp->r_statelock));

	now = gethrtime();

	mi = VTOMI(vp);

	/*
	 * Delta is the number of nanoseconds that we will
	 * cache the attributes of the file.  It is based on
	 * the number of nanoseconds since the last time that
	 * we detected a change.  The assumption is that files
	 * that changed recently are likely to change again.
	 * There is a minimum and a maximum for regular files
	 * and for directories which is enforced though.
	 *
	 * Using the time since last change was detected
	 * eliminates direct comparison or calculation
	 * using mixed client and server times.  NFS does
	 * not make any assumptions regarding the client
	 * and server clocks being synchronized.
	 */
	if (va->va_mtime.tv_sec != rp->r_attr.va_mtime.tv_sec ||
	    va->va_mtime.tv_nsec != rp->r_attr.va_mtime.tv_nsec ||
	    va->va_size != rp->r_attr.va_size)
		rp->r_mtime = now;

	if ((mi->mi_flags & MI_NOAC) || (vp->v_flag & VNOCACHE))
		delta = 0;
	else {
		delta = now - rp->r_mtime;
		if (vp->v_type == VDIR) {
			if (delta < mi->mi_acdirmin)
				delta = mi->mi_acdirmin;
			else if (delta > mi->mi_acdirmax)
				delta = mi->mi_acdirmax;
		} else {
			if (delta < mi->mi_acregmin)
				delta = mi->mi_acregmin;
			else if (delta > mi->mi_acregmax)
				delta = mi->mi_acregmax;
		}
	}
	rp->r_attrtime = now + delta;
	rp->r_attr = *va;
	/*
	 * Update the size of the file if there is no cached data or if
	 * the cached data is clean and there is no data being written
	 * out.
	 */
	if (rp->r_size != va->va_size &&
	    (!vn_has_cached_data(vp) ||
	    (!(rp->r_flags & RDIRTY) && rp->r_count == 0)))
		rp->r_size = va->va_size;
	nfs_setswaplike(vp, va);
	rp->r_flags &= ~RWRITEATTR;
}

/*
 * Fill in attribute from the cache.
 * If valid, then return 0 to indicate that no error occurred,
 * otherwise return 1 to indicate that an error occurred.
 */
static int
nfs_getattr_cache(vnode_t *vp, struct vattr *vap)
{
	rnode_t *rp;
	uint_t mask = vap->va_mask;

	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);
	if (ATTRCACHE_VALID(vp)) {
		/*
		 * Cached attributes are valid
		 */
		*vap = rp->r_attr;
		/*
		 * Set the caller's va_mask to the set of attributes
		 * that were requested ANDed with the attributes that
		 * are available.  If attributes were requested that
		 * are not available, those bits must be turned off
		 * in the callers va_mask.
		 */
		vap->va_mask &= mask;
		mutex_exit(&rp->r_statelock);
		return (0);
	}
	mutex_exit(&rp->r_statelock);
	return (1);
}

/*
 * Get attributes over-the-wire and update attributes cache
 * if no error occurred in the over-the-wire operation.
 * Return 0 if successful, otherwise error.
 */
int
nfs_getattr_otw(vnode_t *vp, struct vattr *vap, cred_t *cr)
{
	int error;
	struct nfsattrstat ns;
	int douprintf;
	mntinfo_t *mi;
	failinfo_t fi;
	hrtime_t t;

	mi = VTOMI(vp);
	fi.vp = vp;
	fi.fhp = NULL;		/* no need to update, filehandle not copied */
	fi.copyproc = nfscopyfh;
	fi.lookupproc = nfslookup;
	fi.xattrdirproc = acl_getxattrdir2;

	if (mi->mi_flags & MI_ACL) {
		error = acl_getattr2_otw(vp, vap, cr);
		if (mi->mi_flags & MI_ACL)
			return (error);
	}

	douprintf = 1;

	t = gethrtime();

	error = rfs2call(mi, RFS_GETATTR,
	    xdr_fhandle, (caddr_t)VTOFH(vp),
	    xdr_attrstat, (caddr_t)&ns, cr,
	    &douprintf, &ns.ns_status, 0, &fi);

	if (!error) {
		error = geterrno(ns.ns_status);
		if (!error)
			error = nfs_cache_fattr(vp, &ns.ns_attr, vap, t, cr);
		else {
			PURGE_STALE_FH(error, vp, cr);
		}
	}

	return (error);
}

/*
 * Return either cached ot remote attributes. If get remote attr
 * use them to check and invalidate caches, then cache the new attributes.
 */
int
nfsgetattr(vnode_t *vp, struct vattr *vap, cred_t *cr)
{
	int error;
	rnode_t *rp;

	/*
	 * If we've got cached attributes, we're done, otherwise go
	 * to the server to get attributes, which will update the cache
	 * in the process.
	 */
	error = nfs_getattr_cache(vp, vap);
	if (error)
		error = nfs_getattr_otw(vp, vap, cr);

	/* Return the client's view of file size */
	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);
	vap->va_size = rp->r_size;
	mutex_exit(&rp->r_statelock);

	return (error);
}

/*
 * Get attributes over-the-wire and update attributes cache
 * if no error occurred in the over-the-wire operation.
 * Return 0 if successful, otherwise error.
 */
int
nfs3_getattr_otw(vnode_t *vp, struct vattr *vap, cred_t *cr)
{
	int error;
	GETATTR3args args;
	GETATTR3vres res;
	int douprintf;
	failinfo_t fi;
	hrtime_t t;

	args.object = *VTOFH3(vp);
	fi.vp = vp;
	fi.fhp = (caddr_t)&args.object;
	fi.copyproc = nfs3copyfh;
	fi.lookupproc = nfs3lookup;
	fi.xattrdirproc = acl_getxattrdir3;
	res.fres.vp = vp;
	res.fres.vap = vap;

	douprintf = 1;

	t = gethrtime();

	error = rfs3call(VTOMI(vp), NFSPROC3_GETATTR,
	    xdr_nfs_fh3, (caddr_t)&args,
	    xdr_GETATTR3vres, (caddr_t)&res, cr,
	    &douprintf, &res.status, 0, &fi);

	if (error)
		return (error);

	error = geterrno3(res.status);
	if (error) {
		PURGE_STALE_FH(error, vp, cr);
		return (error);
	}

	/*
	 * Catch status codes that indicate fattr3 to vattr translation failure
	 */
	if (res.fres.status)
		return (res.fres.status);

	nfs_attr_cache(vp, vap, t, cr);
	return (0);
}

/*
 * Return either cached or remote attributes. If get remote attr
 * use them to check and invalidate caches, then cache the new attributes.
 */
int
nfs3getattr(vnode_t *vp, struct vattr *vap, cred_t *cr)
{
	int error;
	rnode_t *rp;

	/*
	 * If we've got cached attributes, we're done, otherwise go
	 * to the server to get attributes, which will update the cache
	 * in the process.
	 */
	error = nfs_getattr_cache(vp, vap);
	if (error)
		error = nfs3_getattr_otw(vp, vap, cr);

	/* Return the client's view of file size */
	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);
	vap->va_size = rp->r_size;
	mutex_exit(&rp->r_statelock);

	return (error);
}

vtype_t nf_to_vt[] = {
	VNON, VREG, VDIR, VBLK, VCHR, VLNK, VSOCK
};
/*
 * Convert NFS Version 2 over the network attributes to the local
 * virtual attributes.  The mapping between the UID_NOBODY/GID_NOBODY
 * network representation and the local representation is done here.
 * Returns 0 for success, error if failed due to overflow.
 */
int
nattr_to_vattr(vnode_t *vp, struct nfsfattr *na, struct vattr *vap)
{
	/* overflow in time attributes? */
#ifndef _LP64
	if (!NFS2_FATTR_TIME_OK(na))
		return (EOVERFLOW);
#endif

	vap->va_mask = AT_ALL;

	if (na->na_type < NFNON || na->na_type > NFSOC)
		vap->va_type = VBAD;
	else
		vap->va_type = nf_to_vt[na->na_type];
	vap->va_mode = na->na_mode;
	vap->va_uid = (na->na_uid == NFS_UID_NOBODY) ? UID_NOBODY : na->na_uid;
	vap->va_gid = (na->na_gid == NFS_GID_NOBODY) ? GID_NOBODY : na->na_gid;
	vap->va_fsid = vp->v_vfsp->vfs_dev;
	vap->va_nodeid = na->na_nodeid;
	vap->va_nlink = na->na_nlink;
	vap->va_size = na->na_size;	/* keep for cache validation */
	/*
	 * nfs protocol defines times as unsigned so don't extend sign,
	 * unless sysadmin set nfs_allow_preepoch_time.
	 */
	NFS_TIME_T_CONVERT(vap->va_atime.tv_sec, na->na_atime.tv_sec);
	vap->va_atime.tv_nsec = (uint32_t)(na->na_atime.tv_usec * 1000);
	NFS_TIME_T_CONVERT(vap->va_mtime.tv_sec, na->na_mtime.tv_sec);
	vap->va_mtime.tv_nsec = (uint32_t)(na->na_mtime.tv_usec * 1000);
	NFS_TIME_T_CONVERT(vap->va_ctime.tv_sec, na->na_ctime.tv_sec);
	vap->va_ctime.tv_nsec = (uint32_t)(na->na_ctime.tv_usec * 1000);
	/*
	 * Shannon's law - uncompress the received dev_t
	 * if the top half of is zero indicating a response
	 * from an `older style' OS. Except for when it is a
	 * `new style' OS sending the maj device of zero,
	 * in which case the algorithm still works because the
	 * fact that it is a new style server
	 * is hidden by the minor device not being greater
	 * than 255 (a requirement in this case).
	 */
	if ((na->na_rdev & 0xffff0000) == 0)
		vap->va_rdev = nfsv2_expdev(na->na_rdev);
	else
		vap->va_rdev = expldev(na->na_rdev);

	vap->va_nblocks = na->na_blocks;
	switch (na->na_type) {
	case NFBLK:
		vap->va_blksize = DEV_BSIZE;
		break;

	case NFCHR:
		vap->va_blksize = MAXBSIZE;
		break;

	case NFSOC:
	default:
		vap->va_blksize = na->na_blocksize;
		break;
	}
	/*
	 * This bit of ugliness is a hack to preserve the
	 * over-the-wire protocols for named-pipe vnodes.
	 * It remaps the special over-the-wire type to the
	 * VFIFO type. (see note in nfs.h)
	 */
	if (NA_ISFIFO(na)) {
		vap->va_type = VFIFO;
		vap->va_mode = (vap->va_mode & ~S_IFMT) | S_IFIFO;
		vap->va_rdev = 0;
		vap->va_blksize = na->na_blocksize;
	}
	vap->va_seq = 0;
	return (0);
}

/*
 * Convert NFS Version 3 over the network attributes to the local
 * virtual attributes.  The mapping between the UID_NOBODY/GID_NOBODY
 * network representation and the local representation is done here.
 */
vtype_t nf3_to_vt[] = {
	VBAD, VREG, VDIR, VBLK, VCHR, VLNK, VSOCK, VFIFO
};

int
fattr3_to_vattr(vnode_t *vp, fattr3 *na, struct vattr *vap)
{

#ifndef _LP64
	/* overflow in time attributes? */
	if (!NFS3_FATTR_TIME_OK(na))
		return (EOVERFLOW);
#endif
	if (!NFS3_SIZE_OK(na->size))
		/* file too big */
		return (EFBIG);

	vap->va_mask = AT_ALL;

	if (na->type < NF3REG || na->type > NF3FIFO)
		vap->va_type = VBAD;
	else
		vap->va_type = nf3_to_vt[na->type];
	vap->va_mode = na->mode;
	vap->va_uid = (na->uid == NFS_UID_NOBODY) ? UID_NOBODY : (uid_t)na->uid;
	vap->va_gid = (na->gid == NFS_GID_NOBODY) ? GID_NOBODY : (gid_t)na->gid;
	vap->va_fsid = vp->v_vfsp->vfs_dev;
	vap->va_nodeid = na->fileid;
	vap->va_nlink = na->nlink;
	vap->va_size = na->size;

	/*
	 * nfs protocol defines times as unsigned so don't extend sign,
	 * unless sysadmin set nfs_allow_preepoch_time.
	 */
	NFS_TIME_T_CONVERT(vap->va_atime.tv_sec, na->atime.seconds);
	vap->va_atime.tv_nsec = (uint32_t)na->atime.nseconds;
	NFS_TIME_T_CONVERT(vap->va_mtime.tv_sec, na->mtime.seconds);
	vap->va_mtime.tv_nsec = (uint32_t)na->mtime.nseconds;
	NFS_TIME_T_CONVERT(vap->va_ctime.tv_sec, na->ctime.seconds);
	vap->va_ctime.tv_nsec = (uint32_t)na->ctime.nseconds;

	switch (na->type) {
	case NF3BLK:
		vap->va_rdev = makedevice(na->rdev.specdata1,
		    na->rdev.specdata2);
		vap->va_blksize = DEV_BSIZE;
		vap->va_nblocks = 0;
		break;
	case NF3CHR:
		vap->va_rdev = makedevice(na->rdev.specdata1,
		    na->rdev.specdata2);
		vap->va_blksize = MAXBSIZE;
		vap->va_nblocks = 0;
		break;
	case NF3REG:
	case NF3DIR:
	case NF3LNK:
		vap->va_rdev = 0;
		vap->va_blksize = MAXBSIZE;
		vap->va_nblocks = (u_longlong_t)
		    ((na->used + (size3)DEV_BSIZE - (size3)1) /
		    (size3)DEV_BSIZE);
		break;
	case NF3SOCK:
	case NF3FIFO:
	default:
		vap->va_rdev = 0;
		vap->va_blksize = MAXBSIZE;
		vap->va_nblocks = 0;
		break;
	}
	vap->va_seq = 0;
	return (0);
}

/*
 * Asynchronous I/O parameters.  nfs_async_threads is the high-water mark
 * for the demand-based allocation of async threads per-mount.  The
 * nfs_async_timeout is the amount of time a thread will live after it
 * becomes idle, unless new I/O requests are received before the thread
 * dies.  See nfs_async_putpage and nfs_async_start.
 */

int nfs_async_timeout = -1;	/* uninitialized */

static void	nfs_async_start(struct vfs *);
static void	nfs_async_pgops_start(struct vfs *);
static void	nfs_async_common_start(struct vfs *, int);

static void
free_async_args(struct nfs_async_reqs *args)
{
	rnode_t *rp;

	if (args->a_io != NFS_INACTIVE) {
		rp = VTOR(args->a_vp);
		mutex_enter(&rp->r_statelock);
		rp->r_count--;
		if (args->a_io == NFS_PUTAPAGE ||
		    args->a_io == NFS_PAGEIO)
			rp->r_awcount--;
		cv_broadcast(&rp->r_cv);
		mutex_exit(&rp->r_statelock);
		VN_RELE(args->a_vp);
	}
	crfree(args->a_cred);
	kmem_free(args, sizeof (*args));
}

/*
 * Cross-zone thread creation and NFS access is disallowed, yet fsflush() and
 * pageout(), running in the global zone, have legitimate reasons to do
 * VOP_PUTPAGE(B_ASYNC) on other zones' NFS mounts.  We avoid the problem by
 * use of a a per-mount "asynchronous requests manager thread" which is
 * signaled by the various asynchronous work routines when there is
 * asynchronous work to be done.  It is responsible for creating new
 * worker threads if necessary, and notifying existing worker threads
 * that there is work to be done.
 *
 * In other words, it will "take the specifications from the customers and
 * give them to the engineers."
 *
 * Worker threads die off of their own accord if they are no longer
 * needed.
 *
 * This thread is killed when the zone is going away or the filesystem
 * is being unmounted.
 */
void
nfs_async_manager(vfs_t *vfsp)
{
	callb_cpr_t cprinfo;
	mntinfo_t *mi;
	uint_t max_threads;

	mi = VFTOMI(vfsp);

	CALLB_CPR_INIT(&cprinfo, &mi->mi_async_lock, callb_generic_cpr,
	    "nfs_async_manager");

	mutex_enter(&mi->mi_async_lock);
	/*
	 * We want to stash the max number of threads that this mount was
	 * allowed so we can use it later when the variable is set to zero as
	 * part of the zone/mount going away.
	 *
	 * We want to be able to create at least one thread to handle
	 * asynchronous inactive calls.
	 */
	max_threads = MAX(mi->mi_max_threads, 1);
	/*
	 * We don't want to wait for mi_max_threads to go to zero, since that
	 * happens as part of a failed unmount, but this thread should only
	 * exit when the mount/zone is really going away.
	 *
	 * Once MI_ASYNC_MGR_STOP is set, no more async operations will be
	 * attempted: the various _async_*() functions know to do things
	 * inline if mi_max_threads == 0.  Henceforth we just drain out the
	 * outstanding requests.
	 *
	 * Note that we still create zthreads even if we notice the zone is
	 * shutting down (MI_ASYNC_MGR_STOP is set); this may cause the zone
	 * shutdown sequence to take slightly longer in some cases, but
	 * doesn't violate the protocol, as all threads will exit as soon as
	 * they're done processing the remaining requests.
	 */
	for (;;) {
		while (mi->mi_async_req_count > 0) {
			/*
			 * Paranoia: If the mount started out having
			 * (mi->mi_max_threads == 0), and the value was
			 * later changed (via a debugger or somesuch),
			 * we could be confused since we will think we
			 * can't create any threads, and the calling
			 * code (which looks at the current value of
			 * mi->mi_max_threads, now non-zero) thinks we
			 * can.
			 *
			 * So, because we're paranoid, we create threads
			 * up to the maximum of the original and the
			 * current value. This means that future
			 * (debugger-induced) lowerings of
			 * mi->mi_max_threads are ignored for our
			 * purposes, but who told them they could change
			 * random values on a live kernel anyhow?
			 */
			if (mi->mi_threads[NFS_ASYNC_QUEUE] <
			    MAX(mi->mi_max_threads, max_threads)) {
				mi->mi_threads[NFS_ASYNC_QUEUE]++;
				mutex_exit(&mi->mi_async_lock);
				VFS_HOLD(vfsp);	/* hold for new thread */
				(void) zthread_create(NULL, 0, nfs_async_start,
				    vfsp, 0, minclsyspri);
				mutex_enter(&mi->mi_async_lock);
			} else if (mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE] <
			    NUM_ASYNC_PGOPS_THREADS) {
				mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE]++;
				mutex_exit(&mi->mi_async_lock);
				VFS_HOLD(vfsp); /* hold for new thread */
				(void) zthread_create(NULL, 0,
				    nfs_async_pgops_start, vfsp, 0,
				    minclsyspri);
				mutex_enter(&mi->mi_async_lock);
			}
			NFS_WAKE_ASYNC_WORKER(mi->mi_async_work_cv);
			ASSERT(mi->mi_async_req_count != 0);
			mi->mi_async_req_count--;
		}

		mutex_enter(&mi->mi_lock);
		if (mi->mi_flags & MI_ASYNC_MGR_STOP) {
			mutex_exit(&mi->mi_lock);
			break;
		}
		mutex_exit(&mi->mi_lock);

		CALLB_CPR_SAFE_BEGIN(&cprinfo);
		cv_wait(&mi->mi_async_reqs_cv, &mi->mi_async_lock);
		CALLB_CPR_SAFE_END(&cprinfo, &mi->mi_async_lock);
	}
	/*
	 * Let everyone know we're done.
	 */
	mi->mi_manager_thread = NULL;
	cv_broadcast(&mi->mi_async_cv);

	/*
	 * There is no explicit call to mutex_exit(&mi->mi_async_lock)
	 * since CALLB_CPR_EXIT is actually responsible for releasing
	 * 'mi_async_lock'.
	 */
	CALLB_CPR_EXIT(&cprinfo);
	VFS_RELE(vfsp);	/* release thread's hold */
	zthread_exit();
}

/*
 * Signal (and wait for) the async manager thread to clean up and go away.
 */
void
nfs_async_manager_stop(vfs_t *vfsp)
{
	mntinfo_t *mi = VFTOMI(vfsp);

	mutex_enter(&mi->mi_async_lock);
	mutex_enter(&mi->mi_lock);
	mi->mi_flags |= MI_ASYNC_MGR_STOP;
	mutex_exit(&mi->mi_lock);
	cv_broadcast(&mi->mi_async_reqs_cv);
	while (mi->mi_manager_thread != NULL)
		cv_wait(&mi->mi_async_cv, &mi->mi_async_lock);
	mutex_exit(&mi->mi_async_lock);
}

int
nfs_async_readahead(vnode_t *vp, u_offset_t blkoff, caddr_t addr,
	struct seg *seg, cred_t *cr, void (*readahead)(vnode_t *,
	u_offset_t, caddr_t, struct seg *, cred_t *))
{
	rnode_t *rp;
	mntinfo_t *mi;
	struct nfs_async_reqs *args;

	rp = VTOR(vp);
	ASSERT(rp->r_freef == NULL);

	mi = VTOMI(vp);

	/*
	 * If addr falls in a different segment, don't bother doing readahead.
	 */
	if (addr >= seg->s_base + seg->s_size)
		return (-1);

	/*
	 * If we can't allocate a request structure, punt on the readahead.
	 */
	if ((args = kmem_alloc(sizeof (*args), KM_NOSLEEP)) == NULL)
		return (-1);

	/*
	 * If a lock operation is pending, don't initiate any new
	 * readaheads.  Otherwise, bump r_count to indicate the new
	 * asynchronous I/O.
	 */
	if (!nfs_rw_tryenter(&rp->r_lkserlock, RW_READER)) {
		kmem_free(args, sizeof (*args));
		return (-1);
	}
	mutex_enter(&rp->r_statelock);
	rp->r_count++;
	mutex_exit(&rp->r_statelock);
	nfs_rw_exit(&rp->r_lkserlock);

	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	VN_HOLD(vp);
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_READ_AHEAD;
	args->a_nfs_readahead = readahead;
	args->a_nfs_blkoff = blkoff;
	args->a_nfs_seg = seg;
	args->a_nfs_addr = addr;

	mutex_enter(&mi->mi_async_lock);

	/*
	 * If asyncio has been disabled, don't bother readahead.
	 */
	if (mi->mi_max_threads == 0) {
		mutex_exit(&mi->mi_async_lock);
		goto noasync;
	}

	/*
	 * Link request structure into the async list and
	 * wakeup async thread to do the i/o.
	 */
	if (mi->mi_async_reqs[NFS_READ_AHEAD] == NULL) {
		mi->mi_async_reqs[NFS_READ_AHEAD] = args;
		mi->mi_async_tail[NFS_READ_AHEAD] = args;
	} else {
		mi->mi_async_tail[NFS_READ_AHEAD]->a_next = args;
		mi->mi_async_tail[NFS_READ_AHEAD] = args;
	}

	if (mi->mi_io_kstats) {
		mutex_enter(&mi->mi_lock);
		kstat_waitq_enter(KSTAT_IO_PTR(mi->mi_io_kstats));
		mutex_exit(&mi->mi_lock);
	}

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
	return (0);

noasync:
	mutex_enter(&rp->r_statelock);
	rp->r_count--;
	cv_broadcast(&rp->r_cv);
	mutex_exit(&rp->r_statelock);
	VN_RELE(vp);
	crfree(cr);
	kmem_free(args, sizeof (*args));
	return (-1);
}

int
nfs_async_putapage(vnode_t *vp, page_t *pp, u_offset_t off, size_t len,
	int flags, cred_t *cr, int (*putapage)(vnode_t *, page_t *,
	u_offset_t, size_t, int, cred_t *))
{
	rnode_t *rp;
	mntinfo_t *mi;
	struct nfs_async_reqs *args;

	ASSERT(flags & B_ASYNC);
	ASSERT(vp->v_vfsp != NULL);

	rp = VTOR(vp);
	ASSERT(rp->r_count > 0);

	mi = VTOMI(vp);

	/*
	 * If we can't allocate a request structure, do the putpage
	 * operation synchronously in this thread's context.
	 */
	if ((args = kmem_alloc(sizeof (*args), KM_NOSLEEP)) == NULL)
		goto noasync;

	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	VN_HOLD(vp);
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_PUTAPAGE;
	args->a_nfs_putapage = putapage;
	args->a_nfs_pp = pp;
	args->a_nfs_off = off;
	args->a_nfs_len = (uint_t)len;
	args->a_nfs_flags = flags;

	mutex_enter(&mi->mi_async_lock);

	/*
	 * If asyncio has been disabled, then make a synchronous request.
	 * This check is done a second time in case async io was diabled
	 * while this thread was blocked waiting for memory pressure to
	 * reduce or for the queue to drain.
	 */
	if (mi->mi_max_threads == 0) {
		mutex_exit(&mi->mi_async_lock);
		goto noasync;
	}

	/*
	 * Link request structure into the async list and
	 * wakeup async thread to do the i/o.
	 */
	if (mi->mi_async_reqs[NFS_PUTAPAGE] == NULL) {
		mi->mi_async_reqs[NFS_PUTAPAGE] = args;
		mi->mi_async_tail[NFS_PUTAPAGE] = args;
	} else {
		mi->mi_async_tail[NFS_PUTAPAGE]->a_next = args;
		mi->mi_async_tail[NFS_PUTAPAGE] = args;
	}

	mutex_enter(&rp->r_statelock);
	rp->r_count++;
	rp->r_awcount++;
	mutex_exit(&rp->r_statelock);

	if (mi->mi_io_kstats) {
		mutex_enter(&mi->mi_lock);
		kstat_waitq_enter(KSTAT_IO_PTR(mi->mi_io_kstats));
		mutex_exit(&mi->mi_lock);
	}

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
	return (0);

noasync:
	if (args != NULL) {
		VN_RELE(vp);
		crfree(cr);
		kmem_free(args, sizeof (*args));
	}

	if (curproc == proc_pageout || curproc == proc_fsflush) {
		/*
		 * If we get here in the context of the pageout/fsflush,
		 * we refuse to do a sync write, because this may hang
		 * pageout (and the machine). In this case, we just
		 * re-mark the page as dirty and punt on the page.
		 *
		 * Make sure B_FORCE isn't set.  We can re-mark the
		 * pages as dirty and unlock the pages in one swoop by
		 * passing in B_ERROR to pvn_write_done().  However,
		 * we should make sure B_FORCE isn't set - we don't
		 * want the page tossed before it gets written out.
		 */
		if (flags & B_FORCE)
			flags &= ~(B_INVAL | B_FORCE);
		pvn_write_done(pp, flags | B_ERROR);
		return (0);
	}
	if (nfs_zone() != mi->mi_zone) {
		/*
		 * So this was a cross-zone sync putpage.  We pass in B_ERROR
		 * to pvn_write_done() to re-mark the pages as dirty and unlock
		 * them.
		 *
		 * We don't want to clear B_FORCE here as the caller presumably
		 * knows what they're doing if they set it.
		 */
		pvn_write_done(pp, flags | B_ERROR);
		return (EPERM);
	}
	return ((*putapage)(vp, pp, off, len, flags, cr));
}

int
nfs_async_pageio(vnode_t *vp, page_t *pp, u_offset_t io_off, size_t io_len,
	int flags, cred_t *cr, int (*pageio)(vnode_t *, page_t *, u_offset_t,
	size_t, int, cred_t *))
{
	rnode_t *rp;
	mntinfo_t *mi;
	struct nfs_async_reqs *args;

	ASSERT(flags & B_ASYNC);
	ASSERT(vp->v_vfsp != NULL);

	rp = VTOR(vp);
	ASSERT(rp->r_count > 0);

	mi = VTOMI(vp);

	/*
	 * If we can't allocate a request structure, do the pageio
	 * request synchronously in this thread's context.
	 */
	if ((args = kmem_alloc(sizeof (*args), KM_NOSLEEP)) == NULL)
		goto noasync;

	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	VN_HOLD(vp);
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_PAGEIO;
	args->a_nfs_pageio = pageio;
	args->a_nfs_pp = pp;
	args->a_nfs_off = io_off;
	args->a_nfs_len = (uint_t)io_len;
	args->a_nfs_flags = flags;

	mutex_enter(&mi->mi_async_lock);

	/*
	 * If asyncio has been disabled, then make a synchronous request.
	 * This check is done a second time in case async io was diabled
	 * while this thread was blocked waiting for memory pressure to
	 * reduce or for the queue to drain.
	 */
	if (mi->mi_max_threads == 0) {
		mutex_exit(&mi->mi_async_lock);
		goto noasync;
	}

	/*
	 * Link request structure into the async list and
	 * wakeup async thread to do the i/o.
	 */
	if (mi->mi_async_reqs[NFS_PAGEIO] == NULL) {
		mi->mi_async_reqs[NFS_PAGEIO] = args;
		mi->mi_async_tail[NFS_PAGEIO] = args;
	} else {
		mi->mi_async_tail[NFS_PAGEIO]->a_next = args;
		mi->mi_async_tail[NFS_PAGEIO] = args;
	}

	mutex_enter(&rp->r_statelock);
	rp->r_count++;
	rp->r_awcount++;
	mutex_exit(&rp->r_statelock);

	if (mi->mi_io_kstats) {
		mutex_enter(&mi->mi_lock);
		kstat_waitq_enter(KSTAT_IO_PTR(mi->mi_io_kstats));
		mutex_exit(&mi->mi_lock);
	}

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
	return (0);

noasync:
	if (args != NULL) {
		VN_RELE(vp);
		crfree(cr);
		kmem_free(args, sizeof (*args));
	}

	/*
	 * If we can't do it ASYNC, for reads we do nothing (but cleanup
	 * the page list), for writes we do it synchronously, except for
	 * proc_pageout/proc_fsflush as described below.
	 */
	if (flags & B_READ) {
		pvn_read_done(pp, flags | B_ERROR);
		return (0);
	}

	if (curproc == proc_pageout || curproc == proc_fsflush) {
		/*
		 * If we get here in the context of the pageout/fsflush,
		 * we refuse to do a sync write, because this may hang
		 * pageout/fsflush (and the machine). In this case, we just
		 * re-mark the page as dirty and punt on the page.
		 *
		 * Make sure B_FORCE isn't set.  We can re-mark the
		 * pages as dirty and unlock the pages in one swoop by
		 * passing in B_ERROR to pvn_write_done().  However,
		 * we should make sure B_FORCE isn't set - we don't
		 * want the page tossed before it gets written out.
		 */
		if (flags & B_FORCE)
			flags &= ~(B_INVAL | B_FORCE);
		pvn_write_done(pp, flags | B_ERROR);
		return (0);
	}

	if (nfs_zone() != mi->mi_zone) {
		/*
		 * So this was a cross-zone sync pageio.  We pass in B_ERROR
		 * to pvn_write_done() to re-mark the pages as dirty and unlock
		 * them.
		 *
		 * We don't want to clear B_FORCE here as the caller presumably
		 * knows what they're doing if they set it.
		 */
		pvn_write_done(pp, flags | B_ERROR);
		return (EPERM);
	}
	return ((*pageio)(vp, pp, io_off, io_len, flags, cr));
}

void
nfs_async_readdir(vnode_t *vp, rddir_cache *rdc, cred_t *cr,
	int (*readdir)(vnode_t *, rddir_cache *, cred_t *))
{
	rnode_t *rp;
	mntinfo_t *mi;
	struct nfs_async_reqs *args;

	rp = VTOR(vp);
	ASSERT(rp->r_freef == NULL);

	mi = VTOMI(vp);

	/*
	 * If we can't allocate a request structure, do the readdir
	 * operation synchronously in this thread's context.
	 */
	if ((args = kmem_alloc(sizeof (*args), KM_NOSLEEP)) == NULL)
		goto noasync;

	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	VN_HOLD(vp);
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_READDIR;
	args->a_nfs_readdir = readdir;
	args->a_nfs_rdc = rdc;

	mutex_enter(&mi->mi_async_lock);

	/*
	 * If asyncio has been disabled, then make a synchronous request.
	 */
	if (mi->mi_max_threads == 0) {
		mutex_exit(&mi->mi_async_lock);
		goto noasync;
	}

	/*
	 * Link request structure into the async list and
	 * wakeup async thread to do the i/o.
	 */
	if (mi->mi_async_reqs[NFS_READDIR] == NULL) {
		mi->mi_async_reqs[NFS_READDIR] = args;
		mi->mi_async_tail[NFS_READDIR] = args;
	} else {
		mi->mi_async_tail[NFS_READDIR]->a_next = args;
		mi->mi_async_tail[NFS_READDIR] = args;
	}

	mutex_enter(&rp->r_statelock);
	rp->r_count++;
	mutex_exit(&rp->r_statelock);

	if (mi->mi_io_kstats) {
		mutex_enter(&mi->mi_lock);
		kstat_waitq_enter(KSTAT_IO_PTR(mi->mi_io_kstats));
		mutex_exit(&mi->mi_lock);
	}

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
	return;

noasync:
	if (args != NULL) {
		VN_RELE(vp);
		crfree(cr);
		kmem_free(args, sizeof (*args));
	}

	rdc->entries = NULL;
	mutex_enter(&rp->r_statelock);
	ASSERT(rdc->flags & RDDIR);
	rdc->flags &= ~RDDIR;
	rdc->flags |= RDDIRREQ;
	/*
	 * Check the flag to see if RDDIRWAIT is set. If RDDIRWAIT
	 * is set, wakeup the thread sleeping in cv_wait_sig().
	 * The woken up thread will reset the flag to RDDIR and will
	 * continue with the readdir opeartion.
	 */
	if (rdc->flags & RDDIRWAIT) {
		rdc->flags &= ~RDDIRWAIT;
		cv_broadcast(&rdc->cv);
	}
	mutex_exit(&rp->r_statelock);
	rddir_cache_rele(rdc);
}

void
nfs_async_commit(vnode_t *vp, page_t *plist, offset3 offset, count3 count,
	cred_t *cr, void (*commit)(vnode_t *, page_t *, offset3, count3,
	cred_t *))
{
	rnode_t *rp;
	mntinfo_t *mi;
	struct nfs_async_reqs *args;
	page_t *pp;

	rp = VTOR(vp);
	mi = VTOMI(vp);

	/*
	 * If we can't allocate a request structure, do the commit
	 * operation synchronously in this thread's context.
	 */
	if ((args = kmem_alloc(sizeof (*args), KM_NOSLEEP)) == NULL)
		goto noasync;

	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	VN_HOLD(vp);
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_COMMIT;
	args->a_nfs_commit = commit;
	args->a_nfs_plist = plist;
	args->a_nfs_offset = offset;
	args->a_nfs_count = count;

	mutex_enter(&mi->mi_async_lock);

	/*
	 * If asyncio has been disabled, then make a synchronous request.
	 * This check is done a second time in case async io was diabled
	 * while this thread was blocked waiting for memory pressure to
	 * reduce or for the queue to drain.
	 */
	if (mi->mi_max_threads == 0) {
		mutex_exit(&mi->mi_async_lock);
		goto noasync;
	}

	/*
	 * Link request structure into the async list and
	 * wakeup async thread to do the i/o.
	 */
	if (mi->mi_async_reqs[NFS_COMMIT] == NULL) {
		mi->mi_async_reqs[NFS_COMMIT] = args;
		mi->mi_async_tail[NFS_COMMIT] = args;
	} else {
		mi->mi_async_tail[NFS_COMMIT]->a_next = args;
		mi->mi_async_tail[NFS_COMMIT] = args;
	}

	mutex_enter(&rp->r_statelock);
	rp->r_count++;
	mutex_exit(&rp->r_statelock);

	if (mi->mi_io_kstats) {
		mutex_enter(&mi->mi_lock);
		kstat_waitq_enter(KSTAT_IO_PTR(mi->mi_io_kstats));
		mutex_exit(&mi->mi_lock);
	}

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
	return;

noasync:
	if (args != NULL) {
		VN_RELE(vp);
		crfree(cr);
		kmem_free(args, sizeof (*args));
	}

	if (curproc == proc_pageout || curproc == proc_fsflush ||
	    nfs_zone() != mi->mi_zone) {
		while (plist != NULL) {
			pp = plist;
			page_sub(&plist, pp);
			pp->p_fsdata = C_COMMIT;
			page_unlock(pp);
		}
		return;
	}
	(*commit)(vp, plist, offset, count, cr);
}

void
nfs_async_inactive(vnode_t *vp, cred_t *cr,
    void (*inactive)(vnode_t *, cred_t *, caller_context_t *))
{
	mntinfo_t *mi;
	struct nfs_async_reqs *args;

	mi = VTOMI(vp);

	args = kmem_alloc(sizeof (*args), KM_SLEEP);
	args->a_next = NULL;
#ifdef DEBUG
	args->a_queuer = curthread;
#endif
	args->a_vp = vp;
	ASSERT(cr != NULL);
	crhold(cr);
	args->a_cred = cr;
	args->a_io = NFS_INACTIVE;
	args->a_nfs_inactive = inactive;

	/*
	 * Note that we don't check mi->mi_max_threads here, since we
	 * *need* to get rid of this vnode regardless of whether someone
	 * set nfs3_max_threads/nfs_max_threads to zero in /etc/system.
	 *
	 * The manager thread knows about this and is willing to create
	 * at least one thread to accommodate us.
	 */
	mutex_enter(&mi->mi_async_lock);
	if (mi->mi_manager_thread == NULL) {
		rnode_t *rp = VTOR(vp);

		mutex_exit(&mi->mi_async_lock);
		crfree(cr);	/* drop our reference */
		kmem_free(args, sizeof (*args));
		/*
		 * We can't do an over-the-wire call since we're in the wrong
		 * zone, so we need to clean up state as best we can and then
		 * throw away the vnode.
		 */
		mutex_enter(&rp->r_statelock);
		if (rp->r_unldvp != NULL) {
			vnode_t *unldvp;
			char *unlname;
			cred_t *unlcred;

			unldvp = rp->r_unldvp;
			rp->r_unldvp = NULL;
			unlname = rp->r_unlname;
			rp->r_unlname = NULL;
			unlcred = rp->r_unlcred;
			rp->r_unlcred = NULL;
			mutex_exit(&rp->r_statelock);

			VN_RELE(unldvp);
			kmem_free(unlname, MAXNAMELEN);
			crfree(unlcred);
		} else {
			mutex_exit(&rp->r_statelock);
		}
		/*
		 * No need to explicitly throw away any cached pages.  The
		 * eventual rinactive() will attempt a synchronous
		 * VOP_PUTPAGE() which will immediately fail since the request
		 * is coming from the wrong zone, and then will proceed to call
		 * nfs_invalidate_pages() which will clean things up for us.
		 */
		rp_addfree(VTOR(vp), cr);
		return;
	}

	if (mi->mi_async_reqs[NFS_INACTIVE] == NULL) {
		mi->mi_async_reqs[NFS_INACTIVE] = args;
	} else {
		mi->mi_async_tail[NFS_INACTIVE]->a_next = args;
	}
	mi->mi_async_tail[NFS_INACTIVE] = args;
	/*
	 * Don't increment r_count, since we're trying to get rid of the vnode.
	 */

	mi->mi_async_req_count++;
	ASSERT(mi->mi_async_req_count != 0);
	cv_signal(&mi->mi_async_reqs_cv);
	mutex_exit(&mi->mi_async_lock);
}

static void
nfs_async_start(struct vfs *vfsp)
{
	nfs_async_common_start(vfsp, NFS_ASYNC_QUEUE);
}

static void
nfs_async_pgops_start(struct vfs *vfsp)
{
	nfs_async_common_start(vfsp, NFS_ASYNC_PGOPS_QUEUE);
}

/*
 * The async queues for each mounted file system are arranged as a
 * set of queues, one for each async i/o type.  Requests are taken
 * from the queues in a round-robin fashion.  A number of consecutive
 * requests are taken from each queue before moving on to the next
 * queue.  This functionality may allow the NFS Version 2 server to do
 * write clustering, even if the client is mixing writes and reads
 * because it will take multiple write requests from the queue
 * before processing any of the other async i/o types.
 *
 * XXX The nfs_async_common_start thread is unsafe in the light of the present
 * model defined by cpr to suspend the system. Specifically over the
 * wire calls are cpr-unsafe. The thread should be reevaluated in
 * case of future updates to the cpr model.
 */
static void
nfs_async_common_start(struct vfs *vfsp, int async_queue)
{
	struct nfs_async_reqs *args;
	mntinfo_t *mi = VFTOMI(vfsp);
	clock_t time_left = 1;
	callb_cpr_t cprinfo;
	int i;
	int async_types;
	kcondvar_t *async_work_cv;

	if (async_queue == NFS_ASYNC_QUEUE) {
		async_types = NFS_ASYNC_TYPES;
		async_work_cv = &mi->mi_async_work_cv[NFS_ASYNC_QUEUE];
	} else {
		async_types = NFS_ASYNC_PGOPS_TYPES;
		async_work_cv = &mi->mi_async_work_cv[NFS_ASYNC_PGOPS_QUEUE];
	}

	/*
	 * Dynamic initialization of nfs_async_timeout to allow nfs to be
	 * built in an implementation independent manner.
	 */
	if (nfs_async_timeout == -1)
		nfs_async_timeout = NFS_ASYNC_TIMEOUT;

	CALLB_CPR_INIT(&cprinfo, &mi->mi_async_lock, callb_generic_cpr, "nas");

	mutex_enter(&mi->mi_async_lock);
	for (;;) {
		/*
		 * Find the next queue containing an entry.  We start
		 * at the current queue pointer and then round robin
		 * through all of them until we either find a non-empty
		 * queue or have looked through all of them.
		 */
		for (i = 0; i < async_types; i++) {
			args = *mi->mi_async_curr[async_queue];
			if (args != NULL)
				break;
			mi->mi_async_curr[async_queue]++;
			if (mi->mi_async_curr[async_queue] ==
			    &mi->mi_async_reqs[async_types]) {
				mi->mi_async_curr[async_queue] =
				    &mi->mi_async_reqs[0];
			}
		}
		/*
		 * If we didn't find a entry, then block until woken up
		 * again and then look through the queues again.
		 */
		if (args == NULL) {
			/*
			 * Exiting is considered to be safe for CPR as well
			 */
			CALLB_CPR_SAFE_BEGIN(&cprinfo);

			/*
			 * Wakeup thread waiting to unmount the file
			 * system only if all async threads are inactive.
			 *
			 * If we've timed-out and there's nothing to do,
			 * then get rid of this thread.
			 */
			if (mi->mi_max_threads == 0 || time_left <= 0) {
				--mi->mi_threads[async_queue];

				if (mi->mi_threads[NFS_ASYNC_QUEUE] == 0 &&
				    mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE] == 0)
					cv_signal(&mi->mi_async_cv);
				CALLB_CPR_EXIT(&cprinfo);
				VFS_RELE(vfsp);	/* release thread's hold */
				zthread_exit();
				/* NOTREACHED */
			}
			time_left = cv_reltimedwait(async_work_cv,
			    &mi->mi_async_lock, nfs_async_timeout,
			    TR_CLOCK_TICK);

			CALLB_CPR_SAFE_END(&cprinfo, &mi->mi_async_lock);

			continue;
		}
		time_left = 1;

		/*
		 * Remove the request from the async queue and then
		 * update the current async request queue pointer.  If
		 * the current queue is empty or we have removed enough
		 * consecutive entries from it, then reset the counter
		 * for this queue and then move the current pointer to
		 * the next queue.
		 */
		*mi->mi_async_curr[async_queue] = args->a_next;
		if (*mi->mi_async_curr[async_queue] == NULL ||
		    --mi->mi_async_clusters[args->a_io] == 0) {
			mi->mi_async_clusters[args->a_io] =
			    mi->mi_async_init_clusters;
			mi->mi_async_curr[async_queue]++;
			if (mi->mi_async_curr[async_queue] ==
			    &mi->mi_async_reqs[async_types]) {
				mi->mi_async_curr[async_queue] =
				    &mi->mi_async_reqs[0];
			}
		}

		if (args->a_io != NFS_INACTIVE && mi->mi_io_kstats) {
			mutex_enter(&mi->mi_lock);
			kstat_waitq_exit(KSTAT_IO_PTR(mi->mi_io_kstats));
			mutex_exit(&mi->mi_lock);
		}

		mutex_exit(&mi->mi_async_lock);

		/*
		 * Obtain arguments from the async request structure.
		 */
		if (args->a_io == NFS_READ_AHEAD && mi->mi_max_threads > 0) {
			(*args->a_nfs_readahead)(args->a_vp, args->a_nfs_blkoff,
			    args->a_nfs_addr, args->a_nfs_seg,
			    args->a_cred);
		} else if (args->a_io == NFS_PUTAPAGE) {
			(void) (*args->a_nfs_putapage)(args->a_vp,
			    args->a_nfs_pp, args->a_nfs_off,
			    args->a_nfs_len, args->a_nfs_flags,
			    args->a_cred);
		} else if (args->a_io == NFS_PAGEIO) {
			(void) (*args->a_nfs_pageio)(args->a_vp,
			    args->a_nfs_pp, args->a_nfs_off,
			    args->a_nfs_len, args->a_nfs_flags,
			    args->a_cred);
		} else if (args->a_io == NFS_READDIR) {
			(void) ((*args->a_nfs_readdir)(args->a_vp,
			    args->a_nfs_rdc, args->a_cred));
		} else if (args->a_io == NFS_COMMIT) {
			(*args->a_nfs_commit)(args->a_vp, args->a_nfs_plist,
			    args->a_nfs_offset, args->a_nfs_count,
			    args->a_cred);
		} else if (args->a_io == NFS_INACTIVE) {
			(*args->a_nfs_inactive)(args->a_vp, args->a_cred, NULL);
		}

		/*
		 * Now, release the vnode and free the credentials
		 * structure.
		 */
		free_async_args(args);
		/*
		 * Reacquire the mutex because it will be needed above.
		 */
		mutex_enter(&mi->mi_async_lock);
	}
}

void
nfs_async_stop(struct vfs *vfsp)
{
	mntinfo_t *mi = VFTOMI(vfsp);

	/*
	 * Wait for all outstanding async operations to complete and for the
	 * worker threads to exit.
	 */
	mutex_enter(&mi->mi_async_lock);
	mi->mi_max_threads = 0;
	NFS_WAKEALL_ASYNC_WORKERS(mi->mi_async_work_cv);
	while (mi->mi_threads[NFS_ASYNC_QUEUE] != 0 ||
	    mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE] != 0)
		cv_wait(&mi->mi_async_cv, &mi->mi_async_lock);
	mutex_exit(&mi->mi_async_lock);
}

/*
 * nfs_async_stop_sig:
 * Wait for all outstanding putpage operation to complete. If a signal
 * is deliver we will abort and return non-zero. If we can put all the
 * pages we will return 0. This routine is called from nfs_unmount and
 * nfs3_unmount to make these operations interruptible.
 */
int
nfs_async_stop_sig(struct vfs *vfsp)
{
	mntinfo_t *mi = VFTOMI(vfsp);
	ushort_t omax;
	int rval;

	/*
	 * Wait for all outstanding async operations to complete and for the
	 * worker threads to exit.
	 */
	mutex_enter(&mi->mi_async_lock);
	omax = mi->mi_max_threads;
	mi->mi_max_threads = 0;
	/*
	 * Tell all the worker threads to exit.
	 */
	NFS_WAKEALL_ASYNC_WORKERS(mi->mi_async_work_cv);
	while (mi->mi_threads[NFS_ASYNC_QUEUE] != 0 ||
	    mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE] != 0) {
		if (!cv_wait_sig(&mi->mi_async_cv, &mi->mi_async_lock))
			break;
	}
	rval = (mi->mi_threads[NFS_ASYNC_QUEUE] != 0 ||
	    mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE]  != 0); /* Interrupted */
	if (rval)
		mi->mi_max_threads = omax;
	mutex_exit(&mi->mi_async_lock);

	return (rval);
}

int
writerp(rnode_t *rp, caddr_t base, int tcount, struct uio *uio, int pgcreated)
{
	int pagecreate;
	int n;
	int saved_n;
	caddr_t saved_base;
	u_offset_t offset;
	int error;
	int sm_error;
	vnode_t *vp = RTOV(rp);

	ASSERT(tcount <= MAXBSIZE && tcount <= uio->uio_resid);
	ASSERT(nfs_rw_lock_held(&rp->r_rwlock, RW_WRITER));
	if (!vpm_enable) {
		ASSERT(((uintptr_t)base & MAXBOFFSET) + tcount <= MAXBSIZE);
	}

	/*
	 * Move bytes in at most PAGESIZE chunks. We must avoid
	 * spanning pages in uiomove() because page faults may cause
	 * the cache to be invalidated out from under us. The r_size is not
	 * updated until after the uiomove. If we push the last page of a
	 * file before r_size is correct, we will lose the data written past
	 * the current (and invalid) r_size.
	 */
	do {
		offset = uio->uio_loffset;
		pagecreate = 0;

		/*
		 * n is the number of bytes required to satisfy the request
		 *   or the number of bytes to fill out the page.
		 */
		n = (int)MIN((PAGESIZE - (offset & PAGEOFFSET)), tcount);

		/*
		 * Check to see if we can skip reading in the page
		 * and just allocate the memory.  We can do this
		 * if we are going to rewrite the entire mapping
		 * or if we are going to write to or beyond the current
		 * end of file from the beginning of the mapping.
		 *
		 * The read of r_size is now protected by r_statelock.
		 */
		mutex_enter(&rp->r_statelock);
		/*
		 * When pgcreated is nonzero the caller has already done
		 * a segmap_getmapflt with forcefault 0 and S_WRITE. With
		 * segkpm this means we already have at least one page
		 * created and mapped at base.
		 */
		pagecreate = pgcreated ||
		    ((offset & PAGEOFFSET) == 0 &&
		    (n == PAGESIZE || ((offset + n) >= rp->r_size)));

		mutex_exit(&rp->r_statelock);
		if (!vpm_enable && pagecreate) {
			/*
			 * The last argument tells segmap_pagecreate() to
			 * always lock the page, as opposed to sometimes
			 * returning with the page locked. This way we avoid a
			 * fault on the ensuing uiomove(), but also
			 * more importantly (to fix bug 1094402) we can
			 * call segmap_fault() to unlock the page in all
			 * cases. An alternative would be to modify
			 * segmap_pagecreate() to tell us when it is
			 * locking a page, but that's a fairly major
			 * interface change.
			 */
			if (pgcreated == 0)
				(void) segmap_pagecreate(segkmap, base,
				    (uint_t)n, 1);
			saved_base = base;
			saved_n = n;
		}

		/*
		 * The number of bytes of data in the last page can not
		 * be accurately be determined while page is being
		 * uiomove'd to and the size of the file being updated.
		 * Thus, inform threads which need to know accurately
		 * how much data is in the last page of the file.  They
		 * will not do the i/o immediately, but will arrange for
		 * the i/o to happen later when this modify operation
		 * will have finished.
		 */
		ASSERT(!(rp->r_flags & RMODINPROGRESS));
		mutex_enter(&rp->r_statelock);
		rp->r_flags |= RMODINPROGRESS;
		rp->r_modaddr = (offset & MAXBMASK);
		mutex_exit(&rp->r_statelock);

		if (vpm_enable) {
			/*
			 * Copy data. If new pages are created, part of
			 * the page that is not written will be initizliazed
			 * with zeros.
			 */
			error = vpm_data_copy(vp, offset, n, uio,
			    !pagecreate, NULL, 0, S_WRITE);
		} else {
			error = uiomove(base, n, UIO_WRITE, uio);
		}

		/*
		 * r_size is the maximum number of
		 * bytes known to be in the file.
		 * Make sure it is at least as high as the
		 * first unwritten byte pointed to by uio_loffset.
		 */
		mutex_enter(&rp->r_statelock);
		if (rp->r_size < uio->uio_loffset)
			rp->r_size = uio->uio_loffset;
		rp->r_flags &= ~RMODINPROGRESS;
		rp->r_flags |= RDIRTY;
		mutex_exit(&rp->r_statelock);

		/* n = # of bytes written */
		n = (int)(uio->uio_loffset - offset);

		if (!vpm_enable) {
			base += n;
		}
		tcount -= n;
		/*
		 * If we created pages w/o initializing them completely,
		 * we need to zero the part that wasn't set up.
		 * This happens on a most EOF write cases and if
		 * we had some sort of error during the uiomove.
		 */
		if (!vpm_enable && pagecreate) {
			if ((uio->uio_loffset & PAGEOFFSET) || n == 0)
				(void) kzero(base, PAGESIZE - n);

			if (pgcreated) {
				/*
				 * Caller is responsible for this page,
				 * it was not created in this loop.
				 */
				pgcreated = 0;
			} else {
				/*
				 * For bug 1094402: segmap_pagecreate locks
				 * page. Unlock it. This also unlocks the
				 * pages allocated by page_create_va() in
				 * segmap_pagecreate().
				 */
				sm_error = segmap_fault(kas.a_hat, segkmap,
				    saved_base, saved_n,
				    F_SOFTUNLOCK, S_WRITE);
				if (error == 0)
					error = sm_error;
			}
		}
	} while (tcount > 0 && error == 0);

	return (error);
}

int
nfs_putpages(vnode_t *vp, u_offset_t off, size_t len, int flags, cred_t *cr)
{
	rnode_t *rp;
	page_t *pp;
	u_offset_t eoff;
	u_offset_t io_off;
	size_t io_len;
	int error;
	int rdirty;
	int err;

	rp = VTOR(vp);
	ASSERT(rp->r_count > 0);

	if (!vn_has_cached_data(vp))
		return (0);

	ASSERT(vp->v_type != VCHR);

	/*
	 * If ROUTOFSPACE is set, then all writes turn into B_INVAL
	 * writes.  B_FORCE is set to force the VM system to actually
	 * invalidate the pages, even if the i/o failed.  The pages
	 * need to get invalidated because they can't be written out
	 * because there isn't any space left on either the server's
	 * file system or in the user's disk quota.  The B_FREE bit
	 * is cleared to avoid confusion as to whether this is a
	 * request to place the page on the freelist or to destroy
	 * it.
	 */
	if ((rp->r_flags & ROUTOFSPACE) ||
	    (vp->v_vfsp->vfs_flag & VFS_UNMOUNTED))
		flags = (flags & ~B_FREE) | B_INVAL | B_FORCE;

	if (len == 0) {
		/*
		 * If doing a full file synchronous operation, then clear
		 * the RDIRTY bit.  If a page gets dirtied while the flush
		 * is happening, then RDIRTY will get set again.  The
		 * RDIRTY bit must get cleared before the flush so that
		 * we don't lose this information.
		 *
		 * If there are no full file async write operations
		 * pending and RDIRTY bit is set, clear it.
		 */
		if (off == (u_offset_t)0 &&
		    !(flags & B_ASYNC) &&
		    (rp->r_flags & RDIRTY)) {
			mutex_enter(&rp->r_statelock);
			rdirty = (rp->r_flags & RDIRTY);
			rp->r_flags &= ~RDIRTY;
			mutex_exit(&rp->r_statelock);
		} else if (flags & B_ASYNC && off == (u_offset_t)0) {
			mutex_enter(&rp->r_statelock);
			if (rp->r_flags & RDIRTY && rp->r_awcount == 0) {
				rdirty = (rp->r_flags & RDIRTY);
				rp->r_flags &= ~RDIRTY;
			}
			mutex_exit(&rp->r_statelock);
		} else
			rdirty = 0;

		/*
		 * Search the entire vp list for pages >= off, and flush
		 * the dirty pages.
		 */
		error = pvn_vplist_dirty(vp, off, rp->r_putapage,
		    flags, cr);

		/*
		 * If an error occurred and the file was marked as dirty
		 * before and we aren't forcibly invalidating pages, then
		 * reset the RDIRTY flag.
		 */
		if (error && rdirty &&
		    (flags & (B_INVAL | B_FORCE)) != (B_INVAL | B_FORCE)) {
			mutex_enter(&rp->r_statelock);
			rp->r_flags |= RDIRTY;
			mutex_exit(&rp->r_statelock);
		}
	} else {
		/*
		 * Do a range from [off...off + len) looking for pages
		 * to deal with.
		 */
		error = 0;
#ifdef lint
		io_len = 0;
#endif
		eoff = off + len;
		mutex_enter(&rp->r_statelock);
		for (io_off = off; io_off < eoff && io_off < rp->r_size;
		    io_off += io_len) {
			mutex_exit(&rp->r_statelock);
			/*
			 * If we are not invalidating, synchronously
			 * freeing or writing pages use the routine
			 * page_lookup_nowait() to prevent reclaiming
			 * them from the free list.
			 */
			if ((flags & B_INVAL) || !(flags & B_ASYNC)) {
				pp = page_lookup(vp, io_off,
				    (flags & (B_INVAL | B_FREE)) ?
				    SE_EXCL : SE_SHARED);
			} else {
				pp = page_lookup_nowait(vp, io_off,
				    (flags & B_FREE) ? SE_EXCL : SE_SHARED);
			}

			if (pp == NULL || !pvn_getdirty(pp, flags))
				io_len = PAGESIZE;
			else {
				err = (*rp->r_putapage)(vp, pp, &io_off,
				    &io_len, flags, cr);
				if (!error)
					error = err;
				/*
				 * "io_off" and "io_len" are returned as
				 * the range of pages we actually wrote.
				 * This allows us to skip ahead more quickly
				 * since several pages may've been dealt
				 * with by this iteration of the loop.
				 */
			}
			mutex_enter(&rp->r_statelock);
		}
		mutex_exit(&rp->r_statelock);
	}

	return (error);
}

void
nfs_invalidate_pages(vnode_t *vp, u_offset_t off, cred_t *cr)
{
	rnode_t *rp;

	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);
	while (rp->r_flags & RTRUNCATE)
		cv_wait(&rp->r_cv, &rp->r_statelock);
	rp->r_flags |= RTRUNCATE;
	if (off == (u_offset_t)0) {
		rp->r_flags &= ~RDIRTY;
		if (!(rp->r_flags & RSTALE))
			rp->r_error = 0;
	}
	rp->r_truncaddr = off;
	mutex_exit(&rp->r_statelock);
	(void) pvn_vplist_dirty(vp, off, rp->r_putapage,
	    B_INVAL | B_TRUNC, cr);
	mutex_enter(&rp->r_statelock);
	rp->r_flags &= ~RTRUNCATE;
	cv_broadcast(&rp->r_cv);
	mutex_exit(&rp->r_statelock);
}

static int nfs_write_error_to_cons_only = 0;
#define	MSG(x)	(nfs_write_error_to_cons_only ? (x) : (x) + 1)

/*
 * Print a file handle
 */
void
nfs_printfhandle(nfs_fhandle *fhp)
{
	int *ip;
	char *buf;
	size_t bufsize;
	char *cp;

	/*
	 * 13 == "(file handle:"
	 * maximum of NFS_FHANDLE / sizeof (*ip) elements in fh_buf times
	 *	1 == ' '
	 *	8 == maximum strlen of "%x"
	 * 3 == ")\n\0"
	 */
	bufsize = 13 + ((NFS_FHANDLE_LEN / sizeof (*ip)) * (1 + 8)) + 3;
	buf = kmem_alloc(bufsize, KM_NOSLEEP);
	if (buf == NULL)
		return;

	cp = buf;
	(void) strcpy(cp, "(file handle:");
	while (*cp != '\0')
		cp++;
	for (ip = (int *)fhp->fh_buf;
	    ip < (int *)&fhp->fh_buf[fhp->fh_len];
	    ip++) {
		(void) sprintf(cp, " %x", *ip);
		while (*cp != '\0')
			cp++;
	}
	(void) strcpy(cp, ")\n");

	zcmn_err(getzoneid(), CE_CONT, MSG("^%s"), buf);

	kmem_free(buf, bufsize);
}

/*
 * Notify the system administrator that an NFS write error has
 * occurred.
 */

/* seconds between ENOSPC/EDQUOT messages */
clock_t nfs_write_error_interval = 5;

void
nfs_write_error(vnode_t *vp, int error, cred_t *cr)
{
	mntinfo_t *mi;
	clock_t now;

	mi = VTOMI(vp);
	/*
	 * In case of forced unmount or zone shutdown, do not print any
	 * messages since it can flood the console with error messages.
	 */
	if (FS_OR_ZONE_GONE(mi->mi_vfsp))
		return;

	/*
	 * No use in flooding the console with ENOSPC
	 * messages from the same file system.
	 */
	now = ddi_get_lbolt();
	if ((error != ENOSPC && error != EDQUOT) ||
	    now - mi->mi_printftime > 0) {
		zoneid_t zoneid = mi->mi_zone->zone_id;

#ifdef DEBUG
		nfs_perror(error, "NFS%ld write error on host %s: %m.\n",
		    mi->mi_vers, VTOR(vp)->r_server->sv_hostname, NULL);
#else
		nfs_perror(error, "NFS write error on host %s: %m.\n",
		    VTOR(vp)->r_server->sv_hostname, NULL);
#endif
		if (error == ENOSPC || error == EDQUOT) {
			zcmn_err(zoneid, CE_CONT,
			    MSG("^File: userid=%d, groupid=%d\n"),
			    crgetuid(cr), crgetgid(cr));
			if (crgetuid(CRED()) != crgetuid(cr) ||
			    crgetgid(CRED()) != crgetgid(cr)) {
				zcmn_err(zoneid, CE_CONT,
				    MSG("^User: userid=%d, groupid=%d\n"),
				    crgetuid(CRED()), crgetgid(CRED()));
			}
			mi->mi_printftime = now +
			    nfs_write_error_interval * hz;
		}
		nfs_printfhandle(&VTOR(vp)->r_fh);
#ifdef DEBUG
		if (error == EACCES) {
			zcmn_err(zoneid, CE_CONT,
			    MSG("^nfs_bio: cred is%s kcred\n"),
			    cr == kcred ? "" : " not");
		}
#endif
	}
}

/* ARGSUSED */
static void *
nfs_mi_init(zoneid_t zoneid)
{
	struct mi_globals *mig;

	mig = kmem_alloc(sizeof (*mig), KM_SLEEP);
	mutex_init(&mig->mig_lock, NULL, MUTEX_DEFAULT, NULL);
	list_create(&mig->mig_list, sizeof (mntinfo_t),
	    offsetof(mntinfo_t, mi_zone_node));
	mig->mig_destructor_called = B_FALSE;
	return (mig);
}

/*
 * Callback routine to tell all NFS mounts in the zone to stop creating new
 * threads.  Existing threads should exit.
 */
/* ARGSUSED */
static void
nfs_mi_shutdown(zoneid_t zoneid, void *data)
{
	struct mi_globals *mig = data;
	mntinfo_t *mi;

	ASSERT(mig != NULL);
again:
	mutex_enter(&mig->mig_lock);
	for (mi = list_head(&mig->mig_list); mi != NULL;
	    mi = list_next(&mig->mig_list, mi)) {

		/*
		 * If we've done the shutdown work for this FS, skip.
		 * Once we go off the end of the list, we're done.
		 */
		if (mi->mi_flags & MI_DEAD)
			continue;

		/*
		 * We will do work, so not done.  Get a hold on the FS.
		 */
		VFS_HOLD(mi->mi_vfsp);

		/*
		 * purge the DNLC for this filesystem
		 */
		(void) dnlc_purge_vfsp(mi->mi_vfsp, 0);

		mutex_enter(&mi->mi_async_lock);
		/*
		 * Tell existing async worker threads to exit.
		 */
		mi->mi_max_threads = 0;
		NFS_WAKEALL_ASYNC_WORKERS(mi->mi_async_work_cv);
		/*
		 * Set MI_ASYNC_MGR_STOP so the async manager thread starts
		 * getting ready to exit when it's done with its current work.
		 * Also set MI_DEAD to note we've acted on this FS.
		 */
		mutex_enter(&mi->mi_lock);
		mi->mi_flags |= (MI_ASYNC_MGR_STOP|MI_DEAD);
		mutex_exit(&mi->mi_lock);
		/*
		 * Wake up the async manager thread.
		 */
		cv_broadcast(&mi->mi_async_reqs_cv);
		mutex_exit(&mi->mi_async_lock);

		/*
		 * Drop lock and release FS, which may change list, then repeat.
		 * We're done when every mi has been done or the list is empty.
		 */
		mutex_exit(&mig->mig_lock);
		VFS_RELE(mi->mi_vfsp);
		goto again;
	}
	mutex_exit(&mig->mig_lock);
}

static void
nfs_mi_free_globals(struct mi_globals *mig)
{
	list_destroy(&mig->mig_list);	/* makes sure the list is empty */
	mutex_destroy(&mig->mig_lock);
	kmem_free(mig, sizeof (*mig));

}

/* ARGSUSED */
static void
nfs_mi_destroy(zoneid_t zoneid, void *data)
{
	struct mi_globals *mig = data;

	ASSERT(mig != NULL);
	mutex_enter(&mig->mig_lock);
	if (list_head(&mig->mig_list) != NULL) {
		/* Still waiting for VFS_FREEVFS() */
		mig->mig_destructor_called = B_TRUE;
		mutex_exit(&mig->mig_lock);
		return;
	}
	nfs_mi_free_globals(mig);
}

/*
 * Add an NFS mount to the per-zone list of NFS mounts.
 */
void
nfs_mi_zonelist_add(mntinfo_t *mi)
{
	struct mi_globals *mig;

	mig = zone_getspecific(mi_list_key, mi->mi_zone);
	mutex_enter(&mig->mig_lock);
	list_insert_head(&mig->mig_list, mi);
	mutex_exit(&mig->mig_lock);
}

/*
 * Remove an NFS mount from the per-zone list of NFS mounts.
 */
static void
nfs_mi_zonelist_remove(mntinfo_t *mi)
{
	struct mi_globals *mig;

	mig = zone_getspecific(mi_list_key, mi->mi_zone);
	mutex_enter(&mig->mig_lock);
	list_remove(&mig->mig_list, mi);
	/*
	 * We can be called asynchronously by VFS_FREEVFS() after the zone
	 * shutdown/destroy callbacks have executed; if so, clean up the zone's
	 * mi globals.
	 */
	if (list_head(&mig->mig_list) == NULL &&
	    mig->mig_destructor_called == B_TRUE) {
		nfs_mi_free_globals(mig);
		return;
	}
	mutex_exit(&mig->mig_lock);
}

/*
 * NFS Client initialization routine.  This routine should only be called
 * once.  It performs the following tasks:
 *	- Initalize all global locks
 * 	- Call sub-initialization routines (localize access to variables)
 */
int
nfs_clntinit(void)
{
#ifdef DEBUG
	static boolean_t nfs_clntup = B_FALSE;
#endif
	int error;

#ifdef DEBUG
	ASSERT(nfs_clntup == B_FALSE);
#endif

	error = nfs_subrinit();
	if (error)
		return (error);

	error = nfs_vfsinit();
	if (error) {
		/*
		 * Cleanup nfs_subrinit() work
		 */
		nfs_subrfini();
		return (error);
	}
	zone_key_create(&mi_list_key, nfs_mi_init, nfs_mi_shutdown,
	    nfs_mi_destroy);

	nfs4_clnt_init();

#ifdef DEBUG
	nfs_clntup = B_TRUE;
#endif

	return (0);
}

/*
 * This routine is only called if the NFS Client has been initialized but
 * the module failed to be installed. This routine will cleanup the previously
 * allocated/initialized work.
 */
void
nfs_clntfini(void)
{
	(void) zone_key_delete(mi_list_key);
	nfs_subrfini();
	nfs_vfsfini();
	nfs4_clnt_fini();
}

/*
 * nfs_lockrelease:
 *
 * Release any locks on the given vnode that are held by the current
 * process.
 */
void
nfs_lockrelease(vnode_t *vp, int flag, offset_t offset, cred_t *cr)
{
	flock64_t ld;
	struct shrlock shr;
	char *buf;
	int remote_lock_possible;
	int ret;

	ASSERT((uintptr_t)vp > KERNELBASE);

	/*
	 * Generate an explicit unlock operation for the entire file.  As a
	 * partial optimization, only generate the unlock if there is a
	 * lock registered for the file.  We could check whether this
	 * particular process has any locks on the file, but that would
	 * require the local locking code to provide yet another query
	 * routine.  Note that no explicit synchronization is needed here.
	 * At worst, flk_has_remote_locks() will return a false positive,
	 * in which case the unlock call wastes time but doesn't harm
	 * correctness.
	 *
	 * In addition, an unlock request is generated if the process
	 * is listed as possibly having a lock on the file because the
	 * server and client lock managers may have gotten out of sync.
	 * N.B. It is important to make sure nfs_remove_locking_id() is
	 * called here even if flk_has_remote_locks(vp) reports true.
	 * If it is not called and there is an entry on the process id
	 * list, that entry will never get removed.
	 */
	remote_lock_possible = nfs_remove_locking_id(vp, RLMPL_PID,
	    (char *)&(ttoproc(curthread)->p_pid), NULL, NULL);
	if (remote_lock_possible || flk_has_remote_locks(vp)) {
		ld.l_type = F_UNLCK;	/* set to unlock entire file */
		ld.l_whence = 0;	/* unlock from start of file */
		ld.l_start = 0;
		ld.l_len = 0;		/* do entire file */
		ret = VOP_FRLOCK(vp, F_SETLK, &ld, flag, offset, NULL, cr,
		    NULL);

		if (ret != 0) {
			/*
			 * If VOP_FRLOCK fails, make sure we unregister
			 * local locks before we continue.
			 */
			ld.l_pid = ttoproc(curthread)->p_pid;
			lm_register_lock_locally(vp, NULL, &ld, flag, offset);
#ifdef DEBUG
			nfs_perror(ret,
			    "NFS lock release error on vp %p: %m.\n",
			    (void *)vp, NULL);
#endif
		}

		/*
		 * The call to VOP_FRLOCK may put the pid back on the
		 * list.  We need to remove it.
		 */
		(void) nfs_remove_locking_id(vp, RLMPL_PID,
		    (char *)&(ttoproc(curthread)->p_pid), NULL, NULL);
	}

	/*
	 * As long as the vp has a share matching our pid,
	 * pluck it off and unshare it.  There are circumstances in
	 * which the call to nfs_remove_locking_id() may put the
	 * owner back on the list, in which case we simply do a
	 * redundant and harmless unshare.
	 */
	buf = kmem_alloc(MAX_SHR_OWNER_LEN, KM_SLEEP);
	while (nfs_remove_locking_id(vp, RLMPL_OWNER,
	    (char *)NULL, buf, &shr.s_own_len)) {
		shr.s_owner = buf;
		shr.s_access = 0;
		shr.s_deny = 0;
		shr.s_sysid = 0;
		shr.s_pid = curproc->p_pid;

		ret = VOP_SHRLOCK(vp, F_UNSHARE, &shr, flag, cr, NULL);
#ifdef DEBUG
		if (ret != 0) {
			nfs_perror(ret,
			    "NFS share release error on vp %p: %m.\n",
			    (void *)vp, NULL);
		}
#endif
	}
	kmem_free(buf, MAX_SHR_OWNER_LEN);
}

/*
 * nfs_lockcompletion:
 *
 * If the vnode has a lock that makes it unsafe to cache the file, mark it
 * as non cachable (set VNOCACHE bit).
 */

void
nfs_lockcompletion(vnode_t *vp, int cmd)
{
#ifdef DEBUG
	rnode_t *rp = VTOR(vp);

	ASSERT(nfs_rw_lock_held(&rp->r_lkserlock, RW_WRITER));
#endif

	if (cmd == F_SETLK || cmd == F_SETLKW) {
		if (!lm_safemap(vp)) {
			mutex_enter(&vp->v_lock);
			vp->v_flag |= VNOCACHE;
			mutex_exit(&vp->v_lock);
		} else {
			mutex_enter(&vp->v_lock);
			vp->v_flag &= ~VNOCACHE;
			mutex_exit(&vp->v_lock);
		}
	}
	/*
	 * The cached attributes of the file are stale after acquiring
	 * the lock on the file. They were updated when the file was
	 * opened, but not updated when the lock was acquired. Therefore the
	 * cached attributes are invalidated after the lock is obtained.
	 */
	PURGE_ATTRCACHE(vp);
}

/*
 * The lock manager holds state making it possible for the client
 * and server to be out of sync.  For example, if the response from
 * the server granting a lock request is lost, the server will think
 * the lock is granted and the client will think the lock is lost.
 * The client can tell when it is not positive if it is in sync with
 * the server.
 *
 * To deal with this, a list of processes for which the client is
 * not sure if the server holds a lock is attached to the rnode.
 * When such a process closes the rnode, an unlock request is sent
 * to the server to unlock the entire file.
 *
 * The list is kept as a singularly linked NULL terminated list.
 * Because it is only added to under extreme error conditions, the
 * list shouldn't get very big.  DEBUG kernels print a message if
 * the list gets bigger than nfs_lmpl_high_water.  This is arbitrarily
 * choosen to be 8, but can be tuned at runtime.
 */
#ifdef DEBUG
/* int nfs_lmpl_high_water = 8; */
int nfs_lmpl_high_water = 128;
int nfs_cnt_add_locking_id = 0;
int nfs_len_add_locking_id = 0;
#endif /* DEBUG */

/*
 * Record that the nfs lock manager server may be holding a lock on
 * a vnode for a process.
 *
 * Because the nfs lock manager server holds state, it is possible
 * for the server to get out of sync with the client.  This routine is called
 * from the client when it is no longer sure if the server is in sync
 * with the client.  nfs_lockrelease() will then notice this and send
 * an unlock request when the file is closed
 */
void
nfs_add_locking_id(vnode_t *vp, pid_t pid, int type, char *id, int len)
{
	rnode_t *rp;
	lmpl_t *new;
	lmpl_t *cur;
	lmpl_t **lmplp;
#ifdef DEBUG
	int list_len = 1;
#endif /* DEBUG */

#ifdef DEBUG
	++nfs_cnt_add_locking_id;
#endif /* DEBUG */
	/*
	 * allocate new lmpl_t now so we don't sleep
	 * later after grabbing mutexes
	 */
	ASSERT(len < MAX_SHR_OWNER_LEN);
	new = kmem_alloc(sizeof (*new), KM_SLEEP);
	new->lmpl_type = type;
	new->lmpl_pid = pid;
	new->lmpl_owner = kmem_alloc(len, KM_SLEEP);
	bcopy(id, new->lmpl_owner, len);
	new->lmpl_own_len = len;
	new->lmpl_next = (lmpl_t *)NULL;
#ifdef DEBUG
	if (type == RLMPL_PID) {
		ASSERT(len == sizeof (pid_t));
		ASSERT(pid == *(pid_t *)new->lmpl_owner);
	} else {
		ASSERT(type == RLMPL_OWNER);
	}
#endif

	rp = VTOR(vp);
	mutex_enter(&rp->r_statelock);

	/*
	 * Add this id to the list for this rnode only if the
	 * rnode is active and the id is not already there.
	 */
	ASSERT(rp->r_flags & RHASHED);
	lmplp = &(rp->r_lmpl);
	for (cur = rp->r_lmpl; cur != (lmpl_t *)NULL; cur = cur->lmpl_next) {
		if (cur->lmpl_pid == pid &&
		    cur->lmpl_type == type &&
		    cur->lmpl_own_len == len &&
		    bcmp(cur->lmpl_owner, new->lmpl_owner, len) == 0) {
			kmem_free(new->lmpl_owner, len);
			kmem_free(new, sizeof (*new));
			break;
		}
		lmplp = &cur->lmpl_next;
#ifdef DEBUG
		++list_len;
#endif /* DEBUG */
	}
	if (cur == (lmpl_t *)NULL) {
		*lmplp = new;
#ifdef DEBUG
		if (list_len > nfs_len_add_locking_id) {
			nfs_len_add_locking_id = list_len;
		}
		if (list_len > nfs_lmpl_high_water) {
			cmn_err(CE_WARN, "nfs_add_locking_id: long list "
			    "vp=%p is %d", (void *)vp, list_len);
		}
#endif /* DEBUG */
	}

#ifdef DEBUG
	if (share_debug) {
		int nitems = 0;
		int npids = 0;
		int nowners = 0;

		/*
		 * Count the number of things left on r_lmpl after the remove.
		 */
		for (cur = rp->r_lmpl; cur != (lmpl_t *)NULL;
		    cur = cur->lmpl_next) {
			nitems++;
			if (cur->lmpl_type == RLMPL_PID) {
				npids++;
			} else if (cur->lmpl_type == RLMPL_OWNER) {
				nowners++;
			} else {
				cmn_err(CE_PANIC, "nfs_add_locking_id: "
				    "unrecognized lmpl_type %d",
				    cur->lmpl_type);
			}
		}

		cmn_err(CE_CONT, "nfs_add_locking_id(%s): %d PIDs + %d "
		    "OWNs = %d items left on r_lmpl\n",
		    (type == RLMPL_PID) ? "P" : "O", npids, nowners, nitems);
	}
#endif

	mutex_exit(&rp->r_statelock);
}

/*
 * Remove an id from the lock manager id list.
 *
 * If the id is not in the list return 0.  If it was found and
 * removed, return 1.
 */
static int
nfs_remove_locking_id(vnode_t *vp, int type, char *id, char *rid, int *rlen)
{
	lmpl_t *cur;
	lmpl_t **lmplp;
	rnode_t *rp;
	int rv = 0;

	ASSERT(type == RLMPL_PID || type == RLMPL_OWNER);

	rp = VTOR(vp);

	mutex_enter(&rp->r_statelock);
	ASSERT(rp->r_flags & RHASHED);
	lmplp = &(rp->r_lmpl);

	/*
	 * Search through the list and remove the entry for this id
	 * if it is there.  The special case id == NULL allows removal
	 * of the first share on the r_lmpl list belonging to the
	 * current process (if any), without regard to further details
	 * of its identity.
	 */
	for (cur = rp->r_lmpl; cur != (lmpl_t *)NULL; cur = cur->lmpl_next) {
		if (cur->lmpl_type == type &&
		    cur->lmpl_pid == curproc->p_pid &&
		    (id == (char *)NULL ||
		    bcmp(cur->lmpl_owner, id, cur->lmpl_own_len) == 0)) {
			*lmplp = cur->lmpl_next;
			ASSERT(cur->lmpl_own_len < MAX_SHR_OWNER_LEN);
			if (rid != NULL) {
				bcopy(cur->lmpl_owner, rid, cur->lmpl_own_len);
				*rlen = cur->lmpl_own_len;
			}
			kmem_free(cur->lmpl_owner, cur->lmpl_own_len);
			kmem_free(cur, sizeof (*cur));
			rv = 1;
			break;
		}
		lmplp = &cur->lmpl_next;
	}

#ifdef DEBUG
	if (share_debug) {
		int nitems = 0;
		int npids = 0;
		int nowners = 0;

		/*
		 * Count the number of things left on r_lmpl after the remove.
		 */
		for (cur = rp->r_lmpl; cur != (lmpl_t *)NULL;
		    cur = cur->lmpl_next) {
			nitems++;
			if (cur->lmpl_type == RLMPL_PID) {
				npids++;
			} else if (cur->lmpl_type == RLMPL_OWNER) {
				nowners++;
			} else {
				cmn_err(CE_PANIC,
				    "nrli: unrecognized lmpl_type %d",
				    cur->lmpl_type);
			}
		}

		cmn_err(CE_CONT,
		"nrli(%s): %d PIDs + %d OWNs = %d items left on r_lmpl\n",
		    (type == RLMPL_PID) ? "P" : "O",
		    npids,
		    nowners,
		    nitems);
	}
#endif

	mutex_exit(&rp->r_statelock);
	return (rv);
}

void
nfs_free_mi(mntinfo_t *mi)
{
	ASSERT(mi->mi_flags & MI_ASYNC_MGR_STOP);
	ASSERT(mi->mi_manager_thread == NULL);
	ASSERT(mi->mi_threads[NFS_ASYNC_QUEUE] == 0 &&
	    mi->mi_threads[NFS_ASYNC_PGOPS_QUEUE] == 0);

	/*
	 * Remove the node from the global list before we start tearing it down.
	 */
	nfs_mi_zonelist_remove(mi);
	if (mi->mi_klmconfig) {
		lm_free_config(mi->mi_klmconfig);
		kmem_free(mi->mi_klmconfig, sizeof (struct knetconfig));
	}
	mutex_destroy(&mi->mi_lock);
	mutex_destroy(&mi->mi_remap_lock);
	mutex_destroy(&mi->mi_async_lock);
	cv_destroy(&mi->mi_failover_cv);
	cv_destroy(&mi->mi_async_work_cv[NFS_ASYNC_QUEUE]);
	cv_destroy(&mi->mi_async_work_cv[NFS_ASYNC_PGOPS_QUEUE]);
	cv_destroy(&mi->mi_async_reqs_cv);
	cv_destroy(&mi->mi_async_cv);
	zone_rele(mi->mi_zone);
	kmem_free(mi, sizeof (*mi));
}

static int
mnt_kstat_update(kstat_t *ksp, int rw)
{
	mntinfo_t *mi;
	struct mntinfo_kstat *mik;
	vfs_t *vfsp;
	int i;

	/* this is a read-only kstat. Bail out on a write */
	if (rw == KSTAT_WRITE)
		return (EACCES);

	/*
	 * We don't want to wait here as kstat_chain_lock could be held by
	 * dounmount(). dounmount() takes vfs_reflock before the chain lock
	 * and thus could lead to a deadlock.
	 */
	vfsp = (struct vfs *)ksp->ks_private;


	mi = VFTOMI(vfsp);

	mik = (struct mntinfo_kstat *)ksp->ks_data;

	(void) strcpy(mik->mik_proto, mi->mi_curr_serv->sv_knconf->knc_proto);
	mik->mik_vers = (uint32_t)mi->mi_vers;
	mik->mik_flags = mi->mi_flags;
	mik->mik_secmod = mi->mi_curr_serv->sv_secdata->secmod;
	mik->mik_curread = (uint32_t)mi->mi_curread;
	mik->mik_curwrite = (uint32_t)mi->mi_curwrite;
	mik->mik_retrans = mi->mi_retrans;
	mik->mik_timeo = mi->mi_timeo;
	mik->mik_acregmin = HR2SEC(mi->mi_acregmin);
	mik->mik_acregmax = HR2SEC(mi->mi_acregmax);
	mik->mik_acdirmin = HR2SEC(mi->mi_acdirmin);
	mik->mik_acdirmax = HR2SEC(mi->mi_acdirmax);
	for (i = 0; i < NFS_CALLTYPES + 1; i++) {
		mik->mik_timers[i].srtt = (uint32_t)mi->mi_timers[i].rt_srtt;
		mik->mik_timers[i].deviate =
		    (uint32_t)mi->mi_timers[i].rt_deviate;
		mik->mik_timers[i].rtxcur =
		    (uint32_t)mi->mi_timers[i].rt_rtxcur;
	}
	mik->mik_noresponse = (uint32_t)mi->mi_noresponse;
	mik->mik_failover = (uint32_t)mi->mi_failover;
	mik->mik_remap = (uint32_t)mi->mi_remap;
	(void) strcpy(mik->mik_curserver, mi->mi_curr_serv->sv_hostname);

	return (0);
}

void
nfs_mnt_kstat_init(struct vfs *vfsp)
{
	mntinfo_t *mi = VFTOMI(vfsp);

	/*
	 * Create the version specific kstats.
	 *
	 * PSARC 2001/697 Contract Private Interface
	 * All nfs kstats are under SunMC contract
	 * Please refer to the PSARC listed above and contact
	 * SunMC before making any changes!
	 *
	 * Changes must be reviewed by Solaris File Sharing
	 * Changes must be communicated to contract-2001-697 (at) sun.com
	 *
	 */

	mi->mi_io_kstats = kstat_create_zone("nfs", getminor(vfsp->vfs_dev),
	    NULL, "nfs", KSTAT_TYPE_IO, 1, 0, mi->mi_zone->zone_id);
	if (mi->mi_io_kstats) {
		if (mi->mi_zone->zone_id != GLOBAL_ZONEID)
			kstat_zone_add(mi->mi_io_kstats, GLOBAL_ZONEID);
		mi->mi_io_kstats->ks_lock = &mi->mi_lock;
		kstat_install(mi->mi_io_kstats);
	}

	if ((mi->mi_ro_kstats = kstat_create_zone("nfs",
	    getminor(vfsp->vfs_dev), "mntinfo", "misc", KSTAT_TYPE_RAW,
	    sizeof (struct mntinfo_kstat), 0, mi->mi_zone->zone_id)) != NULL) {
		if (mi->mi_zone->zone_id != GLOBAL_ZONEID)
			kstat_zone_add(mi->mi_ro_kstats, GLOBAL_ZONEID);
		mi->mi_ro_kstats->ks_update = mnt_kstat_update;
		mi->mi_ro_kstats->ks_private = (void *)vfsp;
		kstat_install(mi->mi_ro_kstats);
	}
}

nfs_delmapcall_t *
nfs_init_delmapcall()
{
	nfs_delmapcall_t	*delmap_call;

	delmap_call = kmem_alloc(sizeof (nfs_delmapcall_t), KM_SLEEP);
	delmap_call->call_id = curthread;
	delmap_call->error = 0;

	return (delmap_call);
}

void
nfs_free_delmapcall(nfs_delmapcall_t *delmap_call)
{
	kmem_free(delmap_call, sizeof (nfs_delmapcall_t));
}

/*
 * Searches for the current delmap caller (based on curthread) in the list of
 * callers.  If it is found, we remove it and free the delmap caller.
 * Returns:
 *	0 if the caller wasn't found
 *	1 if the caller was found, removed and freed.  *errp is set to what
 * 	the result of the delmap was.
 */
int
nfs_find_and_delete_delmapcall(rnode_t *rp, int *errp)
{
	nfs_delmapcall_t	*delmap_call;

	/*
	 * If the list doesn't exist yet, we create it and return
	 * that the caller wasn't found.  No list = no callers.
	 */
	mutex_enter(&rp->r_statelock);
	if (!(rp->r_flags & RDELMAPLIST)) {
		/* The list does not exist */
		list_create(&rp->r_indelmap, sizeof (nfs_delmapcall_t),
		    offsetof(nfs_delmapcall_t, call_node));
		rp->r_flags |= RDELMAPLIST;
		mutex_exit(&rp->r_statelock);
		return (0);
	} else {
		/* The list exists so search it */
		for (delmap_call = list_head(&rp->r_indelmap);
		    delmap_call != NULL;
		    delmap_call = list_next(&rp->r_indelmap, delmap_call)) {
			if (delmap_call->call_id == curthread) {
				/* current caller is in the list */
				*errp = delmap_call->error;
				list_remove(&rp->r_indelmap, delmap_call);
				mutex_exit(&rp->r_statelock);
				nfs_free_delmapcall(delmap_call);
				return (1);
			}
		}
	}
	mutex_exit(&rp->r_statelock);
	return (0);
}