/* (c) 2005, Tomas Plachetka */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/time.h>
#include <math.h>

#include "tpl.h"
#include "work.h"

/*****************************************************************************
* Local preprocessor defines
******************************************************************************/

#define DEBUG(x)

#define RANK_MASTER 0
#define RANK_WORKER 1

/*****************************************************************************
* Local typedefs
******************************************************************************/

enum 
{
  MSG_JOB_REQ,
  MSG_JOB,
  MSG_QUIT
};

typedef struct job JOB;

struct job
{
  int tile;
  double result;
};

/*****************************************************************************
* Local variables
******************************************************************************/

static int my_rank;
int nr_procs;
int nr_tiles;

/*****************************************************************************
* Static functions
******************************************************************************/

static int match_all(int sender, int tag, void *message);
static void master();
static void worker();

static void do_work(void *tile);
static double f(double arg);

static int match_all(int sender, int tag, void *message)
{
  return(1);
}

static void master(void)
{
  int worker;
  void *message;
  void *msg_job;
  int sender;
  int tag;
  
  int tile;
  
  double integral, job_result;

  DEBUG(printf("master initialised\n");)
  integral = 0.0;
  
  /* Distribute jobs and collect results. */
  tile = 0;
  while (tile < nr_tiles)
  {
    tpl_recv(match_all, &sender, &tag, &message);
    tpl_upkdouble(message, &job_result, 1);
    tpl_destroy(message);
    integral += job_result;

    tpl_create(&msg_job);
    tpl_pkint(msg_job, &tile, 1);
    tpl_send(&sender, 1, MSG_JOB, msg_job);
    
    tile++;
  }
  
  /* Distribute MSG_QUIT messages. */
  for (worker = RANK_WORKER; worker < nr_procs; worker++)
  {
    tpl_recv(match_all, &sender, &tag, &message);
    tpl_upkdouble(message, &job_result, 1);
    tpl_destroy(message);
    integral += job_result;

    tpl_create(&msg_job);
    tpl_send(&sender, 1, MSG_QUIT, msg_job);
  }
  
  printf("PI=%lf\n", integral * 4);
}

static void worker(void)
{
  int sender;
  int tag;
  void *message;
  void *msg_job_req;
  void *msg_job;
  JOB job;
  
  double elapsed;
  
  int rank_master = RANK_MASTER;

  job.result = 0.0;
  do
  {
    /* Send a job request. */
    tpl_create(&msg_job_req);
    tpl_pkdouble(msg_job_req, &(job.result), 1);
    tpl_send(&rank_master, 1, MSG_JOB_REQ, msg_job_req);
    
    /* Receive a job, if there is some. */
    tpl_recv(match_all, &sender, &tag, &msg_job);
    if (tag == MSG_JOB)
    {
      /* Compute the tile. */
      tpl_upkint(msg_job, &(job.tile), 1);
      tpl_destroy(msg_job);
      /* job.result = f(((double) job.tile) / nr_tiles) / nr_tiles; */
      elapsed = work(do_work, (void *) &job);
      /* printf("%lf\n", elapsed); fflush(stdout); */
    }
  } while (tag != MSG_QUIT);
}

static void do_work(void *job)
{
  ((JOB *) job)->result = f(((double) ((JOB *) job)->tile) / 
   nr_tiles) / nr_tiles;
}

static double f(double arg)
{
  return(((double) 1.0) / (1 + arg * arg));
}

int main(int argc, char **argv)
{
  if (argc > 1)
    nr_procs = atoi(argv[1]);
  
  tpl_initialize(&nr_procs, &my_rank, &argc, &argv);

  if (argc != 3)
  {
    printf("Usage: pi <nr_procs> <nr_tiles>\n");
    tpl_deinitialize();
    _exit(1);
  }

  nr_tiles = atoi(argv[2]);
  
  work_initialize();

  switch(my_rank)
  {
    case RANK_MASTER:
      master();
      break;
    
    default:
      worker();
      break;
  }

  /* Terminate the program. */
  work_deinitialize();
  tpl_deinitialize();

  return(0);
}
