Home Research Papers Posters Manuals Talks Download People Help Wiki Internal

arch/net/machine-ammasso.c

Go to the documentation of this file.

#ifndef ALIGN8
#define ALIGN8(x)   (int)((~7)&((x)+7))
#endif

/* DYNAMIC ALLOCATOR: Limit of the allowed pinned memory */
#define MAX_PINNED_MEMORY   100000000
/* DYNAMIC ALLOCATOR END */

#define WASTE_TIME 600
// In order to use CC_POST_CHECK, the last argument to cc_qp_post_sq must be "nWR"
#define CC_POST_CHECK(routine,args,nodeTo) {\
        int retry=10000000; \
        while (ammasso_check_post_err(routine args, #routine, __LINE__, &nWR, nodeTo, retry) == 1) { \
          int i; \
          retry += WASTE_TIME; \
          for (i=0; i<=WASTE_TIME; ++i) { retry --; } \
        } \
      }

#define CC_CHECK(routine,args) \
        ammasso_check_err(routine args, #routine, __LINE__);

static void ammasso_check_err(cc_status_t returnCode,const char *routine,int line) {
  if (returnCode!=CC_OK) {
    char buf[128];
    char *errMsg = cc_status_to_string(returnCode);

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE3(5,"Fatal CC error while executing %s at %s:%d\n", routine, __FILE__, line);
    MACHSTATE2(5,"  Description: %d, %s\n",returnCode,errMsg);
    sprintf(buf,"Fatal CC error while executing %s at %s:%d\n"
            "  Description: %d, %s\n", routine, __FILE__, line,returnCode,errMsg);
    CmiAbort(buf);
  }
}

// We pass the pointer used in the cc_qp_post_sq call as last parameter, since
// when this function is called, cc_pq_post_sq has already been called
static int ammasso_check_post_err(cc_status_t returnCode,const char *routine,int line, int *nWR, int nodeTo, int retry) {
  if (returnCode == CCERR_TOO_MANY_WRS_POSTED && *nWR != 1 && retry>0) {
    cc_wc_t wc;
    // drain the send completion queue and retry
    while (cc_cq_poll(contextBlock->rnic, nodes[nodeTo].send_cq, &wc) == CC_OK) {
      MACHSTATE1(5, "Error posting send request - INFO: Send completed with node %d... now waiting for acknowledge...", nodeTo);
    }
    MACHSTATE(5, "Error posting send request - Retrying...");
    return 1;
  }

  if (returnCode != CC_OK || *nWR != 1) {
    char buf[128];
    char *errMsg = cc_status_to_string(returnCode);

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE3(5,"Fatal CC error while executing %s at %s:%d\n", routine, __FILE__, line);
    MACHSTATE3(5,"  Description: %d, %s (nWR = %d)\n",returnCode,errMsg,nWR);
    sprintf(buf,"Fatal CC error while executing %s at %s:%d\n"
            "  Description: %d, %s (nWR = %d)\n", routine, __FILE__, line,returnCode,errMsg,*nWR);
    CmiAbort(buf);
  }
  return 0;
}


// Function ProtoTypes /////////////////////////////////////////////////////////////////////////////

void CmiMachineInit(char **argv);

void CmiAmmassoNodeAddressesStoreHandler(int pe, struct sockaddr_in *addr, int port);

void AmmassoDoIdle();
void CmiNotifyIdle(void);
static CmiIdleState* CmiNotifyGetState(void);
static void CmiNotifyBeginIdle(CmiIdleState *s);
static void CmiNotifyStillIdle(CmiIdleState *s);

void sendAck(OtherNode node);
AmmassoToken *getQPSendToken(OtherNode node);
int sendDataOnQP(char* data, int len, OtherNode node, char flags);
void DeliverViaNetwork(OutgoingMsg msg, OtherNode otherNode, int rank, unsigned int broot, int copy);
static void CommunicationServer(int withDelayMs, int where);
void CmiMachineExit();

void AsynchronousEventHandler(cc_rnic_handle_t rnic, cc_event_record_t *eventRecord, void *cb);
void CheckRecvBufForMessage(OtherNode node);
//void CompletionEventHandler(cc_rnic_handle_t rnic, cc_cq_handle_t cq, void *cb);
//void CompletionEventHandlerWithAckFlag(cc_rnic_handle_t rnic, cc_cq_handle_t cq, void *cb, int breakOnAck);

void CmiAmmassoOpenQueuePairs();

void processAmmassoControlMessage(char* msg, int len, Tailer *tail, OtherNode from);
int ProcessMessage(char* msg, int len, Tailer *tail, OtherNode from);

OtherNode getNodeFromQPId(cc_qp_id_t qp_id);
OtherNode getNodeFromQPHandle(cc_qp_handle_t qp);

void establishQPConnection(OtherNode node, int reuseQPFlag);
void reestablishQPConnection(OtherNode node);
void closeQPConnection(OtherNode node, int destroyQPFlag);

void BufferAlloc(int n);
void TokenAlloc(int n);
void RequestTokens(OtherNode node, int n);
void GrantTokens(OtherNode node, int n);
void RequestReleaseTokens(OtherNode node, int n);
void ReleaseTokens(OtherNode node, int n);

// Function Bodies /////////////////////////////////////////////////////////////////////////////////

/* Callbacks used by the DYNAMIC ALLOCATOR */
void AllocatorCheck () {
  int i, limit;
  char buf[24];
  for (i=0; i<contextBlock->numNodes; ++i) {
    if (i==contextBlock->myNode) continue;
    limit = nodes[i].num_sendTokens - nodes[i].max_used_tokens - 10;
    CmiPrintf("[%d] AllocatorCheck called: node %d, limit %d\n",CmiMyPe(),i,limit);
    if (limit > 0) {
      ReleaseTokens(&nodes[i], limit);
      CmiPrintf("[%d] Releasing %d tokens to %d\n", CmiMyPe(), limit, i);
      nodes[i].max_used_tokens = 0;
    }
  }
}
/* DYNAMIC ALLOCATOR END */

void BufferAlloc(int n) {
  int i;
  char buf[128];
  cc_stag_index_t newStagIndex;
  AmmassoBuffer *newBuffers;

  MACHSTATE1(3, "Allocating %d new Receive Buffers",n);

  // Try to allocate the memory for n receiving buffers
  newBuffers = (AmmassoBuffer*) CmiAlloc(n*sizeof(AmmassoBuffer));

  if (newBuffers == NULL) {

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE(5, "BufferAlloc() - ERROR: Unable to allocate memory for RECV buffers");
    sprintf(buf, "BufferAlloc() - ERROR: Unable to allocate memory for RECV buffers");
    CmiAbort(buf);
  }

  contextBlock->pinnedMemory += n*sizeof(AmmassoBuffer);
  CC_CHECK(cc_nsmr_register_virt,(contextBlock->rnic,
                                  CC_ADDR_TYPE_VA_BASED,
                                  (cc_byte_t*)newBuffers,
                                  n*sizeof(AmmassoBuffer),
                                  contextBlock->pd_id,
                                  0, 0,
                                  CC_ACF_LOCAL_READ | CC_ACF_LOCAL_WRITE | CC_ACF_REMOTE_WRITE,
                                  &newStagIndex)
           );

  for (i=0; i<n; ++i) {
    newBuffers[i].tail.length = 0;
    newBuffers[i].next = &(newBuffers[i+1]);
    newBuffers[i].stag = newStagIndex;
  }
  newBuffers[n-1].next = NULL;
  if (contextBlock->freeRecvBuffers == NULL) {
    contextBlock->freeRecvBuffers = newBuffers;
  } else {
    contextBlock->last_freeRecvBuffers->next = newBuffers;
  }
  contextBlock->last_freeRecvBuffers = &newBuffers[n-1];
  contextBlock->num_freeRecvBuffers += n;
}

void TokenAlloc(int n) {
  int i;
  char buf[128];
  cc_stag_index_t newStagIndex;
  AmmassoToken *sendToken, *tokenScanner;
  cc_data_addr_t *sendSgl;
  AmmassoBuffer *sendBuffer;

  MACHSTATE1(3, "Allocating %d new Tokens",n);

  // Try to allocate the memory for n sending buffers
  sendBuffer = (AmmassoBuffer*) CmiAlloc(n*sizeof(AmmassoBuffer));

  if (sendBuffer == NULL) {

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE(5, "TokenAlloc() - ERROR: Unable to allocate memory for SEND buffers");
    sprintf(buf, "TokenAlloc() - ERROR: Unable to allocate memory for SEND buffers");
    CmiAbort(buf);
  }

  contextBlock->pinnedMemory += n*sizeof(AmmassoBuffer);
  CC_CHECK(cc_nsmr_register_virt,(contextBlock->rnic,
                                  CC_ADDR_TYPE_VA_BASED,
                                  (cc_byte_t*)sendBuffer,
                                  n*sizeof(AmmassoBuffer),
                                  contextBlock->pd_id,
                                  0, 0,
                                  CC_ACF_LOCAL_READ | CC_ACF_LOCAL_WRITE,
                                  &newStagIndex)
           );

  // Allocate the send tokens
  sendToken = (AmmassoToken*) CmiAlloc(n*ALIGN8(sizeof(AmmassoToken)));

  if (sendToken == NULL) {

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE(5, "TokenAlloc() - ERROR: Unable to allocate memory for send TOKEN buffers");
    sprintf(buf, "TokenAlloc() - ERROR: Unable to allocate memory for send TOKEN buffers");
    CmiAbort(buf);
  }

  sendSgl = (cc_data_addr_t*) CmiAlloc(n*ALIGN8(sizeof(cc_data_addr_t)));

  if (sendSgl == NULL) {

    // Attempt to close the RNIC
    cc_rnic_close(contextBlock->rnic);

    // Let the user know what happened and bail
    MACHSTATE(5, "TokenAlloc() - ERROR: Unable to allocate memory for send SGL buffers");
    sprintf(buf, "TokenAlloc() - ERROR: Unable to allocate memory for send SGL buffers");
    CmiAbort(buf);
  }

  tokenScanner = sendToken;
  for (i=0; i<n; ++i) {
    sendSgl->stag = newStagIndex;
    sendSgl->length = AMMASSO_BUFSIZE + sizeof(Tailer);
    sendSgl->to = (unsigned long)&(sendBuffer[i]);
    tokenScanner->wr.wr_id = (unsigned long)tokenScanner;
    tokenScanner->wr.wr_type = CC_WR_TYPE_RDMA_WRITE;
    tokenScanner->wr.wr_u.rdma_write.local_sgl.sge_count = 1;
    tokenScanner->wr.wr_u.rdma_write.local_sgl.sge_list = sendSgl;
    tokenScanner->wr.signaled = 1;
    tokenScanner->localBuf = (AmmassoBuffer*)&(sendBuffer[i]);
    LIST_ENQUEUE(contextBlock->,freeTokens,tokenScanner);
    sendSgl = (cc_data_addr_t*)(((char*)sendSgl)+ALIGN8(sizeof(cc_data_addr_t)));
    tokenScanner = (AmmassoToken*)(((char*)tokenScanner)+ALIGN8(sizeof(AmmassoToken)));
  }
}

void RequestTokens(OtherNode node, int n) {
  char buf[24];
  *((int*)buf) = n;
  sendDataOnQP(buf, sizeof(int), node, AMMASSO_MOREBUFFERS);
}

void GrantTokens(OtherNode node, int n) {
  int i;
  char *buf;
  AmmassoBuffer *buffer;
  AmmassoBuffer *prebuffer;
  AmmassoTokenDescription *tokenDesc;
  if (node->pending != NULL) return;
  if (n*sizeof(AmmassoTokenDescription) + sizeof(int) > AMMASSO_BUFSIZE) {
    n = (AMMASSO_BUFSIZE-sizeof(int)) / sizeof(AmmassoTokenDescription);
  }
  if (contextBlock->num_freeRecvBuffers < n) {
    int quantity = (n - contextBlock->num_freeRecvBuffers + 1023) & (~1023);
    BufferAlloc(quantity);
  }
  buf = (char*) CmiAlloc(n*sizeof(AmmassoTokenDescription) + sizeof(int));
  *((int*)buf) = n;
  tokenDesc = (AmmassoTokenDescription*)(buf+sizeof(int));
  buffer = contextBlock->freeRecvBuffers;
  for (i=0; i<n; ++i) {
    tokenDesc[i].stag = buffer->stag;
    tokenDesc[i].to = (unsigned long)buffer;
    prebuffer = buffer;
    buffer = buffer->next;
  }
  node->pending = contextBlock->freeRecvBuffers;
  node->last_pending = prebuffer;
  node->num_pending = n;
  prebuffer->next = NULL;
  contextBlock->num_freeRecvBuffers -= n;
  contextBlock->freeRecvBuffers = buffer;
  sendDataOnQP(buf, n*sizeof(AmmassoTokenDescription) + sizeof(int), node, AMMASSO_ALLOCATE);
  CmiFree(buf);
}

void RequestReleaseTokens(OtherNode node, int n) {
  char buf[24];
  *((int*)buf) = n;
  sendDataOnQP(buf, sizeof(int), node, AMMASSO_RELEASE);
}

void ReleaseTokens(OtherNode node, int n) {
  int i, nWR;
  AmmassoToken *token;
  AmmassoBuffer *tokenBuf;
  cc_data_addr_t *tokenSgl;

  if (node->num_sendTokens < n) n = node->num_sendTokens - 1;
  if (n <= 0) return;
  token = node->sendTokens;
  tokenBuf = token->localBuf;

  tokenBuf->tail.length = 1;
  tokenBuf->tail.ack = 0;  // do not send any ACK with this message
  tokenBuf->tail.flags = AMMASSO_RELEASED;

  // Setup the local SGL
  tokenSgl = token->wr.wr_u.rdma_write.local_sgl.sge_list;
  tokenSgl->length = sizeof(Tailer);
  tokenSgl->to = (unsigned long)&tokenBuf->tail;
  token->wr.wr_u.rdma_write.remote_to = (unsigned long)&token->remoteBuf->tail;

  CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, &token->wr, 1, &nWR),node->myNode);

  if (contextBlock->freeTokens == NULL) {
    contextBlock->freeTokens = node->sendTokens;
  } else {
    contextBlock->last_freeTokens->next = node->sendTokens;
  }
  for (i=1; i<n; ++i) token = token->next;
  contextBlock->last_freeTokens = token;
  node->sendTokens = token->next;
  token->next = NULL;
  contextBlock->num_freeTokens += n;
  node->num_sendTokens -= n;
}

/* CmiMachineInit()
 *   This is called as the node is starting up.  It does some initialization of the machine layer.
 */
void CmiMachineInit(char **argv) {

  char buf[128];
  cc_status_t rtn;

  AMMASSO_STATS_INIT

  AMMASSO_STATS_START(MachineInit)

  MACHSTATE(2, "CmiMachineInit() - INFO: (***** Ammasso Specific*****) - Called... Initializing RNIC...");
  MACHSTATE1(1, "CmiMachineInit() - INFO: Cmi_charmrun_pid = %d", Cmi_charmrun_pid);


  //CcdCallOnConditionKeep(CcdPERIODIC, (CcdVoidFn)periodicFunc, NULL);


  // Allocate a context block that will be used throughout this machine layer
  if (contextBlock != NULL) {
    MACHSTATE(5, "CmiMachineInit() - ERROR: contextBlock != NULL");
    sprintf(buf, "CmiMachineInit() - ERROR: contextBlock != NULL");
    CmiAbort(buf);
  }
  contextBlock = (mycb_t*)malloc(sizeof(mycb_t));
  if (contextBlock == NULL) {
    MACHSTATE(5, "CmiMachineInit() - ERROR: Unable to malloc memory for contextBlock");
    sprintf(buf, "CmiMachineInit() - ERROR: Unable to malloc memory for contextBlock");
    CmiAbort(buf);
  }

  // Initialize the contextBlock by zero-ing everything out and then setting anything special
  memset(contextBlock, 0, sizeof(mycb_t));
  contextBlock->rnic = -1;

  MACHSTATE(1, "CmiMachineInit() - INFO: (PRE-OPEN_RNIC)");

  // Check to see if in stand-alone mode
  if (Cmi_charmrun_pid != 0) {

    // Try to Open the RNIC
    //   TODO : Look-up the difference between CC_PBL_PAGE_MODE and CC_PBL_BLOCK_MODE
    //   TODO : Would a call to cc_rnic_enum or cc_rnic_query do any good here?
    rtn = cc_rnic_open(0, CC_PBL_PAGE_MODE, contextBlock, &(contextBlock->rnic));
    if (rtn != CC_OK) {
      MACHSTATE2(5, "CmiMachineInit() - ERROR: Unable to open RNIC: %d, \"%s\"", rtn, cc_status_to_string(rtn));
      sprintf(buf, "CmiMachineInit() - ERROR: Unable to open RNIC: %d, \"%s\"", rtn, cc_status_to_string(rtn));
      CmiAbort(buf);
    }

    MACHSTATE(1, "CmiMachineInit() - INFO: (PRE-SET-ASYNC-HANDLER)");

    // Set the asynchronous event handler function
    CC_CHECK(cc_eh_set_async_handler,(contextBlock->rnic, AsynchronousEventHandler, contextBlock));

    /*
    MACHSTATE(3, "CmiMachineInit() - INFO: (PRE-SET-CE-HANDLER)");

    // Set the Completion Event Handler
    contextBlock->eh_id = 0;
    CC_CHECK(cc_eh_set_ce_handler,(contextBlock->rnic, CompletionEventHandler, &(contextBlock->eh_id)));
    */

    MACHSTATE(1, "CmiMachineInit() - INFO: (PRE-PD-ALLOC)");

    // Allocate the Protection Domain
    CC_CHECK(cc_pd_alloc,(contextBlock->rnic, &(contextBlock->pd_id)));

    MACHSTATE(1, "CmiMachineInit() - INFO: RNIC Open For Business!!!");

  } else {  // Otherwise, not in stand-alone mode

    // Flag the rnic variable as invalid
    contextBlock->rnic = -1;
  }

  MACHSTATE(2, "CmiMachineInit() - INFO: Completed Successfully !!!");

  AMMASSO_STATS_END(MachineInit)
}

void CmiCommunicationInit(char **argv)
{
}

void CmiAmmassoNodeAddressesStoreHandler(int pe, struct sockaddr_in *addr, int port) {

  // DMK : NOTE : The hope is that this can be used to request the RMDA addresses of the other nodes after the
  //              initial addresses from charmrun are given to the node.  Get the address here, use that to request
  //              the RDMA address, use the RDMA address to create the QP connection (in establishQPConnection(), which
  //              only subtracts one from the address at the moment... the way our cluster is setup).

  MACHSTATE1(2, "CmiNodeAddressesStoreHandler() - INFO: pe = %d", pe);
  MACHSTATE1(1, "                                       addr = { sin_family = %d,", addr->sin_family);
  MACHSTATE1(1, "                                                sin_port = %d,", addr->sin_port);
  MACHSTATE4(1, "                                                sin_addr.s_addr = %d.%d.%d.%d }", (addr->sin_addr.s_addr & 0xFF), ((addr->sin_addr.s_addr >> 8) & 0xFF), ((addr->sin_addr.s_addr >> 16) & 0xFF), ((addr->sin_addr.s_addr >> 24) & 0xFF));
  MACHSTATE1(1, "                                       port = %d", port);
}


void AmmassoDoIdle() {

  int i;
  cc_wc_t wc;

  AMMASSO_STATS_START(AmmassoDoIdle)

  /* DYNAMIC ALLOCATOR: Callbacks */
  /*if (contextBlock->conditionRegistered == 0) {
    CcdCallOnConditionKeep(CcdPERIODIC_1s, (CcdVoidFn) AllocatorCheck, NULL);
    //CcdCallFnAfter((CcdVoidFn) AllocatorCheck, NULL, 100);
    contextBlock->conditionRegistered = 1;
    }*/
  /* DYNAMIC ALLOCATOR END */

  for (i = 0; i < contextBlock->numNodes; i++) {
    if (i == contextBlock->myNode) continue;
    CheckRecvBufForMessage(&(nodes[i]));
    while (cc_cq_poll(contextBlock->rnic, nodes[i].send_cq, &wc) == CC_OK) {
      MACHSTATE1(3, "AmmassoDoIdle() - INFO: Send completed with node %d... now waiting for acknowledge...", i);
    }
  }

  AMMASSO_STATS_END(AmmassoDoIdle)
}

void CmiNotifyIdle(void) {
  AmmassoDoIdle();
}

static CmiIdleState* CmiNotifyGetState(void) {
  return NULL;
}

static void CmiNotifyBeginIdle(CmiIdleState *s) {
  AmmassoDoIdle();
}

static void CmiNotifyStillIdle(CmiIdleState *s) {
  AmmassoDoIdle();
}

/* NOTE: if the ack overflows, we cannot use this method of sending, but we need
   to send a special message for the purpose */
void sendAck(OtherNode node) {

  int nWR;

  AMMASSO_STATS_START(sendAck)

  MACHSTATE2(3, "sendAck() - Ammasso - INFO: Called... sending ACK %d to node %d", *node->remoteAck, node->myNode);

  if (*node->remoteAck < ACK_MASK) {

    // Send an ACK message to the specified QP/Connection/Node
    CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, node->ack_sq_wr, 1, &nWR),node->myNode);

  } else {
    // Rare case: happens only after days of run! In this case, do not update
    // directly the ack, but zero it on the other side, wait one second to be
    // safe, and then send a regular message with the ACK
    AmmassoToken *token;
    AmmassoBuffer *tokenBuf;
    int tmp_ack = *node->remoteAck;
    *node->remoteAck = 0;
    CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, node->ack_sq_wr, 1, &nWR),node->myNode);

    sleep(1);

    token = getQPSendToken(node);
    tokenBuf = token->localBuf;
    tokenBuf->tail.ack = tmp_ack;
    tokenBuf->tail.flags = ACK_WRAPPING;
    tokenBuf->tail.length = 1; // So it will be seen by the receiver
    token->wr.wr_u.rdma_write.local_sgl.sge_list->length = sizeof(Tailer);
    token->wr.wr_u.rdma_write.local_sgl.sge_list->to = (unsigned long)&tokenBuf->tail;
    token->wr.wr_u.rdma_write.remote_to = (unsigned long)&token->remoteBuf->tail;
    CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, &token->wr, 1, &nWR),node->myNode);
    LIST_ENQUEUE(node->,usedTokens,token);
    node->max_used_tokens = (node->num_usedTokens>node->max_used_tokens)?node->num_usedTokens:node->max_used_tokens;
    *node->remoteAck = tmp_ack & ACK_MASK;
  }

  node->messagesNotYetAcknowledged = 0;

  AMMASSO_STATS_END(sendAck)
}

/* NOTE, even in SMP versions, only the communication server should be sending
   messages out, thus no locking should be necessary */
/* This function returns the token usable for next communication. If no token is
   available, it blocks until one becomes available */
AmmassoToken *getQPSendToken(OtherNode node) {
  AmmassoToken *token;
  int i;
  cc_wc_t wc;
  ammasso_ack_t newAck;
  while (node->connectionState != QP_CONN_STATE_CONNECTED ||
         node->sendTokens == NULL) {
    // Try to see if an ACK has been sent directly, so we free some tokens The
    // direct token will never be greater than ACK_MASK (by protocol
    // definition), so we do not need to wrap around
    MACHSTATE(3, "getQPSendBuffer() - INFO: No tokens available");
    if (*node->directAck > node->localAck) {
      newAck = *node->directAck;
      for (i=node->localAck; i<newAck; ++i) {
        LIST_DEQUEUE(node->,usedTokens,token);
        LIST_ENQUEUE(node->,sendTokens,token);
      }
      node->localAck = newAck;
    }

    CheckRecvBufForMessage(node);

    while (cc_cq_poll(contextBlock->rnic, node->send_cq, &wc) == CC_OK) {
      MACHSTATE1(3, "getQPSendBuffer() - INFO: Send completed with node %d... now waiting for acknowledge...", node->myNode);
    }
  }
  LIST_DEQUEUE(node->,sendTokens,token);
  return token;
}

/*
int getQPSendBuffer(OtherNode node, char force) {

  int rtnBufIndex, i;
  cc_wc_t wc;

  AMMASSO_STATS_START(getQPSendBuffer)

  MACHSTATE1(3, "getQPSendBuffer() - Ammasso - Called (send to node %d)...", node->myNode);

  while (1) {

    AMMASSO_STATS_START(getQPSendBuffer_loop)

    rtnBufIndex = -1;

    AMMASSO_STATS_START(getQPSendBuffer_lock)

    MACHSTATE(3, "getQPSendBuffer() - INFO: Pre-sendBufLock");
    #if CMK_SHARED_VARS_UNAVAILABLE
      while (node->sendBufLock != 0) { usleep(1); } // Since CmiLock() is not really a lock, actually wait
    #endif
    CmiLock(node->sendBufLock);

    AMMASSO_STATS_END(getQPSendBuffer_lock)
    
    // If force is set, let the message use any of the available send buffers.  Otherwise, there can only
    // be AMMASSO_NUMMSGBUFS_PER_QP message outstanding (haven't gotten an ACK for) so wait for that.
    // VERY IMPORTANT !!! Only use force for sending ACKs !!!!!!!!  If this is done, it is ensured that there
    // will always be at least one buffer available when the force code is executed
    if (node->connectionState == QP_CONN_STATE_CONNECTED) {
      if (force) {
        rtnBufIndex = node->send_UseIndex;
        node->send_UseIndex++;
        if (node->send_UseIndex >= AMMASSO_NUMMSGBUFS_PER_QP * 2)
          node->send_UseIndex = 0;
      } else {
        if (node->send_InUseCounter < AMMASSO_NUMMSGBUFS_PER_QP) {
          rtnBufIndex = node->send_UseIndex;
          node->send_InUseCounter++;
          node->send_UseIndex++;
          if (node->send_UseIndex >= AMMASSO_NUMMSGBUFS_PER_QP * 2)
            node->send_UseIndex = 0;
        }
      }
    }
 
    CmiUnlock(node->sendBufLock);
    MACHSTATE3(3, "getQPSendBuffer() - INFO: Post-sendBufLock - rtnBufIndex = %d, node->connectionState = %d, node->send_UseIndex = %d", rtnBufIndex, node->connectionState, node->send_UseIndex);

    if (rtnBufIndex >= 0) {

      AMMASSO_STATS_END(getQPSendBuffer_loop)
      break;

    } else {

      //usleep(1);

      AMMASSO_STATS_START(getQPSendBuffer_CEH)

      CheckRecvBufForMessage(node);
      //CompletionEventHandlerWithAckFlag(contextBlock->rnic, node->recv_cq, node, 1);

      //CompletionEventHandler(contextBlock->rnic, node->send_cq, node);
      while (cc_cq_poll(contextBlock->rnic, node->send_cq, &wc) == CC_OK) {
        MACHSTATE1(3, "getQPSendBuffer() - INFO: Send completed with node %d... now waiting for acknowledge...", nextNode);
      }


      AMMASSO_STATS_END(getQPSendBuffer_CEH)

      AMMASSO_STATS_END(getQPSendBuffer_loop)
    }
  }

  //node->send_UseIndex++;
  //if (node->send_UseIndex >= AMMASSO_NUMMSGBUFS_PER_QP * 2)
  //  node->send_UseIndex = 0;

  MACHSTATE1(3, "getQPSendBuffer() - Ammasso - Finished (returning buffer index: %d)", rtnBufIndex);

  AMMASSO_STATS_END(getQPSendBuffer)

  return rtnBufIndex;

}
*/

// NOTE: The force parameter can be thought of as an "is ACK" control message flag (see comments in getQPSendBuffer())
int sendDataOnQP(char* data, int len, OtherNode node, char flags) {

  AmmassoToken *sendBufToken;
  AmmassoBuffer *tokenBuf;
  cc_data_addr_t *tokenSgl;
  int toSendLength;
  cc_wc_t wc;
  cc_uint32_t nWR;
  char isFirst = 1;
  char *origMsgStart = data;
  char *sendBufBegin;

  int origSize = len;

  if (origSize <= 1024) {
    AMMASSO_STATS_START(sendDataOnQP_1024)
  } else if (origSize <= 2048) { 
    AMMASSO_STATS_START(sendDataOnQP_2048)
  } else if (origSize <= 4096) { 
    AMMASSO_STATS_START(sendDataOnQP_4096)
  } else if (origSize <= 16384) { 
    AMMASSO_STATS_START(sendDataOnQP_16384)
  } else {
    AMMASSO_STATS_START(sendDataOnQP_over)
  }

  AMMASSO_STATS_START(sendDataOnQP)

  //CompletionEventHandler(contextBlock->rnic, node->recv_cq, node);
  //CompletionEventHandler(contextBlock->rnic, node->send_cq, node);
  while (cc_cq_poll(contextBlock->rnic, node->send_cq, &wc) == CC_OK) {
    MACHSTATE1(3, "sendDataOnQP() - INFO: Send completed with node %d... now waiting for acknowledge...", node->myNode);
  }


  MACHSTATE2(2, "sendDataOnQP() - Ammasso - INFO: Called (send to node %d, len = %d)...", node->myNode, len);

  // Assert that control messages will not be fragmented
  CmiAssert(flags==0 || len<=AMMASSO_BUFSIZE);

  // DMK : For each message that is fragmented, attach another DGRAM header to
  // it, (keeping in mind that the control messages are no where near large
  // enough for this to occur).

  while (len > 0) {

    AMMASSO_STATS_START(sendDataOnQP_pre_send)  

    // Get a free send buffer (NOTE: This call will block until a send buffer is free)
    sendBufToken = getQPSendToken(node);
    tokenBuf = sendBufToken->localBuf;
    // Enqueue the token to the used queue immediately, so it is safe to be
    // interrupted by other calls
    LIST_ENQUEUE(node->,usedTokens,sendBufToken);
    node->max_used_tokens = (node->num_usedTokens>node->max_used_tokens)?node->num_usedTokens:node->max_used_tokens;

    // Copy the contents (up to AMMASSO_BUFSIZE worth) of data into the send buffer

    // The toSendLength includes the DGRAM header size. If the chunk sent is not
    // the first, the initial DGRAM_HEADER_SIZE bytes need to be contructed from
    // the DGRAM header of the original message (instead of just being copied
    // together with the message itself)

    if (isFirst) {

      toSendLength = len > AMMASSO_BUFSIZE ? AMMASSO_BUFSIZE : len;  // MIN of len and AMMASSO_BUFSIZE
      sendBufBegin = tokenBuf->buf + AMMASSO_BUFSIZE - ALIGN8(toSendLength);

      memcpy(sendBufBegin, data, toSendLength);

      MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: Sending 1st Fragment - toSendLength = %d...", toSendLength);

    } else {

      toSendLength = len > (AMMASSO_BUFSIZE - DGRAM_HEADER_SIZE) ? AMMASSO_BUFSIZE : (len+DGRAM_HEADER_SIZE);  // MIN of len and AMMASSO_BUFSIZE
      sendBufBegin = tokenBuf->buf + AMMASSO_BUFSIZE - ALIGN8(toSendLength);

      memcpy(sendBufBegin+DGRAM_HEADER_SIZE, data, toSendLength-DGRAM_HEADER_SIZE);
      
      // This dgram header is the same of the original message, except for the
      // sequence number, so copy the original and just modify the sequence
      // number.

      // NOTE: If the message is large enough that fragmentation needs to
      // happen, the send_next_lock is already owned by the thread executing
      // this code.
      memcpy(sendBufBegin, origMsgStart, DGRAM_HEADER_SIZE);

      ((DgramHeader*)sendBufBegin)->seqno = node->send_next;
      node->send_next = ((node->send_next+1) & DGRAM_SEQNO_MASK);  // Increase the sequence number

      MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: Sending Continuation Fragment - toSendLength = %d...", toSendLength);
    }

    // Write the size of the message at the end of the buffer, with the ack and
    // flags
    tokenBuf->tail.length = toSendLength;
    node->messagesNotYetAcknowledged = 0;
    tokenBuf->tail.ack = *node->remoteAck;
    if (*node->remoteAck > ACK_MASK) {
      // Rare case of ACK wrapping
      tokenBuf->tail.ack = 0;
      // in the rare case that we didn't send an ack with this message because
      // the ack just wrapped around (rare case), send a full ACK message. It is
      // safe to send it now, since the queues are consistent between sender and
      // receiver. The fact that this ack is effectively sent before the regular
      // message is not important, since on the other side it will be discovered
      // only after this one is received
      sendAck(node);
    }
    tokenBuf->tail.flags = flags;

    // Setup the local SGL
    tokenSgl = sendBufToken->wr.wr_u.rdma_write.local_sgl.sge_list;
    tokenSgl->length = ALIGN8(toSendLength) + sizeof(Tailer);
    tokenSgl->to = (unsigned long)sendBufBegin;
    sendBufToken->wr.wr_u.rdma_write.remote_to = (unsigned long)(((char*)sendBufToken->remoteBuf)+AMMASSO_BUFSIZE-ALIGN8(toSendLength));

    // The remote_to and remote_stag are already fixed part of the token
 
    AMMASSO_STATS_END(sendDataOnQP_pre_send)  
    AMMASSO_STATS_START(sendDataOnQP_send)

    MACHSTATE(3, "sendDataOnQP() - Ammasso - INFO: Enqueuing RDMA Write WR...");

    MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: tokenSgl->to = %p", tokenSgl->to);
    MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: sendBufToken->wr.wr_u.rdma_write.remote_to = %p", sendBufToken->wr.wr_u.rdma_write.remote_to);
    MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: tail.ack = %d", tokenBuf->tail.ack);
    MACHSTATE1(1, "sendDataOnQP() - Ammasso - INFO: tail.flags = %d", tokenBuf->tail.flags);

    CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, &sendBufToken->wr, 1, &nWR),node->myNode);
    
    MACHSTATE(1, "sendDataOnQP() - Ammasso - INFO: RDMA Write WR Enqueue Completed");

    AMMASSO_STATS_END(sendDataOnQP_send)
    AMMASSO_STATS_START(sendDataOnQP_post_send)  

    // Update the data and len variables for the next while (if fragmenting is needed)
    data += toSendLength;
    len -= toSendLength;
    if (isFirst == 0) {
      data -= DGRAM_HEADER_SIZE;
      len += DGRAM_HEADER_SIZE;
    }
    isFirst = 0;

    AMMASSO_STATS_END(sendDataOnQP_post_send)  
  }

  AMMASSO_STATS_END(sendDataOnQP)

  if (origSize <= 1024) {
    AMMASSO_STATS_END(sendDataOnQP_1024)
  } else if (origSize <= 2048) { 
    AMMASSO_STATS_END(sendDataOnQP_2048)
  } else if (origSize <= 4096) { 
    AMMASSO_STATS_END(sendDataOnQP_4096)
  } else if (origSize <= 16384) { 
    AMMASSO_STATS_END(sendDataOnQP_16384)
  } else {
    AMMASSO_STATS_END(sendDataOnQP_over)
  }

}


/* DeliverViaNetwork()
 *
 */
void DeliverViaNetwork(OutgoingMsg msg, OtherNode otherNode, int rank, unsigned int broot, int copy) {

  cc_status_t rtn;
  cc_stag_index_t stag;
  cc_data_addr_t sgl;
  cc_sq_wr_t wr;
  cc_uint32_t WRsPosted;

  AMMASSO_STATS_START(DeliverViaNetwork)

  MACHSTATE(2, "DeliverViaNetwork() - Ammasso - INFO: Called...");

  // We don't need to do this since the message data is being copied into the
  // send_buf, the OutgoingMsg can be free'd ASAP

  // The lock will be already held by the calling function in machine.c
  // (CommLock)

  AMMASSO_STATS_START(DeliverViaNetwork_post_lock)

  DgramHeaderMake(msg->data, rank, msg->src, Cmi_charmrun_pid, otherNode->send_next, broot);  // Set DGram Header Fields In-Place
  otherNode->send_next = ((otherNode->send_next+1) & DGRAM_SEQNO_MASK);  // Increase the sequence number

  MACHSTATE1(1, "DeliverViaNetwork() - INFO: Sending message to  node %d", otherNode->myNode);
  MACHSTATE1(1, "DeliverViaNetwork() - INFO:                     rank %d", rank);
  MACHSTATE1(1, "DeliverViaNetwork() - INFO:                    broot %d", broot);

  AMMASSO_STATS_START(DeliverViaNetwork_send)

  sendDataOnQP(msg->data, msg->size, otherNode, 0);

  AMMASSO_STATS_END(DeliverViaNetwork_send)

    //CmiUnlock(otherNode->send_next_lock);
  MACHSTATE(1, "DeliverViaNetwork() - INFO: Post-send_next_lock");


  // DMK : NOTE : I left this in as an example of how to retister the memory with the RNIC on the fly.  Since the ccil
  //              library has a bug which causes it not to de-pin memory that we unregister, it will probably be a better
  //              idea to fragment a message that is too large for a single buffer from the buffer pool.
  /***************************************************************
  // DMK : TODO : This is an important optimization area.  This is registering the memory where the outgoing message
  //              is located with the RNIC.  One balancing act that we will need to do is the cost of copying messages
  //              into memory regions already allocated VS. the cost of registering the memory.  The cost of registering
  //              memory might be constant as the memory doesn't have to traversed.  If the cost of doing a memcpy on a
  //              small message, since memcpy traverses the memory range, is less than the cost of registering the
  //              memory with the RNIC, it would be better to copy the message into a pre-registered memory location.
 
  // Start by registering the memory of the outgoing message with the RNIC
  rtn = cc_nsmr_register_virt(contextBlock->rnic,
                              CC_ADDR_TYPE_VA_BASED,
                              msg->data + DGRAM_HEADER_SIZE,
                              msg->size - DGRAM_HEADER_SIZE,
                              contextBlock->pd_id,
                              0, 0,
                              CC_ACF_LOCAL_READ | CC_ACF_LOCAL_WRITE,
                              &stag
                             );
  if (rtn != CC_OK) {
    // Let the user know what happened
    MACHSTATE2(3, "DeliverViaNetwork() - Ammasso - ERROR - Unable to register OutgoingMsg memory with RNIC: %d, \"%s\"", rtn, cc_status_to_string(rtn));
    return;
  }

  // Setup the Scatter/Gather List
  sgl.stag = stag;
  sgl.to = (cc_uint64_t)(unsigned long)(msg->data + DGRAM_HEADER_SIZE);
  sgl.length = msg->size - DGRAM_HEADER_SIZE;

  // Create the Work Request
  wr.wr_id = (cc_uint64_t)(unsigned long)&(wr);
  wr.wr_type = CC_WR_TYPE_SEND;
  wr.wr_u.send.local_sgl.sge_count = 1;
  wr.wr_u.send.local_sgl.sge_list = &sgl;
  wr.signaled = 1;

  // Post the Work Request to the Send Queue
  // DMK : TODO : Update this code to handle CCERR_QP_TOO_MANY_WRS_POSTED errors (pause breifly and retry)
  rtn = cc_qp_post_sq(contextBlock->rnic, otherNode->qp, &(wr), 1, &(WRsPosted));
  if (rtn != CC_OK || WRsPosted != 1) {
    // Let the user know what happened
    MACHSTATE2(3, "DeliverViaNetwork() - Ammasso - ERROR - Unable post Work Request to Send Queue: %d, \"%s\"", rtn, cc_status_to_string(rtn));
    displayQueueQuery(otherNode->qp, &(otherNode->qp_attrs));
  }
  ***************************************************************/

  MACHSTATE(2, "DeliverViaNetwork() - Ammasso - INFO: Completed.");

  AMMASSO_STATS_END(DeliverViaNetwork_post_lock)
  AMMASSO_STATS_END(DeliverViaNetwork)
}


/****************************************************************************
 *                                                                          
 * CheckSocketsReady
 *
 * Checks both sockets to see which are readable and which are writeable.
 * We check all these things at the same time since this can be done for
 * free with ``select.'' The result is stored in global variables, since
 * this is essentially global state information and several routines need it.
 *
 ***************************************************************************/

int CheckSocketsReady(int withDelayMs)
{   
  int nreadable;
  CMK_PIPE_DECL(withDelayMs);

  CmiStdoutAdd(CMK_PIPE_SUB);
  if (Cmi_charmrun_fd!=-1) CMK_PIPE_ADDREAD(Cmi_charmrun_fd);

  nreadable=CMK_PIPE_CALL();
  ctrlskt_ready_read = 0;
  dataskt_ready_read = 0;
  dataskt_ready_write = 0;
  
  if (nreadable == 0) {
    MACHSTATE(2, "} CheckSocketsReady (nothing readable)");
    return nreadable;
  }
  if (nreadable==-1) {
    CMK_PIPE_CHECKERR();
    MACHSTATE(3, "} CheckSocketsReady (INTERRUPTED!)");
    return CheckSocketsReady(0);
  }
  CmiStdoutCheck(CMK_PIPE_SUB);
  if (Cmi_charmrun_fd!=-1) 
          ctrlskt_ready_read = CMK_PIPE_CHECKREAD(Cmi_charmrun_fd);
  MACHSTATE(3, "} CheckSocketsReady");
  return nreadable;
}

/***********************************************************************
 * CommunicationServer()
 * 
 * This function does the scheduling of the tasks related to the
 * message sends and receives. 
 * It first check the charmrun port for message, and poll the gm event
 * for send complete and outcoming messages.
 *
 ***********************************************************************/

// NOTE: Always called from interrupt
static void ServiceCharmrun_nolock() {

  int again = 1;

  MACHSTATE(2, "ServiceCharmrun_nolock begin {");

  while (again) {
    again = 0;

    CheckSocketsReady(0);
    if (ctrlskt_ready_read) { ctrl_getone(); again=1; }
    if (CmiStdoutNeedsService()) { CmiStdoutService(); }
  }

  MACHSTATE(2, "} ServiceCharmrun_nolock end");
}

void processAmmassoControlMessage(char* msg, int len, Tailer *tail, OtherNode from) {

  int nodeIndex, ctrlType, i, n, nWR;
  AmmassoToken *token, *pretoken;
  AmmassoBuffer *tokenBuf;
  cc_data_addr_t *tokenSgl;
  OtherNode node;
  AmmassoTokenDescription *tokenDesc;

  AMMASSO_STATS_START(processAmmassoControlMessage)

  // Do not check the message, the flags field was set, and this is enough

  // Perform an action based on the control message type
  switch (tail->flags) {

  case AMMASSO_READY:

    // Decrement the node ready count by one
    contextBlock->nodeReadyCount--;
    MACHSTATE1(3, "processAmmassoControlMessage() - Ammasso - INFO: Received READY packet... still waiting for %d more...", contextBlock->nodeReadyCount);

    break;

  case AMMASSO_ALLOCATE: // Sent by the receiver to allocate more tokens

    token = getQPSendToken(from);
    tokenBuf = token->localBuf;

    tokenBuf->tail.length = 1;
    tokenBuf->tail.ack = 0;  // do not send any ACK with this message
    tokenBuf->tail.flags = AMMASSO_ALLOCATED;

    // Setup the local SGL
    tokenSgl = token->wr.wr_u.rdma_write.local_sgl.sge_list;
    tokenSgl->length = sizeof(Tailer);
    tokenSgl->to = (unsigned long)&tokenBuf->tail;
    token->wr.wr_u.rdma_write.remote_to = (unsigned long)&token->remoteBuf->tail;

    CC_POST_CHECK(cc_qp_post_sq,(contextBlock->rnic, node->qp, &token->wr, 1, &nWR),node->myNode);

    LIST_ENQUEUE(from->,usedTokens,token);
    from->max_used_tokens = (from->num_usedTokens>from->