/*
Program: Parallel Sort using a hybrid mergesort
Author: Chris Harper (snakebytestudios.com)
Mod Date: 4/21/08

TODO:
-phase 1
	-merge results back into one array
	-verify final array length and sortedness
	-fix recieve overhead measure
-phase 2
	-get number of processors in each machine
		-give multiprocessor machines more tasks
	-optimize serial sorts for large data sets
		(does quicksort duplicate data for each recursive call?)
	-implement recursive slave execution
		(must be memory efficient for large data sets)

*/

#include <limits.h>
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include "pvm3.h"
#include "stopwatch.c"

//settings
#define LOG_FILE "/home/clh324/mergesort/log.csv"
//#define LOG_FILE "/root/Desktop/log.csv"
#define DEF_TEST_SIZE 25
#define SLAVENAME "slave1"

struct NodeData {
	int Task_ID;
	int part_num;
	char host[255];
	double exec_time;
	long data_length;
};

int short_comp (const void *, const void *);

int main(int argc, char *argv[]) {
	printf("PARALLEL SORT\n");
	printf("\tUsage: executable [# items [# tasks]]\n");
	printf("\t# items - length of array to sort or -1 for default\n");
	printf("\t# tasks - force this many tasks, -1 for default, or 0 for serial\n");
	int k, who;
	stopwatch timer_program, timer_parallel, timer_overhead_send, timer_overhead_rcv, timer_spawn;

	/* command args debug
	for (int i = 0; i < argc; i++) {
		printf("[%d]: %s\n", i, argv[i]);
	}
	*/
	
	//set up the random array data to use
	long test_size;
	if (argc > 1) {
		if (atol(argv[1]) > 0) {
			test_size = atol(argv[1]);
		} else {
			test_size = DEF_TEST_SIZE;
		}
	} else {
		test_size = DEF_TEST_SIZE;
	}
	printf("Alloc %ld items ... ", test_size);
	short *test_set = new short[test_size];
	printf("OK\n");
	printf("\tMemory Use: %.2f MB\n", (test_size * sizeof(short))/1048576.0);

	printf("Randomizing ... ");
	srand(time(NULL));
	for (long i = 0; i < test_size; i++) {
		test_set[i] = rand() % SHRT_MAX;
	}
	printf("OK\n");

	if (test_size <= 50) {
		for (int i = 0; i < test_size; i++) {
			printf("%d ", test_set[i]);
		}
		printf("\n");
	}

	//serial sort
	if ((argc > 2) && (atoi(argv[2]) == 0)) {
		printf("Executing Serial Sort ... ");
		stopwatch timer_serial;
		timer_serial.start();
		qsort(test_set, test_size, sizeof(short), short_comp);
		printf("OK\n");
		printf("\tElapsed Serial Sort Time: %.3f secs\n", timer_serial.stop());

		//write log
		//parts is -1 for serial run
		FILE *pFile;
		pFile = fopen(LOG_FILE,"a");
		if (pFile != NULL) {
			char master_host[255];
			gethostname(master_host, 255);
			fprintf(pFile, "Master,Data_Length,Data_Size,Parts,Total_Exec_Time\n");
			fprintf(pFile, "%s,%ld,%.2f,-1,%.3f\n", master_host, test_size, (test_size * sizeof(short))/1048576.0, timer_serial.read());
			fclose(pFile);
		}
		exit(0);
	}
	
	//start program stopwatch
	timer_program.start();

	 /* enroll in pvm */
	int mytid, nproc, nhost, narch;
	int tids[32];
	pvmhostinfo *hostp;
	mytid = pvm_mytid();
	if (mytid < 0) {
		printf("Fatal PVM error! Exiting.");
		exit(1);
	}
	pvm_config( &nhost, &narch, &hostp );
	printf("%d nodes available\n", nhost);
	if (argc > 2) {
		if ((atoi(argv[2]) > 0) && (atoi(argv[2]) < 32) && (atoi(argv[2]) < test_size)) {
			nproc = atoi(argv[2]);
		} else {
			nproc = nhost;
		}
	} else {
		//nproc = 4;
		nproc = nhost;
	}
	printf("%d tasks selected\n", nproc);

	/* decide on the number of parallel parts */
	int power2 = 0;
	int nparts;
	do {
		nparts = (int)pow(2, power2);
		power2++;
	} while (nparts <= nproc/2);
	printf("Using %d parallel parts\n", nparts);
	//exit(0);

	/* split array into parts */
	short *test_set_parts[nparts];
	long test_set_part_sizes[nparts];
	long remaining_length = test_size;
	long part_length;
	long current_start = 0;
	for (int i = 0; i < nparts; i++) {
		part_length = (int)floor(remaining_length / (nparts-i));
		remaining_length -= part_length;
		printf("\tpart: %ld, pos: %d, len: %ld, left: %ld\n", i, current_start, part_length, remaining_length);
		test_set_parts[i] = &test_set[current_start];
		test_set_part_sizes[i] = part_length;
		current_start += part_length;
	}

	//start parallel stopwatch
	timer_parallel.start();
	timer_spawn.start();

	 /* start up slave tasks */
	printf("Spawning %d worker tasks ... " , nparts);
	int numt;
	numt=pvm_spawn(SLAVENAME, (char**)0, 0, "", nparts, tids);
	if( numt < nparts ){
		printf("\n Trouble spawning slaves. Aborting. Error codes are:\n");
		for (int i = numt; i < nparts; i++) {
			printf("TID %d %d\n",i,tids[i]);
		}
		for(int i = 0 ; i<numt ; i++ ) {
			pvm_kill( tids[i] );
		}
		pvm_exit();
		exit(1);
	}
	printf("OK\n");

	timer_spawn.stop();
	timer_overhead_send.start();
	
	/* Send data to slave tasks */
	printf("Sending data to slave tasks ... ");
	for (int i = 0; i < nparts; i++){
		pvm_initsend(PvmDataDefault);
		pvm_pkint(&i, 1, 1);
		pvm_pklong(&test_set_part_sizes[i], 1, 1);
		pvm_pkshort(test_set_parts[i], test_set_part_sizes[i], 1);
		pvm_send(tids[i], 1);
	}
	printf("OK\n");

	timer_overhead_send.stop();

	//free up memory
	delete(test_set);

	/* Wait for results from slaves */
	NodeData Parts[nparts];
	/*
	int Task_ID;
	int part_num;
	char host[255];
	double exec_time;
	long data_length
	*/
	short *results;
	for (int i = 0; i < nparts; i++){
		pvm_recv( -1, 5 );
		if (timer_overhead_rcv.isRunning() == false) timer_overhead_rcv.start();
		pvm_upkint( &Parts[i].Task_ID, 1, 1 );
		pvm_upkstr(Parts[i].host);
		pvm_upkint( &Parts[i].part_num, 1, 1 );
		pvm_upkdouble( &Parts[i].exec_time, 1, 1 );
		pvm_upklong( &Parts[i].data_length, 1, 1 );
		results = new short[Parts[i].data_length];
		pvm_upkshort( results, Parts[i].data_length, 1 );
		test_set_parts[Parts[i].part_num] = results;
		
		printf("Task %d (%s), Part %d returned. Took %.3f secs.\n", Parts[i].Task_ID, Parts[i].host, Parts[i].part_num, Parts[i].exec_time);
		if (test_size <= 50) {
			for (int j = 0; j < Parts[i].data_length; j++) {
				printf("%d ", results[j]);
			}
			printf("\n");
		}
	}

	timer_overhead_rcv.stop();
	timer_parallel.stop();

	//end stopwatches
	printf("Elapsed Spawn Time: %.3f secs\n", timer_spawn.read());
	printf("Elapsed Tx Overhead Time: %.3f secs\n", timer_overhead_send.read());
	printf("Elapsed Rx Overhead Time: %.3f secs\n", timer_overhead_rcv.read());
	printf("Elapsed Total Comm Overhead Time: %.3f secs\n", timer_overhead_send.read() + timer_overhead_rcv.read());
	printf("Elapsed Parallel Time: %.3f secs\n", timer_parallel.read());
	printf("Elapsed Program Time: %.3f secs\n", timer_program.stop());
	
	/* Program Finished exit PVM before stopping */
	pvm_exit();

	FILE *pFile;
	pFile = fopen(LOG_FILE,"a");
	if (pFile != NULL) {
		char master_host[255];
		gethostname(master_host, 255);
		fprintf(pFile, "Master,Data_Length,Data_Size,Parts,Total_Exec_Time,Parallel_Time,TotalOverhead_Time,SendOverhead_Time,RcvOverhead_Time,Spawn_Time");
		for (int i = 0; i < nparts; i++) {
			fprintf(pFile, ",Part%d_HostName,Part%d_Exec_Time", i, i);
		}
		fprintf(pFile, "\n");
		fprintf(pFile, "%s,%ld,%.2f,%d,%.3f,%.3f,%.3f,%.3f,%.3f,%.3f", master_host, test_size, (test_size * sizeof(short))/1048576.0, nparts, timer_program.read(), timer_parallel.read(), timer_overhead_send.read() + timer_overhead_rcv.read(), timer_overhead_send.read(), timer_overhead_rcv.read(), timer_spawn.read());
		for (int i = 0; i < nparts; i++) {
			fprintf(pFile, ",%s,%.3f", Parts[i].host, Parts[i].exec_time);
		}
		fprintf(pFile, "\n");
		fclose(pFile);
	}
}

int short_comp (const void * a, const void * b) {
	return ( *(short*)a - *(short*)b );
}