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

Comparing ray/src/util/rcomb.c (file contents):
Revision 2.10 by greg, Mon May 20 23:21:29 2024 UTC vs.
Revision 2.17 by greg, Thu May 23 17:13:52 2024 UTC

# Line 543 | Line 543 | output_headinfo(FILE *fp)
543   static int
544   spawned_children(int np)
545   {
546 +        size_t  recsize = 0;
547          int     i, rv;
548  
549   #if defined(_WIN32) || defined(_WIN64)
# Line 558 | Line 559 | spawned_children(int np)
559                  mop[i].imx.nrows = 1;
560                  if (!rmx_prepare(&mop[i].imx))
561                          goto memerror;
562 +                recsize += rmx_array_size(&mop[i].imx);
563                  if (mop[i].rmp != &mop[i].imx) {
564                          mop[i].rmp->nrows = 1;
565                          if (!rmx_prepare(mop[i].rmp))
# Line 598 | Line 600 | spawned_children(int np)
600                  cproc[i].pid = -1;
601                  rv = open_process(&cproc[i], NULL);
602                  if (rv <= 0) break;
603 +                if (!i && 2*rv >= recsize) {
604 +                        fputs("Problem too small for multi-processing\n",
605 +                                        stderr);
606 +                        close_processes(cproc, 1);
607 +                        exit(1);
608 +                }
609          }
602 –        if (rv > 0)
603 –                return(1);      /* parent return value */
610          if (rv < 0) {
611                  perror("fork");
612 +                close_processes(cproc, i);
613                  exit(1);
614          }
615 <        inchild = i;            /* child index */
615 >        if (rv > 0)             /* parent return? */
616 >                return(1);
617 >        inchild = i;            /* our child index */
618 >        while (i-- > 0)         /* don't share siblings' pipes */
619 >                close(cproc[i].w);
620          fpurge(stdin);          /* discard previous matrix input */
621   #ifdef getc_unlocked
622          flockfile(stdin);
623   #endif
624          for (i = 0; i < nmats; i++) {
625                  if (mop[i].infp != stdin)
626 <                        fclose(mop[i].infp);
626 >                        fclose(mop[i].infp);    /* ! pclose() */
627                  mop[i].infp = stdin;
628                  mop[i].imx.dtype = DTdouble;
629          }
# Line 642 | Line 653 | parent_loop()
653   #endif
654          for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
655              FILE        *ofp = outfp[cur_row % nchildren];
656 <            for (i = 0; i < nmats; i++) {
656 >            for (i = 0; i < nmats; i++)
657                  if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
658                          if (cur_row > in_nrows) /* unknown #input rows? */
659                                  break;
# Line 650 | Line 661 | parent_loop()
661                                          mop[i].inspec, cur_row);
662                          return(0);
663                  }
653 –                if (!rmx_write_data(mop[i].imx.mtx, mop[i].imx.ncomp,
654 –                                mop[i].imx.ncols, DTdouble, ofp))
655 –                        return(0);
656 –            }
664              if (i < nmats)
665                  break;
666 +            for (i = 0; i < nmats; i++)
667 +                if (!rmx_write_data(mop[i].imx.mtx, mop[i].imx.ncomp,
668 +                                        mop[i].imx.ncols, DTdouble, ofp))
669 +                        return(0);
670              if (fflush(ofp) == EOF)
671                  return(0);
672          }
673 <        for (i = 0; i < nchildren; i++)
673 >        for (i = 0; i < nchildren; i++) {
674 >                sleep(2);                       /* try to maintain order */
675                  fclose(outfp[i]);
676 +        }
677          free(outfp);
678          i = close_processes(cproc, nchildren);
679          free(cproc); cproc = NULL;
# Line 706 | Line 719 | combine_input()
719                  set_c = varlookup("c") != NULL && !vardefined("c");
720          } else                          /* save a little time */
721                  set_r = set_c = 0;
722 +
723                                          /* read/process row-by-row */
724          for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) {
725              RMATRIX     *mres = NULL;
# Line 758 | Line 772 | combine_input()
772                          return(0);
773              }
774              rmx_free(mres); mres = NULL;
775 <            if (inchild >= 0) {
776 <                i = getc(stdin);        /* child waits for turn to output */
775 >            if (inchild >= 0) {         /* children share stdout */
776 >                i = getc(stdin);        /* signals it's our turn */
777                  if (i != EOF) ungetc(i, stdin);
778              }
779              if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp,
# Line 899 | Line 913 | main(int argc, char *argv[])
913                                  if (n && !isflt(argv[i+1])) {
914                                          mop[nmats].preop.csym = argv[++i];
915                                          mop[nmats].preop.clen = 0;
916 +                                        mcat_last = 0;
917                                          break;
918                                  }
919                                  if (n > MAXCOMP*MAXCOMP) n = MAXCOMP*MAXCOMP;
# Line 989 | Line 1004 | main(int argc, char *argv[])
1004                  return(1);
1005          }
1006          doptimize(1);                   /* optimize definitions */
1007 <        if (spawned_children(nproc))    /* running in children? */
1007 >        if (spawned_children(nproc))    /* running in parent process? */
1008                  return(parent_loop() ? 0 : 1);
1009                                          /* process & write rows */
1010          return(combine_input() ? 0 : 1);

Diff Legend

– Removed lines
+ Added lines
< Changed lines (old)
> Changed lines (new)