src/rt/rc3.c

#ifndef lint
static const char RCSid[] = "$Id: rc3.c,v 2.4 2012/06/11 05:07:55 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 {
        int             ndx;            /* index for this entry */
        int             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 SUBPROC  kida[MAXPROCESS];       /* child processes */
static FILE     *inq_fp[MAXPROCESS];    /* input streams */


/* 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 = 1;
                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 = 1;
        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;
        queue_output(bp);
        output_catchup(0);
}


/* 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()
{
#ifdef _WIN32
        error(WARNING, "multiprocessing unsupported -- running solo");
        nproc = 1;
        return(1);
#else
                                        /* try to fork ourselves */
        while (nchild < nproc) {
                int     p0[2], p1[2];
                int     pid;
                                        /* prepare i/o pipes */
                errno = 0;
                if (pipe(p0) < 0 || pipe(p1) < 0)
                        error(SYSTEM, "pipe() call failed!");
                pid = fork();           /* fork parent process */
                if (pid == 0) {         /* if in child, set up & return true */
                        close(p0[1]); close(p1[0]);
                        lu_doall(&modconttab, set_stdout, NULL);
                        lu_done(&ofiletab);
                        while (nchild--) {      /* don't share other pipes */
                                close(kida[nchild].w);
                                fclose(inq_fp[nchild]);
                        }
                        dup2(p0[0], 0); close(p0[0]);
                        dup2(p1[1], 1); close(p1[1]);
                        inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
                        outfmt = 'd';
                        header = 0;
                        waitflush = xres = 1;
                        yres = 0;
                        raysleft = 0;
                        account = accumulate = 1;
                        return(1);      /* child return value */
                }
                if (pid < 0)
                        error(SYSTEM, "fork() call failed!");
                                        /* connect parent's pipes */
                close(p0[0]); close(p1[1]);
                kida[nchild].r = p1[0];
                kida[nchild].w = p0[1];
                kida[nchild].pid = pid;
                kida[nchild].running = -1;
                inq_fp[nchild] = fdopen(p1[0], "rb");
                if (inq_fp[nchild] == NULL)
                        error(SYSTEM, "out of memory in in_rchild()");
#ifdef getc_unlocked
                flockfile(inq_fp[nchild]);      /* avoid mutex overhead */
#endif
                ++nchild;
        }
        return(0);                      /* parent return value */
#endif
}


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


/* 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, j, n, nr, nqr;

        if (!flushing)                  /* see if there's one free */
                for (i = nchild; i--; )
                        if (kida[i].running < 0)
                                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].running > 0) {
                        FD_SET(kida[i].r, &readset);
                        ++nr;
                }
                FD_SET(kida[i].r, &errset);
                if (kida[i].r >= n)
                        n = kida[i].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--; ) {
                BINQ    *bq;
                if (FD_ISSET(kida[i].r, &errset))
                        error(USER, "rendering process died");
                if (!FD_ISSET(kida[i].r, &readset))
                        continue;
                bq = new_binq();        /* get results holder */
                bq->ndx = kida[i].running;
                                        /* read from child */
                for (j = 0; j < nmods; j++) {
                        nr = bq->mca[j]->nbins;
                        if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), nr,
                                                        inq_fp[i]) != nr)
                                error(SYSTEM, "read error from render process");
                }
                queue_output(bq);       /* add results to output queue */
                kida[i].running = -1;   /* mark child as available */
                n = i;
        }
        return(n);                      /* first available child */
}


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