/*

==========================================================================

jpar.c - a simple parallel computing master program

Copyright (c) 1999 by Jeffrey S. Rosenthal (jeff@math.toronto.edu).

Available from  http://probability.ca/jeff/comp/

Licensed for general copying, distribution and modification according to
the GNU General Public License (http://www.gnu.org/copyleft/gpl.html).

----------------------------------------------------

Compile with "cc jpar.c -o jpar".

----------------------------------------------------

USAGE:  jpar [filename [seed]]

SYNOPSIS:  

jpar will read each line of <filename> (or jpar.data, if "-" or no
filename is specified), and then execute it using the "system" command.
An optional third argument <seed> gives an integer pseudo-random number
seed (or, if <seed> is 0 or non-numeric, the seed value is taken from
the current time microsecond hand) to be appended to the first command,
with incremented-by-1 seeds appended to each succeeding command.

Each command is expected, upon completion, to return two floating-point
numbers.  The first number is its estimate of the unknown quantity,
the second number is the relative weight that should be assigned to
this estimate (e.g. the number of simulations it performed).  Once all
commands have completed, jpar computes and outputs a weighted average
of their estimates.

An example of an appropriate file jpar.data might be:

    nice /path/unifavr 60
    rsh -n machine1 nice /path/unifavr 60
    rsh -n machine2 nice /path/unifavr 60
    rsh -n machine3 nice /path/unifavr 60
    rsh -n machine4 nice /path/unifavr 60

which could then be run with the command

    jpar jpar.data 0

(or simply with the command "jpar" if no random number seeds are required).

==========================================================================

*/


#define MAXNUM 100  /* Maximum number of commands to run; may be modified. */

#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/types.h>


main(int argc, char *argv[])
{
    FILE *fp[MAXNUM];
    int filenum[MAXNUM];
    struct timeval waittime;
    fd_set rfds;
    char c;
    char buffer[80];
    int numread, selectval, tmpval, i;
    char command[MAXNUM][160], tmpstring[20];
    int resultplace[MAXNUM];
    char result[MAXNUM][160];
    int finished[MAXNUM];
    long presenttime(), startingtime, endtime[MAXNUM];
    float newval, newweight;
    double valtot, weighttot;
    FILE *cfp;
    char *filepointer;
    int N, TOGO, ADDSEEDS, seed;

    /* Determine data file name. */
    if ( (argc > 1) && (strcmp(argv[1],"-")) )
	filepointer = argv[1];
    else
	filepointer = "jpar.data";

    /* Determine whether or not to add seed values. */
    ADDSEEDS = ( argc > 2 );
    if ( (ADDSEEDS) && ( (seed=atoi(argv[2])) == 0 ) )
	seed = seedval();

    /* Open data file. */
    if ( (cfp = fopen(filepointer, "r")) == NULL) {
	fprintf(stderr, "Unable to read file %s.\n", filepointer);
	exit(1);
    }

    /* Read in commands. */
    N = 0;
    while (eatwhite(cfp)==0) {
	fgets(command[N], 160, cfp);
	N++;
    }
    fclose(cfp);

    /* Remove trailing new-lines. */
    for (i=0; i<N; i++) {
	if( command[i][strlen(command[i])-1] == '\n' )
	    command[i][strlen(command[i])-1] = '\0';
    }

    /* Initiate the commands. */
    printf("\nInitiating the commands ...\n");
    startingtime = presenttime();
    for (i=0; i<N; i++) {
	if (ADDSEEDS) {
	    sprintf(tmpstring, " %d", seed+i);
	    strcat(command[i], tmpstring);
	}
	if ((fp[i] = popen(command[i], "r")) == NULL) {
	    fprintf(stderr, "Error initiating command %d.\n", i+1);
	    exit(1);
	} else {
	    filenum[i] = fileno(fp[i]);
	    resultplace[i] = finished[i] = 0;
	    printf("%d. %s\n", i+1, command[i]);
	}
    }
    printf("  ... done.\n\n");

    /* Loop, looking for returned characters. */
    TOGO = N;
    while (TOGO > 0) {
	/* Poll each of the N different streams. */
	for (i=0; i<N; i++) {
	    waittime.tv_sec = (long) 0;
	    waittime.tv_usec = (long) 10000;
	    FD_ZERO(&rfds);
	    FD_SET(filenum[i], &rfds);

	    /* Check for characters from i'th command. */
	    while ( (finished[i]==0) &&
			(select(32, &rfds, NULL, NULL, &waittime)) ) {
		if ( (read(filenum[i], buffer, 1)) > 0 ) {

		    /* Read in a character of result. */
		    result[i][resultplace[i]] = buffer[0];
		    resultplace[i]++;

		} else {
		    /* This command has finished! */

		    /* Remove trailing new-line. */
		    if (result[i][resultplace[i]-1] == '\n')
			result[i][resultplace[i]-1] = '\0';
		    else
			result[i][resultplace[i]] = '\0';

		    /* Record finishing of command. */
		    finished[i] = 1;
		    TOGO--;
		    endtime[i] = presenttime();
		    printf("Command %d finished in time %.5f secs.\n", i+1,
			(endtime[i] - startingtime) / 1000000.0 );
		    printf("    Result:  %s\n", result[i]);

		}
		waittime.tv_sec = (long) 0;
		waittime.tv_usec = (long) 10000;
		FD_ZERO(&rfds);
		FD_SET(filenum[i], &rfds);
	    }
	}

    }

    /* Close all the streams, interpret results, and collate. */
    printf("\n--------------------------------------------\n\n");
    printf("SUMMARY OF RESULTS:\n\n");
    weighttot = valtot = 0.0;
    for (i=0; i<N; i++) {
	fclose(fp[i]);
	newval = newweight = 0.0; /* In case of sscanf problems. */
	sscanf(result[i], "%f %f", &newval, &newweight);
	weighttot = weighttot + newweight;
	valtot = valtot + newval*newweight;
	printf("%d. %s\n    (%f sec)    %f    %f \n", i+1, command[i], 
		(endtime[i] - startingtime) / 1000000.0, newval, newweight);
    }

    /* Print out results. */
    if (weighttot == 0.0)
	printf("Oops, weighttot = 0.0!!\n\n");
    else
	printf("\n\tvaltot = %f;   weighttot = %f;\n\n\t\tweighted average = %.8f\n\n",
	    valtot, weighttot, valtot/weighttot);
    printf("\nALL DONE!\n\n");
    printf("Total running time = %f secs.\n\n",
			(presenttime() - startingtime) / 1000000.0 );
    return(0);

}


long presenttime()  /* Return present time in microseconds.  */
{
    struct timeval tmp;
    gettimeofday (&tmp, (struct timezone *)NULL);
    return (tmp.tv_usec + 1000000 * tmp.tv_sec);
}


int eatwhite(FILE *thefp)  /* Returns 1 on EOF, otherwise 0. */
{
    int ch;
    while(isspace(ch=fgetc(thefp)))
	;
    if (ch==EOF) {
	return(1);
    } else {
	ungetc(ch, thefp);
	return(0);
    }
}


/* Provide seed based on current time microsecond hand.  */
int seedval()  
{
    struct timeval tmptv;

    gettimeofday (&tmptv, (struct timezone *)NULL);
    return( (int) (tmptv.tv_usec - 1000000 * (int)
        ( ( (double) tmptv.tv_usec ) / ( (double) 1000000.0 ) ) ) );
}