src/rt/rc3.c

#ifndef lint
static const char RCSid[] = "$Id: rc3.c,v 2.24 2020/09/09 21:28:19 greg Exp $";
#endif
/*
 * Accumulate ray contributions for a set of materials
 * Controlling process for multiple children
 */

#include <signal.h>
#include "rcontrib.h"
#include "selcall.h"

#define MAXIQ           (int)(PIPE_BUF/(sizeof(FVECT)*2))

/* 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 SUBPROC  kidpr[MAXPROCESS];      /* our child processes */

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


/* 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(mcsize(mp->nbins));
                if (bp->mca[i] == NULL)
                        goto memerr;
                memcpy(bp->mca[i], mp, sizeof(MODCONT)-sizeof(DCOLORV));
                /* memset(bp->mca[i]->cbin, 0, DCOLORSIZ*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, DCOLORSIZ*bp->mca[i]->nbins);
*/
        if (bp->next != NULL)
                error(CONSISTENCY, "free_binq() handed list");
        bp->ndx = 0;
        bp->next = free_bq;             /* push onto free list */
        free_bq = bp;
}


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

        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];
                ssrc = mcbin(mpin, mpin->nbins);
                sdst = mcbin(mpout, mpout->nbins);
                while (sdst > mpout->cbin)
                        *--sdst += *--ssrc;
                memset(mpin->cbin, 0, DCOLORSIZ*mpin->nbins);
        }
        out_bq->nadded++;
}


/* Sum one modifier record into another (updates nadded) */
static void
add_modbin(BINQ *dst, BINQ *src)
{
        MODCONT *mpin, *mpout;
        DCOLORV *ssrc, *sdst;
        int     i;

        for (i = nmods; i--; ) {
                mpin = src->mca[i];
                mpout = dst->mca[i];
                ssrc = mcbin(mpin, mpin->nbins);
                sdst = mcbin(mpout, mpout->nbins);
                while (sdst > mpout->cbin)
                        *--sdst += *--ssrc;
        }
        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;                  /* 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;

        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;
                                        /* 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, DCOLORSIZ*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 (getbinary(bq->mca[j]->cbin, DCOLORSIZ, 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 (call only once in parent!) */
int
in_rchild()
{
        int     rval;

        while (nchild < nproc) {        /* fork until target reached */
                errno = 0;
                rval = open_process(&kidpr[nchild], 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(kidpr[nchild].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(kidpr[nchild].r, "rb");
                if (kida[nchild].infp == NULL)
                        error(SYSTEM, "out of memory in in_rchild()");
                kida[nchild++].nr = 0;  /* mark as available */
        }
#ifdef getc_unlocked
        for (rval = nchild; rval--; )   /* avoid mutex overhead */
                flockfile(kida[rval].infp);
#endif
        return(0);                      /* return "false" in parent */
}


/* Close child processes */
void
end_children(int immed)
{
        int     i;

#ifdef SIGKILL                          /* error mode -- quick exit */
        for (i = nchild*immed; i-- > 0; )
                kill(kidpr[nchild].pid, SIGKILL);
#endif
        if ((i = close_processes(kidpr, nchild)) > 0 && !immed) {
                sprintf(errmsg, "rendering process returned bad status (%d)",
                                        i);
                        error(WARNING, errmsg);
        }
        while (nchild-- > 0)
                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(kidpr[i].r, &readset);
                        ++nr;
                }
                FD_SET(kidpr[i].r, &errset);
                if (kidpr[i].r >= n)
                        n = kidpr[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--; ) {
                if (FD_ISSET(kidpr[i].r, &errset))
                        error(USER, "rendering process died");
                if (FD_ISSET(kidpr[i].r, &readset))
                        queue_results(n = i);
        }
        return(n);                      /* first available child */
}


/* Run parental oversight loop */
void
parental_loop()
{
        const 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) {
                const int       zero_ray = orgdir[2*ninq+1][0] == 0.0 &&
                                                (orgdir[2*ninq+1][1] == 0.0) &
                                                (orgdir[2*ninq+1][2] == 0.0);
                ninq += !zero_ray;
                                /* Zero ray cannot go in input queue */
                if (zero_ray ? ninq : ninq >= qlimit ||
                            lastray/accumulate != (lastray+ninq)/accumulate) {
                        i = next_child_nq(0);           /* manages output */
                        n = ninq;
                        if (accumulate > 1)             /* need terminator? */
                                memset(orgdir[2*n++], 0, sizeof(FVECT)*2);
                        n *= sizeof(FVECT)*2;           /* send assignment */
                        if (writebuf(kidpr[i].w, (char *)orgdir, n) != n)
                                error(SYSTEM, "pipe write error");
                        kida[i].r1 = lastray+1;
                        lastray += kida[i].nr = ninq;   /* mark as busy */
                        if (lastray < lastdone) {       /* RNUMBER wrapped? */
                                while (next_child_nq(1) >= 0)
                                        ;
                                lastray -= ninq;
                                lastdone = lastray %= accumulate;
                        }
                        ninq = 0;
                }
                if (zero_ray) {                         /* put bogus record? */
                        if ((yres <= 0) | (xres <= 1) &&
                                        (lastray+1) % accumulate == 0) {
                                while (next_child_nq(1) >= 0)
                                        ;               /* clear the queue */
                                lastdone = lastray = accumulate-1;
                                waitflush = 1;          /* flush next */
                        }
                        put_zero_record(++lastray);
                }
                if (!morays())
                        break;                          /* preemptive EOI */
        }
        while (next_child_nq(1) >= 0)           /* empty results queue */
                ;
        if (account < accumulate) {
                error(WARNING, "partial accumulation in final record");
                free_binq(out_bq);              /* XXX just ignore it */
                out_bq = NULL;
        }
        free_binq(NULL);                        /* clean up */
        lu_done(&ofiletab);
        if (raysleft)
                error(USER, "unexpected EOF on input");
}


/* Wait for the next available child by monitoring "to" pipes */
static int
next_child_ready()
{
        fd_set                  writeset, errset;
        int                     i, n;

        for (i = nchild; i--; )         /* see if there's one free first */
                if (!kida[i].nr)
                        return(i);
                                        /* prepare select() call */
        FD_ZERO(&writeset); FD_ZERO(&errset);
        n = 0;
        for (i = nchild; i--; ) {
                FD_SET(kidpr[i].w, &writeset);
                FD_SET(kidpr[i].r, &errset);
                if (kidpr[i].w >= n)
                        n = kidpr[i].w + 1;
                if (kidpr[i].r >= n)
                        n = kidpr[i].r + 1;
        }
        errno = 0;
        n = select(n, NULL, &writeset, &errset, NULL);
        if (n < 0)
                error(SYSTEM, "select() error in next_child_ready()");
        n = -1;                         /* identify waiting child */
        for (i = nchild; i--; ) {
                if (FD_ISSET(kidpr[i].r, &errset))
                        error(USER, "rendering process died");
                if (FD_ISSET(kidpr[i].w, &writeset))
                        kida[n = i].nr = 0;
        }
        return(n);                      /* first available child */
}


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