Homec4science
Diffusion Dynamic loop scheduling using MPI passive-target remote memory access (master) Browse

dls_MPI_RMA/PSIA_single_thread_master_worker.c
4672720499a7master

Last Change
Raw File

PSIA_single_thread_master_worker.c
View Options

/*********************************************************************************
* Copyright (c) 2018 *
* Ahmed Eleliemy <ahmed.eleliemy@unibas.ch> *
* University of Basel, Switzerland *
* All rights reserved. *
* This code is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* This program is free software; you can redistribute it and/or modify it *
* under the terms of the license (GNU LGPL) which comes with this package. *
*********************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>
#include <math.h>
#include "dls_core.h"
#include "SpinDataStructures.h"
#include "SpinCalculator.h"
#include "ObjectLoader.h"
#define MIN(a,b) (((a)<(b))?(a):(b))
#define MASTER 0
#ifndef CHECK_PERIOD
#define CHECK_PERIOD 1
#endif
#define REQ_WORK 1000
#define ASSI_WORK 1001
#define TERMINATE 1002
double * weights;
//-------- function headers -------
void parallel_single_thread_master_test(int argc, char** argv);
void single_thread_master(int argc, char ** argv, int number_of_workers, int id);
void worker(int argc, char ** argv, int number_of_workers, int id);
void perform_static_assignment(struct Scheduling_params * params, double * total_calc_time, int * executed_at_id, double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle);
void perform_dynamic_assignment(struct Scheduling_params * params, double * total_calc_time, int * executed_at_id, int id, double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle);
void request_static_assignment (int id, int number_of_tasks, double *total_calc_time,double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle);
void request_dynamic_assignment(int number_of_tasks, double * total_request_time, double * total_calc_time, double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle);
//--------- main -------
int main (int argc, char** argv)
{
parallel_single_thread_master_test(argc,argv);
return 0;
}
//------- master-worker functions ---------------
void single_thread_master(int argc, char ** argv, int number_of_workers, int id)
{
// variables for scheduling
int flag=0;
MPI_Status status;
int terminated=0;
int master_is_terminated=0;
int my_start_index=0;
int my_end_index=0;
struct Scheduling_params params;
char dummy='\0';
int indices [2];
if(argc<7)
{
printf("Usgae: ./ObjectTester.o iSpinPoints iImageWidth fBinSize iSupportAngle(PI/x) iDLSMethods(0-STATIC 1-SS 2-FSC 3-GSS 4-FAC 5-TSS) sFileName\n");
MPI_Abort(MPI_COMM_WORLD, -1);
}
int number_of_tasks=atoi(argv[1]);
int method=atoi(argv[5]);
printf("-------------------\n");
printf("Experiment type single-thread non-dedicated master\n");
printf("Number of workers %d \n", number_of_workers);
printf("Number of tasks %d \n", number_of_tasks);
printf("Scheduling method %d \n", method);
printf("Interrupt rate %d \n", CHECK_PERIOD);
printf("------------------- \n");
// variables for the PSIA
int image_width=atoi(argv[2]);
double bin_size=atof(argv[3]);
double support_angle=M_PI/atof(argv[4]);
char * filename=argv[6];
MeshObject* target_object = NULL;
double * images = NULL;
double * results=NULL;
double * all_double_data=NULL;
int * all_int_data=NULL;
params.scheduled_tasks=0;
params.total_tasks=number_of_tasks;
params.total_workers=number_of_workers;
params.step=0;
params.method=method;
params.last_chunk=0;
if(method == 2 && argc == 9)
{
double h = atof(argv[7]);
double sigma = atof(argv[8]);
params.constant=helper_FSC(number_of_workers,number_of_tasks,h,sigma);
}
else if(method ==5 && argc == 9)
{
int F = atoi(argv[7]);
int L = atoi(argv[8]);
int D = helper_TSS(number_of_workers,number_of_tasks, F, L);
params.last_chunk = F + D;
params.constant = D;
params.minimum_chunk=L;
}
else if(method ==5 || method ==2 )
{
printf("Usgae: ./ObjectTester.o iSpinPoints iImageWidth fBinSize iSupportAngle(PI/x) iDLSMethods(2-FSC 5-TSS) sFileName (h,sigma) or (F,L) \n");
MPI_Abort(MPI_COMM_WORLD, -1);
}
double start_time= MPI_Wtime();
images= (double *) calloc(number_of_tasks * image_width * image_width, sizeof(double));
results = (double *) malloc(sizeof(double)*number_of_tasks*image_width*image_width);
target_object=loadofffile(filename);
calcPointsNormal(target_object);
MPI_Bcast(&target_object->iPointCount,1,MPI_INT,MASTER,MPI_COMM_WORLD);
//packing all data
int position=0;
int size_of_double_buffer_in_bytes=12*sizeof(double)*target_object->iPointCount;
int size_of_int_buffer_in_bytes=sizeof(int)*target_object->iPointCount;
all_double_data = (double*) malloc(size_of_double_buffer_in_bytes);
all_int_data = (int *) malloc(size_of_int_buffer_in_bytes);
MPI_Pack(target_object->dptrXPoints, target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
MPI_Pack(target_object->dptrYPoints, target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
MPI_Pack(target_object->dptrZPoints, target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
MPI_Pack(target_object->dptrXNPoints, 3*target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
MPI_Pack(target_object->dptrYNPoints, 3*target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
MPI_Pack(target_object->dptrZNPoints, 3*target_object->iPointCount, MPI_DOUBLE, all_double_data, size_of_double_buffer_in_bytes,&position,MPI_COMM_WORLD);
position=0;
MPI_Pack(target_object->facesPerPoint, target_object->iPointCount, MPI_INT, all_int_data, size_of_int_buffer_in_bytes,&position,MPI_COMM_WORLD);
//sending data
MPI_Bcast(all_double_data,12*target_object->iPointCount,MPI_DOUBLE,MASTER,MPI_COMM_WORLD);
MPI_Bcast(all_int_data,target_object->iPointCount,MPI_INT,MASTER,MPI_COMM_WORLD);
free(all_double_data);
free(all_int_data);
double start_calc_time=0;
double end_calc_time=0;
double total_calc_time=0;
double total_request_time=0;
double max_loop_time=0;
double min_loop_time=0;
int check_request=0;
int to_execute=0;
int * executed_at_id = (int *) calloc(params.total_workers,sizeof(int));
double execution_times[number_of_workers];
for (int i = 0; i < number_of_workers; i++)
{
executed_at_id[i]=0;
}
if(params.method==0)
{
perform_static_assignment(&params, &total_calc_time,executed_at_id,images,target_object,bin_size,image_width,support_angle);
}
else
{
perform_dynamic_assignment(&params,&total_calc_time,executed_at_id,id,images,target_object,bin_size,image_width,support_angle);
}
// get data from workers
// printf("master %d spent %f in loop\n",id ,total_calc_time);
MPI_Reduce(images,results,number_of_tasks*image_width*image_width,MPI_DOUBLE,MPI_SUM,MASTER,MPI_COMM_WORLD);
double end_time= MPI_Wtime();
MPI_Gather(&total_calc_time,1,MPI_DOUBLE,&execution_times,1,MPI_DOUBLE,MASTER,MPI_COMM_WORLD);
//MPI_Reduce(&total_calc_time, &max_loop_time, 1, MPI_DOUBLE, MPI_MAX, MASTER,MPI_COMM_WORLD);
//MPI_Reduce(&total_calc_time, &min_loop_time, 1, MPI_DOUBLE, MPI_MIN, MASTER,MPI_COMM_WORLD);
/* int i,j, c;
for (c = 0; c < 0; c++)
{
printf("image %d: \n",c);
for (i = 0; i < image_width; i++)
{
for (j = 0; j < image_width; j++)
printf("%f ", images[ (c * image_width * image_width) +( i * image_width) + j]);
printf("\n");
}
printf("\n");
}*/
max_loop_time = execution_times[0];
min_loop_time= execution_times[0];
printf("----- ranks calculation times and total scheduled tasks -------\n");
printf("id time(s) total_tasks\n");
printf("%d %lf %d\n",id, execution_times[0], executed_at_id[0]);
for (int i = 1; i < number_of_workers; i++)
{
if(max_loop_time < execution_times[i] )
max_loop_time = execution_times[i];
else if(min_loop_time > execution_times[i])
min_loop_time = execution_times[i];
printf("%d %lf %d\n",i, execution_times[i],executed_at_id[i]);
}
printf("Program execution time %lf\n", end_time-start_time);
printf("Paralell max calculation time %lf\n", max_loop_time);
printf("Paralell min calculation time %lf\n", min_loop_time);
free(executed_at_id);
//printf("average waiting time %lf\n", end_time-start_time);
// free resources
if(results!=NULL)
free(results);
if(images!=NULL)
free(images);
if(target_object!=NULL)
{
free(target_object->facesPerPoint);
free(target_object->dptrXPoints);
free(target_object->dptrYPoints);
free(target_object->dptrZPoints);
free(target_object->dptrXNPoints);
free(target_object->dptrYNPoints);
free(target_object->dptrZNPoints);
free(target_object);
}
}
void worker (int argc, char ** argv,int number_of_workers, int id)
{
// variables scheduling
char dummy='\0';
int indices [2];
int number_of_tasks=atoi(argv[1]);
int method = atoi(argv[5]);
int flag=0;
MPI_Status status;
int terminate=0;
int request_counter=0;
int image_width=atoi(argv[2]);
double bin_size=atof(argv[3]);
double support_angle=M_PI/atof(argv[4]);
char * filename=argv[6];
double * images = NULL;
double * results=NULL;
double * all_double_data=NULL;
int * all_int_data=NULL;
MeshObject *target_object=(MeshObject*) malloc(sizeof(MeshObject));
images= (double *) calloc(number_of_tasks*image_width*image_width,sizeof(double));
MPI_Bcast(&target_object->iPointCount,1,MPI_INT,MASTER,MPI_COMM_WORLD);
int p=0;
int size_of_double_buffer_in_bytes=12*sizeof(double)*target_object->iPointCount;
int size_of_int_buffer_in_bytes=sizeof(int)*target_object->iPointCount;
all_double_data= (double*) malloc(12*sizeof(double)*target_object->iPointCount);
all_int_data= (int *) malloc(sizeof(int)*target_object->iPointCount);
target_object->dptrXPoints =malloc(sizeof(double)*target_object->iPointCount);
target_object->dptrYPoints =malloc(sizeof(double)*target_object->iPointCount);
target_object->dptrZPoints =malloc(sizeof(double)*target_object->iPointCount);
target_object->facesPerPoint =malloc(sizeof(int)*target_object->iPointCount);
target_object->dptrXNPoints =malloc(3*sizeof(double)*target_object->iPointCount);
target_object->dptrYNPoints =malloc(3*sizeof(double)*target_object->iPointCount);
target_object->dptrZNPoints =malloc(3*sizeof(double)*target_object->iPointCount);
MPI_Bcast(all_double_data,12*target_object->iPointCount,MPI_DOUBLE,MASTER,MPI_COMM_WORLD);
MPI_Bcast(all_int_data,target_object->iPointCount,MPI_INT,MASTER,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrXPoints,target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrYPoints,target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrZPoints,target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrXNPoints,3*target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrYNPoints,3*target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
MPI_Unpack(all_double_data,size_of_double_buffer_in_bytes, &p,target_object->dptrZNPoints,3*target_object->iPointCount,MPI_DOUBLE,MPI_COMM_WORLD);
p=0;
MPI_Unpack(all_int_data,size_of_int_buffer_in_bytes, &p,target_object->facesPerPoint,target_object->iPointCount ,MPI_INT,MPI_COMM_WORLD);
free(all_double_data);
free(all_int_data);
double start_request_time=0;
double end_request_time =0;
double start_calc_time=0;
double end_calc_time=0;
double total_calc_time=0;
double total_request_time=0;
// printf("worker %d check which method\n",id);
if(method==0)
{
// printf("worker %d will request static assignmnet\n",id);
request_static_assignment(id, number_of_tasks, &total_calc_time,images,target_object,bin_size,image_width,support_angle);
}
else
{
request_dynamic_assignment(number_of_tasks, &total_request_time,&total_calc_time,images,target_object,bin_size,image_width,support_angle);
}
//send data to master
// printf("worker %d spent %f in loop\n",id ,total_calc_time);
MPI_Reduce(images,NULL,number_of_tasks*image_width*image_width,MPI_DOUBLE,MPI_SUM,MASTER,MPI_COMM_WORLD);
MPI_Gather(&total_calc_time,1,MPI_DOUBLE,NULL,1,MPI_DOUBLE,MASTER,MPI_COMM_WORLD);
//MPI_Reduce(&total_calc_time, NULL, 1, MPI_DOUBLE, MPI_MAX, MASTER,MPI_COMM_WORLD );
//MPI_Reduce(&total_calc_time, NULL, 1, MPI_DOUBLE, MPI_MIN, MASTER,MPI_COMM_WORLD );
// free resources
if(images!=NULL)
free(images);
if(target_object!=NULL)
{
free(target_object->facesPerPoint);
free(target_object->dptrXPoints);
free(target_object->dptrYPoints);
free(target_object->dptrZPoints);
free(target_object->dptrXNPoints);
free(target_object->dptrYNPoints);
free(target_object->dptrZNPoints);
free(target_object);
}
}
void perform_static_assignment(struct Scheduling_params * params, double * total_calc_time, int executed_at_id [], double * images,MeshObject* target_object,double bin_size,int image_width, double support_angle)
{
int terminated=0;
int indices[2];
double start_calc_time =0.0;
double end_calc_time =0.0;
*total_calc_time = 0.0;
char dummy='\0';
MPI_Status status;
while(terminated < params[0].total_workers-1)
{
MPI_Recv(&dummy,1,MPI_CHAR,MPI_ANY_SOURCE,REQ_WORK,MPI_COMM_WORLD,&status);
calculate_next_chunk_stateless(params[0], &indices[0],&indices[1]);
MPI_Send(&indices,2,MPI_INT,status.MPI_SOURCE,ASSI_WORK,MPI_COMM_WORLD);
params[0].scheduled_tasks += indices[1] - indices[0];
params[0].step++;
params[0].last_chunk= indices[1] - indices[0];
terminated++;
executed_at_id[status.MPI_SOURCE]+=indices[1] - indices[0];
//printf("received request from %d assign it %d treminated %d\n",status.MPI_SOURCE,indices[1] - indices[0], terminated );
}
//printf("all workers has been assigned work remaining_tasks %d\n", params[0].total_tasks - params[0].scheduled_tasks );
if(params[0].scheduled_tasks < params[0].total_tasks)
{
calculate_next_chunk_stateless(params[0], &indices[0],&indices[1]);
//printf("master will work on %d\n", indices[1] - indices[0]);
start_calc_time = MPI_Wtime();
calcSpinImages(images,target_object,bin_size, image_width, support_angle,false, params[0].total_tasks,indices[0], indices[1]);
end_calc_time = MPI_Wtime();
*total_calc_time = end_calc_time - start_calc_time;
executed_at_id[0] += indices[1] - indices[0];
}
//printf("master %d finished its part in %f \n",0, *total_calc_time);
}
void perform_dynamic_assignment(struct Scheduling_params * params, double * total_calc_time, int * executed_at_id, int id, double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle)
{
int terminated=0;
int master_is_terminated=0;
int my_start_index=0;
int my_end_index=0;
double start_calc_time=0.0;
double end_calc_time=0.0;
*total_calc_time=0.0;
int check_request=0;
int to_execute=0;
int flag=0;
int dummy='\0';
MPI_Status status;
int indices[2];
start_calc_time = MPI_Wtime();
while(terminated < params[0].total_workers-1 || !master_is_terminated)
{
if( !check_request && !master_is_terminated)
{
if(my_start_index < my_end_index) // this condition is never true if CHECK_PERIOD = chunk_size
{
to_execute = my_end_index - my_start_index;
start_calc_time = MPI_Wtime();
calcSpinImages(images,target_object,bin_size, image_width, support_angle,false, params[0].total_tasks,my_start_index, my_start_index + MIN(CHECK_PERIOD,to_execute));
end_calc_time = MPI_Wtime();
*total_calc_time += end_calc_time - start_calc_time;
my_start_index += MIN(CHECK_PERIOD,to_execute);
}
else if(params[0].scheduled_tasks < params[0].total_tasks)
{
params[0].my_weight= weights[MASTER];
//printf("id %d weight %lf method %d\n",MASTER, weights[MASTER],params[0].method);
calculate_next_chunk_stateless(params[0], &indices[0],&indices[1]);
params[0].scheduled_tasks+= indices[1] - indices[0];
params[0].step++;
params[0].last_chunk= indices[1] - indices[0];
my_start_index=indices[0];
my_end_index=indices[1];
to_execute = my_end_index - my_start_index;
executed_at_id[0] += to_execute;
calcSpinImages(images,target_object,bin_size, image_width, support_angle,false, params[0].total_tasks,my_start_index, my_start_index + MIN(CHECK_PERIOD,to_execute));
my_start_index += MIN(CHECK_PERIOD,to_execute);
//printf("chunk of %d scheduled at master %d \n",indices[1] - indices[0],MASTER);
}
else if(!master_is_terminated)
{
master_is_terminated=1;
// printf("Master terminated itself executed %d \n", executed_at_id[id] );
}
check_request =1;
}
else
{
flag=0;
MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, &status);
while(flag)
{
if(status.MPI_TAG == REQ_WORK)
{
MPI_Recv(&dummy,1,MPI_CHAR,status.MPI_SOURCE,REQ_WORK,MPI_COMM_WORLD,&status);
if(params[0].scheduled_tasks < params[0].total_tasks)
{
if(executed_at_id[status.MPI_SOURCE]!=0 && params[0].method==0)
{
MPI_Send(&dummy,1,MPI_CHAR,status.MPI_SOURCE,TERMINATE,MPI_COMM_WORLD);
terminated++;
}
else
{
params[0].my_weight= weights[status.MPI_SOURCE];
//printf("id %d weight %lf\n",status.MPI_SOURCE, weights[status.MPI_SOURCE]);
calculate_next_chunk_stateless(params[0], &indices[0],&indices[1]);
params[0].scheduled_tasks+= indices[1] - indices[0];
params[0].step++;
params[0].last_chunk = indices[1] - indices[0];
MPI_Send(&indices,2,MPI_INT,status.MPI_SOURCE,ASSI_WORK,MPI_COMM_WORLD);
executed_at_id[status.MPI_SOURCE]+=indices[1] - indices[0];
// printf("chunk of %d scheduled at worker %d \n",indices[1] - indices[0],status.MPI_SOURCE);
}
}
else
{
// printf("Master terminated the following worker %d executed %d \n",status.MPI_SOURCE, executed_at_id[status.MPI_SOURCE]);
MPI_Send(&dummy,1,MPI_CHAR,status.MPI_SOURCE,TERMINATE,MPI_COMM_WORLD);
terminated++;
}
}
MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, &status);
}
check_request = 0;
}
}
end_calc_time = MPI_Wtime();
*total_calc_time += end_calc_time - start_calc_time;
}
void request_static_assignment (int id, int number_of_tasks, double *total_calc_time,double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle)
{
int flag=0;
int indices[2];
double start_calc_time = 0.0;
double end_calc_time = 0.0;
MPI_Status status;
*total_calc_time=0.0;
char dummy='\0';
//printf("before sending\n");
MPI_Send(&dummy,1,MPI_CHAR,MASTER,REQ_WORK,MPI_COMM_WORLD);
//printf("worker %d request a chunk\n",id);
MPI_Recv(&indices,2,MPI_INT,MASTER,ASSI_WORK,MPI_COMM_WORLD,&status);
//printf("worker %d received %d \n", id, indices[1]-indices[0] );
start_calc_time=MPI_Wtime();
calcSpinImages(images,target_object,bin_size, image_width, support_angle,false, number_of_tasks,indices[0], indices[1]);
end_calc_time=MPI_Wtime();
*total_calc_time += end_calc_time - start_calc_time;
//printf("worker %d finished its part in %f \n", id, *total_calc_time);
}
void request_dynamic_assignment(int number_of_tasks, double * total_request_time, double * total_calc_time, double * images, MeshObject* target_object,double bin_size,int image_width, double support_angle)
{
int terminated=0;
double start_request_time=0.0;
double end_request_time=0.0;
double start_calc_time=0.0;
double end_calc_time=0.0;
*total_request_time=0.0;
*total_calc_time=0.0;
char dummy='\0';
int flag=0;
MPI_Status status;
int indices[2];
start_calc_time=MPI_Wtime();
while(!terminated)
{
MPI_Send(&dummy,1,MPI_CHAR,MASTER,REQ_WORK,MPI_COMM_WORLD);
flag=0;
while(!flag)
{
MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, &status);
}
if(status.MPI_TAG==ASSI_WORK)
{
MPI_Recv(&indices,2,MPI_INT,MASTER,ASSI_WORK,MPI_COMM_WORLD,&status);
calcSpinImages(images,target_object,bin_size, image_width, support_angle,false, number_of_tasks,indices[0], indices[1]);
}
else
{
terminated=1;
}
}
end_calc_time=MPI_Wtime();
*total_calc_time += end_calc_time - start_calc_time;
}
//------- testing functions ----------
void parallel_single_thread_master_test(int argc, char ** argv)
{
int id, number_of_workers;
MPI_Init(NULL,NULL);
MPI_Comm_size(MPI_COMM_WORLD, &number_of_workers);
MPI_Comm_rank(MPI_COMM_WORLD,&id);
weights=(double *) calloc(number_of_workers,sizeof(double));
//printf("allocated %d\n",id);
//Ahmed
if(argc>=7)
{
if(atoi(argv[5])==6)
{
char processor_name[30];
int name_len;
MPI_Get_processor_name(processor_name, &name_len);
int knl_count= atoi(argv[7]);
double my_weight= WF_helper(knl_count, processor_name, number_of_workers);
MPI_Gather(&my_weight,1,MPI_DOUBLE,weights,1,MPI_DOUBLE,MASTER,MPI_COMM_WORLD);
}
}
if(id==MASTER)
single_thread_master(argc, argv, number_of_workers,id);
else
worker(argc,argv,number_of_workers,id);
free(weights);
MPI_Finalize();
}
c4science · Help