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 QueueSize = 100, /* maximum block to queue */ 10 }; 11 12 struct Disk { 13 VtLock *lk; 14 int ref; 15 16 int fd; 17 Header h; 18 19 VtRendez *flow; 20 VtRendez *starve; 21 VtRendez *flush; 22 VtRendez *die; 23 24 int nqueue; 25 26 Block *cur; /* block to do on current scan */ 27 Block *next; /* blocks to do next scan */ 28 }; 29 30 31 Disk * 32 diskAlloc(int fd) 33 { 34 u8int buf[HeaderSize]; 35 Header h; 36 Disk *disk; 37 38 if(pread(fd, buf, HeaderSize, HeaderOffset) < HeaderSize){ 39 vtOSError(); 40 return nil; 41 } 42 43 if(!headerUnpack(&h, buf)) 44 return nil; 45 disk = vtMemAllocZ(sizeof(Disk)); 46 disk->lk = vtLockAlloc(); 47 disk->starve = vtRendezAlloc(disk->lk); 48 disk->flow = vtRendezAlloc(disk->lk); 49 disk->flush = vtRendezAlloc(disk->lk); 50 disk->fd = fd; 51 disk->h = h; 52 53 disk->ref = 2; 54 vtThread(diskThread, disk); 55 56 return disk; 57 } 58 59 void 60 diskFree(Disk *disk) 61 { 62 diskFlush(disk); 63 64 /* kill slave */ 65 vtLock(disk->lk); 66 disk->die = vtRendezAlloc(disk->lk); 67 vtWakeup(disk->starve); 68 while(disk->ref > 1) 69 vtSleep(disk->die); 70 vtUnlock(disk->lk); 71 vtRendezFree(disk->flow); 72 vtRendezFree(disk->starve); 73 vtRendezFree(disk->die); 74 vtLockFree(disk->lk); 75 close(disk->fd); 76 vtMemFree(disk); 77 } 78 79 static u32int 80 partStart(Disk *disk, int part) 81 { 82 switch(part){ 83 default: 84 assert(0); 85 case PartSuper: 86 return disk->h.super; 87 case PartLabel: 88 return disk->h.label; 89 case PartData: 90 return disk->h.data; 91 } 92 } 93 94 95 static u32int 96 partEnd(Disk *disk, int part) 97 { 98 switch(part){ 99 default: 100 assert(0); 101 case PartSuper: 102 return disk->h.super+1; 103 case PartLabel: 104 return disk->h.data; 105 case PartData: 106 return disk->h.end; 107 } 108 } 109 110 int 111 diskReadRaw(Disk *disk, int part, u32int addr, uchar *buf) 112 { 113 ulong start, end; 114 u64int offset; 115 int n, nn; 116 117 start = partStart(disk, part); 118 end = partEnd(disk, part); 119 120 if(addr >= end-start){ 121 vtSetError(EBadAddr); 122 return 0; 123 } 124 125 offset = ((u64int)(addr + start))*disk->h.blockSize; 126 n = disk->h.blockSize; 127 while(n > 0){ 128 nn = pread(disk->fd, buf, n, offset); 129 if(nn < 0){ 130 vtOSError(); 131 return 0; 132 } 133 if(nn == 0){ 134 vtSetError(EIO); 135 return 0; 136 } 137 n -= nn; 138 offset += nn; 139 buf += nn; 140 } 141 return 1; 142 } 143 144 int 145 diskWriteRaw(Disk *disk, int part, u32int addr, uchar *buf) 146 { 147 ulong start, end; 148 u64int offset; 149 int n; 150 151 start = partStart(disk, part); 152 end = partEnd(disk, part); 153 154 if(addr >= end-start){ 155 vtSetError(EBadAddr); 156 return 0; 157 } 158 159 offset = ((u64int)(addr + start))*disk->h.blockSize; 160 n = disk->h.blockSize; 161 if(pwrite(disk->fd, buf, n, offset) < n){ 162 vtOSError(); 163 return 0; 164 } 165 166 return 1; 167 } 168 169 static void 170 diskQueue(Disk *disk, Block *b) 171 { 172 Block **bp, *bb; 173 174 vtLock(disk->lk); 175 while(disk->nqueue >= QueueSize) 176 vtSleep(disk->flow); 177 if(disk->cur == nil || b->addr > disk->cur->addr) 178 bp = &disk->cur; 179 else 180 bp = &disk->next; 181 182 for(bb=*bp; bb; bb=*bp){ 183 if(b->addr < bb->addr) 184 break; 185 bp = &bb->ionext; 186 } 187 b->ionext = bb; 188 *bp = b; 189 if(disk->nqueue == 0) 190 vtWakeup(disk->starve); 191 disk->nqueue++; 192 vtUnlock(disk->lk); 193 } 194 195 196 void 197 diskRead(Disk *disk, Block *b) 198 { 199 assert(b->iostate == BioEmpty || b->iostate == BioLabel); 200 blockSetIOState(b, BioReading); 201 diskQueue(disk, b); 202 } 203 204 void 205 diskWrite(Disk *disk, Block *b) 206 { 207 assert(b->iostate == BioDirty); 208 blockSetIOState(b, BioWriting); 209 diskQueue(disk, b); 210 } 211 212 int 213 diskBlockSize(Disk *disk) 214 { 215 return disk->h.blockSize; /* immuttable */ 216 } 217 218 int 219 diskFlush(Disk *disk) 220 { 221 Dir dir; 222 223 vtLock(disk->lk); 224 while(disk->nqueue > 0) 225 vtSleep(disk->flush); 226 vtUnlock(disk->lk); 227 228 /* there really should be a cleaner interface to flush an fd */ 229 nulldir(&dir); 230 if(dirfwstat(disk->fd, &dir) < 0){ 231 vtOSError(); 232 return 0; 233 } 234 return 1; 235 } 236 237 u32int 238 diskSize(Disk *disk, int part) 239 { 240 return partEnd(disk, part) - partStart(disk, part); 241 } 242 243 static void 244 diskThread(void *a) 245 { 246 Disk *disk = a; 247 Block *b; 248 uchar *buf, *p; 249 double t; 250 int nio; 251 252 vtThreadSetName("disk"); 253 254 fprint(2, "diskThread %d\n", getpid()); 255 256 buf = vtMemAlloc(disk->h.blockSize); 257 258 vtLock(disk->lk); 259 nio = 0; 260 t = -nsec(); 261 for(;;){ 262 while(disk->nqueue == 0){ 263 t += nsec(); 264 if(nio >= 10000){ 265 fprint(2, "disk: io=%d at %.3fms\n", nio, t*1e-6/nio); 266 nio = 0; 267 t = 0.; 268 } 269 if(disk->die != nil) 270 goto Done; 271 vtSleep(disk->starve); 272 t -= nsec(); 273 } 274 assert(disk->cur != nil || disk->next != nil); 275 276 if(disk->cur == nil){ 277 disk->cur = disk->next; 278 disk->next = nil; 279 } 280 b = disk->cur; 281 disk->cur = b->ionext; 282 vtUnlock(disk->lk); 283 284 /* 285 * no one should hold onto blocking in the 286 * reading or writing state, so this lock should 287 * not cause deadlock. 288 */ 289 if(0)fprint(2, "diskThread: %d:%d %x\n", getpid(), b->part, b->addr); 290 bwatchLock(b); 291 vtLock(b->lk); 292 assert(b->nlock == 1); 293 294 switch(b->iostate){ 295 default: 296 abort(); 297 case BioReading: 298 if(!diskReadRaw(disk, b->part, b->addr, b->data)){ 299 fprint(2, "diskReadRaw failed: part=%d addr=%ux: %r", b->part, b->addr); 300 blockSetIOState(b, BioReadError); 301 }else 302 blockSetIOState(b, BioClean); 303 break; 304 case BioWriting: 305 p = blockRollback(b, buf); 306 if(!diskWriteRaw(disk, b->part, b->addr, p)){ 307 fprint(2, "diskWriteRaw failed: part=%d addr=%ux: %r", b->part, b->addr); 308 break; 309 } 310 if(p != buf) 311 blockSetIOState(b, BioClean); 312 else 313 blockSetIOState(b, BioDirty); 314 break; 315 } 316 317 blockPut(b); /* remove extra reference, unlock */ 318 vtLock(disk->lk); 319 disk->nqueue--; 320 if(disk->nqueue == QueueSize-1) 321 vtWakeup(disk->flow); 322 if(disk->nqueue == 0) 323 vtWakeup(disk->flush); 324 nio++; 325 } 326 Done: 327 fprint(2, "diskThread done\n"); 328 disk->ref--; 329 vtWakeup(disk->die); 330 vtUnlock(disk->lk); 331 vtMemFree(buf); 332 } 333