--- ray/src/util/rcomb.c 2024/05/23 17:13:52 2.17 +++ ray/src/util/rcomb.c 2025/04/04 18:06:48 2.29 @@ -1,8 +1,13 @@ #ifndef lint -static const char RCSid[] = "$Id: rcomb.c,v 2.17 2024/05/23 17:13:52 greg Exp $"; +static const char RCSid[] = "$Id: rcomb.c,v 2.29 2025/04/04 18:06:48 greg Exp $"; #endif /* * General component matrix combiner, operating on a row at a time. + * + * Multi-processing mode under Unix creates children that each work + * on one input row at a time, fed by the original process. Final conversion + * and output to stdout is sorted by last child while its siblings send it + * their record calculations. */ #include @@ -16,8 +21,6 @@ static const char RCSid[] = "$Id: rcomb.c,v 2.17 2024/ #define M_PI 3.14159265358979323846 #endif -#define MAXCOMP MAXCSAMP /* #components we support */ - /* Unary matrix operation(s) */ typedef struct { double cmat[MAXCOMP*MAXCOMP]; /* component transformation */ @@ -57,9 +60,9 @@ SUBPROC *cproc = NULL; /* child process array */ int nchildren = 0; /* # of child processes */ int inchild = -1; /* our child ID (-1: parent) */ -static int checksymbolic(ROPMAT *rop); +extern int checksymbolic(ROPMAT *rop); -static int +int split_input(ROPMAT *rop) { if (rop->rmp == &rop->imx && !(rop->rmp = rmx_copy(&rop->imx))) { @@ -71,7 +74,7 @@ split_input(ROPMAT *rop) } /* Check/set transform based on a reference input file */ -static int +int checkreffile(ROPMAT *rop) { static const char *curRF = NULL; @@ -134,7 +137,7 @@ checkreffile(ROPMAT *rop) } /* Compute conversion row from spectrum to one channel of RGB */ -static void +void rgbrow(ROPMAT *rop, int r, int p) { const int nc = rop->imx.ncomp; @@ -151,7 +154,7 @@ rgbrow(ROPMAT *rop, int r, int p) } /* Compute conversion row from spectrum to one channel of XYZ */ -static void +void xyzrow(ROPMAT *rop, int r, int p) { const int nc = rop->imx.ncomp; @@ -168,8 +171,8 @@ xyzrow(ROPMAT *rop, int r, int p) } /* Use the spectral sensitivity function to compute matrix coefficients */ -static void -sensrow(ROPMAT *rop, int r, double (*sf)(SCOLOR sc, int ncs, const float wlpt[4])) +void +sensrow(ROPMAT *rop, int r, double (*sf)(const SCOLOR sc, int ncs, const float wlpt[4])) { const int nc = rop->imx.ncomp; int i; @@ -183,7 +186,7 @@ sensrow(ROPMAT *rop, int r, double (*sf)(SCOLOR sc, in } /* Check/set symbolic transform */ -static int +int checksymbolic(ROPMAT *rop) { const int nc = rop->imx.ncomp; @@ -297,7 +300,7 @@ checksymbolic(ROPMAT *rop) return(1); } -static int +int get_component_xfm(ROPMAT *rop) { int i, j; @@ -380,7 +383,7 @@ get_component_xfm(ROPMAT *rop) return(1); } -static int +int apply_op(RMATRIX *dst, const RMATRIX *src, const RUNARYOP *ro) { if (ro->clen > 0) { @@ -399,7 +402,7 @@ apply_op(RMATRIX *dst, const RMATRIX *src, const RUNAR return(1); } -static int +int open_input(ROPMAT *rop) { int outtype; @@ -421,7 +424,7 @@ open_input(ROPMAT *rop) } /* Return nominal wavelength associated with input component (return nm) */ -static double +double l_wavelength(char *nam) { double comp = argument(1); @@ -445,7 +448,7 @@ l_wavelength(char *nam) } /* Return ith input with optional channel selector */ -static double +double l_chanin(char *nam) { double inp = argument(1); @@ -470,7 +473,7 @@ l_chanin(char *nam) return(mop[mi].rmp->mtx[cur_col*in_ncomp + chan]); } -static int +int initialize(RMATRIX *imp) { int i; @@ -483,7 +486,7 @@ initialize(RMATRIX *imp) restype = mop[i].rmp->dtype; if (!imp->dtype || (restype = rmx_newtype(restype, imp->dtype)) > 0) imp->dtype = restype; - else + else if (!nowarn) fprintf(stderr, "%s: warning - data type mismatch\n", mop[i].inspec); if (!i) { @@ -517,7 +520,7 @@ initialize(RMATRIX *imp) return(1); } -static void +void output_headinfo(FILE *fp) { int i; @@ -540,47 +543,47 @@ output_headinfo(FILE *fp) } } -static int +int spawned_children(int np) { - size_t recsize = 0; int i, rv; #if defined(_WIN32) || defined(_WIN64) if (np > 1) { - fputs("Warning: only one process under Windows\n", stderr); + if (!nowarn) + fputs("Warning: only one process under Windows\n", stderr); np = 1; } else #endif - if ((in_nrows > 0) & (np > in_nrows)) - np = in_nrows; + if ((in_nrows > 0) & (np*4 > in_nrows)) + np = in_nrows/4; /* we'll be doing a row at a time */ for (i = 0; i < nmats; i++) { mop[i].imx.nrows = 1; if (!rmx_prepare(&mop[i].imx)) goto memerror; - recsize += rmx_array_size(&mop[i].imx); if (mop[i].rmp != &mop[i].imx) { mop[i].rmp->nrows = 1; if (!rmx_prepare(mop[i].rmp)) goto memerror; } } - /* prep output row buffer */ - if (mcat || mop[nmats].preop.clen > 0) { + /* prep output row buffer(s) */ + if (mop[nmats].preop.clen > 0) { if (!split_input(&mop[nmats])) /* need separate buffer */ - return(0); - if (mop[nmats].preop.clen > 0) - mop[nmats].rmp->ncomp = mop[nmats].preop.clen / - mop[nmats].imx.ncomp; - mop[nmats].rmp->nrows = 1; - if (!mcat | !mcat_last && !rmx_prepare(mop[nmats].rmp)) goto memerror; + mop[nmats].rmp->ncomp = mop[nmats].preop.clen / + mop[nmats].imx.ncomp; } mop[nmats].imx.nrows = 1; if (!rmx_prepare(&mop[nmats].imx)) goto memerror; - if (np <= 1) { /* single process return point */ + if (mop[nmats].rmp != &mop[nmats].imx) { + mop[nmats].rmp->nrows = 1; + if (!rmx_prepare(mop[nmats].rmp)) + goto memerror; + } + if (np <= 1) { /* single process return */ #ifdef getc_unlocked for (i = 0; i < nmats; i++) flockfile(mop[i].infp); @@ -589,122 +592,128 @@ spawned_children(int np) return(0); } fflush(stdout); /* flush header & spawn children */ - cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*np); + nchildren = np + 1; /* extra child to sequence output */ + cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren); if (!cproc) goto memerror; - nchildren = np; - for (i = 0; i < np; i++) { - cproc[i].flags = PF_FILT_OUT; - cproc[i].w = dup(1); - cproc[i].r = 0; - cproc[i].pid = -1; - rv = open_process(&cproc[i], NULL); - if (rv <= 0) break; - if (!i && 2*rv >= recsize) { - fputs("Problem too small for multi-processing\n", - stderr); - close_processes(cproc, 1); - exit(1); - } - } + for (i = nchildren; i--; ) cproc[i] = sp_inactive; + cproc[nchildren-1].flags |= PF_FILT_OUT; + /* start each child from parent */ + for (i = 0; i < nchildren; i++) + if ((rv = open_process(&cproc[i], NULL)) <= 0) + break; /* child breaks here */ if (rv < 0) { - perror("fork"); + perror("fork"); /* WTH? */ close_processes(cproc, i); exit(1); } - if (rv > 0) /* parent return? */ - return(1); - inchild = i; /* our child index */ - while (i-- > 0) /* don't share siblings' pipes */ + if (i != nchildren-1) { /* last child is sole reader */ + int j = i; + while (j-- > 0) { + close(cproc[j].r); + cproc[j].r = -1; + } + } + if (rv > 0) + return(1); /* parent return value */ + + inchild = i; /* else set our child index */ + while (i-- > 0) /* only parent writes siblings */ close(cproc[i].w); - fpurge(stdin); /* discard previous matrix input */ + + i = nmats; /* close matrix streams (carefully) */ + while (i-- > 0) { + if (mop[i].infp != stdin) { + close(fileno(mop[i].infp)); /* avoid lseek() */ + fclose(mop[i].infp); /* ! pclose() */ + } + mop[i].infp = NULL; + } + fpurge(stdin); /* discard previously buffered input */ + + if (inchild == nchildren-1) + return(-1); /* output process return value */ + + i = nmats; /* get matrix rows from parent */ + while (i-- > 0) { + mop[i].infp = stdin; + mop[i].imx.dtype = DTrmx_native; + mop[i].imx.pflags &= ~RMF_SWAPIN; + } #ifdef getc_unlocked flockfile(stdin); #endif - for (i = 0; i < nmats; i++) { - if (mop[i].infp != stdin) - fclose(mop[i].infp); /* ! pclose() */ - mop[i].infp = stdin; - mop[i].imx.dtype = DTdouble; - } - return(0); /* child return */ + mop[nmats].rmp->dtype = DTrmx_native; + return(0); /* worker child return value */ memerror: fputs("Out of memory in spawned_children()\n", stderr); exit(1); } -static int -parent_loop() +int +parent_loop(void) { - FILE **outfp = (FILE **)malloc(nchildren*sizeof(FILE *)); int i; - if (!outfp) goto memerror; - for (i = 0; i < nchildren; i++) { - outfp[i] = fdopen(cproc[i].w, "w"); - if (!outfp[i]) goto memerror; -#ifdef getc_unlocked - flockfile(outfp[i]); -#endif + rmx_reset(&mop[nmats].imx); /* not touching output side */ + if (mop[nmats].rmp != &mop[nmats].imx) { + rmx_free(mop[nmats].rmp); + mop[nmats].rmp = &mop[nmats].imx; } #ifdef getc_unlocked - for (i = 0; i < nmats; i++) + for (i = 0; i < nmats; i++) /* we handle matrix inputs */ flockfile(mop[i].infp); #endif + /* load & send rows to kids */ for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) { - FILE *ofp = outfp[cur_row % nchildren]; + int wfd = cproc[cur_row % (nchildren-1)].w; for (i = 0; i < nmats; i++) if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) { if (cur_row > in_nrows) /* unknown #input rows? */ break; - fprintf(stderr, "%s: read error at row %d\n", + fprintf(stderr, "%s: load error at row %d\n", mop[i].inspec, cur_row); return(0); } if (i < nmats) break; for (i = 0; i < nmats; i++) - if (!rmx_write_data(mop[i].imx.mtx, mop[i].imx.ncomp, - mop[i].imx.ncols, DTdouble, ofp)) - return(0); - if (fflush(ofp) == EOF) - return(0); + if (writebuf(wfd, mop[i].imx.mtx, rmx_array_size(&mop[i].imx)) + != rmx_array_size(&mop[i].imx)) { + fprintf(stderr, "%s: write error at row %d\n", + mop[i].inspec, cur_row); + return(0); + } } - for (i = 0; i < nchildren; i++) { - sleep(2); /* try to maintain order */ - fclose(outfp[i]); - } - free(outfp); - i = close_processes(cproc, nchildren); - free(cproc); cproc = NULL; + i = close_processes(cproc, nchildren); /* collect family */ + free(cproc); cproc = NULL; nchildren = 0; if (i < 0) { - fputs("Warning: missing child in parent_loop()\n", stderr); + if (!nowarn) + fputs("Warning: lost child process\n", stderr); return(1); } if (i > 0) { fprintf(stderr, "Child exited with status %d\n", i); return(0); } - return(1); -memerror: - fputs("Out of memory in parent_loop()\n", stderr); - exit(1); + return(1); /* return success! */ } -static int -combine_input() +int +combine_input(void) { const int row0 = (inchild >= 0)*inchild; - const int rstep = nchildren + !nchildren; + const int rstep = nchildren ? nchildren-1 : 1; ROPMAT *res = &mop[nmats]; int set_r, set_c; RMATRIX *tmp = NULL; int co_set; int i; - if (mcat && mcat_last && - !(tmp = rmx_alloc(1, res->imx.ncols, res->rmp->ncomp))) - goto memerror; + if (mcat_last && !(tmp = rmx_alloc(1, res->imx.ncols, res->rmp->ncomp))) { + fputs("Out of buffer space in combine_input()\n", stderr); + return(0); + } /* figure out what the user set */ co_set = fundefined("co"); if (!co_set) @@ -719,7 +728,6 @@ combine_input() set_c = varlookup("c") != NULL && !vardefined("c"); } else /* save a little time */ set_r = set_c = 0; - /* read/process row-by-row */ for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) { RMATRIX *mres = NULL; @@ -727,7 +735,7 @@ combine_input() if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) { if (cur_row > in_nrows) /* unknown #input rows? */ break; - fprintf(stderr, "%s: read error at row %d\n", + fprintf(stderr, "%s: load error at row %d\n", mop[i].inspec, cur_row); return(0); } @@ -772,26 +780,60 @@ combine_input() return(0); } rmx_free(mres); mres = NULL; - if (inchild >= 0) { /* children share stdout */ - i = getc(stdin); /* signals it's our turn */ - if (i != EOF) ungetc(i, stdin); - } if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp, - res->rmp->ncols, res->rmp->dtype, stdout)) + res->rmp->ncols, res->rmp->dtype, stdout) || + (inchild >= 0 && fflush(stdout) == EOF)) { + fprintf(stderr, "Conversion/write error at row %d\n", + cur_row); return(0); - if (inchild >= 0 && fflush(stdout) == EOF) - return(0); + } } return(inchild >= 0 || fflush(stdout) != EOF); -memerror: - fputs("Out of buffer space in combine_input()\n", stderr); - return(0); multerror: - fputs("Unexpected matrix multiply error in combine_input()\n", stderr); + fputs("Unexpected matrix multiply error\n", stderr); return(0); } -static int +int +output_loop(void) +{ + const size_t row_size = rmx_array_size(mop[nmats].rmp); + int cur_child = 0; + int i = nmats; + + while (i-- > 0) { /* free input buffers */ + rmx_reset(&mop[i].imx); + if (mop[i].rmp != &mop[i].imx) { + rmx_free(mop[i].rmp); + mop[i].rmp = &mop[i].imx; + } + } + if (mop[nmats].rmp != &mop[nmats].imx) /* output is split? */ + rmx_reset(&mop[nmats].imx); +#ifdef getc_unlocked + flockfile(stdout); /* we own this, now */ +#endif + for ( ; ; ) { /* loop until no more */ + ssize_t rv; + rv = readbuf(cproc[cur_child].r, mop[nmats].rmp->mtx, row_size); + if (!rv) /* out of rows? */ + break; + if (rv != row_size) { + fputs("Read error\n", stderr); + return(0); + } /* do final conversion */ + if (!rmx_write_data(mop[nmats].rmp->mtx, mop[nmats].rmp->ncomp, + mop[nmats].rmp->ncols, mop[nmats].rmp->dtype, stdout)) { + fputs("Conversion/write error\n", stderr); + return(0); + } + cur_child++; + cur_child *= (cur_child < inchild); /* loop over workers */ + } + return(fflush(stdout) != EOF); +} + +int get_factors(double da[], int n, char *av[]) { int ac; @@ -801,14 +843,14 @@ get_factors(double da[], int n, char *av[]) return(ac); } -static void +void resize_inparr(int n2alloc) { int i; if (n2alloc == nall) return; - for (i = nall; i > n2alloc; i--) { + for (i = nall; i-- > n2alloc; ) { rmx_reset(&mop[i].imx); if (mop[i].rmp != &mop[i].imx) rmx_free(mop[i].rmp); @@ -903,17 +945,17 @@ main(int argc, char *argv[]) } break; case 'C': + mcat_last = 0; if (!n || isflt(argv[i+1])) goto userr; defCsym = mop[nmats].preop.csym = argv[++i]; mop[nmats].preop.clen = 0; - mcat_last = 0; break; case 'c': + mcat_last = 0; if (n && !isflt(argv[i+1])) { mop[nmats].preop.csym = argv[++i]; mop[nmats].preop.clen = 0; - mcat_last = 0; break; } if (n > MAXCOMP*MAXCOMP) n = MAXCOMP*MAXCOMP; @@ -926,16 +968,15 @@ main(int argc, char *argv[]) goto userr; } mop[nmats].preop.csym = NULL; - mcat_last = 0; break; case 'm': + mcat_last = 1; if (!n) goto userr; if (argv[++i][0] == '-' && !argv[i][1]) { if (stdin_used++) goto stdin_error; mcat_spec = stdin_name; } else mcat_spec = argv[i]; - mcat_last = 1; break; default: fprintf(stderr, "%s: unknown option '%s'\n", @@ -1004,9 +1045,12 @@ main(int argc, char *argv[]) return(1); } doptimize(1); /* optimize definitions */ - if (spawned_children(nproc)) /* running in parent process? */ + i = spawned_children(nproc); /* create multiple processes if requested */ + if (i > 0) /* running in parent process? */ return(parent_loop() ? 0 : 1); - /* process & write rows */ + if (i < 0) /* running in output process? */ + return(output_loop() ? 0 : 1); + /* else we are a worker process */ return(combine_input() ? 0 : 1); stdin_error: fprintf(stderr, "%s: %s used for more than one input\n",