collapse.c

MPI COLLAPSE 用于MPI 研究的初步学习

/****************************************************************
 * collapse.c - Comp5426 - Ass1 
 * <Ke FAN - kfan3839 - 17/04/08>
 * Copyright of Ke FAN
 ****************************************************************/


#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>

int main(int argc, char *argv[])
{
	int number[10000];
	int buffer[10];
	int i = 0;
	int n = 0;
	char c;
	int chunk_size = 100;																/*the length of the distributed array*/
	int myid, numprocs;
	int curvalue = 0;																	/*current position*/
	int end;
	int count = 0;
    int j = 0;
	while(j < 10000)
	{
		number[j] = -1;
		j++;
	}

/* Initiate the environment */
	MPI_Init (&argc, &argv);
	MPI_Comm_size (MPI_COMM_WORLD, &numprocs);
	MPI_Comm_rank (MPI_COMM_WORLD, &myid);

	if(myid == 0)
	{
		
		int* 			lSum = (int *)calloc(numprocs , sizeof(int));					/*the caculated value of each processor*/
		MPI_Request* 	reqs = (MPI_Request*)calloc(numprocs , sizeof(MPI_Request));	/*array of requests */
		int* 			indices = (int *)malloc((numprocs)*sizeof(int));				/*array of index that have completed*/
		MPI_Status * 	stat=(MPI_Status*)malloc(sizeof(MPI_Status)*(numprocs ));		/*array of status objects for operations that completed */
		int 			numRunningProcs;												/*the number of processors that are running*/
		int				result;															/*the result*/

/* the array that indicating the subprocessors that is over */
		while(n < 10)
		{
			buffer[n] = -1;
			n++;
		}		

/* input process*/
		printf("Please input the number: \n");
		while ((c = getc(stdin)) != '\n') 
		{
			if (c < '0' || c > '9') 
			{
				printf("Invalid input.The process will be terminated. \n");
				exit(0);
			}
			number[i] = c - '0';
			i++;
		} 
		printf("Input is finished. i is %d \n" , i);
		number[i] = -1; 
		end = i;

		count = numprocs - 1;
       																/*how many workers are idle*/
		for( n = 1;n <= count;n++)
		{
			indices[n-1] = n;
		}																/*the index of the idle worker*/

		result = 0;				
		numRunningProcs = 0;	
	

        while ((numRunningProcs || number[curvalue + chunk_size] != -1 )&& (10 < (end - curvalue))) 	/*While there is still work to do*/
		{
			

			for(n = 1; n <= count && number[curvalue + chunk_size] != EOF; n++)
			{
				MPI_Send(&number[curvalue], chunk_size, MPI_INT, indices[n-1], 1, MPI_COMM_WORLD);
				printf("Sent a value to %d \n", indices[n-1] );
				numRunningProcs++;
				curvalue += chunk_size;
				/*after sending the work, we open an unblocking socket and wait for return*/
				MPI_Irecv((lSum + indices[n-1]), 1, MPI_INT, indices[n-1] , 2, MPI_COMM_WORLD, reqs + indices[n-1]);
				
			}
			
			MPI_Testsome(numprocs , reqs, &count, indices, stat);

			

			numRunningProcs = numRunningProcs - count;
			/* indices is an array with count entries containing the id's of
				the processes that have finished. We obtain the values that
				the finished processors have returned , and add them to the end of the array */
			
            n = 1;
            while(n <= count)
			{
				number[end] = *(lSum+indices[n-1]);
				end++;
				number[end] = -1;
				n++;
			}	
		}
		/* current position doesn't meet the end*/

		do 
		{
			for (n = curvalue+1; n < end; n++)
			{
				number[n] += number[n-1];
				while (number[n] >= 10)
					{
                                 number[n] = number[n]/10+number[n]%10;
                    }
			}
		}while(curvalue > end);
        	
		for(n = 1; n < numprocs; n++)	/* send the termination signal */
              MPI_Send(buffer, chunk_size, MPI_INT, n, 1, MPI_COMM_WORLD);
		printf("The final result is %d\n", number[end-1]);
		free(lSum);
		free(reqs);
		free(indices);
		free(stat);
		MPI_Finalize();
		return 0;
	}
	else 
	{
		MPI_Status stat;
		MPI_Request req;
		int n;

		printf("process %d is running,\n", myid);
		while (1)
		{
			
			int buffer[chunk_size];

			MPI_Recv(buffer, chunk_size, MPI_INT, 0, 1, MPI_COMM_WORLD, &stat);

			while(buffer[0] == -1)
			{	
				printf("process %d has been terminated! \n", myid);
				break;
			}
			
			n = 1;
			while(n < chunk_size)
			{
				buffer[n] += buffer[n-1];
				while (buffer[n] >= 10)
				{
					buffer[n] = buffer[n]/10+buffer[n]%10;
				}
				n++;
			}

			printf("process %d has caculated the number, the number is %d \n", myid , buffer[n-1]);

			MPI_Isend(&buffer[n-1], 1, MPI_INT, 0, 2, MPI_COMM_WORLD, &req);	
			printf("process %d has sent the number, the number is %d \n", myid , buffer[n-1]);
			MPI_Wait(&req, &stat);
		}
	}
	MPI_Finalize();
	return 0;
}