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