tvbackup.c - plan9port - [fork] Plan 9 from user space
HTML git clone git://src.adamsgaard.dk/plan9port
DIR Log
DIR Files
DIR Refs
DIR README
DIR LICENSE
---
tvbackup.c (14404B)
---
1 /*
2 * vbackup [-Dnv] fspartition [score]
3 *
4 * Copy a file system to a disk image stored on Venti.
5 * Prints a vnfs config line for the copied image.
6 *
7 * -D print debugging
8 * -f 'fast' writes - skip write if block exists on server
9 * -i read old scores incrementally
10 * -m set mount name
11 * -M set mount place
12 * -n nop -- don't actually write blocks
13 * -s print status updates
14 * -v print debugging trace
15 * -w write parallelism
16 *
17 * If score is given on the command line, it should be the
18 * score from a previous vbackup on this fspartition.
19 * In this mode, only the new blocks are stored to Venti.
20 * The result is still a complete image, but requires many
21 * fewer Venti writes in the common case.
22 *
23 * This program is structured as three processes connected
24 * by buffered queues:
25 *
26 * fsysproc | cmpproc | ventiproc
27 *
28 * Fsysproc reads the disk and queues the blocks.
29 * Cmpproc compares the blocks against the SHA1 hashes
30 * in the old image, if any. It discards the unchanged blocks
31 * and queues the changed ones. Ventiproc writes blocks to Venti.
32 *
33 * There is a fourth proc, statusproc, which prints status
34 * updates about how the various procs are progressing.
35 */
36
37 #include <u.h>
38 #include <libc.h>
39 #include <bio.h>
40 #include <thread.h>
41 #include <libsec.h>
42 #include <venti.h>
43 #include <diskfs.h>
44 #include "queue.h"
45
46 enum
47 {
48 STACK = 32768
49 };
50
51 typedef struct WriteReq WriteReq;
52 struct WriteReq
53 {
54 Packet *p;
55 uint type;
56 uchar score[VtScoreSize];
57 };
58
59 Biobuf bscores; /* biobuf filled with block scores */
60 int debug; /* debugging flag (not used) */
61 Disk* disk; /* disk being backed up */
62 RWLock endlk; /* silly synchonization */
63 int errors; /* are we exiting with an error status? */
64 int fastwrites; /* do not write blocks already on server */
65 int fsscanblock; /* last block scanned */
66 Fsys* fsys; /* file system being backed up */
67 int incremental; /* use vscores rather than bscores */
68 int nchange; /* number of changed blocks */
69 int nop; /* don't actually send blocks to venti */
70 int nskip; /* number of blocks skipped (already on server) */
71 int nwritethread; /* number of write-behind threads */
72 Queue* qcmp; /* queue fsys->cmp */
73 Queue* qventi; /* queue cmp->venti */
74 int statustime; /* print status every _ seconds */
75 int verbose; /* print extra stuff */
76 VtFile* vfile; /* venti file being written */
77 VtFile* vscores; /* venti file with block scores */
78 Channel* writechan; /* chan(WriteReq) */
79 VtConn* z; /* connection to venti */
80 VtCache* zcache; /* cache of venti blocks */
81 uchar* zero; /* blocksize zero bytes */
82
83 int nsend, nrecv;
84
85 void cmpproc(void*);
86 void fsysproc(void*);
87 void statusproc(void*);
88 void ventiproc(void*);
89 int timefmt(Fmt*);
90 char* guessmountplace(char *dev);
91
92 void
93 usage(void)
94 {
95 fprint(2, "usage: vbackup [-DVnv] [-m mtpt] [-M mtpl] [-s secs] [-w n] disk [score]\n");
96 threadexitsall("usage");
97 }
98
99 void
100 threadmain(int argc, char **argv)
101 {
102 char *pref, *mountname, *mountplace;
103 uchar score[VtScoreSize], prev[VtScoreSize];
104 int i, fd, csize;
105 vlong bsize;
106 Tm tm;
107 VtEntry e;
108 VtBlock *b;
109 VtCache *c;
110 VtRoot root;
111 char *tmp, *tmpnam;
112
113 fmtinstall('F', vtfcallfmt);
114 fmtinstall('H', encodefmt);
115 fmtinstall('T', timefmt);
116 fmtinstall('V', vtscorefmt);
117
118 mountname = sysname();
119 mountplace = nil;
120 ARGBEGIN{
121 default:
122 usage();
123 break;
124 case 'D':
125 debug++;
126 break;
127 case 'V':
128 chattyventi = 1;
129 break;
130 case 'f':
131 fastwrites = 1;
132 break;
133 case 'i':
134 incremental = 1;
135 break;
136 case 'm':
137 mountname = EARGF(usage());
138 break;
139 case 'M':
140 mountplace = EARGF(usage());
141 i = strlen(mountplace);
142 if(i > 0 && mountplace[i-1] == '/')
143 mountplace[i-1] = 0;
144 break;
145 case 'n':
146 nop = 1;
147 break;
148 case 's':
149 statustime = atoi(EARGF(usage()));
150 break;
151 case 'v':
152 verbose = 1;
153 break;
154 case 'w':
155 nwritethread = atoi(EARGF(usage()));
156 break;
157 }ARGEND
158
159 if(argc != 1 && argc != 2)
160 usage();
161
162 if(statustime)
163 print("# %T vbackup %s %s\n", argv[0], argc>=2 ? argv[1] : "");
164
165 /*
166 * open fs
167 */
168 if((disk = diskopenfile(argv[0])) == nil)
169 sysfatal("diskopen: %r");
170 if((disk = diskcache(disk, 32768, 2*MAXQ+16)) == nil)
171 sysfatal("diskcache: %r");
172 if((fsys = fsysopen(disk)) == nil)
173 sysfatal("fsysopen: %r");
174
175 /*
176 * connect to venti
177 */
178 if((z = vtdial(nil)) == nil)
179 sysfatal("vtdial: %r");
180 if(vtconnect(z) < 0)
181 sysfatal("vtconnect: %r");
182
183 /*
184 * set up venti block cache
185 */
186 zero = vtmallocz(fsys->blocksize);
187 bsize = fsys->blocksize;
188 csize = 50; /* plenty; could probably do with 5 */
189
190 if(verbose)
191 fprint(2, "cache %d blocks\n", csize);
192 c = vtcachealloc(z, bsize*csize);
193 zcache = c;
194
195 /*
196 * parse starting score
197 */
198 memset(prev, 0, sizeof prev);
199 if(argc == 1){
200 vfile = vtfilecreateroot(c, (fsys->blocksize/VtScoreSize)*VtScoreSize,
201 fsys->blocksize, VtDataType);
202 if(vfile == nil)
203 sysfatal("vtfilecreateroot: %r");
204 vtfilelock(vfile, VtORDWR);
205 if(vtfilewrite(vfile, zero, 1, bsize*fsys->nblock-1) != 1)
206 sysfatal("vtfilewrite: %r");
207 if(vtfileflush(vfile) < 0)
208 sysfatal("vtfileflush: %r");
209 }else{
210 if(vtparsescore(argv[1], &pref, score) < 0)
211 sysfatal("bad score: %r");
212 if(pref!=nil && strcmp(pref, fsys->type) != 0)
213 sysfatal("score is %s but fsys is %s", pref, fsys->type);
214 b = vtcacheglobal(c, score, VtRootType, VtRootSize);
215 if(b){
216 if(vtrootunpack(&root, b->data) < 0)
217 sysfatal("bad root: %r");
218 if(strcmp(root.type, fsys->type) != 0)
219 sysfatal("root is %s but fsys is %s", root.type, fsys->type);
220 memmove(prev, score, VtScoreSize);
221 memmove(score, root.score, VtScoreSize);
222 vtblockput(b);
223 }
224 b = vtcacheglobal(c, score, VtDirType, VtEntrySize);
225 if(b == nil)
226 sysfatal("vtcacheglobal %V: %r", score);
227 if(vtentryunpack(&e, b->data, 0) < 0)
228 sysfatal("%V: vtentryunpack failed", score);
229 if(verbose)
230 fprint(2, "entry: size %llud psize %d dsize %d\n",
231 e.size, e.psize, e.dsize);
232 vtblockput(b);
233 if((vfile = vtfileopenroot(c, &e)) == nil)
234 sysfatal("vtfileopenroot: %r");
235 vtfilelock(vfile, VtORDWR);
236 if(e.dsize != bsize)
237 sysfatal("file system block sizes don't match %d %lld", e.dsize, bsize);
238 if(e.size != fsys->nblock*bsize)
239 sysfatal("file system block counts don't match %lld %lld", e.size, fsys->nblock*bsize);
240 }
241
242 tmpnam = nil;
243 if(incremental){
244 if(vtfilegetentry(vfile, &e) < 0)
245 sysfatal("vtfilegetentry: %r");
246 if((vscores = vtfileopenroot(c, &e)) == nil)
247 sysfatal("vtfileopenroot: %r");
248 vtfileunlock(vfile);
249 }else{
250 /*
251 * write scores of blocks into temporary file
252 */
253 if((tmp = getenv("TMP")) != nil){
254 /* okay, good */
255 }else if(access("/var/tmp", 0) >= 0)
256 tmp = "/var/tmp";
257 else
258 tmp = "/tmp";
259 tmpnam = smprint("%s/vbackup.XXXXXX", tmp);
260 if(tmpnam == nil)
261 sysfatal("smprint: %r");
262
263 if((fd = opentemp(tmpnam, ORDWR|ORCLOSE)) < 0)
264 sysfatal("opentemp %s: %r", tmpnam);
265 if(statustime)
266 print("# %T reading scores into %s\n", tmpnam);
267 if(verbose)
268 fprint(2, "read scores into %s...\n", tmpnam);
269
270 Binit(&bscores, fd, OWRITE);
271 for(i=0; i<fsys->nblock; i++){
272 if(vtfileblockscore(vfile, i, score) < 0)
273 sysfatal("vtfileblockhash %d: %r", i);
274 if(Bwrite(&bscores, score, VtScoreSize) != VtScoreSize)
275 sysfatal("Bwrite: %r");
276 }
277 Bterm(&bscores);
278 vtfileunlock(vfile);
279
280 /*
281 * prep scores for rereading
282 */
283 seek(fd, 0, 0);
284 Binit(&bscores, fd, OREAD);
285 }
286
287 /*
288 * start the main processes
289 */
290 if(statustime)
291 print("# %T starting procs\n");
292 qcmp = qalloc();
293 qventi = qalloc();
294
295 rlock(&endlk);
296 proccreate(fsysproc, nil, STACK);
297 rlock(&endlk);
298 proccreate(ventiproc, nil, STACK);
299 rlock(&endlk);
300 proccreate(cmpproc, nil, STACK);
301 if(statustime){
302 rlock(&endlk);
303 proccreate(statusproc, nil, STACK);
304 }
305
306 /*
307 * wait for processes to finish
308 */
309 wlock(&endlk);
310
311 qfree(qcmp);
312 qfree(qventi);
313
314 if(statustime)
315 print("# %T procs exited: %d of %d %d-byte blocks changed, "
316 "%d read, %d written, %d skipped, %d copied\n",
317 nchange, fsys->nblock, fsys->blocksize,
318 vtcachenread, vtcachenwrite, nskip, vtcachencopy);
319
320 /*
321 * prepare root block
322 */
323 if(incremental)
324 vtfileclose(vscores);
325 vtfilelock(vfile, -1);
326 if(vtfileflush(vfile) < 0)
327 sysfatal("vtfileflush: %r");
328 if(vtfilegetentry(vfile, &e) < 0)
329 sysfatal("vtfilegetentry: %r");
330 vtfileunlock(vfile);
331 vtfileclose(vfile);
332
333 b = vtcacheallocblock(c, VtDirType, VtEntrySize);
334 if(b == nil)
335 sysfatal("vtcacheallocblock: %r");
336 vtentrypack(&e, b->data, 0);
337 if(vtblockwrite(b) < 0)
338 sysfatal("vtblockwrite: %r");
339
340 memset(&root, 0, sizeof root);
341 strecpy(root.name, root.name+sizeof root.name, argv[0]);
342 strecpy(root.type, root.type+sizeof root.type, fsys->type);
343 memmove(root.score, b->score, VtScoreSize);
344 root.blocksize = fsys->blocksize;
345 memmove(root.prev, prev, VtScoreSize);
346 vtblockput(b);
347
348 b = vtcacheallocblock(c, VtRootType, VtRootSize);
349 if(b == nil)
350 sysfatal("vtcacheallocblock: %r");
351 vtrootpack(&root, b->data);
352 if(vtblockwrite(b) < 0)
353 sysfatal("vtblockwrite: %r");
354
355 tm = *localtime(time(0));
356 tm.year += 1900;
357 tm.mon++;
358 if(mountplace == nil)
359 mountplace = guessmountplace(argv[0]);
360 print("mount /%s/%d/%02d%02d%s %s:%V %d/%02d%02d/%02d%02d\n",
361 mountname, tm.year, tm.mon, tm.mday,
362 mountplace,
363 root.type, b->score,
364 tm.year, tm.mon, tm.mday, tm.hour, tm.min);
365 print("# %T %s %s:%V\n", argv[0], root.type, b->score);
366 if(statustime)
367 print("# %T venti sync\n");
368 vtblockput(b);
369 if(vtsync(z) < 0)
370 sysfatal("vtsync: %r");
371 if(statustime)
372 print("# %T synced\n");
373
374 fsysclose(fsys);
375 diskclose(disk);
376 vtcachefree(zcache);
377
378 // Vtgoodbye hangs on Linux - not sure why.
379 // Probably vtfcallrpc doesn't quite get the
380 // remote hangup right. Also probably related
381 // to the vtrecvproc problem below.
382 // vtgoodbye(z);
383
384 // Leak here, because I can't seem to make
385 // the vtrecvproc exit.
386 // vtfreeconn(z);
387
388 free(tmpnam);
389 z = nil;
390 zcache = nil;
391 fsys = nil;
392 disk = nil;
393 threadexitsall(nil);
394 }
395
396 void
397 fsysproc(void *dummy)
398 {
399 u32int i;
400 Block *db;
401
402 USED(dummy);
403
404 for(i=0; i<fsys->nblock; i++){
405 fsscanblock = i;
406 if((db = fsysreadblock(fsys, i)) != nil)
407 qwrite(qcmp, db, i);
408 }
409 fsscanblock = i;
410 qclose(qcmp);
411
412 if(statustime)
413 print("# %T fsys proc exiting\n");
414 runlock(&endlk);
415 }
416
417 void
418 cmpproc(void *dummy)
419 {
420 uchar *data;
421 Block *db;
422 u32int bno, bsize;
423 uchar score[VtScoreSize];
424 uchar score1[VtScoreSize];
425
426 USED(dummy);
427
428 if(incremental)
429 vtfilelock(vscores, VtOREAD);
430 bsize = fsys->blocksize;
431 while((db = qread(qcmp, &bno)) != nil){
432 data = db->data;
433 sha1(data, vtzerotruncate(VtDataType, data, bsize), score, nil);
434 if(incremental){
435 if(vtfileblockscore(vscores, bno, score1) < 0)
436 sysfatal("cmpproc vtfileblockscore %d: %r", bno);
437 }else{
438 if(Bseek(&bscores, (vlong)bno*VtScoreSize, 0) < 0)
439 sysfatal("cmpproc Bseek: %r");
440 if(Bread(&bscores, score1, VtScoreSize) != VtScoreSize)
441 sysfatal("cmpproc Bread: %r");
442 }
443 if(memcmp(score, score1, VtScoreSize) != 0){
444 nchange++;
445 if(verbose)
446 print("# block %ud: old %V new %V\n", bno, score1, score);
447 qwrite(qventi, db, bno);
448 }else
449 blockput(db);
450 }
451 qclose(qventi);
452 if(incremental)
453 vtfileunlock(vscores);
454 if(statustime)
455 print("# %T cmp proc exiting\n");
456 runlock(&endlk);
457 }
458
459 void
460 writethread(void *v)
461 {
462 WriteReq wr;
463 char err[ERRMAX];
464
465 USED(v);
466
467 while(recv(writechan, &wr) == 1){
468 nrecv++;
469 if(wr.p == nil)
470 break;
471
472 if(fastwrites && vtread(z, wr.score, wr.type, nil, 0) < 0){
473 rerrstr(err, sizeof err);
474 if(strstr(err, "read too small")){ /* already exists */
475 nskip++;
476 packetfree(wr.p);
477 continue;
478 }
479 }
480 if(vtwritepacket(z, wr.score, wr.type, wr.p) < 0)
481 sysfatal("vtwritepacket: %r");
482 packetfree(wr.p);
483 }
484 }
485
486 int
487 myvtwrite(VtConn *z, uchar score[VtScoreSize], uint type, uchar *buf, int n)
488 {
489 WriteReq wr;
490
491 if(nwritethread == 0){
492 n = vtwrite(z, score, type, buf, n);
493 if(n < 0)
494 sysfatal("vtwrite: %r");
495 return n;
496 }
497
498 wr.p = packetalloc();
499 packetappend(wr.p, buf, n);
500 packetsha1(wr.p, score);
501 memmove(wr.score, score, VtScoreSize);
502 wr.type = type;
503 nsend++;
504 send(writechan, &wr);
505 return 0;
506 }
507
508 void
509 ventiproc(void *dummy)
510 {
511 int i;
512 Block *db;
513 u32int bno;
514 u64int bsize;
515
516 USED(dummy);
517
518 proccreate(vtsendproc, z, STACK);
519 proccreate(vtrecvproc, z, STACK);
520
521 writechan = chancreate(sizeof(WriteReq), 0);
522 for(i=0; i<nwritethread; i++)
523 threadcreate(writethread, nil, STACK);
524 vtcachesetwrite(zcache, myvtwrite);
525
526 bsize = fsys->blocksize;
527 vtfilelock(vfile, -1);
528 while((db = qread(qventi, &bno)) != nil){
529 if(nop){
530 blockput(db);
531 continue;
532 }
533 if(vtfilewrite(vfile, db->data, bsize, bno*bsize) != bsize)
534 sysfatal("ventiproc vtfilewrite: %r");
535 if(vtfileflushbefore(vfile, (bno+1)*bsize) < 0)
536 sysfatal("ventiproc vtfileflushbefore: %r");
537 blockput(db);
538 }
539 vtfileunlock(vfile);
540 vtcachesetwrite(zcache, nil);
541 for(i=0; i<nwritethread; i++)
542 send(writechan, nil);
543 chanfree(writechan);
544 if(statustime)
545 print("# %T venti proc exiting - nsend %d nrecv %d\n", nsend, nrecv);
546 runlock(&endlk);
547 }
548
549 static int
550 percent(u32int a, u32int b)
551 {
552 return (vlong)a*100/b;
553 }
554
555 void
556 statusproc(void *dummy)
557 {
558 int n;
559 USED(dummy);
560
561 for(n=0;;n++){
562 sleep(1000);
563 if(qcmp->closed && qcmp->nel==0 && qventi->closed && qventi->nel==0)
564 break;
565 if(n < statustime)
566 continue;
567 n = 0;
568 print("# %T fsscan=%d%% cmpq=%d%% ventiq=%d%%\n",
569 percent(fsscanblock, fsys->nblock),
570 percent(qcmp->nel, MAXQ),
571 percent(qventi->nel, MAXQ));
572 }
573 print("# %T status proc exiting\n");
574 runlock(&endlk);
575 }
576
577 int
578 timefmt(Fmt *fmt)
579 {
580 vlong ns;
581 Tm tm;
582 ns = nsec();
583 tm = *localtime(time(0));
584 return fmtprint(fmt, "%04d/%02d%02d %02d:%02d:%02d.%03d",
585 tm.year+1900, tm.mon+1, tm.mday, tm.hour, tm.min, tm.sec,
586 (int)(ns%1000000000)/1000000);
587 }
588
589 char*
590 guessmountplace(char *dev)
591 {
592 char *cmd, *q;
593 int p[2], fd[3], n;
594 char buf[100];
595
596 if(pipe(p) < 0)
597 sysfatal("pipe: %r");
598
599 fd[0] = -1;
600 fd[1] = p[1];
601 fd[2] = -1;
602 cmd = smprint("mount | awk 'BEGIN{v=\"%s\"; u=v; sub(/rdisk/, \"disk\", u);} ($1==v||$1==u) && $2 == \"on\" {print $3}'", dev);
603 if(threadspawnl(fd, "sh", "sh", "-c", cmd, nil) < 0)
604 sysfatal("exec mount|awk (to find mtpt of %s): %r", dev);
605 /* threadspawnl closed p[1] */
606 free(cmd);
607 n = readn(p[0], buf, sizeof buf-1);
608 close(p[0]);
609 if(n <= 0)
610 return dev;
611 buf[n] = 0;
612 if((q = strchr(buf, '\n')) == nil)
613 return dev;
614 *q = 0;
615 q = buf+strlen(buf);
616 if(q>buf && *(q-1) == '/')
617 *--q = 0;
618 return strdup(buf);
619 }