[SOWW][Lab2]Async

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

#define DEBUG3

#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;
  }

  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;
  }


  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();

    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;

      // send it to process i
      MPI_Isend(range,2,MPI_DOUBLE,i,DATA,MPI_COMM_WORLD,&SendReq);
      sentcount++;
      range[0]=range[1];
    }

    i = 1;
    do {
      // check the sender and send some more data
      range[1]=range[0]+RANGESIZE;
      if (range[1]>b) range[1]=b;

      MPI_Isend(range,2,MPI_DOUBLE,i,DATA,MPI_COMM_WORLD,&SendReq);
      range[0]=range[1];

      sentcount++;

      i++;
      if(i == proccount) i = 1;
    } while (range[1]<b);

    while(sentcount != 0)
    {
        MPI_Iprobe(MPI_ANY_SOURCE,RESULT,MPI_COMM_WORLD,&flag,&status);
        if(flag && status.MPI_TAG==RESULT)
        {
            MPI_Irecv(resulttemp,2,MPI_DOUBLE,MPI_ANY_SOURCE,RESULT,MPI_COMM_WORLD,&RcvReq);

            result+=resulttemp[0];
            sentcount--;
        }
    }

    // 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_Iprobe(0, DATA, MPI_COMM_WORLD, &flag, &status);
      if(flag && status.MPI_TAG==DATA)
      {
        MPI_Irecv(range,2,MPI_DOUBLE,0,DATA,MPI_COMM_WORLD,&RcvReq);
        queuePut(range, &gRcvQueue, myrank);

#ifdef DEBUG3
      printf("\nSlave #%d received range %f to %f",myrank, range[0],range[1]);
      fflush(stdout);
#endif
      }

      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);
      }

      MPI_Iprobe(0, FINISH, MPI_COMM_WORLD, &flag, &status);
    } while (status.MPI_TAG!=FINISH);
  }
  // Shut down MPI
  MPI_Finalize();
  return 0;
}