[MGR][SEM1][SOWW] Lab2

#include <stdio.h>
#include <mpi.h>
#include <math.h>
#define PRECISION 0.000001
#define RANGESIZE 1
#define DATA 0
#define RESULT 1
#define FINISH 2

//#define DEBUG1

#define QUEUE_SIZE 100
typedef struct _MY_QUEUE {
  double queue[QUEUE_SIZE][2];
  int queueSize;
  int queueHead;
  int queueTail;
  int queueId;
} MY_QUEUE;

int queuePut(double *pItem, MY_QUEUE *pQueue, int myrank)
{
  int result = 0;

  if(pQueue->queueSize < QUEUE_SIZE - 1)
  {
    pQueue->queue[pQueue->queueTail][0] = pItem[0];
    pQueue->queue[pQueue->queueTail][1] = pItem[1];

    pQueue->queueTail++;
    if(pQueue->queueTail == QUEUE_SIZE)
    {
    pQueue->queueTail = 0;
    }
    pQueue->queueSize++;

    result = 1;
  }

#ifdef DEBUG1
      printf("\nProcess #%d queue #%d put. Size %d, Head %d, Tail %d", myrank, pQueue->queueId, pQueue->queueSize, pQueue->queueHead, pQueue->queueTail);
      fflush(stdout);
#endif
  return result;
}

int queuePop(double *pItem, MY_QUEUE *pQueue, int myrank)
{
  int result = 0;

  if(pQueue->queueSize != 0)
  {
    pItem[0] = pQueue->queue[pQueue->queueHead][0];
    pItem[1] = pQueue->queue[pQueue->queueHead][1];

    pQueue->queueHead++;
    if(pQueue->queueHead == QUEUE_SIZE)
    {
    pQueue->queueHead = 0;
    }
    pQueue->queueSize--;

    result = 1;
  }

#ifdef DEBUG1
      printf("\nProcess #%d queue #%d pop. Size %d, Head %d, Tail %d", myrank, pQueue->queueId, pQueue->queueSize, pQueue->queueHead, pQueue->queueTail);
      fflush(stdout);
#endif

  return result;
}

double f(double x) {
  return sin(x)*sin(x)/x;
}

double SimpleIntegration(double a,double b) {
  double i;
  double sum=0;
  for (i=a;i<b;i+=PRECISION)
    sum+=f(i)*PRECISION;
  return sum;
}

int main(int argc, char **argv)
{
  int myrank,proccount;
  double a=1,b=100;
  double range[2];
  double tmpRange[2];
  double result=0;
  double resulttemp[2];
  double tmpResultTemp[2];
  int sentcount=0;
  int i;
  //Slave probe vars
  int flag;
  //\Slave probe vars

  MPI_Request SendReq;
  MPI_Request RcvReq;
  MPI_Status status;
  // Initialize MPI
  MPI_Init(&argc, &argv);
  // find out my rank
  MPI_Comm_rank(MPI_COMM_WORLD, &myrank);
  // find out the number of processes in MPI_COMM_WORLD
  MPI_Comm_size(MPI_COMM_WORLD, &proccount);
  MY_QUEUE gSendQueue;
  MY_QUEUE gRcvQueue;

  gSendQueue.queueSize = 0;
  gSendQueue.queueHead = 0;
  gSendQueue.queueTail = 0;
  gSendQueue.queueId = 0;
  gRcvQueue.queueSize = 0;
  gRcvQueue.queueHead = 0;
  gRcvQueue.queueTail = 0;
  gRcvQueue.queueId = 1;

  if (proccount<2)
  {
    printf("Run with at least 2 processes");
    MPI_Finalize();
#ifdef DEBUG
      printf("\nMaster received result %f from process %d",resulttemp[0],status.MPI_SOURCE);
      fflush(stdout);
#endif
    return -1;
  }
  if (((b-a)/RANGESIZE)<2*(proccount-1))
  {
    printf("More subranges needed");
    MPI_Finalize();
    return -1;
  }
  // now the master will distribute the data and slave processes will perform computations
  if (myrank==0) {
    range[0]=a;
    // first distribute some ranges to all slaves
    for(i=1;i<proccount;i++) {
      range[1]=range[0]+RANGESIZE;
#ifdef DEBUG
      printf("\nMaster sending range %f,%f to process %d",range[0],range[1],i);
      fflush(stdout);
#endif
      // send it to process i
      MPI_Isend(range,2,MPI_DOUBLE,i,DATA,MPI_COMM_WORLD,&SendReq);
      sentcount++;
      range[0]=range[1];
    }
    do {
      // distribute remaining subranges to the processes which have completed their parts
      MPI_Recv(resulttemp,2,MPI_DOUBLE,MPI_ANY_SOURCE,RESULT,MPI_COMM_WORLD,&status);
      result+=resulttemp[0];
#ifdef DEBUG
      printf("\nMaster received result %f from process %d",resulttemp[0],status.MPI_SOURCE);
      fflush(stdout);
#endif
      // check the sender and send some more data
      range[1]=range[0]+RANGESIZE;
      if (range[1]>b) range[1]=b;
#ifdef DEBUG
      printf("\nMaster sending range %f,%f to process %d",range[0],range[1],status.MPI_SOURCE);
      fflush(stdout);
#endif
      MPI_Isend(range,2,MPI_DOUBLE,status.MPI_SOURCE,DATA,MPI_COMM_WORLD,&SendReq);
      range[0]=range[1];
    } while (range[1]<b);
    // now receive results from the processes
    for(i=0;i<(proccount-1);i++) {
      MPI_Recv(resulttemp,2,MPI_DOUBLE,MPI_ANY_SOURCE,RESULT,MPI_COMM_WORLD,&status);
#ifdef DEBUG
      printf("\nMaster received result %f from process %d",resulttemp[0],status.MPI_SOURCE);
      fflush(stdout);
#endif
      result+=resulttemp[0];
    }
    // shut down the slaves
    for(i=1;i<proccount;i++) {
      MPI_Isend(NULL,0,MPI_DOUBLE,i,FINISH,MPI_COMM_WORLD, &SendReq);
    }
    // now display the result
    printf("\nHi, I am process 0, the result is %f\n",result);
  } else { // slave
    // this is easy - just receive data and do the work
    do {
      MPI_Probe(0,MPI_ANY_TAG,MPI_COMM_WORLD,&status);
      if (status.MPI_TAG==DATA) {
    MPI_Iprobe(0, DATA, MPI_COMM_WORLD, &flag, &status);
    if(flag)
    {
      MPI_Irecv(range,2,MPI_DOUBLE,0,DATA,MPI_COMM_WORLD,&RcvReq);
      queuePut(range, &gRcvQueue, myrank);
    }

    if(queuePop(tmpRange, &gRcvQueue, myrank))
    {
      // compute my part
      resulttemp[0]=SimpleIntegration(tmpRange[0],tmpRange[1]);
      // send the result back
      queuePut(resulttemp, &gSendQueue, myrank);
    }

    if(queuePop(tmpResultTemp, &gSendQueue, myrank))
    {
      MPI_Isend(tmpResultTemp,2,MPI_DOUBLE,0,RESULT,MPI_COMM_WORLD, &SendReq);
    }
      }
    } while (status.MPI_TAG!=FINISH);
  }
  // Shut down MPI
  MPI_Finalize();
  return 0;
}