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.19 by greg, Mon Jun 3 18:55:51 2024 UTC vs.
Revision 2.20 by greg, Tue Jun 4 21:23:11 2024 UTC

#	Line 6 | Line 6 | static const char RCSid[] = "$Id$";
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
9	>	* and output to stdout is sorted by last child while its siblings send it
10		* their record calculations.
11		*/
12
#	Line 545 | Line 545 | output_headinfo(FILE *fp)
545		}
546		}
547
548	<	static void
549	<	reset_inputs(void)
548	>	static int
549	>	output_loop(void)
550		{
551	<	int     i = nmats;
551	>	const size_t    row_size = rmx_array_size(mop[nmats].rmp);
552	>	int             i = nmats;
553	>	int             cur_child = 0;
554
555	<	fpurge(stdin);          /* discard previous matrix input */
556	<	while (i--) {
557	<	if (mop[i].infp != stdin)
558	<	fclose(mop[i].infp);    /* ! pclose() */
559	<	mop[i].infp = stdin;
560	<	mop[i].imx.dtype = DTdouble;
561	<	mop[i].imx.pflags &= ~RMF_SWAPIN;
555	>	if (mop[nmats].rmp != &mop[nmats].imx)          /* output is split? */
556	>	rmx_reset(&mop[nmats].imx);
557	>	while (i-- > 0) {                               /* close input matrices */
558	>	fclose(mop[i].infp);            /* ! pclose() */
559	>	mop[i].infp = NULL;
560	>	rmx_reset(&mop[i].imx);
561	>	if (mop[i].rmp != &mop[i].imx) {
562	>	rmx_free(mop[i].rmp);
563	>	mop[i].rmp = &mop[i].imx;
564	>	}
565		}
566	+	#ifdef getc_unlocked
567	+	flockfile(stdout);                              /* we own this, now */
568	+	#endif
569	+	for ( ; ; ) {                                   /* loop until no more */
570	+	ssize_t         rv;
571	+	rv = readbuf(cproc[cur_child].r, mop[nmats].rmp->mtx, row_size);
572	+	if (!rv)                                /* out of rows? */
573	+	break;
574	+	if (rv != row_size) {
575	+	fputs("Read error in output loop\n", stderr);
576	+	return(0);
577	+	}                                       /* do final conversion */
578	+	if (!rmx_write_data(mop[nmats].rmp->mtx, mop[nmats].rmp->ncomp,
579	+	mop[nmats].rmp->ncols, mop[nmats].rmp->dtype, stdout)) {
580	+	fputs("Conversion/write error in output loop\n", stderr);
581	+	return(0);
582	+	}
583	+	cur_child++;
584	+	cur_child *= (cur_child < inchild);
585	+	}
586	+	return(fflush(stdout) != EOF);
587		}
588
589		static int
590		spawned_children(int np)
591		{
566	–	size_t  recsize = 0;
592		int     i, rv;
593
594		#if defined(_WIN32) || defined(_WIN64)
#	Line 579 | Line 604 | spawned_children(int np)
604		mop[i].imx.nrows = 1;
605		if (!rmx_prepare(&mop[i].imx))
606		goto memerror;
582	–	recsize += rmx_array_size(&mop[i].imx);
607		if (mop[i].rmp != &mop[i].imx) {
608		mop[i].rmp->nrows = 1;
609		if (!rmx_prepare(mop[i].rmp))
610		goto memerror;
611		}
612		}
613	<	/* prep output row buffer */
613	>	/* prep output row buffer(s) */
614		if (mcat || mop[nmats].preop.clen > 0) {
615		if (!split_input(&mop[nmats]))  /* need separate buffer */
616		return(0);
617		if (mop[nmats].preop.clen > 0)
618		mop[nmats].rmp->ncomp = mop[nmats].preop.clen /
619		mop[nmats].imx.ncomp;
596	–	mop[nmats].rmp->nrows = 1;
597	–	if (!mcat | !mcat_last && !rmx_prepare(mop[nmats].rmp))
598	–	goto memerror;
620		}
621		mop[nmats].imx.nrows = 1;
622		if (!rmx_prepare(&mop[nmats].imx))
623		goto memerror;
624	+	if (mop[nmats].rmp != &mop[nmats].imx) {
625	+	mop[nmats].rmp->nrows = 1;
626	+	if (!rmx_prepare(mop[nmats].rmp))
627	+	goto memerror;
628	+	}
629		if (np <= 1) {          /* single process return */
630		#ifdef getc_unlocked
631		for (i = 0; i < nmats; i++)
#	Line 609 | Line 635 | spawned_children(int np)
635		return(0);
636		}
637		fflush(stdout);         /* flush header & spawn children */
638	<	nchildren = np+1;
638	>	nchildren = np + 1;     /* extra child to sequence output */
639		cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren);
640		if (!cproc)
641		goto memerror;
642	<	/* first child becomes stdout filter */
643	<	cproc[0].flags = PF_FILT_OUT;
644	<	cproc[0].w = fileno(stdout);
645	<	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;
642	>	for (i = nchildren; i--; ) cproc[i] = sp_inactive;
643	>	cproc[nchildren-1].flags |= PF_FILT_OUT;
644	>	/* start each child */
645	>	for (i = 0; i < nchildren; i++) {
646		rv = open_process(&cproc[i], NULL);
647	+	if (rv <= 0) break;
648		}
649		if (rv < 0) {
650		perror("fork");
651		close_processes(cproc, i);
652		exit(1);
653		}
654	<	if (rv > 0)             /* parent return? */
655	<	return(1);
654	>	if (rv) {               /* are we the parent? */
655	>	i = nchildren-1;        /* last child is sole reader */
656	>	while (i-- > 0)
657	>	close(cproc[i].r);
658	>	return(1);      /* parent return value */
659	>	}
660		inchild = i;            /* our child index */
661	<	while (--i > 0)         /* don't share siblings' pipes */
661	>	while (i-- > 0)         /* only parent writes siblings */
662		close(cproc[i].w);
663	<	reset_inputs();
663	>
664	>	if (inchild == nchildren-1)     /* last child sequences output */
665	>	exit(output_loop() ? 0 : 1);
666	>
667	>	i = inchild;            /* won't read from siblings */
668	>	while (i-- > 0)
669	>	close(cproc[i].r);
670	>	i = nmats;              /* close input matrix streams */
671	>	while (i-- > 0) {
672	>	if (mop[i].infp != stdin)
673	>	fclose(mop[i].infp);    /* ! pclose() */
674	>	mop[i].infp = stdin;
675	>	mop[i].imx.dtype = DTrmx_native;
676	>	mop[i].imx.pflags &= ~RMF_SWAPIN;
677	>	}
678	>	fpurge(stdin);          /* discard any previous matrix input */
679		#ifdef getc_unlocked
680		flockfile(stdin);
681		#endif
682	<	return(0);              /* child return */
682	>	mop[nmats].rmp->dtype = DTrmx_native;
683	>	return(0);              /* worker child return value */
684		memerror:
685		fputs("Out of memory in spawned_children()\n", stderr);
686		exit(1);
#	Line 662 | Line 691 | parent_loop(void)
691		{
692		int     i;
693
694	<	rmx_reset(&mop[nmats].imx);             /* not doing output side */
695	<	if (mop[nmats].rmp != &mop[nmats].imx)
696	<	rmx_reset(mop[nmats].rmp);
694	>	rmx_reset(&mop[nmats].imx);             /* not touching output side */
695	>	if (mop[nmats].rmp != &mop[nmats].imx) {
696	>	rmx_free(mop[nmats].rmp);
697	>	mop[nmats].rmp = &mop[nmats].imx;
698	>	}
699		#ifdef getc_unlocked
700	<	for (i = 0; i < nmats; i++)
700	>	for (i = 0; i < nmats; i++)             /* we handle matrix inputs */
701		flockfile(mop[i].infp);
702		#endif
703		/* load & send rows to kids */
704		for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) {
705	<	int wfd = cproc[1 + cur_row%(nchildren-1)].w;
705	>	int         wfd = cproc[cur_row % (nchildren-1)].w;
706		for (i = 0; i < nmats; i++)
707		if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) {
708		if (cur_row > in_nrows) /* unknown #input rows? */
709		break;
710	<	fprintf(stderr, "%s: read error at row %d\n",
710	>	fprintf(stderr, "%s: parent read error at row %d\n",
711		mop[i].inspec, cur_row);
712		return(0);
713		}
#	Line 688 | Line 719 | parent_loop(void)
719		return(0);
720		}
721		i = close_processes(cproc, nchildren);
722	<	free(cproc); cproc = NULL;
722	>	free(cproc); cproc = NULL; nchildren = 0;
723		if (i < 0) {
724	<	fputs("Warning: missing child in parent_loop()\n", stderr);
724	>	fputs("Warning: lost child in parent_loop()\n", stderr);
725		return(1);
726		}
727		if (i > 0) {
728		fprintf(stderr, "Child exited with status %d\n", i);
729		return(0);
730		}
731	<	return(1);
731	>	return(1);                              /* return success! */
732		memerror:
733		fputs("Out of memory in parent_loop()\n", stderr);
734		exit(1);
#	Line 706 | Line 737 | memerror:
737		static int
738		combine_input(void)
739		{
740	<	const int       row0 = (inchild > 0)*(inchild-1);
741	<	const int       rstep = nchildren + !nchildren;
740	>	const int       row0 = (inchild >= 0)*inchild;
741	>	const int       rstep = nchildren ? nchildren-1 : 1;
742		ROPMAT          *res = &mop[nmats];
743		int             set_r, set_c;
744		RMATRIX         *tmp = NULL;
#	Line 783 | Line 814 | combine_input(void)
814		return(0);
815		}
816		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	–	}
817		if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp,
818		res->rmp->ncols, res->rmp->dtype, stdout))
819		return(0);
820	<	if (inchild > 0 && fflush(stdout) == EOF)
820	>	if (inchild >= 0 && fflush(stdout) == EOF)
821		return(0);
822		}
823	<	return(inchild > 0 || fflush(stdout) != EOF);
823	>	return(inchild >= 0 || fflush(stdout) != EOF);
824		memerror:
825		fputs("Out of buffer space in combine_input()\n", stderr);
826		return(0);

Diff Legend

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