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

# Content
1 #ifndef lint
2 static const char RCSid[] = "$Id: rc3.c,v 2.3 2012/06/10 05:25:42 greg Exp $";
3 #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 int nadded; /* accumulated so far */
18 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 static FILE *inq_fp[MAXPROCESS]; /* input streams */
27
28
29 /* Get new bin queue entry */
30 static BINQ *
31 new_binq()
32 {
33 BINQ *bp;
34 int i;
35
36 if (free_bq != NULL) { /* something already available? */
37 bp = free_bq;
38 free_bq = bp->next;
39 bp->next = NULL;
40 bp->nadded = 1;
41 return(bp);
42 }
43 /* else allocate fresh */
44 bp = (BINQ *)malloc(sizeof(BINQ) + sizeof(MODCONT *)*(nmods-1));
45 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 bp->nadded = 1;
58 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 /* Sum one modifier record into another (updates nadded) */
113 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 dst->nadded += src->nadded;
125 }
126
127
128 /* Queue values for later output */
129 static void
130 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 return;
143 }
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
149 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 if (accumulate == 1) /* no accumulation? */
157 return;
158 b_cur = out_bq; /* else merge accumulation entries */
159 while (b_cur->next != NULL) {
160 if (b_cur->nadded >= accumulate ||
161 (b_cur->ndx-1)/accumulate !=
162 (b_cur->next->ndx-1)/accumulate) {
163 b_cur = b_cur->next;
164 continue;
165 }
166 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 }
172 }
173
174
175 /* 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 static int
196 output_catchup(int nmax)
197 {
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 while (out_bq != NULL && out_bq->nadded >= accumulate
206 && out_bq->ndx == lastdone+1) {
207 if ((nmax > 0) & (nout >= nmax))
208 break;
209 bp = out_bq; /* pop off first entry */
210 out_bq = bp->next;
211 bp->next = NULL;
212 for (i = 0; i < nmods; i++) /* output record */
213 mod_output(bp->mca[i]);
214 end_record();
215 free_binq(bp); /* free this entry */
216 lastdone += accumulate;
217 ++nout;
218 }
219 return(nout);
220 }
221
222
223 /* Put a zero record in results queue & output */
224 void
225 put_zero_record(int ndx)
226 {
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 output_catchup(0);
235 }
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 errno = 0;
262 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 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 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 /* connect parent's pipes */
287 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 inq_fp[nchild] = fdopen(p1[0], "rb");
293 if (inq_fp[nchild] == NULL)
294 error(SYSTEM, "out of memory in in_rchild()");
295 #ifdef getc_unlocked
296 flockfile(inq_fp[nchild]); /* avoid mutex overhead */
297 #endif
298 ++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 while (nchild-- > 0) {
312 kida[nchild].r = -1; /* close(-1) error is ignored */
313 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 fclose(inq_fp[nchild]); /* performs actual close() */
320 }
321 }
322
323
324 /* Wait for the next available child, managing output queue as well */
325 static int
326 next_child_nq(int force_wait)
327 {
328 static struct timeval polling;
329 struct timeval *pmode;
330 fd_set readset, errset;
331 int i, j, n, nr, nqr;
332
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
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 /* 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 if (!nr) /* nothing to wait for? */
362 return(-1);
363 if ((nr > 1) | (pmode == &polling)) {
364 errno = 0;
365 i = select(n, &readset, NULL, &errset, pmode);
366 if (!i) {
367 pmode = NULL; /* try again, blocking this time */
368 goto tryagain;
369 }
370 if (i < 0)
371 error(SYSTEM, "select() error in next_child_nq()");
372 } 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 nr = bq->mca[j]->nbins;
386 if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), nr,
387 inq_fp[i]) != nr)
388 error(SYSTEM, "read error from render process");
389 }
390 queue_output(bq); /* add results to output queue */
391 kida[i].running = -1; /* mark child as available */
392 n = i;
393 }
394 return(n); /* first available child */
395 }
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 int i;
406 /* 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 ; /* clear the queue */
422 lastdone = lastray = 0;
423 }
424 if (d == 0.0) {
425 if ((yres <= 0) | (xres <= 0))
426 waitflush = 1; /* flush right after */
427 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 error(SYSTEM, "pipe write error");
433 kida[i].running = ++lastray;
434 }
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 free_binq(out_bq);
450 out_bq = NULL;
451 }
452 if (raysleft)
453 error(USER, "unexpected EOF on input");
454 free_binq(NULL); /* clean up */
455 lu_done(&ofiletab);
456 }