ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/Development/ray/src/util/rcomb.c

Comparing ray/src/util/rcomb.c (file contents):
Revision 2.8 by greg, Thu May 16 18:59:19 2024 UTC vs.
Revision 2.19 by greg, Mon Jun 3 18:55:51 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 first child while its siblings send it
10	+	* their record calculations.
11		*/
12
8	–	#include <errno.h>
13		#include <math.h>
14		#include "platform.h"
15	+	#include "rtprocess.h"
16		#include "rtio.h"
17		#include "rmatrix.h"
18		#include "calcomp.h"
14	–	#include "paths.h"
19
20		#ifndef M_PI
21		#define M_PI    3.14159265358979323846
#	Line 54 | Line 58 | int            cur_row;                        /* current input/output row */
58		int             cur_col;                        /* current input/output column */
59		int             cur_chan;                       /* if we're looping channels */
60
61	+	SUBPROC         *cproc = NULL;                  /* child process array */
62	+	int             nchildren = 0;                  /* # of child processes */
63	+	int             inchild = -1;                   /* our child ID (-1: parent) */
64	+
65		static int      checksymbolic(ROPMAT *rop);
66
67		static int
#	Line 390 | Line 398 | apply_op(RMATRIX *dst, const RMATRIX *src, const RUNAR
398		return(0);
399		rmx_free(res);
400		} else if (dst != src)
401	<	memcpy(dst->mtx, src->mtx,
394	<	sizeof(double)*dst->ncomp*dst->ncols*dst->nrows);
401	>	memcpy(dst->mtx, src->mtx, rmx_array_size(dst));
402		if (ro->nsf == dst->ncomp)
403		rmx_scale(dst, ro->sca);
404		return(1);
#	Line 538 | Line 545 | output_headinfo(FILE *fp)
545		}
546		}
547
548	+	static void
549	+	reset_inputs(void)
550	+	{
551	+	int     i = nmats;
552	+
553	+	fpurge(stdin);          /* discard previous matrix input */
554	+	while (i--) {
555	+	if (mop[i].infp != stdin)
556	+	fclose(mop[i].infp);    /* ! pclose() */
557	+	mop[i].infp = stdin;
558	+	mop[i].imx.dtype = DTdouble;
559	+	mop[i].imx.pflags &= ~RMF_SWAPIN;
560	+	}
561	+	}
562	+
563		static int
564	<	combine_input(ROPMAT *res, FILE *fout)
564	>	spawned_children(int np)
565		{
566	<	int     set_r, set_c;
567	<	RMATRIX *tmp = NULL;
568	<	int     co_set;
569	<	int     i;
570	<	/* allocate input row buffers */
566	>	size_t  recsize = 0;
567	>	int     i, rv;
568	>
569	>	#if defined(_WIN32) || defined(_WIN64)
570	>	if (np > 1) {
571	>	fputs("Warning: only one process under Windows\n", stderr);
572	>	np = 1;
573	>	} else
574	>	#endif
575	>	if ((in_nrows > 0) & (np*4 > in_nrows))
576	>	np = in_nrows/4;
577	>	/* we'll be doing a row at a time */
578		for (i = 0; i < nmats; i++) {
579	<	mop[i].imx.nrows = 1;   /* we'll be doing a row at a time */
579	>	mop[i].imx.nrows = 1;
580		if (!rmx_prepare(&mop[i].imx))
581		goto memerror;
582	+	recsize += rmx_array_size(&mop[i].imx);
583		if (mop[i].rmp != &mop[i].imx) {
584		mop[i].rmp->nrows = 1;
585		if (!rmx_prepare(mop[i].rmp))
586		goto memerror;
587		}
588		}
589	<	/* prep output row buffers */
590	<	if (mcat || res->preop.clen > 0) {
591	<	if (!split_input(res))  /* need separate buffer */
589	>	/* prep output row buffer */
590	>	if (mcat || mop[nmats].preop.clen > 0) {
591	>	if (!split_input(&mop[nmats]))  /* need separate buffer */
592		return(0);
593	<	if (res->preop.clen > 0)
594	<	res->rmp->ncomp = res->preop.clen / res->imx.ncomp;
595	<	res->rmp->nrows = 1;
596	<	if (!mcat | !mcat_last && !rmx_prepare(res->rmp))
593	>	if (mop[nmats].preop.clen > 0)
594	>	mop[nmats].rmp->ncomp = mop[nmats].preop.clen /
595	>	mop[nmats].imx.ncomp;
596	>	mop[nmats].rmp->nrows = 1;
597	>	if (!mcat | !mcat_last && !rmx_prepare(mop[nmats].rmp))
598		goto memerror;
599		}
600	+	mop[nmats].imx.nrows = 1;
601	+	if (!rmx_prepare(&mop[nmats].imx))
602	+	goto memerror;
603	+	if (np <= 1) {          /* single process return */
604	+	#ifdef getc_unlocked
605	+	for (i = 0; i < nmats; i++)
606	+	flockfile(mop[i].infp);
607	+	flockfile(stdout);
608	+	#endif
609	+	return(0);
610	+	}
611	+	fflush(stdout);         /* flush header & spawn children */
612	+	nchildren = np+1;
613	+	cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren);
614	+	if (!cproc)
615	+	goto memerror;
616	+	/* first child becomes stdout filter */
617	+	cproc[0].flags = PF_FILT_OUT;
618	+	cproc[0].w = fileno(stdout);
619	+	cproc[0].r = 0;
620	+	cproc[0].pid = -1;
621	+	rv = open_process(&cproc[i], NULL);
622	+	if (rv < 0) {
623	+	perror("fork");
624	+	exit(1);
625	+	}
626	+	if (rv == 0) {          /* loop if first child */
627	+	inchild = 0;
628	+	reset_inputs();
629	+	output_loop();
630	+	_exit(0);
631	+	}
632	+	/* start other children */
633	+	for (i = 1; (i < nchildren) & (rv > 0); i++) {
634	+	cproc[i].flags = 0;
635	+	cproc[i].w = 1;
636	+	cproc[i].r = 0;
637	+	cproc[i].pid = -1;
638	+	rv = open_process(&cproc[i], NULL);
639	+	}
640	+	if (rv < 0) {
641	+	perror("fork");
642	+	close_processes(cproc, i);
643	+	exit(1);
644	+	}
645	+	if (rv > 0)             /* parent return? */
646	+	return(1);
647	+	inchild = i;            /* our child index */
648	+	while (--i > 0)         /* don't share siblings' pipes */
649	+	close(cproc[i].w);
650	+	reset_inputs();
651	+	#ifdef getc_unlocked
652	+	flockfile(stdin);
653	+	#endif
654	+	return(0);              /* child return */
655	+	memerror:
656	+	fputs("Out of memory in spawned_children()\n", stderr);
657	+	exit(1);
658	+	}
659	+
660	+	static int
661	+	parent_loop(void)
662	+	{
663	+	int     i;
664	+
665	+	rmx_reset(&mop[nmats].imx);             /* not doing output side */
666	+	if (mop[nmats].rmp != &mop[nmats].imx)
667	+	rmx_reset(mop[nmats].rmp);
668	+	#ifdef getc_unlocked
669	+	for (i = 0; i < nmats; i++)
670	+	flockfile(mop[i].infp);
671	+	#endif
672	+	/* load & send rows to kids */
673	+	for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
674	+	int wfd = cproc[1 + cur_row%(nchildren-1)].w;
675	+	for (i = 0; i < nmats; i++)
676	+	if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
677	+	if (cur_row > in_nrows) /* unknown #input rows? */
678	+	break;
679	+	fprintf(stderr, "%s: read error at row %d\n",
680	+	mop[i].inspec, cur_row);
681	+	return(0);
682	+	}
683	+	if (i < nmats)
684	+	break;
685	+	for (i = 0; i < nmats; i++)
686	+	if (writebuf(wfd, mop[i].imx.mtx, rmx_array_size(&mop[i].imx))
687	+	!= rmx_array_size(&mop[i].imx))
688	+	return(0);
689	+	}
690	+	i = close_processes(cproc, nchildren);
691	+	free(cproc); cproc = NULL;
692	+	if (i < 0) {
693	+	fputs("Warning: missing child in parent_loop()\n", stderr);
694	+	return(1);
695	+	}
696	+	if (i > 0) {
697	+	fprintf(stderr, "Child exited with status %d\n", i);
698	+	return(0);
699	+	}
700	+	return(1);
701	+	memerror:
702	+	fputs("Out of memory in parent_loop()\n", stderr);
703	+	exit(1);
704	+	}
705	+
706	+	static int
707	+	combine_input(void)
708	+	{
709	+	const int       row0 = (inchild > 0)*(inchild-1);
710	+	const int       rstep = nchildren + !nchildren;
711	+	ROPMAT          *res = &mop[nmats];
712	+	int             set_r, set_c;
713	+	RMATRIX         *tmp = NULL;
714	+	int             co_set;
715	+	int             i;
716	+
717		if (mcat && mcat_last &&
718		!(tmp = rmx_alloc(1, res->imx.ncols, res->rmp->ncomp)))
719		goto memerror;
572	–	res->imx.nrows = 1;
573	–	if (!rmx_prepare(&res->imx))
574	–	goto memerror;
720		/* figure out what the user set */
721		co_set = fundefined("co");
722		if (!co_set)
#	Line 587 | Line 732 | combine_input(ROPMAT *res, FILE *fout)
732		} else                          /* save a little time */
733		set_r = set_c = 0;
734		/* read/process row-by-row */
735	<	for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
735	>	for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) {
736		RMATRIX     *mres = NULL;
737	<	for (i = 0; i < nmats; i++) {
737	>	for (i = 0; i < nmats; i++)
738		if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
739		if (cur_row > in_nrows) /* unknown #input rows? */
740	<	goto loop_exit;
740	>	break;
741		fprintf(stderr, "%s: read error at row %d\n",
742		mop[i].inspec, cur_row);
743		return(0);
744		}
745	+	if (i < nmats)
746	+	break;
747	+	for (i = 0; i < nmats; i++)
748		if (!apply_op(mop[i].rmp, &mop[i].imx, &mop[i].preop))
749		return(0);
602	–	}
750		if (set_r) varset("r", '=', cur_row);
751		for (cur_col = 0; cur_col < in_ncols; cur_col++) {
752		if (set_c) varset("c", '=', cur_col);
#	Line 636 | Line 783 | combine_input(ROPMAT *res, FILE *fout)
783		return(0);
784		}
785		rmx_free(mres); mres = NULL;
786	+	if (inchild > 0) {          /* children share stdout */
787	+	i = getc(stdin);        /* signals it's our turn */
788	+	if (i != EOF) ungetc(i, stdin);
789	+	}
790		if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp,
791	<	res->rmp->ncols, res->rmp->dtype, fout))
791	>	res->rmp->ncols, res->rmp->dtype, stdout))
792		return(0);
793	+	if (inchild > 0 && fflush(stdout) == EOF)
794	+	return(0);
795		}
796	<	loop_exit:
644	<	#if 0           /* we're about to exit, so who cares? */
645	<	rmx_free(tmp);                  /* clean up */
646	<	rmx_reset(res->rmp);
647	<	rmx_reset(&res->imx);
648	<	for (i = 0; i < nmats; i++) {
649	<	rmx_reset(mop[i].rmp);
650	<	rmx_reset(&mop[i].imx);
651	<	if (mop[i].inspec[0] == '!')
652	<	pclose(mop[i].infp);
653	<	else if (mop[i].inspec != stdin_name)
654	<	fclose(mop[i].infp);
655	<	mop[i].infp = NULL;
656	<	}
657	<	#endif
658	<	return(fflush(fout) != EOF);
796	>	return(inchild > 0 || fflush(stdout) != EOF);
797		memerror:
798		fputs("Out of buffer space in combine_input()\n", stderr);
799		return(0);
#	Line 705 | Line 843 | main(int argc, char *argv[])
843		const char      *defCsym = NULL;
844		int             echoheader = 1;
845		int             stdin_used = 0;
846	+	int             nproc = 1;
847		const char      *mcat_spec = NULL;
848		int             n2comp = 0;
849		uby8            comp_ndx[128];
#	Line 732 | Line 871 | main(int argc, char *argv[])
871		case 'h':
872		echoheader = !echoheader;
873		break;
874	+	case 'n':
875	+	nproc = atoi(argv[++i]);
876	+	if (nproc <= 0)
877	+	goto userr;
878	+	break;
879		case 'e':
880		if (!n) goto userr;
881		comp_ndx[n2comp++] = i++;
#	Line 780 | Line 924 | main(int argc, char *argv[])
924		if (n && !isflt(argv[i+1])) {
925		mop[nmats].preop.csym = argv[++i];
926		mop[nmats].preop.clen = 0;
927	+	mcat_last = 0;
928		break;
929		}
930		if (n > MAXCOMP*MAXCOMP) n = MAXCOMP*MAXCOMP;
#	Line 870 | Line 1015 | main(int argc, char *argv[])
1015		return(1);
1016		}
1017		doptimize(1);                   /* optimize definitions */
1018	+	if (spawned_children(nproc))    /* running in parent process? */
1019	+	return(parent_loop() ? 0 : 1);
1020		/* process & write rows */
1021	<	return(combine_input(&mop[nmats], stdout) ? 0 : 1);
1021	>	return(combine_input() ? 0 : 1);
1022		stdin_error:
1023		fprintf(stderr, "%s: %s used for more than one input\n",
1024		argv[0], stdin_name);
1025		return(1);
1026		userr:
1027		fprintf(stderr,
1028	<	"Usage: %s [-h][-f{adfc}][-e expr][-f file][-s sf .. | -c ce ..] m1 .. -m mcat > mres\n",
1028	>	"Usage: %s [-h][-f{adfc}][-n nproc][-e expr][-f file][-s sf .. | -c ce ..] m1 .. -m mcat > mres\n",
1029		argv[0]);
1030		return(1);
1031		}

Diff Legend

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