mpe_log_merge.c

fortran并行计算包

/*
   (C) 2001 by Argonne National Laboratory.
       See COPYRIGHT in top-level directory.
*/
/**\ --MPE_Log--
*  * mpe_log_merge.c - routines for performing a parallel merge of
*  *                   all the data logged by each process
*  *
*  * MPE_Log currently represents some code written by Dr. William
*  * Gropp, stolen from Chameleon's 'blog' logging package and
*  * modified by Ed Karrels, as well as some fresh code written
*  * by Ed Karrels.
*  *
*  * All work funded by Argonne National Laboratory
\**/

#ifndef MPE_HAS_PROCID
extern int MPE_Log_procid;
#endif

static MPE_Log_BLOCK *readBlock;
   /* this is used for flexibility with the b->reload stuff. */
   /* the only functions accessing it are MPE_Log_ParallelMerge and */
   /* MPE_Log_ReloadFromData */

#include <math.h>
#ifdef HUGE_VAL
#define TIME_INF HUGE_VAL
#else
#define TIME_INF 1.0e300
#endif
/* merge buffer size should be bigger */

/* Here is a structure to hold the original header and the source procid */
typedef struct {
  MPE_Log_HEADER header;
  int            procid;
} MPE_Log_MERGE_HEADER;


/* 
   Generate the header 
 */
static void MPE_Log_GenerateHeader( fp )
FILE *fp;
{
  int nevents, neventTypes, numProcs, totalnevents,
      totalneventTypes;
  double startTime, endTime, minimumTime, maximumTime;
  MPI_Comm_size(MPI_COMM_WORLD,&numProcs);

  MPE_Log_GetStatistics( &nevents, &neventTypes, &startTime, &endTime );
  MPI_Reduce( &nevents, &totalnevents, 1, MPI_INT, MPI_SUM, 0,
	      MPI_COMM_WORLD );
  MPI_Reduce( &neventTypes, &totalneventTypes, 1, MPI_INT, MPI_SUM, 0,
	      MPI_COMM_WORLD );
    /* get total number of events */
  MPI_Reduce( &startTime, &minimumTime, 1, MPI_DOUBLE, MPI_MIN, 0,
	      MPI_COMM_WORLD );
    /* get min. start time */
  MPI_Reduce( &endTime, &maximumTime, 1, MPI_DOUBLE, MPI_MAX, 0,
	      MPI_COMM_WORLD );
    /* get max. end time */

  if (MPE_Log_procid == 0) {
      char title[101];
#define FULL_TITLE
#ifdef FULL_TITLE
      struct passwd *pw;
      int    ln;
      time_t tloc;
      pw = getpwuid( getuid() );
      if (pw) {
	  sprintf( title, "Created by %s at ", pw->pw_name );
      }
      time( &tloc );
      strcat( title, ctime( &tloc ) );
      /* Must remove trailing newline */
      ln = strlen( title );
      if (title[ln-1] == '\n') title[ln-1] = 0;
#else
      strcpy( title, "Me" );
#endif

    fprintf( fp, "-1 0 0 0 0 0 %s\n", title );
    fprintf( fp, "-2 0 0 %d 0 0\n", totalnevents );
    fprintf( fp, "-3 0 0 %d 0 0\n", numProcs );
    fprintf( fp, "-4 0 0 1 0 0\n" );
    fprintf( fp, "-5 0 0 %d 0 0\n", totalneventTypes );

    /* There have been some problems with bogus minimum and maximum times */
    /* These need to be integers (for upshot) and made smaller */
    /* ALL processors should perform a uniform shift to make the mintime
       0 */
    fprintf( fp, "-6 0 0 0 0 %.0f\n", minimumTime*1000000 );
    fprintf( fp, "-7 0 0 0 0 %.0f\n", maximumTime*1000000 );
    fprintf( fp, "-8 0 0 1 0 0\n" );  /* timer cycles */
    fprintf( fp, "-11 %d 0 0 0 0\n", MPE_Log_procid);  /* 0 rollovers */
  }
}


/* Output from the buffer */    
static void MPE_Log_Output( inBuffer, outBuffer, mesgtag, srcs, fp, parent )
     MPE_Log_MBuf *inBuffer, *outBuffer;
     int      mesgtag, *srcs;
     FILE     *fp;
     int      parent;
{
  int recLen;
  MPE_Log_HEADER *recHdr;
  int recordBuf[MPE_Log_BUF_SIZE]; /* temporary storage for one record */

  recHdr  = (MPE_Log_HEADER *)(inBuffer->p);

#if DEBUG
  fprintf( debug_file, "output record to outBuffer %d %d\n",
	   outBuffer->p - outBuffer->buf, outBuffer->plast - outBuffer->buf );
#endif

  recLen = recHdr->len;
  if (recLen <= 0) {
    fprintf( stderr, "[%d] Error in log file; length of entry = %d\n",
	    MPE_Log_procid, recLen );
#if DEBUG
    fprintf( debug_file, "[%d] Error in log file; length of entry = %d\n",
	    MPE_Log_procid, recLen );
    fflush( debug_file );
#endif
    (*srcs)--;
    return;
  }
  if (parent >= 0) {		 /* if child process, */
    if (recLen + outBuffer->p >= outBuffer->plast) { /* if outBuffer is full */
#if DEBUG>1
      fprintf( debug_file, "Sending buffer to parent:\n" );
      PrintMbuf( debug_file, outBuffer );
      fflush( stderr );
#endif
      MPI_Send( outBuffer->buf, (outBuffer->p - outBuffer->buf) * 
	        sizeof( int ), MPI_BYTE, parent, mesgtag, MPI_COMM_WORLD );
        /* send outBuffer to parent */
        /* send as raw BYTEs, don't want MPI changing the data */
      outBuffer->p = outBuffer->buf; /* mark outBuffer as empty */
#if DEBUG
      fprintf( debug_file, "[%d] sent data to parent\n", MPE_Log_procid );
      fflush( debug_file );
#endif
    }
    memcpy( outBuffer->p, inBuffer->p, recLen*sizeof(int) );
      /* copy data from inBuffer to outBuffer */
    outBuffer->p += recLen;
  }
  else {			 /* if process 0 */
				 /* Repack the buffer */
    if (recHdr->event != MPE_Log_EVENT_SYNC) {
      memcpy( recordBuf, inBuffer->p, sizeof(MPE_Log_HEADER) );
        /* copy header to temp area */
      ((MPE_Log_HEADER *)recordBuf)->len--;
        /* cut out the procid that was inserted */
      memcpy( recordBuf + MPE_Log_HEADERSIZE, inBuffer->p +
	     MPE_Log_HEADERSIZE + 1,
	     (recHdr->len - MPE_Log_HEADERSIZE) * sizeof(int) );
        /* copy the log entry data to the temp area */

      MPE_Log_FormatRecord( fp, inBuffer->p[MPE_Log_HEADERSIZE], recordBuf );
        /* print the log entry to the logfile */
    }
  }
  inBuffer->p    += recLen;	 /* note the data used from the inBuffer */
  if (inBuffer->p >= inBuffer->plast) {	/* if the inBuffer is empty, */
    if (!(*(inBuffer->reload))( inBuffer, srcs )) {
        /* if there's no more data there,  */
      inBuffer->t = TIME_INF;	 /* mark that input source as empty */
      return;
    }
  } else {
    MOVEDBL( &inBuffer->t, &( ((MPE_Log_HEADER *)(inBuffer->p))->time ) );
      /* get next time */
  }
}


#define NUMINTS 4


static void MPE_Log_FormatRecord (fp, procid, rec)
FILE *fp;
int procid, *rec;
{
  MPE_Log_HEADER *hdr;
  MPE_Log_VFIELD *fld;
  int left;	       /* # of ints left in this record to be read */
  int i[NUMINTS], iused;     /* storage ints, and # used */
  char *str;	       /* string data */
  int intsToCopy, j;   /* nubmer of ints to copy, counter */
  double temp_time;

  hdr = (MPE_Log_HEADER *)rec;
  fprintf( fp, "%d %d ", hdr->event, procid );
  MOVEDBL( &temp_time, &hdr->time );
#if DEBUG>1 || DEBUG_FORMATRECORD
  fprintf( debug_file, "printing event %d from %d with size %d\n", hdr->event,
	  procid, hdr->len );
      fflush( debug_file );
#endif
  left = hdr->len;
  fld = (MPE_Log_VFIELD *)(hdr+1);
  left -= (sizeof(MPE_Log_HEADER) / sizeof(int));
  if (left > 0) {
    for (iused=0; iused<NUMINTS; iused++) {
      i[iused] = 0;
    }
    iused = 0;
    str    = "";
      /* clear everything */

    while (left) {
      /* There may be a string or integer data.  It there is integer data,
	 take the first element only */
      if (fld->dtype == MPE_Log_INT) {
#if DEBUG>1 || DEBUG_FORMATRECORD
	fprintf( debug_file, "%d ints left, this field is %d ints long\n",
		left, fld->len - MPE_Log_VFIELDSIZE(0) );
	fflush( debug_file );
#endif
	/* # of ints present */
	intsToCopy = (int)fld->len - MPE_Log_VFIELDSIZE(0); 

	  /* if intsToCopy>space left, just copy what we can */
	for (j=0; intsToCopy > 0 && iused<NUMINTS; j++,iused++,intsToCopy--)
	    /* copy the integers */
	  i[iused]=fld->other[j];

      } else if (fld->dtype == MPE_Log_CHAR) {
	str = (char *)(fld->other);
      }
      left -= fld->len;
#if DEBUG>1 || DEBUG_FORMATRECORD
      fprintf( debug_file, "%d ints left, %d taken away, %d iused\n", left,
	      fld->len, iused );
      fflush( debug_file );
#endif
      fld = (MPE_Log_VFIELD *)((int *)fld + fld->len);
    }

    if (hdr->event == LOG_STATE_DEF) 
      fprintf( fp, "%d %d 0 0 %s\n", i[0], i[1], str );
      /* State events are special - they have two data values  */
    else if (hdr->event == CLOG_MSG_SEND || hdr->event == CLOG_MSG_RECV)
      fprintf( fp, "0 %d 0 %.0f %d %d\n", i[0],
	       temp_time*1000000, i[1], i[2] );
      /* Sends and receives need 3:  otherParty, tag, size*/
    else
	fprintf( fp, "0 %d 0 %.0f %s\n", 
		i[0], temp_time*1000000, str );
  }  /* if (left>0) */
  else
    fprintf( fp, "0 0 0 %.0f\n", temp_time*1000000);
}


static int MPE_Log_ReloadFromData( destBuffer, srcs )
MPE_Log_MBuf *destBuffer;
int      *srcs;
{
  MPE_Log_HEADER *readHdr;
  int             readBufSize, intsUsed, *readPtr, *writePtr;
  /* Destination; repacked as (header)(procid)(fields) */

  /* Note: this function copies all the data from one MPE_Log_BLOCK to */
  /* one MPE_Log_MBuf.  MBufs must be bigger because each entry has an */
  /* extra int */

  if (readBlock) {
    /* We have to insert the procid as an int after the header and before
       any vfields */
    writePtr    = destBuffer->buf;
    readBufSize = readBlock->size;
    readPtr     = (int *)(readBlock + 1);
    intsUsed    = 0;
/*
   fprintf( stderr, "[%d] reading buffer of size %d, srcs = %d\n",
            MPE_Log_procid, readBufSize, *srcs );
*/
    while (intsUsed < readBufSize) {
      readHdr  = (MPE_Log_HEADER *)readPtr;
/*
fprintf( stderr, "[%d] reloading %d\n", MPE_Log_procid, h->event );
*/
      if (readHdr->event <= MAX_HEADER_EVT && readHdr->event >= MIN_HEADER_EVT)
	/* Reserved header events have all times set to zero */
	MPE_Log_ZEROTIME(readHdr);
      memcpy( writePtr, readPtr, sizeof(MPE_Log_HEADER) );
				 /* copy header */
      ((MPE_Log_HEADER *)writePtr)->len = readHdr->len + 1;
				 /* increase record length by 1 */
      writePtr[MPE_Log_HEADERSIZE] = MPE_Log_procid;
				 /* insert procid */
      memcpy( writePtr + MPE_Log_HEADERSIZE + 1, readPtr + MPE_Log_HEADERSIZE, 
	     (readHdr->len - MPE_Log_HEADERSIZE) * sizeof(int) );
				/* copy all the fields */
      /* Increment the lengths (writePtr includes the procid) */
      intsUsed += readHdr->len;
#if DEBUG_RELOAD
      fprintf( debug_file, "reload from: " ); 
      PrintRecord( debug_file, readPtr );
      fflush( debug_file );
#endif
      readPtr  += readHdr->len;
#if DEBUG_RELOAD
      fprintf( debug_file, "reload to: " );
      PrintMbufRecord( debug_file, writePtr );
      fflush( debug_file );
#endif
      writePtr += readHdr->len + 1;
    }
    destBuffer->p = destBuffer->buf; /* reset the pointer in the new buffer */
    destBuffer->plast = writePtr; /* mark the end of the buffer */
    readHdr = (MPE_Log_HEADER *)(destBuffer->p);
    MOVEDBL( &destBuffer->t, &readHdr->time ); /* make a copy of the time */
    readBlock = readBlock->next; /* go to the next block */
#if DEBUG
    PrintMbuf( debug_file, destBuffer );
      fflush( debug_file );
#endif
    return 1;
  }
  else {	/* no more local data */
    destBuffer->t = TIME_INF;
    destBuffer->p = destBuffer->plast = destBuffer->buf;
    (*srcs)--;
#if DEBUG
    fprintf( debug_file, "[%d] done reloading from data\n", MPE_Log_procid );
      fflush( debug_file );
#endif
    return 0;
  }
}


/* 
   There are two routines to reload the buffer.  One gets data from
   another processor (FromChild); the other from the internal buffer 
 */

static int MPE_Log_ReloadFromChild( destBuffer, msgtype, srcs )
int msgtype, *srcs;
MPE_Log_MBuf *destBuffer;
{
  int ln;