--- ray/src/rt/raypcalls.c	2004/09/17 14:27:33	2.7
+++ ray/src/rt/raypcalls.c	2008/02/20 05:21:29	2.19
@@ -1,5 +1,5 @@
 #ifndef lint
-static const char	RCSid[] = "$Id: raypcalls.c,v 2.7 2004/09/17 14:27:33 greg Exp $";
+static const char	RCSid[] = "$Id: raypcalls.c,v 2.19 2008/02/20 05:21:29 greg Exp $";
 #endif
 /*
  *  raypcalls.c - interface for parallel rendering using Radiance
@@ -23,14 +23,17 @@ static const char	RCSid[] = "$Id: raypcalls.c,v 2.7 20
  *  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.  This permits all sorts of queries that
- *  wouldn't be possible otherwise, without causing any real
+ *  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 then traced using a simple
  *  queuing mechanism, explained below.
  *
- *  The ray queue holds as many rays as there are rendering
- *  processes.  Rays are queued and returned by a single
+ *  The ray queue buffers RAYQLEN rays before sending to
+ *  children, each of which may internally buffer RAYQLEN rays.
+ *
+ *  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
@@ -43,7 +46,7 @@ static const char	RCSid[] = "$Id: raypcalls.c,v 2.7 20
  *	myRay.rorg = ( ray origin point )
  *	myRay.rdir = ( normalized ray direction )
  *	myRay.rmax = ( maximum length, or zero for no limit )
- *	rayorigin(&myRay, NULL, PRIMARY, 1.0);
+ *	rayorigin(&myRay, PRIMARY, NULL, NULL);
  *	myRay.rno = ( my personal ray identifier )
  *	if (ray_pqueue(&myRay) == 1)
  *		{ do something with results }
@@ -105,7 +108,7 @@ static const char	RCSid[] = "$Id: raypcalls.c,v 2.7 20
  *  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
+ *  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().
  *
@@ -128,17 +131,15 @@ static const char	RCSid[] = "$Id: raypcalls.c,v 2.7 20
  *  process should not be compromised.
  */
 
-#include <stdio.h>
-#include <sys/types.h>
-#include <sys/wait.h> /* XXX platform */
-
 #include  "rtprocess.h"
 #include  "ray.h"
 #include  "ambient.h"
+#include  <sys/types.h>
+#include  <sys/wait.h>
 #include  "selcall.h"
 
 #ifndef RAYQLEN
-#define RAYQLEN		16		/* # rays to send at once */
+#define RAYQLEN		12		/* # rays to send at once */
 #endif
 
 #ifndef MAX_RPROCS
@@ -170,7 +171,7 @@ static int	r_recv_next;		/* next receive ray placement
 #define sendq_full()	(r_send_next >= RAYQLEN)
 
 static int ray_pflush(void);
-static void ray_pchild(int	fd_in, int	fd_out);
+static void ray_pchild(int fd_in, int fd_out);
 
 
 extern void
@@ -243,8 +244,7 @@ ray_psend(			/* add a ray to our send queue */
 	if (sendq_full() && ray_pflush() <= 0)
 		error(INTERNAL, "ray_pflush failed in ray_psend");
 
-	r_queue[r_send_next] = *r;
-	r_send_next++;
+	r_queue[r_send_next++] = *r;
 }
 
 
@@ -257,25 +257,22 @@ ray_pqueue(			/* queue a ray for computation */
 		return(0);
 					/* check for full send queue */
 	if (sendq_full()) {
-		RAY	mySend;
-		int	rval;
-		mySend = *r;
+		RAY	mySend = *r;
 					/* wait for a result */
-		rval = ray_presult(r, 0);
+		if (ray_presult(r, 0) <= 0)
+			return(-1);
 					/* put new ray in queue */
-		r_queue[r_send_next] = mySend;
-		r_send_next++;
-		return(rval);		/* done */
+		r_queue[r_send_next++] = mySend;
+				/* XXX r_send_next may now be > RAYQLEN */
+		return(1);
 	}
-					/* add ray to send queue */
-	r_queue[r_send_next] = *r;
-	r_send_next++;
+					/* 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];
-	r_recv_first++;
+	*r = r_queue[r_recv_first++];
 	return(1);
 }
 
@@ -295,8 +292,7 @@ ray_presult(		/* check for a completed ray */
 		return(0);
 					/* check queued results first */
 	if (r_recv_first < r_recv_next) {
-		*r = r_queue[r_recv_first];
-		r_recv_first++;
+		*r = r_queue[r_recv_first++];
 		return(1);
 	}
 	n = ray_pnprocs - ray_pnidle;	/* pending before flush? */
@@ -310,6 +306,9 @@ ray_presult(		/* check for a completed ray */
 		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) ||
@@ -370,8 +369,7 @@ getready:				/* any children waiting for us? */
 		rp->slights = NULL;
 	}
 					/* return first ray received */
-	*r = r_queue[r_recv_first];
-	r_recv_first++;
+	*r = r_queue[r_recv_first++];
 	return(1);
 }
 
@@ -399,30 +397,30 @@ ray_pchild(	/* process rays (never returns) */
 {
 	int	n;
 	register 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))
+		if (n < sizeof(RAY))
 			break;
-		n /= sizeof(RAY);
 					/* get smuggled set length */
-		n2 = r_queue[0].crtype - n;
+		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),
-					sizeof(RAY)*n2);
-			if (i != sizeof(RAY)*n2)
+			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].revf = raytrace;
 			samplendx++;
 			rayclear(&r_queue[i]);
 			rayvalue(&r_queue[i]);
@@ -449,8 +447,8 @@ ray_popen(			/* open the specified # processes */
 		nadd = MAX_NPROCS - ray_pnprocs;
 	if (nadd <= 0)
 		return;
-	fflush(stderr);			/* clear pending output */
-	fflush(stdout);
+	ambsync();			/* load any new ambient values */
+	fflush(NULL);			/* clear pending output */
 	while (nadd--) {		/* fork each new process */
 		int	p0[2], p1[2];
 		if (pipe(p0) < 0 || pipe(p1) < 0)
@@ -468,6 +466,12 @@ ray_popen(			/* open the specified # processes */
 		if (r_proc[ray_pnprocs].pid < 0)
 			error(SYSTEM, "cannot fork child process");
 		close(p1[0]); close(p0[1]);
+		/*
+		 * 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;
@@ -500,8 +504,8 @@ ray_pclose(		/* close one or more child processes */
 		ray_pnprocs--;
 		close(r_proc[ray_pnprocs].fd_recv);
 		close(r_proc[ray_pnprocs].fd_send);
-		while (wait(&status) != r_proc[ray_pnprocs].pid)
-			;
+		if (waitpid(r_proc[ray_pnprocs].pid, &status, 0) < 0)
+			status = 127<<8;
 		if (status) {
 			sprintf(errmsg,
 				"rendering process %d exited with code %d",
@@ -518,5 +522,7 @@ void
 quit(ec)			/* make sure exit is called */
 int	ec;
 {
+	if (ray_pnprocs > 0)	/* close children if any */
+		ray_pclose(0);		
 	exit(ec);
 }