Home Research Papers Posters Manuals Talks Download People Help Wiki Internal

arch/mpi/machine.c

Go to the documentation of this file.

/*****************************************************************************
 * $Source: /cvsroot/charm/src/arch/mpi/machine.c,v $
 * $Author: gzheng $
 * $Date: 2008-06-06 21:13:17 $
 * $Revision: 1.111 $
 *****************************************************************************/


#include <stdio.h>
#include <errno.h>
#include "converse.h"
#include <mpi.h>
#if CMK_TIMER_USE_XT3_DCLOCK
#include <catamount/dclock.h>
#endif

#ifdef AMPI
#  warning "We got the AMPI version of mpi.h, instead of the system version--"
#  warning "   Try doing an 'rm charm/include/mpi.h' and building again."
#  error "Can't build Charm++ using AMPI version of mpi.h header"
#endif

/*Support for ++debug: */
#include <unistd.h> /*For getpid()*/
#include <stdlib.h> /*For sleep()*/

#define MULTI_SENDQUEUE    0

#if defined(CMK_SHARED_VARS_POSIX_THREADS_SMP)
#define CMK_SMP 1
#endif

#include "machine.h"

#include "pcqueue.h"

#define FLIPBIT(node,bitnumber) (node ^ (1 << bitnumber))

#if CMK_VERSION_BLUEGENE
#define MAX_QLEN 8
#define NETWORK_PROGRESS_PERIOD_DEFAULT 16
#else
#define NETWORK_PROGRESS_PERIOD_DEFAULT 0
#define MAX_QLEN 200
#endif

#if CMI_MPI_TRACE_USEREVENTS && !defined(CMK_OPTIMIZE) && ! CMK_TRACE_IN_CHARM
CpvStaticDeclare(double, projTraceStart);
# define  START_EVENT()  CpvAccess(projTraceStart) = CmiWallTimer();
# define  END_EVENT(x)   traceUserBracketEvent(x, CpvAccess(projTraceStart), CmiWallTimer());
#else
# define  START_EVENT()
# define  END_EVENT(x)
#endif

/*
    To reduce the buffer used in broadcast and distribute the load from
  broadcasting node, define CMK_BROADCAST_SPANNING_TREE enforce the use of
  spanning tree broadcast algorithm.
    This will use the fourth short in message as an indicator of spanning tree
  root.
*/
#if CMK_SMP
#define CMK_BROADCAST_SPANNING_TREE    0
#else
#define CMK_BROADCAST_SPANNING_TREE    1
#define CMK_BROADCAST_HYPERCUBE        0
#endif

#define BROADCAST_SPANNING_FACTOR      4

#define CMI_BROADCAST_ROOT(msg)          ((CmiMsgHeaderBasic *)msg)->root
#define CMI_GET_CYCLE(msg)               ((CmiMsgHeaderBasic *)msg)->root

#define CMI_DEST_RANK(msg)               ((CmiMsgHeaderBasic *)msg)->rank
#define CMI_MAGIC(msg)                   ((CmiMsgHeaderBasic *)msg)->magic

/* FIXME: need a random number that everyone agrees ! */
#define CHARM_MAGIC_NUMBER               126

#if !CMK_OPTIMIZE
static int checksum_flag = 0;
#define CMI_SET_CHECKSUM(msg, len)      \
        if (checksum_flag)  {   \
          ((CmiMsgHeaderBasic *)msg)->cksum = 0;        \
          ((CmiMsgHeaderBasic *)msg)->cksum = computeCheckSum((unsigned char*)msg, len);        \
        }
#define CMI_CHECK_CHECKSUM(msg, len)    \
        if (checksum_flag)      \
          if (computeCheckSum((unsigned char*)msg, len) != 0)   \
            CmiAbort("Fatal error: checksum doesn't agree!\n");
#else
#define CMI_SET_CHECKSUM(msg, len)
#define CMI_CHECK_CHECKSUM(msg, len)
#endif

#if CMK_BROADCAST_SPANNING_TREE
#  define CMI_SET_BROADCAST_ROOT(msg, root)  CMI_BROADCAST_ROOT(msg) = (root);
#else
#  define CMI_SET_BROADCAST_ROOT(msg, root)
#endif

#if CMK_BROADCAST_HYPERCUBE
#  define CMI_SET_CYCLE(msg, cycle)  CMI_GET_CYCLE(msg) = (cycle);
#else
#  define CMI_SET_CYCLE(msg, cycle)
#endif


#ifdef MPI_POST_RECV
#define MPI_POST_RECV_COUNT 10
#undef MPI_POST_RECV
#endif
#if MPI_POST_RECV_COUNT > 0
#warning "Using MPI posted receives which have not yet been tested"
#ifndef MPI_POST_RECV_SIZE
#define MPI_POST_RECV_SIZE 200
#endif
/* #undef  MPI_POST_RECV_DEBUG  */
CpvDeclare(unsigned long long, Cmi_posted_recv_total);
CpvDeclare(unsigned long long, Cmi_unposted_recv_total);
CpvDeclare(MPI_Request*, CmiPostedRecvRequests); /* An array of request handles for posted recvs */
CpvDeclare(char*,CmiPostedRecvBuffers);
#endif

/*
 to avoid MPI's in order delivery, changing MPI Tag all the time
*/
#define TAG     1375

#if MPI_POST_RECV_COUNT > 0
#define POST_RECV_TAG TAG+1
#define BARRIER_ZERO_TAG TAG
#else
#define BARRIER_ZERO_TAG     1375
#endif

/*
static int mpi_tag = TAG;
#define NEW_MPI_TAG     mpi_tag++; if (mpi_tag == MPI_TAG_UB) mpi_tag=TAG;
*/

int               _Cmi_numpes;
int               _Cmi_mynode;    /* Which address space am I */
int               _Cmi_mynodesize;/* Number of processors in my address space */
int               _Cmi_numnodes;  /* Total number of address spaces */
int               _Cmi_numpes;    /* Total number of processors */
static int        Cmi_nodestart; /* First processor in this address space */
CpvDeclare(void*, CmiLocalQueue);

/*Network progress utility variables. Period controls the rate at
  which the network poll is called */
CpvDeclare(unsigned , networkProgressCount);
int networkProgressPeriod;

int               idleblock = 0;

#define BLK_LEN  512

#if CMK_NODE_QUEUE_AVAILABLE
#define DGRAM_NODEMESSAGE   (0xFB)

#define NODE_BROADCAST_OTHERS (-1)
#define NODE_BROADCAST_ALL    (-2)
#endif

#if 0
static void **recdQueue_blk;
static unsigned int recdQueue_blk_len;
static unsigned int recdQueue_first;
static unsigned int recdQueue_len;
static void recdQueueInit(void);
static void recdQueueAddToBack(void *element);
static void *recdQueueRemoveFromFront(void);
#endif

static void ConverseRunPE(int everReturn);
static void CommunicationServer(int sleepTime);
static void CommunicationServerThread(int sleepTime);

typedef struct msg_list {
     char *msg;
     struct msg_list *next;
     int size, destpe;
     MPI_Request req;
} SMSG_LIST;

int MsgQueueLen=0;
static int request_max;

static SMSG_LIST *sent_msgs=0;
static SMSG_LIST *end_sent=0;

static int Cmi_dim;

static int no_outstanding_sends=0; /*FLAG: consume outstanding Isends in scheduler loop*/

#if NODE_0_IS_CONVHOST
int inside_comm = 0;
#endif

void CmiAbort(const char *message);
static void PerrorExit(const char *msg);

void SendSpanningChildren(int size, char *msg);
void SendHypercube(int size, char *msg);

static void PerrorExit(const char *msg)
{
  perror(msg);
  exit(1);
}

extern unsigned char computeCheckSum(unsigned char *data, int len);

/**************************  TIMER FUNCTIONS **************************/

#if CMK_TIMER_USE_SPECIAL || CMK_TIMER_USE_XT3_DCLOCK

/* MPI calls are not threadsafe, even the timer on some machines */
static CmiNodeLock  timerLock = 0;
static double starttimer = 0;
static int _is_global = 0;

int CmiTimerIsSynchronized()
{
  int  flag;
  void *v;

  /*  check if it using synchronized timer */
  if (MPI_SUCCESS != MPI_Attr_get(MPI_COMM_WORLD, MPI_WTIME_IS_GLOBAL, &v, &flag))
    printf("MPI_WTIME_IS_GLOBAL not valid!\n");
  if (flag) {
    _is_global = *(int*)v;
    if (_is_global && CmiMyPe() == 0)
      printf("Charm++> MPI timer is synchronized!\n");
  }
  return _is_global;
}

void CmiTimerInit()
{
  _is_global = CmiTimerIsSynchronized();

  if (CmiMyRank() == 0) {
    if (_is_global) {
      double minTimer;
#if CMK_TIMER_USE_XT3_DCLOCK
      starttimer = dclock();
#else
      starttimer = MPI_Wtime();
#endif

      MPI_Allreduce(&starttimer, &minTimer, 1, MPI_DOUBLE, MPI_MIN,
                                  MPI_COMM_WORLD );
      starttimer = minTimer;
    }
    else {
      /* we don't have a synchronous timer, set our own start time */
      CmiBarrier();
      CmiBarrier();
      CmiBarrier();
#if CMK_TIMER_USE_XT3_DCLOCK
      starttimer = dclock();
#else
      starttimer = MPI_Wtime();
#endif
    }
    /*  timerLock = CmiCreateLock();  */
  }
  CmiNodeAllBarrier();          /* for smp */
}

double CmiTimer(void)
{
  double t;
#if CMK_SMP
  if (timerLock) CmiLock(timerLock);
#endif
#if CMK_TIMER_USE_XT3_DCLOCK
  t = dclock() - starttimer;
#else
  t = MPI_Wtime() - starttimer;
#endif
#if CMK_SMP
  if (timerLock) CmiUnlock(timerLock);
#endif
  return t;
}

double CmiWallTimer(void)
{
  double t;
#if CMK_SMP
  if (timerLock) CmiLock(timerLock);
#endif
#if CMK_TIMER_USE_XT3_DCLOCK
  t = dclock() - starttimer;
#else
  t = MPI_Wtime() - starttimer;
#endif
#if CMK_SMP
  if (timerLock) CmiUnlock(timerLock);
#endif
  return t;
}

double CmiCpuTimer(void)
{
  double t;
#if CMK_SMP
  if (timerLock) CmiLock(timerLock);
#endif
#if CMK_TIMER_USE_XT3_DCLOCK
  t = dclock() - starttimer;
#else
  t = MPI_Wtime() - starttimer;
#endif
#if CMK_SMP
  if (timerLock) CmiUnlock(timerLock);
#endif
  return t;
}

#endif

int CmiBarrier()
{
  if (CmiMyRank() == 0) {

    START_EVENT();

    if (MPI_SUCCESS != MPI_Barrier(MPI_COMM_WORLD))
        CmiAbort("Timernit: MPI_Barrier failed!\n");

    END_EVENT(10);
  }
  return 0;
}

/* CmiBarrierZero make sure node 0 is the last one exiting the barrier */
int CmiBarrierZero()
{
  int i;
  if (CmiMyRank() == 0) {
    char msg[1];
    MPI_Status sts;
    if (CmiMyNode() == 0)  {
      for (i=0; i<CmiNumNodes()-1; i++) {
         START_EVENT();

         if (MPI_SUCCESS != MPI_Recv(msg,1,MPI_BYTE,MPI_ANY_SOURCE,BARRIER_ZERO_TAG, MPI_COMM_WORLD,&sts))
            CmiPrintf("MPI_Recv failed!\n");

         END_EVENT(30);
      }
    }
    else {
      START_EVENT();

      if (MPI_SUCCESS != MPI_Send((void *)msg,1,MPI_BYTE,0,BARRIER_ZERO_TAG,MPI_COMM_WORLD))
         printf("MPI_Send failed!\n");

      END_EVENT(20);
    }
  }
  CmiNodeAllBarrier();
  return 0;
}

typedef struct ProcState {
#if MULTI_SENDQUEUE
PCQueue      sendMsgBuf;       /* per processor message sending queue */
#endif
CmiNodeLock  recvLock;              /* for cs->recv */
} ProcState;

static ProcState  *procState;

#if CMK_SMP

#if !MULTI_SENDQUEUE
static PCQueue sendMsgBuf;
static CmiNodeLock  sendMsgBufLock = NULL;        /* for sendMsgBuf */
#endif

#endif

/************************************************************
 *
 * Processor state structure
 *
 ************************************************************/

/* fake Cmi_charmrun_fd */
static int Cmi_charmrun_fd = 0;
#include "machine-smp.c"

CsvDeclare(CmiNodeState, NodeState);

#include "immediate.c"

#if ! CMK_SMP
/************ non SMP **************/
static struct CmiStateStruct Cmi_state;
int _Cmi_mype;
int _Cmi_myrank;

void CmiMemLock() {}
void CmiMemUnlock() {}

#define CmiGetState() (&Cmi_state)
#define CmiGetStateN(n) (&Cmi_state)

void CmiYield(void) { sleep(0); }

static void CmiStartThreads(char **argv)
{
  CmiStateInit(Cmi_nodestart, 0, &Cmi_state);
  _Cmi_mype = Cmi_nodestart;
  _Cmi_myrank = 0;
}
#endif  /* non smp */

/*Add a message to this processor's receive queue, pe is a rank */
static void CmiPushPE(int pe,void *msg)
{
  CmiState cs = CmiGetStateN(pe);
  MACHSTATE2(3,"Pushing message into rank %d's queue %p{",pe, cs->recv);
#if CMK_IMMEDIATE_MSG
  if (CmiIsImmediate(msg)) {
/*
CmiPrintf("[node %d] Immediate Message hdl: %d rank: %d {{. \n", CmiMyNode(), CmiGetHandler(msg), pe);
    CmiHandleMessage(msg);
CmiPrintf("[node %d] Immediate Message done.}} \n", CmiMyNode());
*/
    CMI_DEST_RANK(msg) = pe;
    CmiPushImmediateMsg(msg);
    return;
  }
#endif

#if CMK_SMP
  CmiLock(procState[pe].recvLock);
#endif
  PCQueuePush(cs->recv,msg);
#if CMK_SMP
  CmiUnlock(procState[pe].recvLock);
#endif
  CmiIdleLock_addMessage(&cs->idle);
  MACHSTATE1(3,"} Pushing message into rank %d's queue done",pe);
}

#if CMK_NODE_QUEUE_AVAILABLE
/*Add a message to this processor's receive queue */
static void CmiPushNode(void *msg)
{
  MACHSTATE(3,"Pushing message into NodeRecv queue");
#if CMK_IMMEDIATE_MSG
  if (CmiIsImmediate(msg)) {
    CMI_DEST_RANK(msg) = 0;
    CmiPushImmediateMsg(msg);
    return;
  }
#endif
  CmiLock(CsvAccess(NodeState).CmiNodeRecvLock);
  PCQueuePush(CsvAccess(NodeState).NodeRecv,msg);
  CmiUnlock(CsvAccess(NodeState).CmiNodeRecvLock);
  {
  CmiState cs=CmiGetStateN(0);
  CmiIdleLock_addMessage(&cs->idle);
  }
}
#endif

#ifndef CmiMyPe
int CmiMyPe(void)
{
  return CmiGetState()->pe;
}
#endif

#ifndef CmiMyRank
int CmiMyRank(void)
{
  return CmiGetState()->rank;
}
#endif

#ifndef CmiNodeFirst
int CmiNodeFirst(int node) { return node*_Cmi_mynodesize; }
int CmiNodeSize(int node)  { return _Cmi_mynodesize; }
#endif

#ifndef CmiNodeOf
int CmiNodeOf(int pe)      { return (pe/_Cmi_mynodesize); }
int CmiRankOf(int pe)      { return pe%_Cmi_mynodesize; }
#endif

static size_t CmiAllAsyncMsgsSent(void)
{
   SMSG_LIST *msg_tmp = sent_msgs;
   MPI_Status sts;
   int done;

   while(msg_tmp!=0) {
    done = 0;
    if (MPI_SUCCESS != MPI_Test(&(msg_tmp->req), &done, &sts))
      CmiAbort("CmiAllAsyncMsgsSent: MPI_Test failed!\n");
    if(!done)
      return 0;
    msg_tmp = msg_tmp->next;
/*    MsgQueueLen--; ????? */
   }
   return 1;
}

int CmiAsyncMsgSent(CmiCommHandle c) {

  SMSG_LIST *msg_tmp = sent_msgs;
  int done;
  MPI_Status sts;

  while ((msg_tmp) && ((CmiCommHandle)&(msg_tmp->req) != c))
    msg_tmp = msg_tmp->next;
  if(msg_tmp) {
    done = 0;
    if (MPI_SUCCESS != MPI_Test(&(msg_tmp->req), &done, &sts))
      CmiAbort("CmiAsyncMsgSent: MPI_Test failed!\n");
    return ((done)?1:0);
  } else {
    return 1;
  }
}

void CmiReleaseCommHandle(CmiCommHandle c)
{
  return;
}

#if CMK_VERSION_BLUEGENE
extern void MPID_Progress_test();
#endif

void CmiReleaseSentMessages(void)
{
  SMSG_LIST *msg_tmp=sent_msgs;
  SMSG_LIST *prev=0;
  SMSG_LIST *temp;
  int done;
  MPI_Status sts;


#if CMK_VERSION_BLUEGENE
  MPID_Progress_test();
#endif

  MACHSTATE1(2,"CmiReleaseSentMessages begin on %d {", CmiMyPe());
  while(msg_tmp!=0) {
    done =0;
    if(MPI_Test(&(msg_tmp->req), &done, &sts) != MPI_SUCCESS)
      CmiAbort("CmiReleaseSentMessages: MPI_Test failed!\n");
    if(done) {
      MACHSTATE2(3,"CmiReleaseSentMessages release one %d to %d", CmiMyPe(), msg_tmp->destpe);
      MsgQueueLen--;
      /* Release the message */
      temp = msg_tmp->next;
      if(prev==0)  /* first message */
        sent_msgs = temp;
      else
        prev->next = temp;
      CmiFree(msg_tmp->msg);
      CmiFree(msg_tmp);
      msg_tmp = temp;
    } else {
      prev = msg_tmp;
      msg_tmp = msg_tmp->next;
    }
  }
  end_sent = prev;
  MACHSTATE(2,"} CmiReleaseSentMessages end");
}

int PumpMsgs(void)
{
  int nbytes, flg, res;
  char *msg;
  MPI_Status sts;
  int recd=0;

#if CMK_VERSION_BLUEGENE
  MPID_Progress_test();
#endif

  MACHSTATE(2,"PumpMsgs begin {");

  while(1) {
    /* First check posted recvs then do  probe unmatched outstanding messages */
#if MPI_POST_RECV_COUNT > 0 
    int completed_index=-1;
    if(MPI_SUCCESS != MPI_Testany(MPI_POST_RECV_COUNT, CpvAccess(CmiPostedRecvRequests), &completed_index, &flg, &sts))
        CmiAbort("PumpMsgs: MPI_Testany failed!\n");
    if(flg){
        if (MPI_SUCCESS != MPI_Get_count(&sts, MPI_BYTE, &nbytes))
            CmiAbort("PumpMsgs: MPI_Get_count failed!\n");

        recd = 1;
        msg = (char *) CmiAlloc(nbytes);
        memcpy(msg,&(CpvAccess(CmiPostedRecvBuffers)[completed_index*MPI_POST_RECV_SIZE]),nbytes);
        /* and repost the recv */

        START_EVENT();

        if (MPI_SUCCESS != MPI_Irecv(  &(CpvAccess(CmiPostedRecvBuffers)[completed_index*MPI_POST_RECV_SIZE])   ,
            MPI_POST_RECV_SIZE,
            MPI_BYTE,
            MPI_ANY_SOURCE,
            POST_RECV_TAG,
            MPI_COMM_WORLD,
            &(CpvAccess(CmiPostedRecvRequests)[completed_index])  ))
                CmiAbort("PumpMsgs: MPI_Irecv failed!\n");

        END_EVENT(50);

        CpvAccess(Cmi_posted_recv_total)++;
    }
    else {
        res = MPI_Iprobe(MPI_ANY_SOURCE, TAG, MPI_COMM_WORLD, &flg, &sts);
        if(res != MPI_SUCCESS)
        CmiAbort("MPI_Iprobe failed\n");
        if(!flg) break;
        recd = 1;
        MPI_Get_count(&sts, MPI_BYTE, &nbytes);
        msg = (char *) CmiAlloc(nbytes);

        START_EVENT();

        if (MPI_SUCCESS != MPI_Recv(msg,nbytes,MPI_BYTE,sts.MPI_SOURCE,sts.MPI_TAG, MPI_COMM_WORLD,&sts))
            CmiAbort("PumpMsgs: MPI_Recv failed!\n");

        END_EVENT(30);

        CpvAccess(Cmi_unposted_recv_total)++;
    }
#else
    /* Original version */
    res = MPI_Iprobe(MPI_ANY_SOURCE, TAG, MPI_COMM_WORLD, &flg, &sts);
    if(res != MPI_SUCCESS)
      CmiAbort("MPI_Iprobe failed\n");

    if(!flg) break;
    recd = 1;
    MPI_Get_count(&sts, MPI_BYTE, &nbytes);
    msg = (char *) CmiAlloc(nbytes);

    START_EVENT();

    if (MPI_SUCCESS != MPI_Recv(msg,nbytes,MPI_BYTE,sts.MPI_SOURCE,sts.MPI_TAG, MPI_COMM_WORLD,&sts))
      CmiAbort("PumpMsgs: MPI_Recv failed!\n");

    END_EVENT(30);
#endif

    MACHSTATE2(3,"PumpMsgs recv one from node:%d to rank:%d", sts.MPI_SOURCE, CMI_DEST_RANK(msg));
    CMI_CHECK_CHECKSUM(msg, nbytes);
    if (CMI_MAGIC(msg) != CHARM_MAGIC_NUMBER) { /* received a non-charm msg */
      CmiPrintf("Charm++ Abort: Non Charm++ Message Received of size %d. \n", nbytes);
      CmiFree(msg);
      CmiAbort("Abort!\n");
      continue;
    }
#if CMK_NODE_QUEUE_AVAILABLE
    if (CMI_DEST_RANK(msg)==DGRAM_NODEMESSAGE)
      CmiPushNode(msg);
    else
#endif
      CmiPushPE(CMI_DEST_RANK(msg), msg);

#if CMK_BROADCAST_SPANNING_TREE
    if (CMI_BROADCAST_ROOT(msg))
      SendSpanningChildren(nbytes, msg);
#elif CMK_BROADCAST_HYPERCUBE
    if (CMI_GET_CYCLE(msg))
      SendHypercube(nbytes, msg);
#endif
  }
#if CMK_IMMEDIATE_MSG && !CMK_SMP
  CmiHandleImmediate();
#endif
  MACHSTATE(2,"} PumpMsgs end ");
  return recd;
}

/* blocking version */
static void PumpMsgsBlocking(void)
{
  static int maxbytes = 20000000;
  static char *buf = NULL;
  int nbytes, flg;
  MPI_Status sts;
  char *msg;
  int recd=0;

  if (!PCQueueEmpty(CmiGetState()->recv)) return;
  if (!CdsFifo_Empty(CpvAccess(CmiLocalQueue))) return;
  if (!CqsEmpty(CpvAccess(CsdSchedQueue))) return;
  if (sent_msgs)  return;

#if 0
  CmiPrintf("[%d] PumpMsgsBlocking. \n", CmiMyPe());
#endif

  if (buf == NULL) {
    buf = (char *) CmiAlloc(maxbytes);
    _MEMCHECK(buf);
  }


#if MPI_POST_RECV_COUNT > 0
#warning "Using MPI posted receives and PumpMsgsBlocking() will break"
CmiAbort("Unsupported use of PumpMsgsBlocking. This call should be extended to check posted recvs, cancel them all, and then wait on any incoming message, and then re-post the recvs");
#endif

  START_EVENT();

  if (MPI_SUCCESS != MPI_Recv(buf,maxbytes,MPI_BYTE,MPI_ANY_SOURCE,TAG, MPI_COMM_WORLD,&sts))
      CmiAbort("PumpMsgs: PMP_Recv failed!\n");

  END_EVENT(30);

   MPI_Get_count(&sts, MPI_BYTE, &nbytes);
   msg = (char *) CmiAlloc(nbytes);
   memcpy(msg, buf, nbytes);

#if CMK_NODE_QUEUE_AVAILABLE
   if (CMI_DEST_RANK(msg)==DGRAM_NODEMESSAGE)
      CmiPushNode(msg);
   else
#endif
      CmiPushPE(CMI_DEST_RANK(msg), msg);

#if CMK_BROADCAST_SPANNING_TREE
   if (CMI_BROADCAST_ROOT(msg))
      SendSpanningChildren(nbytes, msg);
#elif CMK_BROADCAST_HYPERCUBE
   if (CMI_GET_CYCLE(msg))
      SendHypercube(nbytes, msg);
#endif
}

/********************* MESSAGE RECEIVE FUNCTIONS ******************/

#if CMK_SMP

static int inexit = 0;
static CmiNodeLock  exitLock = 0;

static int MsgQueueEmpty()
{
  int i;
#if MULTI_SENDQUEUE
  for (i=0; i<_Cmi_mynodesize; i++)
    if (!PCQueueEmpty(procState[i].sendMsgBuf)) return 0;
#else
  return PCQueueEmpty(sendMsgBuf);
#endif
  return 1;
}

static int SendMsgBuf();

/* test if all processors recv queues are empty */
static int RecvQueueEmpty()
{
  int i;
  for (i=0; i<_Cmi_mynodesize; i++) {
    CmiState cs=CmiGetStateN(i);
    if (!PCQueueEmpty(cs->recv)) return 0;
  }
  return 1;
}

static void CommunicationServer(int sleepTime)
{
  int static count=0;
/*
  count ++;
  if (count % 10000000==0) MACHSTATE(3, "Entering CommunicationServer {");
*/
  PumpMsgs();
  CmiReleaseSentMessages();
  SendMsgBuf();
/*
  if (count % 10000000==0) MACHSTATE(3, "} Exiting CommunicationServer.");
*/
  if (inexit == CmiMyNodeSize()) {
    MACHSTATE(2, "CommunicationServer exiting {");
#if 0
    while(!MsgQueueEmpty() || !CmiAllAsyncMsgsSent() || !RecvQueueEmpty()) {
#endif
    while(!MsgQueueEmpty() || !CmiAllAsyncMsgsSent()) {
      CmiReleaseSentMessages();
      SendMsgBuf();
      PumpMsgs();
    }
    MACHSTATE(2, "CommunicationServer barrier begin {");

    START_EVENT();

    if (MPI_SUCCESS != MPI_Barrier(MPI_COMM_WORLD))
      CmiAbort("ConverseExit: MPI_Barrier failed!\n");

    END_EVENT(10);

    MACHSTATE(2, "} CommunicationServer barrier end");
#if (CMK_DEBUG_MODE || CMK_WEB_MODE || NODE_0_IS_CONVHOST)
    if (CmiMyNode() == 0){
      CmiPrintf("End of program\n");
    }
#endif
    MACHSTATE(2, "} CommunicationServer EXIT");
#if ! CMK_MPI_VMI
    MPI_Finalize();
#endif
    exit(0);
  }
}

#endif

static void CommunicationServerThread(int sleepTime)
{
#if CMK_SMP
  CommunicationServer(sleepTime);
#endif
#if CMK_IMMEDIATE_MSG
  CmiHandleImmediate();
#endif
}

#if CMK_NODE_QUEUE_AVAILABLE
char *CmiGetNonLocalNodeQ(void)
{
  CmiState cs = CmiGetState();
  char *result = 0;
  CmiIdleLock_checkMessage(&cs->idle);
/*  if(!PCQueueEmpty(CsvAccess(NodeState).NodeRecv)) {  */
    MACHSTATE1(3,"CmiGetNonLocalNodeQ begin %d {", CmiMyPe());
    CmiLock(CsvAccess(NodeState).CmiNodeRecvLock);
    result = (char *) PCQueuePop(CsvAccess(NodeState).NodeRecv);
    CmiUnlock(CsvAccess(NodeState).CmiNodeRecvLock);
    MACHSTATE1(3,"} CmiGetNonLocalNodeQ end %d ", CmiMyPe());
/*  }  */
  return result;
}
#endif

void *CmiGetNonLocal(void)
{
  static int count=0;
  CmiState cs = CmiGetState();
  void *msg;
  CmiIdleLock_checkMessage(&cs->idle);
  /* although it seems that lock is not needed, I found it crashes very often
     on mpi-smp without lock */

#if ! CMK_SMP
  CmiReleaseSentMessages();
  PumpMsgs();
#endif

  /* CmiLock(procState[cs->rank].recvLock); */
  msg =  PCQueuePop(cs->recv);
  /* CmiUnlock(procState[cs->rank].recvLock); */

/*
  if (msg) {
    MACHSTATE2(3,"CmiGetNonLocal done on pe %d for queue %p", CmiMyPe(), cs->recv); }
  else {
    count++;
    if (count%1000000==0) MACHSTATE2(3,"CmiGetNonLocal empty on pe %d for queue %p", CmiMyPe(), cs->recv);
  }
*/
#if ! CMK_SMP
  if (no_outstanding_sends) {
    while (MsgQueueLen>0) {
      CmiReleaseSentMessages();
      PumpMsgs();
    }
  }

  if(!msg) {
    CmiReleaseSentMessages();
    if (PumpMsgs())
      return  PCQueuePop(cs->recv);
    else
      return 0;
  }
#endif
  return msg;
}

/* called in non-smp mode */
void CmiNotifyIdle(void)
{
  CmiReleaseSentMessages();
  if (!PumpMsgs() && idleblock) PumpMsgsBlocking();
}


/********************************************************
    The call to probe immediate messages has been renamed to
    CmiMachineProgressImpl
******************************************************/
/* user call to handle immediate message, only useful in non SMP version
   using polling method to schedule message.
*/
/*
#if CMK_IMMEDIATE_MSG
void CmiProbeImmediateMsg()
{
#if !CMK_SMP
  PumpMsgs();
  CmiHandleImmediate();
#endif
}
#endif
*/

/* Network progress function is used to poll the network when for
   messages. This flushes receive buffers on some  implementations*/
#if CMK_MACHINE_PROGRESS_DEFINED
void CmiMachineProgressImpl()
{
#if !CMK_SMP
    PumpMsgs();
#if CMK_IMMEDIATE_MSG
    CmiHandleImmediate();
#endif
#else
    /*Not implemented yet. Communication server does not seem to be
      thread safe */
    /* CommunicationServerThread(0); */
#endif
}
#endif

/********************* MESSAGE SEND FUNCTIONS ******************/

CmiCommHandle CmiAsyncSendFn_(int destPE, int size, char *msg);

static void CmiSendSelf(char *msg)
{
#if CMK_IMMEDIATE_MSG
    if (CmiIsImmediate(msg)) {
      /* CmiBecomeNonImmediate(msg); */
      CmiPushImmediateMsg(msg);
      CmiHandleImmediate();
      return;
    }
#endif
    CQdCreate(CpvAccess(cQdState), 1);
    CdsFifo_Enqueue(CpvAccess(CmiLocalQueue),msg);
}

void CmiSyncSendFn(int destPE, int size, char *msg)
{
  CmiState cs = CmiGetState();
  char *dupmsg = (char *) CmiAlloc(size);
  memcpy(dupmsg, msg, size);

  CMI_SET_BROADCAST_ROOT(dupmsg, 0);

  if (cs->pe==destPE) {
    CmiSendSelf(dupmsg);
  }
  else
    CmiAsyncSendFn_(destPE, size, dupmsg);
}

#if CMK_SMP

/* called by communication thread in SMP */
static int SendMsgBuf()
{
  SMSG_LIST *msg_tmp;
  char *msg;
  int node, rank, size;
  int i;
  int sent = 0;

  MACHSTATE(2,"SendMsgBuf begin {");
#if MULTI_SENDQUEUE
  for (i=0; i<_Cmi_mynodesize=1; i++)  /* subtle: including comm thread */
  {
    if (!PCQueueEmpty(procState[i].sendMsgBuf))
    {
      msg_tmp = (SMSG_LIST *)PCQueuePop(procState[i].sendMsgBuf);
#else
    /* single message sending queue */
    /* CmiLock(sendMsgBufLock); */
    msg_tmp = (SMSG_LIST *)PCQueuePop(sendMsgBuf);
    /* CmiUnlock(sendMsgBufLock); */
    while (NULL != msg_tmp)
    {
#endif
      node = msg_tmp->destpe;
      size = msg_tmp->size;
      msg = msg_tmp->msg;
      msg_tmp->next = 0;
      while (MsgQueueLen > request_max) {
        CmiReleaseSentMessages();
        PumpMsgs();
      }
      MACHSTATE2(3,"MPI_send to node %d rank: %d{", node, CMI_DEST_RANK(msg));
      CMI_MAGIC(msg) = CHARM_MAGIC_NUMBER;
      CMI_SET_CHECKSUM(msg, size);

#if MPI_POST_RECV_COUNT > 0
        if(size <= MPI_POST_RECV_SIZE){

          START_EVENT();
          if (MPI_SUCCESS != MPI_Isend((void *)msg,size,MPI_BYTE,node,POST_RECV_TAG,MPI_COMM_WORLD,&(msg_tmp->req)))
                CmiAbort("CmiAsyncSendFn: MPI_Isend failed!\n");

          STOP_EVENT(40);
        }
        else {
            START_EVENT();
            if (MPI_SUCCESS != MPI_Isend((void *)msg,size,MPI_BYTE,node,TAG,MPI_COMM_WORLD,&(msg_tmp->req)))
                CmiAbort("CmiAsyncSendFn: MPI_Isend failed!\n");
            STOP_EVENT(40);
        }
#else