1 #include "stdinc.h"
2 #include "dat.h"
3 #include "fns.h"
4 #include "error.h"
5
6 static void diskThread(void *a);
7
8 enum {
9 /*
10 * disable measurement since it gets alignment faults on BG
11 * and the guts used to be commented out.
12 */
13 Timing = 0, /* flag */
14 QueueSize = 100, /* maximum block to queue */
15 };
16
17 struct Disk {
18 QLock lk;
19 int ref;
20
21 int fd;
22 Header h;
23
24 Rendez flow;
25 Rendez starve;
26 Rendez flush;
27 Rendez die;
28
29 int nqueue;
30
31 Block *cur; /* block to do on current scan */
32 Block *next; /* blocks to do next scan */
33 };
34
35 /* keep in sync with Part* enum in dat.h */
36 static char *partname[] = {
37 [PartError] = "error",
38 [PartSuper] = "super",
39 [PartLabel] = "label",
40 [PartData] = "data",
41 [PartVenti] = "venti",
42 };
43
44 Disk *
diskAlloc(int fd)45 diskAlloc(int fd)
46 {
47 u8int buf[HeaderSize];
48 Header h;
49 Disk *disk;
50
51 if(pread(fd, buf, HeaderSize, HeaderOffset) < HeaderSize){
52 werrstr("short read: %r");
53 return nil;
54 }
55
56 if(!headerUnpack(&h, buf)){
57 werrstr("bad disk header");
58 return nil;
59 }
60 disk = vtmallocz(sizeof(Disk));
61 disk->starve.l = &disk->lk;
62 disk->flow.l = &disk->lk;
63 disk->flush.l = &disk->lk;
64 disk->fd = fd;
65 disk->h = h;
66
67 disk->ref = 2;
68 proccreate(diskThread, disk, STACK);
69
70 return disk;
71 }
72
73 void
diskFree(Disk * disk)74 diskFree(Disk *disk)
75 {
76 diskFlush(disk);
77
78 /* kill slave */
79 qlock(&disk->lk);
80 disk->die.l = &disk->lk;
81 rwakeup(&disk->starve);
82 while(disk->ref > 1)
83 rsleep(&disk->die);
84 qunlock(&disk->lk);
85 close(disk->fd);
86 vtfree(disk);
87 }
88
89 static u32int
partStart(Disk * disk,int part)90 partStart(Disk *disk, int part)
91 {
92 switch(part){
93 default:
94 assert(0);
95 case PartSuper:
96 return disk->h.super;
97 case PartLabel:
98 return disk->h.label;
99 case PartData:
100 return disk->h.data;
101 }
102 }
103
104
105 static u32int
partEnd(Disk * disk,int part)106 partEnd(Disk *disk, int part)
107 {
108 switch(part){
109 default:
110 assert(0);
111 case PartSuper:
112 return disk->h.super+1;
113 case PartLabel:
114 return disk->h.data;
115 case PartData:
116 return disk->h.end;
117 }
118 }
119
120 int
diskReadRaw(Disk * disk,int part,u32int addr,uchar * buf)121 diskReadRaw(Disk *disk, int part, u32int addr, uchar *buf)
122 {
123 ulong start, end;
124 u64int offset;
125 int n, nn;
126
127 start = partStart(disk, part);
128 end = partEnd(disk, part);
129
130 if(addr >= end-start){
131 werrstr(EBadAddr);
132 return 0;
133 }
134
135 offset = ((u64int)(addr + start))*disk->h.blockSize;
136 n = disk->h.blockSize;
137 while(n > 0){
138 nn = pread(disk->fd, buf, n, offset);
139 if(nn < 0){
140 werrstr("%r");
141 return 0;
142 }
143 if(nn == 0){
144 werrstr("eof reading disk");
145 return 0;
146 }
147 n -= nn;
148 offset += nn;
149 buf += nn;
150 }
151 return 1;
152 }
153
154 int
diskWriteRaw(Disk * disk,int part,u32int addr,uchar * buf)155 diskWriteRaw(Disk *disk, int part, u32int addr, uchar *buf)
156 {
157 ulong start, end;
158 u64int offset;
159 int n;
160
161 start = partStart(disk, part);
162 end = partEnd(disk, part);
163
164 if(addr >= end - start){
165 werrstr(EBadAddr);
166 return 0;
167 }
168
169 offset = ((u64int)(addr + start))*disk->h.blockSize;
170 n = pwrite(disk->fd, buf, disk->h.blockSize, offset);
171 if(n < 0){
172 werrstr("%r");
173 return 0;
174 }
175 if(n < disk->h.blockSize) {
176 werrstr("short write");
177 return 0;
178 }
179
180 return 1;
181 }
182
183 static void
diskQueue(Disk * disk,Block * b)184 diskQueue(Disk *disk, Block *b)
185 {
186 Block **bp, *bb;
187
188 qlock(&disk->lk);
189 while(disk->nqueue >= QueueSize)
190 rsleep(&disk->flow);
191 if(disk->cur == nil || b->addr > disk->cur->addr)
192 bp = &disk->cur;
193 else
194 bp = &disk->next;
195
196 for(bb=*bp; bb; bb=*bp){
197 if(b->addr < bb->addr)
198 break;
199 bp = &bb->ionext;
200 }
201 b->ionext = bb;
202 *bp = b;
203 if(disk->nqueue == 0)
204 rwakeup(&disk->starve);
205 disk->nqueue++;
206 qunlock(&disk->lk);
207 }
208
209
210 void
diskRead(Disk * disk,Block * b)211 diskRead(Disk *disk, Block *b)
212 {
213 assert(b->iostate == BioEmpty || b->iostate == BioLabel);
214 blockSetIOState(b, BioReading);
215 diskQueue(disk, b);
216 }
217
218 void
diskWrite(Disk * disk,Block * b)219 diskWrite(Disk *disk, Block *b)
220 {
221 assert(b->nlock == 1);
222 assert(b->iostate == BioDirty);
223 blockSetIOState(b, BioWriting);
224 diskQueue(disk, b);
225 }
226
227 void
diskWriteAndWait(Disk * disk,Block * b)228 diskWriteAndWait(Disk *disk, Block *b)
229 {
230 int nlock;
231
232 /*
233 * If b->nlock > 1, the block is aliased within
234 * a single thread. That thread is us.
235 * DiskWrite does some funny stuff with QLock
236 * and blockPut that basically assumes b->nlock==1.
237 * We humor diskWrite by temporarily setting
238 * nlock to 1. This needs to be revisited.
239 */
240 nlock = b->nlock;
241 if(nlock > 1)
242 b->nlock = 1;
243 diskWrite(disk, b);
244 while(b->iostate != BioClean)
245 rsleep(&b->ioready);
246 b->nlock = nlock;
247 }
248
249 int
diskBlockSize(Disk * disk)250 diskBlockSize(Disk *disk)
251 {
252 return disk->h.blockSize; /* immuttable */
253 }
254
255 int
diskFlush(Disk * disk)256 diskFlush(Disk *disk)
257 {
258 Dir dir;
259
260 qlock(&disk->lk);
261 while(disk->nqueue > 0)
262 rsleep(&disk->flush);
263 qunlock(&disk->lk);
264
265 /* there really should be a cleaner interface to flush an fd */
266 nulldir(&dir);
267 if(dirfwstat(disk->fd, &dir) < 0){
268 werrstr("%r");
269 return 0;
270 }
271 return 1;
272 }
273
274 u32int
diskSize(Disk * disk,int part)275 diskSize(Disk *disk, int part)
276 {
277 return partEnd(disk, part) - partStart(disk, part);
278 }
279
280 static uintptr
mypc(int x)281 mypc(int x)
282 {
283 return getcallerpc(&x);
284 }
285
286 static char *
disk2file(Disk * disk)287 disk2file(Disk *disk)
288 {
289 static char buf[256];
290
291 #ifndef PLAN9PORT
292 if (fd2path(disk->fd, buf, sizeof buf) < 0)
293 strncpy(buf, "GOK", sizeof buf);
294 #endif
295 return buf;
296 }
297
298 static void
diskThread(void * a)299 diskThread(void *a)
300 {
301 Disk *disk = a;
302 Block *b;
303 uchar *buf, *p;
304 double t;
305 int nio;
306
307 threadsetname("disk");
308
309 //fprint(2, "diskThread %d\n", getpid());
310
311 buf = vtmalloc(disk->h.blockSize);
312
313 qlock(&disk->lk);
314 if (Timing) {
315 nio = 0;
316 t = -nsec();
317 }
318 for(;;){
319 while(disk->nqueue == 0){
320 if (Timing) {
321 t += nsec();
322 if(nio >= 10000){
323 fprint(2, "disk: io=%d at %.3fms\n",
324 nio, t*1e-6/nio);
325 nio = 0;
326 t = 0;
327 }
328 }
329 if(disk->die.l != nil)
330 goto Done;
331 rsleep(&disk->starve);
332 if (Timing)
333 t -= nsec();
334 }
335 assert(disk->cur != nil || disk->next != nil);
336
337 if(disk->cur == nil){
338 disk->cur = disk->next;
339 disk->next = nil;
340 }
341 b = disk->cur;
342 disk->cur = b->ionext;
343 qunlock(&disk->lk);
344
345 /*
346 * no one should hold onto blocking in the
347 * reading or writing state, so this lock should
348 * not cause deadlock.
349 */
350 if(0)fprint(2, "fossil: diskThread: %d:%d %x\n", getpid(), b->part, b->addr);
351 bwatchLock(b);
352 qlock(&b->lk);
353 b->pc = mypc(0);
354 assert(b->nlock == 1);
355 switch(b->iostate){
356 default:
357 abort();
358 case BioReading:
359 if(!diskReadRaw(disk, b->part, b->addr, b->data)){
360 fprint(2, "fossil: diskReadRaw failed: %s: "
361 "score %V: part=%s block %ud: %r\n",
362 disk2file(disk), b->score,
363 partname[b->part], b->addr);
364 blockSetIOState(b, BioReadError);
365 }else
366 blockSetIOState(b, BioClean);
367 break;
368 case BioWriting:
369 p = blockRollback(b, buf);
370 /* NB: ctime result ends with a newline */
371 if(!diskWriteRaw(disk, b->part, b->addr, p)){
372 fprint(2, "fossil: diskWriteRaw failed: %s: "
373 "score %V: date %s part=%s block %ud: %r\n",
374 disk2file(disk), b->score,
375 ctime(time(0)),
376 partname[b->part], b->addr);
377 break;
378 }
379 if(p != buf)
380 blockSetIOState(b, BioClean);
381 else
382 blockSetIOState(b, BioDirty);
383 break;
384 }
385
386 blockPut(b); /* remove extra reference, unlock */
387 qlock(&disk->lk);
388 disk->nqueue--;
389 if(disk->nqueue == QueueSize-1)
390 rwakeup(&disk->flow);
391 if(disk->nqueue == 0)
392 rwakeup(&disk->flush);
393 if(Timing)
394 nio++;
395 }
396 Done:
397 //fprint(2, "diskThread done\n");
398 disk->ref--;
399 rwakeup(&disk->die);
400 qunlock(&disk->lk);
401 vtfree(buf);
402 }
403