//
// 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/CDDL.txt
// 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/CDDL.txt.
// 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 2008 Sun Microsystems, Inc.  All rights reserved.
// Use is subject to license terms.
//

#pragma ident	"@(#)pxvfs.cc	1.39	08/05/20 SMI"

#include <sys/types.h>
#include <sys/systm.h>
#include <sys/vfs.h>
#include <sys/vnode.h>
#include <sys/file.h>
#include <sys/uio.h>
#include <sys/dnlc.h>
#include <sys/mount.h>
#include <sys/statvfs.h>
#include <sys/debug.h>
#include <sys/cmn_err.h>
#include <sys/fs_subr.h>
#include <sys/pathname.h>
#include <sys/fs/ufs_mount.h>
#include <sys/mntent.h>
#include <kstat.h>
#include <sys/ddi.h>
#include <sys/disp.h>

#include <sys/sol_version.h>
#include <sys/os.h>
#include <sys/sol_conv.h>
#include <solobj/solobj_impl.h>
#include <nslib/ns.h>
#include <orb/fault/fault_injection.h>
#include <orb/monitor/monitor.h>
#include <orb/infrastructure/clusterproc.h>

#include "../version.h"
#include <pxfs/common/pxfslib.h>
#include <pxfs/device/device_replica_impl.h>
#include <pxfs/device/device_service_mgr.h>
#include <pxfs/mount/mount_client_impl.h>
#include <pxfs/mount/mount_debug.h>
#include <pxfs/lib/pxfs_debug.h>
#include <pxfs/lib/pxfs_misc.h>
#include <pxfs/client/pxfobj.h>
#include <pxfs/client/pxfobjplus.h>
#include <pxfs/client/fobj_client_impl.h>
#include <pxfs/client/pxreg.h>
#include <pxfs/client/pxdir.h>
#include <pxfs/client/pxchr.h>
#include <pxfs/client/pxlink.h>
#include <pxfs/client/pxspecial.h>
#include <pxfs/client/pxvfs.h>
#include <pxfs/client/fsmgr_client_impl.h>

#ifndef VXFS_DISABLED
#include <pxfs/server/vxfs_dependent_impl.h>
#endif

#if SOL_VERSION >= __s9
#define	PSARC_2001_038
#endif

//
// For update of the mnttab modification time. The function
// vfs_mnttab_modtimeupd is declared static in vfs.c for
// Solaris 8 and 9. For Solaris 10, it is global.
//
#if SOL_VERSION >= __s10
#define	GLOBAL_MNTTAB_MODTIME_INTERFACE
#else
extern timespec_t vfs_mnttab_mtime;
#endif

//
// Constants for initializing various throttling variables.
//
const int	KILOBYTE = 1024;
const int64_t	MEGABYTE = 1024 * KILOBYTE;
const int	ONE_SECOND = 1000000; // 1 second in microseconds

int64_t DATA_RATE_DEFAULT = 20 * MEGABYTE;	// 20mb per second
int64_t DATA_RATE_MINIMUM = 2  * MEGABYTE;	// 2mb per second
int	THROTTLE_MONITOR_INTERVAL = ONE_SECOND; // 1 second by default

//lint -e666
// PXFS is an extensive user of the inline function get_vp() within
// the vnode macros:
//	error = VOP_LOOKUP(get_vp(), ... );
// There are no side effects to calling get_vp() repeatedly, flexelint
// does not know that, but we do.
//

//
// Static data member initializations.
//
pxvfs_inactive_threadpool
	*pxvfs_inactive_threadpool::the_pxvfs_inactive_threadpool = NULL;

pxvfs_list_t	pxvfs::all_pxvfs;	// list of all pxvfs structures
os::rwlock_t	pxvfs::all_pxvfs_lock;	// protects 'all_pxvfs'
int		pxvfs::pxfstype;	// number assigned by Solaris
bool		pxvfs::unmounts_disabled = false;

// This value must be a non-zero positive number
int		cluster_fs_drain_queue_len = 50;

uint_t		pxvfs::pxfobjhsz = 0;
uint_t		pxvfs::pxfobjhsz_max = 0;
uint_t		pxvfs::pxfobjh_len = 4;
kstat_t		*pxvfs::node_stats = NULL;

//
// The flushing of attributes is done by a kernel thread
// with a priority higher than that of applications.
// The attribute flush operation can starve normal activity.
// These values control the flushing of attributes
// The default values were chosen to allow approx. 100,000 files to be flushed
// in 30 seconds, and to spread the work over that period.
//
// Number of files processed per interval
int		pxvfs::sync_all_attr_throttle = 40;

// No of files processed during a sync before sleeping.
int		pxvfs::sync_filesystem_throttle = 40;

// Amount of sleep between intervals
os::usec_t	pxvfs::sync_all_attr_interval[] = {
			(os::usec_t)(120 * 1000),	// 120ms <10,000 files
			(os::usec_t)(60 * 1000),	// 60ms <20,000 files
			(os::usec_t)(40 * 1000),	// 40ms <30,000 files
			(os::usec_t)(30 * 1000),	// 30ms <40,000 files
			(os::usec_t)(20 * 1000)		// 20ms >=40,000 files
		};

bool		pxvfs::sync_all_attr_thread_running = false;
os::mutex_t	pxvfs::sync_all_attr_lock;

const int	inactive_thread_priority = 65;

#ifdef DEBUG
uint32_t	pxvfs_vget_number_calls = 0;
uint32_t	pxvfs_vget_number_fid_hits = 0;
#endif

// Number of async threads per thread-pool.
int pxfs_async_threads = 15;
uint64_t pxvfs::async_task_count = 0;

//
// This should be pxvfs::pxfobj_hash_bkt but our compiler doesn't understand
// that and our lint checker complains about it.
//lint -e1038
pxvfs::pxfobj_hash_bkt	*pxvfs::pxfobj_hash = NULL;
//lint +e1038

//
// Assign default values for throttling variables. For description see
// declaration of these members in pxvfs's class definition below.
//
int64_t pxvfs::data_rate	 = DATA_RATE_DEFAULT;	// 20mb per second
int64_t pxvfs::data_rate_default = DATA_RATE_DEFAULT;	// 20mb per second
int64_t pxvfs::data_rate_minimum = DATA_RATE_MINIMUM;	// 2mb per second
int64_t	pxvfs::bytes_in_window	 = DATA_RATE_DEFAULT;	// 20mb
int	pxvfs::bandwidth_chunk	 = 8 * KILOBYTE;	// 8kb

// Monitor thread wakes every 1 second by default.
int	pxvfs::throttle_monitor_interval = THROTTLE_MONITOR_INTERVAL;

//
// We don't want to keep more than 16MB worth of I/O requests pending
// per client. Assuming an average size of 128kb per I/O request, that
// is 128 pending I/O requests on the server.
//
int	pxvfs::max_permitted_ios = 128;

// Everything else initialized to zero.
int64_t		pxvfs::bytes_sent_in_second	= 0;
int64_t		pxvfs::bytes_written_in_second	= 0;
timespec_t	pxvfs::window_start		= {0L, 0};

os::mutex_t	pxvfs::io_pending_lock;
os::condvar_t	pxvfs::io_pending_cv;
int64_t		pxvfs::io_pending = 0;
os::condvar_t	pxvfs::bandwidth_cv;
os::mutex_t	pxvfs::bandwidth_lock;
os::mutex_t	pxvfs::data_rate_lock;

//
// class pxvfs_inactive_task methods
//

//
// execute - this does the actual work for the task to clean up
// inactive proxy vnodes.
//
void
pxvfs_inactive_task::execute()
{
	pxvfs	*pxvfsp	= get_pxvfs();

	pxvfsp->flags_lock.lock();

	//
	// Clean up some inactive proxy vnodes
	//
	pxvfsp->empty_inactive_list();

	if (pxvfsp->inactive_list_cnt != 0) {
		//
		// There are still more inactive proxy vnodes.
		// We do not process all inactive proxy vnodes
		// at one time in order to let other file systems clean up.
		// Requeue this work request.
		// It is safe to requeue this task for either of two reasons:
		// 1) this threadpool is single threaded.
		// 2) this work task is already off the work list,
		//	and so will not be queued twice concurrently.
		//
		pxvfs_inactive_threadpool::the().defer_processing(this);
		pxvfsp->flags_lock.unlock();

	} else {
		//
		// There are no more inactive proxy vnodes.
		//
		pxvfsp->flags &= ~pxvfs::PXFS_TASK_QUEUED;
		pxvfsp->flags_lock.unlock();
		VFS_RELE(pxvfsp->fs_vfs);
	}
}

//
// task_done - Method called when the threadpool decides to throw away a task.
// This happens only during shutdown.
// Cannot use the default implementation which does a "delete this",
// because this task is embedded in another object.
//
void
pxvfs_inactive_task::task_done()
{
	// All work tasks should have been processed before shutdown
	ASSERT(0);
}

//
// class pxvfs_inactive_threadpool methods
//

//
// constructor - this threadpool uses two threads.
//
pxvfs_inactive_threadpool::pxvfs_inactive_threadpool() :
	threadpool(false, 2, "pxvfs_inactive_threadpool", 2)
{
}

pxvfs_inactive_threadpool::~pxvfs_inactive_threadpool()
{
	ASSERT(task_count() == 0);
}

//
// startup - this method is called at modload time to initialize
// this object.
//
// static
void
pxvfs_inactive_threadpool::startup()
{
	ASSERT(the_pxvfs_inactive_threadpool == NULL);
	the_pxvfs_inactive_threadpool = new pxvfs_inactive_threadpool;

	//
	// Pxfs consume large amounts of memory.
	// Use a higher thread priority for freeing resources.
	//
	(void) the_pxvfs_inactive_threadpool->
	    set_sched_props(inactive_thread_priority);
}

//
// shutdown - this method is called at modload time to shutdown
// this object.
//
// static
void
pxvfs_inactive_threadpool::shutdown()
{
	delete the_pxvfs_inactive_threadpool;
	the_pxvfs_inactive_threadpool = NULL;
}

//
// class pxvfs methods
//

//
// Constructor.
//
//lint -e668 -e1732 -e1733
pxvfs::pxvfs(PXFS_VER::filesystem_ptr fsptr, fsmgr_client_impl *clientmgrp,
    const PXFS_VER::fs_info *fsinfop, int fstype, vfs_t *vfsp,
    uint32_t server_incarn) :
	pxvfs_list_elem(this),
	fs_rootvp(NULL),
	server_incn(server_incarn),
	flags(0),
	active_cnt(0),
	inactive_list_cnt(0),
	fsmgr_client_implp(clientmgrp),
	_syncdir_on(false),
	_nocto_on(false),
	_forcedirectio_on(false),
	underlying_fs(UNKNOWN),
	blk_reserve_invo_in_progress(false)
{
	ASSERT(clientmgrp != NULL);

#ifdef	PXFS_KSTATS_ENABLED
	char	*stats_name = new char[KSTAT_STRLEN];
	(void) sprintf(stats_name, "client v1 (%d, %d)",
	    (int)getmajor(fsinfop->fsdev), (int)getminor(fsinfop->fsdev));

	//lint +e668 +e1732 +e1733
	stats = kstat_create("pxfs", 0, stats_name,
	    "pxvfs", KSTAT_TYPE_NAMED,
	    PXVFS_STATS_MAX_NUM, KSTAT_FLAG_PERSISTENT);

	delete [] stats_name;

	if (stats != NULL) {
		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_NUM_OPEN_FILES]), "Open Files",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_NUM_OPEN_FILES].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ACCESS_TOKEN_HITS]), "Access Cache Hits",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ACCESS_TOKEN_HITS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ACCESS_TOKEN_MISSES]), "Access Cache Misses",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ACCESS_TOKEN_MISSES].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ACCESS_TOKEN_INVALS]), "Access Cache Invals",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ACCESS_TOKEN_INVALS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ATTR_TOKEN_HITS]), "Attribute Token Hits",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ATTR_TOKEN_HITS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ATTR_TOKEN_MISSES]), "Attribute Token Misses",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ATTR_TOKEN_MISSES].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_ATTR_TOKEN_INVALS]), "Attribute Token Invals",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_ATTR_TOKEN_INVALS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_DATA_TOKEN_HITS]), "Data Token Hits",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_DATA_TOKEN_HITS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_DATA_TOKEN_MISSES]), "Data Token Misses",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_DATA_TOKEN_MISSES].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_DATA_TOKEN_INVALS]), "Data Token Invals",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_DATA_TOKEN_INVALS].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_DATA_ALLOC]), "Number of bmap invocations",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_DATA_ALLOC].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_DATA_TOKEN_RETRIES]), "Data Token Retries",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_DATA_TOKEN_RETRIES].value.ui32 = 0;

		kstat_named_init(&(KSTAT_NAMED_PTR(stats)
		    [PXVFS_STATS_THROTTLING_HITS]), "Mem_async Throttling Hits",
		    KSTAT_DATA_UINT32);
		(KSTAT_NAMED_PTR(stats))[
		    PXVFS_STATS_THROTTLING_HITS].value.ui32 = 0;

		kstat_install(stats);
	}
#else
	stats = NULL;
#endif	/* PXFS_KSTATS_ENABLED */

	vfsp->vfs_data = (caddr_t)this;
	vfsp->vfs_fstype = fstype;

	//
	// The VFS documentation says only that this field is the native block
	// size of the file system. We have successfully used both
	// the PXFS transfer size and that of the underlying file system VFS.
	// We will use the block size of the underlying file system so
	// as to be able to know the granual size of an allocation when
	// filling in a hole.
	//
	vfsp->vfs_bsize = fsinfop->fsbsize;

	vfsp->vfs_dev = fsinfop->fsdev;
	vfsp->vfs_flag = fsinfop->fsflag | VFS_PXFS;

	// XXX:	should be:	fs_vfs->vfs_fsid = conv(fsinfop->fsid);
	//	but haven't been able to figure out how to write a conv method
	//	the compiler will accept.
	vfsp->vfs_fsid.val[0] = fsinfop->fsid.val[0];
	vfsp->vfs_fsid.val[1] = fsinfop->fsid.val[1];

	fs_vfs = vfsp;

	// Check if the underlying filesystem is UFS.
	if (strcmp("ufs", vfssw[fs_vfs->vfs_fstype].vsw_name) == 0) {
		//
		// XXXX Workaround for MO_TAG bug in vfs.c
		//
		mntopt_t *mop = vfs_hasopt(&vfsp->vfs_mntopts, "logging");

		if (mop && (mop->mo_flags & MO_TAG) == 0) {
			mop->mo_flags |= MO_TAG;
		}
		underlying_fs = UFS;
#ifndef VXFS_DISABLED
	} else if (strcmp("vxfs", vfssw[fs_vfs->vfs_fstype].vsw_name)
	    == 0) {
		underlying_fs = VXFS;
#endif
	} else if (strcmp("hsfs", vfssw[fs_vfs->vfs_fstype].vsw_name)
	    == 0) {
		underlying_fs = HSFS;
	}

	//
	// We're storing a reference to the server-side fs,
	// so we need to duplicate it.
	//
	ASSERT(!CORBA::is_nil(fsptr));
	fs_fsobj = PXFS_VER::filesystem::_duplicate(fsptr);

	//
	// The value of the root vp is cached, but we can't pre-fill the
	// cache here because filesystem::getroot() returns a packed
	// vnode.  When we unpack the packed vnode we will call
	// find_pxvfs() looking for this vfs, but it has not yet been
	// constructed, so it will not be on the list of pxvfs structs and
	// we'll end up recursively calling this constructor.
	//

	//
	// Initialize the status of the pxfs client status to
	// PXFS_VER::GREENZONE. It is safe to assume we are in GREENZONE.
	// Because, during the first allocating write or a file creation
	// we would call reserve_blocks(). This routine will contact the
	// server (through get_reservation()) and find out the current status
	// of the filesystem and act accordingly.
	//
	pxvfs_status = PXFS_VER::GREENZONE;

	// Create the async threadpool for this file-system
	mem_async_threadpool =
	    new threadpool(true, 5, "apageout thr", pxfs_async_threads);

	// We must succeed.
	CL_PANIC(mem_async_threadpool != NULL);
}

pxvfs::~pxvfs()
{
	CORBA::release(fs_fsobj);

	// Assert that there are no extant client side objects for this pxvfs.
	ASSERT(fs_rootvp == NULL);
	ASSERT((flags & PXFS_FILE_ACTIVATE) == 0);
	ASSERT(llm_cb_list.empty());
} //lint !e1740 pointers are neither freed nor zero'ed by destructor

//
// get_pxvfs - This method supports the pxvfs_inactive_task
// This is a virtual method, so do not try to make inline.
//
pxvfs *
pxvfs::get_pxvfs()
{
	return (this);
}

//
// Called by mount_client_impl when the initial mount or a remount
// occur.
//
void
pxvfs::set_mntoptions(const char *mntoptions)
{
	if (pxfslib::exists_mntopt(mntoptions, MNTOPT_SYNCDIR, false)) {
		_syncdir_on = true;
	} else {
		_syncdir_on = false;
	}
	if (pxfslib::exists_mntopt(mntoptions, MNTOPT_NOCTO, false)) {
		_nocto_on = true;
	} else {
		_nocto_on = false;
	}
	if (pxfslib::exists_mntopt(mntoptions, MNTOPT_FORCEDIRECTIO, false)) {
		_forcedirectio_on = true;
	} else {
		_forcedirectio_on = false;
	}
}

//
// supported_bdev_fs - returns the whether the specifed name is supported
// by PXFS as a file system.
//
bool
pxvfs::supported_bdev_fs(char *fsname)
{
	//
	// Table for specifying list of block device file systems
	// that are currently supported as wrapped file systems.
	//
	struct supp_bdev_fs {
		char    *fsname;
		bool	supported;
	};
	static struct supp_bdev_fs supp_bdev_fs[] = {
		{ "ufs",  true },
		{ "hsfs", true },
#ifndef VXFS_DISABLED
		{ "vxfs", true }
#endif
	};

	int	nbdev_fs = (int)(sizeof (supp_bdev_fs) /
	    sizeof (struct supp_bdev_fs));

	for (int i = 0; i < nbdev_fs; i++) {
		if (strcmp(fsname, supp_bdev_fs[i].fsname) == 0) {
			return (supp_bdev_fs[i].supported);
		}
	}
	return (false);
}

//
// Mount a global file system.
// At this point, the mount point is locked locally with vn_vfswlock(),
// the file system switch table is locked locally with RLOCK_VFSSW(),
// and the vfs_t is locked locally with vfs_lock() in that order.
//
// static
int
pxvfs::mount(vfs *vfsp, vnode *mvp, struct mounta *uap, cred *cr)
{
	Environment	e;
	int		error;

	ASSERT(uap->flags & MS_GLOBAL);
	ASSERT(vn_vfswlock_held(mvp));
#ifndef PSARC_2001_038 // This assertion is no longer true for S9
	ASSERT(VFSSW_LOCKED());
#endif
	// ASSERT(vfs_lock_held(vfsp));

	//
	// Disallow an overlaid mount on an extant mount point unless it's
	// explicitly requested.
	// XXX This check should be in domount() but requires a change to
	// the VFS_MOUNT() interface.
	// XXX Also, namefs doesn't do this check quite this way.
	// XXX Note that we don't check that vp->v_type == VDIR.
	// This should be checked for most file systems but not "namefs".
	//
	mutex_enter(&mvp->v_lock);
	if ((uap->flags & MS_REMOUNT) == 0 &&
	    (uap->flags & MS_OVERLAY) == 0 &&
	    (mvp->v_count != 1 || (mvp->v_flag & VROOT) != 0)) {
		mutex_exit(&mvp->v_lock);
		return (EBUSY);
	}
	mutex_exit(&mvp->v_lock);

	//
	// Verify that the mount client is already active, returning
	// failure if it isn't.
	//
	if (!mount_client_impl::is_activated()) {
		char		nodename[32];

		(void) sprintf(nodename, "Node (%u)", orb_conf::node_number());
		os::sc_syslog_msg msg(SC_SYSLOG_GLOBAL_MOUNT_TAG,
		    nodename, NULL);
		//
		// SCMSGS
		// @explanation
		// A global mount command is attempted before the node has
		// initialized the global file system name space. Typically
		// this caused by trying to perform a global mount while the
		// system is booted in single user mode.
		// @user_action
		// If the system is not at run level 2 or 3, change to run
		// level 2 or 3 using the init(1M) command. Otherwise, check
		// message logs for errors during boot.
		//
		(void) msg.log(SC_SYSLOG_WARNING, MESSAGE,
		    "pxvfs:mount(): global mounts are not enabled"
		    " (need to run \"clconfig -g\" first)\n");
		return (ENODEV);
	}

	//
	// We read in all the mount data from user space since there
	// are many places where we need to examine the data before
	// passing it to the VFS_MOUNT() of the underlying file system.
	//
	sol::mounta ma;
	CORBA::String_var options;

	if (uap->flags & MS_SYSSPACE) {
		ma.spec = os::strdup(uap->spec);
		ma.dir = os::strdup(uap->dir);
		ma.fstype = os::strdup(uap->fstype);
		ma.flags = uap->flags;
		if (uap->datalen != 0) {
			ma.data.load((uint_t)uap->datalen, (uint_t)uap->datalen,
			    (uint8_t *)uap->dataptr, false);
		}
		if (uap->optlen != 0) {
			ma.options.load((uint_t)uap->optlen,
			    (uint_t)uap->optlen, (uint8_t *)uap->optptr, false);
		}
	} else {
		// Don't support old mount formats.
		if ((uap->flags & (MS_OPTIONSTR | MS_DATA | MS_FSS)) == 0 ||
		    (uintptr_t)uap->fstype < 256) {
			return (EINVAL);
		}

		size_t len;
		char *str = new char [MAXPATHLEN];

		if (uap->spec != NULL) {
			error = copyinstr(uap->spec, str, (size_t)MAXPATHLEN,
			    &len);
			if (error != 0) {
				delete [] str;
				return (error);
			}
			ma.spec = os::strcpy(new char [len], str);
		}

		error = copyinstr(uap->dir, str, (size_t)MAXPATHLEN, &len);
		if (error != 0) {
			delete [] str;
			return (error);
		}
		ma.dir = os::strcpy(new char [len], str);

		ma.fstype = new char [FSTYPSZ];
		error = copyinstr(uap->fstype, ma.fstype, (size_t)FSTYPSZ,
		    &len);
		if (error != 0) {
			delete [] str;
			if (error == ENAMETOOLONG) {
				error = EINVAL;
			}
			return (error);
		}

		if ((uap->flags & MS_DATA) != 0 && uap->datalen != 0) {
			ma.data.length((uint_t)uap->datalen);
			error = copyin(uap->dataptr, ma.data.buffer(),
			    (size_t)uap->datalen);	//lint !e571
			if (error != 0) {
				delete [] str;
				return (error);
			}
		}

		if ((uap->flags & MS_OPTIONSTR) != 0 && uap->optlen != 0) {
			ma.options.length((uint_t)uap->optlen);
			error = copyin(uap->optptr, ma.options.buffer(),
			    (size_t)uap->optlen);	//lint !e571
			if (error != 0) {
				delete [] str;
				return (error);
			}
		}
		MOUNT_DBPRINTF(
		    MOUNT_TRACE_CLIENT,
		    MOUNT_GREEN,
		    ("\nma.spec %s \nma.dir %s \nma.flags %x \nma.fstype %s \n",
		    (char *)ma.spec, (char *)ma.dir, ma.flags,
		    (char *)ma.fstype));

		if (ma.options.length() != 0) {
			MOUNT_DBPRINTF(
			    MOUNT_TRACE_CLIENT,
			    MOUNT_GREEN,
			    (" ma.options %s\n", ma.options.buffer()));
		}

		delete [] str;
		ma.flags = uap->flags | MS_SYSSPACE;

#ifndef VXFS_DISABLED
		if (strcmp(ma.fstype, "vxfs") == 0) {
			if (error = vxfs_copyinargs(ma)) {
				return (error);
			}
		}
#endif
	}

	// Get a reference to the local mount client.
	fs::mount_client_var	clientv = mount_client_impl::get_client_ref();
	solobj::cred_var	credobj = solobj_impl::conv(cr);

	//
	// For remounts, the mount point should be the root vnode for
	// this file system.
	//
	if (ma.flags & MS_REMOUNT) {
		// The mount point should be a PXFS vnode.
		ASSERT(mvp->v_flag & VPXFS);

		//
		// Note: we are sharing the hold on the fobj and file system
		// objects that the proxy vnode/vfs hold so don't release
		// them.
		//
		PXFS_VER::filesystem_ptr	fsptr =
		    VFSTOPXFS(vfsp)->get_fsobj();

		PXFS_VER::fobj_ptr	fobjp =
		    ((pxfobj *)pxnode::VTOPX(mvp))->getfobj();

		uint32_t	vfsflags;

		FAULTPT_PXFS(FAULTNUM_PXFS_REMOUNT_C_B,
				FaultFunctions::generic);

		mount_client_impl::get_server()->remount_v1(fsptr, fobjp, ma,
		    credobj, clientv, vfsflags, options, e);

		FAULTPT_PXFS(FAULTNUM_PXFS_REMOUNT_C_A,
		    FaultFunctions::generic);

		error = pxfslib::get_err(e);

		if (error == 0) {
			//
			// The mount client on each node except this one
			// builds the mount options table and sets the node
			// mount flags. On this node, an options table is
			// built in Solaris by domount, but we do it again
			// to accommodate differences in underlying file
			// systems (ie. VxFS).  Include the remount option
			// for Solaris 10 command level mount option checks.
			//
			vfsp->vfs_flag = vfsflags;
#ifdef GLOBAL_MNTTAB_MODTIME_INTERFACE
			if (!pxfslib::exists_mntopt(options, "remount",
			    false)) {
				char *new_options;
				//lint -e668
				size_t		new_len = strlen(options) +
				    strlen(",remount") + 1;
				new_options = new char [new_len];
				(void) strcpy(new_options, options);
				(void) strcat(new_options, ",remount");
				options = os::strdup(new_options);
				delete [] new_options;
			}
			//
			// vfs_mnttab_modtimeupd is global for Solaris 10 but
			// is declared static for Solaris 8 and 9.
			//
			vfs_list_lock();
			vfs_createopttbl(&vfsp->vfs_mntopts, options);
			vfs_parsemntopts(&vfsp->vfs_mntopts,
			    (char *)options, 1);
			vfs_mnttab_modtimeupd();
			vfs_list_unlock();
			vfs_setmntopt(vfsp, MNTOPT_REMOUNT, NULL,
			    VFS_NODISPLAY);
#else

			vfs_createopttbl(&vfsp->vfs_mntopts, options);
			vfs_parsemntopts(&vfsp->vfs_mntopts, options, 1);
			gethrestime(&vfs_mnttab_mtime);
#endif
			if (ma.options.length() > 0) {
				VFSTOPXFS(vfsp)->set_mntoptions
				    ((const char *)ma.options.buffer());
			}
		}

		//
		// Workaround for MO_TAG bug in vfs.c. pxvfs's constructor.
		//
		if (strcmp("ufs", vfssw[vfsp->vfs_fstype].vsw_name) == 0) {
			mntopt_t *mop = vfs_hasopt(&vfsp->vfs_mntopts,
			    "logging");
			if (mop && (mop->mo_flags & MO_TAG) == 0) {
				mop->mo_flags |= MO_TAG;
			}
		}

		FAULTPT_PXFS(FAULTNUM_PXFS_REMOUNT_C_E,
			FaultFunctions::generic);

		return (error);
	}

	bool			dev_is_ha;
	CORBA::String_var	dev_name;
	sol::nodeid_seq_t	dev_nids;

	//
	// Check that PXFS supports this file system type.
	// Note: ma.spec can be NULL if mounting "namefs".
	// XXX This code will need to change when new file system types
	// are supported by PXFS.
	//
	if (!supported_bdev_fs(ma.fstype)) {
		//
		// Log message and bail out for unsupported file systems.
		//

		char		nodename[32];

		(void) sprintf(nodename, "Node (%u)", orb_conf::node_number());
		os::sc_syslog_msg msg(SC_SYSLOG_GLOBAL_MOUNT_TAG,
		    nodename, NULL);
		//
		// SCMSGS
		// @explanation
		// A global mount is not supported for the specified file
		// system.
		// @user_action
		// Check the release notes and documents about the support
		// of the specified system.
		//
		(void) msg.log(SC_SYSLOG_NOTICE, MESSAGE,
		    "pxvfs:mount(): global mount is not supported for "
		    "filesystem type : '%s'\n", ma.fstype);
		return (ENOTSUP);
	} else {
		//
		// Lookup the device special file to determine the dev_t
		// for the device.
		// If this is a PXFS global device file, then
		// contact DCS to start the device service and get the
		// list of nodes we should create file system replicas on.
		// For non-PXFS special files, just create a file system
		// replica on this node.
		//
		vnode_t		*bvp;
		error = lookupname(ma.spec, UIO_SYSSPACE, FOLLOW, NULL, &bvp);
		if (error != 0) {
			return (error);
		}
		if (bvp->v_type != VBLK) {
			VN_RELE(bvp);
			return (ENOTBLK);
		}
		if ((bvp->v_flag & VPXFS) == 0) {
			//
			// Not a PXFS special file.
			//
			VN_RELE(bvp);
			dev_is_ha = false;
			dev_nids.length(1);
			dev_nids[0] = orb_conf::node_number();
		} else {
			//
			// Contact DCS to get the list of nodes that this
			// device is attached to and whether or not its an
			// HA device. Also, pass a reference to the mount
			// server's dc_callback object so the mount server
			// is notified when this node configuration data
			// changes (due to system administration commands).
			//
			dev_t devid = bvp->v_rdev;
			VN_RELE(bvp);

			sol::nodeid_seq_t_var nodes;

			error = get_configured_nodes(dev_is_ha, dev_name,
			    devid, nodes, e);
			if (error != 0) {
				return (error);
			}
			dev_nids = *nodes;
		}
	}

	//
	// Note: we have to release the read lock on vfssw[] since
	// we may have to instantiate the file system (by calling
	// domount() for the underlying file system) which could
	// try to call WLOCK_VFSSW() and deadlock. We can safely
	// release the read lock because the pxfs module won't
	// be unloaded (we don't allow unloading but if we did,
	// we would have control of that in _fini()).
	// We do need to make sure we return with the read lock held
	// so that domount("pxfs") can unlock it.
	//
	// But first, allocate a vfssw[] entry for the underlying
	// file system type without loading the module.
	//
#ifdef PSARC_2001_038 // Need to grab the lock for S9
	RLOCK_VFSSW();
#endif
	struct vfssw	*vswp = vfs_getvfsswbyname(ma.fstype);
	RUNLOCK_VFSSW();
	if (vswp == NULL) {
		WLOCK_VFSSW();
		if ((vswp = vfs_getvfsswbyname(ma.fstype)) == NULL) {
			vswp = allocate_vfssw(ma.fstype);
		}
		WUNLOCK_VFSSW();
		if (vswp == NULL) {
			RLOCK_VFSSW();
			return (EINVAL);
		}
	}
	int	fstype = (int)(vswp - vfssw);
	ASSERT(fstype); // see 4470243

	//
	// We pass the mount arguments to the mount server which
	// will lock the mount point on all other nodes, create the
	// file system object (instantiate the file system),
	// create proxy file system objects on all other nodes, link
	// the proxy into the file system name space and unlock the
	// mount points. If all that goes well, we will create the proxy,
	// link it into the name space, and unlock on this node after we
	// return to domount().
	// We have to pass the mount point vnode pointer to mount()
	// since mount_client_impl::instantiate() may be called on
	// this node.
	//
	PXFS_VER::filesystem_var	fsobj;
	PXFS_VER::fs_info		fsinfo;

	FAULTPT_PXFS(FAULTNUM_PXFS_MOUNT_C_B, FaultFunctions::generic);

	mount_client_impl::get_server()->mount_v1(ma, (sol::uintptr_t)mvp,
	    credobj, clientv, dev_is_ha, dev_name, dev_nids,
	    fsobj, fsinfo, options, e);

	FAULTPT_PXFS(FAULTNUM_PXFS_MOUNT_C_A, FaultFunctions::generic);

	// Aquire the read lock before returning.
#ifndef PSARC_2001_038 // lock is no longer held, we don't grab it again for S9
	RLOCK_VFSSW();
#endif

	error = pxfslib::get_err(e);

	MOUNT_DBPRINTF(
	    MOUNT_TRACE_CLIENT,
	    (error ? MOUNT_RED : MOUNT_GREEN),
	    ("pxvfs:mount() mount error %d\n",
	    error));

	if (error != 0) {
		return (error);
	}

	//
	// The mount client on each node except this one builds the mount
	// options table. On this node an options table is built in Solaris
	// by domount, but we do it again here to accommodate differences
	// in underlying file systems (ie VxFS).
	//
	vfs_createopttbl(&vfsp->vfs_mntopts, options);
	vfs_parsemntopts(&vfsp->vfs_mntopts, (char *)options, 1);

	//
	// At this point we are committed to creating the proxy file system
	// for this node and linking it into the name space.
	//
	uint32_t	server_incarn;
	ASSERT(!CORBA::is_nil(fsobj));
	fsmgr_client_impl	*clientmgrp = new fsmgr_client_impl();
	PXFS_VER::fsmgr_client_var	clientmgr = clientmgrp->get_objref();

	uint32_t fs_blk_size;
	bool fastwrite;
	PXFS_VER::fsmgr_server_ptr	servermgr_p =
	    fsobj->bind_fs(clientmgr, orb_conf::node_number(),
	    server_incarn, fs_blk_size, fastwrite, e);
	//
	// The file system is already "mounted" on all other nodes.
	// So, go ahead and complete mounting on this node as well.
	//
	if (e.exception()) {
		//
		// We ignore comm failures since this can happen
		// if a node with the pxfs server crashes and
		// then any node joins the global name space. The
		// mount server will try to create a proxy vfs
		// for the dead file system and link it into the
		// name space. We need this to succeed so that
		// the dead file system can be unmounted properly.
		//
		if (CORBA::COMM_FAILURE::_exnarrow(e.exception()) == NULL) {
#ifdef DEBUG
			e.exception()->print_exception(
			    "pxvfs::mount(): ");
#endif
			MOUNT_DBPRINTF(
			    MOUNT_TRACE_CLIENT,
			    MOUNT_RED,
			    ("pxvfs:mount() exception from"
			    " fsobj->bind_fs()\n"));
		} else {
			servermgr_p = PXFS_VER::fsmgr_server::_nil();
			MOUNT_DBPRINTF(
			    MOUNT_TRACE_CLIENT,
			    MOUNT_RED,
			    ("pxvfs:mount() comm failure\n "));
		}

		e.clear();
	}


	pxvfs	*pxvfsp = new pxvfs(fsobj, clientmgrp, &fsinfo, fstype, vfsp,
	    server_incarn);

	pxvfsp->pxfs_bsize = fs_blk_size;
	pxvfsp->fastwrite = fastwrite;
	//
	// No blocks allocated initially.
	// The blocks get allocated during the first allocating write
	// from this node.
	//
	pxvfsp->blocks_available = 0;

	//
	// Mount flags are set by the mount client on each node except
	// this one which initiated the mount(2) call, so we set them now
	// for this node.
	//
	if (ma.options.length() > 0) {
		pxvfsp->set_mntoptions((const char *)ma.options.buffer());
	}

	// Finish initializing the linkages between fsmgr client and server.
	clientmgrp->set_pxvfsp(pxvfsp, servermgr_p);

	//
	// Add ourself to the local list of all PXFS file systems and
	// keep a hold on the vfs struct while on the list.
	//
	all_pxvfs_lock.wrlock();
	ASSERT(search(fsobj) == NULL);
	all_pxvfs.prepend((pxvfs_list_elem *)pxvfsp);
	VFS_HOLD(vfsp);
	all_pxvfs_lock.unlock();

	FAULTPT_PXFS(FAULTNUM_PXFS_MOUNT_C_E, FaultFunctions::generic);

	CORBA::release(servermgr_p);
	return (0);
}

//
// Unmount a global file system.
// At this point, the covered mount point is locked locally with vn_vfswlock(),
// and the vfs_t is locked locally with vfs_lock() in that order.
//
// Note: Purge of all dnlc entries for this vfs has been done in dounmount
// of the pxfs filesystem.
//
int
pxvfs::unmount(int umflags, cred *credp)
{
	Environment	e;

	// ASSERT(vfs_lock_held(fs_vfs));
	ASSERT(!CORBA::is_nil(fs_fsobj));

	//
	// When a cluster node is going down cleanly (ie. with shutdown,
	// poweroff, init, etc), the /sbin/umountall utility is run to
	// unmount all file systems.  But we don't want umountall to unmount
	// PXFS filesystems served out by other nodes and mounted cluster-wide.
	// We take advantage of the Solaris stop script facility associated
	// with the init program; in particluar rc0.d/K30MOUNTGFSYS is run
	// as a result of init 0 (this is done for both shutdown and
	// poweroff).  From the stop script global mounts are disabled using
	// the cladm system call (and ultimately pxvfs::disable_unmounts).
	//
	// When using the scshutdown command to take down all cluster nodes,
	// the global unmounts are of course valid and needed. In this
	// case, scshutdown performs unmounting of global filesytems prior
	// to doing the individual node shutdowns.
	//
	// For node take down commands which do not result in running init
	// (ie. halt, reboot, uadmin), filesystems are unmounted in the
	// kernel and PXFS is made aware that the unmount is occuring.
	// This is done using the PXFS_SYNC_CLOSE flag set during the sync
	// operation before unmount.  If PXFS_SYNC_CLOSE is set, PXFS can set
	// it's PXFS_SHUTDOWN flag.  This is set even if the node going down is
	// the only node serving out the filesystem.  This is desired to prevent
	// applications (which would be unaware of the unmount) from continuing
	// operations at the local mount point.  Rather, after the node serving
	// out the filesystem is down, applications will get EIO for further
	// operations and they can take appropriate action.
	//
	if (unmounts_disabled || flags & PXFS_SHUTDOWN) {
		return (EBUSY);
	}

	flags_lock.lock();
	//
	// Set PXFS_UNMOUNTING to block creation of new proxy vnodes.
	//
	flags |= PXFS_UNMOUNTING;
	if (umflags & MS_FORCE) {
		flags |= PXFS_FORCE_UNMOUNTING;
	}

	if (fs_rootvp != NULL) {
		// Release the cached root vnode.
		vnode_t		*vnodep = fs_rootvp;
		fs_rootvp = NULL;
		flags_lock.unlock();
		VN_RELE(vnodep);
	} else {
		flags_lock.unlock();
	}

	//
	// For a normal unmount, there will be a check for filesystem busy.
	// If busy, we get a return value of true; else the return value is
	// false after waiting for the inactive vnode list to become empty.
	// For forced unmount there is a wait for one pass of processing
	// of the inactive list rather than waiting for an empty list; and
	// the return value is always false.
	//
	if (wait_empty_inactive_list(umflags & MS_FORCE ? true : false)) {
		MOUNT_DBPRINTF(
		    MOUNT_TRACE_CLIENT,
		    MOUNT_RED,
		    ("pxvfs(%p):unmount returning EBUSY\n",
		    this));
		return (EBUSY);
	}

	//
	// Get object references for the mount service
	//
	solobj::cred_var	credobj = solobj_impl::conv(credp);
	fs::mount_client_var	clientv = mount_client_impl::get_client_ref();

	fs::mount_server_var	mount_server_v =
	    mount_client_impl::get_server();

	FAULTPT_PXFS(FAULTNUM_PXFS_UNMOUNT_C_B, FaultFunctions::generic);
	mount_server_v->unmount_v1(fs_fsobj, umflags, credobj, clientv,
	    orb_conf::node_number(), 0, e);
	FAULTPT_PXFS(FAULTNUM_PXFS_UNMOUNT_C_A, FaultFunctions::generic);

	sol::error_t	error = pxfslib::get_err(e);

	MOUNT_DBPRINTF(
	    MOUNT_TRACE_CLIENT,
	    (error ? MOUNT_RED : MOUNT_GREEN),
	    ("pxvfs:unmount(%p) unmount error %d\n",
	    this, error));

	if (error != 0) {
		unmount_failed();
	}
	FAULTPT_PXFS(FAULTNUM_PXFS_UNMOUNT_C_E, FaultFunctions::generic);
	return (error);
}

int
pxvfs::mountroot(enum whymountroot)
{
	MOUNT_DBPRINTF(
	    MOUNT_TRACE_CLIENT,
	    MOUNT_RED,
	    ("pxvfs:mountroot called\n")); // XXX
	return (ENOTSUP);
}

//
// root - An out parameter will contain the root directory proxy vnode,
// and the return value identifies whether the operation succeeded or failed.
//
// Note that the current design imposes the restriction that
// the underlying filesystem sets
// its root once and does not change its root.
//
int
pxvfs::root(vnode **vnodepp)
{
	flags_lock.lock();
	while (flags & PXFS_UNMOUNTING) {
		//
		// Wait for pending unmount operation to complete
		//
		flags |= PXFS_FILE_ACTIVATE;
		flags_cv.wait(&flags_lock);
	}
	if (flags & PXFS_UNMOUNTED) {
		//
		// The file system was unmounted,
		// and hence all files are inaccessible
		//
		flags_lock.unlock();
		*vnodepp = NULL;
		return (EIO);
	}
	if (fs_rootvp != NULL) {
		//
		// The root directory vnode is cached
		//
		*vnodepp = fs_rootvp;
		VN_HOLD(*vnodepp);
		flags_lock.unlock();
		return (0);
	}
	flags_lock.unlock();

	//
	// Call the pxfs server to get a root object.
	// The root directory proxy vnode remains in existence
	// while the proxy file system is mounted.
	// Because the lock is dropped, another thread could
	// concurrently ask for the root proxy directory.
	// So we have to deal with the case where we are not
	// the first to ask for the root directory.
	//
	PXFS_VER::fobj_var	rootobj;
	PXFS_VER::fobj_info	rootinfo;
	PXFS_VER::bind_info	binfo;
	Environment		e;
	uint32_t		server_incn_orig = get_server_incn();

	fobj_client_impl	*client1p = new fobj_client_impl;
	fobj_client_impl	*client2p;

	PXFS_VER::fobj_client_ptr	client1_p = client1p->get_objref();

	PXFS_VER::fobj_client_ptr	client2_p =
	    PXFS_VER::fobj_client::_nil();

	fs_fsobj->getroot(rootobj, rootinfo, binfo, client1_p, client2_p, e);
	sol::error_t	error = pxfslib::get_err(e);
	if (error != 0) {
		CORBA::release(client1_p);
		CORBA::release(client2_p);
		*vnodepp = NULL;
		return (error);
	}
	// The root directory supports caching
	ASSERT(!CORBA::is_nil(client2_p));

	if (client1_p->_equiv(client2_p)) {
		//
		// This is the first thread to request root directory
		//
		// Do not allow the server to change while
		// in the middle of creating a proxy vnode
		//
		server_incn_lock.rdlock();
		if (server_incn_orig == get_server_incn()) {
			// Construct a proxy vnode for the root directory.
			*vnodepp =
			    get_pxfobj(rootobj, rootinfo, &binfo, client1p);

			server_incn_lock.unlock();
		} else {
			//
			// The registration was orphanned by
			// either a failover or switchover
			//
			server_incn_lock.unlock();

			error = connect_again(vnodepp, rootobj, rootinfo, binfo,
			    client1p, client1_p, e);
		}

		if (*vnodepp == NULL) {
			//
			// Error because we are unmounting proxy file system
			//
			CORBA::release(client1_p);
			CORBA::release(client2_p);
			return (EIO);
		}

		flags_lock.lock();
		if (fs_rootvp == NULL) {
			//
			// Cache the root directory proxy vnode.
			// The cache places its own hold on the
			// root directory proxy vnode.
			//
			fs_rootvp = *vnodepp;
			VN_HOLD(*vnodepp);
		}
		flags_lock.unlock();

	} else {
		//
		// There already is a proxy vnode for root directory
		//
		client2p = (fobj_client_impl *)
		    (client2_p->_handler()->get_cookie());

		//
		// The current client may not be ready
		//
		if (!client2p->wait_till_ok()) {
			//
			// Error because we are unmounting proxy file system
			//
			*vnodepp = NULL;
			CORBA::release(client1_p);
			CORBA::release(client2_p);
			return (EIO);
		}
		//
		// The existing client has completed initialization
		// and has a hold placed upon the proxy vnode.
		//
		*vnodepp = pxnode::PXTOV(client2p->get_pxfobjplusp());

		flags_lock.lock();
		if (fs_rootvp == NULL) {
			//
			// Cache the root directory proxy vnode.
			// The cache places its own hold on the
			// root directory proxy vnode.
			//
			fs_rootvp = *vnodepp;
			VN_HOLD(*vnodepp);
		}
		flags_lock.unlock();
	}
	CORBA::release(client1_p);
	CORBA::release(client2_p);

	ASSERT((*vnodepp)->v_flag & VROOT);

	return (0);
}

int
pxvfs::statvfs(struct statvfs64 *sp)
{
	Environment e;

	//
	// We hold the root proxy vnode, because not all calls
	// to VFS_STATVFS hold the root vnode pointer
	// before making the VFS_STATVFS call.
	//

	vnode	*vnodep;
	int	err = root(&vnodep);
	if (err != 0) {
		return (err);
	}

	//lint -e64
	fs_fsobj->get_statistics(conv(*sp), e);
	//lint +e64

	VN_RELE(vnodep);
	return (pxfslib::get_err(e));
}

//
// sync_all_attr
// Walk the list of all pxfobj objects, and flush out dirty attributes.  This is
// triggered by fsflush calling vfs_sync with the SYNC_ATTR flag.
//
// static
void
pxvfs::sync_all_attr(void *)
{
	pxfobj		*pxfobjp;
	pxfobj		*prevp = NULL;
	int		file_count = 0;
	int		sleep_index;

	for (uint_t idx = 0; idx < pxfobjhsz; idx++) {
		//
		// Initialize the iterator while holding the lock.
		//
		pxfobj_hash[idx].hlock.lock();
		pxfobj_list_t::ListIterator	iter(pxfobj_hash[idx].hlist);

		//
		// Determine sleep time.
		// The exact value does not matter.
		// We pretend that the files are evenly distributed across
		// buckets.
		//
		switch ((pxfobj_hash[idx].hlist_cnt * pxfobjhsz) / 10000) {
		case 0:		sleep_index = 0;	// <10,000 files
				break;
		case 1:		sleep_index = 1;	// <20,000 files
				break;
		case 2:		sleep_index = 2;	// <30,000 files
				break;
		case 3:		sleep_index = 3;	// <40,000 files
				break;
		default:	sleep_index = 4;	// >=40,000 files
		};

		for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {

			ASSERT(pxfobjp->is_inhashtable());

			if (pxfobjp->is_inactive() ||
			    !pxfobjp->is_cached()) {
				//
				// Inactive processing will flush any
				// cached attributes.
				// Do not place hold on proxy vnode,
				// as that will stop inactive file cleanup.
				//
				file_count++;
				continue;
			}
			//
			// Ensure that the proxy vnode remains active
			// while doing the sync on this proxy vnode.
			// PXFS does not remove the proxy vnode from
			// this hash table while the proxy vnode remains active.
			// This ensures that the list iterator will be pointing
			// to a member of the hash table.
			//
			VN_HOLD(pxnode::PXTOV(pxfobjp));

			//
			// Drop the lock to allow changes to the hash table.
			// The sync can take a long time. So this is needed.
			//
			pxfobj_hash[idx].hlock.unlock();

			if (prevp) {
				//
				// It is possible that this operation will
				// render this proxy vnode inactive and expel
				// this proxy vnode from the hash table.
				//
				VN_RELE(pxnode::PXTOV(prevp));
			}

			(void) ((pxfobjplus *)pxfobjp)->sync_attr();
			prevp = pxfobjp;

			//
			// This flush can starve other work.
			// So we throttle the flush.
			//
			if (++file_count >= sync_all_attr_throttle) {
				//
				// Allow other work.
				//
				os::usecsleep(
				    sync_all_attr_interval[sleep_index] *
				    ((file_count +
				    (sync_all_attr_throttle - 1))
				    / sync_all_attr_throttle));
				file_count = 0;
			}

			pxfobj_hash[idx].hlock.lock();
		}
		pxfobj_hash[idx].hlock.unlock();
	}
	if (prevp) {
		//
		// Release the hold placed on the proxy vnode that was
		// processed last.
		//
		VN_RELE(pxnode::PXTOV(prevp));
	}
	sync_all_attr_lock.lock();
	sync_all_attr_thread_running = false;
	sync_all_attr_lock.unlock();
}

//
// This is a sync in order to unmount one file system.
//
// XXX We should do an invocation to the server so the sync is seen globally
// and also to sync the underlying file system.
//
int
pxvfs::sync(short flag, cred *credp)
{
	pxfobj	*pxfobjp;
	pxfobj	*prevp;

	//
	// If we are panicing we cannot safely perform any file system
	// operation. The integrity of the kernel is suspect too. To avoid
	// further problems we ignore the sync if a panic is in progress.
	//
	if (panicstr) {
		return (0);
	}

	if (flag & SYNC_ATTR) {
		//
		// Hand off SYNC_ATTR calls by the fsflush thread to a
		// separate thread.  This may take very long, so if the thread
		// is currently running (from a previous call), don't schedule
		// it again.
		//
		// PXFS could already be doing this operation. This avoids
		// duplicate work.
		//
		sync_all_attr_lock.lock();
		if (!sync_all_attr_thread_running) {
			sync_all_attr_thread_running = true;

			// Do 'new's outside locks.
			sync_all_attr_lock.unlock();
			defer_task	*taskp =
			    new work_task(pxvfs::sync_all_attr, NULL);
			sync_all_attr_lock.lock();

			common_threadpool::the().defer_processing(taskp);
		}
		sync_all_attr_lock.unlock();
		return (0);
	}

	//
	// Remember if SYNC_CLOSE is set so we know in unmount()
	// that the unmount is due to the node being shut down.
	//
	if (flag & SYNC_CLOSE) {
		flags_lock.lock();
		flags |= PXFS_SHUTDOWN;
		flags_lock.unlock();
	}

	prevp = NULL;

	for (uint_t idx = 0; idx < pxfobjhsz; idx++) {
		//
		// Make sure the iterator is initialized while holding the
		// lock.
		//
		pxfobj_hash[idx].hlock.lock();
		pxfobj_list_t::ListIterator iter(pxfobj_hash[idx].hlist);

		for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {

			ASSERT(pxfobjp->is_inhashtable());
			VN_HOLD(pxnode::PXTOV(pxfobjp));

			pxfobj_hash[idx].hlock.unlock();

			if (prevp != NULL) {
				VN_RELE(pxnode::PXTOV(prevp));
			}

			//
			// We commit only those vnodes which belong to this
			// filesystem.
			//
			if ((pxnode::PXTOV(pxfobjp)->v_vfsp) == fs_vfs) {
#if	SOL_VERSION >= __s11
				(void) VOP_PUTPAGE(pxnode::PXTOV(pxfobjp),
				    (offset_t)0, (size_t)0, B_ASYNC, credp,
				    NULL);
#else
				(void) VOP_PUTPAGE(pxnode::PXTOV(pxfobjp),
				    (offset_t)0, (size_t)0, B_ASYNC, credp);
#endif
			}

			prevp = pxfobjp;
			pxfobj_hash[idx].hlock.lock();

			//
			// Although the list could have changed while the lock
			// was not held, the list element pointer in the
			// iterator should still be valid since we held the
			// vnode (i.e., it shouldn't be removed from the list).
			// XXX This is based on the knowledge of exact
			// implementation of the ListIterator. This
			// should work as of today, but we need to be careful
			// if iterator implementation changes.
			//
		}
		pxfobj_hash[idx].hlock.unlock();
	}
	if (prevp) {
		VN_RELE(pxnode::PXTOV(prevp));
	}
	return (0);
}

//
// Sync all pxfs file systems.
//
// XXX We should do an invocation to the server so the sync is seen globally
// and also to sync the underlying file system.
//
int
pxvfs::sync_all(short flag, cred *cr)
{
	pxvfs	*pxvfsp;
	int	error = 0;
	int	result = 0;

	all_pxvfs_lock.rdlock();
	for (all_pxvfs.atfirst();
	    (pxvfsp = all_pxvfs.get_current()) != NULL;
	    all_pxvfs.advance()) {

		error = pxvfsp->sync(flag, cr);


		if (result == 0) {
			//
			// If multiple file systems encounter errors,
			// will return the first error encountered.
			//
			result = error;
		}
	}
	all_pxvfs_lock.unlock();

	return (error);
}

//
// fid_to_proxy_file - Find the proxy file object for the specified FID.
//
// The caller must eventually do VN_RELE to undo the VN_HOLD done here.
//
pxfobj *
pxvfs::fid_to_proxy_file(fid_t *wanted_fidp)
{
	pxfobj		*pxfobjp;
	const fid_t	*fidp;

	uint_t		b_idx = pxfs_misc::hash_devt_fid(get_vfsp()->vfs_dev,
			    wanted_fidp, pxfobjhsz);

	pxfobj_hash_bkt	&hbkt = pxfobj_hash[b_idx];

	// Make sure the iterator is initialized while holding the lock.
	hbkt.hlock.lock();
	pxfobj_list_t::ListIterator iter(hbkt.hlist);

	for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {
		fidp = pxfobjp->get_fidp();
		if (fidp != NULL &&
		    fidp->fid_len == wanted_fidp->fid_len &&
		    (bcmp(fidp->fid_data, wanted_fidp->fid_data,
		    (size_t)wanted_fidp->fid_len) == 0)) {
			//
			// Found a match
			//
			// Make sure the proxy file object
			// stays around until we're finished
			//
			VN_HOLD(pxnode::PXTOV(pxfobjp));
			hbkt.hlock.unlock();

			if (pxfobjp->can_cache()) {
				((pxfobjplus *)pxfobjp)->set_recycled();
			}

			return (pxfobjp);
		}
	}
	hbkt.hlock.unlock();

	// No match found
	return (NULL);
}

//
// vget - Lookup a file based on the file ID.
// Return its vnode pointer held.
//
// This operation supports NFS operations.
//
int
pxvfs::vget(vnode **vnodepp, struct fid *ufidp)
{
	pxfobj		*pxfobjp;
	const fid_t	*fidp;

	flags_lock.lock();
	while (flags & PXFS_UNMOUNTING) {
		//
		// Wait for pending unmount operation to complete
		//
		flags |= PXFS_FILE_ACTIVATE;
		flags_cv.wait(&flags_lock);
	}
	if (flags & PXFS_UNMOUNTED) {
		//
		// The file system was unmounted,
		// and hence all files are inaccessible
		//
		flags_lock.unlock();
		*vnodepp = NULL;
		return (EIO);
	}
	flags_lock.unlock();

#ifdef DEBUG
	os::atomic_add_32(&pxvfs_vget_number_calls, 1);
#endif

	uint_t		b_idx = pxfs_misc::hash_devt_fid(get_vfsp()->vfs_dev,
			    ufidp, pxfobjhsz);

	pxfobj_hash_bkt	&hbkt = pxfobj_hash[b_idx];

	// Make sure the iterator is initialized while holding the lock.
	hbkt.hlock.lock();
	pxfobj_list_t::ListIterator iter(hbkt.hlist);

	for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {
		fidp = pxfobjp->get_fidp();
		if (fidp != NULL) {
			//
			// See if the current proxy file object
			// has the fid of the file we are looking for.
			//
			if ((fidp->fid_len == ufidp->fid_len) &&
			    bcmp(fidp->fid_data, ufidp->fid_data,
			    (size_t)ufidp->fid_len) == 0) {
				//
				// Check that the file object of the
				// matched fid, belongs to this filesystem.
				//
				// The current hashing algorithm for
				// a pxfobj uses the dev_t and fid
				// to determine the hash index.
				// In order to retrieve the correct
				// pxfobj from the hash table,
				// both dev_t and fid need to match.
				//
				// Note that fids are unique only for
				// a given filesystem, not across filesystems.
				// It is very much possible that two files
				// belonging to different filesystems,
				// may have the same fid, and may end up
				// on the same hash chain.
				//
				vfs_t	*cur_vfsp =
				    pxnode::PXTOV(pxfobjp)->v_vfsp;
				if (cur_vfsp == get_vfsp()) {
					*vnodepp = pxnode::PXTOV(pxfobjp);
					VN_HOLD(*vnodepp);
					hbkt.hlock.unlock();
					if (pxfobjp->can_cache()) {
						((pxfobjplus *)pxfobjp)->
						    set_recycled();
					}
#ifdef DEBUG
					os::atomic_add_32(
					    &pxvfs_vget_number_fid_hits, 1);
#endif
					return (0);
				} else {
					PXFS_DBPRINTF(PXFS_TRACE_PXVFS,
					    PXFS_AMBER,
					    ("pxvfs:vget(%p) fid %p "
					    "false match vfsp %p\n",
					    this,  fidp->fid_data, cur_vfsp));
				}
			}
		}
	}
	hbkt.hlock.unlock();

	//
	// Go to the pxfs server to do the lookup.
	//
	PXFS_VER::fobj_var	fobj_v;
	PXFS_VER::fobj_info	fobjinfo;
	PXFS_VER::bind_info	binfo;
	Environment		e;
	uint32_t		server_incn_orig = get_server_incn();

	fs::fobjid_t		fobjid(ufidp->fid_len, ufidp->fid_len,
					(uint8_t *)ufidp->fid_data, false);
	//
	// If the new client is not used, unreference will clean it up.
	//
	fobj_client_impl	*client1p = new fobj_client_impl;
	fobj_client_impl	*client2p;

	PXFS_VER::fobj_client_ptr	client1_p = client1p->get_objref();

	PXFS_VER::fobj_client_ptr	client2_p =
	    PXFS_VER::fobj_client::_nil();

	//
	// Contact the server to get the file object for the specified FID
	//
	fs_fsobj->getfobj(fobjid, fobj_v, fobjinfo, binfo,
	    client1_p, client2_p, e);

	sol::error_t	error = pxfslib::get_err(e);
	if (error == 0) {
		if (CORBA::is_nil(client2_p) || client1_p->_equiv(client2_p)) {
			//
			// This means we are creating a new proxy vnode
			//
			if (CORBA::is_nil(client2_p)) {
				//
				// Client side caching is not used
				//
				client1p = NULL;

				*vnodepp = get_pxfobj(fobj_v, fobjinfo, &binfo,
				    client1p);
			} else {
				//
				// Client side caching is used
				//
				// Do not allow the server to change while
				// in the middle of creating a proxy vnode
				//
				server_incn_lock.rdlock();
				if (server_incn_orig == get_server_incn()) {
					*vnodepp = get_pxfobj(fobj_v, fobjinfo,
					    &binfo, client1p);
					server_incn_lock.unlock();
				} else {
					//
					// The registration was orphanned by
					// either a failover or switchover
					//
					server_incn_lock.unlock();

					error = connect_again(vnodepp,
					    fobj_v, fobjinfo, binfo,
					    client1p, client1_p, e);
				}
			}
			if (error == 0 && *vnodepp == NULL) {
				//
				// Although UFS returns a NULL vnode pointer
				// and error == 0,
				// the right thing to do is return EINVAL.
				//
				error = EINVAL;
			}
		} else {
			//
			// There already is a proxy vnode
			//
			client2p = (fobj_client_impl *)
			    (client2_p->_handler()->get_cookie());

			//
			// The current client may not be ready
			//
			if (!client2p->wait_till_ok()) {
				//
				// The current client is going away
				//
				CORBA::release(client2_p);
				client2_p = PXFS_VER::fobj_client::_nil();

				error = connect_again(vnodepp,
				    fobj_v, fobjinfo, binfo,
				    client1p, client1_p, e);
			} else {
				//

				// Existing client has completed initialization
				// and has a hold placed upon it.
				//
				*vnodepp =
				    pxnode::PXTOV(client2p->get_pxfobjplusp());
			}
		}
	} else {
		//
		// This is so that NFS server returns stale file handle error
		// to the client. Current implementation checks for error or
		// NULL vnode pointer, but we set both to make sure we
		// behave the same way as UFS does.
		//
		*vnodepp = NULL;
	}
	CORBA::release(client1_p);
	CORBA::release(client2_p);
	return (error);
}

//
// connect_again - The fobj_client originally given by the server went away.
// This method contacts the server and connects again with a new fobj_client.
//
int
pxvfs::connect_again(vnode **vnodepp,
    PXFS_VER::fobj_ptr fobj_p,
    PXFS_VER::fobj_info &fobjinfo,
    PXFS_VER::bind_info &binfo,
    fobj_client_impl *client1p,
    PXFS_VER::fobj_client_ptr client1_p,
    Environment &e)
{
	fobj_client_impl		*client2p;

	PXFS_VER::fobj_client_ptr	client2_p =
	    PXFS_VER::fobj_client::_nil();

	uint32_t		server_incn_orig;

	//
	// Keep trying until we succeed or encounter a fatal error
	//
	while (true) {
		server_incn_orig = get_server_incn();

		((PXFS_VER::fobjplus_ptr)fobj_p)->
		    cache_new_client(binfo, client1_p, client2_p, e);

		if (e.exception()) {
			//
			// The file server object could not be restored after a
			// node failure.
			//
			e.clear();
			*vnodepp = NULL;
			return (EIO);
		}

		//
		// We only reconnect for those file objects supporting caching.
		//
		ASSERT(!CORBA::is_nil(client2_p));
		if (client1_p->_equiv(client2_p)) {
			//
			// This means we are creating a new proxy vnode
			//
			// Do not allow the server to change while in the
			// middle of creating a proxy vnode.
			//
			server_incn_lock.rdlock();
			if (server_incn_orig != get_server_incn()) {
				//
				// The registration was orphanned by a
				// failover or a switchover
				//
				server_incn_lock.unlock();
				continue;
			}

			*vnodepp = get_pxfobj(fobj_p, fobjinfo, &binfo,
			    client1p);

			server_incn_lock.unlock();

			if (*vnodepp == NULL) {
				//
				// This error can happen on the root directory
				// when the proxy file system is being unmounted
				//
				return (EIO);
			} else {
				return (0);
			}
		} else {
			//
			// There already is a proxy vnode
			//
			client2p = (fobj_client_impl *)
			    (client2_p->_handler()->get_cookie());

			//
			// The current client may not be ready
			//
			if (client2p->wait_till_ok()) {
				//
				// Existing client has completed initialization
				// and has a hold placed upon it.
				//
				*vnodepp =
				    pxnode::PXTOV(client2p->get_pxfobjplusp());

				return (0);
			}

			//
			// The specified client is going away. Try again.
			//
			CORBA::release(client2_p);
			client2_p = PXFS_VER::fobj_client::_nil();
		}
	}
}

int
pxvfs::swapvp(vnode **, char *)
{
	return (ENOSYS);
}

//
// get_pxfobj - Find or create a proxy vnode for the given fobj.
// The vnode is returned held and the caller should call VN_RELE()
// when finished using the vnode.
//
// The routine guarantees that there's at most one proxy vnode
// on a given client for a given fobj.
//
// Warning: this method assumes that the fobj belongs to the file system
// it is invoked on, and will create a proxy vnode for the fobj whether or not
// the fobj is in fact owned by the fs.
//
// Any proxy vnode that supports caching will have a fobj_client,
// and must have a non-null 'binfop'. The reverse is not true.
// For example, a lookup on a device will arrive here with a non-null binfop
// and no fobj_client. The unpack_vnode() method will call this method
// and provide null values for both binfop and the fobj_client.
//
vnode *
pxvfs::get_pxfobj(PXFS_VER::fobj_ptr fobjp, const PXFS_VER::fobj_info &fobjinfo,
    PXFS_VER::bind_info *binfop, fobj_client_impl *clientp)
{
	pxfobj	*pxfobjp;

	ASSERT(!CORBA::is_nil(fobjp));

	//
	// During unmount, we need to be able to say that all vnodes are
	// inactive. Most new vnodes are created via directory operations
	// which are locked out when vn_vfswlock() is called on the mount
	// point. However, VFS_GET() (vget() above) and pxfobj::unpack_vnode()
	// are not blocked and so we have this code here to prevent new
	// pxfobj's from being created until unmount either succeeds or
	// fails completely.
	// Note: if we make it past the unlock, unmount() will return EBUSY
	//
	flags_lock.lock();
	while (flags & PXFS_UNMOUNTING) {
		flags |= PXFS_FILE_ACTIVATE;
		flags_cv.wait(&flags_lock);
	}
	if (flags & PXFS_UNMOUNTED) {
		flags_lock.unlock();
		return (NULL);
	}
	active_cnt++;
	flags_lock.unlock();

	// Allocate a new proxy vnode with a reference count of one.
	pxfobjp = make_pxfobj(fobjp, fobjinfo, clientp);

	if (clientp != NULL) {
		ASSERT(binfop != NULL);

		switch (binfop->_d()) {
		case PXFS_VER::bt_fobj: {
			//
			// Only a proxy vnode of this type caches information
			//
			pxfobjplus	*pxfobjplusp = (pxfobjplus *)pxfobjp;

			//
			// Initialize the attribute cache with
			// the data returned.
			//
			pxfobjplusp->install_attr(binfop->_u.bind_fobj.attr,
			    binfop->_u.bind_fobj.rights);

			// Initialize the cachedata flag when appropriate
			pxfobjplusp->install_cachedata_flag(
			    binfop->_u.bind_fobj.cachedata);

			//
			// Initialize the link from the proxy vnode object to
			// the fobj_client
			//
			ASSERT(clientp != NULL);
			pxfobjplusp->set_client(clientp);

			//
			// Initialize the link from the fobj_client to the
			// proxy vnode.
			//
			// This must be the last part of initialization,
			// because waiting threads are unblocked.
			//
			clientp->set_pxfobjplus(pxfobjplusp);

			break;
		}

		default:
			//
			// No information is provided to the client
			//
			ASSERT(0);
			break;
		}
	}
	// Insert the new proxy vnode into the hash list of all proxy vnodes.
	pxfobj	*pxfobj2p = fobjhash_insert(pxfobjp);
	return (pxnode::PXTOV(pxfobj2p));
}

//
// make_pxfobj
// This is called from get_pxfobj() to actually make a new proxy fobj object.
//
// Note: it should only do memory allocation and initialization for the
// proxy fobj. Other file systems can replace this default implementation
// and therefore common pxfobj initialization should go in get_pxfobj()
// instead.
//
// Proxy vnodes are created with a reference count of one.
//
pxfobj *
pxvfs::make_pxfobj(PXFS_VER::fobj_ptr fobjp,
    const PXFS_VER::fobj_info &fobjinfo,
    fobj_client_impl *clientp)
{
	pxfobj	*pxp;

	ASSERT(!CORBA::is_nil(fobjp));

	//
	// XXX - new() may lead to a deadlock if memory is exhausted
	//	 See the NFS nnode management code how deadlock
	//	 can be prevented.
	//

	switch (fobjinfo.ftype) {
	case PXFS_VER::fobj_io: {
		PXFS_VER::io_var	iop = PXFS_VER::io::_narrow(fobjp);
		ASSERT(!CORBA::is_nil(iop));
		pxp = new pxchr(fs_vfs, iop, fobjinfo);
		break;
	}
	case PXFS_VER::fobj_file: {
		PXFS_VER::file_var	filep = PXFS_VER::file::_narrow(fobjp);
		ASSERT(!CORBA::is_nil(filep));
		pxp = new pxreg(clientp, fs_vfs, filep, fobjinfo);
		break;
	}
	case PXFS_VER::fobj_unixdir: {
		PXFS_VER::unixdir_var	udp = PXFS_VER::unixdir::_narrow(fobjp);
		ASSERT(!CORBA::is_nil(udp));
		pxp = new pxdir(clientp, fs_vfs, udp, fobjinfo);
		break;
	}
	case PXFS_VER::fobj_symbolic_link: {
		PXFS_VER::symbolic_link_var	linkp =
		    PXFS_VER::symbolic_link::_narrow(fobjp);
		ASSERT(!CORBA::is_nil(linkp));
		pxp = new pxlink(clientp, fs_vfs, linkp, fobjinfo);
		break;
	}
	case PXFS_VER::fobj_special: {
		PXFS_VER::special_var	spp = PXFS_VER::special::_narrow(fobjp);
		ASSERT(!CORBA::is_nil(spp));
		pxp = new pxspecial(fs_vfs, spp, fobjinfo);
		break;
	}
	case PXFS_VER::fobj_sobj:
	case PXFS_VER::fobj_fobj:
	case PXFS_VER::fobj_device:
	case PXFS_VER::fobj_procfile:
	default:
		os::panic("pxvfs:make_pxfobj unsupported fobj type %d",
			fobjinfo.ftype);
		// NOTREACHED
	}

	return (pxp);
}

//
// fobjhash_insert
// Insert a new (held) pxfobj into the hash table of all active pxfobj's.
// If a preexisting pxfobj is found, return it (held) instead.
// The caller is responsible for calling VN_RELE() on the associated vnode;
// The caller should not use or release the pointer passed to this
// routine.
//
pxfobj *
pxvfs::fobjhash_insert(pxfobj *new_pxfobjp)
{
	PXFS_VER::fobj_ptr	fobjp = new_pxfobjp->getfobj();
	pxfobj			*pxfobjp;

	//
	// Search the hash table of all fobj objects to see if there is
	// already a proxy for 'fobjp'. We delete the one just
	// created if there is a duplicate since this will hold the
	// pxfobj hash lock for less time and has fewer locking issues
	// than trying to lock, search, create pxfobj, unlock.
	// The assumption is that deleting a duplicate does not
	// happen very often.
	//
	uint_t		b_idx = pxfs_misc::hash_devt_fid(
	    (new_pxfobjp->get_vp())->v_vfsp->vfs_dev,
	    new_pxfobjp->get_fidp(), pxfobjhsz);

	pxfobj_hash_bkt	&hbkt = pxfobj_hash[b_idx];

	hbkt.hlock.lock();
	pxfobj_list_t::ListIterator iter(hbkt.hlist);

	for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {
		if (fobjp->_equiv(pxfobjp->getfobj())) {
			//
			// We found an existing pxfobj.
			// We do a hold before releasing the hash lock so
			// that pxfobj_inactive() will see that we have
			// reclaimed the vnode.
			//
			VN_HOLD(pxnode::PXTOV(pxfobjp));
			hbkt.hlock.unlock();

			//
			// Since we didn't use the new pxfobj, we have to
			// adjust the active count. Objects in the hash
			// table will be accounted for when pxfobj_inactive()
			// removes it from the hash table.
			//
			flags_lock.lock();
			ASSERT(active_cnt >= 2);
			active_cnt--;
			flags_lock.unlock();
			VN_RELE(pxnode::PXTOV(new_pxfobjp));
			return (pxfobjp);
		}
	}

	hbkt.hlist_cnt++;

	//
	// We expect the most recently created proxy file object
	// to be the most likely one to be used next.
	// So we put the newly created one at the front of the list.
	//
	new_pxfobjp->set_inhashtable();
	hbkt.hlist.prepend(new_pxfobjp);

	hbkt.hlock.unlock();

	PXFS_KSTATS(node_stats, ((KSTAT_NAMED_PTR(node_stats))
	    [PXVFS_NODE_STATS_NUM_OPEN_FILES].value.ui32++));
	PXFS_KSTATS(stats, ((KSTAT_NAMED_PTR(stats))
	    [PXVFS_STATS_NUM_OPEN_FILES].value.ui32++));

	return (new_pxfobjp);
}

//
// add_inactivelist - queue this proxy vnode for processing
// to become stale.
//
void
pxvfs::add_inactivelist(pxfobjplus *pxfobjplusp)
{
	flags_lock.lock();

	//
	// Add ourself to the list of inactive proxy vnodes
	//
	inactive_list.append((inactive_list_elem *)pxfobjplusp);
	inactive_list_cnt++;

	if ((flags & PXFS_TASK_QUEUED) == 0) {
		//
		// Prevent pxvfs from being destroyed until work completes.
		//
		VFS_HOLD(fs_vfs);

		//
		// Since the system is not already scheduled
		// to clean up inactive proxy vnodes,
		// make the system clean up inactive proxy vnodes.
		//
		flags |= PXFS_TASK_QUEUED;
		pxvfs_inactive_threadpool::the().defer_processing(this);
	}
	flags_lock.unlock();
}

//
// empty_inactive_list - Drain the inactive proxy vnode list.
// If someone is waiting, this method processes all inactive proxy vnodes.
// Otherwise, process a limited number of inactive proxy vnodes for
// this file system, because we want to allow the worker thread to clean up
// other file systems.
//
void
pxvfs::empty_inactive_list()
{
	ASSERT(cluster_fs_drain_queue_len > 0);
	ASSERT(flags_lock.lock_held());

	pxfobj	*pxfobjp;
	int	drain_count = 0;

	//
	// If a forced unmount operation is waiting, we will drain only
	// a limited amount - but at least what is currently on the list.
	//
	if ((flags & PXFS_FORCE_UNMOUNTING) && (flags & PXFS_INACTIVE_WAIT)) {
		drain_count = - (int)inactive_list_cnt;
	}

	while ((pxfobjp = inactive_list.reapfirst()) != NULL) {
		flags_lock.unlock();
		pxfobjp->cleanup_proxy_vnode();
		flags_lock.lock();

		//
		// Another thread might think the pxvfs object has no
		// work left to clean up inactive proxy vnodes when
		// this count is zero. In which case it might destroy
		// this object. Therefore must decrement this count after
		// all the work has been done and while holding the flags_lock.
		//
		ASSERT(inactive_list_cnt != 0);
		--inactive_list_cnt;

		//
		// Only do a limited amount of processing for each file
		// system by the reaper unless someone is waiting.
		//
		if (++drain_count >= cluster_fs_drain_queue_len &&
		    (((flags & PXFS_INACTIVE_WAIT) == 0) ||
		    (flags & PXFS_FORCE_UNMOUNTING))) {
			break;
		}
	}
	//
	// If the inactive list is empty or a forced unmount is being attempted,
	// then wake up any other thread that is waiting for proxy vnodes
	// to be cleaned up.
	//
	if ((inactive_list_cnt == 0 || flags & PXFS_FORCE_UNMOUNTING) &&
	    ((flags & PXFS_INACTIVE_WAIT) != 0)) {
		//
		// Wake up any other thread that
		// is waiting for proxy vnodes to be cleaned up.
		//
		flags &= ~PXFS_INACTIVE_WAIT;
		flags_cv.broadcast();
	}
}

//
// If this function is called in support of a normal unmount, and the
// filesystem is busy (non-zero active_cnt), then return immediately.
// If this is not for an unmount or if the unmount is forced, then wait
// for processing of the inactive vnode list.
//
// The return value is only significant for normal unmount operations
// (is_unmount == true and forced_umount = false). For this case, a
// value of true is returned if the filesystem is busy. For all other
// cases, false is returned.
//
// If filesystem does not have active files, set a flag to prevent new
// vnodes from being created (see get_pxfobj()).
//
bool
pxvfs::wait_empty_inactive_list(bool forced_unmount)
{
	int	try_count;

	flags_lock.lock();
	if (!forced_unmount) {
		//
		// The only thing holding a vnode active may be something
		// transient, like an asynchronous write. For a regular unmount
		// allow a little more time for transient stuff to complete.
		//
		try_count = 6;
	} else {
		try_count = 1;
	}

	for (; try_count > 0; try_count--) {
		// Wait for inactive proxy vnode processing to finish.
		while (inactive_list_cnt != 0) {
			flags |= PXFS_INACTIVE_WAIT;
			flags_cv.wait(&flags_lock);
			if (forced_unmount) {
				break;
			}
		}
		if (active_cnt != 0 && try_count > 1) {
			flags_lock.unlock();
			// Sleep for 1 second
			os::usecsleep((os::usec_t)1000000);
			flags_lock.lock();
		}
	}
	//
	// Check to see if any proxy vnodes are still in use.
	//
	bool	in_use;
	if (forced_unmount) {
		in_use = false;
	} else {
		in_use = (active_cnt != 0);
		if (in_use) {
			flags &= ~PXFS_UNMOUNTING;
			if (flags & PXFS_FILE_ACTIVATE) {
				flags &= ~PXFS_FILE_ACTIVATE;
				flags_cv.broadcast();
			}
		}
	}
	flags_lock.unlock();
	return (in_use);
}

//
// pxfobj_inactive
// This is called when the last vnode reference to a proxy vnode
// is being released (i.e., the last call to VN_RELE() with v_count == 1).
//
// This supports proxy vnodes that never have dirty information.
// Thus there are no callbacks from the server on these proxy vnodes.
// This kind of proxy vnode can only be found through the hash table.
//
void
pxvfs::pxfobj_inactive(pxfobj *pxfobjp)
{
	//
	// Check to see if we are in the hash table.
	// If we are not, it is because we lost the race in
	// fobjhash_insert(). Once this kind of proxy vnode enters
	// the hash table, the proxy vnode remains there until destroyed.
	//
	if (pxfobjp->is_inhashtable()) {
		uint_t		b_idx = pxfs_misc::hash_devt_fid(
				    (pxfobjp->get_vp())->v_vfsp->vfs_dev,
				    pxfobjp->get_fidp(), pxfobjhsz);

		pxfobj_hash_bkt		&hbkt = pxfobj_hash[b_idx];
		hbkt.hlock.lock();

		vnode_t		*vnodep = pxfobjp->get_vp();

		//
		// Check to see if we are still inactive (not reclaimed).
		//
		mutex_enter(&vnodep->v_lock);
		if (vnodep->v_count > 1) {
			//
			// We were reclaimed by vget() or fobjhash_insert().
			// Account for the missing decrement in vn_rele().
			//
			vnodep->v_count--;
			mutex_exit(&vnodep->v_lock);
			hbkt.hlock.unlock();
			return;
		}
		mutex_exit(&vnodep->v_lock);

		//
		// We are now committed to releasing the vnode.
		//
		bool	removed = hbkt.hlist.erase(pxfobjp);
		CL_PANIC(removed);

		ASSERT(hbkt.hlist_cnt != 0);
		hbkt.hlist_cnt--;

		hbkt.hlock.unlock();
		flags_lock.lock();
		ASSERT(active_cnt != 0);
		active_cnt--;
		pxfobjp->not_inhashtable();
		flags_lock.unlock();

		PXFS_KSTATS(node_stats, ((KSTAT_NAMED_PTR(node_stats))
		    [PXVFS_NODE_STATS_NUM_OPEN_FILES].value.ui32--));
		PXFS_KSTATS(stats, ((KSTAT_NAMED_PTR(stats))
		    [PXVFS_STATS_NUM_OPEN_FILES].value.ui32--));
	}
	pxfobjp->set_stale();
	delete pxfobjp;
}

//
// pxfobjplus_inactive
// Dirty cached information has already been flushed to the server.
//
// If the proxy vnode is not in use, remove the proxy vnode from the
// hash table.
//
// Return Result True = the proxy vnode is not in the hash table
bool
pxvfs::pxfobjplus_inactive(pxfobjplus *pxfobjplusp)
{
	//
	// The proxy vnode may never have entered the hash table.
	//
	if (pxfobjplusp->is_inhashtable()) {
		//
		// The proxy vnode is in the hash table
		//
		uint_t	b_idx = pxfs_misc::hash_devt_fid(
			    (pxfobjplusp->get_vp())->v_vfsp->vfs_dev,
			    pxfobjplusp->get_fidp(), pxfobjhsz);

		pxfobj_hash_bkt		&hbkt = pxfobj_hash[b_idx];
		hbkt.hlock.lock();

		vnode_t		*vnodep = pxfobjplusp->get_vp();

		//
		// Check to see if we are still inactive (not reclaimed).
		//
		mutex_enter(&vnodep->v_lock);
		if (vnodep->v_count > 1) {
			//
			// The proxy vnode was reclaimed.
			// Account for the missing decrement in vn_rele().
			//
			vnodep->v_count--;
			mutex_exit(&vnodep->v_lock);
			hbkt.hlock.unlock();
			return (false);
		}
		mutex_exit(&vnodep->v_lock);

		//
		// We are now committed to releasing the vnode.
		//

		bool	removed = hbkt.hlist.erase((pxfobj *)pxfobjplusp);
		CL_PANIC(removed);

		ASSERT(hbkt.hlist_cnt != 0);
		hbkt.hlist_cnt--;

		hbkt.hlock.unlock();
		flags_lock.lock();
		ASSERT(active_cnt != 0);
		active_cnt--;
		flags_lock.unlock();
		pxfobjplusp->not_inhashtable();

		PXFS_KSTATS(node_stats, ((KSTAT_NAMED_PTR(node_stats))
		    [PXVFS_NODE_STATS_NUM_OPEN_FILES].value.ui32--));
		PXFS_KSTATS(stats, ((KSTAT_NAMED_PTR(stats))
		    [PXVFS_STATS_NUM_OPEN_FILES].value.ui32--));
	}
	pxfobjplusp->set_stale();
	return (true);
}

//
// purge_caches - is called to prepare a filesystem for unmount/removal.
// Return true if file system is still in use (active vnodes present),
// false otherwise. In case of a forced unmount, return false always.
//
bool
pxvfs::purge_caches(bool force_unmount, cred *credp)
{
	// Purge all DNLC entries for this vfs.
	(void) dnlc_purge_vfsp(fs_vfs, 0);

	flags_lock.lock();
	//
	// Set PXFS_UNMOUNTING to block creation of new proxy vnodes.
	//
	flags |= PXFS_UNMOUNTING;
	if (force_unmount) {
		flags |= PXFS_FORCE_UNMOUNTING;
	}

	if (fs_rootvp != NULL) {
		// Release the cached root directory vnode
		vnode_t		*vnodep = fs_rootvp;
		fs_rootvp = NULL;
		flags_lock.unlock();
		VN_RELE(vnodep);
	} else {
		flags_lock.unlock();
	}

	//
	// For normal unmount sync all the data (shortcut to VFS_SYNC(fs_vfs)).
	// Any possiblity of hanging because of I/O problems is avoided for
	// forced unmount.
	//
	if (!force_unmount) {
		(void) sync(0, credp);
	}

	//
	// Lock the vfs to maintain file system status quo during unmount.
	// This has to be done after sync(), because ufs_update tries
	// to acquire the vfs_reflock. Thus we avoid deadlock in
	// traverse(), VFS_ROOT(), get_pxfobj().
	//
	vfs_lock_wait(fs_vfs);

	//
	// Forced Unmount Case - make one pass at processing the inactive list,
	// and the return value is always false.
	//
	// Otherwise wait for the inactive list to be processed,
	// and the return value will be true if there are still active
	// proxy vnodes.
	//
	if (wait_empty_inactive_list(force_unmount)) {
		vfs_unlock(fs_vfs);

		PXFS_DBPRINTF(
		    PXFS_TRACE_PXVFS,
		    PXFS_AMBER,
		    ("pxvfs:purge_caches(%p) active_cnt %d\n",
		    this, active_cnt));

		return (true);
	} else {
		return (false);
	}
}

//
// Called to flush the filesystem's dirty data.
// All the files is the hash bucket are flushed.
// Returns non-zero if an error is encountered; zero otherwise.
//
// revoke : This is set to true if called from revoke_allocation().
//    In which case we call sync_file_revoke() instead of sync_file().
//
int
pxvfs::sync_filesystem(cred *credp, bool revoke)
{
	int	error = 0;
	int	ret = 0;
	pxfobj	*pxfobjp;
	pxfobj	*prevp = NULL;
	int	file_count = 0;

	for (uint_t idx = 0; idx < pxfobjhsz; idx++) {
		//
		// Make sure the iterator is initialized while holding the
		// lock.
		//
		pxfobj_hash[idx].hlock.lock();
		pxfobj_list_t::ListIterator iter(pxfobj_hash[idx].hlist);
		for (; (pxfobjp = iter.get_current()) != NULL; iter.advance()) {

			ASSERT(pxfobjp->is_inhashtable());

			//
			// Place a hold on the vnode so that it is not
			// released during proccessing.
			//
			VN_HOLD(pxnode::PXTOV(pxfobjp));
			pxfobj_hash[idx].hlock.unlock();

			if (prevp != NULL) {
				//
				// It is possible that this operation will
				// render this proxy vnode inactive and expel
				// this proxy vnode from the hash table.
				//
				VN_RELE(pxnode::PXTOV(prevp));
			}

			//
			// We skip those vnodes which don't belong to this
			// filesystem.
			//
			if ((pxnode::PXTOV(pxfobjp)->v_vfsp) != fs_vfs) {
				prevp = pxfobjp;
				pxfobj_hash[idx].hlock.lock();
				continue;
			}

			//
			// Synchronously write out dirty data.
			// Wait for queued up async requests to complete.
			//
			if (revoke) {
				error = pxfobjp->sync_file_revoke();
			} else {
				error = pxfobjp->sync_file();
			}

			if (error) {
				PXFS_DBPRINTF(
				    PXFS_TRACE_PXVFS,
				    PXFS_RED,
				    ("pxvfs::sync_filesystem(%p) sync_file(%p)"
				    " returned error %d \n",
				    this, pxfobjp, error));

				// Return a non zero value.
				ret = 1;
			}

			//
			// We cannot release the hold on this proxy vnode now
			// as the release can result in the vnode getting
			// expelled from the hash table. So we save a pointer
			// to the proxy vnode and do the release later.
			// Note: The release has to be done when the hlock is
			// not held.
			//
			prevp = pxfobjp;

			//
			// This sync can starve other work.
			// So we throttle the sync.
			//
			file_count++;
			if (file_count >= sync_filesystem_throttle) {
				//
				// Allow other work. Sleep for 20 ms.
				//
				os::usecsleep((os::usec_t)(20000));
				file_count = 0;
			}

			pxfobj_hash[idx].hlock.lock();
		}
		pxfobj_hash[idx].hlock.unlock();
	}

	if (prevp != NULL) {
		//
		// Release the hold placed on the proxy vnode that was
		// processed last.
		//
		VN_RELE(pxnode::PXTOV(prevp));
	}

	return (ret);
}

//
// This is called by mount_client_impl::remove_notify() when the
// global unmount succeeds.
//
void
pxvfs::unmount_succeeded()
{
	// ASSERT(vfs_lock_held(fs_vfs));
	fs_vfs->vfs_flag |= VFS_UNMOUNTED;

	//
	// Clear the PXFS_UNMOUNTING flag, set the PXFS_UNMOUNTED flag,
	// and wake up any get_pxfobj() sleepers.
	//
	flags_lock.lock();
	flags &= ~PXFS_UNMOUNTING;
	flags |= PXFS_UNMOUNTED;
	if (flags & PXFS_FILE_ACTIVATE) {
		flags &= ~PXFS_FILE_ACTIVATE;
		flags_cv.broadcast();
	}
	flags_lock.unlock();

	fsmgr_client_implp->unmount_succeeded();
	fsmgr_client_implp = NULL;

	//
	// Remove ourself from the all_pxvfs list (if we're there).
	// XXX how do we make sure that find_pxvfs() doesn't hand out a
	// pointer to us before we remove ourself from the list?
	//
	all_pxvfs_lock.wrlock();
	(void) all_pxvfs.erase((pxvfs_list_elem *)this);
	VFS_RELE(fs_vfs);	// Release list's hold on vfs_t
	all_pxvfs_lock.unlock();

#ifdef	PXFS_KSTATS_ENABLED
	if (stats != NULL) {
		kstat_delete(stats);
	}
#endif
}

//
// This is called by pxvfs::unmount() and mount_client_impl::unmount_failed()
// when a global unmount fails.
//
void
pxvfs::unmount_failed()
{
	// Clear the PXFS_UNMOUNTING flag and wake up any get_pxfobj() sleepers.
	flags_lock.lock();
	flags &= ~PXFS_UNMOUNTING;
	flags &= ~PXFS_FORCE_UNMOUNTING;
	if (flags & PXFS_FILE_ACTIVATE) {
		flags &= ~PXFS_FILE_ACTIVATE;
		flags_cv.broadcast();
	}
	flags_lock.unlock();
}

//
// This routine is called to clean up if the file system server crashes.
// It is called from fsmgr_client_impl::_unreferenced() so it shouldn't
// take too much time.
//
void
pxvfs::cleanup()
{
	// Clean up sleeping locks.
	pxfs_llm_callback_impl *llmp;
	llm_cb_list_lock.lock();
	for (llm_cb_list.atfirst(); (llmp = llm_cb_list.get_current()) != NULL;
	    llm_cb_list.advance()) {
		llmp->signal(EIO);
	}
	llm_cb_list_lock.unlock();
}

//
// new_file_system_primary - do processing needed when the system
// activates a new file system primary.
//
void
pxvfs::new_file_system_primary(uint32_t server_incarn, Environment &)
{
	//
	// The locking ensures that all in-progress client registrations
	// complete before changing the server incarnation.
	// This does not wait for invocations that have yet produced a reply.
	//
	server_incn_lock.wrlock();
	server_incn = server_incarn;
	server_incn_lock.unlock();

	replay_sleeping_locks();
}

//
// Replay all the sleeping locks that originated from this node.
//
void
pxvfs::replay_sleeping_locks()
{
	//
	// Walk through the list of callback objects, and wake them up with
	// 'RETRY_LOCK'.
	//
	pxfs_llm_callback_impl *llmp;
	llm_cb_list_lock.lock();
	for (llm_cb_list.atfirst(); (llmp = llm_cb_list.get_current()) != NULL;
	    llm_cb_list.advance()) {
		llmp->signal(pxfs_llm_callback_impl::RETRY_LOCK);
	}
	llm_cb_list_lock.unlock();
}

//
// Insert callback object into list of callback objects.
//
void
pxvfs::insert_llm_cbobj(pxfs_llm_callback_impl *llmp)
{
	llm_cb_list_lock.lock();
	llm_cb_list.prepend(llmp);
	llm_cb_list_lock.unlock();
}

//
// Remove callback object from list of callback objects.
//
void
pxvfs::remove_llm_cbobj(pxfs_llm_callback_impl *llmp)
{
	llm_cb_list_lock.lock();
	(void) llm_cb_list.erase(llmp);
	llm_cb_list_lock.unlock();
}

//
// Calls made via pxfs/server/nlm_pxfs.cc when lockd on this node dies/restarts.
// When lockd dies, there are two calls, setting the status of the NLM locks
// from this node to 'FLK_NLM_SHUTTING_DOWN', and then 'FLK_NLM_DOWN'.  The
// first call interrupts all sleeping locks, and the second call discards all
// active locks.
//
// static
void
pxvfs::set_nlm_status(int32_t nlmid, PXFS_VER::nlm_status status)
{
	pxvfs		*pxvfsp;
	SList<pxvfs>	tmp_all_pxvfs;

	//
	// Make a copy of 'all_pxvfs' so we don't have to lock/unlock
	// 'all_pxvfs'.
	//
	all_pxvfs_lock.rdlock();
	for (all_pxvfs.atfirst();
	    (pxvfsp = all_pxvfs.get_current()) != NULL;
	    all_pxvfs.advance()) {
		VFS_HOLD(pxvfsp->fs_vfs);
		tmp_all_pxvfs.prepend(pxvfsp);
	}
	all_pxvfs_lock.unlock();

	// Call 'set_nlm_status' on each filesystem.
	Environment	e;
	while ((pxvfsp = tmp_all_pxvfs.reapfirst()) != NULL) {
		pxvfsp->get_fsobj()->set_nlm_status(nlmid, status, e);
		e.clear();
		VFS_RELE(pxvfsp->fs_vfs);
	}
}

//
// Call made via pxfs/server/nlm_pxvfs.cc when statd on the client or server
// node dies and gets restarted.
//
// static
void
pxvfs::remove_file_locks(int32_t sysid)
{
	pxvfs		*pxvfsp;
	SList<pxvfs>	tmp_all_pxvfs;

	//
	// Make a copy of 'all_pxvfs' so we don't have to lock/unlock
	// 'all_pxvfs'.
	//
	all_pxvfs_lock.rdlock();
	for (all_pxvfs.atfirst();
	    (pxvfsp = all_pxvfs.get_current()) != NULL;
	    all_pxvfs.advance()) {
		VFS_HOLD(pxvfsp->fs_vfs);
		tmp_all_pxvfs.prepend(pxvfsp);
	}
	all_pxvfs_lock.unlock();

	// Call 'remove_file_locks' on each filesystem.
	Environment e;
	while ((pxvfsp = tmp_all_pxvfs.reapfirst()) != NULL) {
		pxvfsp->get_fsobj()->remove_file_locks(sysid, e);
		e.clear();
		VFS_RELE(pxvfsp->fs_vfs);
	}
}

//
// find_pxvfs
// Return the pxvfs structure for a given PXFS file system object.
// Return NULL if the proxy could not be found or created.
// Otherwise, the pointer is returned held() and the caller should
// call rele() when finished using the pointer.
//
// If 'fsinfop' is not NULL and 'fsobj' does not already
// have a proxy, then use the file system info to create a new proxy.
//
pxvfs *
pxvfs::find_pxvfs(PXFS_VER::filesystem_ptr fsobj,
    const PXFS_VER::fs_info *fsinfop)
{
	pxvfs		*pxvfsp;
	vfs_t		*vfsp;
	Environment	e;

	//
	// The vast majority of calls will only need the read lock,
	// because the proxy file system vfs already exists.
	//
	all_pxvfs_lock.rdlock();

	// Search the list of proxy file systems
	pxvfsp = search(fsobj);
	if (pxvfsp != NULL) {
		//
		// This proxy file system vfs exists
		//
		ASSERT((pxvfsp->flags & PXFS_UNMOUNTED) == 0);
		VFS_HOLD(pxvfsp->fs_vfs);
		all_pxvfs_lock.unlock();
		return (pxvfsp);
	}

	//
	// If we aren't supposed to create a new proxy,
	// return NULL since we didn't find it.
	//
	if (fsinfop == NULL) {
		all_pxvfs_lock.unlock();
		return (NULL);
	}

	//
	// Try to upgrade the read lock to a write lock. This can fail.
	//
	if (!all_pxvfs_lock.try_upgrade()) {
		// Lock upgrade attempt failed
		all_pxvfs_lock.unlock();
		all_pxvfs_lock.wrlock();

		//
		// Normally nobody else will be attempting to
		// create this file system. But this code takes
		// the safe approach that will always work
		// in spite of orphan requests or anything else,
		// and the cost is minimal.
		//
		// Search the list of proxy file systems
		pxvfsp = search(fsobj);
		if (pxvfsp != NULL) {
			//
			// This proxy file system vfs exists
			//
			ASSERT((pxvfsp->flags & PXFS_UNMOUNTED) == 0);
			VFS_HOLD(pxvfsp->fs_vfs);
			all_pxvfs_lock.unlock();
			return (pxvfsp);
		}
	}

	//
	// Set up the fsmgr_client/fsmgr_server connection.
	//
	fsmgr_client_impl		*clientmgrp = new fsmgr_client_impl();
	PXFS_VER::fsmgr_client_var	clientmgr = clientmgrp->get_objref();

	FAULTPT_PXFS(FAULTNUM_PXFS_MOUNT_BIND_FS_B,
		FaultFunctions::generic);

	uint32_t		server_incarn;
	uint32_t		fs_blk_size;
	bool			fastwrite_flag;
	PXFS_VER::fsmgr_server_ptr	servermgr_p = fsobj->
	    bind_fs(clientmgr, orb_conf::node_number(), server_incarn,
	    fs_blk_size, fastwrite_flag, e);
	if (e.exception()) {
		//
		// We ignore comm failures since this can happen
		// if a node with the pxfs server crashes and
		// then any node joins the global name space. The
		// mount server will try to create a proxy vfs
		// for the dead file system and link it into the
		// name space. We need this to succeed so that
		// the dead file system can be unmounted properly.
		//
		if (CORBA::COMM_FAILURE::_exnarrow(e.exception()) == NULL) {
#ifdef DEBUG
			e.exception()->print_exception(
			    "pxvfs::findpxvfs: ");
#endif
			MOUNT_DBPRINTF(
			    MOUNT_TRACE_CLIENT,
			    MOUNT_RED,
			    ("pxvfs:findpxvfs(): exception from"
			    " fsobj->bind_fs()\n"));
		} else {
			servermgr_p = PXFS_VER::fsmgr_server::_nil();
			MOUNT_DBPRINTF(
			    MOUNT_TRACE_CLIENT,
			    MOUNT_RED,
			    ("pxvfs:findpxvfs(): comm failure\n "));
		}

		e.clear();
	}

	//
	// Allocate a vfssw[] entry for the underlying file system type
	// but don't try to load the module.
	//
	RLOCK_VFSSW();
	struct vfssw	*vswp = vfs_getvfsswbyname((char *)fsinfop->fstype);
	if (vswp == NULL) {
		RUNLOCK_VFSSW();
		WLOCK_VFSSW();
		vswp = vfs_getvfsswbyname((char *)fsinfop->fstype);
		if (vswp == NULL)
			vswp = allocate_vfssw((char *)fsinfop->fstype);
		WUNLOCK_VFSSW();
		if (vswp == NULL) {
			all_pxvfs_lock.unlock();
			CORBA::release(servermgr_p);
			return (NULL);
		}
		RLOCK_VFSSW();
	}
	int	fstype = (int)(vswp - vfssw);
	RUNLOCK_VFSSW();

	//
	// Create a new vfs.
	//
#if	SOL_VERSION >= __s11
	vfsp = vfs_alloc(KM_SLEEP);
#else
	vfsp = (vfs_t *)kmem_alloc(sizeof (vfs_t), KM_SLEEP);
#endif
	VFS_INIT(vfsp, pxfs_vfsopsp, (caddr_t)NULL);
	VFS_HOLD(vfsp);
	pxvfsp = new pxvfs(fsobj, clientmgrp, fsinfop, fstype, vfsp,
	    server_incarn);

	pxvfsp->pxfs_bsize = fs_blk_size;
	pxvfsp->blocks_available = 0;
	pxvfsp->fastwrite = fastwrite_flag;

	// Finish initializing the linkages between fsmgr client and server.
	clientmgrp->set_pxvfsp(pxvfsp, servermgr_p);

	// Add ourself to the list of all PXFS file systems.
	all_pxvfs.prepend((pxvfs_list_elem *)pxvfsp);
	VFS_HOLD(vfsp);
	all_pxvfs_lock.unlock();

	CORBA::release(servermgr_p);
	return (pxvfsp);
}

//
// Search the all_pxvfs list for the given file system object.
// Return NULL if not found.
//
pxvfs *
pxvfs::search(PXFS_VER::filesystem_ptr fsobj)
{
	pxvfs	*p;
	pxvfs_list_t::ListIterator iter(all_pxvfs);

	for (; (p = iter.get_current()) != NULL;
	    iter.advance()) {
		if (fsobj->_equiv(p->get_fsobj())) {
			return (p);
		}
	}
	return (NULL);
}

//
// get_configured_nodes - obtains the nodes configured for this device.
// An important side effect is that DCS will start the device service
// if it is not already started.
//
int
pxvfs::get_configured_nodes(bool &dev_is_ha, CORBA::String_out dev_name,
	dev_t devid, sol::nodeid_seq_t_out nodes, Environment &e)
{
	int	error;

#ifndef PSARC_2001_038 // no longer called with the lock held in Solaris 9
	RUNLOCK_VFSSW();
#endif

	fs::dc_callback_var	callback =
	    mount_client_impl::get_server()->get_dc_callback(e);

#ifndef PSARC_2001_038
	RLOCK_VFSSW();
#endif

	if (e.exception()) {
		return (pxfslib::get_err(e));
	}

	error = dcs_get_configured_nodes(devid, callback,
	    dev_is_ha, dev_name, nodes);
	//lint -e1746

	return (error);
}

void
pxvfs::disable_unmounts()
{
	unmounts_disabled = true;
}

//
// memory_callback - the Memory Monitor executes this method whenever
// the memory state changes. This method purges all pxfs entries from
// the DNLC cache when the system is memory starved
//
// static
void
pxvfs::memory_callback(monitor::system_state_t state)
{
	switch (state) {
	case monitor::MEMORY_STARVED:
		//
		// PXFS enters its proxy vnode into the DNLC.
		// When PXFS is on top of UFS on the same node,
		// the DNLC on this node also contains UFS vnodes.
		// The UFS vnode entry in the DNLC is not removed
		// when the corresponding PXFS server file object goes away.
		// We need to purge both the client proxy vnode
		// and the server file system vnode.
		//
		// The DNLC is an optional performance enhancer.
		// Any file in active use will soon be entered again
		// into the DNLC.
		//
		// The DNLC method for removing a specific vnode
		// requires walking all of the DNLC entries.
		// So purge everything.
		//
		dnlc_purge();

		PXFS_DBPRINTF(
		    PXFS_TRACE_PXVFS,
		    PXFS_AMBER,
		    ("pxvfs:memory_callback: state %d purged dnlc\n",
		    state));
		break;
	default:
		break;
	}
}

//
// update_throughput() does bandwidth calculation. This method is called by
// every thread that does a successful page_out with the number of bytes
// transferred. In the case of async writes, it will be the aio callback
// that calls this method. After every re-calculation it signals all
// threads waiting for more bandwidth.
//
void
pxvfs::update_throughput(int bytes_xfrd)
{
	timespec_t	now = {0L, 0};
	uint64_t	current_rate;
	uint64_t	time_for_xfr;

	gethrestime(&now);

	//
	// If the current window started more than a second ago, store
	// current time, calculate new bytes per second rate and set bytes
	// available over next second to new data rate.
	//
	data_rate_lock.lock();

	// increment total bytes transferred from window_start
	bytes_sent_in_second += bytes_xfrd;

	//
	// The last througput update as less than a second ago, don't
	// recalculate data rate.
	//
	if (diff_timespec(window_start, now) < 1000) {
		data_rate_lock.unlock();
		return;
	}

	gethrestime(&window_start);

	//
	// Update recommended data rate
	//
	monitor::system_state_t sys_state = monitor::the().get_current_state();
	if (sys_state != monitor::MEMORY_PLENTY) {
		//
		// If memory is low we set the data rate to exactly the number
		// of bytes we committed in the last second.
		//
		data_rate = bytes_sent_in_second;

		//
		// If less than configured minimum bytes were written and
		// memory state is not MEMORY_STARVED set data_rate to the
		// configured minimum data rate.
		//
		if (bytes_sent_in_second < data_rate_minimum &&
		    sys_state != monitor::MEMORY_STARVED) {
			data_rate = data_rate_minimum;
		}
	} else {
		//
		// If writers aren't using the available bandwidth, we
		// don't change data rate. Re-calculation happens only if
		// qouta in last second was exceeded. wait_for_bandwidth()
		// allows this over-run when there is plenty of memory.
		//
		if (bytes_written_in_second >= data_rate) {
			if (bytes_sent_in_second < data_rate_minimum) {
				//
				// Bytes written could be low because there
				// weren't enough writes in the last few
				// seconds. Ease off the throttling as we have
				// enough memory to accomodate dirty pages.
				// Set data rate to the configured minimum
				// instead of bytes_sent_in_second.
				//
				data_rate = data_rate_minimum;
			} else {
				//
				// Anticipate an increase by 6.25% over the
				// next second if current data-rate is more
				// than the configured default.
				//
				if (bytes_sent_in_second > data_rate_default) {
					data_rate = bytes_sent_in_second +
						    (bytes_sent_in_second/16);
				} else {
					//
					// If current data rate is below
					// configured default, we must get to
					// the default fast. We increase the
					// data rate at 50% every second.
					//
					data_rate = bytes_sent_in_second +
						    (bytes_sent_in_second/2);
				}
			}
		}
	}

	// Reset per second accumulators..
	bytes_written_in_second = 0;
	bytes_sent_in_second = 0;

	// ..and set byte quota for the current window.
	bytes_in_window = data_rate;

	//
	// If there are no bytes available for writers to consume don't
	// wake anyone.
	//
	if (bytes_in_window != 0) {
		bandwidth_lock.lock();
		// Wakeup any thread waiting for bandwidth.
		bandwidth_cv.broadcast();
		bandwidth_lock.unlock();
	}

	data_rate_lock.unlock();
}

//
// 'throttle_monitor_thread' runs every second, updates the bandwidth and
// wakes up writers waiting for bandwidth. Without a per-second trigger,
// writers can wait indefinitely if there was no i/o scheduled.
//
void
pxvfs::throttle_monitor_thread(void *)
{
	while (true) {
		os::usecsleep(throttle_monitor_interval);
		//
		// Determine new bandwidth for this second with no bytes
		// transferred.
		//
		pxvfs::update_throughput(0);
	}
}

// Create a new thread to monitor bandwidth per second.
int
pxvfs::launch_throttle_monitor_thread()
{
	//
	// Create a kernel thread in the SYS scheduling class.
	//
	if ((clnewlwp(throttle_monitor_thread,
			NULL, MINCLSYSPRI, NULL, NULL)) != 0) {
		return (-1);
	}
	return (0);
}

//
// Called at modload time.
//
int
pxvfs::startup()
{
	//
	// The size of the pxfobj_hash table is computed in a way that
	// is similar to the way ufs calculates the size/max size of its
	// in-core inode hash table.
	//
	//lint -e64 -e419 -e712 -e747 -e534
	if (pxfobjhsz_max == 0) {
		pxfobjhsz_max =
		    (uint_t)1 << os::highbit((uint_t)ncsize / pxfobjh_len);
	}

	if (pxfobjhsz == 0) {
		pxfobjhsz = pxfobjhsz_max;
	}

	if (pxfobjhsz > pxfobjhsz_max) {
		pxfobjhsz = pxfobjhsz_max;
	}

	pxfobj_hash = new pxfobj_hash_bkt[pxfobjhsz];

	//
	// Create the pxvfs_inactive_threadpool,
	// which processes requests to reap inactive proxy vnodes
	//
	pxvfs_inactive_threadpool::startup();

#ifdef	PXFS_KSTATS_ENABLED
	// Create the per-node kstat structure.
	node_stats = kstat_create("pxfs", 0,
	    "Per-node client v1 stats", "pxvfs",
	    KSTAT_TYPE_NAMED, PXVFS_NODE_STATS_MAX_NUM, KSTAT_FLAG_PERSISTENT);

	if (node_stats != NULL) {
		kstat_named_init(&(KSTAT_NAMED_PTR(node_stats)
		    [PXVFS_NODE_STATS_NUM_OPEN_FILES]), "Open Files",
		    KSTAT_DATA_UINT32);
		PXFS_KSTATS(node_stats, ((KSTAT_NAMED_PTR(node_stats))
		    [PXVFS_NODE_STATS_NUM_OPEN_FILES].value.ui32 = 0));

		kstat_install(node_stats);
	}
#else
	node_stats = NULL;
#endif	/* PXFS_KSTATS_ENABLED */

	// Register pxfs purge method with Memory Monitor
	monitor::the().subscribe(memory_callback);

	int error;

	// Start the throttle monitoring thread.
	if ((error = launch_throttle_monitor_thread()) != 0) {
		char		nodename[32];

		(void) sprintf(nodename, "Node (%u)", orb_conf::node_number());
		os::sc_syslog_msg msg(SC_SYSLOG_FILESYSTEM_TAG, nodename, NULL);
		//
		// SCMSGS
		// @explanation
		// Thread to support pxfs throttling could not be launched.
		// @user_action
		// Check if the node is short on resources.
		//
		(void) msg.log(SC_SYSLOG_WARNING, MESSAGE,
		    "pxvfs:startup() Failed to create throttle monitoring"
		    " thread.\n");
		return (error);
	}

	return (0);
}

int
pxvfs::shutdown()
{
	// Deregister pxfs purge method with Memory Monitor
	monitor::the().unsubscribe(memory_callback);

	pxvfs_inactive_threadpool::shutdown();

	delete [] pxfobj_hash;
	return (0);
}

bool
pxvfs::is_unmounted()
{
	if (flags & PXFS_FORCE_UNMOUNTING) {
		return (true);
	} else {
		return (false);
	}
}

//
// This routine tries to allocate blocks from the local cache
// ie. blocks_available.
// If there are not enough blocks to satisfy the current request, it contacts
// the server (get_reservatio()) to get more blocks.
// If the server decides to switch to REDZONE we block here waiting for
// the switch to complete and we return a '0' to indicate the switch.
// If the get_reservation() finds more blocks we do the reservation
// and returns the number of blocks allocated.
//
uint64_t
pxvfs::reserve_blocks(PXFS_VER::blkcnt_t want, bool no_redzone_wait)
{
	PXFS_VER::blkcnt_t refill_block_count = 0;
	PXFS_VER::server_status_t status;
	Environment env;

	for (;;) {
		blocks_reservation.lock();

		//
		// If there is an invocation or redzone switch in progress
		// wait for the invocation or switch to complete.
		//
		while (pxvfs_status == PXFS_VER::SWITCH_TO_REDZONE ||
		    blk_reserve_invo_in_progress) {
			//
			// This was a call for pre-reservation, must
			// not wait for revoke allocations to complete.
			//
			if (no_redzone_wait) {
				blocks_reservation.unlock();
				return (0);
			}
			blocks_reservation_cv.wait(&blocks_reservation);
		}
		if (pxvfs_status == PXFS_VER::REDZONE) {
			blocks_reservation.unlock();
			return (0);
		}
		ASSERT(pxvfs_status == PXFS_VER::GREENZONE);

		if (want < blocks_available) {
			blocks_available = blocks_available - want;
			blocks_reservation.unlock();
			return (want);
		}

		//
		// Clear available blocks and set flag to show block
		// reservation invocation is in progress.
		//
		blocks_available = 0;
		blk_reserve_invo_in_progress = true;

		PXFS_DBPRINTF(PXFS_TRACE_PXVFS,
		    PXFS_AMBER,
		    ("pxvfs:reserve_blocks(%p) Calling get_reservation()\n",
		    this));

		//
		// Get more reservation from server to fulfill the reuqest.
		//
		// Do not hold locks across invocations unless necessary. We
		// must not hold blocks_reservation lock while waiting for
		// server. Any thread needing disk blocks will block until the
		// invocation completes and this thread broadcasts a wakeup.
		//
		blocks_reservation.unlock();
		get_fsobj()->get_reservation(refill_block_count, status, env);
		blocks_reservation.lock();

		env.clear();

		ASSERT(refill_block_count >= 0);

		//
		// Set local block reservation to what the server gave
		// us and clear invocation active flag.
		//
		blocks_available = refill_block_count;
		blk_reserve_invo_in_progress = false;

		if (refill_block_count > 0) {
			//
			// Wake up any threads that came in while we were
			// waiting for server to allocate us blocks.
			//
			blocks_reservation_cv.broadcast();
			blocks_reservation.unlock();
		} else {
			//
			// Server is not able to give away reservation which
			// means server either has switched to REDZONE or
			// is in the process of switching to REDZONE.
			//
			pxvfs_status = status;
			blocks_reservation.unlock();
			PXFS_DBPRINTF(PXFS_TRACE_PXVFS,
			    PXFS_AMBER,
			    ("pxvfs:reserve_blocks(%p) pxvfs_status %d\n",
			    this, status));
		}
	}
}

void
pxvfs::set_server_status(PXFS_VER::server_status_t status)
{
	blocks_reservation.lock();
	pxvfs_status = status;
	blocks_reservation_cv.broadcast();
	blocks_reservation.unlock();
}

//
// Check if there is enough bandwidth to accomodate given bytes. If
// not, wait for throttle update routine to signal us whenever per
// second quota is updated.
//
int
pxvfs::wait_for_bandwidth(int bytes_needed, int &bytes_allocated)
{
	monitor::system_state_t sys_state;

	sys_state = monitor::the().get_current_state();

	// Round up to minimum allowed bandwidth allocation.
	bytes_needed = MAX(bytes_needed, bandwidth_chunk);

	bandwidth_lock.lock();

	while (bytes_needed > bytes_in_window) {

		//
		// If there is plenty of memory, we allow over-run of the
		// calculated bandwidth once without waiting. The next
		// throughput updation will correct this if the server was
		// loaded.
		//
		if (sys_state == monitor::MEMORY_PLENTY &&
		    bytes_in_window != 0) {
			break;
		}

		//
		// Wait for more bandwidth to be available
		//
		if (!bandwidth_cv.wait_sig(&bandwidth_lock)) {
			bandwidth_lock.unlock();
			return (EINTR);
		}
		//
		// If this write is bigger than current bandwidth no sense
		// in waiting for more. If we are under memory pressure
		// don't proceed until the pressure eases off.
		//
		if (bytes_needed > data_rate &&
		    monitor::the().get_current_state() !=
					monitor::MEMORY_STARVED) {
			break;
		}
	}

	if (bytes_needed > bytes_in_window) {
		bytes_in_window = 0;
	} else {
		bytes_in_window -= bytes_needed;
	}

	bytes_written_in_second += bytes_needed;
	bandwidth_lock.unlock();

	bytes_allocated = bytes_needed;
	return (0);
}