return of venti
This commit is contained in:
rsc
parent
88bb285e3d
commit
a0d146edd7
68 changed files with 14443 additions and 2 deletions
376
src/cmd/venti/srv/sortientry.c
Normal file
376
src/cmd/venti/srv/sortientry.c
Normal file
|
|
@ -0,0 +1,376 @@
|
|||
#include "stdinc.h"
|
||||
#include "dat.h"
|
||||
#include "fns.h"
|
||||
#include <bio.h>
|
||||
|
||||
typedef struct IEBuck IEBuck;
|
||||
typedef struct IEBucks IEBucks;
|
||||
|
||||
enum
|
||||
{
|
||||
ClumpChunks = 32*1024
|
||||
};
|
||||
|
||||
struct IEBuck
|
||||
{
|
||||
u32int head; /* head of chain of chunks on the disk */
|
||||
u32int used; /* usage of the last chunk */
|
||||
u64int total; /* total number of bytes in this bucket */
|
||||
u8int *buf; /* chunk of entries for this bucket */
|
||||
};
|
||||
|
||||
struct IEBucks
|
||||
{
|
||||
Part *part;
|
||||
u64int off; /* offset for writing data in the partition */
|
||||
u32int chunks; /* total chunks written to fd */
|
||||
u64int max; /* max bytes entered in any one bucket */
|
||||
int bits; /* number of bits in initial bucket sort */
|
||||
int nbucks; /* 1 << bits, the number of buckets */
|
||||
u32int size; /* bytes in each of the buckets chunks */
|
||||
u32int usable; /* amount usable for IEntry data */
|
||||
u8int *buf; /* buffer for all chunks */
|
||||
u8int *xbuf;
|
||||
IEBuck *bucks;
|
||||
};
|
||||
|
||||
#define U32GET(p) (((p)[0]<<24)|((p)[1]<<16)|((p)[2]<<8)|(p)[3])
|
||||
#define U32PUT(p,v) (p)[0]=(v)>>24;(p)[1]=(v)>>16;(p)[2]=(v)>>8;(p)[3]=(v)
|
||||
|
||||
static IEBucks *initiebucks(Part *part, int bits, u32int size);
|
||||
static int flushiebuck(IEBucks *ib, int b, int reset);
|
||||
static int flushiebucks(IEBucks *ib);
|
||||
static u32int sortiebuck(IEBucks *ib, int b);
|
||||
static u64int sortiebucks(IEBucks *ib);
|
||||
static int sprayientry(IEBucks *ib, IEntry *ie);
|
||||
static u32int readarenainfo(IEBucks *ib, Arena *arena, u64int a, Bloom *b);
|
||||
static u32int readiebuck(IEBucks *ib, int b);
|
||||
static void freeiebucks(IEBucks *ib);
|
||||
|
||||
/*
|
||||
* build a sorted file with all IEntries which should be in ix.
|
||||
* assumes the arenas' directories are up to date.
|
||||
* reads each, converts the entries to index entries,
|
||||
* and sorts them.
|
||||
*/
|
||||
u64int
|
||||
sortrawientries(Index *ix, Part *tmp, u64int *base, Bloom *bloom)
|
||||
{
|
||||
IEBucks *ib;
|
||||
u64int clumps, sorted;
|
||||
u32int n;
|
||||
int i, ok;
|
||||
|
||||
//ZZZ should allow configuration of bits, bucket size
|
||||
ib = initiebucks(tmp, 8, 64*1024);
|
||||
if(ib == nil){
|
||||
seterr(EOk, "can't create sorting buckets: %r");
|
||||
return TWID64;
|
||||
}
|
||||
ok = 0;
|
||||
clumps = 0;
|
||||
fprint(2, "constructing entry list\n");
|
||||
for(i = 0; i < ix->narenas; i++){
|
||||
n = readarenainfo(ib, ix->arenas[i], ix->amap[i].start, bloom);
|
||||
if(n == TWID32){
|
||||
ok = -1;
|
||||
break;
|
||||
}
|
||||
clumps += n;
|
||||
}
|
||||
fprint(2, "sorting %lld entries\n", clumps);
|
||||
if(ok == 0){
|
||||
sorted = sortiebucks(ib);
|
||||
*base = (u64int)ib->chunks * ib->size;
|
||||
if(sorted != clumps){
|
||||
fprint(2, "sorting messed up: clumps=%lld sorted=%lld\n", clumps, sorted);
|
||||
ok = -1;
|
||||
}
|
||||
}
|
||||
freeiebucks(ib);
|
||||
if(ok < 0)
|
||||
return TWID64;
|
||||
return clumps;
|
||||
}
|
||||
|
||||
#define CHECK(cis) if(((ulong*)cis)[-4] != 0xA110C09) xabort();
|
||||
|
||||
void
|
||||
xabort(void)
|
||||
{
|
||||
int *x;
|
||||
|
||||
x = 0;
|
||||
*x = 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* read in all of the arena's clump directory,
|
||||
* convert to IEntry format, and bucket sort based
|
||||
* on the first few bits.
|
||||
*/
|
||||
static u32int
|
||||
readarenainfo(IEBucks *ib, Arena *arena, u64int a, Bloom *b)
|
||||
{
|
||||
IEntry ie;
|
||||
ClumpInfo *ci, *cis;
|
||||
u32int clump;
|
||||
int i, n, ok, nskip;
|
||||
// static Biobuf bout;
|
||||
|
||||
//ZZZ remove fprint?
|
||||
//fprint(2, "ra %s %d %d\n", arena->name, arena->memstats.clumps, arena->diskstats.clumps);
|
||||
if(arena->memstats.clumps)
|
||||
fprint(2, "\tarena %s: %d entries\n", arena->name, arena->memstats.clumps);
|
||||
else
|
||||
fprint(2, "[%s] ", arena->name);
|
||||
|
||||
cis = MKN(ClumpInfo, ClumpChunks);
|
||||
ok = 0;
|
||||
nskip = 0;
|
||||
memset(&ie, 0, sizeof(IEntry));
|
||||
// Binit(&bout, 1, OWRITE);
|
||||
for(clump = 0; clump < arena->memstats.clumps; clump += n){
|
||||
n = ClumpChunks;
|
||||
if(n > arena->memstats.clumps - clump)
|
||||
n = arena->memstats.clumps - clump;
|
||||
if(readclumpinfos(arena, clump, cis, n) != n){
|
||||
seterr(EOk, "arena directory read failed: %r");
|
||||
ok = -1;
|
||||
break;
|
||||
}
|
||||
|
||||
for(i = 0; i < n; i++){
|
||||
ci = &cis[i];
|
||||
ie.ia.type = ci->type;
|
||||
ie.ia.size = ci->uncsize;
|
||||
ie.ia.addr = a;
|
||||
a += ci->size + ClumpSize;
|
||||
ie.ia.blocks = (ci->size + ClumpSize + (1 << ABlockLog) - 1) >> ABlockLog;
|
||||
scorecp(ie.score, ci->score);
|
||||
// Bprint(&bout, "%22lld %V %3d %5d\n",
|
||||
// ie.ia.addr, ie.score, ie.ia.type, ie.ia.size);
|
||||
if(ci->type == VtCorruptType){
|
||||
// print("! %V %22lld %3d %5d %3d\n",
|
||||
// ie.score, ie.ia.addr, ie.ia.type, ie.ia.size, ie.ia.blocks);
|
||||
nskip++;
|
||||
}else
|
||||
sprayientry(ib, &ie);
|
||||
markbloomfilter(b, ie.score);
|
||||
}
|
||||
}
|
||||
// Bterm(&bout);
|
||||
free(cis);
|
||||
if(ok < 0)
|
||||
return TWID32;
|
||||
return clump - nskip;
|
||||
}
|
||||
|
||||
/*
|
||||
* initialize the external bucket sorting data structures
|
||||
*/
|
||||
static IEBucks*
|
||||
initiebucks(Part *part, int bits, u32int size)
|
||||
{
|
||||
IEBucks *ib;
|
||||
int i;
|
||||
|
||||
ib = MKZ(IEBucks);
|
||||
if(ib == nil){
|
||||
seterr(EOk, "out of memory");
|
||||
return nil;
|
||||
}
|
||||
ib->bits = bits;
|
||||
ib->nbucks = 1 << bits;
|
||||
ib->size = size;
|
||||
ib->usable = (size - U32Size) / IEntrySize * IEntrySize;
|
||||
ib->bucks = MKNZ(IEBuck, ib->nbucks);
|
||||
if(ib->bucks == nil){
|
||||
seterr(EOk, "out of memory allocation sorting buckets");
|
||||
freeiebucks(ib);
|
||||
return nil;
|
||||
}
|
||||
ib->xbuf = MKN(u8int, size * ((1 << bits)+1));
|
||||
ib->buf = (u8int*)(((ulong)ib->xbuf+size-1)&~(ulong)(size-1));
|
||||
if(ib->buf == nil){
|
||||
seterr(EOk, "out of memory allocating sorting buckets' buffers");
|
||||
freeiebucks(ib);
|
||||
return nil;
|
||||
}
|
||||
for(i = 0; i < ib->nbucks; i++){
|
||||
ib->bucks[i].head = TWID32;
|
||||
ib->bucks[i].buf = &ib->buf[i * size];
|
||||
}
|
||||
ib->part = part;
|
||||
return ib;
|
||||
}
|
||||
|
||||
static void
|
||||
freeiebucks(IEBucks *ib)
|
||||
{
|
||||
if(ib == nil)
|
||||
return;
|
||||
free(ib->bucks);
|
||||
free(ib->buf);
|
||||
free(ib);
|
||||
}
|
||||
|
||||
/*
|
||||
* initial sort: put the entry into the correct bucket
|
||||
*/
|
||||
static int
|
||||
sprayientry(IEBucks *ib, IEntry *ie)
|
||||
{
|
||||
u32int n;
|
||||
int b;
|
||||
|
||||
b = hashbits(ie->score, ib->bits);
|
||||
n = ib->bucks[b].used;
|
||||
if(n + IEntrySize > ib->usable){
|
||||
/* should be flushed below, but if flush fails, this can happen */
|
||||
seterr(EOk, "out of space in bucket");
|
||||
return -1;
|
||||
}
|
||||
packientry(ie, &ib->bucks[b].buf[n]);
|
||||
n += IEntrySize;
|
||||
ib->bucks[b].used = n;
|
||||
if(n + IEntrySize <= ib->usable)
|
||||
return 0;
|
||||
return flushiebuck(ib, b, 1);
|
||||
}
|
||||
|
||||
/*
|
||||
* finish sorting:
|
||||
* for each bucket, read it in and sort it
|
||||
* write out the the final file
|
||||
*/
|
||||
static u64int
|
||||
sortiebucks(IEBucks *ib)
|
||||
{
|
||||
u64int tot;
|
||||
u32int n;
|
||||
int i;
|
||||
|
||||
if(flushiebucks(ib) < 0)
|
||||
return TWID64;
|
||||
for(i = 0; i < ib->nbucks; i++)
|
||||
ib->bucks[i].buf = nil;
|
||||
ib->off = (u64int)ib->chunks * ib->size;
|
||||
free(ib->xbuf);
|
||||
if(0){
|
||||
fprint(2, "ib->max = %lld\n", ib->max);
|
||||
fprint(2, "ib->chunks = %ud\n", ib->chunks);
|
||||
}
|
||||
ib->buf = MKN(u8int, ib->max + U32Size);
|
||||
if(ib->buf == nil){
|
||||
seterr(EOk, "out of memory allocating final sorting buffer; try more buckets");
|
||||
return TWID64;
|
||||
}
|
||||
tot = 0;
|
||||
for(i = 0; i < ib->nbucks; i++){
|
||||
n = sortiebuck(ib, i);
|
||||
if(n == TWID32)
|
||||
return TWID64;
|
||||
if(n != ib->bucks[i].total/IEntrySize)
|
||||
fprint(2, "bucket %d changed count %d => %d\n",
|
||||
i, (int)(ib->bucks[i].total/IEntrySize), n);
|
||||
tot += n;
|
||||
}
|
||||
return tot;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* sort from bucket b of ib into the output file to
|
||||
*/
|
||||
static u32int
|
||||
sortiebuck(IEBucks *ib, int b)
|
||||
{
|
||||
u32int n;
|
||||
|
||||
n = readiebuck(ib, b);
|
||||
if(n == TWID32)
|
||||
return TWID32;
|
||||
qsort(ib->buf, n, IEntrySize, ientrycmp);
|
||||
if(writepart(ib->part, ib->off, ib->buf, n * IEntrySize) < 0){
|
||||
seterr(EOk, "can't write sorted bucket: %r");
|
||||
return TWID32;
|
||||
}
|
||||
ib->off += n * IEntrySize;
|
||||
return n;
|
||||
}
|
||||
|
||||
/*
|
||||
* write out a single bucket
|
||||
*/
|
||||
static int
|
||||
flushiebuck(IEBucks *ib, int b, int reset)
|
||||
{
|
||||
u32int n;
|
||||
|
||||
if(ib->bucks[b].used == 0)
|
||||
return 0;
|
||||
n = ib->bucks[b].used;
|
||||
U32PUT(&ib->bucks[b].buf[n], ib->bucks[b].head);
|
||||
n += U32Size;
|
||||
USED(n);
|
||||
if(writepart(ib->part, (u64int)ib->chunks * ib->size, ib->bucks[b].buf, ib->size) < 0){
|
||||
seterr(EOk, "can't write sorting bucket to file: %r");
|
||||
xabort();
|
||||
return -1;
|
||||
}
|
||||
ib->bucks[b].head = ib->chunks++;
|
||||
ib->bucks[b].total += ib->bucks[b].used;
|
||||
if(reset)
|
||||
ib->bucks[b].used = 0;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* write out all of the buckets, and compute
|
||||
* the maximum size of any bucket
|
||||
*/
|
||||
static int
|
||||
flushiebucks(IEBucks *ib)
|
||||
{
|
||||
int i;
|
||||
|
||||
for(i = 0; i < ib->nbucks; i++){
|
||||
if(flushiebuck(ib, i, 0) < 0)
|
||||
return -1;
|
||||
if(ib->bucks[i].total > ib->max)
|
||||
ib->max = ib->bucks[i].total;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* read in the chained buffers for bucket b,
|
||||
* and return it's total number of IEntries
|
||||
*/
|
||||
static u32int
|
||||
readiebuck(IEBucks *ib, int b)
|
||||
{
|
||||
u32int head, m, n;
|
||||
|
||||
head = ib->bucks[b].head;
|
||||
n = 0;
|
||||
m = ib->bucks[b].used;
|
||||
if(m == 0)
|
||||
m = ib->usable;
|
||||
// if(ib->bucks[b].total)
|
||||
// fprint(2, "\tbucket %d: %d entries\n", b, ib->bucks[b].total/IEntrySize);
|
||||
while(head != TWID32){
|
||||
if(readpart(ib->part, (u64int)head * ib->size, &ib->buf[n], m + U32Size) < 0){
|
||||
seterr(EOk, "can't read index sort bucket: %r");
|
||||
return TWID32;
|
||||
}
|
||||
n += m;
|
||||
head = U32GET(&ib->buf[n]);
|
||||
m = ib->usable;
|
||||
}
|
||||
if(n != ib->bucks[b].total)
|
||||
fprint(2, "\tbucket %d: expected %d entries, got %d\n",
|
||||
b, (int)ib->bucks[b].total/IEntrySize, n/IEntrySize);
|
||||
return n / IEntrySize;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue