--- old/usr/src/uts/common/avs/ns/rdc/rdc_clnt.c	Fri Jul 27 15:35:57 2018
+++ /dev/null	Fri Jul 27 15:07:21 2018
@@ -1,3381 +0,0 @@
-/*
- * CDDL HEADER START
- *
- * The contents of this file are subject to the terms of the
- * Common Development and Distribution License (the "License").
- * You may not use this file except in compliance with the License.
- *
- * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
- * or http://www.opensolaris.org/os/licensing.
- * See the License for the specific language governing permissions
- * and limitations under the License.
- *
- * When distributing Covered Code, include this CDDL HEADER in each
- * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
- * If applicable, add the following below this CDDL HEADER, with the
- * fields enclosed by brackets "[]" replaced with your own identifying
- * information: Portions Copyright [yyyy] [name of copyright owner]
- *
- * CDDL HEADER END
- */
-/*
- * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
- * Use is subject to license terms.
- */
-
-/* Network data replicator Client side */
-
-
-#include <sys/types.h>
-#include <sys/debug.h>
-#include <sys/ksynch.h>
-#include <sys/cmn_err.h>
-#include <sys/kmem.h>
-#include <sys/cred.h>
-#include <sys/byteorder.h>
-#include <sys/errno.h>
-
-#ifdef _SunOS_2_6
-/*
- * on 2.6 both dki_lock.h and rpc/types.h define bool_t so we
- * define enum_t here as it is all we need from rpc/types.h
- * anyway and make it look like we included it. Yuck.
- */
-#define	_RPC_TYPES_H
-typedef int enum_t;
-#else
-#ifndef DS_DDICT
-#include <rpc/types.h>
-#endif
-#endif /* _SunOS_2_6 */
-
-#ifndef DS_DDICT
-#include <rpc/auth.h>
-#include <rpc/svc.h>
-#include <rpc/xdr.h>
-#endif
-#include <sys/ddi.h>
-
-#include <sys/nsc_thread.h>
-#ifdef DS_DDICT
-#include <sys/nsctl/contract.h>
-#endif
-#include <sys/nsctl/nsctl.h>
-
-#include <sys/sdt.h>		/* dtrace is S10 or later */
-
-#include "rdc_io.h"
-#include "rdc_clnt.h"
-#include "rdc_bitmap.h"
-#include "rdc_diskq.h"
-
-
-kmutex_t rdc_clnt_lock;
-
-#ifdef DEBUG
-int noflush = 0;
-#endif
-
-int rdc_rpc_tmout = RDC_CLNT_TMOUT;
-static void rdc_clnt_free(struct chtab *, CLIENT *);
-static void _rdc_remote_flush(rdc_aio_t *);
-
-void rdc_flush_memq(int index);
-void rdc_flush_diskq(int index);
-int rdc_drain_net_queue(int index);
-void rdc_flusher_thread(int index);
-int  rdc_diskq_enqueue(rdc_k_info_t *krdc, rdc_aio_t *);
-void rdc_init_diskq_header(rdc_group_t *grp, dqheader *hd);
-void rdc_dump_iohdrs(disk_queue *dq);
-rdc_aio_t *rdc_dequeue(rdc_k_info_t *krdc, int *rc);
-void rdc_clr_iohdr(rdc_k_info_t *krdc, nsc_off_t qpos);
-void rdc_close_diskq(rdc_group_t *krdc);
-
-int rdc_writer(int index);
-
-static struct chtab *rdc_chtable = NULL;
-static int rdc_clnt_toomany;
-#ifdef DEBUG
-static int rdc_ooreply;
-#endif
-
-extern void rdc_fail_diskq(rdc_k_info_t *krdc, int wait, int flag);
-extern int _rdc_rsrv_diskq(rdc_group_t *group);
-extern void _rdc_rlse_diskq(rdc_group_t *group);
-
-static enum clnt_stat
-cl_call_sig(struct __client *rh, rpcproc_t proc,
-	    xdrproc_t xargs, caddr_t argsp, xdrproc_t xres,
-	    caddr_t resp, struct timeval secs)
-{
-	enum clnt_stat stat;
-	k_sigset_t smask;
-	sigintr(&smask, 0);
-	rh->cl_nosignal = TRUE;
-	stat = ((*(rh)->cl_ops->cl_call)\
-	    (rh, proc, xargs, argsp, xres, resp, secs));
-	rh->cl_nosignal = FALSE;
-	sigunintr(&smask);
-	return (stat);
-}
-
-int
-rdc_net_getsize(int index, uint64_t *sizeptr)
-{
-	struct timeval t;
-	int err, size;
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	int remote_index = krdc->remote_index;
-
-	*sizeptr = 0;
-	if (krdc->remote_index < 0)
-		return (EINVAL);
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-#ifdef DEBUG
-	if (krdc->intf == NULL)
-		cmn_err(CE_WARN,
-		    "!rdc_net_getsize: null intf for index %d", index);
-#endif
-	if (krdc->rpc_version <= RDC_VERSION5) {
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSIZE,
-		    krdc->rpc_version, xdr_int, (char *)&remote_index,
-		    xdr_int, (char *)&size, &t);
-		if (err == 0)
-			*sizeptr = size;
-	} else {
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSIZE6,
-		    krdc->rpc_version, xdr_int, (char *)&remote_index,
-		    xdr_u_longlong_t, (char *)sizeptr, &t);
-	}
-	return (err);
-}
-
-
-int
-rdc_net_state(int index, int options)
-{
-	struct timeval t;
-	int err;
-	int remote_index = -1;
-	rdc_u_info_t *urdc = &rdc_u_info[index];
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	struct set_state s;
-	struct set_state4 s4;
-	char neta[32], rneta[32];
-	unsigned short *sp;
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-	if (krdc->rpc_version < RDC_VERSION7) {
-		s4.netaddrlen = urdc->primary.addr.len;
-		s4.rnetaddrlen = urdc->secondary.addr.len;
-		bcopy(urdc->primary.addr.buf, s4.netaddr, s4.netaddrlen);
-		bcopy(urdc->secondary.addr.buf, s4.rnetaddr, s4.rnetaddrlen);
-		(void) strncpy(s4.pfile, urdc->primary.file, RDC_MAXNAMLEN);
-		(void) strncpy(s4.sfile, urdc->secondary.file, RDC_MAXNAMLEN);
-		s4.flag = options;
-
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_STATE,
-		    krdc->rpc_version, xdr_set_state4, (char *)&s4, xdr_int,
-		    (char *)&remote_index, &t);
-	} else {
-		s.netaddrlen = urdc->primary.addr.len;
-		s.rnetaddrlen = urdc->secondary.addr.len;
-		s.netaddr.buf = neta;
-		s.rnetaddr.buf = rneta;
-		bcopy(urdc->primary.addr.buf, s.netaddr.buf, s.netaddrlen);
-		bcopy(urdc->secondary.addr.buf, s.rnetaddr.buf, s.rnetaddrlen);
-		s.netaddr.len = urdc->primary.addr.len;
-		s.rnetaddr.len = urdc->secondary.addr.len;
-		s.netaddr.maxlen = urdc->primary.addr.len;
-		s.rnetaddr.maxlen = urdc->secondary.addr.len;
-		sp = (unsigned short *)s.netaddr.buf;
-		*sp = htons(*sp);
-		sp = (unsigned short *)s.rnetaddr.buf;
-		*sp = htons(*sp);
-		s.pfile = urdc->primary.file;
-		s.sfile = urdc->secondary.file;
-		s.flag = options;
-
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_STATE,
-		    krdc->rpc_version, xdr_set_state, (char *)&s, xdr_int,
-		    (char *)&remote_index, &t);
-	}
-
-	if (err)
-		return (-1);
-	else
-		return (remote_index);
-}
-
-
-/*
- * rdc_net_getbmap
- * gets the bitmaps from remote side and or's them  with remote bitmap
- */
-int
-rdc_net_getbmap(int index, int size)
-{
-	struct timeval t;
-	int err;
-	struct bmap b;
-	struct bmap6 b6;
-	rdc_k_info_t *krdc;
-
-	krdc = &rdc_k_info[index];
-
-	if (krdc->remote_index < 0)
-		return (EINVAL);
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-#ifdef DEBUG
-	if (krdc->intf == NULL)
-		cmn_err(CE_WARN,
-		    "!rdc_net_getbmap: null intf for index %d", index);
-#endif
-
-	if (krdc->rpc_version <= RDC_VERSION5) {
-		b.cd = krdc->remote_index;
-		b.dual = index;
-		b.size = size;
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_BMAP,
-		    krdc->rpc_version, xdr_bmap, (char *)&b, xdr_int,
-		    (char *)&err, &t);
-
-	} else {
-		b6.cd = krdc->remote_index;
-		b6.dual = index;
-		b6.size = size;
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_BMAP6,
-		    krdc->rpc_version, xdr_bmap6, (char *)&b6, xdr_int,
-		    (char *)&err, &t);
-	}
-	return (err);
-}
-
-int sndr_proto = 0;
-
-/*
- * return state corresponding to rdc_host
- */
-int
-rdc_net_getstate(rdc_k_info_t *krdc, int *serial_mode, int *use_mirror,
-    int *mirror_down, int network)
-{
-	int err;
-	struct timeval t;
-	int state;
-	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
-	struct set_state s;
-#ifdef sparc
-	struct set_state4 s4;
-#endif
-	char neta[32];
-	char rneta[32];
-	unsigned short *sp;
-	char *setp = (char *)&s;
-	xdrproc_t xdr_proc = xdr_set_state;
-
-	if (krdc->lsrv && (krdc->intf == NULL || krdc->intf->if_down) &&
-	    network) /* fail fast */
-		return (-1);
-
-	s.netaddrlen = urdc->primary.addr.len;
-	s.rnetaddrlen = urdc->secondary.addr.len;
-	s.pfile = urdc->primary.file;
-	s.sfile = urdc->secondary.file;
-	s.netaddr.buf = neta;
-	s.rnetaddr.buf = rneta;
-	bcopy(urdc->primary.addr.buf, s.netaddr.buf, s.netaddrlen);
-	bcopy(urdc->secondary.addr.buf, s.rnetaddr.buf, s.rnetaddrlen);
-	sp = (unsigned short *) s.netaddr.buf;
-	*sp = htons(*sp);
-	sp = (unsigned short *) s.rnetaddr.buf;
-	*sp = htons(*sp);
-	s.netaddr.len = urdc->primary.addr.len;
-	s.rnetaddr.len = urdc->secondary.addr.len;
-	s.netaddr.maxlen = urdc->primary.addr.maxlen;
-	s.rnetaddr.maxlen = urdc->secondary.addr.maxlen;
-	s.flag = 0;
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-	if (sndr_proto)
-		krdc->rpc_version = sndr_proto;
-	else
-		krdc->rpc_version = RDC_VERS_MAX;
-
-again:
-	err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSTATE4, krdc->rpc_version,
-	    xdr_proc, setp, xdr_int, (char *)&state, &t);
-
-	if (err == RPC_PROGVERSMISMATCH && (krdc->rpc_version !=
-	    RDC_VERS_MIN)) {
-		if (krdc->rpc_version-- == RDC_VERSION7) {
-			/* set_state struct changed with v7 of protocol */
-#ifdef sparc
-			s4.netaddrlen = urdc->primary.addr.len;
-			s4.rnetaddrlen = urdc->secondary.addr.len;
-			bcopy(urdc->primary.addr.buf, s4.netaddr,
-			    s4.netaddrlen);
-			bcopy(urdc->secondary.addr.buf, s4.rnetaddr,
-			    s4.rnetaddrlen);
-			(void) strncpy(s4.pfile, urdc->primary.file,
-			    RDC_MAXNAMLEN);
-			(void) strncpy(s4.sfile, urdc->secondary.file,
-			    RDC_MAXNAMLEN);
-			s4.flag = 0;
-			xdr_proc = xdr_set_state4;
-			setp = (char *)&s4;
-#else
-			/* x64 can not use protocols < 7 */
-			return (-1);
-#endif
-		}
-		goto again;
-	}
-#ifdef DEBUG
-	cmn_err(CE_NOTE, "!sndr get_state: Protocol ver %d", krdc->rpc_version);
-#endif
-
-	if (err) {
-		return (-1);
-	}
-
-	if (state == -1)
-		return (-1);
-
-	if (serial_mode)
-		*serial_mode = (state >> 2) & 1;
-	if (use_mirror)
-		*use_mirror = (state >> 1) & 1;
-	if (mirror_down)
-		*mirror_down = state & 1;
-
-	return (0);
-}
-
-
-static struct xdr_discrim rdres_discrim[2] = {
-	{ (int)RDC_OK, xdr_readok },
-	{ __dontcare__, NULL_xdrproc_t }
-};
-
-
-/*
- * Reply from remote read (client side)
- */
-static bool_t
-xdr_rdresult(XDR *xdrs, readres *rr)
-{
-
-	return (xdr_union(xdrs, (enum_t *)&(rr->rr_status),
-	    (caddr_t)&(rr->rr_ok), rdres_discrim, xdr_void));
-}
-
-static int
-rdc_rrstatus_decode(int status)
-{
-	int ret = 0;
-
-	if (status != RDC_OK) {
-		switch (status) {
-		case RDCERR_NOENT:
-			ret = ENOENT;
-			break;
-		case RDCERR_NOMEM:
-			ret = ENOMEM;
-			break;
-		default:
-			ret = EIO;
-			break;
-		}
-	}
-
-	return (ret);
-}
-
-
-int
-rdc_net_read(int local_index, int remote_index, nsc_buf_t *handle,
-    nsc_off_t fba_pos, nsc_size_t fba_len)
-{
-	struct rdcrdresult rr;
-	rdc_k_info_t *krdc;
-	rdc_u_info_t *urdc;
-	struct rread list;
-	struct rread6 list6;
-	struct timeval t;
-	uchar_t *sv_addr;
-	nsc_vec_t *vec;
-	int rpc_flag;
-	nsc_size_t sv_len;
-	int err;
-	int ret;
-	nsc_size_t len;
-	nsc_size_t maxfbas;
-	int transflag;
-
-	if (handle == NULL)
-		return (EINVAL);
-
-	if (!RDC_HANDLE_LIMITS(handle, fba_pos, fba_len)) {
-#ifdef DEBUG
-		cmn_err(CE_NOTE, "!rdc_net_read: handle bounds");
-#endif
-		return (EINVAL);
-	}
-
-	krdc = &rdc_k_info[local_index];
-	urdc = &rdc_u_info[local_index];
-
-	maxfbas = MAX_RDC_FBAS;
-
-	if (krdc->remote_fd && !(rdc_get_vflags(urdc) & RDC_FCAL_FAILED)) {
-		nsc_buf_t *remote_h = NULL;
-		int reserved = 0;
-
-		ret = nsc_reserve(krdc->remote_fd, NSC_MULTI);
-		if (RDC_SUCCESS(ret)) {
-			reserved = 1;
-			ret = nsc_alloc_buf(krdc->remote_fd, fba_pos, fba_len,
-			    NSC_RDBUF, &remote_h);
-		}
-		if (RDC_SUCCESS(ret)) {
-			ret = nsc_copy(remote_h, handle, fba_pos, fba_pos,
-			    fba_len);
-			if (RDC_SUCCESS(ret)) {
-				(void) nsc_free_buf(remote_h);
-				nsc_release(krdc->remote_fd);
-				return (0);
-			}
-		}
-		rdc_group_enter(krdc);
-		rdc_set_flags(urdc, RDC_FCAL_FAILED);
-		rdc_group_exit(krdc);
-		if (remote_h)
-			(void) nsc_free_buf(remote_h);
-		if (reserved)
-			nsc_release(krdc->remote_fd);
-	}
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-	if (rdc_get_vflags(urdc) & RDC_VOL_FAILED)
-		rpc_flag = RDC_RREAD_FAIL;
-	else
-		rpc_flag = 0;
-
-#ifdef DEBUG
-	if (krdc->intf == NULL)
-		cmn_err(CE_WARN,
-		    "!rdc_net_read: null intf for index %d", local_index);
-#endif
-	/*
-	 * switch on proto version.
-	 */
-	len = fba_len;		/* length (FBAs) still to xfer */
-	rr.rr_bufsize = 0;	/* rpc data buffer length (bytes) */
-	rr.rr_data = NULL;	/* rpc data buffer */
-	transflag = rpc_flag | RDC_RREAD_START;	/* setup rpc */
-	if (krdc->rpc_version <= RDC_VERSION5) {
-		ASSERT(fba_pos <= INT32_MAX);
-		list.pos = (int)fba_pos; /* fba position of start of chunk */
-		list.cd = remote_index;	/* remote end cd */
-		/* send setup rpc */
-		list.flag = transflag;
-		ASSERT(len <= INT32_MAX);
-		list.len = (int)len;			/* total fba length */
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
-		    krdc->rpc_version, xdr_rread, (char *)&list, xdr_int,
-		    (char *)&ret, &t);
-
-	} else {
-		list6.pos = fba_pos;	/* fba position of start of chunk */
-		list6.cd = remote_index;	/* remote end cd */
-		/* send setup rpc */
-		list6.flag = transflag;	/* setup rpc */
-		ASSERT(len <= INT32_MAX);
-		list6.len = (int)len;			/* total fba length */
-		err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
-		    krdc->rpc_version, xdr_rread6, (char *)&list6, xdr_int,
-		    (char *)&ret, &t);
-	}
-
-	if (err) {
-#ifdef DEBUG
-		cmn_err(CE_NOTE, "!rdc_net_read: setup err %d", err);
-#endif
-		if (err == RPC_INTR)
-			ret = EINTR;
-		else
-			ret = ENOLINK;
-
-		goto remote_rerror;
-	}
-
-	if (ret == 0) {		/* No valid index from r_net_read */
-#ifdef DEBUG
-		cmn_err(CE_NOTE,
-		    "!rdc_net_read: no valid index from r_net_read");
-#endif
-		return (ENOBUFS);
-	}
-	transflag = rpc_flag | RDC_RREAD_DATA;
-	if (krdc->rpc_version <= RDC_VERSION5) {
-		list.idx = ret;		/* save idx to return to server */
-		list.flag = transflag;
-					/* move onto to data xfer rpcs */
-	} else {
-		list6.idx = ret;	/* save idx to return to server */
-		list6.flag = transflag;
-	}
-
-	/* find starting position in handle */
-
-	vec = handle->sb_vec;
-
-	fba_pos -= handle->sb_pos;
-
-	for (; fba_pos >= FBA_NUM(vec->sv_len); vec++)
-		fba_pos -= FBA_NUM(vec->sv_len);
-
-	sv_addr = vec->sv_addr + FBA_SIZE(fba_pos);	/* data in vector */
-	sv_len = vec->sv_len - FBA_SIZE(fba_pos);	/* bytes in vector */
-
-	while (len) {
-		nsc_size_t translen;
-		if (len > maxfbas) {
-			translen = maxfbas;
-		} else {
-			translen = len;
-		}
-
-		if (FBA_SIZE(translen) > sv_len) {
-			translen = FBA_NUM(sv_len);
-		}
-
-		len -= translen;
-		if (len == 0) {
-			/* last data xfer rpc - tell server to cleanup */
-			transflag |= RDC_RREAD_END;
-		}
-
-		if (!rr.rr_data || (nsc_size_t)rr.rr_bufsize !=
-		    FBA_SIZE(translen)) {
-			if (rr.rr_data)
-				kmem_free(rr.rr_data, rr.rr_bufsize);
-
-			ASSERT(FBA_SIZE(translen) <= INT32_MAX);
-			rr.rr_bufsize = FBA_SIZE(translen);
-			rr.rr_data = kmem_alloc(rr.rr_bufsize, KM_NOSLEEP);
-		}
-
-		if (!rr.rr_data) {
-			/* error */
-#ifdef DEBUG
-			cmn_err(CE_NOTE, "!rdc_net_read: kmem_alloc failed");
-#endif
-			return (ENOMEM);
-		}
-
-		/* get data from remote end */
-
-#ifdef DEBUG
-		if (krdc->intf == NULL)
-			cmn_err(CE_WARN,
-			    "!rdc_net_read: null intf for index %d",
-			    local_index);
-#endif
-		if (krdc->io_kstats) {
-			mutex_enter(krdc->io_kstats->ks_lock);
-			kstat_runq_enter(KSTAT_IO_PTR(krdc->io_kstats));
-			mutex_exit(krdc->io_kstats->ks_lock);
-		}
-		/*CONSTCOND*/
-		ASSERT(RDC_MAXDATA <= INT32_MAX);
-		ASSERT(translen <= RDC_MAXDATA);
-		if (krdc->rpc_version <= RDC_VERSION5) {
-			list.len = (int)translen;
-			list.flag = transflag;
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
-			    krdc->rpc_version, xdr_rread, (char *)&list,
-			    xdr_rdresult, (char *)&rr, &t);
-		} else {
-			list6.len = (int)translen;
-			list6.flag = transflag;
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
-			    krdc->rpc_version, xdr_rread6, (char *)&list6,
-			    xdr_rdresult, (char *)&rr, &t);
-		}
-
-		if (krdc->io_kstats) {
-			mutex_enter(krdc->io_kstats->ks_lock);
-			kstat_runq_exit(KSTAT_IO_PTR(krdc->io_kstats));
-			mutex_exit(krdc->io_kstats->ks_lock);
-		}
-
-		if (err) {
-#ifdef DEBUG
-			cmn_err(CE_NOTE, "!rdc_net_read: rpc err %d", err);
-#endif
-			if (err == RPC_INTR) {
-				ret = EINTR;
-			} else {
-				ret = ENOLINK;
-			}
-
-			goto remote_rerror;
-		}
-
-		if (rr.rr_status != RDC_OK) {
-			ret = rdc_rrstatus_decode(rr.rr_status);
-			if (!ret)
-				ret = EIO;
-
-			goto remote_rerror;
-		}
-
-		/* copy into handle */
-
-		bcopy(rr.rr_data, sv_addr, (size_t)rr.rr_bufsize);
-
-		/* update counters */
-
-		sv_addr += rr.rr_bufsize;
-		if (krdc->rpc_version <= RDC_VERSION5) {
-			list.pos += translen;
-		} else {
-			list6.pos += translen;
-		}
-		if (krdc->io_kstats) {
-			KSTAT_IO_PTR(krdc->io_kstats)->reads++;
-			KSTAT_IO_PTR(krdc->io_kstats)->nread += rr.rr_bufsize;
-		}
-		ASSERT(sv_len <= INT32_MAX);
-		ASSERT(sv_len >= (nsc_size_t)rr.rr_bufsize);
-		sv_len -= rr.rr_bufsize;
-
-		if (sv_len == 0) {
-			/* goto next vector */
-			vec++;
-			sv_addr = vec->sv_addr;
-			sv_len = vec->sv_len;
-		}
-	}
-
-	if (rr.rr_data)
-		kmem_free(rr.rr_data, rr.rr_bufsize);
-
-	return (0);
-
-remote_rerror:
-	if (rr.rr_data)
-		kmem_free(rr.rr_data, rr.rr_bufsize);
-
-	return (ret ? ret : ENOLINK);
-}
-
-/*
- * rdc_net_write
- * Main remote write client side
- * Handles protocol selection as well as requests for remote allocation
- * and data transfer
- * Does local IO for FCAL
- * caller must clear bitmap on success
- */
-
-int
-rdc_net_write(int local_index, int remote_index, nsc_buf_t *handle,
-    nsc_off_t fba_pos, nsc_size_t fba_len, uint_t aseq, int qpos,
-    netwriteres *netres)
-{
-	rdc_k_info_t *krdc;
-	rdc_u_info_t *urdc;
-	struct timeval t;
-	nsc_vec_t *vec;
-	int sv_len;
-	nsc_off_t fpos;
-	int err;
-	struct netwriteres netret;
-	struct netwriteres *netresptr;
-	struct net_data5 dlist5;
-	struct net_data6 dlist6;
-	int ret;
-	nsc_size_t maxfbas;
-	int transflag;
-	int translen;
-	int transendoblk;
-	char *transptr;
-	int vflags;
-
-	if (handle == NULL)
-		return (EINVAL);
-
-	/* if not a diskq buffer */
-	if ((qpos == -1) && (!RDC_HANDLE_LIMITS(handle, fba_pos, fba_len))) {
-#ifdef DEBUG
-		cmn_err(CE_NOTE, "!rdc_net_write: handle bounds");
-#endif
-		return (EINVAL);
-	}
-
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-	krdc = &rdc_k_info[local_index];
-	urdc = &rdc_u_info[local_index];
-
-	maxfbas = MAX_RDC_FBAS;
-
-	/* FCAL IO */
-	if (krdc->remote_fd && !(rdc_get_vflags(urdc) & RDC_FCAL_FAILED)) {
-		nsc_buf_t *remote_h = NULL;
-		int reserved = 0;
-
-		ret = nsc_reserve(krdc->remote_fd, NSC_MULTI);
-		if (RDC_SUCCESS(ret)) {
-			reserved = 1;
-			ret = nsc_alloc_buf(krdc->remote_fd, fba_pos, fba_len,
-			    NSC_WRBUF, &remote_h);
-		}
-		if (RDC_SUCCESS(ret)) {
-			ret = nsc_copy(handle, remote_h, fba_pos, fba_pos,
-			    fba_len);
-			if (RDC_SUCCESS(ret))
-				ret = nsc_write(remote_h, fba_pos, fba_len, 0);
-			if (RDC_SUCCESS(ret)) {
-				(void) nsc_free_buf(remote_h);
-				nsc_release(krdc->remote_fd);
-				return (0);
-			}
-		}
-		rdc_group_enter(krdc);
-		rdc_set_flags(urdc, RDC_FCAL_FAILED);
-		rdc_group_exit(krdc);
-		if (remote_h)
-			(void) nsc_free_buf(remote_h);
-		if (reserved)
-			nsc_release(krdc->remote_fd);
-	}
-
-	/*
-	 * At this point we must decide which protocol we are using and
-	 * do the right thing
-	 */
-	netret.vecdata.vecdata_val = NULL;
-	netret.vecdata.vecdata_len = 0;
-	if (netres) {
-		netresptr = netres;
-	} else {
-		netresptr = &netret;
-	}
-
-	vflags = rdc_get_vflags(urdc);
-
-	if (vflags & (RDC_VOL_FAILED|RDC_BMP_FAILED))
-		transflag = RDC_RWRITE_FAIL;
-	else
-		transflag = 0;
-
-
-#ifdef DEBUG
-	if (krdc->intf == NULL)
-		cmn_err(CE_WARN, "!rdc_net_write: null intf for index %d",
-		    local_index);
-#endif
-
-	vec = handle->sb_vec;
-
-	/*
-	 * find starting position in vector
-	 */
-	if ((qpos == -1) || (handle->sb_user == RDC_NULLBUFREAD))
-		fpos = fba_pos - handle->sb_pos;
-	else
-		fpos = (qpos + 1) - handle->sb_pos;
-
-	for (; fpos >= FBA_NUM(vec->sv_len); vec++)
-		fpos -= FBA_NUM(vec->sv_len);
-	sv_len = vec->sv_len - FBA_SIZE(fpos);	/* bytes in vector */
-	transptr = (char *)vec->sv_addr + FBA_SIZE(fpos);
-
-	if (krdc->rpc_version <= RDC_VERSION5) {
-		dlist5.local_cd = local_index;
-		dlist5.cd = remote_index;
-		ASSERT(fba_len <= INT32_MAX);
-		ASSERT(fba_pos <= INT32_MAX);
-		dlist5.len = (int)fba_len;
-		dlist5.pos = (int)fba_pos;
-		dlist5.idx = -1; /* Starting index */
-		dlist5.flag = transflag;
-		dlist5.seq = aseq;		/* sequence number */
-		dlist5.sfba = (int)fba_pos;	/* starting fba for this xfer */
-	} else {
-		dlist6.local_cd = local_index;
-		dlist6.cd = remote_index;
-		ASSERT(fba_len <= INT32_MAX);
-		dlist6.len = (int)fba_len;
-		dlist6.qpos = qpos;
-		dlist6.pos = fba_pos;
-		dlist6.idx = -1; /* Starting index */
-		dlist6.flag = transflag;
-		dlist6.seq = aseq;		/* sequence number */
-		dlist6.sfba = fba_pos;		/* starting fba for this xfer */
-	}
-
-	transendoblk = 0;
-	while (fba_len) {
-		if (!transptr) {
-#ifdef DEBUG
-			cmn_err(CE_WARN,
-			    "!rdc_net_write: walked off end of handle!");
-#endif
-			ret = EINVAL;
-			goto remote_error;
-		}
-
-		if (fba_len > maxfbas) {
-			ASSERT(maxfbas <= INT32_MAX);
-			translen = (int)maxfbas;
-		} else {
-			ASSERT(fba_len <= INT32_MAX);
-			translen = (int)fba_len;
-		}
-
-		if (FBA_SIZE(translen) > sv_len) {
-			translen = FBA_NUM(sv_len);
-		}
-
-		fba_len -= translen;
-		if (fba_len == 0) {
-			/* last data xfer - tell server to commit */
-			transendoblk = 1;
-		}
-
-
-#ifdef DEBUG
-		if (krdc->intf == NULL)
-			cmn_err(CE_WARN,
-			    "!rdc_net_write: null intf for index %d",
-			    local_index);
-#endif
-		DTRACE_PROBE(rdc_netwrite_clntcall_start);
-
-		if (krdc->io_kstats) {
-			mutex_enter(krdc->io_kstats->ks_lock);
-			kstat_runq_enter(KSTAT_IO_PTR(krdc->io_kstats));
-			mutex_exit(krdc->io_kstats->ks_lock);
-		}
-		if (krdc->rpc_version <= RDC_VERSION5) {
-			ret = 0;
-			dlist5.nfba = translen;
-			dlist5.endoblk = transendoblk;
-			dlist5.data.data_len = FBA_SIZE(translen);
-			dlist5.data.data_val = transptr;
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE5,
-			    krdc->rpc_version, xdr_net_data5,
-			    (char *)&dlist5, xdr_int,
-			    (char *)&ret, &t);
-			if (ret >= 0) {
-				netresptr->result = 0;
-				netresptr->index = ret;
-			} else {
-				netresptr->result = ret;
-			}
-		} else {
-			netresptr->result = 0;
-			dlist6.nfba = translen;
-			dlist6.endoblk = transendoblk;
-			dlist6.data.data_len = FBA_SIZE(translen);
-			dlist6.data.data_val = transptr;
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE6,
-			    krdc->rpc_version, xdr_net_data6,
-			    (char *)&dlist6, xdr_netwriteres,
-			    (char *)netresptr, &t);
-		}
-
-		if (krdc->io_kstats) {
-			mutex_enter(krdc->io_kstats->ks_lock);
-			kstat_runq_exit(KSTAT_IO_PTR(krdc->io_kstats));
-			mutex_exit(krdc->io_kstats->ks_lock);
-		}
-
-		DTRACE_PROBE(rdc_netwrite_clntcall_end);
-		ret = netresptr->result;
-		if (err) {
-			if (err == RPC_INTR)
-				ret = EINTR;
-			else if (err && ret != EPROTO)
-				ret = ENOLINK;
-#ifdef DEBUG
-			cmn_err(CE_NOTE,
-			    "!rdc_net_write(5): cd %d err %d ret %d",
-			    remote_index, err, ret);
-#endif
-			goto remote_error;
-		}
-		/* Error from r_net_write5 */
-		if (netresptr->result < 0) {
-#ifdef DEBUG
-			cmn_err(CE_NOTE,
-			    "!rdc_net_write: r_net_write(5) "
-			    "returned: %d",
-			    -netresptr->result);
-#endif
-			ret = -netresptr->result;
-			if (netret.vecdata.vecdata_val)
-				kmem_free(netret.vecdata.vecdata_val,
-				    netret.vecdata.vecdata_len *
-				    sizeof (net_pendvec_t));
-			goto remote_error;
-		} else if (netresptr->index == 0) {
-#ifdef DEBUG
-			cmn_err(CE_NOTE,
-			    "!rdc_net_write: no valid index from "
-			    "r_net_write(5)");
-#endif
-			ret = ENOBUFS;
-			if (netret.vecdata.vecdata_val)
-				kmem_free(netret.vecdata.vecdata_val,
-				    netret.vecdata.vecdata_len *
-				    sizeof (net_pendvec_t));
-			goto remote_error;
-		}
-		if (krdc->rpc_version <= RDC_VERSION5) {
-			dlist5.idx = netresptr->index;
-			dlist5.sfba += dlist5.nfba;
-		} else {
-			dlist6.idx = netresptr->index;
-			dlist6.sfba += dlist6.nfba;
-		}
-		/* update counters */
-		if (krdc->io_kstats) {
-			KSTAT_IO_PTR(krdc->io_kstats)->writes++;
-			KSTAT_IO_PTR(krdc->io_kstats)->nwritten +=
-			    FBA_SIZE(translen);
-		}
-		transptr += FBA_SIZE(translen);
-		sv_len -= FBA_SIZE(translen);
-
-		if (sv_len <= 0) {
-			/* goto next vector */
-			vec++;
-			transptr = (char *)vec->sv_addr;
-			sv_len = vec->sv_len;
-		}
-	}
-	/*
-	 * this can't happen.....
-	 */
-	if (netret.vecdata.vecdata_val)
-		kmem_free(netret.vecdata.vecdata_val,
-		    netret.vecdata.vecdata_len *
-		    sizeof (net_pendvec_t));
-
-	return (0);
-
-remote_error:
-	return (ret ? ret : ENOLINK);
-}
-
-void
-rdc_fixlen(rdc_aio_t *aio)
-{
-	nsc_vec_t *vecp = aio->qhandle->sb_vec;
-	nsc_size_t len = 0;
-
-	while (vecp->sv_addr) {
-		len += FBA_NUM(vecp->sv_len);
-		vecp++;
-	}
-	aio->qhandle->sb_len = len;
-}
-
-/*
- * rdc_dump_alloc_bufs_cd
- * Dump allocated buffers (rdc_net_hnd's) for the specified cd.
- * this could be the flusher failing, if so, don't do the delay forever
- * Returns: 0 (success), EAGAIN (caller needs to try again).
- */
-int
-rdc_dump_alloc_bufs_cd(int index)
-{
-	rdc_k_info_t *krdc;
-	rdc_aio_t *aio;
-	net_queue *q;
-	disk_queue *dq;
-	kmutex_t *qlock;
-
-	krdc = &rdc_k_info[index];
-
-
-	if (!krdc->c_fd) {
-		/* cannot do anything! */
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_dump_alloc_bufs_cd(%d): c_fd NULL",
-		    index);
-#endif
-		return (0);
-	}
-	rdc_dump_dsets(index);
-
-	dq = &krdc->group->diskq;
-
-	if (RDC_IS_DISKQ(krdc->group)) {
-		qlock = QLOCK(dq);
-		(void) _rdc_rsrv_diskq(krdc->group);
-	} else {
-		qlock = &krdc->group->ra_queue.net_qlock;
-	}
-
-	/*
-	 * Now dump the async queue anonymous buffers
-	 * if we are a diskq, the we are using the diskq mutex.
-	 * However, we are flushing from diskq to memory queue
-	 * so we now need to grab the memory lock also
-	 */
-
-	q = &krdc->group->ra_queue;
-
-	if (RDC_IS_DISKQ(krdc->group)) {
-		mutex_enter(&q->net_qlock);
-		if (q->qfill_sleeping == RDC_QFILL_AWAKE) {
-			int tries = 5;
-#ifdef DEBUG_DISKQ
-			cmn_err(CE_NOTE,
-			    "!dumpalloccd sending diskq->memq flush to sleep");
-#endif
-			q->qfflags |= RDC_QFILLSLEEP;
-			mutex_exit(&q->net_qlock);
-
-			while (q->qfill_sleeping == RDC_QFILL_AWAKE && tries--)
-				delay(5);
-			mutex_enter(&q->net_qlock);
-		}
-	}
-
-	mutex_enter(qlock);
-
-	while ((q->net_qhead != NULL)) {
-		rdc_k_info_t *tmpkrdc;
-		aio = q->net_qhead;
-		tmpkrdc = &rdc_k_info[aio->index];
-
-		if (RDC_IS_DISKQ(krdc->group)) {
-			aio->qhandle->sb_user--;
-			if (aio->qhandle->sb_user == 0) {
-				rdc_fixlen(aio);
-				(void) nsc_free_buf(aio->qhandle);
-				aio->qhandle = NULL;
-				aio->handle = NULL;
-			}
-		} else {
-			if (aio->handle) {
-				(void) nsc_free_buf(aio->handle);
-				aio->handle = NULL;
-			}
-		}
-
-		if (tmpkrdc->io_kstats && !RDC_IS_DISKQ(krdc->group)) {
-			mutex_enter(tmpkrdc->io_kstats->ks_lock);
-			kstat_waitq_exit(KSTAT_IO_PTR(tmpkrdc->io_kstats));
-			mutex_exit(tmpkrdc->io_kstats->ks_lock);
-		}
-		q->net_qhead = q->net_qhead->next;
-		q->blocks -= aio->len;
-		q->nitems--;
-
-		RDC_CHECK_BIT(tmpkrdc, aio->pos, aio->len);
-
-		kmem_free(aio, sizeof (*aio));
-	}
-	q->net_qtail = NULL;
-
-	if (krdc->group->asyncstall) {
-		krdc->group->asyncdis = 1;
-		cv_broadcast(&krdc->group->asyncqcv);
-	}
-	if (krdc->group->sleepq) {
-		rdc_sleepqdiscard(krdc->group);
-	}
-
-	krdc->group->seq = RDC_NEWSEQ;
-	krdc->group->seqack = RDC_NEWSEQ;
-	if (RDC_IS_DISKQ(krdc->group)) {
-		rdc_dump_iohdrs(dq);
-		SET_QNXTIO(dq, QHEAD(dq));
-		SET_QCOALBOUNDS(dq, QHEAD(dq));
-	}
-	mutex_exit(qlock);
-
-	if (RDC_IS_DISKQ(krdc->group)) {
-		mutex_exit(&q->net_qlock);
-		_rdc_rlse_diskq(krdc->group);
-	}
-
-	return (0);
-}
-
-
-/*
- * rdc_dump_alloc_bufs
- * We have an error on the link
- * Try to dump all of the allocated bufs so we can cleanly recover
- * and not hang
- */
-void
-rdc_dump_alloc_bufs(rdc_if_t *ip)
-{
-	rdc_k_info_t *krdc;
-	int repeat;
-	int index;
-
-	for (index = 0; index < rdc_max_sets; index++) {
-		do {
-			krdc = &rdc_k_info[index];
-			repeat = 0;
-			if (krdc->intf == ip) {
-				if (rdc_dump_alloc_bufs_cd(index) == EAGAIN) {
-					repeat = 1;
-					delay(2);
-				}
-			}
-		} while (repeat);
-	}
-}
-
-/*
- * returns 1 if the the throttle should throttle, 0 if not.
- */
-int
-_rdc_diskq_isfull(disk_queue *q, long len)
-{
-	/* ---T----H----N--- */
-	mutex_enter(QLOCK(q));
-
-	if (FITSONQ(q, len + 1)) {
-		mutex_exit(QLOCK(q));
-		return (0);
-	}
-	mutex_exit(QLOCK(q));
-	return (1);
-}
-
-void
-_rdc_async_throttle(rdc_k_info_t *this, long len)
-{
-	rdc_k_info_t *krdc;
-	rdc_u_info_t *urdc;
-	int print_msg = 1;
-	int tries = RDC_FUTILE_ATTEMPTS;
-
-	/*
-	 * Throttle entries on queue
-	 */
-
-	/* Need to take the 1-many case into account, checking all sets */
-
-	/* ADD HANDY HUERISTIC HERE TO SLOW DOWN IO */
-	for (krdc = this; /* CSTYLED */; krdc = krdc->many_next) {
-		urdc = &rdc_u_info[krdc->index];
-
-		/*
-		 * this may be the last set standing in a one to many setup.
-		 * we may also be stuck in unintercept, after marking
-		 * the volume as not enabled, but have not removed it
-		 * from the many list resulting in an endless loop if
-		 * we just continue here. Lets jump over this stuff
-		 * and check to see if we are the only dude here.
-		 */
-		if (!IS_ENABLED(urdc))
-			goto thischeck;
-
-		if (IS_ASYNC(urdc) && RDC_IS_MEMQ(krdc->group)) {
-			net_queue *q = &krdc->group->ra_queue;
-			while ((q->blocks + q->inflbls) > urdc->maxqfbas ||
-			    (q->nitems + q->inflitems) > urdc->maxqitems) {
-
-				if (!IS_ENABLED(urdc)) /* disable race */
-					goto thischeck;
-
-				if (!krdc->group->rdc_writer)
-					(void) rdc_writer(krdc->index);
-				delay(2);
-				q->throttle_delay++;
-			}
-		}
-
-		/* do a much more aggressive delay, get disk flush going */
-		if (IS_ASYNC(urdc) && RDC_IS_DISKQ(krdc->group)) {
-			disk_queue *q = &krdc->group->diskq;
-			while ((!IS_QSTATE(q, RDC_QNOBLOCK)) &&
-			    (_rdc_diskq_isfull(q, len)) &&
-			    (!IS_STATE(urdc, RDC_DISKQ_FAILED))) {
-				if (print_msg) {
-					cmn_err(CE_WARN, "!rdc async throttle:"
-					    " disk queue %s full",
-					    &urdc->disk_queue[0]);
-
-					print_msg = 0;
-				}
-				if (!IS_ENABLED(urdc)) /* disable race */
-					goto thischeck;
-
-				if (!krdc->group->rdc_writer)
-					(void) rdc_writer(krdc->index);
-				delay(10);
-				q->throttle_delay += 10;
-
-				if (!(tries--) && IS_STATE(urdc, RDC_QUEUING)) {
-					cmn_err(CE_WARN, "!SNDR: disk queue "
-					    "%s full & not flushing. giving up",
-					    &urdc->disk_queue[0]);
-					cmn_err(CE_WARN, "!SNDR: %s:%s entering"
-					    " logging mode",
-					    urdc->secondary.intf,
-					    urdc->secondary.file);
-					rdc_fail_diskq(krdc, RDC_WAIT,
-					    RDC_DOLOG | RDC_NOFAIL);
-					mutex_enter(QLOCK(q));
-					cv_broadcast(&q->qfullcv);
-					mutex_exit(QLOCK(q));
-				}
-
-			}
-			if ((IS_QSTATE(q, RDC_QNOBLOCK)) &&
-			    _rdc_diskq_isfull(q, len) &&
-			    !IS_STATE(urdc, RDC_DISKQ_FAILED)) {
-				if (print_msg) {
-					cmn_err(CE_WARN, "!disk queue %s full",
-					    &urdc->disk_queue[0]);
-					print_msg = 0;
-				}
-				rdc_fail_diskq(krdc, RDC_WAIT,
-				    RDC_DOLOG | RDC_NOFAIL);
-				mutex_enter(QLOCK(q));
-				cv_broadcast(&q->qfullcv);
-				mutex_exit(QLOCK(q));
-			}
-		}
-
-thischeck:
-		if (krdc->many_next == this)
-			break;
-	}
-}
-
-int rdc_coalesce = 1;
-static int rdc_joins = 0;
-
-int
-rdc_aio_coalesce(rdc_aio_t *queued, rdc_aio_t *new)
-{
-	nsc_buf_t *h = NULL;
-	int rc;
-	rdc_k_info_t *krdc;
-	uint_t bitmask;
-
-	if (rdc_coalesce == 0)
-		return (0);		/* don't even try */
-
-	if ((queued == NULL) ||
-	    (queued->handle == NULL) ||
-	    (new->handle == NULL)) {
-		return (0);		/* existing queue is empty */
-	}
-	if (queued->index != new->index || queued->len + new->len >
-	    MAX_RDC_FBAS) {
-		return (0);		/* I/O to big */
-	}
-	if ((queued->pos + queued->len == new->pos) ||
-	    (new->pos + new->len == queued->pos)) {
-		rc = nsc_alloc_abuf(queued->pos, queued->len + new->len, 0,
-		    &h);
-		if (!RDC_SUCCESS(rc)) {
-			if (h != NULL)
-				(void) nsc_free_buf(h);
-			return (0);		/* couldn't do coalesce */
-		}
-		rc = nsc_copy(queued->handle, h, queued->pos, queued->pos,
-		    queued->len);
-		if (!RDC_SUCCESS(rc)) {
-			(void) nsc_free_buf(h);
-			return (0);		/* couldn't do coalesce */
-		}
-		rc = nsc_copy(new->handle, h, new->pos, new->pos,
-		    new->len);
-		if (!RDC_SUCCESS(rc)) {
-			(void) nsc_free_buf(h);
-			return (0);		/* couldn't do coalesce */
-		}
-
-		krdc = &rdc_k_info[queued->index];
-
-		RDC_SET_BITMASK(queued->pos, queued->len, &bitmask);
-		RDC_CLR_BITMAP(krdc, queued->pos, queued->len, \
-		    bitmask, RDC_BIT_BUMP);
-
-		RDC_SET_BITMASK(new->pos, new->len, &bitmask);
-		RDC_CLR_BITMAP(krdc, new->pos, new->len, \
-		    bitmask, RDC_BIT_BUMP);
-
-		(void) nsc_free_buf(queued->handle);
-		(void) nsc_free_buf(new->handle);
-		queued->handle = h;
-		queued->len += new->len;
-		bitmask = 0;
-		/*
-		 * bump the ref count back up
-		 */
-
-		RDC_SET_BITMAP(krdc, queued->pos, queued->len, &bitmask);
-		return (1);	/* new I/O succeeds last I/O queued */
-	}
-	return (0);
-}
-
-int
-rdc_memq_enqueue(rdc_k_info_t *krdc, rdc_aio_t *aio)
-{
-	net_queue *q;
-	rdc_group_t *group;
-
-	group = krdc->group;
-	q = &group->ra_queue;
-
-	mutex_enter(&q->net_qlock);
-
-	if (rdc_aio_coalesce(q->net_qtail, aio)) {
-		rdc_joins++;
-		q->blocks += aio->len;
-		kmem_free(aio, sizeof (*aio));
-		goto out;
-	}
-	aio->seq = group->seq++;
-	if (group->seq < aio->seq)
-		group->seq = RDC_NEWSEQ + 1; /* skip magics */
-
-	if (q->net_qhead == NULL) {
-		/* adding to empty q */
-		q->net_qhead = q->net_qtail = aio;
-
-#ifdef DEBUG
-		if (q->blocks != 0 || q->nitems != 0) {
-			cmn_err(CE_PANIC,
-			    "rdc enqueue: q %p, qhead 0, q blocks %" NSC_SZFMT
-			    ", nitems %" NSC_SZFMT,
-			    (void *) q, q->blocks, q->nitems);
-		}
-#endif
-
-	} else {
-		/* discontiguous, add aio to q tail */
-		q->net_qtail->next = aio;
-		q->net_qtail = aio;
-	}
-
-	q->blocks += aio->len;
-	q->nitems++;
-
-	if (krdc->io_kstats) {
-		mutex_enter(krdc->io_kstats->ks_lock);
-		kstat_waitq_enter(KSTAT_IO_PTR(krdc->io_kstats));
-		mutex_exit(krdc->io_kstats->ks_lock);
-	}
-out:
-#ifdef DEBUG
-	/* sum the q and check for sanity */
-	{
-		nsc_size_t qblocks = 0;
-		uint64_t nitems = 0;
-		rdc_aio_t *a;
-
-		for (a = q->net_qhead; a != NULL; a = a->next) {
-			qblocks += a->len;
-			nitems++;
-		}
-
-		if (qblocks != q->blocks || nitems != q->nitems) {
-			cmn_err(CE_PANIC,
-			    "rdc enqueue: q %p, q blocks %" NSC_SZFMT " (%"
-			    NSC_SZFMT "), nitems %" NSC_SZFMT " (%"
-			    NSC_SZFMT ")", (void *) q, q->blocks, qblocks,
-			    q->nitems, nitems);
-		}
-	}
-#endif
-
-	mutex_exit(&q->net_qlock);
-
-	if (q->nitems > q->nitems_hwm) {
-		q->nitems_hwm = q->nitems;
-	}
-
-	if (q->blocks > q->blocks_hwm) {
-		q->blocks_hwm = q->blocks;
-	}
-
-	if (!krdc->group->rdc_writer)
-		(void) rdc_writer(krdc->index);
-
-	return (0);
-}
-
-int
-_rdc_enqueue_write(rdc_k_info_t *krdc, nsc_off_t pos, nsc_size_t len, int flag,
-    nsc_buf_t *h)
-{
-	rdc_aio_t *aio;
-	rdc_group_t *group;
-	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
-	int rc;
-
-	aio = kmem_zalloc(sizeof (*aio), KM_NOSLEEP);
-	if (!aio) {
-		return (ENOMEM);
-	}
-
-	group = krdc->group;
-
-	aio->pos = pos;
-	aio->qpos = -1;
-	aio->len = len;
-	aio->flag = flag;
-	aio->index = krdc->index;
-	aio->handle = h;
-
-	if (group->flags & RDC_MEMQUE) {
-		return (rdc_memq_enqueue(krdc, aio));
-	} else if ((group->flags & RDC_DISKQUE) &&
-	    !IS_STATE(urdc, RDC_DISKQ_FAILED)) {
-		rc = rdc_diskq_enqueue(krdc, aio);
-		kmem_free(aio, sizeof (*aio));
-		return (rc);
-	}
-	return (-1); /* keep lint quiet */
-}
-
-
-
-
-/*
- * Async Network RDC flusher
- */
-
-/*
- * don't allow any new writer threads to start if a member of the set
- * is disable pending
- */
-int
-is_disable_pending(rdc_k_info_t *krdc)
-{
-	rdc_k_info_t *this = krdc;
-	int rc = 0;
-
-	do {
-		if (krdc->type_flag & RDC_DISABLEPEND) {
-			krdc = this;
-			rc = 1;
-			break;
-		}
-		krdc = krdc->group_next;
-
-	} while (krdc != this);
-
-	return (rc);
-}
-
-/*
- * rdc_writer -- spawn new writer if not running already
- *	called after enqueing the dirty blocks
- */
-int
-rdc_writer(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	nsthread_t *t;
-	rdc_group_t	*group;
-	kmutex_t	*qlock;
-	int tries;
-	const int MAX_TRIES = 16;
-
-	group = krdc->group;
-
-	if (RDC_IS_DISKQ(group))
-		qlock = &group->diskq.disk_qlock;
-	else
-		qlock = &group->ra_queue.net_qlock;
-
-	mutex_enter(qlock);
-
-#ifdef DEBUG
-	if (noflush) {
-		mutex_exit(qlock);
-		return (0);
-	}
-#endif
-
-	if ((group->rdc_writer) || is_disable_pending(krdc)) {
-		mutex_exit(qlock);
-		return (0);
-	}
-
-	if ((group->rdc_thrnum >= 1) && (group->seqack == RDC_NEWSEQ)) {
-		/*
-		 * We also need to check if we are starting a new
-		 * sequence, and if so don't create a new thread,
-		 * as we must ensure that the start of new sequence
-		 * requests arrives first to re-init the server.
-		 */
-		mutex_exit(qlock);
-		return (0);
-	}
-	/*
-	 * For version 6,
-	 * see if we can fit in another thread.
-	 */
-	group->rdc_thrnum++;
-
-	if (krdc->intf && (krdc->intf->rpc_version >= RDC_VERSION6)) {
-		rdc_u_info_t *urdc = &rdc_u_info[index];
-		if (group->rdc_thrnum >= urdc->asyncthr)
-			group->rdc_writer = 1;
-	} else {
-		group->rdc_writer = 1;
-	}
-
-	mutex_exit(qlock);
-
-
-	/*
-	 * If we got here, we know that we have not exceeded the allowed
-	 * number of async threads for our group.  If we run out of threads
-	 * in _rdc_flset, we add a new thread to the set.
-	 */
-	tries = 0;
-	do {
-		/* first try to grab a thread from the free list */
-		if (t = nst_create(_rdc_flset, rdc_flusher_thread,
-		    (blind_t)(unsigned long)index, 0)) {
-			break;
-		}
-
-		/* that failed; add a thread to the set and try again */
-		if (nst_add_thread(_rdc_flset, 1) != 1) {
-			cmn_err(CE_WARN, "!rdc_writer index %d nst_add_thread "
-			    "error, tries: %d", index, tries);
-			break;
-		}
-	} while (++tries < MAX_TRIES);
-
-	if (tries) {
-		mutex_enter(&group->addthrnumlk);
-		group->rdc_addthrnum += tries;
-		mutex_exit(&group->addthrnumlk);
-	}
-
-	if (t) {
-		return (1);
-	}
-
-	cmn_err(CE_WARN, "!rdc_writer: index %d nst_create error", index);
-	rdc_many_enter(krdc);
-	mutex_enter(qlock);
-	group->rdc_thrnum--;
-	group->rdc_writer = 0;
-	if ((group->count == 0) && (group->rdc_thrnum == 0)) {
-		mutex_exit(qlock);
-		/*
-		 * Race with remove_from_group while write thread was
-		 * failing to be created.
-		 */
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_writer: group being destroyed");
-#endif
-		rdc_delgroup(group);
-		krdc->group = NULL;
-		rdc_many_exit(krdc);
-		return (-1);
-	}
-	mutex_exit(qlock);
-	rdc_many_exit(krdc);
-	return (-1);
-}
-
-/*
- * Either we need to flush the
- * kmem (net_queue) queue or the disk (disk_queue)
- * determine which, and do it.
- */
-void
-rdc_flusher_thread(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-
-	if (krdc->group->flags & RDC_MEMQUE) {
-		rdc_flush_memq(index);
-		return;
-	} else if (krdc->group->flags & RDC_DISKQUE) {
-		rdc_flush_diskq(index);
-		return;
-	} else { /* uh-oh, big time */
-		cmn_err(CE_PANIC, "flusher trying to flush unknown queue type");
-	}
-
-}
-
-void
-rdc_flush_memq(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	rdc_aio_t *aio;
-	net_queue *q;
-	int dowork;
-	rdc_group_t *group = krdc->group;
-	if (!group || group->count == 0) {
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_memq: no group left!");
-#endif
-		return;
-	}
-
-	if (!krdc->c_fd) {
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_memq: no c_fd!");
-#endif
-		goto thread_death;
-	}
-
-#ifdef DEBUG_DISABLE
-	if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
-		cmn_err(CE_WARN, "!rdc_flush_memq: DISABLE PENDING!");
-		/*
-		 * Need to continue as we may be trying to flush IO
-		 * while trying to disable or suspend
-		 */
-	}
-#endif
-
-	q = &group->ra_queue;
-
-	dowork = 1;
-	/* CONSTCOND */
-	while (dowork) {
-		if (net_exit == ATM_EXIT)
-			break;
-
-		group = krdc->group;
-		if (!group || group->count == 0) {
-#ifdef DEBUG
-			cmn_err(CE_WARN, "!rdc_flush_memq: no group left!");
-#endif
-			break;
-		}
-
-		mutex_enter(&q->net_qlock);
-		aio = q->net_qhead;
-
-		if (aio == NULL) {
-#ifdef DEBUG
-			if (q->nitems != 0 ||
-			    q->blocks != 0 ||
-			    q->net_qtail != 0) {
-				cmn_err(CE_PANIC,
-				    "rdc_flush_memq(1): q %p, q blocks %"
-				    NSC_SZFMT ", nitems %" NSC_SZFMT
-				    ", qhead %p qtail %p",
-				    (void *) q, q->blocks, q->nitems,
-				    (void *) aio, (void *) q->net_qtail);
-			}
-#endif
-			mutex_exit(&q->net_qlock);
-			break;
-		}
-
-		/* aio remove from q */
-
-		q->net_qhead = aio->next;
-		aio->next = NULL;
-
-		if (q->net_qtail == aio)
-			q->net_qtail = q->net_qhead;
-
-		q->blocks -= aio->len;
-		q->nitems--;
-
-		/*
-		 * in flight numbers.
-		 */
-		q->inflbls += aio->len;
-		q->inflitems++;
-
-#ifdef DEBUG
-		if (q->net_qhead == NULL) {
-			if (q->nitems != 0 ||
-			    q->blocks != 0 ||
-			    q->net_qtail != 0) {
-				cmn_err(CE_PANIC,
-				    "rdc_flush_memq(2): q %p, q blocks %"
-				    NSC_SZFMT ", nitems %" NSC_SZFMT
-				    ", qhead %p qtail %p",
-				    (void *) q, q->blocks, q->nitems,
-				    (void *) q->net_qhead,
-				    (void *) q->net_qtail);
-			}
-		}
-
-#ifndef NSC_MULTI_TERABYTE
-		if (q->blocks < 0) {
-			cmn_err(CE_PANIC,
-			    "rdc_flush_memq(3): q %p, q blocks %" NSC_SZFMT
-			    ", nitems %d, qhead %p, qtail %p",
-			    (void *) q, q->blocks, q->nitems,
-			    (void *) q->net_qhead, (void *) q->net_qtail);
-		}
-#else
-		/* blocks and nitems are unsigned for NSC_MULTI_TERABYTE */
-#endif
-#endif
-
-		mutex_exit(&q->net_qlock);
-
-		aio->iostatus = RDC_IO_INIT;
-
-		_rdc_remote_flush(aio);
-
-		mutex_enter(&q->net_qlock);
-		q->inflbls -= aio->len;
-		q->inflitems--;
-		if ((group->seqack == RDC_NEWSEQ) &&
-		    (group->seq != RDC_NEWSEQ + 1)) {
-			if ((q->net_qhead == NULL) ||
-			    (q->net_qhead->seq != RDC_NEWSEQ + 1)) {
-				/*
-				 * We are an old thread, and the
-				 * queue sequence has been reset
-				 * during the network write above.
-				 * As such we mustn't pull another
-				 * job from the queue until the
-				 * first sequence message has been ack'ed.
-				 * Just die instead. Unless this thread
-				 * is the first sequence that has just
-				 * been ack'ed
-				 */
-				dowork = 0;
-			}
-		}
-		mutex_exit(&q->net_qlock);
-
-		if ((aio->iostatus != RDC_IO_DONE) && (group->count)) {
-			rdc_k_info_t *krdctmp = &rdc_k_info[aio->index];
-			if (krdctmp->type_flag & RDC_DISABLEPEND) {
-				kmem_free(aio, sizeof (*aio));
-				goto thread_death;
-			}
-			rdc_group_enter(krdc);
-			ASSERT(krdc->group);
-			rdc_group_log(krdc, RDC_NOFLUSH | RDC_ALLREMOTE,
-			    "memq flush aio status not RDC_IO_DONE");
-			rdc_group_exit(krdc);
-			rdc_dump_queue(aio->index);
-		}
-		kmem_free(aio, sizeof (*aio));
-
-		if (krdc->remote_index < 0 || !krdc->lsrv || !krdc->intf)
-			break;
-	}
-
-thread_death:
-	rdc_many_enter(krdc);
-	mutex_enter(&group->ra_queue.net_qlock);
-	group->rdc_thrnum--;
-	group->rdc_writer = 0;
-	/*
-	 * all threads must be dead.
-	 */
-	if ((group->count == 0) && (group->rdc_thrnum == 0)) {
-		mutex_exit(&group->ra_queue.net_qlock);
-		/*
-		 * Group now empty, so destroy
-		 * Race with remove_from_group while write thread was running
-		 */
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_memq: group being destroyed");
-#endif
-		rdc_delgroup(group);
-		krdc->group = NULL;
-		rdc_many_exit(krdc);
-		return;
-	}
-	mutex_exit(&group->ra_queue.net_qlock);
-	rdc_many_exit(krdc);
-}
-
-/*
- * rdc_flush_diskq
- * disk queue flusher
- */
-void
-rdc_flush_diskq(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	rdc_u_info_t *urdc = &rdc_u_info[index];
-	rdc_aio_t *aio = NULL;
-	disk_queue *q;
-	net_queue *nq;
-	int dowork;
-	int rc;
-	rdc_group_t *group = krdc->group;
-
-	if (!group || group->count == 0) {
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_diskq: no group left!");
-#endif
-		return;
-	}
-
-	if (!krdc->c_fd) {
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_diskq: no c_fd!");
-#endif
-		return;
-	}
-
-#ifdef DEBUG_DISABLE
-	if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
-		cmn_err(CE_WARN, "!rdc_flush_diskq: DISABLE PENDING!");
-		/*
-		 * Need to continue as we may be trying to flush IO
-		 * while trying to disable or suspend
-		 */
-	}
-#endif
-	q = &group->diskq;
-	nq = &group->ra_queue;
-
-	if (IS_QSTATE(q, RDC_QDISABLEPEND) || IS_STATE(urdc, RDC_LOGGING)) {
-#ifdef DEBUG
-		cmn_err(CE_NOTE, "!flusher thread death 1 %x", QSTATE(q));
-#endif
-		goto thread_death;
-	}
-
-	dowork = 1;
-	/* CONSTCOND */
-	while (dowork) {
-		if (net_exit == ATM_EXIT)
-			break;
-
-		group = krdc->group;
-		if (!group || group->count == 0) {
-#ifdef DEBUG
-			cmn_err(CE_WARN, "!rdc_flush_diskq: no group left!");
-#endif
-			break;
-		}
-
-		do {
-			rc = 0;
-			if ((IS_STATE(urdc, RDC_LOGGING)) ||
-			    (IS_STATE(urdc, RDC_SYNCING)) ||
-			    (nq->qfflags & RDC_QFILLSLEEP))
-				goto thread_death;
-
-			aio = rdc_dequeue(krdc, &rc);
-
-			if ((IS_STATE(urdc, RDC_LOGGING)) ||
-			    (IS_STATE(urdc, RDC_SYNCING)) ||
-			    (nq->qfflags & RDC_QFILLSLEEP)) {
-				goto thread_death;
-			}
-			if (rc == EAGAIN) {
-				delay(40);
-			}
-
-		} while (rc == EAGAIN);
-
-		if (aio == NULL) {
-			break;
-		}
-
-		aio->iostatus = RDC_IO_INIT;
-
-		mutex_enter(QLOCK(q));
-		q->inflbls += aio->len;
-		q->inflitems++;
-		mutex_exit(QLOCK(q));
-
-		_rdc_remote_flush(aio);
-
-		mutex_enter(QLOCK(q));
-		q->inflbls -= aio->len;
-		q->inflitems--;
-
-		if ((group->seqack == RDC_NEWSEQ) &&
-		    (group->seq != RDC_NEWSEQ + 1)) {
-			if ((nq->net_qhead == NULL) ||
-			    (nq->net_qhead->seq != RDC_NEWSEQ + 1)) {
-				/*
-				 * We are an old thread, and the
-				 * queue sequence has been reset
-				 * during the network write above.
-				 * As such we mustn't pull another
-				 * job from the queue until the
-				 * first sequence message has been ack'ed.
-				 * Just die instead. Unless of course,
-				 * this thread is the first sequence that
-				 * has just been ack'ed.
-				 */
-				dowork = 0;
-			}
-		}
-		mutex_exit(QLOCK(q));
-
-		if (aio->iostatus == RDC_IO_CANCELLED) {
-			rdc_dump_queue(aio->index);
-			kmem_free(aio, sizeof (*aio));
-			aio = NULL;
-			if (group) { /* seq gets bumped on dequeue */
-				mutex_enter(QLOCK(q));
-				rdc_dump_iohdrs(q);
-				SET_QNXTIO(q, QHEAD(q));
-				SET_QCOALBOUNDS(q, QHEAD(q));
-				group->seq = RDC_NEWSEQ;
-				group->seqack = RDC_NEWSEQ;
-				mutex_exit(QLOCK(q));
-			}
-			break;
-		}
-
-		if ((aio->iostatus != RDC_IO_DONE) && (group->count)) {
-			rdc_k_info_t *krdctmp = &rdc_k_info[aio->index];
-			if (krdctmp->type_flag & RDC_DISABLEPEND) {
-				kmem_free(aio, sizeof (*aio));
-				aio = NULL;
-				goto thread_death;
-			}
-			rdc_group_enter(krdc);
-			rdc_group_log(krdc,
-			    RDC_NOFLUSH | RDC_ALLREMOTE | RDC_QUEUING,
-			    "diskq flush aio status not RDC_IO_DONE");
-			rdc_group_exit(krdc);
-			rdc_dump_queue(aio->index);
-		}
-
-		kmem_free(aio, sizeof (*aio));
-		aio = NULL;
-
-#ifdef DEBUG_DISABLE
-		if (krdc->type_flag & RDC_DISABLEPEND) {
-			cmn_err(CE_WARN,
-			    "!rdc_flush_diskq: DISABLE PENDING after IO!");
-		}
-#endif
-		if (krdc->remote_index < 0 || !krdc->lsrv || !krdc->intf)
-			break;
-
-		if (IS_QSTATE(q, RDC_QDISABLEPEND)) {
-#ifdef DEBUG
-			cmn_err(CE_NOTE, "!flusher thread death 2");
-#endif
-			break;
-		}
-	}
-thread_death:
-	rdc_many_enter(krdc);
-	mutex_enter(QLOCK(q));
-	group->rdc_thrnum--;
-	group->rdc_writer = 0;
-
-	if (aio && aio->qhandle) {
-		aio->qhandle->sb_user--;
-		if (aio->qhandle->sb_user == 0) {
-			(void) _rdc_rsrv_diskq(krdc->group);
-			rdc_fixlen(aio);
-			(void) nsc_free_buf(aio->qhandle);
-			aio->qhandle = NULL;
-			aio->handle = NULL;
-			_rdc_rlse_diskq(krdc->group);
-		}
-	}
-	if ((group->count == 0) && (group->rdc_thrnum == 0)) {
-		mutex_exit(QLOCK(q));
-		/*
-		 * Group now empty, so destroy
-		 * Race with remove_from_group while write thread was running
-		 */
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!rdc_flush_diskq: group being destroyed");
-#endif
-		mutex_enter(&group->diskqmutex);
-		rdc_close_diskq(group);
-		mutex_exit(&group->diskqmutex);
-		rdc_delgroup(group);
-		krdc->group = NULL;
-		rdc_many_exit(krdc);
-		return;
-	}
-	mutex_exit(QLOCK(q));
-	rdc_many_exit(krdc);
-}
-
-/*
- * _rdc_remote_flush
- * Flush a single block ANON block
- * this function will flush from either the disk queue
- * or the memory queue. The appropriate locks must be
- * taken out etc, etc ...
- */
-static void
-_rdc_remote_flush(rdc_aio_t *aio)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[aio->index];
-	rdc_u_info_t *urdc = &rdc_u_info[aio->index];
-	disk_queue *q = &krdc->group->diskq;
-	kmutex_t *qlock;
-	rdc_group_t *group;
-	nsc_buf_t *h = NULL;
-	int reserved = 0;
-	int rtype = RDC_RAW;
-	int rc;
-	uint_t maxseq;
-	struct netwriteres netret;
-	int waitq = 1;
-	int vflags;
-
-	group = krdc->group;
-	netret.vecdata.vecdata_val = NULL;
-	netret.vecdata.vecdata_len = 0;
-
-	/* Where did we get this aio from anyway? */
-	if (RDC_IS_DISKQ(group)) {
-		qlock = &group->diskq.disk_qlock;
-	} else {
-		qlock = &group->ra_queue.net_qlock;
-	}
-
-	/*
-	 * quench transmission if we are too far ahead of the
-	 * server Q, or it will overflow.
-	 * Must fail all requests while asyncdis is set.
-	 * It will be cleared when the last thread to be discarded
-	 * sets the asyncstall counter to zero.
-	 * Note the thread within rdc_net_write
-	 * also bumps the asyncstall counter.
-	 */
-
-	mutex_enter(qlock);
-	if (group->asyncdis) {
-		aio->iostatus = RDC_IO_CANCELLED;
-		mutex_exit(qlock);
-		goto failed;
-	}
-	/* don't go to sleep if we have gone logging! */
-	vflags = rdc_get_vflags(urdc);
-	if ((vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
-		if ((vflags & RDC_LOGGING) && RDC_IS_DISKQ(group))
-			aio->iostatus = RDC_IO_CANCELLED;
-
-		mutex_exit(qlock);
-		goto failed;
-	}
-
-	while (maxseq = group->seqack + RDC_MAXPENDQ + 1,
-	    maxseq = (maxseq < group->seqack) ? maxseq + RDC_NEWSEQ + 1
-	    : maxseq, !RDC_INFRONT(aio->seq, maxseq)) {
-		group->asyncstall++;
-		ASSERT(!IS_STATE(urdc, RDC_LOGGING));
-		cv_wait(&group->asyncqcv, qlock);
-		group->asyncstall--;
-		ASSERT(group->asyncstall >= 0);
-		if (group->asyncdis) {
-			if (group->asyncstall == 0) {
-				group->asyncdis = 0;
-			}
-			aio->iostatus = RDC_IO_CANCELLED;
-			mutex_exit(qlock);
-			goto failed;
-		}
-		/*
-		 * See if we have gone into logging mode
-		 * since sleeping.
-		 */
-		vflags = rdc_get_vflags(urdc);
-		if (vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING)) {
-			if ((vflags & RDC_LOGGING) && RDC_IS_DISKQ(group))
-				aio->iostatus = RDC_IO_CANCELLED;
-
-			mutex_exit(qlock);
-			goto failed;
-		}
-	}
-	mutex_exit(qlock);
-
-	if ((krdc->io_kstats) && (!RDC_IS_DISKQ(krdc->group))) {
-		mutex_enter(krdc->io_kstats->ks_lock);
-		kstat_waitq_exit(KSTAT_IO_PTR(krdc->io_kstats));
-		mutex_exit(krdc->io_kstats->ks_lock);
-		waitq = 0;
-	}
-
-
-	rc = _rdc_rsrv_devs(krdc, rtype, RDC_INTERNAL);
-	if (rc != 0) {
-#ifdef DEBUG
-		cmn_err(CE_WARN, "!_rdc_remote_flush: reserve, index %d, rc %d",
-		    aio->index, rc);
-#endif
-		goto failed;
-	}
-
-	reserved = 1;
-	/*
-	 * Case where we are multihop and calling with no ANON bufs
-	 * Need to do the read to fill the buf.
-	 */
-	if (!aio->handle) {
-		rc = nsc_alloc_buf(RDC_U_FD(krdc), aio->pos, aio->len,
-		    (aio->flag & ~NSC_WRITE) | NSC_READ, &h);
-		if (!RDC_SUCCESS(rc)) {
-#ifdef DEBUG
-			cmn_err(CE_WARN,
-			    "!_rdc_remote_flush: alloc_buf, index %d, pos %"
-			    NSC_SZFMT ", len %" NSC_SZFMT ", rc %d",
-			    aio->index, aio->pos, aio->len, rc);
-#endif
-
-			goto failed;
-		}
-		aio->handle = h;
-		aio->handle->sb_user = RDC_NULLBUFREAD;
-	}
-
-	mutex_enter(qlock);
-	if (group->asyncdis) {
-		if (group->asyncstall == 0) {
-			group->asyncdis = 0;
-		}
-		aio->iostatus = RDC_IO_CANCELLED;
-		mutex_exit(qlock);
-		goto failed;
-	}
-	group->asyncstall++;
-	mutex_exit(qlock);
-
-
-	if (krdc->remote_index < 0) {
-		/*
-		 * this should be ok, we are flushing, not rev syncing.
-		 * remote_index could be -1 if we lost a race with
-		 * resume and the flusher trys to flush an io from
-		 * another set that has not resumed
-		 */
-		krdc->remote_index = rdc_net_state(krdc->index, CCIO_SLAVE);
-		DTRACE_PROBE1(remote_index_negative, int, krdc->remote_index);
-
-	}
-
-	/*
-	 * double check for logging, no check in net_write()
-	 * skip the write if you can, otherwise, if logging
-	 * avoid clearing the bit .. you don't know whose bit it may
-	 * also be.
-	 */
-	if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING)) {
-		aio->iostatus = RDC_IO_CANCELLED;
-		mutex_enter(qlock);
-		group->asyncstall--;
-		mutex_exit(qlock);
-		goto failed;
-	}
-
-	rc = rdc_net_write(krdc->index, krdc->remote_index,
-	    aio->handle, aio->pos, aio->len, aio->seq, aio->qpos, &netret);
-
-	mutex_enter(qlock);
-	group->asyncstall--;
-	if (group->asyncdis) {
-		if (group->asyncstall == 0) {
-			group->asyncdis = 0;
-		}
-		aio->iostatus = RDC_IO_CANCELLED;
-		mutex_exit(qlock);
-		goto failed;
-	}
-
-	if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING)) {
-		mutex_exit(qlock);
-		aio->iostatus = RDC_IO_CANCELLED;
-		goto failed;
-	}
-
-	ASSERT(aio->handle);
-	if (rc != 0) {
-#ifdef DEBUG
-		cmn_err(CE_WARN,
-		    "!_rdc_remote_flush: write, index %d, pos %" NSC_SZFMT
-		    ", len %" NSC_SZFMT ", "
-		    "rc %d seq %u group seq %u seqack %u qpos %" NSC_SZFMT,
-		    aio->index, aio->pos, aio->len, rc, aio->seq,
-		    group->seq, group->seqack, aio->qpos);
-#endif
-		if (rc == ENOLINK) {
-			cmn_err(CE_WARN,
-			    "!Hard timeout detected (%d sec) "
-			    "on SNDR set %s:%s",
-			    rdc_rpc_tmout, urdc->secondary.intf,
-			    urdc->secondary.file);
-		}
-		mutex_exit(qlock);
-		goto failed;
-	} else {
-		aio->iostatus = RDC_IO_DONE;
-	}
-
-	if (RDC_IS_DISKQ(group)) {
-		/* free locally alloc'd handle */
-		if (aio->handle->sb_user == RDC_NULLBUFREAD) {
-			(void) nsc_free_buf(aio->handle);
-			aio->handle = NULL;
-		}
-		aio->qhandle->sb_user--;
-		if (aio->qhandle->sb_user == 0) {
-			(void) _rdc_rsrv_diskq(group);
-			rdc_fixlen(aio);
-			(void) nsc_free_buf(aio->qhandle);
-			aio->qhandle = NULL;
-			aio->handle = NULL;
-			_rdc_rlse_diskq(group);
-		}
-
-	} else {
-		(void) nsc_free_buf(aio->handle);
-		aio->handle = NULL;
-	}
-
-	mutex_exit(qlock);
-
-	_rdc_rlse_devs(krdc, rtype);
-
-	if (netret.result == 0) {
-		vflags = rdc_get_vflags(urdc);
-
-		if (!(vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
-			RDC_CLR_BITMAP(krdc, aio->pos, aio->len, \
-			    0xffffffff, RDC_BIT_BUMP);
-
-			if (RDC_IS_DISKQ(krdc->group)) {
-				if (!IS_STATE(urdc, RDC_LOGGING)) {
-					/* tell queue data has been flushed */
-					rdc_clr_iohdr(krdc, aio->qpos);
-				} else { /* throw away queue, logging */
-					mutex_enter(qlock);
-					rdc_dump_iohdrs(q);
-					SET_QNXTIO(q, QHEAD(q));
-					SET_QCOALBOUNDS(q, QHEAD(q));
-					mutex_exit(qlock);
-				}
-			}
-		}
-
-		mutex_enter(qlock);
-		/*
-		 * Check to see if the reply has arrived out of
-		 * order, if so don't update seqack.
-		 */
-		if (!RDC_INFRONT(aio->seq, group->seqack)) {
-			group->seqack = aio->seq;
-		}
-#ifdef DEBUG
-		else {
-			rdc_ooreply++;
-		}
-#endif
-		if (group->asyncstall) {
-			cv_broadcast(&group->asyncqcv);
-		}
-		mutex_exit(qlock);
-	} else if (netret.result < 0) {
-		aio->iostatus = RDC_IO_FAILED;
-	}
-
-	/*
-	 * see if we have any pending async requests we can mark
-	 * as done.
-	 */
-
-	if (netret.vecdata.vecdata_len) {
-		net_pendvec_t *vecp;
-		net_pendvec_t *vecpe;
-		vecp = netret.vecdata.vecdata_val;
-		vecpe = netret.vecdata.vecdata_val + netret.vecdata.vecdata_len;
-		while (vecp < vecpe) {
-			rdc_k_info_t *krdcp = &rdc_k_info[vecp->pindex];
-			rdc_u_info_t *urdcp = &rdc_u_info[vecp->pindex];
-			/*
-			 * we must always still be in the same group.
-			 */
-			ASSERT(krdcp->group == group);
-			vflags = rdc_get_vflags(urdcp);
-
-			if (!(vflags &
-			    (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
-				RDC_CLR_BITMAP(krdcp, vecp->apos, vecp->alen, \
-				    0xffffffff, RDC_BIT_BUMP);
-				if (RDC_IS_DISKQ(krdcp->group)) {
-					if (!IS_STATE(urdc, RDC_LOGGING)) {
-						/* update queue info */
-						rdc_clr_iohdr(krdc, vecp->qpos);
-					} else { /* we've gone logging */
-						mutex_enter(qlock);
-						rdc_dump_iohdrs(q);
-						SET_QNXTIO(q, QHEAD(q));
-						SET_QCOALBOUNDS(q, QHEAD(q));
-						mutex_exit(qlock);
-					}
-				}
-			}
-
-			/*
-			 * see if we can re-start transmission
-			 */
-			mutex_enter(qlock);
-			if (!RDC_INFRONT(vecp->seq, group->seqack)) {
-				group->seqack = vecp->seq;
-			}
-#ifdef DEBUG
-			else {
-				rdc_ooreply++;
-			}
-#endif
-			DTRACE_PROBE1(pendvec_return, int, vecp->seq);
-
-			if (group->asyncstall) {
-				cv_broadcast(&group->asyncqcv);
-			}
-			mutex_exit(qlock);
-			vecp++;
-		}
-	}
-	if (netret.vecdata.vecdata_val)
-		kmem_free(netret.vecdata.vecdata_val,
-		    netret.vecdata.vecdata_len * sizeof (net_pendvec_t));
-	return;
-failed:
-
-	/* perhaps we have a few threads stuck .. */
-	if (group->asyncstall) {
-		group->asyncdis = 1;
-		cv_broadcast(&group->asyncqcv);
-	}
-	if (netret.vecdata.vecdata_val)
-		kmem_free(netret.vecdata.vecdata_val,
-		    netret.vecdata.vecdata_len * sizeof (net_pendvec_t));
-
-	mutex_enter(qlock);
-	if (RDC_IS_DISKQ(group)) {
-		/* free locally alloc'd hanlde */
-		if ((aio->handle) &&
-		    (aio->handle->sb_user == RDC_NULLBUFREAD)) {
-			(void) nsc_free_buf(aio->handle);
-			aio->handle = NULL;
-		}
-		aio->qhandle->sb_user--;
-		if (aio->qhandle->sb_user == 0) {
-			(void) _rdc_rsrv_diskq(group);
-			rdc_fixlen(aio);
-			(void) nsc_free_buf(aio->qhandle);
-			aio->qhandle = NULL;
-			aio->handle = NULL;
-			_rdc_rlse_diskq(group);
-		}
-	} else {
-		if (aio->handle) {
-			(void) nsc_free_buf(aio->handle);
-			aio->handle = NULL;
-		}
-	}
-	mutex_exit(qlock);
-
-	if (reserved) {
-		_rdc_rlse_devs(krdc, rtype);
-	}
-
-	if ((waitq && krdc->io_kstats) && (!RDC_IS_DISKQ(krdc->group))) {
-		mutex_enter(krdc->io_kstats->ks_lock);
-		kstat_waitq_exit(KSTAT_IO_PTR(krdc->io_kstats));
-		mutex_exit(krdc->io_kstats->ks_lock);
-	}
-
-	/* make sure that the bit is still set */
-	RDC_CHECK_BIT(krdc, aio->pos, aio->len);
-
-	if (aio->iostatus != RDC_IO_CANCELLED)
-		aio->iostatus = RDC_IO_FAILED;
-}
-
-
-/*
- * rdc_drain_disk_queue
- * drain the async network queue for the whole group. Bail out if nothing
- * happens in 20 sec
- * returns -1 if it bails before the queues are drained.
- */
-#define	NUM_RETRIES	15	/* Number of retries to wait if no progress */
-int
-rdc_drain_disk_queue(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	volatile rdc_group_t *group;
-	volatile disk_queue *diskq;
-	int threads, counter;
-	long blocks;
-
-	/* Sanity checking */
-	if (index > rdc_max_sets)
-		return (0);
-
-	/*
-	 * If there is no group or diskq configured, we can leave now
-	 */
-	if (!(group = krdc->group) || !(diskq = &group->diskq))
-		return (0);
-
-	/*
-	 * No need to wait if EMPTY and threads are gone
-	 */
-	counter = 0;
-	while (!QEMPTY(diskq) || group->rdc_thrnum) {
-
-		/*
-		 * Capture counters to determine if progress is being made
-		 */
-		blocks = QBLOCKS(diskq);
-		threads = group->rdc_thrnum;
-
-		/*
-		 * Wait
-		 */
-		delay(HZ);
-
-		/*
-		 * Has the group or disk queue gone away while delayed?
-		 */
-		if (!(group = krdc->group) || !(diskq = &group->diskq))
-			return (0);
-
-		/*
-		 * Are we still seeing progress?
-		 */
-		if (blocks == QBLOCKS(diskq) && threads == group->rdc_thrnum) {
-			/*
-			 * No progress seen, increment retry counter
-			 */
-			if (counter++ > NUM_RETRIES) {
-				return (-1);
-			}
-		} else {
-			/*
-			 * Reset counter, as we've made progress
-			 */
-			counter = 0;
-		}
-	}
-
-	return (0);
-}
-
-/*
- * decide what needs to be drained, disk or core
- * and drain it
- */
-int
-rdc_drain_queue(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	rdc_group_t *group = krdc->group;
-
-	if (!group)
-		return (0);
-
-	if (RDC_IS_DISKQ(group))
-		return (rdc_drain_disk_queue(index));
-	if (RDC_IS_MEMQ(group))
-		return (rdc_drain_net_queue(index));
-	/* oops.. */
-#ifdef DEBUG
-	cmn_err(CE_WARN, "!rdc_drain_queue: "
-	    "attempting drain of unknown Q type");
-#endif
-	return (0);
-}
-
-/*
- * rdc_drain_net_queue
- * drain the async network queue for the whole group. Bail out if nothing
- * happens in 20 sec
- * returns -1 if it bails before the queues are drained.
- */
-int
-rdc_drain_net_queue(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	volatile net_queue *q;
-	int bail = 20;	/* bail out in about 20 secs */
-	nsc_size_t blocks;
-
-	/* Sanity checking */
-	if (index > rdc_max_sets)
-		return (0);
-	if (!krdc->group)
-		return (0);
-	/* LINTED */
-	if (!(q = &krdc->group->ra_queue))
-		return (0);
-
-	/* CONSTCOND */
-	while (1) {
-
-		if (((volatile rdc_aio_t *)q->net_qhead == NULL) &&
-		    (krdc->group->rdc_thrnum == 0)) {
-			break;
-		}
-
-		blocks = q->blocks;
-
-		q = (volatile net_queue *)&krdc->group->ra_queue;
-
-		if ((blocks == q->blocks) &&
-		    (--bail <= 0)) {
-			break;
-		}
-
-		delay(HZ);
-	}
-
-	if (bail <= 0)
-		return (-1);
-
-	return (0);
-}
-
-/*
- * rdc_dump_queue
- * We want to release all the blocks currently on the network flushing queue
- * We already have them logged in the bitmap.
- */
-void
-rdc_dump_queue(int index)
-{
-	rdc_k_info_t *krdc = &rdc_k_info[index];
-	rdc_aio_t *aio;
-	net_queue *q;
-	rdc_group_t *group;
-	disk_queue *dq;
-	kmutex_t *qlock;
-
-	group = krdc->group;
-
-	q = &group->ra_queue;
-	dq = &group->diskq;
-
-	/*
-	 * gotta have both locks here for diskq
-	 */
-
-	if (RDC_IS_DISKQ(group)) {
-		mutex_enter(&q->net_qlock);
-		if (q->qfill_sleeping == RDC_QFILL_AWAKE) {
-			int tries = 3;
-#ifdef DEBUG_DISKQ
-			cmn_err(CE_NOTE,
-			    "!dumpq sending diskq->memq flusher to sleep");
-#endif
-			q->qfflags |= RDC_QFILLSLEEP;
-			mutex_exit(&q->net_qlock);
-			while (q->qfill_sleeping == RDC_QFILL_AWAKE && tries--)
-				delay(5);
-			mutex_enter(&q->net_qlock);
-		}
-	}
-
-	if (RDC_IS_DISKQ(group)) {
-		qlock = &dq->disk_qlock;
-		(void) _rdc_rsrv_diskq(group);
-	} else {
-		qlock = &q->net_qlock;
-	}
-
-	mutex_enter(qlock);
-
-	group->seq = RDC_NEWSEQ;	/* reset the sequence number */
-	group->seqack = RDC_NEWSEQ;
-
-	/* if the q is on disk, dump the q->iohdr chain */
-	if (RDC_IS_DISKQ(group)) {
-		rdc_dump_iohdrs(dq);
-
-		/* back up the nxtio pointer */
-		SET_QNXTIO(dq, QHEAD(dq));
-		SET_QCOALBOUNDS(dq, QHEAD(dq));
-	}
-
-	while (q->net_qhead) {
-		rdc_k_info_t *tmpkrdc;
-		aio = q->net_qhead;
-		tmpkrdc = &rdc_k_info[aio->index];
-
-		if (RDC_IS_DISKQ(group)) {
-			aio->qhandle->sb_user--;
-			if (aio->qhandle->sb_user == 0) {
-				rdc_fixlen(aio);
-				(void) nsc_free_buf(aio->qhandle);
-				aio->qhandle = NULL;
-				aio->handle = NULL;
-			}
-		} else {
-			if (aio->handle) {
-				(void) nsc_free_buf(aio->handle);
-				aio->handle = NULL;
-			}
-		}
-
-		q->net_qhead = aio->next;
-		RDC_CHECK_BIT(tmpkrdc, aio->pos, aio->len);
-
-		kmem_free(aio, sizeof (*aio));
-		if (tmpkrdc->io_kstats && !RDC_IS_DISKQ(group)) {
-			mutex_enter(tmpkrdc->io_kstats->ks_lock);
-			kstat_waitq_exit(KSTAT_IO_PTR(tmpkrdc->io_kstats));
-			mutex_exit(tmpkrdc->io_kstats->ks_lock);
-		}
-
-	}
-
-	q->net_qtail = NULL;
-	q->blocks = 0;
-	q->nitems = 0;
-
-	/*
-	 * See if we have stalled threads.
-	 */
-done:
-	if (group->asyncstall) {
-		group->asyncdis = 1;
-		cv_broadcast(&group->asyncqcv);
-	}
-	mutex_exit(qlock);
-	if (RDC_IS_DISKQ(group)) {
-		mutex_exit(&q->net_qlock);
-		_rdc_rlse_diskq(group);
-	}
-
-}
-
-
-/*
- * rdc_clnt_get
- * Get a CLIENT handle and cache it
- */
-
-static int
-rdc_clnt_get(rdc_srv_t *svp, rpcvers_t vers, struct chtab **rch, CLIENT **clp)
-{
-	uint_t	max_msgsize;
-	int	retries;
-	int ret;
-	struct cred		*cred;
-	int num_clnts = 0;
-	register struct chtab *ch;
-	struct chtab **plistp;
-	CLIENT *client = 0;
-
-	if (rch) {
-		*rch = 0;
-	}
-
-	if (clp) {
-		*clp = 0;
-	}
-
-	retries = 6;	/* Never used for COTS in Solaris */
-	cred = ddi_get_cred();
-	max_msgsize = RDC_RPC_MAX;
-
-	mutex_enter(&rdc_clnt_lock);
-
-	ch = rdc_chtable;
-	plistp = &rdc_chtable;
-
-	/* find the right ch_list chain */
-
-	for (ch = rdc_chtable; ch != NULL; ch = ch->ch_next) {
-		if (ch->ch_prog == RDC_PROGRAM &&
-		    ch->ch_vers == vers &&
-		    ch->ch_dev == svp->ri_knconf->knc_rdev &&
-		    ch->ch_protofmly != NULL &&
-		    strcmp(ch->ch_protofmly,
-		    svp->ri_knconf->knc_protofmly) == 0) {
-			/* found the correct chain to walk */
-			break;
-		}
-		plistp = &ch->ch_next;
-	}
-
-	if (ch != NULL) {
-		/* walk the ch_list and try and find a free client */
-
-		for (num_clnts = 0; ch != NULL; ch = ch->ch_list, num_clnts++) {
-			if (ch->ch_inuse == FALSE) {
-				/* suitable handle to reuse */
-				break;
-			}
-			plistp = &ch->ch_list;
-		}
-	}
-
-	if (ch == NULL && num_clnts >= MAXCLIENTS) {
-		/* alloc a temporary handle and return */
-
-		rdc_clnt_toomany++;
-		mutex_exit(&rdc_clnt_lock);
-
-		ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
-		    RDC_PROGRAM, vers, max_msgsize, retries, cred, &client);
-
-		if (ret != 0) {
-			cmn_err(CE_NOTE,
-			    "!rdc_call: tli_kcreate failed %d", ret);
-			return (ret);
-		}
-
-		*rch = 0;
-		*clp = client;
-		(void) CLNT_CONTROL(client, CLSET_PROGRESS, NULL);
-		return (ret);
-	}
-
-	if (ch != NULL) {
-		/* reuse a cached handle */
-
-		ch->ch_inuse = TRUE;
-		ch->ch_timesused++;
-		mutex_exit(&rdc_clnt_lock);
-
-		*rch = ch;
-
-		if (ch->ch_client == NULL) {
-			ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
-			    RDC_PROGRAM, vers, max_msgsize, retries,
-			    cred, &ch->ch_client);
-			if (ret != 0) {
-				ch->ch_inuse = FALSE;
-				return (ret);
-			}
-
-			(void) CLNT_CONTROL(ch->ch_client, CLSET_PROGRESS,
-			    NULL);
-			*clp = ch->ch_client;
-
-			return (0);
-		} else {
-		/*
-		 * Consecutive calls to CLNT_CALL() on the same client handle
-		 * get the same transaction ID.  We want a new xid per call,
-		 * so we first reinitialise the handle.
-		 */
-			(void) clnt_tli_kinit(ch->ch_client, svp->ri_knconf,
-			    &(svp->ri_addr), max_msgsize, retries, cred);
-
-			*clp = ch->ch_client;
-			return (0);
-		}
-	}
-
-	/* create new handle and cache it */
-	ch = (struct chtab *)kmem_zalloc(sizeof (*ch), KM_SLEEP);
-
-	if (ch) {
-		ch->ch_inuse = TRUE;
-		ch->ch_prog = RDC_PROGRAM;
-		ch->ch_vers = vers;
-		ch->ch_dev = svp->ri_knconf->knc_rdev;
-		ch->ch_protofmly = (char *)kmem_zalloc(
-		    strlen(svp->ri_knconf->knc_protofmly)+1, KM_SLEEP);
-		if (ch->ch_protofmly)
-			(void) strcpy(ch->ch_protofmly,
-			    svp->ri_knconf->knc_protofmly);
-		*plistp = ch;
-	}
-
-	mutex_exit(&rdc_clnt_lock);
-
-	ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
-	    RDC_PROGRAM, vers, max_msgsize, retries, cred, clp);
-
-	if (ret != 0) {
-		if (ch)
-			ch->ch_inuse = FALSE;
-		cmn_err(CE_NOTE, "!rdc_call: tli_kcreate failed %d", ret);
-		return (ret);
-	}
-
-	*rch = ch;
-	if (ch)
-		ch->ch_client = *clp;
-
-	(void) CLNT_CONTROL(*clp, CLSET_PROGRESS, NULL);
-
-	return (ret);
-}
-
-
-long rdc_clnt_count = 0;
-
-/*
- * rdc_clnt_call
- * Arguments:
- *	rdc_srv_t *svp - rdc servinfo
- *	rpcproc_t proc; - rpcid
- *	rpcvers_t vers; - protocol version
- *	xdrproc_t xargs;- xdr function
- *	caddr_t argsp;- args to xdr function
- *	xdrproc_t xres;- xdr function
- *	caddr_t resp;- args to xdr function
- *	struct timeval timeout;
- * Performs RPC client call using specific protocol and version
- */
-
-int
-rdc_clnt_call(rdc_srv_t *svp, rpcproc_t proc, rpcvers_t vers,
-		xdrproc_t xargs, caddr_t argsp,
-		xdrproc_t xres, caddr_t resp, struct timeval *timeout)
-{
-	CLIENT *rh = NULL;
-	int err;
-	int tries = 0;
-	struct chtab *ch = NULL;
-
-	err = rdc_clnt_get(svp, vers, &ch, &rh);
-	if (err || !rh)
-		return (err);
-
-	do {
-		DTRACE_PROBE3(rdc_clnt_call_1,
-		    CLIENT *, rh, rpcproc_t, proc, xdrproc_t, xargs);
-
-		err = cl_call_sig(rh, proc, xargs, argsp, xres, resp, *timeout);
-
-		DTRACE_PROBE1(rdc_clnt_call_end, int, err);
-
-		switch (err) {
-			case RPC_SUCCESS: /* bail now */
-				goto done;
-			case RPC_INTR:	/* No recovery from this */
-				goto done;
-			case RPC_PROGVERSMISMATCH:
-				goto done;
-			case RPC_TLIERROR:
-				/* fall thru */
-			case RPC_XPRTFAILED:
-				/* Delay here to err on side of caution */
-				/* fall thru */
-			case RPC_VERSMISMATCH:
-
-			default:
-				if (IS_UNRECOVERABLE_RPC(err)) {
-					goto done;
-				}
-				tries++;
-			/*
-			 * The call is in progress (over COTS)
-			 * Try the CLNT_CALL again, but don't
-			 * print a noisy error message
-			 */
-				if (err == RPC_INPROGRESS)
-					break;
-				cmn_err(CE_NOTE, "!SNDR client: err %d %s",
-				    err, clnt_sperrno(err));
-			}
-	} while (tries && (tries < 2));
-done:
-	++rdc_clnt_count;
-	rdc_clnt_free(ch, rh);
-	return (err);
-}
-
-
-/*
- * Call an rpc from the client side, not caring which protocol is used.
- */
-int
-rdc_clnt_call_any(rdc_srv_t *svp, rdc_if_t *ip, rpcproc_t proc,
-		xdrproc_t xargs, caddr_t argsp,
-		xdrproc_t xres, caddr_t resp, struct timeval *timeout)
-{
-	rpcvers_t vers;
-	int rc;
-
-	if (ip != NULL) {
-		vers = ip->rpc_version;
-	} else {
-		vers = RDC_VERS_MAX;
-	}
-
-	do {
-		rc = rdc_clnt_call(svp, proc, vers, xargs, argsp,
-		    xres, resp, timeout);
-
-		if (rc == RPC_PROGVERSMISMATCH) {
-			/*
-			 * Downgrade and try again.
-			 */
-			vers--;
-		}
-	} while ((vers >= RDC_VERS_MIN) && (rc == RPC_PROGVERSMISMATCH));
-
-	if ((rc == 0) && (ip != NULL) && (vers != ip->rpc_version)) {
-		mutex_enter(&rdc_ping_lock);
-		ip->rpc_version = vers;
-		mutex_exit(&rdc_ping_lock);
-	}
-
-	return (rc);
-}
-
-/*
- * Call an rpc from the client side, starting with protocol specified
- */
-int
-rdc_clnt_call_walk(rdc_k_info_t *krdc, rpcproc_t proc, xdrproc_t xargs,
-		caddr_t argsp, xdrproc_t xres, caddr_t resp,
-		struct timeval *timeout)
-{
-	int rc;
-	rpcvers_t vers;
-	rdc_srv_t *svp = krdc->lsrv;
-	rdc_if_t *ip = krdc->intf;
-	vers = krdc->rpc_version;
-
-	do {
-		rc = rdc_clnt_call(svp, proc, vers, xargs, argsp,
-		    xres, resp, timeout);
-
-		if (rc == RPC_PROGVERSMISMATCH) {
-			/*
-			 * Downgrade and try again.
-			 */
-			vers--;
-		}
-	} while ((vers >= RDC_VERS_MIN) && (rc == RPC_PROGVERSMISMATCH));
-
-	if ((rc == 0) && (ip != NULL) && (vers != ip->rpc_version)) {
-		mutex_enter(&rdc_ping_lock);
-		ip->rpc_version = vers;
-		mutex_exit(&rdc_ping_lock);
-	}
-
-	return (rc);
-}
-
-/*
- * rdc_clnt_free
- * Free a client structure into the cache, or if this was a temporary
- * handle allocated above MAXCLIENTS, destroy it.
- */
-static void
-rdc_clnt_free(struct chtab *ch, CLIENT *clp)
-{
-	if (ch != NULL) {
-		/* cached client, just clear inuse flag and return */
-		ASSERT(ch->ch_client == clp);
-		ch->ch_inuse = FALSE;
-		return;
-	}
-
-	/* temporary handle allocated above MAXCLIENTS, so destroy it */
-
-	if (clp->cl_auth) {
-		AUTH_DESTROY(clp->cl_auth);
-		clp->cl_auth = 0;
-	}
-
-	CLNT_DESTROY(clp);
-}
-
-
-/*
- * _rdc_clnt_destroy
- * Free a chain (ch_list or ch_next) of cached clients
- */
-static int
-_rdc_clnt_destroy(struct chtab **p, const int list)
-{
-	struct chtab *ch;
-	int leak = 0;
-
-	if (!p)
-		return (0);
-
-	while (*p != NULL) {
-		ch = *p;
-
-		/*
-		 * unlink from the chain
-		 * - this leaks the client if it was inuse
-		 */
-
-		*p = list ? ch->ch_list : ch->ch_next;
-
-		if (!ch->ch_inuse) {
-			/* unused client - destroy it */
-
-			if (ch->ch_client) {
-				if (ch->ch_client->cl_auth) {
-					AUTH_DESTROY(ch->ch_client->cl_auth);
-					ch->ch_client->cl_auth = 0;
-				}
-
-				CLNT_DESTROY(ch->ch_client);
-				ch->ch_client = 0;
-			}
-
-			if (ch->ch_protofmly)
-				kmem_free(ch->ch_protofmly,
-				    strlen(ch->ch_protofmly)+1);
-
-			kmem_free(ch, sizeof (*ch));
-		} else {
-			/* remember client leak */
-			leak++;
-		}
-	}
-
-	return (leak);
-}
-
-
-/*
- * rdc_clnt_destroy
- * Free client caching table on unconfigure
- */
-void
-rdc_clnt_destroy(void)
-{
-	struct chtab *ch;
-	int leak = 0;
-
-	mutex_enter(&rdc_clnt_lock);
-
-	/* destroy each ch_list chain */
-
-	for (ch = rdc_chtable; ch; ch = ch->ch_next) {
-		leak += _rdc_clnt_destroy(&ch->ch_list, 1);
-	}
-
-	/* destroy the main ch_next chain */
-	leak += _rdc_clnt_destroy(&rdc_chtable, 0);
-
-	if (leak) {
-		/* we are about to leak clients */
-		cmn_err(CE_WARN,
-		    "!rdc_clnt_destroy: leaking %d inuse clients", leak);
-	}
-
-	mutex_exit(&rdc_clnt_lock);
-}
-
-#ifdef	DEBUG
-/*
- * Function to send an asynchronous net_data6 request
- * direct to a server to allow the generation of
- * out of order requests for ZatoIchi tests.
- */
-int
-rdc_async6(void *arg, int mode, int *rvp)
-{
-	int			index;
-	rdc_async6_t		async6;
-	struct net_data6	data6;
-	rdc_k_info_t		*krdc;
-	rdc_u_info_t		*urdc;
-	char			*data;
-	int			datasz;
-	char			*datap;
-	int			rc;
-	struct timeval		t;
-	struct netwriteres	netret;
-	int i;
-
-	rc = 0;
-	*rvp = 0;
-	/*
-	 * copyin the user's arguments.
-	 */
-	if (ddi_copyin(arg, &async6, sizeof (async6), mode) < 0) {
-		return (EFAULT);
-	}
-
-	/*
-	 * search by the secondary host and file.
-	 */
-	mutex_enter(&rdc_conf_lock);
-	for (index = 0; index < rdc_max_sets; index++) {
-		urdc = &rdc_u_info[index];
-		krdc = &rdc_k_info[index];
-
-		if (!IS_CONFIGURED(krdc))
-			continue;
-		if (!IS_ENABLED(urdc))
-			continue;
-		if (!IS_ASYNC(urdc))
-			continue;
-		if (krdc->rpc_version < RDC_VERSION6)
-			continue;
-
-		if ((strncmp(urdc->secondary.intf, async6.sechost,
-		    MAX_RDC_HOST_SIZE) == 0) &&
-		    (strncmp(urdc->secondary.file, async6.secfile,
-		    NSC_MAXPATH) == 0)) {
-			break;
-		}
-	}
-	mutex_exit(&rdc_conf_lock);
-	if (index >= rdc_max_sets) {
-		return (ENOENT);
-	}
-
-	if (async6.spos != -1) {
-		if ((async6.spos < async6.pos) ||
-		    ((async6.spos + async6.slen) >
-		    (async6.pos + async6.len))) {
-			cmn_err(CE_WARN, "!Sub task not within range "
-			    "start %d length %d sub start %d sub length %d",
-			    async6.pos, async6.len, async6.spos, async6.slen);
-			return (EIO);
-		}
-	}
-
-	datasz = FBA_SIZE(1);
-	data = kmem_alloc(datasz, KM_SLEEP);
-	datap = data;
-	while (datap < &data[datasz]) {
-		/* LINTED */
-		*datap++ = async6.pat;
-	}
-
-	/*
-	 * Fill in the net databuffer prior to transmission.
-	 */
-
-	data6.local_cd = krdc->index;
-	if (krdc->remote_index == -1) {
-		cmn_err(CE_WARN, "!Remote index not known");
-		kmem_free(data, datasz);
-		return (EIO);
-	} else {
-		data6.cd = krdc->remote_index;
-	}
-	data6.pos = async6.pos;
-	data6.len = async6.len;
-	data6.flag = 0;
-	data6.idx = async6.idx;
-	data6.seq = async6.seq;
-
-	if (async6.spos == -1) {
-		data6.sfba = async6.pos;
-		data6.nfba = async6.len;
-		data6.endoblk = 1;
-
-	} else {
-		data6.sfba = async6.spos;
-		data6.nfba = async6.slen;
-		data6.endoblk = async6.endind;
-	}
-
-	data6.data.data_len = datasz;
-	data6.data.data_val = data;
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-
-	netret.vecdata.vecdata_val = NULL;
-	netret.vecdata.vecdata_len = 0;
-
-
-	rc = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE6, krdc->rpc_version,
-	    xdr_net_data6, (char *)&data6, xdr_netwriteres, (char *)&netret,
-	    &t);
-
-	kmem_free(data, datasz);
-	if (rc == 0) {
-		if (netret.result < 0) {
-			rc = -netret.result;
-		}
-		cmn_err(CE_NOTE, "!async6: seq %u result %d index %d "
-		    "pendcnt %d",
-		    netret.seq, netret.result, netret.index,
-		    netret.vecdata.vecdata_len);
-		for (i = 0; i < netret.vecdata.vecdata_len; i++) {
-			net_pendvec_t pvec;
-			bcopy(netret.vecdata.vecdata_val + i, &pvec,
-			    sizeof (net_pendvec_t));
-			cmn_err(CE_NOTE, "!Seq %u pos %llu len %llu",
-			    pvec.seq, (unsigned long long)pvec.apos,
-			    (unsigned long long)pvec.alen);
-		}
-		if (netret.vecdata.vecdata_val)
-			kmem_free(netret.vecdata.vecdata_val,
-			    netret.vecdata.vecdata_len *
-			    sizeof (net_pendvec_t));
-	} else {
-		cmn_err(CE_NOTE, "!async6: rpc call failed %d", rc);
-	}
-	*rvp = netret.index;
-	return (rc);
-}
-
-/*
- * Function to send an net_read6 request
- * direct to a server to allow the generation of
- * read requests.
- */
-int
-rdc_readgen(void *arg, int mode, int *rvp)
-{
-	int			index;
-	rdc_readgen_t		readgen;
-	rdc_readgen32_t		readgen32;
-	struct rread6		read6;
-	struct rread		read5;
-	rdc_k_info_t		*krdc;
-	int			ret;
-	struct timeval		t;
-	struct rdcrdresult	rr;
-	int			err;
-
-	*rvp = 0;
-	rr.rr_bufsize = 0;	/* rpc data buffer length (bytes) */
-	rr.rr_data = NULL;	/* rpc data buffer */
-	if (ddi_model_convert_from(mode & FMODELS) == DDI_MODEL_ILP32) {
-		if (ddi_copyin(arg, &readgen32, sizeof (readgen32), mode)) {
-			return (EFAULT);
-		}
-		(void) strncpy(readgen.sechost, readgen32.sechost,
-		    MAX_RDC_HOST_SIZE);
-		(void) strncpy(readgen.secfile, readgen32.secfile, NSC_MAXPATH);
-		readgen.len = readgen32.len;
-		readgen.pos = readgen32.pos;
-		readgen.idx = readgen32.idx;
-		readgen.flag = readgen32.flag;
-		readgen.data = (void *)(unsigned long)readgen32.data;
-		readgen.rpcversion = readgen32.rpcversion;
-	} else {
-		if (ddi_copyin(arg, &readgen, sizeof (readgen), mode)) {
-			return (EFAULT);
-		}
-	}
-	switch (readgen.rpcversion) {
-	case 5:
-	case 6:
-		break;
-	default:
-		return (EINVAL);
-	}
-
-	mutex_enter(&rdc_conf_lock);
-	index = rdc_lookup_byhostdev(readgen.sechost, readgen.secfile);
-	if (index >= 0) {
-		krdc = &rdc_k_info[index];
-	}
-	if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
-		mutex_exit(&rdc_conf_lock);
-		return (ENODEV);
-	}
-	/*
-	 * we should really call setbusy here.
-	 */
-	mutex_exit(&rdc_conf_lock);
-
-	t.tv_sec = rdc_rpc_tmout;
-	t.tv_usec = 0;
-	if (krdc->remote_index == -1) {
-		cmn_err(CE_WARN, "!Remote index not known");
-		ret = EIO;
-		goto out;
-	}
-	if (readgen.rpcversion == 6) {
-		read6.cd = krdc->remote_index;
-		read6.len = readgen.len;
-		read6.pos = readgen.pos;
-		read6.idx = readgen.idx;
-		read6.flag = readgen.flag;
-	} else {
-		read5.cd = krdc->remote_index;
-		read5.len = readgen.len;
-		read5.pos = readgen.pos;
-		read5.idx = readgen.idx;
-		read5.flag = readgen.flag;
-	}
-
-	if (readgen.flag & RDC_RREAD_START) {
-		if (readgen.rpcversion == 6) {
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
-			    RDC_VERSION6, xdr_rread6, (char *)&read6,
-			    xdr_int, (char *)&ret, &t);
-		} else {
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
-			    RDC_VERSION5, xdr_rread, (char *)&read5,
-			    xdr_int, (char *)&ret, &t);
-		}
-		if (err == 0) {
-			*rvp = ret;
-			ret = 0;
-		} else {
-			ret = EPROTO;
-		}
-	} else {
-		if (readgen.rpcversion == 6) {
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
-			    RDC_VERSION6, xdr_rread6, (char *)&read6,
-			    xdr_rdresult, (char *)&rr, &t);
-		} else {
-			err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
-			    RDC_VERSION5, xdr_rread, (char *)&read5,
-			    xdr_rdresult, (char *)&rr, &t);
-		}
-		if (err == 0) {
-			if (rr.rr_status != RDC_OK) {
-				ret = EIO;
-				goto out;
-			}
-			*rvp = rr.rr_bufsize;
-			if (ddi_copyout(rr.rr_data, readgen.data,
-			    rr.rr_bufsize, mode) != 0) {
-				ret = EFAULT;
-				goto out;
-			}
-			ret = 0;
-		} else {
-			ret = EPROTO;
-			goto out;
-		}
-	}
-out:
-	if (rr.rr_data) {
-		kmem_free(rr.rr_data, rr.rr_bufsize);
-	}
-	return (ret);
-}
-
-
-#endif
