/*
 * Copyright (c) 2012-2014 The DragonFly Project.  All rights reserved.
 *
 * This code is derived from software contributed to The DragonFly Project
 * by Matthew Dillon <dillon@dragonflybsd.org>
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 * 3. Neither the name of The DragonFly Project nor the names of its
 *    contributors may be used to endorse or promote products derived
 *    from this software without specific, prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
 * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
/*
 * This module allows disk devices to be created and associated with a
 * communications pipe or socket.  You open the device and issue an
 * ioctl() to install a new disk along with its communications descriptor.
 *
 * All further communication occurs via the descriptor using the DMSG
 * LNK_CONN, LNK_SPAN, and BLOCK protocols.  The descriptor can be a
 * direct connection to a remote machine's disk (in-kernenl), to a remote
 * cluster controller, to the local cluster controller, etc.
 *
 * /dev/xdisk is the control device, issue ioctl()s to create the /dev/xa%d
 * devices.  These devices look like raw disks to the system.
 */
#include <sys/param.h>
#include <sys/systm.h>
#include <sys/buf.h>
#include <sys/conf.h>
#include <sys/device.h>
#include <sys/devicestat.h>
#include <sys/disk.h>
#include <sys/kernel.h>
#include <sys/malloc.h>
#include <sys/sysctl.h>
#include <sys/proc.h>
#include <sys/queue.h>
#include <sys/tree.h>
#include <sys/udev.h>
#include <sys/uuid.h>
#include <sys/kern_syscall.h>

#include <sys/dmsg.h>
#include <sys/xdiskioctl.h>

#include <sys/buf2.h>

struct xa_softc;
struct xa_softc_tree;
RB_HEAD(xa_softc_tree, xa_softc);
RB_PROTOTYPE(xa_softc_tree, xa_softc, rbnode, xa_softc_cmp);

static int xa_active;
SYSCTL_INT(_debug, OID_AUTO, xa_active, CTLFLAG_RW, &xa_active, 0,
	   "Number of active xdisk IOs");
static uint64_t xa_last;
SYSCTL_ULONG(_debug, OID_AUTO, xa_last, CTLFLAG_RW, &xa_last, 0,
	   "Offset of last xdisk IO");
static int xa_debug = 1;
SYSCTL_INT(_debug, OID_AUTO, xa_debug, CTLFLAG_RW, &xa_debug, 0,
	   "xdisk debugging");

/*
 * Track a BIO tag
 */
struct xa_tag {
	TAILQ_ENTRY(xa_tag) entry;
	struct xa_softc	*sc;
	dmsg_blk_error_t status;
	kdmsg_state_t	*state;
	struct bio	*bio;
	int		waiting;
	int		async;
	int		done;
};

typedef struct xa_tag	xa_tag_t;

/*
 * Track devices.
 */
struct xa_softc {
	struct kdmsg_state_list spanq;
	RB_ENTRY(xa_softc) rbnode;
	cdev_t		dev;
	struct devstat	stats;
	struct disk_info info;
	struct disk	disk;
	uuid_t		peer_id;
	int		unit;
	int		opencnt;
	int		spancnt;
	uint64_t	keyid;
	int		serializing;
	int		last_error;
	int		terminating;
	char		peer_label[64];	/* from LNK_SPAN host/dev */
	char		pfs_label[64];	/* from LNK_SPAN serno */
	xa_tag_t	*open_tag;
	TAILQ_HEAD(, bio) bioq;		/* pending BIOs */
	TAILQ_HEAD(, xa_tag) tag_freeq;	/* available I/O tags */
	TAILQ_HEAD(, xa_tag) tag_pendq;	/* running I/O tags */
	struct lock	lk;
};

typedef struct xa_softc	xa_softc_t;

struct xa_iocom {
	TAILQ_ENTRY(xa_iocom) entry;
	kdmsg_iocom_t	iocom;
	xa_softc_t	dummysc;
};

typedef struct xa_iocom xa_iocom_t;

static int xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2);
RB_GENERATE(xa_softc_tree, xa_softc, rbnode, xa_softc_cmp);
static struct xa_softc_tree xa_device_tree;

#define MAXTAGS		64	/* no real limit */

static int xdisk_attach(struct xdisk_attach_ioctl *xaioc);
static int xdisk_detach(struct xdisk_attach_ioctl *xaioc);
static void xaio_exit(kdmsg_iocom_t *iocom);
static int xaio_rcvdmsg(kdmsg_msg_t *msg);

static void xa_terminate_check(struct xa_softc *sc);

static xa_tag_t *xa_setup_cmd(xa_softc_t *sc, struct bio *bio);
static void xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async);
static void xa_done(xa_tag_t *tag, int wasbio);
static void xa_release(xa_tag_t *tag, int wasbio);
static uint32_t xa_wait(xa_tag_t *tag);
static int xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
static int xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
static void xa_restart_deferred(xa_softc_t *sc);

#define xa_printf(level, ctl, ...)	\
	if (xa_debug >= (level)) kprintf("xdisk: " ctl, __VA_ARGS__)

MALLOC_DEFINE(M_XDISK, "Networked disk client", "Network Disks");

/*
 * Control device, issue ioctls to create xa devices.
 */
static d_open_t xdisk_open;
static d_close_t xdisk_close;
static d_ioctl_t xdisk_ioctl;

static struct dev_ops xdisk_ops = {
	{ "xdisk", 0, D_MPSAFE | D_TRACKCLOSE },
        .d_open =	xdisk_open,
        .d_close =	xdisk_close,
        .d_ioctl =	xdisk_ioctl
};

/*
 * XA disk devices
 */
static d_open_t xa_open;
static d_close_t xa_close;
static d_ioctl_t xa_ioctl;
static d_strategy_t xa_strategy;
static d_psize_t xa_size;

static struct dev_ops xa_ops = {
	{ "xa", 0, D_DISK | D_CANFREE | D_MPSAFE | D_TRACKCLOSE },
        .d_open =	xa_open,
        .d_close =	xa_close,
        .d_ioctl =	xa_ioctl,
        .d_read =	physread,
        .d_write =	physwrite,
        .d_strategy =	xa_strategy,
	.d_psize =	xa_size
};

static int xdisk_opencount;
static cdev_t xdisk_dev;
struct lock xdisk_lk;
static TAILQ_HEAD(, xa_iocom) xaiocomq;

/*
 * Module initialization
 */
static int
xdisk_modevent(module_t mod, int type, void *data)
{
	switch (type) {
	case MOD_LOAD:
		TAILQ_INIT(&xaiocomq);
		RB_INIT(&xa_device_tree);
		lockinit(&xdisk_lk, "xdisk", 0, 0);
		xdisk_dev = make_dev(&xdisk_ops, 0,
				     UID_ROOT, GID_WHEEL, 0600, "xdisk");
		break;
	case MOD_UNLOAD:
	case MOD_SHUTDOWN:
		if (!RB_EMPTY(&xa_device_tree))
			return (EBUSY);
		if (xdisk_opencount || TAILQ_FIRST(&xaiocomq))
			return (EBUSY);
		if (xdisk_dev) {
			destroy_dev(xdisk_dev);
			xdisk_dev = NULL;
		}
		dev_ops_remove_all(&xdisk_ops);
		dev_ops_remove_all(&xa_ops);
		break;
	default:
		break;
	}
	return 0;
}

DEV_MODULE(xdisk, xdisk_modevent, 0);

static int
xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2)
{
	return(strcmp(sc1->pfs_label, sc2->pfs_label));
}

/*
 * Control device
 */
static int
xdisk_open(struct dev_open_args *ap)
{
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
	++xdisk_opencount;
	lockmgr(&xdisk_lk, LK_RELEASE);
	return(0);
}

static int
xdisk_close(struct dev_close_args *ap)
{
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
	--xdisk_opencount;
	lockmgr(&xdisk_lk, LK_RELEASE);
	return(0);
}

static int
xdisk_ioctl(struct dev_ioctl_args *ap)
{
	int error;

	switch(ap->a_cmd) {
	case XDISKIOCATTACH:
		error = xdisk_attach((void *)ap->a_data);
		break;
	case XDISKIOCDETACH:
		error = xdisk_detach((void *)ap->a_data);
		break;
	default:
		error = ENOTTY;
		break;
	}
	return error;
}

/************************************************************************
 *				DMSG INTERFACE				*
 ************************************************************************/

static int
xdisk_attach(struct xdisk_attach_ioctl *xaioc)
{
	xa_iocom_t *xaio;
	struct file *fp;

	/*
	 * Normalize ioctl params
	 */
	fp = holdfp(curthread, xaioc->fd, -1);
	if (fp == NULL)
		return EINVAL;
	xa_printf(1, "xdisk_attach fp=%p\n", fp);

	/*
	 * See if the serial number is already present.  If we are
	 * racing a termination the disk subsystem may still have
	 * duplicate entries not yet removed so we wait a bit and
	 * retry.
	 */
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);

	xaio = kmalloc(sizeof(*xaio), M_XDISK, M_WAITOK | M_ZERO);
	kdmsg_iocom_init(&xaio->iocom, xaio,
			 KDMSG_IOCOMF_AUTOCONN,
			 M_XDISK, xaio_rcvdmsg);
	xaio->iocom.exit_func = xaio_exit;

	kdmsg_iocom_reconnect(&xaio->iocom, fp, "xdisk");

	/*
	 * Setup our LNK_CONN advertisement for autoinitiate.
	 *
	 * Our filter is setup to only accept PEER_BLOCK advertisements.
	 * XXX no peer_id filter.
	 *
	 * We need a unique pfs_fsid to avoid confusion.
	 */
	xaio->iocom.auto_lnk_conn.peer_type = DMSG_PEER_CLIENT;
	xaio->iocom.auto_lnk_conn.proto_version = DMSG_SPAN_PROTO_1;
	xaio->iocom.auto_lnk_conn.peer_mask = 1LLU << DMSG_PEER_BLOCK;
	ksnprintf(xaio->iocom.auto_lnk_conn.peer_label,
		  sizeof(xaio->iocom.auto_lnk_conn.peer_label),
		  "%s/xdisk",
		  hostname);
	/* kern_uuidgen(&xaio->iocom.auto_lnk_conn.pfs_fsid, 1); */

	/*
	 * Setup our LNK_SPAN advertisement for autoinitiate
	 */
	TAILQ_INSERT_TAIL(&xaiocomq, xaio, entry);
	kdmsg_iocom_autoinitiate(&xaio->iocom, NULL);

	lockmgr(&xdisk_lk, LK_RELEASE);

	return 0;
}

static int
xdisk_detach(struct xdisk_attach_ioctl *xaioc)
{
	return EINVAL;
}

/*
 * Called from iocom core transmit thread upon disconnect.
 */
static
void
xaio_exit(kdmsg_iocom_t *iocom)
{
	xa_iocom_t *xaio = iocom->handle;

	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
	xa_printf(1, "%s", "xdisk_detach [xaio_exit()]\n");
	TAILQ_REMOVE(&xaiocomq, xaio, entry);
	lockmgr(&xdisk_lk, LK_RELEASE);

	kdmsg_iocom_uninit(&xaio->iocom);

	kfree(xaio, M_XDISK);
}

/*
 * Called from iocom core to handle messages that the iocom core does not
 * handle itself and for which a state function callback has not yet been
 * established.
 *
 * We primarily care about LNK_SPAN transactions here.
 */
static int
xaio_rcvdmsg(kdmsg_msg_t *msg)
{
	kdmsg_state_t	*state = msg->state;
	xa_iocom_t	*xaio = state->iocom->handle;
	xa_softc_t	*sc;

	if (state) {
		xa_printf(4,
			"xdisk - rcvmsg state=%p rx=%08x tx=%08x msgcmd=%08x\n",
			state, state->rxcmd, state->txcmd,
			msg->any.head.cmd);
	}
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);

	switch(msg->tcmd) {
	case DMSG_LNK_SPAN | DMSGF_CREATE | DMSGF_DELETE:
		/*
		 * A LNK_SPAN transaction which is opened and closed
		 * degenerately is not useful to us, just ignore it.
		 */
		kdmsg_msg_reply(msg, 0);
		break;
	case DMSG_LNK_SPAN | DMSGF_CREATE:
		/*
		 * Manage the tracking node for the remote LNK_SPAN.
		 *
		 * Return a streaming result, leaving the transaction open
		 * in both directions to allow sub-transactions.
		 */
		bcopy(msg->any.lnk_span.peer_label, xaio->dummysc.peer_label,
		      sizeof(xaio->dummysc.peer_label));
		xaio->dummysc.peer_label[
			sizeof(xaio->dummysc.peer_label) - 1] = 0;

		bcopy(msg->any.lnk_span.pfs_label, xaio->dummysc.pfs_label,
		      sizeof(xaio->dummysc.pfs_label));
		xaio->dummysc.pfs_label[
			sizeof(xaio->dummysc.pfs_label) - 1] = 0;

		xa_printf(3, "LINK_SPAN state %p create for %s\n",
			  msg->state, msg->any.lnk_span.pfs_label);

		sc = RB_FIND(xa_softc_tree, &xa_device_tree, &xaio->dummysc);
		if (sc == NULL) {
			xa_softc_t *sctmp;
			xa_tag_t *tag;
			cdev_t dev;
			int unit;
			int n;

			sc = kmalloc(sizeof(*sc), M_XDISK, M_WAITOK | M_ZERO);
			bcopy(msg->any.lnk_span.peer_label, sc->peer_label,
			      sizeof(sc->peer_label));
			sc->peer_label[sizeof(sc->peer_label) - 1] = 0;
			bcopy(msg->any.lnk_span.pfs_label, sc->pfs_label,
			      sizeof(sc->pfs_label));
			sc->pfs_label[sizeof(sc->pfs_label) - 1] = 0;

			/* XXX FIXME O(N^2) */
			unit = -1;
			do {
				++unit;
				RB_FOREACH(sctmp, xa_softc_tree,
					   &xa_device_tree) {
					if (sctmp->unit == unit)
						break;
				}
			} while (sctmp);

			sc->unit = unit;
			sc->serializing = 1;
			sc->spancnt = 1;
			lockinit(&sc->lk, "xalk", 0, 0);
			TAILQ_INIT(&sc->spanq);
			TAILQ_INIT(&sc->bioq);
			TAILQ_INIT(&sc->tag_freeq);
			TAILQ_INIT(&sc->tag_pendq);

			lockmgr(&sc->lk, LK_EXCLUSIVE);
			RB_INSERT(xa_softc_tree, &xa_device_tree, sc);
			TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
			msg->state->any.xa_sc = sc;

			/*
			 * Setup block device
			 */
			for (n = 0; n < MAXTAGS; ++n) {
				tag = kmalloc(sizeof(*tag),
					      M_XDISK, M_WAITOK|M_ZERO);
				tag->sc = sc;
				TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
			}

			if (sc->dev == NULL) {
				dev = disk_create(unit, &sc->disk, &xa_ops);
				dev->si_drv1 = sc;
				sc->dev = dev;
				devstat_add_entry(&sc->stats, "xa", unit,
						  DEV_BSIZE,
						  DEVSTAT_NO_ORDERED_TAGS,
						  DEVSTAT_TYPE_DIRECT |
						  DEVSTAT_TYPE_IF_OTHER,
						  DEVSTAT_PRIORITY_OTHER);
			}

			sc->info.d_media_blksize =
				msg->any.lnk_span.media.block.blksize;
			if (sc->info.d_media_blksize <= 0)
				sc->info.d_media_blksize = 1;
			sc->info.d_media_blocks =
				msg->any.lnk_span.media.block.bytes /
				sc->info.d_media_blksize;
			sc->info.d_dsflags = DSO_MBRQUIET | DSO_RAWPSIZE;
			sc->info.d_secpertrack = 32;
			sc->info.d_nheads = 64;
			sc->info.d_secpercyl = sc->info.d_secpertrack *
					       sc->info.d_nheads;
			sc->info.d_ncylinders = 0;
			if (sc->pfs_label[0])
				sc->info.d_serialno = sc->pfs_label;
			/*
			 * WARNING! disk_setdiskinfo() must be asynchronous
			 *	    because we are in the rxmsg thread.  If
			 *	    it is synchronous and issues more disk
			 *	    I/Os, we will deadlock.
			 */
			disk_setdiskinfo(&sc->disk, &sc->info);
			xa_restart_deferred(sc);	/* eats serializing */
			lockmgr(&sc->lk, LK_RELEASE);
		} else {
			lockmgr(&sc->lk, LK_EXCLUSIVE);
			++sc->spancnt;
			TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
			msg->state->any.xa_sc = sc;
			if (sc->serializing == 0 && sc->open_tag == NULL) {
				sc->serializing = 1;
				xa_restart_deferred(sc); /* eats serializing */
			}
			lockmgr(&sc->lk, LK_RELEASE);
			if (sc->dev && sc->dev->si_disk) {
				xa_printf(1, "reprobe disk: %s\n",
					  sc->pfs_label);
				disk_msg_send(DISK_DISK_REPROBE,
					      sc->dev->si_disk,
					      NULL);
			}
		}
		xa_printf(2, "sc %p spancnt %d\n", sc, sc->spancnt);
		kdmsg_msg_result(msg, 0);
		break;
	case DMSG_LNK_SPAN | DMSGF_DELETE:
		/*
		 * Manage the tracking node for the remote LNK_SPAN.
		 *
		 * Return a final result, closing our end of the transaction.
		 */
		sc = msg->state->any.xa_sc;
		xa_printf(3, "LINK_SPAN state %p delete for %s (sc=%p)\n",
			  msg->state, (sc ? sc->pfs_label : "(null)"), sc);
		lockmgr(&sc->lk, LK_EXCLUSIVE);
		msg->state->any.xa_sc = NULL;
		TAILQ_REMOVE(&sc->spanq, msg->state, user_entry);
		--sc->spancnt;

		xa_printf(2, "sc %p spancnt %d\n", sc, sc->spancnt);

		/*
		 * Spans can come and go as the graph stabilizes, so if
		 * we lose a span along with sc->open_tag we may be able
		 * to restart the I/Os on a different span.
		 */
		if (sc->spancnt &&
		    sc->serializing == 0 && sc->open_tag == NULL) {
			sc->serializing = 1;
			xa_restart_deferred(sc);
		}
		lockmgr(&sc->lk, LK_RELEASE);
		kdmsg_msg_reply(msg, 0);

#if 0
		/*
		 * Termination
		 */
		if (sc->spancnt == 0)
			xa_terminate_check(sc);
#endif
		break;
	case DMSG_LNK_SPAN | DMSGF_DELETE | DMSGF_REPLY:
		/*
		 * Ignore unimplemented streaming replies on our LNK_SPAN
		 * transaction.
		 */
		xa_printf(3, "LINK_SPAN state %p delete+reply\n",
			  msg->state);
		break;
	case DMSG_LNK_SPAN | DMSGF_REPLY:
		/*
		 * Ignore unimplemented streaming replies on our LNK_SPAN
		 * transaction.
		 */
		xa_printf(3, "LINK_SPAN state %p reply\n",
			  msg->state);
		break;
	case DMSG_DBG_SHELL:
		/*
		 * Execute shell command (not supported atm).
		 *
		 * This is a one-way packet but if not (e.g. if part of
		 * a streaming transaction), we will have already closed
		 * our end.
		 */
		kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
		break;
	case DMSG_DBG_SHELL | DMSGF_REPLY:
		/*
		 * Receive one or more replies to a shell command
		 * that we sent.  Just dump it to the console.
		 *
		 * This is a one-way packet but if not (e.g. if
		 * part of a streaming transaction), we will have
		 * already closed our end.
		 */
		if (msg->aux_data) {
			msg->aux_data[msg->aux_size - 1] = 0;
			xa_printf(0, "DEBUGMSG: %s\n", msg->aux_data);
		}
		break;
	default:
		/*
		 * Unsupported one-way message, streaming message, or
		 * transaction.
		 *
		 * Terminate any unsupported transactions with an error
		 * and ignore any unsupported streaming messages.
		 *
		 * NOTE: This case also includes DMSG_LNK_ERROR messages
		 *	 which might be one-way, replying to those would
		 *	 cause an infinite ping-pong.
		 */
		if (msg->any.head.cmd & DMSGF_CREATE)
			kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
		break;
	}
	lockmgr(&xdisk_lk, LK_RELEASE);

	return 0;
}

/*
 * Determine if we can destroy the xa_softc.
 *
 * Called with xdisk_lk held.
 */
static
void
xa_terminate_check(struct xa_softc *sc)
{
	xa_tag_t *tag;

	/*
	 * Determine if we can destroy the softc.
	 */
	xa_printf(1, "Terminate check xa%d (%d,%d,%d) sc=%p ",
		sc->unit,
		sc->opencnt, sc->serializing, sc->spancnt,
		sc);

	if (sc->opencnt || sc->serializing || sc->spancnt ||
	    TAILQ_FIRST(&sc->bioq) || TAILQ_FIRST(&sc->tag_pendq)) {
		xa_printf(1, "%s", "(leave intact)\n");
		return;
	}

	/*
	 * Remove from device tree, a race with a new incoming span
	 * will create a new softc and disk.
	 */
	RB_REMOVE(xa_softc_tree, &xa_device_tree, sc);
	sc->terminating = 1;

	/*
	 * Device has to go first to prevent device ops races.
	 */
	if (sc->dev) {
		disk_destroy(&sc->disk);
		devstat_remove_entry(&sc->stats);
		sc->dev->si_drv1 = NULL;
		sc->dev = NULL;
	}

	xa_printf(1, "%s", "(remove from tree)\n");
	sc->serializing = 1;
	KKASSERT(sc->opencnt == 0);
	KKASSERT(TAILQ_EMPTY(&sc->tag_pendq));

	while ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
		TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
		tag->sc = NULL;
		kfree(tag, M_XDISK);
	}

	kfree(sc, M_XDISK);
}

/************************************************************************
 *			   XA DEVICE INTERFACE				*
 ************************************************************************/

static int
xa_open(struct dev_open_args *ap)
{
	cdev_t dev = ap->a_head.a_dev;
	xa_softc_t *sc;
	int error;

	dev->si_bsize_phys = 512;
	dev->si_bsize_best = 32768;

	/*
	 * Interlock open with opencnt, wait for attachment operations
	 * to finish.
	 */
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
again:
	sc = dev->si_drv1;
	if (sc == NULL) {
		lockmgr(&xdisk_lk, LK_RELEASE);
		return ENXIO;	/* raced destruction */
	}
	if (sc->serializing) {
		tsleep(sc, 0, "xarace", hz / 10);
		goto again;
	}
	if (sc->terminating) {
		lockmgr(&xdisk_lk, LK_RELEASE);
		return ENXIO;	/* raced destruction */
	}
	sc->serializing = 1;

	/*
	 * Serialize initial open
	 */
	if (sc->opencnt++ > 0) {
		sc->serializing = 0;
		wakeup(sc);
		lockmgr(&xdisk_lk, LK_RELEASE);
		return(0);
	}

	/*
	 * Issue BLK_OPEN if necessary.  ENXIO is returned if we have trouble.
	 */
	if (sc->open_tag == NULL) {
		lockmgr(&sc->lk, LK_EXCLUSIVE);
		xa_restart_deferred(sc); /* eats serializing */
		lockmgr(&sc->lk, LK_RELEASE);
	} else {
		sc->serializing = 0;
		wakeup(sc);
	}
	lockmgr(&xdisk_lk, LK_RELEASE);

	/*
	 * Wait for completion of the BLK_OPEN
	 */
	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
	while (sc->serializing)
		lksleep(sc, &xdisk_lk, 0, "xaopen", hz);

	error = sc->last_error;
	if (error) {
		KKASSERT(sc->opencnt > 0);
		--sc->opencnt;
		xa_terminate_check(sc);
		sc = NULL;	/* sc may be invalid now */
	}
	lockmgr(&xdisk_lk, LK_RELEASE);

	return (error);
}

static int
xa_close(struct dev_close_args *ap)
{
	cdev_t dev = ap->a_head.a_dev;
	xa_softc_t *sc;
	xa_tag_t *tag;

	lockmgr(&xdisk_lk, LK_EXCLUSIVE);
	sc = dev->si_drv1;
	if (sc == NULL)
		return ENXIO;	/* raced destruction */
	if (sc->terminating) {
		lockmgr(&sc->lk, LK_RELEASE);
		return ENXIO;	/* raced destruction */
	}
	lockmgr(&sc->lk, LK_EXCLUSIVE);

	/*
	 * NOTE: Clearing open_tag allows a concurrent open to re-open
	 *	 the device and prevents autonomous completion of the tag.
	 */
	if (sc->opencnt == 1 && sc->open_tag) {
		tag = sc->open_tag;
		sc->open_tag = NULL;
		lockmgr(&sc->lk, LK_RELEASE);
		kdmsg_state_reply(tag->state, 0);	/* close our side */
		xa_wait(tag);				/* wait on remote */
	} else {
		lockmgr(&sc->lk, LK_RELEASE);
	}
	KKASSERT(sc->opencnt > 0);
	--sc->opencnt;
	xa_terminate_check(sc);
	lockmgr(&xdisk_lk, LK_RELEASE);

	return(0);
}

static int
xa_strategy(struct dev_strategy_args *ap)
{
	xa_softc_t *sc = ap->a_head.a_dev->si_drv1;
	xa_tag_t *tag;
	struct bio *bio = ap->a_bio;

	devstat_start_transaction(&sc->stats);
	atomic_add_int(&xa_active, 1);
	xa_last = bio->bio_offset;

	/*
	 * If no tags are available NULL is returned and the bio is
	 * placed on sc->bioq.
	 */
	lockmgr(&sc->lk, LK_EXCLUSIVE);
	tag = xa_setup_cmd(sc, bio);
	if (tag)
		xa_start(tag, NULL, 1);
	lockmgr(&sc->lk, LK_RELEASE);

	return(0);
}

static int
xa_ioctl(struct dev_ioctl_args *ap)
{
	return(ENOTTY);
}

static int
xa_size(struct dev_psize_args *ap)
{
	struct xa_softc *sc;

	if ((sc = ap->a_head.a_dev->si_drv1) == NULL)
		return (ENXIO);
	ap->a_result = sc->info.d_media_blocks;
	return (0);
}

/************************************************************************
 *		    XA BLOCK PROTOCOL STATE MACHINE			*
 ************************************************************************
 *
 * Implement tag/msg setup and related functions.
 * Called with sc->lk held.
 */
static xa_tag_t *
xa_setup_cmd(xa_softc_t *sc, struct bio *bio)
{
	xa_tag_t *tag;

	/*
	 * Only get a tag if we have a valid virtual circuit to the server.
	 */
	if ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
		TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
		tag->bio = bio;
		TAILQ_INSERT_TAIL(&sc->tag_pendq, tag, entry);
	}

	/*
	 * If we can't dispatch now and this is a bio, queue it for later.
	 */
	if (tag == NULL && bio) {
		TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
	}

	return (tag);
}

/*
 * Called with sc->lk held
 */
static void
xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async)
{
	xa_softc_t *sc = tag->sc;

	tag->done = 0;
	tag->async = async;
	tag->status.head.error = DMSG_ERR_IO;	/* fallback error */

	if (msg == NULL) {
		struct bio *bio;
		struct buf *bp;
		kdmsg_state_t *trans;

		if (sc->opencnt == 0 || sc->open_tag == NULL) {
			TAILQ_FOREACH(trans, &sc->spanq, user_entry) {
				if ((trans->rxcmd & DMSGF_DELETE) == 0)
					break;
			}
		} else {
			trans = sc->open_tag->state;
		}
		if (trans == NULL)
			goto skip;

		KKASSERT(tag->bio);
		bio = tag->bio;
		bp = bio->bio_buf;

		switch(bp->b_cmd) {
		case BUF_CMD_READ:
			msg = kdmsg_msg_alloc(trans,
					      DMSG_BLK_READ |
					      DMSGF_CREATE |
					      DMSGF_DELETE,
					      xa_bio_completion, tag);
			msg->any.blk_read.keyid = sc->keyid;
			msg->any.blk_read.offset = bio->bio_offset;
			msg->any.blk_read.bytes = bp->b_bcount;
			break;
		case BUF_CMD_WRITE:
			msg = kdmsg_msg_alloc(trans,
					      DMSG_BLK_WRITE |
					      DMSGF_CREATE | DMSGF_DELETE,
					      xa_bio_completion, tag);
			msg->any.blk_write.keyid = sc->keyid;
			msg->any.blk_write.offset = bio->bio_offset;
			msg->any.blk_write.bytes = bp->b_bcount;
			msg->aux_data = bp->b_data;
			msg->aux_size = bp->b_bcount;
			break;
		case BUF_CMD_FLUSH:
			msg = kdmsg_msg_alloc(trans,
					      DMSG_BLK_FLUSH |
					      DMSGF_CREATE | DMSGF_DELETE,
					      xa_bio_completion, tag);
			msg->any.blk_flush.keyid = sc->keyid;
			msg->any.blk_flush.offset = bio->bio_offset;
			msg->any.blk_flush.bytes = bp->b_bcount;
			break;
		case BUF_CMD_FREEBLKS:
			msg = kdmsg_msg_alloc(trans,
					      DMSG_BLK_FREEBLKS |
					      DMSGF_CREATE | DMSGF_DELETE,
					      xa_bio_completion, tag);
			msg->any.blk_freeblks.keyid = sc->keyid;
			msg->any.blk_freeblks.offset = bio->bio_offset;
			msg->any.blk_freeblks.bytes = bp->b_bcount;
			break;
		default:
			bp->b_flags |= B_ERROR;
			bp->b_error = EIO;
			devstat_end_transaction_buf(&sc->stats, bp);
			atomic_add_int(&xa_active, -1);
			biodone(bio);
			tag->bio = NULL;
			break;
		}
	}

	/*
	 * If no msg was allocated we likely could not find a good span.
	 */
skip:
	if (msg) {
		/*
		 * Message was passed in or constructed.
		 */
		tag->state = msg->state;
		lockmgr(&sc->lk, LK_RELEASE);
		kdmsg_msg_write(msg);
		lockmgr(&sc->lk, LK_EXCLUSIVE);
	} else if (tag->bio &&
		   (tag->bio->bio_buf->b_flags & B_FAILONDIS) == 0) {
		/*
		 * No spans available but BIO is not allowed to fail
		 * on connectivity problems.  Requeue the BIO.
		 */
		TAILQ_INSERT_TAIL(&sc->bioq, tag->bio, bio_act);
		tag->bio = NULL;
		lockmgr(&sc->lk, LK_RELEASE);
		xa_done(tag, 1);
		lockmgr(&sc->lk, LK_EXCLUSIVE);
	} else {
		/*
		 * No spans available, bio is allowed to fail.
		 */
		lockmgr(&sc->lk, LK_RELEASE);
		tag->status.head.error = DMSG_ERR_IO;
		xa_done(tag, 1);
		lockmgr(&sc->lk, LK_EXCLUSIVE);
	}
}

static uint32_t
xa_wait(xa_tag_t *tag)
{
	xa_softc_t *sc = tag->sc;
	uint32_t error;

	lockmgr(&sc->lk, LK_EXCLUSIVE);
	tag->waiting = 1;
	while (tag->done == 0)
		lksleep(tag, &sc->lk, 0, "xawait", 0);
	lockmgr(&sc->lk, LK_RELEASE);

	error = tag->status.head.error;
	tag->waiting = 0;
	xa_release(tag, 0);

	return error;
}

static void
xa_done(xa_tag_t *tag, int wasbio)
{
	KKASSERT(tag->bio == NULL);

	tag->state = NULL;
	tag->done = 1;
	if (tag->waiting)
		wakeup(tag);
	if (tag->async)
		xa_release(tag, wasbio);
}

/*
 * Release a tag.  If everything looks ok and there are pending BIOs
 * (due to all tags in-use), we can use the tag to start the next BIO.
 * Do not try to restart if the connection is currently failed.
 */
static
void
xa_release(xa_tag_t *tag, int wasbio)
{
	xa_softc_t *sc = tag->sc;
	struct bio *bio;

	if ((bio = tag->bio) != NULL) {
		struct buf *bp = bio->bio_buf;

		bp->b_error = EIO;
		bp->b_flags |= B_ERROR;
		devstat_end_transaction_buf(&sc->stats, bp);
		atomic_add_int(&xa_active, -1);
		biodone(bio);
		tag->bio = NULL;
	}

	lockmgr(&sc->lk, LK_EXCLUSIVE);

	if (wasbio && sc->open_tag &&
	    (bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
		TAILQ_REMOVE(&sc->bioq, bio, bio_act);
		tag->bio = bio;
		xa_start(tag, NULL, 1);
	} else {
		TAILQ_REMOVE(&sc->tag_pendq, tag, entry);
		TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
	}
	lockmgr(&sc->lk, LK_RELEASE);
}

/*
 * Handle messages under the BLKOPEN transaction.
 */
static int
xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
{
	xa_tag_t *tag = state->any.any;
	xa_softc_t *sc;
	struct bio *bio;

	/*
	 * If the tag has been cleaned out we already closed our side
	 * of the transaction and we are waiting for the other side to
	 * close.
	 */
	xa_printf(1, "xa_sync_completion: tag %p msg %08x state %p\n",
		  tag, msg->any.head.cmd, msg->state);

	if (tag == NULL) {
		if (msg->any.head.cmd & DMSGF_CREATE)
			kdmsg_state_reply(state, DMSG_ERR_LOSTLINK);
		return 0;
	}
	sc = tag->sc;

	/*
	 * Validate the tag
	 */
	lockmgr(&sc->lk, LK_EXCLUSIVE);

	/*
	 * Handle initial response to our open and restart any deferred
	 * BIOs on success.
	 *
	 * NOTE: DELETE may also be set.
	 */
	if (msg->any.head.cmd & DMSGF_CREATE) {
		switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
		case DMSG_LNK_ERROR | DMSGF_REPLY:
			bzero(&tag->status, sizeof(tag->status));
			tag->status.head = msg->any.head;
			break;
		case DMSG_BLK_ERROR | DMSGF_REPLY:
			tag->status = msg->any.blk_error;
			break;
		}
		sc->last_error = tag->status.head.error;
		xa_printf(1, "blk_open completion status %d\n",
			  sc->last_error);
		if (sc->last_error == 0) {
			while ((bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
				tag = xa_setup_cmd(sc, NULL);
				if (tag == NULL)
					break;
				TAILQ_REMOVE(&sc->bioq, bio, bio_act);
				tag->bio = bio;
				xa_start(tag, NULL, 1);
			}
		}
		sc->serializing = 0;
		wakeup(sc);
	}

	/*
	 * Handle unexpected termination (or lost comm channel) from other
	 * side.  Autonomous completion only if open_tag matches,
	 * otherwise another thread is probably waiting on the tag.
	 *
	 * (see xa_close() for other interactions)
	 */
	if (msg->any.head.cmd & DMSGF_DELETE) {
		kdmsg_state_reply(tag->state, 0);
		if (sc->open_tag == tag) {
			sc->open_tag = NULL;
			xa_done(tag, 0);
		} else {
			tag->async = 0;
			xa_done(tag, 0);
		}
	}
	lockmgr(&sc->lk, LK_RELEASE);

	return (0);
}

static int
xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
{
	xa_tag_t *tag = state->any.any;
	xa_softc_t *sc = tag->sc;
	struct bio *bio;
	struct buf *bp;

	/*
	 * Get the bio from the tag.  If no bio is present we just do
	 * 'done' handling.
	 */
	if ((bio = tag->bio) == NULL)
		goto handle_done;
	bp = bio->bio_buf;

	/*
	 * Process return status
	 */
	switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
	case DMSG_LNK_ERROR | DMSGF_REPLY:
		bzero(&tag->status, sizeof(tag->status));
		tag->status.head = msg->any.head;
		if (tag->status.head.error)
			tag->status.resid = bp->b_bcount;
		else
			tag->status.resid = 0;
		break;
	case DMSG_BLK_ERROR | DMSGF_REPLY:
		tag->status = msg->any.blk_error;
		break;
	}

	/*
	 * If the device is open stall the bio on DMSG errors.  If an
	 * actual I/O error occured on the remote device, DMSG_ERR_IO
	 * will be returned.
	 */
	if (tag->status.head.error &&
	    (msg->any.head.cmd & DMSGF_DELETE) && sc->opencnt) {
		if (tag->status.head.error != DMSG_ERR_IO)
			goto handle_repend;
	}

	/*
	 * Process bio completion
	 *
	 * For reads any returned data is zero-extended if necessary, so
	 * the server can short-cut any all-zeros reads if it desires.
	 */
	switch(bp->b_cmd) {
	case BUF_CMD_READ:
		if (msg->aux_data && msg->aux_size) {
			if (msg->aux_size < bp->b_bcount) {
				bcopy(msg->aux_data, bp->b_data, msg->aux_size);
				bzero(bp->b_data + msg->aux_size,
				      bp->b_bcount - msg->aux_size);
			} else {
				bcopy(msg->aux_data, bp->b_data, bp->b_bcount);
			}
		} else {
			bzero(bp->b_data, bp->b_bcount);
		}
		/* fall through */
	case BUF_CMD_WRITE:
	case BUF_CMD_FLUSH:
	case BUF_CMD_FREEBLKS:
	default:
		if (tag->status.resid > bp->b_bcount)
			tag->status.resid = bp->b_bcount;
		bp->b_resid = tag->status.resid;
		if (tag->status.head.error != 0) {
			bp->b_error = EIO;
			bp->b_flags |= B_ERROR;
		} else {
			bp->b_resid = 0;
		}
		devstat_end_transaction_buf(&sc->stats, bp);
		atomic_add_int(&xa_active, -1);
		biodone(bio);
		tag->bio = NULL;
		break;
	}

	/*
	 * Handle completion of the transaction.  If the bioq is not empty
	 * we can initiate another bio on the same tag.
	 *
	 * NOTE: Most of our transactions will be single-message
	 *	 CREATE+DELETEs, so we won't have to terminate the
	 *	 transaction separately, here.  But just in case they
	 *	 aren't be sure to terminate the transaction.
	 */
handle_done:
	if (msg->any.head.cmd & DMSGF_DELETE) {
		xa_done(tag, 1);
		if ((state->txcmd & DMSGF_DELETE) == 0)
			kdmsg_msg_reply(msg, 0);
	}
	return (0);

	/*
	 * Handle the case where the transaction failed due to a
	 * connectivity issue.  The tag is put away with wasbio=0
	 * and we put the BIO back onto the bioq for a later restart.
	 *
	 * probe I/Os (where the device is not open) will be failed
	 * instead of requeued.
	 */
handle_repend:
	tag->bio = NULL;
	if (bio->bio_buf->b_flags & B_FAILONDIS) {
		xa_printf(1, "xa_strategy: lost link, fail probe bp %p\n",
			  bio->bio_buf);
		bio->bio_buf->b_error = ENXIO;
		bio->bio_buf->b_flags |= B_ERROR;
		biodone(bio);
		bio = NULL;
	} else {
		xa_printf(1, "xa_strategy: lost link, requeue bp %p\n",
			  bio->bio_buf);
	}
	xa_done(tag, 0);
	if ((state->txcmd & DMSGF_DELETE) == 0)
		kdmsg_msg_reply(msg, 0);

	/*
	 * Requeue the bio
	 */
	if (bio) {
		lockmgr(&sc->lk, LK_EXCLUSIVE);
		TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
		lockmgr(&sc->lk, LK_RELEASE);
	}
	return (0);
}

/*
 * Restart as much deferred I/O as we can.  The serializer is set and we
 * eat it (clear it) when done.
 *
 * Called with sc->lk held
 */
static
void
xa_restart_deferred(xa_softc_t *sc)
{
	kdmsg_state_t *span;
	kdmsg_msg_t *msg;
	xa_tag_t *tag;
	int error;

	KKASSERT(sc->serializing);

	/*
	 * Determine if a restart is needed.
	 */
	if (sc->opencnt == 0) {
		/*
		 * Device is not open, nothing to do, eat serializing.
		 */
		sc->serializing = 0;
		wakeup(sc);
	} else if (sc->open_tag == NULL) {
		/*
		 * BLK_OPEN required before we can restart any BIOs.
		 * Select the best LNK_SPAN to issue the BLK_OPEN under.
		 *
		 * serializing interlocks waiting open()s.
		 */
		error = 0;
		TAILQ_FOREACH(span, &sc->spanq, user_entry) {
			if ((span->rxcmd & DMSGF_DELETE) == 0)
				break;
		}
		if (span == NULL)
			error = ENXIO;

		if (error == 0) {
			tag = xa_setup_cmd(sc, NULL);
			if (tag == NULL)
				error = ENXIO;
		}
		if (error == 0) {
			sc->open_tag = tag;
			msg = kdmsg_msg_alloc(span,
					      DMSG_BLK_OPEN |
					      DMSGF_CREATE,
					      xa_sync_completion, tag);
			msg->any.blk_open.modes = DMSG_BLKOPEN_RD;
			xa_printf(1,
				  "BLK_OPEN tag %p state %p "
				  "span-state %p\n",
				  tag, msg->state, span);
			xa_start(tag, msg, 0);
		}
		if (error) {
			sc->serializing = 0;
			wakeup(sc);
		}
		/* else leave serializing set until BLK_OPEN response */
	} else {
		/* nothing to do */
		sc->serializing = 0;
		wakeup(sc);
	}
}