src/rt/raypcalls.c

#ifndef lint
static const char       RCSid[] = "$Id: raypcalls.c,v 2.32 2020/04/30 19:30:48 greg Exp $";
#endif
/*
 *  raypcalls.c - interface for parallel rendering using Radiance
 *
 *  External symbols declared in ray.h
 */

#include "copyright.h"

/*
 *  These calls are designed similarly to the ones in raycalls.c,
 *  but allow for multiple rendering processes on the same host
 *  machine.  There is no sense in specifying more child processes
 *  than you have processor cores, but one child may help by allowing
 *  asynchronous ray computation in an interactive program, and
 *  will protect the caller from fatal rendering errors.
 *
 *  You should first read and understand the header in raycalls.c,
 *  as some things are explained there that are not repated here.
 *
 *  The first step is opening one or more rendering processes
 *  with a call to ray_pinit(oct, nproc).  Before calling fork(),
 *  ray_pinit() loads the octree and data structures into the
 *  caller's memory, and ray_popen() synchronizes the ambient
 *  file, if any.  Shared memory permits all sorts of queries
 *  that wouldn't be possible otherwise without causing any real
 *  memory overhead, since all the static data are shared
 *  between processes.  Rays are traced using a simple
 *  queuing mechanism, explained below.
 *
 *  The ray queue buffers RAYQLEN rays before sending to
 *  children, each of which may internally buffer RAYQLEN rays
 *  during evaluation.  Rays are not returned in the order
 *  they are sent when multiple processes are open.
 *
 *  Rays are queued and returned by a single
 *  ray_pqueue() call.  A ray_pqueue() return
 *  value of 0 indicates that no rays are ready
 *  and the queue is not yet full.  A return value of 1
 *  indicates that a ray was returned, though it is probably
 *  not the one you just requested.  Rays may be identified by
 *  the rno member of the RAY struct, which is incremented
 *  by the rayorigin() call, or may be set explicitly by
 *  the caller.  Below is an example call sequence:
 *
 *      myRay.rorg = ( ray origin point )
 *      myRay.rdir = ( normalized ray direction )
 *      myRay.rmax = ( maximum length, or zero for no limit )
 *      rayorigin(&myRay, PRIMARY, NULL, NULL);
 *      myRay.rno = ( my personal ray identifier )
 *      if (ray_pqueue(&myRay) == 1)
 *              { do something with results }
 *
 *  Note the differences between this and the simpler ray_trace()
 *  call.  In particular, the call may or may not return a value
 *  in the passed ray structure.  Also, you need to call rayorigin()
 *  yourself, which is normally called for you by ray_trace().  The
 *  benefit is that ray_pqueue() will trace rays faster in
 *  proportion to the number of CPUs you have available on your
 *  system.  If the ray queue is full before the call, ray_pqueue()
 *  will block until a result is ready so it can queue this one.
 *  The global int ray_pnidle indicates the number of currently idle
 *  children.  If you want to check for completed rays without blocking,
 *  or get the results from rays that have been queued without
 *  queuing any new ones, the ray_presult() call is for you:
 *
 *      if (ray_presult(&myRay, 1) == 1)
 *              { do something with results }
 *
 *  If the second argument is 1, the call won't block when
 *  results aren't ready, but will immediately return 0.
 *  If the second argument is 0, the call will block
 *  until a value is available, returning 0 only if the
 *  queue is completely empty.  Setting the second argument
 *  to -1 returns 0 unless a ray is ready in the queue and
 *  no system calls are needed.  A negative return value
 *  indicates that a rendering process died.  If this
 *  happens, ray_pclose(0) is automatically called to close
 *  all child processes, and ray_pnprocs is set to zero.
 *
 *  If you just want to fill the ray queue without checking for
 *  results, check ray_pnidle and call ray_psend():
 *
 *      while (ray_pnidle) {
 *              ( set up ray )
 *              ray_psend(&myRay);
 *      }
 *
 *  Note that it is a mistake to call ra_psend() when
 *  ray_pnidle is zero, and nothing will be sent in
 *  this case.  Otherwise, the ray_presult() and/or ray_pqueue()
 *  functions may be called subsequently to read back the results
 *  of rays queued by ray_psend().
 *
 *  When you are done, you may call ray_pdone(1) to close
 *  all child processes and clean up memory used by Radiance.
 *  Any queued ray calculations will be awaited and discarded.
 *  As with ray_done(), ray_pdone(0) hangs onto data files
 *  and fonts that are likely to be used in subsequent renderings.
 *  Whether you need to clean up memory or not, you should
 *  at least call ray_pclose(0) to await the child processes.
 *  The caller should define a quit() function that calls
 *  ray_pclose(0) if ray_pnprocs > 0.
 *
 *  Warning:  You cannot affect any of the rendering processes
 *  by changing global parameter values onece ray_pinit() has
 *  been called.  Changing global parameters will have no effect
 *  until the next call to ray_pinit(), which restarts everything.
 *  If you just want to reap children so that you can alter the
 *  rendering parameters without reloading the scene, use the
 *  ray_pclose(0) and ray_popen(nproc) calls to close
 *  then restart the child processes after the changes are made.
 *
 *  Note:  These routines are written to coordinate with the
 *  definitions in raycalls.c, and in fact depend on them.
 *  If you want to trace a ray and get a result synchronously,
 *  use the ray_trace() call to compute it in the parent process.
 *  This will not interfere with any subprocess calculations,
 *  but beware that a fatal error may end with a call to quit().
 *
 *  Note:  One of the advantages of using separate processes
 *  is that it gives the calling program some immunity from
 *  fatal rendering errors.  As discussed in raycalls.c,
 *  Radiance tends to throw up its hands and exit at the
 *  first sign of trouble, calling quit() to return control
 *  to the top level.  Although you can avoid exit() with
 *  your own longjmp() in quit(), the cleanup afterwards
 *  is always suspect.  Through the use of subprocesses,
 *  we avoid this pitfall by closing the processes and
 *  returning a negative value from ray_pqueue() or
 *  ray_presult().  If you get a negative value from either
 *  of these calls, you can assume that the processes have
 *  been cleaned up with a call to ray_pclose(), though you
 *  will have to call ray_pdone() yourself if you want to
 *  free memory.  Obviously, you cannot continue rendering
 *  without risking further errors, but otherwise your
 *  process should not be compromised.
 */

#include  "rtprocess.h"
#include  "ray.h"
#include  "ambient.h"
#include  <sys/types.h>
#include  <sys/wait.h>
#include  "selcall.h"

#ifndef RAYQLEN
#define RAYQLEN         24              /* # rays to send at once */
#endif

#ifndef MAX_RPROCS
#if (FD_SETSIZE/2-4 < 64) 
#define MAX_NPROCS      (FD_SETSIZE/2-4)
#else
#define MAX_NPROCS      64              /* max. # rendering processes */
#endif
#endif

extern char     *shm_boundary;          /* boundary of shared memory */

int             ray_pnprocs = 0;        /* number of child processes */
int             ray_pnidle = 0;         /* number of idle children */
int             ray_pnbatch = 0;        /* throughput over responsiveness? */

static struct child_proc {
        RT_PID  pid;                            /* child process id */
        int     fd_send;                        /* write to child here */
        int     fd_recv;                        /* read from child here */
        int     npending;                       /* # rays in process */
        RNUMBER rno[RAYQLEN];                   /* working on these rays */
} r_proc[MAX_NPROCS];                   /* our child processes */

static RAY      r_queue[2*RAYQLEN];     /* ray i/o buffer */
static int      r_send_next = 0;        /* next send ray placement */
static int      r_recv_first = RAYQLEN; /* position of first unreported ray */
static int      r_recv_next = RAYQLEN;  /* next received ray placement */

static int      samplestep = 1;         /* sample step size */

#define sendq_full()    (r_send_next >= RAYQLEN)

static int ray_pflush(void);
static void ray_pchild(int fd_in, int fd_out);


void
ray_pinit(              /* initialize ray-tracing processes */
        char    *otnm,
        int     nproc
)
{
        if (nobjects > 0)               /* close old calculation */
                ray_pdone(0);

        ray_init(otnm);                 /* load the shared scene */

        ray_popen(nproc);               /* fork children */
}


static int
ray_pflush(void)                        /* send queued rays to idle children */
{
        int     nc, n, nw, i, sfirst;

        if ((ray_pnidle <= 0) | (r_send_next <= 0))
                return(0);              /* nothing we can send */
        
        sfirst = 0;                     /* divvy up labor */
        nc = ray_pnidle;
        for (i = ray_pnprocs; nc && i--; ) {
                if (r_proc[i].npending > 0)
                        continue;       /* child looks busy */
                n = r_send_next - sfirst;
                if (ray_pnbatch)
                        nc--;           /* maximize bundling for batch calc */
                else
                        n /= nc--;      /* distribute work for interactivity */
                if (!n)
                        continue;
                                        /* smuggle set size in crtype */
                r_queue[sfirst].crtype = n;
                nw = writebuf(r_proc[i].fd_send, (char *)&r_queue[sfirst],
                                sizeof(RAY)*n);
                if (nw != sizeof(RAY)*n)
                        return(-1);     /* write error */
                r_proc[i].npending = n;
                while (n--)             /* record ray IDs */
                        r_proc[i].rno[n] = r_queue[sfirst+n].rno;
                sfirst += r_proc[i].npending;
                ray_pnidle--;           /* now she's busy */
        }
        if (sfirst != r_send_next)
                error(CONSISTENCY, "code screwup in ray_pflush()");
        r_send_next = 0;
        return(sfirst);                 /* return total # sent */
}


int
ray_psend(                      /* add a ray to our send queue */
        RAY     *r
)
{
        int     rv;

        if ((r == NULL) | (ray_pnidle <= 0))
                return(0);
                                        /* flush output if necessary */
        if (sendq_full() && (rv = ray_pflush()) <= 0)
                return(rv);

        r_queue[r_send_next++] = *r;
        return(1);
}


int
ray_pqueue(                     /* queue a ray for computation */
        RAY     *r
)
{
        if (r == NULL)
                return(0);
                                        /* check for full send queue */
        if (sendq_full()) {
                RAY     mySend = *r;
                                        /* wait for a result */
                if (ray_presult(r, 0) <= 0)
                        return(-1);
                                        /* put new ray in queue */
                r_queue[r_send_next++] = mySend;

                return(1);
        }
                                        /* else add ray to send queue */
        r_queue[r_send_next++] = *r;
                                        /* check for returned ray... */
        if (r_recv_first >= r_recv_next)
                return(0);
                                        /* ...one is sitting in queue */
        *r = r_queue[r_recv_first++];
        return(1);
}


int
ray_presult(            /* check for a completed ray */
        RAY     *r,
        int     poll
)
{
        static struct timeval   tpoll;  /* zero timeval struct */
        static fd_set   readset, errset;
        int     n, ok;
        int     pn;

        if (r == NULL)
                return(0);
                                        /* check queued results first */
        if (r_recv_first < r_recv_next) {
                *r = r_queue[r_recv_first++];
                return(1);
        }
        if (poll < 0)                   /* immediate polling mode? */
                return(0);

        n = ray_pnprocs - ray_pnidle;   /* pending before flush? */

        if (ray_pflush() < 0)           /* send new rays to process */
                return(-1);
                                        /* reset receive queue */
        r_recv_first = r_recv_next = RAYQLEN;

        if (!poll)                      /* count newly sent unless polling */
                n = ray_pnprocs - ray_pnidle;
        if (n <= 0)                     /* return if nothing to await */
                return(0);
        if (!poll && ray_pnprocs == 1)  /* one process -> skip select() */
                FD_SET(r_proc[0].fd_recv, &readset);

getready:                               /* any children waiting for us? */
        for (pn = ray_pnprocs; pn--; )
                if (FD_ISSET(r_proc[pn].fd_recv, &readset) ||
                                FD_ISSET(r_proc[pn].fd_recv, &errset))
                        break;
                                        /* call select() if we must */
        if (pn < 0) {
                FD_ZERO(&readset); FD_ZERO(&errset); n = 0;
                for (pn = ray_pnprocs; pn--; ) {
                        if (r_proc[pn].npending > 0)
                                FD_SET(r_proc[pn].fd_recv, &readset);
                        FD_SET(r_proc[pn].fd_recv, &errset);
                        if (r_proc[pn].fd_recv >= n)
                                n = r_proc[pn].fd_recv + 1;
                }
                                        /* find out who is ready */
                while ((n = select(n, &readset, (fd_set *)NULL, &errset,
                                poll ? &tpoll : (struct timeval *)NULL)) < 0)
                        if (errno != EINTR) {
                                error(WARNING,
                                        "select call failed in ray_presult()");
                                ray_pclose(0);
                                return(-1);
                        }
                if (n > 0)              /* go back and get it */
                        goto getready;
                return(0);              /* else poll came up empty */
        }
        if (r_recv_next + r_proc[pn].npending > sizeof(r_queue)/sizeof(RAY))
                error(CONSISTENCY, "buffer shortage in ray_presult()");

                                        /* read rendered ray data */
        n = readbuf(r_proc[pn].fd_recv, (char *)&r_queue[r_recv_next],
                        sizeof(RAY)*r_proc[pn].npending);
        if (n > 0) {
                r_recv_next += n/sizeof(RAY);
                ok = (n == sizeof(RAY)*r_proc[pn].npending);
        } else
                ok = 0;
                                        /* reset child's status */
        FD_CLR(r_proc[pn].fd_recv, &readset);
        if (n <= 0)
                FD_CLR(r_proc[pn].fd_recv, &errset);
        r_proc[pn].npending = 0;
        ray_pnidle++;
                                        /* check for rendering errors */
        if (!ok) {
                ray_pclose(0);          /* process died -- clean up */
                return(-1);
        }
                                        /* preen returned rays */
        for (n = r_recv_next - r_recv_first; n--; ) {
                RAY     *rp = &r_queue[r_recv_first + n];
                rp->rno = r_proc[pn].rno[n];
                rp->parent = NULL;
                rp->newcset = rp->clipset = NULL;
                rp->rox = NULL;
                rp->slights = NULL;
        }
                                        /* return first ray received */
        *r = r_queue[r_recv_first++];
        return(1);
}


void
ray_pdone(              /* reap children and free data */
        int     freall
)
{
        ray_pclose(0);                  /* close child processes */

        if (shm_boundary != NULL) {     /* clear shared memory boundary */
                free((void *)shm_boundary);
                shm_boundary = NULL;
        }

        ray_done(freall);               /* free rendering data */
}


static void
ray_pchild(     /* process rays (never returns) */
        int     fd_in,
        int     fd_out
)
{
        int     n;
        int     i;
                                        /* flag child process for quit() */
        ray_pnprocs = -1;
                                        /* read each ray request set */
        while ((n = read(fd_in, (char *)r_queue, sizeof(r_queue))) > 0) {
                int     n2;
                if (n < sizeof(RAY))
                        break;
                                        /* get smuggled set length */
                n2 = sizeof(RAY)*r_queue[0].crtype - n;
                if (n2 < 0)
                        error(INTERNAL, "buffer over-read in ray_pchild()");
                if (n2 > 0) {           /* read the rest of the set */
                        i = readbuf(fd_in, (char *)r_queue + n, n2);
                        if (i != n2)
                                break;
                        n += n2;
                }
                n /= sizeof(RAY);
                                        /* evaluate rays */
                for (i = 0; i < n; i++) {
                        r_queue[i].crtype = r_queue[i].rtype;
                        r_queue[i].parent = NULL;
                        r_queue[i].clipset = NULL;
                        r_queue[i].slights = NULL;
                        r_queue[i].rlvl = 0;
                        samplendx += samplestep;
                        rayclear(&r_queue[i]);
                        rayvalue(&r_queue[i]);
                }
                                        /* write back our results */
                i = writebuf(fd_out, (char *)r_queue, sizeof(RAY)*n);
                if (i != sizeof(RAY)*n)
                        error(SYSTEM, "write error in ray_pchild()");
        }
        if (n)
                error(SYSTEM, "read error in ray_pchild()");
        ambsync();
        quit(0);                        /* normal exit */
}


void
ray_popen(                      /* open the specified # processes */
        int     nadd
)
{
                                        /* check if our table has room */
        if (ray_pnprocs + nadd > MAX_NPROCS)
                nadd = MAX_NPROCS - ray_pnprocs;
        if (nadd <= 0)
                return;
        ambsync();                      /* load any new ambient values */
        if (shm_boundary == NULL) {     /* first child process? */
                preload_objs();         /* preload auxiliary data */
                                        /* set shared memory boundary */
                shm_boundary = (char *)malloc(16);
                strcpy(shm_boundary, "SHM_BOUNDARY");
        }
        fflush(NULL);                   /* clear pending output */
        samplestep = ray_pnprocs + nadd;
        while (nadd--) {                /* fork each new process */
                int     p0[2], p1[2];
                if (pipe(p0) < 0 || pipe(p1) < 0)
                        error(SYSTEM, "cannot create pipe");
                if ((r_proc[ray_pnprocs].pid = fork()) == 0) {
                        int     pn;     /* close others' descriptors */
                        for (pn = ray_pnprocs; pn--; ) {
                                close(r_proc[pn].fd_send);
                                close(r_proc[pn].fd_recv);
                        }
                        close(p0[0]); close(p1[1]);
                        close(0);       /* don't share stdin */
                                        /* following call never returns */
                        ray_pchild(p1[0], p0[1]);
                }
                if (r_proc[ray_pnprocs].pid < 0)
                        error(SYSTEM, "cannot fork child process");
                close(p1[0]); close(p0[1]);
                if (rand_samp)          /* decorrelate random sequence */
                        srandom(random());
                else
                        samplendx++;
                /*
                 * Close write stream on exec to avoid multiprocessing deadlock.
                 * No use in read stream without it, so set flag there as well.
                 */
                fcntl(p1[1], F_SETFD, FD_CLOEXEC);
                fcntl(p0[0], F_SETFD, FD_CLOEXEC);
                r_proc[ray_pnprocs].fd_send = p1[1];
                r_proc[ray_pnprocs].fd_recv = p0[0];
                r_proc[ray_pnprocs].npending = 0;
                ray_pnprocs++;
                ray_pnidle++;
        }
}


void
ray_pclose(             /* close one or more child processes */
        int     nsub
)
{
        static int      inclose = 0;
        RAY             res;
        int             i, status = 0;
                                        /* check recursion */
        if (inclose)
                return;
        inclose++;
                                        /* check no child / in child */
        if (ray_pnprocs <= 0)
                return;
                                        /* check argument */
        if ((nsub <= 0) | (nsub > ray_pnprocs))
                nsub = ray_pnprocs;
                                        /* clear our ray queue */
        while (ray_presult(&res,0) > 0)
                ;
        r_send_next = 0;                /* hard reset in case of error */
        r_recv_first = r_recv_next = RAYQLEN;
                                        /* close send pipes */
        for (i = ray_pnprocs-nsub; i < ray_pnprocs; i++)
                close(r_proc[i].fd_send);

        if (nsub == 1) {                /* awaiting single process? */
                if (waitpid(r_proc[ray_pnprocs-1].pid, &status, 0) < 0)
                        status = 127<<8;
                close(r_proc[ray_pnprocs-1].fd_recv);
        } else                          /* else unordered wait */
                for (i = 0; i < nsub; ) {
                        int     j, mystatus;
                        RT_PID  pid = wait(&mystatus);
                        for (j = ray_pnprocs-nsub; j < ray_pnprocs; j++)
                                if (r_proc[j].pid == pid) {
                                        if (mystatus)
                                                status = mystatus;
                                        close(r_proc[j].fd_recv);
                                        ++i;
                                }
                }
        ray_pnprocs -= nsub;
        ray_pnidle -= nsub;
        if (status) {
                sprintf(errmsg, "rendering process exited with code %d", status>>8);
                error(WARNING, errmsg);
        }
        inclose--;
}
Revision:	2.33
Committed:	Mon Jun 15 15:44:03 2020 UTC (3 years, 11 months ago) by greg
Content type:	text/plain
Branch:	MAIN
Changes since 2.32:	+7 -2 lines
Log Message:	Optimize rtrace for throughput, especially with ambient cache
#	User	Rev	Content
1	greg	2.1	#ifndef lint
2	greg	2.33	static const char RCSid[] = "$Id: raypcalls.c,v 2.32 2020/04/30 19:30:48 greg Exp $";
3	greg	2.1	#endif
4			/*
5			* raypcalls.c - interface for parallel rendering using Radiance
6			*
7			* External symbols declared in ray.h
8			*/
9
10	greg	2.2	#include "copyright.h"
11	greg	2.1
12			/*
13			* These calls are designed similarly to the ones in raycalls.c,
14			* but allow for multiple rendering processes on the same host
15			* machine. There is no sense in specifying more child processes
16	greg	2.21	* than you have processor cores, but one child may help by allowing
17	greg	2.1	* asynchronous ray computation in an interactive program, and
18			* will protect the caller from fatal rendering errors.
19			*
20	greg	2.21	* You should first read and understand the header in raycalls.c,
21	greg	2.1	* as some things are explained there that are not repated here.
22			*
23			* The first step is opening one or more rendering processes
24			* with a call to ray_pinit(oct, nproc). Before calling fork(),
25			* ray_pinit() loads the octree and data structures into the
26	greg	2.13	* caller's memory, and ray_popen() synchronizes the ambient
27			* file, if any. Shared memory permits all sorts of queries
28	greg	2.21	* that wouldn't be possible otherwise without causing any real
29	greg	2.1	* memory overhead, since all the static data are shared
30	greg	2.21	* between processes. Rays are traced using a simple
31	greg	2.1	* queuing mechanism, explained below.
32			*
33	greg	2.14	* The ray queue buffers RAYQLEN rays before sending to
34	greg	2.21	* children, each of which may internally buffer RAYQLEN rays
35			* during evaluation. Rays are not returned in the order
36			* they are sent when multiple processes are open.
37	greg	2.14	*
38	greg	2.13	* Rays are queued and returned by a single
39	greg	2.1	* ray_pqueue() call. A ray_pqueue() return
40			* value of 0 indicates that no rays are ready
41			* and the queue is not yet full. A return value of 1
42			* indicates that a ray was returned, though it is probably
43			* not the one you just requested. Rays may be identified by
44			* the rno member of the RAY struct, which is incremented
45			* by the rayorigin() call, or may be set explicitly by
46			* the caller. Below is an example call sequence:
47			*
48			* myRay.rorg = ( ray origin point )
49			* myRay.rdir = ( normalized ray direction )
50			* myRay.rmax = ( maximum length, or zero for no limit )
51	greg	2.11	* rayorigin(&myRay, PRIMARY, NULL, NULL);
52	greg	2.1	* myRay.rno = ( my personal ray identifier )
53			* if (ray_pqueue(&myRay) == 1)
54			* { do something with results }
55			*
56			* Note the differences between this and the simpler ray_trace()
57			* call. In particular, the call may or may not return a value
58			* in the passed ray structure. Also, you need to call rayorigin()
59	greg	2.7	* yourself, which is normally called for you by ray_trace(). The
60			* benefit is that ray_pqueue() will trace rays faster in
61	greg	2.1	* proportion to the number of CPUs you have available on your
62			* system. If the ray queue is full before the call, ray_pqueue()
63			* will block until a result is ready so it can queue this one.
64	greg	2.3	* The global int ray_pnidle indicates the number of currently idle
65	greg	2.1	* children. If you want to check for completed rays without blocking,
66			* or get the results from rays that have been queued without
67			* queuing any new ones, the ray_presult() call is for you:
68			*
69			* if (ray_presult(&myRay, 1) == 1)
70			* { do something with results }
71			*
72			* If the second argument is 1, the call won't block when
73			* results aren't ready, but will immediately return 0.
74			* If the second argument is 0, the call will block
75			* until a value is available, returning 0 only if the
76	greg	2.26	* queue is completely empty. Setting the second argument
77			* to -1 returns 0 unless a ray is ready in the queue and
78			* no system calls are needed. A negative return value
79	greg	2.1	* indicates that a rendering process died. If this
80	greg	2.21	* happens, ray_pclose(0) is automatically called to close
81	greg	2.3	* all child processes, and ray_pnprocs is set to zero.
82	greg	2.1	*
83			* If you just want to fill the ray queue without checking for
84	greg	2.3	* results, check ray_pnidle and call ray_psend():
85	greg	2.1	*
86	greg	2.3	* while (ray_pnidle) {
87	greg	2.1	* ( set up ray )
88			* ray_psend(&myRay);
89			* }
90			*
91	greg	2.25	* Note that it is a mistake to call ra_psend() when
92			* ray_pnidle is zero, and nothing will be sent in
93			* this case. Otherwise, the ray_presult() and/or ray_pqueue()
94			* functions may be called subsequently to read back the results
95			* of rays queued by ray_psend().
96	greg	2.1	*
97			* When you are done, you may call ray_pdone(1) to close
98			* all child processes and clean up memory used by Radiance.
99			* Any queued ray calculations will be awaited and discarded.
100			* As with ray_done(), ray_pdone(0) hangs onto data files
101			* and fonts that are likely to be used in subsequent renderings.
102	greg	2.21	* Whether you need to clean up memory or not, you should
103			* at least call ray_pclose(0) to await the child processes.
104	greg	2.23	* The caller should define a quit() function that calls
105			* ray_pclose(0) if ray_pnprocs > 0.
106	greg	2.1	*
107			* Warning: You cannot affect any of the rendering processes
108			* by changing global parameter values onece ray_pinit() has
109			* been called. Changing global parameters will have no effect
110			* until the next call to ray_pinit(), which restarts everything.
111			* If you just want to reap children so that you can alter the
112			* rendering parameters without reloading the scene, use the
113			* ray_pclose(0) and ray_popen(nproc) calls to close
114	greg	2.7	* then restart the child processes after the changes are made.
115	greg	2.1	*
116			* Note: These routines are written to coordinate with the
117			* definitions in raycalls.c, and in fact depend on them.
118			* If you want to trace a ray and get a result synchronously,
119	greg	2.13	* use the ray_trace() call to compute it in the parent process.
120	greg	2.7	* This will not interfere with any subprocess calculations,
121			* but beware that a fatal error may end with a call to quit().
122	greg	2.1	*
123			* Note: One of the advantages of using separate processes
124			* is that it gives the calling program some immunity from
125			* fatal rendering errors. As discussed in raycalls.c,
126			* Radiance tends to throw up its hands and exit at the
127			* first sign of trouble, calling quit() to return control
128	greg	2.7	* to the top level. Although you can avoid exit() with
129	greg	2.1	* your own longjmp() in quit(), the cleanup afterwards
130			* is always suspect. Through the use of subprocesses,
131			* we avoid this pitfall by closing the processes and
132			* returning a negative value from ray_pqueue() or
133			* ray_presult(). If you get a negative value from either
134			* of these calls, you can assume that the processes have
135	greg	2.21	* been cleaned up with a call to ray_pclose(), though you
136	greg	2.1	* will have to call ray_pdone() yourself if you want to
137	greg	2.7	* free memory. Obviously, you cannot continue rendering
138			* without risking further errors, but otherwise your
139			* process should not be compromised.
140	greg	2.1	*/
141
142	schorsch	2.6	#include "rtprocess.h"
143	greg	2.1	#include "ray.h"
144	schorsch	2.6	#include "ambient.h"
145	greg	2.18	#include <sys/types.h>
146			#include <sys/wait.h>
147	greg	2.1	#include "selcall.h"
148
149			#ifndef RAYQLEN
150	greg	2.32	#define RAYQLEN 24 /* # rays to send at once */
151	greg	2.1	#endif
152
153			#ifndef MAX_RPROCS
154			#if (FD_SETSIZE/2-4 < 64)
155			#define MAX_NPROCS (FD_SETSIZE/2-4)
156			#else
157			#define MAX_NPROCS 64 /* max. # rendering processes */
158			#endif
159			#endif
160
161			extern char shm_boundary; / boundary of shared memory */
162
163	greg	2.3	int ray_pnprocs = 0; /* number of child processes */
164			int ray_pnidle = 0; /* number of idle children */
165	greg	2.33	int ray_pnbatch = 0; /* throughput over responsiveness? */
166	greg	2.1
167			static struct child_proc {
168	greg	2.29	RT_PID pid; /* child process id */
169	greg	2.1	int fd_send; /* write to child here */
170			int fd_recv; /* read from child here */
171			int npending; /* # rays in process */
172	greg	2.21	RNUMBER rno[RAYQLEN]; /* working on these rays */
173	greg	2.1	} r_proc[MAX_NPROCS]; /* our child processes */
174
175			static RAY r_queue[2RAYQLEN]; / ray i/o buffer */
176	greg	2.23	static int r_send_next = 0; /* next send ray placement */
177			static int r_recv_first = RAYQLEN; /* position of first unreported ray */
178			static int r_recv_next = RAYQLEN; /* next received ray placement */
179	greg	2.1
180	greg	2.28	static int samplestep = 1; /* sample step size */
181
182	greg	2.1	#define sendq_full() (r_send_next >= RAYQLEN)
183
184	schorsch	2.6	static int ray_pflush(void);
185	greg	2.13	static void ray_pchild(int fd_in, int fd_out);
186	greg	2.1
187	schorsch	2.6
188	greg	2.22	void
189	schorsch	2.6	ray_pinit( /* initialize ray-tracing processes */
190			char *otnm,
191			int nproc
192			)
193	greg	2.1	{
194			if (nobjects > 0) /* close old calculation */
195			ray_pdone(0);
196
197			ray_init(otnm); /* load the shared scene */
198
199			ray_popen(nproc); /* fork children */
200			}
201
202
203			static int
204	schorsch	2.6	ray_pflush(void) /* send queued rays to idle children */
205	greg	2.1	{
206			int nc, n, nw, i, sfirst;
207
208	schorsch	2.5	if ((ray_pnidle <= 0) \| (r_send_next <= 0))
209	greg	2.1	return(0); /* nothing we can send */
210
211			sfirst = 0; /* divvy up labor */
212	greg	2.3	nc = ray_pnidle;
213			for (i = ray_pnprocs; nc && i--; ) {
214	greg	2.1	if (r_proc[i].npending > 0)
215			continue; /* child looks busy */
216	greg	2.33	n = r_send_next - sfirst;
217			if (ray_pnbatch)
218			nc--; /* maximize bundling for batch calc */
219			else
220			n /= nc--; /* distribute work for interactivity */
221	greg	2.1	if (!n)
222			continue;
223			/* smuggle set size in crtype */
224			r_queue[sfirst].crtype = n;
225			nw = writebuf(r_proc[i].fd_send, (char *)&r_queue[sfirst],
226			sizeof(RAY)*n);
227			if (nw != sizeof(RAY)*n)
228			return(-1); /* write error */
229			r_proc[i].npending = n;
230			while (n--) /* record ray IDs */
231			r_proc[i].rno[n] = r_queue[sfirst+n].rno;
232			sfirst += r_proc[i].npending;
233	greg	2.3	ray_pnidle--; /* now she's busy */
234	greg	2.1	}
235			if (sfirst != r_send_next)
236	greg	2.24	error(CONSISTENCY, "code screwup in ray_pflush()");
237	greg	2.1	r_send_next = 0;
238			return(sfirst); /* return total # sent */
239			}
240
241
242	greg	2.25	int
243	schorsch	2.6	ray_psend( /* add a ray to our send queue */
244			RAY *r
245			)
246	greg	2.1	{
247	greg	2.25	int rv;
248
249			if ((r == NULL) \| (ray_pnidle <= 0))
250			return(0);
251	greg	2.1	/* flush output if necessary */
252	greg	2.25	if (sendq_full() && (rv = ray_pflush()) <= 0)
253			return(rv);
254	greg	2.1
255	greg	2.14	r_queue[r_send_next++] = *r;
256	greg	2.25	return(1);
257	greg	2.1	}
258
259
260	greg	2.22	int
261	schorsch	2.6	ray_pqueue( /* queue a ray for computation */
262			RAY *r
263			)
264	greg	2.1	{
265			if (r == NULL)
266			return(0);
267			/* check for full send queue */
268			if (sendq_full()) {
269	greg	2.19	RAY mySend = *r;
270	greg	2.1	/* wait for a result */
271	greg	2.19	if (ray_presult(r, 0) <= 0)
272			return(-1);
273	greg	2.1	/* put new ray in queue */
274	greg	2.14	r_queue[r_send_next++] = mySend;
275	greg	2.25
276	greg	2.19	return(1);
277	greg	2.1	}
278	greg	2.13	/* else add ray to send queue */
279	greg	2.14	r_queue[r_send_next++] = *r;
280	greg	2.1	/* check for returned ray... */
281			if (r_recv_first >= r_recv_next)
282			return(0);
283			/* ...one is sitting in queue */
284	greg	2.14	*r = r_queue[r_recv_first++];
285	greg	2.1	return(1);
286			}
287
288
289	greg	2.22	int
290	schorsch	2.6	ray_presult( /* check for a completed ray */
291			RAY *r,
292			int poll
293			)
294	greg	2.1	{
295			static struct timeval tpoll; /* zero timeval struct */
296			static fd_set readset, errset;
297			int n, ok;
298	greg	2.30	int pn;
299	greg	2.1
300			if (r == NULL)
301			return(0);
302			/* check queued results first */
303			if (r_recv_first < r_recv_next) {
304	greg	2.14	*r = r_queue[r_recv_first++];
305	greg	2.1	return(1);
306			}
307	greg	2.23	if (poll < 0) /* immediate polling mode? */
308			return(0);
309
310	greg	2.3	n = ray_pnprocs - ray_pnidle; /* pending before flush? */
311	greg	2.1
312			if (ray_pflush() < 0) /* send new rays to process */
313			return(-1);
314			/* reset receive queue */
315			r_recv_first = r_recv_next = RAYQLEN;
316
317			if (!poll) /* count newly sent unless polling */
318	greg	2.3	n = ray_pnprocs - ray_pnidle;
319	greg	2.1	if (n <= 0) /* return if nothing to await */
320			return(0);
321	greg	2.16	if (!poll && ray_pnprocs == 1) /* one process -> skip select() */
322			FD_SET(r_proc[0].fd_recv, &readset);
323
324	greg	2.1	getready: /* any children waiting for us? */
325	greg	2.3	for (pn = ray_pnprocs; pn--; )
326	greg	2.1	if (FD_ISSET(r_proc[pn].fd_recv, &readset) \|\|
327			FD_ISSET(r_proc[pn].fd_recv, &errset))
328			break;
329	greg	2.22	/* call select() if we must */
330	greg	2.1	if (pn < 0) {
331			FD_ZERO(&readset); FD_ZERO(&errset); n = 0;
332	greg	2.3	for (pn = ray_pnprocs; pn--; ) {
333	greg	2.1	if (r_proc[pn].npending > 0)
334			FD_SET(r_proc[pn].fd_recv, &readset);
335			FD_SET(r_proc[pn].fd_recv, &errset);
336			if (r_proc[pn].fd_recv >= n)
337			n = r_proc[pn].fd_recv + 1;
338			}
339			/* find out who is ready */
340			while ((n = select(n, &readset, (fd_set *)NULL, &errset,
341			poll ? &tpoll : (struct timeval *)NULL)) < 0)
342			if (errno != EINTR) {
343			error(WARNING,
344	greg	2.24	"select call failed in ray_presult()");
345	greg	2.1	ray_pclose(0);
346			return(-1);
347			}
348			if (n > 0) /* go back and get it */
349			goto getready;
350			return(0); /* else poll came up empty */
351			}
352			if (r_recv_next + r_proc[pn].npending > sizeof(r_queue)/sizeof(RAY))
353			error(CONSISTENCY, "buffer shortage in ray_presult()");
354
355			/* read rendered ray data */
356			n = readbuf(r_proc[pn].fd_recv, (char *)&r_queue[r_recv_next],
357			sizeof(RAY)*r_proc[pn].npending);
358			if (n > 0) {
359			r_recv_next += n/sizeof(RAY);
360			ok = (n == sizeof(RAY)*r_proc[pn].npending);
361			} else
362			ok = 0;
363			/* reset child's status */
364			FD_CLR(r_proc[pn].fd_recv, &readset);
365			if (n <= 0)
366			FD_CLR(r_proc[pn].fd_recv, &errset);
367			r_proc[pn].npending = 0;
368	greg	2.3	ray_pnidle++;
369	greg	2.1	/* check for rendering errors */
370			if (!ok) {
371			ray_pclose(0); /* process died -- clean up */
372			return(-1);
373			}
374			/* preen returned rays */
375			for (n = r_recv_next - r_recv_first; n--; ) {
376	greg	2.30	RAY *rp = &r_queue[r_recv_first + n];
377	greg	2.1	rp->rno = r_proc[pn].rno[n];
378			rp->parent = NULL;
379			rp->newcset = rp->clipset = NULL;
380			rp->rox = NULL;
381			rp->slights = NULL;
382			}
383			/* return first ray received */
384	greg	2.13	*r = r_queue[r_recv_first++];
385	greg	2.1	return(1);
386			}
387
388
389	greg	2.22	void
390	schorsch	2.6	ray_pdone( /* reap children and free data */
391			int freall
392			)
393	greg	2.1	{
394			ray_pclose(0); /* close child processes */
395
396			if (shm_boundary != NULL) { /* clear shared memory boundary */
397			free((void *)shm_boundary);
398			shm_boundary = NULL;
399			}
400	greg	2.23
401	greg	2.1	ray_done(freall); /* free rendering data */
402			}
403
404
405			static void
406	schorsch	2.6	ray_pchild( /* process rays (never returns) */
407			int fd_in,
408			int fd_out
409			)
410	greg	2.1	{
411			int n;
412	greg	2.30	int i;
413	greg	2.15	/* flag child process for quit() */
414			ray_pnprocs = -1;
415	greg	2.1	/* read each ray request set */
416			while ((n = read(fd_in, (char *)r_queue, sizeof(r_queue))) > 0) {
417			int n2;
418	greg	2.12	if (n < sizeof(RAY))
419	greg	2.1	break;
420			/* get smuggled set length */
421	greg	2.12	n2 = sizeof(RAY)*r_queue[0].crtype - n;
422	greg	2.1	if (n2 < 0)
423	greg	2.24	error(INTERNAL, "buffer over-read in ray_pchild()");
424	greg	2.1	if (n2 > 0) { /* read the rest of the set */
425	greg	2.12	i = readbuf(fd_in, (char *)r_queue + n, n2);
426			if (i != n2)
427	greg	2.1	break;
428			n += n2;
429			}
430	greg	2.12	n /= sizeof(RAY);
431	greg	2.1	/* evaluate rays */
432			for (i = 0; i < n; i++) {
433			r_queue[i].crtype = r_queue[i].rtype;
434			r_queue[i].parent = NULL;
435			r_queue[i].clipset = NULL;
436			r_queue[i].slights = NULL;
437	greg	2.21	r_queue[i].rlvl = 0;
438	greg	2.28	samplendx += samplestep;
439	greg	2.1	rayclear(&r_queue[i]);
440			rayvalue(&r_queue[i]);
441			}
442			/* write back our results */
443			i = writebuf(fd_out, (char )r_queue, sizeof(RAY)n);
444			if (i != sizeof(RAY)*n)
445	greg	2.24	error(SYSTEM, "write error in ray_pchild()");
446	greg	2.1	}
447			if (n)
448	greg	2.24	error(SYSTEM, "read error in ray_pchild()");
449	greg	2.1	ambsync();
450			quit(0); /* normal exit */
451			}
452
453
454	greg	2.22	void
455	schorsch	2.6	ray_popen( /* open the specified # processes */
456			int nadd
457			)
458	greg	2.1	{
459			/* check if our table has room */
460	greg	2.3	if (ray_pnprocs + nadd > MAX_NPROCS)
461			nadd = MAX_NPROCS - ray_pnprocs;
462	greg	2.1	if (nadd <= 0)
463			return;
464	greg	2.13	ambsync(); /* load any new ambient values */
465	greg	2.20	if (shm_boundary == NULL) { /* first child process? */
466			preload_objs(); /* preload auxiliary data */
467			/* set shared memory boundary */
468			shm_boundary = (char *)malloc(16);
469			strcpy(shm_boundary, "SHM_BOUNDARY");
470			}
471	greg	2.13	fflush(NULL); /* clear pending output */
472	greg	2.28	samplestep = ray_pnprocs + nadd;
473	greg	2.1	while (nadd--) { /* fork each new process */
474			int p0[2], p1[2];
475			if (pipe(p0) < 0 \|\| pipe(p1) < 0)
476			error(SYSTEM, "cannot create pipe");
477	greg	2.3	if ((r_proc[ray_pnprocs].pid = fork()) == 0) {
478	greg	2.1	int pn; /* close others' descriptors */
479	greg	2.3	for (pn = ray_pnprocs; pn--; ) {
480	greg	2.1	close(r_proc[pn].fd_send);
481			close(r_proc[pn].fd_recv);
482			}
483			close(p0[0]); close(p1[1]);
484	greg	2.24	close(0); /* don't share stdin */
485	greg	2.1	/* following call never returns */
486			ray_pchild(p1[0], p0[1]);
487			}
488	greg	2.3	if (r_proc[ray_pnprocs].pid < 0)
489	greg	2.1	error(SYSTEM, "cannot fork child process");
490			close(p1[0]); close(p0[1]);
491	greg	2.28	if (rand_samp) /* decorrelate random sequence */
492			srandom(random());
493			else
494			samplendx++;
495	greg	2.9	/*
496			* Close write stream on exec to avoid multiprocessing deadlock.
497			* No use in read stream without it, so set flag there as well.
498			*/
499			fcntl(p1[1], F_SETFD, FD_CLOEXEC);
500			fcntl(p0[0], F_SETFD, FD_CLOEXEC);
501	greg	2.3	r_proc[ray_pnprocs].fd_send = p1[1];
502			r_proc[ray_pnprocs].fd_recv = p0[0];
503			r_proc[ray_pnprocs].npending = 0;
504			ray_pnprocs++;
505			ray_pnidle++;
506	greg	2.1	}
507			}
508
509
510	greg	2.22	void
511	schorsch	2.6	ray_pclose( /* close one or more child processes */
512			int nsub
513			)
514	greg	2.1	{
515			static int inclose = 0;
516	greg	2.29	RAY res;
517			int i, status = 0;
518	greg	2.1	/* check recursion */
519			if (inclose)
520			return;
521			inclose++;
522	greg	2.26	/* check no child / in child */
523			if (ray_pnprocs <= 0)
524			return;
525	greg	2.1	/* check argument */
526	schorsch	2.5	if ((nsub <= 0) \| (nsub > ray_pnprocs))
527	greg	2.3	nsub = ray_pnprocs;
528	greg	2.1	/* clear our ray queue */
529			while (ray_presult(&res,0) > 0)
530			;
531	greg	2.23	r_send_next = 0; /* hard reset in case of error */
532			r_recv_first = r_recv_next = RAYQLEN;
533	greg	2.29	/* close send pipes */
534			for (i = ray_pnprocs-nsub; i < ray_pnprocs; i++)
535			close(r_proc[i].fd_send);
536
537			if (nsub == 1) { /* awaiting single process? */
538			if (waitpid(r_proc[ray_pnprocs-1].pid, &status, 0) < 0)
539	greg	2.8	status = 127<<8;
540	greg	2.29	close(r_proc[ray_pnprocs-1].fd_recv);
541			} else /* else unordered wait */
542			for (i = 0; i < nsub; ) {
543			int j, mystatus;
544			RT_PID pid = wait(&mystatus);
545			for (j = ray_pnprocs-nsub; j < ray_pnprocs; j++)
546			if (r_proc[j].pid == pid) {
547			if (mystatus)
548			status = mystatus;
549			close(r_proc[j].fd_recv);
550			++i;
551			}
552	greg	2.1	}
553	greg	2.29	ray_pnprocs -= nsub;
554			ray_pnidle -= nsub;
555			if (status) {
556			sprintf(errmsg, "rendering process exited with code %d", status>>8);
557			error(WARNING, errmsg);
558	greg	2.1	}
559			inclose--;
560			}