src/rt/rc3.c

#ifndef lint
static const char RCSid[] = "$Id: rc3.c,v 2.8 2012/06/12 22:46:45 greg Exp $";
#endif
/*
 * Accumulate ray contributions for a set of materials
 * Controlling process for multiple children
 */

#include "rcontrib.h"
#include "platform.h"
#include "rtprocess.h"
#include "selcall.h"

/* Modifier contribution queue (results waiting to be output) */
typedef struct s_binq {
        RNUMBER         ndx;            /* index for this entry */
        RNUMBER         nadded;         /* accumulated so far */
        struct s_binq   *next;          /* next in queue */
        MODCONT         *mca[1];        /* contrib. array (extends struct) */
} BINQ;

static BINQ     *out_bq = NULL;         /* output bin queue */
static BINQ     *free_bq = NULL;        /* free queue entries */

static struct {
        RNUMBER r1;                     /* assigned ray starting index */
        SUBPROC pr;                     /* PID, i/o descriptors */
        FILE    *infp;                  /* file pointer to read from process */
        int     nr;                     /* number rays to sum (0 if free) */
} kida[MAXPROCESS];             /* our child processes */


/* Get new bin queue entry */
static BINQ *
new_binq()
{
        BINQ    *bp;
        int     i;

        if (free_bq != NULL) {          /* something already available? */
                bp = free_bq;
                free_bq = bp->next;
                bp->next = NULL;
                bp->nadded = 0;
                return(bp);
        }
                                        /* else allocate fresh */
        bp = (BINQ *)malloc(sizeof(BINQ) + sizeof(MODCONT *)*(nmods-1));
        if (bp == NULL)
                goto memerr;
        for (i = nmods; i--; ) {
                MODCONT *mp = (MODCONT *)lu_find(&modconttab,modname[i])->data;
                bp->mca[i] = (MODCONT *)malloc(sizeof(MODCONT) +
                                                sizeof(DCOLOR)*(mp->nbins-1));
                if (bp->mca[i] == NULL)
                        goto memerr;
                memcpy(bp->mca[i], mp, sizeof(MODCONT)-sizeof(DCOLOR));
                /* memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*mp->nbins); */
        }
        bp->ndx = 0;
        bp->nadded = 0;
        bp->next = NULL;
        return(bp);
memerr:
        error(SYSTEM, "out of memory in new_binq()");
        return(NULL);
}


/* Free a bin queue entry */
static void
free_binq(BINQ *bp)
{
        int     i;

        if (bp == NULL) {               /* signal to release our free list */
                while ((bp = free_bq) != NULL) {
                        free_bq = bp->next;
                        for (i = nmods; i--; )
                                free(bp->mca[i]);
                                /* Note: we don't own bp->mca[i]->binv */
                        free(bp);
                }
                return;
        }
                                        /* clear sums for next use */
/*      for (i = nmods; i--; )
                memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
*/
        bp->ndx = 0;
        bp->next = free_bq;             /* push onto free list */
        free_bq = bp;
}


/* Add modifier values to accumulation record in queue and clear */
void
queue_modifiers()
{
        MODCONT *mpin, *mpout;
        int     i, j;

        if ((accumulate > 0) | (out_bq == NULL))
                error(CONSISTENCY, "bad call to queue_modifiers()");

        for (i = nmods; i--; ) {
                mpin = (MODCONT *)lu_find(&modconttab,modname[i])->data;
                mpout = out_bq->mca[i];
                for (j = mpout->nbins; j--; )
                        addcolor(mpout->cbin[j], mpin->cbin[j]);
                memset(mpin->cbin, 0, sizeof(DCOLOR)*mpin->nbins);
        }
        out_bq->nadded++;
}


/* Sum one modifier record into another (updates nadded) */
static void
add_modbin(BINQ *dst, BINQ *src)
{
        int     i, j;
        
        for (i = nmods; i--; ) {
                MODCONT *mpin = src->mca[i];
                MODCONT *mpout = dst->mca[i];
                for (j = mpout->nbins; j--; )
                        addcolor(mpout->cbin[j], mpin->cbin[j]);
        }
        dst->nadded += src->nadded;
}


/* Queue values for later output */
static void
queue_output(BINQ *bp)
{
        BINQ    *b_last, *b_cur;

        if (accumulate <= 0) {          /* just accumulating? */
                if (out_bq == NULL) {
                        bp->next = NULL;
                        out_bq = bp;
                } else {
                        add_modbin(out_bq, bp);
                        free_binq(bp);
                }
                return;
        }
        b_last = NULL;                  /* else insert in output queue */
        for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
                                b_cur = b_cur->next)
                b_last = b_cur;

        if (b_last != NULL) {
                bp->next = b_cur;
                b_last->next = bp;
        } else {
                bp->next = out_bq;
                out_bq = bp;
        }
        if (accumulate == 1)            /* no accumulation? */
                return;
        b_cur = out_bq;                 /* else merge accumulation entries */
        while (b_cur->next != NULL) {
                if (b_cur->nadded >= accumulate ||
                                (b_cur->ndx-1)/accumulate !=
                                (b_cur->next->ndx-1)/accumulate) {
                        b_cur = b_cur->next;
                        continue;
                }
                add_modbin(b_cur, b_cur->next);
                b_last = b_cur->next;
                b_cur->next = b_last->next;
                b_last->next = NULL;
                free_binq(b_last);
        }
}


/* Count number of records ready for output */
static int
queue_ready()
{
        int     nready = 0;
        BINQ    *bp;

        if (accumulate <= 0)            /* just accumulating? */
                return(0);

        for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
                                bp->ndx == lastdone+nready*accumulate+1;
                                bp = bp->next)
                ++nready;

        return(nready);
}


/* Catch up with output queue by producing ready results */
static int
output_catchup(int nmax)
{
        int     nout = 0;
        BINQ    *bp;
        int     i;

        if (accumulate <= 0)            /* just accumulating? */
                return(0);
                                        /* else output ready results */
        while (out_bq != NULL && out_bq->nadded >= accumulate
                                && out_bq->ndx == lastdone+1) {
                if ((nmax > 0) & (nout >= nmax))
                        break;
                bp = out_bq;                    /* pop off first entry */
                out_bq = bp->next;
                bp->next = NULL;
                for (i = 0; i < nmods; i++)     /* output record */
                        mod_output(bp->mca[i]);
                end_record();
                free_binq(bp);                  /* free this entry */
                lastdone += accumulate;
                ++nout;
        }
        return(nout);
}


/* Put a zero record in results queue & output */
void
put_zero_record(int ndx)
{
        BINQ    *bp = new_binq();
        int     i;

        for (i = nmods; i--; )
                memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
        bp->ndx = ndx;
        bp->nadded = 1;
        queue_output(bp);
        output_catchup(0);
}


/* Get results from child process and add to queue */
static void
queue_results(int k)
{
        BINQ    *bq = new_binq();       /* get results holder */
        int     j;

        bq->ndx = kida[k].r1;
        bq->nadded = kida[k].nr;
                                        /* read from child */
        for (j = 0; j < nmods; j++)
                if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), bq->mca[j]->nbins,
                                        kida[k].infp) != bq->mca[j]->nbins)
                        error(SYSTEM, "read error from render process");
                        
        queue_output(bq);               /* put results in output queue */
        kida[k].nr = 0;                 /* mark child as available */
}


/* callback to set output spec to NULL (stdout) */
static int
set_stdout(const LUENT *le, void *p)
{
        (*(MODCONT *)le->data).outspec = NULL;
        return(0);
}


/* Start child processes if we can */
int
in_rchild()
{
        int     rval;

        while (nchild < nproc) {        /* fork until target reached */
                errno = 0;
                rval = open_process(&kida[nchild].pr, NULL);
                if (rval < 0)
                        error(SYSTEM, "open_process() call failed");
                if (rval == 0) {        /* if in child, set up & return true */
                        lu_doall(&modconttab, set_stdout, NULL);
                        lu_done(&ofiletab);
                        while (nchild--) {      /* don't share other pipes */
                                close(kida[nchild].pr.w);
                                fclose(kida[nchild].infp);
                        }
                        inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
                        outfmt = 'd';
                        header = 0;
                        yres = 0;
                        raysleft = 0;
                        if (accumulate == 1) {
                                waitflush = xres = 1;
                                account = accumulate = 1;
                        } else {        /* parent controls accumulation */
                                waitflush = xres = 0;
                                account = accumulate = 0;
                        }
                        return(1);      /* return "true" in child */
                }
                if (rval != PIPE_BUF)
                        error(CONSISTENCY, "bad value from open_process()");
                                        /* connect to child's output */
                kida[nchild].infp = fdopen(kida[nchild].pr.r, "rb");
                if (kida[nchild].infp == NULL)
                        error(SYSTEM, "out of memory in in_rchild()");
#ifdef getc_unlocked
                flockfile(kida[nchild].infp);   /* avoid mutex overhead */
#endif
                kida[nchild++].nr = 0;  /* mark as available */
        }
        return(0);                      /* return "false" in parent */
}


/* Close child processes */
void
end_children()
{
        int     status;
        
        while (nchild > 0) {
                nchild--;
                if ((status = close_process(&kida[nchild].pr)) > 0) {
                        sprintf(errmsg,
                                "rendering process returned bad status (%d)",
                                        status);
                        error(WARNING, errmsg);
                }
                fclose(kida[nchild].infp);
        }
}


/* Wait for the next available child, managing output queue simultaneously */
static int
next_child_nq(int flushing)
{
        static struct timeval   polling;
        struct timeval          *pmode;
        fd_set                  readset, errset;
        int                     i, n, nr, nqr;

        if (!flushing)                  /* see if there's one free */
                for (i = nchild; i--; )
                        if (!kida[i].nr)
                                return(i);

        nqr = queue_ready();            /* choose blocking mode or polling */
        if ((nqr > 0) & !flushing)
                pmode = &polling;
        else
                pmode = NULL;
tryagain:                               /* catch up with output? */
        if (pmode == &polling) {
                if (nqr > nchild)       /* don't get too far behind */
                        nqr -= output_catchup(nqr-nchild);
        } else if (nqr > 0)             /* clear output before blocking */
                nqr -= output_catchup(0);
                                        /* prepare select() call */
        FD_ZERO(&readset); FD_ZERO(&errset);
        n = nr = 0;
        for (i = nchild; i--; ) {
                if (kida[i].nr) {
                        FD_SET(kida[i].pr.r, &readset);
                        ++nr;
                }
                FD_SET(kida[i].pr.r, &errset);
                if (kida[i].pr.r >= n)
                        n = kida[i].pr.r + 1;
        }
        if (!nr)                        /* nothing to wait for? */
                return(-1);
        if ((nr > 1) | (pmode == &polling)) {
                errno = 0;
                nr = select(n, &readset, NULL, &errset, pmode);
                if (!nr) {
                        pmode = NULL;   /* try again, blocking this time */
                        goto tryagain;
                }
                if (nr < 0)
                        error(SYSTEM, "select() error in next_child_nq()");
        } else
                FD_ZERO(&errset);
        n = -1;                         /* read results from child(ren) */
        for (i = nchild; i--; ) {
                if (FD_ISSET(kida[i].pr.r, &errset))
                        error(USER, "rendering process died");
                if (FD_ISSET(kida[i].pr.r, &readset))
                        queue_results(n = i);
        }
        return(n);                      /* first available child */
}


/* Run parental oversight loop */
void
parental_loop()
{
#define MAXIQ           (int)(PIPE_BUF/(sizeof(FVECT)*2))
        static int      ignore_warning_given = 0;
        int             qlimit = (accumulate == 1) ? 1 : MAXIQ-1;
        int             ninq = 0;
        FVECT           orgdir[2*MAXIQ];
        int             i, n;
                                        /* load rays from stdin & process */
#ifdef getc_unlocked
        flockfile(stdin);               /* avoid lock/unlock overhead */
#endif
        while (getvec(orgdir[2*ninq]) == 0 && getvec(orgdir[2*ninq+1]) == 0) {
                if (orgdir[2*ninq+1][0] == 0.0 &&       /* asking for flush? */
                                (orgdir[2*ninq+1][1] == 0.0) &
                                (orgdir[2*ninq+1][2] == 0.0)) {
                        if (accumulate != 1) {
                                if (!ignore_warning_given++)
                                        error(WARNING,
                                "dummy ray(s) ignored during accumulation\n");
                                continue;
                        }
                        while (next_child_nq(1) >= 0)
                                ;                       /* clear the queue */
                        lastdone = lastray = 0;
                        if ((yres <= 0) | (xres <= 0))
                                waitflush = 1;          /* flush next */
                        put_zero_record(++lastray);
                } else if (++ninq >= qlimit || accumulate > 1 &&
                            lastray/accumulate != (lastray+ninq)/accumulate) {
                        i = next_child_nq(0);           /* manages output */
                        n = ninq;
                        if (accumulate != 1)            /* request flush? */
                                memset(orgdir[2*n++], 0, sizeof(FVECT)*2);
                        n *= sizeof(FVECT)*2;           /* send assignment */
                        if (writebuf(kida[i].pr.w, (char *)orgdir, n) != n)
                                error(SYSTEM, "pipe write error");
                        kida[i].r1 = lastray+1;
                        lastray += kida[i].nr = ninq;   /* mark as busy */
                        ninq = 0;
                        if (lastray < lastdone) {       /* RNUMBER wrapped? */
                                while (next_child_nq(1) >= 0)
                                        ;
                                lastdone = lastray = 0;
                        }
                }
                if (raysleft && !--raysleft)
                        break;                          /* preemptive EOI */
        }
        while (next_child_nq(1) >= 0)           /* empty results queue */
                ;
                                                /* output accumulated record */
        if (accumulate <= 0 || account < accumulate) {
                end_children();                 /* frees up file descriptors */
                if (account < accumulate) {
                        error(WARNING, "partial accumulation in final record");
                        accumulate -= account;
                }
                for (i = 0; i < nmods; i++)
                        mod_output(out_bq->mca[i]);
                end_record();
                free_binq(out_bq);
                out_bq = NULL;
        }
        if (raysleft)
                error(USER, "unexpected EOF on input");
        free_binq(NULL);                        /* clean up */
        lu_done(&ofiletab);
#undef MAXIQ
}
Revision:	2.9
Committed:	Thu Jun 14 05:13:25 2012 UTC (11 years, 11 months ago) by greg
Content type:	text/plain
Branch:	MAIN
Changes since 2.8:	+15 -32 lines
Log Message:	Simplified creation of child process
#	User	Rev	Content
1	greg	2.1	#ifndef lint
2	greg	2.9	static const char RCSid[] = "$Id: rc3.c,v 2.8 2012/06/12 22:46:45 greg Exp $";
3	greg	2.1	#endif
4			/*
5			* Accumulate ray contributions for a set of materials
6			* Controlling process for multiple children
7			*/
8
9			#include "rcontrib.h"
10			#include "platform.h"
11			#include "rtprocess.h"
12			#include "selcall.h"
13
14			/* Modifier contribution queue (results waiting to be output) */
15			typedef struct s_binq {
16	greg	2.6	RNUMBER ndx; /* index for this entry */
17			RNUMBER nadded; /* accumulated so far */
18	greg	2.1	struct s_binq next; / next in queue */
19			MODCONT mca[1]; / contrib. array (extends struct) */
20			} BINQ;
21
22			static BINQ out_bq = NULL; / output bin queue */
23			static BINQ free_bq = NULL; / free queue entries */
24
25	greg	2.6	static struct {
26			RNUMBER r1; /* assigned ray starting index */
27			SUBPROC pr; /* PID, i/o descriptors */
28			FILE infp; / file pointer to read from process */
29			int nr; /* number rays to sum (0 if free) */
30			} kida[MAXPROCESS]; /* our child processes */
31	greg	2.1
32
33	greg	2.4	/* Get new bin queue entry */
34	greg	2.1	static BINQ *
35			new_binq()
36			{
37	greg	2.4	BINQ *bp;
38	greg	2.1	int i;
39
40	greg	2.4	if (free_bq != NULL) { /* something already available? */
41			bp = free_bq;
42	greg	2.1	free_bq = bp->next;
43			bp->next = NULL;
44	greg	2.6	bp->nadded = 0;
45	greg	2.1	return(bp);
46			}
47			/* else allocate fresh */
48	greg	2.4	bp = (BINQ )malloc(sizeof(BINQ) + sizeof(MODCONT )*(nmods-1));
49	greg	2.1	if (bp == NULL)
50			goto memerr;
51			for (i = nmods; i--; ) {
52			MODCONT mp = (MODCONT )lu_find(&modconttab,modname[i])->data;
53			bp->mca[i] = (MODCONT *)malloc(sizeof(MODCONT) +
54			sizeof(DCOLOR)*(mp->nbins-1));
55			if (bp->mca[i] == NULL)
56			goto memerr;
57			memcpy(bp->mca[i], mp, sizeof(MODCONT)-sizeof(DCOLOR));
58			/* memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)mp->nbins); /
59			}
60			bp->ndx = 0;
61	greg	2.6	bp->nadded = 0;
62	greg	2.1	bp->next = NULL;
63			return(bp);
64			memerr:
65			error(SYSTEM, "out of memory in new_binq()");
66			return(NULL);
67			}
68
69
70			/* Free a bin queue entry */
71			static void
72			free_binq(BINQ *bp)
73			{
74			int i;
75
76			if (bp == NULL) { /* signal to release our free list */
77			while ((bp = free_bq) != NULL) {
78			free_bq = bp->next;
79			for (i = nmods; i--; )
80			free(bp->mca[i]);
81			/* Note: we don't own bp->mca[i]->binv */
82			free(bp);
83			}
84			return;
85			}
86			/* clear sums for next use */
87			/* for (i = nmods; i--; )
88			memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
89			*/
90			bp->ndx = 0;
91			bp->next = free_bq; /* push onto free list */
92			free_bq = bp;
93			}
94
95
96			/* Add modifier values to accumulation record in queue and clear */
97			void
98			queue_modifiers()
99			{
100			MODCONT mpin, mpout;
101			int i, j;
102
103			if ((accumulate > 0) \| (out_bq == NULL))
104			error(CONSISTENCY, "bad call to queue_modifiers()");
105
106			for (i = nmods; i--; ) {
107			mpin = (MODCONT *)lu_find(&modconttab,modname[i])->data;
108			mpout = out_bq->mca[i];
109			for (j = mpout->nbins; j--; )
110			addcolor(mpout->cbin[j], mpin->cbin[j]);
111			memset(mpin->cbin, 0, sizeof(DCOLOR)*mpin->nbins);
112			}
113	greg	2.5	out_bq->nadded++;
114	greg	2.1	}
115
116
117	greg	2.4	/* Sum one modifier record into another (updates nadded) */
118	greg	2.1	static void
119			add_modbin(BINQ dst, BINQ src)
120			{
121			int i, j;
122
123			for (i = nmods; i--; ) {
124			MODCONT *mpin = src->mca[i];
125			MODCONT *mpout = dst->mca[i];
126			for (j = mpout->nbins; j--; )
127			addcolor(mpout->cbin[j], mpin->cbin[j]);
128			}
129	greg	2.4	dst->nadded += src->nadded;
130	greg	2.1	}
131
132
133	greg	2.2	/* Queue values for later output */
134			static void
135	greg	2.1	queue_output(BINQ *bp)
136			{
137			BINQ b_last, b_cur;
138
139			if (accumulate <= 0) { /* just accumulating? */
140			if (out_bq == NULL) {
141			bp->next = NULL;
142			out_bq = bp;
143			} else {
144			add_modbin(out_bq, bp);
145			free_binq(bp);
146			}
147	greg	2.2	return;
148	greg	2.1	}
149			b_last = NULL; /* else insert in output queue */
150			for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
151			b_cur = b_cur->next)
152			b_last = b_cur;
153	greg	2.4
154	greg	2.1	if (b_last != NULL) {
155			bp->next = b_cur;
156			b_last->next = bp;
157			} else {
158			bp->next = out_bq;
159			out_bq = bp;
160			}
161	greg	2.3	if (accumulate == 1) /* no accumulation? */
162	greg	2.2	return;
163			b_cur = out_bq; /* else merge accumulation entries */
164			while (b_cur->next != NULL) {
165	greg	2.4	if (b_cur->nadded >= accumulate \|\|
166	greg	2.2	(b_cur->ndx-1)/accumulate !=
167			(b_cur->next->ndx-1)/accumulate) {
168			b_cur = b_cur->next;
169			continue;
170	greg	2.1	}
171	greg	2.2	add_modbin(b_cur, b_cur->next);
172			b_last = b_cur->next;
173			b_cur->next = b_last->next;
174			b_last->next = NULL;
175			free_binq(b_last);
176	greg	2.1	}
177	greg	2.2	}
178
179
180	greg	2.4	/* Count number of records ready for output */
181			static int
182			queue_ready()
183			{
184			int nready = 0;
185			BINQ *bp;
186
187			if (accumulate <= 0) /* just accumulating? */
188			return(0);
189
190			for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
191			bp->ndx == lastdone+nready*accumulate+1;
192			bp = bp->next)
193			++nready;
194
195			return(nready);
196			}
197
198
199			/* Catch up with output queue by producing ready results */
200	greg	2.2	static int
201	greg	2.4	output_catchup(int nmax)
202	greg	2.2	{
203			int nout = 0;
204			BINQ *bp;
205			int i;
206
207			if (accumulate <= 0) /* just accumulating? */
208			return(0);
209			/* else output ready results */
210	greg	2.4	while (out_bq != NULL && out_bq->nadded >= accumulate
211			&& out_bq->ndx == lastdone+1) {
212			if ((nmax > 0) & (nout >= nmax))
213			break;
214	greg	2.2	bp = out_bq; /* pop off first entry */
215			out_bq = bp->next;
216			bp->next = NULL;
217	greg	2.1	for (i = 0; i < nmods; i++) /* output record */
218	greg	2.2	mod_output(bp->mca[i]);
219	greg	2.1	end_record();
220	greg	2.2	free_binq(bp); /* free this entry */
221	greg	2.1	lastdone += accumulate;
222			++nout;
223			}
224			return(nout);
225			}
226
227
228	greg	2.3	/* Put a zero record in results queue & output */
229	greg	2.1	void
230	greg	2.3	put_zero_record(int ndx)
231	greg	2.1	{
232			BINQ *bp = new_binq();
233			int i;
234
235			for (i = nmods; i--; )
236			memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
237			bp->ndx = ndx;
238	greg	2.6	bp->nadded = 1;
239	greg	2.1	queue_output(bp);
240	greg	2.4	output_catchup(0);
241	greg	2.1	}
242
243
244	greg	2.6	/* Get results from child process and add to queue */
245			static void
246			queue_results(int k)
247			{
248			BINQ bq = new_binq(); / get results holder */
249			int j;
250
251			bq->ndx = kida[k].r1;
252			bq->nadded = kida[k].nr;
253			/* read from child */
254			for (j = 0; j < nmods; j++)
255			if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), bq->mca[j]->nbins,
256			kida[k].infp) != bq->mca[j]->nbins)
257			error(SYSTEM, "read error from render process");
258
259			queue_output(bq); /* put results in output queue */
260			kida[k].nr = 0; /* mark child as available */
261			}
262
263
264	greg	2.1	/* callback to set output spec to NULL (stdout) */
265			static int
266			set_stdout(const LUENT le, void p)
267			{
268			((MODCONT )le->data).outspec = NULL;
269			return(0);
270			}
271
272
273			/* Start child processes if we can */
274			int
275			in_rchild()
276			{
277	greg	2.9	int rval;
278
279			while (nchild < nproc) { /* fork until target reached */
280	greg	2.3	errno = 0;
281	greg	2.9	rval = open_process(&kida[nchild].pr, NULL);
282			if (rval < 0)
283			error(SYSTEM, "open_process() call failed");
284			if (rval == 0) { /* if in child, set up & return true */
285	greg	2.4	lu_doall(&modconttab, set_stdout, NULL);
286			lu_done(&ofiletab);
287	greg	2.5	while (nchild--) { /* don't share other pipes */
288	greg	2.6	close(kida[nchild].pr.w);
289			fclose(kida[nchild].infp);
290	greg	2.4	}
291	greg	2.1	inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
292			outfmt = 'd';
293			header = 0;
294			yres = 0;
295			raysleft = 0;
296	greg	2.6	if (accumulate == 1) {
297			waitflush = xres = 1;
298			account = accumulate = 1;
299			} else { /* parent controls accumulation */
300			waitflush = xres = 0;
301			account = accumulate = 0;
302			}
303	greg	2.9	return(1); /* return "true" in child */
304	greg	2.1	}
305	greg	2.9	if (rval != PIPE_BUF)
306			error(CONSISTENCY, "bad value from open_process()");
307			/* connect to child's output */
308			kida[nchild].infp = fdopen(kida[nchild].pr.r, "rb");
309	greg	2.6	if (kida[nchild].infp == NULL)
310	greg	2.2	error(SYSTEM, "out of memory in in_rchild()");
311	greg	2.3	#ifdef getc_unlocked
312	greg	2.6	flockfile(kida[nchild].infp); /* avoid mutex overhead */
313	greg	2.3	#endif
314	greg	2.6	kida[nchild++].nr = 0; /* mark as available */
315	greg	2.1	}
316	greg	2.9	return(0); /* return "false" in parent */
317	greg	2.1	}
318
319
320			/* Close child processes */
321			void
322			end_children()
323			{
324			int status;
325
326	greg	2.5	while (nchild > 0) {
327			nchild--;
328	greg	2.6	if ((status = close_process(&kida[nchild].pr)) > 0) {
329	greg	2.1	sprintf(errmsg,
330			"rendering process returned bad status (%d)",
331			status);
332			error(WARNING, errmsg);
333			}
334	greg	2.9	fclose(kida[nchild].infp);
335	greg	2.2	}
336	greg	2.1	}
337
338
339	greg	2.5	/* Wait for the next available child, managing output queue simultaneously */
340	greg	2.1	static int
341	greg	2.5	next_child_nq(int flushing)
342	greg	2.1	{
343	greg	2.2	static struct timeval polling;
344	greg	2.4	struct timeval *pmode;
345	greg	2.2	fd_set readset, errset;
346	greg	2.6	int i, n, nr, nqr;
347	greg	2.1
348	greg	2.5	if (!flushing) /* see if there's one free */
349	greg	2.1	for (i = nchild; i--; )
350	greg	2.6	if (!kida[i].nr)
351	greg	2.1	return(i);
352	greg	2.4
353	greg	2.5	nqr = queue_ready(); /* choose blocking mode or polling */
354			if ((nqr > 0) & !flushing)
355			pmode = &polling;
356			else
357	greg	2.4	pmode = NULL;
358	greg	2.5	tryagain: /* catch up with output? */
359			if (pmode == &polling) {
360			if (nqr > nchild) /* don't get too far behind */
361			nqr -= output_catchup(nqr-nchild);
362			} else if (nqr > 0) /* clear output before blocking */
363			nqr -= output_catchup(0);
364	greg	2.1	/* prepare select() call */
365			FD_ZERO(&readset); FD_ZERO(&errset);
366			n = nr = 0;
367			for (i = nchild; i--; ) {
368	greg	2.6	if (kida[i].nr) {
369			FD_SET(kida[i].pr.r, &readset);
370	greg	2.1	++nr;
371			}
372	greg	2.6	FD_SET(kida[i].pr.r, &errset);
373			if (kida[i].pr.r >= n)
374			n = kida[i].pr.r + 1;
375	greg	2.1	}
376	greg	2.3	if (!nr) /* nothing to wait for? */
377			return(-1);
378	greg	2.2	if ((nr > 1) \| (pmode == &polling)) {
379	greg	2.1	errno = 0;
380	greg	2.5	nr = select(n, &readset, NULL, &errset, pmode);
381			if (!nr) {
382	greg	2.2	pmode = NULL; /* try again, blocking this time */
383			goto tryagain;
384			}
385	greg	2.5	if (nr < 0)
386	greg	2.3	error(SYSTEM, "select() error in next_child_nq()");
387	greg	2.1	} else
388			FD_ZERO(&errset);
389			n = -1; /* read results from child(ren) */
390			for (i = nchild; i--; ) {
391	greg	2.6	if (FD_ISSET(kida[i].pr.r, &errset))
392	greg	2.1	error(USER, "rendering process died");
393	greg	2.6	if (FD_ISSET(kida[i].pr.r, &readset))
394			queue_results(n = i);
395	greg	2.1	}
396	greg	2.3	return(n); /* first available child */
397	greg	2.1	}
398
399
400			/* Run parental oversight loop */
401			void
402			parental_loop()
403			{
404	greg	2.6	#define MAXIQ (int)(PIPE_BUF/(sizeof(FVECT)*2))
405	greg	2.1	static int ignore_warning_given = 0;
406	greg	2.6	int qlimit = (accumulate == 1) ? 1 : MAXIQ-1;
407			int ninq = 0;
408			FVECT orgdir[2*MAXIQ];
409			int i, n;
410	greg	2.1	/* load rays from stdin & process */
411			#ifdef getc_unlocked
412			flockfile(stdin); /* avoid lock/unlock overhead */
413			#endif
414	greg	2.6	while (getvec(orgdir[2ninq]) == 0 && getvec(orgdir[2ninq+1]) == 0) {
415	greg	2.8	if (orgdir[2ninq+1][0] == 0.0 && / asking for flush? */
416			(orgdir[2*ninq+1][1] == 0.0) &
417			(orgdir[2*ninq+1][2] == 0.0)) {
418	greg	2.6	if (accumulate != 1) {
419			if (!ignore_warning_given++)
420			error(WARNING,
421	greg	2.1	"dummy ray(s) ignored during accumulation\n");
422	greg	2.6	continue;
423			}
424	greg	2.1	while (next_child_nq(1) >= 0)
425	greg	2.3	; /* clear the queue */
426	greg	2.1	lastdone = lastray = 0;
427			if ((yres <= 0) \| (xres <= 0))
428	greg	2.5	waitflush = 1; /* flush next */
429	greg	2.3	put_zero_record(++lastray);
430	greg	2.6	} else if (++ninq >= qlimit \|\| accumulate > 1 &&
431			lastray/accumulate != (lastray+ninq)/accumulate) {
432	greg	2.5	i = next_child_nq(0); /* manages output */
433	greg	2.6	n = ninq;
434			if (accumulate != 1) /* request flush? */
435			memset(orgdir[2n++], 0, sizeof(FVECT)2);
436			n = sizeof(FVECT)2; /* send assignment */
437			if (writebuf(kida[i].pr.w, (char *)orgdir, n) != n)
438	greg	2.1	error(SYSTEM, "pipe write error");
439	greg	2.6	kida[i].r1 = lastray+1;
440			lastray += kida[i].nr = ninq; /* mark as busy */
441			ninq = 0;
442			if (lastray < lastdone) { /* RNUMBER wrapped? */
443			while (next_child_nq(1) >= 0)
444			;
445			lastdone = lastray = 0;
446			}
447	greg	2.1	}
448			if (raysleft && !--raysleft)
449			break; /* preemptive EOI */
450			}
451			while (next_child_nq(1) >= 0) /* empty results queue */
452			;
453			/* output accumulated record */
454			if (accumulate <= 0 \|\| account < accumulate) {
455	greg	2.5	end_children(); /* frees up file descriptors */
456	greg	2.1	if (account < accumulate) {
457			error(WARNING, "partial accumulation in final record");
458			accumulate -= account;
459			}
460			for (i = 0; i < nmods; i++)
461			mod_output(out_bq->mca[i]);
462			end_record();
463	greg	2.3	free_binq(out_bq);
464			out_bq = NULL;
465	greg	2.1	}
466			if (raysleft)
467			error(USER, "unexpected EOF on input");
468			free_binq(NULL); /* clean up */
469			lu_done(&ofiletab);
470	greg	2.6	#undef MAXIQ
471	greg	2.1	}