ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/rt/rc3.c
(Generate patch)

Comparing ray/src/rt/rc3.c (file contents):
Revision 2.9 by greg, Thu Jun 14 05:13:25 2012 UTC vs.
Revision 2.21 by greg, Fri Mar 4 02:48:14 2016 UTC

# Line 6 | Line 6 | static const char RCSid[] = "$Id$";
6   * Controlling process for multiple children
7   */
8  
9 + #include <signal.h>
10   #include "rcontrib.h"
10 – #include "platform.h"
11 – #include "rtprocess.h"
11   #include "selcall.h"
12  
13 + #define MAXIQ           (int)(PIPE_BUF/(sizeof(FVECT)*2))
14 +
15   /* Modifier contribution queue (results waiting to be output) */
16   typedef struct s_binq {
17          RNUMBER         ndx;            /* index for this entry */
# Line 22 | Line 23 | typedef struct s_binq {
23   static BINQ     *out_bq = NULL;         /* output bin queue */
24   static BINQ     *free_bq = NULL;        /* free queue entries */
25  
26 + static SUBPROC  kidpr[MAXPROCESS];      /* our child processes */
27 +
28   static struct {
29          RNUMBER r1;                     /* assigned ray starting index */
27 –        SUBPROC pr;                     /* PID, i/o descriptors */
30          FILE    *infp;                  /* file pointer to read from process */
31 <        int     nr;                     /* number rays to sum (0 if free) */
32 < } kida[MAXPROCESS];             /* our child processes */
31 >        int     nr;                     /* number of rays to sum (0 if free) */
32 > } kida[MAXPROCESS];             /* our child process i/o */
33  
34  
35   /* Get new bin queue entry */
# Line 94 | Line 96 | free_binq(BINQ *bp)
96  
97  
98   /* Add modifier values to accumulation record in queue and clear */
99 < void
99 > static void
100   queue_modifiers()
101   {
102          MODCONT *mpin, *mpout;
# Line 146 | Line 148 | queue_output(BINQ *bp)
148                  }
149                  return;
150          }
151 <        b_last = NULL;                  /* else insert in output queue */
151 >        b_last = NULL;                  /* insert in output queue */
152          for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
153                                  b_cur = b_cur->next)
154                  b_last = b_cur;
# Line 184 | Line 186 | queue_ready()
186          int     nready = 0;
187          BINQ    *bp;
188  
187 –        if (accumulate <= 0)            /* just accumulating? */
188 –                return(0);
189 –
189          for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
190                                  bp->ndx == lastdone+nready*accumulate+1;
191                                  bp = bp->next)
# Line 203 | Line 202 | output_catchup(int nmax)
202          int     nout = 0;
203          BINQ    *bp;
204          int     i;
205 <
207 <        if (accumulate <= 0)            /* just accumulating? */
208 <                return(0);
209 <                                        /* else output ready results */
205 >                                        /* output ready results */
206          while (out_bq != NULL && out_bq->nadded >= accumulate
207                                  && out_bq->ndx == lastdone+1) {
208                  if ((nmax > 0) & (nout >= nmax))
# Line 270 | Line 266 | set_stdout(const LUENT *le, void *p)
266   }
267  
268  
269 < /* Start child processes if we can */
269 > /* Start child processes if we can (call only once in parent!) */
270   int
271   in_rchild()
272   {
# Line 278 | Line 274 | in_rchild()
274  
275          while (nchild < nproc) {        /* fork until target reached */
276                  errno = 0;
277 <                rval = open_process(&kida[nchild].pr, NULL);
277 >                rval = open_process(&kidpr[nchild], NULL);
278                  if (rval < 0)
279                          error(SYSTEM, "open_process() call failed");
280                  if (rval == 0) {        /* if in child, set up & return true */
281 <                        lu_doall(&modconttab, set_stdout, NULL);
281 >                        lu_doall(&modconttab, &set_stdout, NULL);
282                          lu_done(&ofiletab);
283                          while (nchild--) {      /* don't share other pipes */
284 <                                close(kida[nchild].pr.w);
284 >                                close(kidpr[nchild].w);
285                                  fclose(kida[nchild].infp);
286                          }
287                          inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
# Line 305 | Line 301 | in_rchild()
301                  if (rval != PIPE_BUF)
302                          error(CONSISTENCY, "bad value from open_process()");
303                                          /* connect to child's output */
304 <                kida[nchild].infp = fdopen(kida[nchild].pr.r, "rb");
304 >                kida[nchild].infp = fdopen(kidpr[nchild].r, "rb");
305                  if (kida[nchild].infp == NULL)
306                          error(SYSTEM, "out of memory in in_rchild()");
311 – #ifdef getc_unlocked
312 –                flockfile(kida[nchild].infp);   /* avoid mutex overhead */
313 – #endif
307                  kida[nchild++].nr = 0;  /* mark as available */
308          }
309 + #ifdef getc_unlocked
310 +        for (rval = nchild; rval--; )   /* avoid mutex overhead */
311 +                flockfile(kida[rval].infp);
312 + #endif
313          return(0);                      /* return "false" in parent */
314   }
315  
316  
317   /* Close child processes */
318   void
319 < end_children()
319 > end_children(int immed)
320   {
321 <        int     status;
322 <        
323 <        while (nchild > 0) {
324 <                nchild--;
325 <                if ((status = close_process(&kida[nchild].pr)) > 0) {
326 <                        sprintf(errmsg,
327 <                                "rendering process returned bad status (%d)",
328 <                                        status);
321 >        int     i;
322 >
323 > #ifdef SIGKILL                          /* error mode -- quick exit */
324 >        for (i = nchild*immed; i-- > 0; )
325 >                kill(kidpr[nchild].pid, SIGKILL);
326 > #endif
327 >        if ((i = close_processes(kidpr, nchild)) > 0 && !immed) {
328 >                sprintf(errmsg, "rendering process returned bad status (%d)",
329 >                                        i);
330                          error(WARNING, errmsg);
333 –                }
334 –                fclose(kida[nchild].infp);
331          }
332 +        while (nchild-- > 0)
333 +                fclose(kida[nchild].infp);
334   }
335  
336  
# Line 366 | Line 364 | tryagain:                              /* catch up with output? */
364          n = nr = 0;
365          for (i = nchild; i--; ) {
366                  if (kida[i].nr) {
367 <                        FD_SET(kida[i].pr.r, &readset);
367 >                        FD_SET(kidpr[i].r, &readset);
368                          ++nr;
369                  }
370 <                FD_SET(kida[i].pr.r, &errset);
371 <                if (kida[i].pr.r >= n)
372 <                        n = kida[i].pr.r + 1;
370 >                FD_SET(kidpr[i].r, &errset);
371 >                if (kidpr[i].r >= n)
372 >                        n = kidpr[i].r + 1;
373          }
374          if (!nr)                        /* nothing to wait for? */
375                  return(-1);
# Line 388 | Line 386 | tryagain:                              /* catch up with output? */
386                  FD_ZERO(&errset);
387          n = -1;                         /* read results from child(ren) */
388          for (i = nchild; i--; ) {
389 <                if (FD_ISSET(kida[i].pr.r, &errset))
389 >                if (FD_ISSET(kidpr[i].r, &errset))
390                          error(USER, "rendering process died");
391 <                if (FD_ISSET(kida[i].pr.r, &readset))
391 >                if (FD_ISSET(kidpr[i].r, &readset))
392                          queue_results(n = i);
393          }
394          return(n);                      /* first available child */
# Line 401 | Line 399 | tryagain:                              /* catch up with output? */
399   void
400   parental_loop()
401   {
402 < #define MAXIQ           (int)(PIPE_BUF/(sizeof(FVECT)*2))
405 <        static int      ignore_warning_given = 0;
406 <        int             qlimit = (accumulate == 1) ? 1 : MAXIQ-1;
402 >        const int       qlimit = (accumulate == 1) ? 1 : MAXIQ-1;
403          int             ninq = 0;
404          FVECT           orgdir[2*MAXIQ];
405          int             i, n;
# Line 412 | Line 408 | parental_loop()
408          flockfile(stdin);               /* avoid lock/unlock overhead */
409   #endif
410          while (getvec(orgdir[2*ninq]) == 0 && getvec(orgdir[2*ninq+1]) == 0) {
411 <                if (orgdir[2*ninq+1][0] == 0.0 &&       /* asking for flush? */
412 <                                (orgdir[2*ninq+1][1] == 0.0) &
413 <                                (orgdir[2*ninq+1][2] == 0.0)) {
414 <                        if (accumulate != 1) {
415 <                                if (!ignore_warning_given++)
416 <                                        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 &&
411 >                const int       zero_ray = orgdir[2*ninq+1][0] == 0.0 &&
412 >                                                (orgdir[2*ninq+1][1] == 0.0) &
413 >                                                (orgdir[2*ninq+1][2] == 0.0);
414 >                ninq += !zero_ray;
415 >                                /* Zero ray cannot go in input queue */
416 >                if (zero_ray ? ninq : ninq >= qlimit ||
417                              lastray/accumulate != (lastray+ninq)/accumulate) {
418                          i = next_child_nq(0);           /* manages output */
419                          n = ninq;
420 <                        if (accumulate != 1)            /* request flush? */
420 >                        if (accumulate > 1)             /* need terminator? */
421                                  memset(orgdir[2*n++], 0, sizeof(FVECT)*2);
422                          n *= sizeof(FVECT)*2;           /* send assignment */
423 <                        if (writebuf(kida[i].pr.w, (char *)orgdir, n) != n)
423 >                        if (writebuf(kidpr[i].w, (char *)orgdir, n) != n)
424                                  error(SYSTEM, "pipe write error");
425                          kida[i].r1 = lastray+1;
426                          lastray += kida[i].nr = ninq;   /* mark as busy */
441 –                        ninq = 0;
427                          if (lastray < lastdone) {       /* RNUMBER wrapped? */
428                                  while (next_child_nq(1) >= 0)
429                                          ;
430 <                                lastdone = lastray = 0;
430 >                                lastray -= ninq;
431 >                                lastdone = lastray %= accumulate;
432                          }
433 +                        ninq = 0;
434                  }
435 +                if (zero_ray) {                         /* put bogus record? */
436 +                        if ((yres <= 0) | (xres <= 1) &&
437 +                                        (lastray+1) % accumulate == 0) {
438 +                                while (next_child_nq(1) >= 0)
439 +                                        ;               /* clear the queue */
440 +                                lastdone = lastray = accumulate-1;
441 +                                waitflush = 1;          /* flush next */
442 +                        }
443 +                        put_zero_record(++lastray);
444 +                }
445                  if (raysleft && !--raysleft)
446                          break;                          /* preemptive EOI */
447          }
448          while (next_child_nq(1) >= 0)           /* empty results queue */
449                  ;
450 <                                                /* output accumulated record */
451 <        if (accumulate <= 0 || account < accumulate) {
452 <                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);
450 >        if (account < accumulate) {
451 >                error(WARNING, "partial accumulation in final record");
452 >                free_binq(out_bq);              /* XXX just ignore it */
453                  out_bq = NULL;
454          }
455 +        free_binq(NULL);                        /* clean up */
456 +        lu_done(&ofiletab);
457          if (raysleft)
458                  error(USER, "unexpected EOF on input");
459 <        free_binq(NULL);                        /* clean up */
459 > }
460 >
461 >
462 > /* Wait for the next available child by monitoring "to" pipes */
463 > static int
464 > next_child_ready()
465 > {
466 >        fd_set                  writeset, errset;
467 >        int                     i, n;
468 >
469 >        for (i = nchild; i--; )         /* see if there's one free first */
470 >                if (!kida[i].nr)
471 >                        return(i);
472 >                                        /* prepare select() call */
473 >        FD_ZERO(&writeset); FD_ZERO(&errset);
474 >        n = 0;
475 >        for (i = nchild; i--; ) {
476 >                FD_SET(kidpr[i].w, &writeset);
477 >                FD_SET(kidpr[i].r, &errset);
478 >                if (kidpr[i].w >= n)
479 >                        n = kidpr[i].w + 1;
480 >                if (kidpr[i].r >= n)
481 >                        n = kidpr[i].r + 1;
482 >        }
483 >        errno = 0;
484 >        n = select(n, NULL, &writeset, &errset, NULL);
485 >        if (n < 0)
486 >                error(SYSTEM, "select() error in next_child_ready()");
487 >        n = -1;                         /* identify waiting child */
488 >        for (i = nchild; i--; ) {
489 >                if (FD_ISSET(kidpr[i].r, &errset))
490 >                        error(USER, "rendering process died");
491 >                if (FD_ISSET(kidpr[i].w, &writeset))
492 >                        kida[n = i].nr = 0;
493 >        }
494 >        return(n);                      /* first available child */
495 > }
496 >
497 >
498 > /* Modified parental loop for full accumulation mode (-c 0) */
499 > void
500 > feeder_loop()
501 > {
502 >        static int      ignore_warning_given = 0;
503 >        int             ninq = 0;
504 >        FVECT           orgdir[2*MAXIQ];
505 >        int             i, n;
506 >                                        /* load rays from stdin & process */
507 > #ifdef getc_unlocked
508 >        flockfile(stdin);               /* avoid lock/unlock overhead */
509 > #endif
510 >        while (getvec(orgdir[2*ninq]) == 0 && getvec(orgdir[2*ninq+1]) == 0) {
511 >                if (orgdir[2*ninq+1][0] == 0.0 &&       /* asking for flush? */
512 >                                (orgdir[2*ninq+1][1] == 0.0) &
513 >                                (orgdir[2*ninq+1][2] == 0.0)) {
514 >                        if (!ignore_warning_given++)
515 >                                error(WARNING,
516 >                                "dummy ray(s) ignored during accumulation\n");
517 >                        continue;
518 >                }
519 >                if (++ninq >= MAXIQ) {
520 >                        i = next_child_ready();         /* get eager child */
521 >                        n = sizeof(FVECT)*2 * ninq;     /* give assignment */
522 >                        if (writebuf(kidpr[i].w, (char *)orgdir, n) != n)
523 >                                error(SYSTEM, "pipe write error");
524 >                        kida[i].r1 = lastray+1;
525 >                        lastray += kida[i].nr = ninq;
526 >                        if (lastray < lastdone)         /* RNUMBER wrapped? */
527 >                                lastdone = lastray = 0;
528 >                        ninq = 0;
529 >                }
530 >                if (raysleft && !--raysleft)
531 >                        break;                          /* preemptive EOI */
532 >        }
533 >        if (ninq) {                             /* polish off input */
534 >                i = next_child_ready();
535 >                n = sizeof(FVECT)*2 * ninq;
536 >                if (writebuf(kidpr[i].w, (char *)orgdir, n) != n)
537 >                        error(SYSTEM, "pipe write error");
538 >                kida[i].r1 = lastray+1;
539 >                lastray += kida[i].nr = ninq;
540 >                ninq = 0;
541 >        }
542 >        memset(orgdir, 0, sizeof(FVECT)*2);     /* get results */
543 >        for (i = nchild; i--; ) {
544 >                writebuf(kidpr[i].w, (char *)orgdir, sizeof(FVECT)*2);
545 >                queue_results(i);
546 >        }
547 >        if (recover)                            /* and from before? */
548 >                queue_modifiers();
549 >        end_children(0);                        /* free up file descriptors */
550 >        for (i = 0; i < nmods; i++)
551 >                mod_output(out_bq->mca[i]);     /* output accumulated record */
552 >        end_record();
553 >        free_binq(out_bq);                      /* clean up */
554 >        out_bq = NULL;
555 >        free_binq(NULL);
556          lu_done(&ofiletab);
557 < #undef MAXIQ
557 >        if (raysleft)
558 >                error(USER, "unexpected EOF on input");
559   }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines