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

#include <sys/types.h>
#include <sys/ksynch.h>
#include <sys/cmn_err.h>
#include <sys/kmem.h>
#include <sys/stat.h>
#include <sys/errno.h>

#include "../solaris/nsc_thread.h"
#ifdef DS_DDICT
#include "../contract.h"
#endif
#include <sys/nsctl/nsctl.h>

#include <sys/kmem.h>
#include <sys/ddi.h>

#include <sys/sdt.h>		/* dtrace is S10 or later */

#include "rdc_io.h"
#include "rdc_bitmap.h"
#include "rdc_diskq.h"
#include "rdc_clnt.h"

#include <sys/unistat/spcs_s.h>
#include <sys/unistat/spcs_s_k.h>
#include <sys/unistat/spcs_errors.h>

extern nsc_io_t *_rdc_io_hc;

int rdc_diskq_coalesce = 0;

int
_rdc_rsrv_diskq(rdc_group_t *group)
{
	int rc = 0;

	mutex_enter(&group->diskqmutex);
	if (group->diskqfd == NULL) {
		mutex_exit(&group->diskqmutex);
		return (EIO);
	} else if ((group->diskqrsrv == 0) &&
	    (rc = nsc_reserve(group->diskqfd, 0)) != 0) {
		cmn_err(CE_WARN,
		    "!rdc: nsc_reserve(%s) failed %d\n",
		    nsc_pathname(group->diskqfd), rc);
	} else {
		group->diskqrsrv++;
	}

	mutex_exit(&group->diskqmutex);
	return (rc);
}

void
_rdc_rlse_diskq(rdc_group_t *group)
{
	mutex_enter(&group->diskqmutex);
	if (group->diskqrsrv > 0 && --group->diskqrsrv == 0) {
		nsc_release(group->diskqfd);
	}
	mutex_exit(&group->diskqmutex);
}

void
rdc_wait_qbusy(disk_queue *q)
{
	ASSERT(MUTEX_HELD(QLOCK(q)));
	while (q->busycnt > 0)
		cv_wait(&q->busycv, QLOCK(q));
}

void
rdc_set_qbusy(disk_queue *q)
{
	ASSERT(MUTEX_HELD(QLOCK(q)));
	q->busycnt++;
}

void
rdc_clr_qbusy(disk_queue *q)
{
	ASSERT(MUTEX_HELD(QLOCK(q)));
	q->busycnt--;
	if (q->busycnt == 0)
		cv_broadcast(&q->busycv);
}

int
rdc_lookup_diskq(char *pathname)
{
	rdc_u_info_t *urdc;
#ifdef DEBUG
	rdc_k_info_t *krdc;
#endif
	int index;

	for (index = 0; index < rdc_max_sets; index++) {
		urdc = &rdc_u_info[index];
#ifdef DEBUG
		krdc = &rdc_k_info[index];
#endif
		ASSERT(krdc->index == index);
		ASSERT(urdc->index == index);
		if (!IS_ENABLED(urdc))
			continue;

		if (strncmp(pathname, urdc->disk_queue,
		    NSC_MAXPATH) == 0)
			return (index);
	}

	return (-1);
}

void
rdc_unintercept_diskq(rdc_group_t *grp)
{
	if (!RDC_IS_DISKQ(grp))
		return;
	if (grp->q_tok)
		(void) nsc_unregister_path(grp->q_tok, 0);
	grp->q_tok = NULL;
}

void
rdc_close_diskq(rdc_group_t *grp)
{

	if (grp == NULL) {
#ifdef DEBUG
		cmn_err(CE_WARN, "!rdc_close_diskq: NULL group!");
#endif
		return;
	}

	if (grp->diskqfd) {
		if (nsc_close(grp->diskqfd) != 0) {
#ifdef DEBUG
			cmn_err(CE_WARN, "!nsc_close on diskq failed");
#else
			;
			/*EMPTY*/
#endif
		}
		grp->diskqfd = 0;
		grp->diskqrsrv = 0;
	}
	bzero(&grp->diskq.disk_hdr, sizeof (diskq_header));
}

/*
 * nsc_open the diskq and attach
 * the nsc_fd to krdc->diskqfd
 */
int
rdc_open_diskq(rdc_k_info_t *krdc)
{
	rdc_u_info_t *urdc;
	rdc_group_t *grp;
	int sts;
	nsc_size_t size;
	char *diskqname;
	int mutexheld = 0;

	grp = krdc->group;
	urdc = &rdc_u_info[krdc->index];

	mutex_enter(&grp->diskqmutex);
	mutexheld++;
	if (&urdc->disk_queue[0] == '\0') {
		goto fail;
	}

	diskqname = &urdc->disk_queue[0];

	if (grp->diskqfd == NULL) {
		grp->diskqfd = nsc_open(diskqname,
		    NSC_RDCHR_ID|NSC_DEVICE|NSC_WRITE, 0, 0, 0);
		if (grp->diskqfd == NULL) {
			cmn_err(CE_WARN, "!rdc_open_diskq: Unable to open %s",
			    diskqname);
			goto fail;
		}
	}
	if (!grp->q_tok)
		grp->q_tok = nsc_register_path(urdc->disk_queue,
		    NSC_DEVICE | NSC_CACHE, _rdc_io_hc);

	grp->diskqrsrv = 0; /* init reserve count */

	mutex_exit(&grp->diskqmutex);
	mutexheld--;
	/* just test a reserve release */
	sts = _rdc_rsrv_diskq(grp);
	if (!RDC_SUCCESS(sts)) {
		cmn_err(CE_WARN, "!rdc_open_diskq: Reserve failed for %s",
		    diskqname);
		goto fail;
	}
	sts = nsc_partsize(grp->diskqfd, &size);
	_rdc_rlse_diskq(grp);

	if ((sts == 0) && (size < 1)) {
		rdc_unintercept_diskq(grp);
		rdc_close_diskq(grp);
		goto fail;
	}

	return (0);

fail:
	bzero(&urdc->disk_queue, NSC_MAXPATH);
	if (mutexheld)
		mutex_exit(&grp->diskqmutex);
	return (-1);

}

/*
 * rdc_count_vecs
 * simply vec++'s until sb_addr is null
 * returns number of vectors encountered
 */
int
rdc_count_vecs(nsc_vec_t *vec)
{
	nsc_vec_t	*vecp;
	int i = 0;
	vecp = vec;
	while (vecp->sv_addr) {
		vecp++;
		i++;
	}
	return (i+1);
}
/*
 * rdc_setid2idx
 * given setid, return index
 */
int
rdc_setid2idx(int setid) {

	int index = 0;

	for (index = 0; index < rdc_max_sets; index++) {
		if (rdc_u_info[index].setid == setid)
			break;
	}
	if (index >= rdc_max_sets)
		index = -1;
	return (index);
}

/*
 * rdc_idx2setid
 * given an index, return its setid
 */
int
rdc_idx2setid(int index)
{
	return (rdc_u_info[index].setid);
}

/*
 * rdc_fill_ioheader
 * fill in all the stuff you want to save on disk
 * at the beginnig of each queued write
 */
void
rdc_fill_ioheader(rdc_aio_t *aio, io_hdr *hd, int qpos)
{
	ASSERT(MUTEX_HELD(&rdc_k_info[aio->index].group->diskq.disk_qlock));

	hd->dat.magic = RDC_IOHDR_MAGIC;
	hd->dat.type = RDC_QUEUEIO;
	hd->dat.pos = aio->pos;
	hd->dat.hpos = aio->pos;
	hd->dat.qpos = qpos;
	hd->dat.len = aio->len;
	hd->dat.flag = aio->flag;
	hd->dat.iostatus = aio->iostatus;
	hd->dat.setid = rdc_idx2setid(aio->index);
	hd->dat.time = nsc_time();
	if (!aio->handle)
		hd->dat.flag |= RDC_NULL_BUF; /* no real data to queue */
}

/*
 * rdc_dump_iohdrs
 * give back the iohdr list
 * and clear out q->lastio
 */
void
rdc_dump_iohdrs(disk_queue *q)
{
	io_hdr *p, *r;

	ASSERT(MUTEX_HELD(QLOCK(q)));

	p = q->iohdrs;
	while (p) {
		r = p->dat.next;
		kmem_free(p, sizeof (*p));
		q->hdrcnt--;
		p = r;
	}
	q->iohdrs = q->hdr_last = NULL;
	q->hdrcnt = 0;
	if (q->lastio->handle)
		(void) nsc_free_buf(q->lastio->handle);
	bzero(&(*q->lastio), sizeof (*q->lastio));
}

/*
 * rdc_fail_diskq
 * set flags, throw away q info
 * clean up what you can
 * wait for flusher threads to stop (taking into account this may be one)
 * takes group_lock, so conf, many, and bitmap may not be held
 */
void
rdc_fail_diskq(rdc_k_info_t *krdc, int wait, int flag)
{
	rdc_k_info_t *p;
	rdc_u_info_t *q = &rdc_u_info[krdc->index];
	rdc_group_t *group = krdc->group;
	disk_queue *dq = &krdc->group->diskq;

	if (IS_STATE(q, RDC_DISKQ_FAILED))
		return;

	if (!(flag & RDC_NOFAIL))
		cmn_err(CE_WARN, "!disk queue %s failure", q->disk_queue);

	if (flag & RDC_DOLOG) {
		rdc_group_enter(krdc);
		rdc_group_log(krdc, RDC_NOFLUSH | RDC_ALLREMOTE,
		    "disk queue failed");
		rdc_group_exit(krdc);
	}
	mutex_enter(QHEADLOCK(dq));
	mutex_enter(QLOCK(dq));
	/*
	 * quick stop of the flushers
	 * other cleanup is done on the un-failing of the diskq
	 */
	SET_QHEAD(dq, RDC_DISKQ_DATA_OFF);
	SET_QTAIL(dq, RDC_DISKQ_DATA_OFF);
	SET_QNXTIO(dq, RDC_DISKQ_DATA_OFF);
	SET_LASTQTAIL(dq, 0);

	rdc_dump_iohdrs(dq);

	mutex_exit(QLOCK(dq));
	mutex_exit(QHEADLOCK(dq));

	bzero(krdc->bitmap_ref, krdc->bitmap_size * BITS_IN_BYTE *
	    BMAP_REF_PREF_SIZE);

	if (flag & RDC_DOLOG) /* otherwise, we already have the conf lock */
		rdc_group_enter(krdc);

	else if (!(flag & RDC_GROUP_LOCKED))
		ASSERT(MUTEX_HELD(&rdc_conf_lock));

	if (!(flag & RDC_NOFAIL)) {
		rdc_set_flags(q, RDC_DISKQ_FAILED);
	}
	rdc_clr_flags(q, RDC_QUEUING);

	for (p = krdc->group_next; p != krdc; p = p->group_next) {
		q = &rdc_u_info[p->index];
		if (!IS_ENABLED(q))
			continue;
		if (!(flag & RDC_NOFAIL)) {
			rdc_set_flags(q, RDC_DISKQ_FAILED);
		}
		rdc_clr_flags(q, RDC_QUEUING);
		bzero(p->bitmap_ref, p->bitmap_size * BITS_IN_BYTE *
		    BMAP_REF_PREF_SIZE);
		/* RDC_QUEUING is cleared in group_log() */
	}

	if (flag & RDC_DOLOG)
		rdc_group_exit(krdc);

	/* can't wait for myself to go away, I'm a flusher */
	if (wait & RDC_WAIT)
		while (group->rdc_thrnum)
			delay(2);

}

/*
 * rdc_stamp_diskq
 * write out diskq header info
 * must have disk_qlock held
 * if rsrvd flag is 0, the nsc_reserve is done
 */
int
rdc_stamp_diskq(rdc_k_info_t *krdc, int rsrvd, int failflags)
{
	nsc_vec_t	vec[2];
	nsc_buf_t	*head = NULL;
	rdc_group_t	*grp;
	rdc_u_info_t	*urdc;
	disk_queue	*q;
	int		rc, flags;

	grp = krdc->group;
	q = &krdc->group->diskq;

	ASSERT(MUTEX_HELD(&q->disk_qlock));

	urdc = &rdc_u_info[krdc->index];

	if (!rsrvd && _rdc_rsrv_diskq(grp)) {
		cmn_err(CE_WARN, "!rdc_stamp_diskq: %s reserve failed",
		    urdc->disk_queue);
		mutex_exit(QLOCK(q));
		rdc_fail_diskq(krdc, RDC_NOWAIT, failflags);
		mutex_enter(QLOCK(q));
		return (-1);
	}
	flags = NSC_WRITE | NSC_NOCACHE | NSC_NODATA;
	rc = nsc_alloc_buf(grp->diskqfd, 0, 1, flags, &head);

	if (!RDC_SUCCESS(rc)) {
		cmn_err(CE_WARN, "!Alloc buf failed for disk queue %s",
		    &urdc->disk_queue[0]);
		mutex_exit(QLOCK(q));
		rdc_fail_diskq(krdc, RDC_NOWAIT, failflags);
		mutex_enter(QLOCK(q));
		return (-1);
	}
	vec[0].sv_len = FBA_SIZE(1);
	vec[0].sv_addr = (uchar_t *)&q->disk_hdr;
	vec[1].sv_len = 0;
	vec[1].sv_addr = NULL;

	head->sb_vec = &vec[0];

#ifdef DEBUG_DISKQ
	cmn_err(CE_NOTE, "!rdc_stamp_diskq: hdr: %p magic: %x state: "
	    "%x head: %d tail: %d size: %d nitems: %d blocks: %d",
	    q, QMAGIC(q), QSTATE(q), QHEAD(q),
	    QTAIL(q), QSIZE(q), QNITEMS(q), QBLOCKS(q));
#endif

	rc = nsc_write(head, 0, 1, 0);

	if (!RDC_SUCCESS(rc)) {
		if (!rsrvd)
			_rdc_rlse_diskq(grp);
		cmn_err(CE_CONT, "!disk queue %s failed rc %d",
		    &urdc->disk_queue[0], rc);
		mutex_exit(QLOCK(q));
		rdc_fail_diskq(krdc, RDC_NOWAIT, failflags);
		mutex_enter(QLOCK(q));
		return (-1);
	}

	(void) nsc_free_buf(head);
	if (!rsrvd)
		_rdc_rlse_diskq(grp);

	return (0);
}

/*
 * rdc_init_diskq_header
 * load initial values into the header
 */
void
rdc_init_diskq_header(rdc_group_t *grp, dqheader *header)
{
	int rc;
	int type = 0;
	disk_queue *q = &grp->diskq;

	ASSERT(MUTEX_HELD(QLOCK(q)));

	/* save q type if this is a failure */
	if (QSTATE(q) & RDC_QNOBLOCK)
		type = RDC_QNOBLOCK;
	bzero(header, sizeof (*header));
	header->h.magic = RDC_DISKQ_MAGIC;
	header->h.vers = RDC_DISKQ_VERS;
	header->h.state |= (RDC_SHUTDOWN_BAD|type); /* SHUTDOWN_OK on suspend */
	header->h.head_offset = RDC_DISKQ_DATA_OFF;
	header->h.tail_offset = RDC_DISKQ_DATA_OFF;
	header->h.nitems = 0;
	header->h.blocks = 0;
	header->h.qwrap = 0;
	SET_QNXTIO(q, QHEAD(q));
	SET_QCOALBOUNDS(q, RDC_DISKQ_DATA_OFF);

	/* do this last, as this might be a failure. get the kernel state ok */
	rc = _rdc_rsrv_diskq(grp);
	if (!RDC_SUCCESS(rc)) {
		cmn_err(CE_WARN, "!init_diskq_hdr: Reserve failed for queue");
		return;
	}
	(void) nsc_partsize(grp->diskqfd, &header->h.disk_size);
	_rdc_rlse_diskq(grp);

}

/*
 * rdc_unfail_diskq
 * the diskq failed for some reason, lets try and re-start it
 * the old stuff has already been thrown away
 * should just be called from rdc_sync
 */
void
rdc_unfail_diskq(rdc_k_info_t *krdc)
{
	rdc_k_info_t *p;
	rdc_u_info_t *q = &rdc_u_info[krdc->index];
	rdc_group_t *group = krdc->group;
	disk_queue *dq = &group->diskq;

	rdc_group_enter(krdc);
	rdc_clr_flags(q, RDC_ASYNC);
	/* someone else won the race... */
	if (!IS_STATE(q, RDC_DISKQ_FAILED)) {
		rdc_group_exit(krdc);
		return;
	}
	rdc_clr_flags(q, RDC_DISKQ_FAILED);
	for (p = krdc->group_next; p != krdc; p = p->group_next) {
		q = &rdc_u_info[p->index];
		if (!IS_ENABLED(q))
			continue;
		rdc_clr_flags(q, RDC_DISKQ_FAILED);
		rdc_clr_flags(q, RDC_ASYNC);
		if (IS_STATE(q, RDC_QUEUING))
			rdc_clr_flags(q, RDC_QUEUING);
	}
	rdc_group_exit(krdc);

	mutex_enter(QLOCK(dq));

	rdc_init_diskq_header(group, &group->diskq.disk_hdr);
	/* real i/o to the queue */
	/* clear RDC_AUXSYNCIP because we cannot halt a sync that's not here */
	krdc->aux_state &= ~RDC_AUXSYNCIP;
	if (rdc_stamp_diskq(krdc, 0, RDC_GROUP_LOCKED | RDC_DOLOG) < 0) {
		mutex_exit(QLOCK(dq));
		goto fail;
	}

	SET_QNXTIO(dq, QHEAD(dq));
	SET_QHDRCNT(dq, 0);
	SET_QSTATE(dq, RDC_SHUTDOWN_BAD); /* only suspend can write good */
	dq->iohdrs = NULL;
	dq->hdr_last = NULL;

	/* should be none, but.. */
	rdc_dump_iohdrs(dq);

	mutex_exit(QLOCK(dq));


fail:
	krdc->aux_state |= RDC_AUXSYNCIP;
	return;

}

int
rdc_read_diskq_header(rdc_k_info_t *krdc)
{
	int rc;
	diskq_header *header;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];

	if (krdc->group->diskqfd == NULL) {
		char buf[NSC_MAXPATH];
		(void) snprintf(buf, NSC_MAXPATH, "%s:%s", urdc->secondary.intf,
		    &urdc->secondary.intf[0]);
		cmn_err(CE_WARN, "!Disk Queue Header read failed for %s",
		    &urdc->group_name[0] == '\0' ? buf:
		    &urdc->group_name[0]);
		return (-1);
	}

	header = &krdc->group->diskq.disk_hdr.h;
	if (_rdc_rsrv_diskq(krdc->group)) {
		return (-1);
	}

	rc = rdc_ns_io(krdc->group->diskqfd, NSC_RDBUF, 0,
	    (uchar_t *)header, sizeof (diskq_header));

	_rdc_rlse_diskq(krdc->group);

	if (!RDC_SUCCESS(rc)) {
		char buf[NSC_MAXPATH];
		(void) snprintf(buf, NSC_MAXPATH, "%s:%s", urdc->secondary.intf,
		    &urdc->secondary.file[0]);
		cmn_err(CE_WARN, "!Disk Queue Header read failed(%d) for %s",
		    rc, &urdc->group_name[0] == '\0' ? buf :
		    &urdc->group_name[0]);
		return (-1);
	}
	return (0);
}

/*
 * rdc_stop_diskq_flusher
 */
void
rdc_stop_diskq_flusher(rdc_k_info_t *krdc)
{
	disk_queue q, *qp;
	rdc_group_t *group;
#ifdef DEBUG
	cmn_err(CE_NOTE, "!stopping flusher threads");
#endif
	group = krdc->group;
	qp = &krdc->group->diskq;

	/* save the queue info */
	q = *qp;

	/* lie a little */
	SET_QTAIL(qp, RDC_DISKQ_DATA_OFF);
	SET_QHEAD(qp, RDC_DISKQ_DATA_OFF);
	SET_QSTATE(qp, RDC_QDISABLEPEND);
	SET_QSTATE(qp, RDC_STOPPINGFLUSH);

	/* drop locks to allow flushers to die */
	mutex_exit(QLOCK(qp));
	mutex_exit(QHEADLOCK(qp));
	rdc_group_exit(krdc);

	while (group->rdc_thrnum)
		delay(2);

	rdc_group_enter(krdc);
	mutex_enter(QHEADLOCK(qp));
	mutex_enter(QLOCK(qp));

	CLR_QSTATE(qp, RDC_STOPPINGFLUSH);
	*qp = q;
}

/*
 * rdc_enable_diskq
 * open the diskq
 * and stamp the header onto it.
 */
int
rdc_enable_diskq(rdc_k_info_t *krdc)
{
	rdc_group_t *group;
	disk_queue *q;

	group = krdc->group;
	q = &group->diskq;

	if (rdc_open_diskq(krdc) < 0)
		goto fail;

	mutex_enter(QLOCK(q));
	rdc_init_diskq_header(group, &group->diskq.disk_hdr);

	if (rdc_stamp_diskq(krdc, 0, RDC_NOLOG) < 0) {
		mutex_exit(QLOCK(q));
		goto fail;
	}

	SET_QNXTIO(q, QHEAD(q));

	mutex_exit(QLOCK(q));
	return (0);

fail:
	mutex_enter(&group->diskqmutex);
	rdc_close_diskq(group);
	mutex_exit(&group->diskqmutex);

	/* caller has to fail diskq after dropping conf & many locks */
	return (RDC_EQNOADD);
}

/*
 * rdc_resume_diskq
 * open the diskq and read the header
 */
int
rdc_resume_diskq(rdc_k_info_t *krdc)
{
	rdc_u_info_t *urdc;
	rdc_group_t *group;
	disk_queue *q;
	int rc = 0;

	urdc = &rdc_u_info[krdc->index];
	group = krdc->group;
	q = &group->diskq;

	if (rdc_open_diskq(krdc) < 0) {
		rc = RDC_EQNOADD;
		goto fail;
	}

	mutex_enter(QLOCK(q));

	rdc_init_diskq_header(group, &group->diskq.disk_hdr);

	if (rdc_read_diskq_header(krdc) < 0) {
		SET_QSTATE(q, RDC_QBADRESUME);
		rc = RDC_EQNOADD;
	}

	/* check diskq magic number */
	if (QMAGIC(q) != RDC_DISKQ_MAGIC) {
		cmn_err(CE_WARN, "!SNDR: unable to resume diskq %s,"
		    " incorrect magic number in header", urdc->disk_queue);
		rdc_init_diskq_header(group, &group->diskq.disk_hdr);
		SET_QSTATE(q, RDC_QBADRESUME);
		rc = RDC_EQNOADD;
	} else switch (QVERS(q)) {
		diskq_header1 h1;	/* version 1 header */
		diskq_header *hc;	/* current header */

#ifdef	NSC_MULTI_TERABYTE
		case RDC_DISKQ_VER_ORIG:
			/* version 1 diskq header, upgrade to 64bit version */
		h1 = *(diskq_header1 *)(&group->diskq.disk_hdr.h);
		hc = &group->diskq.disk_hdr.h;

		cmn_err(CE_WARN, "!SNDR: old version header for diskq %s,"
		    " upgrading to current version", urdc->disk_queue);
		hc->vers = RDC_DISKQ_VERS;
		hc->state = h1.state;
		hc->head_offset = h1.head_offset;
		hc->tail_offset = h1.tail_offset;
		hc->disk_size = h1.disk_size;
		hc->nitems = h1.nitems;
		hc->blocks = h1.blocks;
		hc->qwrap = h1.qwrap;
		hc->auxqwrap = h1.auxqwrap;
		hc->seq_last = h1.seq_last;
		hc->ack_last = h1.ack_last;

		if (hc->nitems > 0) {
			cmn_err(CE_WARN, "!SNDR: unable to resume diskq %s,"
			    " old version Q contains data", urdc->disk_queue);
			rdc_init_diskq_header(group, &group->diskq.disk_hdr);
			SET_QSTATE(q, RDC_QBADRESUME);
			rc = RDC_EQNOADD;
		}
		break;
#else
		case RDC_DISKQ_VER_64BIT:
			cmn_err(CE_WARN, "!SNDR: unable to resume diskq %s,"
			    " diskq header newer than current version",
			    urdc->disk_queue);
			rdc_init_diskq_header(group, &group->diskq.disk_hdr);
			SET_QSTATE(q, RDC_QBADRESUME);
			rc = RDC_EQNOADD;
		break;
#endif
		case RDC_DISKQ_VERS:
			/* okay, current version diskq */
		break;
		default:
			cmn_err(CE_WARN, "!SNDR: unable to resume diskq %s,"
			    " unknown diskq header version", urdc->disk_queue);
			rdc_init_diskq_header(group, &group->diskq.disk_hdr);
			SET_QSTATE(q, RDC_QBADRESUME);
			rc = RDC_EQNOADD;
		break;
	}
	if (IS_QSTATE(q, RDC_SHUTDOWN_BAD)) {
		cmn_err(CE_WARN, "!SNDR: unable to resume diskq %s,"
		    " unsafe shutdown", urdc->disk_queue);
		rdc_init_diskq_header(group, &group->diskq.disk_hdr);
		SET_QSTATE(q, RDC_QBADRESUME);
		rc = RDC_EQNOADD;
	}

	CLR_QSTATE(q, RDC_SHUTDOWN_OK);
	SET_QSTATE(q, RDC_SHUTDOWN_BAD);

	/* bad, until proven not bad */
	if (rdc_stamp_diskq(krdc, 0, RDC_NOLOG) < 0) {
		rdc_fail_diskq(krdc, RDC_NOWAIT, RDC_NOLOG);
		rc = RDC_EQNOADD;
	}

	SET_QNXTIO(q, QHEAD(q));
	group->diskq.nitems_hwm = QNITEMS(q);
	group->diskq.blocks_hwm = QBLOCKS(q);

	mutex_exit(QLOCK(q));

#ifdef DEBUG
	cmn_err(CE_NOTE, "!rdc_resume_diskq: resuming diskq %s \n",
	    urdc->disk_queue);
	cmn_err(CE_NOTE, "!qinfo: " QDISPLAY(q));
#endif
	if (rc == 0)
		return (0);

fail:

	/* caller has to set the diskq failed after dropping it's locks */
	return (rc);

}

int
rdc_suspend_diskq(rdc_k_info_t *krdc)
{
	int rc;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
	disk_queue *q;

	q = &krdc->group->diskq;

	/* grab both diskq locks as we are going to kill the flusher */
	mutex_enter(QHEADLOCK(q));
	mutex_enter(QLOCK(q));

	if ((krdc->group->rdc_thrnum) && (!IS_QSTATE(q, RDC_STOPPINGFLUSH))) {
		SET_QSTATE(q, RDC_STOPPINGFLUSH);
		rdc_stop_diskq_flusher(krdc);
		CLR_QSTATE(q, RDC_STOPPINGFLUSH);
	}

	krdc->group->diskq.disk_hdr.h.state &= ~RDC_SHUTDOWN_BAD;
	krdc->group->diskq.disk_hdr.h.state |= RDC_SHUTDOWN_OK;
	krdc->group->diskq.disk_hdr.h.state &= ~RDC_QBADRESUME;

	/* let's make sure that the flusher has stopped.. */
	if (krdc->group->rdc_thrnum) {
		mutex_exit(QLOCK(q));
		mutex_exit(QHEADLOCK(q));
		rdc_group_exit(krdc);

		while (krdc->group->rdc_thrnum)
			delay(5);

		rdc_group_enter(krdc);
		mutex_enter(QLOCK(q));
		mutex_enter(QHEADLOCK(q));
	}
	/* write refcount to the bitmap */
	if ((rc = rdc_write_refcount(krdc)) < 0) {
		rdc_group_exit(krdc);
		goto fail;
	}

	if (!QEMPTY(q)) {
		rdc_set_flags(urdc, RDC_QUEUING);
	} else {
		rdc_clr_flags(urdc, RDC_QUEUING);
	}

	/* fill in diskq header info */
	krdc->group->diskq.disk_hdr.h.state &= ~RDC_QDISABLEPEND;

#ifdef DEBUG
	cmn_err(CE_NOTE, "!suspending disk queue\n" QDISPLAY(q));
#endif

	/* to avoid a possible deadlock, release in order, and reacquire */
	mutex_exit(QLOCK(q));
	mutex_exit(QHEADLOCK(q));

	if (krdc->group->count > 1) {
		rdc_group_exit(krdc);
		goto fail; /* just stamp on the last suspend */
	}
	rdc_group_exit(krdc); /* in case this stamp fails */
	mutex_enter(QLOCK(q));

	rc = rdc_stamp_diskq(krdc, 0, RDC_NOLOG);

	mutex_exit(QLOCK(q));

fail:
	rdc_group_enter(krdc);

	/* diskq already failed if stamp failed */

	return (rc);
}

/*
 * copy orig aio to copy, including the nsc_buf_t
 */
int
rdc_dup_aio(rdc_aio_t *orig, rdc_aio_t *copy)
{
	int rc;
	bcopy(orig, copy, sizeof (*orig));
	copy->handle = NULL;

	if (orig->handle == NULL) /* no buf to alloc/copy */
		return (0);

	rc = nsc_alloc_abuf(orig->pos, orig->len, 0, &copy->handle);
	if (!RDC_SUCCESS(rc)) {
#ifdef DEBUG
		cmn_err(CE_WARN, "!rdc_dup_aio: alloc_buf failed (%d)", rc);
#endif
		return (rc);
	}
	rc = nsc_copy(orig->handle, copy->handle, orig->pos,
	    orig->pos, orig->len);
	if (!RDC_SUCCESS(rc)) {
		(void) nsc_free_buf(copy->handle);
#ifdef DEBUG
		cmn_err(CE_WARN, "!rdc_dup_aio: copy buf failed (%d)", rc);
#endif
		return (rc);
	}
	return (0);
}

/*
 * rdc_qfill_shldwakeup()
 * 0 if the memory queue has filled, and the low water
 * mark has not been reached. 0 if diskq is empty.
 * 1 if less than low water mark
 * net_queue mutex is already held
 */
int
rdc_qfill_shldwakeup(rdc_k_info_t *krdc)
{
	rdc_group_t *group = krdc->group;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
	net_queue *nq = &group->ra_queue;
	disk_queue *dq = &group->diskq;

	ASSERT(MUTEX_HELD(&nq->net_qlock));

	if (!RDC_IS_DISKQ(krdc->group))
		return (0);

	if (nq->qfill_sleeping != RDC_QFILL_ASLEEP)
		return (0);

	if (nq->qfflags & RDC_QFILLSTOP)
		return (1);

	if (nq->qfflags & RDC_QFILLSLEEP)
		return (0);

	if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING))
		return (0);

	mutex_enter(QLOCK(dq));
	if ((QNXTIO(dq) == QTAIL(dq)) && !IS_QSTATE(dq, RDC_QFULL)) {
		mutex_exit(QLOCK(dq));
		return (0);
	}
	mutex_exit(QLOCK(dq));

	if (nq->qfill_sleeping == RDC_QFILL_ASLEEP) {
		if (nq->hwmhit) {
			if (nq->blocks <= RDC_LOW_QBLOCKS) {
				nq->hwmhit = 0;
			} else {
				return (0);
			}
		}
#ifdef DEBUG_DISKQ_NOISY
		cmn_err(CE_NOTE, "!Waking up diskq->memq flusher, flags 0x%x"
		    " idx: %d", rdc_get_vflags(urdc), urdc->index);
#endif
		return (1);
	}
	return (0);

}

/*
 * rdc_diskq_enqueue
 * enqueue one i/o to the diskq
 * after appending some metadata to the front
 */
int
rdc_diskq_enqueue(rdc_k_info_t *krdc, rdc_aio_t *aio)
{
	nsc_vec_t	*vec = NULL;
	nsc_buf_t	*bp = NULL;
	nsc_buf_t	*qbuf = NULL;
	io_hdr		*iohdr = NULL;
	disk_queue	*q;
	rdc_group_t	*group;
	int		numvecs;
	int		i, j, rc = 0;
	int		retries = 0;
	rdc_u_info_t	*urdc;
	nsc_size_t	iofbas; /* len of io + io header len */
	int		qtail;
	int		delay_time = 2;
	int 		print_msg = 1;

#ifdef DEBUG_WRITER_UBERNOISE
	int		qhead;
#endif
	urdc = &rdc_u_info[krdc->index];
	group = krdc->group;
	q = &group->diskq;

	mutex_enter(QLOCK(q));

	/*
	 * there is a thread that is blocking because the queue is full,
	 * don't try to set up this write until all is clear
	 * check before and after for logging or failed queue just
	 * in case a thread was in flight while the queue was full,
	 * and in the proccess of failing
	 */
	while (IS_QSTATE(q, RDC_QFULL)) {
		if (IS_STATE(urdc, RDC_DISKQ_FAILED) ||
		    (IS_STATE(urdc, RDC_LOGGING) &&
		    !IS_STATE(urdc, RDC_QUEUING))) {
			mutex_exit(QLOCK(q));
			if (aio->handle)
				(void) nsc_free_buf(aio->handle);
			return (-1);
		}
		cv_wait(&q->qfullcv, QLOCK(q));

		if (IS_STATE(urdc, RDC_DISKQ_FAILED) ||
		    (IS_STATE(urdc, RDC_LOGGING) &&
		    !IS_STATE(urdc, RDC_QUEUING))) {
			mutex_exit(QLOCK(q));
			if (aio->handle)
				(void) nsc_free_buf(aio->handle);
			return (-1);
		}

	}

	SET_QSTATE(q, QTAILBUSY);

	if (aio->handle == NULL) {
		/* we're only going to write the header to the queue */
		numvecs = 2; /* kmem_alloc io header + null terminate */
		iofbas = FBA_LEN(sizeof (io_hdr));

	} else {
		/* find out how many vecs */
		numvecs = rdc_count_vecs(aio->handle->sb_vec) + 1;
		iofbas = aio->len + FBA_LEN(sizeof (io_hdr));
	}

	/*
	 * this, in conjunction with QTAILBUSY, will prevent
	 * premature dequeuing
	 */

	SET_LASTQTAIL(q, QTAIL(q));

	iohdr = (io_hdr *) kmem_zalloc(sizeof (io_hdr), KM_NOSLEEP);
	vec = (nsc_vec_t *) kmem_zalloc(sizeof (nsc_vec_t) * numvecs,
	    KM_NOSLEEP);

	if (!vec || !iohdr) {
		if (!vec) {
			cmn_err(CE_WARN, "!vec kmem alloc failed");
		} else {
			cmn_err(CE_WARN, "!iohdr kmem alloc failed");
		}
		if (vec)
			kmem_free(vec, sizeof (*vec));
		if (iohdr)
			kmem_free(iohdr, sizeof (*iohdr));
		CLR_QSTATE(q, QTAILBUSY);
		SET_LASTQTAIL(q, 0);
		mutex_exit(QLOCK(q));
		if (aio->handle)
			(void) nsc_free_buf(aio->handle);
		return (ENOMEM);
	}

	vec[numvecs - 1].sv_len = 0;
	vec[numvecs - 1].sv_addr = 0;

	/* now add the write itself */
	bp = aio->handle;

	for (i = 1, j = 0; bp && bp->sb_vec[j].sv_addr &&
	    i < numvecs; i++, j++) {
		vec[i].sv_len = bp->sb_vec[j].sv_len;
		vec[i].sv_addr = bp->sb_vec[j].sv_addr;
	}

retry:

	/* check for queue wrap, then check for overflow */
	if (IS_STATE(urdc, RDC_DISKQ_FAILED) ||
	    (IS_STATE(urdc, RDC_LOGGING) && !IS_STATE(urdc, RDC_QUEUING))) {
		kmem_free(iohdr, sizeof (*iohdr));
		kmem_free(vec, sizeof (*vec) * numvecs);
		CLR_QSTATE(q, QTAILBUSY);
		SET_LASTQTAIL(q, 0);
		if (IS_QSTATE(q, RDC_QFULL)) { /* wakeup blocked threads */
			CLR_QSTATE(q, RDC_QFULL);
			cv_broadcast(&q->qfullcv);
		}
		mutex_exit(QLOCK(q));
		if (aio->handle)
			(void) nsc_free_buf(aio->handle);

		return (-1);
	}

	if (QTAILSHLDWRAP(q, iofbas)) {
		/*
		 * just go back to the beginning of the disk
		 * it's not worth the trouble breaking up the write
		 */
#ifdef DEBUG_DISKQWRAP
		cmn_err(CE_NOTE, "!wrapping Q tail: " QDISPLAY(q));
#endif
		/*LINTED*/
		WRAPQTAIL(q);
	}

	/*
	 * prepend the write's metadata
	 */
	rdc_fill_ioheader(aio, iohdr, QTAIL(q));

	vec[0].sv_len = FBA_SIZE(1);
	vec[0].sv_addr = (uchar_t *)iohdr;

	/* check for tail < head */

	if (!(FITSONQ(q, iofbas))) {
		/*
		 * don't allow any more writes to start
		 */
		SET_QSTATE(q, RDC_QFULL);
		mutex_exit(QLOCK(q));

		if ((!group->rdc_writer) && !IS_STATE(urdc, RDC_LOGGING))
			(void) rdc_writer(krdc->index);

		delay(delay_time);
		q->throttle_delay += delay_time;
		retries++;
		delay_time *= 2; /* fairly aggressive */
		if ((retries >= 8) || (delay_time >= 256)) {
			delay_time = 2;
			if (print_msg) {
				cmn_err(CE_WARN, "!enqueue: disk queue %s full",
				    &urdc->disk_queue[0]);
				print_msg = 0;
#ifdef DEBUG
				cmn_err(CE_WARN, "!qinfo: " QDISPLAY(q));
#else
				cmn_err(CE_CONT, "!qinfo: " QDISPLAYND(q));
#endif
			}
			/*
			 * if this is a no-block queue, or this is a blocking
			 * queue that is not flushing. reset and log
			 */
			if ((QSTATE(q) & RDC_QNOBLOCK) ||
			    (IS_STATE(urdc, RDC_QUEUING))) {

				if (IS_STATE(urdc, RDC_QUEUING)) {
		cmn_err(CE_WARN, "!SNDR: disk queue %s full and 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);
				kmem_free(iohdr, sizeof (*iohdr));
				kmem_free(vec, sizeof (*vec) * numvecs);
				mutex_enter(QLOCK(q));
				CLR_QSTATE(q, QTAILBUSY | RDC_QFULL);
				cv_broadcast(&q->qfullcv);
				mutex_exit(QLOCK(q));
				SET_LASTQTAIL(q, 0);
				if (aio->handle)
					(void) nsc_free_buf(aio->handle);
				return (ENOMEM);
			}
		}

		mutex_enter(QLOCK(q));
		goto retry;

	}

	qtail = QTAIL(q);
#ifdef DEBUG_WRITER_UBERNOISE
	qhead = QHEAD(q);
#endif

	/* update tail pointer, nitems on queue and blocks on queue */
	INC_QTAIL(q, iofbas); /* increment tail over i/o size + ioheader size */
	INC_QNITEMS(q, 1);
	/* increment counter for i/o blocks only */
	INC_QBLOCKS(q, (iofbas - FBA_LEN(sizeof (io_hdr))));

	if (QNITEMS(q) > q->nitems_hwm)
		q->nitems_hwm = QNITEMS(q);
	if (QBLOCKS(q) > q->blocks_hwm)
		q->blocks_hwm = QBLOCKS(q);

	if (IS_QSTATE(q, RDC_QFULL)) {
		CLR_QSTATE(q, RDC_QFULL);
		cv_broadcast(&q->qfullcv);
	}

	mutex_exit(QLOCK(q));

	/*
	 * 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);
	 * }
	 */

	DTRACE_PROBE(rdc_diskq_rsrv);

	if (_rdc_rsrv_diskq(group)) {
		cmn_err(CE_WARN, "!rdc_enqueue: %s reserve failed",
		    &urdc->disk_queue[0]);
		rdc_fail_diskq(krdc, RDC_WAIT, RDC_DOLOG);
		kmem_free(iohdr, sizeof (*iohdr));
		kmem_free(vec, sizeof (*vec) * numvecs);
		mutex_enter(QLOCK(q));
		CLR_QSTATE(q, QTAILBUSY);
		SET_LASTQTAIL(q, 0);
		mutex_exit(QLOCK(q));
		if (aio->handle)
			(void) nsc_free_buf(aio->handle);
		return (-1);
	}

/* XXX for now do this, but later pre-alloc handle in enable/resume */

	DTRACE_PROBE(rdc_diskq_alloc_start);
	rc = nsc_alloc_buf(group->diskqfd, qtail, iofbas,
	    NSC_NOCACHE | NSC_WRITE | NSC_NODATA, &qbuf);

	DTRACE_PROBE(rdc_diskq_alloc_end);

	if (!RDC_SUCCESS(rc)) {
		cmn_err(CE_WARN, "!disk queue %s alloc failed(%d) %" NSC_SZFMT,
		    &urdc->disk_queue[0], rc, iofbas);
		rdc_fail_diskq(krdc, RDC_WAIT, RDC_DOLOG);
		rc = ENOMEM;
		goto fail;
	}
	/* move vec and write to queue */
	qbuf->sb_vec = &vec[0];

#ifdef DEBUG_WRITER_UBERNOISE

	cmn_err(CE_NOTE, "!about to write to queue, qbuf: %p, qhead: %d, "
	    "qtail: %d, len: %d contents: %c%c%c%c%c",
	    (void *) qbuf, qhead, qtail, iofbas,
	    qbuf->sb_vec[1].sv_addr[0],
	    qbuf->sb_vec[1].sv_addr[1],
	    qbuf->sb_vec[1].sv_addr[2],
	    qbuf->sb_vec[1].sv_addr[3],
	    qbuf->sb_vec[1].sv_addr[4]);
	cmn_err(CE_CONT, "!qinfo: " QDISPLAYND(q));

#endif

	DTRACE_PROBE2(rdc_diskq_nswrite_start, int, qtail, nsc_size_t, iofbas);
	rc = nsc_write(qbuf, qtail, iofbas, 0);
	DTRACE_PROBE2(rdc_diskq_nswrite_end, int, qtail, nsc_size_t, iofbas);

	if (!RDC_SUCCESS(rc)) {
		cmn_err(CE_WARN, "!disk queue %s write failed %d",
		    &urdc->disk_queue[0], rc);
		rdc_fail_diskq(krdc, RDC_WAIT, RDC_DOLOG);
		goto fail;

	}

	mutex_enter(QLOCK(q));

	SET_LASTQTAIL(q, 0);
	CLR_QSTATE(q, QTAILBUSY);

	mutex_exit(QLOCK(q));

fail:

	/*
	 * return what should be returned
	 * the aio is returned in _rdc_write after status is gathered.
	 */

	if (qbuf)
		qbuf->sb_vec = 0;
	(void) nsc_free_buf(qbuf);

	if (aio->handle)
		(void) nsc_free_buf(aio->handle);

	_rdc_rlse_diskq(group);
	DTRACE_PROBE(rdc_diskq_rlse);

	/* free the iohdr and the vecs */

	if (iohdr)
		kmem_free(iohdr, sizeof (*iohdr));
	if (vec)
		kmem_free(vec, sizeof (*vec) * numvecs);

	/* if no flusher running, start one */
	if ((!krdc->group->rdc_writer) && !IS_STATE(urdc, RDC_LOGGING))
		(void) rdc_writer(krdc->index);

	return (rc);
}

/*
 * place this on the pending list of io_hdr's out for flushing
 */
void
rdc_add_iohdr(io_hdr *header, rdc_group_t *group)
{
	disk_queue *q = NULL;
#ifdef DEBUG
	io_hdr *p;
#endif

	q = &group->diskq;

	/* paranoia */
	header->dat.next = NULL;

	mutex_enter(QLOCK(q));
#ifdef DEBUG /* AAAH! double flush!? */
	p = q->iohdrs;
	while (p) {
		if (p->dat.qpos == header->dat.qpos) {
			cmn_err(CE_WARN, "!ADDING DUPLICATE HEADER %" NSC_SZFMT,
			    p->dat.qpos);
			kmem_free(header, sizeof (*header));
			mutex_exit(QLOCK(q));
			return;
		}
		p = p->dat.next;
	}
#endif
	if (q->iohdrs == NULL) {
		q->iohdrs = q->hdr_last = header;
		q->hdrcnt = 1;
		mutex_exit(QLOCK(q));
		return;
	}

	q->hdr_last->dat.next = header;
	q->hdr_last = header;
	q->hdrcnt++;
	mutex_exit(QLOCK(q));
	return;

}

/*
 * mark an io header as flushed. If it is the qhead,
 * then update the qpointers
 * free the io_hdrs
 * called after the bitmap is cleared by flusher
 */
void
rdc_clr_iohdr(rdc_k_info_t *krdc, nsc_size_t qpos)
{
	rdc_group_t *group = krdc->group;
	disk_queue *q = NULL;
	io_hdr	*hp = NULL;
	io_hdr	*p = NULL;
	int found = 0;
	int cnt = 0;

#ifndef NSC_MULTI_TERABYTE
	ASSERT(qpos >= 0);	/* assertion to validate change for 64bit */
	if (qpos < 0) /* not a diskq offset */
		return;
#endif

	q = &group->diskq;
	mutex_enter(QLOCK(q));

	hp = p = q->iohdrs;

	/* find outstanding io_hdr */
	while (hp) {
		if (hp->dat.qpos == qpos) {
			found++;
			break;
		}
		cnt++;
		p = hp;
		hp = hp->dat.next;
	}

	if (!found) {
		if (RDC_BETWEEN(QHEAD(q), QNXTIO(q), qpos)) {
#ifdef DEBUG
			cmn_err(CE_WARN, "!iohdr already cleared? "
			"qpos %" NSC_SZFMT " cnt %d ", qpos, cnt);
			cmn_err(CE_WARN, "!Qinfo: " QDISPLAY(q));
#endif
			mutex_exit(QLOCK(q));
			return;
		}
		mutex_exit(QLOCK(q));
		return;
	}

	/* mark it as flushed */
	hp->dat.iostatus = RDC_IOHDR_DONE;

	/*
	 * if it is the head pointer, travel the list updating the queue
	 * pointers until the next unflushed is reached, freeing on the way.
	 */
	while (hp && (hp->dat.qpos == QHEAD(q)) &&
	    (hp->dat.iostatus == RDC_IOHDR_DONE)) {
#ifdef DEBUG_FLUSHER_UBERNOISE
		cmn_err(CE_NOTE, "!clr_iohdr info: magic %x type %d pos %d"
		    " qpos %d hpos %d len %d flag 0x%x iostatus %x setid %d",
		    hp->dat.magic, hp->dat.type, hp->dat.pos, hp->dat.qpos,
		    hp->dat.hpos, hp->dat.len, hp->dat.flag,
		    hp->dat.iostatus, hp->dat.setid);
#endif
		if (hp->dat.flag & RDC_NULL_BUF) {
			INC_QHEAD(q, FBA_LEN(sizeof (io_hdr)));
		} else {
			INC_QHEAD(q, FBA_LEN(sizeof (io_hdr)) + hp->dat.len);
			DEC_QBLOCKS(q, hp->dat.len);
		}

		DEC_QNITEMS(q, 1);

		if (QHEADSHLDWRAP(q)) { /* simple enough */
#ifdef DEBUG_DISKQWRAP
			cmn_err(CE_NOTE, "!wrapping Q head: " QDISPLAY(q));
#endif
			/*LINTED*/
			WRAPQHEAD(q);
		}

		/* get rid of the iohdr */
		if (hp == q->iohdrs) {
			q->iohdrs = hp->dat.next;
			kmem_free(hp, sizeof (*hp));
			hp = q->iohdrs;
		} else {
			if (hp == q->hdr_last)
				q->hdr_last = p;
			p->dat.next = hp->dat.next;
			kmem_free(hp, sizeof (*hp));
			hp = p->dat.next;
		}
		q->hdrcnt--;
	}

	if (QEMPTY(q) && !IS_QSTATE(q, RDC_QFULL) &&
	    !(IS_QSTATE(q, RDC_QDISABLEPEND))) {
#ifdef DEBUG_FLUSHER_UBERNOISE
		rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
		cmn_err(CE_NOTE, "!clr_iohdr: diskq %s empty, "
		    "resetting defaults", urdc->disk_queue);
#endif

		rdc_init_diskq_header(group, &q->disk_hdr);
		SET_QNXTIO(q, QHEAD(q));
	}

	/* wakeup any blocked enqueue threads */
	cv_broadcast(&q->qfullcv);
	mutex_exit(QLOCK(q));
}

/*
 * put in whatever useful checks we can on the io header
 */
int
rdc_iohdr_ok(io_hdr *hdr)
{
	if (hdr->dat.magic != RDC_IOHDR_MAGIC)
		goto bad;
	return (1);
bad:

#ifdef DEBUG
	cmn_err(CE_WARN, "!Bad io header magic %x type %d pos %" NSC_SZFMT
	    " hpos %" NSC_SZFMT " qpos %" NSC_SZFMT " len %" NSC_SZFMT
	    " flag %d iostatus %d setid %d", hdr->dat.magic,
	    hdr->dat.type, hdr->dat.pos, hdr->dat.hpos, hdr->dat.qpos,
	    hdr->dat.len, hdr->dat.flag, hdr->dat.iostatus, hdr->dat.setid);
#else
	cmn_err(CE_WARN, "!Bad io header retrieved");
#endif
	return (0);
}

/*
 * rdc_netqueue_insert()
 * add an item to a netqueue. No locks necessary as it should only
 * be used in a single threaded manor. If that changes, then
 * a lock or assertion should be done here
 */
void
rdc_netqueue_insert(rdc_aio_t *aio, net_queue *q)
{
	rdc_k_info_t *krdc = &rdc_k_info[aio->index];

	/* paranoid check for bit set */
	RDC_CHECK_BIT(krdc, aio->pos, aio->len);

	if (q->net_qhead == NULL) {
		q->net_qhead = q->net_qtail = aio;

	} else {
		q->net_qtail->next = aio;
		q->net_qtail = aio;
	}
	q->blocks += aio->len;
	q->nitems++;

	if (q->nitems > q->nitems_hwm) {
		q->nitems_hwm = q->nitems;
	}
	if (q->blocks > q->blocks_hwm) {
		q->nitems_hwm = q->blocks;
	}
}

/*
 * rdc_fill_aio(aio, hdr)
 * take the pertinent info from an io_hdr and stick it in
 * an aio, including seq number, abuf.
 */
void
rdc_fill_aio(rdc_group_t *grp, rdc_aio_t *aio, io_hdr *hdr, nsc_buf_t *abuf)
{
	if (hdr->dat.flag & RDC_NULL_BUF) {
		aio->handle = NULL;
	} else {
		aio->handle = abuf;
	}
	aio->qhandle = abuf;
	aio->pos = hdr->dat.pos;
	aio->qpos = hdr->dat.qpos;
	aio->len = hdr->dat.len;
	aio->flag = hdr->dat.flag;
	if ((aio->index = rdc_setid2idx(hdr->dat.setid)) < 0)
		return;
	mutex_enter(&grp->diskq.disk_qlock);
	if (grp->ra_queue.qfflags & RDC_QFILLSLEEP) {
		mutex_exit(&grp->diskq.disk_qlock);
		aio->seq = RDC_NOSEQ;
		return;
	}
	if (abuf && aio->qhandle) {
		abuf->sb_user++;
	}
	aio->seq = grp->seq++;
	if (grp->seq < aio->seq)
		grp->seq = RDC_NEWSEQ + 1;
	mutex_exit(&grp->diskq.disk_qlock);
	hdr->dat.iostatus = aio->seq;

}

#ifdef DEBUG
int maxaios_perbuf = 0;
int midaios_perbuf = 0;
int aveaios_perbuf = 0;
int totaios_perbuf = 0;
int buf2qcalls = 0;

void
calc_perbuf(int items)
{
	if (totaios_perbuf < 0) {
		maxaios_perbuf = 0;
		midaios_perbuf = 0;
		aveaios_perbuf = 0;
		totaios_perbuf = 0;
		buf2qcalls = 0;
	}

	if (items > maxaios_perbuf)
		maxaios_perbuf = items;
	midaios_perbuf = maxaios_perbuf / 2;
	totaios_perbuf += items;
	aveaios_perbuf = totaios_perbuf / buf2qcalls;
}
#endif

/*
 * rdc_discard_tmpq()
 * free up the passed temporary queue
 * NOTE: no cv's or mutexes have been initialized
 */
void
rdc_discard_tmpq(net_queue *q)
{
	rdc_aio_t *aio;

	if (q == NULL)
		return;

	while (q->net_qhead) {
		aio = q->net_qhead;
		q->net_qhead = q->net_qhead->next;
		if (aio->qhandle) {
			aio->qhandle->sb_user--;
			if (aio->qhandle->sb_user == 0) {
				rdc_fixlen(aio);
				(void) nsc_free_buf(aio->qhandle);
			}
		}
		kmem_free(aio, sizeof (*aio));
		q->nitems--;
	}
	kmem_free(q, sizeof (*q));

}

/*
 * rdc_diskq_buf2queue()
 * take a chunk of the diskq, parse it and assemble
 * a chain of rdc_aio_t's.
 * updates QNXTIO()
 */
net_queue *
rdc_diskq_buf2queue(rdc_group_t *grp, nsc_buf_t **abuf, int index)
{
	rdc_aio_t *aio = NULL;
	nsc_vec_t *vecp = NULL;
	uchar_t *vaddr = NULL;
	uchar_t *ioaddr = NULL;
	net_queue *netq = NULL;
	io_hdr  *hdr = NULL;
	nsc_buf_t *buf = *abuf;
	rdc_u_info_t *urdc = &rdc_u_info[index];
	rdc_k_info_t *krdc = &rdc_k_info[index];
	disk_queue *dq = &grp->diskq;
	net_queue *nq = &grp->ra_queue;
	int nullbuf = 0;
	nsc_off_t endobuf;
	nsc_off_t bufoff;
	int vlen;
	nsc_off_t fpos;
	long bufcnt = 0;
	int nullblocks = 0;
	int fail = 1;

	if (buf == NULL)
		return (NULL);

	netq = kmem_zalloc(sizeof (*netq), KM_NOSLEEP);
	if (netq == NULL) {
		cmn_err(CE_WARN, "!SNDR: unable to allocate net queue");
		return (NULL);
	}

	vecp = buf->sb_vec;
	vlen = vecp->sv_len;
	vaddr = vecp->sv_addr;
	bufoff = buf->sb_pos;
	endobuf = bufoff + buf->sb_len;

#ifdef DEBUG_FLUSHER_UBERNOISE
	cmn_err(CE_WARN, "!BUFFOFFENTER %d", bufoff);
#endif
	/* CONSTCOND */
	while (1) {
		if (IS_STATE(urdc, RDC_LOGGING) ||
		    (nq->qfflags & RDC_QFILLSLEEP)) {
			fail = 0;
			goto fail;
		}
#ifdef DEBUG_FLUSHER_UBERNOISE
		cmn_err(CE_WARN, "!BUFFOFF_0 %d", bufoff);
#endif

		if ((vaddr == NULL) || (vlen == 0))
			break;

		if (vlen <= 0) {
			vecp++;
			vaddr = vecp->sv_addr;
			vlen = vecp->sv_len;
			if (vaddr == NULL)
				break;
		}

		/* get the iohdr information */

		hdr = kmem_zalloc(sizeof (*hdr), KM_NOSLEEP);
		if (hdr == NULL) {
			cmn_err(CE_WARN,
			    "!SNDR: unable to alocate net queue header");
			goto fail;
		}

		ioaddr = (uchar_t *)hdr;

		bcopy(vaddr, ioaddr, sizeof (*hdr));

		if (!rdc_iohdr_ok(hdr)) {
			cmn_err(CE_WARN,
			    "!unable to retrieve i/o data from queue %s "
			    "at offset %" NSC_SZFMT " bp: %" NSC_SZFMT " bl: %"
			    NSC_SZFMT, urdc->disk_queue,
			    bufoff, buf->sb_pos, buf->sb_len);
#ifdef DEBUG_DISKQ
			cmn_err(CE_WARN, "!FAILING QUEUE state: %x",
			    rdc_get_vflags(urdc));
			cmn_err(CE_WARN, "!qinfo: " QDISPLAY(dq));
			cmn_err(CE_WARN, "!VADDR %p, IOADDR %p", vaddr, ioaddr);
			cmn_err(CE_WARN, "!BUF %p", buf);
#endif
			cmn_err(CE_WARN, "!qinfo: " QDISPLAYND(dq));

			goto fail;
		}

		nullbuf = hdr->dat.flag & RDC_NULL_BUF;

		bufoff += FBA_NUM(sizeof (*hdr));

		/* out of buffer, set nxtio to re read this last hdr */
		if (!nullbuf && ((bufoff + hdr->dat.len) > endobuf)) {
			break;
		}

		bufcnt += FBA_NUM(sizeof (*hdr));

		aio = kmem_zalloc(sizeof (*aio), KM_NOSLEEP);
		if (aio == NULL) {
			bufcnt -= FBA_NUM(sizeof (*hdr));
			cmn_err(CE_WARN, "!SNDR: net queue aio alloc failed");
			goto fail;
		}

		if (!nullbuf) {
			/* move to next iohdr in big buf */
			bufoff += hdr->dat.len;
			bufcnt += hdr->dat.len;
		}

		rdc_fill_aio(grp, aio, hdr, buf);

		if (aio->index < 0) {
			cmn_err(CE_WARN, "!Set id %d not found or no longer "
			    "enabled, failing disk queue", hdr->dat.setid);
			kmem_free(aio, sizeof (*aio));
			goto fail;
		}
		if (aio->seq == RDC_NOSEQ) {
			kmem_free(aio, sizeof (*aio));
			fail = 0;
			goto fail;
		}
		if (aio->handle == NULL)
			nullblocks += aio->len;

		rdc_add_iohdr(hdr, grp);
		hdr = NULL; /* don't accidentally free on break or fail */
		rdc_netqueue_insert(aio, netq);

		/* no more buffer, skip the below logic */
		if ((bufoff + FBA_NUM(sizeof (*hdr))) >= endobuf) {
			break;
		}

		fpos = bufoff - buf->sb_pos;
		vecp = buf->sb_vec;
		for (; fpos >= FBA_NUM(vecp->sv_len); vecp++)
			fpos -= FBA_NUM(vecp->sv_len);
		vlen = vecp->sv_len - FBA_SIZE(fpos);
		vaddr = vecp->sv_addr + FBA_SIZE(fpos);
		/* abuf = NULL; */

	}

	/* free extraneous header */
	if (hdr) {
		kmem_free(hdr, sizeof (*hdr));
		hdr = NULL;
	}

	/*
	 * probably won't happen, but if we didn't goto fail, but
	 * we don't contain anything meaningful.. return NULL
	 * and let the flusher or the sleep/wakeup routines
	 * decide
	 */
	if (netq && netq->nitems == 0) {
		kmem_free(netq, sizeof (*netq));
		return (NULL);
	}

#ifdef DEBUG
	buf2qcalls++;
	calc_perbuf(netq->nitems);
#endif
	if (IS_STATE(urdc, RDC_LOGGING) ||
	    nq->qfflags & RDC_QFILLSLEEP) {
		fail = 0;
		goto fail;
	}

	mutex_enter(QLOCK(dq));
	INC_QNXTIO(dq, bufcnt);
	mutex_exit(QLOCK(dq));

	netq->net_qtail->orig_len = nullblocks; /* overload */

	return (netq);

fail:

	if (hdr) {
		kmem_free(hdr, sizeof (*hdr));
	}

	if (netq) {
		if (netq->nitems > 0) {
			/* the never can happen case ... */
			if ((netq->nitems == 1) &&
			    (netq->net_qhead->handle == NULL))
				(void) nsc_free_buf(buf);
				*abuf = NULL;

		}
		rdc_discard_tmpq(netq);
	}

	mutex_enter(QLOCK(dq));
	rdc_dump_iohdrs(dq);
	mutex_exit(QLOCK(dq));

	if (fail) { /* real failure, not just state change */
#ifdef DEBUG
		cmn_err(CE_WARN, "!rdc_diskq_buf2queue: failing disk queue %s",
		    urdc->disk_queue);
#endif
		rdc_fail_diskq(krdc, RDC_NOWAIT, RDC_DOLOG);
	}

	return (NULL);

}

/*
 * rdc_diskq_unqueue
 * remove one chunk from the diskq belonging to
 * rdc_k_info[index]
 * updates the head and tail pointers in the disk header
 * but does not write. The header should be written on ack
 * flusher should free whatever..
 */
rdc_aio_t *
rdc_diskq_unqueue(int index)
{
	int rc, rc1, rc2;
	nsc_off_t qhead;
	int nullhandle = 0;
	io_hdr *iohdr;
	rdc_aio_t *aio = NULL;
	nsc_buf_t *buf = NULL;
	nsc_buf_t *abuf = NULL;
	rdc_group_t *group = NULL;
	disk_queue *q = NULL;
	rdc_k_info_t *krdc = &rdc_k_info[index];
	rdc_u_info_t *urdc = &rdc_u_info[index];

	group = krdc->group;
	q = &group->diskq;

	if (group->diskqfd == NULL) /* we've been disabled */
		return (NULL);

	aio = kmem_zalloc(sizeof (*aio), KM_NOSLEEP);
	if (!aio) {
		return (NULL);
	}

	iohdr = kmem_zalloc(sizeof (*iohdr), KM_NOSLEEP);
	if (!iohdr) {
		kmem_free(aio, sizeof (*aio));
		return (NULL);
	}

	mutex_enter(QLOCK(q));
	rdc_set_qbusy(q); /* make sure no one disables the queue */
	mutex_exit(QLOCK(q));

	DTRACE_PROBE(rdc_diskq_unq_rsrv);

	if (_rdc_rsrv_diskq(group)) {
		cmn_err(CE_WARN, "!rdc_unqueue: %s reserve failed",
		    urdc->disk_queue);
		goto fail;
	}

	mutex_enter(QHEADLOCK(q));
	mutex_enter(QLOCK(q));

	if (IS_STATE(urdc, RDC_DISKQ_FAILED) || IS_STATE(urdc, RDC_LOGGING)) {
		rdc_clr_qbusy(q);
		mutex_exit(QLOCK(q));
		mutex_exit(QHEADLOCK(q));
		kmem_free(aio, sizeof (*aio));
		kmem_free(iohdr, sizeof (*iohdr));
		return (NULL);
	}

	if (QNXTIOSHLDWRAP(q)) {
#ifdef DEBUG_DISKQWRAP
		cmn_err(CE_NOTE, "!wrapping Q nxtio: " QDISPLAY(q));
#endif
		/*LINTED*/
		WRAPQNXTIO(q);
	}

	/* read the metainfo at q->nxt_io first */
	if (QNXTIO(q) == QTAIL(q)) { /* empty */

		_rdc_rlse_diskq(group);
		if (q->lastio->handle)
			(void) nsc_free_buf(q->lastio->handle);
		bzero(&(*q->lastio), sizeof (*q->lastio));

		mutex_exit(QHEADLOCK(q));
		rdc_clr_qbusy(q);
		mutex_exit(QLOCK(q));
		kmem_free(aio, sizeof (*aio));
		kmem_free(iohdr, sizeof (*iohdr));
		return (NULL);
	}

	qhead = QNXTIO(q);

	/*
	 * have to drop the lock here, sigh. Cannot block incoming io
	 * we have to wait until after this read to find out how
	 * much to increment QNXTIO. Might as well grab the seq then too
	 */

	while ((qhead == LASTQTAIL(q)) && (IS_QSTATE(q, QTAILBUSY))) {
		mutex_exit(QLOCK(q));
#ifdef DEBUG_DISKQ
		cmn_err(CE_NOTE, "!Qtail busy delay lastqtail: %d", qhead);
#endif
		delay(5);
		mutex_enter(QLOCK(q));
	}
	mutex_exit(QLOCK(q));

	DTRACE_PROBE(rdc_diskq_iohdr_read_start);

	rc = rdc_ns_io(group->diskqfd, NSC_READ, qhead,
	    (uchar_t *)iohdr, FBA_SIZE(1));

	DTRACE_PROBE(rdc_diskq_iohdr_read_end);

	if (!RDC_SUCCESS(rc) || !rdc_iohdr_ok(iohdr)) {
		cmn_err(CE_WARN, "!unable to retrieve i/o data from queue %s"
		    " at offset %" NSC_SZFMT " rc %d", urdc->disk_queue,
		    qhead, rc);
#ifdef DEBUG_DISKQ
		cmn_err(CE_WARN, "!qinfo: " QDISPLAY(q));
#endif
		mutex_exit(QHEADLOCK(q));
		goto fail;
	}

/* XXX process buffer here, creating rdc_aio_t's */

	mutex_enter(QLOCK(q));
	/* update the next pointer */
	if (iohdr->dat.flag == RDC_NULL_BUF) {
		INC_QNXTIO(q, FBA_LEN(sizeof (io_hdr)));
		nullhandle = 1;
	} else {
		INC_QNXTIO(q, (FBA_LEN(sizeof (io_hdr)) + iohdr->dat.len));
	}

	aio->seq = group->seq++;
	if (group->seq < aio->seq)
		group->seq = RDC_NEWSEQ + 1;

	mutex_exit(QLOCK(q));
	mutex_exit(QHEADLOCK(q));

#ifdef DEBUG_FLUSHER_UBERNOISE
	p = &iohdr->dat;
	cmn_err(CE_NOTE, "!unqueued iohdr from %d pos: %d len: %d flag: %d "
	    "iostatus: %d setid: %d time: %d", qhead, p->pos, p->len,
	    p->flag, p->iostatus, p->setid, p->time);
#endif

	if (nullhandle) /* nothing to get from queue */
		goto nullbuf;

	/* now that we know how much to get (iohdr.dat.len), get it */
	DTRACE_PROBE(rdc_diskq_unq_allocbuf1_start);

	rc = nsc_alloc_buf(group->diskqfd, qhead + 1, iohdr->dat.len,
	    NSC_NOCACHE | NSC_READ, &buf);

	DTRACE_PROBE(rdc_diskq_unq_allocbuf1_end);

	/* and get somewhere to keep it for a bit */
	DTRACE_PROBE(rdc_diskq_unq_allocbuf2_start);

	rc1 = nsc_alloc_abuf(qhead + 1, iohdr->dat.len, 0, &abuf);

	DTRACE_PROBE(rdc_diskq_unq_allocbuf2_end);

	if (!RDC_SUCCESS(rc) || !RDC_SUCCESS(rc1)) { /* uh-oh */
		cmn_err(CE_WARN, "!disk queue %s read failure",
		    urdc->disk_queue);
		goto fail;
	}

	/* move it on over... */
	rc2 = nsc_copy(buf, abuf, qhead + 1, qhead + 1, iohdr->dat.len);

	if (!RDC_SUCCESS(rc2)) {
#ifdef DEBUG
		cmn_err(CE_WARN, "!nsc_copy failed for diskq unqueue");
#endif
		goto fail;
	}

	/* let go of the real buf, we've got the abuf  */
	(void) nsc_free_buf(buf);
	buf = NULL;

	aio->handle = abuf;
	/* Hack in the original sb_pos */
	aio->handle->sb_pos = iohdr->dat.hpos;

	/* skip the RDC_HANDLE_LIMITS check */
	abuf->sb_user |= RDC_DISKQUE;

nullbuf:
	if (nullhandle) {
		aio->handle = NULL;
	}

	/* set up the rest of the aio values, seq set above ... */
	aio->pos = iohdr->dat.pos;
	aio->qpos = iohdr->dat.qpos;
	aio->len = iohdr->dat.len;
	aio->flag = iohdr->dat.flag;
	aio->index = rdc_setid2idx(iohdr->dat.setid);
	if (aio->index < 0) { /* uh-oh */
#ifdef DEBUG
		cmn_err(CE_WARN, "!rdc_diskq_unqueue: index < 0");
#endif
		goto fail;
	}


#ifdef DEBUG_FLUSHER_UBERNOISE_STAMP
	h = &q->disk_hdr.h;
	cmn_err(CE_NOTE, "!stamping diskq header:\n"
	    "magic: %x\nstate: %d\nhead_offset: %d\n"
	    "tail_offset: %d\ndisk_size: %d\nnitems: %d\nblocks: %d\n",
	    h->magic, h->state, h->head_offset, h->tail_offset,
	    h->disk_size, h->nitems, h->blocks);
#endif

	_rdc_rlse_diskq(group);

	mutex_enter(QLOCK(q));
	rdc_clr_qbusy(q);
	mutex_exit(QLOCK(q));

	DTRACE_PROBE(rdc_diskq_unq_rlse);

	iohdr->dat.iostatus = aio->seq;
	rdc_add_iohdr(iohdr, group);

#ifdef DEBUG_FLUSHER_UBERNOISE
	if (!nullhandle) {
		cmn_err(CE_NOTE, "!UNQUEUING, %p"
		    " contents: %c%c%c%c%c pos: %d len: %d",
		    (void *)aio->handle,
		    aio->handle->sb_vec[0].sv_addr[0],
		    aio->handle->sb_vec[0].sv_addr[1],
		    aio->handle->sb_vec[0].sv_addr[2],
		    aio->handle->sb_vec[0].sv_addr[3],
		    aio->handle->sb_vec[0].sv_addr[4],
		    aio->handle->sb_pos, aio->handle->sb_len);
	} else {
		cmn_err(CE_NOTE, "!UNQUEUING, NULL " QDISPLAY(q));
	}
	cmn_err(CE_NOTE, "!qinfo: " QDISPLAY(q));
#endif

	return (aio);

fail:
	if (aio)
		kmem_free(aio, sizeof (*aio));
	if (iohdr)
		kmem_free(iohdr, sizeof (*iohdr));
	if (buf)
		(void) nsc_free_buf(buf);
	if (abuf)
		(void) nsc_free_buf(abuf);

	_rdc_rlse_diskq(group);
#ifdef DEBUG
	cmn_err(CE_WARN, "!diskq_unqueue: failing diskq");
#endif
	mutex_enter(QLOCK(q));
	rdc_clr_qbusy(q);
	mutex_exit(QLOCK(q));

	rdc_fail_diskq(krdc, RDC_NOWAIT, RDC_DOLOG);

	return (NULL);
}

int
rdc_diskq_inuse(rdc_set_t *set, char *diskq)
{
	rdc_u_info_t *urdc;
	char *group;
	int index;

	group = set->group_name;

	ASSERT(MUTEX_HELD(&rdc_conf_lock));

	if ((rdc_lookup_bitmap(diskq) >= 0) ||
	    (rdc_lookup_configured(diskq) >= 0)) {
		return (1);
	}
	for (index = 0; index < rdc_max_sets; index++) {
		urdc = &rdc_u_info[index];

		if (!IS_ENABLED(urdc))
			continue;

		/* same diskq different group */
		if ((strcmp(urdc->disk_queue, diskq) == 0) &&
		    (urdc->group_name[0] == '\0' ||
		    strcmp(urdc->group_name, group))) {
			return (1);
		}
	}
	/* last, but not least, lets see if someone is getting really funky */
	if ((strcmp(set->disk_queue, set->primary.file) == 0) ||
	    (strcmp(set->disk_queue, set->primary.bitmap) == 0)) {
		return (1);
	}

	return (0);

}

#ifdef DEBUG
int maxlen = 0;
int avelen = 0;
int totalen = 0;
int lencalls = 0;

void
update_lenstats(int len)
{
	if (lencalls == 0) {
		lencalls = 1;
		avelen = 0;
		maxlen = 0;
		totalen = 0;
	}

	if (len > maxlen)
		maxlen = len;
	totalen += len;
	avelen = totalen / lencalls;
}
#endif

/*
 * rdc_calc_len()
 * returns the size of the diskq that can be read for dequeuing
 * always <= RDC_MAX_DISKQREAD
 */
int
rdc_calc_len(rdc_k_info_t *krdc, disk_queue *dq)
{
	nsc_size_t len = 0;

	ASSERT(MUTEX_HELD(QLOCK(dq)));

	/* ---H-----N-----T--- */
	if (QNXTIO(dq) < QTAIL(dq)) {

		len = min(RDC_MAX_DISKQREAD, QTAIL(dq) - QNXTIO(dq));

	/* ---T-----H-----N--- */
	} else if (QNXTIO(dq) > QTAIL(dq)) {
		if (QWRAP(dq)) {
			len = min(RDC_MAX_DISKQREAD, QWRAP(dq) - QNXTIO(dq));
		} else { /* should never happen */
			len = min(RDC_MAX_DISKQREAD, QSIZE(dq) - QNXTIO(dq));
		}
	} else if (QNXTIO(dq) == QTAIL(dq)) {
		if (QWRAP(dq) && !IS_QSTATE(dq, QNXTIOWRAPD))
			len = min(RDC_MAX_DISKQREAD, QWRAP(dq) - QNXTIO(dq));
	}

	len = min(len, krdc->maxfbas);

#ifdef DEBUG
	lencalls++;
	update_lenstats(len);
#endif

	return ((int)len);
}

/*
 * lie a little if we can, so we don't get tied up in
 * _nsc_wait_dbuf() on the next read. sb_len MUST be
 * restored before nsc_free_buf() however, or we will
 * be looking at memory leak city..
 * so update the entire queue with the info as well
 * and the one that ends up freeing it, can fix the len
 * IMPORTANT: This assumes that we are not cached, in
 * 3.2 caching was turned off for data volumes, if that
 * changes, then this must too
 */
void
rdc_trim_buf(nsc_buf_t *buf, net_queue *q)
{
	rdc_aio_t *p;
	int len;

	if (buf == NULL || q == NULL)
		return;

	if (q && (buf->sb_len >
	    (q->blocks + q->nitems - q->net_qtail->orig_len))) {
		len = buf->sb_len;
		buf->sb_len = (q->blocks + q->nitems - q->net_qtail->orig_len);
	}

	p = q->net_qhead;
	do {
		p->orig_len = len;
		p = p->next;

	} while (p);

}

/*
 * rdc_read_diskq_buf()
 * read a large as possible chunk of the diskq into a nsc_buf_t
 * and convert it to a net_queue of rdc_aio_t's to be appended
 * to the group's netqueue
 */
net_queue *
rdc_read_diskq_buf(int index)
{
	nsc_buf_t *buf = NULL;
	net_queue *tmpnq = NULL;
	disk_queue *dq = NULL;
	rdc_k_info_t *krdc = &rdc_k_info[index];
	rdc_u_info_t *urdc = &rdc_u_info[index];
	rdc_group_t *group = krdc->group;
	net_queue *nq = &group->ra_queue;
	int len = 0;
	int rc;
	int fail = 0;
	int offset = 0;

	if (group == NULL || group->diskqfd == NULL) {
		DTRACE_PROBE(rdc_read_diskq_buf_bail1);
		return (NULL);
	}

	dq = &group->diskq;

	mutex_enter(QLOCK(dq));
	rdc_set_qbusy(dq); /* prevent disables on the queue */
	mutex_exit(QLOCK(dq));

	if (_rdc_rsrv_diskq(group)) {
		cmn_err(CE_WARN, "!rdc_readdiskqbuf: %s reserve failed",
		    urdc->disk_queue);
		mutex_enter(QLOCK(dq));
		rdc_clr_qbusy(dq); /* prevent disables on the queue */
		mutex_exit(QLOCK(dq));
		return (NULL);
	}

	mutex_enter(QHEADLOCK(dq));
	mutex_enter(QLOCK(dq));

	if (IS_STATE(urdc, RDC_DISKQ_FAILED) ||
	    IS_STATE(urdc, RDC_LOGGING) ||
	    (nq->qfflags & RDC_QFILLSLEEP)) {
		mutex_exit(QLOCK(dq));
		mutex_exit(QHEADLOCK(dq));
		DTRACE_PROBE(rdc_read_diskq_buf_bail2);
		goto done;
	}

	/*
	 * real corner case here, we need to let the flusher wrap first.
	 * we've gotten too far ahead, so just delay and try again
	 */
	if (IS_QSTATE(dq, QNXTIOWRAPD) && AUXQWRAP(dq)) {
		mutex_exit(QLOCK(dq));
		mutex_exit(QHEADLOCK(dq));
		goto done;
	}

	if (QNXTIOSHLDWRAP(dq)) {
#ifdef DEBUG_DISKQWRAP
		cmn_err(CE_NOTE, "!wrapping Q nxtio: " QDISPLAY(dq));
#endif
		/*LINTED*/
		WRAPQNXTIO(dq);
	}

	/* read the metainfo at q->nxt_io first */
	if (!QNITEMS(dq)) { /* empty */

		if (dq->lastio->handle)
			(void) nsc_free_buf(dq->lastio->handle);
		bzero(&(*dq->lastio), sizeof (*dq->lastio));
		mutex_exit(QLOCK(dq));
		mutex_exit(QHEADLOCK(dq));
		DTRACE_PROBE(rdc_read_diskq_buf_bail3);
		goto done;
	}


	len = rdc_calc_len(krdc, dq);

	if ((len <= 0) || (IS_STATE(urdc, RDC_LOGGING)) ||
	    (IS_STATE(urdc, RDC_DISKQ_FAILED)) ||
	    (nq->qfflags & RDC_QFILLSLEEP)) {
		mutex_exit(QLOCK(dq));
		mutex_exit(QHEADLOCK(dq));
		/*
		 * a write could be trying to get on the queue, or if
		 * the queue is really really small, a complete image
		 * of it could be on the net queue waiting for flush.
		 * the latter being a fairly stupid scenario and a gross
		 * misconfiguration.. but what the heck, why make the thread
		 * thrash around.. just pause a little here.
		 */
		if (len <= 0)
			delay(50);

		DTRACE_PROBE3(rdc_read_diskq_buf_bail4, int, len,
		    int, rdc_get_vflags(urdc), int, nq->qfflags);

		goto done;
	}

	DTRACE_PROBE2(rdc_calc_len, int, len, int, (int)QNXTIO(dq));

#ifdef DEBUG_FLUSHER_UBERNOISE
	cmn_err(CE_WARN, "!CALC_LEN(%d) h:%d n%d t%d, w%d",
	    len, QHEAD(dq), QNXTIO(dq), QTAIL(dq), QWRAP(dq));
	cmn_err(CE_CONT, "!qinfo: " QDISPLAYND(dq));
#endif
	SET_QCOALBOUNDS(dq, QNXTIO(dq) + len);

	while ((LASTQTAIL(dq) > 0) && !QWRAP(dq) &&
	    ((QNXTIO(dq) + len) >= LASTQTAIL(dq)) &&
	    (IS_QSTATE(dq, QTAILBUSY))) {
		mutex_exit(QLOCK(dq));

#ifdef DEBUG_FLUSHER_UBERNOISE
		cmn_err(CE_NOTE, "!Qtail busy delay nxtio %d len %d "
		    "lastqtail: %d", QNXTIO(dq), len, LASTQTAIL(dq));
#endif
		delay(20);
		mutex_enter(QLOCK(dq));
	}

	offset = QNXTIO(dq);

	/*
	 * one last check to see if we have gone logging, or should.
	 * we may have released the mutex above, so check again
	 */
	if ((IS_STATE(urdc, RDC_LOGGING)) ||
	    (IS_STATE(urdc, RDC_DISKQ_FAILED)) ||
	    (nq->qfflags & RDC_QFILLSLEEP)) {
		mutex_exit(QLOCK(dq));
		mutex_exit(QHEADLOCK(dq));
		goto done;
	}

	mutex_exit(QLOCK(dq));
	mutex_exit(QHEADLOCK(dq));

	DTRACE_PROBE2(rdc_buf2q_preread, int, offset, int, len);

	rc = nsc_alloc_buf(group->diskqfd, offset, len,
	    NSC_NOCACHE | NSC_READ, &buf);

	if (!RDC_SUCCESS(rc)) {
		cmn_err(CE_WARN, "!disk queue %s read failure pos %" NSC_SZFMT
		    " len %d", urdc->disk_queue, QNXTIO(dq), len);
		fail++;
		buf = NULL;
		DTRACE_PROBE(rdc_read_diskq_buf_bail5);
		goto done;
	}

	DTRACE_PROBE2(rdc_buf2q_postread, int, offset, nsc_size_t, buf->sb_len);

	/*
	 * convert buf to a net_queue. buf2queue will
	 * update the QNXTIO pointer for us, based on
	 * the last readable queue item
	 */
	tmpnq = rdc_diskq_buf2queue(group, &buf, index);

#ifdef DEBUG_FLUSHER_UBERNOISE
	cmn_err(CE_NOTE, "!QBUF p: %d l: %d p+l: %d users: %d qblocks: %d ",
	    "qitems: %d WASTED: %d", buf->sb_pos, buf->sb_len,
	    buf->sb_pos+buf->sb_len, buf->sb_user, tmpnq?tmpnq->blocks:-1,
	    tmpnq?tmpnq->nitems:-1,
	    tmpnq?((buf->sb_len-tmpnq->nitems) - tmpnq->blocks):-1);
#endif

	DTRACE_PROBE3(rdc_buf2que_returned, net_queue *, tmpnq?tmpnq:0,
	    uint64_t, tmpnq?tmpnq->nitems:0,
	    uint_t, tmpnq?tmpnq->net_qhead->seq:0);
done:

	/* we don't need to retain the buf */
	if (tmpnq == NULL)
		if (buf) {
			(void) nsc_free_buf(buf);
			buf = NULL;
		}

	rdc_trim_buf(buf, tmpnq);

	mutex_enter(QLOCK(dq));
	rdc_clr_qbusy(dq);
	mutex_exit(QLOCK(dq));

	_rdc_rlse_diskq(group);

	if (fail) {
		rdc_fail_diskq(krdc, RDC_NOWAIT, RDC_DOLOG);
		tmpnq = NULL;
	}

	return (tmpnq);
}

/*
 * rdc_dequeue()
 * removes the head of the memory queue
 */
rdc_aio_t *
rdc_dequeue(rdc_k_info_t *krdc, int *rc)
{
	net_queue *q = &krdc->group->ra_queue;
	disk_queue *dq = &krdc->group->diskq;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
	rdc_aio_t *aio;

	*rc = 0;

	if (q == NULL)
		return (NULL);

	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_dequeue(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);

		if ((!IS_STATE(urdc, RDC_LOGGING)) &&
		    (!(q->qfflags & RDC_QFILLSLEEP)) &&
		    (!IS_STATE(urdc, RDC_SYNCING)) && (QNITEMS(dq) > 0)) {
			*rc = EAGAIN;
		}

		goto done;
	}

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

#ifdef DEBUG
	if (q->net_qhead == NULL) {
		if (q->nitems != 0 || q->blocks != 0 || q->net_qtail != 0) {
			cmn_err(CE_PANIC, "rdc_dequeue(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);
		}
	}
#endif
	mutex_exit(&q->net_qlock);
done:

	mutex_enter(&q->net_qlock);

	if (rdc_qfill_shldwakeup(krdc))
		cv_broadcast(&q->qfcv);

	/*
	 * clear EAGAIN if
	 * logging or q filler thread is sleeping or stopping altogether
	 * or if q filler thread is dead already
	 * or if syncing, this will return a null aio, with no error code set
	 * telling the flusher to die
	 */
	if (*rc == EAGAIN) {
		if (IS_STATE(urdc, RDC_LOGGING) ||
		    (q->qfflags & (RDC_QFILLSLEEP | RDC_QFILLSTOP)) ||
		    (IS_QSTATE(dq, (RDC_QDISABLEPEND | RDC_STOPPINGFLUSH))) ||
		    (q->qfill_sleeping == RDC_QFILL_DEAD) ||
		    (IS_STATE(urdc, RDC_SYNCING)))
			*rc = 0;
	}

	mutex_exit(&q->net_qlock);

	return (aio);

}

/*
 * rdc_qfill_shldsleep()
 * returns 1 if the qfilling code should cv_wait() 0 if not.
 * reasons for going into cv_wait();
 * there is nothing in the diskq to flush to mem.
 * the memory queue has gotten too big and needs more flushing attn.
 */
int
rdc_qfill_shldsleep(rdc_k_info_t *krdc)
{
	net_queue *nq = &krdc->group->ra_queue;
	disk_queue *dq = &krdc->group->diskq;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];

	ASSERT(MUTEX_HELD(&nq->net_qlock));

	if (!RDC_IS_DISKQ(krdc->group))
		return (1);

	if (nq->qfflags & RDC_QFILLSLEEP) {
#ifdef DEBUG_DISKQ_NOISY
	cmn_err(CE_NOTE, "!Sleeping diskq->memq flusher: QFILLSLEEP idx: %d",
	    krdc->index);
#endif
		return (1);
	}

	if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING)) {
#ifdef DEBUG_DISKQ_NOISY
	cmn_err(CE_NOTE, "!Sleeping diskq->memq flusher: Sync|Log (0x%x)"
	    " idx: %d", rdc_get_vflags(urdc), urdc->index);
#endif
		return (1);
	}

	mutex_enter(QLOCK(dq));
	if ((QNXTIO(dq) == QTAIL(dq)) && !IS_QSTATE(dq, RDC_QFULL)) {
#ifdef DEBUG_DISKQ_NOISY
		cmn_err(CE_NOTE, "!Sleeping diskq->memq flusher: QEMPTY");
#endif
		mutex_exit(QLOCK(dq));
		return (1);
	}
	mutex_exit(QLOCK(dq));

	if (nq->blocks >= RDC_MAX_QBLOCKS) {
		nq->hwmhit = 1;
		/* stuck flushers ? */
#ifdef DEBUG_DISKQ_NOISY
		cmn_err(CE_NOTE, "!Sleeping diskq->memq flusher: memq full:"
		    " seq: %d seqack %d", krdc->group->seq,
		    krdc->group->seqack);
#endif
		return (1);
	}

	return (0);
}

/*
 * rdc_join_netqueues(a, b)
 * appends queue b to queue a updating all the queue info
 * as it is assumed queue a is the important one,
 * it's mutex must be held. no one can add to queue b
 */
void
rdc_join_netqueues(net_queue *q, net_queue *tmpq)
{
	ASSERT(MUTEX_HELD(&q->net_qlock));

	if (q->net_qhead == NULL) { /* empty */
#ifdef DEBUG
		if (q->blocks != 0 || q->nitems != 0) {
			cmn_err(CE_PANIC, "rdc filler: q %p, qhead 0, "
			    " q blocks %" NSC_SZFMT ", nitems %" NSC_SZFMT,
			    (void *) q, q->blocks, q->nitems);
		}
#endif
		q->net_qhead = tmpq->net_qhead;
		q->net_qtail = tmpq->net_qtail;
		q->nitems = tmpq->nitems;
		q->blocks = tmpq->blocks;
	} else {
		q->net_qtail->next = tmpq->net_qhead;
		q->net_qtail = tmpq->net_qtail;
		q->nitems += tmpq->nitems;
		q->blocks += tmpq->blocks;
	}

	if (q->nitems > q->nitems_hwm) {
		q->nitems_hwm = q->nitems;
	}

	if (q->blocks > q->blocks_hwm) {
		q->blocks_hwm = q->blocks;
	}
}

/*
 * rdc_qfiller_thr() single thread that moves
 * data from the diskq to a memory queue for
 * the flusher to pick up.
 */
void
rdc_qfiller_thr(rdc_k_info_t *krdc)
{
	rdc_group_t *grp = krdc->group;
	rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
	net_queue *q = &grp->ra_queue;
	net_queue *tmpq = NULL;
	int index = krdc->index;

	q->qfill_sleeping = RDC_QFILL_AWAKE;
	while (!(q->qfflags & RDC_QFILLSTOP)) {
		if (!RDC_IS_DISKQ(grp) ||
		    IS_STATE(urdc, RDC_LOGGING) ||
		    IS_STATE(urdc, RDC_DISKQ_FAILED) ||
		    (q->qfflags & RDC_QFILLSLEEP)) {
			goto nulltmpq;
		}

		DTRACE_PROBE(qfiller_top);
		tmpq = rdc_read_diskq_buf(index);

		if (tmpq == NULL)
			goto nulltmpq;

		if ((q->qfflags & RDC_QFILLSLEEP) ||
		    IS_STATE(urdc, RDC_LOGGING)) {
			rdc_discard_tmpq(tmpq);
			goto nulltmpq;
		}

		mutex_enter(&q->net_qlock);

		/* race with log, redundant yet paranoid */
		if ((q->qfflags & RDC_QFILLSLEEP) ||
		    IS_STATE(urdc, RDC_LOGGING)) {
			rdc_discard_tmpq(tmpq);
			mutex_exit(&q->net_qlock);
			goto nulltmpq;
		}


		rdc_join_netqueues(q, tmpq);
		kmem_free(tmpq, sizeof (*tmpq));
		tmpq = NULL;

		mutex_exit(&q->net_qlock);
nulltmpq:
		/*
		 * sleep for a while if we can.
		 * the enqueuing or flushing code will
		 * wake us if if necessary.
		 */
		mutex_enter(&q->net_qlock);
		while (rdc_qfill_shldsleep(krdc)) {
			q->qfill_sleeping = RDC_QFILL_ASLEEP;
			DTRACE_PROBE(qfiller_sleep);
			cv_wait(&q->qfcv, &q->net_qlock);
			DTRACE_PROBE(qfiller_wakeup);
			q->qfill_sleeping = RDC_QFILL_AWAKE;
			if (q->qfflags & RDC_QFILLSTOP) {
#ifdef DEBUG_DISKQ
			cmn_err(CE_NOTE,
			    "!rdc_qfiller_thr: recieved kill signal");
#endif
				mutex_exit(&q->net_qlock);
				goto done;
			}
		}
		mutex_exit(&q->net_qlock);

	DTRACE_PROBE(qfiller_bottom);
	}
done:
	DTRACE_PROBE(qfiller_done);
	q->qfill_sleeping = RDC_QFILL_DEAD; /* the big sleep */

#ifdef DEBUG
	cmn_err(CE_NOTE, "!rdc_qfiller_thr stopping");
#endif
	q->qfflags &= ~RDC_QFILLSTOP;

}

int
_rdc_add_diskq(int index, char *diskq)
{
	rdc_k_info_t *krdc, *kp;
	rdc_u_info_t *urdc, *up;
	rdc_group_t *group;
	int rc;

	krdc = &rdc_k_info[index];
	urdc = &rdc_u_info[index];
	group = krdc->group;

	if (!diskq || urdc->disk_queue[0]) { /* how'd that happen? */
#ifdef DEBUG
		cmn_err(CE_WARN, "!NULL diskq in _rdc_add_diskq");
#endif
		rc = -1;
		goto fail;
	}

	/* if the enable fails, this is bzero'ed */
	(void) strncpy(urdc->disk_queue, diskq, NSC_MAXPATH);
	group->flags &= ~RDC_MEMQUE;
	group->flags |= RDC_DISKQUE;

#ifdef DEBUG
	cmn_err(CE_NOTE, "!adding diskq to group %s", urdc->group_name);
#endif
	mutex_enter(&rdc_conf_lock);
	rc = rdc_enable_diskq(krdc);
	mutex_exit(&rdc_conf_lock);

	if (rc == RDC_EQNOADD) {
		goto fail;
	}

	RDC_ZERO_BITREF(krdc);
	for (kp = krdc->group_next; kp != krdc; kp = kp->group_next) {
		up = &rdc_u_info[kp->index];
		(void) strncpy(up->disk_queue, diskq, NSC_MAXPATH);
		/* size lives in the diskq structure, already set by enable */
		RDC_ZERO_BITREF(kp);
	}

fail:
	return (rc);

}

/*
 * add a diskq to an existing set/group
 */
int
rdc_add_diskq(rdc_config_t *uparms, spcs_s_info_t kstatus)
{
	char *diskq;
	int rc;
	int index;
	rdc_k_info_t *krdc, *this;
	rdc_u_info_t *urdc;
	rdc_group_t *group;
	nsc_size_t vol_size = 0;
	nsc_size_t req_size = 0;

	mutex_enter(&rdc_conf_lock);
	index = rdc_lookup_byname(uparms->rdc_set);
	mutex_exit(&rdc_conf_lock);
	if (index < 0) {
		spcs_s_add(kstatus, RDC_EALREADY, uparms->rdc_set->primary.file,
		    uparms->rdc_set->secondary.file);
		rc = RDC_EALREADY;
		goto failed;
	}
	urdc = &rdc_u_info[index];
	krdc = &rdc_k_info[index];
	this = &rdc_k_info[index];
	group = krdc->group;
	diskq = uparms->rdc_set->disk_queue;

	if (!IS_ASYNC(urdc)) {
		spcs_s_add(kstatus, RDC_EQWRONGMODE, urdc->primary.intf,
		    urdc->primary.file, urdc->secondary.intf,
		    urdc->secondary.file);
		rc = RDC_EQNOQUEUE;
		goto failed;
	}

	do {
		if (!IS_STATE(urdc, RDC_LOGGING)) {
			spcs_s_add(kstatus, RDC_EQNOTLOGGING,
			    uparms->rdc_set->disk_queue);
			rc = RDC_EQNOTLOGGING;
			goto failed;
		}
		/* make sure that we have enough bitmap vol */
		req_size = RDC_BITMAP_FBA + FBA_LEN(krdc->bitmap_size);
		req_size += FBA_LEN(krdc->bitmap_size * BITS_IN_BYTE);

		rc = _rdc_rsrv_devs(krdc, RDC_BMP, RDC_INTERNAL);

		if (!RDC_SUCCESS(rc)) {
			cmn_err(CE_WARN,
			    "!rdc_open_diskq: Bitmap reserve failed");
			spcs_s_add(kstatus, RDC_EBITMAP,
			    urdc->primary.bitmap);
			rc = RDC_EBITMAP;
			goto failed;
		}

		(void) nsc_partsize(krdc->bitmapfd, &vol_size);

		_rdc_rlse_devs(krdc, RDC_BMP);

		if (vol_size < req_size) {
			spcs_s_add(kstatus, RDC_EBITMAP2SMALL,
			    urdc->primary.bitmap);
			rc = RDC_EBITMAP2SMALL;
			goto failed;
		}

		krdc = krdc->group_next;
		urdc = &rdc_u_info[krdc->index];

	} while (krdc != this);

	if (urdc->disk_queue[0] != '\0') {
		spcs_s_add(kstatus, RDC_EQALREADY, urdc->primary.intf,
		    urdc->primary.file, urdc->secondary.intf,
		    urdc->secondary.file);
		rc = RDC_EQALREADY;
		goto failed;
	}

	if (uparms->options & RDC_OPT_SECONDARY) { /* how'd we get here? */
		spcs_s_add(kstatus, RDC_EQWRONGMODE);
		rc = RDC_EQWRONGMODE;
		goto failed;
	}

	mutex_enter(&rdc_conf_lock);
	if (rdc_diskq_inuse(uparms->rdc_set, uparms->rdc_set->disk_queue)) {
		spcs_s_add(kstatus, RDC_EDISKQINUSE,
		    uparms->rdc_set->disk_queue);
		rc = RDC_EDISKQINUSE;
		mutex_exit(&rdc_conf_lock);
		goto failed;
	}
	mutex_exit(&rdc_conf_lock);

	rdc_group_enter(krdc);
	rc = _rdc_add_diskq(urdc->index, diskq);
	if (rc < 0 || rc == RDC_EQNOADD) {
		group->flags &= ~RDC_DISKQUE;
		group->flags |= RDC_MEMQUE;
		spcs_s_add(kstatus, RDC_EQNOADD, uparms->rdc_set->disk_queue);
		rc = RDC_EQNOADD;
	}
	rdc_group_exit(krdc);
failed:
	return (rc);
}

int
_rdc_init_diskq(rdc_k_info_t *krdc)
{
	rdc_group_t *group = krdc->group;
	disk_queue  *q = &group->diskq;

	rdc_init_diskq_header(group, &group->diskq.disk_hdr);
	SET_QNXTIO(q, QHEAD(q));

	if (rdc_stamp_diskq(krdc, 0, RDC_NOLOG) < 0)
		goto fail;

	return (0);
fail:
	return (-1);
}

/*
 * inititalize the disk queue. This is a destructive
 * operation that will not check for emptiness of the queue.
 */
int
rdc_init_diskq(rdc_config_t *uparms, spcs_s_info_t kstatus)
{
	int rc = 0;
	int index;
	rdc_k_info_t *krdc, *kp;
	rdc_u_info_t *urdc, *up;
	rdc_set_t    *uset;
	rdc_group_t  *group;
	disk_queue   *qp;

	uset = uparms->rdc_set;

	mutex_enter(&rdc_conf_lock);
	index = rdc_lookup_byname(uset);
	mutex_exit(&rdc_conf_lock);
	if (index < 0) {
		spcs_s_add(kstatus, RDC_EALREADY, uset->primary.file,
		    uset->secondary.file);
		rc = RDC_EALREADY;
		goto fail;
	}

	krdc = &rdc_k_info[index];
	urdc = &rdc_u_info[index];
	group = krdc->group;
	qp = &group->diskq;

	if (!IS_STATE(urdc, RDC_SYNCING) && !IS_STATE(urdc, RDC_LOGGING)) {
		spcs_s_add(kstatus, RDC_EQUEISREP, urdc->disk_queue);
		rc = RDC_EQUEISREP;
		goto fail;
	}

	/*
	 * a couple of big "ifs" here. in the first implementation
	 * neither of these will be possible. This will come into
	 * play when we persist the queue across reboots
	 */
	if (!(uparms->options & RDC_OPT_FORCE_QINIT)) {
		if (!QEMPTY(qp)) {
			if (group->rdc_writer) {
				spcs_s_add(kstatus, RDC_EQFLUSHING,
				    urdc->disk_queue);
				rc = RDC_EQFLUSHING;
			} else {
				spcs_s_add(kstatus, RDC_EQNOTEMPTY,
				    urdc->disk_queue);
				rc = RDC_EQNOTEMPTY;
			}
			goto fail;
		}
	}

	mutex_enter(QLOCK(qp));
	if (_rdc_init_diskq(krdc) < 0) {
		mutex_exit(QLOCK(qp));
		goto fail;
	}
	rdc_dump_iohdrs(qp);

	rdc_group_enter(krdc);

	rdc_clr_flags(urdc, RDC_QUEUING);
	for (kp = krdc->group_next; kp != krdc; kp = kp->group_next) {
		up = &rdc_u_info[kp->index];
		rdc_clr_flags(up, RDC_QUEUING);
	}
	rdc_group_exit(krdc);

	mutex_exit(QLOCK(qp));

	return (0);
fail:
	/* generic queue failure */
	if (!rc) {
		spcs_s_add(kstatus, RDC_EQINITFAIL, urdc->disk_queue);
		rc = RDC_EQINITFAIL;
	}

	return (rc);
}

int
_rdc_kill_diskq(rdc_u_info_t *urdc)
{
	rdc_k_info_t *krdc = &rdc_k_info[urdc->index];
	rdc_group_t *group = krdc->group;
	disk_queue *q = &group->diskq;
	rdc_u_info_t *up;
	rdc_k_info_t *p;

	group->flags |= RDC_DISKQ_KILL;
#ifdef DEBUG
	cmn_err(CE_NOTE, "!disabling disk queue %s", urdc->disk_queue);
#endif

	mutex_enter(QLOCK(q));
	rdc_init_diskq_header(group, &q->disk_hdr);
	rdc_dump_iohdrs(q);

	/*
	 * nsc_close the queue and zero out the queue name
	 */
	rdc_wait_qbusy(q);
	rdc_close_diskq(group);
	mutex_exit(QLOCK(q));
	SET_QSIZE(q, 0);
	rdc_clr_flags(urdc, RDC_DISKQ_FAILED);
	bzero(urdc->disk_queue, NSC_MAXPATH);
	for (p = krdc->group_next; p != krdc; p = p->group_next) {
		up = &rdc_u_info[p->index];
		rdc_clr_flags(up, RDC_DISKQ_FAILED);
		bzero(up->disk_queue, NSC_MAXPATH);
	}

#ifdef DEBUG
	cmn_err(CE_NOTE, "!_rdc_kill_diskq: enabling memory queue");
#endif
	group->flags &= ~(RDC_DISKQUE|RDC_DISKQ_KILL);
	group->flags |= RDC_MEMQUE;
	return (0);
}

/*
 * remove this diskq regardless of whether it is draining or not
 * stops the flusher by invalidating the qdata (ie, instant empty)
 * remove the disk qeueue from the group, leaving the group with a memory
 * queue.
 */
int
rdc_kill_diskq(rdc_config_t *uparms, spcs_s_info_t kstatus)
{
	int rc;
	int index;
	rdc_u_info_t *urdc;
	rdc_k_info_t *krdc;
	rdc_set_t *rdc_set = uparms->rdc_set;

	mutex_enter(&rdc_conf_lock);
	index = rdc_lookup_byname(uparms->rdc_set);
	mutex_exit(&rdc_conf_lock);

	if (index < 0) {
		spcs_s_add(kstatus, RDC_EALREADY, rdc_set->primary.file,
		    rdc_set->secondary.file);
		rc = RDC_EALREADY;
		goto failed;
	}

	urdc = &rdc_u_info[index];
	krdc = &rdc_k_info[index];

	if (!RDC_IS_DISKQ(krdc->group)) {
		spcs_s_add(kstatus, RDC_EQNOQUEUE, rdc_set->primary.intf,
		    rdc_set->primary.file, rdc_set->secondary.intf,
		    rdc_set->secondary.file);
		rc = RDC_EQNOQUEUE;
		goto failed;
	}

/*
 *	if (!IS_STATE(urdc, RDC_LOGGING)) {
 *		spcs_s_add(kstatus, RDC_EQNOTLOGGING,
 *		    uparms->rdc_set->disk_queue);
 *		rc = RDC_EQNOTLOGGING;
 *		goto failed;
 *	}
 */
	rdc_unintercept_diskq(krdc->group); /* stop protecting queue */
	rdc_group_enter(krdc); /* to prevent further flushing */
	rc = _rdc_kill_diskq(urdc);
	rdc_group_exit(krdc);

failed:
	return (rc);
}

/*
 * remove a diskq from a group.
 * removal of a diskq from a set, or rather
 * a set from a queue, is done by reconfigging out
 * of the group. This removes the diskq from a whole
 * group and replaces it with a memory based queue
 */
#define	NUM_RETRIES	15	/* Number of retries to wait if no progress */
int
rdc_rem_diskq(rdc_config_t *uparms, spcs_s_info_t kstatus)
{
	int index;
	rdc_u_info_t *urdc;
	rdc_k_info_t *krdc;
	rdc_k_info_t *this;
	volatile rdc_group_t *group;
	volatile disk_queue *diskq;
	int threads, counter;
	long blocks;

	mutex_enter(&rdc_conf_lock);
	index = rdc_lookup_byname(uparms->rdc_set);
	mutex_exit(&rdc_conf_lock);
	if (index < 0) {
		spcs_s_add(kstatus, RDC_EALREADY, uparms->rdc_set->primary.file,
		    uparms->rdc_set->secondary.file);
		return (RDC_EALREADY);
	}

	urdc = &rdc_u_info[index];
	this = &rdc_k_info[index];
	krdc = &rdc_k_info[index];

	do {
		if (!IS_STATE(urdc, RDC_LOGGING)) {
			spcs_s_add(kstatus, RDC_EQNOTLOGGING,
			    urdc->disk_queue);
			return (RDC_EQNOTLOGGING);
		}
		krdc = krdc->group_next;
		urdc = &rdc_u_info[krdc->index];

	} while (krdc != this);

	/*
	 * If there is no group or diskq configured, we can leave now
	 */
	if (!(group = krdc->group) || !(diskq = &group->diskq))
		return (0);


	/*
	 * Wait if not QEMPTY or threads still active
	 */
	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 see, decrement retry counter
			 */
			if (counter++ > NUM_RETRIES) {
				/*
				 * No progress seen, increment retry counter
				 */
				int rc = group->rdc_thrnum ?
				    RDC_EQFLUSHING : RDC_EQNOTEMPTY;
				spcs_s_add(kstatus, rc, urdc->disk_queue);
				return (rc);
			}
		} else {
			/*
			 * Reset counter, as we've made progress
			 */
			counter = 0;
		}
	}

	return (0);
}