1 #include "stdinc.h"
2 #include "dat.h"
3 #include "fns.h"
4 #include <bio.h>
5 
6 typedef struct IEBuck	IEBuck;
7 typedef struct IEBucks	IEBucks;
8 
9 enum
10 {
11 	ClumpChunks	= 32*1024
12 };
13 
14 struct IEBuck
15 {
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 */
20 };
21 
22 struct IEBucks
23 {
24 	Part	*part;
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 */
33 	u8int	*xbuf;
34 	IEBuck	*bucks;
35 };
36 
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)
39 
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);
49 
50 /*
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,
54  * and sorts them.
55  */
56 u64int
57 sortrawientries(Index *ix, Part *tmp, u64int *base, Bloom *bloom)
58 {
59 	IEBucks *ib;
60 	u64int clumps, sorted;
61 	u32int n;
62 	int i, ok;
63 
64 //ZZZ should allow configuration of bits, bucket size
65 	ib = initiebucks(tmp, 8, 64*1024);
66 	if(ib == nil){
67 		seterr(EOk, "can't create sorting buckets: %r");
68 		return TWID64;
69 	}
70 	ok = 0;
71 	clumps = 0;
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);
75 		if(n == TWID32){
76 			ok = -1;
77 			break;
78 		}
79 		clumps += n;
80 	}
81 	fprint(2, "sorting %lld entries\n", clumps);
82 	if(ok == 0){
83 		sorted = sortiebucks(ib);
84 		*base = (u64int)ib->chunks * ib->size;
85 		if(sorted != clumps){
86 			fprint(2, "sorting messed up: clumps=%lld sorted=%lld\n", clumps, sorted);
87 			ok = -1;
88 		}
89 	}
90 	freeiebucks(ib);
91 	if(ok < 0)
92 		return TWID64;
93 	return clumps;
94 }
95 
96 #define CHECK(cis)	if(((ulong*)cis)[-4] != 0xA110C09) xabort();
97 
98 void
99 xabort(void)
100 {
101 	int *x;
102 
103 	x = 0;
104 	*x = 0;
105 }
106 
107 /*
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.
111  */
112 static u32int
113 readarenainfo(IEBucks *ib, Arena *arena, u64int a, Bloom *b)
114 {
115 	IEntry ie;
116 	ClumpInfo *ci, *cis;
117 	u32int clump;
118 	int i, n, ok, nskip;
119 //	static Biobuf bout;
120 
121 //ZZZ remove fprint?
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);
125 	else
126 		fprint(2, "[%s] ", arena->name);
127 
128 	cis = MKN(ClumpInfo, ClumpChunks);
129 	ok = 0;
130 	nskip = 0;
131 	memset(&ie, 0, sizeof(IEntry));
132 //	Binit(&bout, 1, OWRITE);
133 	for(clump = 0; clump < arena->memstats.clumps; clump += n){
134 		n = ClumpChunks;
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");
139 			ok = -1;
140 			break;
141 		}
142 
143 		for(i = 0; i < n; i++){
144 			ci = &cis[i];
145 			ie.ia.type = ci->type;
146 			ie.ia.size = ci->uncsize;
147 			ie.ia.addr = a;
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);
156 				nskip++;
157 			}else
158 				sprayientry(ib, &ie);
159 			markbloomfilter(b, ie.score);
160 		}
161 	}
162 //	Bterm(&bout);
163 	free(cis);
164 	if(ok < 0)
165 		return TWID32;
166 	return clump - nskip;
167 }
168 
169 /*
170  * initialize the external bucket sorting data structures
171  */
172 static IEBucks*
173 initiebucks(Part *part, int bits, u32int size)
174 {
175 	IEBucks *ib;
176 	int i;
177 
178 	ib = MKZ(IEBucks);
179 	if(ib == nil){
180 		seterr(EOk, "out of memory");
181 		return nil;
182 	}
183 	ib->bits = bits;
184 	ib->nbucks = 1 << bits;
185 	ib->size = size;
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");
190 		freeiebucks(ib);
191 		return nil;
192 	}
193 	ib->xbuf = MKN(u8int, size * ((1 << bits)+1));
194 	ib->buf = (u8int*)(((uintptr)ib->xbuf+size-1)&~(uintptr)(size-1));
195 	if(ib->buf == nil){
196 		seterr(EOk, "out of memory allocating sorting buckets' buffers");
197 		freeiebucks(ib);
198 		return nil;
199 	}
200 	for(i = 0; i < ib->nbucks; i++){
201 		ib->bucks[i].head = TWID32;
202 		ib->bucks[i].buf = &ib->buf[i * size];
203 	}
204 	ib->part = part;
205 	return ib;
206 }
207 
208 static void
209 freeiebucks(IEBucks *ib)
210 {
211 	if(ib == nil)
212 		return;
213 	free(ib->bucks);
214 	free(ib->buf);
215 	free(ib);
216 }
217 
218 /*
219  * initial sort: put the entry into the correct bucket
220  */
221 static int
222 sprayientry(IEBucks *ib, IEntry *ie)
223 {
224 	u32int n;
225 	int b;
226 
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");
232 		return -1;
233 	}
234 	packientry(ie, &ib->bucks[b].buf[n]);
235 	n += IEntrySize;
236 	ib->bucks[b].used = n;
237 	if(n + IEntrySize <= ib->usable)
238 		return 0;
239 	return flushiebuck(ib, b, 1);
240 }
241 
242 /*
243  * finish sorting:
244  * for each bucket, read it in and sort it
245  * write out the the final file
246  */
247 static u64int
248 sortiebucks(IEBucks *ib)
249 {
250 	u64int tot;
251 	u32int n;
252 	int i;
253 
254 	if(flushiebucks(ib) < 0)
255 		return TWID64;
256 	for(i = 0; i < ib->nbucks; i++)
257 		ib->bucks[i].buf = nil;
258 	ib->off = (u64int)ib->chunks * ib->size;
259 	free(ib->xbuf);
260 if(0){
261 	fprint(2, "ib->max = %lld\n", ib->max);
262 	fprint(2, "ib->chunks = %ud\n", ib->chunks);
263 }
264 	ib->buf = MKN(u8int, ib->max + U32Size);
265 	if(ib->buf == nil){
266 		seterr(EOk, "out of memory allocating final sorting buffer; try more buckets");
267 		return TWID64;
268 	}
269 	tot = 0;
270 	for(i = 0; i < ib->nbucks; i++){
271 		n = sortiebuck(ib, i);
272 		if(n == TWID32)
273 			return TWID64;
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);
277 		tot += n;
278 	}
279 	return tot;
280 	return 0;
281 }
282 
283 /*
284  * sort from bucket b of ib into the output file to
285  */
286 static u32int
287 sortiebuck(IEBucks *ib, int b)
288 {
289 	u32int n;
290 
291 	n = readiebuck(ib, b);
292 	if(n == TWID32)
293 		return TWID32;
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");
297 		return TWID32;
298 	}
299 	ib->off += n * IEntrySize;
300 	return n;
301 }
302 
303 /*
304  * write out a single bucket
305  */
306 static int
307 flushiebuck(IEBucks *ib, int b, int reset)
308 {
309 	u32int n;
310 
311 	if(ib->bucks[b].used == 0)
312 		return 0;
313 	n = ib->bucks[b].used;
314 	U32PUT(&ib->bucks[b].buf[n], ib->bucks[b].head);
315 	n += U32Size;
316 	USED(n);
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");
319 xabort();
320 		return -1;
321 	}
322 	ib->bucks[b].head = ib->chunks++;
323 	ib->bucks[b].total += ib->bucks[b].used;
324 	if(reset)
325 		ib->bucks[b].used = 0;
326 	return 0;
327 }
328 
329 /*
330  * write out all of the buckets, and compute
331  * the maximum size of any bucket
332  */
333 static int
334 flushiebucks(IEBucks *ib)
335 {
336 	int i;
337 
338 	for(i = 0; i < ib->nbucks; i++){
339 		if(flushiebuck(ib, i, 0) < 0)
340 			return -1;
341 		if(ib->bucks[i].total > ib->max)
342 			ib->max = ib->bucks[i].total;
343 	}
344 	return 0;
345 }
346 
347 /*
348  * read in the chained buffers for bucket b,
349  * and return it's total number of IEntries
350  */
351 static u32int
352 readiebuck(IEBucks *ib, int b)
353 {
354 	u32int head, m, n;
355 
356 	head = ib->bucks[b].head;
357 	n = 0;
358 	m = ib->bucks[b].used;
359 	if(m == 0)
360 		m = ib->usable;
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");
366 			return TWID32;
367 		}
368 		n += m;
369 		head = U32GET(&ib->buf[n]);
370 		m = ib->usable;
371 	}
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;
376 }