6 typedef struct IEBuck IEBuck;
7 typedef struct IEBucks IEBucks;
16 u32int head; /* head of chain of chunks on the disk */
17 u32int used; /* usage of the last chunk */
18 u64int total; /* total number of bytes in this bucket */
19 u8int *buf; /* chunk of entries for this bucket */
25 u64int off; /* offset for writing data in the partition */
26 u32int chunks; /* total chunks written to fd */
27 u64int max; /* max bytes entered in any one bucket */
28 int bits; /* number of bits in initial bucket sort */
29 int nbucks; /* 1 << bits, the number of buckets */
30 u32int size; /* bytes in each of the buckets chunks */
31 u32int usable; /* amount usable for IEntry data */
32 u8int *buf; /* buffer for all chunks */
37 #define U32GET(p) (((p)[0]<<24)|((p)[1]<<16)|((p)[2]<<8)|(p)[3])
38 #define U32PUT(p,v) (p)[0]=(v)>>24;(p)[1]=(v)>>16;(p)[2]=(v)>>8;(p)[3]=(v)
40 static IEBucks *initiebucks(Part *part, int bits, u32int size);
41 static int flushiebuck(IEBucks *ib, int b, int reset);
42 static int flushiebucks(IEBucks *ib);
43 static u32int sortiebuck(IEBucks *ib, int b);
44 static u64int sortiebucks(IEBucks *ib);
45 static int sprayientry(IEBucks *ib, IEntry *ie);
46 static u32int readarenainfo(IEBucks *ib, Arena *arena, u64int a, Bloom *b);
47 static u32int readiebuck(IEBucks *ib, int b);
48 static void freeiebucks(IEBucks *ib);
51 * build a sorted file with all IEntries which should be in ix.
52 * assumes the arenas' directories are up to date.
53 * reads each, converts the entries to index entries,
57 sortrawientries(Index *ix, Part *tmp, u64int *base, Bloom *bloom)
60 u64int clumps, sorted;
64 //ZZZ should allow configuration of bits, bucket size
65 ib = initiebucks(tmp, 8, 64*1024);
67 seterr(EOk, "can't create sorting buckets: %r");
72 fprint(2, "constructing entry list\n");
73 for(i = 0; i < ix->narenas; i++){
74 n = readarenainfo(ib, ix->arenas[i], ix->amap[i].start, bloom);
81 fprint(2, "sorting %lld entries\n", clumps);
83 sorted = sortiebucks(ib);
84 *base = (u64int)ib->chunks * ib->size;
86 fprint(2, "sorting messed up: clumps=%lld sorted=%lld\n", clumps, sorted);
96 #define CHECK(cis) if(((ulong*)cis)[-4] != 0xA110C09) xabort();
108 * read in all of the arena's clump directory,
109 * convert to IEntry format, and bucket sort based
110 * on the first few bits.
113 readarenainfo(IEBucks *ib, Arena *arena, u64int a, Bloom *b)
119 // static Biobuf bout;
122 //fprint(2, "ra %s %d %d\n", arena->name, arena->memstats.clumps, arena->diskstats.clumps);
123 if(arena->memstats.clumps)
124 fprint(2, "\tarena %s: %d entries\n", arena->name, arena->memstats.clumps);
126 fprint(2, "[%s] ", arena->name);
128 cis = MKN(ClumpInfo, ClumpChunks);
131 memset(&ie, 0, sizeof(IEntry));
132 // Binit(&bout, 1, OWRITE);
133 for(clump = 0; clump < arena->memstats.clumps; clump += n){
135 if(n > arena->memstats.clumps - clump)
136 n = arena->memstats.clumps - clump;
137 if(readclumpinfos(arena, clump, cis, n) != n){
138 seterr(EOk, "arena directory read failed: %r");
143 for(i = 0; i < n; i++){
145 ie.ia.type = ci->type;
146 ie.ia.size = ci->uncsize;
148 a += ci->size + ClumpSize;
149 ie.ia.blocks = (ci->size + ClumpSize + (1 << ABlockLog) - 1) >> ABlockLog;
150 scorecp(ie.score, ci->score);
151 // Bprint(&bout, "%22lld %V %3d %5d\n",
152 // ie.ia.addr, ie.score, ie.ia.type, ie.ia.size);
153 if(ci->type == VtCorruptType){
154 // print("! %V %22lld %3d %5d %3d\n",
155 // ie.score, ie.ia.addr, ie.ia.type, ie.ia.size, ie.ia.blocks);
158 sprayientry(ib, &ie);
159 markbloomfilter(b, ie.score);
166 return clump - nskip;
170 * initialize the external bucket sorting data structures
173 initiebucks(Part *part, int bits, u32int size)
180 seterr(EOk, "out of memory");
184 ib->nbucks = 1 << bits;
186 ib->usable = (size - U32Size) / IEntrySize * IEntrySize;
187 ib->bucks = MKNZ(IEBuck, ib->nbucks);
188 if(ib->bucks == nil){
189 seterr(EOk, "out of memory allocation sorting buckets");
193 ib->xbuf = MKN(u8int, size * ((1 << bits)+1));
194 ib->buf = (u8int*)(((uintptr)ib->xbuf+size-1)&~(uintptr)(size-1));
196 seterr(EOk, "out of memory allocating sorting buckets' buffers");
200 for(i = 0; i < ib->nbucks; i++){
201 ib->bucks[i].head = TWID32;
202 ib->bucks[i].buf = &ib->buf[i * size];
209 freeiebucks(IEBucks *ib)
219 * initial sort: put the entry into the correct bucket
222 sprayientry(IEBucks *ib, IEntry *ie)
227 b = hashbits(ie->score, ib->bits);
228 n = ib->bucks[b].used;
229 if(n + IEntrySize > ib->usable){
230 /* should be flushed below, but if flush fails, this can happen */
231 seterr(EOk, "out of space in bucket");
234 packientry(ie, &ib->bucks[b].buf[n]);
236 ib->bucks[b].used = n;
237 if(n + IEntrySize <= ib->usable)
239 return flushiebuck(ib, b, 1);
244 * for each bucket, read it in and sort it
245 * write out the the final file
248 sortiebucks(IEBucks *ib)
254 if(flushiebucks(ib) < 0)
256 for(i = 0; i < ib->nbucks; i++)
257 ib->bucks[i].buf = nil;
258 ib->off = (u64int)ib->chunks * ib->size;
261 fprint(2, "ib->max = %lld\n", ib->max);
262 fprint(2, "ib->chunks = %ud\n", ib->chunks);
264 ib->buf = MKN(u8int, ib->max + U32Size);
266 seterr(EOk, "out of memory allocating final sorting buffer; try more buckets");
270 for(i = 0; i < ib->nbucks; i++){
271 n = sortiebuck(ib, i);
274 if(n != ib->bucks[i].total/IEntrySize)
275 fprint(2, "bucket %d changed count %d => %d\n",
276 i, (int)(ib->bucks[i].total/IEntrySize), n);
284 * sort from bucket b of ib into the output file to
287 sortiebuck(IEBucks *ib, int b)
291 n = readiebuck(ib, b);
294 qsort(ib->buf, n, IEntrySize, ientrycmp);
295 if(writepart(ib->part, ib->off, ib->buf, n*IEntrySize) < 0){
296 seterr(EOk, "can't write sorted bucket: %r");
299 ib->off += n * IEntrySize;
304 * write out a single bucket
307 flushiebuck(IEBucks *ib, int b, int reset)
311 if(ib->bucks[b].used == 0)
313 n = ib->bucks[b].used;
314 U32PUT(&ib->bucks[b].buf[n], ib->bucks[b].head);
317 if(writepart(ib->part, (u64int)ib->chunks * ib->size, ib->bucks[b].buf, ib->size) < 0){
318 seterr(EOk, "can't write sorting bucket to file: %r");
322 ib->bucks[b].head = ib->chunks++;
323 ib->bucks[b].total += ib->bucks[b].used;
325 ib->bucks[b].used = 0;
330 * write out all of the buckets, and compute
331 * the maximum size of any bucket
334 flushiebucks(IEBucks *ib)
338 for(i = 0; i < ib->nbucks; i++){
339 if(flushiebuck(ib, i, 0) < 0)
341 if(ib->bucks[i].total > ib->max)
342 ib->max = ib->bucks[i].total;
348 * read in the chained buffers for bucket b,
349 * and return it's total number of IEntries
352 readiebuck(IEBucks *ib, int b)
356 head = ib->bucks[b].head;
358 m = ib->bucks[b].used;
361 // if(ib->bucks[b].total)
362 // fprint(2, "\tbucket %d: %d entries\n", b, ib->bucks[b].total/IEntrySize);
363 while(head != TWID32){
364 if(readpart(ib->part, (u64int)head * ib->size, &ib->buf[n], m+U32Size) < 0){
365 seterr(EOk, "can't read index sort bucket: %r");
369 head = U32GET(&ib->buf[n]);
372 if(n != ib->bucks[b].total)
373 fprint(2, "\tbucket %d: expected %d entries, got %d\n",
374 b, (int)ib->bucks[b].total/IEntrySize, n/IEntrySize);
375 return n / IEntrySize;