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

Comparing ray/src/util/rcomb.c (file contents):
Revision 2.12 by greg, Tue May 21 17:39:17 2024 UTC vs.
Revision 2.20 by greg, Tue Jun 4 21:23:11 2024 UTC

# Line 3 | Line 3 | static const char RCSid[] = "$Id$";
3   #endif
4   /*
5   * General component matrix combiner, operating on a row at a time.
6 + *
7 + * Multi-processing mode under Unix creates children that each work
8 + * on one input row at a time, fed by the original process.  Final conversion
9 + * and output to stdout is sorted by last child while its siblings send it
10 + * their record calculations.
11   */
12  
8 – #include <signal.h>
13   #include <math.h>
14   #include "platform.h"
15   #include "rtprocess.h"
# Line 57 | Line 61 | int            cur_chan;                       /* if we're looping channels */
61   SUBPROC         *cproc = NULL;                  /* child process array */
62   int             nchildren = 0;                  /* # of child processes */
63   int             inchild = -1;                   /* our child ID (-1: parent) */
60 – int             gpid = -1;                      /* group process ID (parent) */
61 – int             nr_out = 0;                     /* # of rows output by kids */
64  
65   static int      checksymbolic(ROPMAT *rop);
66  
65 – static void
66 – on_sigio(int dummy)
67 – {
68 –        nr_out++;                       /* happens when child outputs row */
69 –        signal(SIGIO, on_sigio);        /* reset to maximize portability */
70 – }
71 –
67   static int
68   split_input(ROPMAT *rop)
69   {
# Line 551 | Line 546 | output_headinfo(FILE *fp)
546   }
547  
548   static int
549 + output_loop(void)
550 + {
551 +        const size_t    row_size = rmx_array_size(mop[nmats].rmp);
552 +        int             i = nmats;
553 +        int             cur_child = 0;
554 +
555 +        if (mop[nmats].rmp != &mop[nmats].imx)          /* output is split? */
556 +                rmx_reset(&mop[nmats].imx);
557 +        while (i-- > 0) {                               /* close input matrices */
558 +                fclose(mop[i].infp);            /* ! pclose() */
559 +                mop[i].infp = NULL;
560 +                rmx_reset(&mop[i].imx);
561 +                if (mop[i].rmp != &mop[i].imx) {
562 +                        rmx_free(mop[i].rmp);
563 +                        mop[i].rmp = &mop[i].imx;
564 +                }
565 +        }
566 + #ifdef getc_unlocked
567 +        flockfile(stdout);                              /* we own this, now */
568 + #endif
569 +        for ( ; ; ) {                                   /* loop until no more */
570 +                ssize_t         rv;
571 +                rv = readbuf(cproc[cur_child].r, mop[nmats].rmp->mtx, row_size);
572 +                if (!rv)                                /* out of rows? */
573 +                        break;
574 +                if (rv != row_size) {
575 +                        fputs("Read error in output loop\n", stderr);
576 +                        return(0);
577 +                }                                       /* do final conversion */
578 +                if (!rmx_write_data(mop[nmats].rmp->mtx, mop[nmats].rmp->ncomp,
579 +                                mop[nmats].rmp->ncols, mop[nmats].rmp->dtype, stdout)) {
580 +                        fputs("Conversion/write error in output loop\n", stderr);
581 +                        return(0);
582 +                }
583 +                cur_child++;
584 +                cur_child *= (cur_child < inchild);
585 +        }
586 +        return(fflush(stdout) != EOF);
587 + }
588 +
589 + static int
590   spawned_children(int np)
591   {
592          int     i, rv;
# Line 561 | Line 597 | spawned_children(int np)
597                  np = 1;
598          } else
599   #endif
600 <        if ((in_nrows > 0) & (np > in_nrows))
601 <                np = in_nrows;
600 >        if ((in_nrows > 0) & (np*4 > in_nrows))
601 >                np = in_nrows/4;
602                                  /* we'll be doing a row at a time */
603          for (i = 0; i < nmats; i++) {
604                  mop[i].imx.nrows = 1;
# Line 574 | Line 610 | spawned_children(int np)
610                                  goto memerror;
611                  }
612          }
613 <                                /* prep output row buffer */
613 >                                /* prep output row buffer(s) */
614          if (mcat || mop[nmats].preop.clen > 0) {
615                  if (!split_input(&mop[nmats]))  /* need separate buffer */
616                          return(0);
617                  if (mop[nmats].preop.clen > 0)
618                          mop[nmats].rmp->ncomp = mop[nmats].preop.clen /
619                                                  mop[nmats].imx.ncomp;
584 –                mop[nmats].rmp->nrows = 1;
585 –                if (!mcat | !mcat_last && !rmx_prepare(mop[nmats].rmp))
586 –                        goto memerror;
620          }
621          mop[nmats].imx.nrows = 1;
622          if (!rmx_prepare(&mop[nmats].imx))
623                  goto memerror;
624 <        if (np <= 1) {          /* single process return point */
624 >        if (mop[nmats].rmp != &mop[nmats].imx) {
625 >                mop[nmats].rmp->nrows = 1;
626 >                if (!rmx_prepare(mop[nmats].rmp))
627 >                        goto memerror;
628 >        }
629 >        if (np <= 1) {          /* single process return */
630   #ifdef getc_unlocked
631                  for (i = 0; i < nmats; i++)
632                          flockfile(mop[i].infp);
# Line 596 | Line 634 | spawned_children(int np)
634   #endif
635                  return(0);
636          }
599 –        gpid = setpgrp();       /* set group process ID */
600 –        signal(SIGIO, on_sigio);
637          fflush(stdout);         /* flush header & spawn children */
638 <        cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*np);
638 >        nchildren = np + 1;     /* extra child to sequence output */
639 >        cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren);
640          if (!cproc)
641                  goto memerror;
642 <        nchildren = np;
643 <        for (i = 0; i < np; i++) {
644 <                cproc[i].flags = PF_FILT_OUT;
645 <                cproc[i].w = dup(1);
609 <                cproc[i].r = 0;
610 <                cproc[i].pid = -1;
642 >        for (i = nchildren; i--; ) cproc[i] = sp_inactive;
643 >        cproc[nchildren-1].flags |= PF_FILT_OUT;
644 >                                /* start each child */
645 >        for (i = 0; i < nchildren; i++) {
646                  rv = open_process(&cproc[i], NULL);
647                  if (rv <= 0) break;
648          }
614 –        if (rv > 0)
615 –                return(1);      /* parent return value */
649          if (rv < 0) {
650                  perror("fork");
651 +                close_processes(cproc, i);
652                  exit(1);
653          }
654 +        if (rv) {               /* are we the parent? */
655 +                i = nchildren-1;        /* last child is sole reader */
656 +                while (i-- > 0)
657 +                        close(cproc[i].r);
658 +                return(1);      /* parent return value */
659 +        }
660          inchild = i;            /* our child index */
661 <        while (i-- > 0)         /* don't share siblings' pipes */
661 >        while (i-- > 0)         /* only parent writes siblings */
662                  close(cproc[i].w);
663 <        fpurge(stdin);          /* discard previous matrix input */
664 < #ifdef getc_unlocked
665 <        flockfile(stdin);
666 < #endif
667 <        for (i = 0; i < nmats; i++) {
663 >
664 >        if (inchild == nchildren-1)     /* last child sequences output */
665 >                exit(output_loop() ? 0 : 1);
666 >
667 >        i = inchild;            /* won't read from siblings */
668 >        while (i-- > 0)
669 >                close(cproc[i].r);
670 >        i = nmats;              /* close input matrix streams */
671 >        while (i-- > 0) {
672                  if (mop[i].infp != stdin)
673                          fclose(mop[i].infp);    /* ! pclose() */
674                  mop[i].infp = stdin;
675 <                mop[i].imx.dtype = DTdouble;
675 >                mop[i].imx.dtype = DTrmx_native;
676 >                mop[i].imx.pflags &= ~RMF_SWAPIN;
677          }
678 <        return(0);              /* child return */
678 >        fpurge(stdin);          /* discard any previous matrix input */
679 > #ifdef getc_unlocked
680 >        flockfile(stdin);
681 > #endif
682 >        mop[nmats].rmp->dtype = DTrmx_native;
683 >        return(0);              /* worker child return value */
684   memerror:
685          fputs("Out of memory in spawned_children()\n", stderr);
686          exit(1);
687   }
688  
689   static int
690 < parent_loop()
690 > parent_loop(void)
691   {
642 –        FILE    **outfp = (FILE **)malloc(nchildren*sizeof(FILE *));
692          int     i;
693  
694 <        if (!outfp) goto memerror;
695 <        for (i = 0; i < nchildren; i++) {
696 <                outfp[i] = fdopen(cproc[i].w, "w");
697 <                if (!outfp[i]) goto memerror;
649 < #ifdef getc_unlocked
650 <                flockfile(outfp[i]);
651 < #endif
694 >        rmx_reset(&mop[nmats].imx);             /* not touching output side */
695 >        if (mop[nmats].rmp != &mop[nmats].imx) {
696 >                rmx_free(mop[nmats].rmp);
697 >                mop[nmats].rmp = &mop[nmats].imx;
698          }
699   #ifdef getc_unlocked
700 <        for (i = 0; i < nmats; i++)
700 >        for (i = 0; i < nmats; i++)             /* we handle matrix inputs */
701                  flockfile(mop[i].infp);
702   #endif
703 +                                                /* load & send rows to kids */
704          for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
705 <            FILE        *ofp = outfp[cur_row % nchildren];
705 >            int         wfd = cproc[cur_row % (nchildren-1)].w;
706              for (i = 0; i < nmats; i++)
707                  if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
708                          if (cur_row > in_nrows) /* unknown #input rows? */
709                                  break;
710 <                        fprintf(stderr, "%s: read error at row %d\n",
710 >                        fprintf(stderr, "%s: parent read error at row %d\n",
711                                          mop[i].inspec, cur_row);
712                          return(0);
713                  }
714              if (i < nmats)
715                  break;
716              for (i = 0; i < nmats; i++)
717 <                if (!rmx_write_data(mop[i].imx.mtx, mop[i].imx.ncomp,
718 <                                        mop[i].imx.ncols, DTdouble, ofp))
719 <                        return(0);
673 <            if (fflush(ofp) == EOF)
674 <                return(0);
717 >                if (writebuf(wfd, mop[i].imx.mtx, rmx_array_size(&mop[i].imx))
718 >                                        != rmx_array_size(&mop[i].imx))
719 >                        return(0);
720          }
676 –        for (i = 0; i < nchildren; i++)
677 –                fclose(outfp[i]);
678 –        free(outfp);
721          i = close_processes(cproc, nchildren);
722 <        free(cproc); cproc = NULL;
722 >        free(cproc); cproc = NULL; nchildren = 0;
723          if (i < 0) {
724 <                fputs("Warning: missing child in parent_loop()\n", stderr);
724 >                fputs("Warning: lost child in parent_loop()\n", stderr);
725                  return(1);
726          }
727          if (i > 0) {
728                  fprintf(stderr, "Child exited with status %d\n", i);
729                  return(0);
730          }
731 <        return(1);
731 >        return(1);                              /* return success! */
732   memerror:
733          fputs("Out of memory in parent_loop()\n", stderr);
734          exit(1);
735   }
736  
737   static int
738 < combine_input()
738 > combine_input(void)
739   {
740          const int       row0 = (inchild >= 0)*inchild;
741 <        const int       rstep = nchildren + !nchildren;
741 >        const int       rstep = nchildren ? nchildren-1 : 1;
742          ROPMAT          *res = &mop[nmats];
743          int             set_r, set_c;
744          RMATRIX         *tmp = NULL;
# Line 772 | Line 814 | combine_input()
814                          return(0);
815              }
816              rmx_free(mres); mres = NULL;
775 –            if (inchild >= 0)           /* children share stdout */
776 –                while (nr_out < cur_row)
777 –                    pause();            /* wait for our turn */
817              if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp,
818                                  res->rmp->ncols, res->rmp->dtype, stdout))
819                  return(0);
820 <            if (inchild >= 0) {         /* flush and notify group */
821 <                if (fflush(stdout) == EOF)
783 <                    return(0);
784 <                killpg(gpid, SIGIO);    /* increments everyone's nr_out */
785 <            }
820 >            if (inchild >= 0 && fflush(stdout) == EOF)
821 >                return(0);
822          }
823          return(inchild >= 0 || fflush(stdout) != EOF);
824   memerror:
# Line 915 | Line 951 | main(int argc, char *argv[])
951                                  if (n && !isflt(argv[i+1])) {
952                                          mop[nmats].preop.csym = argv[++i];
953                                          mop[nmats].preop.clen = 0;
954 +                                        mcat_last = 0;
955                                          break;
956                                  }
957                                  if (n > MAXCOMP*MAXCOMP) n = MAXCOMP*MAXCOMP;

Diff Legend

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