Back to index

radiance  4R0+20100331
rholo2l.c
Go to the documentation of this file.
00001 #ifndef lint
00002 static const char    RCSid[] = "$Id: rholo2l.c,v 3.16 2004/01/01 11:21:55 schorsch Exp $";
00003 #endif
00004 /*
00005  * Routines for local rtrace execution
00006  */
00007 
00008 #include <signal.h>
00009 #include <sys/time.h>
00010 #include <string.h>
00011 
00012 #include "rholo.h"
00013 #include "random.h"
00014 #include "paths.h"
00015 #include "selcall.h"
00016 #include "rtprocess.h"
00017 
00018 #ifndef MAXPROC
00019 #define MAXPROC             64
00020 #endif
00021 
00022 int    nprocs = 0;                        /* running process count */
00023 
00024 static char   pfile[] = TEMPLATE;         /* persist file name */
00025 
00026 static SUBPROC       rtpd[MAXPROC];              /* process descriptors */
00027 static float  *rtbuf = NULL;                     /* allocated i/o buffer */
00028 static int    maxqlen = 0;                /* maximum packets per queue */
00029 
00030 static PACKET *pqueue[MAXPROC];           /* packet queues */
00031 static int    pqlen[MAXPROC];                    /* packet queue lengths */
00032 
00033 static int bestout(void);
00034 static int slots_avail(void);
00035 static void queue_packet(PACKET *p);
00036 static PACKET * get_packets(int poll);
00037 static void killpersist(void);
00038 
00039 
00040 extern int
00041 start_rtrace(void)                 /* start rtrace process */
00042 {
00043        static char   buf1[8];
00044        int    rmaxpack = 0;
00045        int    psiz, n;
00046                                    /* get number of processes */
00047        if (ncprocs <= 0 || nprocs > 0)
00048               return(0);
00049        if (ncprocs > MAXPROC) {
00050               sprintf(errmsg,
00051                      "number of rtrace processes reduced from %d to %d",
00052                             ncprocs, MAXPROC);
00053               error(WARNING, errmsg);
00054               ncprocs = MAXPROC;
00055        }
00056        if (rtargv[rtargc-1] != vval(OCTREE)) {
00057                                           /* add compulsory options */
00058               rtargv[rtargc++] = "-i-";
00059               rtargv[rtargc++] = "-I-";
00060               rtargv[rtargc++] = "-h-";
00061               rtargv[rtargc++] = "-ld-";
00062               sprintf(buf1, "%d", RPACKSIZ);
00063               rtargv[rtargc++] = "-x"; rtargv[rtargc++] = buf1;
00064               rtargv[rtargc++] = "-y"; rtargv[rtargc++] = "0";
00065               rtargv[rtargc++] = "-fff";
00066               rtargv[rtargc++] = vbool(VDIST) ? "-ovl" : "-ovL";
00067               if (nowarn)
00068                      rtargv[rtargc++] = "-w-";
00069               if (ncprocs > 1) {
00070                      mktemp(pfile);
00071                      rtargv[rtargc++] = "-PP"; rtargv[rtargc++] = pfile;
00072               }
00073               rtargv[rtargc++] = vval(OCTREE);
00074               rtargv[rtargc] = NULL;
00075        }
00076        maxqlen = 0;
00077        for (nprocs = 0; nprocs < ncprocs; nprocs++) {   /* spawn children */
00078               psiz = open_process(&rtpd[nprocs], rtargv);
00079               if (psiz <= 0)
00080                      error(SYSTEM, "cannot start rtrace process");
00081               n = psiz/(RPACKSIZ*6*sizeof(float));
00082               if (maxqlen == 0) {
00083                      if (!(maxqlen = n))
00084                             error(INTERNAL,
00085                                    "bad pipe buffer size assumption");
00086                      sleep(2);
00087               } else if (n != maxqlen)
00088                      error(INTERNAL, "varying pipe buffer size!");
00089               rmaxpack += n;
00090        }
00091        rtbuf = (float *)malloc(RPACKSIZ*6*sizeof(float)*maxqlen);
00092        if (rtbuf == NULL)
00093               error(SYSTEM, "malloc failure in start_rtrace");
00094        return(rmaxpack);
00095 }
00096 
00097 
00098 static int
00099 bestout(void)               /* get best process to process packet */
00100 {
00101        int    cnt;
00102        register int  pn, i;
00103 
00104        pn = 0;                     /* find shortest queue */
00105        for (i = 1; i < nprocs; i++)
00106               if (pqlen[i] < pqlen[pn])
00107                      pn = i;
00108                             /* sanity check */
00109        if (pqlen[pn] == maxqlen)
00110               return(-1);
00111        cnt = 0;             /* count number of ties */
00112        for (i = pn; i < nprocs; i++)
00113               if (pqlen[i] == pqlen[pn])
00114                      cnt++;
00115                             /* break ties fairly */
00116        if ((cnt = random() % cnt))
00117               for (i = pn; i < nprocs; i++)
00118                      if (pqlen[i] == pqlen[pn] && !cnt--)
00119                             return(i);
00120        return(pn);
00121 }
00122 
00123 
00124 static int
00125 slots_avail(void)                  /* count packet slots available */
00126 {
00127        register int  nslots = 0;
00128        register int  i;
00129 
00130        for (i = nprocs; i--; )
00131               nslots += maxqlen - pqlen[i];
00132        return(nslots);
00133 }
00134 
00135 
00136 static void
00137 queue_packet(               /* queue up a beam packet */
00138        register PACKET      *p
00139 )
00140 {
00141        int    pn, n;
00142                             /* determine process to write to */
00143        if ((pn = bestout()) < 0)
00144               error(INTERNAL, "rtrace input queues are full!");
00145                             /* write out the packet */
00146        packrays(rtbuf, p);
00147        if ((n = p->nr) < RPACKSIZ) /* add flush block? */
00148               memset((char *)(rtbuf+6*n++), '\0', 6*sizeof(float));
00149        if (writebuf(rtpd[pn].w, (char *)rtbuf, 6*sizeof(float)*n) < 0)
00150               error(SYSTEM, "write error in queue_packet");
00151        p->next = NULL;
00152        if (!pqlen[pn]++)    /* add it to the end of the queue */
00153               pqueue[pn] = p;
00154        else {
00155               register PACKET      *rpl = pqueue[pn];
00156               while (rpl->next != NULL)
00157                      rpl = rpl->next;
00158               rpl->next = p;
00159        }
00160 }
00161 
00162 
00163 static PACKET *
00164 get_packets(         /* read packets from rtrace processes */
00165        int    poll
00166 )
00167 {
00168        static struct timeval       tpoll; /* zero timeval struct */
00169        fd_set readset, errset;
00170        PACKET *pldone = NULL, *plend;
00171        register PACKET      *p;
00172        int    n, nr;
00173        register int  pn;
00174        float  *bp;
00175                                    /* prepare select call */
00176        FD_ZERO(&readset); FD_ZERO(&errset); n = 0;
00177        for (pn = nprocs; pn--; ) {
00178               if (pqlen[pn])
00179                      FD_SET(rtpd[pn].r, &readset);
00180               FD_SET(rtpd[pn].r, &errset);
00181               if (rtpd[pn].r >= n)
00182                      n = rtpd[pn].r + 1;
00183        }
00184                                    /* make the call */
00185        n = select(n, &readset, (fd_set *)NULL, &errset,
00186                      poll ? &tpoll : (struct timeval *)NULL);
00187        if (n < 0) {
00188               if (errno == EINTR)  /* interrupted select call */
00189                      return(NULL);
00190               error(SYSTEM, "select call failure in get_packets");
00191        }
00192        if (n == 0)                 /* is nothing ready? */
00193               return(NULL);
00194                                    /* make read call(s) */
00195        for (pn = 0; pn < nprocs; pn++) {
00196               if (!FD_ISSET(rtpd[pn].r, &readset) &&
00197                             !FD_ISSET(rtpd[pn].r, &errset))
00198                      continue;
00199        reread:
00200               n = read(rtpd[pn].r, (char *)rtbuf,
00201                             4*sizeof(float)*RPACKSIZ*pqlen[pn]);
00202               if (n < 0) {
00203                      if ((errno == EINTR) | (errno == EAGAIN))
00204                             goto reread;
00205                      error(SYSTEM, "read error in get_packets");
00206               }
00207               if (n == 0)
00208                      goto eoferr;
00209               bp = rtbuf;                        /* finish processing */
00210               for (p = pqueue[pn]; n && p != NULL; p = p->next) {
00211                      if ((nr = p->nr) < RPACKSIZ)
00212                             nr++;                /* add flush block */
00213                      n -= 4*sizeof(float)*nr;
00214                      if (n < 0) {                /* get remainder */
00215                             n += readbuf(rtpd[pn].r,
00216                                           (char *)(bp+4*nr)+n, -n);
00217                             if (n)
00218                                    goto eoferr;
00219                      }
00220                      donerays(p, bp);
00221                      bp += 4*nr;
00222                      pqlen[pn]--;
00223               }
00224               if (n)                             /* read past end? */
00225                      error(INTERNAL, "packet sync error in get_packets");
00226                                                  /* take from queue */
00227               if (pldone == NULL)
00228                      pldone = plend = pqueue[pn];
00229               else
00230                      plend->next = pqueue[pn];
00231               while (plend->next != p)
00232                      plend = plend->next;
00233               plend->next = NULL;
00234               pqueue[pn] = p;
00235        }
00236        return(pldone);                           /* return finished packets */
00237 eoferr:
00238        error(USER, "rtrace process died");
00239        return NULL; /* pro forma return */
00240 }
00241 
00242 
00243 extern PACKET *
00244 do_packets(                 /* queue a packet list, return finished */
00245        register PACKET      *pl
00246 )
00247 {
00248        register PACKET      *p;
00249                                    /* consistency check */
00250        if (nprocs < 1)
00251               error(CONSISTENCY, "do_packets called with no active process");
00252                                    /* queue each new packet */
00253        while (pl != NULL) {
00254               p = pl; pl = p->next; p->next = NULL;
00255               queue_packet(p);
00256        }
00257        return(get_packets(slots_avail()));       /* return processed packets */
00258 }
00259 
00260 
00261 extern PACKET *
00262 flush_queue(void)                  /* empty all rtrace queues */
00263 {
00264        PACKET *rpdone = NULL;
00265        register PACKET      *rpl = NULL;
00266        float  *bp;
00267        register PACKET      *p;
00268        int    i, n, nr;
00269 
00270        for (i = 0; i < nprocs; i++)
00271               if (pqlen[i]) {
00272                      if (rpdone == NULL) {              /* tack on queue */
00273                             rpdone = rpl = pqueue[i];
00274                             if ((nr = rpl->nr) < RPACKSIZ) nr++;
00275                      } else {
00276                             rpl->next = pqueue[i];
00277                             nr = 0;
00278                      }
00279                      while (rpl->next != NULL) {
00280                             nr += (rpl = rpl->next)->nr;
00281                             if (rpl->nr < RPACKSIZ)
00282                                    nr++;         /* add flush block */
00283                      }
00284                      n = readbuf(rtpd[i].r, (char *)rtbuf,
00285                                    4*sizeof(float)*nr);
00286                      if (n < 0)
00287                             error(SYSTEM, "read failure in flush_queue");
00288                      bp = rtbuf;                 /* process packets */
00289                      for (p = pqueue[i]; p != NULL; p = p->next) {
00290                             if ((nr = p->nr) < RPACKSIZ)
00291                                    nr++;         /* add flush block */
00292                             n -= 4*sizeof(float)*nr;
00293                             if (n >= 0)
00294                                    donerays(p, bp);
00295                             else
00296                                    p->nr = 0;    /* short data error */
00297                             bp += 4*nr;
00298                      }
00299                      pqueue[i] = NULL;           /* zero this queue */
00300                      pqlen[i] = 0;
00301               }
00302        return(rpdone);             /* return all packets completed */
00303 }
00304 
00305 
00306 static void
00307 killpersist(void)                  /* kill persistent process */
00308 {
00309        FILE   *fp;
00310        int    pid;
00311 
00312        if ((fp = fopen(pfile, "r")) == NULL)
00313               return;
00314        if (fscanf(fp, "%*s %d", &pid) != 1 || kill(pid, SIGALRM) < 0)
00315               unlink(pfile);
00316        fclose(fp);
00317 }
00318 
00319 
00320 extern int
00321 end_rtrace(void)                   /* close rtrace process(es) */
00322 {
00323        int    status = 0, rv;
00324 
00325        if (nprocs > 1)
00326               killpersist();
00327        while (nprocs > 0) {
00328               rv = close_process(&rtpd[--nprocs]);
00329               if (rv > 0)
00330                      status = rv;
00331        }
00332        free((void *)rtbuf);
00333        rtbuf = NULL;
00334        maxqlen = 0;
00335        return(status);
00336 }