1 #include "stdinc.h"
2 #include "dat.h"
3 #include "fns.h"
4 
5 typedef struct IEBuck	IEBuck;
6 typedef struct IEBucks	IEBucks;
7 
8 enum
9 {
10 	ClumpChunks	= 32*1024
11 };
12 
13 struct IEBuck
14 {
15 	u32int	head;		/* head of chain of chunks on the disk */
16 	u32int	used;		/* usage of the last chunk */
17 	u64int	total;		/* total number of bytes in this bucket */
18 	u8int	*buf;		/* chunk of entries for this bucket */
19 };
20 
21 struct IEBucks
22 {
23 	Part	*part;
24 	u64int	off;		/* offset for writing data in the partition */
25 	u32int	chunks;		/* total chunks written to fd */
26 	u64int	max;		/* max bytes entered in any one bucket */
27 	int	bits;		/* number of bits in initial bucket sort */
28 	int	nbucks;		/* 1 << bits, the number of buckets */
29 	u32int	size;		/* bytes in each of the buckets chunks */
30 	u32int	usable;		/* amount usable for IEntry data */
31 	u8int	*buf;		/* buffer for all chunks */
32 	IEBuck	*bucks;
33 };
34 
35 #define	U32GET(p)	(((p)[0]<<24)|((p)[1]<<16)|((p)[2]<<8)|(p)[3])
36 #define	U32PUT(p,v)	(p)[0]=(v)>>24;(p)[1]=(v)>>16;(p)[2]=(v)>>8;(p)[3]=(v)
37 
38 static IEBucks	*initiebucks(Part *part, int bits, u32int size);
39 static int	flushiebuck(IEBucks *ib, int b, int reset);
40 static int	flushiebucks(IEBucks *ib);
41 static u32int	sortiebuck(IEBucks *ib, int b);
42 static u64int	sortiebucks(IEBucks *ib);
43 static int	sprayientry(IEBucks *ib, IEntry *ie);
44 static u32int	readarenainfo(IEBucks *ib, Arena *arena, u64int a);
45 static u32int	readiebuck(IEBucks *ib, int b);
46 static void	freeiebucks(IEBucks *ib);
47 
48 /*
49  * build a sorted file with all Ientries which should be in ix.
50  * assumes the arenas' directories are up to date.
51  * reads each, converts the entries to index entries,
52  * and sorts them.
53  */
54 u64int
55 sortrawientries(Index *ix, Part *tmp, u64int *base)
56 {
57 	IEBucks *ib;
58 	u64int clumps, sorted;
59 	u32int n;
60 	int i, ok;
61 
62 //ZZZ should allow configuration of bits, bucket size
63 	ib = initiebucks(tmp, 8, 64*1024);
64 	if(ib == nil){
65 		seterr(EOk, "can't create sorting buckets: %r");
66 		return TWID64;
67 	}
68 	ok = 0;
69 	clumps = 0;
70 	for(i = 0; i < ix->narenas; i++){
71 		n = readarenainfo(ib, ix->arenas[i], ix->amap[i].start);
72 		if(n == TWID32){
73 			ok = -1;
74 			break;
75 		}
76 		clumps += n;
77 	}
78 fprint(2, "got %lld clumps - starting sort\n", clumps);
79 	if(ok){
80 		sorted = sortiebucks(ib);
81 		*base = (u64int)ib->chunks * ib->size;
82 		if(sorted != clumps){
83 			fprint(2, "sorting messed up: clumps=%lld sorted=%lld\n", clumps, sorted);
84 			ok = -1;
85 		}
86 	}
87 	freeiebucks(ib);
88 	if(ok < 0)
89 		return TWID64;
90 	return clumps;
91 }
92 
93 /*
94  * read in all of the arena's clump directory,
95  * convert to IEntry format, and bucket sort based
96  * on the first few bits.
97  */
98 static u32int
99 readarenainfo(IEBucks *ib, Arena *arena, u64int a)
100 {
101 	IEntry ie;
102 	ClumpInfo *ci, *cis;
103 	u32int clump;
104 	int i, n, ok;
105 
106 //ZZZ remove fprint?
107 	if(arena->clumps)
108 		fprint(2, "reading directory for arena=%s with %d entries\n", arena->name, arena->clumps);
109 
110 	cis = MKN(ClumpInfo, ClumpChunks);
111 	ok = 0;
112 	memset(&ie, 0, sizeof(IEntry));
113 	for(clump = 0; clump < arena->clumps; clump += n){
114 		n = ClumpChunks;
115 		if(n > arena->clumps - clump)
116 			n = arena->clumps - clump;
117 		if(readclumpinfos(arena, clump, cis, n) != n){
118 			seterr(EOk, "arena directory read failed: %r");
119 			ok = -1;
120 			break;
121 		}
122 
123 		for(i = 0; i < n; i++){
124 			ci = &cis[i];
125 			ie.ia.type = ci->type;
126 			ie.ia.size = ci->uncsize;
127 			ie.ia.addr = a;
128 			a += ci->size + ClumpSize;
129 			ie.ia.blocks = (ci->size + ClumpSize + (1 << ABlockLog) - 1) >> ABlockLog;
130 			scorecp(ie.score, ci->score);
131 			sprayientry(ib, &ie);
132 		}
133 	}
134 	free(cis);
135 	if(ok < 0)
136 		return TWID32;
137 	return clump;
138 }
139 
140 /*
141  * initialize the external bucket sorting data structures
142  */
143 static IEBucks*
144 initiebucks(Part *part, int bits, u32int size)
145 {
146 	IEBucks *ib;
147 	int i;
148 
149 	ib = MKZ(IEBucks);
150 	if(ib == nil){
151 		seterr(EOk, "out of memory");
152 		return nil;
153 	}
154 	ib->bits = bits;
155 	ib->nbucks = 1 << bits;
156 	ib->size = size;
157 	ib->usable = (size - U32Size) / IEntrySize * IEntrySize;
158 	ib->bucks = MKNZ(IEBuck, ib->nbucks);
159 	if(ib->bucks == nil){
160 		seterr(EOk, "out of memory allocation sorting buckets");
161 		freeiebucks(ib);
162 		return nil;
163 	}
164 	ib->buf = MKN(u8int, size * (1 << bits));
165 	if(ib->buf == nil){
166 		seterr(EOk, "out of memory allocating sorting buckets' buffers");
167 		freeiebucks(ib);
168 		return nil;
169 	}
170 	for(i = 0; i < ib->nbucks; i++){
171 		ib->bucks[i].head = TWID32;
172 		ib->bucks[i].buf = &ib->buf[i * size];
173 	}
174 	ib->part = part;
175 	return ib;
176 }
177 
178 static void
179 freeiebucks(IEBucks *ib)
180 {
181 	if(ib == nil)
182 		return;
183 	free(ib->bucks);
184 	free(ib->buf);
185 	free(ib);
186 }
187 
188 /*
189  * initial sort: put the entry into the correct bucket
190  */
191 static int
192 sprayientry(IEBucks *ib, IEntry *ie)
193 {
194 	u32int n;
195 	int b;
196 
197 	b = hashbits(ie->score, ib->bits);
198 	n = ib->bucks[b].used;
199 	packientry(ie, &ib->bucks[b].buf[n]);
200 	n += IEntrySize;
201 	ib->bucks[b].used = n;
202 	if(n + IEntrySize <= ib->usable)
203 		return 0;
204 	return flushiebuck(ib, b, 1);
205 }
206 
207 /*
208  * finish sorting:
209  * for each bucket, read it in and sort it
210  * write out the the final file
211  */
212 static u64int
213 sortiebucks(IEBucks *ib)
214 {
215 	u64int tot;
216 	u32int n;
217 	int i;
218 
219 	if(flushiebucks(ib) < 0)
220 		return TWID64;
221 	for(i = 0; i < ib->nbucks; i++)
222 		ib->bucks[i].buf = nil;
223 	ib->off = (u64int)ib->chunks * ib->size;
224 	free(ib->buf);
225 fprint(2, "ib->max = %lld\n", ib->max);
226 fprint(2, "ib->chunks = %ud\n", ib->chunks);
227 	ib->buf = MKN(u8int, ib->max + U32Size);
228 	if(ib->buf == nil){
229 		seterr(EOk, "out of memory allocating final sorting buffer; try more buckets");
230 		return TWID64;
231 	}
232 	tot = 0;
233 	for(i = 0; i < ib->nbucks; i++){
234 		n = sortiebuck(ib, i);
235 		if(n == TWID32)
236 			return TWID64;
237 		tot += n;
238 	}
239 	return tot;
240 	return 0;
241 }
242 
243 /*
244  * sort from bucket b of ib into the output file to
245  */
246 static u32int
247 sortiebuck(IEBucks *ib, int b)
248 {
249 	u32int n;
250 
251 	n = readiebuck(ib, b);
252 	if(n == TWID32)
253 		return TWID32;
254 	qsort(ib->buf, n, IEntrySize, ientrycmp);
255 	if(writepart(ib->part, ib->off, ib->buf, n * IEntrySize) < 0){
256 		seterr(EOk, "can't write sorted bucket: %r");
257 		return TWID32;
258 	}
259 	ib->off += n * IEntrySize;
260 	return n;
261 }
262 
263 /*
264  * write out a single bucket
265  */
266 static int
267 flushiebuck(IEBucks *ib, int b, int reset)
268 {
269 	u32int n;
270 
271 	if(ib->bucks[b].used == 0)
272 		return 0;
273 	n = ib->bucks[b].used;
274 	U32PUT(&ib->bucks[b].buf[n], ib->bucks[b].head);
275 	n += U32Size;
276 	if(writepart(ib->part, (u64int)ib->chunks * ib->size, ib->bucks[b].buf, n) < 0){
277 		seterr(EOk, "can't write sorting bucket to file: %r");
278 		return -1;
279 	}
280 	ib->bucks[b].head = ib->chunks++;
281 	ib->bucks[b].total += ib->bucks[b].used;
282 	if(reset)
283 		ib->bucks[b].used = 0;
284 	return 0;
285 }
286 
287 /*
288  * write out all of the buckets, and compute
289  * the maximum size of any bucket
290  */
291 static int
292 flushiebucks(IEBucks *ib)
293 {
294 	int i;
295 
296 	for(i = 0; i < ib->nbucks; i++){
297 		if(flushiebuck(ib, i, 0) < 0)
298 			return -1;
299 		if(ib->bucks[i].total > ib->max)
300 			ib->max = ib->bucks[i].total;
301 	}
302 	return 0;
303 }
304 
305 /*
306  * read in the chained buffers for bucket b,
307  * and return it's total number of IEntries
308  */
309 static u32int
310 readiebuck(IEBucks *ib, int b)
311 {
312 	u32int head, n, m;
313 
314 	head = ib->bucks[b].head;
315 	n = 0;
316 	m = ib->bucks[b].used;
317 	if(m == 0)
318 		m = ib->usable;
319 fprint(2, "%d total = %lld\n", b, ib->bucks[b].total);
320 	while(head != TWID32){
321 		if(readpart(ib->part, (u64int)head * ib->size, &ib->buf[n], m + U32Size) < 0){
322 fprint(2, "n = %ud\n", n);
323 			seterr(EOk, "can't read index sort bucket: %r");
324 			return TWID32;
325 		}
326 		n += m;
327 		head = U32GET(&ib->buf[n]);
328 		m = ib->usable;
329 	}
330 fprint(2, "n = %ud\n", n);
331 	return n / IEntrySize;
332 }