mpi-isend-m-msg-m-barrier-m-timer.c

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

#define call_mpi( func, ...) { 								\
	int mpi_call_res;										\
	mpi_call_res = func(__VA_ARGS__);						\
	if( mpi_call_res != MPI_SUCCESS ) {						\
		fprintf(stderr, "error on line[%d]\n", __LINE__ );	\
		exit(1); 											\
	}														\
 } 
#define MAX_MSG_SIZE 4194304 // 4 MB
double calculateMean(double data[], int n) {
    double sum = 0.0;
    for (int i = 0; i < n; i++) {
        sum += data[i];
    }
    return sum / n;
}

double calculateStandardDeviation(double data[], int n) {
    double mean = calculateMean(data, n);
    double variance = 0.0;

    for (int i = 0; i < n; i++) {
        variance += pow(data[i] - mean, 2);
    }
    variance /= n; 

    return sqrt(variance);
}
void cbarrier(int rank) {
	// MPI_Barrier(MPI_COMM_WORLD);

	if( rank == 0 ) {
		call_mpi(MPI_Send, NULL, 0, MPI_INT, 1, 1234, MPI_COMM_WORLD);
		call_mpi(MPI_Recv, NULL, 0, MPI_INT, 1, 1234, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
	} else {
		call_mpi(MPI_Recv, NULL, 0, MPI_INT, 0, 1234, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
		call_mpi(MPI_Send, NULL, 0, MPI_INT, 0, 1234, MPI_COMM_WORLD);
		
	}
	return ;
}

int main(int argc, char** argv) {

	int res, size, proc_name_len, rank, msg_size, dest, tag, num_of_iterations, iteration, warmup_iterations;
	char hostname[MPI_MAX_PROCESSOR_NAME];
	char* msg_buf;
	MPI_Status recv_status;
	struct timespec t_start, t_end;
	double xfer_time_usecs;
	double * xfer_times;
	double min_latency = 10000000.0;
	double max_latency = 0.0;
	double deviation;
	MPI_Request issend_request;

	if (argc >= 2)
		num_of_iterations = atoi(argv[1]);
	else
		num_of_iterations = 1000;

	if(num_of_iterations < 0){
		fprintf(stderr, "error: invalid input num_of_iterations  = [%s] . Default value (1000) will be used\n", argv[1] );
		num_of_iterations = 1000;
	}
	warmup_iterations = 100;
	xfer_time_usecs = 0.0;
	msg_buf = (char*)malloc( sizeof(char)*MAX_MSG_SIZE );
	xfer_times = malloc(sizeof(double) * num_of_iterations);

	if( msg_buf == NULL ) {
		fprintf(stderr, "error: failed to malloc at %d\n", __LINE__);
		return 1;
	}
	for(iteration=0; iteration<MAX_MSG_SIZE; ++iteration) {
		msg_buf[iteration] = 0;
	}
	res = MPI_Init(&argc, &argv);
	call_mpi(MPI_Comm_size, MPI_COMM_WORLD, &size ) ;
	call_mpi(MPI_Comm_rank, MPI_COMM_WORLD, &rank);
	call_mpi(MPI_Get_processor_name, hostname, &proc_name_len);

	printf("Rank [%d/%d] runs on [%s]\n", rank, size, hostname);

	tag = 71023;
	for(msg_size = 0; msg_size <= MAX_MSG_SIZE; msg_size = (msg_size ? msg_size*2 : 1) )  {
		
		// warmup runs
		for(iteration = 0; iteration < warmup_iterations; ++iteration) {
			if( rank == 0 ) {
				call_mpi(MPI_Isend, msg_buf, msg_size, MPI_CHAR, 1, tag, MPI_COMM_WORLD, &issend_request );
				call_mpi(MPI_Wait, &issend_request, MPI_STATUS_IGNORE);
			} else {
				call_mpi(MPI_Recv, msg_buf, msg_size, MPI_CHAR, 0, tag, MPI_COMM_WORLD, &recv_status);
			}
		}

		// actual runs
		if( rank == 0 ) {
			
			for(iteration = 0; iteration < num_of_iterations; ++iteration) {

				cbarrier(rank);
				res = clock_gettime(CLOCK_MONOTONIC, &t_start );
				call_mpi(MPI_Isend, msg_buf, msg_size, MPI_CHAR, 1, tag, MPI_COMM_WORLD, &issend_request);
				call_mpi(MPI_Wait, &issend_request, MPI_STATUS_IGNORE);
				res = clock_gettime(CLOCK_MONOTONIC, &t_end );
				xfer_time_usecs = xfer_time_usecs + (t_end.tv_sec - t_start.tv_sec)*1000000.0 + (t_end.tv_nsec - t_start.tv_nsec)/1000.0;
				xfer_times[iteration] =  (t_end.tv_sec - t_start.tv_sec)*1000000.0 + (t_end.tv_nsec - t_start.tv_nsec)/1000.0;
				if (xfer_times[iteration] > max_latency)
					max_latency = xfer_times[iteration];
				if (xfer_times[iteration] < min_latency)
					min_latency = xfer_times[iteration];
			}
			deviation = calculateStandardDeviation(xfer_times, num_of_iterations);
			
		} else { // rank = 1

			for(iteration = 0; iteration < num_of_iterations; ++iteration) {
				cbarrier(rank);
				call_mpi(MPI_Recv, msg_buf, msg_size, MPI_CHAR, 0, tag, MPI_COMM_WORLD, &recv_status);
			}

		}
		
		xfer_time_usecs = xfer_time_usecs/num_of_iterations;
		if( rank == 0) {
			printf("[%s]: avg xfer time: size=%d iters=[%d] avg_latency=%lf usecs min_latency=%lf usecs max_latency=%lf usecs std_dev=%lf \n", argv[0], msg_size, num_of_iterations, xfer_time_usecs, min_latency, max_latency, deviation
			);
		}
	}

	free( msg_buf );
	call_mpi( MPI_Finalize );
	return 0;
}