| 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. |
| 8 |
> |
* on one input row at a time, fed by the original process. Final conversion |
| 9 |
> |
* and output to stdout is sorted by last child while its siblings send it |
| 10 |
> |
* their record calculations. |
| 11 |
|
*/ |
| 12 |
|
|
| 13 |
|
#include <math.h> |
| 546 |
|
} |
| 547 |
|
|
| 548 |
|
static int |
| 549 |
+ |
output_loop(void) |
| 550 |
+ |
{ |
| 551 |
+ |
const size_t row_size = rmx_array_size(mop[nmats].rmp); |
| 552 |
+ |
int i = nmats; |
| 553 |
+ |
int cur_child = 0; |
| 554 |
+ |
|
| 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 |
|
{ |
| 557 |
– |
size_t recsize = 0; |
| 592 |
|
int i, rv; |
| 593 |
|
|
| 594 |
|
#if defined(_WIN32) || defined(_WIN64) |
| 604 |
|
mop[i].imx.nrows = 1; |
| 605 |
|
if (!rmx_prepare(&mop[i].imx)) |
| 606 |
|
goto memerror; |
| 573 |
– |
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; |
| 587 |
– |
mop[nmats].rmp->nrows = 1; |
| 588 |
– |
if (!mcat | !mcat_last && !rmx_prepare(mop[nmats].rmp)) |
| 589 |
– |
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++) |
| 635 |
|
return(0); |
| 636 |
|
} |
| 637 |
|
fflush(stdout); /* flush header & spawn children */ |
| 638 |
< |
cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*np); |
| 638 |
> |
nchildren = np + 1; /* extra child to sequence output */ |
| 639 |
> |
cproc = (SUBPROC *)malloc(sizeof(SUBPROC)*nchildren); |
| 640 |
|
if (!cproc) |
| 641 |
|
goto memerror; |
| 642 |
< |
nchildren = np; |
| 643 |
< |
for (i = 0; i < np; i++) { |
| 644 |
< |
cproc[i].flags = PF_FILT_OUT; |
| 645 |
< |
cproc[i].w = dup(1); |
| 610 |
< |
cproc[i].r = 0; |
| 611 |
< |
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; |
| 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 |
– |
} |
| 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 |
> |
cproc[i].r = -1; |
| 659 |
> |
} |
| 660 |
> |
return(1); /* parent return value */ |
| 661 |
> |
} |
| 662 |
|
inchild = i; /* our child index */ |
| 663 |
< |
while (i-- > 0) /* don't share siblings' pipes */ |
| 663 |
> |
while (i-- > 0) /* only parent writes siblings */ |
| 664 |
|
close(cproc[i].w); |
| 665 |
< |
fpurge(stdin); /* discard previous matrix input */ |
| 666 |
< |
#ifdef getc_unlocked |
| 667 |
< |
flockfile(stdin); |
| 668 |
< |
#endif |
| 669 |
< |
for (i = 0; i < nmats; i++) { |
| 665 |
> |
|
| 666 |
> |
if (inchild == nchildren-1) /* last child sequences output */ |
| 667 |
> |
exit(output_loop() ? 0 : 1); |
| 668 |
> |
|
| 669 |
> |
i = inchild; /* won't read from siblings */ |
| 670 |
> |
while (i-- > 0) |
| 671 |
> |
close(cproc[i].r); |
| 672 |
> |
i = nmats; /* redirect input matrix streams */ |
| 673 |
> |
while (i-- > 0) { |
| 674 |
|
if (mop[i].infp != stdin) |
| 675 |
|
fclose(mop[i].infp); /* ! pclose() */ |
| 676 |
|
mop[i].infp = stdin; |
| 677 |
< |
mop[i].imx.dtype = DTdouble; |
| 677 |
> |
mop[i].imx.dtype = DTrmx_native; |
| 678 |
|
mop[i].imx.pflags &= ~RMF_SWAPIN; |
| 679 |
|
} |
| 680 |
< |
return(0); /* child return */ |
| 680 |
> |
fpurge(stdin); /* discard any previous matrix input */ |
| 681 |
> |
#ifdef getc_unlocked |
| 682 |
> |
flockfile(stdin); |
| 683 |
> |
#endif |
| 684 |
> |
mop[nmats].rmp->dtype = DTrmx_native; |
| 685 |
> |
return(0); /* worker child return value */ |
| 686 |
|
memerror: |
| 687 |
|
fputs("Out of memory in spawned_children()\n", stderr); |
| 688 |
|
exit(1); |
| 689 |
|
} |
| 690 |
|
|
| 691 |
|
static int |
| 692 |
< |
parent_loop() |
| 692 |
> |
parent_loop(void) |
| 693 |
|
{ |
| 651 |
– |
FILE **outfp = (FILE **)malloc(nchildren*sizeof(FILE *)); |
| 694 |
|
int i; |
| 695 |
|
|
| 696 |
< |
if (!outfp) goto memerror; |
| 697 |
< |
for (i = 0; i < nchildren; i++) { |
| 698 |
< |
outfp[i] = fdopen(cproc[i].w, "w"); |
| 699 |
< |
if (!outfp[i]) goto memerror; |
| 658 |
< |
#ifdef getc_unlocked |
| 659 |
< |
flockfile(outfp[i]); |
| 660 |
< |
#endif |
| 696 |
> |
rmx_reset(&mop[nmats].imx); /* not touching output side */ |
| 697 |
> |
if (mop[nmats].rmp != &mop[nmats].imx) { |
| 698 |
> |
rmx_free(mop[nmats].rmp); |
| 699 |
> |
mop[nmats].rmp = &mop[nmats].imx; |
| 700 |
|
} |
| 701 |
|
#ifdef getc_unlocked |
| 702 |
< |
for (i = 0; i < nmats; i++) |
| 702 |
> |
for (i = 0; i < nmats; i++) /* we handle matrix inputs */ |
| 703 |
|
flockfile(mop[i].infp); |
| 704 |
|
#endif |
| 705 |
+ |
/* load & send rows to kids */ |
| 706 |
|
for (cur_row = 0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row++) { |
| 707 |
< |
FILE *ofp = outfp[cur_row % nchildren]; |
| 707 |
> |
int wfd = cproc[cur_row % (nchildren-1)].w; |
| 708 |
|
for (i = 0; i < nmats; i++) |
| 709 |
|
if (!rmx_load_row(mop[i].imx.mtx, &mop[i].imx, mop[i].infp)) { |
| 710 |
|
if (cur_row > in_nrows) /* unknown #input rows? */ |
| 711 |
|
break; |
| 712 |
< |
fprintf(stderr, "%s: read error at row %d\n", |
| 712 |
> |
fprintf(stderr, "%s: parent read error at row %d\n", |
| 713 |
|
mop[i].inspec, cur_row); |
| 714 |
|
return(0); |
| 715 |
|
} |
| 716 |
|
if (i < nmats) |
| 717 |
|
break; |
| 718 |
|
for (i = 0; i < nmats; i++) |
| 719 |
< |
if (!rmx_write_data(mop[i].imx.mtx, mop[i].imx.ncomp, |
| 720 |
< |
mop[i].imx.ncols, DTdouble, ofp)) |
| 721 |
< |
return(0); |
| 682 |
< |
if (fflush(ofp) == EOF) |
| 683 |
< |
return(0); |
| 719 |
> |
if (writebuf(wfd, mop[i].imx.mtx, rmx_array_size(&mop[i].imx)) |
| 720 |
> |
!= rmx_array_size(&mop[i].imx)) |
| 721 |
> |
return(0); |
| 722 |
|
} |
| 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); |
| 723 |
|
i = close_processes(cproc, nchildren); |
| 724 |
< |
free(cproc); cproc = NULL; |
| 724 |
> |
free(cproc); cproc = NULL; nchildren = 0; |
| 725 |
|
if (i < 0) { |
| 726 |
< |
fputs("Warning: missing child in parent_loop()\n", stderr); |
| 726 |
> |
fputs("Warning: lost child in parent_loop()\n", stderr); |
| 727 |
|
return(1); |
| 728 |
|
} |
| 729 |
|
if (i > 0) { |
| 730 |
|
fprintf(stderr, "Child exited with status %d\n", i); |
| 731 |
|
return(0); |
| 732 |
|
} |
| 733 |
< |
return(1); |
| 733 |
> |
return(1); /* return success! */ |
| 734 |
|
memerror: |
| 735 |
|
fputs("Out of memory in parent_loop()\n", stderr); |
| 736 |
|
exit(1); |
| 737 |
|
} |
| 738 |
|
|
| 739 |
|
static int |
| 740 |
< |
combine_input() |
| 740 |
> |
combine_input(void) |
| 741 |
|
{ |
| 742 |
|
const int row0 = (inchild >= 0)*inchild; |
| 743 |
< |
const int rstep = nchildren + !nchildren; |
| 743 |
> |
const int rstep = nchildren ? nchildren-1 : 1; |
| 744 |
|
ROPMAT *res = &mop[nmats]; |
| 745 |
|
int set_r, set_c; |
| 746 |
|
RMATRIX *tmp = NULL; |
| 764 |
|
set_c = varlookup("c") != NULL && !vardefined("c"); |
| 765 |
|
} else /* save a little time */ |
| 766 |
|
set_r = set_c = 0; |
| 734 |
– |
|
| 767 |
|
/* read/process row-by-row */ |
| 768 |
|
for (cur_row = row0; (in_nrows <= 0) | (cur_row < in_nrows); cur_row += rstep) { |
| 769 |
|
RMATRIX *mres = NULL; |
| 816 |
|
return(0); |
| 817 |
|
} |
| 818 |
|
rmx_free(mres); mres = NULL; |
| 787 |
– |
if (inchild >= 0) { /* children share stdout */ |
| 788 |
– |
i = getc(stdin); /* signals it's our turn */ |
| 789 |
– |
if (i != EOF) ungetc(i, stdin); |
| 790 |
– |
} |
| 819 |
|
if (!rmx_write_data(res->rmp->mtx, res->rmp->ncomp, |
| 820 |
|
res->rmp->ncols, res->rmp->dtype, stdout)) |
| 821 |
|
return(0); |
