K-means Parallel

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <math.h>
#include "mpi.h"
#define MASTER 0
/**
 * Estrutura de ponto
 */
typedef struct
	{
		double _x;
		double _y;
	} Point;
/**
 decique o centroid que esta mais perto de um determinado ponto
 */
int getParent(Point point,Point* centroids,int num_centroids)
{
	int daddy=0;
	double distance=0;
	double minDistance=distancePoints(point,centroids[0]);
	int dex;
	#pragma omp parallel for
	for(dex=1;dex<num_centroids;dex++)
	{
		distance=distancePoints(point,centroids[dex]);
		if(minDistance>=distance)
		{
			daddy=dex;
			minDistance=distance;
		}
	}
	return daddy;
}
/**

 calcula os novos centroids após receber o array de centroids de cada processo escravo
 */
void NewCentroids(Point* points,int* data,Point* centroids,int num_clusters,int num_points)
{
	Point* newCentroids=malloc(sizeof(Point)*num_clusters);
	int* population=malloc(sizeof(int)*num_clusters);
	int dex;

	for(dex=0;dex<num_clusters;dex++)
	{
		population[dex]=0;
		newCentroids[dex]._x=0;
		newCentroids[dex]._y=0;
	}
	for(dex=0;dex<num_points;dex++)
	{
		population[data[dex]]++;
		newCentroids[data[dex]]._x+=points[dex]._x;
		newCentroids[data[dex]]._y+=points[dex]._y;
	}
	for(dex=0;dex<num_clusters;dex++)
	{
		if(population[dex]!=0.0)
		{
			newCentroids[dex]._x/=population[dex];
			newCentroids[dex]._y/=population[dex];
		}
	}
	for(dex=0;dex<num_clusters;dex++)
	{
		centroids[dex]._x=newCentroids[dex]._x;
		centroids[dex]._y=newCentroids[dex]._y;
	}
}
/**
 checa se convergiu
 */
int Convergence(int *former_clusters,int *latter_clusters,int num_points)
{
	int dex;
	for(dex=0;dex<num_points;dex++)
		if(former_clusters[dex]!=latter_clusters[dex])
			return -1;
	return 0;
}
/**
calcula a distancia entre dois pontos
 */
double distancePoints(Point point1,Point point2)
{
	return (pow((point1._x-point2._x)*100,2)+pow((point1._y-point2._y)*100,2));
}
/**
 Função responsável por ler os pontos
 */
void readPoints(FILE* input,Point *points,int num_points)
{
	int dex;
	for(dex=0;dex<num_points;dex++)
	{
		fscanf(input,"%lf,%lf",&points[dex]._x,&points[dex]._y);
	}
}
/**
 inicializa os centroids randomicamente
 */
void initialize(Point* centroids,int num_clusters)
{
	int dex;
	srand(time(NULL));
	for(dex=0;dex<num_clusters;dex++)
	{
		centroids[dex]._x=((double)(rand()%1000))/1000;
		centroids[dex]._y=((double)(2*rand()%1000))/1000;
	}
}
/**
 Função responsável por ler o arquivo que contem o numero de pontos e o numero de clusteres
 */
void read(FILE *input,int* num_clusters,int* num_points)
{
	fscanf(input,"%d\n",num_clusters);
	printf("%d\n",*num_clusters);

	fscanf(input,"%d\n",num_points);
	printf("%d\n",*num_points);
}
/**
seta o array de clusteres com o valor -1
 */
int resetData(int *data,int num_points)
{
	int dex;
	for(dex=0;dex<num_points;dex++)
	{
		data[dex]=-1;
	}
}

/**
 main, divide entre processos escravos e mestre
 */
int main(int argc, char* argv[])
{
    int rank;
	int size;
    int num_clusters;
    int num_points;
	int dex;
	int job_size;
	int job_done=0;

	Point* centroids;
	Point* points;
	Point* received_points;
	int  * slave_clusters;
	int  * former_clusters;
	int  * latter_clusters;

	MPI_Init(&argc, &argv);

	MPI_Status status;

    	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    	MPI_Comm_size(MPI_COMM_WORLD, &size);

	//Estrutura MPI
	MPI_Datatype MPI_POINT;
	MPI_Datatype type=MPI_DOUBLE;
	int blocklen=2;
	MPI_Aint disp=0;
	MPI_Type_create_struct(1,&blocklen,&disp,&type,&MPI_POINT);
	MPI_Type_commit(&MPI_POINT);

/******** Processo Master******************************************************/

   	if(rank==MASTER)
  	{
		//leitura de arquivo
		FILE *input;
    		input=fopen(argv[1],"r");
		read(input,&num_clusters,&num_points);
    		points=(Point*)malloc(sizeof(Point)*num_points);
		readPoints(input,points,num_points);
		fclose(input);

		//Alocação de memoria para os clusters
		former_clusters=(int*)malloc(sizeof(int)*num_points);
		latter_clusters=(int*)malloc(sizeof(int)*num_points);
		job_size=num_points/(size-1);
		centroids=malloc(sizeof(Point)*num_clusters);

		//resetando a memoria dos clusteres ja usados
		initialize(centroids,num_clusters);
		resetData(former_clusters,num_points);
		resetData(latter_clusters,num_points);

		//envia data para os processos escravos
		for(dex=1;dex<size;dex++)
		{
			printf("Sending to [%d]\n",dex);
			MPI_Send(&job_size              ,1           , MPI_INT        ,dex,0,MPI_COMM_WORLD);
			MPI_Send(&num_clusters          ,1           , MPI_INT        ,dex,0,MPI_COMM_WORLD);
			MPI_Send(centroids              ,num_clusters, MPI_POINT      ,dex,0,MPI_COMM_WORLD);
			MPI_Send(points+(dex-1)*job_size,job_size    , MPI_POINT      ,dex,0,MPI_COMM_WORLD);
		}
    		printf("Sent!\n");

		MPI_Barrier(MPI_COMM_WORLD);

		//parte principal do processamento
		while(1)
		{
			MPI_Barrier(MPI_COMM_WORLD);

			printf("Master recebendo\n");
			for(dex=1;dex<size;dex++)
				MPI_Recv(latter_clusters+(job_size*(dex-1)),job_size,MPI_INT,dex,0,MPI_COMM_WORLD,&status);

			printf("Master recebeu\n");

			NewCentroids(points,latter_clusters,centroids,num_clusters,num_points);
			printf("Novos Centroids estão prontos!\n");
			if(Convergence(latter_clusters,former_clusters,num_points)==0)
			{
				printf("Convergido!\n");
				job_done=1;
			}
			else
			{
				printf("Não Convergido!\n");
				for(dex=0;dex<num_points;dex++)
					former_clusters[dex]=latter_clusters[dex];
			}

			//informa os slaves que eles não tem mais trabalho
			for(dex=1;dex<size;dex++)
				MPI_Send(&job_done,1, MPI_INT,dex,0,MPI_COMM_WORLD);

			MPI_Barrier(MPI_COMM_WORLD);
			if(job_done==1)
				break;

			//envia os centroids mais recentes
			for(dex=1;dex<size;dex++)
				MPI_Send(centroids,num_clusters, MPI_POINT,dex,0, MPI_COMM_WORLD);

			MPI_Barrier(MPI_COMM_WORLD);
		}

		//saida do arquivo
		FILE* output=fopen(argv[2],"w");
		fprintf(output,"%d\n",num_clusters);
		fprintf(output,"%d\n",num_points);
		for(dex=0;dex<num_clusters;dex++)
			fprintf(output,"%lf,%lf\n",centroids[dex]._x,centroids[dex]._y);
		for(dex=0;dex<num_points;dex++)
			fprintf(output,"%lf,%lf,%d\n",points[dex]._x,points[dex]._y,latter_clusters[dex]+1);
		fclose(output);
	}
/*************Procesos escravos trabalham aqui ************************/
	else
	{
		//recebendo data
		printf("Recebendo\n");
		MPI_Recv(&job_size    ,1           ,MPI_INT  ,MASTER,0,MPI_COMM_WORLD,&status);
		MPI_Recv(&num_clusters,1           ,MPI_INT  ,MASTER,0,MPI_COMM_WORLD,&status);
		centroids=malloc(sizeof(Point)*num_clusters);
		MPI_Recv(centroids    ,num_clusters,MPI_POINT,MASTER,0,MPI_COMM_WORLD,&status);
		printf("part_size =%d\n",job_size);
		received_points=(Point*)malloc(sizeof(Point)*job_size);
		slave_clusters=(int*)malloc(sizeof(int)*job_size);
		MPI_Recv(received_points,job_size,MPI_POINT      ,MASTER,0,MPI_COMM_WORLD,&status);
		printf("Recebido [%d]\n",rank);

		MPI_Barrier(MPI_COMM_WORLD);

		while(1)
		{
			printf("Calculando novo cluster [%d]\n",rank);
			for(dex=0;dex<job_size;dex++)
			{
				slave_clusters[dex]=getParent(received_points[dex],centroids,num_clusters);
			}

			printf("enviando ao Master [%d]\n",rank);
			MPI_Send(slave_clusters,job_size, MPI_INT,MASTER, 0, MPI_COMM_WORLD);
			MPI_Barrier(MPI_COMM_WORLD);
			MPI_Barrier(MPI_COMM_WORLD);
			MPI_Recv(&job_done,1, MPI_INT,MASTER,0,MPI_COMM_WORLD,&status);

			if(job_done==1) //trabalho finalizado
				break;

			//recebendo os centroids do processo Master
			MPI_Recv(centroids,num_clusters,MPI_POINT,MASTER,0, MPI_COMM_WORLD,&status);

			MPI_Barrier(MPI_COMM_WORLD);
		}
	}
	//finaliza tudo
	MPI_Finalize();
    	return 0;
}