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.15 by greg, Thu May 23 01:28:07 2024 UTC vs.
Revision 2.18 by greg, Thu May 23 19:29:41 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 single parent.  Output to
9	+	* a shared stdout is assured because each child waits for parent
10	+	* to be ready with next input row before it outputs the previous result.
11	+	* The assumption (checked in spawn_children()) is that the parent
12	+	* will remain blocked until the child it is feeding has finished
13	+	* flushing the previous row to stdout.  The final output row is
14	+	* less guaranteed to be in order, so the parent sleeps for a few seconds
15	+	* between each child pipe closure, in hopes that this allows enough
16	+	* time for row output to finish in each process.
17		*/
18
8	–	#include <signal.h>
19		#include <math.h>
20		#include "platform.h"
21		#include "rtprocess.h"
#	Line 57 | Line 67 | int            cur_chan;                       /* if we're looping channels */
67		SUBPROC         *cproc = NULL;                  /* child process array */
68		int             nchildren = 0;                  /* # of child processes */
69		int             inchild = -1;                   /* our child ID (-1: parent) */
60	–	int             pgid = -1;                      /* process group ID */
61	–	int             nr_out = 0;                     /* # of rows output by kids */
70
71		static int      checksymbolic(ROPMAT *rop);
72
65	–	static void
66	–	on_sigio(int dummy)
67	–	{
68	–	nr_out++;                       /* happens when child outputs row */
69	–	}
70	–
73		static int
74		split_input(ROPMAT *rop)
75		{
#	Line 552 | Line 554 | output_headinfo(FILE *fp)
554		static int
555		spawned_children(int np)
556		{
557	+	size_t  recsize = 0;
558		int     i, rv;
559
560		#if defined(_WIN32) || defined(_WIN64)
#	Line 560 | Line 563 | spawned_children(int np)
563		np = 1;
564		} else
565		#endif
566	<	if ((in_nrows > 0) & (np > in_nrows))
567	<	np = in_nrows;
566	>	if ((in_nrows > 0) & (np*4 > in_nrows))
567	>	np = in_nrows/4;
568		/* we'll be doing a row at a time */
569		for (i = 0; i < nmats; i++) {
570		mop[i].imx.nrows = 1;
571		if (!rmx_prepare(&mop[i].imx))
572		goto memerror;
573	+	recsize += rmx_array_size(&mop[i].imx);
574		if (mop[i].rmp != &mop[i].imx) {
575		mop[i].rmp->nrows = 1;
576		if (!rmx_prepare(mop[i].rmp))
#	Line 587 | Line 591 | spawned_children(int np)
591		mop[nmats].imx.nrows = 1;
592		if (!rmx_prepare(&mop[nmats].imx))
593		goto memerror;
594	<	if (np <= 1) {          /* single process return point */
594	>	if (np <= 1) {          /* single process return */
595		#ifdef getc_unlocked
596		for (i = 0; i < nmats; i++)
597		flockfile(mop[i].infp);
#	Line 595 | Line 599 | spawned_children(int np)
599		#endif
600		return(0);
601		}
598	–	pgid = setpgrp();       /* set process group ID */
599	–	signal(SIGIO, on_sigio);
602		fflush(stdout);         /* flush header & spawn children */
603		cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*np);
604		if (!cproc)
#	Line 609 | Line 611 | spawned_children(int np)
611		cproc[i].pid = -1;
612		rv = open_process(&cproc[i], NULL);
613		if (rv <= 0) break;
614	+	if (!i && 2*rv >= recsize) {
615	+	fputs("Problem too small for multi-processing\n",
616	+	stderr);
617	+	close_processes(cproc, 1);
618	+	exit(1);
619	+	}
620		}
613	–	if (rv > 0)
614	–	return(1);      /* parent return value */
621		if (rv < 0) {
622		perror("fork");
623	+	close_processes(cproc, i);
624		exit(1);
625		}
626	+	if (rv > 0)             /* parent return? */
627	+	return(1);
628		inchild = i;            /* our child index */
629		while (i-- > 0)         /* don't share siblings' pipes */
630		close(cproc[i].w);
#	Line 628 | Line 637 | spawned_children(int np)
637		fclose(mop[i].infp);    /* ! pclose() */
638		mop[i].infp = stdin;
639		mop[i].imx.dtype = DTdouble;
640	+	mop[i].imx.pflags &= ~RMF_SWAPIN;
641		}
642		return(0);              /* child return */
643		memerror:
#	Line 672 | Line 682 | parent_loop()
682		if (fflush(ofp) == EOF)
683		return(0);
684		}
685	<	for (i = 0; i < nchildren; i++)
685	>	for (i = 0; i < nchildren; i++) {       /* maintain output order */
686	>	sleep(1+(mop[nmats].rmp->ncols*mop[nmats].rmp->ncomp >> 15));
687		fclose(outfp[i]);
688	+	}
689		free(outfp);
690		i = close_processes(cproc, nchildren);
691		free(cproc); cproc = NULL;
#	Line 700 | Line 712 | combine_input()
712		int             set_r, set_c;
713		RMATRIX         *tmp = NULL;
714		int             co_set;
703	–	sigset_t        iomask;
715		int             i;
716
717		if (mcat && mcat_last &&
#	Line 721 | Line 732 | combine_input()
732		} else                          /* save a little time */
733		set_r = set_c = 0;
734
735	<	sigemptyset(&iomask);           /* read/process row-by-row */
725	<	sigaddset(&iomask, SIGIO);
735	>	/* read/process row-by-row */
736		for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) {
737		RMATRIX     *mres = NULL;
728	–	if (inchild >= 0) sigprocmask(SIG_BLOCK, &iomask, NULL);
738		for (i = 0; i < nmats; i++)
739		if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
740		if (cur_row > in_nrows) /* unknown #input rows? */
#	Line 734 | Line 743 | combine_input()
743		mop[i].inspec, cur_row);
744		return(0);
745		}
737	–	if (inchild >= 0) sigprocmask(SIG_UNBLOCK, &iomask, NULL);
746		if (i < nmats)
747		break;
748		for (i = 0; i < nmats; i++)
#	Line 777 | Line 785 | combine_input()
785		}
786		rmx_free(mres); mres = NULL;
787		if (inchild >= 0) {         /* children share stdout */
788	<	while (nr_out < cur_row)
789	<	pause();            /* wait for our turn */
782	<	sigprocmask(SIG_BLOCK, &iomask, NULL);
788	>	i = getc(stdin);        /* signals it's our turn */
789	>	if (i != EOF) ungetc(i, stdin);
790		}
791		if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp,
792		res->rmp->ncols, res->rmp->dtype, stdout))
793		return(0);
794	<	if (inchild >= 0) {         /* flush and notify group */
795	<	if (fflush(stdout) == EOF)
789	<	return(0);
790	<	sigprocmask(SIG_UNBLOCK, &iomask, NULL);
791	<	killpg(pgid, SIGIO);    /* increments everyone's nr_out */
792	<	}
794	>	if (inchild >= 0 && fflush(stdout) == EOF)
795	>	return(0);
796		}
797		return(inchild >= 0 || fflush(stdout) != EOF);
798		memerror:

Diff Legend

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