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.2 by greg, Sat Jun 9 16:47:27 2012 UTC vs.
Revision 2.5 by greg, Tue Jun 12 01:06:59 2012 UTC

# Line 14 | Line 14 | static const char RCSid[] = "$Id$";
14   /* Modifier contribution queue (results waiting to be output) */
15   typedef struct s_binq {
16          int             ndx;            /* index for this entry */
17 <        int             n2add;          /* number left to add */
17 >        int             nadded;         /* accumulated so far */
18          struct s_binq   *next;          /* next in queue */
19          MODCONT         *mca[1];        /* contrib. array (extends struct) */
20   } BINQ;
# Line 26 | Line 26 | static SUBPROC kida[MAXPROCESS];       /* child processes */
26   static FILE     *inq_fp[MAXPROCESS];    /* input streams */
27  
28  
29 < /* Get new (empty) bin queue entry */
29 > /* Get new bin queue entry */
30   static BINQ *
31   new_binq()
32   {
33 <        BINQ    *bp = free_bq;
33 >        BINQ    *bp;
34          int     i;
35  
36 <        if (bp != NULL) {               /* something already available? */
36 >        if (free_bq != NULL) {          /* something already available? */
37 >                bp = free_bq;
38                  free_bq = bp->next;
39                  bp->next = NULL;
40 <                bp->n2add = accumulate-1;
40 >                bp->nadded = 1;
41                  return(bp);
42          }
43                                          /* else allocate fresh */
44 <        bp = (BINQ *)malloc(sizeof(BINQ)+(nmods-1)*sizeof(MODCONT *));
44 >        bp = (BINQ *)malloc(sizeof(BINQ) + sizeof(MODCONT *)*(nmods-1));
45          if (bp == NULL)
46                  goto memerr;
47          for (i = nmods; i--; ) {
# Line 53 | Line 54 | new_binq()
54                  /* memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*mp->nbins); */
55          }
56          bp->ndx = 0;
57 <        bp->n2add = accumulate-1;
57 >        bp->nadded = 1;
58          bp->next = NULL;
59          return(bp);
60   memerr:
# Line 105 | Line 106 | queue_modifiers()
106                          addcolor(mpout->cbin[j], mpin->cbin[j]);
107                  memset(mpin->cbin, 0, sizeof(DCOLOR)*mpin->nbins);
108          }
109 +        out_bq->nadded++;
110   }
111  
112  
113 < /* Sum one modifier record into another (doesn't update n2add) */
113 > /* Sum one modifier record into another (updates nadded) */
114   static void
115   add_modbin(BINQ *dst, BINQ *src)
116   {
# Line 120 | Line 122 | add_modbin(BINQ *dst, BINQ *src)
122                  for (j = mpout->nbins; j--; )
123                          addcolor(mpout->cbin[j], mpin->cbin[j]);
124          }
125 +        dst->nadded += src->nadded;
126   }
127  
128  
# Line 143 | Line 146 | queue_output(BINQ *bp)
146          for (b_cur = out_bq; b_cur != NULL && b_cur->ndx < bp->ndx;
147                                  b_cur = b_cur->next)
148                  b_last = b_cur;
149 +
150          if (b_last != NULL) {
151                  bp->next = b_cur;
152                  b_last->next = bp;
# Line 150 | Line 154 | queue_output(BINQ *bp)
154                  bp->next = out_bq;
155                  out_bq = bp;
156          }
157 <        if (accumulate <= 1)            /* no accumulating? */
157 >        if (accumulate == 1)            /* no accumulation? */
158                  return;
159          b_cur = out_bq;                 /* else merge accumulation entries */
160          while (b_cur->next != NULL) {
161 <                if (b_cur->n2add <= 0 ||
161 >                if (b_cur->nadded >= accumulate ||
162                                  (b_cur->ndx-1)/accumulate !=
163                                  (b_cur->next->ndx-1)/accumulate) {
164                          b_cur = b_cur->next;
165                          continue;
166                  }
167                  add_modbin(b_cur, b_cur->next);
164                b_cur->n2add--;
168                  b_last = b_cur->next;
169                  b_cur->next = b_last->next;
170                  b_last->next = NULL;
# Line 170 | Line 173 | queue_output(BINQ *bp)
173   }
174  
175  
176 < /* Get current with output FIFO by producing ready results */
176 > /* Count number of records ready for output */
177   static int
178 < output_catchup()
178 > queue_ready()
179   {
180 +        int     nready = 0;
181 +        BINQ    *bp;
182 +
183 +        if (accumulate <= 0)            /* just accumulating? */
184 +                return(0);
185 +
186 +        for (bp = out_bq; bp != NULL && bp->nadded >= accumulate &&
187 +                                bp->ndx == lastdone+nready*accumulate+1;
188 +                                bp = bp->next)
189 +                ++nready;
190 +
191 +        return(nready);
192 + }
193 +
194 +
195 + /* Catch up with output queue by producing ready results */
196 + static int
197 + output_catchup(int nmax)
198 + {
199          int     nout = 0;
200          BINQ    *bp;
201          int     i;
# Line 181 | Line 203 | output_catchup()
203          if (accumulate <= 0)            /* just accumulating? */
204                  return(0);
205                                          /* else output ready results */
206 <        while (out_bq != NULL && (out_bq->ndx == lastdone+1) & !out_bq->n2add) {
206 >        while (out_bq != NULL && out_bq->nadded >= accumulate
207 >                                && out_bq->ndx == lastdone+1) {
208 >                if ((nmax > 0) & (nout >= nmax))
209 >                        break;
210                  bp = out_bq;                    /* pop off first entry */
211                  out_bq = bp->next;
212                  bp->next = NULL;
# Line 196 | Line 221 | output_catchup()
221   }
222  
223  
224 < /* Put a zero record in results queue */
224 > /* Put a zero record in results queue & output */
225   void
226 < zero_record(int ndx)
226 > put_zero_record(int ndx)
227   {
228          BINQ    *bp = new_binq();
229          int     i;
# Line 207 | Line 232 | zero_record(int ndx)
232                  memset(bp->mca[i]->cbin, 0, sizeof(DCOLOR)*bp->mca[i]->nbins);
233          bp->ndx = ndx;
234          queue_output(bp);
235 +        output_catchup(0);
236   }
237  
238  
# Line 233 | Line 259 | in_rchild()
259                  int     p0[2], p1[2];
260                  int     pid;
261                                          /* prepare i/o pipes */
262 +                errno = 0;
263                  if (pipe(p0) < 0 || pipe(p1) < 0)
264                          error(SYSTEM, "pipe() call failed!");
265                  pid = fork();           /* fork parent process */
266                  if (pid == 0) {         /* if in child, set up & return true */
267                          close(p0[1]); close(p1[0]);
268 +                        lu_doall(&modconttab, set_stdout, NULL);
269 +                        lu_done(&ofiletab);
270 +                        while (nchild--) {      /* don't share other pipes */
271 +                                close(kida[nchild].w);
272 +                                fclose(inq_fp[nchild]);
273 +                        }
274                          dup2(p0[0], 0); close(p0[0]);
275                          dup2(p1[1], 1); close(p1[1]);
276                          inpfmt = (sizeof(RREAL)==sizeof(double)) ? 'd' : 'f';
# Line 247 | Line 280 | in_rchild()
280                          yres = 0;
281                          raysleft = 0;
282                          account = accumulate = 1;
250                        lu_doall(&modconttab, set_stdout, NULL);
251                        nchild = -1;
283                          return(1);      /* child return value */
284                  }
285                  if (pid < 0)
286                          error(SYSTEM, "fork() call failed!");
287 <                                        /* connect our pipes */
287 >                                        /* connect parent's pipes */
288                  close(p0[0]); close(p1[1]);
289                  kida[nchild].r = p1[0];
290                  kida[nchild].w = p0[1];
# Line 262 | Line 293 | in_rchild()
293                  inq_fp[nchild] = fdopen(p1[0], "rb");
294                  if (inq_fp[nchild] == NULL)
295                          error(SYSTEM, "out of memory in in_rchild()");
296 + #ifdef getc_unlocked
297 +                flockfile(inq_fp[nchild]);      /* avoid mutex overhead */
298 + #endif
299                  ++nchild;
300          }
301          return(0);                      /* parent return value */
# Line 275 | Line 309 | end_children()
309   {
310          int     status;
311          
312 <        while (nchild-- > 0) {
312 >        while (nchild > 0) {
313 >                nchild--;
314                  kida[nchild].r = -1;    /* close(-1) error is ignored */
315                  if ((status = close_process(&kida[nchild])) > 0) {
316                          sprintf(errmsg,
# Line 288 | Line 323 | end_children()
323   }
324  
325  
326 < /* Wait for the next available child, managing output queue as well */
326 > /* Wait for the next available child, managing output queue simultaneously */
327   static int
328 < next_child_nq(int force_wait)
328 > next_child_nq(int flushing)
329   {
330          static struct timeval   polling;
331 <        struct timeval          *pmode = force_wait | (accumulate <= 0) ?
297 <                                        (struct timeval *)NULL : &polling;
331 >        struct timeval          *pmode;
332          fd_set                  readset, errset;
333 <        int                     i, j, n, nr;
333 >        int                     i, j, n, nr, nqr;
334  
335 <        if (!force_wait)                /* see if there's one free */
335 >        if (!flushing)                  /* see if there's one free */
336                  for (i = nchild; i--; )
337                          if (kida[i].running < 0)
338                                  return(i);
339 +
340 +        nqr = queue_ready();            /* choose blocking mode or polling */
341 +        if ((nqr > 0) & !flushing)
342 +                pmode = &polling;
343 +        else
344 +                pmode = NULL;
345 + tryagain:                               /* catch up with output? */
346 +        if (pmode == &polling) {
347 +                if (nqr > nchild)       /* don't get too far behind */
348 +                        nqr -= output_catchup(nqr-nchild);
349 +        } else if (nqr > 0)             /* clear output before blocking */
350 +                nqr -= output_catchup(0);
351                                          /* prepare select() call */
352          FD_ZERO(&readset); FD_ZERO(&errset);
353          n = nr = 0;
# Line 314 | Line 360 | next_child_nq(int force_wait)
360                  if (kida[i].r >= n)
361                          n = kida[i].r + 1;
362          }
363 <        if (!nr)                        /* nothing going on */
363 >        if (!nr)                        /* nothing to wait for? */
364                  return(-1);
319 tryagain:
320        if (pmode == NULL)              /* about to block, so catch up */
321                output_catchup();
365          if ((nr > 1) | (pmode == &polling)) {
366                  errno = 0;
367                  nr = select(n, &readset, NULL, &errset, pmode);
368 <                if (!nr & (pmode == &polling)) {
368 >                if (!nr) {
369                          pmode = NULL;   /* try again, blocking this time */
370                          goto tryagain;
371                  }
372 <                if (nr <= 0)
373 <                        error(SYSTEM, "select call error in next_child_nq()");
372 >                if (nr < 0)
373 >                        error(SYSTEM, "select() error in next_child_nq()");
374          } else
375                  FD_ZERO(&errset);
376          n = -1;                         /* read results from child(ren) */
# Line 341 | Line 384 | tryagain:
384                  bq->ndx = kida[i].running;
385                                          /* read from child */
386                  for (j = 0; j < nmods; j++) {
387 <                        n = bq->mca[j]->nbins;
388 <                        nr = fread(bq->mca[j]->cbin,sizeof(DCOLOR),n,inq_fp[i]);
389 <                        if (nr != n)
387 >                        nr = bq->mca[j]->nbins;
388 >                        if (fread(bq->mca[j]->cbin, sizeof(DCOLOR), nr,
389 >                                                        inq_fp[i]) != nr)
390                                  error(SYSTEM, "read error from render process");
391                  }
392 <                queue_output(bq);       /* put results in output queue */
392 >                queue_output(bq);       /* add results to output queue */
393                  kida[i].running = -1;   /* mark child as available */
394                  n = i;
395          }
396 <        return(n);                      /* last available child */
396 >        return(n);                      /* first available child */
397   }
398  
399  
# Line 361 | Line 404 | parental_loop()
404          static int      ignore_warning_given = 0;
405          FVECT           orgdir[2];
406          double          d;
407 +        int             i;
408                                          /* load rays from stdin & process */
409   #ifdef getc_unlocked
410          flockfile(stdin);               /* avoid lock/unlock overhead */
# Line 376 | Line 420 | parental_loop()
420                  }
421                  if ((d == 0.0) | (lastray+1 < lastray)) {
422                          while (next_child_nq(1) >= 0)
423 <                                ;
423 >                                ;                       /* clear the queue */
424                          lastdone = lastray = 0;
425                  }
426                  if (d == 0.0) {
427                          if ((yres <= 0) | (xres <= 0))
428 <                                waitflush = 1;          /* flush right after */
429 <                        zero_record(++lastray);
430 <                } else {                                /* else assign */
431 <                        int     avail = next_child_nq(0);
432 <                        if (writebuf(kida[avail].w, (char *)orgdir,
433 <                                        sizeof(FVECT)*2) != sizeof(FVECT)*2)
428 >                                waitflush = 1;          /* flush next */
429 >                        put_zero_record(++lastray);
430 >                } else {                                /* else assign ray */
431 >                        i = next_child_nq(0);           /* manages output */
432 >                        if (writebuf(kida[i].w, (char *)orgdir,
433 >                                        sizeof(orgdir)) != sizeof(orgdir))
434                                  error(SYSTEM, "pipe write error");
435 <                        kida[avail].running = ++lastray;
435 >                        kida[i].running = ++lastray;    /* busy now */
436                  }
437                  if (raysleft && !--raysleft)
438                          break;                          /* preemptive EOI */
# Line 397 | Line 441 | parental_loop()
441                  ;
442                                                  /* output accumulated record */
443          if (accumulate <= 0 || account < accumulate) {
444 <                int     i;
444 >                end_children();                 /* frees up file descriptors */
445                  if (account < accumulate) {
446                          error(WARNING, "partial accumulation in final record");
447                          accumulate -= account;
# Line 405 | Line 449 | parental_loop()
449                  for (i = 0; i < nmods; i++)
450                          mod_output(out_bq->mca[i]);
451                  end_record();
452 +                free_binq(out_bq);
453 +                out_bq = NULL;
454          }
455          if (raysleft)
456                  error(USER, "unexpected EOF on input");

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines