1 /*
2  * Write the dirty icache entries to disk.  Random seeks are
3  * so expensive that it makes sense to wait until we have
4  * a lot and then just make a sequential pass over the disk.
5  */
6 #include "stdinc.h"
7 #include "dat.h"
8 #include "fns.h"
9 
10 static void icachewriteproc(void*);
11 static void icachewritecoord(void*);
12 static IEntry *iesort(IEntry*);
13 
14 int icachesleeptime = 1000;	/* milliseconds */
15 int minicachesleeptime = 0;
16 
17 enum
18 {
19 	Bufsize = 8*1024*1024
20 };
21 
22 typedef struct IWrite IWrite;
23 struct IWrite
24 {
25 	Round round;
26 	AState as;
27 };
28 
29 static IWrite iwrite;
30 
31 void
32 initicachewrite(void)
33 {
34 	int i;
35 	Index *ix;
36 
37 	initround(&iwrite.round, "icache", 120*60*1000);
38 	ix = mainindex;
39 	for(i=0; i<ix->nsects; i++){
40 		ix->sects[i]->writechan = chancreate(sizeof(ulong), 1);
41 		ix->sects[i]->writedonechan = chancreate(sizeof(ulong), 1);
42 		vtproc(icachewriteproc, ix->sects[i]);
43 	}
44 	vtproc(icachewritecoord, nil);
45 	vtproc(delaykickroundproc, &iwrite.round);
46 }
47 
48 static u64int
49 ie2diskaddr(Index *ix, ISect *is, IEntry *ie)
50 {
51 	u64int bucket, addr;
52 
53 	bucket = hashbits(ie->score, 32)/ix->div;
54 	addr = is->blockbase + ((bucket - is->start) << is->blocklog);
55 	return addr;
56 }
57 
58 static IEntry*
59 nextchunk(Index *ix, ISect *is, IEntry **pie, u64int *paddr, uint *pnbuf)
60 {
61 	u64int addr, naddr;
62 	uint nbuf;
63 	int bsize;
64 	IEntry *iefirst, *ie, **l;
65 
66 	bsize = 1<<is->blocklog;
67 	iefirst = *pie;
68 	addr = ie2diskaddr(ix, is, iefirst);
69 	nbuf = 0;
70 	for(l = &iefirst->nextdirty; (ie = *l) != nil; l = &(*l)->nextdirty){
71 		naddr = ie2diskaddr(ix, is, ie);
72 		if(naddr - addr >= Bufsize)
73 			break;
74 		nbuf = naddr - addr;
75 	}
76 	nbuf += bsize;
77 
78 	*l = nil;
79 	*pie = ie;
80 	*paddr = addr;
81 	*pnbuf = nbuf;
82 	return iefirst;
83 }
84 
85 static int
86 icachewritesect(Index *ix, ISect *is, u8int *buf)
87 {
88 	int err, i, werr, h, bsize, t;
89 	u32int lo, hi;
90 	u64int addr, naddr;
91 	uint nbuf, off;
92 	DBlock *b;
93 	IBucket ib;
94 	IEntry *ie, *iedirty, **l, *chunk;
95 
96 	lo = is->start * ix->div;
97 	if(TWID32/ix->div < is->stop)
98 		hi = TWID32;
99 	else
100 		hi = is->stop * ix->div - 1;
101 
102 	trace(TraceProc, "icachewritesect enter %ud %ud %llud",
103 		lo, hi, iwrite.as.aa);
104 
105 	iedirty = icachedirty(lo, hi, iwrite.as.aa);
106 	iedirty = iesort(iedirty);
107 	bsize = 1 << is->blocklog;
108 	err = 0;
109 
110 	while(iedirty){
111 		disksched();
112 		while((t = icachesleeptime) == SleepForever){
113 			sleep(1000);
114 			disksched();
115 		}
116 		if(t < minicachesleeptime)
117 			t = minicachesleeptime;
118 		if(t > 0)
119 			sleep(t);
120 		trace(TraceProc, "icachewritesect nextchunk");
121 		chunk = nextchunk(ix, is, &iedirty, &addr, &nbuf);
122 
123 		trace(TraceProc, "icachewritesect readpart 0x%llux+0x%ux",
124 			addr, nbuf);
125 		if(readpart(is->part, addr, buf, nbuf) < 0){
126 			fprint(2, "%s: part %s addr 0x%llux: icachewritesect "
127 				"readpart: %r\n", argv0, is->part->name, addr);
128 			err  = -1;
129 			continue;
130 		}
131 		trace(TraceProc, "icachewritesect updatebuf");
132 		addstat(StatIsectReadBytes, nbuf);
133 		addstat(StatIsectRead, 1);
134 
135 		for(l=&chunk; (ie=*l)!=nil; l=&ie->nextdirty){
136 again:
137 			naddr = ie2diskaddr(ix, is, ie);
138 			off = naddr - addr;
139 			if(off+bsize > nbuf){
140 				fprint(2, "%s: whoops! addr=0x%llux nbuf=%ud "
141 					"addr+nbuf=0x%llux naddr=0x%llux\n",
142 					argv0, addr, nbuf, addr+nbuf, naddr);
143 				assert(off+bsize <= nbuf);
144 			}
145 			unpackibucket(&ib, buf+off, is->bucketmagic);
146 			if(okibucket(&ib, is) < 0){
147 				fprint(2, "%s: bad bucket XXX\n", argv0);
148 				goto skipit;
149 			}
150 			trace(TraceProc, "icachewritesect add %V at 0x%llux",
151 				ie->score, naddr);
152 			h = bucklook(ie->score, ie->ia.type, ib.data, ib.n);
153 			if(h & 1){
154 				h ^= 1;
155 				packientry(ie, &ib.data[h]);
156 			}else if(ib.n < is->buckmax){
157 				memmove(&ib.data[h + IEntrySize], &ib.data[h],
158 					ib.n*IEntrySize - h);
159 				ib.n++;
160 				packientry(ie, &ib.data[h]);
161 			}else{
162 				fprint(2, "%s: bucket overflow XXX\n", argv0);
163 skipit:
164 				err = -1;
165 				*l = ie->nextdirty;
166 				ie = *l;
167 				if(ie)
168 					goto again;
169 				else
170 					break;
171 			}
172 			packibucket(&ib, buf+off, is->bucketmagic);
173 		}
174 
175 		diskaccess(1);
176 
177 		trace(TraceProc, "icachewritesect writepart", addr, nbuf);
178 		werr = 0;
179 		if(writepart(is->part, addr, buf, nbuf) < 0 || flushpart(is->part) < 0)
180 			werr = -1;
181 
182 		for(i=0; i<nbuf; i+=bsize){
183 			if((b = _getdblock(is->part, addr+i, ORDWR, 0)) != nil){
184 				memmove(b->data, buf+i, bsize);
185 				putdblock(b);
186 			}
187 		}
188 
189 		if(werr < 0){
190 			fprint(2, "%s: part %s addr 0x%llux: icachewritesect "
191 				"writepart: %r\n", argv0, is->part->name, addr);
192 			err = -1;
193 			continue;
194 		}
195 
196 		addstat(StatIsectWriteBytes, nbuf);
197 		addstat(StatIsectWrite, 1);
198 		icacheclean(chunk);
199 	}
200 
201 	trace(TraceProc, "icachewritesect done");
202 	return err;
203 }
204 
205 static void
206 icachewriteproc(void *v)
207 {
208 	int ret;
209 	uint bsize;
210 	ISect *is;
211 	Index *ix;
212 	u8int *buf;
213 
214 	ix = mainindex;
215 	is = v;
216 	threadsetname("icachewriteproc:%s", is->part->name);
217 
218 	bsize = 1<<is->blocklog;
219 	buf = emalloc(Bufsize+bsize);
220 	buf = (u8int*)(((uintptr)buf+bsize-1)&~(uintptr)(bsize-1));
221 
222 	for(;;){
223 		trace(TraceProc, "icachewriteproc recv");
224 		recv(is->writechan, 0);
225 		trace(TraceWork, "start");
226 		ret = icachewritesect(ix, is, buf);
227 		trace(TraceProc, "icachewriteproc send");
228 		trace(TraceWork, "finish");
229 		sendul(is->writedonechan, ret);
230 	}
231 }
232 
233 static void
234 icachewritecoord(void *v)
235 {
236 	int i, err;
237 	Index *ix;
238 	AState as;
239 
240 	USED(v);
241 
242 	threadsetname("icachewritecoord");
243 
244 	ix = mainindex;
245 	iwrite.as = icachestate();
246 
247 	for(;;){
248 		trace(TraceProc, "icachewritecoord sleep");
249 		waitforkick(&iwrite.round);
250 		trace(TraceWork, "start");
251 		as = icachestate();
252 		if(as.arena==iwrite.as.arena && as.aa==iwrite.as.aa){
253 			/* will not be able to do anything more than last flush - kick disk */
254 			trace(TraceProc, "icachewritecoord kick dcache");
255 			kickdcache();
256 			trace(TraceProc, "icachewritecoord kicked dcache");
257 			goto SkipWork;	/* won't do anything; don't bother rewriting bloom filter */
258 		}
259 		iwrite.as = as;
260 
261 		trace(TraceProc, "icachewritecoord start flush");
262 		if(iwrite.as.arena){
263 			for(i=0; i<ix->nsects; i++)
264 				send(ix->sects[i]->writechan, 0);
265 			if(ix->bloom)
266 				send(ix->bloom->writechan, 0);
267 
268 			err = 0;
269 			for(i=0; i<ix->nsects; i++)
270 				err |= recvul(ix->sects[i]->writedonechan);
271 			if(ix->bloom)
272 				err |= recvul(ix->bloom->writedonechan);
273 
274 			trace(TraceProc, "icachewritecoord donewrite err=%d", err);
275 			if(err == 0){
276 				setatailstate(&iwrite.as);
277 			}
278 		}
279 	SkipWork:
280 		icacheclean(nil);	/* wake up anyone waiting */
281 		trace(TraceWork, "finish");
282 		addstat(StatIcacheFlush, 1);
283 	}
284 }
285 
286 void
287 flushicache(void)
288 {
289 	trace(TraceProc, "flushicache enter");
290 	kickround(&iwrite.round, 1);
291 	trace(TraceProc, "flushicache exit");
292 }
293 
294 void
295 kickicache(void)
296 {
297 	kickround(&iwrite.round, 0);
298 }
299 
300 void
301 delaykickicache(void)
302 {
303 	delaykickround(&iwrite.round);
304 }
305 
306 static IEntry*
307 iesort(IEntry *ie)
308 {
309 	int cmp;
310 	IEntry **l;
311 	IEntry *ie1, *ie2, *sorted;
312 
313 	if(ie == nil || ie->nextdirty == nil)
314 		return ie;
315 
316 	/* split the lists */
317 	ie1 = ie;
318 	ie2 = ie;
319 	if(ie2)
320 		ie2 = ie2->nextdirty;
321 	if(ie2)
322 		ie2 = ie2->nextdirty;
323 	while(ie1 && ie2){
324 		ie1 = ie1->nextdirty;
325 		ie2 = ie2->nextdirty;
326 		if(ie2)
327 			ie2 = ie2->nextdirty;
328 	}
329 	if(ie1){
330 		ie2 = ie1->nextdirty;
331 		ie1->nextdirty = nil;
332 	}
333 
334 	/* sort the lists */
335 	ie1 = iesort(ie);
336 	ie2 = iesort(ie2);
337 
338 	/* merge the lists */
339 	sorted = nil;
340 	l = &sorted;
341 	cmp = 0;
342 	while(ie1 || ie2){
343 		if(ie1 && ie2)
344 			cmp = scorecmp(ie1->score, ie2->score);
345 		if(ie1==nil || (ie2 && cmp > 0)){
346 			*l = ie2;
347 			l = &ie2->nextdirty;
348 			ie2 = ie2->nextdirty;
349 		}else{
350 			*l = ie1;
351 			l = &ie1->nextdirty;
352 			ie1 = ie1->nextdirty;
353 		}
354 	}
355 	*l = nil;
356 	return sorted;
357 }