Subversion Repositories planix.SVN

#include "stdinc.h"

#include "dat.h"

#include "fns.h"

typedef struct ASum ASum;

struct ASum

{

	Arena	*arena;

	ASum	*next;

};

static void	sealarena(Arena *arena);

static int	okarena(Arena *arena);

static int	loadarena(Arena *arena);

static CIBlock	*getcib(Arena *arena, int clump, int writing, CIBlock *rock);

static void	putcib(Arena *arena, CIBlock *cib);

static void	sumproc(void *);

static void loadcig(Arena *arena);

static QLock	sumlock;

static Rendez	sumwait;

static ASum	*sumq;

static ASum	*sumqtail;

static uchar zero[8192];

int	arenasumsleeptime;

int

initarenasum(void)

{

	needzeroscore();  /* OS X */

	sumwait.l = &sumlock;

	if(vtproc(sumproc, nil) < 0){

		seterr(EOk, "can't start arena checksum slave: %r");

		return -1;

	}

	return 0;

}

/*

 * make an Arena, and initialize it based upon the disk header and trailer.

 */

Arena*

initarena(Part *part, u64int base, u64int size, u32int blocksize)

{

	Arena *arena;

	arena = MKZ(Arena);

	arena->part = part;

	arena->blocksize = blocksize;

	arena->clumpmax = arena->blocksize / ClumpInfoSize;

	arena->base = base + blocksize;

	arena->size = size - 2 * blocksize;

	if(loadarena(arena) < 0){

		seterr(ECorrupt, "arena header or trailer corrupted");

		freearena(arena);

		return nil;

	}

	if(okarena(arena) < 0){

		freearena(arena);

		return nil;

	}

	if(arena->diskstats.sealed && scorecmp(zeroscore, arena->score)==0)

		sealarena(arena);

	return arena;

}

void

freearena(Arena *arena)

{

	if(arena == nil)

		return;

	free(arena);

}

Arena*

newarena(Part *part, u32int vers, char *name, u64int base, u64int size, u32int blocksize)

{

	int bsize;

	Arena *arena;

	if(nameok(name) < 0){

		seterr(EOk, "illegal arena name", name);

		return nil;

	}

	arena = MKZ(Arena);

	arena->part = part;

	arena->version = vers;

	if(vers == ArenaVersion4)

		arena->clumpmagic = _ClumpMagic;

	else{

		do

			arena->clumpmagic = fastrand();

		while(arena->clumpmagic==_ClumpMagic || arena->clumpmagic==0);

	}

	arena->blocksize = blocksize;

	arena->clumpmax = arena->blocksize / ClumpInfoSize;

	arena->base = base + blocksize;

	arena->size = size - 2 * blocksize;

	namecp(arena->name, name);

	bsize = sizeof zero;

	if(bsize > arena->blocksize)

		bsize = arena->blocksize;

	if(wbarena(arena)<0 || wbarenahead(arena)<0

	|| writepart(arena->part, arena->base, zero, bsize)<0){

		freearena(arena);

		return nil;

	}

	return arena;

}

int

readclumpinfo(Arena *arena, int clump, ClumpInfo *ci)

{

	CIBlock *cib, r;

	cib = getcib(arena, clump, 0, &r);

	if(cib == nil)

		return -1;

	unpackclumpinfo(ci, &cib->data->data[cib->offset]);

	putcib(arena, cib);

	return 0;

}

int

readclumpinfos(Arena *arena, int clump, ClumpInfo *cis, int n)

{

	CIBlock *cib, r;

	int i;

	/*

	 * because the clump blocks are laid out

	 * in reverse order at the end of the arena,

	 * it can be a few percent faster to read

	 * the clumps backwards, which reads the

	 * disk blocks forwards.

	 */

	for(i = n-1; i >= 0; i--){

		cib = getcib(arena, clump + i, 0, &r);

		if(cib == nil){

			n = i;

			continue;

		}

		unpackclumpinfo(&cis[i], &cib->data->data[cib->offset]);

		putcib(arena, cib);

	}

	return n;

}

/*

 * write directory information for one clump

 * must be called the arena locked

 */

int

writeclumpinfo(Arena *arena, int clump, ClumpInfo *ci)

{

	CIBlock *cib, r;

	cib = getcib(arena, clump, 1, &r);

	if(cib == nil)

		return -1;

	dirtydblock(cib->data, DirtyArenaCib);

	packclumpinfo(ci, &cib->data->data[cib->offset]);

	putcib(arena, cib);

	return 0;

}

u64int

arenadirsize(Arena *arena, u32int clumps)

{

	return ((clumps / arena->clumpmax) + 1) * arena->blocksize;

}

/*

 * read a clump of data

 * n is a hint of the size of the data, not including the header

 * make sure it won't run off the end, then return the number of bytes actually read

 */

u32int

readarena(Arena *arena, u64int aa, u8int *buf, long n)

{

	DBlock *b;

	u64int a;

	u32int blocksize, off, m;

	long nn;

	if(n == 0)

		return -1;

	qlock(&arena->lock);

	a = arena->size - arenadirsize(arena, arena->memstats.clumps);

	qunlock(&arena->lock);

	if(aa >= a){

		seterr(EOk, "reading beyond arena clump storage: clumps=%d aa=%lld a=%lld -1 clumps=%lld\n", arena->memstats.clumps, aa, a, arena->size - arenadirsize(arena, arena->memstats.clumps - 1));

		return -1;

	}

	if(aa + n > a)

		n = a - aa;

	blocksize = arena->blocksize;

	a = arena->base + aa;

	off = a & (blocksize - 1);

	a -= off;

	nn = 0;

	for(;;){

		b = getdblock(arena->part, a, OREAD);

		if(b == nil)

			return -1;

		m = blocksize - off;

		if(m > n - nn)

			m = n - nn;

		memmove(&buf[nn], &b->data[off], m);

		putdblock(b);

		nn += m;

		if(nn == n)

			break;

		off = 0;

		a += blocksize;

	}

	return n;

}

/*

 * write some data to the clump section at a given offset

 * used to fix up corrupted arenas.

 */

u32int

writearena(Arena *arena, u64int aa, u8int *clbuf, u32int n)

{

	DBlock *b;

	u64int a;

	u32int blocksize, off, m;

	long nn;

	int ok;

	if(n == 0)

		return -1;

	qlock(&arena->lock);

	a = arena->size - arenadirsize(arena, arena->memstats.clumps);

	if(aa >= a || aa + n > a){

		qunlock(&arena->lock);

		seterr(EOk, "writing beyond arena clump storage");

		return -1;

	}

	blocksize = arena->blocksize;

	a = arena->base + aa;

	off = a & (blocksize - 1);

	a -= off;

	nn = 0;

	for(;;){

		b = getdblock(arena->part, a, off != 0 || off + n < blocksize ? ORDWR : OWRITE);

		if(b == nil){

			qunlock(&arena->lock);

			return -1;

		}

		dirtydblock(b, DirtyArena);

		m = blocksize - off;

		if(m > n - nn)

			m = n - nn;

		memmove(&b->data[off], &clbuf[nn], m);

		ok = 0;

		putdblock(b);

		if(ok < 0){

			qunlock(&arena->lock);

			return -1;

		}

		nn += m;

		if(nn == n)

			break;

		off = 0;

		a += blocksize;

	}

	qunlock(&arena->lock);

	return n;

}

/*

 * allocate space for the clump and write it,

 * updating the arena directory

ZZZ question: should this distinguish between an arena

filling up and real errors writing the clump?

 */

u64int

writeaclump(Arena *arena, Clump *c, u8int *clbuf)

{

	DBlock *b;

	u64int a, aa;

	u32int clump, n, nn, m, off, blocksize;

	int ok;

	n = c->info.size + ClumpSize + U32Size;

	qlock(&arena->lock);

	aa = arena->memstats.used;

	if(arena->memstats.sealed

	|| aa + n + U32Size + arenadirsize(arena, arena->memstats.clumps + 1) > arena->size){

		if(!arena->memstats.sealed){

			logerr(EOk, "seal memstats %s", arena->name);

			arena->memstats.sealed = 1;

			wbarena(arena);

		}

		qunlock(&arena->lock);

		return TWID64;

	}

	if(packclump(c, &clbuf[0], arena->clumpmagic) < 0){

		qunlock(&arena->lock);

		return TWID64;

	}

	/*

	 * write the data out one block at a time

	 */

	blocksize = arena->blocksize;

	a = arena->base + aa;

	off = a & (blocksize - 1);

	a -= off;

	nn = 0;

	for(;;){

		b = getdblock(arena->part, a, off != 0 ? ORDWR : OWRITE);

		if(b == nil){

			qunlock(&arena->lock);

			return TWID64;

		}

		dirtydblock(b, DirtyArena);

		m = blocksize - off;

		if(m > n - nn)

			m = n - nn;

		memmove(&b->data[off], &clbuf[nn], m);

		ok = 0;

		putdblock(b);

		if(ok < 0){

			qunlock(&arena->lock);

			return TWID64;

		}

		nn += m;

		if(nn == n)

			break;

		off = 0;

		a += blocksize;

	}

	arena->memstats.used += c->info.size + ClumpSize;

	arena->memstats.uncsize += c->info.uncsize;

	if(c->info.size < c->info.uncsize)

		arena->memstats.cclumps++;

	clump = arena->memstats.clumps;

	if(clump % ArenaCIGSize == 0){

		if(arena->cig == nil){

			loadcig(arena);

			if(arena->cig == nil)

				goto NoCIG;

		}

		/* add aa as start of next cig */

		if(clump/ArenaCIGSize != arena->ncig){

			fprint(2, "bad arena cig computation %s: writing clump %d but %d cigs\n",

				arena->name, clump, arena->ncig);

			arena->ncig = -1;

			vtfree(arena->cig);

			arena->cig = nil;

			goto NoCIG;

		}

		arena->cig = vtrealloc(arena->cig, (arena->ncig+1)*sizeof arena->cig[0]);

		arena->cig[arena->ncig++].offset = aa;

	}

NoCIG:

	arena->memstats.clumps++;

	if(arena->memstats.clumps == 0)

		sysfatal("clumps wrapped");

	arena->wtime = now();

	if(arena->ctime == 0)

		arena->ctime = arena->wtime;

	writeclumpinfo(arena, clump, &c->info);

	wbarena(arena);

	qunlock(&arena->lock);

	return aa;

}

int

atailcmp(ATailStats *a, ATailStats *b)

{

	/* good test */

	if(a->used < b->used)

		return -1;

	if(a->used > b->used)

		return 1;

	/* suspect tests - why order this way? (no one cares) */

	if(a->clumps < b->clumps)

		return -1;

	if(a->clumps > b->clumps)

		return 1;

	if(a->cclumps < b->cclumps)

		return -1;

	if(a->cclumps > b->cclumps)

		return 1;

	if(a->uncsize < b->uncsize)

		return -1;

	if(a->uncsize > b->uncsize)

		return 1;

	if(a->sealed < b->sealed)

		return -1;

	if(a->sealed > b->sealed)

		return 1;

	/* everything matches */

	return 0;

}

void

setatailstate(AState *as)

{

	int i, j, osealed;

	Arena *a;

	Index *ix;

	trace(0, "setatailstate %s 0x%llux clumps %d", as->arena->name, as->aa, as->stats.clumps);

	/*

	 * Look up as->arena to find index.

	 */

	needmainindex();	/* OS X linker */

	ix = mainindex;

	for(i=0; i<ix->narenas; i++)

		if(ix->arenas[i] == as->arena)

			break;

	if(i==ix->narenas || as->aa < ix->amap[i].start || as->aa >= ix->amap[i].stop || as->arena != ix->arenas[i]){

		fprint(2, "funny settailstate 0x%llux\n", as->aa);

		return;

	}

	for(j=0; j<=i; j++){

		a = ix->arenas[j];

		if(atailcmp(&a->diskstats, &a->memstats) == 0)

			continue;

		qlock(&a->lock);

		osealed = a->diskstats.sealed;

		if(j == i)

			a->diskstats = as->stats;

		else

			a->diskstats = a->memstats;

		wbarena(a);

		if(a->diskstats.sealed != osealed && !a->inqueue)

			sealarena(a);

		qunlock(&a->lock);

	}

}

/*

 * once sealed, an arena never has any data added to it.

 * it should only be changed to fix errors.

 * this also syncs the clump directory.

 */

static void

sealarena(Arena *arena)

{

	arena->inqueue = 1;

	backsumarena(arena);

}

void

backsumarena(Arena *arena)

{

	ASum *as;

	if(sumwait.l == nil)

		return;

	as = MK(ASum);

	if(as == nil)

		return;

	qlock(&sumlock);

	as->arena = arena;

	as->next = nil;

	if(sumq)

		sumqtail->next = as;

	else

		sumq = as;

	sumqtail = as;

	rwakeup(&sumwait);

	qunlock(&sumlock);

}

static void

sumproc(void *unused)

{

	ASum *as;

	Arena *arena;

	USED(unused);

	for(;;){

		qlock(&sumlock);

		while(sumq == nil)

			rsleep(&sumwait);

		as = sumq;

		sumq = as->next;

		qunlock(&sumlock);

		arena = as->arena;

		free(as);

		sumarena(arena);

	}

}

void

sumarena(Arena *arena)

{

	ZBlock *b;

	DigestState s;

	u64int a, e;

	u32int bs;

	int t;

	u8int score[VtScoreSize];

	bs = MaxIoSize;

	if(bs < arena->blocksize)

		bs = arena->blocksize;

	/*

	 * read & sum all blocks except the last one

	 */

	flushdcache();

	memset(&s, 0, sizeof s);

	b = alloczblock(bs, 0, arena->part->blocksize);

	e = arena->base + arena->size;

	for(a = arena->base - arena->blocksize; a + arena->blocksize <= e; a += bs){

		disksched();

		while((t=arenasumsleeptime) == SleepForever){

			sleep(1000);

			disksched();

		}

		sleep(t);

		if(a + bs > e)

			bs = arena->blocksize;

		if(readpart(arena->part, a, b->data, bs) < 0)

			goto ReadErr;

		addstat(StatSumRead, 1);

		addstat(StatSumReadBytes, bs);

		sha1(b->data, bs, nil, &s);

	}

	/*

	 * the last one is special, since it may already have the checksum included

	 */

	bs = arena->blocksize;

	if(readpart(arena->part, e, b->data, bs) < 0){

ReadErr:

		logerr(EOk, "sumarena can't sum %s, read at %lld failed: %r", arena->name, a);

		freezblock(b);

		return;

	}

	addstat(StatSumRead, 1);

	addstat(StatSumReadBytes, bs);

	sha1(b->data, bs-VtScoreSize, nil, &s);

	sha1(zeroscore, VtScoreSize, nil, &s);

	sha1(nil, 0, score, &s);

	/*

	 * check for no checksum or the same

	 */

	if(scorecmp(score, &b->data[bs - VtScoreSize]) != 0

	&& scorecmp(zeroscore, &b->data[bs - VtScoreSize]) != 0)

		logerr(EOk, "overwriting mismatched checksums for arena=%s, found=%V calculated=%V",

			arena->name, &b->data[bs - VtScoreSize], score);

	freezblock(b);

	qlock(&arena->lock);

	scorecp(arena->score, score);

	wbarena(arena);

	qunlock(&arena->lock);

}

/*

 * write the arena trailer block to the partition

 */

int

wbarena(Arena *arena)

{

	DBlock *b;

	int bad;

	if((b = getdblock(arena->part, arena->base + arena->size, OWRITE)) == nil){

		logerr(EAdmin, "can't write arena trailer: %r");

		return -1;

	}

	dirtydblock(b, DirtyArenaTrailer);

	bad = okarena(arena)<0 || packarena(arena, b->data)<0;

	scorecp(b->data + arena->blocksize - VtScoreSize, arena->score);

	putdblock(b);

	if(bad)

		return -1;

	return 0;

}

int

wbarenahead(Arena *arena)

{

	ZBlock *b;

	ArenaHead head;

	int bad;

	namecp(head.name, arena->name);

	head.version = arena->version;

	head.size = arena->size + 2 * arena->blocksize;

	head.blocksize = arena->blocksize;

	head.clumpmagic = arena->clumpmagic;

	b = alloczblock(arena->blocksize, 1, arena->part->blocksize);

	if(b == nil){

		logerr(EAdmin, "can't write arena header: %r");

/* ZZZ add error message? */

		return -1;

	}

	/*

	 * this writepart is okay because it only happens

	 * during initialization.

	 */

	bad = packarenahead(&head, b->data)<0 ||

	      writepart(arena->part, arena->base - arena->blocksize, b->data, arena->blocksize)<0 ||

	      flushpart(arena->part)<0;

	freezblock(b);

	if(bad)

		return -1;

	return 0;

}

/*

 * read the arena header and trailer blocks from disk

 */

static int

loadarena(Arena *arena)

{

	ArenaHead head;

	ZBlock *b;

	b = alloczblock(arena->blocksize, 0, arena->part->blocksize);

	if(b == nil)

		return -1;

	if(readpart(arena->part, arena->base + arena->size, b->data, arena->blocksize) < 0){

		freezblock(b);

		return -1;

	}

	if(unpackarena(arena, b->data) < 0){

		freezblock(b);

		return -1;

	}

	if(arena->version != ArenaVersion4 && arena->version != ArenaVersion5){

		seterr(EAdmin, "unknown arena version %d", arena->version);

		freezblock(b);

		return -1;

	}

	scorecp(arena->score, &b->data[arena->blocksize - VtScoreSize]);

	if(readpart(arena->part, arena->base - arena->blocksize, b->data, arena->blocksize) < 0){

		logerr(EAdmin, "can't read arena header: %r");

		freezblock(b);

		return 0;

	}

	if(unpackarenahead(&head, b->data) < 0)

		logerr(ECorrupt, "corrupted arena header: %r");

	else if(namecmp(arena->name, head.name)!=0

	     || arena->clumpmagic != head.clumpmagic

	     || arena->version != head.version

	     || arena->blocksize != head.blocksize

	     || arena->size + 2 * arena->blocksize != head.size){

		if(namecmp(arena->name, head.name)!=0)

			logerr(ECorrupt, "arena tail name %s head %s", 

				arena->name, head.name);

		else if(arena->clumpmagic != head.clumpmagic)

			logerr(ECorrupt, "arena %d tail clumpmagic 0x%lux head 0x%lux",

				debugarena, (ulong)arena->clumpmagic,

				(ulong)head.clumpmagic);

		else if(arena->version != head.version)

			logerr(ECorrupt, "arena tail version %d head version %d",

				arena->version, head.version);

		else if(arena->blocksize != head.blocksize)

			logerr(ECorrupt, "arena tail block size %d head %d",

				arena->blocksize, head.blocksize);

		else if(arena->size+2*arena->blocksize != head.size)

			logerr(ECorrupt, "arena tail size %lud head %lud",

				(ulong)arena->size+2*arena->blocksize, head.size);

		else

			logerr(ECorrupt, "arena header inconsistent with arena data");

	}

	freezblock(b);

	return 0;

}

static int

okarena(Arena *arena)

{

	u64int dsize;

	int ok;

	ok = 0;

	dsize = arenadirsize(arena, arena->diskstats.clumps);

	if(arena->diskstats.used + dsize > arena->size){

		seterr(ECorrupt, "arena %s used > size", arena->name);

		ok = -1;

	}

	if(arena->diskstats.cclumps > arena->diskstats.clumps)

		logerr(ECorrupt, "arena %s has more compressed clumps than total clumps", arena->name);

	/*

	 * This need not be true if some of the disk is corrupted.

	 *

	if(arena->diskstats.uncsize + arena->diskstats.clumps * ClumpSize + arena->blocksize < arena->diskstats.used)

		logerr(ECorrupt, "arena %s uncompressed size inconsistent with used space %lld %d %lld", arena->name, arena->diskstats.uncsize, arena->diskstats.clumps, arena->diskstats.used);

	 */

	/*

	 * this happens; it's harmless.

	 *

	if(arena->ctime > arena->wtime)

		logerr(ECorrupt, "arena %s creation time after last write time", arena->name);

	 */

	return ok;

}

static CIBlock*

getcib(Arena *arena, int clump, int writing, CIBlock *rock)

{

	int mode;

	CIBlock *cib;

	u32int block, off;

	if(clump >= arena->memstats.clumps){

		seterr(EOk, "clump directory access out of range");

		return nil;

	}

	block = clump / arena->clumpmax;

	off = (clump - block * arena->clumpmax) * ClumpInfoSize;

	cib = rock;

	cib->block = block;

	cib->offset = off;

	if(writing){

		if(off == 0 && clump == arena->memstats.clumps-1)

			mode = OWRITE;

		else

			mode = ORDWR;

	}else

		mode = OREAD;

	cib->data = getdblock(arena->part,

		arena->base + arena->size - (block + 1) * arena->blocksize, mode);

	if(cib->data == nil)

		return nil;

	return cib;

}

static void

putcib(Arena *arena, CIBlock *cib)

{

	USED(arena);

	putdblock(cib->data);

	cib->data = nil;

}

/*

 * For index entry readahead purposes, the arenas are 

 * broken into smaller subpieces, called clump info groups

 * or cigs.  Each cig has ArenaCIGSize clumps (ArenaCIGSize

 * is chosen to make the index entries take up about half

 * a megabyte).  The index entries do not contain enough

 * information to determine what the clump index is for

 * a given address in an arena.  That info is needed both for

 * figuring out which clump group an address belongs to 

 * and for prefetching a clump group's index entries from

 * the arena table of contents.  The first time clump groups

 * are accessed, we scan the entire arena table of contents

 * (which might be 10s of megabytes), recording the data 

 * offset of each clump group.

 */

/* 

 * load clump info group information by scanning entire toc.

 */

static void

loadcig(Arena *arena)

{

	u32int i, j, ncig, nci;

	ArenaCIG *cig;

	ClumpInfo *ci;

	u64int offset;

	int ms;

	if(arena->cig || arena->ncig < 0)

		return;

//	fprint(2, "loadcig %s\n", arena->name);

	ncig = (arena->memstats.clumps+ArenaCIGSize-1) / ArenaCIGSize;

	if(ncig == 0){

		arena->cig = vtmalloc(1);

		arena->ncig = 0;

		return;

	}

	ms = msec();

	cig = vtmalloc(ncig*sizeof cig[0]);

	ci = vtmalloc(ArenaCIGSize*sizeof ci[0]);

	offset = 0;

	for(i=0; i<ncig; i++){

		nci = readclumpinfos(arena, i*ArenaCIGSize, ci, ArenaCIGSize);

		cig[i].offset = offset;

		for(j=0; j<nci; j++)

			offset += ClumpSize + ci[j].size;

		if(nci < ArenaCIGSize){

			if(i != ncig-1){

				vtfree(ci);

				vtfree(cig);

				arena->ncig = -1;

				fprint(2, "loadcig %s: got %ud cigs, expected %ud\n", arena->name, i+1, ncig);

				goto out;

			}

		}

	}

	vtfree(ci);

	arena->ncig = ncig;

	arena->cig = cig;

out:

	ms = msec() - ms;

	addstat2(StatCigLoad, 1, StatCigLoadTime, ms);

}

/*

 * convert arena address into arena group + data boundaries.

 */

int

arenatog(Arena *arena, u64int addr, u64int *gstart, u64int *glimit, int *g)

{

	int r, l, m;

	qlock(&arena->lock);

	if(arena->cig == nil)

		loadcig(arena);

	if(arena->cig == nil || arena->ncig == 0){

		qunlock(&arena->lock);

		return -1;

	}

	l = 1;

	r = arena->ncig - 1;

	while(l <= r){

		m = (r + l) / 2;

		if(arena->cig[m].offset <= addr)

			l = m + 1;

		else

			r = m - 1;

	}

	l--;

	*g = l;

	*gstart = arena->cig[l].offset;

	if(l+1 < arena->ncig)

		*glimit = arena->cig[l+1].offset;

	else

		*glimit = arena->memstats.used;

	qunlock(&arena->lock);

	return 0;

}

/*

 * load the clump info for group g into the index entries.

 */

int