--- ray/src/util/rcomb.c	2023/12/18 23:04:05	2.2
+++ ray/src/util/rcomb.c	2024/06/03 18:55:51	2.19
@@ -1,18 +1,21 @@
 #ifndef lint
-static const char RCSid[] = "$Id: rcomb.c,v 2.2 2023/12/18 23:04:05 greg Exp $";
+static const char RCSid[] = "$Id: rcomb.c,v 2.19 2024/06/03 18:55:51 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 first child while its siblings send it
+ * their record calculations.
  */
 
-#include <errno.h>
 #include <math.h>
 #include "platform.h"
+#include "rtprocess.h"
 #include "rtio.h"
-#include "resolu.h"
 #include "rmatrix.h"
 #include "calcomp.h"
-#include "paths.h"
 
 #ifndef M_PI
 #define M_PI	3.14159265358979323846
@@ -55,6 +58,10 @@ int		cur_row;			/* current input/output row */
 int		cur_col;			/* current input/output column */
 int		cur_chan;			/* if we're looping channels */
 
+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);
 
 static int
@@ -186,6 +193,7 @@ checksymbolic(ROPMAT *rop)
 {
 	const int	nc = rop->imx.ncomp;
 	const int	dt = rop->imx.dtype;
+	double		cf = 1;
 	int		i, j;
 					/* check suffix => reference file */
 	if (strchr(rop->preop.csym, '.') > rop->preop.csym)
@@ -206,28 +214,37 @@ checksymbolic(ROPMAT *rop)
 		int	comp = 0;
 		switch (rop->preop.csym[j]) {
 		case 'B':
+		case 'b':
 			++comp;
 			/* fall through */
 		case 'G':
+		case 'g':
 			++comp;
 			/* fall through */
 		case 'R':
+		case 'r':
+			if (rop->preop.csym[j] <= 'Z')
+				cf = 1./WHTEFFICACY;
 			if (dt == DTxyze) {
 				for (i = 3; i--; )
-					rop->preop.cmat[j*nc+i] = 1./WHTEFFICACY *
-							xyz2rgbmat[comp][i];
+					rop->preop.cmat[j*nc+i] = cf*xyz2rgbmat[comp][i];
 			} else if (nc == 3)
 				rop->preop.cmat[j*nc+comp] = 1.;
 			else
 				rgbrow(rop, j, comp);
 			break;
 		case 'Z':
+		case 'z':
 			++comp;
 			/* fall through */
 		case 'Y':
+		case 'y':
 			++comp;
 			/* fall through */
 		case 'X':
+		case 'x':
+			if ((rop->preop.csym[j] <= 'Z') & (dt != DTxyze))
+				cf = WHTEFFICACY;
 			if (dt == DTxyze) {
 				rop->preop.cmat[j*nc+comp] = 1.;
 			} else if (nc == 3) {
@@ -239,20 +256,27 @@ checksymbolic(ROPMAT *rop)
 			else
 				xyzrow(rop, j, comp);
 
-			for (i = nc*(dt != DTxyze); i--; )
-				rop->preop.cmat[j*nc+i] *= WHTEFFICACY;
+			for (i = nc*(cf != 1); i--; )
+				rop->preop.cmat[j*nc+i] *= cf;
 			break;
 		case 'S':		/* scotopic (il)luminance */
+			cf = WHTSCOTOPIC;
+			/* fall through */
+		case 's':
 			sensrow(rop, j, scolor2scotopic);
-			for (i = nc; i--; )
-				rop->preop.cmat[j*nc+i] *= WHTSCOTOPIC;
+			for (i = nc*(cf != 1); i--; )
+				rop->preop.cmat[j*nc+i] *= cf;
 			break;
 		case 'M':		/* melanopic (il)luminance */
+			cf = WHTMELANOPIC;
+			/* fall through */
+		case 'm':
 			sensrow(rop, j, scolor2melanopic);
-			for (i = nc; i--; )
-				rop->preop.cmat[j*nc+i] *= WHTMELANOPIC;
+			for (i = nc*(cf != 1); i--; )
+				rop->preop.cmat[j*nc+i] *= cf;
 			break;
 		case 'A':		/* average component */
+		case 'a':
 			for (i = nc; i--; )
 				rop->preop.cmat[j*nc+i] = 1./(double)nc;
 			break;
@@ -267,10 +291,10 @@ checksymbolic(ROPMAT *rop)
 	memcpy(rop->rmp->wlpart, WLPART, sizeof(rop->rmp->wlpart));
 	rop->rmp->ncomp = rop->preop.clen / nc;
 					/* decide on output type */
-	if (!strcmp(rop->preop.csym, "XYZ")) {
+	if (!strcasecmp(rop->preop.csym, "XYZ")) {
 		if (dt <= DTspec)
 			rop->rmp->dtype = DTxyze;
-	} else if (!strcmp(rop->preop.csym, "RGB")) {
+	} else if (!strcasecmp(rop->preop.csym, "RGB")) {
 		if (dt <= DTspec)
 			rop->rmp->dtype = DTrgbe;
 	} else if (rop->rmp->dtype == DTspec)
@@ -374,8 +398,7 @@ apply_op(RMATRIX *dst, const RMATRIX *src, const RUNAR
 			return(0);
 		rmx_free(res);
 	} else if (dst != src)
-		memcpy(dst->mtx, src->mtx,
-				sizeof(double)*dst->ncomp*dst->ncols*dst->nrows);
+		memcpy(dst->mtx, src->mtx, rmx_array_size(dst));
 	if (ro->nsf == dst->ncomp)
 		rmx_scale(dst, ro->sca);
 	return(1);
@@ -522,40 +545,178 @@ output_headinfo(FILE *fp)
 	}
 }
 
+static void
+reset_inputs(void)
+{
+	int	i = nmats;
+
+	fpurge(stdin);		/* discard previous matrix input */
+	while (i--) {
+		if (mop[i].infp != stdin)
+			fclose(mop[i].infp);	/* ! pclose() */
+		mop[i].infp = stdin;
+		mop[i].imx.dtype = DTdouble;
+		mop[i].imx.pflags &= ~RMF_SWAPIN;
+	}
+}
+
 static int
-combine_input(ROPMAT *res, FILE *fout)
+spawned_children(int np)
 {
-	int	set_r, set_c;
-	RMATRIX	*tmp = NULL;
-	int	co_set;
-	int	i;
-					/* allocate input row buffers */
+	size_t	recsize = 0;
+	int	i, rv;
+
+#if defined(_WIN32) || defined(_WIN64)
+	if (np > 1) {
+		fputs("Warning: only one process under Windows\n", stderr);
+		np = 1;
+	} else
+#endif
+	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;	/* we'll be doing a row at a time */
+		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 buffers */
-	if (mcat || res->preop.clen > 0) {
-		if (!split_input(res))	/* need separate buffer */
+				/* prep output row buffer */
+	if (mcat || mop[nmats].preop.clen > 0) {
+		if (!split_input(&mop[nmats]))	/* need separate buffer */
 			return(0);
-		if (res->preop.clen > 0)
-			res->rmp->ncomp = res->preop.clen / res->imx.ncomp;
-		res->rmp->nrows = 1;
-		if (!mcat | !mcat_last && !rmx_prepare(res->rmp))
+		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].imx.nrows = 1;
+	if (!rmx_prepare(&mop[nmats].imx))
+		goto memerror;
+	if (np <= 1) {		/* single process return */
+#ifdef getc_unlocked
+		for (i = 0; i < nmats; i++)
+			flockfile(mop[i].infp);
+		flockfile(stdout);
+#endif
+		return(0);
+	}
+	fflush(stdout);		/* flush header & spawn children */
+	nchildren = np+1;
+	cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren);
+	if (!cproc)
+		goto memerror;
+				/* first child becomes stdout filter */
+	cproc[0].flags = PF_FILT_OUT;
+	cproc[0].w = fileno(stdout);
+	cproc[0].r = 0;
+	cproc[0].pid = -1;
+	rv = open_process(&cproc[i], NULL);
+	if (rv < 0) {
+		perror("fork");
+		exit(1);
+	}
+	if (rv == 0) {		/* loop if first child */
+		inchild = 0;
+		reset_inputs();
+		output_loop();
+		_exit(0);
+	}
+				/* start other children */
+	for (i = 1; (i < nchildren) & (rv > 0); i++) {
+		cproc[i].flags = 0;
+		cproc[i].w = 1;
+		cproc[i].r = 0;
+		cproc[i].pid = -1;
+		rv = open_process(&cproc[i], NULL);
+	}
+	if (rv < 0) {
+		perror("fork");
+		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 */
+		close(cproc[i].w);
+	reset_inputs();
+#ifdef getc_unlocked
+	flockfile(stdin);
+#endif
+	return(0);		/* child return */
+memerror:
+	fputs("Out of memory in spawned_children()\n", stderr);
+	exit(1);
+}
+
+static int
+parent_loop(void)
+{
+	int	i;
+
+	rmx_reset(&mop[nmats].imx);		/* not doing output side */
+	if (mop[nmats].rmp != &mop[nmats].imx)
+		rmx_reset(mop[nmats].rmp);
+#ifdef getc_unlocked
+	for (i = 0; i < nmats; i++)
+		flockfile(mop[i].infp);
+#endif
+						/* load & send rows to kids */
+	for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
+	    int	wfd = cproc[1 + 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",
+					mop[i].inspec, cur_row);
+			return(0);
+		}
+	    if (i < nmats)
+		break;
+	    for (i = 0; i < nmats; i++)
+	    	if (writebuf(wfd, mop[i].imx.mtx, rmx_array_size(&mop[i].imx))
+	    				!= rmx_array_size(&mop[i].imx))
+			return(0);
+	}
+	i = close_processes(cproc, nchildren);
+	free(cproc); cproc = NULL;
+	if (i < 0) {
+		fputs("Warning: missing child in parent_loop()\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);
+}
+
+static int
+combine_input(void)
+{
+	const int	row0 = (inchild > 0)*(inchild-1);
+	const int	rstep = nchildren + !nchildren;
+	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;
-	res->imx.nrows = 1;
-	if (!rmx_prepare(&res->imx))
-		goto memerror;
 					/* figure out what the user set */
 	co_set = fundefined("co");
 	if (!co_set)
@@ -571,19 +732,21 @@ combine_input(ROPMAT *res, FILE *fout)
 	} else				/* save a little time */
 		set_r = set_c = 0;
 					/* read/process row-by-row */
-	for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
+	for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) {
 	    RMATRIX	*mres = NULL;
-	    for (i = 0; i < nmats; i++) {
+	    for (i = 0; i < nmats; i++)
 		if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
-			if (in_nrows <= 0)	/* normal end? */
-				goto loop_exit;
+			if (cur_row > in_nrows)	/* unknown #input rows? */
+				break;
 			fprintf(stderr, "%s: read error at row %d\n",
 					mop[i].inspec, cur_row);
 			return(0);
 		}
+	    if (i < nmats)
+	    	break;
+	    for (i = 0; i < nmats; i++)
 		if (!apply_op(mop[i].rmp, &mop[i].imx, &mop[i].preop))
 			return(0);
-	    }
 	    if (set_r) varset("r", '=', cur_row);
 	    for (cur_col = 0; cur_col < in_ncols; cur_col++) {
 	    	if (set_c) varset("c", '=', cur_col);
@@ -620,26 +783,17 @@ combine_input(ROPMAT *res, FILE *fout)
 			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, fout))
+	    			res->rmp->ncols, res->rmp->dtype, stdout))
 	    	return(0);
+	    if (inchild > 0 && fflush(stdout) == EOF)
+	    	return(0);
 	}
-loop_exit:
-#if 0		/* we're about to exit, so who cares? */
-	rmx_free(tmp);			/* clean up */
-	rmx_reset(res->rmp);
-	rmx_reset(&res->imx);
-	for (i = 0; i < nmats; i++) {
-		rmx_reset(mop[i].rmp);
-		rmx_reset(&mop[i].imx);
-		if (mop[i].inspec[0] == '!')
-			pclose(mop[i].infp);
-		else if (mop[i].inspec != stdin_name)
-			fclose(mop[i].infp);
-		mop[i].infp = NULL;
-	}
-#endif
-	return(fflush(fout) != EOF);
+	return(inchild > 0 || fflush(stdout) != EOF);
 memerror:
 	fputs("Out of buffer space in combine_input()\n", stderr);
 	return(0);
@@ -663,7 +817,9 @@ resize_inparr(int n2alloc)
 {
 	int	i;
 
-	for (i = nmats; i > n2alloc; i--) {
+	if (n2alloc == nall)
+		return;
+	for (i = nall; i > n2alloc; i--) {
 		rmx_reset(&mop[i].imx);
 		if (mop[i].rmp != &mop[i].imx)
 			rmx_free(mop[i].rmp);
@@ -673,8 +829,8 @@ resize_inparr(int n2alloc)
 		fputs("Out of memory in resize_inparr()\n", stderr);
 		exit(1);
 	}
-	if (n2alloc > nmats)
-		memset(mop+nmats, 0, (n2alloc-nmats)*sizeof(ROPMAT));
+	if (n2alloc > nall)
+		memset(mop+nall, 0, (n2alloc-nall)*sizeof(ROPMAT));
 	nall = n2alloc;
 }
 
@@ -687,6 +843,7 @@ main(int argc, char *argv[])
 	const char	*defCsym = NULL;
 	int		echoheader = 1;
 	int		stdin_used = 0;
+	int		nproc = 1;
 	const char	*mcat_spec = NULL;
 	int		n2comp = 0;
 	uby8		comp_ndx[128];
@@ -714,6 +871,11 @@ main(int argc, char *argv[])
 			case 'h':
 				echoheader = !echoheader;
 				break;
+			case 'n':
+				nproc = atoi(argv[++i]);
+				if (nproc <= 0)
+					goto userr;
+				break;
 			case 'e':
 				if (!n) goto userr;
 				comp_ndx[n2comp++] = i++;
@@ -762,6 +924,7 @@ main(int argc, char *argv[])
 				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;
@@ -851,15 +1014,18 @@ main(int argc, char *argv[])
 		fprintf(stderr, "%s: unsupported output format\n", argv[0]);
 		return(1);
 	}
+	doptimize(1);			/* optimize definitions */
+	if (spawned_children(nproc))	/* running in parent process? */
+		return(parent_loop() ? 0 : 1);
 					/* process & write rows */
-	return(combine_input(&mop[nmats], stdout) ? 0 : 1);
+	return(combine_input() ? 0 : 1);
 stdin_error:
 	fprintf(stderr, "%s: %s used for more than one input\n",
 			argv[0], stdin_name);
 	return(1);
 userr:
 	fprintf(stderr,
-	"Usage: %s [-h][-f{adfc}][-e expr][-f file][-s sf .. | -c ce ..] m1 .. -m mcat > mres\n",
+	"Usage: %s [-h][-f{adfc}][-n nproc][-e expr][-f file][-s sf .. | -c ce ..] m1 .. -m mcat > mres\n",
 			argv[0]);
 	return(1);
 }