authParallelAndDistributedSystemsCourseExercise2/NonBlocking/spaceOptimized/knnMPINonBlocking.c

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

#include "knnMPINonBlockingDeclarations.h"
#include "mpi.h"

/*		Structs		*/
struct timeval startwtime, endwtime
			, calcStartWTime, calcEndWTime;

/*		Global variables		*/
int numberOfPoints = 0, numberOfDimensions = 0, numberOfNeighbors = 0;
char *pointsFilename = NULL, *testFilename = NULL;
double seq_time = 0;

/*		Main		*/
int main(int argc, char **argv){
	int numberOfTasks = 0, taskId = 0, chunksize = 0, myOffset = 0, destination = 0, source = 0;
	double **myPoints, **inBuffer, **outBuffer, calcTimeSum = 0;
	neighbor **sortedNeighborsArray;
	MPI_Status statuses[2];
	MPI_Request requests[2];

	getArguments(argc, argv);

	//Initializes mpi and necessary variables
	MPI_Init(&argc, &argv);
	MPI_Comm_size(MPI_COMM_WORLD, &numberOfTasks);
	if (numberOfPoints % numberOfTasks != 0){
		printf("Quitting. Number of points must be divisible by number of MPI tasks.\n");
		abExit(0);
	}
	MPI_Comm_rank(MPI_COMM_WORLD, &taskId);

	chunksize = numberOfPoints / numberOfTasks; //number of points each task has
	myOffset = taskId * chunksize; //this task's offset from (complete) point's array start
	destination = taskId + 1; //this task's destination task id for sending messages
	if (destination >= numberOfTasks){
		destination = 0;
	}
	source= taskId - 1; //this task's source task id for receiving messages
	if (source < 0){
		source = numberOfTasks - 1;
	}

	init(&myPoints, &inBuffer, &outBuffer, &sortedNeighborsArray
		,chunksize, myOffset * numberOfDimensions);

	MPI_Barrier(MPI_COMM_WORLD);
	if (taskId == MASTER){
		gettimeofday(&startwtime, NULL);
	}

	for (int chunk=-1; chunk<numberOfTasks-1; ++chunk){
		int lastLoop = (chunk == numberOfTasks-1-1) ? 1 : 0;

		if (!lastLoop){ //last loop doesn't send nor receive
			//sends-receives points array blocks
			MPI_Isend((chunk == -1)		//if this is the first loop
				? &(myPoints[0][0])		//send this task's points
				: &(outBuffer[0][0])	//else send the points received from previous loop
				, chunksize * numberOfDimensions, MPI_DOUBLE, destination, taskId, MPI_COMM_WORLD
				, &requests[BLOCK_SENT]);
			MPI_Irecv(&(inBuffer[0][0]), chunksize * numberOfDimensions, MPI_DOUBLE, source, source
				, MPI_COMM_WORLD, &requests[BLOCK_RECEIVED]);	
		}

		//offset of the task that sent new points block from (complete) point's array start
		//used for storing the correct neighbor id's in neighbors array
		int offset = source - chunk;
		if (offset < 0){
			offset += numberOfTasks;
		}
		offset *= chunksize;

		gettimeofday(&calcStartWTime, NULL);
		if (chunk == -1){
			//calculates distances between task's own points
			calculateDistances(&myPoints, &myPoints, &sortedNeighborsArray
				, chunksize, myOffset, taskId * chunksize);
		} else {
			//calculates distances between task's points and new points received
			calculateDistances(&myPoints, &outBuffer, &sortedNeighborsArray
				, chunksize, myOffset, offset);
		}
		gettimeofday(&calcEndWTime, NULL);
		calcTimeSum += (double)((calcEndWTime.tv_usec - calcStartWTime.tv_usec)/1.0e6
			+ calcEndWTime.tv_sec - calcStartWTime.tv_sec);

		if (!lastLoop){ //last loop doesn't send nor receive
			//continues with next loop (and calculations) only after sending and receiving is done
			MPI_Waitall(2, requests, statuses);
		}

		swapPointers(&inBuffer, &outBuffer);
	}

	MPI_Barrier(MPI_COMM_WORLD);
	if (taskId == MASTER){
		gettimeofday(&endwtime, NULL);
		seq_time = (double)((endwtime.tv_usec - startwtime.tv_usec)/1.0e6
			+ endwtime.tv_sec - startwtime.tv_sec);

		printf("Wall clock time = %f\n\n", seq_time);
	}

	if (test(&sortedNeighborsArray, chunksize, myOffset, testFilename)){
		printf("Validity check failed for task %d!\n", taskId);
	} else {
		printf("Validity check success for task %d!\n", taskId);
	}

	if (taskId == MASTER){
		printf("Number of points = %d\n", numberOfPoints);
		printf("Number of dimensions = %d\n", numberOfDimensions);
		printf("Number of neighbors = %d\n", numberOfNeighbors);
		printf("Number of tasks = %d\n", numberOfTasks);
	}
	for(int i=0; i<numberOfTasks; ++i) {
		MPI_Barrier(MPI_COMM_WORLD);
		if (i == taskId) {
			printf("Task %d calculations time sum = %f\n", taskId, calcTimeSum);
			/*
					Uncomment to print the results
			*/
			/*for (int x=0; x<chunksize; ++x){
				for (int y=0; y<numberOfNeighbors; ++y){
					printf("%d to %d = %f\n", x + taskId * chunksize
						, sortedNeighborsArray[x][y].neighborId, sortedNeighborsArray[x][y].distance);
				}
				printf("\n");
			}*/
		}
	}

	cleanUp(&myPoints, &inBuffer, &outBuffer, &sortedNeighborsArray, chunksize);
	MPI_Finalize();
}