shithub: ridefs

ref: 38f2878c44f584d472560c9a426d01cdd3fb99cb
dir: /ridefs.c/

#include <u.h>
#include <libc.h>
#include <fcall.h>
#include <thread.h>
#include <9p.h>
#include <stdio.h>
#include <json.h>


typedef struct Client Client;
typedef struct Rfid Rfid;

struct Client {
	Ref;

	char *addr;
	ulong umask;
	int timeout;

	/* internal use */
	char *user;
	char *pres;     /* previous result */
	char *rinfo;
	char *qctl;
	long time0;
	int id;
	int oio;        /* io opened */
	int iopid;      /* read/write fork */
	int fd;         /* data */
	int cfd;        /* ctl */
};

enum {
	Qroot,
		Qrctl,
		Qclone,
		Qclient,
			Qctl,
			Qio,
			Qrinfo,
	QCOUNT
};

static char *nametab[] = {
	"/",
		"ctl",
		"clone",
		nil,
			"ctl",
			"io",
			"rinfo",
};

struct Rfid {
	Ref;
	int kind;
	int client;
};

#define VERSION 0
static Client *cpool;
static char *mtpt;
static char *service;
static char *net;
static char *user;
static char *qrctl;
static ulong umask;
static long bufsz;
static long time0;
static uint nclients;
static int debug;
static int timeout;

void*
ecalloc(ulong n){
	void *p;

	p = emalloc9p(n);
	setmalloctag(p, getcallerpc(&n));
	memset(p, 0, n);

	return p;
}

void*
erealloc(void *p, ulong n){
	p = erealloc9p(p, n);
	setrealloctag(p, getcallerpc(&n));
	return p;
}

char*
estrdup(char *s){
	if(s == nil)
		return nil;

	s = estrdup9p(s);
	setrealloctag(s, getcallerpc(&s));
	return s;
}

Client*
clientref(int i){
	if(i < 0 || i > nclients)
		return nil;
	return &cpool[i];
}

int
genqrctl(void){
	sprintf(qrctl,
		"version %i\n"
		"bufsz %u\n"
		"nclients %u\n"
		"debug %i\n"
		"timeout %i\n"
		"umask 0%hlo\n",
		VERSION, bufsz, nclients,
		debug, timeout, umask);

	return strlen(qrctl);
}

int
genqctl(int client){
	Client *c;

	c = clientref(client);
	sprintf(c->qctl,
		"%i\n"
		"connect %s\n"
		"timeout %i\n"
		"umask 0%hlo\n",
		client, c->addr, c->timeout, c->umask);

	return strlen(c->qctl);
}

int
mkclient(void){
	Client *c;
	int i;

	for(i = 0; i < nclients; i++)
		if(cpool[i].ref == 0)
			break;
	if(i == nclients)
		return -1;
	c = &cpool[i];

	incref(c);
	c->id = i;
	c->timeout = timeout;
	c->umask = umask;
	c->user = estrdup(getuser());
	c->time0 = time(0);
	c->qctl = ecalloc(bufsz);
	genqctl(i);

	return i;
}

void
rmclient(int i){
	Client *c;

	c = clientref(i);
	if(c == nil || decref(c))
		return;

	if(c->user) free(c->user);
	if(c->pres) free(c->pres);
	if(c->qctl) free(c->qctl);
	if(c->rinfo) free(c->rinfo);
	if(c->fd) close(c->fd);
	if(c->cfd) close(c->cfd);

	memset(c, 0, sizeof(*c));
}

static void
mkqid(Qid *q, int k, int client){
	q->vers = VERSION;
	q->path = ((u64int)client<<32) | k&0xffffffff;

	switch(k){
	case Qroot:
	case Qclient:
		q->type = QTDIR; break;
	default:
		q->type = QTFILE;
	}
}

long
writemsg(int fd, void *pld, long n){
	long len;
	char *r;

	len = n+8;
	r = ecalloc(len+8);
	r[0] = 24>>len & 0xff;
	r[1] = 16>>len & 0xff;
	r[2] =  8>>len & 0xff;
	r[3] =     len & 0xff;
	r[4] = 'R';
	r[5] = 'I';
	r[6] = 'D';
	r[7] = 'E';
	memcpy(&r[8], pld, n);
	len = write(fd, r, len);

	return len;
}

long
readmsg(int fd, char *pld, long n){
	int len;
	char buf[9];

	if(0 > readn(fd, buf, 8))
		return -1;
	buf[9] = '\0';
	if(0 != strcmp(&buf[4], "RIDE"))
		return -2;
	len = -8 + (buf[0]<<24 | buf[1]<<16 | buf[2]<<8 | buf[3]);
	if(len > n)
		return -3;
	if(0 > readn(fd, pld, len))
		return -4;

	pld = erealloc(pld, len+1);
	pld[len] = '\0';
	return len;
}

char *
rideinit(int client){
	int fd;
	char *addr, *buf, *s, *e;
	Client *c;
	JSON *j;
	JSONEl *d;

	c = clientref(client);
	addr = netmkaddr(c->addr, net, "tcp");
	if((fd = dial(addr, nil, nil, &c->cfd)) < 0)
		return "failed to dial addr";
	c = clientref(client);
	c->fd = fd;

	buf = ecalloc(bufsz);
	if(0 > readmsg(fd, buf, bufsz))
		return "failed to read handshake";
	if(0 != strcmp(buf, "SupportedProtocols=2"))
		return "unrecognized protocol";
	free(buf);

	buf = "UsingProtocol=2";
	if(0 > writemsg(fd, buf, strlen(buf)))
		return "failed to write handshake";
		
	buf = "[\"Identify\",{\"apiVersion\":1,\"identity\":1}]";
	if(0 > writemsg(fd, buf, strlen(buf)))
		return "failed to send identification message";

	buf = ecalloc(bufsz);
	if(0 > readmsg(fd, buf, bufsz))
		return "failed to receive identification message";
	j = jsonparse(buf);
	free(buf);
	if(j == nil || j->t != JSONArray || nil == j->first)
		return "unrecognized reply";
	if(nil == (s = jsonstr(j->first->val)) || 0 != strcmp(s, "ReplyIdentify"))
		return "unexpected identification reply";
	if(nil == (d = j->first->next) || d->val->t != JSONObject)
		return "malformed identification reply";

	c->rinfo = s = ecalloc(bufsz);
	buf = ecalloc(bufsz);
	e = s + bufsz-1;
	for(; d != nil; d = d->next){
		switch(d->val->t){
		case JSONBool:
		case JSONNumber:
			sprintf(buf, "%s %i\n", d->name, d->val->n); break;
		case JSONString:
			sprintf(buf, "%s %s\n", d->name, d->val->s); break;
		}
		s = strecpy(s, e, buf);
	}
	free(buf);

	jsonfree(j);
	return nil;
}

long
writeqrctl(char *b, long){
	char *s;
	char sep[7];

	sprintf(sep, " \t\n\f\r\v");
	for(s = strtok(b, sep); s != nil; s = strtok(nil, sep)){
		if(strcmp(s, "bufsz") == 0)
			bufsz = strtoul(strtok(nil, sep), nil, 0);
		if(strcmp(s, "umask") == 0)
			umask = strtoul(strtok(nil, sep), nil, 0);
		if(strcmp(s, "nclients") == 0)
			nclients = strtoul(strtok(nil, sep), nil, 0);
		if(strcmp(s, "debug") == 0)
			debug = atoi(strtok(nil, sep));
		if(strcmp(s, "timeout") == 0)
			timeout = atoi(strtok(nil, sep));
	}

	cpool = erealloc(cpool, nclients*sizeof(*cpool));
	genqrctl();

	return strlen(qrctl);
}

long
writeqctl(char *b, long n, int client){
	Client *c;
	char *s, sep[7];

	c = clientref(client);
	sprintf(sep, " \t\n\f\r\v");
	for(s = strtok(b, sep); s != nil; s = strtok(nil, sep)){
		if(strcmp(s, "connect") == 0)
			c->addr = estrdup(strtok(nil, sep));
		if(strcmp(s, "timeout") == 0)
			c->timeout = atoi(strtok(nil, sep));
		if(strcmp(s, "umask") == 0)
			c->umask = strtoul(strtok(nil, sep), nil, 0);
	}

	genqctl(client);

	return n;
}

long
writeqio(char *b, long, int client){
	Client *c;
	char *pld, *e, *p, *z;
	long sz;

	c = clientref(client);
	pld = z = ecalloc(bufsz);
	e = pld + bufsz - 1;
	sz = -1;
	if(p == (z = strecpy(p = z, e, "[\"Execute\",{\"text\":\"      ")))
		goto end;
	if(p == (z = strecpy(p = z, e, b)))
		goto end;
	if(p == strecpy(p = z, e, "\",\"trace\":0}]"))
		goto end;

	sz = strlen(pld);
	writemsg(c->fd, pld, sz);
	readmsg(c->fd, pld, bufsz);
	c->pres = estrdup(pld);

	end:
	free(pld);
	return sz;
}

void
mkdirent(Dir *d, int kind, int client){
	Client *c;
	char *nm;

	memset(d, 0, sizeof(*d));
	mkqid(&d->qid, kind, client);
	d->mode = 0444 & umask;
	if(nil != (nm = nametab[kind]))
		d->name = estrdup(nm);
	if(kind == Qclient){
		nm = ecalloc(bufsz);
		sprintf(nm, "%i", client);
		d->name = estrdup(nm);
	}
	if(d->qid.type & QTDIR)
		d->mode |= DMDIR | 0111;

	c = clientref(client);
	switch(kind){
	case Qrctl:   d->mode = 0666 & umask; break;
	case Qctl:
	case Qio:     d->mode = 0666 & c->umask; break;
	}

	switch(kind){
	case Qroot:
	case Qrctl:
	case Qclone:
		d->atime = d->mtime = time0;
		d->uid = estrdup(user);
		d->gid = estrdup(user);
		d->muid = estrdup(user);
		break;
	default:
		d->atime = d->mtime = c->time0;
		d->uid = estrdup(c->user);
		d->gid = estrdup(c->user);
		d->muid = estrdup(c->user);
		break;
	}

	switch(kind){
	case Qrctl: d->length = strlen(qrctl); break;
	case Qctl: d->length = strlen(c->qctl); break;
	case Qio: d->length = c->pres == nil ? 0 : strlen(c->pres); break;
	case Qrinfo: d->length = c->rinfo == nil ? 0 : strlen(c->rinfo); break;
	}
}

int
genqroot(int i, Dir *d, void*){
	static int n;
	int j;

	i += Qroot + 1;
	if(i < Qclient){
		mkdirent(d, i, -1);
	} else {
		i -= Qclient;
		if(i == 0)
			n = 0;
		for(j = n; j < nclients && cpool[j].ref == 0; j++);
		if(j == nclients)
			return -1;
		n++;
		mkdirent(d, Qclient, j);
	}

	return 0;
}

int
genqclient(int i, Dir *d, void *aux){
	int client;

	client = (vlong)aux;
	i += Qclient + 1;
	if(i >= QCOUNT)
		return -1;
	mkdirent(d, i, client);
	return 0;
}

static void
fsdestroyfid(Fid *fid){
	Rfid *f;

	f = fid->aux;
	if(f == nil)
		return;

	rmclient(f->client);
	free(f);
}

static void
fsstart(Srv*){
	if(mtpt != nil)
		unmount(nil, mtpt);
}

static void
fsend(Srv*){
	postnote(PNGROUP, getpid(), "shutdown");
	exits(nil);
}

static void
fsattach(Req *r){
	Rfid *f;

	f = ecalloc(sizeof(*f));
	f->kind = Qroot;
	f->client = -1;

	mkqid(&r->fid->qid, f->kind, f->client);
	r->fid->aux = f;
	r->ofcall.qid = r->fid->qid;

	respond(r, nil);
}

static void
fsopen(Req *r){
	int e, pid;
	char *res;
	Rfid *f;
	Client *c;

	f = r->fid->aux;
	c = clientref(f->client);

	switch(f->kind){
	case Qclone:
		if((f->client = mkclient()) == -1){
			respond(r, "reached client limit");
			return;
		}

		f->kind = Qctl;

		mkqid(&r->ofcall.qid, f->kind, f->client);
		r->fid->qid = r->ofcall.qid;

		respond(r, nil);
		break;
	case Qio:
		if(e = c->oio)
			respond(r, "client in use");
		else if(e = (nil == c->addr))
			respond(r, "no server set");
		if(e)
			return;

		switch(pid = rfork(RFPROC|RFNOWAIT|RFMEM)){
		case 0:
			alarm(timeout);
			res = rideinit(f->client);
			alarm(0);
			if(res == nil)
				c->oio = 1;
			c->iopid = 0;
			respond(r, res);
			break;
		case -1:
			respond(r, "failed to init ride");
			break;
		default:
			c->iopid = pid;
		}
		break;
	default:
		respond(r, nil);
	}
}

static void
fsread(Req *r){
	Rfid *f;
	Client *c;
	char *buf;

	buf = nil;
	f = r->fid->aux;
	c = clientref(f->client);
	switch(f->kind){
	case Qroot: dirread9p(r, genqroot, nil); break;
	case Qrctl: buf = estrdup(qrctl); break;
	case Qclone: respond(r, "read prohibited"); return;
	case Qclient: dirread9p(r, genqclient, (void*)f->client); break;
	case Qctl: buf = estrdup(c->qctl); break;
	case Qio: buf = estrdup(c->pres); break;
	case Qrinfo: buf = estrdup(c->rinfo); break;
	}

	if(buf != nil){
		readbuf(r, buf, strlen(buf));
		free(buf);
	}

	respond(r, nil);
}

static void
fswrite(Req *r){
	Rfid *f;
	Client *c;
	int pid;
	char *d, *err;
	long n;

	d = r->ifcall.data;
	n = r->ifcall.count;
	f = r->fid->aux;
	c = clientref(f->client);
	err = nil;

	switch(f->kind){
	case Qrctl:
		r->ofcall.count = writeqrctl(d, n);
		break;
	case Qctl:
		r->ofcall.count = writeqctl(d, n, f->client);
		break;
	case Qio:
		switch(pid = rfork(RFPROC|RFNOWAIT|RFMEM)){
		case 0:
			alarm(c->timeout);
			r->ofcall.count = writeqio(d, n, f->client);
			alarm(0);
			c->iopid = 0;
			break;
		case -1:
			err = "failed to fork write";
			break;
		default:
			c->iopid = pid;
		}
		break;
	default:
		err = "write prohibited";
	}

	respond(r, err);
}

static void
fsflush(Req *r){
	Req *o;
	Rfid *f;
	Client *c;

	if(o = r->oldreq)
	if(f = o->fid->aux)
	if(c = clientref(f->client))
	if(0 < c->iopid){
		postnote(PNPROC, c->iopid, "interrupt");
		respond(r, "interrupted");
	}

	respond(r, nil);
}

static void
fsstat(Req *r){
	Rfid *f;

	f = r->fid->aux;
	mkdirent(&r->d, f->kind, f->client);

	respond(r, nil);
}

static char*
fswalk1(Fid *fid, char *name, Qid *qid){
	Rfid *f;
	Client *c;
	int i, n;
	char *nend;

	if(!(fid->qid.type&QTDIR))
		return "cannot walk from non-directory";

	f = fid->aux;
	n = -1;
	if(strcmp(name, "..") == 0){
		switch(f->kind){
		case Qroot:
			break;
		case Qclient:
			rmclient(f->client);
			f->client = -1;
			break;
		default:
			if(f->kind > Qclient)
				f->kind = Qclient;
		}
	} else {
		for(i = f->kind+1; i<QCOUNT; i++){
			if(nametab[i] && strcmp(name, nametab[i]) == 0)
				break;
			if(i == Qclient){
				n = strtol(name, &nend, 10);
				c = clientref(n);
				if(*nend == 0 && c != nil && c->ref != 0){
					incref(c);
					f->client = n;
					break;
				}	
			}
		}
		if(i >= QCOUNT)
			return "directory entry not found";
		f->kind = i;
	}
	mkqid(qid, f->kind, n);
	fid->qid = *qid;
	return nil;
}

static char*
fsclone(Fid *oldfid, Fid *newfid){
	Rfid *f, *o;

	o = oldfid->aux;
	if(o == nil)
		return "bad fid";

	f = ecalloc(sizeof(*f));
	memmove(f, o, sizeof(*f));

	if(f->client > -1)
		incref(clientref(f->client));
	newfid->aux = f;

	return nil;
}

Srv fs = {
	.destroyfid = fsdestroyfid,
	.start      = fsstart,
	.end        = fsend,
	.attach     = fsattach,
	.open       = fsopen,
	.read       = fsread,
	.write      = fswrite,
	.flush      = fsflush,
	.stat       = fsstat,
	.walk1      = fswalk1,
	.clone      = fsclone,
};

void
usage(void){
	fprintf(stderr, "usage: %s [-Dd] [-T timeout] [-m mtpt] [-s service] [-x net]\n", argv0);
}

void
main(int argc, char **argv){
	user = getuser();
	mtpt = "/mnt/ride";
	umask = 0755;
	bufsz = 4096;
	time0 = time(0);
	nclients = 256;
	timeout = 10000;

	ARGBEGIN{
	case 'D': chatty9p++; break;
	case 'd': debug++; break;
	case 'T': timeout = atoi(EARGF(usage()));
	case 'm': mtpt = EARGF(usage()); break;
	case 's': service = EARGF(usage()); break;
	case 'x': net = EARGF(usage()); break;
	default:  usage(); return;
	}ARGEND

	cpool = ecalloc(nclients*sizeof(*cpool));
	qrctl = ecalloc(bufsz);
	genqrctl();

	if(debug)
		fprintf(stderr,
			"ridefs:\n"
			"\tchatty9p = %i\n"
			"\tdebug = %i\n"
			"\ttimeout = %i\n"
			"\tmtpt = %s\n"
			"\tservice = %s\n"
			"\tnet = %s\n"
			"\tfs = %p\n"
			"\tcpool = %p\n",
			chatty9p, debug, timeout,
			mtpt, service, net,
			&fs, cpool);

	rfork(RFNOTEG);
	postmountsrv(&fs, service, mtpt, MREPL);
	exits(nil);
}