plan9port/src/cmd/fossil/disk.c

#include "stdinc.h"
#include "dat.h"
#include "fns.h"
#include "error.h"

static void diskThread(void *a);

enum {
	/*
	 * disable measurement since it gets alignment faults on BG
	 * and the guts used to be commented out.
	 */
	Timing	= 0,			/* flag */
	QueueSize = 100,		/* maximum block to queue */
};

struct Disk {
	QLock lk;
	int ref;

	int fd;
	Header h;

	Rendez flow;
	Rendez starve;
	Rendez flush;
	Rendez die;

	int nqueue;

	Block *cur;		/* block to do on current scan */
	Block *next;		/* blocks to do next scan */
};

/* keep in sync with Part* enum in dat.h */
static char *partname[] = {
	[PartError]	= "error",
	[PartSuper]	= "super",
	[PartLabel]	= "label",
	[PartData]	= "data",
	[PartVenti]	= "venti",
};

Disk *
diskAlloc(int fd)
{
	u8int buf[HeaderSize];
	Header h;
	Disk *disk;

	if(pread(fd, buf, HeaderSize, HeaderOffset) < HeaderSize){
		werrstr("short read: %r");
		return nil;
	}

	if(!headerUnpack(&h, buf)){
		werrstr("bad disk header");
		return nil;
	}
	disk = vtmallocz(sizeof(Disk));
	disk->starve.l = &disk->lk;
	disk->flow.l = &disk->lk;
	disk->flush.l = &disk->lk;
	disk->fd = fd;
	disk->h = h;

	disk->ref = 2;
	proccreate(diskThread, disk, STACK);

	return disk;
}

void
diskFree(Disk *disk)
{
	diskFlush(disk);

	/* kill slave */
	qlock(&disk->lk);
	disk->die.l = &disk->lk;
	rwakeup(&disk->starve);
	while(disk->ref > 1)
		rsleep(&disk->die);
	qunlock(&disk->lk);
	close(disk->fd);
	vtfree(disk);
}

static u32int
partStart(Disk *disk, int part)
{
	switch(part){
	default:
		assert(0);
	case PartSuper:
		return disk->h.super;
	case PartLabel:
		return disk->h.label;
	case PartData:
		return disk->h.data;
	}
}


static u32int
partEnd(Disk *disk, int part)
{
	switch(part){
	default:
		assert(0);
	case PartSuper:
		return disk->h.super+1;
	case PartLabel:
		return disk->h.data;
	case PartData:
		return disk->h.end;
	}
}

int
diskReadRaw(Disk *disk, int part, u32int addr, uchar *buf)
{
	ulong start, end;
	u64int offset;
	int n, nn;

	start = partStart(disk, part);
	end = partEnd(disk, part);

	if(addr >= end-start){
		werrstr(EBadAddr);
		return 0;
	}

	offset = ((u64int)(addr + start))*disk->h.blockSize;
	n = disk->h.blockSize;
	while(n > 0){
		nn = pread(disk->fd, buf, n, offset);
		if(nn < 0){
			werrstr("%r");
			return 0;
		}
		if(nn == 0){
			werrstr("eof reading disk");
			return 0;
		}
		n -= nn;
		offset += nn;
		buf += nn;
	}
	return 1;
}

int
diskWriteRaw(Disk *disk, int part, u32int addr, uchar *buf)
{
	ulong start, end;
	u64int offset;
	int n;

	start = partStart(disk, part);
	end = partEnd(disk, part);

	if(addr >= end - start){
		werrstr(EBadAddr);
		return 0;
	}

	offset = ((u64int)(addr + start))*disk->h.blockSize;
	n = pwrite(disk->fd, buf, disk->h.blockSize, offset);
	if(n < 0){
		werrstr("%r");
		return 0;
	}
	if(n < disk->h.blockSize) {
		werrstr("short write");
		return 0;
	}

	return 1;
}

static void
diskQueue(Disk *disk, Block *b)
{
	Block **bp, *bb;

	qlock(&disk->lk);
	while(disk->nqueue >= QueueSize)
		rsleep(&disk->flow);
	if(disk->cur == nil || b->addr > disk->cur->addr)
		bp = &disk->cur;
	else
		bp = &disk->next;

	for(bb=*bp; bb; bb=*bp){
		if(b->addr < bb->addr)
			break;
		bp = &bb->ionext;
	}
	b->ionext = bb;
	*bp = b;
	if(disk->nqueue == 0)
		rwakeup(&disk->starve);
	disk->nqueue++;
	qunlock(&disk->lk);
}


void
diskRead(Disk *disk, Block *b)
{
	assert(b->iostate == BioEmpty || b->iostate == BioLabel);
	blockSetIOState(b, BioReading);
	diskQueue(disk, b);
}

void
diskWrite(Disk *disk, Block *b)
{
	assert(b->nlock == 1);
	assert(b->iostate == BioDirty);
	blockSetIOState(b, BioWriting);
	diskQueue(disk, b);
}

void
diskWriteAndWait(Disk *disk, Block *b)
{
	int nlock;

	/*
	 * If b->nlock > 1, the block is aliased within
	 * a single thread.  That thread is us.
	 * DiskWrite does some funny stuff with QLock
	 * and blockPut that basically assumes b->nlock==1.
	 * We humor diskWrite by temporarily setting
	 * nlock to 1.  This needs to be revisited.
	 */
	nlock = b->nlock;
	if(nlock > 1)
		b->nlock = 1;
	diskWrite(disk, b);
	while(b->iostate != BioClean)
		rsleep(&b->ioready);
	b->nlock = nlock;
}

int
diskBlockSize(Disk *disk)
{
	return disk->h.blockSize;	/* immuttable */
}

int
diskFlush(Disk *disk)
{
	Dir dir;

	qlock(&disk->lk);
	while(disk->nqueue > 0)
		rsleep(&disk->flush);
	qunlock(&disk->lk);

	/* there really should be a cleaner interface to flush an fd */
	nulldir(&dir);
	if(dirfwstat(disk->fd, &dir) < 0){
		werrstr("%r");
		return 0;
	}
	return 1;
}

u32int
diskSize(Disk *disk, int part)
{
	return partEnd(disk, part) - partStart(disk, part);
}

static uintptr
mypc(int x)
{
	return getcallerpc(&x);
}

static char *
disk2file(Disk *disk)
{
	static char buf[256];

#ifndef PLAN9PORT
	if (fd2path(disk->fd, buf, sizeof buf) < 0)
		strncpy(buf, "GOK", sizeof buf);
#endif
	return buf;
}

static void
diskThread(void *a)
{
	Disk *disk = a;
	Block *b;
	uchar *buf, *p;
	double t;
	int nio;

	threadsetname("disk");

//fprint(2, "diskThread %d\n", getpid());

	buf = vtmalloc(disk->h.blockSize);

	qlock(&disk->lk);
	if (Timing) {
		nio = 0;
		t = -nsec();
	}
	for(;;){
		while(disk->nqueue == 0){
			if (Timing) {
				t += nsec();
				if(nio >= 10000){
					fprint(2, "disk: io=%d at %.3fms\n",
						nio, t*1e-6/nio);
					nio = 0;
					t = 0;
				}
			}
			if(disk->die.l != nil)
				goto Done;
			rsleep(&disk->starve);
			if (Timing)
				t -= nsec();
		}
		assert(disk->cur != nil || disk->next != nil);

		if(disk->cur == nil){
			disk->cur = disk->next;
			disk->next = nil;
		}
		b = disk->cur;
		disk->cur = b->ionext;
		qunlock(&disk->lk);

		/*
		 * no one should hold onto blocking in the
		 * reading or writing state, so this lock should
		 * not cause deadlock.
		 */
if(0)fprint(2, "fossil: diskThread: %d:%d %x\n", getpid(), b->part, b->addr);
		bwatchLock(b);
		qlock(&b->lk);
		b->pc = mypc(0);
		assert(b->nlock == 1);
		switch(b->iostate){
		default:
			abort();
		case BioReading:
			if(!diskReadRaw(disk, b->part, b->addr, b->data)){
				fprint(2, "fossil: diskReadRaw failed: %s: "
					"score %V: part=%s block %ud: %r\n",
					disk2file(disk), b->score,
					partname[b->part], b->addr);
				blockSetIOState(b, BioReadError);
			}else
				blockSetIOState(b, BioClean);
			break;
		case BioWriting:
			p = blockRollback(b, buf);
			/* NB: ctime result ends with a newline */
			if(!diskWriteRaw(disk, b->part, b->addr, p)){
				fprint(2, "fossil: diskWriteRaw failed: %s: "
				    "score %V: date %s part=%s block %ud: %r\n",
					disk2file(disk), b->score,
					ctime(time(0)),
					partname[b->part], b->addr);
				break;
			}
			if(p != buf)
				blockSetIOState(b, BioClean);
			else
				blockSetIOState(b, BioDirty);
			break;
		}

		blockPut(b);		/* remove extra reference, unlock */
		qlock(&disk->lk);
		disk->nqueue--;
		if(disk->nqueue == QueueSize-1)
			rwakeup(&disk->flow);
		if(disk->nqueue == 0)
			rwakeup(&disk->flush);
		if(Timing)
			nio++;
	}
Done:
//fprint(2, "diskThread done\n");
	disk->ref--;
	rwakeup(&disk->die);
	qunlock(&disk->lk);
	vtfree(buf);
}