1 /*- 2 * SPDX-License-Identifier: BSD-3-Clause 3 * 4 * Copyright (c) 1980, 1991, 1993 5 * The Regents of the University of California. All rights reserved. 6 * 7 * Redistribution and use in source and binary forms, with or without 8 * modification, are permitted provided that the following conditions 9 * are met: 10 * 1. Redistributions of source code must retain the above copyright 11 * notice, this list of conditions and the following disclaimer. 12 * 2. Redistributions in binary form must reproduce the above copyright 13 * notice, this list of conditions and the following disclaimer in the 14 * documentation and/or other materials provided with the distribution. 15 * 3. Neither the name of the University nor the names of its contributors 16 * may be used to endorse or promote products derived from this software 17 * without specific prior written permission. 18 * 19 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND 20 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 21 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 22 * ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE 23 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 24 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 25 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 26 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 27 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 28 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 29 * SUCH DAMAGE. 30 */ 31 32 #include <sys/param.h> 33 #include <sys/socket.h> 34 #include <sys/wait.h> 35 #include <sys/stat.h> 36 37 #include <ufs/ufs/dinode.h> 38 #include <ufs/ffs/fs.h> 39 40 #include <protocols/dumprestore.h> 41 42 #include <assert.h> 43 #include <errno.h> 44 #include <fcntl.h> 45 #include <limits.h> 46 #include <setjmp.h> 47 #include <signal.h> 48 #include <stdio.h> 49 #include <stdlib.h> 50 #include <string.h> 51 #include <time.h> 52 #include <unistd.h> 53 54 #include "dump.h" 55 56 ino_t curino; /* current inumber; used globally */ 57 int newtape; /* new tape flag */ 58 union u_spcl u_spcl; /* mapping of variables in a control block */ 59 60 static int tapefd; /* tape file descriptor */ 61 static long asize; /* number of 0.1" units written on cur tape */ 62 static int writesize; /* size of malloc()ed buffer for tape */ 63 static int64_t lastspclrec = -1; /* tape block number of last written header */ 64 static int trecno = 0; /* next record to write in current block */ 65 static long blocksthisvol; /* number of blocks on current output file */ 66 static char *nexttape; 67 static FILE *popenfp = NULL; 68 69 static int atomic_read(int, void *, int); 70 static int atomic_write(int, const void *, int); 71 static void worker(int, int); 72 static void create_workers(void); 73 static void flushtape(void); 74 static void killall(void); 75 static void rollforward(void); 76 77 /* 78 * Concurrent dump mods (Caltech) - disk block reading and tape writing 79 * are exported to several worker processes. While one worker writes the 80 * tape, the others read disk blocks; they pass control of the tape in 81 * a ring via signals. The parent process traverses the file system and 82 * sends writeheader()'s and lists of daddr's to the workers via pipes. 83 * The following structure defines the instruction packets sent to workers. 84 */ 85 struct req { 86 ufs2_daddr_t dblk; 87 int count; 88 }; 89 static int reqsiz; 90 91 #define WORKERS 3 /* 1 worker writing, 1 reading, 1 for slack */ 92 static struct worker { 93 int64_t tapea; /* header number at start of this chunk */ 94 int64_t firstrec; /* record number of this block */ 95 int count; /* count to next header (used for TS_TAPE */ 96 /* after EOT) */ 97 int inode; /* inode that we are currently dealing with */ 98 int fd; /* FD for this worker */ 99 int pid; /* PID for this worker */ 100 int sent; /* 1 == we've sent this worker requests */ 101 char (*tblock)[TP_BSIZE]; /* buffer for data blocks */ 102 struct req *req; /* buffer for requests */ 103 } workers[WORKERS+1]; 104 static struct worker *wp; 105 106 static char (*nextblock)[TP_BSIZE]; 107 108 static int master; /* pid of master, for sending error signals */ 109 static int tenths; /* length of tape used per block written */ 110 static volatile sig_atomic_t caught; /* have we caught the signal to proceed? */ 111 static volatile sig_atomic_t ready; /* reached the lock point without having */ 112 /* received the SIGUSR2 signal from the prev worker? */ 113 static jmp_buf jmpbuf; /* where to jump to if we are ready when the */ 114 /* SIGUSR2 arrives from the previous worker */ 115 116 int 117 alloctape(void) 118 { 119 int pgoff = getpagesize() - 1; 120 char *buf; 121 int i; 122 123 writesize = ntrec * TP_BSIZE; 124 reqsiz = (ntrec + 1) * sizeof(struct req); 125 /* 126 * CDC 92181's and 92185's make 0.8" gaps in 1600-bpi start/stop mode 127 * (see DEC TU80 User's Guide). The shorter gaps of 6250-bpi require 128 * repositioning after stopping, i.e, streaming mode, where the gap is 129 * variable, 0.30" to 0.45". The gap is maximal when the tape stops. 130 */ 131 if (blocksperfile == 0 && !unlimited) 132 tenths = writesize / density + 133 (cartridge ? 16 : density == 625 ? 5 : 8); 134 /* 135 * Allocate tape buffer contiguous with the array of instruction 136 * packets, so flushtape() can write them together with one write(). 137 * Align tape buffer on page boundary to speed up tape write(). 138 */ 139 for (i = 0; i <= WORKERS; i++) { 140 buf = (char *) 141 malloc((unsigned)(reqsiz + writesize + pgoff + TP_BSIZE)); 142 if (buf == NULL) 143 return(0); 144 workers[i].tblock = (char (*)[TP_BSIZE]) 145 (((long)&buf[ntrec + 1] + pgoff) &~ pgoff); 146 workers[i].req = (struct req *)workers[i].tblock - ntrec - 1; 147 } 148 wp = &workers[0]; 149 wp->count = 1; 150 wp->tapea = 0; 151 wp->firstrec = 0; 152 nextblock = wp->tblock; 153 return(1); 154 } 155 156 void 157 writerec(char *dp, int isspcl) 158 { 159 160 wp->req[trecno].dblk = (ufs2_daddr_t)0; 161 wp->req[trecno].count = 1; 162 /* Can't do a structure assignment due to alignment problems */ 163 bcopy(dp, *(nextblock)++, sizeof (union u_spcl)); 164 if (isspcl) 165 lastspclrec = spcl.c_tapea; 166 trecno++; 167 spcl.c_tapea++; 168 if (trecno >= ntrec) 169 flushtape(); 170 } 171 172 void 173 dumpblock(ufs2_daddr_t blkno, int size) 174 { 175 int avail, tpblks; 176 ufs2_daddr_t dblkno; 177 178 dblkno = fsbtodb(sblock, blkno); 179 tpblks = size >> tp_bshift; 180 while ((avail = MIN(tpblks, ntrec - trecno)) > 0) { 181 wp->req[trecno].dblk = dblkno; 182 wp->req[trecno].count = avail; 183 trecno += avail; 184 spcl.c_tapea += avail; 185 if (trecno >= ntrec) 186 flushtape(); 187 dblkno += avail << (tp_bshift - dev_bshift); 188 tpblks -= avail; 189 } 190 } 191 192 int nogripe = 0; 193 194 void 195 tperror(int signo __unused) 196 { 197 198 if (pipeout) { 199 msg("write error on %s\n", tape); 200 quit("Cannot recover\n"); 201 /* NOTREACHED */ 202 } 203 msg("write error %ld blocks into volume %d\n", blocksthisvol, tapeno); 204 broadcast("DUMP WRITE ERROR!\n"); 205 if (!query("Do you want to restart?")) 206 dumpabort(0); 207 msg("Closing this volume. Prepare to restart with new media;\n"); 208 msg("this dump volume will be rewritten.\n"); 209 killall(); 210 nogripe = 1; 211 close_rewind(); 212 Exit(X_REWRITE); 213 } 214 215 void 216 sigpipe(int signo __unused) 217 { 218 219 quit("Broken pipe\n"); 220 } 221 222 static void 223 flushtape(void) 224 { 225 int i, blks, got; 226 int64_t lastfirstrec; 227 228 int siz = (char *)nextblock - (char *)wp->req; 229 230 wp->req[trecno].count = 0; /* Sentinel */ 231 232 if (atomic_write(wp->fd, (const void *)wp->req, siz) != siz) 233 quit("error writing command pipe: %s\n", strerror(errno)); 234 wp->sent = 1; /* we sent a request, read the response later */ 235 236 lastfirstrec = wp->firstrec; 237 238 if (++wp >= &workers[WORKERS]) 239 wp = &workers[0]; 240 241 /* Read results back from next worker */ 242 if (wp->sent) { 243 if (atomic_read(wp->fd, (void *)&got, sizeof got) 244 != sizeof got) { 245 perror(" DUMP: error reading command pipe in master"); 246 dumpabort(0); 247 } 248 wp->sent = 0; 249 250 /* Check for end of tape */ 251 if (got < writesize) { 252 msg("End of tape detected\n"); 253 254 /* 255 * Drain the results, don't care what the values were. 256 * If we read them here then trewind won't... 257 */ 258 for (i = 0; i < WORKERS; i++) { 259 if (workers[i].sent) { 260 if (atomic_read(workers[i].fd, 261 (void *)&got, sizeof got) 262 != sizeof got) { 263 perror(" DUMP: error reading command pipe in master"); 264 dumpabort(0); 265 } 266 workers[i].sent = 0; 267 } 268 } 269 270 close_rewind(); 271 rollforward(); 272 return; 273 } 274 } 275 276 blks = 0; 277 if (spcl.c_type != TS_END && spcl.c_type != TS_CLRI && 278 spcl.c_type != TS_BITS) { 279 assert(spcl.c_count <= TP_NINDIR); 280 for (i = 0; i < spcl.c_count; i++) 281 if (spcl.c_addr[i] != 0) 282 blks++; 283 } 284 wp->count = lastspclrec + blks + 1 - spcl.c_tapea; 285 wp->tapea = spcl.c_tapea; 286 wp->firstrec = lastfirstrec + ntrec; 287 wp->inode = curino; 288 nextblock = wp->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 < WORKERS; 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 (workers[f].sent) { 318 if (atomic_read(workers[f].fd, (void *)&got, sizeof got) 319 != sizeof got) { 320 perror(" DUMP: error reading command pipe in master"); 321 dumpabort(0); 322 } 323 workers[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(workers[f].fd); 331 } 332 while (wait((int *)NULL) >= 0) /* wait for any signals from workers */ 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 worker *twp; 393 int i, size, got; 394 int64_t savedtapea; 395 union u_spcl *ntb, *otb; 396 twp = &workers[WORKERS]; 397 ntb = (union u_spcl *)twp->tblock[1]; 398 399 /* 400 * Each of the N workers should have requests that need to 401 * be replayed on the next tape. Use the extra worker buffers 402 * (workers[WORKERS]) to construct request lists to be sent to 403 * each worker in turn. 404 */ 405 for (i = 0; i < WORKERS; i++) { 406 q = &twp->req[1]; 407 otb = (union u_spcl *)wp->tblock; 408 409 /* 410 * For each request in the current worker, copy it to twp. 411 */ 412 413 prev = NULL; 414 for (p = wp->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 = &twp->req[0]; 430 if (i == 0) { 431 q->dblk = 0; 432 q->count = 1; 433 trecno = 0; 434 nextblock = twp->tblock; 435 savedtapea = spcl.c_tapea; 436 spcl.c_tapea = wp->tapea; 437 startnewtape(0); 438 spcl.c_tapea = savedtapea; 439 lastspclrec = savedtapea - 1; 440 } 441 size = (char *)ntb - (char *)q; 442 if (atomic_write(wp->fd, (const void *)q, size) != size) { 443 perror(" DUMP: error writing command pipe"); 444 dumpabort(0); 445 } 446 wp->sent = 1; 447 if (++wp >= &workers[WORKERS]) 448 wp = &workers[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 *)twp->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 *)twp->tblock) = *ntb; 468 ntb = (union u_spcl *)twp->tblock[1]; 469 } 470 } 471 wp->req[0] = *q; 472 nextblock = wp->tblock; 473 if (q->dblk == 0) 474 nextblock++; 475 trecno = 1; 476 477 /* 478 * Clear the first workers' response. One hopes that it 479 * worked ok, otherwise the tape is much too short! 480 */ 481 if (wp->sent) { 482 if (atomic_read(wp->fd, (void *)&got, sizeof got) 483 != sizeof got) { 484 perror(" DUMP: error reading command pipe in master"); 485 dumpabort(0); 486 } 487 wp->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 create_workers(); /* Share open tape file descriptor with workers */ 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 = wp->count; 640 /* 641 * measure firstrec in TP_BSIZE units since restore doesn't 642 * know the correct ntrec value... 643 */ 644 spcl.c_firstrec = wp->firstrec; 645 spcl.c_volume++; 646 spcl.c_type = TS_TAPE; 647 writeheader((ino_t)wp->inode); 648 if (tapeno > 1) 649 msg("Volume %d begins with blocks from inode %d\n", 650 tapeno, wp->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(int status) 673 { 674 675 #ifdef TDEBUG 676 msg("pid = %d exits with status %d\n", getpid(), status); 677 #endif /* TDEBUG */ 678 exit(status); 679 } 680 681 /* 682 * proceed - handler for SIGUSR2, used to synchronize IO between the workers. 683 */ 684 void 685 proceed(int signo __unused) 686 { 687 688 if (ready) 689 longjmp(jmpbuf, 1); 690 caught++; 691 } 692 693 void 694 create_workers(void) 695 { 696 int cmd[2]; 697 int i, j; 698 699 master = getpid(); 700 701 signal(SIGTERM, dumpabort); /* Worker sends SIGTERM on dumpabort() */ 702 signal(SIGPIPE, sigpipe); 703 signal(SIGUSR1, tperror); /* Worker sends SIGUSR1 on tape errors */ 704 signal(SIGUSR2, proceed); /* Worker sends SIGUSR2 to next worker */ 705 706 for (i = 0; i < WORKERS; i++) { 707 if (i == wp - &workers[0]) { 708 caught = 1; 709 } else { 710 caught = 0; 711 } 712 713 if (socketpair(AF_UNIX, SOCK_STREAM, 0, cmd) < 0 || 714 (workers[i].pid = fork()) < 0) 715 quit("too many workers, %d (recompile smaller): %s\n", 716 i, strerror(errno)); 717 718 workers[i].fd = cmd[1]; 719 workers[i].sent = 0; 720 if (workers[i].pid == 0) { /* Worker starts up here */ 721 for (j = 0; j <= i; j++) 722 (void) close(workers[j].fd); 723 signal(SIGINT, SIG_IGN); /* Master handles this */ 724 worker(cmd[0], i); 725 Exit(X_FINOK); 726 } 727 } 728 729 for (i = 0; i < WORKERS; i++) 730 (void) atomic_write(workers[i].fd, 731 (const void *) &workers[(i + 1) % WORKERS].pid, 732 sizeof workers[0].pid); 733 734 master = 0; 735 } 736 737 void 738 killall(void) 739 { 740 int i; 741 742 for (i = 0; i < WORKERS; i++) 743 if (workers[i].pid > 0) { 744 (void) kill(workers[i].pid, SIGKILL); 745 workers[i].sent = 0; 746 } 747 } 748 749 /* 750 * Synchronization - each process has a lockfile, and shares file 751 * descriptors to the following process's lockfile. When our write 752 * completes, we release our lock on the following process's lock- 753 * file, allowing the following process to lock it and proceed. We 754 * get the lock back for the next cycle by swapping descriptors. 755 */ 756 static void 757 worker(int cmd, int worker_number) 758 { 759 int nread; 760 int nextworker, size, wrote, eot_count; 761 762 /* 763 * Need our own seek pointer. 764 */ 765 (void) close(diskfd); 766 if ((diskfd = open(disk, O_RDONLY)) < 0) 767 quit("worker couldn't reopen disk: %s\n", strerror(errno)); 768 769 /* 770 * Need the pid of the next worker in the loop... 771 */ 772 if ((nread = atomic_read(cmd, (void *)&nextworker, sizeof nextworker)) 773 != sizeof nextworker) { 774 quit("master/worker protocol botched - didn't get pid of next worker.\n"); 775 } 776 777 /* 778 * Get list of blocks to dump, read the blocks into tape buffer 779 */ 780 while ((nread = atomic_read(cmd, (void *)wp->req, reqsiz)) == reqsiz) { 781 struct req *p = wp->req; 782 783 for (trecno = 0; trecno < ntrec; 784 trecno += p->count, p += p->count) { 785 if (p->dblk) { 786 blkread(p->dblk, wp->tblock[trecno], 787 p->count * TP_BSIZE); 788 } else { 789 if (p->count != 1 || atomic_read(cmd, 790 (void *)wp->tblock[trecno], 791 TP_BSIZE) != TP_BSIZE) 792 quit("master/worker protocol botched.\n"); 793 } 794 } 795 if (setjmp(jmpbuf) == 0) { 796 ready = 1; 797 if (!caught) 798 (void) pause(); 799 } 800 ready = 0; 801 caught = 0; 802 803 /* Try to write the data... */ 804 eot_count = 0; 805 size = 0; 806 807 wrote = 0; 808 while (eot_count < 10 && size < writesize) { 809 #ifdef RDUMP 810 if (host) 811 wrote = rmtwrite(wp->tblock[0]+size, 812 writesize-size); 813 else 814 #endif 815 wrote = write(tapefd, wp->tblock[0]+size, 816 writesize-size); 817 #ifdef WRITEDEBUG 818 printf("worker %d wrote %d\n", worker_number, wrote); 819 #endif 820 if (wrote < 0) 821 break; 822 if (wrote == 0) 823 eot_count++; 824 size += wrote; 825 } 826 827 #ifdef WRITEDEBUG 828 if (size != writesize) 829 printf("worker %d only wrote %d out of %d bytes and gave up.\n", 830 worker_number, size, writesize); 831 #endif 832 833 /* 834 * Handle ENOSPC as an EOT condition. 835 */ 836 if (wrote < 0 && errno == ENOSPC) { 837 wrote = 0; 838 eot_count++; 839 } 840 841 if (eot_count > 0) 842 size = 0; 843 844 if (wrote < 0) { 845 (void) kill(master, SIGUSR1); 846 for (;;) 847 (void) sigpause(0); 848 } else { 849 /* 850 * pass size of write back to master 851 * (for EOT handling) 852 */ 853 (void)atomic_write(cmd, (const void *)&size, 854 sizeof size); 855 } 856 857 /* 858 * If partial write, don't want next worker to go. 859 * Also jolts him awake. 860 */ 861 (void) kill(nextworker, 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 * loop until the count is satisfied (or error). 870 */ 871 static int 872 atomic_read(int fd, void *buf, int count) 873 { 874 int got, need = count; 875 876 while ((got = read(fd, buf, need)) > 0 && (need -= got) > 0) 877 buf += got; 878 return (got < 0 ? got : count - need); 879 } 880 881 /* 882 * Since a write to a pipe may not write all we ask if we get a signal, 883 * loop until the count is satisfied (or error). 884 */ 885 static int 886 atomic_write(int fd, const void *buf, int count) 887 { 888 int got, need = count; 889 890 while ((got = write(fd, buf, need)) > 0 && (need -= got) > 0) 891 buf += got; 892 return (got < 0 ? got : count - need); 893 } 894