/*
 * Client for 9P file servers.
 *
 * Services calls that operate on a remote file tree.
 *
 * Requests are simply forwarded to the fs for processing.
 * The fs should NOT receive version/auth/attach requests. Those are
 * handled by the fs itself while being created.
 *
 * Attach, and recovery, are performed by the reader proc,
 * because this operation might block.
 *
 * This does NOT rewrite fids, and does NOT fix up qids. That's the
 * work of volfs. fs only services the 9P RPC
 *
 * From outside, a dead fs can be seen because it has a -1 fd and
 * always responds with Eio.
 *
 * BUG: Once created, a fs stays around. If there's an I/O error, the file
 * descriptor is closed, but the processes are kept running.
 */

#include <u.h>
#include <libc.h>
#include <fcall.h>
#include <thread.h>
#include <auth.h>
#include <ip.h>
#include "names.h"
#include "vols.h"

enum {
	Request	= 0,
	Reply	= 1,
	Tick	= 2,
};

static QLock	fslck;
static Fs**	fss;
static int	nfss;
static uvlong	qidgen = 1;
static Lock	qidgenlck;
uvlong	Mvolqid	=	0x00000b4900000000ULL;	// B49

char	Eopen[] = "Fid already open";
char	Enotopen[] = "Fid not open";
char	Eio[] = "io error";
char	Eperm[] = "permission denied";
char	Enotexist[]="file does not exist";
char	Eauth[]	= "authentication not required";
char	Euser[]	= "permission denied for that user";
char	Espec[]	= "bad volume specifier";
char	Edupfid[]="fid already in use";
char	Ebadfid[]="bad fid";

static ulong	ndead, ncame, ntmout;
static ulong	nsrvrpcs[Tmax];
static ulong	nclirpcs[Tmax];
static char*	names[Tmax] = {
	[Tattach]	"attach",
	[Twalk]		"walk",
	[Topen]		"open",
	[Tclunk]	"clunk",
	[Tremove]	"remove",
	[Tread]		"read",
	[Twrite]	"write",
	[Tstat]		"stat",
	[Twstat]	"wstat",
	[Tflush]	"flush",
};

void
dumpfss(void)
{
	int	i;

	fprint(2, "fss:\n");
	qlock(&fslck);
	for (i = 0; i < nfss; i++){
		fprint(2, "   %p %s '%s' t=%ld ref=%ld fd=%d pid=%d e=%ld %ldms\n",
			fss[i], fss[i]->addr, fss[i]->spec,fss[i]->time, 
			fss[i]->ref, fss[i]->fd, fss[i]->pid, fss[i]->epoch,
			fss[i]->lat);
		if (fss[i]->fid)
			fprint(2, "\tQ=%llx %X\n", fss[i]->qid.path, fss[i]->fid);
	}
	qunlock(&fslck);
}

void
fsstats(void)
{
	int	i;
	ulong	tsrv, tcli;

	fprint(2, "fs stats:\n");
	fprint(2, "%ld deaths %ld recovers %ld tmouts\n", ndead, ncame, ntmout);
	tsrv = tcli = 0;
	for (i = 0; i < Tmax; i++){
		tsrv += nsrvrpcs[i];
		tcli += nsrvrpcs[i];
		if (names[i])
			fprint(2, "%s:\t%5ld cli\t%5ld srv\n",
				names[i], nclirpcs[i], nsrvrpcs[i]);
	}
	fprint(2,"total:\t%5ld cli\t%5ld srv\n", tcli, tsrv);
}

/* All calls to particular file systems must come through this.
 * This also dispatches to the empty volume and to the ctl fs.
 */
int
fsop(Fs* fs, Frpc* op)
{
	ulong r;

	if (fs == nil){
		// Fake, empty Fs used to present an empty dir
		// when a vol has a dead fs.
		return nilfsops[op->f.type](nil, op);
	}
	if (fs == &ctlfs){
		// fake fs to service the ctl file
		return ctlfsops[op->f.type](nil, op);
	}
	sendp(fs->reqc, op);
	r = recvul(op->sc);
	if (r == ~0)
		r = -1;
	return r;
}

void
fstimer(ulong now)
{
	int	i;
	Fs*	fs;

	qlock(&fslck);
	for (i = 0; i < nfss; i++){
		fs = fss[i];
		nbsendul(fs->tmrc, now);
	}
	qunlock(&fslck);
}

int
putfcall(int fd, Frpc* m)
{
	int	n;

	nsrvrpcs[m->f.type]++;
	Dbgprint(m->fid, "   <= %F\n", &m->f);
	n = convS2M(&m->f, m->sbuf, sizeof(m->sbuf));
	if (n > 0)
		n = write(fd, m->sbuf, n);
	else
		n = -1;
	return n;
}

int
putfrep(int fd, Frpc* m)
{
	int	n;

	if (m->r.type == Rstat && m->d){
		m->r.nstat = convD2M(m->d, m->buf, sizeof(m->buf));
		m->r.stat = m->buf;
		free(m->d);
		m->d = nil;
	}
	dbgprint(m->fid, "-> %F\n", &m->r);
	n = convS2M(&m->r, m->sbuf, sizeof(m->sbuf));
	if (n > 0)
		n = write(fd, m->sbuf, n);
	return n;
}

int
getfcall(int fd, Frpc* m)
{
	int	n;

	n = read9pmsg(fd, m->buf, sizeof(m->buf));
	if (n <= 0)
		return n;
	if (convM2S(m->buf, n, &m->f) == 0)
		return -1;
	nclirpcs[m->f.type]++;
	return n;
}

int
getfrep(int fd, Frpc* m)
{
	int	n;
	char*	p;

	n = read9pmsg(fd, m->buf, sizeof(m->buf));
	if (n <= 0)
		return n;
	if (convM2S(m->buf, n, &m->r) == 0)
		return -1;
	switch(m->r.type){
	case Rstat:
		free(m->d);
		m->d = emalloc(sizeof(Dir)+128);
		p = (char*)(m->d + 1);
		convM2D(m->r.stat, m->r.nstat, m->d, p);
		/* lib9p may insist that name is "/", which is not
		 * a valid file name
		 */
		if (m->d->name[0] == '/' && m->d->name[1] == 0)
			m->d->name = "dev";
		break;
	case Rerror:
		if (!strncmp(m->r.ename, "io ", 3))
			n = -1;
		break;
	}
	return n;
}

void
rpcerr(Frpc* fop, char* err)
{
	fop->r.type = Rerror;
	fop->r.tag = fop->f.tag;
	fop->r.ename = fop->err;
	strecpy(fop->err, fop->err+sizeof(fop->err), err);
}

static void
fsmuxrep(Fs* fs, Frpc* rep)
{
	Frpc*	fop;
	Frpc**	fl;
	int	e;

	for (fl = &fs->rpcs; fop = *fl; fl = &fop->next)
		if (fop->f.tag == rep->r.tag)
			break;
	if (fop == nil){
		rpcfree(rep);
		return;	// no tag. request was aborted.
	}
	*fl = fop->next;
	fop->next = nil;
	if (rep->r.type == Rread)
		memmove(fop->buf, rep->buf, rep->r.count);
	fop->r = rep->r;
	free(fop->d);
	fop->d = rep->d;
	rep->d = nil;
	if (fop->rep)
		rpcfree(fop->rep);
	fop->rep = rep;
	Dbgprint(fop->fid, "   => %F\n", &rep->r);
	e = 0;
	if (fop->r.type == Rerror){
		strecpy(fop->err, fop->err+sizeof(fop->err), fop->r.ename);
		fop->r.ename = fop->err;
		e = ~0;
	}
	sendul(fop->sc, e);	// awake the thread doing the rpc.
}

static int
maytimeout(Frpc* fop)
{
	/* This is here to experiment with removal of
	 * timeouts. Use with care. It may block the whole thing
	 * if a broken/faulty fs is not being timed out.
	 *
	 * As it is now, we timeout all requests but for those in
	 * volumes mounted with -T in spec that might block on
	 * devices like portfs: Tread.
	 */
	if (fop->flushed)
		return 1;

	if (fop->fid != nil && fop->fid->notimeout && fop->f.type == Tread)
		if (fop->fid->qid.path != 0LL)
			return (fop->fid->qid.type&QTDIR);
		else
			return 0;

	return 1;
}

static void
fstmouts(Fs* fs)
{
	Frpc*	fop;
	Frpc**	fl;
	int	some;

	some = 0;
	for (fl = &fs->rpcs; fop = *fl; ){
		if (maytimeout(fop) && fs->time - fop->time > Rpctmout){
			dprint(2, "timeout req tag %d type=%d ot=%uld ft=%uld\n",
				fop->f.tag, fop->f.type, fop->time, fs->time);
			ntmout++;
			*fl = fop->next;
			fop->next = nil;
			if (fop->flushed){
				rpcerr(fop, "interrupt");
			} else {
				if (fs->time - fop->time > 3 * Rpctmout){
					// don't abort the fs in this case.
					// It's likely that RPCs were very
					// old ones but the fs has recover.
					dprint(2, "%s: preposterous PRC tag %d type=%d\n", fs->addr, fop->f.tag, fop->f.type);
				} else
					some++;
				rpcerr(fop, Eio);
			}
			sendul(fop->sc, ~0);
		} else
			fl = &fop->next;
	}
	if (some){
		close(fs->fd);	// Shut this down to clean things up.
		fs->fd = -1;
		incref(&epoch);
	}
}

static void
abortrpcs(Fs* fs)
{
	Frpc*	fop;
	Frpc**	fl;

	for (fl = &fs->rpcs; fop = *fl; ){
		*fl = fop->next;
		fop->next = nil;
		rpcerr(fop, Eio);
		sendul(fop->sc, ~0);
	}
}


static void
fsreadproc(void*a)
{
	Fs*	fs = a;
	Frpc*	rep;
	int	r;
	vlong	t0, t1;
	long	tmout;

	threadsetname("fsread");

	assert(fs->fd < 0);
	fs->lat = Biglatency;
	t0 = nsec();
	if (!amountfs(fs))
		vdprint(2, "failed to mount %s %s\n", fs->addr, fs->spec);
	t1 = nsec();
	fs->lat = (t1 - t0)/1000000;
	tmout = 3000;
	for(;;){
		if (fs->fd < 0){
			vdprint(2, "trying %s %s\n", fs->addr, fs->spec);
			if (!amountfs(fs)){
				vdprint(2, "amount failed: %r\n");
				if (!fs->musthave && tmout < Trytmout * 1000)
					tmout += 1000;
				sleep(tmout);
				continue;
			}
			ncame++;
			vdprint(2, "got fs: %s %s\n", fs->addr, fs->spec);
			incref(&epoch);
			fs->epoch = epoch.ref;
		}
		tmout = 1000;
		rep = rpcalloc();
		r = getfrep(fs->fd, rep);
		if (r < 0){ 			// we're dying.
			vdprint(2, "readproc: dying\n");
			ndead++;
			close(fs->fd);
			fs->fd = -1;
			rpcfree(rep);
			rep = nil;		// Eio to all pending rpcs
			incref(&epoch);	
			fs->epoch = epoch.ref;
		}
		sendp(fs->repc, rep);
	}
	dprint(2, "fsreadproc exiting\n");
	threadexits(nil);
}


void
fsmuxthread(void*arg)
{
	Fs*	fs = arg;
	Frpc*	fop;
	Frpc*	rep;
	ulong	now;
	Alt	a[] = {	{ nil, &fop, CHANRCV},
			{ nil, &rep, CHANRCV},
			{ nil, &now, CHANRCV},
			{ nil, 0, CHANEND }};

	threadsetname("fsmux");
	a[0].c = fs->reqc;
	a[1].c = fs->repc;
	a[2].c = fs->tmrc;
	now = fs->time = time(nil);
	for(;;){
		rep = nil;
		switch(alt(a)){
		case Request:
			if (fs->fd < 0){
				rpcerr(fop, Eio);
				sendul(fop->sc, ~0);
			} else {
				fop->next = fs->rpcs;
				fs->rpcs = fop;
				fop->time = fs->time;
				putfcall(fs->fd, fop);
			}
			break;
		case Reply:
			if (rep != nil)
				fsmuxrep(fs, rep);
			else
				abortrpcs(fs);	// Eio to them all.
			break;
		case Tick:
			fs->time = now;
			fstmouts(fs);
			break;
		default:
			sysfatal("fsmux: alt failed\n");
		}
	}
	fprint(2, "fsmux exiting\n");
	threadexits(nil);
}

void
newfsqid(Qid* q)
{
	lock(&qidgenlck);
	qidgen++;
	q->path = (Mvolqid|(qidgen<<44));
	q->type = QTDIR;
	q->vers = 0;
	unlock(&qidgenlck);
}

static Fs*
addfs(char* addr, char* spec)
{
	int	i;
	Fs*	fs;

	vdprint(2, "adding fs for addr %s spec '%s'\n", addr, spec);
	if ((nfss%16) == 0)
		fss = erealloc(fss, sizeof(Fs*)*(nfss+16));
	i = nfss;
	fs = fss[i] = emalloc(sizeof(Fs));
	memset(fs, 0, sizeof(Fs));
	fs->addr = estrdup(addr);
	fs->spec = estrdup(spec);
	fs->reqc = chancreate(sizeof(Frpc*), 0);
	fs->repc = chancreate(sizeof(Frpc*), 0);
	fs->tmrc = chancreate(sizeof(ulong), 0);
	fs->fid = fidalloc(0);
	fs->fid->fs = fs;	// sic
	fs->fd = -1;
	fs->fdqid.path = -1;	// something weird for a qid.
	newfsqid(&fs->qid);
	/* A call to getfs made by the user may lead to this.
	 * The caller will already count its ref
	 */
	fs->ref = 0;
	fs->epoch = epoch.ref;
	fs->fid->epoch = fs->epoch;
	threadcreate(fsmuxthread, fs, Stack);
	fs->lat = Biglatency;
	fs->pid = proccreate(fsreadproc, fs, Stack);
	nfss++;

	/* We do not give fsreadproc much chance of mounting the fs.
	 * We don't do that ourselves and we dont synchronize
	 * because that might block all the program.
	 * For volume mounts that must be there for the system to
	 * work, mustmount should be specified.
	 * This is ugly, but waiting for all known volumes to be
	 * mounted may slow booting a lot. Also, after the mount
	 * request, the next 9p call is likely to find working vols
	 * already mounted.
	 */
	sleep(1);	// gives a chance.
	return fs;
}

Fs*
getfs(char* addr, char* spec)
{
	int	i;
	Fs*	c;

	qlock(&fslck);
	c = nil;
	for (i = 0; i < nfss; i++){
		if (!strcmp(addr, fss[i]->addr) && !strcmp(spec, fss[i]->spec)){
			c = fss[i];
			break;
		}
	}
	if (c == nil)
		c = addfs(addr, spec);
	if (c)
		incref(c);
	qunlock(&fslck);
	return c;
}

void
putfs(Fs* c)
{
	// The fs always stay there, once mounted.
	// It's likely that we will use them again and we keep them
	// already attached. To change this, unmount when no more refs.
	if (c)
		decref(c);
}