quick.c.txt

Quick Sorting using MPI libraries

#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <math.h>
#include "mpi.h"

int myRank;
int mySize;

const int SingleArrayLength=10000000;
const int MaxArrayElement=1000000;
const int MinArrayElement=100;
int* myArray;

const int WorkerNumber=7;

const int SampleOffset=SingleArrayLength/WorkerNumber;
const int StartSampleOffset=WorkerNumber/2-1;


int* sampleArray;
int**mergeBuf;

MPI_Request* sampleRequests;

const int SAMPLE=100;
const int PIVOT=101;
const int MaxIntegerPrintLength=10;

void myExit()
{
	//printf("\n\nrank %d finishes\n", myRank);
}


void printArray(int* array, int length, char* comment="Array print out:");

void printArray(int* array, int length, char* comment)
{
	char* buf;
	char temp[MaxIntegerPrintLength];
	int commentLength=0;
	commentLength=strlen(comment)+5;
	buf=new char[length*MaxIntegerPrintLength+commentLength];
	sprintf(buf, "rank[%d] %s:*****", myRank, comment);
	for (int i=0; i<length; i++)
	{
		sprintf(temp, "%d,", array[i]);
		strcat(buf, temp);
	}
	strcat(buf, "\n");
	printf(buf);
	delete []buf;
}


void initialize()
{
	int i;
	atexit(myExit);
	srand(myRank*time(0));
	if (myRank==0)
	{
		sampleArray=new int[WorkerNumber*WorkerNumber];
		sampleRequests=new MPI_Request[WorkerNumber];
		myArray=new int[SingleArrayLength*WorkerNumber];
		mergeBuf=new int*[WorkerNumber];
		for (i=0; i<WorkerNumber; i++)
		{
			mergeBuf[i]=new int[SingleArrayLength];
		}
	}
	else
	{
		sampleArray=new int[WorkerNumber];
		sampleRequests=new MPI_Request[1];
		//srand(time(0));
		myArray=new int[SingleArrayLength];
		for (i=0; i<SingleArrayLength; i++)
		{
			myArray[i]=rand()%MaxArrayElement+MinArrayElement;
		}
	}

	

}


int intComp(const void* first, const void* second)
{
	return *(int*)first - *(int*)second;
}

void zeroPhase()
{
	int i;
	double start, end;
	if (myRank==0)
	{
		for (i=0; i<WorkerNumber; i++)
		{
			sampleRequests[i]=MPI_REQUEST_NULL;
			MPI_Irecv(myArray+i*SingleArrayLength, SingleArrayLength, MPI_INT, i+1, 0, MPI_COMM_WORLD, sampleRequests+i);
		}
		MPI_Waitall(WorkerNumber, sampleRequests, MPI_STATUSES_IGNORE);
		start=MPI_Wtime();
		qsort(myArray, WorkerNumber*SingleArrayLength, sizeof(int), intComp);
		end=MPI_Wtime();
		printf("single machine sorting array of length %d takes %f\n",SingleArrayLength*WorkerNumber, end-start);
				
	}
	else
	{
		MPI_Send(myArray, SingleArrayLength, MPI_INT, 0, 0, MPI_COMM_WORLD);
	}
}		



void firstPhase()
{
	if (myRank!=0)
	{
		qsort(myArray, SingleArrayLength, sizeof(int), intComp);
		//printArray(myArray, SingleArrayLength, "worker data array print out");
		//retrieve samples
		for (int i=0; i<WorkerNumber; i++)
		{
			sampleArray[i]=myArray[i*SampleOffset];
		}		
	}
}


void secondPhase()
{
	int i;
	if (myRank==0)
	{
		for (i=0; i<WorkerNumber; i++)
		{
			MPI_Irecv(sampleArray+i*WorkerNumber, WorkerNumber, MPI_INT, i+1, SAMPLE, MPI_COMM_WORLD, sampleRequests+i);
			//MPI_Recv(sampleArray+i*WorkerNumber, WorkerNumber, MPI_INT, i+1, SAMPLE, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
		}
		MPI_Waitall(WorkerNumber, sampleRequests, MPI_STATUSES_IGNORE);
		qsort(sampleArray, WorkerNumber*WorkerNumber, sizeof(int), intComp);
		//printArray(myArray, SingleArrayLength, "worker's data array\n");
		/*
		for (i=0; i<WorkerNumber*WorkerNumber; i++)
		{
			printf("sample[%d]=%d\n", i, sampleArray[i]);
		}
		*/
		//printArray(sampleArray, WorkerNumber*WorkerNumber);
		

		for (i=1; i<WorkerNumber; i++)
		{
			sampleArray[i-1]=sampleArray[WorkerNumber*i+StartSampleOffset];			
		}
		MPI_Bcast(sampleArray, WorkerNumber-1, MPI_INT, 0, MPI_COMM_WORLD);
		//printArray(sampleArray, WorkerNumber-1, "this is the sampel data broadcasted");

	}
	else
	{
		MPI_Ssend(sampleArray, WorkerNumber, MPI_INT, 0, SAMPLE, MPI_COMM_WORLD);
		MPI_Bcast(sampleArray, WorkerNumber-1, MPI_INT, 0, MPI_COMM_WORLD);
	}
	/*
	for (i=0; i<WorkerNumber-1; i++)
	{
		printf("rank[%d][%d]=%d\n", myRank, i, sampleArray[i]);
	}
	*/
	
}

//it returns the smallest index of which the number is bigger than or equal to the key
int binarySearch(int key, int* array, int length)
{
	int front=0, end=length-1;
	if (key>array[end])
	{
		return length;
	}
	if (key<array[front])
	{
		return 0;
	}

	int pos=(front+end+1)/2;;
	while (front<=end)
	{		
		if (key>array[pos])
		{
			front=pos+1;
		}
		else
		{
			if (key<array[pos])
			{
				end=pos-1;
			}
			else
			{
				break;
			}
		}
		pos=(front+end+1)/2;
	}
	
	return pos;
}



void thirdPhase()
{
	int i, flag;
	//do binary search
	if (myRank!=0)
	{
		for (i=0; i<WorkerNumber-1; i++)
		{
			//printf("rank[%d]key=%d\n", myRank, sampleArray[i]);
			sampleArray[i]=binarySearch(sampleArray[i], myArray, SingleArrayLength);			
			//printf("rank[%d][%d]=%d and the data myArray[%d]=%d\n", myRank, i, sampleArray[i],sampleArray[i], myArray[sampleArray[i]] );//for testing
			//printf("before %d and after %d \n", myArray[sampleArray[i]-1], myArray[sampleArray[i]+1]); 
		}
		sampleArray[WorkerNumber-1]=SingleArrayLength;
		MPI_Ssend(sampleArray, WorkerNumber, MPI_INT, 0, PIVOT, MPI_COMM_WORLD);
		//printArray(sampleArray, WorkerNumber, "worker sampleArray print out in third phase");
	}
	else
	{
		for (i=0; i<WorkerNumber; i++)
		{
			sampleRequests[i]=MPI_REQUEST_NULL;
			MPI_Irecv(sampleArray+i*WorkerNumber, WorkerNumber, MPI_INT, i+1, PIVOT, MPI_COMM_WORLD, sampleRequests+i);
			//MPI_Recv(sampleArray+i*WorkerNumber, WorkerNumber, MPI_INT, i+1, PIVOT, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
			//sampleArray[(i+1)*WorkerNumber]=WorkerNumber;
		}
		//printArray(sampleArray, WorkerNumber*WorkerNumber, "master sampleArray print out in third phase");
		MPI_Waitall(WorkerNumber, sampleRequests, MPI_STATUSES_IGNORE);
		/*
		for (i=0; i<WorkerNumber*WorkerNumber; i++)
		{
			printf("sample[%d]=%d\n", i, sampleArray[i]);
		}
		*/
		
		
	}
}

void doMerge(int** mergeBuf, int* lengthArray,int& currentPos)
{
	int indexArray[WorkerNumber];
	int i, candidate, candidateIndex;
	bool beFirst=true, allOver=false;
	for (i=0; i<WorkerNumber; i++)
	{
		indexArray[i]=0;
	}
	do
	{		
		beFirst=true;
		allOver=true;
		for (i=0; i<WorkerNumber; i++)
		{
			if (indexArray[i]<lengthArray[i])
			{
				allOver=false;
				if (beFirst)
				{
					beFirst=false;
					candidate=mergeBuf[i][indexArray[i]];
					candidateIndex=i;
				}
				else
				{
					if (candidate>mergeBuf[i][indexArray[i]])
					{
						candidate=mergeBuf[i][indexArray[i]];
						candidateIndex=i;
					}
				}
			}
		}
		if (allOver)
		{
			break;
		}
		myArray[currentPos]=candidate;
		currentPos++;
		indexArray[candidateIndex]++;
	}
	while (true);
}
		
						
			
		


void fourthPhase()
{
	int i, j, flag;
	int* sizePtr;
	int* dataPtr;
	int currentPos=0;
	int previous, current;
	int length;
	//MPI_Request* tempRequests;
	int lengthArray[WorkerNumber];
	if (myRank==0)
	{
		//tempRequests=new MPI_Request[WorkerNumber*WorkerNumber];
		//printArray(sampleArray, WorkerNumber*WorkerNumber, "before 4th phase, let' see sample Array\n");
		for (i=0; i<WorkerNumber; i++)//the index of  worker node
		{
			for (j=0; j<WorkerNumber; j++)//the index within worker node index
			{
				sizePtr=sampleArray+j*WorkerNumber+i;
				if (i==0)
				{
					previous=0;	
					current=*sizePtr;
				}
				else
				{
					if (i==WorkerNumber-1)
					{
						current=SingleArrayLength;
					}
					else
					{
						current=*sizePtr;
					}
					previous=*(sampleArray+j*WorkerNumber+i-1);
				}
				//printf("\n current=%d, previous=%d\n", current, previous);
				lengthArray[j]=current - previous;
				//currentPos+=length;
				//dataPtr=myArray+currentPos;
				sampleRequests[j]=MPI_REQUEST_NULL;
				//printf("\nmaster begins\n");
				if (lengthArray[j]>0)
				{
					//MPI_Irecv(dataPtr, length, MPI_INT, j+1, j*10+i, MPI_COMM_WORLD, tempRequests+j*WorkerNumber+i);
					//printf("\nmaster begin to recv data from rank %d of length %d\n", j+1, lengthArray[j]);
					MPI_Irecv(mergeBuf[j], lengthArray[j], MPI_INT, j+1, j*10+i, MPI_COMM_WORLD, sampleRequests+j);
					//MPI_Recv(mergeBuf[j], lengthArray[j], MPI_INT, j+1, j*10+i, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
				}

				//printf("\nmaster after prints\n");

			}
			MPI_Waitall(WorkerNumber, sampleRequests, MPI_STATUSES_IGNORE);
			doMerge(mergeBuf, lengthArray, currentPos);
			/*
			printf("\nmaster after tests of %d\n", i+1 );
			for (int k=0; k<WorkerNumber; k++)
			{	
				//printf("\nmaster going to print %d\n", lengthArray[k]);			
				if (lengthArray[k]>0)
				{
					printArray(mergeBuf[k], lengthArray[k], "Master receive segment");
				}
			}
			*/

		}
		//MPI_Testall(WorkerNumber*WorkerNumber, tempRequests, &flag, MPI_STATUSES_IGNORE);
	}
	else
	{
		for (i=0; i<WorkerNumber; i++)
		{
			sizePtr=sampleArray+i;
			if (i==0)
			{
				previous=0;
				current=*sizePtr;
			}
			else
			{
				if (i==WorkerNumber-1)
				{
					current=SingleArrayLength;
				}
				else
				{
					current=*sizePtr;
				}
				previous=*(sampleArray+i-1);

			}
			dataPtr=myArray+previous;
			length=current-previous;
			currentPos+=length;
			if (length>0)
			{
				//printArray(dataPtr, length, "going to send segment");
				MPI_Send(dataPtr, length, MPI_INT, 0, (myRank-1)*10+i, MPI_COMM_WORLD);
			}
		}
	}
}



void testArray()
{
	int previous=myArray[0], current=myArray[0];
	for (int i=1; i<WorkerNumber*SingleArrayLength; i++)
	{		
		
		current=myArray[i];
		if (current<previous)
		{
			printf("sorting error at %d with %d > %d\n", i, previous, current);
			//exit(4);
		}
		previous=current;
		//printf("rank[%d][%d]=%d\n", myRank, i, myArray[i]);
	}
}

		





int main(int argc, char** argv)
{
	double start, end;
	MPI_Init(&argc, &argv);                 
   	MPI_Comm_rank(MPI_COMM_WORLD, &myRank);
    	MPI_Comm_size(MPI_COMM_WORLD, &mySize);  
	initialize();
	
	zeroPhase();
	if (myRank==0)
	{
		start=MPI_Wtime();
	}


	firstPhase();
	//printf("\nfirst phase ends\n");

	secondPhase();	
	//printf("\nsecond phase ends\n");

	thirdPhase();
	//printf("\nthird phase ends\n");

	fourthPhase();

	//printf("\nfourth phase ends\n");

	
	if (myRank==0)
	{
		end=MPI_Wtime();
		printf("distributing system sorting takes %f\n", end-start);
	}
	

	


	return 0;
}