ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/rt/rc3.c
Revision: 2.9
Committed: Thu Jun 14 05:13:25 2012 UTC (11 years, 10 months ago) by greg
Content type: text/plain
Branch: MAIN
Changes since 2.8: +15 -32 lines
Log Message:
Simplified creation of child process

File Contents

# Content
1 #ifndef lint
2 static const char RCSid[] = "$Id: rc3.c,v 2.8 2012/06/12 22:46:45 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 RNUMBER ndx; /* index for this entry */
17 RNUMBER 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 struct {
26 RNUMBER r1; /* assigned ray starting index */
27 SUBPROC pr; /* PID, i/o descriptors */
28 FILE *infp; /* file pointer to read from process */
29 int nr; /* number rays to sum (0 if free) */
30 } kida[MAXPROCESS]; /* our child processes */
31
32
33 /* Get new bin queue entry */
34 static BINQ *
35 new_binq()
36 {
37 BINQ *bp;
38 int i;
39
40 if (free_bq != NULL) { /* something already available? */
41 bp = free_bq;
42 free_bq = bp->next;
43 bp->next = NULL;
44 bp->nadded = 0;
45 return(bp);
46 }
47 /* else allocate fresh */
48 bp = (BINQ *)malloc(sizeof(BINQ) + sizeof(MODCONT *)*(nmods-1));
49 if (bp == NULL)
50 goto memerr;
51 for (i = nmods; i--; ) {
52 MODCONT *mp = (MODCONT *)lu_find(&modconttab,modname[i])->data;
53 bp->mca[i] = (MODCONT *)malloc(sizeof(MODCONT) +
54 sizeof(DCOLOR)*(mp->nbins-1));
55 if (bp->mca[i] == NULL)
56 goto memerr;
57 memcpy(bp->mca[i], mp, sizeof(MODCONT)-sizeof(DCOLOR));
58 /* memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*mp->nbins); */
59 }
60 bp->ndx = 0;
61 bp->nadded = 0;
62 bp->next = NULL;
63 return(bp);
64 memerr:
65 error(SYSTEM, "out of memory in new_binq()");
66 return(NULL);
67 }
68
69
70 /* Free a bin queue entry */
71 static void
72 free_binq(BINQ *bp)
73 {
74 int i;
75
76 if (bp == NULL) { /* signal to release our free list */
77 while ((bp = free_bq) != NULL) {
78 free_bq = bp->next;
79 for (i = nmods; i--; )
80 free(bp->mca[i]);
81 /* Note: we don't own bp->mca[i]->binv */
82 free(bp);
83 }
84 return;
85 }
86 /* clear sums for next use */
87 /* for (i = nmods; i--; )
88 memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
89 */
90 bp->ndx = 0;
91 bp->next = free_bq; /* push onto free list */
92 free_bq = bp;
93 }
94
95
96 /* Add modifier values to accumulation record in queue and clear */
97 void
98 queue_modifiers()
99 {
100 MODCONT *mpin, *mpout;
101 int i, j;
102
103 if ((accumulate > 0) | (out_bq == NULL))
104 error(CONSISTENCY, "bad call to queue_modifiers()");
105
106 for (i = nmods; i--; ) {
107 mpin = (MODCONT *)lu_find(&modconttab,modname[i])->data;
108 mpout = out_bq->mca[i];
109 for (j = mpout->nbins; j--; )
110 addcolor(mpout->cbin[j], mpin->cbin[j]);
111 memset(mpin->cbin, 0, sizeof(DCOLOR)*mpin->nbins);
112 }
113 out_bq->nadded++;
114 }
115
116
117 /* Sum one modifier record into another (updates nadded) */
118 static void
119 add_modbin(BINQ *dst, BINQ *src)
120 {
121 int i, j;
122
123 for (i = nmods; i--; ) {
124 MODCONT *mpin = src->mca[i];
125 MODCONT *mpout = dst->mca[i];
126 for (j = mpout->nbins; j--; )
127 addcolor(mpout->cbin[j], mpin->cbin[j]);
128 }
129 dst->nadded += src->nadded;
130 }
131
132
133 /* Queue values for later output */
134 static void
135 queue_output(BINQ *bp)
136 {
137 BINQ *b_last, *b_cur;
138
139 if (accumulate <= 0) { /* just accumulating? */
140 if (out_bq == NULL) {
141 bp->next = NULL;
142 out_bq = bp;
143 } else {
144 add_modbin(out_bq, bp);
145 free_binq(bp);
146 }
147 return;
148 }
149 b_last = NULL; /* else insert in output queue */
150 for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
151 b_cur = b_cur->next)
152 b_last = b_cur;
153
154 if (b_last != NULL) {
155 bp->next = b_cur;
156 b_last->next = bp;
157 } else {
158 bp->next = out_bq;
159 out_bq = bp;
160 }
161 if (accumulate == 1) /* no accumulation? */
162 return;
163 b_cur = out_bq; /* else merge accumulation entries */
164 while (b_cur->next != NULL) {
165 if (b_cur->nadded >= accumulate ||
166 (b_cur->ndx-1)/accumulate !=
167 (b_cur->next->ndx-1)/accumulate) {
168 b_cur = b_cur->next;
169 continue;
170 }
171 add_modbin(b_cur, b_cur->next);
172 b_last = b_cur->next;
173 b_cur->next = b_last->next;
174 b_last->next = NULL;
175 free_binq(b_last);
176 }
177 }
178
179
180 /* Count number of records ready for output */
181 static int
182 queue_ready()
183 {
184 int nready = 0;
185 BINQ *bp;
186
187 if (accumulate <= 0) /* just accumulating? */
188 return(0);
189
190 for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
191 bp->ndx == lastdone+nready*accumulate+1;
192 bp = bp->next)
193 ++nready;
194
195 return(nready);
196 }
197
198
199 /* Catch up with output queue by producing ready results */
200 static int
201 output_catchup(int nmax)
202 {
203 int nout = 0;
204 BINQ *bp;
205 int i;
206
207 if (accumulate <= 0) /* just accumulating? */
208 return(0);
209 /* else output ready results */
210 while (out_bq != NULL && out_bq->nadded >= accumulate
211 && out_bq->ndx == lastdone+1) {
212 if ((nmax > 0) & (nout >= nmax))
213 break;
214 bp = out_bq; /* pop off first entry */
215 out_bq = bp->next;
216 bp->next = NULL;
217 for (i = 0; i < nmods; i++) /* output record */
218 mod_output(bp->mca[i]);
219 end_record();
220 free_binq(bp); /* free this entry */
221 lastdone += accumulate;
222 ++nout;
223 }
224 return(nout);
225 }
226
227
228 /* Put a zero record in results queue & output */
229 void
230 put_zero_record(int ndx)
231 {
232 BINQ *bp = new_binq();
233 int i;
234
235 for (i = nmods; i--; )
236 memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
237 bp->ndx = ndx;
238 bp->nadded = 1;
239 queue_output(bp);
240 output_catchup(0);
241 }
242
243
244 /* Get results from child process and add to queue */
245 static void
246 queue_results(int k)
247 {
248 BINQ *bq = new_binq(); /* get results holder */
249 int j;
250
251 bq->ndx = kida[k].r1;
252 bq->nadded = kida[k].nr;
253 /* read from child */
254 for (j = 0; j < nmods; j++)
255 if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), bq->mca[j]->nbins,
256 kida[k].infp) != bq->mca[j]->nbins)
257 error(SYSTEM, "read error from render process");
258
259 queue_output(bq); /* put results in output queue */
260 kida[k].nr = 0; /* mark child as available */
261 }
262
263
264 /* callback to set output spec to NULL (stdout) */
265 static int
266 set_stdout(const LUENT *le, void *p)
267 {
268 (*(MODCONT *)le->data).outspec = NULL;
269 return(0);
270 }
271
272
273 /* Start child processes if we can */
274 int
275 in_rchild()
276 {
277 int rval;
278
279 while (nchild < nproc) { /* fork until target reached */
280 errno = 0;
281 rval = open_process(&kida[nchild].pr, NULL);
282 if (rval < 0)
283 error(SYSTEM, "open_process() call failed");
284 if (rval == 0) { /* if in child, set up & return true */
285 lu_doall(&modconttab, set_stdout, NULL);
286 lu_done(&ofiletab);
287 while (nchild--) { /* don't share other pipes */
288 close(kida[nchild].pr.w);
289 fclose(kida[nchild].infp);
290 }
291 inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
292 outfmt = 'd';
293 header = 0;
294 yres = 0;
295 raysleft = 0;
296 if (accumulate == 1) {
297 waitflush = xres = 1;
298 account = accumulate = 1;
299 } else { /* parent controls accumulation */
300 waitflush = xres = 0;
301 account = accumulate = 0;
302 }
303 return(1); /* return "true" in child */
304 }
305 if (rval != PIPE_BUF)
306 error(CONSISTENCY, "bad value from open_process()");
307 /* connect to child's output */
308 kida[nchild].infp = fdopen(kida[nchild].pr.r, "rb");
309 if (kida[nchild].infp == NULL)
310 error(SYSTEM, "out of memory in in_rchild()");
311 #ifdef getc_unlocked
312 flockfile(kida[nchild].infp); /* avoid mutex overhead */
313 #endif
314 kida[nchild++].nr = 0; /* mark as available */
315 }
316 return(0); /* return "false" in parent */
317 }
318
319
320 /* Close child processes */
321 void
322 end_children()
323 {
324 int status;
325
326 while (nchild > 0) {
327 nchild--;
328 if ((status = close_process(&kida[nchild].pr)) > 0) {
329 sprintf(errmsg,
330 "rendering process returned bad status (%d)",
331 status);
332 error(WARNING, errmsg);
333 }
334 fclose(kida[nchild].infp);
335 }
336 }
337
338
339 /* Wait for the next available child, managing output queue simultaneously */
340 static int
341 next_child_nq(int flushing)
342 {
343 static struct timeval polling;
344 struct timeval *pmode;
345 fd_set readset, errset;
346 int i, n, nr, nqr;
347
348 if (!flushing) /* see if there's one free */
349 for (i = nchild; i--; )
350 if (!kida[i].nr)
351 return(i);
352
353 nqr = queue_ready(); /* choose blocking mode or polling */
354 if ((nqr > 0) & !flushing)
355 pmode = &polling;
356 else
357 pmode = NULL;
358 tryagain: /* catch up with output? */
359 if (pmode == &polling) {
360 if (nqr > nchild) /* don't get too far behind */
361 nqr -= output_catchup(nqr-nchild);
362 } else if (nqr > 0) /* clear output before blocking */
363 nqr -= output_catchup(0);
364 /* prepare select() call */
365 FD_ZERO(&readset); FD_ZERO(&errset);
366 n = nr = 0;
367 for (i = nchild; i--; ) {
368 if (kida[i].nr) {
369 FD_SET(kida[i].pr.r, &readset);
370 ++nr;
371 }
372 FD_SET(kida[i].pr.r, &errset);
373 if (kida[i].pr.r >= n)
374 n = kida[i].pr.r + 1;
375 }
376 if (!nr) /* nothing to wait for? */
377 return(-1);
378 if ((nr > 1) | (pmode == &polling)) {
379 errno = 0;
380 nr = select(n, &readset, NULL, &errset, pmode);
381 if (!nr) {
382 pmode = NULL; /* try again, blocking this time */
383 goto tryagain;
384 }
385 if (nr < 0)
386 error(SYSTEM, "select() error in next_child_nq()");
387 } else
388 FD_ZERO(&errset);
389 n = -1; /* read results from child(ren) */
390 for (i = nchild; i--; ) {
391 if (FD_ISSET(kida[i].pr.r, &errset))
392 error(USER, "rendering process died");
393 if (FD_ISSET(kida[i].pr.r, &readset))
394 queue_results(n = i);
395 }
396 return(n); /* first available child */
397 }
398
399
400 /* Run parental oversight loop */
401 void
402 parental_loop()
403 {
404 #define MAXIQ (int)(PIPE_BUF/(sizeof(FVECT)*2))
405 static int ignore_warning_given = 0;
406 int qlimit = (accumulate == 1) ? 1 : MAXIQ-1;
407 int ninq = 0;
408 FVECT orgdir[2*MAXIQ];
409 int i, n;
410 /* load rays from stdin & process */
411 #ifdef getc_unlocked
412 flockfile(stdin); /* avoid lock/unlock overhead */
413 #endif
414 while (getvec(orgdir[2*ninq]) == 0 && getvec(orgdir[2*ninq+1]) == 0) {
415 if (orgdir[2*ninq+1][0] == 0.0 && /* asking for flush? */
416 (orgdir[2*ninq+1][1] == 0.0) &
417 (orgdir[2*ninq+1][2] == 0.0)) {
418 if (accumulate != 1) {
419 if (!ignore_warning_given++)
420 error(WARNING,
421 "dummy ray(s) ignored during accumulation\n");
422 continue;
423 }
424 while (next_child_nq(1) >= 0)
425 ; /* clear the queue */
426 lastdone = lastray = 0;
427 if ((yres <= 0) | (xres <= 0))
428 waitflush = 1; /* flush next */
429 put_zero_record(++lastray);
430 } else if (++ninq >= qlimit || accumulate > 1 &&
431 lastray/accumulate != (lastray+ninq)/accumulate) {
432 i = next_child_nq(0); /* manages output */
433 n = ninq;
434 if (accumulate != 1) /* request flush? */
435 memset(orgdir[2*n++], 0, sizeof(FVECT)*2);
436 n *= sizeof(FVECT)*2; /* send assignment */
437 if (writebuf(kida[i].pr.w, (char *)orgdir, n) != n)
438 error(SYSTEM, "pipe write error");
439 kida[i].r1 = lastray+1;
440 lastray += kida[i].nr = ninq; /* mark as busy */
441 ninq = 0;
442 if (lastray < lastdone) { /* RNUMBER wrapped? */
443 while (next_child_nq(1) >= 0)
444 ;
445 lastdone = lastray = 0;
446 }
447 }
448 if (raysleft && !--raysleft)
449 break; /* preemptive EOI */
450 }
451 while (next_child_nq(1) >= 0) /* empty results queue */
452 ;
453 /* output accumulated record */
454 if (accumulate <= 0 || account < accumulate) {
455 end_children(); /* frees up file descriptors */
456 if (account < accumulate) {
457 error(WARNING, "partial accumulation in final record");
458 accumulate -= account;
459 }
460 for (i = 0; i < nmods; i++)
461 mod_output(out_bq->mca[i]);
462 end_record();
463 free_binq(out_bq);
464 out_bq = NULL;
465 }
466 if (raysleft)
467 error(USER, "unexpected EOF on input");
468 free_binq(NULL); /* clean up */
469 lu_done(&ofiletab);
470 #undef MAXIQ
471 }