msgs.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

  fprintf( stderr, "exit MatchingMsgPosted\n" );
#endif

  return 1;
}




int Msg_Send( msg_data, sender, recver, time, tag, size )
msgData *msg_data;
int sender, recver, tag, size;
double time;
{
  double recvTime;

    /* check if someone says they already received this */
  if (MatchingMsgPosted( msg_data, msg_data->recvq, recver, tag, &recvTime )) {
      /* we have a tachyon! */
    AddMsg( msg_data, sender, time, recver, recvTime, tag, size );
    return 1;
  }

  PostMsg( msg_data, msg_data->sendq, sender, tag, time );
  return 0;
}




int Msg_Recv( msg_data, recver, sender, time, tag, size )
msgData *msg_data;
int sender, recver, tag, size;
double time;
{
  double sendTime;

    /* check if someone says they already sent this */
  if (MatchingMsgPosted( msg_data, msg_data->sendq, sender, tag, &sendTime )) {
    AddMsg( msg_data, sender, sendTime, recver, time, tag, size );
    return 0;
  }

  PostMsg( msg_data, msg_data->recvq, recver, tag, time );
    /* received before being sent -- a tachyon */
  return 1;
}



int Msg_DoneAdding( msg_data )
msgData *msg_data;
{

    /* shrinkwrap the xpandList of message definitions */
  ListShrinkToFit( msg_data->defs, msgDefInfo );

    /* shrinkwrap the master list of messages */
  ListShrinkToFit( msg_data->list, msgInfo );

    /* create various indices */
  CreateIdx( msg_data );

  return 0;
}



int Msg_N( msg_data )
msgData *msg_data;
{
  return ListSize( msg_data->list, msgInfo );
}

int Msg_Get( msg_data, n, type, sender, recver, sendTime, recvTime, size )
msgData *msg_data;
int n, *type, *size, *sender, *recver;
double *sendTime, *recvTime;
{
  msgInfo *m;

  m = ListHeadPtr( msg_data->list, msgInfo ) + n;

  *type     = m->type;
  *size     = m->size;
  *sender   = m->sender;
  *recver   = m->recver;
  *sendTime = m->sendTime;
  *recvTime = m->recvTime;

  return 0;
}


#if USE_QSORT
static msgData *sort_data;

static int SetSortData( data )
msgData *data;
{
  sort_data = data;
  return 0;
}


static int CompareMsgSendTimes( a, b )
#if !defined(sparc) || defined(__STDC__)
const
#endif
void *a, *b;
{
  return (ListItem( sort_data->list, msgInfo, *(int*)a ).sendTime >
	  ListItem( sort_data->list, msgInfo, *(int*)b ).sendTime) ? 1 : -1;
  /* Chances are slim to none that doubles will be equal.  Doesn't
     matter either way if they are.  */
}


static int CompareMsgRecvTimes( a, b )
#if !defined(sparc) || defined(__STDC__)
const
#endif
void *a, *b;
{
  return (ListItem( sort_data->list, msgInfo, *(int*)a ).recvTime >
	  ListItem( sort_data->list, msgInfo, *(int*)b ).recvTime) ? 1 : -1;
  /* Chances are slim to none that doubles will be equal.  Doesn't
     matter either way if they are.  */
}
#endif



static int CreateIdx( msg_data )
msgData *msg_data;
{
  /* Create the following indices:

     idx_send
     idx_recv
     idx_proc_send
     idx_proc_recv

     */

  int i, proc;
  msgInfo *msg;

  ListCreate( msg_data->idx_send, int, msg_data->list->nused );
  ListCreate( msg_data->idx_recv, int, msg_data->list->nused );

  msg_data->idx_proc_send = (xpandList *)
    malloc( sizeof(xpandList) * msg_data->np );
  msg_data->idx_proc_recv = (xpandList *)
    malloc( sizeof(xpandList) * msg_data->np );
    /* one list of index references per process */

  for (proc=0; proc<msg_data->np; proc++) {
    ListCreate( msg_data->idx_proc_send[proc], int,
	        msg_data->list->nused / msg_data->np );
    ListCreate( msg_data->idx_proc_recv[proc], int,
	        msg_data->list->nused / msg_data->np );
  }

  msg = &ListItem( msg_data->list, msgInfo, 0 );
  /* Discard if the msg is to or from proc null */
  if (! (msg->sender < 0 || msg->recver < 0) ) {
      for (i=0; i<msg_data->list->nused; i++,msg++) {
	  ListAddItem( msg_data->idx_send, int, i );
	  ListAddItem( msg_data->idx_recv, int, i );
	  ListAddItem( msg_data->idx_proc_send[msg->sender], int, i );
	  ListAddItem( msg_data->idx_proc_recv[msg->recver], int, i );
      }
  }
  for (proc=0; proc<msg_data->np; proc++) {
    ListShrinkToFit( msg_data->idx_proc_send[proc], int );
    ListShrinkToFit( msg_data->idx_proc_recv[proc], int );
  }

    /* sort the big lists */
#if USE_QSORT
  SetSortData( msg_data );
  qsort( msg_data->idx_send->list, msg_data->idx_send->nused, sizeof(int),
	 CompareMsgSendTimes );
  qsort( msg_data->idx_recv->list, msg_data->idx_recv->nused, sizeof(int),
	 CompareMsgRecvTimes );
#else
  SortSend( msg_data, (int *)msg_data->idx_send->list,
            msg_data->idx_send->nused );
  SortRecv( msg_data, (int *)msg_data->idx_recv->list,
            msg_data->idx_recv->nused );
#endif

    /* sort the per-processor lists */
  for (proc=0; proc<msg_data->np; proc++) {
#if USE_QSORT
    qsort( msg_data->idx_proc_send[proc]->list, msg_data->idx_send->nused,
	   sizeof(int), CompareMsgSendTimes );
    qsort( msg_data->idx_proc_recv[proc]->list, msg_data->idx_recv->nused,
	   sizeof(int), CompareMsgRecvTimes );
#else
    SortSend( msg_data, (int *)msg_data->idx_proc_send[proc]->list,
              msg_data->idx_proc_send[proc]->nused );
    SortRecv( msg_data, (int *)msg_data->idx_proc_recv[proc]->list,
              msg_data->idx_proc_recv[proc]->nused );
#endif
  }

  return 0;
}


static int SortSend( msg_data, a, n )
msgData *msg_data;
int *a, n;     /* array and the number of elements in it */
{
  register int temp;
  register double time;
  int i, m, c;  /* insertPt, movePt, current */

  for (c=1; c<n-1; c++) {
    temp = a[c];
    time = ListItem( msg_data->list, msgInfo, a[c]).sendTime;
    for (i=c; i && time < ListItem( msg_data->list, msgInfo, a[i-1]).sendTime;
	 i--);
    for (m=c-1; m>=i; m--)
      a[m+1] = a[m];
    a[i] = temp;
  }
  return 0;
}

      
static int SortRecv( msg_data, a, n )
msgData *msg_data;
int *a, n;     /* array and the number of elements in it */
{
  register int temp;
  register double time;
  int i, m, c;  /* current, insertPt, movePt */

  for (c=1; c<n-1; c++) {
    temp = a[c];
    time = ListItem( msg_data->list, msgInfo, a[c]).recvTime;
    for (i=c; i && time < ListItem( msg_data->list, msgInfo, a[i-1]).recvTime;
	 i--);
    for (m=c-1; m>=i; m--)
      a[m+1] = a[m];
    a[i] = temp;
  }
  return 0;
}


int Msg_Close( data )
msgData *data;
{
  int proc;

  ListDestroy( data->defs, msgDefInfo );
  ListDestroy( data->list, msgInfo );

    /* if data has begun to be added */
  if (data->sendq) {
    for (proc=0; proc < data->np; proc++) {
      ListDestroy( data->sendq[proc], msgPost );
      ListDestroy( data->recvq[proc], msgPost );
    }
    free( data->sendq );
    free( data->recvq );
  }

    /* if the indices ahve been created */
  if (data->idx_send) {
    for (proc=0; proc < data->np; proc++) {
      ListDestroy( data->idx_proc_send[proc], int );
      ListDestroy( data->idx_proc_recv[proc], int );
    }
    free( data->idx_proc_send );
    free( data->idx_proc_recv );
    ListDestroy( data->idx_send, int );
    ListDestroy( data->idx_recv, int );
  }

  free( data );

  return 0;
}


#if TESTING

static int PrintMsg( msg )
msgInfo *msg;
{
  return fprintf( stderr, "Sent by %d at %f, Recvd by %d at %f, size %d\n",
		  msg->sender, msg->sendTime,
		  msg->recver, msg->recvTime, msg->size );
}

static int PrintMsgIdx( msg_data, idx )
msgData *msg_data;
xpandList /*int*/ idx;
{
  int n, i, *idxPtr;
  msgInfo *head;

  head = ListHeadPtr( msg_data->list, msgInfo );
  idxPtr = ListHeadPtr( idx, int );
  n = ListSize( idx, int );
  for (i=0; i<n; i++,idxPtr++) {
    PrintMsg( head + *idxPtr );
  }
  return 0;
}

int Msg_PrintAll( msg_data )
msgData *msg_data;
{
  msgInfo *msg;
  int nmsgs, i, proc;

  fprintf( stderr, "Master list of messages:\n" );

  nmsgs = ListSize( msg_data->list, msgInfo );
  msg = ListHeadPtr( msg_data->list, msgInfo );
  for (i=0; i<nmsgs; i++,msg++)
    PrintMsg( msg );

  if (!msg_data->idx_send) return 1;
  fprintf( stderr, "Indexed by send time:\n" );
  PrintMsgIdx( msg_data, msg_data->idx_send );

  fprintf( stderr, "Indexed by recv time:\n" );
  PrintMsgIdx( msg_data, msg_data->idx_recv );

  for (proc=0; proc<msg_data->np; proc++) {
    fprintf( stderr, "Messages on process %d:\n", proc );
    fprintf( stderr, "Indexed by send time:\n" );
    PrintMsgIdx( msg_data, msg_data->idx_proc_send[proc] );
    fprintf( stderr, "Indexed by recv time:\n" );
    PrintMsgIdx( msg_data, msg_data->idx_proc_recv[proc] );
  }
  return 0;
}
#endif