123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365 |
- #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;
- 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));
- 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);
- if(ci->type == VtCorruptType){
- if(0) 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);
- }
- }
- 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*)(((uintptr)ib->xbuf+size-1)&~(uintptr)(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);
- 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;
- }
- /*
- * 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(0) if(ib->bucks[b].total)
- fprint(2, "\tbucket %d: %lld 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;
- }
|