Re: [9fans] simple venti demo:

[Russ Cox <rsc@swtch.com>]

i call this 'ventino'.  it's a tiny venti that keeps
the whole index in memory, backed by a text file.
i have not used it in a while (the file is dated may 25, 2009)
but hey, it's a working venti server in 329 lines of code.

#include <u.h>
#include <libc.h>
#include <bio.h>
#include <flate.h>
#include <thread.h>
#include <venti.h>
#include <libsec.h>

typedef uchar byte;
typedef u64int uint64;
typedef u32int uint32;

typedef struct IEntry IEntry;
struct IEntry
{
	IEntry *link;

	// disk data
	byte score[VtScoreSize];
	uint64 offset;
};

typedef struct Chunk Chunk;
struct Chunk
{
	byte score[VtScoreSize];
	uint32 size;
	byte *data;
};

IEntry **ihash;
uint nihash;
uint nientry;

void
rehash(void)
{
	IEntry **new, *e, *next;
	uint i, n;
	uint32 h;

	n = nihash<<1;
	new = vtmallocz(nihash*sizeof new[0]);
	for(i=0; i<nihash; i++) {
		for(e = ihash[i]; e; e = next) {
			next = e->link;
			h = *(uint32*)e->score & (n - 1);
			e->link = new[h];
			new[h] = e;
		}
	}
	free(ihash);
	ihash = new;
	nihash = n;
}

IEntry*
ilookup(byte *score)
{
	uint32 h;
	IEntry *e;

	// Not a great hash; assumes we are
	// seeing all blocks, not just some chosen subset.
	h = *(uint32*)score & (nihash - 1);

	for(e = ihash[h]; e; e = e->link)
		if(memcmp(e->score, score, VtScoreSize) == 0)
			return e;
	return nil;
}

IEntry*
iinsert(byte *score)
{
	uint32 h;
	IEntry *e;

	if(nihash < (1<<28) && nientry > 2*nihash)
		rehash();
	h = *(uint32*)score & (nihash - 1);
	e = vtmallocz(sizeof(IEntry));
	e->link = ihash[h];
	ihash[h] = e;
	memmove(e->score, score, VtScoreSize);
	return e;
}

void
iload(Biobuf *b)
{
	char *p;
	char *f[10];
	int nf;
	byte score[VtScoreSize];
	uint64 offset;
	IEntry *e;

	while((p = Brdline(b, '\n')) != nil) {
		p[Blinelen(b)-1] = '\0';
		nf = tokenize(p, f, nelem(f));
		if(nf != 2 || vtparsescore(f[0], nil, score) < 0 || (offset =
strtoull(f[1], 0, 0)) == 0) {
			sysfatal("malformed index");
			return;
		}
		e = iinsert(score);
		e->offset = offset;
	}
}

void
iwrite(int fd, IEntry *e)
{
	fprint(fd, "%V %-22llud\n", e->score, e->offset);
}

enum {
	ArenaBlock = 1<<30
};

uint64
dwrite(int fd, Chunk *c)
{
	byte *zdat, *w;
	int nzdat;
	uint nw;
	uint64 offset, eoffset;

	zdat = vtmallocz(c->size + 1024);
	nzdat = deflateblock(zdat, c->size + 1024, c->data, c->size, 6, 0);
	if(nzdat < 0 || nzdat > c->size - 512) {
		// don't bother with compression
		w = c->data;
		nw = c->size;
	} else {
		w = zdat;
		nw = nzdat;
	}
	offset = seek(fd, 0, 1);
	eoffset = offset + 2*VtScoreSize + 12 + 12 + 1 + nw;
	if(eoffset / ArenaBlock != offset / ArenaBlock) {
		offset /= ArenaBlock;
		offset++;
		offset *= ArenaBlock;
		seek(fd, offset, 0);
	}
	fprint(fd, "%V %-11ud %-11ud\n", c->score, c->size, nw);
	write(fd, w, nw);
	free(zdat);
	return offset;
}

void
dread(Biobuf *b, Chunk *c)
{
	char *p, *f[10];
	int nf;
	uint zsize;
	byte *r;
	uint64 offset;
	char buf[100];

	offset = Boffset(b);
	p = Brdline(b, '\n');
	if(p == nil || Blinelen(b) >= sizeof buf)
		sysfatal("malformed data - EOF");
	memmove(buf, p, Blinelen(b));
	buf[Blinelen(b)-1] = '\0';
	nf = tokenize(buf, f, nelem(f));
	if(nf != 3 || vtparsescore(f[0], nil, c->score) < 0 || (c->size =
strtoul(f[1], 0, 0)) == 0 || (zsize = strtoul(f[2], 0, 0)) == 0) {
		sysfatal("malformed data at %llud / %d", offset, nf);
		return;
	}
	c->data = vtmalloc(c->size);
	if(c->size == zsize)
		r = c->data;
	else
		r = vtmallocz(zsize);
	Bread(b, r, zsize);
	if(c->size != zsize) {
		if((nf = inflateblock(c->data, c->size, r, zsize)) < 0)
			sysfatal("inflateblock fail %d %d %d %.10H...", c->size, zsize, nf, r);
		free(r);
	}
}

Biobuf *bindexr;
Biobuf *bdatar;
int indexw, dataw;
Biobuf *bsha1;	// TODO
int doCreate;
int verbose;

void
doOpen(char *name, char *what, int *w, Biobuf **r)
{
	int fd;
	char buf[100];
	char *p;

	snprint(buf, sizeof buf, "# sventi %s\n", what);

	if((*w = open(name, OWRITE)) < 0) {
		if(!doCreate)
			sysfatal("open %s: %r", name);
		if((*w = create(name, OWRITE, 0644)) < 0)
			sysfatal("create %s: %r", name);
		write(*w, buf, strlen(buf));
	}
	if((fd = open(name, OREAD)) < 0)
		sysfatal("open %s: %r", name);
	*r = Bfdopen(fd, OREAD);
	p = Brdline(*r, '\n');
	if(p == nil || Blinelen(*r) != strlen(buf) || memcmp(p, buf, strlen(buf)) != 0)
		sysfatal("%s is not %s - %p", name, what);
}

void
usage(void)
{
	fprint(2, "usage: sventi [-cv] [-a address] [-d dir]\n");
	threadexitsall("usage");
}

void
threadmain(int argc, char **argv)
{
	VtReq *r;
	VtSrv *srv;
	char *address, *dir;
	Chunk c;
	IEntry *e;

	nihash = 16;
	ihash = vtmallocz(nihash * sizeof ihash[0]);

	fmtinstall('F', vtfcallfmt);
	fmtinstall('V', vtscorefmt);
	fmtinstall('H', encodefmt);

	address = "tcp!*!venti";
	dir = ".";
	deflateinit();
	inflateinit();

	ARGBEGIN{
	case 'v':
		verbose++;
		break;
	case 'a':
		address = EARGF(usage());
		break;
	case 'c':
		doCreate = 1;
		break;
	case 'd':
		dir = EARGF(usage());
		break;
	default:
		usage();
	}ARGEND

	if(argc != 0)
		usage();

	if(chdir(dir) < 0)
		sysfatal("chdir %s: %r", dir);

	doOpen("sventi.index", "index", &indexw,  &bindexr);
	doOpen("sventi.data", "data", &dataw, &bdatar);
	iload(bindexr);
	Bterm(bindexr);
	seek(indexw, 0, 2);
	seek(dataw, 0, 2);

	srv = vtlisten(address);
	if(srv == nil)
		sysfatal("vtlisten %s: %r", address);

	while((r = vtgetreq(srv)) != nil) {
		r->rx.msgtype = r->tx.msgtype+1;
		if(verbose)
			fprint(2, "<- %F\n", &r->tx);
		switch(r->tx.msgtype) {
		case VtTping:
			break;
		case VtTgoodbye:
			break;
		case VtTsync:
			fsync(dataw);
			fsync(indexw);
			break;
		case VtTread:
			e = ilookup(r->tx.score);
			if(e == nil) {
				r->rx.msgtype = VtRerror;
				r->rx.error = vtstrdup("block not found");
				break;
			}
			Bseek(bdatar, e->offset, 0);
			dread(bdatar, &c);
			if(memcmp(c.score, r->tx.score, VtScoreSize) != 0)
				sysfatal("data/index score mismatch");
			r->rx.data = packetforeign(c.data, c.size, free, c.data);
			packetsha1(r->rx.data, c.score);
			if(memcmp(c.score, r->tx.score, VtScoreSize) != 0)
				sysfatal("data/data score mismatch");
			break;
		case VtTwrite:
			packetsha1(r->tx.data, r->rx.score);
			e = ilookup(r->rx.score);
			if(e != nil)
				break;
			c.size = packetsize(r->tx.data);
			c.data = vtmalloc(c.size);
			packetconsume(r->tx.data, c.data, c.size);
			memmove(c.score, r->rx.score, VtScoreSize);
			e = iinsert(c.score);
			e->offset = dwrite(dataw, &c);
			free(c.data);
			iwrite(indexw, e);
			break;
		}
		if(verbose)
			fprint(2, "-> %F\n", &r->rx);
		vtrespond(r);
	}
	threadexitsall(nil);
}