/*****************************************************************************
 * $Source: /expand6/cvsroot/charm/src/Common/conv-ldb/cldb.spray.c,v $
 * $Author: milind $
 * $Date: 2000/12/25 08:14:12 $
 * $Revision: 1.11 $
 *****************************************************************************/
#include "converse.h"
#include "queueing.h"
#include "cldb.h"
#include <time.h>
#include <stdlib.h>
#include <math.h>

void LoadNotifyFn(int l)
{
}

char *CldGetStrategy(void)
{
  return "spray";
}

#define CYCLE_MILLISECONDS 500
#define DEBUGGING_OUTPUT 0

typedef struct 
{
  int mype;
  int EnqueueHandler;
  int ReduceHandler;
  int AverageHandler;
  int HopHandler;
  double load_reported;
  double load_total;
  int    load_count;
  int    spantree_parent;
  int    spantree_children;
  int    spantree_root;
  int    rebalance;
}
peinfo;

CpvStaticDeclare(peinfo, peinf);

struct loadmsg {
  char core[CmiMsgHeaderSizeBytes];
  double load_total;
};

struct reqmsg {
  char core[CmiMsgHeaderSizeBytes];
};

void CldPropagateLoad(double load);

void CldInitiateReduction()
{
  double load = CldEstimate();
  peinfo *pinf = &(CpvAccess(peinf));
  pinf->load_reported = load;
  CldPropagateLoad(load);
}

void CldPropagateLoad(double load)
{
  struct loadmsg msg;
  peinfo *pinf = &(CpvAccess(peinf));
  pinf->load_total += load;
  pinf->load_count ++;
  if (pinf->load_count == pinf->spantree_children + 1) {
    msg.load_total   = pinf->load_total;
    if (pinf->mype == pinf->spantree_root) {
      if (DEBUGGING_OUTPUT) CmiPrintf("---\n");
      CmiSetHandler(&msg, pinf->AverageHandler);
      CmiSyncBroadcastAll(sizeof(msg), &msg);
    } else {
      CmiSetHandler(&msg, pinf->ReduceHandler);
      CmiSyncSend(pinf->spantree_parent, sizeof(msg), &msg);
    }
    pinf->load_total = 0;
    pinf->load_count = 0;
  }
}

void CldReduceHandler(struct loadmsg *msg)
{
  CldPropagateLoad(msg->load_total);
}

void CldAverageHandler(struct loadmsg *msg)
{
  peinfo *pinf = &(CpvAccess(peinf));
  double load = CldEstimate();
  double average = (msg->load_total / CmiNumPes());
  int rebalance;
  if (load < (average+10) * 1.2) rebalance=0;
  else rebalance = (int)(load - average);
  if (DEBUGGING_OUTPUT)
    CmiPrintf("PE %d load=%6d average=%6d rebalance=%d\n", 
	      CmiMyPe(), CldEstimate(), (int)average, rebalance);
  pinf->rebalance = rebalance;
  CcdCallFnAfter((CcdVoidFn)CldInitiateReduction, 0, CYCLE_MILLISECONDS);
}

void CldEnqueueHandler(char *msg)
{
  int len, queueing, priobits; unsigned int *prioptr;
  CldInfoFn ifn; CldPackFn pfn;
  ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
  CmiSetHandler(msg, CmiGetXHandler(msg));
  CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
}

void CldHopHandler(char *msg)
{
  peinfo *pinf = &(CpvAccess(peinf));
  int len, queueing, priobits; unsigned int *prioptr;
  CldInfoFn ifn; CldPackFn pfn; int pe;

  if (pinf->rebalance) {
    ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
    ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
    if (pfn) {
      pfn(&msg);
      ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
    }
    /* do pe = ((lrand48()&0x7FFFFFFF)%CmiNumPes()); */
    do pe = ((CrnRand()&0x7FFFFFFF)%CmiNumPes());
    while (pe == pinf->mype);
    CmiSyncSendAndFree(pe, len, msg);
    pinf->rebalance--;
  } else {
    CmiSetHandler(msg, CmiGetXHandler(msg));
    CmiHandleMessage(msg);
  }
}

void CldEnqueueMulti(int npes, int *pes, void *msg, int infofn)
{
  int len, queueing, priobits,i; unsigned int *prioptr;
  CldInfoFn ifn = (CldInfoFn)CmiHandlerToFunction(infofn);
  peinfo *pinf = &(CpvAccess(peinf));
  CldPackFn pfn;
  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
  if (pfn) {
    pfn(&msg);
    ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
  }
  CmiSetInfo(msg,infofn);
  CmiSetXHandler(msg, CmiGetHandler(msg));
  CmiSetHandler(msg, pinf->EnqueueHandler);
  for(i=0;i<npes;i++) {
    CmiSyncSend(pes[i], len, msg);
  }
  CmiFree(msg);
}

void CldEnqueue(int pe, void *msg, int infofn)
{
  int len, queueing, priobits; unsigned int *prioptr;
  CldInfoFn ifn; CldPackFn pfn;
  peinfo *pinf = &(CpvAccess(peinf));
  ifn = (CldInfoFn)CmiHandlerToFunction(infofn);
  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
  if (pe != CLD_ANYWHERE) {
    if (pfn && (pe != CmiMyPe())) {
      pfn(&msg);
      ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
    }
    CmiSetInfo(msg, infofn);
    CmiSetXHandler(msg, CmiGetHandler(msg));
    CmiSetHandler(msg, pinf->EnqueueHandler);
    if (pe==CLD_BROADCAST) CmiSyncBroadcastAndFree(len, msg);
    else if (pe==CLD_BROADCAST_ALL) CmiSyncBroadcastAllAndFree(len, msg);
    else CmiSyncSendAndFree(pe, len, msg);
  } else {
    CmiSetInfo(msg, infofn);
    CmiSetXHandler(msg, CmiGetHandler(msg));
    CmiSetHandler(msg, pinf->HopHandler);
    CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
  }
}

void CldModuleInit()
{
  peinfo *pinf;
  CpvInitialize(peinfo, peinf);
  /* srand48(time(0)+CmiMyPe()); */
  CrnSrand((int) (time(0)+CmiMyPe()));
  pinf = &CpvAccess(peinf);
  pinf->mype = CmiMyPe();
  pinf->EnqueueHandler = CmiRegisterHandler(CldEnqueueHandler);
  pinf->ReduceHandler  = CmiRegisterHandler(CldReduceHandler);
  pinf->AverageHandler = CmiRegisterHandler(CldAverageHandler);
  pinf->HopHandler     = CmiRegisterHandler(CldHopHandler);
  pinf->load_total = 0.0;
  pinf->load_count = 0;
  pinf->spantree_children = CmiNumSpanTreeChildren(CmiMyPe());
  pinf->spantree_parent = CmiSpanTreeParent(CmiMyPe());
  pinf->spantree_root = 0;
  pinf->rebalance = 0;
  CldModuleGeneralInit();
  CldInitiateReduction();
}
