/*  * This file is part of the UCB release of Plan 9. It is subject to the license * terms in the LICENSE file found in the top-level directory of this * distribution and at http://akaros.cs.berkeley.edu/files/Plan9License. No * part of the UCB release of Plan 9, including this file, may be copied, * modified, propagated, or distributed except according to the terms contained * in the LICENSE file. */ /* * Write the dirty icache entries to disk.  Random seeks are * so expensive that it makes sense to wait until we have * a lot and then just make a sequential pass over the disk. */#include "stdinc.h"#include "dat.h"#include "fns.h" static void icachewriteproc(void*);static void icachewritecoord(void*);static IEntry *iesort(IEntry*); int icachesleeptime = 1000;	/* milliseconds */int minicachesleeptime = 0; enum{	Bufsize = 8*1024*1024}; typedef struct IWrite IWrite;struct IWrite{	Round round;	AState as;}; static IWrite iwrite; voidiniticachewrite(void){	int i;	Index *ix; 	initround(&iwrite.round, "icache", 120*60*1000);	ix = mainindex;	for(i=0; i<ix->nsects; i++){		ix->sects[i]->writechan = chancreate(sizeof(ulong), 1);		ix->sects[i]->writedonechan = chancreate(sizeof(ulong), 1);		vtproc(icachewriteproc, ix->sects[i]);	}	vtproc(icachewritecoord, nil);	vtproc(delaykickroundproc, &iwrite.round);} static u64intie2diskaddr(Index *ix, ISect *is, IEntry *ie){	u64int bucket, addr; 	bucket = hashbits(ie->score, 32)/ix->div;	addr = is->blockbase + ((bucket - is->start) << is->blocklog);	return addr;} static IEntry*nextchunk(Index *ix, ISect *is, IEntry **pie, u64int *paddr, uint *pnbuf){	u64int addr, naddr;	uint nbuf;	int bsize;	IEntry *iefirst, *ie, **l; 	bsize = 1<<is->blocklog;	iefirst = *pie;	addr = ie2diskaddr(ix, is, iefirst);	nbuf = 0;	for(l = &iefirst->nextdirty; (ie = *l) != nil; l = &(*l)->nextdirty){		naddr = ie2diskaddr(ix, is, ie);		if(naddr - addr >= Bufsize)			break;		nbuf = naddr - addr;	}	nbuf += bsize; 	*l = nil;	*pie = ie;	*paddr = addr;	*pnbuf = nbuf;	return iefirst;}	static inticachewritesect(Index *ix, ISect *is, u8int *buf){	int err, i, werr, h, bsize, t;	u32int lo, hi;	u64int addr, naddr;	uint nbuf, off;	DBlock *b;	IBucket ib;	IEntry *ie, *iedirty, **l, *chunk; 	lo = is->start * ix->div;	if(TWID32/ix->div < is->stop)		hi = TWID32;	else		hi = is->stop * ix->div - 1; 	trace(TraceProc, "icachewritesect enter %ud %ud %llud",		lo, hi, iwrite.as.aa); 	iedirty = icachedirty(lo, hi, iwrite.as.aa);	iedirty = iesort(iedirty);	bsize = 1 << is->blocklog;	err = 0; 	while(iedirty){		disksched();		while((t = icachesleeptime) == SleepForever){			sleep(1000);			disksched();		}		if(t < minicachesleeptime)			t = minicachesleeptime;		if(t > 0)			sleep(t);		trace(TraceProc, "icachewritesect nextchunk");		chunk = nextchunk(ix, is, &iedirty, &addr, &nbuf); 		trace(TraceProc, "icachewritesect readpart 0x%llux+0x%ux",			addr, nbuf);		if(readpart(is->part, addr, buf, nbuf) < 0){			fprint(2, "%s: part %s addr 0x%llux: icachewritesect "				"readpart: %r\n", argv0, is->part->name, addr);			err  = -1;			continue;		}		trace(TraceProc, "icachewritesect updatebuf");		addstat(StatIsectReadBytes, nbuf);		addstat(StatIsectRead, 1); 		for(l=&chunk; (ie=*l)!=nil; l=&ie->nextdirty){again:			naddr = ie2diskaddr(ix, is, ie);			off = naddr - addr;			if(off+bsize > nbuf){				fprint(2, "%s: whoops! addr=0x%llux nbuf=%ud "					"addr+nbuf=0x%llux naddr=0x%llux\n",					argv0, addr, nbuf, addr+nbuf, naddr);				assert(off+bsize <= nbuf);			}			unpackibucket(&ib, buf+off, is->bucketmagic);			if(okibucket(&ib, is) < 0){				fprint(2, "%s: bad bucket XXX\n", argv0);				goto skipit;			}			trace(TraceProc, "icachewritesect add %V at 0x%llux",				ie->score, naddr);			h = bucklook(ie->score, ie->ia.type, ib.data, ib.n);			if(h & 1){				h ^= 1;				packientry(ie, &ib.data[h]);			}else if(ib.n < is->buckmax){				memmove(&ib.data[h + IEntrySize], &ib.data[h],					ib.n*IEntrySize - h);				ib.n++;				packientry(ie, &ib.data[h]);			}else{				fprint(2, "%s: bucket overflow XXX\n", argv0);skipit:				err = -1;				*l = ie->nextdirty;				ie = *l;				if(ie)					goto again;				else					break;			}			packibucket(&ib, buf+off, is->bucketmagic);		} 		diskaccess(1); 		trace(TraceProc, "icachewritesect writepart", addr, nbuf);		werr = 0;		if(writepart(is->part, addr, buf, nbuf) < 0 || flushpart(is->part) < 0)			werr = -1; 		for(i=0; i<nbuf; i+=bsize){			if((b = _getdblock(is->part, addr+i, ORDWR, 0)) != nil){				memmove(b->data, buf+i, bsize);				putdblock(b);			}		} 		if(werr < 0){			fprint(2, "%s: part %s addr 0x%llux: icachewritesect "				"writepart: %r\n", argv0, is->part->name, addr);			err = -1;			continue;		}				addstat(StatIsectWriteBytes, nbuf);		addstat(StatIsectWrite, 1);		icacheclean(chunk);	} 	trace(TraceProc, "icachewritesect done");	return err;} static voidicachewriteproc(void *v){	int ret;	uint bsize;	ISect *is;	Index *ix;	u8int *buf; 	ix = mainindex;	is = v;	threadsetname("icachewriteproc:%s", is->part->name); 	bsize = 1<<is->blocklog;	buf = emalloc(Bufsize+bsize);	buf = (u8int*)(((uintptr)buf+bsize-1)&~(uintptr)(bsize-1)); 	for(;;){		trace(TraceProc, "icachewriteproc recv");		recv(is->writechan, 0);		trace(TraceWork, "start");		ret = icachewritesect(ix, is, buf);		trace(TraceProc, "icachewriteproc send");		trace(TraceWork, "finish");		sendul(is->writedonechan, ret);	}} static voidicachewritecoord(void *v){	int i, err;	Index *ix;	AState as; 	USED(v); 	threadsetname("icachewritecoord"); 	ix = mainindex;	iwrite.as = icachestate(); 	for(;;){		trace(TraceProc, "icachewritecoord sleep");		waitforkick(&iwrite.round);		trace(TraceWork, "start");		as = icachestate();		if(as.arena==iwrite.as.arena && as.aa==iwrite.as.aa){			/* will not be able to do anything more than last flush - kick disk */			trace(TraceProc, "icachewritecoord kick dcache");			kickdcache();			trace(TraceProc, "icachewritecoord kicked dcache");			goto SkipWork;	/* won't do anything; don't bother rewriting bloom filter */		}		iwrite.as = as; 		trace(TraceProc, "icachewritecoord start flush");		if(iwrite.as.arena){			for(i=0; i<ix->nsects; i++)				send(ix->sects[i]->writechan, 0);			if(ix->bloom)				send(ix->bloom->writechan, 0);					err = 0;			for(i=0; i<ix->nsects; i++)				err |= recvul(ix->sects[i]->writedonechan);			if(ix->bloom)				err |= recvul(ix->bloom->writedonechan); 			trace(TraceProc, "icachewritecoord donewrite err=%d", err);			if(err == 0){				setatailstate(&iwrite.as);			}		}	SkipWork:		icacheclean(nil);	/* wake up anyone waiting */		trace(TraceWork, "finish");		addstat(StatIcacheFlush, 1);	}} voidflushicache(void){	trace(TraceProc, "flushicache enter");	kickround(&iwrite.round, 1);	trace(TraceProc, "flushicache exit");} voidkickicache(void){	kickround(&iwrite.round, 0);} voiddelaykickicache(void){	delaykickround(&iwrite.round);} static IEntry*iesort(IEntry *ie){	int cmp;	IEntry **l;	IEntry *ie1, *ie2, *sorted; 	if(ie == nil || ie->nextdirty == nil)		return ie; 	/* split the lists */	ie1 = ie;	ie2 = ie;	if(ie2)		ie2 = ie2->nextdirty;	if(ie2)		ie2 = ie2->nextdirty;	while(ie1 && ie2){		ie1 = ie1->nextdirty;		ie2 = ie2->nextdirty;		if(ie2)			ie2 = ie2->nextdirty;	}	if(ie1){		ie2 = ie1->nextdirty;		ie1->nextdirty = nil;	} 	/* sort the lists */	ie1 = iesort(ie);	ie2 = iesort(ie2); 	/* merge the lists */	sorted = nil;	l = &sorted;	cmp = 0;	while(ie1 || ie2){		if(ie1 && ie2)			cmp = scorecmp(ie1->score, ie2->score);		if(ie1==nil || (ie2 && cmp > 0)){			*l = ie2;			l = &ie2->nextdirty;			ie2 = ie2->nextdirty;		}else{			*l = ie1;			l = &ie1->nextdirty;			ie1 = ie1->nextdirty;		}	}	*l = nil;	return sorted;}