1 /*- 2 * Copyright (c) 1980, 1991, 1993 3 * The Regents of the University of California. All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 8 * 1. Redistributions of source code must retain the above copyright 9 * notice, this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * 4. Neither the name of the University nor the names of its contributors 14 * may be used to endorse or promote products derived from this software 15 * without specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND 18 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE 21 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 22 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 23 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 24 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 25 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 26 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 27 * SUCH DAMAGE. 28 */ 29 30 #ifndef lint 31 #if 0 32 static char sccsid[] = "@(#)tape.c 8.4 (Berkeley) 5/1/95"; 33 #endif 34 static const char rcsid[] = 35 "$FreeBSD$"; 36 #endif /* not lint */ 37 38 #include <sys/param.h> 39 #include <sys/socket.h> 40 #include <sys/wait.h> 41 #include <sys/stat.h> 42 43 #include <ufs/ufs/dinode.h> 44 #include <ufs/ffs/fs.h> 45 46 #include <protocols/dumprestore.h> 47 48 #include <errno.h> 49 #include <fcntl.h> 50 #include <limits.h> 51 #include <setjmp.h> 52 #include <signal.h> 53 #include <stdio.h> 54 #include <stdlib.h> 55 #include <string.h> 56 #include <time.h> 57 #include <unistd.h> 58 59 #include "dump.h" 60 61 int writesize; /* size of malloc()ed buffer for tape */ 62 int64_t lastspclrec = -1; /* tape block number of last written header */ 63 int trecno = 0; /* next record to write in current block */ 64 extern long blocksperfile; /* number of blocks per output file */ 65 long blocksthisvol; /* number of blocks on current output file */ 66 extern int ntrec; /* blocking factor on tape */ 67 extern int cartridge; 68 extern char *host; 69 char *nexttape; 70 FILE *popenfp = NULL; 71 72 static int atomic(ssize_t (*)(), int, char *, int); 73 static void doslave(int, int); 74 static void enslave(void); 75 static void flushtape(void); 76 static void killall(void); 77 static void rollforward(void); 78 79 /* 80 * Concurrent dump mods (Caltech) - disk block reading and tape writing 81 * are exported to several slave processes. While one slave writes the 82 * tape, the others read disk blocks; they pass control of the tape in 83 * a ring via signals. The parent process traverses the file system and 84 * sends writeheader()'s and lists of daddr's to the slaves via pipes. 85 * The following structure defines the instruction packets sent to slaves. 86 */ 87 struct req { 88 ufs2_daddr_t dblk; 89 int count; 90 }; 91 int reqsiz; 92 93 #define SLAVES 3 /* 1 slave writing, 1 reading, 1 for slack */ 94 struct slave { 95 int64_t tapea; /* header number at start of this chunk */ 96 int64_t firstrec; /* record number of this block */ 97 int count; /* count to next header (used for TS_TAPE */ 98 /* after EOT) */ 99 int inode; /* inode that we are currently dealing with */ 100 int fd; /* FD for this slave */ 101 int pid; /* PID for this slave */ 102 int sent; /* 1 == we've sent this slave requests */ 103 char (*tblock)[TP_BSIZE]; /* buffer for data blocks */ 104 struct req *req; /* buffer for requests */ 105 } slaves[SLAVES+1]; 106 struct slave *slp; 107 108 char (*nextblock)[TP_BSIZE]; 109 110 int master; /* pid of master, for sending error signals */ 111 int tenths; /* length of tape used per block written */ 112 static volatile sig_atomic_t caught; /* have we caught the signal to proceed? */ 113 static volatile sig_atomic_t ready; /* reached the lock point without having */ 114 /* received the SIGUSR2 signal from the prev slave? */ 115 static jmp_buf jmpbuf; /* where to jump to if we are ready when the */ 116 /* SIGUSR2 arrives from the previous slave */ 117 118 int 119 alloctape(void) 120 { 121 int pgoff = getpagesize() - 1; 122 char *buf; 123 int i; 124 125 writesize = ntrec * TP_BSIZE; 126 reqsiz = (ntrec + 1) * sizeof(struct req); 127 /* 128 * CDC 92181's and 92185's make 0.8" gaps in 1600-bpi start/stop mode 129 * (see DEC TU80 User's Guide). The shorter gaps of 6250-bpi require 130 * repositioning after stopping, i.e, streaming mode, where the gap is 131 * variable, 0.30" to 0.45". The gap is maximal when the tape stops. 132 */ 133 if (blocksperfile == 0 && !unlimited) 134 tenths = writesize / density + 135 (cartridge ? 16 : density == 625 ? 5 : 8); 136 /* 137 * Allocate tape buffer contiguous with the array of instruction 138 * packets, so flushtape() can write them together with one write(). 139 * Align tape buffer on page boundary to speed up tape write(). 140 */ 141 for (i = 0; i <= SLAVES; i++) { 142 buf = (char *) 143 malloc((unsigned)(reqsiz + writesize + pgoff + TP_BSIZE)); 144 if (buf == NULL) 145 return(0); 146 slaves[i].tblock = (char (*)[TP_BSIZE]) 147 (((long)&buf[ntrec + 1] + pgoff) &~ pgoff); 148 slaves[i].req = (struct req *)slaves[i].tblock - ntrec - 1; 149 } 150 slp = &slaves[0]; 151 slp->count = 1; 152 slp->tapea = 0; 153 slp->firstrec = 0; 154 nextblock = slp->tblock; 155 return(1); 156 } 157 158 void 159 writerec(char *dp, int isspcl) 160 { 161 162 slp->req[trecno].dblk = (ufs2_daddr_t)0; 163 slp->req[trecno].count = 1; 164 /* Can't do a structure assignment due to alignment problems */ 165 bcopy(dp, *(nextblock)++, sizeof (union u_spcl)); 166 if (isspcl) 167 lastspclrec = spcl.c_tapea; 168 trecno++; 169 spcl.c_tapea++; 170 if (trecno >= ntrec) 171 flushtape(); 172 } 173 174 void 175 dumpblock(ufs2_daddr_t blkno, int size) 176 { 177 int avail, tpblks; 178 ufs2_daddr_t dblkno; 179 180 dblkno = fsbtodb(sblock, blkno); 181 tpblks = size >> tp_bshift; 182 while ((avail = MIN(tpblks, ntrec - trecno)) > 0) { 183 slp->req[trecno].dblk = dblkno; 184 slp->req[trecno].count = avail; 185 trecno += avail; 186 spcl.c_tapea += avail; 187 if (trecno >= ntrec) 188 flushtape(); 189 dblkno += avail << (tp_bshift - dev_bshift); 190 tpblks -= avail; 191 } 192 } 193 194 int nogripe = 0; 195 196 void 197 tperror(int signo __unused) 198 { 199 200 if (pipeout) { 201 msg("write error on %s\n", tape); 202 quit("Cannot recover\n"); 203 /* NOTREACHED */ 204 } 205 msg("write error %ld blocks into volume %d\n", blocksthisvol, tapeno); 206 broadcast("DUMP WRITE ERROR!\n"); 207 if (!query("Do you want to restart?")) 208 dumpabort(0); 209 msg("Closing this volume. Prepare to restart with new media;\n"); 210 msg("this dump volume will be rewritten.\n"); 211 killall(); 212 nogripe = 1; 213 close_rewind(); 214 Exit(X_REWRITE); 215 } 216 217 void 218 sigpipe(int signo __unused) 219 { 220 221 quit("Broken pipe\n"); 222 } 223 224 static void 225 flushtape(void) 226 { 227 int i, blks, got; 228 int64_t lastfirstrec; 229 230 int siz = (char *)nextblock - (char *)slp->req; 231 232 slp->req[trecno].count = 0; /* Sentinel */ 233 234 if (atomic(write, slp->fd, (char *)slp->req, siz) != siz) 235 quit("error writing command pipe: %s\n", strerror(errno)); 236 slp->sent = 1; /* we sent a request, read the response later */ 237 238 lastfirstrec = slp->firstrec; 239 240 if (++slp >= &slaves[SLAVES]) 241 slp = &slaves[0]; 242 243 /* Read results back from next slave */ 244 if (slp->sent) { 245 if (atomic(read, slp->fd, (char *)&got, sizeof got) 246 != sizeof got) { 247 perror(" DUMP: error reading command pipe in master"); 248 dumpabort(0); 249 } 250 slp->sent = 0; 251 252 /* Check for end of tape */ 253 if (got < writesize) { 254 msg("End of tape detected\n"); 255 256 /* 257 * Drain the results, don't care what the values were. 258 * If we read them here then trewind won't... 259 */ 260 for (i = 0; i < SLAVES; i++) { 261 if (slaves[i].sent) { 262 if (atomic(read, slaves[i].fd, 263 (char *)&got, sizeof got) 264 != sizeof got) { 265 perror(" DUMP: error reading command pipe in master"); 266 dumpabort(0); 267 } 268 slaves[i].sent = 0; 269 } 270 } 271 272 close_rewind(); 273 rollforward(); 274 return; 275 } 276 } 277 278 blks = 0; 279 if (spcl.c_type != TS_END) { 280 for (i = 0; i < spcl.c_count; i++) 281 if (spcl.c_addr[i] != 0) 282 blks++; 283 } 284 slp->count = lastspclrec + blks + 1 - spcl.c_tapea; 285 slp->tapea = spcl.c_tapea; 286 slp->firstrec = lastfirstrec + ntrec; 287 slp->inode = curino; 288 nextblock = slp->tblock; 289 trecno = 0; 290 asize += tenths; 291 blockswritten += ntrec; 292 blocksthisvol += ntrec; 293 if (!pipeout && !unlimited && (blocksperfile ? 294 (blocksthisvol >= blocksperfile) : (asize > tsize))) { 295 close_rewind(); 296 startnewtape(0); 297 } 298 timeest(); 299 } 300 301 void 302 trewind(void) 303 { 304 struct stat sb; 305 int f; 306 int got; 307 308 for (f = 0; f < SLAVES; f++) { 309 /* 310 * Drain the results, but unlike EOT we DO (or should) care 311 * what the return values were, since if we detect EOT after 312 * we think we've written the last blocks to the tape anyway, 313 * we have to replay those blocks with rollforward. 314 * 315 * fixme: punt for now. 316 */ 317 if (slaves[f].sent) { 318 if (atomic(read, slaves[f].fd, (char *)&got, sizeof got) 319 != sizeof got) { 320 perror(" DUMP: error reading command pipe in master"); 321 dumpabort(0); 322 } 323 slaves[f].sent = 0; 324 if (got != writesize) { 325 msg("EOT detected in last 2 tape records!\n"); 326 msg("Use a longer tape, decrease the size estimate\n"); 327 quit("or use no size estimate at all.\n"); 328 } 329 } 330 (void) close(slaves[f].fd); 331 } 332 while (wait((int *)NULL) >= 0) /* wait for any signals from slaves */ 333 /* void */; 334 335 if (pipeout) 336 return; 337 338 msg("Closing %s\n", tape); 339 340 if (popenout) { 341 tapefd = -1; 342 (void)pclose(popenfp); 343 popenfp = NULL; 344 return; 345 } 346 #ifdef RDUMP 347 if (host) { 348 rmtclose(); 349 while (rmtopen(tape, 0) < 0) 350 sleep(10); 351 rmtclose(); 352 return; 353 } 354 #endif 355 if (fstat(tapefd, &sb) == 0 && S_ISFIFO(sb.st_mode)) { 356 (void)close(tapefd); 357 return; 358 } 359 (void) close(tapefd); 360 while ((f = open(tape, 0)) < 0) 361 sleep (10); 362 (void) close(f); 363 } 364 365 void 366 close_rewind() 367 { 368 time_t tstart_changevol, tend_changevol; 369 370 trewind(); 371 if (nexttape) 372 return; 373 (void)time((time_t *)&(tstart_changevol)); 374 if (!nogripe) { 375 msg("Change Volumes: Mount volume #%d\n", tapeno+1); 376 broadcast("CHANGE DUMP VOLUMES!\a\a\n"); 377 } 378 while (!query("Is the new volume mounted and ready to go?")) 379 if (query("Do you want to abort?")) { 380 dumpabort(0); 381 /*NOTREACHED*/ 382 } 383 (void)time((time_t *)&(tend_changevol)); 384 if ((tstart_changevol != (time_t)-1) && (tend_changevol != (time_t)-1)) 385 tstart_writing += (tend_changevol - tstart_changevol); 386 } 387 388 void 389 rollforward(void) 390 { 391 struct req *p, *q, *prev; 392 struct slave *tslp; 393 int i, size, got; 394 int64_t savedtapea; 395 union u_spcl *ntb, *otb; 396 tslp = &slaves[SLAVES]; 397 ntb = (union u_spcl *)tslp->tblock[1]; 398 399 /* 400 * Each of the N slaves should have requests that need to 401 * be replayed on the next tape. Use the extra slave buffers 402 * (slaves[SLAVES]) to construct request lists to be sent to 403 * each slave in turn. 404 */ 405 for (i = 0; i < SLAVES; i++) { 406 q = &tslp->req[1]; 407 otb = (union u_spcl *)slp->tblock; 408 409 /* 410 * For each request in the current slave, copy it to tslp. 411 */ 412 413 prev = NULL; 414 for (p = slp->req; p->count > 0; p += p->count) { 415 *q = *p; 416 if (p->dblk == 0) 417 *ntb++ = *otb++; /* copy the datablock also */ 418 prev = q; 419 q += q->count; 420 } 421 if (prev == NULL) 422 quit("rollforward: protocol botch"); 423 if (prev->dblk != 0) 424 prev->count -= 1; 425 else 426 ntb--; 427 q -= 1; 428 q->count = 0; 429 q = &tslp->req[0]; 430 if (i == 0) { 431 q->dblk = 0; 432 q->count = 1; 433 trecno = 0; 434 nextblock = tslp->tblock; 435 savedtapea = spcl.c_tapea; 436 spcl.c_tapea = slp->tapea; 437 startnewtape(0); 438 spcl.c_tapea = savedtapea; 439 lastspclrec = savedtapea - 1; 440 } 441 size = (char *)ntb - (char *)q; 442 if (atomic(write, slp->fd, (char *)q, size) != size) { 443 perror(" DUMP: error writing command pipe"); 444 dumpabort(0); 445 } 446 slp->sent = 1; 447 if (++slp >= &slaves[SLAVES]) 448 slp = &slaves[0]; 449 450 q->count = 1; 451 452 if (prev->dblk != 0) { 453 /* 454 * If the last one was a disk block, make the 455 * first of this one be the last bit of that disk 456 * block... 457 */ 458 q->dblk = prev->dblk + 459 prev->count * (TP_BSIZE / DEV_BSIZE); 460 ntb = (union u_spcl *)tslp->tblock; 461 } else { 462 /* 463 * It wasn't a disk block. Copy the data to its 464 * new location in the buffer. 465 */ 466 q->dblk = 0; 467 *((union u_spcl *)tslp->tblock) = *ntb; 468 ntb = (union u_spcl *)tslp->tblock[1]; 469 } 470 } 471 slp->req[0] = *q; 472 nextblock = slp->tblock; 473 if (q->dblk == 0) 474 nextblock++; 475 trecno = 1; 476 477 /* 478 * Clear the first slaves' response. One hopes that it 479 * worked ok, otherwise the tape is much too short! 480 */ 481 if (slp->sent) { 482 if (atomic(read, slp->fd, (char *)&got, sizeof got) 483 != sizeof got) { 484 perror(" DUMP: error reading command pipe in master"); 485 dumpabort(0); 486 } 487 slp->sent = 0; 488 489 if (got != writesize) { 490 quit("EOT detected at start of the tape!\n"); 491 } 492 } 493 } 494 495 /* 496 * We implement taking and restoring checkpoints on the tape level. 497 * When each tape is opened, a new process is created by forking; this 498 * saves all of the necessary context in the parent. The child 499 * continues the dump; the parent waits around, saving the context. 500 * If the child returns X_REWRITE, then it had problems writing that tape; 501 * this causes the parent to fork again, duplicating the context, and 502 * everything continues as if nothing had happened. 503 */ 504 void 505 startnewtape(int top) 506 { 507 int parentpid; 508 int childpid; 509 int status; 510 char *p; 511 sig_t interrupt_save; 512 513 interrupt_save = signal(SIGINT, SIG_IGN); 514 parentpid = getpid(); 515 516 restore_check_point: 517 (void)signal(SIGINT, interrupt_save); 518 /* 519 * All signals are inherited... 520 */ 521 setproctitle(NULL); /* Restore the proctitle. */ 522 childpid = fork(); 523 if (childpid < 0) { 524 msg("Context save fork fails in parent %d\n", parentpid); 525 Exit(X_ABORT); 526 } 527 if (childpid != 0) { 528 /* 529 * PARENT: 530 * save the context by waiting 531 * until the child doing all of the work returns. 532 * don't catch the interrupt 533 */ 534 signal(SIGINT, SIG_IGN); 535 #ifdef TDEBUG 536 msg("Tape: %d; parent process: %d child process %d\n", 537 tapeno+1, parentpid, childpid); 538 #endif /* TDEBUG */ 539 if (waitpid(childpid, &status, 0) == -1) 540 msg("Waiting for child %d: %s\n", childpid, 541 strerror(errno)); 542 if (status & 0xFF) { 543 msg("Child %d returns LOB status %o\n", 544 childpid, status&0xFF); 545 } 546 status = (status >> 8) & 0xFF; 547 #ifdef TDEBUG 548 switch(status) { 549 case X_FINOK: 550 msg("Child %d finishes X_FINOK\n", childpid); 551 break; 552 case X_ABORT: 553 msg("Child %d finishes X_ABORT\n", childpid); 554 break; 555 case X_REWRITE: 556 msg("Child %d finishes X_REWRITE\n", childpid); 557 break; 558 default: 559 msg("Child %d finishes unknown %d\n", 560 childpid, status); 561 break; 562 } 563 #endif /* TDEBUG */ 564 switch(status) { 565 case X_FINOK: 566 Exit(X_FINOK); 567 case X_ABORT: 568 Exit(X_ABORT); 569 case X_REWRITE: 570 goto restore_check_point; 571 default: 572 msg("Bad return code from dump: %d\n", status); 573 Exit(X_ABORT); 574 } 575 /*NOTREACHED*/ 576 } else { /* we are the child; just continue */ 577 #ifdef TDEBUG 578 sleep(4); /* allow time for parent's message to get out */ 579 msg("Child on Tape %d has parent %d, my pid = %d\n", 580 tapeno+1, parentpid, getpid()); 581 #endif /* TDEBUG */ 582 /* 583 * If we have a name like "/dev/rmt0,/dev/rmt1", 584 * use the name before the comma first, and save 585 * the remaining names for subsequent volumes. 586 */ 587 tapeno++; /* current tape sequence */ 588 if (nexttape || strchr(tape, ',')) { 589 if (nexttape && *nexttape) 590 tape = nexttape; 591 if ((p = strchr(tape, ',')) != NULL) { 592 *p = '\0'; 593 nexttape = p + 1; 594 } else 595 nexttape = NULL; 596 msg("Dumping volume %d on %s\n", tapeno, tape); 597 } 598 if (pipeout) { 599 tapefd = STDOUT_FILENO; 600 } else if (popenout) { 601 char volno[sizeof("2147483647")]; 602 603 (void)sprintf(volno, "%d", spcl.c_volume + 1); 604 if (setenv("DUMP_VOLUME", volno, 1) == -1) { 605 msg("Cannot set $DUMP_VOLUME.\n"); 606 dumpabort(0); 607 } 608 popenfp = popen(popenout, "w"); 609 if (popenfp == NULL) { 610 msg("Cannot open output pipeline \"%s\".\n", 611 popenout); 612 dumpabort(0); 613 } 614 tapefd = fileno(popenfp); 615 } else { 616 #ifdef RDUMP 617 while ((tapefd = (host ? rmtopen(tape, 2) : 618 open(tape, O_WRONLY|O_CREAT, 0666))) < 0) 619 #else 620 while ((tapefd = 621 open(tape, O_WRONLY|O_CREAT, 0666)) < 0) 622 #endif 623 { 624 msg("Cannot open output \"%s\".\n", tape); 625 if (!query("Do you want to retry the open?")) 626 dumpabort(0); 627 } 628 } 629 630 enslave(); /* Share open tape file descriptor with slaves */ 631 if (popenout) 632 close(tapefd); /* Give up our copy of it. */ 633 signal(SIGINFO, infosch); 634 635 asize = 0; 636 blocksthisvol = 0; 637 if (top) 638 newtape++; /* new tape signal */ 639 spcl.c_count = slp->count; 640 /* 641 * measure firstrec in TP_BSIZE units since restore doesn't 642 * know the correct ntrec value... 643 */ 644 spcl.c_firstrec = slp->firstrec; 645 spcl.c_volume++; 646 spcl.c_type = TS_TAPE; 647 writeheader((ino_t)slp->inode); 648 if (tapeno > 1) 649 msg("Volume %d begins with blocks from inode %d\n", 650 tapeno, slp->inode); 651 } 652 } 653 654 void 655 dumpabort(int signo __unused) 656 { 657 658 if (master != 0 && master != getpid()) 659 /* Signals master to call dumpabort */ 660 (void) kill(master, SIGTERM); 661 else { 662 killall(); 663 msg("The ENTIRE dump is aborted.\n"); 664 } 665 #ifdef RDUMP 666 rmtclose(); 667 #endif 668 Exit(X_ABORT); 669 } 670 671 void 672 Exit(status) 673 int status; 674 { 675 676 #ifdef TDEBUG 677 msg("pid = %d exits with status %d\n", getpid(), status); 678 #endif /* TDEBUG */ 679 exit(status); 680 } 681 682 /* 683 * proceed - handler for SIGUSR2, used to synchronize IO between the slaves. 684 */ 685 void 686 proceed(int signo __unused) 687 { 688 689 if (ready) 690 longjmp(jmpbuf, 1); 691 caught++; 692 } 693 694 void 695 enslave(void) 696 { 697 int cmd[2]; 698 int i, j; 699 700 master = getpid(); 701 702 signal(SIGTERM, dumpabort); /* Slave sends SIGTERM on dumpabort() */ 703 signal(SIGPIPE, sigpipe); 704 signal(SIGUSR1, tperror); /* Slave sends SIGUSR1 on tape errors */ 705 signal(SIGUSR2, proceed); /* Slave sends SIGUSR2 to next slave */ 706 707 for (i = 0; i < SLAVES; i++) { 708 if (i == slp - &slaves[0]) { 709 caught = 1; 710 } else { 711 caught = 0; 712 } 713 714 if (socketpair(AF_UNIX, SOCK_STREAM, 0, cmd) < 0 || 715 (slaves[i].pid = fork()) < 0) 716 quit("too many slaves, %d (recompile smaller): %s\n", 717 i, strerror(errno)); 718 719 slaves[i].fd = cmd[1]; 720 slaves[i].sent = 0; 721 if (slaves[i].pid == 0) { /* Slave starts up here */ 722 for (j = 0; j <= i; j++) 723 (void) close(slaves[j].fd); 724 signal(SIGINT, SIG_IGN); /* Master handles this */ 725 doslave(cmd[0], i); 726 Exit(X_FINOK); 727 } 728 } 729 730 for (i = 0; i < SLAVES; i++) 731 (void) atomic(write, slaves[i].fd, 732 (char *) &slaves[(i + 1) % SLAVES].pid, 733 sizeof slaves[0].pid); 734 735 master = 0; 736 } 737 738 void 739 killall(void) 740 { 741 int i; 742 743 for (i = 0; i < SLAVES; i++) 744 if (slaves[i].pid > 0) { 745 (void) kill(slaves[i].pid, SIGKILL); 746 slaves[i].sent = 0; 747 } 748 } 749 750 /* 751 * Synchronization - each process has a lockfile, and shares file 752 * descriptors to the following process's lockfile. When our write 753 * completes, we release our lock on the following process's lock- 754 * file, allowing the following process to lock it and proceed. We 755 * get the lock back for the next cycle by swapping descriptors. 756 */ 757 static void 758 doslave(int cmd, int slave_number) 759 { 760 int nread; 761 int nextslave, size, wrote, eot_count; 762 763 /* 764 * Need our own seek pointer. 765 */ 766 (void) close(diskfd); 767 if ((diskfd = open(disk, O_RDONLY)) < 0) 768 quit("slave couldn't reopen disk: %s\n", strerror(errno)); 769 770 /* 771 * Need the pid of the next slave in the loop... 772 */ 773 if ((nread = atomic(read, cmd, (char *)&nextslave, sizeof nextslave)) 774 != sizeof nextslave) { 775 quit("master/slave protocol botched - didn't get pid of next slave.\n"); 776 } 777 778 /* 779 * Get list of blocks to dump, read the blocks into tape buffer 780 */ 781 while ((nread = atomic(read, cmd, (char *)slp->req, reqsiz)) == reqsiz) { 782 struct req *p = slp->req; 783 784 for (trecno = 0; trecno < ntrec; 785 trecno += p->count, p += p->count) { 786 if (p->dblk) { 787 bread(p->dblk, slp->tblock[trecno], 788 p->count * TP_BSIZE); 789 } else { 790 if (p->count != 1 || atomic(read, cmd, 791 (char *)slp->tblock[trecno], 792 TP_BSIZE) != TP_BSIZE) 793 quit("master/slave protocol botched.\n"); 794 } 795 } 796 if (setjmp(jmpbuf) == 0) { 797 ready = 1; 798 if (!caught) 799 (void) pause(); 800 } 801 ready = 0; 802 caught = 0; 803 804 /* Try to write the data... */ 805 eot_count = 0; 806 size = 0; 807 808 wrote = 0; 809 while (eot_count < 10 && size < writesize) { 810 #ifdef RDUMP 811 if (host) 812 wrote = rmtwrite(slp->tblock[0]+size, 813 writesize-size); 814 else 815 #endif 816 wrote = write(tapefd, slp->tblock[0]+size, 817 writesize-size); 818 #ifdef WRITEDEBUG 819 printf("slave %d wrote %d\n", slave_number, wrote); 820 #endif 821 if (wrote < 0) 822 break; 823 if (wrote == 0) 824 eot_count++; 825 size += wrote; 826 } 827 828 #ifdef WRITEDEBUG 829 if (size != writesize) 830 printf("slave %d only wrote %d out of %d bytes and gave up.\n", 831 slave_number, size, writesize); 832 #endif 833 834 /* 835 * Handle ENOSPC as an EOT condition. 836 */ 837 if (wrote < 0 && errno == ENOSPC) { 838 wrote = 0; 839 eot_count++; 840 } 841 842 if (eot_count > 0) 843 size = 0; 844 845 if (wrote < 0) { 846 (void) kill(master, SIGUSR1); 847 for (;;) 848 (void) sigpause(0); 849 } else { 850 /* 851 * pass size of write back to master 852 * (for EOT handling) 853 */ 854 (void) atomic(write, cmd, (char *)&size, sizeof size); 855 } 856 857 /* 858 * If partial write, don't want next slave to go. 859 * Also jolts him awake. 860 */ 861 (void) kill(nextslave, SIGUSR2); 862 } 863 if (nread != 0) 864 quit("error reading command pipe: %s\n", strerror(errno)); 865 } 866 867 /* 868 * Since a read from a pipe may not return all we asked for, 869 * or a write may not write all we ask if we get a signal, 870 * loop until the count is satisfied (or error). 871 */ 872 static int 873 atomic(ssize_t (*func)(), int fd, char *buf, int count) 874 { 875 int got, need = count; 876 877 while ((got = (*func)(fd, buf, need)) > 0 && (need -= got) > 0) 878 buf += got; 879 return (got < 0 ? got : count - need); 880 } 881