57#include <unistd.h>
58
59#include "dump.h"
60
61int writesize; /* size of malloc()ed buffer for tape */
62int64_t lastspclrec = -1; /* tape block number of last written header */
63int trecno = 0; /* next record to write in current block */
64extern long blocksperfile; /* number of blocks per output file */
65long blocksthisvol; /* number of blocks on current output file */
66extern int ntrec; /* blocking factor on tape */
67extern int cartridge;
68extern char *host;
69char *nexttape;
70FILE *popenfp = NULL;
71
72static int atomic(ssize_t (*)(), int, char *, int);
73static void doslave(int, int);
74static void enslave(void);
75static void flushtape(void);
76static void killall(void);
77static 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 */
87struct req {
88 ufs2_daddr_t dblk;
89 int count;
90};
91int reqsiz;
92
93#define SLAVES 3 /* 1 slave writing, 1 reading, 1 for slack */
94struct 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];
106struct slave *slp;
107
108char (*nextblock)[TP_BSIZE];
109
110int master; /* pid of master, for sending error signals */
111int tenths; /* length of tape used per block written */
112static volatile sig_atomic_t caught; /* have we caught the signal to proceed? */
113static volatile sig_atomic_t ready; /* reached the lock point without having */
114 /* received the SIGUSR2 signal from the prev slave? */
115static jmp_buf jmpbuf; /* where to jump to if we are ready when the */
116 /* SIGUSR2 arrives from the previous slave */
117
118int
119alloctape(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
158void
159writerec(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
174void
175dumpblock(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
194int nogripe = 0;
195
196void
197tperror(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
217void
218sigpipe(int signo __unused)
219{
220
221 quit("Broken pipe\n");
222}
223
224static void
225flushtape(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
301void
302trewind(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
365void
366close_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
388void
389rollforward(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 */
504void
505startnewtape(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
516restore_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
654void
655dumpabort(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
671void
672Exit(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 */
685void
686proceed(int signo __unused)
687{
688
689 if (ready)
690 longjmp(jmpbuf, 1);
691 caught++;
692}
693
694void
695enslave(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
738void
739killall(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 */
757static void
758doslave(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 */
872static int
873atomic(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}
| 57#include <unistd.h>
58
59#include "dump.h"
60
61int writesize; /* size of malloc()ed buffer for tape */
62int64_t lastspclrec = -1; /* tape block number of last written header */
63int trecno = 0; /* next record to write in current block */
64extern long blocksperfile; /* number of blocks per output file */
65long blocksthisvol; /* number of blocks on current output file */
66extern int ntrec; /* blocking factor on tape */
67extern int cartridge;
68extern char *host;
69char *nexttape;
70FILE *popenfp = NULL;
71
72static int atomic(ssize_t (*)(), int, char *, int);
73static void doslave(int, int);
74static void enslave(void);
75static void flushtape(void);
76static void killall(void);
77static 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 */
87struct req {
88 ufs2_daddr_t dblk;
89 int count;
90};
91int reqsiz;
92
93#define SLAVES 3 /* 1 slave writing, 1 reading, 1 for slack */
94struct 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];
106struct slave *slp;
107
108char (*nextblock)[TP_BSIZE];
109
110int master; /* pid of master, for sending error signals */
111int tenths; /* length of tape used per block written */
112static volatile sig_atomic_t caught; /* have we caught the signal to proceed? */
113static volatile sig_atomic_t ready; /* reached the lock point without having */
114 /* received the SIGUSR2 signal from the prev slave? */
115static jmp_buf jmpbuf; /* where to jump to if we are ready when the */
116 /* SIGUSR2 arrives from the previous slave */
117
118int
119alloctape(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
158void
159writerec(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
174void
175dumpblock(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
194int nogripe = 0;
195
196void
197tperror(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
217void
218sigpipe(int signo __unused)
219{
220
221 quit("Broken pipe\n");
222}
223
224static void
225flushtape(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
301void
302trewind(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
365void
366close_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
388void
389rollforward(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 */
504void
505startnewtape(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
516restore_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
654void
655dumpabort(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
671void
672Exit(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 */
685void
686proceed(int signo __unused)
687{
688
689 if (ready)
690 longjmp(jmpbuf, 1);
691 caught++;
692}
693
694void
695enslave(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
738void
739killall(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 */
757static void
758doslave(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 */
872static int
873atomic(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}
|