ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/rt/rc3.c
Revision: 2.4
Committed: Mon Jun 11 05:07:55 2012 UTC (11 years, 10 months ago) by greg
Content type: text/plain
Branch: MAIN
Changes since 2.3: +59 -23 lines
Log Message:
Fixed a number of bugs -- still testing

File Contents

# User Rev Content
1 greg 2.1 #ifndef lint
2 greg 2.4 static const char RCSid[] = "$Id: rc3.c,v 2.3 2012/06/10 05:25:42 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     }
110    
111    
112 greg 2.4 /* Sum one modifier record into another (updates nadded) */
113 greg 2.1 static void
114     add_modbin(BINQ *dst, BINQ *src)
115     {
116     int i, j;
117    
118     for (i = nmods; i--; ) {
119     MODCONT *mpin = src->mca[i];
120     MODCONT *mpout = dst->mca[i];
121     for (j = mpout->nbins; j--; )
122     addcolor(mpout->cbin[j], mpin->cbin[j]);
123     }
124 greg 2.4 dst->nadded += src->nadded;
125 greg 2.1 }
126    
127    
128 greg 2.2 /* Queue values for later output */
129     static void
130 greg 2.1 queue_output(BINQ *bp)
131     {
132     BINQ *b_last, *b_cur;
133    
134     if (accumulate <= 0) { /* just accumulating? */
135     if (out_bq == NULL) {
136     bp->next = NULL;
137     out_bq = bp;
138     } else {
139     add_modbin(out_bq, bp);
140     free_binq(bp);
141     }
142 greg 2.2 return;
143 greg 2.1 }
144     b_last = NULL; /* else insert in output queue */
145     for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
146     b_cur = b_cur->next)
147     b_last = b_cur;
148 greg 2.4
149 greg 2.1 if (b_last != NULL) {
150     bp->next = b_cur;
151     b_last->next = bp;
152     } else {
153     bp->next = out_bq;
154     out_bq = bp;
155     }
156 greg 2.3 if (accumulate == 1) /* no accumulation? */
157 greg 2.2 return;
158     b_cur = out_bq; /* else merge accumulation entries */
159     while (b_cur->next != NULL) {
160 greg 2.4 if (b_cur->nadded >= accumulate ||
161 greg 2.2 (b_cur->ndx-1)/accumulate !=
162     (b_cur->next->ndx-1)/accumulate) {
163     b_cur = b_cur->next;
164     continue;
165 greg 2.1 }
166 greg 2.2 add_modbin(b_cur, b_cur->next);
167     b_last = b_cur->next;
168     b_cur->next = b_last->next;
169     b_last->next = NULL;
170     free_binq(b_last);
171 greg 2.1 }
172 greg 2.2 }
173    
174    
175 greg 2.4 /* Count number of records ready for output */
176     static int
177     queue_ready()
178     {
179     int nready = 0;
180     BINQ *bp;
181    
182     if (accumulate <= 0) /* just accumulating? */
183     return(0);
184    
185     for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
186     bp->ndx == lastdone+nready*accumulate+1;
187     bp = bp->next)
188     ++nready;
189    
190     return(nready);
191     }
192    
193    
194     /* Catch up with output queue by producing ready results */
195 greg 2.2 static int
196 greg 2.4 output_catchup(int nmax)
197 greg 2.2 {
198     int nout = 0;
199     BINQ *bp;
200     int i;
201    
202     if (accumulate <= 0) /* just accumulating? */
203     return(0);
204     /* else output ready results */
205 greg 2.4 while (out_bq != NULL && out_bq->nadded >= accumulate
206     && out_bq->ndx == lastdone+1) {
207     if ((nmax > 0) & (nout >= nmax))
208     break;
209 greg 2.2 bp = out_bq; /* pop off first entry */
210     out_bq = bp->next;
211     bp->next = NULL;
212 greg 2.1 for (i = 0; i < nmods; i++) /* output record */
213 greg 2.2 mod_output(bp->mca[i]);
214 greg 2.1 end_record();
215 greg 2.2 free_binq(bp); /* free this entry */
216 greg 2.1 lastdone += accumulate;
217     ++nout;
218     }
219     return(nout);
220     }
221    
222    
223 greg 2.3 /* Put a zero record in results queue & output */
224 greg 2.1 void
225 greg 2.3 put_zero_record(int ndx)
226 greg 2.1 {
227     BINQ *bp = new_binq();
228     int i;
229    
230     for (i = nmods; i--; )
231     memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
232     bp->ndx = ndx;
233     queue_output(bp);
234 greg 2.4 output_catchup(0);
235 greg 2.1 }
236    
237    
238     /* callback to set output spec to NULL (stdout) */
239     static int
240     set_stdout(const LUENT *le, void *p)
241     {
242     (*(MODCONT *)le->data).outspec = NULL;
243     return(0);
244     }
245    
246    
247     /* Start child processes if we can */
248     int
249     in_rchild()
250     {
251     #ifdef _WIN32
252     error(WARNING, "multiprocessing unsupported -- running solo");
253     nproc = 1;
254     return(1);
255     #else
256     /* try to fork ourselves */
257     while (nchild < nproc) {
258     int p0[2], p1[2];
259     int pid;
260     /* prepare i/o pipes */
261 greg 2.3 errno = 0;
262 greg 2.1 if (pipe(p0) < 0 || pipe(p1) < 0)
263     error(SYSTEM, "pipe() call failed!");
264     pid = fork(); /* fork parent process */
265     if (pid == 0) { /* if in child, set up & return true */
266     close(p0[1]); close(p1[0]);
267 greg 2.4 lu_doall(&modconttab, set_stdout, NULL);
268     lu_done(&ofiletab);
269     while (nchild--) {
270     close(kida[nchild].w);
271     fclose(inq_fp[nchild]);
272     }
273 greg 2.1 dup2(p0[0], 0); close(p0[0]);
274     dup2(p1[1], 1); close(p1[1]);
275     inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
276     outfmt = 'd';
277     header = 0;
278     waitflush = xres = 1;
279     yres = 0;
280     raysleft = 0;
281     account = accumulate = 1;
282     return(1); /* child return value */
283     }
284     if (pid < 0)
285     error(SYSTEM, "fork() call failed!");
286 greg 2.3 /* connect parent's pipes */
287 greg 2.1 close(p0[0]); close(p1[1]);
288     kida[nchild].r = p1[0];
289     kida[nchild].w = p0[1];
290     kida[nchild].pid = pid;
291     kida[nchild].running = -1;
292 greg 2.2 inq_fp[nchild] = fdopen(p1[0], "rb");
293     if (inq_fp[nchild] == NULL)
294     error(SYSTEM, "out of memory in in_rchild()");
295 greg 2.3 #ifdef getc_unlocked
296     flockfile(inq_fp[nchild]); /* avoid mutex overhead */
297     #endif
298 greg 2.1 ++nchild;
299     }
300     return(0); /* parent return value */
301     #endif
302     }
303    
304    
305     /* Close child processes */
306     void
307     end_children()
308     {
309     int status;
310    
311 greg 2.2 while (nchild-- > 0) {
312     kida[nchild].r = -1; /* close(-1) error is ignored */
313 greg 2.1 if ((status = close_process(&kida[nchild])) > 0) {
314     sprintf(errmsg,
315     "rendering process returned bad status (%d)",
316     status);
317     error(WARNING, errmsg);
318     }
319 greg 2.2 fclose(inq_fp[nchild]); /* performs actual close() */
320     }
321 greg 2.1 }
322    
323    
324 greg 2.2 /* Wait for the next available child, managing output queue as well */
325 greg 2.1 static int
326     next_child_nq(int force_wait)
327     {
328 greg 2.2 static struct timeval polling;
329 greg 2.4 struct timeval *pmode;
330 greg 2.2 fd_set readset, errset;
331 greg 2.4 int i, j, n, nr, nqr;
332 greg 2.1
333     if (!force_wait) /* see if there's one free */
334     for (i = nchild; i--; )
335     if (kida[i].running < 0)
336     return(i);
337 greg 2.4
338     nqr = queue_ready(); /* wait mode or polling? */
339     if (!nqr | force_wait | (accumulate <= 0))
340     pmode = NULL;
341     else
342     pmode = &polling;
343     tryagain:
344     n = 0; /* catch up with output? */
345     if ((pmode == &polling) & (nqr > nchild))
346     n = nqr - nchild;
347     if ((pmode == NULL) & (nqr > 0) | (n > 0))
348     nqr -= output_catchup(n);
349 greg 2.1 /* prepare select() call */
350     FD_ZERO(&readset); FD_ZERO(&errset);
351     n = nr = 0;
352     for (i = nchild; i--; ) {
353     if (kida[i].running > 0) {
354     FD_SET(kida[i].r, &readset);
355     ++nr;
356     }
357     FD_SET(kida[i].r, &errset);
358     if (kida[i].r >= n)
359     n = kida[i].r + 1;
360     }
361 greg 2.3 if (!nr) /* nothing to wait for? */
362     return(-1);
363 greg 2.2 if ((nr > 1) | (pmode == &polling)) {
364 greg 2.1 errno = 0;
365 greg 2.3 i = select(n, &readset, NULL, &errset, pmode);
366     if (!i) {
367 greg 2.2 pmode = NULL; /* try again, blocking this time */
368     goto tryagain;
369     }
370 greg 2.3 if (i < 0)
371     error(SYSTEM, "select() error in next_child_nq()");
372 greg 2.1 } else
373     FD_ZERO(&errset);
374     n = -1; /* read results from child(ren) */
375     for (i = nchild; i--; ) {
376     BINQ *bq;
377     if (FD_ISSET(kida[i].r, &errset))
378     error(USER, "rendering process died");
379     if (!FD_ISSET(kida[i].r, &readset))
380     continue;
381     bq = new_binq(); /* get results holder */
382     bq->ndx = kida[i].running;
383     /* read from child */
384     for (j = 0; j < nmods; j++) {
385 greg 2.3 nr = bq->mca[j]->nbins;
386     if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), nr,
387     inq_fp[i]) != nr)
388 greg 2.1 error(SYSTEM, "read error from render process");
389     }
390 greg 2.3 queue_output(bq); /* add results to output queue */
391 greg 2.1 kida[i].running = -1; /* mark child as available */
392     n = i;
393     }
394 greg 2.3 return(n); /* first available child */
395 greg 2.1 }
396    
397    
398     /* Run parental oversight loop */
399     void
400     parental_loop()
401     {
402     static int ignore_warning_given = 0;
403     FVECT orgdir[2];
404     double d;
405 greg 2.3 int i;
406 greg 2.1 /* load rays from stdin & process */
407     #ifdef getc_unlocked
408     flockfile(stdin); /* avoid lock/unlock overhead */
409     #endif
410     while (getvec(orgdir[0]) == 0 && getvec(orgdir[1]) == 0) {
411     d = normalize(orgdir[1]);
412     /* asking for flush? */
413     if ((d == 0.0) & (accumulate != 1)) {
414     if (!ignore_warning_given++)
415     error(WARNING,
416     "dummy ray(s) ignored during accumulation\n");
417     continue;
418     }
419     if ((d == 0.0) | (lastray+1 < lastray)) {
420     while (next_child_nq(1) >= 0)
421 greg 2.3 ; /* clear the queue */
422 greg 2.1 lastdone = lastray = 0;
423     }
424     if (d == 0.0) {
425     if ((yres <= 0) | (xres <= 0))
426     waitflush = 1; /* flush right after */
427 greg 2.3 put_zero_record(++lastray);
428     } else { /* else assign ray */
429     i = next_child_nq(0);
430     if (writebuf(kida[i].w, (char *)orgdir,
431     sizeof(orgdir)) != sizeof(orgdir))
432 greg 2.1 error(SYSTEM, "pipe write error");
433 greg 2.3 kida[i].running = ++lastray;
434 greg 2.1 }
435     if (raysleft && !--raysleft)
436     break; /* preemptive EOI */
437     }
438     while (next_child_nq(1) >= 0) /* empty results queue */
439     ;
440     /* output accumulated record */
441     if (accumulate <= 0 || account < accumulate) {
442     if (account < accumulate) {
443     error(WARNING, "partial accumulation in final record");
444     accumulate -= account;
445     }
446     for (i = 0; i < nmods; i++)
447     mod_output(out_bq->mca[i]);
448     end_record();
449 greg 2.3 free_binq(out_bq);
450     out_bq = NULL;
451 greg 2.1 }
452     if (raysleft)
453     error(USER, "unexpected EOF on input");
454     free_binq(NULL); /* clean up */
455     lu_done(&ofiletab);
456     }