Home Research Papers Posters Manuals Talks Download People Help Wiki Internal

arch/elan/machine.c

Go to the documentation of this file.

/* Charm++ Machine Layer for ELAN network interface 
Developed by Sameer Kumar
*/

#include <stdio.h>
#include <sys/time.h>
#include <assert.h>
#include <errno.h>
#include "converse.h"
#include <elan/elan.h>
/*#include <elan3/elan3.h>*/

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

#include "machine.h"
#include "pcqueue.h"

#if CMK_PERSISTENT_COMM
#include "persist_impl.h"
#endif

/* copy from elan/version.h */
#ifndef QSNETLIBS_VERSION_CODE
#define QSNETLIBS_VERSION(a,b,c)        (((a) << 16) + ((b) << 8) + (c))
#define QSNETLIBS_VERSION_CODE          QSNETLIBS_VERSION(1,3,0)
#endif

#define MAX_QLEN 1000
#define MAX_BYTES 1000000

#define USE_SHM 0

/*
  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.
*/
#define CMK_BROADCAST_SPANNING_TREE    1
#define BROADCAST_SPANNING_FACTOR      4

#define CMI_BROADCAST_ROOT(msg)          ((CmiMsgHeaderBasic *)msg)->root
#define CMI_DEST_RANK(msg)               ((CmiMsgHeaderBasic *)msg)->rank
#define CMI_MESSAGE_SIZE(msg)            ((CmiMsgHeaderBasic *)msg)->size

#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

ELAN_BASE     *elan_base;
ELAN_TPORT    *elan_port;
ELAN_QUEUE    *elan_q;

int enableGetBasedSend = 1;
int enableBufferPooling = 0;

int SMALL_MESSAGE_SIZE=4080;  /* Smallest message size queue 
                                 used for receiving short messages */
                                     
int MID_MESSAGE_SIZE=65536;     /* Queue for larger messages 
                                   which need pre posted receives
                                   Message sizes greater will be 
                                   probe received adding 5us overhead*/
#define SYNC_MESSAGE_SIZE MID_MESSAGE_SIZE * 10
                               /* Message sizes greater will be 
                                  sent synchronously thus avoiding copying*/

#define NON_BLOCKING_MSG  4     /* Message sizes greater 
                                    than this will be sent asynchronously*/
#define RECV_MSG_Q_SIZE  8   //Maximim queue size for short messages
#define MID_MSG_Q_SIZE   4   //Maximum queue size for mid-range messages

//Actual sizes, can also be set from the command line
int smallQSize = RECV_MSG_Q_SIZE;
int midQSize = MID_MSG_Q_SIZE;

ELAN_EVENT *esmall[RECV_MSG_Q_SIZE], *emid[MID_MSG_Q_SIZE], *elarge;

#define TAG_SMALL 0x1
#define TAG_LARGE_HEADER 0x3     /* Header that a large message is coming*/
#define TAG_GET_BASED_SEND 0x5     /* Header that a large message is coming
                                    as a get request*/
#define TAG_MID   0x10
#define TAG_LARGE 0x100

/*Release sent messages status to check for*/
#define BASIC_SEND 0
#define GET_BASED_SEND 1
#define RECEIVE_GET   2
#define GET_FINISHED_RECEIVE 3

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

#define BLK_LEN  512

static int MsgQueueLen=0;
static int MsgQueueBytes=0;
static int request_max;
static int request_bytes;

#include "pcqueue.h"
PCQueue localSmallBufferQueue;
PCQueue localMidBufferQueue;

int outstandingMsgs[3000];

int stretchFlag = 0;
int blockingReceiveFlag = 0;

static void ConverseRunPE(int everReturn);

typedef struct {
    char header[CmiMsgHeaderSizeBytes];
    int size;
    char* src_addr;
    char* flag_addr;
} GetHeader;

typedef struct msg_list {
    ELAN_EVENT *e;
    char *msg;
    struct msg_list *next;
    int size, destpe;
    int sent;
    
    //Fields for get based send
    int status;
    long done;
    long *flag_addr;
    char *gmsg;
    char *newmsg;       
    int is_broadcast;
} SMSG_LIST;


static int Cmi_dim;

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

void ElanSendQueuedMessages();
static int CmiReleaseSentMessages();

void ElanGetBasedSend(SMSG_LIST *ptr);
void handleGetHeader(char *msg, int src);
void processGetEnv(SMSG_LIST *ptr);

#if NODE_0_IS_CONVHOST
int inside_comm = 0;
#endif

double starttimer;

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

void SendSpanningChildren(int size, char *msg);

typedef struct __elanChunkHeader {
  int type;
  int size;
} ElanChunkHeader;

typedef struct __chunkHeader {
  ElanChunkHeader elan;
  CmiChunkHeader conv;
} ChunkHeader;

#define TYPE_FIELD(buf)       (((ChunkHeader*)(buf))->elan.type)
#define SIZE_FIELD(buf)       (((ChunkHeader*)(buf))->elan.size)
#define CONV_SIZE_FIELD(buf)  (((ChunkHeader*)(buf))->conv.size)
#define REF_FIELD(buf)        (((ChunkHeader*)(buf))->conv.ref)

// CONV_BUF_START moves the res from pointing to the start of the elan chunk to the start of the converse chunk
#define CONV_BUF_START(res)     ((char*)(res) + sizeof(ElanChunkHeader))

// MACHINE_BUF_START moves the res from pointing to the start of the converse chunk to the start of the elan chunk
#define MACHINE_BUF_START(res)  ((char*)(res) - sizeof(ElanChunkHeader))

// USER_BUF_START moves the res from pointing to the start of the payload to the start of the elan chunk
#define USER_BUF_START(res)     ((char*)(res) - sizeof(ChunkHeader))

#define DYNAMIC_MESSAGE 0
#define STATIC_MESSAGE 1
#define ELAN_MESSAGE 3

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

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

#if CMK_TIMER_USE_SPECIAL

#include <sys/timers.h>
void CmiTimerInit(void)
{
    starttimer =  elan_clock(elan_base->state); 
}

double CmiTimer(void)
{
    return (elan_clock(elan_base->state) - starttimer)/1e9;
}

double CmiWallTimer(void)
{
    return (elan_clock(elan_base->state) - starttimer)/1e9;
}

double CmiCpuTimer(void)
{
    return (elan_clock(elan_base->state) - starttimer)/1e9;
}

#endif

static PCQueue   msgBuf;

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

/*****
      SMP version Extend later, currently only NON SMP version 
***************/

#include "machine-smp.c"

CsvDeclare(CmiNodeState, NodeState);

static struct CmiStateStruct Cmi_state;
int _Cmi_mype;
int _Cmi_myrank;

#include "immediate.c"

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

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

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

/*Add a message to this processor's receive queue */
static void CmiPushPE(int pe,void *msg)
{
  CmiState cs=CmiGetStateN(pe);
  MACHSTATE1(2,"Pushing message into %d's queue",pe);
#if CMK_IMMEDIATE_MSG
  if (CmiIsImmediate(msg)) {
      /*CmiPrintf("[node %d] Immediate Message %d %d {{. \n", CmiMyNode(), CmiGetHandler(msg), _ImmediateMsgHandlerIdx);*/
      /*CmiHandleMessage(msg);*/
      CmiPushImmediateMsg(msg);
      /*CmiPrintf("[node %d] Immediate Message done.}} \n", CmiMyNode());*/
      return;
  }
#endif
  CmiIdleLock_addMessage(&cs->idle); 
  PCQueuePush(cs->recv,msg);
}

#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

int CmiAllAsyncMsgsSent(void)
{
   SMSG_LIST *msg_tmp = NULL; 

   int done;
   
   CmiReleaseSentMessages();

   msg_tmp = sent_msgs;
   while((msg_tmp != cur_unsent) && (msg_tmp->e != NULL)){
       done = 0;
    
       if(elan_tportTxDone(msg_tmp->e))
           done = 1;
       else
#if USE_SHM 
           elan_deviceCheck(elan_base->state);
#else 
       ;
#endif

       if(!done)
           return 0;
       msg_tmp = msg_tmp->next;
       //    MsgQueueLen--;
   }
   return 1;
}

int CmiAsyncMsgSent(CmiCommHandle c) {
     
  SMSG_LIST *msg_tmp = sent_msgs;
  int done;

  while ((msg_tmp) && (msg_tmp ->e != NULL) && 
         ((CmiCommHandle)(msg_tmp->e) != c))
      msg_tmp = msg_tmp->next;
  
  if(msg_tmp) {
    done = 0;
    
    if(elan_tportTxDone(msg_tmp->e))
        done = 1;
    else 
#if USE_SHM 
           elan_deviceCheck(elan_base->state);
#else 
       ;
#endif
    
    return ((done)?1:0);
  } else {
      return 1;
  }
}

void CmiReleaseCommHandle(CmiCommHandle c)
{
  return;
}

void release_pmsg_list();

#define MAX_RELEASE_POLL 4096

static int CmiReleaseSentMessages(void)
{
    SMSG_LIST *msg_tmp=sent_msgs;
    SMSG_LIST *prev=0;
    SMSG_LIST *temp;
    int done;
    int locked = 0;
    
#ifndef CMK_OPTIMIZE 
    double rel_start_time = CmiWallTimer();
#endif

#if CMK_PERSISTENT_COMM
  release_pmsg_list();
#endif

  int ncheck = MAX_RELEASE_POLL;

  while(msg_tmp != NULL && ncheck > 0){
      if(msg_tmp->sent) {
          done =0;

          if(msg_tmp->status == BASIC_SEND) {
              if(elan_tportTxDone(msg_tmp->e)) {
                  elan_tportTxWait(msg_tmp->e);
                  done = 1;
              }          
              else 
#if USE_SHM 
                  elan_deviceCheck(elan_base->state);
#else 
              ;
#endif
          }
          else {
              processGetEnv(msg_tmp);
              done = msg_tmp->done;
          }
          
          if(done) {
              MsgQueueLen--;
              MsgQueueBytes -= msg_tmp->size;
              
              outstandingMsgs[msg_tmp->destpe] = 0;
              
              /* 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;
              ncheck --;
          }
      }
      else {
          prev = msg_tmp;
          msg_tmp = msg_tmp->next;
      }
  }
  
  //if(msg_tmp)
  //  elan_deviceCheck(elan_base->state);

  end_sent = prev;

#if CMK_PERSISTENT_COMM
  release_pmsg_list();
#endif

#ifndef CMK_OPTIMIZE 
#if ! CMK_TRACE_IN_CHARM
  {
  double rel_end_time = CmiWallTimer();
  if(rel_end_time > rel_start_time + 50/1e6)
      traceUserBracketEvent(20, rel_start_time, rel_end_time);
  }
#endif
#endif

  //So messages not finished sending
  if(msg_tmp != cur_unsent)
      return 0;
  else
      //All messages sent
      return 1;
}

/* retflag = 0, receive as many messages as can be 
   and then post another receive
   retflag = 1, receive the first message and return 
   retflag =3 blocking receives

   Pump Msgs posts a circular queue of receives. The main idea 
   is that if a large number of receives are being posted only 
   minimum receive events should be polled. 

   So there are two indices, event_idx and post_idx. event_idx points
   to the first posted receive. So if a large number of messages come
   for a particular tag then this position in the queue will receive
   the first message. The new receives should be posted from
   post_idx. Notice that event_idx and post_idx are not the same. If a
   large number of messages are received then receives should be
   posted from the position of where first message was received.

*/
int PumpMsgs(int retflag)
{

    static char recv_small_done[RECV_MSG_Q_SIZE]; /*A list of flags which 
                                                    tells if a particular 
                                                    slot in the queue has 
                                                    a receive posted or not. */
    static char recv_mid_done[MID_MSG_Q_SIZE];  
    static int recv_large_done = 0;
    
    static char *sbuf[RECV_MSG_Q_SIZE];    /* Buffer of pointer to the 
                                              messages received */
    static char *mbuf[MID_MSG_Q_SIZE];
    static char *lbuf;
    
    static int event_idx = 0;             /* As defined earlier */
    static int post_idx = 0;

    static int event_m_idx = 0;
    static int post_m_idx = 0;

    static int nlarge_torecv = 0; /*this variable specifies how many
                                    large messages need to be received
                                    before we can block again.*/

    static int step1 = 0;   /* Large message are received in two
                               steps, first the envelope is probed by
                               posting a receive and then memory is
                               allocated for the message and the
                               message is finally received */

    int flg, res, rcount, mcount;
    char *msg = 0;
    
    int recd=0;
#if QSNETLIBS_VERSION_CODE > QSNETLIBS_VERSION(1,4,0)
    unsigned long size= 0;
#else
    int size= 0;
#endif
    int tag=0;
    int src=-1;
    
#ifndef CMK_OPTIMIZE 
    double pmp_start_time = CmiWallTimer();
#endif

    int ecount = 0, emcount = 0;

#if CMK_PERSISTENT_COMM
    if (PumpPersistent()) return 1;
#endif
        
    while(1) {
        msg = 0;
        
        ecount = 0;
        for(rcount = 0; rcount < smallQSize; rcount ++){
            ecount = (rcount + post_idx) % smallQSize;
            if(!recv_small_done[ecount]) {
                sbuf[ecount] = (char *) CmiAlloc(SMALL_MESSAGE_SIZE);
                
                esmall[ecount] = elan_tportRxStart(elan_port, 0, 0, 0, 1, 
                                                   TAG_SMALL, sbuf[ecount], 
                                                   SMALL_MESSAGE_SIZE);
                recv_small_done[ecount] = 1;
            }
            else {
                ecount = (ecount + smallQSize - 1) % smallQSize;
                break;
            }
        }
        post_idx = ecount + 1;

        emcount = 0;
        for(mcount = 0; mcount < midQSize; mcount ++){
            emcount = (mcount + post_m_idx) % midQSize;
            if(!recv_mid_done[emcount]) {
                mbuf[emcount] = (char *) CmiAlloc(MID_MESSAGE_SIZE);
                                                
                emid[emcount] = elan_tportRxStart(elan_port, 0, 0, 0, -1, 
                                                  TAG_MID, mbuf[emcount], 
                                                  MID_MESSAGE_SIZE);
                recv_mid_done[emcount] = 1;
            }
            else {
                emcount = (emcount + midQSize - 1) % midQSize;
                break;
            }
        }
        post_m_idx = emcount + 1;        
        
        if(!recv_large_done) {
            elarge = elan_tportRxStart(elan_port, ELAN_TPORT_RXPROBE, 0, 0, 
                                       -1, TAG_LARGE, NULL, 0);
            recv_large_done = 1;
        }
    
        if(!step1 && elan_tportRxDone(elarge)) {
            elan_tportRxWait(elarge, NULL, NULL, &size );
      
            lbuf = (char *) CmiAlloc(size);
            elarge = elan_tportRxStart(elan_port, 0, 0, 0, -1, TAG_LARGE, 
                                       lbuf,size);
            step1 = 1;
        }
        
        if(step1 && elan_tportRxDone(elarge)) {
            elan_tportRxWait(elarge, NULL, NULL, &size);
            
            msg = lbuf;
            recv_large_done = 0;
            flg = 1;
            
            CmiPushPE(CMI_DEST_RANK(msg), msg);
#if CMK_BROADCAST_SPANNING_TREE
            if (CMI_BROADCAST_ROOT(msg))
                SendSpanningChildren(size, msg);
#endif
            step1 = 0;
            
            if(blockingReceiveFlag)
                nlarge_torecv --;
        }

        emcount = 0;
        for(mcount = 0; mcount < midQSize; mcount ++){
            emcount = (mcount + event_m_idx) % midQSize;
            if(elan_tportRxDone(emid[emcount])) {
                elan_tportRxWait(emid[emcount], NULL, NULL, &size);
                
                msg = mbuf[emcount];
                mbuf[emcount] = NULL;
                
                recv_mid_done[emcount] = 0;
                flg = 1;
                
                CmiPushPE(CMI_DEST_RANK(msg), msg);
                
#if CMK_BROADCAST_SPANNING_TREE
                if (CMI_BROADCAST_ROOT(msg))
                    SendSpanningChildren(size, msg);
#endif
                if(blockingReceiveFlag)
                    nlarge_torecv --;
            }
            else {
#if USE_SHM 
                elan_deviceCheck(elan_base->state);
#else 
                ;
#endif
                emcount = (emcount + midQSize - 1) % midQSize;
                break;
            }
        }
        event_m_idx = emcount + 1;
        
        ecount = 0;
        for(rcount = 0; rcount < smallQSize; rcount ++){
            ecount = (rcount + event_idx) % smallQSize;
            if(elan_tportRxDone(esmall[ecount]) || 
               (retflag == 3 && nlarge_torecv == 0 && !flg)) {
                elan_tportRxWait(esmall[ecount], &src, &tag, &size );
                
                msg = sbuf[ecount];
                sbuf[ecount] = NULL;
                
                recv_small_done[ecount] = 0;
                    
                if(tag == TAG_SMALL) {
                    flg = 1;
                    CmiPushPE(CMI_DEST_RANK(msg), msg);                
#if CMK_BROADCAST_SPANNING_TREE
                    if (CMI_BROADCAST_ROOT(msg))
                        SendSpanningChildren(size, msg);
#endif
                }
                else if(tag == TAG_LARGE_HEADER) {
                    //CmiPrintf("[%d] Received Header\n", CmiMyPe());
                    nlarge_torecv ++;
                    CmiFree(msg);
                }
                else if(tag == TAG_GET_BASED_SEND) {
                    handleGetHeader(msg, src);
                }                    

                if(retflag == 3)
                    retflag = 1;
            }
            else {
#if USE_SHM 
                elan_deviceCheck(elan_base->state);
#else 
                ;
#endif
                ecount = (ecount + smallQSize - 1) % smallQSize;
                break;
            }
        }
        event_idx = ecount + 1;
        
#if CMK_PERSISTENT_COMM
        PumpPersistent();
#endif
        
        if(!flg) {
#ifndef CMK_OPTIMIZE 
#if ! CMK_TRACE_IN_CHARM

            double pmp_end_time = CmiWallTimer();
            if(pmp_end_time > pmp_start_time + 50/1e6)
                traceUserBracketEvent(10, pmp_start_time, pmp_end_time);
#endif
#endif
#if CMK_IMMEDIATE_MSG && !CMK_SMP
            CmiHandleImmediate();
#endif
            return recd;    
        }
        
        if (retflag) {
#ifndef CMK_OPTIMIZE 
#if ! CMK_TRACE_IN_CHARM
            double pmp_end_time = CmiWallTimer();
            if(pmp_end_time > pmp_start_time + 50/1e6)
                traceUserBracketEvent(10, pmp_start_time, pmp_end_time);
#endif
#endif
#if CMK_IMMEDIATE_MSG && !CMK_SMP
            CmiHandleImmediate();
#endif
            return flg;
        }
        
        recd = 1;
        flg = 0;
    }
#if CMK_IMMEDIATE_MSG && !CMK_SMP
    CmiHandleImmediate();
#endif
    return recd;
}

void *remote_get(void * srcptr, void *destptr, int size, int srcPE){
    return (void *)elan_get(elan_base->state, srcptr, destptr, size, srcPE);
}

int remote_get_done(void *e){
    ELAN_EVENT *evt = (ELAN_EVENT *)e;

    int flag = elan_poll(evt, ELAN_POLL_EVENT);
    return flag;
}

/*
void remote_get_wait_all(){
    elan_getWaitAll(elan_base->state, ELAN_WAIT_EVENT);
}

void remote_put_wait_all(){
    elan_putWaitAll(elan_base->state, ELAN_WAIT_EVENT);
}
*/

/********************* MESSAGE RECEIVE FUNCTIONS ******************/
void *CmiGetNonLocal(void)
{
    register CmiState cs = CmiGetState();
    register void *msg = NULL;
    CmiIdleLock_checkMessage(&cs->idle);
    
    msg =  PCQueuePop(cs->recv); 
    
    if(msg) {
        return msg;
    }

    //get new messages and flush receive buffers
    PumpMsgs(1);           // PumpMsgs(0)
    //we are idle do more work
    CmiReleaseSentMessages();
    ElanSendQueuedMessages();

    msg =  PCQueuePop(cs->recv); 
    return msg;
}

void CmiPing() {
    CmiReleaseSentMessages();
    PumpMsgs(0);
    ElanSendQueuedMessages();
}


void enableBlockingReceives(){
    blockingReceiveFlag = 1;
}

void disableBlockingReceives(){
    blockingReceiveFlag = 0;
}

static int toggle = 0;  //Blocking receive posted only after all idle
                        //handlers are called

void CmiNotifyIdle(void)
{
    static int previousSleepTime = 0;
    CmiReleaseSentMessages();
    ElanSendQueuedMessages();

    PumpMsgs(1);    
    toggle = 0;
}

void CmiNotifyStillIdle(void)
{
    static int previousSleepTime = 0;
    CmiReleaseSentMessages();
    ElanSendQueuedMessages();
    
    if(!PumpMsgs(1) && blockingReceiveFlag && toggle){
        if (!PCQueueEmpty(CmiGetState()->recv)) return; 
        if (!CdsFifo_Empty(CpvAccess(CmiLocalQueue))) return;
        if (!CqsEmpty(CpvAccess(CsdSchedQueue))) return;
        if (sent_msgs) return;
        if (cur_unsent) return;
        PumpMsgs(3); 
    }
    toggle = 1;
}


 
#if CMK_IMMEDIATE_MSG
void CmiProbeImmediateMsg()
{
    PumpMsgs(0);
}
#endif
/********************* MESSAGE SEND FUNCTIONS ******************/

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

void CmiSyncSendFn(int destPE, int size, char *msg)
{
    CmiState cs = CmiGetState();
    
    char *dupmsg;
    dupmsg = (char *) CmiAlloc(size);
    memcpy(dupmsg, msg, size);
    
    //  CmiPrintf("Setting root to %d\n", 0);
    CMI_SET_BROADCAST_ROOT(dupmsg, 0);
    
    if (cs->pe==destPE)
        CmiSendSelf(dupmsg);
    else
        CmiAsyncSendFn(destPE, size, dupmsg);
}

void ElanBasicSendFn(SMSG_LIST * ptr){
    int tag = 0, sync_mode = 0;
    int tiny_msg = 0;
    
    ptr->status = BASIC_SEND;
    
    if (ptr->size <= SMALL_MESSAGE_SIZE)
        tag = TAG_SMALL;
    else if (ptr->size < MID_MESSAGE_SIZE)
        tag = TAG_MID;
    else {        
        if(!ptr->is_broadcast && enableGetBasedSend) {
            ElanGetBasedSend(ptr);
            return;
        }
        
        tag = TAG_LARGE;
    }

    //if(ptr->size > SYNC_MESSAGE_SIZE)
    //  sync_mode = ELAN_TPORT_TXSYNC;
    
    tiny_msg = 0; //A sizeof(int) byte message 
    //sent to wake up a blocked process
    
    //WAKE A PROCESS SLEEPING ON A BLOCKING RECEIVE UP,
    //WITH A SMALL MESSAGE THAT MATCHES THE TAG OF A SMALL MESSAGE
    //BUT IS ACTUALLY A MID OR LARGE MESSAGE
    if(ptr->size > SMALL_MESSAGE_SIZE && blockingReceiveFlag) {
        elan_tportTxWait(elan_tportTxStart(elan_port, 0, ptr->destpe, 
                                           CmiMyPe(), TAG_LARGE_HEADER, 
                                           &tiny_msg, sizeof(int)));
    }

    ptr->e = elan_tportTxStart(elan_port, sync_mode, ptr->destpe, CmiMyPe(),
                               tag, ptr->msg, ptr->size);
    ptr->sent = 1;
    
    MsgQueueLen++;
    MsgQueueBytes += ptr->size;
    
    outstandingMsgs[ptr->destpe] = 1;
}

CmiCommHandle ElanSendFn(int destPE, int size, char *msg, int flag)
{
    CmiState cs = CmiGetState();
    SMSG_LIST *msg_tmp;
    CmiUInt2  rank, node;
    
    if(destPE == cs->pe) {
        char *dupmsg = (char *) CmiAlloc(size);
        memcpy(dupmsg, msg, size);
        CmiSendSelf(dupmsg);
        return 0;
    }
    
    CQdCreate(CpvAccess(cQdState), 1);
#if CMK_PERSISTENT_COMM
    if (phs) {
        CmiAssert(phsSize == 1);
        CmiSendPersistentMsg(*phs, destPE, size, msg);
        return NULL;
    }
#endif
    
    msg_tmp = (SMSG_LIST *) CmiAlloc(sizeof(SMSG_LIST));
    msg_tmp->msg = msg;
    msg_tmp->next = 0;
    msg_tmp->size = size;
    msg_tmp->sent = 0;
    msg_tmp->e = NULL;
    msg_tmp->destpe = destPE;
    msg_tmp->is_broadcast = flag;

    if ((MsgQueueLen > request_max || MsgQueueBytes > request_bytes) 
        && (!flag)) {
        CmiReleaseSentMessages();
        PumpMsgs(1); //PumpMsgs(0) 
    }

    ElanSendQueuedMessages();

    if(MsgQueueLen > request_max || MsgQueueBytes > request_bytes 
       || outstandingMsgs[destPE]){
        
        if(sent_msgs==0)
            sent_msgs = msg_tmp;
        else
            end_sent->next = msg_tmp;
        end_sent = msg_tmp;
        
        if(cur_unsent == 0)
            cur_unsent = msg_tmp;
        
    }
    else{        
        ElanBasicSendFn(msg_tmp);

        if(sent_msgs==0)
            sent_msgs = msg_tmp;
        else
            end_sent->next = msg_tmp;
        end_sent = msg_tmp;
        
        return (CmiCommHandle) msg_tmp->e;
    }
    return NULL;
}

void ElanSendQueuedMessages() {
    SMSG_LIST * ptr = cur_unsent, *new_unsent = NULL;
    while (MsgQueueLen <= request_max && MsgQueueBytes <= request_bytes 
           && ptr != NULL) {

        if(!outstandingMsgs[ptr->destpe] && !ptr->sent)
            ElanBasicSendFn(ptr);
        else if ((!ptr->sent) && (new_unsent == NULL))
            new_unsent = ptr;
        
        ptr = ptr->next;
    }
    
    if(new_unsent)
        cur_unsent = new_unsent;
    else
        cur_unsent = ptr;
}

void ElanGetBasedSend(SMSG_LIST *msg_tmp){
    //CmiPrintf("using get based send\n");
    GetHeader *gmsg = (GetHeader *) CmiAlloc(sizeof(GetHeader));

    gmsg->src_addr = msg_tmp->msg;
    gmsg->size = msg_tmp->size;
    CMI_SET_BROADCAST_ROOT(gmsg, 0);
    
    msg_tmp->sent = 1;
    msg_tmp->e = NULL;
    msg_tmp->status = GET_BASED_SEND;
    msg_tmp->done = 0;
    
    msg_tmp->gmsg = (char *)gmsg;
    gmsg->flag_addr = (char *)&(msg_tmp->done);
    
    msg_tmp->e = elan_tportTxStart(elan_port, 0, msg_tmp->destpe, CmiMyPe(), 
                                   TAG_GET_BASED_SEND, gmsg, sizeof(GetHeader));
    
    MsgQueueLen++;
    MsgQueueBytes += msg_tmp->size;
    
    outstandingMsgs[msg_tmp->destpe] = 1;
}

void handleGetHeader(char *msg, int src){
    GetHeader *gmsg = (GetHeader *) msg;

    char *newmsg = CmiAlloc(gmsg->size);
    
    SMSG_LIST *msg_tmp = (SMSG_LIST *)CmiAlloc(sizeof(SMSG_LIST));
    msg_tmp->msg = msg;
    msg_tmp->next = 0;
    msg_tmp->size = gmsg->size;
    msg_tmp->sent = 1;
    msg_tmp->destpe = src;
    msg_tmp->status = RECEIVE_GET;
    msg_tmp->done = 0;
    msg_tmp->flag_addr = (long *)gmsg->flag_addr;
    msg_tmp->newmsg = newmsg;
    
    msg_tmp->e = elan_get(elan_base->state, gmsg->src_addr, msg_tmp->newmsg, gmsg->size, src);
    
    if(sent_msgs==0) {
        sent_msgs = msg_tmp;
        end_sent = msg_tmp;
    }
    else {
        msg_tmp->next = sent_msgs;
        sent_msgs = msg_tmp;
    }
}

long trueFlag = 1;
void processGetEnv(SMSG_LIST *ptr){

    if(ptr->status == BASIC_SEND)
        return;

    if(ptr->status == GET_BASED_SEND) {
        if(ptr->gmsg != NULL) {