#include "converse.h"
#include <stdio.h>

/* flag to turn content checking on or off */

#define CHECKCONTENTS 1

CpvDeclare(int, hndl_id);
CpvDeclare(int, sendPE);
CpvDeclare(int, MSG_SIZ);
CpvDeclare(int, total_messages);
CpvDeclare(int, num_messages);
CpvDeclare(double, start_time);

void hndlr(int *);
void ConverseRing(int, int);

#if CHECKCONTENTS
void setMessage(int *, int);
int checkMessage(int *, int);
#endif

#define TSCOUNT 4

void lbw_main(argc, argv)
    int argc; char **argv;
{
  CpvInitialize(int, hndl_id);
  CpvInitialize(int, sendPE);
  CpvInitialize(int, MSG_SIZ);
  CpvInitialize(int, total_messages);
  CpvInitialize(int, num_messages);
  CpvInitialize(double, start_time);
  CpvAccess(hndl_id) = CmiRegisterHandler(hndlr);
  ConverseRing(1<<8,1024);
  CsdScheduler(-1); 
  ConverseRing(1<<10,512);
  CsdScheduler(-1); 
  ConverseRing(1<<12,256);
  CsdScheduler(-1); 
  ConverseRing(1<<14,128);
  CsdScheduler(-1); 
  ConverseRing(1<<16,64);
  CsdScheduler(-1); 
  ConverseRing(1<<18,32);
  CsdScheduler(-1); 
  ConverseRing(1<<20,16);
}

void main(int argc, char **argv)
{
  ConverseInit(argc,argv,lbw_main,0,0);
}



void hndlr(msg)
int *msg;
{
  double per_ping, msg_time;

  CpvAccess(num_messages)++;

#if CHECKCONTENTS
  if(!checkMessage(msg, CpvAccess(MSG_SIZ)-CmiMsgHeaderSizeBytes)) {
    CmiPrintf("Message Contents Corrupted.\n");
    exit(1);
  }
#endif

  if(CpvAccess(num_messages) < CpvAccess(total_messages)) {
      if(CpvAccess(num_messages) == TSCOUNT)
	CpvAccess(start_time) = CmiWallTimer();
      CmiSyncSendAndFree(CpvAccess(sendPE), CpvAccess(MSG_SIZ), msg);
  } else {
    if(CmiMyPe() != 0)
      CmiSyncSendAndFree(CpvAccess(sendPE), CpvAccess(MSG_SIZ), msg);
    else {
      msg_time =  CmiWallTimer()-CpvAccess(start_time);
      per_ping =  msg_time/(CmiNumPes()*(CpvAccess(total_messages)-TSCOUNT));
      CmiPrintf("%d\t%.19lf\n", CpvAccess(MSG_SIZ)-CmiMsgHeaderSizeBytes, per_ping);
    }
    CsdExitScheduler();
  }
}

#if CHECKCONTENTS
void setMessage(int *msg, int size)
{
  int i;

  msg = (int *) ((char *)msg + CmiMsgHeaderSizeBytes);
  for(i=0;i<(size/sizeof(int));i++)
    msg[i] = i+13456;
}

int checkMessage(int *msg, int size)
{
  int i;

  msg = (int *) ((char *)msg + CmiMsgHeaderSizeBytes);
  for(i=0;i<(size/sizeof(int));i++)
    if(msg[i]!= i+13456)
      return 0;
  return 1;
}
#endif

void ConverseRing(size, repetitions)
int size, repetitions;
{
  int i, *my_msg;
  
  CpvAccess(total_messages) = TSCOUNT + repetitions;
  CpvAccess(MSG_SIZ) = size+CmiMsgHeaderSizeBytes;
  CpvAccess(num_messages) = 0;
  CpvAccess(sendPE) = (CmiMyPe() + 1) % CmiNumPes();

  if(CmiMyPe() == 0) { 
    my_msg = (int *)CmiAlloc(CpvAccess(MSG_SIZ));
#if CHECKCONTENTS
    setMessage(my_msg, size);
#endif
    CmiSetHandler(my_msg, CpvAccess(hndl_id));
    CmiSyncSendAndFree(CpvAccess(sendPE), CpvAccess(MSG_SIZ), my_msg);
  }
}
